# VarTree/VarMap large-value cached read — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Eliminate the `extract_from_block` panic for values whose serialized entry fits `write_buffer_size` but spans more than two 4 KiB cache blocks, by routing such values through the encryption-aware locked-path fallback.

**Architecture:** Split each collection's `read_value_locked` into a `_result` variant returning `DbResult<ByteView>` plus a thin `Option<ByteView>` wrapper that preserves existing call-sites. Add an early size check in `read_value_cached_inner` (placed **before** the block-cache lookup) that delegates large values (`start + len > 8192`) to the `_result` helper under the shard lock. `extract_from_block` keeps its two-block signature.

**Tech Stack:** Rust 2024 edition, `armdb` crate, features `var-collections` (default) and `encryption` (gated test). Test runner: `cargo nextest`.

**Spec:** `armdb/docs/superpowers/specs/26-05-17-vartree-varmap-large-value-cached-read.md` (commit `a46d4c4`).

---

## File Map

| File | Responsibility | Change |
|------|----------------|--------|
| `armdb/src/var_tree.rs` | VarTree read path | Refactor `read_value_locked`, gate `read_value_cached_inner` on large values, add tests |
| `armdb/src/var_map.rs` | VarMap read path | Same refactor + gating + tests |

No new files. No new dependencies.

---

## Task 1: VarTree — refactor `read_value_locked` into `_result` + Option wrapper

**Files:**
- Modify: `armdb/src/var_tree.rs:1013-1052`
- Test: `armdb/src/var_tree.rs` (tests module starting at line 1410)

This task introduces the `read_value_locked_result` helper that propagates `StaleDiskLoc`, and rewrites the existing `read_value_locked` as a thin wrapper that preserves the `Option<ByteView>` contract (and the existing `tracing::error!` line) for all current call-sites: `var_tree.rs:230, 267, 920, 986, 1188`.

- [ ] **Step 1: Write the failing tests for `read_value_locked_result` and the wrapper**

Add at the end of the `mod tests` block in `armdb/src/var_tree.rs` (before the closing `}` of `mod tests`):

```rust
    /// `_result` returns Ok when the value sits in the active write buffer.
    #[test]
    fn read_value_locked_result_ok_from_write_buf() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 1u64.to_be_bytes();
        let payload = b"in-write-buffer-value";
        tree.put(&key, payload).expect("put");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = *node.load_disk();
        drop(guard);

        let shard = &tree.engine.shards()[disk.shard_id as usize];
        let inner = shard.lock();
        let v = tree
            .read_value_locked_result(&disk, &inner)
            .expect("write-buf read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// `_result` returns Ok when the value is on an immutable file and the
    /// entry sits within a single 4 KiB block.
    #[test]
    fn read_value_locked_result_ok_from_disk_immutable() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 2u64.to_be_bytes();
        let payload = b"single-block-immutable";
        tree.put(&key, payload).expect("put");
        // Force rotation so the entry moves out of the active write buffer.
        for i in 0..6u8 {
            tree.put(&i.to_be_bytes(), &[i; 256]).expect("rotator");
        }

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = *node.load_disk();
        drop(guard);

        // `open_test_tree` uses the default CacheConfig (max_size = 0), so the
        // cache is disabled and step 2 (single-block cache lookup) returns
        // None unconditionally — the read naturally exercises step 3 (disk).

        let shard = &tree.engine.shards()[disk.shard_id as usize];
        let inner = shard.lock();
        let v = tree
            .read_value_locked_result(&disk, &inner)
            .expect("disk read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// `_result` propagates `StaleDiskLoc` from the disk read.
    #[test]
    fn read_value_locked_result_propagates_stale_disk_loc() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 3u64.to_be_bytes();
        let snap = put_until_compactable(&tree, key);

        let shard = &tree.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let inner = shard.lock();
        match tree.read_value_locked_result(&snap, &inner) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    /// Backward-compat: the `Option` wrapper still returns `None` on
    /// `StaleDiskLoc`, so callers like `cas`/`update` keep their contract.
    /// We intentionally do NOT assert on the tracing output (no capture
    /// helper exists in this crate); the `None` is the load-bearing
    /// invariant.
    #[test]
    fn read_value_locked_returns_none_on_stale() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 4u64.to_be_bytes();
        let snap = put_until_compactable(&tree, key);

        let shard = &tree.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let inner = shard.lock();
        assert!(tree.read_value_locked(&snap, &inner).is_none());
    }
```

- [ ] **Step 2: Run the new tests and verify they fail to compile**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(read_value_locked_result) | test(read_value_locked_returns_none_on_stale)'
```

Expected: build failure with `no method named 'read_value_locked_result' found for ... 'VarTree'`.

- [ ] **Step 3: Add `read_value_locked_result` and convert `read_value_locked` to wrapper**

In `armdb/src/var_tree.rs`, replace the entire existing block at `var_tree.rs:1013-1052`:

```rust
    /// Read value when shard lock is already held. Used by CAS/update.
    fn read_value_locked(&self, disk: &DiskLoc, inner: &ShardInner) -> Option<ByteView> {
        let len = disk.len as usize;

        // 1. Write buffer (for unflushed data)
        if inner.active.file_id == disk.file_id as u32
            && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
        {
            return Some(ByteView::new(bytes));
        }

        // 2. Block cache (lock-free)
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        if start + len <= 4096 {
            let cache_key = BlockKey {
                shard_id: disk.shard_id,
                file_id: disk.file_id as u32,
                block_offset,
            };
            if let Some(block) = self.cache.get(&cache_key) {
                return Some(ByteView::new(&block[start..start + len]));
            }
        }

        // 3. Disk read via encryption-aware helper (no re-lock).
        let bytes = match inner.read_value_from_disk_locked(disk) {
            Ok(b) => b,
            Err(DbError::StaleDiskLoc) => {
                tracing::error!(
                    file_id = disk.file_id,
                    shard_id = disk.shard_id,
                    "stale DiskLoc under shard lock - programming bug",
                );
                return None;
            }
            Err(_) => return None,
        };
        Some(ByteView::new(&bytes))
    }
