# Compaction Stale DiskLoc 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 race in `VarTree`/`VarMap` lock-free reads where compaction removes an immutable file between the reader's `DiskLoc` snapshot and the disk read, causing a live key to return `None`. Replace the silent `DbError::Io(NotFound)` path with a discrete `DbError::StaleDiskLoc` variant; retry up to 3 times against a fresh index. As a structural side effect, fix two pre-existing bugs: encrypted `Shard::read_block` returning ciphertext under the same race, and `VarTree::first` dropping its `seize` guard before reading through the raw node pointer.

**Architecture:** A new `DbError::StaleDiskLoc` variant is produced by `Shard::read_block` and `ShardInner::read_value_from_disk_locked` when the requested `file_id` is no longer in `inner.immutable`. The block helpers (`get_or_read_block`/`extract_from_block`) are refactored from `Option` to `DbResult` so the variant is not swallowed across multi-block reads. The lock-free `read_value_cached` is split into an inner `DbResult<Option<ByteView>>` body and an outer retry wrapper that takes `(&VarNode<K>, &seize::LocalGuard)`; the analogous wrapper in `VarMap` re-acquires the shard's index mutex on retry. `Shard::read_block`'s encrypted-decode path is restructured to capture `Arc<TagFile>` atomically with `Arc<ImmutableFile>` inside lock 1, eliminating the existing second lock and the silent-skip-decryption window.

**Tech Stack:** Rust 2024, `parking_lot` mutex, `seize` epoch-based reclamation, `metrics` counters, `tracing`, `cargo nextest` for tests (per project `CLAUDE.md`).

**Spec:** `armdb/docs/superpowers/specs/26-05-16-compaction-stale-diskloc.md`.

---

## File Structure

Files modified:

- `armdb/src/error.rs` — add `DbError::StaleDiskLoc` variant.
- `armdb/src/shard.rs` —
  - `ActiveFile.tag_file: Option<TagFile>` → `Option<Arc<TagFile>>`.
  - `ImmutableFile.tag_file: Option<TagFile>` → `Option<Arc<TagFile>>`.
  - `Shard::read_block` — Change A (Stale on missing file_id) + Change B (capture `Arc<TagFile>` in lock 1, remove lock 2).
  - `ShardInner::read_value_from_disk_locked` — Stale on missing file_id.
  - `#[cfg(test)] mod tests` — three new unit tests.
- `armdb/src/recovery.rs` — adjust to wrap `TagFile` in `Arc` at construction.
- `armdb/src/compaction.rs` — adjust to wrap `TagFile` in `Arc` at construction.
- `armdb/src/var_tree.rs` —
  - `get_or_read_block` / `extract_from_block` — `Option` → `DbResult`.
  - `read_value_cached` → split into `read_value_cached_inner` (DbResult) + outer retry wrapper taking `(&VarNode<K>, &seize::LocalGuard)`.
  - Call sites: `get`, `first` (remove `let _ = guard;`), `migrate`, `VarIter::next`/`next_back`.
  - `read_value_locked` — log Stale as programming bug, otherwise unchanged.
  - `#[cfg(test)] mod tests` — seven new unit tests.
- `armdb/src/var_map.rs` —
  - `get_or_read_block` / `extract_from_block` — `Option` → `DbResult`.
  - `read_value_cached` → split, `get` retry loop.
  - `read_value_locked` — log Stale.
  - `#[cfg(test)] mod tests` — three new unit tests.

No on-disk format change. No new public API surface other than the new `DbError` variant. No change to `compact_shard_inner`'s observable phases.

---

## Task 1: Add `DbError::StaleDiskLoc` variant

**Files:**
- Modify: `armdb/src/error.rs`

- [ ] **Step 1: Add the variant**

In `armdb/src/error.rs`, inside the `pub enum DbError` block, add the new variant alongside the others (placement: directly after `KeyExists` is fine):

```rust
/// A copied DiskLoc points to a file that has just been removed from
/// inner.immutable by compaction. The data is logically still live;
/// the caller should retry with a fresh DiskLoc from the index.
#[error("stale DiskLoc - file removed by compaction")]
StaleDiskLoc,
```

ASCII hyphen in the message. No `#[from]`, no fields.

- [ ] **Step 2: Build the crate to confirm the variant compiles**

Run: `cargo check -p armdb --features 'var-collections'`
Expected: clean build, no warnings about unused variants (the variant is unused for now; `dead_code` will not fire because `DbError` is `pub`).

- [ ] **Step 3: Commit**

```bash
git add armdb/src/error.rs
git commit -m "feat(armdb/error): add DbError::StaleDiskLoc variant

Discrete signal for the compaction race in lock-free reads where a
DiskLoc copy outlives the immutable file it points at."
```

---

## Task 2: Wrap `TagFile` in `Arc` at storage sites

**Files:**
- Modify: `armdb/src/shard.rs:119`, `armdb/src/shard.rs:129` (struct fields).
- Modify: `armdb/src/shard.rs` (all `Some(TagFile::open_...(...)?)` constructions and `.tag_file` accessors that pass to `pread_value_encrypted` / `read_tag` / `sync` / `write_tags`).
- Modify: `armdb/src/recovery.rs:68` (the existing `Arc::new(TagFile::open_read(&tp)?)` pattern; ensure consistent).
- Modify: `armdb/src/compaction.rs:274` (`Some(TagFile::open_read(&final_tags_path)?)`).

Mechanical change. No behavior change yet.

- [ ] **Step 1: Change field types**

In `armdb/src/shard.rs`:

```rust
pub(crate) struct ActiveFile {
    pub(crate) file: std::fs::File,
    pub(crate) read_file: Arc<std::fs::File>,
    pub(crate) file_id: u32,
    pub(crate) write_offset: u64,
    pub(crate) path: PathBuf,
    #[cfg(feature = "encryption")]
    pub(crate) tag_file: Option<Arc<TagFile>>,
}

pub(crate) struct ImmutableFile {
    pub(crate) file: std::fs::File,
    pub(crate) file_id: u32,
    #[cfg(feature = "encryption")]
    pub(crate) path: PathBuf,
    pub(crate) total_bytes: u64,
    #[cfg(feature = "encryption")]
    pub(crate) tag_file: Option<Arc<TagFile>>,
}
```

