armdb 0.11.0

//! Post-open lifecycle shared by every typed `Db::open_*` recipe.
//!
//! Deliberately separate from [`crate::armour::Collection`]: the migration
//! path needs a durable engine flush (`durable_flush` / inherent `flush`),
//! which is not what `Collection::flush` does on Typed / VarTyped /
//! Zero-Bitcask (those flush userspace buffers ± hints).

use crate::armour::migration::TypedMigrationFn;
use crate::codec::Codec;
use crate::error::DbResult;
use crate::hook::TypedWriteHook;
use crate::key::Key;
use crate::tree_meta::CollectionMeta;
use crate::{TypedMap, TypedTree};

pub(crate) trait ManagedOpen {
    type Key;
    type Val;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize>;

    /// Durable flush between migration steps. See module docs.
    fn flush_for_migration(&self) -> DbResult<()>;

    fn replay_init(&self);

    /// Replay pending transaction-log operations for this collection before
    /// migrations run. Bitcask participants override; LSM and unsupported
    /// backends keep the default no-op.
    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        let _ = (db, stored_fingerprint);
        Ok(0)
    }

    /// Register the `Arc` handle transaction deferred replay needs.
    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        let _ = arc;
    }

    /// Install transaction metadata (projection target ranks, etc.).
    fn register_tx_metadata(&self, _db: &crate::armour::Db) -> DbResult<()> {
        Ok(())
    }

    /// Build the in-memory TTL index from live RAM state. Default no-op.
    fn build_ttl_index(&self) -> DbResult<()> {
        Ok(())
    }

    /// Whether this collection's hook participates in secondary-index projection.
    fn projections_enabled(&self) -> bool {
        false
    }

    /// Rebuild declared index targets from a full source scan (migration reconciliation).
    fn reconcile_projections(&self) -> DbResult<()> {
        Ok(())
    }

    /// Bind the `counters_dirty` marker (T9 Task 7) into this collection.
    /// Called by `Db::open_managed`, after construction and before the
    /// collection is published — a direct open never reaches this (see
    /// `counter_dirty::check_no_counters`).
    ///
    /// Default no-op; overridden by the six typed families (`TypedTree`,
    /// `TypedMap`, `ZeroTree`, `ZeroMap`, `VarTypedTree`, `VarTypedMap`) to
    /// reach through to the collection struct that owns the marker.
    fn bind_counter_dirty(&self, _db: &crate::armour::Db, _name: &'static str) {}

    /// Persist `counters_dirty = true` for this collection, right after
    /// [`bind_counter_dirty`](Self::bind_counter_dirty) and still before the
    /// collection is published. `Db::open_managed` calls it for a **counted
    /// source** only; the flag then stands for the whole session and a
    /// successful `close`/`clean_shutdown` clears it. See `counter_dirty.rs`
    /// for why the mark lives at open rather than on the write path.
    ///
    /// Default no-op (`Ok(())`); overridden by the six typed families.
    fn mark_counter_dirty(&self) -> DbResult<()> {
        Ok(())
    }

    /// Open-time / on-demand counter recount (T9 Task 8) — the two-pass walk
    /// (`counter::recount`) that recomputes every declared target row's
    /// aggregate from this collection's current live data. Called by
    /// `Db::open_managed` when this collection or one of its declared counter
    /// targets is dirty, and by `Db::recount_counters` on demand. Returns the
    /// number of target rows repaired.
    ///
    /// Default no-op (`Ok(0)`); overridden by the six typed families to reach
    /// through to the carrier's hook. `LsmTree` keeps the default — it does
    /// not participate in the counter system yet, matching
    /// `reconcile_projections`'s default no-op for the same reason.
    fn reconcile_counters(&self) -> DbResult<u64> {
        Ok(0)
    }

    /// The collections whose `counters_dirty` flag this collection must also
    /// consult at open time (T9 Task 8) — one entry per declared counter
    /// target. Default empty; overridden by the six typed families.
    fn counter_target_names(&self) -> Vec<&'static str> {
        Vec::new()
    }
}