```

with:

```rust
    /// Read value when shard lock is already held, propagating `DbError`.
    /// Used by both `read_value_locked` (Option wrapper) and the large-value
    /// fallback in `read_value_cached_inner`.
    fn read_value_locked_result(
        &self,
        disk: &DiskLoc,
        inner: &ShardInner,
    ) -> DbResult<ByteView> {
        let len = disk.len as usize;

        // 1. Write buffer (for unflushed data)
        if inner.active.file_id == disk.file_id as u32
            && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
        {
            return Ok(ByteView::new(bytes));
        }

        // 2. Block cache (lock-free, single-block fast path)
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        if start + len <= 4096 {
            let cache_key = BlockKey {
                shard_id: disk.shard_id,
                file_id: disk.file_id as u32,
                block_offset,
            };
            if let Some(block) = self.cache.get(&cache_key) {
                return Ok(ByteView::new(&block[start..start + len]));
            }
        }

        // 3. Disk read via encryption-aware helper (no re-lock).
        let bytes = inner.read_value_from_disk_locked(disk)?;
        Ok(ByteView::new(&bytes))
    }

    /// Read value when shard lock is already held. Used by CAS/update.
    ///
    /// Thin `Option` wrapper around `read_value_locked_result` that preserves
    /// the existing call-site contract: `StaleDiskLoc` is logged at error
    /// level and collapsed to `None`; other errors collapse to `None` silently.
    fn read_value_locked(&self, disk: &DiskLoc, inner: &ShardInner) -> Option<ByteView> {
        match self.read_value_locked_result(disk, inner) {
            Ok(v) => Some(v),
            Err(DbError::StaleDiskLoc) => {
                tracing::error!(
                    file_id = disk.file_id,
                    shard_id = disk.shard_id,
                    "stale DiskLoc under shard lock - programming bug",
                );
                None
            }
            Err(_) => None,
        }
    }
```

- [ ] **Step 4: Run the new tests and verify they pass**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(read_value_locked_result) | test(read_value_locked_returns_none_on_stale)'
```

Expected: 4 tests passed.

- [ ] **Step 5: Run the full VarTree test suite to confirm no regression**

Run:

```sh
cargo nextest run -p armdb --features var-collections -E 'test(var_tree)'
```

Expected: all VarTree tests pass.

- [ ] **Step 6: Commit**

