From bcc9c525d35cf6614dbd608f84b82180faed0c43 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Mon, 4 May 2026 12:39:51 +0000 Subject: [PATCH] nostr: per-peer NAT-traversal failure suppression and clock-skew handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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` 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": , "in_cooldown": , "cooldown_until_ms": , "last_observed_skew_ms": } 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 | 25 +++ src/config/node.rs | 43 ++++ src/control/queries.rs | 37 ++++ src/discovery/nostr/failure_state.rs | 311 +++++++++++++++++++++++++++ src/discovery/nostr/mod.rs | 8 +- src/discovery/nostr/runtime.rs | 198 ++++++++++++++++- src/discovery/nostr/signal.rs | 101 +++++++-- src/discovery/nostr/tests.rs | 172 ++++++++++++++- src/discovery/nostr/types.rs | 42 ++++ src/node/lifecycle.rs | 71 +++++- src/node/mod.rs | 7 + 11 files changed, 987 insertions(+), 28 deletions(-) create mode 100644 src/discovery/nostr/failure_state.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e5e04df..b27e449 100644 --- a/CHANGELOG.md +++ b/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:`) instead of just the bare onion hostname. The publisher previously emitted a diff --git a/src/config/node.rs b/src/config/node.rs index 6bfcfd5..9b95748 100644 --- a/src/config/node.rs +++ b/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.*`). diff --git a/src/control/queries.rs b/src/control/queries.rs index 7547192..b7d3575 100644 --- a/src/control/queries.rs +++ b/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 = 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 = 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!({ diff --git a/src/discovery/nostr/failure_state.rs b/src/discovery/nostr/failure_state.rs new file mode 100644 index 0000000..57d2b0e --- /dev/null +++ b/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, + /// When the extended cooldown was applied; while + /// `cooldown_until_ms.is_some_and(|t| t > now)`, retries are + /// suppressed. + pub cooldown_until_ms: Option, + /// 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, +} + +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, + /// 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>, + 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 { + 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, 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")); + } +} diff --git a/src/discovery/nostr/mod.rs b/src/discovery/nostr/mod.rs index c08f6f0..881fb2e 100644 --- a/src/discovery/nostr/mod.rs +++ b/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, }; diff --git a/src/discovery/nostr/runtime.rs b/src/discovery/nostr/runtime.rs index 6dd4844..a08bf88 100644 --- a/src/discovery/nostr/runtime.rs +++ b/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>, notify_task: Mutex>>, advertise_task: Mutex>>, + 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 { + 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 { + 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 { 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, } } diff --git a/src/discovery/nostr/signal.rs b/src/discovery/nostr/signal.rs index bf57053..b3d018b 100644 --- a/src/discovery/nostr/signal.rs +++ b/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 { 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 { 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, punch: Option, reason: Option, + offer_received_at: Option, ) -> 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 { 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 { + 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 { + 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 + } } diff --git a/src/discovery/nostr/tests.rs b/src/discovery/nostr/tests.rs index ec7f77c..b412058 100644 --- a/src/discovery/nostr/tests.rs +++ b/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(); diff --git a/src/discovery/nostr/types.rs b/src/discovery/nostr/types.rs index 592444e..3d76419 100644 --- a/src/discovery/nostr/types.rs +++ b/src/discovery/nostr/types.rs @@ -164,6 +164,16 @@ pub struct TraversalAnswer { pub stun_server: Option, pub punch: Option, pub reason: Option, + /// 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, } #[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, + /// 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, + pub last_observed_skew_ms: Option, +} + +/// Outcome of `NostrDiscovery::refetch_advert_for_stale_check` (B6). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NostrRefetchOutcome { + Evicted, + Refreshed, + SameAdvert, + Skipped, +} diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index b98f7fd..55b546d 100644 --- a/src/node/lifecycle.rs +++ b/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" ); diff --git a/src/node/mod.rs b/src/node/mod.rs index 3d95d47..7744546 100644 --- a/src/node/mod.rs +++ b/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 { self.peers.keys()