fix(relay): run a handler's ordered prefix on the connection loop

.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`.

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",<id>,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();

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<KindSelector> {
     let Some(settings) = settings else {
@@ -186,20 +204,51 @@ pub fn restricted_read_kinds(settings: Option<&Nip42Settings>) -> Vec<KindSelect
 /// One handler's future, boxed so a connection task can spawn it without naming its type.
 pub type HandlerFuture = Pin<Box<dyn Future<Output = ()> + 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<HandlerContext>,
-) -> HandlerFuture {
+    context: &Arc<HandlerContext>,
+) -> Option<HandlerFuture> {
     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
+        }
+    }
 }

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<HandlerContext>) -> Option<HandlerFuture> {
+    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<Filter>)> {
     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

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;
 

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()),
         }

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<String> {
         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<String> {
                 },
                 &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}"
+    );
+}