# FixedStore Replication — Design

**Date:** 2026-04-17
**Status:** Design complete, ready for implementation
**Scope:** `armdb/src/fixed_replication/` (new module), changes to `armdb/src/fixed/{slot,shard,engine}.rs`

---

## 1. Цель и контекст

FixedStore (fixed-slot pwrite backend) пока не поддерживает репликацию (см. `armdb/docs/fixed_store_design.md`, раздел «Ограничения»). Bitcask replication основана на GSN + log shipping — эта модель неприменима к Fixed'у, потому что:

- Нет append-only лога — слоты перезаписываются in-place.
- Нет GSN — нет глобального монотонного счётчика через процесс.
- Compaction/hint файлов — нет.

Задача: **добавить репликацию для FixedStore** методом **SPSC streaming + slot-version resync** с модульной (wrapping) 30-битной версией per slot.

### Non-goals (v1)

- Admin-triggered `SlotReset` RPC для принудительной resync конкретного слота.
- Wide version (u62) конфигурируемая опция.
- Broadcast replication (несколько активных streaming follower'ов на один shard).
- Shared port с Bitcask replication.
- Encryption replicated slots.

---

## 2. Высокоуровневая архитектура

- **Отдельный модуль** `armdb/src/fixed_replication/` — не пересекается с `armdb/src/replication/` (Bitcask). Feature `replication` покрывает оба.
- **Отдельный порт** на leader'е. Клиенты подключаются per-shard.
- **Per-slot version** (30 бит) в meta-поле слота — заменяет GSN для упорядочивания writes.
- **Modular comparison**: `is_newer(new, old) = (new.wrapping_sub(old) & 0x3FFF_FFFF) < (1 << 29)`. Толерирует follower lag до 2^29 версий на слот.
- **Catch-up stateless** — leader делает full scan всех слотов per shard при подключении follower'а. Нет persistent per-follower state на leader'е.
- **Streaming через SPSC** — follower получает live events через rtrb ring buffer (как Bitcask).
- **Один streaming follower на shard** (ограничение v1). Второй и далее — catch-up only on (re)connect.
- **Lazy SPSC install** — producer'ы ставятся в `FixedShardInner.replication_tx` только после первого follower connect'а (`producers_installed: AtomicBool`). До этого — никакого write-path overhead'а.
- **Overflow → silent drop** + metric. Catch-up at next reconnect подтянет silently lost updates. Hot slots с частыми updates исправляются через streaming (latest event overwrites).

---

## 3. Формат слота v2 (breaking change — нет production данных)

### Layout

```
┌───────────────────────────────────────────────┐
│  meta: u32 (4 bytes LE)                       │
│    bits 0-1:   status (FREE=0, OCC=1, DEL=2)  │
│    bits 2-31:  version (30 бит)               │
│  crc32: u32 (4 bytes LE)                      │
│    OCCUPIED: CRC32(meta_bytes + key + value)  │
│    DELETED:  stale (не проверяется при read)  │
│  key:   [u8; K]                                │
│  value: [u8; V] (физически остаётся и после   │
│                  DELETE — pwrite на delete     │
│                  пишет только 4 байта meta)    │
│  pad:   (0..7 байт до 8-byte alignment)       │
└───────────────────────────────────────────────┘

SLOT_HEADER_SIZE = 8 байт
slot_size(K, V) = align_up(8 + K + V, 8)
```

### Константы и helpers (`fixed/slot.rs`)

```rust
pub const STATUS_FREE:     u8 = 0;
pub const STATUS_OCCUPIED: u8 = 1;
pub const STATUS_DELETED:  u8 = 2;
const STATUS_MASK: u32 = 0b11;
const VERSION_SHIFT: u32 = 2;
const VERSION_MASK: u32 = 0xFFFF_FFFC;

pub const VERSION_WARN_THRESHOLD: u32 = 0x3E00_0000; // ~97% от 2^30

pub fn pack_meta(status: u8, version: u32) -> u32 {
    (version << VERSION_SHIFT) | (status as u32 & STATUS_MASK)
}
pub fn status_of(meta: u32) -> u8 { (meta & STATUS_MASK) as u8 }
pub fn version_of(meta: u32) -> u32 { meta >> VERSION_SHIFT }

pub fn bump_version(meta: u32) -> u32 {
    // +4 инкрементирует version на 1, сохраняет status. wrapping_add обеспечивает
    // wrap at 2^30 автоматически.
    meta.wrapping_add(1 << VERSION_SHIFT)
}

pub fn with_status(meta: u32, new_status: u8) -> u32 {
    (meta & VERSION_MASK) | (new_status as u32 & STATUS_MASK)
}

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

### CRC (изменение)

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

`read_slot` возвращает `None` для `status != STATUS_OCCUPIED` без проверки CRC. Для OCCUPIED — `crc == CRC(meta + key + value)`.

### Write/Delete поведение

- **Write slot**: full pwrite `slot_size` байт с `meta = bump_version(old_meta)` + новый CRC + key + value.
- **Delete slot**: pwrite **только 4 байта** (`new_meta` с status=DELETED). Key и value на диске остаются от предыдущего OCCUPIED. CRC становится stale (но не проверяется при read). `bitmap.clear(slot_id)`.

Atomicity 4-byte delete: aligned u32 store атомарен на CPU и page cache; на SSD sector (≥512B) пишется целиком, торн-write aligned 4B на healthy disk не случается. Даже при гипотетическом torn write результат либо old (OCCUPIED) либо new (DELETED), промежуточных состояний нет.

### db.meta и fixed.data version bump

- `db.meta`: `META_VERSION` 1 → 2.
- `fixed.data` header: `VERSION` 1 → 2.

Breaking change. Старые файлы откажутся открываться с `FormatMismatch`. Production данных нет — ok.

### Замена `fixed.bitmap` на `fixed.versions`

Для clean-shutdown оптимизации: вместо sidecar `fixed.bitmap` (128 KB/млн слотов) пишем `fixed.versions` — `slot_count × 4B` LE u32 (4 MB/млн слотов). Из `versions[]` тривиально восстанавливается bitmap (`status_of(versions[i]) == OCCUPIED`). Cost: +3.9 MB на миллион слотов — приемлемо.

### Целостность sidecar'а (trailer + CRC)

Файл `fixed.versions` уязвим к частичной записи (power loss во время `fdatasync`, disk full). Чтобы отличить валидный sidecar от повреждённого, добавляем trailer:

```
[versions: slot_count × u32 LE]   // 4 × slot_count байт
[slot_count_check: u32 LE]        // дублирует header.slot_count
[crc32: u32 LE]                   // CRC32 всех предыдущих байт (включая slot_count_check)
```

**На load (clean startup):**
1. Читаем header → узнаём expected `slot_count`.
2. Читаем sidecar. Проверяем `file_size == slot_count * 4 + 8`.
3. Читаем trailing `slot_count_check`, сверяем с header.slot_count.
4. Считаем CRC32 первых `(file_size - 4)` байт, сверяем с stored CRC.

**При любом mismatch:** удаляем sidecar, fallback на dirty full scan. Без паники — корректность восстанавливается, только startup медленнее на один запуск.

**На clean shutdown:**
1. Write all `versions[]` bytes.
2. Append `slot_count` (u32 LE).
3. Compute CRC32 of everything written so far.
4. Append CRC.
5. `fsync`.

---

## 4. In-memory state

### `FixedShardInner` — новые поля

```rust
pub struct FixedShardInner {
    file: File,
    dir: PathBuf,
    pub(crate) bitmap: Bitmap,
    pub(crate) versions: Vec<u32>,         // ← НОВОЕ: meta cache per slot_id
    pub(crate) slot_size: u16,
    pub(crate) slot_count: u32,
    key_len: u16,
    value_len: u16,
    shard_id: u8,
    grow_step: u32,
    pending_writes: u32,
    sync_batch_size: u32,
    last_sync: Instant,
    sync_interval: Duration,
    enable_fsync: bool,
    #[cfg(feature = "replication")]
    pub(crate) replication_tx: Option<rtrb::Producer<FixedReplicationEvent>>,
}
```

Память: `versions` = `slot_count × 4B`. Для 1M слотов × 32 шарда = 128 MB на instance.

### Populate `versions` на startup

Одинаковое поведение для leader'а и follower'а. Роль движка (leader vs follower) определяется только запуском `FixedReplicationServer` / `FixedReplicationClient`, не зависит от recovery path'а.

**Clean shutdown** (`clean_shutdown=1`, sidecar `fixed.versions` валиден):
1. Load `fixed.versions` → `versions: Vec<u32>` (с trailer validation, см. §3).
2. Derive bitmap из versions (bit[i] = status_of == OCCUPIED).
3. Для каждого OCCUPIED: pread слот → key/value → index insert.

**Dirty startup (crash или sidecar trailer invalid)**: full slot scan:
1. Для каждого slot_id: read_exact_at slot_size bytes.
2. `versions[i] = meta` (сохраняем всегда — даже для DELETED и для slot'ов с CRC mismatch; см. §9 про torn slot handling).
3. Bitmap.set + index.insert только если OCCUPIED + CRC match.

На clean shutdown оба mode'а пишут sidecar. 100M слотов × 4B = 400MB sidecar — на NVMe fsync занимает ~200-500ms, приемлемо как разовая цена за быстрый restart.

---

## 5. Write hook на leader'е

### SPSC event type

```rust
// fixed_replication/event.rs
pub enum FixedReplicationEvent {
    Write {
        slot_id: u32,
        payload: Vec<u8>,        // full slot_size bytes (meta+crc+key+value)
    },
    Delete {
        slot_id: u32,
        meta: u32,
        key: Vec<u8>,            // для follower index.remove без pread'а
    },
}
```

`Write.payload` — тот же Vec, что использован для pwrite (move, zero-extra-alloc).
`Delete.key` — свежий Vec (alloc приемлем, delete редок в target workload'ах).

### Hook в `write_slot`

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

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

    self.file.write_all_at(&buf, self.slot_offset(slot_id))?;
    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()?; }

    #[cfg(feature = "replication")]
    if let Some(tx) = &mut self.replication_tx {
        let _ = tx.push(FixedReplicationEvent::Write { slot_id, payload: buf });
        // Err (overflow) → event lost. Recovery через full scan на reconnect.
    }

    Ok(new_meta)
}
```

### Hook в `delete_slot`

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

    // Только 4 байта pwrite (aligned u32 atomic).
    self.file.write_all_at(&new_meta.to_le_bytes(), self.slot_offset(slot_id))?;
    self.versions[slot_id as usize] = new_meta;
    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(FixedReplicationEvent::Delete {
            slot_id, meta: new_meta, key: key.to_vec(),
        });
    }

    Ok(new_meta)
}
```

### Wrap-proximity monitoring

```rust
fn maybe_warn_wrap(&mut 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"
        );
    }
}
```

В v1 — без rate limiting. Если логи шумят — добавим позже.

---

## 6. Wire protocol

Отдельный `FixedMessageType` enum (в `fixed_replication/protocol.rs`). Frame format идентичный Bitcask: `[type:u8][len:u32 LE][payload]`.

```rust
#[repr(u8)]
pub enum FixedMessageType {
    SyncRequest = 1,
    ShardInfo   = 2,
    SlotBatch   = 3,
    CaughtUp    = 4,
    Ack         = 5,
    Heartbeat   = 6,
    Error       = 255,
}
```

### Сообщения

**SyncRequest (F→L):** `[shard_id:u8][protocol_version:u8][flags:u8]` — PROTOCOL_VERSION=1 в v1.

Флаги (bits):
- `bit 0 (FLAG_EMPTY_STATE)` — follower сообщает, что его локальное состояние для этого shard'а пустое (0 occupied слотов). Leader МОЖЕТ пропустить отправку DELETED слотов в phase-1 catch-up — экономит трафик (до 2× в базах с высокой churn rate). Если флаг снят — leader шлёт DELETED слоты как обычно (нужно для корректной синхронизации у follower'а с состоянием).
- bits 1-7: reserved (zero).

Клиент выставляет bit 0 когда `inner.bitmap.occupied() == 0` для конкретного shard'а на момент SyncRequest'а. После первого успешного sync'а — флаг уже не ставится на reconnect'ах, потому что follower больше не empty.

**ShardInfo (L→F):** `[shard_count:u8][key_len:u16][value_len:u16][slot_size:u16][current_slot_count:u32]`

**SlotBatch (L→F):** `[shard_id:u8][count:u32][events]`, где каждый event:
- `[slot_id:u32][meta:u32][key:K][value:V if status==OCCUPIED else 0 bytes]`
- Размер OCCUPIED event = `8 + K + V`. DELETED event = `8 + K`.
- Дискриминатор OCCUPIED/DELETED — low 2 бита meta.

**CaughtUp (L→F):** `[shard_id:u8][total_scanned:u64]`

**Ack (F→L):** `[shard_id:u8][applied_count:u64][max_version_seen:u32]`

**Heartbeat / Error:** идентично Bitcask.

### Batch параметры

- `BATCH_MAX_ENTRIES = 256`
- `BATCH_MAX_BYTES = 64 * 1024`
- `TAIL_POLL_MS = 1`
- `HEARTBEAT_INTERVAL = 5s`
- `ACK_INTERVAL = 1000 events`
- `PROTOCOL_VERSION = 1`

---

## 7. Leader side (server)

### Lazy producer install (Mitigation B)

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

impl FixedReplicationServer {
    pub fn start<E: FixedEngineAccess + 'static>(
        bind_addr: SocketAddr,
        engine: Arc<E>,
        signal: ShutdownSignal,
    ) -> DbResult<Self>;
}
```

