common/database/

write_guard.rs

// Generated by Qleany v1.11.0 from write_guard.tera

//! Enforces the single-write-transaction-per-store invariant.
//!
//! ## Why this exists
//!
//! [`Transaction::begin_write_transaction`](super::transactions::Transaction::begin_write_transaction)
//! takes an unconditional WHOLE-STORE savepoint via
//! `HashMapStore::create_savepoint`, whose own doc comment states the
//! assumption directly: taken "on the single writer thread with no
//! concurrent writer". `rollback()` and the `Drop` safety net on
//! `Transaction` both restore that savepoint wholesale — not just the rows
//! the transaction itself touched. That is intended behaviour for a single
//! writer; it silently breaks the moment two write transactions on the same
//! store are ever open at once. If a second write transaction were opened
//! concurrently with one already in flight, the second transaction's own
//! `commit`/`rollback`/`Drop` would silently roll back to a savepoint taken
//! *before* the first transaction's edits ever existed, erasing them with no
//! error and no event.
//!
//! [`WriteTransactionGuard`] turns that assumption into an assertion. Every
//! generated `CommandUnitOfWork::begin_transaction` acquires one for the
//! entire span its write transaction may be open (see `commit`/`rollback` on
//! any generated `*WriteUoW`/`*UnitOfWork`, which clear it). A second
//! concurrent acquisition **on the same store** panics in debug builds —
//! loud, immediate, naming the call site — and returns an error in release
//! builds, so a shipped build degrades to a reported failure instead of
//! silently corrupting the store.
//!
//! Because the acquire/release is generated into every write-transaction
//! call site's `begin_transaction`/`commit`/`rollback` (both the per-entity
//! `direct_access` `WriteUoW`s and the per-use-case feature `UnitOfWork`s),
//! this coverage never drifts out of sync with the rest of the generated
//! backend the way a hand-maintained list of call sites would: a new entity
//! or use case picks up the guard the moment it is generated, with nothing
//! left for the application developer to remember.
//!
//! ## Why scoped per store, not one bare global flag
//!
//! A production app constructs exactly one `DbContext` (hence one store) for
//! the whole process, so a single process-wide flag would be an equally
//! correct proxy for "a write transaction is open on THE store" there. But
//! tests often build a fresh, independent `DbContext`/store per `#[test]`
//! fn, and `cargo test`'s default runner executes those fns concurrently on
//! separate threads — two fns opening a write transaction on their OWN
//! unrelated stores can legitimately overlap in wall-clock time. A bare
//! process-wide flag would make those spuriously trip each other's guard.
//! Keying by the store's `Arc` pointer identity models the real invariant
//! (one writer *per store*, not one writer *ever*) without that flakiness —
//! and needs no test-only carve-out to stay correct.
//!
//! ## Holding the `Arc` — no stale-pointer ABA
//!
//! The map is keyed by `Arc::as_ptr(db_context.get_store())`, a raw address —
//! not by anything that keeps that address meaningful on its own. Storing
//! only that `usize` without cloning the `Arc` would be unsound: if the
//! `DbContext` a guard was acquired from is dropped (its last other clone
//! going out of scope) while the guard still lives, the allocator is free to
//! reuse that exact address for an unrelated, brand-new `HashMapStore`. The
//! new store's *first* real write transaction would then find the address
//! already marked as held — tripping the guard for a store that never had a
//! concurrent writer — and worse, when the stale guard eventually dropped it
//! would remove the *new* store's entry, silently re-opening the door the
//! guard exists to close.
//!
//! [`WriteTransactionGuard`] therefore clones the `Arc<HashMapStore>` itself
//! (the `store` field below) and holds that clone for its entire lifetime, in
//! addition to using its address as the map key. Holding the clone keeps the
//! allocation alive — the address cannot be freed, let alone reused for a
//! different store, for as long as any guard referencing it exists. The
//! pointer is still fine to use as a `HashMap` key (it is still unique among
//! *currently live* stores); what changed is that the guard now makes that
//! uniqueness durable for its own lifetime instead of merely observing it at
//! acquire time.
//!
//! ## Diagnosing contention: naming both the current holder and the new claimant
//!
//! Each slot records not just "held", but who holds it: the owning thread's
//! [`ThreadId`] and name, plus the `site` string passed
//! to `acquire`. A contending `acquire` call reads that record before failing,
//! so the panic/error names **both** parties — "`{new_site}` on thread {new}
//! collided with `{holder_site}` still held by thread {holder}" — not just the
//! new claimant. That is enough to `grep` straight to the offending call site
//! without attaching a debugger, which is the entire point of a guard whose
//! job is to fail loudly instead of silently.
//!
//! This module deliberately does **not** add any machinery to reclaim a slot
//! left held by a leaked or `mem::forget`-ed guard (a timeout, a generation
//! counter, a "steal the lock" escape hatch). Two things make that
//! unnecessary rather than merely deferred:
//!
//! * The ABA fix above already rules out the scenario that would make a stale
//!   key *dangerous* — a leaked guard cannot cause a *different, live* store to
//!   be wrongly blocked, because it holds the `Arc` that keeps its own store's
//!   address from ever being handed to a different store while the leak
//!   persists. A leak only ever poisons the one store it actually leaked on.
//! * For that one store, "a write-transaction slot that is held and never
//!   released" and "a write transaction that is still legitimately in
//!   progress" are the same observable state from this module's point of view.
//!   Any reclaim heuristic (e.g. "assume held-for-N-seconds means leaked") is a
//!   race against a legitimately long transaction and would trade a loud,
//!   honest failure for an occasional silent one — reintroducing exactly the
//!   failure mode this guard exists to prevent. The recorded thread id/site
//!   above turn that failure into an immediately diagnosable one (which call
//!   site leaked it), which is the correct fix for a bug, not a runtime
//!   work-around for one.
//!
//! ## Re-entrant `begin_transaction` on the same unit of work is diagnosed honestly
//!
//! Every generated `CommandUnitOfWork::begin_transaction` follows the same
//! shape:
//!
//! ```ignore
//! self.write_guard = Some(WriteTransactionGuard::acquire(&self.context, "some_use_case")?);
//! ```
//!
//! Assignment evaluates its right-hand side — the whole `acquire` call —
//! *before* dropping whatever `self.write_guard` held previously. So if
//! `begin_transaction` were ever called again on the *same* `CommandUnitOfWork`
//! instance while its own previous guard is still `Some` (a re-entrant call, or
//! a retry loop that calls `begin_transaction` again without going through
//! `commit`/`rollback` first — both of which clear the field), the new
//! `acquire` runs while the old guard is still registered as held. Naively that
//! looks identical to a genuine second writer, and the resulting panic/error
//! would misidentify a single UoW talking to itself as two concurrent writers
//! — plausible-sounding, and wrong, which is worse than no diagnostic at all.
//!
//! `acquire` tells the two apart using exactly the holder record described
//! above: if the slot is already held **by the current thread**, the message
//! says so explicitly — "this thread already holds this store's slot,
//! acquired at `{holder_site}`, this looks like a re-entrant/retried
//! `begin_transaction`, not a second writer" — instead of the generic
//! cross-thread contention wording. It still fails (the underlying invariant
//! violation — two live guards on one store — is exactly as real as in the
//! cross-thread case, and this module cannot see into the caller's struct to
//! silently drop the stale one for it), but the diagnostic now points at the
//! actual bug (a UoW re-entering `begin_transaction` without clearing its own
//! previous guard) instead of a phantom concurrent writer.
//!
//! ## RAII — cannot be defeated by an early return or a panic
//!
//! [`WriteTransactionGuard`] is a plain owned struct with only a `Drop` impl to
//! release its slot; there is no separate "release" method to forget to call.
//! Binding it to a named local for the whole span a write transaction may be
//! open means the slot is freed on every exit path — the happy path, an early
//! `?` return, or an unwinding panic — exactly like `Transaction`'s own `Drop`
//! safety net that it sits beside.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::thread::ThreadId;