- [ ] **Step 2: Wrap every construction site**

Every `Some(TagFile::open_write(...)?)` or `Some(TagFile::open_read(...)?)` that targets `ActiveFile.tag_file` or `ImmutableFile.tag_file` becomes `Some(Arc::new(TagFile::open_write(...)?))` / `Some(Arc::new(TagFile::open_read(...)?))`. Sites (current line numbers in `armdb/src/shard.rs`):

- `armdb/src/shard.rs:215` (initial file)
- `armdb/src/shard.rs:271` (immutable scan during open)
- `armdb/src/shard.rs:300` (active file in open with existing files)
- `armdb/src/shard.rs:812` and `:847` (rotate — both `new_tag_file` and `imm_tag_file` end up stored)
- `armdb/src/compaction.rs:274` (`final_tag_file` written into `ImmutableFile`)

For `armdb/src/shard.rs:812`:

```rust
let new_tag_file = if self.cipher.is_some() {
    Some(Arc::new(TagFile::open_write(&tags::tags_path_for_data(&new_path))?))
} else {
    None
};
```

For `armdb/src/shard.rs:847`:

```rust
let imm_tag_file = if self.cipher.is_some() {
    let tp = tags::tags_path_for_data(&old_path);
    if tp.exists() {
        Some(Arc::new(TagFile::open_read(&tp)?))
    } else {
        None
    }
} else {
    None
};
```

For `armdb/src/compaction.rs:271-277`:

```rust
let final_tag_file = if let Some(ref tp) = tmp_tags_path {
    let final_tags_path = shard.dir().join(format!("{new_file_id:06}.tags"));
    fs::rename(tp, &final_tags_path)?;
    Some(Arc::new(TagFile::open_read(&final_tags_path)?))
} else {
    None
};
```

- [ ] **Step 3: Fix `pread_value_encrypted` call sites that pass `&TagFile`**

`pread_value_encrypted(file, tag_file: &TagFile, ...)` keeps its current signature. When the caller now holds `Option<Arc<TagFile>>` and matches `Some(tag_file)`, pass `tag_file.as_ref()` (or `&**tag_file`) — `Arc<T>` does not coerce to `&T` in pattern bindings directly. Concretely, in `armdb/src/shard.rs:942` and `:967`:

```rust
if let (Some(cipher), Some(tag_file)) = (&self.cipher, &self.active.tag_file) {
    return crate::io::direct::pread_value_encrypted(
        &self.active.read_file,
        tag_file.as_ref(),
        cipher,
        target_file_id,
        loc.offset as u64,
        len,
    );
}
```

Apply the same `.as_ref()` change in `armdb/src/shard.rs:521` (flush path), `:861` (rotate path), and any other call site `grep -n 'pread_value_encrypted\|encrypt_page_in_place\|decrypt_page' armdb/src/shard.rs` reports.

- [ ] **Step 4: Fix non-encryption call sites that called methods on `&TagFile`**

The `tag_file.sync()`, `tag_file.write_tags(...)`, `tag_file.read_tag(...)` methods are defined on `TagFile`. Through `Arc<TagFile>`'s `Deref`, `tag_file.sync()` continues to compile when `tag_file` binds to `&Arc<TagFile>`. No change required at these sites — Rust auto-derefs `&Arc<T>` to `&T` for method dispatch.

- [ ] **Step 5: Build**

Run: `cargo check -p armdb --features 'var-collections encryption'`
Expected: clean build. Any remaining mismatch (e.g., a place that constructs a non-`Arc` `TagFile` and stores it in a field) is the compile error to fix.

- [ ] **Step 6: Run existing tests to confirm no behavior regression**

Run: `cargo nextest run -p armdb --features 'var-collections encryption'`
Expected: same delta as `main` — pre-existing failures `test_encrypted_no_key_fails` and `test_unencrypted_with_key_fails` remain (not introduced by this change); all other tests pass.

- [ ] **Step 7: Commit**

```bash
git add armdb/src/shard.rs armdb/src/compaction.rs armdb/src/recovery.rs
git commit -m "refactor(armdb): wrap ActiveFile/ImmutableFile tag_file in Arc

Mechanical rewrap of Option<TagFile> to Option<Arc<TagFile>>. Enables
atomic capture of the tag-file reference inside read_block's first lock,
eliminating the second-lock encrypted-decode pass in the next commit."
```

---

## Task 3: `Shard::read_block` — Stale signal and single-lock encrypted decode

**Files:**
- Modify: `armdb/src/shard.rs:401-469` (the `read_block` body).
- Test: `armdb/src/shard.rs#[cfg(test)] mod tests` (new).

- [ ] **Step 1: Restructure the lock-1 capture block**

Replace lines 406-431 of the current `read_block`:

```rust
pub fn read_block(
    &self,
    file_id: u32,
    block_offset: u64,
) -> DbResult<(crate::io::aligned_buf::AlignedBuf, bool)> {
    #[cfg(feature = "encryption")]
    let cipher_ref;
    #[cfg(feature = "encryption")]
    let tag_file_arc_opt: Option<Arc<crate::io::tags::TagFile>>;

    let (active_read, immutable_arc, immutable_total) = {
        let inner = sync::lock(&self.inner);

        #[cfg(feature = "encryption")]
        {
            cipher_ref = inner.cipher.clone();
        }

        if inner.active.file_id == file_id {
            #[cfg(feature = "encryption")]
            {
                tag_file_arc_opt = inner.active.tag_file.clone();
            }
            (Some(inner.active.read_file.clone()), None, 0)
        } else {
            let arc = inner
                .immutable
                .iter()
                .find(|f| f.file_id == file_id)
                .ok_or(DbError::StaleDiskLoc)?
                .clone();
            let total = arc.total_bytes;
            #[cfg(feature = "encryption")]
            {
                tag_file_arc_opt = arc.tag_file.clone();
            }
            (None, Some(arc), total)
        }
    };
```

- [ ] **Step 2: Replace the second-lock decrypt block with the captured Arc**

Replace lines 448-466 (the current `#[cfg(feature = "encryption")] if let Some(cipher) = &cipher_ref { ... }` block) with:

```rust
    #[cfg(feature = "encryption")]
    if let Some(cipher) = &cipher_ref {
        let page_number = block_offset / 4096;
        match &tag_file_arc_opt {
            Some(tf) => {
                let tag = tf.read_tag(page_number)?;
                cipher.decrypt_page(file_id, page_number, &mut buf, &tag)?;
            }
            None => {
                // Encrypted DB but no tag file for this file_id under the
                // captured immutable Arc. The first lookup succeeded, so this
                // is either the active file with encryption configured but no
                // active tag file (a programming bug, not a stale race), or
                // an immutable whose tag handle was lost. Either way, refuse
                // to return ciphertext as plaintext.
                if active_read.is_none() {
                    return Err(DbError::StaleDiskLoc);
                }
                // Active file: continue with plain bytes only if cipher is
                // legitimately not set up for this file (unreachable today;
                // active rotation always creates a tag file when cipher is on).
                return Err(DbError::StaleDiskLoc);
            }
        }
    }

    Ok((buf, is_full_block))
}
```

- [ ] **Step 3: Build**

Run: `cargo check -p armdb --features 'var-collections encryption'` and `cargo check -p armdb --features 'var-collections'`
Expected: both clean. The non-encryption build naturally elides the `tag_file_arc_opt` binding.

- [ ] **Step 4: Write the unit test module skeleton in `armdb/src/shard.rs`**

Append to the end of `armdb/src/shard.rs`:

```rust
#[cfg(test)]
mod tests {
    use super::*;
    use crate::error::DbError;
    use std::sync::atomic::AtomicU64;
    use tempfile::tempdir;

    fn open_test_shard(dir: &std::path::Path) -> Shard {
        let gsn = Arc::new(AtomicU64::new(0));
        Shard::open(0, dir, 1 << 20, 64 * 1024, false, gsn).expect("open test shard")
    }

    #[test]
    fn read_block_returns_stale_for_unknown_file_id() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        // The freshly opened shard has only file_id == 1 as active. file_id 9999
        // exists in neither active nor immutable.
        match shard.read_block(9999, 0) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }
}
```

- [ ] **Step 5: Run the test**

Run: `cargo nextest run -p armdb --features 'var-collections' shard::tests::read_block_returns_stale_for_unknown_file_id`
Expected: PASS.

- [ ] **Step 6: Add the encrypted captured-tag-file test**

In the same `mod tests` block in `armdb/src/shard.rs`:

```rust
    #[cfg(feature = "encryption")]
    #[test]
    fn read_block_returns_stale_when_immutable_missing_under_encryption() {
        use crate::crypto::PageCipher;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Some(Arc::new(PageCipher::new([0x42; 32])));
        let shard = Shard::open_encrypted(0, dir.path(), 1 << 20, 64 * 1024, false, cipher, gsn)
            .expect("open encrypted shard");

        // The shard has only active file_id=1 right now. read_block on a
        // non-existent file_id must hit Change A and return Stale before any
        // encrypted-decode path runs. We pick file_id=42 to make the intent
        // explicit (it is not 1 and not anything in immutable).
        match shard.read_block(42, 0) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }
```

(The richer scenario — write a value, force rotation so it lives in an immutable file, snapshot the file_id, compact, then call `read_block(F_old, off)` — is covered end-to-end by Tasks 6/9. This local test verifies Change A alone wires through the encrypted feature path.)

Run: `cargo nextest run -p armdb --features 'var-collections encryption' shard::tests::`
Expected: both new tests PASS.

- [ ] **Step 7: Commit**

```bash
git add armdb/src/shard.rs
git commit -m "fix(armdb/shard): return StaleDiskLoc on missing file_id, drop encrypted re-lock

Change A: the immutable lookup in Shard::read_block now returns
DbError::StaleDiskLoc instead of Io(NotFound) when the requested file_id
is no longer in inner.immutable. Change B: the encrypted-decode path
captures Arc<TagFile> atomically with the immutable Arc inside lock 1,
eliminating the second sync::lock and the silent-skip-decrypt window
where compaction between locks could have returned ciphertext as
plaintext and cached it.

Two unit tests cover Change A directly; encrypted end-to-end coverage
lives in var_tree/var_map tests in subsequent commits."
```

---

## Task 4: `ShardInner::read_value_from_disk_locked` — Stale signal

**Files:**
- Modify: `armdb/src/shard.rs:955-964`.
- Test: `armdb/src/shard.rs#[cfg(test)] mod tests`.

- [ ] **Step 1: Change the error variant**

Replace lines 955-964 of the current body:

```rust
let imm = self
    .immutable
    .iter()
    .find(|f| f.file_id == target_file_id)
    .ok_or(crate::error::DbError::StaleDiskLoc)?;
```

(Drop the `Io(NotFound)` constructor.)

- [ ] **Step 2: Add the test**

In `armdb/src/shard.rs#[cfg(test)] mod tests`:

```rust
    #[test]
    fn read_value_from_disk_locked_returns_stale_for_unknown_file_id() {
        use crate::shard::DiskLoc;

        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        let fake = DiskLoc::new(9999, 0, 0);
        let inner = shard.lock();
        match inner.read_value_from_disk_locked(&fake) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }
```

If `DiskLoc` is not at `crate::shard::DiskLoc`, locate it via `grep -n 'pub struct DiskLoc' armdb/src/` and adjust the import. (Likely at `armdb/src/types.rs` or similar — it is the `Cid::DiskLoc` used in `var_tree.rs`.)

- [ ] **Step 3: Build + run**

Run: `cargo nextest run -p armdb --features 'var-collections' shard::tests::`
Expected: all three tests PASS.

- [ ] **Step 4: Commit**

```bash
git add armdb/src/shard.rs
git commit -m "fix(armdb/shard): StaleDiskLoc from read_value_from_disk_locked

Mirrors Shard::read_block Change A under the shard lock. Under the lock
this branch is logically unreachable (compaction takes the same lock for
its immutable-retain phase), but the variant gives the assumption a
visible signal if it ever fires."
```

---

## Task 5: `VarTree` block helpers — `Option` → `DbResult`

