perf: reduce redundant cache path work

ARCHITECTURE.md

@@ -126,7 +126,9 @@ Delete allocates a sequence number, performs one bounded in-memory index removal
 
 When free capacity is low, a reclaim worker takes the oldest sealed Region, reads its used prefix sequentially, and walks the self-sized records. A mapping is changed only if the index still points at that exact Region address.
 
-The first L2 candidate access sets `seen`; a later access sets `hot`. Reclaim clears these volatile bits. Cold current mappings are removed. Hot current records may be rewritten through an existing append shard while preserving their logical sequence number. Payload validation produces a borrowed payload and checksum that staging reuses without scanning the payload again; the replacement header receives its own checksum. Reinsertion is capped at roughly one eighth of the reclaimed bytes and rotates successive source Regions across append shards. An empty Free queue disables reinsertion for that reclaim pass so foreground write capacity is restored first; staging or budget pressure drops the remaining candidates.
+The first L2 candidate access sets `seen`; a later access sets `hot`. Reclaim clears these volatile bits. Cold current mappings are removed. Hot current records may be rewritten through an existing append shard while preserving their logical sequence number. Payload validation produces a borrowed payload and checksum that staging reuses without scanning the payload again; the replacement header receives its own checksum.
+
+Reinsertion is capped at roughly one eighth of the reclaimed bytes and rotates successive source Regions across append shards. An empty Free queue disables reinsertion for that reclaim pass so foreground write capacity is restored first. Staging pressure stops further reinsertion from that source Region; the byte budget is checked per record. Once reinsertion stops, records within the byte budget skip payload checksums but retain header validation and address-conditional removal. Payloads exceeding the budget are still validated so the budget-skipped count includes only valid records.
 
 The source Region stays pinned until accepted replacement writes finish. Conditional index replacement lets a newer foreground put or delete win over a delayed reinsert. Only then is the Region returned to the free FIFO. Reclaim uses background I/O lanes and bounded single-pass work. `RuntimeOptions::reclaim_io_timeout` bounds each reclaim read from I/O admission through completion, defaulting to five seconds; foreground I/O deadlines are independent. Expiration enters reversible background I/O recovery. Long recovery can exhaust free capacity, but close interrupts recovery before draining workers.
 

CHANGELOG.md

@@ -13,7 +13,7 @@
 - A zero-capacity L1 bypasses shard locking and directory work for lookups, fills, promotions, and deletes.
 - Memory accounting skips redundant atomic updates when transient reads stay within the recorded peak or L1 transfers leave the charged byte total unchanged.
 - L2 index updates and cold-record reclamation avoid redundant atomic writes for already-clear heat bits and validated index pages.
-- Hot-record reclamation reuses the checksum computed during payload validation for its rewrite, avoiding a second payload scan while preserving record validation and the on-disk format.
+- Hot-record reclamation reuses validated payload checksums for rewrites and skips payload checksum scans for records within the byte budget after reinsertion stops, while preserving record header validation, conditional removal, budget statistics, and the on-disk format.
 - L1 replacement and eviction reuse detached, exclusively owned allocations of the same size under CLOCK and S3-FIFO, reducing repeated allocation during fixed-size turnover without an additional buffer pool. Caller-held values remain immutable and retain their separate memory charge. When activity statistics are enabled, mutations with no eviction also avoid a shared zero-value counter update.
 
 ## v0.6.0 (2026-09-21)

cache2/src/cache/runtime/mod.rs

@@ -1871,29 +1871,20 @@ fn reclaim_worker_result(
             } else {
                 None
             };
