fix(replication): stop the first-audit scheduler from targeting the local peer

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,
@@ -100,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,
@@ -145,15 +170,34 @@ 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), 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<PeerId, MonetizedPinEvent>,
     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 => {
@@ -453,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<mpsc::UnboundedReceiver<MonetizedPinEvent>>,
+    /// 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<FirstAuditObservability>,
     /// Shutdown token.
     shutdown: CancellationToken,
     /// Background task handles.
@@ -523,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(),
         };
@@ -639,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.
     ///
@@ -910,7 +977,8 @@ 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 {
             // Bounded dedup of pins that have ALREADY been given their
@@ -961,7 +1029,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 +1054,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 +1075,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 +1094,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 +1116,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 +5636,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 +5654,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 +5675,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();