armdb 0.14.1

//! 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 std::sync::Arc;

use crate::error::{DbError, DbResult};
use crate::format_migration::plan::{MigrationPlan, TransformBuffers, TransformOutcome};
use crate::txlog::record::{OpKind, VersionTag};
use crate::unique::UniqueOwner;

/// One row emitted by [`TxParticipant::tx_project`] for admission and logging.
pub(crate) struct TxDerivedWrite {
    pub participant: Arc<dyn TxParticipant>,
    pub key: Vec<u8>,
    /// `None` is a delete.
    pub value: Option<Vec<u8>>,
}

/// Absolute queue rows and post-success accounting produced by one collapsed
/// change-stream transition.  The log only sees `writes`; callbacks deliberately
/// stay outside replay because counters and wakeups describe live process state.
pub(crate) struct TxStreamWrites {
    pub writes: Vec<TxDerivedWrite>,
    pub after_success: Vec<Box<dyn FnOnce() + Send>>,
    pub queue_depth_admissions: Vec<crate::queue::QueueDepthAdmission>,
}

impl TxStreamWrites {
    pub(crate) fn empty() -> Self {
        Self {
            writes: Vec::new(),
            after_success: Vec::new(),
            queue_depth_admissions: Vec::new(),
        }
    }
}

/// One counter delta emitted by [`TxParticipant::tx_counter_deltas`] for a
/// decided source write.
///
/// A delta is **not** what reaches the transaction log. The log holds absolute
/// values — the replay rule ("has this operation already landed?") can only
/// answer that for an absolute write, and that is what makes replay
/// exactly-once. So the commit path sums every delta for one `(target, key)`
/// across the whole transaction, then resolves the sum against the target's
/// current value **under the target's own lock**, turning it into an ordinary
/// put (or delete) in the same record as the source write. The record format
/// is untouched.
///
/// `target` does the arithmetic ([`ErasedCounterTarget::resolve`]);
/// `participant` is the same collection seen as a transaction participant, and
/// is what the commit path matches against the planned entries.
///
/// [`ErasedCounterTarget::resolve`]: crate::counter::ErasedCounterTarget::resolve
pub(crate) struct TxCounterDelta {
    pub target: Arc<dyn crate::counter::ErasedCounterTarget>,
    pub participant: Arc<dyn TxParticipant>,
    pub key: Vec<u8>,
    pub delta: i128,
}

/// What a [`ClaimOp`] means for the unique-namespace row it targets (T10
/// Task 6).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ClaimKind {
    /// Take `key` permanently (`UniqueRow::expires_at == 0`).
    Claim,
    /// Give `key` up — the source's old value claimed it, its new one
    /// doesn't (or the row was deleted).
    Release,
    /// Hold `key` until `deadline_ms`, then let the TTL sweep it unless it is
    /// converted to a permanent [`Self::Claim`] first. Not produced by
    /// [`crate::unique::claim_delta`]/[`crate::unique::claim_delta_bytes`] —
    /// only [`crate::txn::Tx::reserve`] builds one (T10 Task 8), via
    /// [`crate::unique::ExplicitClaim`].
    ///
    /// **Meaning of `deadline_ms` depends on where this value is read.** On
    /// [`crate::unique::ExplicitClaim`] and on
    /// [`crate::txn::projection_commit::PendingClaim`] (i.e. anywhere
    /// *before* `resolve_claims` has run) it is the requested TTL in
    /// milliseconds, relative to a clock read `resolve_claims` has not taken
    /// yet. `resolve_claims` stamps it into an absolute epoch-ms deadline —
    /// from one clock read, under the namespace's own lock — before ever
    /// constructing the `Reserve` that reaches [`crate::unique::resolve::decide`],
    /// which treats the field as the literal value to write.
    Reserve { deadline_ms: u64 },
}

