alkgit-substrate 0.1.0

//! The refs co-tenant row table (substrate.md §refs, ADR-019 §2) — the
//! object-storage family's third trait ([`GitRefs`]) in three shapes:
//!
//! - [`InMemoryRefTable`] — the CAS-semantics reference impl, engine-free
//!   and unconditional in the crate's base (the completeness shape; no
//!   `mem` feature exists — ADR-025's placement rule).
//! - [`SqliteRefTable`] (the `sqlite` feature) — the row-CAS twin over
//!   the co-tenant `git_refs` rows
//!   ([`CoTenantSqlite`](crate::CoTenantSqlite)): every CAS check and
//!   apply inside one `BEGIN IMMEDIATE` transaction — the
//!   one-transaction home that makes `atomic` correct (ADRs 013 §7 / 019
//!   §2; POC-O verified stale/atomic/delete/force against real `git
//!   push` behaviors, poco-findings DQ-2).
//! - `PgRefTable` (the `postgres` feature) — the pg row-CAS twin
//!   (`crate::postgres::CoTenantPg`): the same one
//!   transaction per `apply_updates` call, the pg transaction class
//!   standing in for sqlite's `BEGIN IMMEDIATE` — the call's
//!   transaction holds the schema's advisory CAS lock (the pg analog
//!   of the file write lock), so two clashing applies — cross-connection
//!   included — serialize and the loser sees the committed winner.
//!
//! # The CAS semantics (the reference shapes both impls pin)
//!
//! Per update, decided in request order against the row state with
//! in-call visibility (each decision and write sees the updates already
//! applied within the call); the whole call commits only under the
//! non-atomic flag — atomic aborts whole on any denial. The wire-path
//! input always carries distinct ref names per call (duplicate commands
//! fail closed at the parse — ADR-013 §3), so in practice every update
//! addresses its own row and the semantics below evaluate cleanly in
//! the git-observed classes:
//!
//! - **Stale** — [`CAS_STALE_REASON`]: the `expected` CAS failed. An
//!   absent ref never matches an `expected: Some(..)`; a present row
//!   never matches `expected: None` (create-over-existing) or a
//!   mismatched oid; a symref row's oid is unborn — it matches nothing
//!   (the seeded HEAD can be neither created-over nor CAS-deleted —
//!   git parity: a push never touches a repo's HEAD row).
//! - **Create** — `expected: None` applies onto an absent name only.
//! - **Delete** — `new: None` applies onto an oid row whose oid matches
//!   `expected`. The degenerate `expected: None` + `new: None` shape is
//!   a no-op success on an absent name (nothing to delete), stale on a
//!   present one.
//! - **Force** — no fast-forward check exists (the v1 CAS-only policy,
//!   ADR-013 §7): an update whose `expected` matches applies with any
//!   new value, non-fast-forward included (the client's forced-update
//!   shape is client-displayed; the table's contract is exact-match
//!   CAS and nothing more).
//! - **Atomic** — one transaction whole (ADR-013 §10): any denial
//!   aborts — nothing applies; the denied refs keep their own reasons
//!   and every other ref of the call reports
//!   [`ATOMIC_PUSH_FAILURE_REASON`]. An all-passing call commits whole.
//! - **Non-atomic** — per-update outcomes, each applying when its own
//!   CAS passes; denials never block the others.
//!
//! Per-ref denials are [`RefOutcome`] values, never `Err` (a stale ref
//! is a business result of receive-pack — ADR-013 §7–8). `Err` is for
//! substrate failure only: an engine failure rolls the call's
//! transaction back whole and rides [`StorageError::Io`]
//! (message-only — no engine source types on the surface, the family
//! posture).
//!
//! # Name validation at the table layer (the pinned mapping)
//!
//! Every update's ref name runs [`alkgit::validate_ref_name`] — the
//! core's `gix_validate::reference` composition plus the
//! `RESERVED_REF_NAMESPACES` deny-list and `.lock` rule, reused (no
//! second deny-list; ADR-013 §9's validation homes at the CAS layer —
//! substrate.md §refs). The mapping this task pins: a rejected name is
//! a per-ref denial ([`RefOutcome`] with `FUNNY_REFNAME_REASON`),
//! never `Err`, whatever the row state; under `atomic` the
//! invalid-name refs keep `funny refname` while the other refs report
//! `atomic push failure` (the same upstream shape the wire layer's
//! abort branch pins, ADR-013 §10). The wire layer already validated;
//! this is the CAS-home defense. `list_refs` does not re-validate:
//! rows can only arrive through the validated CAS path or the seeding
//! rule (substrate.md §refs), so read-back names are trusted — but a
//! structurally invalid row state is [`StorageError::Invalid`]
//! (mis-state, not a business denial).
//!
//! # The row shapes
//!
//! `(repo_id, name) → oid, symref_target` rows (POC-O-validated table,
//! DQ-2). An oid row lists as the plain [`RefLine`]; a symref row (oid
//! `NULL`, target set — the seeded `HEAD → refs/heads/main`) lists as
//! the **unborn symref line** `RefLine { oid: None, symref_target:
//! Some(..) }` — the unborn-HEAD shape `ls-refs=unborn` serves (the
//! OQ-19 rider (a) fixture falls out of the seeding rule; the seeding
//! itself lands in `substrate/w6-registry-sqlite`'s `put_repo`, which
//! writes the HEAD symref row in the record-write transaction — this
//! task's `list_refs` surfaces the shape on both twins). Listing is
//! deterministic: sorted by ref name (HEAD sorts before the `refs/`
//! subtree). `peeled` is never produced by a row table — peel needs
//! object access the refs surface does not have (ls-refs-arg data).
//!
//! Ref-table writes ride their own durable surface — no stream events,
//! no registry-seq coupling (ADR-019 §7: pushes anchor the event/
//! watermark legs with the composition-internal mutation marker, the
//! receive composition's business; ref-table writes are not
//! registry-family writes).

use std::collections::BTreeMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;

use alkgit::{
    validate_ref_name, GitRefs, RefLine, RefOutcome, RefUpdate, RepoRecord, StorageError,
    ATOMIC_PUSH_FAILURE_REASON, FUNNY_REFNAME_REASON,
};
use async_trait::async_trait;
use gix_hash::ObjectId;

use crate::liveness::{fingerprint_rows, RefRow, RefRowState, RefRows, RefRowsSnapshot};

/// The pinned per-ref CAS denial reason (ADR-013 §7–8: CAS-stale
/// failures carry a server-chosen reason — ours is upstream's
/// `stale info`-style internal text, the same text the core's contract
/// doubles pin).
pub const CAS_STALE_REASON: &str = "stale info";

/// One refs-table row's state as the CAS and `list_refs` read it back.
#[derive(Clone, Debug, PartialEq, Eq)]
enum RowState {
    /// The ordinary ref row: an oid.
    Oid(ObjectId),
    /// The symref row (the seeded `HEAD → refs/heads/main`): the unborn
    /// shape — no oid of its own, a set target.
    Symref(String),
}

impl RowState {
    /// A symref row has no oid of its own — the `None` is exactly the
    /// unborn property that makes the row never match a CAS
    /// expectation.
    fn oid(&self) -> Option<ObjectId> {
        match self {
            RowState::Oid(oid) => Some(*oid),
            RowState::Symref(_) => None,
        }
    }
}

/// The write a passing CAS decision commands (create and update are
/// the same row write — set the oid; delete removes the row).
enum CasWrite {
    Set { oid: ObjectId },
    Delete,
}

/// The two denial classes of the CAS check — [`RefOutcome`] reasons,
/// never errors. The reason text is the respective pinned constant.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum CasDenial {
    /// The name failed the table-layer validation.
    FunnyRefname,
    /// The `expected` CAS failed (the [`CAS_STALE_REASON`] class).
    Stale,
}

impl CasDenial {
    fn reason(self) -> &'static str {
        match self {
            CasDenial::FunnyRefname => FUNNY_REFNAME_REASON,
            CasDenial::Stale => CAS_STALE_REASON,
        }
    }
}

/// The shared CAS decision — the semantics core both twins run (the
/// module doc's CAS section is this function's pin). One call per
/// update, against the update's current row state: the name validation
/// first (whatever the row state), then the exact-match CAS.
///
/// - `expected` must match the row exactly (the CAS contract's
///   match matrix): absent ⇔ `None`, an oid row ⇔ its own oid only; a
///   symref row never matches (unborn ⇒ no oid to CAS against).
/// - A pass commands the write: `new: Some` sets the oid (create and
///   update are identical at the row level), `new: None` deletes.
/// - The degenerate `expected: None` + `new: None` shape on an absent
///   name passes with no write (`None` decision).
fn cas_decide(
    current: Option<RowState>,
    update: &RefUpdate,
) -> Result<Option<CasWrite>, CasDenial> {
    if validate_ref_name(&update.name).is_err() {
        return Err(CasDenial::FunnyRefname);
    }
    let current_oid = current.as_ref().and_then(RowState::oid);
    let matched = match (&current, update.expected) {
        (None, None) => true,
        (None, Some(_)) => false,
        (Some(_), None) => false,
        (Some(_), Some(_)) => current_oid.is_some() && update.expected == current_oid,
    };
    if !matched {
        return Err(CasDenial::Stale);
    }
    Ok(match update.new {
        Some(new) => Some(CasWrite::Set { oid: new }),
        None => Some(CasWrite::Delete),
    })
}

