armdb 0.14.2

//! T10 Task 8: reservations and the deadline policy.
//!
//! `Tx::reserve`/`Tx::release` queue claim-stage intents — not raw writes —
//! that `resolve_claims` decides under the same two-level partition a
//! projected claim goes through (T10 Task 7). A reservation is *the same
//! owner row with a deadline*: the takeover when the source row finally
//! claims the same key is the ordinary "owner is me, has a deadline" branch
//! of `unique::resolve::decide`, not a special case.
//!
//! Fixture mirrors `unique_tx.rs`'s `typed_map` shape (`User`/`UsernameKey`/
//! `Usernames`) — same types, so a reader who knows that file's fixture
//! recognizes this one immediately.

#![cfg(all(
    feature = "armour",
    feature = "schema",
    feature = "typed-tree",
    feature = "rapira-codec"
))]

use std::sync::Arc;
use std::time::Duration;

use armdb::armour::{Db, WorkerIntervals};
use armdb::{Config, FixedConfig, NoHook, Projected, RapiraCodec, UniqueIndex, UniqueTargets};
use armour_core::{Fuid, GetType};
use rapira::Rapira;

armour_core::const_hasher_or!(user_id, "USR", "dev-key-user_id", brand = "user");
type UserId = Fuid<user_id::Hasher>;

fn id(n: u64) -> UserId {
    Fuid::from_be_bytes(n.to_be_bytes())
}

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    Hash,
    zerocopy::FromBytes,
    zerocopy::IntoBytes,
    zerocopy::Immutable,
    zerocopy::KnownLayout,
)]
#[repr(C)]
pub struct UsernameKey([u8; 24]);
armdb::impl_key_zerocopy!(
    UsernameKey,
    armour_core::KeyScheme::Typed(&[armour_core::KeyType::Array(24)])
);

impl UsernameKey {
    fn new(s: &str) -> Self {
        let mut b = [0u8; 24];
        b[..s.len()].copy_from_slice(s.as_bytes());
        Self(b)
    }
}

armdb::unique_index!(
    Usernames,
    key = UsernameKey,
    name = "unique_reservation_usernames"
);

fn open_usernames(db: &Arc<Db>) -> Arc<UniqueIndex<Usernames>> {
    db.open_unique_index::<Usernames>(FixedConfig::test())
        .expect("open usernames namespace")
}

/// `avatar` is along for the ride — mirrors `unique_tx.rs`'s `User`.
#[derive(Clone, Debug, PartialEq, GetType, Rapira)]
struct User {
    username: Option<String>,
    avatar: u32,
}
impl armdb::CollectionMeta for User {
    type SelfId = UserId;
    const NAME: &'static str = "unique_reservation_users";
}

/// Claims `value.username` (when set) into `usernames` — the projected half
/// of the "explicit reserve + projected claim" interaction this task's
/// matrix exercises.
type UserHook = Projected<UserId, User, NoHook>;

fn claim_hook_user(usernames: &Arc<UniqueIndex<Usernames>>) -> UserHook {
    let targets = UniqueTargets::new().with(usernames);
    let ns = Arc::clone(usernames);
    Projected::<UserId, User>::new(armdb::IndexTargets::new(), |_, _, _| Ok(())).and_unique(
        targets,
        move |_k, v: &User, sink| {
            if let Some(u) = &v.username {
                sink.claim(&ns, UsernameKey::new(u))?;
            }
            Ok(())
        },
    )
}

type Coll = armdb::typed_map!(User, RapiraCodec, UserHook);

armour_core::const_hasher_or!(session_id, "SESS", "dev-key-session_id", brand = "session");
type SessionId = Fuid<session_id::Hasher>;

fn sid(n: u64) -> SessionId {
    Fuid::from_be_bytes(n.to_be_bytes())
}

/// `Result::unwrap_err` requires the `Ok` side to be `Debug`, and `Tx` isn't
/// — mirrors `unique_nontx.rs`'s helper of the same name/shape.
fn take_err<T>(r: armdb::DbResult<T>) -> armdb::DbError {
    match r {
        Ok(_) => panic!("expected an Err, got Ok"),
        Err(e) => e,
    }
}

/// A collection that never declares `usernames` as a claim target — I1's
/// "wrong source" fixture: `reserve`/`release` must refuse it, the same way
/// `ClaimSink::claim` refuses an undeclared namespace on the projected path.
#[derive(Clone, Debug, PartialEq, GetType, Rapira)]
struct Session {
    active: bool,
}
impl armdb::CollectionMeta for Session {
    type SelfId = SessionId;
    const NAME: &'static str = "unique_reservation_sessions";
}

