diff --git a/.changeset/relay-concurrency-fixes.md b/.changeset/relay-concurrency-fixes.md new file mode 100644 index 000000000..05a36891f --- /dev/null +++ b/.changeset/relay-concurrency-fixes.md @@ -0,0 +1,26 @@ +--- +--- + +fix(relay): make a re-REQ keep its read, abort a read during the invite mint, and bound the outbound frame queue + +Three concurrency defects in `nostream-relay`, found after the wire port and fixed +together because they meet in one per-socket task: + +- A re-`REQ` replaces the read registered under a subscription id, and the replaced + read's cleanup removed the replacement's cancellation sender, so a re-subscribed + client could stop before `EOSE`. Reads now carry a connection-wide generation and + `end_read` removes only its own entry. +- A `CLOSE` that arrived while the NIP-43 invite mint was in flight aborted nothing: + the read was registered only after the mint returned. The read is now registered + before the mint's first `await` and its cancellation is checked before the stored + read starts. The invite `EVENT` is still emitted in that window, matching the + capture. +- The outbound frame queue was unbounded, and its only consumer was the loop that + performs the socket write. The queue is now bounded in bytes, a frame that does not + fit drops and flags the overflow, the client is closed with 1013 and a reason, and + every write the connection task performs is bounded by a write timeout so a blocked + write cannot starve the liveness probe. + +The TypeScript tree is untouched: no `src/` or `test/` file changes, so the published +package has no user-visible change. Decisions are recorded in +`docs/rewrite/decisions/0005-connection-concurrency.md`. diff --git a/crates/nostream-relay/src/connection.rs b/crates/nostream-relay/src/connection.rs index 5120468c5..48b827ea0 100644 --- a/crates/nostream-relay/src/connection.rs +++ b/crates/nostream-relay/src/connection.rs @@ -38,7 +38,47 @@ pub struct Connection { pub remote_address: String, subscriptions: Mutex>, session: Mutex, - in_flight: Mutex>>, + in_flight: Mutex, +} + +/// The reads running on one socket, each owned by the generation it was given. +/// +/// A re-`REQ` replaces the read for its id and signals the old one to abort; the old read +/// then returns through [`Connection::end_read`] carrying the generation it began with, and +/// that call must not remove its replacement's entry. The counter is connection-wide and +/// never reset, so a generation is unique for the socket's life and a fresh read under an +/// id an aborted read is still winding down from cannot be handed the same number (ADR +/// 0005). +#[derive(Debug, Default)] +struct InFlightReads { + /// The generation the next read is given. + next_generation: u64, + /// The live read per subscription id, with the generation that owns the entry. + by_subscription: HashMap)>, +} + +/// One read's ownership of its entry in the in-flight table: the subscription it streams, +/// the generation that owns the entry, and the cancellation signal it watches. +/// +/// It is passed to [`Connection::end_read`] as the read's last act, so ending a read is +/// something a read can do exactly once and only to its own entry. +#[derive(Debug)] +pub struct InFlightRead { + subscription_id: String, + generation: u64, + cancellation: watch::Receiver, +} + +impl InFlightRead { + /// The subscription this read streams. + pub fn subscription_id(&self) -> &str { + &self.subscription_id + } + + /// The cancellation signal: `true` once this read has been aborted. + pub fn cancellation(&mut self) -> &mut watch::Receiver { + &mut self.cancellation + } } impl Connection { @@ -84,7 +124,7 @@ impl Connection { remote_address, subscriptions: Mutex::new(HashMap::new()), session: Mutex::new(session), - in_flight: Mutex::new(HashMap::new()), + in_flight: Mutex::new(InFlightReads::default()), } } @@ -204,30 +244,58 @@ impl Connection { .authenticate(pubkey, auth_event_id, now) } - /// Starts an abortable read for `subscription_id` and returns its cancellation handle. + /// Starts an abortable read for `subscription_id` and returns the read's handle. /// /// A previous read for the same id is aborted first, so a re-REQ cannot leave two - /// streams writing to one subscription (`onClientMessage` replaces the handler). - pub fn begin_read(&self, subscription_id: &str) -> watch::Receiver { - let (sender, receiver) = watch::channel(false); + /// streams writing to one subscription (`onClientMessage` replaces the handler). The + /// new read is given the next generation, which is what makes + /// [`Connection::end_read`] remove only its own entry. + pub fn begin_read(&self, subscription_id: &str) -> InFlightRead { + let (sender, cancellation) = watch::channel(false); let mut in_flight = self .in_flight .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - if let Some(previous) = in_flight.insert(subscription_id.to_owned(), sender) { + let generation = in_flight.next_generation; + // A generation is unique for the socket's life; the wrap is unreachable, and a + // panic on it would be a panic in the accept path. + in_flight.next_generation = in_flight.next_generation.wrapping_add(1); + + if let Some((_, previous)) = in_flight + .by_subscription + .insert(subscription_id.to_owned(), (generation, sender)) + { let _ = previous.send(true); } - receiver + InFlightRead { + subscription_id: subscription_id.to_owned(), + generation, + cancellation, + } } /// Ends a read without aborting it (the stream finished). - pub fn end_read(&self, subscription_id: &str) { - self.in_flight + /// + /// The entry is removed only while `read`'s generation still owns it: a read that a + /// re-`REQ` replaced has nothing left to remove, and leaving the replacement's entry + /// alone is what keeps the replacement streaming to its `EOSE`. + pub fn end_read(&self, read: InFlightRead) { + let mut in_flight = self + .in_flight .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .remove(subscription_id); + .unwrap_or_else(|poisoned| poisoned.into_inner()); + + let owns_entry = in_flight + .by_subscription + .get(read.subscription_id()) + .map(|(generation, _)| *generation) + == Some(read.generation); + + if owns_entry { + in_flight.by_subscription.remove(read.subscription_id()); + } } /// Aborts the read running for `subscription_id`, if any (`abortInFlightSubscriptionReads`). @@ -237,7 +305,7 @@ impl Connection { .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - if let Some(sender) = in_flight.remove(subscription_id) { + if let Some((_, sender)) = in_flight.by_subscription.remove(subscription_id) { let _ = sender.send(true); } } @@ -247,6 +315,7 @@ impl Connection { self.in_flight .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) + .by_subscription .contains_key(subscription_id) } @@ -268,7 +337,7 @@ impl Connection { .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); - for (_, sender) in in_flight.drain() { + for (_, (_, sender)) in in_flight.by_subscription.drain() { let _ = sender.send(true); } } diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index 11d5581b9..53e1a10e0 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -4,7 +4,17 @@ //! `canSubscribe` limits with the capture's strings and its two typos, closes an //! all-restricted REQ from an unauthenticated client, registers the subscription, mints a //! NIP-43 invite code when the filters ask for one, then streams the stored events and ends -//! with `EOSE`. A `CLOSE` or a tripped message limiter aborts the read with no further frame. +//! with `EOSE`. A `CLOSE`, a socket close or a tripped message limiter aborts the read with +//! no further frame. +//! +//! The in-flight read is registered **before the mint's one `await`**, as the capture sets +//! `activeSubscriptionId` as its first statement (`subscribe-message-handler.ts:53`), so a +//! `CLOSE` handled after that registration aborts the stored read and the subscription +//! streams nothing (a `CLOSE` handled ahead of its own `REQ` finds nothing to abort: +//! cameri/nostream#958). The invite `EVENT` is emitted in that case: the capture's +//! emit is unguarded (`subscribe-message-handler.ts:111`) and attaches its abort signal only +//! to the stored read (`:230`), and parity wins where the two disagree absent a decision +//! record (ADR 0005). use futures_util::StreamExt; use nostream_core::expiration; @@ -13,6 +23,7 @@ use nostream_core::messages::{EoseMessage, OutgoingEventMessage, RelayMessage, S use nostream_core::nip42; use nostream_core::{Filter, Kind}; +use crate::connection::InFlightRead; use crate::error::RelayError; use crate::handlers::HandlerContext; @@ -54,9 +65,12 @@ pub async fn handle(message: SubscribeMessage, context: &HandlerContext) { filters: filters.clone(), }); + // Registered before the mint's only `await`, so the read a `CLOSE` aborts exists by then. + let read = context.connection.begin_read(&subscription_id); + maybe_issue_invite_code(&subscription_id, &filters, context).await; - fetch_and_send(&subscription_id, &filters, context).await; + fetch_and_send(read, &filters, context).await; } /// NIP-43: kind 28935 is never published by a client and never stored. A REQ for it asks the @@ -199,14 +213,24 @@ async fn mint_invite_code( } /// `fetchAndSend` (`:208-252`): the stored read, abortable, ending in `EOSE` unless aborted. -async fn fetch_and_send(subscription_id: &str, filters: &[Filter], context: &HandlerContext) { - let mut cancellation = context.connection.begin_read(subscription_id); +/// +/// The read is already in flight when this runs: it is registered before the invite mint, so +/// a `CLOSE` that landed during the mint has set its cancellation. +async fn fetch_and_send(mut read: InFlightRead, filters: &[Filter], context: &HandlerContext) { + let subscription_id = read.subscription_id().to_owned(); + + // A cancellation set before the stream was opened (a `CLOSE` during the mint) ends the + // read here: no `find_by_filters`, no frames, no `EOSE`. + if *read.cancellation().borrow() || read.cancellation().has_changed().is_err() { + context.connection.end_read(read); + return; + } let stream = match context.stores.events.find_by_filters(filters).await { Ok(stream) => stream, Err(error) => { tracing::error!(error = %error, "unable to stream events"); - context.connection.end_read(subscription_id); + context.connection.end_read(read); return; } }; @@ -218,16 +242,16 @@ async fn fetch_and_send(subscription_id: &str, filters: &[Filter], context: &Han |event: &nostream_core::Event| filters.iter().any(|filter| filter.matches(event)); loop { - if *cancellation.borrow() { - context.connection.end_read(subscription_id); + if *read.cancellation().borrow() { + context.connection.end_read(read); return; } let record = tokio::select! { - changed = cancellation.changed() => { + changed = read.cancellation().changed() => { // An aborted read sends no frame at all, not even EOSE. - if changed.is_err() || *cancellation.borrow() { - context.connection.end_read(subscription_id); + if changed.is_err() || *read.cancellation().borrow() { + context.connection.end_read(read); return; } @@ -261,7 +285,7 @@ async fn fetch_and_send(subscription_id: &str, filters: &[Filter], context: &Han context .sink .send(&RelayMessage::Event(OutgoingEventMessage { - subscription_id: subscription_id.to_owned(), + subscription_id: subscription_id.clone(), event: std::sync::Arc::new(event), })); } @@ -272,11 +296,11 @@ async fn fetch_and_send(subscription_id: &str, filters: &[Filter], context: &Han } } - context.connection.end_read(subscription_id); + context.connection.end_read(read); - context.sink.send(&RelayMessage::Eose(EoseMessage { - subscription_id: subscription_id.to_owned(), - })); + context + .sink + .send(&RelayMessage::Eose(EoseMessage { subscription_id })); } /// `canSubscribe` (`:258-301`): six reasons, the capture's strings — including diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index c2db2c52b..4d15a8356 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -33,8 +33,12 @@ use crate::relay_key::RelayKeySource; use crate::sink::FrameSink; use crate::transport; -/// The knobs a [`Relay`] is built with that are not settings (`liveness` and the NIP-56 -/// refresh are intervals the capture fixes in code, and tests shorten both). +/// 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. #[derive(Debug, Clone)] pub struct RelayConfig { /// `WSS_CLIENT_HEALTH_PROBE_INTERVAL` (`web-socket-server-adapter.ts:18`). @@ -43,16 +47,32 @@ pub struct RelayConfig { pub hidden_refresh_interval: Duration, /// `WS_DRAIN_TIMEOUT_MS`, resolved by `getWsDrainTimeoutMs`. pub drain_timeout: Duration, + /// The byte bound on one socket's outbound frame queue. The capture has no counterpart: + /// it queues every frame without a limit (`web-socket-adapter.ts:150-155`), so the bound + /// is the port's own policy (ADR 0005). + pub frame_queue_bytes: usize, + /// How long one socket write may take before the connection ends. The capture has no + /// counterpart either: its liveness probe is a process-wide `setInterval` a blocked write + /// cannot starve (`web-socket-server-adapter.ts:18,46`), while this loop performs the + /// write itself. + pub write_timeout: Duration, } impl Default for RelayConfig { /// The capture's own constants, and no environment read: `WS_DRAIN_TIMEOUT_MS` is - /// resolved by [`transport::drain_timeout`], which the composition root calls (ADR 0004). + /// resolved by [`transport::drain_timeout`], which the composition root calls (ADR 0004); + /// the queue bound and the write timeout are the port's additions (ADR 0005). fn default() -> Self { Self { liveness_interval: Duration::from_millis(120_000), hidden_refresh_interval: DEFAULT_REFRESH_INTERVAL, drain_timeout: Duration::from_millis(transport::DEFAULT_WS_DRAIN_TIMEOUT_MS), + // 8 MiB: above any single outbound frame, and a few hundred times below the + // 405 MiB an unbounded queue reached in the measurement that motivated it. + frame_queue_bytes: 8 * 1024 * 1024, + // 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), } } } @@ -87,8 +107,9 @@ 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 capture's constants, - /// and the environment read is [`transport::drain_timeout`] at the call site. + /// 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. pub fn new( settings: Settings, stores: Arc, @@ -255,7 +276,7 @@ impl Relay { peer: std::net::SocketAddr, shutdown: ShutdownState, ) { - let (sink, mut frames) = FrameSink::channel(); + let (sink, mut frames) = FrameSink::channel(self.config.frame_queue_bytes); let connection_context = ConnectionContext { settings: Arc::clone(&self.settings), @@ -296,11 +317,28 @@ impl Relay { break; } + // 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; + break; + } + tokio::select! { frame = frames.recv() => { let Some(text) = frame else { break }; - if transport::send_text(&mut transport, text).await.is_err() { + let bytes = text.len(); + let written = tokio::time::timeout( + self.config.write_timeout, + transport::send_text(&mut transport, text), + ) + .await; + + sink.release(bytes); + + if !matches!(written, Ok(Ok(()))) { break; } } @@ -318,7 +356,13 @@ impl Relay { // alive (`web-socket-adapter.ts:342-346`). alive = true; - if transport::send_pong(&mut transport, payload).await.is_err() { + let pong = tokio::time::timeout( + self.config.write_timeout, + transport::send_pong(&mut transport, payload), + ) + .await; + + if !matches!(pong, Ok(Ok(()))) { break; } } @@ -358,7 +402,13 @@ impl Relay { alive = false; - if transport.send(Message::Ping(Vec::new().into())).await.is_err() { + let ping = tokio::time::timeout( + self.config.write_timeout, + transport.send(Message::Ping(Vec::new().into())), + ) + .await; + + if !matches!(ping, Ok(Ok(()))) { break; } } @@ -371,6 +421,28 @@ 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()`). + /// + /// 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( + &self, + transport: &mut AcceptedTransport, + peer: std::net::SocketAddr, + ) { + tracing::debug!(%peer, "closing a client that cannot keep up with its frame queue"); + + let _ = tokio::time::timeout( + self.config.write_timeout, + transport.close(Some(transport::close_frame( + CloseCode::Again, + transport::SLOW_CONSUMER_REASON, + ))), + ) + .await; + } + /// The close frame an inbound error implies: 1009 for a message over the limit, 1002 for /// a protocol violation (an RSV bit with no negotiated extension), nothing otherwise. async fn close_on_error( @@ -380,12 +452,20 @@ impl Relay { ) { if transport::is_message_too_long(error) { tracing::debug!("client sent a payload too large"); - let _ = transport.close(Some(transport::too_big_frame())).await; + + let _ = tokio::time::timeout( + self.config.write_timeout, + transport.close(Some(transport::too_big_frame())), + ) + .await; } else if transport::is_protocol_error(error) { tracing::debug!(error = %error, "websocket protocol error"); - let _ = transport - .close(Some(transport::protocol_error_frame())) - .await; + + let _ = tokio::time::timeout( + self.config.write_timeout, + transport.close(Some(transport::protocol_error_frame())), + ) + .await; } else { tracing::debug!(error = %error, "websocket closed"); } @@ -464,20 +544,25 @@ impl Relay { params: None, }); - if transport::send_text(transport, frame.to_text()) - .await - .is_err() - { + let written = tokio::time::timeout( + self.config.write_timeout, + transport::send_text(transport, frame.to_text()), + ) + .await; + + if !matches!(written, Ok(Ok(()))) { return; } } - let _ = transport - .close(Some(transport::close_frame( + let _ = tokio::time::timeout( + self.config.write_timeout, + transport.close(Some(transport::close_frame( CloseCode::Away, transport::DRAIN_REASON, - ))) - .await; + ))), + ) + .await; } } diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index 2dfa8be7f..9b8ed5400 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -13,6 +13,20 @@ //! //! A send on a socket that is no longer open is dropped silently, which is what the //! capture's `readyState !== OPEN` guard does: it returns without an error. +//! +//! **The queue is bounded in bytes**, which the capture does not do: it queues every frame +//! and never pushes back (`src/utils/stream.ts:11-18`), so a stored read of large events can +//! grow the process without limit while a socket drains at link speed. Here a frame is +//! admitted only while the queued bytes stay within the bound the sink was built with, and a +//! frame larger than the bound is still admitted into an empty queue so one big event does +//! not close every client. A frame that does not fit is dropped, the sink records the +//! overflow and each racing sender that sets the flag warns once with the bound and the +//! 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). + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use tokio::sync::mpsc; @@ -22,27 +36,41 @@ use nostream_core::messages::RelayMessage; #[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, } impl FrameSink { - /// A sink whose frames go to `sender`. - pub fn new(sender: mpsc::UnboundedSender) -> Self { - Self { - sender: Some(sender), - } - } - - /// Creates the queue and its sink. The receiver is drained by the connection task. - pub fn channel() -> (Self, mpsc::UnboundedReceiver) { + /// 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) { let (sender, receiver) = mpsc::unbounded_channel(); - (Self::new(sender), receiver) + ( + Self { + sender: Some(sender), + bound_bytes, + queued_bytes: Arc::new(AtomicUsize::new(0)), + overflowed: Arc::new(AtomicBool::new(false)), + }, + receiver, + ) } /// A sink that is already closed: every send is dropped silently. Used by tests that /// assert the "not open" behaviour without a socket. pub fn closed() -> Self { - Self { sender: None } + Self { + sender: None, + bound_bytes: 0, + queued_bytes: Arc::new(AtomicUsize::new(0)), + overflowed: Arc::new(AtomicBool::new(false)), + } } /// Sends one frame. A closed socket drops the frame and is not an error. @@ -51,8 +79,185 @@ impl FrameSink { 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); + + // 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) { + 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); + tracing::warn!( + bound_bytes = self.bound_bytes, + queued_bytes = queued, + "outbound frame queue full: dropping frames and closing the client" + ); + return; + } + + match self.queued_bytes.compare_exchange_weak( + queued, + queued + length, + Ordering::SeqCst, + Ordering::SeqCst, + ) { + Ok(_) => break, + // Another sender got in first: judge the bound against what it left. + Err(actual) => queued = actual, + } + } + // 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(message.to_text()); + let _ = sender.send(text); + } + + /// Releases a frame's bytes once it has been written to the socket. + pub(crate) fn release(&self, bytes: usize) { + let _ = self + .queued_bytes + .try_update(Ordering::SeqCst, Ordering::SeqCst, |queued| { + Some(queued.saturating_sub(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) + } + + /// The bytes admitted into the queue and not yet written. + pub fn queued_bytes(&self) -> usize { + self.queued_bytes.load(Ordering::SeqCst) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use nostream_core::messages::NoticeMessage; + + fn notice(message: &str) -> RelayMessage { + RelayMessage::Notice(NoticeMessage { + message: message.to_owned(), + }) + } + + #[test] + fn the_queue_is_bounded_in_bytes_and_reports_the_overflow() { + let frame = notice("bounded"); + 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); + + sink.send(&frame); + sink.send(&frame); + + assert!(!sink.overflowed()); + assert_eq!(sink.queued_bytes(), 2 * len); + assert!(frames.try_recv().is_ok()); + assert!(frames.try_recv().is_ok()); + assert!(frames.try_recv().is_err()); + + sink.send(&frame); + + assert!(sink.overflowed()); + assert_eq!( + sink.queued_bytes(), + 2 * len, + "the dropped frame is not counted" + ); + assert!( + frames.try_recv().is_err(), + "the dropped frame was never queued" + ); + + // Once the flag is set every later send is dropped as well. + sink.send(&frame); + + assert!(sink.overflowed()); + 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); + + sink.send(&frame); + + assert!(!sink.overflowed()); + assert_eq!(sink.queued_bytes(), len); + assert!(frames.try_recv().is_ok()); + + sink.send(&frame); + + assert!(sink.overflowed()); + assert_eq!(sink.queued_bytes(), len); + assert!(frames.try_recv().is_err()); + + // Several in-flight reads share one sink, so the accounting has to survive concurrent + // senders: every frame fits the bound and every frame's bytes are counted once. The + // barrier makes the sends overlap but cannot force the interleaving a lost update + // needs, so this is a regression guard rather than a proof. + const SENDERS: usize = 8; + + let (sink, mut frames) = FrameSink::channel(SENDERS * len); + let start = std::sync::Barrier::new(SENDERS); + + std::thread::scope(|scope| { + for _ in 0..SENDERS { + let sink = sink.clone(); + let frame = &frame; + let start = &start; + + scope.spawn(move || { + start.wait(); + sink.send(frame); + }); + } + }); + + assert!(!sink.overflowed(), "every frame fits the bound"); + assert_eq!(sink.queued_bytes(), SENDERS * len); + + for _ in 0..SENDERS { + assert!(frames.try_recv().is_ok()); + } + + assert!(frames.try_recv().is_err()); + + // `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); + + sink.send(&frame); + sink.send(&frame); + sink.release(2 * len); + + assert_eq!(sink.queued_bytes(), 0); + assert!(!sink.overflowed()); + + sink.send(&frame); + sink.send(&frame); + + assert_eq!(sink.queued_bytes(), 2 * len); + assert!(frames.try_recv().is_ok()); + assert!(frames.try_recv().is_ok()); + assert!(frames.try_recv().is_ok()); + assert!(frames.try_recv().is_ok()); + assert!(frames.try_recv().is_err()); } } diff --git a/crates/nostream-relay/src/testing.rs b/crates/nostream-relay/src/testing.rs index af2192671..d03d90350 100644 --- a/crates/nostream-relay/src/testing.rs +++ b/crates/nostream-relay/src/testing.rs @@ -638,6 +638,10 @@ impl Nip05VerificationRepository for InMemoryNip05Repository { #[derive(Debug, Default)] pub struct InMemoryInviteCodeRepository { codes: Mutex>, + /// When set, `create` holds until a permit is released, which is how a test keeps the + /// invite mint's one `await` open long enough for a `CLOSE` to land in it. + create_gate: Mutex>>, + creates: AtomicU64, } impl InMemoryInviteCodeRepository { @@ -646,6 +650,27 @@ impl InMemoryInviteCodeRepository { Arc::new(Self::default()) } + /// Makes every `create` wait for a permit. Grant one with `Semaphore::add_permits` to + /// let the mint through. + /// + /// A gate rather than a sleeping double: `RateLimits::invite_limited` is synchronous and + /// a sleep in it would block the runtime a single-threaded `#[tokio::test]` runs on. + pub fn gate_creates(&self) -> Arc { + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + + *self + .create_gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(Arc::clone(&gate)); + + gate + } + + /// How many `create` calls have started, including one a gate is holding. + pub fn creates_started(&self) -> u64 { + self.creates.load(Ordering::SeqCst) + } + /// The codes minted, in code order, with the options each was minted with. pub fn minted(&self) -> Vec<(String, CreateInviteCodeOptions)> { self.codes @@ -664,6 +689,18 @@ impl InviteCodeRepository for InMemoryInviteCodeRepository { options: &'a CreateInviteCodeOptions, ) -> StorageFuture<'a, InviteCode> { Box::pin(async move { + self.creates.fetch_add(1, Ordering::SeqCst); + + let gate = self + .create_gate + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(); + + if let Some(gate) = gate { + let _ = gate.acquire().await; + } + if self .codes .lock() diff --git a/crates/nostream-relay/src/transport.rs b/crates/nostream-relay/src/transport.rs index 3f11440be..e88706f81 100644 --- a/crates/nostream-relay/src/transport.rs +++ b/crates/nostream-relay/src/transport.rs @@ -34,6 +34,10 @@ pub const DEFAULT_WS_DRAIN_TIMEOUT_MS: u64 = 30_000; /// The close reason a draining server sends (`drainAndClose`). pub const DRAIN_REASON: &str = "relay shutting down"; +/// The close reason a client whose outbound frame queue overflowed is closed with: the port's +/// own policy, with no counterpart in the capture (ADR 0005). +pub const SLOW_CONSUMER_REASON: &str = "slow consumer: outbound frame queue full"; + /// `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/common/mod.rs b/crates/nostream-relay/tests/common/mod.rs index ddc732cda..5958d79d0 100644 --- a/crates/nostream-relay/tests/common/mod.rs +++ b/crates/nostream-relay/tests/common/mod.rs @@ -113,6 +113,7 @@ pub fn fast_config() -> RelayConfig { liveness_interval: Duration::from_secs(600), hidden_refresh_interval: Duration::from_secs(600), drain_timeout: Duration::from_millis(300), + ..RelayConfig::default() } } diff --git a/crates/nostream-relay/tests/connection.rs b/crates/nostream-relay/tests/connection.rs index f4c19e49c..43f01c1b4 100644 --- a/crates/nostream-relay/tests/connection.rs +++ b/crates/nostream-relay/tests/connection.rs @@ -185,20 +185,49 @@ fn a_read_is_in_flight_until_it_ends_and_an_abort_notifies_it() { assert!(!connection.read_in_flight("s1")); - let cancelled = connection.begin_read("s1"); + let mut read = connection.begin_read("s1"); assert!(connection.read_in_flight("s1")); connection.abort_read("s1"); - assert!(*cancelled.borrow()); + assert!(*read.cancellation().borrow()); assert!(!connection.read_in_flight("s1")); - let cancelled = connection.begin_read("s2"); - connection.end_read("s2"); + let read = connection.begin_read("s2"); + connection.end_read(read); - assert!(!*cancelled.borrow()); assert!(!connection.read_in_flight("s2")); + + let mut read = connection.begin_read("s3"); + + assert!(!*read.cancellation().borrow()); + connection.end_read(read); +} + +#[test] +fn a_finished_read_does_not_remove_the_read_that_replaced_it() { + let connection = connection(); + + let mut first = connection.begin_read("s1"); + let mut second = connection.begin_read("s1"); + + assert!( + *first.cancellation().borrow(), + "the replaced read is told to abort" + ); + + connection.end_read(first); + + assert!( + connection.read_in_flight("s1"), + "the replacement is still in flight" + ); + assert!( + second.cancellation().has_changed().is_ok(), + "the replacement's sender is alive" + ); + assert!(!*second.cancellation().borrow()); } #[test] @@ -211,14 +240,17 @@ fn close_clears_the_table_the_session_and_the_reads() { filters: vec![Filter::default()], }); connection.authenticate(pubkey, EventId::from_byte_array([1; 32]), NOW_SECS); - let cancelled = connection.begin_read("s1"); + let mut read = connection.begin_read("s1"); connection.close(); assert_eq!(connection.subscription_count(), 0); assert!(connection.authenticated_pubkeys(NOW_SECS).is_empty()); assert!(!connection.read_in_flight("s1")); - assert!(*cancelled.borrow(), "an in-flight read is aborted"); + assert!( + *read.cancellation().borrow(), + "an in-flight read is aborted" + ); } // --------------------------------------------------------------------------------------- @@ -378,7 +410,8 @@ async fn closing_the_socket_aborts_an_in_flight_read() { .seed_event(signed_event(&CLIENT_SECRET, 1, vec![], "stored", NOW_SECS)) .await; - let (sink, mut frames) = FrameSink::channel(); + // The queue's bound plays no part here: the frames are a few hundred bytes. + let (sink, mut frames) = FrameSink::channel(1024 * 1024); let connection = Arc::new(Connection::new( "127.0.0.1".to_owned(), nostream_core::nip42::Nip42Session::with_challenge("challenge", 0), @@ -458,7 +491,7 @@ async fn restricted_read_with(authenticate_mid_read: bool) -> Vec { )) .await; - let (sink, mut frames) = FrameSink::channel(); + let (sink, mut frames) = FrameSink::channel(1024 * 1024); // 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. diff --git a/crates/nostream-relay/tests/handlers.rs b/crates/nostream-relay/tests/handlers.rs index fab86e40c..5ace1fede 100644 --- a/crates/nostream-relay/tests/handlers.rs +++ b/crates/nostream-relay/tests/handlers.rs @@ -1090,6 +1090,73 @@ async fn an_invite_request_mints_one_kind_28935_event_before_eose() { assert_eq!(frames[1], r#"["EOSE","invite"]"#); } +#[tokio::test] +async fn a_close_during_the_invite_mint_leaves_the_event_but_not_the_read() { + let mut settings = settings(); + settings.nip43 = Some(nip43_settings()); + + let fixture = Fixture::new(settings); + let gate = fixture.invites.gate_creates(); + fixture + .seed_event(signed_event(&CLIENT_SECRET, 1, vec![], "stored", NOW_SECS)) + .await; + + let server = start(fixture).await; + let (mut client, challenge) = server.connect().await; + + let auth = auth_event(&challenge, NOW_SECS); + client + .send(&format!( + r#"["AUTH",{}]"#, + serde_json::to_string(&auth).expect("auth json") + )) + .await; + assert!(client.expect_frame().await.ends_with(r#"true,""]"#)); + + client.send(r#"["REQ","invite",{"kinds":[28935,1]}]"#).await; + + // The mint's one `await` is held, which is the window a `CLOSE` lands in. + for _ in 0..200 { + if server.fixture.invites.creates_started() == 1 { + break; + } + + tokio::time::sleep(Duration::from_millis(5)).await; + } + + assert_eq!( + server.fixture.invites.creates_started(), + 1, + "the invite mint is in flight" + ); + + client.send(r#"["CLOSE","invite"]"#).await; + tokio::time::sleep(Duration::from_millis(50)).await; + gate.add_permits(1); + + // The capture emits the invite event for a subscription the client has closed + // (`subscribe-message-handler.ts:111`), and that stays. + let first = client.expect_frame().await; + assert!(first.starts_with(r#"["EVENT","invite","#), "{first}"); + + let event: serde_json::Value = serde_json::from_str( + first + .strip_prefix(r#"["EVENT","invite","#) + .and_then(|rest| rest.strip_suffix(']')) + .expect("an EVENT frame"), + ) + .expect("event json"); + assert_eq!(event["kind"], 28_935); + + // Nothing else: the stored read was cancelled with the subscription, so no stored + // event and no `EOSE`. + assert!( + client.next_message().await.is_none(), + "a closed subscription streams nothing" + ); + assert_eq!(server.fixture.invites.minted().len(), 1); +} + #[tokio::test] async fn an_unauthenticated_invite_request_gets_eose_only() { let mut settings = settings(); diff --git a/crates/nostream-relay/tests/limits.rs b/crates/nostream-relay/tests/limits.rs index 28ac9407d..80263b24f 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(); + let (sink, mut frames) = nostream_relay::sink::FrameSink::channel(1024 * 1024); 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/transport.rs b/crates/nostream-relay/tests/transport.rs index 0a10e4b4f..f3da1d547 100644 --- a/crates/nostream-relay/tests/transport.rs +++ b/crates/nostream-relay/tests/transport.rs @@ -17,7 +17,9 @@ mod common; use std::time::Duration; -use common::{Client, Server, fast_config, liveness_config, relay_keys}; +use common::{ + CLIENT_SECRET, Client, NOW_SECS, Server, fast_config, liveness_config, relay_keys, signed_event, +}; use futures_util::SinkExt; use nostream_relay::testing::{Fixture, settings}; use nostream_relay::transport::{self, DEFAULT_MAX_PAYLOAD}; @@ -586,3 +588,97 @@ async fn a_connection_with_no_subscriptions_is_closed_without_a_closed_frame() { other => panic!("expected only the 1001 close, got {other:?}"), } } + +// --------------------------------------------------------------------------------------- +// The bounded frame queue (#952) +// --------------------------------------------------------------------------------------- + +#[tokio::test] +async fn a_client_that_stops_reading_is_closed_with_1013() { + // 32 MiB of events against the default 8 MiB queue: more than any autotuned loopback + // buffer can absorb, so the bound is reached whatever the kernel's window is. The + // timestamps differ so every event is a distinct row. + let content = "x".repeat(32 * 1024); + let fixture = Fixture::new(settings()); + // One permit per stored record, so the prefix below is a frame the client can read rather + // than one of the whole 32 MiB the read would otherwise queue in a single poll. + let gate = fixture.events.gate_reads_per_item(); + + for index in 0..1024 { + fixture + .seed_event(signed_event( + &CLIENT_SECRET, + 1, + vec![], + &content, + NOW_SECS + index, + )) + .await; + } + + let server = Server::start_with(fixture, relay_keys(), fast_config()).await; + 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 + // connection: read it here so nothing below counts it as a stored event. + let challenge = raw.next_frame().await.expect("the AUTH challenge"); + assert_eq!(challenge.opcode, 0x1); + assert!( + challenge.text().starts_with(r#"["AUTH","#), + "the first frame is the challenge: {}", + challenge.text() + ); + + raw.send_text(r#"["REQ","bulk",{"kinds":[1]}]"#).await; + + // Three stored events while the client is still reading: these are the frames AC6 step 5 + // means by "at least one `EVENT` frame preceded it". + gate.add_permits(3); + + let mut delivered = Vec::new(); + + for _ in 0..3 { + let frame = tokio::time::timeout(Duration::from_secs(5), raw.next_frame()) + .await + .expect("a stored event arrives while the client reads") + .expect("a stored event frame"); + + assert_eq!(frame.opcode, 0x1, "{frame:?}"); + delivered.push(frame.text()); + } + + for text in &delivered { + let event: serde_json::Value = serde_json::from_str( + text.strip_prefix(r#"["EVENT","bulk","#) + .and_then(|rest| rest.strip_suffix(']')) + .expect("an EVENT frame for bulk"), + ) + .expect("event json"); + + assert_eq!(event["kind"], 1, "the stored events are the ones asked for"); + } + + // The client stops reading and the rest of the read runs at once: the queue takes what + // the socket cannot, overflows, and the loop closes the client as soon as the blocked + // write returns. + gate.add_permits(1024); + tokio::time::sleep(Duration::from_millis(500)).await; + + let frames = tokio::time::timeout(Duration::from_secs(10), raw.frames_until(0x8)) + .await + .expect("the close frame arrives once the queue overflows"); + let last = frames.last().expect("a close frame"); + + assert_eq!(last.opcode, 0x8, "the last frame is a close frame"); + assert!( + frames.iter().all(|frame| { + frame.opcode == 0x8 || frame.text().starts_with(r#"["EVENT","bulk","#) + }), + "everything the closed client still receives is a stored event: {:?}", + frames.iter().map(Frame::text).collect::>() + ); + assert_eq!( + close_of(last), + (1013, transport::SLOW_CONSUMER_REASON.to_owned()) + ); +} diff --git a/crates/nostream-relay/tests/wire_parity.rs b/crates/nostream-relay/tests/wire_parity.rs index 03fc642c2..6b37f4e31 100644 --- a/crates/nostream-relay/tests/wire_parity.rs +++ b/crates/nostream-relay/tests/wire_parity.rs @@ -245,6 +245,7 @@ impl Server { ), hidden_refresh_interval: Duration::from_secs(600), drain_timeout: Duration::from_millis(300), + ..RelayConfig::default() }; let relay = Relay::new( diff --git a/docs/rewrite/decisions/0005-connection-concurrency.md b/docs/rewrite/decisions/0005-connection-concurrency.md new file mode 100644 index 000000000..a4c9f090a --- /dev/null +++ b/docs/rewrite/decisions/0005-connection-concurrency.md @@ -0,0 +1,152 @@ +# ADR 0005: Connection concurrency — read ownership, a bounded frame queue, and liveness under a blocked write (job 010) + +- **Status:** accepted (job 010, Phase 2) +- **Deciders:** the rewrite team, on the evidence in job 010's recon and spec +- **Date:** 2026-10-24 + +## Context + +One task per socket owns the transport, and three of its concurrent relationships were +wrong. A re-`REQ` replaces the read registered for a subscription id and aborts the old +one, and the old read's own cleanup then removed the *replacement's* cancellation sender, +so a re-subscribed client could go silent without an `EOSE`. A `CLOSE` that landed while +the NIP-43 invite mint was in flight aborted nothing, because the read was registered only +after that mint returned, so the handler streamed a stored read the client had already +closed. And the outbound frame queue was an unbounded channel whose only consumer was the +same select loop that performs the socket write, so a client that stopped reading let the +queue grow without limit while the liveness probe could not run. + +Each fix had more than one plausible shape. This ADR records the shape chosen, the shape +rejected, and why — the code shows the first and cannot show the second. + +## Read ownership — a connection-wide generation, not a token + +The in-flight table records an owner with each entry: one mutex over a connection-wide +`next_generation: u64` and a map from subscription id to `(generation, watch::Sender)`. +`begin_read` hands back a handle carrying its id, its generation and the receiver; +`end_read` takes the handle by value and removes the entry only while the stored +generation still matches, so a replaced read's last act is a no-op and the replacement +keeps its sender. `abort_read`, `read_in_flight` and `close` keep their present +signatures and behaviour. + +The generation is a counter and not a per-read token. Identity comparison is available — +`watch::Sender::same_channel` and `watch::Receiver::same_channel` — so an entry holding the +sender could end a read by comparing the returning read's handle against it. What rejects +that route is cost, not impossibility: `watch::channel(false)` allocates its shared state +once per read either way, and identity would add a *second* watch handle to every entry (the +sender, plus a receiver the entry never awaits), while the generation adds one `u64` beside +the one handle the table already holds. The generation is also the value a returning read can +still carry after its own handle has been replaced, and the value that names the owner in the +table; the rule is tested through behaviour (`tests/connection.rs:208-231`), not by value. It +fits the locking the file already has: `impl Connection` contains no `.await`, so no lock is +ever held across a suspension point, and the counter is read and written inside the same +`insert` that replaces the previous sender. No second lock, no allocation outside the guard, +and no new state for a caller to keep. + +The counter is connection-wide and never reset, rather than derived per id from the +previous entry. `abort_read` removes an entry for a read that is still winding down, so a +per-id "previous + 1" would restart at one and hand a fresh read the same number as the +aborted one — the defect again, one call later. Wrapping is unreachable (2^64 reads on one +socket) and is written with `wrapping_add` rather than a panic, because a panic there +would be a panic in the accept path. + +The alternative shape — a guard owning the entry and releasing it on `Drop` — was +rejected as larger for the same property: it changes the call sites and the lifetime of +the cancel signal, and gives nothing the generation does not, since a read already has +exactly one point where it ends. + +## The outbound frame queue — a byte bound and a disconnect, not a pause + +`FrameSink` counts the bytes it has admitted and drops a frame that would exceed its +bound, unless the queue is empty, so one frame larger than the whole bound still reaches the +client instead of closing every client on a large event. A dropped frame sets an overflow +flag and the senders that set it each log one warning carrying the bound and the current +depth (one line per racing sender, not one per connection); once the flag is set every later +send is dropped without queueing. The draining side releases a frame's bytes when its write +completes. The default bound is 8 MiB, which is above any single outbound frame and a few +hundred times below the 405 MiB an unbounded queue reached when 200 000 frames were queued +with nothing draining. + +The full-queue policy is **disconnect**: the connection loop closes the client with 1013 and +the fixed reason `slow consumer: outbound frame queue full` once its socket accepts bytes +again — a client whose socket never does is ended by `write_timeout` instead, with no close +frame — and then falls through to the existing cleanup. + +What the client experiences at that point is a truncation, not a clean stop: the frames already +written to its socket still arrive, in order, and then the close; frames the sink had admitted +and this run never wrote are frames it will never read, and nothing in the stream marks where +the cut fell — no `EOSE`, no count, only the close frame. Draining is not the alternative for +the client this bound exists for — one that has stopped reading — because draining means writing +to a socket that is not taking bytes, which is the condition that produced the overflow; the +relay would be waiting on precisely the client it has already failed to serve. That is the trade +the policy makes: some frames one client will never read, in exchange for a relay that keeps +serving everyone else. + +Three alternatives lost: + +- **Pause the producer** (a bounded channel and a permit the sender waits for). It is a + self-deadlock here: `FrameSink::send` is synchronous and `dispatch_live` — which sends to + every matching subscription — is called from the very loop that drains the queue, so a + permit wait on that path waits on the task that would release it. Working around it + needs a second write path, which ADR 0004 forbids: one function turns a frame into + outbound bytes. And it would not fix liveness on its own; the same write timeout below + would still be needed, so it is more change for the same answer. +- **Silent break** (the shape the liveness path uses for a missed pong). The client cannot + tell a slow link from a relay that forgot it, and a test cannot tell a policy from a + crash. A close frame with a code and a reason is observable to both. +- **A reader task and a writer task per socket.** A larger restructuring of the connection + loop and the drain for the same liveness bound the write timeout gives for a fraction of + the diff; it also moves the send half of the socket out of the task that owns the + connection's state, which every handler's sink would have to follow. + +The bound and the timeout are `RelayConfig` fields and not settings. The capture has no +counterpart for either — it queues every frame without a limit +(`src/adapters/web-socket-adapter.ts:150-155`; its pipeline never pushes back, +`src/utils/stream.ts:11-18`) and its heartbeat is a process-wide interval — 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 — with the tests setting the cadences they +need — and these two port-only policies, which every test so far runs at their defaults. + +## Liveness under a blocked write — every write carries a timeout + +The write happens inside the select arm, so while it is pending the same task cannot poll +the liveness tick, the next inbound frame, the live-event ring or the drain flag. A +blocked write therefore starves the probe that exists to notice a dead client; an +inbound-only probe cannot help, because a client that stops reading is exactly a client +whose socket accepts nothing. + +Every write the connection task performs — a queued frame, the pong reply, the liveness +ping, each `CLOSED` frame and the close frame in the drain — is wrapped in +`tokio::time::timeout(config.write_timeout, ..)`, and a write that elapses ends the +connection. The decision is that a write unable to complete within `write_timeout` is +itself evidence that the client is not reading, so waiting for the liveness arm would only +delay the same outcome. The default is 30 s against the capture's 120 s liveness interval: +shorter, so the write path ends a stuck client before a missed pong could, and long enough +for a large frame on a slow link. Because no write the connection task performs can hold it +longer than that timeout — and the drain's own writes are further bounded by `drain_timeout` +from the serve loop — the tick can be delayed by at most 30 s, which is the interval's own +order of magnitude rather than the unbounded stall it was. + +The overflow flag is read at the top of every loop iteration. No `await` sits between a +write returning and that check, so a flag a producer set while the loop was parked is acted +on as soon as the write completes. + +## The invite emit stays unguarded + +The read is registered before the invite mint's only `await`, and its cancellation is +checked before the stored read opens, so a `CLOSE` that is *handled after that registration* +ends the subscription with no stored event and no `EOSE`; one whose handler is scheduled +ahead of its own `REQ` finds nothing to abort, which is pre-existing and tracked as +cameri/nostream#958. The invite `EVENT` is nonetheless still +emitted for that closed subscription. That is deliberate parity, not an oversight: the +capture emits the invite event unguarded +(`src/handlers/subscribe-message-handler.ts:111`) and attaches its abort signal only to the +stored read (`:230`), while it sets the abortable subscription id as the handler's first +statement (`:53`). The TypeScript tree wins where the two trees disagree absent a decision +record, so the fix must not reach for the emit; the exposure it leaves is one stray +`EVENT`, which is the capture's own. + +## Not decided here — the port's `CLOSE` semantics + +Whether a plain `CLOSE` aborts a stored read is out of scope for this record; it is tracked +as cameri/nostream#957.