# FixedStore Replication 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:** Add replication to the FixedStore backend of armdb using SPSC streaming + per-slot version resync (modular 30-bit version comparison).

**Architecture:** New module `armdb/src/fixed_replication/` mirrors the existing `armdb/src/replication/` (Bitcask) but is isolated — its own port, wire protocol, and event pipeline. Per-slot monotonic version (packed into 4-byte `meta` alongside 2-bit status) replaces GSN. Catch-up is a stateless full slot scan on each (re)connect; steady state streams events via per-shard rtrb ring buffers.

**Tech Stack:** Rust 2024, feature-gated under `replication` (already enables `rtrb`). No new crate deps. Integration with existing `FixedEngine`, `ConstTree<K, V, H, Fixed>`, `ConstMap<K, V, H, Fixed>`.

**Design spec:** `armdb/docs/superpowers/specs/2026-04-17-fixed-replication-design.md` (commit `b715160`). Tasks below reference spec section numbers (§N) for semantic details.

**Testing tool:** Use `cargo nextest run -p armdb` (not `cargo test`) per `CLAUDE.md`. Filter by test name expr, e.g. `cargo nextest run -p armdb -E 'test(fixed)'`.

---

## File structure

| Path | Responsibility |
|---|---|
| `armdb/src/fixed/slot.rs` | Slot format v2: `meta: u32` (2-bit status + 30-bit version), CRC over `(meta+key+value)`, `read_slot` returns `None` for DELETED. |
| `armdb/src/fixed/shard.rs` | `FixedShardInner`: adds `versions: Vec<u32>`, `replication_tx: Option<Producer>`, `apply_foreign_*`, `read_slot_header_and_key`, partial-4B delete, wrap warnings. Sidecar read/write moves from `fixed.bitmap` to `fixed.versions` with CRC trailer. |
| `armdb/src/fixed/engine.rs` | db.meta `VERSION` 1→2, `install_replication_producers`, implements `FixedEngineAccess`. |
| `armdb/src/fixed/bitmap.rs` | `from_versions(&[u32])` helper to derive bitmap from versions array. |
| `armdb/src/fixed/mod.rs` | Re-export `FixedReplicationTarget` (conditional on feature). |
| `armdb/src/fixed/config.rs` | No change. |
| `armdb/src/durability.rs` | `Fixed::recover_entries` populates `versions`; handles torn slot (CRC mismatch → FREE+preserve version). Writes sidecar on `close()`. |
| `armdb/src/lib.rs` | `#[cfg(feature = "replication")] pub mod fixed_replication;`. |
| `armdb/src/fixed_replication/mod.rs` | Public exports for server/client/target/event. |
| `armdb/src/fixed_replication/event.rs` | `FixedReplicationEvent` enum (SPSC payload). |
| `armdb/src/fixed_replication/engine_access.rs` | `FixedEngineAccess` trait + impl for `FixedEngine`. |
| `armdb/src/fixed_replication/protocol.rs` | `FixedMessageType`, `Frame`, message encode/decode. Flags: `FLAG_EMPTY_STATE`. |
| `armdb/src/fixed_replication/cursor.rs` | `FixedReplicationCursor` — diagnostic only (applied_total, max_version_seen, last_ack_at). |
| `armdb/src/fixed_replication/apply.rs` | `FixedReplicationTarget` trait + `ApplyOutcome` enum + blanket impls for `ConstTree<Fixed>` / `ConstMap<Fixed>`. |
| `armdb/src/fixed_replication/server.rs` | `FixedReplicationServer`: acceptor (lazy producer install), per-connection handler, full scan phase 1, SPSC streaming phase 2. |
| `armdb/src/fixed_replication/client.rs` | `FixedReplicationClient`: per-shard reconnect threads, handshake with `FLAG_EMPTY_STATE`, apply loop. |
| `armdb/tests/fixed_replication.rs` | End-to-end integration tests. |
| `armdb/src/skiplist/node.rs` | Verify `ConstNode` has required accessors (probably no change). |
| `armdb/src/const_tree.rs` / `const_map.rs` | Possibly add `pub(crate)` helpers for replication-initiated index updates. |

---

## Phase 1: Slot format v2

### Task 1: Add `meta`/`status`/`version` helpers to `fixed/slot.rs`

**Files:**
- Modify: `armdb/src/fixed/slot.rs`
- Test: same file (unit tests inline)

- [ ] **Step 1.1: Write failing tests for pack/unpack helpers**

Add at the bottom of `armdb/src/fixed/slot.rs` (inside `mod tests`):

```rust
    #[test]
    fn test_pack_meta_roundtrip() {
        let m = pack_meta(STATUS_OCCUPIED, 0x12345);
        assert_eq!(status_of(m), STATUS_OCCUPIED);
        assert_eq!(version_of(m), 0x12345);
    }

    #[test]
    fn test_pack_meta_max_version() {
        let max_v = (1u32 << 30) - 1;
        let m = pack_meta(STATUS_DELETED, max_v);
        assert_eq!(status_of(m), STATUS_DELETED);
        assert_eq!(version_of(m), max_v);
    }

    #[test]
    fn test_bump_version_preserves_status() {
        let m = pack_meta(STATUS_OCCUPIED, 5);
        let b = bump_version(m);
        assert_eq!(status_of(b), STATUS_OCCUPIED);
        assert_eq!(version_of(b), 6);
    }

    #[test]
    fn test_bump_version_wraps_at_2_pow_30() {
        let m = pack_meta(STATUS_OCCUPIED, (1 << 30) - 1);
        let b = bump_version(m);
        assert_eq!(version_of(b), 0, "version should wrap from 2^30-1 to 0");
        assert_eq!(status_of(b), STATUS_OCCUPIED);
    }

    #[test]
    fn test_with_status_preserves_version() {
        let m = pack_meta(STATUS_OCCUPIED, 42);
        let changed = with_status(m, STATUS_DELETED);
        assert_eq!(status_of(changed), STATUS_DELETED);
        assert_eq!(version_of(changed), 42);
    }
```

- [ ] **Step 1.2: Run tests; they should fail (functions not defined)**

```sh
cargo nextest run -p armdb -E 'test(slot::tests::test_pack_meta_roundtrip)'
```

Expected: compile error — unresolved names `pack_meta`, `status_of`, etc.

- [ ] **Step 1.3: Add constants and helpers above existing `mod tests`**

Replace the top of `armdb/src/fixed/slot.rs` (existing status constants stay as values, but we may adjust them):

```rust
/// Slot status: the slot has never been written (packed in low 2 bits of meta).
pub const STATUS_FREE: u8 = 0;
/// Slot status: the slot contains a valid key-value pair.
pub const STATUS_OCCUPIED: u8 = 1;
/// Slot status: the slot was previously occupied but has been deleted.
pub const STATUS_DELETED: u8 = 2;

const STATUS_MASK: u32 = 0b11;
const VERSION_SHIFT: u32 = 2;
const VERSION_MASK: u32 = 0xFFFF_FFFC;

/// Warning threshold — at ~97% of 2^30 we log/metric that a hot slot is
/// approaching version wrap.
pub const VERSION_WARN_THRESHOLD: u32 = 0x3E00_0000;

/// Maximum 30-bit version value (2^30 - 1).
pub const VERSION_MAX: u32 = (1 << 30) - 1;

/// Size of the slot header: [meta: 4] [crc32: 4] = 8 bytes.
pub const SLOT_HEADER_SIZE: usize = 8;

#[inline]
pub fn pack_meta(status: u8, version: u32) -> u32 {
    (version << VERSION_SHIFT) | (status as u32 & STATUS_MASK)
}

#[inline]
pub fn status_of(meta: u32) -> u8 {
    (meta & STATUS_MASK) as u8
}

#[inline]
pub fn version_of(meta: u32) -> u32 {
    meta >> VERSION_SHIFT
}

/// Increment `version` by 1, preserving `status`. Wraps at 2^30.
#[inline]
pub fn bump_version(meta: u32) -> u32 {
    // Adding `1 << VERSION_SHIFT` = 4 increments version bits and leaves
    // status bits intact. wrapping_add handles 2^30 wrap.
    meta.wrapping_add(1 << VERSION_SHIFT) & (VERSION_MASK | STATUS_MASK)
}

/// Replace status, preserving version.
#[inline]
pub fn with_status(meta: u32, new_status: u8) -> u32 {
    (meta & VERSION_MASK) | (new_status as u32 & STATUS_MASK)
}
```