```sh
git add armdb/src/var_tree.rs
git commit -m "$(cat <<'EOF'
refactor(armdb/var_tree): split read_value_locked into _result + wrapper

Introduces read_value_locked_result returning DbResult<ByteView> so the
lock-free cached path can retry on StaleDiskLoc. The existing
read_value_locked is kept as a thin Option wrapper that preserves the
contract for cas/update/put/delete/get call-sites.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 2: VarMap — refactor `read_value_locked` into `_result` + Option wrapper

**Files:**
- Modify: `armdb/src/var_map.rs:649-688`
- Test: `armdb/src/var_map.rs` (tests module starting at line 821)

Structurally identical to Task 1. Call-sites that keep the `Option` contract: `var_map.rs:203, 239, 270, 326, 802`.

- [ ] **Step 1: Write the failing tests**

Add at the end of the `mod tests` block in `armdb/src/var_map.rs`:

```rust
    /// `_result` returns Ok when the value sits in the active write buffer.
    #[test]
    fn read_value_locked_result_ok_from_write_buf() {
        let dir = tempdir().unwrap();
        let map = open_test_map(dir.path());

        let key = 1u64.to_be_bytes();
        let payload = b"in-write-buffer-value";
        map.put(&key, payload).expect("put");

        let disk = *sync::lock(&map.indexes[0]).get(&key).expect("indexed");
        let shard = &map.engine.shards()[disk.shard_id as usize];
        let inner = shard.lock();
        let v = map
            .read_value_locked_result(&disk, &inner)
            .expect("write-buf read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// `_result` returns Ok when the value is on an immutable file and the
    /// entry sits within a single 4 KiB block.
    #[test]
    fn read_value_locked_result_ok_from_disk_immutable() {
        let dir = tempdir().unwrap();
        let map = open_test_map(dir.path());

        let key = 2u64.to_be_bytes();
        let payload = b"single-block-immutable";
        map.put(&key, payload).expect("put");
        for i in 0..6u8 {
            map.put(&i.to_be_bytes(), &[i; 256]).expect("rotator");
        }

        let disk = *sync::lock(&map.indexes[0]).get(&key).expect("indexed");
        // `open_test_map` uses default CacheConfig (max_size = 0) — cache is
        // disabled, so step 2 returns None and the read exercises step 3 (disk).

        let shard = &map.engine.shards()[disk.shard_id as usize];
        let inner = shard.lock();
        let v = map
            .read_value_locked_result(&disk, &inner)
            .expect("disk read must succeed");
        assert_eq!(v.as_bytes(), payload);
    }

    /// `_result` propagates `StaleDiskLoc` from the disk read.
    #[test]
    fn read_value_locked_result_propagates_stale_disk_loc() {
        let dir = tempdir().unwrap();
        let map = open_test_map(dir.path());

        let key = 3u64.to_be_bytes();
        let snap = put_until_compactable(&map, key);

        let shard = &map.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &map, 0.0).expect("compaction");

        let inner = shard.lock();
        match map.read_value_locked_result(&snap, &inner) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    /// Backward-compat: Option wrapper still returns None on StaleDiskLoc.
    /// No tracing assertion: the None is the load-bearing invariant.
    #[test]
    fn read_value_locked_returns_none_on_stale() {
        let dir = tempdir().unwrap();
        let map = open_test_map(dir.path());

        let key = 4u64.to_be_bytes();
        let snap = put_until_compactable(&map, key);

        let shard = &map.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &map, 0.0).expect("compaction");

        let inner = shard.lock();
        assert!(map.read_value_locked(&snap, &inner).is_none());
    }
```

- [ ] **Step 2: Run the new tests and verify they fail to compile**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(read_value_locked_result) | test(read_value_locked_returns_none_on_stale)'
```

Expected: build failure with `no method named 'read_value_locked_result' found for ... 'VarMap'`.

- [ ] **Step 3: Add `read_value_locked_result` and convert `read_value_locked` to wrapper**

In `armdb/src/var_map.rs`, replace the entire existing block at `var_map.rs:649-688`:

```rust
    /// Read value when shard lock is already held. Used by CAS/update.
    fn read_value_locked(&self, disk: &DiskLoc, inner: &ShardInner) -> Option<ByteView> {
        let len = disk.len as usize;

        // 1. Write buffer (for unflushed data)
        if inner.active.file_id == disk.file_id as u32
            && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
        {
            return Some(ByteView::new(bytes));
        }

        // 2. Block cache (lock-free)
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        if start + len <= 4096 {
            let cache_key = BlockKey {
                shard_id: disk.shard_id,
                file_id: disk.file_id as u32,
                block_offset,
            };
            if let Some(block) = self.cache.get(&cache_key) {
                return Some(ByteView::new(&block[start..start + len]));
            }
        }

        // 3. Disk read via encryption-aware helper (no re-lock).
        let bytes = match inner.read_value_from_disk_locked(disk) {
            Ok(b) => b,
            Err(DbError::StaleDiskLoc) => {
                tracing::error!(
                    file_id = disk.file_id,
                    shard_id = disk.shard_id,
                    "stale DiskLoc under shard lock - programming bug",
                );
                return None;
            }
            Err(_) => return None,
        };
        Some(ByteView::new(&bytes))
    }
```

with:

```rust
    /// Read value when shard lock is already held, propagating `DbError`.
    /// Used by both `read_value_locked` (Option wrapper) and the large-value
    /// fallback in `read_value_cached_inner`.
    fn read_value_locked_result(
        &self,
        disk: &DiskLoc,
        inner: &ShardInner,
    ) -> DbResult<ByteView> {
        let len = disk.len as usize;

        // 1. Write buffer (for unflushed data)
        if inner.active.file_id == disk.file_id as u32
            && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
        {
            return Ok(ByteView::new(bytes));
        }

        // 2. Block cache (lock-free, single-block fast path)
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        if start + len <= 4096 {
            let cache_key = BlockKey {
                shard_id: disk.shard_id,
                file_id: disk.file_id as u32,
                block_offset,
            };
            if let Some(block) = self.cache.get(&cache_key) {
                return Ok(ByteView::new(&block[start..start + len]));
            }
        }

        // 3. Disk read via encryption-aware helper (no re-lock).
        let bytes = inner.read_value_from_disk_locked(disk)?;
        Ok(ByteView::new(&bytes))
    }

    /// Read value when shard lock is already held. Used by CAS/update.
    ///
    /// Thin `Option` wrapper around `read_value_locked_result` that preserves
    /// the existing call-site contract: `StaleDiskLoc` is logged at error
    /// level and collapsed to `None`; other errors collapse to `None` silently.
    fn read_value_locked(&self, disk: &DiskLoc, inner: &ShardInner) -> Option<ByteView> {
        match self.read_value_locked_result(disk, inner) {
            Ok(v) => Some(v),
            Err(DbError::StaleDiskLoc) => {
                tracing::error!(
                    file_id = disk.file_id,
                    shard_id = disk.shard_id,
                    "stale DiskLoc under shard lock - programming bug",
                );
                None
            }
            Err(_) => None,
        }
    }
```

- [ ] **Step 4: Run the new tests and verify they pass**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(read_value_locked_result) | test(read_value_locked_returns_none_on_stale)'
```

Expected: 8 tests passed (4 VarTree from Task 1 + 4 VarMap).

- [ ] **Step 5: Run the full VarMap test suite to confirm no regression**

Run:

```sh
cargo nextest run -p armdb --features var-collections -E 'test(var_map)'
```

Expected: all VarMap tests pass.

- [ ] **Step 6: Commit**

```sh
git add armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
refactor(armdb/var_map): split read_value_locked into _result + wrapper

Mirror of the var_tree refactor: introduces read_value_locked_result
returning DbResult<ByteView> so the lock-free cached path can retry on
StaleDiskLoc. The existing read_value_locked is kept as a thin Option
wrapper that preserves the contract for cas/update/put/delete/get
call-sites.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 3: VarTree — route large values through locked fallback before cache lookup

**Files:**
- Modify: `armdb/src/var_tree.rs:784-819`
- Test: `armdb/src/var_tree.rs` (tests module)

Add the early `if start + len > 8192` branch to `read_value_cached_inner` **before** any cache lookup. The branch must take the shard lock and call `read_value_locked_result`; on `StaleDiskLoc` the error propagates so the outer retry loop (`read_value_cached`) can take a fresh `*node.load_disk()` snapshot under its `LocalGuard`.

- [ ] **Step 1: Write the failing test for first-block-cached large value**

Append to the `mod tests` block in `armdb/src/var_tree.rs`:

```rust
    /// Pins the "size check sits before cache lookup" invariant.
    ///
    /// Manually inject the first 4 KiB block of a large value into the
    /// cache. Without the early `start + len > 8192` check, the cached path
    /// would call `extract_from_block`, which only supports two blocks and
    /// would panic on `next[..second_len]` for `second_len > 4096`.
    ///
    /// With the early check, the call routes to the locked path and returns
    /// the correct bytes.
    #[test]
    fn large_value_with_first_block_cached_uses_fallback() {
        let dir = tempdir().unwrap();
        // Custom geometry: write_buffer_size big enough for 20 KiB entry,
        // small max_file_size so a follow-up write rotates it to immutable.
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        // Enable the block cache so `cache.insert` actually takes effect.
        // Default `CacheConfig::default` has `max_size = 0` which makes the
        // cache a no-op; we need real inserts to set up the warm-cache scenario.
        cfg.cache.max_size = 1 << 20;
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open");

        let key = 42u64.to_be_bytes();
        // Build a value whose body spans ~5 blocks: > 8192 bytes.
        let payload: Vec<u8> = (0..20_000u32).map(|i| i as u8).collect();
        tree.put(&key, &payload).expect("put large");
        // Force rotation so the large entry moves to an immutable file.
        tree.put(&0u64.to_be_bytes(), b"trigger-rotate")
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let disk = *node.load_disk();
        drop(guard);

        // Sanity: this is a multi-block large value.
        let start = (disk.offset & 4095) as usize;
        assert!(
            start + disk.len as usize > 8192,
            "test precondition: large value must span >2 blocks",
        );

        // Inject the first block (zeroes) into the cache to ensure the cache
        // lookup would succeed for the first block — exactly the warm-cache
        // scenario that would have triggered the old panic.
        let block_offset = disk.offset as u64 & !4095;
        let cache_key = BlockKey {
            shard_id: disk.shard_id,
            file_id: disk.file_id as u32,
            block_offset,
        };
        tree.cache
            .insert(cache_key, Arc::new(AlignedBuf::zeroed(4096)));
        // Sanity: insert took effect.
        assert!(
            tree.cache.get(&cache_key).is_some(),
            "cache must be enabled for warm-cache scenario",
        );

        let v = tree
            .read_value_cached_inner(&disk)
            .expect("large value must read via locked fallback");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }
```

- [ ] **Step 2: Run the test and verify it panics**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(large_value_with_first_block_cached_uses_fallback)'
```

Expected: panic at `extract_from_block` (`next[..second_len]` slice out of bounds) or test failure with mismatched bytes — either way it must not return the correct payload.

- [ ] **Step 3: Add the early large-value routing in `read_value_cached_inner`**

In `armdb/src/var_tree.rs`, replace the body of `read_value_cached_inner` at `var_tree.rs:784-819`:

```rust
    fn read_value_cached_inner(&self, disk: &DiskLoc) -> DbResult<ByteView> {
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        let len = disk.len as usize;
        let cache_key = BlockKey {
            shard_id: disk.shard_id,
            file_id: disk.file_id as u32,
            block_offset,
        };

        // 1. Block Cache (lock-free)
        if let Some(block) = self.cache.get(&cache_key) {
            metrics::counter!("armdb.cache.hit").increment(1);
            return Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
            });
        }

        // 2. Write buffer — for unflushed data in active file (brief shard lock)
        {
            let shard = &self.engine.shards()[disk.shard_id as usize];
            let inner = shard.lock();
            if inner.active.file_id == disk.file_id as u32
                && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
            {
                return Ok(ByteView::new(bytes));
            }
        }

        // 3. Disk read + cache
        metrics::counter!("armdb.cache.miss").increment(1);
        let block = self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset)?;
        Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
        })
    }
