From d827c9780c24ce8fb4236db4cf99dc13987619db Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 24 Oct 2026 23:42:35 -0400 Subject: [PATCH 01/12] feat(relay): bound the total queued frames and the sockets one address may hold (#959) The outbound queue was bounded per socket and their sum was bounded by nothing: one address could hold 8 MiB for every socket it kept, and a closed socket's queued bytes were never given back. Every socket now counts its queued payload against one process-wide FrameBudget (256 MiB, RelayConfig::frame_memory_budget_bytes), refunded when the socket's last sink handle drops, and one address may hold at most max_connections_per_ip live sockets (64) as a gauge beside the attempt limiter. Both refuse with an immediate 1013 close, a fixed reason and their own counter. See docs/rewrite/decisions/0008-frame-memory-budget.md and #959. --- .changeset/relay-frame-memory-budget.md | 22 + crates/nostream-relay/src/connection_cap.rs | 168 ++++++++ crates/nostream-relay/src/lib.rs | 1 + crates/nostream-relay/src/relay.rs | 126 +++++- crates/nostream-relay/src/sink.rs | 439 ++++++++++++++++++-- crates/nostream-relay/src/transport.rs | 8 + crates/nostream-relay/tests/connection.rs | 6 +- crates/nostream-relay/tests/limits.rs | 2 +- crates/nostream-relay/tests/strategies.rs | 2 +- crates/nostream-relay/tests/transport.rs | 274 ++++++++++++ 10 files changed, 978 insertions(+), 70 deletions(-) create mode 100644 .changeset/relay-frame-memory-budget.md create mode 100644 crates/nostream-relay/src/connection_cap.rs diff --git a/.changeset/relay-frame-memory-budget.md b/.changeset/relay-frame-memory-budget.md new file mode 100644 index 00000000..2292cfa7 --- /dev/null +++ b/.changeset/relay-frame-memory-budget.md @@ -0,0 +1,22 @@ +--- +--- + +feat(relay): bound the relay's total queued frames and the sockets one address may hold + +The outbound queue was bounded per socket and the number of sockets was bounded by nothing: +one address that opened sockets and stopped reading could hold 8 MiB for every socket it kept, +measured as 384 MiB from the shipped attempt limits, and nothing gave a closed socket's queued +bytes back. Two bounds now sit beside that one, both recorded in +`docs/rewrite/decisions/0008-frame-memory-budget.md`: + +- `RelayConfig::frame_memory_budget_bytes` (256 MiB) is one process-wide budget every socket's + queued payload is counted against, admitted frame by frame at the sink and refunded when a + socket's last handle drops. +- `RelayConfig::max_connections_per_ip` (64) caps the live sockets one address may hold, as a + gauge beside the attempt limiter rather than folded into it. + +Both refuse the way the policy comment on #959 asks: an immediate `1013` close with a fixed +reason and its own counter, never a text frame and never a silent drop. See #959. + +The TypeScript tree is untouched: no `src/` or `test/` file changes, so the published package +has no user-visible change. diff --git a/crates/nostream-relay/src/connection_cap.rs b/crates/nostream-relay/src/connection_cap.rs new file mode 100644 index 00000000..48641393 --- /dev/null +++ b/crates/nostream-relay/src/connection_cap.rs @@ -0,0 +1,168 @@ +//! The per-IP live-connection cap: a gauge, beside the attempt limiter and not part of it. +//! +//! `limits.rs` answers "how fast may this address open connections" over a sliding window; +//! this answers "how many may it hold at once". The capture has neither question +//! (`src/@types/settings.ts:163-167`: `ConnectionLimits` is `rateLimits`, `ipWhitelist` and +//! `ipBlacklist`), so the cap is the port's own policy with no counterpart there (ADR 0008). +//! +//! Two things it deliberately is not. It is not a second `RateLimits` store: no window, no +//! clock, no whitelist, nothing to expire — a count that only `Drop` changes. And it is not +//! advisory: the connection task holds a [`ConnectionSlot`] for as long as the socket lives, +//! so an address at the cap is refused before the NIP-42 challenge is sent rather than after +//! the client has already been spoken to. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +/// The live sockets per source address, with the cap each is checked against (ADR 0008). +#[derive(Debug)] +pub struct ConnectionCap { + max_per_ip: usize, + live: Mutex>, + refusals: AtomicU64, +} + +impl ConnectionCap { + /// A cap of `max_per_ip` sockets per address, with nothing counted. + pub fn new(max_per_ip: usize) -> Self { + Self { + max_per_ip, + live: Mutex::new(HashMap::new()), + refusals: AtomicU64::new(0), + } + } + + /// The most sockets one address may hold at once. + pub fn max_per_ip(&self) -> usize { + self.max_per_ip + } + + /// How many sockets `ip` holds now. + pub fn live(&self, ip: &str) -> usize { + self.live + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .get(ip) + .copied() + .unwrap_or(0) + } + + /// How many addresses hold a socket: an address that releases its last one leaves no entry. + pub fn tracked_addresses(&self) -> usize { + self.live + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .len() + } + + /// How many sockets were refused because their address was at the cap. + pub fn refusals(&self) -> u64 { + self.refusals.load(Ordering::SeqCst) + } + + /// Takes a slot for `ip`, or counts the refusal and returns `None` at the cap. + /// + /// The slot releases on `Drop`, so a caller that holds it for the whole connection + /// releases it on every exit — the normal one, a registration failure, or one added + /// later — without a decrement at each. + pub fn acquire(self: &Arc, ip: &str) -> Option { + let mut live = self + .live + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + let held = live.get(ip).copied().unwrap_or(0); + + if held >= self.max_per_ip { + // Nothing is inserted for a refused address: an entry that must not be there + // would read as a socket held. + drop(live); + self.refusals.fetch_add(1, Ordering::SeqCst); + + return None; + } + + live.insert(ip.to_owned(), held + 1); + + Some(ConnectionSlot { + cap: Arc::clone(self), + ip: ip.to_owned(), + }) + } +} + +/// One socket's slot in a [`ConnectionCap`], given back when it drops. +#[derive(Debug)] +pub struct ConnectionSlot { + cap: Arc, + ip: String, +} + +impl Drop for ConnectionSlot { + fn drop(&mut self) { + let mut live = self + .cap + .live + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + + if let Some(held) = live.get_mut(&self.ip) { + *held = held.saturating_sub(1); + + if *held == 0 { + live.remove(&self.ip); + } + } + } +} + +#[cfg(test)] +mod tests { + #![allow(clippy::expect_used)] + + use super::*; + + #[test] + fn the_cap_counts_live_slots_and_releases_each_once() { + let cap = Arc::new(ConnectionCap::new(2)); + + let first = cap.acquire("127.0.0.1").expect("a slot below the cap"); + let second = cap + .acquire("127.0.0.1") + .expect("the last slot below the cap"); + + assert_eq!(cap.live("127.0.0.1"), 2); + assert_eq!(cap.tracked_addresses(), 1); + + assert!(cap.acquire("127.0.0.1").is_none()); + assert_eq!(cap.refusals(), 1); + assert_eq!(cap.live("127.0.0.1"), 2, "a refusal holds no slot"); + + // A second address has its own count, and an address at the cap does not affect it. + let elsewhere = cap + .acquire("10.0.0.1") + .expect("another address is unaffected"); + + assert_eq!(cap.live("10.0.0.1"), 1); + assert_eq!(cap.tracked_addresses(), 2); + + drop(first); + + assert_eq!(cap.live("127.0.0.1"), 1); + assert_eq!(cap.tracked_addresses(), 2); + + drop(second); + + assert_eq!(cap.live("127.0.0.1"), 0); + assert_eq!( + cap.tracked_addresses(), + 1, + "an address that releases its last slot leaves no entry" + ); + + drop(elsewhere); + + assert_eq!(cap.tracked_addresses(), 0); + assert_eq!(cap.refusals(), 1); + } +} diff --git a/crates/nostream-relay/src/lib.rs b/crates/nostream-relay/src/lib.rs index 54de5777..5d1752cb 100644 --- a/crates/nostream-relay/src/lib.rs +++ b/crates/nostream-relay/src/lib.rs @@ -44,6 +44,7 @@ pub mod broadcast; pub mod connection; +pub mod connection_cap; pub mod error; pub mod handlers; pub mod hidden_content; diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index 70fb1613..ce6722e3 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -28,20 +28,21 @@ use nostream_core::messages::RelayMessage; use nostream_core::{Event, messages}; use crate::connection::{AcceptedTransport, Connection, ConnectionContext}; +use crate::connection_cap::ConnectionCap; use crate::handlers::{self, HandlerContext, Stores}; use crate::hidden_content::{DEFAULT_REFRESH_INTERVAL, HiddenContentCache}; use crate::limits::RateLimits; use crate::relay_key::RelayKeySource; -use crate::sink::FrameSink; +use crate::sink::{FrameBudget, FrameSink, Refusal}; use crate::transport; use crate::wot::WotGraph; /// The knobs a [`Relay`] is built with that are not settings: a constant the capture fixes in /// code (`liveness`, the NIP-56 refresh) or a policy the port adds with no counterpart there -/// (`frame_queue_bytes`, `write_timeout`; ADR 0005). Tests set the cadences they need -/// (`tests/common/mod.rs::fast_config`); the two policies run at their shipped defaults in -/// every test so far. A value the capture's settings document carries belongs in [`Settings`], -/// never here. +/// (`frame_queue_bytes`, `write_timeout`, the frame-memory budget and the per-IP connection +/// cap; ADR 0005, ADR 0008). Tests set the cadences they need (`tests/common/mod.rs::fast_config`); +/// the policies run at their shipped defaults except where a test sets one deliberately. A +/// value the capture's settings document carries belongs in [`Settings`], never here. #[derive(Debug, Clone)] pub struct RelayConfig { /// `WSS_CLIENT_HEALTH_PROBE_INTERVAL` (`web-socket-server-adapter.ts:18`). @@ -59,6 +60,13 @@ pub struct RelayConfig { /// cannot starve (`web-socket-server-adapter.ts:18,46`), while this loop performs the /// write itself. pub write_timeout: Duration, + /// The queued outbound payload bytes every socket of this relay shares. The capture has + /// no counterpart: it bounds neither one socket's queue nor their sum (ADR 0008). + pub frame_memory_budget_bytes: usize, + /// The live sockets one source address may hold at once. The capture has no counterpart + /// either: its connection limits are rates, not counts (`src/@types/settings.ts:163-167`; + /// ADR 0008). + pub max_connections_per_ip: usize, } impl Default for RelayConfig { @@ -76,6 +84,14 @@ impl Default for RelayConfig { // Shorter than the 120 s liveness interval, so a client that stops reading is // closed by the write path before a missed pong could close it. write_timeout: Duration::from_secs(30), + // 256 MiB: 32 sockets at the `frame_queue_bytes` ceiling can be fully buffered + // before the budget binds, which is below the 48 a single address gets through + // the shipped attempt limits alone (ADR 0008). + frame_memory_budget_bytes: 256 * 1024 * 1024, + // 64: four times the concurrent total one well-behaved client needs. The cap + // bounds sockets, the budget bounds bytes, so its own worst case (64 × 8 MiB) sits + // above the budget on purpose (ADR 0008). + max_connections_per_ip: 64, } } } @@ -92,6 +108,8 @@ pub struct Relay { hidden: Arc, hub: broadcast::Sender>, pow: Arc>, + frame_budget: Arc, + connection_cap: Arc, config: RelayConfig, } @@ -112,8 +130,8 @@ impl Relay { /// Everything the relay runs with is an argument, so a relay that mints from /// `SystemRandom` or resolves `WS_DRAIN_TIMEOUT_MS` from the environment is one whose /// composition root asked for it: `RelayConfig::default()` is the port's own defaults — the - /// capture's constants plus the two policies ADR 0005 adds — and the environment read is - /// [`transport::drain_timeout`] at the call site. + /// capture's constants plus the policies ADR 0005 and ADR 0008 add — and the environment + /// read is [`transport::drain_timeout`] at the call site. pub fn new( settings: Settings, stores: Arc, @@ -125,6 +143,8 @@ impl Relay { ) -> Arc { let settings = Arc::new(settings); let (hub, _receiver) = broadcast::channel(1024); + let frame_budget = Arc::new(FrameBudget::new(config.frame_memory_budget_bytes)); + let connection_cap = Arc::new(ConnectionCap::new(config.max_connections_per_ip)); Arc::new(Self { limits: Arc::new(RateLimits::new(&settings, Arc::clone(&clock))), @@ -137,6 +157,8 @@ impl Relay { hidden: HiddenContentCache::new(), hub, pow: Arc::new(Mutex::new(AdaptivePowState::default())), + frame_budget, + connection_cap, config, }) } @@ -147,6 +169,18 @@ impl Relay { &self.hidden } + /// The process-wide bound on queued outbound payload bytes. The counter it carries is what + /// a metrics exporter would read once the admin `/metrics` route is implemented; the warn + /// line on a refusal is what an operator has today (ADR 0008). + pub fn frame_budget(&self) -> &Arc { + &self.frame_budget + } + + /// The live-connection cap per source address, and its refusal counter. + pub fn connection_cap(&self) -> &Arc { + &self.connection_cap + } + /// The context one socket's handlers run in. fn handler_context(&self, connection: Arc, sink: FrameSink) -> HandlerContext { HandlerContext { @@ -242,7 +276,7 @@ impl Relay { ) { let max_payload = transport::max_payload_size(&self.settings); - let transport = match transport::upgrade(stream, max_payload).await { + let mut transport = match transport::upgrade(stream, max_payload).await { Ok(transport) => transport, Err(error) => { tracing::debug!(%peer, error = %error, "websocket handshake failed"); @@ -255,6 +289,30 @@ impl Relay { return; } + // After the attempt limiter, so a rate-limited socket is still dropped silently as it + // always was and takes no slot; before `run_connection`, because registering the + // `Connection` sends the NIP-42 challenge as a text frame and a refused client must + // receive no text frame (ADR 0008). The slot is held until this function returns, + // which covers the registration failure inside `run_connection` as well as its + // normal exit. + let ip = peer.ip().to_string(); + let Some(_slot) = self.connection_cap.acquire(&ip) else { + tracing::debug!( + %peer, + max_connections_per_ip = self.connection_cap.max_per_ip(), + live = self.connection_cap.live(&ip), + "client terminated: connection cap" + ); + self.close_frame( + &mut transport, + CloseCode::Again, + transport::CONNECTION_CAP_REASON, + ) + .await; + + return; + }; + self.run_connection(transport, peer, shutdown).await; } @@ -283,7 +341,10 @@ impl Relay { peer: std::net::SocketAddr, shutdown: ShutdownState, ) { - let (sink, mut frames) = FrameSink::channel(self.config.frame_queue_bytes); + let (sink, mut frames) = FrameSink::channel( + self.config.frame_queue_bytes, + Some(Arc::clone(&self.frame_budget)), + ); let connection_context = ConnectionContext { settings: Arc::clone(&self.settings), @@ -327,8 +388,8 @@ impl Relay { // Read at the top of every iteration, so a flag the producers set while the loop // was parked on a write is acted on as soon as that write returns. No `await` // sits between a write completing and this check. - if sink.overflowed() { - self.close_slow_consumer(&mut transport, peer).await; + if let Some(refusal) = sink.refusal() { + self.close_refused(&mut transport, peer, refusal).await; break; } @@ -428,24 +489,47 @@ impl Relay { connection.close(); } - /// Closes a client whose outbound frame queue overflowed: 1013 with the reason, then the - /// loop's cleanup (`handlers.abort_all()` and `connection.close()`). + /// Closes a client whose frame was refused, with the close its bound implies: 1013 and the + /// reason for that bound, then the loop's cleanup (`handlers.abort_all()` and + /// `connection.close()`). /// - /// The policy is a disconnect, not a silent break: a client cut mid-`REQ` with no frame - /// cannot tell a slow link from a relay that forgot it (ADR 0005). - async fn close_slow_consumer( + /// The policy is a disconnect, not a silent break, for both bounds: a client cut mid-`REQ` + /// with no frame cannot tell a slow link from a relay that forgot it (ADR 0005), and a + /// refusal that looked like a network fault is exactly what the two reason strings and + /// their counters exist to keep out of the logs (ADR 0008). + async fn close_refused( &self, transport: &mut AcceptedTransport, peer: std::net::SocketAddr, + refusal: Refusal, ) { - tracing::debug!(%peer, "closing a client that cannot keep up with its frame queue"); + match refusal { + Refusal::QueueFull => { + tracing::debug!(%peer, "closing a client that cannot keep up with its frame queue"); + + self.close_frame(transport, CloseCode::Again, transport::SLOW_CONSUMER_REASON) + .await; + } + Refusal::BudgetExhausted => { + tracing::debug!(%peer, "closing a client that exhausted the frame memory budget"); + + self.close_frame(transport, CloseCode::Again, transport::FRAME_BUDGET_REASON) + .await; + } + } + } + /// Writes one close frame straight to the transport, bounded by the write timeout so a + /// client that stopped reading cannot hold the task open. + async fn close_frame( + &self, + transport: &mut AcceptedTransport, + code: CloseCode, + reason: &'static str, + ) { let _ = tokio::time::timeout( self.config.write_timeout, - transport.close(Some(transport::close_frame( - CloseCode::Again, - transport::SLOW_CONSUMER_REASON, - ))), + transport.close(Some(transport::close_frame(code, reason))), ) .await; } diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index 9b8ed540..dfb236d1 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -24,39 +24,193 @@ //! depth — one line per sender, not one per connection — and every later send is dropped //! without queueing. The connection task reads the flag and closes the client with 1013, so //! the policy is a disconnect rather than a silent break (ADR 0005). +//! +//! **The sum is bounded too**, which the per-socket bound cannot do: one address that opens +//! sockets and stops reading held a multiple of the per-socket bound, so every socket's +//! queue is also counted against one process-wide [`FrameBudget`] (ADR 0008). An admission +//! needs room in both bounds; the budget has no empty-queue exception, so a frame that does +//! not fit the budget is refused however empty its own queue. The socket's share of the +//! budget is refunded when its last sink handle drops — see [`Account`] — because the next +//! socket's admission depends on it. use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU8, AtomicU64, AtomicUsize, Ordering}; use tokio::sync::mpsc; use nostream_core::messages::RelayMessage; +/// Why a frame was refused: the bound that could not hold it. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Refusal { + /// The frame did not fit this socket's `frame_queue_bytes`. + QueueFull, + /// The frame did not fit the relay's `frame_memory_budget_bytes`. + BudgetExhausted, +} + +/// No refusal recorded yet; the two values below are this file's only refusal causes. +const NO_REFUSAL: u8 = 0; +const QUEUE_FULL: u8 = 1; +const BUDGET_EXHAUSTED: u8 = 2; + +/// The process-wide bound on queued outbound payload bytes, shared by every socket of one +/// relay (ADR 0008). +/// +/// The unit is `text.len()`, the same bytes `frame_queue_bytes` counts: not `String` +/// capacity, and not resident memory. +#[derive(Debug)] +pub struct FrameBudget { + limit_bytes: usize, + accounted_bytes: AtomicUsize, + refusals: AtomicU64, +} + +impl FrameBudget { + /// A budget of `limit_bytes`, with nothing accounted. + pub fn new(limit_bytes: usize) -> Self { + Self { + limit_bytes, + accounted_bytes: AtomicUsize::new(0), + refusals: AtomicU64::new(0), + } + } + + /// The byte bound. + pub fn limit(&self) -> usize { + self.limit_bytes + } + + /// The bytes admitted across every socket and not yet written or refunded. + pub fn accounted(&self) -> usize { + self.accounted_bytes.load(Ordering::SeqCst) + } + + /// How many sockets were refused because their frame did not fit. + pub fn refusals(&self) -> u64 { + self.refusals.load(Ordering::SeqCst) + } + + /// Admits `bytes` against the budget, or reports that they do not fit. + /// + /// A compare-and-exchange loop for the same reason [`FrameSink::send`] uses one: every + /// handler of every socket shares this counter, and a load-then-store would lose an + /// increment and admit more than the budget. + fn admit(&self, bytes: usize) -> bool { + let mut accounted = self.accounted_bytes.load(Ordering::SeqCst); + + loop { + // Strict: the whole frame has to fit what is left, with no exception for an + // empty budget — unlike the per-socket bound, whose exception keeps one large + // event from closing every client. + if bytes > self.limit_bytes.saturating_sub(accounted) { + return false; + } + + match self.accounted_bytes.compare_exchange_weak( + accounted, + accounted + bytes, + Ordering::SeqCst, + Ordering::SeqCst, + ) { + Ok(_) => return true, + Err(actual) => accounted = actual, + } + } + } + + /// Returns bytes to the budget: a written frame's, or a closing socket's remainder. + fn release(&self, bytes: usize) { + let _ = self + .accounted_bytes + .try_update(Ordering::SeqCst, Ordering::SeqCst, |accounted| { + Some(accounted.saturating_sub(bytes)) + }); + } + + /// Counts one socket refused for the budget. + fn refuse(&self) { + self.refusals.fetch_add(1, Ordering::SeqCst); + } +} + +/// One socket's accounting: the bytes it has admitted, its refusal cause, and the budget +/// those bytes are counted against when the relay has one. +/// +/// Every clone of the socket's sink shares this value, so the `Drop` that runs here is the +/// socket's accounting ending: a send holds a sink and therefore cannot be in flight or +/// begin once the drop starts. That is what makes the refund exactly this socket's +/// outstanding bytes, once, on every exit from the connection loop — the drain arm's +/// [`FrameSink::release`] keeps handling the bytes it wrote on the way out. +#[derive(Debug)] +struct Account { + queued_bytes: AtomicUsize, + refusal: AtomicU8, + budget: Option>, +} + +impl Account { + /// The refusal recorded for this socket, if a sender has been refused. + fn refusal(&self) -> Option { + match self.refusal.load(Ordering::SeqCst) { + QUEUE_FULL => Some(Refusal::QueueFull), + BUDGET_EXHAUSTED => Some(Refusal::BudgetExhausted), + _ => None, + } + } + + /// Records the first refusal for this socket, reporting whether this sender set it. + fn refuse(&self, refusal: Refusal) -> bool { + let cause = match refusal { + Refusal::QueueFull => QUEUE_FULL, + Refusal::BudgetExhausted => BUDGET_EXHAUSTED, + }; + + self.refusal + .compare_exchange(NO_REFUSAL, cause, Ordering::SeqCst, Ordering::SeqCst) + .is_ok() + } +} + +impl Drop for Account { + fn drop(&mut self) { + if let Some(budget) = &self.budget { + budget.release(self.queued_bytes.load(Ordering::SeqCst)); + } + } +} + /// A handle to one connection's outbound frame queue. #[derive(Debug, Clone)] pub struct FrameSink { sender: Option>, /// The byte bound the queue was built with. bound_bytes: usize, - /// The bytes admitted and not yet released, shared with the draining clone. - queued_bytes: Arc, - /// Set once a frame was dropped for the bound: from then on every send is dropped. - overflowed: Arc, + /// The socket's accounting, shared with every clone and with the draining task. + account: Arc, } impl FrameSink { - /// Creates the queue and its sink, admitting at most `bound_bytes` of queued frames. The - /// receiver is drained by the connection task, which [`FrameSink::release`]s each frame's - /// bytes once its write completes. - pub fn channel(bound_bytes: usize) -> (Self, mpsc::UnboundedReceiver) { + /// Creates the queue and its sink, admitting at most `bound_bytes` of queued frames. + /// + /// `budget` is the relay's process-wide bound, shared by every socket it serves: a frame + /// has to fit both. The receiver is drained by the connection task, which + /// [`FrameSink::release`]s each frame's bytes once its write completes. + pub fn channel( + bound_bytes: usize, + budget: Option>, + ) -> (Self, mpsc::UnboundedReceiver) { let (sender, receiver) = mpsc::unbounded_channel(); ( Self { sender: Some(sender), bound_bytes, - queued_bytes: Arc::new(AtomicUsize::new(0)), - overflowed: Arc::new(AtomicBool::new(false)), + account: Arc::new(Account { + queued_bytes: AtomicUsize::new(0), + refusal: AtomicU8::new(NO_REFUSAL), + budget, + }), }, receiver, ) @@ -68,37 +222,40 @@ impl FrameSink { Self { sender: None, bound_bytes: 0, - queued_bytes: Arc::new(AtomicUsize::new(0)), - overflowed: Arc::new(AtomicBool::new(false)), + account: Arc::new(Account { + queued_bytes: AtomicUsize::new(0), + refusal: AtomicU8::new(NO_REFUSAL), + budget: None, + }), } } /// Sends one frame. A closed socket drops the frame and is not an error. + /// + /// A queue-full refusal latches the socket shut, as the overflow always has: every later + /// send is dropped. A budget refusal does not — it is one frame that did not fit at that + /// instant, and the connection task closes the socket at its next iteration whatever the + /// sender does — so the accounting stays exact and a frame that fits later is admitted. pub fn send(&self, message: &RelayMessage) { let Some(sender) = &self.sender else { return; }; - if self.overflowed.load(Ordering::SeqCst) { - return; - } - let text = message.to_text(); let length = text.len(); - let mut queued = self.queued_bytes.load(Ordering::SeqCst); + let mut queued = self.account.queued_bytes.load(Ordering::SeqCst); // Admission is a compare-and-exchange, not a load-then-store: an in-flight read runs // in its own task and several readers share this sink, so two sends that interleaved // a load and a store would lose one increment and admit more than the bound. loop { - if self.overflowed.load(Ordering::SeqCst) { + if self.account.refusal() == Some(Refusal::QueueFull) { return; } - // An empty queue admits its frame whatever the size, so an event larger than the // whole bound still reaches the client instead of closing it. if queued > 0 && queued + length > self.bound_bytes { - self.overflowed.store(true, Ordering::SeqCst); + self.account.refuse(Refusal::QueueFull); tracing::warn!( bound_bytes = self.bound_bytes, queued_bytes = queued, @@ -107,7 +264,7 @@ impl FrameSink { return; } - match self.queued_bytes.compare_exchange_weak( + match self.account.queued_bytes.compare_exchange_weak( queued, queued + length, Ordering::SeqCst, @@ -119,29 +276,61 @@ impl FrameSink { } } + // The budget comes second so a socket that is already over its own bound refuses for + // that reason, as it did before this counter existed. It has no empty-queue + // exception: a frame that does not fit the budget is refused even on an idle socket. + if let Some(budget) = &self.account.budget + && !budget.admit(length) + { + // The frame is never queued, so the bytes counted a moment ago will never be + // released by the drain arm: take them back here, exactly what was counted. + let _ = self.account.queued_bytes.try_update( + Ordering::SeqCst, + Ordering::SeqCst, + |queued| Some(queued.saturating_sub(length)), + ); + + if self.account.refuse(Refusal::BudgetExhausted) { + budget.refuse(); + tracing::warn!( + budget_bytes = budget.limit(), + accounted_bytes = budget.accounted(), + "frame memory budget exhausted: dropping frames and closing the client" + ); + } + + return; + } + // The receiver is gone exactly when the connection task ended: the socket is not // open, and the capture's guard returns without sending. let _ = sender.send(text); } - /// Releases a frame's bytes once it has been written to the socket. + /// Releases a frame's bytes once it has been written to the socket, for this socket and + /// for the budget it counts against. pub(crate) fn release(&self, bytes: usize) { - let _ = self - .queued_bytes - .try_update(Ordering::SeqCst, Ordering::SeqCst, |queued| { - Some(queued.saturating_sub(bytes)) - }); + let _ = + self.account + .queued_bytes + .try_update(Ordering::SeqCst, Ordering::SeqCst, |queued| { + Some(queued.saturating_sub(bytes)) + }); + + if let Some(budget) = &self.account.budget { + budget.release(bytes); + } } - /// Whether a frame has been dropped for the bound. The connection task closes the client - /// once this is set. - pub fn overflowed(&self) -> bool { - self.overflowed.load(Ordering::SeqCst) + /// Why a frame was refused, if the connection is over its bound. The connection task + /// closes the client once this is set. + pub fn refusal(&self) -> Option { + self.account.refusal() } /// The bytes admitted into the queue and not yet written. pub fn queued_bytes(&self) -> usize { - self.queued_bytes.load(Ordering::SeqCst) + self.account.queued_bytes.load(Ordering::SeqCst) } } @@ -162,12 +351,12 @@ mod tests { let len = frame.to_text().len(); // Two frames fit exactly, and a third is dropped rather than queued. - let (sink, mut frames) = FrameSink::channel(2 * len); + let (sink, mut frames) = FrameSink::channel(2 * len, None); sink.send(&frame); sink.send(&frame); - assert!(!sink.overflowed()); + assert_eq!(sink.refusal(), None); assert_eq!(sink.queued_bytes(), 2 * len); assert!(frames.try_recv().is_ok()); assert!(frames.try_recv().is_ok()); @@ -175,7 +364,7 @@ mod tests { sink.send(&frame); - assert!(sink.overflowed()); + assert_eq!(sink.refusal(), Some(Refusal::QueueFull)); assert_eq!( sink.queued_bytes(), 2 * len, @@ -189,22 +378,22 @@ mod tests { // Once the flag is set every later send is dropped as well. sink.send(&frame); - assert!(sink.overflowed()); + assert_eq!(sink.refusal(), Some(Refusal::QueueFull)); assert_eq!(sink.queued_bytes(), 2 * len); // A frame larger than the bound still goes out while the queue is empty, so one // large event does not close every client. - let (sink, mut frames) = FrameSink::channel(len - 1); + let (sink, mut frames) = FrameSink::channel(len - 1, None); sink.send(&frame); - assert!(!sink.overflowed()); + assert_eq!(sink.refusal(), None); assert_eq!(sink.queued_bytes(), len); assert!(frames.try_recv().is_ok()); sink.send(&frame); - assert!(sink.overflowed()); + assert_eq!(sink.refusal(), Some(Refusal::QueueFull)); assert_eq!(sink.queued_bytes(), len); assert!(frames.try_recv().is_err()); @@ -214,7 +403,7 @@ mod tests { // needs, so this is a regression guard rather than a proof. const SENDERS: usize = 8; - let (sink, mut frames) = FrameSink::channel(SENDERS * len); + let (sink, mut frames) = FrameSink::channel(SENDERS * len, None); let start = std::sync::Barrier::new(SENDERS); std::thread::scope(|scope| { @@ -230,7 +419,7 @@ mod tests { } }); - assert!(!sink.overflowed(), "every frame fits the bound"); + assert_eq!(sink.refusal(), None, "every frame fits the bound"); assert_eq!(sink.queued_bytes(), SENDERS * len); for _ in 0..SENDERS { @@ -241,14 +430,14 @@ mod tests { // `release` is the drain side's half of the balance: bytes come back as a frame is // written, so a drained queue admits again and the count never drifts upwards. - let (sink, mut frames) = FrameSink::channel(2 * len); + let (sink, mut frames) = FrameSink::channel(2 * len, None); sink.send(&frame); sink.send(&frame); sink.release(2 * len); assert_eq!(sink.queued_bytes(), 0); - assert!(!sink.overflowed()); + assert_eq!(sink.refusal(), None); sink.send(&frame); sink.send(&frame); @@ -260,4 +449,166 @@ mod tests { assert!(frames.try_recv().is_ok()); assert!(frames.try_recv().is_err()); } + + #[test] + fn the_budget_binds_across_sockets_and_is_shared() { + let frame = notice("shared"); + let len = frame.to_text().len(); + let budget = Arc::new(FrameBudget::new(2 * len)); + + // Three sockets of a relay whose budget only two frames of this size fit in. + let (first, _first_frames) = FrameSink::channel(4 * len, Some(Arc::clone(&budget))); + let (second, _second_frames) = FrameSink::channel(4 * len, Some(Arc::clone(&budget))); + let (third, mut third_frames) = FrameSink::channel(4 * len, Some(Arc::clone(&budget))); + + first.send(&frame); + second.send(&frame); + + assert_eq!(budget.accounted(), budget.limit()); + assert!(budget.accounted() <= budget.limit()); + + third.send(&frame); + + assert_eq!(third.refusal(), Some(Refusal::BudgetExhausted)); + assert_eq!( + third.queued_bytes(), + 0, + "a refused frame is not counted against the socket either" + ); + assert_eq!( + budget.accounted(), + 2 * len, + "the refused frame is not accounted" + ); + assert_eq!(budget.refusals(), 1); + assert!(third_frames.try_recv().is_err()); + assert!(budget.accounted() <= budget.limit()); + + // The drain arm returning one written frame's bytes is all the room a waiting socket + // needs: the bound is bytes, not a queue slot. + first.release(len); + + third.send(&frame); + + assert_eq!(budget.accounted(), 2 * len); + assert_eq!( + budget.refusals(), + 1, + "the second frame needed no second refusal" + ); + assert_eq!(third.queued_bytes(), len); + assert!(third_frames.try_recv().is_ok()); + assert!(budget.accounted() <= budget.limit()); + } + + #[test] + fn the_budget_release_refunds_only_the_closing_sockets_bytes() { + let frame = notice("refunded"); + let len = frame.to_text().len(); + let budget = Arc::new(FrameBudget::new(4 * len)); + + let (first, _first_frames) = FrameSink::channel(len, Some(Arc::clone(&budget))); + let (second, _second_frames) = FrameSink::channel(len, Some(Arc::clone(&budget))); + + first.send(&frame); + second.send(&frame); + + assert_eq!(budget.accounted(), 2 * len); + assert_eq!(first.queued_bytes(), len); + assert_eq!(second.queued_bytes(), len); + + // Neither frame is released: this is the state of every exit from the connection loop + // except the drain arm, where the frames a socket admitted were never written. + drop(first); + + assert_eq!( + budget.accounted(), + len, + "the refund is the closing socket's bytes, not the whole counter" + ); + + drop(second); + + assert_eq!(budget.accounted(), 0); + + // Both handles are gone and nothing underflowed: the budget still admits what it + // should, which a race that subtracted the same bytes twice could not. + let (third, mut third_frames) = FrameSink::channel(len, Some(Arc::clone(&budget))); + + third.send(&frame); + + assert_eq!(third.refusal(), None); + assert_eq!(budget.accounted(), len); + assert!(third_frames.try_recv().is_ok()); + } + + #[test] + fn the_budget_is_never_over_committed_by_racing_senders() { + // The shape of the racing-sender guard above: a barrier makes the sends overlap but + // cannot force the interleaving a lost update needs, so this is a regression guard + // rather than a proof. + let frame = notice("raced"); + let len = frame.to_text().len(); + const SENDERS: usize = 8; + let budget = Arc::new(FrameBudget::new(SENDERS * len)); + let max_seen = AtomicUsize::new(0); + let start = std::sync::Barrier::new(SENDERS + 1); + + let mut sinks = Vec::new(); + + for _ in 0..SENDERS { + sinks.push(FrameSink::channel(len, Some(Arc::clone(&budget))).0); + } + + std::thread::scope(|scope| { + for sink in &sinks { + let frame = &frame; + let start = &start; + + scope.spawn(move || { + start.wait(); + sink.send(frame); + }); + } + + // A sampler reading the counter while the sends run: it stops when the budget is + // full, so it cannot outlive the sends it observes. + { + let budget = Arc::clone(&budget); + let max_seen = &max_seen; + let start = &start; + + scope.spawn(move || { + start.wait(); + + loop { + let accounted = budget.accounted(); + max_seen.fetch_max(accounted, Ordering::SeqCst); + + if accounted == SENDERS * len { + break; + } + + std::hint::spin_loop(); + } + }); + } + }); + + assert!(max_seen.load(Ordering::SeqCst) <= budget.limit()); + assert_eq!( + budget.accounted(), + SENDERS * len, + "every frame was admitted once" + ); + assert_eq!(budget.refusals(), 0); + + for sink in &sinks { + assert_eq!(sink.queued_bytes(), len); + } + + drop(sinks); + + assert_eq!(budget.accounted(), 0); + } } diff --git a/crates/nostream-relay/src/transport.rs b/crates/nostream-relay/src/transport.rs index e88706f8..b0080868 100644 --- a/crates/nostream-relay/src/transport.rs +++ b/crates/nostream-relay/src/transport.rs @@ -38,6 +38,14 @@ pub const DRAIN_REASON: &str = "relay shutting down"; /// own policy, with no counterpart in the capture (ADR 0005). pub const SLOW_CONSUMER_REASON: &str = "slow consumer: outbound frame queue full"; +/// The close reason a client whose outbound frame did not fit the relay's frame-memory budget +/// is closed with: the port's own policy, with no counterpart in the capture (ADR 0008). +pub const FRAME_BUDGET_REASON: &str = "frame memory budget exhausted"; + +/// The close reason a client above `max_connections_per_ip` is closed with: the port's own +/// policy, with no counterpart in the capture (ADR 0008). +pub const CONNECTION_CAP_REASON: &str = "connection cap: too many connections from this address"; + /// `network.maxPayloadSize` (`network.max_payload_size` in the shipped YAML), else 131072. pub fn max_payload_size(settings: &Settings) -> usize { settings diff --git a/crates/nostream-relay/tests/connection.rs b/crates/nostream-relay/tests/connection.rs index 5b6ee276..1654e8f6 100644 --- a/crates/nostream-relay/tests/connection.rs +++ b/crates/nostream-relay/tests/connection.rs @@ -411,7 +411,7 @@ async fn closing_the_socket_aborts_an_in_flight_read() { .await; // The queue's bound plays no part here: the frames are a few hundred bytes. - let (sink, mut frames) = FrameSink::channel(1024 * 1024); + let (sink, mut frames) = FrameSink::channel(1024 * 1024, None); let connection = Arc::new(Connection::new( "127.0.0.1".to_owned(), nostream_core::nip42::Nip42Session::with_challenge("challenge", 0), @@ -493,7 +493,7 @@ async fn restricted_read_with(authenticate_mid_read: bool) -> Vec { )) .await; - let (sink, mut frames) = FrameSink::channel(1024 * 1024); + let (sink, mut frames) = FrameSink::channel(1024 * 1024, None); // The session outlives the read: this connection is built directly rather than by // `Connection::accept`, so the TTL is the one named here and the settings' expiry (a // zero TTL) plays no part. @@ -621,7 +621,7 @@ async fn a_restricted_read_is_judged_against_the_authenticated_set_at_each_event async fn a_handler_prefix_runs_before_its_continuation_is_polled() { let fixture = Fixture::new(settings()); let connection = connection(); - let (sink, mut frames) = FrameSink::channel(1024 * 1024); + let (sink, mut frames) = FrameSink::channel(1024 * 1024, None); let context = Arc::new(fixture.context( Arc::new(TestClock::new(NOW_MS)), relay_keys(), diff --git a/crates/nostream-relay/tests/limits.rs b/crates/nostream-relay/tests/limits.rs index 80263b24..f0431175 100644 --- a/crates/nostream-relay/tests/limits.rs +++ b/crates/nostream-relay/tests/limits.rs @@ -969,7 +969,7 @@ async fn a_failing_event_limiter_limits_the_event_without_a_backoff() { ..Default::default() })); - let (sink, mut frames) = nostream_relay::sink::FrameSink::channel(1024 * 1024); + let (sink, mut frames) = nostream_relay::sink::FrameSink::channel(1024 * 1024, None); let clock = Arc::new(TestClock::new(NOW_MS)); let connection = Arc::new(nostream_relay::connection::Connection::new( "127.0.0.1".to_owned(), diff --git a/crates/nostream-relay/tests/strategies.rs b/crates/nostream-relay/tests/strategies.rs index 83cc663f..0fbb605e 100644 --- a/crates/nostream-relay/tests/strategies.rs +++ b/crates/nostream-relay/tests/strategies.rs @@ -100,7 +100,7 @@ fn keys() -> RelayKeySource { /// A context whose sink a test can read, with its frames. fn socket(fixture: &Fixture) -> (HandlerContext, mpsc::UnboundedReceiver) { - let (sink, frames) = FrameSink::channel(64 * 1024); + let (sink, frames) = FrameSink::channel(64 * 1024, None); let connection = Arc::new(Connection::new( "127.0.0.1".to_owned(), Nip42Session::with_challenge("challenge", 0), diff --git a/crates/nostream-relay/tests/transport.rs b/crates/nostream-relay/tests/transport.rs index f3da1d54..76327fbb 100644 --- a/crates/nostream-relay/tests/transport.rs +++ b/crates/nostream-relay/tests/transport.rs @@ -21,6 +21,7 @@ use common::{ CLIENT_SECRET, Client, NOW_SECS, Server, fast_config, liveness_config, relay_keys, signed_event, }; use futures_util::SinkExt; +use nostream_relay::RelayConfig; use nostream_relay::testing::{Fixture, settings}; use nostream_relay::transport::{self, DEFAULT_MAX_PAYLOAD}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -244,6 +245,16 @@ fn close_of(frame: &Frame) -> (u16, String) { ) } +/// Asserts that nothing was written to the client before its close frame: a frame the client +/// will never read cannot be the way to say why it is being closed. +fn assert_no_text_frame(frames: &[Frame]) { + assert!( + frames.iter().all(|frame| frame.opcode == 0x8), + "a refusal sends no text frame: {:?}", + frames.iter().map(Frame::text).collect::>() + ); +} + fn settings_with_payload(size: u64) -> nostream_config::settings::Settings { let mut settings = settings(); settings.network.max_payload_size = Some(size); @@ -251,6 +262,43 @@ fn settings_with_payload(size: u64) -> nostream_config::settings::Settings { settings } +/// A config with only the two bounds changed, on the fast intervals the suite runs with. +fn bounded_config(frame_memory_budget_bytes: usize, max_connections_per_ip: usize) -> RelayConfig { + RelayConfig { + frame_memory_budget_bytes, + max_connections_per_ip, + ..fast_config() + } +} + +/// Waits up to two seconds for `condition`: a closing socket's slot and bytes are given back +/// by its own task, which the client's close does not wait for. +async fn eventually(condition: impl Fn() -> bool) -> bool { + for _ in 0..200 { + if condition() { + return true; + } + + tokio::time::sleep(Duration::from_millis(10)).await; + } + + condition() +} + +/// The next text frame's payload, ignoring any other frame before it. +async fn next_text(client: &mut RawClient) -> String { + loop { + let frame = tokio::time::timeout(Duration::from_secs(5), client.next_frame()) + .await + .expect("a frame arrives") + .expect("the stream is open"); + + if frame.opcode == 0x1 { + return frame.text(); + } + } +} + // --------------------------------------------------------------------------------------- // The payload limit (R27) // --------------------------------------------------------------------------------------- @@ -617,6 +665,13 @@ async fn a_client_that_stops_reading_is_closed_with_1013() { } let server = Server::start_with(fixture, relay_keys(), fast_config()).await; + + assert_eq!( + server.relay.frame_budget().accounted(), + 0, + "nothing is counted before this client asks for anything" + ); + let (mut raw, _head) = RawClient::connect(server.address, None).await; // The socket's first frame is the NIP-42 challenge, which is a text frame on every @@ -681,4 +736,223 @@ async fn a_client_that_stops_reading_is_closed_with_1013() { close_of(last), (1013, transport::SLOW_CONSUMER_REASON.to_owned()) ); + + // This exit is the loop's overflow arm, not the drain arm's `release`: the frames this + // socket admitted and never wrote were still counted when the connection ended, so the + // refund on its last sink is the only thing that gives them back. This is the + // behaviour-level form of the criterion — a subsequent connection is admitted — and does + // not by itself tell a refund from a counter that happened to be zero; + // `sink.rs::the_budget_release_refunds_only_the_closing_sockets_bytes` is the test that does. + assert!( + eventually(|| server.relay.frame_budget().accounted() == 0).await, + "a closed socket's bytes return to the budget, found {}", + server.relay.frame_budget().accounted() + ); + + let (mut fresh, _head) = RawClient::connect(server.address, None).await; + let auth = tokio::time::timeout(Duration::from_secs(5), fresh.next_frame()) + .await + .expect("the fresh client is served") + .expect("the fresh client's challenge"); + + assert_eq!(auth.opcode, 0x1); + assert!( + auth.text().starts_with(r#"["AUTH","#), + "a fresh client is served after the refund: {}", + auth.text() + ); +} + +// --------------------------------------------------------------------------------------- +// The frame-memory budget and the per-IP cap (#959, ADR 0008) +// --------------------------------------------------------------------------------------- + +#[tokio::test] +async fn the_default_bounds_are_the_policy_numbers() { + let config = RelayConfig::default(); + + assert_eq!(config.frame_memory_budget_bytes, 268_435_456); + assert_eq!(config.max_connections_per_ip, 64); + + let server = Server::start_with(Fixture::new(settings()), relay_keys(), config).await; + + assert_eq!( + server.relay.frame_budget().limit(), + RelayConfig::default().frame_memory_budget_bytes + ); + assert_eq!( + server.relay.connection_cap().max_per_ip(), + RelayConfig::default().max_connections_per_ip + ); + assert_eq!(server.relay.frame_budget().accounted(), 0); + assert_eq!(server.relay.frame_budget().refusals(), 0); +} + +#[tokio::test] +async fn a_socket_above_the_per_ip_cap_is_closed_with_1013_and_no_text_frame() { + let server = Server::start_with( + Fixture::new(settings()), + relay_keys(), + bounded_config(256 * 1024 * 1024, 2), + ) + .await; + + let (mut first, first_head) = RawClient::connect(server.address, None).await; + let (mut second, second_head) = RawClient::connect(server.address, None).await; + + // Both admitted sockets get the challenge, which is the relay's first text frame. + for head in [&first_head, &second_head] { + assert!(head.starts_with("HTTP/1.1 101"), "{head}"); + } + + for client in [&mut first, &mut second] { + let challenge = tokio::time::timeout(Duration::from_secs(5), client.next_frame()) + .await + .expect("a challenge arrives") + .expect("a challenge frame"); + + assert_eq!(challenge.opcode, 0x1); + assert!(challenge.text().starts_with(r#"["AUTH","#)); + } + + let (mut refused, refused_head) = RawClient::connect(server.address, None).await; + + assert!(refused_head.starts_with("HTTP/1.1 101"), "{refused_head}"); + + let frames = tokio::time::timeout(Duration::from_secs(5), refused.frames_until(0x8)) + .await + .expect("the refused socket is closed"); + + assert_no_text_frame(&frames); + + assert_eq!(frames.len(), 1, "exactly one frame: the close: {frames:?}"); + assert_eq!(frames[0].opcode, 0x8); + assert_eq!( + close_of(&frames[0]), + (1013, transport::CONNECTION_CAP_REASON.to_owned()) + ); + + // The two admitted sockets are still serving. + first.send_text(r#"["REQ","s",{"kinds":[1]}]"#).await; + + assert_eq!(next_text(&mut first).await, r#"["EOSE","s"]"#); + + assert_eq!(server.relay.connection_cap().refusals(), 1); + assert_eq!(server.relay.connection_cap().live("127.0.0.1"), 2); +} + +#[tokio::test] +async fn the_per_ip_cap_releases_a_slot_when_a_socket_closes() { + let server = Server::start_with( + Fixture::new(settings()), + relay_keys(), + bounded_config(256 * 1024 * 1024, 1), + ) + .await; + + let (mut first, head) = RawClient::connect(server.address, None).await; + + assert!(head.starts_with("HTTP/1.1 101"), "{head}"); + assert!(next_text(&mut first).await.starts_with(r#"["AUTH","#)); + + let (mut second, _head) = RawClient::connect(server.address, None).await; + let frames = tokio::time::timeout(Duration::from_secs(5), second.frames_until(0x8)) + .await + .expect("the second socket is closed"); + + assert_no_text_frame(&frames); + + assert_eq!( + close_of(frames.last().expect("a close frame")), + (1013, transport::CONNECTION_CAP_REASON.to_owned()) + ); + assert_eq!(server.relay.connection_cap().live("127.0.0.1"), 1); + + drop(first); + + assert!( + eventually(|| server.relay.connection_cap().live("127.0.0.1") == 0).await, + "a closed socket's slot is released" + ); + assert_eq!( + server.relay.connection_cap().tracked_addresses(), + 0, + "an address that released its last slot leaves no entry" + ); + + let (mut third, head) = RawClient::connect(server.address, None).await; + + assert!(head.starts_with("HTTP/1.1 101"), "{head}"); + assert!(next_text(&mut third).await.starts_with(r#"["AUTH","#)); + assert_eq!( + server.relay.connection_cap().refusals(), + 1, + "the released slot admits again without a second refusal" + ); +} + +#[tokio::test] +async fn the_per_ip_cap_does_not_lapse_with_time() { + let server = Server::start_with( + Fixture::new(settings()), + relay_keys(), + bounded_config(256 * 1024 * 1024, 2), + ) + .await; + + let (mut first, _head) = RawClient::connect(server.address, None).await; + let (mut second, _head) = RawClient::connect(server.address, None).await; + + assert!(next_text(&mut first).await.starts_with(r#"["AUTH","#)); + assert!(next_text(&mut second).await.starts_with(r#"["AUTH","#)); + + // Past every rate window the shipped limits use, and past the liveness interval too: the + // cap is a count of live sockets, so time passing changes nothing. + server.clock.advance(3_600_000); + + let (mut refused, _head) = RawClient::connect(server.address, None).await; + let frames = tokio::time::timeout(Duration::from_secs(5), refused.frames_until(0x8)) + .await + .expect("the third socket is closed"); + + assert_no_text_frame(&frames); + + assert_eq!( + close_of(frames.last().expect("a close frame")), + (1013, transport::CONNECTION_CAP_REASON.to_owned()) + ); + assert_eq!(server.relay.connection_cap().live("127.0.0.1"), 2); +} + +#[tokio::test] +async fn a_frame_that_exceeds_the_budget_closes_the_socket_with_1013_and_no_text_frame() { + // One byte of budget: not even the AUTH challenge fits, so the very first frame a socket + // would send is refused and nothing is ever queued. + let server = Server::start_with( + Fixture::new(settings()), + relay_keys(), + bounded_config(1, 64), + ) + .await; + + for refusals in 1..=2 { + let (mut client, head) = RawClient::connect(server.address, None).await; + + assert!(head.starts_with("HTTP/1.1 101"), "{head}"); + + let frames = tokio::time::timeout(Duration::from_secs(5), client.frames_until(0x8)) + .await + .expect("the refused socket is closed"); + + assert_no_text_frame(&frames); + + assert_eq!(frames.len(), 1, "exactly one frame: the close: {frames:?}"); + assert_eq!(frames[0].opcode, 0x8); + assert_eq!( + close_of(&frames[0]), + (1013, transport::FRAME_BUDGET_REASON.to_owned()) + ); + assert_eq!(server.relay.frame_budget().refusals(), refusals); + assert_eq!(server.relay.frame_budget().accounted(), 0); + } } From 7f1635140f49e9557b89acc9c72efe3ac0481147 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 24 Oct 2026 23:42:35 -0400 Subject: [PATCH 02/12] docs(rewrite): record the frame-memory budget and the per-IP cap (ADR 0008) Both numbers are deviations from the capture, with their arithmetic, the rejected accept-time shape, the refund on the last handle's Drop, and what an operator can see today (#959). --- .../decisions/0008-frame-memory-budget.md | 229 ++++++++++++++++++ 1 file changed, 229 insertions(+) create mode 100644 docs/rewrite/decisions/0008-frame-memory-budget.md diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md new file mode 100644 index 00000000..762c0188 --- /dev/null +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -0,0 +1,229 @@ +# ADR 0008: A global frame-memory budget and a per-IP live-connection cap (job 016) + +- **Status:** accepted (job 016, Phase 2) +- **Deciders:** the rewrite team, on the operator's policy comment on #959 and the evidence in job 016's recon and spec +- **Date:** 2026-10-24 + +## Context + +The outbound frame queue was bounded per socket and nothing bounded their sum. +`RelayConfig::frame_queue_bytes` (8 MiB; `crates/nostream-relay/src/relay.rs:57`) is enforced by +`FrameSink::send`'s compare-and-exchange on the socket's own counter +(`crates/nostream-relay/src/sink.rs:239-322`), and `Relay::serve` accepted every connection with +no count at all. The only per-address bookkeeping was the **attempt** limiter, which throttles how +fast connections open, not how many are held (`crates/nostream-relay/src/limits.rs`, windows +consumed at `crates/nostream-relay/src/relay.rs:327`). + +The operator's policy comment on #959 chose the numbers and the refusal and asked for two +consequences a reviewer can hold the change to: both bounds refuse the same way, and the budget is +released when a socket closes ("a leak here is worse than no bound"). + +**What the change is worth, measured.** One address that opens sockets and then stops reading held +`N × 8 MiB`. With the shipped attempt windows — 12 per 1 000 ms and 48 per 60 000 ms +(`resources/default-settings.yaml:215-224`) — one address gets 48 sockets a minute indefinitely, so +the pre-change worst case is **384 MiB**, and that figure is **arithmetic from named constants, not a measurement**; it is the same derivation the performance lens made, marked there as arithmetic. + +**The failure the counter exists to avoid was measured on the wire.** Twenty sockets opened back to +back on a fresh daemon: the first twelve received `HTTP/1.1 101 Switching Protocols` and then a text +frame starting `["AUTH","…`, and the remaining eight were upgraded and dropped with **no websocket +close frame at all** — the attempt limiter's silent return +(`crates/nostream-relay/src/relay.rs:287-290`). A refusal indistinguishable from a network fault is +exactly what the new bounds are not allowed to be, so each of them closes with a code, a fixed +reason and its own counter. + +## The two numbers, as deviations from the capture, with the arithmetic + +The capture has neither bound. Its connection limits are `ConnectionLimits` — `rateLimits`, +`ipWhitelist`, `ipBlacklist` and no count (`src/@types/settings.ts:163-167`) — and its outbound +queue is unbounded (`src/adapters/web-socket-adapter.ts:150-155`). Both numbers below are therefore +deviations from the capture, and both are correct because the operator's comment names them, not +because a captured transcript shows them. There is no parity oracle for either: `wire_parity` cannot +see a bound the capture does not have. + +| knob | default | the arithmetic | +|---|---|---| +| `frame_memory_budget_bytes` | **256 MiB** (`268_435_456` bytes) | 32 sockets at the 8 MiB `frame_queue_bytes` ceiling can be fully buffered before the budget binds — below the 48 a single address gets through the shipped attempt limits alone | +| `max_connections_per_ip` | **64** sockets | four times the concurrent total one well-behaved client needs, and above the 48 the attempt limiter alone allows in a minute | +| refusal | `1013`, immediate, no text frame | a frame the client will never read cannot be the way to say "you are not reading" | + +The cap's own worst case is 64 × 8 MiB = **512 MiB**, deliberately **above** the budget. That is the +policy's own division of labour restated: the cap bounds sockets, the budget bounds bytes, and a +process whose sockets are all small can hold more of them than the budget-implied 32 without being +refused for it. Neither number is a total-memory ceiling; see the next section. + +## What "frame memory" means here + +Accounted **outbound payload bytes**: the `text.len()` that `frame_queue_bytes` already counts +(`crates/nostream-relay/src/sink.rs:244-245`), not `String` capacity, not the channel node, and not +resident memory. The nearest measurement of the ratio is job 010's probe D, which saw roughly 2 KiB +resident per 1 KiB queued at 1 KiB frames, so **256 MiB accounted is not a process RSS ceiling** and +should not be read as one. + +Nothing else is counted. Inbound is bounded per message by `max_payload_size` (131 072 by default, +`crates/nostream-relay/src/transport.rs:50-57`), handler state is not counted at all, and the +hidden-content mirror and the event stores are outside the budget on purpose: they are not frames, +and folding them in would make one knob mean two things. + +## The config surface — `RelayConfig`, not settings + +Both fields are `RelayConfig` fields, `frame_memory_budget_bytes` at +`crates/nostream-relay/src/relay.rs:65` and `max_connections_per_ip` at `:69`, built into the relay's +two shared values in `Relay::new` (`:146-147`). ADR 0005 already decided the question for this class +of knob: + +> The bound and the timeout are `RelayConfig` fields and not settings. The capture has no +> counterpart for either … so a settings key would be an operator-facing surface mirroring nothing. +> `RelayConfig` is where this branch already keeps the knobs the capture fixes in code … + +— `docs/rewrite/decisions/0005-connection-concurrency.md:104-110` + +**The honest cost:** an operator cannot change either number without a rebuild. "Configurable" here +means the value is readable and settable through a `RelayConfig` literal rather than buried in a +call site — a settings key, a YAML entry and a validator/guided-catalogue entry are **#989** and out +of scope for this job. Say it plainly rather than letting "configurable" read as a knob an operator +can edit in production; today they cannot. + +`nostreamd` keeps `..Default::default()` in its `RelayConfig` literal +(`crates/nostreamd/src/main.rs:120-123`), and all five `Relay::new` call sites already did, so no +call site had to change. + +## The refusal — `1013`, immediate, no text frame + +Both bounds close the socket with `CloseCode::Again` (1013) written straight to the transport rather +than through the sink, bounded by `config.write_timeout` exactly as the existing slow-consumer close +is (`Relay::close_frame`, `crates/nostream-relay/src/relay.rs:524-535`). Three fixed reason strings +now exist, each lowercase and resource-stating, per the convention +`SLOW_CONSUMER_REASON` set (`crates/nostream-relay/src/transport.rs:37-47`): + +- `SLOW_CONSUMER_REASON` — "slow consumer: outbound frame queue full", unchanged. +- `FRAME_BUDGET_REASON` — "frame memory budget exhausted". +- `CONNECTION_CAP_REASON` — "connection cap: too many connections from this address". + +**Why not a `NOTICE`.** A NOTICE is a text frame, and an over-budget client is by definition one that +is not reading: the frame would sit in the very queue being refused, and a client at the cap has not +earned a JSON frame it did not ask for. The refusal has to be legible in the close frame, which every +websocket client reads, rather than in a payload it may never consume. + +**Neither refusal is a timeout, and neither is the attempt limiter's.** The attempt limiter's refusal +is still the silent upgrade-then-drop measured above; this job leaves `RateLimits`, its windows and +its whitelist untouched. The two new refusals are distinguishable from it and from each other by +code, reason string, and one counter each (`FrameBudget::refusals`, +`crates/nostream-relay/src/sink.rs:90`; `ConnectionCap::refusals`, +`crates/nostream-relay/src/connection_cap.rs:60`), which is what keeps them out of the "looks like a +network fault" bucket that started this. + +**A budget refusal refuses the frame, not the socket.** The refusal cause is recorded once — so the +counter counts one refused socket, not one per frame, and the warn line is written once — but only a +queue-full refusal latches the sink shut. A budget refusal that latched would also have to be +un-latched when the drain side frees room, and it buys nothing: the connection loop closes the socket +at its next iteration whatever the sender does. The accounting therefore stays exact under either +choice, and the unit test that proves room lets the next frame through +(`sink.rs::the_budget_binds_across_sockets_and_is_shared`) can say so. + +## Where the budget binds: per-frame admission, not an accept-time reservation + +The budget is enforced where the per-socket bound already is: in `FrameSink::send`, as a second +compare-and-exchange loop against `accounted + length > limit` +(`crates/nostream-relay/src/sink.rs:99-135`). Admission requires room in both bounds; the budget has +**no empty-queue exception** — unlike `frame_queue_bytes`, whose exception keeps one large event from +closing every client — because 256 MiB exceeds any single outbound frame the protocol can produce by +three orders of magnitude, and strictness is what makes a one-byte test budget meaningful. + +**The rejected shape: an accept-time reservation.** Refusing new sockets when the budget is already +full is the reading a "global budget" invites, and it is wrong for two reasons. It does not bound what +an admitted socket may queue — a socket admitted into an empty budget can still queue its 8 MiB, so +the bound has to exist at the sink regardless, and adding an accept-time gate buys a second bound +beside it rather than replacing it. And a reservation needs its own lifecycle: take bytes at accept, +refund them on every exit, and get the partial case right when a socket used less than it reserved — +which is a second refund path of exactly the kind section 7 exists to eliminate. Per-frame admission +has one lifecycle (bytes are counted when a frame is admitted, uncounted when it is written or when +the socket's accounting ends) and refuses the frames that actually need the memory. + +## The refund, and the shape rejected + +**Chosen: refund on the last handle's `Drop`.** `FrameSink` holds an `Arc` +(`crates/nostream-relay/src/sink.rs:185-191`); the account carries the socket's outstanding bytes, +its refusal cause and the shared budget, and `impl Drop for Account` +(`crates/nostream-relay/src/sink.rs:175-181`) returns whatever is left of that socket's bytes to the +budget. Every clone of the socket's sink shares the account, so the drop that runs is the last one: a +send holds a sink and therefore cannot be in flight or begin once that drop starts. That is what makes +the refund **exactly this socket's bytes, once**, whatever the connection loop did on the way out. + +**The rejected shape: a decrement at each of the loop's exits.** Today `release` has exactly one +caller, the drain arm (`crates/nostream-relay/src/relay.rs:407`), and every other exit leaves whatever +was still queued counted: the overflow close (`:391-394`), the `None` receiver (`:398`), a failed +write (`:409-411`), the inbound and close arms (`:414`, `:433-435`, `:440`), the lagged hub (`:460`), the +liveness arms (`:464-469`, `:479-481`) and the final cleanup (`:488-489`). Per socket that leak is invisible +because the counter dies with the sink; a process-wide budget makes it a leak of up to 8 MiB on every +disconnect. Writing the decrement at each of those ten sites is the change that looks right and rots: +the next exit added to the loop silently leaks, and no unit test can reach the exits that need a +failed handshake or a poisoned store to trigger. `Drop` covers all of them by construction, including +the registration failure inside `run_connection` — which matters, because a refund that ran only on +the happy exit would fail AC8's test. + +The drain arm's `release` is unchanged and still runs: it returns written bytes promptly, so the +budget is reusable while the socket lives, and the `Drop` refund then finds only what was never +written. + +## The per-IP cap + +A `ConnectionCap` (`crates/nostream-relay/src/connection_cap.rs`) counts **live sockets** per +`peer.ip().to_string()` — the same string the attempt limiter's buckets are keyed on — and is +checked in `handle_socket` **after** `is_connection_rate_limited` and **before** `run_connection`: + +- after the limiter, so a rate-limited socket is still dropped silently as it always was, takes no + slot, and needs no refund on that path; +- before `run_connection`, because `Connection::accept`'s first act is to send the NIP-42 challenge + as a text frame (`crates/nostream-relay/src/connection.rs:109-115`) — a client refused afterwards + would already have received a text frame, which the policy forbids. The window is + `crates/nostream-relay/src/relay.rs:292-312`. + +Acquisition returns a `ConnectionSlot` held for the rest of `handle_socket` +(`crates/nostream-relay/src/relay.rs:299`), so the normal exit, the registration failure inside +`run_connection`, and any early return added later all release it through `Drop` +(`crates/nostream-relay/src/connection_cap.rs:101-116`). A count that leaks makes the cap permanent +for that address, which is a defect and not a memory leak. The map entry is removed when its count +reaches zero (`:111-114`), so an address that once connected leaves nothing behind and the map is +bounded by live addresses. + +The cap applies to **every address, including whitelisted ones**. An exempt address would bypass the +memory bound the cap protects, and the limiter's `ipWhitelist` behaviour is untouched: the whitelist +answers "skip the rate window", not "hold unlimited sockets". + +This is a gauge, not a window: no store, no clock, no expiry. `RateLimits` and `limits.rs` are +unchanged, and the measured 20-socket behaviour — 12 served, 8 dropped by the limiter — is unchanged +with it. + +## What an operator can see today + +Two accessors in the `hidden_content()` style, `Relay::frame_budget()` and +`Relay::connection_cap()` (`crates/nostream-relay/src/relay.rs:175-182`), and the counters behind +them: `accounted()`, `limit()`, `refusals()` on the budget; `live(ip)`, `max_per_ip()`, +`refusals()`, `tracked_addresses()` on the cap. Plus two structured log lines, one per refusal — a +`warn!` carrying `budget_bytes` and `accounted_bytes` from the sink +(`crates/nostream-relay/src/sink.rs:294-298`), and a `debug!` carrying `peer`, +`max_connections_per_ip` and `live` from the connection task +(`crates/nostream-relay/src/relay.rs:300-305`). + +That is the whole surface, because **a Prometheus endpoint is out of reach right now**: the OTLP +exporter is a stub returning `NotImplemented` (`crates/nostreamd/src/telemetry.rs:47-52`) and the +admin `/metrics` controller is a Phase-1 stub (`crates/nostream-admin/src/controllers/get_metrics.rs:5-9`). +There is no registry to publish into. When `/metrics` lands, an exporter reads these two counters and +nothing here needs to change; until then the counters and the two log lines are the deliverable, and +this job deliberately adds no metrics subsystem. + +## What this does not decide + +- **The operator-facing settings key** for either number: #989, with the YAML entry, the validator + and the guided catalogue it drags in. +- **Any exporter, dashboard or alerting.** No registry, no `/metrics` controller, no admin or + telemetry change. +- **Anything per listener.** One `Relay` carries one budget, and a second listener served by the same + `Arc` shares it (`Acceptor::run` calls `relay.serve(listener)`, + `crates/nostreamd/src/acceptor.rs:56-60`). A `static` would give the same sharing and cost the test + suite the ability to run two relays with different budgets in one process; two `Relay::new` calls + are the only way to get two budgets, and production makes one. +- **The per-socket bound's own empty-queue exception**, its default, and `SLOW_CONSUMER_REASON`. +- **Inbound memory or handler state**, neither of which the budget counts. +- **Any validation of the two numbers.** A zero budget or a zero cap is a literal the operator wrote; + policing it is a settings decision this job does not own. From bb71b72745cd44f204a1ef515c53dd63a201b057 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 24 Oct 2026 23:44:57 -0400 Subject: [PATCH 03/12] docs(relay): name the two new knobs where RelayConfig's list is, and the budget in the sink doc The RelayConfig doc enumerates the port-only policies, so it has to name all four; the sink's module doc is where a reader learns what the sum is bounded by. --- crates/nostream-relay/src/relay.rs | 8 ++++---- crates/nostream-relay/src/sink.rs | 10 +++++----- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index ce6722e3..a5b9d9e1 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -39,10 +39,10 @@ use crate::wot::WotGraph; /// The knobs a [`Relay`] is built with that are not settings: a constant the capture fixes in /// code (`liveness`, the NIP-56 refresh) or a policy the port adds with no counterpart there -/// (`frame_queue_bytes`, `write_timeout`, the frame-memory budget and the per-IP connection -/// cap; ADR 0005, ADR 0008). Tests set the cadences they need (`tests/common/mod.rs::fast_config`); -/// the policies run at their shipped defaults except where a test sets one deliberately. A -/// value the capture's settings document carries belongs in [`Settings`], never here. +/// (`frame_queue_bytes`, `write_timeout`, `frame_memory_budget_bytes`, +/// `max_connections_per_ip`; ADR 0005, ADR 0008). Tests set the cadences they need +/// (`tests/common/mod.rs::fast_config`); the policies run at their shipped defaults. A value +/// the capture's settings document carries belongs in [`Settings`], never here. #[derive(Debug, Clone)] pub struct RelayConfig { /// `WSS_CLIENT_HEALTH_PROBE_INTERVAL` (`web-socket-server-adapter.ts:18`). diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index dfb236d1..f7c3fb06 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -27,11 +27,11 @@ //! //! **The sum is bounded too**, which the per-socket bound cannot do: one address that opens //! sockets and stops reading held a multiple of the per-socket bound, so every socket's -//! queue is also counted against one process-wide [`FrameBudget`] (ADR 0008). An admission -//! needs room in both bounds; the budget has no empty-queue exception, so a frame that does -//! not fit the budget is refused however empty its own queue. The socket's share of the -//! budget is refunded when its last sink handle drops — see [`Account`] — because the next -//! socket's admission depends on it. +//! queue also counts against one process-wide frame memory budget ([`FrameBudget`], +//! ADR 0008). An admission needs room in both bounds; the budget has no empty-queue +//! exception, so a frame that does not fit the budget is refused however empty its own +//! queue. The socket's share of the budget is refunded when its last sink handle drops — +//! see [`Account`] — because the next socket's admission depends on it. use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU64, AtomicUsize, Ordering}; From 74ff50d2b2494723e4fa4587d89f0293651fd5a2 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sat, 24 Oct 2026 23:45:00 -0400 Subject: [PATCH 04/12] docs(rewrite): correct ADR 0008's line citations to the final tree --- docs/rewrite/decisions/0008-frame-memory-budget.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index 762c0188..aac2e086 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -176,7 +176,7 @@ checked in `handle_socket` **after** `is_connection_rate_limited` and **before** - before `run_connection`, because `Connection::accept`'s first act is to send the NIP-42 challenge as a text frame (`crates/nostream-relay/src/connection.rs:109-115`) — a client refused afterwards would already have received a text frame, which the policy forbids. The window is - `crates/nostream-relay/src/relay.rs:292-312`. + `crates/nostream-relay/src/relay.rs:292-315`. Acquisition returns a `ConnectionSlot` held for the rest of `handle_socket` (`crates/nostream-relay/src/relay.rs:299`), so the normal exit, the registration failure inside @@ -201,7 +201,7 @@ Two accessors in the `hidden_content()` style, `Relay::frame_budget()` and them: `accounted()`, `limit()`, `refusals()` on the budget; `live(ip)`, `max_per_ip()`, `refusals()`, `tracked_addresses()` on the cap. Plus two structured log lines, one per refusal — a `warn!` carrying `budget_bytes` and `accounted_bytes` from the sink -(`crates/nostream-relay/src/sink.rs:294-298`), and a `debug!` carrying `peer`, +(`crates/nostream-relay/src/sink.rs:295-299`), and a `debug!` carrying `peer`, `max_connections_per_ip` and `live` from the connection task (`crates/nostream-relay/src/relay.rs:300-305`). From 8ae2009a3c6e98e0d39ec6035680a0e2c70553bc Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 00:00:25 -0400 Subject: [PATCH 05/12] fix(relay): wake the connection loop on a refusal, and encode only frames it can queue Three static-review findings on the frame-memory budget, in one commit because they meet in FrameSink::send: - A refusal recorded a flag but woke nothing. A shared-budget refusal can arrive while this socket's own queue is empty, the socket idle and its read parked, so the 1013 waited for the next liveness tick instead of being immediate. The first refusal now signals a tokio::Notify the account owns, and the connection loop selects on FrameSink::refused(); notify_one stores a permit, so a refusal recorded between the loop's check and its select is not lost. - The queue-full check ran after to_text(), so a latching socket kept serialising events nobody would read. The early return is back before the encoding, with the in-loop check kept for racing senders. - The racing-sender test's sampler stopped only at the expected total, so the accounting bug it exists to catch could spin forever inside thread::scope. It now stops when the last sender finishes, and fails with numbers. The new wire test holds the budget from a socket that never reads and loses a frame on a socket whose own queue is empty: it times out without the signal and closes in 1.3 s with it. --- crates/nostream-relay/src/relay.rs | 6 ++ crates/nostream-relay/src/sink.rs | 66 ++++++++++++----- crates/nostream-relay/tests/transport.rs | 91 ++++++++++++++++++++++++ 3 files changed, 147 insertions(+), 16 deletions(-) diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index a5b9d9e1..1b733954 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -481,6 +481,12 @@ impl Relay { } } _ = drained.changed() => drain_seen = *drained.borrow_and_update(), + // A refusal can arrive with nothing else to wake this loop: the budget is + // shared, so the frame that did not fit may have left this socket's own queue + // empty, the socket itself idle, and its read parked on the store. The + // producer signals when it records the refusal, and the next iteration's + // check closes the client — immediate, not one liveness tick away (ADR 0008). + _ = sink.refused() => {} Some(_) = handlers.join_next(), if !handlers.is_empty() => {} } } diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index f7c3fb06..9cc26072 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -33,10 +33,11 @@ //! queue. The socket's share of the budget is refunded when its last sink handle drops — //! see [`Account`] — because the next socket's admission depends on it. +use std::future::Future; use std::sync::Arc; use std::sync::atomic::{AtomicU8, AtomicU64, AtomicUsize, Ordering}; -use tokio::sync::mpsc; +use tokio::sync::{Notify, mpsc}; use nostream_core::messages::RelayMessage; @@ -146,6 +147,14 @@ impl FrameBudget { struct Account { queued_bytes: AtomicUsize, refusal: AtomicU8, + /// Resolved when the first refusal is recorded. A refusal can arrive while this socket's + /// own queue is empty — the budget is shared, so another socket's frames are what left no + /// room — and the connection task may then have nothing else to wake it: signalling here + /// is what makes the close immediate rather than the next liveness tick. + /// + /// `notify_one` stores a permit when nothing is waiting, so a refusal recorded between the + /// loop's check and its `select!` is not lost, and two racing refusals leave one permit. + refusal_signal: Notify, budget: Option>, } @@ -166,9 +175,16 @@ impl Account { Refusal::BudgetExhausted => BUDGET_EXHAUSTED, }; - self.refusal + let first = self + .refusal .compare_exchange(NO_REFUSAL, cause, Ordering::SeqCst, Ordering::SeqCst) - .is_ok() + .is_ok(); + + if first { + self.refusal_signal.notify_one(); + } + + first } } @@ -209,6 +225,7 @@ impl FrameSink { account: Arc::new(Account { queued_bytes: AtomicUsize::new(0), refusal: AtomicU8::new(NO_REFUSAL), + refusal_signal: Notify::new(), budget, }), }, @@ -225,6 +242,7 @@ impl FrameSink { account: Arc::new(Account { queued_bytes: AtomicUsize::new(0), refusal: AtomicU8::new(NO_REFUSAL), + refusal_signal: Notify::new(), budget: None, }), } @@ -233,14 +251,21 @@ impl FrameSink { /// Sends one frame. A closed socket drops the frame and is not an error. /// /// A queue-full refusal latches the socket shut, as the overflow always has: every later - /// send is dropped. A budget refusal does not — it is one frame that did not fit at that - /// instant, and the connection task closes the socket at its next iteration whatever the - /// sender does — so the accounting stays exact and a frame that fits later is admitted. + /// send is dropped, and the check comes before the frame is encoded — a stored read can + /// keep producing events until the connection task closes it, and each discarded event + /// would otherwise cost an allocation and a serialisation that nothing reads. A budget + /// refusal does not latch — it is one frame that did not fit at that instant, and the + /// connection task closes the socket at its next iteration whatever the sender does — so + /// the accounting stays exact and a frame that fits later is admitted. pub fn send(&self, message: &RelayMessage) { let Some(sender) = &self.sender else { return; }; + if self.account.refusal() == Some(Refusal::QueueFull) { + return; + } + let text = message.to_text(); let length = text.len(); let mut queued = self.account.queued_bytes.load(Ordering::SeqCst); @@ -252,6 +277,7 @@ impl FrameSink { if self.account.refusal() == Some(Refusal::QueueFull) { return; } + // An empty queue admits its frame whatever the size, so an event larger than the // whole bound still reaches the client instead of closing it. if queued > 0 && queued + length > self.bound_bytes { @@ -328,6 +354,14 @@ impl FrameSink { self.account.refusal() } + /// Resolves when a refusal is recorded on this socket, so the connection task can close it + /// without waiting for a read, a write or a liveness tick to wake it. A permit is stored + /// when nothing is waiting, so a refusal recorded before the loop reaches its `select!` is + /// delivered rather than lost. + pub fn refused(&self) -> impl Future + '_ { + self.account.refusal_signal.notified() + } + /// The bytes admitted into the queue and not yet written. pub fn queued_bytes(&self) -> usize { self.account.queued_bytes.load(Ordering::SeqCst) @@ -552,6 +586,7 @@ mod tests { const SENDERS: usize = 8; let budget = Arc::new(FrameBudget::new(SENDERS * len)); let max_seen = AtomicUsize::new(0); + let finished = AtomicUsize::new(0); let start = std::sync::Barrier::new(SENDERS + 1); let mut sinks = Vec::new(); @@ -564,31 +599,30 @@ mod tests { for sink in &sinks { let frame = &frame; let start = &start; + let finished = &finished; scope.spawn(move || { start.wait(); sink.send(frame); + finished.fetch_add(1, Ordering::SeqCst); }); } - // A sampler reading the counter while the sends run: it stops when the budget is - // full, so it cannot outlive the sends it observes. + // A sampler reading the counter while the sends run. It stops when the last sender + // has finished rather than when the counter reaches the expected total, so the + // accounting bug this test exists to catch fails it with numbers instead of + // leaving it spinning inside the scope. { let budget = Arc::clone(&budget); let max_seen = &max_seen; + let finished = &finished; let start = &start; scope.spawn(move || { start.wait(); - loop { - let accounted = budget.accounted(); - max_seen.fetch_max(accounted, Ordering::SeqCst); - - if accounted == SENDERS * len { - break; - } - + while finished.load(Ordering::SeqCst) < SENDERS { + max_seen.fetch_max(budget.accounted(), Ordering::SeqCst); std::hint::spin_loop(); } }); diff --git a/crates/nostream-relay/tests/transport.rs b/crates/nostream-relay/tests/transport.rs index 76327fbb..1258236e 100644 --- a/crates/nostream-relay/tests/transport.rs +++ b/crates/nostream-relay/tests/transport.rs @@ -956,3 +956,94 @@ async fn a_frame_that_exceeds_the_budget_closes_the_socket_with_1013_and_no_text assert_eq!(server.relay.frame_budget().accounted(), 0); } } + +#[tokio::test] +async fn a_shared_budget_refusal_closes_the_socket_without_waiting_for_a_tick() { + // The budget is shared, so the frame that does not fit may be refused while *this* + // socket's own queue is empty: nothing here is ready to wake its loop — no queued frame, + // no inbound message, no handler finishing — and the close must not wait for the 600 s + // liveness tick. One socket holds the budget and never reads, so its write blocks with + // ~7 MiB of the 8 MiB still counted; the second socket's 2 MiB frame then does not fit + // what is left (7 + 2 > 8). + const BUDGET: usize = 8 * 1024 * 1024; + const HELD: usize = 7 * 1024 * 1024; + + let fixture = Fixture::new(settings()); + let held_content = "h".repeat(HELD); + let refused_content = "r".repeat(2 * 1024 * 1024); + + fixture + .seed_event(signed_event( + &CLIENT_SECRET, + 1, + vec![], + &held_content, + NOW_SECS, + )) + .await; + fixture + .seed_event(signed_event( + &CLIENT_SECRET, + 2, + vec![], + &refused_content, + NOW_SECS + 1, + )) + .await; + // A second kind-2 event, so the read that loses its frame stays in flight afterwards + // instead of finishing and waking the loop through `handlers.join_next()`. + fixture + .seed_event(signed_event( + &CLIENT_SECRET, + 2, + vec![], + "tail", + NOW_SECS + 2, + )) + .await; + + let server = Server::start_with(fixture, relay_keys(), bounded_config(BUDGET, 64)).await; + + // The socket that holds the budget: it never reads, so the loop's write of its one frame + // cannot finish and the bytes stay counted. + let (mut holder, _head) = RawClient::connect(server.address, None).await; + assert!(next_text(&mut holder).await.starts_with(r#"["AUTH","#)); + + holder.send_text(r#"["REQ","hold",{"kinds":[1]}]"#).await; + + assert!( + eventually(|| server.relay.frame_budget().accounted() > BUDGET / 2).await, + "the socket that never reads holds its frame in the budget, found {}", + server.relay.frame_budget().accounted() + ); + assert_eq!(server.relay.frame_budget().refusals(), 0); + + // From here every stored read waits on the gate, so the second socket's first event is + // delivered from its read task — with this loop parked in `select!` — rather than inside + // the `REQ`'s ordered prefix. + let gate = server.fixture.events.gate_reads_per_item(); + + let (mut refused, _head) = RawClient::connect(server.address, None).await; + assert!(next_text(&mut refused).await.starts_with(r#"["AUTH","#)); + + refused.send_text(r#"["REQ","now",{"kinds":[2]}]"#).await; + // The loop answers a ping in place, so the pong proves the `REQ` was taken and its read is + // parked: nothing in this socket's loop is ready after that. + refused.send_frame(true, false, 0x9, b"budget").await; + tokio::time::timeout(Duration::from_secs(5), refused.frames_until(0xa)) + .await + .expect("the loop answers the ping"); + + gate.add_permits(1); + + let frames = tokio::time::timeout(Duration::from_secs(2), refused.frames_until(0x8)) + .await + .expect("the refusal closes the socket without waiting for a liveness tick"); + + assert_no_text_frame(&frames); + assert_eq!( + close_of(frames.last().expect("a close frame")), + (1013, transport::FRAME_BUDGET_REASON.to_owned()) + ); + assert_eq!(server.relay.frame_budget().refusals(), 1); +} From c5bc366c2554f7927dd27c123d16e702ceeb5470 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 00:00:59 -0400 Subject: [PATCH 06/12] docs(rewrite): record the refusal signal in ADR 0008 and retarget its line citations --- .../decisions/0008-frame-memory-budget.md | 29 ++++++++++++++----- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index aac2e086..3bafffdb 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -9,7 +9,7 @@ The outbound frame queue was bounded per socket and nothing bounded their sum. `RelayConfig::frame_queue_bytes` (8 MiB; `crates/nostream-relay/src/relay.rs:57`) is enforced by `FrameSink::send`'s compare-and-exchange on the socket's own counter -(`crates/nostream-relay/src/sink.rs:239-322`), and `Relay::serve` accepted every connection with +(`crates/nostream-relay/src/sink.rs:260-334`), and `Relay::serve` accepted every connection with no count at all. The only per-address bookkeeping was the **attempt** limiter, which throttles how fast connections open, not how many are held (`crates/nostream-relay/src/limits.rs`, windows consumed at `crates/nostream-relay/src/relay.rs:327`). @@ -54,7 +54,7 @@ refused for it. Neither number is a total-memory ceiling; see the next section. ## What "frame memory" means here Accounted **outbound payload bytes**: the `text.len()` that `frame_queue_bytes` already counts -(`crates/nostream-relay/src/sink.rs:244-245`), not `String` capacity, not the channel node, and not +(`crates/nostream-relay/src/sink.rs:269-270`), not `String` capacity, not the channel node, and not resident memory. The nearest measurement of the ratio is job 010's probe D, which saw roughly 2 KiB resident per 1 KiB queued at 1 KiB frames, so **256 MiB accounted is not a process RSS ceiling** and should not be read as one. @@ -91,7 +91,7 @@ call site had to change. Both bounds close the socket with `CloseCode::Again` (1013) written straight to the transport rather than through the sink, bounded by `config.write_timeout` exactly as the existing slow-consumer close -is (`Relay::close_frame`, `crates/nostream-relay/src/relay.rs:524-535`). Three fixed reason strings +is (`Relay::close_frame`, `crates/nostream-relay/src/relay.rs:530-541`). Three fixed reason strings now exist, each lowercase and resource-stating, per the convention `SLOW_CONSUMER_REASON` set (`crates/nostream-relay/src/transport.rs:37-47`): @@ -108,7 +108,7 @@ websocket client reads, rather than in a payload it may never consume. is still the silent upgrade-then-drop measured above; this job leaves `RateLimits`, its windows and its whitelist untouched. The two new refusals are distinguishable from it and from each other by code, reason string, and one counter each (`FrameBudget::refusals`, -`crates/nostream-relay/src/sink.rs:90`; `ConnectionCap::refusals`, +`crates/nostream-relay/src/sink.rs:91`; `ConnectionCap::refusals`, `crates/nostream-relay/src/connection_cap.rs:60`), which is what keeps them out of the "looks like a network fault" bucket that started this. @@ -120,11 +120,24 @@ at its next iteration whatever the sender does. The accounting therefore stays e choice, and the unit test that proves room lets the next frame through (`sink.rs::the_budget_binds_across_sockets_and_is_shared`) can say so. +**The refusal wakes the loop.** Recording the cause is not enough on its own. A budget refusal +can arrive while *this* socket's queue is empty — the bytes that left no room are another +socket's — with the socket idle and its read parked on the store, and then nothing in the +connection task is ready: no frame to write, no inbound message, no handler finishing. The +account owns a `tokio::sync::Notify`, the first refusal signals it, and `run_connection` selects +on `FrameSink::refused()`, so the close is one iteration away rather than one liveness interval +away. `notify_one` stores a permit when nothing is waiting, which is what makes a refusal +recorded between the loop's check and its `select!` survive instead of being lost; two racing +refusals leave one permit, and the loop rechecks the cause it already reads at the top of every +iteration. The wire test that holds this down +(`tests/transport.rs::a_shared_budget_refusal_closes_the_socket_without_waiting_for_a_tick`) +times out at two seconds without the signal and closes in about 1.3 s with it. + ## Where the budget binds: per-frame admission, not an accept-time reservation The budget is enforced where the per-socket bound already is: in `FrameSink::send`, as a second compare-and-exchange loop against `accounted + length > limit` -(`crates/nostream-relay/src/sink.rs:99-135`). Admission requires room in both bounds; the budget has +(`crates/nostream-relay/src/sink.rs:100-136`). Admission requires room in both bounds; the budget has **no empty-queue exception** — unlike `frame_queue_bytes`, whose exception keeps one large event from closing every client — because 256 MiB exceeds any single outbound frame the protocol can produce by three orders of magnitude, and strictness is what makes a one-byte test budget meaningful. @@ -142,9 +155,9 @@ the socket's accounting ends) and refuses the frames that actually need the memo ## The refund, and the shape rejected **Chosen: refund on the last handle's `Drop`.** `FrameSink` holds an `Arc` -(`crates/nostream-relay/src/sink.rs:185-191`); the account carries the socket's outstanding bytes, +(`crates/nostream-relay/src/sink.rs:201-207`); the account carries the socket's outstanding bytes, its refusal cause and the shared budget, and `impl Drop for Account` -(`crates/nostream-relay/src/sink.rs:175-181`) returns whatever is left of that socket's bytes to the +(`crates/nostream-relay/src/sink.rs:191-197`) returns whatever is left of that socket's bytes to the budget. Every clone of the socket's sink shares the account, so the drop that runs is the last one: a send holds a sink and therefore cannot be in flight or begin once that drop starts. That is what makes the refund **exactly this socket's bytes, once**, whatever the connection loop did on the way out. @@ -201,7 +214,7 @@ Two accessors in the `hidden_content()` style, `Relay::frame_budget()` and them: `accounted()`, `limit()`, `refusals()` on the budget; `live(ip)`, `max_per_ip()`, `refusals()`, `tracked_addresses()` on the cap. Plus two structured log lines, one per refusal — a `warn!` carrying `budget_bytes` and `accounted_bytes` from the sink -(`crates/nostream-relay/src/sink.rs:295-299`), and a `debug!` carrying `peer`, +(`crates/nostream-relay/src/sink.rs:321-325`), and a `debug!` carrying `peer`, `max_connections_per_ip` and `live` from the connection task (`crates/nostream-relay/src/relay.rs:300-305`). From 1479c6c84ca016d956ed5e39ff26164c8e2d5da9 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 00:34:22 -0400 Subject: [PATCH 07/12] fix(relay): stop a stored read at the first refused frame, and correct the record Review round 1: a performance finding and two documentation defects, plus four notes. - A budget refusal did not stop the encode. The read loop in fetch_and_send encoded one frame per event and discarded each refused one; the window is the connection loop's write plus close_frame (30 s each) and is the normal case, since the budget binds when queues are full and loops are parked in writes. The read now ends at the top of its loop when the sink has a refusal, before pulling or encoding the next record; the sink still does not latch a budget refusal, so AC6's re-admission holds. New test in subscribe.rs: with a 64-byte budget every event frame is refused and the EOSE would fit, so an empty queue proves the read stopped (without the check it holds the 12-byte EOSE). - Two ADR citations were stale by six lines and one named the wrong arm; every citation in the file was re-read against the final tree. - The ADR presented the cap's refusal line as operator-visible; it is a debug! line and the daemon ships the info filter, so the record now says so, names the switch, and calls the gap a gap. Notes: the two new refusals are separated by the reason string, not by 1013; the shared budget's blast radius (one address below its cap can refuse every other client, including a newcomer's AUTH) is recorded; the cap's key is an address, so a [::] bind keys per /128; and RelayConfig's doc no longer claims the policies never move from their defaults. --- .../nostream-relay/src/handlers/subscribe.rs | 93 +++++++++++++++++++ crates/nostream-relay/src/relay.rs | 4 +- .../decisions/0008-frame-memory-budget.md | 73 +++++++++++++-- 3 files changed, 158 insertions(+), 12 deletions(-) diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index b0ca6cf0..0d95448a 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -268,6 +268,16 @@ async fn fetch_and_send(mut read: InFlightRead, filters: &[Filter], context: &Ha return; } + // A refused frame is never queued, and the connection loop is about to close this + // socket whatever this read does. Encoding the rest of the stream would only discard + // it, and with a shared budget the loop can be a whole blocked write away from + // running, so the window is not short. The sink must not latch on a budget refusal — + // a frame that fits later is admitted — which is why the stop lives here. + if context.sink.refusal().is_some() { + context.connection.end_read(read); + return; + } + let record = tokio::select! { changed = read.cancellation().changed() => { // An aborted read sends no frame at all, not even EOSE. @@ -408,3 +418,86 @@ fn can_subscribe( None } + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + + use nostream_core::clock::TestClock; + use nostream_core::event::{EventId, Pubkey, Signature}; + use nostream_core::nip42::Nip42Session; + + use super::*; + use crate::connection::Connection; + use crate::relay_key::RelayKeySource; + use crate::sink::{FrameBudget, FrameSink, Refusal}; + use crate::testing::{Fixture, settings}; + + /// An event the fixture can hold, signed by `secret`. + fn signed_event(secret: &[u8; 32], created_at: i64) -> nostream_core::Event { + let mut event = nostream_core::Event { + id: EventId::from_byte_array([0; 32]), + pubkey: Pubkey::from_byte_array([0; 32]), + created_at, + kind: Kind(1), + tags: Vec::new(), + content: "stored".to_owned(), + sig: Signature::from_byte_array([0; 64]), + }; + + nostream_core::crypto::sign_event(secret, &mut event).expect("sign the test event"); + + event + } + + #[tokio::test] + async fn a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it() { + // Three stored events and a budget of 64 bytes: every event frame is refused and the + // read's closing `EOSE` would fit. So the queue says which read ran — a read that + // stopped at the first refusal leaves nothing in it, and a read that encoded and + // discarded the rest of the stream ends with the `EOSE` that fits. + let fixture = Fixture::new(settings()); + + for index in 0..3 { + fixture + .seed_event(signed_event(&[7u8; 32], 1_700_000_000 + index)) + .await; + } + + let budget = Arc::new(FrameBudget::new(64)); + let (sink, mut frames) = FrameSink::channel(64 * 1024, Some(Arc::clone(&budget))); + let refused = sink.clone(); + let context = Arc::new(fixture.context( + Arc::new(TestClock::new(1_700_000_000_000)), + RelayKeySource::from_values(None, None), + Arc::new(Connection::new( + "127.0.0.1".to_owned(), + Nip42Session::with_challenge("challenge", 0), + )), + sink, + )); + + let message = SubscribeMessage { + subscription_id: "s".to_owned(), + filters: vec![Filter { + kinds: Some(vec![Kind(1)]), + ..Filter::default() + }], + }; + + let read = begin(message, &context).expect("the subscription is registered"); + read.await; + + assert_eq!(refused.refusal(), Some(Refusal::BudgetExhausted)); + assert_eq!(refused.queued_bytes(), 0); + assert!( + frames.try_recv().is_err(), + "the read stopped at the first refusal: no event frame and no EOSE reached the queue" + ); + assert_eq!( + budget.refusals(), + 1, + "one refusal for the socket, not one per event" + ); + } +} diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index 1b733954..2ec579ac 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -40,8 +40,8 @@ use crate::wot::WotGraph; /// The knobs a [`Relay`] is built with that are not settings: a constant the capture fixes in /// code (`liveness`, the NIP-56 refresh) or a policy the port adds with no counterpart there /// (`frame_queue_bytes`, `write_timeout`, `frame_memory_budget_bytes`, -/// `max_connections_per_ip`; ADR 0005, ADR 0008). Tests set the cadences they need -/// (`tests/common/mod.rs::fast_config`); the policies run at their shipped defaults. A value +/// `max_connections_per_ip`; ADR 0005, ADR 0008). Tests set the cadences they need and the +/// bounds a case asks for; every value here runs at its shipped default otherwise. A value /// the capture's settings document carries belongs in [`Settings`], never here. #[derive(Debug, Clone)] pub struct RelayConfig { diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index 3bafffdb..feaad0f5 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -106,11 +106,16 @@ websocket client reads, rather than in a payload it may never consume. **Neither refusal is a timeout, and neither is the attempt limiter's.** The attempt limiter's refusal is still the silent upgrade-then-drop measured above; this job leaves `RateLimits`, its windows and -its whitelist untouched. The two new refusals are distinguishable from it and from each other by -code, reason string, and one counter each (`FrameBudget::refusals`, -`crates/nostream-relay/src/sink.rs:91`; `ConnectionCap::refusals`, -`crates/nostream-relay/src/connection_cap.rs:60`), which is what keeps them out of the "looks like a -network fault" bucket that started this. +its whitelist untouched. What separates the two new refusals from the limiter's silent drop is that +they say something: a close frame with **code `1013`** and a fixed reason string. What separates +them **from each other** is the reason string and the counter, not the code — plus the log line's +fields (`budget_bytes`/`accounted_bytes` against `peer`/`max_connections_per_ip`/`live`) and, for a +test driving the relay directly, the accessor (`FrameBudget::refusals` at +`crates/nostream-relay/src/sink.rs:91` or `ConnectionCap::refusals` at +`crates/nostream-relay/src/connection_cap.rs:60`). Both bounds use `CloseCode::Again`, so `1013` on +its own tells a client that it was refused and says nothing about which bound did it; anything that +wants to tell the two apart reads the reason string. That is what keeps both out of the "looks like +a network fault" bucket that started this. **A budget refusal refuses the frame, not the socket.** The refusal cause is recorded once — so the counter counts one refused socket, not one per frame, and the warn line is written once — but only a @@ -120,6 +125,21 @@ at its next iteration whatever the sender does. The accounting therefore stays e choice, and the unit test that proves room lets the next frame through (`sink.rs::the_budget_binds_across_sockets_and_is_shared`) can say so. +**The refusal stops the reader as well.** The connection loop closes the socket at its next +iteration, but a stored read is a task of its own and keeps producing events until it notices: +`fetch_and_send` (`crates/nostream-relay/src/handlers/subscribe.rs:276-279`) encodes one frame per event, and +each refused frame is encoded and discarded. The window is not short — the loop can be parked in a +write and then in `close_frame`, each bounded only by the 30 s `write_timeout`, and +`handlers.abort_all()` runs only after both — and it is the normal case, because the budget binds +exactly when queues are full and loops are parked in writes. So the read checks +`context.sink.refusal()` at the top of its loop and ends there, before pulling or encoding the next +record, with no `EOSE` (there is no socket left to tell). The sink itself must not latch a budget +refusal — a frame that fits later is admitted, which the shared-budget test relies on — so the stop +belongs to the reader. `subscribe.rs::a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it` +holds it down: with a 64-byte budget every event frame is refused and the `EOSE` would fit, so the +queue stays empty only if the read stopped at the first refusal (without the check it holds the +12-byte `EOSE`). + **The refusal wakes the loop.** Recording the cause is not enough on its own. A budget refusal can arrive while *this* socket's queue is empty — the bytes that left no room are another socket's — with the socket idle and its read parked on the store, and then nothing in the @@ -152,6 +172,16 @@ which is a second refund path of exactly the kind section 7 exists to eliminate. has one lifecycle (bytes are counted when a frame is admitted, uncounted when it is written or when the socket's accounting ends) and refuses the frames that actually need the memory. +**The blast radius is one address, and it is new.** The budget is shared, so it is only as fair as +the sockets already holding it. Eight sockets at the 8 MiB `frame_queue_bytes` ceiling are 64 MiB and +32 of them are 256 MiB, so **one address stays below its own 64-socket cap and still exhausts the +budget** — and then every other client's next frame is refused and its socket closed, including a +new client's AUTH challenge, which is a text frame like any other and is refused before that client +has been spoken to. Before this change there was no shared bound to exhaust: one address could hold +`N × 8 MiB` of its own, but it could not make another client's connection fail. That is the shape the +operator's comment chose — a shared bound over a per-address one — and it is recorded here as a cost +of it, not as a defect. + ## The refund, and the shape rejected **Chosen: refund on the last handle's `Drop`.** `FrameSink` holds an `Arc` @@ -164,11 +194,13 @@ the refund **exactly this socket's bytes, once**, whatever the connection loop d **The rejected shape: a decrement at each of the loop's exits.** Today `release` has exactly one caller, the drain arm (`crates/nostream-relay/src/relay.rs:407`), and every other exit leaves whatever -was still queued counted: the overflow close (`:391-394`), the `None` receiver (`:398`), a failed -write (`:409-411`), the inbound and close arms (`:414`, `:433-435`, `:440`), the lagged hub (`:460`), the -liveness arms (`:464-469`, `:479-481`) and the final cleanup (`:488-489`). Per socket that leak is invisible -because the counter dies with the sink; a process-wide budget makes it a leak of up to 8 MiB on every -disconnect. Writing the decrement at each of those ten sites is the change that looks right and rots: +was still queued counted: the drain itself (`:384-385`), the refusal close (`:391-394`), the `None` +receiver (`:398`), a failed write (`:409-411`), the inbound arms (`:414`, `:433-435`, `:440`, +`:443-445`), the closed hub (`:460`), the liveness arms (`:463-469`, `:479-481`) and the final +cleanup (`:494-495`); the lagged hub (`:455-457`) warns and does not leave. Per socket that leak is +invisible because the counter dies with the sink; a process-wide budget makes it a leak of up to +8 MiB on every disconnect. Writing the decrement at each of those sites is the change that looks +right and rots: the next exit added to the loop silently leaks, and no unit test can reach the exits that need a failed handshake or a poisoned store to trigger. `Drop` covers all of them by construction, including the registration failure inside `run_connection` — which matters, because a refund that ran only on @@ -207,6 +239,14 @@ This is a gauge, not a window: no store, no clock, no expiry. `RateLimits` and ` unchanged, and the measured 20-socket behaviour — 12 served, 8 dropped by the limiter — is unchanged with it. +**The key is an address, not an identity.** The cap counts source addresses, which is what the +policy's name says, and it holds on the shipped bind: `nostreamd` listens on +`SocketAddr::from((Ipv4Addr::UNSPECIFIED, port))` (`crates/nostreamd/src/main.rs:90`), so every +client has a distinct IPv4 source address. The relay is a library, though, and a listener bound to +`[::]` keys an IPv6 peer by its full address: one host holding a /64 may present 2^64 source +addresses, one socket each, and the cap then bounds nothing. Closing that is a policy about +prefixes, and it is not this job's. + ## What an operator can see today Two accessors in the `hidden_content()` style, `Relay::frame_budget()` and @@ -218,6 +258,19 @@ them: `accounted()`, `limit()`, `refusals()` on the budget; `live(ip)`, `max_per `max_connections_per_ip` and `live` from the connection task (`crates/nostream-relay/src/relay.rs:300-305`). +**Those two lines are not visible at the same level, and the daemon ships `info`.** The filter is +`--log-level`, then `LOG_LEVEL`, then `RUST_LOG`, else `info` +(`crates/nostreamd/src/telemetry.rs:10-23`). At the shipped default an operator therefore sees the +budget's `warn!` and **not** the cap's `debug!`: a socket refused at the cap leaves no line an +operator will find, which is the silent-invisibility failure this job exists to remove, restored on +one of the two paths. Seeing it takes `--log-level debug` (or `LOG_LEVEL=debug`, or a `RUST_LOG` +directive for `nostream_relay`). The reason the line is `debug!` and not something louder is that it +carries per-socket detail — `peer`, `live` — and would be log-amplifiable by exactly the client that +trips it; the process-wide fact belongs to the counter instead. Which leaves the counters as the +surface that does not depend on a log level, and they are readable only from a `Relay` handle today, +i.e. from a test or an embedder. **Making either refusal visible to an operator who has not raised +the filter is an open gap, and #989 or the `/metrics` work is where it closes.** + That is the whole surface, because **a Prometheus endpoint is out of reach right now**: the OTLP exporter is a stub returning `NotImplemented` (`crates/nostreamd/src/telemetry.rs:47-52`) and the admin `/metrics` controller is a Phase-1 stub (`crates/nostream-admin/src/controllers/get_metrics.rs:5-9`). From f688ae4cf9ae5ef5016bd495ab6fbc54604b06b8 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 00:35:24 -0400 Subject: [PATCH 08/12] docs(rewrite): point ADR 0008 at the cap's removal block, not the blank line above it --- docs/rewrite/decisions/0008-frame-memory-budget.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index feaad0f5..af08ec49 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -228,7 +228,7 @@ Acquisition returns a `ConnectionSlot` held for the rest of `handle_socket` `run_connection`, and any early return added later all release it through `Drop` (`crates/nostream-relay/src/connection_cap.rs:101-116`). A count that leaks makes the cap permanent for that address, which is a defect and not a memory leak. The map entry is removed when its count -reaches zero (`:111-114`), so an address that once connected leaves nothing behind and the map is +reaches zero (`:112-114`), so an address that once connected leaves nothing behind and the map is bounded by live addresses. The cap applies to **every address, including whitelisted ones**. An exempt address would bypass the From 5b23eebd40ed43a202aacb06d6150e264df0663a Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 00:47:26 -0400 Subject: [PATCH 09/12] fix(relay): hold a refused read until the close retires it, so the wake-up is what closes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Static round 2 showed the parked-read test passed with the refusal signal deleted: the read returned at the guard, its handler completed, and handlers.join_next() woke the loop. The two changes had cancelled each other's discriminating power, and the close's promptness came from a handler finishing rather than from the signal. A refused read now waits for the close that retires it (the cancellation watch, then end_read) instead of unwinding: the connection loop's abort_all/close stays the single owner of the teardown, and the close is prompt because the refusal woke the loop. The read's own test is restructured for the park — pending after the refusal, one record taken, no EOSE queued, retired by the connection's close — and the two tests now fail on their own mechanism: deleting the refused() arm makes the wire test time out at 2 s, and deleting the read's guard makes the unit test fail with the park message. --- .../nostream-relay/src/handlers/subscribe.rs | 55 ++++++++++++++----- 1 file changed, 41 insertions(+), 14 deletions(-) diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index 0d95448a..6462ac11 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -273,7 +273,15 @@ async fn fetch_and_send(mut read: InFlightRead, filters: &[Filter], context: &Ha // it, and with a shared budget the loop can be a whole blocked write away from // running, so the window is not short. The sink must not latch on a budget refusal — // a frame that fits later is admitted — which is why the stop lives here. + // + // The read then waits for the close that retires it instead of unwinding: it has + // nothing left to send, and the connection loop's `abort_all` and `close` are the one + // owner of the teardown, so this handler must not run `end_read` against the same + // state while the loop is closing the socket. Parking also keeps the close's + // promptness a property of the refusal signal rather than of this handler happening + // to finish. if context.sink.refusal().is_some() { + let _ = read.cancellation().changed().await; context.connection.end_read(read); return; } @@ -423,6 +431,8 @@ fn can_subscribe( mod tests { #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + use std::time::Duration; + use nostream_core::clock::TestClock; use nostream_core::event::{EventId, Pubkey, Signature}; use nostream_core::nip42::Nip42Session; @@ -452,11 +462,13 @@ mod tests { #[tokio::test] async fn a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it() { - // Three stored events and a budget of 64 bytes: every event frame is refused and the - // read's closing `EOSE` would fit. So the queue says which read ran — a read that - // stopped at the first refusal leaves nothing in it, and a read that encoded and - // discarded the rest of the stream ends with the `EOSE` that fits. + // Three stored events behind a per-item gate, and a 64-byte budget: every event frame + // is refused, so the read must neither take another record nor encode one. The gate is + // the measure — two permits left means one record was taken and the stream was not + // drained — and the queue is the other: a read that ran on to the end would leave the + // 12-byte `EOSE` in it, which does fit. let fixture = Fixture::new(settings()); + let gate = fixture.events.gate_reads_per_item(); for index in 0..3 { fixture @@ -467,13 +479,14 @@ mod tests { let budget = Arc::new(FrameBudget::new(64)); let (sink, mut frames) = FrameSink::channel(64 * 1024, Some(Arc::clone(&budget))); let refused = sink.clone(); + let connection = Arc::new(Connection::new( + "127.0.0.1".to_owned(), + Nip42Session::with_challenge("challenge", 0), + )); let context = Arc::new(fixture.context( Arc::new(TestClock::new(1_700_000_000_000)), RelayKeySource::from_values(None, None), - Arc::new(Connection::new( - "127.0.0.1".to_owned(), - Nip42Session::with_challenge("challenge", 0), - )), + Arc::clone(&connection), sink, )); @@ -485,19 +498,33 @@ mod tests { }], }; - let read = begin(message, &context).expect("the subscription is registered"); - read.await; + let mut read = begin(message, &context).expect("the subscription is registered"); + gate.add_permits(3); - assert_eq!(refused.refusal(), Some(Refusal::BudgetExhausted)); - assert_eq!(refused.queued_bytes(), 0); assert!( - frames.try_recv().is_err(), - "the read stopped at the first refusal: no event frame and no EOSE reached the queue" + tokio::time::timeout(Duration::from_millis(200), &mut read) + .await + .is_err(), + "the read parks once its frame is refused instead of draining the stream" ); + assert_eq!(refused.refusal(), Some(Refusal::BudgetExhausted)); + assert_eq!( + gate.available_permits(), + 2, + "one record was taken and the rest of the stream was never asked for" + ); + assert!(frames.try_recv().is_err(), "no EOSE reached the queue"); + assert_eq!(refused.queued_bytes(), 0); assert_eq!( budget.refusals(), 1, "one refusal for the socket, not one per event" ); + + // The close the read is waiting for ends it. + connection.close(); + tokio::time::timeout(Duration::from_secs(1), read) + .await + .expect("the connection's close ends the read"); } } From ff144dc08dfed2287dbea5ca352a5e8d7a86d821 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 01:07:29 -0400 Subject: [PATCH 10/12] docs(rewrite): say that a refused read waits for the close, and what that holds Round 2 (code, security, architecture) found one ADR paragraph that 5b23eeb had made false, the code comment copying it, and an ownership claim one clause too strong. - The read-stop paragraph said a refused read 'ends there'. It stops producing there and then waits for the close that retires it, so it now says that, names what a refused client therefore still holds, and no longer contradicts its own account of the two 30 s write windows. - The encode citation moved from the park comment to the send (subscribe.rs:325-330), and the guard is cited where it is (:284-288). - 'its read parked on the store' in the wake-up paragraph was the pre-1479c6c state; the relay.rs comment copied it. Both now say stopped and waiting for the close. - Section 7 records that the park holds the store stream to the close: latent today (unavailable::stores), but sqlx's &Pool executor holds a pooled connection per stream and DB_MAX_POOL_SIZE is 3, so whoever wires the store leaves the loop before parking. - The ownership comment claimed the loop's abort_all/close was the sole owner of teardown while end_read is reachable from six places. It now says what makes that safe: end_read removes the entry only while the generation still matches (connection.rs:290-297). Every line reference in the ADR was re-read from the final tree: 53 of them, all resolving. --- .../nostream-relay/src/handlers/subscribe.rs | 11 ++-- crates/nostream-relay/src/relay.rs | 4 +- .../decisions/0008-frame-memory-budget.md | 52 +++++++++++++------ 3 files changed, 44 insertions(+), 23 deletions(-) diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index 6462ac11..cd29b1ba 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -275,11 +275,12 @@ async fn fetch_and_send(mut read: InFlightRead, filters: &[Filter], context: &Ha // a frame that fits later is admitted — which is why the stop lives here. // // The read then waits for the close that retires it instead of unwinding: it has - // nothing left to send, and the connection loop's `abort_all` and `close` are the one - // owner of the teardown, so this handler must not run `end_read` against the same - // state while the loop is closing the socket. Parking also keeps the close's - // promptness a property of the refusal signal rather than of this handler happening - // to finish. + // nothing left to send, and parking keeps the close's promptness a property of the + // refusal signal rather than of this handler happening to finish. The `end_read` + // below is the same call every other exit of this function makes, and it is safe + // beside the loop's own `close` because it removes the entry only while the generation + // it carries still matches (`connection.rs:290-297`): whichever of the two runs + // second finds no entry of its own and does nothing. if context.sink.refusal().is_some() { let _ = read.cancellation().changed().await; context.connection.end_read(read); diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index 2ec579ac..c776c025 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -483,8 +483,8 @@ impl Relay { _ = drained.changed() => drain_seen = *drained.borrow_and_update(), // A refusal can arrive with nothing else to wake this loop: the budget is // shared, so the frame that did not fit may have left this socket's own queue - // empty, the socket itself idle, and its read parked on the store. The - // producer signals when it records the refusal, and the next iteration's + // empty, the socket idle, and its read stopped and waiting for this close. + // The producer signals when it records the refusal, and the next iteration's // check closes the client — immediate, not one liveness tick away (ADR 0008). _ = sink.refused() => {} Some(_) = handlers.join_next(), if !handlers.is_empty() => {} diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index af08ec49..58bd5e89 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -125,25 +125,33 @@ at its next iteration whatever the sender does. The accounting therefore stays e choice, and the unit test that proves room lets the next frame through (`sink.rs::the_budget_binds_across_sockets_and_is_shared`) can say so. -**The refusal stops the reader as well.** The connection loop closes the socket at its next -iteration, but a stored read is a task of its own and keeps producing events until it notices: -`fetch_and_send` (`crates/nostream-relay/src/handlers/subscribe.rs:276-279`) encodes one frame per event, and -each refused frame is encoded and discarded. The window is not short — the loop can be parked in a -write and then in `close_frame`, each bounded only by the 30 s `write_timeout`, and -`handlers.abort_all()` runs only after both — and it is the normal case, because the budget binds -exactly when queues are full and loops are parked in writes. So the read checks -`context.sink.refusal()` at the top of its loop and ends there, before pulling or encoding the next -record, with no `EOSE` (there is no socket left to tell). The sink itself must not latch a budget -refusal — a frame that fits later is admitted, which the shared-budget test relies on — so the stop -belongs to the reader. `subscribe.rs::a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it` -holds it down: with a 64-byte budget every event frame is refused and the `EOSE` would fit, so the -queue stays empty only if the read stopped at the first refusal (without the check it holds the -12-byte `EOSE`). +**The refusal stops the reader too, and the reader then waits for the close.** The connection loop +closes the socket at its next iteration, but a stored read is a task of its own and keeps producing +events until it notices: `fetch_and_send` encodes one frame per event +(`crates/nostream-relay/src/handlers/subscribe.rs:325-330`), and each refused frame was encoded and +discarded. The window is not short — the loop can be parked in a write and then in `close_frame`, +each bounded only by the 30 s `write_timeout`, and `handlers.abort_all()` runs only after both — and +it is the normal case, because the budget binds exactly when queues are full and loops are parked in +writes. So the read checks `context.sink.refusal()` at the top of its loop, before pulling or +encoding the next record (`crates/nostream-relay/src/handlers/subscribe.rs:284-288`), and sends +nothing more: no further event, and no `EOSE`, which would be a frame this client is not reading. + +What the read does **not** do is finish there. It waits for the close that retires it — the +cancellation watch, then `end_read` — and what delivers that close is `handlers.abort_all()` and +`connection.close()` on the loop's way out (`crates/nostream-relay/src/relay.rs:494-495`). So a +refused client at this head holds two things until its socket actually closes: the read task, and +the store stream inside it (§7 records the second). The sink itself must not latch a budget refusal — +a frame that fits later is admitted, which the shared-budget test relies on — so the stop belongs to +the reader. `subscribe.rs::a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it` holds it +down: with a 64-byte budget every event frame is refused and the `EOSE` would fit, so the queue stays +empty only if the read stopped at the first refusal, and the store's per-item gate shows it asked for +exactly one record (without the check it drains the stream and holds the 12-byte `EOSE`). **The refusal wakes the loop.** Recording the cause is not enough on its own. A budget refusal can arrive while *this* socket's queue is empty — the bytes that left no room are another -socket's — with the socket idle and its read parked on the store, and then nothing in the -connection task is ready: no frame to write, no inbound message, no handler finishing. The +socket's — with the socket idle and the read that lost the frame waiting for the very close that +recording the refusal is supposed to start, and then nothing in the connection task is ready: no +frame to write, no inbound message, no handler finishing. The account owns a `tokio::sync::Notify`, the first refusal signals it, and `run_connection` selects on `FrameSink::refused()`, so the close is one iteration away rather than one liveness interval away. `notify_one` stores a permit when nothing is waiting, which is what makes a refusal @@ -210,6 +218,18 @@ The drain arm's `release` is unchanged and still runs: it returns written bytes budget is reusable while the socket lives, and the `Drop` refund then finds only what was never written. +**The parked read holds its store stream, and that is what must change when a store is wired.** +Stopping at the refusal without leaving the loop keeps the stream from `find_by_filters` alive until +the close retires the read; the version before this one dropped it at the refusal. Latent today — the +daemon wires `unavailable::stores()` (`crates/nostreamd/src/main.rs:113`), whose reads are stubs — +but not for ever: sqlx's `&Pool` executor draws a pooled connection for as long as a query stream is +alive, and `DB_MAX_POOL_SIZE` defaults to **3** (`crates/nostream-storage/src/postgres/mod.rs:362`), +so three refused reads parked at once would hold every connection in the pool and stall every other +read in the process. Whoever wires the store leaves the loop before parking — returning from +`fetch_and_send`, or dropping the stream first — and in doing so takes the close's promptness back +onto the refusal signal, because a handler that returns is a handler whose completion wakes the loop +(static round 2 is what that looked like). The choice is recorded here so it is made on purpose. + ## The per-IP cap A `ConnectionCap` (`crates/nostream-relay/src/connection_cap.rs`) counts **live sockets** per From f67e726555d6a2f9ba8cfce72725c71f852958fd Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 01:07:47 -0400 Subject: [PATCH 11/12] docs(rewrite): record the park's test-power trade the right way round Returning from the loop before parking does not put the close's promptness onto the refusal signal - it removes the signal's necessity, because a returned handler wakes the loop on completion. Say that, and that the wire test stops discriminating in that world. --- docs/rewrite/decisions/0008-frame-memory-budget.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index 58bd5e89..3cbd5502 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -226,9 +226,11 @@ but not for ever: sqlx's `&Pool` executor draws a pooled connection for as long alive, and `DB_MAX_POOL_SIZE` defaults to **3** (`crates/nostream-storage/src/postgres/mod.rs:362`), so three refused reads parked at once would hold every connection in the pool and stall every other read in the process. Whoever wires the store leaves the loop before parking — returning from -`fetch_and_send`, or dropping the stream first — and in doing so takes the close's promptness back -onto the refusal signal, because a handler that returns is a handler whose completion wakes the loop -(static round 2 is what that looked like). The choice is recorded here so it is made on purpose. +`fetch_and_send`, or dropping the stream first — and pays for it in test power: a handler that +returns wakes the loop on completion, so the close arrives promptly even with the refusal signal +deleted, and the wire test below then holds down nothing (static round 2 is exactly what that looked +like). That trade — a pooled connection against the signal being load-bearing — is recorded here so +it is made on purpose rather than by accident. ## The per-IP cap From 04186d4e7dea8b4a6cf72fa2d8b1b7679d91cd74 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 25 Oct 2026 01:22:28 -0400 Subject: [PATCH 12/12] docs(rewrite): say that the abort, not the watch, retires a refused read Round 3 (code, architecture) measured the same clause independently: every connection-loop exit is a break, handlers.abort_all() runs before connection.close(), and a JoinSet-aborted task is dropped without being polled again. So on the refusal path the parked read dies inside the cancellation watch, its end_read never runs, and the store stream goes with the abort; the watch is the path a client CLOSE or a re-REQ takes, and the one the unit test exercises. Also: the refused frame is the only one encoded per read (and a queue-latched socket skips that too), the wire-test pointer no longer says 'below' with nothing below it, and section 7 gives the deployed pool size (prod compose 64, no replica, so held connections stall writes as well) beside the code default of 3. The two code comments that described the watch as the retirement now match. --- .../nostream-relay/src/handlers/subscribe.rs | 15 ++-- crates/nostream-relay/src/relay.rs | 2 +- .../decisions/0008-frame-memory-budget.md | 73 +++++++++++-------- 3 files changed, 53 insertions(+), 37 deletions(-) diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index cd29b1ba..6644a8f3 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -274,13 +274,14 @@ async fn fetch_and_send(mut read: InFlightRead, filters: &[Filter], context: &Ha // running, so the window is not short. The sink must not latch on a budget refusal — // a frame that fits later is admitted — which is why the stop lives here. // - // The read then waits for the close that retires it instead of unwinding: it has - // nothing left to send, and parking keeps the close's promptness a property of the - // refusal signal rather than of this handler happening to finish. The `end_read` - // below is the same call every other exit of this function makes, and it is safe - // beside the loop's own `close` because it removes the entry only while the generation - // it carries still matches (`connection.rs:290-297`): whichever of the two runs - // second finds no entry of its own and does nothing. + // The read then parks instead of unwinding: it has nothing left to send, and parking + // keeps the close's promptness a property of the refusal signal rather than of this + // handler happening to finish. On the refusal path the loop's `abort_all` drops the + // parked task before the watch below ever resolves, so the `end_read` after it is the + // path a client `CLOSE` or a re-`REQ` takes, not the refusal's. It is the same call + // every other exit of this function makes, and it is safe beside the loop's own + // `close` because it removes the entry only while the generation it carries still + // matches (`connection.rs:290-297`): whichever of the two runs second does nothing. if context.sink.refusal().is_some() { let _ = read.cancellation().changed().await; context.connection.end_read(read); diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index c776c025..2dcd1f3b 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -483,7 +483,7 @@ impl Relay { _ = drained.changed() => drain_seen = *drained.borrow_and_update(), // A refusal can arrive with nothing else to wake this loop: the budget is // shared, so the frame that did not fit may have left this socket's own queue - // empty, the socket idle, and its read stopped and waiting for this close. + // empty, the socket idle, and its read stopped and parked with nothing to run it. // The producer signals when it records the refusal, and the next iteration's // check closes the client — immediate, not one liveness tick away (ADR 0008). _ = sink.refused() => {} diff --git a/docs/rewrite/decisions/0008-frame-memory-budget.md b/docs/rewrite/decisions/0008-frame-memory-budget.md index 3cbd5502..962b8814 100644 --- a/docs/rewrite/decisions/0008-frame-memory-budget.md +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -125,26 +125,37 @@ at its next iteration whatever the sender does. The accounting therefore stays e choice, and the unit test that proves room lets the next frame through (`sink.rs::the_budget_binds_across_sockets_and_is_shared`) can say so. -**The refusal stops the reader too, and the reader then waits for the close.** The connection loop +**The refusal stops the reader too, and the abort eventually retires it.** The connection loop closes the socket at its next iteration, but a stored read is a task of its own and keeps producing events until it notices: `fetch_and_send` encodes one frame per event -(`crates/nostream-relay/src/handlers/subscribe.rs:325-330`), and each refused frame was encoded and -discarded. The window is not short — the loop can be parked in a write and then in `close_frame`, -each bounded only by the 30 s `write_timeout`, and `handlers.abort_all()` runs only after both — and -it is the normal case, because the budget binds exactly when queues are full and loops are parked in -writes. So the read checks `context.sink.refusal()` at the top of its loop, before pulling or -encoding the next record (`crates/nostream-relay/src/handlers/subscribe.rs:284-288`), and sends -nothing more: no further event, and no `EOSE`, which would be a frame this client is not reading. - -What the read does **not** do is finish there. It waits for the close that retires it — the -cancellation watch, then `end_read` — and what delivers that close is `handlers.abort_all()` and -`connection.close()` on the loop's way out (`crates/nostream-relay/src/relay.rs:494-495`). So a -refused client at this head holds two things until its socket actually closes: the read task, and -the store stream inside it (§7 records the second). The sink itself must not latch a budget refusal — -a frame that fits later is admitted, which the shared-budget test relies on — so the stop belongs to -the reader. `subscribe.rs::a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it` holds it -down: with a 64-byte budget every event frame is refused and the `EOSE` would fit, so the queue stays -empty only if the read stopped at the first refusal, and the store's per-item gate shows it asked for +(`crates/nostream-relay/src/handlers/subscribe.rs:326-331`), so a frame that does not fit has been +encoded by the time the sink refuses it. The window is not short — the loop can be parked in a write +and then in `close_frame`, each bounded only by the 30 s `write_timeout`, and `handlers.abort_all()` +runs only after both — and it is the normal case, because the budget binds exactly when queues are +full and loops are parked in writes. So the read checks `context.sink.refusal()` at the top of its +loop, before pulling or encoding the next record +(`crates/nostream-relay/src/handlers/subscribe.rs:285-289`), and sends nothing more: no further +event, and no `EOSE`, which would be a frame this client is not reading. That leaves the frame that +was refused as the only one encoded and thrown away per read — one, not one per event — and a socket +the queue bound has already latched does not even pay that, because `send` returns before `to_text()` +(`crates/nostream-relay/src/sink.rs:260-266`). + +The read does not finish there: it waits on the read's cancellation watch and then calls `end_read`. +**On the refusal path that wait is never resolved.** Every exit of the connection loop is a `break`, +`handlers.abort_all()` runs before `connection.close()` (`crates/nostream-relay/src/relay.rs:494-495`), +and a task aborted by the `JoinSet` it was spawned into (`crates/nostream-relay/src/relay.rs:377`, +`:602-603`) is dropped without being polled again: the parked read is dropped inside the watch, the +`end_read` below it never runs, and the store stream goes with it. What retires a refused read in +production is therefore the abort, not the watch — the watch is the path a client can still take, by +a `CLOSE` or a re-`REQ`, and it is the one the unit test exercises, because the test closes the +connection rather than aborting the task. So a refused client holds two things until the loop aborts +it, a close frame later: the read task, and the store stream inside it (§7 records the second). + +The sink itself must not latch a budget refusal — a frame that fits later is admitted, which the +shared-budget test relies on — so the stop belongs to the reader. +`subscribe.rs::a_refused_sink_stops_the_read_instead_of_encoding_the_rest_of_it` holds it down with a +64-byte budget, where every event frame is refused and the `EOSE` would fit: the queue stays empty +only if the read stopped at the first refusal, and the store's per-item gate shows it asked for exactly one record (without the check it drains the stream and holds the 12-byte `EOSE`). **The refusal wakes the loop.** Recording the cause is not enough on its own. A budget refusal @@ -220,17 +231,21 @@ written. **The parked read holds its store stream, and that is what must change when a store is wired.** Stopping at the refusal without leaving the loop keeps the stream from `find_by_filters` alive until -the close retires the read; the version before this one dropped it at the refusal. Latent today — the -daemon wires `unavailable::stores()` (`crates/nostreamd/src/main.rs:113`), whose reads are stubs — -but not for ever: sqlx's `&Pool` executor draws a pooled connection for as long as a query stream is -alive, and `DB_MAX_POOL_SIZE` defaults to **3** (`crates/nostream-storage/src/postgres/mod.rs:362`), -so three refused reads parked at once would hold every connection in the pool and stall every other -read in the process. Whoever wires the store leaves the loop before parking — returning from -`fetch_and_send`, or dropping the stream first — and pays for it in test power: a handler that -returns wakes the loop on completion, so the close arrives promptly even with the refusal signal -deleted, and the wire test below then holds down nothing (static round 2 is exactly what that looked -like). That trade — a pooled connection against the signal being load-bearing — is recorded here so -it is made on purpose rather than by accident. +the loop aborts the read, a close frame later; the version before this one dropped it at the refusal. +Latent today — the daemon wires `unavailable::stores()` (`crates/nostreamd/src/main.rs:113`), whose +reads are stubs — but not for ever: sqlx's `&Pool` executor draws a pooled connection for as long as a +query stream is alive. The size that matters is the deployed one, not the code's: `DB_MAX_POOL_SIZE` +defaults to **3** (`crates/nostream-storage/src/postgres/mod.rs:362`) but +`deploy/docker-compose.prod.yml` sets **64** (`:16`) and configures no replica, so the connection a +refused read holds is one the master cannot serve a *write* with either — and the pool goes at one +connection per parked read: three read only because the code default is 3, sixty-four because the +deployed one is 64. Whoever wires the store leaves the +loop before parking — returning from `fetch_and_send`, or dropping the stream first — and pays for it +in test power: a handler that returns wakes the loop on completion, so the close arrives promptly +even with the refusal signal deleted, and the wire test in the wake-up paragraph above then holds +down nothing (static round 2 is exactly what that looked like). That trade — a pooled connection +against the signal being load-bearing — is recorded here so it is made on purpose rather than by +accident. ## The per-IP cap