logdb 0.6.0

//! # logdb — Embedded Append-Only Log Database
//!
//! logdb is an **embedded, append-only, crash-recoverable, optionally tamper-proof,
//! optionally remotely-pushable** local log database.
//!
//! ## Features
//!
//! - **High-throughput append**: lock-free fast path with CAS-based ring buffer
//! - **Crash recovery**: automatic torn-write detection and truncation on restart
//! - **Optional hash chain**: SHA-256 forward-linking for tamper detection (`hash-chain` feature)
//! - **Segment management**: automatic rolling, configurable retention
//! - **Segment pre-allocation**: next segment is pre-created at 80% capacity,
//!   reducing roll-time blocking to a single `fdatasync` call
//!
//! ## Performance: Inline vs Spill
//!
//! Records ≤ `INLINE_CAP` (256) bytes take the **inline
//! fast path**: zero heap allocation, zero extra memcpy. p50 is typically <100ns.
//!
//! Records > 256 bytes take the **spill path**: a heap allocation in the append
//! thread. The spill path is ~4x slower in throughput with ~80x higher p99.9
//! tail latency due to allocator jitter. Keep latency-sensitive records ≤ 256B.

// ── Public API surface ─────────────────────────────────────────────────────
//
// logdb exposes a narrow, intentional public API from the crate root (see the
// `pub use` re-exports below). Implementation modules are `pub(crate)` so they
// are NOT part of the supported public API / semver surface — internal
// refactors do not break downstream callers.
//
// The off-by-default `testing` feature re-exposes those modules (as
// `#[doc(hidden)] pub`) so the deployed test binary (`examples/testsuite.rs`)
// and the `tests/fuzz` integration target can exercise internals. It is not a
// supported public API.

/// Declare an implementation module.
///
/// `pub(crate)` normally; `#[doc(hidden)] pub` under the `testing` feature.
macro_rules! internal_mod {
    ($name:ident) => {
        // Crate-private in normal builds. The module's full API surface is
        // exercised under the `testing` feature (deployed test binary +
        // `tests/fuzz`), so silence dead_code for the whole subtree rather than
        // cfg-gating every test-only helper individually. Under `testing` the
        // module is `pub` (and doc-hidden) and its items are used, so no allow.
        #[cfg(not(feature = "testing"))]
        #[allow(dead_code)]
        pub(crate) mod $name;
        #[cfg(feature = "testing")]
        #[doc(hidden)]
        pub mod $name;
    };
}

// Observability shim — `tracing` when the feature is on, no-ops otherwise.
// Declared first with `#[macro_use]` so the `log_*!` macros are in scope for
// every module below.
#[macro_use]
mod observe;

internal_mod!(config);
internal_mod!(error);
internal_mod!(health);
internal_mod!(pipeline);
internal_mod!(platform);
internal_mod!(reader);
internal_mod!(record);
internal_mod!(ring);
internal_mod!(shard);
internal_mod!(storage);

mod pusher;
internal_mod!(recovery);
internal_mod!(tailer);

// The supported public surface, re-exported at the crate root. These are the
// only paths callers should depend on.
pub use config::{
    Config, DurabilityMode, IoBackend, QueueFullPolicy, RetentionPolicy, WaitStrategy,
};
pub use error::{
    AppendError, ConfigError, FlushError, OpenError, ReadError, ShutdownError, ShutdownReport,
    TailerError,
};
pub use reader::ScanIter;
pub use record::{Record, RecordId};
pub use shard::{decode_record_id, encode_record_id, shard_bits};
pub use tailer::Tailer;

/// Recovery report returned by [`LogDb::recovery_report`].
#[derive(Debug, Clone)]
pub struct RecoveryReport {
    /// First sequence to replay (the last checkpoint).
    pub from_sequence: u64,
    /// Last durable sequence.
    pub to_sequence: u64,
    /// Number of records to replay.
    pub count: u64,
}

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use zeroize::Zeroizing;

use health::HealthState;

/// Shared handle to the encryption key. The inner `[u8; 32]` is wrapped in
/// [`Zeroizing`] so the memory is scrubbed when the last holder drops the
/// [`Arc`]. All internal components (SegmentManager, Reader, Tailer, …) hold a
/// clone of this handle — no [`Copy`] duplicates left in memory.
pub(crate) type KeyHandle = Arc<zeroize::Zeroizing<[u8; 32]>>;

/// A resolved, in-memory set of AES-256-GCM keys supporting rotation **without
/// a disk-format change** (cr-032).
///
/// `active` encrypts new writes (and seeds the hash-chain MAC). On read, every
/// key in `decrypt_keys` — which always begins with `active`, followed by prior
/// keys still in the read window — is tried until one authenticates: AES-GCM is
/// an AEAD, so a wrong key deterministically fails the auth tag and the first
/// success is unambiguously correct. Rotation is therefore "add a new `active`
/// while keeping the old key in `decrypt_keys`"; retirement is "drop a key from
/// `decrypt_keys`" (its data becomes unreadable).
///
/// Each key carries a 128-bit `id` (cr-032 Phase 3). Written into the segment
/// header, it lets recovery pick the decrypt key in O(1) and lets an operator
/// see which key a segment depends on (explicit retirement, backup manifests).
/// A header `key_id == 0` means "absent" (no id recorded) — recovery falls back
/// to try-in-order. The id is opaque to the core; the provider assigns it
/// (e.g. a hash of the config `key_id` string).
///
/// The library **never** knows where keys come from (KMS, file, env). It holds
/// only this resolved ring; provider/vendor logic lives entirely in the server,
/// so the core carries no vendor dependency (cr-032 design rule).
pub struct KeyRing {
    /// Encrypts new writes + derives the hash-chain key.
    active: KeyHandle,
    /// The active key's 128-bit id (written to segment headers).
    active_id: u128,
    /// All keys valid for decryption. `[0] == (active_id, active)`; the rest are
    /// prior keys still in the read window. Tried in order, so the common case
    /// (current key) decrypts in one AEAD attempt.
    pub(crate) decrypt_keys: Vec<(u128, KeyHandle)>,
}

impl KeyRing {
    /// Single-key ring (no rotation) — the common case. The key gets a stable
    /// id derived from its bytes.
    pub fn single(key: [u8; 32]) -> Arc<Self> {
        let id = key_id_from_bytes(&key);
        let h: KeyHandle = Arc::new(Zeroizing::new(key));
        Arc::new(Self {
            active: h.clone(),
            active_id: id,
            decrypt_keys: vec![(id, h)],
        })
    }

    /// Rotation-capable ring: `active` encrypts new writes; each key in `prior`
    /// (older keys that must remain readable) is appended after `active`. The
    /// decrypt ring is therefore `[active, ...prior]`. Ids are derived from the
    /// key bytes (stable across reopens with the same keys).
    pub fn new(active: [u8; 32], prior: Vec<[u8; 32]>) -> Arc<Self> {
        let active_id = key_id_from_bytes(&active);
        let active_h: KeyHandle = Arc::new(Zeroizing::new(active));
        let mut decrypt_keys = vec![(active_id, active_h.clone())];
        for k in prior {
            let id = key_id_from_bytes(&k);
            decrypt_keys.push((id, Arc::new(Zeroizing::new(k))));
        }
        Arc::new(Self {
            active: active_h,
            active_id,
            decrypt_keys,
        })
    }

    /// Rotation-capable ring with explicit ids (one per key). Used by providers
    /// that source ids from config (e.g. a hash of the `key_id` string). `active`
    /// carries `active_id`; each `(id, key)` in `prior` follows. `prior` ids must
    /// not collide with `active_id` or each other.
    pub fn with_ids(active_id: u128, active: [u8; 32], prior: Vec<(u128, [u8; 32])>) -> Arc<Self> {
        let active_h: KeyHandle = Arc::new(Zeroizing::new(active));
        let mut decrypt_keys = vec![(active_id, active_h.clone())];
        for (id, k) in prior {
            decrypt_keys.push((id, Arc::new(Zeroizing::new(k))));
        }
        Arc::new(Self {
            active: active_h,
            active_id,
            decrypt_keys,
        })
    }

    /// The active key slice — used for encryption and hash-chain derivation.
    /// (Deref coercion — Arc → Zeroizing → [u8; 32] — yields the slice from the
    /// handle; no manual dereferencing needed.)
    pub(crate) fn active_slice(&self) -> &[u8; 32] {
        &self.active
    }

    /// The active key's 128-bit id (written to segment headers).
    pub(crate) fn active_id(&self) -> u128 {
        self.active_id
    }

    /// Look up a key by its 128-bit id (O(1)). `None` if the id is not in the
    /// ring (e.g. the key has been retired). Currently used by hash-chain
    /// recovery to unmask the chain key from a segment's `key_id` hint; available
    /// for future O(1) read paths.
    #[cfg_attr(not(feature = "hash-chain"), allow(dead_code))]
    pub(crate) fn key_for_id(&self, id: u128) -> Option<&KeyHandle> {
        self.decrypt_keys
            .iter()
            .find(|(kid, _)| *kid == id)
            .map(|(_, k)| k)
    }
}

/// A stable, 128-bit id for a key, derived from its bytes via FNV-1a (zero-dep,
/// always available — unlike BLAKE3, which is `hash-chain`-gated). Used when no
/// config id is supplied; providers may instead hash the `key_id` string. Only
/// needs to be collision-resistant among the (small) set of configured keys.
fn key_id_from_bytes(b: &[u8]) -> u128 {
    // FNV-1a 128-bit offset basis and prime.
    let mut h: u128 = 0x6c62_272e_07bb_0142_62b8_2175_6295_c58d;
    for &byte in b {
        h ^= byte as u128;
        h = h.wrapping_mul(0x0000_0000_0100_0000_0000_0000_0000_013b);
    }
    // Guarantee non-zero: a zero id is the "absent" sentinel in segment headers.
    if h == 0 {
        1
    } else {
        h
    }
}

impl std::fmt::Debug for KeyRing {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        // Never emit key material.
        f.debug_struct("KeyRing")
            .field("active", &"<redacted>")
            .field("decrypt_keys", &self.decrypt_keys.len())
            .finish()
    }
}

use pipeline::signal::{FlushSignal, ShutdownState};
use pipeline::trigger::CommitTrigger;
use ring::Ring;
use shard::ShardMap;
use storage::SegmentManager;

/// The main log database handle.
pub struct LogDb {
    inner: Arc<LogDbInner>,
}

struct LogDbInner {
    config: config::Config,
    encryption_keys: Option<Arc<KeyRing>>,
    shards: ShardMap,
    health: Arc<HealthState>,
    flush: Arc<FlushSignal>,
    shutdown: Arc<ShutdownState>,
    committer_handle: Option<std::thread::JoinHandle<()>>,
    #[cfg(feature = "hash-chain")]
    sealer_handles: Vec<std::thread::JoinHandle<()>>,
    data_dir: std::path::PathBuf,
    /// WAL checkpoint: records with sequence < this are safe to truncate.
    checkpoint_sequence: Arc<AtomicU64>,
    /// Per-shard cached segment listings (one per shard dir).
    manifests: Vec<Arc<Mutex<reader::SegmentManifest>>>,
}

