armdb 0.14.2

#![allow(clippy::type_complexity)]

use std::hash::Hash;
use std::sync::{Arc, Weak};
use std::time::{Duration, Instant};
use std::{collections::BTreeMap, path::PathBuf};

use crate::sync::{Mutex, lock};
use armour_core::persist::Persist;
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

use crate::compaction::Compactor;
use crate::error::SchemaMismatchKind;
use crate::flusher::Flusher;
use crate::format_migration::{
    StorageDescriptor,
    bitcask::BackendMigrationOutcome,
    fixed::migrate_fixed,
    marker::{
        FormatMarker, cleanup_stale_format_temps, read_collection_marker, remove_collection_marker,
        write_collection_marker,
    },
    plan::{MigrationPlan, build_format_only_plan, build_plan, build_replay_plan},
};
use crate::hook::TypedWriteHook;
#[cfg(all(feature = "typed-tree", feature = "armour"))]
use crate::queue::wait::QueueWakeService;
use crate::shutdown::ShutdownSignal;
use crate::sweeper::Sweeper;
use crate::ttl::{SweepOutcome, log_sweep_failures};
use crate::{
    Codec, CollectionMeta, Config, DbError, DbResult, FixedConfig, Key, NoHook, TreeMeta, TtlHook,
};
use crate::{
    TypedMap, TypedTree, UniqueIndex, UniqueNamespace, UniqueRow, UniqueTargets, ZeroMap, ZeroTree,
};

use super::collection::Collection;
use super::inspection::InspectionHandler;
use super::migration::TypedMigration;
use super::seq::SeqGen;

#[cfg(all(feature = "typed-tree", feature = "armour"))]
use crate::queue::{Queue, QueueConfig, QueueSpec, retention::QueueRetentionPass};

#[derive(Serialize, Deserialize, Default, Debug, Clone)]
pub struct DbInfo {
    #[serde(default)]
    pub version: u32,
    pub collections: BTreeMap<String, CollectionInfo>,
    /// Set to `true` by `Db::close`; cleared to `false` at open-time.
    /// A `false` value on open means the previous session was unclean (crash
    /// or missing `close()`) and persisted `CollectionInfo::seq` counts are
    /// stale — callers that care must re-derive the count from live data.
    #[serde(default)]
    pub clean: bool,
}

/// The managed-LSM capability retained for one keyspace mutation. It owns the
/// root lease as well as `db.info`, so a managed tree cannot write after its
/// originating [`Db`] releases exclusive ownership of the root directory.
#[cfg(feature = "fjall")]
pub(crate) struct ManagedLsmLifetime {
    pub(crate) db_info: Persist<DbInfo>,
    pub(crate) _root_lease: Arc<std::fs::File>,
}

#[derive(Serialize, Deserialize, Default, Debug, Clone)]
pub struct CollectionInfo {
    pub version: u16,
    /// Schema type fingerprint (`meta.ty.h()`). Value `0` is the
    /// [`TYP_HASH_MIGRATING`] sentinel (interrupted multi-step migration).
    pub typ_hash: u64,
    #[serde(default)]
    pub seq: u64,
    #[serde(default)]
    pub projections_dirty: bool,
    /// Counter deltas for this collection may have been lost. Persisted when a
    /// counted **source** is opened, and cleared only by a successful close.
    /// On a *target* it means the row content moved under the counter
    /// (migration), and only a source's own open consumes that reading.
    #[serde(default)]
    pub counters_dirty: bool,
    #[serde(default)]
    pub gsn_high: u64,
    #[serde(default)]
    pub shard_gsn_high: Vec<u64>,
    #[serde(default)]
    pub unique_dirty: bool,
}

#[derive(Serialize, Deserialize, Default, Debug, Clone)]
struct UserMeta {
    db: Option<serde_json::Value>,
    collections: BTreeMap<String, serde_json::Value>,
}

const DEFAULT_COMPACTION_INTERVAL: Duration = Duration::from_secs(60);
const DEFAULT_FLUSH_INTERVAL: Duration = Duration::from_secs(1);
const DEFAULT_TTL_SWEEP_INTERVAL: Duration = Duration::from_secs(60);
const DEFAULT_TTL_BUDGET: usize = 4096;

/// Intervals and budgets for `Db` background workers.
///
/// [`Default`] is what [`Db::open`] uses: compaction 60 s, flush 1 s, TTL sweep
/// 60 s with a 4096-row budget. [`Db::open_test`] disables all three workers.
#[derive(Debug, Clone, Copy)]
pub struct WorkerIntervals {
    /// Compaction tick; `None` disables the compaction worker. Default 60 s.
    pub compaction: Option<Duration>,
    /// Flush tick — the durability cadence. Each pass runs
    /// [`Collection::periodic_flush`](super::collection::Collection::periodic_flush)
    /// on every registered collection. On Bitcask and FixedStore that is a
    /// dirty-shard fsync/fdatasync; on LSM it is the configured `PersistPolicy`.
    /// A Bitcask fsync stage records `armdb.flush.barriers` with a `result` of
    /// `ok` or `failed`; drain failures remain in the pass's attempted/failed
    /// counts because they never reach a fsync stage. A shard that needs no
    /// barrier records `result = "skipped"`.
    ///
    /// Every pass — clean ones included — publishes two gauges:
    /// `armdb.flush.dirty_shards` (shards this pass had to barrier) and
    /// `armdb.flush.unsynced_shards` (shards whose barrier failed, so they stay
    /// in the loss window until the next tick). With the worker disabled
    /// (`flush: None`) nothing publishes them; [`Db::dirty_shards`] answers the
    /// same question on demand.
    ///
    /// `Config::sync_interval` is checked only as a write returns. This tick is
    /// therefore the deadline that covers pending writes while a collection is
    /// silent; it is not merely a fallback for a missed timer.
    ///
    /// Loss window for ordinary non-transactional writes, when passes complete
    /// and barriers succeed: `interval + 2 × pass duration` (+ scheduling delay).
    /// `None` disables **this worker only** — explicit `flush`, the checkpoint,
    /// rotation, compaction, and FixedStore write-triggered sync still run.
    /// Bitcask write buffers then reach the page cache only when full
    /// (`write_buffer_size` per shard), which is a byte bound, not a time bound.
    /// Default 1 s.
    pub flush: Option<Duration>,
    /// TTL sweep tick; `None` disables the sweep worker entirely, leaving
    /// expiry to explicit [`Db::sweep_expired_now`] calls. Default 60 s.
    ///
    /// A tick that finds more work than `ttl_budget` allows backs off to 1 s
    /// instead of the configured interval until the backlog drains.
    pub ttl_sweep: Option<Duration>,
    /// Maximum rows deleted per collection per sweep pass. Default 4096.
    ///
    /// This is what bounds the blast radius of a bad policy: a mistake that
    /// marks everything due deletes at most this many rows per pass, and the
    /// 1 s catch-up backoff caps how fast passes can follow one another.
    pub ttl_budget: usize,
}

impl Default for WorkerIntervals {
    fn default() -> Self {
        Self {
            compaction: Some(DEFAULT_COMPACTION_INTERVAL),
            flush: Some(DEFAULT_FLUSH_INTERVAL),
            ttl_sweep: Some(DEFAULT_TTL_SWEEP_INTERVAL),
            ttl_budget: DEFAULT_TTL_BUDGET,
        }
    }
}

/// Names reserved for internal `Db` state. A user collection may not take one of
/// these: `__seq` is the `SeqGen` directory, `db.info`/`db.user` are the metadata
/// sidecars, `txlog` is the transaction log — a same-named collection would
/// collide with them under `tree_path`.
const RESERVED_COLLECTION_NAMES: [&str; 5] = [
    "__seq",
    "db.info",
    "db.user",
    crate::lock::LOCK_FILE_NAME,
    "txlog",
];

/// Sentinel `typ_hash` written on intermediate multi-step migration steps:
/// binary cannot know intermediate version type hash. V-005 skips comparison
/// for this value. Reserved; collision with real xxh3 hash negligibly likely
/// and only skips one check, not data corruption.
pub const TYP_HASH_MIGRATING: u64 = 0;

#[cfg(test)]
mod format_migration_metadata_tests {
    use super::CollectionInfo;

    #[test]
    fn collection_info_new_fields_default_when_absent_from_stored_json() {
        let info: CollectionInfo = serde_json::from_str(
            r#"{"version":1,"typ_hash":2,"seq":3,"projections_dirty":true,"counters_dirty":true}"#,
        )
        .unwrap();
        assert_eq!(info.gsn_high, 0);
        assert!(info.shard_gsn_high.is_empty());
        assert!(!info.unique_dirty);
    }
}

/// A transaction record still present in the log.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingTx {
    pub tx_id: u64,
    pub collections: Vec<String>,
}

/// Manager for multiple named trees with persistent metadata and sequence generation.
pub struct Db {
    pub path: PathBuf,
    /// Private on purpose: `Persist` is `Clone` over an `Arc`, so a public field
    /// would let a caller outlive the `Db` with a live handle and write
    /// `db.info` after the root lease is released. Reach it through
    /// [`db_info`](Self::db_info) / [`db_info_update`](Self::db_info_update),
    /// which borrow `&self` and cannot escape.
    db_info: Persist<DbInfo>,
    /// Separate `Arc` over the same `db.info` handle, held only so
    /// [`counter_dirty_weak`](Self::counter_dirty_weak) can hand out a `Weak`
    /// that tracks *this `Db`'s* lifetime specifically — `Persist` is already
    /// `Clone`-over-`Arc` internally, but that inner `Arc<Inner<DbInfo>>` isn't
    /// something we can downgrade from outside `armour_core::persist`. Nothing
    /// else clones this field, so its strong count is exactly the `Db`'s own
    /// lifetime. The only caller of `CounterDirty::mark()` is `open_managed`,
    /// which holds `&Db` across it, so the upgrade always succeeds there — a
    /// dead `Weak` is an internal error, not a no-op (see `counter_dirty.rs`).
    db_info_arc: Arc<Persist<DbInfo>>,
    user_meta: Persist<UserMeta>,
    pub seq: Arc<SeqGen>,
    pub(crate) shutdown: ShutdownSignal,
    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    queue_wake_service: Arc<QueueWakeService>,
    pub(crate) collections: Arc<Mutex<Vec<Arc<dyn Collection>>>>,
    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    queue_retentions: Arc<Mutex<Vec<Arc<dyn QueueRetentionPass>>>>,
    pub(crate) inspection_handlers: Arc<Mutex<BTreeMap<&'static str, Arc<dyn InspectionHandler>>>>,
    /// P1-5: names of collections whose open is currently in flight. A name is
    /// inserted here — atomically checked against the live `collections` and
    /// other in-flight reservations under this lock — *before* the expensive
    /// open/recovery/migration, and removed only once the collection is published
    /// into `collections` and `inspection_handlers` (and RPC `handlers` when
    /// enabled). This closes the TOCTOU window in which two concurrent `open_*`
    /// calls for the same name could both pass the uniqueness check and build
    /// aliasing indexes over one shard set.
    reserved_names: Mutex<std::collections::HashSet<&'static str>>,
    #[cfg(feature = "rpc")]
    pub(crate) handlers: Arc<Mutex<Vec<(u64, Arc<dyn super::handler::RpcHandler>)>>>,
    #[cfg(feature = "rpc")]
    pub(crate) rpc_handles: Mutex<Vec<super::rpc::RpcHandle>>,
    compactor: Option<Compactor>,
    flusher: Option<Flusher>,
    sweeper: Option<Sweeper>,
    ttl_budget: usize,
    #[cfg(feature = "fjall")]
    lsm_db: Mutex<Option<(fjall::Database, crate::lsm::LsmDbConfig)>>,
    /// The only strong managed-LSM lifetime owner. Mutation guards upgrade its
    /// Weak reference and retain it through their fjall operation.
    #[cfg(feature = "fjall")]
    #[allow(dead_code)]
    managed_lsm_lifetime: Arc<ManagedLsmLifetime>,
    /// One clean-state marker shared by every LSM tree opened through this Db.
    #[cfg(feature = "fjall")]
    lsm_clean_marker: Arc<crate::lsm::ManagedLsmCleanMarker>,
    /// `true` iff the previous session ended with a successful finalization.
    /// Used by `open_lsm_tree` to decide whether to trust the persisted seq
    /// counter or re-seed it from `ks.len()`.
    #[cfg(feature = "fjall")]
    opened_clean: bool,
    tx_config: crate::txn::TxConfig,
    txlog: Arc<crate::txlog::TxLog>,
    /// Collection names whose open-time tx replay has finished this session.
    /// Used to mark a pending record applied once every participant it names
    /// has been opened (and therefore had its ops applied or skipped).
    tx_replay_done: Mutex<std::collections::HashSet<&'static str>>,
    /// Lock round for projection targets (`1`); sources stay at `0`.
    tx_ranks: Arc<Mutex<std::collections::HashMap<&'static str, u8>>>,
    /// `(namespace, owner tag) -> source name` for every claiming source
    /// opened on this `Db` (T10 review F5) — the managed-open twin of
    /// `SchemaRegistry::register_owner_tag`, populated by
    /// [`check_claim_owner_tags`](Self::check_claim_owner_tags) so a source
    /// declared through the raw `Projected::and_unique` path (no
    /// `SchemaRegistry` involved) still has its owner tag checked for a
    /// collision against every other claiming source opened on this same
    /// `Db`. See that method's doc for why a `Db`-lifetime map is the right
    /// home for this, rather than a registry-lifetime one.
    claim_owner_tags: Mutex<std::collections::HashMap<(&'static str, u64), &'static str>>,
    /// Counter **targets** whose `counters_dirty` flag an open-time recount
    /// consumed this session (T9). A target's flag is set by its own migration
    /// and has no in-memory marker behind it, so `close`/`clean_shutdown`
    /// cannot tell it apart from evidence nobody has acted on yet and refuse
    /// to clear it — which left a migrated target re-triggering a full recount
    /// at every open, forever. Recording the consumption here gives the close
    /// exactly the proof it was missing.
    ///
    /// Deferred to the close rather than cleared at the recount because
    /// several sources may feed one target: every one of them must still see
    /// the flag when it opens later in the same session.
    counter_targets_consumed: Mutex<std::collections::HashSet<&'static str>>,
    /// Exclusive lease on the root directory.
    ///
    /// Declared last on purpose: Rust drops fields in declaration order, so the
    /// lock is released only after `db_info`/`db.user` are closed, after
    /// `Db::drop` → `stop_workers()` has joined the compactor and flusher, and after
    /// the `fjall::Database` held in `lsm_db` is dropped (so its keyspace files
    /// are flushed before the next process takes the lease). Moving this field
    /// up would open a window where the next process holds the lock while this
    /// one is still writing.
    _lock: Arc<std::fs::File>,
    /// Private transaction services retained by queues that can outlive `Db`.
    pub(crate) tx_runtime: Arc<crate::txn::TxRuntime>,
}

#[derive(Clone, Copy)]
enum FinalizeMode {
    Close,
    CleanShutdown,
}

fn note_first_error(first: &mut Option<DbError>, error: DbError) {
    if first.is_none() {
        *first = Some(error);
    }
}

/// RAII reservation of a collection name held for the duration of an in-flight
/// open (P1-5). Dropping it — whether the open succeeded (the collection is by
/// then in `Db::collections`) or failed — frees the name from `reserved_names`.
/// The name is therefore continuously owned by the reservation and then the
/// live list and inspection registry (and RPC handlers when enabled), with no
/// gap a concurrent open could slip through.
struct NameReservation<'a> {
    db: &'a Db,
    name: &'static str,
}

struct StagedManaged<'a, S, W> {
    meta: TreeMeta,
    collection: Arc<S>,
    wire: W,
    _name_guard: NameReservation<'a>,
}

struct StagedAdapter<'a> {
    name: &'static str,
    collection: Arc<dyn Collection>,
    inspection: Arc<dyn InspectionHandler>,
    #[cfg(feature = "rpc")]
    rpc: Option<(u64, Arc<dyn super::handler::RpcHandler>)>,
    _name_guard: NameReservation<'a>,
}

/// Publication surface for a managed collection.
///
/// Internal collections still participate in local collection management and
/// read-only inspection, but do not receive a generic RPC adapter. Engine-owned
/// structures use this when arbitrary remote mutation would violate their
/// higher-level invariants.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ManagedCollectionAccess {
    Public,
    Internal,
}

impl Drop for NameReservation<'_> {
    fn drop(&mut self) {
        lock(&self.db.reserved_names).remove(self.name);
    }
}

/// Run one collection TTL sweep, catching extractor panics so one bad policy
/// cannot stop the background worker.
fn ttl_sweep_collection(
    c: &Arc<dyn Collection>,
    now_ms: u64,
    budget: usize,
) -> DbResult<SweepOutcome> {
    let name = c.name();
    match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        c.sweep_expired(now_ms, budget)
    })) {
        Ok(Ok(out)) => {
            log_sweep_failures(Some(name), &out);
            Ok(out)
        }
        Ok(Err(e)) => Err(e),
        Err(_) => {
            tracing::error!(
                collection = name,
                "ttl sweep panicked in deadline extractor; skipping collection for this pass"
            );
            Ok(SweepOutcome::default())
        }
    }
}

impl Db {
    pub(crate) fn queue_wake_service(&self) -> Arc<QueueWakeService> {
        Arc::clone(&self.queue_wake_service)
    }

    /// Open an engine-owned durable queue.
    pub fn open_queue<N>(&self, config: QueueConfig) -> DbResult<Arc<Queue<N>>>
    where
        N: QueueSpec,
        N::Mode: crate::queue::QueueOpen<N>,
    {
        <N::Mode as crate::queue::QueueOpen<N>>::open(self, config.build()?)
    }

    /// Full control over background workers and transaction configuration.
    /// The `open*` constructors below are presets over this builder.
    pub fn builder(path: impl AsRef<std::path::Path>) -> DbBuilder {
        DbBuilder {
            path: path.as_ref().to_path_buf(),
            workers: WorkerIntervals::default(),
            sync_persist: true,
            tx_config: crate::txn::TxConfig::default(),
        }
    }

    /// Open with default background workers: compaction every 60 s and flush
    /// every 1 s. Both stop together on [`shutdown`](Self::shutdown) / drop.
    /// Use [`open_with_intervals`](Self::open_with_intervals) to tune or disable
    /// either.
    ///
    /// A preset over [`Db::builder`].
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[doc = include_str!("../../docs/exclusive-access-root.md")]
    pub fn open(path: impl AsRef<std::path::Path>) -> DbResult<Self> {
        Self::builder(path).open()
    }

    /// Open with explicit background worker intervals and TTL sweep budget.
    ///
    /// A preset over [`Db::builder`].
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[doc = include_str!("../../docs/exclusive-access-root.md")]
    pub fn open_with_workers(
        path: impl AsRef<std::path::Path>,
        workers: WorkerIntervals,
    ) -> DbResult<Self> {
        Self::builder(path).workers(workers).open()
    }

    /// Open for tests: no compaction or flush worker, and `db.info`/`db.user`
    /// are persisted asynchronously. Use [`open`](Self::open) in production.
    ///
    /// A preset over [`Db::builder`].
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[doc = include_str!("../../docs/exclusive-access-root.md")]
    pub fn open_test(path: impl AsRef<std::path::Path>) -> DbResult<Self> {
        Self::builder(path).for_tests().open()
    }

    /// Open with a custom transaction configuration. The existing constructors
    /// use [`crate::txn::TxConfig::default`].
    ///
    /// A preset over [`Db::builder`].
    pub fn open_with_tx_config(
        path: impl AsRef<std::path::Path>,
        tx_config: crate::txn::TxConfig,
    ) -> DbResult<Self> {
        Self::builder(path).tx_config(tx_config).open()
    }

    /// Open with a custom compaction interval (`None` disables compaction); the
    /// flush worker keeps its 1 s default. Use
    /// [`open_with_intervals`](Self::open_with_intervals) to also control flush.
    ///
    /// A preset over [`Db::builder`].
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[doc = include_str!("../../docs/exclusive-access-root.md")]
    pub fn open_with_compaction(
        path: impl AsRef<std::path::Path>,
        compaction_interval: Option<Duration>,
    ) -> DbResult<Self> {
        Self::builder(path)
            .workers(WorkerIntervals {
                compaction: compaction_interval,
                ..WorkerIntervals::default()
            })
            .open()
    }

    /// Open with explicit compaction and flush intervals. `None` disables the
    /// corresponding background worker. The flush worker periodically runs each
    /// collection's [`Collection::periodic_flush`](super::collection::Collection::periodic_flush)
    /// on every registered collection. On Bitcask and FixedStore that is a
    /// dirty-shard fsync/fdatasync; on LSM it is the configured `PersistPolicy`.
    ///
    /// A preset over [`Db::builder`].
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[doc = include_str!("../../docs/exclusive-access-root.md")]
    pub fn open_with_intervals(
        path: impl AsRef<std::path::Path>,
        compaction_interval: Option<Duration>,
        flush_interval: Option<Duration>,
    ) -> DbResult<Self> {
        Self::builder(path)
            .workers(WorkerIntervals {
                compaction: compaction_interval,
                flush: flush_interval,
                ttl_sweep: Some(DEFAULT_TTL_SWEEP_INTERVAL),
                ttl_budget: DEFAULT_TTL_BUDGET,
            })
            .open()
    }

