fix(relay): emit BUZZ_SYNC_REQUIRED gap signal on fan-out backpressure drops

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

@@ -88,7 +88,7 @@ where
             .expect("fan-out frame cache covers every recipient subscription id");
         if !state
             .conn_manager
-            .send_to_text_bytes(conn_id, Arc::clone(frame))
+            .send_fanout_frame(conn_id, Arc::clone(frame))
         {
             drop_count += 1;
         }
@@ -2121,6 +2121,88 @@ mod tests {
             conn_id
         }
 
+        /// Register a connection and hand back its channel receivers so tests
+        /// can observe what fan-out actually delivered.
+        fn register_conn_with_buffers(
+            state: &AppState,
+            data_buffer: usize,
+            ctrl_buffer: usize,
+        ) -> (
+            Uuid,
+            mpsc::Receiver<axum::extract::ws::Message>,
+            mpsc::Receiver<axum::extract::ws::Message>,
+            CancellationToken,
+            Arc<AtomicU8>,
+        ) {
+            let conn_id = Uuid::new_v4();
+            let (tx, rx) = mpsc::channel(data_buffer);
+            let (ctrl_tx, ctrl_rx) = mpsc::channel(ctrl_buffer);
+            let cancel = CancellationToken::new();
+            let backpressure_count = Arc::new(AtomicU8::new(0));
+            state.conn_manager.register(
+                conn_id,
+                tx,
+                ctrl_tx,
+                None,
+                cancel.clone(),
+                buzz_core::tenant::CommunityId::from_uuid(Uuid::nil()),
+                Arc::clone(&backpressure_count),
+                Arc::new(Mutex::new(HashMap::new())),
+                3,
+            );
+            (conn_id, rx, ctrl_rx, cancel, backpressure_count)
+        }
+
+        #[tokio::test]
+        async fn send_fanout_frames_drop_signals_sync_required_on_ctrl() {
+            // End-to-end through the real fan-out path: `send_fanout_frames`
+            // → `ConnectionManager::send_fanout_frame`. With a 1-slot data
+            // buffer the second recipient frame for this connection drops.
+            let state = test_state().await;
+            let (conn_id, mut rx, mut ctrl_rx, cancel, backpressure_count) =
+                register_conn_with_buffers(&state, 1, 8);
+
+            let event_json = r#"{"id":"abc"}"#;
+            let frames = crate::handlers::event::fanout_frame_cache(["sub-a", "sub-b"], event_json);
+
+            let drop_count = crate::handlers::event::send_fanout_frames(
+                &state,
+                [(conn_id, "sub-a"), (conn_id, "sub-b")],
+                &frames,
+            );
+            assert_eq!(drop_count, 1, "second frame drops on the full buffer");
+
+            // Gap signal queued on the priority control channel, exact bytes.
+            match ctrl_rx.try_recv().expect("gap signal on ctrl channel") {
+                axum::extract::ws::Message::Text(text) => assert_eq!(
+                    text.to_string(),
+                    r#"["BUZZ_SYNC_REQUIRED","backpressure"]"#,
+                    "exact machine frame contract"
+                ),
+                other => panic!("expected BUZZ_SYNC_REQUIRED text frame, got {other:?}"),
+            }
+            assert!(ctrl_rx.try_recv().is_err(), "exactly one gap signal");
+
+            // Data channel holds exactly the one delivered EVENT frame.
+            match rx.try_recv().expect("delivered EVENT frame") {
+                axum::extract::ws::Message::Text(text) => assert_eq!(
+                    text.to_string(),
+                    format!(r#"["EVENT","sub-a",{event_json}]"#)
+                ),
+                other => panic!("expected EVENT frame, got {other:?}"),
+            }
+            assert!(rx.try_recv().is_err(), "dropped frame must not reach data");
+
+            assert_eq!(
+                backpressure_count.load(std::sync::atomic::Ordering::Relaxed),
+                1
+            );
+            assert!(
+                !cancel.is_cancelled(),
+                "below the grace limit the socket stays live once the signal is queued"
+            );
+        }
+
         fn channel_event(channel_id: Option<Uuid>) -> StoredEvent {
             let event = EventBuilder::new(Kind::Custom(9), "{}")
                 .sign_with_keys(&Keys::generate())

crates/buzz-relay/src/protocol.rs

@@ -1,4 +1,29 @@
 //! NIP-01 client/relay message parsing and formatting.
+//!
+//! # Buzz extension frames
+//!
+//! Alongside the NIP-01 relay→client messages ([`RelayMessage`]), the relay
+//! emits one Buzz-specific extension frame:
+//!
+//! ## `BUZZ_SYNC_REQUIRED`
+//!
+//! ```text
+//! ["BUZZ_SYNC_REQUIRED","<reason>"]
+//! ```
+//!
+//! Machine-readable signal that the client's view has a gap and it should
+//! resynchronize (replay from its watermark). Emitted today with reason
+//! `backpressure` when an EVENT fan-out frame for this connection was dropped
+//! because the connection's outbound data channel was full. Delivery rules:
+//!
+//! - Sent on the connection's priority control channel, never on the data
+//!   channel it signals about, and never as a human-readable `NOTICE`.
+//! - Emitted only for live connections whose data channel is full. Closed or
+//!   already-gone connections get no signal — reconnect replay is their
+//!   fail-safe.
+//! - Clients that do not recognize the frame MUST ignore it (unknown
+//!   relay→client array heads are non-fatal per NIP-01 client convention);
+//!   the reconnect-replay machinery remains the backstop either way.
 
 use nostr::{Event, Filter};
 use serde_json::Value;
@@ -214,6 +239,19 @@ impl RelayMessage {
     pub fn count(sub_id: &str, count: u64) -> String {
         serde_json::json!(["COUNT", sub_id, {"count": count}]).to_string()
     }
+
+    /// Format a `BUZZ_SYNC_REQUIRED` extension frame (see module docs).
+    ///
+    /// Machine-readable gap signal delivered on the connection's priority
+    /// control channel. `"backpressure"` — an EVENT fan-out frame was dropped
+    /// because the connection's data channel was full — is the only reason
+    /// the wire contract defines. The constructor is deliberately monomorphic
+    /// so the relay cannot emit a frame the contract does not define; if a
+    /// new reason ever appears, grow this into an enum and extend the
+    /// contract deliberately.
+    pub fn sync_required() -> String {
+        serde_json::json!(["BUZZ_SYNC_REQUIRED", "backpressure"]).to_string()
+    }
 }
 
 #[cfg(test)]
@@ -448,6 +486,18 @@ mod tests {
                     assert_eq!(v[2], "auth-required: not authenticated");
                 }),
             ),
+            (
+                "sync_required",
+                Box::new(|| {
+                    let msg = RelayMessage::sync_required();
+                    // Exact wire bytes: clients match on this precise shape,
+                    // and the constructor cannot produce any other reason.
+                    assert_eq!(msg, r#"["BUZZ_SYNC_REQUIRED","backpressure"]"#);
+                    let v: Value = serde_json::from_str(&msg).unwrap();
+                    assert_eq!(v[0], "BUZZ_SYNC_REQUIRED");
+                    assert_eq!(v[1], "backpressure");
+                }),
+            ),
         ];
 
         for (name, check) in cases {