impl LogDb {
    /// Open or create a logdb instance.
    ///
    /// Validates `config`, recovers each shard's on-disk state (or creates a
    /// fresh log), and spawns the background Committer (and Sealer, under
    /// `hash-chain`). Returns a structured [`OpenError`] on failure so callers
    /// can match on the category (invalid config, recovery, segment, thread).
    pub fn open(mut config: Config) -> Result<Self, OpenError> {
        config.validate()?;

        let data_dir = config.data_dir.clone();
        let hash_enabled = config.hash_enabled;

        // Take the resolved key ring out of the config. From here on every
        // component that needs the keys gets an Arc clone of the ring — no raw
        // [u8; 32] copies left in memory (each key is wrapped in Zeroizing, so
        // the last holder's drop scrubs the bytes).
        let encryption_keys: Option<Arc<KeyRing>> = config.encryption_keys.take();

        let num_shards = config.shards;
        let shard_bits = shard::shard_bits(num_shards);

        // Build per-shard SegmentManagers, recovering each shard dir independently.
        // Each shard is an independent recoverable log; within a shard, record ids
        // are monotonic with stride `1 << shard_bits`. shards=1: flat data_dir,
        // shard_bits=0 (identity encoding, stride 1). shards>1: data_dir/s<shard>/.
        let mut seg_mgrs: Vec<SegmentManager> = Vec::with_capacity(num_shards);
        let mut initial_seqs: Vec<u64> = Vec::with_capacity(num_shards);
        // Per-shard hash chain state (one entry per shard).
        let mut shard_hash_inits: Vec<[u8; 32]> = Vec::with_capacity(num_shards);
        let mut shard_last_hashes: Vec<[u8; 32]> = Vec::with_capacity(num_shards);

        for s in 0..num_shards {
            let sdir = if num_shards == 1 {
                data_dir.clone()
            } else {
                data_dir.join(format!("s{}", s))
            };
            let has_data = sdir.exists() && sdir.join("segment-00000001.log").exists();

            if has_data {
                let st = recovery::recover_shard(
                    &sdir,
                    shard_bits,
                    config.segment_size,
                    config.retention.clone(),
                    encryption_keys.clone(),
                )
                .map_err(|reason| OpenError::Recovery { shard: s, reason })?;
                // Resume this shard's ring at the LOCAL seq after the last recovered
                // record. An empty shard (recovered_count == 0) resumes at 0.
                let initial_local = if st.recovered_count == 0 {
                    0
                } else {
                    shard::decode_record_id(st.last_sequence, shard_bits).1 + 1
                };
                shard_hash_inits.push(st.hash_init);
                shard_last_hashes.push(st.last_hash);
                seg_mgrs.push(st.segment_manager);
                initial_seqs.push(initial_local);
            } else {
                // Fresh shard. `hi` is the stable per-shard hash-chain key
                // (cr-032 Phase 3): a random secret, independent of which
                // encryption key is active, so rotating the key no longer severs
                // the chain. It is stored MASKED on disk (chain_key ⊕
                // derive(active)) so it stays off-disk-plaintext — secret, as
                // before — and recovery unmasks it. Without encryption it is the
                // plaintext header seed (the released unencrypted hash-chain
                // behavior, unchanged).
                let hi = generate_hash_init();

                shard_hash_inits.push(hi);
                shard_last_hashes.push([0u8; 32]);

                #[cfg(feature = "hash-chain")]
                let header_hash_init = match &encryption_keys {
                    Some(kr) if hash_enabled => xor32(hi, derive_hash_init(kr.active_slice())),
                    _ => hi,
                };
                #[cfg(not(feature = "hash-chain"))]
                let header_hash_init = hi;
                let sm = SegmentManager::create(
                    sdir,
                    config.segment_size,
                    hash_enabled,
                    config.compression_enabled,
                    encryption_keys.clone(),
                    header_hash_init,
                    config.retention.clone(),
                    0,
                )
                .map_err(OpenError::SegmentCreate)?;
                seg_mgrs.push(sm);
                initial_seqs.push(0);
            }
        }

        // Apply the configured sparse-index stride before the Committer starts
        // appending (active index is still empty here). Also tag each manager
        // with its shard id/bits so segment base_sequence headers are written as
        // global record ids (required for correct point reads / truncation /
        // recovery under shards>1).
        for (s, m) in seg_mgrs.iter_mut().enumerate() {
            m.set_index_stride(config.index_stride);
            m.set_shard(s, shard_bits);
        }

        // Create shared state
        let shards = ShardMap::new_with_initial(
            config.shards,
            config.ring_size,
            hash_enabled,
            &initial_seqs,
        );
        let health = Arc::new(HealthState::new());
        let flush = Arc::new(FlushSignal::new(num_shards));
        let shutdown = Arc::new(ShutdownState::new());

        let trigger = CommitTrigger {
            bytes: 256 * 1024,
            records: 1024,
            interval: Duration::from_millis(10),
            durability: config.durability_mode,
        };
        let wait = config.wait_strategy;

        // Spawn Committer — passes all rings + per-shard managers for multi-shard polling
        let committer_rings = shards.all_rings().to_vec();
        let committer_flush = Arc::clone(&flush);
        let committer_shutdown = Arc::clone(&shutdown);
        let committer_health = Arc::clone(&health);
        let checkpoint = Arc::new(AtomicU64::new(Self::load_checkpoint(&data_dir)));
        let committer_checkpoint = Arc::clone(&checkpoint);
        let committer_handle = std::thread::Builder::new()
            .name("logdb-committer".into())
            .spawn(move || {
                pipeline::committer::run_committer(
                    committer_rings,
                    seg_mgrs,
                    shard_bits,
                    trigger,
                    committer_flush,
                    committer_shutdown,
                    committer_health,
                    committer_checkpoint,
                    wait,
                );
            })
            .map_err(OpenError::ThreadSpawn)?;

        // Spawn one Sealer per shard (only when hash-chain is enabled).
        // Each shard has its own independent hash chain with its own
        // hash_init and last_hash, recovered from on-disk state or generated
        // fresh above. Multi-shard hashing is fully supported.
        #[cfg(feature = "hash-chain")]
        let mut sealer_handles: Vec<std::thread::JoinHandle<()>> = Vec::new();
        #[cfg(feature = "hash-chain")]
        if hash_enabled {
            for s in 0..num_shards {
                let sealer_ring = Arc::clone(shards.ring(s));
                let sealer_shutdown = Arc::clone(&shutdown);
                let hi = shard_hash_inits[s];
                let lh = shard_last_hashes[s];
                let iseq = initial_seqs[s];
                let name = format!("logdb-sealer-{}", s);
                sealer_handles.push(
                    std::thread::Builder::new()
                        .name(name)
                        .spawn(move || {
                            pipeline::sealer::run_sealer(
                                sealer_ring,
                                hi,
                                lh,
                                iseq,
                                sealer_shutdown,
                                wait,
                            );
                        })
                        .map_err(OpenError::ThreadSpawn)?,
                );
            }
        }

        let manifests: Vec<Arc<Mutex<reader::SegmentManifest>>> = (0..num_shards)
            .map(|s| {
                let dir = if num_shards == 1 {
                    data_dir.clone()
                } else {
                    data_dir.join(format!("s{}", s))
                };
                Arc::new(Mutex::new(reader::SegmentManifest::new(dir)))
            })
            .collect();

        Ok(Self {
            inner: Arc::new(LogDbInner {
                config,
                encryption_keys,
                shards,
                health,
                flush,
                shutdown,
                committer_handle: Some(committer_handle),
                #[cfg(feature = "hash-chain")]
                sealer_handles,
                data_dir,
                checkpoint_sequence: checkpoint,
                manifests,
            }),
        })
    }

    /// Append multiple records atomically. All records in the batch are
    /// committed together — either all visible after crash, or none.
    /// Returns the sequence number of the first record in the batch.
    ///
    /// All `contents.len()` sequences are reserved in one atomic
    /// `claim_batch` (no partial reservation),
    /// so consecutive batches never overwrite each other's slots.
    pub fn append_batch(&self, contents: &[&[u8]]) -> Result<u64, AppendError> {
        if contents.is_empty() {
            return Err(AppendError::EmptyBatch);
        }
        let inner = &self.inner;
        if let Some(code) = inner.health.check() {
            return Err(match code {
                health::HEALTH_DISK_FULL => AppendError::DiskFull,
                _ => AppendError::Io("unhealthy".into()),
            });
        }
        // Validate ALL contents BEFORE reserving sequences. A too-large record
        // found after a partial claim_batch would leave reserved-but-unwritten
        // slots (a gap the Committer can't cross).
        for content in contents {
            if content.len() > inner.config.max_content_size {
                return Err(AppendError::ContentTooLarge {
                    size: content.len(),
                    max: inner.config.max_content_size,
                });
            }
        }
        if !inner.shutdown.enter() {
            return Err(AppendError::ShuttingDown);
        }
        let _guard = scopeguard::guard((), |_| inner.shutdown.leave());

        // Reserve the whole batch atomically (producer_cursor += n).
        let n = contents.len() as u64;
        let (first_id, shard_id, local_first) = inner
            .shards
            .claim_batch(n, inner.config.queue_full_policy)?;
        let ts = platform::clock_realtime_coarse_ns();
        let ring = inner.shards.ring(shard_id);
        let shard_bits = inner.shards.shard_bits();
        for (i, content) in contents.iter().enumerate() {
            let local_seq = local_first + i as u64;
            let global_id = shard::encode_record_id(shard_id, local_seq, shard_bits);
            // Safety: claim_batch reserved the LOCAL range [local_first, local_first+n)
            // exclusively. Slot is indexed by LOCAL seq; record_id stores the GLOBAL id.
            unsafe {
                ring.slot(local_seq).producer_write(global_id, ts, content);
            }
            ring.slot(local_seq).publish(local_seq);
        }
        metric_counter!("logdb.appends", n);
        Ok(first_id)
    }

    /// Append a record to the log, routed by thread affinity.
    /// Returns the global record_id.
    pub fn append(&self, content: &[u8]) -> Result<u64, AppendError> {
        let shard_id = self.inner.shards.select_shard();
        self.append_routed(content, shard_id)
    }

    /// Append a record to the log, routed by a caller-supplied key.
    ///
    /// Same `shard_key` ⇒ same shard (deterministic CRC32C routing via
    /// [`ShardMap::select_shard_by_key`]), so all records for one entity
    /// (session/user id) land on one shard and stay ordered. This is the
    /// partitioning model the logdb-broker (cr-037) builds consumer-group work
    /// distribution on. Returns the global record_id.
    pub fn append_with_key(&self, content: &[u8], shard_key: &[u8]) -> Result<u64, AppendError> {
        let shard_id = self.inner.shards.select_shard_by_key(shard_key);
        self.append_routed(content, shard_id)
    }

    /// Shared append path: validate, claim a slot on `shard_id`, write, publish.
    ///
    /// `shard_id` must be `< num_shards()` (enforced by
    /// [`ShardMap::claim_on_shard`]). The two public entry points
    /// ([`append`](Self::append) via thread affinity, [`append_with_key`](Self::append_with_key)
    /// via key) only differ in how `shard_id` is chosen.
    fn append_routed(&self, content: &[u8], shard_id: usize) -> Result<u64, AppendError> {
        let inner = &self.inner;

        // Health check (self-healing)
        if let Some(code) = inner.health.check() {
            return Err(match code {
                health::HEALTH_DISK_FULL => AppendError::DiskFull,
                _ => AppendError::Io("health check failed".into()),
            });
        }

        // Content size check
        if content.len() > inner.config.max_content_size {
            return Err(AppendError::ContentTooLarge {
                size: content.len(),
                max: inner.config.max_content_size,
            });
        }

        // Shutdown guard
        if !inner.shutdown.enter() {
            return Err(AppendError::ShuttingDown);
        }
        let _guard = scopeguard::guard((), |_| inner.shutdown.leave());

        // CAS claim on the selected shard.
        let (global_id, _, local_seq) = inner
            .shards
            .claim_on_shard(shard_id, inner.config.queue_full_policy)?;

        // Write slot (safety: claim guarantees exclusive access). Slot is indexed
        // by LOCAL seq; record_id stores the GLOBAL id so read-back by global id
        // works under sharding (shards=1: global == local, behavior unchanged).
        let ts = platform::clock_realtime_coarse_ns();
        let ring = inner.shards.ring(shard_id);
        unsafe {
            ring.slot(local_seq).producer_write(global_id, ts, content);
        }

        // Publish
        ring.slot(local_seq).publish(local_seq);

        metric_counter!("logdb.appends", 1);
        Ok(global_id)
    }

