armdb 0.14.2

//! Engine-owned unique namespaces (T10): a `key -> owner` collection whose
//! rows the engine writes, never the application directly.
//!
//! This module (Task 1) built only the row/owner types and the opaque
//! [`UniqueIndex`] wrapper — reads only. Task 2 added the declaration side:
//! the set of namespaces a source may claim into ([`UniqueTargets`]), the
//! sink a projection closure emits claims through ([`ClaimSink`]), and the
//! crate-private [`ClaimTarget`] trait through which — and *only* through
//! which — anything writes to a namespace.
//!
//! Task 5 adds the **non-transactional rule**: outside `db.tx()` the
//! engine cannot refuse a write after the fact (the source row is applied
//! under the shard lock and projections run only after the lock is
//! released), so [`guard_nontx_write`] / [`guard_nontx_write_bytes`] run
//! *inside* the critical section, before any mutation, and refuse a write
//! that would change the claim set with
//! [`DbError::UniqueIndexRequiresTx`](crate::error::DbError::UniqueIndexRequiresTx).
//! Claims are never written outside a transaction — an unchanged set has
//! nothing to write, a changed one is refused — so that task did not touch
//! [`PendingProjection`](crate::projection::PendingProjection).
//!
//! Task 7 (`resolve` submodule) adds the piece that actually enforces
//! uniqueness: [`resolve::decide`] is the pure `(current row, claimant,
//! kind) -> decision` rule, and `txn::projection_commit::resolve_claims`
//! runs it under the namespaces' own locks during `db.tx()` commit, turning
//! [`ClaimIntent`]s buffered by this module into namespace writes — or
//! refusing the whole commit with [`DbError::UniqueViolation`](crate::error::DbError::UniqueViolation).
//!
//! Task 9 adds the backstop for everything enforcement cannot have caused —
//! data written before adoption, a source written outside the registry, or a
//! bug: `schema::validate` walks every declared namespace through
//! [`ClaimTarget::scan_rows`], unions every declared source's claims into one
//! expected set per namespace (`schema::expected::build_unique`), and reports
//! `UniqueMissing`/`UniqueConflict`/`UniqueOrphan`/`UniqueRowCorrupt`. Since a
//! namespace has no public write path at all — the entire point of this
//! module — the validator's own tests need a bypass to construct states a
//! live commit could never produce; `armdb::testing::force_put` (backed by
//! [`ClaimTarget::write_row`]) is that bypass, test-only and reachable only
//! with the same crate-private trait every other write in this module goes
//! through.
//!
//! **The guard must not fire inside `db.tx()`.** A transaction's commit
//! applies each logged operation through `txn::held::TxHeld::apply`, which
//! calls the very same per-family `Shard::put`/`delete` the facade and
//! `atomic()` call — there is no way to tell those two callers apart from
//! inside the locked write body itself. Claim *admission* inside a
//! transaction is `resolve_claims`'s job (`txn::projection_commit`, beside
//! `resolve_counters`, the same way T9 resolves counter deltas during commit
//! instead of diffing them after the fact); `TxHeld::apply` is still the
//! thing that applies the decided source row, and still needs the
//! non-transactional guard suppressed while it does —
//! **`SkipClaimGuard` is permanent**, not a placeholder Task 7 removes.
//! [`SkipClaimGuard`] is the signal `TxHeld::apply` raises around one
//! `mutate` call so [`guard_nontx_write`]/[`guard_nontx_write_bytes`] know
//! to stand down for it — and, since crash-recovery replay (`txn::replay`)
//! re-applies logged operations through the same `TxLocked::apply_put`/
//! `apply_delete` entry point, replay is covered for free. The flag is
//! **thread-global**, not scoped to the one collection `apply` is applying —
//! see [`SkipClaimGuard`]'s own doc for what that means and why `enter`
//! saves and restores the previous value rather than clearing it.
//!
//! [`UniqueIndex`] is deliberately **not** a [`crate::ProjectableIndex`] —
//! see `index.rs` for why; [`ClaimTarget`] is the narrow, crate-private
//! substitute that grants only the engine itself a write path.