type SessionColl = armdb::typed_map!(Session, RapiraCodec);

fn open_sessions(db: &Arc<Db>) -> Arc<SessionColl> {
    db.open_typed_map::<Session, RapiraCodec, _>(Config::test(), NoHook, &[])
        .unwrap()
}

/// Assert `deadline` sits within a generous window of `now_before_ms + ttl_ms`
/// — tight enough to catch a unit-scaling bug (e.g. seconds mistaken for
/// milliseconds, a ×1000 error), loose enough to tolerate real commit and
/// scheduling latency between reading `now_before_ms` and the resolver's own
/// clock read.
fn assert_deadline_near(deadline: u64, now_before_ms: u64, ttl_ms: u64) {
    let target = now_before_ms + ttl_ms;
    let window_ms = 10_000;
    assert!(
        deadline + window_ms >= target && deadline <= target + window_ms,
        "deadline {deadline} not within {window_ms}ms of now_before({now_before_ms}) + ttl({ttl_ms}) = {target}"
    );
}

fn fixture() -> (
    tempfile::TempDir,
    Arc<Db>,
    Arc<UniqueIndex<Usernames>>,
    Arc<Coll>,
) {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let usernames = open_usernames(&db);
    let hook = claim_hook_user(&usernames);
    let users = db
        .open_typed_map::<User, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();
    (dir, db, usernames, users)
}

fn workers(interval_ms: u64, budget: usize) -> WorkerIntervals {
    WorkerIntervals {
        compaction: None,
        flush: None,
        ttl_sweep: Some(Duration::from_millis(interval_ms)),
        ttl_budget: budget,
    }
}

/// Same shape as `fixture()`, but the `Db` runs a real background TTL sweep
/// worker on a short interval instead of `Db::open_test`'s disabled one — for
/// the sweep tests, which need an actual pass to run rather than a
/// synchronous `sweep_expired(now_ms, budget)` call.
fn fixture_with_fast_sweeper() -> (
    tempfile::TempDir,
    Arc<Db>,
    Arc<UniqueIndex<Usernames>>,
    Arc<Coll>,
) {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_with_workers(dir.path(), workers(20, 4096)).unwrap());
    let usernames = open_usernames(&db);
    let hook = claim_hook_user(&usernames);
    let users = db
        .open_typed_map::<User, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();
    (dir, db, usernames, users)
}

// ===========================================================================
// Core matrix (task brief, Step 1).
// ===========================================================================

#[test]
fn reservation_holds_the_key_and_is_taken_over_by_its_own_source_row() {
    let (_dir, db, usernames, users) = fixture();
    let now_before_ms = armdb::testing::now_ms();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    let owner = usernames.owner_of(&UsernameKey::new("rail")).unwrap();
    assert!(owner.is_reservation());
    assert_eq!(owner.belongs_to::<User>(), Some(id(1)));
    // Review round M5(d): the deadline's *magnitude*, not just its presence —
    // the only other numeric check in this file is on a 1 ms ttl, which a
    // ×1000 unit-scaling bug (e.g. `now + ttl.as_secs()`) would pass just as
    // well as a correct implementation.
    assert_deadline_near(owner.expires_at().unwrap(), now_before_ms, 900_000);

    // A second owner cannot take it.
    assert!(matches!(
        db.tx()
            .put(
                &users,
                &id(2),
                &User {
                    username: Some("rail".into()),
                    avatar: 1,
                },
            )
            .unwrap()
            .commit()
            .unwrap_err(),
        armdb::DbError::UniqueViolation { .. }
    ));

    // The rightful owner's row clears the deadline.
    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();
    assert!(
        !usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .is_reservation()
    );
}

#[test]
fn release_is_idempotent_and_refuses_a_foreign_owner() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    // Wrong owner: refused, row survives.
    assert!(matches!(
        db.tx()
            .release(&usernames, UsernameKey::new("rail"), &users, &id(2))
            .unwrap()
            .commit()
            .unwrap_err(),
        armdb::DbError::UniqueViolation { .. }
    ));
    assert!(usernames.contains(&UsernameKey::new("rail")));

    // Right owner: gone, and a second release is a no-op success.
    db.tx()
        .release(&usernames, UsernameKey::new("rail"), &users, &id(1))
        .unwrap()
        .commit()
        .unwrap();
    assert!(!usernames.contains(&UsernameKey::new("rail")));
    db.tx()
        .release(&usernames, UsernameKey::new("rail"), &users, &id(1))
        .unwrap()
        .commit()
        .unwrap();
}