/// The row shape's read-back construction from the row's two columns
/// (the row twins' shared path; the in-memory impl holds the same
/// shape as an enum). Exactly-one discipline: an oid row is `(oid, no target)`,
/// a symref row is `(no oid, target)` — anything else is structurally
/// invalid row state ([`StorageError::Invalid`]).
#[cfg(any(feature = "sqlite", feature = "postgres"))]
fn row_state_from_columns(
    repo_id: &str,
    name: &str,
    oid_hex: Option<String>,
    symref_target: Option<String>,
) -> Result<RowState, StorageError> {
    match (oid_hex, symref_target) {
        (Some(hex), None) => {
            let oid = ObjectId::from_hex(hex.as_bytes()).map_err(|_| {
                StorageError::Invalid(format!(
                    "git_refs row {repo_id}/{name}: oid_hex is not a hex sha1"
                ))
            })?;
            Ok(RowState::Oid(oid))
        }
        (None, Some(target)) => Ok(RowState::Symref(target)),
        (None, None) => Err(StorageError::Invalid(format!(
            "git_refs row {repo_id}/{name}: neither oid nor symref target — structurally invalid row state"
        ))),
        (Some(_), Some(_)) => Err(StorageError::Invalid(format!(
            "git_refs row {repo_id}/{name}: both oid and symref target — structurally invalid row state"
        ))),
    }
}

/// The listed [`RefLine`] of a row (the module doc's row shapes: an oid
/// row lists plain; the symref row lists the unborn line).
fn row_to_line(name: &str, state: &RowState) -> RefLine {
    match state {
        RowState::Oid(oid) => RefLine {
            name: name.to_owned(),
            oid: Some(*oid),
            peeled: None,
            symref_target: None,
        },
        RowState::Symref(target) => RefLine {
            name: name.to_owned(),
            oid: None,
            peeled: None,
            symref_target: Some(target.clone()),
        },
    }
}

/// The refs surface's row home behind [`cas_decide`], as both impls
/// drive it: one call of [`run_cas`], per-update, in request order.
/// Async surface — the postgres twin is natively async (the
/// tokio-postgres client has no blocking face); the blocking twins'
/// method bodies run on their bridged blocking threads (the sqlite
/// twin polls the whole [`run_cas`] under `Handle::block_on` from one
/// `spawn_blocking` leg, so its blocking-SQL bodies never touch the
/// executor).
trait CasTable {
    /// The named row's current state (`None` = absent), in-call visible
    /// (the updates already applied within the call are seen).
    async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError>;

    /// Apply the commanded CAS write.
    async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError>;

    /// The call's end transition: commit the whole call (`abort =
    /// false`) or discard it (`abort` — the atomic denial: nothing the
    /// call applied survives; the outcomes already carry the denial
    /// mapping at the caller).
    async fn finish(&mut self, abort: bool) -> Result<(), StorageError>;
}

/// The call driver all twins share: one pass over the updates, each
/// decided and (on pass) applied in request order, then the call's end
/// transition. Atomic denies map to the pinned reason matrix (denied
/// refs keep their reasons, the rest report
/// [`ATOMIC_PUSH_FAILURE_REASON`]); the abort discards every applied
/// write via [`CasTable::finish`]. An engine failure propagates early —
/// the impl's transaction drops rolled back and [`StorageError::Io`]
/// surfaces (substrate failure is not a business outcome).
async fn run_cas<T: CasTable + ?Sized>(
    table: &mut T,
    updates: &[RefUpdate],
    atomic: bool,
) -> Result<Vec<RefOutcome>, StorageError> {
    let mut any_denial = false;
    let mut outcomes = Vec::with_capacity(updates.len());
    for update in updates {
        let outcome = match cas_decide(table.state(&update.name).await?, update) {
            Err(denial) => {
                any_denial = true;
                RefOutcome {
                    name: update.name.clone(),
                    result: Err(denial.reason().to_owned()),
                }
            }
            Ok(None) => RefOutcome {
                name: update.name.clone(),
                result: Ok(()),
            },
            Ok(Some(op)) => {
                table.write(&update.name, &op).await?;
                RefOutcome {
                    name: update.name.clone(),
                    result: Ok(()),
                }
            }
        };
        outcomes.push(outcome);
    }
    table.finish(atomic && any_denial).await?;
    if atomic && any_denial {
        for outcome in &mut outcomes {
            if outcome.result.is_ok() {
                outcome.result = Err(ATOMIC_PUSH_FAILURE_REASON.to_owned());
            }
        }
    }
    Ok(outcomes)
}

/// The in-memory refs table — the CAS-semantics reference impl
/// (engine-free, single-process, stamp-free by construction: it builds
/// fresh every time and versions nothing durable — ADR-024 §6). Rows
/// are keyed on the resolved repo id (`repo_id → name → row`), name-sorted
/// maps (deterministic listing); the whole call runs under one
/// short-held write lock — the in-memory "transaction" — with the
/// atomic abort as simply not committing the call's working copy
/// (per-update decisions inside still see their in-call predecessors).
/// The lock is the async [`tokio::sync::RwLock`]: its guard is held
/// across the (never-yielding) CAS futures' awaits and is `Send` —
/// the [`GitRefs`] `Send`-future surface holds (a blocking guard
/// across an await would not).
pub struct InMemoryRefTable {
    rows: tokio::sync::RwLock<BTreeMap<String, BTreeMap<String, RowState>>>,
}

impl Default for InMemoryRefTable {
    fn default() -> Self {
        Self::new()
    }
}

impl InMemoryRefTable {
    /// A fresh, empty table.
    pub fn new() -> Self {
        InMemoryRefTable {
            rows: tokio::sync::RwLock::new(BTreeMap::new()),
        }
    }

    /// The composition's record-transaction seeding leg (substrate.md
    /// §refs, git parity: `git init --bare` creates the dangling HEAD
    /// at init, not at first push) — the twins' same-named SQL leg, the
    /// degenerate shape's row write now that the composition wires this
    /// table. CAS-unreachable by git parity (a unborn symref row
    /// matches nothing).
    pub(crate) async fn seed_head(&self, repo_id: &str, target: &str) {
        self.rows
            .write()
            .await
            .entry(repo_id.to_owned())
            .or_default()
            .insert("HEAD".to_owned(), RowState::Symref(target.to_owned()));
    }

    /// The composition's removal leg (backend.md: repo deletion deletes
    /// the registry row, the repo's ref rows, and the repo's
    /// mutation-marker rows together — the twins' same-named SQL leg,
    /// the degenerate shape's row write).
    pub(crate) async fn drop_repo_rows(&self, repo_id: &str) {
        self.rows.write().await.remove(repo_id);
    }
}

/// One [`InMemoryRefTable`] call's working copy — the in-memory
/// "transaction": decisions and writes land on the copy taken at call
/// start; [`CasTable::finish`] marks the abort. The write-back happens
/// only on a committed call (never an abort, never a substrate `Err` —
/// an early error return drops the working copy whole).
struct InMemoryCasTable {
    working: BTreeMap<String, RowState>,
    aborted: bool,
}

impl CasTable for InMemoryCasTable {
    async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError> {
        Ok(self.working.get(name).cloned())
    }

    async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError> {
        match op {
            CasWrite::Set { oid } => {
                self.working.insert(name.to_owned(), RowState::Oid(*oid));
            }
            CasWrite::Delete => {
                self.working.remove(name);
            }
        }
        Ok(())
    }

    async fn finish(&mut self, abort: bool) -> Result<(), StorageError> {
        if abort {
            self.working.clear();
            self.aborted = true;
        }
        Ok(())
    }
}

#[async_trait]
impl GitRefs for InMemoryRefTable {
    async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
        let guard = self.rows.read().await;
        match guard.get(&repo.repo_id) {
            None => Ok(Vec::new()),
            Some(tier) => Ok(tier
                .iter()
                .map(|(name, state)| row_to_line(name, state))
                .collect()),
        }
    }

    async fn apply_updates(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        atomic: bool,
    ) -> Result<Vec<RefOutcome>, StorageError> {
        let mut rows = self.rows.write().await;
        let mut table = InMemoryCasTable {
            working: rows.get(&repo.repo_id).cloned().unwrap_or_default(),
            aborted: false,
        };
        let outcomes = run_cas(&mut table, &updates, atomic).await?;
        if !table.aborted && (!table.working.is_empty() || rows.contains_key(&repo.repo_id)) {
            rows.insert(repo.repo_id.clone(), table.working);
        }
        Ok(outcomes)
    }
}

/// The sqlite row-CAS twin over the co-tenant `git_refs` rows
/// (substrate.md §refs, ADR-019 §2) — the `atomic` realization as the
/// one-transaction home: every [`GitRefs::apply_updates`] call runs all
/// its CAS check-and-applies inside one `BEGIN IMMEDIATE` transaction
/// on the co-tenant connection (the migration-checked schema from
/// `substrate/w6-migrations`), so two clashing applies — from this
/// process or a co-tenant connection of another — serialize on the row
/// write lock and the loser sees the committed winner (`stale info`),
/// while an atomic denial rolls its transaction back whole, leaving the
/// rows exactly as they were (POC-O's demonstrated shapes, DQ-2).
///
/// Blocking SQL bridges off the async executor
/// (`tokio::task::spawn_blocking`) per the composition's concurrency
/// model; the co-tenant handle's pinned busy timeout absorbs the
/// cross-process write-lock waits. Construction takes the
/// already-migrated handle ([`CoTenantSqlite::open`](crate::migrations::CoTenantSqlite::open)
/// runs the migrations) — no construction I/O here.
#[cfg(feature = "sqlite")]
pub struct SqliteRefTable {
    handle: crate::migrations::CoTenantSqlite,
}