- `start()` создаёт SPSC pairs (capacity 8192), держит их в pending pool.
- Producers НЕ инсталлируются в shards до первого follower connect'а.
- При первом accept: `producers_installed.swap(true, SeqCst)` → install all producers.
- До первого follower'а — нулевой overhead на leader write path.

### Per-connection flow

1. Read `SyncRequest`, запомнить `flags` (в т.ч. `FLAG_EMPTY_STATE`).
2. Send `ShardInfo` (валидационные поля + current_slot_count).
3. Попытаться взять consumer для shard_id из pending. Если `None` — catch-up only mode.
4. Spawn serve thread:
   - **Phase 1**: `full_scan_shard(skip_deleted = req.flags & FLAG_EMPTY_STATE != 0)` — iterate `0..slot_count` в chunk'ах по 64 KiB, read, skip FREE, serialize OCCUPIED (всегда) и DELETED (если `skip_deleted == false`) events в batch, flush + wait Ack.
   - После scan: send `CaughtUp`.
   - **Phase 2** (только если consumer был взят): `stream_from_spsc` — pop events, serialize, batch'им, отправляем. Non-blocking ack read. Heartbeat на idle.

### Full scan detail

- Read-ahead 64 KiB chunk через `FixedEngineAccess::read_shard_chunk(shard_id, start_slot, count)`.
- Для каждого слота в chunk'е: parse meta → skip если FREE → skip если DELETED и `skip_deleted=true` → иначе serialize event (OCCUPIED с key/value, DELETED с только key).
- Concurrent writes leader'а во время scan'а корректны: modular compare на follower'е разруливает duplicate events (scan + streaming могут доставить один и тот же слот дважды).
- **Empty follower optimization**: когда `skip_deleted=true`, трафик и I/O phase-1 пропорциональны OCCUPIED count'у, не slot_count'у. В DB с 50% churn это ~2× экономия. Empty follower'у DELETED слоты бесполезны (у него индекс пустой — удалять нечего).

