fix(relay): broadcast connection state only when it actually changes

docs/OPTIMIZATION_PLAN.md

@@ -250,14 +250,24 @@ Each PR stands alone; none requires Phase 3's redesign.
 - **Verify:** Rust test: ingest terminal status ⇒ order leaves the book; long-session memory
   stays flat.
 
-### PR 2.5 — Fix the connection-state resubscribe storm `fix(relay)`
-- **Evidence:** `relay_pool.rs:211-214` broadcasts on **any** relay status change even when the
-  derived state is unchanged (`Online → Online`); each event drives a 10 s `fetch_events`, an
-  outbox flush and full resubscribes (`rust/src/api/nostr.rs:51-69`). One flapping relay at
-  the 2 s poll interval (`relay_pool.rs:22`) reproduces this indefinitely.
-- **Fix:** only send when the derived `ConnectionState` actually changed; debounce the
-  Online handler.
-- **Verify:** Rust test with a mock flapping relay: exactly one resubscribe cycle.
+### PR 2.5 — Fix the connection-state resubscribe storm `fix(relay)` — #364
+- **Evidence:** `relay_pool.rs` broadcast on **any** relay status change even when the
+  derived state was unchanged (`Online → Online`); each event drives a 10 s `fetch_events`, an
+  outbox flush and full resubscribes (`rust/src/api/nostr.rs`, the `Online` handler). The
+  trigger is a relay that **connects and drops** while another stays up — the cadence is the
+  SDK's reconnect backoff (~8/min measured), not the 2 s poll. A relay that is simply
+  unreachable settles into `Disconnected` and produces no storm (review of #364).
+- **Fix (done in #364):** every publisher (`new`, add/remove, status monitor) goes through one
+  shared gate that sends only when the derived `ConnectionState` actually changed. The gate
+  must be shared: a monitor-local one dropped a genuine `Online` after a direct `Offline`
+  from `remove_relay`, leaving subscribers believing the pool was down (reproduced in review).
+- **Remaining gap (not done):** debouncing the `Online` handler. With a single relay, or every
+  relay flapping in lockstep, the derived state genuinely oscillates and each real `Online`
+  still re-runs the whole sequence. Related pre-existing gap surfaced by the fix: the outbox
+  has retry backoff fields but nothing schedules a retry, and `fetch_and_set_node_capabilities`
+  has no retry either — the storm was the only thing re-driving both.
+- **Verify:** Rust tests in `relay_pool.rs`: unchanged state not rebroadcast, real transitions
+  pass in both directions, direct and monitor publishers share one gate.
 
 ### PR 2.6 — Close relay-side subscriptions on task exit `fix(relay)`
 - **Evidence:** `subscribe_daemon_messages` (`orders.rs:1203`) and `subscribe_single_order`

rust/src/nostr/relay_pool.rs

@@ -13,7 +13,7 @@ use nostr_sdk::prelude::*;
 // conflicting with our internal `RelayStatus` from `crate::api::types`.
 use nostr_sdk::prelude::RelayStatus as SdkRelayStatus;
 use std::collections::HashSet;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::sync::{broadcast, RwLock};
 
@@ -31,6 +31,10 @@ pub struct RelayPool {
     /// without this the removed relay would come straight back.
     blacklist: RwLock<HashSet<String>>,
     conn_tx: broadcast::Sender<ConnectionState>,
+    /// The last state actually sent on `conn_tx`, shared by every path that
+    /// publishes (`new`, add/remove and the status monitor) so a subscriber
+    /// only ever sees real transitions — see `broadcast_if_changed`.
+    last_broadcast: Arc<Mutex<Option<ConnectionState>>>,
     relay_tx: broadcast::Sender<RelayInfo>,
 }
 
@@ -49,6 +53,7 @@ impl RelayPool {
             relays: Arc::new(RwLock::new(Vec::new())),
             blacklist: RwLock::new(HashSet::new()),
             conn_tx,
+            last_broadcast: Arc::new(Mutex::new(None)),
             relay_tx,
         });
 
@@ -233,8 +238,8 @@ impl RelayPool {
     // ── Internal helpers ──────────────────────────────────────────────────────
 
     async fn broadcast_connection_state(&self) {
-        let state = derive_connection_state(&self.relays.read().await);
-        let _ = self.conn_tx.send(state);
+        let relays = self.relays.read().await;
+        broadcast_if_changed(&self.last_broadcast, &self.conn_tx, &relays);
     }
 
     /// Spawn a background task that polls each relay's SDK status every
@@ -249,6 +254,7 @@ impl RelayPool {
         let client = self.client.clone();
         let relays = self.relays.clone();
         let conn_tx = self.conn_tx.clone();
+        let last_broadcast = self.last_broadcast.clone();
         let relay_tx = self.relay_tx.clone();
 
         crate::rt::spawn(async move {
@@ -299,8 +305,8 @@ impl RelayPool {
                 }
 
                 if any_changed {
-                    let state = derive_connection_state(&relays.read().await);
-                    let _ = conn_tx.send(state);
+                    let relays_r = relays.read().await;
+                    broadcast_if_changed(&last_broadcast, &conn_tx, &relays_r);
                 }
             }
         });
@@ -309,6 +315,54 @@ impl RelayPool {
 
 // ── Pure helpers ──────────────────────────────────────────────────────────────
 
+/// Derive the state from `relays` and send it on `tx` only if it differs
+/// from what was last sent.
+///
+/// Every publisher must go through here. The gate has to be shared: if only
+/// the monitor deduplicated, an add/remove sending directly would leave the
+/// monitor's view stale, and its next genuine transition back (e.g. `Offline`
+/// after a removal, then the relay reconnecting to `Online`) would be dropped
+/// as a duplicate — leaving subscribers believing the pool is down for the
+/// rest of the session.
+///
+/// Takes the relay list rather than a derived state so the caller derives and
+/// sends while still holding the `relays` read guard: no writer can slip a
+/// newer observation in between, so a newer state is never sent before an
+/// older snapshot of it. The gate lock is never held across an await.
+fn broadcast_if_changed(
+    last: &Mutex<Option<ConnectionState>>,
+    tx: &broadcast::Sender<ConnectionState>,
+    relays: &[RelayInfo],
+) {
+    let state = derive_connection_state(relays);
+    let next = next_broadcast(&mut last.lock().unwrap_or_else(|e| e.into_inner()), state);
+    if let Some(state) = next {
+        let _ = tx.send(state);
+    }
+}
+
+/// The state to broadcast, or `None` when it has not actually changed.
+///
+/// The monitor broadcasts whenever any relay's status moved, and a relay that
+/// connects and drops (the SDK's reconnect backoff makes that a few times a
+/// minute) ticks `any_changed` on each move while the derived state stays
+/// `Online` as long as another relay is up. A relay that is simply
+/// unreachable settles into `Disconnected` and is harmless. Every `Online`
+/// reaching the subscriber in `api/nostr.rs` costs a capability fetch, an
+/// outbox flush and a full resubscribe of orders and chats — so rebroadcasting
+/// an unchanged state turns one flapping relay into a permanent background
+/// storm.
+fn next_broadcast(
+    last: &mut Option<ConnectionState>,
+    current: ConnectionState,
+) -> Option<ConnectionState> {
+    if last.as_ref() == Some(&current) {
+        return None;
+    }
+    *last = Some(current.clone());
+    Some(current)
+}
+
 /// Which of `announced` should be added: not configured yet, not
 /// blacklisted, first occurrence only. Order is preserved so the user sees
 /// them in the order the node listed them.
@@ -365,6 +419,139 @@ use crate::rt::unix_now;
 mod tests {
     use super::*;
 
+    /// One relay connecting and dropping while another stays connected
+    /// changes a relay's status without changing the derived state.
+    /// Broadcasting that re-ran a capability fetch, an outbox flush and a
+    /// full resubscribe of orders and chats on every reconnect, indefinitely.
+    #[test]
+    fn an_unchanged_state_is_not_rebroadcast() {
+        let mut last = None;
+
+        assert_eq!(
+            next_broadcast(&mut last, ConnectionState::Online),
+            Some(ConnectionState::Online),
+            "the first observation is always a transition"
+        );
+        assert_eq!(
+            next_broadcast(&mut last, ConnectionState::Online),
+            None,
+            "a flapping relay that leaves the pool online must stay silent"
+        );
+        assert_eq!(next_broadcast(&mut last, ConnectionState::Online), None);
+    }
+
+    /// Real transitions must still get through, in both directions.
+    #[test]
+    fn a_real_transition_is_broadcast() {
+        let mut last = Some(ConnectionState::Online);
+
+        assert_eq!(
+            next_broadcast(&mut last, ConnectionState::Offline),
+            Some(ConnectionState::Offline)
+        );
+        assert_eq!(
+            next_broadcast(&mut last, ConnectionState::Reconnecting),
+            Some(ConnectionState::Reconnecting)
+        );
+        assert_eq!(
+            next_broadcast(&mut last, ConnectionState::Online),
+            Some(ConnectionState::Online),
+            "coming back online must re-arm the subscriber's recovery work"
+        );
+    }
+
+    /// Review scenario (PR #364): the monitor has published `Online`, then
+    /// the user removes the only connected relay — `remove_relay` publishes
+    /// directly. When the remaining relay later connects, the monitor's
+    /// `Online` is a genuine transition and must not be dropped because the
+    /// monitor never saw the removal's broadcast. All publishers share one
+    /// gate.
+    #[tokio::test]
+    async fn direct_and_monitor_publishers_share_one_gate() {
+        let pool = RelayPool::new(vec![
+            "ws://127.0.0.1:1".to_string(),
+            "ws://127.0.0.1:2".to_string(),
+        ])
+        .await
+        .unwrap();
+        let mut rx = pool.subscribe_connection_state();
+
+        // Stand in for the monitor observing a connection: the pool is up.
+        pool.relays.write().await[0].status = RelayStatus::Connected;
+        broadcast_if_changed(
+            &pool.last_broadcast,
+            &pool.conn_tx,
+            &pool.relays.read().await,
+        );
+        assert_eq!(rx.try_recv().unwrap(), ConnectionState::Online);
+
+        // The connected relay is removed and the other is still `Connecting`
+        // in our view, so the removal derives `Reconnecting` — a real
+        // transition, sent directly.
+        pool.remove_relay("ws://127.0.0.1:1").await.unwrap();
+        assert_eq!(rx.try_recv().unwrap(), ConnectionState::Reconnecting);
+
+        // The surviving relay connects (what the monitor would observe) and
+        // the monitor's `Online` must get through: with a monitor-local gate
+        // its stale `Online` would suppress it.
+        pool.relays.write().await[0].status = RelayStatus::Connected;
+        broadcast_if_changed(
+            &pool.last_broadcast,
+            &pool.conn_tx,
+            &pool.relays.read().await,
+        );
+        assert_eq!(rx.try_recv().unwrap(), ConnectionState::Online);
+
+        // And the reverse: adding a relay while already online must not
+        // re-emit `Online` and re-run the whole recovery sequence.
+        pool.add_relay("ws://127.0.0.1:3").await.unwrap();
+        assert!(rx.try_recv().is_err(), "unchanged state must stay silent");
+    }
+
+    /// A publisher that has already observed the relay list must send before
+    /// a newer observation can be written and sent — otherwise subscribers
+    /// would end on a stale `Offline` after a genuine `Online`.
+    #[tokio::test]
+    async fn an_older_snapshot_is_never_sent_after_a_newer_one() {
+        let pool = RelayPool::new(vec!["ws://127.0.0.1:1".to_string()])
+            .await
+            .unwrap();
+        let mut rx = pool.subscribe_connection_state();
+
+        // Publisher A observes the relay drop but has not sent yet.
+        pool.relays.write().await[0].status = RelayStatus::Disconnected;
+        let snapshot = pool.relays.read().await;
+
+        // Publisher B observes the relay connect and wants to send `Online`.
+        // Its write must wait for A.
+        let b_pool = pool.clone();
+        let b = tokio::spawn(async move {
+            b_pool.relays.write().await[0].status = RelayStatus::Connected;
+            let relays = b_pool.relays.read().await;
+            broadcast_if_changed(&b_pool.last_broadcast, &b_pool.conn_tx, &relays);
+        });
+        for _ in 0..8 {
+            tokio::task::yield_now().await;
+        }
+        assert!(
+            !b.is_finished(),
+            "B must not publish while A holds its observation"
+        );
+        assert!(rx.try_recv().is_err());
+
+        broadcast_if_changed(&pool.last_broadcast, &pool.conn_tx, &snapshot);
+        drop(snapshot);
+        b.await.unwrap();
+
+        assert_eq!(rx.try_recv().unwrap(), ConnectionState::Offline);
+        assert_eq!(
+            rx.try_recv().unwrap(),
+            ConnectionState::Online,
+            "the newer state wins"
+        );
+        assert!(rx.try_recv().is_err());
+    }
+
     fn relay(url: &str, source: RelaySource) -> RelayInfo {
         RelayInfo {
             url: url.to_string(),