#[cfg(feature = "sqlite")]
impl SqliteRefTable {
    /// Mount the refs row CAS over a co-tenant handle (the same handle
    /// the sqlite registry twin consumes).
    pub fn new(handle: crate::migrations::CoTenantSqlite) -> Self {
        SqliteRefTable { handle }
    }
}

#[cfg(feature = "sqlite")]
mod sqlite_twin {
    use super::*;

    /// One sqlite [`GitRefs::apply_updates`] call's in-transaction
    /// plumbing: the `BEGIN IMMEDIATE` transaction taken up front, the
    /// decisions and row writes inside it, the commit-or-rollback at
    /// [`CasTable::finish`]. An engine failure at any leg propagates
    /// out of [`run_cas`]; the dropped transaction rolls the whole call
    /// back (substrate failure, not a business outcome). The methods'
    /// bodies are the blocking rusqlite calls — polled on the
    /// blocking thread under the [`run_cas`] bridge, never the
    /// executor.
    struct SqliteCasTable<'a> {
        tx: Option<rusqlite::Transaction<'a>>,
        repo_id: String,
    }

    impl SqliteCasTable<'_> {
        fn tx(&self) -> Result<&rusqlite::Transaction<'_>, StorageError> {
            self.tx.as_ref().ok_or_else(|| {
                StorageError::Io("refs CAS transaction already finalized".to_owned())
            })
        }

        fn tx_take(&mut self) -> Result<rusqlite::Transaction<'_>, StorageError> {
            self.tx.take().ok_or_else(|| {
                StorageError::Io("refs CAS transaction already finalized".to_owned())
            })
        }
    }

    impl CasTable for SqliteCasTable<'_> {
        async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError> {
            let tx = self.tx()?;
            let queried = tx.query_row(
                "SELECT oid_hex, symref_target FROM git_refs WHERE repo_id = ?1 AND name = ?2",
                rusqlite::params![self.repo_id, name],
                |row| {
                    Ok((
                        row.get::<_, Option<String>>(0)?,
                        row.get::<_, Option<String>>(1)?,
                    ))
                },
            );
            match queried {
                Ok((oid_hex, symref_target)) => {
                    let state =
                        row_state_from_columns(&self.repo_id, name, oid_hex, symref_target)?;
                    Ok(Some(state))
                }
                Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
                Err(e) => Err(storage_io("refs CAS state read", e)),
            }
        }

        async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError> {
            let tx = self.tx()?;
            match op {
                CasWrite::Set { oid } => {
                    tx.execute(
                        "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
                         VALUES (?1, ?2, ?3, NULL) \
                         ON CONFLICT(repo_id, name) DO UPDATE SET oid_hex = ?3, symref_target = NULL",
                        rusqlite::params![self.repo_id, name, oid.to_hex().to_string()],
                    )
                    .map_err(|e| storage_io("refs CAS row set", e))?;
                }
                CasWrite::Delete => {
                    tx.execute(
                        "DELETE FROM git_refs WHERE repo_id = ?1 AND name = ?2",
                        rusqlite::params![self.repo_id, name],
                    )
                    .map_err(|e| storage_io("refs CAS row delete", e))?;
                }
            }
            Ok(())
        }

        async fn finish(&mut self, abort: bool) -> Result<(), StorageError> {
            let tx = self.tx_take()?;
            if abort {
                tx.rollback()
                    .map_err(|e| storage_io("refs CAS rollback", e))
            } else {
                tx.commit().map_err(|e| storage_io("refs CAS commit", e))
            }
        }
    }

    fn storage_io(context: &str, error: rusqlite::Error) -> StorageError {
        StorageError::Io(format!("{context}: {error}"))
    }

    #[async_trait]
    impl GitRefs for SqliteRefTable {
        async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
            let handle = self.handle.clone();
            let repo_id = repo.repo_id.clone();
            let listed = tokio::task::spawn_blocking(move || {
                let conn = handle.lock();
                let mut stmt = conn
                    .prepare(
                        "SELECT name, oid_hex, symref_target FROM git_refs \
                         WHERE repo_id = ?1 ORDER BY name",
                    )
                    .map_err(|e| storage_io("refs listing prepare", e))?;
                let rows = stmt
                    .query_map(rusqlite::params![repo_id], |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, Option<String>>(1)?,
                            row.get::<_, Option<String>>(2)?,
                        ))
                    })
                    .map_err(|e| storage_io("refs listing query", e))?;
                let mut lines = Vec::new();
                for row in rows {
                    let (name, oid_hex, symref_target) =
                        row.map_err(|e| storage_io("refs listing row", e))?;
                    let state = row_state_from_columns(&repo_id, &name, oid_hex, symref_target)?;
                    lines.push(row_to_line(&name, &state));
                }
                Ok(lines)
            })
            .await
            .map_err(|e| StorageError::Io(format!("refs listing blocking task: {e}")))?;
            listed
        }

        async fn apply_updates(
            &self,
            repo: &RepoRecord,
            updates: Vec<RefUpdate>,
            atomic: bool,
        ) -> Result<Vec<RefOutcome>, StorageError> {
            let handle = self.handle.clone();
            let repo_id = repo.repo_id.clone();
            let runtime = tokio::runtime::Handle::current();
            let applied = tokio::task::spawn_blocking(move || {
                let mut conn = handle.lock();
                let tx = conn
                    .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
                    .map_err(|e| storage_io("refs CAS BEGIN IMMEDIATE", e))?;
                let mut table = SqliteCasTable {
                    tx: Some(tx),
                    repo_id,
                };
                runtime.block_on(run_cas(&mut table, &updates, atomic))
            })
            .await
            .map_err(|e| StorageError::Io(format!("refs CAS blocking task: {e}")))?;
            applied
        }
    }

    /// The liveness seam's row read: one ordered query over the
    /// co-tenant rows (`ORDER BY repo_id, name` — the canonical
    /// order the fingerprint folds), bridged off the executor like
    /// every blocking-SQL leg.
    #[async_trait]
    impl RefRows for SqliteRefTable {
        async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
            let handle = self.handle.clone();
            let rows = tokio::task::spawn_blocking(move || {
                let conn = handle.lock();
                let mut stmt = conn
                    .prepare(
                        "SELECT repo_id, name, oid_hex, symref_target FROM git_refs \
                         ORDER BY repo_id, name",
                    )
                    .map_err(|e| storage_io("refs rows snapshot prepare", e))?;
                let queried = stmt
                    .query_map([], |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, Option<String>>(2)?,
                            row.get::<_, Option<String>>(3)?,
                        ))
                    })
                    .map_err(|e| storage_io("refs rows snapshot query", e))?;
                let mut rows = Vec::new();
                for row in queried {
                    let (repo_id, name, oid_hex, symref_target) =
                        row.map_err(|e| storage_io("refs rows snapshot row", e))?;
                    let state = row_state_from_columns(&repo_id, &name, oid_hex, symref_target)?;
                    rows.push(RefRow {
                        repo_id,
                        name,
                        state: row_state_to_seam(&state),
                    });
                }
                Ok(rows)
            })
            .await
            .map_err(|e| StorageError::Io(format!("refs rows snapshot task: {e}")))?;
            let rows = rows?;
            Ok(RefRowsSnapshot {
                fingerprint: fingerprint_rows(&rows),
                rows,
            })
        }
    }
}

/// The postgres row-CAS twin over the co-tenant `git_refs` rows
/// (substrate.md §refs, ADR-019 §3 — the fleet/durable-shared engine
/// assembly's `GitRefs`, the `postgres` feature; the wave-6 gate's N-c
/// rider). The [`SqliteRefTable`] semantics translated engine-swapped
/// over [`CoTenantPg`](crate::postgres::CoTenantPg): every
/// [`GitRefs::apply_updates`] call runs all its CAS check-and-applies
/// inside **one transaction**, the pg transaction class standing in
/// for sqlite's `BEGIN IMMEDIATE`.
///
/// The transaction's serialization home: the call takes the schema's
/// advisory CAS lock (`pg_advisory_xact_lock` — the pg analog of the
/// sqlite file's write lock) as the transaction's first statement, so
/// two clashing applies — cross-connection included — serialize
/// before their reads: the loser proceeds only after the winner's
/// commit and sees committed state, which under the shared
/// `cas_decide` matrix is exactly the [`CAS_STALE_REASON`] denial.
/// (The same-handle shape serializes on the co-tenant client mutex
/// first anyway; the advisory lock is the whole-fleet home.) Within
/// the transaction, the decisions ride `READ COMMITTED` reads of the
/// co-tenant rows with in-call visibility — the CAS loop's own
/// applies are visible because they are this transaction's writes.
///
/// Registry-family writes are untouched (the `git_refs` rows' other
/// writers — the record transactions' HEAD seed/removal legs — run
/// their own transactions; ref-table writes are not registry-family
/// writes, ADR-019 §7: no `registry_seq` coupling here).
///
/// Natively async — every leg is a straight await on the co-tenant
/// handle's client (the tokio-postgres model; the blocking bridge is
/// the *other* engines'). Construction takes the already-migrated
/// handle ([`CoTenantPg::open`](crate::postgres::CoTenantPg::open)
/// runs the migrations) — no construction I/O here.
#[cfg(feature = "postgres")]
pub struct PgRefTable {
    handle: crate::postgres::CoTenantPg,
}

