# Var Collections Critical Correctness 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:** Fix five critical correctness bugs in `VarTree`/`VarMap` and the shared SkipList node implementation without changing the on-disk format.

**Architecture:** The fixes are narrow and additive: replace fragile pointer-int casts with explicit atomic fields, remove unsound iterator `Send` impls, plug an `InsertResult::Exists` leak in `VarTree::insert_locked`, reject oversized entries before they touch the write buffer, and route locked reads through an encryption-aware helper.

**Tech Stack:** Rust, `armdb` workspace crate, `seize` (epoch-style reclamation), `loom`/`miri` for concurrent SkipList tests, `cargo-nextest` for running armdb tests.

**Reference spec:** `armdb/docs/superpowers/specs/26-05-15-var-collections-critical-correctness.md`

---

## File Structure

Files modified by this plan and their responsibilities:

- `armdb/src/skiplist/node.rs` — adds explicit `mark: AtomicBool` field to `ConstNode`, `VarNode`, `TypedNode`; rewrites `mark()` / `is_marked()` against that field; tower pointers remain plain `AtomicPtr`.
- `armdb/src/skiplist/mod.rs` — `strip_mark` becomes an identity function (no `as usize & !1` cast). All existing call sites remain valid.
- `armdb/src/var_tree.rs` — removes `unsafe impl Send for VarIter`, fixes `VarTree::insert_locked` to free the rejected node and return `KeyExists` on race, replaces `pread_value` call in `read_value_locked` with the new encrypted-aware helper.
- `armdb/src/var_map.rs` — replaces `pread_value` call in `read_value_locked` with the new encrypted-aware helper.
- `armdb/src/const_tree.rs` — removes `unsafe impl Send for ConstIter`.
- `armdb/src/typed_tree.rs` — removes `unsafe impl Send for TypedIter`.
- `armdb/src/shard.rs` — `ShardInner::append_entry` rejects oversized entries with `DbError::Client` before consuming a GSN; adds a `pub(crate)` helper `ShardInner::read_value_from_disk_locked` that performs the plain or encrypted read using already-locked state.
- `armdb/tests/var_correctness_tests.rs` — new file with regression tests for duplicate insert, oversized values, and encrypted locked reads.

---

## Task 1: Replace tagged-pointer mark with explicit `AtomicBool` field

**Files:**
- Modify: `armdb/src/skiplist/node.rs`
- Modify: `armdb/src/skiplist/mod.rs`

Tower pointers currently double-encode mark state in their low bit. `mark()` reinterprets `&AtomicPtr<T>` as `&AtomicUsize` and `is_marked()` reads the same low bit. Replace this with a dedicated `AtomicBool` field per node and make `strip_mark` an identity function.

- [ ] **Step 1.1: Add `mark: AtomicBool` field to `ConstNode`**

In `armdb/src/skiplist/node.rs`, import `AtomicBool` alongside the existing atomics (with both `loom` and `std` cfg branches), and add a `mark` field. The `ConstNode` struct currently looks like:

```rust
#[repr(C)]
pub struct ConstNode<K: Key, const V: usize, L: Location = DiskLoc> {
    pub key: K,
    seq: AtomicU64,
    loc: UnsafeCell<L>,
    value: UnsafeCell<[u8; V]>,
    height: u8,
    tower_ptr: *const AtomicPtr<Self>,
}
```

Change it to:

```rust
#[repr(C)]
pub struct ConstNode<K: Key, const V: usize, L: Location = DiskLoc> {
    pub key: K,
    seq: AtomicU64,
    loc: UnsafeCell<L>,
    value: UnsafeCell<[u8; V]>,
    mark: AtomicBool,
    height: u8,
    tower_ptr: *const AtomicPtr<Self>,
}
```

Update the import line:

```rust
#[cfg(feature = "loom")]
use loom::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, AtomicUsize, Ordering, fence};
#[cfg(not(feature = "loom"))]
use std::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, AtomicUsize, Ordering, fence};
```

`AtomicUsize` is no longer used in this file after the next steps but keep it imported until Step 1.5 to avoid intermediate compile errors.

- [ ] **Step 1.2: Initialize `mark` in `ConstNode::alloc`**

