alkgit-substrate 0.1.0

//! The LFS pointer-lifecycle row store (wave 9) — the durable
//! reference map ADR-022 §4 places alkgit-side: one row per reference,
//! `(repo_id, oid, size)`, over the co-tenant `lfs_objects` table that
//! ships day-one in both engine step sets (ADR-024 §5 — the table is
//! unconditional in the schema, the `lfs` feature gates this *surface*
//! consuming it, so the stamp stays unit-shaped). The repo_id is the
//! registry-resolved id (the ops namespace on the batch's ACL path);
//! the oid is the LFS media oid (64 lowercase hex — the raw sha256 of
//! the content, byte-exact the `GitObjectSha256` digest under alkblobs
//! ADR-018 §1, though the row store itself carries no key arithmetic);
//! the surface is consumed by the batch adapter (assertion at
//! check/upload time, per-repo reads) and the LFS liveness source (the
//! full row read feeding the sweep chain).
//!
//! # The reference surface ([`LfsRows`])
//!
//! - [`LfsRows::assert`] — insert a reference, or the refcount-aware
//!   upsert: re-asserting the same `(repo_id, oid)` pair in one repo
//!   is one row (the pair is the primary key; no second row, no
//!   refcount bump) and the row's size refreshes to the asserted
//!   value (the oid is the content digest, so honest traffic asserts
//!   the same size — the last assertion's metadata wins).
//! - [`LfsRows::remove`] — remove one reference row. An absent row is
//!   a success no-op (removal is a cleanup op — no error path there).
//! - [`LfsRows::remove_repo`] — the repo-deletion leg: the repo's
//!   rows go with it. The composition's record-removal transactions
//!   carry the twins' in-SQL copies of this leg unconditionally
//!   (lfs.md §3: rows go with the repo, same removal as
//!   refs/registry); the mem shape's leg rides its record-transaction
//!   legs in the composition entry ([`crate::compose`] — the
//!   `MemSeededBackend` pattern). The unconditional posture: the
//!   table is unconditional in both step sets, the rows only exist
//!   when the `lfs` surface wrote them, and a mixed deployment that
//!   turned the feature off must not leak a removed repo's rows.
//! - [`LfsRows::rows`] / [`LfsRows::repo_rows`] — the liveness read
//!   (all rows / per-repo rows, canonical `(repo_id, oid)` order):
//!   the shape the LFS liveness source and the batch adapter consume.
//!
//! # The refcount semantics (the cross-reference re-protection
//! contract, ADR-022 §4)
//!
//! The CAS pool is global — the same content referenced by two repos
//! is one key with two rows — so the liveness source must re-protect
//! the key while *any* row references it (POC-L's refcount
//! requirement, replacing the POC's crude second live-row scan). The
//! `refcount` column carries that data:
//!
//! - **What counts**: a row's refcount is the number of rows
//!   referencing the row's oid (itself included) — with the pair
//!   primary key, the number of distinct repos referencing the oid.
//! - **What bumps it**: exactly one event — a newly asserted row
//!   (a previously-unrepresented `(repo_id, oid)` pair). The
//!   refcount-aware upsert (a same-pair re-assert) changes nothing.
//! - **What decrements it**: exactly one event — a removed row
//!   ([`LfsRows::remove`], or the repo-deletion legs). Every
//!   surviving row of the oid carries the recomputed count, so every
//!   row of an oid reads the same current count whatever the engine.
//! - **What reads it**: the liveness reads ([`LfsRow::refcount`) —
//!   consumed by the wave-9 liveness source (the re-protection
//!   contract's data: while a key's count is ≥ 1 the key stays
//!   referenced; the last row's removal makes the media
//!   sweep-eligible) and the OQ-20 deferred per-oid pruning's future
//!   queries.
//!
//! The engine twins maintain the column transactionally: one
//! transaction per call (the refs-rows posture, ADR-019 §2 — the
//! sqlite twin's `BEGIN IMMEDIATE`, the pg twin's transaction *and*
//! its schema advisory lock, the sqlite write-lock class's analog
//! serializing the same-oid recompute across co-tenant connections),
//! recompute-after-write so concurrent mutations converge on exact
//! counts. The in-memory impl recomputes from the row map —
//! exact-by-construction.
//!
//! # Repo-lifetime scoping v1
//!
//! Rows live from assertion (batch check/upload — the moment the
//! server learns the pair) to repo deletion; no per-oid pruning —
//! OQ-20 stands deferred (first concrete row-volume need). There is
//! no sweep, no TTL, and *no* mutation of rows anywhere outside this
//! surface's ops and the two removal legs.
//!
//! # Errors
//!
//! The `StorageError` family (the refs tables' posture — the
//! co-tenant row stores speak the core's object-storage errors):
//! table-layer input validation is [`StorageError::Invalid`] (a
//! non-64-hex-lowercase oid, an empty repo id, a size over the
//! engine's integer range), engine failures are
//! [`StorageError::Io`] (message-only — no engine source types on
//! the surface).

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

use alkgit::StorageError;
use async_trait::async_trait;

/// One LFS pointer-lifecycle row as the surface reads it back: the
/// resolved repo id, the 64-hex LFS oid, the asserted size, and the
/// oid's current refcount (the module doc's refcount semantics).
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LfsRow {
    /// The registry-resolved repo id the reference belongs to.
    pub repo_id: String,
    /// The LFS media oid (64 lowercase hex — the raw sha256 of the
    /// content).
    pub oid: String,
    /// The asserted media size (the batch's mandatory size; the last
    /// assertion's value).
    pub size: u64,
    /// The oid's current reference count — rows referencing the oid,
    /// itself included (the re-protection contract's data).
    pub refcount: u64,
}

/// The LFS pointer-lifecycle row surface (substrate.md §public API's
/// "the LFS liveness/pointer-lifecycle surface under `lfs`" — the
/// row-op half; the exact names this module pins are the OQ-03
/// substrate-side inventory of the wave-9 rows task). Three impls in
/// the refs-tables' pattern:
/// [`InMemoryLfsRows`] (the reference impl, engine-free), the sqlite
/// twin ([`SqliteLfsRows`], the `sqlite` feature), and the postgres
/// twin ([`PgLfsRows`], the `postgres` + `lfs` features).
#[async_trait]
pub trait LfsRows: Send + Sync {
    /// Assert a reference: insert the `(repo_id, oid)` row with the
    /// asserted size, or — when the pair already has a row — refresh
    /// that row's size and change nothing else (the refcount-aware
    /// upsert: one oid re-asserted in one repo is one row).
    async fn assert(&self, repo_id: &str, oid: &str, size: u64) -> Result<(), StorageError>;