#[cfg(feature = "postgres")]
impl PgRefTable {
    /// Mount the refs row CAS over a co-tenant handle (the same handle
    /// the pg registry twin consumes; one instance serves both).
    pub fn new(handle: crate::postgres::CoTenantPg) -> Self {
        PgRefTable { handle }
    }
}

#[cfg(feature = "postgres")]
mod pg_twin {
    use super::*;
    use crate::migrations::{git_refs_lock_key, qualify};
    use tokio_postgres::Transaction;

    fn pg_io(context: &str, error: tokio_postgres::Error) -> StorageError {
        StorageError::Io(format!("{context}: {error}"))
    }

    /// One pg [`GitRefs::apply_updates`] call's in-transaction
    /// plumbing: the transaction taken (and the advisory CAS lock
    /// held) by the caller up front, the decisions and row writes
    /// inside it, the commit-or-rollback at [`CasTable::finish`]. An
    /// engine failure at any leg propagates out of [`run_cas`]; the
    /// dropped transaction rolls the whole call back (the
    /// tokio-postgres transaction's drop performs it — substrate
    /// failure, not a business outcome).
    struct PgCasTable<'a> {
        tx: Option<Transaction<'a>>,
        repo_id: String,
        schema: String,
    }

    impl PgCasTable<'_> {
        fn tx(&self) -> Result<&Transaction<'_>, StorageError> {
            self.tx.as_ref().ok_or_else(|| {
                StorageError::Io("refs CAS transaction already finalized".to_owned())
            })
        }
    }

    impl CasTable for PgCasTable<'_> {
        async fn state(&mut self, name: &str) -> Result<Option<RowState>, StorageError> {
            let select = format!(
                "SELECT oid_hex, symref_target FROM {} \
                 WHERE repo_id = $1 AND name = $2",
                qualify(&self.schema, "git_refs")
            );
            let tx = self.tx()?;
            match tx.query_opt(&select, &[&self.repo_id, &name]).await {
                Ok(Some(row)) => {
                    let oid_hex: Option<String> = row
                        .try_get(0)
                        .map_err(|e| pg_io("refs CAS state read", e))?;
                    let symref_target: Option<String> = row
                        .try_get(1)
                        .map_err(|e| pg_io("refs CAS state read", e))?;
                    let state =
                        row_state_from_columns(&self.repo_id, name, oid_hex, symref_target)?;
                    Ok(Some(state))
                }
                Ok(None) => Ok(None),
                Err(e) => Err(pg_io("refs CAS state read", e)),
            }
        }

        async fn write(&mut self, name: &str, op: &CasWrite) -> Result<(), StorageError> {
            let tx = self.tx()?;
            match op {
                CasWrite::Set { oid } => {
                    let upsert = format!(
                        "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
                         VALUES ($1, $2, $3, NULL) \
                         ON CONFLICT (repo_id, name) DO UPDATE \
                         SET oid_hex = $3, symref_target = NULL",
                        qualify(&self.schema, "git_refs")
                    );
                    let oid_hex = oid.to_hex().to_string();
                    tx.execute(&upsert, &[&self.repo_id, &name, &oid_hex])
                        .await
                        .map_err(|e| pg_io("refs CAS row set", e))?;
                }
                CasWrite::Delete => {
                    let delete = format!(
                        "DELETE FROM {} WHERE repo_id = $1 AND name = $2",
                        qualify(&self.schema, "git_refs")
                    );
                    tx.execute(&delete, &[&self.repo_id, &name])
                        .await
                        .map_err(|e| pg_io("refs CAS row delete", e))?;
                }
            }
            Ok(())
        }

        async fn finish(&mut self, abort: bool) -> Result<(), StorageError> {
            let tx = self.tx.take().ok_or_else(|| {
                StorageError::Io("refs CAS transaction already finalized".to_owned())
            })?;
            if abort {
                tx.rollback()
                    .await
                    .map_err(|e| pg_io("refs CAS rollback", e))
            } else {
                tx.commit().await.map_err(|e| pg_io("refs CAS commit", e))
            }
        }
    }

    #[async_trait]
    impl GitRefs for PgRefTable {
        async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
            let select = format!(
                "SELECT name, oid_hex, symref_target FROM {} \
                 WHERE repo_id = $1 ORDER BY name",
                qualify(self.handle.schema(), "git_refs")
            );
            let guard = self.handle.lock().await;
            let rows = guard
                .query(&select, &[&repo.repo_id])
                .await
                .map_err(|e| pg_io("refs listing query", e))?;
            let mut lines = Vec::with_capacity(rows.len());
            for row in &rows {
                let name: String = row.try_get(0).map_err(|e| pg_io("refs listing row", e))?;
                let oid_hex: Option<String> =
                    row.try_get(1).map_err(|e| pg_io("refs listing row", e))?;
                let symref_target: Option<String> =
                    row.try_get(2).map_err(|e| pg_io("refs listing row", e))?;
                let state = row_state_from_columns(&repo.repo_id, &name, oid_hex, symref_target)?;
                lines.push(row_to_line(&name, &state));
            }
            Ok(lines)
        }

        async fn apply_updates(
            &self,
            repo: &RepoRecord,
            updates: Vec<RefUpdate>,
            atomic: bool,
        ) -> Result<Vec<RefOutcome>, StorageError> {
            let mut guard = self.handle.lock().await;
            let repo_id = repo.repo_id.clone();
            let schema = self.handle.schema().to_owned();
            let tx = guard
                .transaction()
                .await
                .map_err(|e| pg_io("refs CAS BEGIN", e))?;
            tx.execute(
                "SELECT pg_advisory_xact_lock($1)",
                &[&git_refs_lock_key(&schema)],
            )
            .await
            .map_err(|e| pg_io("refs CAS advisory lock", e))?;
            let mut table = PgCasTable {
                tx: Some(tx),
                repo_id,
                schema,
            };
            run_cas(&mut table, &updates, atomic).await
        }
    }

    /// The liveness seam's row read: one ordered query over the
    /// co-tenant rows (`ORDER BY repo_id, name` — the canonical
    /// order the fingerprint folds), a straight await (natively
    /// async).
    #[async_trait]
    impl RefRows for PgRefTable {
        async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
            let select = format!(
                "SELECT repo_id, name, oid_hex, symref_target FROM {} \
                 ORDER BY repo_id, name",
                qualify(self.handle.schema(), "git_refs")
            );
            let guard = self.handle.lock().await;
            let rows = guard
                .query(&select, &[])
                .await
                .map_err(|e| pg_io("refs rows snapshot query", e))?;
            let mut snapshot = Vec::new();
            for row in &rows {
                let repo_id: String = row
                    .try_get(0)
                    .map_err(|e| pg_io("refs rows snapshot row", e))?;
                let name: String = row
                    .try_get(1)
                    .map_err(|e| pg_io("refs rows snapshot row", e))?;
                let oid_hex: Option<String> = row
                    .try_get(2)
                    .map_err(|e| pg_io("refs rows snapshot row", e))?;
                let symref_target: Option<String> = row
                    .try_get(3)
                    .map_err(|e| pg_io("refs rows snapshot row", e))?;
                let state = row_state_from_columns(&repo_id, &name, oid_hex, symref_target)?;
                snapshot.push(RefRow {
                    repo_id,
                    name,
                    state: row_state_to_seam(&state),
                });
            }
            drop(guard);
            Ok(RefRowsSnapshot {
                fingerprint: fingerprint_rows(&snapshot),
                rows: snapshot,
            })
        }
    }
}

// The liveness seam's row reads (crate::liveness — the ref-table
// closure's durable state): a plain ordered read of the rows the CAS
// writes, plus the canonical fingerprint over them. Both tables carry
// it; the write-counter wrapper forwards.
//
// The fingerprint's purpose (memo invalidation, cross-process) lives
// at the seam: this is the durable-marker option realized read-side
// — no schema change, the rows themselves are the marker.

fn row_state_to_seam(state: &RowState) -> RefRowState {
    match state {
        RowState::Oid(oid) => RefRowState::Oid(*oid),
        RowState::Symref(target) => RefRowState::Symref(target.clone()),
    }
}

#[async_trait]
impl RefRows for InMemoryRefTable {
    async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
        let guard = self.rows.read().await;
        let rows: Vec<RefRow> = guard
            .iter()
            .flat_map(|(repo_id, tier)| {
                tier.iter().map(move |(name, state)| RefRow {
                    repo_id: repo_id.clone(),
                    name: name.clone(),
                    state: row_state_to_seam(state),
                })
            })
            .collect();
        Ok(RefRowsSnapshot {
            fingerprint: fingerprint_rows(&rows),
            rows,
        })
    }
}

