From af05b35698ce44c96bbf839bfe384ee126c8e971 Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Mon, 14 Sep 2026 14:40:44 +0100 Subject: [PATCH 1/2] Write SSE frames to the socket as they are pushed Queuing an entire publication burst before writing could close subscribers at the 64 KiB cap while their sockets still had room. ChunkedResponse::deliver attempts to drain output after each accepted push and returns the required readiness interests. Its drain_into method also serves the readiness handler, preserving FIFO order and send-deadline accounting. The beacon API applies registration changes and closes connections on cap, write or registration errors. When a write drains an existing backlog, restore READABLE alone. This also handles publication before the response head's writable event. Subscribers whose buffers were already empty keep their registration. The cap remains 64 KiB and is checked before each push. Keep PENDING_MAX private to httpcore. Test delivery through impl Write with a scripted writer, so burst survival does not depend on kernel buffer capacity or reader scheduling. The tests cover a 40-frame burst exceeding the cap, ordered output, and returned interests after draining the head or a blocked write. They also exercise partial writes, interruptions, stall-clock updates, cap errors, zero-length writes and write errors. The beacon API checks delivery and ordering over a Unix socket using a burst that fits entirely in the application buffer. Two Linux-only integration tests check epoll registration after draining an idle subscriber's frame or its unsent response head. These retain their Linux readiness assumptions without requiring capacity for a large socket burst. Formatting and all-feature Clippy passed. All 108 HTTP-core, 141 beacon API and 14 application-boundary tests passed. The workspace suite's only failure was finalized_state_loads rejecting an expired checkpoint. Assisted-by: Claude:claude-fable-5-1 Assisted-by: Codex:GPT-6 --- crates/beacon_api/src/server.rs | 145 +++++++++++---- crates/httpcore/src/chunked_response.rs | 223 +++++++++++++++++++++++- crates/httpcore/src/lib.rs | 2 +- docs/adr/0004-sync-materialized-api.md | 8 + 4 files changed, 345 insertions(+), 33 deletions(-) diff --git a/crates/beacon_api/src/server.rs b/crates/beacon_api/src/server.rs index 849d9df3..82cdc64b 100644 --- a/crates/beacon_api/src/server.rs +++ b/crates/beacon_api/src/server.rs @@ -9,8 +9,8 @@ use mio::{Events, Interest, Registry, Token, event::Event}; use silver_beacon_state_data::{BeaconStateReader, SpecConfig}; use silver_common::{Enr, Identify, Keypair}; use silver_httpcore::{ - AfterResponse, Bind, ChunkedResponse, Listener, ParsedRequest, ServerConnection, Stream, - TokenRange, + AfterResponse, Bind, ChunkedResponse, Closed, Listener, ParsedRequest, ServerConnection, + Stream, TokenRange, }; use crate::{ @@ -255,18 +255,7 @@ impl Subscription { return Ok(true); } - if event.is_writable() { - while !self.body.pending_write().is_empty() { - match stream.write(self.body.pending_write()) { - Ok(0) => { - return Err(io::Error::new(io::ErrorKind::WriteZero, "write returned 0")) - } - Ok(n) => self.body.commit_write(n, now), - Err(e) if would_block(&e) => return Ok(false), - Err(e) if interrupted(&e) => continue, - Err(e) => return Err(e), - } - } + if event.is_writable() && self.body.drain_into(stream, now)? { registry.reregister(stream, event.token(), Interest::READABLE)?; } @@ -435,26 +424,36 @@ impl BeaconApi { let Self { connections, registry, .. } = self; let mut pushed = false; connections.retain(|token, conn| { - let State::Subscription(subscription) = &mut conn.state else { return true }; + let Connection { stream, state: State::Subscription(subscription) } = conn else { + return true; + }; if !wants(subscription) { return true; } - if !subscription.body.push(chunk, now) { - tracing::warn!( - "beacon api subscriber would exceed send cap with {} bytes already pending, closing", - subscription.body.pending_write().len() - ); - let _ = registry.deregister(&mut conn.stream); - return false; - } - pushed = true; - let interest = Interest::READABLE | Interest::WRITABLE; - if let Err(e) = registry.reregister(&mut conn.stream, *token, interest) { - tracing::warn!("beacon api subscriber lost: {e}"); - let _ = registry.deregister(&mut conn.stream); - return false; + let outcome = subscription.body.deliver(stream, chunk, now).and_then(|interest| { + pushed = true; + match interest { + Some(interest) => { + registry.reregister(stream, *token, interest).map_err(Closed::Lost) + } + None => Ok(()), + } + }); + match outcome { + Ok(()) => true, + Err(Closed::AtCap { pending }) => { + tracing::warn!( + "beacon api subscriber would exceed send cap with {pending} bytes already pending, closing" + ); + let _ = registry.deregister(stream); + false + } + Err(Closed::Lost(e)) => { + tracing::warn!("beacon api subscriber lost: {e}"); + let _ = registry.deregister(stream); + false + } } - true }); pushed } @@ -1447,6 +1446,92 @@ mod tests { assert_same_bytes(&got, &expected); } + fn burst_frame(index: usize, len: usize) -> Vec { + let mut frame = format!("event: burst\ndata: {index:02}").into_bytes(); + frame.resize(len - 2, b'c'); + frame.extend_from_slice(b"\n\n"); + frame + } + + /// The burst fits in the application buffer even if the socket initially + /// accepts no bytes. Delivery therefore does not require a particular + /// kernel send-buffer capacity. + #[test] + fn a_burst_reaches_a_reading_subscriber_in_order() { + let dir = tempfile::tempdir().unwrap(); + let socket = dir.path().join("api.sock"); + let mut server = server_bound_to(&[Bind::Unix(socket.clone())], 64, LONG_TIMEOUT); + let mut client = connect_uds(&socket); + subscribe(&mut client, "block"); + pump_until(&mut server, "subscribed and head sent", |server| { + subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0 + }); + + let frames: Vec<_> = (0..24).map(|index| burst_frame(index, 2300)).collect(); + let mut expected = SSE_HEAD.to_vec(); + frames.iter().for_each(|frame| expected.extend(chunk(frame))); + let now = Instant::now(); + for frame in &frames { + assert!(server.api.fan_out(|_| true, frame, now), "queued for the subscriber"); + } + assert_eq!(subscribers(&server), 1); + + let got = serve(&mut server, read_exactly(client, expected.len()), "the burst"); + assert_same_bytes(&got, &expected); + } + + /// On Linux, registering `WRITABLE` on a writable socket queues an epoll + /// event. The small frame keeps this probe independent of burst capacity. + #[cfg(target_os = "linux")] + #[test] + fn a_frame_written_whole_to_an_idle_subscriber_leaves_nothing_to_report() { + let mut server = server_with(64, LONG_TIMEOUT); + let mut client = connect(tcp_addr(&server)); + subscribe(&mut client, "block"); + pump_until(&mut server, "subscribed and head sent", |server| { + subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0 + }); + + server.api.publish_block(7, &[0x77; 32]); + assert_eq!(bytes_waiting_for_subscribers(&server), 0, "the socket took the frame"); + server.readiness.wait(Duration::ZERO); + assert_eq!(server.readiness.events().iter().count(), 0); + drop(client); + } + + /// The queued response head leaves `WRITABLE` registered. On Linux, a + /// publish that drains the head must remove that interest before polling + /// and retain interest in inbound bytes. + #[cfg(target_os = "linux")] + #[test] + fn a_publish_that_drains_the_unsent_head_leaves_nothing_to_report() { + let dir = tempfile::tempdir().unwrap(); + let socket = dir.path().join("api.sock"); + let mut server = server_bound_to(&[Bind::Unix(socket.clone())], 64, LONG_TIMEOUT); + let mut client = connect_uds(&socket); + subscribe(&mut client, "block"); + pump_until(&mut server, "subscribed", |server| subscribers(server) == 1); + assert!(bytes_waiting_for_subscribers(&server) > 0, "the head is still queued"); + + server.api.publish_block(5, &[0x55; 32]); + assert_eq!(bytes_waiting_for_subscribers(&server), 0, "the publish drained the head"); + server.readiness.wait(Duration::ZERO); + assert_eq!(server.readiness.events().iter().count(), 0, "no writable event remains"); + + client.write_all(b"GET /eth/v1/node/version HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); + assert!(server.pump(), "inbound bytes produce a readable event"); + assert_eq!( + subscribers(&server), + 1, + "a request behind the subscribe is dropped, not answered" + ); + + drop(client); + pump_until(&mut server, "hung-up subscriber removed", |server| { + server.api.connections.is_empty() + }); + } + #[test] fn a_subscriber_that_stops_reading_is_closed_at_the_cap() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/httpcore/src/chunked_response.rs b/crates/httpcore/src/chunked_response.rs index c6dd5b01..81a1f822 100644 --- a/crates/httpcore/src/chunked_response.rs +++ b/crates/httpcore/src/chunked_response.rs @@ -1,8 +1,10 @@ use std::{ - io::Write, + io::{self, ErrorKind, Write}, time::{Duration, Instant}, }; +use mio::Interest; + // Reserve the full output allowance at construction so accepted pushes do // not reallocate. The response head and chunk framing count against it. const PENDING_MAX: usize = 64 << 10; @@ -57,6 +59,44 @@ impl ChunkedResponse { true } + /// Attempts to drain pending output after accepting a chunk within the cap. + /// Returns replacement readiness interests for the caller to register. + /// `None` leaves an already-empty response's `READABLE` registration + /// unchanged. Errors require the caller to close the connection. + pub fn deliver( + &mut self, + stream: &mut impl Write, + chunk: &[u8], + now: Instant, + ) -> Result, Closed> { + let backlog = !self.pending_write().is_empty(); + if !self.push(chunk, now) { + return Err(Closed::AtCap { pending: self.pending_write().len() }); + } + // An empty buffer already has READABLE alone. A backlog may retain + // WRITABLE from the response head or an earlier blocked write. + Ok(match (self.drain_into(stream, now).map_err(Closed::Lost)?, backlog) { + (true, false) => None, + (true, true) => Some(Interest::READABLE), + (false, _) => Some(Interest::READABLE | Interest::WRITABLE), + }) + } + + /// Returns `true` when no output remains, or `false` on `WouldBlock`. + /// On `true`, readiness-driven callers restore `READABLE` alone. + pub fn drain_into(&mut self, stream: &mut impl Write, now: Instant) -> io::Result { + while !self.pending_write().is_empty() { + match stream.write(self.pending_write()) { + Ok(0) => return Err(io::Error::new(ErrorKind::WriteZero, "write returned 0")), + Ok(n) => self.commit_write(n, now), + Err(e) if e.kind() == ErrorKind::WouldBlock => return Ok(false), + Err(e) if e.kind() == ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + } + } + Ok(true) + } + pub fn pending_write(&self) -> &[u8] { &self.pending[self.write_pos..] } @@ -86,6 +126,12 @@ impl ChunkedResponse { } } +#[derive(Debug)] +pub enum Closed { + AtCap { pending: usize }, + Lost(io::Error), +} + fn hex_digits(n: usize) -> usize { (usize::BITS - n.leading_zeros()).div_ceil(4).max(1) as usize } @@ -102,7 +148,7 @@ pub fn frame_chunked_head(out: &mut Vec, content_type: &str, headers: &[(&st #[cfg(test)] mod tests { - use std::cell::Cell; + use std::{cell::Cell, collections::VecDeque}; use super::*; use crate::{ParsedRequest, ServerConnection, frame_response}; @@ -157,6 +203,179 @@ mod tests { vec![b'e'; n] } + fn burst_frame(index: usize, len: usize) -> Vec { + let mut frame = format!("event: burst\ndata: {index:02}").into_bytes(); + frame.resize(len - 2, b'c'); + frame.extend_from_slice(b"\n\n"); + frame + } + + #[derive(Clone, Copy)] + enum Step { + Take(usize), + WouldBlock, + Interrupted, + Zero, + Broken, + } + + /// After the scripted steps are consumed, each write follows `then`. + struct ScriptedSocket { + steps: VecDeque, + then: Step, + taken: Vec, + } + + impl ScriptedSocket { + fn taking_everything() -> Self { + Self { steps: VecDeque::new(), then: Step::Take(usize::MAX), taken: Vec::new() } + } + + fn refusing_everything() -> Self { + Self { then: Step::WouldBlock, ..Self::taking_everything() } + } + + fn script(&mut self, steps: impl IntoIterator) { + self.steps.extend(steps); + } + } + + impl Write for ScriptedSocket { + fn write(&mut self, buf: &[u8]) -> io::Result { + match self.steps.pop_front().unwrap_or(self.then) { + Step::Take(n) => { + let n = n.min(buf.len()); + self.taken.extend_from_slice(&buf[..n]); + Ok(n) + } + Step::WouldBlock => Err(ErrorKind::WouldBlock.into()), + Step::Interrupted => Err(ErrorKind::Interrupted.into()), + Step::Zero => Ok(0), + Step::Broken => Err(ErrorKind::BrokenPipe.into()), + } + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + const BOTH: Interest = Interest::READABLE.add(Interest::WRITABLE); + + /// The queued head supplies the initial backlog. Once drained, later + /// deliveries need no registration change while the writer accepts output. + #[test] + fn a_burst_past_the_cap_is_written_as_it_is_pushed() { + let t0 = Instant::now(); + let mut stream = subscribed(t0); + let mut socket = ScriptedSocket::taking_everything(); + let frames: Vec<_> = (0..40).map(|index| burst_frame(index, 2300)).collect(); + let mut expected = HEAD.to_vec(); + frames.iter().for_each(|frame| expected.extend(framed(frame))); + assert!(expected.len() - HEAD.len() > PENDING_MAX, "the burst passes the cap"); + + let (first, rest) = frames.split_first().unwrap(); + assert_eq!(stream.deliver(&mut socket, first, t0).unwrap(), Some(Interest::READABLE)); + for frame in rest { + assert_eq!(stream.deliver(&mut socket, frame, t0).unwrap(), None); + } + assert_eq!(socket.taken, expected); + assert!(stream.pending_write().is_empty()); + } + + /// A blocked write requests `WRITABLE`; a delivery that drains the backlog + /// requests `READABLE` alone. The readiness path reports a complete drain + /// for the caller to make the same registration change. + #[test] + fn a_refused_write_arms_writable_until_the_backlog_drains() { + let t0 = Instant::now(); + let mut stream = subscribed(t0); + let mut socket = ScriptedSocket::taking_everything(); + assert!(stream.drain_into(&mut socket, t0).unwrap()); + let frames = [burst_frame(1, 300), burst_frame(2, 300), burst_frame(3, 300)]; + + socket.script([Step::WouldBlock]); + assert_eq!(stream.deliver(&mut socket, &frames[0], t0).unwrap(), Some(BOTH)); + assert_eq!(stream.pending_write(), framed(&frames[0])); + let by_publish = stream.deliver(&mut socket, &frames[1], t0).unwrap(); + assert_eq!(by_publish, Some(Interest::READABLE)); + assert!(stream.pending_write().is_empty()); + + socket.script([Step::WouldBlock]); + assert_eq!(stream.deliver(&mut socket, &frames[2], t0).unwrap(), Some(BOTH)); + assert!(stream.drain_into(&mut socket, t0).unwrap(), "the loop drains the rest"); + + let expected: Vec<_> = frames.iter().flat_map(|frame| framed(frame)).collect(); + assert_eq!(socket.taken[HEAD.len()..], expected); + } + + /// Partial writes reset the stall clock. Interruptions are retried, and + /// draining the remaining bytes clears the clock. + #[test] + fn partial_writes_continue_until_the_socket_refuses() { + let t0 = Instant::now(); + let mut stream = subscribed(t0); + let mut socket = ScriptedSocket::taking_everything(); + let frame = burst_frame(0, 1000); + let expected = [HEAD, &framed(&frame)].concat(); + + socket.script([Step::Take(100), Step::Take(100), Step::WouldBlock]); + let t1 = t0 + Duration::from_secs(1); + assert_eq!(stream.deliver(&mut socket, &frame, t1).unwrap(), Some(BOTH)); + assert_eq!(socket.taken, expected[..200]); + assert_eq!(stream.pending_write(), &expected[200..]); + assert!(!stream.stalled(t1 + DEADLINE, DEADLINE)); + assert!(stream.stalled(t1 + DEADLINE + Duration::from_millis(1), DEADLINE)); + + socket.script([Step::Interrupted, Step::Take(50)]); + assert!(stream.drain_into(&mut socket, t1).unwrap()); + assert_eq!(socket.taken, expected); + assert!(!stream.stalled(t1 + DEADLINE * 100, DEADLINE)); + } + + /// The cap error reports bytes already pending, including the response + /// head. + #[test] + fn pushes_the_socket_never_takes_close_at_the_cap() { + let t0 = Instant::now(); + let mut stream = subscribed(t0); + let mut socket = ScriptedSocket::refusing_everything(); + let frame = burst_frame(0, 2300); + let chunk = framed(&frame).len(); + let fit = (PENDING_MAX - HEAD.len()) / chunk; + + for pushed in 0..fit { + let interest = stream.deliver(&mut socket, &frame, t0).unwrap(); + assert_eq!(interest, Some(BOTH), "push {pushed} waits for the socket"); + } + let closed = stream.deliver(&mut socket, &frame, t0).unwrap_err(); + let at_cap = + matches!(closed, Closed::AtCap { pending } if pending == HEAD.len() + fit * chunk); + assert!(at_cap, "{closed:?}"); + assert!(socket.taken.is_empty()); + } + + /// Both zero-length writes and write errors require the caller to close. + #[test] + fn a_dead_socket_ends_the_stream() { + let t0 = Instant::now(); + let frame = burst_frame(0, 100); + + let mut zero = ScriptedSocket::taking_everything(); + zero.script([Step::Zero]); + let Err(Closed::Lost(e)) = subscribed(t0).deliver(&mut zero, &frame, t0) else { + panic!("a zero-length write is an error") + }; + assert_eq!(e.kind(), ErrorKind::WriteZero); + + let mut broken = ScriptedSocket::taking_everything(); + broken.script([Step::Broken]); + let Err(Closed::Lost(e)) = subscribed(t0).deliver(&mut broken, &frame, t0) else { + panic!("a failed write is an error") + }; + assert_eq!(e.kind(), ErrorKind::BrokenPipe); + } + #[test] fn the_head_leaves_first_ahead_of_chunks_pushed_before_the_first_drain() { let t0 = Instant::now(); diff --git a/crates/httpcore/src/lib.rs b/crates/httpcore/src/lib.rs index 4aae12e9..b8d8aa63 100644 --- a/crates/httpcore/src/lib.rs +++ b/crates/httpcore/src/lib.rs @@ -6,7 +6,7 @@ mod server; mod stream; mod token_range; -pub use chunked_response::{ChunkedResponse, frame_chunked_head}; +pub use chunked_response::{ChunkedResponse, Closed, frame_chunked_head}; pub use client::{ClientConnection, frame_request}; pub use query::Query; pub use readiness::Readiness; diff --git a/docs/adr/0004-sync-materialized-api.md b/docs/adr/0004-sync-materialized-api.md index 9ee38058..efcf2623 100644 --- a/docs/adr/0004-sync-materialized-api.md +++ b/docs/adr/0004-sync-materialized-api.md @@ -121,3 +121,11 @@ If the cache has overwritten an object's bytes, the boundary logs a warning and emits no event for that request. It also skips sidecars whose length and column offset identify neither supported layout. These failures do not cancel the publication request. + +Amended 2026-09-14: each accepted push immediately attempts to write the +subscription's pending output. A burst can therefore drain into the socket +without waiting for the next readiness event. The 64 KiB cap applies before +each push, including the response head and chunk framing. A push that exceeds +the cap closes the connection before another write is attempted. Successful +writes reset the send deadline while output remains pending; draining it +clears the deadline. Progress means kernel acceptance, not peer consumption. From 8909135d3356216055b28e634d282f290d0f80de Mon Sep 17 00:00:00 2001 From: Bronek Kozicki Date: Mon, 14 Sep 2026 16:06:40 +0100 Subject: [PATCH 2/2] Raise the SSE send cap to 512 KiB One block's 128 data_column_sidecar events are expected to total about 290 KiB when each carries 21 Fulu commitments. Eager writes can reduce pending output, but the 64 KiB cap cannot hold that burst if the socket accepts no bytes. Raise the cap to 512 KiB to accommodate the burst without relying on kernel send-buffer capacity. Earlier pending events or repeated publications can still exhaust the allowance. At 64 subscriptions, the pending-output allowance grows from 4 MiB to 32 MiB. Buffers retain their allocations until the subscriptions close, including after a full drain. Add a scripted-writer test that accepts all 128 representative frames while every write returns WouldBlock. The response head also counts against the cap. Derive the eager-write test's burst length from the cap so queuing without writing would still exceed it after this increase. Expand the real-socket delivery test to 128 frames and check their order. Add a small publication test that checks for no pending bytes before pumping, independently of whether the larger burst fits the buffer. Document the allowance and allocation lifetime in ADR-0004. Assisted-by: Claude:claude-fable-5-1 Assisted-by: Codex:GPT-6 --- crates/beacon_api/src/server.rs | 27 +++++++++++++++++++++++-- crates/httpcore/src/chunked_response.rs | 24 ++++++++++++++++++++-- docs/adr/0004-sync-materialized-api.md | 10 +++++++++ 3 files changed, 57 insertions(+), 4 deletions(-) diff --git a/crates/beacon_api/src/server.rs b/crates/beacon_api/src/server.rs index 82cdc64b..572d81c7 100644 --- a/crates/beacon_api/src/server.rs +++ b/crates/beacon_api/src/server.rs @@ -1455,7 +1455,9 @@ mod tests { /// The burst fits in the application buffer even if the socket initially /// accepts no bytes. Delivery therefore does not require a particular - /// kernel send-buffer capacity. + /// kernel send-buffer capacity. The frames approximate one block's column + /// events with commitments; any unsent remainder drains through the + /// readiness loop. #[test] fn a_burst_reaches_a_reading_subscriber_in_order() { let dir = tempfile::tempdir().unwrap(); @@ -1467,7 +1469,7 @@ mod tests { subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0 }); - let frames: Vec<_> = (0..24).map(|index| burst_frame(index, 2300)).collect(); + let frames: Vec<_> = (0..128).map(|index| burst_frame(index, 2300)).collect(); let mut expected = SSE_HEAD.to_vec(); frames.iter().for_each(|frame| expected.extend(chunk(frame))); let now = Instant::now(); @@ -1480,6 +1482,27 @@ mod tests { assert_same_bytes(&got, &expected); } + /// Checks that publication attempts delivery before the readiness loop, + /// independently of whether a larger burst fits the application buffer. + #[test] + fn a_publish_to_a_reading_subscriber_leaves_nothing_pending() { + let dir = tempfile::tempdir().unwrap(); + let socket = dir.path().join("api.sock"); + let mut server = server_bound_to(&[Bind::Unix(socket.clone())], 64, LONG_TIMEOUT); + let mut client = connect_uds(&socket); + subscribe(&mut client, "block"); + pump_until(&mut server, "subscribed and head sent", |server| { + subscribers(server) == 1 && bytes_waiting_for_subscribers(server) == 0 + }); + + for slot in 1..=3 { + server.api.publish_block(slot, &[0x33; 32]); + } + assert_eq!(bytes_waiting_for_subscribers(&server), 0); + assert_eq!(subscribers(&server), 1); + drop(client); + } + /// On Linux, registering `WRITABLE` on a writable socket queues an epoll /// event. The small frame keeps this probe independent of burst capacity. #[cfg(target_os = "linux")] diff --git a/crates/httpcore/src/chunked_response.rs b/crates/httpcore/src/chunked_response.rs index 81a1f822..e1ff6e33 100644 --- a/crates/httpcore/src/chunked_response.rs +++ b/crates/httpcore/src/chunked_response.rs @@ -7,7 +7,9 @@ use mio::Interest; // Reserve the full output allowance at construction so accepted pushes do // not reallocate. The response head and chunk framing count against it. -const PENDING_MAX: usize = 64 << 10; +// Allows one block's 128 column events with 21 commitments each, about +// 290 KiB in total, even when the writer accepts no bytes. +const PENDING_MAX: usize = 512 << 10; const DISCARD_LEN: usize = 4096; /// Does not emit a terminal chunk; the caller ends the stream by closing @@ -269,7 +271,8 @@ mod tests { let t0 = Instant::now(); let mut stream = subscribed(t0); let mut socket = ScriptedSocket::taking_everything(); - let frames: Vec<_> = (0..40).map(|index| burst_frame(index, 2300)).collect(); + let past_the_cap = PENDING_MAX / framed(&burst_frame(0, 2300)).len() + 1; + let frames: Vec<_> = (0..past_the_cap).map(|index| burst_frame(index, 2300)).collect(); let mut expected = HEAD.to_vec(); frames.iter().for_each(|frame| expected.extend(framed(frame))); assert!(expected.len() - HEAD.len() > PENDING_MAX, "the burst passes the cap"); @@ -333,6 +336,23 @@ mod tests { assert!(!stream.stalled(t1 + DEADLINE * 100, DEADLINE)); } + /// Synthetic frames approximate column events with 21 commitments. + /// The queued response head also counts against the allowance. + #[test] + fn one_blocks_column_events_fit_the_cap_with_the_socket_taking_nothing() { + let t0 = Instant::now(); + let mut stream = subscribed(t0); + let mut socket = ScriptedSocket::refusing_everything(); + let frames: Vec<_> = (0..128).map(|index| burst_frame(index, 2300)).collect(); + + for frame in &frames { + assert_eq!(stream.deliver(&mut socket, frame, t0).unwrap(), Some(BOTH)); + } + let mut expected = HEAD.to_vec(); + frames.iter().for_each(|frame| expected.extend(framed(frame))); + assert_eq!(stream.pending_write(), expected); + } + /// The cap error reports bytes already pending, including the response /// head. #[test] diff --git a/docs/adr/0004-sync-materialized-api.md b/docs/adr/0004-sync-materialized-api.md index efcf2623..e94b9014 100644 --- a/docs/adr/0004-sync-materialized-api.md +++ b/docs/adr/0004-sync-materialized-api.md @@ -129,3 +129,13 @@ each push, including the response head and chunk framing. A push that exceeds the cap closes the connection before another write is attempted. Successful writes reset the send deadline while output remains pending; draining it clears the deadline. Progress means kernel acceptance, not peer consumption. + +Amended 2026-09-14: the send cap is 512 KiB. This accommodates one block's +128 `data_column_sidecar` events with 21 commitments each, estimated at +290 KiB, even when the socket accepts no bytes. Earlier pending events or +repeated publications can still exhaust the allowance. + +At 64 subscriptions, the pending-output allowance totals 32 MiB. Each +subscription reserves its buffer at construction and retains the allocation +until it closes. A full drain reuses the buffer from its beginning without +releasing it. Partial drains can advance through the entire allocation.