    fn open_with_options(
        path: impl AsRef<std::path::Path>,
        workers: WorkerIntervals,
        sync_persist: bool,
        tx_config: crate::txn::TxConfig,
    ) -> DbResult<Self> {
        let WorkerIntervals {
            compaction: compaction_interval,
            flush: flush_interval,
            ttl_sweep: ttl_sweep_interval,
            ttl_budget,
        } = workers;
        let path = path.as_ref().to_path_buf();
        std::fs::create_dir_all(&path).map_err(crate::DbError::Io)?;
        // Exclusive lease on the root, taken before any state is read: db.info
        // and db.user are plain JSON sidecars, so a second process here would
        // silently lose one side's updates. (`__seq` is itself a FixedMap and
        // is covered by its own collection lease.)
        let _lock = Arc::new(crate::lock::acquire(&path)?);
        let txlog = Arc::new(crate::txlog::TxLog::open(
            &path,
            crate::txlog::TxLogConfig {
                segment_bytes: tx_config.segment_bytes,
                id_window: 4096,
                #[cfg(feature = "encryption")]
                log_encryption_key: tx_config.log_encryption_key,
            },
        )?);
        let db_info: Persist<DbInfo> = if sync_persist {
            Persist::open(path.join("db.info"))?
        } else {
            Persist::open_no_sync(path.join("db.info"))?
        };
        // See the `db_info_arc` field doc: a separate `Arc` over the same
        // handle, so `counter_dirty_weak` can hand out a `Weak` tied to this
        // `Db`'s own lifetime.
        let db_info_arc = Arc::new(db_info.clone());
        let user_meta = if sync_persist {
            Persist::open(path.join("db.user"))?
        } else {
            Persist::open_no_sync(path.join("db.user"))?
        };
        let seq = SeqGen::open(path.join("__seq"))?;
        let collections: Arc<Mutex<Vec<Arc<dyn Collection>>>> = Arc::new(Mutex::new(Vec::new()));
        #[cfg(all(feature = "typed-tree", feature = "armour"))]
        let queue_retentions: Arc<Mutex<Vec<Arc<dyn QueueRetentionPass>>>> =
            Arc::new(Mutex::new(Vec::new()));
        let shutdown = ShutdownSignal::new();
        #[cfg(all(feature = "typed-tree", feature = "armour"))]
        let queue_wake_service = QueueWakeService::new(shutdown.clone());

        let compactor = compaction_interval.map(|interval| {
            let cols = collections.clone();
            Compactor::start_with_signal(
                move || {
                    let snapshot = lock(&cols).clone();
                    let mut total = 0;
                    for c in &snapshot {
                        total += c.compact()?;
                    }
                    Ok(total)
                },
                interval,
                shutdown.clone(),
            )
        });

        let flusher = flush_interval.map(|interval| {
            let cols = collections.clone();
            Flusher::start_with_signal(
                move || {
                    let snapshot = lock(&cols).clone();
                    crate::flush_pass::begin();
                    let started = Instant::now();
                    // Flush every collection even if one fails: a flush is a
                    // durability action, so an unhealthy collection must not
                    // starve the others. Return the first error — the Flusher
                    // logs it and retries every collection next tick.
                    let mut first_err = None;
                    for c in &snapshot {
                        if let Err(e) = c.periodic_flush()
                            && first_err.is_none()
                        {
                            first_err = Some(e);
                        }
                    }
                    crate::flush_pass::finish(interval, started.elapsed());
                    match first_err {
                        Some(e) => Err(e),
                        None => Ok(()),
                    }
                },
                interval,
                shutdown.clone(),
            )
        });

        let sweeper = ttl_sweep_interval.map(|interval| {
            let cols = collections.clone();
            #[cfg(all(feature = "typed-tree", feature = "armour"))]
            let retentions = queue_retentions.clone();
            Sweeper::start_with_signal(
                move |budget| {
                    let now_ms = std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .map_err(|_| DbError::Internal("system clock before unix epoch"))?
                        .as_millis() as u64;
                    let snapshot = lock(&cols).clone();
                    let mut deleted = 0;
                    let mut more = false;
                    let mut first_err = None;
                    for c in &snapshot {
                        match ttl_sweep_collection(c, now_ms, budget) {
                            Ok(out) => {
                                deleted += out.deleted;
                                if out.more {
                                    more = true;
                                }
                            }
                            Err(e) if first_err.is_none() => first_err = Some(e),
                            Err(_) => {}
                        }
                    }
                    #[cfg(all(feature = "typed-tree", feature = "armour"))]
                    for retention in &*lock(&retentions) {
                        if let Err(e) = retention.run_retention()
                            && first_err.is_none()
                        {
                            first_err = Some(e);
                        }
                    }
                    match first_err {
                        Some(e) => Err(e),
                        None => Ok(crate::sweeper::SweepPassResult { deleted, more }),
                    }
                },
                interval,
                ttl_budget,
                shutdown.clone(),
            )
        });

        // Clean-close marker: record whether the previous session ended cleanly,
        // then immediately mark this session as dirty. A crash before `close()`
        // leaves `clean=false`, telling `open_lsm_tree` to rescan counts.
        #[cfg(feature = "fjall")]
        let opened_clean = {
            let was_clean = db_info.cloned().clean;
            db_info.update(|i| i.clean = false)?;
            was_clean
        };
        #[cfg(feature = "fjall")]
        let managed_lsm_lifetime = Arc::new(ManagedLsmLifetime {
            db_info: db_info.clone(),
            _root_lease: Arc::clone(&_lock),
        });
        #[cfg(feature = "fjall")]
        let lsm_clean_marker = Arc::new(crate::lsm::ManagedLsmCleanMarker::new(Arc::downgrade(
            &managed_lsm_lifetime,
        )));

        let tx_ranks = Arc::new(Mutex::new(std::collections::HashMap::new()));
        let tx_runtime = Arc::new(crate::txn::TxRuntime {
            config: tx_config,
            log: Arc::clone(&txlog),
            collections: Arc::clone(&collections),
            ranks: Arc::clone(&tx_ranks),
            post_log_queue_producer_permits: Mutex::new(Vec::new()),
            _root_lease: Arc::clone(&_lock),
        });
        Ok(Self {
            path,
            db_info,
            db_info_arc,
            user_meta,
            seq,
            shutdown,
            #[cfg(all(feature = "typed-tree", feature = "armour"))]
            queue_wake_service,
            collections,
            #[cfg(all(feature = "typed-tree", feature = "armour"))]
            queue_retentions,
            inspection_handlers: Arc::new(Mutex::new(BTreeMap::new())),
            reserved_names: Mutex::new(std::collections::HashSet::new()),
            counter_targets_consumed: Mutex::new(std::collections::HashSet::new()),
            #[cfg(feature = "rpc")]
            handlers: Arc::new(Mutex::new(Vec::new())),
            #[cfg(feature = "rpc")]
            rpc_handles: Mutex::new(Vec::new()),
            compactor,
            flusher,
            sweeper,
            ttl_budget,
            #[cfg(feature = "fjall")]
            lsm_db: Mutex::new(None),
            #[cfg(feature = "fjall")]
            managed_lsm_lifetime,
            #[cfg(feature = "fjall")]
            lsm_clean_marker,
            #[cfg(feature = "fjall")]
            opened_clean,
            tx_config,
            txlog,
            tx_replay_done: Mutex::new(std::collections::HashSet::new()),
            tx_ranks,
            claim_owner_tags: Mutex::new(std::collections::HashMap::new()),
            _lock,
            tx_runtime,
        })
    }