```

with:

```rust
    fn read_value_cached_inner(&self, disk: &DiskLoc) -> DbResult<ByteView> {
        let len = disk.len as usize;
        let start = (disk.offset & 4095) as usize;

        // Large values (>2 blocks) bypass the cached path. This check MUST
        // sit before the cache lookup: a warm first block would otherwise
        // route a large value into `extract_from_block`, which only supports
        // two blocks and would panic on `next[..second_len]` when
        // `second_len > 4096`. StaleDiskLoc propagates so the outer retry
        // loop (`read_value_cached`) takes a fresh DiskLoc snapshot.
        if start + len > 8192 {
            let shard = &self.engine.shards()[disk.shard_id as usize];
            let inner = shard.lock();
            return self.read_value_locked_result(disk, &inner);
        }

        let block_offset = disk.offset as u64 & !4095;
        let cache_key = BlockKey {
            shard_id: disk.shard_id,
            file_id: disk.file_id as u32,
            block_offset,
        };

        // 1. Block Cache (lock-free)
        if let Some(block) = self.cache.get(&cache_key) {
            metrics::counter!("armdb.cache.hit").increment(1);
            return Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
            });
        }

        // 2. Write buffer — for unflushed data in active file (brief shard lock)
        {
            let shard = &self.engine.shards()[disk.shard_id as usize];
            let inner = shard.lock();
            if inner.active.file_id == disk.file_id as u32
                && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
            {
                return Ok(ByteView::new(bytes));
            }
        }

        // 3. Disk read + cache
        metrics::counter!("armdb.cache.miss").increment(1);
        let block = self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset)?;
        Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
        })
    }
```

- [ ] **Step 4: Run the test and verify it passes**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(large_value_with_first_block_cached_uses_fallback)'
```

Expected: 1 test passed.

- [ ] **Step 5: Run the full VarTree test suite**

Run:

```sh
cargo nextest run -p armdb --features var-collections -E 'test(var_tree)'
```

Expected: all VarTree tests pass.

- [ ] **Step 6: Commit**