impl<T, C, H> ManagedOpen for TypedTree<T::SelfId, T, C, H>
where
    T: CollectionMeta + Clone + Send + Sync + 'static,
    T::SelfId: Key + Ord + Send + Sync,
    C: Codec<T> + Sync + 'static,
    H: TypedWriteHook<T::SelfId, T> + 'static,
{
    type Key = T::SelfId;
    type Val = T;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize> {
        TypedTree::migrate_inner(self, mfn, dry_run)
    }

    fn flush_for_migration(&self) -> DbResult<()> {
        // Today's opener calls durable_flush(), NOT Collection::flush.
        self.durable_flush()
    }

    fn replay_init(&self) {
        TypedTree::replay_init(self)
    }

    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        crate::txn::replay::replay_pending(db, self, stored_fingerprint)
    }

    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        TypedTree::bind_tx_self(arc);
    }

    fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_install_projection_metadata(db)
    }

    fn build_ttl_index(&self) -> DbResult<()> {
        TypedTree::build_ttl_index(self)
    }

    fn projections_enabled(&self) -> bool {
        H::PROJECTS
    }

    fn reconcile_projections(&self) -> DbResult<()> {
        TypedTree::reconcile_projections(self)
    }

    fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
        let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
        let _ = self.counter_dirty.set(dirty);
    }

    fn mark_counter_dirty(&self) -> DbResult<()> {
        TypedTree::mark_counter_dirty(self)
    }

    fn reconcile_counters(&self) -> DbResult<u64> {
        TypedTree::reconcile_counters(self)
    }

    fn counter_target_names(&self) -> Vec<&'static str> {
        TypedTree::counter_target_names(self)
    }
}

impl<T, C, H> ManagedOpen for TypedMap<T::SelfId, T, C, H>
where
    T: CollectionMeta + Clone + Send + Sync + 'static,
    T::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
    C: Codec<T> + Sync + 'static,
    H: TypedWriteHook<T::SelfId, T> + 'static,
{
    type Key = T::SelfId;
    type Val = T;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize> {
        TypedMap::migrate_inner(self, mfn, dry_run)
    }

    fn flush_for_migration(&self) -> DbResult<()> {
        self.durable_flush()
    }

    fn replay_init(&self) {
        TypedMap::replay_init(self)
    }

    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        crate::txn::replay::replay_pending(db, self, stored_fingerprint)
    }

    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        TypedMap::bind_tx_self(arc);
    }

    fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_install_projection_metadata(db)
    }

    fn build_ttl_index(&self) -> DbResult<()> {
        TypedMap::build_ttl_index(self)
    }

    fn projections_enabled(&self) -> bool {
        H::PROJECTS
    }

    fn reconcile_projections(&self) -> DbResult<()> {
        TypedMap::reconcile_projections(self)
    }

    fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
        let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
        let _ = self.counter_dirty.set(dirty);
    }

    fn mark_counter_dirty(&self) -> DbResult<()> {
        TypedMap::mark_counter_dirty(self)
    }

    fn reconcile_counters(&self) -> DbResult<u64> {
        TypedMap::reconcile_counters(self)
    }

    fn counter_target_names(&self) -> Vec<&'static str> {
        TypedMap::counter_target_names(self)
    }
}

use crate::durability::{Bitcask, Fixed};
use crate::{ZeroMap, ZeroTree};

impl<T, const V: usize, H> ManagedOpen for ZeroTree<T::SelfId, V, T, H, 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,
{
    type Key = T::SelfId;
    type Val = T;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize> {
        ZeroTree::migrate_inner(self, mfn, dry_run)
    }

    fn flush_for_migration(&self) -> DbResult<()> {
        // Today's opener calls the inherent ZeroTree::flush (engine), not
        // Collection::flush (buffers ± hints on Bitcask).
        ZeroTree::flush(self)
    }

