diff --git a/Cargo.lock b/Cargo.lock index 7eb5b398..d158230a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -861,9 +861,8 @@ dependencies = [ [[package]] name = "ant-protocol" -version = "2.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b665880152b4e124094acc1faaff5f20a275594e6364b8a729b50f3bc326674" +version = "3.0.0" +source = "git+https://github.com/grumbach/ant-protocol?rev=2015113270bbde8204a1740674bed2ada52d2ca6#2015113270bbde8204a1740674bed2ada52d2ca6" dependencies = [ "blake3", "bytes", diff --git a/Cargo.toml b/Cargo.toml index ca894682..4dc1d29e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,7 +39,7 @@ mimalloc = "0.1" # Until then, the git pin tracks the matching saorsa-core lineage # (the rc-2026.4.2 branch) so Cargo can unify the wire types here # with ant-protocol's re-exports. -ant-protocol = "2.3.0" +ant-protocol = "3.0.0" # Core (provides EVERYTHING: networking, DHT, security, trust, storage) saorsa-core = "0.26.2" @@ -196,3 +196,11 @@ unused_async = "allow" cognitive_complexity = "allow" # Allow non-const functions during initial development (may need runtime features later) missing_const_for_fn = "allow" + +# ADR-0005 coordinated cutover: pin ant-protocol to the immutable rev of the +# audit-report wire-types PR (github.com/WithAutonomi/ant-protocol/pull/19). +# An immutable rev (not a mutable branch) keeps the build reproducible; the +# committed Cargo.lock records the exact resolution. STRIP BEFORE MERGE — +# replace with the published ant-protocol 3.0.0 once it lands on crates.io. +[patch.crates-io] +ant-protocol = { git = "https://github.com/grumbach/ant-protocol", rev = "2015113270bbde8204a1740674bed2ada52d2ca6" } diff --git a/docs/adr/ADR-0005-earned-reward-eligibility.md b/docs/adr/ADR-0005-earned-reward-eligibility.md new file mode 100644 index 00000000..7f48d1d9 --- /dev/null +++ b/docs/adr/ADR-0005-earned-reward-eligibility.md @@ -0,0 +1,249 @@ +# ADR-0005: Earned reward eligibility + +- **Status:** Proposed +- **Date:** 2026-07-10 +- **Decision owners:** Anselme (@grumbach) +- **Reviewers:** +- **Supersedes:** none +- **Superseded by:** none +- **Related:** ADR-0002 (gossip-triggered contiguous-subtree storage audit — the work this ADR reads), ADR-0003 (full-node detection and eviction — the trust/eviction lane), ADR-0004 (commitment-bound quote pricing — bounds the per-payment cheat window this ADR builds on) + +## Framing: earned future eligibility, not slashing + +This ADR is an **incentive to keep serving**, layered on the network's existing +lost-trust model. It gates a node's eligibility for **future** rewards on a record +of useful storage work. It does **not** slash, claw back, or remove any settled +payment, and it introduces no new penalty on-chain. The only thing a node can lose +is the *opportunity to be paid next time*, until it has earned it back. "Caught +cheating" here means "this observer stops vouching for you until you re-earn a +week," never "money is taken from you." + +## Context + +ADR-0004 shrank the window in which a node can cheat on a single payment to +minutes. But a node's identity is just a free keypair: join, cheat, get caught, +throw the identity away, rejoin clean. A hit-and-run costs nothing to repeat. + +What's missing is a **sunk cost**: real work a node must put in before it can be +paid, so abandoning an identity means abandoning that work. This ADR makes that +work "roughly a week of storing real data and passing audits on it." + +We do this without three things we deliberately rule out: **no fee to join** (it +would destroy node operator UX), **no new blockchain machinery** (the chain stays +a payment rail), and **no reputation score**. A score is a complexity bomb: its +thresholds quietly break as the network's audit rate drifts, and a subjective +rating is a bad thing to gate money on. Instead we count plain facts: how many +times a node passed an audit, and when. + +## Decision + +**A node must earn its place before it can be paid**: about a week of audited +storage at the size it wants to charge for. Only clients enforce this, by +choosing who they pay; nodes and the chain are unchanged. + +- **Each node keeps a simple record of what it audited.** Every node already + audits its neighbours (ADR-0002/0004). It now writes down the result: for each + peer, how many audits it passed each day, and the largest amount of data it + proved it was holding. These are plain counts, saved to disk like any other + network state, nothing weighted or decayed, nothing to tune. Two things can + knock a peer's record down. If a node is *caught cheating* (a failed audit) its + record is wiped and it has to start the week over; that mark sticks for the full + week even as new passes come in, so getting caught costs exactly one week's + re-earning, never a permanent ban. If a node simply *goes quiet* on data it was + paid to keep, its record stops counting until it answers an audit again: silence + costs it one vouch for as long as it lasts, and nothing more. + +- **When a node quotes a price, it also shares what it has witnessed.** A quote + reply now carries the node's own audit record for the peers near that address: + positive facts only, no accusations about anyone. It's signed, and tied to a + fresh random token the client puts in each request, so an old record can't be + replayed. Quote replies are the only place these records travel; they're never + passed along in a payment, and no node ever acts on another node's record. + +- **The client decides who's eligible, from those shared records.** A node's + judges are the *other* nodes that answered the same quote request; a node never + vouches for itself. This set is exactly the responders to one request, not a + standing neighbourhood or "anyone who ever audited it" — and it rotates per + address, so an attacker cannot pre-position where its judges will be. Among the + judges that have any record of the node, it is eligible when **more than half of + them vouch for it** (but never fewer than 3 judges). A judge vouches when its + record shows the node passed audits on at least ~7 different days in the last two + weeks, at the size being quoted (with some slack), with the most recent pass + inside the last day. That last-day requirement is kept alive by the node's + routine hourly commitment refresh, which re-gossips a freshly signed commitment + and draws a fresh audit whether or not the node took in any new data, so an + honest node that is simply idle does not fall out of eligibility. Judges with no + record of the node simply abstain: a brand-new or just-arrived neighbour neither + helps nor hurts, so there's no knob to tune for them. + +- **A majority, not a fixed count.** The rule cuts both ways: to exclude a node, + more than half its judges must decline to vouch, meaning it was genuinely caught + by half the neighbourhood, or half the neighbourhood is colluding, which is the + same "capture the neighbourhood" bar the network already lives with elsewhere. A + fixed count of vouches would not do this: a node caught by nearly half its judges + could still clear a small fixed bar, whereas a majority scales the exclusion to + the size of the neighbourhood that actually watched it. + +- **The check happens when the client gathers quotes, not when it pays.** The + client already asks more nodes than it needs; it now prefers eligible ones when + filling its list. If not enough are eligible (which fast, network-wide growth can + cause, since nobody has a full week at the newly grown size), it prefers the + nodes with the most dues done, a clean audited week at any size, before falling + back to today's ungated rules. Falling back is a degraded-security mode, not a + neutral one — earned eligibility is being bypassed — so it is logged and surfaced + as a metric, letting operators see how often and where the network is running + ungated. This keeps out the two things the gate is for, fresh identities and + caught cheaters, even when it cannot enforce size, and ADR-0004 still forces a + current-size proof at payment. How the winning payee is chosen and how payment is + verified do not change at all: if the whole list is eligible, whoever wins is + eligible. + +- **Nodes that aren't yet eligible still store data.** That unpaid storage *is* the + week of dues, and nodes already store for their neighbourhood without being paid + for every chunk. + +- **This leans on ADR-0004.** ADR-0004 already forces an audit on any data a node + got paid to keep, on top of the random ones, and already treats a wrong answer + there as a definite cheat. This ADR just records those outcomes: a wrong answer + wipes the record, a silent one freezes it. So the fastest, most certain + detections land exactly on the data a node was paid to keep: the forced audit + needs no lottery, and it's the other nodes storing alongside the payee that run + it. A node that takes payment and then deletes fails that audit every time, so + "caught" marks pile up across the neighbourhood fast, while its old vouches age + out. + +- **Newcomers can't be frozen out by incumbents.** A node is judged only on its + own record, never ranked against others, so a newcomer earns its place even when + the existing neighbours are "old friends" who might prefer to keep it out. One + node abstaining can never keep anyone out. + +## Policy reference (the client aggregation rule, written out) + +So that clients and nodes share one exact meaning, the eligibility predicate a +client evaluates over the signed reports it collected is: + +- **Reporters (judges).** The other nodes that answered this client's quote + request for this address. The subject is never its own judge (self-vouch is + discarded). Reporters rotate per address by construction. +- **Opinions vs abstention.** A reporter is an *opinion* only if its report carries + a row for the subject; a reporter with no row **abstains** and is neither in the + numerator nor the denominator. Abstention can never suppress a node. +- **Vouch (numerator).** A reporter's row vouches when it is unfenced, unconvicted, + and shows ≥ **D** distinct days (default 7), each carrying at least one pass at a + proven size that covers the quoted size within a **slack** factor (default 2×), + all inside a trailing **window** (default 14 days), with the most recent covering + day no older than the **recency** bound (default 1 day). +- **Bar.** Eligible when vouches > half of opinions, floored at **3** opinions + (`bar = max(3, opinions/2 + 1)`). Fewer than 3 opinions → cannot be size-eligible + on this rule alone; it falls to the dues tier / ungated fallback below. +- **`fenced`.** Set by a monetized-pin audit timeout on the reporting node; a + fenced row does not vouch until the subject passes a fresh audit, at which point + it clears. A fence is that one observer's withheld vouch — never portable guilt. +- **`convicted`.** Set by a *confirmed* (non-timeout) subtree-audit failure. It + zeroes the day history and stays **sticky for the dues period D** even as fresh + passes accrue underneath, so one catch costs exactly one week's re-earning and + never compounds into a permanent ban. (Node and protocol MUST agree on this + sticky semantics — a convicted row does *not* clear on the next pass.) +- **Newcomer / reset path.** A node with no history simply isn't yet size-eligible; + it is not marked malicious. Under the two-tier fallback it can still be selected + via the dues tier (a clean audited week at any size) or the ungated fallback, so + honest new and churned-in nodes are never frozen out. Suppressing a qualified + newcomer requires more than half its judges to lie, i.e. neighbourhood capture. +- **Missing / invalid reports.** Distinguished and handled as: *not upgraded* (no + report field) and *no testimony* (report present, no row) → abstain; + *unreachable* → not a reporter; *invalid* (bad signature, wrong nonce, over-cap) + → dropped before parse, contributes nothing; *fenced* / *convicted* → counted as + a non-vouching opinion (in the denominator, not the numerator). + +## Consequences + +**Positive** +- Walking away from an identity throws away ≈ a week of real, audited storage; + parked or empty nodes earn nothing. +- No reputation score to tune into a fragile equilibrium: standing is plain counts + a human can read ("passed audits all week at this size"), and the client-side + settings can be adjusted without shipping new node software. +- The network never stalls: data is always stored, and if too few nodes are + eligible, which can happen under fast network-wide growth or heavy churn, + payments fall back to the most dues done and then to today's behaviour. + +**Negative / trade-offs** +- Honest new nodes earn nothing for their first ≈ week. This will need clear + operator messaging. Nodes that grow fast wait a little longer for full-size + eligibility, since the size has to have been held for the week too, though the + dues fallback still lets them earn in the meantime. +- **Some settings are client-side, some aren't.** The client can freely retune the + ~7-day requirement, the size slack, the recency window, and the enforce switch. + But the two-week memory window and how long a "caught" mark sticks are baked into + the node software, so changing *those* needs a node release. This is the one + partial exception to "nothing to tune": if audit cadence drifts far enough, those + two constants would need re-fitting in a release rather than as client policy. +- **A node's shared record has a size limit.** It reports on at most a couple dozen + peers, preferring the most recently active. In a big or fast-churning + neighbourhood that can silently leave some peers out of a given report, which + turns those judges into abstainers and makes the "more than half" count harder to + reach. An unlucky node could dip out of eligibility for a spell, but its next + audits put it straight back on the reports, so any such dip is brief and + self-correcting. +- **Reports are testimony, not proof.** A signed row is a reporter's own claim; it + carries no portable audit proof a third party could re-verify. The defence + against colluding reporters is structural, not cryptographic: the judges are the + per-address responder set (not attacker-chosen), self-vouch is excluded, and the + majority bar means smuggling a cheater in or freezing an honest node out both + require capturing more than half that set — the same neighbourhood-capture + boundary ADR-0004's median already accepts. + +**Neutral / operational** +- Each quote reply grows by the shared record: ~7 KB at a ~20-peer neighbourhood, + hard-capped at 16 KB; an over-cap report is dropped whole, never trimmed on the + wire. The producing node enforces the row, day, and serialized-byte caps before + emitting a report, so an over-cap report is never put on the wire. A single + upload gathers one report per quote responder, about 7 while the gate is only + observing and up to ~20 once it is enforcing; a batched "merkle" upload gathers + up to ~32, one from each node it over-queries. All of it stays on the client. +- **Hard wire cutover.** The new quote-request nonce and quote-response report + fields change the wire format. Because the shared payment types are + postcard-encoded (non-self-describing), old and new peers do not interoperate, so + this ships as a coordinated cutover with a bumped protocol identifier / version — + not a mixed-fleet rollout. Nodes that don't upgrade lose access to rewards, so + the incentive to migrate is built in. +- **Day length is fixed in production.** The tally's notion of "a day" is a fixed + 86400 seconds in any release build; the compressed-time knob used by local + testnets is compiled out (feature-gated) so a reporter can never redefine "a day" + in production eligibility evidence. +- Recommended starting values, to be finalised against real production numbers: + ~7 audited days required, two-week memory window, most-recent pass within ~1 day, + size slack ~2×, majority of judges with a floor of 3, "caught" mark sticks ~7 + days. +- Rolled out in stages: nodes start keeping and sharing the record, then clients + watch and log who they *would* exclude (observe-only shadow eligibility, to + calibrate against churn / NAT / newcomer cases), then clients turn enforcement on + only once that data looks right. + +## Validation + +This decision is correct only if the following hold; each must be shown before the +ADR is Accepted: + +- **Honest nodes qualify and stay eligible.** A node storing real data qualifies in + about a week, an honest node that holds steady data without new uploads does not + quietly fall out of eligibility, and enforcement causes no upload failures. +- **Cheaters are excluded quickly.** A node that takes payment and then deletes is + dropped within about a day of its first payment, whether or not it keeps + receiving fresh data afterward. +- **Fence and conviction behave as specified.** A monetized-pin timeout fences and + clears on the next pass; a confirmed failure convicts and stays sticky for the + dues period even as passes accrue; a restart preserves the tally safely. +- **Collusion is as hard as capturing a neighbourhood.** Withholding vouches or + faking catches can only exclude an honest node, or keep a cheater in, by + controlling more than half its judges: the same bar the network already accepts + for pricing and audits. +- **The audit cadence in production actually supports the ~7-day rule**, measured on + the real fleet under observe-only shadow eligibility before enforcement is turned + on. + +## Notes for AI-assisted work + +AI tools may help draft this ADR, but **must not mark it Accepted without human +review**. Accepted ADRs are immutable: create a new superseding ADR rather than +editing an Accepted one. diff --git a/src/node.rs b/src/node.rs index fcd1cb27..b582f3ee 100644 --- a/src/node.rs +++ b/src/node.rs @@ -170,6 +170,15 @@ impl NodeBuilder { let concrete = Arc::clone(engine.commitment_state()); let source: Arc = concrete; protocol.attach_commitment_source(source); + // ADR-0005: wire the engine's audit tally as the quote + // generator's report source so quote responses carry + // this node's signed audit report. + let report_source: Arc = + Arc::new(crate::replication::audit_tally::TallyReportSource::new( + engine.audit_tally_handle(), + *identity.peer_id().as_bytes(), + )); + protocol.attach_report_source(report_source); // ADR-0004: share the engine's gossip commitment // cache with the verifier so the cross-check can // resolve quote pins against neighbours' commitments. diff --git a/src/payment/quote.rs b/src/payment/quote.rs index 74f18333..08b2fdaa 100644 --- a/src/payment/quote.rs +++ b/src/payment/quote.rs @@ -8,7 +8,7 @@ //! and will be fully integrated when the node is initialized. use crate::error::{Error, Result}; -use crate::logging::debug; +use crate::logging::{debug, warn}; use crate::payment::metrics::QuotingMetricsTracker; use crate::payment::pricing::calculate_price; use evmlib::merkle_payments::MerklePaymentCandidateNode; @@ -66,6 +66,20 @@ pub trait CommitmentSource: Send + Sync { fn commitment_blob_for_pin(&self, pin: [u8; 32]) -> Option>; } +/// Source of the node's audit-tally rows for the ADR-0005 report. +/// +/// Shipped on quote responses. Implemented by the replication engine's tally; +/// decouples [`QuoteGenerator`] from replication internals the same way +/// [`CommitmentSource`] does. +pub trait AuditReportSource: Send + Sync { + /// This node's own peer id (the report's `reporter_peer_id`). + fn reporter_peer_id(&self) -> [u8; 32]; + + /// The tally rows as they stand right now (pruned to the window and capped + /// to the wire limits). Empty when this observer has no testimony yet. + fn report_rows(&self) -> Vec; +} + /// Quote generator for creating payment quotes. /// /// Uses the node's signing capabilities to sign quotes, which clients @@ -93,6 +107,10 @@ pub struct QuoteGenerator { sign_fn: Option, /// Public key bytes for the quote. pub_key: Vec, + /// ADR-0005 report source: the audit tally this node testifies from on + /// quote responses. `None` until [`Self::attach_report_source`] is called + /// (pre-wiring / unit tests), in which case responses carry no report. + report_source: RwLock>>, } impl QuoteGenerator { @@ -112,7 +130,70 @@ impl QuoteGenerator { commitment_source: RwLock::new(None), sign_fn: None, pub_key: Vec::new(), + report_source: RwLock::new(None), + } + } + + /// Attach the ADR-0005 report source so quote responses can ship this + /// node's signed audit report. Idempotent, interior mutability — same + /// contract as [`Self::attach_commitment_source`]. + pub fn attach_report_source(&self, source: Arc) { + *self.report_source.write() = Some(source); + debug!("QuoteGenerator: ADR-0005 audit-report source attached"); + } + + /// Build this node's signed audit report for a quote response (ADR-0005). + /// + /// `nonce` is the client's request nonce, echoed and signed so the report + /// is provably generated for this request. Returns the rmp-serialized + /// [`ant_protocol::payment::AuditReport`], or `None` when no source is + /// attached, the tally is empty (nothing to testify), or the node cannot + /// sign — a response without a report is valid, it just doesn't vouch. + #[must_use] + pub fn build_audit_report(&self, nonce: [u8; 32]) -> Option> { + use ant_protocol::payment::{MAX_AUDIT_REPORT_BYTES, MAX_REPORT_DAYS, MAX_REPORT_ROWS}; + let source = self.report_source.read().as_ref().map(Arc::clone)?; + let sign_fn = self.sign_fn.as_ref()?; + let mut rows = source.report_rows(); + if rows.is_empty() { + return None; + } + // Producer-side cap enforcement (ADR-0005): the source truncates, but the + // producer must NOT rely on that discipline — it enforces the structural + // caps itself so a malformed or oversized report is never signed or put on + // the wire. Verifiers reject an over-cap report wholesale, so emitting one + // would only cost this node its own vouch. + rows.truncate(MAX_REPORT_ROWS); + for row in &mut rows { + row.days.truncate(MAX_REPORT_DAYS); + } + let reporter_peer_id = source.reporter_peer_id(); + let payload = + ant_protocol::payment::audit_report_signed_payload(&reporter_peer_id, &nonce, &rows); + let signature = sign_fn(&payload); + if signature.is_empty() { + return None; + } + let report = ant_protocol::payment::AuditReport { + reporter_peer_id, + nonce, + rows, + signature, + }; + let bytes = rmp_serde::to_vec(&report).ok()?; + // Enforce the serialized byte cap the verifier expects (a report over + // MAX_AUDIT_REPORT_BYTES is dropped whole on receipt): never put an + // over-cap report on the wire. With the row/day caps above this is + // unreachable for an honest tally, but the guard makes the invariant local. + if bytes.len() > MAX_AUDIT_REPORT_BYTES { + warn!( + "ADR-0005: built audit report {} B exceeds cap {} B; omitting", + bytes.len(), + MAX_AUDIT_REPORT_BYTES + ); + return None; } + Some(bytes) } /// Attach the ADR-0004 commitment source so quotes bind their price to the @@ -472,6 +553,96 @@ mod tests { generator } + /// Fixed-rows [`AuditReportSource`] for tests. + struct FixedReportSource { + reporter_peer_id: [u8; 32], + rows: Vec, + } + impl AuditReportSource for FixedReportSource { + fn reporter_peer_id(&self) -> [u8; 32] { + self.reporter_peer_id + } + fn report_rows(&self) -> Vec { + self.rows.clone() + } + } + + /// ADR-0005 report round-trip with REAL keys: the node-side + /// `build_audit_report` output must verify with the client-side + /// `verify_audit_report` against the same pubkey, reporter id, and nonce — + /// this is the guard against signer/verifier payload-framing drift. Also + /// asserts the nonce binding (wrong nonce fails) and the empty-tally and + /// no-source `None` paths. + #[test] + fn test_audit_report_round_trip_real_keys() { + use ant_protocol::payment::{verify_audit_report, AuditReport, AuditReportDay}; + + let ml_dsa = MlDsa65::new(); + let (public_key, secret_key) = ml_dsa.generate_keypair().expect("keypair generation"); + let pub_key_bytes = public_key.as_bytes().to_vec(); + // The reporter id is the peer binding the client already checked for + // the quote: BLAKE3(pub_key). + let reporter_peer_id = *blake3::hash(&pub_key_bytes).as_bytes(); + + let mut generator = QuoteGenerator::new( + RewardsAddress::new([3u8; 20]), + QuotingMetricsTracker::new(10), + ); + let sk_bytes = secret_key.as_bytes().to_vec(); + generator.set_signer(pub_key_bytes.clone(), move |msg| { + let sk = MlDsaSecretKey::from_bytes(&sk_bytes).expect("secret key parse"); + MlDsa65::new() + .sign(&sk, msg) + .expect("signing") + .as_bytes() + .to_vec() + }); + + let nonce = [9u8; 32]; + assert!( + generator.build_audit_report(nonce).is_none(), + "no source attached -> no report" + ); + + generator.attach_report_source(Arc::new(FixedReportSource { + reporter_peer_id, + rows: Vec::new(), + })); + assert!( + generator.build_audit_report(nonce).is_none(), + "empty tally -> no report (nothing to testify)" + ); + + let rows = vec![ant_protocol::payment::AuditReportRow { + subject_peer_id: [5u8; 32], + days: vec![AuditReportDay { + age_days: 0, + passes: 4, + max_passed_key_count: 1234, + }], + fenced: false, + convicted: false, + }]; + generator.attach_report_source(Arc::new(FixedReportSource { + reporter_peer_id, + rows: rows.clone(), + })); + + let bytes = generator + .build_audit_report(nonce) + .expect("report with rows"); + let report: AuditReport = rmp_serde::from_slice(&bytes).expect("report parses"); + assert_eq!(report.rows, rows, "rows survive the wire encoding"); + assert!( + verify_audit_report(&report, &pub_key_bytes, &reporter_peer_id, &nonce), + "node-signed report must verify client-side" + ); + assert!( + !verify_audit_report(&report, &pub_key_bytes, &reporter_peer_id, &[0u8; 32]), + "a different nonce must not verify (no precompute/replay)" + ); + } + /// Fixed-binding [`CommitmentSource`] for tests: always reports the same /// `(key_count, pin)` so we can assert the forced-price wiring exactly. struct FixedCommitmentSource { diff --git a/src/replication/audit.rs b/src/replication/audit.rs index 3bfccb65..7b377c97 100644 --- a/src/replication/audit.rs +++ b/src/replication/audit.rs @@ -37,6 +37,10 @@ pub enum AuditTickResult { challenged_peer: PeerId, /// Number of keys verified. keys_checked: usize, + /// Key count of the audited storage commitment — `Some` only for the + /// subtree (commitment) audit lane, which is the only lane that feeds + /// the ADR-0005 audit tally ("audited clean at this size"). + commitment_key_count: Option, }, /// Audit found failures (after responsibility confirmation). Failed { @@ -577,6 +581,9 @@ async fn verify_digests( return AuditTickResult::Passed { challenged_peer: *challenged_peer, keys_checked: keys.len(), + // Responsible-chunk audits challenge single chunks, not a + // commitment — they never feed the ADR-0005 tally. + commitment_key_count: None, }; } diff --git a/src/replication/audit_tally.rs b/src/replication/audit_tally.rs new file mode 100644 index 00000000..1c0f1b5e --- /dev/null +++ b/src/replication/audit_tally.rs @@ -0,0 +1,540 @@ +//! Per-peer audit tally — the node-side standing record for ADR-0005. +//! +//! Every subtree (commitment) audit outcome this node observes is remembered +//! here as plain counts: passes per day and the largest committed key count +//! that passed, plus two row-local markers — a deterministic conviction (zeroes +//! the row and stays sticky for one dues period) and an unanswered challenge on +//! a pin the peer monetized (fences the row until a fresh pass clears it). +//! Nothing is weighted, decayed, or tuned; the client aggregates a majority of +//! these rows into the eligibility decision. +//! +//! Only the *subtree* audit lane feeds the tally: its passes carry the audited +//! commitment's key count (the "at this size" half of the dues) and its +//! confirmed failures are deterministic convictions. The responsible-chunk +//! audit lane stays out — its timeouts are not deterministic and its passes +//! carry no committed size. +//! +//! Two codecs, do not confuse them: the on-disk SNAPSHOT here uses **postcard** +//! ([`AuditTally::from_bytes`]/[`AuditTally::snapshot_bytes`], mirroring +//! `commitment_retention.bin`, ADR-0004 A1 — write-on-change, atomic temp-file +//! rename); the WIRE report shipped on a quote response is **`MessagePack`** +//! (`rmp_serde`, built in `payment::quote::build_audit_report`). Losing the +//! snapshot only costs this observer's testimony window — the subject re-earns +//! it through ordinary audits. + +use std::collections::HashMap; + +use crate::logging::warn; +use saorsa_core::identity::PeerId; +use serde::{Deserialize, Serialize}; + +use ant_protocol::payment::{AuditReportDay, AuditReportRow, MAX_REPORT_DAYS, MAX_REPORT_ROWS}; + +/// Tally window, in day buckets. Day entries older than this are pruned. +pub const TALLY_WINDOW_DAYS: u64 = 14; + +/// How long a conviction marker stays sticky (ADR-0005 v4): one dues period. +/// +/// A pass does NOT clear it — each observer that catches a peer withholds its +/// vouch for a full re-grind, so partial catches bite under the relative bar. +pub const CONVICTION_STICKY_DAYS: u64 = 7; + +/// Cap on tracked rows. The neighbour-sync observation scope is ~20 peers, so +/// this leaves generous slack; when full, the stalest row is evicted. +pub const MAX_TALLY_ROWS: usize = 64; + +/// Current unix time in whole seconds. +/// +/// 0 only if the system clock is before the epoch (never panics — tally +/// timestamps degrade, nothing else). +#[must_use] +pub fn unix_now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()) +} + +/// Length of one tally day bucket in seconds. +#[must_use] +pub fn tally_day_secs() -> u64 { + 86_400 +} + +/// One day bucket of outcomes for one observed peer. +#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)] +struct DayCell { + /// Subtree audit passes recorded in this bucket. + passes: u32, + /// Largest committed key count that passed in this bucket. + max_passed_key_count: u32, +} + +/// The tally row for one observed peer. +#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)] +struct TallyRow { + /// Day-index → outcomes. Day index = `unix_secs / tally_day_secs()`. + days: Vec<(u64, DayCell)>, + /// Day index of the last deterministic conviction, if any. Sticky for + /// [`CONVICTION_STICKY_DAYS`]; passes do NOT clear it. + convicted_day: Option, + /// Day index the row was fenced by an unanswered monetized-pin challenge, + /// if outstanding. Cleared by the next recorded pass. Also aged out in + /// [`TallyRow::prune`] once the window has fully passed AND no day-evidence + /// remains: an eternal fence with no underlying passes would outlast the + /// evidence it was meant to freeze, so it is bounded to the window. + fenced_day: Option, + /// Day index of the most recent activity, for row eviction. + last_touched_day: u64, +} + +impl TallyRow { + fn prune(&mut self, today: u64) { + self.days + .retain(|(day, _)| today.saturating_sub(*day) < TALLY_WINDOW_DAYS); + if let Some(day) = self.convicted_day { + if today.saturating_sub(day) >= CONVICTION_STICKY_DAYS { + self.convicted_day = None; + } + } + // Age out a fence that has outlasted the window with no day-evidence + // left underneath it. The fence normally clears on the peer's next + // pass; but a peer that goes silent forever would otherwise keep this + // observer's row fenced indefinitely — longer than the evidence it + // froze. Once the window has fully passed AND no pass days remain, + // there is nothing left to freeze, so drop the fence (the row then + // ages out via `is_empty`). + if let Some(day) = self.fenced_day { + if self.days.is_empty() && today.saturating_sub(day) >= TALLY_WINDOW_DAYS { + self.fenced_day = None; + } + } + } + + fn is_empty(&self) -> bool { + self.days.is_empty() && self.convicted_day.is_none() && self.fenced_day.is_none() + } +} + +/// Snapshot format tag: magic + version, prepended to the postcard body. +/// +/// Postcard is non-self-describing and decodes positionally, so a struct-layout +/// change (e.g. the v5 `fenced`/`convicted` `bool` → `Option` retype) makes +/// an older on-disk blob mis-decode into garbage rather than fail cleanly. The +/// tag turns that into a deliberate, safe rejection: [`AuditTally::from_bytes`] +/// discards any snapshot whose tag it does not recognise and the observer starts +/// fresh — cheap, since a lost snapshot only costs this observer's testimony +/// window, which it re-earns through ordinary audits. Bump [`SNAPSHOT_VERSION`] +/// on any change to the serialized layout of `AuditTally` or its fields. +const SNAPSHOT_MAGIC: &[u8; 4] = b"ATLY"; +const SNAPSHOT_VERSION: u8 = 1; + +/// The audit tally: per-peer rows of subtree-audit outcomes. +/// +/// Interior state only — callers wrap it in their own lock and drive +/// persistence with [`AuditTally::snapshot_bytes`]. +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct AuditTally { + rows: HashMap, +} + +impl AuditTally { + /// Restore from persisted snapshot bytes; `None` if the tag is unrecognised + /// (wrong magic or an older/newer layout version) or the body is corrupt. + /// An unrecognised snapshot is discarded rather than mis-decoded, so the + /// observer safely starts fresh across a format change. + #[must_use] + pub fn from_bytes(bytes: &[u8]) -> Option { + let body = bytes + .strip_prefix(SNAPSHOT_MAGIC.as_slice()) + .and_then(|rest| rest.split_first()) + .filter(|(version, _)| **version == SNAPSHOT_VERSION) + .map(|(_, body)| body)?; + postcard::from_bytes(body).ok() + } + + /// Snapshot for persistence: the format tag followed by the postcard body. + /// `None` only on serialization failure. + #[must_use] + pub fn snapshot_bytes(&self) -> Option> { + let body = postcard::to_allocvec(self).ok()?; + let mut out = Vec::with_capacity(SNAPSHOT_MAGIC.len() + 1 + body.len()); + out.extend_from_slice(SNAPSHOT_MAGIC.as_slice()); + out.push(SNAPSHOT_VERSION); + out.extend_from_slice(&body); + Some(out) + } + + /// Record a subtree audit pass for `peer` on a commitment of + /// `commitment_key_count` keys. Clears any fence. Does NOT clear an + /// outstanding conviction (ADR-0005 v4): the marker stays sticky for + /// [`CONVICTION_STICKY_DAYS`], so a catch costs this observer's vouch for + /// a full dues period while the zeroed history re-accumulates underneath. + /// + /// # Precondition (event ordering) + /// + /// `now_secs` MUST be the observer's real-time clock at the moment the audit + /// outcome is processed — never a timestamp carried on the wire. All three + /// record paths pass [`unix_now_secs`], so events are stamped in processing + /// order and `now_secs` is monotonic non-decreasing. This is what makes + /// "a pass clears the fence" safe: a pass can only clear a fence that was set + /// at or before the same wall-clock instant, so an out-of-order/replayed + /// pass cannot retroactively lift a newer fence. Do not feed this a + /// caller-supplied or attacker-influenced timestamp. + pub fn record_pass(&mut self, peer: PeerId, commitment_key_count: u32, now_secs: u64) { + let today = now_secs / tally_day_secs(); + self.evict_if_full(&peer, today); + let row = self.rows.entry(peer).or_default(); + row.prune(today); + row.fenced_day = None; + row.last_touched_day = today; + match row.days.iter_mut().find(|(day, _)| *day == today) { + Some((_, cell)) => { + cell.passes = cell.passes.saturating_add(1); + cell.max_passed_key_count = cell.max_passed_key_count.max(commitment_key_count); + } + None => { + row.days.push(( + today, + DayCell { + passes: 1, + max_passed_key_count: commitment_key_count, + }, + )); + } + } + } + + /// Record a deterministic conviction: the row is zeroed — the peer starts + /// its dues over in this observer's eyes — and the marker stays sticky for + /// [`CONVICTION_STICKY_DAYS`] regardless of new passes. Stickiness equals + /// the dues period, so the total cost of a catch is one full re-grind at + /// this observer; it can never compound into a permanent ban. + /// + /// Clearing `fenced_day` here is intentional, not a weakening: a conviction + /// strictly subsumes a fence. A convicted row does not vouch at all (the + /// client checks `convicted` independently), and the conviction's stickiness + /// covers the full dues period — longer than any fence would have lasted. On + /// expiry the peer must re-earn its entire week regardless. Keeping a + /// now-redundant fence marker alongside the conviction would buy no extra + /// suppression and only complicate the age-out invariant. + pub fn record_conviction(&mut self, peer: PeerId, now_secs: u64) { + let today = now_secs / tally_day_secs(); + self.evict_if_full(&peer, today); + let row = self.rows.entry(peer).or_default(); + row.days.clear(); + row.fenced_day = None; + row.convicted_day = Some(today); + row.last_touched_day = today; + } + + /// Record an unanswered audit challenge on a pin the peer monetized: the + /// row is fenced (its testimony stops counting) until the next pass. + pub fn record_unanswered_monetized_challenge(&mut self, peer: PeerId, now_secs: u64) { + let today = now_secs / tally_day_secs(); + self.evict_if_full(&peer, today); + let row = self.rows.entry(peer).or_default(); + row.fenced_day = Some(today); + row.last_touched_day = today; + } + + /// Build wire report rows as of `now_secs`: prune the window, convert day + /// indexes to relative ages, and cap rows/days to the wire limits (rows + /// with the most recent activity win the cap). + #[must_use] + pub fn report_rows(&mut self, now_secs: u64) -> Vec { + let today = now_secs / tally_day_secs(); + self.rows.retain(|_, row| { + row.prune(today); + !row.is_empty() + }); + let mut rows: Vec<(&PeerId, &TallyRow)> = self.rows.iter().collect(); + rows.sort_by_key(|(_, row)| std::cmp::Reverse(row.last_touched_day)); + rows.truncate(MAX_REPORT_ROWS); + rows.into_iter() + .map(|(peer, row)| { + let mut days: Vec = row + .days + .iter() + .map(|(day, cell)| AuditReportDay { + age_days: u16::try_from(today.saturating_sub(*day)).unwrap_or(u16::MAX), + passes: cell.passes, + max_passed_key_count: cell.max_passed_key_count, + }) + .collect(); + days.sort_by_key(|d| d.age_days); + days.truncate(MAX_REPORT_DAYS); + AuditReportRow { + subject_peer_id: *peer.as_bytes(), + days, + fenced: row.fenced_day.is_some(), + convicted: row.convicted_day.is_some(), + } + }) + .collect() + } + + /// Number of tracked rows (diagnostics/tests). + #[must_use] + pub fn row_count(&self) -> usize { + self.rows.len() + } + + /// Evict the stalest row when at capacity and `peer` is not yet tracked. + fn evict_if_full(&mut self, peer: &PeerId, today: u64) { + if self.rows.len() < MAX_TALLY_ROWS || self.rows.contains_key(peer) { + return; + } + // Prefer dropping rows that fell out of the window entirely. + self.rows.retain(|_, row| { + row.prune(today); + !row.is_empty() + }); + if self.rows.len() < MAX_TALLY_ROWS { + return; + } + if let Some(stalest) = self + .rows + .iter() + .min_by_key(|(_, row)| row.last_touched_day) + .map(|(p, _)| *p) + { + self.rows.remove(&stalest); + } + } +} + +/// Adapter exposing an [`AuditTally`] as the quote generator's +/// [`crate::payment::quote::AuditReportSource`] (ADR-0005 assembly glue). +pub struct TallyReportSource { + tally: std::sync::Arc>, + reporter_peer_id: [u8; 32], +} + +impl TallyReportSource { + /// Bundle the engine's tally handle with this node's own peer id. + #[must_use] + pub fn new( + tally: std::sync::Arc>, + reporter_peer_id: [u8; 32], + ) -> Self { + Self { + tally, + reporter_peer_id, + } + } +} + +impl crate::payment::quote::AuditReportSource for TallyReportSource { + fn reporter_peer_id(&self) -> [u8; 32] { + self.reporter_peer_id + } + + fn report_rows(&self) -> Vec { + self.tally.write().map_or_else( + |_| { + // A poisoned lock means a writer panicked mid-update. Say so + // loudly: "no testimony" must be distinguishable from + // "tally disabled by poison". + warn!("Audit tally lock poisoned — reporting no rows until restart"); + Vec::new() + }, + |mut tally| tally.report_rows(unix_now_secs()), + ) + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] +mod tests { + use super::*; + + const DAY: u64 = 86_400; + + fn peer(byte: u8) -> PeerId { + PeerId::from_bytes([byte; 32]) + } + + #[test] + fn passes_accumulate_per_day_with_max_size() { + let mut t = AuditTally::default(); + t.record_pass(peer(1), 100, 10 * DAY + 5); + t.record_pass(peer(1), 900, 10 * DAY + 600); + t.record_pass(peer(1), 300, 11 * DAY); + let rows = t.report_rows(11 * DAY + 1); + assert_eq!(rows.len(), 1); + let row = &rows[0]; + assert_eq!(row.subject_peer_id, [1u8; 32]); + assert_eq!(row.days.len(), 2); + // Newest first: today (age 0) then yesterday (age 1). + let [d0, d1] = &row.days[..] else { + panic!("expected two days, got {}", row.days.len()); + }; + assert_eq!(d0.age_days, 0); + assert_eq!(d0.passes, 1); + assert_eq!(d0.max_passed_key_count, 300); + assert_eq!(d1.age_days, 1); + assert_eq!(d1.passes, 2); + assert_eq!(d1.max_passed_key_count, 900); + assert!(!row.fenced); + assert!(!row.convicted); + } + + #[test] + fn conviction_zeroes_history_and_stays_sticky_for_dues_period() { + let mut t = AuditTally::default(); + for d in 0..7 { + t.record_pass(peer(2), 500, d * DAY + 1); + } + t.record_conviction(peer(2), 7 * DAY); + let rows = t.report_rows(7 * DAY + 1); + assert_eq!(rows.len(), 1); + assert!(rows[0].days.is_empty(), "conviction must zero the history"); + assert!(rows[0].convicted, "outstanding conviction is reported"); + // v4: passes do NOT clear the marker — it stays for a full dues + // period while the fresh history re-accumulates underneath. + for d in 8..14 { + t.record_pass(peer(2), 500, d * DAY); + } + let rows = t.report_rows(13 * DAY + 1); + assert!(!rows[0].days.is_empty(), "history re-accumulates"); + assert!( + rows[0].convicted, + "the marker survives passes inside the sticky period" + ); + // After CONVICTION_STICKY_DAYS the marker expires; the re-earned + // history stands on its own. + t.record_pass(peer(2), 500, (7 + CONVICTION_STICKY_DAYS) * DAY + 1); + let rows = t.report_rows((7 + CONVICTION_STICKY_DAYS) * DAY + 2); + assert!( + !rows[0].convicted, + "the marker expires after one dues period" + ); + assert!( + !rows[0].days.is_empty(), + "re-earned days survive the expiry" + ); + } + + #[test] + fn conviction_row_without_passes_expires_with_the_sticky_period() { + let mut t = AuditTally::default(); + t.record_conviction(peer(3), 0); + let rows = t.report_rows(1); + assert_eq!(rows.len(), 1); + assert!(rows[0].convicted); + // Never re-proves anything: the marker (and the empty row) age out + // after the sticky period. + let rows = t.report_rows(CONVICTION_STICKY_DAYS * DAY + 1); + assert!(rows.is_empty(), "an abandoned convicted row expires"); + } + + #[test] + fn fence_blocks_until_next_pass() { + let mut t = AuditTally::default(); + t.record_pass(peer(4), 500, DAY); + t.record_unanswered_monetized_challenge(peer(4), DAY + 10); + let rows = t.report_rows(DAY + 20); + let row = rows.first().expect("one row"); + assert!(row.fenced); + assert_eq!(row.days.len(), 1, "fence keeps history, blocks it"); + t.record_pass(peer(4), 500, DAY + 30); + let rows = t.report_rows(DAY + 40); + assert!( + !rows.first().expect("one row").fenced, + "a fresh pass clears the fence" + ); + } + + #[test] + fn fence_ages_out_after_window_with_no_evidence() { + // A peer that goes silent forever after a monetized-pin timeout: no + // fresh pass ever clears the fence, but once the window passes and no + // day-evidence remains, the fence (and the row) age out — an eternal + // fence would outlast the evidence it froze. + let mut t = AuditTally::default(); + t.record_pass(peer(5), 500, DAY); + t.record_unanswered_monetized_challenge(peer(5), 2 * DAY); + // Still fenced within the window (day-evidence at day 1 still present). + assert!( + t.report_rows(3 * DAY).first().expect("row").fenced, + "fence holds while evidence is in-window" + ); + // Well past the window: day-1 pass pruned, fence has nothing to freeze. + let rows = t.report_rows((TALLY_WINDOW_DAYS + 3) * DAY); + assert!( + rows.is_empty(), + "an eternal fence with no evidence ages out with the row" + ); + } + + #[test] + fn old_days_prune_out_of_the_window() { + let mut t = AuditTally::default(); + t.record_pass(peer(5), 500, 0); + t.record_pass(peer(5), 500, (TALLY_WINDOW_DAYS - 1) * DAY); + let rows = t.report_rows(TALLY_WINDOW_DAYS * DAY); + assert_eq!(rows.len(), 1); + assert_eq!( + rows[0].days.len(), + 1, + "the day-0 bucket must have pruned out" + ); + } + + #[test] + fn snapshot_roundtrip_preserves_rows() { + let mut t = AuditTally::default(); + t.record_pass(peer(6), 123, 5 * DAY); + t.record_unanswered_monetized_challenge(peer(7), 5 * DAY); + let bytes = t.snapshot_bytes().unwrap(); + let mut restored = AuditTally::from_bytes(&bytes).unwrap(); + assert_eq!(restored.row_count(), 2); + let rows = restored.report_rows(5 * DAY + 1); + assert_eq!(rows.len(), 2); + } + + #[test] + fn corrupt_snapshot_returns_none() { + assert!(AuditTally::from_bytes(&[0xFF, 0x00, 0x13, 0x37]).is_none()); + } + + #[test] + fn snapshot_tag_rejects_untagged_and_wrong_version() { + let mut t = AuditTally::default(); + t.record_pass(peer(6), 123, 5 * DAY); + let tagged = t.snapshot_bytes().unwrap(); + + // A raw postcard body with NO tag (what an older format wrote) must be + // rejected — mis-decoding it into garbage is exactly what the tag + // prevents. + let untagged = postcard::to_allocvec(&t).unwrap(); + assert!(AuditTally::from_bytes(&untagged).is_none()); + + // Right magic, wrong version → rejected. + let mut wrong_version = tagged.clone(); + wrong_version[SNAPSHOT_MAGIC.len()] = SNAPSHOT_VERSION.wrapping_add(1); + assert!(AuditTally::from_bytes(&wrong_version).is_none()); + + // Wrong magic → rejected. + let mut wrong_magic = tagged.clone(); + wrong_magic[0] ^= 0xFF; + assert!(AuditTally::from_bytes(&wrong_magic).is_none()); + + // The correctly tagged snapshot still round-trips. + assert!(AuditTally::from_bytes(&tagged).is_some()); + } + + #[test] + fn row_cap_evicts_when_full() { + let mut t = AuditTally::default(); + let base = 100 * DAY; + for i in 0..MAX_TALLY_ROWS { + let mut id = [u8::try_from(i % 250).unwrap(); 32]; + id[0] = u8::try_from(i / 250).unwrap(); + t.record_pass(PeerId::from_bytes(id), 1, base + u64::try_from(i).unwrap()); + } + assert_eq!(t.row_count(), MAX_TALLY_ROWS); + t.record_pass(peer(255), 1, base + 999); + assert_eq!(t.row_count(), MAX_TALLY_ROWS, "cap holds under insert"); + } +} diff --git a/src/replication/config.rs b/src/replication/config.rs index f1544664..e8acd44d 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -382,6 +382,18 @@ pub const AUDIT_ON_GOSSIP_PROBABILITY: f64 = 0.2; /// seconds. Bounds how often any one peer is audited regardless of gossip rate. pub const AUDIT_ON_GOSSIP_COOLDOWN_SECS: u64 = 30 * 60; +/// Per-peer gossip-audit cooldown, in seconds. +#[must_use] +pub fn audit_on_gossip_cooldown_secs() -> u64 { + AUDIT_ON_GOSSIP_COOLDOWN_SECS +} + +/// Gossip-audit lottery probability. +#[must_use] +pub fn audit_on_gossip_probability() -> f64 { + AUDIT_ON_GOSSIP_PROBABILITY +} + /// ADR-0004: first-audit drainer retry cadence for cooldown-pending pins. /// /// How often the drainer retries pins it kept pending because their peer was on diff --git a/src/replication/mod.rs b/src/replication/mod.rs index bd827210..ada4ecbb 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -16,6 +16,7 @@ pub mod admission; pub mod audit; +pub mod audit_tally; pub mod bootstrap; pub mod commitment; pub mod commitment_state; @@ -159,6 +160,11 @@ const BOOTSTRAP_DRAIN_CHECK_SECS: u64 = 5; /// "rotated past" case. const COMMITMENT_ROTATION_INTERVAL_SECS: u64 = 3600; +/// Effective rotation interval. +fn commitment_rotation_interval() -> std::time::Duration { + std::time::Duration::from_secs(COMMITMENT_ROTATION_INTERVAL_SECS) +} + /// How often the responder retention snapshot is flushed to disk (ADR-0004 A1). /// Short relative to the answerability TTL (3 h) so a gossip-stamp refresh is /// durable well before it could matter to a restart, while the write-on-change @@ -207,6 +213,11 @@ fn quote_within_audit_window(quote_ts: SystemTime, now: SystemTime) -> bool { /// (the 10-20 min neighbor-sync round on each peer). const COMMITMENT_SIG_VERIFY_MIN_INTERVAL: Duration = Duration::from_secs(60); +/// The effective per-peer sig-verify interval. +fn commitment_sig_verify_min_interval() -> Duration { + COMMITMENT_SIG_VERIFY_MIN_INTERVAL +} + /// Hard cap on the size of `last_commitment_by_peer`. /// /// Bounds the per-process memory cost of the auditor's per-peer @@ -361,6 +372,17 @@ pub struct ReplicationEngine { /// ADR-0004: receiver half of the monetized-pin channel, taken by /// `start_first_audit_drainer`. monetized_pin_rx: Option>, + /// ADR-0005: per-peer tally of subtree-audit outcomes — the raw facts a + /// quote response's signed audit report testifies with. A std (not tokio) + /// lock because the quote handler reads it synchronously; every critical + /// section is short and never held across an await. + audit_tally: Arc>, + /// Wakes the persistence loop for negative audit-tally evidence that must + /// not wait for the periodic retention flush cadence. + audit_tally_persist_trigger: Arc, + /// Path to the persisted tally snapshot (`{root_dir}/audit_tally.bin`), + /// written on change alongside the retention snapshot. + audit_tally_path: PathBuf, /// Shutdown token. shutdown: CancellationToken, /// Background task handles. @@ -431,6 +453,9 @@ impl ReplicationEngine { possession_check_rx: Some(possession_check_rx), monetized_pin_tx, monetized_pin_rx: Some(monetized_pin_rx), + audit_tally: Arc::new(std::sync::RwLock::new(audit_tally::AuditTally::default())), + audit_tally_persist_trigger: Arc::new(Notify::new()), + audit_tally_path: root_dir.join("audit_tally.bin"), shutdown, task_handles: Vec::new(), }; @@ -439,6 +464,9 @@ impl ReplicationEngine { // pins from the first audit it serves, and the persist loop never races // an empty snapshot over the good on-disk file. load_commitment_retention(&engine.commitment_state, &engine.retention_path).await; + // ADR-0005: reload the audit tally the same way — this observer's + // testimony about its neighbours survives a restart. + load_audit_tally(&engine.audit_tally, &engine.audit_tally_path).await; Ok(engine) } @@ -450,6 +478,13 @@ impl ReplicationEngine { self.monetized_pin_tx.clone() } + /// ADR-0005: shared handle to the audit tally, for wiring the quote + /// generator's report source at node assembly. + #[must_use] + pub fn audit_tally_handle(&self) -> Arc> { + Arc::clone(&self.audit_tally) + } + /// Get a reference to the `PaidList`. #[must_use] pub fn paid_list(&self) -> &Arc { @@ -814,6 +849,8 @@ impl ReplicationEngine { recent_provers: Arc::clone(&self.recent_provers), sync_state: Arc::clone(&self.sync_state), cooldown: Arc::clone(&self.audit_on_gossip_cooldown), + audit_tally: Arc::clone(&self.audit_tally), + audit_tally_persist_trigger: Arc::clone(&self.audit_tally_persist_trigger), }; let shutdown = self.shutdown.clone(); @@ -825,7 +862,7 @@ impl ReplicationEngine { 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 + // MEMORY-ONLY 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 @@ -835,6 +872,10 @@ impl ReplicationEngine { // 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. + // TODO(ADR-0005 durability): Persist this queue, plus enough + // first-audit launch/dedup state to avoid re-auditing completed pins, + // so a restart cannot forget a paid pin before its deterministic + // first audit. let mut pending: LruCache = LruCache::new( NonZeroUsize::new(MAX_LAST_COMMITMENT_BY_PEER).unwrap_or(NonZeroUsize::MIN), ); @@ -949,8 +990,38 @@ impl ReplicationEngine { &trigger.sync_state, &trigger.recent_provers, &trigger.config, + &trigger.audit_tally, + &trigger.audit_tally_persist_trigger, ) .await; + // ADR-0005: this is the MONETIZED first-audit lane. A + // timeout here is an unanswered challenge on a pin the + // peer was just paid for — fence its tally row (this + // observer's testimony stops counting until the peer + // passes a fresh audit). Row-local by design: silence + // is not portable evidence, so it silences one voucher + // and nothing more. + if let AuditTickResult::Failed { + evidence: + FailureEvidence::AuditFailure { + challenged_peer, + reason: AuditFailureReason::Timeout, + .. + }, + } = &result + { + let recorded_fence = + trigger.audit_tally.write().is_ok_and(|mut tally| { + tally.record_unanswered_monetized_challenge( + *challenged_peer, + audit_tally::unix_now_secs(), + ); + true + }); + if recorded_fence { + trigger.audit_tally_persist_trigger.notify_one(); + } + } }); } } @@ -982,6 +1053,8 @@ impl ReplicationEngine { let recent_provers = Arc::clone(&self.recent_provers); let sig_verify_attempts = Arc::clone(&self.sig_verify_attempts); let audit_on_gossip_cooldown = Arc::clone(&self.audit_on_gossip_cooldown); + let audit_tally = Arc::clone(&self.audit_tally); + let audit_tally_persist_trigger = Arc::clone(&self.audit_tally_persist_trigger); let sync_state = Arc::clone(&self.sync_state); let audit_responder_semaphore = Arc::clone(&self.audit_responder_semaphore); let audit_responder_inflight = Arc::clone(&self.audit_responder_inflight); @@ -994,6 +1067,8 @@ impl ReplicationEngine { recent_provers: Arc::clone(&recent_provers), sync_state: Arc::clone(&sync_state), cooldown: Arc::clone(&audit_on_gossip_cooldown), + audit_tally: Arc::clone(&audit_tally), + audit_tally_persist_trigger: Arc::clone(&audit_tally_persist_trigger), }; let handle = tokio::spawn(async move { @@ -1161,6 +1236,8 @@ impl ReplicationEngine { recent_provers: Arc::clone(&self.recent_provers), sync_state: Arc::clone(&sync_state), cooldown: Arc::clone(&self.audit_on_gossip_cooldown), + audit_tally: Arc::clone(&self.audit_tally), + audit_tally_persist_trigger: Arc::clone(&self.audit_tally_persist_trigger), }; let handle = tokio::spawn(async move { @@ -1364,9 +1441,7 @@ impl ReplicationEngine { loop { tokio::select! { () = shutdown.cancelled() => break, - () = tokio::time::sleep( - std::time::Duration::from_secs(COMMITMENT_ROTATION_INTERVAL_SECS) - ) => { + () = tokio::time::sleep(commitment_rotation_interval()) => { if let Err(e) = rebuild_and_rotate_commitment( &storage, &identity, @@ -1395,22 +1470,32 @@ impl ReplicationEngine { self.task_handles.push(handle); } - /// ADR-0004 A1: periodically flush the responder retention snapshot to disk - /// (write-on-change) so answerability — including gossip-stamp refreshes and - /// rotations — survives a restart. Flushes once immediately, then every - /// `RETENTION_PERSIST_INTERVAL_SECS`, and once more on shutdown. + /// ADR-0004 A1 / ADR-0005: flush responder retention and the observer audit + /// tally to disk with write-on-change snapshots. Both flush once + /// immediately, then every `RETENTION_PERSIST_INTERVAL_SECS`, and once more + /// on shutdown; negative audit-tally evidence also wakes an immediate tally + /// flush. fn start_retention_persist_loop(&mut self) { let commitment_state = Arc::clone(&self.commitment_state); let retention_path = self.retention_path.clone(); + let audit_tally = Arc::clone(&self.audit_tally); + let audit_tally_persist_trigger = Arc::clone(&self.audit_tally_persist_trigger); + let audit_tally_path = self.audit_tally_path.clone(); let shutdown = self.shutdown.clone(); let handle = tokio::spawn(async move { let mut last: Option> = None; + let mut last_tally: Option> = None; persist_retention_if_changed(&commitment_state, &retention_path, &mut last).await; + persist_audit_tally_if_changed(&audit_tally, &audit_tally_path, &mut last_tally).await; loop { tokio::select! { () = shutdown.cancelled() => { persist_retention_if_changed(&commitment_state, &retention_path, &mut last) .await; + persist_audit_tally_if_changed( + &audit_tally, &audit_tally_path, &mut last_tally, + ) + .await; break; } () = tokio::time::sleep(std::time::Duration::from_secs( @@ -1418,10 +1503,21 @@ impl ReplicationEngine { )) => { persist_retention_if_changed(&commitment_state, &retention_path, &mut last) .await; + persist_audit_tally_if_changed( + &audit_tally, &audit_tally_path, &mut last_tally, + ) + .await; + } + () = audit_tally_persist_trigger.notified() => { + persist_audit_tally_if_changed( + &audit_tally, &audit_tally_path, &mut last_tally, + ) + .await; } } } debug!("Commitment retention persist loop shut down"); + debug!("Audit tally persist loop shut down"); }); self.task_handles.push(handle); } @@ -4081,6 +4177,7 @@ fn first_failed_key_label(confirmed_failed_keys: &[XorName]) -> String { /// legitimately time out on slow or loaded honest peers, so it never touches the /// responsible-chunk audit path or its timeout accounting. Confirmed subtree /// failures still penalise immediately and revoke holder credit. +#[allow(clippy::too_many_arguments)] async fn handle_subtree_failed_audit( challenged_peer: &PeerId, confirmed_failed_key_count: usize, @@ -4088,6 +4185,8 @@ async fn handle_subtree_failed_audit( p2p_node: &Arc, sync_state: &Arc>, recent_provers: &Arc>, + audit_tally: &std::sync::RwLock, + audit_tally_persist_trigger: &Notify, ) { if matches!(reason, AuditFailureReason::Timeout) { debug!( @@ -4097,6 +4196,16 @@ async fn handle_subtree_failed_audit( return; } + // ADR-0005: a confirmed subtree failure is a deterministic conviction — + // the peer's tally row is zeroed and its dues start over. + let recorded_conviction = audit_tally.write().is_ok_and(|mut tally| { + tally.record_conviction(*challenged_peer, audit_tally::unix_now_secs()); + true + }); + if recorded_conviction { + audit_tally_persist_trigger.notify_one(); + } + // The caller already logged the rich failure line with reason + per-category // summary; avoid a redundant second error log here. let _ = confirmed_failed_key_count; @@ -4123,13 +4232,23 @@ async fn handle_subtree_audit_result( sync_state: &Arc>, recent_provers: &Arc>, config: &ReplicationConfig, + audit_tally: &std::sync::RwLock, + audit_tally_persist_trigger: &Notify, ) { match result { AuditTickResult::Passed { challenged_peer, keys_checked, + commitment_key_count, } => { debug!("Audit passed for {challenged_peer} ({keys_checked} keys)"); + // ADR-0005: a subtree pass is a tally fact — audited clean at the + // committed size, today. Also clears any fence on the row. + if let Some(key_count) = commitment_key_count { + if let Ok(mut tally) = audit_tally.write() { + tally.record_pass(*challenged_peer, *key_count, audit_tally::unix_now_secs()); + } + } // Peer responded normally — clear the active bootstrap claim while // retaining history so a later claim is treated as repeated abuse. { @@ -4172,6 +4291,8 @@ async fn handle_subtree_audit_result( p2p_node, sync_state, recent_provers, + audit_tally, + audit_tally_persist_trigger, ) .await; } @@ -4249,6 +4370,9 @@ async fn handle_audit_result( AuditTickResult::Passed { challenged_peer, keys_checked, + // Responsible-chunk audits never feed the ADR-0005 tally (no + // committed size, non-deterministic timeout lane). + commitment_key_count: _, } => { debug!("Audit passed for {challenged_peer} ({keys_checked} keys)"); { @@ -4379,6 +4503,10 @@ struct GossipAuditTrigger { recent_provers: Arc>, sync_state: Arc>, cooldown: Arc>>, + /// ADR-0005: subtree-audit outcomes land here as tally facts. + audit_tally: Arc>, + /// Wakes the audit-tally persist loop after negative evidence. + audit_tally_persist_trigger: Arc, } /// What a gossip ingest yields for the audit trigger: the commitment hash to @@ -4423,7 +4551,7 @@ pub struct MonetizedPinEvent { /// without a live node: a burst of gossips from one peer yields at most one /// `true` per cooldown window. fn cooldown_allows_audit(map: &mut HashMap, peer: &PeerId, now: Instant) -> bool { - let cooldown = Duration::from_secs(config::AUDIT_ON_GOSSIP_COOLDOWN_SECS); + let cooldown = Duration::from_secs(config::audit_on_gossip_cooldown_secs()); let known = match map.get(peer) { Some(&last) => { if now.saturating_duration_since(last) < cooldown { @@ -4483,7 +4611,7 @@ async fn maybe_trigger_gossip_audit( // `audit_launch_decision` so the ordering is shared with its test. Sample // the lottery here, then let the helper apply it AFTER the cooldown stamp. let now = Instant::now(); - let lottery_wins = rand::thread_rng().gen_bool(config::AUDIT_ON_GOSSIP_PROBABILITY); + let lottery_wins = rand::thread_rng().gen_bool(config::audit_on_gossip_probability()); { let mut map = trigger.cooldown.write().await; if !audit_launch_decision(&mut map, peer, now, lottery_wins) { @@ -4512,6 +4640,8 @@ async fn maybe_trigger_gossip_audit( &trigger.sync_state, &trigger.recent_provers, &trigger.config, + &trigger.audit_tally, + &trigger.audit_tally_persist_trigger, ) .await; }); @@ -4532,7 +4662,7 @@ async fn sig_verify_rate_limit_ok( ) -> bool { let mut attempts = sig_verify_attempts.write().await; if let Some(&last) = attempts.get(source) { - if now.saturating_duration_since(last) < COMMITMENT_SIG_VERIFY_MIN_INTERVAL { + if now.saturating_duration_since(last) < commitment_sig_verify_min_interval() { return false; } } @@ -4831,6 +4961,77 @@ async fn load_commitment_retention(state: &ResponderCommitmentState, path: &Path } } +/// Reload the persisted ADR-0005 audit tally, mirroring +/// `load_commitment_retention`: missing file is a fresh start, corrupt file is +/// logged and ignored (this observer's testimony re-accumulates naturally). +async fn load_audit_tally(tally: &std::sync::RwLock, path: &Path) { + let bytes = match tokio::fs::read(path).await { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + debug!( + "Audit tally: no snapshot at {} (fresh start)", + path.display() + ); + return; + } + Err(e) => { + warn!("Audit tally: failed to read {}: {e}", path.display()); + return; + } + }; + match audit_tally::AuditTally::from_bytes(&bytes) { + Some(restored) => { + let rows = restored.row_count(); + match tally.write() { + Ok(mut guard) => { + *guard = restored; + info!( + "Audit tally: reloaded {rows} row(s) from {}", + path.display() + ); + } + Err(_) => { + warn!("Audit tally: lock poisoned during reload; keeping empty tally"); + } + } + } + None => { + warn!( + "Audit tally: corrupt snapshot at {}; starting empty", + path.display() + ); + } + } +} + +/// Persist the ADR-0005 audit tally IF it changed since `last` — same +/// write-on-change + atomic-rename discipline as the retention snapshot. +async fn persist_audit_tally_if_changed( + tally: &std::sync::RwLock, + path: &Path, + last: &mut Option>, +) { + // Scope the guard so the !Send read guard is provably dead before the + // awaits below (tokio::spawn requires a Send future). + let bytes = { + let Ok(guard) = tally.read() else { + warn!("Audit tally: lock poisoned; skipping persistence this round"); + return; + }; + guard.snapshot_bytes() + }; + let Some(bytes) = bytes else { + warn!("Audit tally: serialization failed; keeping previous snapshot"); + return; + }; + if last.as_deref() == Some(bytes.as_slice()) { + return; + } + if write_retention_atomic(path, bytes.clone(), "Audit tally").await { + *last = Some(bytes); + } +} + /// Persist the responder retention snapshot IF it changed since `last` (ADR-0004 /// A1). Write-on-change keeps frequent gossip-stamp refreshes durable without /// needless disk writes on idle nodes. On success updates `last` to the bytes @@ -4848,16 +5049,16 @@ async fn persist_retention_if_changed( if last.as_deref() == Some(bytes.as_slice()) { return; } - if write_retention_atomic(path, bytes.clone()).await { + if write_retention_atomic(path, bytes.clone(), "Commitment retention").await { *last = Some(bytes); } } /// Durably write `bytes` to `path`: temp file → fsync temp → atomic rename → /// fsync parent dir (so the rename itself survives a crash). Returns `true` on -/// success. Only the retention-persist loop writes this path, so a fixed temp -/// name is safe. -async fn write_retention_atomic(path: &Path, bytes: Vec) -> bool { +/// success. Only the persistence loop writes each target path, so a fixed temp +/// name per target is safe. +async fn write_retention_atomic(path: &Path, bytes: Vec, label: &'static str) -> bool { let path = path.to_path_buf(); let res = tokio::task::spawn_blocking(move || -> std::io::Result<()> { let tmp = path.with_extension("tmp"); @@ -4878,11 +5079,11 @@ async fn write_retention_atomic(path: &Path, bytes: Vec) -> bool { match res { Ok(Ok(())) => true, Ok(Err(e)) => { - warn!("Commitment retention: persist failed: {e}"); + warn!("{label}: persist failed: {e}"); false } Err(e) => { - warn!("Commitment retention: persist task join failed: {e}"); + warn!("{label}: persist task join failed: {e}"); false } } diff --git a/src/replication/storage_commitment_audit.rs b/src/replication/storage_commitment_audit.rs index 14e5104f..febf286e 100644 --- a/src/replication/storage_commitment_audit.rs +++ b/src/replication/storage_commitment_audit.rs @@ -748,28 +748,40 @@ async fn verify_subtree_response( warn!("Audit: {challenged_peer} failed subtree audit ({reason:?})"); failed(challenged_peer, challenge_id, reason) } - AuditVerdict::Pass { checked } => { - // Closeness (ADR-0002, soft/observe-only) — see observe_closeness. - observe_closeness(ctx.p2p_node, ctx.config, challenged_peer, proof).await; - // Credit the peer as a proven holder of its committed keys. - if let (Some(credit), Some(pin)) = (ctx.credit, commitment_hash(commitment)) { - let now = std::time::Instant::now(); - let mut provers = credit.recent_provers.write().await; - for leaf in &proof.leaves { - provers.record_proof(leaf.key, *challenged_peer, pin, now); - } - } - info!( - "Audit: peer {challenged_peer} passed subtree audit ({} leaves, {checked} \ - byte-checked)", - proof.leaves.len() - ); - AuditTickResult::Passed { - challenged_peer: *challenged_peer, - keys_checked: checked, - } + AuditVerdict::Pass { checked } => credit_and_pass(ctx, commitment, proof, checked).await, + } +} + +/// Full-pass epilogue of [`verify_subtree_response`]: soft closeness +/// observation, proven-holder credit, and the `Passed` result carrying the +/// audited commitment's key count (the ADR-0005 "at this size" fact). +async fn credit_and_pass( + ctx: &AuditCtx<'_>, + commitment: &StorageCommitment, + proof: &SubtreeProof, + checked: usize, +) -> AuditTickResult { + let challenged_peer = ctx.challenged_peer; + // Closeness (ADR-0002, soft/observe-only) — see observe_closeness. + observe_closeness(ctx.p2p_node, ctx.config, challenged_peer, proof).await; + // Credit the peer as a proven holder of its committed keys. + if let (Some(credit), Some(pin)) = (ctx.credit, commitment_hash(commitment)) { + let now = std::time::Instant::now(); + let mut provers = credit.recent_provers.write().await; + for leaf in &proof.leaves { + provers.record_proof(leaf.key, *challenged_peer, pin, now); } } + info!( + "Audit: peer {challenged_peer} passed subtree audit ({} leaves, {checked} \ + byte-checked)", + proof.leaves.len() + ); + AuditTickResult::Passed { + challenged_peer: *challenged_peer, + keys_checked: checked, + commitment_key_count: Some(commitment.key_count), + } } /// Soft, density-aware closeness observation (ADR-0002). Logs — never fails — diff --git a/src/storage/handler.rs b/src/storage/handler.rs index 9e5e7d0c..749b0ca8 100644 --- a/src/storage/handler.rs +++ b/src/storage/handler.rs @@ -177,6 +177,13 @@ impl AntProtocol { self.quote_generator.attach_commitment_source(source); } + /// ADR-0005: wire the replication engine's audit tally as the quote + /// generator's report source, so quote responses testify with this node's + /// signed audit report. + pub fn attach_report_source(&self, source: Arc) { + self.quote_generator.attach_report_source(source); + } + /// ADR-0004: return the proof with any commitment sidecars stripped, so a /// replicated/persisted receipt carries only the pin and count (stored /// proofs do not grow). Handles BOTH proof types — single-node and @@ -583,12 +590,20 @@ impl AntProtocol { let commitment = quote .commitment_pin .and_then(|pin| self.quote_generator.commitment_blob_for_pin(pin)); + // ADR-0005: this node's signed audit report (its tally rows, + // bound to the client's request nonce). `None` when there is + // nothing to testify — the response is valid, it just doesn't + // vouch for anyone. + let audit_report = self + .quote_generator + .build_audit_report(request.report_nonce); // Serialize the quote match rmp_serde::to_vec("e) { Ok(quote_bytes) => ChunkQuoteResponse::Success { quote: quote_bytes, already_stored, commitment, + audit_report, }, Err(e) => ChunkQuoteResponse::Error(ProtocolError::QuoteFailed(format!( "Failed to serialize quote: {e}" @@ -639,10 +654,15 @@ impl AntProtocol { let commitment = candidate_node .commitment_pin .and_then(|pin| self.quote_generator.commitment_blob_for_pin(pin)); + // ADR-0005: same signed audit report as the single-node path. + let audit_report = self + .quote_generator + .build_audit_report(request.report_nonce); match rmp_serde::to_vec(&candidate_node) { Ok(bytes) => MerkleCandidateQuoteResponse::Success { candidate_node: bytes, commitment, + audit_report, }, Err(e) => MerkleCandidateQuoteResponse::Error(ProtocolError::QuoteFailed( format!("Failed to serialize merkle candidate node: {e}"), @@ -775,6 +795,109 @@ mod tests { (protocol, temp_dir) } + /// Like [`create_test_protocol`] but also returns the identity's public + /// key bytes, for tests that need the peer binding (ADR-0005 reports). + async fn create_test_protocol_with_identity() -> (AntProtocol, TempDir, Vec) { + let temp_dir = TempDir::new().expect("create temp dir"); + let storage_config = LmdbStorageConfig { + root_dir: temp_dir.path().to_path_buf(), + disk_reserve: 0, + ..LmdbStorageConfig::test_default() + }; + let storage = Arc::new( + LmdbStorage::new(storage_config) + .await + .expect("create storage"), + ); + let rewards_address = RewardsAddress::new([1u8; 20]); + let payment_config = PaymentVerifierConfig { + evm: EvmVerifierConfig::default(), + cache_capacity: 100_000, + close_group_size: crate::ant_protocol::CLOSE_GROUP_SIZE, + local_rewards_address: rewards_address, + }; + let payment_verifier = Arc::new(PaymentVerifier::new(payment_config)); + let mut quote_generator = + QuoteGenerator::new(rewards_address, QuotingMetricsTracker::new(100)); + let identity = NodeIdentity::generate().expect("generate identity"); + let pub_key_bytes = identity.public_key().as_bytes().to_vec(); + let sk_bytes = identity.secret_key_bytes().to_vec(); + let sk = MlDsaSecretKey::from_bytes(&sk_bytes).expect("deserialize secret key"); + quote_generator.set_signer(pub_key_bytes.clone(), move |msg| { + use saorsa_pqc::pqc::MlDsaOperations; + MlDsa65::new() + .sign(&sk, msg) + .map_or_else(|_| vec![], |sig| sig.as_bytes().to_vec()) + }); + let protocol = AntProtocol::new(storage, payment_verifier, Arc::new(quote_generator)); + (protocol, temp_dir, pub_key_bytes) + } + + /// ADR-0005 end-to-end through the handler: a quote request whose node has + /// tally rows yields a response carrying a signed report that echoes the + /// request nonce and verifies against the quote's own key; with no rows the + /// response carries no report. + #[tokio::test] + async fn quote_response_carries_verifiable_audit_report() { + use crate::payment::quote::AuditReportSource; + use ant_protocol::payment::{verify_audit_report, AuditReport, AuditReportDay}; + + struct OneRow { + reporter_peer_id: [u8; 32], + } + impl AuditReportSource for OneRow { + fn reporter_peer_id(&self) -> [u8; 32] { + self.reporter_peer_id + } + fn report_rows(&self) -> Vec { + vec![ant_protocol::payment::AuditReportRow { + subject_peer_id: [6u8; 32], + days: vec![AuditReportDay { + age_days: 0, + passes: 2, + max_passed_key_count: 77, + }], + fenced: false, + convicted: false, + }] + } + } + + let (protocol, _temp, pub_key_bytes) = create_test_protocol_with_identity().await; + let reporter_peer_id = *blake3::hash(&pub_key_bytes).as_bytes(); + + // No report source attached -> no report on the wire. + let request = ChunkQuoteRequest { + address: [0x21u8; 32], + data_size: 100, + data_type: DATA_TYPE_CHUNK, + report_nonce: [0xABu8; 32], + }; + let ChunkQuoteResponse::Success { audit_report, .. } = protocol.handle_quote(&request) + else { + panic!("expected quote success"); + }; + assert!(audit_report.is_none(), "no source -> no report"); + + // Attach a source with one row; the response must testify with it. + protocol.attach_report_source(Arc::new(OneRow { reporter_peer_id })); + + let ChunkQuoteResponse::Success { audit_report, .. } = protocol.handle_quote(&request) + else { + panic!("expected quote success"); + }; + let bytes = audit_report.expect("report attached"); + let report: AuditReport = rmp_serde::from_slice(&bytes).expect("report parses"); + assert_eq!( + report.nonce, [0xABu8; 32], + "report echoes the request nonce" + ); + assert!( + verify_audit_report(&report, &pub_key_bytes, &reporter_peer_id, &[0xABu8; 32]), + "handler-built report verifies against the quote key" + ); + } + #[tokio::test] async fn test_put_and_get_chunk() { let (protocol, _temp) = create_test_protocol().await; @@ -1185,6 +1308,7 @@ mod tests { data_type: DATA_TYPE_CHUNK, data_size: 4096, merkle_payment_timestamp: timestamp, + report_nonce: [0u8; 32], }; let msg = ChunkMessage { request_id: 600, @@ -1268,6 +1392,7 @@ mod tests { address, data_size: content.len() as u64, data_type: DATA_TYPE_CHUNK, + report_nonce: [0u8; 32], }; let quote_msg = ChunkMessage { request_id: 301, @@ -1299,6 +1424,7 @@ mod tests { address: new_address, data_size: 100, data_type: DATA_TYPE_CHUNK, + report_nonce: [0u8; 32], }; let quote_msg2 = ChunkMessage { request_id: 302, @@ -1333,6 +1459,7 @@ mod tests { address: [0xAAu8; 32], // a quote-only probe, not one of the stored chunks data_size: 100, data_type: DATA_TYPE_CHUNK, + report_nonce: [0u8; 32], }; let _ = protocol.handle_quote("e_request); protocol.priced_records_stored()