    /// Start a transaction. Operations are buffered; nothing touches a
    /// collection until [`crate::txn::Tx::commit`] has made the log record
    /// durable.
    ///
    /// Crash-atomic across collections — **not** an isolation mechanism: a
    /// concurrent reader can see one participant's write before another's. Only
    /// the six managed families opened through this same `Db` may take part;
    /// Bitcask and FixedStore backends may be mixed in one transaction.
    /// An encrypted collection needs [`TxConfig::log_encryption_key`]; without
    /// it `commit()` refuses, so a value is never logged in the clear. See
    /// `armdb/docs/transactions.md` for the full contract, the limits, and the
    /// reentrancy rules for `upsert`/`guard` closures.
    ///
    /// [`TxConfig::log_encryption_key`]: crate::txn::TxConfig::log_encryption_key
    pub fn tx(&self) -> crate::txn::Tx<'_> {
        crate::txn::Tx {
            context: crate::txn::TxContext::Borrowed(self),
            limits: crate::txn::TxLimits::from(&self.tx_config),
            ops: Vec::new(),
            claim_ops: Vec::new(),
            durable: self.tx_config.durable_by_default,
            after_commit: Vec::new(),
            queue_producer_permits: Vec::new(),
            queue_depth_admissions: crate::queue::QueueDepthAdmissions::default(),
        }
    }

    /// Transaction limits and durability defaults for [`crate::txn::Tx`].
    pub fn tx_config(&self) -> &crate::txn::TxConfig {
        &self.tx_config
    }

    /// Suppress the apply phase after a successful log append (tests only): the
    /// on-disk state is then exactly what a crash between the commit marker and
    /// the first write leaves behind.
    #[doc(hidden)]
    #[cfg(any(test, feature = "test-faults"))]
    pub fn tx_debug_skip_apply(on: bool) {
        crate::txn::SKIP_APPLY.store(on, std::sync::atomic::Ordering::SeqCst);
    }

    /// Suppress post-migration projection reconciliation (tests only).
    #[doc(hidden)]
    #[cfg(any(test, feature = "test-faults"))]
    pub fn tx_debug_skip_projection_reconcile(on: bool) {
        crate::txn::SKIP_PROJECTION_RECONCILE.store(on, std::sync::atomic::Ordering::SeqCst);
    }

    /// Make every value read inside a Var reconcile scan fail (tests only).
    ///
    /// The scan drives `delete_orphans`, so a row it cannot read must abort it
    /// rather than shrink the expected set — otherwise that row's live index
    /// entries are deleted and the dirty flag is cleared, losing them for good.
    #[doc(hidden)]
    #[cfg(any(test, feature = "test-faults"))]
    pub fn debug_fail_reconcile_value_read(on: bool) {
        crate::projection::FAIL_RECONCILE_VALUE_READ.store(on, std::sync::atomic::Ordering::SeqCst);
    }

    pub(crate) fn txlog(&self) -> &crate::txlog::TxLog {
        &self.txlog
    }

    /// Position of a participant in the published collection list — the
    /// canonical `collection_id`. Identity is by allocation, so a handle from
    /// another `Db` (or one never published) is rejected.
    pub(crate) fn tx_collection_id(
        &self,
        participant: &dyn crate::txn::participant::TxParticipant,
    ) -> DbResult<u32> {
        let needle = participant as *const dyn crate::txn::participant::TxParticipant as *const ();
        let collections = lock(&self.collections);
        collections
            .iter()
            .position(|c| std::sync::Arc::as_ptr(c) as *const () == needle)
            .map(|i| i as u32)
            .ok_or(DbError::Client("collection was not opened through this Db"))
    }

    /// Lock round for a published participant (`0` = source, `1` = index target).
    pub(crate) fn tx_rank_of(
        &self,
        participant: &dyn crate::txn::participant::TxParticipant,
    ) -> u8 {
        lock(&self.tx_ranks)
            .get(participant.name())
            .copied()
            .unwrap_or(0)
    }

    /// Called when opening a projecting source: verify targets are on this `Db`
    /// and mark them rank `1` for the two-round commit protocol.
    pub(crate) fn tx_register_projection_targets(
        &self,
        targets: &crate::projection::IndexTargets,
    ) -> DbResult<()> {
        for handle in targets.erased_targets() {
            let participant =
                targets
                    .resolve_tx_participant(handle.name)
                    .ok_or(DbError::Internal(
                        "declared index target missing participant",
                    ))?;
            self.tx_collection_id(participant.as_ref()).map_err(|_| {
                DbError::Client("secondary index target is not published on this Db")
            })?;
            lock(&self.tx_ranks).insert(handle.name, 1);
        }
        Ok(())
    }

    /// Counter twin of [`tx_register_projection_targets`](Self::tx_register_projection_targets)
    /// (T9 Task 9): a counter target is written under a lock the commit takes
    /// in its **second** round, so it must be rank `1` for the same reason an
    /// index target is.
    ///
    /// Without this, a transaction that also writes the target directly would
    /// plan that write at rank 0 and never merge the counter row's shard into
    /// the target's lock set — the derived write would then hit an unlocked
    /// shard.
    ///
    /// Rank 1 has a cost, and it is deliberate: **a rank-1 participant does not
    /// run its own projections** (`txn::projection_commit::eval_rank1` collects
    /// no projection inputs — a target's own derived writes would need a third
    /// lock round). So marking a collection rank 1 makes it illegal to write
    /// directly inside a transaction *if it is itself a projecting or counting
    /// source*; `plan` rejects that topology with
    /// [`DbError::TxTargetIsAlsoASource`] rather than let its indexes/counters
    /// drift silently. A plain target — the ordinary case — is unaffected.
    pub(crate) fn tx_register_counter_targets(
        &self,
        targets: &crate::counter::CounterTargets,
    ) -> DbResult<()> {
        for member in targets.all() {
            let participant = member.as_tx_participant();
            self.tx_collection_id(participant.as_ref())
                .map_err(|_| DbError::Client("counter target is not published on this Db"))?;
            lock(&self.tx_ranks).insert(member.collection_name(), 1);
        }
        Ok(())
    }

    /// Unique-namespace twin of
    /// [`tx_register_projection_targets`](Self::tx_register_projection_targets)/
    /// [`tx_register_counter_targets`](Self::tx_register_counter_targets)
    /// (T10 Task 6): a namespace row is written under a lock the commit takes
    /// in its **second** round, so every declared namespace must be rank `1`
    /// for the same reason an index or counter target is — and, per the same
    /// [`DbError::TxTargetIsAlsoASource`] rule, a unique namespace must not
    /// itself be a projecting/counting/claiming source.
    pub(crate) fn tx_register_claim_targets(&self, targets: &UniqueTargets) -> DbResult<()> {
        for target in targets.all() {
            let participant = target.as_tx_participant();
            self.tx_collection_id(participant.as_ref()).map_err(|_| {
                DbError::Client("unique namespace target is not published on this Db")
            })?;
            lock(&self.tx_ranks).insert(target.collection_name(), 1);
        }
        Ok(())
    }

    pub(crate) fn tx_register_stream_targets(&self, targets: &crate::QueueTargets) -> DbResult<()> {
        for participant in targets.tx_participants() {
            self.tx_collection_id(participant.as_ref()).map_err(|_| {
                DbError::Client("change stream queue target is not published on this Db")
            })?;
            lock(&self.tx_ranks).insert(participant.name(), 1);
        }
        Ok(())
    }

    /// Validate a declared change stream before the source is staged. Queue
    /// destinations must already be managed collections of this exact Db;
    /// variable queues contribute both their header and payload participants.
    fn check_stream_open<T, H>(&self, hook: &H) -> DbResult<()>
    where
        T: CollectionMeta,
        H: TypedWriteHook<T::SelfId, T>,
    {
        if !H::EMITS {
            return Ok(());
        }
        if T::TTL.is_some() {
            return Err(DbError::Client("change stream source cannot use TTL"));
        }
        let targets = hook
            .stream_targets()
            .ok_or(DbError::Internal("stream carrier missing targets"))?;
        targets.check_managed(self).map_err(|_| {
            DbError::Client("change stream queue target is not published on this Db")
        })?;
        Ok(())
    }

    /// Record `T`'s owner tag (`unique::collection_tag(T::NAME)`) against
    /// every namespace it claims into, on this `Db` (T10 review F5) — the
    /// managed-open twin of `SchemaRegistry::register_owner_tag`. Called from
    /// every managed opener beside [`crate::unique::check_no_ttl_claims`] and
    /// [`crate::unique::check_claim_key_width`], so the same collision the
    /// registry path already rejects at declaration time is also rejected for
    /// a source that reaches a managed open through the raw
    /// `Projected::and_unique` path with no `SchemaRegistry` involved.
    ///
    /// A `Db`-lifetime map (not a registry-lifetime one) is the right home for
    /// this: two sources without a shared registry can still collide if — and
    /// only if — they are opened on the same `Db`, so the check has to live
    /// exactly as long as the `Db` does. Idempotent for the same
    /// `(namespace, source)` pair, mirroring `register_owner_tag`: opening the
    /// same collection twice (e.g. a test reopening a `Db`) is not itself a
    /// collision.
    fn check_claim_owner_tags<T, H>(&self, hook: &H) -> DbResult<()>
    where
        T: CollectionMeta,
        H: TypedWriteHook<T::SelfId, T>,
    {
        let Some(targets) = hook.claim_targets() else {
            return Ok(());
        };
        let tag = crate::unique::collection_tag(T::NAME);
        self.register_claim_owner_tag(tag, T::NAME, targets)
    }

    /// The tag-parameterized core of [`check_claim_owner_tags`](Self::check_claim_owner_tags),
    /// split out so a unit test can drive a collision with a hand-picked
    /// `tag` — mirroring why `SchemaRegistry::register_owner_tag`'s own tests
    /// call it directly with a synthetic tag rather than through `finish()`:
    /// a genuine 64-bit FNV-1a collision cannot be produced from two real
    /// collection names, so testing only the public, name-driven path would
    /// test nothing.
    fn register_claim_owner_tag(
        &self,
        tag: u64,
        name: &'static str,
        targets: &UniqueTargets,
    ) -> DbResult<()> {
        let mut map = lock(&self.claim_owner_tags);
        for target in targets.all() {
            let namespace = target.collection_name();
            match map.entry((namespace, tag)) {
                std::collections::hash_map::Entry::Occupied(e) => {
                    if *e.get() != name {
                        return Err(DbError::ClaimOwnerTagCollision {
                            namespace,
                            a: e.get(),
                            b: name,
                        });
                    }
                }
                std::collections::hash_map::Entry::Vacant(e) => {
                    e.insert(name);
                }
            }
        }
        Ok(())
    }

    /// Pending transactions recovered from the log and not yet fully replayed.
    pub fn pending_transactions(&self) -> Vec<PendingTx> {
        self.txlog
            .pending()
            .into_iter()
            .map(|r| PendingTx {
                tx_id: r.tx_id,
                collections: r.ops.iter().map(|o| o.name.clone()).collect(),
            })
            .collect()
    }

    /// Record that open-time tx replay finished for `name`, then mark any
    /// pending records whose every participant has now been opened this session.
    pub(crate) fn note_tx_replay_done(&self, name: &'static str) {
        lock(&self.tx_replay_done).insert(name);
        self.settle_pending_transactions();
    }

    /// Strategy A: a recovered record is fully settled once every collection it
    /// names has completed open-time replay in this process (each open either
    /// applied or skipped that collection's ops via the GSN watermark rule).
    fn settle_pending_transactions(&self) {
        let done = lock(&self.tx_replay_done);
        for record in self.txlog.pending() {
            // Only records recovered from disk. `tx_replay_done` accumulates for
            // the whole session, so after the first transaction every one of its
            // participants is in it — without this gate, a record left pending by
            // a *partial apply in this session* would be marked applied by the
            // next collection open and retired by the next cut, losing the half
            // that never landed. Records this process appended settle through
            // their own commit.
            if !self.txlog.is_recovered(record.tx_id) {
                continue;
            }
            let names: std::collections::HashSet<&str> =
                record.ops.iter().map(|o| o.name.as_str()).collect();
            if names.is_empty() {
                continue;
            }
            if names.iter().all(|n| done.iter().any(|d| d == n)) {
                self.txlog.mark_applied(record.tx_id);
            }
        }
    }

    /// Run a transaction-log checkpoint now (durable-flush, then retire applied
    /// records at or below the cut). Called from [`close`](Self::close).
    pub(crate) fn checkpoint_now(&self) -> DbResult<()> {
        let _ = crate::txn::checkpoint::run(self)?;
        Ok(())
    }

    /// Current stored database version.
    pub fn version(&self) -> u32 {
        self.db_info.cloned().version
    }

    /// Update version via callback. Returns the new version.
    pub fn set_version(&self, f: impl FnOnce(u32) -> u32) -> DbResult<u32> {
        let mut new_version = 0;
        self.db_info.update(|info| {
            new_version = f(info.version);
            info.version = new_version;
        })?;
        Ok(new_version)
    }

    /// Snapshot of the stored database info.
    pub fn db_info(&self) -> DbInfo {
        self.db_info.cloned()
    }

    /// Whether `name`'s counter deltas may have been lost this session (T9).
    /// Persisted when `name` is opened as a counted source; cleared only by a
    /// fully successful `close`/`clean_shutdown`. A name with no stored entry
    /// reads as clean (`false`).
    #[doc(hidden)]
    pub fn debug_counters_dirty(&self, name: &str) -> bool {
        self.db_info
            .cloned()
            .collections
            .get(name)
            .is_some_and(|c| c.counters_dirty)
    }

    /// Whether a format migration may have left `name`'s unique-namespace
    /// claims stale. The marker is durable and can only be retired by the
    /// schema registry's complete namespace repair pass.
    pub fn unique_dirty(&self, name: &str) -> bool {
        self.db_info
            .cloned()
            .collections
            .get(name)
            .is_some_and(|c| c.unique_dirty)
    }

    /// A `Weak` handle to `db.info`, tied to this `Db`'s own lifetime — see the
    /// `db_info_arc` field doc. Bound into a collection's carrier at managed
    /// open via `ManagedOpen::bind_counter_dirty`.
    pub(crate) fn counter_dirty_weak(&self) -> Weak<Persist<DbInfo>> {
        Arc::downgrade(&self.db_info_arc)
    }

    /// Edit `db.info` in place.
    ///
    /// White-box hook for migration tests; not part of the supported API. Use
    /// [`set_version`](Self::set_version) and the collection openers instead —
    /// they maintain the invariants this bypasses.
    #[doc(hidden)]
    pub fn db_info_update(&self, f: impl FnOnce(&mut DbInfo)) -> DbResult<()> {
        self.db_info.update(f).map_err(Into::into)
    }

    /// Read database metadata, deserializing into M.
    /// Returns None if not set or deserialization fails.
    pub fn metadata<M: DeserializeOwned>(&self) -> Option<M> {
        let um = self.user_meta.cloned();
        um.db.and_then(|v| serde_json::from_value(v).ok())
    }

    /// Update database metadata via callback. Atomic (single lock).
    /// Returns the new value.
    pub fn set_metadata<M: Serialize + DeserializeOwned>(
        &self,
        f: impl FnOnce(Option<M>) -> M,
    ) -> DbResult<M> {
        let mut result = None;
        // A failing `Serialize` (e.g. a map with non-string keys) must not panic
        // under the Persist lock — capture the error and surface it as a
        // `DbError` instead. On serialization failure the stored metadata is left
        // untouched. Atomicity is preserved: the current value is read and the
        // new value written under the same lock. Serializing under the lock is
        // negligible next to the fsync `update` already performs while holding it.
        let mut ser_err = None;
        self.user_meta.update(|um| {
            let current = um
                .db
                .as_ref()
                .and_then(|v| serde_json::from_value(v.clone()).ok());
            let new_val = f(current);
            match serde_json::to_value(&new_val) {
                Ok(v) => {
                    um.db = Some(v);
                    result = Some(new_val);
                }
                Err(e) => ser_err = Some(e),
            }
        })?;
        if let Some(e) = ser_err {
            return Err(DbError::FormatMismatch(e.to_string()));
        }
        Ok(result.expect("update executed"))
    }

    /// Read collection metadata by name.
    /// Returns None if not set or deserialization fails.
    pub fn collection_metadata<M: DeserializeOwned>(&self, name: &str) -> Option<M> {
        let um = self.user_meta.cloned();
        um.collections
            .get(name)
            .and_then(|v| serde_json::from_value(v.clone()).ok())
    }

    /// Update collection metadata via callback. Atomic (single lock).
    /// Returns the new value.
    pub fn set_collection_metadata<M: Serialize + DeserializeOwned>(
        &self,
        name: &str,
        f: impl FnOnce(Option<M>) -> M,
    ) -> DbResult<M> {
        let mut result = None;
        // See `set_metadata`: capture a failing `Serialize` instead of panicking
        // under the Persist lock; leave the stored value untouched on failure.
        let mut ser_err = None;
        self.user_meta.update(|um| {
            let current = um
                .collections
                .get(name)
                .and_then(|v| serde_json::from_value(v.clone()).ok());
            let new_val = f(current);
            match serde_json::to_value(&new_val) {
                Ok(v) => {
                    um.collections.insert(name.to_owned(), v);
                    result = Some(new_val);
                }
                Err(e) => ser_err = Some(e),
            }
        })?;
        if let Some(e) = ser_err {
            return Err(DbError::FormatMismatch(e.to_string()));
        }
        Ok(result.expect("update executed"))
    }

    pub fn tree_path(&self, name: &str) -> PathBuf {
        self.path.join(name)
    }

    /// Run only the physical format prefix for a Bitcask collection.
    /// Value migrations, replay, and projection reconciliation are deferred to
    /// the next managed open.
    ///
    /// Every non-empty explicit rewrite conservatively sets `unique_dirty`
    /// because this API has neither a hook nor a schema registry with which to
    /// prove that the source declares no unique claims. Only registry-aware
    /// schema repair clears that marker; builds without that repair path should
    /// prefer automatic managed migration for sources known to be claim-free.
    pub fn migrate_format<T: CollectionMeta>(
        &self,
        config: &Config,
    ) -> DbResult<crate::FormatMigrationReport> {
        config.validate()?;
        let _reservation = self.reserve_name(T::NAME)?;
        let descriptor = StorageDescriptor::Bitcask {
            config,
            key_len: std::mem::size_of::<T::SelfId>(),
            claims: true,
        };
        descriptor.validate()?;
        descriptor.validate_persisted(&self.tree_path(T::NAME))?;
        let stored = self.stored_info(T::NAME);
        let installed_version = stored.as_ref().map_or(T::VERSION, |info| info.version);
        let Some(plan) = build_format_only_plan::<T>(
            T::NAME,
            stored.as_ref().map(|info| (info.version, info.typ_hash)),
        )?
        else {
            return Ok(crate::FormatMigrationReport::empty(
                T::NAME,
                installed_version,
            ));
        };
        let replay_plan = build_replay_plan::<T::SelfId, T>(
            T::NAME,
            stored.as_ref().map(|info| (info.version, info.typ_hash)),
            &[],
        )?;
        let meta = TreeMeta::of::<T>();
        if let Some(replay_plan) = replay_plan.as_ref() {
            self.guard_ambiguous_format_replay(T::NAME, T::TYPE.h(), replay_plan)?;
        }
        self.validate_format_backend_layout(&meta, &plan, descriptor)?;
        if let Some(shard_count) = descriptor.shard_count() {
            cleanup_stale_format_temps(&self.tree_path(T::NAME), shard_count)?;
        }
        self.set_projections_dirty(T::NAME, true)?;
        write_collection_marker(
            &self.tree_path(T::NAME),
            FormatMarker {
                from: plan.stored_version,
                to: plan.format_to_version,
            },
        )?;
        let mut outcome = self.migrate_format_backend(&meta, &plan, descriptor)?;
        outcome
            .report
            .warnings
            .push("explicit migration requires registry-aware unique repair".to_owned());
        self.commit_format_outcome(T::NAME, T::TYPE.h(), &plan, true, &outcome)?;
        Ok(outcome.report)
    }

    /// Run only the physical format prefix for a FixedStore collection.
    ///
    /// Every non-empty explicit rewrite conservatively sets `unique_dirty`
    /// because this API has neither a hook nor a schema registry with which to
    /// prove that the source declares no unique claims. Only registry-aware
    /// schema repair clears that marker; builds without that repair path should
    /// prefer automatic managed migration for sources known to be claim-free.
    pub fn migrate_format_fixed<T: CollectionMeta, const V: usize>(
        &self,
        config: &FixedConfig,
    ) -> DbResult<crate::FormatMigrationReport> {
        config.validate()?;
        let _reservation = self.reserve_name(T::NAME)?;
        let value_len =
            u16::try_from(V).map_err(|_| DbError::Config("fixed value length exceeds u16::MAX"))?;
        let descriptor = StorageDescriptor::Fixed {
            config,
            key_len: std::mem::size_of::<T::SelfId>(),
            value_len,
            claims: true,
        };
        descriptor.validate()?;
        descriptor.validate_persisted(&self.tree_path(T::NAME))?;
        let stored = self.stored_info(T::NAME);
        let installed_version = stored.as_ref().map_or(T::VERSION, |info| info.version);
        let Some(plan) = build_format_only_plan::<T>(
            T::NAME,
            stored.as_ref().map(|info| (info.version, info.typ_hash)),
        )?
        else {
            return Ok(crate::FormatMigrationReport::empty(
                T::NAME,
                installed_version,
            ));
        };
        let replay_plan = build_replay_plan::<T::SelfId, T>(
            T::NAME,
            stored.as_ref().map(|info| (info.version, info.typ_hash)),
            &[],
        )?;
        let meta = TreeMeta::of::<T>();
        if let Some(replay_plan) = replay_plan.as_ref() {
            self.guard_ambiguous_format_replay(T::NAME, T::TYPE.h(), replay_plan)?;
        }
        self.validate_format_backend_layout(&meta, &plan, descriptor)?;
        if let Some(shard_count) = descriptor.shard_count() {
            cleanup_stale_format_temps(&self.tree_path(T::NAME), shard_count)?;
        }
        self.set_projections_dirty(T::NAME, true)?;
        write_collection_marker(
            &self.tree_path(T::NAME),
            FormatMarker {
                from: plan.stored_version,
                to: plan.format_to_version,
            },
        )?;
        let mut outcome = self.migrate_format_backend(&meta, &plan, descriptor)?;
        outcome
            .report
            .warnings
            .push("explicit migration requires registry-aware unique repair".to_owned());
        self.commit_format_outcome(T::NAME, T::TYPE.h(), &plan, true, &outcome)?;
        Ok(outcome.report)
    }

    /// Generate the next sequential ID for a named collection.
    pub fn next_id(&self, name: &str) -> DbResult<u64> {
        self.seq.next_id(name)
    }

    /// Get the last saved collection length from db.info.
    pub fn collection_len<T: CollectionMeta>(&self) -> u64 {
        self.stored_info(T::NAME).map_or(0, |ci| ci.seq)
    }

    /// Names of all opened collections (for schema completeness checks).
    pub fn collection_names(&self) -> Vec<String> {
        lock(&self.collections)
            .iter()
            .map(|c| c.name().to_string())
            .collect()
    }

    /// Flush all collection write buffers, durably barrier dirty collection data,
    /// persist collection lengths to db.info, and flush the sequence generator.
    ///
    /// Does **not** write FixedStore sidecars or the clean-shutdown flag — use
    /// [`clean_shutdown`](Self::clean_shutdown) for Fixed fast-recovery.
    ///
    /// Synchronously disables TTL sweep on every collection before the final
    /// flush (waits for in-flight passes) so the background sweeper cannot emit
    /// tombstones during/after length persistence. The sweeper thread itself is
    /// not joined — that remains [`shutdown`](Self::shutdown).
    ///
    /// **TTL stays disabled after this returns.** This method tolerates writes
    /// afterwards (they downgrade the affected shard back to `Dirty` recovery),
    /// but expiry does not resume on its own: a `Db` used past a `close`
    /// checkpoint must re-enable it per collection with
    /// [`set_ttl_sweep(name, true)`](Self::set_ttl_sweep). The disable is logged
    /// at `info`.
    ///
    /// # Comparison with other shutdown paths
    ///
    /// | Method | Hints | Fixed sidecar/flag | Final fsync | Txlog checkpoint | Stops workers | Error |
    /// |---|---|---|---|---|---|---|
    /// | periodic_flush | no | no | dirty Bitcask/Fixed; LSM policy | no | no | logged |
    /// | close | writes | no | dirty shards only | yes | no | returned |
    /// | clean_shutdown | writes | yes | yes | yes | no | returned |
    /// | shutdown(&mut self) | via clean_shutdown | yes | yes | yes | yes | returned |
    /// | Drop | via close | no | dirty shards only | yes | yes | logged only |
    ///
    /// Hinted Bitcask collections re-barrier once per shard during [`Drop`] after
    /// `clean_shutdown`; clean no-hints shards issue zero fsyncs.
    ///
    /// This is a power-loss barrier for collection data, but it deliberately
    /// does not write FixedStore clean-recovery sidecars. Use
    /// [`clean_shutdown`](Self::clean_shutdown) when the next FixedStore open
    /// should use Clean recovery.
    pub fn close(&self) -> DbResult<()> {
        let collections = lock(&self.collections).clone();
        self.finalize(&collections, FinalizeMode::Close)
    }

    /// How many shards across every registered collection hold writes no
    /// durability barrier covers yet — the current power-loss window.
    ///
    /// This is the live counterpart of the `armdb.flush.dirty_shards` gauge the
    /// flush worker emits once per tick: it answers on demand, so it is also
    /// the only reading available when the worker is disabled (`flush: None`).
    ///
    /// Collections whose backend has no shard-level dirty state — `LsmTree`,
    /// which persists by its own `PersistPolicy` — report nothing and are not
    /// part of the sum; [`Collection::dirty_shards`] per collection
    /// distinguishes "clean" from "not reported".
    ///
    /// Sampled collection by collection without a consistent cut: concurrent
    /// writers and barriers move the answer while it is being counted.
    pub fn dirty_shards(&self) -> usize {
        lock(&self.collections)
            .iter()
            .filter_map(|c| c.dirty_shards())
            .sum()
    }

    /// Full clean shutdown for every registered collection.
    ///
    /// Heavier than [`close`](Self::close): writes Bitcask hint files + fsync,
    /// and for FixedStore collections writes the `fixed.versions` sidecar and sets
    /// the clean-shutdown flag so the next open uses Clean recovery.
    ///
    /// **Caller contract:** stop all application writers before calling. Does not
    /// stop background compactor/flusher workers — they are not write sources for
    /// FixedStore. The TTL sweeper *is* a write source; this method disables TTL
    /// sweep on every collection synchronously (waits for in-flight passes)
    /// before finalization, but does not join the worker thread — that remains
    /// [`shutdown`](Self::shutdown). In async contexts, call via `spawn_blocking`
    /// (tens of shards with fsync can take seconds).
    ///
    /// On partial failure: attempts all collections, logs `tracing::warn!` per
    /// failure, returns the first error at the end.
    ///
    /// # Comparison with other shutdown paths
    ///
    /// | Method | Hints | Fixed sidecar/flag | Final fsync | Txlog checkpoint | Stops workers | Error |
    /// |---|---|---|---|---|---|---|
    /// | periodic_flush | no | no | dirty Bitcask/Fixed; LSM policy | no | no | logged |
    /// | close | writes | no | dirty shards only | yes | no | returned |
    /// | clean_shutdown | writes | yes | yes | yes | no | returned |
    /// | shutdown(&mut self) | via clean_shutdown | yes | yes | yes | yes | returned |
    /// | Drop | via close | no | dirty shards only | yes | yes | logged only |
    ///
    /// Hinted Bitcask collections re-barrier once per shard during [`Drop`] after
    /// `clean_shutdown`; clean no-hints shards issue zero fsyncs.
    ///
    pub fn clean_shutdown(&self) -> DbResult<()> {
        let collections = lock(&self.collections).clone();
        self.finalize(&collections, FinalizeMode::CleanShutdown)
    }

    fn finalize(&self, collections: &[Arc<dyn Collection>], mode: FinalizeMode) -> DbResult<()> {
        #[cfg(feature = "fjall")]
        let mut lsm_finalization = self.lsm_clean_marker.begin_finalization();
        self.disable_all_ttl_sweeps(collections);
        let start = Instant::now();
        let mut first_err: Option<DbError> = None;
        let mut failed = 0usize;
        for c in collections.iter() {
            match mode {
                FinalizeMode::Close => {
                    if let Err(e) = c.flush() {
                        tracing::warn!(collection = c.name(), error = %e, "close flush failed");
                        failed += 1;
                        note_first_error(&mut first_err, e);
                    }
                    if let Err(e) = c.durable_flush() {
                        tracing::warn!(collection = c.name(), error = %e, "close durable_flush failed");
                        failed += 1;
                        note_first_error(&mut first_err, e);
                    }
                }
                FinalizeMode::CleanShutdown => {
                    if let Err(e) = c.clean_shutdown() {
                        tracing::warn!(collection = c.name(), error = %e, "clean_shutdown failed");
                        failed += 1;
                        note_first_error(&mut first_err, e);
                    }
                }
            }
            if let Err(e) = self.save_collection_len(c.name(), c.len() as u64) {
                tracing::warn!(collection = c.name(), error = %e, "finalization save_collection_len failed");
                failed += 1;
                note_first_error(&mut first_err, e);
            }
        }
        // Clear counters_dirty only after every collection operation above
        // succeeded, and only for a collection this session accounted for.
        if first_err.is_none()
            && let Err(e) = self.clear_consumed_counter_flags(collections)
        {
            failed += 1;
            note_first_error(&mut first_err, e);
        }
        if let Err(e) = self.seq.flush() {
            tracing::warn!(error = %e, "seq flush failed");
            failed += 1;
            note_first_error(&mut first_err, e);
        }
        let pending = self.pending_transactions();
        if !pending.is_empty() {
            tracing::warn!(
                count = pending.len(),
                "finalizing with pending transactions still in the log"
            );
        }
        if let Err(e) = self.checkpoint_now() {
            tracing::warn!(error = %e, "finalization checkpoint failed");
            failed += 1;
            note_first_error(&mut first_err, e);
        }
        // Publish the LSM clean marker only after every collection operation,
        // metadata update, sequence flush, and final checkpoint succeeded. A
        // failed checkpoint must leave the next open in recovery mode, where it
        // re-seeds collection lengths from live keyspaces instead of trusting
        // potentially stale db.info counts.
        #[cfg(feature = "fjall")]
        if first_err.is_none() {
            match self.db_info.update(|i| i.clean = true) {
                Ok(()) => lsm_finalization.note_clean_persisted(),
                Err(e) => {
                    tracing::warn!(error = %e, "finalization db_info clean flag failed");
                    failed += 1;
                    note_first_error(&mut first_err, e.into());
                }
            }
        }
        #[cfg(feature = "fjall")]
        if first_err.is_none() {
            lsm_finalization.publish_clean();
        } else if let Err(e) = lsm_finalization.finish_dirty() {
            tracing::warn!(error = %e, "finalization LSM clean-marker dirty publish failed");
            failed += 1;
            note_first_error(&mut first_err, e);
        }
        tracing::info!(
            collections = collections.len(),
            failed,
            elapsed_ms = start.elapsed().as_millis() as u64,
            mode = match mode {
                FinalizeMode::Close => "close",
                FinalizeMode::CleanShutdown => "clean_shutdown",
            },
            "db finalization"
        );
        first_err.map_or(Ok(()), Err)
    }

    /// Graceful shutdown: stop background workers, perform clean finalization,
    /// then close watches. Callers must stop application writers first.
    pub fn shutdown(&mut self) -> DbResult<()> {
        self.stop_workers();
        let result = self.clean_shutdown();
        self.close_all_watches();
        result
    }

    fn stop_workers(&mut self) {
        self.shutdown.shutdown();
        #[cfg(all(feature = "typed-tree", feature = "armour"))]
        self.queue_wake_service.join();
        #[cfg(feature = "rpc")]
        for h in lock(&self.rpc_handles).drain(..) {
            drop(h); // RpcHandle::drop broadcasts + joins
        }
        if let Some(ref mut c) = self.compactor {
            c.stop();
        }
        if let Some(ref mut f) = self.flusher {
            f.stop();
        }
        if let Some(ref mut s) = self.sweeper {
            s.stop();
        }
    }

    fn close_all_watches(&self) {
        let collections = lock(&self.collections).clone();
        for collection in collections {
            collection.close_watches();
        }
    }

    /// Disable TTL sweep on every collection, waiting for in-flight passes.
    ///
    /// Used by [`close`](Self::close) / [`clean_shutdown`](Self::clean_shutdown)
    /// so the sweeper cannot write tombstones during finalization. Failures are
    /// logged and skipped — disable must not abort durability flush.
    fn disable_all_ttl_sweeps(&self, collections: &[Arc<dyn Collection>]) {
        if collections.is_empty() {
            return;
        }
        // Deliberately not re-enabled afterwards: `close` may be followed by more
        // writes, and silently resuming expiry mid-finalization is the worse
        // failure. Say so at info level — an application that uses `close` as a
        // checkpoint and keeps running would otherwise see TTL simply stop, with
        // nothing in the log to explain it.
        tracing::info!(
            collections = collections.len(),
            "ttl sweep disabled for finalization; re-enable per collection with \
             Db::set_ttl_sweep(name, true) if this Db keeps running"
        );
        for c in collections.iter() {
            if let Err(e) = c.set_ttl_sweep(false) {
                tracing::warn!(
                    collection = c.name(),
                    error = %e,
                    "disable ttl sweep failed"
                );
            }
        }
    }

    /// Return a clone of the shutdown signal for passing to replication
    /// workers or custom background tasks.
    pub fn shutdown_signal(&self) -> ShutdownSignal {
        self.shutdown.clone()
    }

    /// Trigger a compaction pass for all collections.
    pub fn compact(&self) -> DbResult<usize> {
        let snapshot = lock(&self.collections).clone();
        let mut total = 0;
        for c in &snapshot {
            total += c.compact()?;
        }
        Ok(total)
    }

    #[cfg(all(feature = "typed-tree", feature = "armour"))]
    pub(crate) fn register_queue_retention(&self, retention: Arc<dyn QueueRetentionPass>) {
        lock(&self.queue_retentions).push(retention);
    }

    /// Sweep expired rows across all collections using the current wall clock.
    pub fn sweep_expired_now(&self) -> DbResult<usize> {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_err(|_| DbError::Internal("system clock before unix epoch"))?
            .as_millis() as u64;
        let snapshot = lock(&self.collections).clone();
        let mut total_deleted = 0;
        let mut first_err = None;
        for c in &snapshot {
            match ttl_sweep_collection(c, now_ms, self.ttl_budget) {
                Ok(out) => total_deleted += out.deleted,
                Err(e) if first_err.is_none() => first_err = Some(e),
                Err(_) => {}
            }
        }
        #[cfg(all(feature = "typed-tree", feature = "armour"))]
        for retention in &*lock(&self.queue_retentions) {
            if let Err(e) = retention.run_retention()
                && first_err.is_none()
            {
                first_err = Some(e);
            }
        }
        match first_err {
            Some(e) => Err(e),
            None => Ok(total_deleted),
        }
    }

    /// Enable or disable TTL sweep for a named collection.
    pub fn set_ttl_sweep(&self, name: &str, enabled: bool) -> DbResult<()> {
        // Clone the handle out before acting: disabling waits for an in-flight
        // pass, and holding the Db-wide `collections` lock across that wait would
        // block every `open_*` / `flush_all` / `close` for the length of a sweep.
        self.collection_by_name(name)?.set_ttl_sweep(enabled)
    }

    /// Rebuild the TTL index for a named collection.
    pub fn rebuild_ttl_index(&self, name: &str) -> DbResult<()> {
        // Cloned out for the same reason as `set_ttl_sweep`: a rebuild is a full
        // O(N) walk, hundreds of ms on a large Var collection.
        self.collection_by_name(name)?.rebuild_ttl_index()
    }

    /// Recount `collection`'s declared counter targets synchronously (T9
    /// Task 8): the same two-pass walk the open-time trigger runs
    /// (`counter::recount`), callable on an already-open collection without a
    /// flag or a reopen. Returns the number of target rows repaired.
    ///
    /// This is the promotion entry point: a promoted replica calls it for
    /// every counted source before accepting writes, because replication runs
    /// per collection, so the source and counter streams can settle at
    /// independent positions on a fresh promotion. A dirty-flag trigger would
    /// not fire here — the collections are already open, not being opened —
    /// and a clean close would clear the flag before anything used it.
    ///
    /// **Leaves `counters_dirty` exactly as it found it** — it neither clears
    /// the flag nor consumes the in-memory marker. Unlike the open-time
    /// trigger, this runs on an **already-open** collection that may be taking
    /// concurrent writes, and against a live writer the recount is not
    /// guaranteed to converge: a write landing after its source row was
    /// scanned increments the target before the repair reads it, and the
    /// repair then subtracts that legitimate increment to match its own stale
    /// snapshot. Clearing the flag afterwards would erase the only evidence
    /// that the drift it just introduced is owed a repair, and the drift would
    /// survive the restart. Leaving the flag costs one recount at the next
    /// unclean open, which is idempotent.
    ///
    /// Quiescing writers is still the caller's job for an *accurate*
    /// promotion recount; not clearing the flag only stops an inaccurate one
    /// from becoming permanent.
    ///
    /// Refuses [`DbError::NotACounterSource`] when `collection` declares no
    /// counter targets of its own: for a plain counter *target*, a recount
    /// here would walk nothing at all, and the caller almost certainly means
    /// the source. Mirrors the open-time trigger, which likewise only ever
    /// recounts a source.
    pub fn recount_counters(&self, collection: &str) -> DbResult<u64> {
        let coll = self.collection_by_name(collection)?;
        if coll.counter_target_names().is_empty() {
            return Err(DbError::NotACounterSource {
                collection: collection.to_string(),
            });
        }
        coll.reconcile_counters()
    }

    /// Look up an open collection by name, cloning the handle out of the registry.
    fn collection_by_name(&self, name: &str) -> DbResult<Arc<dyn Collection>> {
        let cols = lock(&self.collections);
        cols.iter()
            .find(|c| c.name() == name)
            .cloned()
            .ok_or(DbError::Client("unknown collection"))
    }

    // -- TypedTree ---------------------------------------------------------

    /// Open a [`TypedTree`] with auto-migration.
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::replication::ReplicationTarget`]. Master and follower nodes
    /// each run their own migration on their own on-disk data. The migration
    /// callback must be deterministic (state converges across replication
    /// nodes) **and idempotent**: a crash between the durable data flush and
    /// the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is not
    /// provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_typed_tree<T, C, H>(
        &self,
        config: Config,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<TypedTree<T::SelfId, T, C, TtlHook<T, H>>>>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        C: Codec<T> + Default + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        self.open_managed(
            |meta| TypedTree::open_inner(self.tree_path(meta.name), config, C::default(), hook),
            crate::access::CodecWire(C::default()),
            migrations,
            StorageDescriptor::Bitcask {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                claims,
            },
        )
    }

    // -- TypedMap ----------------------------------------------------------

    /// Open a [`TypedMap`] with auto-migration.
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::replication::ReplicationTarget`]. Master and follower nodes
    /// each run their own migration on their own on-disk data. The migration
    /// callback must be deterministic (state converges across replication
    /// nodes) **and idempotent**: a crash between the durable data flush and
    /// the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is not
    /// provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_typed_map<T, C, H>(
        &self,
        config: Config,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<TypedMap<T::SelfId, T, C, TtlHook<T, H>>>>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Send + Sync + Hash + Eq,
        C: Codec<T> + Default + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        self.open_managed(
            |meta| TypedMap::open_inner(self.tree_path(meta.name), config, C::default(), hook),
            crate::access::CodecWire(C::default()),
            migrations,
            StorageDescriptor::Bitcask {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                claims,
            },
        )
    }

    // -- VarTypedTree ------------------------------------------------------

    /// Open a [`VarTypedTree`](crate::VarTypedTree) with auto-migration (disk-resident typed values).
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::replication::ReplicationTarget`]. Master and follower nodes
    /// each run their own migration on their own on-disk data. The migration
    /// callback must be deterministic (state converges across replication
    /// nodes) **and idempotent**: a crash between the durable data flush and
    /// the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is not
    /// provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[cfg(feature = "var-collections")]
    pub fn open_var_typed_tree<T, C, H>(
        &self,
        config: Config,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<crate::VarTypedTree<T::SelfId, T, C, TtlHook<T, H>>>>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        C: Codec<T> + Clone + Default + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        self.open_managed(
            |meta| {
                crate::VarTypedTree::open_hooked_inner(
                    self.tree_path(meta.name),
                    config,
                    C::default(),
                    hook,
                )
            },
            crate::access::CodecWire(C::default()),
            migrations,
            StorageDescriptor::Bitcask {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                claims,
            },
        )
    }

    // -- VarTypedMap -------------------------------------------------------

    /// Open a [`VarTypedMap`](crate::VarTypedMap) with auto-migration (disk-resident typed values).
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::replication::ReplicationTarget`]. Master and follower nodes
    /// each run their own migration on their own on-disk data. The migration
    /// callback must be deterministic (state converges across replication
    /// nodes) **and idempotent**: a crash between the durable data flush and
    /// the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is not
    /// provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[cfg(feature = "var-collections")]
    pub fn open_var_typed_map<T, C, H>(
        &self,
        config: Config,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<crate::VarTypedMap<T::SelfId, T, C, TtlHook<T, H>>>>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Send + Sync + Hash + Eq,
        C: Codec<T> + Clone + Default + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        self.open_managed(
            |meta| {
                crate::VarTypedMap::open_hooked_inner(
                    self.tree_path(meta.name),
                    config,
                    C::default(),
                    hook,
                )
            },
            crate::access::CodecWire(C::default()),
            migrations,
            StorageDescriptor::Bitcask {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                claims,
            },
        )
    }

    // -- ZeroTree ---------------------------------------------------------

    /// Open a [`ZeroTree`] (Bitcask backend) with auto-migration.
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::replication::ReplicationTarget`]. Master and follower nodes
    /// each run their own migration on their own on-disk data. The migration
    /// callback must be deterministic (state converges across replication
    /// nodes) **and idempotent**: a crash between the durable data flush and
    /// the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is not
    /// provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_zero_tree<T, const V: usize, H>(
        &self,
        config: Config,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<ZeroTree<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Bitcask>>>
    where
        T: CollectionMeta
            + Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + Send
            + Sync
            + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        self.open_managed(
            |meta| ZeroTree::open_hooked_inner(self.tree_path(meta.name), config, hook),
            crate::access::ZeroWire::<V>,
            migrations,
            StorageDescriptor::Bitcask {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                claims,
            },
        )
    }

    pub(crate) fn open_zero_tree_fixed_internal<T, const V: usize, H>(
        &self,
        config: FixedConfig,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
        after_stage: impl FnOnce(
            &Arc<ZeroTree<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Fixed>>,
        ) -> DbResult<()>,
    ) -> DbResult<Arc<ZeroTree<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Fixed>>>
    where
        T: CollectionMeta
            + Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + Send
            + Sync
            + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.open_zero_tree_fixed_with_access(
            config,
            hook,
            migrations,
            ManagedCollectionAccess::Internal,
            after_stage,
        )
    }

    fn open_zero_tree_fixed_with_access<T, const V: usize, H>(
        &self,
        config: FixedConfig,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
        access: ManagedCollectionAccess,
        after_stage: impl FnOnce(
            &Arc<ZeroTree<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Fixed>>,
        ) -> DbResult<()>,
    ) -> DbResult<Arc<ZeroTree<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Fixed>>>
    where
        T: CollectionMeta
            + Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + Send
            + Sync
            + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        let value_len =
            u16::try_from(V).map_err(|_| DbError::Config("fixed value length exceeds u16::MAX"))?;
        self.open_managed_with_access_after_stage(
            |meta| ZeroTree::open_with_hook_inner(self.tree_path(meta.name), config, hook),
            crate::access::ZeroWire::<V>,
            migrations,
            StorageDescriptor::Fixed {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                value_len,
                claims,
            },
            access,
            after_stage,
        )
    }

    // -- ZeroTree (Fixed) -------------------------------------------------

    /// Open a [`ZeroTree`] with Fixed (fixed-slot) backend.
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::fixed_replication::FixedReplicationTarget`]. Master and
    /// follower nodes each run their own migration on their own on-disk data.
    /// The migration callback must be deterministic (state converges across
    /// replication nodes) **and idempotent**: a crash between the durable data
    /// flush and the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is
    /// not provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_zero_tree_fixed<T, const V: usize, H>(
        &self,
        config: FixedConfig,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<ZeroTree<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Fixed>>>
    where
        T: CollectionMeta
            + Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + Send
            + Sync
            + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.open_zero_tree_fixed_with_access(
            config,
            hook,
            migrations,
            ManagedCollectionAccess::Public,
            |_| Ok(()),
        )
    }

    // -- ZeroMap ----------------------------------------------------------

    /// Open a [`ZeroMap`] (Bitcask backend) with auto-migration.
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::replication::ReplicationTarget`]. Master and follower nodes
    /// each run their own migration on their own on-disk data. The migration
    /// callback must be deterministic (state converges across replication
    /// nodes) **and idempotent**: a crash between the durable data flush and
    /// the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is not
    /// provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_zero_map<T, const V: usize, H>(
        &self,
        config: Config,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<ZeroMap<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Bitcask>>>
    where
        T: CollectionMeta
            + Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + Send
            + Sync
            + 'static,
        T::SelfId: Key + Send + Sync + Hash + Eq,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        self.open_managed(
            |meta| ZeroMap::open_hooked_inner(self.tree_path(meta.name), config, hook),
            crate::access::ZeroWire::<V>,
            migrations,
            StorageDescriptor::Bitcask {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                claims,
            },
        )
    }

    // -- ZeroMap (Fixed) --------------------------------------------------

    /// Open a [`ZeroMap`] with Fixed (fixed-slot) backend.
    ///
    /// **Replication-migration discipline**: when migrations are registered,
    /// they are applied locally during this call, before the caller wires any
    /// [`crate::fixed_replication::FixedReplicationTarget`]. Master and
    /// follower nodes each run their own migration on their own on-disk data.
    /// The migration callback must be deterministic (state converges across
    /// replication nodes) **and idempotent**: a crash between the durable data
    /// flush and the version commit re-runs the step on already-migrated data.
    /// Cross-node coordination of migrations is
    /// not provided; see
    /// `docs/superpowers/specs/26-05-22-armdb-armour-integration-deferred.md` D5.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_zero_map_fixed<T, const V: usize, H>(
        &self,
        config: FixedConfig,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<ZeroMap<T::SelfId, V, T, TtlHook<T, H>, crate::durability::Fixed>>>
    where
        T: CollectionMeta
            + Copy
            + zerocopy::IntoBytes
            + zerocopy::FromBytes
            + zerocopy::Immutable
            + Send
            + Sync
            + 'static,
        T::SelfId: Key + Send + Sync + Hash + Eq,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        self.check_stream_open::<T, H>(&hook)?;
        crate::unique::check_no_ttl_claims::<T, H>(&hook)?;
        crate::unique::check_claim_key_width::<T, H>(&hook)?;
        self.check_claim_owner_tags::<T, H>(&hook)?;
        let hook = TtlHook::<T, H>::new(hook);
        let format_config = config.clone();
        let claims = H::CLAIMS
            || hook
                .claim_targets()
                .is_some_and(|targets| !targets.is_empty());
        let value_len =
            u16::try_from(V).map_err(|_| DbError::Config("fixed value length exceeds u16::MAX"))?;
        self.open_managed(
            |meta| ZeroMap::open_with_hook_inner(self.tree_path(meta.name), config, hook),
            crate::access::ZeroWire::<V>,
            migrations,
            StorageDescriptor::Fixed {
                config: &format_config,
                key_len: std::mem::size_of::<T::SelfId>(),
                value_len,
                claims,
            },
        )
    }

    // -- UniqueIndex (unique namespace) ------------------------------------

    /// Open a unique namespace declared with
    /// [`unique_index!`](crate::unique_index).
    ///
    /// A thin wrapper over [`Self::open_zero_map_fixed`]: the namespace is a
    /// FixedStore [`ZeroMap`] whose 80-byte row the engine owns, returned as
    /// the opaque [`UniqueIndex`] rather than the raw handle — there is **no
    /// public write API on it, ever**. A source declares claims into it with
    /// [`crate::projection::Projected::and_unique`]
    /// (typically via [`crate::schema::SchemaRegistry::unique_index`]), and
    /// `db.tx()`'s commit path (`resolve_claims`) is the only thing that ever
    /// writes a row here — under CAS-style uniqueness enforcement, never a
    /// bare put. `owner_of`/`contains` read whatever that resolution has
    /// written so far.
    ///
    #[doc = include_str!("../../docs/exclusive-access.md")]
    pub fn open_unique_index<N: UniqueNamespace>(
        &self,
        config: FixedConfig,
    ) -> DbResult<Arc<UniqueIndex<N>>>
    where
        N::SelfId: Key + Send + Sync + Hash + Eq,
    {
        let inner =
            self.open_zero_map_fixed::<N, { size_of::<UniqueRow>() }, NoHook>(config, NoHook, &[])?;
        Ok(Arc::new(UniqueIndex::from_inner(inner)))
    }

    // -- LsmTree -----------------------------------------------------------

    #[cfg(feature = "fjall")]
    fn lsm_database(
        &self,
        db_config: Option<crate::lsm::LsmDbConfig>,
    ) -> DbResult<(fjall::Database, crate::lsm::PersistPolicy)> {
        let mut guard = lock(&self.lsm_db);
        if let Some((db, existing)) = guard.as_ref() {
            if let Some(req) = db_config
                && *existing != req
            {
                return Err(DbError::Config(
                    "LsmDbConfig differs from the already-open shared fjall Database (first-open-wins)",
                ));
            }
            return Ok((db.clone(), existing.persist));
        }
        let cfg = db_config.unwrap_or_default();
        let root = self.path.join("lsm");
        let db = crate::lsm::map_fjall(
            fjall::Database::builder(&root)
                .cache_size(cfg.cache_size)
                .open(),
        )?;
        *guard = Some((db.clone(), cfg));
        Ok((db, cfg.persist))
    }

    /// Open a [`crate::lsm::LsmTree`] backed by the shared `fjall::Database`.
    ///
    /// On the FIRST `open_lsm_tree`, `db_config` establishes the shared database
    /// config; `None` uses [`crate::lsm::LsmDbConfig::default`]. On later calls a
    /// differing `db_config` returns `DbError::Config` (first-open-wins); `None`
    /// inherits the established config.
    ///
    /// `migrations` are applied like every other `open_*`: on a version bump the
    /// registered step fns run and the new version is committed durably; a
    /// typ_hash drift at the same version or a downgrade returns
    /// `DbError::SchemaMismatch`.
    #[cfg(feature = "fjall")]
    pub fn open_lsm_tree<T, C, H>(
        &self,
        db_config: Option<crate::lsm::LsmDbConfig>,
        config: crate::lsm::LsmConfig,
        hook: H,
        migrations: &[TypedMigration<T::SelfId, T>],
    ) -> DbResult<Arc<crate::lsm::LsmTree<T::SelfId, T, C, H>>>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Ord + Send + Sync + 'static,
        C: Codec<T> + Clone + Default + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        // An `LsmTree` is not a transaction participant, so a change-stream
        // half attached to one would never run: no emission, no
        // `ChangeStreamRequiresTx` guard on its non-transactional writes, and
        // `schema::validate` has no queue check that would surface it. Unlike
        // T9/T10's silent exclusions this one is cheap to close, so it is an
        // open-time error rather than a documented trap.
        self.check_stream_open::<T, H>(&hook)?;
        if H::WATCHES {
            return Err(DbError::WatchUnsupported);
        }
        self.open_managed(
            |meta| {
                let (fdb, persist) = self.lsm_database(db_config)?;
                // Trust the persisted count only when the previous session closed
                // cleanly; after a crash opened_clean=false so we pass None and let
                // open_with_persist rescan via ks.len().
                let initial_count = if self.opened_clean {
                    self.stored_info(meta.name).as_ref().map(|i| i.seq)
                } else {
                    None
                };
                crate::lsm::LsmTree::<T::SelfId, T, C, H>::open_with_persist(
                    fdb,
                    config,
                    persist,
                    C::default(),
                    hook,
                    initial_count,
                    Arc::clone(&self.lsm_clean_marker),
                )
            },
            crate::access::CodecWire(C::default()),
            migrations,
            StorageDescriptor::Lsm,
        )
    }

    // -- Private helpers --------------------------------------------------

    fn stored_info(&self, name: &str) -> Option<CollectionInfo> {
        self.db_info.cloned().collections.get(name).cloned()
    }

    #[cfg(all(test, feature = "fjall"))]
    pub(crate) fn has_lsm_database_for_test(&self) -> bool {
        lock(&self.lsm_db).is_some()
    }

    /// Atomically reserve a collection name for the duration of an open, or
    /// reject a duplicate/reserved one. Two collections mapped to the same
    /// `tree_path` would drive two independent in-memory indexes over the same
    /// shard files, silently losing each other's updates; the reserved names
    /// collide with internal `Db` state directories.
    ///
    /// P1-5: the live `collections` list is only updated *after* the expensive
    /// open/recovery/migration, so a bare check-then-open is a TOCTOU race — two
    /// concurrent opens of the same name could both pass. The check and the
    /// insertion into `reserved_names` happen together under the `reserved_names`
    /// lock, so at most one open can hold a given name at a time. The returned
    /// guard frees the reservation on drop; hold it (bind to a named local, not
    /// `_`) until *after* `publish_collection` succeeds, so the name is
    /// continuously owned by either the reservation or the published registries.
    fn reserve_name(&self, name: &'static str) -> DbResult<NameReservation<'_>> {
        if RESERVED_COLLECTION_NAMES.contains(&name) {
            return Err(DbError::DuplicateCollection(name));
        }
        // Lock order is always reserved_names -> collections (never the reverse),
        // so this nested acquisition cannot deadlock.
        let mut reserved = lock(&self.reserved_names);
        if lock(&self.collections).iter().any(|c| c.name() == name) {
            return Err(DbError::DuplicateCollection(name));
        }
        if !reserved.insert(name) {
            return Err(DbError::DuplicateCollection(name));
        }
        Ok(NameReservation { db: self, name })
    }

    /// Shared open → migrate → replay → publish recipe for every typed family.
    ///
    /// `open` constructs the collection however the family needs (path + Config,
    /// FixedConfig, or the shared LSM keyspace). The recipe never assumes
    /// `(path, Config, hook)`.
    fn open_managed<T, S, W>(
        &self,
        open: impl FnOnce(&TreeMeta) -> DbResult<S>,
        wire: W,
        migrations: &[TypedMigration<T::SelfId, T>],
        descriptor: StorageDescriptor<'_>,
    ) -> DbResult<Arc<S>>
    where
        T: CollectionMeta + Send + Sync + 'static,
        S: crate::access::Access<Key = T::SelfId, Val = T>
            + crate::access::ManagedOpen<Key = T::SelfId, Val = T>
            + Collection
            + 'static,
        W: crate::access::Wire<T>,
    {
        descriptor.validate()?;
        self.open_managed_with_access(
            open,
            wire,
            migrations,
            descriptor,
            ManagedCollectionAccess::Public,
        )
    }

    fn open_managed_with_access<T, S, W>(
        &self,
        open: impl FnOnce(&TreeMeta) -> DbResult<S>,
        wire: W,
        migrations: &[TypedMigration<T::SelfId, T>],
        descriptor: StorageDescriptor<'_>,
        access: ManagedCollectionAccess,
    ) -> DbResult<Arc<S>>
    where
        T: CollectionMeta + Send + Sync + 'static,
        S: crate::access::Access<Key = T::SelfId, Val = T>
            + crate::access::ManagedOpen<Key = T::SelfId, Val = T>
            + Collection
            + 'static,
        W: crate::access::Wire<T>,
    {
        self.open_managed_with_access_after_stage(
            open,
            wire,
            migrations,
            descriptor,
            access,
            |_| Ok(()),
        )
    }

    /// [`Self::open_managed_with_access`] plus a callback that runs after
    /// migrations and replay but **before** the collection is published.
    ///
    /// A collection is visible to the TTL sweeper the moment it is published,
    /// with its write hook already live. Anything that has to observe the
    /// collection's contents exactly once — seeding a counter the hook then
    /// maintains, for instance — has to run here, or a concurrent sweep can
    /// interleave with it. Failing the callback leaves the collection
    /// unpublished, exactly as the paired opener's `after_stage` does.
    fn open_managed_with_access_after_stage<T, S, W>(
        &self,
        open: impl FnOnce(&TreeMeta) -> DbResult<S>,
        wire: W,
        migrations: &[TypedMigration<T::SelfId, T>],
        descriptor: StorageDescriptor<'_>,
        access: ManagedCollectionAccess,
        after_stage: impl FnOnce(&Arc<S>) -> DbResult<()>,
    ) -> DbResult<Arc<S>>
    where
        T: CollectionMeta + Send + Sync + 'static,
        S: crate::access::Access<Key = T::SelfId, Val = T>
            + crate::access::ManagedOpen<Key = T::SelfId, Val = T>
            + Collection
            + 'static,
        W: crate::access::Wire<T>,
    {
        descriptor.validate()?;
        let meta = TreeMeta::of::<T>();
        let staged = self.stage_managed(
            open,
            wire,
            migrations,
            descriptor,
            self.reserve_name(meta.name)?,
        )?;
        after_stage(&staged.collection)?;
        let collection = Arc::clone(&staged.collection);
        self.publish_staged_adapters(vec![self.stage_to_adapter(staged, access)])?;
        Ok(collection)
    }

    #[allow(clippy::too_many_arguments)] // paired staging keeps both managed-open contracts explicit
    pub(crate) fn open_managed_pair<T1, S1, W1, T2, S2, W2>(
        &self,
        open_first: impl FnOnce(&TreeMeta) -> DbResult<S1>,
        wire_first: W1,
        migrations_first: &[TypedMigration<T1::SelfId, T1>],
        descriptor_first: StorageDescriptor<'_>,
        open_second: impl FnOnce(&TreeMeta) -> DbResult<S2>,
        wire_second: W2,
        migrations_second: &[TypedMigration<T2::SelfId, T2>],
        descriptor_second: StorageDescriptor<'_>,
        after_stage: impl FnOnce(&Arc<S1>, &Arc<S2>) -> DbResult<()>,
        first_access: ManagedCollectionAccess,
        second_access: ManagedCollectionAccess,
    ) -> DbResult<(Arc<S1>, Arc<S2>)>
    where
        T1: CollectionMeta + Send + Sync + 'static,
        S1: crate::access::Access<Key = T1::SelfId, Val = T1>
            + crate::access::ManagedOpen<Key = T1::SelfId, Val = T1>
            + Collection
            + 'static,
        W1: crate::access::Wire<T1>,
        T2: CollectionMeta + Send + Sync + 'static,
        S2: crate::access::Access<Key = T2::SelfId, Val = T2>
            + crate::access::ManagedOpen<Key = T2::SelfId, Val = T2>
            + Collection
            + 'static,
        W2: crate::access::Wire<T2>,
    {
        descriptor_first.validate()?;
        descriptor_second.validate()?;
        let first_meta = TreeMeta::of::<T1>();
        let second_meta = TreeMeta::of::<T2>();
        descriptor_first.validate_persisted(&self.tree_path(first_meta.name))?;
        descriptor_second.validate_persisted(&self.tree_path(second_meta.name))?;
        let first_guard = self.reserve_name(first_meta.name)?;
        let second_guard = self.reserve_name(second_meta.name)?;
        let first = self.stage_managed(
            open_first,
            wire_first,
            migrations_first,
            descriptor_first,
            first_guard,
        )?;
        let second = self.stage_managed(
            open_second,
            wire_second,
            migrations_second,
            descriptor_second,
            second_guard,
        )?;
        after_stage(&first.collection, &second.collection)?;
        let collections = (
            Arc::clone(&first.collection),
            Arc::clone(&second.collection),
        );
        self.publish_staged_adapters(vec![
            self.stage_to_adapter(first, first_access),
            self.stage_to_adapter(second, second_access),
        ])?;
        Ok(collections)
    }

    fn stage_managed<'a, T, S, W>(
        &self,
        open: impl FnOnce(&TreeMeta) -> DbResult<S>,
        wire: W,
        migrations: &[TypedMigration<T::SelfId, T>],
        descriptor: StorageDescriptor<'_>,
        name_guard: NameReservation<'a>,
    ) -> DbResult<StagedManaged<'a, S, W>>
    where
        T: CollectionMeta + Send + Sync + 'static,
        S: crate::access::Access<Key = T::SelfId, Val = T>
            + crate::access::ManagedOpen<Key = T::SelfId, Val = T>
            + Collection
            + 'static,
        W: crate::access::Wire<T>,
    {
        let meta = TreeMeta::of::<T>();
        descriptor.validate_persisted(&self.tree_path(meta.name))?;
        let mut stored = self.stored_info(meta.name);
        let has_pending_tx_ops = self
            .txlog()
            .pending()
            .iter()
            .any(|record| record.ops.iter().any(|op| op.name == meta.name));

        // A same-version type drift normally fails before physical open. The
        // one exception is a fail-closed pending Put whose logged fingerprint
        // already disagrees with the stored fingerprint: if that operation
        // genuinely needs replay, replay owns the more specific
        // `TxSchemaMismatch`. A stale record that already landed is skipped by
        // `needs_replay`; in that case restore the ordinary TypHash error
        // immediately after replay and before migration/publication.
        let deferred_typ_hash = stored.as_ref().and_then(|info| {
            let expected = meta.ty.h();
            let candidate = info.version == meta.version
                && info.typ_hash != expected
                && !self.tx_config().discard_pending_on_schema_mismatch
                && self.txlog().pending().iter().any(|record| {
                    record.ops.iter().any(|op| {
                        op.name == meta.name
                            && matches!(op.kind, crate::txlog::record::OpKind::Put)
                            && op.schema_fingerprint != info.typ_hash
                    })
                });
            candidate.then_some((info.typ_hash, expected))
        });
        let migration_plan = if deferred_typ_hash.is_some() {
            None
        } else {
            build_plan::<T::SelfId, T>(
                meta.name,
                stored.as_ref().map(|info| (info.version, info.typ_hash)),
                migrations,
            )?
        };
        if let Some(plan) = migration_plan
            .as_ref()
            .filter(|plan| !plan.format_steps.is_empty())
        {
            crate::format_migration::refuse_if_required(descriptor.policy(), meta.name, plan)?;
            if matches!(descriptor, StorageDescriptor::Lsm) {
                return Err(DbError::FormatMigrationUnsupported {
                    collection: meta.name,
                    backend: "lsm",
                });
            }
        }

        // Replay admission must precede every durable migration mutation:
        // layout recovery, dirty flags, markers, file-id burns, and physical
        // shard rewrites.
        // The retained plan spans the complete declared format history, not
        // just the physical suffix still pending from the installed version.
        let replay_plan = if deferred_typ_hash.is_some() {
            None
        } else {
            build_replay_plan::<T::SelfId, T>(
                meta.name,
                stored.as_ref().map(|info| (info.version, info.typ_hash)),
                migrations,
            )?
        };
        if has_pending_tx_ops && let Some(replay_plan) = replay_plan.as_ref() {
            self.guard_ambiguous_format_replay(meta.name, meta.ty.h(), replay_plan)?;
        }

        if let Some(plan) = migration_plan
            .as_ref()
            .filter(|plan| !plan.format_steps.is_empty())
        {
            self.validate_format_backend_layout(&meta, plan, descriptor)?;
        }
        if let Some(shard_count) = descriptor.shard_count() {
            cleanup_stale_format_temps(&self.tree_path(meta.name), shard_count)?;
        }
        self.remove_committed_format_marker(meta.name, stored.as_ref())?;

        if let Some(plan) = migration_plan
            .as_ref()
            .filter(|plan| !plan.format_steps.is_empty())
        {
            self.set_projections_dirty(meta.name, true)?;
            write_collection_marker(
                &self.tree_path(meta.name),
                FormatMarker {
                    from: plan.stored_version,
                    to: plan.format_to_version,
                },
            )?;
            let outcome = self.migrate_format_backend(&meta, plan, descriptor)?;
            self.commit_format_outcome(
                meta.name,
                meta.ty.h(),
                plan,
                descriptor.claims(),
                &outcome,
            )?;
            stored = self.stored_info(meta.name);
        }

        // `migration_plan` describes only physical work still required from
        // the installed version. Once that work commits (or an explicit
        // `migrate_format*` call already committed it), rebuilding that plan
        // returns `None`. Transaction records are intentionally not rewritten,
        // so reconstruct the full retained checked format prefix separately
        // and keep it alive through replay only.
        let coll = open(&meta)?;
        let coll = Arc::new(coll);
        S::bind_tx_arc(&coll);
        if let Some(info) = stored.as_ref() {
            // Format recovery's durable high-water marks outlive rewritten
            // files. Install them before replay so an old pending operation
            // cannot be mistaken for an unapplied transaction.
            crate::access::ManagedOpen::floor_recovered_gsns(
                &*coll,
                info.gsn_high,
                &info.shard_gsn_high,
            );
        }
        // Replay before typed value migrations. Old logged Put bytes are
        // transformed in memory through the retained format prefix; the txlog
        // itself stays byte-identical and physical format migration never runs
        // from this replay-only plan.
        let replayed = coll.replay_pending_tx_log(
            self,
            stored.as_ref().map(|info| info.typ_hash),
            replay_plan.as_ref(),
        )?;
        if let Some((stored, expected)) = deferred_typ_hash {
            return Err(DbError::SchemaMismatch {
                name: meta.name.to_owned(),
                kind: SchemaMismatchKind::TypHash { stored, expected },
            });
        }
        if replayed > 0 {
            tracing::info!(
                collection = meta.name,
                replayed,
                "replayed pending transaction operations"
            );
        }
        let mut projections_dirty = stored.as_ref().is_some_and(|info| info.projections_dirty);
        if Self::migration_will_run(stored.as_ref(), &meta) && coll.projections_enabled() {
            projections_dirty = true;
            self.set_projections_dirty(meta.name, true)?;
        }
        self.run_migration(
            &meta,
            stored.as_ref(),
            migrations,
            |mfn| coll.migrate_inner(mfn, false),
            || coll.flush_for_migration(),
        )?;
        let skip_reconcile = {
            #[cfg(any(test, feature = "test-faults"))]
            {
                crate::txn::SKIP_PROJECTION_RECONCILE.load(std::sync::atomic::Ordering::SeqCst)
            }
            #[cfg(not(any(test, feature = "test-faults")))]
            {
                false
            }
        };
        if projections_dirty && !skip_reconcile {
            coll.reconcile_projections()?;
            projections_dirty = false;
        }
        coll.replay_init();
        coll.build_ttl_index()?;

        // T9 Task 8: open-time counter recount. Read `counters_dirty` fresh
        // here (`self.stored_info`, not the `stored` snapshot captured before
        // `run_migration` ran above) — a migration on THIS collection may
        // have just persisted the flag (see `run_migration`'s `mutated_any`
        // tail). Only a collection that is itself a counted source (declares
        // counter targets) ever runs the recount or clears its own flag
        // here: a plain *target*'s flag — set by its own migration, to warn a
        // source that its maintained aggregate may have been lost — must
        // survive this collection's own open untouched. It is consumed by
        // the source's check below (the source opens later — targets are
        // always opened first) and otherwise cleared the normal way, at the
        // next successful `close`/`clean_shutdown`.
        // Disambiguates against `Collection::counter_target_names` (the
        // by-name entry point `Db::recount_counters` uses): `coll` here is
        // the concrete `S: ManagedOpen + Collection`, so both are in scope.
        let target_names = crate::access::ManagedOpen::counter_target_names(&*coll);
        let mut counters_dirty = self
            .stored_info(meta.name)
            .is_some_and(|i| i.counters_dirty);
        if !target_names.is_empty() {
            // Only the targets whose flag was actually set drove this
            // recount — a target with no flag was not migrated and consuming
            // its (absent) evidence would only widen the retirement window
            // below for no benefit.
            let dirty_targets: Vec<&'static str> = target_names
                .iter()
                .copied()
                .filter(|n| self.stored_info(n).is_some_and(|i| i.counters_dirty))
                .collect();
            if counters_dirty || !dirty_targets.is_empty() {
                // Disambiguates against `Collection::reconcile_counters` (the
                // by-name entry point `Db::recount_counters` uses): `coll`
                // here is the concrete `S: ManagedOpen + Collection`, so both
                // are in scope.
                let repaired = crate::access::ManagedOpen::reconcile_counters(&*coll)?;
                tracing::info!(collection = meta.name, repaired, "counters recounted");
                self.clear_counters_dirty(meta.name)?;
                counters_dirty = false;
                // The recount also consumed the *dirty* targets' evidence —
                // their flags say "my content moved under the counter", and
                // this walk just recomputed every one of their rows from the
                // source. Retiring them is what stops a migrated target from
                // re-triggering a full recount at every subsequent open,
                // forever: a target's flag has no marker behind it, so
                // `close`/`clean_shutdown` deliberately refuse to clear it.
                //
                // Noted for the close, not cleared here: several sources may
                // feed one target (different fields of the same row — the
                // topology `SchemaRegistry::finish`'s `(collection, field)`
                // duplicate rule exists to allow). Clearing at this open would
                // hide the target's flag from every *later* source's open in
                // the same session, and those sources would silently skip the
                // recount the migration owed them. Deferring to a successful
                // finalisation lets every source see the flag, and an unclean
                // exit simply recounts again — idempotent.
                lock(&self.counter_targets_consumed).extend(dirty_targets);
            }
        }
        self.save_info(&meta, coll.len() as u64, projections_dirty, counters_dirty)?;
        // Bind the counters_dirty marker now: the collection is constructed
        // but not yet published (no writer can reach it before this call
        // returns), and this is the only place holding both the collection and
        // `db.info`. A direct open with real counter targets never reaches
        // here — it is rejected earlier by `check_no_counters`.
        coll.bind_counter_dirty(self, meta.name);
        // …and, for a counted source, persist the evidence right away, still
        // before publication. The flag then stands for the whole session
        // rather than being taken lazily on the first delta: the write path
        // runs *after* the shard lock is released, so a mark that failed there
        // could no longer abort the write it was supposed to witness. Here a
        // failure fails the open, and a source that cannot record evidence
        // never takes a write. See `counter_dirty.rs`.
        if !target_names.is_empty() {
            crate::access::ManagedOpen::mark_counter_dirty(&*coll)?;
        }

        coll.register_tx_metadata(self)?;
        Ok(StagedManaged {
            meta,
            collection: coll,
            wire,
            _name_guard: name_guard,
        })
    }

    fn migrate_format_backend(
        &self,
        meta: &TreeMeta,
        plan: &MigrationPlan,
        descriptor: StorageDescriptor<'_>,
    ) -> DbResult<BackendMigrationOutcome> {
        let dir = self.tree_path(meta.name);
        match descriptor {
            StorageDescriptor::Bitcask {
                config, key_len, ..
            } => crate::format_migration::bitcask::migrate_bitcask(
                &dir,
                meta.name,
                plan.stored_version,
                plan,
                config,
                key_len,
            ),
            StorageDescriptor::Fixed {
                config,
                key_len,
                value_len,
                ..
            } => migrate_fixed(
                &dir,
                meta.name,
                plan.stored_version,
                plan,
                config,
                key_len,
                value_len,
            ),
            StorageDescriptor::Lsm => Err(DbError::FormatMigrationUnsupported {
                collection: meta.name,
                backend: "lsm",
            }),
        }
    }

    fn validate_format_backend_layout(
        &self,
        meta: &TreeMeta,
        plan: &MigrationPlan,
        descriptor: StorageDescriptor<'_>,
    ) -> DbResult<()> {
        let shard_count = match descriptor {
            StorageDescriptor::Bitcask { config, .. } => config.shard_count,
            StorageDescriptor::Fixed { config, .. } => config.shard_count,
            StorageDescriptor::Lsm => return Ok(()),
        };
        crate::format_migration::marker::configured_shards(
            &self.tree_path(meta.name),
            shard_count,
            plan.stored_version,
            plan.format_to_version,
        )?;
        Ok(())
    }

    fn guard_ambiguous_format_replay(
        &self,
        name: &'static str,
        target_fingerprint: u64,
        plan: &MigrationPlan,
    ) -> DbResult<()> {
        let pending = self.txlog().pending();
        let ambiguous = |op: &crate::txlog::record::TxOp| {
            op.name == name
                && plan.logged_fingerprint_is_ambiguous(op.schema_fingerprint, target_fingerprint)
        };
        let Some((first_record, first_op)) = pending.iter().find_map(|record| {
            record
                .ops
                .iter()
                .find(|op| ambiguous(op))
                .map(|op| (record, op))
        }) else {
            return Ok(());
        };
        if !self.tx_config().discard_pending_on_schema_mismatch {
            return Err(DbError::TxFormatAmbiguous {
                collection: name,
                tx_id: first_record.tx_id,
                fingerprint: first_op.schema_fingerprint,
            });
        }

        let mut discarded = 0usize;
        for record in pending {
            if record.ops.iter().any(&ambiguous) {
                self.txlog().mark_applied(record.tx_id);
                discarded += 1;
            }
        }
        // This is the durable admission barrier: all already-published sibling
        // buffers reach disk before the affected immutable txlog records can
        // retire. Only after it returns may physical migration begin.
        let _ = crate::txn::checkpoint::run(self)?;
        tracing::error!(
            collection = name,
            discarded,
            "dropped replay-ambiguous pending transactions before format migration"
        );
        Ok(())
    }

    fn remove_committed_format_marker(
        &self,
        name: &'static str,
        stored: Option<&CollectionInfo>,
    ) -> DbResult<()> {
        let Some(marker) = read_collection_marker(&self.tree_path(name))? else {
            return Ok(());
        };
        if stored.is_some_and(|info| info.version == marker.to) {
            remove_collection_marker(&self.tree_path(name))?;
        }
        Ok(())
    }

    fn commit_format_outcome(
        &self,
        name: &'static str,
        typ_hash: u64,
        plan: &MigrationPlan,
        claims: bool,
        outcome: &BackendMigrationOutcome,
    ) -> DbResult<()> {
        #[cfg(any(test, feature = "test-faults"))]
        crate::format_migration::failpoints::hit(
            crate::format_migration::failpoints::BEFORE_INFO_COMMIT,
        );
        self.update_collection_info(name, |entry| {
            entry.version = plan.format_to_version;
            entry.typ_hash = typ_hash;
            entry.gsn_high = entry.gsn_high.max(outcome.gsn_high);
            if entry.shard_gsn_high.len() < outcome.shard_gsn_high.len() {
                entry.shard_gsn_high.resize(outcome.shard_gsn_high.len(), 0);
            }
            for (index, high) in outcome.shard_gsn_high.iter().copied().enumerate() {
                entry.shard_gsn_high[index] = entry.shard_gsn_high[index].max(high);
            }
            entry.projections_dirty = true;
            entry.counters_dirty = true;
            entry.unique_dirty |= claims;
        })?;
        #[cfg(any(test, feature = "test-faults"))]
        crate::format_migration::failpoints::hit(
            crate::format_migration::failpoints::AFTER_INFO_COMMIT,
        );
        // The marker can only be removed after the in-memory persisted-state
        // handle observes the committed target version. This is a
        // post-condition assertion; `Persist::update` above owns the durable
        // write rather than this clone acting as an independent disk reread.
        let committed = self.stored_info(name).ok_or_else(|| {
            DbError::FormatMismatch(format!("missing committed format metadata for `{name}`"))
        })?;
        if committed.version != plan.format_to_version || committed.typ_hash != typ_hash {
            return Err(DbError::FormatMismatch(format!(
                "format metadata readback for `{name}` did not reach version {}",
                plan.format_to_version
            )));
        }
        #[cfg(any(test, feature = "test-faults"))]
        crate::format_migration::failpoints::hit(
            crate::format_migration::failpoints::BEFORE_MARKER_REMOVE,
        );
        remove_collection_marker(&self.tree_path(name))
    }

    /// Run pending schema migrations one step at a time.
    ///
    /// Each step: migrate data → durable flush → update `db.info` version. A crash
    /// after flush but before the version commit leaves `typ_hash =
    /// [`TYP_HASH_MIGRATING`] (sentinel `0`) on the intermediate step; the next
    /// open re-runs that step on already-migrated data — migration callbacks
    /// must be idempotent.
    fn run_migration<K, T>(
        &self,
        meta: &TreeMeta,
        stored: Option<&CollectionInfo>,
        migrations: &[TypedMigration<K, T>],
        mut migrate_fn: impl FnMut(&super::migration::TypedMigrationFn<K, T>) -> DbResult<usize>,
        flush_fn: impl Fn() -> DbResult<()>,
    ) -> DbResult<bool> {
        // First-open shortcut: nothing stored yet → no migration needed.
        let Some(stored) = stored else {
            return Ok(false);
        };

        // V-005: typ_hash drift at the same version (schema changed without bumping VERSION).
        if stored.version == meta.version && stored.typ_hash != meta.ty.h() {
            if stored.typ_hash == TYP_HASH_MIGRATING {
                tracing::warn!(
                    name = meta.name,
                    version = stored.version,
                    "typ_hash is the migration sentinel — skipping V-005 (interrupted multi-step migration)"
                );
            } else {
                return Err(DbError::SchemaMismatch {
                    name: meta.name.into(),
                    kind: SchemaMismatchKind::TypHash {
                        stored: stored.typ_hash,
                        expected: meta.ty.h(),
                    },
                });
            }
        }

        // V-006: downgrade (stored version is newer than the binary's version).
        if stored.version > meta.version {
            return Err(DbError::SchemaMismatch {
                name: meta.name.into(),
                kind: SchemaMismatchKind::Downgrade {
                    stored: stored.version,
                    requested: meta.version,
                },
            });
        }

        // Versions match → nothing to do.
        if stored.version == meta.version {
            return Ok(false);
        }

        // V-001 + V-002: walk one step at a time, committing each step's version
        // durably before advancing so a crash leaves the on-disk state resumable.
        let mut current = stored.version;
        let mut mutated_any = false;
        while current < meta.version {
            let (_, mfn) = migrations
                .iter()
                .find(|(v, _)| *v == current)
                .ok_or_else(|| DbError::SchemaMismatch {
                    name: meta.name.into(),
                    kind: SchemaMismatchKind::MissingStep { from: current },
                })?;
            // Migrations mutate persisted rows through the family apply path,
            // but must remain hook-free: schema upgrades neither require a
            // user transaction nor emit a change-stream queue row.
            let _stream_apply_guard = crate::txn_depth::StreamApplyGuard::enter();
            let mutated = migrate_fn(mfn)?;
            flush_fn()?;
            let step_hash = if current + 1 == meta.version {
                meta.ty.h()
            } else {
                TYP_HASH_MIGRATING
            };
            self.update_schema_version(meta.name, current + 1, step_hash)?;
            tracing::info!(mutated, from = current, to = current + 1, "migration step");
            current += 1;
            mutated_any = true;
        }
        // T9 Task 8: any collection whose migration ran at least one step is
        // marked dirty — it may itself be a counter SOURCE (its own migration
        // could have invalidated its contribution computation) or a counter
        // TARGET (the migration may have rewritten/lost the maintained
        // aggregate). This is the only durable evidence that survives to a
        // source's later open: `open_managed` computes migration need from
        // the collection being opened and this return value is otherwise
        // discarded, so without persisting the flag here the fact that a
        // migration ran at all would not outlive this call.
        if mutated_any {
            self.set_counters_dirty(meta.name, true)?;
        }
        Ok(mutated_any)
    }

    fn update_schema_version(&self, name: &str, version: u16, typ_hash: u64) -> DbResult<()> {
        self.update_collection_info(name, |entry| {
            entry.version = version;
            entry.typ_hash = typ_hash;
            // seq (collection length) is preserved; not updated here.
        })?;
        Ok(())
    }

    fn save_info(
        &self,
        meta: &TreeMeta,
        seq: u64,
        projections_dirty: bool,
        counters_dirty: bool,
    ) -> DbResult<()> {
        let typ_hash = meta.ty.h();
        self.update_collection_info(meta.name, |entry| {
            entry.version = meta.version;
            entry.typ_hash = typ_hash;
            entry.seq = seq;
            entry.projections_dirty = projections_dirty;
            entry.counters_dirty = counters_dirty;
        })?;
        Ok(())
    }

    fn update_collection_info(
        &self,
        name: &str,
        update: impl FnOnce(&mut CollectionInfo),
    ) -> DbResult<()> {
        self.db_info
            .update(|info| update(info.collections.entry(name.to_owned()).or_default()))?;
        Ok(())
    }

    fn set_projections_dirty(&self, name: &str, dirty: bool) -> DbResult<()> {
        self.db_info.update(|info| {
            let entry = info.collections.entry(name.to_owned()).or_default();
            entry.projections_dirty = dirty;
        })?;
        Ok(())
    }

    /// Clear `counters_dirty` for one collection. Called from `close` /
    /// `clean_shutdown`, only when every collection's flush succeeded this
    /// session — see the call sites for why.
    /// Clear every `counters_dirty` flag this session can prove it accounted
    /// for (T9), shared verbatim by [`close`](Self::close) and
    /// [`clean_shutdown`](Self::clean_shutdown) — both call it only after
    /// finalisation succeeded for every collection.
    ///
    /// Two sources of proof, and nothing else:
    ///
    /// - a collection whose own marker is set — a counted **source**, marked
    ///   when it was opened (`open_managed`);
    /// - a counter **target** whose flag an open-time recount consumed this
    ///   session (`counter_targets_consumed`). Without this half a migrated
    ///   target's flag would never be retired and every later open of its
    ///   source would recount again, forever.
    ///
    /// **Cross-session requirement this retirement imposes:** a migration
    /// touching a counted target must run in a session that opens **every**
    /// source of that target, not just the one the migration cares about.
    /// This function only sees the sources that opened *in this session*
    /// (`counter_targets_consumed` is populated by their open-time recount,
    /// above); the full set of a target's sources lives one layer up, in the
    /// schema registry, which this function has no access to. If target `T`
    /// is migrated and sources `S1`/`S2` maintain different fields of `T`, a
    /// session that opens only `S1` recounts `S1`, records `T` as consumed,
    /// and this function retires `T`'s flag at that session's clean close —
    /// silently, from `S2`'s perspective. `S2`, opening in a *later* session,
    /// finds no flag on `T` (already cleared) and never recounts, so its
    /// field keeps the migrated value (e.g. zeroed) forever. Escape hatch for
    /// a session that could not open every source:
    /// [`Db::recount_counters`](Self::recount_counters) on each source that
    /// was left out. See `docs/schema-metadata.md` ("Recovery:
    /// `counters_dirty`") and
    /// `docs/breaking-changes/26-08-09-armdb-engine-maintained-counters.md`
    /// (§9) for the full writeup.
    ///
    /// Both are *taken*: a second close proves nothing and clears nothing.
    /// Returns the first error, having attempted every entry.
    fn clear_consumed_counter_flags(&self, collections: &[Arc<dyn Collection>]) -> DbResult<()> {
        let mut first_err: Option<DbError> = None;
        let consumed = std::mem::take(&mut *lock(&self.counter_targets_consumed));
        for c in collections.iter() {
            if !c.reset_counter_dirty_marker() {
                continue;
            }
            if let Err(e) = self.clear_counters_dirty(c.name()) {
                tracing::warn!(collection = c.name(), error = %e, "clear counters_dirty failed");
                first_err.get_or_insert(e);
            }
        }
        for name in consumed {
            if let Err(e) = self.clear_counters_dirty(name) {
                tracing::warn!(collection = name, error = %e, "clear consumed target counters_dirty failed");
                first_err.get_or_insert(e);
            }
        }
        first_err.map_or(Ok(()), Err)
    }

    fn clear_counters_dirty(&self, name: &str) -> DbResult<()> {
        self.db_info.update(|info| {
            let entry = info.collections.entry(name.to_owned()).or_default();
            entry.counters_dirty = false;
        })?;
        Ok(())
    }

    /// Set `counters_dirty` for one collection (T9 Task 8). Called from
    /// `run_migration` when a migration step actually ran, so the fact
    /// survives past that call's own return value — see the call site.
    fn set_counters_dirty(&self, name: &str, dirty: bool) -> DbResult<()> {
        self.db_info.update(|info| {
            let entry = info.collections.entry(name.to_owned()).or_default();
            entry.counters_dirty = dirty;
        })?;
        Ok(())
    }

    /// Clear a completed namespace repair set in one metadata persistence
    /// operation. A persistence failure therefore cannot retire only a prefix
    /// of the sources whose repair depended on the same union scan.
    pub(crate) fn clear_unique_dirty(&self, names: &[&str]) -> DbResult<()> {
        self.db_info.update(|info| {
            for name in names {
                let entry = info.collections.entry((*name).to_owned()).or_default();
                entry.unique_dirty = false;
            }
        })?;
        Ok(())
    }

    /// Returns whether `run_migration` would walk at least one version step.
    fn migration_will_run(stored: Option<&CollectionInfo>, meta: &TreeMeta) -> bool {
        let Some(stored) = stored else {
            return false;
        };
        stored.version < meta.version
    }

    fn save_collection_len(&self, name: &str, len: u64) -> DbResult<()> {
        self.db_info.update(|info| {
            if let Some(ci) = info.collections.get_mut(name) {
                ci.seq = len;
            }
        })?;
        Ok(())
    }

    /// Borrowed read-only inspector over every collection opened through this `Db`.
    ///
    /// # Examples
    ///
    /// ```
    /// # use armdb::{CollectionMeta, Config, NoHook};
    /// # use armdb::armour::{Db, CountMode};
    /// # use armour_core::GetType;
    /// # use zerocopy::{FromBytes, Immutable, IntoBytes};
    /// # #[derive(Clone, Copy, Debug, PartialEq, GetType, FromBytes, Immutable, IntoBytes)]
    /// # #[repr(C)]
    /// # struct Item { n: u64 }
    /// # impl CollectionMeta for Item {
    /// #     type SelfId = [u8; 8];
    /// #     const NAME: &'static str = "items";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// # let dir = tempfile::tempdir().unwrap();
    /// # let db = Db::open_test(dir.path()).unwrap();
    /// # let _tree = db.open_zero_tree::<Item, { std::mem::size_of::<Item>() }, _>(Config::test(), NoHook, &[]).unwrap();
    /// let inspector = db.inspector();
    /// let count = inspector.count(Item::NAME, CountMode::Current).unwrap();
    /// assert_eq!(count, 0);
    /// ```
    ///
    /// Mutation is not available on the inspector facade:
    ///
    /// ```compile_fail
    /// # use armdb::{CollectionMeta, Config, NoHook};
    /// # use armdb::armour::Db;
    /// # use armour_core::GetType;
    /// # use zerocopy::{FromBytes, Immutable, IntoBytes};
    /// # #[derive(Clone, Copy, Debug, PartialEq, GetType, FromBytes, Immutable, IntoBytes)]
    /// # #[repr(C)]
    /// # struct Item { n: u64 }
    /// # impl CollectionMeta for Item {
    /// #     type SelfId = [u8; 8];
    /// #     const NAME: &'static str = "items";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// # let dir = tempfile::tempdir().unwrap();
    /// # let db = Db::open_test(dir.path()).unwrap();
    /// # let _tree = db.open_zero_tree::<Item, { std::mem::size_of::<Item>() }, _>(Config::test(), NoHook, &[]).unwrap();
    /// let inspector = db.inspector();
    /// inspector.put(Item::NAME, &[0u8; 8], &[]);
    /// ```
    ///
    /// Handlers and registries are crate-private and cannot be extracted:
    ///
    /// ```compile_fail
    /// # use armdb::{CollectionMeta, Config, NoHook};
    /// # use armdb::armour::Db;
    /// # use armour_core::GetType;
    /// # use zerocopy::{FromBytes, Immutable, IntoBytes};
    /// # #[derive(Clone, Copy, Debug, PartialEq, GetType, FromBytes, Immutable, IntoBytes)]
    /// # #[repr(C)]
    /// # struct Item { n: u64 }
    /// # impl CollectionMeta for Item {
    /// #     type SelfId = [u8; 8];
    /// #     const NAME: &'static str = "items";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// # let dir = tempfile::tempdir().unwrap();
    /// # let db = Db::open_test(dir.path()).unwrap();
    /// # let _tree = db.open_zero_tree::<Item, { std::mem::size_of::<Item>() }, _>(Config::test(), NoHook, &[]).unwrap();
    /// let inspector = db.inspector();
    /// let _handlers = inspector.db.inspection_handlers.clone();
    /// ```
    ///
    /// An inspector cannot outlive its `Db`:
    ///
    /// ```compile_fail
    /// # use armdb::{CollectionMeta, Config, NoHook};
    /// # use armdb::armour::{Db, Inspector};
    /// # use armour_core::GetType;
    /// # use zerocopy::{FromBytes, Immutable, IntoBytes};
    /// # #[derive(Clone, Copy, Debug, PartialEq, GetType, FromBytes, Immutable, IntoBytes)]
    /// # #[repr(C)]
    /// # struct Item { n: u64 }
    /// # impl CollectionMeta for Item {
    /// #     type SelfId = [u8; 8];
    /// #     const NAME: &'static str = "items";
    /// #     const VERSION: u16 = 1;
    /// # }
    /// fn leak_inspector() -> Inspector<'static> {
    ///     let dir = tempfile::tempdir().unwrap();
    ///     let db = Db::open_test(dir.path()).unwrap();
    ///     let _tree = db.open_zero_tree::<Item, { std::mem::size_of::<Item>() }, _>(Config::test(), NoHook, &[]).unwrap();
    ///     db.inspector()
    /// }
    /// ```
    pub fn inspector(&self) -> super::inspection::Inspector<'_> {
        super::inspection::Inspector { db: self }
    }

    /// Atomically publish a collection into the live registries.
    ///
    /// Lock order: `collections` → `inspection_handlers` → RPC `handlers`.
    /// Validates exact-name and RPC-hash uniqueness before any mutation.
    #[cfg(test)]
    fn publish_collection(
        &self,
        name: &'static str,
        collection: Arc<dyn Collection>,
        inspection: Arc<dyn InspectionHandler>,
        #[cfg(feature = "rpc")] rpc: Option<(u64, Arc<dyn super::handler::RpcHandler>)>,
    ) -> DbResult<()> {
        let mut collections = lock(&self.collections);
        let mut inspections = lock(&self.inspection_handlers);
        #[cfg(feature = "rpc")]
        let mut handlers = lock(&self.handlers);

        if collections.iter().any(|c| c.name() == name) {
            return Err(DbError::DuplicateCollection(name));
        }
        if inspections.contains_key(name) {
            return Err(DbError::DuplicateCollection(name));
        }
        #[cfg(feature = "rpc")]
        if let Some((hash, ref handler)) = rpc
            && let Some((_, existing)) = handlers.iter().find(|(k, _)| *k == hash)
        {
            tracing::error!(
                collection = name,
                conflicting = handler.name(),
                existing = existing.name(),
                hash,
                "xxh3-64 collection name collision — refusing to register a second \
                 handler under the same hash"
            );
            return Err(DbError::DuplicateCollection(
                "rpc handler hash collision (see log for the two collection names)",
            ));
        }

        collections.push(collection);
        inspections.insert(name, inspection);
        #[cfg(feature = "rpc")]
        if let Some((hash, handler)) = rpc {
            handlers.push((hash, handler));
        }
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn publish_collection_for_test(
        &self,
        name: &'static str,
        collection: Arc<dyn Collection>,
        inspection: Arc<dyn InspectionHandler>,
        #[cfg(feature = "rpc")] rpc: Option<Arc<dyn super::handler::RpcHandler>>,
        forced_hash: Option<u64>,
    ) -> DbResult<()> {
        #[cfg(feature = "rpc")]
        let rpc = rpc.map(|handler| {
            let hash = forced_hash.unwrap_or_else(|| xxhash_rust::xxh3::xxh3_64(name.as_bytes()));
            (hash, handler)
        });
        #[cfg(not(feature = "rpc"))]
        let _ = forced_hash;
        self.publish_collection(
            name,
            collection,
            inspection,
            #[cfg(feature = "rpc")]
            rpc,
        )
    }

    fn stage_to_adapter<'a, S, W>(
        &self,
        staged: StagedManaged<'a, S, W>,
        access: ManagedCollectionAccess,
    ) -> StagedAdapter<'a>
    where
        S: crate::access::Access + Collection + 'static,
        W: crate::access::Wire<S::Val>,
    {
        let StagedManaged {
            meta,
            collection,
            wire,
            _name_guard,
        } = staged;
        let adapter = Arc::new(crate::access::Adapter::new(
            &meta,
            collection.clone(),
            wire,
            self.seq.clone(),
        ));
        let inspection: Arc<dyn InspectionHandler> = adapter.clone();
        let collection: Arc<dyn Collection> = collection;
        #[cfg(not(feature = "rpc"))]
        let _ = access;
        #[cfg(feature = "rpc")]
        let rpc = (access == ManagedCollectionAccess::Public).then(|| {
            (
                xxhash_rust::xxh3::xxh3_64(meta.name.as_bytes()),
                adapter as Arc<dyn super::handler::RpcHandler>,
            )
        });
        StagedAdapter {
            name: meta.name,
            collection,
            inspection,
            #[cfg(feature = "rpc")]
            rpc,
            _name_guard,
        }
    }

    /// Publish a fully staged group while holding every name reservation.
    /// All conflict checks run before the first registry mutation, so a failed
    /// group leaves no collection reachable from this `Db`.
    fn publish_staged_adapters(&self, staged: Vec<StagedAdapter<'_>>) -> DbResult<()> {
        let mut collections = lock(&self.collections);
        let mut inspections = lock(&self.inspection_handlers);
        #[cfg(feature = "rpc")]
        let mut handlers = lock(&self.handlers);

        for adapter in &staged {
            if collections.iter().any(|c| c.name() == adapter.name)
                || inspections.contains_key(adapter.name)
                || staged
                    .iter()
                    .filter(|other| other.name == adapter.name)
                    .count()
                    != 1
            {
                return Err(DbError::DuplicateCollection(adapter.name));
            }
            #[cfg(feature = "rpc")]
            if let Some((hash, ref handler)) = adapter.rpc
                && (handlers.iter().any(|(existing, _)| *existing == hash)
                    || staged
                        .iter()
                        .filter(|other| {
                            other
                                .rpc
                                .as_ref()
                                .is_some_and(|(other_hash, _)| *other_hash == hash)
                        })
                        .count()
                        != 1)
            {
                tracing::error!(
                    collection = adapter.name,
                    conflicting = handler.name(),
                    hash,
                    "RPC handler collision while publishing managed collection group"
                );
                return Err(DbError::DuplicateCollection(
                    "rpc handler hash collision (see log for the two collection names)",
                ));
            }
        }

        let names: Vec<_> = staged.iter().map(|adapter| adapter.name).collect();
        for adapter in staged {
            collections.push(adapter.collection);
            inspections.insert(adapter.name, adapter.inspection);
            #[cfg(feature = "rpc")]
            if let Some((hash, handler)) = adapter.rpc {
                handlers.push((hash, handler));
            }
        }
        #[cfg(feature = "rpc")]
        drop(handlers);
        drop(inspections);
        drop(collections);
        for name in names {
            self.note_tx_replay_done(name);
        }
        Ok(())
    }
}

