fix(replication): require aged repair hints before audits

src/replication/audit.rs

@@ -4,6 +4,7 @@
 
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
+use std::time::Instant;
 
 use crate::logging::{debug, info, warn};
 use rand::seq::SliceRandom;
@@ -167,11 +168,13 @@ pub async fn audit_tick_with_repair_proofs(
 
     let peer_keys = {
         let mut proofs = repair_proofs.write().await;
+        let now = Instant::now();
         mature_audit_keys_for_peer(
             &challenged_peer,
             sampled_key_groups,
             &mut proofs,
             current_sync_epoch,
+            now,
         )
     };
 
@@ -349,12 +352,19 @@ fn mature_audit_keys_for_peer(
     sampled_key_groups: Vec<(XorName, HashSet<PeerId>)>,
     repair_proofs: &mut RepairProofs,
     current_sync_epoch: u64,
+    now: Instant,
 ) -> Vec<XorName> {
     sampled_key_groups
         .into_iter()
         .filter_map(|(key, close_peers)| {
             repair_proofs
-                .has_mature_replica_hint(challenged_peer, &key, &close_peers, current_sync_epoch)
+                .has_mature_replica_hint(
+                    challenged_peer,
+                    &key,
+                    &close_peers,
+                    current_sync_epoch,
+                    now,
+                )
                 .then_some(key)
         })
         .collect()
@@ -720,6 +730,7 @@ pub async fn handle_audit_challenge(
 #[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
 mod tests {
     use super::*;
+    use crate::replication::config::REPAIR_HINT_MIN_AGE;
     use crate::replication::protocol::compute_audit_digest;
     use crate::replication::types::{BootstrapClaimObservation, NeighborSyncState};
     use crate::storage::LmdbStorageConfig;
@@ -1367,6 +1378,7 @@ mod tests {
         const MISSING_PROOF_KEY_BYTE: u8 = 0xB3;
         const STABLE_CHURN_KEY_BYTE: u8 = 0xB4;
         const EVICTED_KEY_BYTE: u8 = 0xB5;
+        const FRESH_HINT_KEY_BYTE: u8 = 0xB6;
         const XOR_NAME_LEN: usize = 32;
 
         let challenged_peer = peer_id_from_bytes([CHALLENGED_PEER_BYTE; XOR_NAME_LEN]);
@@ -1377,34 +1389,50 @@ mod tests {
         let missing_proof_key = [MISSING_PROOF_KEY_BYTE; XOR_NAME_LEN];
         let stable_churn_key = [STABLE_CHURN_KEY_BYTE; XOR_NAME_LEN];
         let evicted_key = [EVICTED_KEY_BYTE; XOR_NAME_LEN];
+        let fresh_hint_key = [FRESH_HINT_KEY_BYTE; XOR_NAME_LEN];
         let close_group = HashSet::from([challenged_peer, other_peer]);
         let changed_close_group = HashSet::from([challenged_peer, new_peer]);
         let evicted_close_group = HashSet::from([other_peer, new_peer]);
         let mut repair_proofs = RepairProofs::new();
+        let mature_hinted_at = Instant::now();
+        let now = mature_hinted_at
+            .checked_add(REPAIR_HINT_MIN_AGE)
+            .unwrap_or(mature_hinted_at);
 
-        assert!(repair_proofs.record_replica_hint_sent(
+        assert!(repair_proofs.record_replica_hint_sent_at(
             challenged_peer,
             mature_key,
             &close_group,
             HINT_EPOCH,
+            mature_hinted_at,
         ));
-        assert!(repair_proofs.record_replica_hint_sent(
+        assert!(repair_proofs.record_replica_hint_sent_at(
             challenged_peer,
             same_epoch_key,
             &close_group,
             CURRENT_EPOCH,
+            mature_hinted_at,
         ));
-        assert!(repair_proofs.record_replica_hint_sent(
+        assert!(repair_proofs.record_replica_hint_sent_at(
             challenged_peer,
             stable_churn_key,
             &close_group,
             HINT_EPOCH,
+            mature_hinted_at,
         ));
-        assert!(repair_proofs.record_replica_hint_sent(
+        assert!(repair_proofs.record_replica_hint_sent_at(
             challenged_peer,
             evicted_key,
             &close_group,
             HINT_EPOCH,
+            mature_hinted_at,
+        ));
+        assert!(repair_proofs.record_replica_hint_sent_at(
+            challenged_peer,
+            fresh_hint_key,
+            &close_group,
+            HINT_EPOCH,
+            now,
         ));
 
         let sampled_key_groups = vec![
@@ -1413,18 +1441,20 @@ mod tests {
             (missing_proof_key, close_group.clone()),
             (stable_churn_key, changed_close_group),
             (evicted_key, evicted_close_group),
+            (fresh_hint_key, close_group.clone()),
         ];
         let peer_keys = mature_audit_keys_for_peer(
             &challenged_peer,
             sampled_key_groups,
             &mut repair_proofs,
             CURRENT_EPOCH,
+            now,
         );
 
         assert_eq!(
             peer_keys,
             vec![mature_key, stable_churn_key],
-            "mature proofs for stable close-group peers should become audit keys, while same-epoch, missing, and evicted-peer proofs should not"
+            "mature proofs for stable close-group peers should become audit keys, while same-epoch, fresh, missing, and evicted-peer proofs should not"
         );
     }
 

src/replication/config.rs

@@ -58,6 +58,13 @@ const NEIGHBOR_SYNC_COOLDOWN_SECS: u64 = 60 * 60; // 1 hour
 /// Per-peer minimum spacing between successive syncs with the same peer.
 pub const NEIGHBOR_SYNC_COOLDOWN: Duration = Duration::from_secs(NEIGHBOR_SYNC_COOLDOWN_SECS);
 
+/// Minimum age for a replica repair hint before the hinted peer can be audited
+/// for that key.
+const REPAIR_HINT_MIN_AGE_SECS: u64 = 60 * 60; // 1 hour
+/// Minimum age for a replica repair hint before the hinted peer can be audited
+/// for that key.
+pub const REPAIR_HINT_MIN_AGE: Duration = Duration::from_secs(REPAIR_HINT_MIN_AGE_SECS);
+
 /// Minimum self-lookup cadence.
 const SELF_LOOKUP_INTERVAL_MIN_SECS: u64 = 5 * 60;
 /// Maximum self-lookup cadence.

src/replication/mod.rs

@@ -1675,8 +1675,9 @@ async fn run_neighbor_sync_round(
     // prune pass and DHT snapshot so other tasks are not starved.
     let cycle_complete = sync_state.read().await.is_cycle_complete();
     if cycle_complete {
-        // A completed local neighbor-sync cycle matures key-specific repair
-        // proofs recorded in earlier epochs.
+        // A completed local neighbor-sync cycle advances the epoch component
+        // of repair-proof maturity. The per-key wall-clock minimum age is
+        // checked when audits are selected.
         {
             let mut history = sync_history.write().await;
             for record in history.values_mut() {
@@ -1702,6 +1703,8 @@ async fn run_neighbor_sync_round(
             sync_state,
             repair_proofs,
             current_sync_epoch,
+            #[cfg(any(test, feature = "test-utils"))]
+            repair_proof_now: None,
             allow_remote_prune_audits,
         })
         .await;

src/replication/pruning.rs

@@ -81,8 +81,11 @@ pub struct PrunePassContext<'a> {
     pub sync_state: &'a Arc<RwLock<NeighborSyncState>>,
     /// Key-specific repair proofs used to gate prune-confirmation audits.
     pub repair_proofs: &'a Arc<RwLock<RepairProofs>>,
-    /// Current local neighbor-sync cycle epoch.
+    /// 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<Instant>,
     /// Whether remote prune-confirmation audits are allowed this pass.
     pub allow_remote_prune_audits: bool,
 }
@@ -229,6 +232,8 @@ pub async fn run_prune_pass(
         sync_state,
         repair_proofs: &repair_proofs,
         current_sync_epoch: 0,
+        #[cfg(any(test, feature = "test-utils"))]
+        repair_proof_now: None,
         allow_remote_prune_audits,
     })
     .await
@@ -413,12 +418,17 @@ async fn evaluate_record_prune_key(
     // 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<PeerId> = closest.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,
         &current_close_peers,
         ctx.repair_proofs,
         ctx.current_sync_epoch,
+        repair_proof_now,
     )
     .await;
     let proofs_needed = prune_proofs_needed(target_peers.len());
@@ -791,12 +801,13 @@ async fn peers_with_mature_repair_proofs(
     current_close_peers: &HashSet<PeerId>,
     repair_proofs: &Arc<RwLock<RepairProofs>>,
     current_sync_epoch: u64,
+    now: Instant,
 ) -> Vec<PeerId> {
     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)
+            proofs.has_mature_replica_hint(peer, key, current_close_peers, current_sync_epoch, now)
         })
         .copied()
         .collect()