#[test]
fn reservation_deadline_conversion_is_bounded() {
    let (_dir, db, usernames, users) = fixture();
    // Zero is rejected: an expired-on-arrival reservation is indistinguishable
    // from a leak, and the caller means `release`.
    assert!(matches!(
        db.tx().reserve(
            &usernames,
            UsernameKey::new("a"),
            &users,
            &id(1),
            Duration::ZERO
        ),
        Err(armdb::DbError::UniqueReservationInvalid {
            reason: "zero duration",
            ..
        })
    ));
    // Sub-millisecond rounds up rather than landing in the past. `now_ms` is
    // read *before* the reserve/commit, not after: the deadline is stamped
    // once, inside `commit()`, and a real commit's own log/apply work can
    // easily take longer than the 1 ms TTL itself — reading "now" again only
    // after `commit()` returns would race that work and could fail even
    // though the deadline was computed correctly. Reading it first still
    // pins the property this case exists to check: the deadline is not
    // somewhere in the past relative to when the reservation was requested.
    let now_before_ms = armdb::testing::now_ms();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("b"),
            &users,
            &id(1),
            Duration::from_micros(1),
        )
        .unwrap()
        .commit()
        .unwrap();
    let deadline = usernames
        .owner_of(&UsernameKey::new("b"))
        .unwrap()
        .expires_at()
        .unwrap();
    assert!(
        deadline >= now_before_ms,
        "deadline {deadline} must not land before now_before_ms {now_before_ms}"
    );
    // Sanity upper bound — generous, to stay clear of any real scheduling
    // jitter, but still tight enough to catch a deadline computed from the
    // wrong units (e.g. seconds instead of milliseconds).
    assert!(
        deadline <= now_before_ms + 60_000,
        "deadline {deadline} is implausibly far from now_before_ms {now_before_ms} for a 1 ms ttl"
    );
    // Overflow is rejected, not wrapped.
    assert!(matches!(
        db.tx().reserve(
            &usernames,
            UsernameKey::new("c"),
            &users,
            &id(1),
            Duration::from_secs(u64::MAX),
        ),
        Err(armdb::DbError::UniqueReservationInvalid {
            reason: "deadline overflow",
            ..
        })
    ));
}

/// The builder's own `Duration -> ms` conversion only ever overflows when the
/// *ttl itself* does not fit `u64` milliseconds (pinned above). `now + ttl`
/// is a distinct overflow the resolver checks, under the namespace lock, at
/// `commit()` — reachable with a `ttl` that fits `u64` on its own but leaves
/// no headroom once added to the real wall clock.
#[test]
fn reservation_deadline_overflow_at_resolve_time_is_rejected() {
    let (_dir, db, usernames, users) = fixture();
    let huge = Duration::from_millis(u64::MAX - 1_000);

    let err = db
        .tx()
        .reserve(&usernames, UsernameKey::new("huge"), &users, &id(1), huge)
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(
        matches!(
            err,
            armdb::DbError::UniqueReservationInvalid {
                reason: "deadline overflow",
                ..
            }
        ),
        "{err}"
    );
    assert!(!usernames.contains(&UsernameKey::new("huge")));
}

/// Review round I2: the original version of this test polled
/// `!usernames.contains(key)` right after `commit()` and never checked the
/// row existed first — had `reserve` silently written nothing, the test
/// would have passed on the very first poll, in microseconds, having proven
/// nothing about the sweeper. Fixed two ways: assert the row is actually
/// there before doing anything that could remove it, and drive the sweep
/// through `Db::sweep_expired_now` (`ttl_sweep.rs`'s
/// `db_sweep_expired_now_aggregates` pattern) — a synchronous call that
/// returns a deleted count, works against `fixture()`'s disabled background
/// worker, and turns "gone by some deadline" into a direct assertion that
/// *the sweeper* did the deleting.
#[test]
fn expired_reservation_is_swept_and_the_key_becomes_free() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_millis(1),
        )
        .unwrap()
        .commit()
        .unwrap();
    assert!(
        usernames.contains(&UsernameKey::new("rail")),
        "the reservation must actually have been written before its removal proves anything"
    );

    // The 1 ms TTL above only needs to have elapsed by the time
    // `sweep_expired_now` reads the wall clock — a real commit (log append,
    // apply) already takes longer than that on its own, and this sleep adds
    // a wide margin on top.
    std::thread::sleep(Duration::from_millis(50));
    let deleted = db.sweep_expired_now().unwrap();
    assert!(
        deleted >= 1,
        "sweep_expired_now must have deleted the expired reservation"
    );
    assert!(!usernames.contains(&UsernameKey::new("rail")));

    // And a different owner can now take it.
    db.tx()
        .put(
            &users,
            &id(2),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();
}

