# VarTree / VarMap migrate hook semantics — 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:** Привести `VarTree::migrate` / `VarMap::migrate` к контракту хуков из `armdb/docs/hooks.md` (на миграции — `on_init` под `H::NEEDS_INIT`, `on_write` не фитится), добавить `replay_init()` для пути «открытие без миграции», обновить `Db::open_var_typed_*` и доки.

**Architecture:** Hook-free `_locked` методы уже существуют — `migrate` перестаёт звать публичные `put`/`delete` и зовёт `put_locked`/`delete_locked` напрямую. `replay_init()` — pub(crate), последовательно обходит индекс и зовёт `on_init`; для `VarMap` — через snapshot ключей по шардам + `self.get(key)`, чтобы не нарушить порядок lock'ов (shard → index, как у public writes).

**Tech Stack:** Rust, armdb crate (feature `var-collections`), `seize::LocalGuard`, `parking_lot::Mutex`, `tempfile`, `cargo nextest`. Все тесты под `feature = "var-collections"` (+`typed-tree` для Db-orchestration).

---

## File Structure

| Файл | Что меняется |
|---|---|
| `armdb/src/var_tree.rs` | `migrate` переписан; добавлен `pub(crate) fn replay_init`; в `mod tests` — тестовый `CountingHook` + 6 новых тестов |
| `armdb/src/var_map.rs` | `migrate` переписан; добавлен `pub(crate) fn replay_init`; в `mod tests` — `CountingHook` + 6 новых тестов |
| `armdb/src/var_typed_tree.rs` | добавлен `pub(crate) fn replay_init`, делегирующий на `self.inner.replay_init()` |
| `armdb/src/var_typed_map.rs` | то же |
| `armdb/src/armour/db.rs` | `open_var_typed_tree` / `open_var_typed_map`: `let _ = run_migration` → `let migrated = …` + `if !migrated { … .replay_init() }`; добавлены Db-orchestration тесты в `mod tests` (если есть) или в `armdb/tests/var_typed_replay_init.rs` |
| `armdb/docs/hooks.md` | таблица «Какие коллекции поддерживают хуки» (line 221-234) — VarTree/VarMap on_init: `да`; удалена сноска; уточнение в lifecycle (line 93-125) |

Все исходники Var* уже под `#[cfg(feature = "var-collections")]` (lib.rs:306-321), поэтому новые тесты тоже автоматически попадают под этот gate.

---

## Task 1: VarTree::replay_init() (raw + tests)

**Files:**
- Modify: `armdb/src/var_tree.rs` — добавить `pub(crate) fn replay_init` после `migrate` (~ var_tree.rs:736), добавить тесты в `mod tests` (var_tree.rs:1368+).

### Step 1.1: Добавить `CountingHook` в `mod tests` в `var_tree.rs`

- [ ] Открыть `armdb/src/var_tree.rs`. Найти `mod tests {` (var_tree.rs:1368). Сразу после `use tempfile::tempdir;` (var_tree.rs:1372) добавить:

```rust
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    /// Тестовый хук со счётчиками вызовов `on_write` / `on_init` и
    /// последнее наблюдённое `new`-значение для `on_write`.
    /// `NEEDS_INIT` и `NEEDS_OLD_VALUE` параметризуются константами.
    #[derive(Default)]
    struct CountingHook<const NEEDS_INIT: bool, const NEEDS_OLD: bool> {
        writes: AtomicUsize,
        writes_with_old: AtomicUsize,
        inits: AtomicUsize,
        last_write_new: parking_lot::Mutex<Option<Vec<u8>>>,
        last_init_value: parking_lot::Mutex<Option<Vec<u8>>>,
    }

    impl<const NEEDS_INIT: bool, const NEEDS_OLD: bool> WriteHook<[u8; 8]>
        for CountingHook<NEEDS_INIT, NEEDS_OLD>
    {
        const NEEDS_OLD_VALUE: bool = NEEDS_OLD;
        const NEEDS_INIT: bool = NEEDS_INIT;

        fn on_write(&self, _key: &[u8; 8], old: Option<&[u8]>, new: Option<&[u8]>) {
            self.writes.fetch_add(1, AtomicOrdering::Relaxed);
            if old.is_some() {
                self.writes_with_old.fetch_add(1, AtomicOrdering::Relaxed);
            }
            *self.last_write_new.lock() = new.map(<[u8]>::to_vec);
        }

        fn on_init(&self, _key: &[u8; 8], value: &[u8]) {
            self.inits.fetch_add(1, AtomicOrdering::Relaxed);
            *self.last_init_value.lock() = Some(value.to_vec());
        }
    }
```

- [ ] **Step 1.2: Добавить helper `open_test_tree_hooked`** сразу после `open_test_tree` (var_tree.rs:1374-1381):

```rust
    fn open_test_tree_hooked<const NEEDS_INIT: bool, const NEEDS_OLD: bool>(
        dir: &std::path::Path,
        hook: CountingHook<NEEDS_INIT, NEEDS_OLD>,
    ) -> VarTree<[u8; 8], CountingHook<NEEDS_INIT, NEEDS_OLD>> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 8192;
        cfg.compaction_threshold = 0.0;
        VarTree::open_hooked(dir, cfg, hook).expect("open hooked test tree")
    }
```