In `ConstNode::alloc`, after the existing `std::ptr::write(&raw mut (*ptr).value, ...)` line, write:

```rust
std::ptr::write(&raw mut (*ptr).mark, AtomicBool::new(false));
```

- [ ] **Step 1.3: Rewrite `ConstNode` `SkipNode` impl mark methods**

Replace these methods in `impl SkipNode for ConstNode`:

```rust
#[inline]
fn is_marked(&self) -> bool {
    self.tower(0).load(Ordering::Acquire) as usize & 1 != 0
}

fn mark(&self) -> bool {
    let atomic_usize: &AtomicUsize =
        unsafe { &*(self.tower(0) as *const AtomicPtr<Self> as *const AtomicUsize) };
    let old = atomic_usize.fetch_or(1, Ordering::AcqRel);
    old & 1 == 0
}
```

with:

```rust
#[inline]
fn is_marked(&self) -> bool {
    self.mark.load(Ordering::Acquire)
}

fn mark(&self) -> bool {
    !self.mark.swap(true, Ordering::AcqRel)
}
```

- [ ] **Step 1.4: Repeat field, init, and method changes for `VarNode`**

In the `#[cfg(feature = "var-collections")] #[repr(C)] pub struct VarNode<K: Key>` block, add `mark: AtomicBool,` between `pub disk: AtomicPtr<DiskLoc>,` and `height: u8,`.

In `VarNode::alloc`, after `std::ptr::write(&raw mut (*ptr).disk, ...)`, add:

```rust
std::ptr::write(&raw mut (*ptr).mark, AtomicBool::new(false));
```

In `impl SkipNode for VarNode`, replace `is_marked` and `mark` with the same pattern:

```rust
#[inline]
fn is_marked(&self) -> bool {
    self.mark.load(Ordering::Acquire)
}

fn mark(&self) -> bool {
    !self.mark.swap(true, Ordering::AcqRel)
}
```

- [ ] **Step 1.5: Repeat field, init, and method changes for `TypedNode`**

In the `#[cfg(feature = "typed-tree")] #[repr(C)] pub struct TypedNode<K: Key, T>` block, add `mark: AtomicBool,` between `pub data: AtomicPtr<TypedData<T>>,` and `height: u8,`.

In `TypedNode::alloc`, after `std::ptr::write(&raw mut (*ptr).data, ...)`, add:

```rust
std::ptr::write(&raw mut (*ptr).mark, AtomicBool::new(false));
```

In the `alloc_head` impl for `TypedNode`, after the `data` write, also write `mark` as `AtomicBool::new(false)`:

```rust
std::ptr::write(&raw mut (*ptr).mark, AtomicBool::new(false));
```

In `impl SkipNode for TypedNode`, replace `is_marked` and `mark` with the same `AtomicBool` versions used in Steps 1.3 and 1.4.

Now `AtomicUsize` is unused in `node.rs`. Remove it from the import line:

```rust
#[cfg(feature = "loom")]
use loom::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, Ordering, fence};
#[cfg(not(feature = "loom"))]
use std::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, Ordering, fence};
```

- [ ] **Step 1.6: Make `strip_mark` an identity function**

In `armdb/src/skiplist/mod.rs`, replace:

```rust
#[inline(always)]
pub(crate) fn strip_mark<T>(ptr: *mut T) -> *mut T {
    (ptr as usize & !1) as *mut T
}
```

with:

```rust
#[inline(always)]
pub(crate) fn strip_mark<T>(ptr: *mut T) -> *mut T {
    ptr
}
```

Do not delete it — its many call sites in `mod.rs`, `iter.rs`, `var_tree.rs`, `var_map.rs`, etc. continue to compile and serve as documentation that the caller is reading the bottom-level pointer for traversal.

- [ ] **Step 1.7: Build and verify**

Run: `cargo check -p armdb --features var-collections,typed-tree`
Expected: PASS (no warnings about unused `AtomicUsize`).

Run: `cargo nextest run -p armdb -E 'test(skiplist)'`
Expected: PASS for all SkipList tests.

Run: `cargo nextest run -p armdb -E 'test(basic) or test(iter)'`
Expected: PASS — node lifecycle and iteration behavior unchanged.

- [ ] **Step 1.8: Commit**