    /// Remove one reference row. An absent row is a success no-op;
    /// the oid's surviving rows recompute their refcount.
    async fn remove(&self, repo_id: &str, oid: &str) -> Result<(), StorageError>;

    /// The repo-deletion leg: the repo's rows go with the repo (its
    /// oids' surviving rows in other repos recompute their refcount).
    async fn remove_repo(&self, repo_id: &str) -> Result<(), StorageError>;

    /// The liveness read — all rows, canonical `(repo_id, oid)`
    /// order.
    async fn rows(&self) -> Result<Vec<LfsRow>, StorageError>;

    /// The liveness read — one repo's rows, canonical oid order.
    async fn repo_rows(&self, repo_id: &str) -> Result<Vec<LfsRow>, StorageError>;
}

/// The table-layer input validation (the pinned mapping):
/// the repo id is the registry-resolved id (non-empty),
/// the oid is a 64-lowercase-hex sha256 form, and the size fits the
/// engine twins' integer range. Everything else the caller's surface
/// layer validated.
fn validated_repo_id(repo_id: &str) -> Result<(), StorageError> {
    if repo_id.is_empty() {
        return Err(StorageError::Invalid(
            "lfs reference assertion: the repo id is empty (the registry id is resolved per \
             call)"
                .to_owned(),
        ));
    }
    Ok(())
}

fn validated_oid(oid: &str) -> Result<(), StorageError> {
    let hex_ok = oid.len() == 64 && oid.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'));
    if !hex_ok {
        return Err(StorageError::Invalid(format!(
            "lfs reference assertion: the oid is not a 64-lowercase-hex sha256 form: {oid:?}"
        )));
    }
    Ok(())
}

fn validated_size(size: u64) -> Result<(), StorageError> {
    if size > i64::MAX as u64 {
        return Err(StorageError::Invalid(format!(
            "lfs reference assertion: the size exceeds the engine integer range: {size}"
        )));
    }
    Ok(())
}

/// The in-memory LFS row store — the semantically exact reference
/// impl, engine-free and single-process in the crate's base (the
/// refs-tables' pattern: no engine dependency, no feature beyond the
/// `lfs` gate this surface carries — ADR-025's placement rule). Rows
/// live in a `(repo_id, oid) → size` sorted map (canonical order
/// falls out of the BTreeMap); the refcounts recompute from the map
/// on every mutation (exact-by-construction). The whole call runs
/// under one short-held write lock (the in-memory "transaction"; the
/// async `tokio::sync::RwLock`) over **both** maps — the rows and
/// the refcounts mutate in the same locked section, so a concurrent
/// pair of same-oid mutations serializes with exact recomputes (a
/// two-lock split would let interleaved phases land in reversed
/// order: a stale one-off undercount — the wave-9 gate's fix).
pub struct InMemoryLfsRows {
    state: tokio::sync::RwLock<InMemoryLfsState>,
}

#[derive(Default)]
struct InMemoryLfsState {
    rows: BTreeMap<(String, String), u64>,
    refcounts: BTreeMap<String, u64>,
}

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

impl InMemoryLfsRows {
    /// A fresh, empty store.
    pub fn new() -> Self {
        InMemoryLfsRows {
            state: tokio::sync::RwLock::new(InMemoryLfsState::default()),
        }
    }

    /// The oid's recomputed count after a mutation (the map's rows
    /// referencing the oid).
    fn recount(rows: &BTreeMap<(String, String), u64>, oid: &str) -> u64 {
        rows.keys().filter(|(_, o)| o == oid).count() as u64
    }
}

#[async_trait]
impl LfsRows for InMemoryLfsRows {
    async fn assert(&self, repo_id: &str, oid: &str, size: u64) -> Result<(), StorageError> {
        validated_repo_id(repo_id)?;
        validated_oid(oid)?;
        validated_size(size)?;
        let mut state = self.state.write().await;
        let newly = state
            .rows
            .insert((repo_id.to_owned(), oid.to_owned()), size)
            .is_none();
        if newly {
            let count = Self::recount(&state.rows, oid);
            state.refcounts.insert(oid.to_owned(), count);
        }
        Ok(())
    }

    async fn remove(&self, repo_id: &str, oid: &str) -> Result<(), StorageError> {
        validated_repo_id(repo_id)?;
        validated_oid(oid)?;
        let mut state = self.state.write().await;
        if state
            .rows
            .remove(&(repo_id.to_owned(), oid.to_owned()))
            .is_some()
        {
            let count = Self::recount(&state.rows, oid);
            if count == 0 {
                state.refcounts.remove(oid);
            } else {
                state.refcounts.insert(oid.to_owned(), count);
            }
        }
        Ok(())
    }

    async fn remove_repo(&self, repo_id: &str) -> Result<(), StorageError> {
        validated_repo_id(repo_id)?;
        let mut state = self.state.write().await;
        let affected: Vec<String> = state
            .rows
            .keys()
            .filter(|(r, _)| r == repo_id)
            .map(|(_, o)| o.clone())
            .collect();
        state.rows.retain(|r, _| r.0 != repo_id);
        for oid in affected {
            let count = Self::recount(&state.rows, &oid);
            if count == 0 {
                state.refcounts.remove(&oid);
            } else {
                state.refcounts.insert(oid, count);
            }
        }
        Ok(())
    }

    async fn rows(&self) -> Result<Vec<LfsRow>, StorageError> {
        Ok(self.read_all().await)
    }

    async fn repo_rows(&self, repo_id: &str) -> Result<Vec<LfsRow>, StorageError> {
        validated_repo_id(repo_id)?;
        Ok(self
            .read_all()
            .await
            .into_iter()
            .filter(|row| row.repo_id == repo_id)
            .collect())
    }
}