mod diff;
mod index;
/// The pure claim-resolution rule (T10 Task 7) — see [`resolve::decide`].
/// `pub(crate)`, not private: `txn::projection_commit::resolve_claims` is the
/// rule's production caller and lives outside this module.
pub(crate) mod resolve;
mod row;

pub use index::{UniqueIndex, UniqueNamespace};
pub use row::{UNIQUE_OWNER_KEY_CAP, UniqueOwner, UniqueRow, collection_tag};

/// `#[cfg(any(test, feature = "test-faults"))]` (T10 review I1) — see
/// [`row::corrupt_unique_row`]'s own doc for why this bypass is gated.
#[cfg(any(test, feature = "test-faults"))]
#[doc(hidden)]
pub use row::corrupt_unique_row;
#[doc(hidden)]
pub use row::unique_owner_for;

/// `claim_set` backs `first_changed` (`diff.rs`), which is what calls into
/// [`guard_nontx_write`]'s non-transactional production path — re-exported
/// here only for symmetry with `claim_delta`; nothing outside `diff.rs` calls
/// it directly yet.
#[allow(unused_imports)]
pub(crate) use diff::claim_set;
/// Claim-set diff (T10 Task 4). `claim_delta`/`claim_delta_bytes` are the six
/// families' `tx_claims_for` input and back [`into_tx_claim_ops`] below — the
/// ordered release-then-claim split
/// [`TxParticipant::tx_claims`](crate::txn::participant::TxParticipant::tx_claims)
/// needs, not a plain boolean (see `diff.rs`'s module doc for why the earlier
/// boolean-only `claims_changed` was deleted rather than kept alongside it).
pub(crate) use diff::{ClaimDelta, claim_delta, claim_delta_bytes};

use std::sync::Arc;

use crate::Key;
use crate::error::{DbError, DbResult};
use crate::hook::{TypedWriteHook, WriteHook};

/// Reject a claim carrier (`hook.claim_targets().is_some()`) opened without a
/// `Db` (T10 review I6) — mirrors
/// [`counter_dirty::check_no_counters`](crate::counter_dirty::check_no_counters)
/// in both shape and placement. A direct open is never `bind_tx_arc`'d, so it
/// can never join `db.tx()`: every claim-changing write on such a collection
/// would be refused forever by [`DbError::UniqueIndexRequiresTx`], and the
/// error it raises would carry an empty `collection` (the source is never
/// registered with any `counter_dirty`/claim bookkeeping either). Refusing at
/// open is cheaper than discovering that at the first write.
///
/// Runs only on the direct `open_hooked`/`open_with_hook` entry points, the
/// same eight call sites `check_no_counters` already guards; a managed open
/// (`Db::open_*`) calls the private `*_inner` constructor directly and never
/// observes this check.
pub(crate) fn check_no_claims<K: Key, T, H: TypedWriteHook<K, T>>(hook: &H) -> DbResult<()> {
    if hook.claim_targets().is_some() {
        return Err(DbError::ClaimsRequireManagedOpen);
    }
    Ok(())
}