/// The refs write counter — the liveness source's in-process
/// invalidation half (crate::liveness "the arbitration window"):
/// bumped on every CAS call through the composition's refs impl, read
/// by the source's arbitration consult. See [`VersionedRefTable`].
///
/// The wave-6 refs rows carry no version counter; the wrapper is the
/// counter's home (the POC-O table-embedded counter's shape, moved
/// into the composition wiring — the tables stay row-homes).
pub struct VersionedRefTable<T: ?Sized> {
    inner: Arc<T>,
    version: Arc<AtomicU64>,
}

impl<T: ?Sized> VersionedRefTable<T> {
    /// The wrapper over the table and the shared write counter.
    pub fn new(inner: Arc<T>, version: Arc<AtomicU64>) -> Self {
        VersionedRefTable { inner, version }
    }

    /// The wrapped table (the rows' home — the liveness snapshot
    /// forwards through it).
    pub fn inner(&self) -> &Arc<T> {
        &self.inner
    }

    /// The write counter's current value (the liveness source's
    /// arbitration epoch signal — the [`GitRefs`] impl's verbatim
    /// delegation adds only the bump).
    pub fn version(&self) -> u64 {
        self.version.load(Ordering::Relaxed)
    }
}

#[async_trait]
impl<T> GitRefs for VersionedRefTable<T>
where
    T: GitRefs + ?Sized,
{
    async fn list_refs(&self, repo: &RepoRecord) -> Result<Vec<RefLine>, StorageError> {
        self.inner.list_refs(repo).await
    }

    async fn apply_updates(
        &self,
        repo: &RepoRecord,
        updates: Vec<RefUpdate>,
        atomic: bool,
    ) -> Result<Vec<RefOutcome>, StorageError> {
        // The epoch discipline's bump-after-commit half (the
        // [`crate::liveness`] docs): the mark consult's counter load
        // rides ahead of its snapshot read, so every write's bump must
        // land strictly after its committed rows — the only pairing
        // that keeps a memo from pairing an included bump with a
        // snapshot read that missed the write's rows. Every green call
        // bumps — the per-ref denials ride the outcome list, and an
        // all-denied call landing nothing still moves the counter past
        // any consult racing it (over-invalidation, the safe
        // direction); only `Err` bumps nothing.
        let outcomes = self.inner.apply_updates(repo, updates, atomic).await?;
        self.version.fetch_add(1, Ordering::Relaxed);
        Ok(outcomes)
    }
}

#[async_trait]
impl<T> RefRows for VersionedRefTable<T>
where
    T: RefRows + ?Sized,
{
    async fn snapshot(&self) -> Result<RefRowsSnapshot, StorageError> {
        self.inner.snapshot().await
    }
}

#[cfg(test)]
pub(crate) mod battery {
    use alkgit::{
        GitRefs, RefLine, RefOutcome, RefUpdate, RepoRecord, ATOMIC_PUSH_FAILURE_REASON,
        FUNNY_REFNAME_REASON,
    };
    use gix_hash::ObjectId;

    use super::CAS_STALE_REASON;

    fn oid(b: u8) -> ObjectId {
        ObjectId::from([b; 20])
    }

    fn up(name: &str, expected: Option<u8>, new: Option<u8>) -> RefUpdate {
        RefUpdate {
            name: name.to_owned(),
            expected: expected.map(oid),
            new: new.map(oid),
        }
    }

    fn ok(name: &str) -> RefOutcome {
        RefOutcome {
            name: name.to_owned(),
            result: Ok(()),
        }
    }

    fn deny(name: &str, reason: &str) -> RefOutcome {
        RefOutcome {
            name: name.to_owned(),
            result: Err(reason.to_owned()),
        }
    }

    async fn line_of(table: &dyn GitRefs, repo: &RepoRecord, name: &str) -> Option<RefLine> {
        table
            .list_refs(repo)
            .await
            .unwrap()
            .into_iter()
            .find(|line| line.name == name)
    }

    /// The empty-set shapes: an untouched table lists empty; an empty
    /// update call is an empty outcome set (both flags) and nothing
    /// else.
    pub(super) async fn empty_shapes(table: &dyn GitRefs, repo: &RepoRecord) {
        assert!(table.list_refs(repo).await.unwrap().is_empty());
        for atomic in [false, true] {
            let outcomes = table.apply_updates(repo, Vec::new(), atomic).await.unwrap();
            assert!(outcomes.is_empty(), "atomic={atomic}");
        }
        assert!(table.list_refs(repo).await.unwrap().is_empty());
    }

    /// The stale classes (ADR-013 §7): a wrong-`expected` update, a
    /// create-over-existing, a wrong-`expected` delete — each a denied
    /// [`RefOutcome`] (the pinned `stale info` reason), never `Err`,
    /// with the row untouched.
    pub(super) async fn stale_shapes_deny_as_outcomes(table: &dyn GitRefs, repo: &RepoRecord) {
        table
            .apply_updates(repo, vec![up("refs/stale/main", None, Some(1))], false)
            .await
            .unwrap();

        for (shape, update) in [
            (
                "wrong-expected update",
                up("refs/stale/main", Some(2), Some(3)),
            ),
            ("create over existing", up("refs/stale/main", None, Some(3))),
            (
                "wrong-expected delete",
                up("refs/stale/main", Some(9), None),
            ),
        ] {
            let outcomes = table
                .apply_updates(repo, vec![update.clone()], false)
                .await
                .unwrap();
            assert_eq!(
                outcomes,
                vec![deny("refs/stale/main", CAS_STALE_REASON)],
                "{shape} denies as the stale outcome"
            );
            assert_eq!(
                line_of(table, repo, "refs/stale/main").await.unwrap().oid,
                Some(oid(1)),
                "{shape} left the row untouched"
            );
        }
    }

    /// Delete and the force shape: an `expected`-matched delete applies
    /// (the row is gone); a non-fast-forward update applies with no
    /// fast-forward check (the v1 CAS-only policy, ADR-013 §7) — the
    /// unrelated new value lands on a matching `expected`.
    pub(super) async fn delete_applies_and_force_has_no_fast_forward_check(
        table: &dyn GitRefs,
        repo: &RepoRecord,
    ) {
        table
            .apply_updates(repo, vec![up("refs/force/main", None, Some(1))], false)
            .await
            .unwrap();

        let outcomes = table
            .apply_updates(repo, vec![up("refs/force/main", Some(1), None)], false)
            .await
            .unwrap();
        assert_eq!(outcomes, vec![ok("refs/force/main")]);
        assert!(line_of(table, repo, "refs/force/main").await.is_none());

        table
            .apply_updates(repo, vec![up("refs/force/head", None, Some(4))], false)
            .await
            .unwrap();
        let outcomes = table
            .apply_updates(repo, vec![up("refs/force/head", Some(4), Some(5))], false)
            .await
            .unwrap();
        assert_eq!(outcomes, vec![ok("refs/force/head")]);
        assert_eq!(
            line_of(table, repo, "refs/force/head").await.unwrap().oid,
            Some(oid(5)),
            "the non-fast-forward new oid applied — no FF check at the policy leg"
        );
    }

    /// Non-atomic (ADR-018 §6): per-update outcomes in request order —
    /// each ok applies, denials never block the others; plus the
    /// in-call visibility: two updates to one name in one call see
    /// each other.
    pub(super) async fn non_atomic_reports_per_ref_with_in_call_visibility(
        table: &dyn GitRefs,
        repo: &RepoRecord,
    ) {
        table
            .apply_updates(repo, vec![up("refs/na/main", None, Some(1))], false)
            .await
            .unwrap();

        let updates = vec![
            up("refs/na/main", Some(1), Some(2)),
            up("refs/na/topic", Some(3), Some(4)),
            up("refs/na/parked", None, Some(5)),
        ];
        let names = updates.iter().map(|u| u.name.clone()).collect::<Vec<_>>();
        let outcomes = table.apply_updates(repo, updates, false).await.unwrap();
        assert_eq!(
            outcomes.iter().map(|o| o.name.clone()).collect::<Vec<_>>(),
            names,
            "outcomes arrive in request order"
        );
        assert_eq!(outcomes[0].result, Ok(()));
        assert_eq!(outcomes[1].result, Err(CAS_STALE_REASON.to_owned()));
        assert_eq!(outcomes[2].result, Ok(()));
        assert_eq!(
            line_of(table, repo, "refs/na/main").await.unwrap().oid,
            Some(oid(2)),
            "the ok applied even though a sibling was denied"
        );
        assert!(
            line_of(table, repo, "refs/na/topic").await.is_none(),
            "the denied update applied nothing"
        );
        assert!(line_of(table, repo, "refs/na/parked").await.is_some());

        let outcomes = table
            .apply_updates(
                repo,
                vec![
                    up("refs/na/inline", None, Some(1)),
                    up("refs/na/inline", Some(1), Some(2)),
                ],
                false,
            )
            .await
            .unwrap();
        assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
        assert_eq!(
            line_of(table, repo, "refs/na/inline").await.unwrap().oid,
            Some(oid(2)),
            "the second update saw the first's applied row (in-call visibility)"
        );
    }