    /// Replicate a record at an EXACT sequence number.
    ///
    /// Used by logdbd standby nodes to write records received from the primary
    /// at the primary's own sequence, preserving the global offset space so
    /// that consumers can fail over primary → standby without re-mapping
    /// offsets. Unlike [`append`](LogDb::append), this does NOT claim a fresh
    /// sequence; it writes directly to the slot for `sequence`.
    ///
    /// Constraints (standby contract):
    /// - **Single-shard only.** Replication maps the primary's linear sequence
    ///   1:1 onto shard 0, so `shards` must be 1.
    /// - **In-order.** `sequence` must equal the current producer cursor (the
    ///   next expected sequence). Gaps return an error so the caller retries.
    /// - **Idempotent.** A `sequence` already replicated (below the cursor) is
    ///   a no-op, so duplicate/replayed Sync RPCs are safe.
    /// - **Backpressured.** Refuses to overwrite a live (uncommitted) slot via
    ///   the same consume-watermark gate as `claim`, returning `QueueFull`.
    ///
    /// The record is published for the Committer to serialize and fsync like
    /// any other; `producer_cursor` is advanced so `flush`/`shutdown` compute
    /// the correct durability target.
    pub fn replicate(
        &self,
        sequence: u64,
        timestamp_ns: u64,
        content: &[u8],
    ) -> Result<(), AppendError> {
        let inner = &self.inner;

        // Replication is a linear stream onto shard 0.
        if inner.shards.num_shards() != 1 {
            return Err(AppendError::Io("replicate requires shards=1".into()));
        }
        if content.len() > inner.config.max_content_size {
            return Err(AppendError::ContentTooLarge {
                size: content.len(),
                max: inner.config.max_content_size,
            });
        }
        if let Some(code) = inner.health.check() {
            return Err(match code {
                health::HEALTH_DISK_FULL => AppendError::DiskFull,
                _ => AppendError::Io("health check failed".into()),
            });
        }
        if !inner.shutdown.enter() {
            return Err(AppendError::ShuttingDown);
        }
        let _guard = scopeguard::guard((), |_| inner.shutdown.leave());

        let ring = inner.shards.ring(0);
        let ring_size = ring.ring_size() as u64;

        // Idempotency: already replicated past this sequence.
        let cur = ring.producer_cursor.inner.load(Ordering::Acquire);
        if sequence < cur {
            return Ok(());
        }
        // In-order: sequence must be exactly the next expected slot.
        if sequence != cur {
            return Err(AppendError::Io(format!(
                "replicate out of order: expected {}, got {}",
                cur, sequence
            )));
        }
        // Backpressure: do not overwrite a slot the Committer has not drained.
        // Same invariant as Ring::claim (seq - watermark < ring_size).
        let wm = ring.consume_watermark();
        if sequence.wrapping_sub(wm) >= ring_size {
            return Err(AppendError::QueueFull);
        }

        // Safety: the standby serializes Sync RPCs externally (and local writes
        // are rejected on a standby), so this slot is not being written by any
        // other producer. `sequence` was validated to equal the monotonic
        // cursor, so the slot has not yet been consumed (it is below committed
        // only by the watermark gate above). Mirrors append()'s proof.
        unsafe {
            ring.slot(sequence)
                .producer_write(sequence, timestamp_ns, content);
        }
        ring.slot(sequence).publish(sequence);

        // Advance the cursor so flush/shutdown target these records for fsync.
        // Single-writer on the standby (no local appends); a plain Release
        // store is correct and races are prevented by the caller's lock.
        ring.producer_cursor
            .inner
            .store(sequence + 1, Ordering::Release);

        Ok(())
    }

    /// Force all previously appended records to durable storage.
    ///
    /// Waits for `durable_cursor` (NOT committed_cursor — fix C4).
    pub fn flush(&self) -> Result<(), FlushError> {
        let _t0 = Instant::now();
        let inner = &self.inner;

        // Per-shard snapshot: flush completes when EVERY shard's durable reaches
        // its own producer-cursor snapshot (handles uneven sharded loads).
        let targets = inner.shards.producer_cursors();
        if targets.iter().all(|&t| t == 0) {
            metric_histogram!("logdb.flush.duration", _t0.elapsed());
            return Ok(());
        }

        // If hash enabled (shards=1 only), wait for Sealer first.
        #[cfg(feature = "hash-chain")]
        if inner.config.hash_enabled {
            let target0 = targets[0];
            wait_until(
                &inner.shutdown,
                || inner.shards.ring(0).sealed_cursor.load(Ordering::Acquire) >= target0,
                inner.config.flush_timeout,
            )?;
        }

        inner.flush.request(&targets);
        wait_until(
            &inner.shutdown,
            || inner.flush.is_done(&targets),
            inner.config.flush_timeout,
        )?;

        metric_histogram!("logdb.flush.duration", _t0.elapsed());
        Ok(())
    }

    /// Read a single record by `record_id` (a global id).
    ///
    /// Decodes the global id to its owning shard, gates on that shard's
    /// durable cursor (per-shard visibility), and reads from that shard's
    /// manifest. shards=1: shard 0, local == record_id (zero-regression).
    pub fn read(&self, record_id: u64) -> Result<Option<Record>, ReadError> {
        let inner = &self.inner;
        let (shard, local) = shard::decode_record_id(record_id, inner.shards.shard_bits());
        let durable_s = inner
            .shards
            .durable_cursors()
            .get(shard)
            .copied()
            .unwrap_or(0);
        if local >= durable_s {
            return Ok(None);
        }
        let reader = reader::Reader::new(
            Arc::clone(&inner.manifests[shard]),
            inner.encryption_keys.clone(),
        );
        reader.read(record_id)
    }

    /// Force-refresh every shard's segment manifest — re-scan the data
    /// directory, ignoring the mtime cache.
    ///
    /// Read paths self-heal when a *cached* segment's file goes missing
    /// (truncation/retention), but a **brand-new** segment whose directory
    /// mtime hasn't ticked can stay invisible until mtime propagates. This
    /// removes that lag deterministically. Use on coarse-mtime filesystems
    /// (WSL2, some network FS) or after out-of-band changes (backup restore,
    /// manual segment deletion).
    pub fn refresh_manifests(&self) -> Result<(), ReadError> {
        for manifest in &self.inner.manifests {
            manifest.lock().unwrap().force_refresh()?;
        }
        Ok(())
    }

    /// Read many records by id in one call — a multi-get. Faster than N
    /// individual [`read`](LogDb::read)s when several ids share a segment:
    /// `read_batch` opens each segment file and loads its sparse index **once
    /// per segment**, not once per record.
    ///
    /// Result order matches `ids`. Ids that don't exist, or whose records are
    /// not yet durable, yield `None` at their position.
    pub fn read_batch(&self, ids: &[u64]) -> Result<Vec<Option<Record>>, ReadError> {
        let mut results: Vec<Option<Record>> = vec![None; ids.len()];
        if ids.is_empty() {
            return Ok(results);
        }
        let inner = &self.inner;
        let bits = inner.shards.shard_bits();
        let durable = inner.shards.durable_cursors();

        // Group result slots by shard (each shard has its own manifest/reader).
        let mut by_shard: std::collections::HashMap<usize, Vec<usize>> =
            std::collections::HashMap::new();
        for (i, &id) in ids.iter().enumerate() {
            let (shard, _local) = shard::decode_record_id(id, bits);
            by_shard.entry(shard).or_default().push(i);
        }
        for (shard, slots) in by_shard {
            let durable_s = durable.get(shard).copied().unwrap_or(0);
            // Keep only ids that are durable in this shard; the rest stay None.
            let live: Vec<(usize, u64)> = slots
                .iter()
                .map(|&slot| (slot, ids[slot]))
                .filter(|&(_slot, id)| shard::decode_record_id(id, bits).1 < durable_s)
                .collect();
            if live.is_empty() {
                continue;
            }
            let reader = reader::Reader::new(
                Arc::clone(&inner.manifests[shard]),
                inner.encryption_keys.clone(),
            );
            let shard_ids: Vec<u64> = live.iter().map(|&(_, id)| id).collect();
            let mut batch = reader.read_batch(&shard_ids)?;
            for (k, (slot, _id)) in live.iter().enumerate() {
                results[*slot] = batch[k].take(); // move (no Record clone)
            }
        }
        Ok(results)
    }

    // ── Internal state accessors (for diagnostics/benchmarking) ─────────

    /// Get the maximum producer cursor across all shards.
    pub fn producer_cursor(&self) -> u64 {
        self.inner.shards.max_producer_cursor()
    }

    /// Get the minimum committed cursor across all shards.
    pub fn committed_cursor(&self) -> u64 {
        self.inner.shards.min_committed_cursor()
    }

    /// Get the minimum durable cursor across all shards.
    pub fn durable_cursor(&self) -> u64 {
        self.inner.shards.min_durable_cursor()
    }

    /// Current health state: `None` if healthy, `Some(code)` if degraded
    /// (e.g. `health::HEALTH_DISK_FULL` on ENOSPC). The state self-heals —
    /// `clear_if_recovered` is checked on each append, so a transient disk-full
    /// clears once the filesystem has space again. Hosts (e.g. logdbd) poll
    /// this to drive a readiness/liveness probe.
    pub fn health_code(&self) -> Option<u8> {
        self.inner.health.check()
    }

    /// Sample the current operational gauges into the `metrics` facade
    /// (no-op without the `metrics` feature). A host that exports metrics
    /// (e.g. a Prometheus scraper) should call this on its scrape interval.
    ///
    /// Emits:
    /// - `logdb.durable_lag`  — `producer_cursor − durable_cursor` (records
    ///   appended but not yet fsynced; the durability backlog).
    /// - `logdb.queue_depth`  — `producer_cursor − committed_cursor` (records
    ///   claimed but not yet serialized by the Committer).
    /// - `logdb.wal_bytes`    — total size of segment files (`wal_usage().0`).
    pub fn record_gauges(&self) {
        let _producer = self.producer_cursor();
        let _durable = self.durable_cursor();
        let _committed = self.committed_cursor();
        metric_gauge!("logdb.durable_lag", _producer.saturating_sub(_durable));
        metric_gauge!("logdb.queue_depth", _producer.saturating_sub(_committed));
        metric_gauge!("logdb.wal_bytes", self.wal_usage().0);
    }

    /// Get the total ring capacity across all shards.
    pub fn ring_size(&self) -> usize {
        self.inner.shards.num_shards() * self.inner.shards.ring(0).ring_size()
    }

    /// Create a named tailer (consumer) with independent read progress.
    ///
    /// Progress is tracked per shard and persisted to `tailer_<name>.dat` via
    /// `commit()`. See [`Tailer`] for the sharding semantics
    /// (per-shard progress, merged-batch delivery, best-effort cross-batch
    /// ordering when a shard stalls).
    pub fn new_tailer(&self, name: &str) -> Tailer {
        let rings: Vec<Arc<Ring>> = self
            .inner
            .shards
            .all_rings()
            .iter()
            .map(Arc::clone)
            .collect();
        let manifests: Vec<Arc<std::sync::Mutex<reader::SegmentManifest>>> =
            self.inner.manifests.iter().map(Arc::clone).collect();
        crate::tailer::Tailer::open(
            manifests,
            rings,
            self.inner.shards.shard_bits(),
            name,
            self.inner.encryption_keys.clone(),
            self.inner.data_dir.clone(),
        )
    }

    /// Scan records in range `[from_id, to_id)` across ALL shards, ordered by
    /// ascending global id. `shards=1` returns a single cross-segment stream;
    /// `shards>1` k-way-merges the per-shard streams. An empty range yields an
    /// empty iterator (no error).
    pub fn scan(&self, from_id: u64, to_id: u64) -> Result<ScanIter, ReadError> {
        let manifests = self.inner.manifests.iter().map(Arc::clone).collect();
        reader::ScanIter::build(manifests, self.inner.encryption_keys.clone(), from_id, to_id)
    }

