From a091d8747f0e8cf32aa00a461f616e6d72355344 Mon Sep 17 00:00:00 2001 From: grunch Date: Mon, 31 Aug 2026 08:42:51 -0300 Subject: [PATCH 1/3] fix(relay): broadcast connection state only when it actually changes The status monitor polls every relay every 2s and sent the derived connection state whenever any single relay's status changed -- including when the aggregate was unchanged, Online to Online. Every Online reaching the subscriber in api/nostr.rs runs a 10s capability fetch, an outbox flush, subscribe_orders, and a full resubscribe of chats and dispute chats. So one unreachable relay flapping on the poll interval reproduced all of that indefinitely, in the background, for the life of the session. The monitor now remembers the last state it broadcast and sends only on a real transition. Per-relay updates on relay_tx are untouched -- the UI relay list still reflects each relay individually. --- rust/src/nostr/relay_pool.rs | 72 +++++++++++++++++++++++++++++++++++- 1 file changed, 71 insertions(+), 1 deletion(-) diff --git a/rust/src/nostr/relay_pool.rs b/rust/src/nostr/relay_pool.rs index ae4a1a37..f42c88d8 100644 --- a/rust/src/nostr/relay_pool.rs +++ b/rust/src/nostr/relay_pool.rs @@ -162,6 +162,10 @@ impl RelayPool { let relay_tx = self.relay_tx.clone(); crate::rt::spawn(async move { + // Last state actually broadcast, so a poll that changes a relay + // without changing the aggregate stays silent. + let mut last_state: Option = None; + loop { crate::rt::time::sleep(Duration::from_secs(STATUS_POLL_INTERVAL_SECS)).await; @@ -210,7 +214,9 @@ impl RelayPool { if any_changed { let state = derive_connection_state(&relays.read().await); - let _ = conn_tx.send(state); + if let Some(state) = next_broadcast(&mut last_state, state) { + let _ = conn_tx.send(state); + } } } }); @@ -219,6 +225,24 @@ impl RelayPool { // ── Pure helpers ────────────────────────────────────────────────────────────── +/// The state to broadcast, or `None` when it has not actually changed. +/// +/// The monitor polls every relay, so one flapping relay ticks `any_changed` +/// on every poll while the derived state stays put. 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 bad relay into a permanent background storm. +fn next_broadcast( + last: &mut Option, + current: ConnectionState, +) -> Option { + if last.as_ref() == Some(¤t) { + return None; + } + *last = Some(current.clone()); + Some(current) +} + fn derive_connection_state(relays: &[RelayInfo]) -> ConnectionState { let any_connected = relays .iter() @@ -250,3 +274,49 @@ fn map_sdk_status(s: SdkRelayStatus) -> RelayStatus { } use crate::rt::unix_now; + +#[cfg(test)] +mod tests { + use super::*; + + /// One relay flapping while another stays connected changes a relay's + /// status on every poll without changing the derived state. Broadcasting + /// that re-ran a capability fetch, an outbox flush and a full resubscribe + /// of orders and chats every two seconds, 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" + ); + } +} From 138e9e10b44b46c8d5ad25d26250492bf1077aba Mon Sep 17 00:00:00 2001 From: grunch Date: Thu, 3 Sep 2026 14:41:00 -0300 Subject: [PATCH 2/3] fix(relay): share the connection-state gate across all publishers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round 1 (Codex P2 + Catrya's reproduction). The dedupe state lived inside the status monitor task, while `new`, `add_relay` and `remove_relay` still sent on `conn_tx` unconditionally. Two consequences: adding a relay while already online re-emitted `Online` (the storm this PR is named after, on a different path), and a direct `Offline` from `remove_relay` left the monitor's view stale so its next genuine `Online` was dropped as a duplicate — subscribers believed the pool was down for the rest of the session and the outbox never flushed. `last_broadcast` now lives on `RelayPool` and every publisher goes through `broadcast_if_changed`. New pool-level test reproduces the review scenario (direct `Reconnecting` after removal, then the monitor's `Online` must pass; an add while online must stay silent). Also corrects the description of the trigger, per the measurement in review: the storm needs a relay that connects and drops (cadence = SDK reconnect backoff), not one that is merely unreachable. Plan item 2.5 updated with what shipped and the debounce half recorded as the remaining gap. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013gRMS1h8Nuux1CARsLd8db --- docs/OPTIMIZATION_PLAN.md | 26 +++++++--- rust/src/nostr/relay_pool.rs | 97 +++++++++++++++++++++++++++++------- 2 files changed, 97 insertions(+), 26 deletions(-) diff --git a/docs/OPTIMIZATION_PLAN.md b/docs/OPTIMIZATION_PLAN.md index 297965b2..e7e71a84 100644 --- a/docs/OPTIMIZATION_PLAN.md +++ b/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` diff --git a/rust/src/nostr/relay_pool.rs b/rust/src/nostr/relay_pool.rs index d00eceb3..3d8e41b4 100644 --- a/rust/src/nostr/relay_pool.rs +++ b/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>, conn_tx: broadcast::Sender, + /// 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>>, relay_tx: broadcast::Sender, } @@ -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, }); @@ -234,7 +239,7 @@ impl RelayPool { async fn broadcast_connection_state(&self) { let state = derive_connection_state(&self.relays.read().await); - let _ = self.conn_tx.send(state); + broadcast_if_changed(&self.last_broadcast, &self.conn_tx, state); } /// Spawn a background task that polls each relay's SDK status every @@ -249,13 +254,10 @@ 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 { - // Last state actually broadcast, so a poll that changes a relay - // without changing the aggregate stays silent. - let mut last_state: Option = None; - loop { crate::rt::time::sleep(Duration::from_secs(STATUS_POLL_INTERVAL_SECS)).await; @@ -304,9 +306,7 @@ impl RelayPool { if any_changed { let state = derive_connection_state(&relays.read().await); - if let Some(state) = next_broadcast(&mut last_state, state) { - let _ = conn_tx.send(state); - } + broadcast_if_changed(&last_broadcast, &conn_tx, state); } } }); @@ -315,13 +315,36 @@ impl RelayPool { // ── Pure helpers ────────────────────────────────────────────────────────────── +/// Send `state` 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. +fn broadcast_if_changed( + last: &Mutex>, + tx: &broadcast::Sender, + state: ConnectionState, +) { + 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 polls every relay, so one flapping relay ticks `any_changed` -/// on every poll while the derived state stays put. 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 bad relay into a permanent background storm. +/// 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, current: ConnectionState, @@ -389,10 +412,10 @@ use crate::rt::unix_now; mod tests { use super::*; - /// One relay flapping while another stays connected changes a relay's - /// status on every poll without changing the derived state. Broadcasting - /// that re-ran a capability fetch, an outbox flush and a full resubscribe - /// of orders and chats every two seconds, indefinitely. + /// 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; @@ -430,6 +453,44 @@ mod tests { ); } + /// 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: the pool is up. + broadcast_if_changed(&pool.last_broadcast, &pool.conn_tx, ConnectionState::Online); + assert_eq!(rx.try_recv().unwrap(), ConnectionState::Online); + + // The remaining relay is still `Connecting` in our view, so the + // removal derives `Reconnecting` — a real transition, sent directly. + pool.remove_relay("ws://127.0.0.1:2").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, ConnectionState::Online); + 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"); + } + fn relay(url: &str, source: RelaySource) -> RelayInfo { RelayInfo { url: url.to_string(), From cc0de7e38ed43c678dcef8bc708c46f51dd474a8 Mon Sep 17 00:00:00 2001 From: grunch Date: Thu, 3 Sep 2026 14:51:40 -0300 Subject: [PATCH 3/3] fix(relay): derive the connection state under the relays read guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round 2 (CodeRabbit). `broadcast_connection_state` and the monitor derived the state under the `relays` read lock, released it, and only then took the gate. On the multi-threaded runtime two publishers could interleave there: a newer observation written, derived and sent before an older snapshot, which then landed last and left subscribers on a stale state. `broadcast_if_changed` now takes the relay list and derives inside, so every caller derives and sends while still holding the read guard — no writer can slip in between. The gate mutex is still never held across an await. New test holds one publisher's observation, has a second one write and publish a newer state, and asserts the older snapshot is delivered first. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_013gRMS1h8Nuux1CARsLd8db --- rust/src/nostr/relay_pool.rs | 85 +++++++++++++++++++++++++++++++----- 1 file changed, 73 insertions(+), 12 deletions(-) diff --git a/rust/src/nostr/relay_pool.rs b/rust/src/nostr/relay_pool.rs index 3d8e41b4..275ef58e 100644 --- a/rust/src/nostr/relay_pool.rs +++ b/rust/src/nostr/relay_pool.rs @@ -238,8 +238,8 @@ impl RelayPool { // ── Internal helpers ────────────────────────────────────────────────────── async fn broadcast_connection_state(&self) { - let state = derive_connection_state(&self.relays.read().await); - broadcast_if_changed(&self.last_broadcast, &self.conn_tx, 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 @@ -305,8 +305,8 @@ impl RelayPool { } if any_changed { - let state = derive_connection_state(&relays.read().await); - broadcast_if_changed(&last_broadcast, &conn_tx, state); + let relays_r = relays.read().await; + broadcast_if_changed(&last_broadcast, &conn_tx, &relays_r); } } }); @@ -315,7 +315,8 @@ impl RelayPool { // ── Pure helpers ────────────────────────────────────────────────────────────── -/// Send `state` on `tx` only if it differs from what was last sent. +/// 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 @@ -323,11 +324,17 @@ impl RelayPool { /// 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>, tx: &broadcast::Sender, - state: 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); @@ -469,20 +476,30 @@ mod tests { .unwrap(); let mut rx = pool.subscribe_connection_state(); - // Stand in for the monitor: the pool is up. - broadcast_if_changed(&pool.last_broadcast, &pool.conn_tx, ConnectionState::Online); + // 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 remaining relay is still `Connecting` in our view, so the - // removal derives `Reconnecting` — a real transition, sent directly. - pool.remove_relay("ws://127.0.0.1:2").await.unwrap(); + // 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, ConnectionState::Online); + 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 @@ -491,6 +508,50 @@ mod tests { 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(),