-            let mut accepting_reinserts = reinsert_operation.is_some();
-            let mut staged_reinsert = false;
-            let stats = state.regions.scan_reclaim(receipt, bytes, |record| {
-                if !accepting_reinserts {
-                    return Ok(false);
-                }
-                match state
-                    .regions
-                    .try_stage_reinsert(&state.staging, reinsert_shard, record)?
-                {
-                    RegionStageValue::Staged { .. } => {
-                        staged_reinsert = true;
-                        Ok(true)
-                    }
-                    RegionStageValue::Contended
-                    | RegionStageValue::NeedsFlush
-                    | RegionStageValue::NeedsRotation => {
-                        accepting_reinserts = false;
-                        Ok(false)
-                    }
-                }
-            })?;
-            if staged_reinsert {
+            let stats = state.regions.scan_reclaim(
+                receipt,
+                bytes,
+                reinsert_operation.is_some(),
+                |record| {
+                    Ok(matches!(
+                        state
+                            .regions
+                            .try_stage_reinsert(&state.staging, reinsert_shard, record,)?,
+                        RegionStageValue::Staged { .. }
+                    ))
+                },
+            )?;
+            if stats.reinsert_records != 0 {
                 let generation = state.append_controls[reinsert_shard].request_drain(false)?;
                 state.append_controls[reinsert_shard].wait_for_drain(generation)?;
             }

cache2/src/memory/mod.rs

@@ -729,7 +729,7 @@ impl MemoryShard {
         let Ok(memory_value) = MemoryValue::try_new(key, value, seqno, charge, recycled) else {
             return MemoryInsertResult::rejected(evictions);
         };
-        let previous_head = self.directory_head(hash);
+        let previous_head = self.directory_upsert(hash, packed_index);
         let entry = MemoryEntry {
             value: memory_value,
             hash_next: previous_head.unwrap_or(NO_SLOT_INDEX),
@@ -741,8 +741,6 @@ impl MemoryShard {
         self.resident_bytes = self.resident_bytes.saturating_add(charged_bytes);
         self.eviction
             .insert(&mut self.policy_slots, index, hash, charged_bytes);
-        let previous = self.directory_upsert(hash, packed_index);
-        debug_assert_eq!(previous, previous_head);
         MemoryInsertResult {
             slot: Some(packed_index),
             evictions,

cache2/src/region/index/storage/mod.rs

@@ -764,15 +764,14 @@ impl IndexStorage {
         // The exclusive partition guard keeps the state observed by the probe
         // stable, so updating its physical count needs no second slot decode.
         let global_slot = self.global_slot(slot)?;
-        let previous = IndexSlot::from_state(previous);
+        let previous_kind = match previous {
+            IndexSlotState::Empty => SlotPhysicalKind::Empty,
+            IndexSlotState::Value { .. } => SlotPhysicalKind::Value,
+        };
         let value = IndexSlot::from_state(state);
         let next_stats = self
             .physical_stats
-            .transitioned(
-                previous.physical_kind(),
-                value.physical_kind(),
-                self.range.slot_count,
-            )
+            .transitioned(previous_kind, value.physical_kind(), self.range.slot_count)
             .ok_or(IndexStorageError::InvalidPhysicalStats)?;
         self.core.write_slot(global_slot, value)?;
         self.physical_stats = next_stats;

cache2/src/region/mod.rs

@@ -461,12 +461,14 @@ impl RegionStore {
     }
 
     /// Scans one exact sealed prefix, removes cold mappings, and offers each
-    /// hot current record once to a bounded reinsertion sink. The
-    /// source Region remains exclusively pinned until [`Self::complete_reclaim`].
+    /// hot current record once to a bounded reinsertion sink. A false sink
+    /// result stops further offers from this source Region. The source stays
+    /// exclusively pinned until [`Self::complete_reclaim`].
     pub fn scan_reclaim(
         &self,
         receipt: RegionReclaimReceipt,
         bytes: &[u8],
+        mut accepting_reinserts: bool,
         mut try_reinsert: impl FnMut(RegionReinsertRecord<'_>) -> io::Result<bool>,
     ) -> io::Result<RegionReclaimStats> {
         if bytes.len() as u64 != receipt.used_offset {
@@ -563,33 +565,36 @@ impl RegionStore {
                                 "reclaim replacement record is too large",
                             )
                         })?;
-                    let payload = RecordPayload::validate(key, value, header.payload_crc);
                     let within_budget = u64::from(rewrite_bytes) <= reinsert_budget;
-                    let budget_exhausted = payload.is_some() && !within_budget;
-                    let reinserted = if let Some(payload) = payload
-                        && within_budget
+                    // Once the sink stops, in-budget records will only be removed.
+                    // Over-budget records still need validation because the
+                    // budget-skipped counter excludes corrupt payloads.
+                    let mut reinserted = false;
+                    if (accepting_reinserts || !within_budget)
+                        && let Some(payload) =
+                            RecordPayload::validate(key, value, header.payload_crc)
                     {
-                        try_reinsert(RegionReinsertRecord {
-                            hash: header.key_hash,
-                            previous_location: location,
-                            logical_seqno: header.seqno,
-                            record_bytes: rewrite_bytes,
-                            payload,
-                        })?
-                    } else {
-                        false
-                    };
+                        if within_budget {
+                            reinserted = try_reinsert(RegionReinsertRecord {
+                                hash: header.key_hash,
+                                previous_location: location,
+                                logical_seqno: header.seqno,
+                                record_bytes: rewrite_bytes,
+                                payload,
+                            })?;
+                            accepting_reinserts = reinserted;
+                        } else {
+                            stats.reinsert_budget_skipped =
+                                stats.reinsert_budget_skipped.saturating_add(1);
+                        }
+                    }
                     if reinserted {
                         reinsert_budget = reinsert_budget.saturating_sub(u64::from(rewrite_bytes));
                         stats.reinsert_records = stats.reinsert_records.saturating_add(1);
                         stats.reinsert_bytes = stats
                             .reinsert_bytes
                             .saturating_add(u64::from(rewrite_bytes));
                     } else {
-                        if budget_exhausted {
-                            stats.reinsert_budget_skipped =
-                                stats.reinsert_budget_skipped.saturating_add(1);
-                        }
                         if self
                             .index
                             .remove_if_match(header.key_hash, location)