diff --git a/src/replication/audit.rs b/src/replication/audit.rs index 3bfccb65..64b192de 100644 --- a/src/replication/audit.rs +++ b/src/replication/audit.rs @@ -174,7 +174,7 @@ pub async fn audit_tick_with_repair_proofs( return AuditTickResult::Idle; } - let sample_count = ReplicationConfig::audit_sample_count(all_keys.len()); + let sample_count = ReplicationConfig::responsible_audit_key_limit(all_keys.len()); let sampled_keys: Vec = { let mut rng = rand::thread_rng(); all_keys diff --git a/src/replication/config.rs b/src/replication/config.rs index cb56905d..ed42f470 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -143,7 +143,7 @@ pub const MAX_CONCURRENT_REPLICATION_SENDS: usize = 3; /// gossip-audit cooldown is 30 min, so genuine concurrent audits are few) while /// bounding the byte round's worst-case resident bytes /// (`N × MAX_BYTE_CHALLENGE_KEYS × MAX_CHUNK_SIZE`). -pub const MAX_CONCURRENT_AUDIT_RESPONSES: usize = 16; +pub const MAX_CONCURRENT_AUDIT_RESPONSES: usize = 32; /// Maximum concurrent in-flight audit-responder tasks from any SINGLE peer. /// @@ -153,9 +153,9 @@ pub const MAX_CONCURRENT_AUDIT_RESPONSES: usize = 16; /// timeout verdicts on the challenged peers). This per-peer cap guarantees no /// source holds more than its share, so a flood self-throttles. Audits are /// cooldown-gated (one -/// gossip-triggered audit per peer per 30 min), so 2 in-flight per peer -/// comfortably covers the legitimate round-1 + round-2 overlap. -pub const MAX_AUDIT_RESPONSES_PER_PEER: u32 = 2; +/// gossip-triggered audit per peer per 30 min), so 4 in-flight per peer leaves +/// headroom beyond the legitimate round-1 + round-2 overlap. +pub const MAX_AUDIT_RESPONSES_PER_PEER: u32 = 4; /// Concurrent fetches cap, derived from hardware thread count. /// @@ -629,15 +629,30 @@ impl ReplicationConfig { sqrt.max(1).min(total_keys) } + /// Maximum number of keys this node should send in one responsible-chunk + /// [`AuditChallenge`]. + /// + /// This is the sender-side limit used by the normal responsible audit path: + /// one challenge samples at most `sqrt(local_stored_keys)` keys. Other paths + /// that reuse the same `AuditChallenge` wire message, such as + /// prune-confirmation audits, should chunk to this same limit instead of + /// inventing a larger batch size. + /// + /// [`AuditChallenge`]: crate::replication::protocol::AuditChallenge + #[must_use] + pub fn responsible_audit_key_limit(local_stored_keys: usize) -> usize { + Self::audit_sample_count(local_stored_keys) + } + /// Maximum number of keys to accept in an incoming audit challenge. /// - /// Scales dynamically: `2 * audit_sample_count(stored_chunks)`. The 2x - /// margin accounts for the challenger having a larger store than us and - /// therefore sampling more keys. + /// Scales dynamically from the same responsible-audit sender limit, with a + /// 2x margin to account for the challenger having a larger local store than + /// us and therefore sampling more keys. #[must_use] pub fn max_incoming_audit_keys(stored_chunks: usize) -> usize { // Allow at least 1 key so a newly-joined node can still be audited. - (2 * Self::audit_sample_count(stored_chunks)).max(1) + (2 * Self::responsible_audit_key_limit(stored_chunks)).max(1) } /// Compute the audit response timeout for a challenge with @@ -1016,21 +1031,32 @@ mod tests { assert_eq!(ReplicationConfig::audit_sample_count(1_000_000), 1_000); } + #[test] + fn responsible_audit_key_limit_matches_audit_sample_count() { + for stored_keys in [0, 1, 3, 4, 25, 100, 1_000, 10_000, 1_000_000] { + assert_eq!( + ReplicationConfig::responsible_audit_key_limit(stored_keys), + ReplicationConfig::audit_sample_count(stored_keys), + "responsible audit sender limit must stay identical to audit sample count" + ); + } + } + #[test] fn max_incoming_audit_keys_scales_dynamically() { // Empty store: at least 1 key accepted. assert_eq!(ReplicationConfig::max_incoming_audit_keys(0), 1); - // 1 chunk: 2 * sqrt(1) = 2. + // 1 chunk: 2 * responsible_audit_key_limit(1) = 2. assert_eq!(ReplicationConfig::max_incoming_audit_keys(1), 2); - // 100 chunks: 2 * sqrt(100) = 20. + // 100 chunks: 2 * responsible_audit_key_limit(100) = 20. assert_eq!(ReplicationConfig::max_incoming_audit_keys(100), 20); - // 1M chunks: 2 * sqrt(1_000_000) = 2_000. + // 1M chunks: 2 * responsible_audit_key_limit(1_000_000) = 2_000. assert_eq!(ReplicationConfig::max_incoming_audit_keys(1_000_000), 2_000); - // 5M chunks: 2 * sqrt(5_000_000) = 4_472. + // 5M chunks: 2 * responsible_audit_key_limit(5_000_000) = 4_472. assert_eq!(ReplicationConfig::max_incoming_audit_keys(5_000_000), 4_472); } diff --git a/src/replication/mod.rs b/src/replication/mod.rs index cdb5cef7..4cb83d5d 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -34,6 +34,7 @@ pub mod subtree; pub mod types; use std::collections::{HashMap, HashSet}; +use std::fmt; use std::num::NonZeroUsize; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; @@ -2081,6 +2082,77 @@ impl ReplicationEngine { // Free functions for background tasks // =========================================================================== +/// Which ceiling rejected an audit-responder admission attempt. +/// +/// Stable, machine-readable so a production log-scrape can bucket drops by +/// cause. A 24 h node log collapsed 117 dropped responsible replies into a +/// single opaque "capacity reached" line; this splits the two distinct causes +/// so the next such investigation can tell a global-pool exhaustion (the whole +/// node is saturated) from a per-peer cap hit (one source is self-throttling) +/// without re-instrumenting. +/// +/// This branch runs a SINGLE shared audit-responder pool for all challenge +/// kinds (responsible / subtree / byte), so there is no separate slow/fast +/// pool to distinguish here — the `kind=` log field already separates the +/// responsible (fast-path) challenge from the heavier subtree/byte ones. If a +/// dedicated slow pool is later split out, add its variants here. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AuditResponderRejectReason { + /// The global [`MAX_CONCURRENT_AUDIT_RESPONSES`] semaphore had no permit + /// free; the per-peer cap was not the binding constraint. + GlobalPoolFull, + /// `source` already held its [`MAX_AUDIT_RESPONSES_PER_PEER`] in-flight + /// share, so the global permit was never attempted. + PerPeerCapFull, +} + +impl AuditResponderRejectReason { + /// Stable token emitted as `reason=` in drop logs. Keep these values + /// frozen — production log tooling greps for them. + fn as_str(self) -> &'static str { + match self { + Self::GlobalPoolFull => "global_pool_full", + Self::PerPeerCapFull => "per_peer_cap_full", + } + } +} + +/// Why an audit-responder admission attempt failed, with the decision-time +/// capacity counters that let a drop be logged with full context. +/// +/// `global_inflight`/`peer_inflight` are best-effort snapshots taken as the +/// decision was made (the two ceilings are read under different locks, so they +/// are not a single atomic view), but they are exact enough to tell a saturated +/// node from a single self-throttling flooder. +#[derive(Debug, Clone, Copy)] +struct AuditResponderAdmissionFailure { + reason: AuditResponderRejectReason, + /// Global permits in use across the whole engine at decision time. + global_inflight: usize, + /// Configured global ceiling ([`MAX_CONCURRENT_AUDIT_RESPONSES`]). + global_limit: usize, + /// In-flight audit responders already held for `source` at decision time. + peer_inflight: u32, + /// Configured per-peer ceiling ([`MAX_AUDIT_RESPONSES_PER_PEER`]). + peer_limit: u32, +} + +impl fmt::Display for AuditResponderAdmissionFailure { + /// Renders the stable `reason=... global_inflight=... global_limit=... + /// peer_inflight=... peer_limit=...` suffix appended to every drop log. + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "reason={} global_inflight={} global_limit={} peer_inflight={} peer_limit={}", + self.reason.as_str(), + self.global_inflight, + self.global_limit, + self.peer_inflight, + self.peer_limit, + ) + } +} + /// RAII admission for one audit-responder task: holds the GLOBAL permit and, /// on drop, decrements the PER-PEER in-flight count. Moving this into the /// spawned task ties both bounds to the task's exact lifetime — no manual @@ -2127,39 +2199,69 @@ impl Drop for AuditResponderGuard { } /// Try to admit one audit-responder task for `source`: take a global permit AND -/// a per-peer slot (both bounded). Returns `None` (caller drops the challenge, -/// leaving the remote auditor to apply that audit path's timeout policy) if -/// either ceiling is hit, so one flooder can neither exhaust the global pool's -/// effect on others nor exceed its own per-peer share (codex-r2 A). +/// a per-peer slot (both bounded). Returns `Err` with the binding ceiling and +/// its decision-time counters (caller drops the challenge, leaving the remote +/// auditor to apply that audit path's timeout policy) if either ceiling is hit, +/// so one flooder can neither exhaust the global pool's effect on others nor +/// exceed its own per-peer share (codex-r2 A). The `Err` reason lets the caller +/// log exactly WHY the drop happened rather than a single opaque "capacity +/// reached". async fn admit_audit_responder( semaphore: &Arc, inflight: &Arc>>, source: &PeerId, -) -> Option { +) -> std::result::Result { + let global_limit = MAX_CONCURRENT_AUDIT_RESPONSES; + let peer_limit = MAX_AUDIT_RESPONSES_PER_PEER; + // `available_permits()` is a cheap atomic load; `global_limit - available` + // is the best-effort in-flight count at decision time. Not synchronized with + // the per-peer lock, so it is a snapshot, not a single atomic view. + let global_inflight = |sem: &Semaphore| global_limit.saturating_sub(sem.available_permits()); + // Per-peer cap first (cheap, and the fairness-critical bound), committed // under the write lock so concurrent challenges from the same peer can't // both slip past the cap. { let mut map = inflight.write().await; let entry = map.entry(*source).or_insert(0); - if *entry >= MAX_AUDIT_RESPONSES_PER_PEER { - return None; + if *entry >= peer_limit { + let peer_inflight = *entry; + drop(map); // release before the (unrelated) semaphore read + return Err(AuditResponderAdmissionFailure { + reason: AuditResponderRejectReason::PerPeerCapFull, + global_inflight: global_inflight(semaphore), + global_limit, + peer_inflight, + peer_limit, + }); } *entry += 1; } // Then the global ceiling. If it's exhausted, give back the per-peer slot we // just claimed so it isn't leaked. let Ok(permit) = Arc::clone(semaphore).try_acquire_owned() else { - let mut map = inflight.write().await; - if let Some(n) = map.get_mut(source) { - *n = n.saturating_sub(1); - if *n == 0 { - map.remove(source); - } - } - return None; + let peer_inflight = { + let mut map = inflight.write().await; + map.remove(source).map_or(0, |n| { + // Report the per-peer occupancy AFTER releasing our rolled-back + // slot: the share still held by this source's other in-flight + // tasks (below the cap, since the per-peer check passed). + let remaining = n.saturating_sub(1); + if remaining > 0 { + map.insert(*source, remaining); + } + remaining + }) + }; + return Err(AuditResponderAdmissionFailure { + reason: AuditResponderRejectReason::GlobalPoolFull, + global_inflight: global_inflight(semaphore), + global_limit, + peer_inflight, + peer_limit, + }); }; - Some(AuditResponderGuard { + Ok(AuditResponderGuard { _permit: permit, inflight: Arc::clone(inflight), peer: *source, @@ -2303,15 +2405,21 @@ async fn handle_replication_message( // is hit. Responsible/prune audit timeouts are penalised by the // caller, so the caps must remain high enough for honest audit load; // the per-peer share still prevents one flooder from starving others. - let Some(guard) = - admit_audit_responder(audit_responder_semaphore, audit_responder_inflight, source) - .await - else { - warn!( - "Audit challenge reply not sent: kind=responsible response=dropped \ - source={source} (audit-responder capacity reached)" - ); - return Ok(()); + let guard = match admit_audit_responder( + audit_responder_semaphore, + audit_responder_inflight, + source, + ) + .await + { + Ok(guard) => guard, + Err(failure) => { + warn!( + "Audit challenge reply not sent: kind=responsible response=dropped \ + source={source} {failure}" + ); + return Ok(()); + } }; let bootstrapping = *is_bootstrapping.read().await; let storage = Arc::clone(storage); @@ -2354,15 +2462,21 @@ async fn handle_replication_message( "Audit challenge received: kind=subtree source={source} request_response={}", rr_message_id.is_some(), ); - let Some(guard) = - admit_audit_responder(audit_responder_semaphore, audit_responder_inflight, source) - .await - else { - warn!( - "Audit challenge reply not sent: kind=subtree response=dropped \ - source={source} (audit-responder capacity reached)" - ); - return Ok(()); + let guard = match admit_audit_responder( + audit_responder_semaphore, + audit_responder_inflight, + source, + ) + .await + { + Ok(guard) => guard, + Err(failure) => { + warn!( + "Audit challenge reply not sent: kind=subtree response=dropped \ + source={source} {failure}" + ); + return Ok(()); + } }; let bootstrapping = *is_bootstrapping.read().await; let storage = Arc::clone(storage); @@ -2416,15 +2530,21 @@ async fn handle_replication_message( "Audit challenge received: kind=byte source={source} request_response={}", rr_message_id.is_some(), ); - let Some(guard) = - admit_audit_responder(audit_responder_semaphore, audit_responder_inflight, source) - .await - else { - warn!( - "Audit challenge reply not sent: kind=byte response=dropped \ - source={source} (audit-responder capacity reached)" - ); - return Ok(()); + let guard = match admit_audit_responder( + audit_responder_semaphore, + audit_responder_inflight, + source, + ) + .await + { + Ok(guard) => guard, + Err(failure) => { + warn!( + "Audit challenge reply not sent: kind=byte response=dropped \ + source={source} {failure}" + ); + return Ok(()); + } }; let bootstrapping = *is_bootstrapping.read().await; let storage = Arc::clone(storage); @@ -2480,12 +2600,18 @@ async fn handle_replication_message( // cap) so a flood of fetches cannot drive unbounded commitment // clone/encode/send work; over-limit is dropped, which the fetching // peer graces exactly like a missed audit response. - let Some(_guard) = - admit_audit_responder(audit_responder_semaphore, audit_responder_inflight, source) - .await - else { - debug!("GetCommitmentByPin from {source} dropped: responder capacity reached"); - return Ok(()); + let _guard = match admit_audit_responder( + audit_responder_semaphore, + audit_responder_inflight, + source, + ) + .await + { + Ok(guard) => guard, + Err(failure) => { + debug!("GetCommitmentByPin from {source} dropped: {failure}"); + return Ok(()); + } }; let response = my_commitment_state.lookup_by_hash(&request.pin).map_or( protocol::GetCommitmentByPinResponse::NotRetained { pin: request.pin }, @@ -5313,6 +5439,58 @@ mod tests { k } + #[tokio::test] + async fn audit_responder_admission_reports_per_peer_cap_full() { + let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT_AUDIT_RESPONSES)); + let inflight = Arc::new(RwLock::new(HashMap::new())); + let peer = test_peer(0xA1); + + let mut guards = Vec::new(); + for _ in 0..MAX_AUDIT_RESPONSES_PER_PEER { + match admit_audit_responder(&semaphore, &inflight, &peer).await { + Ok(guard) => guards.push(guard), + Err(err) => panic!("unexpected admission failure before peer cap: {err:?}"), + } + } + + let Err(err) = admit_audit_responder(&semaphore, &inflight, &peer).await else { + panic!("admission should fail once per-peer cap is full"); + }; + assert_eq!(err.reason, AuditResponderRejectReason::PerPeerCapFull); + assert_eq!(err.peer_inflight, MAX_AUDIT_RESPONSES_PER_PEER); + assert_eq!(err.peer_limit, MAX_AUDIT_RESPONSES_PER_PEER); + assert_eq!(err.global_limit, MAX_CONCURRENT_AUDIT_RESPONSES); + + drop(guards); + } + + #[tokio::test] + async fn audit_responder_admission_reports_global_pool_full() { + let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT_AUDIT_RESPONSES)); + let inflight = Arc::new(RwLock::new(HashMap::new())); + let peer = test_peer(0xA2); + + let mut held_global_permits = Vec::new(); + for _ in 0..MAX_CONCURRENT_AUDIT_RESPONSES { + held_global_permits.push( + Arc::clone(&semaphore) + .try_acquire_owned() + .expect("test should be able to exhaust the global pool"), + ); + } + + let Err(err) = admit_audit_responder(&semaphore, &inflight, &peer).await else { + panic!("admission should fail once global pool is full"); + }; + assert_eq!(err.reason, AuditResponderRejectReason::GlobalPoolFull); + assert_eq!(err.global_inflight, MAX_CONCURRENT_AUDIT_RESPONSES); + assert_eq!(err.global_limit, MAX_CONCURRENT_AUDIT_RESPONSES); + assert_eq!(err.peer_inflight, 0); + assert_eq!(err.peer_limit, MAX_AUDIT_RESPONSES_PER_PEER); + + drop(held_global_permits); + } + #[test] fn first_audit_terminal_outcomes_are_stable() { let peer = test_peer(1); diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index b5e5f76e..87bf1f26 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -462,6 +462,7 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune stats.audits_attempted = candidates.len(); let present_by_key = collect_record_prune_proofs( &candidates, + stored_keys.len(), ctx.storage, ctx.p2p_node, ctx.config, @@ -1083,6 +1084,7 @@ async fn delete_stored_records( /// stored keys, including out-of-range keys retained by hysteresis. async fn collect_record_prune_proofs( candidates: &[RecordPruneCandidate], + local_stored_key_count: usize, storage: &Arc, p2p_node: &Arc, config: &ReplicationConfig, @@ -1092,24 +1094,29 @@ async fn collect_record_prune_proofs( return HashMap::new(); } + let max_keys_per_challenge = + ReplicationConfig::responsible_audit_key_limit(local_stored_key_count); let report_state = PruneAuditReportState::default(); - let mut requests = stream::iter(build_peer_audit_challenges(candidates)) - .map(|(peer, key)| { - peer_proves_record( - peer, - key, - storage, - p2p_node, - config, - sync_state, - &report_state, - ) - }) - .buffer_unordered(MAX_CONCURRENT_PRUNE_AUDIT_CHALLENGES); + let mut requests = stream::iter(build_peer_audit_challenges( + candidates, + max_keys_per_challenge, + )) + .map(|(peer, keys)| { + peer_proves_records( + peer, + keys, + storage, + p2p_node, + config, + sync_state, + &report_state, + ) + }) + .buffer_unordered(MAX_CONCURRENT_PRUNE_AUDIT_CHALLENGES); let mut present_by_key = HashMap::>::new(); - while let Some(proof) = requests.next().await { - if let Some((peer, key)) = proof { + while let Some(proofs) = requests.next().await { + for (peer, key) in proofs { present_by_key.entry(key).or_default().insert(peer); } } @@ -1248,13 +1255,27 @@ fn revalidate_record_prune_candidate( } } -fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(PeerId, XorName)> { - let mut challenges = Vec::new(); +fn build_peer_audit_challenges( + candidates: &[RecordPruneCandidate], + max_keys_per_challenge: usize, +) -> Vec<(PeerId, Vec)> { + let max_keys_per_challenge = max_keys_per_challenge.max(1); + let mut keys_by_peer: HashMap> = HashMap::new(); for candidate in candidates { for peer in &candidate.target_peers { - challenges.push((*peer, candidate.key)); + keys_by_peer.entry(*peer).or_default().push(candidate.key); } } + + let mut challenges = Vec::new(); + for (peer, mut keys) in keys_by_peer { + keys.sort_unstable(); + keys.dedup(); + challenges.extend( + keys.chunks(max_keys_per_challenge) + .map(|chunk| (peer, chunk.to_vec())), + ); + } challenges } @@ -1321,52 +1342,101 @@ fn target_peers_reported_present( proven >= proofs_needed } -/// Challenge a peer to prove it holds the exact record bytes for `key`. -/// `None` means the peer failed to provide usable proof. -async fn peer_proves_record( +/// Challenge a peer to prove it holds the exact record bytes for one or more keys. +/// +/// Batching by peer prevents a prune pass from firing many simultaneous one-key +/// `AuditChallenge`s at the same target. The responder already supports +/// multi-key challenges, so we preserve per-key proof accounting while reducing +/// per-peer request bursts. +async fn peer_proves_records( peer: PeerId, - key: XorName, + keys: Vec, storage: &Arc, p2p_node: &Arc, config: &ReplicationConfig, sync_state: &Arc>, report_state: &PruneAuditReportState, -) -> Option<(PeerId, XorName)> { - let local_bytes = local_record_bytes(&key, storage).await?; - +) -> Vec<(PeerId, XorName)> { let (challenge_id, nonce) = { let mut rng = rand::thread_rng(); (rng.gen::(), rng.gen::<[u8; 32]>()) }; - let (encoded, key_count) = encode_prune_audit_challenge(&peer, key, challenge_id, nonce)?; + let mut challenge_material = Vec::new(); + for key in keys { + if let Some(expected_digest) = local_record_digest(&peer, &key, &nonce, storage).await { + challenge_material.push((key, expected_digest)); + } + } + if challenge_material.is_empty() { + return Vec::new(); + } + + let challenge_keys: Vec = challenge_material.iter().map(|(key, _)| *key).collect(); + let Some((encoded, key_count)) = + encode_prune_audit_challenge(&peer, &challenge_keys, challenge_id, nonce) + else { + return Vec::new(); + }; let Some(decoded) = - send_prune_audit_challenge(&peer, &key, encoded, key_count, p2p_node, config).await + send_prune_audit_challenge(&peer, encoded, key_count, p2p_node, config).await else { // No decoded response means a timeout or malformed reply. Prune // confirmation reuses `AuditChallenge` semantics, so this is an immediate - // audit failure just like a decoded bad proof below. - report_prune_audit_failure_once(&peer, &key, p2p_node, config, report_state).await; - return None; + // audit failure just like a decoded bad proof below. Keep the historical + // one-report-per-peer-per-pass guard by attempting each key against the + // shared `report_state`. + let mut audit_failure_reported = false; + for key in &challenge_keys { + if report_prune_audit_failure_once(&peer, key, p2p_node, config, report_state).await { + audit_failure_reported = true; + break; + } + } + if audit_failure_reported { + debug!("Prune audit: reported one failure for timed-out/malformed batch from {peer}"); + } + return Vec::new(); }; - let status = - prune_audit_response_status(decoded, challenge_id, &peer, &key, &nonce, &local_bytes); - if prune_audit_response_clears_bootstrap_claim(status) { - clear_prune_bootstrap_claim(&peer, sync_state).await; - } + let statuses = prune_audit_response_statuses(decoded, challenge_id, &peer, &challenge_material); + let mut clear_bootstrap_claim = false; + let mut audit_failure_reported = false; + let mut proven = Vec::new(); - match status { - PruneAuditStatus::Proven => Some((peer, key)), - PruneAuditStatus::Bootstrapping => { - report_prune_bootstrap_claim(&peer, &key, p2p_node, config, sync_state, report_state) - .await; - None + for (key, status) in statuses { + if prune_audit_response_clears_bootstrap_claim(status) { + clear_bootstrap_claim = true; } - PruneAuditStatus::Failed => { - report_prune_audit_failure_once(&peer, &key, p2p_node, config, report_state).await; - None + + match status { + PruneAuditStatus::Proven => proven.push((peer, key)), + PruneAuditStatus::Bootstrapping => { + report_prune_bootstrap_claim( + &peer, + &key, + p2p_node, + config, + sync_state, + report_state, + ) + .await; + } + PruneAuditStatus::Failed => { + if !audit_failure_reported + && report_prune_audit_failure_once(&peer, &key, p2p_node, config, report_state) + .await + { + audit_failure_reported = true; + } + } } } + + if clear_bootstrap_claim { + clear_prune_bootstrap_claim(&peer, sync_state).await; + } + + proven } fn prune_audit_response_clears_bootstrap_claim(status: PruneAuditStatus) -> bool { @@ -1377,18 +1447,20 @@ fn prune_audit_response_clears_bootstrap_claim(status: PruneAuditStatus) -> bool // challenges, which reuse the same wire message) lives in // `super::handle_audit_challenge_msg` -> `audit::handle_audit_challenge`, the // responsible-chunk audit responder. No separate prune-only responder is needed. - fn encode_prune_audit_challenge( peer: &PeerId, - key: XorName, + keys: &[XorName], challenge_id: u64, nonce: [u8; 32], ) -> Option<(Vec, usize)> { + if keys.is_empty() { + return None; + } let challenge = AuditChallenge { challenge_id, nonce, challenged_peer_id: *peer.as_bytes(), - keys: vec![key], + keys: keys.to_vec(), }; let key_count = challenge.keys.len(); let msg = ReplicationMessage { @@ -1399,8 +1471,8 @@ fn encode_prune_audit_challenge( Ok(data) => data, Err(e) => { warn!( - "Failed to encode prune audit challenge for {} against {peer}: {e}", - hex::encode(key), + "Failed to encode prune audit challenge with {} keys against {peer}: {e}", + keys.len(), ); return None; } @@ -1410,7 +1482,6 @@ fn encode_prune_audit_challenge( async fn send_prune_audit_challenge( peer: &PeerId, - key: &XorName, encoded: Vec, key_count: usize, p2p_node: &Arc, @@ -1423,10 +1494,7 @@ async fn send_prune_audit_challenge( { Ok(response) => response, Err(e) => { - debug!( - "Prune audit challenge for {} against {peer} failed: {e}", - hex::encode(key) - ); + debug!("Prune audit challenge with {key_count} keys against {peer} failed: {e}"); return None; } }; @@ -1442,59 +1510,76 @@ async fn send_prune_audit_challenge( Some(decoded) } -fn prune_audit_response_status( +fn prune_audit_response_statuses( decoded: ReplicationMessage, challenge_id: u64, peer: &PeerId, - key: &XorName, - nonce: &[u8; 32], - local_bytes: &[u8], -) -> PruneAuditStatus { + challenge_material: &[(XorName, [u8; 32])], +) -> Vec<(XorName, PruneAuditStatus)> { + let failed_all = |reason: &str| { + warn!( + "Prune audit proof batch from {peer} failed for {} keys: {reason}", + challenge_material.len() + ); + challenge_material + .iter() + .map(|(key, _)| (*key, PruneAuditStatus::Failed)) + .collect() + }; + match decoded.body { ReplicationMessageBody::AuditResponse(AuditResponse::Digests { challenge_id: resp_id, digests, }) => { if resp_id != challenge_id { - warn!("Prune audit challenge ID mismatch from {peer}"); - return PruneAuditStatus::Failed; + return failed_all("challenge id mismatch"); } - let [digest] = digests.as_slice() else { - warn!( - "Prune audit response from {peer} returned {} digests for one challenged key", + if digests.len() != challenge_material.len() { + return failed_all(&format!( + "returned {} digests for {} challenged keys", digests.len(), - ); - return PruneAuditStatus::Failed; - }; - if *digest == ABSENT_KEY_DIGEST { - warn!( - "Prune audit proof from {peer} failed for {}: peer reports key absent", - hex::encode(key) - ); - return PruneAuditStatus::Failed; - } - if audit_digest_proves_key(peer, key, nonce, local_bytes, digest) { - PruneAuditStatus::Proven - } else { - warn!( - "Prune audit proof from {peer} failed for {}: digest mismatch", - hex::encode(key) - ); - PruneAuditStatus::Failed + challenge_material.len() + )); } + + challenge_material + .iter() + .zip(digests.iter()) + .map(|((key, expected_digest), digest)| { + if *digest == ABSENT_KEY_DIGEST { + warn!( + "Prune audit proof from {peer} failed for {}: peer reports key absent", + hex::encode(key) + ); + return (*key, PruneAuditStatus::Failed); + } + if digest == expected_digest { + (*key, PruneAuditStatus::Proven) + } else { + warn!( + "Prune audit proof from {peer} failed for {}: digest mismatch", + hex::encode(key) + ); + (*key, PruneAuditStatus::Failed) + } + }) + .collect() } ReplicationMessageBody::AuditResponse(AuditResponse::Bootstrapping { challenge_id: resp_id, }) => { if resp_id == challenge_id { warn!( - "Prune audit proof for {} blocked by bootstrap claim from {peer}", - hex::encode(key) + "Prune audit proof batch for {} keys blocked by bootstrap claim from {peer}", + challenge_material.len() ); - PruneAuditStatus::Bootstrapping + challenge_material + .iter() + .map(|(key, _)| (*key, PruneAuditStatus::Bootstrapping)) + .collect() } else { - warn!("Prune audit challenge ID mismatch on Bootstrapping from {peer}"); - PruneAuditStatus::Failed + failed_all("challenge id mismatch on Bootstrapping") } } ReplicationMessageBody::AuditResponse(AuditResponse::Rejected { @@ -1503,21 +1588,32 @@ fn prune_audit_response_status( }) => { if resp_id == challenge_id { warn!( - "Prune audit proof for {} rejected by {peer}: {reason}", - hex::encode(key) + "Prune audit proof batch for {} keys rejected by {peer}: {reason}", + challenge_material.len() ); } else { warn!("Prune audit challenge ID mismatch on Rejected from {peer}"); } - PruneAuditStatus::Failed - } - _ => { - warn!("Unexpected prune audit response type from {peer}"); - PruneAuditStatus::Failed + challenge_material + .iter() + .map(|(key, _)| (*key, PruneAuditStatus::Failed)) + .collect() } + _ => failed_all("unexpected response type"), } } +async fn local_record_digest( + peer: &PeerId, + key: &XorName, + nonce: &[u8; 32], + storage: &Arc, +) -> Option<[u8; 32]> { + local_record_bytes(key, storage) + .await + .map(|bytes| compute_audit_digest(nonce, peer.as_bytes(), key, &bytes)) +} + async fn local_record_bytes(key: &XorName, storage: &Arc) -> Option> { match storage.get_raw(key).await { Ok(Some(bytes)) => Some(bytes), @@ -1538,6 +1634,7 @@ async fn local_record_bytes(key: &XorName, storage: &Arc) -> Option } } +#[cfg(test)] fn audit_digest_proves_key( peer: &PeerId, key: &XorName, @@ -1697,7 +1794,7 @@ mod tests { } #[test] - fn prune_audit_challenges_are_one_per_candidate_peer() { + fn prune_audit_challenges_are_batched_by_target_peer() { let peer_a = peer_id_from_byte(1); let peer_b = peer_id_from_byte(2); let key_a = key_from_byte(0xA); @@ -1707,14 +1804,77 @@ mod tests { candidate(key_b, vec![peer_b]), ]; - let mut challenges = build_peer_audit_challenges(&candidates); - challenges.sort_unstable_by_key(|(peer, key)| (*peer.as_bytes(), *key)); + let mut challenges = build_peer_audit_challenges(&candidates, 2); + for (_, keys) in &mut challenges { + keys.sort_unstable(); + } + challenges.sort_unstable_by_key(|(peer, keys)| (*peer.as_bytes(), keys.clone())); - let mut expected = vec![(peer_a, key_a), (peer_b, key_a), (peer_b, key_b)]; - expected.sort_unstable_by_key(|(peer, key)| (*peer.as_bytes(), *key)); + let mut expected = vec![(peer_a, vec![key_a]), (peer_b, vec![key_a, key_b])]; + expected.sort_unstable_by_key(|(peer, keys)| (*peer.as_bytes(), keys.clone())); assert_eq!(challenges, expected); } + #[test] + fn prune_audit_challenges_split_peer_batches_at_responsible_audit_limit() { + let peer = peer_id_from_byte(1); + let candidates = vec![ + candidate(key_from_byte(0xA), vec![peer]), + candidate(key_from_byte(0xB), vec![peer]), + candidate(key_from_byte(0xC), vec![peer]), + candidate(key_from_byte(0xD), vec![peer]), + candidate(key_from_byte(0xE), vec![peer]), + ]; + + let mut challenges = build_peer_audit_challenges(&candidates, 2); + for (_, keys) in &mut challenges { + keys.sort_unstable(); + } + challenges.sort_unstable_by_key(|(_, keys)| keys.clone()); + + assert_eq!( + challenges, + vec![ + (peer, vec![key_from_byte(0xA), key_from_byte(0xB)]), + (peer, vec![key_from_byte(0xC), key_from_byte(0xD)]), + (peer, vec![key_from_byte(0xE)]), + ] + ); + } + + #[test] + fn prune_audit_batched_digest_response_is_evaluated_per_key() { + let peer = peer_id_from_byte(7); + let key_a = key_from_byte(0xA); + let key_b = key_from_byte(0xB); + let nonce = [0x7A; 32]; + let bytes_a = b"record-a".to_vec(); + let expected_a = compute_audit_digest(&nonce, peer.as_bytes(), &key_a, &bytes_a); + let expected_b = compute_audit_digest(&nonce, peer.as_bytes(), &key_b, b"record-b"); + let msg = ReplicationMessage { + request_id: 42, + body: ReplicationMessageBody::AuditResponse(AuditResponse::Digests { + challenge_id: 42, + digests: vec![expected_a, ABSENT_KEY_DIGEST], + }), + }; + + let statuses = prune_audit_response_statuses( + msg, + 42, + &peer, + &[(key_a, expected_a), (key_b, expected_b)], + ); + + assert_eq!( + statuses, + vec![ + (key_a, PruneAuditStatus::Proven), + (key_b, PruneAuditStatus::Failed), + ] + ); + } + #[test] fn confirmed_keys_require_quorum_of_target_peers_present() { let peer_a = peer_id_from_byte(1); @@ -2591,14 +2751,14 @@ mod tests { } fn graded_status(peer: &PeerId, key: &XorName, msg: ReplicationMessage) -> PruneAuditStatus { - prune_audit_response_status( - msg, - TEST_CHALLENGE_ID, - peer, - key, - &TEST_NONCE, - TEST_RECORD_BYTES, - ) + let expected = compute_audit_digest(&TEST_NONCE, peer.as_bytes(), key, TEST_RECORD_BYTES); + let statuses = + prune_audit_response_statuses(msg, TEST_CHALLENGE_ID, peer, &[(*key, expected)]); + statuses + .into_iter() + .next() + .map(|(_, status)| status) + .expect("single-key batch returns exactly one status") } #[test]