```bash
git add armdb/src/skiplist/node.rs armdb/src/skiplist/mod.rs
git commit -m "$(cat <<'EOF'
fix(armdb/skiplist): replace tagged-pointer mark with explicit AtomicBool

Eliminates the unsafe AtomicPtr↔AtomicUsize aliasing used to encode the
deletion bit in the low bit of tower[0]. Each SkipList node now carries
a dedicated AtomicBool mark; strip_mark is an identity function. No
on-disk format change.

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

---

## Task 2: Remove unsound `Send` impls from `VarIter`, `ConstIter`, `TypedIter`

**Files:**
- Modify: `armdb/src/var_tree.rs:1173-1175`
- Modify: `armdb/src/const_tree.rs:1136`
- Modify: `armdb/src/typed_tree.rs:1086`

Each iterator stores a `seize::LocalGuard<'a>`. `LocalGuard` is intentionally `!Send` because it pins the current thread into the `seize::Collector` epoch — moving it to another thread would let that thread read freed nodes. The existing `unsafe impl Send` blocks are unsound.

Other types that ship `unsafe impl Send` (`SkipList`, `VarNode`, `VarTypedShard`, `ByteView`, `AlignedBuf`) are out of scope.

- [ ] **Step 2.1: Remove `unsafe impl Send for VarIter`**

In `armdb/src/var_tree.rs`, delete:

```rust
// SAFETY: VarIter holds a seize guard that prevents node reclamation.
// The raw pointer is only dereferenced while the guard is alive.
unsafe impl<K: Key, H: WriteHook<K>> Send for VarIter<'_, K, H> {}
```

- [ ] **Step 2.2: Remove `unsafe impl Send for ConstIter`**

In `armdb/src/const_tree.rs`, find and delete the `unsafe impl Send for ConstIter` line near the bottom of the file (around line 1136). Use grep to confirm the exact text first:

```bash
grep -n "unsafe impl.*Send for ConstIter" armdb/src/const_tree.rs
```

Then delete the matched line and any SAFETY comment immediately above it.

- [ ] **Step 2.3: Remove `unsafe impl Send for TypedIter`**

Same procedure for `armdb/src/typed_tree.rs`:

```bash
grep -n "unsafe impl.*Send for TypedIter" armdb/src/typed_tree.rs
```

Delete the line and its SAFETY comment.

- [ ] **Step 2.4: Build**

Run: `cargo check -p armdb --features var-collections,typed-tree`
Expected: PASS. If any caller relied on iterator `Send`, the compile failure must be addressed in a follow-up plan — the spec accepts this as a breaking trait-bound change.

Run: `cargo nextest run -p armdb`
Expected: PASS — no runtime regression test required; the absence of the impl is the fix.

- [ ] **Step 2.5: Commit**

```bash
git add armdb/src/var_tree.rs armdb/src/const_tree.rs armdb/src/typed_tree.rs
git commit -m "$(cat <<'EOF'
fix(armdb): drop unsound Send impls on iterators with seize::LocalGuard

VarIter, ConstIter, and TypedIter each hold a seize::LocalGuard, which
is !Send by design — moving it across threads would allow reads of freed
nodes after the originating thread leaves the collector epoch. Remove
the unsafe Send impls; let the compiler enforce thread-locality.

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

---

## Task 3: Fix `VarTree::insert_locked` to honor `InsertResult::Exists`

**Files:**
- Modify: `armdb/src/var_tree.rs:360-377`
- Test: `armdb/tests/var_correctness_tests.rs` (new file, created in Task 3)

`VarTree::insert_locked` checks `self.index.get(...)` and returns `KeyExists` if found. Otherwise it appends an entry, allocates a `VarNode`, then calls `self.index.insert(node_ptr, guard)` and discards the result. Between the initial `get` and `insert`, another shard can race the same key in via `apply_entry` (replication) or future internal paths, leaving the allocated node dangling and the just-written entry uncounted as dead bytes.

The fix mirrors the `put_locked`/`Exists` slow path but returns `KeyExists` instead of overwriting. The rejected node has never been published, so `dealloc_node` is safe without the `disk.store(null)` trick used in `put_locked` (which is only there because that path *did* hand the `DiskLoc` to the live existing node).

- [ ] **Step 3.1: Create the regression test file**

Create new file `armdb/tests/var_correctness_tests.rs`:

```rust
//! Regression tests for the Var Collections Critical Correctness fix slice.
//! See armdb/docs/superpowers/specs/26-05-15-var-collections-critical-correctness.md.

