diff --git a/src/nostr/advert.rs b/src/nostr/advert.rs new file mode 100644 index 0000000..f3eaa28 --- /dev/null +++ b/src/nostr/advert.rs @@ -0,0 +1,505 @@ +//! Synchronous decision core for the Nostr overlay-advert lifecycle. +//! +//! `AdvertMachine` owns the advert-related state that previously lived +//! directly on `NostrRendezvous` — the peer advert cache, the local +//! advert we publish, and the id of our most recently published advert +//! event — and hosts the *decision* logic for publishing, caching, +//! fetching, and pruning adverts. +//! +//! Following the sans-IO shape used by `failure_state`, every method here +//! is synchronous, performs no network I/O and no `.await`, holds its +//! state behind `std::sync::Mutex`, and takes the current time as an +//! explicit `now_ms: u64` input rather than reading a clock. The async +//! driver on `NostrRendezvous` reads the clock at the call site, invokes +//! these methods, and performs the actual relay I/O (`send_event_to`, +//! `fetch_events_from`, gift-wrap crypto), event signing, NIP-09 deletes, +//! and `Notify` wakeups described by the returned decisions. + +use std::collections::HashMap; +use std::sync::Mutex; + +use nostr::prelude::{Event, EventId}; + +use super::runtime::{NostrRendezvous, endpoint_advert_is_publicly_usable}; +use super::types::{ + ADVERT_IDENTIFIER, ADVERT_VERSION, BootstrapError, CachedOverlayAdvert, OverlayAdvert, + OverlayEndpointAdvert, +}; + +/// What the async driver should do to satisfy a publish request. Returned +/// by [`AdvertMachine::plan_publish`]; the machine performs the pure +/// decision (which advert body, or a delete, or nothing) and the driver +/// executes the corresponding relay I/O. +#[derive(Debug, Clone)] +pub(super) enum PublishPlan { + /// Nothing to publish (advertising disabled with no prior event, no + /// local advert yet, or the advert has no publicly usable endpoints). + Nothing, + /// Advertising is disabled but a prior advert event exists; the driver + /// should emit a NIP-09 delete for `EventId` then call + /// [`AdvertMachine::clear_event_id`]. + Delete(EventId), + /// Publish this fully-prepared advert body. The driver builds the + /// tags/expiration, signs, sends, then records the new event id via + /// [`AdvertMachine::set_event_id`]. + Publish(OverlayAdvert), +} + +pub(super) struct AdvertMachine { + /// Our own npub. Used to avoid logging/attributing self-authored + /// adverts as peer discoveries. + npub: String, + /// Whether this node advertises at all (`config.advertise`). + advertise: bool, + /// Grace-extended max age for a cached advert, in ms + /// (`advert_ttl_secs * 1000 * stale-grace-multiplier`). + advert_max_age_ms: u64, + /// Size cap for the peer advert cache. + cache_max_entries: usize, + /// Peer advert cache keyed by author npub. + cache: Mutex>, + /// The advert body we currently want to publish, if any. + local_advert: Mutex>, + /// Id of our most recently published advert event (for NIP-09 delete + /// on withdrawal). + current_event_id: Mutex>, +} + +impl AdvertMachine { + pub(super) fn new( + npub: String, + advertise: bool, + advert_max_age_ms: u64, + cache_max_entries: usize, + ) -> Self { + Self { + npub, + advertise, + advert_max_age_ms, + cache_max_entries, + cache: Mutex::new(HashMap::new()), + local_advert: Mutex::new(None), + current_event_id: Mutex::new(None), + } + } + + // --- validity (time-injected) -------------------------------------- + + /// Compute the validity horizon of an advert event, or `None` if it is + /// already stale. Thin time-injected wrapper over the pure + /// `compute_advert_valid_until_ms`. + pub(super) fn event_valid_until_ms(&self, event: &Event, now_ms: u64) -> Option { + NostrRendezvous::compute_advert_valid_until_ms(event, self.advert_max_age_ms, now_ms) + } + + // --- cache: prune / observe / fetch -------------------------------- + + /// TTL + size-cap eviction. Drops entries past their validity horizon, + /// then evicts the oldest (by `valid_until_ms`) beyond the cap. + /// + /// Returns `Some((evicted, retained))` when a size-cap eviction + /// occurred so the driver can log it; `None` otherwise. + pub(super) fn prune(&self, now_ms: u64) -> Option<(usize, usize)> { + let mut cache = self.lock_cache(); + cache.retain(|_, entry| entry.valid_until_ms > now_ms); + if cache.len() <= self.cache_max_entries { + return None; + } + + let mut oldest = cache + .iter() + .map(|(npub, entry)| (npub.clone(), entry.valid_until_ms)) + .collect::>(); + oldest.sort_by_key(|(_, ts)| *ts); + let overflow = cache.len().saturating_sub(self.cache_max_entries); + for (npub, _) in oldest.into_iter().take(overflow) { + cache.remove(&npub); + } + Some((overflow, cache.len())) + } + + /// Observe an advert event received on the notify loop. Replaces the + /// cached entry iff its `created_at` is newer-or-equal to the cached + /// one (or none is cached). + /// + /// Returns `true` when the caller should log a "peer cached" line — + /// i.e. the entry was (re)cached *and* it is not our own advert. + pub(super) fn observe_advert( + &self, + author_npub: &str, + advert: OverlayAdvert, + created_at: u64, + valid_until_ms: u64, + ) -> bool { + let mut cache = self.lock_cache(); + let should_replace = cache + .get(author_npub) + .map(|existing| existing.created_at <= created_at) + .unwrap_or(true); + if !should_replace { + return false; + } + let is_peer = author_npub != self.npub; + cache.insert( + author_npub.to_string(), + CachedOverlayAdvert { + author_npub: author_npub.to_string(), + advert, + created_at, + valid_until_ms, + }, + ); + is_peer + } + + /// Cache-hit lookup for the fetch path: return the cached advert body + /// if present, `None` if the driver must fetch from relays. + pub(super) fn cached_advert(&self, peer_npub: &str) -> Option { + self.lock_cache() + .get(peer_npub) + .map(|cached| cached.advert.clone()) + } + + /// The `created_at` of a cached advert, if any. Used by the stale-check + /// refetch path to decide whether a relay result is newer. + pub(super) fn cached_created_at(&self, peer_npub: &str) -> Option { + self.lock_cache() + .get(peer_npub) + .map(|cached| cached.created_at) + } + + /// Insert a freshly-fetched advert into the cache (fetch-miss path and + /// stale-check refresh). + pub(super) fn insert_fetched(&self, peer_npub: &str, cached: CachedOverlayAdvert) { + self.lock_cache().insert(peer_npub.to_string(), cached); + } + + /// Remove a peer's cached advert (stale-check eviction). + pub(super) fn remove(&self, peer_npub: &str) { + self.lock_cache().remove(peer_npub); + } + + /// Validity-filtered snapshot of cacheable peers for open discovery: + /// entries authored by someone other than us and still valid at + /// `now_ms`. + pub(super) fn open_discovery_candidates( + &self, + max: usize, + now_ms: u64, + ) -> Vec<(String, Vec, u64)> { + let cache = self.lock_cache(); + cache + .values() + .filter(|entry| entry.author_npub != self.npub) + .filter(|entry| entry.valid_until_ms > now_ms) + .map(|entry| { + ( + entry.author_npub.clone(), + entry.advert.endpoints.clone(), + entry.created_at, + ) + }) + .take(max) + .collect() + } + + // --- local advert / publish ---------------------------------------- + + /// Set the local advert we want to publish. Returns `true` when the + /// value changed (so the driver should request a republish). + pub(super) fn set_local_advert(&self, advert: Option) -> bool { + let mut slot = self.lock_local(); + if *slot == advert { + false + } else { + *slot = advert; + true + } + } + + /// Build the publish decision: which advert body to publish, a delete + /// to emit, or nothing. Pure logic — the driver performs the relay I/O + /// and event signing. + pub(super) fn plan_publish(&self) -> Result { + let previous_event_id = *self.lock_event_id(); + if !self.advertise { + return Ok(match previous_event_id { + Some(event_id) => PublishPlan::Delete(event_id), + None => PublishPlan::Nothing, + }); + } + + let mut advert = match self.lock_local().clone() { + Some(advert) => advert, + // Transient absence (e.g., a single tick during startup where + // build_overlay_advert briefly returns None). Don't proactively + // emit a NIP-09 delete: the next publish supersedes the old + // event via parameterized-replaceable semantics, and the NIP-40 + // expiration tag bounds the worst case if we never re-publish. + None => return Ok(PublishPlan::Nothing), + }; + + advert.identifier = ADVERT_IDENTIFIER.to_string(); + advert.version = ADVERT_VERSION; + advert.endpoints.retain(endpoint_advert_is_publicly_usable); + // Defensive: build_overlay_advert returns None on empty endpoints, + // so this is only reachable from non-lifecycle callers. + if advert.endpoints.is_empty() { + return Ok(PublishPlan::Nothing); + } + + if advert.has_udp_nat_endpoint() { + if advert + .signal_relays + .as_ref() + .is_none_or(|relays| relays.is_empty()) + { + return Err(BootstrapError::InvalidAdvert( + "udp:nat endpoint requires non-empty signalRelays".to_string(), + )); + } + if advert + .stun_servers + .as_ref() + .is_none_or(|servers| servers.is_empty()) + { + return Err(BootstrapError::InvalidAdvert( + "udp:nat endpoint requires non-empty stunServers".to_string(), + )); + } + } else { + advert.signal_relays = None; + advert.stun_servers = None; + } + + Ok(PublishPlan::Publish(advert)) + } + + // --- current advert event id (NIP-09 delete-on-withdraw) ----------- + + /// Record the id of a just-published advert event. + pub(super) fn set_event_id(&self, event_id: EventId) { + *self.lock_event_id() = Some(event_id); + } + + /// Clear the recorded advert event id (after emitting a delete). + pub(super) fn clear_event_id(&self) { + *self.lock_event_id() = None; + } + + /// Take and clear the recorded advert event id (shutdown path). + pub(super) fn take_event_id(&self) -> Option { + self.lock_event_id().take() + } + + // --- lock helpers --------------------------------------------------- + + fn lock_cache(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.cache + .lock() + .expect("advert-machine cache mutex poisoned") + } + + fn lock_local(&self) -> std::sync::MutexGuard<'_, Option> { + self.local_advert + .lock() + .expect("advert-machine local-advert mutex poisoned") + } + + fn lock_event_id(&self) -> std::sync::MutexGuard<'_, Option> { + self.current_event_id + .lock() + .expect("advert-machine event-id mutex poisoned") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::nostr::types::OverlayTransportKind; + + fn ep(addr: &str) -> OverlayEndpointAdvert { + OverlayEndpointAdvert { + transport: OverlayTransportKind::Udp, + addr: addr.to_string(), + } + } + + fn advert(endpoints: Vec) -> OverlayAdvert { + OverlayAdvert { + identifier: ADVERT_IDENTIFIER.to_string(), + version: ADVERT_VERSION, + endpoints, + signal_relays: None, + stun_servers: None, + } + } + + fn cached(author: &str, created_at: u64, valid_until_ms: u64) -> CachedOverlayAdvert { + CachedOverlayAdvert { + author_npub: author.to_string(), + advert: advert(vec![ep("1.2.3.4:9000")]), + created_at, + valid_until_ms, + } + } + + fn machine() -> AdvertMachine { + // npub=self, advertise=true, max-age huge, cap=3 + AdvertMachine::new("npub1self".to_string(), true, 10_000_000, 3) + } + + #[test] + fn observe_advert_replaces_only_when_newer_and_flags_peer() { + let m = machine(); + // Fresh peer advert -> cached, log flagged (peer). + assert!(m.observe_advert("npub1peer", advert(vec![ep("1.2.3.4:9000")]), 100, 5000)); + // Older created_at -> not replaced, no log. + assert!(!m.observe_advert("npub1peer", advert(vec![ep("1.2.3.4:9001")]), 50, 5000)); + assert_eq!(m.cached_created_at("npub1peer"), Some(100)); + // Newer created_at -> replaced, log flagged. + assert!(m.observe_advert("npub1peer", advert(vec![ep("1.2.3.4:9002")]), 200, 5000)); + assert_eq!(m.cached_created_at("npub1peer"), Some(200)); + } + + #[test] + fn observe_advert_self_author_caches_but_does_not_flag_log() { + let m = machine(); + // Own advert is still cached (should_replace true) but must NOT be + // flagged as a peer-cached log line. + assert!(!m.observe_advert("npub1self", advert(vec![ep("1.2.3.4:9000")]), 100, 5000)); + assert_eq!(m.cached_created_at("npub1self"), Some(100)); + } + + #[test] + fn prune_drops_expired_and_reports_no_eviction_under_cap() { + let m = machine(); + m.insert_fetched("npub1a", cached("npub1a", 1, 1000)); + m.insert_fetched("npub1b", cached("npub1b", 1, 3000)); + // now=2000 -> npub1a expired, npub1b retained, under cap -> None. + assert_eq!(m.prune(2000), None); + assert_eq!(m.cached_created_at("npub1a"), None); + assert!(m.cached_created_at("npub1b").is_some()); + } + + #[test] + fn prune_size_cap_evicts_oldest_by_valid_until() { + let m = machine(); // cap = 3 + // Four still-valid entries; oldest valid_until must be evicted. + m.insert_fetched("npub1a", cached("npub1a", 1, 1000)); + m.insert_fetched("npub1b", cached("npub1b", 1, 2000)); + m.insert_fetched("npub1c", cached("npub1c", 1, 3000)); + m.insert_fetched("npub1d", cached("npub1d", 1, 4000)); + let evicted = m.prune(500); + assert_eq!(evicted, Some((1, 3))); + // Oldest validity (npub1a) evicted; newest kept. + assert_eq!(m.cached_created_at("npub1a"), None); + assert!(m.cached_created_at("npub1d").is_some()); + } + + #[test] + fn open_discovery_candidates_filters_self_and_expired() { + let m = machine(); + m.insert_fetched("npub1self", cached("npub1self", 1, 9000)); + m.insert_fetched("npub1peer", cached("npub1peer", 1, 9000)); + m.insert_fetched("npub1stale", cached("npub1stale", 1, 1000)); + let out = m.open_discovery_candidates(10, 2000); + assert_eq!(out.len(), 1, "only the valid non-self peer survives"); + assert_eq!(out[0].0, "npub1peer"); + } + + #[test] + fn open_discovery_candidates_respects_max() { + let m = machine(); + for i in 0..5 { + let npub = format!("npub1p{i}"); + m.insert_fetched(&npub, cached(&npub, 1, 9000)); + } + assert_eq!(m.open_discovery_candidates(2, 1000).len(), 2); + } + + #[test] + fn set_local_advert_detects_change() { + let m = machine(); + let a = advert(vec![ep("1.2.3.4:9000")]); + assert!(m.set_local_advert(Some(a.clone())), "first set is a change"); + assert!( + !m.set_local_advert(Some(a.clone())), + "identical set is no change" + ); + assert!(m.set_local_advert(None), "clearing is a change"); + } + + #[test] + fn plan_publish_strips_relays_for_non_nat_advert() { + let m = machine(); + let mut a = advert(vec![ep("1.2.3.4:9000")]); + a.signal_relays = Some(vec!["wss://relay".to_string()]); + a.stun_servers = Some(vec!["stun:host:3478".to_string()]); + m.set_local_advert(Some(a)); + match m.plan_publish().expect("plan ok") { + PublishPlan::Publish(out) => { + assert!( + out.signal_relays.is_none(), + "non-nat advert strips signalRelays" + ); + assert!( + out.stun_servers.is_none(), + "non-nat advert strips stunServers" + ); + assert_eq!(out.identifier, ADVERT_IDENTIFIER); + assert_eq!(out.version, ADVERT_VERSION); + } + other => panic!("expected Publish, got {other:?}"), + } + } + + #[test] + fn plan_publish_keeps_relays_for_nat_advert() { + let m = machine(); + let mut a = advert(vec![ep("nat")]); + a.signal_relays = Some(vec!["wss://relay".to_string()]); + a.stun_servers = Some(vec!["stun:host:3478".to_string()]); + m.set_local_advert(Some(a)); + match m.plan_publish().expect("plan ok") { + PublishPlan::Publish(out) => { + assert!(out.has_udp_nat_endpoint()); + assert_eq!(out.signal_relays.as_deref().map(<[_]>::len), Some(1)); + assert_eq!(out.stun_servers.as_deref().map(<[_]>::len), Some(1)); + } + other => panic!("expected Publish, got {other:?}"), + } + } + + #[test] + fn plan_publish_nat_without_relays_errors() { + let m = machine(); + m.set_local_advert(Some(advert(vec![ep("nat")]))); + assert!(matches!( + m.plan_publish(), + Err(BootstrapError::InvalidAdvert(_)) + )); + } + + #[test] + fn plan_publish_nothing_when_no_local_advert() { + let m = machine(); + assert!(matches!(m.plan_publish(), Ok(PublishPlan::Nothing))); + } + + #[test] + fn plan_publish_nothing_when_disabled_without_prior_event() { + // advertise=false, no prior event id -> Nothing. + let m = AdvertMachine::new("npub1self".to_string(), false, 10_000_000, 3); + m.set_local_advert(Some(advert(vec![ep("1.2.3.4:9000")]))); + assert!(matches!(m.plan_publish(), Ok(PublishPlan::Nothing))); + } + + #[test] + fn event_id_set_clear_take_roundtrip() { + let m = machine(); + assert_eq!(m.take_event_id(), None); + m.clear_event_id(); + assert_eq!(m.take_event_id(), None); + } +} diff --git a/src/nostr/mod.rs b/src/nostr/mod.rs index 24ec3cf..af300ad 100644 --- a/src/nostr/mod.rs +++ b/src/nostr/mod.rs @@ -1,3 +1,4 @@ +mod advert; mod driver; mod failure_state; mod handoff; @@ -5,6 +6,7 @@ mod runtime; mod signal; mod stun; mod traversal; +mod traversal_machine; mod types; #[cfg(test)] diff --git a/src/nostr/runtime.rs b/src/nostr/runtime.rs index e2e1a16..e7cc123 100644 --- a/src/nostr/runtime.rs +++ b/src/nostr/runtime.rs @@ -16,6 +16,7 @@ use tokio::sync::{Mutex, Notify, RwLock, Semaphore, broadcast, mpsc, oneshot}; use tokio::task::JoinHandle; use tracing::{debug, info, trace, warn}; +use super::advert::{AdvertMachine, PublishPlan}; use super::failure_state::FailureState; use super::handoff::EstablishedTraversal; use super::signal::{ @@ -25,14 +26,15 @@ use super::signal::{ }; use super::stun::observe_traversal_addresses; use super::traversal::{nonce, now_ms, planned_remote_endpoints, run_punch_attempt}; +use super::traversal_machine::{OfferDisposition, SeenDecision, TraversalMachine}; use super::types::{ ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, BootstrapError, BootstrapEvent, CachedOverlayAdvert, NostrFailureDecision, NostrPeerFailureView, NostrRefetchOutcome, OverlayAdvert, OverlayEndpointAdvert, PROTOCOL_VERSION, PunchHint, SIGNAL_KIND, TraversalAnswer, TraversalOffer, }; +use crate::PeerIdentity; use crate::config::{NostrRendezvousConfig, PeerConfig}; -use crate::{NodeAddr, PeerIdentity}; const ADVERT_CACHE_STALE_GRACE_MULTIPLIER: u64 = 2; @@ -51,42 +53,6 @@ fn short_id(id: &str) -> String { } } -/// Decide whether an incoming-offer responder session should be suppressed -/// in favour of our own already-running outbound initiator session. -/// -/// Two peers that each have the other as `auto_connect` simultaneously run an -/// initiator traversal *and* a responder traversal for the same peer, binding a -/// separate UDP socket per session. Each node then emits two -/// `BootstrapEvent::Established` events and `adopt_established_traversal` keeps -/// only the first on a non-deterministic race; when the two nodes' independent -/// races resolve to mismatched sessions, each side's Noise msg1 lands on a peer -/// port the peer already stopped draining and both handshakes stall (root cause -/// of ISSUE-2026-0031). -/// -/// To collapse the four-socket dance to a single, guaranteed-matching socket -/// pair, both nodes deterministically keep the session **initiated by the -/// smaller `NodeAddr`** — reusing the project's existing NodeAddr tie-breaker -/// convention (`cross_connection_winner`, the rekey dual-init resolution, and -/// the dual-cross-init adopt path in `lifecycle.rs`). -/// -/// This is evaluated on the responder path, where the session being handled is -/// *peer-initiated*. It returns `true` (suppress this responder session) only -/// when genuine duplication exists — i.e. we also have an in-flight outbound -/// initiator for this same peer (`have_active_initiator`) — and our own -/// initiator session is the preferred one (`our_addr < peer_addr`). When there -/// is no co-active initiator (the asymmetric / one-sided `auto_connect` case, -/// where only one session exists at all) it never suppresses, so connectivity -/// is preserved. The `our_addr == peer_addr` case (self / loopback) and any -/// caller that cannot derive a peer `NodeAddr` likewise fall through to "do not -/// suppress". -pub(super) fn suppress_responder_for_own_initiator( - our_addr: &NodeAddr, - peer_addr: &NodeAddr, - have_active_initiator: bool, -) -> bool { - have_active_initiator && our_addr < peer_addr -} - fn endpoint_summary(endpoints: &[OverlayEndpointAdvert]) -> String { endpoints .iter() @@ -117,7 +83,7 @@ fn is_unroutable_direct_advert_ip(ip: std::net::IpAddr) -> bool { } } -fn endpoint_advert_is_publicly_usable(endpoint: &OverlayEndpointAdvert) -> bool { +pub(super) fn endpoint_advert_is_publicly_usable(endpoint: &OverlayEndpointAdvert) -> bool { let addr = endpoint.addr.trim(); if addr.is_empty() { return false; @@ -183,12 +149,9 @@ pub struct NostrRendezvous { pubkey: PublicKey, npub: String, config: NostrRendezvousConfig, - advert_cache: RwLock>, - local_advert: RwLock>, - current_advert_event_id: RwLock>, + advert: AdvertMachine, + traversal: TraversalMachine, pending_answers: Mutex>>>, - active_initiators: Mutex>, - seen_sessions: Mutex>, offer_slots: Arc, event_tx: mpsc::UnboundedSender, event_rx: Mutex>, @@ -249,18 +212,25 @@ impl NostrRendezvous { config.failure_state_max_entries, ); + let advert = AdvertMachine::new( + npub.clone(), + config.advertise, + config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER, + config.advert_cache_max_entries, + ); + let traversal = TraversalMachine::new( + config.replay_window_secs * 1000, + config.seen_sessions_max_entries, + ); let runtime = Arc::new(Self { client, keys, pubkey, npub, config, - advert_cache: RwLock::new(HashMap::new()), - local_advert: RwLock::new(None), - current_advert_event_id: RwLock::new(None), + advert, + traversal, pending_answers: Mutex::new(HashMap::new()), - active_initiators: Mutex::new(HashSet::new()), - seen_sessions: Mutex::new(HashMap::new()), offer_slots, event_tx, event_rx: Mutex::new(event_rx), @@ -309,11 +279,8 @@ impl NostrRendezvous { pub async fn request_connect(self: &Arc, peer_config: PeerConfig) { let peer_npub = peer_config.npub.clone(); - { - let mut active = self.active_initiators.lock().await; - if !active.insert(peer_npub.clone()) { - return; - } + if !self.traversal.begin_initiator(&peer_npub) { + return; } let runtime = Arc::clone(self); @@ -326,7 +293,7 @@ impl NostrRendezvous { }, }; let _ = runtime.event_tx.send(event); - runtime.active_initiators.lock().await.remove(&peer_npub); + runtime.traversal.end_initiator(&peer_npub); }); } @@ -507,12 +474,7 @@ impl NostrRendezvous { 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 cached_created_at = self.advert.cached_created_at(peer_npub); let events = match self .client @@ -541,7 +503,7 @@ impl NostrRendezvous { 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.advert.remove(peer_npub); self.failure_state.reset_streak_after_refresh(peer_npub); return NostrRefetchOutcome::Evicted; }; @@ -561,10 +523,7 @@ impl NostrRendezvous { created_at: relay_created_at, valid_until_ms, }; - self.advert_cache - .write() - .await - .insert(peer_npub.to_string(), updated); + self.advert.insert_fetched(peer_npub, updated); self.failure_state.reset_streak_after_refresh(peer_npub); NostrRefetchOutcome::Refreshed } @@ -584,19 +543,9 @@ impl NostrRendezvous { self: &Arc, advert: Option, ) -> Result<(), BootstrapError> { - let changed = { - let mut slot = self.local_advert.write().await; - if *slot == advert { - false - } else { - *slot = advert; - true - } - }; - if !changed { - return Ok(()); + if self.advert.set_local_advert(advert) { + self.request_publish_advert(); } - self.request_publish_advert(); Ok(()) } @@ -617,22 +566,8 @@ impl NostrRendezvous { &self, max: usize, ) -> Vec<(String, Vec, u64)> { - self.prune_advert_cache().await; - let now = now_ms(); - let cache = self.advert_cache.read().await; - cache - .values() - .filter(|entry| entry.author_npub != self.npub) - .filter(|entry| entry.valid_until_ms > now) - .map(|entry| { - ( - entry.author_npub.clone(), - entry.advert.endpoints.clone(), - entry.created_at, - ) - }) - .take(max) - .collect() + self.prune_advert_cache(); + self.advert.open_discovery_candidates(max, now_ms()) } pub async fn shutdown(&self) -> Result<(), BootstrapError> { @@ -655,7 +590,7 @@ impl NostrRendezvous { // permanent shutdown. An explicit retraction races with the next // daemon's republish on strict relays (e.g. Damus rate-limits the // burst, leaving the advert deleted and never restored). - let _ = self.current_advert_event_id.write().await.take(); + let _ = self.advert.take_event_id(); if let Some(handle) = self.notify_task.lock().await.take() { handle.abort(); @@ -701,32 +636,23 @@ impl NostrRendezvous { && let Ok(advert) = Self::parse_overlay_advert_event(&event, &self.config.app) { - let mut cache = self.advert_cache.write().await; - let should_replace = cache - .get(&author_npub) - .map(|existing| existing.created_at <= event.created_at.as_secs()) - .unwrap_or(true); - if should_replace && author_npub != self.npub { + let endpoints = endpoint_summary(&advert.endpoints); + let created_at = event.created_at.as_secs(); + if self.advert.observe_advert( + &author_npub, + advert, + created_at, + valid_until_ms, + ) { debug!( peer = %short_npub(&author_npub), - endpoints = %endpoint_summary(&advert.endpoints), + endpoints = %endpoints, event = %short_id(&event.id.to_string()), "advert: peer cached" ); } - if should_replace { - cache.insert( - author_npub.clone(), - CachedOverlayAdvert { - author_npub, - advert, - created_at: event.created_at.as_secs(), - valid_until_ms, - }, - ); - } } - self.prune_advert_cache().await; + self.prune_advert_cache(); continue; } @@ -949,59 +875,17 @@ impl NostrRendezvous { } async fn publish_advert(&self) -> Result<(), BootstrapError> { - let previous_event_id = self.current_advert_event_id.read().await.to_owned(); - if !self.config.advertise { - if let Some(event_id) = previous_event_id { + let advert = match self.advert.plan_publish()? { + PublishPlan::Nothing => return Ok(()), + PublishPlan::Delete(event_id) => { self.publish_delete(&self.config.advert_relays, [event_id]) .await?; - *self.current_advert_event_id.write().await = None; + self.advert.clear_event_id(); + return Ok(()); } - return Ok(()); - } - - let mut advert = match self.local_advert.read().await.clone() { - Some(advert) => advert, - // Transient absence (e.g., a single tick during startup where - // build_overlay_advert briefly returns None). Don't proactively - // emit a NIP-09 delete: the next publish supersedes the old - // event via parameterized-replaceable semantics, and the NIP-40 - // expiration tag bounds the worst case if we never re-publish. - None => return Ok(()), + PublishPlan::Publish(advert) => advert, }; - advert.identifier = ADVERT_IDENTIFIER.to_string(); - advert.version = ADVERT_VERSION; - advert.endpoints.retain(endpoint_advert_is_publicly_usable); - // Defensive: build_overlay_advert returns None on empty endpoints, - // so this is only reachable from non-lifecycle callers. - if advert.endpoints.is_empty() { - return Ok(()); - } - - if advert.has_udp_nat_endpoint() { - if advert - .signal_relays - .as_ref() - .is_none_or(|relays| relays.is_empty()) - { - return Err(BootstrapError::InvalidAdvert( - "udp:nat endpoint requires non-empty signalRelays".to_string(), - )); - } - if advert - .stun_servers - .as_ref() - .is_none_or(|servers| servers.is_empty()) - { - return Err(BootstrapError::InvalidAdvert( - "udp:nat endpoint requires non-empty stunServers".to_string(), - )); - } - } else { - advert.signal_relays = None; - advert.stun_servers = None; - } - let expires_at = now_ms() + self.config.advert_ttl_secs * 1000; let tags = vec![ Tag::identifier(ADVERT_IDENTIFIER.to_string()), @@ -1031,7 +915,7 @@ impl NostrRendezvous { // NIP-09 delete here is redundant and races with the replacement // publish, which strict relays (e.g. Damus) honor by removing the // new advert too. - *self.current_advert_event_id.write().await = Some(event.id); + self.advert.set_event_id(event.id); Ok(()) } @@ -1270,17 +1154,17 @@ impl NostrRendezvous { // single matching socket pair survives on both sides. Asymmetric / // one-sided `auto_connect` (no co-active initiator) is never suppressed, // preserving connectivity. See `suppress_responder_for_own_initiator`. - if self.active_initiators.lock().await.contains(&sender_npub) { - match ( - PeerIdentity::from_npub(&self.npub), - PeerIdentity::from_npub(&sender_npub), - ) { - (Ok(ours), Ok(theirs)) => { - if suppress_responder_for_own_initiator( - ours.node_addr(), - theirs.node_addr(), - true, - ) { + match ( + PeerIdentity::from_npub(&self.npub), + PeerIdentity::from_npub(&sender_npub), + ) { + (Ok(ours), Ok(theirs)) => { + match self.traversal.classify_incoming_offer( + &sender_npub, + ours.node_addr(), + theirs.node_addr(), + ) { + OfferDisposition::Suppress => { debug!( peer = %peer_short, session = %short_id(&offer.session_id), @@ -1288,20 +1172,38 @@ impl NostrRendezvous { ); return Ok(()); } + OfferDisposition::Proceed => {} } - _ => { - // Could not derive a NodeAddr for one side; fall through and - // answer rather than risk suppressing the only session. - trace!( - peer = %peer_short, - "traversal: could not derive NodeAddr for dedup, answering offer" + } + _ => { + // Could not derive a NodeAddr for one side; fall through and + // answer rather than risk suppressing the only session. + trace!( + peer = %peer_short, + "traversal: could not derive NodeAddr for dedup, answering offer" + ); + } + } + + match self + .traversal + .note_session_seen(&offer.session_id, now_ms()) + { + SeenDecision::Replay => { + return Err(BootstrapError::Replay(offer.session_id.clone())); + } + SeenDecision::Fresh { evicted } => { + if let Some((evicted, retained)) = evicted { + debug!( + evicted = evicted, + retained = retained, + cap = self.config.seen_sessions_max_entries, + "seen-sessions cache overflow; evicted oldest entries" ); } } } - self.mark_session_seen(&offer.session_id).await?; - let base_socket = std::net::UdpSocket::bind(("0.0.0.0", 0))?; base_socket.set_nonblocking(true)?; let (reflexive_address, local_addresses, stun_server) = observe_traversal_addresses( @@ -1396,15 +1298,15 @@ impl NostrRendezvous { peer_npub: &str, target_pubkey: PublicKey, ) -> Result { - self.prune_advert_cache().await; - if let Some(cached) = self.advert_cache.read().await.get(peer_npub).cloned() { + self.prune_advert_cache(); + if let Some(advert) = self.advert.cached_advert(peer_npub) { debug!( peer = %short_npub(peer_npub), source = "cache", - endpoints = %endpoint_summary(&cached.advert.endpoints), + endpoints = %endpoint_summary(&advert.endpoints), "advert: resolved" ); - return Ok(cached.advert); + return Ok(advert); } let events = self @@ -1453,11 +1355,8 @@ impl NostrRendezvous { endpoints = %endpoint_summary(&cached.advert.endpoints), "advert: resolved" ); - self.advert_cache - .write() - .await - .insert(peer_npub.to_string(), cached.clone()); - self.prune_advert_cache().await; + self.advert.insert_fetched(peer_npub, cached.clone()); + self.prune_advert_cache(); Ok(cached.advert) } @@ -1603,39 +1502,19 @@ impl NostrRendezvous { Ok(advert) } - async fn prune_advert_cache(&self) { - let now = now_ms(); - let mut cache = self.advert_cache.write().await; - cache.retain(|_, entry| entry.valid_until_ms > now); - if cache.len() <= self.config.advert_cache_max_entries { - return; + fn prune_advert_cache(&self) { + if let Some((evicted, retained)) = self.advert.prune(now_ms()) { + debug!( + evicted, + retained, + cap = self.config.advert_cache_max_entries, + "advert cache overflow; evicted oldest entries" + ); } - - let mut oldest = cache - .iter() - .map(|(npub, entry)| (npub.clone(), entry.valid_until_ms)) - .collect::>(); - oldest.sort_by_key(|(_, ts)| *ts); - let overflow = cache - .len() - .saturating_sub(self.config.advert_cache_max_entries); - for (npub, _) in oldest.into_iter().take(overflow) { - cache.remove(&npub); - } - debug!( - evicted = overflow, - retained = cache.len(), - cap = self.config.advert_cache_max_entries, - "advert cache overflow; evicted oldest entries" - ); - } - - fn advert_max_age_ms(&self) -> u64 { - self.config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER } fn event_valid_until_ms(&self, event: &Event) -> Option { - Self::compute_advert_valid_until_ms(event, self.advert_max_age_ms(), now_ms()) + self.advert.event_valid_until_ms(event, now_ms()) } pub(super) fn compute_advert_valid_until_ms( @@ -1699,37 +1578,6 @@ impl NostrRendezvous { .map_err(|e| BootstrapError::Nostr(e.to_string()))?; Ok(()) } - - async fn mark_session_seen(&self, session_id: &str) -> Result<(), BootstrapError> { - let now = now_ms(); - let expiry = now + self.config.replay_window_secs * 1000; - let mut seen = self.seen_sessions.lock().await; - seen.retain(|_, expires_at| *expires_at > now); - if seen.contains_key(session_id) { - return Err(BootstrapError::Replay(session_id.to_string())); - } - seen.insert(session_id.to_string(), expiry); - if seen.len() > self.config.seen_sessions_max_entries { - let mut oldest = seen - .iter() - .map(|(session, expires_at)| (session.clone(), *expires_at)) - .collect::>(); - oldest.sort_by_key(|(_, expires_at)| *expires_at); - let overflow = seen - .len() - .saturating_sub(self.config.seen_sessions_max_entries); - for (session, _) in oldest.into_iter().take(overflow) { - seen.remove(&session); - } - debug!( - evicted = overflow, - retained = seen.len(), - cap = self.config.seen_sessions_max_entries, - "seen-sessions cache overflow; evicted oldest entries" - ); - } - Ok(()) - } } #[cfg(test)] @@ -1755,18 +1603,25 @@ impl NostrRendezvous { config.warn_log_interval_secs, config.failure_state_max_entries, ); + let advert = AdvertMachine::new( + npub.clone(), + config.advertise, + config.advert_ttl_secs * 1000 * ADVERT_CACHE_STALE_GRACE_MULTIPLIER, + config.advert_cache_max_entries, + ); + let traversal = TraversalMachine::new( + config.replay_window_secs * 1000, + config.seen_sessions_max_entries, + ); Self { client, keys, pubkey, npub, config, - advert_cache: RwLock::new(HashMap::new()), - local_advert: RwLock::new(None), - current_advert_event_id: RwLock::new(None), + advert, + traversal, pending_answers: Mutex::new(HashMap::new()), - active_initiators: Mutex::new(HashSet::new()), - seen_sessions: Mutex::new(HashMap::new()), offer_slots, event_tx, event_rx: Mutex::new(event_rx), @@ -1806,8 +1661,7 @@ impl NostrRendezvous { /// Insert a cached advert directly into the in-memory cache. Used by /// unit tests to set up consumer-side state without needing live relays. pub(crate) async fn insert_advert_for_test(&self, npub: String, advert: CachedOverlayAdvert) { - let mut cache = self.advert_cache.write().await; - cache.insert(npub, advert); + self.advert.insert_fetched(&npub, advert); } /// Queue a bootstrap event directly for lifecycle tests without live relays diff --git a/src/nostr/tests.rs b/src/nostr/tests.rs index 4bda840..1bcf307 100644 --- a/src/nostr/tests.rs +++ b/src/nostr/tests.rs @@ -1,6 +1,6 @@ use nostr::prelude::{EventBuilder, Kind, Tag, Timestamp}; -use super::runtime::{NostrRendezvous, suppress_responder_for_own_initiator}; +use super::runtime::NostrRendezvous; use super::signal::{ FreshnessOutcome, build_signal_event, create_traversal_answer, create_traversal_offer, estimate_clock_skew, validate_offer_freshness, validate_traversal_answer_for_offer, @@ -10,6 +10,7 @@ use super::traversal::{ PunchStrategy, build_punch_packet, parse_punch_packet, plan_punch_targets, planned_remote_endpoints, session_hash, }; +use super::traversal_machine::suppress_responder_for_own_initiator; use super::{ ADVERT_IDENTIFIER, ADVERT_KIND, ADVERT_VERSION, OverlayAdvert, OverlayEndpointAdvert, OverlayTransportKind, PunchHint, PunchPacketKind, TraversalAddress, diff --git a/src/nostr/traversal.rs b/src/nostr/traversal.rs index 49f875a..d867262 100644 --- a/src/nostr/traversal.rs +++ b/src/nostr/traversal.rs @@ -177,17 +177,15 @@ pub(super) async fn run_punch_attempt( let Ok(Ok((len, remote))) = recv else { break Err(BootstrapError::PunchTimeout(session_id.to_string())); }; - let Ok(packet) = parse_punch_packet(&buf[..len]) else { - continue; - }; - if packet.session_hash != expected_hash { - continue; + match classify_punch_packet(&buf[..len], expected_hash) { + PunchAction::Ignore => continue, + PunchAction::Ack { sequence } => { + let ack = build_punch_packet(PunchPacketKind::Ack, sequence, session_id); + let _ = udp.send_to(&ack, remote).await; + break Ok(remote); + } + PunchAction::Matched => break Ok(remote), } - if packet.kind == PunchPacketKind::Probe { - let ack = build_punch_packet(PunchPacketKind::Ack, packet.sequence, session_id); - let _ = udp.send_to(&ack, remote).await; - } - break Ok(remote); }; send_handle.abort(); result @@ -274,3 +272,82 @@ pub(super) fn parse_punch_packet(bytes: &[u8]) -> Result PunchAction { + let Ok(packet) = parse_punch_packet(bytes) else { + return PunchAction::Ignore; + }; + if packet.session_hash != expected_hash { + return PunchAction::Ignore; + } + if packet.kind == PunchPacketKind::Probe { + PunchAction::Ack { + sequence: packet.sequence, + } + } else { + PunchAction::Matched + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const SESSION: &str = "session-classify-vectors"; + + #[test] + fn classify_ignores_unparseable_bytes() { + // P1: too short to parse. + assert_eq!( + classify_punch_packet(&[0u8; 4], session_hash(SESSION)), + PunchAction::Ignore + ); + } + + #[test] + fn classify_ignores_mismatched_session_hash() { + // P2: parseable, but hash is for a different session. + let packet = build_punch_packet(PunchPacketKind::Probe, 7, SESSION); + let other_hash = session_hash("some-other-session"); + assert_eq!( + classify_punch_packet(&packet, other_hash), + PunchAction::Ignore + ); + } + + #[test] + fn classify_probe_matching_hash_acks_with_sequence() { + // P3: matching probe -> Ack carrying the packet's sequence. + let packet = build_punch_packet(PunchPacketKind::Probe, 42, SESSION); + assert_eq!( + classify_punch_packet(&packet, session_hash(SESSION)), + PunchAction::Ack { sequence: 42 } + ); + } + + #[test] + fn classify_ack_matching_hash_is_matched() { + // P4: matching non-probe (ack) -> Matched. + let packet = build_punch_packet(PunchPacketKind::Ack, 3, SESSION); + assert_eq!( + classify_punch_packet(&packet, session_hash(SESSION)), + PunchAction::Matched + ); + } +} diff --git a/src/nostr/traversal_machine.rs b/src/nostr/traversal_machine.rs new file mode 100644 index 0000000..844c0af --- /dev/null +++ b/src/nostr/traversal_machine.rs @@ -0,0 +1,333 @@ +//! Synchronous decision core for the Nostr NAT-traversal control state. +//! +//! `TraversalMachine` owns the engine-scoped, cross-session state that +//! previously lived directly on `NostrRendezvous` — the set of in-flight +//! outbound initiators, and the replay/seen-sessions cache — and hosts the +//! *decisions* over that state: initiator dedup, the dual-`auto_connect` +//! responder election, and replay rejection. +//! +//! Following the sans-IO shape used by `advert` / `failure_state`, every +//! method here is synchronous, performs no network I/O and no `.await`, +//! holds its state behind `std::sync::Mutex`, and takes the current time as +//! an explicit `now_ms: u64` input rather than reading a clock. The async +//! driver on `NostrRendezvous` reads the clock at the call site, derives the +//! `NodeAddr`s from npubs, and performs the actual socket/STUN/relay I/O. +//! +//! The inherently-async concurrency primitives stay driver-side: the +//! `pending_answers` oneshot routing and the `offer_slots` semaphore +//! admission are not modeled here, and neither is the punch send-cadence. + +use std::collections::{HashMap, HashSet}; +use std::sync::Mutex; + +use crate::NodeAddr; + +/// Decide whether an incoming-offer responder session should be suppressed +/// in favour of our own already-running outbound initiator session. +/// +/// Two peers that each have the other as `auto_connect` simultaneously run an +/// initiator traversal *and* a responder traversal for the same peer, binding a +/// separate UDP socket per session. Each node then emits two +/// `BootstrapEvent::Established` events and `adopt_established_traversal` keeps +/// only the first on a non-deterministic race; when the two nodes' independent +/// races resolve to mismatched sessions, each side's Noise msg1 lands on a peer +/// port the peer already stopped draining and both handshakes stall. +/// +/// To collapse the four-socket dance to a single, guaranteed-matching socket +/// pair, both nodes deterministically keep the session **initiated by the +/// smaller `NodeAddr`** — reusing the project's existing NodeAddr tie-breaker +/// convention (`cross_connection_winner`, the rekey dual-init resolution, and +/// the dual-cross-init adopt path in `lifecycle.rs`). +/// +/// This is evaluated on the responder path, where the session being handled is +/// *peer-initiated*. It returns `true` (suppress this responder session) only +/// when genuine duplication exists — i.e. we also have an in-flight outbound +/// initiator for this same peer (`have_active_initiator`) — and our own +/// initiator session is the preferred one (`our_addr < peer_addr`). When there +/// is no co-active initiator (the asymmetric / one-sided `auto_connect` case, +/// where only one session exists at all) it never suppresses, so connectivity +/// is preserved. The `our_addr == peer_addr` case (self / loopback) and any +/// caller that cannot derive a peer `NodeAddr` likewise fall through to "do not +/// suppress". +pub(super) fn suppress_responder_for_own_initiator( + our_addr: &NodeAddr, + peer_addr: &NodeAddr, + have_active_initiator: bool, +) -> bool { + have_active_initiator && our_addr < peer_addr +} + +/// Result of the dual-init responder election. Returned by +/// [`TraversalMachine::classify_incoming_offer`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum OfferDisposition { + /// Answer this offer normally. + Proceed, + /// Decline this responder session; our own outbound initiator wins. + Suppress, +} + +/// Result of the replay / seen-sessions check. Returned by +/// [`TraversalMachine::note_session_seen`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum SeenDecision { + /// First time we've seen this session id within the replay window. If a + /// cap eviction occurred, `evicted` carries `(evicted, retained)` for the + /// driver's overflow debug line. + Fresh { evicted: Option<(usize, usize)> }, + /// The session id is already present within the replay window. + Replay, +} + +pub(super) struct TraversalMachine { + /// Replay window length in ms (`replay_window_secs * 1000`). + replay_window_ms: u64, + /// Size cap for the seen-sessions cache. + seen_max_entries: usize, + /// npubs of peers we currently have an in-flight outbound initiator for. + active_initiators: Mutex>, + /// Replay cache: session id -> expiry (ms). + seen_sessions: Mutex>, +} + +impl TraversalMachine { + pub(super) fn new(replay_window_ms: u64, seen_max_entries: usize) -> Self { + Self { + replay_window_ms, + seen_max_entries, + active_initiators: Mutex::new(HashSet::new()), + seen_sessions: Mutex::new(HashMap::new()), + } + } + + // --- initiator lifecycle ------------------------------------------- + + /// Register an in-flight outbound initiator for `npub`. Returns `false` + /// when one is already active (dedup — the driver should not start a + /// second). + pub(super) fn begin_initiator(&self, npub: &str) -> bool { + self.lock_initiators().insert(npub.to_string()) + } + + /// Clear the in-flight initiator for `npub` on task completion. + pub(super) fn end_initiator(&self, npub: &str) { + self.lock_initiators().remove(npub); + } + + // --- responder dual-init election ---------------------------------- + + /// Decide whether an incoming offer from `sender_npub` should be answered + /// or suppressed in favour of our own outbound initiator. The driver + /// derives both `NodeAddr`s (keeping today's derivation-failure "answer + /// anyway" fallthrough) and passes them in; the machine reads its own + /// active-initiator membership under its lock. + pub(super) fn classify_incoming_offer( + &self, + sender_npub: &str, + our_addr: &NodeAddr, + peer_addr: &NodeAddr, + ) -> OfferDisposition { + let have_active = self.lock_initiators().contains(sender_npub); + if suppress_responder_for_own_initiator(our_addr, peer_addr, have_active) { + OfferDisposition::Suppress + } else { + OfferDisposition::Proceed + } + } + + // --- replay / seen-sessions ---------------------------------------- + + /// Record that `session_id` was seen at `now_ms`. Prunes expired entries, + /// rejects a replay, inserts the fresh id, then applies the size cap. + /// + /// Returns [`SeenDecision::Replay`] in place of the old + /// `Err(BootstrapError::Replay)`, or [`SeenDecision::Fresh`] carrying the + /// `(evicted, retained)` pair when a cap eviction occurred. + pub(super) fn note_session_seen(&self, session_id: &str, now_ms: u64) -> SeenDecision { + let expiry = now_ms + self.replay_window_ms; + let mut seen = self.lock_seen(); + seen.retain(|_, expires_at| *expires_at > now_ms); + if seen.contains_key(session_id) { + return SeenDecision::Replay; + } + seen.insert(session_id.to_string(), expiry); + if seen.len() > self.seen_max_entries { + let mut oldest = seen + .iter() + .map(|(session, expires_at)| (session.clone(), *expires_at)) + .collect::>(); + oldest.sort_by_key(|(_, expires_at)| *expires_at); + let overflow = seen.len().saturating_sub(self.seen_max_entries); + for (session, _) in oldest.into_iter().take(overflow) { + seen.remove(&session); + } + return SeenDecision::Fresh { + evicted: Some((overflow, seen.len())), + }; + } + SeenDecision::Fresh { evicted: None } + } + + // --- lock helpers --------------------------------------------------- + + fn lock_initiators(&self) -> std::sync::MutexGuard<'_, HashSet> { + self.active_initiators + .lock() + .expect("traversal-machine active-initiators mutex poisoned") + } + + fn lock_seen(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.seen_sessions + .lock() + .expect("traversal-machine seen-sessions mutex poisoned") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn node_addr(first_byte: u8) -> NodeAddr { + let mut bytes = [0u8; 16]; + bytes[0] = first_byte; + NodeAddr::from_bytes(bytes) + } + + fn machine() -> TraversalMachine { + TraversalMachine::new(1_000_000, 3) + } + + // --- election ordering (classify_incoming_offer) ------------------- + + #[test] + fn election_vectors() { + let smaller = node_addr(0x01); + let larger = node_addr(0x02); + let peer = "npub1peer"; + + // V1: co-active initiator, our addr smaller -> Suppress. + let m = machine(); + assert!(m.begin_initiator(peer)); + assert_eq!( + m.classify_incoming_offer(peer, &smaller, &larger), + OfferDisposition::Suppress + ); + + // V2: co-active initiator, our addr larger -> Proceed. + let m = machine(); + assert!(m.begin_initiator(peer)); + assert_eq!( + m.classify_incoming_offer(peer, &larger, &smaller), + OfferDisposition::Proceed + ); + + // V3: no co-active initiator, our addr smaller -> Proceed + // (asymmetric one-sided auto_connect). + let m = machine(); + assert_eq!( + m.classify_incoming_offer(peer, &smaller, &larger), + OfferDisposition::Proceed + ); + + // V4: co-active initiator, equal addresses -> Proceed (self/loopback). + let m = machine(); + assert!(m.begin_initiator(peer)); + assert_eq!( + m.classify_incoming_offer(peer, &smaller, &smaller), + OfferDisposition::Proceed + ); + + // V5: classify BEFORE begin_initiator -> Proceed (offer-first race). + let m = machine(); + assert_eq!( + m.classify_incoming_offer(peer, &smaller, &larger), + OfferDisposition::Proceed + ); + + // V6: begin then end then classify -> Proceed (initiator finished). + let m = machine(); + assert!(m.begin_initiator(peer)); + m.end_initiator(peer); + assert_eq!( + m.classify_incoming_offer(peer, &smaller, &larger), + OfferDisposition::Proceed + ); + } + + // --- initiator dedup ------------------------------------------------ + + #[test] + fn initiator_dedup() { + let m = machine(); + assert!(m.begin_initiator("npub1p"), "first is a fresh initiator"); + assert!( + !m.begin_initiator("npub1p"), + "second for same npub is a dup" + ); + m.end_initiator("npub1p"); + assert!( + m.begin_initiator("npub1p"), + "fresh again after end_initiator" + ); + } + + // --- replay (note_session_seen) ------------------------------------ + + #[test] + fn replay_first_then_repeat() { + // R1: first id Fresh; same id within window Replay. + let m = machine(); + assert_eq!( + m.note_session_seen("s1", 1000), + SeenDecision::Fresh { evicted: None } + ); + assert_eq!(m.note_session_seen("s1", 1500), SeenDecision::Replay); + } + + #[test] + fn replay_prunes_expired() { + // R2: an entry past its expiry is pruned, so re-seeing it is Fresh. + let m = machine(); // replay_window_ms = 1_000_000 + assert_eq!( + m.note_session_seen("s1", 1000), + SeenDecision::Fresh { evicted: None } + ); + // now well past s1's expiry (1000 + 1_000_000): s1 pruned by retain, + // so s1 is Fresh again rather than Replay. + assert_eq!( + m.note_session_seen("s1", 5_000_000), + SeenDecision::Fresh { evicted: None } + ); + } + + #[test] + fn replay_cap_evicts_oldest_by_expiry() { + // R3: cap overflow evicts oldest-by-expiry, returns (evicted, retained). + let m = machine(); // cap = 3, window huge so nothing expires here + assert_eq!( + m.note_session_seen("s1", 1), + SeenDecision::Fresh { evicted: None } + ); + assert_eq!( + m.note_session_seen("s2", 2), + SeenDecision::Fresh { evicted: None } + ); + assert_eq!( + m.note_session_seen("s3", 3), + SeenDecision::Fresh { evicted: None } + ); + assert_eq!( + m.note_session_seen("s4", 4), + SeenDecision::Fresh { + evicted: Some((1, 3)) + } + ); + // Oldest expiry (s1) was evicted, so re-seeing it is Fresh. + assert_eq!( + m.note_session_seen("s1", 5), + SeenDecision::Fresh { + evicted: Some((1, 3)) + } + ); + } +}