    fn replay_init(&self) {
        ZeroTree::replay_init(self)
    }

    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        crate::txn::replay::replay_pending(db, self, stored_fingerprint)
    }

    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        ZeroTree::bind_tx_self(arc);
    }

    fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_install_projection_metadata(db)
    }

    fn build_ttl_index(&self) -> DbResult<()> {
        ZeroTree::build_ttl_index(self)
    }

    fn projections_enabled(&self) -> bool {
        H::PROJECTS
    }

    fn reconcile_projections(&self) -> DbResult<()> {
        ZeroTree::reconcile_projections(self)
    }

    fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
        let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
        let _ = self.tx_inner().counter_dirty.set(dirty);
    }

    fn mark_counter_dirty(&self) -> DbResult<()> {
        self.tx_inner().mark_counter_dirty()
    }

    fn reconcile_counters(&self) -> DbResult<u64> {
        self.tx_inner().reconcile_counters()
    }

    fn counter_target_names(&self) -> Vec<&'static str> {
        self.tx_inner().counter_target_names()
    }
}

impl<T, const V: usize, H> ManagedOpen for ZeroTree<T::SelfId, V, T, H, 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,
{
    type Key = T::SelfId;
    type Val = T;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize> {
        ZeroTree::migrate_inner(self, mfn, dry_run)
    }

    fn flush_for_migration(&self) -> DbResult<()> {
        ZeroTree::flush(self)
    }

    fn replay_init(&self) {
        ZeroTree::replay_init(self)
    }

    // `build_ttl_index` is generic over `D`, so a FixedStore `ZeroTree` needs it
    // exactly as much as a Bitcask one: without it, rows with a deadline never
    // expire after a restart. `ZeroMap<Fixed>` below always had it; splitting
    // this impl by backend dropped it here.
    fn build_ttl_index(&self) -> DbResult<()> {
        ZeroTree::build_ttl_index(self)
    }

    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        crate::txn::replay::replay_pending(db, self, stored_fingerprint)
    }

    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        ZeroTree::bind_tx_self(arc);
    }

    fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_install_projection_metadata(db)
    }

    fn projections_enabled(&self) -> bool {
        H::PROJECTS
    }

    fn reconcile_projections(&self) -> DbResult<()> {
        ZeroTree::reconcile_projections(self)
    }

    fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
        let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
        let _ = self.tx_inner().counter_dirty.set(dirty);
    }

    fn mark_counter_dirty(&self) -> DbResult<()> {
        self.tx_inner().mark_counter_dirty()
    }

    fn reconcile_counters(&self) -> DbResult<u64> {
        self.tx_inner().reconcile_counters()
    }

    fn counter_target_names(&self) -> Vec<&'static str> {
        self.tx_inner().counter_target_names()
    }
}

impl<T, const V: usize, H> ManagedOpen for ZeroMap<T::SelfId, V, T, H, Bitcask>
where
    T: CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    T::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
    H: TypedWriteHook<T::SelfId, T> + 'static,
{
    type Key = T::SelfId;
    type Val = T;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize> {
        ZeroMap::migrate_inner(self, mfn, dry_run)
    }

    fn flush_for_migration(&self) -> DbResult<()> {
        ZeroMap::flush(self)
    }

    fn replay_init(&self) {
        ZeroMap::replay_init(self)
    }

    fn build_ttl_index(&self) -> DbResult<()> {
        ZeroMap::build_ttl_index(self)
    }

    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        crate::txn::replay::replay_pending(db, self, stored_fingerprint)
    }

    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        ZeroMap::bind_tx_self(arc);
    }

    fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_install_projection_metadata(db)
    }

    fn projections_enabled(&self) -> bool {
        H::PROJECTS
    }

    fn reconcile_projections(&self) -> DbResult<()> {
        ZeroMap::reconcile_projections(self)
    }

    fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
        let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
        let _ = self.tx_inner().counter_dirty.set(dirty);
    }

    fn mark_counter_dirty(&self) -> DbResult<()> {
        self.tx_inner().mark_counter_dirty()
    }

    fn reconcile_counters(&self) -> DbResult<u64> {
        self.tx_inner().reconcile_counters()
    }

    fn counter_target_names(&self) -> Vec<&'static str> {
        self.tx_inner().counter_target_names()
    }
}