#![cfg(feature = "var-collections")]

use armdb::{Config, DbError, VarMap, VarTree};
use tempfile::tempdir;

#[test]
fn var_tree_duplicate_insert_returns_key_exists_and_preserves_value() {
    let dir = tempdir().unwrap();
    let tree = VarTree::<[u8; 8]>::open(dir.path(), Config::test()).unwrap();

    let key = 1u64.to_be_bytes();
    tree.insert(&key, b"original").unwrap();

    let err = tree
        .insert(&key, b"replacement")
        .expect_err("duplicate insert must fail");
    assert!(matches!(err, DbError::KeyExists));

    let got = tree.get(&key).expect("original value must still exist");
    assert_eq!(got.as_bytes(), b"original");
}
```

- [ ] **Step 3.2: Run the test and confirm current behavior**

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree_duplicate_insert_returns_key_exists_and_preserves_value)'`
Expected: PASS on current code (the public `insert()` API already returns `KeyExists` via the early-return in `insert_locked`). This test exists to lock in user-visible behavior and serve as a regression anchor; the actual fix targets the post-`get` race path.

- [ ] **Step 3.3: Rewrite `VarTree::insert_locked`**

In `armdb/src/var_tree.rs`, replace the current body of `insert_locked`:

```rust
fn insert_locked(
    &self,
    shard_id: usize,
    inner: &mut ShardInner,
    guard: &seize::LocalGuard<'_>,
    key: &K,
    value: &[u8],
) -> DbResult<()> {
    if self.index.get(key.as_bytes(), guard).is_some() {
        return Err(DbError::KeyExists);
    }

    let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), value, false)?;
    let height = random_height();
    let node_ptr = VarNode::alloc(*key, disk_loc, height);
    self.index.insert(node_ptr, guard);
    Ok(())
}
```

with:

```rust
fn insert_locked(
    &self,
    shard_id: usize,
    inner: &mut ShardInner,
    guard: &seize::LocalGuard<'_>,
    key: &K,
    value: &[u8],
) -> DbResult<()> {
    if self.index.get(key.as_bytes(), guard).is_some() {
        return Err(DbError::KeyExists);
    }

    let (disk_loc, _gsn) = inner.append_entry(shard_id as u8, key.as_bytes(), value, false)?;
    let height = random_height();
    let node_ptr = VarNode::alloc(*key, disk_loc, height);

    match self.index.insert(node_ptr, guard) {
        InsertResult::Inserted => Ok(()),
        InsertResult::Exists(_existing) => {
            // Race: another path inserted this key after our `get` check and
            // before `index.insert`. Account for the entry we already appended
            // as dead bytes, free the unpublished node, and report KeyExists.
            inner.add_dead_bytes(
                disk_loc.file_id as u32,
                crate::entry::entry_size(size_of::<K>(), disk_loc.len),
            );
            // SAFETY: node_ptr was just allocated by VarNode::alloc and has
            // not been published into the SkipList. It uniquely owns its
            // freshly-allocated DiskLoc box.
            unsafe { VarNode::<K>::dealloc_node(node_ptr) };
            Err(DbError::KeyExists)
        }
    }
}
```

`InsertResult` is already imported at the top of the file (`use crate::skiplist::{InsertResult, SkipList};`).

- [ ] **Step 3.4: Run the regression test again**

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree_duplicate_insert_returns_key_exists_and_preserves_value)'`
Expected: PASS.

Run the broader VarTree test set to confirm no regression:

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var)'`
Expected: PASS.

- [ ] **Step 3.5: Commit**