/// Assembles the options `Db::open*` presets fix.
///
/// Every constructor on [`Db`] is a preset over this builder; reach for it when
/// you need a combination none of them offers — most often background worker
/// intervals **and** a [`TxConfig`](crate::txn::TxConfig) at the same time.
///
/// ```no_run
/// # use armdb::armour::{Db, WorkerIntervals};
/// # use armdb::TxConfig;
/// # fn main() -> armdb::DbResult<()> {
/// let db = Db::builder("/var/lib/app/armdb")
///     .workers(WorkerIntervals::default())
///     .tx_config(TxConfig::default())
///     .open()?;
/// # Ok(()) }
/// ```
pub struct DbBuilder {
    path: std::path::PathBuf,
    workers: WorkerIntervals,
    sync_persist: bool,
    tx_config: crate::txn::TxConfig,
}

impl DbBuilder {
    /// Background worker intervals and TTL sweep budget.
    #[must_use]
    pub fn workers(mut self, workers: WorkerIntervals) -> Self {
        self.workers = workers;
        self
    }

    /// Transaction log configuration — limits, checkpoint sizing, and the log
    /// encryption key.
    #[must_use]
    pub fn tx_config(mut self, tx_config: crate::txn::TxConfig) -> Self {
        self.tx_config = tx_config;
        self
    }

