fix(nip-ar): send no EOSE after a resync recorded during its wait

The initial phase checked for a pending resync before EOSE waited for queue room. A held-list overflow during that wait still let EOSE out. EOSE is now queued under the inbox lock, after the owner check, so a resync recorded before it enters the queue ends the sync with CLOSED and no EOSE (rule 1).

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Signed-off-by: Logan Johnson <loganj@squareup.com>

crates/buzz-relay/src/handlers/artifact_sync.rs

@@ -114,13 +114,6 @@ impl SyncHandle {
         self.notify.notify_one();
     }
 
-    fn resync_reason(&self) -> Option<&'static str> {
-        self.pending
-            .lock()
-            .unwrap_or_else(PoisonError::into_inner)
-            .resync
-    }
-
     /// Record a fanned-out event that falls in this sync's channels.
     fn signal(&self, stored: &StoredEvent) {
         let kind = stored.event.kind.as_u16();
@@ -401,11 +394,8 @@ pub async fn handle_sync(
     // the handler slot past the budget or get EOSE after retirement.
     let initial = async {
         send_initial_set(&sub_id, owner, &filter, &reader, &conn, &state).await?;
-        // A resync requested during the read also ends the sync before EOSE.
-        if let Some(reason) = handle.resync_reason() {
-            return Err(Stop::Close(reason));
-        }
-        send_owned(&sub_id, owner, RelayMessage::eose(&sub_id), &conn).await
+        // A resync requested before EOSE is queued ends the sync without it.
+        send_eose(&sub_id, owner, &handle, &conn).await
     };
     let outcome = tokio::select! {
         biased;
@@ -574,6 +564,36 @@ async fn send_owned(
     owner: u64,
     frame: String,
     conn: &ConnectionState,
+) -> Result<(), Stop> {
+    send_reserved(sub_id, owner, frame, None, conn).await
+}
+
+/// Queue `EOSE` like `send_owned`, unless the sync must resync. The pending
+/// resync is checked under the inbox lock while the frame is queued, so a
+/// resync recorded at any time before `EOSE` enters the queue ends the sync
+/// with `CLOSED` and no `EOSE` (rule 1); a later one closes after it.
+async fn send_eose(
+    sub_id: &str,
+    owner: u64,
+    handle: &SyncHandle,
+    conn: &ConnectionState,
+) -> Result<(), Stop> {
+    send_reserved(
+        sub_id,
+        owner,
+        RelayMessage::eose(sub_id),
+        Some(handle),
+        conn,
+    )
+    .await
+}
+
+async fn send_reserved(
+    sub_id: &str,
+    owner: u64,
+    frame: String,
+    unless_resync: Option<&SyncHandle>,
+    conn: &ConnectionState,
 ) -> Result<(), Stop> {
     let slot = conn
         .send_tx
@@ -584,6 +604,15 @@ async fn send_owned(
     if subs.get(sub_id) != Some(&owner) {
         return Err(Stop::Retired);
     }
+    let pending = unless_resync.map(|handle| {
+        handle
+            .pending
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+    });
+    if let Some(reason) = pending.as_ref().and_then(|pending| pending.resync) {
+        return Err(Stop::Close(reason));
+    }
     slot.send(WsMessage::Text(frame.into()));
     Ok(())
 }
@@ -874,16 +903,13 @@ mod tests {
                 .try_send(WsMessage::Text("fill".into()))
                 .unwrap();
         }
-        let waiting = tokio::spawn({
-            let conn = Arc::clone(&conn);
-            async move { send_owned("s", 1, "eose".into(), &conn).await }
-        });
-        tokio::task::yield_now().await;
-        assert!(!waiting.is_finished());
+        let mut waiting = std::pin::pin!(send_owned("s", 1, "eose".into(), &conn));
+        // Polled once, the frame is now waiting in `reserve`.
+        assert!(futures_util::poll!(&mut waiting).is_pending());
         // A replacing REQ takes the ID while the frame waits.
         conn.subscriptions.lock().await.insert("s".into(), 2);
         rx.recv().await.unwrap();
-        assert!(matches!(waiting.await.unwrap(), Err(Stop::Retired)));
+        assert!(matches!(waiting.await, Err(Stop::Retired)));
         let mut rest = 0;
         while let Ok(WsMessage::Text(text)) = rx.try_recv() {
             assert_eq!(text.as_str(), "fill");
@@ -892,6 +918,43 @@ mod tests {
         assert_eq!(rest, capacity - 1);
     }
 
+    /// A resync recorded while `EOSE` waits for queue room ends the sync
+    /// without `EOSE` (rule 1). Mutation: check the pending resync before
+    /// `reserve` instead of while queueing → `EOSE` is queued → RED.
+    #[tokio::test]
+    async fn resync_during_the_eose_wait_sends_no_eose() {
+        let (conn, mut rx) = conn_with_sub(1);
+        let handle = SyncHandle::new();
+        let capacity = conn.send_tx.capacity();
+        for _ in 0..capacity {
+            conn.send_tx
+                .try_send(WsMessage::Text("fill".into()))
+                .unwrap();
+        }
+        let mut waiting = std::pin::pin!(send_eose("s", 1, &handle, &conn));
+        assert!(futures_util::poll!(&mut waiting).is_pending());
+        // The held list overflows while EOSE waits.
+        for _ in 0..=MAX_SYNC_EVENTS {
+            handle.signal(&stored(45010, Uuid::new_v4()));
+        }
+        rx.recv().await.unwrap();
+        assert!(matches!(
+            waiting.await,
+            Err(Stop::Close("resync: too many changes"))
+        ));
+        while let Ok(WsMessage::Text(text)) = rx.try_recv() {
+            assert_eq!(text.as_str(), "fill");
+        }
+
+        // Without a pending resync, EOSE is queued.
+        let clean = SyncHandle::new();
+        assert!(send_eose("s", 1, &clean, &conn).await.is_ok());
+        let Ok(WsMessage::Text(text)) = rx.try_recv() else {
+            panic!("no EOSE");
+        };
+        assert!(text.as_str().starts_with(r#"["EOSE""#), "{text}");
+    }
+
     #[test]
     fn resync_all_reaches_every_sync() {
         let syncs = ArtifactSyncs::default();