```bash
git add armdb/src/var_tree.rs armdb/tests/var_correctness_tests.rs
git commit -m "$(cat <<'EOF'
fix(armdb/var_tree): handle InsertResult::Exists in insert_locked

VarTree::insert_locked relied on a get-then-insert pattern that ignored
the InsertResult of the SkipList insert. A concurrent insert via another
path (e.g. replication apply) could land between the two and leak the
freshly allocated VarNode and DiskLoc box, plus leave the just-written
entry uncounted toward compaction's dead-bytes total. Match on
InsertResult, free the rejected node, account for the wasted entry, and
return DbError::KeyExists.

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

---

## Task 4: Reject oversized variable entries with `DbError::Client`

**Files:**
- Modify: `armdb/src/shard.rs:723-775`
- Test: `armdb/tests/var_correctness_tests.rs` (append)

`ShardInner::append_entry` serializes the entry into a `Vec`, asks `write_buf.is_full(buf.len())` whether a flush is needed, flushes if so, then `append`s. If the serialized entry is larger than the buffer capacity, `is_full` reports true, the flush runs (turning a flushed-empty buffer into a flushed-empty buffer), and the subsequent `append` panics because `self.buf` is too small. The GSN counter has already been consumed by `fetch_add`.

The fix performs a strict bounds check at the very top of `append_entry`, before consuming a GSN or touching the write buffer. The same bounds check protects `append_raw_entry` so the replication path stays consistent.

- [ ] **Step 4.1: Add the regression tests**

Append to `armdb/tests/var_correctness_tests.rs`:

```rust
#[test]
fn var_tree_oversized_value_returns_client_error() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.write_buffer_size = 4096;
    let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();

    let key = 7u64.to_be_bytes();
    // write_buffer_size = 4096; a 4096-byte value plus header+key always
    // exceeds the buffer. Should be rejected without panic.
    let big_value = vec![0xAB; 4096];
    let err = tree
        .put(&key, &big_value)
        .expect_err("oversized value must be rejected");
    assert!(
        matches!(err, DbError::Client(msg) if msg.contains("write_buffer_size")),
        "expected DbError::Client describing write_buffer_size, got: {err:?}",
    );

    // The tree must still be usable for normal-sized writes after rejection.
    tree.put(&key, b"normal").unwrap();
    let got = tree.get(&key).expect("normal put must succeed");
    assert_eq!(got.as_bytes(), b"normal");
}