#[test]
fn a_permanent_row_is_never_swept() {
    let (_dir, db, usernames, users) = fixture_with_fast_sweeper();
    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();
    std::thread::sleep(Duration::from_millis(300));
    assert!(usernames.contains(&UsernameKey::new("rail")));
}

// ===========================================================================
// Explicit reserve + projected claim, same transaction.
//
// A wrong design here silently degrades to last-write-wins (see this
// module's own doc and `txn::projection_commit::resolve_claims`'s). The
// engine places every explicit acquisition *ahead* of every projected one,
// regardless of the order the caller chained `.reserve()`/`.put()` — these
// three tests are that guarantee, checked from the outside.
// ===========================================================================

/// `.reserve(..).put(..)` in one transaction: the reservation resolves
/// first, then the source row's own projected claim finds "owner is me, has
/// a deadline" and promotes it — same as the two-transaction case, just
/// collapsed into one.
#[test]
fn same_transaction_reserve_then_claim_promotes_to_permanent() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();

    let owner = usernames.owner_of(&UsernameKey::new("rail")).unwrap();
    assert!(
        !owner.is_reservation(),
        "the source row's own claim must promote the reservation to permanent"
    );
    assert_eq!(owner.belongs_to::<User>(), Some(id(1)));
}

/// The same outcome with the builder calls in the **opposite** order:
/// `.put(..).reserve(..)`. If resolution order followed builder call order
/// instead of the engine's own explicit-before-projected rule, this would
/// instead leave the row a reservation — `decide`'s `Reserve` arm on an
/// already-owned row unconditionally refreshes the deadline, so a
/// reservation resolving *after* its own takeover claim would silently turn
/// a permanent claim back into an expiring one.
#[test]
fn same_transaction_claim_then_reserve_still_promotes_to_permanent() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    let owner = usernames.owner_of(&UsernameKey::new("rail")).unwrap();
    assert!(
        !owner.is_reservation(),
        "builder call order must not change the outcome: the row must still end up permanent"
    );
    assert_eq!(owner.belongs_to::<User>(), Some(id(1)));
}

