armdb 0.8.0

//! The erased seam a transaction sees.
//!
//! Type erasure happens at the *call site* (`Tx::put` encodes with the
//! collection's codec while the type is still known), so nothing below this
//! line is generic over the value type — which is what removes the per-arity
//! ceiling `atomicN` had.

use crate::error::DbResult;
use crate::txlog::record::{OpKind, VersionTag};

/// Which durability backend a participant is on — it selects the version tag
/// written to the log and, in part 2, gates participation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TxBackend {
    Bitcask,
    Fixed,
}

/// One resolved operation handed to [`TxLocked::reserve`], in apply order.
pub(crate) struct PlannedOp<'a> {
    pub key: &'a [u8],
    pub delete: bool,
}

/// Value types that carry a collection name log and replay under [`CollectionMeta::NAME`].
pub(crate) trait TxCollectionName {
    fn tx_collection_name() -> &'static str {
        ""
    }
}

impl<T: crate::CollectionMeta> TxCollectionName for T {
    fn tx_collection_name() -> &'static str {
        T::NAME
    }
}

/// A collection that can take part in a transaction. Implemented for the six
/// managed families; `Db` hands out the trait object at `commit()`.
pub(crate) trait TxParticipant: Send + Sync {
    fn name(&self) -> &'static str;
    fn backend(&self) -> TxBackend;
    /// `typ_hash` of the stored type — the log records it per operation so a
    /// replay after a schema change fails closed instead of injecting stale
    /// encodings.
    fn schema_fingerprint(&self) -> u64;
    fn encrypted(&self) -> bool;
    /// Largest `key + value + header` this participant's write path accepts, or
    /// `None` when it has no such ceiling (FixedStore writes a fixed-size slot,
    /// and the value length is pinned by the collection's type).
    ///
    /// Checked **before** the log append. Past that point the record is durable,
    /// so a write the collection cannot take is not merely a failed operation:
    /// replay re-attempts it at every subsequent open and the database stops
    /// opening at all.
    fn max_entry_bytes(&self) -> Option<usize>;
    fn shard_of(&self, key: &[u8]) -> DbResult<usize>;
    /// Lock `shards` (ascending, deduplicated) and keep the guards.
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>>;
}