#[test]
fn var_map_oversized_value_returns_client_error() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.write_buffer_size = 4096;
    let map = VarMap::<[u8; 8]>::open(dir.path(), config).unwrap();

    let key = 11u64.to_be_bytes();
    let big_value = vec![0xCD; 4096];
    let err = map
        .put(&key, &big_value)
        .expect_err("oversized value must be rejected");
    assert!(
        matches!(err, DbError::Client(msg) if msg.contains("write_buffer_size")),
        "expected DbError::Client describing write_buffer_size, got: {err:?}",
    );

    map.put(&key, b"normal").unwrap();
    let got = map.get(&key).expect("normal put must succeed");
    assert_eq!(got.as_bytes(), b"normal");
}
```

- [ ] **Step 4.2: Run the tests and confirm they fail (or panic) today**

Run: `cargo nextest run -p armdb --features var-collections -E 'test(oversized_value)'`
Expected: FAIL or panic — current code does not guard the oversized case.

If the test process aborts with a panic instead of returning `Err`, nextest will report a test failure, which is the expected pre-fix state.

- [ ] **Step 4.3: Reject oversized entries in `append_entry`**

In `armdb/src/shard.rs`, modify `ShardInner::append_entry`. The current top of the function is:

```rust
pub fn append_entry(
    &mut self,
    shard_id: u8,
    key: &[u8],
    value: &[u8],
    tombstone: bool,
) -> DbResult<(DiskLoc, u64)> {
    let gsn = self.gsn.fetch_add(1, Ordering::Relaxed);

    let buf = serialize_entry(gsn, key, value, tombstone);

    // Flush if buffer can't fit this entry
    if self.write_buf.is_full(buf.len()) {
        self.flush_write_buf()?;
    }
```

Replace it with:

```rust
pub fn append_entry(
    &mut self,
    shard_id: u8,
    key: &[u8],
    value: &[u8],
    tombstone: bool,
) -> DbResult<(DiskLoc, u64)> {
    let needed = crate::entry::entry_size(key.len(), value.len() as u32) as usize;
    if needed > self.write_buf.buf.len() {
        return Err(DbError::Client("entry exceeds write_buffer_size"));
    }

    let gsn = self.gsn.fetch_add(1, Ordering::Relaxed);

    let buf = serialize_entry(gsn, key, value, tombstone);

    // Flush if buffer can't fit this entry
    if self.write_buf.is_full(buf.len()) {
        self.flush_write_buf()?;
    }
```

`DbError` is already in scope in this module (used elsewhere in the file). `entry::entry_size` is the same helper used downstream for dead-byte accounting and returns a `u64` aligned to the same scheme as `serialize_entry`'s output length.

Confirm `WriteBuffer::buf` is reachable from `ShardInner::append_entry`. Looking at the struct, `WriteBuffer` is a sibling type in `shard.rs` with a `pub(crate) buf: AlignedBuf` (or similar). If `buf` is private to `WriteBuffer`, add a helper method on `WriteBuffer`:

```rust
#[inline]
fn capacity(&self) -> usize {
    self.buf.len()
}
```

and use `self.write_buf.capacity()` in the check. Choose whichever matches existing visibility; the simplest path is to add the `capacity` accessor.

- [ ] **Step 4.4: Apply the same guard to `append_raw_entry`**

Locate the `#[cfg(feature = "replication")] pub fn append_raw_entry` near line 893. At the top of the function, before the existing `if self.write_buf.is_full(data.len())`, add:

```rust
if data.len() > self.write_buf.capacity() {
    return Err(DbError::Client("replicated entry exceeds write_buffer_size"));
}
```

(If you went with the direct `self.write_buf.buf.len()` style in Step 4.3, mirror that here.)

- [ ] **Step 4.5: Re-run the regression tests and broader var suite**

Run: `cargo nextest run -p armdb --features var-collections -E 'test(oversized_value)'`
Expected: PASS.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var)'`
Expected: PASS — existing var tests use the default 1 MiB buffer and small values, so they remain well below the new threshold.

- [ ] **Step 4.6: Commit**

```bash
git add armdb/src/shard.rs armdb/tests/var_correctness_tests.rs
git commit -m "$(cat <<'EOF'
fix(armdb/shard): reject oversized entries with DbError::Client

ShardInner::append_entry previously consumed a GSN and only checked the
write buffer via is_full, which silently allowed a single entry larger
than write_buffer_size to flow through to write_buf.append and panic.
Add an upfront bounds check against write_buf.capacity() in both
append_entry and append_raw_entry; on overflow return
DbError::Client("entry exceeds write_buffer_size") and leave on-disk and
in-memory state unchanged.

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

---

## Task 5: Route `read_value_locked` through an encryption-aware helper

**Files:**
- Modify: `armdb/src/shard.rs` (add `ShardInner::read_value_from_disk_locked`)
- Modify: `armdb/src/var_tree.rs:934-970`
- Modify: `armdb/src/var_map.rs:599-635`
- Test: `armdb/tests/var_correctness_tests.rs` (append)

Both `VarTree::read_value_locked` and `VarMap::read_value_locked` currently fall through to `direct::pread_value` for any data that has flushed to disk. `direct::pread_value` is encryption-unaware: when `feature = "encryption"` is enabled and the data file is encrypted, the bytes it returns are ciphertext. CAS, `update`, `fetch_update`, and `delete`'s old-value hook then compare against ciphertext, producing silent mismatches.

The fix adds a `pub(crate)` helper on `ShardInner` that selects the active or immutable file from already-locked state and dispatches to `pread_value_encrypted` (when a cipher is configured) or `pread_value` otherwise. The helper must NOT re-acquire the shard mutex — `VarTree`/`VarMap` callers already hold it. Both callers replace their final disk-read branch with a call to the helper; the write-buffer and block-cache branches stay as-is.

- [ ] **Step 5.1: Add the encrypted-read regression test**

Append to `armdb/tests/var_correctness_tests.rs`:

```rust
#[cfg(feature = "encryption")]
#[test]
fn var_tree_locked_read_returns_plaintext_after_flush() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    // Force everything to flush quickly so the locked read hits disk.
    config.write_buffer_size = 4096;
    config.encryption_key = Some([0x42; 32]);
    let tree = VarTree::<[u8; 8]>::open(dir.path(), config).unwrap();

    let key = 99u64.to_be_bytes();
    // Value designed to cross a 4096-byte block boundary so read_value_locked
    // bypasses the single-block cache fast path and exercises the disk read.
    let value = vec![0xEFu8; 4096];

    tree.put(&key, &value).unwrap();

    // Trigger a flush by writing another sizeable entry.
    let filler_key = 100u64.to_be_bytes();
    tree.put(&filler_key, &vec![0u8; 3000]).unwrap();

    // cas() goes through read_value_locked under the shard lock. If the
    // path returns ciphertext, the expected comparison fails and cas
    // returns CasMismatch instead of swapping.
    let new_value = b"plain-new".to_vec();
    tree.cas(&key, &value, &new_value)
        .expect("cas with plaintext expected value must succeed");

    let got = tree.get(&key).expect("key must still exist");
    assert_eq!(got.as_bytes(), new_value.as_slice());
}