/// Reject a claim carrier on a collection whose value type declares a TTL
/// (T10 review C1). [`crate::schema::SchemaRegistry::unique_index_on`]
/// already asserts `SV::TTL.is_none()` at declaration time — but that is only
/// one of two reachable declaration paths: [`crate::projection::Projected::and_unique`],
/// [`UniqueTargets`] and [`crate::armour::Db::open_unique_index`] are all
/// `pub`, so a carrier assembled by hand and passed straight to a managed
/// `Db::open_*` never meets that assert. Called from every managed opener,
/// where [`crate::TtlHook::EXPIRES`] = `T::TTL.is_some()` is already read —
/// before the raw hook is wrapped in `TtlHook`, since a claim carrier with a
/// TTL must never be reachable at all, not merely disarmed.
///
/// Left unrejected, the sweeper's guard placement disagrees by family:
/// `TypedTree`/`TypedMap`/`ZeroTree`/`ZeroMap` retry the expiry delete
/// forever (the guard lives *inside* the locked delete, so `sweep_expired`
/// hits `UniqueIndexRequiresTx`, counts a failure and reschedules —
/// the row never expires), while `VarTypedTree`/`VarTypedMap` call
/// `delete_locked` directly from the sweep and delete the row with no guard
/// at all — silently orphaning its claim, a name locked forever with no
/// owner (spec decision 13). Registering this check once, here, closes both
/// paths at their common root instead of patching each family's sweeper to
/// agree.
pub(crate) fn check_no_ttl_claims<T, H>(hook: &H) -> DbResult<()>
where
    T: crate::CollectionMeta,
    H: TypedWriteHook<T::SelfId, T>,
{
    if T::TTL.is_some() && hook.claim_targets().is_some() {
        return Err(DbError::ClaimingSourceHasTtl {
            collection: T::NAME,
        });
    }
    Ok(())
}

/// Reject a claim carrier on a collection whose own key does not fit a unique
/// namespace owner row (T10 review F3). [`crate::schema::SchemaRegistry::unique_index_on`]
/// already has a `const { assert!(size_of::<SV::SelfId>() <= UNIQUE_OWNER_KEY_CAP) }`
/// at declaration time — but, exactly like the TTL check above, that assert is
/// only reachable through the registry path: [`crate::projection::Projected::and_unique`],
/// [`UniqueTargets`] and every managed `Db::open_*` are all `pub`, so a
/// hand-assembled carrier can reach a managed open without ever meeting it.
///
/// Left unrejected, such a collection opens fine and then fails **every**
/// claim-changing transaction with `DbError::Internal("claim source key
/// exceeds the owner capacity")` from [`into_tx_claim_ops`] — a working open
/// followed by a permanently broken write path. Refusing here, beside the TTL
/// check, moves the failure to where the unsupported configuration is
/// created.
pub(crate) fn check_claim_key_width<T, H>(hook: &H) -> DbResult<()>
where
    T: crate::CollectionMeta,
    H: TypedWriteHook<T::SelfId, T>,
{
    if hook.claim_targets().is_some() && size_of::<T::SelfId>() > UNIQUE_OWNER_KEY_CAP {
        return Err(DbError::ClaimingSourceKeyTooWide {
            collection: T::NAME,
        });
    }
    Ok(())
}

/// Crate-private write seam for a unique namespace. Implemented **only** for
/// [`UniqueIndex<N>`] (in `index.rs`, where the wrapper's private inner
/// handle is in scope) — the sole path by which anything, including this
/// crate's own commit path, writes an engine-owned `key -> owner` row. No
/// public trait grants this; `UniqueIndex` itself exposes no write method.
///
/// `as_tx_participant` and `collection_name` are reachable via `db.tx()`
/// wiring (T10 Task 6). Task 7's resolver decides namespace writes through
/// the generic [`crate::txn::participant::TxLocked`] seam instead (mirroring
/// `resolve_counters`), so `write_row`/`delete_row`/`scan_rows` were
/// unreachable until Task 9 (the schema validator): `scan_rows` now backs
/// `schema::validate`'s namespace pass, and `write_row` backs
/// `armdb::testing::force_put` (there being no other way to construct a
/// namespace row a live commit could never produce — an orphan or a corrupt
/// one — for that pass's own tests). `delete_row` was reserved for Task 10's
/// repair pass (`schema::unique_repair::repair_namespaces`), which now
/// deletes a plain orphan row through it — the same seam `write_row` and
/// `scan_rows` already go through, never a public API.
pub(crate) trait ClaimTarget: Send + Sync {
    fn collection_name(&self) -> &'static str;
    /// Unconditionally write `row` at `key`. CAS/uniqueness enforcement is
    /// the caller's concern — `resolve_claims` (`txn::projection_commit`) is
    /// the only production caller, and it decides *before* calling this; this
    /// method only performs the write.
    fn write_row(&self, key: &[u8], row: UniqueRow) -> DbResult<()>;
    fn delete_row(&self, key: &[u8]) -> DbResult<()>;
    /// Full scan of live rows — validation/reconciliation use.
    fn scan_rows(&self, f: &mut dyn FnMut(&[u8], UniqueRow) -> DbResult<()>) -> DbResult<()>;
    /// This namespace seen as a transaction participant (T10 `db.tx()` wiring).
    fn as_tx_participant(&self) -> Arc<dyn crate::txn::participant::TxParticipant>;
}

