alkgit/substrate/src/postgres.rs

//! The postgres registry twin — the family adapter pattern's
//! fleet/durable-shared engine assembly (substrate.md §assembly
//! shapes; ADR-019 §3: "the postgres twin (`postgres` feature)") — the
//! `w6-registry-postgres` landing of POC-R's pg leg. The machinery
//! this module shares with the ADR-024 home — the [`CoTenantPg`]
//! co-tenant connection handle, the pg migration runner, and the pg
//! SQL plumbing ([`DEFAULT_SCHEMA`], `crate::migrations::qualify`, the
//! advisory-lock scope keys) — lives in [`crate::migrations`] beside
//! the sqlite twin's (the settled, engine-symmetric placement); this
//! module carries the twin itself plus its options/assembly.
//!
//! - [`PgRegistry`] — the adapter twin: the durable write side
//!   ([`RegistryWrite`]) over the co-tenant `git_repos` rows — the
//!   identical shape the sqlite twin carries. The `registry_seq`
//!   singleton bump rides **inside the row-write transaction**
//!   (transactionally exact — the watermark's anchor, ADR-019 §5;
//!   failed mutations consume none), and the row commits carry the
//!   refs-table legs the composition owes: **put seeds
//!   `HEAD → refs/heads/main`**
//!   ([`crate::registry::HEAD_SEED_TARGET`]; `remove` deletes the
//!   record row + the repo's ref rows + the repo's mutation-marker
//!   rows). Plus `PgRegistry::record_mutation` — the per-push
//!   mutation-marker write (ADR-019 §7, composition-internal: bumps
//!   the seq/watermark legs without touching the record). Concurrency
//!   model: every mutation takes the `registry_seq` singleton's row
//!   lock first (the `UPDATE … RETURNING` bump), then
//!   existence-reads committed state — concurrent clashing mutations
//!   serialize and map to the closed error set
//!   ([`RegistryError::AlreadyExists`] / [`RegistryError::NotFound`]),
//!   never a raw unique-violation.
//! - [`PgRegistryAssembly`] — the adapter constructor: one fallible
//!   assembly call (loud — ADR-024 §1) that opens the co-tenant
//!   connection + its migrations, opens the alkstore-postgres engine,
//!   loads the cached read provider's initial snapshot, and wires the
//!   reactive halves ([`Reactor`]) plus the announced write side
//!   ([`AnnouncingRegistry`]) and the
//!   [`RegistryBackend::with_cached_provider`] binding over the
//!   announced side — the sqlite twin's composition, engine-swapped.
//!   Post-commit announce wired by default (ADR-019 §5). The wave-8
//!   single composition entry unifies construction; this is the
//!   per-piece constructor (the recorded interpretation).
//!
//! # The LISTEN/NOTIFY leg — load-bearing on postgres
//!
//! The reactive composition is engine-generic: the announce's notify
//! leg fires the wake channel through the engine handle (the pg
//! engine's forwarder delivers it over its dedicated LISTEN
//! connection), and the wake runner listens the same way. **No
//! file-level bump exists on postgres** — a lost/failed hint leaves
//! caches stale until the reconcile tick: **the tick's cadence is the
//! real staleness bound on postgres** (substrate.md §the reactive
//! halves' designed-in asymmetry, carried by the composition and the
//! docs). The asymmetry probe is test-pinned here: a notify-free
//! write on the co-tenant connection wakes nothing (the provider
//! stays stale past a wake-sized window), the reconcile sweep closes
//! the gap, and the announced write's NOTIFY demonstrably delivers
//! (the leg is live when wired).
//!
//! # The live-pg harness (the alkblobs convention)
//!
//! The tests run against the family harness container
//! (localhost:15432 — env-gated, skip with a clear message when the
//! harness is absent; the alkblobs `tests/readme_examples.rs` skip
//! posture). The env names pinned here: `ALKGIT_SUBSTRATE_PG_` +
//! `HOST`/`PORT`/`USER`/`PASSWORD`/`DB`, each falling back to the
//! same-named `ALKBLOBS_PG_*` harness variable (one container, one
//! convention), then to the harness defaults
//! (`15432`/`postgres`/`poc`/`blobs`). The harness host's absence
//! means `--all-features` keeps green everywhere (a clean skip).
//! Every live test works in a unique, freshly-migrated PostgreSQL
//! schema (process + tag keyed; the engine in its own unique schema
//! beside it) and drops both after — the harness database stays
//! untouched. The tests serialize on one harness gate: the
//! `LISTEN/NOTIFY` wake channel is database-wide, and a probe that
//! asserts silence must not read a sibling test's announce.

#![allow(clippy::too_many_lines)]

use std::sync::Arc;

use alkgit::{RegistryError, RepoRecord};
use async_trait::async_trait;

use crate::error::OpenError;
use crate::migrations::{git_refs_lock_key, lfs_objects_lock_key, qualify};
use crate::reactor::{AnnouncingRegistry, Reactor, RegistryEvent};
use crate::registry::{
    row_to_record, validate_record, visibility_str, CachedRegistryProvider, RegistryBackend,
    RegistryRow, RegistryWrite, HEAD_SEED_TARGET,
};

pub use crate::migrations::{CoTenantPg, DEFAULT_SCHEMA};

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

/// The postgres twin — the family pair's durable write side
/// ([`RegistryWrite`]) over the co-tenant `git_repos` rows. Reads
/// ride the cached provider in the composed assemblies — the hot
/// path never touches this side except the refresh reads.
///
/// Natively async — every method is a straight round trip on the
/// co-tenant handle's client (no blocking bridge; the sqlite twin's
/// `spawn_blocking` model is the *other* engine's, and the POC-R pg
/// leg's shape). Construction takes the already-migrated handle
/// ([`CoTenantPg::open`] runs the migrations) — no construction I/O
/// here.
pub struct PgRegistry {
    handle: CoTenantPg,
}

impl PgRegistry {
    /// Mount the twin over a co-tenant handle.
    pub fn new(handle: CoTenantPg) -> Self {
        PgRegistry { handle }
    }

    /// The marker write rides its own transaction
    /// (`durable_record_mutation`): the marker row upserted with a
    /// `registry_seq` bump inside one transaction — the watermark's
    /// anchor moves on every push while the record row and its schema
    /// stay untouched (`rows_since` never carries markers; the seq is
    /// the reconcile watermark's currency). The record must exist (the
    /// receive flow resolved it) — a removed repo's marker is refused
    /// ([`RegistryError::NotFound`]). Consuming composition: the
    /// assembly's fused mutation legs ([`crate::pins::RegistryMutations`]
    /// — the receive flow's post-CAS legs), which announce the write
    /// post-commit.
    pub(crate) async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let mut guard = self.handle.lock().await;
        durable_record_mutation(&mut guard, self.handle.schema(), repo_id).await
    }
}

#[async_trait]
impl RegistryWrite for PgRegistry {
    async fn put(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
        validate_record(record)?;
        let grants = serde_json::to_string(&record.grants).map_err(|_| RegistryError::Invalid)?;
        let mut guard = self.handle.lock().await;
        durable_put(&mut guard, self.handle.schema(), record, &grants).await
    }

    async fn update(&self, record: &RepoRecord) -> Result<i64, RegistryError> {
        validate_record(record)?;
        let grants = serde_json::to_string(&record.grants).map_err(|_| RegistryError::Invalid)?;
        let mut guard = self.handle.lock().await;
        durable_update(&mut guard, self.handle.schema(), record, &grants).await
    }