use crate::database::db_context::DbContext;
use crate::database::hashmap_store::HashMapStore;

/// Who currently holds a store's write-transaction slot — enough to name them
/// in a contention panic/error (see the module doc's "Diagnosing contention"
/// section) instead of just saying "someone already holds it".
#[derive(Clone, Debug)]
struct Holder {
    thread_id: ThreadId,
    thread_name: Option<String>,
    site: &'static str,
}

impl Holder {
    fn describe(&self) -> String {
        match &self.thread_name {
            Some(name) => format!("{name:?} ({:?})", self.thread_id),
            None => format!("<unnamed> ({:?})", self.thread_id),
        }
    }
}

/// Which stores (keyed by `Arc<HashMapStore>` pointer identity) currently have a
/// write transaction in flight, and who holds each one. `None` until the first
/// acquire; lazily initialised rather than built from a const-unfriendly
/// `HashMap::new()` in a `static` initializer.
static WRITERS: Mutex<Option<HashMap<usize, Holder>>> = Mutex::new(None);

/// RAII token proving "I hold the write-transaction slot for this store." See
/// the module doc for the full rationale.
#[must_use = "the slot is released when this drops — binding it to `_` drops it \
              immediately and defeats the guard; bind it to a named local (e.g. \
              `write_guard`) that lives for the whole write-transaction span"]