impl InMemoryLfsRows {
    async fn read_all(&self) -> Vec<LfsRow> {
        let state = self.state.read().await;
        state
            .rows
            .iter()
            .map(|((repo_id, oid), size)| LfsRow {
                repo_id: repo_id.clone(),
                oid: oid.clone(),
                size: *size,
                refcount: state.refcounts.get(oid).copied().unwrap_or(0),
            })
            .collect()
    }
}

/// The sqlite row twin over the co-tenant `lfs_objects` rows
/// (substrate.md §refs' row-store pattern, ADR-019 §2): every write
/// call ([`LfsRows::assert`], [`LfsRows::remove`],
/// [`LfsRows::remove_repo`]) runs its check-and-writes inside **one**
/// `BEGIN IMMEDIATE` transaction on the co-tenant connection — two
/// clashing calls (cross-connection included) serialize on the write
/// lock and the refcount recompute reads the committed winner. The
/// refcount recompute is the transaction's trailing statement — the
/// touched oid's rows all carry the same current count at commit.
/// Blocking SQL bridges off the async executor
/// (`tokio::task::spawn_blocking`) per the composition's concurrency
/// model; construction takes the already-migrated
/// ([`CoTenantSqlite::open`](crate::migrations::CoTenantSqlite::open))
/// handle — no construction I/O here. The record-removal leg that
/// takes a repo's rows with the record rides the registry twin's own
/// in-SQL copy ([`crate::sqlite`] — unconditional, see the module
/// doc); this impl's [`LfsRows::remove_repo`] is the surface's own
/// leg for the same row set.
#[cfg(feature = "sqlite")]
pub struct SqliteLfsRows {
    handle: crate::migrations::CoTenantSqlite,
}

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

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

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

    /// The refcount recompute — the touched oid's rows all carry the
    /// count of rows referencing the oid at commit (the module doc's
    /// "what decrements" half; trailing statement of the write
    /// transaction).
    fn recount_sql() -> &'static str {
        "UPDATE lfs_objects \
         SET refcount = (SELECT count(*) FROM lfs_objects AS c WHERE c.oid = lfs_objects.oid) \
         WHERE oid = ?1"
    }

    /// The repo-deletion recompute: the affected oids are exactly the
    /// ones any other repo may share, so one recount statement over
    /// the whole table is the exact single-statement form (no capture
    /// needed — the deleted rows' oids are recomputed wherever rows
    /// survive).
    fn recount_all_sql() -> &'static str {
        "UPDATE lfs_objects \
         SET refcount = (SELECT count(*) FROM lfs_objects AS c WHERE c.oid = lfs_objects.oid)"
    }

    fn read_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<LfsRow> {
        Ok(LfsRow {
            repo_id: row.get(0)?,
            oid: row.get(1)?,
            size: row.get::<_, i64>(2)? as u64,
            refcount: row.get::<_, i64>(3)? as u64,
        })
    }

    async fn run_blocking<T: Send + 'static>(
        handle: crate::migrations::CoTenantSqlite,
        work: impl FnOnce(&mut rusqlite::Connection) -> Result<T, StorageError> + Send + 'static,
    ) -> Result<T, StorageError> {
        tokio::task::spawn_blocking(move || {
            let mut conn = handle.lock();
            work(&mut conn)
        })
        .await
        .map_err(|e| StorageError::Io(format!("lfs rows blocking task: {e}")))?
    }

    async fn run_transaction<T: Send + 'static>(
        handle: crate::migrations::CoTenantSqlite,
        work: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<T, StorageError> + Send + 'static,
    ) -> Result<T, StorageError> {
        run_blocking(handle, move |conn| {
            let tx = conn
                .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
                .map_err(|e| storage_io("lfs rows BEGIN IMMEDIATE", e))?;
            let value = work(&tx)?;
            tx.commit().map_err(|e| storage_io("lfs rows commit", e))?;
            Ok(value)
        })
        .await
    }

    #[async_trait]
    impl LfsRows for SqliteLfsRows {
        async fn assert(&self, repo_id: &str, oid: &str, size: u64) -> Result<(), StorageError> {
            validated_repo_id(repo_id)?;
            validated_oid(oid)?;
            validated_size(size)?;
            let handle = self.handle.clone();
            let (repo_id, oid) = (repo_id.to_owned(), oid.to_owned());
            run_transaction(handle, move |tx| {
                tx.execute(
                    "INSERT INTO lfs_objects (repo_id, oid, size, refcount) \
                     VALUES (?1, ?2, ?3, 1) \
                     ON CONFLICT(repo_id, oid) DO UPDATE SET size = excluded.size",
                    rusqlite::params![repo_id, oid, size as i64],
                )
                .map_err(|e| storage_io("lfs assert upsert", e))?;
                tx.execute(recount_sql(), rusqlite::params![oid])
                    .map_err(|e| storage_io("lfs assert refcount recount", e))?;
                Ok(())
            })
            .await
        }

        async fn remove(&self, repo_id: &str, oid: &str) -> Result<(), StorageError> {
            validated_repo_id(repo_id)?;
            validated_oid(oid)?;
            let handle = self.handle.clone();
            let (repo_id, oid) = (repo_id.to_owned(), oid.to_owned());
            run_transaction(handle, move |tx| {
                tx.execute(
                    "DELETE FROM lfs_objects WHERE repo_id = ?1 AND oid = ?2",
                    rusqlite::params![repo_id, oid],
                )
                .map_err(|e| storage_io("lfs reference remove", e))?;
                tx.execute(recount_sql(), rusqlite::params![oid])
                    .map_err(|e| storage_io("lfs remove refcount recount", e))?;
                Ok(())
            })
            .await
        }

        async fn remove_repo(&self, repo_id: &str) -> Result<(), StorageError> {
            validated_repo_id(repo_id)?;
            let handle = self.handle.clone();
            let repo_id = repo_id.to_owned();
            run_transaction(handle, move |tx| {
                tx.execute(
                    "DELETE FROM lfs_objects WHERE repo_id = ?1",
                    rusqlite::params![repo_id],
                )
                .map_err(|e| storage_io("lfs repo removal", e))?;
                tx.execute(recount_all_sql(), [])
                    .map_err(|e| storage_io("lfs repo removal recount", e))?;
                Ok(())
            })
            .await
        }

        async fn rows(&self) -> Result<Vec<LfsRow>, StorageError> {
            let handle = self.handle.clone();
            run_blocking(handle, move |conn| {
                let mut stmt = conn
                    .prepare(
                        "SELECT repo_id, oid, size, refcount FROM lfs_objects \
                              ORDER BY repo_id, oid",
                    )
                    .map_err(|e| storage_io("lfs rows listing prepare", e))?;
                let rows = stmt
                    .query_map([], read_row)
                    .map_err(|e| storage_io("lfs rows listing query", e))?;
                rows.collect::<Result<Vec<_>, _>>()
                    .map_err(|e| storage_io("lfs rows listing row", e))
            })
            .await
        }

        async fn repo_rows(&self, repo_id: &str) -> Result<Vec<LfsRow>, StorageError> {
            validated_repo_id(repo_id)?;
            let handle = self.handle.clone();
            let repo_id = repo_id.to_owned();
            run_blocking(handle, move |conn| {
                let mut stmt = conn
                    .prepare(
                        "SELECT repo_id, oid, size, refcount FROM lfs_objects \
                              WHERE repo_id = ?1 ORDER BY oid",
                    )
                    .map_err(|e| storage_io("lfs rows listing prepare", e))?;
                let rows = stmt
                    .query_map(rusqlite::params![repo_id], read_row)
                    .map_err(|e| storage_io("lfs rows listing query", e))?;
                rows.collect::<Result<Vec<_>, _>>()
                    .map_err(|e| storage_io("lfs rows listing row", e))
            })
            .await
        }
    }
}