### Consumer lifecycle

- Первый follower для shard_id берёт consumer → streaming available.
- При его disconnect → consumer dropped → SPSC продолжает заполняться → overflow.
- Новый follower: consumer уже None → catch-up only.
- v1 ограничение. Документируем.

---

## 8. Follower side (client + apply)

### `FixedReplicationTarget` trait

```rust
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 in the shard. Used by client 
    /// to decide whether to send FLAG_EMPTY_STATE in SyncRequest.
    fn shard_occupied_count(&self, shard_id: u8) -> u32;
    fn grow_shard_to(&self, shard_id: u8, min_slot_count: u32) -> DbResult<()>;
    fn apply_occupied(&self, shard_id: u8, slot_id: u32, meta: u32,
                      key: &[u8], value: &[u8]) -> DbResult<ApplyOutcome>;
    fn apply_deleted(&self, shard_id: u8, slot_id: u32, meta: u32,
                     key: &[u8]) -> DbResult<ApplyOutcome>;
}

pub enum ApplyOutcome { Applied, Skipped }
```

Реализуется для `ConstTree<K, V, H, Fixed>` и `ConstMap<K, V, H, Fixed>`.

### Apply логика (OCCUPIED)

Три ортогональных шага: (1) очистка залежавшейся записи старого ключа 
в индексе, (2) pwrite содержимого слота, (3) upsert нового ключа.