- [ ] **Step 1.3: Написать failing-тест на `replay_init` (NEEDS_INIT=true)**. В конце `mod tests` добавить:

```rust
    #[test]
    fn var_tree_replay_init_fires_on_init_per_live_key_raw() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        for i in 0u64..5 {
            tree.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        // Удалим один — replay_init не должен его видеть.
        tree.delete(&3u64.to_be_bytes()).expect("delete");

        // Сбросим счётчики, чтобы измерить только эффект replay_init.
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        tree.replay_init();

        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 4, "4 live keys");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 0, "no on_write");
    }
```

- [ ] **Step 1.4: Прогнать тест и подтвердить FAIL** (метод `replay_init` ещё не существует).

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree_replay_init_fires_on_init_per_live_key_raw)'`
Expected: компиляция падает с `no method named 'replay_init'`.

- [ ] **Step 1.5: Реализовать `replay_init`**. В `impl<K: Key, H: WriteHook<K>> VarTree<K, H>` (находится в var_tree.rs:66) сразу после метода `migrate` (заканчивается ~ var_tree.rs:735) добавить:

```rust
    /// Replay `on_init` for every live entry. Used when no migration runs
    /// (Db calls this after `run_migration` returns `false`). Public users
    /// should invoke `migrate(|_, _| MigrateAction::Keep)` instead.
    pub(crate) fn replay_init(&self) {
        if !H::NEEDS_INIT {
            return;
        }
        let guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let value = match self.read_value_cached(node, &guard) {
                Some(v) => v,
                None => continue,
            };
            self.hook.on_init(&node.key, &value);
        }
    }
```

- [ ] **Step 1.6: Прогнать тест и подтвердить PASS**.

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

- [ ] **Step 1.7: Добавить no-op тест для `NoHook`**. В `mod tests`:

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

        for i in 0u64..3 {
            tree.put(&i.to_be_bytes(), &[i as u8; 8]).expect("put");
        }
        // Должно завершиться мгновенно и без эффектов.
        tree.replay_init();
        // sanity: дерево не повреждено.
        assert!(tree.get(&0u64.to_be_bytes()).is_some());
    }
```

- [ ] **Step 1.8: Прогнать оба теста**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree_replay_init)'`
Expected: 2 теста PASS.

- [ ] **Step 1.9: Commit**.

```bash
git add armdb/src/var_tree.rs
git commit -m "$(cat <<'EOF'
feat(armdb/var_tree): add pub(crate) replay_init for NEEDS_INIT hooks

Walks the SkipList and fires on_init for every live entry. Used by Db
after run_migration returns false. Compiles to a no-op for NoHook.

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

---

## Task 2: VarTree::migrate — on_init + hook-free writes

**Files:**
- Modify: `armdb/src/var_tree.rs` — переписать тело `migrate` (var_tree.rs:693-735); добавить 4 теста.

### Step 2.1: Написать failing-тесты migrate (Keep/Update/Delete/sanity-on_write)

- [ ] В `mod tests` (`var_tree.rs:1368+`) добавить:

```rust
    #[test]
    fn var_tree_migrate_keep_fires_on_init_not_on_write_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        for i in 0u64..4 {
            tree.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        let mutated = tree.migrate(|_, _| MigrateAction::Keep).expect("migrate");

        assert_eq!(mutated, 0);
        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 4, "4 keeps -> 4 on_init");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 0, "Keep must not fire on_write");
    }

    #[test]
    fn var_tree_migrate_update_fires_on_init_with_new_value_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 42u64.to_be_bytes();
        tree.put(&key, b"old-value").expect("put");
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        let new = ByteView::new(b"new-value");
        let mutated = tree
            .migrate(move |_, _| MigrateAction::Update(new.clone()))
            .expect("migrate");

        assert_eq!(mutated, 1);
        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 1);
        assert_eq!(
            tree.hook.last_init_value.lock().as_deref(),
            Some(b"new-value".as_ref()),
            "on_init must receive the NEW value"
        );
        assert_eq!(
            tree.hook.writes.load(AtomicOrdering::Relaxed),
            0,
            "Update must NOT fire on_write (was double-firing through self.put)"
        );
        assert_eq!(tree.get(&key).unwrap().as_bytes(), b"new-value");
    }

    #[test]
    fn var_tree_migrate_delete_fires_no_hooks_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 7u64.to_be_bytes();
        tree.put(&key, b"x").expect("put");
        tree.hook.writes.store(0, AtomicOrdering::Relaxed);
        tree.hook.inits.store(0, AtomicOrdering::Relaxed);

        let mutated = tree.migrate(|_, _| MigrateAction::Delete).expect("migrate");

        assert_eq!(mutated, 1);
        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 0);
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 0);
        assert!(tree.get(&key).is_none());
    }

    #[test]
    fn var_tree_public_put_still_fires_on_write_once() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        tree.put(&1u64.to_be_bytes(), b"v").expect("put");
        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }
```

- [ ] **Step 2.2: Прогнать и подтвердить FAIL** (текущий migrate фитит on_write для Keep, double-fires для Update).

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree_migrate_keep_fires) | test(var_tree_migrate_update_fires) | test(var_tree_migrate_delete_fires)'`
Expected: 3 теста FAIL (Keep — fires `on_write`; Update — `writes != 0`; Delete — `writes != 0`).

