Merge pull request #207 from grumbach/fix/capacity-aware-verification-cycle

fix(replication): stop a write-blocked node probing its close group

Author: jacderida

docs/adr/ADR-0011-capacity-gated-source-discovery.md

@@ -626,6 +626,45 @@ const PENDING_VERIFY_MAX_AGE_SECS: u64 = 30 * 60;
 /// Maximum age for pending-verification entries before stale eviction.
 pub const PENDING_VERIFY_MAX_AGE: Duration = Duration::from_secs(PENDING_VERIFY_MAX_AGE_SECS);
 
+/// How long a key waits for another look once this node's disk is **full**.
+///
+/// Only a full disk, not any refused write: a space query that fails says
+/// nothing about available space and keeps the ordinary retry schedule.
+///
+/// A node that cannot write cannot finish an acquisition, so before this gate
+/// the key came straight back. At [`VERIFICATION_REQUEST_TIMEOUT`] that was a
+/// close-group probe as often as every 15 s for every key the node owes — the
+/// requested delay, so cycle polling, round duration and bounded per-cycle
+/// selection make it an upper rate rather than an observed cadence. The owed set
+/// grows with the network rather than with anything this node does — 25.6% of a
+/// 195-service testnet, held full, produced 99.6% of every verification-request
+/// byte on the network (V2-987).
+///
+/// What this schedules is a *look*: the cycle re-reads local capacity and only
+/// probes if space has returned. A key that stays full and authorized sends no
+/// probes at all, because the gate stops the round before it is sent.
+///
+/// Five minutes is chosen against [`PENDING_VERIFY_MAX_AGE`], which it is well
+/// inside. That is not a guarantee that freed space is noticed while the entry
+/// lives: one deferred inside its final five minutes expires first, and a
+/// backlog can delay selection further.
+///
+/// This is deliberately a constant rather than a [`ReplicationConfig`] field.
+/// The struct is publicly re-exported and is not `#[non_exhaustive]`, so a new
+/// field would break downstream exhaustive construction for a knob nothing
+/// needs to tune at runtime.
+///
+/// Applied flat, through the ordinary `defer_pending`. A key deferred inside the
+/// last five minutes of its entry's life therefore expires at
+/// [`PENDING_VERIFY_MAX_AGE`] without a further look and comes back on the next
+/// neighbour-sync hint. An earlier revision clamped the delay to half the
+/// entry's remaining life to avoid that; it was cut because the extra looks it
+/// bought near expiry can themselves become ungated quorum rounds.
+const CAPACITY_BLOCKED_RETRY_SECS: u64 = 5 * 60;
+/// How long a key waits for another look once this node's disk is full.
+pub(crate) const CAPACITY_BLOCKED_RETRY: Duration =
+    Duration::from_secs(CAPACITY_BLOCKED_RETRY_SECS);
+
 /// Trust event weight for confirmed audit failures.
 pub const AUDIT_FAILURE_TRUST_WEIGHT: f64 = 5.0;
 
@@ -1803,4 +1842,28 @@ mod tests {
             "audit intervals should exhibit randomized jitter across samples"
         );
     }
+
+    /// The capacity stand-down has to be an order of magnitude above the retry
+    /// it replaces and still below the life of the entry it defers.
+    ///
+    /// A stand-down only a little above the request timeout would leave the
+    /// repeat cost the same order as before, which is the cost this change
+    /// exists to remove. At or past `PENDING_VERIFY_MAX_AGE` every deferral
+    /// would instead become an eviction, which is a different design with
+    /// different failure modes: the key would only return on a fresh
+    /// neighbour-sync hint rather than on its own schedule.
+    ///
+    /// What this does not show: that either gate uses the constant. It pins the
+    /// policy the constant encodes; the e2e proves the gates.
+    #[test]
+    fn capacity_blocked_retry_is_an_order_above_the_request_timeout_and_below_the_entry_lifetime() {
+        assert!(
+            CAPACITY_BLOCKED_RETRY >= VERIFICATION_REQUEST_TIMEOUT * 10,
+            "a stand-down near the request timeout leaves the repeat cost unchanged in order"
+        );
+        assert!(
+            CAPACITY_BLOCKED_RETRY < PENDING_VERIFY_MAX_AGE,
+            "a deferral at or past the entry lifetime is an eviction, not a deferral"
+        );
+    }
 }

src/replication/config.rs

@@ -97,7 +97,7 @@ use crate::replication::types::{
     NeighborSyncState, PeerSyncRecord, PresenceEvidence, RepairProofs, VerificationEntry,
     VerificationState,
 };
