nostr: per-peer NAT-traversal failure suppression and clock-skew handling

Public-test daemons with populous open-discovery caches generate
sustained NAT-traversal-failure WARN volume (~140/hour, ~3500/day)
against cache-learned peers that have gone offline — their adverts
are absent from major Nostr relays but cached entries persist until
their advertised `valid_until` expires. The daemon kept publishing
offers indefinitely under exponential backoff with no per-peer
suppression, drowning operator signal and hammering relays. A
parallel concern: the strict freshness check at signal.rs silently
rejected offers under modest clock skew (now_ms() anchors to
SystemTime once at startup, so post-startup NTP step adjustments
don't propagate on long-uptime daemons), indistinguishable from
"peer is offline."

Six independent improvements layered on the existing retry logic.

Per-npub WARN log rate-limit
----------------------------
New `FailureState` struct on `NostrDiscovery` records per-npub
`last_warn_at_ms`. Subsequent failures inside `warn_log_interval_secs`
(default 5 min) emit DEBUG instead of WARN. Each WARN now also
carries `consecutive_failures` and remaining `cooldown_secs` so
operators can read the trajectory without grepping multiple lines.

Per-npub consecutive-failure counter + extended cooldown
--------------------------------------------------------
After `failure_streak_threshold` (default 5) consecutive failures
against a peer, the next `extended_cooldown_secs` (default 1800)
of attempts are suppressed by pushing
`retry_pending[npub].retry_after_ms` past the cooldown wall. The
open-discovery sweep also consults `cooldown_until` and increments
a new `skipped_cooldown` counter so a peer whose `retry_pending`
was cleared by max_retries doesn't get re-enqueued during the
cooldown window. Caps offer-publish rate per dead peer regardless
of how often the sweep tries to re-enqueue.

Stale-advert eviction on streak-threshold transition
----------------------------------------------------
On the threshold-crossing transition (one-shot, not every
subsequent failure), `tokio::spawn` an active re-fetch of the
peer's Kind 37195 advert from `advert_relays`. Three outcomes:
- absent on relays → cache evicted; sweep won't re-enqueue
  (peer is genuinely gone).
- newer `created_at` → cache refreshed + streak reset
  (peer republished; allowed to retry immediately).
- same → cache untouched; cooldown stands.

Cost: ~one fetch per dead peer per 30-min cooldown cycle, vs
hundreds of offer publishes/hour today.

Clock-skew tolerance on freshness check
---------------------------------------
`signal.rs` `validate_offer_freshness` and
`validate_traversal_answer_for_offer` now allow ±60s grace beyond
strict TTL. Both return a new `FreshnessOutcome` enum so callers
can DEBUG-log when an offer/answer was only accepted via the grace
window. `FRESHNESS_SKEW_TOLERANCE_MS` is hard-coded — loosening
this past minutes erodes the freshness/replay security boundary
and operators tend to tune in the wrong direction.

NTP-style skew estimate (offer_received_at echo)
------------------------------------------------
Added optional `offerReceivedAt: Option<u64>` field to
`TraversalAnswer` payload. Responder fills it with `now_ms()` at
offer-receipt time. Initiator computes the standard NTP offset
formula `((T2-T1) + (T3-T4)) / 2` against the round-trip and
DEBUG-logs when `|skew| ≥ 30s`. Skew is also stashed in
`FailureState` and surfaced in `show_peers`. Non-breaking — older
responders that don't fill the field still produce valid answers,
and `estimate_clock_skew` returns `None`.

Per-peer state in `show_peers` JSON
-----------------------------------
Each peer entry in `show_peers` now carries:

  "nostr_traversal": {
    "consecutive_failures": <u32>,
    "in_cooldown": <bool>,
    "cooldown_until_ms": <u64 | null>,
    "last_observed_skew_ms": <i64 | null>
  }

Always emitted (schema-stable); values populated when discovery is
enabled and the npub has a recorded entry. Required a new public
`Node::nostr_discovery_handle()` accessor and refactored
`FailureState`'s internal Mutex from `tokio::sync` to `std::sync`
(operations never hold across await), which lets the synchronous
`show_peers` handler call `snapshot()` directly without the
dispatcher becoming async.

New config knobs (under `node.discovery.nostr`)
-----------------------------------------------
  failure_streak_threshold: 5
  extended_cooldown_secs: 1800
  warn_log_interval_secs: 300
  failure_state_max_entries: 4096

Tests
-----
12 new unit tests:
- 5 in `tests.rs` covering freshness strict / tolerated / rejected
  outcomes, NTP skew estimation, and the backward-compat None case
  when the responder didn't fill `offer_received_at`.
- 7 in `failure_state.rs` covering streak/warn-rate-limit state
  transitions, cooldown active vs expired semantics,
  success-resets-streak, observed-skew records, and size-cap
  eviction by oldest `last_failure_at`.

CHANGELOG entries added under `[Unreleased]` Fixed.

CHANGELOG.md

@@ -274,6 +274,31 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Nostr-discovery now tolerates ±60s of clock skew on offer/answer
+  freshness checks so a responder whose wall clock leads the
+  initiator's by less than that no longer silently rejects every
+  offer. Previously, a public-test daemon with un-NTP'd peers (or
+  long uptime — `now_ms()` anchors to `SystemTime` once at startup,
+  then advances monotonically; post-startup NTP step adjustments
+  don't propagate) would see ~100% signal-timeout rate against
+  skewed peers, indistinguishable from "peer is offline." New
+  optional `offerReceivedAt` field on the answer payload lets the
+  initiator log per-peer NTP-style skew estimates (DEBUG when ≥30s)
+  for operator visibility. Backward-compatible — older responders
+  that don't fill the field still produce valid answers
+- Nostr-discovery NAT-traversal failure suppression: per-npub
+  consecutive-failure counter triggers a 30-min extended cooldown
+  after 5 failures, preventing the daemon from hammering Nostr
+  relays with offers to peers that have gone away. WARN log lines
+  rate-limited to one per peer per 5 min (subsequent failures
+  emit DEBUG with `consecutive_failures` + remaining `cooldown_secs`).
+  Threshold-crossing also fires a one-shot active re-check of the
+  peer's Kind 37195 advert against `advert_relays`; absent →
+  evict cache; newer → refresh + reset streak; same → cooldown
+  stands. New `failure_streak_threshold`, `extended_cooldown_secs`,
+  `warn_log_interval_secs`, `failure_state_max_entries` config
+  fields under `node.discovery.nostr`. Per-peer state visible in
+  `fipsctl show peers` JSON under `nostr_traversal`
 - Tor onion adverts published over Nostr overlay discovery now
   include the public-facing port (`<onion>.onion:<port>`) instead of
   just the bare onion hostname. The publisher previously emitted a

src/config/node.rs

@@ -365,6 +365,29 @@ pub struct NostrDiscoveryConfig {
     /// `policy: open`. Default: 3600 (1 hour).
     #[serde(default = "NostrDiscoveryConfig::default_startup_sweep_max_age_secs")]
     pub startup_sweep_max_age_secs: u64,
+    /// Number of consecutive NAT-traversal failures against a peer before
+    /// an extended cooldown is applied to throttle further offer publishes.
+    /// At this threshold the daemon also actively re-fetches the peer's
+    /// advert from `advert_relays` to evict cache entries for peers that
+    /// have gone away. Default: 5.
+    #[serde(default = "NostrDiscoveryConfig::default_failure_streak_threshold")]
+    pub failure_streak_threshold: u32,
+    /// Cooldown applied to a peer once `failure_streak_threshold` is hit.
+    /// Suppresses both open-discovery sweep enqueues and per-attempt
+    /// retry firings until elapsed. Default: 1800 (30 minutes).
+    #[serde(default = "NostrDiscoveryConfig::default_extended_cooldown_secs")]
+    pub extended_cooldown_secs: u64,
+    /// Minimum interval between `NAT traversal failed` WARN log lines for
+    /// the same peer. Subsequent failures inside the window log at DEBUG.
+    /// Reduces log spam on public-test nodes with many cache-learned
+    /// peers. Default: 300 (5 minutes).
+    #[serde(default = "NostrDiscoveryConfig::default_warn_log_interval_secs")]
+    pub warn_log_interval_secs: u64,
+    /// Maximum entries retained in the per-npub failure-state map.
+    /// Bounds memory under high cache turnover. Oldest entries (by last
+    /// failure time) evicted when the cap is exceeded. Default: 4096.
+    #[serde(default = "NostrDiscoveryConfig::default_failure_state_max_entries")]
+    pub failure_state_max_entries: usize,
 }
 
 impl Default for NostrDiscoveryConfig {
@@ -392,6 +415,10 @@ impl Default for NostrDiscoveryConfig {
             advert_refresh_secs: Self::default_advert_refresh_secs(),
             startup_sweep_delay_secs: Self::default_startup_sweep_delay_secs(),
             startup_sweep_max_age_secs: Self::default_startup_sweep_max_age_secs(),
+            failure_streak_threshold: Self::default_failure_streak_threshold(),
+            extended_cooldown_secs: Self::default_extended_cooldown_secs(),
+            warn_log_interval_secs: Self::default_warn_log_interval_secs(),
+            failure_state_max_entries: Self::default_failure_state_max_entries(),
         }
     }
 }