- [ ] **Step 2.3: Переписать `VarTree::migrate`**. Заменить тело метода `pub fn migrate` (var_tree.rs:693-735) на:

```rust
    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// The callback receives each (key, value_bytes) and returns `MigrateAction`:
    /// - `Keep` — no change (fires `on_init` if `H::NEEDS_INIT`)
    /// - `Update(new_value)` — replace value (hook-free write, fires `on_init`)
    /// - `Delete` — remove entry (hook-free tombstone, no hooks)
    ///
    /// `on_write` is NEVER fired during migrate — see `armdb/docs/hooks.md`.
    /// Returns the number of mutated entries.
    pub fn migrate(
        &self,
        f: impl Fn(&K, &[u8]) -> crate::MigrateAction<ByteView>,
    ) -> DbResult<usize> {
        use crate::MigrateAction;

        let guard = self.index.collector().enter();
        let mut current = crate::skiplist::strip_mark(unsafe {
            (*self.index.head_ptr())
                .tower(0)
                .load(std::sync::atomic::Ordering::Acquire)
        });
        let mut count = 0;
        while !current.is_null() {
            let node = unsafe { &*current };
            current = crate::skiplist::strip_mark(
                node.tower(0).load(std::sync::atomic::Ordering::Acquire),
            );
            if node.is_marked() {
                continue;
            }
            let value = match self.read_value_cached(node, &guard) {
                Some(v) => v,
                None => continue,
            };
            match f(&node.key, &value) {
                MigrateAction::Keep => {
                    if H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &value);
                    }
                }
                MigrateAction::Update(new_value) => {
                    if H::NEEDS_INIT {
                        self.hook.on_init(&node.key, &new_value);
                    }
                    let shard_id = self.shard_for(&node.key);
                    let mut inner = self.engine.shards()[shard_id].lock();
                    let g = self.index.collector().enter();
                    self.put_locked(shard_id, &mut inner, &g, &node.key, &new_value)?;
                    count += 1;
                }
                MigrateAction::Delete => {
                    let shard_id = self.shard_for(&node.key);
                    let mut inner = self.engine.shards()[shard_id].lock();
                    let g = self.index.collector().enter();
                    self.delete_locked(shard_id, &mut inner, &g, &node.key)?;
                    count += 1;
                }
            }
        }

        tracing::info!(mutations = count, "var_tree migration complete");
        Ok(count)
    }
```

Ключевые отличия от старого тела:

- `MigrateAction::Keep` теперь не зовёт `on_write`; зовёт `on_init` под `H::NEEDS_INIT`.
- `MigrateAction::Update(...)` зовёт `on_init` (под `H::NEEDS_INIT`) до записи, потом берёт shard lock + новый guard и зовёт `put_locked` напрямую (а не `self.put`, который дополнительно фитил бы `on_write`).
- `MigrateAction::Delete` зовёт `delete_locked` напрямую вместо `self.delete`.

- [ ] **Step 2.4: Прогнать новые тесты и подтвердить PASS**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree_migrate_keep_fires) | test(var_tree_migrate_update_fires) | test(var_tree_migrate_delete_fires) | test(var_tree_public_put_still_fires)'`
Expected: 4 PASS.

- [ ] **Step 2.5: Прогнать все тесты `var_tree` и убедиться, что нет регрессий** (особенно вокруг `migrate` и iter/get).

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree)'`
Expected: все PASS. Если какой-то старый тест опирался на старое поведение `migrate Keep → on_write` — починить ожидания.

- [ ] **Step 2.6: Commit**.