-use crate::storage::LmdbStorage;
+use crate::storage::{CapacityVerdict, LmdbStorage};
 use saorsa_core::identity::{NodeIdentity, PeerId};
 use saorsa_core::{DhtNetworkEvent, P2PEvent, P2PNode, TrustEvent};
 use saorsa_pqc::api::sig::{MlDsaSecretKey, MlDsaVariant};
@@ -2149,6 +2149,42 @@ impl ReplicationEngine {
         self.queues.read().await.contains_key(key)
     }
 
+    /// Test-only: place `key` into pending verification as though `hinter` had
+    /// just advertised it with a replica hint. Returns whether it was admitted.
+    ///
+    /// Enters the pipeline one stage earlier than
+    /// [`Self::enqueue_fetch_for_test`], so what the verification cycle itself
+    /// decides about the key is observable.
+    #[cfg(any(test, feature = "test-utils"))]
+    pub async fn enqueue_pending_verify_for_test(&self, key: XorName, hinter: PeerId) -> bool {
+        let now = Instant::now();
+        let entry = VerificationEntry {
+            state: VerificationState::PendingVerify,
+            verified_sources: Vec::new(),
+            tried_sources: HashSet::new(),
+            created_at: now,
+            next_verify_at: now,
+            hint_sources: HashSet::from([hinter]),
+            replica_hint_sources: HashSet::from([hinter]),
+        };
+        self.queues
+            .write()
+            .await
+            .add_pending_verify(key, entry)
+            .admitted()
+    }
+
+    /// Test-only: how far ahead `key`'s next verification round is scheduled,
+    /// or `None` when the key is not pending verification.
+    #[cfg(any(test, feature = "test-utils"))]
+    pub async fn pending_verify_delay_for_test(&self, key: &XorName) -> Option<Duration> {
+        self.queues.read().await.get_pending(key).map(|entry| {
+            entry
+                .next_verify_at
+                .saturating_duration_since(Instant::now())
+        })
+    }
+
     /// Start all background tasks.
     ///
     /// `dht_events` must be subscribed **before** `P2PNode::start()` so that
@@ -6546,6 +6582,99 @@ async fn handle_neighbor_sync_request(
     Ok(())
 }
 
