diff --git a/.changeset/relay-inbound-dispatch-order.md b/.changeset/relay-inbound-dispatch-order.md new file mode 100644 index 00000000..95800469 --- /dev/null +++ b/.changeset/relay-inbound-dispatch-order.md @@ -0,0 +1,21 @@ +--- +--- + +fix(relay): order an inbound frame's prefix before the next frame is read + +A per-socket handler ran as an independently scheduled task, so two frames read in order +could run in either order: a `CLOSE` handled ahead of its own `REQ` found nothing to abort +and the subscription kept streaming live events to a closed id, and a restricted `REQ` +pipelined behind its `AUTH` was refused for want of a session that had been sent but not +yet handled (cameri/nostream#958). + +An inbound frame now runs its ordering-sensitive prefix — `subscribe` through its read +registration, `unsubscribe`, and `auth` — on the connection loop before the next frame is +read. The rest of the handler is spawned and stays concurrent, so a `CLOSE` during an +invite mint still aborts the stored read and the invite `EVENT` is still emitted. `EVENT` +and `COUNT` are unchanged. + +The TypeScript tree is untouched: no `src/` or `test/` file changes, so the published +package has no user-visible change. The decision, the two rejected shapes and the +guarantee's boundary are recorded in +`docs/rewrite/decisions/0005-connection-concurrency.md`. diff --git a/crates/nostream-relay/src/handlers/auth.rs b/crates/nostream-relay/src/handlers/auth.rs index 193e4e17..196867e5 100644 --- a/crates/nostream-relay/src/handlers/auth.rs +++ b/crates/nostream-relay/src/handlers/auth.rs @@ -10,7 +10,11 @@ use crate::handlers::HandlerContext; /// The seven outcomes are `nip42::authenticate`'s failures in the capture's order, then the /// replay check, then `["OK",,true,""]`. The challenge is never rotated by AUTH /// (`web-socket-adapter.ts:186-196`). -pub async fn handle(message: AuthMessage, context: &HandlerContext) { +/// +/// It contains no `.await`, so it runs whole as the message's prefix, on the connection loop: +/// a restricted `REQ` behind this `AUTH` in the stream is judged against the session it +/// records. +pub fn handle(message: AuthMessage, context: &HandlerContext) { let event = message.event; let now = context.now_secs(); let challenge = context.connection.challenge(); diff --git a/crates/nostream-relay/src/handlers/mod.rs b/crates/nostream-relay/src/handlers/mod.rs index 5b6a3cb4..e2e3a78a 100644 --- a/crates/nostream-relay/src/handlers/mod.rs +++ b/crates/nostream-relay/src/handlers/mod.rs @@ -5,6 +5,24 @@ //! write-point sink, the read-through settings, the clock, the connection state and the //! repository seams. +//! [`begin`] runs a message's **ordering-sensitive prefix** on the caller's task — the +//! connection loop — before the loop reads the next frame, and returns the rest of the +//! handler as a future to spawn. The rule for which handler takes a prefix: it is the +//! per-connection state a later frame's handler reads and expects to be there. `subscribe` +//! registers its subscription and its read, `unsubscribe` clears them, and `auth` records the +//! session a restricted `REQ` is judged against, so those three are prefixes. For `auth` this is +//! stronger than the capture, whose handler awaits two checks before recording the session, so a +//! `REQ` pipelined behind an `AUTH` may be refused there and is served here; ADR 0005 records the +//! deviation. `event` and `count` keep their whole body as one future: neither establishes +//! per-connection state that a later frame looks up, and the one shared table they touch is +//! relay-wide — the event limiter, keyed on the event's author and window +//! (`RateLimits::event_backoff`). So two pipelined `EVENT`s from one author record their hits in +//! whichever order their tasks are polled, and which of the two is refused is not fixed by +//! arrival order. The capture races the same way: +//! the socket fires each `'message'` listener without awaiting it (`web-socket-adapter.ts:79`, +//! `:198`), and the handler is awaited inside that listener (`:229`), so two pipelined `EVENT`s +//! run as two interleaved chains there too. + pub mod auth; pub mod count; pub mod event; @@ -162,7 +180,7 @@ impl ReadGuard { } } -/// `getRestrictedReadKinds` (`src/utils/nip42.ts:15-23`): the configured kinds when +/// `getRestrictedReadKinds` (`src/utils/nip42.ts:17-24`): the configured kinds when /// restricted reads are enabled, the NIP-42 defaults when no list is given, none when off. pub fn restricted_read_kinds(settings: Option<&Nip42Settings>) -> Vec { let Some(settings) = settings else { @@ -186,20 +204,51 @@ pub fn restricted_read_kinds(settings: Option<&Nip42Settings>) -> Vec + Send>>; -/// Runs one decoded client message as the capture's `messageHandlerFactory` switch does. -pub fn dispatch( +/// Runs one decoded client message's ordering-sensitive prefix on the caller's task, as the +/// capture's `messageHandlerFactory` switch does, and returns the rest of its handler. +/// +/// Nothing here can enforce when it is called: the guarantee holds because the connection loop +/// is the only caller in `src`, and it calls this before reading the next frame. For a message +/// with a prefix, every statement the prefix executes has run by the time this returns. `None` +/// means the prefix handled the message and nothing is left to run — it writes whatever frames +/// its handler writes, and `CLOSE` writes none; `Some(future)` is the rest of the handler, which +/// is spawned into the socket's handler set and runs unordered with its neighbours. +/// +/// The postcondition is therefore about the prefix, not the handler: the `Event` and `Count` +/// arms box the whole handler, so none of their body is ordered. +pub fn begin( message: nostream_core::messages::ClientMessage, - context: Arc, -) -> HandlerFuture { + context: &Arc, +) -> Option { use nostream_core::messages::ClientMessage; - Box::pin(async move { - match message { - ClientMessage::Event(event) => event::handle(event, &context).await, - ClientMessage::Subscribe(message) => subscribe::handle(message, &context).await, - ClientMessage::Count(message) => count::handle(message, &context).await, - ClientMessage::Unsubscribe(message) => unsubscribe::handle(message, &context), - ClientMessage::Auth(message) => auth::handle(message, &context).await, + match message { + ClientMessage::Event(event) => { + let context = Arc::clone(context); + + // Whole body by rule: `event` needs no prefix (see the module doc). + Some(Box::pin( + async move { event::handle(event, &context).await }, + )) + } + ClientMessage::Count(message) => { + let context = Arc::clone(context); + + // Whole body by rule: `count` needs no prefix (see the module doc). + Some(Box::pin( + async move { count::handle(message, &context).await }, + )) } - }) + ClientMessage::Subscribe(message) => subscribe::begin(message, context), + ClientMessage::Unsubscribe(message) => { + unsubscribe::handle(message, context); + + None + } + ClientMessage::Auth(message) => { + auth::handle(message, context); + + None + } + } } diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index 53e1a10e..b0ca6cf0 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -9,12 +9,15 @@ //! //! 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). +//! `CLOSE` aborts the stored read and the subscription streams nothing. Everything through +//! that registration is the handler's ordering-sensitive prefix and runs on the connection +//! loop, so a `CLOSE` behind this `REQ` in the stream cannot be handled ahead of it — the +//! defect cameri/nostream#958 reported, closed by the decision recorded in ADR 0005. The +//! invite `EVENT` is emitted even then: 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 std::sync::Arc; use futures_util::StreamExt; use nostream_core::expiration; @@ -25,13 +28,33 @@ use nostream_core::{Filter, Kind}; use crate::connection::InFlightRead; use crate::error::RelayError; -use crate::handlers::HandlerContext; +use crate::handlers::{HandlerContext, HandlerFuture}; /// `EventKinds.NIP43_INVITE_REQUEST`. const INVITE_REQUEST_KIND: Kind = Kind(28_935); /// The handler's entry point (`handleMessage`, `:51-82`). -pub async fn handle(message: SubscribeMessage, context: &HandlerContext) { +/// +/// Its ordering-sensitive prefix — everything through `begin_read` — runs on the caller's +/// task, so the connection loop has the subscription and its read registered before it reads +/// the next frame. The continuation starts at the invite mint and is spawned into the +/// socket's handler set, where it interleaves with the frames that follow. +pub fn begin(message: SubscribeMessage, context: &Arc) -> Option { + let (read, filters) = register(message, context)?; + let context = Arc::clone(context); + + Some(Box::pin(async move { + maybe_issue_invite_code(read.subscription_id(), &filters, &context).await; + + fetch_and_send(read, &filters, &context).await; + })) +} + +/// The prefix (`handleMessage`, `:51-69`): what the capture runs before its first `await`. +fn register( + message: SubscribeMessage, + context: &HandlerContext, +) -> Option<(InFlightRead, Vec)> { let subscription_id = message.subscription_id; let search_enabled = context .settings @@ -43,7 +66,7 @@ pub async fn handle(message: SubscribeMessage, context: &HandlerContext) { if let Some(reason) = can_subscribe(&subscription_id, &filters, context) { context.send_notice(&format!("Subscription rejected: {reason}")); - return; + return None; } let restricted = context.restricted_read_kinds(); @@ -55,7 +78,7 @@ pub async fn handle(message: SubscribeMessage, context: &HandlerContext) { "auth-required: authentication is required to request these event kinds", None, ); - return; + return None; } context @@ -68,9 +91,7 @@ pub async fn handle(message: SubscribeMessage, context: &HandlerContext) { // 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(read, &filters, context).await; + Some((read, filters)) } /// NIP-43: kind 28935 is never published by a client and never stored. A REQ for it asks the diff --git a/crates/nostream-relay/src/handlers/unsubscribe.rs b/crates/nostream-relay/src/handlers/unsubscribe.rs index 51b422a8..02746b5d 100644 --- a/crates/nostream-relay/src/handlers/unsubscribe.rs +++ b/crates/nostream-relay/src/handlers/unsubscribe.rs @@ -2,6 +2,9 @@ //! //! A `CLOSE` removes the subscription, aborts its in-flight read, and emits **no frame** //! (`:9-11`; measured: `CLOSE s1` → no frame). +//! +//! It runs whole as the message's prefix, on the connection loop, so a `CLOSE` behind its own +//! `REQ` in the stream always finds the subscription and the read it aborts. use nostream_core::messages::UnsubscribeMessage; diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index 003bd689..70fb1613 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -4,9 +4,11 @@ //! //! Each connection task owns its socket and the state machine R26 lists: `alive`, the //! subscription table, the NIP-42 session, the in-flight handler set and the sink. Outbound -//! frames travel through [`FrameSink`]; inbound text is rate-limited, decoded, and handed to -//! the handler `messageHandlerFactory` would pick. Accepted events fan out to this process's -//! connections through one broadcast channel (cross-worker fan-out stays a stub). +//! frames travel through [`FrameSink`]; inbound text is rate-limited, decoded, and given to +//! the handler `messageHandlerFactory` would pick — the handler's ordering-sensitive prefix +//! runs here, on the loop, before the next frame is read, and the rest of it is spawned into +//! the per-socket handler set. Accepted events fan out to this process's connections through +//! one broadcast channel (cross-worker fan-out stays a stub). use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -478,6 +480,9 @@ impl Relay { /// One inbound text frame: the message limiter first, then the decoder, then the handler /// (`web-socket-adapter.ts:198-240`). + /// + /// The handler's prefix runs here, before the loop reads the next frame; its continuation + /// joins the socket's handler set. fn on_client_message( &self, text: &str, @@ -504,8 +509,9 @@ impl Relay { match messages::decode(text) { Ok(message) => { - let context = Arc::clone(context); - handlers.spawn(handlers::dispatch(message, context)); + if let Some(handler) = handlers::begin(message, context) { + handlers.spawn(handler); + } } Err(error) => context.send_notice(&error.notice()), } diff --git a/crates/nostream-relay/tests/connection.rs b/crates/nostream-relay/tests/connection.rs index 43f01c1b..5b6ee276 100644 --- a/crates/nostream-relay/tests/connection.rs +++ b/crates/nostream-relay/tests/connection.rs @@ -18,7 +18,7 @@ use std::sync::Arc; use common::{CLIENT_SECRET, NOW_MS, NOW_SECS, Server, fast_config, relay_keys, signed_event}; use nostream_core::clock::TestClock; use nostream_core::filter::Filter; -use nostream_core::messages::SubscribeMessage; +use nostream_core::messages::{AuthMessage, ClientMessage, SubscribeMessage, UnsubscribeMessage}; use nostream_core::{EventId, Pubkey}; use nostream_relay::connection::{Connection, Subscription}; use nostream_relay::sink::FrameSink; @@ -427,7 +427,7 @@ async fn closing_the_socket_aborts_an_in_flight_read() { let context = Arc::clone(&context); async move { - nostream_relay::handlers::subscribe::handle( + let continuation = nostream_relay::handlers::subscribe::begin( SubscribeMessage { subscription_id: "s1".to_owned(), filters: vec![Filter { @@ -437,7 +437,9 @@ async fn closing_the_socket_aborts_an_in_flight_read() { }, &context, ) - .await; + .expect("the REQ registers its subscription"); + + continuation.await; } }); @@ -510,7 +512,7 @@ async fn restricted_read_with(authenticate_mid_read: bool) -> Vec { let context = Arc::clone(&context); async move { - nostream_relay::handlers::subscribe::handle( + let continuation = nostream_relay::handlers::subscribe::begin( SubscribeMessage { subscription_id: "restricted".to_owned(), // A filter that names no kind is accepted while unauthenticated; the @@ -520,7 +522,9 @@ async fn restricted_read_with(authenticate_mid_read: bool) -> Vec { }, &context, ) - .await; + .expect("the REQ registers its subscription"); + + continuation.await; } }); @@ -604,3 +608,81 @@ async fn a_restricted_read_is_judged_against_the_authenticated_set_at_each_event assert!(authenticated[1].contains("encrypted")); assert_eq!(authenticated[2], r#"["EOSE","restricted"]"#); } + +// --------------------------------------------------------------------------------------- +// Handler dispatch order +// --------------------------------------------------------------------------------------- + +/// A message's prefix has run by the time `begin` returns, before its continuation is polled. +/// +/// No scheduler is involved: every assertion sits between two synchronous calls on this task, +/// so `Connection` either holds the registration or it does not. +#[tokio::test] +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 context = Arc::new(fixture.context( + Arc::new(TestClock::new(NOW_MS)), + relay_keys(), + Arc::clone(&connection), + sink, + )); + + let continuation = nostream_relay::handlers::begin( + ClientMessage::Subscribe(SubscribeMessage { + subscription_id: "s1".to_owned(), + filters: vec![Filter { + kinds: Some(vec![nostream_core::Kind(1)]), + ..Default::default() + }], + }), + &context, + ); + + // Nothing above was awaited and nothing was polled, yet the REQ has registered. + assert!(continuation.is_some(), "the REQ has a continuation"); + assert!(connection.read_in_flight("s1")); + assert!(connection.subscription("s1").is_some()); + + // Dropped unpolled: the registration is the whole of what this test needs from it. + drop(continuation); + + assert!( + nostream_relay::handlers::begin( + ClientMessage::Unsubscribe(UnsubscribeMessage { + subscription_id: "s1".to_owned(), + }), + &context, + ) + .is_none(), + "a CLOSE completes inside its prefix" + ); + assert!(!connection.read_in_flight("s1")); + assert!(connection.subscription("s1").is_none()); + + let event = signed_event( + &CLIENT_SECRET, + 22_242, + vec![ + vec!["relay", "wss://relay.example.com"], + vec!["challenge", "challenge"], + ], + "", + NOW_SECS, + ); + + assert!( + nostream_relay::handlers::begin(ClientMessage::Auth(AuthMessage { event }), &context) + .is_none(), + "an AUTH completes inside its prefix" + ); + + // The OK is already in the sink, so the AUTH's frame was written on this task. + let ok = frames.try_recv().expect("the AUTH's OK frame"); + + assert!( + ok.starts_with(r#"["OK","#) && ok.ends_with(r#"true,""]"#), + "{ok}" + ); +} diff --git a/crates/nostream-relay/tests/handlers.rs b/crates/nostream-relay/tests/handlers.rs index 5ace1fed..ffaec0c6 100644 --- a/crates/nostream-relay/tests/handlers.rs +++ b/crates/nostream-relay/tests/handlers.rs @@ -67,6 +67,30 @@ async fn send_event(server: &Server, event: &Event) -> String { client.expect_frame().await } +/// The text frames that arrive within `window`, read straight off the socket so a test can +/// bound a silence shorter than `Client::next_message`'s two-second deadline. +async fn frames_within(client: &mut common::Client, window: Duration) -> Vec { + use futures_util::StreamExt; + use tokio_tungstenite::tungstenite::Message; + + let deadline = tokio::time::Instant::now() + window; + let stream = client.stream(); + let mut frames = Vec::new(); + + while let Ok(Some(Ok(message))) = tokio::time::timeout_at(deadline, stream.next()).await { + if let Message::Text(text) = message { + frames.push(text.as_str().to_owned()); + } + } + + frames +} + +/// Whether a frame names the closed subscription: a live dispatch to it, or its `EOSE`. +fn names_s1(frame: &str) -> bool { + frame.starts_with(r#"["EVENT","s1""#) || frame == r#"["EOSE","s1"]"# +} + // --------------------------------------------------------------------------------------- // EVENT (R9) // --------------------------------------------------------------------------------------- @@ -1025,6 +1049,89 @@ async fn a_close_aborts_an_in_flight_read_without_a_frame() { assert_eq!(frames[1], r#"["EOSE","s1"]"#); } +/// A `CLOSE` must not be handled ahead of the `REQ` it closes. +/// +/// The `REQ` and the `CLOSE` leave in one flush, followed by a frame that cannot be decoded: +/// its `NOTICE` is written by the connection loop itself, so receiving it proves the loop has +/// read the two frames ahead of it. No wall-clock wait stands in for an observation — the +/// barrier orders the live event after both frames, and the live event's own `OK` (on the +/// client's deadline) is the only other anchor. The stored read stays parked at its gate until +/// the test releases it, which is what keeps the assertion about ordering: without the gate +/// the in-memory read finishes in one poll and its frames arrive even when the ordering is +/// right. `multi_thread, worker_threads = 1` is the flavour the defect was measured on (30/30); +/// on `current_thread` a spawned handler cannot be polled before the spawning task yields, so +/// the test would pass on a wrong tree. +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn a_close_that_overtakes_its_req_streams_nothing() { + use futures_util::SinkExt; + use tokio_tungstenite::tungstenite::Message; + + let fixture = Fixture::new(settings()); + let gate = fixture.events.gate_reads(); + 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 live = signed_event(&CLIENT_SECRET, 1, vec![], "live", NOW_SECS); + let accepted = format!(r#"["OK","{}",true,""]"#, live.id); + + // One flush, so the loop reads the three frames in the order they are written here. + let stream = client.stream(); + stream + .feed(Message::Text( + r#"["REQ","s1",{"kinds":[1]}]"#.to_owned().into(), + )) + .await + .expect("feed the REQ"); + stream + .feed(Message::Text(r#"["CLOSE","s1"]"#.to_owned().into())) + .await + .expect("feed the CLOSE"); + stream + .feed(Message::Text(r#"{"kind":"#.to_owned().into())) + .await + .expect("feed the undecodable frame"); + stream.flush().await.expect("flush the three frames"); + + // The loop's own answer to the third frame: the `REQ` and the `CLOSE` have been read. + let barrier = client.expect_frame().await; + assert!(barrier.starts_with(r#"["NOTICE","invalid: "#), "{barrier}"); + + // The live event's `OK` is the second anchor: nothing for `s1` may reach the client on + // the way to it, and the client's own deadline bounds the wait. + client.send(&event_frame(&live)).await; + + loop { + let Some(message) = client.next_message().await else { + panic!("the live event's OK never arrived"); + }; + let Message::Text(frame) = message else { + continue; + }; + + assert!( + !names_s1(frame.as_str()), + "a closed subscription streamed: {frame}" + ); + + if frame.as_str() == accepted { + break; + } + } + + // Releasing the read decides nothing new: the read either never opened or was aborted. + gate.add_permits(4); + + let frames = frames_within(&mut client, Duration::from_millis(300)).await; + + assert!( + frames.iter().all(|frame| !names_s1(frame)), + "a closed subscription streamed: {frames:?}" + ); +} + #[tokio::test] async fn a_new_filter_set_replaces_the_subscription() { let fixture = Fixture::new(settings()); @@ -1489,6 +1596,53 @@ async fn nip42_sessions_expire_at_the_ttl_boundary() { assert!(forever.is_authenticated(&pubkey, i64::MAX)); } +/// An `AUTH` must be handled before the restricted `REQ` that follows it in the stream. +/// +/// The existing NIP-42 tests read the `OK` before sending their `REQ`, which serialises the +/// pair; pipelined in one flush they exercise the loop's ordering. Two workers, the flavour +/// this symptom was measured on (10/10 refused before the ordering fix). +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_pipelined_auth_before_a_restricted_req_is_ordered() { + use futures_util::SinkExt; + use tokio_tungstenite::tungstenite::Message; + + let mut settings = settings(); + settings.nip42 = Some(nip42_settings()); + let server = start(Fixture::new(settings)).await; + let (mut client, challenge) = server.connect().await; + let auth = auth_event(&challenge, NOW_SECS); + + // One flush, so both frames are buffered before the loop reads either of them. + let stream = client.stream(); + stream + .feed(Message::Text( + format!( + r#"["AUTH",{}]"#, + serde_json::to_string(&auth).expect("auth json") + ) + .into(), + )) + .await + .expect("feed the AUTH"); + stream + .feed(Message::Text( + r#"["REQ","s1",{"kinds":[4]}]"#.to_owned().into(), + )) + .await + .expect("feed the REQ"); + stream.flush().await.expect("flush both frames"); + + let first = client.expect_frame().await; + let second = client.expect_frame().await; + + assert_eq!(first, format!(r#"["OK","{}",true,""]"#, auth.id)); + assert_eq!(second, r#"["EOSE","s1"]"#); + assert!( + !first.starts_with(r#"["CLOSED","s1""#) && !second.starts_with(r#"["CLOSED","s1""#), + "the REQ was refused: {first} {second}" + ); +} + // --------------------------------------------------------------------------------------- // Dispatch and the message limiter (R14) // --------------------------------------------------------------------------------------- diff --git a/docs/rewrite/decisions/0005-connection-concurrency.md b/docs/rewrite/decisions/0005-connection-concurrency.md index a4c9f090..4a4b5240 100644 --- a/docs/rewrite/decisions/0005-connection-concurrency.md +++ b/docs/rewrite/decisions/0005-connection-concurrency.md @@ -1,20 +1,22 @@ -# ADR 0005: Connection concurrency — read ownership, a bounded frame queue, and liveness under a blocked write (job 010) +# ADR 0005: Connection concurrency — read ownership, a bounded frame queue, liveness under a blocked write, and inbound dispatch order (jobs 010 and 012) -- **Status:** accepted (job 010, Phase 2) -- **Deciders:** the rewrite team, on the evidence in job 010's recon and spec +- **Status:** accepted (job 010 and job 012, Phase 2) +- **Deciders:** the rewrite team, on the evidence in job 010's recon and spec, and job 012's for §"Inbound dispatch order" - **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 task per socket owns the transport, and four 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. +queue grow without limit while the liveness probe could not run. And the handlers themselves +were spawned whole into that task's `JoinSet`, so two frames read in order could run in either +order (§"Inbound dispatch order"). 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. @@ -134,11 +136,11 @@ 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 +checked before the stored read opens, so a `CLOSE` ends the subscription with no stored +event and no `EOSE`. Since §"Inbound dispatch order" below, that `CLOSE` cannot be handled +ahead of its own `REQ` either: a handler's prefix runs on the connection loop before the next +frame is read, which is what closed 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 @@ -146,6 +148,67 @@ statement (`:53`). The TypeScript tree wins where the two trees disagree absent record, so the fix must not reach for the emit; the exposure it leaves is one stray `EVENT`, which is the capture's own. +## Inbound dispatch order — a prefix on the loop, not a serialized handler + +An inbound frame's handler was spawned whole into the per-socket `JoinSet`, so two frames +read in order ran in whichever order the scheduler picked: a `CLOSE` could be handled ahead +of its own `REQ` and find nothing to abort, and a restricted `REQ` could be refused before +the `AUTH` that preceded it had been handled. Both were measured through a real socket — +30/30 runs at one worker for the `REQ`/`CLOSE` pair, 10/10 at two for `AUTH`/`REQ` — and +neither was visible to the suite, because `#[tokio::test]` defaults to the one runtime flavour +in which a spawned handler cannot overtake the task that spawned it. + +The decision is the capture's own division of a message: its **ordering-sensitive prefix** +runs before the next frame is read, and its continuations stay concurrent. +`handlers::begin` runs each prefix on the connection loop and returns the continuation for +the `JoinSet`. The prefixes are `subscribe` (everything through `begin_read`, so the +subscription and its read exist before a following `CLOSE` is looked at), `unsubscribe` (the +whole handler) and `auth` (the whole handler, which has no `await`). + +`AUTH` is the one prefix that is deliberately **stronger** than the capture rather than parity +with it. The capture's `AuthMessageHandler.handleMessage` awaits `isEventIdValid` +(`src/handlers/auth-message-handler.ts:30`) and `isEventSignatureValid` (`:35`) before it +records the session in `addAuthenticatedPubkey` (`:65`), and the listener awaits the whole +handler, so a `REQ` pipelined behind an `AUTH` can be judged against a session that has not +been recorded yet and be refused as needing authentication. Rust verifies and records on the +loop (`handlers::auth::handle` is synchronous), so the session exists by the time the next +frame is read. The difference is one-sided — a pipelined `REQ` the capture may refuse, this +relay serves — and it is what a client reads NIP-42 as promising, so it is taken rather than +reproduced. The cost is real and named: the read path waits for two signature checks it could +have overlapped, once per `AUTH` frame rather than per connection. + +`EVENT` and `COUNT` keep their whole body as one spawned future: neither writes +per-connection state a later frame looks up, so nothing they do before their first `await` has +to be ordered. The one shared table they touch is relay-wide: the event limiter, keyed on the +event's author and window. Two pipelined `EVENT`s from one author therefore record their hits +in whichever order their tasks are polled, so which of the two is refused is not fixed by +arrival order — parity, because the socket fires +each `'message'` listener without awaiting it (`src/adapters/web-socket-adapter.ts:79`, `:198`) +and the handler is awaited inside that listener (`:229`), so its hits are unordered too. + +Two shapes were rejected. **Serializing whole handlers** — an ordered queue drained in the +loop — would run a `REQ`'s invite mint to completion before a `CLOSE` already in the buffer +is looked at, which the capture permits and `tests/handlers.rs` asserts, and it would put +every database write in front of the socket's whole read path. **An ordering token per +connection** — a per-id "closed at sequence N" consulted by `begin_read` — needs the second +structure in `InFlightReads` this record argues against in §"Read ownership", must be bounded +or evicted, needs a rule for what an "older" `REQ` means under a re-`REQ`, and orders +`AUTH` before a restricted `REQ` only with a second, unrelated rule. + +What this does not decide: it is a guarantee about prefixes, not about handler execution. +Continuations interleave as before, so a `CLOSE` during a gated invite mint still aborts the +read and the invite `EVENT` is still emitted. The cost accepted is that the prefixes run on the +loop task: `REQ`'s `normalise_subscription` and `can_subscribe`, bounded by +`transport::max_payload_size` and, where configured, `max_filters` / `max_filter_values`, and +`AUTH`'s whole `nip42::authenticate`. Measured for this branch on the 4-vCPU development box: +`AUTH` 34.3 µs per message (a 32.2–32.5 µs signature verify plus a 1.28 µs id hash), `REQ` +0.72–0.76 µs (0.11 µs normalisation plus 0.61–0.65 µs of subscription-table work, measured +separately), `CLOSE` 0.053 µs. The capture does the same work on its socket's thread, but it +pays in awaited chunks that let other sockets interleave, where a prefix does not yield at all; +what the port buys is blast radius — one socket's read path rather than 1.57–1.85 ms/op of the +capture's only thread. The cost inside that radius is real: the loop is occupied for those +34 µs, and a client that pipelines `AUTH` is capped at roughly 29 000 per second. + ## 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