```bash
git add armdb/src/var_tree.rs
git commit -m "$(cat <<'EOF'
fix(armdb/var_tree): migrate fires on_init under NEEDS_INIT, never on_write

Keep/Update fire on_init via H::NEEDS_INIT; Update/Delete call hook-free
put_locked/delete_locked directly instead of public put/delete (which
were double-firing on_write). Matches TypedTree::migrate contract from
armdb/docs/hooks.md.

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

---

## Task 3: VarMap::replay_init() (raw + tests)

**Files:**
- Modify: `armdb/src/var_map.rs` — добавить `replay_init`; добавить `CountingHook` + helper + тесты в `mod tests` (var_map.rs:778+).

### Step 3.1: Добавить `CountingHook` и helper в `mod tests` var_map.rs

- [ ] Открыть `armdb/src/var_map.rs`. В `mod tests {` (var_map.rs:778) сразу после `use tempfile::tempdir;` (var_map.rs:782) добавить тот же `CountingHook` и helper:

```rust
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    #[derive(Default)]
    struct CountingHook<const NEEDS_INIT: bool, const NEEDS_OLD: bool> {
        writes: AtomicUsize,
        writes_with_old: AtomicUsize,
        inits: AtomicUsize,
        last_write_new: parking_lot::Mutex<Option<Vec<u8>>>,
        last_init_value: parking_lot::Mutex<Option<Vec<u8>>>,
    }

    impl<const NEEDS_INIT: bool, const NEEDS_OLD: bool> WriteHook<[u8; 8]>
        for CountingHook<NEEDS_INIT, NEEDS_OLD>
    {
        const NEEDS_OLD_VALUE: bool = NEEDS_OLD;
        const NEEDS_INIT: bool = NEEDS_INIT;

        fn on_write(&self, _key: &[u8; 8], old: Option<&[u8]>, new: Option<&[u8]>) {
            self.writes.fetch_add(1, AtomicOrdering::Relaxed);
            if old.is_some() {
                self.writes_with_old.fetch_add(1, AtomicOrdering::Relaxed);
            }
            *self.last_write_new.lock() = new.map(<[u8]>::to_vec);
        }

        fn on_init(&self, _key: &[u8; 8], value: &[u8]) {
            self.inits.fetch_add(1, AtomicOrdering::Relaxed);
            *self.last_init_value.lock() = Some(value.to_vec());
        }
    }

    fn open_test_map_hooked<const NEEDS_INIT: bool, const NEEDS_OLD: bool>(
        dir: &std::path::Path,
        hook: CountingHook<NEEDS_INIT, NEEDS_OLD>,
    ) -> VarMap<[u8; 8], CountingHook<NEEDS_INIT, NEEDS_OLD>> {
        let mut cfg = Config::test();
        cfg.shard_count = 1;
        cfg.max_file_size = 4096;
        cfg.write_buffer_size = 8192;
        cfg.compaction_threshold = 0.0;
        VarMap::open_hooked(dir, cfg, hook).expect("open hooked test map")
    }
```

### Step 3.2: Написать failing-тест на `replay_init`

- [ ] В `mod tests` добавить:

```rust
    #[test]
    fn var_map_replay_init_fires_on_init_per_live_key_raw() {
        let dir = tempdir().unwrap();
        let map = open_test_map_hooked::<true, false>(dir.path(), CountingHook::default());

        for i in 0u64..5 {
            map.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        map.delete(&3u64.to_be_bytes()).expect("delete");

        map.hook.writes.store(0, AtomicOrdering::Relaxed);
        map.hook.inits.store(0, AtomicOrdering::Relaxed);

        map.replay_init();

        assert_eq!(map.hook.inits.load(AtomicOrdering::Relaxed), 4);
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 0);
    }

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

        for i in 0u64..3 {
            map.put(&i.to_be_bytes(), &[i as u8; 8]).expect("put");
        }
        map.replay_init();
        assert!(map.get(&0u64.to_be_bytes()).is_some());
    }
```

- [ ] **Step 3.3: Прогнать и подтвердить FAIL** (метод отсутствует).

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_map_replay_init)'`
Expected: компиляция падает с `no method named 'replay_init'`.

- [ ] **Step 3.4: Реализовать `replay_init`** в `impl<K: Key + Hash + Eq, H: WriteHook<K>> VarMap<K, H>`. Найти метод `migrate` (var_map.rs:407) и сразу после его закрывающей `}` добавить:

```rust
    /// Replay `on_init` for every live entry. Used when no migration runs
    /// (Db calls this after `run_migration` returns `false`).
    ///
    /// Lock order: snapshot keys under per-shard index lock, release the
    /// lock, then call `self.get(key)` which re-acquires the index lock
    /// briefly to copy the `DiskLoc` and reads the value WITHOUT holding
    /// the index lock. This matches the lock order of public writes
    /// (shard → index) and avoids the inverted order that would deadlock
    /// against an active writer.
    pub(crate) fn replay_init(&self) {
        if !H::NEEDS_INIT {
            return;
        }
        for i in 0..self.engine.shards().len() {
            let keys: Vec<K> = {
                let index = sync::lock(&self.indexes[i]);
                index.keys().copied().collect()
            };
            for key in keys {
                let value = match self.get(&key) {
                    Some(v) => v,
                    None => continue,
                };
                self.hook.on_init(&key, &value);
            }
        }
    }
```

- [ ] **Step 3.5: Прогнать тесты и подтвердить PASS**.

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

- [ ] **Step 3.6: Commit**.