/// The postgres row twin over the co-tenant `lfs_objects` rows
/// (substrate.md §refs' row-store pattern, ADR-019 §3): every write
/// call runs its check-and-writes inside **one** transaction that
/// takes the schema's advisory row-store lock
/// (`pg_advisory_xact_lock` — the sqlite twin's `BEGIN IMMEDIATE`
/// write-lock class's whole-fleet analog), so the refcount recompute
/// of two clashing calls — cross-connection included — serializes
/// before their reads and every surviving row of an oid carries the
/// same current count at commit. Natively async — every leg is a
/// straight await on the co-tenant handle's client (the
/// tokio-postgres model). Construction takes the already-migrated
/// ([`CoTenantPg::open`](crate::postgres::CoTenantPg::open)) handle —
/// no construction I/O here. The record-removal leg's in-SQL copy
/// rides [`crate::postgres`]'s registry removal transaction
/// (unconditional, see the module doc).
#[cfg(feature = "postgres")]
pub struct PgLfsRows {
    handle: crate::postgres::CoTenantPg,
}

#[cfg(feature = "postgres")]
impl PgLfsRows {
    /// Mount the row store over a co-tenant handle (the same handle
    /// the pg registry and refs twins consume; one instance per
    /// schema serves all three).
    pub fn new(handle: crate::postgres::CoTenantPg) -> Self {
        PgLfsRows { handle }
    }
}

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

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

    /// The refcount recompute — the touched oid's rows all carry the
    /// count at commit (the sqlite twin's trailing statement,
    /// engine-swapped; the target alias keeps the outer reference
    /// unambiguous). The qualified table comes in as a parameter —
    /// the statement composes from it, no string surgery.
    fn recount_sql(table: &str) -> String {
        format!(
            "UPDATE {table} AS t0 \
             SET refcount = (SELECT count(*) FROM {table} c WHERE c.oid = t0.oid) \
             WHERE t0.oid = $1"
        )
    }

    fn recount_all_sql(table: &str) -> String {
        format!(
            "UPDATE {table} AS t0 \
             SET refcount = (SELECT count(*) FROM {table} c WHERE c.oid = t0.oid)"
        )
    }

    #[async_trait]
    impl LfsRows for PgLfsRows {
        async fn assert(&self, repo_id: &str, oid: &str, size: u64) -> Result<(), StorageError> {
            validated_repo_id(repo_id)?;
            validated_oid(oid)?;
            validated_size(size)?;
            let schema = self.handle.schema().to_owned();
            let size = size as i64;
            let (repo_id, oid) = (repo_id.to_owned(), oid.to_owned());
            let mut guard = self.handle.lock().await;
            let tx = guard
                .transaction()
                .await
                .map_err(|e| pg_io("lfs rows BEGIN", e))?;
            tx.execute(
                "SELECT pg_advisory_xact_lock($1)",
                &[&lfs_objects_lock_key(&schema)],
            )
            .await
            .map_err(|e| pg_io("lfs rows advisory lock", e))?;
            let upsert = format!(
                "INSERT INTO {} (repo_id, oid, size, refcount) \
                 VALUES ($1, $2, $3, 1) \
                 ON CONFLICT (repo_id, oid) DO UPDATE SET size = excluded.size",
                qualify(&schema, "lfs_objects")
            );
            tx.execute(&upsert, &[&repo_id, &oid, &size])
                .await
                .map_err(|e| pg_io("lfs assert upsert", e))?;
            tx.execute(&recount_sql(&qualify(&schema, "lfs_objects")), &[&oid])
                .await
                .map_err(|e| pg_io("lfs assert refcount recount", e))?;
            tx.commit().await.map_err(|e| pg_io("lfs rows commit", e))?;
            Ok(())
        }

        async fn remove(&self, repo_id: &str, oid: &str) -> Result<(), StorageError> {
            validated_repo_id(repo_id)?;
            validated_oid(oid)?;
            let schema = self.handle.schema().to_owned();
            let (repo_id, oid) = (repo_id.to_owned(), oid.to_owned());
            let mut guard = self.handle.lock().await;
            let tx = guard
                .transaction()
                .await
                .map_err(|e| pg_io("lfs rows BEGIN", e))?;
            tx.execute(
                "SELECT pg_advisory_xact_lock($1)",
                &[&lfs_objects_lock_key(&schema)],
            )
            .await
            .map_err(|e| pg_io("lfs rows advisory lock", e))?;
            let delete = format!(
                "DELETE FROM {} WHERE repo_id = $1 AND oid = $2",
                qualify(&schema, "lfs_objects")
            );
            tx.execute(&delete, &[&repo_id, &oid])
                .await
                .map_err(|e| pg_io("lfs reference remove", e))?;
            tx.execute(&recount_sql(&qualify(&schema, "lfs_objects")), &[&oid])
                .await
                .map_err(|e| pg_io("lfs remove refcount recount", e))?;
            tx.commit().await.map_err(|e| pg_io("lfs rows commit", e))?;
            Ok(())
        }

        async fn remove_repo(&self, repo_id: &str) -> Result<(), StorageError> {
            validated_repo_id(repo_id)?;
            let schema = self.handle.schema().to_owned();
            let repo_id = repo_id.to_owned();
            let mut guard = self.handle.lock().await;
            let tx = guard
                .transaction()
                .await
                .map_err(|e| pg_io("lfs rows BEGIN", e))?;
            tx.execute(
                "SELECT pg_advisory_xact_lock($1)",
                &[&lfs_objects_lock_key(&schema)],
            )
            .await
            .map_err(|e| pg_io("lfs rows advisory lock", e))?;
            let delete = format!(
                "DELETE FROM {} WHERE repo_id = $1",
                qualify(&schema, "lfs_objects")
            );
            tx.execute(&delete, &[&repo_id])
                .await
                .map_err(|e| pg_io("lfs repo removal", e))?;
            let recount = recount_all_sql(&qualify(&schema, "lfs_objects"));
            tx.execute(&recount, &[])
                .await
                .map_err(|e| pg_io("lfs repo removal recount", e))?;
            tx.commit().await.map_err(|e| pg_io("lfs rows commit", e))?;
            Ok(())
        }

        async fn rows(&self) -> Result<Vec<LfsRow>, StorageError> {
            let select = format!(
                "SELECT repo_id, oid, size, refcount FROM {} ORDER BY repo_id, oid",
                qualify(self.handle.schema(), "lfs_objects")
            );
            let guard = self.handle.lock().await;
            let rows = guard
                .query(&select, &[])
                .await
                .map_err(|e| pg_io("lfs rows listing query", e))?;
            flatten_rows(&rows)
        }

        async fn repo_rows(&self, repo_id: &str) -> Result<Vec<LfsRow>, StorageError> {
            validated_repo_id(repo_id)?;
            let select = format!(
                "SELECT repo_id, oid, size, refcount FROM {} WHERE repo_id = $1 ORDER BY oid",
                qualify(self.handle.schema(), "lfs_objects")
            );
            let guard = self.handle.lock().await;
            let rows = guard
                .query(&select, &[&repo_id])
                .await
                .map_err(|e| pg_io("lfs rows listing query", e))?;
            flatten_rows(&rows)
        }
    }

    fn flatten_rows(rows: &[tokio_postgres::Row]) -> Result<Vec<LfsRow>, StorageError> {
        rows.iter()
            .map(|row| {
                Ok(LfsRow {
                    repo_id: row
                        .try_get(0)
                        .map_err(|e| pg_io("lfs rows listing row", e))?,
                    oid: row
                        .try_get(1)
                        .map_err(|e| pg_io("lfs rows listing row", e))?,
                    size: row
                        .try_get::<_, i64>(2)
                        .map_err(|e| pg_io("lfs rows listing row", e))?
                        as u64,
                    refcount: row
                        .try_get::<_, i64>(3)
                        .map_err(|e| pg_io("lfs rows listing row", e))?
                        as u64,
                })
            })
            .collect()
    }
}

