Merge #c1b384df: fix(sync): use bounded cooldown for transient negentro…

fix(sync): use bounded cooldown for transient negentropy failures

nostr:nevent1qgsx2lyl2e4zvfadwcvkd9fkrcwczj7mf858hy85mwqclwgut8wpg2spz3mhxue69uhhyetvv9ujumn8d96zuer9wcq3yamnwvaz7tm8d96xummnw3ezucm0d5q3kamnwvaz7tmwva5hgtnyv9hxxmmwwashjer9wchxxmmdqqsvrvuyml2nqtsnpfjz0vrmrfnw52apdvggskpwpr82pqkt6aaq0dqy9k7hy

PR-Author: DanConwayDev's Agent
nostr:npub1v47f74n2ycn66asev62nv8sas99akj0g0wg0fkup37u3ckwuzs4q7cwtp0

CoverNote:

One transient negentropy diff error permanently disabled NIP-77 for the relay connection. On gitnostr.com (2026-08-04, PR commit 256a9912) a single client-side "channel lagged by 3" during the startup burst marked the bootstrap relay non-NIP-77 within seconds of startup; 20 relays were marked in the first minutes and reconciliation collapsed from 739 runs to 24 after 14:30 UTC. 93% of observed failures (timeout, channel lagged, blocked/rate-limit) carry no information about NIP-77 support, and the affected relays advertise NIP-77 in their NIP-11 documents.

This PR classifies diff failures: only explicit unsupported signals ("negentropy not supported", "server does not support our negentropy protocol version") permanently mark the connection; everything else applies an escalating per-relay cooldown (60s/5m/30m/2h) that a successful diff resets. Failures arriving while a cooldown is active come from diffs already in flight when it started and do not escalate the backoff. Per-batch REQ+EOSE fallback is unchanged, so sync progress never depends on the classification. The concurrent-abort error no longer fabricates a NOTICE-based detection.

## Production verification (gitnostr.com)

Deployed from this PR's commit `ecb6c8b68caa77ebb8b894145101af55124006c5` at 2026-08-04 19:24:49 UTC, before merge. Soak window 19:24-19:51 UTC, covering service startup — the window in which the failure always fired:

- The misclassification signature ("does not support NIP-77 (detected via NOTICE)") occurred 0 times; on the previous revision it fired 98 times on the bootstrap relay within seconds of startup.
- The bootstrap relay wss://relay.ngit.dev was neither marked unsupported nor placed in cooldown; 845 negentropy reconciliations completed, continuing past startup instead of collapsing fleet-wide to REQ+EOSE.
- Escalating cooldown observed end-to-end in production: wss://wheat.happytavern.co timed out at 19:26:22 → 60s cooldown → retried after expiry → timed out again at 19:31:39 → escalated to 300s.
- Genuine unsupported detection still works: wss://relay.primal.net sent NOTICE "bad msg: negentropy disabled" and was permanently marked.
- Startup NEG bursts absorbed: 13 transient-cooldown warnings total across 10 relays (previously hundreds of cascading failures); nos.lol's 34 concurrent NEG-request rejections produced a single 60s cooldown.
- No new WARN/ERROR signatures versus the previous revision.

Remaining observation for a future fix: the startup historic sync still opens enough concurrent NEG requests to draw rate-limit NOTICEs ("ERROR: too many concurrent NEG requests" from nos.lol, 34 at startup) and per-filter timeouts; the cooldown now contains the damage, but bounding sync concurrency at startup would remove the cause.

Part of the sync-stabilisation tracker: nostr:nevent1qqspxxvcaqj96zt8sdj520h9nl4rtq3873m6za6keg2gm2xn6dgp55cpz3mhxue69uhhyetvv9ujumn8d96zuer9wc85zw09

Cargo.toml

@@ -82,6 +82,7 @@ tempfile = "3"
 # Testing
 grasp-audit = { path = "grasp-audit", version = "0.2.0" }
 tempfile = "3"