```
lock shard;
if slot_count < slot_id+1: grow;
if !is_newer(new_version, old_version): return Skipped;

// ── Step 1: cleanup old key at this slot_id ───────────────────────
// Если прошлое состояние слота было OCCUPIED, в индексе МОЖЕТ 
// висеть ссылка old_key → slot_id. Это независимо от того, есть 
// ли new_key где-то ещё в индексе.
if status_of(versions[slot_id]) == OCCUPIED {
    // Fast-path: если new_key уже указывает на slot_id, значит 
    // old_key == new_key (обычный update) — чистить не нужно.
    if index.get_slot_id(&new_key) != Some(slot_id) {
        // Нужен pread, чтобы узнать, какой old_key сидит на диске.
        let own_buf = read_slot_header_and_key(slot_id);
        let old_key = decode_key(&own_buf[8..8+K]);
        if old_key != new_key {
            // Важно: old_key мог "переехать" на другой слот — 
            // удаляем ТОЛЬКО если индекс всё ещё указывает на slot_id.
            if index.get_slot_id(&old_key) == Some(slot_id) {
                index.remove(&old_key);
                metrics::counter!("armdb.fixed.stale_cleanup_count").increment(1);
            }
        }
    }
}

// ── Step 2: persist slot to disk ──────────────────────────────────
pwrite full slot bytes;
versions[slot_id] = new_meta;
bitmap.set(slot_id);

// ── Step 3: insert/update new_key → slot_id ───────────────────────
// upsert: заменяет существующий маппинг для new_key (если был — 
// например, ключ "переехал" с другого слота сюда), обновляет inline 
// value через SeqLock.
index.upsert(new_key, new_value, slot_id);

return Applied;
```

