diff --git a/.changeset/long-poll-reactor.md b/.changeset/long-poll-reactor.md new file mode 100644 index 0000000000..fe499a7d07 --- /dev/null +++ b/.changeset/long-poll-reactor.md @@ -0,0 +1,5 @@ +--- +'@electric-ax/durable-streams-server-rust': patch +--- + +Serve caught-up long-poll waits from the epoll reactor too, sharing the machinery with SSE: a blocked poll hands its socket to the reactor (freeing the connection task's future) and is handed back for keep-alive after the response. Linux only; other platforms keep the inline wait. diff --git a/packages/durable-streams-rust/src/api.rs b/packages/durable-streams-rust/src/api.rs index 79f62cce51..568533a780 100644 --- a/packages/durable-streams-rust/src/api.rs +++ b/packages/durable-streams-rust/src/api.rs @@ -48,6 +48,25 @@ pub struct SseReg { pub client_cursor: Option, } +/// Caught-up long-poll registration: everything the reactor needs to wait at the +/// tail and write the single long-poll response once data arrives (or on close / +/// timeout). Constructed only on Linux, by `handle_long_poll` when the poll would +/// block on a reactor-eligible stream (root, no tiering); see `Body::LongPollWait`. +#[cfg(target_os = "linux")] +pub struct LongPollReg { + pub st: Arc, + /// Offset the poll is caught up at: the response delivers `from..tail` once + /// the tail advances past it. + pub from: u64, + pub client_cursor: Option, + /// Honour HTTP keep-alive: when set, the reactor hands the socket back to the + /// async connection loop after writing the response instead of closing it, so + /// a polling client's next request reuses the connection (no per-poll churn). + /// Filled in by the engine at hand-off — the handler can't see the + /// connection's keep-alive state. + pub keep_alive: bool, +} + /// A streamed (chunked) response body plus an abort signal. /// /// `rx` carries the body frames. `failed` is set by the producer if it ends the @@ -128,6 +147,13 @@ pub enum Body { /// Pull-based streaming body driven inline on the connection task (SSE). /// No spawned task, no channel — see `EventSource`. Sse(Box), + /// Linux: a caught-up long-poll that would block at the tail. The engine + /// hands the socket to the epoll reactor (freeing this connection task's + /// future for the up-to-30s wait, the dominant per-poll cost) instead of + /// awaiting inline; the reactor writes the single response when data arrives + /// or on close/timeout. Never serialized by `write_response`. + #[cfg(target_os = "linux")] + LongPollWait(LongPollReg), /// Byte ranges of data files plus optional framing (JSON `[` / `]`). /// Total body length is prefix + Σ segment.len + suffix. FileRange { @@ -150,6 +176,8 @@ impl Body { Body::Full(b) => Some(b.len() as u64), Body::Channel(_) => None, Body::Sse(_) => None, + #[cfg(target_os = "linux")] + Body::LongPollWait(_) => None, Body::FileRange { segments, prefix, diff --git a/packages/durable-streams-rust/src/engine_raw.rs b/packages/durable-streams-rust/src/engine_raw.rs index 1e7558ea05..fbaee02404 100644 --- a/packages/durable-streams-rust/src/engine_raw.rs +++ b/packages/durable-streams-rust/src/engine_raw.rs @@ -114,7 +114,36 @@ pub async fn drain(grace: std::time::Duration) { let _ = tokio::time::timeout(grace, sema.acquire_many(MAX_CONNECTIONS as u32)).await; } +/// Resume an async connection loop on a socket the long-poll reactor handed back +/// after writing a keep-alive response. Called from a reactor thread (NOT a tokio +/// worker), so it spawns onto the runtime handle captured in `serve`. `leftover` +/// is any data the client pipelined behind its poll. On any failure (no runtime — +/// only in unit tests — or the fd can't be adopted) everything is dropped, which +/// closes the socket and releases the permit. +#[cfg(target_os = "linux")] +pub(crate) fn resume_after_longpoll( + store: Arc, + std_stream: std::net::TcpStream, + leftover: BytesMut, + permit: tokio::sync::OwnedSemaphorePermit, +) { + let Some(handle) = CONN_RUNTIME.get() else { return }; + handle.spawn(async move { + if let Ok(stream) = TcpStream::from_std(std_stream) { + let _ = conn_loop(store, stream, leftover, permit).await; + } + }); +} + +/// Runtime handle captured at startup so the long-poll reactor (running on its +/// own non-tokio threads) can hand a socket back to the async connection loop +/// for keep-alive. See `resume_after_longpoll`. +#[cfg(target_os = "linux")] +static CONN_RUNTIME: std::sync::OnceLock = std::sync::OnceLock::new(); + pub async fn serve(store: Arc, listener: TcpListener) { + #[cfg(target_os = "linux")] + let _ = CONN_RUNTIME.set(tokio::runtime::Handle::current()); let conns = conn_limiter().clone(); loop { let (stream, _) = match listener.accept().await { @@ -178,7 +207,7 @@ pub async fn serve(store: Arc, listener: TcpListener) { // 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; + let _ = conn_loop(store, stream, BytesMut::with_capacity(4 * 1024), permit).await; }); } } @@ -186,18 +215,19 @@ pub async fn serve(store: Arc, listener: TcpListener) { async fn conn_loop( store: Arc, mut stream: TcpStream, + mut buf: BytesMut, 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] = b"HTTP/1.1 413 Payload Too Large\r\ncontent-length: 0\r\nconnection: close\r\n\r\n"; - // Initial capacity for the per-connection read buffer. Small on purpose: - // it persists for the whole (possibly long-lived, idle) connection, and - // `read_buf` grows it on demand, so a request head/body larger than this - // still parses — it just triggers a reserve. 4 KiB covers a typical request - // head in one read while keeping idle SSE subscribers cheap. - let mut buf = BytesMut::with_capacity(4 * 1024); + // The per-connection read buffer is passed in (a fresh 4 KiB for a new + // connection; for a connection handed back from the long-poll reactor after + // keep-alive, any bytes the client pipelined behind its poll). Small on + // purpose: it persists for the whole (possibly long-lived, idle) connection, + // and `read_buf` grows it on demand, so a larger request head/body still + // parses — it just triggers a reserve. loop { // ---- read request head ---- let head = match http1::read_head(&mut stream, &mut buf).await? { @@ -331,6 +361,25 @@ async fn conn_loop( body, }; let resp = handlers::handle(store.clone(), req).await; + // Long-poll (Linux): a caught-up poll on a reactor-eligible stream comes + // back as `LongPollWait` instead of having blocked inline. Hand the socket + // to the reactor — which parks for the wait without holding this conn_loop + // future, writes the single response, and (for keep-alive) hands the socket + // back so the next poll reuses the connection. `buf` carries any bytes the + // client pipelined behind the poll, fed back on resume. + #[cfg(target_os = "linux")] + let mut resp = resp; + #[cfg(target_os = "linux")] + if !is_head && matches!(resp.body, Body::LongPollWait(_)) { + let Body::LongPollWait(mut reg) = + std::mem::replace(&mut resp.body, Body::Empty) + else { + unreachable!() + }; + reg.keep_alive = keep_alive; + crate::reactor::register_long_poll(stream, reg, buf, store, permit); + return Ok(()); + } // 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 @@ -351,7 +400,7 @@ async fn conn_loop( if let Some(reg) = src.reactor_reg() { let mut head: Vec = Vec::with_capacity(256); http1::write_head(&mut head, resp.status, &resp.headers, None, keep_alive); - crate::sse_reactor::register(stream, head, reg, permit); + crate::reactor::register(stream, head, reg, permit); return Ok(()); } } @@ -398,6 +447,10 @@ async fn write_response( } match resp.body { + // A caught-up long-poll on Linux is intercepted in `conn_loop` and handed + // to the reactor before reaching here, so this is never serialized inline. + #[cfg(target_os = "linux")] + Body::LongPollWait(_) => unreachable!("LongPollWait is handed to the reactor"), Body::Empty => { stream.write_all(&head).await?; } diff --git a/packages/durable-streams-rust/src/handlers.rs b/packages/durable-streams-rust/src/handlers.rs index 061e684f72..1dffd57bbc 100644 --- a/packages/durable-streams-rust/src/handlers.rs +++ b/packages/durable-streams-rust/src/handlers.rs @@ -103,7 +103,7 @@ pub(crate) mod test_support { } } -fn long_poll_timeout_dur() -> Duration { +pub(crate) fn long_poll_timeout_dur() -> Duration { Duration::from_millis(LONG_POLL_TIMEOUT_MS.load(std::sync::atomic::Ordering::Relaxed)) } @@ -776,7 +776,7 @@ fn publish_durable_tail(st: &StreamState, tail: u64, wire: &Bytes) { st.tail_tx.send_replace(Tail { bytes: tail, closed }); // Wake any reactor-served subscribers of this stream (no-op when none). #[cfg(target_os = "linux")] - crate::sse_reactor::wake_stream(st); + crate::reactor::wake_stream(st); } // ---------- POST (append) ---------- @@ -1155,7 +1155,7 @@ async fn handle_append_inner(store: Arc, req: Req, path: String) -> (Resp }; st.tail_tx.send_replace(Tail { bytes: tail, closed: true }); #[cfg(target_os = "linux")] - crate::sse_reactor::wake_stream(&st); + crate::reactor::wake_stream(&st); } else { st.schedule_meta_flush(); } @@ -1355,7 +1355,7 @@ where }; st.tail_tx.send_replace(Tail { bytes: tail, closed }); #[cfg(target_os = "linux")] - crate::sse_reactor::wake_stream(&st); + crate::reactor::wake_stream(&st); // Shared response builder: captures `st`, `producer` by ref. let make_ok = || { @@ -1768,6 +1768,26 @@ async fn handle_long_poll( return long_poll_close(t0.bytes, cursor); } + // Caught up at the live tail with the stream still open: this poll must wait. + // On Linux, hand the wait to the epoll reactor (which frees this connection + // task's future for the up-to-30s park — the dominant per-poll cost) when the + // stream is reactor-eligible; the reactor writes the single response on + // data/close/timeout and hands the socket back for keep-alive. Cold/forked/ + // tiered streams and non-Linux fall through to the inline wait below. + #[cfg(target_os = "linux")] + if reactor_eligible_long_poll(&st) { + // Status is a placeholder: the engine intercepts `LongPollWait` before + // serialization and the reactor builds the real response head itself. + let mut resp = Resp::new(200); + resp.body = Body::LongPollWait(crate::api::LongPollReg { + st, + from, + client_cursor, + keep_alive: false, + }); + return resp; + } + // Wait for new data / closure / timeout. let mut rx = st.tail_tx.subscribe(); let deadline = Instant::now() + long_poll_timeout_dur(); @@ -1806,8 +1826,22 @@ async fn long_poll_data( hot: bool, cache_hit: &mut bool, ) -> Resp { - let cursor = compute_cursor(client_cursor); let body = read_range_body(st, from, t.bytes, hot, "long-poll", cache_hit).await; + long_poll_data_resp(st, from, t, client_cursor, body) +} + +/// Build the `200` response for a long-poll that delivered `from..t.bytes`, given +/// an already-materialized body. Sync and free of any read, so it is shared by +/// the inline path (async body via `read_range_body`) and the reactor (sync body +/// materialized by hand) — both must emit byte-identical headers. +pub(crate) fn long_poll_data_resp( + st: &StreamState, + from: u64, + t: Tail, + client_cursor: Option, + body: Body, +) -> Resp { + let cursor = compute_cursor(client_cursor); let mut b = ResponseBuilder::new(200) .h("content-type", st.config.content_type.clone()) .h(H_NEXT_OFFSET, format_offset(t.bytes)) @@ -1821,7 +1855,26 @@ async fn long_poll_data( b.body(body) } -fn long_poll_close(tail: u64, cursor: u64) -> Resp { +/// Wrap the wire bytes of a delivered long-poll range (`from..tail`) as a +/// fixed-length body, matching `read_range_body`'s framing: JSON streams strip +/// the trailing record comma and wrap the records in `[ ]`; others pass through. +/// Used by the reactor, which materializes the bytes itself rather than producing +/// a `Body::FileRange`. +#[cfg(target_os = "linux")] +pub(crate) fn long_poll_full_body(is_json: bool, data: Bytes) -> Body { + if is_json { + let inner = &data[..data.len().saturating_sub(1)]; + let mut out = BytesMut::with_capacity(inner.len() + 2); + out.put_u8(b'['); + out.put_slice(inner); + out.put_u8(b']'); + Body::Full(out.freeze()) + } else { + Body::Full(data) + } +} + +pub(crate) fn long_poll_close(tail: u64, cursor: u64) -> Resp { ResponseBuilder::new(204) .h(H_NEXT_OFFSET, format_offset(tail)) .h(H_CURSOR, cursor.to_string()) @@ -1831,7 +1884,7 @@ fn long_poll_close(tail: u64, cursor: u64) -> Resp { .body(empty()) } -fn long_poll_timeout(tail: u64, cursor: u64, closed: bool) -> Resp { +pub(crate) fn long_poll_timeout(tail: u64, cursor: u64, closed: bool) -> Resp { let mut b = ResponseBuilder::new(204) .h(H_NEXT_OFFSET, format_offset(tail)) .h(H_CURSOR, cursor.to_string()) @@ -1843,6 +1896,18 @@ fn long_poll_timeout(tail: u64, cursor: u64, closed: bool) -> Resp { b.body(empty()) } +/// Whether a caught-up long-poll on `st` can wait in the epoll reactor. Mirrors +/// the SSE live-tail gate (`SseSource::reactor_reg`): the reactor serves the live +/// data file by `pread`, so a forked (`parent`) or tiered (`blobstore`) stream +/// stays on the inline path. No `file_base` check is needed — a caught-up poll's +/// `from == tail >= file_base`, and the bytes it waits for are appended to the +/// live file (a rare concurrent compaction past `from` is handled at produce time +/// by ending the connection so the client retries through the cold path). +#[cfg(target_os = "linux")] +fn reactor_eligible_long_poll(st: &StreamState) -> bool { + st.parent.is_none() && st.blobstore.is_none() +} + // ---------- SSE ---------- #[derive(Clone, Copy)] diff --git a/packages/durable-streams-rust/src/main.rs b/packages/durable-streams-rust/src/main.rs index aa4bc7b748..ef59a7035b 100644 --- a/packages/durable-streams-rust/src/main.rs +++ b/packages/durable-streams-rust/src/main.rs @@ -4,7 +4,7 @@ mod engine_raw; mod handlers; mod http1; #[cfg(target_os = "linux")] -mod sse_reactor; +mod reactor; mod store; mod telemetry; mod tier; @@ -353,7 +353,7 @@ fn main() { // Close reactor-served SSE subscribers first so their permits are // released and `drain` doesn't wait out the full grace period. #[cfg(target_os = "linux")] - sse_reactor::shutdown(); + reactor::shutdown(); engine_raw::drain(std::time::Duration::from_secs(25)).await; telemetry_guard.shutdown(); } diff --git a/packages/durable-streams-rust/src/sse_reactor.rs b/packages/durable-streams-rust/src/reactor.rs similarity index 50% rename from packages/durable-streams-rust/src/sse_reactor.rs rename to packages/durable-streams-rust/src/reactor.rs index 549fe27b6c..06bc39aaf5 100644 --- a/packages/durable-streams-rust/src/sse_reactor.rs +++ b/packages/durable-streams-rust/src/reactor.rs @@ -1,27 +1,38 @@ -//! Per-core epoll reactor for live-tail SSE subscribers (Linux only). +//! Per-core epoll reactor for live-tail readers — SSE subscribers and caught-up +//! long-polls (Linux only). //! -//! A live SSE subscriber otherwise costs a whole connection task future (sized to -//! the largest request handler) parked for up to `SSE_MAX_DURATION`. At -//! fan-out scale that per-connection future is the dominant resident cost. This -//! reactor instead holds each subscriber as a compact slab entry on a fixed pool -//! of `N = available_parallelism()` dedicated threads (one epoll instance each): -//! task count stays constant, and per-subscriber userspace collapses to the slab -//! entry plus the kernel socket. See -//! `docs/superpowers/specs/2026-06-29-sse-reactor-flat-userspace-design.md`. +//! A live reader otherwise costs a whole connection task future (sized to the +//! largest request handler) parked for the wait — up to `SSE_MAX_DURATION` for an +//! SSE subscriber, up to the long-poll timeout for a caught-up poll. At fan-out +//! scale that per-connection future is the dominant resident cost. This reactor +//! instead holds each reader as a compact slab entry on a fixed pool of +//! `N = available_parallelism()` dedicated threads (one epoll instance each): task +//! count stays constant, and per-reader userspace collapses to the slab entry plus +//! the kernel socket. //! -//! Only the live-tail case runs here (root stream, tiering off, start at/after -//! the live file base). Cold catch-up keeps the proven inline hand-off path in -//! `engine_raw`. +//! The two modes share all the machinery (slab, epoll arm/flush/backpressure, +//! cross-thread wake, lifetime sweep) and differ only in what they produce on a +//! wake and what they do when finished (`Kind`): +//! - **SSE** streams chunked `data`/`control` events for the connection's whole +//! lifetime, then closes (the client reconnects). +//! - **long-poll** writes one complete fixed-length response when the tail +//! advances (or on close/timeout), then hands the socket back to the async +//! connection loop so a keep-alive client's next poll reuses it. +//! +//! Only the live-tail case runs here (root stream, tiering off, start at/after the +//! live file base). Cold catch-up keeps the proven inline path in `engine_raw`. -use std::os::fd::{IntoRawFd, RawFd}; +use std::os::fd::{FromRawFd, IntoRawFd, RawFd}; use std::os::unix::fs::FileExt; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, Instant}; -use crate::api::SseReg; +use bytes::{Bytes, BytesMut}; + +use crate::api::{LongPollReg, Resp, SseReg}; use crate::handlers::SseEncoding; -use crate::store::{StreamState, StreamSubs, SubHandle}; +use crate::store::{Store, StreamState, StreamSubs, SubHandle, Tail}; /// Idle heartbeat interval and total subscriber lifetime — mirror the inline /// path (`handlers::SSE_KEEPALIVE` / `SSE_MAX_DURATION`) so client behaviour is @@ -54,41 +65,82 @@ struct Shard { wake: Mutex>, } -/// A subscriber handed from a connection task to the reactor. +/// A reader handed from a connection task to the reactor. struct Registration { fd: RawFd, st: Arc, - write_off: u64, - encoding: SseEncoding, - client_cursor: Option, - /// Raw HTTP response head (status line + headers), sent before the chunked - /// body. Becomes the initial `pending`. + /// Initial `pending` bytes: the raw chunked-response head for SSE (sent before + /// the first event), empty for long-poll (which builds its whole response, + /// head included, only once it has an outcome). head: Vec, permit: tokio::sync::OwnedSemaphorePermit, + kind: RegKind, } -/// Resident per-subscriber state. `pending`/`sent` are the only growable parts, -/// and only while a socket is backpressured; a caught-up subscriber carries an -/// empty buffer. +/// Mode-specific registration payload. The reactor thread turns this into a +/// `Kind` at `insert` (stamping per-reader timestamps then). +enum RegKind { + Sse { + write_off: u64, + encoding: SseEncoding, + client_cursor: Option, + }, + LongPoll { + from: u64, + client_cursor: Option, + keep_alive: bool, + /// Any bytes the client pipelined behind its poll; handed back to the async + /// loop with the socket so it re-parses them as the next request. + leftover: BytesMut, + store: Arc, + }, +} + +/// Resident per-reader state. `pending`/`sent` are the only growable parts, and +/// only while a socket is backpressured; an idle reader carries an empty buffer. struct Sub { fd: RawFd, st: Arc, + /// Framed bytes not yet written; `pending[sent..]` is the unwritten tail. + pending: Vec, + sent: usize, + /// Response fully queued; finish (close or hand-back) once `pending` drains. + done: bool, + epollout_armed: bool, + kind: Kind, + _permit: tokio::sync::OwnedSemaphorePermit, +} + +/// Per-reader mode + its state. Sized to the larger variant; the extra bytes over +/// a single-mode `Sub` are dwarfed by `pending` and the 64 KiB kernel socket. +enum Kind { + Sse(SseState), + LongPoll(LongPollState), +} + +struct SseState { /// Stream byte offset delivered (framed) so far. write_off: u64, /// Offset at registration — gates the one-shot initial caught-up control. start: u64, - /// Framed bytes not yet written; `pending[sent..]` is the unwritten tail. - pending: Vec, - sent: usize, encoding: SseEncoding, client_cursor: Option, registered_at: Instant, last_event: Instant, sent_initial: bool, - /// Terminating chunk queued; close once `pending` drains. - done: bool, - epollout_armed: bool, - _permit: tokio::sync::OwnedSemaphorePermit, +} + +struct LongPollState { + /// Offset the poll is caught up at; the response delivers `from..tail`. + from: u64, + client_cursor: Option, + keep_alive: bool, + /// When an idle poll returns the up-to-date `204` timeout. + deadline: Instant, + /// Bytes the client pipelined behind its poll; handed back with the socket so + /// the async loop re-parses them as the next request. + leftover: BytesMut, + store: Arc, } struct Slot { @@ -105,6 +157,56 @@ struct Slot { /// tokio task returns immediately afterward. Best-effort: a fd-extraction failure /// just drops the connection. pub fn register(stream: tokio::net::TcpStream, head: Vec, reg: SseReg, permit: tokio::sync::OwnedSemaphorePermit) { + enqueue( + stream, + head, + reg.st, + RegKind::Sse { + write_off: reg.start, + encoding: reg.encoding, + client_cursor: reg.client_cursor, + }, + permit, + ); +} + +/// Hand a caught-up long-poll socket to the reactor. The reactor waits at the +/// tail and writes the single response on data/close/timeout (no head is sent +/// up front — it depends on the outcome), then hands the socket back to the async +/// loop for keep-alive (or closes it). Takes ownership of the socket fd and the +/// connection-limiter permit; best-effort, like `register`. +pub fn register_long_poll( + stream: tokio::net::TcpStream, + reg: LongPollReg, + leftover: BytesMut, + store: Arc, + permit: tokio::sync::OwnedSemaphorePermit, +) { + enqueue( + stream, + Vec::new(), + reg.st, + RegKind::LongPoll { + from: reg.from, + client_cursor: reg.client_cursor, + keep_alive: reg.keep_alive, + leftover, + store, + }, + permit, + ); +} + +/// Shared hand-off: extract the raw fd, make it non-blocking, and queue it on a +/// shard. Best-effort — a fd-extraction failure or an in-progress shutdown drops +/// the connection (and releases the permit). +fn enqueue( + stream: tokio::net::TcpStream, + head: Vec, + st: Arc, + kind: RegKind, + permit: tokio::sync::OwnedSemaphorePermit, +) { let std_stream = match stream.into_std() { Ok(s) => s, Err(_) => return, @@ -124,12 +226,10 @@ pub fn register(stream: tokio::net::TcpStream, head: Vec, reg: SseReg, permi let shard = &pool[(fd as usize) % pool.len()]; shard.intake.lock().unwrap().push(Registration { fd, - st: reg.st, - write_off: reg.start, - encoding: reg.encoding, - client_cursor: reg.client_cursor, + st, head, permit, + kind, }); signal(shard.eventfd); } @@ -140,8 +240,8 @@ pub fn wake_stream(st: &StreamState) { let mut to_signal: Vec = Vec::new(); { // Check for subscribers BEFORE touching the pool: a stream (or whole - // server) with no SSE subscribers never spawns a reactor thread. - let guard = st.sse_subs.lock().unwrap(); + // server) with no live readers never spawns a reactor thread. + let guard = st.reactor_subs.lock().unwrap(); let Some(list) = guard.as_ref() else { return }; // Subscribers exist only because `register` ran, so the pool is live. let pool = pool(); @@ -185,9 +285,9 @@ fn pool() -> &'static Vec> { }); let me = shard.clone(); std::thread::Builder::new() - .name(format!("sse-reactor-{i}")) + .name(format!("reactor-{i}")) .spawn(move || Reactor::new(me, i as u16).run()) - .expect("spawn sse reactor thread"); + .expect("spawn reactor thread"); shards.push(shard); } shards @@ -300,25 +400,48 @@ impl Reactor { let gen = self.slab[key as usize].gen; let now = Instant::now(); let st = reg.st.clone(); + let kind = match reg.kind { + RegKind::Sse { + write_off, + encoding, + client_cursor, + } => Kind::Sse(SseState { + write_off, + start: write_off, + encoding, + client_cursor, + registered_at: now, + last_event: now, + sent_initial: false, + }), + RegKind::LongPoll { + from, + client_cursor, + keep_alive, + leftover, + store, + } => Kind::LongPoll(LongPollState { + from, + client_cursor, + keep_alive, + deadline: now + crate::handlers::long_poll_timeout_dur(), + leftover, + store, + }), + }; self.slab[key as usize].sub = Some(Sub { fd: reg.fd, st: reg.st, - write_off: reg.write_off, - start: reg.write_off, pending: reg.head, sent: 0, - encoding: reg.encoding, - client_cursor: reg.client_cursor, - registered_at: now, - last_event: now, - sent_initial: false, done: false, epollout_armed: false, + kind, _permit: reg.permit, }); // Link onto the stream so the append path can find and wake this sub. { - let mut g = st.sse_subs.lock().unwrap(); + let mut g = st.reactor_subs.lock().unwrap(); g.get_or_insert_with(|| Box::new(StreamSubs { subs: Vec::new() })) .subs .push(SubHandle { @@ -339,99 +462,175 @@ impl Reactor { self.flush(key); } + /// Produce the reader's next bytes into `pending`, dispatched by mode. Called + /// on registration and on every wake; the caller flushes afterwards. + fn produce(&mut self, key: u32) { + match self.slab.get(key as usize).and_then(|s| s.sub.as_ref()) { + Some(sub) if sub.done => {} + Some(sub) => match &sub.kind { + Kind::Sse(_) => self.sse_produce(key), + Kind::LongPoll(_) => self.longpoll_produce(key), + }, + None => {} + } + } + /// Append the subscriber's next SSE frame(s) — catch-up data + control, a /// final close control, or the one-shot initial caught-up control — to /// `pending`. A faithful synchronous port of `handlers::SseSource::next` /// (minus the async idle wait, which the keepalive sweep covers). - fn produce(&mut self, key: u32) { + fn sse_produce(&mut self, key: u32) { let sub = match self.slab[key as usize].sub.as_mut() { Some(s) if !s.done => s, _ => return, }; + let Kind::Sse(ss) = &mut sub.kind else { return }; loop { let (file, file_base, tail, closed) = { let s = sub.st.shared.read().unwrap(); (s.file.clone(), s.file_base, s.durable_tail, s.closed_durable) }; - if tail > sub.write_off { - if sub.write_off < file_base { + if tail > ss.write_off { + if ss.write_off < file_base { // The needed bytes were compacted out of the live file; end the // stream (the client reconnects, routing through the cold path). sub.done = true; frame_terminator(&mut sub.pending); return; } - let want = (tail - sub.write_off) as usize; + let want = (tail - ss.write_off) as usize; let mut data = vec![0u8; want]; - if pread_exact(&file, sub.write_off - file_base, &mut data).is_err() { + if pread_exact(&file, ss.write_off - file_base, &mut data).is_err() { sub.done = true; frame_terminator(&mut sub.pending); return; } let mut ev = String::new(); - crate::handlers::sse_encode_data(&mut ev, &data, sub.encoding); - sub.write_off = tail; + crate::handlers::sse_encode_data(&mut ev, &data, ss.encoding); + ss.write_off = tail; let cur = sub.st.tail(); - let up_to_date = sub.write_off >= cur.bytes; - let closed_now = closed && sub.write_off >= tail; + let up_to_date = ss.write_off >= cur.bytes; + let closed_now = closed && ss.write_off >= tail; crate::handlers::sse_control_event( &mut ev, - sub.write_off, - crate::store::compute_cursor(sub.client_cursor), + ss.write_off, + crate::store::compute_cursor(ss.client_cursor), up_to_date, closed_now, ); push_frame(&mut sub.pending, &mut sub.sent, ev.as_bytes()); - sub.last_event = Instant::now(); + ss.last_event = Instant::now(); if closed_now { sub.done = true; frame_terminator(&mut sub.pending); return; } - if backlog(sub) > PENDING_CAP { + if sub.pending.len() - sub.sent > PENDING_CAP { sub.done = true; return; } continue; } - if closed && sub.write_off >= tail { + if closed && ss.write_off >= tail { let mut ev = String::new(); crate::handlers::sse_control_event( &mut ev, - sub.write_off, - crate::store::compute_cursor(sub.client_cursor), + ss.write_off, + crate::store::compute_cursor(ss.client_cursor), true, true, ); push_frame(&mut sub.pending, &mut sub.sent, ev.as_bytes()); - sub.last_event = Instant::now(); + ss.last_event = Instant::now(); sub.done = true; frame_terminator(&mut sub.pending); return; } // Caught up at the live tail with nothing pending: emit the one-shot // initial control (parity with the inline path) then go idle. - if !sub.sent_initial - && sub.write_off == sub.start - && tail == sub.start - && !closed - { + if !ss.sent_initial && ss.write_off == ss.start && tail == ss.start && !closed { let mut ev = String::new(); crate::handlers::sse_control_event( &mut ev, - sub.write_off, - crate::store::compute_cursor(sub.client_cursor), + ss.write_off, + crate::store::compute_cursor(ss.client_cursor), true, false, ); push_frame(&mut sub.pending, &mut sub.sent, ev.as_bytes()); - sub.sent_initial = true; - sub.last_event = Instant::now(); + ss.sent_initial = true; + ss.last_event = Instant::now(); } return; } } + /// Build the single long-poll response into `pending` once the wait resolves: + /// `200` with `from..tail` when the tail advanced, `204` stream-closed when the + /// writer closed at the tail. Stays idle (queues nothing) while still caught up + /// and open — the deadline timeout is handled by `longpoll_tick`. Mirrors + /// `handlers::handle_long_poll`'s post-wait branches via the shared builders. + fn longpoll_produce(&mut self, key: u32) { + let (st, from, client_cursor, keep_alive) = { + let sub = match self.slab[key as usize].sub.as_ref() { + Some(s) if !s.done => s, + _ => return, + }; + let Kind::LongPoll(lp) = &sub.kind else { return }; + (sub.st.clone(), lp.from, lp.client_cursor, lp.keep_alive) + }; + let (file, file_base, tail, closed) = { + let s = st.shared.read().unwrap(); + (s.file.clone(), s.file_base, s.durable_tail, s.closed_durable) + }; + if tail > from { + if from < file_base { + // Concurrent compaction dropped the needed bytes from the live + // file. Close so the client retries and is served the gap through + // the cold catch-up path. + self.close(key); + return; + } + let data = match read_hot_range(&st, &file, file_base, from, tail) { + Some(d) => d, + None => { + self.close(key); + return; + } + }; + let body = crate::handlers::long_poll_full_body(st.is_json, data); + let resp = crate::handlers::long_poll_data_resp( + &st, + from, + Tail { bytes: tail, closed }, + client_cursor, + body, + ); + self.finalize_longpoll(key, resp, keep_alive); + } else if closed { + let resp = + crate::handlers::long_poll_close(tail, crate::store::compute_cursor(client_cursor)); + self.finalize_longpoll(key, resp, keep_alive); + } + } + + /// Queue a fully-built long-poll `Resp` as this reader's complete response and + /// mark it done; the caller flushes (then `finish` hands the socket back or + /// closes it). + fn finalize_longpoll(&mut self, key: u32, resp: Resp, keep_alive: bool) { + let bytes = serialize_resp(resp, keep_alive); + if let Some(sub) = self.slab[key as usize].sub.as_mut() { + // `pending` is normally empty here (long-poll queues nothing before its + // outcome); compact any sent prefix defensively before appending. + if sub.sent > 0 { + sub.pending.drain(0..sub.sent); + sub.sent = 0; + } + sub.pending.extend_from_slice(&bytes); + sub.done = true; + } + } + /// Write as much of `pending` as the socket accepts. Arms EPOLLOUT on /// backpressure, disarms it once drained, and closes the sub on a fatal write /// error or after the terminating chunk flushes. @@ -480,7 +679,7 @@ impl Reactor { self.disarm_epollout(key); } if done { - self.close(key); + self.finish(key); } } @@ -516,43 +715,95 @@ impl Reactor { } } - /// Keepalive heartbeat + total-lifetime cap sweep, driven by the epoll_wait - /// timeout. Linear over the shard's subs; fine at the per-core sub counts this - /// targets. + /// Per-mode time-driven sweep, driven by the epoll_wait timeout: SSE keepalive + /// and lifetime cap, long-poll idle timeout. Linear over the shard's subs; fine + /// at the per-core sub counts this targets. fn tick(&mut self) { let now = Instant::now(); for key in 0..self.slab.len() as u32 { - let Some(sub) = self.slab[key as usize].sub.as_ref() else { - continue; - }; - if now.duration_since(sub.registered_at) >= MAX_DURATION { - // Lifetime cap: end cleanly; the client reconnects from its offset. - let sub = self.slab[key as usize].sub.as_mut().unwrap(); - if !sub.done { - sub.done = true; - frame_terminator(&mut sub.pending); - } - self.flush(key); - continue; + match self.slab[key as usize].sub.as_ref().map(|s| &s.kind) { + Some(Kind::Sse(_)) => self.sse_tick(key, now), + Some(Kind::LongPoll(_)) => self.longpoll_tick(key, now), + None => {} } - if backlog(sub) == 0 - && !sub.done - && now.duration_since(sub.last_event) >= KEEPALIVE - { - let sub = self.slab[key as usize].sub.as_mut().unwrap(); - let mut ev = String::new(); - crate::handlers::sse_control_event( - &mut ev, - sub.write_off, - crate::store::compute_cursor(sub.client_cursor), - true, - false, - ); - push_frame(&mut sub.pending, &mut sub.sent, ev.as_bytes()); - sub.last_event = now; - self.flush(key); + } + } + + /// SSE keepalive heartbeat + total-lifetime cap for one subscriber. + fn sse_tick(&mut self, key: u32, now: Instant) { + let (registered_at, last_event) = match self.slab[key as usize].sub.as_ref() { + Some(sub) => match &sub.kind { + Kind::Sse(ss) => (ss.registered_at, ss.last_event), + _ => return, + }, + None => return, + }; + if now.duration_since(registered_at) >= MAX_DURATION { + // Lifetime cap: end cleanly; the client reconnects from its offset. + let sub = self.slab[key as usize].sub.as_mut().unwrap(); + if !sub.done { + sub.done = true; + frame_terminator(&mut sub.pending); } + self.flush(key); + return; + } + let idle = { + let sub = self.slab[key as usize].sub.as_ref().unwrap(); + sub.pending.len() == sub.sent && !sub.done && now.duration_since(last_event) >= KEEPALIVE + }; + if idle { + let sub = self.slab[key as usize].sub.as_mut().unwrap(); + let Kind::Sse(ss) = &mut sub.kind else { return }; + let mut ev = String::new(); + crate::handlers::sse_control_event( + &mut ev, + ss.write_off, + crate::store::compute_cursor(ss.client_cursor), + true, + false, + ); + ss.last_event = now; + push_frame(&mut sub.pending, &mut sub.sent, ev.as_bytes()); + self.flush(key); + } + } + + /// Long-poll idle timeout: once the deadline passes, deliver any data/close + /// that arrived right at the wire (an append whose wake hasn't been drained + /// yet), otherwise write the up-to-date `204` timeout. Idle ticks before the + /// deadline return immediately. + fn longpoll_tick(&mut self, key: u32, now: Instant) { + let deadline = match self.slab[key as usize].sub.as_ref() { + Some(sub) if !sub.done => match &sub.kind { + Kind::LongPoll(lp) => lp.deadline, + _ => return, + }, + _ => return, + }; + if now < deadline { + return; } + self.longpoll_produce(key); + let needs_timeout = matches!( + self.slab[key as usize].sub.as_ref(), + Some(sub) if !sub.done + ); + if needs_timeout { + let (st, client_cursor, keep_alive) = { + let sub = self.slab[key as usize].sub.as_ref().unwrap(); + let Kind::LongPoll(lp) = &sub.kind else { return }; + (sub.st.clone(), lp.client_cursor, lp.keep_alive) + }; + let t = st.tail(); + let resp = crate::handlers::long_poll_timeout( + t.bytes, + crate::store::compute_cursor(client_cursor), + t.closed, + ); + self.finalize_longpoll(key, resp, keep_alive); + } + self.flush(key); } /// Tear down a subscriber: drop it from epoll, close the socket, unlink it @@ -568,7 +819,52 @@ impl Reactor { } let shard_idx = self.shard_idx; { - let mut g = sub.st.sse_subs.lock().unwrap(); + let mut g = sub.st.reactor_subs.lock().unwrap(); + if let Some(list) = g.as_mut() { + list.subs.retain(|h| !(h.shard == shard_idx && h.key == key)); + if list.subs.is_empty() { + *g = None; + } + } + } + self.slab[key as usize].gen = self.slab[key as usize].gen.wrapping_add(1); + self.free.push(key); + } + + /// A reader's response fully flushed: a keep-alive long-poll hands its socket + /// back to the async connection loop so the client's next request is read and + /// dispatched there (re-registering with the reactor if it is another caught-up + /// poll); otherwise (SSE, or a `Connection: close` poll) the socket is torn down. + fn finish(&mut self, key: u32) { + let keep_alive = matches!( + self.slab[key as usize].sub.as_ref(), + Some(Sub { kind: Kind::LongPoll(lp), .. }) if lp.keep_alive + ); + if keep_alive { + self.handback(key); + } else { + self.close(key); + } + } + + /// Hand a keep-alive long-poll socket back to the async connection loop once its + /// response has flushed, so the client's next request is read and dispatched + /// there. Drop it from epoll WITHOUT closing the fd, unlink + free the slot (as + /// `close` does), then adopt the live fd as a std stream and resume `conn_loop`, + /// seeding it with any bytes the client pipelined behind its poll so they are + /// re-parsed. The permit moves with it, so the connection stays counted across + /// the hand-back. + fn handback(&mut self, key: u32) { + let Some(sub) = self.slab[key as usize].sub.take() else { + return; + }; + let fd = sub.fd; + unsafe { + libc::epoll_ctl(self.epfd, libc::EPOLL_CTL_DEL, fd, std::ptr::null_mut()); + } + let shard_idx = self.shard_idx; + { + let mut g = sub.st.reactor_subs.lock().unwrap(); if let Some(list) = g.as_mut() { list.subs.retain(|h| !(h.shard == shard_idx && h.key == key)); if list.subs.is_empty() { @@ -578,6 +874,15 @@ impl Reactor { } self.slab[key as usize].gen = self.slab[key as usize].gen.wrapping_add(1); self.free.push(key); + let Kind::LongPoll(lp) = sub.kind else { + // finish only hands back long-polls; close the fd if this is reached. + unsafe { + libc::close(fd); + } + return; + }; + let std_stream = unsafe { std::net::TcpStream::from_raw_fd(fd) }; + crate::engine_raw::resume_after_longpoll(lp.store, std_stream, lp.leftover, sub._permit); } fn close_all(&mut self) { @@ -597,8 +902,40 @@ impl Reactor { } } -fn backlog(sub: &Sub) -> usize { - sub.pending.len() - sub.sent +/// Serialize a complete long-poll response (head + optional fixed-length body) to +/// raw socket bytes. Byte-identical to the async engine's `write_response` for the +/// `Empty`/`Full` bodies that long-poll produces (`204` head-only, `200` with +/// `content-length`); never chunked, since the body is fully materialized. +fn serialize_resp(resp: Resp, keep_alive: bool) -> Vec { + let body_len = resp.body.len(); + let mut out = Vec::with_capacity(256); + crate::http1::write_head(&mut out, resp.status, &resp.headers, body_len, keep_alive); + if !crate::http1::status_has_no_body(resp.status) { + if let crate::api::Body::Full(b) = resp.body { + out.extend_from_slice(&b); + } + } + out +} + +/// Materialize the wire bytes of a delivered long-poll range from the live data +/// file: the resident tail chunk when it covers the range (shared across readers, +/// no syscall — matching `read_range_body`'s fast path), else a positioned read. +/// `None` on a read error, so the caller closes and the client retries. +fn read_hot_range( + st: &StreamState, + file: &std::fs::File, + file_base: u64, + from: u64, + tail: u64, +) -> Option { + if let Some(b) = st.tail_chunk_slice(from, tail) { + return Some(b); + } + let want = (tail - from) as usize; + let mut data = vec![0u8; want]; + pread_exact(file, from - file_base, &mut data).ok()?; + Some(Bytes::from(data)) } /// Append `frame` as a chunked-transfer frame to `pending`, compacting already @@ -684,4 +1021,53 @@ mod tests { frame_terminator(&mut p); assert_eq!(p, b"0\r\n\r\n"); } + + // A long-poll timeout serializes as a bodyless 204, keep-alive honoured. + #[test] + fn longpoll_timeout_serializes_204_keepalive() { + let resp = crate::handlers::long_poll_timeout(42, 7, false); + let bytes = serialize_resp(resp, true); + let s = String::from_utf8(bytes).unwrap(); + assert!(s.starts_with("HTTP/1.1 204 No Content\r\n"), "{s:?}"); + assert!( + s.contains(&format!("stream-next-offset: {}\r\n", crate::store::format_offset(42))), + "{s:?}" + ); + assert!(s.contains("stream-up-to-date: true\r\n"), "{s:?}"); + // 204 carries no body and no content-length; keep-alive omits `close`. + assert!(!s.contains("content-length"), "{s:?}"); + assert!(!s.contains("connection: close"), "{s:?}"); + assert!(s.ends_with("\r\n\r\n")); + } + + // A non-keep-alive outcome must add `Connection: close`. + #[test] + fn longpoll_close_serializes_connection_close() { + let resp = crate::handlers::long_poll_close(99, 7); + let bytes = serialize_resp(resp, false); + let s = String::from_utf8(bytes).unwrap(); + assert!(s.starts_with("HTTP/1.1 204 No Content\r\n"), "{s:?}"); + assert!(s.contains("stream-closed: true\r\n"), "{s:?}"); + assert!(s.contains("connection: close\r\n"), "{s:?}"); + } + + // JSON streams strip the trailing record comma and wrap in `[ ]`; others pass + // the wire bytes through unchanged. Matches `read_range_body`'s framing. + #[test] + fn longpoll_body_json_wraps_and_strips_comma() { + let data = Bytes::from_static(b"{\"a\":1},{\"b\":2},"); + match crate::handlers::long_poll_full_body(true, data) { + crate::api::Body::Full(b) => assert_eq!(&b[..], b"[{\"a\":1},{\"b\":2}]"), + _ => panic!("expected Full body"), + } + } + + #[test] + fn longpoll_body_non_json_passthrough() { + let data = Bytes::from_static(b"\x00\x01\x02raw"); + match crate::handlers::long_poll_full_body(false, data.clone()) { + crate::api::Body::Full(b) => assert_eq!(b, data), + _ => panic!("expected Full body"), + } + } } diff --git a/packages/durable-streams-rust/src/store.rs b/packages/durable-streams-rust/src/store.rs index c3ba17f5b8..73ff63b02f 100644 --- a/packages/durable-streams-rust/src/store.rs +++ b/packages/durable-streams-rust/src/store.rs @@ -182,9 +182,9 @@ pub struct StreamState { /// Reactor-served SSE subscribers of this stream (Linux). `None` while the /// stream has none — the common case — so idle streams cost only the lock + /// a null pointer; the list (and its allocation) exist only while subscribers - /// are attached. See sse_reactor.rs. + /// are attached. See reactor.rs. #[cfg(target_os = "linux")] - pub sse_subs: StdMutex>>, + pub reactor_subs: StdMutex>>, } /// Reactor subscriber list for one stream — populated only while subscribers are @@ -601,7 +601,7 @@ impl Store { // cell starts clear; the next meta write persists the cleared marker. compaction: StdMutex::new(None), #[cfg(target_os = "linux")] - sse_subs: StdMutex::new(None), + reactor_subs: StdMutex::new(None), config: StreamConfig { content_type: meta.content_type.clone(), ttl_seconds: meta.ttl_seconds, @@ -764,7 +764,7 @@ impl Store { blobstore: self.blobstore.clone(), compaction: StdMutex::new(None), #[cfg(target_os = "linux")] - sse_subs: StdMutex::new(None), + reactor_subs: StdMutex::new(None), config, }); match self.streams.entry(path.to_string()) {