#[derive(Clone)]
struct ErasedClaimTarget {
    name: &'static str,
    target: Arc<dyn ClaimTarget>,
}

/// The set of unique namespaces one source may claim into, declared on a
/// [`crate::schema`]-style registry before the source is opened. Mirrors
/// [`crate::IndexTargets`] in shape: a small `Vec` of erased handles, looked
/// up by [`UniqueIndex::name`].
#[derive(Clone, Default)]
pub struct UniqueTargets {
    handles: Vec<ErasedClaimTarget>,
}

impl UniqueTargets {
    pub fn new() -> Self {
        Self {
            handles: Vec::new(),
        }
    }

    /// Declare `ns` as a claim target. Builder-style, mirrors
    /// [`crate::IndexTargets::with_index`].
    pub fn with<N>(mut self, ns: &Arc<UniqueIndex<N>>) -> Self
    where
        N: UniqueNamespace,
        N::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
    {
        self.handles.push(ErasedClaimTarget {
            name: ns.name(),
            target: Arc::clone(ns) as Arc<dyn ClaimTarget>,
        });
        self
    }

    pub fn contains(&self, name: &str) -> bool {
        self.handles.iter().any(|h| h.name == name)
    }

    pub fn is_empty(&self) -> bool {
        self.handles.is_empty()
    }

    pub fn len(&self) -> usize {
        self.handles.len()
    }

    pub(crate) fn first_name(&self) -> &'static str {
        self.handles.first().map_or("", |target| target.name)
    }

    /// Declared claim targets, in registration order — mirrors
    /// [`crate::counter::CounterTargets::all`] /
    /// [`crate::projection::IndexTargets::erased_targets`]. Used by
    /// [`crate::schema::SchemaRegistry::unique_index`]'s topology and
    /// owner-tag checks (`feature = "schema"`), and by
    /// [`crate::armour::Db::tx_register_claim_targets`] (T10 Task 6).
    pub(crate) fn all(&self) -> impl Iterator<Item = &Arc<dyn ClaimTarget>> {
        self.handles.iter().map(|h| &h.target)
    }

    /// Resolve a declared namespace to its write seam. `None` when `name`
    /// was never declared with [`Self::with`]. Every production caller
    /// resolves a namespace through [`Self::resolve_tx_participant`] instead
    /// (claims are decided and written through the transaction commit path,
    /// never directly against a [`ClaimTarget`]), so this is genuinely
    /// unused — kept as the raw-write-seam counterpart to
    /// [`crate::projection::IndexTargets::resolve_target`] in case a future
    /// non-transactional caller needs it.
    #[allow(dead_code)]
    pub(crate) fn resolve(&self, name: &str) -> Option<Arc<dyn ClaimTarget>> {
        self.handles
            .iter()
            .find(|h| h.name == name)
            .map(|h| Arc::clone(&h.target))
    }

    /// Resolve a declared namespace to its transaction participant (T10
    /// `db.tx()` admission, mirrors [`crate::IndexTargets::resolve_tx_participant`]).
    /// Used by [`into_tx_claim_ops`] to fill in each [`ClaimOp`](crate::txn::participant::ClaimOp)'s
    /// namespace participant (T10 Task 6).
    pub(crate) fn resolve_tx_participant(
        &self,
        name: &str,
    ) -> Option<Arc<dyn crate::txn::participant::TxParticipant>> {
        self.handles
            .iter()
            .find(|h| h.name == name)
            .map(|h| h.target.as_tx_participant())
    }
}

