From d04c5aa6c4943ecb17d040721235c00d3c2f4e1c Mon Sep 17 00:00:00 2001 From: grumbach Date: Thu, 11 Jun 2026 13:32:35 +0900 Subject: [PATCH 1/4] fix(pruning): tolerate one lagging peer in the prune proof gate Prune deletion required every peer in the current close group to prove possession, but fresh replication is fire-and-forget and uploads/repair only guarantee QUORUM_THRESHOLD placement (4 of 7). One lagging peer vetoed the prune forever, leaving prod nodes unable to free disk while audit-proof WARNs accumulated. A record that is out of the node's closest 7 now prunes once all but one of the current close group (6 of 7 at production parameters) prove they hold the exact bytes. This keeps a single absent peer from blocking deletion forever while still requiring near-full placement before the extra copy is dropped. Groups of one or two peers still require every proof. The mature-repair-proof precondition follows the same rule: a never-synced close-group peer reduces the audit pool instead of vetoing the prune, and only hinted peers are ever audited. PaidForList entry pruning gets its own remote gate: an out-of-range entry was previously removed on local state alone once the hysteresis elapsed; it is now retained until three quarters of the current paid close group (15 of 20 at production parameters) confirm the key in their own paid lists. Confirmations are revalidated against the current paid close group after the network round, the scan rotates through a dedicated cursor so capped passes cannot starve later entries, and chunk pruning checks only chunk possession while paid pruning checks only paid-list confirmations. Also split the failed-proof WARN into key-absent vs digest-mismatch so the two failure modes can be told apart in production logs. --- src/replication/mod.rs | 2 + src/replication/pruning.rs | 546 +++++++++++++++++++++++++++++++++---- src/replication/quorum.rs | 2 +- src/replication/types.rs | 4 + tests/e2e/replication.rs | 253 +++++++++++++++++ 5 files changed, 760 insertions(+), 47 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 81ea8ec5..a19fb57e 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -1695,11 +1695,13 @@ async fn run_neighbor_sync_round( let old_bootstrap_claims = std::mem::take(&mut state.bootstrap_claims); let old_bootstrap_claim_history = std::mem::take(&mut state.bootstrap_claim_history); let old_prune_cursor = state.prune_cursor; + let old_paid_prune_cursor = state.paid_prune_cursor; *state = NeighborSyncState::new_cycle(neighbors); state.last_sync_times = old_sync_times; state.bootstrap_claims = old_bootstrap_claims; state.bootstrap_claim_history = old_bootstrap_claim_history; state.prune_cursor = old_prune_cursor; + state.paid_prune_cursor = old_paid_prune_cursor; } } diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 4618ab09..19177d9a 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -26,13 +26,22 @@ use crate::replication::protocol::{ compute_audit_digest, AuditChallenge, AuditResponse, ReplicationMessage, ReplicationMessageBody, ABSENT_KEY_DIGEST, }; -use crate::replication::types::{BootstrapClaimObservation, NeighborSyncState, RepairProofs}; +use crate::replication::quorum::{self, VerificationTargets}; +use crate::replication::types::{ + BootstrapClaimObservation, KeyVerificationEvidence, NeighborSyncState, PaidListEvidence, + RepairProofs, +}; use crate::storage::LmdbStorage; use super::REPLICATION_TRUST_WEIGHT; const MAX_CONCURRENT_PRUNE_AUDIT_CHALLENGES: usize = 32; +/// Maximum expired `PaidForList` entries verified against the paid close +/// group per prune pass. Bounds the per-pass verification fan-out the same +/// way `MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS` bounds record audits. +const MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS: usize = 32; + // --------------------------------------------------------------------------- // Result type // --------------------------------------------------------------------------- @@ -140,13 +149,15 @@ struct PruneAuditReportState { /// For each stored record K: /// - If `IsResponsible(self, K)`: clear `RecordOutOfRangeFirstSeen`. /// - If not responsible: set timestamp if not already set; delete if the -/// timestamp is at least `PRUNE_HYSTERESIS_DURATION` old and the current -/// close group proves it stores the record. +/// timestamp is at least `PRUNE_HYSTERESIS_DURATION` old and all but one +/// of the current close group prove they store the record. /// /// For each `PaidForList` entry K: /// - If self is in `PaidCloseGroup(K)`: clear `PaidOutOfRangeFirstSeen`. /// - If not in group: set timestamp if not already set; remove entry if the -/// timestamp is at least `PRUNE_HYSTERESIS_DURATION` old. +/// timestamp is at least `PRUNE_HYSTERESIS_DURATION` old and three +/// quarters of the current paid close group (15 of 20 at production +/// parameters) confirm the key in their own `PaidForList`. /// /// Compatibility wrapper for callers that have not adopted repair-proof /// tracking. It preserves the original public signature, but it has no proof @@ -182,8 +193,15 @@ pub async fn run_prune_pass( pub async fn run_prune_pass_with_context(ctx: PrunePassContext<'_>) -> PruneResult { let (stored_count, record_stats) = prune_stored_records(&ctx).await; let now = Instant::now(); - let (paid_count, paid_stats) = - prune_paid_entries(ctx.self_id, ctx.paid_list, ctx.p2p_node, ctx.config, now).await; + let (paid_count, paid_stats) = prune_paid_entries( + ctx.self_id, + ctx.paid_list, + ctx.p2p_node, + ctx.config, + ctx.sync_state, + now, + ) + .await; let result = PruneResult { records_pruned: record_stats.pruned, @@ -346,32 +364,37 @@ async fn evaluate_record_prune_key( return outcome; } + // Only peers we have hinted (mature repair proof) may be audited; the + // proof threshold must be reachable among them. A never-synced peer in + // the close group reduces the audit pool instead of vetoing the prune. let current_close_peers: HashSet = closest.iter().map(|node| node.peer_id).collect(); - if !target_peers_have_mature_repair_proofs( + let audit_targets = peers_with_mature_repair_proofs( key, &target_peers, ¤t_close_peers, ctx.repair_proofs, ctx.current_sync_epoch, ) - .await - { + .await; + let proofs_needed = prune_proofs_needed(target_peers.len()); + if proofs_needed == 0 || audit_targets.len() < proofs_needed { debug!( - "Deferring prune for {} until current close group has mature repair proofs", + "Deferring prune for {} until enough of the close group has mature \ + repair proofs", hex::encode(key) ); return outcome; } - if target_peers.len() > *audit_challenge_budget { + if audit_targets.len() > *audit_challenge_budget { outcome.state = RecordPruneKeyState::BudgetDeferred; return outcome; } - *audit_challenge_budget -= target_peers.len(); + *audit_challenge_budget -= audit_targets.len(); outcome.state = RecordPruneKeyState::Candidate(RecordPruneCandidate { key: *key, - target_peers, + target_peers: audit_targets, }); outcome } @@ -381,6 +404,7 @@ async fn prune_paid_entries( paid_list: &Arc, p2p_node: &Arc, config: &ReplicationConfig, + sync_state: &Arc>, now: Instant, ) -> (usize, PaidPruneStats) { let paid_keys = match paid_list.all_keys() { @@ -393,9 +417,17 @@ async fn prune_paid_entries( let dht = p2p_node.dht_manager(); let mut stats = PaidPruneStats::default(); - let mut paid_keys_to_delete = Vec::new(); + let mut expired_candidates: Vec<(XorName, Vec)> = Vec::new(); + let mut verification_deferred = 0usize; + // Rotate the scan start so expired entries beyond the per-pass cap are + // not starved by the same head-of-list entries every pass. + let scan_start = paid_prune_scan_start(sync_state, paid_keys.len()).await; + let mut last_selected_offset = None; - for key in &paid_keys { + for offset in 0..paid_keys.len() { + let Some(key) = paid_keys.get((scan_start + offset) % paid_keys.len()) else { + continue; + }; let closest: Vec = dht .find_closest_nodes_local_with_self(key, config.paid_list_close_group_size) .await; @@ -417,12 +449,46 @@ async fn prune_paid_entries( .checked_duration_since(first_seen) .unwrap_or(Duration::ZERO); if elapsed >= config.prune_hysteresis_duration { - paid_keys_to_delete.push(*key); + if expired_candidates.len() < MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS { + let target_peers = remote_close_group_peers(&closest, self_id); + expired_candidates.push((*key, target_peers)); + last_selected_offset = Some(offset); + } else { + verification_deferred = verification_deferred.saturating_add(1); + } } } } } + advance_paid_prune_cursor( + sync_state, + paid_keys.len(), + scan_start, + last_selected_offset, + ) + .await; + + if verification_deferred > 0 { + debug!( + "Deferred {verification_deferred} expired PaidForList entries beyond the \ + per-pass verification cap ({MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS})" + ); + } + + let confirmed_by_key = + collect_paid_prune_confirmations(&expired_candidates, p2p_node, config).await; + let (paid_keys_to_delete, revalidated_cleared) = revalidated_paid_prune_keys( + &expired_candidates, + &confirmed_by_key, + self_id, + paid_list, + p2p_node, + config, + ) + .await; + stats.cleared += revalidated_cleared; + if !paid_keys_to_delete.is_empty() { match paid_list.remove_batch(&paid_keys_to_delete).await { Ok(count) => { @@ -438,6 +504,104 @@ async fn prune_paid_entries( (paid_keys.len(), stats) } +async fn paid_prune_scan_start( + sync_state: &Arc>, + paid_key_count: usize, +) -> usize { + if paid_key_count == 0 { + return 0; + } + sync_state.read().await.paid_prune_cursor % paid_key_count +} + +async fn advance_paid_prune_cursor( + sync_state: &Arc>, + paid_key_count: usize, + scan_start: usize, + last_selected_offset: Option, +) { + if paid_key_count == 0 { + sync_state.write().await.paid_prune_cursor = 0; + return; + } + + let advance_by = last_selected_offset.map_or(1, |offset| offset.saturating_add(1)); + sync_state.write().await.paid_prune_cursor = (scan_start + advance_by) % paid_key_count; +} + +/// Re-check each confirmed candidate against current local state before +/// deletion. +/// +/// The network round in [`collect_paid_prune_confirmations`] takes time; +/// the paid close group may have changed underneath it, including self +/// moving back into range. Mirrors [`revalidated_record_prune_keys`]: +/// confirmations only count from peers still in the current paid close +/// group, against a threshold computed from that current group. +async fn revalidated_paid_prune_keys( + expired_candidates: &[(XorName, Vec)], + confirmed_by_key: &HashMap>, + self_id: &PeerId, + paid_list: &Arc, + p2p_node: &Arc, + config: &ReplicationConfig, +) -> (Vec, usize) { + let dht = p2p_node.dht_manager(); + let mut keys_to_delete = Vec::new(); + let mut cleared = 0; + let now = Instant::now(); + + for (key, _) in expired_candidates { + let closest: Vec = dht + .find_closest_nodes_local_with_self(key, config.paid_list_close_group_size) + .await; + + if closest.iter().any(|n| n.peer_id == *self_id) { + if paid_list.paid_out_of_range_since(key).is_some() { + paid_list.clear_paid_out_of_range(key); + cleared += 1; + } + continue; + } + + let Some(first_seen) = paid_list.paid_out_of_range_since(key) else { + continue; + }; + let elapsed = now + .checked_duration_since(first_seen) + .unwrap_or(Duration::ZERO); + if elapsed < config.prune_hysteresis_duration { + continue; + } + + let current_target_peers = remote_close_group_peers(&closest, self_id); + if current_target_peers.is_empty() { + warn!( + "Cannot prune paid entry {}: current paid close group has no remote peers", + hex::encode(key) + ); + continue; + } + + let confirmations_needed = paid_prune_confirmations_needed(current_target_peers.len()); + if target_peers_reported_present( + key, + ¤t_target_peers, + confirmed_by_key, + confirmations_needed, + ) { + keys_to_delete.push(*key); + } else { + debug!( + "Deferring paid-entry prune for {} until enough of the current paid \ + close group confirm it", + hex::encode(key) + ); + } + } + + (keys_to_delete, cleared) +} + fn remote_close_group_peers(close_group: &[DHTNode], self_id: &PeerId) -> Vec { close_group .iter() @@ -446,17 +610,112 @@ fn remote_close_group_peers(close_group: &[DHTNode], self_id: &PeerId) -> Vec usize { + (3 * group_size).div_ceil(4) +} + +/// Ask the current paid close group whether they track each expired key in +/// their `PaidForList`, and return the confirming peers per key. +/// +/// The deletion decision happens afterwards in +/// [`revalidated_paid_prune_keys`], against the paid close group as it +/// stands once the network round has completed. +async fn collect_paid_prune_confirmations( + expired_candidates: &[(XorName, Vec)], + p2p_node: &Arc, + config: &ReplicationConfig, +) -> HashMap> { + if expired_candidates.is_empty() { + return HashMap::new(); + } + + let mut targets = VerificationTargets::default(); + let mut keys = Vec::new(); + for (key, target_peers) in expired_candidates { + if target_peers.is_empty() { + warn!( + "Cannot prune paid entry {}: current paid close group has no remote peers", + hex::encode(key) + ); + continue; + } + keys.push(*key); + for peer in target_peers { + targets.all_peers.insert(*peer); + targets.peer_to_keys.entry(*peer).or_default().push(*key); + targets + .peer_to_paid_keys + .entry(*peer) + .or_default() + .insert(*key); + } + targets.paid_targets.insert(*key, target_peers.clone()); + targets.paid_group_sizes.insert(*key, target_peers.len()); + } + for keys_list in targets.peer_to_keys.values_mut() { + keys_list.sort_unstable(); + keys_list.dedup(); + } + + let evidence = quorum::run_verification_round(&keys, &targets, p2p_node, config).await; + paid_confirmations_by_key(expired_candidates, &evidence) +} + +/// Aggregate `Confirmed` paid-list evidence into per-key peer sets. +/// +/// Only peers in the candidate's own target set count; `NotFound` and +/// `Unresolved` answers never confirm. +fn paid_confirmations_by_key( + expired_candidates: &[(XorName, Vec)], + evidence: &HashMap, +) -> HashMap> { + let mut confirmed_by_key: HashMap> = HashMap::new(); + for (key, target_peers) in expired_candidates { + let Some(key_evidence) = evidence.get(key) else { + continue; + }; + let confirmed: HashSet = key_evidence + .paid_list + .iter() + .filter(|(peer, status)| { + **status == PaidListEvidence::Confirmed && target_peers.contains(peer) + }) + .map(|(peer, _)| *peer) + .collect(); + if !confirmed.is_empty() { + confirmed_by_key.insert(*key, confirmed); + } + } + confirmed_by_key +} + +/// Filter `target_peers` down to those with a mature repair proof for `key`. +/// +/// Per design rule 20, peers without a key-specific mature repair hint proof +/// are never audited for that key. +async fn peers_with_mature_repair_proofs( key: &XorName, target_peers: &[PeerId], current_close_peers: &HashSet, repair_proofs: &Arc>, current_sync_epoch: u64, -) -> bool { +) -> Vec { let mut proofs = repair_proofs.write().await; - target_peers.iter().all(|peer| { - proofs.has_mature_replica_hint(peer, key, current_close_peers, current_sync_epoch) - }) + target_peers + .iter() + .filter(|peer| { + proofs.has_mature_replica_hint(peer, key, current_close_peers, current_sync_epoch) + }) + .copied() + .collect() } async fn prune_scan_start( @@ -512,10 +771,14 @@ async fn delete_stored_records( /// Collect positive presence reports for prune candidates. /// -/// Peers that fail to prove storage block pruning for their keys. The -/// retained local record continues to participate in normal neighbor-sync -/// repair because replica hint construction walks all locally stored keys, -/// including out-of-range keys retained by hysteresis. +/// A key is deleted once all but one of the current close group prove +/// possession ([`prune_proofs_needed`]). Requiring unanimous proofs left +/// out-of-range records undeletable whenever a single close-group peer +/// lagged, while the all-but-one threshold still demands more copies than +/// the storage quorum used elsewhere. Keys below the proof threshold stay +/// local, and the retained record continues to participate in normal +/// neighbor-sync repair because replica hint construction walks all locally +/// stored keys, including out-of-range keys retained by hysteresis. async fn collect_record_prune_proofs( candidates: &[RecordPruneCandidate], storage: &Arc, @@ -600,11 +863,18 @@ async fn revalidated_record_prune_keys( continue; } - if target_peers_reported_present(&candidate.key, ¤t_target_peers, present_by_key) { + let proofs_needed = prune_proofs_needed(current_target_peers.len()); + if target_peers_reported_present( + &candidate.key, + ¤t_target_peers, + present_by_key, + proofs_needed, + ) { keys_to_delete.push(candidate.key); } else { debug!( - "Deferring prune for {} until current close group reports it", + "Deferring prune for {} until all but one of the current close group \ + report it", hex::encode(candidate.key) ); } @@ -627,25 +897,63 @@ fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(Peer fn confirmed_keys_from_presence( candidates: &[RecordPruneCandidate], present_by_key: &HashMap>, + proofs_needed: usize, ) -> Vec { candidates .iter() .filter(|candidate| { - target_peers_reported_present(&candidate.key, &candidate.target_peers, present_by_key) + target_peers_reported_present( + &candidate.key, + &candidate.target_peers, + present_by_key, + proofs_needed, + ) }) .map(|candidate| candidate.key) .collect() } +/// Proofs required before deleting an out-of-range record: all but one of +/// the close group (6 of 7 at production parameters). +/// +/// Stricter than the storage quorum (`QuorumNeeded`) because pruning only +/// runs after `PRUNE_HYSTERESIS_DURATION` out of range, by which time many +/// sync cycles should have replicated the record across the whole close +/// group. Tolerating exactly one lagging peer keeps a single absent peer +/// from vetoing deletion forever without accepting under-replication. +/// Groups of one or two peers require every proof: tolerating a miss there +/// would allow deletion on a single attestation. +fn prune_proofs_needed(group_size: usize) -> usize { + if group_size <= 2 { + group_size + } else { + group_size - 1 + } +} + +/// Whether enough target peers proved possession to allow deletion. +/// +/// `proofs_needed == 0` means confirmation is impossible (no targets), not +/// trivially met. fn target_peers_reported_present( key: &XorName, target_peers: &[PeerId], present_by_key: &HashMap>, + proofs_needed: usize, ) -> bool { + if proofs_needed == 0 { + return false; + } let Some(present_peers) = present_by_key.get(key) else { return false; }; - target_peers.iter().all(|peer| present_peers.contains(peer)) + // Count distinct proven peers: iterating the present set keeps a + // duplicated entry in `target_peers` from being counted twice. + let proven = present_peers + .iter() + .filter(|peer| target_peers.contains(peer)) + .count(); + proven >= proofs_needed } /// Challenge a peer to prove it holds the exact record bytes for `key`. @@ -782,19 +1090,25 @@ fn prune_audit_response_status( warn!("Prune audit challenge ID mismatch from {peer}"); return PruneAuditStatus::Failed; } - if digests.len() != 1 { + let [digest] = digests.as_slice() else { warn!( "Prune audit response from {peer} returned {} digests for one challenged key", 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, &digests[0]) { + if audit_digest_proves_key(peer, key, nonce, local_bytes, digest) { PruneAuditStatus::Proven } else { warn!( - "Prune audit proof from {peer} failed for {}", + "Prune audit proof from {peer} failed for {}: digest mismatch", hex::encode(key) ); PruneAuditStatus::Failed @@ -1027,38 +1341,178 @@ mod tests { } #[test] - fn confirmed_keys_require_all_target_peers_present() { + fn confirmed_keys_require_quorum_of_target_peers_present() { let peer_a = peer_id_from_byte(1); let peer_b = peer_id_from_byte(2); + let peer_c = peer_id_from_byte(3); let key = key_from_byte(0xC); - let candidates = vec![candidate(key, vec![peer_a, peer_b])]; + let candidates = vec![candidate(key, vec![peer_a, peer_b, peer_c])]; let mut present_by_key = HashMap::new(); present_by_key.insert(key, HashSet::from([peer_a, peer_b])); - let confirmed = confirmed_keys_from_presence(&candidates, &present_by_key); - + // Two of three proofs meet a quorum of 2 even though one peer is + // missing — unanimity is not required. + let confirmed = confirmed_keys_from_presence(&candidates, &present_by_key, 2); assert_eq!(confirmed, vec![key]); + + // The same evidence fails a quorum of 3. + let confirmed = confirmed_keys_from_presence(&candidates, &present_by_key, 3); + assert!(confirmed.is_empty()); } #[test] - fn confirmed_keys_defer_absent_or_missing_peer_evidence() { + fn confirmed_keys_defer_below_quorum_or_missing_peer_evidence() { let peer_a = peer_id_from_byte(1); let peer_b = peer_id_from_byte(2); - let complete_key = key_from_byte(0xD); - let absent_key = key_from_byte(0xE); + let quorum_key = key_from_byte(0xD); + let below_quorum_key = key_from_byte(0xE); let missing_key = key_from_byte(0xF); let candidates = vec![ - candidate(complete_key, vec![peer_a, peer_b]), - candidate(absent_key, vec![peer_a, peer_b]), + candidate(quorum_key, vec![peer_a, peer_b]), + candidate(below_quorum_key, vec![peer_a, peer_b]), candidate(missing_key, vec![peer_a, peer_b]), ]; let mut present_by_key = HashMap::new(); - present_by_key.insert(complete_key, HashSet::from([peer_a, peer_b])); - present_by_key.insert(absent_key, HashSet::from([peer_a])); + present_by_key.insert(quorum_key, HashSet::from([peer_a, peer_b])); + present_by_key.insert(below_quorum_key, HashSet::from([peer_a])); + + let confirmed = confirmed_keys_from_presence(&candidates, &present_by_key, 2); - let confirmed = confirmed_keys_from_presence(&candidates, &present_by_key); + assert_eq!(confirmed, vec![quorum_key]); + } - assert_eq!(confirmed, vec![complete_key]); + #[test] + fn prune_proofs_needed_tolerates_exactly_one_lagging_peer() { + assert_eq!(prune_proofs_needed(0), 0); + // Tiny groups require every proof. + assert_eq!(prune_proofs_needed(1), 1); + assert_eq!(prune_proofs_needed(2), 2); + assert_eq!(prune_proofs_needed(3), 2); + assert_eq!(prune_proofs_needed(5), 4); + // Production close group: 6 of 7 proofs required. + assert_eq!(prune_proofs_needed(7), 6); + } + + #[test] + fn paid_prune_confirmations_are_three_quarters_rounded_up() { + assert_eq!(paid_prune_confirmations_needed(0), 0); + assert_eq!(paid_prune_confirmations_needed(1), 1); + assert_eq!(paid_prune_confirmations_needed(2), 2); + assert_eq!(paid_prune_confirmations_needed(4), 3); + // Production paid close group: 15 of 20 confirmations required. + assert_eq!(paid_prune_confirmations_needed(20), 15); + } + + #[test] + fn paid_confirmations_count_only_confirmed_target_peers() { + let confirmed_peer = peer_id_from_byte(1); + let not_found_peer = peer_id_from_byte(2); + let unresolved_peer = peer_id_from_byte(3); + let outsider = peer_id_from_byte(4); + let key = key_from_byte(0x21); + let candidates = vec![(key, vec![confirmed_peer, not_found_peer, unresolved_peer])]; + + let mut evidence = HashMap::new(); + evidence.insert( + key, + KeyVerificationEvidence { + presence: HashMap::new(), + paid_list: HashMap::from([ + (confirmed_peer, PaidListEvidence::Confirmed), + (not_found_peer, PaidListEvidence::NotFound), + (unresolved_peer, PaidListEvidence::Unresolved), + // Confirmation from a peer outside the target set. + (outsider, PaidListEvidence::Confirmed), + ]), + }, + ); + + let confirmed_by_key = paid_confirmations_by_key(&candidates, &evidence); + + assert_eq!( + confirmed_by_key.get(&key), + Some(&HashSet::from([confirmed_peer])), + "only Confirmed answers from target peers may count", + ); + } + + #[test] + fn paid_confirmations_skip_keys_without_evidence() { + let peer = peer_id_from_byte(1); + let key = key_from_byte(0x22); + let candidates = vec![(key, vec![peer])]; + + let confirmed_by_key = paid_confirmations_by_key(&candidates, &HashMap::new()); + + assert!(confirmed_by_key.is_empty()); + } + + #[tokio::test] + async fn paid_prune_cursor_advances_past_selected_window() { + let state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); + state.write().await.paid_prune_cursor = 2; + + let start = paid_prune_scan_start(&state, 10).await; + advance_paid_prune_cursor(&state, 10, start, Some(3)).await; + + assert_eq!(state.read().await.paid_prune_cursor, 6); + } + + #[tokio::test] + async fn paid_prune_cursor_advances_even_when_nothing_selected() { + let state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); + state.write().await.paid_prune_cursor = 9; + + let start = paid_prune_scan_start(&state, 10).await; + advance_paid_prune_cursor(&state, 10, start, None).await; + + assert_eq!(state.read().await.paid_prune_cursor, 0); + } + + #[test] + fn zero_quorum_never_confirms() { + let peer_a = peer_id_from_byte(1); + let key = key_from_byte(0x10); + let mut present_by_key = HashMap::new(); + present_by_key.insert(key, HashSet::from([peer_a])); + + assert!(!target_peers_reported_present( + &key, + &[peer_a], + &present_by_key, + 0 + )); + } + + #[test] + fn proofs_from_non_target_peers_do_not_count_toward_quorum() { + let target = peer_id_from_byte(1); + let outsider = peer_id_from_byte(2); + let key = key_from_byte(0x11); + let mut present_by_key = HashMap::new(); + present_by_key.insert(key, HashSet::from([outsider])); + + assert!(!target_peers_reported_present( + &key, + &[target], + &present_by_key, + 1 + )); + } + + #[test] + fn duplicated_target_peer_counts_once_toward_quorum() { + let peer = peer_id_from_byte(1); + let key = key_from_byte(0x12); + let mut present_by_key = HashMap::new(); + present_by_key.insert(key, HashSet::from([peer])); + + assert!(!target_peers_reported_present( + &key, + &[peer, peer], + &present_by_key, + 2 + )); } #[test] diff --git a/src/replication/quorum.rs b/src/replication/quorum.rs index 5f4d99af..4a3f1405 100644 --- a/src/replication/quorum.rs +++ b/src/replication/quorum.rs @@ -22,7 +22,7 @@ use crate::replication::types::{KeyVerificationEvidence, PaidListEvidence, Prese // --------------------------------------------------------------------------- /// Targets for verifying a set of keys. -#[derive(Debug)] +#[derive(Debug, Default)] pub struct VerificationTargets { /// Per-key: closest `CLOSE_GROUP_SIZE` peers (excluding self) for presence /// quorum. diff --git a/src/replication/types.rs b/src/replication/types.rs index ec74e76d..a2f60190 100644 --- a/src/replication/types.rs +++ b/src/replication/types.rs @@ -455,6 +455,9 @@ pub struct NeighborSyncState { /// Cursor used by post-cycle pruning to rotate through stored records when /// the per-pass prune-confirmation budget is exhausted. pub prune_cursor: usize, + /// Cursor used by post-cycle pruning to rotate through `PaidForList` + /// entries when the per-pass paid-verification cap is exhausted. + pub paid_prune_cursor: usize, } impl NeighborSyncState { @@ -468,6 +471,7 @@ impl NeighborSyncState { bootstrap_claims: HashMap::new(), bootstrap_claim_history: HashMap::new(), prune_cursor: 0, + paid_prune_cursor: 0, } } diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index 83fc792f..121cc8c0 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -626,6 +626,259 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { harness.teardown().await.expect("teardown"); } +/// The prune proof gate tolerates exactly one lagging close-group peer, at +/// production parameters (close group 7, 6 proofs required). +/// +/// Fresh replication is fire-and-forget and uploads/repair succeed at +/// `QUORUM_THRESHOLD` (4 of 7), so a record's placement routinely sits below +/// full unanimity. When the prune gate demanded unanimous proofs, such +/// records were audited every pass, failed (absent peers answer +/// `ABSENT_KEY_DIGEST`), and were never deleted — the production "pruning is +/// hardly taking place" incident. The all-but-one threshold keeps a single +/// absent peer from vetoing deletion while still demanding near-full +/// placement before the local copy is dropped. +/// +/// This test pins both sides of the gate: +/// - below the threshold (5 of 7 proofs) the record must never be deleted, +/// no matter how many passes run; +/// - at the threshold (6 of 7 proofs) the record prunes even though one +/// peer still lacks the bytes. +/// +/// Repair proofs are recorded only for the eventual holders: the +/// never-hinted peer must reduce the audit pool rather than veto the prune +/// at the repair-proof gate. +#[tokio::test] +#[serial] +async fn prune_deletes_at_proof_threshold_and_retains_below_it() { + const HINT_EPOCH: u64 = 7; + const CURRENT_EPOCH: u64 = HINT_EPOCH + 1; + /// Production close-group size (`CLOSE_GROUP_SIZE` in ant-protocol). + const PROD_CLOSE_GROUP_SIZE: usize = 7; + /// Prune proof threshold at production parameters: all but one, 6 of 7. + const PRUNE_PROOFS_NEEDED: usize = 6; + + let harness = TestHarness::setup_small().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let pruner_idx = 3; + let config = ReplicationConfig { + close_group_size: PROD_CLOSE_GROUP_SIZE, + paid_list_close_group_size: 1, + prune_hysteresis_duration: Duration::ZERO, + ..ReplicationConfig::default() + }; + let sync_state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); + let repair_proofs = Arc::new(RwLock::new(RepairProofs::new())); + + let pruner = harness.test_node(pruner_idx).expect("pruner"); + let pruner_p2p = Arc::clone(pruner.p2p_node.as_ref().expect("pruner p2p")); + let pruner_protocol = pruner.ant_protocol.as_ref().expect("pruner protocol"); + let pruner_storage = pruner_protocol.storage(); + let pruner_paid_list = Arc::clone( + pruner + .replication_engine + .as_ref() + .expect("pruner replication engine") + .paid_list(), + ); + let pruner_peer = *pruner_p2p.peer_id(); + + let (content, address, targets) = + find_remote_prune_candidate(&harness, pruner_idx, PROD_CLOSE_GROUP_SIZE, "quorum-stored") + .await; + pruner_storage + .put(&address, &content) + .await + .expect("put record on pruner"); + + // Replicate below the threshold: only 5 of 7 peers hold the bytes. + // Repair proofs cover only the eventual holders; the remaining + // close-group peer was never hinted and stays outside the audit pool. + store_record_on_peers( + &harness, + &targets[..PRUNE_PROOFS_NEEDED - 1], + &address, + &content, + ) + .await; + record_repair_proofs_for_peers( + &repair_proofs, + &pruner_p2p, + &config, + &targets[..PRUNE_PROOFS_NEEDED], + &address, + HINT_EPOCH, + ) + .await; + + // Below the threshold the local copy is load-bearing: deleting it would + // shrink the proven replica set past the prune safety bar, so every + // pass must retain it. + for pass in 0..3 { + let result = pruning::run_prune_pass_with_context(pruning::PrunePassContext { + self_id: &pruner_peer, + storage: &pruner_storage, + paid_list: &pruner_paid_list, + p2p_node: &pruner_p2p, + config: &config, + sync_state: &sync_state, + repair_proofs: &repair_proofs, + current_sync_epoch: CURRENT_EPOCH, + allow_remote_prune_audits: true, + }) + .await; + assert_eq!( + result.records_pruned, 0, + "pass {pass}: a record below the proof threshold must never prune", + ); + assert!( + pruner_storage.exists(&address).expect("exists"), + "pass {pass}: record should remain on the out-of-range node", + ); + } + + // One more holder reaches the threshold (6 of 7). The prune must now + // proceed even though one close-group peer still lacks the bytes: + // demanding unanimity left prod fleets unable to prune at all. + store_record_on_peer( + &harness, + targets + .get(PRUNE_PROOFS_NEEDED - 1) + .expect("threshold target"), + &address, + &content, + ) + .await; + let at_threshold = pruning::run_prune_pass_with_context(pruning::PrunePassContext { + self_id: &pruner_peer, + storage: &pruner_storage, + paid_list: &pruner_paid_list, + p2p_node: &pruner_p2p, + config: &config, + sync_state: &sync_state, + repair_proofs: &repair_proofs, + current_sync_epoch: CURRENT_EPOCH, + allow_remote_prune_audits: true, + }) + .await; + assert_eq!( + at_threshold.records_pruned, 1, + "a record proven on all but one of the close group must prune", + ); + assert!(!pruner_storage.exists(&address).expect("exists")); + + harness.teardown().await.expect("teardown"); +} + +/// Paid-list entry pruning requires confirmations from the current paid +/// close group (three quarters rounded up, 15 of 20 at production +/// parameters), independent of chunk possession. +/// +/// An out-of-range `PaidForList` entry used to be removed on local state +/// alone once the hysteresis elapsed. It is now retained until enough of +/// the current paid close group confirm they track the key in their own +/// paid lists, so a node never forgets an authorization the group has not +/// already absorbed. Chunk pruning and paid pruning check their own gates +/// only: this test stores no chunk anywhere. +/// +/// Run with a 2-peer paid close group, where the threshold is both peers. +#[tokio::test] +#[serial] +async fn paid_prune_requires_paid_close_group_confirmations() { + const PAID_GROUP: usize = 2; + + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let pruner_idx = 3; + let config = ReplicationConfig { + close_group_size: 2, + quorum_threshold: 1, + paid_list_close_group_size: PAID_GROUP, + prune_hysteresis_duration: Duration::ZERO, + ..ReplicationConfig::default() + }; + let sync_state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); + let repair_proofs = Arc::new(RwLock::new(RepairProofs::new())); + + let pruner = harness.test_node(pruner_idx).expect("pruner"); + let pruner_p2p = Arc::clone(pruner.p2p_node.as_ref().expect("pruner p2p")); + let pruner_protocol = pruner.ant_protocol.as_ref().expect("pruner protocol"); + let pruner_storage = pruner_protocol.storage(); + let pruner_peer = *pruner_p2p.peer_id(); + + // Standalone paid list so the engine's own paid state stays untouched. + let paid_dir = tempfile::tempdir().expect("tempdir"); + let paid_list = Arc::new( + ant_node::replication::paid_list::PaidList::new(paid_dir.path()) + .await + .expect("paid list"), + ); + + let (_content, address, targets) = + find_remote_prune_candidate(&harness, pruner_idx, PAID_GROUP, "paid-prune").await; + paid_list.insert(&address).await.expect("insert paid key"); + + // The paid close group does not track the key yet: the entry must be + // retained even though it is out of range and past hysteresis. + let unconfirmed = pruning::run_prune_pass_with_context(pruning::PrunePassContext { + self_id: &pruner_peer, + storage: &pruner_storage, + paid_list: &paid_list, + p2p_node: &pruner_p2p, + config: &config, + sync_state: &sync_state, + repair_proofs: &repair_proofs, + current_sync_epoch: 1, + allow_remote_prune_audits: true, + }) + .await; + assert_eq!( + unconfirmed.paid_entries_pruned, 0, + "a paid entry without paid close-group confirmations must never prune", + ); + assert!( + paid_list.contains(&address).expect("contains"), + "unconfirmed paid entry should remain tracked", + ); + + // Once the whole paid close group confirms the key in their paid lists, + // the entry prunes. + for peer in &targets { + let idx = node_index_for_peer(&harness, peer).expect("target in harness"); + let engine = harness + .test_node(idx) + .expect("target node") + .replication_engine + .as_ref() + .expect("target engine"); + engine.paid_list().insert(&address).await.expect("insert"); + } + + let confirmed = pruning::run_prune_pass_with_context(pruning::PrunePassContext { + self_id: &pruner_peer, + storage: &pruner_storage, + paid_list: &paid_list, + p2p_node: &pruner_p2p, + config: &config, + sync_state: &sync_state, + repair_proofs: &repair_proofs, + current_sync_epoch: 1, + allow_remote_prune_audits: true, + }) + .await; + assert_eq!( + confirmed.paid_entries_pruned, 1, + "a paid entry confirmed by the paid close group must prune", + ); + assert!( + !paid_list.contains(&address).expect("contains"), + "confirmed paid entry should be removed", + ); + + harness.teardown().await.expect("teardown"); +} + /// Fetch not-found returns `NotFound`. /// /// Request a key that does not exist on the target node and verify From 11976d9d953424ba42c5d07d409213ceeedc4c41 Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Thu, 11 Jun 2026 15:38:52 +0200 Subject: [PATCH 2/4] fix(pruning): keep paid prune cursor internal --- src/replication/mod.rs | 2 - src/replication/paid_list.rs | 80 ++++++++++++++++++++++++++++++++++++ src/replication/pruning.rs | 69 ++----------------------------- src/replication/types.rs | 4 -- 4 files changed, 84 insertions(+), 71 deletions(-) diff --git a/src/replication/mod.rs b/src/replication/mod.rs index a19fb57e..81ea8ec5 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -1695,13 +1695,11 @@ async fn run_neighbor_sync_round( let old_bootstrap_claims = std::mem::take(&mut state.bootstrap_claims); let old_bootstrap_claim_history = std::mem::take(&mut state.bootstrap_claim_history); let old_prune_cursor = state.prune_cursor; - let old_paid_prune_cursor = state.paid_prune_cursor; *state = NeighborSyncState::new_cycle(neighbors); state.last_sync_times = old_sync_times; state.bootstrap_claims = old_bootstrap_claims; state.bootstrap_claim_history = old_bootstrap_claim_history; state.prune_cursor = old_prune_cursor; - state.paid_prune_cursor = old_paid_prune_cursor; } } diff --git a/src/replication/paid_list.rs b/src/replication/paid_list.rs index 86760368..e329ff81 100644 --- a/src/replication/paid_list.rs +++ b/src/replication/paid_list.rs @@ -54,6 +54,9 @@ pub struct PaidList { /// In-memory: when each stored record first went out of /// storage-responsibility range. record_out_of_range: RwLock>, + /// Cursor used by paid-list pruning to rotate through expired entries when + /// the per-pass remote confirmation cap is exhausted. + paid_prune_cursor: RwLock, } impl PaidList { @@ -107,6 +110,7 @@ impl PaidList { db, paid_out_of_range: RwLock::new(HashMap::new()), record_out_of_range: RwLock::new(HashMap::new()), + paid_prune_cursor: RwLock::new(0), }; let count = paid_list.count()?; @@ -325,6 +329,36 @@ impl PaidList { self.record_out_of_range.read().get(key).copied() } + /// Starting offset for the next paid-list prune scan. + /// + /// LMDB iteration order is stable, so a bounded prune pass must rotate its + /// verification window or later expired entries can be starved behind + /// earlier unconfirmed entries. + pub(crate) fn paid_prune_scan_start(&self, paid_key_count: usize) -> usize { + if paid_key_count == 0 { + return 0; + } + + *self.paid_prune_cursor.read() % paid_key_count + } + + /// Advance the paid-list prune cursor after one pass. + pub(crate) fn advance_paid_prune_cursor( + &self, + paid_key_count: usize, + scan_start: usize, + last_selected_offset: Option, + ) { + let mut cursor = self.paid_prune_cursor.write(); + if paid_key_count == 0 { + *cursor = 0; + return; + } + + let advance_by = last_selected_offset.map_or(1, |offset| offset.saturating_add(1)); + *cursor = (scan_start + advance_by) % paid_key_count; + } + /// Remove multiple keys in a single write transaction. /// /// Also clears any in-memory out-of-range timestamps for removed keys. @@ -642,6 +676,52 @@ mod tests { assert_eq!(removed, 0); } + #[tokio::test] + async fn paid_prune_cursor_advances_past_selected_window() { + const PAID_KEY_COUNT: usize = 10; + const START_CURSOR: usize = 2; + const LAST_SELECTED_OFFSET: usize = 3; + const EXPECTED_CURSOR: usize = 6; + + let (pl, _temp) = create_test_paid_list().await; + *pl.paid_prune_cursor.write() = START_CURSOR; + + let scan_start = pl.paid_prune_scan_start(PAID_KEY_COUNT); + pl.advance_paid_prune_cursor(PAID_KEY_COUNT, scan_start, Some(LAST_SELECTED_OFFSET)); + + assert_eq!(*pl.paid_prune_cursor.read(), EXPECTED_CURSOR); + } + + #[tokio::test] + async fn paid_prune_cursor_advances_even_without_selected_entry() { + const PAID_KEY_COUNT: usize = 10; + const START_CURSOR: usize = 9; + const EXPECTED_CURSOR: usize = 0; + + let (pl, _temp) = create_test_paid_list().await; + *pl.paid_prune_cursor.write() = START_CURSOR; + + let scan_start = pl.paid_prune_scan_start(PAID_KEY_COUNT); + pl.advance_paid_prune_cursor(PAID_KEY_COUNT, scan_start, None); + + assert_eq!(*pl.paid_prune_cursor.read(), EXPECTED_CURSOR); + } + + #[tokio::test] + async fn paid_prune_cursor_resets_for_empty_paid_list() { + const STALE_CURSOR: usize = 7; + const EMPTY_PAID_KEY_COUNT: usize = 0; + const EXPECTED_CURSOR: usize = 0; + + let (pl, _temp) = create_test_paid_list().await; + *pl.paid_prune_cursor.write() = STALE_CURSOR; + + let scan_start = pl.paid_prune_scan_start(EMPTY_PAID_KEY_COUNT); + pl.advance_paid_prune_cursor(EMPTY_PAID_KEY_COUNT, scan_start, Some(STALE_CURSOR)); + + assert_eq!(*pl.paid_prune_cursor.read(), EXPECTED_CURSOR); + } + // -- Scenario tests ------------------------------------------------------- /// #50: Key goes out of range. `set_record_out_of_range` called. diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 19177d9a..6280946d 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -193,15 +193,8 @@ pub async fn run_prune_pass( pub async fn run_prune_pass_with_context(ctx: PrunePassContext<'_>) -> PruneResult { let (stored_count, record_stats) = prune_stored_records(&ctx).await; let now = Instant::now(); - let (paid_count, paid_stats) = prune_paid_entries( - ctx.self_id, - ctx.paid_list, - ctx.p2p_node, - ctx.config, - ctx.sync_state, - now, - ) - .await; + let (paid_count, paid_stats) = + prune_paid_entries(ctx.self_id, ctx.paid_list, ctx.p2p_node, ctx.config, now).await; let result = PruneResult { records_pruned: record_stats.pruned, @@ -404,7 +397,6 @@ async fn prune_paid_entries( paid_list: &Arc, p2p_node: &Arc, config: &ReplicationConfig, - sync_state: &Arc>, now: Instant, ) -> (usize, PaidPruneStats) { let paid_keys = match paid_list.all_keys() { @@ -421,7 +413,7 @@ async fn prune_paid_entries( let mut verification_deferred = 0usize; // Rotate the scan start so expired entries beyond the per-pass cap are // not starved by the same head-of-list entries every pass. - let scan_start = paid_prune_scan_start(sync_state, paid_keys.len()).await; + let scan_start = paid_list.paid_prune_scan_start(paid_keys.len()); let mut last_selected_offset = None; for offset in 0..paid_keys.len() { @@ -461,13 +453,7 @@ async fn prune_paid_entries( } } - advance_paid_prune_cursor( - sync_state, - paid_keys.len(), - scan_start, - last_selected_offset, - ) - .await; + paid_list.advance_paid_prune_cursor(paid_keys.len(), scan_start, last_selected_offset); if verification_deferred > 0 { debug!( @@ -504,31 +490,6 @@ async fn prune_paid_entries( (paid_keys.len(), stats) } -async fn paid_prune_scan_start( - sync_state: &Arc>, - paid_key_count: usize, -) -> usize { - if paid_key_count == 0 { - return 0; - } - sync_state.read().await.paid_prune_cursor % paid_key_count -} - -async fn advance_paid_prune_cursor( - sync_state: &Arc>, - paid_key_count: usize, - scan_start: usize, - last_selected_offset: Option, -) { - if paid_key_count == 0 { - sync_state.write().await.paid_prune_cursor = 0; - return; - } - - let advance_by = last_selected_offset.map_or(1, |offset| offset.saturating_add(1)); - sync_state.write().await.paid_prune_cursor = (scan_start + advance_by) % paid_key_count; -} - /// Re-check each confirmed candidate against current local state before /// deletion. /// @@ -1447,28 +1408,6 @@ mod tests { assert!(confirmed_by_key.is_empty()); } - #[tokio::test] - async fn paid_prune_cursor_advances_past_selected_window() { - let state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); - state.write().await.paid_prune_cursor = 2; - - let start = paid_prune_scan_start(&state, 10).await; - advance_paid_prune_cursor(&state, 10, start, Some(3)).await; - - assert_eq!(state.read().await.paid_prune_cursor, 6); - } - - #[tokio::test] - async fn paid_prune_cursor_advances_even_when_nothing_selected() { - let state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); - state.write().await.paid_prune_cursor = 9; - - let start = paid_prune_scan_start(&state, 10).await; - advance_paid_prune_cursor(&state, 10, start, None).await; - - assert_eq!(state.read().await.paid_prune_cursor, 0); - } - #[test] fn zero_quorum_never_confirms() { let peer_a = peer_id_from_byte(1); diff --git a/src/replication/types.rs b/src/replication/types.rs index a2f60190..ec74e76d 100644 --- a/src/replication/types.rs +++ b/src/replication/types.rs @@ -455,9 +455,6 @@ pub struct NeighborSyncState { /// Cursor used by post-cycle pruning to rotate through stored records when /// the per-pass prune-confirmation budget is exhausted. pub prune_cursor: usize, - /// Cursor used by post-cycle pruning to rotate through `PaidForList` - /// entries when the per-pass paid-verification cap is exhausted. - pub paid_prune_cursor: usize, } impl NeighborSyncState { @@ -471,7 +468,6 @@ impl NeighborSyncState { bootstrap_claims: HashMap::new(), bootstrap_claim_history: HashMap::new(), prune_cursor: 0, - paid_prune_cursor: 0, } } From 1570733d9887c728b013e850f314a87269e86109 Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Thu, 11 Jun 2026 17:19:06 +0200 Subject: [PATCH 3/4] fix(pruning): bound paid prune verification --- src/replication/pruning.rs | 237 ++++++++++++++++++++++++++++++++----- 1 file changed, 205 insertions(+), 32 deletions(-) diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 6280946d..ff97f1f5 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -37,10 +37,12 @@ use super::REPLICATION_TRUST_WEIGHT; const MAX_CONCURRENT_PRUNE_AUDIT_CHALLENGES: usize = 32; -/// Maximum expired `PaidForList` entries verified against the paid close -/// group per prune pass. Bounds the per-pass verification fan-out the same -/// way `MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS` bounds record audits. +/// Maximum expired `PaidForList` entries selected for verification per prune +/// pass. The unique peer fan-out for those entries is capped separately. const MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS: usize = 32; +/// Maximum unique peers contacted for paid-list verification per prune pass. +/// `quorum::run_verification_round` sends one request per target peer. +const MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS: usize = MAX_CONCURRENT_PRUNE_AUDIT_CHALLENGES; // --------------------------------------------------------------------------- // Result type @@ -106,6 +108,41 @@ struct PaidPruneStats { pruned: usize, } +#[derive(Debug, Default)] +struct PaidPruneDeferredCounts { + entry_budget: usize, + remote_gate: usize, + peer_budget: usize, +} + +impl PaidPruneDeferredCounts { + fn log(&self) { + if self.entry_budget > 0 { + debug!( + "Deferred {} expired PaidForList entries beyond the per-pass verification cap \ + ({MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS})", + self.entry_budget, + ); + } + + if self.remote_gate > 0 { + debug!( + "Deferred {} expired PaidForList entries until bootstrap drain allows remote \ + paid-prune verification", + self.remote_gate, + ); + } + + if self.peer_budget > 0 { + debug!( + "Deferred {} expired PaidForList entries beyond the per-pass paid-prune peer cap \ + ({MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS})", + self.peer_budget, + ); + } + } +} + #[derive(Debug, Clone)] struct RecordPruneCandidate { key: XorName, @@ -134,6 +171,14 @@ enum RecordPruneKeyState { Candidate(RecordPruneCandidate), } +enum PaidPruneKeyState { + None, + RemoteDeferred, + EntryBudgetDeferred, + PeerBudgetDeferred, + Candidate(Vec), +} + #[derive(Default)] struct PruneAuditReportState { audit_failures: RwLock>, @@ -193,8 +238,15 @@ pub async fn run_prune_pass( pub async fn run_prune_pass_with_context(ctx: PrunePassContext<'_>) -> PruneResult { let (stored_count, record_stats) = prune_stored_records(&ctx).await; let now = Instant::now(); - let (paid_count, paid_stats) = - prune_paid_entries(ctx.self_id, ctx.paid_list, ctx.p2p_node, ctx.config, now).await; + let (paid_count, paid_stats) = prune_paid_entries( + ctx.self_id, + ctx.paid_list, + ctx.p2p_node, + ctx.config, + now, + ctx.allow_remote_prune_audits, + ) + .await; let result = PruneResult { records_pruned: record_stats.pruned, @@ -398,6 +450,7 @@ async fn prune_paid_entries( p2p_node: &Arc, config: &ReplicationConfig, now: Instant, + allow_remote_prune_audits: bool, ) -> (usize, PaidPruneStats) { let paid_keys = match paid_list.all_keys() { Ok(keys) => keys, @@ -410,7 +463,8 @@ async fn prune_paid_entries( let dht = p2p_node.dht_manager(); let mut stats = PaidPruneStats::default(); let mut expired_candidates: Vec<(XorName, Vec)> = Vec::new(); - let mut verification_deferred = 0usize; + let mut deferred_counts = PaidPruneDeferredCounts::default(); + let mut selected_verification_peers = HashSet::new(); // Rotate the scan start so expired entries beyond the per-pass cap are // not starved by the same head-of-list entries every pass. let scan_start = paid_list.paid_prune_scan_start(paid_keys.len()); @@ -441,12 +495,31 @@ async fn prune_paid_entries( .checked_duration_since(first_seen) .unwrap_or(Duration::ZERO); if elapsed >= config.prune_hysteresis_duration { - if expired_candidates.len() < MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS { - let target_peers = remote_close_group_peers(&closest, self_id); - expired_candidates.push((*key, target_peers)); - last_selected_offset = Some(offset); - } else { - verification_deferred = verification_deferred.saturating_add(1); + match select_paid_prune_candidate( + key, + &closest, + self_id, + allow_remote_prune_audits, + expired_candidates.len(), + &mut selected_verification_peers, + ) { + PaidPruneKeyState::None => {} + PaidPruneKeyState::RemoteDeferred => { + deferred_counts.remote_gate = + deferred_counts.remote_gate.saturating_add(1); + } + PaidPruneKeyState::EntryBudgetDeferred => { + deferred_counts.entry_budget = + deferred_counts.entry_budget.saturating_add(1); + } + PaidPruneKeyState::PeerBudgetDeferred => { + deferred_counts.peer_budget = + deferred_counts.peer_budget.saturating_add(1); + } + PaidPruneKeyState::Candidate(target_peers) => { + expired_candidates.push((*key, target_peers)); + last_selected_offset = Some(offset); + } } } } @@ -454,13 +527,7 @@ async fn prune_paid_entries( } paid_list.advance_paid_prune_cursor(paid_keys.len(), scan_start, last_selected_offset); - - if verification_deferred > 0 { - debug!( - "Deferred {verification_deferred} expired PaidForList entries beyond the \ - per-pass verification cap ({MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS})" - ); - } + deferred_counts.log(); let confirmed_by_key = collect_paid_prune_confirmations(&expired_candidates, p2p_node, config).await; @@ -474,20 +541,58 @@ async fn prune_paid_entries( ) .await; stats.cleared += revalidated_cleared; + stats.pruned = delete_paid_entries(&paid_keys_to_delete, paid_list).await; - if !paid_keys_to_delete.is_empty() { - match paid_list.remove_batch(&paid_keys_to_delete).await { - Ok(count) => { - stats.pruned = count; - debug!("Pruned {count} out-of-range PaidForList entries"); - } - Err(e) => { - warn!("Failed to prune PaidForList entries: {e}"); - } - } + (paid_keys.len(), stats) +} + +fn select_paid_prune_candidate( + key: &XorName, + closest: &[DHTNode], + self_id: &PeerId, + allow_remote_prune_audits: bool, + selected_candidate_count: usize, + selected_verification_peers: &mut HashSet, +) -> PaidPruneKeyState { + if !allow_remote_prune_audits { + return PaidPruneKeyState::RemoteDeferred; } - (paid_keys.len(), stats) + let target_peers = remote_close_group_peers(closest, self_id); + if target_peers.is_empty() { + warn!( + "Cannot prune paid entry {}: current paid close group has no remote peers", + hex::encode(key) + ); + return PaidPruneKeyState::None; + } + + if selected_candidate_count >= MAX_PAID_PRUNE_VERIFICATIONS_PER_PASS { + return PaidPruneKeyState::EntryBudgetDeferred; + } + + if !reserve_paid_prune_peer_budget(&target_peers, selected_verification_peers) { + return PaidPruneKeyState::PeerBudgetDeferred; + } + + PaidPruneKeyState::Candidate(target_peers) +} + +async fn delete_paid_entries(keys_to_delete: &[XorName], paid_list: &Arc) -> usize { + if keys_to_delete.is_empty() { + return 0; + } + + match paid_list.remove_batch(keys_to_delete).await { + Ok(count) => { + debug!("Pruned {count} out-of-range PaidForList entries"); + count + } + Err(e) => { + warn!("Failed to prune PaidForList entries: {e}"); + 0 + } + } } /// Re-check each confirmed candidate against current local state before @@ -583,6 +688,26 @@ fn paid_prune_confirmations_needed(group_size: usize) -> usize { (3 * group_size).div_ceil(4) } +fn reserve_paid_prune_peer_budget( + target_peers: &[PeerId], + selected_verification_peers: &mut HashSet, +) -> bool { + let new_peer_count = target_peers + .iter() + .filter(|peer| !selected_verification_peers.contains(peer)) + .count(); + if selected_verification_peers + .len() + .saturating_add(new_peer_count) + > MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS + { + return false; + } + + selected_verification_peers.extend(target_peers.iter().copied()); + true +} + /// Ask the current paid close group whether they track each expired key in /// their `PaidForList`, and return the confirming peers per key. /// @@ -646,8 +771,8 @@ fn paid_confirmations_by_key( let confirmed: HashSet = key_evidence .paid_list .iter() - .filter(|(peer, status)| { - **status == PaidListEvidence::Confirmed && target_peers.contains(peer) + .filter(|&(peer, status)| { + *status == PaidListEvidence::Confirmed && target_peers.contains(peer) }) .map(|(peer, _)| *peer) .collect(); @@ -1278,6 +1403,12 @@ mod tests { [b; 32] } + fn peer_ids(count: usize) -> Vec { + (0..count) + .map(|idx| peer_id_from_byte(u8::try_from(idx + 1).expect("peer byte"))) + .collect() + } + fn candidate(key: XorName, target_peers: Vec) -> RecordPruneCandidate { RecordPruneCandidate { key, target_peers } } @@ -1364,6 +1495,48 @@ mod tests { assert_eq!(paid_prune_confirmations_needed(20), 15); } + #[test] + fn paid_prune_peer_budget_allows_overlapping_targets() { + let peers = peer_ids(MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS); + let mut selected_peers = HashSet::new(); + + assert!(reserve_paid_prune_peer_budget(&peers, &mut selected_peers)); + assert_eq!( + selected_peers.len(), + MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS, + ); + + let overlapping_targets = vec![peers[0], peers[1]]; + assert!(reserve_paid_prune_peer_budget( + &overlapping_targets, + &mut selected_peers, + )); + assert_eq!( + selected_peers.len(), + MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS, + ); + } + + #[test] + fn paid_prune_peer_budget_rejects_new_peers_past_cap() { + let peers = peer_ids(MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS + 1); + let mut selected_peers = HashSet::new(); + + assert!(reserve_paid_prune_peer_budget( + &peers[..MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS], + &mut selected_peers, + )); + assert!(!reserve_paid_prune_peer_budget( + &peers[MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS..], + &mut selected_peers, + )); + assert_eq!( + selected_peers.len(), + MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS, + ); + assert!(!selected_peers.contains(&peers[MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS])); + } + #[test] fn paid_confirmations_count_only_confirmed_target_peers() { let confirmed_peer = peer_id_from_byte(1); From 479aceda4c7f0e29f6252d15b9f8e6ec9b35eb51 Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Thu, 11 Jun 2026 17:59:29 +0200 Subject: [PATCH 4/4] chore(pruning): clean up paid prune review nits --- src/replication/pruning.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index ff97f1f5..d72a7504 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -471,9 +471,7 @@ async fn prune_paid_entries( let mut last_selected_offset = None; for offset in 0..paid_keys.len() { - let Some(key) = paid_keys.get((scan_start + offset) % paid_keys.len()) else { - continue; - }; + let key = &paid_keys[(scan_start + offset) % paid_keys.len()]; let closest: Vec = dht .find_closest_nodes_local_with_self(key, config.paid_list_close_group_size) .await; @@ -1017,7 +1015,7 @@ fn prune_proofs_needed(group_size: usize) -> usize { } } -/// Whether enough target peers proved possession to allow deletion. +/// Whether enough target peers supplied positive evidence to allow deletion. /// /// `proofs_needed == 0` means confirmation is impossible (no targets), not /// trivially met.