#[cfg(feature = "encryption")]
#[test]
fn var_map_locked_read_returns_plaintext_after_flush() {
    let dir = tempdir().unwrap();
    let mut config = Config::test();
    config.write_buffer_size = 4096;
    config.encryption_key = Some([0x77; 32]);
    let map = VarMap::<[u8; 8]>::open(dir.path(), config).unwrap();

    let key = 33u64.to_be_bytes();
    let value = vec![0xABu8; 4096];

    map.put(&key, &value).unwrap();
    map.put(&44u64.to_be_bytes(), &vec![0u8; 3000]).unwrap();

    let new_value = b"plain-new".to_vec();
    map.cas(&key, &value, &new_value)
        .expect("cas with plaintext expected value must succeed");

    let got = map.get(&key).expect("key must still exist");
    assert_eq!(got.as_bytes(), new_value.as_slice());
}
```

Confirm `tree.cas` and `map.cas` exist with this signature. If the project's CAS method is named differently (e.g. `compare_and_swap` or `update`), use whichever method's documented behavior is "read current under lock, compare with expected, swap if equal" — the goal is exercising `read_value_locked`. Inspect `var_tree.rs` / `var_map.rs` for the exact public method (search for `read_value_locked` callers) and adjust the test calls if needed.

- [ ] **Step 5.2: Run the tests and confirm they fail under encryption**

Run: `cargo nextest run -p armdb --features 'var-collections encryption' -E 'test(locked_read_returns_plaintext_after_flush)'`
Expected: FAIL — current code feeds ciphertext to CAS, so the comparison mismatches.

- [ ] **Step 5.3: Add `ShardInner::read_value_from_disk_locked`**

In `armdb/src/shard.rs`, add a new method on `impl ShardInner` (place it near the other read-side helpers). The method receives the inner state by `&self`, picks the right file handle from `active` or `immutable`, and dispatches to encrypted or plain reads:

```rust
#[cfg(feature = "var-collections")]
impl ShardInner {
    /// Read a value from disk while the shard mutex is already held.
    ///
    /// Returns the plaintext bytes when encryption is enabled. The caller is
    /// responsible for checking the write buffer and block cache first; this
    /// helper only inspects the active and immutable files.
    pub(crate) fn read_value_from_disk_locked(&self, loc: &DiskLoc) -> DbResult<Vec<u8>> {
        let len = loc.len as usize;
        let target_file_id = loc.file_id as u32;

        if self.active.file_id == target_file_id {
            #[cfg(feature = "encryption")]
            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,
                    cipher,
                    target_file_id,
                    loc.offset as u64,
                    len,
                );
            }
            return crate::io::direct::pread_value(
                &self.active.read_file,
                loc.offset as u64,
                len,
            );
        }

        let imm = self
            .immutable
            .iter()
            .find(|f| f.file_id == target_file_id)
            .ok_or_else(|| {
                DbError::Io(std::io::Error::new(
                    std::io::ErrorKind::NotFound,
                    "data file not found",
                ))
            })?;

        #[cfg(feature = "encryption")]
        if let (Some(cipher), Some(tag_file)) = (&self.cipher, &imm.tag_file) {
            return crate::io::direct::pread_value_encrypted(
                &imm.file,
                tag_file,
                cipher,
                target_file_id,
                loc.offset as u64,
                len,
            );
        }
        crate::io::direct::pread_value(&imm.file, loc.offset as u64, len)
    }
}
```

`DiskLoc`, `DbError`, and the `direct` module are already in scope in this file. Keep the `#[cfg(feature = "var-collections")]` gate — the helper is only used by `VarTree`/`VarMap` and avoids adding code under feature combinations that don't enable it.