/// One claim emitted by a projection closure through [`ClaimSink::claim`].
/// The owner is **not** supplied here — the sink records only
/// `(namespace, key)`; the owner is filled in by the caller of the
/// projection, which knows the source collection and row key, not by the
/// application closure (see [`into_tx_claim_ops`], T10 Task 6).
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct ClaimIntent {
    pub(crate) namespace: &'static str,
    pub(crate) key: Vec<u8>,
}

/// Buffer + budgets + declared target table for one claim-projection
/// evaluation. Mirrors [`crate::ProjectionSink`] in shape and in how it
/// accounts its budget: one row, `key.len() + size_of::<UniqueRow>()` bytes
/// per claim, against the same [`DbError::ProjectionBudgetExceeded`].
pub struct ClaimSink<'a> {
    targets: &'a UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    row_count: usize,
    byte_count: usize,
    claims: Vec<ClaimIntent>,
}

impl<'a> ClaimSink<'a> {
    pub(crate) fn new(targets: &'a UniqueTargets, max_rows: usize, max_bytes: usize) -> Self {
        Self {
            targets,
            max_rows,
            max_bytes,
            row_count: 0,
            byte_count: 0,
            claims: Vec::new(),
        }
    }

    /// Buffered claims so far.
    pub(crate) fn claims(&self) -> &[ClaimIntent] {
        &self.claims
    }

    /// Emit a claim on `ns` for `key`. Errors when `ns` was not declared on
    /// the [`UniqueTargets`] this sink was built from, or when the claim
    /// would exceed the sink's row/byte budget.
    pub fn claim<N>(&mut self, ns: &Arc<UniqueIndex<N>>, key: N::SelfId) -> DbResult<()>
    where
        N: UniqueNamespace,
        N::SelfId: crate::Key + Send + Sync + std::hash::Hash + Eq,
    {
        let namespace = ns.name();
        if !self.targets.contains(namespace) {
            return Err(DbError::UndeclaredUniqueTarget { namespace });
        }
        self.record(namespace, crate::Key::as_bytes(&key))
    }

    fn record(&mut self, namespace: &'static str, key: &[u8]) -> DbResult<()> {
        let added_bytes = key.len() + size_of::<UniqueRow>();
        let next_rows = self.row_count + 1;
        let next_bytes = self.byte_count + added_bytes;
        if next_rows > self.max_rows || next_bytes > self.max_bytes {
            return Err(DbError::ProjectionBudgetExceeded {
                rows: next_rows,
                bytes: next_bytes,
            });
        }
        self.row_count = next_rows;
        self.byte_count = next_bytes;
        self.claims.push(ClaimIntent {
            namespace,
            key: key.to_vec(),
        });
        Ok(())
    }
}

/// Turn one write's [`ClaimDelta`] into transaction claim ops (T10 Task 6) —
/// the claim twin of [`crate::projection::PendingProjection::into_tx_derived`]
/// / [`crate::counter::into_tx_counter_deltas`]. `name`/`key` are the
/// *source* collection's own name and row key — every op built here shares
/// one [`UniqueOwner`], since every claim or release in one write's delta
/// comes from the same source row. Releases are emitted before claims,
/// preserving [`ClaimDelta`]'s own release-before-claim order (Task 7's
/// resolver relies on it).
pub(crate) fn into_tx_claim_ops(
    targets: &UniqueTargets,
    name: &'static str,
    key: &[u8],
    delta: ClaimDelta,
) -> DbResult<Vec<crate::txn::participant::ClaimOp>> {
    let owner = UniqueOwner::new(collection_tag(name), key, 0).ok_or(DbError::Internal(
        "claim source key exceeds the owner capacity",
    ))?;
    let mut out = Vec::with_capacity(delta.releases.len() + delta.claims.len());
    for intent in delta.releases {
        out.push(build_claim_op(
            targets,
            owner,
            crate::txn::participant::ClaimKind::Release,
            intent,
        )?);
    }
    for intent in delta.claims {
        out.push(build_claim_op(
            targets,
            owner,
            crate::txn::participant::ClaimKind::Claim,
            intent,
        )?);
    }
    Ok(out)
}

