Axum 0.8 SSE Stream Disconnect Lifecycle and Broadcast Backpressure Invariant

--- description: Axum 0.8 SSE Stream Disconnect Lifecycle and Broadcast Backpressure Invariant - Mandates deterministic client disconnect propagation and Tokio broadcast subscriber teardown for Server-Sent Events (SSE) endpoints in Axum 0.8. Eliminates zombie worker tasks and ghost compute leaks by binding stream drop semantics directly to Tokio cancellation tokens and handling broadcast lag errors defensively. globs: src/**/*.rs alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/axum-sse-backpressure-disconnect-cleanup" --- # Axum 0.8 SSE Stream Disconnect Lifecycle and Broadcast Backpressure Invariant > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning Unlinked Long-Polling Background Tasks and Leaking Tokio Broadcast Receivers on Client Abort ## System Constraints You are a Principal Rust Systems Architect enforcing production-grade Axum 0.8 Server-Sent Events (SSE) stream lifecycles. You must prevent ghost compute leaks, unbounded subscriber queues, and orphaned background worker tasks by mandating RAII stream drops, Tokio CancellationToken integration, and explicit backpressure handling on broadcast channels. ## Strict Guidelines - Engineers MUST ALWAYS bind SSE stream lifecycles to an RAII drop guard or a Tokio CancellationToken to terminate upstream processing immediately when the HTTP connection drops. - Engineers MUST NOT spawn detached Tokio tasks (`tokio::spawn`) inside SSE handlers without retaining and terminating an `AbortHandle` or polling a `CancellationToken` linked to the connection drop. - Code MUST handle `tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged` explicitly instead of silently dropping the stream or panicking, either by forwarding a lag notification or dropping gracefully. - Streams MUST yield `Result<axum::response::sse::Event, std::convert::Infallible>` or a defined error type that Axum's SSE implementation cleanly flushes before terminating. - Handlers MUST configure `axum::response::Sse::keep_alive` with a bounded interval to proactively detect TCP half-open connections and broken peer sockets. ## Bad vs Good Implementations ### Leaking Detached Background Task and Swallowing Broadcast Lag vs. RAII Cancellation Guard #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawns an unlinked background worker via `tokio::spawn` with zero drop propagation when the HTTP client disconnects. Additionally, `filter_map` blindly drops `BroadcastStreamRecvError::Lagged` events without emitting backpressure warnings, causing client desynchronization and persistent resource leakage. ```rust use axum::{response::sse::{Event, Sse}, extract::State}; use tokio::sync::broadcast; use tokio_stream::StreamExt; use tokio_stream::wrappers::BroadcastStream; use std::convert::Infallible; // ANTI-PATTERN: Detached task runs indefinitely after client disconnects. // BroadcastStreamRecvError::Lagged is ignored or causes unhandled termination. pub async fn sse_handler( State(tx): State<broadcast::Sender<String>>, ) -> Sse<impl tokio_stream::Stream<Item = Result<Event, Infallible>>> { let rx = tx.subscribe(); // Detached task has no visibility into HTTP disconnects: GHOST COMPUTE LEAK tokio::spawn(async move { loop { tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; // Heavy compute or database polling continues forever after peer closes connection } }); let stream = BroadcastStream::new(rx).filter_map(|res| { // Swallowing errors causes silently stalled streams res.ok().map(|msg| Ok(Event::default().data(msg))) }); Sse::new(stream) } ``` #### Production Standard (Correct) > **Standard**: Uses a custom RAII `StreamGuard` carrying a Tokio `CancellationToken`. When Axum drops the SSE stream upon HTTP peer disconnect, the guard executes `drop()`, signalling all child tasks to terminate instantly. Explicitly handles `BroadcastStreamRecvError::Lagged` and provisions keep-alive heartbeats to reliably detect broken TCP sockets. ```rust use axum::{response::sse::{Event, KeepAlive, Sse}, extract::State}; use tokio::sync::broadcast; use tokio_stream::{Stream, StreamExt}; use tokio_stream::wrappers::{BroadcastStream, errors::BroadcastStreamRecvError}; use tokio_util::sync::CancellationToken; use std::{convert::Infallible, time::Duration}; // PRODUCTION INVARIANT: Stream drop immediately cancels worker via RAII token. pub async fn sse_handler( State(tx): State<broadcast::Sender<String>>, ) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { let rx = tx.subscribe(); let cancel_token = CancellationToken::new(); let child_token = cancel_token.child_token(); // Bound background tasks strictly to the SSE stream lifespan tokio::spawn(async move { tokio::select! { _ = child_token.cancelled() => { // Clean worker teardown on disconnect } _ = async { // Scoped long-polling or calculation workload } => {} } }); let broadcast_stream = BroadcastStream::new(rx); let mapped_stream = broadcast_stream.filter_map(|result| match result { Ok(msg) => Some(Ok(Event::default().data(msg))), Err(BroadcastStreamRecvError::Lagged(missed)) => { tracing::warn!(missed_events = missed, "SSE subscriber lagged behind"); Some(Ok(Event::default().event("system_error").data("stream_lagged"))) } }); // Guard struct invokes cancel_token.cancel() when Axum drops the stream struct StreamGuard<S> { inner: S, token: CancellationToken, } impl<S: Stream + Unpin> Stream for StreamGuard<S> { type Item = S::Item; fn poll_next( mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll<Option<Self::Item>> { std::pin::Pin::new(&mut self.inner).poll_next(cx) } } impl<S> Drop for StreamGuard<S> { fn drop(&mut self) { self.token.cancel(); } } let guarded_stream = StreamGuard { inner: mapped_stream, token: cancel_token, }; Sse::new(guarded_stream) .keep_alive(KeepAlive::new().interval(Duration::from_secs(15))) } ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: Axum 0.8 SSE Stream Disconnect Lifecycle and Broadcast Backpressure Invariant

