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/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index b0ca6cf0..6644a8f3 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -268,6 +268,26 @@ 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. + // + // 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); + return; + } + let record = tokio::select! { changed = read.cancellation().changed() => { // An aborted read sends no frame at all, not even EOSE. @@ -408,3 +428,105 @@ fn can_subscribe( None } + +#[cfg(test)] +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; + + 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 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 + .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 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::clone(&connection), + sink, + )); + + let message = SubscribeMessage { + subscription_id: "s".to_owned(), + filters: vec![Filter { + kinds: Some(vec![Kind(1)]), + ..Filter::default() + }], + }; + + let mut read = begin(message, &context).expect("the subscription is registered"); + gate.add_permits(3); + + assert!( + 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"); + } +} 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..2dcd1f3b 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`, `frame_memory_budget_bytes`, +/// `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 { /// `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; } @@ -420,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 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() => {} Some(_) = handlers.join_next(), if !handlers.is_empty() => {} } } @@ -428,24 +495,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..9cc26072 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -24,39 +24,210 @@ //! 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 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::future::Future; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU8, AtomicU64, AtomicUsize, Ordering}; -use tokio::sync::mpsc; +use tokio::sync::{Notify, 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, + /// 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>, +} + +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, + }; + + let first = self + .refusal + .compare_exchange(NO_REFUSAL, cause, Ordering::SeqCst, Ordering::SeqCst) + .is_ok(); + + if first { + self.refusal_signal.notify_one(); + } + + first + } +} + +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), + refusal_signal: Notify::new(), + budget, + }), }, receiver, ) @@ -68,37 +239,49 @@ 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), + refusal_signal: Notify::new(), + 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, 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.overflowed.load(Ordering::SeqCst) { + if self.account.refusal() == Some(Refusal::QueueFull) { 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 +290,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 +302,69 @@ 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); + } + } + + /// 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() } - /// 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) + /// 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.queued_bytes.load(Ordering::SeqCst) + self.account.queued_bytes.load(Ordering::SeqCst) } } @@ -162,12 +385,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 +398,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 +412,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 +437,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 +453,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 +464,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 +483,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 finished = 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; + 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 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(); + + while finished.load(Ordering::SeqCst) < SENDERS { + max_seen.fetch_max(budget.accounted(), Ordering::SeqCst); + 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..1258236e 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,314 @@ 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); + } +} + +#[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); } 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..962b8814 --- /dev/null +++ b/docs/rewrite/decisions/0008-frame-memory-budget.md @@ -0,0 +1,332 @@ +# 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: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`). + +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: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. + +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: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`): + +- `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. 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 +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. + +**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: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 +can arrive while *this* socket's queue is empty — the bytes that left no room are another +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 +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: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. + +**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 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` +(`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: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. + +**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 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 +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 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 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 + +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-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 +`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 (`: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 +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. + +**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 +`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: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`). + +**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`). +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.