    /// Scan a single shard in `[from_id, to_id)`. Used for incremental
    /// per-shard recovery (e.g. seq-map checkpoint replay).
    pub fn scan_shard(
        &self,
        shard_id: usize,
        from_id: u64,
        to_id: u64,
    ) -> Result<ScanIter, ReadError> {
        let manifest = Arc::clone(&self.inner.manifests[shard_id]);
        reader::ScanIter::build_single(
            manifest,
            self.inner.encryption_keys.clone(),
            from_id,
            to_id,
        )
    }

    /// Mark `sequence` as the WAL checkpoint.
    ///
    /// Records with sequence < checkpoint are safe to delete. Old segments
    /// fully covered by the checkpoint will be truncated on the next roll.
    pub fn checkpoint(&self, sequence: u64) {
        let mut cur = self.inner.checkpoint_sequence.load(Ordering::Acquire);
        while sequence > cur {
            match self.inner.checkpoint_sequence.compare_exchange_weak(
                cur,
                sequence,
                Ordering::Release,
                Ordering::Acquire,
            ) {
                Ok(_) => break,
                Err(v) => cur = v,
            }
        }
        // Persist to disk so crash recovery can read it
        let _ = save_checkpoint(&self.inner.data_dir, sequence);
    }

    /// Get the current checkpoint sequence.
    pub fn checkpoint_sequence(&self) -> u64 {
        self.inner.checkpoint_sequence.load(Ordering::Acquire)
    }

    /// Recover the checkpoint from disk (called during startup).
    pub(crate) fn load_checkpoint(data_dir: &std::path::Path) -> u64 {
        let path = data_dir.join("checkpoint.dat");
        match std::fs::read(&path) {
            Ok(data) if data.len() == 12 => {
                let seq = u64::from_le_bytes([
                    data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7],
                ]);
                let crc = u32::from_le_bytes([data[8], data[9], data[10], data[11]]);
                if crc32c::crc32c(&data[..8]) == crc {
                    seq
                } else {
                    0
                }
            }
            _ => 0,
        }
    }

    /// Get WAL space usage: (used_bytes, total_bytes_configured).
    ///
    /// `used_bytes` is the sum of all segment file sizes — flat in `data_dir`
    /// for `shards == 1`, and across every `s<shard>/` subdir for `shards > 1`.
    pub fn wal_usage(&self) -> (u64, u64) {
        let mut total = count_log_bytes(&self.inner.data_dir);
        // shards>1 lays segments under data_dir/s<shard>/ — sum those too.
        if let Ok(entries) = std::fs::read_dir(&self.inner.data_dir) {
            for e in entries.flatten() {
                let path = e.path();
                if path.is_dir() {
                    total += count_log_bytes(&path);
                }
            }
        }
        (total, self.inner.config.segment_size)
    }

    /// Recovery report for database WAL replay.
    ///
    /// `from_sequence` is the WAL checkpoint (global id). `count` is the number
    /// of durable records with global id `>= from_sequence`, summed across ALL
    /// shards. `to_sequence` is the global durable watermark (the smallest
    /// not-yet-durable global id — an informational lower bound). For
    /// `shards == 1` this reduces exactly to the legacy `durable - checkpoint`.
    pub fn recovery_report(&self) -> RecoveryReport {
        let cp = self.checkpoint_sequence();
        let shards = &self.inner.shards;
        let bits = shards.shard_bits();
        let stride: u64 = 1u64 << bits;
        let mut count: u64 = 0;
        let mut watermark = u64::MAX;
        for s in 0..shards.num_shards() {
            let durable_s = shards.ring(s).durable_cursor.load(Ordering::Acquire);
            // First local seq in shard s whose global id (local*stride + s) >= cp.
            let first_local: u64 = if cp <= s as u64 {
                0
            } else {
                (cp - s as u64 + stride - 1) / stride // ceil((cp - s) / stride)
            };
            count += durable_s.saturating_sub(first_local);
            let wm = (durable_s << bits) | s as u64; // first not-yet-durable gid of shard s
            if wm < watermark {
                watermark = wm;
            }
        }
        RecoveryReport {
            from_sequence: cp,
            to_sequence: if watermark == u64::MAX { 0 } else { watermark },
            count,
        }
    }

    /// Replay records from `sequence` (inclusive) to the end of the log, across
    /// all shards, ordered by ascending global id.
    pub fn replay_from(&self, sequence: u64) -> Result<ScanIter, ReadError> {
        self.scan(sequence, u64::MAX)
    }

    /// Drain in-flight appends and flush all published records to durable
    /// storage — WITHOUT consuming the handle or joining background threads.
    ///
    /// This is the shared-safe drain path: unlike [`shutdown`](LogDb::shutdown)
    /// it takes `&self`, so it works when the `LogDb` is shared via `Arc`
    /// (e.g. inside a long-running service like logdbd). It enters the drain
    /// phase (rejecting new appends with `ShuttingDown`), waits for in-flight
    /// appends to publish, then waits for the Committer to fsync everything up
    /// to the producer cursor.
    ///
    /// After this returns `Ok(Clean)`, every record appended before the call is
    /// durable. The background threads keep running; the process may then exit
    /// (threads are aborted on drop, harmlessly, since data is already durable),
    /// or [`LogDb::shutdown`] may be called to join them.
    pub fn drain(&self, timeout: Duration) -> Result<ShutdownReport, FlushError> {
        let inner = &self.inner;
        let deadline = Instant::now() + timeout;

        // Phase 1: Drain — reject new appends.
        inner.shutdown.start_drain();

        // Wait for all in-flight appends to publish.
        loop {
            if inner.shutdown.in_flight.load(Ordering::Acquire) == 0 {
                break;
            }
            if Instant::now() >= deadline {
                inner.shutdown.abort();
                return Err(FlushError::Timeout);
            }
            std::hint::spin_loop();
        }

        // Phase 2: flush everything published up to each shard's producer cursor.
        let targets = inner.shards.producer_cursors();
        // drain_target is a single best-effort signal (max across shards); the
        // committer drains per-shard via the FlushSignal targets.
        let max_target = targets.iter().copied().max().unwrap_or(0);
        inner
            .shutdown
            .drain_target
            .store(max_target, Ordering::Release);
        inner.flush.request(&targets);

        let remaining = deadline.saturating_duration_since(Instant::now());
        let durable_ok =
            wait_until(&inner.shutdown, || inner.flush.is_done(&targets), remaining).is_ok();

        Ok(if durable_ok {
            ShutdownReport::Clean
        } else {
            ShutdownReport::PartialDurable
        })
    }

    /// Shut down gracefully with timeout: drain (flush all to durable) then
    /// join the background threads. Consumes the handle and requires it be the
    /// only strong reference (so the Committer/Sealer can be joined). For
    /// shared handles (e.g. inside a service), use [`drain`](LogDb::drain).
    pub fn shutdown(mut self, timeout: Duration) -> Result<ShutdownReport, ShutdownError> {
        // Drain first (shared-safe path).
        let report = self.drain(timeout).map_err(|_| ShutdownError::Timeout)?;

        // Join threads — requires exclusive access.
        let inner = match Arc::get_mut(&mut self.inner) {
            Some(i) => i,
            None => return Err(ShutdownError::JoinError("LogDb still referenced".into())),
        };
        if let Some(h) = inner.committer_handle.take() {
            let _ = h.join();
        }
        #[cfg(feature = "hash-chain")]
        for h in inner.sealer_handles.drain(..) {
            let _ = h.join();
        }

        Ok(report)
    }
}

/// Unified wait with timeout and abort checking.
fn wait_until(
    shutdown: &ShutdownState,
    cond: impl Fn() -> bool,
    timeout: Duration,
) -> Result<(), FlushError> {
    let deadline = Instant::now() + timeout;
    let mut spins: u32 = 0;
    loop {
        if cond() {
            return Ok(());
        }
        if shutdown.aborted() {
            return Err(FlushError::Aborted);
        }
        if Instant::now() >= deadline {
            log_warn!("logdb flush/drain timed out waiting for the Committer");
            return Err(FlushError::Timeout);
        }
        spins = spins.saturating_add(1);
        if spins <= 64 {
            std::hint::spin_loop();
        } else if spins <= 256 {
            std::thread::yield_now();
        } else {
            std::thread::sleep(Duration::from_micros(100));
            spins = 128;
        }
    }
}

fn save_checkpoint(dir: &std::path::Path, seq: u64) -> std::io::Result<()> {
    let path = dir.join("checkpoint.dat");
    let tmp = dir.join("checkpoint.tmp");
    let mut buf = [0u8; 12];
    buf[0..8].copy_from_slice(&seq.to_le_bytes());
    let crc = crc32c::crc32c(&buf[..8]);
    buf[8..12].copy_from_slice(&crc.to_le_bytes());
    let mut f = std::fs::File::create(&tmp)?;
    std::io::Write::write_all(&mut f, &buf)?;
    platform::fdatasync(&f)?;
    drop(f);
    std::fs::rename(&tmp, &path)?;
    let d = std::fs::File::open(dir)?;
    platform::sync_dir(&d)?;
    Ok(())
}

/// Sum the sizes of all `*.log` files directly in `dir` (non-recursive).
fn count_log_bytes(dir: &std::path::Path) -> u64 {
    let mut total = 0u64;
    if let Ok(entries) = std::fs::read_dir(dir) {
        for e in entries.flatten() {
            if e.file_name()
                .to_str()
                .map_or(false, |n| n.ends_with(".log"))
            {
                if let Ok(meta) = e.metadata() {
                    total += meta.len();
                }
            }
        }
    }
    total
}

/// Derive the hash-chain key from the encryption key (BLAKE3 keyed KDF).
///
/// When an encryption key is configured, the chain key is derived from it so an
/// attacker who reads the segment file — but not the key — cannot recompute it
/// and forge the chain: a real MAC, vs the clock-seeded tamper-evidence used
/// without a key. The derived value is NEVER written to the segment header
/// (zeros are stored there instead); the sealer and recovery re-derive it from
/// the encryption key.
#[cfg(feature = "hash-chain")]
pub(crate) fn derive_hash_init(key: &[u8; 32]) -> [u8; 32] {
    *blake3::keyed_hash(key, b"logdb-hash-chain-init-v1").as_bytes()
}

/// XOR two 32-byte values. Used to mask the hash-chain key into the segment
/// header under encryption (cr-032 Phase 3): `stored = chain_key ⊕
/// derive(active)`. Recovery inverts it (`chain_key = stored ⊕ derive(key)`),
/// so the chain key never appears in plaintext on disk yet is stable across key
/// rotation (it is no longer derived from the active key).
#[cfg(feature = "hash-chain")]
pub(crate) fn xor32(a: [u8; 32], b: [u8; 32]) -> [u8; 32] {
    let mut out = [0u8; 32];
    for i in 0..32 {
        out[i] = a[i] ^ b[i];
    }
    out
}

fn generate_hash_init() -> [u8; 32] {
    #[cfg(feature = "hash-chain")]
    {
        // Used ONLY when no encryption key is configured. Seeds the chain from
        // non-secret entropy (wall clock) — tamper-EVIDENCE, not authenticity,
        // because the seed is reproducible. With an encryption key, the chain
        // key is derived from it instead (see `derive_hash_init`).
        let mut hasher = blake3::Hasher::new();
        hasher.update(&platform::clock_realtime_coarse_ns().to_le_bytes());
        hasher.update(b"logdb-hash-init-v0.2.0");
        *hasher.finalize().as_bytes()
    }
    #[cfg(not(feature = "hash-chain"))]
    {
        [0u8; 32]
    }
}

