From ec8b12be06f3049ad1015a52044a28792033a7b9 Mon Sep 17 00:00:00 2001 From: grumbach Date: Thu, 9 Jul 2026 14:07:24 +0900 Subject: [PATCH 1/2] Add ADR-0005 node-side audit tally for earned reward eligibility Every node already audits its neighbours (ADR-0002/0004); this records the outcomes so a client can decide, at quote collection, whether a quoter has earned a clean audited week at the size it wants to charge for. - audit_tally: per-peer, per-day tally of subtree-audit passes and the size proven, with sticky convictions and a fence for unanswered monetized-pin challenges. Rows and the fence age out with the window; the on-disk snapshot carries a magic+version tag so a format change is rejected cleanly rather than mis-decoded, and the observer starts fresh. - quote: attach the signed audit report (bound to the client's report_nonce) to quote responses. - replication/storage: wire pass/conviction/fence recording into the audit paths and persist the tally with write-on-change atomic rename. - Cargo: temporarily pin ant-protocol to the PR branch carrying the wire types (revert to a published bump once released). --- Cargo.toml | 8 + src/node.rs | 9 + src/payment/quote.rs | 148 ++++++ src/replication/audit.rs | 7 + src/replication/audit_tally.rs | 551 ++++++++++++++++++++ src/replication/config.rs | 68 ++- src/replication/mod.rs | 346 +++++++++++- src/replication/storage_commitment_audit.rs | 52 +- src/storage/handler.rs | 139 +++++ 9 files changed, 1297 insertions(+), 31 deletions(-) create mode 100644 src/replication/audit_tally.rs diff --git a/Cargo.toml b/Cargo.toml index ca894682..14a3843c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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 depends on the audit-report wire types added to ant-protocol in +# https://github.com/WithAutonomi/ant-protocol/pull/19, not yet on crates.io. +# Point the pin at that PR branch so this builds in CI. Revert to a published +# ant-protocol version bump once the wire types are released (evmlib is +# re-exported through ant-protocol, so no separate evmlib pin is needed). +[patch.crates-io] +ant-protocol = { git = "https://github.com/grumbach/ant-protocol.git", branch = "adr-0005-earned-reward-eligibility" } 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..92c9d99c 100644 --- a/src/payment/quote.rs +++ b/src/payment/quote.rs @@ -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,47 @@ 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> { + let source = self.report_source.read().as_ref().map(Arc::clone)?; + let sign_fn = self.sign_fn.as_ref()?; + let rows = source.report_rows(); + if rows.is_empty() { + return None; + } + 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, + }; + rmp_serde::to_vec(&report).ok() } /// Attach the ADR-0004 commitment source so quotes bind their price to the @@ -472,6 +530,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..4c814d66 --- /dev/null +++ b/src/replication/audit_tally.rs @@ -0,0 +1,551 @@ +//! 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. +/// +/// Defaults to a wall-clock day; the `ADR5_DAY_SECS` environment variable +/// shortens it for tests and local testnets (time compression: a "week" of +/// dues in minutes). Read once. +pub fn tally_day_secs() -> u64 { + use std::sync::OnceLock; + static DAY_SECS: OnceLock = OnceLock::new(); + *DAY_SECS.get_or_init(|| { + std::env::var("ADR5_DAY_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0) + .unwrap_or(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..eecc480d 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -382,6 +382,62 @@ 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; +/// ADR-0005 testnet-only audit-cadence compression knobs. Production never +/// sets these; a time-compressed local testnet ("a day" is `ADR5_DAY_SECS`) +/// sets them so audits outpace compressed days. Each reads its env var once. +fn adr5_test_env_u64(name: &'static str) -> Option { + std::env::var(name) + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0) +} + +/// `ADR5_TEST_AUDIT_TICK_SECS`: fixed audit tick interval override (both ends +/// of the production 10-20 min band). NOTE: the tick drives the +/// responsible-chunk audit lane, which deliberately does NOT feed the +/// ADR-0005 tally — compress it for general audit liveness, but the +/// tally-producing cadence is the gossip lane below. +pub(crate) fn adr5_test_audit_tick() -> Option { + use std::sync::OnceLock; + static V: OnceLock> = OnceLock::new(); + (*V.get_or_init(|| adr5_test_env_u64("ADR5_TEST_AUDIT_TICK_SECS"))).map(Duration::from_secs) +} + +/// `ADR5_TEST_NEIGHBOR_SYNC_SECS`: fixed neighbor-sync interval override +/// (both ends of the production 10-20 min band) — ALSO overrides the 1 h +/// per-peer sync cooldown, which would otherwise cap any pair at one sync per +/// hour regardless of interval. This is the cadence that feeds the ADR-0005 +/// tally: every sync ingests peers' commitments, and each VALID ingest rolls +/// the (audit-cooldown-gated) subtree-audit lottery whose passes and +/// convictions the tally records. +pub(crate) fn adr5_test_neighbor_sync() -> Option { + use std::sync::OnceLock; + static V: OnceLock> = OnceLock::new(); + (*V.get_or_init(|| adr5_test_env_u64("ADR5_TEST_NEIGHBOR_SYNC_SECS"))).map(Duration::from_secs) +} + +/// `ADR5_TEST_AUDIT_COOLDOWN_SECS`: per-peer gossip-audit cooldown override. +pub fn audit_on_gossip_cooldown_secs() -> u64 { + use std::sync::OnceLock; + static V: OnceLock = OnceLock::new(); + *V.get_or_init(|| { + adr5_test_env_u64("ADR5_TEST_AUDIT_COOLDOWN_SECS").unwrap_or(AUDIT_ON_GOSSIP_COOLDOWN_SECS) + }) +} + +/// `ADR5_TEST_AUDIT_PROBABILITY`: gossip-audit lottery override, parsed as a +/// float and clamped to [0, 1]. +pub fn audit_on_gossip_probability() -> f64 { + use std::sync::OnceLock; + static V: OnceLock = OnceLock::new(); + *V.get_or_init(|| { + std::env::var("ADR5_TEST_AUDIT_PROBABILITY") + .ok() + .and_then(|v| v.parse::().ok()) + .map_or(AUDIT_ON_GOSSIP_PROBABILITY, |p| p.clamp(0.0, 1.0)) + }) +} + /// 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 @@ -495,13 +551,15 @@ impl Default for ReplicationConfig { paid_list_close_group_size: PAID_LIST_CLOSE_GROUP_SIZE, neighbor_sync_scope: NEIGHBOR_SYNC_SCOPE, neighbor_sync_peer_count: NEIGHBOR_SYNC_PEER_COUNT, - neighbor_sync_interval_min: NEIGHBOR_SYNC_INTERVAL_MIN, - neighbor_sync_interval_max: NEIGHBOR_SYNC_INTERVAL_MAX, - neighbor_sync_cooldown: NEIGHBOR_SYNC_COOLDOWN, + neighbor_sync_interval_min: adr5_test_neighbor_sync() + .unwrap_or(NEIGHBOR_SYNC_INTERVAL_MIN), + neighbor_sync_interval_max: adr5_test_neighbor_sync() + .unwrap_or(NEIGHBOR_SYNC_INTERVAL_MAX), + neighbor_sync_cooldown: adr5_test_neighbor_sync().unwrap_or(NEIGHBOR_SYNC_COOLDOWN), self_lookup_interval_min: SELF_LOOKUP_INTERVAL_MIN, self_lookup_interval_max: SELF_LOOKUP_INTERVAL_MAX, - audit_tick_interval_min: AUDIT_TICK_INTERVAL_MIN, - audit_tick_interval_max: AUDIT_TICK_INTERVAL_MAX, + audit_tick_interval_min: adr5_test_audit_tick().unwrap_or(AUDIT_TICK_INTERVAL_MIN), + audit_tick_interval_max: adr5_test_audit_tick().unwrap_or(AUDIT_TICK_INTERVAL_MAX), audit_response_floor: Duration::from_secs(AUDIT_RESPONSE_FLOOR_SECS), audit_honest_read_bps: AUDIT_HONEST_READ_BPS, audit_response_honest_multiplier: AUDIT_RESPONSE_HONEST_MULTIPLIER, diff --git a/src/replication/mod.rs b/src/replication/mod.rs index bd827210..4af50954 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,86 @@ const BOOTSTRAP_DRAIN_CHECK_SECS: u64 = 5; /// "rotated past" case. const COMMITMENT_ROTATION_INTERVAL_SECS: u64 = 3600; +/// ADR-0005 run-3 pure-freeloader flag (test only): set when the fat-pin +/// cheat fires, so the node refuses ALL new storage afterwards (no re-supply +/// can rebuild its audit passes). Process-global, inert in production because +/// nothing sets it unless `ADR5_TEST_CHEAT_DELETE_AFTER_SECS` triggers. +static ADR5_CHEATING: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); + +/// Whether this node has entered fat-pin cheat mode (test only). Checked on +/// the storage-admission path so a cheater stays a pure freeloader. +#[must_use] +pub fn adr5_is_cheating() -> bool { + ADR5_CHEATING.load(std::sync::atomic::Ordering::Relaxed) +} + +/// ADR-0005 idle-honest flag (test only): set when the idle seam fires. An +/// idle node KEEPS all its earned data (so it keeps passing audits) but +/// accepts no NEW storage — the quiescent-honest case. Inert in production. +static ADR5_IDLE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); + +/// Whether this node has gone idle-honest (test only): refuses new storage but +/// serves everything it already holds, so it stays auditable and eligible. +#[must_use] +pub fn adr5_is_idle() -> bool { + ADR5_IDLE.load(std::sync::atomic::Ordering::Relaxed) +} + +/// Whether this node refuses new storage (test only). +/// +/// True for a fat-pin freeloader (which also deleted its data) or an +/// idle-honest node (which keeps it). One predicate for the three +/// storage-admission boundaries. Inert in production. +#[must_use] +pub fn adr5_refuses_new_storage() -> bool { + adr5_is_cheating() || adr5_is_idle() +} + +/// ADR-0005 idle-honest delay (test only): seconds of uptime after which the +/// node stops accepting new storage (but keeps serving what it has). `None` in +/// production (env unset). +fn adr5_test_idle_after() -> Option { + use std::sync::OnceLock; + static V: OnceLock> = OnceLock::new(); + (*V.get_or_init(|| { + std::env::var("ADR5_TEST_IDLE_AFTER_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0) + })) + .map(std::time::Duration::from_secs) +} + +/// ADR-0005 run-2 fat-pin cheater delay (test only): seconds of uptime after +/// which the rotation loop deletes all chunks and freezes the fat commitment. +/// `None` in production (env unset) — the seam is inert. +fn adr5_test_cheat_delete_after() -> Option { + use std::sync::OnceLock; + static V: OnceLock> = OnceLock::new(); + (*V.get_or_init(|| { + std::env::var("ADR5_TEST_CHEAT_DELETE_AFTER_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0) + })) + .map(std::time::Duration::from_secs) +} + +/// Effective rotation interval: production hourly, or the ADR-0005 testnet +/// override (`ADR5_TEST_ROTATION_SECS`) so freshly-stored chunks enter a +/// commitment — and therefore the audit/tally lane — within compressed days. +fn commitment_rotation_interval() -> std::time::Duration { + use std::sync::OnceLock; + static V: OnceLock = OnceLock::new(); + std::time::Duration::from_secs(*V.get_or_init(|| { + std::env::var("ADR5_TEST_ROTATION_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0) + .unwrap_or(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 +288,22 @@ 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: production value above, or the +/// compressed neighbor-sync interval when the ADR-0005 testnet knob is set — +/// otherwise the 60 s cap would silently throttle the compressed gossip lane +/// back to ~1 valid ingest/min/pair and starve the tally cadence. +fn commitment_sig_verify_min_interval() -> Duration { + use std::sync::OnceLock; + static V: OnceLock = OnceLock::new(); + *V.get_or_init(|| { + std::env::var("ADR5_TEST_NEIGHBOR_SYNC_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .filter(|v| *v > 0) + .map_or(COMMITMENT_SIG_VERIFY_MIN_INTERVAL, Duration::from_secs) + }) +} + /// Hard cap on the size of `last_commitment_by_peer`. /// /// Bounds the per-process memory cost of the auditor's per-peer @@ -361,6 +458,14 @@ 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>, + /// 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 +536,8 @@ 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_path: root_dir.join("audit_tally.bin"), shutdown, task_handles: Vec::new(), }; @@ -439,6 +546,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 +560,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 +931,7 @@ 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), }; let shutdown = self.shutdown.clone(); @@ -949,8 +1067,32 @@ impl ReplicationEngine { &trigger.sync_state, &trigger.recent_provers, &trigger.config, + &trigger.audit_tally, ) .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 + { + if let Ok(mut tally) = trigger.audit_tally.write() { + tally.record_unanswered_monetized_challenge( + *challenged_peer, + audit_tally::unix_now_secs(), + ); + } + } }); } } @@ -982,6 +1124,7 @@ 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 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 +1137,7 @@ 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), }; let handle = tokio::spawn(async move { @@ -1161,6 +1305,7 @@ 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), }; let handle = tokio::spawn(async move { @@ -1361,12 +1506,80 @@ impl ReplicationEngine { } else { sync_trigger.notify_one(); } + // ADR-0005 run-2 fat-pin cheater seam (test only). When + // `ADR5_TEST_CHEAT_DELETE_AFTER_SECS` is set, after that uptime the + // node DELETES all its chunks once and then FREEZES its commitment + // (skips every rotation) — so it keeps quoting and answering the + // pin it can no longer serve. Its next byte challenge is a + // confirmed failure -> deterministic conviction, exactly the + // delete-and-keep-collecting attacker. No effect without the env. + let cheat_after = adr5_test_cheat_delete_after(); + // ADR-0005 idle-honest seam (test only): after `ADR5_TEST_IDLE_AFTER_SECS` + // the node stops accepting NEW storage but keeps everything it has, + // so it stays auditable and should stay eligible — the quiescent + // honest case. No delete, no freeze; rotation continues normally. + let idle_after = adr5_test_idle_after(); + let started = tokio::time::Instant::now(); + let mut cheating = false; + let mut idle = false; 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 Some(after) = idle_after { + if !idle && started.elapsed() >= after { + ADR5_IDLE.store(true, std::sync::atomic::Ordering::Relaxed); + warn!("ADR5 TEST IDLE: node going quiescent (keeps data, refuses new storage)"); + idle = true; + } + } + if let Some(after) = cheat_after { + if !cheating && started.elapsed() >= after { + // Become a pure freeloader: the flag gates every + // storage-REQUEST boundary (PUT handler, + // replication offer, fetch/repair worker), so no + // re-supply can rebuild audit passes. Set the + // flag FIRST, then purge all keys once. A store + // already admitted before the flag flip can leave + // a few stragglers, but the subtree audit's round + // 1 reads EVERY leaf of a >= ceil(sqrt(N)) subtree + // — one absent leaf fails it confirmed — so fewer + // than ceil(sqrt(N)) survivors cannot pass any + // audit, and the measurement is sound (a delete + // error or a residual reaching ceil(sqrt(N)) + // invalidates the run, guarded by the runner). + ADR5_CHEATING + .store(true, std::sync::atomic::Ordering::Relaxed); + let mut deleted = 0usize; + let mut delete_errs = 0usize; + match storage.all_keys().await { + Ok(keys) => { + for k in &keys { + match storage.delete(k).await { + Ok(_) => deleted += 1, + Err(e) => { + delete_errs += 1; + warn!("ADR5 TEST CHEAT: delete failed: {e}"); + } + } + } + } + Err(e) => { + delete_errs += 1; + warn!("ADR5 TEST CHEAT: all_keys failed: {e}"); + } + } + warn!( + "ADR5 TEST CHEAT: purged {deleted} chunk(s) ({delete_errs} \ + error(s)), freezing fat commitment, refusing all new storage" + ); + cheating = true; + } + if cheating { + // Frozen: keep the fat pin, never rebuild. + continue; + } + } if let Err(e) = rebuild_and_rotate_commitment( &storage, &identity, @@ -1402,15 +1615,23 @@ impl ReplicationEngine { 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_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,6 +1639,10 @@ 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; } } } @@ -2480,6 +2705,12 @@ async fn handle_fresh_offer( } } + // ADR-0005 (test only): a fat-pin freeloader OR an idle-honest node + // refuses replicated offers at the request boundary. Inert in production. + if adr5_refuses_new_storage() { + return Ok(()); + } + // Rule 6: add to PaidForList. if let Err(e) = paid_list.insert(&offer.key).await { warn!("Failed to add key to PaidForList: {e}"); @@ -3968,6 +4199,15 @@ async fn execute_single_fetch( }; } + // ADR-0005 (test only): a freeloader or idle-honest node does + // not store fetched/repaired records either. Inert in prod. + if adr5_refuses_new_storage() { + return FetchOutcome { + key, + result: FetchResult::SourceFailed, + }; + } + if let Err(e) = storage.put(&resp_key, &data).await { warn!( "Failed to store fetched record {}: {e}", @@ -4088,6 +4328,7 @@ async fn handle_subtree_failed_audit( p2p_node: &Arc, sync_state: &Arc>, recent_provers: &Arc>, + audit_tally: &std::sync::RwLock, ) { if matches!(reason, AuditFailureReason::Timeout) { debug!( @@ -4097,6 +4338,12 @@ 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. + if let Ok(mut tally) = audit_tally.write() { + tally.record_conviction(*challenged_peer, audit_tally::unix_now_secs()); + } + // 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 +4370,22 @@ async fn handle_subtree_audit_result( sync_state: &Arc>, recent_provers: &Arc>, config: &ReplicationConfig, + audit_tally: &std::sync::RwLock, ) { 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 +4428,7 @@ async fn handle_subtree_audit_result( p2p_node, sync_state, recent_provers, + audit_tally, ) .await; } @@ -4249,6 +4506,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 +4639,8 @@ struct GossipAuditTrigger { recent_provers: Arc>, sync_state: Arc>, cooldown: Arc>>, + /// ADR-0005: subtree-audit outcomes land here as tally facts. + audit_tally: Arc>, } /// What a gossip ingest yields for the audit trigger: the commitment hash to @@ -4423,7 +4685,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 +4745,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 +4774,7 @@ async fn maybe_trigger_gossip_audit( &trigger.sync_state, &trigger.recent_provers, &trigger.config, + &trigger.audit_tally, ) .await; }); @@ -4532,7 +4795,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 +5094,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()).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 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..7ae0ad15 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 @@ -301,6 +308,18 @@ impl AntProtocol { let addr_hex = hex::encode(address); debug!("Handling PUT request for {addr_hex}"); + // ADR-0005 run-3 fat-pin cheater (test only): a freeloader refuses new + // storage at the REQUEST boundary — before the chunk reaches the store + // — so no re-supply rebuilds its audit passes. A request already + // admitted before the flag flip may still land (bounded stragglers); + // that does not corrupt the measurement (see the cheat branch in + // replication::mod). Inert in production (flag only set by env-gated cheat). + if crate::replication::adr5_refuses_new_storage() { + return ChunkPutResponse::Error(ProtocolError::StorageFailed( + "node in ADR5 test cheat mode: refusing storage".to_string(), + )); + } + // 1. Validate chunk size if request.content.len() > MAX_CHUNK_SIZE { return ChunkPutResponse::Error(ProtocolError::ChunkTooLarge { @@ -583,12 +602,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 +666,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 +807,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 +1320,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 +1404,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 +1436,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 +1471,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() From 111320385aa5a4eb8d90a19c13e63a7aaf172af4 Mon Sep 17 00:00:00 2001 From: grumbach Date: Fri, 10 Jul 2026 07:52:27 +0900 Subject: [PATCH 2/2] Remove ADR-0005 test seams; make production node minimal and durable Delete every test/testnet-only ADR-0005 seam from the node so no release binary carries them: - the destructive fat-pin cheater and idle-honest seams (ADR5_TEST_CHEAT_DELETE_AFTER_SECS / ADR5_TEST_IDLE_AFTER_SECS), which could delete all stored chunks or refuse storage if an env var leaked; - the compressed-time knobs (ADR5_DAY_SECS and the ADR5_TEST_* cadence overrides), so an untrusted reporter can never redefine a "day" and the tally day is a fixed 86400s in production. The seam functions collapse to their production form and the storage admission paths always proceed, so production behaviour is unchanged; the CI tests direct-drive the tally, so nothing depended on the seams. Also harden persistence and the report producer: - flush the tally eagerly on a conviction or fence, not only on the 30s periodic loop, so a crash cannot lose negative evidence; - give the audit-tally persist/load path its own log labels instead of reusing the commitment-retention ones; - enforce the report row/day/byte caps in the producer before signing, so an over-cap report is never put on the wire. Add the ADR-0005 source document and pin ant-protocol to the immutable PR rev for a reproducible build. --- Cargo.lock | 5 +- Cargo.toml | 14 +- .../adr/ADR-0005-earned-reward-eligibility.md | 249 +++++++++++++++++ src/payment/quote.rs | 29 +- src/replication/audit_tally.rs | 15 +- src/replication/config.rs | 68 +---- src/replication/mod.rs | 253 +++++------------- src/storage/handler.rs | 12 - 8 files changed, 357 insertions(+), 288 deletions(-) create mode 100644 docs/adr/ADR-0005-earned-reward-eligibility.md 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 14a3843c..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" @@ -197,10 +197,10 @@ cognitive_complexity = "allow" # Allow non-const functions during initial development (may need runtime features later) missing_const_for_fn = "allow" -# ADR-0005 depends on the audit-report wire types added to ant-protocol in -# https://github.com/WithAutonomi/ant-protocol/pull/19, not yet on crates.io. -# Point the pin at that PR branch so this builds in CI. Revert to a published -# ant-protocol version bump once the wire types are released (evmlib is -# re-exported through ant-protocol, so no separate evmlib pin is needed). +# 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.git", branch = "adr-0005-earned-reward-eligibility" } +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/payment/quote.rs b/src/payment/quote.rs index 92c9d99c..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; @@ -151,12 +151,22 @@ impl QuoteGenerator { /// 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 rows = source.report_rows(); + 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); @@ -170,7 +180,20 @@ impl QuoteGenerator { rows, signature, }; - rmp_serde::to_vec(&report).ok() + 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 diff --git a/src/replication/audit_tally.rs b/src/replication/audit_tally.rs index 4c814d66..1c0f1b5e 100644 --- a/src/replication/audit_tally.rs +++ b/src/replication/audit_tally.rs @@ -55,20 +55,9 @@ pub fn unix_now_secs() -> u64 { } /// Length of one tally day bucket in seconds. -/// -/// Defaults to a wall-clock day; the `ADR5_DAY_SECS` environment variable -/// shortens it for tests and local testnets (time compression: a "week" of -/// dues in minutes). Read once. +#[must_use] pub fn tally_day_secs() -> u64 { - use std::sync::OnceLock; - static DAY_SECS: OnceLock = OnceLock::new(); - *DAY_SECS.get_or_init(|| { - std::env::var("ADR5_DAY_SECS") - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|v| *v > 0) - .unwrap_or(86_400) - }) + 86_400 } /// One day bucket of outcomes for one observed peer. diff --git a/src/replication/config.rs b/src/replication/config.rs index eecc480d..e8acd44d 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -382,60 +382,16 @@ 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; -/// ADR-0005 testnet-only audit-cadence compression knobs. Production never -/// sets these; a time-compressed local testnet ("a day" is `ADR5_DAY_SECS`) -/// sets them so audits outpace compressed days. Each reads its env var once. -fn adr5_test_env_u64(name: &'static str) -> Option { - std::env::var(name) - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|v| *v > 0) -} - -/// `ADR5_TEST_AUDIT_TICK_SECS`: fixed audit tick interval override (both ends -/// of the production 10-20 min band). NOTE: the tick drives the -/// responsible-chunk audit lane, which deliberately does NOT feed the -/// ADR-0005 tally — compress it for general audit liveness, but the -/// tally-producing cadence is the gossip lane below. -pub(crate) fn adr5_test_audit_tick() -> Option { - use std::sync::OnceLock; - static V: OnceLock> = OnceLock::new(); - (*V.get_or_init(|| adr5_test_env_u64("ADR5_TEST_AUDIT_TICK_SECS"))).map(Duration::from_secs) -} - -/// `ADR5_TEST_NEIGHBOR_SYNC_SECS`: fixed neighbor-sync interval override -/// (both ends of the production 10-20 min band) — ALSO overrides the 1 h -/// per-peer sync cooldown, which would otherwise cap any pair at one sync per -/// hour regardless of interval. This is the cadence that feeds the ADR-0005 -/// tally: every sync ingests peers' commitments, and each VALID ingest rolls -/// the (audit-cooldown-gated) subtree-audit lottery whose passes and -/// convictions the tally records. -pub(crate) fn adr5_test_neighbor_sync() -> Option { - use std::sync::OnceLock; - static V: OnceLock> = OnceLock::new(); - (*V.get_or_init(|| adr5_test_env_u64("ADR5_TEST_NEIGHBOR_SYNC_SECS"))).map(Duration::from_secs) -} - -/// `ADR5_TEST_AUDIT_COOLDOWN_SECS`: per-peer gossip-audit cooldown override. +/// Per-peer gossip-audit cooldown, in seconds. +#[must_use] pub fn audit_on_gossip_cooldown_secs() -> u64 { - use std::sync::OnceLock; - static V: OnceLock = OnceLock::new(); - *V.get_or_init(|| { - adr5_test_env_u64("ADR5_TEST_AUDIT_COOLDOWN_SECS").unwrap_or(AUDIT_ON_GOSSIP_COOLDOWN_SECS) - }) + AUDIT_ON_GOSSIP_COOLDOWN_SECS } -/// `ADR5_TEST_AUDIT_PROBABILITY`: gossip-audit lottery override, parsed as a -/// float and clamped to [0, 1]. +/// Gossip-audit lottery probability. +#[must_use] pub fn audit_on_gossip_probability() -> f64 { - use std::sync::OnceLock; - static V: OnceLock = OnceLock::new(); - *V.get_or_init(|| { - std::env::var("ADR5_TEST_AUDIT_PROBABILITY") - .ok() - .and_then(|v| v.parse::().ok()) - .map_or(AUDIT_ON_GOSSIP_PROBABILITY, |p| p.clamp(0.0, 1.0)) - }) + AUDIT_ON_GOSSIP_PROBABILITY } /// ADR-0004: first-audit drainer retry cadence for cooldown-pending pins. @@ -551,15 +507,13 @@ impl Default for ReplicationConfig { paid_list_close_group_size: PAID_LIST_CLOSE_GROUP_SIZE, neighbor_sync_scope: NEIGHBOR_SYNC_SCOPE, neighbor_sync_peer_count: NEIGHBOR_SYNC_PEER_COUNT, - neighbor_sync_interval_min: adr5_test_neighbor_sync() - .unwrap_or(NEIGHBOR_SYNC_INTERVAL_MIN), - neighbor_sync_interval_max: adr5_test_neighbor_sync() - .unwrap_or(NEIGHBOR_SYNC_INTERVAL_MAX), - neighbor_sync_cooldown: adr5_test_neighbor_sync().unwrap_or(NEIGHBOR_SYNC_COOLDOWN), + neighbor_sync_interval_min: NEIGHBOR_SYNC_INTERVAL_MIN, + neighbor_sync_interval_max: NEIGHBOR_SYNC_INTERVAL_MAX, + neighbor_sync_cooldown: NEIGHBOR_SYNC_COOLDOWN, self_lookup_interval_min: SELF_LOOKUP_INTERVAL_MIN, self_lookup_interval_max: SELF_LOOKUP_INTERVAL_MAX, - audit_tick_interval_min: adr5_test_audit_tick().unwrap_or(AUDIT_TICK_INTERVAL_MIN), - audit_tick_interval_max: adr5_test_audit_tick().unwrap_or(AUDIT_TICK_INTERVAL_MAX), + audit_tick_interval_min: AUDIT_TICK_INTERVAL_MIN, + audit_tick_interval_max: AUDIT_TICK_INTERVAL_MAX, audit_response_floor: Duration::from_secs(AUDIT_RESPONSE_FLOOR_SECS), audit_honest_read_bps: AUDIT_HONEST_READ_BPS, audit_response_honest_multiplier: AUDIT_RESPONSE_HONEST_MULTIPLIER, diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 4af50954..ada4ecbb 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -160,84 +160,9 @@ const BOOTSTRAP_DRAIN_CHECK_SECS: u64 = 5; /// "rotated past" case. const COMMITMENT_ROTATION_INTERVAL_SECS: u64 = 3600; -/// ADR-0005 run-3 pure-freeloader flag (test only): set when the fat-pin -/// cheat fires, so the node refuses ALL new storage afterwards (no re-supply -/// can rebuild its audit passes). Process-global, inert in production because -/// nothing sets it unless `ADR5_TEST_CHEAT_DELETE_AFTER_SECS` triggers. -static ADR5_CHEATING: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); - -/// Whether this node has entered fat-pin cheat mode (test only). Checked on -/// the storage-admission path so a cheater stays a pure freeloader. -#[must_use] -pub fn adr5_is_cheating() -> bool { - ADR5_CHEATING.load(std::sync::atomic::Ordering::Relaxed) -} - -/// ADR-0005 idle-honest flag (test only): set when the idle seam fires. An -/// idle node KEEPS all its earned data (so it keeps passing audits) but -/// accepts no NEW storage — the quiescent-honest case. Inert in production. -static ADR5_IDLE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); - -/// Whether this node has gone idle-honest (test only): refuses new storage but -/// serves everything it already holds, so it stays auditable and eligible. -#[must_use] -pub fn adr5_is_idle() -> bool { - ADR5_IDLE.load(std::sync::atomic::Ordering::Relaxed) -} - -/// Whether this node refuses new storage (test only). -/// -/// True for a fat-pin freeloader (which also deleted its data) or an -/// idle-honest node (which keeps it). One predicate for the three -/// storage-admission boundaries. Inert in production. -#[must_use] -pub fn adr5_refuses_new_storage() -> bool { - adr5_is_cheating() || adr5_is_idle() -} - -/// ADR-0005 idle-honest delay (test only): seconds of uptime after which the -/// node stops accepting new storage (but keeps serving what it has). `None` in -/// production (env unset). -fn adr5_test_idle_after() -> Option { - use std::sync::OnceLock; - static V: OnceLock> = OnceLock::new(); - (*V.get_or_init(|| { - std::env::var("ADR5_TEST_IDLE_AFTER_SECS") - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|v| *v > 0) - })) - .map(std::time::Duration::from_secs) -} - -/// ADR-0005 run-2 fat-pin cheater delay (test only): seconds of uptime after -/// which the rotation loop deletes all chunks and freezes the fat commitment. -/// `None` in production (env unset) — the seam is inert. -fn adr5_test_cheat_delete_after() -> Option { - use std::sync::OnceLock; - static V: OnceLock> = OnceLock::new(); - (*V.get_or_init(|| { - std::env::var("ADR5_TEST_CHEAT_DELETE_AFTER_SECS") - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|v| *v > 0) - })) - .map(std::time::Duration::from_secs) -} - -/// Effective rotation interval: production hourly, or the ADR-0005 testnet -/// override (`ADR5_TEST_ROTATION_SECS`) so freshly-stored chunks enter a -/// commitment — and therefore the audit/tally lane — within compressed days. +/// Effective rotation interval. fn commitment_rotation_interval() -> std::time::Duration { - use std::sync::OnceLock; - static V: OnceLock = OnceLock::new(); - std::time::Duration::from_secs(*V.get_or_init(|| { - std::env::var("ADR5_TEST_ROTATION_SECS") - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|v| *v > 0) - .unwrap_or(COMMITMENT_ROTATION_INTERVAL_SECS) - })) + std::time::Duration::from_secs(COMMITMENT_ROTATION_INTERVAL_SECS) } /// How often the responder retention snapshot is flushed to disk (ADR-0004 A1). @@ -288,20 +213,9 @@ 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: production value above, or the -/// compressed neighbor-sync interval when the ADR-0005 testnet knob is set — -/// otherwise the 60 s cap would silently throttle the compressed gossip lane -/// back to ~1 valid ingest/min/pair and starve the tally cadence. +/// The effective per-peer sig-verify interval. fn commitment_sig_verify_min_interval() -> Duration { - use std::sync::OnceLock; - static V: OnceLock = OnceLock::new(); - *V.get_or_init(|| { - std::env::var("ADR5_TEST_NEIGHBOR_SYNC_SECS") - .ok() - .and_then(|v| v.parse::().ok()) - .filter(|v| *v > 0) - .map_or(COMMITMENT_SIG_VERIFY_MIN_INTERVAL, Duration::from_secs) - }) + COMMITMENT_SIG_VERIFY_MIN_INTERVAL } /// Hard cap on the size of `last_commitment_by_peer`. @@ -463,6 +377,9 @@ pub struct ReplicationEngine { /// 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, @@ -537,6 +454,7 @@ impl ReplicationEngine { 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(), @@ -932,6 +850,7 @@ impl ReplicationEngine { 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(); @@ -943,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 @@ -953,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), ); @@ -1068,6 +991,7 @@ impl ReplicationEngine { &trigger.recent_provers, &trigger.config, &trigger.audit_tally, + &trigger.audit_tally_persist_trigger, ) .await; // ADR-0005: this is the MONETIZED first-audit lane. A @@ -1086,11 +1010,16 @@ impl ReplicationEngine { }, } = &result { - if let Ok(mut tally) = trigger.audit_tally.write() { - tally.record_unanswered_monetized_challenge( - *challenged_peer, - audit_tally::unix_now_secs(), - ); + 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(); } } }); @@ -1125,6 +1054,7 @@ impl ReplicationEngine { 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); @@ -1138,6 +1068,7 @@ impl ReplicationEngine { 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 { @@ -1306,6 +1237,7 @@ impl ReplicationEngine { 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 { @@ -1506,80 +1438,10 @@ impl ReplicationEngine { } else { sync_trigger.notify_one(); } - // ADR-0005 run-2 fat-pin cheater seam (test only). When - // `ADR5_TEST_CHEAT_DELETE_AFTER_SECS` is set, after that uptime the - // node DELETES all its chunks once and then FREEZES its commitment - // (skips every rotation) — so it keeps quoting and answering the - // pin it can no longer serve. Its next byte challenge is a - // confirmed failure -> deterministic conviction, exactly the - // delete-and-keep-collecting attacker. No effect without the env. - let cheat_after = adr5_test_cheat_delete_after(); - // ADR-0005 idle-honest seam (test only): after `ADR5_TEST_IDLE_AFTER_SECS` - // the node stops accepting NEW storage but keeps everything it has, - // so it stays auditable and should stay eligible — the quiescent - // honest case. No delete, no freeze; rotation continues normally. - let idle_after = adr5_test_idle_after(); - let started = tokio::time::Instant::now(); - let mut cheating = false; - let mut idle = false; loop { tokio::select! { () = shutdown.cancelled() => break, () = tokio::time::sleep(commitment_rotation_interval()) => { - if let Some(after) = idle_after { - if !idle && started.elapsed() >= after { - ADR5_IDLE.store(true, std::sync::atomic::Ordering::Relaxed); - warn!("ADR5 TEST IDLE: node going quiescent (keeps data, refuses new storage)"); - idle = true; - } - } - if let Some(after) = cheat_after { - if !cheating && started.elapsed() >= after { - // Become a pure freeloader: the flag gates every - // storage-REQUEST boundary (PUT handler, - // replication offer, fetch/repair worker), so no - // re-supply can rebuild audit passes. Set the - // flag FIRST, then purge all keys once. A store - // already admitted before the flag flip can leave - // a few stragglers, but the subtree audit's round - // 1 reads EVERY leaf of a >= ceil(sqrt(N)) subtree - // — one absent leaf fails it confirmed — so fewer - // than ceil(sqrt(N)) survivors cannot pass any - // audit, and the measurement is sound (a delete - // error or a residual reaching ceil(sqrt(N)) - // invalidates the run, guarded by the runner). - ADR5_CHEATING - .store(true, std::sync::atomic::Ordering::Relaxed); - let mut deleted = 0usize; - let mut delete_errs = 0usize; - match storage.all_keys().await { - Ok(keys) => { - for k in &keys { - match storage.delete(k).await { - Ok(_) => deleted += 1, - Err(e) => { - delete_errs += 1; - warn!("ADR5 TEST CHEAT: delete failed: {e}"); - } - } - } - } - Err(e) => { - delete_errs += 1; - warn!("ADR5 TEST CHEAT: all_keys failed: {e}"); - } - } - warn!( - "ADR5 TEST CHEAT: purged {deleted} chunk(s) ({delete_errs} \ - error(s)), freezing fat commitment, refusing all new storage" - ); - cheating = true; - } - if cheating { - // Frozen: keep the fat pin, never rebuild. - continue; - } - } if let Err(e) = rebuild_and_rotate_commitment( &storage, &identity, @@ -1608,14 +1470,16 @@ 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 { @@ -1644,9 +1508,16 @@ impl ReplicationEngine { ) .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); } @@ -2705,12 +2576,6 @@ async fn handle_fresh_offer( } } - // ADR-0005 (test only): a fat-pin freeloader OR an idle-honest node - // refuses replicated offers at the request boundary. Inert in production. - if adr5_refuses_new_storage() { - return Ok(()); - } - // Rule 6: add to PaidForList. if let Err(e) = paid_list.insert(&offer.key).await { warn!("Failed to add key to PaidForList: {e}"); @@ -4199,15 +4064,6 @@ async fn execute_single_fetch( }; } - // ADR-0005 (test only): a freeloader or idle-honest node does - // not store fetched/repaired records either. Inert in prod. - if adr5_refuses_new_storage() { - return FetchOutcome { - key, - result: FetchResult::SourceFailed, - }; - } - if let Err(e) = storage.put(&resp_key, &data).await { warn!( "Failed to store fetched record {}: {e}", @@ -4321,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, @@ -4329,6 +4186,7 @@ async fn handle_subtree_failed_audit( sync_state: &Arc>, recent_provers: &Arc>, audit_tally: &std::sync::RwLock, + audit_tally_persist_trigger: &Notify, ) { if matches!(reason, AuditFailureReason::Timeout) { debug!( @@ -4340,8 +4198,12 @@ async fn handle_subtree_failed_audit( // ADR-0005: a confirmed subtree failure is a deterministic conviction — // the peer's tally row is zeroed and its dues start over. - if let Ok(mut tally) = audit_tally.write() { + 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 @@ -4371,6 +4233,7 @@ async fn handle_subtree_audit_result( recent_provers: &Arc>, config: &ReplicationConfig, audit_tally: &std::sync::RwLock, + audit_tally_persist_trigger: &Notify, ) { match result { AuditTickResult::Passed { @@ -4429,6 +4292,7 @@ async fn handle_subtree_audit_result( sync_state, recent_provers, audit_tally, + audit_tally_persist_trigger, ) .await; } @@ -4641,6 +4505,8 @@ struct GossipAuditTrigger { 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 @@ -4775,6 +4641,7 @@ async fn maybe_trigger_gossip_audit( &trigger.recent_provers, &trigger.config, &trigger.audit_tally, + &trigger.audit_tally_persist_trigger, ) .await; }); @@ -5160,7 +5027,7 @@ async fn persist_audit_tally_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(), "Audit tally").await { *last = Some(bytes); } } @@ -5182,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"); @@ -5212,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/storage/handler.rs b/src/storage/handler.rs index 7ae0ad15..749b0ca8 100644 --- a/src/storage/handler.rs +++ b/src/storage/handler.rs @@ -308,18 +308,6 @@ impl AntProtocol { let addr_hex = hex::encode(address); debug!("Handling PUT request for {addr_hex}"); - // ADR-0005 run-3 fat-pin cheater (test only): a freeloader refuses new - // storage at the REQUEST boundary — before the chunk reaches the store - // — so no re-supply rebuilds its audit passes. A request already - // admitted before the flag flip may still land (bounded stragglers); - // that does not corrupt the measurement (see the cheat branch in - // replication::mod). Inert in production (flag only set by env-gated cheat). - if crate::replication::adr5_refuses_new_storage() { - return ChunkPutResponse::Error(ProtocolError::StorageFailed( - "node in ADR5 test cheat mode: refusing storage".to_string(), - )); - } - // 1. Validate chunk size if request.content.len() > MAX_CHUNK_SIZE { return ChunkPutResponse::Error(ProtocolError::ChunkTooLarge {