    /// Atomic denial (ADR-013 §10): nothing applied — the denied refs
    /// keep their own reasons and the other refs of the call report
    /// `atomic push failure`; an all-ok atomic call commits whole
    /// (delete + create in one transaction).
    pub(super) async fn atomic_denials_abort_and_others_report_the_atomic_reason(
        table: &dyn GitRefs,
        repo: &RepoRecord,
    ) {
        table
            .apply_updates(repo, vec![up("refs/atomic/base", None, Some(1))], false)
            .await
            .unwrap();

        let updates = vec![
            up("refs/atomic/base", Some(2), Some(3)),
            up("refs/atomic/topic", None, Some(4)),
            up("refs/atomic/parked", None, Some(5)),
        ];
        let outcomes = table.apply_updates(repo, updates, true).await.unwrap();
        assert_eq!(outcomes.len(), 3);
        assert_eq!(outcomes[0].result, Err(CAS_STALE_REASON.to_owned()));
        assert_eq!(
            outcomes[1].result,
            Err(ATOMIC_PUSH_FAILURE_REASON.to_owned())
        );
        assert_eq!(
            outcomes[2].result,
            Err(ATOMIC_PUSH_FAILURE_REASON.to_owned())
        );
        assert!(
            line_of(table, repo, "refs/atomic/topic").await.is_none(),
            "the atomic abort left the applied leg discarded"
        );
        assert!(
            line_of(table, repo, "refs/atomic/parked").await.is_none(),
            "the atomic abort left the table untouched"
        );
        assert_eq!(
            line_of(table, repo, "refs/atomic/base").await.unwrap().oid,
            Some(oid(1))
        );

        let outcomes = table
            .apply_updates(
                repo,
                vec![
                    up("refs/atomic/base", Some(1), None),
                    up("refs/atomic/topic", None, Some(4)),
                ],
                true,
            )
            .await
            .unwrap();
        assert!(outcomes.iter().all(|outcome| outcome.result.is_ok()));
        assert!(line_of(table, repo, "refs/atomic/base").await.is_none());
        assert_eq!(
            line_of(table, repo, "refs/atomic/topic").await.unwrap().oid,
            Some(oid(4)),
            "an all-ok atomic call committed whole"
        );
        table
            .apply_updates(repo, vec![up("refs/atomic/topic", Some(4), None)], false)
            .await
            .unwrap();
    }

    /// The table-layer name validation's pinned mapping (ADR-013 §9/
    /// §10, substrate.md §refs): a rejected name denies the ref with
    /// `funny refname` — an outcome, never `Err`, whatever the row
    /// state; under atomic the funny refs keep their reason while the
    /// other refs report `atomic push failure` and nothing applies.
    pub(super) async fn invalid_names_deny_as_outcomes(table: &dyn GitRefs, repo: &RepoRecord) {
        table
            .apply_updates(repo, vec![up("refs/funny/main", None, Some(1))], false)
            .await
            .unwrap();

        let outcomes = table
            .apply_updates(
                repo,
                vec![
                    up("refs/funny/main.lock", None, Some(2)),
                    up("refs/alkgit/host", Some(1), None),
                    up("refs/funny/main", Some(1), Some(2)),
                ],
                false,
            )
            .await
            .unwrap();
        assert_eq!(
            outcomes[0],
            deny("refs/funny/main.lock", FUNNY_REFNAME_REASON),
            "the .lock deny-list rule denies at the table layer"
        );
        assert_eq!(
            outcomes[1],
            deny("refs/alkgit/host", FUNNY_REFNAME_REASON),
            "the reserved namespace denies at the table layer"
        );
        assert_eq!(outcomes[2].result, Ok(()));
        assert_eq!(
            line_of(table, repo, "refs/funny/main").await.unwrap().oid,
            Some(oid(2)),
            "the valid ref of the non-atomic call applied"
        );

        let before = table.list_refs(repo).await.unwrap();
        let outcomes = table
            .apply_updates(
                repo,
                vec![
                    up("refs/funny/main.lock", None, Some(9)),
                    up("refs/funny/valid", None, Some(3)),
                ],
                true,
            )
            .await
            .unwrap();
        assert_eq!(
            outcomes[0],
            deny("refs/funny/main.lock", FUNNY_REFNAME_REASON)
        );
        assert_eq!(
            outcomes[1],
            deny("refs/funny/valid", ATOMIC_PUSH_FAILURE_REASON)
        );
        assert_eq!(
            table.list_refs(repo).await.unwrap(),
            before,
            "nothing applied"
        );
    }

    /// The seeded `HEAD → refs/heads/main` symref row lists as the
    /// unborn line (`oid: None`, the target set) — the row shape
    /// `ls-refs=unborn` serves (the OQ-19 rider (a) fixture),
    /// name-sorted before the `refs/` rows; the listing stays
    /// deterministic as rows are added.
    pub(super) async fn unborn_head_lists_the_unborn_line_sorted(
        table: &dyn GitRefs,
        repo: &RepoRecord,
    ) {
        assert_eq!(
            table.list_refs(repo).await.unwrap(),
            vec![RefLine {
                name: "HEAD".to_owned(),
                oid: None,
                peeled: None,
                symref_target: Some("refs/heads/main".to_owned()),
            }]
        );

        table
            .apply_updates(repo, vec![up("refs/heads/zebra", None, Some(1))], false)
            .await
            .unwrap();
        table
            .apply_updates(repo, vec![up("refs/heads/alpaca", None, Some(2))], false)
            .await
            .unwrap();

        let listed = table.list_refs(repo).await.unwrap();
        assert_eq!(
            listed
                .iter()
                .map(|line| line.name.as_str())
                .collect::<Vec<_>>(),
            vec!["HEAD", "refs/heads/alpaca", "refs/heads/zebra"],
            "deterministic name-sorted listing, the unborn HEAD first"
        );
        assert_eq!(listed[0].oid, None);
        assert_eq!(listed[0].symref_target.as_deref(), Some("refs/heads/main"));
        assert_eq!(listed[1].oid, Some(oid(2)));
        assert_eq!(listed[2].oid, Some(oid(1)));
    }

