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 4618ab09..d72a7504 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -26,13 +26,24 @@ 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 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 // --------------------------------------------------------------------------- @@ -97,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, @@ -125,6 +171,14 @@ enum RecordPruneKeyState { Candidate(RecordPruneCandidate), } +enum PaidPruneKeyState { + None, + RemoteDeferred, + EntryBudgetDeferred, + PeerBudgetDeferred, + Candidate(Vec), +} + #[derive(Default)] struct PruneAuditReportState { audit_failures: RwLock>, @@ -140,13 +194,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 +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, @@ -346,32 +409,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 } @@ -382,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, @@ -393,9 +462,16 @@ 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 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()); + let mut last_selected_offset = None; - for key in &paid_keys { + for offset in 0..paid_keys.len() { + 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; @@ -417,25 +493,177 @@ 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); + 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); + } + } } } } } - 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_list.advance_paid_prune_cursor(paid_keys.len(), scan_start, last_selected_offset); + deferred_counts.log(); + + 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; + stats.pruned = delete_paid_entries(&paid_keys_to_delete, paid_list).await; + + (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; + } + + 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 +/// 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) + ); } } - (paid_keys.len(), stats) + (keys_to_delete, cleared) } fn remote_close_group_peers(close_group: &[DHTNode], self_id: &PeerId) -> Vec { @@ -446,17 +674,132 @@ fn remote_close_group_peers(close_group: &[DHTNode], self_id: &PeerId) -> Vec 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. +/// +/// 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 +855,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 +947,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 +981,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 supplied positive evidence 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 +1174,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 @@ -1003,6 +1401,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 } } @@ -1027,38 +1431,198 @@ 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); + + assert_eq!(confirmed, vec![quorum_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_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, + ); + } - let confirmed = confirmed_keys_from_presence(&candidates, &present_by_key); + #[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_eq!(confirmed, vec![complete_key]); + 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); + 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()); + } + + #[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/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