    async fn remove(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let mut guard = self.handle.lock().await;
        durable_remove(&mut guard, self.handle.schema(), repo_id).await
    }

    async fn watermark(&self) -> Result<i64, RegistryError> {
        let guard = self.handle.lock().await;
        durable_watermark(&guard, self.handle.schema()).await
    }

    async fn rows_since(&self, seq: i64) -> Result<Vec<RegistryRow>, RegistryError> {
        let guard = self.handle.lock().await;
        durable_rows_since(&guard, self.handle.schema(), seq).await
    }
}

fn seq_bump_sql(schema: &str) -> String {
    format!(
        "UPDATE {} SET value = value + 1 WHERE id = 1 RETURNING value",
        qualify(schema, "registry_seq")
    )
}

/// The singleton bump's `RETURNING` row → the integer.
fn seq_of(row: &tokio_postgres::Row, context: &str) -> Result<i64, RegistryError> {
    row.try_get(0).map_err(|e| io_err(context, e))
}

/// The `put`'s transaction: the `registry_seq` singleton's row lock
/// taken first (the writers' serialization point — the
/// `BEGIN IMMEDIATE` analog), then the existence check (a taken id is
/// [`RegistryError::AlreadyExists`], no seq consumed), the record row,
/// and the HEAD seed row — commit or roll back whole. The HEAD seed
/// rides the `git_refs` advisory lock the refs CAS twins hold for
/// their whole call ([`crate::refs::PgRefTable`]'s
/// `{schema}\0git_refs` scope) — every transaction touching the
/// refs table serializes with the CAS storm, the same whole-fleet
/// tradeoff the removal's `lfs_objects` leg accepts (rare
/// construction legs queueing behind the storm, not the storm behind
/// them).
async fn durable_put(
    client: &mut tokio_postgres::Client,
    schema: &str,
    record: &RepoRecord,
    grants: &str,
) -> Result<i64, RegistryError> {
    let bump = seq_bump_sql(schema);
    let exists_sql = format!(
        "SELECT true FROM {} WHERE repo_id = $1",
        qualify(schema, "git_repos")
    );
    let insert_sql = format!(
        "INSERT INTO {} (repo_id, visibility, grants, seq) VALUES ($1, $2, $3, $4)",
        qualify(schema, "git_repos")
    );
    let seed_sql = format!(
        "INSERT INTO {} (repo_id, name, oid_hex, symref_target) VALUES ($1, 'HEAD', NULL, $2)",
        qualify(schema, "git_refs")
    );
    let tx = client
        .transaction()
        .await
        .map_err(|e| io_err("registry put BEGIN", e))?;
    tx.execute(
        "SELECT pg_advisory_xact_lock($1)",
        &[&git_refs_lock_key(schema)],
    )
    .await
    .map_err(|e| io_err("registry put git_refs advisory lock", e))?;
    let seq: i64 = tx
        .query_one(&bump, &[])
        .await
        .map_err(|e| io_err("registry put seq bump", e))
        .and_then(|row| seq_of(&row, "registry put seq bump"))?;
    let taken = tx
        .query_opt(&exists_sql, &[&record.repo_id])
        .await
        .map_err(|e| io_err("registry put existence read", e))?;
    if taken.is_some() {
        tx.rollback().await.ok();
        return Err(RegistryError::AlreadyExists);
    }
    let visibility = visibility_str(record.visibility);
    tx.execute(&insert_sql, &[&record.repo_id, &visibility, &grants, &seq])
        .await
        .map_err(|e| io_err("registry put record row", e))?;
    tx.execute(&seed_sql, &[&record.repo_id, &HEAD_SEED_TARGET])
        .await
        .map_err(|e| io_err("registry put HEAD seed", e))?;
    tx.commit()
        .await
        .map_err(|e| io_err("registry put commit", e))?;
    Ok(seq)
}

/// The `update`'s transaction: the seq-row lock first, existence
/// ([`RegistryError::NotFound`]), the record row's replace. The
/// refs-table rows are untouched (HEAD belongs to the record's
/// creation).
async fn durable_update(
    client: &mut tokio_postgres::Client,
    schema: &str,
    record: &RepoRecord,
    grants: &str,
) -> Result<i64, RegistryError> {
    let bump = seq_bump_sql(schema);
    let exists_sql = format!(
        "SELECT true FROM {} WHERE repo_id = $1",
        qualify(schema, "git_repos")
    );
    let update_sql = format!(
        "UPDATE {} SET visibility = $2, grants = $3, seq = $4 WHERE repo_id = $1",
        qualify(schema, "git_repos")
    );
    let tx = client
        .transaction()
        .await
        .map_err(|e| io_err("registry update BEGIN", e))?;
    let seq: i64 = tx
        .query_one(&bump, &[])
        .await
        .map_err(|e| io_err("registry update seq bump", e))
        .and_then(|row| seq_of(&row, "registry update seq bump"))?;
    let exists = tx
        .query_opt(&exists_sql, &[&record.repo_id])
        .await
        .map_err(|e| io_err("registry update existence read", e))?;
    if exists.is_none() {
        tx.rollback().await.ok();
        return Err(RegistryError::NotFound);
    }
    let visibility = visibility_str(record.visibility);
    tx.execute(&update_sql, &[&record.repo_id, &visibility, &grants, &seq])
        .await
        .map_err(|e| io_err("registry update record row", e))?;
    tx.commit()
        .await
        .map_err(|e| io_err("registry update commit", e))?;
    Ok(seq)
}