**Files:**
- Modify: `armdb/src/var_tree.rs:778-795` (`extract_from_block`).
- Modify: `armdb/src/var_tree.rs:798-830` (`get_or_read_block`).
- Test: `armdb/src/var_tree.rs#[cfg(test)] mod tests`.

This task changes signatures but does not yet touch the outer `read_value_cached` — that lives in Task 6, which is the only consumer.

- [ ] **Step 1: Rewrite `extract_from_block` to take and return `DbResult`**

Replace lines 778-795 with:

```rust
    fn extract_from_block(
        block: &AlignedBuf,
        start: usize,
        len: usize,
        next_block: impl FnOnce() -> DbResult<Arc<AlignedBuf>>,
    ) -> DbResult<ByteView> {
        if start + len <= 4096 {
            Ok(ByteView::new(&block[start..start + len]))
        } else {
            let next = next_block()?;
            let first_part = &block[start..];
            let second_len = len - first_part.len();
            let mut combined = Vec::with_capacity(len);
            combined.extend_from_slice(first_part);
            combined.extend_from_slice(&next[..second_len]);
            Ok(ByteView::from_vec(combined))
        }
    }
```

- [ ] **Step 2: Rewrite `get_or_read_block` to return `DbResult`**

Replace lines 798-830 with:

```rust
    fn get_or_read_block(
        &self,
        shard_id: u8,
        file_id: u32,
        block_offset: u64,
    ) -> DbResult<Arc<AlignedBuf>> {
        let key = BlockKey {
            shard_id,
            file_id,
            block_offset,
        };
        if let Some(cached) = self.cache.get(&key) {
            return Ok(cached);
        }
        let shard = &self.engine.shards()[shard_id as usize];
        let (buf, is_full_block) = shard.read_block(file_id, block_offset)?;
        let arc = Arc::new(buf);
        if is_full_block {
            self.cache.insert(key, arc.clone());
        }
        Ok(arc)
    }
```

The `#[cfg(feature = "hot-path-tracing")]` error trace is removed — `?` propagates the error, and the caller (Task 6) decides what to log.

- [ ] **Step 3: Build (expect failures from current `read_value_cached` body, which still uses `Option`)**

Run: `cargo check -p armdb --features 'var-collections'`
Expected: compile errors at lines 754-757 and 772-775 (inside `read_value_cached`) because they call helpers that now return `DbResult`. Leave those errors in place for Task 6 to fix; **do not commit yet**.

- [ ] **Step 4: Hold this task open and proceed to Task 6**

The two refactors (block helpers + `read_value_cached`) are interlocked: each forces the other to change. We commit them in a single logical step at the end of Task 6 to keep the crate compilable. Document this in the running task list.

(No commit at the end of Task 5.)

---

## Task 6: `VarTree::read_value_cached` — split + retry + call-site fixes

**Files:**
- Modify: `armdb/src/var_tree.rs:741-776` (the current `read_value_cached`).
- Modify: `armdb/src/var_tree.rs:209-211` (`get` call site).
- Modify: `armdb/src/var_tree.rs:436-441` (`first` call site, including the buggy `let _ = guard;`).
- Modify: `armdb/src/var_tree.rs:715-718` (`migrate` call site).
- Modify: `armdb/src/var_tree.rs:1205-1207`, `:1249-1251` (`VarIter::next`/`next_back`).
- Test: `armdb/src/var_tree.rs#[cfg(test)] mod tests`.

- [ ] **Step 1: Add the `MAX_STALE_RETRIES` constant**

At the top of the `impl<K: Key, H: WriteHook<K>> VarTree<K, H>` block (or as a module-level `const`), add:

```rust
const MAX_STALE_RETRIES: usize = 3;
```

Module-level is fine: place it directly after the existing `use` statements in `armdb/src/var_tree.rs`.

- [ ] **Step 2: Rewrite `read_value_cached`**

Replace lines 739-776 with:

```rust
    /// Lock-free read body. Returns `Ok(Some(v))` on success, `Ok(None)` if
    /// the value is genuinely absent at the supplied DiskLoc (e.g. the cache
    /// returned a partial block at end-of-file), and `Err(StaleDiskLoc)` if
    /// compaction removed the file referenced by `disk` between the caller's
    /// snapshot and the read. Other `DbError`s indicate I/O or decryption
    /// failure.
    fn read_value_cached_inner(&self, disk: &DiskLoc) -> DbResult<Option<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);
            let v = Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
            })?;
            return Ok(Some(v));
        }

        // 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(Some(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)?;
        let v = Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
        })?;
        Ok(Some(v))
    }

    /// Lock-free read with bounded retry on compaction race. The caller must
    /// keep `guard` alive for the duration of the call; `node` is re-read each
    /// iteration so the retry observes the post-compaction DiskLoc.
    fn read_value_cached(
        &self,
        node: &VarNode<K>,
        guard: &seize::LocalGuard<'_>,
    ) -> Option<ByteView> {
        let _ = guard; // explicit anchor for the caller's epoch
        for _ in 0..MAX_STALE_RETRIES {
            let disk = *node.load_disk();
            match self.read_value_cached_inner(&disk) {
                Ok(Some(v)) => return Some(v),
                Ok(None) => return None,
                Err(DbError::StaleDiskLoc) => {
                    metrics::counter!("armdb.read.stale_retry", "tree" => "var_tree")
                        .increment(1);
                    continue;
                }
                Err(_e) => {
                    #[cfg(feature = "hot-path-tracing")]
                    tracing::error!("VarTree read_value_cached error: {:?}", _e);
                    return None;
                }
            }
        }
        None
    }
```

`DbError` is already imported as `crate::error::DbError` in this file (or `super::DbError` from the `use` block); check the existing imports near the top and adjust if missing.

- [ ] **Step 3: Fix the `get` call site (`armdb/src/var_tree.rs:209-211`)**

Replace:

```rust
        let disk = *node.load_disk();
        self.read_value_cached(&disk)
```

with:

```rust
        self.read_value_cached(node, &guard)
```

(The `guard` binding is already in scope from line 197.)

- [ ] **Step 4: Fix `first` — remove the pre-existing UB-window `let _ = guard;`**

Replace lines 436-447 with:

```rust
        while !ptr.is_null() {
            let node = unsafe { &*ptr };
            if !node.is_marked() {
                return self
                    .read_value_cached(node, &guard)
                    .map(|v| (node.key, v));
            }
            ptr = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
        }
```

