alkblobs 0.2.1

//! The pin clause family (the fleet host's pins): `pin_commit` (the
//! joint tx's terminal step, publish + protection in one commit),
//! `pin_renew` (the successor-time shape), `pin_check`, the maintenance
//! step (`pin_reap`), and delete-time SQL arbitration folding the
//! `FOR UPDATE` finger-lock — plus the engine-side liveness-table
//! registration the arbitration consults. The sweep lease's
//! acquire/release ride here too, under the
//! [`LeaseUnlockGuard`]'s cancellation defense (review-001 F-2: the
//! cancelled acquire/release windows cannot strand the advisory lock
//! on a pooled session).

use std::time::SystemTime;

use deadpool_postgres::Object;
use parking_lot::Mutex;

use super::joint::sweep_lock_key;
use super::{PgJointTx, PostgresBackend};
use crate::gc::{EngineFleetState, PinRow};
use crate::pg_common::{detached_batch_execute, err, pg_err, quote_identifier, time_to_nanos};
use crate::{Error, PinState};

impl PostgresBackend {
    /// The engine-side registration of named liveness-table references
    /// (ADR-012 §6.6's fleet registration's home): the store core's
    /// fleet registration verb forwards here. The reference is a
    /// (schema-qualified) table name, quoted at registration and held
    /// as a reference only — the engine never creates the table (the
    /// schema-ownership rule) and never learns liveness semantics; the
    /// [`EngineFleetState::arbitrate`] clause is the only consumer.
    /// Re-registering the same reference is idempotent.
    pub(crate) fn register_liveness_source(&self, reference: &str) -> Result<(), Error> {
        let quoted = quote_table_reference(reference)?;
        let mut sources = self.liveness_sources.lock();
        if !sources.contains(&quoted) {
            sources.push(quoted);
        }
        Ok(())
    }

    /// The engine-side removal (the correction flow's other half,
    /// ADR-012 §6.4): quote the reference the same way
    /// [`Self::register_liveness_source`] did and drop it from the
    /// consulted list. `false` when the reference (as quoted) is not
    /// currently registered — a consumed or foreign reference is not
    /// an error, deregistration stays a bool discipline.
    pub(crate) fn unregister_liveness_source(&self, reference: &str) -> Result<bool, Error> {
        let quoted = quote_table_reference(reference)?;
        let mut sources = self.liveness_sources.lock();
        let before = sources.len();
        sources.retain(|held| *held != quoted);
        Ok(sources.len() != before)
    }
}

/// Quote a (possibly schema-qualified) liveness-table reference at
/// registration: every dot-component rides the quoted-identifier
/// boundary; an empty component is a malformed reference, surfaced as
/// the engine error rather than interpolated unquoted.
fn quote_table_reference(reference: &str) -> Result<String, Error> {
    let mut quoted = String::new();
    for part in reference.split('.') {
        if part.is_empty() {
            return Err(Error::Io(format!(
                "postgres: invalid liveness table reference {reference:?}"
            )));
        }
        if !quoted.is_empty() {
            quoted.push('.');
        }
        quoted.push_str(&quote_identifier(part));
    }
    Ok(quoted)
}

/// The fleet family (ADR-012 §5's postgres row — all yes): the joint
/// entry+pin transaction, the pin clauses, delete-time SQL arbitration,
/// and the sweep lease. Fleet activation is the constructor
/// declaration (`fleet: true` — the `_fleet` constructors' explicit
/// declaration, ADR-012 §3) — under `fleet: false` the store core
/// never calls these; the clauses exist and are contract-tested per
/// their applicability, and the store core's put/sweep drivers
/// consume them as the fleet mechanism's tasks land.
impl EngineFleetState for PostgresBackend {
    type JointTx = PgJointTx;

    /// Open the joint tx on a pooled connection: checkout held for the
    /// tx's life, session-level `BEGIN` issued, uncommitted until
    /// [`Self::pin_commit`].
    async fn begin_joint_tx(&self) -> Result<PgJointTx, Error> {
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        Ok(PgJointTx {
            client: Some(client),
            blobs_table: self.table.clone(),
            pins_table: self.pins_table.clone(),
            #[cfg(test)]
            fail_commit: self.fail_joint_commit,
        })
    }

