perf: reuse validated payload checksums during reclaim

ARCHITECTURE.md

@@ -126,7 +126,7 @@ 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. 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 or budget pressure drops the remaining candidates.
 
 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

@@ -10,6 +10,7 @@
 
 ### Improvements
 
+- 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.
 - 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/region/mod.rs

@@ -246,8 +246,9 @@ pub struct RegionReinsertRecord<'a> {
     pub previous_location: PackedLocation,
     pub logical_seqno: u64,
     pub record_bytes: u32,
-    pub key: &'a [u8],
-    pub value: &'a [u8],
+    // Constructed only after payload validation; the same borrowed bytes and
+    // checksum travel together into staging without a second payload scan.
+    payload: RecordPayload<'a>,
 }
 
 /// Short shard-local transaction gate. `mutation` makes manager receipts and
@@ -562,20 +563,22 @@ impl RegionStore {
                                 "reclaim replacement record is too large",
                             )
                         })?;
-                    let payload_valid =
-                        crc32c(&[&bytes[header_end..payload_end]]) == header.payload_crc;
+                    let payload = RecordPayload::validate(key, value, header.payload_crc);
                     let within_budget = u64::from(rewrite_bytes) <= reinsert_budget;
-                    let budget_exhausted = payload_valid && !within_budget;
-                    let reinserted = payload_valid
+                    let budget_exhausted = payload.is_some() && !within_budget;
+                    let reinserted = if let Some(payload) = payload
                         && within_budget
