From 9d180fcbc56c790e061816ac342d7e05787adaf4 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 9 Jul 2026 14:58:08 +0200 Subject: [PATCH 1/6] chore: improve audit responder capacity tracking and logging --- src/replication/mod.rs | 293 +++++++++++++++++++++++++++++++++-------- 1 file changed, 240 insertions(+), 53 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 6495dc11..1cd78fb4 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,73 @@ 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); + let peer_inflight = { + let mut map = inflight.write().await; + match map.get_mut(source) { + // 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). + Some(n) => { + *n = n.saturating_sub(1); + let remaining = *n; + if *n == 0 { + map.remove(source); + } + remaining + } + None => 0, } - } - return None; + }; + 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 +2409,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 +2466,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 +2534,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 +2604,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 }, @@ -5306,12 +5436,13 @@ async fn rebuild_and_rotate_commitment( #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod tests { use super::{ - apply_audit_failure_credit_revocation, audit_failure_clears_bootstrap_claim, - audit_failure_revokes_holder_credit, audit_launch_decision, config, cooldown_allows_audit, - first_audit_terminal_outcome, first_failed_key_label, fresh_offer_payment_context, - paid_notify_payment_context, queue_first_audit_event, quote_within_audit_window, + admit_audit_responder, apply_audit_failure_credit_revocation, + audit_failure_clears_bootstrap_claim, audit_failure_revokes_holder_credit, + audit_launch_decision, config, cooldown_allows_audit, first_audit_terminal_outcome, + first_failed_key_label, fresh_offer_payment_context, paid_notify_payment_context, + queue_first_audit_event, quote_within_audit_window, AuditResponderRejectReason, FirstAuditQueueOutcome, FirstAuditTerminalOutcome, MonetizedPinEvent, - MONETIZED_AUDIT_SKEW_MARGIN, + MAX_AUDIT_RESPONSES_PER_PEER, MAX_CONCURRENT_AUDIT_RESPONSES, MONETIZED_AUDIT_SKEW_MARGIN, }; use crate::payment::VerificationContext; use crate::replication::audit::AuditTickResult; @@ -5321,9 +5452,11 @@ mod tests { use saorsa_core::identity::PeerId; use std::collections::HashMap; use std::num::NonZeroUsize; + use std::sync::Arc; use std::time::Duration; use std::time::Instant; use std::time::SystemTime; + use tokio::sync::{RwLock, Semaphore}; fn test_peer(b: u8) -> PeerId { let mut bytes = [0u8; 32]; @@ -5337,6 +5470,60 @@ 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 = match admit_audit_responder(&semaphore, &inflight, &peer).await { + Ok(_) => panic!("admission should fail once per-peer cap is full"), + Err(err) => err, + }; + 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 = match admit_audit_responder(&semaphore, &inflight, &peer).await { + Ok(_) => panic!("admission should fail once global pool is full"), + Err(err) => err, + }; + 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); From 426ef16dcf5edf92ccc8744aaaa49c0e688eafcb Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:13:54 +0200 Subject: [PATCH 2/6] fix(replication): raise audit responder capacity caps --- src/replication/config.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/replication/config.rs b/src/replication/config.rs index cb56905d..8a7dd746 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. /// From 622cdb068f58f6d1d2fac3eb1b8789f4967be2fd Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 9 Jul 2026 17:26:13 +0200 Subject: [PATCH 3/6] fix(replication): batch record prune audit challenges --- src/replication/pruning.rs | 272 +++++++++++++++++++++++++------------ 1 file changed, 185 insertions(+), 87 deletions(-) diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 79db6c9a..8db38c6e 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -958,10 +958,10 @@ async fn collect_record_prune_proofs( let report_state = PruneAuditReportState::default(); let mut requests = stream::iter(build_peer_audit_challenges(candidates)) - .map(|(peer, key)| { - peer_proves_record( + .map(|(peer, keys)| { + peer_proves_records( peer, - key, + keys, storage, p2p_node, config, @@ -972,8 +972,8 @@ async fn collect_record_prune_proofs( .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); } } @@ -1061,14 +1061,22 @@ async fn revalidated_record_prune_keys( (keys_to_delete, cleared) } -fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(PeerId, XorName)> { - let mut challenges = Vec::new(); +fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(PeerId, Vec)> { + 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); } } - challenges + + keys_by_peer + .into_iter() + .map(|(peer, mut keys)| { + keys.sort_unstable(); + keys.dedup(); + (peer, keys) + }) + .collect() } #[cfg(test)] @@ -1134,52 +1142,89 @@ 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 mut challenge_material = Vec::new(); + for key in keys { + if let Some(local_bytes) = local_record_bytes(&key, storage).await { + challenge_material.push((key, local_bytes)); + } + } + if challenge_material.is_empty() { + return Vec::new(); + } + let challenge_keys: Vec = challenge_material.iter().map(|(key, _)| *key).collect(); 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 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`. + for key in &challenge_keys { + report_prune_audit_failure_once(&peer, key, p2p_node, config, report_state).await; + } + 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, &nonce, &challenge_material); + let mut clear_bootstrap_claim = 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 => { + report_prune_audit_failure_once(&peer, &key, p2p_node, config, report_state).await; + } } } + + if clear_bootstrap_claim { + clear_prune_bootstrap_claim(&peer, sync_state).await; + } + + proven } fn prune_audit_response_clears_bootstrap_claim(status: PruneAuditStatus) -> bool { @@ -1190,18 +1235,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 { @@ -1212,8 +1259,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; } @@ -1223,7 +1270,6 @@ fn encode_prune_audit_challenge( async fn send_prune_audit_challenge( peer: &PeerId, - key: &XorName, encoded: Vec, key_count: usize, p2p_node: &Arc, @@ -1236,10 +1282,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; } }; @@ -1255,59 +1298,77 @@ 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, Vec)], +) -> 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, local_bytes), 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 audit_digest_proves_key(peer, key, nonce, local_bytes, 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 { @@ -1316,18 +1377,18 @@ 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"), } } @@ -1510,7 +1571,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); @@ -1521,13 +1582,50 @@ mod tests { ]; let mut challenges = build_peer_audit_challenges(&candidates); - challenges.sort_unstable_by_key(|(peer, key)| (*peer.as_bytes(), *key)); + 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_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 bytes_b = b"record-b".to_vec(); + let digest_a = compute_audit_digest(&nonce, peer.as_bytes(), &key_a, &bytes_a); + let msg = ReplicationMessage { + request_id: 42, + body: ReplicationMessageBody::AuditResponse(AuditResponse::Digests { + challenge_id: 42, + digests: vec![digest_a, ABSENT_KEY_DIGEST], + }), + }; + + let statuses = prune_audit_response_statuses( + msg, + 42, + &peer, + &nonce, + &[(key_a, bytes_a), (key_b, bytes_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); From c7b0626fff95e0f8181f1718712a5b100b039e94 Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:08:17 +0200 Subject: [PATCH 4/6] fix(replication): align prune audit key limits --- src/replication/audit.rs | 2 +- src/replication/config.rs | 42 +++++++++++++++---- src/replication/pruning.rs | 86 ++++++++++++++++++++++++++++---------- 3 files changed, 98 insertions(+), 32 deletions(-) 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 8a7dd746..ed42f470 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -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/pruning.rs b/src/replication/pruning.rs index 8db38c6e..b69ba3d0 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -373,6 +373,7 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune let present_by_key = collect_record_prune_proofs( &candidates, + stored_keys.len(), ctx.storage, ctx.p2p_node, ctx.config, @@ -947,6 +948,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, @@ -956,20 +958,25 @@ 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, keys)| { - peer_proves_records( - peer, - keys, - 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(proofs) = requests.next().await { @@ -1061,7 +1068,11 @@ async fn revalidated_record_prune_keys( (keys_to_delete, cleared) } -fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(PeerId, Vec)> { +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 { @@ -1069,14 +1080,16 @@ fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(Peer } } - keys_by_peer - .into_iter() - .map(|(peer, mut keys)| { - keys.sort_unstable(); - keys.dedup(); - (peer, keys) - }) - .collect() + 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 } #[cfg(test)] @@ -1581,7 +1594,7 @@ mod tests { candidate(key_b, vec![peer_b]), ]; - let mut challenges = build_peer_audit_challenges(&candidates); + let mut challenges = build_peer_audit_challenges(&candidates, 2); for (_, keys) in &mut challenges { keys.sort_unstable(); } @@ -1592,6 +1605,33 @@ mod tests { 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); From a00b38e31e199ce81d52ffb6e845e31c90be609f Mon Sep 17 00:00:00 2001 From: Mick van Dijke <12992260+mickvandijke@users.noreply.github.com> Date: Thu, 9 Jul 2026 18:20:45 +0200 Subject: [PATCH 5/6] fix(replication): address prune audit review feedback --- src/replication/mod.rs | 26 +++++++---------- src/replication/pruning.rs | 60 ++++++++++++++++++++++++++------------ 2 files changed, 51 insertions(+), 35 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 1cd78fb4..49595958 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -2242,20 +2242,16 @@ async fn admit_audit_responder( let Ok(permit) = Arc::clone(semaphore).try_acquire_owned() else { let peer_inflight = { let mut map = inflight.write().await; - match map.get_mut(source) { + 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). - Some(n) => { - *n = n.saturating_sub(1); - let remaining = *n; - if *n == 0 { - map.remove(source); - } - remaining + let remaining = n.saturating_sub(1); + if remaining > 0 { + map.insert(*source, remaining); } - None => 0, - } + remaining + }) }; return Err(AuditResponderAdmissionFailure { reason: AuditResponderRejectReason::GlobalPoolFull, @@ -5484,9 +5480,8 @@ mod tests { } } - let err = match admit_audit_responder(&semaphore, &inflight, &peer).await { - Ok(_) => panic!("admission should fail once per-peer cap is full"), - Err(err) => 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); @@ -5511,9 +5506,8 @@ mod tests { ); } - let err = match admit_audit_responder(&semaphore, &inflight, &peer).await { - Ok(_) => panic!("admission should fail once global pool is full"), - Err(err) => err, + 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); diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index b69ba3d0..baa12b14 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -1170,10 +1170,14 @@ async fn peer_proves_records( sync_state: &Arc>, report_state: &PruneAuditReportState, ) -> Vec<(PeerId, XorName)> { + let (challenge_id, nonce) = { + let mut rng = rand::thread_rng(); + (rng.gen::(), rng.gen::<[u8; 32]>()) + }; let mut challenge_material = Vec::new(); for key in keys { - if let Some(local_bytes) = local_record_bytes(&key, storage).await { - challenge_material.push((key, local_bytes)); + if let Some(expected_digest) = local_record_digest(&peer, &key, &nonce, storage).await { + challenge_material.push((key, expected_digest)); } } if challenge_material.is_empty() { @@ -1181,10 +1185,6 @@ async fn peer_proves_records( } let challenge_keys: Vec = challenge_material.iter().map(|(key, _)| *key).collect(); - let (challenge_id, nonce) = { - let mut rng = rand::thread_rng(); - (rng.gen::(), rng.gen::<[u8; 32]>()) - }; let Some((encoded, key_count)) = encode_prune_audit_challenge(&peer, &challenge_keys, challenge_id, nonce) else { @@ -1198,15 +1198,22 @@ async fn peer_proves_records( // 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 { - report_prune_audit_failure_once(&peer, key, p2p_node, config, report_state).await; + 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 statuses = - prune_audit_response_statuses(decoded, challenge_id, &peer, &nonce, &challenge_material); + 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(); for (key, status) in statuses { @@ -1228,7 +1235,12 @@ async fn peer_proves_records( .await; } PruneAuditStatus::Failed => { - report_prune_audit_failure_once(&peer, &key, p2p_node, config, report_state).await; + if !audit_failure_reported + && report_prune_audit_failure_once(&peer, &key, p2p_node, config, report_state) + .await + { + audit_failure_reported = true; + } } } } @@ -1315,8 +1327,7 @@ fn prune_audit_response_statuses( decoded: ReplicationMessage, challenge_id: u64, peer: &PeerId, - nonce: &[u8; 32], - challenge_material: &[(XorName, Vec)], + challenge_material: &[(XorName, [u8; 32])], ) -> Vec<(XorName, PruneAuditStatus)> { let failed_all = |reason: &str| { warn!( @@ -1348,7 +1359,7 @@ fn prune_audit_response_statuses( challenge_material .iter() .zip(digests.iter()) - .map(|((key, local_bytes), digest)| { + .map(|((key, expected_digest), digest)| { if *digest == ABSENT_KEY_DIGEST { warn!( "Prune audit proof from {peer} failed for {}: peer reports key absent", @@ -1356,7 +1367,7 @@ fn prune_audit_response_statuses( ); return (*key, PruneAuditStatus::Failed); } - if audit_digest_proves_key(peer, key, nonce, local_bytes, digest) { + if digest == expected_digest { (*key, PruneAuditStatus::Proven) } else { warn!( @@ -1405,6 +1416,17 @@ fn prune_audit_response_statuses( } } +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), @@ -1425,6 +1447,7 @@ async fn local_record_bytes(key: &XorName, storage: &Arc) -> Option } } +#[cfg(test)] fn audit_digest_proves_key( peer: &PeerId, key: &XorName, @@ -1639,13 +1662,13 @@ mod tests { let key_b = key_from_byte(0xB); let nonce = [0x7A; 32]; let bytes_a = b"record-a".to_vec(); - let bytes_b = b"record-b".to_vec(); - let digest_a = compute_audit_digest(&nonce, peer.as_bytes(), &key_a, &bytes_a); + 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![digest_a, ABSENT_KEY_DIGEST], + digests: vec![expected_a, ABSENT_KEY_DIGEST], }), }; @@ -1653,8 +1676,7 @@ mod tests { msg, 42, &peer, - &nonce, - &[(key_a, bytes_a), (key_b, bytes_b)], + &[(key_a, expected_a), (key_b, expected_b)], ); assert_eq!( From 57a11550d07a6be521cd055d14fcda741bfbfd16 Mon Sep 17 00:00:00 2001 From: Chris O'Neil Date: Mon, 13 Jul 2026 21:55:04 +0100 Subject: [PATCH 6/6] fix(replication): adapt merged prune-audit grading tests to batch API The merge of upstream/rc-2026.7.1 combined our branch's refactor (singular prune_audit_response_status replaced by the batch prune_audit_response_statuses) with upstream's grading tests, which still called the removed singular function. Adapt the graded_status test helper to delegate to the batch function with a single-key challenge slice so upstream's coverage is preserved. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/replication/pruning.rs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 4aca55d6..87bf1f26 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -2751,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]