```bash
git add armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
feat(armdb/var_map): add pub(crate) replay_init for NEEDS_INIT hooks

Per-shard snapshot of keys then self.get(key) outside the index lock.
This preserves the public shard→index lock order and avoids deadlock
against concurrent writers. Compiles to a no-op for NoHook.

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

---

## Task 4: VarMap::migrate — on_init + hook-free writes

**Files:**
- Modify: `armdb/src/var_map.rs` — переписать тело `migrate` (var_map.rs:407-442); добавить 4 теста.

### Step 4.1: Написать failing-тесты

- [ ] В `mod tests` var_map.rs добавить:

```rust
    #[test]
    fn var_map_migrate_keep_fires_on_init_not_on_write_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let map = open_test_map_hooked::<true, false>(dir.path(), CountingHook::default());

        for i in 0u64..4 {
            map.put(&i.to_be_bytes(), &[i as u8; 16]).expect("put");
        }
        map.hook.writes.store(0, AtomicOrdering::Relaxed);
        map.hook.inits.store(0, AtomicOrdering::Relaxed);

        let mutated = map.migrate(|_, _| MigrateAction::Keep).expect("migrate");

        assert_eq!(mutated, 0);
        assert_eq!(map.hook.inits.load(AtomicOrdering::Relaxed), 4);
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 0);
    }

    #[test]
    fn var_map_migrate_update_fires_on_init_with_new_value_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let map = open_test_map_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 42u64.to_be_bytes();
        map.put(&key, b"old-value").expect("put");
        map.hook.writes.store(0, AtomicOrdering::Relaxed);
        map.hook.inits.store(0, AtomicOrdering::Relaxed);

        let new = ByteView::new(b"new-value");
        let mutated = map
            .migrate(move |_, _| MigrateAction::Update(new.clone()))
            .expect("migrate");

        assert_eq!(mutated, 1);
        assert_eq!(map.hook.inits.load(AtomicOrdering::Relaxed), 1);
        assert_eq!(
            map.hook.last_init_value.lock().as_deref(),
            Some(b"new-value".as_ref())
        );
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 0);
        assert_eq!(map.get(&key).unwrap().as_bytes(), b"new-value");
    }

    #[test]
    fn var_map_migrate_delete_fires_no_hooks_raw() {
        use crate::MigrateAction;
        let dir = tempdir().unwrap();
        let map = open_test_map_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 7u64.to_be_bytes();
        map.put(&key, b"x").expect("put");
        map.hook.writes.store(0, AtomicOrdering::Relaxed);
        map.hook.inits.store(0, AtomicOrdering::Relaxed);

        let mutated = map.migrate(|_, _| MigrateAction::Delete).expect("migrate");

        assert_eq!(mutated, 1);
        assert_eq!(map.hook.inits.load(AtomicOrdering::Relaxed), 0);
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 0);
        assert!(map.get(&key).is_none());
    }

    #[test]
    fn var_map_public_put_still_fires_on_write_once() {
        let dir = tempdir().unwrap();
        let map = open_test_map_hooked::<true, false>(dir.path(), CountingHook::default());

        map.put(&1u64.to_be_bytes(), b"v").expect("put");
        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 1);
    }
```

- [ ] **Step 4.2: Прогнать и подтвердить FAIL**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_map_migrate_keep_fires) | test(var_map_migrate_update_fires) | test(var_map_migrate_delete_fires)'`
Expected: 3 FAIL.

- [ ] **Step 4.3: Переписать `VarMap::migrate`**. Заменить тело метода `pub fn migrate` (var_map.rs:407-442) на:

```rust
    /// Iterate all entries and optionally mutate them. Call once at startup.
    ///
    /// - `Keep` — no change (fires `on_init` if `H::NEEDS_INIT`)
    /// - `Update(new_value)` — replace value (hook-free write, fires `on_init`)
    /// - `Delete` — remove entry (hook-free tombstone, no hooks)
    ///
    /// `on_write` is NEVER fired during migrate.
    /// Returns the number of mutated entries.
    pub fn migrate(
        &self,
        f: impl Fn(&K, &[u8]) -> crate::MigrateAction<ByteView>,
    ) -> DbResult<usize> {
        use crate::MigrateAction;

        let mut count = 0;
        for i in 0..self.engine.shards().len() {
            let keys: Vec<K> = {
                let index = sync::lock(&self.indexes[i]);
                index.keys().copied().collect()
            };
            for key in keys {
                let value = match self.get(&key) {
                    Some(v) => v,
                    None => continue,
                };
                match f(&key, &value) {
                    MigrateAction::Keep => {
                        if H::NEEDS_INIT {
                            self.hook.on_init(&key, &value);
                        }
                    }
                    MigrateAction::Update(new_value) => {
                        if H::NEEDS_INIT {
                            self.hook.on_init(&key, &new_value);
                        }
                        let shard_id = self.shard_for(&key);
                        let mut inner = self.engine.shards()[shard_id].lock();
                        let mut index = sync::lock(&self.indexes[shard_id]);
                        self.put_locked(shard_id, &mut inner, &mut index, &key, &new_value)?;
                        count += 1;
                    }
                    MigrateAction::Delete => {
                        let shard_id = self.shard_for(&key);
                        let mut inner = self.engine.shards()[shard_id].lock();
                        let mut index = sync::lock(&self.indexes[shard_id]);
                        self.delete_locked(shard_id, &mut inner, &mut index, &key)?;
                        count += 1;
                    }
                }
            }
        }

        tracing::info!(mutations = count, "var_map migration complete");
        Ok(count)
    }
```

Лок-порядок (shard → index) совпадает с public `put`/`delete` (var_map.rs:199-200, 222-223, 235-236). Snapshot ключей делается под отдельным коротким lock'ом на `indexes[i]` (без shard lock), потом отпускается до чтения значения через `self.get` — тот же паттерн, что в текущей миграции (var_map.rs:413-422).

- [ ] **Step 4.4: Прогнать новые тесты и подтвердить PASS**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_map_migrate_keep_fires) | test(var_map_migrate_update_fires) | test(var_map_migrate_delete_fires) | test(var_map_public_put_still_fires)'`
Expected: 4 PASS.

- [ ] **Step 4.5: Прогнать все тесты `var_map`** на регрессии.

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

- [ ] **Step 4.6: Commit**.

