perf(server-rust): cut SSE fan-out per-subscriber memory ~60% (#4661)

## What

The Rust durable-streams server added **linear memory per SSE
subscriber** in the fan-out scenario (1 stream, 1 writer, many
subscribers), while the ursula reference keeps it flat — see [ds-bench
results](https://github.com/electric-sql/ds-bench/blob/main/results/sse/sse-comparison.md).
This trims the per-subscriber footprint by ~60%.

### Root cause
With the socket-buffer cap already in place, the residual growth was
**userspace heap per connection**. Each live SSE subscriber:
- spawned a **producer task** + allocated an **mpsc channel**, and
- kept the whole `conn_loop` state machine future resident while parked
(up to `SSE_MAX_DURATION`),
- plus a pinned 4 KiB read buffer.

≈ **14 KiB/subscriber**.

### Fix
- **Inline SSE production** via a new pull-based `Body::Sse` /
`EventSource` — no per-subscriber producer task, no channel. All
caught-up subscribers still share the one resident tail chunk.
- **Hand off to a dedicated streaming task** when a request is SSE, so
the large `conn_loop` future is freed while a subscriber is parked. The
connection permit moves with it (stays counted); the socket closes when
the stream ends (the client reconnects from its last offset).

An idle subscriber's resident state collapses to roughly a cursor over
the shared stream tail.

## Results

Isolated server-process RSS on Linux/cgroup (the ds-bench methodology),
wal mode:

| | per-subscriber slope | @ 2000 subs | idle baseline |
|---|---|---|---|
| before | ~13.6 KiB/sub | ~30 MiB | 3 MiB |
| **after** | **~5 KiB/sub** | **~13.5 MiB** | 3 MiB |

This keeps our low idle baseline (vs ursula's ~15 MiB mimalloc floor)
while bringing the slope into the same regime, so we're comparable at
scale instead of worse.

> Note: the official ds-bench SSE suite (`scripts/run-sse.sh`) is
GKE-only; the numbers above are from a faithful local Docker
reproduction using the same cgroup working-set metric. Re-running the
real suite on GKE is recommended to confirm the slope drop at scale.

### Allocators evaluated (and rejected)
mimalloc and jemalloc were both measured — both are **worse** for our
allocation shape (mimalloc: ~6× higher absolute, ~3× worse slope),
because our per-connection state is kilobyte-sized rather than
cursor-sized. glibc stays the default.

## Verification
- `cargo test --release`: 87 passed
- Conformance suite: 326 passed across wal / tail-cache-on / memory
(Linux) modes
- Single correct `Connection: keep-alive` header on the wire; keep-alive
reuse intact

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: balegas

.changeset/sse-fanout-per-subscriber-memory.md

@@ -0,0 +1,7 @@
+---
+'@electric-ax/durable-streams-server-rust': patch
+---
+
+Cut SSE fan-out per-subscriber memory by ~60%. Each live subscriber used to spawn a producer task and an mpsc channel and keep the whole connection state machine resident while parked. SSE is now produced inline (new pull-based `Body::Sse`) and the connection is handed to a small dedicated streaming task, so an idle subscriber's resident footprint collapses to roughly a cursor over the shared stream tail.
+
+Live-tail SSE subscribers are then served from a fixed pool of epoll reactor threads instead of a parked connection task per subscriber. Each subscriber becomes a compact slab entry, so per-subscriber resident memory drops from ~7 KiB to ~0.6 KiB and stops scaling with the number of active connections. Linux only; other platforms keep the existing path.

packages/durable-streams-rust/README.md

@@ -208,6 +208,8 @@ publishing is re-enabled.
 
 ## Benchmarks
 
-Numbers from **[ds-bench](https://github.com/electric-sql/ds-bench)**, a reproducible single-node harness. All figures below are the default **`wal` mode** (group-commit fsync, resident tail cache off). One server node (`c4d-standard-16-lssd`) pinned to **4 CPUs**, a Kubernetes client fleet driving 256-byte binary appends. Throughput is the saturation ceiling; latency is fleet-wide p50 / p99; memory is the pod cgroup working set (anon + active page cache), peak / p50.
+Numbers from **[ds-bench](https://github.com/electric-sql/ds-bench)**, a reproducible single-node harness — full matrix, methodology, and the comparison systems are in the **[2026-06-30 report](https://github.com/electric-sql/ds-bench/blob/feat/read-scalability-workload/results-2026-06-30/REPORT.md)** (measured on this build). One server node (`c4d-standard-16-lssd`) pinned to **4 CPUs**, a Kubernetes client fleet driving 256-byte binary appends. Throughput is the saturation ceiling; latency is fleet-wide p50 / p99; memory is the pod cgroup working set (anon + active page cache), peak / p50.
 
-**Writes** — peaks at **860,000 append/s** at 4 CPUs, scales cleanly to **100k streams**, with median append latency staying sub-ms → ~1.5 ms. Memory tracks **stream count, not bytes** (each stream is a lean record plus its open file; data lives on disk / in the page cache, never resident), so it stays in tens–hundreds of MiB even at 100k streams, with p50 ≪ peak.
+**Writes** (`wal` mode, group-commit fsync) — peak **~928,000 append/s** at 4 CPUs, scaling cleanly to **100k streams**, with median append latency sub-ms → ~1.5 ms through 10k streams. Memory tracks **stream count, not bytes** (each stream is a lean record plus its open file; data lives on disk / in the page cache, never resident), so it stays in tens–hundreds of MiB even at 100k streams, with p50 ≪ peak — versus the 1–3.5 GB a log-resident design holds at the same load.
+
+**Live-tail reads (SSE reactor)** — the per-core epoll reactor serves SSE live tail at **p99 ~0.5–2.5 ms from 64 to 2048 concurrent connections**, the flattest-scaling read path (long-poll holds ~5–7 ms at the same fan-out; catch-up's resident re-scan does not scale past ~64 connections). **Fan-out memory is shared, not per-subscriber**: 1000 SSE subscribers on one stream cost **~27 MiB** total — one resident tail served to all — so subscriber count is decoupled from memory.

packages/durable-streams-rust/src/api.rs

@@ -5,6 +5,8 @@
 // optional framing bytes, and the engine decides how to serve them (buffered
 // copy, or sendfile where the platform can).
 
+use std::future::Future;
+use std::pin::Pin;
 use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 
@@ -13,6 +15,39 @@ use tokio::sync::mpsc;
 
 use crate::store::Segment;
 
+/// A pull-based streaming body, driven INLINE on the connection task: the engine
+/// calls `next_chunk` in a loop and writes each returned frame, ending when it
+/// yields `None`. Unlike `Body::Channel` (a spawned producer task feeding an
+/// mpsc channel), this adds NO per-response task and NO channel — the producer
+/// runs on the connection's own task. SSE fan-out uses it so each subscriber
+/// costs only its connection's existing footprint, not an extra ~4–5 KiB of
+/// resident task+channel heap (the source of per-subscriber linear memory).
+pub trait EventSource: Send {
+    /// Produce the next framed chunk, or `None` to end the stream. The returned
+    /// future borrows `self`, so state lives in the source (no channel buffer).
+    fn next_chunk(&mut self) -> Pin<Box<dyn Future<Output = Option<Bytes>> + Send + '_>>;
+
+    /// Reactor eligibility (Linux). When this returns `Some`, the connection task
+    /// hands the socket to the epoll reactor instead of driving the stream
+    /// inline, freeing its future — the per-subscriber cost. Default: not
+    /// eligible (inline / hand-off, as for every non-SSE source).
+    #[cfg(target_os = "linux")]
+    fn reactor_reg(&self) -> Option<SseReg> {
+        None
+    }
+}
+
+/// Live-tail SSE registration: everything the reactor needs to serve a
+/// subscriber once the connection task has handed off its socket. Constructed
+/// only on Linux (`SseSource::reactor_reg`).
+#[cfg(target_os = "linux")]
+pub struct SseReg {
+    pub st: Arc<crate::store::StreamState>,
+    pub start: u64,
+    pub encoding: crate::handlers::SseEncoding,
+    pub client_cursor: Option<u64>,
+}
+
 /// A streamed (chunked) response body plus an abort signal.
 ///
 /// `rx` carries the body frames. `failed` is set by the producer if it ends the
@@ -27,16 +62,6 @@ pub struct StreamBody {
     pub failed: Arc<AtomicBool>,
 }
 
-impl StreamBody {
-    /// A stream with no failure path (e.g. SSE, where the client reconnects).
-    pub fn infallible(rx: mpsc::Receiver<Bytes>) -> Self {
-        StreamBody {
-            rx,
-            failed: Arc::new(AtomicBool::new(false)),
-        }
-    }
-}
-
 /// Maximum accepted request body size; larger bodies get `413 Payload Too
 /// Large`.
 pub const MAX_BODY_BYTES: usize = 1024 * 1024 * 1024; // 1 GiB safety cap
@@ -95,10 +120,14 @@ impl Req {
 pub enum Body {
     Empty,
     Full(Bytes),
-    /// Streaming body (SSE / large dynamic responses); engine frames it
-    /// (e.g. chunked transfer-encoding) and ends when the channel closes. The
-    /// `failed` flag tells the engine to abort rather than terminate cleanly.
+    /// Streaming body for large dynamic responses (e.g. cold-tier reads): a
+    /// spawned producer feeds frames over a channel. The `failed` flag tells the
+    /// engine to abort rather than terminate cleanly. For SSE fan-out, prefer
+    /// `Body::Sse`, which avoids the per-response task + channel.
     Channel(StreamBody),
+    /// Pull-based streaming body driven inline on the connection task (SSE).
+    /// No spawned task, no channel — see `EventSource`.
+    Sse(Box<dyn EventSource>),
     /// Byte ranges of data files plus optional framing (JSON `[` / `]`).
     /// Total body length is prefix + Σ segment.len + suffix.
     FileRange {
@@ -120,6 +149,7 @@ impl Body {
             Body::Empty => Some(0),
             Body::Full(b) => Some(b.len() as u64),
             Body::Channel(_) => None,
+            Body::Sse(_) => None,
             Body::FileRange {
                 segments,
                 prefix,

packages/durable-streams-rust/src/engine_raw.rs

@@ -174,13 +174,20 @@ pub async fn serve(store: Arc<Store>, listener: TcpListener) {
         };
         let store = store.clone();
         tokio::spawn(async move {
-            let _permit = permit; // released when the connection ends
-            let _ = conn_loop(store, stream).await;
+            // The permit is moved INTO conn_loop so it can hand it on to a
+            // detached SSE streaming task (keeping the connection counted while
+            // the big conn_loop future is freed); otherwise it is released when
+            // conn_loop returns.
+            let _ = conn_loop(store, stream, permit).await;
         });
     }
 }
 
-async fn conn_loop(store: Arc<Store>, mut stream: TcpStream) -> std::io::Result<()> {
+async fn conn_loop(
+    store: Arc<Store>,
+    mut stream: TcpStream,
+    permit: tokio::sync::OwnedSemaphorePermit,
+) -> std::io::Result<()> {
     const BAD_REQUEST: &[u8] =
         b"HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n";
     const TOO_LARGE: &[u8] =
@@ -324,6 +331,49 @@ async fn conn_loop(store: Arc<Store>, mut stream: TcpStream) -> std::io::Result<
             body,
         };
         let resp = handlers::handle(store.clone(), req).await;
+        // SSE: hand the connection off to a DETACHED, minimal streaming task and
+        // return. An SSE subscriber parks for up to SSE_MAX_DURATION; driving it
+        // inline would keep this whole `conn_loop` future (sized to the largest
+        // request handler) resident the entire time — the dominant per-subscriber
+        // cost. The hand-off task's future holds only what streaming needs (the
+        // socket, the event source, a small frame buffer), so an idle subscriber's
+        // resident footprint collapses to ~that, mirroring a per-subscriber cursor
+        // rather than a parked request state machine. The permit moves with it so
+        // the connection stays counted; the stream closes when it ends (the client
+        // reconnects from its last offset — SSE is already capped + reconnecting).
+        // SSE live-tail fast path (Linux): a reactor-eligible source hands its
+        // socket to the epoll reactor and frees this connection task's future
+        // entirely — the dominant per-subscriber cost. HEAD requests and cold
+        // catch-up fall through to the inline hand-off below.
+        #[cfg(target_os = "linux")]
+        if !is_head {
+            if let Body::Sse(ref src) = resp.body {
+                if let Some(reg) = src.reactor_reg() {
+                    let mut head: Vec<u8> = Vec::with_capacity(256);
+                    http1::write_head(&mut head, resp.status, &resp.headers, None, keep_alive);
+                    crate::sse_reactor::register(stream, head, reg, permit);
+                    return Ok(());
+                }
+            }
+        }
+        if matches!(resp.body, Body::Sse(_)) {
+            tokio::spawn(async move {
+                let _permit = permit; // keep the connection counted until it ends
+                // Pass the request's `keep_alive` (true for a normal SSE GET) so
+                // the framing matches the inline path: `handle_sse` already sets
+                // `Connection: keep-alive`, and `write_head` must not also emit a
+                // `Connection: close`. The socket is closed when this task ends
+                // (it never loops back), so the stream still terminates cleanly.
+                let _ = write_response(&mut stream, resp, is_head, keep_alive).await;
+            });
+            return Ok(());
+        }
+        // A cold-tier `Body::Channel` stream also parks; release the 4 KiB read
+        // buffer first so it isn't pinned for the response (it re-grows on the
+        // next keep-alive request). Safe only with no pipelined bytes buffered.
+        if buf.is_empty() && matches!(resp.body, Body::Channel(_)) {
+            buf = BytesMut::new();
+        }
         write_response(&mut stream, resp, is_head, keep_alive).await?;
         if !keep_alive {
             return Ok(());
@@ -374,6 +424,23 @@ async fn write_response(
                 stream.write_all(suffix).await?;
             }
         }
+        Body::Sse(mut src) => {
+            // Pull-based SSE driven INLINE on this connection task — no spawned
+            // producer task, no mpsc channel (those cost ~4–5 KiB of resident
+            // heap per subscriber, the source of per-subscriber fan-out memory).
+            stream.write_all(&head).await?;
+            debug_assert!(body_len.is_none());
+            // SSE events are small; `frame_chunk` grows this on demand.
+            let mut frame: Vec<u8> = Vec::with_capacity(512);
+            while let Some(b) = src.next_chunk().await {
+                frame.clear();
+                http1::frame_chunk(&mut frame, &b);
+                stream.write_all(&frame).await?;
+            }
+            // SSE is infallible (the client reconnects from its last offset), so
+            // always terminate cleanly.
+            stream.write_all(b"0\r\n\r\n").await?;
+        }
         Body::Channel(crate::api::StreamBody { mut rx, failed }) => {
             stream.write_all(&head).await?;
             debug_assert!(body_len.is_none());