    /// The entry half, executed inside the held tx: the engine's put
    /// mechanics (the `ON CONFLICT DO NOTHING` CAS clause), same
    /// first-writer-wins semantics as the public `put`.
    async fn joint_put(&self, tx: &mut PgJointTx, key: &[u8], value: &[u8]) -> Result<(), Error> {
        let sql = format!(
            "INSERT INTO {} (key, value, size) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
            tx.blobs_table
        );
        let client = tx.client()?;
        let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?;
        let size = value.len() as i64;
        client
            .execute(&stmt, &[&key, &value, &size])
            .await
            .map_err(pg_err)?;
        Ok(())
    }

    /// The joint terminal step: the pin-row inserts **and** the commit
    /// — publish and protection land in one step. Every failure path
    /// rolls the entire tx back (no pin rows, no published entry, zero
    /// residue) before surfacing the error; consumes the tx.
    async fn pin_commit(&self, mut tx: PgJointTx, entries: &[PinRow]) -> Result<(), Error> {
        let Some(client) = tx.client.take() else {
            return Err(Error::Io("postgres: joint tx already consumed".to_owned()));
        };
        #[cfg(test)]
        if tx.fail_commit {
            client.batch_execute("ROLLBACK").await.map_err(pg_err)?;
            return Err(Error::Io(
                "engine joint commit failure (postgres, injected)".to_owned(),
            ));
        }
        let sql = format!(
            "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3) \
             ON CONFLICT (key, owner) DO UPDATE SET expiry = excluded.expiry",
            tx.pins_table
        );
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(err(e));
            }
        };
        for entry in entries {
            let expiry = time_to_nanos(entry.expiry);
            match client
                .execute(&stmt, &[&entry.key, &entry.owner, &expiry])
                .await
            {
                Ok(_) => {}
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    return Err(err(e));
                }
            }
        }
        match client.batch_execute("COMMIT").await {
            Ok(_) => Ok(()),
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(err(e))
            }
        }
    }

    /// Renew the `(key, owner)` pin row: `expiry = now + pin_ttl` (the
    /// recorded successor-time source — the TTL the backend was
    /// constructed with: the shipped default, or the fleet
    /// constructor's value through `new_with_pin_ttl`) — and only
    /// against a live row: the update's predicate requires the
    /// existing row un-expired, so an expired or absent row is never
    /// resurrected (`false`, nothing updated).
    async fn pin_renew(&self, key: &[u8], owner: &[u8]) -> Result<bool, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let now = SystemTime::now();
        let successor = now
            .checked_add(self.pin_ttl)
            .ok_or_else(|| Error::Io("postgres: pin expiry overflow".to_owned()))?;
        let successor_nanos = time_to_nanos(successor);
        let now_nanos = time_to_nanos(now);
        let sql = format!(
            "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3 AND expiry > $4",
            self.pins_table
        );
        let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?;
        let updated = client
            .execute(&stmt, &[&successor_nanos, &key, &owner, &now_nanos])
            .await
            .map_err(pg_err)?;
        Ok(updated == 1)
    }

    /// Classify against un-expired pin rows: any un-expired `(key, *)`
    /// row exists → `Protected`, else `Unprotected`. The engine never
    /// decides deletion — it only classifies.
    async fn pin_check(&self, key: &[u8]) -> Result<PinState, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let now_nanos = time_to_nanos(SystemTime::now());
        let sql = format!(
            "SELECT EXISTS(SELECT 1 FROM {} WHERE key = $1 AND expiry > $2)",
            self.pins_table
        );
        let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?;
        let present = client
            .query_one(&stmt, &[&key, &now_nanos])
            .await
            .map_err(pg_err)?
            .try_get::<_, bool>(0)
            .map_err(pg_err)?;
        Ok(if present {
            PinState::Protected
        } else {
            PinState::Unprotected
        })
    }

    /// The maintenance step (ADR-008 §2's GC-of-GC cleanup): one
    /// statement — `DELETE … WHERE expiry <= $1` — reclaiming exactly
    /// the expired pin rows (the boundary matches `pin_renew`'s live
    /// predicate: `expiry > now` is live, so `expiry <= now` is the
    /// abandoned set). Un-expired rows are never touched; returns the
    /// count reaped. The fleet sweep drives it under the lease.
    async fn pin_reap(&self, now: SystemTime) -> Result<u64, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let now_nanos = time_to_nanos(now);
        let sql = format!("DELETE FROM {} WHERE expiry <= $1", self.pins_table);
        let stmt = client.prepare_cached(&sql).await.map_err(pg_err)?;
        client.execute(&stmt, &[&now_nanos]).await.map_err(pg_err)
    }

    /// The full delete-time arbitration, folded under the `FOR UPDATE`
    /// finger-lock in one `BEGIN … COMMIT` step: un-expired pin rows
    /// for the key first, then each registered liveness-table
    /// reference, all keyed at delete time — `true` = liveness protects
    /// the key at this instant. The registration list is snapshotted
    /// under its (process-local) mutex before the server round trip.
    async fn arbitrate(&self, key: &[u8]) -> Result<bool, Error> {
        let sources = self.liveness_sources.lock().clone();
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        let folded: Result<bool, Error> = async {
            let now_nanos = time_to_nanos(SystemTime::now());
            let pins_sql = format!(
                "SELECT 1 FROM {} WHERE key = $1 AND expiry > $2 FOR UPDATE LIMIT 1",
                self.pins_table
            );
            let stmt = client.prepare_cached(&pins_sql).await.map_err(pg_err)?;
            let mut protected = client
                .query_opt(&stmt, &[&key, &now_nanos])
                .await
                .map_err(pg_err)?
                .is_some();
            if !protected {
                for table in &sources {
                    let sql = format!("SELECT 1 FROM {table} WHERE key = $1 FOR UPDATE LIMIT 1");
                    let liveness_stmt = client.prepare_cached(&sql).await.map_err(pg_err)?;
                    let live = client
                        .query_opt(&liveness_stmt, &[&key])
                        .await
                        .map_err(pg_err)?
                        .is_some();
                    if live {
                        protected = true;
                        break;
                    }
                }
            }
            client.batch_execute("COMMIT").await.map_err(pg_err)?;
            Ok(protected)
        }
        .await;
        if folded.is_err() {
            let _ = client.batch_execute("ROLLBACK").await;
        }
        folded
    }

    /// Acquire the sweep lease: the session-scoped pg advisory lock,
    /// atomic test-and-set. `Ok(true)` = held (the lease's checkout is
    /// stored so release rides the same session); `Ok(false)` =
    /// contention — either a second call on this backend (the one-
    /// sweeper-per-instance mutex answers before any pool checkout) or
    /// another session on the same schema holding the lock. Session
    /// loss reaps a held lease server-side and **in-instance**: a lease
    /// whose held checkout is closed is dropped here (ADR-013 §2's
    /// holdership rule — the acquisition answers against a live
    /// session only, so a fresh acquire succeeds without a prior
    /// release); owner death reaps it at checkout drop.
    ///
    /// **The cancellation window (review-001 F-2)**: the whole span
    /// from the pool checkout through the `pg_try_advisory_lock` round
    /// trip to the lease-store line is guarded by the returned-to-
    /// caller [`LeaseUnlockGuard`] — a cancellation dropping the future
    /// anywhere in that span (the lock may have been granted server-
    /// side while the answer was still in flight) leaves the guard
    /// dropping with the checkout in hand, and its spawned best-effort
    /// unlock clears the advisory lock before the checkout returns to
    /// the pool. The store line is the one await-free seam between the
    /// lock's grant and the lease's record; the guard is defused by
    /// `into_hold` there.
    async fn sweep_lock_acquire(&self) -> Result<bool, Error> {
        {
            let mut lease = self.lease.lock();
            if lease.as_ref().is_some_and(|client| !client.is_closed()) {
                return Ok(false);
            }
            *lease = None;
        }
        let client = self.pool.get().await.map_err(err)?;
        let guard = LeaseUnlockGuard::new(client, &self.schema);
        let key = sweep_lock_key(&self.schema);
        let session = guard
            .session()
            .ok_or_else(|| Error::Io("postgres: lease checkout already consumed".to_owned()))?;
        let stmt = session
            .prepare_cached("SELECT pg_try_advisory_lock($1)")
            .await
            .map_err(pg_err)?;
        let held = session
            .query_one(&stmt, &[&key])
            .await
            .map_err(pg_err)?
            .try_get::<_, bool>(0)
            .map_err(pg_err)?;
        if held {
            #[cfg(test)]
            crate::postgres::testing::lease_cancel_window_park(&self.schema).await;
            guard.into_hold(&self.lease);
        } else {
            guard.release_unheld();
        }
        Ok(held)
    }

    /// Release the sweep lease on the same session it was acquired on
    /// (the mutex-held checkout). Releasing an unheld lease is a
    /// no-op; a holder whose session died is the reaping case itself —
    /// the call never errors, holdership has ended either way (an
    /// unlock failure is that reaping, not a fault to surface).
    ///
    /// **The cancellation window (review-001 F-2)**: after the `take()`
    /// the client is guarded by a fresh [`LeaseUnlockGuard`] through
    /// the unlock's round trip — a cancellation before the unlock
    /// completes leaves the guard dropping owed, its spawned
    /// best-effort unlock clearing the lock before the checkout
    /// returns to the pool.
    async fn sweep_lock_release(&self) -> Result<(), Error> {
        let held = self.lease.lock().take();
        let Some(client) = held else {
            return Ok(());
        };
        let guard = LeaseUnlockGuard::new(client, &self.schema);
        #[cfg(test)]
        crate::postgres::testing::lease_cancel_window_park(&self.schema).await;
        guard.unlock(sweep_lock_key(&self.schema)).await;
        Ok(())
    }
}