impl<T, const V: usize, H> ManagedOpen for ZeroMap<T::SelfId, V, T, H, Fixed>
where
    T: CollectionMeta
        + Copy
        + zerocopy::IntoBytes
        + zerocopy::FromBytes
        + zerocopy::Immutable
        + Send
        + Sync
        + 'static,
    T::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
    H: TypedWriteHook<T::SelfId, T> + 'static,
{
    type Key = T::SelfId;
    type Val = T;

    fn migrate_inner(
        &self,
        mfn: &TypedMigrationFn<Self::Key, Self::Val>,
        dry_run: bool,
    ) -> DbResult<usize> {
        ZeroMap::migrate_inner(self, mfn, dry_run)
    }

    fn flush_for_migration(&self) -> DbResult<()> {
        ZeroMap::flush(self)
    }

    fn replay_init(&self) {
        ZeroMap::replay_init(self)
    }

    fn build_ttl_index(&self) -> DbResult<()> {
        ZeroMap::build_ttl_index(self)
    }

    fn replay_pending_tx_log(
        &self,
        db: &crate::armour::Db,
        stored_fingerprint: Option<u64>,
    ) -> crate::error::DbResult<usize> {
        crate::txn::replay::replay_pending(db, self, stored_fingerprint)
    }

    fn bind_tx_arc(arc: &std::sync::Arc<Self>)
    where
        Self: Sized,
    {
        ZeroMap::bind_tx_self(arc);
    }

    fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
        self.tx_install_projection_metadata(db)
    }

    fn projections_enabled(&self) -> bool {
        H::PROJECTS
    }

    fn reconcile_projections(&self) -> DbResult<()> {
        ZeroMap::reconcile_projections(self)
    }

    fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
        let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
        let _ = self.tx_inner().counter_dirty.set(dirty);
    }

    fn mark_counter_dirty(&self) -> DbResult<()> {
        self.tx_inner().mark_counter_dirty()
    }

    fn reconcile_counters(&self) -> DbResult<u64> {
        self.tx_inner().reconcile_counters()
    }

    fn counter_target_names(&self) -> Vec<&'static str> {
        self.tx_inner().counter_target_names()
    }
}

#[cfg(feature = "var-collections")]
mod var_typed_managed {
    use super::*;
    use crate::VarTypedMap;
    use crate::VarTypedTree;

    impl<T, C, H> ManagedOpen for VarTypedTree<T::SelfId, T, C, H>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Ord + Send + Sync,
        C: Codec<T> + Clone + Sync + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        type Key = T::SelfId;
        type Val = T;

        fn migrate_inner(
            &self,
            mfn: &TypedMigrationFn<Self::Key, Self::Val>,
            dry_run: bool,
        ) -> DbResult<usize> {
            VarTypedTree::migrate_inner(self, mfn, dry_run)
        }

        fn flush_for_migration(&self) -> DbResult<()> {
            self.durable_flush()
        }

        fn replay_init(&self) {
            VarTypedTree::replay_init(self)
        }

        fn bind_tx_arc(arc: &std::sync::Arc<Self>)
        where
            Self: Sized,
        {
            VarTypedTree::bind_tx_self(arc);
        }

        fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
            self.tx_install_projection_metadata(db)
        }

        fn replay_pending_tx_log(
            &self,
            db: &crate::armour::Db,
            stored_fingerprint: Option<u64>,
        ) -> DbResult<usize> {
            crate::txn::replay::replay_pending(db, self, stored_fingerprint)
        }

        fn build_ttl_index(&self) -> DbResult<()> {
            VarTypedTree::build_ttl_index(self)
        }

        fn projections_enabled(&self) -> bool {
            H::PROJECTS
        }

        fn reconcile_projections(&self) -> DbResult<()> {
            VarTypedTree::reconcile_projections(self)
        }

        fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
            let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
            let _ = self.as_inner().counter_dirty.set(dirty);
        }

        fn mark_counter_dirty(&self) -> DbResult<()> {
            self.as_inner().mark_counter_dirty()
        }

        fn reconcile_counters(&self) -> DbResult<u64> {
            self.as_inner().reconcile_counters()
        }

        fn counter_target_names(&self) -> Vec<&'static str> {
            self.as_inner().counter_target_names()
        }
    }

    impl<T, C, H> ManagedOpen for VarTypedMap<T::SelfId, T, C, H>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
        C: Codec<T> + Clone + Sync + 'static,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        type Key = T::SelfId;
        type Val = T;

        fn migrate_inner(
            &self,
            mfn: &TypedMigrationFn<Self::Key, Self::Val>,
            dry_run: bool,
        ) -> DbResult<usize> {
            VarTypedMap::migrate_inner(self, mfn, dry_run)
        }

        fn flush_for_migration(&self) -> DbResult<()> {
            self.durable_flush()
        }

        fn replay_init(&self) {
            VarTypedMap::replay_init(self)
        }

        fn bind_tx_arc(arc: &std::sync::Arc<Self>)
        where
            Self: Sized,
        {
            VarTypedMap::bind_tx_self(arc);
        }

        fn register_tx_metadata(&self, db: &crate::armour::Db) -> DbResult<()> {
            self.tx_install_projection_metadata(db)
        }

        fn replay_pending_tx_log(
            &self,
            db: &crate::armour::Db,
            stored_fingerprint: Option<u64>,
        ) -> DbResult<usize> {
            crate::txn::replay::replay_pending(db, self, stored_fingerprint)
        }

        fn build_ttl_index(&self) -> DbResult<()> {
            VarTypedMap::build_ttl_index(self)
        }

        fn projections_enabled(&self) -> bool {
            H::PROJECTS
        }

        fn reconcile_projections(&self) -> DbResult<()> {
            VarTypedMap::reconcile_projections(self)
        }

        fn bind_counter_dirty(&self, db: &crate::armour::Db, name: &'static str) {
            let dirty = crate::counter_dirty::CounterDirty::new(name, db.counter_dirty_weak());
            let _ = self.as_inner().counter_dirty.set(dirty);
        }

        fn mark_counter_dirty(&self) -> DbResult<()> {
            self.as_inner().mark_counter_dirty()
        }

        fn reconcile_counters(&self) -> DbResult<u64> {
            self.as_inner().reconcile_counters()
        }

        fn counter_target_names(&self) -> Vec<&'static str> {
            self.as_inner().counter_target_names()
        }
    }
}

#[cfg(feature = "fjall")]
mod lsm_managed {
    use super::*;
    use crate::lsm::LsmTree;

    impl<T, C, H> ManagedOpen for LsmTree<T::SelfId, T, C, H>
    where
        T: CollectionMeta + Clone + Send + Sync + 'static,
        T::SelfId: Key + Ord + Send + Sync + 'static,
        C: Codec<T> + Clone + Sync,
        H: TypedWriteHook<T::SelfId, T> + 'static,
    {
        type Key = T::SelfId;
        type Val = T;

        fn migrate_inner(
            &self,
            mfn: &TypedMigrationFn<Self::Key, Self::Val>,
            dry_run: bool,
        ) -> DbResult<usize> {
            LsmTree::migrate_inner(self, mfn, dry_run)
        }

        fn flush_for_migration(&self) -> DbResult<()> {
            LsmTree::flush(self)
        }

        fn replay_init(&self) {
            LsmTree::replay_init(self)
        }
    }
}