-                        && try_reinsert(RegionReinsertRecord {
+                    {
+                        try_reinsert(RegionReinsertRecord {
                             hash: header.key_hash,
                             previous_location: location,
                             logical_seqno: header.seqno,
                             record_bytes: rewrite_bytes,
-                            key,
-                            value,
-                        })?;
+                            payload,
+                        })?
+                    } else {
+                        false
+                    };
                     if reinserted {
                         reinsert_budget = reinsert_budget.saturating_sub(u64::from(rewrite_bytes));
                         stats.reinsert_records = stats.reinsert_records.saturating_add(1);
@@ -860,7 +863,9 @@ impl RegionStore {
         key: &[u8],
         value: &[u8],
     ) -> io::Result<RegionStageValue> {
-        self.try_stage_record(staging, shard_id, hash, record_bytes, key, value, None)
+        self.validate_stage(staging, record_bytes)?;
+        let payload = RecordPayload::new(key, value);
+        self.try_stage_record(staging, shard_id, hash, record_bytes, &payload, None)
     }
 
     pub fn try_stage_reinsert(
@@ -869,36 +874,37 @@ impl RegionStore {
         shard_id: usize,
         record: RegionReinsertRecord<'_>,
     ) -> io::Result<RegionStageValue> {
+        self.validate_stage(staging, record.record_bytes)?;
         self.try_stage_record(
             staging,
             shard_id,
             record.hash,
             record.record_bytes,
-            record.key,
-            record.value,
+            &record.payload,
             Some((record.logical_seqno, record.previous_location)),
         )
     }
 
-    #[allow(clippy::too_many_arguments)]
+    fn validate_stage(&self, staging: &AppendStaging, record_bytes: u32) -> io::Result<()> {
+        self.health.require_healthy()?;
+        if record_bytes as usize > staging.chunk_bytes() {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "value exceeds one Region staging buffer",
+            ));
+        }
+        Ok(())
+    }
+
     fn try_stage_record(
         &self,
         staging: &AppendStaging,
         shard_id: usize,
         hash: u64,
         record_bytes: u32,
-        key: &[u8],
-        value: &[u8],
+        payload: &RecordPayload<'_>,
         reinsert: Option<(u64, PackedLocation)>,
     ) -> io::Result<RegionStageValue> {
-        self.health.require_healthy()?;
-        if record_bytes as usize > staging.chunk_bytes() {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidInput,
-                "value exceeds one Region staging buffer",
-            ));
-        }
-        let payload = RecordPayload::new(key, value);
         let _shard_mutation = match self.append_gate(shard_id)?.mutation.try_lock() {
             Ok(guard) => guard,
             Err(TryLockError::WouldBlock) => return Ok(RegionStageValue::Contended),
@@ -970,11 +976,11 @@ impl RegionStore {
                     receipt,
                     hash,
                     record_bytes,
-                    &payload,
+                    payload,
                     logical_seqno,
                 )?,
                 None => {
-                    encode_value_into_hashed(destination, receipt, hash, record_bytes, &payload)?
+                    encode_value_into_hashed(destination, receipt, hash, record_bytes, payload)?
                 }
             };
             Ok::<StagedRecord, RecordEncodeError>(match reinsert {

cache2/src/region/persistence/tests.rs

@@ -57,11 +57,14 @@ use crate::region::index::storage::IndexSlot;
 use crate::region::index::storage::IndexSlotState;
 use crate::region::index::storage::page_format::INDEX_IMAGE_SLOTS_PER_PAGE;
 use crate::region::manager::RegionManager;
+use crate::region::manager::RegionReclaimReceipt;
 use crate::region::reader::ReadCandidate;
 use crate::region::reader::ReadCompletion;
 use crate::region::reader::ReadDesc;
 use crate::region::reader::describe_read;
 use crate::region::record::RECORD_ALIGNMENT;
+use crate::region::record::RECORD_HEADER_SIZE;
+use crate::region::record::RecordHeader;
 use crate::region::record::codec::hash_key;
 use crate::region::record::codec::required_record_bytes;
 use crate::region::recovery::DATA_REGION_AREA_OFFSET;
@@ -998,6 +1001,140 @@ fn completed_owned_span_publishes_index_without_a_steady_state_sync() {
     engine.shutdown().unwrap();
 }
 
+#[test]
+fn reclaim_validates_payload_before_staging_and_counts_rewrites() {
+    let key = b"retained-reclaim";
+    for (value_len, corrupt_byte, reinserted, budget_skips) in [
+        (16 * 1024 + 7, None, true, 0),
+        (16 * 1024 + 7, Some(0), false, 0),
+        (16 * 1024 + 7, Some(key.len()), false, 0),
+        (64 * 1024, None, false, 1),
+        (64 * 1024, Some(key.len()), false, 0),
+    ] {
+        let (data, regions, staging) = foreground_stage_fixture();
+        let value = vec![0xa7; value_len];
+        let cold_key = b"cold-reclaim";
+        let cold_value = vec![0x31; 192 * 1024];
+        let cold_len = required_record_bytes(cold_key.len(), cold_value.len()).unwrap() as usize;
+        let rewrite_bytes = required_record_bytes(key.len(), value.len()).unwrap();
+        let used = (cold_len + rewrite_bytes as usize).next_multiple_of(RECOVERY_PAGE_SIZE);
+        let mut bytes = vec![0_u8; used];
+        // Model an already-read source prefix. Region selection and release are
+        // covered separately; this test exercises scanning through staged I/O.
+        let receipt = RegionReclaimReceipt {
+            region_id: 1,
+            created_seqno: 2,
+            used_offset: used as u64,
+            physical_record_count: 2,
+        };
+        for (offset, length, key, value, seqno) in [
+            (0, cold_len, &cold_key[..], &cold_value[..], 2),
+            (cold_len, used - cold_len, &key[..], &value[..], 3),
+        ] {
+            let payload = [key, value].concat();
+            let header = RecordHeader {
+                key_len: key.len() as u16,
+                value_len: value.len() as u32,
+                seqno,
+                key_hash: hash_key(data.hash_seed, key),
+                payload_crc: crc_fast::crc32_iscsi(&payload),
+                region_generation: receipt.created_seqno,
+                record_len: length as u32,
+            };
+            bytes[offset..offset + RECORD_HEADER_SIZE].copy_from_slice(&header.encode());
+            bytes[offset + RECORD_HEADER_SIZE..offset + RECORD_HEADER_SIZE + payload.len()]
+                .copy_from_slice(&payload);
+            regions
+                .index
+                .upsert(
+                    header.key_hash,
+                    IndexEntry {
+                        location: PackedLocation::new(
+                            receipt.region_id,
+                            offset as u32,
+                            length as u32,
+                        )
+                        .unwrap(),
+                    },
+                )
+                .unwrap();
+        }
+        let hash = hash_key(data.hash_seed, key);
+        assert!(regions.begin_point_read(hash).is_some());
+        assert!(regions.begin_point_read(hash).is_some());
+        let previous = regions.lookup_snapshot(hash).unwrap();
+        if let Some(corrupt_byte) = corrupt_byte {
+            bytes[cold_len + RECORD_HEADER_SIZE + corrupt_byte] ^= 1;
+        }
+        let stats = regions
+            .scan_reclaim(receipt, &bytes, |record| {
+                assert!(reinserted, "invalid or over-budget payload reached staging");
+                assert!(matches!(
+                    regions.try_stage_reinsert(&staging, 0, record)?,
+                    RegionStageValue::Staged { seqno, .. } if seqno > 3
+                ));
+                Ok(true)
+            })
+            .unwrap();
+        assert_eq!(stats.bytes_read, receipt.used_offset);
+        assert_eq!(stats.records_scanned, 2);
+        assert_eq!(stats.records_removed, 2 - u64::from(reinserted));
+        assert_eq!(stats.reinsert_records, u64::from(reinserted));
+        assert_eq!(
+            stats.reinsert_bytes,
+            u64::from(reinserted) * u64::from(rewrite_bytes)
+        );
+        assert_eq!(stats.reinsert_skipped, u64::from(!reinserted));
+        assert_eq!(stats.reinsert_budget_skipped, budget_skips);
+        if !reinserted {
+            assert_eq!(regions.lookup_snapshot(hash).unwrap(), None);
+            assert!(staging.shard_fill_snapshot(0).unwrap().is_none());
+            continue;
+        }
+        assert_eq!(regions.lookup_snapshot(hash).unwrap(), previous);
+        assert!(
+            used - cold_len > rewrite_bytes as usize,
+            "source has flush padding"
+        );
+        // Destroy the source before I/O: staging must own the validated bytes.
+        bytes.fill(0xcc);
+        drop(bytes);
+        let directory = TestDirectory::new();
+        let (file, _) = FaultFile::open(&directory.paths.data).unwrap();
+        file.set_len(data.geometry.data_file_len).unwrap();
+        let engine = IoEngine::for_test(Arc::new(file), 1).unwrap();
+        regions
+            .flush_staging_shard(
+                &staging,
+                &engine,
+                0,
+                crate::io::background::BackgroundIoAttempt::new(
+                    &crate::io::background::IoRecovery::new(Some(Duration::ZERO)),
+                    None,
+                ),
+            )
+            .unwrap()
+            .unwrap();
+        let current = regions.begin_point_read(hash).unwrap();
+        assert_ne!(current.region_generation, receipt.created_seqno);
+        let desc = describe_read(data.geometry, hash, current, false).unwrap();
+        let memory = data_path_memory();
+        let hit = regions
+            .read_value_from_desc(
+                &engine,
+                engine.try_reserve_read().unwrap(),
+                memory.try_read_buffer(desc.read_len).unwrap(),
+                desc,
+                key,
+            )
+            .unwrap()
+            .unwrap();
+        assert_eq!(hit.value(), value);
+        assert_eq!(hit.seqno(), 3);
+        engine.shutdown().unwrap();
+    }
+}
+
 #[test]
 fn read_availability_errors_do_not_latch_miss_only() {
     let data = data_path_superblock();

cache2/src/region/record/codec.rs

@@ -100,6 +100,13 @@ impl<'a> RecordPayload<'a> {
             crc: crc32c(&[key, value]),
         }
     }
+
+    /// Validates persisted payload bytes and retains their checksum for a rewrite.
+    /// The shared borrows keep the validated bytes immutable through encoding.
+    pub fn validate(key: &'a [u8], value: &'a [u8], expected_crc: u32) -> Option<Self> {
+        let payload = Self::new(key, value);
+        (payload.crc == expected_crc).then_some(payload)
+    }
 }
 
 /// Returns the minimum 32-byte-aligned envelope for one logical value.
@@ -303,6 +310,23 @@ mod tests {
         assert!(!expected.is_multiple_of(DIRECT_IO_ALIGNMENT));
     }
 
+    #[test]
+    fn payload_validation_checks_all_bytes_before_rewrite() {
+        for (key, value) in [
+            (&b""[..], &b""[..]),
+            (&b"key"[..], &b""[..]),
+            (&b""[..], &b"value"[..]),
+            (&b"key"[..], &b"value"[..]),
+        ] {
+            let expected_crc = crc_fast::crc32_iscsi(&[key, value].concat());
+            let payload = RecordPayload::validate(key, value, expected_crc).unwrap();
+            assert_eq!(payload.key, key);
+            assert_eq!(payload.value, value);
+            assert_eq!(payload.crc, expected_crc);
+            assert!(RecordPayload::validate(key, value, expected_crc ^ 1).is_none());
+        }
+    }
+
     #[test]
     fn reinsertion_preserves_logical_seqno_across_a_new_region_generation() {
         let key = b"key";
@@ -318,12 +342,14 @@ mod tests {
         };
         let mut destination = vec![0_u8; required as usize];
 
-        let payload = RecordPayload::new(key, value);
+        let expected_crc = crc_fast::crc32_iscsi(&[&key[..], &value[..]].concat());
+        let payload = RecordPayload::validate(key, value, expected_crc).unwrap();
         encode_reinsert_into_hashed(&mut destination, reservation, 7, required, &payload, 11)
             .unwrap();
         let header = RecordHeader::decode(&destination[..RECORD_HEADER_SIZE]).unwrap();
         assert_eq!(header.seqno, 11);
         assert_eq!(header.region_generation, 100);
+        assert_eq!(header.payload_crc, expected_crc);
     }
 
     #[test]