`guard` is now naturally alive until the function returns.

- [ ] **Step 5: Fix `migrate` (`armdb/src/var_tree.rs:715-719`)**

Rename `_guard` to `guard` at line 700, then replace:

```rust
            let disk = *node.load_disk();
            let value = match self.read_value_cached(&disk) {
                Some(v) => v,
                None => continue,
            };
```

with:

```rust
            let value = match self.read_value_cached(node, &guard) {
                Some(v) => v,
                None => continue,
            };
```

(The `disk` local is unused after this and can be removed.)

- [ ] **Step 6: Fix `VarIter::next` (`armdb/src/var_tree.rs:1205-1207`)**

Replace:

```rust
            let disk = *node.load_disk();
            match self.tree.read_value_cached(&disk) {
                Some(value) => return Some((node.key, value)),
```

with:

```rust
            match self.tree.read_value_cached(node, &self._guard) {
                Some(value) => return Some((node.key, value)),
```

Drop the `disk` local. Same for `VarIter::next_back` at lines 1249-1251.

- [ ] **Step 7: Build**

Run: `cargo check -p armdb --features 'var-collections'`
Expected: clean. Note: this is the first checkpoint where the crate compiles again after Task 5.

- [ ] **Step 8: Run existing tests**

Run: `cargo nextest run -p armdb --features 'var-collections'`
Expected: same delta as `main` (pre-existing encryption failures remain; all other tests pass). This validates that the refactor preserves behavior for non-stale read paths.

- [ ] **Step 9: Add unit tests in `armdb/src/var_tree.rs#[cfg(test)] mod tests`**

Append to the end of `armdb/src/var_tree.rs`:

```rust
#[cfg(test)]
mod tests {
    use super::*;
    use crate::Config;
    use crate::compaction::compact_shard;
    use tempfile::tempdir;

    fn open_test_tree(dir: &std::path::Path) -> VarTree<[u8; 8]> {
        // Tight geometry so a handful of overwrites accumulate dead bytes:
        // max_file_size=4096 forces rotation per put, compaction_threshold=0.0
        // makes every dead byte eligible.
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 8192;
        cfg.compaction_threshold = 0.0;
        VarTree::open(dir, cfg).expect("open test tree")
    }

    fn put_until_compactable(tree: &VarTree<[u8; 8]>, key: [u8; 8]) -> DiskLoc {
        // Overwrite enough times to leave dead bytes in at least one immutable
        // file. Final put is the live one.
        for i in 0..6u8 {
            tree.put(&key, &[i; 256]).expect("put");
        }
        tree.put(&key, b"final-value-payload-XX").expect("final put");
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        *node.load_disk()
    }

    #[test]
    fn read_value_cached_inner_returns_stale_after_compaction() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

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

        // Run compaction synchronously. After it returns, the immutable file
        // referenced by `snap.file_id` is gone and the index points to a new
        // file_id. The captured `snap` is exactly the stale value a racing
        // reader would have copied.
        let shard = &tree.engine.shards()[snap.shard_id as usize];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        match tree.read_value_cached_inner(&snap) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }

    #[test]
    fn read_value_cached_returns_some_after_compaction() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 11u64.to_be_bytes();
        let _snap = put_until_compactable(&tree, key);
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        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("post-compaction value must be readable");
        assert_eq!(v.as_bytes(), b"final-value-payload-XX");
    }

    #[test]
    fn get_during_compaction_returns_some() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 13u64.to_be_bytes();
        let _snap = put_until_compactable(&tree, key);
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        // Public API works end-to-end after compaction; we are not asserting
        // the retry counter here because compact_shard already updated the
        // index before the get() acquires the guard, so the loop's first
        // iteration succeeds. See test 14 for retry-limit coverage.
        let v = tree.get(&key).expect("post-compaction get");
        assert_eq!(v.as_bytes(), b"final-value-payload-XX");
    }

    #[test]
    fn iter_during_compaction_yields_all_live_keys() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        // Three live keys, each overwritten to force dead-byte accumulation.
        for k in 1u64..=3 {
            put_until_compactable(&tree, k.to_be_bytes());
        }
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        let collected: std::collections::BTreeMap<[u8; 8], Vec<u8>> = tree
            .iter()
            .map(|(k, v)| (k, v.as_bytes().to_vec()))
            .collect();
        assert_eq!(collected.len(), 3);
        for k in 1u64..=3 {
            let bytes = collected
                .get(&k.to_be_bytes())
                .expect("every original key must remain");
            assert_eq!(bytes.as_slice(), b"final-value-payload-XX");
        }
    }

    #[test]
    fn get_or_read_block_returns_stale_for_unknown_file_id() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        match tree.get_or_read_block(0, 9999, 0) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }

    #[test]
    fn extract_from_block_propagates_next_block_error() {
        // Pure-function test: 4096-byte block, start/len chosen so the value
        // crosses the boundary. The `next_block` closure returns Stale.
        let block = AlignedBuf::zeroed(4096);
        let start = 4090;
        let len = 32;
        let result: DbResult<ByteView> =
            VarTree::<[u8; 8]>::extract_from_block(&block, start, len, || {
                Err(DbError::StaleDiskLoc)
            });
        match result {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }

    #[test]
    fn extract_from_block_multi_block_first_cached_second_stale() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 21u64.to_be_bytes();
        // Value > 4072 bytes guarantees crossing a 4096 boundary inside the
        // file. Use the same geometry as armdb/tests/var_correctness_tests.rs.
        let value = vec![0xCDu8; 4073];
        tree.put(&key, &value).expect("initial put");
        // One filler write to force rotation so `key`'s data file is immutable.
        tree.put(&22u64.to_be_bytes(), &vec![0u8; 1000]).expect("filler");

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

        // Run compaction; the immutable referenced by `snap` is removed.
        let shard = &tree.engine.shards()[0];
        let _ = compact_shard(shard, &tree, 0.0).expect("compaction");

        // Manually seed only the FIRST block of `snap` into the cache so the
        // first-block read hits the cache, then the second-block read goes to
        // disk and triggers Stale.
        let block_offset = snap.offset as u64 & !4095;
        let first_block = AlignedBuf::zeroed(4096);
        let cache_key = BlockKey {
            shard_id: snap.shard_id,
            file_id: snap.file_id as u32,
            block_offset,
        };
        tree.cache.insert(cache_key, Arc::new(first_block));

        match tree.read_value_cached_inner(&snap) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc from second block, got {other:?}"),
        }
    }
}
```