+tokio = { version = "1.35", features = ["full", "test-util"] }
 reqwest = { version = "0.13", default-features = false, features = ["native-tls"] }
 tokio-tungstenite = "0.28.0"
 

src/sync/relay_connection.rs

@@ -30,6 +30,43 @@ use crate::outbound::{OutboundTargetKind, OutboundTargetPolicy, RelayTargetSourc
 /// exchange alive indefinitely by continuing to send reconciliation messages.
 const NEGENTROPY_DIFF_TIMEOUT: Duration = Duration::from_secs(15);
 
+/// Cooldown schedule for transient negentropy failures.
+///
+/// Indexed by the number of consecutive failed attempts (capped at the last
+/// entry). Transient failures pause NIP-77 for this relay temporarily instead
+/// of disabling it for the connection lifetime: a client-side channel overflow
+/// or a relay rate limit says nothing about whether the relay speaks NIP-77.
+const NEGENTROPY_TRANSIENT_BACKOFF: [Duration; 4] = [
+    Duration::from_secs(60),
+    Duration::from_secs(300),
+    Duration::from_secs(1800),
+    Duration::from_secs(7200),
+];
+
+/// How a failed negentropy diff should affect future NIP-77 attempts.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum NegentropyFailure {
+    /// The relay explicitly reported that it cannot speak NIP-77.
+    Unsupported,
+    /// Timeouts, rate limits, client-side channel overflow and other errors
+    /// that carry no information about NIP-77 support.
+    Transient,
+}
+
+/// Classify a negentropy diff error string.
+///
+/// Only explicit "not supported" signals from the relay justify permanently
+/// disabling NIP-77. Everything else is treated as transient; the relay stays
+/// eligible for negentropy after a bounded cooldown.
+fn classify_negentropy_failure(error: &str) -> NegentropyFailure {
+    let lower = error.to_lowercase();
+    if lower.contains("not support") || lower.contains("unsupported") {
+        NegentropyFailure::Unsupported
+    } else {
+        NegentropyFailure::Transient
+    }
+}
+
 /// Interval between relay-status checks while a connection attempt is in flight.
 ///
 /// nostr-sdk may return from `try_connect_relay` while another task still has
@@ -116,8 +153,12 @@ pub struct RelayConnection {
     database: Option<SharedDatabase>,
     /// Whether we've logged NIP-77 not supported for this relay (log once)
     nip77_warning_logged: std::sync::Arc<std::sync::atomic::AtomicBool>,
-    /// Whether this relay supports NIP-77 negentropy (None = unknown, Some(false) = confirmed not supported)
+    /// Whether this relay supports NIP-77 negentropy (0 = unknown, 2 = confirmed not supported)
     nip77_supported: std::sync::Arc<std::sync::atomic::AtomicU8>,
+    /// Consecutive transient negentropy failures (drives the cooldown schedule)
+    nip77_transient_failures: std::sync::Arc<std::sync::atomic::AtomicU32>,
+    /// Deadline before which negentropy is not attempted (transient-failure cooldown)
+    nip77_cooldown_until: std::sync::Arc<std::sync::Mutex<Option<tokio::time::Instant>>>,
 }
 
 impl RelayConnection {
@@ -169,6 +210,8 @@ impl RelayConnection {
             database: None,
             nip77_warning_logged: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
             nip77_supported: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(0)),
+            nip77_transient_failures: std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)),
+            nip77_cooldown_until: std::sync::Arc::new(std::sync::Mutex::new(None)),
         }
     }
 
@@ -199,6 +242,8 @@ impl RelayConnection {
             database: Some(database),
             nip77_warning_logged: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
             nip77_supported: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(0)),
+            nip77_transient_failures: std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)),
+            nip77_cooldown_until: std::sync::Arc::new(std::sync::Mutex::new(None)),
         }
     }
 