- [ ] **Step 5.4: Route `VarTree::read_value_locked` through the helper**

In `armdb/src/var_tree.rs`, the current disk branch is:

```rust
// 3. Disk read (file handle from inner, no re-lock)
let file: &std::fs::File = if inner.active.file_id == disk.file_id as u32 {
    &inner.active.read_file
} else {
    let imm = inner
        .immutable
        .iter()
        .find(|f| f.file_id == disk.file_id as u32)?;
    &imm.file
};
let bytes = direct::pread_value(file, disk.offset as u64, len).ok()?;
Some(ByteView::new(&bytes))
```

Replace it with:

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

Remove the now-unused `file` local. The unused `len` binding above (it is still passed through `disk.len`) stays as written — re-check the surrounding code after the edit to make sure no helper variables become dead.

- [ ] **Step 5.5: Route `VarMap::read_value_locked` through the helper**

In `armdb/src/var_map.rs`, the disk branch is structurally identical (around line 623). Apply the same replacement:

```rust
let bytes = inner.read_value_from_disk_locked(disk).ok()?;
Some(ByteView::new(&bytes))
```

- [ ] **Step 5.6: Build with encryption enabled**

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

Run: `cargo check -p armdb --features var-collections`
Expected: PASS — the helper's non-encrypted branch is the only path active.

- [ ] **Step 5.7: Run the regression tests**

Run: `cargo nextest run -p armdb --features 'var-collections encryption' -E 'test(locked_read_returns_plaintext_after_flush)'`
Expected: PASS.

Run the broader encryption and var suites:

Run: `cargo nextest run -p armdb --features 'var-collections encryption' -E 'test(encryption) or test(var)'`
Expected: PASS.

- [ ] **Step 5.8: Commit**

```bash
git add armdb/src/shard.rs armdb/src/var_tree.rs armdb/src/var_map.rs armdb/tests/var_correctness_tests.rs
git commit -m "$(cat <<'EOF'
fix(armdb/var_*): decrypt values in locked reads

VarTree::read_value_locked and VarMap::read_value_locked fell through
to direct::pread_value for any flushed data, returning ciphertext when
the data file is encrypted. CAS/update/fetch_update then compared
against ciphertext and silently mismatched. Add
ShardInner::read_value_from_disk_locked, which dispatches to
pread_value_encrypted when a cipher is configured and reuses the
already-locked active/immutable state without taking the shard mutex
again. Route both callers through it.

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

---

## Task 6: Full-workspace verification

**Files:**
- (none — verification only)

- [ ] **Step 6.1: Format**

Run: `cargo fmt`
Expected: no diff, or only whitespace-level changes; commit if any.

- [ ] **Step 6.2: Lint**

Run: `cargo clippy -p armdb --features 'var-collections typed-tree encryption' -- -D warnings`
Expected: PASS. Address any new clippy findings introduced by the changes; do not touch unrelated warnings (per `CLAUDE.md` rules).

- [ ] **Step 6.3: Full armdb test run**

Run: `cargo nextest run -p armdb --features 'var-collections typed-tree'`
Expected: PASS.

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

- [ ] **Step 6.4: Optional — workspace-wide check**

Run: `cargo check`
Expected: PASS. The removed iterator `Send` impls are a breaking trait-bound change; if a workspace consumer fails to compile, the fix scope (`armdb`) is intact but the consumer needs a follow-up. Surface the failure to the user rather than reverting.

---

## Follow-Up Work (out of scope, mirrored from the spec)

- `migrate` / hook semantics for `VarTree` and `VarMap`.
- `DiskLoc.file_id` overflow and file-format decisions.
- Compaction behavior for readers holding stale copied `DiskLoc` values.
- Direct-write or multi-block cache support for values larger than `write_buffer_size`.
- Replication `VarTree::apply_entry` revision.
