diff --git a/docs/REPLICATION_DESIGN.md b/docs/REPLICATION_DESIGN.md index 1f45be20..8d83094e 100644 --- a/docs/REPLICATION_DESIGN.md +++ b/docs/REPLICATION_DESIGN.md @@ -105,9 +105,17 @@ Parameter safety constraints (MUST hold): 17. A `PaidNotify(K)` only whitelists key `K` after receiver-side proof verification succeeds; sender assertions never whitelist by themselves. 18. Neighbor-sync paid hints are non-authoritative and carry no PoP; receivers MUST only whitelist by paid-list majority verification (`>= ConfirmNeeded(K)`) or close-group replica majority (Section 7.2 rule 4), never by hint claims alone. Paid-hint-only processing MAY enqueue record fetch only after authorization succeeds and `IsResponsible(self, K)` is true; otherwise it updates `PaidForList(self)` only. 19. Storage-proof audits start only after `BootstrapDrained(self)` becomes true. -20. Storage-proof audits target only peers derived from closest-peer lookups for sampled local keys, filtered through local authenticated routing state (`LocalRT(self)`), and further filtered to `(peer, key)` pairs for which mature `RepairProof` and `RepairOpportunity` hold; random global peers, never-synced peers, peers without a key-specific repair hint proof, peers whose key proof predates a close-group change, and peers whose proof was cleared by routing-table removal are never audited for that key. +20. Responsible-chunk storage audits (the general anti-outsourcing audit tick, Section 15) target only peers derived from closest-peer lookups for sampled local keys, filtered through local authenticated routing state (`LocalRT(self)`), and further filtered to `(peer, key)` pairs for which mature `RepairProof` and `RepairOpportunity` hold; random global peers, never-synced peers, peers without a key-specific repair hint proof, peers whose key proof predates a close-group change, and peers whose proof was cleared by routing-table removal are never audited for that key. This repair-proof prerequisite applies to the responsible-chunk audit ONLY: stored-record prune candidacy and prune-confirmation target selection (rule 22) never depend on `RepairProof` or `RepairOpportunity` — prune confirmation concerns the peers currently closest to the KEY, while repair hints are sent to the peers closest to SELF, and nothing guarantees those sets overlap for an out-of-range key. 21. Verification-request batching is mandatory for unknown-key neighbor-sync verification and preserves per-key quorum semantics: each key receives explicit per-key evidence, and missing/timeout evidence is unresolved per key. -22. On every `NeighborSyncCycleComplete(self)`, node MUST run a prune pass using current `SelfInclusiveRT(self)`: for stored records where `IsResponsible(self, K)` is false, record `RecordOutOfRangeFirstSeen` if not already set and delete only when `now - RecordOutOfRangeFirstSeen >= PRUNE_HYSTERESIS_DURATION`, every currently-known `CloseGroup(K)` member has a mature `RepairProof(peer, K)` tied to the current close-group snapshot, and every currently-known `CloseGroup(K)` member returns a positive nonce-bound audit proof for the record. If any close-group peer lacks mature repair proof or does not prove storage, pruning is deferred and the retained local record continues participating in normal neighbor-sync repair. Prune-confirmation failures emit trust penalties after fresh responsibility confirmation and key-specific mature repair proof; first-time bootstrap claims use the same one-time `BOOTSTRAP_CLAIM_GRACE_PERIOD` tracking before abuse penalties apply. Clear `RecordOutOfRangeFirstSeen` when back in range. For `PaidForList` entries where `self ∉ PaidCloseGroup(K)`, record `PaidOutOfRangeFirstSeen` if not already set and delete only when `now - PaidOutOfRangeFirstSeen >= PRUNE_HYSTERESIS_DURATION`; clear `PaidOutOfRangeFirstSeen` when back in range. The two timestamps are independent. +22. On every `NeighborSyncCycleComplete(self)`, node MUST run a prune pass over all stored records and `PaidForList` entries using the current local routing table. For stored records, retention responsibility is evaluated at the storage-retention width `CLOSE_GROUP_SIZE + STORAGE_ADMISSION_MARGIN` (9 at reference parameters): + - In range: clear `RecordOutOfRangeFirstSeen(self, K)`. + - Out of range: set `RecordOutOfRangeFirstSeen(self, K)` to `now` if not already set — immediately, even when an older retained commitment currently contains `K` (commitment retention vetoes deletion, never the start of hysteresis). The timestamp is process-local and need not survive restarts. + - Once continuously out of range for `>= PRUNE_HYSTERESIS_DURATION`, the record is a prune candidate UNCONDITIONALLY: candidacy never depends on repair-hint proofs, bootstrap state, audit budget, or prior neighbor-sync contact. + - Candidates are processed subject to operational scheduling: bootstrap state (rule 19) may defer the prune audit, the bounded per-pass challenge budget may defer the prune audit, and a retained answerable commitment vetoes deletion. These are deferrals only — they never remove candidate status or restart hysteresis. + - Prune confirmation is a direct current-close-group possession check: challenge the current strict `CloseGroup(K)` (taken directly from the local routing table, never filtered through `RepairProofs` or prior neighbor-sync hints) with the nonce-bound audit proof, and delete the local record only when all but one of the current close group (6 of 7 at reference parameters) return valid positive possession proofs. Missing, timed-out, bootstrapping, malformed, absent, and digest-mismatching responses never count positively. Below the threshold, the record, its timestamp, and its candidacy are retained and retried on later passes, and the retained record continues participating in normal neighbor-sync repair. + - Immediately before deletion, revalidate against the current routing table: the key must still be outside the retention width, must not be held by an answerable retained commitment, and the positive reports must still satisfy the CURRENT strict close group (membership churn invalidates stale reports). + - Prune-confirmation failures emit trust penalties only after fresh responsibility confirmation; first-time bootstrap claims use the same one-time `BOOTSTRAP_CLAIM_GRACE_PERIOD` tracking before abuse penalties apply. + For `PaidForList` entries where `self ∉ PaidCloseGroup(K)`, record `PaidOutOfRangeFirstSeen` if not already set and delete only when `now - PaidOutOfRangeFirstSeen >= PRUNE_HYSTERESIS_DURATION` and three quarters of the current paid close group (15 of 20 at reference parameters) confirm the key in their own paid lists; clear `PaidOutOfRangeFirstSeen` when back in range. The two timestamps are independent. 23. Peers claiming bootstrap status are skipped for sync and audit without penalty for up to `BOOTSTRAP_CLAIM_GRACE_PERIOD` from first observation. After the grace period, each continued bootstrap claim emits `BootstrapClaimAbuse` evidence to `TrustEngine` (via `report_trust_event` with `ApplicationFailure(weight)`). If a peer stops claiming bootstrap and later claims bootstrap again, the repeated claim emits `BootstrapClaimAbuse` immediately; the grace period is not reset. 24. Audit trust-penalty signals require responsibility confirmation: on audit failure, challenger MUST perform fresh local RT closest-peer lookups for each challenged key and only penalize the peer for keys where it is confirmed responsible. @@ -361,15 +369,25 @@ This check is evaluated per-key at decision points: 2. Post-cycle pruning eligibility (prune stored records where node is no longer responsible). 3. Post-cycle paid-list retention eligibility (drop `PaidForList` entries for keys where node is no longer in `PaidCloseGroup(K)`). +Distinct widths and mechanisms govern a stored record's lifecycle; they must not be conflated (reference profile values in parentheses): + +- **Storage admission**: how close self must be to accept a write. The client PUT self-closeness gate accepts up to the K-bucket width (20); neighbor-sync replica-hint admission uses the storage-admission width `CLOSE_GROUP_SIZE + STORAGE_ADMISSION_MARGIN` (9). +- **Retained storage responsibility**: a stored record is retained while self is within the storage-retention width `CLOSE_GROUP_SIZE + STORAGE_ADMISSION_MARGIN` (9) of the key. +- **Commitment inclusion**: storage commitments cover keys for which self is in the strict `CLOSE_GROUP_SIZE` (7) closest, so only a subset of physically stored chunks appears in commitments — being uncommitted is normal for retained-but-not-strictly-close records. +- **Prune candidacy**: a stored record continuously outside the retention width (9) for `PRUNE_HYSTERESIS_DURATION` is a prune candidate, unconditionally. +- **Prune-audit scheduling**: candidates are audited subject to bootstrap drain and the bounded per-pass challenge budget; a retained answerable commitment vetoes deletion. All of these are deferrals — none removes candidacy or restarts hysteresis. +- **Prune deletion confirmation**: a direct possession check of the current strict `CloseGroup(K)` (7), requiring all-but-one (6 of 7) valid positive proofs; targets come straight from the local routing table. +- **General anti-outsourcing storage audits** (Section 15): a separate mechanism that retains its mature repair-proof prerequisite (rule 20). Prune candidacy and prune-confirmation target selection never consult repair proofs. + Post-cycle responsibility pruning (triggered by `NeighborSyncCycleComplete(self)`): -1. For each locally stored key `K`, recompute `IsResponsible(self, K)` using current `SelfInclusiveRT(self)`: +1. For each locally stored key `K`, recompute retention responsibility (storage-retention width) using current `SelfInclusiveRT(self)`: a. If in range: clear `RecordOutOfRangeFirstSeen(self, K)` (set to `None`). - b. If out of range: if `RecordOutOfRangeFirstSeen(self, K)` is `None`, set it to `now`. Once the hysteresis duration has elapsed, request a nonce-bound audit proof only from current `CloseGroup(K)` peers for which a mature `RepairProof(peer, K)` exists for the current close-group snapshot, subject to a bounded per-pass prune-confirmation budget. If any current close-group peer lacks mature repair proof or does not prove storage, defer pruning and emit trust penalties only after fresh responsibility confirmation plus key-specific mature repair proof; first-time bootstrap claims are penalized only after the bootstrap grace period, while repeated bootstrap claims are penalized immediately. Delete the local record only when `now - RecordOutOfRangeFirstSeen(self, K) >= PRUNE_HYSTERESIS_DURATION`, every current close-group peer has mature repair proof for `K`, and every current close-group peer proves storage for `K`. + b. If out of range: if `RecordOutOfRangeFirstSeen(self, K)` is `None`, set it to `now` — immediately, even if a retained commitment currently contains `K`. Once the hysteresis duration has elapsed, `K` is a prune candidate. Request a nonce-bound audit proof from the current strict `CloseGroup(K)` peers (taken directly from the local routing table, never filtered through repair proofs), subject to a bounded per-pass prune-confirmation budget; bootstrap state may defer the audit and a retained answerable commitment vetoes deletion, without removing candidacy. Delete the local record only when all but one of the current close group (6 of 7) return valid positive possession proofs AND an immediate pre-deletion revalidation confirms `K` is still out of range, not held by an answerable retained commitment, and the positive reports still satisfy the current close group. Anything less retains the record, its timestamp, and its candidacy for later passes. Trust penalties for failed prune proofs are emitted only after fresh responsibility confirmation; first-time bootstrap claims are penalized only after the bootstrap grace period, while repeated bootstrap claims are penalized immediately. 2. For each key `K` in `PaidForList(self)`, recompute `PaidCloseGroup(K)` membership using current `SelfInclusiveRT(self)`: a. If `self ∈ PaidCloseGroup(K)`: clear `PaidOutOfRangeFirstSeen(self, K)` (set to `None`). - b. If `self ∉ PaidCloseGroup(K)`: if `PaidOutOfRangeFirstSeen(self, K)` is `None`, set it to `now`. Delete the entry only when `now - PaidOutOfRangeFirstSeen(self, K) >= PRUNE_HYSTERESIS_DURATION`. -3. Paid-list entry pruning remains local-state-only and does not require remote confirmations. + b. If `self ∉ PaidCloseGroup(K)`: if `PaidOutOfRangeFirstSeen(self, K)` is `None`, set it to `now`. Delete the entry only when `now - PaidOutOfRangeFirstSeen(self, K) >= PRUNE_HYSTERESIS_DURATION` and three quarters of the current paid close group (15 of 20) confirm the key in their own paid lists. +3. Paid-list entry pruning is gated on paid-list confirmations from the current paid close group; it never requires chunk-possession proofs (and chunk pruning never requires paid-list confirmations). Effect: @@ -612,7 +630,7 @@ Each scenario should assert exact expected outcomes and state transitions. 35. Neighbor-sync round-robin batch selection with cooldown skip: - With more than `NEIGHBOR_SYNC_PEER_COUNT` eligible peers, consecutive rounds scan forward from cursor, skip and remove cooldown peers, and sync the next batch of up to `NEIGHBOR_SYNC_PEER_COUNT` non-cooldown peers. Cycle completes when all snapshot peers have been synced, skipped (cooldown), or removed (unreachable). 36. Post-cycle responsibility pruning with time-based hysteresis: -- When a full neighbor-sync round-robin cycle completes, node runs one prune pass using current `SelfInclusiveRT(self)` (`LocalRT(self) ∪ {self}`): stored keys with `IsResponsible(self, K)=false` have `RecordOutOfRangeFirstSeen` recorded (if not already set) but are deleted only when `now - RecordOutOfRangeFirstSeen >= PRUNE_HYSTERESIS_DURATION`, every current close-group peer has mature repair proof for the current close-group snapshot, and every current close-group peer returns a positive nonce-bound audit proof for the key. Missing or failed proofs defer pruning and emit trust penalties after fresh responsibility confirmation; first-time bootstrap claims are penalized only after the bootstrap grace period, while repeated bootstrap claims are penalized immediately. Prune-confirmation work is bounded per pass. Keys that are in range have their `RecordOutOfRangeFirstSeen` cleared. Same hysteresis timing applies independently to `PaidForList` entries where `self ∉ PaidCloseGroup(K)` using `PaidOutOfRangeFirstSeen`, but paid-list pruning does not require remote storage confirmation. +- When a full neighbor-sync round-robin cycle completes, node runs one prune pass using current `SelfInclusiveRT(self)` (`LocalRT(self) ∪ {self}`): stored keys outside the storage-retention width have `RecordOutOfRangeFirstSeen` recorded immediately (if not already set, and even while a retained commitment still contains the key). A key continuously out of range for `>= PRUNE_HYSTERESIS_DURATION` is a prune candidate unconditionally; it is deleted only when all but one of the current strict close group (6 of 7 at reference parameters) return valid positive nonce-bound possession proofs, with targets taken directly from the local routing table (never filtered through repair proofs). Missing or failed proofs defer deletion — preserving the candidate, its timestamp, and its retry on later passes — and emit trust penalties only after fresh responsibility confirmation; first-time bootstrap claims are penalized only after the bootstrap grace period, while repeated bootstrap claims are penalized immediately. Prune-confirmation work is bounded per pass. Keys that are in range have their `RecordOutOfRangeFirstSeen` cleared. Same hysteresis timing applies independently to `PaidForList` entries where `self ∉ PaidCloseGroup(K)` using `PaidOutOfRangeFirstSeen`, gated on paid-list confirmations from the current paid close group rather than storage confirmation. 37. Non-`LocalRT` inbound sync behavior: - If a peer opens sync while not in receiver `LocalRT(self)`, receiver may still send hints to that peer, but receiver drops all inbound replica/paid hints from that peer. 38. Neighbor-sync priority under peer join: @@ -640,7 +658,7 @@ Each scenario should assert exact expected outcomes and state transitions. 49. Bootstrap active claim cleared on normal response, with history retained: - Peer `P` previously claimed bootstrapping. `P` later responds normally to a sync or audit request. Node clears the active bootstrap claim but retains `BootstrapClaimFirstSeen(self, P)` history. If `P` later claims bootstrapping again, node emits `BootstrapClaimAbuse` immediately instead of granting a second grace period. 50. Prune hysteresis prevents premature deletion: -- Key `K` goes out of range at time `T`. `RecordOutOfRangeFirstSeen(self, K)` is set to `T`. Key is NOT deleted. At `T + 24h` (less than `PRUNE_HYSTERESIS_DURATION`), key is still retained. At `T + 3 days` (`>= PRUNE_HYSTERESIS_DURATION`), key is eligible for deletion on the next prune pass only if all current close-group peers have mature repair proof for the current close-group snapshot and return positive nonce-bound audit proofs. +- Key `K` goes out of range at time `T`. `RecordOutOfRangeFirstSeen(self, K)` is set to `T`. Key is NOT deleted. At `T + 24h` (less than `PRUNE_HYSTERESIS_DURATION`), key is still retained. At `T + 3 days` (`>= PRUNE_HYSTERESIS_DURATION`), key becomes a prune candidate on the next prune pass and is deleted only when all but one of the current strict close group (6 of 7 at reference parameters) return valid positive nonce-bound possession proofs. 51. Prune hysteresis timestamp reset on partition heal: - Key `K` goes out of range at time `T`. `RecordOutOfRangeFirstSeen(self, K)` is set to `T`. At `T + 4h`, partition heals, peers return, `K` is back in range. `RecordOutOfRangeFirstSeen` is cleared. Key is retained. If `K` later goes out of range again, the clock restarts from zero. 52. Prune hysteresis applies to paid-list entries: diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 6495dc11..cdb5cef7 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -57,7 +57,7 @@ use crate::payment::{PaymentVerifier, VerificationContext}; use crate::replication::audit::AuditTickResult; use crate::replication::commitment::{commitment_hash, StorageCommitment}; use crate::replication::commitment_state::{ - PeerCommitmentRecord, PersistedRetention, ResponderCommitmentState, + PeerCommitmentRecord, PersistedRetention, ResponderCommitmentState, GOSSIP_ANSWERABILITY_TTL, }; use crate::replication::config::{ max_parallel_fetch, storage_admission_width, ReplicationConfig, MAX_AUDIT_RESPONSES_PER_PEER, @@ -78,6 +78,7 @@ use crate::replication::types::{ use crate::storage::LmdbStorage; use saorsa_core::identity::{NodeIdentity, PeerId}; use saorsa_core::{DhtNetworkEvent, P2PEvent, P2PNode, TrustEvent}; +use saorsa_pqc::api::sig::{MlDsaSecretKey, MlDsaVariant}; #[derive(Default)] struct FirstAuditObservability { @@ -280,8 +281,7 @@ fn quote_within_audit_window(quote_ts: SystemTime, now: SystemTime) -> bool { let too_future = quote_ts .duration_since(now) .is_ok_and(|ahead| ahead > MONETIZED_AUDIT_SKEW_MARGIN); - let audit_cutoff = crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL - .saturating_sub(MONETIZED_AUDIT_SKEW_MARGIN); + let audit_cutoff = GOSSIP_ANSWERABILITY_TTL.saturating_sub(MONETIZED_AUDIT_SKEW_MARGIN); let too_old = now .duration_since(quote_ts) .is_ok_and(|age| age >= audit_cutoff); @@ -3239,15 +3239,15 @@ async fn run_neighbor_sync_round( record.cycles_since_sync = record.cycles_since_sync.saturating_add(1); } } - let current_sync_epoch = { + { let mut epoch = sync_cycle_epoch.write().await; *epoch = epoch.saturating_add(1); - *epoch - }; + } // Post-cycle pruning (Section 11) — runs without holding sync_state. - // Remote prune-confirmation audits are storage-proof audits and only - // run after bootstrap has drained. + // Prune candidacy is unconditional once the hysteresis elapses; + // bootstrap state only defers the remote prune-confirmation audits + // until bootstrap has drained. let allow_remote_prune_audits = !bootstrapping && bootstrap_state.read().await.is_drained(); pruning::run_prune_pass_with_context(pruning::PrunePassContext { self_id: &self_id, @@ -3257,9 +3257,6 @@ async fn run_neighbor_sync_round( config, sync_state, repair_proofs, - current_sync_epoch, - #[cfg(any(test, feature = "test-utils"))] - repair_proof_now: None, allow_remote_prune_audits, commitment_state: Some(commitment_state), }) @@ -4859,8 +4856,7 @@ async fn handle_commitment_downgrade( } let last = rec.last_commitment()?; let pin = rec.commitment_hash()?; - let fresh = now.saturating_duration_since(rec.received_at) - < crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL; + let fresh = now.saturating_duration_since(rec.received_at) < GOSSIP_ANSWERABILITY_TTL; Some((pin, last.key_count, fresh)) }) }; @@ -4883,8 +4879,8 @@ async fn handle_commitment_downgrade( // from this peer may have refreshed `received_at` in the gap between // our read and write locks; if so, leave its fresh commitment intact. if let Some(rec) = last_commitment_by_peer.write().await.get_mut(source) { - let still_stale = now.saturating_duration_since(rec.received_at) - >= crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL; + let still_stale = + now.saturating_duration_since(rec.received_at) >= GOSSIP_ANSWERABILITY_TTL; if still_stale { rec.clear_commitment(); debug!( @@ -5158,8 +5154,6 @@ async fn rebuild_and_rotate_commitment( p2p: &Arc, config: &Arc, ) -> Result<()> { - use saorsa_pqc::api::sig::{MlDsaSecretKey, MlDsaVariant}; - let stored_keys = storage .all_keys() .await @@ -5305,25 +5299,7 @@ async fn rebuild_and_rotate_commitment( #[cfg(test)] #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] mod tests { - use super::{ - apply_audit_failure_credit_revocation, audit_failure_clears_bootstrap_claim, - audit_failure_revokes_holder_credit, audit_launch_decision, config, cooldown_allows_audit, - first_audit_terminal_outcome, first_failed_key_label, fresh_offer_payment_context, - paid_notify_payment_context, queue_first_audit_event, quote_within_audit_window, - FirstAuditQueueOutcome, FirstAuditTerminalOutcome, MonetizedPinEvent, - MONETIZED_AUDIT_SKEW_MARGIN, - }; - use crate::payment::VerificationContext; - use crate::replication::audit::AuditTickResult; - use crate::replication::recent_provers::RecentProvers; - use crate::replication::types::{AuditFailureReason, AuditFailureSummary, FailureEvidence}; - use lru::LruCache; - use saorsa_core::identity::PeerId; - use std::collections::HashMap; - use std::num::NonZeroUsize; - use std::time::Duration; - use std::time::Instant; - use std::time::SystemTime; + use super::*; fn test_peer(b: u8) -> PeerId { let mut bytes = [0u8; 32]; @@ -5349,7 +5325,7 @@ mod tests { challenge_id: 1, challenged_peer: peer, confirmed_failed_keys: vec![test_key(1)], - summary: AuditFailureSummary::default(), + summary: crate::replication::types::AuditFailureSummary::default(), reason: AuditFailureReason::Timeout, }, }; @@ -5442,7 +5418,6 @@ mod tests { /// stale or future/skewed client-forwarded quote cannot frame an honest node. #[test] fn monetized_quote_audit_window_fails_closed_both_ends() { - use crate::replication::commitment_state::GOSSIP_ANSWERABILITY_TTL; let now = SystemTime::now(); // Fresh (just quoted) and small future/past skew -> audited. assert!(quote_within_audit_window(now, now)); diff --git a/src/replication/pruning.rs b/src/replication/pruning.rs index 79db6c9a..b5e5f76e 100644 --- a/src/replication/pruning.rs +++ b/src/replication/pruning.rs @@ -3,6 +3,41 @@ //! On `NeighborSyncCycleComplete`: prune stored records and `PaidForList` //! entries that have been continuously out of range for at least //! `PRUNE_HYSTERESIS_DURATION`. +//! +//! # Stored-record prune lifecycle +//! +//! Each stored record is classified per pass against the current local +//! routing table: +//! +//! - `InRange`: self is within the storage-retention width +//! (`close_group_size + STORAGE_ADMISSION_MARGIN`, 9 at production +//! parameters); any out-of-range state is cleared. +//! - `HysteresisPending`: outside the retention width, but not yet for the +//! full `PRUNE_HYSTERESIS_DURATION`. The first-seen timestamp is recorded +//! immediately on leaving range — a retained commitment vetoes DELETION, +//! never the start of this clock. The timestamp is process-local and is +//! not persisted across restarts. +//! - `Candidate`: continuously outside the retention width for the full +//! hysteresis. Candidacy is unconditional: it never depends on repair-hint +//! proofs, bootstrap state, audit budget, or prior neighbor-sync contact. +//! - `HeldByCommitment` / `BootstrapDeferred` / `BudgetDeferred`: scheduling +//! dispositions of a candidate. They defer the prune audit (and deletion), +//! but never remove candidacy or restart the hysteresis clock. +//! - `AuditFailed`: an audited candidate whose current strict close group +//! returned fewer than the required positive possession proofs +//! (`prune_proofs_needed`, 6 of 7 at production parameters). The record +//! and its first-seen timestamp are retained and retried on later passes. +//! - `Pruned`: deleted after the audit round re-passed every check against +//! the then-current routing table (see +//! `revalidate_record_prune_candidate`). +//! +//! Prune-confirmation audits challenge the CURRENT strict closest +//! `close_group_size` peers to the key, taken directly from the local +//! routing table — never filtered through `RepairProofs` or prior +//! neighbor-sync hints. The repair-proof maturity gate remains a +//! prerequisite for the responsible-chunk storage audit (see +//! `audit_tick_with_repair_proofs`), which is a different mechanism with a +//! different threat model. use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -34,6 +69,10 @@ use crate::replication::types::{ }; use crate::storage::LmdbStorage; +// `RepairProofs` remains in the prune-pass context only so records deleted by +// pruning also drop their (audit-path) repair-proof entries; it plays no part +// in prune candidacy or prune-audit target selection. + use super::REPLICATION_TRUST_WEIGHT; const MAX_CONCURRENT_PRUNE_AUDIT_CHALLENGES: usize = 32; @@ -52,12 +91,32 @@ const MAX_PAID_PRUNE_VERIFICATION_PEERS_PER_PASS: usize = MAX_CONCURRENT_PRUNE_A /// Summary of a prune pass. #[derive(Debug, Default)] pub struct PruneResult { + /// Total stored records scanned. + pub records_total: usize, + /// Number of records for which self is within the storage-retention width. + pub records_in_range: usize, /// Number of records deleted from storage. pub records_pruned: usize, /// Number of records with out-of-range timestamp newly set. pub records_marked_out_of_range: usize, /// Number of records with out-of-range timestamp cleared (back in range). pub records_cleared: usize, + /// Out-of-range records still inside the hysteresis window. + pub records_hysteresis_pending: usize, + /// Records continuously out of range for the full hysteresis (candidates), + /// regardless of whether their audit could be scheduled this pass. + pub records_candidates: usize, + /// Candidates whose deletion (and audit) is vetoed by a retained + /// recently-gossiped commitment. + pub records_held_by_commitment: usize, + /// Candidates whose audit is deferred until bootstrap drains. + pub records_bootstrap_deferred: usize, + /// Candidates whose audit is deferred by the per-pass challenge budget. + pub records_budget_deferred: usize, + /// Candidates audited against their current strict close group this pass. + pub records_audits_attempted: usize, + /// Audited candidates confirmed below the proof threshold (retained). + pub records_audit_below_threshold: usize, /// Number of `PaidForList` entries removed. pub paid_entries_pruned: usize, /// Number of `PaidForList` entries with out-of-range timestamp newly set. @@ -80,13 +139,9 @@ pub struct PrunePassContext<'a> { pub config: &'a ReplicationConfig, /// Neighbor-sync state, including prune cursor and bootstrap claims. pub sync_state: &'a Arc>, - /// Key-specific repair proofs used to gate prune-confirmation audits. + /// Repair-proof table, consulted ONLY to drop a deleted record's proof + /// entries. Prune candidacy and prune-audit target selection never read it. pub repair_proofs: &'a Arc>, - /// Current local neighbor-sync cycle epoch for repair-proof maturity. - pub current_sync_epoch: u64, - /// Test-only clock override for repair-proof maturity checks. - #[cfg(any(test, feature = "test-utils"))] - pub repair_proof_now: Option, /// Whether remote prune-confirmation audits are allowed this pass. pub allow_remote_prune_audits: bool, /// Responder commitment state, used to veto deleting a chunk still held @@ -105,10 +160,17 @@ enum PruneAuditStatus { #[derive(Debug, Default)] struct RecordPruneStats { + in_range: usize, marked: usize, cleared: usize, - pruned: usize, + hysteresis_pending: usize, + candidates: usize, held_by_commitment: usize, + bootstrap_deferred: usize, + budget_deferred: usize, + audits_attempted: usize, + audit_below_threshold: usize, + pruned: usize, } #[derive(Debug, Default)] @@ -153,6 +215,9 @@ impl PaidPruneDeferredCounts { } } +/// A prune candidate scheduled for a prune-confirmation audit this pass. +/// `target_peers` is the current strict close group for the key (self +/// excluded), taken directly from the local routing table. #[derive(Debug, Clone)] struct RecordPruneCandidate { key: XorName, @@ -160,29 +225,66 @@ struct RecordPruneCandidate { } struct RecordPruneKeyOutcome { + /// Whether the out-of-range timestamp was newly set for this key. marked: bool, state: RecordPruneKeyState, } -impl Default for RecordPruneKeyOutcome { - fn default() -> Self { - Self { - marked: false, - state: RecordPruneKeyState::None, - } - } +/// Per-pass lifecycle classification of one stored record (see module docs). +enum RecordPruneKeyState { + /// Self is within the storage-retention width for this key. `cleared` is + /// true when a stale out-of-range timestamp was removed. + InRange { cleared: bool }, + /// Outside the retention width, but not yet for the full hysteresis. + HysteresisPending, + /// Continuously outside the retention width for at least the hysteresis + /// duration. Candidacy is unconditional; the disposition only says how the + /// candidate is scheduled this pass. + Candidate(PruneCandidateDisposition), } -enum RecordPruneKeyState { - None, - Cleared, +/// How a prune candidate is scheduled within one pass. Every non-`Auditable` +/// variant is a deferral: the record, its first-seen timestamp, and its +/// candidacy are all retained for later passes. +enum PruneCandidateDisposition { + /// Still committed under a recently-gossiped commitment: a neighbour can + /// pin that root and demand the bytes in a round-2 byte challenge, so + /// deletion is vetoed (and the audit skipped) until the key ages out of + /// the retention window. Bounded reprieve: the commitment rebuild only + /// commits to keys we are still responsible for, so the key drops out of + /// the next rebuilt commitment and `is_held` flips false within at most + /// `RETAINED_GOSSIPED_COMMITMENTS` gossip rotations. + HeldByCommitment, + /// Remote prune-confirmation audits are not allowed yet (bootstrap has + /// not drained). BootstrapDeferred, + /// The per-pass audit challenge budget is exhausted. BudgetDeferred, - /// Out of range, but still committed under a recently-gossiped commitment: - /// deletion is vetoed (and the out-of-range hysteresis clock is not even - /// started) until the key ages out of the last-2-gossiped window. + /// The current close group has no remote peers to audit; retain + /// conservatively. + Unauditable, + /// Audit the current strict close group this pass. + Auditable(RecordPruneCandidate), +} + +/// Outcome of revalidating one audited candidate immediately before deletion. +enum PruneRevalidationOutcome { + /// Every check re-passed against the current routing table: delete. + Delete, + /// Self moved back inside the retention width; state cleared, record kept. + ClearedBackInRange, + /// The hysteresis condition no longer holds (timestamp cleared or reset + /// concurrently); record kept. + HysteresisPending, + /// (Re-)committed under a retained commitment; deletion vetoed. HeldByCommitment, - Candidate(RecordPruneCandidate), + /// Fewer than the required positive proofs from the CURRENT strict close + /// group — including when the group's membership changed after the audit + /// round, which invalidates stale positive reports. Record kept, retried + /// on later passes. + AuditFailed, + /// The current close group has no remote peers; retain conservatively. + Unauditable, } enum PaidPruneKeyState { @@ -206,7 +308,7 @@ struct PruneAuditReportState { /// Execute post-cycle responsibility pruning. /// /// For each stored record K: -/// - If `self` is within the storage-admission group +/// - If `self` is within the storage-retention group /// (`close_group_size + STORAGE_ADMISSION_MARGIN`): clear /// `RecordOutOfRangeFirstSeen`. /// - If not in that group: set timestamp if not already set; delete if the @@ -220,12 +322,10 @@ struct PruneAuditReportState { /// 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 -/// table or advanced sync epoch to pass into record prune-confirmation audits. -/// Out-of-range records are therefore marked/deferred rather than deleted via -/// remote confirmation. The replication engine calls -/// [`run_prune_pass_with_context`] so it can pass real repair proofs. +/// Convenience wrapper over [`run_prune_pass_with_context`] with a throwaway +/// repair-proof table (only used to drop proofs for deleted keys) and no +/// responder commitment state (so no commitment-retention deletion veto). +/// The replication engine calls [`run_prune_pass_with_context`] directly. pub async fn run_prune_pass( self_id: &PeerId, storage: &Arc, @@ -244,16 +344,13 @@ pub async fn run_prune_pass( config, sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: 0, - #[cfg(any(test, feature = "test-utils"))] - repair_proof_now: None, allow_remote_prune_audits, commitment_state: None, }) .await } -/// Execute one prune pass with repair-proof-gated remote confirmations. +/// Execute one prune pass (see the module docs for the record lifecycle). 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(); @@ -268,25 +365,50 @@ pub async fn run_prune_pass_with_context(ctx: PrunePassContext<'_>) -> PruneResu .await; let result = PruneResult { + records_total: stored_count, + records_in_range: record_stats.in_range, records_pruned: record_stats.pruned, records_marked_out_of_range: record_stats.marked, records_cleared: record_stats.cleared, + records_hysteresis_pending: record_stats.hysteresis_pending, + records_candidates: record_stats.candidates, + records_held_by_commitment: record_stats.held_by_commitment, + records_bootstrap_deferred: record_stats.bootstrap_deferred, + records_budget_deferred: record_stats.budget_deferred, + records_audits_attempted: record_stats.audits_attempted, + records_audit_below_threshold: record_stats.audit_below_threshold, paid_entries_pruned: paid_stats.pruned, paid_entries_marked: paid_stats.marked, paid_entries_cleared: paid_stats.cleared, }; + // One aggregate line per pass: the full lifecycle census (never per-chunk). info!( - "Prune pass complete: records={}/{} pruned, paid={}/{} pruned", - result.records_pruned, stored_count, result.paid_entries_pruned, paid_count, + "Prune pass complete: records total={} in_range={} newly_marked={} cleared={} \ + hysteresis_pending={} candidates={} held_by_commitment={} bootstrap_deferred={} \ + budget_deferred={} audits_attempted={} audit_below_threshold={} pruned={}; \ + paid total={} marked={} cleared={} pruned={}", + result.records_total, + result.records_in_range, + result.records_marked_out_of_range, + result.records_cleared, + result.records_hysteresis_pending, + result.records_candidates, + result.records_held_by_commitment, + result.records_bootstrap_deferred, + result.records_budget_deferred, + result.records_audits_attempted, + result.records_audit_below_threshold, + result.records_pruned, + paid_count, + result.paid_entries_marked, + result.paid_entries_cleared, + result.paid_entries_pruned, ); result } -// Combines main's prune-proof/admission gate with ADR-0002's commitment -// retention veto in one pass; the merged body runs just over the line budget. -#[allow(clippy::too_many_lines)] async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPruneStats) { let stored_keys = match ctx.storage.all_keys().await { Ok(keys) => keys, @@ -300,44 +422,32 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune let mut stats = RecordPruneStats::default(); let mut candidates = Vec::new(); let mut audit_challenge_budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; - let mut budget_deferred = 0usize; - let mut bootstrap_deferred = 0usize; + let deps = RecordPruneKeyDeps { + self_id: ctx.self_id, + paid_list: ctx.paid_list, + config: ctx.config, + allow_remote_prune_audits: ctx.allow_remote_prune_audits, + commitment_state: ctx.commitment_state, + }; let scan_start = prune_scan_start(ctx.sync_state, stored_keys.len()).await; let mut last_selected_offset = None; for offset in 0..stored_keys.len() { let key = &stored_keys[(scan_start + offset) % stored_keys.len()]; - let (storage_admission_group, strict_close_group) = + let (storage_admission_peers, strict_close_peers) = record_prune_lookup_groups(key, ctx.p2p_node, ctx.config).await; let outcome = evaluate_record_prune_key( - ctx, + &deps, key, - &storage_admission_group, - &strict_close_group, + &storage_admission_peers, + &strict_close_peers, now, &mut audit_challenge_budget, - ) - .await; - if outcome.marked { - stats.marked += 1; - } - match outcome.state { - RecordPruneKeyState::None => {} - RecordPruneKeyState::Cleared => stats.cleared += 1, - RecordPruneKeyState::BootstrapDeferred => { - bootstrap_deferred = bootstrap_deferred.saturating_add(1); - } - RecordPruneKeyState::BudgetDeferred => { - budget_deferred = budget_deferred.saturating_add(1); - } - RecordPruneKeyState::HeldByCommitment => { - stats.held_by_commitment = stats.held_by_commitment.saturating_add(1); - } - RecordPruneKeyState::Candidate(candidate) => { - last_selected_offset = Some(offset); - candidates.push(candidate); - } + ); + if let Some(candidate) = tally_record_prune_outcome(&mut stats, outcome, key) { + last_selected_offset = Some(offset); + candidates.push(candidate); } } @@ -349,28 +459,7 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune ) .await; - if bootstrap_deferred > 0 { - debug!( - "Deferred {bootstrap_deferred} prune candidates until bootstrap drain allows \ - remote prune-confirmation audits" - ); - } - - if budget_deferred > 0 { - debug!( - "Deferred {budget_deferred} prune candidates due to per-pass audit budget \ - ({MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS} challenges)" - ); - } - - if stats.held_by_commitment > 0 { - debug!( - "Vetoed {} prune candidate(s) still committed under a recently-gossiped \ - commitment (bounded reprieve until they age out of the retention window)", - stats.held_by_commitment - ); - } - + stats.audits_attempted = candidates.len(); let present_by_key = collect_record_prune_proofs( &candidates, ctx.storage, @@ -379,17 +468,19 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune ctx.sync_state, ) .await; - let (keys_to_delete, revalidated_cleared) = revalidated_record_prune_keys( - &candidates, - &present_by_key, - ctx.self_id, - ctx.paid_list, - ctx.p2p_node, - ctx.config, - ctx.commitment_state, - ) - .await; + let (keys_to_delete, revalidated_cleared, audit_below_threshold) = + revalidated_record_prune_keys( + &candidates, + &present_by_key, + ctx.self_id, + ctx.paid_list, + ctx.p2p_node, + ctx.config, + ctx.commitment_state, + ) + .await; stats.cleared += revalidated_cleared; + stats.audit_below_threshold = audit_below_threshold; stats.pruned = delete_stored_records( &keys_to_delete, ctx.storage, @@ -401,128 +492,189 @@ async fn prune_stored_records(ctx: &PrunePassContext<'_>) -> (usize, RecordPrune (stored_keys.len(), stats) } +/// Fold one record's per-pass classification into the pass stats. Returns the +/// auditable candidate when the record was scheduled for a prune audit. +fn tally_record_prune_outcome( + stats: &mut RecordPruneStats, + outcome: RecordPruneKeyOutcome, + key: &XorName, +) -> Option { + if outcome.marked { + stats.marked += 1; + } + match outcome.state { + RecordPruneKeyState::InRange { cleared } => { + stats.in_range += 1; + if cleared { + stats.cleared += 1; + } + None + } + RecordPruneKeyState::HysteresisPending => { + stats.hysteresis_pending += 1; + None + } + RecordPruneKeyState::Candidate(disposition) => { + stats.candidates += 1; + match disposition { + PruneCandidateDisposition::HeldByCommitment => { + stats.held_by_commitment += 1; + None + } + PruneCandidateDisposition::BootstrapDeferred => { + stats.bootstrap_deferred += 1; + None + } + PruneCandidateDisposition::BudgetDeferred => { + stats.budget_deferred += 1; + None + } + PruneCandidateDisposition::Unauditable => { + debug!( + "Cannot prune-audit {}: current close group has no remote peers", + hex::encode(key) + ); + None + } + PruneCandidateDisposition::Auditable(candidate) => Some(candidate), + } + } + } +} + +/// Current self-inclusive storage-retention group and strict close group for +/// `key`, from the local routing table, reduced to peer ids. async fn record_prune_lookup_groups( key: &XorName, p2p_node: &Arc, config: &ReplicationConfig, -) -> (Vec, Vec) { +) -> (Vec, Vec) { let dht = p2p_node.dht_manager(); - let storage_admission_group = dht + let storage_admission_group: Vec = dht .find_closest_nodes_local_with_self(key, storage_admission_width(config.close_group_size)) .await; - let strict_close_group = dht + let strict_close_group: Vec = dht .find_closest_nodes_local_with_self(key, config.close_group_size) .await; - (storage_admission_group, strict_close_group) + ( + storage_admission_group + .iter() + .map(|node| node.peer_id) + .collect(), + strict_close_group.iter().map(|node| node.peer_id).collect(), + ) } -async fn evaluate_record_prune_key( - ctx: &PrunePassContext<'_>, +/// The subset of [`PrunePassContext`] needed to classify one stored record. +/// Split out (with routing-table lookups precomputed by the caller) so unit +/// tests can drive the classification without a live `P2PNode`. +struct RecordPruneKeyDeps<'a> { + self_id: &'a PeerId, + paid_list: &'a Arc, + config: &'a ReplicationConfig, + allow_remote_prune_audits: bool, + commitment_state: Option<&'a Arc>, +} + +/// Classify one stored record for this pass (see the module docs). +/// +/// The out-of-range timestamp is recorded the moment self is outside the +/// storage-retention width — even while the key is still held by a retained +/// commitment. Candidacy (past-hysteresis) is unconditional; commitment +/// retention, bootstrap state, and the audit budget only defer the audit or +/// veto the deletion. +fn evaluate_record_prune_key( + deps: &RecordPruneKeyDeps<'_>, key: &XorName, - storage_admission_group: &[DHTNode], - strict_close_group: &[DHTNode], + storage_admission_peers: &[PeerId], + strict_close_peers: &[PeerId], now: Instant, audit_challenge_budget: &mut usize, ) -> RecordPruneKeyOutcome { - let mut outcome = RecordPruneKeyOutcome::default(); - let is_responsible = storage_admission_group - .iter() - .any(|node| node.peer_id == *ctx.self_id); - - if is_responsible { - if ctx.paid_list.record_out_of_range_since(key).is_some() { - ctx.paid_list.clear_record_out_of_range(key); - outcome.state = RecordPruneKeyState::Cleared; - } - return outcome; - } - - // Retention veto: the key has left our close group, but if it is still - // committed under a recently-gossiped commitment a neighbour can pin that - // root and demand its bytes in a round-2 byte challenge. Deleting it now - // would turn an honest node's response into `Absent` → a confirmed audit - // failure. Veto deletion AND do not even start the out-of-range hysteresis - // clock yet: the commitment rebuild only commits to keys we are still - // responsible for, so this key drops out of the next rebuilt commitment and - // ages out of the last-2-gossiped window within at most - // `RETAINED_GOSSIPED_COMMITMENTS` gossip rotations, after which `is_held` - // returns false and the key prunes through the normal path. This is a - // bounded reprieve, not a permanent pin. - if let Some(cs) = ctx.commitment_state { - if cs.is_held(key) { - outcome.state = RecordPruneKeyState::HeldByCommitment; - return outcome; + if storage_admission_peers.contains(deps.self_id) { + let cleared = deps.paid_list.record_out_of_range_since(key).is_some(); + if cleared { + deps.paid_list.clear_record_out_of_range(key); } + return RecordPruneKeyOutcome { + marked: false, + state: RecordPruneKeyState::InRange { cleared }, + }; } - if ctx.paid_list.record_out_of_range_since(key).is_none() { - outcome.marked = true; - } - ctx.paid_list.set_record_out_of_range(key); + // Outside the retention width: start (or continue) the hysteresis clock + // immediately. A retained commitment vetoes DELETION further down, never + // the timer — otherwise commitment retention would postpone when a record + // can become a candidate instead of only protecting answerability. + let marked = deps.paid_list.record_out_of_range_since(key).is_none(); + deps.paid_list.set_record_out_of_range(key); - let Some(first_seen) = ctx.paid_list.record_out_of_range_since(key) else { - return outcome; + let Some(first_seen) = deps.paid_list.record_out_of_range_since(key) else { + // The timestamp was just set; its absence means a concurrent clear. + return RecordPruneKeyOutcome { + marked, + state: RecordPruneKeyState::HysteresisPending, + }; }; let elapsed = now .checked_duration_since(first_seen) .unwrap_or(Duration::ZERO); - if elapsed < ctx.config.prune_hysteresis_duration { - return outcome; + if elapsed < deps.config.prune_hysteresis_duration { + return RecordPruneKeyOutcome { + marked, + state: RecordPruneKeyState::HysteresisPending, + }; + } + + RecordPruneKeyOutcome { + marked, + state: RecordPruneKeyState::Candidate(schedule_prune_candidate( + deps, + key, + strict_close_peers, + audit_challenge_budget, + )), + } +} + +/// Decide how a prune candidate is scheduled this pass. Deferrals retain the +/// record, its first-seen timestamp, and its candidacy. +/// +/// Audit targets are the CURRENT strict close group from the local routing +/// table — never filtered through `RepairProofs` or prior neighbor-sync +/// contact: prune confirmation concerns the peers now closest to the KEY, +/// while neighbor sync contacts the peers closest to SELF, and nothing +/// guarantees those sets overlap for an out-of-range key. +fn schedule_prune_candidate( + deps: &RecordPruneKeyDeps<'_>, + key: &XorName, + strict_close_peers: &[PeerId], + audit_challenge_budget: &mut usize, +) -> PruneCandidateDisposition { + if let Some(cs) = deps.commitment_state { + if cs.is_held(key) { + return PruneCandidateDisposition::HeldByCommitment; + } } - if !ctx.allow_remote_prune_audits { - outcome.state = RecordPruneKeyState::BootstrapDeferred; - return outcome; + if !deps.allow_remote_prune_audits { + return PruneCandidateDisposition::BootstrapDeferred; } - let target_peers = remote_close_group_peers(strict_close_group, ctx.self_id); + let target_peers = remote_close_group_peers(strict_close_peers, deps.self_id); if target_peers.is_empty() { - warn!( - "Cannot prune {}: current close group has no remote peers", - hex::encode(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 = - strict_close_group.iter().map(|node| node.peer_id).collect(); - #[cfg(any(test, feature = "test-utils"))] - let repair_proof_now = ctx.repair_proof_now.unwrap_or(now); - #[cfg(not(any(test, feature = "test-utils")))] - let repair_proof_now = now; - let audit_targets = peers_with_mature_repair_proofs( - key, - &target_peers, - ¤t_close_peers, - ctx.repair_proofs, - ctx.current_sync_epoch, - repair_proof_now, - ) - .await; - let proofs_needed = prune_proofs_needed(target_peers.len()); - if proofs_needed == 0 || audit_targets.len() < proofs_needed { - debug!( - "Deferring prune for {} until enough of the close group has mature \ - repair proofs", - hex::encode(key) - ); - return outcome; + return PruneCandidateDisposition::Unauditable; } - if audit_targets.len() > *audit_challenge_budget { - outcome.state = RecordPruneKeyState::BudgetDeferred; - return outcome; + if target_peers.len() > *audit_challenge_budget { + return PruneCandidateDisposition::BudgetDeferred; } - *audit_challenge_budget -= audit_targets.len(); - outcome.state = RecordPruneKeyState::Candidate(RecordPruneCandidate { + *audit_challenge_budget -= target_peers.len(); + PruneCandidateDisposition::Auditable(RecordPruneCandidate { key: *key, - target_peers: audit_targets, - }); - outcome + target_peers, + }) } async fn prune_paid_entries( @@ -553,10 +705,13 @@ async fn prune_paid_entries( for offset in 0..paid_keys.len() { let key = &paid_keys[(scan_start + offset) % paid_keys.len()]; - let closest: Vec = dht + let closest: Vec = dht .find_closest_nodes_local_with_self(key, config.paid_list_close_group_size) - .await; - let in_paid_group = closest.iter().any(|n| n.peer_id == *self_id); + .await + .iter() + .map(|node: &DHTNode| node.peer_id) + .collect(); + let in_paid_group = closest.contains(self_id); if in_paid_group { if paid_list.paid_out_of_range_since(key).is_some() { @@ -627,7 +782,7 @@ async fn prune_paid_entries( fn select_paid_prune_candidate( key: &XorName, - closest: &[DHTNode], + closest: &[PeerId], self_id: &PeerId, allow_remote_prune_audits: bool, selected_candidate_count: usize, @@ -696,11 +851,14 @@ async fn revalidated_paid_prune_keys( let now = Instant::now(); for (key, _) in expired_candidates { - let closest: Vec = dht + let closest: Vec = dht .find_closest_nodes_local_with_self(key, config.paid_list_close_group_size) - .await; + .await + .iter() + .map(|node: &DHTNode| node.peer_id) + .collect(); - if closest.iter().any(|n| n.peer_id == *self_id) { + if closest.contains(self_id) { if paid_list.paid_out_of_range_since(key).is_some() { paid_list.clear_paid_out_of_range(key); cleared += 1; @@ -747,11 +905,11 @@ async fn revalidated_paid_prune_keys( (keys_to_delete, cleared) } -fn remote_close_group_peers(close_group: &[DHTNode], self_id: &PeerId) -> Vec { +fn remote_close_group_peers(close_group: &[PeerId], self_id: &PeerId) -> Vec { close_group .iter() - .filter(|node| node.peer_id != *self_id) - .map(|node| node.peer_id) + .filter(|peer| *peer != self_id) + .copied() .collect() } @@ -862,28 +1020,6 @@ fn paid_confirmations_by_key( 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, - now: Instant, -) -> Vec { - let mut proofs = repair_proofs.write().await; - target_peers - .iter() - .filter(|peer| { - proofs.has_mature_replica_hint(peer, key, current_close_peers, current_sync_epoch, now) - }) - .copied() - .collect() -} - async fn prune_scan_start( sync_state: &Arc>, stored_key_count: usize, @@ -981,6 +1117,16 @@ async fn collect_record_prune_proofs( present_by_key } +/// Re-check every audited candidate against current local state immediately +/// before deletion, returning the keys to delete, the number of out-of-range +/// timestamps cleared (self back in range), and the number of candidates +/// confirmed below the proof threshold. +/// +/// The audit round takes time; the routing table may have changed underneath +/// it, including self moving back into range or the strict close group +/// changing membership. Positive reports only count from peers still in the +/// CURRENT strict close group, against a threshold computed from that +/// current group. async fn revalidated_record_prune_keys( candidates: &[RecordPruneCandidate], present_by_key: &HashMap>, @@ -989,76 +1135,117 @@ async fn revalidated_record_prune_keys( p2p_node: &Arc, config: &ReplicationConfig, commitment_state: Option<&Arc>, -) -> (Vec, usize) { +) -> (Vec, usize, usize) { let mut keys_to_delete = Vec::new(); let mut cleared = 0; + let mut audit_below_threshold = 0; let now = Instant::now(); for candidate in candidates { - // TOCTOU guard: a rotation/gossip may have (re-)committed this key - // between candidate selection and now. Re-check retention immediately - // before scheduling deletion so we never delete bytes a recently - // gossiped commitment still owes in a round-2 byte challenge. - if let Some(cs) = commitment_state { - if cs.is_held(&candidate.key) { - continue; - } - } - - let (storage_admission_group, strict_close_group) = + let (storage_admission_peers, strict_close_peers) = record_prune_lookup_groups(&candidate.key, p2p_node, config).await; + let held_by_commitment = commitment_state.is_some_and(|cs| cs.is_held(&candidate.key)); + let inputs = PruneRevalidationInputs { + self_id, + first_seen: paid_list.record_out_of_range_since(&candidate.key), + prune_hysteresis_duration: config.prune_hysteresis_duration, + held_by_commitment, + storage_admission_peers: &storage_admission_peers, + strict_close_peers: &strict_close_peers, + now, + }; - if storage_admission_group - .iter() - .any(|n| n.peer_id == *self_id) - { - if paid_list - .record_out_of_range_since(&candidate.key) - .is_some() - { + match revalidate_record_prune_candidate(candidate, present_by_key, &inputs) { + PruneRevalidationOutcome::Delete => keys_to_delete.push(candidate.key), + PruneRevalidationOutcome::ClearedBackInRange => { paid_list.clear_record_out_of_range(&candidate.key); cleared += 1; } - continue; + PruneRevalidationOutcome::AuditFailed => { + audit_below_threshold += 1; + debug!( + "Deferring prune for {} until all but one of the current close group \ + report it", + hex::encode(candidate.key) + ); + } + PruneRevalidationOutcome::HysteresisPending + | PruneRevalidationOutcome::HeldByCommitment => {} + PruneRevalidationOutcome::Unauditable => { + debug!( + "Cannot prune {}: current close group has no remote peers", + hex::encode(candidate.key) + ); + } } + } - let Some(first_seen) = paid_list.record_out_of_range_since(&candidate.key) else { - continue; - }; - let elapsed = now - .checked_duration_since(first_seen) - .unwrap_or(Duration::ZERO); - if elapsed < config.prune_hysteresis_duration { - continue; - } + (keys_to_delete, cleared, audit_below_threshold) +} - let current_target_peers = remote_close_group_peers(&strict_close_group, self_id); - if current_target_peers.is_empty() { - warn!( - "Cannot prune {}: current close group has no remote peers", - hex::encode(candidate.key) - ); - continue; - } +/// Inputs for revalidating one audited candidate immediately before deletion. +struct PruneRevalidationInputs<'a> { + self_id: &'a PeerId, + /// The candidate's out-of-range first-seen timestamp as it stands NOW. + first_seen: Option, + prune_hysteresis_duration: Duration, + /// Whether a retained commitment holds the key NOW (TOCTOU re-check: a + /// rotation/gossip may have re-committed it since candidate selection). + held_by_commitment: bool, + /// Current self-inclusive storage-retention group for the key. + storage_admission_peers: &'a [PeerId], + /// Current strict close group for the key. + strict_close_peers: &'a [PeerId], + now: Instant, +} - 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 all but one of the current close group \ - report it", - hex::encode(candidate.key) - ); - } +/// Pure deletion decision for one audited candidate (see +/// [`revalidated_record_prune_keys`]). Deletion requires that, against the +/// CURRENT routing table: self is still outside the storage-retention width, +/// the hysteresis still holds, no retained commitment holds the key, and at +/// least [`prune_proofs_needed`] of the current strict close group supplied +/// positive possession proofs. Stale positive reports from peers no longer in +/// the current close group never count. +fn revalidate_record_prune_candidate( + candidate: &RecordPruneCandidate, + present_by_key: &HashMap>, + inputs: &PruneRevalidationInputs<'_>, +) -> PruneRevalidationOutcome { + if inputs.storage_admission_peers.contains(inputs.self_id) { + return PruneRevalidationOutcome::ClearedBackInRange; } - (keys_to_delete, cleared) + if inputs.held_by_commitment { + return PruneRevalidationOutcome::HeldByCommitment; + } + + let Some(first_seen) = inputs.first_seen else { + return PruneRevalidationOutcome::HysteresisPending; + }; + let elapsed = inputs + .now + .checked_duration_since(first_seen) + .unwrap_or(Duration::ZERO); + if elapsed < inputs.prune_hysteresis_duration { + return PruneRevalidationOutcome::HysteresisPending; + } + + let current_target_peers = remote_close_group_peers(inputs.strict_close_peers, inputs.self_id); + if current_target_peers.is_empty() { + return PruneRevalidationOutcome::Unauditable; + } + + 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, + ) { + PruneRevalidationOutcome::Delete + } else { + PruneRevalidationOutcome::AuditFailed + } } fn build_peer_audit_challenges(candidates: &[RecordPruneCandidate]) -> Vec<(PeerId, XorName)> { @@ -1857,4 +2044,663 @@ mod tests { assert!(reported.contains(&peer)); assert!(reported.contains(&other_peer)); } + + // -- Prune lifecycle classification (see module docs) -------------------- + + /// Production strict close-group size. + const PROD_CLOSE_GROUP: usize = 7; + /// Production storage-retention width (`close_group_size + margin`). + const PROD_RETENTION_WIDTH: usize = 9; + /// A self id outside every `peer_ids(..)` helper group. + const SELF_BYTE: u8 = 99; + + async fn test_paid_list() -> (Arc, tempfile::TempDir) { + let dir = tempfile::tempdir().expect("tempdir"); + let paid_list = Arc::new(PaidList::new(dir.path()).await.expect("paid list")); + (paid_list, dir) + } + + fn record_deps<'a>( + self_id: &'a PeerId, + paid_list: &'a Arc, + config: &'a ReplicationConfig, + allow_remote_prune_audits: bool, + commitment_state: Option<&'a Arc>, + ) -> RecordPruneKeyDeps<'a> { + RecordPruneKeyDeps { + self_id, + paid_list, + config, + allow_remote_prune_audits, + commitment_state, + } + } + + /// A responder commitment state whose retained (gossiped) commitment + /// contains exactly `key`, so `is_held(key)` is true. + fn held_commitment_state(key: XorName, content: &[u8]) -> Arc { + let (pk, sk) = saorsa_pqc::api::sig::ml_dsa_65() + .generate_keypair() + .expect("keypair"); + let bytes_hash = *blake3::hash(content).as_bytes(); + let built = crate::replication::commitment_state::BuiltCommitment::build( + vec![(key, bytes_hash)], + &[0; 32], + &sk, + &pk.to_bytes(), + ) + .expect("build commitment"); + let hash = built.hash(); + let state = ResponderCommitmentState::new(); + state.rotate(built); + state.mark_gossiped(hash); + Arc::new(state) + } + + fn instant_after(base: Instant, delta: Duration) -> Instant { + base.checked_add(delta).expect("test instant overflow") + } + + /// #1 + #4: a record outside the retention width is marked the moment it + /// is observed there — even while a retained commitment still holds the + /// key. Retention vetoes deletion, never the hysteresis clock. + #[tokio::test] + async fn out_of_range_record_is_marked_immediately_even_when_held_by_commitment() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA0); + let commitment = held_commitment_state(key, b"held bytes"); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, true, Some(&commitment)); + let mut budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; + + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + Instant::now(), + &mut budget, + ); + + assert!( + outcome.marked, + "first out-of-range observation must set the timestamp" + ); + assert!(matches!( + outcome.state, + RecordPruneKeyState::HysteresisPending + )); + assert!( + paid_list.record_out_of_range_since(&key).is_some(), + "a retained commitment must not delay the start of hysteresis" + ); + } + + /// #2: a record outside the retention width for less than the hysteresis + /// duration is not a prune candidate. + #[tokio::test] + async fn record_outside_range_within_hysteresis_is_not_a_candidate() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA1); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, true, None); + let mut budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; + + paid_list.set_record_out_of_range(&key); + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + Instant::now(), + &mut budget, + ); + + assert!(!outcome.marked, "timestamp was already set"); + assert!(matches!( + outcome.state, + RecordPruneKeyState::HysteresisPending + )); + assert_eq!( + budget, MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS, + "no audit may be scheduled inside the hysteresis window" + ); + } + + /// #3 + #8: after the full hysteresis the record becomes an auditable + /// candidate whose targets are the CURRENT strict close group — with an + /// empty `RepairProofs` table and no prior neighbor-sync contact. + #[tokio::test] + async fn record_past_hysteresis_is_candidate_targeting_current_close_group() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA2); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, true, None); + let mut budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; + + paid_list.set_record_out_of_range(&key); + let first_seen = paid_list + .record_out_of_range_since(&key) + .expect("first seen"); + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + instant_after(first_seen, config.prune_hysteresis_duration), + &mut budget, + ); + + let RecordPruneKeyState::Candidate(PruneCandidateDisposition::Auditable(candidate)) = + outcome.state + else { + panic!("record past hysteresis must be an auditable candidate"); + }; + let targets: HashSet = candidate.target_peers.iter().copied().collect(); + let expected: HashSet = strict_close_peers.iter().copied().collect(); + assert_eq!( + targets, expected, + "audit targets must be the current strict close group, unfiltered by repair proofs" + ); + assert_eq!( + budget, + MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS - PROD_CLOSE_GROUP, + "the scheduled audit must consume challenge budget" + ); + } + + /// #5: a retained commitment vetoes deletion after candidacy without + /// touching the first-seen timestamp or the audit budget. + #[tokio::test] + async fn held_commitment_defers_candidate_without_losing_state() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA3); + let commitment = held_commitment_state(key, b"still committed"); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, true, Some(&commitment)); + let mut budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; + + paid_list.set_record_out_of_range(&key); + let first_seen = paid_list + .record_out_of_range_since(&key) + .expect("first seen"); + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + instant_after(first_seen, config.prune_hysteresis_duration), + &mut budget, + ); + + assert!(matches!( + outcome.state, + RecordPruneKeyState::Candidate(PruneCandidateDisposition::HeldByCommitment) + )); + assert_eq!( + paid_list.record_out_of_range_since(&key), + Some(first_seen), + "the retention veto must not restart hysteresis" + ); + assert_eq!(budget, MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS); + } + + /// #6: bootstrap state defers the audit but preserves candidacy and the + /// first-seen timestamp. + #[tokio::test] + async fn bootstrap_gate_defers_audit_but_preserves_candidacy() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA4); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, false, None); + let mut budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; + + paid_list.set_record_out_of_range(&key); + let first_seen = paid_list + .record_out_of_range_since(&key) + .expect("first seen"); + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + instant_after(first_seen, config.prune_hysteresis_duration), + &mut budget, + ); + + assert!(matches!( + outcome.state, + RecordPruneKeyState::Candidate(PruneCandidateDisposition::BootstrapDeferred) + )); + assert_eq!( + paid_list.record_out_of_range_since(&key), + Some(first_seen), + "bootstrap deferral must preserve the first-seen time" + ); + assert_eq!(budget, MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS); + } + + /// #7: an exhausted per-pass challenge budget defers the audit but + /// preserves candidacy and the first-seen timestamp. + #[tokio::test] + async fn exhausted_audit_budget_defers_audit_but_preserves_candidacy() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA5); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, true, None); + let mut budget = PROD_CLOSE_GROUP - 1; + + paid_list.set_record_out_of_range(&key); + let first_seen = paid_list + .record_out_of_range_since(&key) + .expect("first seen"); + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + instant_after(first_seen, config.prune_hysteresis_duration), + &mut budget, + ); + + assert!(matches!( + outcome.state, + RecordPruneKeyState::Candidate(PruneCandidateDisposition::BudgetDeferred) + )); + assert_eq!( + paid_list.record_out_of_range_since(&key), + Some(first_seen), + "budget deferral must preserve the first-seen time" + ); + assert_eq!(budget, PROD_CLOSE_GROUP - 1, "no budget may be consumed"); + } + + /// #12 (part): moving back inside the retention width clears the + /// out-of-range state. + #[tokio::test] + async fn record_back_in_range_clears_out_of_range_state() { + let (paid_list, _dir) = test_paid_list().await; + let config = ReplicationConfig::default(); + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xA6); + let mut admission_peers = peer_ids(PROD_RETENTION_WIDTH - 1); + admission_peers.push(self_id); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let deps = record_deps(&self_id, &paid_list, &config, true, None); + let mut budget = MAX_PRUNE_AUDIT_CHALLENGES_PER_PASS; + + paid_list.set_record_out_of_range(&key); + let outcome = evaluate_record_prune_key( + &deps, + &key, + &admission_peers, + &strict_close_peers, + Instant::now(), + &mut budget, + ); + + assert!(matches!( + outcome.state, + RecordPruneKeyState::InRange { cleared: true } + )); + assert!( + paid_list.record_out_of_range_since(&key).is_none(), + "re-entering range must clear the out-of-range timestamp" + ); + } + + // -- Pre-deletion revalidation ------------------------------------------- + + fn revalidation_inputs<'a>( + self_id: &'a PeerId, + first_seen: Option, + held_by_commitment: bool, + storage_admission_peers: &'a [PeerId], + strict_close_peers: &'a [PeerId], + now: Instant, + ) -> PruneRevalidationInputs<'a> { + PruneRevalidationInputs { + self_id, + first_seen, + prune_hysteresis_duration: ReplicationConfig::default().prune_hysteresis_duration, + held_by_commitment, + storage_admission_peers, + strict_close_peers, + now, + } + } + + fn matured(first_seen: Instant) -> Instant { + instant_after( + first_seen, + ReplicationConfig::default().prune_hysteresis_duration, + ) + } + + /// #9 + #10: six positive proofs from the current seven-strong close group + /// permit deletion; five do not. + #[test] + fn revalidation_deletes_at_six_of_seven_and_retains_at_five() { + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xB0); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let candidate = candidate(key, strict_close_peers.clone()); + let first_seen = Instant::now(); + let now = matured(first_seen); + + for (proofs, expect_delete) in [(6usize, true), (5usize, false)] { + let mut present_by_key = HashMap::new(); + present_by_key.insert( + key, + strict_close_peers[..proofs] + .iter() + .copied() + .collect::>(), + ); + let inputs = revalidation_inputs( + &self_id, + Some(first_seen), + false, + &admission_peers, + &strict_close_peers, + now, + ); + let outcome = revalidate_record_prune_candidate(&candidate, &present_by_key, &inputs); + if expect_delete { + assert!( + matches!(outcome, PruneRevalidationOutcome::Delete), + "6 of 7 proofs must permit deletion" + ); + } else { + assert!( + matches!(outcome, PruneRevalidationOutcome::AuditFailed), + "5 of 7 proofs must retain the record" + ); + } + } + } + + /// #12: self moving back inside the retention width between the audit and + /// deletion clears the state and prevents deletion, even with full proofs. + #[test] + fn revalidation_clears_and_prevents_deletion_when_back_in_range() { + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xB1); + let mut admission_peers = peer_ids(PROD_RETENTION_WIDTH - 1); + admission_peers.push(self_id); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let candidate = candidate(key, strict_close_peers.clone()); + let mut present_by_key = HashMap::new(); + present_by_key.insert(key, strict_close_peers.iter().copied().collect()); + let first_seen = Instant::now(); + let inputs = revalidation_inputs( + &self_id, + Some(first_seen), + false, + &admission_peers, + &strict_close_peers, + matured(first_seen), + ); + + let outcome = revalidate_record_prune_candidate(&candidate, &present_by_key, &inputs); + + assert!(matches!( + outcome, + PruneRevalidationOutcome::ClearedBackInRange + )); + } + + /// #12: a strict-close-group membership change after the audit round + /// invalidates stale positive reports — only proofs from CURRENT members + /// count toward the CURRENT group's threshold. + #[test] + fn revalidation_rejects_proofs_from_stale_close_group() { + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xB2); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let old_close_group = peer_ids(PROD_CLOSE_GROUP); + // Six positive proofs — a passing audit against the OLD group. + let mut present_by_key = HashMap::new(); + present_by_key.insert( + key, + old_close_group[..6].iter().copied().collect::>(), + ); + // The close group churned: only three audited peers remain members. + let churned_close_group: Vec = old_close_group[4..7] + .iter() + .copied() + .chain((20..24).map(peer_id_from_byte)) + .collect(); + let candidate = candidate(key, old_close_group); + let first_seen = Instant::now(); + let inputs = revalidation_inputs( + &self_id, + Some(first_seen), + false, + &admission_peers, + &churned_close_group, + matured(first_seen), + ); + + let outcome = revalidate_record_prune_candidate(&candidate, &present_by_key, &inputs); + + assert!( + matches!(outcome, PruneRevalidationOutcome::AuditFailed), + "stale proofs must not satisfy the churned current close group" + ); + } + + /// A commitment re-gossiped between candidate selection and deletion + /// vetoes the deletion even with a full set of proofs (TOCTOU re-check). + #[test] + fn revalidation_vetoes_deletion_for_recommitted_key() { + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xB3); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let candidate = candidate(key, strict_close_peers.clone()); + let mut present_by_key = HashMap::new(); + present_by_key.insert(key, strict_close_peers.iter().copied().collect()); + let first_seen = Instant::now(); + let inputs = revalidation_inputs( + &self_id, + Some(first_seen), + true, + &admission_peers, + &strict_close_peers, + matured(first_seen), + ); + + let outcome = revalidate_record_prune_candidate(&candidate, &present_by_key, &inputs); + + assert!(matches!( + outcome, + PruneRevalidationOutcome::HeldByCommitment + )); + } + + /// A concurrently cleared or reset out-of-range timestamp downgrades the + /// candidate back to hysteresis-pending instead of deleting. + #[test] + fn revalidation_requires_hysteresis_to_still_hold() { + let self_id = peer_id_from_byte(SELF_BYTE); + let key = key_from_byte(0xB4); + let admission_peers = peer_ids(PROD_RETENTION_WIDTH); + let strict_close_peers = peer_ids(PROD_CLOSE_GROUP); + let candidate = candidate(key, strict_close_peers.clone()); + let mut present_by_key = HashMap::new(); + present_by_key.insert( + key, + strict_close_peers + .iter() + .copied() + .collect::>(), + ); + let now = Instant::now(); + + for first_seen in [None, Some(now)] { + let inputs = revalidation_inputs( + &self_id, + first_seen, + false, + &admission_peers, + &strict_close_peers, + now, + ); + let outcome = revalidate_record_prune_candidate(&candidate, &present_by_key, &inputs); + assert!(matches!( + outcome, + PruneRevalidationOutcome::HysteresisPending + )); + } + } + + // -- Prune-audit response grading (#11) ----------------------------------- + + const TEST_CHALLENGE_ID: u64 = 0x00C0_FFEE; + const TEST_NONCE: [u8; 32] = [0xAB; 32]; + const TEST_RECORD_BYTES: &[u8] = b"prune audit record bytes"; + + fn digests_response(challenge_id: u64, digests: Vec<[u8; 32]>) -> ReplicationMessage { + ReplicationMessage { + request_id: challenge_id, + body: ReplicationMessageBody::AuditResponse(AuditResponse::Digests { + challenge_id, + digests, + }), + } + } + + fn graded_status(peer: &PeerId, key: &XorName, msg: ReplicationMessage) -> PruneAuditStatus { + prune_audit_response_status( + msg, + TEST_CHALLENGE_ID, + peer, + key, + &TEST_NONCE, + TEST_RECORD_BYTES, + ) + } + + #[test] + fn prune_audit_status_accepts_only_a_matching_digest() { + let peer = peer_id_from_byte(1); + let key = key_from_byte(0xC0); + let valid = compute_audit_digest(&TEST_NONCE, peer.as_bytes(), &key, TEST_RECORD_BYTES); + + let status = graded_status( + &peer, + &key, + digests_response(TEST_CHALLENGE_ID, vec![valid]), + ); + + assert_eq!(status, PruneAuditStatus::Proven); + } + + #[test] + fn prune_audit_status_rejects_absent_malformed_and_mismatching_responses() { + let peer = peer_id_from_byte(1); + let key = key_from_byte(0xC1); + let valid = compute_audit_digest(&TEST_NONCE, peer.as_bytes(), &key, TEST_RECORD_BYTES); + + // Absent-key sentinel is a negative answer. + let absent = digests_response(TEST_CHALLENGE_ID, vec![ABSENT_KEY_DIGEST]); + assert_eq!(graded_status(&peer, &key, absent), PruneAuditStatus::Failed); + + // A digest over different bytes does not prove possession. + let mismatch = digests_response(TEST_CHALLENGE_ID, vec![[0x11; 32]]); + assert_eq!( + graded_status(&peer, &key, mismatch), + PruneAuditStatus::Failed + ); + + // A malformed digest count never counts. + let wrong_count = digests_response(TEST_CHALLENGE_ID, vec![valid, valid]); + assert_eq!( + graded_status(&peer, &key, wrong_count), + PruneAuditStatus::Failed + ); + + // A response bound to a different challenge never counts. + let stale_challenge = digests_response(TEST_CHALLENGE_ID + 1, vec![valid]); + assert_eq!( + graded_status(&peer, &key, stale_challenge), + PruneAuditStatus::Failed + ); + + // An explicit rejection never counts. + let rejected = ReplicationMessage { + request_id: TEST_CHALLENGE_ID, + body: ReplicationMessageBody::AuditResponse(AuditResponse::Rejected { + challenge_id: TEST_CHALLENGE_ID, + reason: "test".to_string(), + }), + }; + assert_eq!( + graded_status(&peer, &key, rejected), + PruneAuditStatus::Failed + ); + + // An unexpected message type never counts. + let unexpected = ReplicationMessage { + request_id: TEST_CHALLENGE_ID, + body: ReplicationMessageBody::AuditChallenge(AuditChallenge { + challenge_id: TEST_CHALLENGE_ID, + nonce: TEST_NONCE, + challenged_peer_id: *peer.as_bytes(), + keys: vec![key], + }), + }; + assert_eq!( + graded_status(&peer, &key, unexpected), + PruneAuditStatus::Failed + ); + } + + #[test] + fn prune_audit_status_bootstrap_claim_is_not_a_positive_proof() { + let peer = peer_id_from_byte(1); + let key = key_from_byte(0xC2); + + let bootstrapping = ReplicationMessage { + request_id: TEST_CHALLENGE_ID, + body: ReplicationMessageBody::AuditResponse(AuditResponse::Bootstrapping { + challenge_id: TEST_CHALLENGE_ID, + }), + }; + assert_eq!( + graded_status(&peer, &key, bootstrapping), + PruneAuditStatus::Bootstrapping + ); + + let mismatched = ReplicationMessage { + request_id: TEST_CHALLENGE_ID, + body: ReplicationMessageBody::AuditResponse(AuditResponse::Bootstrapping { + challenge_id: TEST_CHALLENGE_ID + 1, + }), + }; + assert_eq!( + graded_status(&peer, &key, mismatched), + PruneAuditStatus::Failed + ); + } } diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index df9e05a6..14d3b5a7 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -10,7 +10,7 @@ use super::TestHarness; use ant_node::client::compute_address; use ant_node::replication::commitment_state::{BuiltCommitment, ResponderCommitmentState}; use ant_node::replication::config::{ - storage_admission_width, K_BUCKET_SIZE, REPAIR_HINT_MIN_AGE, REPLICATION_PROTOCOL_ID, + storage_admission_width, K_BUCKET_SIZE, REPLICATION_PROTOCOL_ID, }; use ant_node::replication::protocol::{ compute_audit_digest, AuditChallenge, AuditResponse, FetchRequest, FetchResponse, @@ -24,9 +24,8 @@ use ant_node::ReplicationConfig; use saorsa_core::identity::PeerId; use saorsa_core::{P2PNode, TrustEvent}; use serial_test::serial; -use std::collections::HashSet; use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Duration; use tokio::sync::RwLock; /// Maximum time to wait for replication propagation in tests. @@ -164,42 +163,6 @@ async fn store_record_on_peers( } } -async fn record_repair_proofs_for_peers( - repair_proofs: &Arc>, - p2p_node: &Arc, - config: &ReplicationConfig, - peers: &[PeerId], - key: &[u8; 32], - hinted_at_epoch: u64, -) -> Instant { - let close_peers: HashSet = p2p_node - .dht_manager() - .find_closest_nodes_local_with_self(key, config.close_group_size) - .await - .iter() - .map(|node| node.peer_id) - .collect(); - let mut proofs = repair_proofs.write().await; - let hinted_at = Instant::now(); - let repair_proof_now = hinted_at - .checked_add(REPAIR_HINT_MIN_AGE) - .unwrap_or(hinted_at); - for peer in peers { - assert!( - proofs.record_replica_hint_sent_at( - *peer, - *key, - &close_peers, - hinted_at_epoch, - hinted_at - ), - "test target should be in close group for repair-proof recording" - ); - } - drop(proofs); - repair_proof_now -} - /// Fresh write happy path (Section 18 #1). /// /// Store a chunk on a node that has a `ReplicationEngine`, manually call @@ -940,13 +903,12 @@ async fn test_audit_absent_key_returns_sentinel() { /// Prune pass requires remote storage proofs before deleting local records. /// /// This drives `run_prune_pass` against live nodes so prune-confirmation audits -/// travel over the real replication request/response path. +/// travel over the real replication request/response path. The repair-proof +/// table stays EMPTY throughout: prune candidacy and prune-audit target +/// selection must not depend on prior neighbor-sync repair hints. #[tokio::test] #[serial] async fn test_prune_pass_requires_remote_confirmation_before_delete() { - const HINT_EPOCH: u64 = 7; - const CURRENT_EPOCH: u64 = HINT_EPOCH + 1; - let harness = TestHarness::setup_minimal().await.expect("setup"); harness.warmup_dht().await.expect("warmup"); @@ -970,7 +932,8 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { let pruner_peer = *pruner_p2p.peer_id(); // Even with all target peers storing the record, pruning must be blocked - // while remote prune audits are not allowed during bootstrap/drain. + // while remote prune audits are not allowed during bootstrap/drain — but + // the deferral must preserve candidacy and the first-seen timestamp. let (gate_content, gate_address, gate_targets) = find_remote_prune_candidate(&harness, pruner_idx, close_group_size, "bootstrap-gate").await; pruner_storage @@ -978,15 +941,6 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { .await .expect("put gate record on pruner"); store_record_on_peers(&harness, &gate_targets, &gate_address, &gate_content).await; - let gate_repair_proof_now = record_repair_proofs_for_peers( - &repair_proofs, - &pruner_p2p, - &config, - &gate_targets, - &gate_address, - HINT_EPOCH, - ) - .await; let blocked = pruning::run_prune_pass_with_context(pruning::PrunePassContext { self_id: &pruner_peer, @@ -996,17 +950,43 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, - repair_proof_now: Some(gate_repair_proof_now), allow_remote_prune_audits: false, commitment_state: None, }) .await; assert_eq!(blocked.records_pruned, 0); + assert_eq!( + blocked.records_candidates, 1, + "the bootstrap gate must not remove candidate status" + ); + assert_eq!(blocked.records_bootstrap_deferred, 1); assert!( pruner_storage.exists(&gate_address).expect("exists"), "record must not be pruned before remote audits are allowed" ); + let first_seen = pruner_paid_list + .record_out_of_range_since(&gate_address) + .expect("bootstrap-deferred record keeps its out-of-range timestamp"); + + // A second blocked pass must not restart the hysteresis clock. + let blocked_again = 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, + allow_remote_prune_audits: false, + commitment_state: None, + }) + .await; + assert_eq!(blocked_again.records_pruned, 0); + assert_eq!( + pruner_paid_list.record_out_of_range_since(&gate_address), + Some(first_seen), + "repeated deferrals must preserve the original first-seen time" + ); let confirmed = pruning::run_prune_pass_with_context(pruning::PrunePassContext { self_id: &pruner_peer, @@ -1016,12 +996,11 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, - repair_proof_now: Some(gate_repair_proof_now), allow_remote_prune_audits: true, commitment_state: None, }) .await; + assert_eq!(confirmed.records_audits_attempted, 1); assert_eq!(confirmed.records_pruned, 1); assert!( !pruner_storage.exists(&gate_address).expect("exists"), @@ -1043,15 +1022,6 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { &missing_content, ) .await; - let missing_repair_proof_now = record_repair_proofs_for_peers( - &repair_proofs, - &pruner_p2p, - &config, - &missing_targets, - &missing_address, - HINT_EPOCH, - ) - .await; let incomplete = pruning::run_prune_pass_with_context(pruning::PrunePassContext { self_id: &pruner_peer, @@ -1061,13 +1031,15 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, - repair_proof_now: Some(missing_repair_proof_now), allow_remote_prune_audits: true, commitment_state: None, }) .await; assert_eq!(incomplete.records_pruned, 0); + assert_eq!( + incomplete.records_audit_below_threshold, 1, + "a below-threshold audit must be reported as such, not silently dropped" + ); assert!( pruner_storage.exists(&missing_address).expect("exists"), "record must remain local while any target peer lacks proof" @@ -1089,8 +1061,6 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, - repair_proof_now: Some(missing_repair_proof_now), allow_remote_prune_audits: true, commitment_state: None, }) @@ -1108,16 +1078,14 @@ async fn test_prune_pass_requires_remote_confirmation_before_delete() { /// for, but which is still committed under a recently-gossiped commitment, must /// NOT be deleted — the storage-commitment audit's round-2 byte challenge could /// still demand it, and deleting would turn an honest node's reply into an -/// `Absent` confirmed failure. Once it is no longer committed (e.g. it has aged -/// out of the retention window, simulated here by passing `None`), the same -/// out-of-range record becomes prunable. Drives the real `run_prune_pass` -/// against live nodes. +/// `Absent` confirmed failure. The veto applies to DELETION only: the +/// out-of-range timer starts immediately even while the key is held. Once it is +/// no longer committed (e.g. it has aged out of the retention window, simulated +/// here by passing `None`), the same out-of-range record becomes prunable. +/// Drives the real `run_prune_pass` against live nodes. #[tokio::test] #[serial] async fn test_prune_veto_for_committed_out_of_range_key() { - const HINT_EPOCH: u64 = 7; - const CURRENT_EPOCH: u64 = HINT_EPOCH + 1; - let harness = TestHarness::setup_minimal().await.expect("setup"); harness.warmup_dht().await.expect("warmup"); @@ -1148,18 +1116,6 @@ async fn test_prune_veto_for_committed_out_of_range_key() { .await .expect("put record on pruner"); store_record_on_peers(&harness, &targets, &address, &content).await; - // Mature repair proofs (hinted_at + REPAIR_HINT_MIN_AGE) so main's prune-proof - // gate treats the record as fully prunable — leaving the retention veto as the - // ONLY thing that can keep it. Same pattern as the prune-threshold tests below. - let repair_proof_now = record_repair_proofs_for_peers( - &repair_proofs, - &pruner_p2p, - &config, - &targets, - &address, - HINT_EPOCH, - ) - .await; // A retained commitment that COMMITS to the out-of-range key (as if we // gossiped it just before the key left our range). A throwaway keypair is @@ -1182,7 +1138,7 @@ async fn test_prune_veto_for_committed_out_of_range_key() { assert!(committed.is_held(&address), "test setup: key must be held"); // With the key still committed, an otherwise-fully-prunable out-of-range - // record is VETOED. + // record is VETOED — but the out-of-range timer still starts. let vetoed = pruning::run_prune_pass_with_context(pruning::PrunePassContext { self_id: &pruner_peer, storage: &pruner_storage, @@ -1191,9 +1147,7 @@ async fn test_prune_veto_for_committed_out_of_range_key() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, allow_remote_prune_audits: true, - repair_proof_now: Some(repair_proof_now), commitment_state: Some(&committed), }) .await; @@ -1201,6 +1155,14 @@ async fn test_prune_veto_for_committed_out_of_range_key() { vetoed.records_pruned, 0, "a key still committed under a recent commitment must not be pruned" ); + assert_eq!(vetoed.records_marked_out_of_range, 1); + assert_eq!(vetoed.records_held_by_commitment, 1); + assert!( + pruner_paid_list + .record_out_of_range_since(&address) + .is_some(), + "a retained commitment must not prevent the out-of-range timer from starting" + ); assert!( pruner_storage.exists(&address).expect("exists"), "the vetoed record must remain on disk" @@ -1216,9 +1178,7 @@ async fn test_prune_veto_for_committed_out_of_range_key() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, allow_remote_prune_audits: true, - repair_proof_now: Some(repair_proof_now), commitment_state: None, }) .await; @@ -1234,32 +1194,26 @@ async fn test_prune_veto_for_committed_out_of_range_key() { 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). +/// Production-width prune liveness: close group 7, storage-retention width 9 +/// (7 + `STORAGE_ADMISSION_MARGIN`), and a chunk the pruner holds (as if +/// admitted under the wider client-PUT gate) that is now outside width 9. /// -/// 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. +/// The prune-confirmation audit must reach the CURRENT key-nearest close +/// group directly from the routing table, with a completely EMPTY +/// `RepairProofs` table: no overlap between the pruner's neighbor-sync set +/// (self-nearest peers) and the key-nearest close group may be assumed. The +/// old repair-proof gate silently deferred candidacy here forever — the +/// production "pruning is hardly taking place" incident. /// -/// This test pins both sides of the gate: +/// This test also pins both sides of the 6-of-7 possession threshold: /// - below the threshold (5 of 7 proofs) the record must never be deleted, -/// no matter how many passes run; +/// no matter how many passes run (absent peers answer +/// `ABSENT_KEY_DIGEST`, which never counts positively); /// - 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. @@ -1276,6 +1230,8 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { ..ReplicationConfig::default() }; let sync_state = Arc::new(RwLock::new(NeighborSyncState::new_cycle(vec![]))); + // Deliberately empty and never populated: candidacy and target selection + // must not depend on neighbor-sync repair hints. let repair_proofs = Arc::new(RwLock::new(RepairProofs::new())); let pruner = harness.test_node(pruner_idx).expect("pruner"); @@ -1291,6 +1247,8 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { ); let pruner_peer = *pruner_p2p.peer_id(); + // `find_remote_prune_candidate` guarantees the pruner is outside the + // width-9 storage-retention group for the key (and outside the strict 7). let (content, address, targets) = find_remote_prune_candidate(&harness, pruner_idx, PROD_CLOSE_GROUP_SIZE, "quorum-stored") .await; @@ -1300,8 +1258,6 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { .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], @@ -1309,19 +1265,10 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { &content, ) .await; - let repair_proof_now = 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. + // pass must retain it — while still ATTEMPTING the audit each pass. for pass in 0..3 { let result = pruning::run_prune_pass_with_context(pruning::PrunePassContext { self_id: &pruner_peer, @@ -1331,16 +1278,22 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, allow_remote_prune_audits: true, - repair_proof_now: Some(repair_proof_now), commitment_state: None, }) .await; + assert_eq!( + result.records_audits_attempted, 1, + "pass {pass}: the candidate must be audited despite the empty repair-proof table", + ); assert_eq!( result.records_pruned, 0, "pass {pass}: a record below the proof threshold must never prune", ); + assert_eq!( + result.records_audit_below_threshold, 1, + "pass {pass}: the failed confirmation must surface in telemetry", + ); assert!( pruner_storage.exists(&address).expect("exists"), "pass {pass}: record should remain on the out-of-range node", @@ -1367,9 +1320,7 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: CURRENT_EPOCH, allow_remote_prune_audits: true, - repair_proof_now: Some(repair_proof_now), commitment_state: None, }) .await; @@ -1382,6 +1333,121 @@ async fn prune_deletes_at_proof_threshold_and_retains_below_it() { harness.teardown().await.expect("teardown"); } +/// Hysteresis lifecycle against live nodes: a record outside the retention +/// width is marked immediately, is NOT a candidate while the hysteresis is +/// running, and becomes a candidate that prunes — with an empty +/// `RepairProofs` table — once continuously out of range for the full +/// (test-shortened) hysteresis. +#[tokio::test] +#[serial] +async fn prune_marks_immediately_and_candidacy_waits_for_hysteresis() { + /// Test stand-in for the 3-day production hysteresis. Long enough that the + /// two pre-maturity prune passes reliably run inside the window, short + /// enough to keep the test fast. + const TEST_HYSTERESIS: Duration = Duration::from_secs(2); + + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let pruner_idx = 3; + let close_group_size = 2; + let config = ReplicationConfig { + prune_hysteresis_duration: TEST_HYSTERESIS, + ..prune_test_config(close_group_size) + }; + 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_storage = pruner.ant_protocol.as_ref().expect("protocol").storage(); + let pruner_paid_list = Arc::clone( + pruner + .replication_engine + .as_ref() + .expect("engine") + .paid_list(), + ); + let pruner_peer = *pruner_p2p.peer_id(); + + let (content, address, targets) = + find_remote_prune_candidate(&harness, pruner_idx, close_group_size, "hysteresis").await; + pruner_storage + .put(&address, &content) + .await + .expect("put record on pruner"); + store_record_on_peers(&harness, &targets, &address, &content).await; + + // Pass 1: the record is marked out-of-range immediately, but the running + // hysteresis keeps it from candidacy. + let marked = 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, + allow_remote_prune_audits: true, + commitment_state: None, + }) + .await; + assert_eq!(marked.records_marked_out_of_range, 1); + assert_eq!(marked.records_hysteresis_pending, 1); + assert_eq!(marked.records_candidates, 0); + assert_eq!(marked.records_pruned, 0); + let first_seen = pruner_paid_list + .record_out_of_range_since(&address) + .expect("out-of-range timestamp set on first observation"); + + // Pass 2 (immediately after): still pending, timer NOT restarted. + let pending = 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, + allow_remote_prune_audits: true, + commitment_state: None, + }) + .await; + assert_eq!(pending.records_marked_out_of_range, 0); + assert_eq!(pending.records_hysteresis_pending, 1); + assert_eq!(pending.records_pruned, 0); + assert_eq!( + pruner_paid_list.record_out_of_range_since(&address), + Some(first_seen), + "continuously out-of-range records keep their original first-seen time" + ); + + // Pass 3 (after the hysteresis): candidate, audited, and pruned — with an + // empty repair-proof table. + tokio::time::sleep(TEST_HYSTERESIS + Duration::from_millis(100)).await; + let matured = 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, + allow_remote_prune_audits: true, + commitment_state: None, + }) + .await; + assert_eq!(matured.records_candidates, 1); + assert_eq!(matured.records_audits_attempted, 1); + assert_eq!( + matured.records_pruned, 1, + "the matured candidate must prune without any repair-hint history" + ); + 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. @@ -1441,9 +1507,7 @@ async fn paid_prune_requires_paid_close_group_confirmations() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: 1, allow_remote_prune_audits: true, - repair_proof_now: None, commitment_state: None, }) .await; @@ -1477,9 +1541,7 @@ async fn paid_prune_requires_paid_close_group_confirmations() { config: &config, sync_state: &sync_state, repair_proofs: &repair_proofs, - current_sync_epoch: 1, allow_remote_prune_audits: true, - repair_proof_now: None, commitment_state: None, }) .await;