```bash
git add armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
fix(armdb/var_map): migrate fires on_init under NEEDS_INIT, never on_write

Keep/Update fire on_init via H::NEEDS_INIT; Update/Delete call hook-free
put_locked/delete_locked directly instead of public put/delete (which
were double-firing on_write). Matches TypedMap::migrate contract from
armdb/docs/hooks.md.

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

---

## Task 5: Sanity-тесты для `atomic()` (Var* hook-free contract)

**Files:**
- Modify: `armdb/src/var_tree.rs`, `armdb/src/var_map.rs` — добавить по одному тесту.

### Step 5.1: Тест `var_tree_atomic_does_not_fire_hooks`

- [ ] В `mod tests` var_tree.rs добавить:

```rust
    #[test]
    fn var_tree_atomic_does_not_fire_hooks() {
        let dir = tempdir().unwrap();
        let tree = open_test_tree_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 1u64.to_be_bytes();
        tree.atomic(&key, |shard| {
            shard.put(&key, b"a")?;
            shard.delete(&key)?;
            Ok(())
        })
        .expect("atomic");

        assert_eq!(tree.hook.writes.load(AtomicOrdering::Relaxed), 0);
        assert_eq!(tree.hook.inits.load(AtomicOrdering::Relaxed), 0);
    }
```

- [ ] **Step 5.2: Тест `var_map_atomic_does_not_fire_hooks`** в var_map.rs:

```rust
    #[test]
    fn var_map_atomic_does_not_fire_hooks() {
        let dir = tempdir().unwrap();
        let map = open_test_map_hooked::<true, false>(dir.path(), CountingHook::default());

        let key = 1u64.to_be_bytes();
        map.atomic(&key, |shard| {
            shard.put(&key, b"a")?;
            shard.delete(&key)?;
            Ok(())
        })
        .expect("atomic");

        assert_eq!(map.hook.writes.load(AtomicOrdering::Relaxed), 0);
        assert_eq!(map.hook.inits.load(AtomicOrdering::Relaxed), 0);
    }
```

- [ ] **Step 5.3: Прогнать**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(atomic_does_not_fire_hooks)'`
Expected: 2 PASS (поведение `_locked` уже hook-free, тесты только закрепляют контракт).

- [ ] **Step 5.4: Commit**.

```bash
git add armdb/src/var_tree.rs armdb/src/var_map.rs
git commit -m "$(cat <<'EOF'
test(armdb/var_*): atomic() does not fire write or init hooks

Pin the existing hook-free contract for VarShard / VarMapShard methods
called inside atomic() so future refactors don't reintroduce double-fire.

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

---

## Task 6: Wire `replay_init` через `VarTypedTree` / `VarTypedMap` и в `Db`

**Files:**
- Modify: `armdb/src/var_typed_tree.rs` — добавить `pub(crate) fn replay_init` рядом с `migrate` (var_typed_tree.rs:357-375).
- Modify: `armdb/src/var_typed_map.rs` — то же рядом с `migrate` (var_typed_map.rs:229-247).
- Modify: `armdb/src/armour/db.rs` — две правки в `open_var_typed_tree` (db.rs:333) и `open_var_typed_map` (db.rs:368).

### Step 6.1: Добавить `replay_init` в `VarTypedTree`

- [ ] В `armdb/src/var_typed_tree.rs` в `impl<...> VarTypedTree<...>` сразу после метода `migrate` (заканчивается на var_typed_tree.rs:375 `}`) добавить:

```rust
    /// Replay `on_init` for every live entry. Used by Db when no migration ran.
    pub(crate) fn replay_init(&self) {
        self.inner.replay_init();
    }
```

### Step 6.2: Добавить `replay_init` в `VarTypedMap`

- [ ] В `armdb/src/var_typed_map.rs` в `impl<...> VarTypedMap<...>` сразу после метода `migrate` (заканчивается на var_typed_map.rs:247 `}`) добавить:

```rust
    /// Replay `on_init` for every live entry. Used by Db when no migration ran.
    pub(crate) fn replay_init(&self) {
        self.inner.replay_init();
    }
```

### Step 6.3: Wire в `Db::open_var_typed_tree`

- [ ] В `armdb/src/armour/db.rs` найти строку 333:

```rust
        let _ = self.run_migration(&meta, &stored, migrations, |mfn| tree.migrate(mfn))?;
```

Заменить на:

```rust
        let migrated = self.run_migration(&meta, &stored, migrations, |mfn| tree.migrate(mfn))?;
        if !migrated {
            tree.replay_init();
        }
```

### Step 6.4: Wire в `Db::open_var_typed_map`

- [ ] В `armdb/src/armour/db.rs` найти строку 368:

```rust
        let _ = self.run_migration(&meta, &stored, migrations, |mfn| map.migrate(mfn))?;
```

Заменить на:

```rust
        let migrated = self.run_migration(&meta, &stored, migrations, |mfn| map.migrate(mfn))?;
        if !migrated {
            map.replay_init();
        }
```

### Step 6.5: Db-orchestration интеграционные тесты

- [ ] Создать файл `armdb/tests/var_typed_replay_init.rs`:

```rust
#![cfg(all(feature = "var-collections", feature = "typed-tree", feature = "armour"))]

use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

use armdb::{
    Codec, CollectionMeta, Config, Db, NoHook, RapiraCodec, TypedWriteHook,
};

/// Тестовый Record для VarTypedTree/Map. Каждый тест-кейс открывает
/// фиктивный schema (имя + версия). Версия меняется ВРУЧНУЮ через
/// fixture'ы для проверки migrate-пути.
#[derive(Clone, Debug, PartialEq)]
struct V1 {
    payload: u32,
}