Axum 0.8 SSE Stream Disconnect Lifecycle and Broadcast Backpressure Invariant defines an industrial-grade behavioral standard for backend engineering. Mandates deterministic client disconnect propagation and Tokio broadcast subscriber teardown for Server-Sent Events (SSE) endpoints in Axum 0.8. Eliminates zombie worker tasks and ghost compute leaks by binding stream drop semantics directly to Tokio cancellation tokens and handling broadcast lag errors defensively. Modern software architectures require strict adherence to deterministic patterns to guarantee high availability, maintainability, and zero security vulnerabilities.

Framework Paradigm Shift & Drift MechanicsLLMs are trained on millions of open-source repositories containing outdated code, legacy idioms, and unpatched antipatterns. When coding assistants generate implementations for this stack, they default to common statistical tokens: Spawning Unlinked Long-Polling Background Tasks and Leaking Tokio Broadcast Receivers on Client Abort Without an explicit system instruction file, developers spend hours debugging silent failures.

Failure Modes & Production Blast Radius

Hazard Analysis: What Breaks Without This Rule

The following failure modes have been cataloged from hundreds of hours of production code reviews where AI assistants produced non-compliant implementations:

Failure Mode	AI Symptom (The Anti-Pattern)	Severity	Production Impact & Remedy
AI Hallucination & Deprecation Drift	Spawning Unlinked Long-Polling Background Tasks and Leaking Tokio Broadcast Receivers on Client Abort	Critical	Violating this industrial standard introduces regression bugs, security risks, and unexpected runtime failures in production. Enforce explicit behavioral system directives across all AI coding assistants in your repository.
Silent Architecture Inconsistency	AI assistants produce mixed paradigms across files, creating fragmented codebase conventions.	Medium	Increases cognitive load during code reviews and leads to maintenance debt across engineering teams. Pin project conventions in .cursor/rules, CLAUDE.md, and AGENTS.md so all team agents follow identical guardrails.

Architectural Directives in Detail

Step-by-Step Production Directives

Each rule constraint below is injected into the AI agent's system prompt to establish deterministic boundaries:

Engineers MUST ALWAYS bind SSE stream lifecycles to an RAII drop guard or a Tokio CancellationToken to terminate upstream processing immediately when the HTTP connection drops.

Rationale:AI models frequently generate unconstrained code here because statistical language distributions lack project-specific architectural context. Mandating this directive ensures that generated code adheres to verified industrial production requirements.