Notes for the engineer:
- `tree.index`, `tree.engine`, `tree.cache` are private. The `mod tests` block is inside the same module, so it has full access.
- `BlockKey`, `AlignedBuf`, `DbError`, `DbResult`, `Arc` need to be in scope from `super::*`. If they are not exported at module level, add explicit `use super::*` plus targeted imports.
- The `put_until_compactable` helper is shared across tests; place it just below `open_test_tree`.

- [ ] **Step 10: Run the new tests**

Run: `cargo nextest run -p armdb --features 'var-collections' var_tree::tests::`
Expected: all seven tests PASS.

- [ ] **Step 11: Run the full var-collections test set**

Run: `cargo nextest run -p armdb --features 'var-collections'`
Expected: same delta as `main` for unrelated tests; all new tests pass.

- [ ] **Step 12: Commit Tasks 5+6 together**

```bash
git add armdb/src/var_tree.rs
git commit -m "fix(armdb/var_tree): retry lock-free reads on StaleDiskLoc

Splits read_value_cached into a DbResult-returning inner body and an
outer retry wrapper that takes (&VarNode<K>, &seize::LocalGuard) and
re-reads node.load_disk() up to MAX_STALE_RETRIES=3 times. Refactors
get_or_read_block and extract_from_block to propagate DbResult so
StaleDiskLoc is not swallowed across multi-block reads.

Fixes a pre-existing UB window in VarTree::first where 'let _ = guard;'
dropped the seize guard before reading through the raw node pointer.

Adds seven unit tests covering the stale signal, the retry, the helper
DbResult propagation, multi-block stale-on-second-block, and end-to-end
get/iter behavior across a synchronous compaction."
```

---

## Task 7: `VarTree::read_value_locked` — log Stale as programming bug

**Files:**
- Modify: `armdb/src/var_tree.rs:973-975`.

- [ ] **Step 1: Replace the disk-read branch**

In `read_value_locked` (current line 974):

```rust
        // 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))
```

(ASCII hyphen in the log message.)

- [ ] **Step 2: Build + run**

Run: `cargo nextest run -p armdb --features 'var-collections'`
Expected: same delta as before. No new test is added — under the lock, Stale is logically impossible, and adding a forced-failure test would require manually corrupting `inner.immutable`, which Task 11 (the retry-limit test) already does for `VarTree::read_value_cached`.

- [ ] **Step 3: Commit**

```bash
git add armdb/src/var_tree.rs
git commit -m "fix(armdb/var_tree): log StaleDiskLoc under shard lock

Defensive branch in read_value_locked. Stale is impossible under the
shard lock by design (compaction takes the same lock), but if the
invariant breaks the log records it instead of silently returning None."
```

---

## Task 8: `VarMap` block helpers — `Option` → `DbResult`

**Files:**
- Modify: `armdb/src/var_map.rs:547-595`.
- Test: `armdb/src/var_map.rs#[cfg(test)] mod tests`.

Same interlock as Task 5/6: hold the commit until Task 9 completes the consumer refactor.

- [ ] **Step 1: Rewrite `extract_from_block`**

Replace lines 547-564 with the same `DbResult` shape as `VarTree` (the body is byte-for-byte identical):

```rust
    fn extract_from_block(
        block: &AlignedBuf,
        start: usize,
        len: usize,
        next_block: impl FnOnce() -> DbResult<Arc<AlignedBuf>>,
    ) -> DbResult<ByteView> {
        if start + len <= 4096 {
            Ok(ByteView::new(&block[start..start + len]))
        } else {
            let next = next_block()?;
            let first_part = &block[start..];
            let second_len = len - first_part.len();
            let mut combined = Vec::with_capacity(len);
            combined.extend_from_slice(first_part);
            combined.extend_from_slice(&next[..second_len]);
            Ok(ByteView::from_vec(combined))
        }
    }
```

- [ ] **Step 2: Rewrite `get_or_read_block`**

Replace lines 566-595 with:

```rust
    fn get_or_read_block(
        &self,
        shard_id: u8,
        file_id: u32,
        block_offset: u64,
    ) -> DbResult<Arc<AlignedBuf>> {
        let key = BlockKey {
            shard_id,
            file_id,
            block_offset,
        };
        if let Some(cached) = self.cache.get(&key) {
            return Ok(cached);
        }
        let shard = &self.engine.shards()[shard_id as usize];
        let (buf, is_full_block) = shard.read_block(file_id, block_offset)?;
        let arc = Arc::new(buf);
        if is_full_block {
            self.cache.insert(key, arc.clone());
        }
        Ok(arc)
    }
```

- [ ] **Step 3: Build — expect failures from `read_value_cached`**

Run: `cargo check -p armdb --features 'var-collections'`
Expected: compile errors at the current `read_value_cached` body that calls the helpers. Proceed to Task 9; **do not commit yet**.

---

## Task 9: `VarMap::read_value_cached` — split + `get` retry

**Files:**
- Modify: `armdb/src/var_map.rs:510-545` (the current `read_value_cached`).
- Modify: `armdb/src/var_map.rs:157-166` (`get`).
- Test: `armdb/src/var_map.rs#[cfg(test)] mod tests`.

- [ ] **Step 1: Add `MAX_STALE_RETRIES` to `var_map.rs`**

Either reuse the constant from `var_tree.rs` via `use crate::var_tree::MAX_STALE_RETRIES;` (only works if Task 6 made it `pub(crate)`), or define a private duplicate at the top of `var_map.rs`:

```rust
const MAX_STALE_RETRIES: usize = 3;
```

Pick the duplicate path; it keeps the modules independent and the constant tiny. (Tweak Task 6 if you prefer a shared constant.)

- [ ] **Step 2: Rewrite `read_value_cached`**

Replace lines 510-545 with:

```rust
    fn read_value_cached_inner(&self, disk: &DiskLoc) -> DbResult<Option<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);
            let v = Self::extract_from_block(&block, start, len, || {
                self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
            })?;
            return Ok(Some(v));
        }

        {
            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(Some(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)?;
        let v = Self::extract_from_block(&block, start, len, || {
            self.get_or_read_block(disk.shard_id, disk.file_id as u32, block_offset + 4096)
        })?;
        Ok(Some(v))
    }
```

- [ ] **Step 3: Rewrite `get` with retry loop**

Replace lines 157-166 with:

```rust
    pub fn get(&self, key: &K) -> Option<ByteView> {
        metrics::counter!("armdb.ops", "op" => "get", "tree" => "var_map").increment(1);
        #[cfg(feature = "hot-path-tracing")]
        tracing::trace!("var_map.get");
        let shard_id = self.shard_for(key);
        for _ in 0..MAX_STALE_RETRIES {
            let disk = {
                let index = sync::lock(&self.indexes[shard_id]);
                match index.get(key) {
                    Some(d) => *d,
                    None => return None,
                }
            };
            match self.read_value_cached_inner(&disk) {
                Ok(Some(v)) => return Some(v),
                Ok(None) => return None,
                Err(DbError::StaleDiskLoc) => {
                    metrics::counter!("armdb.read.stale_retry", "tree" => "var_map")
                        .increment(1);
                    continue;
                }
                Err(_e) => {
                    #[cfg(feature = "hot-path-tracing")]
                    tracing::error!("VarMap read_value_cached error: {:?}", _e);
                    return None;
                }
            }
        }
        None
    }
```

- [ ] **Step 4: Build**

Run: `cargo check -p armdb --features 'var-collections'`
Expected: clean.

- [ ] **Step 5: Existing tests**

Run: `cargo nextest run -p armdb --features 'var-collections'`
Expected: same delta as `main`; all previous `var_correctness_tests.rs` tests pass.

- [ ] **Step 6: Add unit tests in `armdb/src/var_map.rs#[cfg(test)] mod tests`**

Append to the end of `armdb/src/var_map.rs`:

```rust
#[cfg(test)]
mod tests {
    use super::*;
    use crate::Config;
    use crate::compaction::compact_shard;
    use tempfile::tempdir;

    fn open_test_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 = 8192;
        cfg.compaction_threshold = 0.0;
        VarMap::open(dir, cfg).expect("open test map")
    }

    fn put_until_compactable(map: &VarMap<[u8; 8]>, key: [u8; 8]) -> DiskLoc {
        for i in 0..6u8 {
            map.put(&key, &[i; 256]).expect("put");
        }
        map.put(&key, b"final-value-payload-XX").expect("final put");
        *sync::lock(&map.indexes[0]).get(&key).expect("indexed")
    }

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

        let key = 7u64.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");

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

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

        let key = 11u64.to_be_bytes();
        let _snap = put_until_compactable(&map, key);
        let shard = &map.engine.shards()[0];
        let _ = compact_shard(shard, &map, 0.0).expect("compaction");

        let v = map.get(&key).expect("post-compaction get");
        assert_eq!(v.as_bytes(), b"final-value-payload-XX");
    }

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

        match map.get_or_read_block(0, 9999, 0) {
            Err(DbError::StaleDiskLoc) => {}
            other => panic!("expected StaleDiskLoc, got {other:?}"),
        }
    }
}
```

Notes:
- `map.engine`, `map.indexes`, `map.cache` are private; `mod tests` has full access.
- The `CompactionIndex` trait must be implemented for `&VarMap<K>` (it is, per `var_map.rs` existing impls); `compact_shard(shard, &map, ...)` mirrors how `VarMap::compact` calls it internally.

- [ ] **Step 7: Run the new tests**

Run: `cargo nextest run -p armdb --features 'var-collections' var_map::tests::`
Expected: all three tests PASS.

- [ ] **Step 8: Commit Tasks 8+9 together**

```bash
git add armdb/src/var_map.rs
git commit -m "fix(armdb/var_map): retry lock-free reads on StaleDiskLoc

Splits read_value_cached into a DbResult-returning inner body and lifts
the retry loop into VarMap::get, which re-acquires the shard's index
HashMap mutex on retry to observe the post-compaction DiskLoc. Refactors
get_or_read_block and extract_from_block to propagate DbResult, mirroring
the VarTree change.

Adds three unit tests covering the stale signal, the helper, and
end-to-end get behavior across a synchronous compaction."
```

---

## Task 10: `VarMap::read_value_locked` — log Stale

**Files:**
- Modify: `armdb/src/var_map.rs` (the disk-read branch in `read_value_locked`).

- [ ] **Step 1: Locate the disk-read branch**

Run: `grep -n 'read_value_from_disk_locked\|read_value_locked' armdb/src/var_map.rs | head`
Expected: a single disk branch at the end of `read_value_locked`. Replace it analogously to Task 7:

```rust
        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))
```

- [ ] **Step 2: Build + tests**

Run: `cargo nextest run -p armdb --features 'var-collections'`
Expected: same delta as before.

- [ ] **Step 3: Commit**

```bash
git add armdb/src/var_map.rs
git commit -m "fix(armdb/var_map): log StaleDiskLoc under shard lock

Defensive branch in read_value_locked. Mirrors the VarTree change."
```

---

## Task 11: Retry-limit smoke test

**Files:**
- Modify: `armdb/src/var_tree.rs#[cfg(test)] mod tests`.

- [ ] **Step 1: Add the test**

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