#[cfg(test)]
pub(crate) mod battery {
    use super::*;

    /// The pinned oid fixture: a valid 64-hex form (the digits spell
    /// out the row's identity in the tests).
    pub(crate) fn oid_hex(tag: u8) -> String {
        let base = format!("{tag:02x}").repeat(32);
        base
    }

    /// Assertion + the refcount-aware upsert: a new pair inserts one
    /// row with refcount 1; a same-pair re-assert is the same row
    /// (refcount unmoved, the asserted size refreshed); a different
    /// repo asserting the same oid bumps both rows' refcount.
    pub(crate) async fn assert_and_the_refcount_aware_upsert(
        table: &dyn LfsRows,
    ) -> Result<(), StorageError> {
        table.assert("repo/a", &oid_hex(1), 128).await?;
        table.assert("repo/a", &oid_hex(1), 128).await?;
        assert_eq!(
            table.repo_rows("repo/a").await?,
            vec![LfsRow {
                repo_id: "repo/a".to_owned(),
                oid: oid_hex(1),
                size: 128,
                refcount: 1,
            }],
            "the same oid re-asserted in one repo is one row"
        );

        table.assert("repo/a", &oid_hex(1), 256).await?;
        assert_eq!(
            table.repo_rows("repo/a").await?[0].size,
            256,
            "the re-assert refreshes the row's size"
        );
        assert_eq!(table.repo_rows("repo/a").await?[0].refcount, 1);

        // A second repo references the same content: two rows, one
        // key — both carry the current count of 2.
        table.assert("repo/b", &oid_hex(1), 256).await?;
        let shared: Vec<LfsRow> = table
            .rows()
            .await?
            .into_iter()
            .filter(|row| row.oid == oid_hex(1))
            .collect();
        assert_eq!(shared.len(), 2);
        for row in &shared {
            assert_eq!(
                row.refcount, 2,
                "every row of an oid reads the same count: {row:?}"
            );
        }
        Ok(())
    }

    /// The re-protection contract, test-pinned (ADR-022 §4): two
    /// repos referencing one oid — one row removed ⇒ the key stays
    /// referenced by the survivor (its refcount decremented to 1);
    /// all rows removed ⇒ nothing references the key (sweep-eligible
    /// — no rows; the liveness source's sweep contract realizes the
    /// deletion end-to-end, wave-9 liveness task).
    pub(crate) async fn cross_reference_reprotection(
        table: &dyn LfsRows,
    ) -> Result<(), StorageError> {
        table.assert("repo/one", &oid_hex(2), 64).await?;
        table.assert("repo/two", &oid_hex(2), 64).await?;
        assert_eq!(
            table
                .rows()
                .await?
                .into_iter()
                .filter(|row| row.oid == oid_hex(2))
                .count(),
            2
        );

        table.remove("repo/one", &oid_hex(2)).await?;
        let survivor = table.repo_rows("repo/two").await?;
        assert_eq!(
            survivor,
            vec![LfsRow {
                repo_id: "repo/two".to_owned(),
                oid: oid_hex(2),
                size: 64,
                refcount: 1,
            }],
            "one row removed — the survivor carries the decremented count; the key stays referenced"
        );

        table.remove("repo/two", &oid_hex(2)).await?;
        assert!(
            table
                .rows()
                .await?
                .into_iter()
                .all(|row| row.oid != oid_hex(2)),
            "all rows removed — nothing references the key (sweep-eligible)"
        );
        Ok(())
    }

