armdb 0.14.2

//! The pure claim-resolution rule (T10 Task 7).
//!
//! [`decide`] turns one claim/release/reservation *intent* plus the
//! namespace row's *current* state into a *decision* — write, delete,
//! no-op, or conflict. It is deliberately storage-agnostic: the caller
//! supplies `current` from wherever it reads it and turns the decision into
//! an actual write (or, for a read-only caller, a finding). Two callers share
//! it today, and are built to never disagree by construction:
//! `txn::projection_commit::resolve_claims`, which reads `current` from a
//! transaction's own pending write first and storage only otherwise (the
//! same overlay rule `resolve_counters` uses), and
//! `schema::registry::run_decls`'s unique-namespace pass (T10 Task 9), which
//! reads `current` from a plain [`crate::unique::ClaimTarget::scan_rows`] and
//! turns `ClaimDecision::Conflict`/`Write` into
//! `Finding::UniqueConflict`/`UniqueMissing` instead of a write. `read_row`
//! (a point-read seam the validator could have used instead) was removed as
//! dead code before this landed — `scan_rows` already hands the validator
//! every row it needs more cheaply than a point read per expected key would.
//!
//! The rule is a straight table on `(current owner, kind)`:
//!
//! | current | kind | decision |
//! |---|---|---|
//! | absent | Claim | write the claimant, permanent |
//! | absent | Reserve | write the claimant, with the deadline |
//! | absent | Release | no-op |
//! | owner is someone else | any | conflict |
//! | owner is me, no deadline | Claim | no-op |
//! | owner is me, has a deadline | Claim | rewrite as permanent |
//! | owner is me, has a deadline | Reserve | rewrite with the new deadline (refresh) |
//! | owner is me, no deadline | Reserve | no-op — a permanent claim outranks a reservation |
//! | owner is me | Release | delete |
//!
//! The "someone else holds it" arm sits **above** the release and claim
//! arms in [`decide`]'s `match`, so a foreign owner is decided once for
//! every [`ClaimKind`] — a release must never delete a third party's row.
//! A permanent claim outranking a reservation (review round C1) is what keeps
//! a re-issued/retried `reserve` on an already-promoted key idempotent
//! instead of destructive: without it, `reserve` reads as safe to retry (its
//! own doc says it "refreshes the deadline") but would silently demote a
//! permanent claim back into an expiring reservation, which the TTL sweeper
//! then deletes out from under a live, promoted claim.

use crate::txn::participant::ClaimKind;
use crate::unique::{UniqueOwner, UniqueRow};

/// What [`decide`] says to do with one namespace row.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum ClaimDecision {
    /// Write this row — a fresh claim, a fresh reservation, a refreshed
    /// reservation, or a reservation becoming a permanent claim.
    Write(UniqueRow),
    /// The claimant releases a row it owns: delete it.
    Delete,
    /// Nothing changes: a release of an already-absent row, or a claim the
    /// claimant already holds permanently.
    NoOp,
    /// Someone else holds the row. Carries the current owner, for the
    /// caller's [`crate::error::DbError::UniqueViolation`] (or, later, a
    /// validator finding).
    Conflict(UniqueOwner),
}