+/// Test-only record of who *sent* a verification request covering a watched key.
+///
+/// The capacity gate's entire effect is a request that is never sent, and a
+/// request that was not sent leaves no production counter anywhere — not on the
+/// sender, whose traffic counters are process-global and so cannot separate one
+/// node of a single-process testnet from another, and not on the responder,
+/// which simply never hears from it. Recording who asked is what lets a test
+/// tell "held the key back" apart from "probed the close group, was refused at
+/// the dial, and then held the key back", which are otherwise identical from the
+/// queue's point of view.
+///
+/// Arming replaces the previous watch, so the outer map holds exactly the keys
+/// the current test named, and each inner map holds at most one entry per node
+/// in the process. Nothing is recorded until a test arms it, and the armed path
+/// costs one read lock on the sending side.
+///
+/// Counting is per key because a node that cannot write legitimately keeps
+/// sending verification requests for keys it has not yet authorized — that is
+/// `PaidForList` convergence, which the gate leaves alone — so a per-peer total
+/// would assert something untrue.
+///
+/// It counts *send attempts*, deliberately, recorded immediately before the wire
+/// call. A responder sheds requests at admission and as stale, so a
+/// receiver-side count would report zero for probes that were really sent, and
+/// an assertion that this node asked nobody would pass on the strength of the
+/// receiver dropping the question. It is not proof of delivery: a send that
+/// fails immediately, with no route or during shutdown, is still counted.
+///
+/// `OnceLock` rather than `LazyLock`, which needs a newer Rust than this crate's
+/// MSRV.
+#[cfg(any(test, feature = "test-utils"))]
+type VerificationWatch = HashMap<XorName, HashMap<PeerId, usize>>;
+
+#[cfg(any(test, feature = "test-utils"))]
+static VERIFICATION_WATCH: std::sync::OnceLock<std::sync::RwLock<VerificationWatch>> =
+    std::sync::OnceLock::new();
+
+#[cfg(any(test, feature = "test-utils"))]
+fn verification_watch() -> &'static std::sync::RwLock<VerificationWatch> {
+    VERIFICATION_WATCH.get_or_init(|| std::sync::RwLock::new(HashMap::new()))
+}
+
+/// Test-only: start counting verification requests for `keys`, from zero.
+///
+/// Replaces any previous watch rather than adding to it, so the map is bounded
+/// by the keys of the test that armed it last and nothing accumulates across a
+/// process running many tests.
+#[cfg(any(test, feature = "test-utils"))]
+pub fn watch_verification_requests_for_test(keys: &[XorName]) {
+    if let Ok(mut watch) = verification_watch().write() {
+        watch.clear();
+        for key in keys {
+            watch.insert(*key, HashMap::new());
+        }
+    }
+}
+
+/// Record that `requester` sent a verification request covering `keys`, for
+/// whichever of them are watched. A poisoned lock is ignored rather than
+/// propagated: this is observation for tests and must never change behaviour.
+#[cfg(any(test, feature = "test-utils"))]
+pub(crate) fn record_verification_request_sent(requester: &PeerId, keys: &[XorName]) {
+    // Fast path under a read lock: with nothing watched — every production
+    // build, and every test that did not ask — this is all the sender pays.
+    match verification_watch().read() {
+        Ok(watch) if watch.is_empty() => return,
+        Ok(watch) if !keys.iter().any(|key| watch.contains_key(key)) => return,
+        Ok(_) => {}
+        Err(_) => return,
+    }
+    if let Ok(mut watch) = verification_watch().write() {
+        for key in keys {
+            if let Some(by_peer) = watch.get_mut(key) {
+                *by_peer.entry(*requester).or_insert(0) += 1;
+            }
+        }
+    }
+}
+
+/// Test-only: how many times `requester` has asked this process about `key`.
+/// Zero unless the key was registered with `watch_verification_requests_for_test`.
+#[cfg(any(test, feature = "test-utils"))]
+#[must_use]
+pub fn verification_requests_for_key_from_for_test(requester: &PeerId, key: &XorName) -> usize {
+    verification_watch().read().map_or(0, |watch| {
+        watch
+            .get(key)
+            .and_then(|by_peer| by_peer.get(requester))
+            .copied()
+            .unwrap_or(0)
+    })
+}
+
 async fn handle_verification_request(
     source: &PeerId,
     request: &protocol::VerificationRequest,
@@ -7752,6 +7881,30 @@ async fn run_verification_cycle(ctx: VerificationCycleContext<'_>) {
     }
     let initial_pending_count = pending_keys.len();
 
+    // Capacity verdict for this cycle — the same pre-check the PUT handler and
+    // the fresh-offer path already run (V2-411), read once and reused below.
+    //
+    // It gates only the two steps that exist to enable a fetch: the presence
+    // probe that discovers holders, and the promotion that queues the download.
+    // Everything a cycle does that does not need a local write still runs on a
+    // full node — the local paid-list fast path, `PaidForList` convergence
+    // through the quorum round, and the terminal checks that retire a key this
+    // node already holds or is no longer admitted for. Gating earlier than this
+    // would stop a full node learning which of its keys were paid for, and
+    // would strand keys that should have retired, which keeps bootstrap drain
+    // pending and audits disabled until stale eviction.
+    // Only a *full* disk is a standing condition worth minutes of backoff. A
+    // failed `statvfs` says nothing about available space and may have cleared
+    // by the next cycle, so it is treated as writable here and left to the
+    // pre-check at the dial, which queries again and may well permit the write.
+    let write_blocked = storage.capacity_verdict() == CapacityVerdict::Full;
+    // Counted per gate rather than combined. `local_paid_probe` below counts
+    // probes actually sent, so it reads zero once the first gate fires; keeping
+    // the two deferral counts apart is what lets an operator see which branch a
+    // full node's keys are taking without adding a code path to find out.
+    let mut capacity_deferred_probe = 0usize;
+    let mut capacity_deferred_promote = 0usize;
+
     let self_id = *p2p_node.peer_id();
 
     // Step 1: Check local PaidForList for fast-path authorization (Section 9,
@@ -7802,6 +7955,21 @@ async fn run_verification_cycle(ctx: VerificationCycleContext<'_>) {
                 terminal_keys.push(key);
             }
         }
+
+        // Capacity gate, first of two. Everything above this point has already
+        // run: authorization succeeded via the local `PaidForList` hit, and the
+        // keys that should retire (already held, or no longer storage-admitted)
+        // have retired. What is left is a probe whose only purpose is finding a
+        // holder to download from, and this node cannot write what it would
+        // download.
+        if write_blocked && !local_paid_presence_probe_keys.is_empty() {
+            let mut q = queues.write().await;
+            for key in std::mem::take(&mut local_paid_presence_probe_keys) {
+                if q.defer_pending(&key, config::CAPACITY_BLOCKED_RETRY) {
+                    capacity_deferred_probe += 1;
+                }
+            }
+        }
     }
 
     let local_paid_probe_count = local_paid_presence_probe_keys.len();