    /// Removing an absent row: a success no-op — no error path, no
    /// refcount disturbance.
    pub(crate) async fn removal_of_an_absent_row_is_a_noop(
        table: &dyn LfsRows,
    ) -> Result<(), StorageError> {
        table.assert("repo/noop", &oid_hex(3), 16).await?;
        table.remove("repo/absent", &oid_hex(3)).await?;
        table.remove("repo/noop", &oid_hex(0x3f)).await?;
        let rows = table.repo_rows("repo/noop").await?;
        assert_eq!(rows.len(), 1);
        assert_eq!(
            rows[0].refcount, 1,
            "the no-op removals left the count exact"
        );
        Ok(())
    }

    /// The repo-deletion leg: the repo's rows go with it (all of
    /// them — multi-oid), the surviving repos' shared oids recompute,
    /// a shared oid fully owned by the repo leaves no refcount entry.
    pub(crate) async fn repo_deletion_takes_the_rows_and_recounts(
        table: &dyn LfsRows,
    ) -> Result<(), StorageError> {
        table.assert("victim", &oid_hex(4), 32).await?;
        table.assert("victim", &oid_hex(5), 48).await?;
        table.assert("survivor", &oid_hex(4), 32).await?;
        table.assert("survivor", &oid_hex(9), 72).await?;

        table.remove_repo("victim").await?;

        let touched: Vec<LfsRow> = table
            .rows()
            .await?
            .into_iter()
            .filter(|row| row.repo_id == "survivor" || row.repo_id == "victim")
            .collect();
        assert_eq!(
            touched,
            vec![
                LfsRow {
                    repo_id: "survivor".to_owned(),
                    oid: oid_hex(4),
                    size: 32,
                    refcount: 1,
                },
                LfsRow {
                    repo_id: "survivor".to_owned(),
                    oid: oid_hex(9),
                    size: 72,
                    refcount: 1,
                },
            ],
            "the victim's rows went with the repo; the survivor's shared oid recomputed to 1"
        );

        // Re-asserting the victim's name starts a fresh row set.
        table.assert("victim", &oid_hex(4), 32).await?;
        assert_eq!(
            table.repo_rows("victim").await?[0].refcount,
            2,
            "a shared oid re-referenced by the former victim counts both rows again"
        );
        Ok(())
    }

    /// The liveness read's shape: canonical `(repo_id, oid)` order
    /// over all rows; `repo_rows` filters one repo's view.
    pub(crate) async fn the_reads_list_canonically(
        table: &dyn LfsRows,
    ) -> Result<(), StorageError> {
        table.assert("repo/z", &oid_hex(6), 8).await?;
        table.assert("repo/z", &oid_hex(7), 9).await?;
        table.assert("repo/a", &oid_hex(8), 10).await?;

        let all = table.rows().await?;
        let keys: Vec<(String, String)> = all
            .iter()
            .map(|r| (r.repo_id.clone(), r.oid.clone()))
            .collect();
        let mut sorted = keys.clone();
        sorted.sort();
        assert_eq!(
            keys, sorted,
            "all rows arrive in canonical (repo_id, oid) order"
        );

        let mine = table.repo_rows("repo/z").await?;
        assert_eq!(
            mine.iter().map(|r| r.oid.clone()).collect::<Vec<_>>(),
            vec![oid_hex(6), oid_hex(7)],
            "the per-repo view filters and stays oid-ordered"
        );
        Ok(())
    }

