armdb 0.11.0

//! T9: the open-time `counters_dirty` marker.
//!
//! The existing global `DbInfo.clean` flag cannot stand in for this:
//! `Db::close` is documented to tolerate writes after it returns, it writes
//! `clean = true` even when collection flushes accumulated errors, and it runs
//! the transaction checkpoint *after* that mark. All three holes survive any
//! attempt to reuse it for counter recovery.
//!
//! So T9 owns a per-collection `counters_dirty` flag ([`CollectionInfo`](crate::armour::CollectionInfo)),
//! persisted when a **counted source is opened** — in `Db::open_managed`,
//! before the collection is published — and cleared only when finalisation
//! (`Db::close` / `Db::clean_shutdown`) actually succeeded **and** this
//! session's open is what set it ([`CounterDirty::take_marked`]).
//!
//! # Why the mark is at open, not on the write path
//!
//! The mark used to run in the write-path teardown, immediately before the
//! first delta of a session applied. That placement could not deliver what it
//! promised: `Ctx::with_shard` runs the locked mutation, *releases the lock*,
//! and only then calls the projection/counter teardown — so by the time a
//! failing mark could return an error, the source row was already written.
//! Aborting there did not abort the write; it left a committed row whose delta
//! was skipped, and a retry of the same `put` produced `old == new` and hence
//! no delta at all. The counter was then permanently stale with no evidence
//! that a recount was owed — precisely the loss the flag exists to prevent.
//!
//! Marking at open is the same evidence, taken at a point where it can still
//! be enforced: the collection is not yet published, so a failed mark **fails
//! the open** and the source never takes a single write. The `Db` is also
//! provably alive there, which is why the write path no longer touches the
//! marker at all.
//!
//! The cost is deliberate: the flag stands for the whole session, so an
//! unclean shutdown recounts even when no counted write ever happened. A clean
//! close still clears it, so an ordinary restart does not recount. The
//! symmetric cost is that a write *after* `close()` — which the crate
//! tolerates — is no longer re-marked; stop writers before closing.
//!
//! Direct opens (`TypedTree::open_hooked` and its siblings) have no `Db` and
//! therefore nowhere to persist the flag, so a carrier with counters is
//! rejected there ([`check_no_counters`]) — otherwise "the engine repairs at
//! the next open" would be a promise the code cannot keep.

use std::sync::Weak;
use std::sync::atomic::{AtomicBool, Ordering};

use armour_core::persist::Persist;

use crate::Key;
use crate::armour::db::DbInfo;
use crate::error::{DbError, DbResult};
use crate::hook::TypedWriteHook;

/// One collection's `counters_dirty` marker.
///
/// Bound once, at managed open, into an `OnceLock<CounterDirty>` field on the
/// collection that owns the write path — `TypedTree`/`TypedMap` directly, or
/// `ConstTree`/`ConstMap`/`VarTree`/`VarMap` for the `ZeroTree`/`ZeroMap`/
/// `VarTypedTree`/`VarTypedMap` families they back. It lives on the collection
/// struct rather than behind the hook chain so that a user-authored wrapper
/// hook cannot silently disconnect it by forgetting a forwarding method (this
/// crate already hit that failure class once — `TtlHook` not forwarding
/// `COUNTS`, T9 Task 5).
///
/// It carries exactly two things:
///
/// - [`mark`](Self::mark) — persist `counters_dirty = true` for this
///   collection. Called once per session, from `Db::open_managed`, for a
///   counted **source**.
/// - `marked` — "this session's open is what persisted the flag", read (and
///   consumed) by `Db::close`/`Db::clean_shutdown` through
///   [`take_marked`](Self::take_marked). That distinction is load-bearing: the
///   same durable bit also carries a second, marker-less meaning set by
///   `run_migration` on a counter *target* ("my content moved under the
///   counter"), which only a source's later open consumes. A close that
///   cleared it would destroy the sole evidence a recount is owed.
pub(crate) struct CounterDirty {
    name: &'static str,
    marked: AtomicBool,
    info: Weak<Persist<DbInfo>>,
}