    /// The preset behind [`Db::open_test`]: no compaction, flush or TTL worker,
    /// and `db.info`/`db.user` persisted asynchronously.
    ///
    /// It sets **two** things, so calls after it override only what they name —
    /// `.for_tests().workers(w)` keeps the asynchronous persistence and brings
    /// the workers back.
    #[must_use]
    pub fn for_tests(mut self) -> Self {
        self.workers = WorkerIntervals {
            compaction: None,
            flush: None,
            ttl_sweep: None,
            ttl_budget: DEFAULT_TTL_BUDGET,
        };
        self.sync_persist = false;
        self
    }

    #[doc = include_str!("../../docs/exclusive-access.md")]
    #[doc = include_str!("../../docs/exclusive-access-root.md")]
    pub fn open(self) -> DbResult<Db> {
        Db::open_with_options(self.path, self.workers, self.sync_persist, self.tx_config)
    }
}

#[cfg(feature = "rpc")]
impl Db {
    /// Look up a registered RPC handler by collection name.
    ///
    /// `name` is the collection's `CollectionMeta::NAME`; returns `None` if no
    /// handler is registered for it.
    pub fn rpc_handler_for(&self, name: &str) -> Option<Arc<dyn super::handler::RpcHandler>> {
        let h = xxhash_rust::xxh3::xxh3_64(name.as_bytes());
        lock(&self.handlers)
            .iter()
            .find(|(k, _)| *k == h)
            .map(|(_, v)| v.clone())
    }