#[derive(Debug)]
pub struct WriteTransactionGuard {
    key: usize,
    /// Never read directly — its only job is to keep the store's allocation
    /// (and hence `key`'s validity as a pointer identity) alive for as long as
    /// this guard exists. See the module doc's "Holding the `Arc`" section.
    #[allow(dead_code)]
    store: Arc<HashMapStore>,
}

impl WriteTransactionGuard {
    /// Claim the write-transaction slot for `db_context`'s store. `site` names
    /// the call site, used only in the panic/error message.
    ///
    /// # Panics
    /// In debug builds, panics immediately if this store's slot is already
    /// held — some other write transaction is already open on it. Turns the
    /// silent-data-loss failure mode this guards against into a loud one, at
    /// the exact call site that introduced the second writer (or, for a
    /// same-thread re-entry, the call site that never released its own
    /// previous guard — see the module doc).
    ///
    /// # Errors
    /// In release builds, returns an error instead of panicking — a shipped
    /// build should degrade to a reported failure, not abort the process.
    pub fn acquire(db_context: &DbContext, site: &'static str) -> anyhow::Result<Self> {
        // Clone (not just dereference) the `Arc`: the guard keeps this clone
        // alive for its whole lifetime so `key` can never be handed out to a
        // different store by the allocator while a guard referencing it
        // exists. See the module doc's "Holding the `Arc`" section.
        let store = Arc::clone(db_context.get_store());
        let key = Arc::as_ptr(&store) as usize;

        let current_thread = std::thread::current();
        let current_thread_id = current_thread.id();
        let current_thread_name = current_thread.name().map(str::to_owned);

        // Take the lock, mutate, and drop it again *before* possibly panicking —
        // panicking while holding a `std::sync::Mutex` poisons it for every
        // other store's acquire/drop for the rest of the process, which would
        // turn one store's legitimate bug into spurious failures for every
        // other, wholly unrelated store.
        let existing_holder = {
            let mut writers = WRITERS.lock().unwrap_or_else(|e| e.into_inner());
            let map = writers.get_or_insert_with(HashMap::new);
            match map.get(&key) {
                Some(holder) => Some(holder.clone()),
                None => {
                    map.insert(
                        key,
                        Holder {
                            thread_id: current_thread_id,
                            thread_name: current_thread_name,
                            site,
                        },
                    );
                    None
                }
            }
        };

        if let Some(holder) = existing_holder {
            let new_thread_desc = current_thread.name().map_or_else(
                || format!("{current_thread_id:?}"),
                |n| format!("{n:?} ({current_thread_id:?})"),
            );

            let message = if holder.thread_id == current_thread_id {
                format!(
                    "WriteTransactionGuard: `{site}` (thread {new_thread_desc}) tried to open a \
                     write transaction on a store THIS SAME THREAD already holds the slot for — \
                     acquired earlier at `{holder_site}` and never released. This is NOT a \
                     second concurrent writer; it is a re-entrant or retried `begin_transaction` \
                     on the same (or a nested) unit of work whose previous \
                     `WriteTransactionGuard` was never dropped — either a missing \
                     commit()/rollback() before the retry, or `self.write_guard = Some(new)` \
                     built `new` (running `acquire` again) before the old guard's `Drop` had a \
                     chance to run. Drop the previous guard (commit/rollback it, or explicitly \
                     `self.write_guard = None;`) before acquiring a new one.",
                    holder_site = holder.site,
                )
            } else {
                format!(
                    "WriteTransactionGuard: a second write transaction started at `{site}` \
                     (thread {new_thread_desc}) while another was already open on this store, \
                     held by thread {holder} since `{holder_site}`. \
                     Transaction::begin_write_transaction takes a WHOLE-STORE savepoint \
                     (common/src/database/transactions.rs) — a second concurrent writer's \
                     rollback/Drop would silently erase the first writer's committed edits. \
                     Fix the caller; do not remove or weaken this guard.",
                    holder = holder.describe(),
                    holder_site = holder.site,
                )
            };

            if cfg!(debug_assertions) {
                panic!("{message}");
            }
            anyhow::bail!(message);
        }

        Ok(WriteTransactionGuard { key, store })
    }
}