fn build_claim_op(
    targets: &UniqueTargets,
    owner: UniqueOwner,
    kind: crate::txn::participant::ClaimKind,
    intent: ClaimIntent,
) -> DbResult<crate::txn::participant::ClaimOp> {
    let participant = targets.resolve_tx_participant(intent.namespace).ok_or(
        DbError::UndeclaredUniqueTarget {
            namespace: intent.namespace,
        },
    )?;
    Ok(crate::txn::participant::ClaimOp {
        participant,
        namespace: intent.namespace,
        key: intent.key,
        owner,
        kind,
    })
}

/// Write `row` directly into `ns` at `key`, bypassing `resolve_claims`
/// entirely: no CAS, no owner check, no requirement that `row` even be
/// well-formed. `UniqueIndex` has no public write path at all (the whole
/// point of this module), so this is the only way to construct a namespace
/// state a live `db.tx()` commit could never produce — a permanent owner row
/// with no claiming source row (an "orphan", for `schema::validate`'s own
/// tests), or a corrupt one ([`row::corrupt_unique_row`]). Test-only:
/// reachable at `armdb::testing::force_put`, not part of the public API.
///
/// `#[cfg(any(test, feature = "test-faults"))]` (T10 review I1): this was
/// previously `#[doc(hidden)] pub` with no `cfg` at all — a real, compiling
/// write path into a collection whose entire premise (spec decision 4) is
/// that no such path exists, shipping in every ordinary build. Gating it
/// behind `test`/`test-faults` keeps the bypass available to this crate's own
/// tests and to `armdb/tests/*.rs` integration binaries built with
/// `--features test-faults`, and removes it from a plain `cargo build`/
/// `cargo add armdb` entirely.
#[cfg(any(test, feature = "test-faults"))]
#[doc(hidden)]
pub fn force_put<N>(ns: &Arc<UniqueIndex<N>>, key: &N::SelfId, row: UniqueRow) -> DbResult<()>
where
    N: UniqueNamespace,
    N::SelfId: Key + Send + Sync + std::hash::Hash + Eq,
{
    ClaimTarget::write_row(ns.as_ref(), Key::as_bytes(key), row)
}

/// One explicit `reserve`/`release` op queued directly on a
/// [`crate::txn::Tx`] builder (T10 Task 8) — the claim-stage twin of an
/// ordinary builder `put`/`delete`, except it never becomes a raw write: it
/// is merged into the same list [`crate::txn::participant::ClaimOp`]s from a
/// projecting source feed (`txn::projection_commit::pending_claims_from_explicit`),
/// so `resolve_claims` decides it under the same two-level partition
/// (release before acquisition) instead of letting an explicit `reserve` and
/// a projected claim on the same key race through an ordinary
/// last-write-wins overlay (`txn::projection_commit::set_write`).
///
/// `kind`'s `Reserve { deadline_ms }` carries the **requested TTL** in
/// milliseconds here — not yet an absolute deadline. `resolve_claims` stamps
/// that into an absolute epoch-ms deadline, from one clock read, immediately
/// before handing the kind to [`resolve::decide`] (which treats
/// `deadline_ms` as the literal value to write) — see that function's own
/// doc.
pub(crate) struct ExplicitClaim {
    pub participant: Arc<dyn crate::txn::participant::TxParticipant>,
    pub namespace: &'static str,
    pub key: Vec<u8>,
    pub owner: UniqueOwner,
    pub kind: crate::txn::participant::ClaimKind,
}