    /// Build the RPC routing table from all registered handlers.
    pub fn build_tree_map(&self) -> super::rpc::TreeMap {
        Arc::new(
            lock(&self.handlers)
                .iter()
                .map(|(h, handler)| (*h, handler.clone()))
                .collect(),
        )
    }
}

impl Drop for Db {
    fn drop(&mut self) {
        self.stop_workers();
        if let Err(error) = self.close() {
            tracing::error!(
                error = %error,
                "Db drop: close failed; durability of the last writes is not guaranteed"
            );
        }
        self.close_all_watches();
    }
}

#[cfg(all(test, feature = "rapira-codec"))]
mod tests {
    use super::*;
    use armour_core::GetType;
    use rapira::Rapira;
    use tempfile::tempdir;

    use crate::armour::migration::TypedMigration;
    use crate::{MigrateAction, NoHook, RapiraCodec};

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct User {
        id: u64,
        name: String,
    }

    impl CollectionMeta for User {
        type SelfId = [u8; 8];
        const NAME: &'static str = "iterable_toggle_users";
        const VERSION: u16 = 1;
    }

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct StagedFirst;
    impl CollectionMeta for StagedFirst {
        type SelfId = [u8; 8];
        const NAME: &'static str = "staged_pair_first";
    }

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct StagedSecond;
    impl CollectionMeta for StagedSecond {
        type SelfId = [u8; 8];
        const NAME: &'static str = "staged_pair_second";
    }

    fn key(id: u64) -> [u8; 8] {
        id.to_be_bytes()
    }

    fn demo_user() -> User {
        User {
            id: 1,
            name: "alice".into(),
        }
    }

    // AI-10: opening the same collection name twice on one Db must error — two
    // in-memory indexes over the same shard files would silently drop each
    // other's updates.
    #[test]
    fn open_same_collection_twice_errors() {
        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let _first = db
            .open_typed_map::<User, RapiraCodec, _>(Config::test(), NoHook, &[])
            .expect("first open succeeds");
        let err = db
            .open_typed_map::<User, RapiraCodec, _>(Config::test(), NoHook, &[])
            .err()
            .expect("second open of the same name must fail");
        assert!(
            matches!(err, DbError::DuplicateCollection("iterable_toggle_users")),
            "got {err:?}"
        );
    }

    #[test]
    fn failed_after_stage_callback_leaves_pair_unpublished_and_retryable() {
        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let result = db.open_managed_pair(
            |meta| {
                TypedMap::<[u8; 8], StagedFirst, RapiraCodec, TtlHook<StagedFirst, NoHook>>::open_inner(
                    db.tree_path(meta.name), Config::test(), RapiraCodec, TtlHook::new(NoHook),
                )
            },
            crate::access::CodecWire(RapiraCodec),
            &[],
            StorageDescriptor::Bitcask {
                config: &Config::test(),
                key_len: 8,
                claims: false,
            },
            |meta| {
                TypedMap::<[u8; 8], StagedSecond, RapiraCodec, TtlHook<StagedSecond, NoHook>>::open_inner(
                    db.tree_path(meta.name), Config::test(), RapiraCodec, TtlHook::new(NoHook),
                )
            },
            crate::access::CodecWire(RapiraCodec),
            &[],
            StorageDescriptor::Bitcask {
                config: &Config::test(),
                key_len: 8,
                claims: false,
            },
            |_, _| Err(DbError::Client("staged cleanup failed")),
            ManagedCollectionAccess::Public,
            ManagedCollectionAccess::Public,
        );
        assert!(matches!(
            result,
            Err(DbError::Client("staged cleanup failed"))
        ));
        assert!(lock(&db.collections).is_empty());
        db.open_typed_map::<StagedFirst, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        db.open_typed_map::<StagedSecond, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
    }

    #[cfg(feature = "var-collections")]
    mod format_pair_descriptor_tests {
        use super::*;

        const FIXED_NAME: &str = "format_pair_fixed_header";
        const PAYLOAD_NAME: &str = "format_pair_bitcask_payload";