/// A pooled checkout whose `pg_advisory_unlock` is owed if dropped —
/// the sweep lease's cancellation guard (review-001 F-2, N-6 riding
/// through the release path): it owns the lease's checkout across the
/// windows where the advisory lock is (or may be) held server-side but
/// the backend's lease record does not yet (or no longer) answer for
/// it. Every completed path leaves the guard holding nothing — the
/// lease's success path moves the checkout into the backend's
/// `lease` record (`into_hold`), the release path's unlock (`unlock`)
/// consumes it — so the guard dropping with the checkout in hand means
/// exactly "cancelled between the lock's grant and the lease's
/// record", or "cancelled mid-unlock": the Drop body then hands the
/// checkout to a spawned best-effort unlock
/// ([`crate::pg_common::detached_batch_execute`]) so the advisory
/// lock never rides an idle session into the pool. The unlock's
/// result is ignored by contract (an unlock failure is the reaping
/// case, ADR-013 §2).
struct LeaseUnlockGuard {
    /// The held checkout — `Some` while the unlock is owed on drop.
    client: Option<Object>,
    /// The backend's schema — the advisory lock key's namespaced
    /// derivation input ([`sweep_lock_key`]).
    schema: String,
}

impl LeaseUnlockGuard {
    fn new(client: Object, schema: &str) -> Self {
        Self {
            client: Some(client),
            schema: schema.to_owned(),
        }
    }

