fix(relay): read ownership, cancellation before the mint, and a bounded outbound queue (#950, #951, #952)

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

crates/nostream-relay/src/connection.rs

@@ -38,7 +38,47 @@ pub struct Connection {
     pub remote_address: String,
     subscriptions: Mutex<HashMap<String, Subscription>>,
     session: Mutex<Nip42Session>,
-    in_flight: Mutex<HashMap<String, watch::Sender<bool>>>,
+    in_flight: Mutex<InFlightReads>,
+}
+
+/// 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<String, (u64, watch::Sender<bool>)>,
+}
+
+/// 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<bool>,
+}
+
+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<bool> {
+        &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<bool> {
-        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);
         }
     }

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