Pread (step 1 fast-path miss) срабатывает только при истинной 
divergence — когда slot_id уже был OCCUPIED другим ключом, а 
обычный streaming путь (DELETE→OCCUPIED) не успел долететь 
(SPSC overflow или catch-up с out-of-order по времени). В 
нормальной работе: либо статус был FREE/DELETED (step 1 skip), 
либо new_key уже указывает на slot_id (fast-path hit, обычный 
update one-hot).

### Apply логика (DELETED)

```
lock shard;
if slot_count < slot_id+1: grow;
if !is_newer(new_version, old_version): return Skipped;
pwrite 4 байта meta;
versions[slot_id] = new_meta;
bitmap.clear(slot_id);

// ВАЖНО: удаляем ключ из индекса ТОЛЬКО если он всё ещё указывает 
// на этот slot_id. Ключ мог "переехать" на другой слот (delete+insert 
// с reuse); в catch-up режиме события приходят в порядке slot_id, 
// не темпорально — DELETED для старого slot'а может прийти уже после 
// OCCUPIED для нового.
if index.get_slot_id(&key) == Some(slot_id) {
    index.remove(&key);
}
return Applied;
```

### Client structure

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

impl FixedReplicationClient {
    pub fn start(
        leader_addr: SocketAddr,
        target: Arc<dyn FixedReplicationTarget>,
        signal: ShutdownSignal,
    ) -> DbResult<Self>;
}
```

Per-shard threads. Каждый поток:
- Reconnect loop с exponential backoff (1s → 30s).
- `connect_and_sync`: 
   - Определить `flags`: если `target.shard_occupied_count(shard_id) == 0` → выставить `FLAG_EMPTY_STATE`. Это дешёвый check (bitmap.occupied() — O(1) counter).
   - Handshake: send `SyncRequest { shard_id, protocol_version, flags }`, receive `ShardInfo`.
   - `grow_shard_to(info.current_slot_count)`.
   - apply loop.
- Apply loop: read frame, decode, per-event `target.apply_occupied/deleted`, periodic Ack, periodic cursor save.
- Heartbeat раз в 5 сек на idle.
- Timeout 1s на read — чтобы ловить shutdown signal.

Note: `FLAG_EMPTY_STATE` проверяется на КАЖДОМ reconnect'е, но естественно становится false после первого успешного sync'а — bitmap уже не пустой. Это даёт fast-initial-sync для fresh follower'ов и обычный behavior для подлежавших reconnect'ов.

### Cursor (diagnostic only)

```rust
// fixed_replication/cursor.rs
pub struct FixedReplicationCursor {
    pub shard_id: u8,
    pub applied_total: u64,
    pub max_version_seen: u32,
    pub last_ack_at: SystemTime,
}
```

File `shard_NNN/fixed_repl.cursor`, 40 байт. **НЕ** используется для resume (catch-up всегда full scan) — только для диагностики. Save каждые 1000 applied events.

---

## 9. Recovery и startup scenarios

### Follower restart

1. `FixedEngine::open` — clean (если прошлый shutdown был clean и sidecar валиден) или dirty path. `versions[]` populated и на диске, и в памяти.
2. `FixedReplicationClient::start` → per-shard connect → full scan from leader → apply (modular compare отбрасывает события, которые follower уже имеет) → streaming.

### Leader restart

1. `FixedEngine::open` — clean startup если shutdown был clean, иначе full scan. `versions[]` populated (critical — нужно для правильного `bump_version` от предыдущих значений).
2. `FixedReplicationServer::start` — заново, `producers_installed = false`, lazy install на первый новый connect.
3. Live follower'ы получают disconnect → reconnect → новый full scan → sync converges.

### Обработка torn slot'а при dirty recovery

При full scan CRC слота может не совпасть (torn write при сбое питания в середине slot-size pwrite). Важно: **нельзя сбрасывать `versions[slot_id] = 0`** — это сломает modular compare. Если leader потом bump'нет version с 0 → 1, а у какого-то follower'а для этого slot_id уже был version=N (большой), `is_newer(1, N)` вернёт false (обычно ложно older), и все последующие writes от leader'а для этого slot'а будут отвергнуты — **permanent divergence**.

**Правильное поведение:**

```rust
// meta занимает первые 4 байта, aligned u32 — атомарна на уровне
// page cache и NVMe. Практически не рвётся.
// Обычный torn сценарий: meta валидна, key/value недописаны → CRC mismatch.
// Сохраняем version из meta (надёжная), но сбрасываем status в FREE,
// чтобы аллокатор мог переиспользовать слот.
let meta = u32::from_le_bytes(buf[0..4].try_into().unwrap());
bitmap.clear(slot_id);
versions[slot_id] = pack_meta(STATUS_FREE, version_of(meta));
// Redo тоже не нужен — allocator на следующем write сделает bump_version 
// от сохранённой version → монотонность сохраняется, followers примут 
// новый write корректно.
```

**Если сама meta битая** (теоретически, редко): `version_of(meta)` — случайное 30-битное число. Максимум — один event от leader'а будет воспринят follower'ами "не в том порядке", но modular compare в пределах 2^29 самоисправляется следующим writeʼом.

### Slot count divergence

- Leader > follower → follower grow'ится на handshake.
- Leader < follower → warn + ignore, follower не truncate'ит.

### Promotion follower → leader

- Stop `FixedReplicationClient`.
- Start `FixedReplicationServer`.
- User writes: `bump_version(versions[slot_id])` продолжает с того version'а, который был last-seen. Нет специального promotion API (per-slot version'а достаточно).

---

## 10. Correctness properties

1. **Eventual consistency.** После того как все события доставлены (streaming или catch-up), follower состояние идентично leader'скому (modular compare обеспечивает idempotency).
2. **Idempotent apply.** Duplicate events (scan + streaming) безопасны: modular compare отфильтровывает.
3. **No causal consistency между слотами.** Per-slot linearizability есть, cross-slot ordering — только в streaming mode (SPSC FIFO); в catch-up порядок определяется slot_id.
4. **Torn write safety.** CRC на (meta+key+value) ловит torn write (partial pwrite slot'а). DELETED's 4-byte pwrite атомарен на aligned u32.
5. **Overflow safety.** Silent event drop на SPSC overflow безопасен: catch-up на reconnect'е lib-лечит. Hot slots (часто обновляемые) самоисправляются на следующем streaming event'е.
6. **Missed DELETE recovery.** Rare stale cleanup через pread при detection.

### Known limitations v1

- Wrap beyond 2^29 lag per hot slot = undefined behavior (modular compare ambiguity). В практике 10K writes/sec/slot = ~15 hours до ambiguity. Mitigated monitoring warn'ом.
- Один streaming follower на shard. Остальные — catch-up only on reconnect.
- No encryption replicated slots.
- No `SlotReset` admin RPC для экстренной resync.

---

## 11. Monitoring и metrics

### Leader

| Metric | Type | Labels |
|---|---|---|
| `armdb.fixed.spsc_overflow_count` | counter | `shard` |
| `armdb.fixed.version_near_wrap` | counter | `shard` |
| `armdb.fixed.followers_connected` | gauge | — |
| `armdb.fixed.catchup_slots_scanned` | counter | `shard` |
| `armdb.fixed.streaming_events_sent` | counter | `shard` |
| `armdb.fixed.min_replicated_version` | gauge | `shard` |

### Follower

| Metric | Type | Labels |
|---|---|---|
| `armdb.fixed.applied_events` | counter | `shard`, `kind` |
| `armdb.fixed.skipped_events` | counter | `shard` |
| `armdb.fixed.stale_cleanup_count` | counter | `shard` |
| `armdb.fixed.reconnect_count` | counter | `shard` |
| `armdb.fixed.applied_total` | counter | `shard` |
| `armdb.fixed.max_version_seen` | gauge | `shard` |

### Tracing

- Info: first-follower-connected, catch-up-complete, connected-to-leader.
- Warn: version-near-wrap, stale-index-cleanup, follower-has-more-slots, spsc-consumer-taken.
- Error: shard-stream-error, topology-mismatch.

---

## 12. File changes summary

| Path | Action |
|---|---|
| `armdb/src/fixed/slot.rs` | MODIFY — meta u32, helpers, CRC over (meta+key+value), DELETED read returns None |
| `armdb/src/fixed/shard.rs` | MODIFY — `versions: Vec<u32>`, 4-byte delete, apply_foreign_*, read_slot_header_and_key, replication_tx, versions sidecar |
| `armdb/src/fixed/engine.rs` | MODIFY — db.meta v2, fixed.data v2, install_replication_producers, impl FixedEngineAccess |
| `armdb/src/fixed/bitmap.rs` | MODIFY — from_versions helper |
| `armdb/src/fixed/mod.rs` | MODIFY — re-export FixedReplicationTarget |
| `armdb/src/fixed/config.rs` | no change |
| `armdb/src/durability.rs` | MODIFY — recover_entries populates versions + uses fixed.versions sidecar |
| `armdb/src/fixed_replication/mod.rs` | CREATE |
| `armdb/src/fixed_replication/apply.rs` | CREATE — trait + impl for ConstTree/ConstMap<Fixed> |
| `armdb/src/fixed_replication/protocol.rs` | CREATE |
| `armdb/src/fixed_replication/server.rs` | CREATE |
| `armdb/src/fixed_replication/client.rs` | CREATE |
| `armdb/src/fixed_replication/cursor.rs` | CREATE |
| `armdb/src/fixed_replication/event.rs` | CREATE |
| `armdb/src/fixed_replication/engine_access.rs` | CREATE |
| `armdb/src/lib.rs` | MODIFY — `pub mod fixed_replication;` |
| `armdb/src/error.rs` | no change (DbError::Replication, ShardCountMismatch существуют) |
| `armdb/Cargo.toml` | no change (feature `replication` уже `dep:rtrb`) |
| `armdb/src/skiplist/node.rs` | MAYBE MODIFY — `update_value` method if missing |
| `armdb/src/const_tree.rs` / `const_map.rs` | MAYBE MODIFY — hooks for replicated apply (`pub(crate)` helpers) |

---

## 13. Implementation order

1. Slot format + meta helpers (`fixed/slot.rs`) + unit tests.
2. `versions: Vec<u32>` в shard + `fixed.versions` sidecar + recovery population + tests.
3. Modular compare + boundary tests.
4. db.meta / fixed.data version bump + reopen-with-old-version test.
5. `FixedReplicationEvent` type + SPSC hook в write_slot/delete_slot + tests.
6. Protocol (encode/decode) + unit roundtrip tests.
7. `FixedReplicationTarget` trait + impl для ConstTree/ConstMap<Fixed> + apply-path unit tests with mock engine.
8. `FixedReplicationServer` — acceptor + lazy install + full scan + integration test (fresh sync).
9. `FixedReplicationClient` — reconnect + apply loop + integration tests (streaming, reconnect, chaos disconnects).
10. `FixedReplicationCursor` + diagnostic assertions.
11. End-to-end property-based + stress tests.

Каждый этап — отдельный коммит, прошедший `cargo check -p armdb` и `cargo nextest run -p armdb -E 'test(fixed)'`.

---

## 14. Testing strategy

### Unit tests
- Slot format: pack/unpack, bump, is_newer (+wrap boundary), CRC coverage, DELETED read returns None.
- Shard: versions population, delete partial write, apply_foreign_slot, wrap warning.
- Protocol: roundtrip для каждого message type (в т.ч. `FLAG_EMPTY_STATE`), truncated payload errors.
- Cursor: save/load roundtrip, absent/truncated file.
- Sidecar: save/load `fixed.versions` с trailer, corruption scenarios (truncate → fallback to dirty scan, CRC mismatch → fallback, slot_count mismatch → fallback).
- Dirty recovery torn slot: CRC mismatch → `versions[i] = pack_meta(FREE, version_of(meta))` preserving version.

### Integration tests
- `test_initial_sync_empty_follower` — fresh follower отправляет `FLAG_EMPTY_STATE`, leader не шлёт DELETED слоты, финальное состояние совпадает.
- `test_initial_sync_non_empty_follower_gets_deleted` — follower с existing state получает DELETED события в catch-up.
- `test_streaming_live_writes`
- `test_mixed_catchup_and_streaming` (duplicate events)
- `test_reconnect_after_disconnect`
- `test_overflow_during_catchup`
- `test_shard_count_mismatch_error`
- `test_kv_len_mismatch_error`
- `test_modular_compare_at_boundary` (forced wrap)
- `test_missed_delete_triggers_stale_cleanup` — OCCUPIED apply корректно очищает old_key когда DELETE был пропущен.
- `test_delete_does_not_remove_moved_key` (Bug 1 regression) — DELETED event для slot_A не удаляет ключ из индекса, если тот уже указывает на slot_B.
- `test_occupied_cleans_stale_when_new_key_maps_elsewhere` (Bug 2 regression) — OCCUPIED(slot_A, key=Y) при наличии `key_X → slot_A` и `key_Y → slot_C` правильно убирает `key_X`.
- `test_grow_during_streaming`
- `test_two_followers_one_streaming_one_catchup`
- `test_leader_restart`
- `test_follower_restart_dirty`
- `test_follower_restart_clean_uses_sidecar` — clean shutdown → reopen быстрее, sidecar trailer validated.
- `test_follower_restart_corrupted_sidecar_falls_back` — truncate `fixed.versions` → reopen делает full scan без данных потерь.
- `test_torn_slot_preserves_version` — симулируем torn write, recovery → версия сохранена, последующий write корректен.
- `test_promotion_follower_to_leader`

### Property-based (proptest)
- `prop_modular_compare_transitive` (within 2^29 window).
- `prop_slot_serialization_roundtrip` (random meta/key/value).
- `prop_replication_convergence` (random ops + disconnects → final equality).

### Stress
- `stress_1m_writes_with_follower` — throughput + convergence.
- `chaos_random_disconnects` — 100K+ writes с random (re)connects, final equality.