        #[derive(
            Clone,
            Copy,
            Debug,
            Eq,
            PartialEq,
            zerocopy::FromBytes,
            zerocopy::IntoBytes,
            zerocopy::Immutable,
        )]
        #[repr(C)]
        struct FixedHeaderV1 {
            value: u32,
        }

        impl GetType for FixedHeaderV1 {
            const TYPE: armour_core::Typ = <u32 as GetType>::TYPE;
        }

        impl CollectionMeta for FixedHeaderV1 {
            type SelfId = [u8; 8];
            const NAME: &'static str = FIXED_NAME;
            const VERSION: u16 = 1;
        }

        #[derive(
            Clone,
            Copy,
            Debug,
            Eq,
            PartialEq,
            zerocopy::FromBytes,
            zerocopy::IntoBytes,
            zerocopy::Immutable,
        )]
        #[repr(C)]
        struct FixedHeaderV2 {
            value: u64,
        }

        impl GetType for FixedHeaderV2 {
            const TYPE: armour_core::Typ = <u64 as GetType>::TYPE;
        }

        fn widen_header(
            _key: &[u8],
            old: &[u8],
            out: &mut Vec<u8>,
        ) -> DbResult<crate::FormatAction> {
            let value = u32::from_ne_bytes(old.try_into().expect("v1 header width"));
            out.extend_from_slice(&u64::from(value).to_ne_bytes());
            Ok(crate::FormatAction::Write)
        }

        const HEADER_FORMAT: crate::FormatMigration = crate::FormatMigration::raw_checked(
            1,
            &FixedHeaderV1::TYPE,
            &FixedHeaderV2::TYPE,
            widen_header,
        );

        impl CollectionMeta for FixedHeaderV2 {
            type SelfId = [u8; 8];
            const NAME: &'static str = FIXED_NAME;
            const VERSION: u16 = 2;
            const FORMAT_MIGRATIONS: &'static [crate::FormatMigration] = &[HEADER_FORMAT];
        }

        #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
        struct PayloadV1 {
            bytes: Vec<u8>,
        }

        impl CollectionMeta for PayloadV1 {
            type SelfId = [u8; 8];
            const NAME: &'static str = PAYLOAD_NAME;
            const VERSION: u16 = 1;
        }

        #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
        struct PayloadV2 {
            bytes: Vec<u8>,
            generation: u16,
        }

        const PAYLOAD_FORMAT: crate::FormatMigration = crate::format_step!(
            1,
            PayloadV1 => PayloadV2,
            RapiraCodec,
            |old| Ok::<_, DbError>(Some(PayloadV2 {
                bytes: old.bytes,
                generation: 2,
            }))
        );

        impl CollectionMeta for PayloadV2 {
            type SelfId = [u8; 8];
            const NAME: &'static str = PAYLOAD_NAME;
            const VERSION: u16 = 2;
            const FORMAT_MIGRATIONS: &'static [crate::FormatMigration] = &[PAYLOAD_FORMAT];
        }

        #[test]
        fn format_migration_paired_fixed_header_and_bitcask_payload_routes_descriptors() {
            let dir = tempdir().expect("tempdir");
            let key = 21u64.to_be_bytes();
            let payload_bytes = vec![0x3c; 16 * 1024 + 9];
            let db = Db::open_test(dir.path()).expect("open v1 db");
            let header = db
                .open_zero_tree_fixed::<FixedHeaderV1, { std::mem::size_of::<FixedHeaderV1>() }, _>(
                    crate::FixedConfig::test(),
                    NoHook,
                    &[],
                )
                .expect("open v1 fixed header");
            let payload = db
                .open_var_typed_tree::<PayloadV1, RapiraCodec, _>(Config::test(), NoHook, &[])
                .expect("open v1 bitcask payload");
            header
                .put(&key, &FixedHeaderV1 { value: 21 })
                .expect("seed fixed header");
            payload
                .put(
                    &key,
                    &PayloadV1 {
                        bytes: payload_bytes.clone(),
                    },
                )
                .expect("seed bitcask payload");
            drop(header);
            drop(payload);
            db.close().expect("close v1 db");
            drop(db);

            let db = Db::open_test(dir.path()).expect("reopen v2 db");
            let fixed_config = crate::FixedConfig::test();
            let fixed_descriptor_config = fixed_config.clone();
            let payload_config = Config::test();
            let payload_descriptor_config = payload_config.clone();
            let (header, payload) = db
                .open_managed_pair(
                    |meta| {
                        crate::ZeroTree::<
                            [u8; 8],
                            { std::mem::size_of::<FixedHeaderV2>() },
                            FixedHeaderV2,
                            TtlHook<FixedHeaderV2, NoHook>,
                            crate::durability::Fixed,
                        >::open_with_hook_inner(
                            db.tree_path(meta.name),
                            fixed_config,
                            TtlHook::new(NoHook),
                        )
                    },
                    crate::access::ZeroWire::<{ std::mem::size_of::<FixedHeaderV2>() }>,
                    &[],
                    StorageDescriptor::Fixed {
                        config: &fixed_descriptor_config,
                        key_len: std::mem::size_of::<[u8; 8]>(),
                        value_len: std::mem::size_of::<FixedHeaderV2>() as u16,
                        claims: false,
                    },
                    |meta| {
                        crate::VarTypedTree::<
                            [u8; 8],
                            PayloadV2,
                            RapiraCodec,
                            TtlHook<PayloadV2, NoHook>,
                        >::open_hooked_inner(
                            db.tree_path(meta.name),
                            payload_config,
                            RapiraCodec,
                            TtlHook::new(NoHook),
                        )
                    },
                    crate::access::CodecWire(RapiraCodec),
                    &[],
                    StorageDescriptor::Bitcask {
                        config: &payload_descriptor_config,
                        key_len: std::mem::size_of::<[u8; 8]>(),
                        claims: false,
                    },
                    |_, _| Ok(()),
                    ManagedCollectionAccess::Internal,
                    ManagedCollectionAccess::Internal,
                )
                .expect("migrate paired managed collections");

            assert_eq!(header.get(&key), Some(FixedHeaderV2 { value: 21 }));
            assert_eq!(
                payload.get(&key),
                Some(PayloadV2 {
                    bytes: payload_bytes,
                    generation: 2,
                })
            );
        }
    }

    // T10 review F3: a hand-assembled claim carrier (never touching
    // `SchemaRegistry::unique_index_on`, whose own `const { assert!(...) }`
    // never runs) must still be refused at managed open when the source's own
    // key does not fit a unique namespace owner row — mirrors
    // `open_managed_rejects_a_hand_assembled_ttl_carrying_claim_carrier` in
    // `tests/unique_namespace.rs`, but for the key-width guard instead of the
    // TTL one.
    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        Hash,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
        zerocopy::KnownLayout,
    )]
    #[repr(C)]
    struct OverCapKey([u8; crate::unique::UNIQUE_OWNER_KEY_CAP + 8]);
    crate::impl_key_zerocopy!(
        OverCapKey,
        armour_core::KeyScheme::Typed(&[armour_core::KeyType::Array(
            crate::unique::UNIQUE_OWNER_KEY_CAP + 8
        )])
    );

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct WideKeyUser {
        name: String,
    }
    impl CollectionMeta for WideKeyUser {
        type SelfId = OverCapKey;
        const NAME: &'static str = "f3_wide_key_users";
    }

    #[test]
    fn open_managed_rejects_a_hand_assembled_carrier_whose_key_is_too_wide() {
        let (_dir, db, ns) = crate::unique_tests::open_claim_ns();

        let hook = crate::projection::Projected::new(crate::IndexTargets::new(), |_, _, _| Ok(()))
            .and_unique(
                crate::UniqueTargets::new().with(&ns),
                move |_k, _v, sink| sink.claim(&ns, crate::unique_tests::ClaimFwdKey::new(1)),
            );

        let result = db.open_typed_map::<WideKeyUser, RapiraCodec, _>(Config::test(), hook, &[]);
        let Err(err) = result else {
            panic!("a claim carrier whose source key exceeds the owner capacity must be rejected");
        };
        assert!(
            matches!(
                err,
                DbError::ClaimingSourceKeyTooWide {
                    collection: "f3_wide_key_users"
                }
            ),
            "expected ClaimingSourceKeyTooWide, got {err:?}"
        );
    }

    // T10 review F5: two distinct source collections declaring claims into the
    // same namespace under the *same* owner tag must be refused — the raw
    // `Projected::and_unique` path has no `SchemaRegistry` in the picture, so
    // nothing else would ever run this check for it.
    // `register_claim_owner_tag` (not the name-driven `check_claim_owner_tags`)
    // is exercised directly with a hand-picked tag, mirroring
    // `SchemaRegistry`'s own `owner_tag_tests`: a genuine 64-bit FNV-1a
    // collision between two real collection names cannot be constructed, so
    // testing only the public, name-driven path would test nothing.
    #[test]
    fn claim_owner_tag_collision_between_two_sources_is_rejected() {
        let (_dir, db, ns) = crate::unique_tests::open_claim_ns();
        let targets = crate::UniqueTargets::new().with(&ns);

        db.register_claim_owner_tag(42, "src_a", &targets)
            .expect("first source registers cleanly");
        // Idempotent for the *same* source under the same tag.
        db.register_claim_owner_tag(42, "src_a", &targets)
            .expect("re-registering the same source is not a collision");

        let err = db
            .register_claim_owner_tag(42, "src_b", &targets)
            .expect_err("a different source under the same tag must collide");
        assert!(
            matches!(
                err,
                DbError::ClaimOwnerTagCollision {
                    namespace: "claim_forwarding_test_ns",
                    a: "src_a",
                    b: "src_b",
                }
            ),
            "expected ClaimOwnerTagCollision, got {err:?}"
        );
    }

    // P1-5: two concurrent opens of the same name race the check-then-register
    // window. Exactly one must win; the rest get DuplicateCollection. Without the
    // atomic name reservation every racer passes the (still-empty) uniqueness
    // check and builds its own index over the same shard files.
    #[test]
    fn concurrent_open_same_name_exactly_one_wins() {
        use std::sync::{Arc, Barrier};
        let dir = tempdir().unwrap();
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        const N: usize = 8;
        let barrier = Arc::new(Barrier::new(N));
        let handles: Vec<_> = (0..N)
            .map(|_| {
                let db = Arc::clone(&db);
                let barrier = Arc::clone(&barrier);
                std::thread::spawn(move || {
                    barrier.wait();
                    db.open_typed_map::<User, RapiraCodec, _>(Config::test(), NoHook, &[])
                        .is_ok()
                })
            })
            .collect();
        let wins = handles
            .into_iter()
            .map(|h| h.join().unwrap())
            .filter(|&ok| ok)
            .count();
        assert_eq!(
            wins, 1,
            "expected exactly one concurrent open to win, got {wins}"
        );
    }

    // AI-10: a user collection may not claim an internal reserved name.
    #[test]
    fn open_reserved_name_errors() {
        #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
        struct SeqNamed {
            x: u64,
        }
        impl CollectionMeta for SeqNamed {
            type SelfId = [u8; 8];
            const NAME: &'static str = "__seq";
            const VERSION: u16 = 1;
        }

        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let err = db
            .open_typed_map::<SeqNamed, RapiraCodec, _>(Config::test(), NoHook, &[])
            .err()
            .expect("reserved name must be rejected");
        assert!(
            matches!(err, DbError::DuplicateCollection("__seq")),
            "got {err:?}"
        );
    }

    // AI-12: a failing `Serialize` in metadata must surface as a `DbError`, not a
    // panic under the Persist lock.
    #[test]
    fn set_metadata_serialize_failure_returns_err_not_panic() {
        use serde::de::{Deserialize, Deserializer};
        use serde::ser::{Error as _, Serialize, Serializer};

        #[derive(Debug)]
        struct FailSer;
        impl Serialize for FailSer {
            fn serialize<S: Serializer>(&self, _s: S) -> std::result::Result<S::Ok, S::Error> {
                Err(S::Error::custom("intentional serialize failure"))
            }
        }
        impl<'de> Deserialize<'de> for FailSer {
            fn deserialize<D: Deserializer<'de>>(_d: D) -> std::result::Result<Self, D::Error> {
                Ok(FailSer)
            }
        }

        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let res = db.set_metadata::<FailSer>(|_current| FailSer);
        assert!(
            matches!(res, Err(DbError::FormatMismatch(_))),
            "expected FormatMismatch, got {res:?}"
        );
    }

    #[test]
    fn db_map_iterable_toggles_across_reopen() {
        let dir = tempdir().unwrap();
        {
            let db = Db::open_test(dir.path()).unwrap();
            let users = db
                .open_typed_map::<User, RapiraCodec, _>(Config::test(), NoHook, &[])
                .unwrap();
            assert!(users.iter_view().is_none());
            users.put(&key(1), demo_user()).unwrap();
            db.close().unwrap();
        }
        let db = Db::open_test(dir.path()).unwrap();
        let cfg = Config::balanced()
            .shard_count(2)
            .hints(true)
            .iterable(true)
            .build();
        let users = db
            .open_typed_map::<User, RapiraCodec, _>(cfg, NoHook, &[])
            .unwrap();
        assert!(users.iter_view().is_some());
        assert_eq!(users.iter_view().unwrap().iter().count(), 1);
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
    )]
    #[repr(C)]
    struct FxCounter {
        n: u32,
    }

    impl CollectionMeta for FxCounter {
        type SelfId = [u8; 8];
        const NAME: &'static str = "fixed_reversed_db";
        const VERSION: u16 = 1;
    }

    // A distinct name for the map so tree and map don't alias the same directory
    // on one Db (rejected since AI-10 — two indexes over the same shard files).
    impl CollectionMeta for FxCounterMap {
        type SelfId = [u8; 8];
        const NAME: &'static str = "fixed_reversed_db_map";
        const VERSION: u16 = 1;
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
    )]
    #[repr(C)]
    struct FxCounterMap {
        n: u32,
    }

    #[test]
    fn db_fixed_openers_honor_reversed() {
        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();

        // Tree: default DESC.
        let tree = db
            .open_zero_tree_fixed::<FxCounter, 4, _>(FixedConfig::test(), NoHook, &[])
            .unwrap();
        for i in 1u64..=3 {
            tree.put(&i.to_be_bytes(), &FxCounter { n: i as u32 })
                .unwrap();
        }
        let tk: Vec<u64> = tree.iter().map(|(k, _)| u64::from_be_bytes(k)).collect();
        assert_eq!(tk, [3, 2, 1], "Db fixed tree default DESC");

        // Map: ASC via reversed=false (+ iterable for iter_view).
        let map_cfg = FixedConfig {
            iterable: true,
            reversed: false,
            ..FixedConfig::test()
        };
        let map = db
            .open_zero_map_fixed::<FxCounterMap, 4, _>(map_cfg, NoHook, &[])
            .unwrap();
        for i in 1u64..=3 {
            map.put(&i.to_be_bytes(), &FxCounterMap { n: i as u32 })
                .unwrap();
        }
        let mk: Vec<u64> = map
            .iter_view()
            .unwrap()
            .keys()
            .map(u64::from_be_bytes)
            .collect();
        assert_eq!(mk, [1, 2, 3], "Db fixed map reversed=false ASC");
    }

    /// D12/AI-01: flush_fn, который opener передаёт в run_migration, обязан
    /// доходить до fsync-пути (Engine::flush), а не только flush_buffers.
    /// Счётчик глобальный: при параллельном cargo test другие тесты могут его
    /// инкрементить (ложно-зелёный риск); под nextest (процесс-на-тест) — точно.
    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct FlushMigItem {
        value: i64,
    }
    impl CollectionMeta for FlushMigItem {
        type SelfId = [u8; 8];
        const NAME: &'static str = "flush_mig_items";
        const VERSION: u16 = 2;
    }

    fn keep_flush(_k: &[u8; 8], _v: &FlushMigItem) -> MigrateAction<FlushMigItem> {
        MigrateAction::Keep
    }

    #[test]
    fn migration_flush_fn_reaches_engine_durable_flush() {
        use std::sync::atomic::Ordering;
        let dir = tempdir().unwrap();
        // Phase 1: v2-данные, патч stored.version → 1.
        {
            let db = Db::open_test(dir.path()).unwrap();
            let tree = db
                .open_typed_tree::<FlushMigItem, RapiraCodec, _>(Config::test(), NoHook, &[])
                .unwrap();
            tree.put(&1u64.to_be_bytes(), FlushMigItem { value: 7 })
                .unwrap();
            db.db_info
                .update(|info| {
                    info.collections
                        .get_mut(FlushMigItem::NAME)
                        .unwrap()
                        .version = 1;
                })
                .unwrap();
        }
        let before = crate::engine::ENGINE_DURABLE_FLUSHES.load(Ordering::Relaxed);
        // Phase 2: reopen с миграцией (1 → 2).
        {
            let db = Db::open_test(dir.path()).unwrap();
            let migrations: &[TypedMigration<[u8; 8], FlushMigItem>] = &[(1u16, keep_flush)];
            let _tree = db
                .open_typed_tree::<FlushMigItem, RapiraCodec, _>(Config::test(), NoHook, migrations)
                .unwrap();
        }
        let after = crate::engine::ENGINE_DURABLE_FLUSHES.load(Ordering::Relaxed);
        assert!(
            after > before,
            "migration flush_fn must hit Engine::flush (fsync)"
        );
    }

    /// D12/AI-01 (zero opener): flush_fn у zero-opener'ов — `|| tree.flush()`,
    /// и durable он только потому, что inherent `ZeroTree::flush` затеняет
    /// трейтовый `Collection::flush` (flush_buffers + sync_hints, без fsync на
    /// Bitcask). Тест фиксирует это разрешение метода: пере-разрешение в
    /// трейтовый вариант снова откроет AI-01 без ошибки компиляции.
    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        zerocopy::FromBytes,
        zerocopy::IntoBytes,
        zerocopy::Immutable,
    )]
    #[repr(C)]
    struct FlushMigZero {
        n: u32,
    }
    impl CollectionMeta for FlushMigZero {
        type SelfId = [u8; 8];
        const NAME: &'static str = "flush_mig_zero";
        const VERSION: u16 = 2;
    }

    fn keep_flush_zero(_k: &[u8; 8], _v: &FlushMigZero) -> MigrateAction<FlushMigZero> {
        MigrateAction::Keep
    }

    #[test]
    fn zero_opener_migration_flush_fn_reaches_engine_durable_flush() {
        use std::sync::atomic::Ordering;
        const V: usize = std::mem::size_of::<FlushMigZero>();
        let dir = tempdir().unwrap();
        // Phase 1: v2-данные, патч stored.version → 1.
        {
            let db = Db::open_test(dir.path()).unwrap();
            let tree = db
                .open_zero_tree::<FlushMigZero, V, _>(Config::test(), NoHook, &[])
                .unwrap();
            tree.put(&1u64.to_be_bytes(), &FlushMigZero { n: 7 })
                .unwrap();
            db.db_info
                .update(|info| {
                    info.collections
                        .get_mut(FlushMigZero::NAME)
                        .unwrap()
                        .version = 1;
                })
                .unwrap();
        }
        let before = crate::engine::ENGINE_DURABLE_FLUSHES.load(Ordering::Relaxed);
        // Phase 2: reopen с миграцией (1 → 2).
        {
            let db = Db::open_test(dir.path()).unwrap();
            let migrations: &[TypedMigration<[u8; 8], FlushMigZero>] = &[(1u16, keep_flush_zero)];
            let _tree = db
                .open_zero_tree::<FlushMigZero, V, _>(Config::test(), NoHook, migrations)
                .unwrap();
        }
        let after = crate::engine::ENGINE_DURABLE_FLUSHES.load(Ordering::Relaxed);
        assert!(
            after > before,
            "zero opener flush_fn must resolve to the durable inherent ZeroTree::flush (Engine::flush fsync)"
        );
    }

    /// D6: завершённая миграция пишет реальный `typ_hash` на финальном шаге,
    /// промежуточные шаги — sentinel. `run_migration` проверяется напрямую,
    /// до `save_info` — иначе self-heal маскирует регрессию «всегда sentinel».
    #[test]
    fn run_migration_final_step_writes_real_typ_hash() {
        #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
        struct TwoStep {
            value: i64,
        }
        impl CollectionMeta for TwoStep {
            type SelfId = [u8; 8];
            const NAME: &'static str = "run_mig_two_step";
            const VERSION: u16 = 3;
        }
        fn keep(_k: &[u8; 8], _v: &TwoStep) -> MigrateAction<TwoStep> {
            MigrateAction::Keep
        }

        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let meta = TreeMeta::of::<TwoStep>();
        // version != meta.version → V-005 не срабатывает, typ_hash не важен.
        let stored = CollectionInfo {
            version: 1,
            typ_hash: meta.ty.h(),
            seq: 0,
            projections_dirty: false,
            counters_dirty: false,
            ..CollectionInfo::default()
        };
        let migrations: &[TypedMigration<[u8; 8], TwoStep>] = &[(1, keep), (2, keep)];

        // Хэш в db.info, видимый в начале каждого шага: на шаге 2 виден коммит
        // шага 1 (промежуточный → sentinel); до шага 1 записи ещё нет.
        let mut seen = Vec::new();
        let mutated = db
            .run_migration(
                &meta,
                Some(&stored),
                migrations,
                |_mfn| {
                    if let Some(ci) = db.db_info.cloned().collections.get(TwoStep::NAME) {
                        seen.push(ci.typ_hash);
                    }
                    Ok(1)
                },
                || Ok(()),
            )
            .unwrap();
        assert!(mutated);
        assert_eq!(seen, vec![TYP_HASH_MIGRATING]);

        let info = db.db_info.cloned();
        let ci = info.collections.get(TwoStep::NAME).unwrap();
        assert_eq!(ci.version, 3);
        assert_eq!(
            ci.typ_hash,
            meta.ty.h(),
            "финальный шаг обязан записать реальный typ_hash, не sentinel"
        );
        assert_ne!(ci.typ_hash, TYP_HASH_MIGRATING);
    }

    #[test]
    fn db_open_twice_same_path_errors() {
        let dir = tempdir().unwrap();
        let _first = Db::open_test(dir.path()).expect("first open");
        let err = Db::open_test(dir.path())
            .err()
            .expect("second open of a live root must fail");
        assert!(matches!(err, DbError::Locked { .. }), "got {err:?}");
    }

    #[test]
    fn db_lock_released_on_drop() {
        let dir = tempdir().unwrap();
        let first = Db::open_test(dir.path()).expect("first open");
        drop(first);
        Db::open_test(dir.path()).expect("reopen after drop must succeed");
    }

    // The lease must outlive the background workers: releasing it before the
    // compactor/flusher threads join would let the next process in mid-write.
    #[test]
    fn db_lock_released_after_background_workers() {
        let dir = tempdir().unwrap();
        let db = Db::open_with_intervals(
            dir.path(),
            Some(std::time::Duration::from_millis(10)),
            Some(std::time::Duration::from_millis(10)),
        )
        .expect("open with background workers");
        drop(db);
        Db::open_test(dir.path()).expect("reopen immediately after workers stop");
    }

    #[test]
    fn db_lock_collection_name_reserved() {
        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let err = db
            .reserve_name("db.lock")
            .err()
            .expect("db.lock must be reserved: tree_path would alias the root lock file");
        assert!(
            matches!(err, DbError::DuplicateCollection("db.lock")),
            "got {err:?}"
        );
    }
}

#[cfg(test)]
mod clean_shutdown_tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    struct MockCollection {
        name: &'static str,
        fail: bool,
        calls: Arc<AtomicUsize>,
    }

    impl Collection for MockCollection {
        fn name(&self) -> &str {
            self.name
        }
        fn len(&self) -> usize {
            0
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            if self.fail {
                Err(DbError::Io(std::io::Error::other(
                    "mock clean_shutdown failure",
                )))
            } else {
                Ok(())
            }
        }
    }

    #[test]
    fn test_db_clean_shutdown_attempt_all() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let calls = Arc::new(AtomicUsize::new(0));
        let calls2 = Arc::clone(&calls);

        lock(&db.collections).push(Arc::new(MockCollection {
            name: "fail_first",
            fail: true,
            calls,
        }));
        lock(&db.collections).push(Arc::new(MockCollection {
            name: "ok_second",
            fail: false,
            calls: calls2.clone(),
        }));

        let err = db.clean_shutdown().unwrap_err();
        assert!(
            matches!(err, DbError::Io(_)),
            "expected first collection error, got {err:?}"
        );
        assert_eq!(
            calls2.load(Ordering::SeqCst),
            2,
            "both collections must be attempted"
        );
    }
}

#[cfg(test)]
mod close_tests {
    use super::*;
    #[cfg(feature = "fjall")]
    use crate::RapiraCodec;
    use armour_core::GetType;
    #[cfg(feature = "fjall")]
    use rapira::Rapira;
    use std::sync::atomic::{AtomicUsize, Ordering};

    struct FlushMock {
        name: &'static str,
        flush_fail: bool,
        flush_calls: Arc<AtomicUsize>,
        durable_calls: Arc<AtomicUsize>,
    }

    impl Collection for FlushMock {
        fn name(&self) -> &str {
            self.name
        }
        fn len(&self) -> usize {
            0
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            self.flush_calls.fetch_add(1, Ordering::SeqCst);
            if self.flush_fail {
                Err(DbError::Io(std::io::Error::other("mock flush failure")))
            } else {
                Ok(())
            }
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            self.durable_calls.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            Ok(())
        }
    }

    // AI-03: `close` must flush every collection (and the seq generator) even when
    // one flush fails, returning the first error at the end. Pre-fix it returned on
    // the first `?`, starving later collections of their final flush/hint.
    #[test]
    fn test_db_close_flushes_all_despite_first_failure() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let calls = Arc::new(AtomicUsize::new(0));

        lock(&db.collections).push(Arc::new(FlushMock {
            name: "fail_first",
            flush_fail: true,
            flush_calls: calls.clone(),
            durable_calls: Arc::new(AtomicUsize::new(0)),
        }));
        lock(&db.collections).push(Arc::new(FlushMock {
            name: "ok_second",
            flush_fail: false,
            flush_calls: calls.clone(),
            durable_calls: Arc::new(AtomicUsize::new(0)),
        }));

