Axum 0.8 Tokio Streaming Backpressure & SSE Connection Draining Standard

--- description: Axum 0.8 Tokio Streaming Backpressure & SSE Connection Draining Standard - Enforces bounded channels, consumer-driven stream backpressure, and coordinated connection draining for Server-Sent Events (SSE) and WebSockets during zero-downtime rolling deploys in Axum 0.8. Prevents memory leaks caused by unbounded queueing and abrupt client socket terminations using CancellationToken and graceful shutdown barriers. globs: src/**/*.rs alwaysApply: true verified_by: "agentrules.org" license: "CC-BY-4.0" source: "https://agentrules.org/rules/axum-08-tokio-streaming-backpressure-sse-draining" --- # Axum 0.8 Tokio Streaming Backpressure & SSE Connection Draining Standard > **Severity**: CRITICAL | **Category**: backend > **Anti-Pattern Warning**: Unbounded mpsc Channel Buffering and Abrupt Connection Severing on SIGTERM ## System Constraints You are a Principal Rust Distributed Systems Engineer enforcing Axum 0.8 and Tokio asynchronous streaming invariants. Mandate bounded channel backpressure, CancellationToken propagation for long-lived SSE/WebSocket connections, and explicit drain phases before hyper runtime termination during SIGTERM events. ## Strict Guidelines - Engineers MUST ALWAYS use bounded channels (`tokio::sync::mpsc::channel`) with explicit capacity and handle backpressure failures instead of `unbounded_channel`. - Engineers MUST propagate `tokio_util::sync::CancellationToken` or watch receivers to all long-lived SSE and WebSocket connection handler loops. - Services MUST emit a final event (e.g., SSE retry or close frame) and cleanly flush pending buffer state when a shutdown signal is received before severing TCP sockets. - Engineers MUST configure `axum::serve::Serve::with_graceful_shutdown` and coordinate active stream completion using a tracker like `tokio_util::task::TaskTracker`. - Stream adapters MUST NOT swallow `std::io::ErrorKind::BrokenPipe` or disconnect signals, and MUST terminate the underlying producer task when consumer poll yields `Poll::Ready(None)`. ## Bad vs Good Implementations ### SSE Connection Handling with Backpressure and Coordinated Draining #### Bad Antipattern (AI Common Hallucination) > **Why it fails**: Uses an unbounded channel that leads to unbounded memory growth if the client has slow TCP window updates. Completely ignores server shutdown signals, causing abrupt socket drops during rolling deployments without signaling clients to reconnect. ```rust use axum::{response::sse::{Event, KeepAlive, Sse}, routing::get, Router}; use futures_util::stream::Stream; use std::convert::Infallible; use tokio::sync::mpsc; use tokio_stream::wrappers::UnboundedReceiverStream; pub async fn leak_and_drop_sse() -> Sse<impl Stream<Item = Result<Event, Infallible>>> { // ANTI-PATTERN: Unbounded channels cause OOM under slow consumers. // No shutdown hook/cancellation token means abrupt termination on SIGTERM. let (tx, rx) = mpsc::unbounded_channel(); tokio::spawn(async move { let mut count = 0; loop { tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; count += 1; let _ = tx.send(Event::default().data(count.to_string())); } }); Sse::new(UnboundedReceiverStream::new(rx)).keep_alive(KeepAlive::default()) } ``` #### Production Standard (Correct) > **Standard**: Applies explicit backpressure via bounded mpsc channel (capacity 64). Uses CancellationToken with tokio::select! to detect rolling deployment shutdown, sends a clean closure event, and drops the producer channel cleanly when the client disconnects. ```rust use axum::{ extract::State, response::sse::{Event, KeepAlive, Sse}, routing::get, Router, }; use futures_util::stream::Stream; use std::{convert::Infallible, sync::Arc, time::Duration}; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; #[derive(Clone)] pub struct AppState { pub shutdown_token: CancellationToken, } pub async fn drained_sse_handler( State(state): State<Arc<AppState>>, ) -> Sse<impl Stream<Item = Result<Event, Infallible>>> { // Bounded channel to enforce backpressure against slow network consumers let (tx, rx) = mpsc::channel::<Result<Event, Infallible>>(64); let token = state.shutdown_token.clone(); tokio::spawn(async move { let mut ticker = tokio::time::interval(Duration::from_millis(500)); let mut seq: u64 = 0; loop { tokio::select! { biased; _ = token.cancelled() => { // Gracefully signal termination and exit producer loop let close_event = Event::default() .event("shutdown") .data("draining_connection"); let _ = tx.send(Ok(close_event)).await; break; } _ = ticker.tick() => { seq += 1; let event = Event::default().id(seq.to_string()).data("heartbeat"); // Backpressure: waits if receiver buffer is full if tx.send(Ok(event)).await.is_err() { // Consumer disconnected; terminate producer task immediately break; } } } } }); Sse::new(ReceiverStream::new(rx)).keep_alive(KeepAlive::new().interval(Duration::from_secs(15))) } ```

Architectural Specification & Invariant Rationale

Architectural Invariant Design: Axum 0.8 Tokio Streaming Backpressure & SSE Connection Draining Standard

Axum 0.8 Tokio Streaming Backpressure & SSE Connection Draining Standard defines an industrial-grade behavioral standard for backend engineering. Enforces bounded channels, consumer-driven stream backpressure, and coordinated connection draining for Server-Sent Events (SSE) and WebSockets during zero-downtime rolling deploys in Axum 0.8. Prevents memory leaks caused by unbounded queueing and abrupt client socket terminations using CancellationToken and graceful shutdown barriers. 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: Unbounded mpsc Channel Buffering and Abrupt Connection Severing on SIGTERM 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	Unbounded mpsc Channel Buffering and Abrupt Connection Severing on SIGTERM	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 use bounded channels (`tokio

Engineers MUST ALWAYS use bounded channels (`tokio::sync::mpsc::channel`) with explicit capacity and handle backpressure failures instead of `unbounded_channel`.

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.