/// One claim operation a decided source write emits into a declared unique
/// namespace ([`TxParticipant::tx_claims`], T10 Task 6) — carries enough for
/// the resolver (T10 Task 7) to decide it under the namespace's own lock.
///
/// `participant` is the **namespace's** own `TxParticipant` (the unique
/// index collection this op targets), matching [`TxDerivedWrite::participant`]/
/// [`TxCounterDelta::participant`]'s shape — not the source's. `owner` is the
/// same for every op built from one source write (see
/// [`crate::unique::into_tx_claim_ops`]): the source collection's name and
/// row key, which is what makes it the one thing a released/claimed
/// namespace row remembers about who holds it.
///
/// Unlike [`TxCounterDelta`], claim ops are **not** coalesced — two claims on
/// one key from the same transaction are exactly the conflict the resolver
/// exists to catch, so summing or replacing them would erase it.
pub(crate) struct ClaimOp {
    pub participant: Arc<dyn TxParticipant>,
    pub namespace: &'static str,
    pub key: Vec<u8>,
    pub owner: UniqueOwner,
    pub kind: ClaimKind,
}

impl ClaimOp {
    pub(crate) fn is_claim(&self) -> bool {
        matches!(self.kind, ClaimKind::Claim)
    }
    pub(crate) fn is_release(&self) -> bool {
        matches!(self.kind, ClaimKind::Release)
    }
}

/// 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>;
    fn shard_count(&self) -> usize {
        1
    }
    /// Lock `shards` (ascending, deduplicated) and keep the guards.
    fn lock_shards<'a>(&'a self, shards: &[usize]) -> DbResult<Box<dyn TxLocked + 'a>>;
    /// Secondary-index rows for a decided source write (`old`/`new` in stored bytes).
    fn tx_project(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<TxDerivedWrite>> {
        let _ = (key, old, new);
        Ok(Vec::new())
    }
    /// Change-stream rows for one collapsed source transition.  Implementors
    /// return already encoded absolute queue writes, so replay never evaluates
    /// user stream code.
    fn tx_stream_writes(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<TxStreamWrites> {
        let _ = (key, old, new);
        Ok(TxStreamWrites::empty())
    }
    /// Counter deltas for a decided source write (`old`/`new` in stored bytes)
    /// — the counter twin of [`tx_project`](Self::tx_project).
    ///
    /// Returns *deltas*, not writes: the absolute value they resolve to is only
    /// knowable under the target's lock, which the commit path takes in its
    /// rank-1 round. See [`TxCounterDelta`].
    fn tx_counter_deltas(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<TxCounterDelta>> {
        let _ = (key, old, new);
        Ok(Vec::new())
    }
    /// Claim operations a decided source write emits into declared unique
    /// namespaces (`old`/`new` in stored bytes) — the claim twin of
    /// [`tx_project`](Self::tx_project)/[`tx_counter_deltas`](Self::tx_counter_deltas)
    /// (T10 Task 6).
    ///
    /// Unlike a counter delta, a claim op is not resolved here: `project_rank0`
    /// (`txn::projection_commit`) collects every op verbatim into a
    /// [`crate::txn::projection_commit::PendingClaim`] slot, without coalescing,
    /// for the resolver (T10 Task 7) to decide under the namespace's own lock.
    fn tx_claims(
        &self,
        key: &[u8],
        old: Option<&[u8]>,
        new: Option<&[u8]>,
    ) -> DbResult<Vec<ClaimOp>> {
        let _ = (key, old, new);
        Ok(Vec::new())
    }
    /// Declared projection targets for commit planning (rank `1` participants)
    /// — secondary-index targets, counter targets, **and** unique-namespace
    /// targets, since all three are written under a lock this commit must
    /// take in its second round.
    fn tx_preplan_projection_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        Ok(Vec::new())
    }
    /// Queue destinations whose emitted key is unknown until stream lowering.
    /// These are locked across all shards before invoking user stream code.
    fn tx_preplan_stream_targets(&self) -> DbResult<Vec<Arc<dyn TxParticipant>>> {
        Ok(Vec::new())
    }
    /// Whether this participant declares a change stream at all.
    ///
    /// The commit path uses this to skip materializing stream inputs for the
    /// overwhelming majority of transactions, which have no stream half. The
    /// default answer is derived from the target list, which for a source
    /// without a stream is an empty `Vec` and therefore allocation-free.
    fn tx_emits(&self) -> DbResult<bool> {
        Ok(!self.tx_preplan_stream_targets()?.is_empty())
    }
    fn tx_prepare_streams(&self) -> DbResult<Vec<crate::queue::retention::ProducerPermit>> {
        Ok(Vec::new())
    }
}

/// 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>>>;
    /// Whether a row is present, without reading or decoding its payload.
    /// Used by `insert` so an occupied row remains `KeyExists` even when its
    /// value cannot be read or re-encoded.
    fn contains(&mut self, key: &[u8]) -> DbResult<bool>;
    /// 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.
    ///
    /// `fenced` collects `(shard, effective gsn)` for every shard this
    /// participant fenced — non-empty only past the log point, and only for
    /// Bitcask-backed families.
    ///
    /// `wakes` collects this participant's subscriber wakers. It is an
    /// out-parameter rather than a return value because a waker taken off a
    /// subscriber is not re-registered by a later push: losing the batch on the
    /// error path would park an async consumer forever on an event that is
    /// already in its ring.
    fn finish(
        self: Box<Self>,
        fenced: &mut Vec<(usize, u64)>,
        wakes: &mut crate::watch::WatchWakeBatch,
    ) -> 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);
}