    /// The table-layer validation (the pinned mapping): oid shape,
    /// repo id, and the engine integer range are
    /// [`StorageError::Invalid`]; the row state is untouched.
    pub(crate) async fn validation_rejects_as_storage_invalid(
        table: &dyn LfsRows,
    ) -> Result<(), StorageError> {
        let before = table.rows().await?.len();

        for oid in [
            "",
            "nothex",
            &"AB".repeat(32),
            &"1".repeat(63),
            &"1".repeat(65),
        ] {
            assert!(
                matches!(
                    table.assert("repo/a", oid, 8).await,
                    Err(StorageError::Invalid(_))
                ),
                "the oid shape is a table-layer input: {oid:?}"
            );
        }
        assert!(
            matches!(
                table.assert("", &oid_hex(1), 8).await,
                Err(StorageError::Invalid(_))
            ),
            "an empty repo id is table-layer invalid"
        );
        assert!(
            matches!(
                table
                    .assert("repo/a", &oid_hex(1), (i64::MAX as u64) + 1)
                    .await,
                Err(StorageError::Invalid(_))
            ),
            "a size over the engine integer range is table-layer invalid"
        );
        assert_eq!(
            table.rows().await?.len(),
            before,
            "the rejected inputs wrote nothing"
        );
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "sqlite")]
    use std::sync::Arc;

    use super::battery;
    use super::*;

    #[cfg(feature = "sqlite")]
    use crate::testutil::scratch_dir;

    async fn in_memory_reference() -> InMemoryLfsRows {
        InMemoryLfsRows::new()
    }

    /// The reference impl runs the full battery — the semantics
    /// pinned here are exactly what the engine twins re-run
    /// engine-swapped.
    #[tokio::test]
    async fn the_in_memory_reference_runs_the_battery() {
        let table = in_memory_reference().await;
        battery::assert_and_the_refcount_aware_upsert(&table)
            .await
            .unwrap();
        battery::cross_reference_reprotection(&table).await.unwrap();
        battery::removal_of_an_absent_row_is_a_noop(&table)
            .await
            .unwrap();
        battery::repo_deletion_takes_the_rows_and_recounts(&table)
            .await
            .unwrap();
        battery::the_reads_list_canonically(&table).await.unwrap();
        battery::validation_rejects_as_storage_invalid(&table)
            .await
            .unwrap();
    }

    /// The refcount reads back through the rows read exactly as the
    /// engine twins' recomputed SQL column would (the shared read
    /// shape) — pinned over the maintained map, including the
    /// empty-oid absence (no rows ⇒ no entry).
    #[tokio::test]
    async fn the_in_memory_refcounts_track_the_rows() {
        let table = in_memory_reference().await;
        let refcount_of = |rows: &Vec<LfsRow>, repo: &str| -> u64 {
            rows.iter()
                .find(|row| row.repo_id == repo)
                .map(|row| row.refcount)
                .expect("the row exists")
        };
        table.assert("r", &battery::oid_hex(1), 8).await.unwrap();
        table.assert("r", &battery::oid_hex(1), 8).await.unwrap();
        assert_eq!(
            refcount_of(&table.rows().await.unwrap(), "r"),
            1,
            "the re-assert did not bump the count"
        );
        table.assert("r2", &battery::oid_hex(1), 8).await.unwrap();
        assert_eq!(refcount_of(&table.rows().await.unwrap(), "r2"), 2);
        table.remove("r", &battery::oid_hex(1)).await.unwrap();
        assert_eq!(refcount_of(&table.rows().await.unwrap(), "r2"), 1);
        table.remove("r2", &battery::oid_hex(1)).await.unwrap();
        assert!(
            table.rows().await.unwrap().is_empty(),
            "the last removal leaves no row, no refcount entry"
        );
    }

    /// The versioned wrapper forwards the battery exactly and bumps
    /// the shared counter per **committed** `assert` — re-asserts
    /// included — while rejected asserts and removals read back
    /// without a bump: the epoch discipline's bump-after-commit half
    /// (a bump must never land ahead of the rows it covers, or a
    /// mark's load-before-read consult could pair the bump with a
    /// rows read that missed the in-flight commit). A rejected assert
    /// makes no coverage change, so a missed bump is exact there, and
    /// the mark's next rebuild sees the truth either way.
    #[tokio::test]
    async fn the_versioned_wrapper_forwards_and_bumps_on_assert() {
        let inner = Arc::new(InMemoryLfsRows::new());
        let counter = Arc::new(AtomicU64::new(0));
        let wrapped = VersionedLfsRows::new(inner.clone(), Arc::clone(&counter));
        let count = || counter.load(Ordering::Relaxed);

        battery::assert_and_the_refcount_aware_upsert(&wrapped)
            .await
            .unwrap();
        assert_eq!(
            count(),
            4,
            "every assert bumps — same-pair re-asserts included"
        );

        let before = count();
        battery::cross_reference_reprotection(&wrapped)
            .await
            .unwrap();
        assert_eq!(
            count() - before,
            2,
            "the two asserts bumped; the two removals read back without one"
        );

        let before = count();
        battery::removal_of_an_absent_row_is_a_noop(&wrapped)
            .await
            .unwrap();
        assert_eq!(
            count() - before,
            1,
            "the one assert bumped — the no-op removals did not"
        );
        let before = count();
        battery::the_reads_list_canonically(&wrapped).await.unwrap();
        assert_eq!(count() - before, 3, "the reads themselves do not bump");

        let before = count();
        battery::validation_rejects_as_storage_invalid(&wrapped)
            .await
            .unwrap();
        assert_eq!(
            count() - before,
            0,
            "the rejected asserts bump nothing (no rows landed — the bump-after-commit discipline)"
        );

        let before = count();
        battery::removal_of_an_absent_row_is_a_noop(&wrapped)
            .await
            .unwrap();
        assert_eq!(count(), before + 1, "removals never bump");
    }

    /// The sqlite twin runs the same battery over the co-tenant
    /// `lfs_objects` rows (one transaction per call).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn the_sqlite_twin_runs_the_battery() {
        let dir = scratch_dir("lfs", "sqlite-battery");
        let handle = crate::migrations::CoTenantSqlite::open(dir.join("rows.sqlite3")).unwrap();
        let table = SqliteLfsRows::new(handle);

        battery::assert_and_the_refcount_aware_upsert(&table)
            .await
            .unwrap();
        battery::cross_reference_reprotection(&table).await.unwrap();
        battery::removal_of_an_absent_row_is_a_noop(&table)
            .await
            .unwrap();
        battery::repo_deletion_takes_the_rows_and_recounts(&table)
            .await
            .unwrap();
        battery::the_reads_list_canonically(&table).await.unwrap();
        battery::validation_rejects_as_storage_invalid(&table)
            .await
            .unwrap();

        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }

    /// The recomputed refcount column is exact across concurrent
    /// same-oid assertion storms (two connections asserting the same
    /// oid — the write lock serializes, every row ends at the same
    /// exact count).
    #[cfg(feature = "sqlite")]
    #[tokio::test(flavor = "multi_thread")]
    async fn the_sqlite_concurrent_asserts_converge_on_exact_counts() {
        let dir = scratch_dir("lfs", "sqlite-race");
        let path = dir.join("race.sqlite3");
        let first = Arc::new(SqliteLfsRows::new(
            crate::migrations::CoTenantSqlite::open(&path).unwrap(),
        ));
        let second = Arc::new(SqliteLfsRows::new(
            crate::migrations::CoTenantSqlite::open(&path).unwrap(),
        ));

        let oid = battery::oid_hex(2);
        let mut joins = Vec::new();
        for store in [first.clone(), second.clone()] {
            for repo in ["repo/a", "repo/b", "repo/c", "repo/d"] {
                joins.push(tokio::spawn({
                    let store = store.clone();
                    let oid = oid.clone();
                    let repo = repo.to_owned();
                    async move { store.assert(&repo, &oid, 64).await }
                }));
            }
        }
        for join in joins {
            join.await.unwrap().expect("every assert lands");
        }

        let rows = first.rows().await.unwrap();
        assert_eq!(rows.len(), 4, "four repos — four rows, one key");
        for row in &rows {
            assert_eq!(
                row.refcount, 4,
                "every row carries the exact count: {row:?}"
            );
        }

        std::fs::remove_dir_all(&dir).unwrap_or_default();
    }
}