@@ -484,6 +511,22 @@ impl NostrDiscoveryConfig {
     fn default_startup_sweep_max_age_secs() -> u64 {
         3_600
     }
+
+    fn default_failure_streak_threshold() -> u32 {
+        5
+    }
+
+    fn default_extended_cooldown_secs() -> u64 {
+        1_800
+    }
+
+    fn default_warn_log_interval_secs() -> u64 {
+        300
+    }
+
+    fn default_failure_state_max_entries() -> usize {
+        4_096
+    }
 }
 
 /// Spanning tree (`node.tree.*`).

src/control/queries.rs

@@ -114,6 +114,18 @@ pub fn show_peers(node: &Node) -> Value {
     let parent_id = *tree.my_declaration().parent_id();
     let is_root = tree.is_root();
 
+    // Per-npub Nostr-traversal failure-state snapshot, indexed by npub
+    // for O(1) per-peer lookup. Empty if Nostr discovery is disabled.
+    let nostr_state: std::collections::HashMap<String, _> = node
+        .nostr_discovery_handle()
+        .map(|d| {
+            d.failure_state_snapshot()
+                .into_iter()
+                .map(|view| (view.npub.clone(), view))
+                .collect()
+        })
+        .unwrap_or_default();
+
     let peers: Vec<Value> = node
         .peers()
         .map(|peer| {
@@ -175,6 +187,31 @@ pub fn show_peers(node: &Node) -> Value {
             peer_json["replay_suppressed"] = json!(peer.replay_suppressed_count());
             peer_json["consecutive_decrypt_failures"] = json!(peer.consecutive_decrypt_failures());
 
+            // Nostr-traversal state if this peer's npub appears in
+            // failure-state. Always emitted (even null) so the schema
+            // stays stable; values populated only when Nostr discovery
+            // is enabled and the npub has been seen.
+            let npub = peer.npub();
+            let mut nostr_obj = json!({
+                "consecutive_failures": 0,
+                "in_cooldown": false,
+                "cooldown_until_ms": Value::Null,
+                "last_observed_skew_ms": Value::Null,
+            });
+            if let Some(state) = nostr_state.get(&npub) {
+                nostr_obj["consecutive_failures"] = json!(state.consecutive_failures);
+                nostr_obj["in_cooldown"] = json!(state.cooldown_until_ms.is_some());
+                nostr_obj["cooldown_until_ms"] = state
+                    .cooldown_until_ms
+                    .map(|t| json!(t))
+                    .unwrap_or(Value::Null);
+                nostr_obj["last_observed_skew_ms"] = state
+                    .last_observed_skew_ms
+                    .map(|s| json!(s))
+                    .unwrap_or(Value::Null);
+            }
+            peer_json["nostr_traversal"] = nostr_obj;
+
             // Noise session counters (rekey urgency, replay window state)
             if let Some(session) = peer.noise_session() {
                 peer_json["noise"] = json!({

src/discovery/nostr/failure_state.rs

@@ -0,0 +1,311 @@
+//! Per-npub NAT-traversal failure tracking.
+//!
+//! Records consecutive offer/answer signal-timeout (and other) failures
+//! against each peer. Drives three operator-visible behaviors:
+//!
+//! - **WARN log rate-limit (B1).** Suppresses repeat WARN lines for the
+//!   same peer inside a configurable window; subsequent failures inside
+//!   the window log at DEBUG instead.
+//! - **Extended cooldown (B2).** Once a peer accumulates
+//!   `failure_streak_threshold` consecutive failures, the next
+//!   `extended_cooldown_secs` worth of attempts are suppressed by pushing
+//!   the retry timer out, capping how aggressively a public-test node can
+//!   hammer Nostr relays with offers to dead peers.
+//! - **Stale-advert eviction (B6).** At streak threshold, the daemon
+//!   actively re-fetches the peer's advert; outcomes (`Evicted`,
+//!   `Refreshed`, `SameAdvert`, `Skipped`) feed back into the cache so
+//!   peers that have actually disappeared stop being retried after
+//!   eviction (`prune_advert_cache` semantics).
+//!
+//! Also stores last-observed clock skew (from B5a) so operators can
+//! surface it via `fipsctl show peers` (B3).
+
+use std::collections::HashMap;
+use std::sync::Mutex;
+
+/// One peer's failure-tracking state. Keyed by bech32 npub string in the
+/// owning `FailureState` map.
+#[derive(Debug, Clone)]
+pub(super) struct NpubFailureRecord {
+    /// Number of consecutive failures since the last success (or fresh
+    /// advert that reset the streak).
+    pub consecutive_failures: u32,
+    /// When this entry was last touched, used for size-cap eviction.
+    pub last_failure_at_ms: u64,
+    /// When the last WARN was emitted for this peer; controls WARN
+    /// rate-limit (B1).
+    pub last_warn_at_ms: Option<u64>,
+    /// When the extended cooldown was applied; while
+    /// `cooldown_until_ms.is_some_and(|t| t > now)`, retries are
+    /// suppressed.
+    pub cooldown_until_ms: Option<u64>,
+    /// Most recent NTP-style skew estimate (B5a), in ms (positive =
+    /// peer ahead of us). `None` if the peer hasn't successfully
+    /// answered an offer with `offerReceivedAt` populated, or if
+    /// successful traversal cleared the streak (we keep the last-seen
+    /// skew, but only on records that are still in the map).
+    pub last_observed_skew_ms: Option<i64>,
+}
+
+impl NpubFailureRecord {
+    fn new(now_ms: u64) -> Self {
+        Self {
+            consecutive_failures: 0,
+            last_failure_at_ms: now_ms,
+            last_warn_at_ms: None,
+            cooldown_until_ms: None,
+            last_observed_skew_ms: None,
+        }
+    }
+}
+
+/// What the lifecycle layer should do based on the recorded failure.
+#[derive(Debug, Clone, Copy)]
+pub(super) struct FailureDecision {
+    /// Updated streak count (post-increment).
+    pub consecutive_failures: u32,
+    /// True iff lifecycle should log at WARN; false → log at DEBUG.
+    pub should_warn: bool,
+    /// If set, retry_after_ms for this peer should not fire before this
+    /// wall-clock ms.
+    pub cooldown_until_ms: Option<u64>,
+    /// True only on the streak-threshold-crossing transition. Lifecycle
+    /// should run a one-shot stale-advert check (B6) when this fires.
+    pub crossed_threshold: bool,
+}
+
+pub(super) struct FailureState {
+    inner: Mutex<HashMap<String, NpubFailureRecord>>,
+    threshold: u32,
+    extended_cooldown_ms: u64,
+    warn_log_interval_ms: u64,
+    max_entries: usize,
+}
+
+impl FailureState {
+    pub(super) fn new(
+        threshold: u32,
+        extended_cooldown_secs: u64,
+        warn_log_interval_secs: u64,
+        max_entries: usize,
+    ) -> Self {
+        Self {
+            inner: Mutex::new(HashMap::new()),
+            threshold,
+            extended_cooldown_ms: extended_cooldown_secs.saturating_mul(1000),
+            warn_log_interval_ms: warn_log_interval_secs.saturating_mul(1000),
+            max_entries,
+        }
+    }
+
+    /// Record a traversal failure against `npub`. Returns the resulting
+    /// FailureDecision for the lifecycle layer to act on.
+    pub(super) fn record_failure(&self, npub: &str, now_ms: u64) -> FailureDecision {
+        let mut map = self.inner.lock().expect("failure-state mutex poisoned");
+        let entry = map
+            .entry(npub.to_string())
+            .or_insert_with(|| NpubFailureRecord::new(now_ms));
+        entry.consecutive_failures = entry.consecutive_failures.saturating_add(1);
+        entry.last_failure_at_ms = now_ms;
+
+        let crossed_threshold = entry.consecutive_failures == self.threshold;
+        let cooldown_until_ms = if entry.consecutive_failures >= self.threshold {
+            let cooldown = now_ms.saturating_add(self.extended_cooldown_ms);
+            entry.cooldown_until_ms = Some(cooldown);
+            Some(cooldown)
+        } else {
+            None
+        };
+
+        let should_warn = !matches!(
+            entry.last_warn_at_ms,
+            Some(last) if now_ms.saturating_sub(last) < self.warn_log_interval_ms
+        );
+        if should_warn {
+            entry.last_warn_at_ms = Some(now_ms);
+        }
+
+        let decision = FailureDecision {
+            consecutive_failures: entry.consecutive_failures,
+            should_warn,
+            cooldown_until_ms,
+            crossed_threshold,
+        };
+
+        if map.len() > self.max_entries {
+            evict_oldest(&mut map, self.max_entries);
+        }
+
+        decision
+    }
+
+    /// Record a successful traversal — clears the streak and cooldown.
+    /// Last-observed skew is retained until next eviction since it's
+    /// useful to display in `show_peers` even for healthy peers.
+    pub(super) fn record_success(&self, npub: &str, now_ms: u64) {
+        let mut map = self.inner.lock().expect("failure-state mutex poisoned");
+        if let Some(entry) = map.get_mut(npub) {
+            entry.consecutive_failures = 0;
+            entry.cooldown_until_ms = None;
+            entry.last_failure_at_ms = now_ms;
+        }
+        // No insert if absent — successful peers don't need a record.
+    }
+
+    /// Record an observed clock-skew estimate from a successful answer
+    /// receipt (B5a). Creates an entry if needed so we can surface the
+    /// skew via `show_peers` even when the peer is healthy.
+    pub(super) fn note_observed_skew(&self, npub: &str, skew_ms: i64, now_ms: u64) {
+        let mut map = self.inner.lock().expect("failure-state mutex poisoned");
+        let entry = map
+            .entry(npub.to_string())
+            .or_insert_with(|| NpubFailureRecord::new(now_ms));
+        entry.last_observed_skew_ms = Some(skew_ms);
+
+        if map.len() > self.max_entries {
+            evict_oldest(&mut map, self.max_entries);
+        }
+    }
+
+    /// Reset streak/cooldown after a successful B6 advert refresh.
+    pub(super) fn reset_streak_after_refresh(&self, npub: &str) {
+        let mut map = self.inner.lock().expect("failure-state mutex poisoned");
+        if let Some(entry) = map.get_mut(npub) {
+            entry.consecutive_failures = 0;
+            entry.cooldown_until_ms = None;
+        }
+    }
+
+    /// Return cooldown_until_ms if the peer is currently in extended
+    /// cooldown.
+    pub(super) fn cooldown_until(&self, npub: &str, now_ms: u64) -> Option<u64> {
+        let map = self.inner.lock().expect("failure-state mutex poisoned");
+        map.get(npub)
+            .and_then(|e| e.cooldown_until_ms)
+            .filter(|&t| t > now_ms)
+    }
+
+    /// Snapshot for `show_peers` rendering (B3).
+    pub(super) fn snapshot(&self) -> Vec<(String, NpubFailureRecord)> {
+        let map = self.inner.lock().expect("failure-state mutex poisoned");
+        map.iter()
+            .map(|(npub, rec)| (npub.clone(), rec.clone()))
+            .collect()
+    }
+}
+
+fn evict_oldest(map: &mut HashMap<String, NpubFailureRecord>, target: usize) {
+    if map.len() <= target {
+        return;
+    }
+    let overflow = map.len() - target;
+    let mut entries: Vec<(String, u64)> = map
+        .iter()
+        .map(|(k, v)| (k.clone(), v.last_failure_at_ms))
+        .collect();
+    entries.sort_by_key(|(_, t)| *t);
+    for (k, _) in entries.into_iter().take(overflow) {
+        map.remove(&k);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn fs() -> FailureState {
+        // threshold=3, cooldown=10s, warn-interval=5s, cap=8
+        FailureState::new(3, 10, 5, 8)
+    }
+
+    #[test]
+    fn first_failure_warns_and_no_cooldown() {
+        let s = fs();
+        let d = s.record_failure("npub1a", 1000);
+        assert_eq!(d.consecutive_failures, 1);
+        assert!(d.should_warn);
+        assert!(d.cooldown_until_ms.is_none());
+        assert!(!d.crossed_threshold);
+    }
+
+    #[test]
+    fn warn_suppressed_inside_window_then_unsuppressed_after() {
+        let s = fs();
+        let d1 = s.record_failure("npub1a", 1000);
+        let d2 = s.record_failure("npub1a", 1500);
+        assert!(d1.should_warn);
+        assert!(
+            !d2.should_warn,
+            "second failure inside 5s window must DEBUG"
+        );
+        // 5s warn-interval = 5000 ms; bump beyond.
+        let d3 = s.record_failure("npub1a", 1000 + 5_500);
+        assert!(d3.should_warn, "after window, must WARN again");
+    }
+
+    #[test]
+    fn streak_threshold_triggers_cooldown_and_signals_crossing() {
+        let s = fs();
+        let _ = s.record_failure("npub1a", 1000);
+        let _ = s.record_failure("npub1a", 1100);
+        let d3 = s.record_failure("npub1a", 1200);
+        assert_eq!(d3.consecutive_failures, 3);
+        assert!(d3.crossed_threshold);
+        assert_eq!(d3.cooldown_until_ms, Some(1200 + 10_000));
+        // Subsequent failure does NOT re-fire crossed_threshold.
+        let d4 = s.record_failure("npub1a", 1300);
+        assert!(!d4.crossed_threshold);
+        assert!(d4.cooldown_until_ms.is_some());
+    }
+
+    #[test]
+    fn record_success_clears_streak() {
+        let s = fs();
+        for t in [1000u64, 1100, 1200, 1300] {
+            let _ = s.record_failure("npub1a", t);
+        }
+        s.record_success("npub1a", 2000);
+        let d = s.record_failure("npub1a", 3000);
+        assert_eq!(d.consecutive_failures, 1, "streak reset after success");
+        assert!(!d.crossed_threshold);
+    }
+
+    #[test]
+    fn cooldown_until_returns_only_active_cooldowns() {
+        let s = fs();
+        for t in [1000u64, 1100, 1200] {
+            let _ = s.record_failure("npub1a", t);
+        }
+        // Mid-cooldown
+        assert!(s.cooldown_until("npub1a", 5_000).is_some());
+        // Past cooldown
+        assert!(s.cooldown_until("npub1a", 1200 + 10_001).is_none());
+    }
+
+    #[test]
+    fn note_observed_skew_creates_entry_for_healthy_peer() {
+        let s = fs();
+        s.note_observed_skew("npub1healthy", 250, 1000);
+        let snap = s.snapshot();
+        assert_eq!(snap.len(), 1);
+        let (npub, rec) = &snap[0];
+        assert_eq!(npub, "npub1healthy");
+        assert_eq!(rec.last_observed_skew_ms, Some(250));
+        assert_eq!(rec.consecutive_failures, 0);
+    }
+
+    #[test]
+    fn size_cap_evicts_oldest_by_last_failure_at() {
+        let s = fs(); // cap = 8
+        for i in 0..10 {
+            let npub = format!("npub1{i}");
+            let _ = s.record_failure(&npub, 1000 + i as u64);
+        }
+        let snap = s.snapshot();
+        assert!(snap.len() <= 8, "cap not enforced: {}", snap.len());
+        // Oldest two (npub10, npub11) should be evicted; newer kept.
+        let names: std::collections::HashSet<_> = snap.iter().map(|(n, _)| n.clone()).collect();
+        assert!(!names.contains("npub10"));
+        assert!(names.contains("npub19"));
+    }
+}

src/discovery/nostr/mod.rs

@@ -1,3 +1,4 @@
+mod failure_state;
 mod runtime;
 mod signal;
 mod stun;
@@ -10,7 +11,8 @@ mod tests;
 pub use runtime::NostrDiscovery;
 pub use types::{
     ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent,
-    CachedOverlayAdvert, OverlayAdvert, OverlayEndpointAdvert, OverlayTransportKind,
-    PROTOCOL_VERSION, PUNCH_ACK_MAGIC, PUNCH_MAGIC, PunchHint, PunchPacket, PunchPacketKind,
-    SIGNAL_KIND, TraversalAddress, TraversalAnswer, TraversalOffer,
+    CachedOverlayAdvert, NostrFailureDecision, NostrPeerFailureView, NostrRefetchOutcome,
+    OverlayAdvert, OverlayEndpointAdvert, OverlayTransportKind, PROTOCOL_VERSION, PUNCH_ACK_MAGIC,
+    PUNCH_MAGIC, PunchHint, PunchPacket, PunchPacketKind, SIGNAL_KIND, TraversalAddress,
+    TraversalAnswer, TraversalOffer,
 };

src/discovery/nostr/runtime.rs

@@ -14,16 +14,19 @@ use tokio::sync::{Mutex, RwLock, Semaphore, mpsc, oneshot};
 use tokio::task::JoinHandle;
 use tracing::{debug, trace, warn};
 
+use super::failure_state::FailureState;
 use super::signal::{
-    SignalEnvelope, build_signal_event, create_traversal_answer, create_traversal_offer,
-    unwrap_signal_event, validate_offer_freshness, validate_traversal_answer_for_offer,
+    FreshnessOutcome, SignalEnvelope, build_signal_event, create_traversal_answer,
+    create_traversal_offer, estimate_clock_skew, unwrap_signal_event, validate_offer_freshness,
+    validate_traversal_answer_for_offer,
 };
 use super::stun::observe_traversal_addresses;
 use super::traversal::{nonce, now_ms, planned_remote_endpoints, run_punch_attempt};
 use super::types::{
     ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent,
-    CachedOverlayAdvert, OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION, PunchHint,
-    SIGNAL_KIND, TraversalAnswer, TraversalOffer,
+    CachedOverlayAdvert, NostrFailureDecision, NostrPeerFailureView, NostrRefetchOutcome,
+    OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION, PunchHint, SIGNAL_KIND,
+    TraversalAnswer, TraversalOffer,
 };
 use crate::config::{NostrDiscoveryConfig, PeerConfig};
 use crate::discovery::EstablishedTraversal;
@@ -70,6 +73,7 @@ pub struct NostrDiscovery {
     event_rx: Mutex<mpsc::UnboundedReceiver<BootstrapEvent>>,
     notify_task: Mutex<Option<JoinHandle<()>>>,
     advertise_task: Mutex<Option<JoinHandle<()>>>,
+    failure_state: FailureState,
 }
 
 impl NostrDiscovery {
@@ -104,6 +108,13 @@ impl NostrDiscovery {
         let (event_tx, event_rx) = mpsc::unbounded_channel();
         let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers));
 
+        let failure_state = FailureState::new(
+            config.failure_streak_threshold,
+            config.extended_cooldown_secs,
+            config.warn_log_interval_secs,
+            config.failure_state_max_entries,
+        );
+
         let runtime = Arc::new(Self {
             client,
             keys,
@@ -121,6 +132,7 @@ impl NostrDiscovery {
             event_rx: Mutex::new(event_rx),
             notify_task: Mutex::new(None),
             advertise_task: Mutex::new(None),
+            failure_state,
         });
 
         runtime.subscribe().await?;
@@ -154,6 +166,121 @@ impl NostrDiscovery {
         });
     }
 
+    /// Record a NAT-traversal failure for `npub`, returning the
+    /// resulting decision (WARN suppression + extended cooldown +
+    /// threshold-crossing flag for the B6 re-fetch).
+    pub fn record_traversal_failure(&self, npub: &str, now_ms: u64) -> NostrFailureDecision {
+        let d = self.failure_state.record_failure(npub, now_ms);
+        NostrFailureDecision {
+            consecutive_failures: d.consecutive_failures,
+            should_warn: d.should_warn,
+            cooldown_until_ms: d.cooldown_until_ms,
+            crossed_threshold: d.crossed_threshold,
+        }
+    }
+
+    /// Record a successful traversal — clears the streak/cooldown.
+    pub fn record_traversal_success(&self, npub: &str, now_ms: u64) {
+        self.failure_state.record_success(npub, now_ms);
+    }
+
+    /// Cooldown wall-clock ms if the peer is currently suppressed,
+    /// else None. Used by the open-discovery sweep to skip enqueue.
+    pub fn cooldown_until(&self, npub: &str, now_ms: u64) -> Option<u64> {
+        self.failure_state.cooldown_until(npub, now_ms)
+    }
+
+    /// Snapshot of per-npub failure state for `show_peers` rendering.
+    pub fn failure_state_snapshot(&self) -> Vec<NostrPeerFailureView> {
+        self.failure_state
+            .snapshot()
+            .into_iter()
+            .map(|(npub, rec)| NostrPeerFailureView {
+                npub,
+                consecutive_failures: rec.consecutive_failures,
+                cooldown_until_ms: rec.cooldown_until_ms,
+                last_observed_skew_ms: rec.last_observed_skew_ms,
+            })
+            .collect()
+    }
+
+    /// Stale-advert re-check (B6). Called by lifecycle on the
+    /// streak-threshold transition. Actively re-queries the peer's
+    /// Kind 37195 advert from `advert_relays`; evicts the cache entry
+    /// if absent, refreshes if newer than the cached `created_at`,
+    /// otherwise leaves the cache untouched.
+    pub async fn refetch_advert_for_stale_check(&self, peer_npub: &str) -> NostrRefetchOutcome {
+        let target_pubkey = match PublicKey::parse(peer_npub) {
+            Ok(p) => p,
+            Err(_) => return NostrRefetchOutcome::Skipped,
+        };
+        if self.config.advert_relays.is_empty() {
+            return NostrRefetchOutcome::Skipped;
+        }
+        let cached_created_at = self
+            .advert_cache
+            .read()
+            .await
+            .get(peer_npub)
+            .map(|c| c.created_at);
+
+        let events = match self
+            .client
+            .fetch_events_from(
+                self.config.advert_relays.clone(),
+                Filter::new()
+                    .author(target_pubkey)
+                    .kind(Kind::Custom(ADVERT_KIND))
+                    .identifier(ADVERT_IDENTIFIER),
+                Duration::from_secs(2),
+            )
+            .await
+        {
+            Ok(e) => e,
+            Err(_) => return NostrRefetchOutcome::Skipped,
+        };
+
+        let mut newest: Option<(u64, &Event)> = None;
+        for ev in events.iter() {
+            let ts = ev.created_at.as_secs();
+            match newest {
+                Some((cur, _)) if ts <= cur => {}
+                _ => newest = Some((ts, ev)),
+            }
+        }
+
+        let Some((relay_created_at, ev)) = newest else {
+            // Absent on relays. Evict any stale cache entry.
+            self.advert_cache.write().await.remove(peer_npub);
+            self.failure_state.reset_streak_after_refresh(peer_npub);
+            return NostrRefetchOutcome::Evicted;
+        };
+
+        match cached_created_at {
+            Some(cached) if relay_created_at <= cached => NostrRefetchOutcome::SameAdvert,
+            _ => {
+                let Some(valid_until_ms) = self.event_valid_until_ms(ev) else {
+                    return NostrRefetchOutcome::Skipped;
+                };
+                let Ok(advert) = Self::parse_overlay_advert_event(ev, &self.config.app) else {
+                    return NostrRefetchOutcome::Skipped;
+                };
+                let updated = CachedOverlayAdvert {
+                    author_npub: peer_npub.to_string(),
+                    advert,
+                    created_at: relay_created_at,
+                    valid_until_ms,
+                };
+                self.advert_cache
+                    .write()
+                    .await
+                    .insert(peer_npub.to_string(), updated);
+                self.failure_state.reset_streak_after_refresh(peer_npub);
+                NostrRefetchOutcome::Refreshed
+            }
+        }
+    }
+
     pub async fn drain_events(&self) -> Vec<BootstrapEvent> {
         let mut out = Vec::new();
         let mut rx = self.event_rx.lock().await;
@@ -591,6 +718,7 @@ impl NostrDiscovery {
             }
         };
 
+        let answer_received_at = now_ms();
         debug!(
             peer = %peer_short,
             session = %short_id(&offer.session_id),
@@ -599,14 +727,47 @@ impl NostrDiscovery {
             local = answer.payload.local_addresses.len(),
             "traversal: answer received"
         );
-        validate_traversal_answer_for_offer(
+        if let Some(observed_skew_ms) =
+            estimate_clock_skew(&offer, &answer.payload, answer_received_at)
+        {
+            self.failure_state.note_observed_skew(
+                &peer_config.npub,
+                observed_skew_ms,
+                answer_received_at,
+            );
+            let abs_skew = observed_skew_ms.unsigned_abs();
+            // 30s threshold: well below the 60s SKEW_TOLERANCE wall but loud
+            // enough to surface a real clock problem on either side.
+            if abs_skew >= 30_000 {
+                debug!(
+                    peer = %peer_short,
+                    session = %short_id(&offer.session_id),
+                    skew_ms = observed_skew_ms,
+                    "traversal: significant peer clock skew observed"
+                );
+            } else {
+                trace!(
+                    peer = %peer_short,
+                    skew_ms = observed_skew_ms,
+                    "traversal: peer clock skew within nominal range"
+                );
+            }
+        }
+        let outcome = validate_traversal_answer_for_offer(
             &offer,
             &answer.payload,
-            now_ms(),
+            answer_received_at,
             self.config.signal_ttl_secs * 1000,
             &answer.sender_npub,
             &self.npub,
         )?;
+        if outcome == FreshnessOutcome::FreshWithinSkewTolerance {
+            debug!(
+                peer = %peer_short,
+                session = %short_id(&offer.session_id),
+                "traversal: answer accepted within clock-skew tolerance"
+            );
+        }
         if !answer.payload.accepted {
             return Err(BootstrapError::Protocol(
                 answer
@@ -643,6 +804,9 @@ impl NostrDiscovery {
             .publish_delete(&relays, [offer_event.id, answer.event_id])
             .await;
 
+        self.failure_state
+            .record_success(&peer_config.npub, now_ms());
+
         Ok(
             EstablishedTraversal::new(session_id, peer_config.npub, remote_addr, base_socket)
                 .with_transport_name("nostr-nat"),
@@ -656,6 +820,7 @@ impl NostrDiscovery {
         sender_npub: String,
     ) -> Result<(), BootstrapError> {
         let peer_short = short_npub(&sender_npub);
+        let offer_received_at = now_ms();
         debug!(
             peer = %peer_short,
             session = %short_id(&offer.session_id),
@@ -663,13 +828,22 @@ impl NostrDiscovery {
             local = offer.local_addresses.len(),
             "traversal: offer received"
         );
-        validate_offer_freshness(
+        let outcome = validate_offer_freshness(
             &offer,
-            now_ms(),
+            offer_received_at,
             self.config.signal_ttl_secs * 1000,
             &sender_npub,
             &self.npub,
         )?;
+        if outcome == FreshnessOutcome::FreshWithinSkewTolerance {
+            debug!(
+                peer = %peer_short,
+                session = %short_id(&offer.session_id),
+                offer_issued_at = offer.issued_at,
+                offer_received_at = offer_received_at,
+                "traversal: offer accepted within clock-skew tolerance"
+            );
+        }
         self.mark_session_seen(&offer.session_id).await?;
 
         let base_socket = std::net::UdpSocket::bind(("0.0.0.0", 0))?;
@@ -703,6 +877,7 @@ impl NostrDiscovery {
             stun_server,
             accepted.then(|| self.punch_hint()),
             (!accepted).then_some("no-usable-addresses".to_string()),
+            Some(offer_received_at),
         );
         let relays = self.preferred_signal_relays(sender, None).await?;
         let answer_event = self.send_signal(&relays, sender, &answer).await?;
@@ -1118,6 +1293,12 @@ impl NostrDiscovery {
         let config = NostrDiscoveryConfig::default();
         let offer_slots = Arc::new(Semaphore::new(config.max_concurrent_incoming_offers));
         let (event_tx, event_rx) = mpsc::unbounded_channel();
+        let failure_state = FailureState::new(
+            config.failure_streak_threshold,
+            config.extended_cooldown_secs,
+            config.warn_log_interval_secs,
+            config.failure_state_max_entries,
+        );
         Self {
             client,
             keys,
@@ -1135,6 +1316,7 @@ impl NostrDiscovery {
             event_rx: Mutex::new(event_rx),
             notify_task: Mutex::new(None),
             advertise_task: Mutex::new(None),
+            failure_state,
         }
     }
 

src/discovery/nostr/signal.rs

@@ -6,6 +6,13 @@ use nostr::prelude::{
 
 use super::types::{BootstrapError, PunchHint, SIGNAL_KIND, TraversalAnswer, TraversalOffer};
 
+/// Wall-clock skew tolerance applied to offer/answer freshness checks, in
+/// milliseconds. Constant rather than configurable because loosening this
+/// past ~minutes erodes the freshness guarantee that backstops session-id
+/// replay protection. Tightening it below the size of a typical un-NTP'd
+/// drift defeats the purpose. 60s sits comfortably between those.
+pub(super) const FRESHNESS_SKEW_TOLERANCE_MS: u64 = 60_000;
+
 pub(super) struct SignalEnvelope<T> {
     pub(super) payload: T,
     pub(super) event_id: EventId,
@@ -75,23 +82,34 @@ pub(super) async fn unwrap_signal_event(
     })
 }
 
+/// Result of a freshness check. `Fresh` means the offer/answer is within the
+/// strict TTL window; `FreshWithinSkewTolerance` means it was only accepted
+/// after applying `FRESHNESS_SKEW_TOLERANCE_MS` grace, which is a useful
+/// signal for operators to know clock skew is in play.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum FreshnessOutcome {
+    Fresh,
+    FreshWithinSkewTolerance,
+}
+
 pub(super) fn validate_offer_freshness(
     offer: &TraversalOffer,
     now: u64,
     signal_ttl_ms: u64,
     actual_sender_npub: &str,
     local_npub: &str,
-) -> Result<(), BootstrapError> {
+) -> Result<FreshnessOutcome, BootstrapError> {
     if offer.message_type != "offer" {
         return Err(BootstrapError::Protocol("invalid-offer".to_string()));
     }
-    if offer.expires_at <= now || now.saturating_sub(offer.issued_at) > signal_ttl_ms {
-        return Err(BootstrapError::Protocol("expired-offer".to_string()));
-    }
+    let outcome = match check_freshness(offer.issued_at, offer.expires_at, now, signal_ttl_ms) {
+        Some(o) => o,
+        None => return Err(BootstrapError::Protocol("expired-offer".to_string())),
+    };
     if offer.sender_npub != actual_sender_npub || offer.recipient_npub != local_npub {
         return Err(BootstrapError::Protocol("identity-mismatch".to_string()));
     }
-    Ok(())
+    Ok(outcome)
 }
 
 #[allow(clippy::too_many_arguments)]
@@ -135,6 +153,7 @@ pub(super) fn create_traversal_answer(
     stun_server: Option<String>,
     punch: Option<PunchHint>,
     reason: Option<String>,
+    offer_received_at: Option<u64>,
 ) -> TraversalAnswer {
     TraversalAnswer {
         message_type: "answer".to_string(),
@@ -151,6 +170,7 @@ pub(super) fn create_traversal_answer(
         stun_server,
         punch,
         reason,
+        offer_received_at,
     }
 }
 
@@ -161,16 +181,20 @@ pub(super) fn validate_traversal_answer_for_offer(
     signal_ttl_ms: u64,
     actual_sender_npub: &str,
     local_npub: &str,
-) -> Result<(), BootstrapError> {
+) -> Result<FreshnessOutcome, BootstrapError> {
     if answer.message_type != "answer" {
         return Err(BootstrapError::Protocol("invalid-answer".to_string()));
     }
-    if offer.expires_at <= now
-        || answer.expires_at <= now
-        || now.saturating_sub(answer.issued_at) > signal_ttl_ms
+    let offer_outcome = match check_freshness(offer.issued_at, offer.expires_at, now, signal_ttl_ms)
     {
-        return Err(BootstrapError::Protocol("expired-answer".to_string()));
-    }
+        Some(o) => o,
+        None => return Err(BootstrapError::Protocol("expired-answer".to_string())),
+    };
+    let answer_outcome =
+        match check_freshness(answer.issued_at, answer.expires_at, now, signal_ttl_ms) {
+            Some(o) => o,
+            None => return Err(BootstrapError::Protocol("expired-answer".to_string())),
+        };
     if offer.session_id != answer.session_id || answer.in_reply_to != offer.nonce {
         return Err(BootstrapError::Protocol("session-mismatch".to_string()));
     }
@@ -189,5 +213,58 @@ pub(super) fn validate_traversal_answer_for_offer(
             "missing-rejection-reason".to_string(),
         ));
     }
-    Ok(())
+    // Surface skew if either side was tolerated. The strict-Fresh case wins
+    // when both are strict; otherwise tolerance applied somewhere.
+    Ok(
+        if offer_outcome == FreshnessOutcome::Fresh && answer_outcome == FreshnessOutcome::Fresh {
+            FreshnessOutcome::Fresh
+        } else {
+            FreshnessOutcome::FreshWithinSkewTolerance
+        },
+    )
+}
+
+/// NTP-style clock-skew estimate from a completed offer/answer round-trip.
+/// Returns the responder's apparent offset relative to the initiator in
+/// milliseconds (positive = responder clock ahead). Requires the responder
+/// to have populated `answer.offer_received_at`; older responders won't, in
+/// which case this returns `None`.
+///
+/// Symmetric one-way-delay assumption (the standard NTP offset formula):
+///   offset = ((T2 - T1) + (T3 - T4)) / 2
+/// where T1 = offer.issued_at, T2 = answer.offer_received_at,
+///       T3 = answer.issued_at, T4 = answer_received_at.
+pub(super) fn estimate_clock_skew(
+    offer: &TraversalOffer,
+    answer: &TraversalAnswer,
+    answer_received_at: u64,
+) -> Option<i64> {
+    let t1 = offer.issued_at as i64;
+    let t2 = answer.offer_received_at? as i64;
+    let t3 = answer.issued_at as i64;
+    let t4 = answer_received_at as i64;
+    Some(((t2 - t1) + (t3 - t4)) / 2)
+}
+
+/// Returns Some(outcome) if the (issued_at, expires_at) pair is acceptable
+/// against `now` under the configured TTL plus `FRESHNESS_SKEW_TOLERANCE_MS`
+/// of clock-skew grace on each side. Returns None if the message is
+/// genuinely outside the tolerated window.
+fn check_freshness(
+    issued_at: u64,
+    expires_at: u64,
+    now: u64,
+    signal_ttl_ms: u64,
+) -> Option<FreshnessOutcome> {
+    let strict_ok = expires_at > now && now.saturating_sub(issued_at) <= signal_ttl_ms;
+    if strict_ok {
+        return Some(FreshnessOutcome::Fresh);
+    }
+    let tolerated_ok = expires_at.saturating_add(FRESHNESS_SKEW_TOLERANCE_MS) > now
+        && now.saturating_sub(issued_at) <= signal_ttl_ms + FRESHNESS_SKEW_TOLERANCE_MS;
+    if tolerated_ok {
+        Some(FreshnessOutcome::FreshWithinSkewTolerance)
+    } else {
+        None
+    }
 }

src/discovery/nostr/tests.rs

@@ -2,8 +2,8 @@ use nostr::prelude::{EventBuilder, Kind, Tag, Timestamp};
 
 use super::runtime::NostrDiscovery;
 use super::signal::{
-    build_signal_event, create_traversal_answer, create_traversal_offer, validate_offer_freshness,
-    validate_traversal_answer_for_offer,
+    FreshnessOutcome, build_signal_event, create_traversal_answer, create_traversal_offer,
+    estimate_clock_skew, validate_offer_freshness, validate_traversal_answer_for_offer,
 };
 use super::stun::{parse_stun_binding_success, parse_stun_url};
 use super::traversal::{
@@ -240,6 +240,7 @@ fn validates_offer_answer_pair() {
             duration_ms: 10_000,
         }),
         None,
+        Some(1_700_000_000_400),
     );
 
     assert!(
@@ -311,6 +312,7 @@ fn rejects_answer_with_mismatched_actual_sender() {
             duration_ms: 10_000,
         }),
         None,
+        Some(1_700_000_000_400),
     );
 
     let result = validate_traversal_answer_for_offer(
@@ -386,6 +388,172 @@ fn planned_remote_endpoints_include_private_and_reflexive_paths() {
     assert!(endpoints.contains(&"198.51.100.20:63000".parse().unwrap()));
 }
 
+/// B4: strict-fresh path returns Fresh; the offer is well within TTL and
+/// not expired.
+#[test]
+fn freshness_strict_returns_fresh_outcome() {
+    let offer = create_traversal_offer(
+        "sess-1".to_string(),
+        1_700_000_000_000,
+        60_000,
+        "offer-1".to_string(),
+        "npub1client".to_string(),
+        "npub1server".to_string(),
+        Some(addr("203.0.113.10", 62000)),
+        vec![addr("192.168.1.10", 62000)],
+        Some("stun:example.org:3478".to_string()),
+    );
+
+    let result = validate_offer_freshness(
+        &offer,
+        1_700_000_000_500,
+        60_000,
+        "npub1client",
+        "npub1server",
+    )
+    .expect("strict-fresh offer should validate");
+    assert_eq!(result, FreshnessOutcome::Fresh);
+}
+
+/// B4: an offer whose `expires_at` has already passed by < SKEW_TOL is
+/// accepted but flagged FreshWithinSkewTolerance — emulates the case where
+/// the responder's clock is ahead of the initiator's.
+#[test]
+fn freshness_responder_clock_ahead_within_tolerance_is_tolerated() {
+    let offer = create_traversal_offer(
+        "sess-1".to_string(),
+        1_700_000_000_000,
+        60_000, // expires_at = 1_700_000_060_000
+        "offer-1".to_string(),
+        "npub1client".to_string(),
+        "npub1server".to_string(),
+        Some(addr("203.0.113.10", 62000)),
+        vec![addr("192.168.1.10", 62000)],
+        None,
+    );
+
+    // now 90s past issued_at — 30s past strict expiry, but inside the 60s
+    // SKEW_TOL grace.
+    let result = validate_offer_freshness(
+        &offer,
+        1_700_000_090_000,
+        60_000,
+        "npub1client",
+        "npub1server",
+    )
+    .expect("offer just past strict expiry should be tolerated");
+    assert_eq!(result, FreshnessOutcome::FreshWithinSkewTolerance);
+}
+
+/// B4: an offer beyond TTL + SKEW_TOL is rejected as expired.
+#[test]
+fn freshness_responder_clock_far_ahead_is_rejected() {
+    let offer = create_traversal_offer(
+        "sess-1".to_string(),
+        1_700_000_000_000,
+        60_000,
+        "offer-1".to_string(),
+        "npub1client".to_string(),
+        "npub1server".to_string(),
+        Some(addr("203.0.113.10", 62000)),
+        vec![addr("192.168.1.10", 62000)],
+        None,
+    );
+
+    // 130s past issued_at: 70s past strict expiry, 10s past tolerated expiry.
+    let err = validate_offer_freshness(
+        &offer,
+        1_700_000_130_000,
+        60_000,
+        "npub1client",
+        "npub1server",
+    )
+    .expect_err("offer past tolerated expiry should be rejected");
+    assert!(err.to_string().contains("expired-offer"), "{}", err);
+}
+
+/// B5a: the NTP-style skew estimator returns the responder's apparent
+/// clock offset relative to the initiator. Symmetric one-way delays of
+/// 50ms each plus a +500ms responder skew should yield ≈+500ms.
+#[test]
+fn estimate_clock_skew_matches_responder_offset() {
+    // T1 (initiator sent)
+    let offer = create_traversal_offer(
+        "sess-1".to_string(),
+        1_700_000_000_000,
+        60_000,
+        "offer-1".to_string(),
+        "npub1client".to_string(),
+        "npub1server".to_string(),
+        None,
+        vec![addr("192.168.1.10", 62000)],
+        None,
+    );
+    // Wire takes 50ms, responder clock is +500ms ahead, so:
+    //   T2 = 1_700_000_000_000 + 50 + 500 = 1_700_000_000_550
+    //   T3 = 1_700_000_000_550 (no processing time for this synthetic case)
+    //   T4 = T1 + 50 + (T3 - T2 + 500_skew_corrected) + 50 wire return
+    //      For simplicity: T4 = T1 + 100ms wire + 0 responder processing
+    //                       = 1_700_000_000_100 (initiator wall clock)
+    let answer = create_traversal_answer(
+        "sess-1".to_string(),
+        1_700_000_000_550, // T3
+        60_000,
+        "answer-1".to_string(),
+        "npub1server".to_string(),
+        "npub1client".to_string(),
+        "offer-1".to_string(),
+        true,
+        Some(addr("198.51.100.20", 63000)),
+        vec![],
+        None,
+        None,
+        None,
+        Some(1_700_000_000_550), // T2
+    );
+    let answer_received_at = 1_700_000_000_100; // T4
+
+    let skew = estimate_clock_skew(&offer, &answer, answer_received_at)
+        .expect("offer_received_at populated -> Some");
+    // ((550 - 0) + (550 - 100)) / 2 = (550 + 450) / 2 = 500
+    assert_eq!(skew, 500);
+}
+
+/// B5a: backward-compat — when the responder did not populate
+/// `offer_received_at` (older daemon), skew estimation returns None
+/// and callers should silently skip logging it.
+#[test]
+fn estimate_clock_skew_returns_none_without_responder_timestamp() {
+    let offer = create_traversal_offer(
+        "sess-1".to_string(),
+        1_700_000_000_000,
+        60_000,
+        "offer-1".to_string(),
+        "npub1client".to_string(),
+        "npub1server".to_string(),
+        None,
+        vec![],
+        None,
+    );
+    let answer = create_traversal_answer(
+        "sess-1".to_string(),
+        1_700_000_000_500,
+        60_000,
+        "answer-1".to_string(),
+        "npub1server".to_string(),
+        "npub1client".to_string(),
+        "offer-1".to_string(),
+        true,
+        Some(addr("198.51.100.20", 63000)),
+        vec![],
+        None,
+        None,
+        None,
+        None, // older responder
+    );
+    assert!(estimate_clock_skew(&offer, &answer, 1_700_000_000_900).is_none());
+}
+
 #[tokio::test]
 async fn signal_events_use_current_timestamps() {
     let sender = nostr::Keys::generate();

src/discovery/nostr/types.rs

@@ -164,6 +164,16 @@ pub struct TraversalAnswer {
     pub stun_server: Option<String>,
     pub punch: Option<PunchHint>,
     pub reason: Option<String>,
+    /// Responder's local wall-clock (Unix ms) at the moment it received the
+    /// offer. Optional / non-breaking: the initiator uses this to derive an
+    /// NTP-style clock-skew estimate against the offer's `issued_at`. Older
+    /// responders that don't fill this in still produce valid answers.
+    #[serde(
+        rename = "offerReceivedAt",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub offer_received_at: Option<u64>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -178,3 +188,35 @@ pub struct PunchPacket {
     pub sequence: u32,
     pub session_hash: [u8; 16],
 }
+
+/// Outcome of `NostrDiscovery::record_traversal_failure`.
+#[derive(Debug, Clone, Copy)]
+pub struct NostrFailureDecision {
+    pub consecutive_failures: u32,
+    /// True iff the lifecycle should log at WARN; false → DEBUG (rate-limit).
+    pub should_warn: bool,
+    /// Wall-clock ms before which retries should not fire, if cooldown
+    /// is in effect.
+    pub cooldown_until_ms: Option<u64>,
+    /// True only on the streak-threshold-crossing transition. Lifecycle
+    /// should run a one-shot stale-advert re-check (B6) when set.
+    pub crossed_threshold: bool,
+}
+
+/// Snapshot row for `show_peers` rendering of per-npub Nostr-traversal state.
+#[derive(Debug, Clone)]
+pub struct NostrPeerFailureView {
+    pub npub: String,
+    pub consecutive_failures: u32,
+    pub cooldown_until_ms: Option<u64>,
+    pub last_observed_skew_ms: Option<i64>,
+}
+
+/// Outcome of `NostrDiscovery::refetch_advert_for_stale_check` (B6).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum NostrRefetchOutcome {
+    Evicted,
+    Refreshed,
+    SameAdvert,
+    Skipped,
+}

src/node/lifecycle.rs

@@ -407,7 +407,56 @@ impl Node {
                     peer_config,
                     reason,
                 } => {
-                    warn!(npub = %peer_config.npub, error = %reason, "NAT traversal failed");
+                    let now_ms = Self::now_ms();
+                    let decision = bootstrap.record_traversal_failure(&peer_config.npub, now_ms);
+                    if decision.should_warn {
+                        warn!(
+                            npub = %peer_config.npub,
+                            error = %reason,
+                            consecutive_failures = decision.consecutive_failures,
+                            cooldown_secs = decision
+                                .cooldown_until_ms
+                                .map(|t| t.saturating_sub(now_ms) / 1000),
+                            "NAT traversal failed"
+                        );
+                    } else {
+                        debug!(
+                            npub = %peer_config.npub,
+                            error = %reason,
+                            consecutive_failures = decision.consecutive_failures,
+                            "NAT traversal failed (suppressed by warn-rate-limit)"
+                        );
+                    }
+
+                    // B6: stale-advert eviction on the streak-threshold
+                    // crossing. Fire-and-forget; the outcome is logged so
+                    // operators can see when peers get cleaned up.
+                    if decision.crossed_threshold {
+                        let bootstrap = bootstrap.clone();
+                        let npub = peer_config.npub.clone();
+                        tokio::spawn(async move {
+                            let outcome = bootstrap.refetch_advert_for_stale_check(&npub).await;
+                            match outcome {
+                                crate::discovery::nostr::NostrRefetchOutcome::Evicted => info!(
+                                    npub = %npub,
+                                    "stale-advert sweep: peer evicted from advert cache"
+                                ),
+                                crate::discovery::nostr::NostrRefetchOutcome::Refreshed => info!(
+                                    npub = %npub,
+                                    "stale-advert sweep: peer republished, cache refreshed and streak reset"
+                                ),
+                                crate::discovery::nostr::NostrRefetchOutcome::SameAdvert => debug!(
+                                    npub = %npub,
+                                    "stale-advert sweep: advert unchanged, cooldown stands"
+                                ),
+                                crate::discovery::nostr::NostrRefetchOutcome::Skipped => debug!(
+                                    npub = %npub,
+                                    "stale-advert sweep: skipped (relay error or no advert_relays)"
+                                ),
+                            }
+                        });
+                    }
+
                     let peer_identity = match PeerIdentity::from_npub(&peer_config.npub) {
                         Ok(identity) => identity,
                         Err(_) => continue,
@@ -421,7 +470,16 @@ impl Node {
                         continue;
                     }
 
-                    self.schedule_retry(*peer_identity.node_addr(), Self::now_ms());
+                    let node_addr = *peer_identity.node_addr();
+                    self.schedule_retry(node_addr, now_ms);
+                    if let Some(cooldown_until_ms) = decision.cooldown_until_ms
+                        && let Some(state) = self.retry_pending.get_mut(&node_addr)
+                    {
+                        // Push the next retry past the cooldown so the
+                        // open-discovery sweep doesn't re-enqueue and the
+                        // per-attempt backoff doesn't fire sooner.
+                        state.retry_after_ms = state.retry_after_ms.max(cooldown_until_ms);
+                    }
                 }
             }
         }
@@ -1204,6 +1262,7 @@ impl Node {
         let mut skipped_connecting = 0usize;
         let mut skipped_no_endpoints = 0usize;
         let mut skipped_invalid_npub = 0usize;
+        let mut skipped_cooldown = 0usize;
 
         for (npub, endpoints, created_at_secs) in candidates {
             if enqueue_budget == 0 {
@@ -1242,6 +1301,10 @@ impl Node {
                 skipped_retry_pending = skipped_retry_pending.saturating_add(1);
                 continue;
             }
+            if bootstrap.cooldown_until(&npub, now_ms).is_some() {
+                skipped_cooldown = skipped_cooldown.saturating_add(1);
+                continue;
+            }
             let connecting = self.connections.values().any(|conn| {
                 conn.expected_identity()
                     .map(|id| id.node_addr() == &node_addr)
@@ -1309,7 +1372,8 @@ impl Node {
             + skipped_retry_pending
             + skipped_connecting
             + skipped_no_endpoints
-            + skipped_invalid_npub;
+            + skipped_invalid_npub
+            + skipped_cooldown;
         let should_summarize = caller == "startup" || enqueued > 0;
         if should_summarize {
             info!(
@@ -1324,6 +1388,7 @@ impl Node {
                 skipped_connecting = skipped_connecting,
                 skipped_no_endpoints = skipped_no_endpoints,
                 skipped_invalid_npub = skipped_invalid_npub,
+                skipped_cooldown = skipped_cooldown,
                 skipped_total = total_skipped,
                 "open-discovery sweep complete"
             );

src/node/mod.rs

@@ -1539,6 +1539,13 @@ impl Node {
         self.peers.values()
     }
 
+    /// Reference to the Nostr discovery handle if discovery is enabled.
+    /// Used by control queries (`show_peers` per-peer Nostr-traversal
+    /// state) to read failure-state without taking shared ownership.
+    pub fn nostr_discovery_handle(&self) -> Option<&crate::discovery::nostr::NostrDiscovery> {
+        self.nostr_discovery.as_deref()
+    }
+
     /// Iterate over all peer node IDs.
     pub fn peer_ids(&self) -> impl Iterator<Item = &NodeAddr> {
         self.peers.keys()