    /// The held checkout, for the caller's statement surface.
    fn session(&self) -> Option<&Object> {
        self.client.as_ref()
    }

    /// The acquire path's success arm: the lock is held and the
    /// backend's lease record now answers for it — move the checkout
    /// into the record so the Drop body has nothing owed.
    fn into_hold(self, lease: &Mutex<Option<Object>>) {
        let mut this = self;
        *lease.lock() = this.client.take();
    }

    /// The acquire path's contention arm: the try answered `false` —
    /// no advisory lock was granted on this session, nothing owed on
    /// drop (the checkout drops clean, straight back to the pool).
    fn release_unheld(mut self) {
        let _ = self.client.take();
    }

    /// The release path's body: issue the unlock best-effort, leaving
    /// the guard holding nothing (the checkout drops with the guard,
    /// clean, back to the pool). Never errors — by the call's end
    /// holdership has ended either way.
    async fn unlock(self, key: i64) {
        let mut this = self;
        if let Some(client) = this.client.take() {
            let _ = client
                .batch_execute(&format!("SELECT pg_advisory_unlock({key})"))
                .await;
        }
    }
}

impl Drop for LeaseUnlockGuard {
    fn drop(&mut self) {
        if let Some(client) = self.client.take() {
            let key = sweep_lock_key(&self.schema);
            detached_batch_execute(client, format!("SELECT pg_advisory_unlock({key})"));
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::time::{Duration, SystemTime};

    use crate::pg_common::time_to_nanos;
    use crate::postgres::pool::{DEFAULT_PIN_TTL, DEFAULT_PIN_TTL_SECS};

    use crate::postgres::testing::{
        arc_backend, at, harness_dsn, harness_gc, harness_with_ttl, pin_row, pin_rows,
        raw_connection, schema_tables, CONTENT,
    };

    /// `pin_renew`'s expiry semantics (ADR-012 §2 verbatim): a live row
    /// renews to the successor time (`now + DEFAULT_PIN_TTL`, the
    /// recorded source); an expired or absent row is never resurrected
    /// — `false`, nothing updated.
    #[tokio::test]
    async fn pin_renew_never_resurrects_an_expired_or_absent_row() {
        let Some(backend) = harness_gc("renew").await else {
            return;
        };
        let key = b"\x01renew-key".to_vec();
        let owner = b"\x01instance-id".to_vec();

        assert!(
            !backend.pin_renew(&key, &owner).await.unwrap(),
            "absent row: no renewal, no creation"
        );
        assert_eq!(pin_rows(&backend).await, vec![], "absent stays absent");

        let live = SystemTime::now() + Duration::from_secs(30);
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)",
                backend.pins_table
            );
            client
                .execute(sql.as_str(), &[&key, &owner, &time_to_nanos(live)])
                .await
                .unwrap();
        }
        assert!(
            backend.pin_renew(&key, &owner).await.unwrap(),
            "live row renews"
        );
        let after = pin_rows(&backend).await;
        let expected =
            time_to_nanos(SystemTime::now()) + DEFAULT_PIN_TTL_SECS as i64 * 1_000_000_000;
        let got = after[0].2;
        assert!(
            (got - expected).abs() < 3_000_000_000,
            "renewed to the successor time now + {DEFAULT_PIN_TTL_SECS}s: {got} vs {expected}"
        );