/// The `remove`'s transaction: the seq-row lock first, existence
/// ([`RegistryError::NotFound`]), then the record row, the repo's ref
/// rows (the HEAD seed included), the repo's mutation-marker rows,
/// and the repo's LFS liveness rows — the whole repo's co-tenant
/// state in one removal. The `lfs_objects` leg is unconditional (the
/// table ships unconditionally in the step set; the rows only exist
/// when the `lfs` surface wrote them, and a mixed deployment that
/// turned the feature off must not leak), followed by the shared
/// oids' refcount recompute (`substrate/src/lfs.rs`'s semantics —
/// every surviving row of an oid reads the same count at commit).
/// The whole transaction holds the row-store advisory lock the LFS
/// twins take ([`crate::lfs::PgLfsRows`]'s
/// `{schema}\0lfs_objects` scope) — the recount serializes against
/// cross-connection same-oid asserts/removals, the twins' storm
/// discipline including the removal leg. It holds the refs-table
/// advisory lock the same way ([`crate::refs::PgRefTable`]'s
/// `{schema}\0git_refs` scope): the refs delete leg serializes with
/// the CAS storm, whose in-flight transactions otherwise could commit
/// decided writes after the removal's delete — orphan rows the
/// removal's contract ("the whole repo's co-tenant state in one
/// removal") exists to prevent. Objects and packs are not
/// touched (deletion is a GC consequence).
async fn durable_remove(
    client: &mut tokio_postgres::Client,
    schema: &str,
    repo_id: &str,
) -> Result<i64, RegistryError> {
    let bump = seq_bump_sql(schema);
    let exists_sql = format!(
        "SELECT true FROM {} WHERE repo_id = $1",
        qualify(schema, "git_repos")
    );
    let delete_record = format!(
        "DELETE FROM {} WHERE repo_id = $1",
        qualify(schema, "git_repos")
    );
    let delete_refs = format!(
        "DELETE FROM {} WHERE repo_id = $1",
        qualify(schema, "git_refs")
    );
    let delete_markers = format!(
        "DELETE FROM {} WHERE repo_id = $1",
        qualify(schema, "repo_mutation_markers")
    );
    let delete_lfs = format!(
        "DELETE FROM {} WHERE repo_id = $1",
        qualify(schema, "lfs_objects")
    );
    let recount_lfs = format!(
        "UPDATE {} AS t0 \
         SET refcount = (SELECT count(*) FROM {} c WHERE c.oid = t0.oid)",
        qualify(schema, "lfs_objects"),
        qualify(schema, "lfs_objects")
    );
    let tx = client
        .transaction()
        .await
        .map_err(|e| io_err("registry remove BEGIN", e))?;
    tx.execute(
        "SELECT pg_advisory_xact_lock($1)",
        &[&lfs_objects_lock_key(schema)],
    )
    .await
    .map_err(|e| io_err("registry remove lfs advisory lock", e))?;
    tx.execute(
        "SELECT pg_advisory_xact_lock($1)",
        &[&git_refs_lock_key(schema)],
    )
    .await
    .map_err(|e| io_err("registry remove git_refs advisory lock", e))?;
    let seq: i64 = tx
        .query_one(&bump, &[])
        .await
        .map_err(|e| io_err("registry remove seq bump", e))
        .and_then(|row| seq_of(&row, "registry remove seq bump"))?;
    let exists = tx
        .query_opt(&exists_sql, &[&repo_id])
        .await
        .map_err(|e| io_err("registry remove existence read", e))?;
    if exists.is_none() {
        tx.rollback().await.ok();
        return Err(RegistryError::NotFound);
    }
    tx.execute(&delete_record, &[&repo_id])
        .await
        .map_err(|e| io_err("registry remove record row", e))?;
    tx.execute(&delete_refs, &[&repo_id])
        .await
        .map_err(|e| io_err("registry remove ref rows", e))?;
    tx.execute(&delete_markers, &[&repo_id])
        .await
        .map_err(|e| io_err("registry remove marker rows", e))?;
    tx.execute(&delete_lfs, &[&repo_id])
        .await
        .map_err(|e| io_err("registry remove lfs rows", e))?;
    tx.execute(&recount_lfs, &[])
        .await
        .map_err(|e| io_err("registry remove lfs recount", e))?;
    tx.commit()
        .await
        .map_err(|e| io_err("registry remove commit", e))?;
    Ok(seq)
}

async fn durable_watermark(
    client: &tokio_postgres::Client,
    schema: &str,
) -> Result<i64, RegistryError> {
    let sql = format!(
        "SELECT value FROM {} WHERE id = 1",
        qualify(schema, "registry_seq")
    );
    client
        .query_one(&sql, &[])
        .await
        .map_err(|e| io_err("registry watermark read", e))?
        .try_get(0)
        .map_err(|e| io_err("registry watermark read", e))
}

async fn durable_rows_since(
    client: &tokio_postgres::Client,
    schema: &str,
    seq: i64,
) -> Result<Vec<RegistryRow>, RegistryError> {
    let sql = format!(
        "SELECT repo_id, visibility, grants, seq FROM {} WHERE seq > $1 ORDER BY seq",
        qualify(schema, "git_repos")
    );
    let rows = client
        .query(&sql, &[&seq])
        .await
        .map_err(|e| io_err("registry rows_since query", e))?;
    let mut since = Vec::with_capacity(rows.len());
    for row in &rows {
        let repo_id: String = row
            .try_get(0)
            .map_err(|e| io_err("registry rows_since row", e))?;
        let visibility: String = row
            .try_get(1)
            .map_err(|e| io_err("registry rows_since row", e))?;
        let grants: String = row
            .try_get(2)
            .map_err(|e| io_err("registry rows_since row", e))?;
        let row_seq: i64 = row
            .try_get(3)
            .map_err(|e| io_err("registry rows_since row", e))?;
        since.push(row_to_record(&repo_id, &visibility, &grants, row_seq)?);
    }
    Ok(since)
}

/// The marker row upsert inside its one transaction: the seq-row lock
/// first, existence (the receive flow resolved the record), the
/// marker row.
async fn durable_record_mutation(
    client: &mut tokio_postgres::Client,
    schema: &str,
    repo_id: &str,
) -> Result<i64, RegistryError> {
    let bump = seq_bump_sql(schema);
    let exists_sql = format!(
        "SELECT true FROM {} WHERE repo_id = $1",
        qualify(schema, "git_repos")
    );
    let upsert_sql = format!(
        "INSERT INTO {} (repo_id, seq) VALUES ($1, $2) \
         ON CONFLICT (repo_id) DO UPDATE SET seq = excluded.seq",
        qualify(schema, "repo_mutation_markers")
    );
    let tx = client
        .transaction()
        .await
        .map_err(|e| io_err("mutation marker BEGIN", e))?;
    let seq: i64 = tx
        .query_one(&bump, &[])
        .await
        .map_err(|e| io_err("mutation marker seq bump", e))
        .and_then(|row| seq_of(&row, "mutation marker seq bump"))?;
    let exists = tx
        .query_opt(&exists_sql, &[&repo_id])
        .await
        .map_err(|e| io_err("mutation marker existence read", e))?;
    if exists.is_none() {
        tx.rollback().await.ok();
        return Err(RegistryError::NotFound);
    }
    tx.execute(&upsert_sql, &[&repo_id, &seq])
        .await
        .map_err(|e| io_err("mutation marker row", e))?;
    tx.commit()
        .await
        .map_err(|e| io_err("mutation marker commit", e))?;
    Ok(seq)
}

/// The postgres twin's engine options — the assembly constructor's
/// input ([`PgRegistryAssembly::open`]): the co-tenant schema the
/// alkgit tables live in and the alkstore-postgres engine's own
/// options.
///
/// The connection settings are not part of this struct — they are the
/// DSN string both opens share (the engine's `open(url, PgOpts)` pin;
/// engine-configuration concerns never touch the contract surface —
/// ADR-008 §6, the family posture).
#[derive(Clone, Debug)]
pub struct PgRegistryOptions {
    /// The PostgreSQL schema the alkgit co-tenant tables (and the
    /// stamp table) live in. Empty falls back to [`DEFAULT_SCHEMA`].
    pub schema: String,
    /// The alkstore-postgres engine's options (its schema, pool size,
    /// durability knob).
    pub engine: alkstore_postgres::PgOpts,
    /// The alkblobs postgres kv engine's options (the composition
    /// entry's object-store half rides them — the fleet/durable-shared
    /// assembly's object pool).
    pub blobs: alkblobs::PgOpts,
}

impl Default for PgRegistryOptions {
    fn default() -> Self {
        PgRegistryOptions {
            schema: DEFAULT_SCHEMA.to_string(),
            engine: alkstore_postgres::PgOpts::default(),
            blobs: alkblobs::PgOpts::default(),
        }
    }
}