@@ -614,7 +659,8 @@ impl RelayConnection {
     /// - The nostr-sdk 0.44 API for relay document access varies
     ///
     /// # Returns
-    /// * `false` if we've confirmed this relay doesn't support NIP-77
+    /// * `false` if we've confirmed this relay doesn't support NIP-77, or a
+    ///   transient-failure cooldown is active
     /// * `true` if unknown or supported (will attempt and handle failure)
     pub async fn supports_negentropy(&self) -> bool {
         // 0 = unknown (try it), 1 = supported, 2 = confirmed not supported
@@ -628,16 +674,26 @@ impl RelayConnection {
             return false;
         }
 
+        // Transient-failure cooldown: pause negentropy without ruling it out
+        let cooldown_until = *self
+            .nip77_cooldown_until
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        if matches!(cooldown_until, Some(until) if tokio::time::Instant::now() < until) {
+            tracing::trace!(relay = %self.url, "Skipping negentropy - transient-failure cooldown active");
+            return false;
+        }
+
         // Unknown or supported - try it
         true
     }
 
     /// Mark this relay as not supporting NIP-77 negentropy
     ///
-    /// Called when we detect negentropy isn't working for this relay:
-    /// - NOTICE message contains negentropy-related error
-    /// - negentropy_sync_diff() fails
-    /// - Negentropy retry returns zero events
+    /// Called only when the relay explicitly signals it cannot speak NIP-77
+    /// (see [`classify_negentropy_failure`]), or when a negentropy retry
+    /// returns zero events (see zero-progress handling in `sync::mod`).
+    /// Transient failures use a bounded cooldown instead.
     ///
     /// Future batches will skip negentropy and use REQ+EOSE directly.
     pub fn mark_negentropy_unsupported(&self) {
@@ -645,6 +701,44 @@ impl RelayConnection {
             .store(2, std::sync::atomic::Ordering::Relaxed);
     }
 
+    /// Record a transient negentropy failure and start (or keep) a cooldown.
+    ///
+    /// Failures that arrive while a cooldown is already active come from
+    /// diffs that were in flight when the cooldown started (e.g. a burst of
+    /// concurrent per-filter syncs sharing one root cause); they do not
+    /// escalate the backoff.
+    ///
+    /// # Returns
+    /// The newly applied cooldown, or `None` if an existing cooldown
+    /// absorbed the failure.
+    fn record_negentropy_transient_failure(&self) -> Option<Duration> {
+        let mut cooldown_until = self
+            .nip77_cooldown_until
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        let now = tokio::time::Instant::now();
+        if matches!(*cooldown_until, Some(until) if now < until) {
+            return None;
+        }
+        let failures = self
+            .nip77_transient_failures
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+        let index = (failures as usize).min(NEGENTROPY_TRANSIENT_BACKOFF.len() - 1);
+        let cooldown = NEGENTROPY_TRANSIENT_BACKOFF[index];
+        *cooldown_until = Some(now + cooldown);
+        Some(cooldown)
+    }
+
+    /// Record a successful negentropy diff: clear the transient-failure state.
+    fn record_negentropy_success(&self) {
+        self.nip77_transient_failures
+            .store(0, std::sync::atomic::Ordering::Relaxed);
+        *self
+            .nip77_cooldown_until
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
+    }
+
     /// Perform a negentropy sync diff (dry run) to identify missing events
     ///
     /// This method performs NIP-77 negentropy reconciliation without downloading events.
@@ -713,9 +807,9 @@ impl RelayConnection {
             loop {
                 let status = nip77_status.load(std::sync::atomic::Ordering::Relaxed);
                 if status == 2 {
-                    // Relay confirmed not to support NIP-77 (via NOTICE or other means)
+                    // A concurrent sync received an explicit unsupported signal
                     return Err(format!(
-                        "Relay {} does not support NIP-77 (detected via NOTICE)",
+                        "Relay {} was marked as not supporting NIP-77 by a concurrent sync",
                         url
                     ));
                 }
@@ -744,6 +838,7 @@ impl RelayConnection {
 
         match result {
             Ok(reconciliation) => {
+                self.record_negentropy_success();
                 tracing::debug!(
                     relay = %self.url,
                     local_count = reconciliation.local.len(),
@@ -753,18 +848,40 @@ impl RelayConnection {
                 Ok(reconciliation)
             }
             Err(e) => {
-                self.mark_negentropy_unsupported();
+                match classify_negentropy_failure(&e) {
+                    NegentropyFailure::Unsupported => {
+                        self.mark_negentropy_unsupported();
 
-                // Log warning only once per relay to avoid spam
-                if !self
-                    .nip77_warning_logged
-                    .swap(true, std::sync::atomic::Ordering::Relaxed)
-                {
-                    tracing::warn!(
-                        relay = %self.url,
-                        error = %e,
-                        "Negentropy diff failed, will fall back to REQ+EOSE"
-                    );
+                        // Log warning only once per relay to avoid spam
+                        if !self
+                            .nip77_warning_logged
+                            .swap(true, std::sync::atomic::Ordering::Relaxed)
+                        {
+                            tracing::warn!(
+                                relay = %self.url,
+                                error = %e,
+                                "Relay does not support NIP-77, will fall back to REQ+EOSE"
+                            );
+                        }
+                    }
+                    NegentropyFailure::Transient => {
+                        // One warning per cooldown; in-flight failures sharing
+                        // the same root cause are logged at debug level
+                        if let Some(cooldown) = self.record_negentropy_transient_failure() {
+                            tracing::warn!(
+                                relay = %self.url,
+                                error = %e,
+                                cooldown_secs = cooldown.as_secs(),
+                                "Transient negentropy failure, pausing NIP-77 and falling back to REQ+EOSE"
+                            );
+                        } else {
+                            tracing::debug!(
+                                relay = %self.url,
+                                error = %e,
+                                "Negentropy diff failed during existing cooldown"
+                            );
+                        }
+                    }
                 }
                 Err(e)
             }
@@ -834,8 +951,8 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn hung_negentropy_diff_times_out_and_disables_future_attempts() {
+    #[tokio::test(start_paused = true)]
+    async fn hung_negentropy_diff_times_out_and_pauses_attempts_until_cooldown() {
         let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
         assert!(connection.supports_negentropy().await);
 
@@ -851,7 +968,167 @@ mod tests {
             .contains("timed out"));
         assert!(
             !connection.supports_negentropy().await,
-            "a timed-out relay must use REQ+EOSE for future historic batches"
+            "a timed-out relay must use REQ+EOSE while the cooldown is active"
+        );
+
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[0] + Duration::from_secs(1)).await;
+        assert!(
+            connection.supports_negentropy().await,
+            "a timeout must not disable NIP-77 permanently"
+        );
+    }
+
+    /// Run a diff seeded with a fixed error and assert it surfaces as a failure.
+    async fn failing_diff(connection: &RelayConnection, error: &str) {
+        let error = error.to_string();
+        connection
+            .run_negentropy_diff_with_timeout(async move { Err(error) }, Duration::from_secs(5))
+            .await
+            .expect_err("diff seeded with an error must fail");
+    }
+
+    #[test]
+    fn production_failure_strings_classify_as_transient_or_unsupported() {
+        // Transient errors carry no information about NIP-77 support
+        // (all observed on gitnostr.com, 2026-08-04)
+        for error in [
+            r#"Negentropy diff had failures: {RelayUrl("wss://relay.ngit.dev"): "channel lagged by 3"}"#,
+            r#"Negentropy diff had failures: {RelayUrl("wss://nos.lol"): "timeout"}"#,
+            r#"Negentropy diff had failures: {RelayUrl("wss://relay.cyberguy.fyi"): "blocked: too many subscriptions"}"#,
+            "Negentropy diff timed out after 15.000s",
+        ] {
+            assert_eq!(
+                classify_negentropy_failure(error),
+                NegentropyFailure::Transient,
+                "{error}"
+            );
+        }
+
+        // Explicit unsupported signals from the relay
+        for error in [
+            r#"Negentropy diff had failures: {RelayUrl("wss://example.com"): "negentropy not supported"}"#,
+            r#"Negentropy diff had failures: {RelayUrl("wss://example.com"): "server does not support our negentropy protocol version"}"#,
+        ] {
+            assert_eq!(
+                classify_negentropy_failure(error),
+                NegentropyFailure::Unsupported,
+                "{error}"
+            );
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn transient_diff_failure_pauses_negentropy_only_until_cooldown_elapses() {
+        let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
+
+        failing_diff(
+            &connection,
+            r#"Negentropy diff had failures: {RelayUrl("wss://relay.example.com"): "channel lagged by 3"}"#,
+        )
+        .await;
+
+        assert!(
+            !connection.supports_negentropy().await,
+            "a transient failure must pause NIP-77 while the cooldown is active"
+        );
+
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[0] + Duration::from_secs(1)).await;
+        assert!(
+            connection.supports_negentropy().await,
+            "a transient failure must not disable NIP-77 permanently"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn concurrent_burst_of_transient_failures_does_not_escalate_backoff() {
+        let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
+
+        for _ in 0..40 {
+            failing_diff(
+                &connection,
+                r#"Negentropy diff had failures: {RelayUrl("wss://relay.example.com"): "timeout"}"#,
+            )
+            .await;
+        }
+
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[0] + Duration::from_secs(1)).await;
+        assert!(
+            connection.supports_negentropy().await,
+            "in-flight failures sharing one root cause must not escalate the cooldown"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn repeated_transient_failures_escalate_and_success_resets_backoff() {
+        let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
+        let transient =
+            r#"Negentropy diff had failures: {RelayUrl("wss://relay.example.com"): "timeout"}"#;
+
+        // First failed attempt starts the first cooldown
+        failing_diff(&connection, transient).await;
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[0] + Duration::from_secs(1)).await;
+        assert!(connection.supports_negentropy().await);
+
+        // Second consecutive failed attempt backs off longer
+        failing_diff(&connection, transient).await;
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[0] + Duration::from_secs(1)).await;
+        assert!(
+            !connection.supports_negentropy().await,
+            "consecutive failed attempts must back off longer"
+        );
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[1]).await;
+        assert!(connection.supports_negentropy().await);
+
+        // A successful diff resets the schedule to the first cooldown
+        connection
+            .run_negentropy_diff_with_timeout(
+                async { Ok(nostr_sdk::client::SyncSummary::default()) },
+                Duration::from_secs(5),
+            )
+            .await
+            .expect("successful diff");
+        failing_diff(&connection, transient).await;
+        tokio::time::advance(NEGENTROPY_TRANSIENT_BACKOFF[0] + Duration::from_secs(1)).await;
+        assert!(
+            connection.supports_negentropy().await,
+            "a successful diff must reset the backoff schedule"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn explicit_unsupported_signal_disables_negentropy_for_the_connection() {
+        let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
+
+        failing_diff(
+            &connection,
+            r#"Negentropy diff had failures: {RelayUrl("wss://relay.example.com"): "negentropy not supported"}"#,
+        )
+        .await;
+
+        assert!(!connection.supports_negentropy().await);
+        tokio::time::advance(Duration::from_secs(48 * 3600)).await;
+        assert!(
+            !connection.supports_negentropy().await,
+            "an explicit unsupported signal is permanent for the connection lifetime"
+        );
+    }
+
+    #[tokio::test]
+    async fn concurrent_unsupported_marking_aborts_diff_without_claiming_a_notice() {
+        let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
+        connection.mark_negentropy_unsupported();
+
+        let error = connection
+            .run_negentropy_diff_with_timeout(
+                pending::<Result<nostr_sdk::client::SyncSummary, String>>(),
+                Duration::from_secs(5),
+            )
+            .await
+            .expect_err("a relay marked unsupported must abort in-flight diffs");
+
+        assert!(
+            !error.contains("NOTICE"),
+            "abort reason must not fabricate a NOTICE-based detection: {error}"
         );
     }