    /// The symref row is CAS-unreachable: create-over, update, and
    /// delete against `HEAD` all deny stale (a symref row's oid is
    /// unborn — it matches nothing; git parity: pushes never touch the
    /// repo's HEAD row).
    pub(super) async fn symref_row_is_cas_unreachable(table: &dyn GitRefs, repo: &RepoRecord) {
        for shape in [
            up("HEAD", None, Some(1)),
            up("HEAD", Some(1), Some(2)),
            up("HEAD", Some(1), None),
        ] {
            let outcomes = table.apply_updates(repo, vec![shape], false).await.unwrap();
            assert_eq!(
                outcomes,
                vec![deny("HEAD", CAS_STALE_REASON)],
                "the HEAD symref row never matches a CAS expectation"
            );
        }
        let head = table
            .list_refs(repo)
            .await
            .unwrap()
            .into_iter()
            .find(|line| line.name == "HEAD")
            .expect("the HEAD symref row survived every denied shape");
        assert_eq!(head.oid, None);
        assert_eq!(head.symref_target.as_deref(), Some("refs/heads/main"));
    }
}

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

    use crate::testutil::private_record as repo;
    #[cfg(feature = "sqlite")]
    use crate::testutil::unique_db_path;

    fn oid(b: u8) -> ObjectId {
        ObjectId::from([b; 20])
    }

    #[test]
    fn the_denial_and_atomic_reasons_are_the_pinned_texts() {
        assert_eq!(CAS_STALE_REASON, "stale info");
        assert_eq!(alkgit::ATOMIC_PUSH_FAILURE_REASON, "atomic push failure");
        assert_eq!(alkgit::FUNNY_REFNAME_REASON, "funny refname");
    }

    /// The battery over the engine-free reference impl — the CAS
    /// semantics this task pins (ADR-013 §7/§10's shapes).
    #[tokio::test]
    async fn inmemory_table_carries_the_cas_battery() {
        let table: Arc<dyn GitRefs> = Arc::new(InMemoryRefTable::new());
        let doubled = repo("owner/doubled");

        battery::empty_shapes(table.as_ref(), &doubled).await;
        battery::stale_shapes_deny_as_outcomes(table.as_ref(), &doubled).await;
        battery::delete_applies_and_force_has_no_fast_forward_check(table.as_ref(), &doubled).await;
        battery::non_atomic_reports_per_ref_with_in_call_visibility(table.as_ref(), &doubled).await;
        battery::atomic_denials_abort_and_others_report_the_atomic_reason(table.as_ref(), &doubled)
            .await;
        battery::invalid_names_deny_as_outcomes(table.as_ref(), &doubled).await;
    }

    /// The unborn-HEAD shapes over the seeded row (each test writes the
    /// HEAD symref row directly — the `substrate/w6-registry-sqlite`
    /// seeding rule's row shape; put_repo's seeding is not this task's).
    #[tokio::test]
    async fn inmemory_unborn_head_row_lists_and_holds_cas() {
        let table = Arc::new(InMemoryRefTable::new());
        let unborn = repo("owner/unborn");

        table.rows.write().await.insert(
            unborn.repo_id.clone(),
            [(
                "HEAD".to_owned(),
                RowState::Symref("refs/heads/main".to_owned()),
            )]
            .into_iter()
            .collect(),
        );

        battery::unborn_head_lists_the_unborn_line_sorted(table.as_ref(), &unborn).await;
        battery::symref_row_is_cas_unreachable(table.as_ref(), &unborn).await;
    }

    /// Non-atomic deletes clear the call's own tier too — a committed
    /// delete-only call's working copy replaces the stored rows (a
    /// delete-emptied repo lists empty, never stale leftovers).
    #[tokio::test]
    async fn inmemory_committed_delete_all_empties_the_tier() {
        let table = InMemoryRefTable::new();
        let doubled = repo("owner/delete-all");
        table
            .apply_updates(
                &doubled,
                vec![
                    RefUpdate {
                        name: "refs/heads/main".to_owned(),
                        expected: None,
                        new: Some(oid(1)),
                    },
                    RefUpdate {
                        name: "refs/other".to_owned(),
                        expected: None,
                        new: Some(oid(2)),
                    },
                ],
                false,
            )
            .await
            .unwrap();
        assert_eq!(table.list_refs(&doubled).await.unwrap().len(), 2);

        table
            .apply_updates(
                &doubled,
                vec![
                    RefUpdate {
                        name: "refs/heads/main".to_owned(),
                        expected: Some(oid(1)),
                        new: None,
                    },
                    RefUpdate {
                        name: "refs/other".to_owned(),
                        expected: Some(oid(2)),
                        new: None,
                    },
                ],
                false,
            )
            .await
            .unwrap();
        assert!(
            table.list_refs(&doubled).await.unwrap().is_empty(),
            "the delete-only call emptied the tier"
        );
    }

    /// The battery over the sqlite row-CAS twin — the same semantics in
    /// rows, one `BEGIN IMMEDIATE` transaction per call.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn sqlite_table_carries_the_cas_battery() {
        use crate::migrations::CoTenantSqlite;

        let path = unique_db_path("refs", "battery");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let table: Arc<dyn GitRefs> = Arc::new(SqliteRefTable::new(handle.clone()));
        let doubled = repo("owner/doubled");

        battery::empty_shapes(table.as_ref(), &doubled).await;
        battery::stale_shapes_deny_as_outcomes(table.as_ref(), &doubled).await;
        battery::delete_applies_and_force_has_no_fast_forward_check(table.as_ref(), &doubled).await;
        battery::non_atomic_reports_per_ref_with_in_call_visibility(table.as_ref(), &doubled).await;
        battery::atomic_denials_abort_and_others_report_the_atomic_reason(table.as_ref(), &doubled)
            .await;
        battery::invalid_names_deny_as_outcomes(table.as_ref(), &doubled).await;

        let _ = std::fs::remove_file(&path);
    }

    /// The unborn-HEAD fixture over the seeded rows (the seeding rule's
    /// row shape seeded directly).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn sqlite_unborn_head_row_lists_and_holds_cas() {
        use crate::migrations::CoTenantSqlite;

        let path = unique_db_path("refs", "unborn");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let unborn = repo("owner/unborn");
        handle
            .lock()
            .execute(
                "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
                 VALUES (?1, 'HEAD', NULL, 'refs/heads/main')",
                rusqlite::params![unborn.repo_id],
            )
            .unwrap();
        let table: Arc<dyn GitRefs> = Arc::new(SqliteRefTable::new(handle.clone()));

        battery::unborn_head_lists_the_unborn_line_sorted(table.as_ref(), &unborn).await;
        battery::symref_row_is_cas_unreachable(table.as_ref(), &unborn).await;

        let _ = std::fs::remove_file(&path);
    }

    /// POC-O's demonstrated contention shape (DQ-2): two clashing
    /// applies from two co-tenant connections of one file — the loser
    /// sees the committed winner (`stale info`); the transactions
    /// serialize on the row write lock, never erroring (pinned 5000 ms
    /// busy timeout).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn sqlite_two_clashing_applies_from_two_connections() {
        use crate::migrations::CoTenantSqlite;

        let path = unique_db_path("refs", "clash");
        let first =
            Arc::new(SqliteRefTable::new(CoTenantSqlite::open(&path).unwrap())) as Arc<dyn GitRefs>;
        let second =
            Arc::new(SqliteRefTable::new(CoTenantSqlite::open(&path).unwrap())) as Arc<dyn GitRefs>;
        let clashers = repo("owner/clash");

        let a = tokio::spawn({
            let first = first.clone();
            let clashers = clashers.clone();
            async move {
                first
                    .apply_updates(
                        &clashers,
                        vec![RefUpdate {
                            name: "refs/heads/main".to_owned(),
                            expected: None,
                            new: Some(oid(1)),
                        }],
                        false,
                    )
                    .await
            }
        });
        let b = tokio::spawn({
            let second = second.clone();
            let clashers = clashers.clone();
            async move {
                second
                    .apply_updates(
                        &clashers,
                        vec![RefUpdate {
                            name: "refs/heads/main".to_owned(),
                            expected: None,
                            new: Some(oid(2)),
                        }],
                        false,
                    )
                    .await
            }
        });
        let (a_out, b_out) = tokio::join!(a, b);
        let (a_out, b_out) = (a_out.unwrap().unwrap(), b_out.unwrap().unwrap());
        let a_won = a_out[0].result.is_ok();
        assert!(
            a_won != b_out[0].result.is_ok(),
            "two clashing creates on one row: exactly one wins"
        );
        let loser = if a_won {
            b_out[0].clone()
        } else {
            a_out[0].clone()
        };
        assert_eq!(loser.result, Err(CAS_STALE_REASON.to_owned()));
        let winner_oid = if a_won { Some(oid(1)) } else { Some(oid(2)) };
        let winner = if a_won { first.clone() } else { second.clone() };
        let winner: Arc<dyn GitRefs> = winner;
        assert_eq!(
            winner.list_refs(&clashers).await.unwrap()[0].oid,
            winner_oid,
            "the committed winner's row is the row the table carries"
        );

        drop(first);
        drop(second);
        let _ = std::fs::remove_file(&path);
    }

    /// The corrupted-row defenses: neither column set, both columns
    /// set, or a non-hex oid — structurally invalid row state maps
    /// [`StorageError::Invalid`] on listing. The in-memory impl carries
    /// the same shape as a two-arm enum — the mis-state is
    /// unexpressible there by construction.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn sqlite_corrupted_row_states_map_storage_invalid() {
        use crate::migrations::CoTenantSqlite;

        let path = unique_db_path("refs", "corrupt");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let table = SqliteRefTable::new(handle.clone());
        let corrupt = repo("owner/corrupt");

        handle
            .lock()
            .execute(
                "INSERT INTO git_refs (repo_id, name, oid_hex, symref_target) \
                 VALUES (?1, 'refs/heads/bent', NULL, NULL)",
                rusqlite::params![corrupt.repo_id],
            )
            .unwrap();
        assert!(matches!(
            table.list_refs(&corrupt).await.unwrap_err(),
            StorageError::Invalid(_)
        ));

        handle
            .lock()
            .execute(
                "UPDATE git_refs SET oid_hex = ?1, symref_target = 'refs/heads/main' \
             WHERE repo_id = ?2 AND name = 'refs/heads/bent'",
                rusqlite::params![oid(1).to_string(), corrupt.repo_id],
            )
            .unwrap();
        assert!(matches!(
            table.list_refs(&corrupt).await.unwrap_err(),
            StorageError::Invalid(_)
        ));

        handle
            .lock()
            .execute(
                "UPDATE git_refs SET oid_hex = 'not-hex', symref_target = NULL \
             WHERE repo_id = ?1 AND name = 'refs/heads/bent'",
                rusqlite::params![corrupt.repo_id],
            )
            .unwrap();
        assert!(matches!(
            table.list_refs(&corrupt).await.unwrap_err(),
            StorageError::Invalid(_)
        ));

        drop(table);
        drop(handle);
        let _ = std::fs::remove_file(&path);
    }
}

/// The pg twin's battery — the same CAS script the in-memory reference
/// and the sqlite twin run, over the live-pg harness (the
/// `ALKBLOBS_PG_*`/`ALKGIT_SUBSTRATE_PG_*` convention — one container,
/// one convention; clean skip when the host env is absent, so
/// `--all-features` stays green everywhere). Every test works in a
/// unique, freshly-migrated PostgreSQL schema and drops it after — the
/// harness database stays untouched.
#[cfg(all(test, feature = "postgres"))]
mod pg_twin_tests {
    use crate::postgres::live_harness::{drop_schema, live_pg, unique_schema};
    use crate::testutil::private_record as repo;
    use alkgit::{GitRefs, StorageError};
    use gix_hash::ObjectId;

    use super::battery;

    fn oid(b: u8) -> ObjectId {
        ObjectId::from([b; 20])
    }

    use super::PgRefTable;