/// The postgres twin assembly — the pieces one adapter constructor
/// wires over the fleet/durable-shared engine assembly (substrate.md
/// §assembly shapes: `PgStore`/`PgOpts` schema-namespaced + the pg
/// twin; **the registry's own connection is the fourth pool** of the
/// assembly's connection budget):
///
/// - [`PgRegistryAssembly::store`] — the opened alkstore engine handle
///   (the reactive halves' substrate; the LISTEN/NOTIFY leg rides its
///   dedicated listener connection).
/// - [`PgRegistryAssembly::write`] — the **announced write side**
///   ([`AnnouncingRegistry`]): the pair with the post-commit announce
///   wired by default (ADR-019 §5 — no non-reactive shipped path).
/// - [`PgRegistryAssembly::durable`] — the bare twin (announce free —
///   the probe legs run here; the asymmetry probe's notify-free
///   write).
/// - [`PgRegistryAssembly::provider`] — the cached read provider,
///   loaded once from the durable side at construction (loud — a
///   registry serving an empty snapshot over a populated table would
///   contradict the definite-miss read contract).
/// - [`PgRegistryAssembly::reactor`] — the wake runner/stream
///   consumer/reconcile machinery, holding the durable side.
/// - [`PgRegistryAssembly::backend`] — the core-traits binding
///   ([`RegistryBackend::with_cached_provider`]) over the announced
///   write side: the store methods announce post-commit and write
///   through.
/// - [`PgRegistryAssembly::lfs_rows`] — the LFS pointer-lifecycle
///   row CAS over the same co-tenant handle, write-counter-wrapped
///   ([`VersionedLfsRows`](crate::lfs::VersionedLfsRows) — the
///   refs-twins' wrap posture; the composition entry composes it
///   unchanged) — the counter [`PgRegistryAssembly::lfs_version`]
///   exposes is the LFS liveness source's arbitration-window signal.
///
/// The refs row-CAS twin's pg side is
/// [`crate::refs::PgRefTable`] (the refs workstream's file — the
/// same CAS core the sqlite twin runs, engine-swapped; the wave-7
/// N-c rider) on the same co-tenant handle type, wrapped in the
/// write-counter shape the liveness source consults
/// ([`Self::refs`] / [`Self::refs_version`] — the sqlite assembly's
/// accessor shapes mirrored; the wave-8 composition entry consumers).
pub struct PgRegistryAssembly {
    store: Arc<dyn alkstore::Store>,
    durable: Arc<PgRegistry>,
    registry: Arc<dyn RegistryWrite>,
    provider: Arc<CachedRegistryProvider>,
    reactor: Arc<Reactor>,
    backend: Arc<RegistryBackend>,
    refs: Arc<crate::refs::VersionedRefTable<crate::refs::PgRefTable>>,
    refs_version: Arc<std::sync::atomic::AtomicU64>,
    #[cfg(feature = "lfs")]
    lfs: Arc<crate::lfs::VersionedLfsRows<crate::lfs::PgLfsRows>>,
    #[cfg(feature = "lfs")]
    lfs_version: Arc<std::sync::atomic::AtomicU64>,
}

impl PgRegistryAssembly {
    /// The adapter constructor: open the co-tenant connection at
    /// `config` (migrations at construction, loud — a half-migrated
    /// store refuses service), open the alkstore-postgres engine
    /// (its own schema-namespaced bootstrap + listener), load the
    /// provider's initial snapshot, and wire the reactive
    /// composition. Every failure is [`OpenError`].
    pub async fn open(config: &str, opts: PgRegistryOptions) -> Result<Self, OpenError> {
        let handle = CoTenantPg::open(config, &opts.schema).await?;
        let store: Arc<dyn alkstore::Store> = Arc::from(
            alkstore_postgres::open(config, opts.engine)
                .await
                .map_err(|e| OpenError::Engine(format!("alkstore-postgres open: {e}")))?,
        );
        let refs_handle = handle.clone();
        let durable: Arc<PgRegistry> = Arc::new(PgRegistry::new(handle));

        let provider = Arc::new(CachedRegistryProvider::new());
        let rows = durable
            .rows_since(0)
            .await
            .map_err(|e| OpenError::Engine(format!("cached provider initial load: {e}")))?;
        provider.install(rows);

        let durable_side: Arc<dyn RegistryWrite> = durable.clone();
        let reactor = Arc::new(Reactor::new(
            store.clone(),
            durable_side.clone(),
            provider.clone(),
        ));
        let registry: Arc<dyn RegistryWrite> = Arc::new(AnnouncingRegistry::new(
            durable_side.clone(),
            reactor.clone(),
        ));
        let backend = Arc::new(RegistryBackend::with_cached_provider(
            registry.clone(),
            provider.clone(),
        ));
        #[cfg(feature = "lfs")]
        let lfs_version = Arc::new(std::sync::atomic::AtomicU64::new(0));
        #[cfg(feature = "lfs")]
        let lfs = Arc::new(crate::lfs::VersionedLfsRows::new(
            Arc::new(crate::lfs::PgLfsRows::new(refs_handle.clone())),
            lfs_version.clone(),
        ));
        let refs_version = Arc::new(std::sync::atomic::AtomicU64::new(0));
        let refs = Arc::new(crate::refs::VersionedRefTable::new(
            Arc::new(crate::refs::PgRefTable::new(refs_handle)),
            refs_version.clone(),
        ));
        Ok(PgRegistryAssembly {
            store,
            durable,
            registry,
            provider,
            reactor,
            backend,
            refs,
            refs_version,
            #[cfg(feature = "lfs")]
            lfs,
            #[cfg(feature = "lfs")]
            lfs_version,
        })
    }

    /// The opened alkstore engine handle (the reactive halves'
    /// substrate; the LISTEN/NOTIFY leg rides its listener
    /// connection — the wave-7 maintenance chain rides the same).
    pub fn store(&self) -> &Arc<dyn alkstore::Store> {
        &self.store
    }

    /// The announced write side — the pair with the post-commit
    /// announce wired (the write surface the composition's callers
    /// use).
    pub fn write(&self) -> &Arc<dyn RegistryWrite> {
        &self.registry
    }

    /// The bare twin (announce-free — the transactional legs' probe
    /// surface).
    pub fn durable(&self) -> &Arc<PgRegistry> {
        &self.durable
    }

    /// The cached read provider (the hot read).
    pub fn provider(&self) -> &Arc<CachedRegistryProvider> {
        &self.provider
    }

    /// The reactor (the wake runner / stream consumer / reconcile
    /// sweep run through it).
    pub fn reactor(&self) -> &Arc<Reactor> {
        &self.reactor
    }

    /// The core-traits binding (the engine-twin binding announced and
    /// write-through wired).
    pub fn backend(&self) -> &Arc<RegistryBackend> {
        &self.backend
    }

    /// The refs row CAS over the same co-tenant handle — the
    /// write-counter-wrapped composition refs ([`crate::refs::VersionedRefTable`]:
    /// the CAS semantics are unchanged (verbatim delegation), every
    /// CAS call bumps the shared counter the liveness source's
    /// arbitration consult reads (`crate::liveness` — the
    /// composition's invalidation half). The wave-7 twin's record: the
    /// refs twin's fleet wiring is the composition entry's — this
    /// accessor is the legs it wires (the wave-8 entry consumes these
    /// pieces, the sqlite assembly's accessor shapes mirrored).
    pub fn refs(&self) -> &Arc<crate::refs::VersionedRefTable<crate::refs::PgRefTable>> {
        &self.refs
    }

    /// The refs writes' shared counter — the same handle
    /// [`Self::refs`] bumps and the `crate::liveness::RefClosure`
    /// source consults; hand it to the closure's constructor.
    pub fn refs_version(&self) -> &Arc<std::sync::atomic::AtomicU64> {
        &self.refs_version
    }