```sh
git add armdb/src/var_tree.rs
git commit -m "$(cat <<'EOF'
fix(armdb/var_tree): route large values to locked fallback before cache lookup

read_value_cached_inner now checks (start + len > 8192) before any cache
lookup. Values whose body spans more than two 4 KiB blocks take the
shard lock and call read_value_locked_result. Previously a warm first
block would route a large value into extract_from_block (two-block
specialization), which panicked on next[..second_len] when
second_len > 4096.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 4: VarMap — route large values through locked fallback before cache lookup

**Files:**
- Modify: `armdb/src/var_map.rs:573-605`
- Test: `armdb/src/var_map.rs` (tests module)

Mirror of Task 3.

- [ ] **Step 1: Write the failing test for first-block-cached large value**

Append to the `mod tests` block in `armdb/src/var_map.rs`:

```rust
    /// Pins the "size check sits before cache lookup" invariant for VarMap.
    /// Without the early `start + len > 8192` check, the cached path would
    /// hit `extract_from_block` and panic on `next[..second_len]`.
    #[test]
    fn large_value_with_first_block_cached_uses_fallback() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        // Enable the block cache so `cache.insert` actually takes effect
        // (default CacheConfig has max_size = 0, making the cache a no-op).
        cfg.cache.max_size = 1 << 20;
        let map: VarMap<[u8; 8]> = VarMap::open(dir.path(), cfg).expect("open");

        let key = 42u64.to_be_bytes();
        let payload: Vec<u8> = (0..20_000u32).map(|i| i as u8).collect();
        map.put(&key, &payload).expect("put large");
        map.put(&0u64.to_be_bytes(), b"trigger-rotate")
            .expect("rotate");

        let disk = *sync::lock(&map.indexes[0]).get(&key).expect("indexed");
        let start = (disk.offset & 4095) as usize;
        assert!(
            start + disk.len as usize > 8192,
            "test precondition: large value must span >2 blocks",
        );

        let block_offset = disk.offset as u64 & !4095;
        let cache_key = BlockKey {
            shard_id: disk.shard_id,
            file_id: disk.file_id as u32,
            block_offset,
        };
        map.cache
            .insert(cache_key, Arc::new(AlignedBuf::zeroed(4096)));
        assert!(
            map.cache.get(&cache_key).is_some(),
            "cache must be enabled for warm-cache scenario",
        );

        let v = map
            .read_value_cached_inner(&disk)
            .expect("large value must read via locked fallback");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }
```

- [ ] **Step 2: Run the test and verify it panics or returns wrong bytes**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(large_value_with_first_block_cached_uses_fallback)'
```

Expected: panic at `extract_from_block` for the VarMap test.

- [ ] **Step 3: Add the early large-value routing in `read_value_cached_inner`**

In `armdb/src/var_map.rs`, replace the body of `read_value_cached_inner` at `var_map.rs:573-605`:

```rust
    fn read_value_cached_inner(&self, disk: &DiskLoc) -> DbResult<ByteView> {
        let block_offset = disk.offset as u64 & !4095;
        let start = (disk.offset & 4095) as usize;
        let len = disk.len as usize;
        let cache_key = BlockKey {
            shard_id: disk.shard_id,
            file_id: disk.file_id as u32,
            block_offset,
        };

        if let Some(block) = self.cache.get(&cache_key) {
            metrics::counter!("armdb.cache.hit").increment(1);
            return Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
            });
        }

        {
            let shard = &self.engine.shards()[disk.shard_id as usize];
            let inner = shard.lock();
            if inner.active.file_id == disk.file_id as u32
                && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
            {
                return Ok(ByteView::new(bytes));
            }
        }

        metrics::counter!("armdb.cache.miss").increment(1);
        let block = self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset)?;
        Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
        })
    }
```

with:

```rust
    fn read_value_cached_inner(&self, disk: &DiskLoc) -> DbResult<ByteView> {
        let len = disk.len as usize;
        let start = (disk.offset & 4095) as usize;

        // Large values (>2 blocks) bypass the cached path. This check MUST
        // sit before the cache lookup: a warm first block would otherwise
        // route a large value into `extract_from_block`, which only supports
        // two blocks and would panic on `next[..second_len]` when
        // `second_len > 4096`. StaleDiskLoc propagates so VarMap::get's
        // per-shard-index retry loop takes a fresh DiskLoc snapshot.
        if start + len > 8192 {
            let shard = &self.engine.shards()[disk.shard_id as usize];
            let inner = shard.lock();
            return self.read_value_locked_result(disk, &inner);
        }

        let block_offset = disk.offset as u64 & !4095;
        let cache_key = BlockKey {
            shard_id: disk.shard_id,
            file_id: disk.file_id as u32,
            block_offset,
        };

        if let Some(block) = self.cache.get(&cache_key) {
            metrics::counter!("armdb.cache.hit").increment(1);
            return Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
            });
        }

        {
            let shard = &self.engine.shards()[disk.shard_id as usize];
            let inner = shard.lock();
            if inner.active.file_id == disk.file_id as u32
                && let Some(bytes) = inner.write_buf.read(disk.offset as u64, len)
            {
                return Ok(ByteView::new(bytes));
            }
        }

        metrics::counter!("armdb.cache.miss").increment(1);
        let block = self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset)?;
        Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
        })
    }
```

- [ ] **Step 4: Run the test and verify it passes**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(large_value_with_first_block_cached_uses_fallback)'
```

Expected: 2 tests passed (VarTree + VarMap).

- [ ] **Step 5: Run the full VarMap test suite**

Run:

```sh
cargo nextest run -p armdb --features var-collections -E 'test(var_map)'
```

Expected: all VarMap tests pass.

- [ ] **Step 6: Commit**

```sh
git add armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
fix(armdb/var_map): route large values to locked fallback before cache lookup

Mirror of the var_tree fix. read_value_cached_inner now checks
(start + len > 8192) before any cache lookup. StaleDiskLoc propagates so
VarMap::get's per-shard-index retry loop takes a fresh DiskLoc snapshot
when compaction races with the read.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 5: VarTree — pin `extract_from_block` boundaries with unit tests

**Files:**
- Test only: `armdb/src/var_tree.rs` (tests module)

No production code change. These tests pin the contract callers now rely on after Task 3: `extract_from_block` is only invoked for `start + len <= 8192`. Existing `extract_from_block_propagates_next_block_error` and `extract_from_block_multi_block_first_cached_second_stale` stay untouched.

- [ ] **Step 1: Write the three boundary tests**

Append to the `mod tests` block in `armdb/src/var_tree.rs`:

```rust
    /// Value entirely in one block: `start + len <= 4096`.
    #[test]
    fn extract_from_block_single_block() {
        let mut block = AlignedBuf::zeroed(4096);
        for (i, byte) in block.iter_mut().enumerate() {
            *byte = i as u8;
        }
        let v = VarTree::<[u8; 8]>::extract_from_block(&block, 100, 50, || {
            panic!("next_block must not be called for single-block reads")
        })
        .expect("ok");
        let expected: Vec<u8> = (100u8..150u8).map(|i| i).collect();
        assert_eq!(v.as_bytes(), expected.as_slice());
    }

    /// Value ends exactly at the second block's end: `start + len == 8192`.
    #[test]
    fn extract_from_block_two_blocks_exact() {
        let mut first = AlignedBuf::zeroed(4096);
        for byte in first.iter_mut() {
            *byte = 0xAA;
        }
        let mut second = AlignedBuf::zeroed(4096);
        for byte in second.iter_mut() {
            *byte = 0xBB;
        }
        // start = 4096 - 1 (last byte of first block); len = 4097.
        // start + len = 8192 — the boundary that must remain supported.
        let v = VarTree::<[u8; 8]>::extract_from_block(&first, 4095, 4097, || {
            Ok(Arc::new(second))
        })
        .expect("ok");
        let bytes = v.as_bytes();
        assert_eq!(bytes.len(), 4097);
        assert_eq!(bytes[0], 0xAA);
        assert_eq!(bytes[1], 0xBB);
        assert_eq!(bytes[4096], 0xBB);
    }

    /// Value spans two blocks partially: `4096 < start + len < 8192`.
    #[test]
    fn extract_from_block_two_blocks_partial() {
        let mut first = AlignedBuf::zeroed(4096);
        for byte in first.iter_mut() {
            *byte = 0x11;
        }
        let mut second = AlignedBuf::zeroed(4096);
        for byte in second.iter_mut() {
            *byte = 0x22;
        }
        // start = 4000, len = 200 -> first 96 bytes from first, next 104 from second.
        let v = VarTree::<[u8; 8]>::extract_from_block(&first, 4000, 200, || {
            Ok(Arc::new(second))
        })
        .expect("ok");
        let bytes = v.as_bytes();
        assert_eq!(bytes.len(), 200);
        assert!(bytes[..96].iter().all(|b| *b == 0x11));
        assert!(bytes[96..].iter().all(|b| *b == 0x22));
    }
```

- [ ] **Step 2: Run the boundary tests**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(extract_from_block_single_block) | test(extract_from_block_two_blocks_exact) | test(extract_from_block_two_blocks_partial)'
```

Expected: 3 tests passed.

- [ ] **Step 3: Commit**

```sh
git add armdb/src/var_tree.rs
git commit -m "$(cat <<'EOF'
test(armdb/var_tree): pin extract_from_block boundary cases

Adds single_block, two_blocks_exact (start+len == 8192), and
two_blocks_partial unit tests. These pin the contract that callers now
rely on after read_value_cached_inner filters out start+len > 8192:
extract_from_block remains a two-block specialization.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 6: VarMap — pin `extract_from_block` boundaries with unit tests

**Files:**
- Test only: `armdb/src/var_map.rs` (tests module)

Mirror of Task 5.

- [ ] **Step 1: Write the three boundary tests**

Append to the `mod tests` block in `armdb/src/var_map.rs`:

```rust
    #[test]
    fn extract_from_block_single_block() {
        let mut block = AlignedBuf::zeroed(4096);
        for (i, byte) in block.iter_mut().enumerate() {
            *byte = i as u8;
        }
        let v = VarMap::<[u8; 8]>::extract_from_block(&block, 100, 50, || {
            panic!("next_block must not be called for single-block reads")
        })
        .expect("ok");
        let expected: Vec<u8> = (100u8..150u8).map(|i| i).collect();
        assert_eq!(v.as_bytes(), expected.as_slice());
    }

    #[test]
    fn extract_from_block_two_blocks_exact() {
        let mut first = AlignedBuf::zeroed(4096);
        for byte in first.iter_mut() {
            *byte = 0xAA;
        }
        let mut second = AlignedBuf::zeroed(4096);
        for byte in second.iter_mut() {
            *byte = 0xBB;
        }
        let v = VarMap::<[u8; 8]>::extract_from_block(&first, 4095, 4097, || {
            Ok(Arc::new(second))
        })
        .expect("ok");
        let bytes = v.as_bytes();
        assert_eq!(bytes.len(), 4097);
        assert_eq!(bytes[0], 0xAA);
        assert_eq!(bytes[1], 0xBB);
        assert_eq!(bytes[4096], 0xBB);
    }

    #[test]
    fn extract_from_block_two_blocks_partial() {
        let mut first = AlignedBuf::zeroed(4096);
        for byte in first.iter_mut() {
            *byte = 0x11;
        }
        let mut second = AlignedBuf::zeroed(4096);
        for byte in second.iter_mut() {
            *byte = 0x22;
        }
        let v = VarMap::<[u8; 8]>::extract_from_block(&first, 4000, 200, || {
            Ok(Arc::new(second))
        })
        .expect("ok");
        let bytes = v.as_bytes();
        assert_eq!(bytes.len(), 200);
        assert!(bytes[..96].iter().all(|b| *b == 0x11));
        assert!(bytes[96..].iter().all(|b| *b == 0x22));
    }
```

- [ ] **Step 2: Run the boundary tests**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(extract_from_block_single_block) | test(extract_from_block_two_blocks_exact) | test(extract_from_block_two_blocks_partial)'
```

Expected: 6 tests passed (3 VarTree + 3 VarMap).

- [ ] **Step 3: Commit**

```sh
git add armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
test(armdb/var_map): pin extract_from_block boundary cases

Mirror of the var_tree boundary tests: single_block,
two_blocks_exact (start+len == 8192), two_blocks_partial.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 7: VarTree — integration tests for large-value end-to-end

**Files:**
- Test only: `armdb/src/var_tree.rs` (tests module)

