Axum SSE Channel Disconnection & Task Cancellation Standard

--- description: Axum SSE Channel Disconnection & Task Cancellation Standard - Enforces proactive cancellation propagation via Tokio AbortHandles or CancellationToken during Axum Server-Sent Events (SSE) stream termination. Prevents orphaned background worker tasks, zombie database connections, and unbounded memory leaks caused by decoupled mpsc channels when HTTP clients abruptly disconnect. globs: src/**/*.rs alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/axum-sse-task-cancellation-standard" --- # Axum SSE Channel Disconnection & Task Cancellation Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Spawning detached Tokio tasks feeding unmonitored SSE channels without drop-detection or cancellation tokens ## System Constraints You are a Principal Rust Architect enforcing strict lifecycle safety and task cancellation invariants in Axum SSE endpoints. You must reject any code where an asynchronous producer is spawned without direct binding to the SSE response stream's lifecycle via Tokio CancellationToken, tokio_util::sync::DropGuard, or explicit stream termination hooks. ## Strict Guidelines - Engineers MUST ALWAYS tie the lifecycle of any background producer task to the client SSE stream lifetime using `tokio_util::sync::CancellationToken` or RAII drop guards. - Engineers MUST NOT spawn detached tasks (`tokio::spawn(async move { ... })`) that pipe items into an SSE stream without a cancellation listener (`select!` or `cancellation_token.cancelled()`). - Services MUST monitor channel send errors (`tx.send().await.is_err()`); when the consumer receiver drops due to client disconnection, the producer task MUST terminate immediately. - Code SHOULD prefer converting streams directly using `tokio_stream::wrappers::ReceiverStream` or `async_stream::stream!` bound to the request scope rather than spawning detached worker tasks. - Producer loops MUST abort in-flight database cursors and release pool connections immediately upon client disconnect detection. ## Bad vs Good Implementations ### SSE Background Task Lifecycle Binding #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Spawns a detached Tokio task without cancellation signaling. If `tx.send().await` is ignored or buffered work occurs between sends, the background worker leaks indefinitely, keeping database handles and memory allocations alive long after the client disconnects. ```rust use axum::response::sse::{Event, Sse}; use futures::stream::Stream; use std::convert::Infallible; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; pub async fn bad_sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> { let (tx, rx) = mpsc::channel::<Result<Event, Infallible>>(10); // ANTI-PATTERN: Detached task spawned with no cancellation token. // If the client disconnects, `rx` drops, but this task might continue // executing costly DB work or loop indefinitely if send isn't checked. tokio::spawn(async move { let mut counter = 0; loop { tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; counter += 1; // Ignoring result or failing to clean up resources let _ = tx.send(Ok(Event::default().data(counter.to_string()))).await; } }); Sse::new(ReceiverStream::new(rx)) } ``` #### Production Standard (Correct) > **Standard**: Implements cooperative cancellation via `CancellationToken` and `drop_guard`. When the HTTP client disconnects, Axum drops the stream, firing the drop guard and terminating the Tokio worker immediately, preventing resource leaks. ```rust use axum::response::sse::{Event, Sse}; use futures::stream::Stream; use std::convert::Infallible; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; pub async fn good_sse_handler() -> Sse<impl Stream<Item = Result<Event, Infallible>>> { let (tx, rx) = mpsc::channel::<Result<Event, Infallible>>(10); let token = CancellationToken::new(); let child_token = token.child_token(); tokio::spawn(async move { let mut counter = 0; loop { tokio::select! { _ = child_token.cancelled() => { tracing::info!("Client disconnected, terminating worker."); break; } _ = tokio::time::sleep(tokio::time::Duration::from_secs(1)) => { counter += 1; let event = Ok(Event::default().data(counter.to_string())); if tx.send(event).await.is_err() { tracing::info!("SSE receiver dropped, terminating worker."); break; } } } } }); // Guard triggers token cancellation when the Sse wrapper drops the stream let stream = ReceiverStream::new(rx); let guarded_stream = async_stream::stream! { let _guard = token.drop_guard(); for await item in stream { yield item; } }; Sse::new(guarded_stream) } ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: Axum SSE Channel Disconnection & Task Cancellation Standard

Axum SSE Channel Disconnection & Task Cancellation Standard defines an industrial-grade behavioral standard for backend engineering. Enforces proactive cancellation propagation via Tokio AbortHandles or CancellationToken during Axum Server-Sent Events (SSE) stream termination. Prevents orphaned background worker tasks, zombie database connections, and unbounded memory leaks caused by decoupled mpsc channels when HTTP clients abruptly disconnect. 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 detached Tokio tasks feeding unmonitored SSE channels without drop-detection or cancellation tokens 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 detached Tokio tasks feeding unmonitored SSE channels without drop-detection or cancellation tokens	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 tie the lifecycle of any background producer task to the client SSE stream lifetime using `tokio_util

Engineers MUST ALWAYS tie the lifecycle of any background producer task to the client SSE stream lifetime using `tokio_util::sync::CancellationToken` or RAII drop guards.

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.