diff --git a/src/replication/mod.rs b/src/replication/mod.rs index c913b3a0..98a44dc3 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -103,6 +103,9 @@ struct FirstAuditObservability { coalesced: AtomicU64, duplicates: AtomicU64, capacity_evicted: AtomicU64, + /// An event targeting the local peer itself, dropped at ingress: the node + /// cannot network-audit itself (no dialable address for the local peer). + self_target_skipped: AtomicU64, /// A strictly-lower-count same-peer nomination that was dropped so a /// higher-count pending pin survived. A sustained rise is the signal of an /// attempted "erase the inflated pin with a cheaper one" self-suppression. @@ -121,6 +124,30 @@ struct FirstAuditObservability { inflight: AtomicU64, } +/// Test-only snapshot of the first-audit scheduler counters. +/// +/// Lets e2e tests assert on the scheduler's decisions (e.g. that a +/// self-targeting monetized pin was dropped and never launched) instead of +/// scraping log lines. +#[cfg(any(test, feature = "test-utils"))] +#[derive(Debug, Clone, Copy)] +pub struct FirstAuditStats { + /// Events ingested from the monetized-pin channel. + pub received: u64, + /// Events accepted into the pending first-audit queue. + pub queued: u64, + /// Events dropped because they targeted the local peer. + pub self_target_skipped: u64, + /// Audits launched. + pub launched: u64, + /// Launched audits that passed. + pub passed: u64, + /// Launched audits that timed out (non-response lane). + pub timed_out: u64, + /// Launched audits that ended in a confirmed failure. + pub failed: u64, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum FirstAuditTerminalOutcome { Passed, @@ -484,10 +511,20 @@ struct FirstAuditScheduler { reserved: Option, /// Alternating launch lane, flipped on every PROMOTION (real launch). oldest_first_lane: bool, + /// The local node's own peer ID. A verified payment's quote list includes + /// the node's own quote, so the verifier emits a monetized-pin event for + /// the local peer on every payment it verifies. The node cannot + /// network-audit itself (there is no dialable address for the local peer, + /// so the challenge fails instantly and is miscounted as a timeout), while + /// any other payee that verifies the same payment schedules its own first + /// audit of this node's pin — a self-dial adds no coverage either way. + /// Such an event is dropped at ingress: never queued, and hence never + /// launched nor marked first-audited. + self_peer: PeerId, } impl FirstAuditScheduler { - fn new(now: Instant) -> Self { + fn new(now: Instant, self_peer: PeerId) -> Self { let cap = NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN); Self { first_audited: LruCache::new(cap), @@ -495,6 +532,7 @@ impl FirstAuditScheduler { limiter: FirstAuditLimiter::new(now), reserved: None, oldest_first_lane: false, + self_peer, } } @@ -570,15 +608,20 @@ impl FirstAuditScheduler { self.reserved.as_ref().map(|r| r.event.peer) } - /// Admit a monetized nomination into `pending`. Dropped as a duplicate if - /// already first-audited; the window screen is bypassed for the currently - /// reserved peer (so a successor is retained across the reservation, never - /// window-dropped); otherwise window-screened. Coalescing is - /// highest-count-per-peer (newest on a tie) — a lower-count successor never - /// displaces a higher-count pending pin. An incumbent that has aged past - /// the answerability horizon is dropped (and accounted as an expiry) before - /// coalescing, so a dead pin never vetoes a live nomination. + /// Admit a monetized nomination into `pending`. Dropped at ingress if it + /// targets the local peer (see [`Self::self_peer`]); dropped as a duplicate + /// if already first-audited; the window screen is bypassed for the + /// currently reserved peer (so a successor is retained across the + /// reservation, never window-dropped); otherwise window-screened. + /// Coalescing is highest-count-per-peer (newest on a tie) — a lower-count + /// successor never displaces a higher-count pending pin. An incumbent that + /// has aged past the answerability horizon is dropped (and accounted as an + /// expiry) before coalescing, so a dead pin never vetoes a live nomination. fn enqueue(&mut self, event: MonetizedPinEvent, obs: &Arc) { + if event.peer == self.self_peer { + obs.self_target_skipped.fetch_add(1, Ordering::Relaxed); + return; + } if self.first_audited.contains(&event.pin) { obs.duplicates.fetch_add(1, Ordering::Relaxed); return; @@ -1102,6 +1145,10 @@ pub struct ReplicationEngine { /// ADR-0004: receiver half of the monetized-pin channel, taken by /// `start_first_audit_drainer`. monetized_pin_rx: Option>, + /// Counters shared with the first-audit drainer task, so the scheduler's + /// decisions (queued / launched / self-target skipped / outcome) stay + /// observable from the engine after the drainer takes the receiver. + first_audit_observability: Arc, /// Shutdown token. shutdown: CancellationToken, /// Background task handles. @@ -1177,6 +1224,7 @@ impl ReplicationEngine { possession_check_rx: Some(possession_check_rx), monetized_pin_tx, monetized_pin_rx: Some(monetized_pin_rx), + first_audit_observability: Arc::new(FirstAuditObservability::default()), shutdown, task_handles: Vec::new(), }; @@ -1294,6 +1342,24 @@ impl ReplicationEngine { .await } + /// Test-only: snapshot the first-audit scheduler's counters. Lets e2e + /// tests assert what the live drainer decided for an injected + /// [`MonetizedPinEvent`] (dropped as self-target vs queued and launched). + #[cfg(any(test, feature = "test-utils"))] + #[must_use] + pub fn first_audit_stats(&self) -> FirstAuditStats { + let o = &self.first_audit_observability; + FirstAuditStats { + received: o.received.load(Ordering::Relaxed), + queued: o.queued.load(Ordering::Relaxed), + self_target_skipped: o.self_target_skipped.load(Ordering::Relaxed), + launched: o.launched.load(Ordering::Relaxed), + passed: o.passed.load(Ordering::Relaxed), + timed_out: o.timed_out.load(Ordering::Relaxed), + failed: o.failed.load(Ordering::Relaxed), + } + } + /// Test-only: run the possession check immediately for `key` against /// `peers`, bypassing the scheduler's randomised 5-15 minute settle delay. /// @@ -1566,7 +1632,8 @@ impl ReplicationEngine { lottery_attempts: Arc::clone(&self.gossip_lottery_attempts), }; let shutdown = self.shutdown.clone(); - let observability = Arc::new(FirstAuditObservability::default()); + let observability = Arc::clone(&self.first_audit_observability); + let self_peer = *self.p2p_node.peer_id(); let handle = tokio::spawn(async move { // ADR-0004 Amendment 2 (E'): the drainer-owned first-audit @@ -1574,7 +1641,7 @@ impl ReplicationEngine { // them at a fixed per-node rate, and durable suppression is stamped // only at PROMOTION (after an authoritative post-jitter answerability // + cooldown check), so a cancelled reservation leaves nothing behind. - let mut scheduler = FirstAuditScheduler::new(Instant::now()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), self_peer); // Periodic retry tick so budget/cooldown-deferred pins get retried // even when no new nomination arrives. `Skip` collapses a backlog. let mut tick = tokio::time::interval(config::FIRST_AUDIT_RETRY_INTERVAL); @@ -1750,7 +1817,7 @@ impl ReplicationEngine { ); } info!( - "First-audit scheduler summary: audit_trigger=first_monetized ingress_dropped={} received={} queued={} coalesced={} suppressed_lower={} duplicates={} capacity_evicted={} cooldown_deferred_attempts={} rate_deferred_attempts={} window_deduped={} launched={} passed={} timeout={} failed={} bootstrap_claims={} idle={} insufficient_keys={} outside_answerability_window={} pending={} oldest_pending_quote_age_ms={} inflight={} tokens={}", + "First-audit scheduler summary: audit_trigger=first_monetized ingress_dropped={} received={} queued={} coalesced={} suppressed_lower={} duplicates={} capacity_evicted={} self_target_skipped={} cooldown_deferred_attempts={} rate_deferred_attempts={} window_deduped={} launched={} passed={} timeout={} failed={} bootstrap_claims={} idle={} insufficient_keys={} outside_answerability_window={} pending={} oldest_pending_quote_age_ms={} inflight={} tokens={}", FIRST_AUDIT_INGRESS_DROPPED.load(Ordering::Relaxed), observability.received.load(Ordering::Relaxed), observability.queued.load(Ordering::Relaxed), @@ -1758,6 +1825,7 @@ impl ReplicationEngine { observability.suppressed_lower.load(Ordering::Relaxed), observability.duplicates.load(Ordering::Relaxed), observability.capacity_evicted.load(Ordering::Relaxed), + observability.self_target_skipped.load(Ordering::Relaxed), observability.cooldown_deferred_attempts.load(Ordering::Relaxed), observability.rate_deferred_attempts.load(Ordering::Relaxed), observability.window_deduped.load(Ordering::Relaxed), @@ -6497,7 +6565,7 @@ mod tests { #[test] fn first_audit_stale_incumbent_does_not_suppress_live_lower_count() { let obs = Arc::new(FirstAuditObservability::default()); - let mut scheduler = FirstAuditScheduler::new(Instant::now()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); let peer = test_peer(1); let dead_quote = SystemTime::now() @@ -6546,7 +6614,7 @@ mod tests { #[test] fn first_audit_live_incumbent_still_suppresses_lower_count() { let obs = Arc::new(FirstAuditObservability::default()); - let mut scheduler = FirstAuditScheduler::new(Instant::now()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); let peer = test_peer(1); let live_high = MonetizedPinEvent { @@ -6582,7 +6650,7 @@ mod tests { fn first_audit_sweep_expired_collects_dead_entries_without_tokens() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); let dead_quote = SystemTime::now() .checked_sub(GOSSIP_ANSWERABILITY_TTL) @@ -6639,7 +6707,7 @@ mod tests { fn first_audit_answerability_cancel_is_state_neutral_and_retains_successor() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); // A pre-existing window sentinel for an UNRELATED peer must survive the // cancel byte-for-byte (cancel never touches `recent`). @@ -6784,7 +6852,7 @@ mod tests { fn first_audit_cooldown_race_requeue_preserves_newer_successor() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); let peer = test_peer(1); // Reserve A. @@ -6835,7 +6903,7 @@ mod tests { #[test] fn first_audit_oldest_pending_quote_age_tracks_the_oldest() { let now = SystemTime::now(); - let mut scheduler = FirstAuditScheduler::new(Instant::now()); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), test_peer(99)); assert_eq!( scheduler.oldest_pending_quote_age_ms(now), 0, @@ -6862,7 +6930,7 @@ mod tests { ); // A future-dated quote (clock skew) saturates to zero, never panics. - let mut skewed = FirstAuditScheduler::new(Instant::now()); + let mut skewed = FirstAuditScheduler::new(Instant::now(), test_peer(99)); let _ = coalesce_first_audit_event( &mut skewed.pending, MonetizedPinEvent { @@ -6964,7 +7032,7 @@ mod tests { fn first_audit_cooldown_race_requeue_counts_capacity_eviction() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); // Pending capacity 1 so a requeue of a different peer must evict. scheduler.pending = LruCache::new(NonZeroUsize::new(1).unwrap()); let reserved_peer = test_peer(1); @@ -7021,7 +7089,7 @@ mod tests { fn first_audit_pending_lower_count_does_not_replace_higher() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); let peer = test_peer(1); // Inflated (high-count) sidecar pin lands in pending. @@ -7077,7 +7145,7 @@ mod tests { fn first_audit_cooldown_race_requeues_reserved_higher_over_lower_successor() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); let peer = test_peer(1); // Reserve the inflated pin. @@ -7156,7 +7224,7 @@ mod tests { fn first_audit_lane_alternates_across_promotions() { let obs = Arc::new(FirstAuditObservability::default()); let mono = Instant::now(); - let mut scheduler = FirstAuditScheduler::new(mono); + let mut scheduler = FirstAuditScheduler::new(mono, test_peer(99)); let mut cooldown: HashMap = HashMap::new(); // Two distinct peers; the newest-inserted is the MRU (newest lane end). @@ -7219,6 +7287,44 @@ mod tests { assert!(!first_audit_count_jump(u32::MAX, u32::MAX)); } + /// A verified payment's quote list includes the local node's own quote, so + /// the verifier emits a monetized-pin event for the local peer on every + /// payment it verifies. The node cannot network-audit itself, so the + /// scheduler must drop such an event at ingress: never queued, and hence + /// never launched nor marked first-audited. + #[test] + fn first_audit_queue_drops_self_targeting_events() { + let obs = Arc::new(FirstAuditObservability::default()); + let self_peer = test_peer(1); + let mut scheduler = FirstAuditScheduler::new(Instant::now(), self_peer); + let self_event = MonetizedPinEvent { + peer: self_peer, + pin: [7; 32], + key_count: 1, + quote_ts: SystemTime::now(), + }; + + scheduler.enqueue(self_event, &obs); + assert_eq!(obs.self_target_skipped.load(Ordering::Relaxed), 1); + assert_eq!(obs.queued.load(Ordering::Relaxed), 0); + assert_eq!(scheduler.pending_len(), 0, "self-target must never queue"); + + // A remote peer's event still queues normally under the same filter. + let remote_event = MonetizedPinEvent { + peer: test_peer(2), + pin: [8; 32], + ..self_event + }; + scheduler.enqueue(remote_event, &obs); + assert_eq!(obs.queued.load(Ordering::Relaxed), 1); + assert_eq!(scheduler.pending_len(), 1); + assert_eq!( + obs.self_target_skipped.load(Ordering::Relaxed), + 1, + "remote event must not count as a self-target skip" + ); + } + #[test] fn fresh_offer_runs_store_admission_payment_checks() { let context = fresh_offer_payment_context(); diff --git a/tests/e2e/subtree_audit_testnet.rs b/tests/e2e/subtree_audit_testnet.rs index c21dce66..e13a2370 100644 --- a/tests/e2e/subtree_audit_testnet.rs +++ b/tests/e2e/subtree_audit_testnet.rs @@ -15,9 +15,32 @@ #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] +use std::time::{Duration, SystemTime}; + use super::TestHarness; use ant_node::replication::audit::AuditTickResult; +use ant_node::replication::{FirstAuditStats, MonetizedPinEvent, ReplicationEngine}; use serial_test::serial; +use tokio::time::sleep; + +/// Poll `engine`'s first-audit scheduler counters until `done(stats)` holds or +/// `deadline` elapses; returns the last snapshot either way. The drainer is a +/// live background task, so tests observe it converge instead of sleeping a +/// fixed amount. +async fn wait_for_first_audit_stats( + engine: &ReplicationEngine, + deadline: Duration, + done: impl Fn(&FirstAuditStats) -> bool, +) -> FirstAuditStats { + let start = std::time::Instant::now(); + loop { + let stats = engine.first_audit_stats(); + if done(&stats) || start.elapsed() >= deadline { + return stats; + } + sleep(Duration::from_millis(200)).await; + } +} /// Store the same `n` chunks on both `a` (the audited holder) and `b` (the /// auditor — NOT because verification needs them: round 2 demands the bytes @@ -159,6 +182,125 @@ async fn data_deleting_node_fails_subtree_audit() { harness.teardown().await.expect("teardown"); } +/// SELF FILTER: a monetized-pin event targeting the node's own peer ID (the +/// payment verifier emits one for every payment it verifies, because the +/// node's own quote is in the payment's quote list) must be dropped by the +/// live first-audit drainer at ingress — never queued, never launched — and +/// counted as `self_target_skipped`. +/// +/// Without the filter this exact scenario launches an audit the node addresses +/// to itself; the dial fails instantly with `Peer not found` and is miscounted +/// as a subtree audit timeout. +#[tokio::test] +#[serial] +async fn first_audit_drainer_drops_self_targeting_monetized_pin() { + let harness = TestHarness::setup_small().await.expect("setup"); + + let node = harness.test_node(4).expect("node"); + let engine = node.replication_engine.as_ref().expect("engine"); + let self_peer = *node.p2p_node.as_ref().expect("p2p").peer_id(); + + let before = engine.first_audit_stats(); + assert_eq!(before.received, 0, "no events expected before injection"); + + engine + .monetized_pin_sender() + .try_send(MonetizedPinEvent { + peer: self_peer, + pin: [0x42; 32], + key_count: 8, + quote_ts: SystemTime::now(), + }) + .expect("drainer alive"); + + let stats = wait_for_first_audit_stats(engine, Duration::from_secs(10), |s| { + s.received >= 1 && s.self_target_skipped >= 1 + }) + .await; + + assert_eq!(stats.received, 1, "drainer must have ingested the event"); + assert_eq!( + stats.self_target_skipped, 1, + "a self-targeting event must be counted as skipped, got {stats:?}" + ); + assert_eq!( + stats.queued, 0, + "a self-targeting event must never enter the pending queue, got {stats:?}" + ); + assert_eq!( + stats.launched, 0, + "the scheduler must never launch an audit against the local peer, got {stats:?}" + ); + + harness.teardown().await.expect("teardown"); +} + +/// REMOTE STILL AUDITED: the same injection for a REMOTE peer's real +/// commitment passes through the self filter, gets queued, launches over the +/// live wire, and completes as a passed first audit — proving the filter does +/// not suppress the deterministic first audit of legitimate monetized pins. +#[tokio::test] +#[serial] +async fn first_audit_drainer_launches_and_passes_remote_monetized_pin() { + let harness = TestHarness::setup_small().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let (a_idx, b_idx) = (1, 2); + commit_and_seed(&harness, a_idx, b_idx, 64).await; + + let a = harness.test_node(a_idx).expect("a"); + let a_peer = *a.p2p_node.as_ref().expect("a p2p").peer_id(); + let a_built = a + .replication_engine + .as_ref() + .expect("a engine") + .commitment_state() + .current() + .expect("a has a commitment"); + let (a_pin, a_key_count) = (a_built.hash(), a_built.commitment().key_count); + + let b_engine = harness + .test_node(b_idx) + .expect("b") + .replication_engine + .as_ref() + .expect("b engine"); + + b_engine + .monetized_pin_sender() + .try_send(MonetizedPinEvent { + peer: a_peer, + pin: a_pin, + key_count: a_key_count, + quote_ts: SystemTime::now(), + }) + .expect("drainer alive"); + + let stats = wait_for_first_audit_stats(b_engine, Duration::from_secs(120), |s| { + s.passed >= 1 || s.timed_out >= 1 || s.failed >= 1 + }) + .await; + + assert_eq!( + stats.self_target_skipped, 0, + "a remote event must not be misfiltered, got {stats:?}" + ); + assert_eq!( + stats.queued, 1, + "remote event must be queued, got {stats:?}" + ); + assert_eq!( + stats.launched, 1, + "remote event must launch an audit, got {stats:?}" + ); + assert_eq!( + stats.passed, 1, + "the live-wire first audit of an honest holder must pass, got {stats:?}" + ); + + harness.teardown().await.expect("teardown"); +} + /// NO FALSE POSITIVE: auditing an honest node repeatedly (different nonces) /// never produces a confirmed failure. #[tokio::test]