/// The locked view of one participant during a commit. Every method works in
/// **stored bytes**: the codec ran at the call site.
pub(crate) trait TxLocked {
    /// Current stored value, or `None` when absent. Used by `upsert`/`guard`
    /// evaluation and by nothing else.
    fn read(&mut self, key: &[u8]) -> DbResult<Option<Vec<u8>>>;
    /// Reserve a version per operation, in apply order, before the record is
    /// logged. Bitcask reserves a contiguous GSN block per shard.
    fn reserve(&mut self, ops: &[PlannedOp<'_>]) -> DbResult<Vec<VersionTag>>;
    /// Apply one write and report **how many hook events it buffered**.
    ///
    /// Not always one: a mutation that changed nothing — a delete of an absent
    /// key, or any write on a family whose hook does not want writes — buffers
    /// none. The caller uses the count to keep its replay order aligned with
    /// [`DeferredReplay::event_count`]; assuming one event per applied write
    /// silently misaligns the hooks.
    fn apply_put(&mut self, key: &[u8], value: &[u8], tag: VersionTag) -> DbResult<usize>;
    fn apply_delete(&mut self, key: &[u8], tag: VersionTag) -> DbResult<usize>;
    /// Does this logged operation still need to be applied to this collection?
    ///
    /// Answered under the shard lock, because FixedStore reads the slot's live
    /// state to decide. Bitcask compares the reserved GSN with the shard's
    /// recovered watermark.
    fn needs_replay(&mut self, key: &[u8], kind: OpKind, tag: VersionTag) -> DbResult<bool>;
    /// Announce that this transaction's record is durable.
    ///
    /// After this call a reservation is owned by replay: `abort` (and the `Drop`
    /// that a panic unwinds through) must leave the reserved slot allocated, or
    /// an unrelated key takes it and replay overwrites that key's row. Call it
    /// on every locked participant as soon as `TxLog::append` returns.
    fn mark_logged(&mut self);
    /// Release the locks, undoing any reservation. Nothing was written —
    /// unless [`mark_logged`](Self::mark_logged) already ran, in which case the
    /// reservations stay for replay.
    fn abort(self: Box<Self>);
    /// Reservations taken but not yet applied (tests only).
    #[cfg(test)]
    fn unapplied_reservations(&self) -> usize;
    /// Arm a reserved version without running the mutation (tests only): the
    /// state a panic between arming and mutating leaves behind.
    #[cfg(test)]
    fn tx_debug_arm(&mut self, key: &[u8], tag: VersionTag) -> DbResult<()>;
    /// Release the locks and hand back the owner of the deferred hook work.
    ///
    /// It must **not** return a plain event vector: for the Typed families an
    /// event holds raw pointers whose validity depends on a `seize::LocalGuard`
    /// entered before the first lock, and hooks fire after this call.
    fn finish(self: Box<Self>) -> DbResult<Box<dyn DeferredReplay>>;
}

/// Owns whatever must stay alive between lock release and the last hook —
/// the family's buffered events plus, for the Typed families, the seize guard.
/// Retirement happens on `Drop`, after every `replay_one`.
pub(crate) trait DeferredReplay {
    fn event_count(&self) -> usize;
    fn replay_one(&mut self, local: usize);
}

#[cfg(test)]
mod tests {
    use super::*;

    /// A participant made of nothing: proves the traits are object-safe and
    /// that a `DeferredReplay` can outlive the locked object it came from.
    /// Mirrors the `compile_gate` stub in `ctx.rs`.
    struct StubParticipant;
    struct StubLocked {
        events: Vec<&'static str>,
    }
    struct StubDeferred {
        events: Vec<&'static str>,
        fired: std::rc::Rc<std::cell::RefCell<Vec<&'static str>>>,
    }

    impl TxParticipant for StubParticipant {
        fn name(&self) -> &'static str {
            "stub"
        }
        fn backend(&self) -> TxBackend {
            TxBackend::Bitcask
        }
        fn schema_fingerprint(&self) -> u64 {
            1
        }
        fn encrypted(&self) -> bool {
            false
        }
        fn max_entry_bytes(&self) -> Option<usize> {
            None
        }
        fn shard_of(&self, _key: &[u8]) -> DbResult<usize> {
            Ok(0)
        }
        fn lock_shards<'a>(&'a self, _shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>> {
            Ok(Box::new(StubLocked { events: Vec::new() }))
        }
    }

    impl TxLocked for StubLocked {
        fn read(&mut self, _key: &[u8]) -> DbResult<Option<Vec<u8>>> {
            Ok(None)
        }
        fn reserve(&mut self, ops: &[PlannedOp<'_>]) -> DbResult<Vec<VersionTag>> {
            Ok((0..ops.len())
                .map(|i| VersionTag::Bitcask { gsn: i as u64 })
                .collect())
        }
        fn apply_put(&mut self, _key: &[u8], _value: &[u8], _tag: VersionTag) -> DbResult<usize> {
            self.events.push("put");
            Ok(1)
        }
        fn apply_delete(&mut self, _key: &[u8], _tag: VersionTag) -> DbResult<usize> {
            self.events.push("delete");
            Ok(1)
        }
        fn needs_replay(&mut self, _key: &[u8], _kind: OpKind, _tag: VersionTag) -> DbResult<bool> {
            Ok(false)
        }
        fn mark_logged(&mut self) {}
        fn abort(self: Box<Self>) {}
        #[cfg(test)]
        fn unapplied_reservations(&self) -> usize {
            0
        }
        #[cfg(test)]
        fn tx_debug_arm(&mut self, _key: &[u8], _tag: VersionTag) -> DbResult<()> {
            Ok(())
        }
        fn finish(self: Box<Self>) -> DbResult<Box<dyn DeferredReplay>> {
            Ok(Box::new(StubDeferred {
                events: self.events,
                fired: std::rc::Rc::new(std::cell::RefCell::new(Vec::new())),
            }))
        }
    }

    impl DeferredReplay for StubDeferred {
        fn event_count(&self) -> usize {
            self.events.len()
        }
        fn replay_one(&mut self, local: usize) {
            self.fired.borrow_mut().push(self.events[local]);
        }
    }

    #[test]
    fn tx_seam_is_object_safe_and_survives_finish() {
        let p = StubParticipant;
        let mut locked: Box<dyn TxLocked> = p.lock_shards(&[0]).unwrap();
        let tags = locked
            .reserve(&[
                PlannedOp {
                    key: b"a",
                    delete: false,
                },
                PlannedOp {
                    key: b"b",
                    delete: true,
                },
            ])
            .unwrap();
        locked.apply_put(b"a", b"v", tags[0]).unwrap();
        locked.apply_delete(b"b", tags[1]).unwrap();

        let mut deferred = locked.finish().unwrap();
        assert_eq!(deferred.event_count(), 2);
        deferred.replay_one(0);
        deferred.replay_one(1);
    }
}