/// An explicit reservation for one owner and a projected claim for a
/// *different* owner, same key, same transaction: a genuine conflict, and
/// the whole transaction is refused — neither the reservation nor the
/// competing claim lands.
#[test]
fn same_transaction_reserve_and_a_different_owners_claim_conflict() {
    let (_dir, db, usernames, users) = fixture();
    let err = db
        .tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .put(
            &users,
            &id(2),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(
        matches!(err, armdb::DbError::UniqueViolation { .. }),
        "{err}"
    );

    // The whole transaction is refused: neither row landed.
    assert_eq!(usernames.owner_of(&UsernameKey::new("rail")), None);
    assert!(users.get(&id(1)).is_none());
    assert!(users.get(&id(2)).is_none());
}

// ===========================================================================
// Review round fixes.
// ===========================================================================

/// Review round C1 (Critical): `reserve` on a key its own caller already
/// promoted to a permanent claim must not demote it back into an expiring
/// reservation. This is the end-to-end version of the four-step failure the
/// review described: reserve, the real row lands and promotes it (signup
/// done), an at-least-once job or client retry re-issues the same reserve —
/// `reserve`'s own doc says it "refreshes the deadline", so it reads as safe
/// to retry — and without the fix that retry would silently hand a live,
/// permanent claim back to the TTL sweeper.
#[test]
fn reserve_on_a_key_i_already_own_permanently_does_not_demote_it() {
    let (_dir, db, usernames, users) = fixture();

    // 1. Reserve.
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    // 2. The real row lands and promotes it to permanent.
    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();
    assert!(
        !usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .is_reservation(),
        "fixture must actually be permanent before the retry, or this test proves nothing"
    );

    // 3. A retried/duplicate reserve re-issues the same request.
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    // 4. Still permanent — not silently demoted to an expiring reservation.
    let owner = usernames.owner_of(&UsernameKey::new("rail")).unwrap();
    assert!(
        !owner.is_reservation(),
        "a permanent claim must outrank a reservation: the retried reserve must not demote it"
    );
    assert_eq!(owner.belongs_to::<User>(), Some(id(1)));
}

/// Review round I1 (Important): `reserve`/`release` must refuse a `source`
/// that never declared `ns` as a claim target — the explicit-op twin of
/// `ClaimSink::claim`'s `UndeclaredUniqueTarget` on the projected path.
/// Without this check the row would be written tagged with the wrong
/// source's collection, block the real claiming source with
/// `UniqueViolation` for the whole TTL, never promote, then vanish — with
/// nothing reporting it at any point.
#[test]
fn reserve_and_release_refuse_a_source_that_never_declared_the_namespace() {
    let (_dir, db, usernames, _users) = fixture();
    let sessions = open_sessions(&db);

    let err = take_err(db.tx().reserve(
        &usernames,
        UsernameKey::new("rail"),
        &sessions,
        &sid(1),
        Duration::from_secs(900),
    ));
    assert!(
        matches!(
            err,
            armdb::DbError::UndeclaredUniqueTarget {
                namespace: "unique_reservation_usernames"
            }
        ),
        "{err}"
    );

    let err = take_err(
        db.tx()
            .release(&usernames, UsernameKey::new("rail"), &sessions, &sid(1)),
    );
    assert!(
        matches!(
            err,
            armdb::DbError::UndeclaredUniqueTarget {
                namespace: "unique_reservation_usernames"
            }
        ),
        "{err}"
    );

    assert!(!usernames.contains(&UsernameKey::new("rail")));
}

/// Review round M5(a): a second `reserve` on a key the same owner already
/// holds refreshes the deadline through the real `db.tx()` stack (not just
/// `decide`'s own unit test) — and the refreshed deadline reflects the
/// *second* call's ttl, not merely elapsed real time since the first.
#[test]
fn reserve_refreshes_the_deadline_through_the_real_stack() {
    let (_dir, db, usernames, users) = fixture();

    let now1 = armdb::testing::now_ms();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(60),
        )
        .unwrap()
        .commit()
        .unwrap();
    let deadline1 = usernames
        .owner_of(&UsernameKey::new("rail"))
        .unwrap()
        .expires_at()
        .unwrap();
    assert_deadline_near(deadline1, now1, 60_000);

    let now2 = armdb::testing::now_ms();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();
    let deadline2 = usernames
        .owner_of(&UsernameKey::new("rail"))
        .unwrap()
        .expires_at()
        .unwrap();
    assert_deadline_near(deadline2, now2, 900_000);
    assert!(
        deadline2 > deadline1,
        "the second reserve must move the deadline forward"
    );
}

/// Review round M5(b): release one owner's reservation and reserve the same
/// key for a different owner, in one transaction — releases resolve before
/// acquisitions (T10 Task 7), so this hand-off must succeed.
#[test]
fn release_then_reserve_the_same_key_in_one_transaction() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(1),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    db.tx()
        .release(&usernames, UsernameKey::new("rail"), &users, &id(1))
        .unwrap()
        .reserve(
            &usernames,
            UsernameKey::new("rail"),
            &users,
            &id(2),
            Duration::from_secs(900),
        )
        .unwrap()
        .commit()
        .unwrap();

    let owner = usernames.owner_of(&UsernameKey::new("rail")).unwrap();
    assert!(owner.is_reservation());
    assert_eq!(owner.belongs_to::<User>(), Some(id(2)));
}

/// Review round M5(c): `release` is spec-sanctioned even for a *permanent*
/// claim (`Tx::release`'s own doc: "permanent claim or reservation alike").
/// It only ever touches the namespace row — the source row it names is
/// never written, so a released permanent claim leaves `users` holding a
/// value the namespace no longer protects.
#[test]
fn release_of_a_permanent_claim_deletes_only_the_namespace_row() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();
    assert!(
        !usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .is_reservation(),
        "fixture must be a permanent claim, or this test cannot tell release-of-permanent \
         from release-of-reservation"
    );

    db.tx()
        .release(&usernames, UsernameKey::new("rail"), &users, &id(1))
        .unwrap()
        .commit()
        .unwrap();

    assert!(!usernames.contains(&UsernameKey::new("rail")));
    assert_eq!(
        users.get(&id(1)).and_then(|u| u.username.clone()),
        Some("rail".into()),
        "release only touches the namespace row — the source row still carries the value"
    );
}