impl Drop for WriteTransactionGuard {
    fn drop(&mut self) {
        if let Some(map) = WRITERS.lock().unwrap_or_else(|e| e.into_inner()).as_mut() {
            map.remove(&self.key);
        }
    }
}

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

    fn ctx() -> DbContext {
        DbContext::new().unwrap()
    }

    /// Extracts the panic payload as a string regardless of whether it was
    /// boxed as `&'static str` (a literal) or `String` (an interpolated
    /// `format!`/`panic!("{message}")` payload, which is what this module's
    /// contention messages are).
    fn panic_message(payload: &(dyn std::any::Any + Send)) -> String {
        if let Some(s) = payload.downcast_ref::<&str>() {
            (*s).to_string()
        } else if let Some(s) = payload.downcast_ref::<String>() {
            s.clone()
        } else {
            panic!("panic payload was neither &str nor String");
        }
    }

    /// Drive a contending `acquire` and return its failure message, whichever
    /// way this build reports failure.
    ///
    /// `acquire` deliberately fails *differently* per profile — it panics
    /// under `debug_assertions` and returns an error without them — while
    /// producing an identical message either way. Tests that hard-coded the
    /// debug behaviour would fail the moment a generated app ran
    /// `cargo test --release`, and since this module is generated, a
    /// downstream developer could not fix those tests without their edit
    /// being erased by the next `qleany generate`. Absorbing the difference
    /// here keeps the suite green under both profiles and, more usefully,
    /// means the release path is genuinely exercised rather than assumed.
    fn contention_message(db: &DbContext, site: &'static str) -> String {
        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            WriteTransactionGuard::acquire(db, site)
        }));

        match outcome {
            Err(payload) => {
                // `if !cfg!(..) { panic!(..) }` rather than `assert!(cfg!(..), ..)`:
                // the condition is a compile-time constant, which clippy's
                // `assertions_on_constants` rejects under `-D warnings`. The
                // check is worth keeping — it is what proves the profile and the
                // observed behaviour agree — so it is restated, not removed.
                if !cfg!(debug_assertions) {
                    panic!(
                        "`{site}`: a release build must report contention by returning an \
                         error, not by panicking"
                    );
                }
                panic_message(payload.as_ref())
            }
            Ok(Err(error)) => {
                if cfg!(debug_assertions) {
                    panic!(
                        "`{site}`: a debug build must report contention by panicking loudly \
                         at the offending call site, not by returning an error"
                    );
                }
                format!("{error}")
            }
            Ok(Ok(_)) => panic!(
                "`{site}` acquired a slot that was already held — the guard did not fire \
                 at all, which is the exact silent failure it exists to prevent"
            ),
        }
    }

    #[test]
    fn sequential_acquire_release_never_trips_the_guard() {
        let db = ctx();
        for _ in 0..3 {
            let g = WriteTransactionGuard::acquire(&db, "test").unwrap();
            drop(g);
        }
    }

    #[test]
    fn two_different_stores_can_each_acquire_independently() {
        let a = ctx();
        let b = ctx();
        let _ga = WriteTransactionGuard::acquire(&a, "a").unwrap();
        let _gb = WriteTransactionGuard::acquire(&b, "b").unwrap();
        // No panic reaching here is the assertion: two unrelated stores must
        // never contend on each other's slot.
    }

    #[test]
    fn dropping_the_first_guard_frees_the_slot_for_a_second_acquire() {
        let db = ctx();
        let first = WriteTransactionGuard::acquire(&db, "first").unwrap();
        drop(first);
        let second = WriteTransactionGuard::acquire(&db, "second");
        assert!(
            second.is_ok(),
            "the slot must be free once the first guard dropped"
        );
    }

    #[test]
    fn a_second_concurrent_acquire_on_the_same_store_fails() {
        let db = ctx();
        let _first = WriteTransactionGuard::acquire(&db, "first").unwrap();
        let message = contention_message(&db, "second");
        assert!(
            message.contains("second"),
            "message must name the call site that was refused: {message}"
        );

        // The failure must not have poisoned the global map for anyone else: a
        // fresh, unrelated store can still acquire immediately afterward.
        let other = ctx();
        assert!(WriteTransactionGuard::acquire(&other, "other").is_ok());
    }

    #[test]
    fn the_guard_keeps_the_store_allocation_alive_after_the_db_context_drops() {
        let db = ctx();
        let store_ref = Arc::clone(db.get_store());
        // `db` itself + our `store_ref` clone.
        assert_eq!(Arc::strong_count(&store_ref), 2);

        let guard = WriteTransactionGuard::acquire(&db, "test").unwrap();
        // Acquiring must have taken its own clone: three owners now.
        assert_eq!(
            Arc::strong_count(&store_ref),
            3,
            "WriteTransactionGuard::acquire must clone the store's Arc, not just read its pointer"
        );

        drop(db);
        // With the DbContext gone, only the guard's clone (+ ours) keep the
        // allocation alive. If the guard did NOT hold a clone, this would
        // drop to 1 and the address would become free for the allocator to
        // hand to an unrelated, brand-new store — the ABA scenario the
        // module doc describes.
        assert_eq!(
            Arc::strong_count(&store_ref),
            2,
            "the guard must still be holding the store alive once the DbContext drops"
        );

        drop(guard);
        assert_eq!(
            Arc::strong_count(&store_ref),
            1,
            "the guard's clone must be released when the guard drops"
        );
    }

    #[test]
    fn cross_thread_contention_names_both_the_holder_and_the_new_claimant() {
        let db = ctx();
        let _first = WriteTransactionGuard::acquire(&db, "first_site_xyz").unwrap();

        let db_for_thread = db.clone();
        let handle = std::thread::Builder::new()
            .name("contender-thread".to_string())
            .spawn(move || contention_message(&db_for_thread, "second_site_xyz"))
            .unwrap();

        let message = handle
            .join()
            .expect("the contending thread must fail through the guard, not around it");

        assert!(
            message.contains("first_site_xyz"),
            "message must name the current holder's call site: {message}"
        );
        assert!(
            message.contains("second_site_xyz"),
            "message must name the new claimant's call site: {message}"
        );
        assert!(
            message.contains("contender-thread"),
            "message must name the new claimant's thread: {message}"
        );
        assert!(
            !message.contains("SAME THREAD"),
            "a genuine cross-thread writer must not be misdiagnosed as same-thread \
             re-entry: {message}"
        );
    }

    #[test]
    fn same_thread_reacquire_is_diagnosed_as_self_reentry_not_a_phantom_second_writer() {
        let db = ctx();
        let _first = WriteTransactionGuard::acquire(&db, "begin_transaction_first_call").unwrap();

        let message = contention_message(&db, "begin_transaction_retry");

        assert!(
            message.contains("begin_transaction_first_call"),
            "message must name the site that still holds the slot: {message}"
        );
        assert!(
            message.contains("begin_transaction_retry"),
            "message must name the re-entrant call site: {message}"
        );
        assert!(
            message.contains("SAME THREAD") || message.contains("re-entrant"),
            "message must say this is the same thread re-entering, not blame a second \
             writer: {message}"
        );
        assert!(
            !message.contains("Fix the caller; do not remove or weaken this guard"),
            "must not reuse the generic cross-thread-second-writer wording for a \
             same-thread self re-entry: {message}"
        );
    }
}