    /// The battery over the pg row-CAS twin — the semantics pinned on
    /// the reference impl, re-run engine-swapped (no drift possible:
    /// the same shared `cas_decide` + `run_cas` decide both).
    #[tokio::test]
    async fn pg_table_carries_the_cas_battery() {
        let dsn = live_pg!();
        let schema = unique_schema("refs-battery");
        let table = PgRefTable::new(
            crate::postgres::CoTenantPg::open(&dsn, &schema)
                .await
                .unwrap(),
        );
        let doubled = repo("owner/doubled");

        let table = std::sync::Arc::new(table);
        let table: std::sync::Arc<dyn GitRefs> = table;

        battery::empty_shapes(table.as_ref(), &doubled).await;
        battery::stale_shapes_deny_as_outcomes(table.as_ref(), &doubled).await;
        battery::delete_applies_and_force_has_no_fast_forward_check(table.as_ref(), &doubled).await;
        battery::non_atomic_reports_per_ref_with_in_call_visibility(table.as_ref(), &doubled).await;
        battery::atomic_denials_abort_and_others_report_the_atomic_reason(table.as_ref(), &doubled)
            .await;
        battery::invalid_names_deny_as_outcomes(table.as_ref(), &doubled).await;

        drop(table);
        drop_schema(&dsn, &schema).await;
    }

    /// The unborn-HEAD fixture over the seeded row (the seeding rule's
    /// row shape, seeded directly — put_repo's seeding is not this
    /// task's; the pg twin's migrations already carry the table).
    #[tokio::test]
    async fn pg_unborn_head_row_lists_and_holds_cas() {
        let dsn = live_pg!();
        let schema = unique_schema("refs-unborn");
        let handle = crate::postgres::CoTenantPg::open(&dsn, &schema)
            .await
            .unwrap();
        let unborn = repo("owner/unborn");
        handle
            .lock()
            .await
            .execute(
                &format!(
                    "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
                     VALUES ($1, 'HEAD', NULL, 'refs/heads/main')",
                    crate::migrations::qualify(&schema, "git_refs")
                ),
                &[&unborn.repo_id],
            )
            .await
            .unwrap();
        let table = std::sync::Arc::new(PgRefTable::new(handle));
        let table: std::sync::Arc<dyn GitRefs> = table;

        battery::unborn_head_lists_the_unborn_line_sorted(table.as_ref(), &unborn).await;
        battery::symref_row_is_cas_unreachable(table.as_ref(), &unborn).await;

        drop(table);
        drop_schema(&dsn, &schema).await;
    }

    /// POC-O's demonstrated contention shape (DQ-2) over pg: two
    /// clashing applies from two co-tenant connections of one schema —
    /// the advisory CAS lock serializes the transactions before their
    /// reads, the loser sees the committed winner (`stale info`), and
    /// neither errors.
    #[tokio::test]
    async fn pg_two_clashing_applies_from_two_connections() {
        use super::CAS_STALE_REASON;

        let dsn = live_pg!();
        let schema = unique_schema("refs-clash");
        let first = std::sync::Arc::new(PgRefTable::new(
            crate::postgres::CoTenantPg::open(&dsn, &schema)
                .await
                .unwrap(),
        )) as std::sync::Arc<dyn GitRefs>;
        let second = std::sync::Arc::new(PgRefTable::new(
            crate::postgres::CoTenantPg::open(&dsn, &schema)
                .await
                .unwrap(),
        )) as std::sync::Arc<dyn GitRefs>;
        let clashers = repo("owner/clash");

        let a = tokio::spawn({
            let first = first.clone();
            let clashers = clashers.clone();
            async move {
                first
                    .apply_updates(
                        &clashers,
                        vec![alkgit::RefUpdate {
                            name: "refs/heads/main".to_owned(),
                            expected: None,
                            new: Some(oid(1)),
                        }],
                        false,
                    )
                    .await
            }
        });
        let b = tokio::spawn({
            let second = second.clone();
            let clashers = clashers.clone();
            async move {
                second
                    .apply_updates(
                        &clashers,
                        vec![alkgit::RefUpdate {
                            name: "refs/heads/main".to_owned(),
                            expected: None,
                            new: Some(oid(2)),
                        }],
                        false,
                    )
                    .await
            }
        });
        let (a_out, b_out) = tokio::join!(a, b);
        let (a_out, b_out) = (a_out.unwrap().unwrap(), b_out.unwrap().unwrap());
        let a_won = a_out[0].result.is_ok();
        assert!(
            a_won != b_out[0].result.is_ok(),
            "two clashing creates on one row: exactly one wins"
        );
        let loser = if a_won {
            b_out[0].clone()
        } else {
            a_out[0].clone()
        };
        assert_eq!(loser.result, Err(CAS_STALE_REASON.to_owned()));
        let winner_oid = if a_won { oid(1) } else { oid(2) };
        let winner = if a_won { first } else { second };
        assert_eq!(
            winner.list_refs(&clashers).await.unwrap()[0].oid,
            Some(winner_oid),
            "the committed winner's row is the row the table carries"
        );

        drop(winner);
        drop_schema(&dsn, &schema).await;
    }

    /// The corrupted-row defenses: neither column set, both columns
    /// set, or a non-hex oid — structurally invalid row state maps
    /// [`StorageError::Invalid`] on listing (the mis-states are the pg
    /// rows' expressible shape; the in-memory two-arm enum cannot
    /// express them).
    #[tokio::test]
    async fn pg_corrupted_row_states_map_storage_invalid() {
        let dsn = live_pg!();
        let schema = unique_schema("refs-corrupt");
        let handle = crate::postgres::CoTenantPg::open(&dsn, &schema)
            .await
            .unwrap();
        let table = PgRefTable::new(handle.clone());
        let corrupt = repo("owner/corrupt");
        let table_name = crate::migrations::qualify(&schema, "git_refs");

        let second = crate::postgres::live_harness::connect_direct(&dsn).await;
        second
            .execute(
                &format!(
                    "INSERT INTO {table_name} (repo_id, name, oid_hex, symref_target) \
                     VALUES ($1, 'refs/heads/bent', NULL, NULL)"
                ),
                &[&corrupt.repo_id],
            )
            .await
            .unwrap();
        assert!(
            matches!(
                table.list_refs(&corrupt).await.unwrap_err(),
                StorageError::Invalid(_)
            ),
            "neither column set is structurally invalid"
        );

        second
            .execute(
                &format!(
                    "UPDATE {table_name} SET oid_hex = $1, symref_target = 'refs/heads/main' \
                     WHERE repo_id = $2 AND name = 'refs/heads/bent'"
                ),
                &[&oid(1).to_string(), &corrupt.repo_id],
            )
            .await
            .unwrap();
        assert!(
            matches!(
                table.list_refs(&corrupt).await.unwrap_err(),
                StorageError::Invalid(_)
            ),
            "both columns set is structurally invalid"
        );

        second
            .execute(
                &format!(
                    "UPDATE {table_name} SET oid_hex = 'not-hex', symref_target = NULL \
                     WHERE repo_id = $1 AND name = 'refs/heads/bent'"
                ),
                &[&corrupt.repo_id],
            )
            .await
            .unwrap();
        assert!(matches!(
            table.list_refs(&corrupt).await.unwrap_err(),
            StorageError::Invalid(_)
        ));

        drop(second);
        drop_schema(&dsn, &schema).await;
    }

    /// The parity discipline (the wave-6 posture extended to the refs
    /// twin): the same canonical row set snapshots identically over
    /// the pg twin and the in-memory reference — rows and the
    /// fingerprint both (the liveness memo's cross-engine rebuild key
    /// is one value whatever the engine).
    #[tokio::test]
    async fn pg_snapshot_matches_the_reference_impl_canonically() {
        use super::InMemoryRefTable;
        use crate::liveness::RefRows;

        let dsn = live_pg!();
        let schema = unique_schema("refs-parity");
        let handle = crate::postgres::CoTenantPg::open(&dsn, &schema)
            .await
            .unwrap();
        let seeded = repo("owner/parity");
        let sister = repo("owner/parity-sister");

        handle
            .lock()
            .await
            .execute(
                &format!(
                    "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
                     VALUES ($1, 'HEAD', NULL, 'refs/heads/main'), \
                            ($1, 'refs/heads/main', $2, NULL)",
                    crate::migrations::qualify(&schema, "git_refs")
                ),
                &[&seeded.repo_id, &oid(1).to_string()],
            )
            .await
            .unwrap();
        handle
            .lock()
            .await
            .execute(
                &format!(
                    "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
                     VALUES ($1, 'refs/tags/v1', $2, NULL)",
                    crate::migrations::qualify(&schema, "git_refs")
                ),
                &[&sister.repo_id, &oid(9).to_string()],
            )
            .await
            .unwrap();

        let mem = InMemoryRefTable::new();
        {
            use super::RowState;
            let mut rows = mem.rows.write().await;
            rows.insert(
                seeded.repo_id.clone(),
                [
                    (
                        "HEAD".to_owned(),
                        RowState::Symref("refs/heads/main".to_owned()),
                    ),
                    ("refs/heads/main".to_owned(), RowState::Oid(oid(1))),
                ]
                .into_iter()
                .collect(),
            );
            rows.insert(
                sister.repo_id.clone(),
                [("refs/tags/v1".to_owned(), RowState::Oid(oid(9)))]
                    .into_iter()
                    .collect(),
            );
        }

        let pg = PgRefTable::new(handle);
        assert_eq!(
            pg.snapshot().await.unwrap(),
            mem.snapshot().await.unwrap(),
            "the same rows snapshot identically — one canonical form, \
             one fingerprint, whatever the engine"
        );
        assert_eq!(
            pg.list_refs(&seeded).await.unwrap(),
            mem.list_refs(&seeded).await.unwrap()
        );

        drop(pg);
        drop(mem);
        drop_schema(&dsn, &schema).await;
    }
}