armdb 0.8.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 (`flush_durable` / 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;
    }

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

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 flush_durable(), NOT Collection::flush.
        self.flush_durable()
    }

    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 build_ttl_index(&self) -> DbResult<()> {
        TypedTree::build_ttl_index(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.flush_durable()
    }

    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 build_ttl_index(&self) -> DbResult<()> {
        TypedMap::build_ttl_index(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 build_ttl_index(&self) -> DbResult<()> {
        ZeroTree::build_ttl_index(self)
    }
}

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

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

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

#[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.flush_durable()
        }

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

    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.flush_durable()
        }

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

#[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)
        }
    }
}