        let err = db.close().unwrap_err();
        assert!(
            matches!(err, DbError::Io(_)),
            "expected the first flush error to propagate, got {err:?}"
        );
        assert_eq!(
            calls.load(Ordering::SeqCst),
            2,
            "both collections must be flushed even though the first failed"
        );
    }

    #[test]
    fn close_is_barrier_after_a_flush_error() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let flush_calls = Arc::new(AtomicUsize::new(0));
        let durable_calls = Arc::new(AtomicUsize::new(0));

        lock(&db.collections).push(Arc::new(FlushMock {
            name: "flush_fails_but_barriers",
            flush_fail: true,
            flush_calls,
            durable_calls: Arc::clone(&durable_calls),
        }));

        assert!(db.close().is_err(), "the flush error remains the result");
        assert_eq!(
            durable_calls.load(Ordering::SeqCst),
            1,
            "close must still attempt the durable barrier after the hint/flush failure"
        );
    }

    struct LengthFailureMock {
        name: &'static str,
        len: usize,
    }

    impl Collection for LengthFailureMock {
        fn name(&self) -> &str {
            self.name
        }
        fn len(&self) -> usize {
            self.len
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Err(DbError::Internal("injected close failure"))
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            Err(DbError::Internal("injected clean shutdown failure"))
        }
    }

    fn assert_length_persists_after_collection_failure(mode: FinalizeMode) {
        const NAME: &str = "length_persists_after_failure";
        const OLD_LEN: u64 = 3;
        const NEW_LEN: u64 = 7;

        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        db.db_info_update(|info| {
            info.collections.entry(NAME.to_owned()).or_default().seq = OLD_LEN;
        })
        .unwrap();
        lock(&db.collections).push(Arc::new(LengthFailureMock {
            name: NAME,
            len: NEW_LEN as usize,
        }));

        let result = match mode {
            FinalizeMode::Close => db.close(),
            FinalizeMode::CleanShutdown => db.clean_shutdown(),
        };
        assert!(result.is_err());
        assert_eq!(
            db.db_info().collections[NAME].seq,
            NEW_LEN,
            "length persistence must run after a collection finalization failure"
        );
    }

    #[test]
    fn close_persists_length_after_collection_failure() {
        assert_length_persists_after_collection_failure(FinalizeMode::Close);
    }

    #[test]
    fn clean_shutdown_persists_length_after_collection_failure() {
        assert_length_persists_after_collection_failure(FinalizeMode::CleanShutdown);
    }

    #[derive(
        Clone, Copy, GetType, zerocopy::FromBytes, zerocopy::Immutable, zerocopy::IntoBytes,
    )]
    #[repr(C)]
    struct BarrierRow {
        n: u64,
    }

    impl CollectionMeta for BarrierRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "close_barrier_rows";
    }

    fn single_shard_no_hints() -> Config {
        let mut config = Config::test_no_hints();
        config.shard_count = 1;
        config
    }

    #[test]
    fn close_is_barrier_for_dirty_no_hints_and_second_close_skips_clean_shards() {
        crate::flush_test::reset();
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let tree = db
            .open_zero_tree::<BarrierRow, 8, _>(single_shard_no_hints(), NoHook, &[])
            .unwrap();
        tree.put(&1u64.to_be_bytes(), &BarrierRow { n: 1 }).unwrap();

        db.close().unwrap();
        let after_first_close = crate::flush_test::data_attempts();
        assert_eq!(
            after_first_close, 1,
            "close must barrier a dirty shard even when no checkpoint record exists"
        );

        db.close().unwrap();
        assert_eq!(
            crate::flush_test::data_attempts(),
            after_first_close,
            "a second close must skip a clean no-hints shard"
        );
    }

    #[test]
    fn close_hint_failure_still_durable_flushes_and_preserves_that_error() {
        crate::flush_test::reset();
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let mut config = Config::test();
        config.shard_count = 1;
        config.hints = Some(true);
        let tree = db
            .open_zero_tree::<BarrierRow, 8, _>(config, NoHook, &[])
            .unwrap();
        tree.put(&1u64.to_be_bytes(), &BarrierRow { n: 1 }).unwrap();

        let hint_path = db
            .tree_path(BarrierRow::NAME)
            .join("shard_000")
            .join("000001.hint");
        let _fault = crate::hint::fail_hint_write_for(
            hint_path,
            std::io::Error::other("test: close hint failure"),
        );

        let error = db.close().unwrap_err();
        assert!(
            error.to_string().contains("test: close hint failure"),
            "the hint failure must retain first-error precedence: {error}"
        );
        assert_eq!(
            crate::flush_test::data_attempts(),
            1,
            "close must still durable-flush after a failed hint write"
        );
    }

    #[test]
    fn drop_barriers_dirty_bitcask_and_skips_clean_bitcask() {
        crate::flush_test::reset();
        let dirty_dir = tempfile::tempdir().unwrap();
        let dirty_db = Db::open_test(dirty_dir.path()).unwrap();
        let dirty_tree = dirty_db
            .open_zero_tree::<BarrierRow, 8, _>(single_shard_no_hints(), NoHook, &[])
            .unwrap();
        dirty_tree
            .put(&1u64.to_be_bytes(), &BarrierRow { n: 1 })
            .unwrap();
        drop(dirty_tree);
        drop(dirty_db);
        assert_eq!(
            crate::flush_test::data_attempts(),
            1,
            "Drop must barrier a dirty Bitcask shard"
        );

        crate::flush_test::reset();
        let clean_dir = tempfile::tempdir().unwrap();
        let clean_db = Db::open_test(clean_dir.path()).unwrap();
        let clean_tree = clean_db
            .open_zero_tree::<BarrierRow, 8, _>(single_shard_no_hints(), NoHook, &[])
            .unwrap();
        clean_tree
            .put(&1u64.to_be_bytes(), &BarrierRow { n: 1 })
            .unwrap();
        clean_db.close().unwrap();
        drop(clean_tree);
        drop(clean_db);
        assert_eq!(
            crate::flush_test::data_attempts(),
            1,
            "Drop must skip an already-clean Bitcask shard"
        );
    }

    struct WorkerShutdownMock {
        clean_calls: Arc<AtomicUsize>,
        watch_closes: Arc<AtomicUsize>,
    }

    impl Collection for WorkerShutdownMock {
        fn name(&self) -> &str {
            "worker_shutdown_mock"
        }
        fn len(&self) -> usize {
            0
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            self.clean_calls.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
        fn close_watches(&self) {
            self.watch_closes.fetch_add(1, Ordering::SeqCst);
        }
    }

    #[test]
    fn shutdown_is_idempotent_with_workers_and_drop_closes_watches() {
        let dir = tempfile::tempdir().unwrap();
        let mut db = Db::open_with_intervals(
            dir.path(),
            Some(std::time::Duration::from_millis(10)),
            Some(std::time::Duration::from_millis(10)),
        )
        .unwrap();
        let clean_calls = Arc::new(AtomicUsize::new(0));
        let watch_closes = Arc::new(AtomicUsize::new(0));
        lock(&db.collections).push(Arc::new(WorkerShutdownMock {
            clean_calls: Arc::clone(&clean_calls),
            watch_closes: Arc::clone(&watch_closes),
        }));

        db.shutdown().unwrap();
        db.shutdown().unwrap();
        drop(db);

        assert_eq!(clean_calls.load(Ordering::SeqCst), 2);
        assert_eq!(
            watch_closes.load(Ordering::SeqCst),
            3,
            "both shutdown calls and Drop must terminally close watches"
        );
    }

    #[cfg(feature = "fjall")]
    #[derive(Clone, Debug, PartialEq, GetType, Rapira)]
    struct LsmRow {
        n: u64,
    }

    #[cfg(feature = "fjall")]
    impl CollectionMeta for LsmRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "finalize_lsm_rows";
    }

    #[cfg(feature = "fjall")]
    struct FailingLsmLengthProbe;

    #[cfg(feature = "fjall")]
    impl Collection for FailingLsmLengthProbe {
        fn name(&self) -> &str {
            LsmRow::NAME
        }
        fn len(&self) -> usize {
            1
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Err(DbError::Internal("injected LSM barrier failure"))
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            Ok(())
        }
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn fjall_finalization_error_keeps_clean_false_and_reopens_from_live_length() {
        let dir = tempfile::tempdir().unwrap();
        {
            let db = Db::open_test(dir.path()).unwrap();
            let rows = db
                .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                    Some(crate::lsm::LsmDbConfig::default()),
                    crate::lsm::LsmConfig::default(),
                    NoHook,
                    &[],
                )
                .unwrap();
            for n in 1u64..=3 {
                rows.put(&n.to_be_bytes(), &LsmRow { n }).unwrap();
            }
            lock(&db.collections).push(Arc::new(FailingLsmLengthProbe));

            assert!(db.close().is_err());
            let info = db.db_info();
            assert!(!info.clean, "a failed finalization cannot publish clean");
            assert_eq!(info.collections[LsmRow::NAME].seq, 1);
            drop(rows);
        }

        let db = Db::open_test(dir.path()).unwrap();
        assert!(
            !db.opened_clean,
            "reopen must not trust stale db.info length"
        );
        let rows = db
            .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                None,
                crate::lsm::LsmConfig::default(),
                NoHook,
                &[],
            )
            .unwrap();
        assert_eq!(rows.len(), 3, "unclean reopen must rescan the LSM keyspace");
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn fjall_write_after_close_demotes_clean_before_a_failed_second_finalization() {
        let dir = tempfile::tempdir().unwrap();
        {
            let db = Db::open_test(dir.path()).unwrap();
            let rows = db
                .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                    Some(crate::lsm::LsmDbConfig::default()),
                    crate::lsm::LsmConfig::default(),
                    NoHook,
                    &[],
                )
                .unwrap();
            rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
            rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }).unwrap();
            db.close().unwrap();
            assert!(db.db_info().clean, "successful close publishes clean");

            rows.put(&3u64.to_be_bytes(), &LsmRow { n: 3 }).unwrap();
            assert!(
                !db.db_info().clean,
                "the first post-close LSM write must persistently demote clean before mutation"
            );

            lock(&db.collections).push(Arc::new(FailingLsmLengthProbe));
            assert!(
                db.close().is_err(),
                "the second finalization is injected to fail"
            );
            assert_eq!(db.db_info().collections[LsmRow::NAME].seq, 1);
            drop(rows);
        }

        let db = Db::open_test(dir.path()).unwrap();
        assert!(!db.opened_clean);
        let rows = db
            .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                None,
                crate::lsm::LsmConfig::default(),
                NoHook,
                &[],
            )
            .unwrap();
        assert_eq!(rows.len(), 3, "reopen must rescan rather than trust seq=1");
        assert_eq!(rows.get(&3u64.to_be_bytes()), Some(LsmRow { n: 3 }));
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn fjall_clean_marker_demotion_failure_aborts_write_and_retries() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let rows = db
            .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                Some(crate::lsm::LsmDbConfig::default()),
                crate::lsm::LsmConfig::default(),
                NoHook,
                &[],
            )
            .unwrap();
        rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
        db.close().unwrap();
        assert!(db.db_info().clean);

        rows.fail_next_clean_marker_demotion_for_test();
        assert!(rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }).is_err());
        assert!(rows.get(&2u64.to_be_bytes()).is_none());
        assert!(
            db.db_info().clean,
            "a failed demotion leaves the shared marker clean so the next write retries"
        );

        rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }).unwrap();
        assert_eq!(rows.get(&2u64.to_be_bytes()), Some(LsmRow { n: 2 }));
        assert!(!db.db_info().clean);
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn fjall_unwind_after_clean_persistence_retries_demotion_before_mutation() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let rows = db
            .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                Some(crate::lsm::LsmDbConfig::default()),
                crate::lsm::LsmConfig::default(),
                NoHook,
                &[],
            )
            .unwrap();
        rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
        rows.fail_next_clean_marker_demotion_for_test();
        rows.panic_after_clean_marker_persistence_for_test();

        assert!(std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| db.close())).is_err());
        assert!(
            db.db_info().clean,
            "the injected unwind demotion failure leaves the persisted marker clean"
        );
        assert!(rows.get(&2u64.to_be_bytes()).is_none());

        rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }).unwrap();
        assert_eq!(rows.get(&2u64.to_be_bytes()), Some(LsmRow { n: 2 }));
        assert!(
            !db.db_info().clean,
            "the next writer must retry durable demotion before its mutation"
        );
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn fjall_finalization_blocks_a_write_until_clean_publication_then_demotes() {
        use std::sync::mpsc;

        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let rows = db
            .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                Some(crate::lsm::LsmDbConfig::default()),
                crate::lsm::LsmConfig::default(),
                NoHook,
                &[],
            )
            .unwrap();
        rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
        let publication = rows.hold_clean_marker_publication_for_test();
        let writer_entry = rows.hold_clean_marker_mutation_entry_for_test();
        let (write_done_tx, write_done_rx) = mpsc::channel();

        std::thread::scope(|scope| {
            let finalizer = scope.spawn(|| db.close());
            publication.wait_until_blocked();
            assert!(
                db.db_info().clean,
                "the persistent clean marker is written before state publication"
            );

            let writer_rows = Arc::clone(&rows);
            let writer = scope.spawn(move || {
                let result = writer_rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 });
                write_done_tx.send(result).unwrap();
            });
            writer_entry.wait_until_entered();
            assert!(
                write_done_rx
                    .recv_timeout(std::time::Duration::from_millis(100))
                    .is_err(),
                "a managed write must not pass the finalization transition lock"
            );
            assert!(rows.get(&2u64.to_be_bytes()).is_none());

            publication.release();
            finalizer.join().unwrap().unwrap();
            write_done_rx
                .recv_timeout(std::time::Duration::from_secs(1))
                .unwrap()
                .unwrap();
            writer.join().unwrap();
        });

        assert_eq!(rows.get(&2u64.to_be_bytes()), Some(LsmRow { n: 2 }));
        assert!(
            !db.db_info().clean,
            "the released write must durably demote clean before landing"
        );
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn pre_admitted_managed_lsm_write_blocks_finalization_until_keyspace_operation() {
        use std::sync::mpsc;

        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let rows = db
            .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                Some(crate::lsm::LsmDbConfig::default()),
                crate::lsm::LsmConfig::default(),
                NoHook,
                &[],
            )
            .unwrap();
        rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
        let operation = rows.hold_clean_marker_mutation_operation_for_test();
        let finalizer_entry = rows.hold_clean_marker_finalization_entry_for_test();
        let (writer_done_tx, writer_done_rx) = mpsc::channel();
        let (finalizer_done_tx, finalizer_done_rx) = mpsc::channel();

        std::thread::scope(|scope| {
            let writer_rows = Arc::clone(&rows);
            let writer = scope.spawn(move || {
                writer_done_tx
                    .send(writer_rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }))
                    .unwrap();
            });
            operation.wait_until_blocked();

            let finalizer = scope.spawn(|| {
                finalizer_done_tx.send(db.close()).unwrap();
            });
            finalizer_entry.wait_until_entered();
            let early_finalization = finalizer_done_rx
                .recv_timeout(std::time::Duration::from_millis(100))
                .ok();
            operation.release();
            writer_done_rx
                .recv_timeout(std::time::Duration::from_secs(1))
                .unwrap()
                .unwrap();
            match early_finalization.as_ref() {
                Some(result) => assert!(result.is_ok()),
                None => finalizer_done_rx
                    .recv_timeout(std::time::Duration::from_secs(1))
                    .unwrap()
                    .unwrap(),
            }
            writer.join().unwrap();
            finalizer.join().unwrap();
            assert!(
                early_finalization.is_none(),
                "finalization must wait for a Dirty writer admitted before its barriers"
            );
        });

        assert_eq!(rows.get(&2u64.to_be_bytes()), Some(LsmRow { n: 2 }));
        assert!(
            db.db_info().clean,
            "finalization must include the admitted write"
        );
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn retained_managed_lsm_tree_cannot_mutate_after_db_drop() {
        let dir = tempfile::tempdir().unwrap();
        let info_path = dir.path().join("db.info");
        let rows = {
            let db = Db::open_test(dir.path()).unwrap();
            let rows = db
                .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                    Some(crate::lsm::LsmDbConfig::default()),
                    crate::lsm::LsmConfig::default(),
                    NoHook,
                    &[],
                )
                .unwrap();
            rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
            db.close().unwrap();
            assert!(db.db_info().clean);
            rows
        };

        assert!(rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }).is_err());
        assert!(rows.get(&2u64.to_be_bytes()).is_none());
        let info: DbInfo = serde_json::from_slice(&std::fs::read(info_path).unwrap()).unwrap();
        assert!(
            info.clean,
            "the retained tree must not rewrite db.info after Db drop"
        );
    }

    #[cfg(feature = "fjall")]
    #[test]
    fn retained_managed_lsm_tree_cannot_mutate_after_dirty_db_drop() {
        let dir = tempfile::tempdir().unwrap();
        let rows = {
            let db = Db::open_test(dir.path()).unwrap();
            let rows = db
                .open_lsm_tree::<LsmRow, RapiraCodec, NoHook>(
                    Some(crate::lsm::LsmDbConfig::default()),
                    crate::lsm::LsmConfig::default(),
                    NoHook,
                    &[],
                )
                .unwrap();
            rows.put(&1u64.to_be_bytes(), &LsmRow { n: 1 }).unwrap();
            lock(&db.collections).push(Arc::new(FailingLsmLengthProbe));
            assert!(
                db.close().is_err(),
                "injected close leaves the marker dirty"
            );
            assert!(!db.db_info().clean);
            rows
        };

        assert!(rows.put(&2u64.to_be_bytes(), &LsmRow { n: 2 }).is_err());
        assert!(rows.get(&2u64.to_be_bytes()).is_none());
    }

    struct ShutdownMock {
        clean_calls: Arc<AtomicUsize>,
        watch_closes: Arc<AtomicUsize>,
    }

    impl Collection for ShutdownMock {
        fn name(&self) -> &str {
            "shutdown_mock"
        }
        fn len(&self) -> usize {
            0
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            self.clean_calls.fetch_add(1, Ordering::SeqCst);
            Err(DbError::Io(std::io::Error::other(
                "mock clean shutdown failure",
            )))
        }
        fn close_watches(&self) {
            self.watch_closes.fetch_add(1, Ordering::SeqCst);
        }
    }

    #[test]
    fn shutdown_uses_clean_finalization_and_closes_watches_after_error() {
        let dir = tempfile::tempdir().unwrap();
        let mut db = Db::open_test(dir.path()).unwrap();
        let clean_calls = Arc::new(AtomicUsize::new(0));
        let watch_closes = Arc::new(AtomicUsize::new(0));
        lock(&db.collections).push(Arc::new(ShutdownMock {
            clean_calls: Arc::clone(&clean_calls),
            watch_closes: Arc::clone(&watch_closes),
        }));

        assert!(
            db.shutdown().is_err(),
            "clean finalization failure must surface"
        );
        assert_eq!(clean_calls.load(Ordering::SeqCst), 1);
        assert_eq!(
            watch_closes.load(Ordering::SeqCst),
            1,
            "watches must close even when finalization fails"
        );
    }
}

#[cfg(all(test, feature = "rapira-codec"))]
mod inspection_registry_tests {
    use super::*;
    use armour_core::GetType;
    use rapira::Rapira;
    use tempfile::tempdir;

    use crate::{NoHook, RapiraCodec, TypedTree};

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct AlphaItem {
        n: u64,
    }
    impl CollectionMeta for AlphaItem {
        type SelfId = [u8; 8];
        const NAME: &'static str = "alpha_items";
        const VERSION: u16 = 1;
    }

    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct ZuluItem {
        n: u64,
    }
    impl CollectionMeta for ZuluItem {
        type SelfId = [u8; 8];
        const NAME: &'static str = "zulu_items";
        const VERSION: u16 = 1;
    }

    #[test]
    fn inspection_registry_only_db_opened_collections() {
        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let _via_db = db
            .open_typed_tree::<AlphaItem, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        let _direct = TypedTree::<[u8; 8], AlphaItem, RapiraCodec, NoHook>::open_hooked(
            dir.path().join("direct_only"),
            Config::test(),
            RapiraCodec,
            NoHook,
        )
        .unwrap();

        let names: Vec<_> = lock(&db.inspection_handlers).keys().copied().collect();
        assert_eq!(names, vec![AlphaItem::NAME]);
        assert!(!names.contains(&"direct_only"));
    }

    #[test]
    fn inspection_registry_collections_sorted_by_name() {
        let dir = tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        let _zulu = db
            .open_typed_tree::<ZuluItem, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        let _alpha = db
            .open_typed_tree::<AlphaItem, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();

        let names: Vec<_> = db
            .inspector()
            .collections()
            .unwrap()
            .into_iter()
            .map(|c| c.name)
            .collect();
        assert_eq!(
            names,
            vec![AlphaItem::NAME.to_owned(), ZuluItem::NAME.to_owned()]
        );
    }
}

#[cfg(test)]
mod inspection_rollback_tests {
    use super::*;
    use std::num::NonZeroU32;

    use crate::armour::inspection::{
        InspectionCollection, InspectionEntry, InspectionError, InspectionHandler, InspectionPage,
    };

    struct StubCollection(&'static str);
    impl Collection for StubCollection {
        fn name(&self) -> &str {
            self.0
        }
        fn len(&self) -> usize {
            0
        }
        fn compact(&self) -> DbResult<usize> {
            Ok(0)
        }
        fn flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn periodic_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn durable_flush(&self) -> DbResult<()> {
            Ok(())
        }
        fn clean_shutdown(&self) -> DbResult<()> {
            Ok(())
        }
    }

    struct StubInspection;
    impl InspectionHandler for StubInspection {
        fn metadata(&self) -> Result<InspectionCollection, InspectionError> {
            Err(InspectionError::CollectionNotFound {
                name: "metadata-fail".into(),
            })
        }
        fn get_raw(&self, _: &[u8]) -> DbResult<Option<Vec<u8>>> {
            Ok(None)
        }
        fn exact_count(&self) -> DbResult<u64> {
            Ok(0)
        }
        fn ordered_page(
            &self,
            _: &crate::armour::inspection::NormalizedBounds,
            _: NonZeroU32,
        ) -> DbResult<InspectionPage> {
            Ok(InspectionPage {
                entries: vec![],
                next_key: None,
            })
        }
        fn unordered_snapshot(
            &self,
            _: NonZeroU32,
        ) -> Result<Vec<InspectionEntry>, InspectionError> {
            Ok(vec![])
        }
    }

    #[cfg(feature = "rpc")]
    mod rpc_stubs {
        use super::*;
        use std::ops::Bound;

        use crate::armour::handler::{KeyBytes, RowWriter, RpcHandler, ValueBytes};
        use armour_rpc::protocol::WriteKey;

        pub(super) struct StubRpc(pub &'static str);
        impl RpcHandler for StubRpc {
            fn name(&self) -> &str {
                self.0
            }
            fn info(&self) -> (u64, u16) {
                (0, 0)
            }
            fn schema(&self) -> armour_rpc::SchemaResponse {
                armour_rpc::SchemaResponse {
                    name: self.0.to_owned(),
                    version: 0,
                    typ: <u8 as armour_core::GetType>::TYPE,
                    key_scheme: armour_core::KeyScheme::Bytes,
                }
            }
            fn get_into(&self, _: &[u8], _: &mut Vec<u8>) -> DbResult<bool> {
                Ok(false)
            }
            fn contains(&self, _: &[u8]) -> DbResult<bool> {
                Ok(false)
            }
            fn entry_len(&self, _: &[u8]) -> DbResult<Option<u32>> {
                Ok(None)
            }
            fn first_into(&self, _: &mut RowWriter<'_>) -> DbResult<bool> {
                Ok(false)
            }
            fn last_into(&self, _: &mut RowWriter<'_>) -> DbResult<bool> {
                Ok(false)
            }
            fn range_into(
                &self,
                _: Bound<KeyBytes>,
                _: Bound<KeyBytes>,
                _: &mut RowWriter<'_>,
            ) -> DbResult<u32> {
                Ok(0)
            }
            fn range_keys_into(
                &self,
                _: Bound<KeyBytes>,
                _: Bound<KeyBytes>,
                _: &mut RowWriter<'_>,
            ) -> DbResult<u32> {
                Ok(0)
            }
            fn write(&self, _: WriteKey, _: Option<bool>, _: ValueBytes) -> DbResult<ValueBytes> {
                Ok(vec![])
            }
            fn remove(&self, _: &[u8], _: bool) -> DbResult<()> {
                Ok(())
            }
            fn take_into(&self, _: &[u8], _: bool, _: &mut Vec<u8>) -> DbResult<bool> {
                Ok(false)
            }
            fn count(&self, _: bool) -> DbResult<u64> {
                Ok(0)
            }
            fn apply_batch(&self, _: Vec<(KeyBytes, Option<ValueBytes>)>) -> DbResult<()> {
                Ok(())
            }
        }
    }

    fn publish_stub(db: &Db, name: &'static str, forced_hash: Option<u64>) -> DbResult<()> {
        let collection: Arc<dyn Collection> = Arc::new(StubCollection(name));
        let inspection: Arc<dyn InspectionHandler> = Arc::new(StubInspection);
        #[cfg(feature = "rpc")]
        let rpc =
            Some(Arc::new(rpc_stubs::StubRpc(name)) as Arc<dyn crate::armour::handler::RpcHandler>);
        db.publish_collection_for_test(
            name,
            collection,
            inspection,
            #[cfg(feature = "rpc")]
            rpc,
            forced_hash,
        )
    }

    // The forced hash only reaches a registry under `rpc` — that is the one
    // registry keyed by hash rather than by name, so without it `publish_stub`
    // has no collision to detect and the call simply succeeds.
    #[cfg(feature = "rpc")]
    #[test]
    fn inspection_rollback_hash_collision_leaves_registries_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        publish_stub(&db, "alpha", None).unwrap();

        let before_collections = lock(&db.collections).len();
        let before_inspections = lock(&db.inspection_handlers).len();
        #[cfg(feature = "rpc")]
        let before_rpc = lock(&db.handlers).len();
        let forced = xxhash_rust::xxh3::xxh3_64(b"alpha");
        let error = publish_stub(&db, "beta", Some(forced)).unwrap_err();

        assert!(matches!(error, DbError::DuplicateCollection(_)));
        assert_eq!(lock(&db.collections).len(), before_collections);
        assert_eq!(lock(&db.inspection_handlers).len(), before_inspections);
        #[cfg(feature = "rpc")]
        assert_eq!(lock(&db.handlers).len(), before_rpc);
    }

    #[test]
    fn inspection_rollback_exact_name_collision_leaves_registries_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_test(dir.path()).unwrap();
        publish_stub(&db, "alpha", None).unwrap();

        let before_collections = lock(&db.collections).len();
        let before_inspections = lock(&db.inspection_handlers).len();
        #[cfg(feature = "rpc")]
        let before_rpc = lock(&db.handlers).len();
        let error = publish_stub(&db, "alpha", None).unwrap_err();

        assert!(matches!(error, DbError::DuplicateCollection(_)));
        assert_eq!(lock(&db.collections).len(), before_collections);
        assert_eq!(lock(&db.inspection_handlers).len(), before_inspections);
        #[cfg(feature = "rpc")]
        assert_eq!(lock(&db.handlers).len(), before_rpc);
    }
}

#[cfg(all(test, feature = "rapira-codec"))]
mod db_builder_tests {
    use std::sync::Arc;

    use super::*;
    use crate::{CollectionMeta, Config, NoHook, RapiraCodec, TxConfig};
    use armour_core::GetType;
    use rapira::Rapira;

    #[derive(Clone, Copy, Debug, PartialEq, Eq, GetType, Rapira)]
    struct Row {
        n: u32,
    }
    impl CollectionMeta for Row {
        type SelfId = [u8; 8];
        const NAME: &'static str = "rows";
    }

    /// The builder's defaults are `Db::open`'s: a database opened either way
    /// behaves the same.
    #[test]
    fn builder_defaults_match_open() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(Db::builder(dir.path()).open().unwrap());
        let rows = db
            .open_typed_map::<Row, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        rows.put(&1u64.to_be_bytes(), Row { n: 7 }).unwrap();
        assert_eq!(rows.get(&1u64.to_be_bytes()).map(|r| r.n), Some(7));
    }

    /// The point of the builder: worker intervals and transaction config at the
    /// same time. `max_ops` is the cheapest observable `TxConfig` field — the
    /// limit is checked before any lock is taken.
    #[test]
    fn builder_carries_workers_and_tx_config_together() {
        let dir = tempfile::tempdir().unwrap();
        let db = Arc::new(
            Db::builder(dir.path())
                .for_tests()
                .tx_config(TxConfig {
                    max_ops: 2,
                    ..TxConfig::default()
                })
                .open()
                .unwrap(),
        );
        let rows = db
            .open_typed_map::<Row, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();

        // `Tx` doesn't implement `Debug`, so `.unwrap_err()` on the chain (whose
        // success type is `Tx`) can't be used here; match instead.
        let err = match db
            .tx()
            .put(&rows, &1u64.to_be_bytes(), &Row { n: 1 })
            .unwrap()
            .put(&rows, &2u64.to_be_bytes(), &Row { n: 2 })
            .unwrap()
            .put(&rows, &3u64.to_be_bytes(), &Row { n: 3 })
        {
            Ok(_) => panic!("expected the third put to exceed max_ops"),
            Err(e) => e,
        };
        // `check_op_limits` runs inside `put` itself (`txn/mod.rs:183`), so the
        // third builder call is where it fails — not `commit`.
        assert!(matches!(err, crate::DbError::Client(_)), "{err}");
        assert!(rows.get(&1u64.to_be_bytes()).is_none(), "nothing written");
    }

    /// A `TxConfig` field with no other test path at all today.
    #[test]
    fn builder_carries_discard_pending_on_schema_mismatch() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::builder(dir.path())
            .for_tests()
            .tx_config(TxConfig {
                discard_pending_on_schema_mismatch: true,
                ..TxConfig::default()
            })
            .open()
            .unwrap();
        assert!(db.tx_config().discard_pending_on_schema_mismatch);
    }

    /// `for_tests` is a preset over two fields, so a later `workers` call wins
    /// on the worker half: `.for_tests().workers(WorkerIntervals::default())`
    /// brings the compactor, flusher and TTL sweeper back.
    #[test]
    fn for_tests_is_a_preset_and_later_calls_override_it() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::builder(dir.path())
            .for_tests()
            .workers(WorkerIntervals::default())
            .open()
            .unwrap();
        assert!(db.compactor.is_some(), "workers() re-enabled compaction");
        assert!(
            db.flusher.is_some(),
            "workers() re-enabled the flush worker"
        );
        assert!(db.sweeper.is_some(), "workers() re-enabled the TTL sweeper");
        drop(db);
    }
}