/// Validate and convert a reservation's requested TTL (T10 Task 8): rejects a
/// zero duration — an expired-on-arrival reservation is indistinguishable
/// from a leak, and the caller means [`crate::txn::Tx::release`] — and a
/// duration whose millisecond count does not fit `u64`. A sub-millisecond,
/// non-zero duration rounds **up** to 1 ms (review round M3: genuinely
/// rounded, via `div_ceil` on the nanosecond count — not truncated
/// `as_millis()` clamped to a 1 ms floor, which would silently round a
/// 1.5 ms request *down* to 1 ms), so it can never resolve to "no deadline"
/// ([`UniqueRow::deadline`] reads `expires_at == 0` as permanent) or land in
/// the past.
///
/// Runs in the builder ([`crate::txn::Tx::reserve`]), before any lock is
/// taken — `namespace` names the error the same way every other T10 error in
/// this module does.
pub(crate) fn reservation_ttl_ms(
    ttl: std::time::Duration,
    namespace: &'static str,
) -> DbResult<u64> {
    if ttl.is_zero() {
        return Err(DbError::UniqueReservationInvalid {
            namespace,
            reason: "zero duration",
        });
    }
    let ms = ttl.as_nanos().div_ceil(1_000_000);
    u64::try_from(ms).map_err(|_| DbError::UniqueReservationInvalid {
        namespace,
        reason: "deadline overflow",
    })
}

/// Wall-clock milliseconds since the Unix epoch, for stamping a
/// reservation's absolute deadline (T10 Task 8,
/// `txn::projection_commit::resolve_claims`). A clock reading before the
/// epoch is treated as the epoch itself: `now + ttl` should describe a
/// sensible future deadline, not fail a whole commit over a misconfigured
/// clock.
pub(crate) fn reservation_now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or(std::time::Duration::ZERO)
        .as_millis() as u64
}