impl CounterDirty {
    pub(crate) fn new(name: &'static str, info: Weak<Persist<DbInfo>>) -> Self {
        Self {
            name,
            marked: AtomicBool::new(false),
            info,
        }
    }

    /// Persist `counters_dirty = true` for this collection, at most once per
    /// session.
    ///
    /// The only caller is `Db::open_managed`, which holds `&Db` across the
    /// call — so a dead `Weak` here is not a tolerated race but a broken
    /// invariant, and is reported as an error rather than silently succeeding.
    /// The error fails the open, which is the whole point: a counted source
    /// that cannot record evidence must not start taking writes.
    pub(crate) fn mark(&self) -> DbResult<()> {
        if self.marked.load(Ordering::Relaxed) {
            return Ok(());
        }
        counter_mark_fault()?;
        let Some(info) = self.info.upgrade() else {
            return Err(DbError::Internal(
                "counters_dirty marker outlived its db.info handle",
            ));
        };
        info.update(|i| {
            i.collections
                .entry(self.name.to_owned())
                .or_default()
                .counters_dirty = true;
        })?;
        self.marked.store(true, Ordering::Relaxed);
        Ok(())
    }

    /// Consume the "this session's open persisted the flag" bit and report
    /// whether it was set.
    ///
    /// `Db::close`/`Db::clean_shutdown` clear the durable `counters_dirty`
    /// only for a collection that answers `true` here — see the type's
    /// rustdoc for why a target's marker-less flag must survive a close.
    pub(crate) fn take_marked(&self) -> bool {
        self.marked.swap(false, Ordering::Relaxed)
    }

    /// This collection's own name.
    ///
    /// `ManagedOpen::bind_counter_dirty` sets this **unconditionally** for
    /// every managed-opened collection (not only counted ones), so it is
    /// already exactly "this collection's identity, bound once at managed
    /// open, off the hook chain" — the non-transactional claim guard (T10
    /// Task 5) reuses it for `DbError::UniqueIndexRequiresTx { collection,
    /// .. }` rather than inventing a second, claims-only binding for the same
    /// fact. A direct (`open_hooked`) open never binds this marker at all —
    /// see `check_no_counters` — so a claiming hook there has no name to
    /// report; callers fall back to `""`.
    pub(crate) fn name(&self) -> &'static str {
        self.name
    }
}

/// Fault injection for [`CounterDirty::mark`] — lets a test prove that a
/// failed mark **fails the open** of a counted source, so it never takes a
/// write whose delta could be lost with no durable evidence behind it.
/// Mirrors [`crate::projection::FAIL_COLLECTION_FLUSH`] and
/// `crate::counter::FAIL_COUNTER_DELTA`.
#[cfg(any(test, feature = "test-faults"))]
pub(crate) static FAIL_COUNTER_MARK: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// Simulate a `db.info` write that fails while persisting `counters_dirty`.
/// Compiles away entirely without `test-faults` (a non-default feature).
#[inline(always)]
fn counter_mark_fault() -> DbResult<()> {
    #[cfg(any(test, feature = "test-faults"))]
    if FAIL_COUNTER_MARK.load(std::sync::atomic::Ordering::SeqCst) {
        return Err(DbError::Internal("injected counters_dirty mark failure"));
    }
    Ok(())
}

/// Reject a counter carrier (`hook.counter_targets().is_some()`) opened
/// without a `Db` — there is nowhere to persist the dirty marker, so "the
/// engine repairs at the next open" would be a promise the code cannot keep.
///
/// Runs only on the direct `open_hooked`/`open_with_hook` entry points; a
/// managed open (`Db::open_*`) calls the private constructor directly and
/// binds the marker afterward via `ManagedOpen::bind_counter_dirty`, so it
/// never observes this check.
pub(crate) fn check_no_counters<K: Key, T, H: TypedWriteHook<K, T>>(hook: &H) -> DbResult<()> {
    if hook.counter_targets().is_some() {
        return Err(DbError::CountersRequireManagedOpen);
    }
    Ok(())
}