/// The pure claim-resolution rule (T10 Task 7). See the module docs for the
/// full table; `current` is the namespace row's state as the caller sees it
/// — this function does not read or write storage itself.
pub(crate) fn decide(
    current: Option<UniqueRow>,
    claimant: &UniqueOwner,
    kind: ClaimKind,
) -> ClaimDecision {
    match (current.map(|r| r.owner()), kind) {
        // `Claim` is always permanent — explicit here via `with_deadline(0)`,
        // not inherited from whatever `expires_at` the claimant happens to
        // carry (review round M1). `into_tx_claim_ops` builds a plain claim's
        // owner with `expires_at = 0` today, so this is currently a no-op on
        // its input either way — but Task 8 wires reserve->claim promotion,
        // where the natural call *is* `decide(.., &reservation_owner,
        // ClaimKind::Claim)`. Trusting the claimant's own deadline there
        // would silently leave the row a reservation for the TTL sweeper to
        // delete out from under what is supposed to be a permanent claim.
        (None, ClaimKind::Claim) => ClaimDecision::Write(UniqueRow::new(claimant.with_deadline(0))),
        (None, ClaimKind::Reserve { deadline_ms }) => {
            ClaimDecision::Write(UniqueRow::new(claimant.with_deadline(deadline_ms)))
        }
        (None, ClaimKind::Release) => ClaimDecision::NoOp,
        // A foreign owner is decided once, above every kind below — in
        // particular, a Release must never delete a row that is not the
        // claimant's own.
        (Some(cur), _) if !cur.same_holder(claimant) => ClaimDecision::Conflict(cur),
        (Some(_), ClaimKind::Release) => ClaimDecision::Delete,
        (Some(cur), ClaimKind::Claim) => {
            // Mine already: a reservation becomes permanent ownership,
            // otherwise this is a genuine no-op. See the absent-row arm
            // above for why the write is `with_deadline(0)` rather than
            // `*claimant` verbatim.
            if cur.is_reservation() {
                ClaimDecision::Write(UniqueRow::new(claimant.with_deadline(0)))
            } else {
                ClaimDecision::NoOp
            }
        }
        (Some(cur), ClaimKind::Reserve { deadline_ms }) => {
            // Mine already: a reservation refreshes its own deadline, but a
            // *permanent* claim outranks a reservation — `reserve` must not
            // be able to demote a permanent claim back into an expiring one.
            // This is what makes a re-issued/retried `reserve` on a key its
            // caller already promoted idempotent rather than destructive: a
            // "hold this name 15 minutes" request from someone who already
            // holds it forever is already satisfied, not an error (that
            // would break the legitimate retry) and not a write (nothing to
            // change).
            if cur.is_reservation() {
                ClaimDecision::Write(UniqueRow::new(claimant.with_deadline(deadline_ms)))
            } else {
                ClaimDecision::NoOp
            }
        }
    }
}

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

    fn owner(tag: u64, key: &[u8]) -> UniqueOwner {
        UniqueOwner::new(tag, key, 0).unwrap()
    }

    #[test]
    fn absent_row_claim_writes_a_permanent_owner() {
        let me = owner(1, b"a");
        let decision = decide(None, &me, ClaimKind::Claim);
        match decision {
            ClaimDecision::Write(row) => {
                assert_eq!(row.owner(), me);
                assert!(!row.owner().is_reservation());
                assert_eq!(row.deadline(), None);
            }
            other => panic!("expected Write, got {other:?}"),
        }
    }

    /// Review round M1: `Claim` must write a permanent row even when the
    /// *claimant itself* carries a reservation deadline — `decide` must not
    /// trust `expires_at` on its input for a `Claim`, only on `Reserve`.
    /// Today `into_tx_claim_ops` never builds a claimant with a nonzero
    /// deadline, so this only distinguishes the fix from the bug the review
    /// flagged (Task 8's reserve->claim promotion is the real caller that
    /// will pass a reservation owner here).
    #[test]
    fn claim_ignores_a_deadline_the_claimant_itself_happens_to_carry() {
        let reservation_owner = owner(1, b"a").with_deadline(999);
        assert!(
            reservation_owner.is_reservation(),
            "fixture must actually carry a deadline, or this test cannot tell \
             'decide clears it' from 'there was nothing to clear'"
        );
        match decide(None, &reservation_owner, ClaimKind::Claim) {
            ClaimDecision::Write(row) => {
                assert_eq!(
                    row.deadline(),
                    None,
                    "Claim must always write a permanent row, regardless of what \
                     deadline the claimant itself carries"
                );
            }
            other => panic!("expected Write, got {other:?}"),
        }
    }

    #[test]
    fn absent_row_reserve_writes_a_reservation_with_the_deadline() {
        let me = owner(1, b"a");
        let decision = decide(None, &me, ClaimKind::Reserve { deadline_ms: 42 });
        match decision {
            ClaimDecision::Write(row) => {
                assert_eq!(row.deadline(), Some(42));
                assert!(row.owner().same_holder(&me));
            }
            other => panic!("expected Write, got {other:?}"),
        }
    }

    #[test]
    fn absent_row_release_is_a_noop() {
        let me = owner(1, b"a");
        assert_eq!(decide(None, &me, ClaimKind::Release), ClaimDecision::NoOp);
    }

    #[test]
    fn my_own_permanent_row_claim_is_a_noop() {
        let me = owner(1, b"a");
        let current = UniqueRow::new(me);
        assert_eq!(
            decide(Some(current), &me, ClaimKind::Claim),
            ClaimDecision::NoOp
        );
    }

    #[test]
    fn my_own_reservation_claim_rewrites_it_permanent() {
        let me = owner(1, b"a");
        let reserved = me.with_deadline(99);
        let current = UniqueRow::new(reserved);
        match decide(Some(current), &me, ClaimKind::Claim) {
            ClaimDecision::Write(row) => {
                assert_eq!(
                    row.deadline(),
                    None,
                    "a claim clears the reservation deadline"
                );
                assert_eq!(row.owner(), me);
            }
            other => panic!("expected Write, got {other:?}"),
        }
    }

    #[test]
    fn my_own_row_reserve_refreshes_the_deadline() {
        let me = owner(1, b"a");
        let current = UniqueRow::new(me.with_deadline(10));
        match decide(Some(current), &me, ClaimKind::Reserve { deadline_ms: 20 }) {
            ClaimDecision::Write(row) => assert_eq!(row.deadline(), Some(20)),
            other => panic!("expected Write, got {other:?}"),
        }
    }

    /// Review round C1: `reserve` on a key its own caller already holds
    /// *permanently* must not demote it back into an expiring reservation —
    /// a permanent claim outranks a reservation. Without this, `reserve`'s
    /// own doc ("refreshes the deadline") reads as safe to retry, but a
    /// retried `reserve` after the real claim already landed would silently
    /// hand the row back to the TTL sweeper.
    #[test]
    fn my_own_permanent_row_reserve_is_a_noop_not_a_demotion() {
        let me = owner(1, b"a");
        let current = UniqueRow::new(me); // permanent: no deadline
        assert_eq!(
            decide(Some(current), &me, ClaimKind::Reserve { deadline_ms: 999 }),
            ClaimDecision::NoOp,
            "a permanent claim must outrank a reservation, not be overwritten by one"
        );
    }

    #[test]
    fn my_own_row_release_deletes_it() {
        let me = owner(1, b"a");
        let current = UniqueRow::new(me);
        assert_eq!(
            decide(Some(current), &me, ClaimKind::Release),
            ClaimDecision::Delete
        );
    }

    #[test]
    fn a_foreign_owner_conflicts_on_claim() {
        let mine = owner(1, b"me");
        let foreign = owner(1, b"someone-else");
        let current = UniqueRow::new(foreign);
        match decide(Some(current), &mine, ClaimKind::Claim) {
            ClaimDecision::Conflict(cur) => assert_eq!(cur, foreign),
            other => panic!("expected Conflict, got {other:?}"),
        }
    }

    /// The rule table's most load-bearing line: a release must never delete
    /// a row someone else holds — a stray release racing a foreign claim
    /// must not destroy that claim.
    #[test]
    fn a_foreign_owner_conflicts_on_release_rather_than_being_deleted() {
        let mine = owner(1, b"me");
        let foreign = owner(1, b"someone-else");
        let current = UniqueRow::new(foreign);
        match decide(Some(current), &mine, ClaimKind::Release) {
            ClaimDecision::Conflict(cur) => assert_eq!(cur, foreign),
            other => panic!("expected Conflict, not a delete: got {other:?}"),
        }
    }

    #[test]
    fn a_foreign_owner_conflicts_on_reserve() {
        let mine = owner(1, b"me");
        let foreign = owner(1, b"someone-else");
        let current = UniqueRow::new(foreign);
        match decide(Some(current), &mine, ClaimKind::Reserve { deadline_ms: 5 }) {
            ClaimDecision::Conflict(cur) => assert_eq!(cur, foreign),
            other => panic!("expected Conflict, got {other:?}"),
        }
    }

    /// Same key bytes under a *different* collection tag are a different
    /// holder — `same_holder` compares the tag too, not just the key.
    #[test]
    fn a_different_collection_tag_with_the_same_key_bytes_still_conflicts() {
        let a = owner(1, b"same-key");
        let b = owner(2, b"same-key");
        let current = UniqueRow::new(a);
        match decide(Some(current), &b, ClaimKind::Claim) {
            ClaimDecision::Conflict(cur) => assert_eq!(cur, a),
            other => panic!("expected Conflict, got {other:?}"),
        }
    }
}