    /// The LFS pointer-lifecycle row CAS over the same co-tenant
    /// handle — the write-counter-wrapped rows
    /// ([`crate::lfs::VersionedLfsRows`]; the refs twins' wrap posture,
    /// the per-piece parity posture the per-piece-path embedder
    /// composes): the row semantics are unchanged (verbatim
    /// delegation), every `assert` bumps the shared counter the LFS
    /// liveness source's arbitration consult reads (`crate::liveness`
    /// — the composition's invalidation half). The record-removal leg
    /// takes a removed repo's rows with the record transactionally
    /// (the twins' unconditional in-SQL copy).
    #[cfg(feature = "lfs")]
    pub fn lfs_rows(&self) -> &Arc<crate::lfs::VersionedLfsRows<crate::lfs::PgLfsRows>> {
        &self.lfs
    }

    /// The LFS rows' asserts' shared counter — the same handle
    /// [`Self::lfs_rows`] bumps and the
    /// [`crate::liveness::LfsMedia`] source consults; hand it to the
    /// media source's constructor.
    #[cfg(feature = "lfs")]
    pub fn lfs_version(&self) -> &Arc<std::sync::atomic::AtomicU64> {
        &self.lfs_version
    }
}

/// The per-push mutation-marker write, announced
/// ([`RegistryEvent::Mutation`] post-commit, fail-soft) — the receive
/// flow's watermark leg (ADR-019 §7; the composition-internal
/// [`RegistryMutations`](crate::pins::RegistryMutations) legs the
/// wrapped refs implement rides — [`crate::pins::CommittedRefs`]; the
/// record schema untouched). The durable leg carries no marker-driven
/// record change: `rows_since` never carries markers.
#[async_trait]
impl crate::pins::RegistryMutations for PgRegistryAssembly {
    async fn record_mutation(&self, repo_id: &str) -> Result<i64, RegistryError> {
        let seq = self.durable.record_mutation(repo_id).await?;
        self.reactor
            .announce(&RegistryEvent::Mutation {
                seq,
                repo_id: repo_id.to_owned(),
            })
            .await;
        Ok(seq)
    }
}

/// The live-pg test harness, shared by this crate's pg-gated test
/// modules (this module's suite, the refs twin's battery in
/// `crate::refs`) — the alkblobs convention's env contract, one
/// container: `ALKGIT_SUBSTRATE_PG_*` per field, falling back to the
/// same-named `ALKBLOBS_PG_*` harness variable, then to the harness
/// defaults (`15432`/`postgres`/`poc`/`blobs`). The harness host's
/// absence is the clean-skip signal (`--all-features` stays green
/// everywhere). Not runtime API — cfg(test) only.
#[cfg(all(test, feature = "postgres"))]
pub(crate) mod live_harness {
    use alkstore_postgres::quote_identifier;
    use tokio_postgres::NoTls;

    /// The harness DSN. The harness host's absence is the clean-skip
    /// signal.
    pub(crate) fn harness_dsn() -> Option<String> {
        let pick = |alkgit_key: &str, alkblobs_key: &str, default: &str| {
            std::env::var(alkgit_key)
                .ok()
                .filter(|v| !v.is_empty())
                .or_else(|| std::env::var(alkblobs_key).ok().filter(|v| !v.is_empty()))
                .unwrap_or_else(|| default.to_owned())
        };
        let host = std::env::var("ALKGIT_SUBSTRATE_PG_HOST")
            .ok()
            .filter(|v| !v.is_empty())
            .or_else(|| {
                std::env::var("ALKBLOBS_PG_HOST")
                    .ok()
                    .filter(|v| !v.is_empty())
            })?;
        let port = pick("ALKGIT_SUBSTRATE_PG_PORT", "ALKBLOBS_PG_PORT", "15432");
        let user = pick("ALKGIT_SUBSTRATE_PG_USER", "ALKBLOBS_PG_USER", "postgres");
        let password = pick(
            "ALKGIT_SUBSTRATE_PG_PASSWORD",
            "ALKBLOBS_PG_PASSWORD",
            "poc",
        );
        let db = pick("ALKGIT_SUBSTRATE_PG_DB", "ALKBLOBS_PG_DB", "blobs");
        Some(format!(
            "host={host} port={port} user={user} password={password} dbname={db}"
        ))
    }

    /// The skip or nothing — the alkblobs convention's skip posture
    /// (a clear message, a passing test).
    macro_rules! live_pg {
        () => {
            match $crate::postgres::live_harness::harness_dsn() {
                Some(dsn) => dsn,
                None => {
                    eprintln!(
                        "skipping (no postgres harness server): set \
                         ALKGIT_SUBSTRATE_PG_HOST (or the ALKBLOBS_PG_* \
                         harness env)"
                    );
                    return;
                }
            }
        };
    }
    pub(crate) use live_pg;

    pub(crate) fn unique_schema(tag: &str) -> String {
        static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        format!("alkgit_substrate_pg_{}_{}_{}", std::process::id(), n, tag)
    }

    pub(crate) async fn drop_schema(dsn: &str, schema: &str) {
        if let Ok((client, conn)) = tokio_postgres::connect(dsn, NoTls).await {
            tokio::spawn(async move {
                let _ = conn.await;
            });
            let _ = client
                .batch_execute(&format!(
                    "DROP SCHEMA IF EXISTS {} CASCADE",
                    quote_identifier(schema)
                ))
                .await;
        }
    }

    pub(crate) async fn connect_direct(dsn: &str) -> tokio_postgres::Client {
        let (client, conn) = tokio_postgres::connect(dsn, NoTls).await.unwrap();
        tokio::spawn(async move {
            let _ = conn.await;
        });
        client
    }

    /// The live tests' one gate, shared crate-wide: the `LISTEN/NOTIFY`
    /// wake channel is database-wide on the harness container, so any
    /// test doing announced writes (the registry's put/marker legs fire
    /// the wake channel) must serialize against any test that asserts
    /// silence (the asymmetry probe) — not only the tests inside the
    /// same module.
    pub(crate) static LIVE_GATE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use alkgit::AccessAction;

    use crate::reactor::{AppliedWatermark, MAINTENANCE_QUEUE, WAKE_CHANNEL};
    use crate::registry::RegistryRead;

    use alkgit::{GitRegistry, GitRegistryStore};
    use alkstore::{QueueOpts, StreamEvent};

    use super::*;

    use super::live_harness::{connect_direct, drop_schema, unique_schema, LIVE_GATE};
    use crate::migrations::pg_twin::{build_steps, STAMP_TABLE, STEP_VERSIONS};
    use crate::postgres::live_harness::live_pg;
    use crate::testutil::{private_record, record, spawn_wake_runner, stop_and_join, wait_until};

    fn assembly_opts(schema: &str) -> PgRegistryOptions {
        PgRegistryOptions {
            schema: schema.to_owned(),
            engine: alkstore_postgres::PgOpts {
                schema: format!("{schema}_engine"),
                ..alkstore_postgres::PgOpts::default()
            },
            ..PgRegistryOptions::default()
        }
    }

    async fn count(
        client: &tokio_postgres::Client,
        schema: &str,
        table: &str,
        repo_id: &str,
    ) -> i64 {
        client
            .query_one(
                &format!(
                    "SELECT count(*) FROM {} WHERE repo_id = $1",
                    qualify(schema, table)
                ),
                &[&repo_id],
            )
            .await
            .unwrap()
            .get::<_, i64>(0)
    }