/// The LFS rows' write counter — the LFS liveness source's in-process
/// invalidation half ([`crate::liveness::LfsMedia`]'s arbitration
/// epoch signal): bumped on every `assert` through the composition's
/// rows surface, read by the source's arbitration consult. See
/// [`VersionedLfsRows`].
///
/// The counter's discipline is narrower than the refs counter's
/// ([`crate::refs::VersionedRefTable`]): `assert` bumps, removals read
/// back without one. The bump covers the one dangerous direction — a
/// row landing
/// between a sweep's mark and a staged candidate's arbitration (the
/// media is *newly* referenced; the stale index says unreferenced; the
/// moved counter says unknown-is-live and cancels). A removal moves
/// coverage only downward — a stale "live" answer is the safe side and
/// a stale "dead" answer is exact for a candidate whose rows left —
/// and the mark's rebuild-per-mark consult sees the truth at the next
/// sweep either way.
///
/// The bump follows the inner call (the epoch discipline's
/// bump-after-commit half, shared with
/// [`crate::refs::VersionedRefTable`]): the mark consult's counter
/// load rides **ahead** of its rows read, so only the
/// load-before-read / bump-after-commit pairing is sound — bump
/// lands strictly after the rows commit, and any write whose rows
/// the mark's read missed necessarily moves the counter past the
/// memo's captured epoch (the arbitration consult answers
/// conservatively-live). The bump-before shape was reviewed out at
/// the wave-9 gate: it let an assert's bump land while its sqlite/pg
/// write transaction was still in flight, letting a mark's row read
/// miss the uncommitted rows while catching the bump — a memo keyed
/// on pre-assert rows and a post-bump epoch, the poisoned shape that
/// answers exact-dead for just-referenced media once the commit
/// lands.
///
/// The twins' record-removal legs (the in-SQL copies inside the
/// registry removal transactions) never touch the counter — they are
/// removals (coverage downward, above); the mem shape's
/// [`LfsRows::remove_repo`] leg rides this surface and follows the
/// same discipline.
pub struct VersionedLfsRows<T: ?Sized> {
    inner: Arc<T>,
    version: Arc<AtomicU64>,
}

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

    /// The wrapped rows store.
    pub fn inner(&self) -> &Arc<T> {
        &self.inner
    }

    /// The counter's current value (the liveness source's arbitration
    /// epoch signal).
    pub fn version(&self) -> u64 {
        self.version.load(Ordering::Relaxed)
    }
}

#[async_trait]
impl<T> LfsRows for VersionedLfsRows<T>
where
    T: LfsRows + ?Sized,
{
    async fn assert(&self, repo_id: &str, oid: &str, size: u64) -> Result<(), StorageError> {
        self.inner.assert(repo_id, oid, size).await?;
        self.version.fetch_add(1, Ordering::Relaxed);
        Ok(())
    }

    async fn remove(&self, repo_id: &str, oid: &str) -> Result<(), StorageError> {
        self.inner.remove(repo_id, oid).await
    }

    async fn remove_repo(&self, repo_id: &str) -> Result<(), StorageError> {
        self.inner.remove_repo(repo_id).await
    }

    async fn rows(&self) -> Result<Vec<LfsRow>, StorageError> {
        self.inner.rows().await
    }

    async fn repo_rows(&self, repo_id: &str) -> Result<Vec<LfsRow>, StorageError> {
        self.inner.repo_rows(repo_id).await
    }
}

/// The pg twin's battery — the same script the reference and the
/// sqlite twin run, over the live-pg harness (the
/// `ALKBLOBS_PG_*`/`ALKGIT_SUBSTRATE_PG_*` convention; clean skip
/// when the host env is absent, so `--all-features` stays green
/// everywhere). Works in a unique, freshly-migrated schema and drops
/// it after.
#[cfg(all(test, feature = "postgres"))]
mod pg_twin_tests {
    use std::sync::Arc;

    use crate::postgres::live_harness::{drop_schema, live_pg, unique_schema};

    use super::battery;
    use super::*;

    /// The battery over the pg row twin — the semantics pinned on the
    /// reference impl, re-run engine-swapped.
    #[tokio::test]
    async fn pg_rows_twin_carries_the_battery() {
        let dsn = live_pg!();
        let schema = unique_schema("lfs-battery");
        let table = PgLfsRows::new(
            crate::postgres::CoTenantPg::open(&dsn, &schema)
                .await
                .unwrap(),
        );

        battery::assert_and_the_refcount_aware_upsert(&table)
            .await
            .unwrap();
        battery::cross_reference_reprotection(&table).await.unwrap();
        battery::removal_of_an_absent_row_is_a_noop(&table)
            .await
            .unwrap();
        battery::repo_deletion_takes_the_rows_and_recounts(&table)
            .await
            .unwrap();
        battery::the_reads_list_canonically(&table).await.unwrap();
        battery::validation_rejects_as_storage_invalid(&table)
            .await
            .unwrap();

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

    /// The advisory lock serializes the concurrent same-oid
    /// assertions from two co-tenant connections — every row ends at
    /// the exact count (the fleet-shape convergence).
    #[tokio::test]
    async fn pg_concurrent_asserts_converge_on_exact_counts() {
        let dsn = live_pg!();
        let schema = unique_schema("lfs-race");
        let oid = battery::oid_hex(2);
        let mut joins = Vec::new();
        for index in 0..4 {
            let table = Arc::new(PgLfsRows::new(
                crate::postgres::CoTenantPg::open(&dsn, &schema)
                    .await
                    .unwrap(),
            ));
            let oid = oid.clone();
            joins.push(tokio::spawn(async move {
                table
                    .assert(&format!("repo/{index}"), &oid, 64)
                    .await
                    .expect("the assert lands")
            }));
        }
        for join in joins {
            join.await.unwrap();
        }

        let table = PgLfsRows::new(
            crate::postgres::CoTenantPg::open(&dsn, &schema)
                .await
                .unwrap(),
        );
        for row in table.rows().await.unwrap() {
            assert_eq!(
                row.refcount, 4,
                "every row carries the exact count: {row:?}"
            );
        }

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