impl Drop for LogDb {
    fn drop(&mut self) {
        // Best-effort durability on drop: flush already-published records.
        // Skipped during panic unwinding (I/O during unwind is risky) and
        // bounded by `DROP_DRAIN_TIMEOUT` so a stuck Committer can never hang
        // drop. For a guaranteed-clean shutdown, call `shutdown()` / `drain()`
        // explicitly — drop is a safety net, not a contract.
        if !std::thread::panicking() {
            match self.drain(DROP_DRAIN_TIMEOUT) {
                Ok(ShutdownReport::Clean) => {}
                Ok(_) => {
                    log_warn!(
                        timeout_secs = DROP_DRAIN_TIMEOUT.as_secs(),
                        "logdb dropped with records not fully fsynced; call shutdown()/drain() for guaranteed durability"
                    );
                }
                Err(_) => {
                    log_warn!(
                        timeout_secs = DROP_DRAIN_TIMEOUT.as_secs(),
                        "logdb best-effort drain on drop failed/timed out; in-flight records may be lost"
                    );
                }
            }
        }
        self.inner.shutdown.abort();
    }
}

/// Bound on the best-effort drain performed in `Drop`. Long enough to let the
/// Committer flush a typical in-flight batch, short enough that dropping a
/// `LogDb` never visibly stalls.
const DROP_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);

#[cfg(test)]
mod tests {
    use super::*;
    use config::DurabilityMode;
    use std::sync::Arc;

    #[test]
    fn open_and_append_and_read() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let db = LogDb::open(config).unwrap();
        let id = db.append(b"hello logdb").unwrap();
        db.flush().unwrap();
        std::thread::sleep(Duration::from_millis(100));

