From 11e4c18d3569add1472d38b136ca0037736c03f8 Mon Sep 17 00:00:00 2001 From: grumbach Date: Tue, 29 Sep 2026 14:55:27 +0900 Subject: [PATCH 1/2] fix(pointer): serve the record round 1 bound, however many updates follow A storage audit binds, in round 1, a nonced root over each pointer record a node holds, and round 2 must serve the bytes that reproduce it. The node kept only the record the last update replaced, so two paid updates to a pointer between the rounds made an honest holder fail round 2 with DigestMismatch, a confirmed failure that feeds the trust penalty. An owner who is also one of the holder's auditors knows exactly when its round 1 has been answered, so this was a cheap way to penalise a chosen honest neighbour. The round-1 session now keeps the root reported for each pointer leaf, the pointer store keeps every record an update replaces rather than only the last one, and round 2 serves the one record, held or replaced, whose root matches. Replaced records are kept for ten minutes, longer than a round 1 over the largest subtree an auditor waits for plus the session its round 2 must arrive within. Roots are capped at 65,536 across all live sessions, the oldest sessions giving theirs up first. When a node cannot serve the record round 1 read, because it aged out, was evicted, or the session kept no root and the pointer has been updated since, round 2 is rejected as Transient instead of guessing: no trust penalty, only the credit of that audit. A node that holds nothing at all for the pointer is still reported absent, as before. The auditor, the wire format and the subtree-audit protocol id are unchanged. ADR-0017 records the change and amends one point of ADR-0016, which is left as written. --- ...audits-serve-the-record-round-one-bound.md | 154 ++++++++ src/pointer/store.rs | 142 ++++++-- src/replication/config.rs | 12 + src/replication/mod.rs | 178 +++++++-- src/replication/protocol.rs | 14 +- src/replication/storage_commitment_audit.rs | 339 ++++++++++++++++-- tests/e2e/pointer_replication.rs | 159 +++++++- 7 files changed, 914 insertions(+), 84 deletions(-) create mode 100644 docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md diff --git a/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md b/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md new file mode 100644 index 00000000..bb9396a4 --- /dev/null +++ b/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md @@ -0,0 +1,154 @@ +# ADR-0017: Pointer audits serve the record round 1 bound + +- **Status:** Proposed +- **Date:** 2026-09-29 +- **Decision owners:** Anselme (@grumbach) +- **Reviewers:** +- **Supersedes:** none. It amends one point of ADR-0016, "Updates between the + rounds", and leaves the rest of ADR-0016 as written. +- **Superseded by:** none +- **Related:** ADR-0002 (audit), ADR-0009 (audit families), ADR-0016 (pointers) + +## Context + +A storage audit is two rounds (ADR-0002, ADR-0009). Round 1 reports, for every +leaf of the audited subtree, a root over the bytes the node holds, keyed by the +audit's fresh nonce. Round 2 then opens a few of those leaves, and the node must +serve bytes that reproduce the root round 1 reported. + +For a pointer (ADR-0016) those bytes are the whole signed record, and the owner +may replace the record at any time with a paid update. ADR-0016 covers one +update between the rounds: the store keeps the record an update replaced for +five minutes, and round 2 serves it beside the new one. It accepts the case of +two updates: + +> Two updates to one pointer inside the same audit would fail an honest holder, +> and would take the owner two paid updates within seconds of each other. + +Two things make that worth closing rather than accepting. + +- **The failure is charged to the wrong party.** The auditor reports + `DigestMismatch`, a confirmed failure, and the holder takes the trust penalty + for its owner's activity. Nothing the holder did was wrong. +- **It can be aimed.** An owner who is also a close-group auditor of the holder + knows exactly when its round 1 has been answered, and two paid updates cost + two writes, about 0.013 ANT each on the 990-node pointer testnet. ADR-0016 + already notes that an owner can grind a node id beside its own pointer. That + turns an accepted edge case into a cheap way to penalise a chosen honest + neighbour. + +The cause is narrow. Round 2 serves the record held now and the one last +replaced, because the node has not kept what round 1 bound and so cannot tell +which record it owes. After two updates neither of those is the one round 1 +read. + +## Decision Drivers + +- An honest holder must not take a confirmed failure because its owner updated + the pointer, however often. +- No wire change and no change to what the auditor accepts. Pointers have not + shipped in any release yet, but the smaller change is still the better one. +- Memory stays bounded against an auditor that opens many round-1 sessions, and + running out of it must never turn into a confirmed failure either. + +## Considered Options + +1. **Keep ADR-0016 as written.** Cheapest, and it leaves the failure above. +2. **Serve every record held since round 1.** It needs a larger per-item cap + than `MAX_POINTER_RECORDS_PER_ITEM`, so the auditor's check changes, and any + cap is still one more paid update away from failing. +3. **Remember what round 1 bound, and serve that record.** Round 1 already + reports a nonced root per pointer leaf. The node keeps those roots in the + round-1 session it already holds, and round 2 serves the one record, + current or replaced, whose root matches. + +## Decision + +We will take option 3. + +- **The session keeps round 1's roots.** When a round-1 proof is about to be + sent, the single-use session it opens keeps the nonced root reported for each + pointer leaf, keyed by address: 64 bytes of key and root a pointer, and + nothing for chunks. A session whose proof then fails to send keeps them until + it expires, as it keeps its place today. +- **The store keeps every replaced record, for longer.** Every record an update + replaces is kept in memory, not only the last one, for ten minutes rather + than five. A round 1 can read a pointer at its start and take as long as the + auditor waits for the largest subtree (1,024 leaves), about seven minutes by + default, before its session even opens, and round 2 then has the session's + two minutes. A test ties the ten minutes to those two figures. The overall + cap stays 2,048 records, about 11 MB, oldest first wherever it is. +- **Round 2 serves the record that matches.** Among the record held now and the + replaced records kept for that address, it serves the one whose nonced root, + under the audit's own nonce, is the root round 1 reported. That is a single + record, so the auditor's check and the item cap are unchanged. +- **When it cannot, it says so.** If the root matches nothing kept, because the + record aged out or was evicted, round 2 is rejected as `Transient`: no trust + penalty, and the holder loses the credit this audit would have given it, as + for a local read error. A node that holds nothing at all for the pointer + still reports it absent, which is a confirmed failure, as before. +- **The roots are bounded.** Every live session together keeps at most + `MAX_SESSION_POINTER_BINDINGS` (65,536) roots, 4 MiB of payload before the + maps' own overhead. An honest round 2 follows its round 1 within seconds, so + to make room the oldest sessions give theirs up first. A session left without + roots rejects round 2 as `Transient` for any pointer it opens. It cannot show + that no update came since round 1 read the record: the replaced records kept + are capped, so an empty history proves nothing, and serving the record held + now would be a guess that fails an honest node when it is wrong. + +## Consequences + +### Positive + +- Updates between the rounds no longer fail an honest holder, however many. + What remains are local limits, and each is reported as `Transient`, never as + a confirmed failure: the bound record evicted by more than 2,048 paid updates + across the node's pointers inside ten minutes, or the session's roots given up + because newer sessions filled the budget. +- Round 2 serves one pointer record where it could serve two, so it is smaller. +- The auditor, the wire format and the subtree-audit protocol id are unchanged. + +### Negative / Trade-offs + +- Round-1 sessions carry state they did not before, bounded by the budget above. + An auditor that opens sessions faster than honest ones complete can make an + honest holder's round 2 go `Transient` for any pointer it opens: that costs + the holder the credit of one audit, not trust. +- A responder that returns `Transient` is not proved wrong. That was already + so, since any responder can report a local read error, so this gives a + dishonest node no answer it did not have. +- One address can hold many replaced records inside the window, and round 2 + hashes each candidate it checks for that address. The global cap bounds that + work to about 11 MB of keyed BLAKE3 per opened pointer, and filling it takes + that many paid updates. +- Replaced records are kept twice as long, so under a high update rate the + 2,048-record cap is reached sooner. The memory bound itself is unchanged. + +### Neutral / Operational + +- A restart drops every session, as before, so a round 2 that follows one goes + to the graced timeout lane, as it already did. + +## Validation + +- `several_updates_between_the_rounds_do_not_fail_an_honest_holder`: three + updates between the rounds fail with `DigestMismatch` before this change and + pass after it, and round 2 serves exactly the record round 1 read. +- `round_two_serves_the_record_round_one_read_across_several_updates` (e2e): + both rounds sent over QUIC to a live node, with three updates between them. + It fails if the node stops handing round 1's roots to its session. +- `a_bound_record_no_longer_held_is_unavailable_not_failed`, + `without_the_bound_root_a_pointer_is_unavailable_not_failed`, + `subtree_session_carries_pointer_bindings_within_the_budget`, + `past_the_cap_the_oldest_replaced_record_anywhere_goes_first` and + `a_replaced_record_outlives_the_slowest_audit` cover the limits above. +- In production, a pointer holder's `DigestMismatch` rate should not rise with + the update rate of the pointers it holds. A rise in `Transient` round-2 + rejections naming a pointer means the retention or session budget is being + reached. + +## 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 ADR. diff --git a/src/pointer/store.rs b/src/pointer/store.rs index 3628d173..ec15cbf9 100644 --- a/src/pointer/store.rs +++ b/src/pointer/store.rs @@ -45,7 +45,9 @@ //! lock file under `{root}/pointers/` keeps two processes from keeping two //! indexes over one set of files. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; + +use bytes::Bytes; use std::fs::{File, OpenOptions}; use std::io::{Read, Write}; use std::path::{Path, PathBuf}; @@ -81,16 +83,26 @@ const SHARD_COUNT: u16 = 256; /// /// A storage audit binds the record a node holds in its first round and asks /// for it in the second (ADR-0016). An owner updating the pointer between the -/// two would otherwise fail the honest node that took the update, so the -/// record the update replaced stays servable for longer than an audit session -/// lives. -pub const SUPERSEDED_RETENTION: Duration = Duration::from_mins(5); - -/// Most replaced records kept at once, about 11 MB at the cap. Past it the -/// oldest goes first; reaching it inside [`SUPERSEDED_RETENTION`] takes that -/// many paid updates to pointers this node holds. +/// two would otherwise fail the honest node that took the update, so every +/// record an update replaces stays servable, however many updates follow it +/// (ADR-0017), for longer than the slowest audit can take: a round 1 over the +/// largest subtree an auditor will wait for, then the session its round 2 +/// must arrive within. Round 1 can read a pointer at its very start and take +/// that long to finish, so the time counts from the read, not from the +/// session. +pub const SUPERSEDED_RETENTION: Duration = Duration::from_mins(10); + +/// Most replaced records kept at once, across every address, about 11 MB at +/// the cap. Past it the oldest goes first; reaching it inside +/// [`SUPERSEDED_RETENTION`] takes that many paid updates to pointers this +/// node holds. const MAX_SUPERSEDED: usize = 2048; +/// Every record a recent update replaced, by address, oldest first, each with +/// when it was replaced (see [`SUPERSEDED_RETENTION`]). Held as [`Bytes`] so +/// handing them out copies nothing. +type Superseded = HashMap>; + /// The name of the shard directory `address` lives in: its last byte in hex. fn shard_name(address: &XorName) -> String { let last = address.last().copied().unwrap_or_default(); @@ -258,9 +270,8 @@ struct Inner { generation: AtomicU64, /// What the store has done, for telemetry. counters: Counters, - /// The record each recent update replaced, by address, with when (see - /// [`SUPERSEDED_RETENTION`]). - superseded: Mutex)>>, + /// Every record a recent update replaced (see [`Superseded`]). + superseded: Mutex, /// Held for the store's lifetime; releasing it releases the directory. _lock_file: File, } @@ -519,17 +530,23 @@ impl PointerStore { .map(|entry| entry.state.state_id) } - /// The record an update at `address` replaced within the last - /// [`SUPERSEDED_RETENTION`], if any: what a storage audit that bound it - /// before the update is still owed. + /// Every record updates at `address` replaced within the last + /// [`SUPERSEDED_RETENTION`], newest first: what a storage audit that bound + /// one of them before the updates is still owed. #[must_use] - pub fn superseded(&self, address: &XorName) -> Option> { + pub fn superseded(&self, address: &XorName) -> Vec { self.inner .superseded .lock() .get(address) - .filter(|(at, _)| at.elapsed() < SUPERSEDED_RETENTION) - .map(|(_, bytes)| bytes.clone()) + .map(|kept| { + kept.iter() + .rev() + .filter(|(at, _)| at.elapsed() < SUPERSEDED_RETENTION) + .map(|(_, bytes)| bytes.clone()) + .collect() + }) + .unwrap_or_default() } /// The bytes of the record held at `address`, read from disk without @@ -738,23 +755,39 @@ impl PointerStore { } impl Inner { - /// Keep `bytes` as the record just replaced at `address`, dropping what - /// has aged out and, past the cap, the oldest. + /// Keep `bytes` beside whatever else was recently replaced at `address`, + /// dropping what has aged out and, past the cap, the oldest anywhere. fn keep_superseded(&self, address: XorName, bytes: Vec) { let now = Instant::now(); let mut superseded = self.superseded.lock(); - superseded.retain(|_, (at, _)| now.duration_since(*at) < SUPERSEDED_RETENTION); - while superseded.len() >= MAX_SUPERSEDED { + superseded.retain(|_, kept| { + kept.retain(|(at, _)| now.duration_since(*at) < SUPERSEDED_RETENTION); + !kept.is_empty() + }); + let mut held: usize = superseded.values().map(VecDeque::len).sum(); + while held >= MAX_SUPERSEDED { + // Each address keeps its records oldest first, so the oldest + // anywhere is the first of one of them. let Some(oldest) = superseded .iter() - .min_by_key(|(_, (at, _))| *at) - .map(|(address, _)| *address) + .filter_map(|(address, kept)| kept.front().map(|(at, _)| (*at, *address))) + .min() + .map(|(_, address)| address) else { break; }; - superseded.remove(&oldest); + if let Some(kept) = superseded.get_mut(&oldest) { + kept.pop_front(); + if kept.is_empty() { + superseded.remove(&oldest); + } + } + held = held.saturating_sub(1); } - superseded.insert(address, (now, bytes)); + superseded + .entry(address) + .or_default() + .push_back((now, Bytes::from(bytes))); } /// Remove the file and the index entry for `address` under one lock, so a @@ -1276,9 +1309,8 @@ mod tests { let (store, _dir) = store().await; let first = signed(1, 1, 1); store.put_bytes(&first.to_bytes()).await.expect("put"); - assert_eq!( - store.superseded(&first.address()), - None, + assert!( + store.superseded(&first.address()).is_empty(), "a creation replaces nothing" ); @@ -1289,7 +1321,7 @@ mod tests { ); assert_eq!( store.superseded(&first.address()), - Some(first.to_bytes()), + vec![Bytes::from(first.to_bytes())], "the replaced record, exactly as it was held" ); assert_eq!( @@ -1300,7 +1332,55 @@ mod tests { // A stale arrival replaces nothing, so it keeps nothing. store.put_bytes(&first.to_bytes()).await.expect("put"); - assert_eq!(store.superseded(&first.address()), Some(first.to_bytes())); + assert_eq!( + store.superseded(&first.address()), + vec![Bytes::from(first.to_bytes())] + ); + + // A later update keeps the one before it too, newest first: an audit + // may have bound either. + let third = signed(1, 3, 3); + store.put_bytes(&third.to_bytes()).await.expect("put"); + assert_eq!( + store.superseded(&first.address()), + vec![ + Bytes::from(second.to_bytes()), + Bytes::from(first.to_bytes()) + ] + ); + } + + /// Past the cap the oldest record kept goes first, whichever address it + /// is for, and an address whose last record goes is forgotten. + #[tokio::test] + async fn past_the_cap_the_oldest_replaced_record_anywhere_goes_first() { + let (store, _dir) = store().await; + let (first, second) = ([1u8; 32], [2u8; 32]); + store.inner.keep_superseded(first, vec![1]); + for i in 0..MAX_SUPERSEDED - 1 { + store + .inner + .keep_superseded(second, (i as u64).to_le_bytes().to_vec()); + } + assert_eq!(store.superseded(&first), vec![Bytes::from(vec![1])]); + assert_eq!(store.superseded(&second).len(), MAX_SUPERSEDED - 1); + + // One more: the first address held the oldest record, so it goes. + store.inner.keep_superseded(second, vec![0xFF]); + assert!(store.superseded(&first).is_empty()); + assert!(!store.inner.superseded.lock().contains_key(&first)); + assert_eq!(store.superseded(&second).len(), MAX_SUPERSEDED); + + // And the next goes from the front of the second address's history. + store.inner.keep_superseded(first, vec![2]); + let kept = store.superseded(&second); + assert_eq!(kept.len(), MAX_SUPERSEDED - 1); + assert_eq!(kept.first(), Some(&Bytes::from(vec![0xFF])), "newest first"); + assert_eq!( + kept.last(), + Some(&Bytes::from(1u64.to_le_bytes().to_vec())), + "the oldest of them went" + ); } #[tokio::test] diff --git a/src/replication/config.rs b/src/replication/config.rs index 276ba1bb..a6722305 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -258,6 +258,18 @@ pub const SUBTREE_SESSION_TTL: Duration = Duration::from_mins(2); /// peers open sessions; oldest are evicted past this). pub const MAX_SUBTREE_SESSIONS: usize = 4 * MAX_CONCURRENT_SUBTREE_ROUND1 * 256; +/// Most pointer bindings every live round-1 session holds together +/// (ADR-0017): 4 MiB of keys and roots at the cap, before the maps' own +/// overhead. +/// +/// A session keeps one per pointer its round 1 proved, and a round-1 subtree +/// can hold about a thousand leaves, so [`MAX_SUBTREE_SESSIONS`] full sessions +/// would otherwise hold two million. Past the cap the oldest sessions give +/// theirs up first. A session without them still opens, and its round 2 +/// reports each pointer it opens as a transient failure rather than guessing, +/// so the cap bounds memory and never becomes a confirmed failure. +pub const MAX_SESSION_POINTER_BINDINGS: usize = 1 << 16; + /// Sustained rate at which the responder-wide round-1 work budget refills, in /// bytes of chunk content per second. /// diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 73d91f45..a33f4005 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -82,9 +82,9 @@ use crate::replication::config::{ max_parallel_fetch, storage_admission_width, ReplicationConfig, MAX_AUDIT_RESPONSES_PER_PEER, MAX_CONCURRENT_AUDIT_RESPONSES, MAX_CONCURRENT_REPLICATION_SENDS, MAX_DIGEST_AUDIT_RESPONSES_PER_PEER, MAX_INCOMING_VERIFICATION_KEYS, - MAX_SUBTREE_ROUND1_PER_PEER, MAX_SUBTREE_SESSIONS, MAX_VERIFICATION_KEYS_PER_CYCLE, - REPLICATION_PROTOCOL_ID, SUBTREE_AUDIT_PROTOCOL_ID, SUBTREE_ROUND1_WORK_BURST_BYTES, - SUBTREE_ROUND1_WORK_REFILL_BYTES_PER_SEC, SUBTREE_SESSION_TTL, + MAX_SESSION_POINTER_BINDINGS, MAX_SUBTREE_ROUND1_PER_PEER, MAX_SUBTREE_SESSIONS, + MAX_VERIFICATION_KEYS_PER_CYCLE, REPLICATION_PROTOCOL_ID, SUBTREE_AUDIT_PROTOCOL_ID, + SUBTREE_ROUND1_WORK_BURST_BYTES, SUBTREE_ROUND1_WORK_REFILL_BYTES_PER_SEC, SUBTREE_SESSION_TTL, }; use crate::replication::paid_list::PaidList; use crate::replication::protocol::{ @@ -94,6 +94,7 @@ use crate::replication::protocol::{ use crate::replication::quorum::KeyVerificationOutcome; use crate::replication::recent_provers::RecentProvers; use crate::replication::scheduling::{CapacityDisplacement, DeferralOutcome, ReplicationQueues}; +use crate::replication::storage_commitment_audit::PointerBindings; use crate::replication::types::{ AuditFailureReason, BootstrapClaimObservation, BootstrapState, FailureEvidence, NeighborSyncState, PeerSyncRecord, PresenceEvidence, RepairProofs, VerificationEntry, @@ -4624,6 +4625,9 @@ struct SubtreeSession { commitment_hash: [u8; 32], nonce: [u8; 32], inserted: Instant, + /// What round 1 bound for each pointer it proved, so round 2 serves that + /// record however many updates land in between (ADR-0017). + pointer_bindings: PointerBindings, } /// Responder-wide token bucket over the chunk bytes round-1 proof building may @@ -4787,12 +4791,21 @@ impl SubtreeRound1Limiter { /// Record a single-use session once a round-1 proof is built and about to be /// sent, so the matching round 2 is admitted exactly once. + /// + /// The session keeps what round 1 bound for each pointer it proved, while + /// every live session together holds no more than + /// [`MAX_SESSION_POINTER_BINDINGS`] of them. An honest round 2 follows its + /// round 1 within seconds, so to make room the oldest sessions give theirs + /// up first. A session left without them still opens, and its round 2 + /// reports each pointer it opens as a transient failure rather than + /// guessing which record round 1 read (ADR-0017). async fn open_session( &self, source: PeerId, challenge_id: u64, commitment_hash: [u8; 32], nonce: [u8; 32], + pointer_bindings: PointerBindings, ) { let now = Instant::now(); let mut sessions = self.sessions.write().await; @@ -4806,27 +4819,55 @@ impl SubtreeRound1Limiter { sessions.remove(&oldest); } } + let mut held: usize = sessions.values().map(|e| e.pointer_bindings.len()).sum(); + if pointer_bindings.len() <= MAX_SESSION_POINTER_BINDINGS + && held.saturating_add(pointer_bindings.len()) > MAX_SESSION_POINTER_BINDINGS + { + let mut oldest_first: Vec<_> = sessions + .iter() + .filter(|(_, e)| !e.pointer_bindings.is_empty()) + .map(|(k, e)| (e.inserted, *k)) + .collect(); + oldest_first.sort_unstable(); + for (_, k) in oldest_first { + if held.saturating_add(pointer_bindings.len()) <= MAX_SESSION_POINTER_BINDINGS { + break; + } + if let Some(e) = sessions.get_mut(&k) { + held = held.saturating_sub(e.pointer_bindings.len()); + e.pointer_bindings = PointerBindings::new(); + } + } + } + let pointer_bindings = + if held.saturating_add(pointer_bindings.len()) > MAX_SESSION_POINTER_BINDINGS { + PointerBindings::new() + } else { + pointer_bindings + }; sessions.insert( (source, challenge_id), SubtreeSession { commitment_hash, nonce, inserted: now, + pointer_bindings, }, ); } - /// Atomically consume the round-2 session for this exchange. `true` iff a + /// Atomically consume the round-2 session for this exchange. `Some` iff a /// live session matching `(source, challenge_id, commitment_hash, nonce)` - /// existed (and is now removed); a miss silently drops round 2 to the graced - /// timeout lane (sessions are ephemeral and can be lost across a restart). + /// existed (and is now removed), carrying what its round 1 bound for each + /// pointer; a miss silently drops round 2 to the graced timeout lane + /// (sessions are ephemeral and can be lost across a restart). async fn consume_session( &self, source: &PeerId, challenge_id: u64, commitment_hash: &[u8; 32], nonce: &[u8; 32], - ) -> bool { + ) -> Option { let mut sessions = self.sessions.write().await; let matches = sessions.get(&(*source, challenge_id)).is_some_and(|e| { Instant::now().duration_since(e.inserted) < SUBTREE_SESSION_TTL @@ -4834,17 +4875,21 @@ impl SubtreeRound1Limiter { && &e.nonce == nonce }); if matches { - sessions.remove(&(*source, challenge_id)); + sessions + .remove(&(*source, challenge_id)) + .map(|e| e.pointer_bindings) + } else { + None } - matches } } /// Outcome of admitting a round-2 slice challenge. enum SliceAdmission { /// Admitted: the guard holds the global permit and the per-peer slot, and - /// the single-use round-1 session has been consumed. - Admitted(AuditResponderGuard), + /// the single-use round-1 session has been consumed, yielding what its + /// round 1 bound for each pointer. + Admitted(AuditResponderGuard, PointerBindings), /// Refused at a responder ceiling. The round-1 session is left INTACT. Capacity(AuditResponderAdmissionFailure), /// No live round-1 session matched this challenge. @@ -4885,7 +4930,7 @@ async fn admit_slice_challenge( Ok(guard) => guard, Err(failure) => return SliceAdmission::Capacity(failure), }; - if !round1 + let Some(pointer_bindings) = round1 .consume_session( source, challenge.challenge_id, @@ -4893,13 +4938,13 @@ async fn admit_slice_challenge( &challenge.nonce, ) .await - { + else { // Release the permit and per-peer slot before the caller replies: no // chunk work follows, so holding them would shrink the pool for nothing. drop(guard); return SliceAdmission::NoSession; - } - SliceAdmission::Admitted(guard) + }; + SliceAdmission::Admitted(guard, pointer_bindings) } /// Try to admit one audit-responder task for `source`: take a global permit AND @@ -5242,6 +5287,7 @@ async fn handle_replication_message( challenge.challenge_id, challenge.expected_commitment_hash, challenge.nonce, + storage_commitment_audit::pointer_bindings(&response), ) .await; } @@ -5286,7 +5332,7 @@ async fn handle_replication_message( "Audit challenge received: kind=slice source={source} request_response={}", rr_message_id.is_some(), ); - let guard = match admit_slice_challenge( + let (guard, pointer_bindings) = match admit_slice_challenge( &ctx.audit_responder_semaphore, &ctx.audit_responder_inflight, &ctx.subtree_round1, @@ -5295,7 +5341,7 @@ async fn handle_replication_message( ) .await { - SliceAdmission::Admitted(guard) => guard, + SliceAdmission::Admitted(guard, pointer_bindings) => (guard, pointer_bindings), SliceAdmission::Capacity(failure) => { protocol::record_audit_drop(protocol::AuditDropKind::Slice); audit_metrics::record_admission_drop(class); @@ -5395,6 +5441,7 @@ async fn handle_replication_message( &challenge, &storage, pointer_store.as_ref(), + &pointer_bindings, p2p_node.peer_id(), bootstrapping, Some(&my_commitment_state), @@ -10723,15 +10770,94 @@ mod tests { // Session: opened by round 1, consumed exactly once by the matching round 2. let hash = [7u8; 32]; let nonce = [9u8; 32]; - limiter.open_session(peer, 42, hash, nonce).await; + limiter + .open_session(peer, 42, hash, nonce, PointerBindings::new()) + .await; // Wrong nonce / commitment does not match. - assert!(!limiter.consume_session(&peer, 42, &hash, &[0u8; 32]).await); - assert!(!limiter.consume_session(&peer, 42, &[0u8; 32], &nonce).await); + assert!(limiter + .consume_session(&peer, 42, &hash, &[0u8; 32]) + .await + .is_none()); + assert!(limiter + .consume_session(&peer, 42, &[0u8; 32], &nonce) + .await + .is_none()); // A round 2 with no prior round 1 (wrong challenge_id) misses. - assert!(!limiter.consume_session(&peer, 99, &hash, &nonce).await); + assert!(limiter + .consume_session(&peer, 99, &hash, &nonce) + .await + .is_none()); // The matching round 2 consumes it — and only once (single-use). - assert!(limiter.consume_session(&peer, 42, &hash, &nonce).await); - assert!(!limiter.consume_session(&peer, 42, &hash, &nonce).await); + assert!(limiter + .consume_session(&peer, 42, &hash, &nonce) + .await + .is_some()); + assert!(limiter + .consume_session(&peer, 42, &hash, &nonce) + .await + .is_none()); + } + + // A session carries what its round 1 bound for each pointer to round 2, + // and every live session together stays under the binding budget: to make + // room the oldest sessions give theirs up, and a session larger than the + // whole budget keeps none rather than being refused. + #[tokio::test(start_paused = true)] + async fn subtree_session_carries_pointer_bindings_within_the_budget() { + let limiter = SubtreeRound1Limiter::new(Duration::ZERO, 1); + let (hash, nonce) = ([1u8; 32], [2u8; 32]); + let bindings = |from: u64, count: usize| -> PointerBindings { + (from..) + .take(count) + .map(|i| { + let mut key = [0u8; 32]; + key[..8].copy_from_slice(&i.to_le_bytes()); + (key, [0xAB; 32]) + }) + .collect() + }; + let open = |peer: u8, id: u64, kept: PointerBindings| { + let limiter = limiter.clone(); + async move { + limiter + .open_session(test_peer(peer), id, hash, nonce, kept) + .await; + // Sessions are ordered by when they opened. + tokio::time::advance(Duration::from_millis(1)).await; + } + }; + let kept = |peer: u8, id: u64| { + let limiter = limiter.clone(); + async move { + limiter + .consume_session(&test_peer(peer), id, &hash, &nonce) + .await + .map(|b| b.len()) + } + }; + + let half = MAX_SESSION_POINTER_BINDINGS / 2; + let first = bindings(0, half); + open(1, 1, first.clone()).await; + open(2, 2, bindings(1 << 40, half)).await; + // The budget is full. The next session takes the oldest one's room. + open(3, 3, bindings(1 << 41, 1)).await; + // One larger than the whole budget keeps nothing, and costs no one. + open(4, 4, bindings(1 << 42, MAX_SESSION_POINTER_BINDINGS + 1)).await; + + assert_eq!(kept(1, 1).await, Some(0), "the oldest gave its bindings up"); + assert_eq!(kept(2, 2).await, Some(half)); + assert_eq!(kept(3, 3).await, Some(1)); + assert_eq!(kept(4, 4).await, Some(0), "still opened, with none kept"); + + // With room, round 2 gets exactly what round 1 bound. + open(5, 5, first.clone()).await; + assert_eq!( + limiter + .consume_session(&test_peer(5), 5, &hash, &nonce) + .await, + Some(first) + ); } // The concurrency pool and the per-peer cooldown are both keyed by peer id, @@ -10945,7 +11071,9 @@ mod tests { let (id, hash, nonce) = (77u64, [3u8; 32], [4u8; 32]); let challenge = slice_challenge(id, hash, nonce); - round1.open_session(peer, id, hash, nonce).await; + round1 + .open_session(peer, id, hash, nonce, PointerBindings::new()) + .await; // Saturate this peer's share so the next admission must be refused. let mut hold = Vec::new(); @@ -10970,7 +11098,7 @@ mod tests { let retried = admit_slice_challenge(&semaphore, &inflight, &round1, &peer, &challenge).await; assert!( - matches!(retried, SliceAdmission::Admitted(_)), + matches!(retried, SliceAdmission::Admitted(..)), "the round-1 session must survive a capacity refusal so the retry succeeds" ); diff --git a/src/replication/protocol.rs b/src/replication/protocol.rs index b2ca7962..4fa80bd9 100644 --- a/src/replication/protocol.rs +++ b/src/replication/protocol.rs @@ -1374,16 +1374,18 @@ pub enum SubtreeSliceItem { PointerRecord { /// The requested key: the pointer's address. key: XorName, - /// The record held now, and the one an update replaced since round 1 - /// if there was one, each in its canonical encoding. At most - /// [`MAX_POINTER_RECORDS_PER_ITEM`]: round 1 bound one of them, and - /// the responder cannot tell which without keeping round 1's answer. + /// The record round 1 read, in its canonical encoding, found by the + /// nonced root round 1 reported over it (ADR-0017). At most + /// [`MAX_POINTER_RECORDS_PER_ITEM`], and the auditor accepts whichever + /// reproduces that root. records: Vec>, }, } -/// Most records one [`SubtreeSliceItem::PointerRecord`] may carry: the one -/// held now and the one it replaced. +/// Most records one [`SubtreeSliceItem::PointerRecord`] may carry. +/// +/// A responder that keeps what round 1 bound serves one. Before it did, it +/// served the record held now and the one an update last replaced (ADR-0016). pub const MAX_POINTER_RECORDS_PER_ITEM: usize = 2; /// Response to a [`SubtreeSliceChallenge`] (round 2). diff --git a/src/replication/storage_commitment_audit.rs b/src/replication/storage_commitment_audit.rs index c20b8e60..2b47b853 100644 --- a/src/replication/storage_commitment_audit.rs +++ b/src/replication/storage_commitment_audit.rs @@ -14,6 +14,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use crate::logging::{debug, info, warn}; +use bytes::Bytes; use rand::Rng; use crate::ant_protocol::XorName; @@ -748,6 +749,27 @@ const _: () = assert!( "a replaced pointer record must outlive the audit session that may be owed it" ); +/// What round 1 bound for each committed pointer it proved, by address. +/// +/// Each is the nonced root round 1 reported over the record it read +/// (ADR-0017). Round 2 is owed that record, whatever the pointer holds by the +/// time it asks. +pub type PointerBindings = HashMap; + +/// The pointer bindings a round-1 response reports. Only a proof reports any. +#[must_use] +pub fn pointer_bindings(response: &SubtreeAuditResponse) -> PointerBindings { + match response { + SubtreeAuditResponse::Proof { proof, .. } => proof + .leaves + .iter() + .filter(|leaf| is_pointer_leaf(leaf)) + .map(|leaf| (leaf.key, leaf.nonced_root)) + .collect(), + _ => PointerBindings::new(), + } +} + /// Whether a round-1 leaf commits a pointer (ADR-0016): committed under /// [`pointer_leaf_hash`] of its key, at the fixed record length. fn is_pointer_leaf(leaf: &SubtreeLeaf) -> bool { @@ -977,7 +999,7 @@ pub(crate) fn verify_slice_response( // belong at the committed address, and be the record round 1 bound its // nonced root over, which the responder had to read before it knew what // would be sampled. An update between the rounds does not fail an - // honest holder: it serves the record it held then beside the new one. + // honest holder: it serves the record round 1 read (ADR-0017). if is_pointer_leaf(leaf) { if let Err(reason) = verify_pointer_item(nonce, challenged_peer_bytes, leaf, items) { return AuditVerdict::Fail(reason); @@ -1733,6 +1755,7 @@ pub async fn handle_subtree_slice_challenge( challenge, storage, None, + &PointerBindings::new(), self_peer_id, is_bootstrapping, commitment_state, @@ -1741,12 +1764,14 @@ pub async fn handle_subtree_slice_challenge( } /// [`handle_subtree_slice_challenge`] for a node that also commits pointers -/// (ADR-0016): a committed pointer is answered with its whole signed record. -#[allow(clippy::too_many_lines)] +/// (ADR-0016): a committed pointer is answered with its whole signed record, +/// the one `bound` says round 1 read (ADR-0017). +#[allow(clippy::too_many_lines, clippy::too_many_arguments)] pub async fn handle_subtree_slice_challenge_with_pointers( challenge: &SubtreeSliceChallenge, storage: &ChunkStore, pointers: Option<&PointerStore>, + bound: &PointerBindings, self_peer_id: &PeerId, is_bootstrapping: bool, commitment_state: Option<&Arc>, @@ -1882,10 +1907,10 @@ pub async fn handle_subtree_slice_challenge_with_pointers( for key in key_order { let indices = indices_by_key.remove(&key).unwrap_or_default(); if built.tree().commits_pointer(&key) { - // The bytes held now, exactly as round 1 read them, and the record - // an update replaced since, if any: round 1 bound one of the two. - // The auditor verifies whichever it checks, so nothing is verified - // here. + // The record round 1 read, found by the root it reported over it + // among what is held now and every record updates have replaced + // since. The auditor verifies whatever is served, so nothing is + // verified here. let Some(store) = pointers else { items.push(SubtreeSliceItem::Absent { key }); continue; @@ -1900,11 +1925,30 @@ pub async fn handle_subtree_slice_challenge_with_pointers( } } }; - let records: Vec> = current.into_iter().chain(store.superseded(&key)).collect(); - if records.is_empty() { - items.push(SubtreeSliceItem::Absent { key }); - } else { - items.push(SubtreeSliceItem::PointerRecord { key, records }); + match pointer_records( + challenge, + &key, + bound.get(&key), + current, + store.superseded(&key), + ) { + PointerServe::Record(record) => { + items.push(SubtreeSliceItem::PointerRecord { + key, + records: vec![record], + }); + } + PointerServe::Absent => items.push(SubtreeSliceItem::Absent { key }), + PointerServe::Unavailable => { + return SubtreeSliceResponse::Rejected { + challenge_id: challenge.challenge_id, + kind: RejectKind::Transient, + reason: format!( + "cannot serve the record round 1 read for pointer {}", + hex::encode(key) + ), + } + } } continue; } @@ -1921,6 +1965,54 @@ pub async fn handle_subtree_slice_challenge_with_pointers( } } +/// What round 2 serves for a committed pointer. +enum PointerServe { + /// The record round 1 read. + Record(Vec), + /// Nothing at all is held for it, which is a lost pointer. + Absent, + /// The pointer is held, but this node cannot serve the record round 1 + /// read: it has aged out or been evicted to keep memory bounded, or the + /// session gave round 1's root up to stay in budget. A local limit, not a + /// lost pointer, so it is reported as one rather than proved wrong. + Unavailable, +} + +/// Choose what round 2 serves for the pointer at `key` (ADR-0017), from the +/// record held now and the records updates replaced: the one that reproduces +/// the root round 1 reported. +/// +/// Without that root nothing is served for a pointer still held. Neither the +/// record held now nor the replaced records kept can show that no update came +/// since round 1 read it, since the kept ones are capped, so serving one would +/// be a guess that fails the node when it is wrong. +fn pointer_records( + challenge: &SubtreeSliceChallenge, + key: &XorName, + bound: Option<&[u8; 32]>, + current: Option>, + replaced: Vec, +) -> PointerServe { + if current.is_none() && replaced.is_empty() { + return PointerServe::Absent; + } + let Some(root) = bound else { + return PointerServe::Unavailable; + }; + let reproduces = |record: &[u8]| { + nonced_block_root(&challenge.nonce, &challenge.challenged_peer_id, key, record) == *root + }; + if let Some(current) = current.filter(|record| reproduces(record)) { + return PointerServe::Record(current); + } + replaced + .into_iter() + .find(|record| reproduces(record)) + .map_or(PointerServe::Unavailable, |record| { + PointerServe::Record(Vec::from(record)) + }) +} + /// Outcome of serving all requested openings for one committed key. enum KeyServe { /// Openings built for this key; append to the response. @@ -2749,7 +2841,9 @@ mod tests { mod pointer_audit_tests { use super::*; use crate::replication::commitment::MerkleTree; + use crate::replication::commitment::MAX_COMMITMENT_KEY_COUNT; use crate::replication::commitment_state::BuiltCommitment; + use crate::replication::subtree::max_subtree_leaves; use crate::storage::ChunkStoreConfig; use ant_protocol::pointer::{PointerTarget, PointerTargetKind}; use saorsa_pqc::api::sig::ml_dsa_65; @@ -2849,11 +2943,40 @@ mod pointer_audit_tests { .response } + /// Round 2 as the engine serves it: with what round 1 bound for each + /// pointer it opens, as the live session carries it. async fn round2( &self, nonce: [u8; 32], openings: &[(SubtreeLeaf, u32)], ) -> Vec { + let bound = openings + .iter() + .map(|(leaf, _)| leaf) + .filter(|leaf| is_pointer_leaf(leaf)) + .map(|leaf| (leaf.key, leaf.nonced_root)) + .collect(); + self.round2_bound(nonce, openings, &bound).await + } + + async fn round2_bound( + &self, + nonce: [u8; 32], + openings: &[(SubtreeLeaf, u32)], + bound: &PointerBindings, + ) -> Vec { + match self.round2_response(nonce, openings, bound).await { + SubtreeSliceResponse::Items { items, .. } => items, + other => panic!("expected items, got {other:?}"), + } + } + + async fn round2_response( + &self, + nonce: [u8; 32], + openings: &[(SubtreeLeaf, u32)], + bound: &PointerBindings, + ) -> SubtreeSliceResponse { let challenge = SubtreeSliceChallenge { challenge_id: CHALLENGE_ID, nonce, @@ -2867,19 +2990,16 @@ mod pointer_audit_tests { }) .collect(), }; - match handle_subtree_slice_challenge_with_pointers( + handle_subtree_slice_challenge_with_pointers( &challenge, &self.storage, Some(&self.pointers), + bound, &self.peer, false, Some(&self.state), ) .await - { - SubtreeSliceResponse::Items { items, .. } => items, - other => panic!("expected items, got {other:?}"), - } } /// Round 1 as the auditor sees it: the proof, checked against the pin. @@ -2978,6 +3098,128 @@ mod pointer_audit_tests { ); } + /// A record an update replaces is kept for longer than the slowest audit + /// can take: a round 1 over the largest subtree an auditor waits for, + /// which may have read the pointer at its start, then the session its + /// round 2 must arrive within. + #[test] + fn a_replaced_record_outlives_the_slowest_audit() { + let largest = + usize::try_from(max_subtree_leaves(MAX_COMMITMENT_KEY_COUNT)).expect("fits a usize"); + let slowest = + ReplicationConfig::default().audit_response_timeout(largest) + SUBTREE_SESSION_TTL; + assert!( + SUPERSEDED_RETENTION > slowest, + "{SUPERSEDED_RETENTION:?} must outlive {slowest:?}" + ); + } + + /// Round 1 binds each pointer it proves, and nothing else. + #[tokio::test] + async fn round_one_binds_exactly_the_pointers_it_proves() { + let responder = Responder::new(24, 24).await; + let nonce = mixed_nonce(responder.committed().tree()); + let response = responder.round1(nonce).await; + let SubtreeAuditResponse::Proof { proof, .. } = &response else { + panic!("expected a proof, got {response:?}"); + }; + let expected: PointerBindings = proof + .leaves + .iter() + .filter(|leaf| responder.committed().tree().commits_pointer(&leaf.key)) + .map(|leaf| (leaf.key, leaf.nonced_root)) + .collect(); + assert!(!expected.is_empty(), "the subtree holds a pointer"); + assert_eq!(pointer_bindings(&response), expected); + assert!( + pointer_bindings(&SubtreeAuditResponse::Bootstrapping { + challenge_id: CHALLENGE_ID + }) + .is_empty(), + "only a proof binds anything" + ); + } + + /// Without the root round 1 reported, as when its session gave it up to + /// stay in budget, a pointer still held is reported as a transient + /// failure, updated or not: the node cannot show which record round 1 + /// read, and guessing would risk a confirmed failure it did not earn. + #[tokio::test] + async fn without_the_bound_root_a_pointer_is_unavailable_not_failed() { + let responder = Responder::new(24, 24).await; + let nonce = mixed_nonce(responder.committed().tree()); + let openings = openings(&responder.proved_leaves(nonce).await); + let unavailable = |response: SubtreeSliceResponse| { + matches!( + response, + SubtreeSliceResponse::Rejected { + kind: RejectKind::Transient, + .. + } + ) + }; + + assert!(unavailable( + responder + .round2_response(nonce, &openings, &PointerBindings::new()) + .await + )); + + let updated = first_pointer(&openings); + let owner = (0..24u8) + .find(|owner| pointer(*owner, 1).address() == updated) + .expect("the opened pointer is one of ours"); + responder + .pointers + .put_bytes(&pointer(owner, 2).to_bytes()) + .await + .expect("update"); + assert!(unavailable( + responder + .round2_response(nonce, &openings, &PointerBindings::new()) + .await + )); + } + + /// A root round 1 reported that nothing held now reproduces, as when the + /// record it read has been evicted to keep memory bounded, is reported as + /// a transient failure while the pointer is still held, and as absent + /// once nothing at all is. + #[tokio::test] + async fn a_bound_record_no_longer_held_is_unavailable_not_failed() { + let responder = Responder::new(24, 24).await; + let nonce = mixed_nonce(responder.committed().tree()); + let openings = openings(&responder.proved_leaves(nonce).await); + let target = first_pointer(&openings); + let evicted: PointerBindings = openings + .iter() + .map(|(leaf, _)| leaf) + .filter(|leaf| is_pointer_leaf(leaf)) + .map(|leaf| (leaf.key, [0xEE; 32])) + .collect(); + + assert!(matches!( + responder.round2_response(nonce, &openings, &evicted).await, + SubtreeSliceResponse::Rejected { + kind: RejectKind::Transient, + .. + } + )); + + for (leaf, _) in openings.iter().filter(|(leaf, _)| is_pointer_leaf(leaf)) { + responder.pointers.delete(&leaf.key).await.expect("delete"); + } + let items = responder.round2_bound(nonce, &openings, &evicted).await; + assert!(items + .iter() + .any(|item| matches!(item, SubtreeSliceItem::Absent { key } if *key == target))); + assert_eq!( + verify_slice_response(&openings, &nonce, &responder.peer_bytes, &items), + AuditVerdict::Fail(AuditFailureReason::KeyAbsent), + "a pointer lost outright is still a confirmed failure" + ); + } + /// The commitment binds which pointers are held, not their state, so an /// owner updating a pointer mid-audit cannot fail the node holding it. #[tokio::test] @@ -3003,6 +3245,59 @@ mod pointer_audit_tests { )); } + /// However many paid updates land between the rounds, the record round 1 + /// bound is the one round 2 serves: the owner's activity is not the + /// holder's failure. + #[tokio::test] + async fn several_updates_between_the_rounds_do_not_fail_an_honest_holder() { + let responder = Responder::new(24, 24).await; + let nonce = mixed_nonce(responder.committed().tree()); + let openings = openings(&responder.proved_leaves(nonce).await); + let updated = first_pointer(&openings); + + let owner = (0..24u8) + .find(|owner| pointer(*owner, 1).address() == updated) + .expect("the opened pointer is one of ours"); + for counter in 2..=4 { + responder + .pointers + .put_bytes(&pointer(owner, counter).to_bytes()) + .await + .expect("update"); + } + + let items = responder.round2(nonce, &openings).await; + assert!( + matches!( + verify_slice_response(&openings, &nonce, &responder.peer_bytes, &items), + AuditVerdict::Pass { .. } + ), + "three updates between the rounds must not fail the holder, got {:?}", + verify_slice_response(&openings, &nonce, &responder.peer_bytes, &items) + ); + let served = items.iter().find_map(|item| match item { + SubtreeSliceItem::PointerRecord { key, records } if *key == updated => Some(records), + _ => None, + }); + assert_eq!( + served, + Some(&vec![pointer_record_bound_in_round_one(&responder, owner)]), + "exactly the record round 1 read, and nothing else" + ); + } + + /// The record a responder held for `owner`'s pointer before any update: + /// counter 1, as [`Responder::new`] stored it. + fn pointer_record_bound_in_round_one(responder: &Responder, owner: u8) -> Vec { + let address = pointer(owner, 1).address(); + responder + .pointers + .superseded(&address) + .last() + .map(|record| record.to_vec()) + .expect("the first record is kept") + } + #[tokio::test] async fn a_node_that_lost_a_committed_pointer_fails_round_one() { let responder = Responder::new(24, 24).await; @@ -3080,14 +3375,16 @@ mod pointer_audit_tests { async fn a_relay_that_fetches_records_only_in_round_two_fails() { let responder = Responder::new(24, 24).await; let nonce = mixed_nonce(responder.committed().tree()); - let mut leaves = responder.proved_leaves(nonce).await; + let genuine = responder.proved_leaves(nonce).await; + let held = openings(&genuine); // What a relay can say in round 1 without the bytes. + let mut leaves = genuine; for leaf in leaves.iter_mut().filter(|leaf| is_pointer_leaf(leaf)) { leaf.nonced_root = [0u8; 32]; } let openings = openings(&leaves); // And in round 2 it serves the genuine records, fetched on demand. - let items = responder.round2(nonce, &openings).await; + let items = responder.round2(nonce, &held).await; assert_eq!( verify_slice_response(&openings, &nonce, &responder.peer_bytes, &items), AuditVerdict::Fail(AuditFailureReason::DigestMismatch) @@ -3121,8 +3418,8 @@ mod pointer_audit_tests { ); } - /// A pointer item may carry the record held now and the one it replaced, - /// never more. + /// A pointer item may carry at most two records. This build serves one; + /// two is what a responder that kept no round-1 roots served (ADR-0016). #[tokio::test] async fn a_pointer_item_with_more_than_two_records_is_malformed() { let responder = Responder::new(24, 24).await; diff --git a/tests/e2e/pointer_replication.rs b/tests/e2e/pointer_replication.rs index c031d719..5489f38e 100644 --- a/tests/e2e/pointer_replication.rs +++ b/tests/e2e/pointer_replication.rs @@ -17,7 +17,13 @@ use ant_node::pointer::PointerStore; use ant_node::replication::audit::AuditTickResult; use ant_node::replication::commitment::pointer_leaf_hash; use ant_node::replication::commitment_state::{BuiltCommitment, ResponderCommitmentState}; +use ant_node::replication::config::SUBTREE_AUDIT_PROTOCOL_ID; use ant_node::replication::pointer::{PointerFreshWrite, PointerReplication}; +use ant_node::replication::protocol::{ + ReplicationMessage, ReplicationMessageBody, SubtreeAuditChallenge, SubtreeAuditResponse, + SubtreeSliceChallenge, SubtreeSliceItem, SubtreeSliceOpening, SubtreeSliceResponse, +}; +use ant_node::replication::slice::nonced_block_root; use ant_node::ReplicationConfig; use ant_protocol::pointer::{Pointer, PointerState, PointerTarget, PointerTargetKind}; use bytes::Bytes; @@ -33,6 +39,9 @@ const SETTLE: Duration = Duration::from_secs(30); /// How often to look while waiting. const POLL: Duration = Duration::from_millis(200); +/// The id a hand-driven storage audit uses for both of its rounds. +const CHALLENGE_ID: u64 = 0x5EED; + /// A proof the receivers never parse: the state is pre-marked as paid in each /// verifier's cache, so verification answers from the cache. const DUMMY_PROOF: [u8; 64] = [0x01; 64]; @@ -632,13 +641,23 @@ async fn commit_pointers( auditor: usize, count: usize, ) -> Vec { - let holder_node = harness.test_node(holder).expect("holder"); let records: Vec = (0..count) .map(|_| { let (pk, sk) = owner(); signed(&pk, &sk, 1, 1) }) .collect(); + commit_records(harness, holder, auditor, records).await +} + +/// [`commit_pointers`] over records the caller signed. +async fn commit_records( + harness: &TestHarness, + holder: usize, + auditor: usize, + records: Vec, +) -> Vec { + let holder_node = harness.test_node(holder).expect("holder"); for record in &records { store(holder_node) .put_bytes(&record.to_bytes()) @@ -704,6 +723,144 @@ async fn a_node_holding_its_committed_pointers_passes_the_storage_audit() { harness.teardown().await.expect("teardown"); } +/// Several paid updates between the two rounds of a storage audit do not fail +/// the node holding the pointer: round 2 serves the record round 1 read, found +/// by the root round 1 reported over it (ADR-0017). Driven one round at a time +/// against the holder's live engine, so it is the round-1 session that carries +/// what round 1 bound across to round 2. +#[tokio::test] +#[serial] +async fn round_two_serves_the_record_round_one_read_across_several_updates() { + let harness = TestHarness::setup_small().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + let (holder, auditor) = (7, 8); + let owners: Vec<(MlDsaPublicKey, MlDsaSecretKey)> = (0..24).map(|_| owner()).collect(); + let records = owners.iter().map(|(pk, sk)| signed(pk, sk, 1, 1)).collect(); + commit_records(&harness, holder, auditor, records).await; + + let holder_node = harness.test_node(holder).expect("holder"); + let holder_peer = peer(holder_node); + let committed = commitments(holder_node) + .current() + .expect("a current commitment"); + let pointer_keys = committed.pointer_leaf_keys(); + let auditor_p2p = harness + .test_node(auditor) + .expect("auditor") + .p2p_node + .as_ref() + .expect("p2p") + .clone(); + let ask = |body: ReplicationMessageBody| { + let auditor_p2p = Arc::clone(&auditor_p2p); + async move { + let request = ReplicationMessage { + request_id: CHALLENGE_ID, + body, + } + .encode() + .expect("encode"); + let response = auditor_p2p + .send_request( + &holder_peer, + SUBTREE_AUDIT_PROTOCOL_ID, + request, + Duration::from_secs(60), + ) + .await + .expect("a response"); + ReplicationMessage::decode_subtree_audit_response(&response.data) + .expect("decode") + .body + } + }; + + // Round 1: the holder binds the record it holds for each pointer. + let nonce = [0x5A; 32]; + let round1 = ask(ReplicationMessageBody::SubtreeAuditChallenge( + SubtreeAuditChallenge { + challenge_id: CHALLENGE_ID, + nonce, + challenged_peer_id: *holder_peer.as_bytes(), + expected_commitment_hash: committed.hash(), + }, + )) + .await; + let ReplicationMessageBody::SubtreeAuditResponse(SubtreeAuditResponse::Proof { proof, .. }) = + round1 + else { + panic!("expected a round-1 proof, got {round1:?}"); + }; + let opened: Vec<_> = proof + .leaves + .iter() + .filter(|leaf| pointer_keys.contains(&leaf.key)) + .take(5) + .cloned() + .collect(); + assert!(!opened.is_empty(), "round 1 proved a pointer"); + + // Between the rounds, the owner of every pointer about to be opened + // updates it three times, each a state the holder accepts. + let holder_store = store(holder_node); + for leaf in &opened { + let (pk, sk) = owners + .iter() + .find(|(pk, sk)| signed(pk, sk, 1, 1).address() == leaf.key) + .expect("an owner for every committed pointer"); + for counter in 2..=4 { + holder_store + .put_bytes(&signed(pk, sk, counter, 2).to_bytes()) + .await + .expect("update"); + } + } + + // Round 2: each opened pointer is proved by the record round 1 read. + let round2 = ask(ReplicationMessageBody::SubtreeSliceChallenge( + SubtreeSliceChallenge { + challenge_id: CHALLENGE_ID, + nonce, + challenged_peer_id: *holder_peer.as_bytes(), + expected_commitment_hash: committed.hash(), + openings: opened + .iter() + .map(|leaf| SubtreeSliceOpening { + key: leaf.key, + block_index: 0, + }) + .collect(), + }, + )) + .await; + let ReplicationMessageBody::SubtreeSliceResponse(SubtreeSliceResponse::Items { items, .. }) = + round2 + else { + panic!("expected round-2 items, got {round2:?}"); + }; + for leaf in &opened { + let served = items + .iter() + .find_map(|item| match item { + SubtreeSliceItem::PointerRecord { key, records } if *key == leaf.key => { + Some(records.as_slice()) + } + _ => None, + }) + .expect("a record for every opened pointer"); + assert!( + served.iter().any(|record| { + nonced_block_root(&nonce, holder_peer.as_bytes(), &leaf.key, record) + == leaf.nonced_root + && Pointer::from_bytes(record).is_ok_and(|p| p.address() == leaf.key) + }), + "round 2 must serve the record round 1 bound, after three updates" + ); + } + + harness.teardown().await.expect("teardown"); +} + /// A node that dropped the pointers it committed to fails the storage audit, /// exactly as a node that dropped its chunks does. #[tokio::test] From 32255961c45c579de638f8f90cafe8dbe2487d3d Mon Sep 17 00:00:00 2001 From: grumbach Date: Tue, 29 Sep 2026 18:55:02 +0900 Subject: [PATCH 2/2] fix(pointer): withhold round 1 rather than evict pointer bindings Review of the first version found three ways a node could still lose the record an audit was owed. - A flood of pointer-heavy round-1 sessions made older sessions give up their roots to make room, so an honest holder's round 2 went transient. A round 1 whose roots do not fit now withholds its proof, exactly as a round 1 refused for capacity already does, and no live session gives its roots up. - Keeping a replaced record could evict another one before an update that then failed. Eviction now waits for the rename to succeed, and a failed rename takes back the record it kept. - A round 2 could read the new record before the one it replaced was kept. The replaced record is now kept before the rename, under the same lock. ADR-0017 now states what a transient round 2 costs (the auditor forgets the holder's standing for the whole pinned commitment, with no trust penalty) and that the ten-minute retention is sized for the default configuration. --- ...audits-serve-the-record-round-one-bound.md | 55 +++-- src/pointer/store.rs | 113 ++++++--- src/replication/config.rs | 7 +- src/replication/mod.rs | 218 +++++++++++------- src/replication/storage_commitment_audit.rs | 18 +- 5 files changed, 270 insertions(+), 141 deletions(-) diff --git a/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md b/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md index bb9396a4..63298c41 100644 --- a/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md +++ b/docs/adr/ADR-0017-pointer-audits-serve-the-record-round-one-bound.md @@ -74,46 +74,55 @@ We will take option 3. - **The store keeps every replaced record, for longer.** Every record an update replaces is kept in memory, not only the last one, for ten minutes rather than five. A round 1 can read a pointer at its start and take as long as the - auditor waits for the largest subtree (1,024 leaves), about seven minutes by - default, before its session even opens, and round 2 then has the session's - two minutes. A test ties the ten minutes to those two figures. The overall - cap stays 2,048 records, about 11 MB, oldest first wherever it is. + auditor waits for the largest subtree (1,024 leaves), about seven minutes + with the default configuration, before its session even opens, and round 2 + then has the session's two minutes. A test ties the ten minutes to those two + figures for the default configuration; an auditor configured to wait longer + than that can outlast the record. The overall cap stays 2,048 records, about + 11 MB, oldest first wherever it is. - **Round 2 serves the record that matches.** Among the record held now and the replaced records kept for that address, it serves the one whose nonced root, under the audit's own nonce, is the root round 1 reported. That is a single record, so the auditor's check and the item cap are unchanged. - **When it cannot, it says so.** If the root matches nothing kept, because the - record aged out or was evicted, round 2 is rejected as `Transient`: no trust - penalty, and the holder loses the credit this audit would have given it, as - for a local read error. A node that holds nothing at all for the pointer - still reports it absent, which is a confirmed failure, as before. -- **The roots are bounded.** Every live session together keeps at most - `MAX_SESSION_POINTER_BINDINGS` (65,536) roots, 4 MiB of payload before the - maps' own overhead. An honest round 2 follows its round 1 within seconds, so - to make room the oldest sessions give theirs up first. A session left without - roots rejects round 2 as `Transient` for any pointer it opens. It cannot show - that no update came since round 1 read the record: the replaced records kept - are capped, so an empty history proves nothing, and serving the record held - now would be a guess that fails an honest node when it is wrong. + record aged out or was evicted, round 2 is rejected as `Transient`, as for a + local read error. That is the auditor's timeout lane: no trust penalty, but + the auditor forgets the holder's standing as a proven holder of every key + under the pinned commitment, until the holder passes again. A node that + holds nothing at all for the pointer still reports it absent, which is a + confirmed failure, as before. +- **The roots are bounded, by admission.** Every live session together keeps + at most `MAX_SESSION_POINTER_BINDINGS` (65,536) roots, 4 MiB of payload + before the maps' own overhead. A round 1 whose roots would not fit withholds + its proof, exactly as a round 1 refused for capacity does, so the auditor + sees a timeout. Roots are never stripped from a live session to make room: + its round 2 is owed them. A whole session can still be evicted when the + session count reaches `MAX_SUBTREE_SESSIONS`, as before this change, and its + round 2 then goes to the timeout lane. Without a root for a pointer, round 2 + would reject it + as `Transient` rather than guess, since the replaced records kept are capped + and an empty history proves nothing, but a session this node opened always + holds a root for every pointer it proved. ## Consequences ### Positive - Updates between the rounds no longer fail an honest holder, however many. - What remains are local limits, and each is reported as `Transient`, never as - a confirmed failure: the bound record evicted by more than 2,048 paid updates - across the node's pointers inside ten minutes, or the session's roots given up - because newer sessions filled the budget. + What remains are local limits, and none is a confirmed failure: the bound + record evicted by more than 2,048 paid updates across the node's pointers + inside ten minutes is reported as `Transient`, and a round 1 over the roots + budget goes unanswered, as one over the round-1 capacity already does. - Round 2 serves one pointer record where it could serve two, so it is smaller. - The auditor, the wire format and the subtree-audit protocol id are unchanged. ### Negative / Trade-offs - Round-1 sessions carry state they did not before, bounded by the budget above. - An auditor that opens sessions faster than honest ones complete can make an - honest holder's round 2 go `Transient` for any pointer it opens: that costs - the holder the credit of one audit, not trust. + Auditors that open pointer-heavy sessions faster than they complete can fill + it, and later round 1s then go unanswered until it drains. That costs the + holder those audits' credit, not trust, and is the same exposure the round-1 + concurrency and work budgets already have. - A responder that returns `Transient` is not proved wrong. That was already so, since any responder can report a local read error, so this gives a dishonest node no answer it did not have. diff --git a/src/pointer/store.rs b/src/pointer/store.rs index ec15cbf9..2074c683 100644 --- a/src/pointer/store.rs +++ b/src/pointer/store.rs @@ -85,11 +85,11 @@ const SHARD_COUNT: u16 = 256; /// for it in the second (ADR-0016). An owner updating the pointer between the /// two would otherwise fail the honest node that took the update, so every /// record an update replaces stays servable, however many updates follow it -/// (ADR-0017), for longer than the slowest audit can take: a round 1 over the -/// largest subtree an auditor will wait for, then the session its round 2 -/// must arrive within. Round 1 can read a pointer at its very start and take -/// that long to finish, so the time counts from the read, not from the -/// session. +/// (ADR-0017), for longer than the slowest audit takes with the default +/// configuration: a round 1 over the largest subtree an auditor will wait +/// for, then the session its round 2 must arrive within. Round 1 can read a +/// pointer at its very start and take that long to finish, so the time counts +/// from the read, not from the session. pub const SUPERSEDED_RETENTION: Duration = Duration::from_mins(10); /// Most replaced records kept at once, across every address, about 11 MB at @@ -756,7 +756,11 @@ impl PointerStore { impl Inner { /// Keep `bytes` beside whatever else was recently replaced at `address`, - /// dropping what has aged out and, past the cap, the oldest anywhere. + /// dropping what has aged out. + /// + /// Nothing is evicted to make room here: the update that replaced it may + /// yet fail, and a record evicted for an update that never happened would + /// be lost for nothing. [`Self::trim_superseded`] does that once it has. fn keep_superseded(&self, address: XorName, bytes: Vec) { let now = Instant::now(); let mut superseded = self.superseded.lock(); @@ -764,8 +768,17 @@ impl Inner { kept.retain(|(at, _)| now.duration_since(*at) < SUPERSEDED_RETENTION); !kept.is_empty() }); + superseded + .entry(address) + .or_default() + .push_back((now, Bytes::from(bytes))); + } + + /// Past the cap, drop the oldest kept records, wherever they are. + fn trim_superseded(&self) { + let mut superseded = self.superseded.lock(); let mut held: usize = superseded.values().map(VecDeque::len).sum(); - while held >= MAX_SUPERSEDED { + while held > MAX_SUPERSEDED { // Each address keeps its records oldest first, so the oldest // anywhere is the first of one of them. let Some(oldest) = superseded @@ -784,10 +797,20 @@ impl Inner { } held = held.saturating_sub(1); } - superseded - .entry(address) - .or_default() - .push_back((now, Bytes::from(bytes))); + drop(superseded); + } + + /// Take back the record [`Self::keep_superseded`] just kept for `address`, + /// when the update that replaced it did not happen after all. Called under + /// the index lock that kept it, so nothing was kept for `address` since. + fn unkeep_superseded(&self, address: &XorName) { + let mut superseded = self.superseded.lock(); + if let Some(kept) = superseded.get_mut(address) { + kept.pop_back(); + if kept.is_empty() { + superseded.remove(address); + } + } } /// Remove the file and the index entry for `address` under one lock, so a @@ -896,18 +919,26 @@ impl Inner { } // What this replaces, read under the lock so it is the record the - // index names. Kept for an audit that bound it; a record that - // cannot be read is not kept, and an audit owed it fails as it - // would have on the lost file. - let previous = if replacing { - read_record_file(&path).ok().flatten() - } else { - None - }; + // index names. Kept for an audit that bound it, and kept before + // the rename makes the new record visible, so a round 2 reading + // the new record always finds the old one kept beside it. A record + // that cannot be read is not kept, and an audit owed it fails as + // it would have on the lost file. + let mut kept = false; + if replacing { + if let Some(previous) = read_record_file(&path).ok().flatten() { + self.keep_superseded(address, previous); + kept = true; + } + } // The rename is the commit point: nothing fallible happens between // it and the index update, and both are under this one lock. if let Err(e) = rename_with_retry(&temp, &path) { + // Nothing was replaced, so nothing replaced is kept. + if kept { + self.unkeep_superseded(&address); + } if std::fs::remove_file(&temp).is_err() { settle(reservation); } @@ -917,11 +948,11 @@ impl Inner { path.display() ))); } + if kept { + self.trim_superseded(); + } let generation = self.generation.fetch_add(1, Ordering::Relaxed); index.insert(address, IndexEntry::of(record, generation)); - if let Some(previous) = previous { - self.keep_superseded(address, previous); - } // A file is on the disk now. Charge it whether or not this replaced // one: telling those apart would mean trusting an observation taken // before the rename, and that observation can be wrong in the one @@ -1352,27 +1383,53 @@ mod tests { /// Past the cap the oldest record kept goes first, whichever address it /// is for, and an address whose last record goes is forgotten. + /// Keeping a record evicts nothing: an update that then fails takes back + /// what it kept and leaves every other kept record where it was. #[tokio::test] - async fn past_the_cap_the_oldest_replaced_record_anywhere_goes_first() { + async fn a_record_kept_for_an_update_that_fails_costs_no_other_record() { let (store, _dir) = store().await; - let (first, second) = ([1u8; 32], [2u8; 32]); - store.inner.keep_superseded(first, vec![1]); + let (owed, failing) = ([1u8; 32], [2u8; 32]); + store.inner.keep_superseded(owed, vec![1]); for i in 0..MAX_SUPERSEDED - 1 { store .inner - .keep_superseded(second, (i as u64).to_le_bytes().to_vec()); + .keep_superseded(failing, (i as u64).to_le_bytes().to_vec()); + } + // The cap is full. The next update keeps its record first ... + store.inner.keep_superseded(failing, vec![0xFF]); + // ... and its rename fails, so it takes that record back. + store.inner.unkeep_superseded(&failing); + assert_eq!( + store.superseded(&owed), + vec![Bytes::from(vec![1])], + "the record an audit may be owed is still kept" + ); + assert_eq!(store.superseded(&failing).len(), MAX_SUPERSEDED - 1); + } + + #[tokio::test] + async fn past_the_cap_the_oldest_replaced_record_anywhere_goes_first() { + let (store, _dir) = store().await; + let (first, second) = ([1u8; 32], [2u8; 32]); + let keep = |address: XorName, bytes: Vec| { + store.inner.keep_superseded(address, bytes); + store.inner.trim_superseded(); + }; + keep(first, vec![1]); + for i in 0..MAX_SUPERSEDED - 1 { + keep(second, (i as u64).to_le_bytes().to_vec()); } assert_eq!(store.superseded(&first), vec![Bytes::from(vec![1])]); assert_eq!(store.superseded(&second).len(), MAX_SUPERSEDED - 1); // One more: the first address held the oldest record, so it goes. - store.inner.keep_superseded(second, vec![0xFF]); + keep(second, vec![0xFF]); assert!(store.superseded(&first).is_empty()); assert!(!store.inner.superseded.lock().contains_key(&first)); assert_eq!(store.superseded(&second).len(), MAX_SUPERSEDED); // And the next goes from the front of the second address's history. - store.inner.keep_superseded(first, vec![2]); + keep(first, vec![2]); let kept = store.superseded(&second); assert_eq!(kept.len(), MAX_SUPERSEDED - 1); assert_eq!(kept.first(), Some(&Bytes::from(vec![0xFF])), "newest first"); diff --git a/src/replication/config.rs b/src/replication/config.rs index a6722305..53415423 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -264,10 +264,9 @@ pub const MAX_SUBTREE_SESSIONS: usize = 4 * MAX_CONCURRENT_SUBTREE_ROUND1 * 256; /// /// A session keeps one per pointer its round 1 proved, and a round-1 subtree /// can hold about a thousand leaves, so [`MAX_SUBTREE_SESSIONS`] full sessions -/// would otherwise hold two million. Past the cap the oldest sessions give -/// theirs up first. A session without them still opens, and its round 2 -/// reports each pointer it opens as a transient failure rather than guessing, -/// so the cap bounds memory and never becomes a confirmed failure. +/// would otherwise hold two million. A round 1 whose bindings would not fit +/// withholds its proof, as a round 1 refused for capacity does, and no +/// session already answered gives its bindings up. pub const MAX_SESSION_POINTER_BINDINGS: usize = 1 << 16; /// Sustained rate at which the responder-wide round-1 work budget refills, in diff --git a/src/replication/mod.rs b/src/replication/mod.rs index a33f4005..c373b9fa 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -4794,11 +4794,10 @@ impl SubtreeRound1Limiter { /// /// The session keeps what round 1 bound for each pointer it proved, while /// every live session together holds no more than - /// [`MAX_SESSION_POINTER_BINDINGS`] of them. An honest round 2 follows its - /// round 1 within seconds, so to make room the oldest sessions give theirs - /// up first. A session left without them still opens, and its round 2 - /// reports each pointer it opens as a transient failure rather than - /// guessing which record round 1 read (ADR-0017). + /// [`MAX_SESSION_POINTER_BINDINGS`] of them. A session whose bindings do + /// not fit is not opened and `false` is returned, so its proof is not + /// sent: the bindings of sessions already answered are never given up, + /// because their round 2 is owed them (ADR-0017). async fn open_session( &self, source: PeerId, @@ -4806,10 +4805,16 @@ impl SubtreeRound1Limiter { commitment_hash: [u8; 32], nonce: [u8; 32], pointer_bindings: PointerBindings, - ) { + ) -> bool { let now = Instant::now(); let mut sessions = self.sessions.write().await; sessions.retain(|_, e| now.duration_since(e.inserted) < SUBTREE_SESSION_TTL); + // Checked before anything is evicted, so a session refused here costs + // no other session its place. + let held: usize = sessions.values().map(|e| e.pointer_bindings.len()).sum(); + if held.saturating_add(pointer_bindings.len()) > MAX_SESSION_POINTER_BINDINGS { + return false; + } if sessions.len() >= MAX_SUBTREE_SESSIONS { if let Some(oldest) = sessions .iter() @@ -4819,32 +4824,6 @@ impl SubtreeRound1Limiter { sessions.remove(&oldest); } } - let mut held: usize = sessions.values().map(|e| e.pointer_bindings.len()).sum(); - if pointer_bindings.len() <= MAX_SESSION_POINTER_BINDINGS - && held.saturating_add(pointer_bindings.len()) > MAX_SESSION_POINTER_BINDINGS - { - let mut oldest_first: Vec<_> = sessions - .iter() - .filter(|(_, e)| !e.pointer_bindings.is_empty()) - .map(|(k, e)| (e.inserted, *k)) - .collect(); - oldest_first.sort_unstable(); - for (_, k) in oldest_first { - if held.saturating_add(pointer_bindings.len()) <= MAX_SESSION_POINTER_BINDINGS { - break; - } - if let Some(e) = sessions.get_mut(&k) { - held = held.saturating_sub(e.pointer_bindings.len()); - e.pointer_bindings = PointerBindings::new(); - } - } - } - let pointer_bindings = - if held.saturating_add(pointer_bindings.len()) > MAX_SESSION_POINTER_BINDINGS { - PointerBindings::new() - } else { - pointer_bindings - }; sessions.insert( (source, challenge_id), SubtreeSession { @@ -4854,6 +4833,7 @@ impl SubtreeRound1Limiter { pointer_bindings, }, ); + true } /// Atomically consume the round-2 session for this exchange. `Some` iff a @@ -5281,7 +5261,7 @@ async fn handle_replication_message( // a live round-1 exchange. if let crate::replication::protocol::SubtreeAuditResponse::Proof { .. } = &response { - subtree_round1 + let opened = subtree_round1 .open_session( source, challenge.challenge_id, @@ -5290,6 +5270,24 @@ async fn handle_replication_message( storage_commitment_audit::pointer_bindings(&response), ) .await; + // A proof round 2 could not be answered for is not sent. + // Withholding it is the round-1 capacity drop the auditor + // already treats as a timeout (ADR-0017). + if !opened { + protocol::record_audit_drop(protocol::AuditDropKind::Subtree); + warn!( + target: "ant_node::replication::audit_responder", + event = "admission_dropped", + kind = "subtree", + responder_class = class.as_str(), + source = %source, + challenge_id = challenge.challenge_id, + request_response = rr_message_id.is_some(), + reason = "pointer_binding_budget", + "Audit responder admission dropped" + ); + return; + } } let response_kind = subtree_audit_response_kind(&response); let work_items = subtree_audit_response_work_items(&response); @@ -10770,9 +10768,11 @@ mod tests { // Session: opened by round 1, consumed exactly once by the matching round 2. let hash = [7u8; 32]; let nonce = [9u8; 32]; - limiter - .open_session(peer, 42, hash, nonce, PointerBindings::new()) - .await; + assert!( + limiter + .open_session(peer, 42, hash, nonce, PointerBindings::new()) + .await + ); // Wrong nonce / commitment does not match. assert!(limiter .consume_session(&peer, 42, &hash, &[0u8; 32]) @@ -10799,10 +10799,10 @@ mod tests { } // A session carries what its round 1 bound for each pointer to round 2, - // and every live session together stays under the binding budget: to make - // room the oldest sessions give theirs up, and a session larger than the - // whole budget keeps none rather than being refused. - #[tokio::test(start_paused = true)] + // and every live session together stays under the binding budget. A + // session that does not fit is refused, and no session already opened + // gives up its bindings or its place for it. + #[tokio::test] async fn subtree_session_carries_pointer_bindings_within_the_budget() { let limiter = SubtreeRound1Limiter::new(Duration::ZERO, 1); let (hash, nonce) = ([1u8; 32], [2u8; 32]); @@ -10816,47 +10816,107 @@ mod tests { }) .collect() }; - let open = |peer: u8, id: u64, kept: PointerBindings| { - let limiter = limiter.clone(); - async move { - limiter - .open_session(test_peer(peer), id, hash, nonce, kept) - .await; - // Sessions are ordered by when they opened. - tokio::time::advance(Duration::from_millis(1)).await; - } - }; - let kept = |peer: u8, id: u64| { - let limiter = limiter.clone(); - async move { - limiter - .consume_session(&test_peer(peer), id, &hash, &nonce) - .await - .map(|b| b.len()) - } - }; let half = MAX_SESSION_POINTER_BINDINGS / 2; let first = bindings(0, half); - open(1, 1, first.clone()).await; - open(2, 2, bindings(1 << 40, half)).await; - // The budget is full. The next session takes the oldest one's room. - open(3, 3, bindings(1 << 41, 1)).await; - // One larger than the whole budget keeps nothing, and costs no one. - open(4, 4, bindings(1 << 42, MAX_SESSION_POINTER_BINDINGS + 1)).await; - - assert_eq!(kept(1, 1).await, Some(0), "the oldest gave its bindings up"); - assert_eq!(kept(2, 2).await, Some(half)); - assert_eq!(kept(3, 3).await, Some(1)); - assert_eq!(kept(4, 4).await, Some(0), "still opened, with none kept"); - - // With room, round 2 gets exactly what round 1 bound. - open(5, 5, first.clone()).await; + assert!( + limiter + .open_session(test_peer(1), 1, hash, nonce, first.clone()) + .await + ); + assert!( + limiter + .open_session(test_peer(2), 2, hash, nonce, bindings(1 << 40, half)) + .await + ); + // The budget is full: one more binding is refused, but a round 1 with + // no pointers in it still opens. + assert!( + !limiter + .open_session(test_peer(3), 3, hash, nonce, bindings(1 << 41, 1)) + .await + ); + assert!( + limiter + .open_session(test_peer(4), 4, hash, nonce, PointerBindings::new()) + .await + ); + assert_eq!( limiter - .consume_session(&test_peer(5), 5, &hash, &nonce) + .consume_session(&test_peer(1), 1, &hash, &nonce) .await, - Some(first) + Some(first.clone()), + "an opened session keeps every binding it was given" + ); + assert!( + limiter + .consume_session(&test_peer(3), 3, &hash, &nonce) + .await + .is_none(), + "a refused session was never opened" + ); + assert_eq!( + limiter + .consume_session(&test_peer(2), 2, &hash, &nonce) + .await + .map(|b| b.len()), + Some(half) + ); + + // Consumed sessions free their room. + assert!( + limiter + .open_session(test_peer(5), 5, hash, nonce, bindings(1 << 42, half)) + .await + ); + } + + // A session refused for the binding budget is refused before the session + // cap evicts anything: with both full, the refusal costs no session its + // place. + #[tokio::test] + async fn a_session_refused_for_bindings_evicts_no_other_session() { + let limiter = SubtreeRound1Limiter::new(Duration::ZERO, 1); + let (hash, nonce) = ([1u8; 32], [2u8; 32]); + let full: PointerBindings = (0u64..) + .take(MAX_SESSION_POINTER_BINDINGS) + .map(|i| { + let mut key = [0u8; 32]; + key[..8].copy_from_slice(&i.to_le_bytes()); + (key, [0xCD; 32]) + }) + .collect(); + assert!( + limiter + .open_session(test_peer(0), 0, hash, nonce, full) + .await + ); + for id in 1..MAX_SUBTREE_SESSIONS as u64 { + assert!( + limiter + .open_session(test_peer(1), id, hash, nonce, PointerBindings::new()) + .await + ); + } + + let one: PointerBindings = std::iter::once(([0xEE; 32], [0xEE; 32])).collect(); + assert!( + !limiter + .open_session(test_peer(2), u64::MAX, hash, nonce, one) + .await + ); + assert_eq!( + limiter.sessions.read().await.len(), + MAX_SUBTREE_SESSIONS, + "every session is still there" + ); + assert!( + limiter + .consume_session(&test_peer(0), 0, &hash, &nonce) + .await + .is_some(), + "including the oldest" ); } @@ -11071,9 +11131,11 @@ mod tests { let (id, hash, nonce) = (77u64, [3u8; 32], [4u8; 32]); let challenge = slice_challenge(id, hash, nonce); - round1 - .open_session(peer, id, hash, nonce, PointerBindings::new()) - .await; + assert!( + round1 + .open_session(peer, id, hash, nonce, PointerBindings::new()) + .await + ); // Saturate this peer's share so the next admission must be refused. let mut hold = Vec::new(); diff --git a/src/replication/storage_commitment_audit.rs b/src/replication/storage_commitment_audit.rs index 2b47b853..02006033 100644 --- a/src/replication/storage_commitment_audit.rs +++ b/src/replication/storage_commitment_audit.rs @@ -1972,9 +1972,10 @@ enum PointerServe { /// Nothing at all is held for it, which is a lost pointer. Absent, /// The pointer is held, but this node cannot serve the record round 1 - /// read: it has aged out or been evicted to keep memory bounded, or the - /// session gave round 1's root up to stay in budget. A local limit, not a - /// lost pointer, so it is reported as one rather than proved wrong. + /// read: it has aged out or been evicted to keep memory bounded, or no + /// root for it was kept, which a session this node opened never lacks. A + /// local limit, not a lost pointer, so it is reported as one rather than + /// proved wrong. Unavailable, } @@ -3103,7 +3104,7 @@ mod pointer_audit_tests { /// which may have read the pointer at its start, then the session its /// round 2 must arrive within. #[test] - fn a_replaced_record_outlives_the_slowest_audit() { + fn a_replaced_record_outlives_the_slowest_audit_by_default() { let largest = usize::try_from(max_subtree_leaves(MAX_COMMITMENT_KEY_COUNT)).expect("fits a usize"); let slowest = @@ -3140,10 +3141,11 @@ mod pointer_audit_tests { ); } - /// Without the root round 1 reported, as when its session gave it up to - /// stay in budget, a pointer still held is reported as a transient - /// failure, updated or not: the node cannot show which record round 1 - /// read, and guessing would risk a confirmed failure it did not earn. + /// Without the root round 1 reported, which a session this node opened + /// never lacks but a direct caller can, a pointer still held is reported + /// as a transient failure, updated or not: the node cannot show which + /// record round 1 read, and guessing would risk a confirmed failure it did + /// not earn. #[tokio::test] async fn without_the_bound_root_a_pointer_is_unavailable_not_failed() { let responder = Responder::new(24, 24).await;