From cceed19763e8c4603b7b72e660cea3cccf077940 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 12 Jul 2026 00:21:24 +0100 Subject: [PATCH 1/4] feat(replication): add first-audit scheduler observability --- src/replication/config.rs | 6 ++ src/replication/mod.rs | 188 ++++++++++++++++++++++++++++++++++++-- 2 files changed, 186 insertions(+), 8 deletions(-) diff --git a/src/replication/config.rs b/src/replication/config.rs index f1544664..cb56905d 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -390,6 +390,12 @@ pub const AUDIT_ON_GOSSIP_COOLDOWN_SECS: u64 = 30 * 60; /// re-checks a small per-peer map, so the tick is cheap. pub const FIRST_AUDIT_RETRY_INTERVAL: Duration = Duration::from_secs(60); +/// Interval for the cumulative first-audit scheduler observability summary. +/// +/// Deliberately low frequency: this is intended for fleet-level Elasticsearch +/// aggregation without recreating the high-volume logging load it measures. +pub const FIRST_AUDIT_SUMMARY_INTERVAL: Duration = Duration::from_secs(5 * 60); + /// ADR-0004: max monetized-pin events the first-audit drainer drains from its /// channel per wake before it must run the audit-launch phase. /// diff --git a/src/replication/mod.rs b/src/replication/mod.rs index bd827210..7b1edccb 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -36,6 +36,7 @@ pub mod types; use std::collections::{HashMap, HashSet}; use std::num::NonZeroUsize; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; @@ -78,6 +79,55 @@ use crate::storage::LmdbStorage; use saorsa_core::identity::{NodeIdentity, PeerId}; use saorsa_core::{DhtNetworkEvent, P2PEvent, P2PNode, TrustEvent}; +#[derive(Default)] +struct FirstAuditObservability { + received: AtomicU64, + queued: AtomicU64, + coalesced: AtomicU64, + duplicates: AtomicU64, + cooldown_deferred: AtomicU64, + launched: AtomicU64, + passed: AtomicU64, + timed_out: AtomicU64, + failed: AtomicU64, + expired: AtomicU64, + inflight: AtomicU64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FirstAuditTerminalOutcome { + Passed, + Timeout, + Failed, +} + +impl FirstAuditTerminalOutcome { + const fn as_str(self) -> &'static str { + match self { + Self::Passed => "passed", + Self::Timeout => "timeout", + Self::Failed => "failed", + } + } +} + +fn first_audit_terminal_outcome(result: &AuditTickResult) -> FirstAuditTerminalOutcome { + match result { + AuditTickResult::Passed { .. } => FirstAuditTerminalOutcome::Passed, + AuditTickResult::Failed { + evidence: + FailureEvidence::AuditFailure { + reason: AuditFailureReason::Timeout, + .. + }, + } => FirstAuditTerminalOutcome::Timeout, + AuditTickResult::Failed { .. } + | AuditTickResult::BootstrapClaim { .. } + | AuditTickResult::Idle + | AuditTickResult::InsufficientKeys => FirstAuditTerminalOutcome::Failed, + } +} + // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- @@ -816,6 +866,7 @@ impl ReplicationEngine { cooldown: Arc::clone(&self.audit_on_gossip_cooldown), }; let shutdown = self.shutdown.clone(); + let observability = Arc::new(FirstAuditObservability::default()); let handle = tokio::spawn(async move { // Bounded dedup of pins that have ALREADY been given their @@ -842,6 +893,7 @@ impl ReplicationEngine { // once; `Skip` so a backlog of missed ticks collapses to one. let mut tick = tokio::time::interval(config::FIRST_AUDIT_RETRY_INTERVAL); tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut last_summary = Instant::now(); loop { // Wake on: shutdown, a new monetized pin, OR a periodic tick so // pending (cooldown-blocked) pins get retried once their window @@ -850,6 +902,7 @@ impl ReplicationEngine { () = shutdown.cancelled() => break, event = rx.recv() => match event { Some(e) => { + observability.received.fetch_add(1, Ordering::Relaxed); // Newest-per-peer: a fresher pin replaces the older one, // BUT only if it is not already first-audited — an // already-audited duplicate must never overwrite an @@ -857,15 +910,38 @@ impl ReplicationEngine { // be dropped as "done" and the unaudited pin lost). Cap // the per-wake batch drain (FIRST_AUDIT_DRAIN_BATCH) so a // sustained flood can't starve the audit-launch phase. - if !first_audited.contains(&e.pin) { - pending.put(e.peer, e); + if first_audited.contains(&e.pin) { + observability.duplicates.fetch_add(1, Ordering::Relaxed); + } else if pending.put(e.peer, e).is_some() { + observability.coalesced.fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=coalesced peer={} pin={} key_count={} pending={}", + e.peer, hex::encode(e.pin), e.key_count, pending.len() + ); + } else { + observability.queued.fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=queued peer={} pin={} key_count={} pending={}", + e.peer, hex::encode(e.pin), e.key_count, pending.len() + ); } let mut drained = 1usize; while drained < config::FIRST_AUDIT_DRAIN_BATCH { match rx.try_recv() { Ok(e) => { - if !first_audited.contains(&e.pin) { - pending.put(e.peer, e); + observability.received.fetch_add(1, Ordering::Relaxed); + if first_audited.contains(&e.pin) { + observability + .duplicates + .fetch_add(1, Ordering::Relaxed); + } else if pending.put(e.peer, e).is_some() { + observability + .coalesced + .fetch_add(1, Ordering::Relaxed); + } else { + observability + .queued + .fetch_add(1, Ordering::Relaxed); } drained += 1; } @@ -880,6 +956,25 @@ impl ReplicationEngine { }; let _ = drained_new; + if last_summary.elapsed() >= config::FIRST_AUDIT_SUMMARY_INTERVAL { + info!( + "First-audit scheduler summary: audit_trigger=first_monetized received={} queued={} coalesced={} duplicates={} cooldown_deferred={} launched={} passed={} timed_out={} failed={} expired={} pending={} inflight={}", + observability.received.load(Ordering::Relaxed), + observability.queued.load(Ordering::Relaxed), + observability.coalesced.load(Ordering::Relaxed), + observability.duplicates.load(Ordering::Relaxed), + observability.cooldown_deferred.load(Ordering::Relaxed), + observability.launched.load(Ordering::Relaxed), + observability.passed.load(Ordering::Relaxed), + observability.timed_out.load(Ordering::Relaxed), + observability.failed.load(Ordering::Relaxed), + observability.expired.load(Ordering::Relaxed), + pending.len(), + observability.inflight.load(Ordering::Relaxed), + ); + last_summary = Instant::now(); + } + if pending.is_empty() { continue; } @@ -897,6 +992,7 @@ impl ReplicationEngine { for (peer, event) in snapshot { // Dedup: a pin already first-audited is dropped (done). if first_audited.contains(&event.pin) { + observability.duplicates.fetch_add(1, Ordering::Relaxed); continue; } // ADR-0004 A1 (guardrail A): only first-audit a pin whose SIGNED @@ -907,8 +1003,10 @@ impl ReplicationEngine { // payment (quote_ts ≈ now); a skipped pin can still be // gossip-lottery audited. if !quote_within_audit_window(event.quote_ts, SystemTime::now()) { + observability.expired.fetch_add(1, Ordering::Relaxed); debug!( - "First audit skipped for {peer}: quote_ts outside the answerability window" + "First-audit scheduler: audit_trigger=first_monetized outcome=expired peer={peer} pin={} key_count={} pending={}", + hex::encode(event.pin), event.key_count, pending.len() ); continue; } @@ -924,13 +1022,29 @@ impl ReplicationEngine { let mut map = gossip_audit.cooldown.write().await; if !cooldown_allows_audit(&mut map, &peer, now) { pending.put(peer, event); + observability + .cooldown_deferred + .fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=cooldown_deferred peer={peer} pin={} key_count={} pending={}", + hex::encode(event.pin), event.key_count, pending.len() + ); continue; } } // Audit is launching: now mark the pin first-audited. first_audited.put(event.pin, ()); + observability.launched.fetch_add(1, Ordering::Relaxed); + observability.inflight.fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=launched peer={peer} pin={} key_count={} pending={} inflight={}", + hex::encode(event.pin), event.key_count, pending.len(), + observability.inflight.load(Ordering::Relaxed) + ); let trigger = gossip_audit.clone(); + let audit_observability = Arc::clone(&observability); tokio::spawn(async move { + let started = Instant::now(); let credit = storage_commitment_audit::AuditCredit { recent_provers: &trigger.recent_provers, }; @@ -943,6 +1057,28 @@ impl ReplicationEngine { Some(&credit), ) .await; + let outcome = first_audit_terminal_outcome(&result); + match outcome { + FirstAuditTerminalOutcome::Passed => { + audit_observability.passed.fetch_add(1, Ordering::Relaxed); + } + FirstAuditTerminalOutcome::Timeout => { + audit_observability + .timed_out + .fetch_add(1, Ordering::Relaxed); + } + FirstAuditTerminalOutcome::Failed => { + audit_observability.failed.fetch_add(1, Ordering::Relaxed); + } + } + audit_observability.inflight.fetch_sub(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome={} peer={} pin={} key_count={} elapsed_ms={} inflight={}", + outcome.as_str(), + event.peer, hex::encode(event.pin), event.key_count, + started.elapsed().as_millis(), + audit_observability.inflight.load(Ordering::Relaxed) + ); handle_subtree_audit_result( &result, &trigger.p2p_node, @@ -5057,12 +5193,14 @@ mod tests { use super::{ apply_audit_failure_credit_revocation, audit_failure_clears_bootstrap_claim, audit_failure_revokes_holder_credit, audit_launch_decision, config, cooldown_allows_audit, - first_failed_key_label, fresh_offer_payment_context, paid_notify_payment_context, - quote_within_audit_window, MONETIZED_AUDIT_SKEW_MARGIN, + first_audit_terminal_outcome, first_failed_key_label, fresh_offer_payment_context, + paid_notify_payment_context, quote_within_audit_window, FirstAuditTerminalOutcome, + MONETIZED_AUDIT_SKEW_MARGIN, }; use crate::payment::VerificationContext; + use crate::replication::audit::AuditTickResult; use crate::replication::recent_provers::RecentProvers; - use crate::replication::types::AuditFailureReason; + use crate::replication::types::{AuditFailureReason, AuditFailureSummary, FailureEvidence}; use saorsa_core::identity::PeerId; use std::collections::HashMap; use std::time::Duration; @@ -5081,6 +5219,40 @@ mod tests { k } + #[test] + fn first_audit_terminal_outcomes_are_stable() { + let peer = test_peer(1); + let passed = AuditTickResult::Passed { + challenged_peer: peer, + keys_checked: 1, + }; + let timed_out = AuditTickResult::Failed { + evidence: FailureEvidence::AuditFailure { + challenge_id: 1, + challenged_peer: peer, + confirmed_failed_keys: vec![test_key(1)], + summary: AuditFailureSummary::default(), + reason: AuditFailureReason::Timeout, + }, + }; + + assert_eq!( + first_audit_terminal_outcome(&passed), + FirstAuditTerminalOutcome::Passed + ); + assert_eq!( + first_audit_terminal_outcome(&timed_out), + FirstAuditTerminalOutcome::Timeout + ); + assert_eq!( + first_audit_terminal_outcome(&AuditTickResult::Idle), + FirstAuditTerminalOutcome::Failed + ); + assert_eq!(FirstAuditTerminalOutcome::Passed.as_str(), "passed"); + assert_eq!(FirstAuditTerminalOutcome::Timeout.as_str(), "timeout"); + assert_eq!(FirstAuditTerminalOutcome::Failed.as_str(), "failed"); + } + #[test] fn fresh_offer_runs_client_put_payment_checks() { assert_eq!( From 82a90017e37aab57509fd5bbbee81bbec1dc9bf7 Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 12 Jul 2026 00:25:36 +0100 Subject: [PATCH 2/4] fix(replication): correct first-audit telemetry taxonomy --- src/replication/mod.rs | 197 +++++++++++++++++++++++++++++++++++------ 1 file changed, 168 insertions(+), 29 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 7b1edccb..18b7a432 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -85,12 +85,16 @@ struct FirstAuditObservability { queued: AtomicU64, coalesced: AtomicU64, duplicates: AtomicU64, - cooldown_deferred: AtomicU64, + capacity_evicted: AtomicU64, + cooldown_deferred_attempts: AtomicU64, launched: AtomicU64, passed: AtomicU64, timed_out: AtomicU64, failed: AtomicU64, - expired: AtomicU64, + bootstrap_claims: AtomicU64, + idle: AtomicU64, + insufficient_keys: AtomicU64, + outside_answerability_window: AtomicU64, inflight: AtomicU64, } @@ -99,6 +103,9 @@ enum FirstAuditTerminalOutcome { Passed, Timeout, Failed, + BootstrapClaim, + Idle, + InsufficientKeys, } impl FirstAuditTerminalOutcome { @@ -107,6 +114,9 @@ impl FirstAuditTerminalOutcome { Self::Passed => "passed", Self::Timeout => "timeout", Self::Failed => "failed", + Self::BootstrapClaim => "bootstrap_claim", + Self::Idle => "idle", + Self::InsufficientKeys => "insufficient_keys", } } } @@ -121,10 +131,35 @@ fn first_audit_terminal_outcome(result: &AuditTickResult) -> FirstAuditTerminalO .. }, } => FirstAuditTerminalOutcome::Timeout, - AuditTickResult::Failed { .. } - | AuditTickResult::BootstrapClaim { .. } - | AuditTickResult::Idle - | AuditTickResult::InsufficientKeys => FirstAuditTerminalOutcome::Failed, + AuditTickResult::Failed { .. } => FirstAuditTerminalOutcome::Failed, + AuditTickResult::BootstrapClaim { .. } => FirstAuditTerminalOutcome::BootstrapClaim, + AuditTickResult::Idle => FirstAuditTerminalOutcome::Idle, + AuditTickResult::InsufficientKeys => FirstAuditTerminalOutcome::InsufficientKeys, + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum FirstAuditQueueOutcome { + Queued, + Coalesced, + CapacityEvicted { peer: PeerId, pin: [u8; 32] }, +} + +/// Insert newest-per-peer work while exposing the bounded LRU's otherwise +/// silent capacity eviction. This preserves `LruCache::put` semantics exactly. +fn queue_first_audit_event( + pending: &mut LruCache, + event: MonetizedPinEvent, +) -> FirstAuditQueueOutcome { + match pending.push(event.peer, event) { + None => FirstAuditQueueOutcome::Queued, + Some((replaced_peer, _)) if replaced_peer == event.peer => { + FirstAuditQueueOutcome::Coalesced + } + Some((evicted_peer, evicted)) => FirstAuditQueueOutcome::CapacityEvicted { + peer: evicted_peer, + pin: evicted.pin, + }, } } @@ -912,18 +947,37 @@ impl ReplicationEngine { // sustained flood can't starve the audit-launch phase. if first_audited.contains(&e.pin) { observability.duplicates.fetch_add(1, Ordering::Relaxed); - } else if pending.put(e.peer, e).is_some() { - observability.coalesced.fetch_add(1, Ordering::Relaxed); debug!( - "First-audit scheduler: audit_trigger=first_monetized outcome=coalesced peer={} pin={} key_count={} pending={}", + "First-audit scheduler: audit_trigger=first_monetized outcome=duplicate peer={} pin={} key_count={} pending={}", e.peer, hex::encode(e.pin), e.key_count, pending.len() ); } else { - observability.queued.fetch_add(1, Ordering::Relaxed); - debug!( - "First-audit scheduler: audit_trigger=first_monetized outcome=queued peer={} pin={} key_count={} pending={}", - e.peer, hex::encode(e.pin), e.key_count, pending.len() - ); + match queue_first_audit_event(&mut pending, e) { + FirstAuditQueueOutcome::Queued => { + observability.queued.fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=queued peer={} pin={} key_count={} pending={}", + e.peer, hex::encode(e.pin), e.key_count, pending.len() + ); + } + FirstAuditQueueOutcome::Coalesced => { + observability.coalesced.fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=coalesced peer={} pin={} key_count={} pending={}", + e.peer, hex::encode(e.pin), e.key_count, pending.len() + ); + } + FirstAuditQueueOutcome::CapacityEvicted { peer, pin } => { + observability.queued.fetch_add(1, Ordering::Relaxed); + observability + .capacity_evicted + .fetch_add(1, Ordering::Relaxed); + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=capacity_evicted evicted_peer={peer} evicted_pin={} replacement_peer={} replacement_pin={} pending={}", + hex::encode(pin), e.peer, hex::encode(e.pin), pending.len() + ); + } + } } let mut drained = 1usize; while drained < config::FIRST_AUDIT_DRAIN_BATCH { @@ -934,14 +988,27 @@ impl ReplicationEngine { observability .duplicates .fetch_add(1, Ordering::Relaxed); - } else if pending.put(e.peer, e).is_some() { - observability - .coalesced - .fetch_add(1, Ordering::Relaxed); } else { - observability - .queued - .fetch_add(1, Ordering::Relaxed); + match queue_first_audit_event(&mut pending, e) { + FirstAuditQueueOutcome::Queued => { + observability + .queued + .fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::Coalesced => { + observability + .coalesced + .fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::CapacityEvicted { .. } => { + observability + .queued + .fetch_add(1, Ordering::Relaxed); + observability + .capacity_evicted + .fetch_add(1, Ordering::Relaxed); + } + } } drained += 1; } @@ -958,17 +1025,21 @@ impl ReplicationEngine { if last_summary.elapsed() >= config::FIRST_AUDIT_SUMMARY_INTERVAL { info!( - "First-audit scheduler summary: audit_trigger=first_monetized received={} queued={} coalesced={} duplicates={} cooldown_deferred={} launched={} passed={} timed_out={} failed={} expired={} pending={} inflight={}", + "First-audit scheduler summary: audit_trigger=first_monetized received={} queued={} coalesced={} duplicates={} capacity_evicted={} cooldown_deferred_attempts={} launched={} passed={} timed_out={} 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.cooldown_deferred.load(Ordering::Relaxed), + observability.capacity_evicted.load(Ordering::Relaxed), + observability.cooldown_deferred_attempts.load(Ordering::Relaxed), observability.launched.load(Ordering::Relaxed), observability.passed.load(Ordering::Relaxed), observability.timed_out.load(Ordering::Relaxed), observability.failed.load(Ordering::Relaxed), - observability.expired.load(Ordering::Relaxed), + observability.bootstrap_claims.load(Ordering::Relaxed), + observability.idle.load(Ordering::Relaxed), + observability.insufficient_keys.load(Ordering::Relaxed), + observability.outside_answerability_window.load(Ordering::Relaxed), pending.len(), observability.inflight.load(Ordering::Relaxed), ); @@ -1003,9 +1074,11 @@ impl ReplicationEngine { // payment (quote_ts ≈ now); a skipped pin can still be // gossip-lottery audited. if !quote_within_audit_window(event.quote_ts, SystemTime::now()) { - observability.expired.fetch_add(1, Ordering::Relaxed); + observability + .outside_answerability_window + .fetch_add(1, Ordering::Relaxed); debug!( - "First-audit scheduler: audit_trigger=first_monetized outcome=expired peer={peer} pin={} key_count={} pending={}", + "First-audit scheduler: audit_trigger=first_monetized outcome=outside_answerability_window peer={peer} pin={} key_count={} pending={}", hex::encode(event.pin), event.key_count, pending.len() ); continue; @@ -1023,7 +1096,7 @@ impl ReplicationEngine { if !cooldown_allows_audit(&mut map, &peer, now) { pending.put(peer, event); observability - .cooldown_deferred + .cooldown_deferred_attempts .fetch_add(1, Ordering::Relaxed); debug!( "First-audit scheduler: audit_trigger=first_monetized outcome=cooldown_deferred peer={peer} pin={} key_count={} pending={}", @@ -1070,6 +1143,19 @@ impl ReplicationEngine { FirstAuditTerminalOutcome::Failed => { audit_observability.failed.fetch_add(1, Ordering::Relaxed); } + FirstAuditTerminalOutcome::BootstrapClaim => { + audit_observability + .bootstrap_claims + .fetch_add(1, Ordering::Relaxed); + } + FirstAuditTerminalOutcome::Idle => { + audit_observability.idle.fetch_add(1, Ordering::Relaxed); + } + FirstAuditTerminalOutcome::InsufficientKeys => { + audit_observability + .insufficient_keys + .fetch_add(1, Ordering::Relaxed); + } } audit_observability.inflight.fetch_sub(1, Ordering::Relaxed); debug!( @@ -5194,15 +5280,18 @@ mod tests { apply_audit_failure_credit_revocation, audit_failure_clears_bootstrap_claim, audit_failure_revokes_holder_credit, audit_launch_decision, config, cooldown_allows_audit, first_audit_terminal_outcome, first_failed_key_label, fresh_offer_payment_context, - paid_notify_payment_context, quote_within_audit_window, FirstAuditTerminalOutcome, + paid_notify_payment_context, queue_first_audit_event, quote_within_audit_window, + FirstAuditQueueOutcome, FirstAuditTerminalOutcome, MonetizedPinEvent, MONETIZED_AUDIT_SKEW_MARGIN, }; use crate::payment::VerificationContext; use crate::replication::audit::AuditTickResult; use crate::replication::recent_provers::RecentProvers; use crate::replication::types::{AuditFailureReason, AuditFailureSummary, FailureEvidence}; + use lru::LruCache; use saorsa_core::identity::PeerId; use std::collections::HashMap; + use std::num::NonZeroUsize; use std::time::Duration; use std::time::Instant; use std::time::SystemTime; @@ -5246,11 +5335,61 @@ mod tests { ); assert_eq!( first_audit_terminal_outcome(&AuditTickResult::Idle), - FirstAuditTerminalOutcome::Failed + FirstAuditTerminalOutcome::Idle ); assert_eq!(FirstAuditTerminalOutcome::Passed.as_str(), "passed"); assert_eq!(FirstAuditTerminalOutcome::Timeout.as_str(), "timeout"); assert_eq!(FirstAuditTerminalOutcome::Failed.as_str(), "failed"); + assert_eq!(FirstAuditTerminalOutcome::Idle.as_str(), "idle"); + assert_eq!( + FirstAuditTerminalOutcome::InsufficientKeys.as_str(), + "insufficient_keys" + ); + assert_eq!( + FirstAuditTerminalOutcome::BootstrapClaim.as_str(), + "bootstrap_claim" + ); + } + + #[test] + fn first_audit_queue_exposes_coalescing_and_capacity_eviction() { + let mut pending = LruCache::new(NonZeroUsize::new(1).unwrap()); + let first = MonetizedPinEvent { + peer: test_peer(1), + pin: [1; 32], + key_count: 1, + quote_ts: SystemTime::now(), + }; + let replacement = MonetizedPinEvent { + pin: [2; 32], + ..first + }; + let other_peer = MonetizedPinEvent { + peer: test_peer(2), + pin: [3; 32], + ..first + }; + + assert_eq!( + queue_first_audit_event(&mut pending, first), + FirstAuditQueueOutcome::Queued + ); + assert_eq!( + queue_first_audit_event(&mut pending, replacement), + FirstAuditQueueOutcome::Coalesced + ); + assert_eq!( + queue_first_audit_event(&mut pending, other_peer), + FirstAuditQueueOutcome::CapacityEvicted { + peer: first.peer, + pin: replacement.pin, + } + ); + assert_eq!(pending.len(), 1); + assert_eq!( + pending.peek(&other_peer.peer).map(|event| event.pin), + Some([3; 32]) + ); } #[test] From ebb9a018d4cfb526fb222c949895a565d57fddee Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 12 Jul 2026 01:06:08 +0100 Subject: [PATCH 3/4] fix(observability): align first-audit outcome labels --- src/replication/mod.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 18b7a432..62016746 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -973,7 +973,7 @@ impl ReplicationEngine { .capacity_evicted .fetch_add(1, Ordering::Relaxed); debug!( - "First-audit scheduler: audit_trigger=first_monetized outcome=capacity_evicted evicted_peer={peer} evicted_pin={} replacement_peer={} replacement_pin={} pending={}", + "First-audit scheduler: audit_trigger=first_monetized outcome=queued capacity_evicted=true evicted_peer={peer} evicted_pin={} replacement_peer={} replacement_pin={} pending={}", hex::encode(pin), e.peer, hex::encode(e.pin), pending.len() ); } @@ -1025,7 +1025,7 @@ 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={} timed_out={} 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={} 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), From 30a01fd430b5ed7404f5cd52e7d120151f82588c Mon Sep 17 00:00:00 2001 From: Hermes Agent Date: Sun, 12 Jul 2026 01:17:45 +0100 Subject: [PATCH 4/4] fix(build): gate first-audit labels with logging --- src/replication/mod.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 62016746..fba808ae 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -109,6 +109,7 @@ enum FirstAuditTerminalOutcome { } impl FirstAuditTerminalOutcome { + #[cfg(any(feature = "logging", test))] const fn as_str(self) -> &'static str { match self { Self::Passed => "passed",