impl MigrationPlan {
    pub(crate) fn accepts_logged_fingerprint(&self, fingerprint: u64) -> bool {
        self.format_steps
            .first()
            .and_then(|step| step.from_typ)
            .is_some_and(|typ| typ.h() == fingerprint)
            || self
                .accepted_fingerprints
                .iter()
                .any(|(accepted, _)| *accepted == fingerprint)
    }

    /// Transform a value held by a pending transaction record from the
    /// fingerprint recorded at commit time into this plan's target format.
    ///
    /// The transaction log itself is immutable during recovery: this only
    /// supplies the bytes to the collection's normal replay write path.
    pub(crate) fn transform_logged<'a>(
        &self,
        fingerprint: u64,
        key: &[u8],
        value: &'a [u8],
        buffers: &'a mut TransformBuffers,
    ) -> DbResult<TransformOutcome<'a>> {
        let target = self
            .format_steps
            .last()
            .and_then(|step| step.to_typ)
            .map(armour_core::Typ::h);
        let from_first = self
            .format_steps
            .first()
            .and_then(|step| step.from_typ)
            .map(armour_core::Typ::h);
        let from_version = if Some(fingerprint) == target {
            self.format_to_version
        } else if Some(fingerprint) == from_first {
            self.stored_version
        } else if let Some((_, suffix_from_version)) = self
            .accepted_fingerprints
            .iter()
            .find(|(accepted, _)| *accepted == fingerprint)
        {
            *suffix_from_version as u16
        } else {
            return Err(DbError::TxSchemaMismatch {
                collection: self.collection,
                tx_id: 0,
                stored: target.unwrap_or(fingerprint),
                logged: fingerprint,
            });
        };
        self.transform_from_version(from_version, key, value, buffers)
    }
}

#[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 contains(&mut self, _key: &[u8]) -> DbResult<bool> {
            Ok(false)
        }
        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>,
            _fenced: &mut Vec<(usize, u64)>,
            _wakes: &mut crate::watch::WatchWakeBatch,
        ) -> 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(
                &mut Vec::new(),
                &mut crate::watch::WatchWakeBatch::default(),
            )
            .unwrap();
        assert_eq!(deferred.event_count(), 2);
        deferred.replay_one(0);
        deferred.replay_one(1);
    }
}