@@ -8023,7 +8191,16 @@ async fn run_verification_cycle(ctx: VerificationCycleContext<'_>) {
                     let mut fetch_sources = sources;
                     add_replica_hint_sources(&mut fetch_sources, &replica_hint_sources);
                     let fetch_eligible = fetch_allowed_keys.contains(&key);
-                    if fetch_eligible && !fetch_sources.is_empty() {
+                    if fetch_eligible && write_blocked {
+                        // Capacity gate, second of two, and deliberately after
+                        // Step 4: the key is now recorded in `PaidForList` and
+                        // its verification stands. Only the download is held,
+                        // because `execute_single_fetch` would refuse it and
+                        // hand the key straight back here.
+                        if q.defer_pending(&key, config::CAPACITY_BLOCKED_RETRY) {
+                            capacity_deferred_promote += 1;
+                        }
+                    } else if fetch_eligible && !fetch_sources.is_empty() {
                         let distance =
                             crate::client::xor_distance(&key, p2p_node.peer_id().as_bytes());
                         // Atomic remove+enqueue: on fetch_queue capacity miss
@@ -8098,12 +8275,12 @@ async fn run_verification_cycle(ctx: VerificationCycleContext<'_>) {
     if elapsed_ms >= VERIFICATION_CYCLE_SLOW_LOG_MS {
         info!(
             target: "ant_node::replication::verification",
-            "Slow replication verification cycle: pending_start={initial_pending_count}, local_paid_probe={local_paid_probe_count}, network_verify={keys_needing_network_count}, terminal={terminal_key_count}, pending_after={pending_after}, fetch_after={fetch_after}, in_flight_after={in_flight_after}, elapsed_ms={elapsed_ms}",
+            "Slow replication verification cycle: pending_start={initial_pending_count}, capacity_deferred_probe={capacity_deferred_probe}, capacity_deferred_promote={capacity_deferred_promote}, local_paid_probe={local_paid_probe_count}, network_verify={keys_needing_network_count}, terminal={terminal_key_count}, pending_after={pending_after}, fetch_after={fetch_after}, in_flight_after={in_flight_after}, elapsed_ms={elapsed_ms}",
         );
     } else {
         debug!(
             target: "ant_node::replication::verification",
-            "Replication verification cycle: pending_start={initial_pending_count}, local_paid_probe={local_paid_probe_count}, network_verify={keys_needing_network_count}, terminal={terminal_key_count}, pending_after={pending_after}, fetch_after={fetch_after}, in_flight_after={in_flight_after}, elapsed_ms={elapsed_ms}",
+            "Replication verification cycle: pending_start={initial_pending_count}, capacity_deferred_probe={capacity_deferred_probe}, capacity_deferred_promote={capacity_deferred_promote}, local_paid_probe={local_paid_probe_count}, network_verify={keys_needing_network_count}, terminal={terminal_key_count}, pending_after={pending_after}, fetch_after={fetch_after}, in_flight_after={in_flight_after}, elapsed_ms={elapsed_ms}",
         );
     }
 }
@@ -8342,6 +8519,12 @@ fn apply_fetch_result(
         // stranded — it comes back once capacity allows — and the fallthrough
         // to `Terminal` when there is no retry metadata is what keeps
         // bootstrap drain accounting correct, exactly as for a source failure.
+        //
+        // The ordinary requeue delay, deliberately. A standing capacity block
+        // does not need a longer one here: the key returns to pending, and the
+        // gate at the head of the next cycle defers it for minutes. A separate
+        // backoff on this path would duplicate that to save one 15 s round, on
+        // a race the gate already makes rare.
         FetchResult::LocalWriteFailed => {
             if q.requeue_fetch_for_verification(key, verification_retry_after) {
                 FetchFollowUp::RequeuedForVerification

src/replication/mod.rs

@@ -600,6 +600,16 @@ fn spawn_verification_batch_task(
             }
         };
 
+        // Recorded at the wire call, after the permit and the encode, so the
+        // count is send attempts rather than scheduled batches — recording where
+        // a batch is merely queued would stay positive if the encode or the task
+        // itself regressed. It is not proof of delivery: a send that fails
+        // immediately is still counted. Recording on the responder would be
+        // worse, reading zero for requests the receiver shed at admission or as
+        // stale.
+        #[cfg(any(test, feature = "test-utils"))]
+        crate::replication::record_verification_request_sent(p2p.peer_id(), &requested_keys);
+
         let response = match p2p
             .send_request(&peer, REPLICATION_PROTOCOL_ID, encoded, timeout)
             .await