        let expired_nanos = time_to_nanos(at(-1_000_000_000));
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3",
                backend.pins_table
            );
            client
                .execute(sql.as_str(), &[&expired_nanos, &key, &owner])
                .await
                .unwrap();
        }
        assert!(
            !backend.pin_renew(&key, &owner).await.unwrap(),
            "expired row: never resurrected"
        );
        assert_eq!(
            pin_rows(&backend).await,
            vec![(key, owner, expired_nanos)],
            "the expired row's expiry is untouched"
        );
    }

    /// The maintenance step (`pin_reap`, ADR-008 §2's GC-of-GC
    /// cleanup): exactly the expired set is reclaimed — rows whose
    /// expiry is past `now` delete (count returned exact), un-expired
    /// rows survive untouched, and per-row composition holds (a key
    /// with one expired and one live row keeps the live row —
    /// protection composes by row, the reaper reclaims by row). A
    /// second reap of the same instant reclaims nothing.
    #[tokio::test]
    async fn pin_reap_reaps_exactly_the_expired_set() {
        let Some(backend) = harness_gc("reap").await else {
            return;
        };
        let now = SystemTime::now();
        let expired_solo = pin_row(
            b"\x01reap-expired-solo",
            b"\x01owner-a",
            now - Duration::from_secs(1),
        );
        let expired_pair = pin_row(
            b"\x01reap-pair-key",
            b"\x01owner-b",
            now - Duration::from_secs(300),
        );
        let live_pair = pin_row(
            b"\x01reap-pair-key",
            b"\x01owner-c",
            now + Duration::from_secs(30),
        );
        let live_solo = pin_row(
            b"\x01reap-live-solo",
            b"\x01owner-d",
            now + Duration::from_secs(400),
        );

        {
            let mut tx = backend.begin_joint_tx().await.unwrap();
            EngineFleetState::joint_put(&*backend, &mut tx, &expired_solo.key, CONTENT)
                .await
                .unwrap();
            backend
                .pin_commit(
                    tx,
                    &[
                        expired_solo.clone(),
                        expired_pair.clone(),
                        live_pair.clone(),
                    ],
                )
                .await
                .unwrap();
        }
        {
            let tx = backend.begin_joint_tx().await.unwrap();
            backend
                .pin_commit(tx, std::slice::from_ref(&live_solo))
                .await
                .unwrap();
        }

        let reaped = EngineFleetState::pin_reap(&*backend, now).await.unwrap();
        assert_eq!(
            reaped, 2,
            "exactly the expired rows reclaimed: the solo and the pair-key's expired owner"
        );

        let after = pin_rows(&backend).await;
        let mut expected: Vec<(Vec<u8>, Vec<u8>, i64)> = vec![
            (
                live_pair.key.clone(),
                live_pair.owner.clone(),
                time_to_nanos(live_pair.expiry),
            ),
            (
                live_solo.key.clone(),
                live_solo.owner.clone(),
                time_to_nanos(live_solo.expiry),
            ),
        ];
        expected.sort();
        assert_eq!(
            after, expected,
            "un-expired rows survive untouched — including the live row over the reaped key"
        );

        let second = EngineFleetState::pin_reap(&*backend, now).await.unwrap();
        assert_eq!(
            second, 0,
            "a second reap of the same instant reclaims nothing"
        );
        assert_eq!(pin_rows(&backend).await, expected, "no further change");
    }

    /// The constructor-carried pin TTL reaches the renewal body: a
    /// backend opened through `new_with_pin_ttl` (the fleet
    /// constructors' instrument) renews a live row to
    /// `now + pin_ttl` — the constructor's value, not the shipped
    /// default (the wave-4 recorded design point, landed with the
    /// fleet declaration).
    #[tokio::test]
    async fn pin_renew_renews_to_the_constructor_carried_ttl() {
        let Some(backend) = harness_with_ttl("renew-ttl", Duration::from_secs(3600)).await else {
            return;
        };
        let key = b"\x01renew-ttl-key".to_vec();
        let owner = b"\x0ainstance-id".to_vec();

        let live = SystemTime::now() + Duration::from_secs(30);
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)",
                backend.pins_table
            );
            client
                .execute(sql.as_str(), &[&key, &owner, &time_to_nanos(live)])
                .await
                .unwrap();
        }
        assert!(
            backend.pin_renew(&key, &owner).await.unwrap(),
            "live row renews"
        );
        let after = pin_rows(&backend).await;
        let expected = time_to_nanos(SystemTime::now()) + 3600i64 * 1_000_000_000;
        let got = after[0].2;
        assert!(
            (got - expected).abs() < 3_000_000_000,
            "renewed to the constructor-carried successor time now + 3600s: {got} vs {expected}"
        );
    }

    /// `pin_check`'s two-variant classification against un-expired pin
    /// rows: absent or only-expired rows → `Unprotected`; any live row
    /// → `Protected`. Protection composes by row, so any owner's live
    /// row counts.
    #[tokio::test]
    async fn pin_check_classifies_the_two_variants() {
        let Some(backend) = harness_gc("check").await else {
            return;
        };
        let key = b"\x01check-key".to_vec();
        let owner = b"\x01instance-id".to_vec();
        let other_owner = b"\x02instance-id".to_vec();

        assert_eq!(
            backend.pin_check(&key).await.unwrap(),
            PinState::Unprotected,
            "no rows: Unprotected"
        );
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)",
                backend.pins_table
            );
            client
                .execute(
                    sql.as_str(),
                    &[&key, &owner, &time_to_nanos(at(-5_000_000_000))],
                )
                .await
                .unwrap();
        }
        assert_eq!(
            backend.pin_check(&key).await.unwrap(),
            PinState::Unprotected,
            "only an expired row: Unprotected (abandoned pins never protect)"
        );
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)",
                backend.pins_table
            );
            client
                .execute(
                    sql.as_str(),
                    &[
                        &key,
                        &other_owner,
                        &time_to_nanos(SystemTime::now() + DEFAULT_PIN_TTL),
                    ],
                )
                .await
                .unwrap();
        }
        assert_eq!(
            backend.pin_check(&key).await.unwrap(),
            PinState::Protected,
            "any owner's live row protects (protection composes by row)"
        );
    }

    /// `arbitrate` folds the full answer under the `FOR UPDATE`
    /// finger-lock: un-expired pin rows plus registered liveness-table
    /// references — here a test-owned embedder table the engine never
    /// created. `true` = liveness protects the key at this instant;
    /// expired pins and absent rows protect nothing.
    #[tokio::test]
    async fn arbitrate_folds_pins_and_registered_liveness_tables() {
        let Some(backend) = harness_gc("arbitrate").await else {
            return;
        };
        let key = b"\x01arbitrate-key".to_vec();
        let owner = b"\x01instance-id".to_vec();

        let embedder_table = format!("{}.alive_keys", backend.schema);
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "CREATE TABLE {} (key bytea PRIMARY KEY, seq bigint)",
                quote_table_reference(&embedder_table).unwrap()
            );
            client.execute(sql.as_str(), &[]).await.unwrap();
        }
        backend.register_liveness_source(&embedder_table).unwrap();
        backend.register_liveness_source(&embedder_table).unwrap();
        let tables = schema_tables(&backend).await;
        assert_eq!(
            tables,
            vec!["alive_keys", "blobs", "gc_pins", "gc_window"],
            "the embedder's table exists only because the test (the embedder) created it"
        );

        assert!(
            !backend.arbitrate(&key).await.unwrap(),
            "nothing protects an absent key"
        );

        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, seq) VALUES ($1, $2)",
                quote_table_reference(&embedder_table).unwrap()
            );
            client.execute(sql.as_str(), &[&key, &1_i64]).await.unwrap();
        }
        assert!(
            backend.arbitrate(&key).await.unwrap(),
            "a registered liveness row protects the key"
        );

        {
            let client = backend.pool.get().await.unwrap();
            client
                .execute(
                    &format!(
                        "DELETE FROM {} WHERE key = $1",
                        quote_table_reference(&embedder_table).unwrap()
                    ),
                    &[&key],
                )
                .await
                .unwrap();
        }
        assert!(
            !backend.arbitrate(&key).await.unwrap(),
            "liveness row gone, no pin rows: unprotected"
        );

        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)",
                backend.pins_table
            );
            client
                .execute(
                    sql.as_str(),
                    &[&key, &owner, &time_to_nanos(at(-1_000_000_000))],
                )
                .await
                .unwrap();
        }
        assert!(
            !backend.arbitrate(&key).await.unwrap(),
            "an expired pin row is abandoned: it protects nothing"
        );

        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3",
                backend.pins_table
            );
            client
                .execute(
                    sql.as_str(),
                    &[
                        &time_to_nanos(SystemTime::now() + DEFAULT_PIN_TTL),
                        &key,
                        &owner,
                    ],
                )
                .await
                .unwrap();
        }
        assert!(
            backend.arbitrate(&key).await.unwrap(),
            "a live pin row protects the key"
        );
    }

    /// The `FOR UPDATE` finger-lock is real: a raw session holding the
    /// pin row's row lock at delete time blocks a concurrent
    /// `pin_renew` (which targets the same row) until the holder
    /// commits — the arbitration's exclusion of the renewal path.
    #[tokio::test]
    async fn arbitrate_finger_lock_blocks_a_concurrent_renewal() {
        let Some(backend) = harness_gc("finger-lock").await else {
            return;
        };
        let key = b"\x01finger-lock-key".to_vec();
        let owner = b"\x01instance-id".to_vec();
        {
            let client = backend.pool.get().await.unwrap();
            let sql = format!(
                "INSERT INTO {} (key, owner, expiry) VALUES ($1, $2, $3)",
                backend.pins_table
            );
            client
                .execute(
                    sql.as_str(),
                    &[
                        &key,
                        &owner,
                        &time_to_nanos(SystemTime::now() + DEFAULT_PIN_TTL),
                    ],
                )
                .await
                .unwrap();
        }

        let client = backend.pool.get().await.unwrap();
        client.batch_execute("BEGIN").await.unwrap();
        let lock_sql = format!(
            "SELECT expiry FROM {} WHERE key = $1 AND owner = $2 FOR UPDATE",
            backend.pins_table
        );
        let stmt = client.prepare(&lock_sql).await.unwrap();
        let locked = client.query(&stmt, &[&key, &owner]).await.unwrap().len();
        assert_eq!(locked, 1, "the raw session holds the finger-lock");

        let (renewed_tx, mut renewed_rx) = tokio::sync::oneshot::channel();
        tokio::spawn(async move {
            let renewed = backend.pin_renew(&key, &owner).await;
            let _ = renewed_tx.send(renewed);
        });
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
        assert!(
            matches!(
                renewed_rx.try_recv(),
                Err(tokio::sync::oneshot::error::TryRecvError::Empty)
            ),
            "the renewal is still blocked on the finger-lock"
        );

        client.batch_execute("COMMIT").await.unwrap();
        assert!(
            renewed_rx.await.unwrap().unwrap(),
            "renew lands once the holder releases"
        );
    }

    /// The acquire window's cancellation (review-001 F-2): aborting the
    /// acquire between the advisory lock's grant and the lease record's
    /// store (the test-instrumented parked point) leaves the guard
    /// dropping owed — its spawned best-effort unlock frees the lock
    /// before the checkout returns to the pool, and the backend can
    /// acquire again.
    #[tokio::test]
    async fn cancelled_acquire_cannot_strand_the_advisory_lock() {
        let Some(backend) = harness_gc("lease-cancel-acquire").await else {
            return;
        };
        let acquirer = arc_backend(&backend);
        let dsn = harness_dsn().unwrap();
        let (raw, _conn) = raw_connection(&dsn).await;
        let key = sweep_lock_key(&backend.schema);

        crate::postgres::testing::arm_lease_cancel_window(&backend.schema);
        let task = tokio::spawn({
            let acquirer = Arc::clone(&acquirer);
            async move { acquirer.sweep_lock_acquire().await }
        });
        crate::postgres::testing::wait_for_advisory(
            &raw,
            key,
            true,
            "the acquire granted server-side, parked in the window",
        )
        .await;
        task.abort();
        let _ = task.await;
        crate::postgres::testing::disarm_lease_cancel_window();

        crate::postgres::testing::wait_for_advisory(
            &raw,
            key,
            false,
            "the guard's spawned unlock cleared the granted lock",
        )
        .await;
        assert!(
            acquirer.lease.lock().is_none(),
            "the cancelled acquire stored no lease record"
        );
        assert!(
            backend.sweep_lock_acquire().await.unwrap(),
            "the backend acquires again — nothing stranded"
        );
        backend.sweep_lock_release().await.unwrap();
    }

    /// The release window's cancellation (review-001 F-2): aborting the
    /// release after its `take()` (the lease record emptied, the unlock
    /// not yet issued — the test-instrumented parked point) leaves the
    /// guard dropping owed; its spawned unlock frees the advisory lock.
    #[tokio::test]
    async fn cancelled_release_cannot_strand_the_advisory_lock() {
        let Some(backend) = harness_gc("lease-cancel-release").await else {
            return;
        };
        let releaser = arc_backend(&backend);
        assert!(releaser.sweep_lock_acquire().await.unwrap(), "held");
        let dsn = harness_dsn().unwrap();
        let (raw, _conn) = raw_connection(&dsn).await;
        let key = sweep_lock_key(&backend.schema);

        crate::postgres::testing::arm_lease_cancel_window(&backend.schema);
        let task = tokio::spawn({
            let releaser = Arc::clone(&releaser);
            async move { releaser.sweep_lock_release().await }
        });
        let mut taken = false;
        for _ in 0..100 {
            if releaser.lease.lock().is_none() {
                taken = true;
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert!(taken, "the release took the checkout out of the record");
        crate::postgres::testing::wait_for_advisory(
            &raw,
            key,
            true,
            "the parked release has not unlocked yet",
        )
        .await;
        task.abort();
        let _ = task.await;
        crate::postgres::testing::disarm_lease_cancel_window();

        crate::postgres::testing::wait_for_advisory(
            &raw,
            key,
            false,
            "the guard's spawned unlock cleared the parked release's lock",
        )
        .await;
        assert!(
            backend.sweep_lock_acquire().await.unwrap(),
            "the backend acquires again — nothing stranded"
        );
        backend.sweep_lock_release().await.unwrap();
    }
}