Three integration tests with pinned geometry that forces the entry to an immutable file, exercising the locked-path disk read (and `pread_value_encrypted` under `cfg(feature = "encryption")`). Plus a deterministic stale-DiskLoc test mirroring the existing `read_value_cached_inner_returns_stale_after_compaction` pattern at `var_tree.rs:1491-1507`.

- [ ] **Step 1: Add a helper for the large-value config**

Append to the `mod tests` block in `armdb/src/var_tree.rs` (helper goes near the existing `open_test_tree` so it is reachable from later tests):

```rust
    /// Fixed geometry that forces large entries onto an immutable file:
    /// `write_buffer_size` fits the 50 KiB entry, `max_file_size` rotates
    /// after one large write + one small trigger.
    fn open_large_value_tree(dir: &std::path::Path) -> VarTree<[u8; 8]> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        VarTree::open(dir, cfg).expect("open large-value test tree")
    }

    fn build_large_payload(seed: u8) -> Vec<u8> {
        (0..50_000u32).map(|i| (i as u8).wrapping_add(seed)).collect()
    }
```

- [ ] **Step 2: Write `large_value_read_via_locked_fallback`**

Append to the same module:

```rust
    /// End-to-end: 50 KiB value, force rotation to immutable, then `get`
    /// must return correct bytes via read_value_from_disk_locked.
    ///
    /// `open_large_value_tree` uses the default CacheConfig (max_size = 0),
    /// so the block cache is disabled and the read goes through the
    /// large-value locked fallback → disk read on every call.
    #[test]
    fn large_value_read_via_locked_fallback() {
        let dir = tempdir().unwrap();
        let tree = open_large_value_tree(dir.path());

        let key = 100u64.to_be_bytes();
        let payload = build_large_payload(0);
        tree.put(&key, &payload).expect("put large");
        tree.put(&0u64.to_be_bytes(), b"trigger-rotate")
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("read must succeed via locked fallback");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }
```

- [ ] **Step 3: Write `large_value_read_encrypted` (encryption feature only)**

Append to the same module:

```rust
    #[cfg(feature = "encryption")]
    #[test]
    fn large_value_read_encrypted() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.encryption_key = Some([7u8; 32]);
        // Cache stays at default (max_size = 0, disabled). The locked
        // fallback always goes to step 3 -> read_value_from_disk_locked,
        // which routes to pread_value_encrypted under this feature gate.
        let tree: VarTree<[u8; 8]> = VarTree::open(dir.path(), cfg).expect("open enc");

        let key = 101u64.to_be_bytes();
        let payload = build_large_payload(0xAB);
        tree.put(&key, &payload).expect("put encrypted large");
        tree.put(&0u64.to_be_bytes(), b"trigger-rotate")
            .expect("rotate");

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("encrypted large read must succeed via pread_value_encrypted");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }
```

- [ ] **Step 4: Write `large_value_stale_disk_loc_deterministic`**

Append to the same module:

```rust
    /// Deterministic stale-DiskLoc test for large values: build a large
    /// entry, capture its DiskLoc, overwrite enough to make the file fully
    /// dead, compact (removes the file), then the captured DiskLoc must
    /// surface as StaleDiskLoc via the locked fallback. The public
    /// retry path then returns the live value.
    #[test]
    fn large_value_stale_disk_loc_deterministic() {
        let dir = tempdir().unwrap();
        let tree = open_large_value_tree(dir.path());

        let key = 102u64.to_be_bytes();
        let payload = build_large_payload(0x42);
        tree.put(&key, &payload).expect("first large put");

        // Snapshot the DiskLoc of the first put — points at the file we
        // will erase via compaction.
        let snap = {
            let guard = tree.index.collector().enter();
            let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
            *node.load_disk()
        };

        // Overwrite many times with small payloads to (a) move the live
        // pointer off the original file and (b) force file rotations so the
        // original file becomes immutable AND fully dead.
        for i in 1..20u8 {
            tree.put(&key, &[i; 256]).expect("overwrite");
        }
        tree.put(&key, b"live-after-compaction").expect("final put");

        let shard = &tree.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        // Direct call on the stale snapshot — must propagate StaleDiskLoc
        // through the large-value locked fallback.
        match tree.read_value_cached_inner(&snap) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }

        // Public retry path picks up the live value.
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let v = tree
            .read_value_cached(node, &guard)
            .expect("public path must retry and return live value");
        assert_eq!(v.as_bytes(), b"live-after-compaction");
    }
```

- [ ] **Step 5: Run the integration tests (default features)**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(large_value_read_via_locked_fallback) | test(large_value_stale_disk_loc_deterministic)'
```

Expected: 2 tests passed.

- [ ] **Step 6: Run the encrypted variant**

Run:

```sh
cargo nextest run -p armdb --features 'var-collections encryption' \
    -E 'test(large_value_read_encrypted)'
```

Expected: 1 test passed.

- [ ] **Step 7: Commit**

```sh
git add armdb/src/var_tree.rs
git commit -m "$(cat <<'EOF'
test(armdb/var_tree): large-value end-to-end integration tests

Adds three integration tests with pinned geometry
(shard_count=1, write_buffer_size=128 KiB, max_file_size=4096) that
force 50 KiB entries onto an immutable file:

- large_value_read_via_locked_fallback: cold-cache disk read.
- large_value_read_encrypted (cfg(feature = "encryption")): exercises
  pread_value_encrypted on a multi-block payload.
- large_value_stale_disk_loc_deterministic: pins the StaleDiskLoc
  propagation + retry contract.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 8: VarMap — integration tests for large-value end-to-end

**Files:**
- Test only: `armdb/src/var_map.rs` (tests module)

Mirror of Task 7.

- [ ] **Step 1: Add helpers and the three tests**

Append to the `mod tests` block in `armdb/src/var_map.rs`:

```rust
    fn open_large_value_map(dir: &std::path::Path) -> VarMap<[u8; 8]> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        VarMap::open(dir, cfg).expect("open large-value test map")
    }

    fn build_large_payload(seed: u8) -> Vec<u8> {
        (0..50_000u32).map(|i| (i as u8).wrapping_add(seed)).collect()
    }

    /// `open_large_value_map` uses default CacheConfig (max_size = 0), so
    /// the block cache is disabled and the read goes through the large-value
    /// locked fallback → disk read.
    #[test]
    fn large_value_read_via_locked_fallback() {
        let dir = tempdir().unwrap();
        let map = open_large_value_map(dir.path());

        let key = 100u64.to_be_bytes();
        let payload = build_large_payload(0);
        map.put(&key, &payload).expect("put large");
        map.put(&0u64.to_be_bytes(), b"trigger-rotate")
            .expect("rotate");

        let v = map.get(&key).expect("read must succeed via locked fallback");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn large_value_read_encrypted() {
        let dir = tempdir().unwrap();
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 128 * 1024;
        cfg.compaction_threshold = 0.0;
        cfg.encryption_key = Some([7u8; 32]);
        // Cache stays at default (max_size = 0, disabled), so the locked
        // fallback always routes through pread_value_encrypted.
        let map: VarMap<[u8; 8]> = VarMap::open(dir.path(), cfg).expect("open enc");

        let key = 101u64.to_be_bytes();
        let payload = build_large_payload(0xAB);
        map.put(&key, &payload).expect("put encrypted large");
        map.put(&0u64.to_be_bytes(), b"trigger-rotate")
            .expect("rotate");

        let v = map
            .get(&key)
            .expect("encrypted large read must succeed via pread_value_encrypted");
        assert_eq!(v.as_bytes(), payload.as_slice());
    }

    #[test]
    fn large_value_stale_disk_loc_deterministic() {
        let dir = tempdir().unwrap();
        let map = open_large_value_map(dir.path());

        let key = 102u64.to_be_bytes();
        let payload = build_large_payload(0x42);
        map.put(&key, &payload).expect("first large put");

        let snap = *sync::lock(&map.indexes[0]).get(&key).expect("indexed");

        for i in 1..20u8 {
            map.put(&key, &[i; 256]).expect("overwrite");
        }
        map.put(&key, b"live-after-compaction").expect("final put");

        let shard = &map.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &map, 0.0).expect("compaction");

        match map.read_value_cached_inner(&snap) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(v) => panic!("expected StaleDiskLoc, got Ok({:?})", v.as_bytes()),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }

        let v = map.get(&key).expect("public path must retry and return live value");
        assert_eq!(v.as_bytes(), b"live-after-compaction");
    }
```

- [ ] **Step 2: Run the integration tests (default features)**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(large_value_read_via_locked_fallback) | test(large_value_stale_disk_loc_deterministic)'
```

Expected: 4 tests passed (2 VarTree + 2 VarMap).

- [ ] **Step 3: Run the encrypted variant**

Run:

```sh
cargo nextest run -p armdb --features 'var-collections encryption' \
    -E 'test(large_value_read_encrypted)'
```

Expected: 2 tests passed (1 VarTree + 1 VarMap).

- [ ] **Step 4: Commit**

```sh
git add armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
test(armdb/var_map): large-value end-to-end integration tests

Mirror of the var_tree integration tests:
- large_value_read_via_locked_fallback: cold-cache disk read.
- large_value_read_encrypted (cfg(feature = "encryption")).
- large_value_stale_disk_loc_deterministic: pins StaleDiskLoc
  propagation + retry through VarMap::get's per-shard-index retry loop.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
EOF
)"
```

---

## Task 9: Final verification

**Files:** none (verification only)

- [ ] **Step 1: Type-check with var-collections**

Run:

```sh
cargo check -p armdb --features var-collections
```

Expected: clean build.

- [ ] **Step 2: Full armdb test suite (var-collections)**

Run:

```sh
cargo nextest run -p armdb --features var-collections \
    -E 'test(var_tree) | test(var_map)'
```

Expected: all VarTree and VarMap tests pass.

- [ ] **Step 3: Targeted run including encryption-only tests**

Run:

```sh
cargo nextest run -p armdb --features 'var-collections encryption' \
    -E 'test(large_value) | test(extract_from_block) | test(read_value_locked)'
```

Expected: all targeted tests pass (including the encrypted variants).

- [ ] **Step 4: Clippy with -D warnings (var-collections scope only)**

Run:

```sh
cargo clippy -p armdb --features var-collections -- -D warnings
```

Expected: no warnings introduced by the new code. If clippy flags pre-existing unrelated warnings, leave them — the project rule is "do not fix warnings unrelated to the task."

- [ ] **Step 5: Format check**

Run:

```sh
cargo fmt --check
```

Expected: no diff.

- [ ] **Step 6: Whole-workspace test sweep (smoke)**

Run:

```sh
cargo nextest run -p armdb
```

Expected: no regressions outside the var-collections scope.

- [ ] **Step 7: Report results**

Summarize in the conversation: number of tests added, number of tests passing, any unexpected warnings. No commit (verification only). The next step is `superpowers:finishing-a-development-branch`.

---

## Notes for the executor

- **Scope discipline.** Do not touch unrelated files; do not fix unrelated warnings. The project's `CLAUDE.md` enforces this explicitly.
- **Test geometry matters.** The existing `open_test_tree` / `open_test_map` set `write_buffer_size = 8192`, which is too small to hold a 50 KiB entry. Tasks 7 and 8 introduce `open_large_value_tree` / `open_large_value_map` with `write_buffer_size = 128 * 1024`. Do not collapse the helpers.
- **The early size check goes _before_ the cache lookup.** This is the load-bearing position invariant; the spec and the `large_value_with_first_block_cached_uses_fallback` test pin it.
- **Wrapper preserves contract.** `read_value_locked` must continue to return `None` (not `Err`) on stale, with the existing `tracing::error!` line. All 11 existing call-sites (5 in VarTree, 6 in VarMap incl. the shard handles) remain unchanged.
- **`extract_from_block` keeps its two-block signature.** Do not extend it to N blocks; that is out of scope.