impl armdb::Key for u64 {}

impl CollectionMeta for V1 {
    const NAME: &'static str = "var_typed_replay_init_v1";
    const VERSION: u32 = 1;
    type SelfId = u64;
}

// Rapira-codec implementation: подставить ровно тот импорт/derive,
// которым код проекта подключает RapiraCodec в существующих тестах.
// Пример паттерна: см. armdb/tests/rpc_schema.rs:19-33 (impl CollectionMeta,
// затем .open_typed_tree::<TestUser, RapiraCodec, _>).

struct InitCounter {
    inits: AtomicUsize,
    writes: AtomicUsize,
}

impl InitCounter {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            inits: AtomicUsize::new(0),
            writes: AtomicUsize::new(0),
        })
    }
}

impl TypedWriteHook<u64, V1> for Arc<InitCounter> {
    const NEEDS_OLD_VALUE: bool = false;
    const NEEDS_INIT: bool = true;

    fn on_write(&self, _key: &u64, _old: Option<&V1>, _new: Option<&V1>) {
        self.writes.fetch_add(1, AtomicOrdering::Relaxed);
    }

    fn on_init(&self, _key: &u64, _value: &V1) {
        self.inits.fetch_add(1, AtomicOrdering::Relaxed);
    }
}

#[test]
fn db_open_var_typed_tree_calls_replay_init_when_no_migration() {
    let dir = tempfile::tempdir().unwrap();
    let cfg = Config::test();

    // 1) Заполним базу версии 1 без хука (NoHook).
    {
        let db = Db::open(dir.path(), Config::test()).expect("open");
        let tree = db
            .open_var_typed_tree::<V1, RapiraCodec, _>(cfg.clone(), NoHook, &[])
            .expect("open v1");
        for i in 0u64..3 {
            tree.put(&i, &V1 { payload: i as u32 }).expect("put");
        }
    }

    // 2) Откроем снова — версия та же, миграции нет → должен вызваться replay_init.
    let db = Db::open(dir.path(), Config::test()).expect("open2");
    let counter = InitCounter::new();
    let _tree = db
        .open_var_typed_tree::<V1, RapiraCodec, _>(cfg, counter.clone(), &[])
        .expect("reopen v1 hooked");

    assert_eq!(counter.inits.load(AtomicOrdering::Relaxed), 3,
        "replay_init must fire on_init for each live key");
    assert_eq!(counter.writes.load(AtomicOrdering::Relaxed), 0,
        "no writes occurred — on_write must stay at 0");
}

#[test]
fn db_open_var_typed_tree_skips_replay_init_when_migration_ran() {
    // Этот тест проверяет, что при schema bump (миграция запущена)
    // replay_init дополнительно НЕ зовётся: on_init фитится ровно
    // один раз — внутри migrate. Если в проекте на момент исполнения
    // плана нет инфраструктуры для schema bump в одном процессе,
    // выразить тест через два cfg-варианта (V1 и V2: CollectionMeta
    // с тем же NAME но VERSION=2). При невозможности оставить TODO,
    // см. armdb/tests/rpc_schema.rs паттерны.
    //
    // ОЖИДАНИЕ: counter.inits = N (от migrate), не 2*N (replay не
    // должен дублировать).
    //
    // Минимальная реализация: открыть V1 без хука → закрыть → открыть
    // V2 с миграцией KeepAll (или Update(noop)) и хуком → проверить inits.
    //
    // Если требует существенной инфраструктуры — может быть отложено
    // и помечено как `#[ignore]` с TODO; основное покрытие даёт первый
    // тест и raw migrate-тесты из Task 2/4. Принять решение во время
    // имплементации.
}

// Аналогичные 2 теста для VarTypedMap.
```

> **Note для имплементера:** RapiraCodec / Db / armour-feature setup — см. `armdb/tests/rpc_schema.rs:1-40` как опорный пример (там уже `impl CollectionMeta`, `Db::open`, `open_typed_tree::<…, RapiraCodec, _>`). Адаптировать те же import'ы и зависимости. Второй тест («skips_replay_init_when_migration_ran») может потребовать `TypedMigration` setup — если в crate нет ready-made helper'ов для schema bump в одном процессе, упростить до проверки «migrate отработал — counter ровно = N, не 2N» через `MigrateAction::Keep`-замыкание.

- [ ] **Step 6.6: Прогнать**.

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

- [ ] **Step 6.7: Зеркальные тесты для VarTypedMap** — добавить в тот же файл по образцу `tree`-варианта, заменив `open_var_typed_tree` → `open_var_typed_map`.

- [ ] **Step 6.8: Прогнать все Var*-тесты целиком**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(var_tree) | test(var_map) | test(var_typed)'`
Expected: все PASS.

- [ ] **Step 6.9: Commit**.

