diff --git a/docs/adr/ADR-0004-commitment-bound-quote-pricing.md b/docs/adr/ADR-0004-commitment-bound-quote-pricing.md index 97411d16..682ac5c0 100644 --- a/docs/adr/ADR-0004-commitment-bound-quote-pricing.md +++ b/docs/adr/ADR-0004-commitment-bound-quote-pricing.md @@ -334,3 +334,147 @@ attributing malicious silence/transient conditions network-wide is the (out-of-scope) distributed non-response problem. This supersedes the "unanswerable quoted pin is graced, never confirmed" rule and deletes `RejectKind::is_graced`; the regression tests are updated accordingly. + +--- + +## Amendment 2 (2026-07-14): first audits are bounded best-effort sampling — payments nominate, the clock launches + +The original decision made the monetized first audit deterministic: every pinned +quote in every verified client-put proof entered the first-audit queue at every +verifying storer, gated only by the per-peer 30-minute cooldown. That per-peer +gate bounds one observer against one peer but places no bound on the aggregate: +fleet-wide launch pressure scaled as +`uploads x pinned-quotes-per-proof x verifying-storers`, with hourly commitment +rotation re-arming pin-level dedup every rotation. In the v0.14.3 production +rollout this amplification saturated the audit-responder admission pools; +overflow challenges were dropped and recorded by auditors as Timeout failures +(545k Timeout errors/24h fleet-wide, ~31% slower downloads). A matched staging +ablation attributed 97.7% of storage-commitment audit traffic and 99.955% of +one-key timeout failures to this path, and a per-service concurrency cap alone +(without a launch-rate budget) reproduced the storm. Amendment 1's answerability +guarantees are unaffected; this amendment changes only who is nominated and how +fast nominations launch. + +**Nomination narrows to paid pins.** Only the candidate whose on-chain +settlement actually verified is nominated: the single-node path nominates the +settlement-verified median candidate (not the whole bundle), and the merkle +path nominates the contract-paid indices (not the whole pool). Unpaid quotes +earned nothing; their gossiped commitments remain under the ADR-0002 lottery. +The storer-side cross-check (arithmetic re-check and mismatch evidence) still +runs on every quote and candidate. + +**Launches are budgeted, not deterministic.** Each node's first-audit drainer +runs a launch limiter: a token bucket (burst 2, one token per 5 minutes — a +sustained per-node rate of 12 launches/hour, plus at most the burst of 2 in +the first hour), an in-flight cap (2), and a uniform +0-30s launch jitter so the storers of one chunk, which all verify the same +payment at the same instant, do not challenge the paid peer simultaneously. +Fleet-wide first-audit pressure is therefore `nodes x refill-rate`, independent +of upload volume. Budget deferral is penalty-free: a deferred pin stays pending +(bounded, one entry per peer) and launches when tokens allow; only audits that +actually launch have consequences. + +**Pending coalescing keeps the highest count per peer, not merely the newest.** +The per-peer pending slot retains the pin that most needs auditing — the +HIGHEST committed key count (newest on an equal-count tie). A strictly-lower- +count nomination for a peer never displaces a higher-count pending pin, and a +suppressed lower nomination does not disturb the retained pin's queue position. +Without this rule a peer could erase an inflated (audit-worthy) commitment for +the cost of one cheaper same-peer settlement, since the count-jump override +only compares against the last AUDITED count and a sidecar-only inflated pin +has no gossip-lottery backstop. The same rule governs the requeue of a +reservation that loses the shared-cooldown race. Accepted residual: if the +retained higher pin later ages out of the answerability window (un-auditable +once aged, grace removed), the single per-peer slot has already discarded the +lower fallback — a small coverage reduction versus keeping both, not a revival +of the lower-count-displacement bypass. + +**Per-peer re-audit window survives pin rotation.** After a first audit +launches at a peer, further nominations for that peer are dropped for 2 hours +(inside the 3h answerability TTL) — unless the new pin's committed key count +exceeds the audited one by more than 1.5x, which re-nominates immediately. The +jump override preserves the anti-inflation property this queue exists for: an +inflated commitment delivered only as a quote sidecar is visible to payment +verifiers alone, so no gossip-lottery audit can ever select it; a peer that +passes an audit on an honest count and then mints a much larger sidecar-only +commitment is re-audited at once. Ordinary rotations with a stable count stop +re-arming the fleet. + +**Coverage restated.** "The latest commitment earning money for a peer always +faces an audit soon" becomes: the FIRST commitment earning money for a peer, +and any later commitment whose claimed count materially jumps, faces an audit +soon (minutes, from multiple independent storers, each within its own budget); +commitments rotated without material count change rely on the ADR-0002 lottery +for re-audit. First-audit coverage is best-effort supplementary sampling under +an explicit load budget, not a per-payment guarantee. The scheduler's funnel +(received / queued / coalesced / suppressed_lower / duplicates / +capacity_evicted / window_deduped / rate_deferred / cooldown_deferred / +launched / terminal outcomes, plus tokens and in-flight gauges) is exported in +the periodic scheduler summary so this coverage is measurable in production; +`suppressed_lower` in particular surfaces attempted cheaper-pin self-erasure. + +**Every pipeline stage is bounded, and the invariants hold by construction.** +The verifier-to-drainer nomination channel is bounded (producers `try_send` +and drop on full — penalty-free), the pending set is a bounded +highest-count-per-peer LRU, and launches are token-bucketed. Durable +suppression (the first-audited pin dedup, the per-peer re-audit window, the +shared cooldown stamp, the lane flip, and the launch count) is committed ONLY +at promotion, after an authoritative answerability + cooldown check-and-stamp +taken immediately before the wire challenge; a reservation cancelled during +its jitter refunds its token, releases its in-flight slot, and leaves no +suppression behind. The answerability screen runs both as a schedule-time +prefilter (over the whole jitter horizon) and authoritatively at promotion, so +a pin can never be challenged outside its window regardless of how deferral +time, jitter, and the skew margin compose. + +**Nomination scope and known coverage limits (measured in staging).** (a) On +the single-node path only the FIRST settlement-verified median candidate is +nominated; if a client settles several tied-median candidates, gossiped extras +retain the ADR-0002 lottery but sidecar-only extra settled pins do not and are +an accepted best-effort residual. The merkle path nominates every +contract-paid index. (b) The per-peer re-audit window is stamped at LAUNCH, not +at a passing outcome: a peer that answers with a `Transient`/silence reaches +the ADR-0002 timeout lane and is not automatically re-launched by this +scheduler (automatic retries would change the load bound and need their own +policy). Both effects reduce coverage without weakening the price ceiling and +are to be quantified against the terminal-outcome counters in the matched +staging run. + +**Accepted residuals.** (1) *Budget-exhaustion starvation, mitigated by +alternating lanes:* an attacker can try to keep observers' launch budgets +saturated with newer settled decoy nominations so an older pending pin +expires un-first-audited. Stated precisely: suppressing one pin requires +outcompeting roughly 31 launch opportunities per observer over the 150-minute +eligibility window; a single decoy settlement consumes budget at every storer +of its chunk simultaneously (a cohort, not per-observer, cost), a merkle +settlement nominates every paid index, principal recycles through +sybil-controlled reward addresses (the real costs are gas, liquidity, and +close-group positioning, not the transferred amount), and a sidecar-only +target pin has NO lottery backstop. The scheduler therefore does not rely on +economics alone: consecutive LAUNCHES strictly alternate between a +newest-first and an oldest-first lane. The lane advances on every committed +launch — not per scheduling pass (most pass over an empty bucket, and a pass +is triggered by ingress, so per-pass parity would be attacker-steerable) and +not once per launching pass (a full burst spends two tokens in one pass, and +those two must alternate with each other). Fresh decoy nominations therefore +cannot capture the oldest lane, and an aging pin is served there. + +Two suppression routes remain, both requiring real settled payments at scale. +An attacker holding pre-aged pending decoys at every observer can occupy the +oldest lane too — though that lane drains them, and the per-peer re-audit +window blocks the same issuers from refreshing them. And 4096 fresher +distinct-peer nominations at an observer evict the target from that +observer's pending LRU, which is permanent for that nomination: coverage then +depends on the peer's NEXT settled payment (or, for a gossiped commitment, on +the ADR-0002 lottery). We accept this: both routes cost per-observer sybil +positioning plus a stream of genuine on-chain settlements, and neither is +free or silent — the scheduler funnel exports capacity evictions and lane +outcomes. (2) *Settlement overwrite (pre-existing, +contract-level):* the vault's `payForQuotes` unconditionally overwrites +`completedPayments[quoteHash]`, so a third party who learns a quote hash can +overwrite the record (already breaking the long-standing amount check with a +1-wei payment, on all node versions) and now equivalently the +rewards-address binding. This griefing race predates this amendment and +needs a vault-side fix (reject or accumulate on existing entries); it is +tracked as follow-up work, not a property this ADR can enforce from the node +side. diff --git a/src/payment/verifier.rs b/src/payment/verifier.rs index a48da2a3..8f7314e1 100644 --- a/src/payment/verifier.rs +++ b/src/payment/verifier.rs @@ -221,6 +221,11 @@ type CommitmentCache = Arc< /// handle without borrowing the verifier. type PinFetchNegativeCache = Arc>>; +/// Test-override shape for one on-chain settlement: the settled amount and +/// the recorded `bytes16` rewards-address prefix (`None` = matches the quote). +#[cfg(any(test, feature = "test-utils"))] +type TestSettlementOverride = (Amount, Option<[u8; 16]>); + /// Main payment verifier for ant-node. /// /// Uses: @@ -262,9 +267,13 @@ pub struct PaymentVerifier { /// Test-only override for `completedPayments(quote_hash)`. /// /// Production always queries the payment vault; unit tests use this to - /// exercise the full verifier path without starting an EVM chain. + /// exercise the full verifier path without starting an EVM chain. Maps a + /// quote hash to the settled amount plus the recorded rewards-address + /// prefix (`None` means "matches the quote", the honest-settlement + /// default). The prefix is 16 bytes because the vault packs the address + /// as `bytes16` in the `completedPayments` slot. #[cfg(any(test, feature = "test-utils"))] - test_completed_payments_override: RwLock>, + test_completed_payments_override: RwLock>, // NOTE: the test-only own-peer-id override was removed with the ADR-retired // quote-freshness/staleness gate (ADR-0004 binds price to the committed // count instead), so it no longer has any reader. @@ -289,7 +298,7 @@ pub struct PaymentVerifier { /// `None` until [`Self::attach_monetized_pin_sender`] (unit tests, or /// pre-replication startup), in which case no first audit is scheduled. monetized_pin_tx: - RwLock>>, + RwLock>>, /// Configuration. config: PaymentVerifierConfig, } @@ -427,7 +436,7 @@ impl PaymentVerifier { /// absent (unit tests / pre-replication) no first audit is scheduled. pub fn attach_monetized_pin_sender( &self, - tx: tokio::sync::mpsc::UnboundedSender, + tx: tokio::sync::mpsc::Sender, ) { *self.monetized_pin_tx.write() = Some(tx); debug!("PaymentVerifier: ADR-0004 monetized-pin sender attached"); @@ -500,12 +509,40 @@ impl PaymentVerifier { self.set_paid_quote_k_closest_for_tests(peer_ids); } - /// Test-only setter for an on-chain completed payment amount. + /// Test-only setter for an on-chain completed payment amount. The recorded + /// rewards address is treated as matching the quote (honest settlement). #[cfg(any(test, feature = "test-utils"))] pub fn set_completed_payment_for_tests(&self, quote_hash: QuoteHash, amount: Amount) { self.test_completed_payments_override .write() - .insert(quote_hash, amount); + .insert(quote_hash, (amount, None)); + } + + /// Test-only setter for an on-chain completed payment with an explicit + /// recorded rewards address, to exercise the settlement-redirect rejection. + #[cfg(any(test, feature = "test-utils"))] + pub fn set_completed_payment_with_address_for_tests( + &self, + quote_hash: QuoteHash, + amount: Amount, + rewards_address: evmlib::RewardsAddress, + ) { + self.test_completed_payments_override.write().insert( + quote_hash, + (amount, Some(Self::rewards_address_prefix(&rewards_address))), + ); + } + + /// The vault's `completedPayments` slot packs the payee as + /// `bytes16(bytes20(rewardsAddress))` — the leading 16 bytes of the + /// address (Solidity `bytesN` conversions truncate on the right). This is + /// the form the on-chain record is compared in. + fn rewards_address_prefix(rewards_address: &evmlib::RewardsAddress) -> [u8; 16] { + let mut prefix = [0u8; 16]; + if let Some(head) = rewards_address.as_slice().get(..16) { + prefix.copy_from_slice(head); + } + prefix } /// Check if payment is required for the given `XorName`. @@ -796,22 +833,34 @@ impl PaymentVerifier { Self::validate_quote_arithmetic(payment)?; let candidates = Self::legacy_median_candidates(payment)?; let mut failures = Vec::with_capacity(candidates.len()); - let mut verified_paid_quote = false; + // ADR-0004 Amendment 2: remember WHICH candidate's on-chain settlement + // verified — only that peer's commitment actually earned money, so only + // its pin is nominated for a deterministic first audit below. The + // non-median quotes merely locate the median and stay covered by the + // gossip-lottery audit path. + let mut paid_peer: Option<[u8; 32]> = None; for candidate in candidates { + let candidate_peer = *candidate.encoded_peer_id.as_bytes(); match self .verify_legacy_median_candidate(xorname, candidate) .await { Ok(()) => { - verified_paid_quote = true; + // First settlement-verified median candidate wins the paid + // slot and the sole first-audit nomination. If a client + // settled several TIED-median candidates, only this one is + // first-audited; gossiped extras keep the ADR-0002 lottery, + // sidecar-only extras are an accepted best-effort residual + // (ADR-0004 Amendment 2). The honest client pays one. + paid_peer = Some(candidate_peer); break; } Err(err) => failures.push(err.to_string()), } } - if !verified_paid_quote { + if paid_peer.is_none() { let xorname_hex = hex::encode(xorname); let details = if failures.is_empty() { "no median-priced candidates were available".to_string() @@ -834,7 +883,8 @@ impl PaymentVerifier { // (but signed) bundle can never enqueue audits or drive pin fetches — // closing the free-amplification path. Fresh client-put bundles only. if context == VerificationContext::ClientPut { - self.cross_check_quotes(payment, commitment_sidecars).await; + self.cross_check_quotes(payment, commitment_sidecars, paid_peer) + .await; } if crate::logging::enabled!(crate::logging::Level::INFO) { @@ -901,9 +951,29 @@ impl PaymentVerifier { Self::validate_paid_quote_signature(candidate).await?; - let on_chain_amount = self - .completed_payment_amount(candidate.quote.hash()) + let (on_chain_amount, recorded_rewards_prefix) = self + .completed_payment_settlement(candidate.quote.hash()) .await?; + // ADR-0004 Amendment 2: the settlement must be RECORDED FOR THE QUOTE'S + // rewards address, not merely under its quote hash. The vault stores + // whatever `(rewardsAddress, amount)` the payer supplied for the hash, + // so an amount-only check would accept a payment the client redirected + // to its own wallet — the issuer would be treated (and first-audited) + // as "paid" without ever being compensated. Honest clients build the + // payment from the quote itself, so this never rejects a legit upload. + // The vault packs the payee as `bytes16`, so the comparison is on the + // leading 16 bytes (128 bits — far beyond grinding range). + if let Some(recorded) = recorded_rewards_prefix { + let expected = Self::rewards_address_prefix(&candidate.quote.rewards_address); + if recorded != expected { + return Err(Error::Payment(format!( + "Median-priced quote settlement for peer {:?} was redirected: recorded rewards address prefix {} does not match the quote's {}", + candidate.encoded_peer_id, + hex::encode(recorded), + hex::encode(expected) + ))); + } + } if on_chain_amount >= candidate.expected_amount { return Ok(()); } @@ -945,7 +1015,15 @@ impl PaymentVerifier { .map_err(|e| Error::Payment(format!("Signature verification task failed: {e}")))? } - async fn completed_payment_amount(&self, quote_hash: QuoteHash) -> Result { + /// Look up the on-chain settlement recorded for `quote_hash`: the settled + /// amount and the `bytes16` rewards-address prefix it was recorded for. + /// The prefix is `None` only on the test-override path (meaning "matches + /// the quote"); the production contract read always returns the recorded + /// prefix so the caller's redirect check always applies. + async fn completed_payment_settlement( + &self, + quote_hash: QuoteHash, + ) -> Result<(Amount, Option<[u8; 16]>)> { #[cfg(any(test, feature = "test-utils"))] { let completed_payment_override = { @@ -954,8 +1032,8 @@ impl PaymentVerifier { .get("e_hash) .copied() }; - if let Some(amount) = completed_payment_override { - return Ok(amount); + if let Some((amount, recorded)) = completed_payment_override { + return Ok((amount, recorded)); } } @@ -969,7 +1047,7 @@ impl PaymentVerifier { .await .map_err(|e| Error::Payment(format!("completedPayments lookup failed: {e}")))?; - Ok(Amount::from(result.amount)) + Ok((Amount::from(result.amount), Some(result.rewardsAddress.0))) } fn validate_paid_quote_peer_binding( @@ -1369,7 +1447,12 @@ impl PaymentVerifier { /// missing cache or absent `P2PNode` degrades to "resolve nothing", never an /// error on the payment path — the synchronous arithmetic gate and the /// later audit remain the load-bearing checks. - async fn cross_check_quotes(&self, payment: &ProofOfPayment, commitment_sidecars: &[Vec]) { + async fn cross_check_quotes( + &self, + payment: &ProofOfPayment, + commitment_sidecars: &[Vec], + paid_peer: Option<[u8; 32]>, + ) { let now = std::time::Instant::now(); let ttl = crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL; let p2p = self.p2p_node.read().as_ref().map(Arc::clone); @@ -1393,17 +1476,32 @@ impl PaymentVerifier { }; let peer_id = PeerId::from_bytes(*encoded_peer_id.as_bytes()); - // ADR-0004: this commitment backed a payment — route it for a - // deterministic first audit (the drainer dedups by pin and respects - // the cooldown). Best-effort: a closed channel just means no first - // audit is scheduled, never an error on the payment path. - if let Some(ref tx) = monetized_pin_tx { - let _ = tx.send(crate::replication::MonetizedPinEvent { - peer: peer_id, - pin, - key_count: quote.committed_key_count, - quote_ts: quote.timestamp, - }); + // ADR-0004 Amendment 2: only the PAID candidate's commitment earned + // money, so only its pin is routed for a deterministic first audit + // (the drainer dedups, rate-budgets, and respects the cooldown). + // Pre-amendment this fired for every pinned quote in the bundle — + // up to CLOSE_GROUP_SIZE audits per proof for peers that were never + // paid — a 7x term in the fleet-wide amplification that degraded + // v0.14.3. Best-effort: a closed channel just means no first audit + // is scheduled, never an error on the payment path. + let is_paid = paid_peer.is_some_and(|paid| paid == *encoded_peer_id.as_bytes()); + if is_paid { + if let Some(ref tx) = monetized_pin_tx { + // Bounded queue: drop on full (best-effort, penalty-free; + // the peer's next settled payment re-nominates it). Count a + // Full drop so ingress saturation is observable; a Closed + // channel just means the engine is shutting down. + if let Err(tokio::sync::mpsc::error::TrySendError::Full(_)) = + tx.try_send(crate::replication::MonetizedPinEvent { + peer: peer_id, + pin, + key_count: quote.committed_key_count, + quote_ts: quote.timestamp, + }) + { + crate::replication::note_monetized_ingress_drop(); + } + } } // Resolution order: sidecar (synchronous, no state) -> gossip cache // (fresh within TTL) -> fetch fallback (collected as unresolved). @@ -2462,10 +2560,19 @@ impl PaymentVerifier { // cross-check + first-audit funnel as single-node quotes, AFTER on-chain // verification has succeeded (so an unpaid pool cannot drive audits or // fetches). ClientPut only — a replication receipt's pins have aged out. + // Amendment 2: only the candidates the contract actually PAID + // (`paid_node_addresses` indices, verified above) are nominated for + // first audits; the rest of the pool merely established the median. if context == VerificationContext::ClientPut { + let paid_indices: std::collections::HashSet = payment_info + .paid_node_addresses + .iter() + .map(|(_, idx, _)| *idx) + .collect(); self.cross_check_merkle_candidates( &merkle_proof.winner_pool, &merkle_proof.commitment_sidecars, + &paid_indices, ) .await; } @@ -2484,6 +2591,7 @@ impl PaymentVerifier { &self, pool: &evmlib::merkle_payments::MerklePaymentCandidatePool, commitment_sidecars: &[Vec], + paid_indices: &std::collections::HashSet, ) { let now = std::time::Instant::now(); let ttl = crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL; @@ -2495,23 +2603,37 @@ impl PaymentVerifier { let sidecar_map = Self::index_valid_sidecars(commitment_sidecars); let mut unresolved: Vec<(PeerId, [u8; 32], u32, Vec)> = Vec::new(); - for candidate in &pool.candidate_nodes { + for (idx, candidate) in pool.candidate_nodes.iter().enumerate() { let Some(pin) = candidate.commitment_pin else { continue; // baseline candidate pins nothing }; let peer_id = PeerId::from_bytes(*blake3::hash(&candidate.pub_key).as_bytes()); - if let Some(ref tx) = monetized_pin_tx { - let _ = tx.send(crate::replication::MonetizedPinEvent { - peer: peer_id, - pin, - key_count: candidate.committed_key_count, - quote_ts: std::time::UNIX_EPOCH - .checked_add(std::time::Duration::from_secs( - candidate.merkle_payment_timestamp, - )) - .unwrap_or(std::time::UNIX_EPOCH), - }); + // ADR-0004 Amendment 2: nominate only the candidates the contract + // actually paid — the unpaid pool members earned nothing and stay + // covered by the gossip-lottery audit path. Pre-amendment every + // pool candidate (16) was nominated per verified proof. + if paid_indices.contains(&idx) { + if let Some(ref tx) = monetized_pin_tx { + // Bounded queue: drop on full (best-effort, penalty-free; + // the peer's next settled payment re-nominates it). Count a + // Full drop so ingress saturation is observable; a Closed + // channel just means the engine is shutting down. + if let Err(tokio::sync::mpsc::error::TrySendError::Full(_)) = + tx.try_send(crate::replication::MonetizedPinEvent { + peer: peer_id, + pin, + key_count: candidate.committed_key_count, + quote_ts: std::time::UNIX_EPOCH + .checked_add(std::time::Duration::from_secs( + candidate.merkle_payment_timestamp, + )) + .unwrap_or(std::time::UNIX_EPOCH), + }) + { + crate::replication::note_monetized_ingress_drop(); + } + } } let resolved = match sidecar_map.get(&(peer_id, pin)) { @@ -2977,6 +3099,43 @@ mod tests { ); } + /// ADR-0004 Amendment 2: a settlement recorded under the median quote's + /// hash but redirected to a DIFFERENT rewards address must be rejected — + /// otherwise a client could "pay" its own wallet while the issuer is + /// treated (and first-audited) as paid without ever being compensated. + #[tokio::test] + async fn test_legacy_paid_median_redirected_settlement_rejected() { + let verifier = create_test_verifier(); + let xorname = [0xA2u8; 32]; + let peer_quotes = make_signed_legacy_bundle(xorname, unique_test_prices()); + mark_k_closest_paid_candidates(&verifier, &peer_quotes); + let expected_amount = expected_median_payment(&peer_quotes); + let paid_quote = median_test_candidates(&peer_quotes) + .first() + .expect("median candidate") + .1 + .clone(); + // Full amount, but recorded for an address that is not the quote's. + let attacker_address = RewardsAddress::new([0xEEu8; 20]); + assert_ne!(attacker_address, paid_quote.rewards_address); + verifier.set_completed_payment_with_address_for_tests( + paid_quote.hash(), + expected_amount, + attacker_address, + ); + + let proof_bytes = serialize_proof(peer_quotes); + let result = verifier + .verify_payment(&xorname, Some(&proof_bytes), VerificationContext::ClientPut) + .await; + + let err = result.expect_err("redirected settlement must be rejected"); + assert!( + format!("{err}").contains("redirected"), + "Error should mention the settlement redirect: {err}" + ); + } + #[tokio::test] async fn test_legacy_single_quote_proof_accepted() { let verifier = create_test_verifier(); @@ -4206,6 +4365,130 @@ mod tests { (merkle_proof, pool_hash, xorname, timestamp) } + /// ADR-0004 Amendment 2: only the PAID (settlement-verified) quote's pin is + /// nominated for a deterministic first audit; the unpaid bundle quotes are + /// cross-checked but never enqueue audits. + #[tokio::test] + async fn adr0004_first_audit_nominates_only_paid_single_node_quote() { + use evmlib::{EncodedPeerId, RewardsAddress}; + + let verifier = create_test_verifier(); + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + verifier.attach_monetized_pin_sender(tx); + + let ids: Vec<[u8; 32]> = (1..=3u8).map(|b| [b; 32]).collect(); + let payment = ProofOfPayment { + peer_quotes: ids + .iter() + .enumerate() + .map(|(i, id)| { + let pin_byte = u8::try_from(i).expect("small bundle") + 1; + let mut quote = make_fake_quote( + [0xD0; 32], + SystemTime::now(), + RewardsAddress::new([pin_byte; 20]), + ); + quote.commitment_pin = Some([pin_byte; 32]); + quote.committed_key_count = 100; + (EncodedPeerId::new(*id), quote) + }) + .collect(), + }; + + // The middle quote is the candidate whose on-chain settlement verified. + let paid = ids.get(1).copied().expect("paid id"); + verifier.cross_check_quotes(&payment, &[], Some(paid)).await; + + let event = rx.try_recv().expect("the paid quote must be nominated"); + assert_eq!(event.peer, PeerId::from_bytes(paid)); + assert_eq!(event.pin, [2u8; 32]); + assert!( + rx.try_recv().is_err(), + "unpaid bundle quotes must not be nominated for first audits" + ); + } + + /// ADR-0004 Amendment 2: with no verified paid candidate, nothing is + /// nominated (defensive: the caller never reaches the cross-check on a + /// failed verification, but the emission itself must also fail closed). + #[tokio::test] + async fn adr0004_first_audit_nominates_nothing_without_paid_peer() { + use evmlib::{EncodedPeerId, RewardsAddress}; + + let verifier = create_test_verifier(); + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + verifier.attach_monetized_pin_sender(tx); + + let mut quote = + make_fake_quote([0xD1; 32], SystemTime::now(), RewardsAddress::new([1; 20])); + quote.commitment_pin = Some([1u8; 32]); + let payment = ProofOfPayment { + peer_quotes: vec![(EncodedPeerId::new([1u8; 32]), quote)], + }; + + verifier.cross_check_quotes(&payment, &[], None).await; + assert!(rx.try_recv().is_err(), "no paid peer, no nomination"); + } + + /// ADR-0004 Amendment 2, merkle path: only the candidates at the + /// contract-verified PAID indices are nominated; the rest of the pool + /// merely established the median and must not enqueue audits. + #[tokio::test] + async fn adr0004_first_audit_nominates_only_paid_merkle_candidates() { + use evmlib::merkle_payments::{MerklePaymentCandidatePool, MerkleTree}; + + let verifier = create_test_verifier(); + let (tx, mut rx) = tokio::sync::mpsc::channel(8); + verifier.attach_monetized_pin_sender(tx); + + let timestamp = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system time") + .as_secs(); + let mut candidate_nodes = make_candidate_nodes(timestamp); + for (i, candidate) in candidate_nodes.iter_mut().enumerate() { + candidate.commitment_pin = Some([u8::try_from(i & 0xFF).expect("byte"); 32]); + candidate.committed_key_count = 50; + } + let addresses: Vec = (0..4u8) + .map(|i| xor_name::XorName::from_content(&[i])) + .collect(); + let tree = MerkleTree::from_xornames(addresses).expect("tree"); + let midpoint_proof = tree + .reward_candidates(timestamp) + .expect("reward candidates") + .first() + .expect("at least one candidate") + .clone(); + let pool = MerklePaymentCandidatePool { + midpoint_proof, + candidate_nodes, + }; + + let paid_indices: std::collections::HashSet = [1usize, 3].into_iter().collect(); + verifier + .cross_check_merkle_candidates(&pool, &[], &paid_indices) + .await; + + let mut nominated = Vec::new(); + while let Ok(event) = rx.try_recv() { + nominated.push(event); + } + assert_eq!( + nominated.len(), + 2, + "only the PAID candidates are nominated, not the whole pool" + ); + for (event, idx) in nominated.iter().zip([1usize, 3]) { + let candidate = pool.candidate_nodes.get(idx).expect("paid candidate"); + assert_eq!( + event.peer, + PeerId::from_bytes(*blake3::hash(&candidate.pub_key).as_bytes()) + ); + assert_eq!(event.pin, [u8::try_from(idx).expect("byte"); 32]); + } + } + /// Helper: build a minimal valid `MerklePaymentProof` with real ML-DSA-65 /// signatures. Returns `(xorname, serialized_tagged_proof, pool_hash, timestamp)`. fn make_valid_merkle_proof_bytes() -> ( diff --git a/src/replication/config.rs b/src/replication/config.rs index ed42f470..2ce8f8c3 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -404,6 +404,86 @@ pub const FIRST_AUDIT_SUMMARY_INTERVAL: Duration = Duration::from_secs(5 * 60); /// events the drainer processes audits, then loops back to drain more. pub const FIRST_AUDIT_DRAIN_BATCH: usize = 64; +/// ADR-0004 Amendment 2: token-bucket refill interval for monetized first-audit +/// launches — one launch token per interval, per node. +/// +/// This is the load-bearing bound: fleet-wide first-audit pressure becomes +/// `nodes x (1 / interval)` regardless of upload volume, where the pre-amendment +/// scheduler scaled with `uploads x pinned-quotes-per-proof x verifying-storers` +/// (measured at ~110 storage-commitment audit events/hour/service in the +/// DEV-01 staging run, with 77.7% of launches timing out in the DEV-03 +/// per-service-concurrency attempt). At 5 minutes the sustained rate is +/// 12 launches/hour/node (a drained burst bucket adds at most +/// [`FIRST_AUDIT_BUDGET_BURST`] more in the first hour); steady-state demand +/// sits far below it because nomination is paid-pin-only and re-nominations +/// are suppressed by [`FIRST_AUDIT_PEER_REAUDIT_INTERVAL`]. +pub const FIRST_AUDIT_LAUNCH_INTERVAL: Duration = Duration::from_secs(5 * 60); + +/// ADR-0004 Amendment 2: token-bucket capacity for monetized first-audit +/// launches. +/// +/// Lets a node absorb a small burst of NEW earning peers (several distinct +/// paid pins verified back-to-back) without waiting a full refill interval per +/// launch, while keeping the sustained rate pinned to +/// [`FIRST_AUDIT_LAUNCH_INTERVAL`]. +pub const FIRST_AUDIT_BUDGET_BURST: u32 = 2; + +/// ADR-0004 Amendment 2: max concurrently in-flight monetized first audits per +/// node. +/// +/// A first audit can hold a slot for the full size-scaled two-round deadline, +/// so without this cap a burst of slow/timing-out targets would pile up +/// concurrent subtree audits even under the launch-rate budget. Deferral is +/// penalty-free: the pin stays pending and is retried when a slot frees. +pub const FIRST_AUDIT_MAX_INFLIGHT: u64 = 2; + +/// ADR-0004 Amendment 2: max uniform random delay applied before sending a +/// monetized first-audit challenge. +/// +/// Every storer of a chunk verifies the same payment at the same instant, so +/// unjittered first audits from the whole close group would hit the paid peer +/// simultaneously and trip its per-peer responder admission cap (drops that +/// auditors then record as Timeout). The jitter decorrelates the observers. +pub const FIRST_AUDIT_LAUNCH_JITTER_MAX: Duration = Duration::from_secs(30); + +/// ADR-0004 Amendment 2: per-peer suppression window between monetized first +/// audits, across commitment rotations. +/// +/// Pin-level dedup alone is defeated by the hourly commitment rotation (every +/// rotation mints a fresh pin, re-arming every observer). After a first audit +/// LAUNCHES at a peer, further monetized nominations for that peer are dropped +/// for this window — unless the new pin's committed key count exceeds the +/// audited one by more than [`FIRST_AUDIT_COUNT_JUMP_NUM`]/ +/// [`FIRST_AUDIT_COUNT_JUMP_DEN`] (a peer that passes an audit on an honest +/// count and then mints a much larger commitment must be re-nominated +/// immediately: inflated SIDECAR-ONLY pins are visible to payment verifiers +/// only, so no gossip-lottery audit can ever cover them). Kept comfortably +/// inside the 3h answerability TTL so a re-nomination after the window still +/// lands in-window. Gossip-lottery re-audits are unaffected. +pub const FIRST_AUDIT_PEER_REAUDIT_INTERVAL: Duration = Duration::from_secs(2 * 3600); + +/// ADR-0004 Amendment 2: committed-count jump that overrides the per-peer +/// re-audit window, as a ratio (`new > old * NUM / DEN`, integer math). +/// +/// 3/2: a >1.5x growth in claimed committed keys within the suppression window +/// re-nominates the peer despite a recent first audit. +pub const FIRST_AUDIT_COUNT_JUMP_NUM: u64 = 3; +/// Denominator of the count-jump override ratio. See +/// [`FIRST_AUDIT_COUNT_JUMP_NUM`]. +pub const FIRST_AUDIT_COUNT_JUMP_DEN: u64 = 2; + +/// ADR-0004 Amendment 2: capacity of the bounded verifier-to-drainer +/// monetized-pin channel. +/// +/// Nominations are gated behind SETTLED on-chain payments, so legitimate +/// ingress is tiny; the drainer drains every wake (batched at +/// [`FIRST_AUDIT_DRAIN_BATCH`]) and coalesces newest-per-peer, so a backlog +/// this deep implies the drainer is starved, not that work is arriving fast. +/// Producers `try_send` and drop on full: a dropped nomination is +/// penalty-free; the peer's gossiped commitments stay lottery-covered and +/// its next settled payment re-nominates the paid pin. +pub const FIRST_AUDIT_INGRESS_CAPACITY: usize = 1024; + /// Number of subtree leaves spot-checked against real chunk bytes per audit /// (ADR-0002 real-bytes layer). /// diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 4cb83d5d..30daf1a7 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -81,6 +81,21 @@ use saorsa_core::identity::{NodeIdentity, PeerId}; use saorsa_core::{DhtNetworkEvent, P2PEvent, P2PNode, TrustEvent}; use saorsa_pqc::api::sig::{MlDsaSecretKey, MlDsaVariant}; +/// Count of monetized-pin nominations DROPPED at the bounded ingress channel +/// because it was full (Amendment 2). Process-global because the producer is +/// the payment verifier (a different module) and the drop happens before the +/// per-drainer `received` counter. A non-zero value is the rollout signal that +/// nomination ingress is saturating — benign (penalty-free, lottery/next- +/// payment covered) but worth watching. `Closed` (engine shut down) is not +/// counted: it is not a saturation signal. +static FIRST_AUDIT_INGRESS_DROPPED: AtomicU64 = AtomicU64::new(0); + +/// Record one ingress-full drop. Called by the payment verifier's `try_send` +/// sites; read by the drainer's periodic summary. +pub(crate) fn note_monetized_ingress_drop() { + FIRST_AUDIT_INGRESS_DROPPED.fetch_add(1, Ordering::Relaxed); +} + #[derive(Default)] struct FirstAuditObservability { received: AtomicU64, @@ -88,7 +103,13 @@ struct FirstAuditObservability { coalesced: AtomicU64, duplicates: AtomicU64, capacity_evicted: 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. + suppressed_lower: AtomicU64, cooldown_deferred_attempts: AtomicU64, + rate_deferred_attempts: AtomicU64, + window_deduped: AtomicU64, launched: AtomicU64, passed: AtomicU64, timed_out: AtomicU64, @@ -143,22 +164,62 @@ fn first_audit_terminal_outcome(result: &AuditTickResult) -> FirstAuditTerminalO #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum FirstAuditQueueOutcome { + /// New peer inserted with free capacity. Queued, + /// Collapsed with an existing same-peer entry; the incoming won (higher + /// count, or equal count and newer) and replaced it. Coalesced, + /// Collapsed with an existing same-peer entry; the incoming LOST because it + /// had a STRICTLY LOWER count and was dropped, leaving the higher-count pin + /// in place. This is the attempted cheaper-pin self-erasure signal. + SuppressedLower, + /// Collapsed with an existing same-peer entry of EQUAL count; the incoming + /// lost the freshness tie (it was not the newer of the two) and the + /// existing entry was retained. Benign, not an attack signal. + RetainedOnTie, + /// A DIFFERENT peer's entry was evicted by the bounded LRU to make room. 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( +/// Coalesce a monetized nomination into the per-peer pending queue with a +/// SECURITY-AWARE rule (ADR-0004 Amendment 2): keep the pin that most needs +/// auditing — the HIGHEST committed key count for that peer, newest on a tie. +/// +/// A strictly-lower-count incoming must NOT displace a higher-count pending pin, +/// otherwise a peer can erase an inflated (audit-worthy) commitment by simply +/// monetizing a cheaper one right after — and a sidecar-only inflated pin has no +/// gossip-lottery backstop. When the incoming loses, the retained entry's LRU +/// recency is left UNTOUCHED (via `peek`), so a flood of low-count nominations +/// cannot promote the retained pin's lane position. +/// +/// `incoming_is_newer` distinguishes ordinary enqueue (the incoming arrived +/// last, so it wins an equal-count tie for freshness) from a cooldown-race +/// requeue of an older reserved event (the pending successor is newer, so it +/// wins the tie). +fn coalesce_first_audit_event( pending: &mut LruCache, - event: MonetizedPinEvent, + incoming: MonetizedPinEvent, + incoming_is_newer: bool, ) -> FirstAuditQueueOutcome { - match pending.push(event.peer, event) { - None => FirstAuditQueueOutcome::Queued, - Some((replaced_peer, _)) if replaced_peer == event.peer => { - FirstAuditQueueOutcome::Coalesced + if let Some(existing) = pending.peek(&incoming.peer) { + // Strictly lower -> the incoming loses and is dropped WITHOUT touching + // the retained pin's recency (the security-relevant self-erasure case). + if incoming.key_count < existing.key_count { + return FirstAuditQueueOutcome::SuppressedLower; } + // Equal count -> keep the fresher; an older incoming loses a benign tie. + if incoming.key_count == existing.key_count && !incoming_is_newer { + return FirstAuditQueueOutcome::RetainedOnTie; + } + // Strictly higher, or equal-and-newer: the incoming wins. `push` updates + // the value and bumps MRU; replacing an existing key never evicts a + // different peer. + let _ = pending.push(incoming.peer, incoming); + return FirstAuditQueueOutcome::Coalesced; + } + // No same-peer entry: a `push` at capacity evicts the LRU (a DIFFERENT peer). + match pending.push(incoming.peer, incoming) { + None => FirstAuditQueueOutcome::Queued, Some((evicted_peer, evicted)) => FirstAuditQueueOutcome::CapacityEvicted { peer: evicted_peer, pin: evicted.pin, @@ -166,6 +227,517 @@ fn queue_first_audit_event( } } +/// ADR-0004 Amendment 2 (E′): slack added to the max launch jitter when +/// prefiltering a nomination's answerability at schedule time, covering the +/// spawn/dispatch latency between the timer firing and the wire challenge so a +/// jitter==MAX pin is not admitted only to fail the authoritative check by a +/// few milliseconds. Tiny against the multi-hour answerability window. +const FIRST_AUDIT_SEND_LATENCY_SLACK: Duration = Duration::from_secs(1); + +/// A first audit the limiter recently launched at a peer: when, and the +/// committed key count that was audited (for the count-jump override). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct RecentFirstAudit { + launched_at: Instant, + key_count: u32, +} + +/// Holds one first-audit in-flight slot; decrements the gauge on drop so a +/// panicking or cancelled audit task can never leak a slot and wedge the +/// [`config::FIRST_AUDIT_MAX_INFLIGHT`] cap shut. +struct FirstAuditInflightSlot(Arc); + +impl FirstAuditInflightSlot { + fn acquire(observability: &Arc) -> Self { + observability.inflight.fetch_add(1, Ordering::Relaxed); + Self(Arc::clone(observability)) + } +} + +impl Drop for FirstAuditInflightSlot { + fn drop(&mut self) { + self.0.inflight.fetch_sub(1, Ordering::Relaxed); + } +} + +/// ADR-0004 Amendment 2: whether `new_count` exceeds `audited_count` by more +/// than the [`config::FIRST_AUDIT_COUNT_JUMP_NUM`]/ +/// [`config::FIRST_AUDIT_COUNT_JUMP_DEN`] ratio (`new > old * NUM / DEN`, +/// overflow-free integer math). A jump re-nominates a peer despite a recent +/// first audit: an inflated SIDECAR-ONLY pin is visible to payment verifiers +/// only, so no gossip-lottery audit can ever cover it. +const fn first_audit_count_jump(audited_count: u32, new_count: u32) -> bool { + (new_count as u64) * config::FIRST_AUDIT_COUNT_JUMP_DEN + > (audited_count as u64) * config::FIRST_AUDIT_COUNT_JUMP_NUM +} + +/// The launch limiter's verdict for one pending monetized pin. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum LimiterVerdict { + /// Within budget and window: the caller may reserve a launch (and consumes + /// a token via [`FirstAuditLimiter::reserve_token`] only if it does). + Admit, + /// Launch-rate budget or in-flight cap exhausted. Penalty-free: keep the + /// pin pending and retry on a later tick. + RateDeferred, + /// The peer had a first audit within + /// [`config::FIRST_AUDIT_PEER_REAUDIT_INTERVAL`] and the new pin's count + /// shows no [`first_audit_count_jump`]. Drop the nomination (the + /// gossip-lottery re-audit path is unaffected). + WindowDeduped, +} + +/// ADR-0004 Amendment 2: the first-audit launch limiter — a token bucket +/// (launch rate), an in-flight cap, and a per-peer re-audit window that +/// survives pin rotation. +/// +/// This is the load-bearing aggregate bound the original scheduler lacked: +/// fleet-wide first-audit pressure becomes `nodes x refill-rate` instead of +/// `uploads x pinned-quotes-per-proof x verifying-storers`. Pure over passed +/// `now`/`inflight` values so every decision is unit-testable without a +/// runtime. +struct FirstAuditLimiter { + /// Launch tokens available, at most [`config::FIRST_AUDIT_BUDGET_BURST`]. + tokens: u32, + /// Refill anchor. While the bucket is full this tracks `now` (a full + /// bucket accrues nothing); while below capacity it advances only by + /// whole refill intervals so fractional elapsed time is never lost. + last_refill: Instant, + /// Peers given a first audit recently, with the audited key count. + /// Bounded like the drainer's other per-peer maps. + recent: LruCache, +} + +impl FirstAuditLimiter { + fn new(now: Instant) -> Self { + Self { + tokens: config::FIRST_AUDIT_BUDGET_BURST, + last_refill: now, + recent: LruCache::new( + NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN), + ), + } + } + + /// Whether the per-peer re-audit window admits a nomination for `peer` + /// carrying `key_count` at `now`. Read-only (`peek`), so it is safe to + /// call at ENQUEUE time — suppressed nominations never occupy pending + /// slots — without disturbing LRU recency. + fn window_allows(&self, peer: &PeerId, key_count: u32, now: Instant) -> bool { + self.recent.peek(peer).map_or(true, |prev| { + now.saturating_duration_since(prev.launched_at) + >= config::FIRST_AUDIT_PEER_REAUDIT_INTERVAL + || first_audit_count_jump(prev.key_count, key_count) + }) + } + + /// Refill the token bucket for the time elapsed up to `now`. + fn refill(&mut self, now: Instant) { + if self.tokens >= config::FIRST_AUDIT_BUDGET_BURST { + // Full bucket accrues nothing; keep the anchor current so the + // next consumption starts its interval from here. + self.last_refill = now; + return; + } + let interval = config::FIRST_AUDIT_LAUNCH_INTERVAL; + let elapsed = now.saturating_duration_since(self.last_refill); + let earned = elapsed.as_nanos() / interval.as_nanos().max(1); + if earned == 0 { + return; + } + let capacity_gap = u128::from(config::FIRST_AUDIT_BUDGET_BURST - self.tokens); + if earned >= capacity_gap { + self.tokens = config::FIRST_AUDIT_BUDGET_BURST; + self.last_refill = now; + } else { + // earned < capacity_gap <= u32::MAX, so both casts are lossless. + self.tokens = self + .tokens + .saturating_add(u32::try_from(earned).unwrap_or(u32::MAX)); + let advance = interval.saturating_mul(u32::try_from(earned).unwrap_or(u32::MAX)); + self.last_refill = self.last_refill.checked_add(advance).unwrap_or(now); + } + } + + /// Decide whether a pending pin may launch now. Consumes NOTHING: the + /// caller runs the remaining (per-peer cooldown) gate and calls + /// [`Self::commit_launch`] only for a launch that actually happens, so a + /// deferral elsewhere never burns budget or stamps the window. + fn assess( + &mut self, + peer: &PeerId, + key_count: u32, + now: Instant, + inflight: u64, + ) -> LimiterVerdict { + if !self.window_allows(peer, key_count, now) { + return LimiterVerdict::WindowDeduped; + } + self.refill(now); + if inflight >= config::FIRST_AUDIT_MAX_INFLIGHT || self.tokens == 0 { + return LimiterVerdict::RateDeferred; + } + LimiterVerdict::Admit + } + + /// Consume one launch token for a RESERVATION (ADR-0004 Amendment 2 E′). + /// Does NOT stamp the per-peer window: the durable `recent` stamp happens + /// only at [`Self::promote`], after the authoritative post-jitter + /// answerability check, so a reservation that is later cancelled leaves no + /// suppression behind. + fn reserve_token(&mut self) { + self.tokens = self.tokens.saturating_sub(1); + } + + /// Return a token consumed by a reservation that was cancelled before it + /// launched (answerability lapsed or a concurrent gossip audit won the + /// cooldown). Capped at the burst so a spurious double-refund cannot exceed + /// the bucket. + fn refund_token(&mut self) { + self.tokens = (self.tokens + 1).min(config::FIRST_AUDIT_BUDGET_BURST); + } + + /// Stamp the per-peer re-audit window for a launch that is ACTUALLY firing + /// (promotion). Separated from token consumption so suppression is only + /// ever recorded for a real send. + fn promote(&mut self, peer: PeerId, key_count: u32, now: Instant) { + self.recent.put( + peer, + RecentFirstAudit { + launched_at: now, + key_count, + }, + ); + } + + /// Test convenience: the pre-E′ atomic "a launch happened" — consume a token + /// and stamp the window in one call. Production splits these across + /// reservation and promotion; the limiter's own budget/window unit tests do + /// not model the jitter reservation and use this shorthand. + #[cfg(test)] + fn commit_launch(&mut self, peer: PeerId, key_count: u32, now: Instant) { + self.reserve_token(); + self.promote(peer, key_count, now); + } +} + +/// ADR-0004 Amendment 2 (E′ B-prefilter): whether a monetized pin is answerable +/// across the ENTIRE launch-jitter window ending at +/// `now + FIRST_AUDIT_LAUNCH_JITTER_MAX + slack`, so committing scheduling state +/// for it cannot later require aborting an out-of-window challenge. The +/// too-future bound is enforced at `now` (a jitter delay only ages a pin +/// forward, never toward the future); the too-old bound is enforced at the +/// latest possible send time. This is a conservative admission prefilter; the +/// authoritative answerability check still runs at promotion against the real +/// send-time wall clock, so A1 (no false conviction) holds regardless of how +/// jitter and the answerability margin are sized. `checked_add` overflow fails +/// closed (skip the pin). +fn quote_answerable_through_nominal_jitter(quote_ts: SystemTime, now: SystemTime) -> bool { + let Some(latest_send) = now + .checked_add(config::FIRST_AUDIT_LAUNCH_JITTER_MAX) + .and_then(|t| t.checked_add(FIRST_AUDIT_SEND_LATENCY_SLACK)) + else { + return false; + }; + quote_within_audit_window(quote_ts, now) && quote_within_audit_window(quote_ts, latest_send) +} + +/// A far-future `Instant` used to effectively DISABLE the promotion-timer +/// select arm when no reservation is outstanding. The drainer still wakes at +/// least every [`config::FIRST_AUDIT_RETRY_INTERVAL`] via its tick, so this only +/// needs to be comfortably past the next tick. +fn first_audit_far_future() -> Instant { + Instant::now() + .checked_add(Duration::from_secs(3600)) + .unwrap_or_else(Instant::now) +} + +/// ADR-0004 Amendment 2 (E′): one outstanding first-audit reservation. Holds +/// its in-flight slot and launch token from schedule time until the jitter +/// timer fires; the durable suppression (`recent` + `first_audited`) is stamped +/// only if the authoritative post-jitter answerability + cooldown checks pass at +/// promotion, so a cancelled reservation leaves NO suppression behind. +struct FirstAuditReservation { + event: MonetizedPinEvent, + ready_at: Instant, + inflight: FirstAuditInflightSlot, +} + +/// ADR-0004 Amendment 2 (E′): the drainer-owned first-audit scheduler. Owns the +/// pending queue, the dedup set, the launch limiter, the alternating lane, and +/// the single outstanding reservation. All mutation is single-threaded in the +/// drainer task; the only asynchrony is the spawned audit I/O (which just holds +/// the moved-in in-flight slot). Kept as a struct so the reserve/promote/cancel +/// state machine is unit-testable with injected clocks. +struct FirstAuditScheduler { + /// Pins already given a first audit (dedup). A pin enters only at PROMOTION + /// (a real send), never at reservation, so a cancelled reservation can be + /// re-nominated. + first_audited: LruCache<[u8; 32], ()>, + /// Newest-per-peer pending nominations not yet launched. + pending: LruCache, + /// Token bucket + per-peer re-audit window. + limiter: FirstAuditLimiter, + /// The single outstanding reservation (E′ serializes reservations so the + /// per-launch lane alternation is preserved and at most one jitter timer is + /// live). + reserved: Option, + /// Alternating launch lane, flipped on every PROMOTION (real launch). + oldest_first_lane: bool, +} + +impl FirstAuditScheduler { + fn new(now: Instant) -> Self { + let cap = NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN); + Self { + first_audited: LruCache::new(cap), + pending: LruCache::new(cap), + limiter: FirstAuditLimiter::new(now), + reserved: None, + oldest_first_lane: false, + } + } + + /// Observability getter: used by the scheduler summary log and unit tests, + /// both absent from a release `--no-default-features` build, so it is dead + /// only in that configuration. + #[cfg_attr(not(feature = "logging"), allow(dead_code))] + fn pending_len(&self) -> usize { + self.pending.len() + } + + /// Observability getter (see [`Self::pending_len`]). + #[cfg_attr(not(feature = "logging"), allow(dead_code))] + fn tokens(&self) -> u32 { + self.limiter.tokens + } + + fn has_reservation(&self) -> bool { + self.reserved.is_some() + } + + /// When the outstanding reservation becomes eligible for promotion. + fn reserved_ready_at(&self) -> Option { + self.reserved.as_ref().map(|r| r.ready_at) + } + + /// The peer of the outstanding reservation, if any. + fn reserved_peer(&self) -> Option { + 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. + fn enqueue(&mut self, event: MonetizedPinEvent, obs: &Arc) { + if self.first_audited.contains(&event.pin) { + obs.duplicates.fetch_add(1, Ordering::Relaxed); + return; + } + let reserved_peer = self.reserved_peer(); + let is_reserved_peer = reserved_peer == Some(event.peer); + if !is_reserved_peer + && !self + .limiter + .window_allows(&event.peer, event.key_count, Instant::now()) + { + obs.window_deduped.fetch_add(1, Ordering::Relaxed); + return; + } + // Ordinary enqueue: the incoming arrived last, so it wins an equal-count + // tie. + match coalesce_first_audit_event(&mut self.pending, event, true) { + FirstAuditQueueOutcome::Queued => { + obs.queued.fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::Coalesced | FirstAuditQueueOutcome::RetainedOnTie => { + obs.coalesced.fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::SuppressedLower => { + obs.suppressed_lower.fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::CapacityEvicted { .. } => { + obs.queued.fetch_add(1, Ordering::Relaxed); + obs.capacity_evicted.fetch_add(1, Ordering::Relaxed); + } + } + } + + /// Attempt to create the single reservation from `pending` (E′ reserve). + /// Scans in the current lane order and reserves the FIRST eligible pin: + /// consumes one token, acquires one in-flight slot, and sets `ready_at = + /// mono_now + jitter`. Does NOT stamp `recent`/`first_audited` and does NOT + /// flip the lane — those happen only at promotion. Returns whether a + /// reservation was made. `cooldown` is a read-only snapshot (the + /// authoritative check-and-stamp is at promotion). + fn try_reserve( + &mut self, + mono_now: Instant, + inflight: u64, + jitter: Duration, + cooldown: &HashMap, + obs: &Arc, + ) -> bool { + if self.reserved.is_some() || self.pending.is_empty() { + return false; + } + self.limiter.refill(mono_now); + if inflight >= config::FIRST_AUDIT_MAX_INFLIGHT || self.limiter.tokens == 0 { + obs.rate_deferred_attempts.fetch_add(1, Ordering::Relaxed); + return false; + } + // MRU->LRU order; reverse for the oldest-first lane so the preferred + // end is at the front. + let mut ordered: Vec<(usize, PeerId, MonetizedPinEvent)> = self + .pending + .iter() + .enumerate() + .map(|(i, (p, e))| (i, *p, *e)) + .collect(); + self.pending.clear(); + if self.oldest_first_lane { + ordered.reverse(); + } + let mut chosen: Option = None; + let mut kept: Vec<(usize, PeerId, MonetizedPinEvent)> = Vec::new(); + for (idx, peer, event) in ordered { + if chosen.is_some() { + kept.push((idx, peer, event)); + continue; + } + if self.first_audited.contains(&event.pin) { + obs.duplicates.fetch_add(1, Ordering::Relaxed); + continue; // drop: already audited + } + if !quote_answerable_through_nominal_jitter(event.quote_ts, SystemTime::now()) { + obs.outside_answerability_window + .fetch_add(1, Ordering::Relaxed); + continue; // drop: cannot stay answerable through the jitter + } + // Window + budget + inflight. Tokens do not decrease during the + // scan (reserve happens after the loop) and `inflight` is fixed, so + // after the upfront budget gate `assess` never returns RateDeferred + // here; a defensive RateDeferred keeps the pin. + match self + .limiter + .assess(&peer, event.key_count, mono_now, inflight) + { + LimiterVerdict::WindowDeduped => { + obs.window_deduped.fetch_add(1, Ordering::Relaxed); + continue; // drop: recently first-audited, no count jump + } + LimiterVerdict::RateDeferred => { + kept.push((idx, peer, event)); + continue; // keep (defensive; unreachable after upfront gate) + } + LimiterVerdict::Admit => {} + } + if !cooldown_would_allow(cooldown, &peer, mono_now) { + obs.cooldown_deferred_attempts + .fetch_add(1, Ordering::Relaxed); + kept.push((idx, peer, event)); + continue; // keep: on shared cooldown, retry later + } + chosen = Some(event); + } + // Restore relative recency (oldest re-put first -> newest stays MRU). + kept.sort_unstable_by_key(|(idx, _, _)| std::cmp::Reverse(*idx)); + for (_, peer, event) in kept { + self.pending.put(peer, event); + } + let Some(event) = chosen else { + return false; + }; + self.limiter.reserve_token(); + let inflight_slot = FirstAuditInflightSlot::acquire(obs); + let ready_at = mono_now.checked_add(jitter).unwrap_or(mono_now); + self.reserved = Some(FirstAuditReservation { + event, + ready_at, + inflight: inflight_slot, + }); + true + } + + /// Take the outstanding reservation if its jitter has elapsed at `mono_now`. + fn take_due_reservation(&mut self, mono_now: Instant) -> Option { + if self + .reserved + .as_ref() + .is_some_and(|r| mono_now >= r.ready_at) + { + self.reserved.take() + } else { + None + } + } + + /// Authoritative promotion of a due reservation (E′). The caller holds the + /// shared cooldown write lock and passes the real send-time `wall_now` and + /// `mono_now`. Returns `Some((event, slot))` to spawn the audit on a real + /// launch, or `None` when the launch was cancelled (answerability lapsed + /// during jitter) or requeued (a concurrent gossip audit won the cooldown). + /// On both `None` paths the token is refunded, the in-flight slot released, + /// and NO suppression is recorded. + fn resolve( + &mut self, + reservation: FirstAuditReservation, + wall_now: SystemTime, + mono_now: Instant, + cooldown: &mut HashMap, + obs: &Arc, + ) -> Option<(MonetizedPinEvent, FirstAuditInflightSlot)> { + let FirstAuditReservation { + event, inflight, .. + } = reservation; + // Authoritative answerability at the REAL send time. + if !quote_within_audit_window(event.quote_ts, wall_now) { + self.limiter.refund_token(); + obs.outside_answerability_window + .fetch_add(1, Ordering::Relaxed); + drop(inflight); + return None; // cancelled; nothing stamped, nothing to roll back + } + // Authoritative shared-cooldown check-and-stamp. Losing this race to a + // concurrent gossip audit requeues the reserved event through the SAME + // security-aware coalescing: the reserved event is OLDER than any + // same-peer successor (`incoming_is_newer = false`), so a higher-count + // reserved event still wins over a lower-count successor (the inflated + // pin must be audited), while an equal/higher successor is preserved. + if !cooldown_allows_audit(cooldown, &event.peer, mono_now) { + self.limiter.refund_token(); + obs.cooldown_deferred_attempts + .fetch_add(1, Ordering::Relaxed); + drop(inflight); + // Account for the requeue outcome (the nomination itself was already + // counted at ingress, so `queued` is not re-incremented): a capacity + // eviction of a DIFFERENT peer and a suppressed reserved event are + // both observable per the ADR funnel. + match coalesce_first_audit_event(&mut self.pending, event, false) { + FirstAuditQueueOutcome::CapacityEvicted { .. } => { + obs.capacity_evicted.fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::SuppressedLower => { + obs.suppressed_lower.fetch_add(1, Ordering::Relaxed); + } + FirstAuditQueueOutcome::Queued + | FirstAuditQueueOutcome::Coalesced + | FirstAuditQueueOutcome::RetainedOnTie => {} + } + return None; + } + // Promote: stamp durable suppression, flip the lane, count the launch. + self.limiter.promote(event.peer, event.key_count, mono_now); + self.first_audited.put(event.pin, ()); + self.oldest_first_lane = !self.oldest_first_lane; + obs.launched.fetch_add(1, Ordering::Relaxed); + Some((event, inflight)) + } +} + // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- @@ -448,11 +1020,15 @@ pub struct ReplicationEngine { possession_check_rx: Option>, /// ADR-0004: sender the payment verifier clones to surface monetized pins /// for a deterministic first audit. The matching receiver is drained by - /// `start_first_audit_drainer`. - monetized_pin_tx: mpsc::UnboundedSender, + /// `start_first_audit_drainer`. BOUNDED (Amendment 2): the producer + /// `try_send`s and drops on a full queue, so ingress memory is capped just + /// like launches; a dropped nomination is penalty-free — the peer's + /// gossiped commitments stay lottery-covered and its next settled payment + /// re-nominates the paid pin. + monetized_pin_tx: mpsc::Sender, /// ADR-0004: receiver half of the monetized-pin channel, taken by /// `start_first_audit_drainer`. - monetized_pin_rx: Option>, + monetized_pin_rx: Option>, /// Shutdown token. shutdown: CancellationToken, /// Background task handles. @@ -490,7 +1066,11 @@ impl ReplicationEngine { let (possession_check_tx, possession_check_rx) = mpsc::unbounded_channel(); // ADR-0004: monetized-pin channel (verifier -> first-audit drainer). - let (monetized_pin_tx, monetized_pin_rx) = mpsc::unbounded_channel(); + // Bounded (Amendment 2): every stage of the first-audit pipeline is + // now capacity-limited — ingress queue here, pending set (LRU), and + // launch rate (token bucket). + let (monetized_pin_tx, monetized_pin_rx) = + mpsc::channel(config::FIRST_AUDIT_INGRESS_CAPACITY); let engine = Self { config: Arc::clone(&config), @@ -535,10 +1115,11 @@ impl ReplicationEngine { } /// ADR-0004: a sender the payment verifier uses to surface monetized pins - /// (commitments that backed a payment) for a deterministic first audit. - /// Cloneable; the engine drains the matching receiver. + /// (commitments that backed a payment) for a first audit. Cloneable; the + /// engine drains the matching receiver. Bounded: senders must `try_send` + /// and treat a full queue as a benign drop (Amendment 2 best-effort). #[must_use] - pub fn monetized_pin_sender(&self) -> mpsc::UnboundedSender { + pub fn monetized_pin_sender(&self) -> mpsc::Sender { self.monetized_pin_tx.clone() } @@ -913,134 +1494,187 @@ impl ReplicationEngine { let observability = Arc::new(FirstAuditObservability::default()); let handle = tokio::spawn(async move { - // Bounded dedup of pins that have ALREADY been given their - // deterministic first audit. A pin is inserted ONLY when an audit is - // actually launched (never on a cooldown skip), so a pin skipped now - // can still be first-audited later. - let mut first_audited: LruCache<[u8; 32], ()> = LruCache::new( - NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN), - ); - // PERSISTENT pending queue: the most-recently-monetized pin per peer - // that has NOT yet been first-audited. A pin stays here until it is - // ACTUALLY first-audited (enters `first_audited`) — never removed for - // any weaker reason (e.g. a cooldown stamp), so an unaudited monetized - // pin is never silently forgotten. Newest-per-peer: a fresher pin for - // the same peer replaces the older one. Memory is bounded by an LRU: - // each entry needs a SETTLED on-chain payment, so the realistic count - // is tiny; the LRU is a pure DoS backstop that, only under an absurd - // flood, evicts the LEAST-RECENTLY-MONETIZED peer (the one most likely - // already superseded) — never the newest. - let mut pending: LruCache = LruCache::new( - NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN), - ); - // Periodic retry tick for pending (cooldown-blocked) pins. Created - // once; `Skip` so a backlog of missed ticks collapses to one. + // ADR-0004 Amendment 2 (E'): the drainer-owned first-audit + // scheduler. Payments only NOMINATE pins; the token bucket launches + // 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()); + // 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); 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 - // reopens even if no new pin arrives. - let drained_new = tokio::select! { + // The reservation's jitter deadline is a wake source: if a + // reservation is outstanding, sleep until it is due; otherwise + // a far-future deadline effectively disables that arm and only + // shutdown/rx/tick wake the loop. Recreated each iteration so a + // newly reserved (earlier) deadline is honoured next turn. + let promotion_due = scheduler + .reserved_ready_at() + .unwrap_or_else(first_audit_far_future); + tokio::select! { () = 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 - // unaudited pending pin for the same peer (it would then - // 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) { - observability.duplicates.fetch_add(1, Ordering::Relaxed); - debug!( - "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 { - 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=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() - ); - } - } - } + scheduler.enqueue(e, &observability); + // Drain a bounded burst so a flood cannot starve the + // launch phase. let mut drained = 1usize; while drained < config::FIRST_AUDIT_DRAIN_BATCH { match rx.try_recv() { Ok(e) => { observability.received.fetch_add(1, Ordering::Relaxed); - if first_audited.contains(&e.pin) { - observability - .duplicates - .fetch_add(1, Ordering::Relaxed); - } else { - 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); - } - } - } + scheduler.enqueue(e, &observability); drained += 1; } Err(_) => break, } } - true } None => break, }, - _ = tick.tick() => false, - }; - let _ = drained_new; + _ = tick.tick() => {} + () = tokio::time::sleep_until(promotion_due.into()) => {} + } + + // 1) Promote a due reservation. Resolved after EVERY wake (not + // only when the timer arm wins) so a continuously-ready `rx` + // cannot indefinitely delay a due promotion. The authoritative + // answerability + cooldown check-and-stamp happens here under + // the shared cooldown write lock, immediately before the send. + if let Some(reservation) = scheduler.take_due_reservation(Instant::now()) { + let promoted = { + let mut cooldown = gossip_audit.cooldown.write().await; + scheduler.resolve( + reservation, + SystemTime::now(), + Instant::now(), + &mut cooldown, + &observability, + ) + }; + if let Some((event, inflight_slot)) = promoted { + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=launched peer={} pin={} key_count={} inflight={}", + event.peer, hex::encode(event.pin), event.key_count, + observability.inflight.load(Ordering::Relaxed) + ); + let trigger = gossip_audit.clone(); + let audit_observability = Arc::clone(&observability); + tokio::spawn(async move { + // The jitter already elapsed as the reservation's + // timer; the slot is held for the audit's duration + // and released on drop (panic-safe). + let inflight_slot = inflight_slot; + let started = Instant::now(); + let credit = storage_commitment_audit::AuditCredit { + recent_provers: &trigger.recent_provers, + }; + let result = storage_commitment_audit::run_subtree_audit( + &trigger.p2p_node, + &trigger.config, + &event.peer, + event.pin, + event.key_count, + 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); + } + 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); + } + } + drop(inflight_slot); + 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, + &trigger.sync_state, + &trigger.recent_provers, + &trigger.config, + ) + .await; + }); + } + } + + // 2) Open the single reservation from the pending queue if none + // is outstanding (read-only cooldown snapshot; the + // authoritative stamp is at promotion). Serializing + // reservations preserves per-launch lane alternation and keeps + // at most one jitter timer live. + if !scheduler.has_reservation() { + let jitter = Duration::from_millis( + rand::thread_rng().gen_range( + 0..=u64::try_from(config::FIRST_AUDIT_LAUNCH_JITTER_MAX.as_millis()) + .unwrap_or(u64::MAX), + ), + ); + let inflight = observability.inflight.load(Ordering::Relaxed); + let reserved = { + let cooldown = gossip_audit.cooldown.read().await; + scheduler.try_reserve( + Instant::now(), + inflight, + jitter, + &cooldown, + &observability, + ) + }; + if reserved { + if let Some(peer) = scheduler.reserved_peer() { + debug!( + "First-audit scheduler: audit_trigger=first_monetized outcome=reserved peer={peer} pending={}", + scheduler.pending_len() + ); + } + } + } 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 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={} inflight={} tokens={}", + FIRST_AUDIT_INGRESS_DROPPED.load(Ordering::Relaxed), observability.received.load(Ordering::Relaxed), observability.queued.load(Ordering::Relaxed), observability.coalesced.load(Ordering::Relaxed), + observability.suppressed_lower.load(Ordering::Relaxed), observability.duplicates.load(Ordering::Relaxed), observability.capacity_evicted.load(Ordering::Relaxed), observability.cooldown_deferred_attempts.load(Ordering::Relaxed), + observability.rate_deferred_attempts.load(Ordering::Relaxed), + observability.window_deduped.load(Ordering::Relaxed), observability.launched.load(Ordering::Relaxed), observability.passed.load(Ordering::Relaxed), observability.timed_out.load(Ordering::Relaxed), @@ -1049,141 +1683,12 @@ impl ReplicationEngine { observability.idle.load(Ordering::Relaxed), observability.insufficient_keys.load(Ordering::Relaxed), observability.outside_answerability_window.load(Ordering::Relaxed), - pending.len(), + scheduler.pending_len(), observability.inflight.load(Ordering::Relaxed), + scheduler.tokens(), ); last_summary = Instant::now(); } - - if pending.is_empty() { - continue; - } - - // Try to launch an audit for each pending peer; keep the ones - // still blocked by cooldown for the next tick. Drain into a vec - // first so we can re-insert the still-blocked ones afterwards - // (LruCache has no drain). `iter()` yields most- to least-recently- - // used; we reverse so re-inserting blocked entries below restores - // their relative recency (oldest re-put first → stays the eviction - // victim, newest stays most-recently-used). - let snapshot: Vec<(PeerId, MonetizedPinEvent)> = - pending.iter().rev().map(|(p, e)| (*p, *e)).collect(); - pending.clear(); - 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 - // quote_ts lands inside the answerability window. With grace - // removed, auditing an out-of-window pin the responder may have - // aged out would false-convict, so a stale OR far-future/skewed - // quote is skipped. Legit first-audits fire moments after - // payment (quote_ts ≈ now); a skipped pin can still be - // gossip-lottery audited. - if !quote_within_audit_window(event.quote_ts, SystemTime::now()) { - observability - .outside_answerability_window - .fetch_add(1, Ordering::Relaxed); - debug!( - "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; - } - // Cooldown: if the peer's per-peer audit window is closed, keep - // this pin pending and retry on a later tick once it reopens. - // We do NOT treat "cooldown closed" as "already audited" (a - // losing gossip lottery can stamp the window without auditing), - // so the pin stays pending until it gets a REAL first audit; it - // is only ever evicted by the LRU memory backstop above, which - // drops the least-recently-monetized peer, not this newest one. - { - let now = Instant::now(); - let mut map = gossip_audit.cooldown.write().await; - if !cooldown_allows_audit(&mut map, &peer, now) { - pending.put(peer, event); - observability - .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={}", - 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, - }; - let result = storage_commitment_audit::run_subtree_audit( - &trigger.p2p_node, - &trigger.config, - &event.peer, - event.pin, - event.key_count, - 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); - } - 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!( - "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, - &trigger.sync_state, - &trigger.recent_provers, - &trigger.config, - ) - .await; - }); - } } debug!("First-audit drainer shut down"); }); @@ -4818,6 +5323,18 @@ fn cooldown_allows_audit(map: &mut HashMap, peer: &PeerId, now: true } +/// Read-only companion to [`cooldown_allows_audit`]: whether `peer` is OUTSIDE +/// its cooldown at `now`, WITHOUT stamping. Used by the first-audit reserve gate +/// (ADR-0004 Amendment 2 E′) to avoid reserving a peer that a recent audit +/// already covered; the authoritative check-and-stamp still runs at promotion, +/// so this is only an optimization and never the security boundary. +fn cooldown_would_allow(map: &HashMap, peer: &PeerId, now: Instant) -> bool { + let cooldown = Duration::from_secs(config::AUDIT_ON_GOSSIP_COOLDOWN_SECS); + map.get(peer).map_or(true, |&last| { + now.saturating_duration_since(last) >= cooldown + }) +} + /// The gossip-audit launch decision in ONE place so the ordering is shared /// between production and its test (ADR-0002 "occasional surprise exams"). /// @@ -5426,6 +5943,11 @@ async fn rebuild_and_rotate_commitment( #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod tests { use super::*; + use crate::payment::VerificationContext; + use crate::replication::audit::AuditTickResult; + use crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL; + use crate::replication::recent_provers::RecentProvers; + use crate::replication::types::{AuditFailureReason, FailureEvidence}; fn test_peer(b: u8) -> PeerId { let mut bytes = [0u8; 32]; @@ -5535,44 +6057,811 @@ mod tests { } #[test] - fn first_audit_queue_exposes_coalescing_and_capacity_eviction() { + fn first_audit_coalescing_keeps_highest_count_and_exposes_eviction() { let mut pending = LruCache::new(NonZeroUsize::new(1).unwrap()); - let first = MonetizedPinEvent { - peer: test_peer(1), + let peer = test_peer(1); + let base = MonetizedPinEvent { + peer, pin: [1; 32], - key_count: 1, + key_count: 100, quote_ts: SystemTime::now(), }; - let replacement = MonetizedPinEvent { + + // First insert into an empty slot: Queued. + assert_eq!( + coalesce_first_audit_event(&mut pending, base, true), + FirstAuditQueueOutcome::Queued + ); + + // A strictly LOWER-count same-peer nomination must NOT displace it. + let lower = MonetizedPinEvent { pin: [2; 32], - ..first + key_count: 50, + ..base + }; + assert_eq!( + coalesce_first_audit_event(&mut pending, lower, true), + FirstAuditQueueOutcome::SuppressedLower + ); + assert_eq!( + pending.peek(&peer).map(|e| (e.pin, e.key_count)), + Some(([1; 32], 100)), + "the higher-count pin is retained" + ); + + // A HIGHER-count same-peer nomination wins (the inflated pin to audit). + let higher = MonetizedPinEvent { + pin: [3; 32], + key_count: 400, + ..base + }; + assert_eq!( + coalesce_first_audit_event(&mut pending, higher, true), + FirstAuditQueueOutcome::Coalesced + ); + assert_eq!( + pending.peek(&peer).map(|e| (e.pin, e.key_count)), + Some(([3; 32], 400)) + ); + + // EQUAL count: an ordinary (newer) enqueue replaces for freshness... + let equal_newer = MonetizedPinEvent { + pin: [4; 32], + key_count: 400, + ..base + }; + assert_eq!( + coalesce_first_audit_event(&mut pending, equal_newer, true), + FirstAuditQueueOutcome::Coalesced + ); + assert_eq!(pending.peek(&peer).map(|e| e.pin), Some([4; 32])); + // ...but an equal-count OLDER requeue (incoming_is_newer=false) does not. + let equal_older = MonetizedPinEvent { + pin: [5; 32], + key_count: 400, + ..base }; + assert_eq!( + coalesce_first_audit_event(&mut pending, equal_older, false), + FirstAuditQueueOutcome::RetainedOnTie + ); + assert_eq!(pending.peek(&peer).map(|e| e.pin), Some([4; 32])); + + // A different peer at capacity 1 evicts the LRU (a DIFFERENT peer). let other_peer = MonetizedPinEvent { peer: test_peer(2), - pin: [3; 32], - ..first + pin: [6; 32], + key_count: 100, + ..base }; + assert_eq!( + coalesce_first_audit_event(&mut pending, other_peer, true), + FirstAuditQueueOutcome::CapacityEvicted { peer, pin: [4; 32] } + ); + assert_eq!(pending.len(), 1); + assert_eq!(pending.peek(&other_peer.peer).map(|e| e.pin), Some([6; 32])); + } + + // -- ADR-0004 Amendment 2: first-audit launch limiter -------------------- + #[test] + fn first_audit_limiter_enforces_burst_then_refills() { + let base = Instant::now(); + let mut limiter = FirstAuditLimiter::new(base); + let interval = config::FIRST_AUDIT_LAUNCH_INTERVAL; + + // The full burst is admitted back-to-back (distinct peers). + for i in 0..config::FIRST_AUDIT_BUDGET_BURST { + let peer = test_peer(u8::try_from(i).expect("small burst")); + assert_eq!(limiter.assess(&peer, 10, base, 0), LimiterVerdict::Admit); + limiter.commit_launch(peer, 10, base); + } + // Bucket empty: the next distinct peer is deferred, never dropped. + let extra = test_peer(0xEE); assert_eq!( - queue_first_audit_event(&mut pending, first), - FirstAuditQueueOutcome::Queued + limiter.assess(&extra, 10, base, 0), + LimiterVerdict::RateDeferred ); + + // One full interval later exactly one token is available again. + let later = base + interval; + assert_eq!(limiter.assess(&extra, 10, later, 0), LimiterVerdict::Admit); + limiter.commit_launch(extra, 10, later); + let extra2 = test_peer(0xEF); assert_eq!( - queue_first_audit_event(&mut pending, replacement), - FirstAuditQueueOutcome::Coalesced + limiter.assess(&extra2, 10, later, 0), + LimiterVerdict::RateDeferred + ); + } + + #[test] + fn first_audit_limiter_refill_keeps_fractional_remainder() { + let base = Instant::now(); + let mut limiter = FirstAuditLimiter::new(base); + let interval = config::FIRST_AUDIT_LAUNCH_INTERVAL; + for i in 0..config::FIRST_AUDIT_BUDGET_BURST { + let peer = test_peer(u8::try_from(i).expect("small burst")); + assert_eq!(limiter.assess(&peer, 1, base, 0), LimiterVerdict::Admit); + limiter.commit_launch(peer, 1, base); + } + // 1.5 intervals later one token is earned and the half interval is + // NOT lost to drift... + let at_1_5 = base + interval + interval / 2; + let p = test_peer(0xAA); + assert_eq!(limiter.assess(&p, 1, at_1_5, 0), LimiterVerdict::Admit); + limiter.commit_launch(p, 1, at_1_5); + // ...so the next token arrives at 2.0 intervals, not 2.5. + let at_2_0 = base + interval * 2; + let q = test_peer(0xAB); + assert_eq!(limiter.assess(&q, 1, at_2_0, 0), LimiterVerdict::Admit); + } + + #[test] + fn first_audit_limiter_inflight_cap_defers_until_slot_frees() { + let base = Instant::now(); + let mut limiter = FirstAuditLimiter::new(base); + let peer = test_peer(1); + assert_eq!( + limiter.assess(&peer, 1, base, config::FIRST_AUDIT_MAX_INFLIGHT), + LimiterVerdict::RateDeferred + ); + // A freed slot admits without any clock movement. + assert_eq!( + limiter.assess(&peer, 1, base, config::FIRST_AUDIT_MAX_INFLIGHT - 1), + LimiterVerdict::Admit + ); + } + + #[test] + fn first_audit_limiter_assess_consumes_nothing() { + let base = Instant::now(); + let mut limiter = FirstAuditLimiter::new(base); + let peer = test_peer(3); + // Repeated assessment must not burn budget or stamp the window: only + // `commit_launch` consumes (the cooldown gate between assess and + // commit can defer, and that deferral must be free). + for _ in 0..10 { + assert_eq!(limiter.assess(&peer, 5, base, 0), LimiterVerdict::Admit); + } + for i in 0..config::FIRST_AUDIT_BUDGET_BURST { + let p = test_peer(0x20 + u8::try_from(i).expect("small burst")); + assert_eq!(limiter.assess(&p, 5, base, 0), LimiterVerdict::Admit); + limiter.commit_launch(p, 5, base); + } + } + + #[test] + fn first_audit_limiter_window_dedups_rotated_pins_and_count_jump_overrides() { + let base = Instant::now(); + let mut limiter = FirstAuditLimiter::new(base); + let peer = test_peer(7); + assert_eq!(limiter.assess(&peer, 100, base, 0), LimiterVerdict::Admit); + limiter.commit_launch(peer, 100, base); + + // A rotated pin with a similar count inside the window is dropped... + let soon = base + Duration::from_secs(60); + assert_eq!( + limiter.assess(&peer, 100, soon, 0), + LimiterVerdict::WindowDeduped + ); + // ...even at the exact jump boundary (new*DEN == old*NUM is no jump)... + assert_eq!( + limiter.assess(&peer, 150, soon, 0), + LimiterVerdict::WindowDeduped + ); + // ...but a >1.5x committed-count jump re-nominates immediately (an + // inflated sidecar-only pin is invisible to the gossip lottery, so + // the window must not shield it). + assert_eq!(limiter.assess(&peer, 151, soon, 0), LimiterVerdict::Admit); + + // Window expiry re-admits an unchanged count. + let expired = base + config::FIRST_AUDIT_PEER_REAUDIT_INTERVAL; + assert_eq!( + limiter.assess(&peer, 100, expired, 0), + LimiterVerdict::Admit + ); + } + + #[test] + fn first_audit_limiter_window_verdict_outranks_empty_budget() { + // A window-deduped nomination must be DROPPED, not kept pending as + // rate-deferred, even when the bucket is also empty: re-queuing a + // suppressed rotation would hold a pending slot for two hours. + let base = Instant::now(); + let mut limiter = FirstAuditLimiter::new(base); + let peer = test_peer(9); + assert_eq!(limiter.assess(&peer, 10, base, 0), LimiterVerdict::Admit); + limiter.commit_launch(peer, 10, base); + let other = test_peer(10); + assert_eq!(limiter.assess(&other, 10, base, 0), LimiterVerdict::Admit); + limiter.commit_launch(other, 10, base); + assert_eq!( + limiter.assess(&peer, 10, base, 0), + LimiterVerdict::WindowDeduped + ); + } + + /// ADR-0004 Amendment 2 (E'): the B horizon prefilter rejects a quote that + /// is answerable now but would age out of the answerability window during + /// the launch jitter, so scheduling state is only ever committed for a pin + /// that can still be challenged when it actually sends. + #[test] + fn first_audit_horizon_prefilter_boundary() { + let now = SystemTime::now(); + // C = the too-old cutoff; H = the worst-case send horizon. + let c = GOSSIP_ANSWERABILITY_TTL.saturating_sub(MONETIZED_AUDIT_SKEW_MARGIN); + let h = config::FIRST_AUDIT_LAUNCH_JITTER_MAX + FIRST_AUDIT_SEND_LATENCY_SLACK; + // A quote whose age is exactly C at `now + H` (in the past, since C > H). + let boundary = now + .checked_add(h) + .and_then(|t| t.checked_sub(c)) + .expect("boundary time"); + + // In window at `now` (age = C - H < C)... + assert!(quote_within_audit_window(boundary, now)); + // ...but the horizon prefilter rejects it (age == C at now + H). + assert!(!quote_answerable_through_nominal_jitter(boundary, now)); + + // A hair newer stays answerable through the horizon; a hair older does + // not. Use 1µs (not 1ns): Windows `SystemTime` has 100ns granularity, so + // a nanosecond step would round to the same instant there. + let newer = boundary + .checked_add(Duration::from_micros(1)) + .expect("newer"); + let older = boundary + .checked_sub(Duration::from_micros(1)) + .expect("older"); + assert!(quote_answerable_through_nominal_jitter(newer, now)); + assert!(!quote_answerable_through_nominal_jitter(older, now)); + + // A quote too far in the FUTURE is rejected at `now`, independent of the + // horizon. + let future = now + .checked_add(MONETIZED_AUDIT_SKEW_MARGIN) + .and_then(|t| t.checked_add(Duration::from_secs(60))) + .expect("future"); + assert!(!quote_answerable_through_nominal_jitter(future, now)); + } + + /// ADR-0004 Amendment 2 (E'): a reservation whose AUTHORITATIVE post-jitter + /// answerability check fails at promotion is fully state-neutral — it stamps + /// no `first_audited`, no per-peer window, refunds its token, releases its + /// in-flight slot, does not flip the lane, and does not count a launch — and + /// a same-peer, same-count successor enqueued DURING the reservation is + /// retained and becomes the next reservation after the cancel. This is the + /// exact hole the reviewer flagged: suppression must never outlive a launch + /// that did not send. + #[test] + #[allow(clippy::too_many_lines)] + 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); + + // A pre-existing window sentinel for an UNRELATED peer must survive the + // cancel byte-for-byte (cancel never touches `recent`). + let sentinel_peer = test_peer(0xAA); + scheduler.limiter.promote(sentinel_peer, 500, mono); + let sentinel_before = scheduler + .limiter + .recent + .peek(&sentinel_peer) + .copied() + .expect("sentinel present"); + + let peer = test_peer(1); + let a = MonetizedPinEvent { + peer, + pin: [1; 32], + key_count: 100, + quote_ts: SystemTime::now(), // fresh: passes the horizon prefilter now + }; + scheduler.enqueue(a, &obs); + + let tokens_before = scheduler.tokens(); + let lane_before = scheduler.oldest_first_lane; + let launched_before = obs.launched.load(Ordering::Relaxed); + let mut cooldown: HashMap = HashMap::new(); + + // Reserve A (jitter 0 so it is immediately due). + let inflight0 = obs.inflight.load(Ordering::Relaxed); + assert!(scheduler.try_reserve(mono, inflight0, Duration::ZERO, &cooldown, &obs)); + assert_eq!(scheduler.reserved_peer(), Some(peer)); + assert_eq!( + scheduler.tokens(), + tokens_before - 1, + "reserve consumes a token" + ); + assert_eq!( + obs.inflight.load(Ordering::Relaxed), + 1, + "reserve holds a slot" + ); + + // A same-peer, same-count successor arrives DURING the reservation. It + // must be retained (bypasses the window for the reserved peer) and must + // NOT create a second reservation. + let b = MonetizedPinEvent { + peer, + pin: [2; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }; + scheduler.enqueue(b, &obs); + assert_eq!(scheduler.pending_len(), 1, "successor retained in pending"); + assert!( + !scheduler.try_reserve( + mono, + obs.inflight.load(Ordering::Relaxed), + Duration::ZERO, + &cooldown, + &obs + ), + "no second reservation while one is outstanding" + ); + + // Resolve A with an injected wall time PAST A's answerability cutoff. + let reservation = scheduler + .take_due_reservation(mono) + .expect("A is due at jitter 0"); + let wall_fail = a + .quote_ts + .checked_add(GOSSIP_ANSWERABILITY_TTL) + .and_then(|t| t.checked_add(Duration::from_secs(1))) + .expect("past-cutoff wall time"); + let promoted = scheduler.resolve(reservation, wall_fail, mono, &mut cooldown, &obs); + assert!( + promoted.is_none(), + "answerability lapsed -> cancelled, not promoted" + ); + + // State-neutral cancel. + assert!(scheduler.first_audited.is_empty(), "no pin marked audited"); + assert!( + scheduler.limiter.recent.peek(&peer).is_none(), + "cancel stamps no per-peer window" + ); + assert_eq!( + scheduler.limiter.recent.peek(&sentinel_peer).copied(), + Some(sentinel_before), + "unrelated window sentinel untouched" + ); + assert!(!cooldown.contains_key(&peer), "cancel stamps no cooldown"); + assert_eq!( + scheduler.tokens(), + tokens_before, + "token refunded on cancel" + ); + assert_eq!( + obs.inflight.load(Ordering::Relaxed), + 0, + "in-flight slot released" + ); + assert_eq!(scheduler.oldest_first_lane, lane_before, "lane not flipped"); + assert_eq!( + obs.launched.load(Ordering::Relaxed), + launched_before, + "no launch counted" + ); + assert_eq!( + obs.outside_answerability_window.load(Ordering::Relaxed), + 1, + "the cancel reason is recorded" + ); + + // The successor is still pending and is now fully schedulable: a fresh + // reserve makes B the next reservation (proving eligibility, not merely + // that the limiter would admit it). + assert_eq!( + scheduler.pending_len(), + 1, + "successor still pending after cancel" + ); + assert!(scheduler.reserved.is_none()); + assert!(scheduler.try_reserve( + mono, + obs.inflight.load(Ordering::Relaxed), + Duration::ZERO, + &cooldown, + &obs + )); + assert_eq!( + scheduler.reserved_peer(), + Some(peer), + "the retained successor becomes the next reservation" + ); + } + + /// ADR-0004 Amendment 2 (E'): when a promotion loses the shared-cooldown + /// race, the reserved event is requeued ONLY if no same-peer successor is + /// already pending. A successor arrived after the reservation, so it is the + /// newer nomination (e.g. a count jump) and must not be overwritten by the + /// older reserved event. + #[test] + 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 peer = test_peer(1); + + // Reserve A. + let a = MonetizedPinEvent { + peer, + pin: [1; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }; + scheduler.enqueue(a, &obs); + let cooldown_reserve: HashMap = HashMap::new(); + assert!(scheduler.try_reserve(mono, 0, Duration::ZERO, &cooldown_reserve, &obs)); + + // A newer same-peer successor B (a count jump) arrives during the + // reservation and is retained. + let b = MonetizedPinEvent { + peer, + pin: [2; 32], + key_count: 400, + quote_ts: SystemTime::now(), + }; + scheduler.enqueue(b, &obs); + assert_eq!(scheduler.pending_len(), 1); + + // Resolve A: answerability PASSES (fresh quote) but the shared cooldown + // is already stamped for the peer, so promotion loses the race. + let reservation = scheduler.take_due_reservation(mono).expect("due"); + let mut cooldown: HashMap = HashMap::new(); + cooldown.insert(peer, mono); // freshly on cooldown + let promoted = scheduler.resolve(reservation, SystemTime::now(), mono, &mut cooldown, &obs); + assert!(promoted.is_none(), "cooldown race -> not promoted"); + + // B (newer) is preserved; A did NOT overwrite it. + assert_eq!(scheduler.pending_len(), 1, "still exactly one pending"); + assert_eq!( + scheduler.pending.peek(&peer).map(|e| e.pin), + Some([2; 32]), + "the newer successor B is retained, not the older reserved A" + ); + assert!(scheduler.first_audited.is_empty()); + assert!(scheduler.limiter.recent.peek(&peer).is_none()); + } + + /// A flood of strictly-lower-count same-peer nominations must neither + /// displace the retained higher pin NOR disturb its LRU position (each is + /// suppressed via `peek`, no `push`), and each must be counted as + /// `suppressed_lower` — the attempted cheaper-pin self-erasure signal. + #[test] + fn first_audit_suppressed_lower_flood_leaves_recency_and_counts() { + let mut pending: LruCache = + LruCache::new(NonZeroUsize::new(2).unwrap()); + let victim = test_peer(1); + let other = test_peer(2); + // Victim (high count) inserted first (older), then `other` (newer/MRU). + let _ = coalesce_first_audit_event( + &mut pending, + MonetizedPinEvent { + peer: victim, + pin: [1; 32], + key_count: 400, + quote_ts: SystemTime::now(), + }, + true, ); + let _ = coalesce_first_audit_event( + &mut pending, + MonetizedPinEvent { + peer: other, + pin: [9; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + true, + ); + + // Flood the victim with cheaper nominations. + let mut suppressed = 0u64; + for i in 0..8u8 { + let out = coalesce_first_audit_event( + &mut pending, + MonetizedPinEvent { + peer: victim, + pin: [i; 32], + key_count: 50, + quote_ts: SystemTime::now(), + }, + true, + ); + assert_eq!(out, FirstAuditQueueOutcome::SuppressedLower); + suppressed += 1; + } + assert_eq!(suppressed, 8); + // Victim pin/count unchanged. assert_eq!( - queue_first_audit_event(&mut pending, other_peer), + pending.peek(&victim).map(|e| (e.pin, e.key_count)), + Some(([1; 32], 400)) + ); + // Recency unchanged: `other` is still MRU (a new distinct peer at cap 2 + // would evict the LRU; the victim must be the LRU, so `third` evicts the + // victim, not `other`). + let third = test_peer(3); + let out = coalesce_first_audit_event( + &mut pending, + MonetizedPinEvent { + peer: third, + pin: [7; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + true, + ); + assert_eq!( + out, FirstAuditQueueOutcome::CapacityEvicted { - peer: first.peer, - pin: replacement.pin, - } + peer: victim, + pin: [1; 32] + }, + "the suppressed-lower flood must not have promoted the victim above `other`" ); - assert_eq!(pending.len(), 1); + } + + /// The cooldown-race requeue counts a genuine different-peer capacity + /// eviction (the ADR promises capacity loss is observable). + #[test] + 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); + // 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); + let other_peer = test_peer(2); + + // Reserve peer 1. + scheduler.enqueue( + MonetizedPinEvent { + peer: reserved_peer, + pin: [1; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + &obs, + ); + let cooldown_read: HashMap = HashMap::new(); + assert!(scheduler.try_reserve(mono, 0, Duration::ZERO, &cooldown_read, &obs)); + + // A DIFFERENT peer fills the single pending slot during the reservation. + scheduler.enqueue( + MonetizedPinEvent { + peer: other_peer, + pin: [2; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + &obs, + ); + assert_eq!(scheduler.pending_len(), 1); + + // Resolve peer 1: cooldown race -> requeue peer 1, evicting peer 2. + let reservation = scheduler.take_due_reservation(mono).expect("due"); + let mut cooldown: HashMap = HashMap::new(); + cooldown.insert(reserved_peer, mono); + let cap_before = obs.capacity_evicted.load(Ordering::Relaxed); + assert!(scheduler + .resolve(reservation, SystemTime::now(), mono, &mut cooldown, &obs) + .is_none()); + assert_eq!( + obs.capacity_evicted.load(Ordering::Relaxed), + cap_before + 1, + "the requeue eviction of a different peer is counted" + ); + assert_eq!( + scheduler.pending.peek(&reserved_peer).map(|e| e.pin), + Some([1; 32]) + ); + } + + /// ADR-0004 Amendment 2 (reviewer blocker): a strictly-lower-count same-peer + /// nomination arriving while an inflated pin is PENDING must not displace it. + /// The inflated pin stays and is the one launched. + #[test] + 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 peer = test_peer(1); + + // Inflated (high-count) sidecar pin lands in pending. + scheduler.enqueue( + MonetizedPinEvent { + peer, + pin: [1; 32], + key_count: 400, + quote_ts: SystemTime::now(), + }, + &obs, + ); + // A cheaper same-peer settlement arrives right after. + scheduler.enqueue( + MonetizedPinEvent { + peer, + pin: [2; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + &obs, + ); + + assert_eq!(scheduler.pending_len(), 1); + assert_eq!( + scheduler.pending.peek(&peer).map(|e| (e.pin, e.key_count)), + Some(([1; 32], 400)), + "the inflated pin must not be erased by the cheaper successor" + ); + // The dropped cheaper nomination is counted through the enqueue path. + assert_eq!( + obs.suppressed_lower.load(Ordering::Relaxed), + 1, + "the attempted cheaper-pin self-erasure is observable" + ); + + // It reserves and promotes as the inflated pin/count. + let cooldown_read: HashMap = HashMap::new(); + assert!(scheduler.try_reserve(mono, 0, Duration::ZERO, &cooldown_read, &obs)); + let reservation = scheduler.take_due_reservation(mono).expect("due"); + let mut cooldown: HashMap = HashMap::new(); + let (event, _slot) = scheduler + .resolve(reservation, SystemTime::now(), mono, &mut cooldown, &obs) + .expect("promotes"); + assert_eq!((event.pin, event.key_count), ([1; 32], 400)); + } + + /// ADR-0004 Amendment 2 (reviewer blocker): a RESERVED inflated pin that + /// loses the cooldown race must be requeued OVER a lower-count same-peer + /// successor that arrived during its jitter, and must remain launchable once + /// the shared cooldown expires — the cheaper successor cannot suppress it. + #[test] + 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 peer = test_peer(1); + + // Reserve the inflated pin. + scheduler.enqueue( + MonetizedPinEvent { + peer, + pin: [1; 32], + key_count: 400, + quote_ts: SystemTime::now(), + }, + &obs, + ); + let cooldown_read: HashMap = HashMap::new(); + assert!(scheduler.try_reserve(mono, 0, Duration::ZERO, &cooldown_read, &obs)); + + // A cheaper successor arrives during the reservation (bypasses the + // window as the reserved peer) and sits in pending. + scheduler.enqueue( + MonetizedPinEvent { + peer, + pin: [2; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + &obs, + ); + assert_eq!(scheduler.pending_len(), 1); + + // Resolve: answerability passes but the shared cooldown is already + // stamped, so the reservation loses the race and requeues. + let reservation = scheduler.take_due_reservation(mono).expect("due"); + let mut cooldown: HashMap = HashMap::new(); + cooldown.insert(peer, mono); + assert!(scheduler + .resolve(reservation, SystemTime::now(), mono, &mut cooldown, &obs) + .is_none()); + + // The inflated pin (400) replaced the cheaper successor (100). + assert_eq!(scheduler.pending_len(), 1); assert_eq!( - pending.peek(&other_peer.peer).map(|event| event.pin), - Some([3; 32]) + scheduler.pending.peek(&peer).map(|e| (e.pin, e.key_count)), + Some(([1; 32], 400)), + "the inflated reserved pin must survive the requeue over the cheaper successor" ); + assert!(scheduler.first_audited.is_empty()); + assert!(scheduler.limiter.recent.peek(&peer).is_none()); + + // Once the shared cooldown expires, the inflated pin reserves and + // promotes with its intended pin/count. + let later = mono + .checked_add(Duration::from_secs( + config::AUDIT_ON_GOSSIP_COOLDOWN_SECS + 1, + )) + .expect("later"); + let cooldown_read_later: HashMap = HashMap::new(); + assert!(scheduler.try_reserve(later, 0, Duration::ZERO, &cooldown_read_later, &obs)); + let reservation = scheduler.take_due_reservation(later).expect("due"); + let mut cooldown_later: HashMap = HashMap::new(); + let (event, _slot) = scheduler + .resolve( + reservation, + SystemTime::now(), + later, + &mut cooldown_later, + &obs, + ) + .expect("promotes after cooldown"); + assert_eq!((event.pin, event.key_count), ([1; 32], 400)); + } + + /// ADR-0004 Amendment 2 (E'): consecutive PROMOTIONS strictly alternate the + /// launch lane, driven through the real scheduler (reserve -> resolve -> + /// promote), so a stream of fresh nominations cannot keep every launch on + /// the newest lane and starve the oldest. + #[test] + 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 cooldown: HashMap = HashMap::new(); + + // Two distinct peers; the newest-inserted is the MRU (newest lane end). + let oldest_peer = test_peer(1); + let newest_peer = test_peer(2); + scheduler.enqueue( + MonetizedPinEvent { + peer: oldest_peer, + pin: [1; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + &obs, + ); + scheduler.enqueue( + MonetizedPinEvent { + peer: newest_peer, + pin: [2; 32], + key_count: 100, + quote_ts: SystemTime::now(), + }, + &obs, + ); + + let mut launched_peers = Vec::new(); + for _ in 0..2 { + assert!(scheduler.try_reserve( + mono, + obs.inflight.load(Ordering::Relaxed), + Duration::ZERO, + &cooldown, + &obs + )); + let reservation = scheduler.take_due_reservation(mono).expect("due"); + // Answerable wall time == the quote's own time (age 0). + let (event, _slot) = scheduler + .resolve(reservation, SystemTime::now(), mono, &mut cooldown, &obs) + .expect("fresh in-window quote promotes"); + launched_peers.push(event.peer); + } + + // First launch takes the newest lane (lane starts false = newest), the + // second takes the oldest lane: strict alternation. + assert_eq!( + launched_peers, + vec![newest_peer, oldest_peer], + "consecutive promotions alternate newest-then-oldest lane" + ); + } + + #[test] + fn first_audit_count_jump_boundaries() { + // Exactly 1.5x is NOT a jump; strictly above is. + assert!(!first_audit_count_jump(100, 150)); + assert!(first_audit_count_jump(100, 151)); + // Anything beats an audited zero; zero never jumps. + assert!(first_audit_count_jump(0, 1)); + assert!(!first_audit_count_jump(0, 0)); + // Equal max counts must not jump (and must not overflow). + assert!(!first_audit_count_jump(u32::MAX, u32::MAX)); } #[test] diff --git a/tests/e2e/first_audit_ab.rs b/tests/e2e/first_audit_ab.rs new file mode 100644 index 00000000..2425a363 --- /dev/null +++ b/tests/e2e/first_audit_ab.rs @@ -0,0 +1,537 @@ +//! ADR-0004 first-audit A/B workload driver (NOT a pass/fail regression test). +//! +//! Drives a production-shaped paid-upload workload against a local testnet +//! with real on-chain payment verification (Anvil), so the monetized +//! first-audit pipeline runs end-to-end exactly as in production: +//! quote collection (with commitment sidecars) -> median payment on chain -> +//! `ChunkPutRequest` with the full proof -> storer-side `verify_payment` -> +//! monetized-pin nomination -> first-audit drainer -> subtree audits over QUIC. +//! +//! Run the SAME file against two builds (baseline `main` vs the scheduler-fix +//! branch) and compare the emitted `AB-METRIC` lines plus log-event counts +//! (audit launches, timeouts, responder drops). Skipped entirely unless +//! `FIRST_AUDIT_AB=1` is set, so it never runs in CI. +//! +//! ```bash +//! FIRST_AUDIT_AB=1 RUST_LOG=info,ant_node::replication=debug \ +//! cargo test --test e2e first_audit_ab_workload -- --nocapture +//! ``` +//! +//! Knobs (env): `AB_NODES`, `AB_SEED_UPLOADS`, `AB_UPLOADS`, `AB_GAP_MS`, +//! `AB_REBUILD_EVERY`, `AB_DOWNLOADS`, `AB_TAIL_SECS`, `AB_CHUNK_KIB`. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::unnested_or_patterns, + clippy::cast_precision_loss, + clippy::redundant_closure_for_method_calls +)] + +use ant_node::ant_protocol::{ + ChunkGetRequest, ChunkGetResponse, ChunkMessage, ChunkMessageBody, ChunkPutRequest, + ChunkPutResponse, ChunkQuoteRequest, ChunkQuoteResponse, CLOSE_GROUP_SIZE, +}; +use ant_node::client::send_and_await_chunk_response; +use ant_node::payment::{serialize_single_node_proof, PaymentProof}; +use evmlib::common::Amount; +use evmlib::{EncodedPeerId, ProofOfPayment}; +use rand::Rng; +use saorsa_core::identity::PeerId; +use saorsa_core::P2PNode; +use serial_test::serial; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use super::anvil::TestAnvil; +use super::testnet::{TestNetwork, TestNetworkConfig}; + +/// The verifier accepts a settlement of at least 3x the median quote price. +/// Mirrors `PAID_QUOTE_PAYMENT_MULTIPLIER` (private to the verifier). +const PAYMENT_MULTIPLIER: u64 = 3; + +/// Per-request timeout for quote/PUT/GET protocol round trips. +const REQUEST_TIMEOUT: Duration = Duration::from_secs(20); + +fn env_usize(name: &str, default: usize) -> usize { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +fn env_u64(name: &str, default: u64) -> u64 { + std::env::var(name) + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(default) +} + +/// One collected quote: the issuing peer, the decoded quote, and the raw +/// commitment sidecar blob (when the quote pinned a commitment). +struct CollectedQuote { + peer: PeerId, + quote: evmlib::PaymentQuote, + sidecar: Option>, +} + +/// Request a signed quote (plus ADR-0004 commitment sidecar) from one peer. +async fn request_quote( + client: &Arc, + peer: &PeerId, + address: [u8; 32], + data_size: u64, +) -> Result { + let request_id: u64 = rand::thread_rng().gen(); + let message = ChunkMessage { + request_id, + body: ChunkMessageBody::QuoteRequest(ChunkQuoteRequest { + address, + data_size, + data_type: 0, + }), + }; + let message_bytes = message + .encode() + .map_err(|e| format!("encode quote request: {e}"))?; + + let (quote_bytes, sidecar) = send_and_await_chunk_response( + client, + peer, + message_bytes, + request_id, + REQUEST_TIMEOUT, + &[], + |body| match body { + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Success { + quote, + already_stored: _, + commitment, + }) => Some(Ok((quote, commitment))), + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(e)) => { + Some(Err(format!("quote error: {e:?}"))) + } + _ => None, + }, + |e| format!("send quote request: {e}"), + || "quote request timed out".to_string(), + ) + .await?; + + let quote: evmlib::PaymentQuote = + rmp_serde::from_slice("e_bytes).map_err(|e| format!("decode quote: {e}"))?; + Ok(CollectedQuote { + peer: *peer, + quote, + sidecar, + }) +} + +/// Send a paid PUT to one peer; `Ok(true)` when it stored (or already had it). +async fn paid_put( + client: &Arc, + peer: &PeerId, + address: [u8; 32], + data: &[u8], + proof_bytes: Vec, +) -> Result { + let request_id: u64 = rand::thread_rng().gen(); + let message = ChunkMessage { + request_id, + body: ChunkMessageBody::PutRequest(ChunkPutRequest::with_payment( + address, + bytes::Bytes::copy_from_slice(data), + proof_bytes, + )), + }; + let message_bytes = message + .encode() + .map_err(|e| format!("encode put request: {e}"))?; + + send_and_await_chunk_response( + client, + peer, + message_bytes, + request_id, + REQUEST_TIMEOUT, + &[], + |body| match body { + ChunkMessageBody::PutResponse(ChunkPutResponse::Success { .. }) + | ChunkMessageBody::PutResponse(ChunkPutResponse::AlreadyExists { .. }) => { + Some(Ok(true)) + } + ChunkMessageBody::PutResponse(ChunkPutResponse::PaymentRequired { message }) => { + Some(Err(format!("payment required: {message}"))) + } + ChunkMessageBody::PutResponse(ChunkPutResponse::Error(e)) => { + Some(Err(format!("put error: {e:?}"))) + } + _ => None, + }, + |e| format!("send put: {e}"), + || "put timed out".to_string(), + ) + .await +} + +/// Fetch a chunk from one peer; `Ok(true)` when the bytes came back. +async fn get_chunk( + client: &Arc, + peer: &PeerId, + address: [u8; 32], +) -> Result { + let request_id: u64 = rand::thread_rng().gen(); + let message = ChunkMessage { + request_id, + body: ChunkMessageBody::GetRequest(ChunkGetRequest::new(address)), + }; + let message_bytes = message + .encode() + .map_err(|e| format!("encode get request: {e}"))?; + + send_and_await_chunk_response( + client, + peer, + message_bytes, + request_id, + REQUEST_TIMEOUT, + &[], + |body| match body { + ChunkMessageBody::GetResponse(ChunkGetResponse::Success { .. }) => Some(Ok(true)), + ChunkMessageBody::GetResponse(ChunkGetResponse::NotFound { .. }) => Some(Ok(false)), + ChunkMessageBody::GetResponse(ChunkGetResponse::Error(e)) => { + Some(Err(format!("get error: {e:?}"))) + } + _ => None, + }, + |e| format!("send get: {e}"), + || "get timed out".to_string(), + ) + .await +} + +/// Outcome of one full paid upload. +struct UploadOutcome { + duration: Duration, + pinned_quotes: usize, + total_quotes: usize, + stored_acks: usize, +} + +/// One production-shaped paid upload: collect quotes from the close group, +/// pay the median issuer 3x on chain, then PUT the chunk (with the full quote +/// bundle + sidecars) to every close-group peer. +async fn paid_upload( + client: &Arc, + client_peer: &PeerId, + wallet: &evmlib::wallet::Wallet, + data: &[u8], +) -> Result<(UploadOutcome, [u8; 32]), String> { + let address = ant_node::compute_address(data); + let started = Instant::now(); + + let closest = client + .dht() + .find_closest_nodes(&address, CLOSE_GROUP_SIZE + 1) + .await + .map_err(|e| format!("find closest: {e}"))?; + let storers: Vec = closest + .iter() + .map(|n| n.peer_id) + .filter(|p| p != client_peer) + .take(CLOSE_GROUP_SIZE) + .collect(); + if storers.is_empty() { + return Err("no close-group peers found".to_string()); + } + + let mut quotes: Vec = Vec::with_capacity(storers.len()); + for peer in &storers { + match request_quote(client, peer, address, data.len() as u64).await { + Ok(q) => quotes.push(q), + Err(e) => eprintln!("AB-WARN quote from {peer} failed: {e}"), + } + } + if quotes.is_empty() { + return Err("no quotes collected".to_string()); + } + + // Median selection mirrors the node verifier: sort by price, index len/2, + // settle 3x that price to the median issuer. + quotes.sort_by_key(|q| q.quote.price); + let median = quotes + .get(quotes.len() / 2) + .ok_or_else(|| "median index out of range".to_string())?; + let amount = median + .quote + .price + .checked_mul(Amount::from(PAYMENT_MULTIPLIER)) + .ok_or_else(|| "payment amount overflow".to_string())?; + let (tx_by_quote, _gas) = wallet + .pay_for_quotes(vec![( + median.quote.hash(), + median.quote.rewards_address, + amount, + )]) + .await + .map_err(|e| format!("pay_for_quotes: {e:?}"))?; + let tx_hashes: Vec = tx_by_quote.values().copied().collect(); + + let pinned_quotes = quotes + .iter() + .filter(|q| q.quote.commitment_pin.is_some()) + .count(); + let total_quotes = quotes.len(); + let proof = PaymentProof { + proof_of_payment: ProofOfPayment { + peer_quotes: quotes + .iter() + .map(|q| (EncodedPeerId::new(*q.peer.as_bytes()), q.quote.clone())) + .collect(), + }, + tx_hashes, + commitment_sidecars: quotes.iter().filter_map(|q| q.sidecar.clone()).collect(), + }; + let proof_bytes = + serialize_single_node_proof(&proof).map_err(|e| format!("serialize proof: {e}"))?; + + let mut stored_acks = 0usize; + for peer in &storers { + match paid_put(client, peer, address, data, proof_bytes.clone()).await { + Ok(true) => stored_acks += 1, + Ok(false) => {} + Err(e) => eprintln!("AB-WARN put to {peer} failed: {e}"), + } + } + if stored_acks == 0 { + return Err("no storer accepted the paid put".to_string()); + } + + Ok(( + UploadOutcome { + duration: started.elapsed(), + pinned_quotes, + total_quotes, + stored_acks, + }, + address, + )) +} + +/// Rebuild every node's storage commitment (simulates the hourly rotation +/// that re-arms pin-level dedup in production). +async fn rebuild_all_commitments(network: &TestNetwork) { + for node in network.nodes() { + if let Some(engine) = node.replication_engine.as_ref() { + if let Err(e) = engine.rebuild_commitment_now().await { + eprintln!( + "AB-WARN commitment rebuild failed on node {}: {e}", + node.index + ); + } + } + } +} + +fn percentile_ms(sorted: &[Duration], pct: f64) -> u128 { + if sorted.is_empty() { + return 0; + } + let last = sorted.len() - 1; + #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] + let idx = ((sorted.len() as f64 * pct) as usize).min(last); + sorted.get(idx).map_or(0, |d| d.as_millis()) +} + +#[tokio::test(flavor = "multi_thread")] +#[serial] +#[allow(clippy::too_many_lines)] +async fn first_audit_ab_workload() { + if std::env::var("FIRST_AUDIT_AB").ok().as_deref() != Some("1") { + eprintln!("first_audit_ab_workload skipped: set FIRST_AUDIT_AB=1 to run"); + return; + } + // Route node logs to stderr so the runner script can count audit events. + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .with_writer(std::io::stderr) + .try_init(); + + let nodes = env_usize("AB_NODES", 20); + let seed_uploads = env_usize("AB_SEED_UPLOADS", 25); + let uploads = env_usize("AB_UPLOADS", 40); + let gap_ms = env_u64("AB_GAP_MS", 2000); + let rebuild_every = env_usize("AB_REBUILD_EVERY", 15); + let downloads = env_usize("AB_DOWNLOADS", 15); + let tail_secs = env_u64("AB_TAIL_SECS", 90); + let chunk_kib = env_usize("AB_CHUNK_KIB", 64); + + eprintln!( + "AB-CONFIG nodes={nodes} seed_uploads={seed_uploads} uploads={uploads} gap_ms={gap_ms} \ + rebuild_every={rebuild_every} downloads={downloads} tail_secs={tail_secs} chunk_kib={chunk_kib}" + ); + + // Anvil FIRST so nodes verify payments against it for real. + let anvil = TestAnvil::new().await.expect("start anvil"); + let wallet = anvil.create_funded_wallet().expect("funded wallet"); + + let mut config = TestNetworkConfig { + node_count: nodes, + payment_enforcement: true, + ..TestNetworkConfig::default() + }; + config.evm_network = Some(anvil.to_network()); + let mut network = TestNetwork::new(config).await.expect("build network"); + network.start().await.expect("start network"); + network.warmup_dht().await.expect("warmup dht"); + + // Wire each node exactly like production `node.rs`: commitment source into + // the quote generator, gossip cache into the verifier, and the + // monetized-pin sender into the verifier (the trio the e2e testnet does + // not wire by default). + for node in network.nodes() { + if let (Some(protocol), Some(engine)) = + (node.ant_protocol.as_ref(), node.replication_engine.as_ref()) + { + let concrete = Arc::clone(engine.commitment_state()); + let source: Arc = concrete; + protocol.attach_commitment_source(source); + protocol + .payment_verifier_arc() + .attach_commitment_cache(Arc::clone(engine.last_commitment_by_peer())); + protocol + .payment_verifier_arc() + .attach_monetized_pin_sender(engine.monetized_pin_sender()); + } + } + + // The last node acts as the uploading/downloading client. + let client_index = nodes.saturating_sub(1); + let client = network + .nodes() + .get(client_index) + .and_then(|n| n.p2p_node.as_ref()) + .map(Arc::clone) + .expect("client node p2p"); + let client_peer = *client.peer_id(); + + let chunk_bytes = chunk_kib * 1024; + let mut rng = rand::thread_rng(); + let mut make_chunk = move || { + let mut data = vec![0u8; chunk_bytes]; + rng.fill(&mut data[..]); + data + }; + + // ---- Seed phase: give nodes stored chunks, then build commitments so + // subsequent quotes carry pins (production steady state). + let mut seeded = 0usize; + for _ in 0..seed_uploads { + let data = make_chunk(); + match paid_upload(&client, &client_peer, &wallet, &data).await { + Ok(_) => seeded += 1, + Err(e) => eprintln!("AB-WARN seed upload failed: {e}"), + } + } + eprintln!("AB-METRIC phase=seed uploads_ok={seeded}/{seed_uploads}"); + rebuild_all_commitments(&network).await; + tokio::time::sleep(Duration::from_secs(5)).await; + + // ---- Measurement phase: steady paid uploads, with periodic commitment + // rotation (the production re-arming mechanism). + let mut upload_durations: Vec = Vec::with_capacity(uploads); + let mut uploaded_addresses: Vec<[u8; 32]> = Vec::with_capacity(uploads); + let mut pinned_total = 0usize; + let mut quotes_total = 0usize; + let mut acks_total = 0usize; + let mut failures = 0usize; + let measure_started = Instant::now(); + for i in 0..uploads { + if i > 0 && rebuild_every > 0 && i % rebuild_every == 0 { + rebuild_all_commitments(&network).await; + eprintln!("AB-METRIC phase=measure rotation_at_upload={i}"); + } + let data = make_chunk(); + match paid_upload(&client, &client_peer, &wallet, &data).await { + Ok((outcome, address)) => { + eprintln!( + "AB-METRIC phase=measure upload={i} ms={} pinned={}/{} acks={}", + outcome.duration.as_millis(), + outcome.pinned_quotes, + outcome.total_quotes, + outcome.stored_acks + ); + pinned_total += outcome.pinned_quotes; + quotes_total += outcome.total_quotes; + acks_total += outcome.stored_acks; + upload_durations.push(outcome.duration); + uploaded_addresses.push(address); + } + Err(e) => { + failures += 1; + eprintln!("AB-WARN measure upload {i} failed: {e}"); + } + } + tokio::time::sleep(Duration::from_millis(gap_ms)).await; + } + + // ---- Download phase. + let mut download_durations: Vec = Vec::with_capacity(downloads); + let mut download_failures = 0usize; + for (i, address) in uploaded_addresses.iter().rev().take(downloads).enumerate() { + let started = Instant::now(); + let closest = client + .dht() + .find_closest_nodes(address, CLOSE_GROUP_SIZE) + .await + .unwrap_or_default(); + let mut found = false; + for peer in closest.iter().map(|n| n.peer_id) { + if peer == client_peer { + continue; + } + if get_chunk(&client, &peer, *address).await == Ok(true) { + found = true; + break; + } + } + if found { + let elapsed = started.elapsed(); + eprintln!( + "AB-METRIC phase=download get={i} ms={}", + elapsed.as_millis() + ); + download_durations.push(elapsed); + } else { + download_failures += 1; + eprintln!("AB-WARN download {i} failed"); + } + } + + // ---- Tail: let in-flight and pending audits play out before teardown so + // their outcomes land in the captured logs. + eprintln!("AB-METRIC phase=tail sleeping_secs={tail_secs}"); + tokio::time::sleep(Duration::from_secs(tail_secs)).await; + + upload_durations.sort_unstable(); + download_durations.sort_unstable(); + eprintln!( + "AB-SUMMARY uploads_ok={} uploads_failed={failures} upload_p50_ms={} upload_p90_ms={} \ + downloads_ok={} downloads_failed={download_failures} download_p50_ms={} download_p90_ms={} \ + pinned_quote_ratio={pinned_total}/{quotes_total} storer_acks={acks_total} \ + measure_wall_secs={}", + upload_durations.len(), + percentile_ms(&upload_durations, 0.50), + percentile_ms(&upload_durations, 0.90), + download_durations.len(), + percentile_ms(&download_durations, 0.50), + percentile_ms(&download_durations, 0.90), + measure_started.elapsed().as_secs(), + ); + + network.shutdown().await.expect("shutdown network"); +} diff --git a/tests/e2e/mod.rs b/tests/e2e/mod.rs index 994dc31f..d5e9ae5c 100644 --- a/tests/e2e/mod.rs +++ b/tests/e2e/mod.rs @@ -66,6 +66,9 @@ mod security_attacks; #[cfg(test)] mod subtree_audit_testnet; +#[cfg(test)] +mod first_audit_ab; + pub use anvil::TestAnvil; pub use harness::TestHarness; pub use testnet::{NetworkState, NodeState, TestNetwork, TestNetworkConfig, TestNode};