```rust
    #[test]
    fn retry_limit_returns_none_on_persistent_stale() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree(dir.path());

        let key = 99u64.to_be_bytes();
        tree.put(&key, b"payload").expect("put");

        // Force every read_value_cached_inner call to see Stale by manually
        // emptying inner.immutable AND moving the active file_id out of the
        // way. Simpler approach: clear immutable AND rotate the active file
        // until node.disk's file_id is no longer current. The cleanest path
        // is to read the node's file_id, then write enough to push past it.
        //
        // Instead, take the shard lock and forcibly clear the immutable Vec
        // and reassign active.file_id to something different from the node's
        // file_id. The node's load_disk() still points at the now-unreachable
        // file_id, so every retry returns Stale.
        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("indexed");
        let snap = *node.load_disk();
        drop(guard);

        {
            let shard = &tree.engine.shards()[snap.shard_id as usize];
            let mut inner = shard.lock();
            inner.immutable.clear();
            // Reassign active.file_id to a value that does not equal snap.file_id.
            // snap.file_id is currently active because we wrote once into a
            // fresh tree; pick file_id u32::MAX as a sentinel value clearly not
            // referenced by any DiskLoc.
            inner.active.file_id = u32::MAX;
        }

        let guard = tree.index.collector().enter();
        let node = tree.index.get(key.as_bytes(), &guard).expect("still indexed");
        assert!(
            tree.read_value_cached(node, &guard).is_none(),
            "MAX_STALE_RETRIES must terminate the loop and return None"
        );
    }
```

Notes:
- `inner.immutable.clear()` is permitted because `immutable: Vec<Arc<ImmutableFile>>` is `pub(crate)`.
- Reassigning `active.file_id` mutates `ActiveFile.file_id` (also `pub(crate)`).
- After the test, the tree is dropped along with `dir`; no cleanup needed.

- [ ] **Step 2: Run**

Run: `cargo nextest run -p armdb --features 'var-collections' var_tree::tests::retry_limit_returns_none_on_persistent_stale`
Expected: PASS.

- [ ] **Step 3: Commit**

```bash
git add armdb/src/var_tree.rs
git commit -m "test(armdb/var_tree): retry-limit terminates on persistent stale"
```

---

## Task 12: Final verification

**Files:** none (verification only).

- [ ] **Step 1: Format**

Run: `cargo fmt --all`
Expected: no diff if previous tasks were committed cleanly; otherwise commit the formatting in a single follow-up.

- [ ] **Step 2: Clippy on the var-collections + encryption feature set**

Run: `cargo clippy -p armdb --features 'var-collections typed-tree encryption' -- -D warnings`
Expected: zero warnings.

- [ ] **Step 3: Workspace clippy (sanity check)**

Run: `cargo clippy -- -D warnings`
Expected: zero new warnings on the touched files. Pre-existing warnings in unrelated crates are out of scope per `CLAUDE.md`.

- [ ] **Step 4: Test the var-collections feature set**

Run: `cargo nextest run -p armdb --features 'var-collections'`
Expected: same delta as `main` (pre-existing encryption-feature failures don't appear here because the feature is off); all new tests pass.

- [ ] **Step 5: Test the var-collections + encryption feature set**

Run: `cargo nextest run -p armdb --features 'var-collections encryption'`
Expected: pre-existing failures `test_encrypted_no_key_fails` and `test_unencrypted_with_key_fails` remain (unrelated to this spec). All other tests pass, including the new `read_block_returns_stale_when_immutable_missing_under_encryption`.

- [ ] **Step 6: Confirm the test set deliberately exercises the unfixed path**

For a single representative test, temporarily revert the StaleDiskLoc change in `Shard::read_block` (the `.ok_or(DbError::StaleDiskLoc)?` becomes `.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::NotFound, "data file not found"))?`), re-run `cargo nextest run -p armdb --features 'var-collections' var_tree::tests::read_value_cached_inner_returns_stale_after_compaction`, and verify it FAILS (the test expects `Err(DbError::StaleDiskLoc)` but receives `Err(DbError::Io(...))`). Then restore the change.

This guards against the codex review's #1 concern: tests must not be accidentally green on unfixed code.

- [ ] **Step 7: Final commit if any formatting drift**

```bash
git status
# if formatting-only diffs exist:
git commit -am "style: cargo fmt"
```

- [ ] **Step 8: Push the branch and open a PR**

Per `/Users/railka/.claude/CLAUDE.md`: on a new feature branch, `finishing-a-development-branch` does Push + PR immediately.

Run: `git push -u origin compaction-stale-diskloc`
Then open the PR. PR body should:
- Reference the spec: `armdb/docs/superpowers/specs/26-05-16-compaction-stale-diskloc.md`.
- Reference the codex review: `armdb/docs/superpowers/reviews/26-05-16-compaction-stale-diskloc.md`.
- Call out the two pre-existing bugs fixed as structural side effects (encrypted ciphertext race in `Shard::read_block`; `VarTree::first` UB window).

---

## Spec coverage cross-check

| Spec requirement | Task |
|---|---|
| `DbError::StaleDiskLoc` variant | 1 |
| `Shard::read_block` Change A (Stale on missing file_id) | 3 |
| `Shard::read_block` Change B (capture Arc<TagFile> in lock 1) | 2 (rewrap) + 3 (single-lock refactor) |
| `ShardInner::read_value_from_disk_locked` Stale | 4 |
| `VarTree::read_value_cached` split + retry, `(&VarNode, &guard)` | 6 |
| `VarTree::first` UB fix (`let _ = guard;` removed) | 6 |
| `VarTree::migrate` traversal | 6 |
| `VarIter::next` / `next_back` updated | 6 |
| Block helpers `Option` → `DbResult` (VarTree) | 5 (paired with 6) |
| `VarMap::get` retry loop | 9 |
| Block helpers `Option` → `DbResult` (VarMap) | 8 (paired with 9) |
| `read_value_locked` Stale log | 7 (VarTree), 10 (VarMap) |
| `MAX_STALE_RETRIES = 3` | 6, 9 |
| `armdb.read.stale_retry` metric | 6, 9 |
| Test 1: `read_block` stale | 3 |
| Test 2: `read_value_from_disk_locked` stale | 4 |
| Test 3: encrypted captured-tag-file | 3 (light); end-to-end via 6/9 |
| Tests 4-7 (VarTree stale, retry, get, iter) | 6 |
| Test 8: `get_or_read_block` stale (VarTree) | 6 |
| Test 9: `extract_from_block` propagates next-block error | 6 |
| Test 10: multi-block first-cached-second-stale | 6 |
| Tests 11-13 (VarMap mirror) | 9 |
| Test 14: retry-limit terminates | 11 |
| `warmup` Non-Goal (propagates Stale via `?`) | already correct — `warmup` uses `?` and is unchanged in code |
| `cargo fmt` / `clippy` / `nextest` | 12 |

Every line item in the spec maps to at least one task.