```bash
git add armdb/src/var_typed_tree.rs armdb/src/var_typed_map.rs armdb/src/armour/db.rs armdb/tests/var_typed_replay_init.rs
git commit -m "$(cat <<'EOF'
feat(armdb): Db wires replay_init for VarTypedTree/VarTypedMap

Mirrors TypedTree/TypedMap orchestration: if run_migration returns
false, call replay_init() so hooks with NEEDS_INIT see every live key
exactly once. Pub(crate) replay_init on the typed wrappers delegates to
the inner VarTree/VarMap; VarTypedHookAdapter already decodes on_init.

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

---

## Task 7: Обновить `armdb/docs/hooks.md`

**Files:**
- Modify: `armdb/docs/hooks.md` — три точечные правки.

### Step 7.1: Таблица «Какие коллекции поддерживают хуки» (line 221-234)

- [ ] Открыть `armdb/docs/hooks.md`. Заменить строки 229-234 (для VarTree/VarMap):

Старое:
```
| VarTree | да | — | нет* |
| VarMap | да | — | нет* |

*VarTree/VarMap поддерживают `on_write`, но `on_init` не реализован —
значения хранятся на диске, а не в in-memory индексе, поэтому
hint-based recovery не содержит байты значений.
```

Новое:
```
| VarTree | да | — | да |
| VarMap | да | — | да |
```

(сноска `*` целиком удаляется.)

### Step 7.2: Раздел «Жизненный цикл при открытии коллекции» (line 93-125)

- [ ] В разделе «Жизненный цикл при открытии коллекции» добавить уточнение в конец описания шагов 2a/2b (после строки 108 `replay_init для каждой live-записи`):

```
VarTree/VarMap следуют той же модели: `migrate` использует hook-free
`put_locked`/`delete_locked` и фитит `on_init` для `Keep`/`Update` под
`H::NEEDS_INIT`. `replay_init` для Var* читает значения с диска через
существующий cached-read path (snapshot ключей по шардам + `get`).
```

### Step 7.3: Таблица «Когда вызываются хуки» (line 67-80) — без правок

- [ ] Подтвердить, что текущие строки таблицы уже описывают целевое
поведение (`migrate() Keep → on_init да, on_write нет` и т.д.) — изменения не требуются.

### Step 7.4: Commit

```bash
git add armdb/docs/hooks.md
git commit -m "$(cat <<'EOF'
docs(armdb/hooks): VarTree/VarMap now support on_init via replay_init

Update the support matrix and remove the obsolete footnote. Note in the
lifecycle section that Var* mirror the TypedTree/TypedMap model with
disk-backed value reads during replay_init.

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

---

## Task 8: Полная верификация

**Files:** —

- [ ] **Step 8.1: Формат**.

Run: `cargo fmt --check`
Expected: PASS (нет дифа).

- [ ] **Step 8.2: Clippy под var-collections**.

Run: `cargo clippy -p armdb --features var-collections -- -D warnings`
Expected: 0 предупреждений.

- [ ] **Step 8.3: Clippy под var-collections + encryption**.

Run: `cargo clippy -p armdb --features "var-collections encryption" -- -D warnings`
Expected: 0 предупреждений.

- [ ] **Step 8.4: Тесты под var-collections**.

Run: `cargo nextest run -p armdb --features var-collections`
Expected: всё PASS.

- [ ] **Step 8.5: Тесты под var-collections + encryption**.

Run: `cargo nextest run -p armdb --features "var-collections encryption"`
Expected: всё PASS.

- [ ] **Step 8.6: Прицельный прогон новых тестов**.

Run: `cargo nextest run -p armdb --features var-collections -E 'test(replay_init) | test(migrate_keep_fires) | test(migrate_update_fires) | test(migrate_delete_fires) | test(atomic_does_not_fire_hooks) | test(public_put_still_fires)'`
Expected: все новые тесты PASS.

- [ ] **Step 8.7: Финальный коммит** (если фикстуры/форматирование что-то поправили).

Если `cargo fmt` или подсказки clippy потребовали правок — закоммитить отдельным коммитом:

```bash
git add -A
git commit -m "chore(armdb): fmt/clippy follow-ups after migrate-hooks fix"
```

Если правок не было — пропустить.

---

## Self-Review Checklist (для имплементера)

После всех тасков убедиться:

- [ ] `VarTree::migrate Keep` больше не зовёт `on_write` (Task 2).
- [ ] `VarTree::migrate Update` зовёт `on_init` (под NEEDS_INIT) с **новым** значением и `on_write` ровно 0 раз (Task 2).
- [ ] `VarTree::migrate Delete` не зовёт ни `on_write`, ни `on_init` (Task 2).
- [ ] То же для `VarMap::migrate` (Task 4).
- [ ] `VarTree::replay_init` и `VarMap::replay_init` существуют, `pub(crate)`, корректно гейтятся `NEEDS_INIT` (Task 1, 3).
- [ ] `VarMap::replay_init` использует snapshot-then-get (нет index lock'а во время `read_value_cached_inner`) — Task 3.
- [ ] `VarTypedTree::replay_init` и `VarTypedMap::replay_init` существуют и делегируют (Task 6).
- [ ] `Db::open_var_typed_tree` / `open_var_typed_map` зовут `replay_init()` при `!migrated` (Task 6).
- [ ] Публичные `put`/`insert`/`delete` фитят `on_write` ровно один раз (тесты `*_public_put_still_fires_on_write_once`) — Task 2, 4.
- [ ] `atomic()` не фитит хуки (Task 5).
- [ ] `armdb/docs/hooks.md`: VarTree/VarMap on_init = `да`, сноска удалена, lifecycle упоминает Var* (Task 7).
- [ ] Все verification команды из Task 8 проходят.