    /// Step order monotone across the twins (ADR-024's release
    /// discipline): the pg step set carries the same ordered integers
    /// sqlite's shipped set does, contiguous from 1.
    #[test]
    fn the_pg_step_set_is_monotone_across_the_twins() {
        assert_eq!(
            STEP_VERSIONS,
            &[1][..],
            "step order monotone across the twins — \
             sqlite's set carries the same integers"
        );
        let steps = build_steps("alkgit");
        assert_eq!(
            steps.iter().map(|s| s.version).collect::<Vec<_>>(),
            STEP_VERSIONS.to_vec(),
            "the built step set carries the pinned versions"
        );
        assert!(
            steps
                .iter()
                .enumerate()
                .all(|(i, s)| s.version == i as i64 + 1),
            "the step set is contiguous from 1"
        );

        #[cfg(feature = "sqlite")]
        assert_eq!(
            crate::migrations::CoTenantSqlite::step_count(),
            1,
            "the sqlite twin's shipped set — the same ordered integers"
        );
    }

    /// The family write semantics (the sqlite twin's shape, engine-
    /// swapped): the seq exact inside the row-write transaction (a
    /// second connection reads the committed bump), the record
    /// round-trips, the removal advances the watermark and leaves no
    /// row, and the seq singleton is durable across reopen.
    #[tokio::test]
    async fn pg_twin_carries_the_family_write_semantics_and_an_exact_seq() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("family");
        let handle = CoTenantPg::open(&dsn, &schema).await.unwrap();
        assert_eq!(
            handle.stamp().await.unwrap(),
            1,
            "a fresh schema — the initial step stamped"
        );
        assert_eq!(CoTenantPg::step_count(), 1);
        let twin = PgRegistry::new(handle.clone());

        let put_seq = twin.put(&record("owner/r")).await.unwrap();
        assert_eq!(put_seq, 1, "first mutation carries seq 1");