Note: the existing `SLOT_HEADER_SIZE` declaration should be removed (it's reintroduced above).

- [ ] **Step 1.4: Run tests; they should pass**

```sh
cargo nextest run -p armdb -E 'test(fixed::slot::tests)'
```

Expected: all new pack/bump/with_status tests PASS.

- [ ] **Step 1.5: Commit**

```sh
git add armdb/src/fixed/slot.rs
git commit -m "feat(armdb/fixed): add meta pack/unpack helpers for slot v2"
```

---

### Task 2: Add `is_newer` modular comparison to `fixed/slot.rs`

**Files:**
- Modify: `armdb/src/fixed/slot.rs`

- [ ] **Step 2.1: Write failing tests**

Append to the `mod tests` block:

```rust
    #[test]
    fn test_is_newer_basic() {
        assert!(is_newer(5, 3));
        assert!(!is_newer(3, 5));
        assert!(!is_newer(7, 7), "equal is not newer");
    }

    #[test]
    fn test_is_newer_wrap_near_max() {
        // new=0 with old=2^30-1: diff = 1, newer
        assert!(is_newer(0, (1 << 30) - 1));
        // new=5 with old=2^30-3: diff = 8 (after wrap), newer
        assert!(is_newer(5, (1 << 30) - 3));
    }

    #[test]
    fn test_is_newer_boundary() {
        // diff exactly at 2^29 is ambiguous → NOT newer (window is strictly <)
        assert!(!is_newer(1 << 29, 0));
        // diff one below boundary → newer
        assert!(is_newer((1 << 29) - 1, 0));
    }

    #[test]
    fn test_is_newer_too_far() {
        // diff > 2^29 → treat as older (went backwards in ring)
        assert!(!is_newer(0, (1 << 29) + 5));
    }
```

- [ ] **Step 2.2: Run tests; expect failure (function missing)**

```sh
cargo nextest run -p armdb -E 'test(is_newer)'
```

- [ ] **Step 2.3: Add `is_newer` to `fixed/slot.rs`**

Append after `with_status`:

```rust
/// Modular comparison: is `new_version` newer than `old_version` in the
/// 30-bit version ring? Tolerates lag up to 2^29 versions — beyond that,
/// comparison is ambiguous and the event is treated as stale.
#[inline]
pub fn is_newer(new_version: u32, old_version: u32) -> bool {
    let diff = new_version.wrapping_sub(old_version) & 0x3FFF_FFFF;
    diff != 0 && diff < (1 << 29)
}
```

- [ ] **Step 2.4: Run tests; should pass**

```sh
cargo nextest run -p armdb -E 'test(is_newer)'
```

- [ ] **Step 2.5: Commit**

```sh
git add armdb/src/fixed/slot.rs
git commit -m "feat(armdb/fixed): add is_newer modular comparison (30-bit ring, 2^29 tolerance)"
```

---

### Task 3: Update `compute_crc` to include `meta`; update `serialize_slot`/`read_slot`

**Files:**
- Modify: `armdb/src/fixed/slot.rs`

Per spec §3: CRC now covers `meta + key + value`; DELETED slots return `None` from `read_slot` without checking CRC (their CRC is stale).

- [ ] **Step 3.1: Write failing tests**

Append to `mod tests`:

```rust
    #[test]
    fn test_crc_covers_meta() {
        let m1 = pack_meta(STATUS_OCCUPIED, 1);
        let m2 = pack_meta(STATUS_OCCUPIED, 2);
        let c1 = compute_crc(m1, b"k", b"v");
        let c2 = compute_crc(m2, b"k", b"v");
        assert_ne!(c1, c2, "different meta must produce different CRC");
    }

    #[test]
    fn test_serialize_read_roundtrip_with_meta() {
        let key = b"hello";
        let value = b"world!!!";
        let meta = pack_meta(STATUS_OCCUPIED, 123);
        let size = slot_size(key.len(), value.len());
        let mut buf = vec![0u8; size];
        serialize_slot(&mut buf, meta, key, value);
        let (m, k, v) = read_slot(&buf, key.len(), value.len())
            .expect("valid slot should read back");
        assert_eq!(m, meta);
        assert_eq!(k, key);
        assert_eq!(v, value);
    }

    #[test]
    fn test_read_slot_returns_none_for_deleted() {
        let size = slot_size(4, 4);
        let mut buf = vec![0u8; size];
        // Write OCCUPIED first
        serialize_slot(&mut buf, pack_meta(STATUS_OCCUPIED, 1), b"abcd", b"1234");
        // Change status to DELETED (first 4 bytes = meta)
        let deleted_meta = pack_meta(STATUS_DELETED, 2);
        buf[0..4].copy_from_slice(&deleted_meta.to_le_bytes());
        // Read: CRC no longer matches (we changed meta without recomputing),
        // but we should skip CRC check because status == DELETED and return None.
        assert!(read_slot(&buf, 4, 4).is_none());
    }

    #[test]
    fn test_read_slot_returns_none_for_free() {
        let size = slot_size(4, 4);
        let buf = vec![0u8; size]; // all zeros = status FREE
        assert!(read_slot(&buf, 4, 4).is_none());
    }
```

- [ ] **Step 3.2: Run tests; expect failures (signatures changed)**

```sh
cargo nextest run -p armdb -E 'test(fixed::slot::tests)'
```

- [ ] **Step 3.3: Update `compute_crc`, `serialize_slot`, replace `validate_slot` with `read_slot`**

In `armdb/src/fixed/slot.rs`, replace the existing `compute_crc`, `serialize_slot`, `serialize_deleted_slot`, `slot_status`, `validate_slot` with:

```rust
/// Compute CRC32 over (meta bytes + key + value). Used for OCCUPIED slots;
/// DELETED slots store a stale CRC that is intentionally ignored at read.
pub fn compute_crc(meta: u32, key: &[u8], value: &[u8]) -> u32 {
    let mut h = crc32fast::Hasher::new();
    h.update(&meta.to_le_bytes());
    h.update(key);
    h.update(value);
    h.finalize()
}

/// Serialize an occupied slot into `buf`.
///
/// Layout: `[meta LE u32] [crc32 LE u32] [key] [value] [0-pad to align]`.
///
/// # Panics
/// If `buf.len() < slot_size(key.len(), value.len())`.
pub fn serialize_slot(buf: &mut [u8], meta: u32, key: &[u8], value: &[u8]) {
    let size = slot_size(key.len(), value.len());
    assert!(buf.len() >= size, "buffer too small: {} < {}", buf.len(), size);
    buf[..size].fill(0);
    buf[0..4].copy_from_slice(&meta.to_le_bytes());
    let crc = compute_crc(meta, key, value);
    buf[4..8].copy_from_slice(&crc.to_le_bytes());
    buf[SLOT_HEADER_SIZE..SLOT_HEADER_SIZE + key.len()].copy_from_slice(key);
    let val_off = SLOT_HEADER_SIZE + key.len();
    buf[val_off..val_off + value.len()].copy_from_slice(value);
}

/// Read the raw meta bytes from a slot buffer (first 4 bytes LE).
/// Useful for decisions that don't need key/value and cannot assume status.
#[inline]
pub fn meta_of(buf: &[u8]) -> u32 {
    u32::from_le_bytes(buf[0..4].try_into().expect("slot buffer >= 4 bytes"))
}

/// Read a slot and validate it. Returns `Some((meta, key, value))` only
/// for OCCUPIED slots with a matching CRC. Returns `None` for FREE,
/// DELETED, or CRC-mismatched slots.
pub fn read_slot(buf: &[u8], key_len: usize, value_len: usize)
    -> Option<(u32, &[u8], &[u8])>
{
    let size = slot_size(key_len, value_len);
    if buf.len() < size {
        return None;
    }
    let meta = meta_of(buf);
    if status_of(meta) != STATUS_OCCUPIED {
        return None;
    }
    let stored_crc = u32::from_le_bytes(buf[4..8].try_into().ok()?);
    let key = &buf[SLOT_HEADER_SIZE..SLOT_HEADER_SIZE + key_len];
    let val_off = SLOT_HEADER_SIZE + key_len;
    let value = &buf[val_off..val_off + value_len];
    if stored_crc != compute_crc(meta, key, value) {
        return None;
    }
    Some((meta, key, value))
}
```

Delete the old `serialize_deleted_slot`, `slot_status`, `validate_slot` fns (they are replaced).

- [ ] **Step 3.4: Update any callers in existing tests of `slot.rs`**

Look at the existing `mod tests` block: `test_serialize_validate_roundtrip`, `test_corrupted_slot_fails_validation`, `test_free_slot_not_valid`, `test_deleted_slot`. Rewrite them to use the new API:

```rust
    #[test]
    fn test_serialize_read_roundtrip() {
        let key = b"hello";
        let value = b"world!!!";
        let size = slot_size(key.len(), value.len());
        let mut buf = vec![0u8; size];
        let meta = pack_meta(STATUS_OCCUPIED, 1);
        serialize_slot(&mut buf, meta, key, value);
        let (m, k, v) = read_slot(&buf, key.len(), value.len())
            .expect("validation should pass");
        assert_eq!(m, meta);
        assert_eq!(k, key);
        assert_eq!(v, value);
    }

    #[test]
    fn test_corrupted_slot_fails_validation() {
        let key = b"abc";
        let value = b"defgh";
        let size = slot_size(key.len(), value.len());
        let mut buf = vec![0u8; size];
        serialize_slot(&mut buf, pack_meta(STATUS_OCCUPIED, 1), key, value);
        buf[SLOT_HEADER_SIZE + key.len()] ^= 0xFF;
        assert!(read_slot(&buf, key.len(), value.len()).is_none());
    }
```

Delete the old free/deleted tests that relied on `slot_status` (they overlap with the new `test_read_slot_returns_none_*` tests).

- [ ] **Step 3.5: Run tests; all pass**

```sh
cargo nextest run -p armdb -E 'test(fixed::slot)'
```

- [ ] **Step 3.6: Commit**

```sh
git add armdb/src/fixed/slot.rs
git commit -m "refactor(armdb/fixed): slot v2 — meta:u32, CRC over (meta+key+value), DELETED skips CRC"
```

---

## Phase 2: Shard-level state and sidecar

### Task 4: Add `versions: Vec<u32>` field to `FixedShardInner`; update `open`

**Files:**
- Modify: `armdb/src/fixed/shard.rs`

Per spec §4.

- [ ] **Step 4.1: Add the field**

Edit `FixedShardInner` struct in `armdb/src/fixed/shard.rs`:

```rust
pub struct FixedShardInner {
    file: File,
    dir: PathBuf,
    pub(crate) bitmap: Bitmap,
    /// Per-slot `meta` cache (4 bytes × slot_count).  Populated on open;
    /// kept in sync with disk on every write/delete.  Source of truth for
    /// replication modular version comparisons.
    pub(crate) versions: Vec<u32>,
    pub(crate) slot_size: u16,
    pub(crate) slot_count: u32,
    key_len: u16,
    value_len: u16,
    pub(crate) shard_id: u8,
    grow_step: u32,
    pending_writes: u32,
    sync_batch_size: u32,
    last_sync: Instant,
    sync_interval: Duration,
    enable_fsync: bool,
}
```

Note: `shard_id` is promoted to `pub(crate)` (was private) — needed for wrap-warning metric label.

- [ ] **Step 4.2: Initialize `versions` in both branches of `FixedShardInner::open`**

In the `exists` branch, after computing `stored_slot_count`, add:

```rust
            let bitmap = Bitmap::new(stored_slot_count);
            let versions = vec![0u32; stored_slot_count as usize];

            Ok(Self {
                file,
                dir,
                bitmap,
                versions,
                // ... rest of fields (slot_size, slot_count, etc.) as before
                // ensure `slot_count: stored_slot_count,` passes through.
```

In the `else` branch (new file), after `let bitmap = Bitmap::new(initial_slots);`:

```rust
            let versions = vec![0u32; initial_slots as usize];

            Ok(Self {
                file,
                dir,
                bitmap,
                versions,
                // ... other fields
```

Full struct literal must include `versions`; compiler will flag any missing field.

- [ ] **Step 4.3: Update `grow` to extend `versions` vec**

In `FixedShardInner::grow`:

```rust
    pub fn grow(&mut self) -> DbResult<()> {
        let new_count = self
            .slot_count
            .checked_add(self.grow_step)
            .ok_or(DbError::Internal("slot_count overflow on grow"))?;
        let new_size = HEADER_SIZE + new_count as u64 * self.slot_size as u64;

        self.file.set_len(new_size)?;
        self.bitmap.grow(new_count);
        self.versions.resize(new_count as usize, 0);   // ← new line
        self.slot_count = new_count;

        self.file.write_all_at(&new_count.to_le_bytes(), 8)?;
        self.file.sync_data()?;
        Ok(())
    }
```

- [ ] **Step 4.4: Run existing shard tests; they should still compile and pass**

```sh
cargo nextest run -p armdb -E 'test(fixed::shard)'
```

Existing tests (e.g. `test_create_and_reopen`) use `write_slot` which we're about to change — they may fail in a future task but must pass now. If they fail now, the missing field was the cause.

- [ ] **Step 4.5: Commit**

```sh
git add armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed): add versions cache field to FixedShardInner"
```

---

### Task 5: Rewrite `write_slot` and `delete_slot` with version bump

**Files:**
- Modify: `armdb/src/fixed/shard.rs`

Per spec §3 (delete = 4-byte pwrite) and §5 (leader hook).

- [ ] **Step 5.1: Write failing tests**

Add to the `mod tests` block of `armdb/src/fixed/shard.rs`:

```rust
    #[test]
    fn test_write_slot_bumps_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        let m1 = shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();
        assert_eq!(slot::status_of(m1), slot::STATUS_OCCUPIED);
        assert_eq!(slot::version_of(m1), 1, "first write → version 1");
        let m2 = shard.write_slot(id, b"key_0001", b"value___00000002").unwrap();
        assert_eq!(slot::version_of(m2), 2, "second write → version 2");
        assert_eq!(shard.versions[id as usize], m2);
    }

    #[test]
    fn test_delete_slot_4byte_partial_write() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();

        let m = shard.delete_slot(id, b"key_0001").unwrap();
        assert_eq!(slot::status_of(m), slot::STATUS_DELETED);
        assert_eq!(slot::version_of(m), 2);

        // key bytes on disk should still be the original 'key_0001' — only
        // first 4 bytes (meta) were overwritten.
        let buf = shard.read_slot(id).unwrap();
        assert_eq!(&buf[slot::SLOT_HEADER_SIZE..slot::SLOT_HEADER_SIZE + 8], b"key_0001");
        assert_eq!(slot::meta_of(&buf), m);
        // read_slot helper must return None for DELETED.
        assert!(slot::read_slot(&buf, 8, 16).is_none());
    }

    #[test]
    fn test_delete_then_write_continues_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard.write_slot(id, b"key_0001", b"val1_0000_0000_00").unwrap();
        shard.delete_slot(id, b"key_0001").unwrap();
        let m3 = shard.write_slot(id, b"key_0002", b"val2_0000_0000_00").unwrap();
        assert_eq!(slot::version_of(m3), 3, "version continues across DELETE");
    }
```

- [ ] **Step 5.2: Run tests; expect compile errors**

```sh
cargo nextest run -p armdb -E 'test(test_write_slot_bumps_version)'
```

- [ ] **Step 5.3: Rewrite `write_slot` and `delete_slot`**

Replace existing `write_slot` and `delete_slot` bodies in `armdb/src/fixed/shard.rs`:

```rust
    /// Write an OCCUPIED slot: bumps version from `self.versions[slot_id]`,
    /// serializes full slot, pwrite, updates cache. Returns the new `meta`.
    pub fn write_slot(&mut self, slot_id: u32, key: &[u8], value: &[u8]) -> DbResult<u32> {
        let old_meta = self.versions[slot_id as usize];
        let new_meta = slot::pack_meta(
            slot::STATUS_OCCUPIED,
            slot::version_of(old_meta).wrapping_add(1) & ((1 << 30) - 1),
        );

        let size = self.slot_size as usize;
        let mut buf = vec![0u8; size];
        slot::serialize_slot(&mut buf, new_meta, key, value);

        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&buf, offset)?;
        self.versions[slot_id as usize] = new_meta;
        self.maybe_warn_wrap(slot_id, slot::version_of(new_meta));

        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        Ok(new_meta)
    }

    /// Mark a slot as DELETED: 4-byte pwrite of the new `meta` only.
    /// Key and value on disk remain intact (will be overwritten on next
    /// OCCUPIED write to this slot_id). Returns the new `meta`.
    pub fn delete_slot(&mut self, slot_id: u32, _key: &[u8]) -> DbResult<u32> {
        let old_meta = self.versions[slot_id as usize];
        let new_meta = slot::pack_meta(
            slot::STATUS_DELETED,
            slot::version_of(old_meta).wrapping_add(1) & ((1 << 30) - 1),
        );

        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&new_meta.to_le_bytes(), offset)?;
        self.versions[slot_id as usize] = new_meta;

        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        Ok(new_meta)
    }

    /// Placeholder — filled in Task 8.
    #[inline]
    fn maybe_warn_wrap(&self, _slot_id: u32, _version: u32) {}
```

The `_key` parameter on `delete_slot` is unused at disk level but kept so the SPSC replication hook (Task 11/12) can carry it. Existing callers pass a key already.

- [ ] **Step 5.4: Update `durability.rs`'s `write_tombstone` for `Fixed`**

`armdb/src/durability.rs` line ≈275:

Current:
```rust
    fn write_tombstone(&mut self, _shard_id: u8, old_loc: u32, key: &[u8]) -> DbResult<()> {
        let _ = key;
        self.delete_slot(old_loc)?;
        self.bitmap.clear(old_loc);
        Ok(())
    }
```

Change to:
```rust
    fn write_tombstone(&mut self, _shard_id: u8, old_loc: u32, key: &[u8]) -> DbResult<()> {
        self.delete_slot(old_loc, key)?;
        self.bitmap.clear(old_loc);
        Ok(())
    }
```

- [ ] **Step 5.5: Fix existing test `test_delete_slot` in `shard.rs`**

It currently expects `slot::slot_status(&buf) == SLOT_DELETED` and calls `slot::validate_slot` — both removed. Rewrite:

```rust
    #[test]
    fn test_delete_slot() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();
        shard.delete_slot(id, b"key_0001").unwrap();
        let buf = shard.read_slot(id).unwrap();
        assert_eq!(slot::status_of(slot::meta_of(&buf)), slot::STATUS_DELETED);
        assert!(slot::read_slot(&buf, 8, 16).is_none());
    }
```

And `test_write_read_slot` — update to use `read_slot`:

```rust
    #[test]
    fn test_write_read_slot() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let slot_id = shard.alloc_slot().unwrap();
        let key = b"key_0001";
        let value = b"value___00000001";
        shard.write_slot(slot_id, key, value).unwrap();
        let buf = shard.read_slot(slot_id).unwrap();
        let (_m, k, v) = slot::read_slot(&buf, key.len(), value.len()).expect("CRC should match");
        assert_eq!(k, key);
        assert_eq!(v, value);
    }
```

- [ ] **Step 5.6: Run tests**

```sh
cargo nextest run -p armdb -E 'test(fixed::shard)'
```

All shard tests should pass.

- [ ] **Step 5.7: Commit**

```sh
git add armdb/src/fixed/shard.rs armdb/src/durability.rs
git commit -m "feat(armdb/fixed): write_slot bumps version, delete_slot is 4-byte pwrite"
```

---

### Task 6: Add `apply_foreign_slot`, `apply_foreign_delete`, `read_slot_header_and_key`

**Files:**
- Modify: `armdb/src/fixed/shard.rs`

Per spec §5. These are used on follower only — write foreign `meta` as-is (no bump).

- [ ] **Step 6.1: Write failing tests**

Append to `mod tests` in `shard.rs`:

```rust
    #[test]
    fn test_apply_foreign_slot_overwrites_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let foreign_meta = slot::pack_meta(slot::STATUS_OCCUPIED, 500);
        shard.apply_foreign_slot(0, foreign_meta, b"key_0001", b"value___00000001").unwrap();
        assert_eq!(shard.versions[0], foreign_meta);
        let buf = shard.read_slot(0).unwrap();
        assert_eq!(slot::meta_of(&buf), foreign_meta);
    }

    #[test]
    fn test_apply_foreign_delete_sets_meta() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let occ = slot::pack_meta(slot::STATUS_OCCUPIED, 100);
        shard.apply_foreign_slot(0, occ, b"keyabcde", b"0123456701234567").unwrap();
        let del = slot::pack_meta(slot::STATUS_DELETED, 101);
        shard.apply_foreign_delete(0, del).unwrap();
        assert_eq!(shard.versions[0], del);
        let buf = shard.read_slot(0).unwrap();
        assert_eq!(slot::meta_of(&buf), del);
        // key bytes preserved
        assert_eq!(&buf[slot::SLOT_HEADER_SIZE..slot::SLOT_HEADER_SIZE + 8], b"keyabcde");
    }

    #[test]
    fn test_read_slot_header_and_key() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();
        let buf = shard.read_slot_header_and_key(id, 8).unwrap();
        assert_eq!(buf.len(), slot::SLOT_HEADER_SIZE + 8);
        assert_eq!(&buf[slot::SLOT_HEADER_SIZE..], b"key_0001");
    }
```

- [ ] **Step 6.2: Implement the three methods**

Add to the `impl FixedShardInner` block:

```rust
    /// Apply a slot event from a leader — writes `meta` as-is (no bump).
    /// Used by follower replication apply path.
    pub fn apply_foreign_slot(
        &mut self,
        slot_id: u32,
        meta: u32,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<()> {
        let size = self.slot_size as usize;
        let mut buf = vec![0u8; size];
        slot::serialize_slot(&mut buf, meta, key, value);
        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&buf, offset)?;
        self.versions[slot_id as usize] = meta;
        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        Ok(())
    }

    /// Apply a DELETE event from a leader — 4-byte pwrite of `meta`.
    pub fn apply_foreign_delete(&mut self, slot_id: u32, meta: u32) -> DbResult<()> {
        let offset = self.slot_offset(slot_id);
        self.file.write_all_at(&meta.to_le_bytes(), offset)?;
        self.versions[slot_id as usize] = meta;
        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        Ok(())
    }

    /// Read only the first `SLOT_HEADER_SIZE + key_len` bytes of a slot.
    /// Used rarely on follower stale-cleanup path (spec §8).
    pub fn read_slot_header_and_key(&self, slot_id: u32, key_len: usize) -> DbResult<Vec<u8>> {
        let n = slot::SLOT_HEADER_SIZE + key_len;
        let mut buf = vec![0u8; n];
        self.file.read_exact_at(&mut buf, self.slot_offset(slot_id))?;
        Ok(buf)
    }
```

- [ ] **Step 6.3: Run tests**

```sh
cargo nextest run -p armdb -E 'test(fixed::shard) and (test(apply_foreign) or test(read_slot_header))'
```

- [ ] **Step 6.4: Commit**

```sh
git add armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed): add apply_foreign_slot/delete and read_slot_header_and_key"
```

---

### Task 7: Replace `fixed.bitmap` sidecar with `fixed.versions` (with CRC trailer)

**Files:**
- Modify: `armdb/src/fixed/shard.rs`
- Modify: `armdb/src/fixed/bitmap.rs`

Per spec §3 (sidecar integrity) and §4.

- [ ] **Step 7.1: Add `Bitmap::from_versions` helper**

In `armdb/src/fixed/bitmap.rs`, add method:

```rust
impl Bitmap {
    /// Build a bitmap by marking every slot whose `meta`'s status is OCCUPIED.
    pub fn from_versions(versions: &[u32]) -> Self {
        use crate::fixed::slot;
        let mut b = Bitmap::new(versions.len() as u32);
        for (i, &m) in versions.iter().enumerate() {
            if slot::status_of(m) == slot::STATUS_OCCUPIED {
                b.set(i as u32);
            }
        }
        b
    }
}
```

Add unit test:

```rust
    #[test]
    fn test_from_versions() {
        use crate::fixed::slot::{pack_meta, STATUS_OCCUPIED, STATUS_DELETED, STATUS_FREE};
        let v = vec![
            pack_meta(STATUS_OCCUPIED, 1),
            pack_meta(STATUS_DELETED, 2),
            pack_meta(STATUS_FREE, 0),
            pack_meta(STATUS_OCCUPIED, 5),
        ];
        let b = Bitmap::from_versions(&v);
        assert!(b.is_set(0));
        assert!(!b.is_set(1));
        assert!(!b.is_set(2));
        assert!(b.is_set(3));
        assert_eq!(b.occupied(), 2);
    }
```

Run:
```sh
cargo nextest run -p armdb -E 'test(fixed::bitmap::tests::test_from_versions)'
```

- [ ] **Step 7.2: Write failing test for versions sidecar roundtrip**

Add to `armdb/src/fixed/shard.rs` `mod tests`:

```rust
    #[test]
    fn test_versions_sidecar_roundtrip() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id0 = shard.alloc_slot().unwrap();
            let id1 = shard.alloc_slot().unwrap();
            shard.write_slot(id0, b"key_0001", b"value___00000001").unwrap();
            shard.write_slot(id1, b"key_0002", b"value___00000002").unwrap();
            shard.delete_slot(id1, b"key_0002").unwrap();
            shard.clean_shutdown().unwrap();
        }

        // Reopen: sidecar present, load versions, derive bitmap.
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        assert!(shard.has_clean_shutdown());
        shard.load_versions_sidecar().unwrap();
        assert_eq!(shard.versions.len(), shard.slot_count() as usize);
        assert_eq!(slot::status_of(shard.versions[0]), slot::STATUS_OCCUPIED);
        assert_eq!(slot::status_of(shard.versions[1]), slot::STATUS_DELETED);
        // derived bitmap: only slot 0 is OCCUPIED
        let b = crate::fixed::bitmap::Bitmap::from_versions(&shard.versions);
        assert!(b.is_set(0));
        assert!(!b.is_set(1));
    }

    #[test]
    fn test_versions_sidecar_trailer_validation_fails_on_corruption() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();

        {
            let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
            let id = shard.alloc_slot().unwrap();
            shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();
            shard.clean_shutdown().unwrap();
        }

        // Truncate the sidecar (strip trailer).
        let sidecar = shard_dir.join("fixed.versions");
        let data = std::fs::read(&sidecar).unwrap();
        std::fs::write(&sidecar, &data[..data.len() - 4]).unwrap();

        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let err = shard.load_versions_sidecar().unwrap_err();
        match err {
            DbError::FormatMismatch(msg) => assert!(msg.contains("fixed.versions"), "got: {msg}"),
            other => panic!("expected FormatMismatch, got {other:?}"),
        }
    }
```

- [ ] **Step 7.3: Implement sidecar write + read + validate**

Replace `clean_shutdown`, `has_clean_shutdown`, `load_bitmap_sidecar` in `armdb/src/fixed/shard.rs` with:

```rust
    const SIDECAR_FILE: &str = "fixed.versions";

    /// Perform a clean shutdown: sync data, write versions sidecar with
    /// CRC trailer, set clean_shutdown flag.
    pub fn clean_shutdown(&mut self) -> DbResult<()> {
        self.file.sync_data()?;
        self.write_versions_sidecar()?;
        self.file.write_all_at(&[1u8], CLEAN_SHUTDOWN_OFFSET)?;
        self.file.sync_data()?;
        Ok(())
    }

    fn write_versions_sidecar(&self) -> DbResult<()> {
        let path = self.dir.join(Self::SIDECAR_FILE);
        let expected_len = (self.slot_count as usize) * 4 + 8;
        let mut buf: Vec<u8> = Vec::with_capacity(expected_len);
        for &m in &self.versions {
            buf.extend_from_slice(&m.to_le_bytes());
        }
        buf.extend_from_slice(&self.slot_count.to_le_bytes());
        let mut h = crc32fast::Hasher::new();
        h.update(&buf);
        let crc = h.finalize();
        buf.extend_from_slice(&crc.to_le_bytes());
        std::fs::write(&path, &buf)?;
        Ok(())
    }

    /// Returns `true` when a prior run cleanly shut down AND a versions
    /// sidecar is on disk.  Does NOT validate the sidecar — caller must
    /// call `load_versions_sidecar` which does the validation.
    pub fn has_clean_shutdown(&self) -> bool {
        let mut buf = [0u8; 1];
        if self
            .file
            .read_exact_at(&mut buf, CLEAN_SHUTDOWN_OFFSET)
            .is_err()
        {
            return false;
        }
        buf[0] == 1 && self.dir.join(Self::SIDECAR_FILE).exists()
    }

    /// Load the versions sidecar, validating its trailer (slot_count + CRC32).
    /// On success, `self.versions` is filled and `self.bitmap` is derived.
    /// On any mismatch, returns `DbError::FormatMismatch` — caller should
    /// remove the sidecar file and fall back to a full slot scan.
    pub fn load_versions_sidecar(&mut self) -> DbResult<()> {
        let path = self.dir.join(Self::SIDECAR_FILE);
        let data = std::fs::read(&path)?;
        let expected_len = (self.slot_count as usize) * 4 + 8;
        if data.len() != expected_len {
            return Err(DbError::FormatMismatch(format!(
                "fixed.versions size mismatch: expected {expected_len}, got {}",
                data.len()
            )));
        }
        let versions_bytes = &data[..self.slot_count as usize * 4];
        let trailer = &data[data.len() - 8..];
        let stored_slot_count = u32::from_le_bytes(trailer[0..4].try_into().unwrap());
        let stored_crc = u32::from_le_bytes(trailer[4..8].try_into().unwrap());
        if stored_slot_count != self.slot_count {
            return Err(DbError::FormatMismatch(format!(
                "fixed.versions slot_count mismatch: stored {stored_slot_count}, header {}",
                self.slot_count
            )));
        }
        let mut h = crc32fast::Hasher::new();
        h.update(&data[..data.len() - 4]);
        let actual_crc = h.finalize();
        if actual_crc != stored_crc {
            return Err(DbError::FormatMismatch(format!(
                "fixed.versions CRC mismatch: expected {stored_crc:#x}, got {actual_crc:#x}"
            )));
        }
        self.versions.clear();
        self.versions.reserve(self.slot_count as usize);
        for chunk in versions_bytes.chunks_exact(4) {
            self.versions.push(u32::from_le_bytes(chunk.try_into().unwrap()));
        }
        self.bitmap = crate::fixed::bitmap::Bitmap::from_versions(&self.versions);
        Ok(())
    }
```

Delete the old `load_bitmap_sidecar` method.

- [ ] **Step 7.4: Update `clear_clean_shutdown` (probably unchanged) and sidecar path references elsewhere**

The existing `clear_clean_shutdown` method in `shard.rs` already writes `[0u8]` at `CLEAN_SHUTDOWN_OFFSET` — keep as-is. Remove any reference to `fixed.bitmap` path; if the old shard code created or deleted `fixed.bitmap`, replace with `fixed.versions`.

- [ ] **Step 7.5: Run tests**

```sh
cargo nextest run -p armdb -E 'test(fixed::shard::tests)'
```

- [ ] **Step 7.6: Commit**

```sh
git add armdb/src/fixed/shard.rs armdb/src/fixed/bitmap.rs
git commit -m "feat(armdb/fixed): replace fixed.bitmap with fixed.versions sidecar (CRC trailer)"
```

---

## Phase 3: Recovery with versions population

### Task 8: Update `Fixed::recover_entries` to populate `versions` + handle torn slots

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

Per spec §4 and §9.

- [ ] **Step 8.1: Write failing tests**

Create `armdb/tests/fixed_recovery.rs` (new integration test file) — see below. If this file already exists, append tests; if not, create it:

```rust
//! Integration tests for Fixed backend recovery + versions population.

use armdb::fixed::{FixedConfig};
use armdb::fixed::slot::{self, pack_meta, version_of, status_of, STATUS_OCCUPIED, STATUS_DELETED, STATUS_FREE};
// NOTE: many of these types are pub(crate) — if needed, expose via Fixed or
// build tests through the public Db API (open_zero_tree_fixed).
//
// Concrete assertions go in unit tests inside durability.rs / shard.rs.
```

Instead of a new integration file, add unit tests inside `armdb/src/durability.rs` at the bottom:

```rust
#[cfg(test)]
mod tests {
    use super::*;
    use crate::fixed::config::FixedConfig;
    use tempfile::tempdir;

    fn test_fixed_config() -> FixedConfig {
        FixedConfig {
            shard_count: 1,
            grow_step: 16,
            ..FixedConfig::test()
        }
    }

    #[test]
    fn test_recover_populates_versions_dirty() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        // First run: write a few entries, simulate crash (no clean shutdown).
        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = fixed.engine.shards()[0].inner.lock();
                let id0 = inner.alloc_slot().unwrap();
                inner.write_slot(id0, b"keyaaaaa", b"val_0000_0000_00").unwrap();
                let id1 = inner.alloc_slot().unwrap();
                inner.write_slot(id1, b"keybbbbb", b"val_0001_0000_00").unwrap();
                inner.delete_slot(id1, b"keybbbbb").unwrap();
                inner.sync().unwrap();
            }
            // Drop without clean_shutdown.
            drop(fixed);
        }

        // Second run: reopen, recover_entries must populate versions.
        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let mut entries = Vec::new();
        fixed.recover_entries(|_shard, k, v, slot_id| {
            entries.push((k.to_vec(), v.to_vec(), slot_id));
        }).unwrap();
        assert_eq!(entries.len(), 1, "only one OCCUPIED entry");
        assert_eq!(entries[0].0, b"keyaaaaa");

        // versions for OCCUPIED / DELETED slots must be populated.
        let inner = fixed.engine.shards()[0].inner.lock();
        assert_eq!(status_of(inner.versions[0]), STATUS_OCCUPIED);
        assert_eq!(status_of(inner.versions[1]), STATUS_DELETED);
    }

    #[test]
    fn test_recover_populates_versions_clean() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        // First run: write, clean shutdown (sidecar written).
        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = fixed.engine.shards()[0].inner.lock();
                let id = inner.alloc_slot().unwrap();
                inner.write_slot(id, b"keyaaaaa", b"val_0000_0000_00").unwrap();
            }
            fixed.engine.close().unwrap();
        }

        // Second run: sidecar present → clean path.
        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let mut entries = Vec::new();
        fixed.recover_entries(|_shard, k, _v, _id| {
            entries.push(k.to_vec());
        }).unwrap();
        assert_eq!(entries.len(), 1);
        let inner = fixed.engine.shards()[0].inner.lock();
        assert_eq!(status_of(inner.versions[0]), STATUS_OCCUPIED);
        assert_eq!(version_of(inner.versions[0]), 1);
    }

    #[test]
    fn test_torn_slot_preserves_version() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("db");

        // Write one slot, then corrupt its value bytes (simulating torn write),
        // close without clean shutdown.
        {
            let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
            {
                let mut inner = fixed.engine.shards()[0].inner.lock();
                let id = inner.alloc_slot().unwrap();
                // Write with version_of = 1.
                inner.write_slot(id, b"keyaaaaa", b"val_0000_0000_00").unwrap();
                inner.sync().unwrap();
            }
            drop(fixed);
        }

        // Corrupt the value bytes directly on disk (skip meta + crc + key).
        let shard_data = path.join("shard_000").join("fixed.data");
        use std::os::unix::fs::FileExt;
        let f = std::fs::OpenOptions::new().write(true).open(&shard_data).unwrap();
        // slot_size for K=8 V=16 = 32; offset = HEADER(4096) + 0*32 + 8(header) + 8(key) = 4112
        f.write_all_at(&[0xFFu8; 16], 4112).unwrap();
        f.sync_data().unwrap();
        drop(f);

        // Reopen dirty → recover → slot has CRC mismatch.
        let fixed = Fixed::open(&path, test_fixed_config(), 8, 16).unwrap();
        let mut entries = Vec::new();
        fixed.recover_entries(|_shard, k, _v, _id| {
            entries.push(k.to_vec());
        }).unwrap();
        assert!(entries.is_empty(), "torn slot must be skipped");

        // versions[0] must keep the original version (1) but with status=FREE.
        let inner = fixed.engine.shards()[0].inner.lock();
        assert_eq!(status_of(inner.versions[0]), STATUS_FREE);
        assert_eq!(version_of(inner.versions[0]), 1,
            "version must be preserved so next bump continues monotonically");
    }
}
```

These tests depend on `fixed.engine` being `pub(crate)` accessible from `durability.rs` (it is).

- [ ] **Step 8.2: Rewrite `Fixed::recover_entries`**

Replace the body of `Fixed::recover_entries` in `armdb/src/durability.rs`:

```rust
    /// Recover entries from all shards. Populates `versions` for every
    /// shard (regardless of clean/dirty path). Calls `visitor` for each
    /// valid OCCUPIED slot.
    pub fn recover_entries(
        &self,
        mut visitor: impl FnMut(usize, &[u8], &[u8], u32),
    ) -> DbResult<u32> {
        let shards = self.engine.shards();
        let mut total_recovered = 0u32;

        for (shard_idx, shard) in shards.iter().enumerate() {
            let mut inner = shard.inner.lock();
            let key_len = inner.key_len() as usize;
            let value_len = inner.value_len() as usize;
            let slot_count = inner.slot_count();

            let used_sidecar = inner.has_clean_shutdown() && {
                match inner.load_versions_sidecar() {
                    Ok(()) => true,
                    Err(e) => {
                        tracing::warn!(
                            shard = shard_idx, error = %e,
                            "fixed.versions sidecar invalid; falling back to full scan"
                        );
                        let _ = std::fs::remove_file(inner.dir().join("fixed.versions"));
                        false
                    }
                }
            };

            if used_sidecar {
                for slot_id in 0..slot_count {
                    if slot::status_of(inner.versions[slot_id as usize]) != slot::STATUS_OCCUPIED {
                        continue;
                    }
                    let buf = inner.read_slot(slot_id)?;
                    match slot::read_slot(&buf, key_len, value_len) {
                        Some((_m, k, v)) => {
                            inner.bitmap.set(slot_id);
                            visitor(shard_idx, k, v, slot_id);
                            total_recovered += 1;
                        }
                        None => {
                            // Unexpected: sidecar said OCCUPIED but CRC fails.
                            // Treat as torn slot: mark FREE, keep version.
                            let meta = inner.versions[slot_id as usize];
                            inner.versions[slot_id as usize] =
                                slot::pack_meta(slot::STATUS_FREE, slot::version_of(meta));
                            inner.bitmap.clear(slot_id);
                        }
                    }
                }
            } else {
                // Dirty path: read every slot, populate versions unconditionally.
                for slot_id in 0..slot_count {
                    let buf = inner.read_slot(slot_id)?;
                    let meta = slot::meta_of(&buf);
                    let status = slot::status_of(meta);
                    inner.versions[slot_id as usize] = meta;

                    if status == slot::STATUS_OCCUPIED {
                        if let Some((_m, k, v)) = slot::read_slot(&buf, key_len, value_len) {
                            inner.bitmap.set(slot_id);
                            visitor(shard_idx, k, v, slot_id);
                            total_recovered += 1;
                        } else {
                            // Torn slot: keep version, clear status to FREE.
                            inner.versions[slot_id as usize] =
                                slot::pack_meta(slot::STATUS_FREE, slot::version_of(meta));
                            // bitmap bit already 0 by default.
                        }
                    }
                    // DELETED: keep versions[i] = meta, bitmap stays 0.
                    // FREE: keep versions[i] = meta (likely 0), bitmap stays 0.
                }
            }

            inner.clear_clean_shutdown()?;
        }

        Ok(total_recovered)
    }
```

Also add a `dir()` accessor on `FixedShardInner` if missing:

```rust
    pub fn dir(&self) -> &std::path::Path {
        &self.dir
    }
```

- [ ] **Step 8.3: Add `Fixed::close` that calls `clean_shutdown` on each shard**

Currently `Fixed::close` calls `self.engine.flush()`. Change to:

```rust
    fn close(&self) -> DbResult<()> {
        self.engine.close()
    }
```

Ensure `FixedEngine::close` calls `inner.clean_shutdown()?;` on each shard (it probably does — spec §8 already uses this, verify).

- [ ] **Step 8.4: Run tests**

```sh
cargo nextest run -p armdb -E 'test(durability::tests)'
```

- [ ] **Step 8.5: Commit**

```sh
git add armdb/src/durability.rs armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed): recover_entries populates versions; torn slot preserves version"
```

---

### Task 9: Add `maybe_warn_wrap` with metric + tracing

**Files:**
- Modify: `armdb/src/fixed/shard.rs`

Per spec §5.

- [ ] **Step 9.1: Replace the placeholder `maybe_warn_wrap`**

```rust
    #[inline]
    fn maybe_warn_wrap(&self, slot_id: u32, version: u32) {
        if version >= slot::VERSION_WARN_THRESHOLD {
            metrics::counter!(
                "armdb.fixed.version_near_wrap",
                "shard" => self.shard_id.to_string()
            )
            .increment(1);
            tracing::warn!(
                shard_id = self.shard_id,
                slot_id,
                version,
                "FixedStore slot version approaching 30-bit wrap"
            );
        }
    }
```

Add import at top of `shard.rs` if missing:
```rust
use crate::fixed::slot;
```
(probably already present).

- [ ] **Step 9.2: Add a unit test** (optional — metric assertion is hard; just check it compiles and the branch executes)

```rust
    #[test]
    fn test_warn_wrap_threshold_triggers() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let id = shard.alloc_slot().unwrap();
        // Force versions[id] to just below wrap threshold so next write triggers.
        shard.versions[id as usize] =
            slot::pack_meta(slot::STATUS_OCCUPIED, slot::VERSION_WARN_THRESHOLD - 1);
        // No panic expected; warning goes to tracing subscriber (test doesn't assert).
        shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();
        assert!(slot::version_of(shard.versions[id as usize]) >= slot::VERSION_WARN_THRESHOLD);
    }
```

Run:
```sh
cargo nextest run -p armdb -E 'test(test_warn_wrap_threshold_triggers)'
```

- [ ] **Step 9.3: Commit**

```sh
git add armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed): emit warn + metric when slot version approaches wrap"
```

---

## Phase 4: Format version bumps

### Task 10: Bump `db.meta` version and `fixed.data` header version to 2

**Files:**
- Modify: `armdb/src/fixed/engine.rs`
- Modify: `armdb/src/fixed/shard.rs`

Per spec §3. Breaking change — old files refuse to open with `FormatMismatch` (no production data yet, per user).

- [ ] **Step 10.1: Bump `META_VERSION` in `engine.rs`**

In `armdb/src/fixed/engine.rs`:
```rust
const META_VERSION: u8 = 2;  // was 1
```

- [ ] **Step 10.2: Bump `fixed.data` header `VERSION` in `shard.rs`**

In `armdb/src/fixed/shard.rs`:
```rust
const VERSION: u16 = 2;  // was 1
```

- [ ] **Step 10.3: Add test asserting old-version file is rejected**

Append to `armdb/src/fixed/shard.rs` `mod tests`:

```rust
    #[test]
    fn test_reject_old_version() {
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        std::fs::create_dir_all(&shard_dir).unwrap();

        // Write a valid header with VERSION = 1 manually.
        let slot_size = slot::slot_size(8, 16) as u16;
        let mut header = [0u8; 4096];
        header[0..4].copy_from_slice(b"FIXD");
        header[4..6].copy_from_slice(&1u16.to_le_bytes());  // OLD VERSION
        header[6..8].copy_from_slice(&slot_size.to_le_bytes());
        header[8..12].copy_from_slice(&10u32.to_le_bytes());
        header[12..14].copy_from_slice(&8u16.to_le_bytes());
        header[14..16].copy_from_slice(&16u16.to_le_bytes());
        header[16] = 0; // shard_id
        let data_path = shard_dir.join("fixed.data");
        std::fs::write(&data_path, header).unwrap();
        // Extend file to match header's declared slot_count.
        let f = std::fs::OpenOptions::new().write(true).open(&data_path).unwrap();
        f.set_len(4096 + 10 * slot_size as u64).unwrap();

        let cfg = test_config();
        let err = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap_err();
        match err {
            DbError::FormatMismatch(msg) => assert!(msg.contains("version"), "got: {msg}"),
            other => panic!("expected FormatMismatch, got {other:?}"),
        }
    }
```

- [ ] **Step 10.4: Run tests**

```sh
cargo nextest run -p armdb -E 'test(test_reject_old_version)'
```

- [ ] **Step 10.5: Commit**

```sh
git add armdb/src/fixed/engine.rs armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed): bump db.meta and fixed.data format version to 2"
```

---

## Phase 5: Replication scaffolding

### Task 11: Create `fixed_replication` module skeleton

**Files:**
- Create: `armdb/src/fixed_replication/mod.rs`
- Create: `armdb/src/fixed_replication/event.rs`
- Modify: `armdb/src/lib.rs`

- [ ] **Step 11.1: Create `event.rs` with `FixedReplicationEvent`**

Write to `armdb/src/fixed_replication/event.rs`:

```rust
//! SPSC event payload pushed by leader's write_slot / delete_slot
//! and consumed by the replication server.

/// Event carried over per-shard rtrb SPSC from leader write path to
/// replication server thread. Producer is `FixedShardInner.replication_tx`.
pub enum FixedReplicationEvent {
    /// OCCUPIED write. `payload` is the full serialized slot (slot_size
    /// bytes: meta + crc + key + value) — same Vec that was pwrite'd.
    Write {
        slot_id: u32,
        payload: Vec<u8>,
    },
    /// DELETED mark. Key is carried separately so the follower can
    /// update its index without a pread on its own slot.
    Delete {
        slot_id: u32,
        meta: u32,
        key: Vec<u8>,
    },
}
```

- [ ] **Step 11.2: Create `mod.rs`**

Write to `armdb/src/fixed_replication/mod.rs`:

```rust
//! Replication for the FixedStore backend.
//!
//! Separate module from `crate::replication` (Bitcask). Separate port,
//! protocol, and lifecycle. See `armdb/docs/superpowers/specs/2026-04-17-fixed-replication-design.md`.

pub mod event;

pub use event::FixedReplicationEvent;
```

- [ ] **Step 11.3: Gate module in `lib.rs`**

Add to `armdb/src/lib.rs` near other `pub mod` lines:

```rust
#[cfg(feature = "replication")]
pub mod fixed_replication;
```

- [ ] **Step 11.4: Verify compilation**

```sh
cargo check -p armdb
cargo check -p armdb --no-default-features --features replication
```

Both should compile.

- [ ] **Step 11.5: Commit**

```sh
git add armdb/src/fixed_replication/ armdb/src/lib.rs
git commit -m "feat(armdb): scaffold fixed_replication module and FixedReplicationEvent"
```

---

### Task 12: Add `replication_tx` field to `FixedShardInner` + SPSC hook in write/delete

**Files:**
- Modify: `armdb/src/fixed/shard.rs`

- [ ] **Step 12.1: Add the field and constructor wiring**

In `armdb/src/fixed/shard.rs`, add to the struct:

```rust
    #[cfg(feature = "replication")]
    pub(crate) replication_tx: Option<rtrb::Producer<crate::fixed_replication::FixedReplicationEvent>>,
```

In both branches of `FixedShardInner::open`, include `#[cfg(feature = "replication")] replication_tx: None,` in the struct literal.

- [ ] **Step 12.2: Hook in `write_slot`**

After the existing body of `write_slot` (after `if self.enable_fsync { ... }`), but BEFORE `Ok(new_meta)`:

```rust
        #[cfg(feature = "replication")]
        if let Some(tx) = &mut self.replication_tx {
            let _ = tx.push(
                crate::fixed_replication::FixedReplicationEvent::Write {
                    slot_id,
                    payload: buf,
                },
            );
            // Err (overflow) → event dropped silently; reconnect catch-up
            // will reconcile via full scan.
        }
```

**Caveat:** `buf` is moved here. If `enable_fsync` is on, the `file.sync_data()?` call earlier already happened, so moving `buf` after is fine. Re-order: sync_data → hook, or move the sync before the hook:

```rust
        self.pending_writes += 1;
        if self.enable_fsync {
            self.file.sync_data()?;
        }
        #[cfg(feature = "replication")]
        if let Some(tx) = &mut self.replication_tx {
            let _ = tx.push(
                crate::fixed_replication::FixedReplicationEvent::Write {
                    slot_id,
                    payload: buf,
                },
            );
        }
        Ok(new_meta)
```

- [ ] **Step 12.3: Hook in `delete_slot`**

At the end of `delete_slot`, before `Ok(new_meta)`:

```rust
        #[cfg(feature = "replication")]
        if let Some(tx) = &mut self.replication_tx {
            let _ = tx.push(
                crate::fixed_replication::FixedReplicationEvent::Delete {
                    slot_id,
                    meta: new_meta,
                    key: _key.to_vec(),
                },
            );
        }
```

Rename the `_key` parameter to `key` (drop leading underscore) — it's now used.

- [ ] **Step 12.4: Write a unit test that exercises the hook**

Add to `mod tests`:

```rust
    #[cfg(feature = "replication")]
    #[test]
    fn test_replication_hook_pushes_events() {
        use crate::fixed_replication::FixedReplicationEvent;
        let dir = tempdir().unwrap();
        let shard_dir = dir.path().join("shard_0");
        let cfg = test_config();
        let mut shard = FixedShardInner::open(&shard_dir, 0, 8, 16, &cfg).unwrap();
        let (producer, mut consumer) = rtrb::RingBuffer::new(8);
        shard.replication_tx = Some(producer);

        let id = shard.alloc_slot().unwrap();
        shard.write_slot(id, b"key_0001", b"value___00000001").unwrap();
        shard.delete_slot(id, b"key_0001").unwrap();

        // Expect two events.
        match consumer.pop().unwrap() {
            FixedReplicationEvent::Write { slot_id, payload } => {
                assert_eq!(slot_id, id);
                assert_eq!(payload.len(), shard.slot_size as usize);
                assert_eq!(slot::status_of(slot::meta_of(&payload)), slot::STATUS_OCCUPIED);
            }
            _ => panic!("expected Write"),
        }
        match consumer.pop().unwrap() {
            FixedReplicationEvent::Delete { slot_id, meta, key } => {
                assert_eq!(slot_id, id);
                assert_eq!(slot::status_of(meta), slot::STATUS_DELETED);
                assert_eq!(key, b"key_0001");
            }
            _ => panic!("expected Delete"),
        }
    }
```

Make sure `rtrb` is visible in tests; it already is (since replication feature is on by default).

- [ ] **Step 12.5: Run tests**

```sh
cargo nextest run -p armdb -E 'test(test_replication_hook_pushes_events)'
```

- [ ] **Step 12.6: Commit**

```sh
git add armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed): SPSC hook in write_slot/delete_slot for replication"
```

---

### Task 13: Add `install_replication_producers` to `FixedEngine`

**Files:**
- Modify: `armdb/src/fixed/engine.rs`

Per spec §5 / §7 (server uses this on first follower connect).

- [ ] **Step 13.1: Add method**

Add inside `impl FixedEngine`:

```rust
    /// Install SPSC producers into each shard's `replication_tx`.
    /// Called by `FixedReplicationServer` on the first follower accept.
    /// Panics if any shard already has a producer installed or if the
    /// producers slice length mismatches `shard_count`.
    #[cfg(feature = "replication")]
    pub fn install_replication_producers(
        &self,
        producers: Vec<rtrb::Producer<crate::fixed_replication::FixedReplicationEvent>>,
    ) {
        assert_eq!(
            producers.len(),
            self.shards.len(),
            "producer count must match shard count"
        );
        for (shard, tx) in self.shards.iter().zip(producers) {
            let mut inner = shard.inner.lock();
            assert!(
                inner.replication_tx.is_none(),
                "producer already installed for shard {}",
                shard.id
            );
            inner.replication_tx = Some(tx);
        }
    }
```

- [ ] **Step 13.2: Commit**

```sh
git add armdb/src/fixed/engine.rs
git commit -m "feat(armdb/fixed): install_replication_producers on FixedEngine"
```

---

### Task 14: Create `FixedEngineAccess` trait + impl for `FixedEngine`

**Files:**
- Create: `armdb/src/fixed_replication/engine_access.rs`
- Modify: `armdb/src/fixed_replication/mod.rs`
- Modify: `armdb/src/fixed/engine.rs` (impl)

Per spec §7 (`FixedEngineAccess`).

- [ ] **Step 14.1: Define trait**

Write to `armdb/src/fixed_replication/engine_access.rs`:

```rust
//! Trait abstracting the leader engine for the replication server.
//! Lets tests mock engine behavior and keeps server code decoupled
//! from `FixedEngine`.

use std::sync::Arc;

use crate::error::DbResult;
use crate::fixed_replication::event::FixedReplicationEvent;

pub trait FixedEngineAccess: Send + Sync + 'static {
    fn shard_count(&self) -> usize;
    fn key_len(&self) -> usize;
    fn value_len(&self) -> usize;
    fn slot_size(&self) -> u16;
    fn current_slot_count(&self, shard_id: usize) -> u32;

    /// Read a contiguous chunk of `count` slots starting at `start_slot`
    /// into a fresh Vec<u8>. Byte length = `count * slot_size`.
    fn read_shard_chunk(
        &self,
        shard_id: usize,
        start_slot: u32,
        count: usize,
    ) -> DbResult<Vec<u8>>;

    /// Install SPSC producers into shards (lazy on first follower).
    fn install_replication_producers(
        &self,
        producers: Vec<rtrb::Producer<FixedReplicationEvent>>,
    );

    /// Hint for monitoring — follower Acked this version. Used only for metrics.
    fn update_min_replicated_version(&self, shard_id: usize, version: u32);
}

/// Convenience wrapper for `Arc<dyn FixedEngineAccess>` used across
/// server threads.
pub type ArcEngine = Arc<dyn FixedEngineAccess>;
```

- [ ] **Step 14.2: Re-export from `mod.rs`**

Update `armdb/src/fixed_replication/mod.rs`:

```rust
pub mod event;
pub mod engine_access;

pub use event::FixedReplicationEvent;
pub use engine_access::{ArcEngine, FixedEngineAccess};
```

- [ ] **Step 14.3: Implement for `FixedEngine`**

Add at the bottom of `armdb/src/fixed/engine.rs`:

```rust
#[cfg(feature = "replication")]
impl crate::fixed_replication::FixedEngineAccess for FixedEngine {
    fn shard_count(&self) -> usize {
        self.shards.len()
    }
    fn key_len(&self) -> usize {
        // All shards share the same key_len; read from shard 0.
        self.shards[0].inner.lock().key_len() as usize
    }
    fn value_len(&self) -> usize {
        self.shards[0].inner.lock().value_len() as usize
    }
    fn slot_size(&self) -> u16 {
        self.shards[0].inner.lock().slot_size
    }
    fn current_slot_count(&self, shard_id: usize) -> u32 {
        self.shards[shard_id].inner.lock().slot_count()
    }
    fn read_shard_chunk(
        &self,
        shard_id: usize,
        start_slot: u32,
        count: usize,
    ) -> crate::error::DbResult<Vec<u8>> {
        let inner = self.shards[shard_id].inner.lock();
        let slot_size = inner.slot_size as usize;
        let mut buf = vec![0u8; count * slot_size];
        let offset = 4096u64 + start_slot as u64 * slot_size as u64;
        use std::os::unix::fs::FileExt;
        // inner.file is private — expose via a small accessor; see Step 14.4.
        inner.read_chunk_at(offset, &mut buf)?;
        Ok(buf)
    }
    fn install_replication_producers(
        &self,
        producers: Vec<rtrb::Producer<crate::fixed_replication::FixedReplicationEvent>>,
    ) {
        FixedEngine::install_replication_producers(self, producers);
    }
    fn update_min_replicated_version(&self, _shard_id: usize, _version: u32) {
        // Metric-only hook; a full impl would update a per-shard atomic.
        // For v1 we just no-op; acks still arrive for monitoring via tracing.
    }
}
```

- [ ] **Step 14.4: Add `read_chunk_at` helper on `FixedShardInner`**

In `armdb/src/fixed/shard.rs`:

```rust
    /// Read raw bytes from the data file at absolute offset.
    /// Used by the replication server's full scan.
    pub fn read_chunk_at(&self, offset: u64, buf: &mut [u8]) -> DbResult<()> {
        use std::os::unix::fs::FileExt;
        self.file.read_exact_at(buf, offset)?;
        Ok(())
    }
```

- [ ] **Step 14.5: Compile check**

```sh
cargo check -p armdb
```

- [ ] **Step 14.6: Commit**

```sh
git add armdb/src/fixed_replication/engine_access.rs armdb/src/fixed_replication/mod.rs armdb/src/fixed/engine.rs armdb/src/fixed/shard.rs
git commit -m "feat(armdb/fixed_replication): FixedEngineAccess trait + impl for FixedEngine"
```

---

## Phase 6: Wire protocol

### Task 15: Define `FixedMessageType` enum + `Frame` encode/decode

**Files:**
- Create: `armdb/src/fixed_replication/protocol.rs`
- Modify: `armdb/src/fixed_replication/mod.rs`

Per spec §6.

- [ ] **Step 15.1: Scaffold `protocol.rs`**

Write to `armdb/src/fixed_replication/protocol.rs`:

```rust
//! Wire protocol for FixedStore replication.
//!
//! Frame format: `[type: u8][len: u32 LE][payload: len bytes]`.
//! All integers are little-endian. See spec §6 for the full message table.

use std::io::{self, Read, Write};

pub const PROTOCOL_VERSION: u8 = 1;
pub const BATCH_MAX_ENTRIES: usize = 256;
pub const BATCH_MAX_BYTES: usize = 64 * 1024;
pub const TAIL_POLL_MS: u64 = 1;
pub const HEARTBEAT_INTERVAL_SECS: u64 = 5;
pub const ACK_INTERVAL: usize = 1000;

/// Flags carried in SyncRequest.
pub const FLAG_EMPTY_STATE: u8 = 1 << 0;

#[repr(u8)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FixedMessageType {
    SyncRequest = 1,
    ShardInfo = 2,
    SlotBatch = 3,
    CaughtUp = 4,
    Ack = 5,
    Heartbeat = 6,
    Error = 255,
}

impl FixedMessageType {
    pub fn from_u8(v: u8) -> Option<Self> {
        Some(match v {
            1 => Self::SyncRequest,
            2 => Self::ShardInfo,
            3 => Self::SlotBatch,
            4 => Self::CaughtUp,
            5 => Self::Ack,
            6 => Self::Heartbeat,
            255 => Self::Error,
            _ => return None,
        })
    }
}

pub struct Frame {
    pub msg_type: FixedMessageType,
    pub payload: Vec<u8>,
}

pub fn write_frame(w: &mut impl Write, frame: &Frame) -> io::Result<()> {
    w.write_all(&[frame.msg_type as u8])?;
    w.write_all(&(frame.payload.len() as u32).to_le_bytes())?;
    w.write_all(&frame.payload)?;
    w.flush()
}

pub fn read_frame(r: &mut impl Read) -> io::Result<Frame> {
    let mut t = [0u8; 1];
    r.read_exact(&mut t)?;
    let msg_type = FixedMessageType::from_u8(t[0]).ok_or_else(|| {
        io::Error::new(io::ErrorKind::InvalidData, format!("unknown message type: {}", t[0]))
    })?;
    let mut len_buf = [0u8; 4];
    r.read_exact(&mut len_buf)?;
    let len = u32::from_le_bytes(len_buf) as usize;
    let mut payload = vec![0u8; len];
    if len > 0 {
        r.read_exact(&mut payload)?;
    }
    Ok(Frame { msg_type, payload })
}
```

Add `pub mod protocol;` to `armdb/src/fixed_replication/mod.rs`.

- [ ] **Step 15.2: Add frame roundtrip test**

At the bottom of `protocol.rs`:

```rust
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_frame_roundtrip() {
        let f = Frame {
            msg_type: FixedMessageType::Heartbeat,
            payload: vec![1, 2, 3],
        };
        let mut buf = Vec::new();
        write_frame(&mut buf, &f).unwrap();
        let parsed = read_frame(&mut &buf[..]).unwrap();
        assert_eq!(parsed.msg_type, FixedMessageType::Heartbeat);
        assert_eq!(parsed.payload, vec![1, 2, 3]);
    }

    #[test]
    fn test_frame_unknown_type() {
        let buf = [42u8, 0, 0, 0, 0];
        let err = read_frame(&mut &buf[..]).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
    }
}
```

- [ ] **Step 15.3: Run**

```sh
cargo nextest run -p armdb -E 'test(fixed_replication::protocol)'
```

- [ ] **Step 15.4: Commit**

```sh
git add armdb/src/fixed_replication/protocol.rs armdb/src/fixed_replication/mod.rs
git commit -m "feat(armdb/fixed_replication): FixedMessageType + Frame encode/decode"
```

---

### Task 16: Encode/decode `SyncRequest` and `ShardInfo`

**Files:**
- Modify: `armdb/src/fixed_replication/protocol.rs`

- [ ] **Step 16.1: Write failing tests**

Append to `mod tests`:

```rust
    #[test]
    fn test_sync_request_roundtrip() {
        let r = SyncRequest {
            shard_id: 7,
            protocol_version: PROTOCOL_VERSION,
            flags: FLAG_EMPTY_STATE,
        };
        let frame = r.encode();
        let decoded = SyncRequest::decode(&frame.payload).unwrap();
        assert_eq!(decoded.shard_id, 7);
        assert_eq!(decoded.protocol_version, PROTOCOL_VERSION);
        assert_eq!(decoded.flags, FLAG_EMPTY_STATE);
    }

    #[test]
    fn test_sync_request_truncated() {
        let err = SyncRequest::decode(&[1u8]).unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
    }

    #[test]
    fn test_shard_info_roundtrip() {
        let s = ShardInfo {
            shard_count: 16,
            key_len: 8,
            value_len: 32,
            slot_size: 48,
            current_slot_count: 1_000_000,
        };
        let frame = s.encode();
        let decoded = ShardInfo::decode(&frame.payload).unwrap();
        assert_eq!(decoded.shard_count, 16);
        assert_eq!(decoded.key_len, 8);
        assert_eq!(decoded.value_len, 32);
        assert_eq!(decoded.slot_size, 48);
        assert_eq!(decoded.current_slot_count, 1_000_000);
    }
```

- [ ] **Step 16.2: Add structs + encode/decode**

Append to `protocol.rs`:

```rust
/// F→L: ask leader to sync the given shard.
pub struct SyncRequest {
    pub shard_id: u8,
    pub protocol_version: u8,
    pub flags: u8, // FLAG_EMPTY_STATE etc.
}

impl SyncRequest {
    pub fn encode(&self) -> Frame {
        let payload = vec![self.shard_id, self.protocol_version, self.flags];
        Frame { msg_type: FixedMessageType::SyncRequest, payload }
    }
    pub fn decode(payload: &[u8]) -> io::Result<Self> {
        if payload.len() < 3 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "SyncRequest too short"));
        }
        Ok(Self {
            shard_id: payload[0],
            protocol_version: payload[1],
            flags: payload[2],
        })
    }
}

/// L→F: leader advertises shard topology after accepting a SyncRequest.
pub struct ShardInfo {
    pub shard_count: u8,
    pub key_len: u16,
    pub value_len: u16,
    pub slot_size: u16,
    pub current_slot_count: u32,
}

impl ShardInfo {
    pub fn encode(&self) -> Frame {
        let mut p = Vec::with_capacity(11);
        p.push(self.shard_count);
        p.extend_from_slice(&self.key_len.to_le_bytes());
        p.extend_from_slice(&self.value_len.to_le_bytes());
        p.extend_from_slice(&self.slot_size.to_le_bytes());
        p.extend_from_slice(&self.current_slot_count.to_le_bytes());
        Frame { msg_type: FixedMessageType::ShardInfo, payload: p }
    }
    pub fn decode(payload: &[u8]) -> io::Result<Self> {
        if payload.len() < 11 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "ShardInfo too short"));
        }
        Ok(Self {
            shard_count: payload[0],
            key_len: u16::from_le_bytes(payload[1..3].try_into().unwrap()),
            value_len: u16::from_le_bytes(payload[3..5].try_into().unwrap()),
            slot_size: u16::from_le_bytes(payload[5..7].try_into().unwrap()),
            current_slot_count: u32::from_le_bytes(payload[7..11].try_into().unwrap()),
        })
    }
}
```

- [ ] **Step 16.3: Run tests**

```sh
cargo nextest run -p armdb -E 'test(fixed_replication::protocol)'
```

- [ ] **Step 16.4: Commit**

```sh
git add armdb/src/fixed_replication/protocol.rs
git commit -m "feat(armdb/fixed_replication): encode/decode SyncRequest (w/ flags), ShardInfo"
```

---

### Task 17: Encode/decode `SlotBatch` (variable-length events)

**Files:**
- Modify: `armdb/src/fixed_replication/protocol.rs`

Per spec §6. Each event: `[slot_id:u32][meta:u32][key:K][value:V if status==OCCUPIED else 0 bytes]`.

- [ ] **Step 17.1: Write failing tests**

```rust
    #[test]
    fn test_slot_batch_mixed_roundtrip() {
        let key_len = 4usize;
        let value_len = 8usize;
        let mut batch = SlotBatchEncoder::new(5, key_len, value_len);
        // OCCUPIED event: meta with status_of = 1
        let occ_meta = crate::fixed::slot::pack_meta(crate::fixed::slot::STATUS_OCCUPIED, 42);
        batch.add_occupied(100, occ_meta, b"key0", b"12345678");
        // DELETED event
        let del_meta = crate::fixed::slot::pack_meta(crate::fixed::slot::STATUS_DELETED, 43);
        batch.add_deleted(101, del_meta, b"key1");
        let frame = batch.finish();
        assert_eq!(frame.msg_type, FixedMessageType::SlotBatch);

        let decoded = SlotBatchDecoder::new(&frame.payload, key_len, value_len).unwrap();
        assert_eq!(decoded.shard_id, 5);
        let events: Vec<_> = decoded.iter().collect::<Result<_, _>>().unwrap();
        assert_eq!(events.len(), 2);
        assert_eq!(events[0].slot_id, 100);
        assert_eq!(events[0].meta, occ_meta);
        assert_eq!(events[0].key, b"key0");
        assert_eq!(events[0].value, b"12345678");
        assert_eq!(events[1].slot_id, 101);
        assert_eq!(events[1].meta, del_meta);
        assert_eq!(events[1].key, b"key1");
        assert!(events[1].value.is_empty());
    }
```

- [ ] **Step 17.2: Implement `SlotBatchEncoder` and `SlotBatchDecoder`**

Append to `protocol.rs`:

```rust
use crate::fixed::slot::{status_of, STATUS_OCCUPIED};

/// Incremental encoder — caller adds events one at a time to avoid building
/// an intermediate Vec of event structs.
pub struct SlotBatchEncoder {
    payload: Vec<u8>,
    count: u32,
    shard_id: u8,
    key_len: usize,
    value_len: usize,
    count_offset: usize,
}

impl SlotBatchEncoder {
    pub fn new(shard_id: u8, key_len: usize, value_len: usize) -> Self {
        let mut payload = Vec::with_capacity(BATCH_MAX_BYTES);
        payload.push(shard_id);
        let count_offset = payload.len();
        payload.extend_from_slice(&0u32.to_le_bytes()); // placeholder count
        Self {
            payload,
            count: 0,
            shard_id,
            key_len,
            value_len,
            count_offset,
        }
    }

    pub fn add_occupied(&mut self, slot_id: u32, meta: u32, key: &[u8], value: &[u8]) {
        debug_assert_eq!(key.len(), self.key_len);
        debug_assert_eq!(value.len(), self.value_len);
        debug_assert_eq!(status_of(meta), STATUS_OCCUPIED);
        self.payload.extend_from_slice(&slot_id.to_le_bytes());
        self.payload.extend_from_slice(&meta.to_le_bytes());
        self.payload.extend_from_slice(key);
        self.payload.extend_from_slice(value);
        self.count += 1;
    }

    pub fn add_deleted(&mut self, slot_id: u32, meta: u32, key: &[u8]) {
        debug_assert_eq!(key.len(), self.key_len);
        self.payload.extend_from_slice(&slot_id.to_le_bytes());
        self.payload.extend_from_slice(&meta.to_le_bytes());
        self.payload.extend_from_slice(key);
        self.count += 1;
    }

    pub fn len(&self) -> u32 { self.count }
    pub fn bytes(&self) -> usize { self.payload.len() }
    pub fn is_empty(&self) -> bool { self.count == 0 }

    pub fn finish(mut self) -> Frame {
        self.payload[self.count_offset..self.count_offset + 4]
            .copy_from_slice(&self.count.to_le_bytes());
        Frame { msg_type: FixedMessageType::SlotBatch, payload: self.payload }
    }
}

/// Zero-copy decoder reading events out of a payload slice.
pub struct SlotBatchDecoder<'a> {
    pub shard_id: u8,
    pub count: u32,
    pub events_bytes: &'a [u8],
    key_len: usize,
    value_len: usize,
}

pub struct SlotEventRef<'a> {
    pub slot_id: u32,
    pub meta: u32,
    pub key: &'a [u8],
    pub value: &'a [u8], // empty if DELETED
}

impl<'a> SlotBatchDecoder<'a> {
    pub fn new(payload: &'a [u8], key_len: usize, value_len: usize) -> io::Result<Self> {
        if payload.len() < 5 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "SlotBatch too short"));
        }
        let shard_id = payload[0];
        let count = u32::from_le_bytes(payload[1..5].try_into().unwrap());
        Ok(Self {
            shard_id,
            count,
            events_bytes: &payload[5..],
            key_len,
            value_len,
        })
    }

    pub fn iter(&self) -> SlotEventIter<'a> {
        SlotEventIter {
            remaining: self.events_bytes,
            left: self.count,
            key_len: self.key_len,
            value_len: self.value_len,
        }
    }
}

pub struct SlotEventIter<'a> {
    remaining: &'a [u8],
    left: u32,
    key_len: usize,
    value_len: usize,
}

impl<'a> Iterator for SlotEventIter<'a> {
    type Item = io::Result<SlotEventRef<'a>>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.left == 0 {
            return None;
        }
        if self.remaining.len() < 8 + self.key_len {
            return Some(Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "SlotBatch event truncated at header/key",
            )));
        }
        let slot_id = u32::from_le_bytes(self.remaining[0..4].try_into().unwrap());
        let meta = u32::from_le_bytes(self.remaining[4..8].try_into().unwrap());
        let key_start = 8;
        let key_end = key_start + self.key_len;
        let key = &self.remaining[key_start..key_end];
        let value_len_if_occ = if status_of(meta) == STATUS_OCCUPIED { self.value_len } else { 0 };
        let value_end = key_end + value_len_if_occ;
        if self.remaining.len() < value_end {
            return Some(Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "SlotBatch event truncated at value",
            )));
        }
        let value = &self.remaining[key_end..value_end];
        self.remaining = &self.remaining[value_end..];
        self.left -= 1;
        Some(Ok(SlotEventRef { slot_id, meta, key, value }))
    }
}
```

- [ ] **Step 17.3: Run tests**

```sh
cargo nextest run -p armdb -E 'test(fixed_replication::protocol)'
```

- [ ] **Step 17.4: Commit**

```sh
git add armdb/src/fixed_replication/protocol.rs
git commit -m "feat(armdb/fixed_replication): SlotBatchEncoder + zero-copy SlotBatchDecoder"
```

---

### Task 18: Encode/decode `CaughtUp`, `Ack`, `Heartbeat`, `Error`

**Files:**
- Modify: `armdb/src/fixed_replication/protocol.rs`

- [ ] **Step 18.1: Append structs + tests**

```rust
pub struct CaughtUp {
    pub shard_id: u8,
    pub total_scanned: u64,
}

impl CaughtUp {
    pub fn encode(&self) -> Frame {
        let mut p = Vec::with_capacity(9);
        p.push(self.shard_id);
        p.extend_from_slice(&self.total_scanned.to_le_bytes());
        Frame { msg_type: FixedMessageType::CaughtUp, payload: p }
    }
    pub fn decode(payload: &[u8]) -> io::Result<Self> {
        if payload.len() < 9 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "CaughtUp too short"));
        }
        Ok(Self {
            shard_id: payload[0],
            total_scanned: u64::from_le_bytes(payload[1..9].try_into().unwrap()),
        })
    }
}

pub struct Ack {
    pub shard_id: u8,
    pub applied_count: u64,
    pub max_version_seen: u32,
}

impl Ack {
    pub fn encode(&self) -> Frame {
        let mut p = Vec::with_capacity(13);
        p.push(self.shard_id);
        p.extend_from_slice(&self.applied_count.to_le_bytes());
        p.extend_from_slice(&self.max_version_seen.to_le_bytes());
        Frame { msg_type: FixedMessageType::Ack, payload: p }
    }
    pub fn decode(payload: &[u8]) -> io::Result<Self> {
        if payload.len() < 13 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "Ack too short"));
        }
        Ok(Self {
            shard_id: payload[0],
            applied_count: u64::from_le_bytes(payload[1..9].try_into().unwrap()),
            max_version_seen: u32::from_le_bytes(payload[9..13].try_into().unwrap()),
        })
    }
}

pub fn encode_heartbeat() -> Frame {
    Frame { msg_type: FixedMessageType::Heartbeat, payload: Vec::new() }
}

pub fn encode_error(msg: &str) -> Frame {
    Frame { msg_type: FixedMessageType::Error, payload: msg.as_bytes().to_vec() }
}

pub fn decode_error(payload: &[u8]) -> String {
    String::from_utf8_lossy(payload).into_owned()
}
```

Tests:

```rust
    #[test]
    fn test_caught_up_roundtrip() {
        let c = CaughtUp { shard_id: 3, total_scanned: 123_456_789 };
        let f = c.encode();
        let d = CaughtUp::decode(&f.payload).unwrap();
        assert_eq!(d.shard_id, 3);
        assert_eq!(d.total_scanned, 123_456_789);
    }

    #[test]
    fn test_ack_roundtrip() {
        let a = Ack { shard_id: 4, applied_count: 1000, max_version_seen: 42 };
        let f = a.encode();
        let d = Ack::decode(&f.payload).unwrap();
        assert_eq!(d.shard_id, 4);
        assert_eq!(d.applied_count, 1000);
        assert_eq!(d.max_version_seen, 42);
    }
```

- [ ] **Step 18.2: Run tests**

```sh
cargo nextest run -p armdb -E 'test(fixed_replication::protocol)'
```

- [ ] **Step 18.3: Commit**

```sh
git add armdb/src/fixed_replication/protocol.rs
git commit -m "feat(armdb/fixed_replication): CaughtUp, Ack, Heartbeat, Error messages"
```

---

## Phase 7: Cursor (diagnostic)

### Task 19: `FixedReplicationCursor` save/load

**Files:**
- Create: `armdb/src/fixed_replication/cursor.rs`
- Modify: `armdb/src/fixed_replication/mod.rs`

Per spec §8. Diagnostic only — does **not** affect correctness.

- [ ] **Step 19.1: Write struct + save/load + tests**

Write to `armdb/src/fixed_replication/cursor.rs`:

```rust
//! Per-shard diagnostic cursor for the FixedStore replication follower.
//!
//! Unlike Bitcask's cursor (which resumes from a GSN/file offset), the
//! FixedStore cursor does NOT influence correctness — catch-up is always
//! a full slot scan on reconnect. The cursor exists purely for diagnostics:
//! applied event count, max version observed, last-Ack timestamp.

use std::fs;
use std::path::Path;
use std::time::{Duration, SystemTime};

use crate::error::DbResult;

pub const CURSOR_FILE: &str = "fixed_repl.cursor";
pub const CURSOR_SIZE: usize = 40;
// Layout: shard_id(1) + pad(3) + applied_total(8) + max_version(4) +
//         pad(4) + unix_ms(8) + reserved(12).
pub const CURSOR_SAVE_INTERVAL: usize = 1000;

pub struct FixedReplicationCursor {
    pub shard_id: u8,
    pub applied_total: u64,
    pub max_version_seen: u32,
    pub last_ack_at: SystemTime,
}

impl FixedReplicationCursor {
    pub fn new(shard_id: u8) -> Self {
        Self {
            shard_id,
            applied_total: 0,
            max_version_seen: 0,
            last_ack_at: SystemTime::UNIX_EPOCH,
        }
    }

    pub fn load(shard_id: u8, shard_dir: &Path) -> Option<Self> {
        let path = shard_dir.join(CURSOR_FILE);
        let data = fs::read(&path).ok()?;
        if data.len() < CURSOR_SIZE {
            return None;
        }
        let applied_total = u64::from_le_bytes(data[4..12].try_into().ok()?);
        let max_version_seen = u32::from_le_bytes(data[12..16].try_into().ok()?);
        let unix_ms = u64::from_le_bytes(data[20..28].try_into().ok()?);
        let last_ack_at = SystemTime::UNIX_EPOCH + Duration::from_millis(unix_ms);
        Some(Self {
            shard_id,
            applied_total,
            max_version_seen,
            last_ack_at,
        })
    }

    pub fn save(&self, shard_dir: &Path) -> DbResult<()> {
        let path = shard_dir.join(CURSOR_FILE);
        let mut buf = [0u8; CURSOR_SIZE];
        buf[0] = self.shard_id;
        buf[4..12].copy_from_slice(&self.applied_total.to_le_bytes());
        buf[12..16].copy_from_slice(&self.max_version_seen.to_le_bytes());
        let unix_ms = self
            .last_ack_at
            .duration_since(SystemTime::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0);
        buf[20..28].copy_from_slice(&unix_ms.to_le_bytes());
        fs::write(&path, buf)?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::tempdir;

    #[test]
    fn test_save_load_roundtrip() {
        let dir = tempdir().unwrap();
        let mut c = FixedReplicationCursor::new(3);
        c.applied_total = 42;
        c.max_version_seen = 0xABCD;
        c.last_ack_at = SystemTime::UNIX_EPOCH + Duration::from_millis(1_700_000_000_000);
        c.save(dir.path()).unwrap();

        let loaded = FixedReplicationCursor::load(3, dir.path()).unwrap();
        assert_eq!(loaded.shard_id, 3);
        assert_eq!(loaded.applied_total, 42);
        assert_eq!(loaded.max_version_seen, 0xABCD);
        assert_eq!(loaded.last_ack_at, c.last_ack_at);
    }

    #[test]
    fn test_load_absent() {
        let dir = tempdir().unwrap();
        assert!(FixedReplicationCursor::load(0, dir.path()).is_none());
    }

    #[test]
    fn test_load_truncated() {
        let dir = tempdir().unwrap();
        std::fs::write(dir.path().join(CURSOR_FILE), &[0u8; 10]).unwrap();
        assert!(FixedReplicationCursor::load(0, dir.path()).is_none());
    }
}
```

Add `pub mod cursor;` and `pub use cursor::FixedReplicationCursor;` to `mod.rs`.

- [ ] **Step 19.2: Run**

```sh
cargo nextest run -p armdb -E 'test(fixed_replication::cursor)'
```

- [ ] **Step 19.3: Commit**

```sh
git add armdb/src/fixed_replication/cursor.rs armdb/src/fixed_replication/mod.rs
git commit -m "feat(armdb/fixed_replication): FixedReplicationCursor (diagnostic-only)"
```

---

## Phase 8: Apply path (`FixedReplicationTarget`)

### Task 20: Define `FixedReplicationTarget` trait

**Files:**
- Create: `armdb/src/fixed_replication/apply.rs`
- Modify: `armdb/src/fixed_replication/mod.rs`

Per spec §8.

- [ ] **Step 20.1: Write trait + enum**

Write to `armdb/src/fixed_replication/apply.rs`:

```rust
//! Follower apply trait and blanket implementations for
//! `ConstTree<K, V, H, Fixed>` and `ConstMap<K, V, H, Fixed>`.

use std::path::PathBuf;

use crate::error::DbResult;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApplyOutcome {
    /// Leader's version is newer — event applied.
    Applied,
    /// Own version ≥ incoming (modular compare) — event skipped.
    Skipped,
}

pub trait FixedReplicationTarget: Send + Sync {
    fn shard_count(&self) -> usize;
    fn key_len(&self) -> usize;
    fn value_len(&self) -> usize;

    /// Current number of OCCUPIED slots for `shard_id`. Used by the client
    /// to decide whether to set `FLAG_EMPTY_STATE` in SyncRequest.
    fn shard_occupied_count(&self, shard_id: u8) -> u32;

    /// Directory for per-shard cursor files.
    fn shard_dir(&self, shard_id: u8) -> PathBuf;

    /// Grow the shard to at least `min_slot_count` slots.
    /// No-op when follower already has more slots.
    fn grow_shard_to(&self, shard_id: u8, min_slot_count: u32) -> DbResult<()>;

    /// Apply an OCCUPIED slot event. Handles modular version compare,
    /// stale-cleanup of old key (rare path), slot pwrite, and index upsert.
    fn apply_occupied(
        &self,
        shard_id: u8,
        slot_id: u32,
        meta: u32,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<ApplyOutcome>;

    /// Apply a DELETED slot event. Conditional `index.remove` — only if
    /// the index still points to this slot_id.
    fn apply_deleted(
        &self,
        shard_id: u8,
        slot_id: u32,
        meta: u32,
        key: &[u8],
    ) -> DbResult<ApplyOutcome>;
}
```

Add to `armdb/src/fixed_replication/mod.rs`:
```rust
pub mod apply;
pub use apply::{ApplyOutcome, FixedReplicationTarget};
```

- [ ] **Step 20.2: Compile check**

```sh
cargo check -p armdb
```

- [ ] **Step 20.3: Commit**

```sh
git add armdb/src/fixed_replication/apply.rs armdb/src/fixed_replication/mod.rs
git commit -m "feat(armdb/fixed_replication): FixedReplicationTarget trait + ApplyOutcome"
```

---

### Task 21: Add `pub(crate)` helpers on `ConstTree` / `ConstMap` needed for replication apply

**Files:**
- Modify: `armdb/src/const_tree.rs`
- Modify: `armdb/src/const_map.rs`
- Modify: `armdb/src/skiplist/mod.rs` (possibly)

We need index operations beyond the current public API:
- `get_slot_id(&key) -> Option<u32>` — read-only lookup of the slot_id behind a key.
- `remove_key_if_slot_matches(&key, slot_id) -> bool` — conditional remove (for Bug 1 fix).
- `upsert_replicated(&key, value, slot_id)` — insert-or-update without triggering the write path (no pwrite; disk is already updated by `apply_foreign_slot`).

These all operate on the in-memory index only and don't touch `durability`.

- [ ] **Step 21.1: Inspect current index API**

Open `armdb/src/const_tree.rs` and `armdb/src/skiplist/mod.rs`. Locate:
- The SkipList `get`/`insert`/`remove` methods on `self.index`.
- How values are inlined in `ConstNode<K, V, L>`.

For SkipList (in `skiplist/mod.rs` or similar), find methods like:
```rust
pub fn get(&self, key: &[u8], guard: &Guard) -> Option<&ConstNode<...>>;
pub fn insert(&self, key: &[u8], node: ...);
pub fn remove(&self, key: &[u8]) -> bool;
```

And on `ConstNode`:
```rust
pub fn read_value(&self) -> [u8; V];
pub fn write_value(&self, value: [u8; V]);  // under SeqLock
pub fn loc(&self) -> L;
pub fn set_loc(&self, loc: L);
```

If these exact methods don't exist, check closest analogs and add shim methods named appropriately.

**Deliverable for this step:** make a short note (can be a code comment in a file scratch location) listing which methods exist vs need creating.

- [ ] **Step 21.2: Add `get_slot_id` to `ConstTree<Fixed>`**

In `armdb/src/const_tree.rs`, inside the generic impl block (or a new `#[cfg(feature = "replication")]` block specialized for `Fixed`):

```rust
impl<K: Key, const V: usize, H: WriteHook<K>> ConstTree<K, V, H, crate::durability::Fixed> {
    #[cfg(feature = "replication")]
    pub(crate) fn get_slot_id(&self, key: &K) -> Option<u32> {
        let guard = self.index.collector().enter();
        let node = self.index.get(key.as_bytes(), &guard)?;
        Some(node.loc())
    }

    #[cfg(feature = "replication")]
    pub(crate) fn remove_key_if_slot_matches(&self, key: &K, slot_id: u32) -> bool {
        let guard = self.index.collector().enter();
        match self.index.get(key.as_bytes(), &guard) {
            Some(node) if node.loc() == slot_id => {
                drop(node);
                self.index.remove(key.as_bytes());
                true
            }
            _ => false,
        }
    }

    #[cfg(feature = "replication")]
    pub(crate) fn upsert_replicated(&self, key: &K, value: [u8; V], slot_id: u32) {
        let guard = self.index.collector().enter();
        if let Some(node) = self.index.get(key.as_bytes(), &guard) {
            if node.loc() == slot_id {
                node.write_value(value);
                return;
            }
            // Key moved to different slot; remove then insert.
            drop(node);
            self.index.remove(key.as_bytes());
        }
        self.index.insert_with_loc(key, value, slot_id);
    }
}
```

If `ConstNode::loc()`, `write_value`, or `SkipList::insert_with_loc` don't exist, add them. They're straightforward — `loc` mirrors the existing `DiskLoc` accessor, generic over `L`. `write_value` exists (SeqLock write). `insert_with_loc` may need adding — an insert that takes both value and location rather than going through the usual insert/put path.

- [ ] **Step 21.3: Add equivalents on `ConstMap<Fixed>`**

Symmetric impl in `armdb/src/const_map.rs`:

```rust
impl<K: Key, const V: usize, H: WriteHook<K>> ConstMap<K, V, H, crate::durability::Fixed> {
    #[cfg(feature = "replication")]
    pub(crate) fn get_slot_id(&self, key: &K) -> Option<u32> {
        let guard = self.index.collector().enter();
        let node = self.index.get(key.as_bytes(), &guard)?;
        Some(node.loc())
    }

    #[cfg(feature = "replication")]
    pub(crate) fn remove_key_if_slot_matches(&self, key: &K, slot_id: u32) -> bool {
        let guard = self.index.collector().enter();
        match self.index.get(key.as_bytes(), &guard) {
            Some(node) if node.loc() == slot_id => {
                drop(node);
                self.index.remove(key.as_bytes());
                true
            }
            _ => false,
        }
    }

    #[cfg(feature = "replication")]
    pub(crate) fn upsert_replicated(&self, key: &K, value: [u8; V], slot_id: u32) {
        let guard = self.index.collector().enter();
        if let Some(node) = self.index.get(key.as_bytes(), &guard) {
            if node.loc() == slot_id {
                node.write_value(value);
                return;
            }
            drop(node);
            self.index.remove(key.as_bytes());
        }
        self.index.insert_with_loc(key, value, slot_id);
    }
}
```

`ConstMap` uses a HashMap-backed index — likely needs its own `insert_with_loc` helper.

- [ ] **Step 21.4: Add unit tests**

In `armdb/src/const_tree.rs` `mod tests` (or inline near the new methods):

```rust
    #[cfg(feature = "replication")]
    #[test]
    fn test_get_slot_id_remove_conditional_upsert() {
        use crate::fixed::config::FixedConfig;
        use crate::Fuid;
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("db");
        let fixed = crate::durability::Fixed::open(
            &path,
            FixedConfig::test(),
            std::mem::size_of::<Fuid>(),
            4,
        ).unwrap();
        // ... construct ConstTree<Fuid, 4, NoHook, Fixed> — reuse existing test helpers.
        // Write key k1 at slot S1, verify get_slot_id returns S1.
        // Call remove_key_if_slot_matches(k1, S1 + 1) → false (slot mismatch).
        // Call remove_key_if_slot_matches(k1, S1) → true, key removed.
        // Call upsert_replicated(k2, value, S2) → index contains k2 → S2.
    }
```

The concrete test body mirrors existing ConstTree tests — adapt from `const_tree.rs` existing test helpers.

- [ ] **Step 21.5: Run tests + cargo check**

```sh
cargo check -p armdb
cargo nextest run -p armdb -E 'test(const_tree) and test(replicated)'
```

- [ ] **Step 21.6: Commit**

```sh
git add armdb/src/const_tree.rs armdb/src/const_map.rs armdb/src/skiplist
git commit -m "feat(armdb): add pub(crate) index helpers for replication apply path"
```

---

### Task 22: Implement `FixedReplicationTarget` for `ConstTree<Fixed>`

**Files:**
- Modify: `armdb/src/fixed_replication/apply.rs`

Per spec §8 — with Bug 1 / Bug 2 fixes.

- [ ] **Step 22.1: Add impl**

Append to `armdb/src/fixed_replication/apply.rs`:

```rust
use std::path::PathBuf;

use crate::const_tree::ConstTree;
use crate::durability::{Durability, Fixed};
use crate::fixed::slot::{self, is_newer, status_of, version_of, STATUS_OCCUPIED};
use crate::hook::WriteHook;
use crate::key::Key;

impl<K, const V: usize, H> FixedReplicationTarget for ConstTree<K, V, H, Fixed>
where
    K: Key + 'static,
    H: WriteHook<K> + 'static,
{
    fn shard_count(&self) -> usize {
        self.durability.shard_count()
    }
    fn key_len(&self) -> usize {
        K::LEN
    }
    fn value_len(&self) -> usize {
        V
    }

    fn shard_occupied_count(&self, shard_id: u8) -> u32 {
        let inner = self.durability.lock_shard(shard_id as usize);
        inner.bitmap.occupied()
    }

    fn shard_dir(&self, shard_id: u8) -> PathBuf {
        self.durability.engine.path().join(format!("shard_{:03}", shard_id))
    }

    fn grow_shard_to(&self, shard_id: u8, min_slot_count: u32) -> DbResult<()> {
        let mut inner = self.durability.lock_shard(shard_id as usize);
        while inner.slot_count() < min_slot_count {
            inner.grow()?;
        }
        if inner.slot_count() > min_slot_count {
            tracing::warn!(
                shard_id,
                follower_count = inner.slot_count(),
                leader_count = min_slot_count,
                "follower has more slots than leader — ignoring (no truncate)"
            );
        }
        Ok(())
    }

    fn apply_occupied(
        &self,
        shard_id: u8,
        slot_id: u32,
        meta: u32,
        key: &[u8],
        value: &[u8],
    ) -> DbResult<ApplyOutcome> {
        debug_assert_eq!(key.len(), K::LEN);
        debug_assert_eq!(value.len(), V);

        let key_typed = K::from_bytes(key);
        let mut value_arr = [0u8; V];
        value_arr.copy_from_slice(value);

        let mut inner = self.durability.lock_shard(shard_id as usize);
        while inner.slot_count() <= slot_id {
            inner.grow()?;
        }

        // Modular version compare.
        let old_meta = inner.versions[slot_id as usize];
        if !is_newer(version_of(meta), version_of(old_meta)) {
            metrics::counter!("armdb.fixed.skipped_events",
                "shard" => shard_id.to_string()).increment(1);
            return Ok(ApplyOutcome::Skipped);
        }

        // Step 1: stale cleanup of old key if slot was previously OCCUPIED
        // with a different key (Bug 2 fix — orthogonal to where new_key is).
        if status_of(old_meta) == STATUS_OCCUPIED
            && self.get_slot_id(&key_typed) != Some(slot_id)
        {
            let buf = inner.read_slot_header_and_key(slot_id, K::LEN)?;
            let old_key_bytes = &buf[slot::SLOT_HEADER_SIZE..slot::SLOT_HEADER_SIZE + K::LEN];
            let old_key = K::from_bytes(old_key_bytes);
            if old_key.as_bytes() != key_typed.as_bytes() {
                // Only remove if the index still points to this slot_id
                // (old_key may have "moved" to a different slot already).
                let removed = self.remove_key_if_slot_matches(&old_key, slot_id);
                if removed {
                    metrics::counter!("armdb.fixed.stale_cleanup_count",
                        "shard" => shard_id.to_string()).increment(1);
                }
            }
        }

        // Step 2: persist slot.
        inner.apply_foreign_slot(slot_id, meta, key, value)?;
        inner.bitmap.set(slot_id);

        // Step 3: upsert index.
        self.upsert_replicated(&key_typed, value_arr, slot_id);

        metrics::counter!("armdb.fixed.applied_events",
            "shard" => shard_id.to_string(),
            "kind" => "occupied").increment(1);
        Ok(ApplyOutcome::Applied)
    }

    fn apply_deleted(
        &self,
        shard_id: u8,
        slot_id: u32,
        meta: u32,
        key: &[u8],
    ) -> DbResult<ApplyOutcome> {
        debug_assert_eq!(key.len(), K::LEN);
        let key_typed = K::from_bytes(key);

        let mut inner = self.durability.lock_shard(shard_id as usize);
        while inner.slot_count() <= slot_id {
            inner.grow()?;
        }

        let old_meta = inner.versions[slot_id as usize];
        if !is_newer(version_of(meta), version_of(old_meta)) {
            metrics::counter!("armdb.fixed.skipped_events",
                "shard" => shard_id.to_string()).increment(1);
            return Ok(ApplyOutcome::Skipped);
        }

        inner.apply_foreign_delete(slot_id, meta)?;
        inner.bitmap.clear(slot_id);

        // Bug 1 fix: only remove from index if it still points to this slot.
        // Otherwise we'd wipe a valid mapping for a key that moved elsewhere.
        let _removed = self.remove_key_if_slot_matches(&key_typed, slot_id);

        metrics::counter!("armdb.fixed.applied_events",
            "shard" => shard_id.to_string(),
            "kind" => "deleted").increment(1);
        Ok(ApplyOutcome::Applied)
    }
}
```

Notes:
- `self.durability.engine.path()` — needs `FixedEngine::path()` to be `pub(crate)` visible to this module. Already is.
- `K::LEN` and `K::from_bytes` assumed; check `armdb::key::Key` trait.
- `inner.bitmap.occupied()` — verify bitmap has this counter (it does; saw in `bitmap.rs`).

If any of the assumed methods don't exist, add them (e.g., `Key::LEN` may be `Key::SIZE`; adjust).

- [ ] **Step 22.2: Compile check**

```sh
cargo check -p armdb --all-features
```

Fix any name mismatches surfaced by the compiler.

- [ ] **Step 22.3: Commit**

```sh
git add armdb/src/fixed_replication/apply.rs
git commit -m "feat(armdb/fixed_replication): apply_occupied/deleted for ConstTree<Fixed>"
```

---

### Task 23: Implement `FixedReplicationTarget` for `ConstMap<Fixed>`

**Files:**
- Modify: `armdb/src/fixed_replication/apply.rs`

- [ ] **Step 23.1: Add impl (symmetric to `ConstTree`)**

Append:

```rust
use crate::const_map::ConstMap;

impl<K, const V: usize, H> FixedReplicationTarget for ConstMap<K, V, H, Fixed>
where
    K: Key + 'static,
    H: WriteHook<K> + 'static,
{
    fn shard_count(&self) -> usize { self.durability.shard_count() }
    fn key_len(&self) -> usize { K::LEN }
    fn value_len(&self) -> usize { V }

    fn shard_occupied_count(&self, shard_id: u8) -> u32 {
        self.durability.lock_shard(shard_id as usize).bitmap.occupied()
    }
    fn shard_dir(&self, shard_id: u8) -> PathBuf {
        self.durability.engine.path().join(format!("shard_{:03}", shard_id))
    }
    fn grow_shard_to(&self, shard_id: u8, min_slot_count: u32) -> DbResult<()> {
        let mut inner = self.durability.lock_shard(shard_id as usize);
        while inner.slot_count() < min_slot_count {
            inner.grow()?;
        }
        Ok(())
    }

    fn apply_occupied(&self, shard_id: u8, slot_id: u32, meta: u32, key: &[u8], value: &[u8])
        -> DbResult<ApplyOutcome>
    {
        // Same logic as ConstTree; copy the body adapting the index ops
        // to ConstMap methods (get_slot_id / remove_key_if_slot_matches /
        // upsert_replicated defined in Task 21).
        // ... (paste & adjust — identical structure to ConstTree) ...
        todo!("copy from ConstTree apply_occupied and use self's index helpers")
    }

    fn apply_deleted(&self, shard_id: u8, slot_id: u32, meta: u32, key: &[u8])
        -> DbResult<ApplyOutcome>
    {
        // Same logic as ConstTree::apply_deleted.
        todo!("copy from ConstTree apply_deleted")
    }
}
```

**Replace the `todo!()` with actual copy-paste of the `ConstTree` bodies.** DO NOT leave `todo!()` — reviewer catches placeholder code. Concrete expansion:

```rust
    fn apply_occupied(
        &self, shard_id: u8, slot_id: u32, meta: u32, key: &[u8], value: &[u8],
    ) -> DbResult<ApplyOutcome> {
        debug_assert_eq!(key.len(), K::LEN);
        debug_assert_eq!(value.len(), V);
        let key_typed = K::from_bytes(key);
        let mut value_arr = [0u8; V];
        value_arr.copy_from_slice(value);
        let mut inner = self.durability.lock_shard(shard_id as usize);
        while inner.slot_count() <= slot_id { inner.grow()?; }
        let old_meta = inner.versions[slot_id as usize];
        if !is_newer(version_of(meta), version_of(old_meta)) {
            metrics::counter!("armdb.fixed.skipped_events",
                "shard" => shard_id.to_string()).increment(1);
            return Ok(ApplyOutcome::Skipped);
        }
        if status_of(old_meta) == STATUS_OCCUPIED
            && self.get_slot_id(&key_typed) != Some(slot_id)
        {
            let buf = inner.read_slot_header_and_key(slot_id, K::LEN)?;
            let old_key_bytes = &buf[slot::SLOT_HEADER_SIZE..slot::SLOT_HEADER_SIZE + K::LEN];
            let old_key = K::from_bytes(old_key_bytes);
            if old_key.as_bytes() != key_typed.as_bytes()
                && self.remove_key_if_slot_matches(&old_key, slot_id)
            {
                metrics::counter!("armdb.fixed.stale_cleanup_count",
                    "shard" => shard_id.to_string()).increment(1);
            }
        }
        inner.apply_foreign_slot(slot_id, meta, key, value)?;
        inner.bitmap.set(slot_id);
        self.upsert_replicated(&key_typed, value_arr, slot_id);
        metrics::counter!("armdb.fixed.applied_events",
            "shard" => shard_id.to_string(),
            "kind" => "occupied").increment(1);
        Ok(ApplyOutcome::Applied)
    }

    fn apply_deleted(
        &self, shard_id: u8, slot_id: u32, meta: u32, key: &[u8],
    ) -> DbResult<ApplyOutcome> {
        debug_assert_eq!(key.len(), K::LEN);
        let key_typed = K::from_bytes(key);
        let mut inner = self.durability.lock_shard(shard_id as usize);
        while inner.slot_count() <= slot_id { inner.grow()?; }
        let old_meta = inner.versions[slot_id as usize];
        if !is_newer(version_of(meta), version_of(old_meta)) {
            metrics::counter!("armdb.fixed.skipped_events",
                "shard" => shard_id.to_string()).increment(1);
            return Ok(ApplyOutcome::Skipped);
        }
        inner.apply_foreign_delete(slot_id, meta)?;
        inner.bitmap.clear(slot_id);
        let _ = self.remove_key_if_slot_matches(&key_typed, slot_id);
        metrics::counter!("armdb.fixed.applied_events",
            "shard" => shard_id.to_string(),
            "kind" => "deleted").increment(1);
        Ok(ApplyOutcome::Applied)
    }
```

- [ ] **Step 23.2: Compile check**

```sh
cargo check -p armdb --all-features
```

- [ ] **Step 23.3: Commit**

```sh
git add armdb/src/fixed_replication/apply.rs
git commit -m "feat(armdb/fixed_replication): apply impl for ConstMap<Fixed>"
```

---

## Phase 9: Server

### Task 24: `FixedReplicationServer` skeleton with acceptor + lazy producer install

**Files:**
- Create: `armdb/src/fixed_replication/server.rs`
- Modify: `armdb/src/fixed_replication/mod.rs`

Per spec §7.

- [ ] **Step 24.1: Write skeleton**

Write to `armdb/src/fixed_replication/server.rs`:

```rust
//! Leader-side replication server for FixedStore.
//!
//! - Spawns a TCP acceptor thread.
//! - On the first accepted follower connection, installs SPSC producers
//!   into every shard (lazy install — Mitigation B of spec §5/§7).
//! - Each accepted connection spawns a per-shard serve thread that does
//!   Phase 1 (full scan catch-up) then Phase 2 (SPSC streaming) if the
//!   per-shard consumer is still available.

use std::io::{BufWriter, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use rtrb::{Consumer, Producer, RingBuffer};

use crate::error::{DbError, DbResult};
use crate::shutdown::ShutdownSignal;

use super::engine_access::ArcEngine;
use super::event::FixedReplicationEvent;
use super::protocol::*;

pub const SPSC_CAPACITY: usize = 8192;

type PendingSlot = crate::sync::Mutex<(Option<Producer<FixedReplicationEvent>>, Option<Consumer<FixedReplicationEvent>>)>;

pub struct FixedReplicationServer {
    stop: ShutdownSignal,
    acceptor_handle: Option<JoinHandle<()>>,
    handler_handles: Arc<crate::sync::Mutex<Vec<JoinHandle<()>>>>,
    producers_installed: Arc<AtomicBool>,
}

impl FixedReplicationServer {
    pub fn start(
        bind_addr: SocketAddr,
        engine: ArcEngine,
        signal: ShutdownSignal,
    ) -> DbResult<Self> {
        let shard_count = engine.shard_count();
        let mut pending: Vec<PendingSlot> = Vec::with_capacity(shard_count);
        for _ in 0..shard_count {
            let (p, c) = RingBuffer::new(SPSC_CAPACITY);
            pending.push(crate::sync::Mutex::new((Some(p), Some(c))));
        }
        let pending = Arc::new(pending);
        let producers_installed = Arc::new(AtomicBool::new(false));
        let handler_handles = Arc::new(crate::sync::Mutex::new(Vec::new()));

        let listener = TcpListener::bind(bind_addr).map_err(DbError::from)?;
        listener.set_nonblocking(true).ok();

        let acceptor_handle = {
            let engine = engine.clone();
            let pending = pending.clone();
            let producers_installed = producers_installed.clone();
            let stop = signal.clone();
            let hh = handler_handles.clone();
            thread::spawn(move || {
                acceptor_loop(listener, engine, pending, producers_installed, hh, stop);
            })
        };

        Ok(Self {
            stop: signal,
            acceptor_handle: Some(acceptor_handle),
            handler_handles,
            producers_installed,
        })
    }

    pub fn stop(&self) {
        self.stop.shutdown();
    }
}

impl Drop for FixedReplicationServer {
    fn drop(&mut self) {
        self.stop.shutdown();
        if let Some(h) = self.acceptor_handle.take() {
            let _ = h.join();
        }
        let mut handles = crate::sync::lock(&self.handler_handles);
        for h in handles.drain(..) {
            let _ = h.join();
        }
    }
}

fn acceptor_loop(
    listener: TcpListener,
    engine: ArcEngine,
    pending: Arc<Vec<PendingSlot>>,
    producers_installed: Arc<AtomicBool>,
    handler_handles: Arc<crate::sync::Mutex<Vec<JoinHandle<()>>>>,
    stop: ShutdownSignal,
) {
    while !stop.is_shutdown() {
        match listener.accept() {
            Ok((stream, addr)) => {
                tracing::info!(%addr, "follower connected");
                stream.set_nodelay(true).ok();

                if !producers_installed.swap(true, Ordering::SeqCst) {
                    tracing::info!("first follower — installing SPSC producers");
                    let mut producers = Vec::with_capacity(pending.len());
                    for slot in pending.iter() {
                        let mut guard = crate::sync::lock(slot);
                        producers.push(guard.0.take().expect("producer present on first install"));
                    }
                    engine.install_replication_producers(producers);
                }

                handle_connection(stream, engine.clone(), pending.clone(), stop.clone(), handler_handles.clone());
            }
            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                stop.wait_timeout(Duration::from_millis(50));
            }
            Err(e) => {
                tracing::error!(error = %e, "accept error");
                stop.wait_timeout(Duration::from_millis(100));
            }
        }
    }
    tracing::info!("fixed replication acceptor stopped");
}

fn handle_connection(
    stream: TcpStream,
    engine: ArcEngine,
    pending: Arc<Vec<PendingSlot>>,
    stop: ShutdownSignal,
    handler_handles: Arc<crate::sync::Mutex<Vec<JoinHandle<()>>>>,
) {
    let handle = thread::spawn(move || {
        if let Err(e) = serve_connection(stream, engine, pending, stop) {
            tracing::error!(error = %e, "fixed replication connection error");
        }
    });
    crate::sync::lock(&handler_handles).push(handle);
}

fn serve_connection(
    stream: TcpStream,
    engine: ArcEngine,
    pending: Arc<Vec<PendingSlot>>,
    stop: ShutdownSignal,
) -> DbResult<()> {
    // Implementation in Task 25.
    let _ = (stream, engine, pending, stop);
    Ok(())
}
```

Update `mod.rs`:
```rust
pub mod server;
pub use server::FixedReplicationServer;
```

- [ ] **Step 24.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 24.3: Commit**

```sh
git add armdb/src/fixed_replication/server.rs armdb/src/fixed_replication/mod.rs
git commit -m "feat(armdb/fixed_replication): server skeleton with lazy SPSC install"
```

---

### Task 25: Per-connection handler — SyncRequest → ShardInfo → spawn serve thread

**Files:**
- Modify: `armdb/src/fixed_replication/server.rs`

- [ ] **Step 25.1: Fill `serve_connection`**

Replace the skeleton body:

```rust
fn serve_connection(
    stream: TcpStream,
    engine: ArcEngine,
    pending: Arc<Vec<PendingSlot>>,
    stop: ShutdownSignal,
) -> DbResult<()> {
    let mut reader = stream.try_clone().map_err(DbError::from)?;
    let mut writer = BufWriter::new(stream);

    // Read initial SyncRequest.
    let frame = read_frame(&mut reader).map_err(DbError::from)?;
    if frame.msg_type != FixedMessageType::SyncRequest {
        return Err(DbError::Replication(format!(
            "expected SyncRequest, got {:?}", frame.msg_type
        )));
    }
    let req = SyncRequest::decode(&frame.payload).map_err(DbError::from)?;
    let shard_id = req.shard_id as usize;
    if shard_id >= engine.shard_count() {
        write_frame(&mut writer, &encode_error("invalid shard_id"))
            .map_err(DbError::from)?;
        return Err(DbError::Replication(format!("invalid shard_id {shard_id}")));
    }

    // Send ShardInfo.
    let info = ShardInfo {
        shard_count: engine.shard_count() as u8,
        key_len: engine.key_len() as u16,
        value_len: engine.value_len() as u16,
        slot_size: engine.slot_size(),
        current_slot_count: engine.current_slot_count(shard_id),
    };
    write_frame(&mut writer, &info.encode()).map_err(DbError::from)?;

    let skip_deleted = (req.flags & FLAG_EMPTY_STATE) != 0;
    tracing::info!(shard_id, skip_deleted, "fixed follower handshake complete");

    // Take the SPSC consumer for this shard (first follower only).
    let consumer = {
        let mut guard = crate::sync::lock(&pending[shard_id]);
        guard.1.take()
    };

    serve_shard(reader, writer, engine, shard_id, consumer, skip_deleted, stop)
}

fn serve_shard(
    mut reader: TcpStream,
    mut writer: BufWriter<TcpStream>,
    engine: ArcEngine,
    shard_id: usize,
    consumer: Option<Consumer<FixedReplicationEvent>>,
    skip_deleted: bool,
    stop: ShutdownSignal,
) -> DbResult<()> {
    // Phase 1: full scan — implemented in Task 26.
    let total = phase1_full_scan(&engine, shard_id, &mut writer, &mut reader, skip_deleted, &stop)?;
    write_frame(&mut writer, &CaughtUp {
        shard_id: shard_id as u8,
        total_scanned: total,
    }.encode()).map_err(DbError::from)?;
    tracing::info!(shard_id, total, "fixed catch-up complete");

    // Phase 2: streaming — implemented in Task 27.
    if let Some(consumer) = consumer {
        phase2_streaming(&engine, shard_id, consumer, &mut writer, &mut reader, &stop)?;
    } else {
        tracing::warn!(shard_id, "no SPSC consumer — catch-up only mode");
        idle_until_disconnect(&mut reader, &mut writer, &stop)?;
    }
    Ok(())
}

fn phase1_full_scan(
    _engine: &ArcEngine,
    _shard_id: usize,
    _writer: &mut BufWriter<TcpStream>,
    _reader: &mut TcpStream,
    _skip_deleted: bool,
    _stop: &ShutdownSignal,
) -> DbResult<u64> {
    // Task 26.
    Ok(0)
}

fn phase2_streaming(
    _engine: &ArcEngine,
    _shard_id: usize,
    _consumer: Consumer<FixedReplicationEvent>,
    _writer: &mut BufWriter<TcpStream>,
    _reader: &mut TcpStream,
    _stop: &ShutdownSignal,
) -> DbResult<()> {
    // Task 27.
    Ok(())
}

fn idle_until_disconnect(
    reader: &mut TcpStream,
    writer: &mut BufWriter<TcpStream>,
    stop: &ShutdownSignal,
) -> DbResult<()> {
    reader.set_read_timeout(Some(Duration::from_secs(1))).ok();
    loop {
        if stop.is_shutdown() { return Ok(()); }
        match read_frame(reader) {
            Ok(f) if f.msg_type == FixedMessageType::Heartbeat => {
                write_frame(writer, &encode_heartbeat()).map_err(DbError::from)?;
            }
            Ok(_) => {}
            Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut
                       || e.kind() == std::io::ErrorKind::WouldBlock => { continue; }
            Err(_) => return Ok(()),
        }
    }
}
```

- [ ] **Step 25.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 25.3: Commit**

```sh
git add armdb/src/fixed_replication/server.rs
git commit -m "feat(armdb/fixed_replication): handshake + serve_shard dispatcher"
```

---

### Task 26: Phase 1 — full slot scan

**Files:**
- Modify: `armdb/src/fixed_replication/server.rs`

Per spec §7 "Full scan detail". Uses `read_shard_chunk` in 64 KiB chunks.

- [ ] **Step 26.1: Implement `phase1_full_scan`**

Replace the stub:

```rust
const SCAN_CHUNK_BYTES: usize = 64 * 1024;

fn phase1_full_scan(
    engine: &ArcEngine,
    shard_id: usize,
    writer: &mut BufWriter<TcpStream>,
    reader: &mut TcpStream,
    skip_deleted: bool,
    stop: &ShutdownSignal,
) -> DbResult<u64> {
    let slot_size = engine.slot_size() as usize;
    let key_len = engine.key_len();
    let value_len = engine.value_len();
    let slot_count = engine.current_slot_count(shard_id);
    let slots_per_chunk = (SCAN_CHUNK_BYTES / slot_size).max(1);

    let mut total_scanned = 0u64;
    let mut batch = SlotBatchEncoder::new(shard_id as u8, key_len, value_len);

    let mut slot_id = 0u32;
    while slot_id < slot_count {
        if stop.is_shutdown() { return Ok(total_scanned); }
        let remaining = slot_count - slot_id;
        let this_chunk = remaining.min(slots_per_chunk as u32) as usize;
        let chunk = engine.read_shard_chunk(shard_id, slot_id, this_chunk)?;

        for i in 0..this_chunk {
            let off = i * slot_size;
            let slot_buf = &chunk[off..off + slot_size];
            let meta = u32::from_le_bytes(slot_buf[0..4].try_into().unwrap());
            let status = crate::fixed::slot::status_of(meta);

            if status == crate::fixed::slot::STATUS_FREE { continue; }
            if status == crate::fixed::slot::STATUS_DELETED && skip_deleted { continue; }

            let key = &slot_buf[crate::fixed::slot::SLOT_HEADER_SIZE
                                .. crate::fixed::slot::SLOT_HEADER_SIZE + key_len];
            let current_slot = slot_id + i as u32;
            if status == crate::fixed::slot::STATUS_OCCUPIED {
                let value = &slot_buf[crate::fixed::slot::SLOT_HEADER_SIZE + key_len
                                      .. crate::fixed::slot::SLOT_HEADER_SIZE + key_len + value_len];
                batch.add_occupied(current_slot, meta, key, value);
            } else {
                batch.add_deleted(current_slot, meta, key);
            }
            total_scanned += 1;

            if batch.len() as usize >= BATCH_MAX_ENTRIES
                || batch.bytes() >= BATCH_MAX_BYTES
            {
                flush_and_wait_ack(writer, reader, batch, engine, shard_id)?;
                batch = SlotBatchEncoder::new(shard_id as u8, key_len, value_len);
            }
        }

        slot_id += this_chunk as u32;
    }

    if !batch.is_empty() {
        flush_and_wait_ack(writer, reader, batch, engine, shard_id)?;
    }

    metrics::counter!("armdb.fixed.catchup_slots_scanned",
        "shard" => shard_id.to_string()).increment(total_scanned);

    Ok(total_scanned)
}

fn flush_and_wait_ack(
    writer: &mut BufWriter<TcpStream>,
    reader: &mut TcpStream,
    batch: SlotBatchEncoder,
    engine: &ArcEngine,
    shard_id: usize,
) -> DbResult<()> {
    let frame = batch.finish();
    write_frame(writer, &frame).map_err(DbError::from)?;
    writer.flush().map_err(DbError::from)?;
    // Wait for Ack — blocks follower flow control.
    let ack_frame = read_frame(reader).map_err(DbError::from)?;
    if ack_frame.msg_type == FixedMessageType::Ack {
        let ack = Ack::decode(&ack_frame.payload).map_err(DbError::from)?;
        engine.update_min_replicated_version(shard_id, ack.max_version_seen);
    }
    Ok(())
}
```

- [ ] **Step 26.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 26.3: Commit**

```sh
git add armdb/src/fixed_replication/server.rs
git commit -m "feat(armdb/fixed_replication): phase 1 full scan with FLAG_EMPTY_STATE skip"
```

---

### Task 27: Phase 2 — SPSC streaming loop

**Files:**
- Modify: `armdb/src/fixed_replication/server.rs`

- [ ] **Step 27.1: Implement**

Replace `phase2_streaming` stub:

```rust
fn phase2_streaming(
    engine: &ArcEngine,
    shard_id: usize,
    mut consumer: Consumer<FixedReplicationEvent>,
    writer: &mut BufWriter<TcpStream>,
    reader: &mut TcpStream,
    stop: &ShutdownSignal,
) -> DbResult<()> {
    let key_len = engine.key_len();
    let value_len = engine.value_len();
    let slot_size = engine.slot_size() as usize;
    let mut last_heartbeat = std::time::Instant::now();
    let hb_interval = Duration::from_secs(HEARTBEAT_INTERVAL_SECS);

    reader.set_nonblocking(true).ok();

    loop {
        if stop.is_shutdown() { return Ok(()); }

        let mut batch = SlotBatchEncoder::new(shard_id as u8, key_len, value_len);
        while (batch.len() as usize) < BATCH_MAX_ENTRIES && batch.bytes() < BATCH_MAX_BYTES {
            match consumer.pop() {
                Ok(FixedReplicationEvent::Write { slot_id, payload }) => {
                    debug_assert_eq!(payload.len(), slot_size);
                    let meta = u32::from_le_bytes(payload[0..4].try_into().unwrap());
                    let key = &payload[crate::fixed::slot::SLOT_HEADER_SIZE
                                       .. crate::fixed::slot::SLOT_HEADER_SIZE + key_len];
                    let value = &payload[crate::fixed::slot::SLOT_HEADER_SIZE + key_len
                                         .. crate::fixed::slot::SLOT_HEADER_SIZE + key_len + value_len];
                    batch.add_occupied(slot_id, meta, key, value);
                }
                Ok(FixedReplicationEvent::Delete { slot_id, meta, key }) => {
                    batch.add_deleted(slot_id, meta, &key);
                }
                Err(_) => break,
            }
        }

        if !batch.is_empty() {
            let frame_events = batch.len() as u64;
            let frame = batch.finish();
            write_frame(writer, &frame).map_err(DbError::from)?;
            writer.flush().map_err(DbError::from)?;
            metrics::counter!("armdb.fixed.streaming_events_sent",
                "shard" => shard_id.to_string()).increment(frame_events);

            // Non-blocking Ack check.
            match read_frame(reader) {
                Ok(f) if f.msg_type == FixedMessageType::Ack => {
                    if let Ok(ack) = Ack::decode(&f.payload) {
                        engine.update_min_replicated_version(shard_id, ack.max_version_seen);
                    }
                }
                Ok(_) => {}
                Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
                Err(e) => {
                    return Err(DbError::Replication(format!("ack read error: {e}")));
                }
            }
        } else {
            if last_heartbeat.elapsed() >= hb_interval {
                write_frame(writer, &encode_heartbeat()).map_err(DbError::from)?;
                writer.flush().map_err(DbError::from)?;
                last_heartbeat = std::time::Instant::now();
            }
            thread::sleep(Duration::from_millis(TAIL_POLL_MS));
        }
    }
}
```

- [ ] **Step 27.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 27.3: Commit**

```sh
git add armdb/src/fixed_replication/server.rs
git commit -m "feat(armdb/fixed_replication): phase 2 SPSC streaming with heartbeat"
```

---

## Phase 10: Client

### Task 28: `FixedReplicationClient` skeleton + per-shard reconnect

**Files:**
- Create: `armdb/src/fixed_replication/client.rs`
- Modify: `armdb/src/fixed_replication/mod.rs`

Per spec §8.

- [ ] **Step 28.1: Write skeleton**

Write to `armdb/src/fixed_replication/client.rs`:

```rust
//! Follower-side replication client for FixedStore.

use std::io::{BufReader, BufWriter, Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::Arc;
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant, SystemTime};

use crate::error::{DbError, DbResult};
use crate::shutdown::ShutdownSignal;

use super::apply::{ApplyOutcome, FixedReplicationTarget};
use super::cursor::{FixedReplicationCursor, CURSOR_SAVE_INTERVAL};
use super::protocol::*;

pub const RECONNECT_BASE_MS: u64 = 1000;
pub const RECONNECT_MAX_MS: u64 = 30_000;

pub struct FixedReplicationClient {
    stop: ShutdownSignal,
    handles: Vec<JoinHandle<()>>,
}

impl FixedReplicationClient {
    pub fn start(
        leader_addr: SocketAddr,
        target: Arc<dyn FixedReplicationTarget>,
        signal: ShutdownSignal,
    ) -> DbResult<Self> {
        let shard_count = target.shard_count();
        let mut handles = Vec::with_capacity(shard_count);
        for shard_id in 0..shard_count {
            let target = target.clone();
            let stop = signal.clone();
            let h = thread::spawn(move || {
                run_shard_client(leader_addr, shard_id as u8, target, stop);
            });
            handles.push(h);
        }
        Ok(Self { stop: signal, handles })
    }

    pub fn stop(&self) {
        self.stop.shutdown();
    }
}

impl Drop for FixedReplicationClient {
    fn drop(&mut self) {
        self.stop.shutdown();
        for h in self.handles.drain(..) {
            let _ = h.join();
        }
    }
}

fn run_shard_client(
    leader_addr: SocketAddr,
    shard_id: u8,
    target: Arc<dyn FixedReplicationTarget>,
    stop: ShutdownSignal,
) {
    let mut backoff_ms = RECONNECT_BASE_MS;
    loop {
        if stop.is_shutdown() { return; }
        match connect_and_sync(leader_addr, shard_id, &target, &stop) {
            Ok(()) => {
                tracing::info!(shard_id, "fixed replication stream ended cleanly");
                return;
            }
            Err(e) => {
                tracing::warn!(shard_id, error = %e, backoff_ms,
                    "fixed replication error, reconnecting");
                metrics::counter!("armdb.fixed.reconnect_count",
                    "shard" => shard_id.to_string()).increment(1);
                if stop.wait_timeout(Duration::from_millis(backoff_ms)) { return; }
                backoff_ms = (backoff_ms * 2).min(RECONNECT_MAX_MS);
            }
        }
    }
}

fn connect_and_sync(
    _leader_addr: SocketAddr,
    _shard_id: u8,
    _target: &Arc<dyn FixedReplicationTarget>,
    _stop: &ShutdownSignal,
) -> DbResult<()> {
    // Implemented in Tasks 29–30.
    Ok(())
}
```

Update `mod.rs`:
```rust
pub mod client;
pub use client::FixedReplicationClient;
```

- [ ] **Step 28.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 28.3: Commit**

```sh
git add armdb/src/fixed_replication/client.rs armdb/src/fixed_replication/mod.rs
git commit -m "feat(armdb/fixed_replication): client skeleton with reconnect backoff"
```

---

### Task 29: Client — handshake with `FLAG_EMPTY_STATE` and topology validation

**Files:**
- Modify: `armdb/src/fixed_replication/client.rs`

- [ ] **Step 29.1: Fill handshake**

Replace `connect_and_sync` with:

```rust
fn connect_and_sync(
    leader_addr: SocketAddr,
    shard_id: u8,
    target: &Arc<dyn FixedReplicationTarget>,
    stop: &ShutdownSignal,
) -> DbResult<()> {
    let stream = TcpStream::connect_timeout(&leader_addr, Duration::from_secs(5))
        .map_err(DbError::from)?;
    stream.set_nodelay(true).ok();
    let mut writer = stream.try_clone().map_err(DbError::from)?;
    let mut reader = BufReader::new(stream);

    // Decide flags.
    let mut flags = 0u8;
    if target.shard_occupied_count(shard_id) == 0 {
        flags |= FLAG_EMPTY_STATE;
    }

    write_frame(&mut writer, &SyncRequest {
        shard_id,
        protocol_version: PROTOCOL_VERSION,
        flags,
    }.encode()).map_err(DbError::from)?;

    // Receive ShardInfo.
    let info_frame = read_frame(&mut reader).map_err(DbError::from)?;
    if info_frame.msg_type != FixedMessageType::ShardInfo {
        return Err(DbError::Replication(format!(
            "expected ShardInfo, got {:?}", info_frame.msg_type
        )));
    }
    let info = ShardInfo::decode(&info_frame.payload).map_err(DbError::from)?;

    // Validate topology.
    if info.shard_count as usize != target.shard_count() {
        return Err(DbError::ShardCountMismatch {
            leader: info.shard_count as usize,
            follower: target.shard_count(),
        });
    }
    if info.key_len as usize != target.key_len()
        || info.value_len as usize != target.value_len()
    {
        return Err(DbError::Replication(format!(
            "K/V size mismatch: leader ({}, {}), follower ({}, {})",
            info.key_len, info.value_len, target.key_len(), target.value_len()
        )));
    }

    target.grow_shard_to(shard_id, info.current_slot_count)?;

    tracing::info!(shard_id, flags, current_slot_count = info.current_slot_count,
                   "fixed follower connected to leader");

    apply_loop(reader, writer, shard_id, info, target, stop)
}

fn apply_loop(
    _reader: BufReader<TcpStream>,
    _writer: TcpStream,
    _shard_id: u8,
    _info: ShardInfo,
    _target: &Arc<dyn FixedReplicationTarget>,
    _stop: &ShutdownSignal,
) -> DbResult<()> {
    // Task 30.
    Ok(())
}
```

- [ ] **Step 29.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 29.3: Commit**

```sh
git add armdb/src/fixed_replication/client.rs
git commit -m "feat(armdb/fixed_replication): client handshake with FLAG_EMPTY_STATE"
```

---

### Task 30: Client apply loop — decode batches, dispatch, periodic Ack + cursor save

**Files:**
- Modify: `armdb/src/fixed_replication/client.rs`

- [ ] **Step 30.1: Implement `apply_loop`**

Replace the stub:

```rust
fn apply_loop(
    mut reader: BufReader<TcpStream>,
    mut writer: TcpStream,
    shard_id: u8,
    info: ShardInfo,
    target: &Arc<dyn FixedReplicationTarget>,
    stop: &ShutdownSignal,
) -> DbResult<()> {
    let shard_dir = target.shard_dir(shard_id);
    let mut cursor = FixedReplicationCursor::load(shard_id, &shard_dir)
        .unwrap_or_else(|| FixedReplicationCursor::new(shard_id));

    let mut events_since_ack: u64 = 0;
    let mut events_since_save: u64 = 0;
    let mut applied_total = cursor.applied_total;
    let mut max_version_seen = cursor.max_version_seen;
    let mut last_heartbeat_sent = Instant::now();
    let hb = Duration::from_secs(HEARTBEAT_INTERVAL_SECS);

    // Read-side timeout so we can exit on stop signal.
    reader.get_mut().set_read_timeout(Some(Duration::from_secs(1))).ok();

    loop {
        if stop.is_shutdown() {
            if events_since_save > 0 {
                cursor.applied_total = applied_total;
                cursor.max_version_seen = max_version_seen;
                cursor.last_ack_at = SystemTime::now();
                let _ = cursor.save(&shard_dir);
            }
            return Ok(());
        }

        let frame = match read_frame(&mut reader) {
            Ok(f) => f,
            Err(ref e) if e.kind() == std::io::ErrorKind::TimedOut
                       || e.kind() == std::io::ErrorKind::WouldBlock => { continue; }
            Err(e) => return Err(DbError::from(e)),
        };

        match frame.msg_type {
            FixedMessageType::SlotBatch => {
                let decoded = SlotBatchDecoder::new(&frame.payload,
                    info.key_len as usize,
                    info.value_len as usize,
                ).map_err(DbError::from)?;

                for event in decoded.iter() {
                    let event = event.map_err(DbError::from)?;
                    let outcome = if crate::fixed::slot::status_of(event.meta)
                        == crate::fixed::slot::STATUS_OCCUPIED
                    {
                        target.apply_occupied(shard_id, event.slot_id, event.meta,
                            event.key, event.value)?
                    } else {
                        target.apply_deleted(shard_id, event.slot_id, event.meta, event.key)?
                    };
                    if matches!(outcome, ApplyOutcome::Applied) {
                        applied_total += 1;
                        let v = crate::fixed::slot::version_of(event.meta);
                        if v > max_version_seen { max_version_seen = v; }
                    }
                    events_since_ack += 1;
                    events_since_save += 1;
                }

                if events_since_ack >= ACK_INTERVAL as u64 {
                    write_frame(&mut writer, &Ack {
                        shard_id,
                        applied_count: applied_total,
                        max_version_seen,
                    }.encode()).map_err(DbError::from)?;
                    events_since_ack = 0;
                }

                if events_since_save >= CURSOR_SAVE_INTERVAL as u64 {
                    cursor.applied_total = applied_total;
                    cursor.max_version_seen = max_version_seen;
                    cursor.last_ack_at = SystemTime::now();
                    cursor.save(&shard_dir)?;
                    metrics::gauge!("armdb.fixed.max_version_seen",
                        "shard" => shard_id.to_string()).set(max_version_seen as f64);
                    events_since_save = 0;
                }
            }
            FixedMessageType::CaughtUp => {
                let cu = CaughtUp::decode(&frame.payload).map_err(DbError::from)?;
                tracing::info!(shard_id, total_scanned = cu.total_scanned,
                    "fixed catch-up complete, entering streaming");
            }
            FixedMessageType::Heartbeat => {
                write_frame(&mut writer, &encode_heartbeat()).map_err(DbError::from)?;
            }
            FixedMessageType::Error => {
                let msg = decode_error(&frame.payload);
                return Err(DbError::Replication(msg));
            }
            other => {
                return Err(DbError::Replication(format!("unexpected message: {other:?}")));
            }
        }

        if last_heartbeat_sent.elapsed() >= hb {
            write_frame(&mut writer, &encode_heartbeat()).map_err(DbError::from)?;
            last_heartbeat_sent = Instant::now();
        }
    }
}
```

- [ ] **Step 30.2: Compile check**

```sh
cargo check -p armdb --features replication
```

- [ ] **Step 30.3: Commit**

```sh
git add armdb/src/fixed_replication/client.rs
git commit -m "feat(armdb/fixed_replication): client apply loop with Ack + cursor save"
```

---

## Phase 11: End-to-end integration tests

Integration tests live in `armdb/tests/fixed_replication.rs`. Each test sets up a real `FixedEngine` for leader and follower (two temp dirs), runs server + client on a localhost port, drives writes on leader, asserts convergence on follower.

### Task 31: Test harness helpers

**Files:**
- Create: `armdb/tests/fixed_replication.rs`

- [ ] **Step 31.1: Write helpers and one smoke test**

```rust
//! End-to-end integration tests for Fixed replication.
//!
//! Each test spins up a leader Db + server and a follower Db + client,
//! drives traffic through the public API, then waits for convergence.

use std::net::{SocketAddr, TcpListener};
use std::sync::Arc;
use std::time::{Duration, Instant};

use armdb::Db;
use armdb::fixed::FixedConfig;
use armdb::fixed_replication::{FixedReplicationClient, FixedReplicationServer, FixedReplicationTarget};
use armdb::shutdown::ShutdownSignal;
use tempfile::TempDir;

fn pick_port() -> u16 {
    // Bind to 0 on loopback and return the assigned port.
    let l = TcpListener::bind("127.0.0.1:0").unwrap();
    l.local_addr().unwrap().port()
}

struct Pair {
    _leader_dir: TempDir,
    _follower_dir: TempDir,
    pub leader_db: Db,
    pub follower_db: Db,
    pub server: FixedReplicationServer,
    pub client: FixedReplicationClient,
    pub stop_leader: ShutdownSignal,
    pub stop_follower: ShutdownSignal,
}

fn small_config() -> FixedConfig {
    FixedConfig {
        shard_count: 2,
        grow_step: 64,
        ..FixedConfig::test()
    }
}

/// Poll `cond` until true or timeout.
fn wait_for(timeout: Duration, mut cond: impl FnMut() -> bool) -> bool {
    let start = Instant::now();
    while start.elapsed() < timeout {
        if cond() { return true; }
        std::thread::sleep(Duration::from_millis(20));
    }
    false
}

#[test]
fn test_smoke_harness_compiles() {
    // Just verify imports and helpers compile; real tests follow in subsequent tasks.
    let _ = pick_port();
    let _ = small_config();
}
```

Depending on the concrete `Db` API for Fixed collections (`open_zero_tree_fixed<K, V>` per CLAUDE.md), the harness will need to instantiate a tree. Refer to existing integration tests in `armdb/tests/` for how Db is constructed for Fixed.

- [ ] **Step 31.2: Run**

```sh
cargo nextest run -p armdb -E 'package(armdb) and binary(fixed_replication)' --no-fail-fast
```

- [ ] **Step 31.3: Commit**

```sh
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): harness helpers + smoke test"
```

---

### Task 32: Integration — `test_initial_sync_empty_follower`

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

Per spec §14. Fresh follower with `FLAG_EMPTY_STATE` → leader skips DELETED events; follower converges.

- [ ] **Step 32.1: Add test**

```rust
#[test]
fn test_initial_sync_empty_follower() {
    // Leader: open Fixed tree, write 100 keys, delete 10.
    let leader_dir = TempDir::new().unwrap();
    let leader_db = Db::open_fixed(leader_dir.path(), small_config(), /*K*/8, /*V*/16).unwrap();
    let tree = leader_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    for i in 0..100u64 {
        let key = armdb::Fuid::from(i);
        let val = [i as u8; 16];
        tree.put(&key, &val).unwrap();
    }
    for i in 0..10u64 {
        tree.delete(&armdb::Fuid::from(i)).unwrap();
    }
    leader_db.flush().unwrap();

    // Start server.
    let port = pick_port();
    let stop_leader = ShutdownSignal::new();
    let server = FixedReplicationServer::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        leader_db.fixed_engine_access(),  // needs helper; see below
        stop_leader.clone(),
    ).unwrap();

    // Fresh follower.
    let follower_dir = TempDir::new().unwrap();
    let follower_db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
    let follower_tree = follower_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    let stop_follower = ShutdownSignal::new();
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(follower_tree.clone());
    let client = FixedReplicationClient::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        target,
        stop_follower.clone(),
    ).unwrap();

    // Wait for convergence: all 90 remaining keys visible on follower.
    assert!(wait_for(Duration::from_secs(10), || {
        (10..100u64).all(|i| follower_tree.get(&armdb::Fuid::from(i)).is_some())
    }), "follower did not converge");

    // Keys 0..10 should be absent.
    for i in 0..10u64 {
        assert!(follower_tree.get(&armdb::Fuid::from(i)).is_none());
    }

    stop_follower.shutdown();
    drop(client);
    stop_leader.shutdown();
    drop(server);
}
```

**Dependencies this test surfaces:**
- `Db::open_fixed` — existing or needs a shim.
- `Db::fixed_engine_access` — helper returning `Arc<dyn FixedEngineAccess>`. Add to `Db` if missing:
  ```rust
  #[cfg(feature = "replication")]
  pub fn fixed_engine_access(&self) -> Arc<dyn armdb::fixed_replication::FixedEngineAccess> {
      // wrap inner FixedEngine in Arc
  }
  ```

Adapt variable names to existing `Db` API — inspect `armdb/src/lib.rs` for the real constructors.

- [ ] **Step 32.2: Run**

```sh
cargo nextest run -p armdb -E 'test(test_initial_sync_empty_follower)'
```

- [ ] **Step 32.3: Commit**

```sh
git add armdb/tests/fixed_replication.rs armdb/src/
git commit -m "test(armdb/fixed_replication): initial sync on empty follower"
```

---

### Task 33: Integration — `test_streaming_live_writes`

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

- [ ] **Step 33.1: Add test**

```rust
#[test]
fn test_streaming_live_writes() {
    // Connect first, write after — exercise the streaming path.
    let leader_dir = TempDir::new().unwrap();
    let leader_db = Db::open_fixed(leader_dir.path(), small_config(), 8, 16).unwrap();
    let tree = leader_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");

    let port = pick_port();
    let stop_leader = ShutdownSignal::new();
    let _server = FixedReplicationServer::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        leader_db.fixed_engine_access(),
        stop_leader.clone(),
    ).unwrap();

    let follower_dir = TempDir::new().unwrap();
    let follower_db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
    let follower_tree = follower_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    let stop_follower = ShutdownSignal::new();
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(follower_tree.clone());
    let _client = FixedReplicationClient::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        target,
        stop_follower.clone(),
    ).unwrap();

    // Give handshake time to complete.
    std::thread::sleep(Duration::from_millis(200));

    // Now write on leader — all events should stream.
    for i in 0..500u64 {
        let key = armdb::Fuid::from(i);
        let val = [(i * 2) as u8; 16];
        tree.put(&key, &val).unwrap();
    }
    leader_db.flush().unwrap();

    assert!(wait_for(Duration::from_secs(10), || {
        (0..500u64).all(|i| follower_tree.get(&armdb::Fuid::from(i)) == Some([(i * 2) as u8; 16]))
    }), "streaming did not converge");

    stop_follower.shutdown();
    stop_leader.shutdown();
}
```

- [ ] **Step 33.2: Run**

```sh
cargo nextest run -p armdb -E 'test(test_streaming_live_writes)'
```

- [ ] **Step 33.3: Commit**

```sh
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): streaming live writes converge"
```

---

### Task 34: Integration — Bug 1 / Bug 2 regression tests

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

Per spec §8 critical bug fixes.

- [ ] **Step 34.1: Bug 1 test — `test_delete_does_not_remove_moved_key`**

```rust
#[test]
fn test_delete_does_not_remove_moved_key() {
    // Scenario: follower has key_X → slot_2 from earlier. A catch-up DELETE
    // for slot_1 with key_X must NOT remove the valid slot_2 mapping.

    // We construct this by directly manipulating follower state via
    // the FixedReplicationTarget trait to avoid orchestrating a complex
    // live leader.

    let dir = TempDir::new().unwrap();
    let db = Db::open_fixed(dir.path(), small_config(), 8, 16).unwrap();
    let tree = db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(tree.clone());

    let key_x = armdb::Fuid::from(0x0101_0101_0101_0101u64);
    let meta_v10 = armdb::fixed::slot::pack_meta(
        armdb::fixed::slot::STATUS_OCCUPIED, 10);

    // Apply OCCUPIED at slot 2.
    target.apply_occupied(0, 2, meta_v10, key_x.as_bytes(), &[0x55u8; 16]).unwrap();
    assert!(tree.get(&key_x).is_some());

    // Now apply DELETED for slot 1 with key_x — must be ignored (wrong slot).
    let del_meta = armdb::fixed::slot::pack_meta(
        armdb::fixed::slot::STATUS_DELETED, 11);
    target.apply_deleted(0, 1, del_meta, key_x.as_bytes()).unwrap();

    // key_x still present.
    assert!(tree.get(&key_x).is_some(), "Bug 1 regression: key_x was wrongly removed");
}
```

- [ ] **Step 34.2: Bug 2 test — `test_occupied_cleans_stale_when_new_key_maps_elsewhere`**

```rust
#[test]
fn test_occupied_cleans_stale_when_new_key_maps_elsewhere() {
    // Scenario: follower index has
    //   key_X → slot_A (from earlier), versions[slot_A].status = OCCUPIED
    //   key_Y → slot_C
    // Now leader sends OCCUPIED(slot_A, key_Y) — slot_A was reused.
    // Expected: key_X removed from index, key_Y points to slot_A.

    let dir = TempDir::new().unwrap();
    let db = Db::open_fixed(dir.path(), small_config(), 8, 16).unwrap();
    let tree = db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(tree.clone());

    let key_x = armdb::Fuid::from(0xAAu64);
    let key_y = armdb::Fuid::from(0xBBu64);

    // Setup: key_X → slot 1, key_Y → slot 3.
    target.apply_occupied(0, 1,
        armdb::fixed::slot::pack_meta(armdb::fixed::slot::STATUS_OCCUPIED, 1),
        key_x.as_bytes(), &[0x11u8; 16]).unwrap();
    target.apply_occupied(0, 3,
        armdb::fixed::slot::pack_meta(armdb::fixed::slot::STATUS_OCCUPIED, 1),
        key_y.as_bytes(), &[0x33u8; 16]).unwrap();

    // Leader reuses slot 1 for key_Y (stale OCCUPIED).
    target.apply_occupied(0, 1,
        armdb::fixed::slot::pack_meta(armdb::fixed::slot::STATUS_OCCUPIED, 2),
        key_y.as_bytes(), &[0x77u8; 16]).unwrap();

    // key_X must be gone; key_Y points to slot 1 with value [0x77;16].
    assert!(tree.get(&key_x).is_none(), "Bug 2 regression: stale key_X not cleaned up");
    assert_eq!(tree.get(&key_y), Some([0x77u8; 16]));
}
```

- [ ] **Step 34.3: Run**

```sh
cargo nextest run -p armdb -E 'test(test_delete_does_not_remove_moved_key) + test(test_occupied_cleans_stale)'
```

- [ ] **Step 34.4: Commit**

```sh
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): Bug 1 and Bug 2 regression tests"
```

---

### Task 35: Integration — `test_reconnect_after_disconnect`

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

- [ ] **Step 35.1: Add test**

```rust
#[test]
fn test_reconnect_after_disconnect() {
    let leader_dir = TempDir::new().unwrap();
    let leader_db = Db::open_fixed(leader_dir.path(), small_config(), 8, 16).unwrap();
    let tree = leader_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");

    let port = pick_port();
    let stop_leader = ShutdownSignal::new();
    let _server = FixedReplicationServer::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        leader_db.fixed_engine_access(),
        stop_leader.clone(),
    ).unwrap();

    let follower_dir = TempDir::new().unwrap();
    let follower_db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
    let follower_tree = follower_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");

    // Phase 1: connect, write a few, disconnect.
    let stop_f1 = ShutdownSignal::new();
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(follower_tree.clone());
    let client1 = FixedReplicationClient::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        target.clone(),
        stop_f1.clone(),
    ).unwrap();
    for i in 0..50u64 {
        tree.put(&armdb::Fuid::from(i), &[1u8; 16]).unwrap();
    }
    assert!(wait_for(Duration::from_secs(5), || {
        follower_tree.get(&armdb::Fuid::from(49)).is_some()
    }));
    stop_f1.shutdown();
    drop(client1);

    // Phase 2: leader writes 50 more while follower is disconnected.
    for i in 50..100u64 {
        tree.put(&armdb::Fuid::from(i), &[2u8; 16]).unwrap();
    }
    leader_db.flush().unwrap();

    // Phase 3: reconnect — follower catches up.
    let stop_f2 = ShutdownSignal::new();
    let _client2 = FixedReplicationClient::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        target,
        stop_f2.clone(),
    ).unwrap();
    assert!(wait_for(Duration::from_secs(10), || {
        follower_tree.get(&armdb::Fuid::from(99)).is_some()
    }), "follower did not reconverge after reconnect");

    stop_f2.shutdown();
    stop_leader.shutdown();
}
```

- [ ] **Step 35.2: Run + commit**

```sh
cargo nextest run -p armdb -E 'test(test_reconnect_after_disconnect)'
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): reconnect after disconnect catches up"
```

---

### Task 36: Integration — `test_modular_compare_at_boundary`

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

- [ ] **Step 36.1: Add test**

```rust
#[test]
fn test_modular_compare_at_boundary() {
    // Force a slot's version near 2^30, then apply event with version that
    // wraps; follower must accept it via is_newer modular compare.

    let dir = TempDir::new().unwrap();
    let db = Db::open_fixed(dir.path(), small_config(), 8, 16).unwrap();
    let tree = db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(tree.clone());

    let key = armdb::Fuid::from(0xDEADBEEFu64);
    let high = (1u32 << 30) - 5;  // near max
    target.apply_occupied(0, 0,
        armdb::fixed::slot::pack_meta(armdb::fixed::slot::STATUS_OCCUPIED, high),
        key.as_bytes(), &[0x11u8; 16]).unwrap();

    // Event with version wrapped past 2^30 — version 3 (diff = 8, newer).
    target.apply_occupied(0, 0,
        armdb::fixed::slot::pack_meta(armdb::fixed::slot::STATUS_OCCUPIED, 3),
        key.as_bytes(), &[0x22u8; 16]).unwrap();

    assert_eq!(tree.get(&key), Some([0x22u8; 16]));

    // Stale event (version = high - 2) must be skipped.
    target.apply_occupied(0, 0,
        armdb::fixed::slot::pack_meta(armdb::fixed::slot::STATUS_OCCUPIED, high - 2),
        key.as_bytes(), &[0x99u8; 16]).unwrap();
    assert_eq!(tree.get(&key), Some([0x22u8; 16]), "stale event must be skipped");
}
```

- [ ] **Step 36.2: Run + commit**

```sh
cargo nextest run -p armdb -E 'test(test_modular_compare_at_boundary)'
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): modular compare at 2^30 boundary"
```

---

### Task 37: Integration — `test_topology_mismatch_errors`

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

- [ ] **Step 37.1: Add test**

```rust
#[test]
fn test_topology_mismatch_errors() {
    // Leader: 2 shards, K=8, V=16.
    let leader_dir = TempDir::new().unwrap();
    let leader_db = Db::open_fixed(leader_dir.path(), small_config(), 8, 16).unwrap();
    let _leader_tree = leader_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");

    let port = pick_port();
    let stop_leader = ShutdownSignal::new();
    let _server = FixedReplicationServer::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        leader_db.fixed_engine_access(),
        stop_leader.clone(),
    ).unwrap();

    // Follower with different shard count.
    let follower_dir = TempDir::new().unwrap();
    let mut bad_cfg = small_config();
    bad_cfg.shard_count = 4;
    let follower_db = Db::open_fixed(follower_dir.path(), bad_cfg, 8, 16).unwrap();
    let follower_tree = follower_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");

    let stop_f = ShutdownSignal::new();
    let target: Arc<dyn FixedReplicationTarget> = Arc::new(follower_tree.clone());
    let _client = FixedReplicationClient::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        target,
        stop_f.clone(),
    ).unwrap();

    // Give it time — the client will error and backoff-retry.
    std::thread::sleep(Duration::from_millis(1500));
    // There's no direct handle to read the error; instead assert the
    // reconnect metric incremented.
    // NOTE: replace with a metric scrape helper if available in your test infra.
    // For v1 we just sleep and check no crash + follower remains empty.
    assert_eq!(follower_tree.get(&armdb::Fuid::from(0)), None);

    stop_f.shutdown();
    stop_leader.shutdown();
}
```

- [ ] **Step 37.2: Run + commit**

```sh
cargo nextest run -p armdb -E 'test(test_topology_mismatch_errors)'
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): topology mismatch triggers backoff, no crash"
```

---

### Task 38: Integration — `test_follower_restart_dirty_and_clean`

**Files:**
- Modify: `armdb/tests/fixed_replication.rs`

Per spec §9.

- [ ] **Step 38.1: Add test**

```rust
#[test]
fn test_follower_restart_clean_uses_sidecar() {
    let leader_dir = TempDir::new().unwrap();
    let leader_db = Db::open_fixed(leader_dir.path(), small_config(), 8, 16).unwrap();
    let tree = leader_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");

    let port = pick_port();
    let stop_leader = ShutdownSignal::new();
    let _server = FixedReplicationServer::start(
        format!("127.0.0.1:{port}").parse().unwrap(),
        leader_db.fixed_engine_access(),
        stop_leader.clone(),
    ).unwrap();

    let follower_dir = TempDir::new().unwrap();

    // Phase 1: follower syncs, then cleanly shuts down.
    {
        let follower_db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
        let follower_tree = follower_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
        let stop_f = ShutdownSignal::new();
        let target: Arc<dyn FixedReplicationTarget> = Arc::new(follower_tree.clone());
        let _client = FixedReplicationClient::start(
            format!("127.0.0.1:{port}").parse().unwrap(),
            target,
            stop_f.clone(),
        ).unwrap();

        for i in 0..30u64 {
            tree.put(&armdb::Fuid::from(i), &[7u8; 16]).unwrap();
        }
        assert!(wait_for(Duration::from_secs(5), || {
            follower_tree.get(&armdb::Fuid::from(29)).is_some()
        }));

        stop_f.shutdown();
        // Drop client, then drop follower_db — drop order triggers clean shutdown.
        drop(follower_db);
    }

    // Phase 2: reopen follower — must load sidecar.
    let follower_db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
    let follower_tree = follower_db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    // State restored from sidecar + recovery.
    for i in 0..30u64 {
        assert_eq!(follower_tree.get(&armdb::Fuid::from(i)), Some([7u8; 16]));
    }

    // Verify sidecar file exists.
    let sidecar = follower_dir.path().join("shard_000").join("fixed.versions");
    assert!(sidecar.exists(), "fixed.versions sidecar should exist after clean shutdown");

    stop_leader.shutdown();
}

#[test]
fn test_follower_restart_corrupted_sidecar_falls_back() {
    let follower_dir = TempDir::new().unwrap();
    {
        let db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
        let tree = db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
        tree.put(&armdb::Fuid::from(1), &[0x42u8; 16]).unwrap();
        // clean shutdown via drop
    }

    // Corrupt the sidecar trailer.
    let sidecar = follower_dir.path().join("shard_000").join("fixed.versions");
    let data = std::fs::read(&sidecar).unwrap();
    // Truncate 4 bytes (the CRC trailer).
    std::fs::write(&sidecar, &data[..data.len() - 4]).unwrap();

    // Reopen should fall back to full scan, NOT error.
    let db = Db::open_fixed(follower_dir.path(), small_config(), 8, 16).unwrap();
    let tree = db.open_zero_tree_fixed::<armdb::Fuid, 16>("data");
    assert_eq!(tree.get(&armdb::Fuid::from(1)), Some([0x42u8; 16]));
    // Sidecar should be removed after fallback.
    assert!(!sidecar.exists(), "corrupted sidecar should be removed");
}
```

- [ ] **Step 38.2: Run + commit**

```sh
cargo nextest run -p armdb -E 'test(test_follower_restart)'
git add armdb/tests/fixed_replication.rs
git commit -m "test(armdb/fixed_replication): follower restart (clean sidecar + corrupted fallback)"
```

---

### Task 39: Integration — `test_torn_slot_preserves_version`

Already added as unit test in Task 8 (via `durability.rs` tests). Nothing to add here unless we want an end-to-end version that also asserts replication convergence after recovery. Low priority for v1 — skip to self-review.

---

## Final

### Task 40: Self-review checklist

- [ ] **Step 40.1: Placeholder scan**

Run in the repo root:
```sh
rg 'TBD|TODO|FIXME|XXX|todo!\(\)|unimplemented!\(\)' armdb/src/fixed armdb/src/fixed_replication armdb/tests/fixed_replication.rs
```

Expected: no matches.

- [ ] **Step 40.2: Full build + test**

```sh
cargo check -p armdb
cargo check -p armdb --no-default-features --features replication
cargo clippy -p armdb --all-targets -- -D warnings
cargo nextest run -p armdb
```

All green.

- [ ] **Step 40.3: Metric enumeration**

Run:
```sh
rg 'metrics::counter!\("armdb\.fixed' armdb/src/fixed_replication armdb/src/fixed | sort -u
```

Verify all spec §11 metrics are emitted:
- `armdb.fixed.spsc_overflow_count` — note: currently we drop silently via `let _ = tx.push(...)`; add explicit `.inspect_err(|_| metrics::counter!("armdb.fixed.spsc_overflow_count", ...).increment(1))` in Task 12 hook if missing.
- `armdb.fixed.version_near_wrap`
- `armdb.fixed.catchup_slots_scanned`
- `armdb.fixed.streaming_events_sent`
- `armdb.fixed.applied_events` (labels `kind=occupied|deleted`)
- `armdb.fixed.skipped_events`
- `armdb.fixed.stale_cleanup_count`
- `armdb.fixed.reconnect_count`
- `armdb.fixed.max_version_seen`

If any are missing, add them before completing.

- [ ] **Step 40.4: Doc update (optional)**

Update `armdb/docs/fixed_store_design.md` Ограничения section to mark replication as now supported. Short sentence; not strictly required to ship.

- [ ] **Step 40.5: Final commit**

```sh
git add -A
git commit -m "chore(armdb/fixed_replication): self-review pass"
```

---

## Appendix: Key references

- **Design spec:** `armdb/docs/superpowers/specs/2026-04-17-fixed-replication-design.md` (commit `b715160`).
- **Review notes:** `armdb/docs/reviews/2026-04-17-fixed-replication-design.md` — both critical bugs are addressed in Tasks 22, 23, and 34.
- **Existing Bitcask replication** (reference model): `armdb/src/replication/*.rs`.
- **Test runner note:** Use `cargo nextest run -p armdb` per repo `CLAUDE.md` — `cargo test` hits fd limits with many shards.




