From e444952ae2667ed2b0b0a88dd8f0be72958abe28 Mon Sep 17 00:00:00 2001 From: grumbach Date: Thu, 23 Jul 2026 12:51:05 +0900 Subject: [PATCH 1/2] fix(replication): drop self-targeting monetized first-audit events The payment verifier walks every quote in a verified payment, and the local node's own quote is among them, so each verified payment emitted a monetized-pin event targeting the local peer. The first-audit scheduler launched these audits without a self check, so the node dialed itself, which fails instantly (no dialable address for the local peer) and was miscounted as a subtree audit timeout. On 400-node testnets these self-targets were ~8% of all first-monetized audit starts. Drop self-targets at queue ingress: never queued, never launched, never marked first-audited, counted under a new self_target_skipped stat in the scheduler summary line. The pin still receives its deterministic first audit from the payment's other payees, which schedule their own audits of this node. --- src/replication/mod.rs | 86 ++++++++++++++++++++++++++++++++++++++---- 1 file changed, 79 insertions(+), 7 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 89badb4f..d812f38d 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -88,6 +88,7 @@ struct FirstAuditObservability { coalesced: AtomicU64, duplicates: AtomicU64, capacity_evicted: AtomicU64, + self_target_skipped: AtomicU64, cooldown_deferred_attempts: AtomicU64, launched: AtomicU64, passed: AtomicU64, @@ -145,15 +146,33 @@ fn first_audit_terminal_outcome(result: &AuditTickResult) -> FirstAuditTerminalO enum FirstAuditQueueOutcome { Queued, Coalesced, - CapacityEvicted { peer: PeerId, pin: [u8; 32] }, + CapacityEvicted { + peer: PeerId, + pin: [u8; 32], + }, + /// The event targets the local peer itself — dropped, never queued. + SelfTargetSkipped, } /// Insert newest-per-peer work while exposing the bounded LRU's otherwise /// silent capacity eviction. This preserves `LruCache::put` semantics exactly. +/// +/// An event targeting the local peer is dropped instead of queued: the +/// verifier walks every quote in a verified payment and the node's own quote +/// is one of them, so each verified payment surfaces a self-targeting event. +/// The node cannot audit itself over the network (there is no dialable address +/// for the local peer, so the challenge fails instantly and is miscounted as a +/// timeout), and the pin still receives its deterministic first audit from the +/// payment's other payees. A self-target is therefore never queued — and hence +/// never launched nor marked first-audited. fn queue_first_audit_event( pending: &mut LruCache, event: MonetizedPinEvent, + self_peer: &PeerId, ) -> FirstAuditQueueOutcome { + if event.peer == *self_peer { + return FirstAuditQueueOutcome::SelfTargetSkipped; + } match pending.push(event.peer, event) { None => FirstAuditQueueOutcome::Queued, Some((replaced_peer, _)) if replaced_peer == event.peer => { @@ -911,6 +930,7 @@ impl ReplicationEngine { }; let shutdown = self.shutdown.clone(); let observability = Arc::new(FirstAuditObservability::default()); + let self_peer = *self.p2p_node.peer_id(); let handle = tokio::spawn(async move { // Bounded dedup of pins that have ALREADY been given their @@ -961,7 +981,7 @@ impl ReplicationEngine { e.peer, hex::encode(e.pin), e.key_count, pending.len() ); } else { - match queue_first_audit_event(&mut pending, e) { + match queue_first_audit_event(&mut pending, e, &self_peer) { FirstAuditQueueOutcome::Queued => { observability.queued.fetch_add(1, Ordering::Relaxed); debug!( @@ -986,6 +1006,15 @@ impl ReplicationEngine { hex::encode(pin), e.peer, hex::encode(e.pin), pending.len() ); } + FirstAuditQueueOutcome::SelfTargetSkipped => { + observability + .self_target_skipped + .fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=self_target_skipped peer={} pin={} key_count={} pending={}", + e.peer, hex::encode(e.pin), e.key_count, pending.len() + ); + } } } let mut drained = 1usize; @@ -998,7 +1027,7 @@ impl ReplicationEngine { .duplicates .fetch_add(1, Ordering::Relaxed); } else { - match queue_first_audit_event(&mut pending, e) { + match queue_first_audit_event(&mut pending, e, &self_peer) { FirstAuditQueueOutcome::Queued => { observability .queued @@ -1017,6 +1046,11 @@ impl ReplicationEngine { .capacity_evicted .fetch_add(1, Ordering::Relaxed); } + FirstAuditQueueOutcome::SelfTargetSkipped => { + observability + .self_target_skipped + .fetch_add(1, Ordering::Relaxed); + } } } drained += 1; @@ -1034,12 +1068,13 @@ impl ReplicationEngine { if last_summary.elapsed() >= config::FIRST_AUDIT_SUMMARY_INTERVAL { info!( - "First-audit scheduler summary: audit_trigger=first_monetized received={} queued={} coalesced={} duplicates={} capacity_evicted={} cooldown_deferred_attempts={} launched={} passed={} timeout={} failed={} bootstrap_claims={} idle={} insufficient_keys={} outside_answerability_window={} pending={} inflight={}", + "First-audit scheduler summary: audit_trigger=first_monetized received={} queued={} coalesced={} duplicates={} capacity_evicted={} self_target_skipped={} cooldown_deferred_attempts={} launched={} passed={} timeout={} failed={} bootstrap_claims={} idle={} insufficient_keys={} outside_answerability_window={} pending={} inflight={}", observability.received.load(Ordering::Relaxed), observability.queued.load(Ordering::Relaxed), observability.coalesced.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.launched.load(Ordering::Relaxed), observability.passed.load(Ordering::Relaxed), @@ -5553,6 +5588,7 @@ mod tests { #[test] fn first_audit_queue_exposes_coalescing_and_capacity_eviction() { let mut pending = LruCache::new(NonZeroUsize::new(1).unwrap()); + let self_peer = test_peer(9); let first = MonetizedPinEvent { peer: test_peer(1), pin: [1; 32], @@ -5570,15 +5606,15 @@ mod tests { }; assert_eq!( - queue_first_audit_event(&mut pending, first), + queue_first_audit_event(&mut pending, first, &self_peer), FirstAuditQueueOutcome::Queued ); assert_eq!( - queue_first_audit_event(&mut pending, replacement), + queue_first_audit_event(&mut pending, replacement, &self_peer), FirstAuditQueueOutcome::Coalesced ); assert_eq!( - queue_first_audit_event(&mut pending, other_peer), + queue_first_audit_event(&mut pending, other_peer, &self_peer), FirstAuditQueueOutcome::CapacityEvicted { peer: first.peer, pin: replacement.pin, @@ -5591,6 +5627,42 @@ mod tests { ); } + /// 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 mut pending = LruCache::new(NonZeroUsize::new(4).unwrap()); + let self_peer = test_peer(1); + let self_event = MonetizedPinEvent { + peer: self_peer, + pin: [7; 32], + key_count: 1, + quote_ts: SystemTime::now(), + }; + + assert_eq!( + queue_first_audit_event(&mut pending, self_event, &self_peer), + FirstAuditQueueOutcome::SelfTargetSkipped + ); + assert!(pending.is_empty()); + + // 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 + }; + assert_eq!( + queue_first_audit_event(&mut pending, remote_event, &self_peer), + FirstAuditQueueOutcome::Queued + ); + assert_eq!(pending.len(), 1); + assert!(pending.peek(&self_peer).is_none()); + } + #[test] fn fresh_offer_runs_store_admission_payment_checks() { let context = fresh_offer_payment_context(); From fb64e5b05224d7574dad4c28557058653863903d Mon Sep 17 00:00:00 2001 From: grumbach Date: Thu, 23 Jul 2026 13:17:30 +0900 Subject: [PATCH 2/2] test(replication): prove the first-audit self filter on a live testnet Add a test-utils-gated first_audit_stats() snapshot of the scheduler counters (previously created inside the drainer task and unobservable from tests), and two e2e tests against a running multi-node testnet: - a self-targeting monetized-pin event injected into the live drainer is ingested and dropped (received=1, self_target_skipped=1, queued=0, launched=0); - a remote peer's real commitment pin passes the filter and completes a passing live-wire subtree audit (queued=1, launched=1, passed=1). With the guard ablated the self event reproduces the production bug exactly (queued=1, launched=1, timed_out=1: the node dials itself and the instant failure is miscounted as an audit timeout), so both the unit and e2e layers go red on a regression. Also soften the queue helper's comment: other payees' first audits of the pin are best-effort, not guaranteed. --- src/replication/mod.rs | 54 ++++++++++- tests/e2e/subtree_audit_testnet.rs | 142 +++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+), 3 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index d812f38d..6967341b 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -101,6 +101,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, @@ -162,8 +186,9 @@ enum FirstAuditQueueOutcome { /// is one of them, so each verified payment surfaces a self-targeting event. /// The node cannot audit itself over the network (there is no dialable address /// for the local peer, so the challenge fails instantly and is miscounted as a -/// timeout), and the pin still receives its deterministic first audit from the -/// payment's other payees. A self-target is therefore never queued — and hence +/// timeout), while any other payee that verifies the same payment schedules its +/// own deterministic first audit of this node's pin — a self-dial adds no +/// coverage either way. A self-target is therefore never queued — and hence /// never launched nor marked first-audited. fn queue_first_audit_event( pending: &mut LruCache, @@ -472,6 +497,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. @@ -542,6 +571,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(), }; @@ -658,6 +688,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. /// @@ -929,7 +977,7 @@ impl ReplicationEngine { cooldown: Arc::clone(&self.audit_on_gossip_cooldown), }; 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 { diff --git a/tests/e2e/subtree_audit_testnet.rs b/tests/e2e/subtree_audit_testnet.rs index c21dce66..33f6ec40 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() + .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() + .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]