        let record = db.read(id).unwrap().unwrap();
        assert_eq!(record.id.sequence, id);
        assert_eq!(record.content, b"hello logdb");
    }

    /// With hash-chain + an encryption key, the chain key is DERIVED from the
    /// key (real MAC) and the segment header must store ZEROS — not the derived
    /// key — so an attacker who reads the file cannot recompute the chain.
    /// Also verifies a round-trip: reopen with the same key reads everything back.
    #[cfg(all(feature = "hash-chain", feature = "encryption"))]
    #[test]
    fn hash_chain_with_key_derives_mac_and_hides_it() {
        use crate::storage::format::{SEGMENT_HEADER_SIZE, SegmentHeader};
        use std::io::Read;

        let key = [0x99u8; 32];
        let dir = tempfile::tempdir().unwrap();
        let data_dir = dir.path().to_path_buf();
        let mk = || {
            let mut c = Config::default();
            c.data_dir = data_dir.clone();
            c.hash_enabled = true;
            c.encryption_keys = Some(KeyRing::single(key));
            c.ring_size = 64;
            c.durability_mode = DurabilityMode::Sync;
            c.flush_timeout = Duration::from_secs(5);
            c
        };

        // Write 5 records, durable.
        {
            let db = LogDb::open(mk()).unwrap();
            for i in 0..5u64 {
                db.append(format!("r-{}", i).as_bytes()).unwrap();
            }
            db.flush().unwrap();
            for _ in 0..100 {
                if db.durable_cursor() >= 5 {
                    break;
                }
                std::thread::sleep(Duration::from_millis(100));
            }
        }

        // cr-032 Phase 3: the chain key is a stable per-shard secret stored
        // MASKED on disk (`chain_key ⊕ derive(active)`) — never plaintext, never
        // zeros. An attacker with the disk but not the key sees only the mask;
        // recovery unmasks it (verifying against the first record) to rebuild
        // the chain. So the header must hold a non-zero, non-plain value.
        let seg = data_dir.join("segment-00000001.log");
        let mut buf = [0u8; SEGMENT_HEADER_SIZE];
        let mut f = std::fs::File::open(&seg).unwrap();
        f.read_exact(&mut buf).unwrap();
        let header = SegmentHeader::deserialize(&buf).unwrap();
        assert_ne!(
            header.hash_init, [0u8; 32],
            "header must not store zeros — the chain key is now masked, not derived at runtime"
        );
        let derived = crate::derive_hash_init(&key);
        assert_ne!(
            header.hash_init, derived,
            "header must store the MASKED chain key, not the bare key-derived value"
        );
        // The active key's id is stamped into the header (cr-032 Phase 3).
        assert_ne!(
            header.encryption_key_id, 0,
            "header must carry the active key id for O(1) recovery / retirement"
        );

        // Reopen with the SAME key: chain re-derives, records read back.
        let db = LogDb::open(mk()).unwrap();
        for i in 0..5u64 {
            let rec = db.read(i).unwrap().expect("record readable after reopen");
            assert_eq!(rec.content, format!("r-{}", i).as_bytes());
        }
    }

    #[test]
    fn read_batch_matches_single_reads() {
        // read_batch must return the same records as N individual reads, in the
        // same order, with None for missing / not-yet-durable ids.
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4; // exercise the per-shard routing inside read_batch
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        // Spread writes across shards (thread-affine routing).
        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..8u64)
                    .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= all_ids.len() {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // A shuffled batch that includes a missing id (u64::MAX) and reorders.
        let mut batch_ids = all_ids.clone();
        batch_ids.push(u64::MAX); // not present -> None
        batch_ids.sort_by_key(|_| std::cmp::Reverse(0)); // keep order, but exercise non-sorted
        batch_ids.reverse();

        let batch = db.read_batch(&batch_ids).unwrap();
        assert_eq!(batch.len(), batch_ids.len());

        // Cross-check each slot against an individual read.
        for (i, &id) in batch_ids.iter().enumerate() {
            let single = db.read(id).unwrap();
            match (&batch[i], &single) {
                (Some(a), Some(b)) => {
                    assert_eq!(a.content, b.content, "content mismatch id {}", id)
                }
                (None, None) => {}
                other => panic!(
                    "slot {} (id {}): batch {:?} vs single {:?}",
                    i, id, other.0, other.1
                ),
            }
        }
        // The missing id is None.
        let missing_pos = batch_ids.iter().position(|&id| id == u64::MAX).unwrap();
        assert!(batch[missing_pos].is_none());
    }

    #[test]
    fn read_batch_empty() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        let db = LogDb::open(config).unwrap();
        assert!(db.read_batch(&[]).unwrap().is_empty());
    }

    #[test]
    fn append_rejected_content_too_large() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.max_content_size = 100;

        let db = LogDb::open(config).unwrap();
        let err = db.append(&vec![0u8; 200]).unwrap_err();
        assert!(matches!(err, AppendError::ContentTooLarge { .. }));
    }

    #[test]
    fn open_rejects_invalid_config_with_structured_error() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.ring_size = 8; // below the power-of-two >= 16 floor
        assert!(matches!(
            LogDb::open(config),
            Err(OpenError::InvalidConfig(ConfigError::InvalidRingSize(8)))
        ));
    }

    #[test]
    fn append_batch_rejects_empty_with_structured_error() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        let db = LogDb::open(config).unwrap();
        let err = db.append_batch(&[]).unwrap_err();
        assert!(matches!(err, AppendError::EmptyBatch));
    }

    #[test]
    fn shutdown_clean() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let db = LogDb::open(config).unwrap();
        for i in 0..10 {
            db.append(format!("r-{}", i).as_bytes()).unwrap();
        }
        let report = db.shutdown(Duration::from_secs(5)).unwrap();
        assert!(matches!(report, ShutdownReport::Clean));
    }

    #[test]
    fn drain_flushes_to_durable_and_rejects_appends_after() {
        // drain() is the shared-safe path logdbd uses on graceful shutdown:
        // it must flush all in-flight records to durable without consuming the
        // handle, and reject further appends once draining.
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let db = LogDb::open(config).unwrap();
        for i in 0..20 {
            db.append(format!("r-{}", i).as_bytes()).unwrap();
        }

        let report = db.drain(Duration::from_secs(5)).unwrap();
        assert!(
            matches!(report, ShutdownReport::Clean),
            "drain must complete clean"
        );
        assert!(
            db.durable_cursor() >= 20,
            "all appended records must be durable after drain"
        );
        for i in 0..20 {
            assert!(
                db.read(i).unwrap().is_some(),
                "record {} readable after drain",
                i
            );
        }

        // Drain phase rejects new appends.
        let err = db.append(b"after-drain").unwrap_err();
        assert!(
            matches!(err, AppendError::ShuttingDown),
            "append after drain must be rejected"
        );
    }

    #[test]
    fn replicate_preserves_sequence_and_is_readable() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let db = LogDb::open(config).unwrap();
        // Replicate 5 records at exact sequences 0..5 with arbitrary timestamps.
        for i in 0..5u64 {
            db.replicate(i, 1_000_000 + i, format!("replica-{}", i).as_bytes())
                .unwrap();
        }
        assert_eq!(db.producer_cursor(), 5, "producer cursor must advance");
        db.flush().unwrap();
        for _ in 0..20 {
            std::thread::sleep(Duration::from_millis(25));
            if db.durable_cursor() >= 5 {
                break;
            }
        }
        assert!(db.durable_cursor() >= 5);

        // Sequences must be EXACTLY preserved (offset semantics).
        for i in 0..5u64 {
            let rec = db.read(i).unwrap().unwrap();
            assert_eq!(rec.id.sequence, i);
            assert_eq!(rec.timestamp_ns, 1_000_000 + i);
            assert_eq!(rec.content, format!("replica-{}", i).as_bytes());
        }
    }

    #[test]
    fn replicate_rejects_out_of_order_and_is_idempotent() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let db = LogDb::open(config).unwrap();
        db.replicate(0, 0, b"a").unwrap();
        // Skipping ahead to 2 (gap at 1) must error — caller retries in order.
        let err = db.replicate(2, 0, b"c").unwrap_err();
        assert!(
            matches!(err, AppendError::Io(_)),
            "expected out-of-order error"
        );
        // Filling the gap succeeds.
        db.replicate(1, 0, b"b").unwrap();
        // Re-replicating an already-applied sequence is a no-op (idempotent).
        db.replicate(0, 0, b"REPLAY").unwrap();
        // Only seq 0 and 1 were applied (seq 2 was rejected) → cursor at 2,
        // unchanged by the idempotent replay.
        assert_eq!(db.producer_cursor(), 2);
    }

    // ── cr-003 Phase 1: sharded write/durability ─────────────────────────

    // Read back every durable record from a single shard's directory by pointing
    // the existing single-shard Reader at it. Within one shard, global ids are
    // monotonic, so the existing reader works (cross-shard routing is Phase 2).
    fn read_all_in_dir(dir: &std::path::Path) -> Vec<Vec<u8>> {
        use std::sync::{Arc, Mutex};
        let manifest = Arc::new(Mutex::new(reader::SegmentManifest::new(dir.to_path_buf())));
        let reader = reader::Reader::new(manifest, None);
        let mut out = Vec::new();
        if let Ok(iter) = reader.scan(0, u64::MAX) {
            for r in iter {
                if let Ok(rec) = r {
                    out.push(rec.content);
                }
            }
        }
        out
    }

    #[test]
    fn append_under_sharding_is_durable_per_shard() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        // Single-threaded appends are thread-affine → all land on one shard
        // (uneven load). flush must still complete and every record be durable.
        for i in 0..6u64 {
            db.append(format!("rec-{}", i).as_bytes()).unwrap();
        }
        db.flush().unwrap();

        let mut got: Vec<Vec<u8>> = (0..2u32)
            .flat_map(|s| read_all_in_dir(&dir.path().join(format!("s{}", s))))
            .collect();
        got.sort();
        let mut want: Vec<Vec<u8>> = (0..6u64)
            .map(|i| format!("rec-{}", i).into_bytes())
            .collect();
        want.sort();
        assert_eq!(
            got, want,
            "all appended records must be durable per-shard under sharding"
        );
    }

    #[test]
    fn append_batch_under_sharding_is_durable_per_shard() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let batch: Vec<&[u8]> = vec![b"alpha", b"beta", b"gamma", b"delta"];
        db.append_batch(&batch).unwrap();
        db.flush().unwrap(); // cr-003: previously timed out / lost the batch

        let mut got: Vec<Vec<u8>> = (0..2u32)
            .flat_map(|s| read_all_in_dir(&dir.path().join(format!("s{}", s))))
            .collect();
        got.sort();
        let mut want: Vec<Vec<u8>> = batch.iter().map(|b| b.to_vec()).collect();
        want.sort();
        assert_eq!(
            got, want,
            "append_batch records must all be durable per-shard under sharding"
        );
    }

    // ── cr-003 Phase 2: sharded read (point lookup) ──────────────────────

    #[test]
    fn read_under_sharding_returns_record_by_global_id() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        // Single-threaded appends are thread-affine → one shard. Capture each
        // returned global id and read it back by that id.
        let mut ids = Vec::new();
        for i in 0..6u64 {
            let id = db.append(format!("rec-{}", i).as_bytes()).unwrap();
            ids.push((id, format!("rec-{}", i).into_bytes()));
        }
        db.flush().unwrap();

        for (id, want) in &ids {
            let rec = db.read(*id).unwrap().expect("readable by global id");
            assert_eq!(&rec.content, want);
            assert_eq!(rec.id.sequence, *id);
        }
    }

    #[test]
    fn read_under_sharding_append_batch_first_record() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let batch: Vec<&[u8]> = vec![b"alpha", b"beta", b"gamma", b"delta"];
        let first = db.append_batch(&batch).unwrap();
        db.flush().unwrap();

        // append_batch returns the first record's global id; read it back.
        let rec = db
            .read(first)
            .unwrap()
            .expect("first batch record readable");
        assert_eq!(rec.content, b"alpha");
        assert_eq!(rec.id.sequence, first);
    }

    #[test]
    fn read_under_sharding_not_visible_before_flush() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Async; // fsync only on explicit flush
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let id = db.append(b"not-yet-durable").unwrap();
        // In Async mode the committer does not fsync without a flush, so the
        // per-shard durable cursor has not advanced → read returns None.
        assert!(db.read(id).unwrap().is_none(), "not visible before durable");
        db.flush().unwrap();
        let rec = db.read(id).unwrap().expect("visible after flush");
        assert_eq!(rec.content, b"not-yet-durable");
    }

    // ── cr-003 Phase 3: cross-shard + cross-segment scan ─────────────────

    #[test]
    fn scan_under_sharding_is_complete_and_ordered() {
        // Multi-thread appends spread across shards (thread-affine routing).
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let db = Arc::new(db);
        let mut handles = Vec::new();
        let mut all_ids = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                let mut ids = Vec::new();
                for i in 0..10u64 {
                    ids.push(db.append(format!("t{}-{}", t, i).as_bytes()).unwrap());
                }
                ids
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();

        // Every appended record must be visible; ids strictly ascending.
        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            scanned.len(),
            all_ids.len(),
            "scan must see every record across shards"
        );
        assert!(
            all_ids.iter().all(|id| scanned.contains(id)),
            "scan missing some ids"
        );
        assert!(
            scanned.windows(2).all(|w| w[0] < w[1]),
            "scan must be strictly ascending"
        );
    }

    #[test]
    fn scan_under_sharding_respects_range() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 128;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for t in 0..2u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..12u64)
                    .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();

        all_ids.sort();
        let from = all_ids[5];
        let to = all_ids[15];
        let got: Vec<u64> = db
            .scan(from, to)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        let want: Vec<u64> = all_ids
            .iter()
            .copied()
            .filter(|&id| id >= from && id < to)
            .collect();
        assert_eq!(
            got, want,
            "scan([from,to)) must clip to the global-id range"
        );
    }

    #[test]
    fn scan_crosses_segment_boundary_single_shard() {
        // Force a segment roll: tiny segment_size, write >1 segment of data.
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 1;
        config.segment_size = 1 * 1024 * 1024; // 1MB minimum
        config.ring_size = 8192;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(60);
        let db = LogDb::open(config).unwrap();

        let payload = vec![0xA5u8; 64 * 1024]; // 64KB each -> ~17 records fill >1MB
        let n = 20u64;
        for _ in 0..n {
            db.append(&payload).unwrap();
        }
        db.flush().unwrap();

        // More than one segment file must exist (the roll happened).
        let segs = std::fs::read_dir(dir.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| {
                e.file_name()
                    .to_str()
                    .map_or(false, |n| n.ends_with(".log"))
            })
            .count();
        assert!(
            segs >= 2,
            "expected a segment roll, found {} segment files",
            segs
        );

        // scan must return ALL records across both segments, ascending.
        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            scanned.len(),
            n as usize,
            "scan must cross the segment boundary"
        );
        assert!(scanned.windows(2).all(|w| w[0] < w[1]));
        assert_eq!((0..n).collect::<Vec<_>>(), scanned);
    }

    #[test]
    fn replay_from_under_sharding_returns_tail_ordered() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..8u64)
                    .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();

        all_ids.sort();
        let pivot = all_ids[10];
        let tail: Vec<u64> = db
            .replay_from(pivot)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        let want: Vec<u64> = all_ids.iter().copied().filter(|&id| id >= pivot).collect();
        assert_eq!(
            tail, want,
            "replay_from must return the ordered tail across shards"
        );
    }

    // ── cr-003 Phase 4: per-shard crash recovery ──────────────────────────

    #[test]
    fn reopen_under_sharding_preserves_all_records() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        // Session 1: spread appends across all 4 shards (thread-affine routing).
        let mut all_ids = Vec::new();
        {
            let db = Arc::new(LogDb::open(config.clone()).unwrap());
            let mut handles = Vec::new();
            for t in 0..4u64 {
                let db = Arc::clone(&db);
                handles.push(std::thread::spawn(move || {
                    (0..10u64)
                        .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                        .collect::<Vec<_>>()
                }));
            }
            for h in handles {
                all_ids.extend(h.join().unwrap());
            }
            db.flush().unwrap();
            let db = Arc::try_unwrap(db).ok().unwrap();
            db.shutdown(Duration::from_secs(5)).unwrap();
        }

        // Session 2: reopen — recovery must run per shard and preserve every record.
        let db = LogDb::open(config).unwrap();
        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            scanned.len(),
            all_ids.len(),
            "reopen must preserve every record across shards"
        );
        assert!(
            all_ids.iter().all(|id| scanned.contains(id)),
            "reopen lost some ids"
        );
        assert!(
            scanned.windows(2).all(|w| w[0] < w[1]),
            "scanned ids must be strictly ascending"
        );
    }

    #[test]
    fn reopen_under_sharding_single_thread_handles_empty_shards() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let ids: Vec<u64> = {
            let db = LogDb::open(config.clone()).unwrap();
            // Single-threaded appends are thread-affine → all land on ONE shard;
            // the other three shards get only an empty first segment.
            let mut ids = Vec::new();
            for i in 0..6u64 {
                ids.push(db.append(format!("r-{}", i).as_bytes()).unwrap());
            }
            db.flush().unwrap();
            db.shutdown(Duration::from_secs(5)).unwrap();
            ids
        };

        let db = LogDb::open(config).unwrap();
        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            scanned, ids,
            "empty shards must not lose the written shard's records"
        );
        // The empty shards resume at local 0; a fresh append must still produce a
        // collision-free global id and be readable.
        let nid = db.append(b"after-reopen").unwrap();
        db.flush().unwrap();
        assert!(
            db.read(nid).unwrap().is_some(),
            "post-reopen append must read back"
        );
    }

    #[test]
    fn reopen_under_sharding_then_append_no_id_collision() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let mut all_ids = Vec::new();
        {
            let db = Arc::new(LogDb::open(config.clone()).unwrap());
            let mut handles = Vec::new();
            for t in 0..4u64 {
                let db = Arc::clone(&db);
                handles.push(std::thread::spawn(move || {
                    (0..8u64)
                        .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                        .collect::<Vec<_>>()
                }));
            }
            for h in handles {
                all_ids.extend(h.join().unwrap());
            }
            db.flush().unwrap();
            let db = Arc::try_unwrap(db).ok().unwrap();
            db.shutdown(Duration::from_secs(5)).unwrap();
        }

        let db = Arc::new(LogDb::open(config).unwrap());
        // Append more from multiple threads after reopen — per-shard resume points
        // must keep the global id space collision-free.
        let mut new_ids = Vec::new();
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..5u64)
                    .map(|i| db.append(format!("n{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        for h in handles {
            new_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();

        let mut seen = std::collections::HashSet::new();
        for id in all_ids.iter().chain(new_ids.iter()) {
            assert!(seen.insert(*id), "global id collision after reopen: {}", id);
        }
        // Every old id is still readable after the post-reopen appends.
        for id in &all_ids {
            assert!(
                db.read(*id).unwrap().is_some(),
                "old id {} lost after reopen+append",
                id
            );
        }
    }

    #[test]
    fn reopen_under_sharding_detects_torn_write() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.ring_size = 128;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        // Single-threaded → all 20 records land on ONE shard; the other is empty.
        {
            let db = LogDb::open(config.clone()).unwrap();
            for i in 0..20u64 {
                db.append(format!("r-{}", i).as_bytes()).unwrap();
            }
            db.flush().unwrap();
            db.shutdown(Duration::from_secs(5)).unwrap();
        }

        // Corrupt the non-empty shard's active segment: chop 5 bytes off the end
        // (mid-record) → the last record becomes a torn tail.
        let victim = (0..2u32)
            .map(|s| {
                dir.path()
                    .join(format!("s{}", s))
                    .join("segment-00000001.log")
            })
            .find(|p| p.exists() && std::fs::metadata(p).map(|m| m.len() > 200).unwrap_or(false))
            .expect("some shard must have received records");
        let len = std::fs::metadata(&victim).unwrap().len();
        std::fs::OpenOptions::new()
            .write(true)
            .open(&victim)
            .unwrap()
            .set_len(len - 5)
            .unwrap();

        // Recovery must succeed: the torn tail is truncated, the rest survives.
        // (The pre-fix stride bug truncated to a single record; this guards that.)
        let db = LogDb::open(config).unwrap();
        let scanned: Vec<Vec<u8>> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.content)
            .collect();
        assert!(
            scanned.len() >= 18,
            "torn-write recovery lost too many records: {}",
            scanned.len()
        );
    }

    #[test]
    fn reopen_shards1_still_recovers() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.ring_size = 64;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);

        let id = {
            let db = LogDb::open(config.clone()).unwrap();
            let id = db.append(b"shards1-recovery").unwrap();
            db.flush().unwrap();
            db.shutdown(Duration::from_secs(5)).unwrap();
            id
        };
        // shards=1 goes through the same unified per-shard loop (shard_bits=0).
        let db = LogDb::open(config).unwrap();
        let rec = db
            .read(id)
            .unwrap()
            .expect("shards=1 record must survive reopen");
        assert_eq!(rec.content, b"shards1-recovery");
        // Resume + append, no collision.
        let nid = db.append(b"after").unwrap();
        db.flush().unwrap();
        assert!(db.read(nid).unwrap().is_some());
    }

    // ── cr-003 Phase 5: cross-shard tailer (per-shard progress + merge) ───

    #[test]
    fn tailer_under_sharding_reads_all_shards_merged() {
        // Multi-thread appends spread across shards (thread-affine routing).
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let total = 40u64; // 4 threads × 10
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..10u64)
                    .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        let mut all_ids = Vec::new();
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();
        // Wait until every shard is durable (min durable cursor advances).
        for _ in 0..100 {
            if db.durable_cursor() >= 10 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        let mut t = db.new_tailer("merge");
        let mut got: Vec<u64> = Vec::new();
        // Drain in batches; bound the wait so the test fails fast if broken.
        for _ in 0..200 {
            match t.next_batch(1000).unwrap() {
                Some(batch) => got.extend(batch.iter().map(|r| r.id.sequence)),
                None => break,
            }
            std::thread::sleep(Duration::from_millis(5));
        }

        assert_eq!(
            got.len(),
            total as usize,
            "tailer must see every record across ALL shards"
        );
        assert!(
            got.windows(2).all(|w| w[0] < w[1]),
            "tailer batch must be ascending global id"
        );
        assert!(
            all_ids.iter().all(|id| got.contains(id)),
            "tailer missing some ids: got={:?}",
            got
        );
        // Records must come from more than one shard (proves cross-shard merge).
        let shards_seen: std::collections::HashSet<usize> = got
            .iter()
            .map(|&g| crate::shard::decode_record_id(g, 2).0)
            .collect();
        assert!(
            shards_seen.len() > 1,
            "tailer should have merged multiple shards, saw {:?}",
            shards_seen
        );
    }

    #[test]
    fn tailer_under_sharding_single_shard_is_delivered() {
        // Single-thread appends all land on ONE shard (thread-affine routing).
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let n = 60u64;
        let mut ids = Vec::new();
        for i in 0..n {
            ids.push(db.append(format!("s-{}", i).as_bytes()).unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.durable_cursor() >= 1 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        let mut t = db.new_tailer("one-shard");
        let mut got: Vec<u64> = Vec::new();
        for _ in 0..200 {
            match t.next_batch(1000).unwrap() {
                Some(b) => got.extend(b.iter().map(|r| r.id.sequence)),
                None => break,
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(
            got.len(),
            n as usize,
            "single-thread writes (one shard) must all be delivered"
        );
        assert!(ids.iter().all(|id| got.contains(id)));
    }

    #[test]
    fn tailer_under_sharding_persists_per_shard_progress() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..10u64)
                    .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.durable_cursor() >= 10 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Drain 12 records, commit, then reopen.
        let mut delivered = Vec::new();
        {
            let mut t = db.new_tailer("persist");
            while delivered.len() < 12 {
                match t.next_batch(12).unwrap() {
                    Some(b) => delivered.extend(b.iter().map(|r| r.id.sequence)),
                    None => break,
                }
            }
            t.commit().unwrap();
        }
        let saved_positions = db.new_tailer("persist").positions().to_vec();
        assert!(
            saved_positions.iter().any(|&p| p > 0),
            "some shard progress must have persisted"
        );

        // Reopen and drain the rest; total must be all 40, exactly once.
        let mut t = db.new_tailer("persist");
        assert_eq!(
            t.positions(),
            saved_positions.as_slice(),
            "reopened tailer must restore per-shard positions"
        );
        let mut rest = Vec::new();
        for _ in 0..200 {
            match t.next_batch(1000).unwrap() {
                Some(b) => rest.extend(b.iter().map(|r| r.id.sequence)),
                None => break,
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        let mut all = delivered.clone();
        all.extend(rest);
        all.sort();
        all_ids.sort();
        assert_eq!(
            all, all_ids,
            "commit+reopen must deliver every record exactly once"
        );
    }

    #[test]
    fn tailer_under_sharding_resumes_without_loss() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let spawn = |db: &Arc<LogDb>| -> Vec<u64> {
            let mut handles = Vec::new();
            for t in 0..4u64 {
                let db = Arc::clone(db);
                handles.push(std::thread::spawn(move || {
                    (0..8u64)
                        .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                        .collect::<Vec<_>>()
                }));
            }
            let mut ids = Vec::new();
            for h in handles {
                ids.extend(h.join().unwrap());
            }
            ids
        };

        let first = spawn(&db);
        db.flush().unwrap();
        for _ in 0..100 {
            if db.durable_cursor() >= 8 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Drain the first wave fully and commit.
        {
            let mut t = db.new_tailer("resume");
            let mut got = Vec::new();
            for _ in 0..200 {
                match t.next_batch(1000).unwrap() {
                    Some(b) => got.extend(b.iter().map(|r| r.id.sequence)),
                    None => break,
                }
                std::thread::sleep(Duration::from_millis(5));
            }
            assert_eq!(got.len(), first.len());
            t.commit().unwrap();
        }

        // Append a second wave.
        let second = spawn(&db);
        db.flush().unwrap();
        for _ in 0..100 {
            if db.durable_cursor() >= 8 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Reopen: must deliver ONLY the second wave (no re-delivery, no loss).
        let mut t = db.new_tailer("resume");
        let mut got = Vec::new();
        for _ in 0..200 {
            match t.next_batch(1000).unwrap() {
                Some(b) => got.extend(b.iter().map(|r| r.id.sequence)),
                None => break,
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        let mut want = second.clone();
        let mut got_sorted = got.clone();
        want.sort();
        got_sorted.sort();
        assert_eq!(
            got_sorted, want,
            "reopened tailer must deliver only the newly-appended records"
        );
    }

    #[test]
    fn tailer_crosses_segment_under_sharding() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.segment_size = 1 * 1024 * 1024; // 1MB → rolls quickly with big payloads
        config.ring_size = 8192;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(60);
        let db = Arc::new(LogDb::open(config).unwrap());

        // Spread big records across both shards so at least one shard rolls.
        let mut handles = Vec::new();
        for t in 0..2u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                let payload = vec![0xA5u8 + t as u8; 64 * 1024]; // 64KB each
                for _ in 0..20u64 {
                    db.append(&payload).unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.durable_cursor() >= 1 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        let mut t = db.new_tailer("xseg");
        let mut count = 0usize;
        for _ in 0..300 {
            match t.next_batch(1000).unwrap() {
                Some(b) => count += b.len(),
                None => break,
            }
            std::thread::sleep(Duration::from_millis(5));
        }
        assert_eq!(
            count,
            2 * 20,
            "tailer must cross segment boundaries within each shard"
        );
    }

    // ── cr-003 Phase 6: hardening + production-ready docs ─────────────────

    #[test]
    fn wal_usage_under_sharding_sums_all_shard_dirs() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                for i in 0..10u64 {
                    db.append(format!("t{}-{}", t, i).as_bytes()).unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.durable_cursor() >= 10 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        let (used, total) = db.wal_usage();
        assert!(total > 0, "segment_size total should be reported");
        assert!(
            used > 0,
            "wal_usage must sum segment files across ALL shard subdirs (got {})",
            used
        );
    }

    #[test]
    fn recovery_report_under_sharding_counts_across_shards() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let total = 40u64;
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                for i in 0..10u64 {
                    db.append(format!("t{}-{}", t, i).as_bytes()).unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        db.flush().unwrap();
        // flush() is the durability sync point; wait until every record is on disk.
        let scan_count = loop {
            let n = db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count();
            if n >= total as usize {
                break n;
            }
            std::thread::sleep(Duration::from_millis(100));
        };

        // checkpoint defaults to 0 → every durable record (gid >= 0) counts.
        // count must equal the full-scan ground truth (sum across ALL shards),
        // not the min per-shard durable cursor.
        let report = db.recovery_report();
        assert_eq!(report.from_sequence, 0);
        assert_eq!(
            report.count as usize, scan_count,
            "recovery_report.count must equal total durable records across shards (got {}, scan={})",
            report.count, scan_count
        );
    }

    #[test]
    // FIXME: consistently fails on WSL2 (scan sees 45/60 records after
    // flush+roll+truncation). Underlying issue: per-shard manifests may not
    // refresh segment listings after a segment roll that occurs during the
    // second write batch. Works reliably on bare-metal Linux. See cr-026.
    #[serial_test::serial]
    fn checkpoint_truncation_under_sharding_preserves_post_checkpoint_data() {
        // Truncation removes only fully-checkpointed segments, per shard. After
        // a roll past a checkpoint, every record with gid >= checkpoint must
        // still be readable (no over-truncation, no data loss).
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.segment_size = 1 * 1024 * 1024; // 1MB → rolls with 64KB payloads
        config.ring_size = 8192;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(60);
        let db = Arc::new(LogDb::open(config).unwrap());

        let payload = vec![0xA5u8; 64 * 1024];
        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for _ in 0..2u64 {
            let db = Arc::clone(&db);
            let p = payload.clone();
            handles.push(std::thread::spawn(move || {
                let mut ids = Vec::new();
                for _ in 0..20u64 {
                    ids.push(db.append(&p).unwrap());
                }
                ids
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();
        // flush made the records durable; force a manifest rescan so the
        // newly-rolled segments are visible without waiting for directory
        // mtime propagation (the old polling loop flaked under load — see
        // `LogDb::refresh_manifests`).
        db.refresh_manifests().unwrap();
        let count = db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count();
        assert_eq!(
            count,
            all_ids.len(),
            "all appended records must be visible after flush + refresh"
        );

        all_ids.sort();
        // Checkpoint past the first segment of every shard, then append more to
        // force rolls that trigger per-shard truncation.
        let cp = all_ids[all_ids.len() / 2];
        db.checkpoint(cp);
        let mut more_ids = Vec::new();
        for _ in 0..20u64 {
            more_ids.push(db.append(&payload).unwrap());
        }
        db.flush().unwrap();
        // Truncation deleted segments fully before the checkpoint; refresh so
        // the manifest reflects the deletions, then assert exactly the
        // surviving (>= checkpoint) records are visible — no polling.
        db.refresh_manifests().unwrap();
        let surviving = all_ids.iter().filter(|&&id| id >= cp).count() + more_ids.len();
        let count = db
            .scan(cp, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .count();
        assert!(
            count >= surviving,
            "expected at least {surviving} records at/after checkpoint, got {count}"
        );

        // Every surviving record must be individually readable.
        for id in all_ids
            .iter()
            .chain(more_ids.iter())
            .filter(|&&id| id >= cp)
        {
            assert!(
                db.read(*id).unwrap().is_some(),
                "post-checkpoint id {} lost after truncation",
                id
            );
        }
    }

    #[test]
    fn point_read_across_segment_roll_under_sharding() {
        // Directly exercises the base_sequence fix: under shards>1, a point read
        // by GLOBAL id must route to the correct segment even after a roll.
        // Before the fix, pre-allocated segment headers stored the LOCAL ring
        // seq as base_sequence, so SegmentManifest::find misrouted reads.
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.segment_size = 1 * 1024 * 1024; // 1MB → rolls with 64KB payloads
        config.ring_size = 8192;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(60);
        let db = Arc::new(LogDb::open(config).unwrap());

        let payload = vec![0xA5u8; 64 * 1024];
        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for _ in 0..2u64 {
            let db = Arc::clone(&db);
            let p = payload.clone();
            handles.push(std::thread::spawn(move || {
                let mut ids = Vec::new();
                for _ in 0..20u64 {
                    ids.push(db.append(&p).unwrap());
                }
                ids
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= all_ids.len() {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // At least two segment files exist in some shard dir (a roll happened).
        let rolled = (0..2).any(|s| {
            std::fs::read_dir(dir.path().join(format!("s{}", s)))
                .map(|rd| {
                    rd.filter_map(|e| e.ok())
                        .filter(|e| {
                            e.file_name()
                                .to_str()
                                .map_or(false, |n| n.ends_with(".log"))
                        })
                        .count()
                        >= 2
                })
                .unwrap_or(false)
        });
        assert!(rolled, "test should have triggered a segment roll");

        // Every appended id must be readable by global id, including those in
        // non-first segments (this is what the base_sequence fix preserves).
        for id in &all_ids {
            assert!(
                db.read(*id).unwrap().is_some(),
                "point read of id {} failed across segment roll",
                id
            );
        }
    }

    #[test]
    fn retention_maxbytes_applies_per_shard() {
        use crate::config::RetentionPolicy;
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.segment_size = 1 * 1024 * 1024;
        config.ring_size = 8192;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(60);
        config.retention = RetentionPolicy::MaxBytes(2 * 1024 * 1024); // 2MB cap
        let db = Arc::new(LogDb::open(config).unwrap());

        let payload = vec![0xA5u8; 64 * 1024];
        let mut handles = Vec::new();
        for _ in 0..2u64 {
            let db = Arc::clone(&db);
            let p = payload.clone();
            handles.push(std::thread::spawn(move || {
                for _ in 0..40u64 {
                    db.append(&p).unwrap();
                }
            }));
        }
        for h in handles {
            h.join().unwrap();
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= 40 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Each shard dir is retained independently; neither grows unbounded.
        for s in 0..2 {
            let sd = dir.path().join(format!("s{}", s));
            let bytes: u64 = std::fs::read_dir(&sd)
                .unwrap()
                .filter_map(|e| e.ok())
                .filter(|e| {
                    e.file_name()
                        .to_str()
                        .map_or(false, |n| n.ends_with(".log"))
                })
                .filter_map(|e| e.metadata().ok())
                .map(|m| m.len())
                .sum();
            // Retention evicts whole segments only after a roll; with a 2MB cap
            // and 1MB segments, each shard holds at most a few segments.
            assert!(
                bytes <= 4 * 1024 * 1024,
                "shard {} retained {} bytes (retention not applied)",
                s,
                bytes
            );
        }
    }

    #[test]
    fn non_power_of_two_shards_end_to_end() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 3; // non-power-of-two → shard_bits=2, shard id 3 unused
        config.ring_size = 256;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config.clone()).unwrap());

        let mut all_ids = Vec::new();
        let mut handles = Vec::new();
        for t in 0..6u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                (0..5u64)
                    .map(|i| db.append(format!("t{}-{}", t, i).as_bytes()).unwrap())
                    .collect::<Vec<_>>()
            }));
        }
        for h in handles {
            all_ids.extend(h.join().unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= all_ids.len() {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Every id decodes to a valid shard (< 3) and reads back.
        for id in &all_ids {
            let (shard, _local) = crate::shard::decode_record_id(*id, 2);
            assert!(
                shard < 3,
                "non-power-of-2 must not produce shard id >= num_shards"
            );
            assert!(db.read(*id).unwrap().is_some(), "id {} must read back", id);
        }
        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(scanned.len(), all_ids.len());
        assert!(scanned.windows(2).all(|w| w[0] < w[1]));

        // Reopen preserves everything (per-shard recovery).
        drop(db);
        let db = LogDb::open(config).unwrap();
        let rescanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            rescanned.len(),
            all_ids.len(),
            "non-power-of-2 shards must survive reopen"
        );
    }

    #[test]
    fn concurrent_append_scan_tailer_under_sharding() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 512;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config).unwrap());

        let total = 80u64;
        let db2 = Arc::clone(&db);
        let writer = std::thread::spawn(move || {
            let mut ids = Vec::new();
            for i in 0..total {
                ids.push(db2.append(format!("x-{}", i).as_bytes()).unwrap());
            }
            db2.flush().unwrap();
            ids
        });

        // Concurrent reader: scan repeatedly while writing; must never panic.
        let db3 = Arc::clone(&db);
        let reader = std::thread::spawn(move || {
            let mut last = 0usize;
            for _ in 0..20 {
                let n = db3
                    .scan(0, u64::MAX)
                    .unwrap()
                    .filter_map(|r| r.ok())
                    .count();
                if n > last {
                    last = n;
                }
                std::thread::sleep(Duration::from_millis(2));
            }
            last
        });

        let ids = writer.join().unwrap();
        let _seen = reader.join().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= total as usize {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Tailer must eventually deliver every record (lossless).
        let mut t = db.new_tailer("stress");
        let mut got: Vec<u64> = Vec::new();
        for _ in 0..400 {
            match t.next_batch(1000).unwrap() {
                Some(b) => got.extend(b.iter().map(|r| r.id.sequence)),
                None => break,
            }
            std::thread::sleep(Duration::from_millis(2));
        }
        assert_eq!(
            got.len(),
            total as usize,
            "tailer must deliver all under concurrency"
        );
        assert!(ids.iter().all(|id| got.contains(id)));
    }

    // ── cr-004: scan optimization characterization tests ─────────────────

    #[test]
    fn scan_raw_large_records_across_chunk_boundary() {
        // Records larger than the read chunk force the buffer to compact/grow.
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 1;
        config.ring_size = 4096;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let big = vec![0xA5u8; 80 * 1024]; // 80KB each — > 64KB chunk
        let n = 40u64;
        let mut ids = Vec::new();
        for _ in 0..n {
            ids.push(db.append(&big).unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= n as usize {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        let scanned: Vec<(u64, Vec<u8>)> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| (r.id.sequence, r.content.clone()))
            .collect();
        assert_eq!(scanned.len(), n as usize);
        for (id, content) in &scanned {
            assert_eq!(content, &big, "large-record content must survive buffering");
            assert!(ids.contains(id));
        }
    }

    #[test]
    fn scan_raw_respects_range_after_buffering() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 1;
        config.ring_size = 4096;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();

        let mut ids = Vec::new();
        for i in 0..200u64 {
            ids.push(db.append(format!("r-{}", i).as_bytes()).unwrap());
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= 200 {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        ids.sort();
        let from = ids[50];
        let to = ids[150];
        let got: Vec<u64> = db
            .scan(from, to)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        let want: Vec<u64> = ids
            .iter()
            .copied()
            .filter(|&id| id >= from && id < to)
            .collect();
        assert_eq!(got, want, "range scan must be exact after buffering");
    }

    // ── cr-005: feature × shards matrix coverage (release-prep) ──────────

    #[test]
    fn replicate_rejects_shards_gt_1() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 2;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = LogDb::open(config).unwrap();
        let err = db.replicate(0, 0, b"x").unwrap_err();
        assert!(
            matches!(err, crate::AppendError::Io(_)),
            "replicate must reject shards>1"
        );
        assert!(
            format!("{}", err).contains("shards=1"),
            "error should explain the constraint, got: {}",
            err
        );
    }

    #[cfg(feature = "compression")]
    #[test]
    fn sharded_compressed_log_end_to_end() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.compression_enabled = true;
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config.clone()).unwrap());

        // Spread compressed (frame-mode) writes across all shards.
        let mut by_id = std::collections::HashMap::<u64, Vec<u8>>::new();
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                let mut v = Vec::new();
                for i in 0..10u64 {
                    let c = format!("t{}-{}", t, i).into_bytes();
                    let id = db.append(&c).unwrap();
                    v.push((id, c));
                }
                v
            }));
        }
        for h in handles {
            for (id, c) in h.join().unwrap() {
                by_id.insert(id, c);
            }
        }
        let all_ids: Vec<u64> = by_id.keys().copied().collect();
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= all_ids.len() {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // Frame-mode scan under shards>1: every record, ascending, readable.
        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            scanned.len(),
            all_ids.len(),
            "compressed scan must see all shards"
        );
        assert!(scanned.windows(2).all(|w| w[0] < w[1]));
        for (id, expected) in &by_id {
            let rec = db
                .read(*id)
                .unwrap()
                .expect("compressed point read under shards>1");
            assert_eq!(&rec.content, expected, "content mismatch for id {}", id);
        }

        // Reopen → frame-mode recovery under shards>1.
        drop(db);
        let db = LogDb::open(config).unwrap();
        let rescanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            rescanned.len(),
            all_ids.len(),
            "compressed shards>1 must survive reopen"
        );
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn sharded_encrypted_log_end_to_end() {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 4;
        config.ring_size = 256;
        config.encryption_keys = Some(KeyRing::single([0x42u8; 32]));
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(5);
        let db = Arc::new(LogDb::open(config.clone()).unwrap());

        let mut by_id = std::collections::HashMap::<u64, Vec<u8>>::new();
        let mut handles = Vec::new();
        for t in 0..4u64 {
            let db = Arc::clone(&db);
            handles.push(std::thread::spawn(move || {
                let mut v = Vec::new();
                for i in 0..10u64 {
                    let c = format!("t{}-{}", t, i).into_bytes();
                    let id = db.append(&c).unwrap();
                    v.push((id, c));
                }
                v
            }));
        }
        for h in handles {
            for (id, c) in h.join().unwrap() {
                by_id.insert(id, c);
            }
        }
        let all_ids: Vec<u64> = by_id.keys().copied().collect();
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= all_ids.len() {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        let scanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            scanned.len(),
            all_ids.len(),
            "encrypted scan must see all shards"
        );
        assert!(scanned.windows(2).all(|w| w[0] < w[1]));
        for (id, expected) in &by_id {
            let rec = db
                .read(*id)
                .unwrap()
                .expect("encrypted point read under shards>1");
            assert_eq!(&rec.content, expected, "content mismatch for id {}", id);
        }

        drop(db);
        let db = LogDb::open(config).unwrap();
        let rescanned: Vec<u64> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| r.id.sequence)
            .collect();
        assert_eq!(
            rescanned.len(),
            all_ids.len(),
            "encrypted shards>1 must survive reopen"
        );
    }

    #[cfg(feature = "compression")]
    #[test]
    fn compressed_scan_crosses_segment_roll() {
        compressed_or_encrypted_scan_crosses_roll(true, None);
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn encrypted_scan_crosses_segment_roll() {
        compressed_or_encrypted_scan_crosses_roll(false, Some([0x99u8; 32]));
    }

    /// Shared harness: shards=1 frame-mode (compressed and/or encrypted),
    /// force a segment roll with large payloads, scan all records across it.
    fn compressed_or_encrypted_scan_crosses_roll(compressed: bool, key: Option<[u8; 32]>) {
        let dir = tempfile::tempdir().unwrap();
        let mut config = Config::default();
        config.data_dir = dir.path().to_path_buf();
        config.shards = 1;
        config.segment_size = 1 * 1024 * 1024; // 1MB → rolls with 64KB payloads
        config.ring_size = 8192;
        config.compression_enabled = compressed;
        config.encryption_keys = key.map(KeyRing::single);
        config.durability_mode = DurabilityMode::Sync;
        config.flush_timeout = Duration::from_secs(60);
        let db = LogDb::open(config).unwrap();

        // Each record gets a UNIQUE incompressible payload (splitmix64 over
        // record_index*65536 + byte_index). Uniqueness matters under compression:
        // reusing one payload would let zstd deduplicate the 40 records to almost
        // nothing and never roll. Size-preserving for the encrypted path.
        let smix = |x: u64| -> u8 {
            let mut z = x;
            z = (z ^ (z >> 30)).wrapping_mul(0xbf58476d1ce4e5b9);
            z = (z ^ (z >> 27)).wrapping_mul(0x94d049bb133111eb);
            (z ^ (z >> 31)) as u8
        };
        let n = 40u64; // 40 × 64KB = 2.5MB > 1MB → at least one roll (even with compression)
        let mut payloads: Vec<Vec<u8>> = Vec::with_capacity(n as usize);
        for r in 0..n {
            let base = r * (64 * 1024) as u64;
            let p: Vec<u8> = (0..64 * 1024).map(|j| smix(base + j as u64)).collect();
            db.append(&p).unwrap();
            payloads.push(p);
        }
        db.flush().unwrap();
        for _ in 0..100 {
            if db.scan(0, u64::MAX).unwrap().filter_map(|r| r.ok()).count() >= n as usize {
                break;
            }
            std::thread::sleep(Duration::from_millis(100));
        }

        // At least two segment files exist (a roll happened).
        let segs = std::fs::read_dir(dir.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .filter(|e| {
                e.file_name()
                    .to_str()
                    .map_or(false, |n| n.ends_with(".log"))
            })
            .count();
        assert!(
            segs >= 2,
            "expected a frame-mode segment roll, found {}",
            segs
        );

        // Frame-mode RecordIter must cross the roll and return all records,
        // with correct (decrypted/decompressed) content.
        let scanned: Vec<(u64, Vec<u8>)> = db
            .scan(0, u64::MAX)
            .unwrap()
            .filter_map(|r| r.ok())
            .map(|r| (r.id.sequence, r.content.clone()))
            .collect();
        assert_eq!(
            scanned.len(),
            n as usize,
            "frame-mode scan must cross the segment roll"
        );
        assert_eq!(
            scanned.iter().map(|(id, _)| *id).collect::<Vec<_>>(),
            (0..n).collect::<Vec<_>>(),
        );
        for (id, content) in &scanned {
            assert_eq!(
                content, &payloads[*id as usize],
                "content mismatch for id {}",
                id
            );
        }
    }
}