        let mut granted = private_record("owner/r");
        granted
            .grants
            .insert("u-7731".to_owned(), vec![AccessAction::Write]);
        let update_seq = twin.update(&granted).await.unwrap();
        assert_eq!(update_seq, 2);
        let rows = twin.rows_since(0).await.unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].seq, update_seq);
        assert_eq!(
            rows[0].record, granted,
            "the record round-trips the columns"
        );

        assert_eq!(twin.watermark().await.unwrap(), 2);
        let remove_seq = twin.remove("owner/r").await.unwrap();
        assert_eq!(remove_seq, 3);
        assert_eq!(
            twin.watermark().await.unwrap(),
            3,
            "the committed removal advances the watermark though it \
             leaves no row"
        );
        assert!(twin.rows_since(0).await.unwrap().is_empty());

        let second = connect_direct(&dsn).await;
        let seq_now: i64 = second
            .query_one(
                &format!(
                    "SELECT value FROM {} WHERE id = 1",
                    qualify(&schema, "registry_seq")
                ),
                &[],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(
            seq_now, 3,
            "the second connection sees the committed seq — the bump \
             rode the row-write transaction"
        );
        drop(second);

        let reopened = PgRegistry::new(CoTenantPg::open(&dsn, &schema).await.unwrap());
        assert_eq!(
            reopened.watermark().await.unwrap(),
            3,
            "the seq singleton is durable"
        );

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

    /// The closed registry family error set exactly as the mem and
    /// sqlite bindings': the not-found rejections, the taken id, the
    /// malformed records — each rejection consumes no seq.
    #[tokio::test]
    async fn pg_twin_error_map_matches_the_family() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("errors");
        let twin = PgRegistry::new(CoTenantPg::open(&dsn, &schema).await.unwrap());

        assert_eq!(
            twin.update(&record("ghost")).await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            twin.remove("ghost").await.unwrap_err(),
            RegistryError::NotFound
        );
        assert_eq!(
            twin.record_mutation("ghost").await.unwrap_err(),
            RegistryError::NotFound,
            "a marker write is refused for a removed repo"
        );

        assert_eq!(twin.put(&record("owner/r")).await.unwrap(), 1);
        assert_eq!(
            twin.put(&record("owner/r")).await.unwrap_err(),
            RegistryError::AlreadyExists
        );
        assert_eq!(
            twin.watermark().await.unwrap(),
            1,
            "the failed duplicate put consumed no seq"
        );

        let malformed = record("Owner/name");
        assert_eq!(
            twin.put(&malformed).await.unwrap_err(),
            RegistryError::Invalid
        );
        let mut empty_key = record("empty");
        empty_key
            .grants
            .insert(String::new(), vec![AccessAction::Read]);
        assert_eq!(
            twin.put(&empty_key).await.unwrap_err(),
            RegistryError::Invalid
        );
        assert_eq!(
            twin.watermark().await.unwrap(),
            1,
            "the rejections consumed no seq"
        );

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

    /// The refs-table legs ride the record-write transactions: put
    /// seeds the unborn `HEAD → refs/heads/main` symref row, update
    /// leaves it alone, and remove clears the record row, the repo's
    /// ref rows (the seed included), and the repo's mutation-marker
    /// rows together — the whole co-tenant repo state in one removal,
    /// while a sister repo's rows survive.
    #[tokio::test]
    async fn put_seeds_the_unborn_head_and_remove_clears_the_repo_rows() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("seed-head");
        let twin = PgRegistry::new(CoTenantPg::open(&dsn, &schema).await.unwrap());

        twin.put(&record("owner/seeded")).await.unwrap();
        let second = connect_direct(&dsn).await;
        let head: Option<String> = second
            .query_one(
                &format!(
                    "SELECT symref_target FROM {} WHERE repo_id = $1 AND name = 'HEAD'",
                    qualify(&schema, "git_refs")
                ),
                &[&"owner/seeded"],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(
            head.as_deref(),
            Some(HEAD_SEED_TARGET),
            "put seeded the unborn symref row in the record's transaction"
        );

        twin.update(&private_record("owner/seeded")).await.unwrap();
        let heads = count(&second, &schema, "git_refs", "owner/seeded").await;
        assert_eq!(
            heads, 1,
            "the update's transaction did not touch the seeded row"
        );

        twin.record_mutation("owner/seeded").await.unwrap();
        second
            .execute(
                &format!(
                    "INSERT INTO {} (repo_id, name, oid_hex, symref_target) \
                     VALUES ('owner/seeded', 'refs/heads/main', \
                     '0123456789abcdef0123456789abcdef01234567', NULL)",
                    qualify(&schema, "git_refs")
                ),
                &[],
            )
            .await
            .unwrap();

        twin.put(&record("owner/survivor")).await.unwrap();
        twin.remove("owner/seeded").await.unwrap();

        for (label, table) in [
            ("the record row", "git_repos"),
            ("the ref rows (the HEAD seed included)", "git_refs"),
            ("the mutation-marker rows", "repo_mutation_markers"),
        ] {
            assert_eq!(
                count(&second, &schema, table, "owner/seeded").await,
                0,
                "{label} removed with the registry removal"
            );
        }
        assert_eq!(
            count(&second, &schema, "git_repos", "owner/survivor").await,
            1,
            "a sister repo's row survived the removal"
        );
        drop(second);

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

    /// The per-push mutation-marker write: the seq/watermark legs move
    /// without touching the record (ADR-019 §7 — rows_since never
    /// carries markers), and the marker row upserts.
    #[tokio::test]
    async fn record_mutation_bumps_the_watermark_without_touching_the_record() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("marker");
        let handle = CoTenantPg::open(&dsn, &schema).await.unwrap();
        let twin = PgRegistry::new(handle.clone());

        let put_seq = twin.put(&record("owner/m")).await.unwrap();
        assert_eq!(put_seq, 1);
        let seq = twin.record_mutation("owner/m").await.unwrap();
        assert_eq!(seq, 2);
        assert_eq!(twin.watermark().await.unwrap(), 2);
        let rows = twin.rows_since(0).await.unwrap();
        assert_eq!(
            rows.iter().map(|row| row.seq).collect::<Vec<_>>(),
            vec![put_seq],
            "the marker left no record row — the record schema is \
             untouched"
        );

        let marker: i64 = handle
            .lock()
            .await
            .query_one(
                &format!(
                    "SELECT seq FROM {} WHERE repo_id = $1",
                    qualify(&schema, "repo_mutation_markers")
                ),
                &[&"owner/m"],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(marker, seq, "the marker row carries its seq");

        let again = twin.record_mutation("owner/m").await.unwrap();
        assert_eq!(again, 3, "the marker row upserts, never duplicates");

        twin.remove("owner/m").await.unwrap();
        let second = connect_direct(&dsn).await;
        assert_eq!(
            count(&second, &schema, "repo_mutation_markers", "owner/m").await,
            0,
            "the removal took the repo's marker rows with it"
        );
        drop(second);

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

    /// The announced store binding fires the composition post-commit
    /// over the live pg engine: the Put event is keyed in the feed,
    /// the LISTEN/NOTIFY wake lands on a listener, the
    /// pack-housekeeping job is on the maintenance queue, and the
    /// resolve rides the write-through provider.
    #[tokio::test]
    async fn announced_store_binding_fires_the_composition_post_commit() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("announce");
        let assembly = PgRegistryAssembly::open(&dsn, assembly_opts(&schema))
            .await
            .unwrap();
        let mut listener = assembly.store().listen(WAKE_CHANNEL).await.unwrap();

        let mut granted = record("owner/a");
        granted
            .grants
            .insert("u-1".to_owned(), vec![AccessAction::Read]);
        assembly.backend().put_repo(&granted).await.unwrap();

        let events: Vec<StreamEvent> = assembly
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .read_since(0, 10)
            .await
            .unwrap();
        assert_eq!(events.len(), 1, "one announced put = one feed event");
        assert_eq!(events[0].key.as_deref(), Some("owner/a"));
        assert_eq!(
            crate::reactor::RegistryEvent::decode_event(&events[0]),
            Some(crate::reactor::RegistryEvent::Put {
                seq: 1,
                record: granted.clone(),
            })
        );

        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect("the announce's NOTIFY delivered over the LISTEN leg");
        assert_eq!(wake.channel, WAKE_CHANNEL);

        let queue = assembly
            .store()
            .queue(MAINTENANCE_QUEUE, QueueOpts::default())
            .await
            .unwrap();
        let claimed = queue
            .claim_one("probe")
            .await
            .unwrap()
            .expect("the pack-housekeeping job");
        assert_eq!(
            claimed
                .job()
                .payload_as::<crate::reactor::MaintenanceJob>()
                .unwrap(),
            crate::reactor::MaintenanceJob::PackHousekeeping {
                repo_id: "owner/a".to_owned(),
            }
        );

        assert_eq!(
            assembly.backend().resolve("owner/a").await.unwrap(),
            granted,
            "the write-through provider serves the mutation"
        );

        assembly.backend().remove_repo("owner/a").await.unwrap();
        let events: Vec<StreamEvent> = assembly
            .store()
            .stream(crate::reactor::EVENTS_STREAM)
            .await
            .unwrap()
            .read_since(1, 10)
            .await
            .unwrap();
        assert_eq!(
            events
                .first()
                .and_then(crate::reactor::RegistryEvent::decode_event),
            Some(crate::reactor::RegistryEvent::Removed {
                seq: 2,
                repo_id: "owner/a".to_owned(),
            })
        );

        drop(listener);
        drop_schema(&dsn, &schema).await;
        drop_schema(&dsn, &format!("{schema}_engine")).await;
    }

    /// The LFS liveness rows go with the removed repo — the
    /// unconditional `lfs_objects` leg inside the record-removal
    /// transaction (the rows only exist when the `lfs` surface wrote
    /// them; the leg is unconditional so a mixed deployment that
    /// turned the feature off must not leak). The rows seed directly
    /// here (the leg's SQL is unconditional — the test holds with the
    /// surface off too), a shared oid's surviving row recomputes, and
    /// the victim's oids leave no row at all.
    #[tokio::test]
    async fn remove_repo_takes_the_lfs_rows_and_recounts_the_survivors() {
        let dsn = live_pg!();
        let schema = unique_schema("remove-lfs");
        let assembly = PgRegistryAssembly::open(&dsn, assembly_opts(&schema))
            .await
            .unwrap();
        assembly.durable().put(&record("victim")).await.unwrap();
        assembly.durable().put(&record("survivor")).await.unwrap();

        let shared_oid = "01".repeat(32);
        let victim_only_oid = "02".repeat(32);
        {
            let guard = assembly.durable().handle.lock().await;
            guard
                .execute(
                    &format!(
                        "INSERT INTO {} (repo_id, oid, size, refcount) \
                         VALUES ('victim', $1, 8, 2), ('survivor', $1, 8, 2), \
                                 ('victim', $2, 4, 1)",
                        qualify(&schema, "lfs_objects")
                    ),
                    &[&shared_oid, &victim_only_oid],
                )
                .await
                .unwrap();
        }

        assembly.backend().remove_repo("victim").await.unwrap();

        let second = connect_direct(&dsn).await;
        let victim_only: i64 = second
            .query_one(
                &format!(
                    "SELECT count(*) FROM {} WHERE oid = $1",
                    qualify(&schema, "lfs_objects")
                ),
                &[&victim_only_oid],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(victim_only, 0, "the victim's own oid left no row at all");
        let shared_count: i64 = second
            .query_one(
                &format!(
                    "SELECT refcount FROM {} WHERE repo_id = 'survivor' AND oid = $1",
                    qualify(&schema, "lfs_objects")
                ),
                &[&shared_oid],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(
            shared_count, 1,
            "the surviving row carries the recomputed count (the re-protection contract's data)"
        );
        drop(second);

        #[cfg(feature = "lfs")]
        assert_eq!(
            crate::lfs::LfsRows::repo_rows(assembly.lfs_rows().as_ref(), "survivor")
                .await
                .unwrap(),
            vec![crate::lfs::LfsRow {
                repo_id: "survivor".to_owned(),
                oid: shared_oid,
                size: 8,
                refcount: 1,
            }],
            "the surface reads back the same state (the composition's row store)"
        );

        drop_schema(&dsn, &schema).await;
        drop_schema(&dsn, &format!("{schema}_engine")).await;
    }

    /// The asymmetry probe (the POC-R pg gate, re-landed): a
    /// notify-free write on the co-tenant twin wakes nothing — no
    /// file-level bump exists on postgres — and the cache stays stale
    /// until the reconcile sweep closes the gap; the tick's cadence
    /// is the real staleness bound. Then the announced write's NOTIFY
    /// demonstrably delivers (the load-bearing leg is live when
    /// wired).
    #[tokio::test]
    async fn the_asymmetry_probe_notify_free_write_wakes_nothing_and_reconcile_closes_the_gap() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("asymmetry");
        let assembly = PgRegistryAssembly::open(&dsn, assembly_opts(&schema))
            .await
            .unwrap();
        let mut listener = assembly.store().listen(WAKE_CHANNEL).await.unwrap();
        let (stop, runner) = spawn_wake_runner(assembly.reactor().clone());
        tokio::time::sleep(Duration::from_millis(200)).await;

        let seq = assembly
            .durable()
            .put(&private_record("owner/silent"))
            .await
            .unwrap();

        let wake = listener
            .recv_timeout(Duration::from_millis(700))
            .await
            .unwrap();
        assert!(
            wake.is_none(),
            "a notify-free write wakes nothing on postgres — no \
             file-level bump exists (LISTEN/NOTIFY is per-notify)"
        );
        assert_eq!(
            assembly.provider().resolve("owner/silent"),
            None,
            "the cache is stale — the reconcile tick's cadence is the \
             staleness bound"
        );

        let applied = AppliedWatermark::new();
        let caught = assembly.reactor().reconcile(&applied).await.unwrap();
        assert_eq!(
            caught
                .iter()
                .map(|row| row.record.repo_id.as_str())
                .collect::<Vec<_>>(),
            vec!["owner/silent"],
            "the reconcile sweep closed the gap in one pass"
        );
        assert_eq!(
            assembly.provider().resolve("owner/silent"),
            Some(private_record("owner/silent"))
        );
        assert_eq!(
            assembly.durable().watermark().await.unwrap(),
            seq,
            "the watermark advanced though nothing announced"
        );

        assembly
            .write()
            .update(&record("owner/silent"))
            .await
            .unwrap();
        let wake = listener
            .recv_timeout(Duration::from_secs(2))
            .await
            .unwrap()
            .expect("the announced write's NOTIFY delivered");
        assert_eq!(wake.channel, WAKE_CHANNEL);
        wait_until(|| {
            assembly
                .provider()
                .resolve("owner/silent")
                .as_ref()
                .is_some_and(|resolved| resolved == &record("owner/silent"))
        })
        .await;

        drop(listener);
        stop_and_join(stop, runner).await;
        drop_schema(&dsn, &schema).await;
        drop_schema(&dsn, &format!("{schema}_engine")).await;
    }

    /// The assembly constructor is loud and loads the persisted
    /// registry: a reopened assembly resolves the stored record with
    /// no refresh, and the seq singleton persists.
    #[tokio::test]
    async fn assembly_open_is_loud_and_loads_the_persisted_registry() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("reopen");
        {
            let assembly = PgRegistryAssembly::open(&dsn, assembly_opts(&schema))
                .await
                .unwrap();
            assembly
                .backend()
                .put_repo(&private_record("owner/kept"))
                .await
                .unwrap();
            assert_eq!(assembly.durable().watermark().await.unwrap(), 1);
        }

        let reopened = PgRegistryAssembly::open(&dsn, assembly_opts(&schema))
            .await
            .unwrap();
        assert_eq!(
            reopened.backend().resolve("owner/kept").await.unwrap(),
            private_record("owner/kept"),
            "the initial snapshot load carried the persisted table"
        );
        assert_eq!(
            reopened.durable().watermark().await.unwrap(),
            1,
            "the seq singleton persisted and continues the mutation count"
        );
        assert_eq!(
            reopened.provider().refresh_count(),
            1,
            "load only — no refreshes at construction beyond the \
             initial snapshot"
        );

        drop_schema(&dsn, &schema).await;
        drop_schema(&dsn, &format!("{schema}_engine")).await;
    }

    /// The corrupt row defenses: a row the composition did not write
    /// (malformed grants JSON, corrupt visibility text) is
    /// structurally invalid row state — `RegistryError::Invalid` on
    /// the catch-up read, never a poison row wedging the reconcile.
    #[tokio::test]
    async fn corrupt_record_rows_map_registry_invalid() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("corrupt");
        let handle = CoTenantPg::open(&dsn, &schema).await.unwrap();
        let twin = PgRegistry::new(handle.clone());
        twin.put(&record("owner/ok")).await.unwrap();
        let second = connect_direct(&dsn).await;
        second
            .execute(
                &format!(
                    "INSERT INTO {} (repo_id, visibility, grants, seq) \
                     VALUES ('owner/bent', 'public', 'not-json', 2)",
                    qualify(&schema, "git_repos")
                ),
                &[],
            )
            .await
            .unwrap();
        drop(second);

        let err = twin.rows_since(0).await.unwrap_err();
        assert_eq!(err, RegistryError::Invalid, "{err:?}");

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

    /// Concurrent-open safety (ADR-024 §3, the pg arm): two
    /// concurrent opens of one schema — the advisory migration lock
    /// serializes them, the loser sees the winner's committed stamp
    /// and skips — both converge on the stamped schema.
    #[tokio::test]
    async fn concurrent_open_of_one_schema_races_to_one_stamped_schema() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("concurrent");
        let (a, b) = tokio::join!(
            CoTenantPg::open(&dsn, &schema),
            CoTenantPg::open(&dsn, &schema)
        );
        let a = a.expect("no opener fails");
        let b = b.expect("no opener fails");
        assert_eq!(a.stamp().await.unwrap(), 1);
        assert_eq!(
            b.stamp().await.unwrap(),
            1,
            "the benign race is swallowed — both openers converge on \
             the stamped schema"
        );

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

    /// No refuse-gate ships in v1 (the coexistence contract, ADR-024
    /// §4): a stamp advanced by a forward step is safe to open
    /// against and left untouched; the schema is intact.
    #[tokio::test]
    async fn an_unknown_forward_stamp_is_safe_to_open_against() {
        let _gate = LIVE_GATE.lock().await;
        let dsn = live_pg!();
        let schema = unique_schema("forward-stamp");
        {
            let handle = CoTenantPg::open(&dsn, &schema).await.unwrap();
            assert_eq!(handle.stamp().await.unwrap(), 1, "the initial step stamped");
            let second = connect_direct(&dsn).await;
            second
                .execute(
                    &format!(
                        "UPDATE {} SET stamp = 99 WHERE id = 1",
                        qualify(&schema, STAMP_TABLE)
                    ),
                    &[],
                )
                .await
                .unwrap();
            drop(second);
        }

        let reopened = CoTenantPg::open(&dsn, &schema).await.unwrap();
        assert_eq!(
            reopened.stamp().await.unwrap(),
            99,
            "the forward stamp is left untouched"
        );
        let second = connect_direct(&dsn).await;
        let rows: i64 = second
            .query_one(
                &format!("SELECT count(*) FROM {}", qualify(&schema, "git_repos")),
                &[],
            )
            .await
            .unwrap()
            .get(0);
        assert_eq!(rows, 0, "the schema is intact — the tables still serve");
        drop(second);

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

    /// The constructor fails loudly on an unreachable server and
    /// before touching a server at all on an unusable schema name —
    /// the fallible-constructor posture (ADR-024 §1).
    #[tokio::test]
    async fn open_is_loud_on_unreachable_and_unusable_input() {
        assert!(
            matches!(
                CoTenantPg::open(
                    "host=127.0.0.1 port=1 user=postgres password=x dbname=x",
                    ""
                )
                .await,
                Err(OpenError::Engine(_))
            ),
            "an unreachable server fails loud at construction"
        );
        assert!(
            matches!(
                CoTenantPg::open("", "\"bent\"").await,
                Err(OpenError::Invalid(_))
            ),
            "the schema-name boundary rejects before any connection"
        );
    }
}