thread_local! {
    /// See [`SkipClaimGuard`].
    static SKIP_CLAIM_GUARD: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

/// RAII scope that suppresses [`guard_nontx_write`]/[`guard_nontx_write_bytes`]
/// for its lifetime. The **only** caller is `txn::held::TxHeld::apply`,
/// wrapped tightly around the single `mutate` call every `apply_put`/
/// `apply_delete` makes.
///
/// The flag is **thread-global**, not scoped to the collection being
/// applied: while it is set, *any* non-transactional write on *any* claiming
/// collection reached from `mutate`'s call graph (e.g. `hook.deadline`, the
/// codec's `encode_to`) is also exempted, not just the one write `apply`
/// intended. Today's call graph never nests one `apply` inside another on
/// the same thread, but that is a property of the current code, not of this
/// type — so `enter`/`drop` save and restore the *previous* value (like a
/// saved/restored interrupt mask) rather than unconditionally clearing it
/// (T10 Task 5 review, F3). Unconditional clearing would let a hypothetical
/// nested `enter` disarm the guard for the *rest* of an outer, still-live
/// scope: the inner `Drop` would clear the flag while the outer call is
/// still logically "inside" its own scope, re-arming the guard mid-commit —
/// and a refusal at that point is a `TxFailedAfterLog` after the record is
/// already durable, which must not be retried.
pub(crate) struct SkipClaimGuard(bool);

impl SkipClaimGuard {
    #[inline]
    pub(crate) fn enter() -> Self {
        let previous = SKIP_CLAIM_GUARD.with(|f| f.replace(true));
        Self(previous)
    }
}

impl Drop for SkipClaimGuard {
    #[inline]
    fn drop(&mut self) {
        SKIP_CLAIM_GUARD.with(|f| f.set(self.0));
    }
}

#[inline]
fn skip_claim_guard() -> bool {
    SKIP_CLAIM_GUARD.with(std::cell::Cell::get)
}

/// T10 Task 5 — the non-transactional rule: refuse a write (facade or
/// `atomic()`) that would change `collection`'s claim set, **before** any
/// mutation. Called from the two typed families' (`TypedTree`/`TypedMap`)
/// locked write bodies, immediately after the old-value read those bodies
/// already do and before their first `append_entry` / index mutation — the
/// shard lock is already held there and nothing has been written yet, which
/// is what makes the refusal real rather than advisory.
///
/// `H::CLAIMS` is a compile-time constant, but it does **not** by itself mean
/// "this source declared claims" — [`crate::projection::Projected`]'s
/// `TypedWriteHook`/`WriteHook` impls hardcode `const CLAIMS: bool = true`
/// unconditionally (mirrors `PROJECTS`/`COUNTS`), so a plain secondary-index
/// or counter source built through `Projected` also has `H::CLAIMS = true`
/// even when it never called `.and_unique(...)`. A source with `H = NoHook`
/// (or any hand-rolled hook with `CLAIMS = false`) is eliminated entirely at
/// compile time by the `if !H::CLAIMS` check below; a `Projected` source that
/// simply never declared a claim target pays one cheap runtime `Option`
/// check (`hook.claim_targets().is_none()`, just below) instead — every
/// caller with a genuinely additive pre-read (e.g. `put_locked_inner`) gates
/// on that same runtime check *before* reading, not only before calling in
/// here. [`SkipClaimGuard`] (checked next) stands the guard down specifically
/// inside a `db.tx()` commit/replay — see the module docs.
///
/// A claiming hook must set `NEEDS_OLD_VALUE = true` — the guard cannot see a
/// claim-set change without the old snapshot. Every claiming hook is built
/// through [`crate::projection::Projected::and_unique`], which hardcodes
/// `NEEDS_OLD_VALUE = true` unconditionally, so this holds by construction;
/// the `const` assertion below turns a hand-rolled violation of that into a
/// compile error at the call site's monomorphization instead of a silently
/// wrong (always-`None`) old snapshot.
#[inline]
pub(crate) fn guard_nontx_write<K: Key, T, H: TypedWriteHook<K, T>>(
    collection: &'static str,
    hook: &H,
    limits: (usize, usize),
    key: &K,
    old: Option<&T>,
    new: Option<&T>,
) -> DbResult<()> {
    if !H::CLAIMS {
        return Ok(());
    }
    const { assert!(!H::CLAIMS || H::NEEDS_OLD_VALUE) };
    if skip_claim_guard() {
        return Ok(());
    }
    let Some(targets) = hook.claim_targets() else {
        return Ok(());
    };
    if let Some(namespace) =
        crate::unique::diff::first_changed(hook, targets, limits.0, limits.1, key, old, new)?
    {
        return Err(DbError::UniqueIndexRequiresTx {
            collection,
            namespace,
        });
    }
    Ok(())
}

/// Byte-snapshot twin of [`guard_nontx_write`] for the `WriteHook`
/// (Const/Var) families — called from the locked write bodies backing
/// `ZeroTree`/`ZeroMap` (`ConstTree`/`ConstMap`) and `VarTypedTree`/
/// `VarTypedMap` (`VarTree`/`VarMap`), which own their hook as
/// `H: WriteHook<K>` and work in `Option<&[u8]>` snapshots rather than a
/// decoded `T`. Same placement rule, same `H::CLAIMS` compile-time
/// short-circuit, same `NEEDS_OLD_VALUE` invariant as the typed twin.
#[inline]
pub(crate) fn guard_nontx_write_bytes<K: Key, H: WriteHook<K>>(
    collection: &'static str,
    hook: &H,
    limits: (usize, usize),
    key: &K,
    old: Option<&[u8]>,
    new: Option<&[u8]>,
) -> DbResult<()> {
    if !H::CLAIMS {
        return Ok(());
    }
    const { assert!(!H::CLAIMS || H::NEEDS_OLD_VALUE) };
    if skip_claim_guard() {
        return Ok(());
    }
    let Some(targets) = hook.claim_targets() else {
        return Ok(());
    };
    if let Some(namespace) =
        crate::unique::diff::first_changed_bytes(hook, targets, limits.0, limits.1, key, old, new)?
    {
        return Err(DbError::UniqueIndexRequiresTx {
            collection,
            namespace,
        });
    }
    Ok(())
}