alkgit_substrate/

migrations.rs

//! The registry schema migrations (ADR-024) and the co-tenant
//! connection handle.
//!
//! # The machinery (the shared pattern)
//!
//! One alkgit-owned stamp table versions the co-tenant schema *as a
//! unit* — the record rows, ref rows, the `registry_seq` singleton,
//! the mutation-marker rows, and the LFS liveness rows ship and evolve
//! together, so one integer stamp covers them all (ADR-024 §2). Steps
//! are ordered, hand-written, and monomorphic *per engine twin* (the
//! twins already carry engine-specific SQL; migrations ride them — no
//! framework, no builder). The runner's contract, identical for both
//! twins:
//!
//! - **One transaction per step** — the step's DDL and its stamp
//!   advance commit together; a failed transaction rolls back whole
//!   and nothing is left half-applied by the failed step.
//! - **Construction-run, loud failure** — migrations run when the
//!   adapter constructs its connection, and any unverified failure
//!   rides [`OpenError::Migration`]: a
//!   half-migrated store refuses service rather than degrading
//!   silently (the fallible-constructor posture ADR-024 §1). Every
//!   construction re-runs the runner, so a half-migrated store keeps
//!   refusing until the state is real or the store is reset.
//! - **Concurrent-open-safe** — the stamp is read inside each step's
//!   own `BEGIN IMMEDIATE` transaction (the write lock serializes
//!   racing openers; a loser sees the winner's committed stamp and
//!   skips). A step whose SQL nevertheless fails verifies state
//!   against committed data — the alkstore ADR-012 §5 re-key pattern:
//!   the step's effect present ⇒ benign race swallowed (and the stamp
//!   healed, converging the store); absent ⇒ propagate. Load-bearing
//!   on sqlite, where multi-process co-tenancy of one file is native.
//! - **Absent stamp ⇔ fresh store** — the stamp table bootstrap
//!   creates the stamp row at value 0; a fresh store runs the whole
//!   step set and stamps it.
//! - **Additive-first, no refuse-gate in v1** — an unknown-forward
//!   stamp is safe to open against, and the runner leaves it untouched
//!   (the coexistence contract: explicit column lists, unknown
//!   columns/tables unread — ADR-024 §4). A non-additive step is
//!   major-version territory and owns its refusal semantics then.
//!
//! # The engine-neutral logical DDL (the shared schema, both twins)
//!
//! The initial co-tenant schema, stamped step 1 (ADR-024 §2/§5), in
//! engine-neutral terms — both engines' SQL lands in this module over
//! exactly this logical schema: the sqlite twin's step set is the
//! `sqlite` module below; the postgres twin writes step 1's pg SQL
//! (BIGINT sequence columns, CHECK'd visibility text) in its own
//! module:
//!
//! - `alkgit_schema_stamp (id = 1, stamp)` — the alkgit-owned stamp
//!   row (the machinery's meta table, beside its own tables).
//! - `git_repos (repo_id PK, visibility in {'public','private'},
//!   grants, seq)` — the record rows: the backend.md §`RepoRecord`
//!   serde shape serialized (`grants` is the JSON object
//!   identity-id → action array; `seq` is the row's mutation sequence
//!   number, assigned inside the row-write transaction).
//! - `git_refs (repo_id, name) → oid, symref_target` — the ref rows
//!   (POC-O-validated shape; ADR-019 §2). HEAD is a symref row: `oid`
//!   NULL with a set target — exactly the unborn
//!   `RefLine { oid: None }` shape.
//! - `registry_seq (id = 1, value)` — the watermark singleton; every
//!   registry mutation bumps it inside the row-write transaction
//!   (transactionally exact, ADR-019 §5).
//! - `repo_mutation_markers (repo_id PK, seq)` — the per-push
//!   mutation-marker rows (composition-internal, ADR-019 §7: pushes
//!   do not touch `RepoRecord`; the receive flow's watermark leg; the
//!   record schema is untouched).
//! - `lfs_objects (repo_id, oid) → size, refcount` — the LFS liveness
//!   rows with the **day-one refcount column** (ADR-024 §5 — nothing
//!   published predates it, so no migration is owed and none should be
//!   written; the column's consumption is wave 9). The table ships in
//!   the step set unconditionally — the tables evolve as one
//!   composition; the `lfs` feature gates the surface consuming it, so
//!   an inert table is harmless and the stamp stays unit-shaped.
//!
//! The stamp table itself is bootstrapped idempotently before the
//! first read (`CREATE TABLE IF NOT EXISTS` + `INSERT OR IGNORE` at
//! 0), so "absent stamp ⇔ fresh store" holds under concurrent opens.
//!
//! # The co-tenant connection handles (one per engine twin)
//!
//! The handles' placement is symmetric across engines — one file-local
//! module per engine twin in this, the ADR-024 machinery home:
//!
//! - [`CoTenantSqlite`] (the `sqlite` feature) — the thin wrapper over
//!   the rusqlite connection that the registry adapter and the refs
//!   layer both consume: it opens the co-tenant connection (WAL,
//!   NORMAL synchronous, pinned 5000 ms busy timeout — substrate.md
//!   §assembly tuning datums) beside the engine-owned schemas on the
//!   same file (alkblobs ADR-013 §3's composition posture), runs the
//!   migrations at construction, and hands out the connection for the
//!   twins' blocking SQL (which the composition's concurrency model
//!   bridges off the async executor).
//! - [`CoTenantPg`] (the `postgres` feature) — the shape's other half:
//!   the thin wrapper over the tokio-postgres client the postgres
//!   twins consume, natively async (no blocking bridge — the shared
//!   client lives under a `tokio::sync::Mutex` whose guard is held for
//!   one statement batch or transaction, the twin's short-held-lock
//!   model). Its engine-locally-realized runner (below) rides the
//!   advisory migration lock instead of `BEGIN IMMEDIATE`.
//!
//! The postgres twin's pg machinery also carries the crate's advisory
//! lock-key plumbing over the co-tenant schema: `advisory_lock_key`
//! (the deterministic key over a scope's identity string) plus the
//! named scope-key constructors beside it — the stamp-table/migration
//! runner's key, the `git_refs` CAS scope's key (`git_refs_lock_key`,
//! the refs CAS twins' and every registry transaction touching the
//! refs table's serialization point), and the `lfs_objects` rows
//! scope's key (`lfs_objects_lock_key`) — every transaction touching
//! a co-tenant table holds that table's advisory lock (the wave-9
//! lock-discipline invariant), and the composition spells the scope
//! identity through these constructors only, never hand-inlined. The
//! runner's step set is the pg step builder (`build_steps`), step
//! order monotone with sqlite's (`STEP_VERSIONS`).

#[cfg(feature = "sqlite")]
use crate::error::OpenError;

#[cfg(feature = "sqlite")]
mod sqlite_twin {
    use super::OpenError;
    use std::path::Path;
    use std::sync::Arc;

    /// The alkgit-owned stamp table — one row versioning the
    /// co-tenant schema as a unit (ADR-024 §2).
    const STAMP_TABLE: &str = "alkgit_schema_stamp";

    /// The pinned co-tenant busy timeout, in milliseconds
    /// (substrate.md §assembly tuning datums — contracted at both
    /// substrate crates; the co-tenant connection carries it too).
    pub(crate) const BUSY_TIMEOUT_MS: i64 = 5_000;

    /// The stamp table's idempotent bootstrap: create the table if
    /// absent, seed the single stamp row at 0 (absent stamp ⇔ fresh
    /// store). Safe under concurrent opens — sqlite serializes the
    /// autocommit statements and both statements are
    /// existence-guarded.
    pub(crate) const STAMP_TABLE_BOOTSTRAP: &str = "
        CREATE TABLE IF NOT EXISTS alkgit_schema_stamp (
            id    INTEGER PRIMARY KEY CHECK (id = 1),
            stamp INTEGER NOT NULL
        );
        INSERT OR IGNORE INTO alkgit_schema_stamp (id, stamp) VALUES (1, 0);";

    /// One ordered migration step: its version (the stamp value it
    /// marks — the step's position in the twin's global ordered set),
    /// its monomorphic SQL applied inside one transaction, plus its
    /// presence predicate — a query returning a single integer, truthy
    /// exactly when the step's complete effect is present on the
    /// engine. The predicate is the race-verification re-key (the
    /// alkstore ADR-012 §5 pattern): on a step failure, its presence
    /// decides swallow-vs-propagate.
    pub(crate) struct Step {
        pub(crate) version: i64,
        pub(crate) name: &'static str,
        pub(crate) sql: &'static str,
        pub(crate) verify_sql: &'static str,
    }

    /// The sqlite step set. Step 1 is the initial co-tenant schema
    /// (the engine-neutral DDL of the module doc, in sqlite SQL).
    /// Post-publish additions append — additive-first (ADR-024 §4).
    const SQLITE_STEPS: &[Step] = &[Step {
        version: 1,
        name: "initial co-tenant schema (records, refs, seq, mutation markers, LFS liveness rows)",
        sql: "CREATE TABLE git_repos (
                  repo_id        TEXT PRIMARY KEY,
                  visibility     TEXT NOT NULL CHECK (visibility IN ('public', 'private')),
                  grants         TEXT NOT NULL,
                  seq            INTEGER NOT NULL
              );
              CREATE TABLE git_refs (
                  repo_id        TEXT NOT NULL,
                  name           TEXT NOT NULL,
                  oid_hex        TEXT,
                  symref_target  TEXT,
                  PRIMARY KEY (repo_id, name)
              );
              CREATE TABLE registry_seq (
                  id    INTEGER PRIMARY KEY CHECK (id = 1),
                  value INTEGER NOT NULL
              );
              INSERT INTO registry_seq (id, value) VALUES (1, 0);
              CREATE TABLE repo_mutation_markers (
                  repo_id  TEXT PRIMARY KEY,
                  seq      INTEGER NOT NULL
              );
              CREATE TABLE lfs_objects (
                  repo_id   TEXT NOT NULL,
                  oid       TEXT NOT NULL,
                  size      INTEGER NOT NULL,
                  refcount  INTEGER NOT NULL,
                  PRIMARY KEY (repo_id, oid)
              );",
        verify_sql: "SELECT count(*) = 5 FROM sqlite_master WHERE type = 'table' \
                     AND name IN ('git_repos', 'git_refs', 'registry_seq', \
                                  'repo_mutation_markers', 'lfs_objects')",
    }];

    /// The co-tenant connection handle — the thin wrapper over the
    /// rusqlite connection that the sqlite registry adapter and the
    /// refs layer both consume (alkgit owns its tables and its
    /// connection, co-tenant beside the engine-owned schemas on the
    /// same file — alkblobs ADR-013 §3's composition posture).
    ///
    /// Construction is the assembly-time fallible surface: the open,
    /// the WAL/NORMAL/busy-timeout pinning, and the full
    /// construction-time migration run all fail loudly into
    /// [`OpenError`](crate::OpenError) — a half-migrated store refuses
    /// service (ADR-024 §1). The connection is shared under a
    /// short-held lock; consumers run their blocking SQL under
    /// [`CoTenantSqlite::lock`], bridged off the async executor per
    /// the composition's concurrency model (the trait impls'
    /// `spawn_blocking` home).
    #[derive(Clone)]
    pub struct CoTenantSqlite {
        conn: Arc<parking_lot::Mutex<rusqlite::Connection>>,
    }

    impl CoTenantSqlite {
        /// Open the co-tenant connection at `path` — the same file the
        /// alkstore engine was opened at — run the migrations, and
        /// return the handle. In-memory paths are rejected: `:memory:`
        /// confers no co-tenancy (per-connection databases; WAL cannot
        /// engage) — ephemeral assemblies compose the mem engines
        /// (substrate.md §assembly shapes, DQ-7).
        pub fn open(path: impl AsRef<Path>) -> Result<Self, OpenError> {
            let path = path.as_ref();
            if path == Path::new(":memory:") {
                return Err(OpenError::Invalid(
                    "sqlite memory mode confers no co-tenancy — compose the \
                     mem engines for ephemeral assemblies"
                        .into(),
                ));
            }
            let conn = rusqlite::Connection::open(path).map_err(|e| {
                OpenError::Engine(format!("co-tenant sqlite open ({}): {e}", path.display()))
            })?;
            conn.pragma_update(None, "busy_timeout", BUSY_TIMEOUT_MS)
                .map_err(|e| OpenError::Engine(format!("co-tenant sqlite busy_timeout: {e}")))?;
            if let Err(e) = conn.pragma_update(None, "journal_mode", "WAL") {
                // Concurrent opens race on the journal-mode switch (it
                // takes the file lock briefly and SQLITE_BUSY is not
                // retried by the busy handler). Bounded verify-or-retry:
                // WAL ⇒ a concurrent opener set it — benign; a still-not-
                // WAL file gets a few bounded re-attempts of its own
                // (the mode is never un-set once WAL, so persistence past
                // the window is a real failure) — then propagate loud.
                const VERIFY_RETRIES: usize = 25;
                let mut became_wal = false;
                for _ in 0..VERIFY_RETRIES {
                    let mode: Result<String, _> =
                        conn.query_row("PRAGMA journal_mode", [], |r| r.get(0));
                    if mode.is_ok_and(|m| m.eq_ignore_ascii_case("wal")) {
                        became_wal = true;
                        break;
                    }
                    if conn.pragma_update(None, "journal_mode", "WAL").is_ok() {
                        became_wal = true;
                        break;
                    }
                    std::thread::sleep(std::time::Duration::from_millis(10));
                }
                if !became_wal {
                    return Err(OpenError::Engine(format!(
                        "co-tenant sqlite journal_mode: {e}"
                    )));
                }
            }
            conn.pragma_update(None, "synchronous", "NORMAL")
                .map_err(|e| OpenError::Engine(format!("co-tenant sqlite synchronous: {e}")))?;
            let mut conn = conn;
            migrate(&mut conn)?;
            Ok(CoTenantSqlite {
                conn: Arc::new(parking_lot::Mutex::new(conn)),
            })
        }

        /// Hold the connection for blocking SQL work. The guard is
        /// short-held — one statement batch or transaction per hold,
        /// bridged off the async executor by the caller.
        pub fn lock(&self) -> parking_lot::MutexGuard<'_, rusqlite::Connection> {
            self.conn.lock()
        }

        /// The store's current co-tenant schema stamp — the version
        /// support and diagnostics report (an ADR-024 consequence: a
        /// real version to report). 0 means no step applied yet; a
        /// value beyond the shipped step set is the safe
        /// unknown-forward coexistence shape (ADR-024 §4).
        pub fn stamp(&self) -> Result<i64, OpenError> {
            read_stamp(&self.lock())
        }

        /// The shipped step count — the highest stamp this binary's
        /// step set owns (an honest coexistence datum beside
        /// [`CoTenantSqlite::stamp`]).
        pub fn step_count() -> usize {
            SQLITE_STEPS.len()
        }
    }

    /// The construction-time migration run: the stamp-table bootstrap
    /// plus every pending step, loud on unverified failure. Returns
    /// the post-run stamp.
    pub(crate) fn migrate(conn: &mut rusqlite::Connection) -> Result<i64, OpenError> {
        run_steps(conn, SQLITE_STEPS)
    }

    /// The step runner — the shared pattern of the module doc, over
    /// any ordered step set (the tests drive it with probe steps; the
    /// postgres twin mirrors the pattern with its own SQL).
    pub(crate) fn run_steps(
        conn: &mut rusqlite::Connection,
        steps: &[Step],
    ) -> Result<i64, OpenError> {
        conn.execute_batch(STAMP_TABLE_BOOTSTRAP)
            .map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("stamp-table bootstrap ({STAMP_TABLE}): {e}"),
            })?;
        for step in steps {
            apply_step(conn, step)?;
        }
        read_stamp(conn)
    }

    /// Apply one step if pending. The stamp is read inside the step's
    /// own BEGIN IMMEDIATE transaction — the write lock serializes
    /// concurrent openers, so a loser observes the winner's committed
    /// stamp and skips without ever executing duplicate DDL.
    fn apply_step(conn: &mut rusqlite::Connection, step: &Step) -> Result<(), OpenError> {
        let step_no = step.version as usize;
        let tx = conn
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
            .map_err(|e| OpenError::Migration {
                step: step_no,
                reason: format!("step {step_no} ({}): BEGIN IMMEDIATE: {e}", step.name),
            })?;
        let stamp_now: i64 = tx
            .query_row(read_stamp_sql().as_str(), [], |r| r.get(0))
            .map_err(|e| OpenError::Migration {
                step: step_no,
                reason: format!("step {step_no} ({}): stamp read: {e}", step.name),
            })?;
        if stamp_now >= step.version {
            return Ok(());
        }
        let outcome = tx
            .execute_batch(step.sql)
            .and_then(|_| tx.execute(write_stamp_sql().as_str(), rusqlite::params![step.version]))
            .and_then(|_| tx.commit());
        match outcome {
            Ok(()) => Ok(()),
            Err(e) => verify_race(conn, step, &e),
        }
    }

    /// The failure re-key (the ADR-024 §3 race-verification swallow —
    /// the alkstore ADR-012 §5 pattern): the failed transaction rolled
    /// back whole, so only a *concurrently committed* application (or
    /// an externally created equivalent state) can explain a present
    /// effect. Present ⇒ benign race swallowed (healing the stamp when
    /// the concurrent applier's own stamp write has not landed);
    /// absent ⇒ propagate loud.
    fn verify_race(
        conn: &rusqlite::Connection,
        step: &Step,
        failure: &rusqlite::Error,
    ) -> Result<(), OpenError> {
        if step_effect_present(conn, step)? {
            let step_no = step.version as usize;
            conn.execute(mark_stamp_sql().as_str(), rusqlite::params![step.version])
                .map_err(|e| OpenError::Migration {
                    step: step_no,
                    reason: format!(
                        "step {step_no} ({}): race-swallow stamp heal: {e}",
                        step.name
                    ),
                })?;
            Ok(())
        } else {
            let step_no = step.version as usize;
            Err(OpenError::Migration {
                step: step_no,
                reason: format!("step {step_no} ({}): {failure}", step.name),
            })
        }
    }

    /// The step's presence predicate: truthy single-integer row iff
    /// the step's complete effect is present. A query failure is loud
    /// (unverifiable state must not be swallowed).
    fn step_effect_present(conn: &rusqlite::Connection, step: &Step) -> Result<bool, OpenError> {
        let step_no = step.version as usize;
        match conn.query_row(step.verify_sql, [], |r| r.get::<_, i64>(0)) {
            Ok(truthy) => Ok(truthy > 0),
            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(false),
            Err(e) => Err(OpenError::Migration {
                step: step_no,
                reason: format!("step {step_no} ({}): verification query: {e}", step.name),
            }),
        }
    }

    fn read_stamp_sql() -> String {
        format!("SELECT stamp FROM {STAMP_TABLE} WHERE id = 1")
    }

    fn write_stamp_sql() -> String {
        format!("UPDATE {STAMP_TABLE} SET stamp = ?1 WHERE id = 1")
    }

    fn mark_stamp_sql() -> String {
        format!("UPDATE {STAMP_TABLE} SET stamp = ?1 WHERE id = 1 AND stamp < ?1")
    }

    pub(crate) fn read_stamp(conn: &rusqlite::Connection) -> Result<i64, OpenError> {
        conn.query_row(read_stamp_sql().as_str(), [], |r| r.get(0))
            .map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("stamp read ({STAMP_TABLE}): {e}"),
            })
    }

    #[cfg(test)]
    pub(crate) fn presence_sql(table: &str) -> &'static str {
        Box::leak(
            format!(
                "SELECT count(*) = 1 FROM sqlite_master \
                 WHERE type = 'table' AND name = '{table}'"
            )
            .into_boxed_str(),
        )
    }

    #[cfg(test)]
    pub(crate) fn step_probe(table: &str, name: &'static str, version: i64) -> Step {
        Step {
            version,
            name,
            sql: Box::leak(format!("CREATE TABLE {table} (x TEXT, y TEXT)").into_boxed_str()),
            verify_sql: presence_sql(table),
        }
    }

    #[cfg(test)]
    pub(crate) fn failing_step_probe(table: &str, name: &'static str, version: i64) -> Step {
        Step {
            version,
            name,
            sql: Box::leak(
                format!(
                    "CREATE TABLE {table} (x TEXT, y TEXT); CREATE TABLE {table} (x TEXT, y TEXT)"
                )
                .into_boxed_str(),
            ),
            verify_sql: presence_sql(table),
        }
    }
}

#[cfg(feature = "sqlite")]
pub use sqlite_twin::CoTenantSqlite;

#[cfg(feature = "postgres")]
pub(crate) mod pg_twin {
    use std::sync::Arc;

    use alkstore_postgres::quote_identifier;
    use tokio_postgres::NoTls;

    use crate::error::OpenError;

    /// The co-tenant schema the pg machinery's tables live in by
    /// default — alkgit's own PostgreSQL schema, beside the engine's
    /// (`alkstore-postgres`' default `alkstore`) and the alkblobs kv
    /// shapes' on the same database (the co-tenancy composition posture,
    /// alkblobs ADR-013 §3).
    pub const DEFAULT_SCHEMA: &str = "alkgit";

    /// The alkgit-owned stamp table — one row versioning the co-tenant
    /// schema as a unit (ADR-024 §2), beside the co-tenant tables in the
    /// alkgit schema.
    pub(crate) const STAMP_TABLE: &str = "alkgit_schema_stamp";

    /// The shipped pg step set's versions, in order (step order
    /// monotone across the twins — sqlite's set carries the same
    /// integers, ADR-024's release discipline; the parity test pins it).
    pub(crate) const STEP_VERSIONS: &[i64] = &[1];

    /// The module's schema-name boundary: non-empty names validated
    /// conservatively (no controls, no quotes, ≤ 63 bytes, not
    /// digit-leading) so every later SQL build composes from
    /// [`quote_identifier`]'s escaping alone; an empty option means
    /// [`DEFAULT_SCHEMA`].
    fn validated_schema(schema: &str) -> Result<Arc<str>, OpenError> {
        let schema = if schema.is_empty() {
            DEFAULT_SCHEMA
        } else {
            schema
        };
        if schema.len() > 63
            || schema.starts_with(|c: char| c.is_ascii_digit())
            || schema
                .chars()
                .any(|c| c.is_control() || c == '"' || c == '\0')
        {
            return Err(OpenError::Invalid(format!(
                "postgres co-tenant schema name is not a usable identifier: {schema:?}"
            )));
        }
        Ok(Arc::from(schema))
    }

    /// The qualified `schema.table` identifier (crate-internal SQL
    /// composition — the refs twin's legs compose through it too).
    pub(crate) fn qualify(schema: &str, table: &str) -> String {
        format!("{}.{}", quote_identifier(schema), quote_identifier(table))
    }

    /// The advisory-lock key over a scope's identity string (schema +
    /// what it protects): deterministic, so concurrent openers/writers of
    /// one schema serialize on the same key and differently-scoped
    /// schemas never serialize on each other.
    fn advisory_lock_key(scope: &str) -> i64 {
        let mut hash: u64 = 0xcbf29ce484222325;
        for byte in scope.as_bytes() {
            hash ^= u64::from(*byte);
            hash = hash.wrapping_mul(0x100000001b3);
        }
        hash as i64
    }

    /// The schema-stamp scope's key — the migration runner's
    /// serialization (deterministically the same for every opener of
    /// one schema, so concurrent opens serialize exactly the way the
    /// sqlite machinery's `BEGIN IMMEDIATE` write lock serializes its
    /// openers; differently-named co-tenant schemas — assembly tests,
    /// sibling deployments — never serialize on each other).
    fn migration_lock_key(schema: &str) -> i64 {
        advisory_lock_key(&format!("{schema}\0{STAMP_TABLE}"))
    }

    /// The refs-table scope's key — the refs CAS twins hold it for the
    /// call's whole transaction, and every registry transaction
    /// touching `git_refs` (the HEAD seed, the removal's refs leg)
    /// takes it first statement.
    pub(crate) fn git_refs_lock_key(schema: &str) -> i64 {
        advisory_lock_key(&format!("{schema}\0git_refs"))
    }

    /// The LFS rows-table scope's key — the LFS row twins take it for
    /// each write transaction, and every transaction touching
    /// `lfs_objects` (the removal's LFS leg with its recount) takes it
    /// first statement.
    pub(crate) fn lfs_objects_lock_key(schema: &str) -> i64 {
        advisory_lock_key(&format!("{schema}\0lfs_objects"))
    }

    /// The pg co-tenant connection handle — the thin wrapper over the
    /// tokio-postgres client that the postgres registry twin (and any
    /// later pg co-tenant consumers) hold: it connects the alkgit-owned
    /// connection beside the engine-owned schemas on the same database
    /// (alkblobs ADR-013 §3's composition posture), drives its migration
    /// run at construction, and hands out the client for the twin's SQL.
    ///
    /// Natively async: the driver's client is `Send + Sync` and every
    /// surface is a straight `.await` (the engine's ADR-004/007 posture,
    /// inherited — no `spawn_blocking` bridge here). The client is
    /// shared under a tokio `tokio::sync::Mutex`; the guard is
    /// short-held — one statement batch or transaction per hold — and
    /// writers serialize further inside each mutation transaction on the
    /// `registry_seq` row lock.
    #[derive(Clone)]
    pub struct CoTenantPg {
        client: Arc<tokio::sync::Mutex<tokio_postgres::Client>>,
        schema: Arc<str>,
    }

    impl CoTenantPg {
        /// Open the co-tenant connection to the engine's database at
        /// `config` (a tokio-postgres connection string) and run the
        /// migrations — loud ([`OpenError`]): a half-migrated store
        /// refuses service. An empty `schema` means [`DEFAULT_SCHEMA`].
        /// The connection driver task is spawned by the handle; its
        /// lifetime is the handle's.
        pub async fn open(config: &str, schema: &str) -> Result<Self, OpenError> {
            let schema = validated_schema(schema)?;
            let cfg = config
                .parse::<tokio_postgres::Config>()
                .map_err(|e| OpenError::Engine(format!("co-tenant postgres config: {e}")))?;
            let (client, conn) = cfg
                .connect(NoTls)
                .await
                .map_err(|e| OpenError::Engine(format!("co-tenant postgres connect: {e}")))?;
            tokio::spawn(async move {
                let _ = conn.await;
            });
            let mut client = client;
            migrate(&mut client, &schema).await?;
            Ok(CoTenantPg {
                client: Arc::new(tokio::sync::Mutex::new(client)),
                schema,
            })
        }

        /// The co-tenant schema name ([`DEFAULT_SCHEMA`] when opened
        /// empty; qualified SQL composes through [`quote_identifier`]).
        pub fn schema(&self) -> &str {
            &self.schema
        }

        /// Hold the co-tenant client for SQL work. The guard is
        /// short-held — one statement batch or transaction per hold —
        /// awaited (the async model, never held across an unrelated
        /// mutation).
        pub async fn lock(&self) -> tokio::sync::MutexGuard<'_, tokio_postgres::Client> {
            self.client.lock().await
        }

        /// The store's current co-tenant schema stamp — the version
        /// support and diagnostics report (an ADR-024 consequence: a real
        /// version to report). 0 means no step applied yet; a value
        /// beyond the shipped step set is the safe unknown-forward
        /// coexistence shape (ADR-024 §4).
        pub async fn stamp(&self) -> Result<i64, OpenError> {
            let read = format!(
                "SELECT stamp FROM {} WHERE id = 1",
                qualify(&self.schema, STAMP_TABLE)
            );
            let guard = self.client.lock().await;
            stamp_of(&guard, &read).await
        }

        /// The shipped step count — the highest stamp this binary's step
        /// set owns (the honest coexistence datum beside
        /// [`CoTenantPg::stamp`]).
        pub fn step_count() -> usize {
            STEP_VERSIONS.len()
        }
    }

    /// The row's integer read for the stamp (and every other single
    /// integer the runner reads).
    async fn stamp_of<R: tokio_postgres::ToStatement>(
        client: &tokio_postgres::Client,
        read: &R,
    ) -> Result<i64, OpenError> {
        client
            .query_one(read, &[])
            .await
            .map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("stamp read ({STAMP_TABLE}): {e}"),
            })
            .and_then(|row| {
                row.try_get::<_, i64>(0).map_err(|e| OpenError::Migration {
                    step: 0,
                    reason: format!("stamp read ({STAMP_TABLE}): {e}"),
                })
            })
    }

    /// One ordered pg migration step: its version (the stamp value it
    /// marks), its monomorphic SQL applied inside one transaction, plus
    /// its presence-predicate table set — the race-verification re-key
    /// (the alkstore ADR-012 §5 pattern): on a step failure, the set's
    /// presence decides swallow-vs-propagate.
    pub(crate) struct PgStep {
        pub(crate) version: i64,
        name: &'static str,
        sql: String,
        tables: &'static [&'static str],
    }

    /// The pg step set for the schema. Step 1 is the initial co-tenant
    /// schema — the engine-neutral logical DDL of the module doc, in
    /// pg SQL (BIGINT sequence columns, CHECK'd visibility text).
    /// Step order monotone with the sqlite set ([`STEP_VERSIONS`]);
    /// post-publish additions append — additive-first (ADR-024 §4).
    pub(crate) fn build_steps(schema: &str) -> Vec<PgStep> {
        let q = |table: &str| qualify(schema, table);
        vec![PgStep {
            version: 1,
            name:
                "initial co-tenant schema (records, refs, seq, mutation markers, LFS liveness rows)",
            sql: format!(
                "CREATE TABLE {} (
                     repo_id        TEXT PRIMARY KEY,
                     visibility     TEXT NOT NULL CHECK (visibility IN ('public', 'private')),
                     grants         TEXT NOT NULL,
                     seq            BIGINT NOT NULL
                 );
                 CREATE TABLE {} (
                     repo_id        TEXT NOT NULL,
                     name           TEXT NOT NULL,
                     oid_hex        TEXT,
                     symref_target  TEXT,
                     PRIMARY KEY (repo_id, name)
                 );
                 CREATE TABLE {} (
                     id    INTEGER PRIMARY KEY CHECK (id = 1),
                     value BIGINT NOT NULL
                 );
                 INSERT INTO {} (id, value) VALUES (1, 0);
                 CREATE TABLE {} (
                     repo_id  TEXT PRIMARY KEY,
                     seq      BIGINT NOT NULL
                 );
                 CREATE TABLE {} (
                     repo_id   TEXT NOT NULL,
                     oid       TEXT NOT NULL,
                     size      BIGINT NOT NULL,
                     refcount  BIGINT NOT NULL,
                     PRIMARY KEY (repo_id, oid)
                 );",
                q("git_repos"),
                q("git_refs"),
                q("registry_seq"),
                q("registry_seq"),
                q("repo_mutation_markers"),
                q("lfs_objects"),
            ),
            tables: &[
                "git_repos",
                "git_refs",
                "registry_seq",
                "repo_mutation_markers",
                "lfs_objects",
            ],
        }]
    }

    /// The stamp table's idempotent bootstrap: create the schema and the
    /// stamp table if absent, seed the single stamp row at 0 (absent
    /// stamp ⇔ fresh store). One transaction under the schema's advisory
    /// migration lock ([`migration_lock_key`]). On failure the re-key
    /// verifies state against committed data: the effect present ⇒ a
    /// benign race swallowed; absent ⇒ propagate loud. The advisory
    /// lock releases at the transaction's end.
    async fn bootstrap_stamp_table(
        client: &mut tokio_postgres::Client,
        schema: &str,
    ) -> Result<(), OpenError> {
        let stamp_table = qualify(schema, STAMP_TABLE);
        let bootstrap = format!(
            "CREATE SCHEMA IF NOT EXISTS {};
             CREATE TABLE IF NOT EXISTS {} (
                 id    INTEGER PRIMARY KEY CHECK (id = 1),
                 stamp BIGINT NOT NULL
             );
             INSERT INTO {} (id, stamp) VALUES (1, 0) ON CONFLICT (id) DO NOTHING;",
            quote_identifier(schema),
            stamp_table,
            stamp_table,
        );
        let ctx = || format!("stamp-table bootstrap ({STAMP_TABLE})");
        let tx = client
            .transaction()
            .await
            .map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("{}: BEGIN: {e}", ctx()),
            })?;
        tx.execute(
            "SELECT pg_advisory_xact_lock($1)",
            &[&migration_lock_key(schema)],
        )
        .await
        .map_err(|e| OpenError::Migration {
            step: 0,
            reason: format!("{}: {e}", ctx()),
        })?;
        match tx.batch_execute(&bootstrap).await {
            Ok(()) => tx.commit().await.map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("{}: commit: {e}", ctx()),
            }),
            Err(e) => {
                drop(tx);
                if stamp_table_present(client, schema).await? {
                    return Ok(());
                }
                Err(OpenError::Migration {
                    step: 0,
                    reason: format!("{}: {e}", ctx()),
                })
            }
        }
    }

    /// The stamp table's presence — the bootstrap's re-key.
    async fn stamp_table_present(
        client: &tokio_postgres::Client,
        schema: &str,
    ) -> Result<bool, OpenError> {
        client
            .query_one(
                "SELECT to_regclass($1) IS NOT NULL",
                &[&format!("{schema}.{STAMP_TABLE}")],
            )
            .await
            .map(|row| row.get::<_, bool>(0))
            .map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("stamp-table presence query ({STAMP_TABLE}): {e}"),
            })
    }

    /// The step's presence predicate: truthy exactly when the step's
    /// complete table set is present in the schema's catalog. A query
    /// failure is loud (unverifiable state must not be swallowed).
    async fn step_effect_present(
        client: &tokio_postgres::Client,
        schema: &str,
        step: &PgStep,
    ) -> Result<bool, OpenError> {
        client
            .query_one(
                "SELECT (SELECT count(*) FROM pg_class c \
                 JOIN pg_namespace n ON n.oid = c.relnamespace \
                 WHERE n.nspname = $1 AND c.relname = ANY($2)) = $3",
                &[&schema, &step.tables, &(step.tables.len() as i64)],
            )
            .await
            .map(|row| row.get::<_, bool>(0))
            .map_err(|e| OpenError::Migration {
                step: step.version as usize,
                reason: format!(
                    "step {} ({}): verification query: {e}",
                    step.version, step.name
                ),
            })
    }

    /// The construction-time migration run: the stamp-table bootstrap
    /// plus every pending step, loud on unverified failure. Returns the
    /// post-run stamp.
    async fn migrate(client: &mut tokio_postgres::Client, schema: &str) -> Result<i64, OpenError> {
        bootstrap_stamp_table(client, schema).await?;
        for step in build_steps(schema) {
            apply_step(client, schema, &step).await?;
        }
        let read = format!(
            "SELECT stamp FROM {} WHERE id = 1",
            qualify(schema, STAMP_TABLE)
        );
        client
            .query_one(&read, &[])
            .await
            .map(|row| row.get::<_, i64>(0))
            .map_err(|e| OpenError::Migration {
                step: 0,
                reason: format!("stamp read ({STAMP_TABLE}) in {schema}: {e}"),
            })
    }

    /// Apply one step if pending. The stamp is read inside the step's
    /// own transaction, under the schema's advisory migration lock —
    /// concurrent openers serialize; a loser sees the winner's committed
    /// stamp and skips without ever executing duplicate DDL. A step whose
    /// SQL nevertheless fails verifies state against committed data (the
    /// re-key): the effect present ⇒ benign race swallowed (the stamp
    /// healed when the concurrent applier's own advance has not landed);
    /// absent ⇒ propagate loud.
    async fn apply_step(
        client: &mut tokio_postgres::Client,
        schema: &str,
        step: &PgStep,
    ) -> Result<(), OpenError> {
        let step_no = step.version as usize;
        let stamp_read = format!(
            "SELECT stamp FROM {} WHERE id = 1",
            qualify(schema, STAMP_TABLE)
        );
        let stamp_advance = format!(
            "UPDATE {} SET stamp = $1 WHERE id = 1",
            qualify(schema, STAMP_TABLE)
        );
        let stamp_heal = format!(
            "UPDATE {} SET stamp = $1 WHERE id = 1 AND stamp < $1",
            qualify(schema, STAMP_TABLE)
        );
        let step_ctx = || format!("step {step_no} ({})", step.name);
        let tx = client
            .transaction()
            .await
            .map_err(|e| OpenError::Migration {
                step: step_no,
                reason: format!("{}: BEGIN: {e}", step_ctx()),
            })?;
        tx.execute(
            "SELECT pg_advisory_xact_lock($1)",
            &[&migration_lock_key(schema)],
        )
        .await
        .map_err(|e| OpenError::Migration {
            step: step_no,
            reason: format!("{}: {e}", step_ctx()),
        })?;
        let stamp_now: i64 = tx
            .query_one(&stamp_read, &[])
            .await
            .map_err(|e| OpenError::Migration {
                step: step_no,
                reason: format!("{}: stamp read: {e}", step_ctx()),
            })?
            .try_get(0)
            .map_err(|e| OpenError::Migration {
                step: step_no,
                reason: format!("{}: stamp read: {e}", step_ctx()),
            })?;
        if stamp_now >= step.version {
            tx.commit().await.ok();
            return Ok(());
        }
        let outcome = tx.batch_execute(&step.sql).await;
        let outcome = match outcome {
            Ok(()) => tx.execute(&stamp_advance, &[&step.version]).await,
            Err(e) => Err(e),
        };
        match outcome {
            Ok(_) => tx.commit().await.map_err(|e| OpenError::Migration {
                step: step_no,
                reason: format!("{}: commit: {e}", step_ctx()),
            }),
            Err(e) => {
                drop(tx);
                if step_effect_present(client, schema, step).await? {
                    client
                        .execute(&stamp_heal, &[&step.version])
                        .await
                        .map_err(|e| OpenError::Migration {
                            step: step_no,
                            reason: format!("{}: race-swallow stamp heal: {e}", step_ctx()),
                        })?;
                    return Ok(());
                }
                Err(OpenError::Migration {
                    step: step_no,
                    reason: format!("{}: {e}", step_ctx()),
                })
            }
        }
    }
}

#[cfg(feature = "postgres")]
pub use pg_twin::{CoTenantPg, DEFAULT_SCHEMA};

#[cfg(feature = "postgres")]
pub(crate) use pg_twin::{git_refs_lock_key, lfs_objects_lock_key, qualify};

#[cfg(all(test, feature = "sqlite"))]
mod tests {
    use super::sqlite_twin::{
        failing_step_probe, read_stamp, run_steps, step_probe, BUSY_TIMEOUT_MS,
        STAMP_TABLE_BOOTSTRAP,
    };
    use super::*;
    use crate::registry::{InMemoryRegistry, RegistryRead, RegistryWrite};
    use crate::testutil::{mem_registry_pair, unique_db_path};

    fn columns_of(conn: &rusqlite::Connection, table: &str) -> Vec<String> {
        let mut stmt = conn
            .prepare(&format!("PRAGMA table_info({table})"))
            .unwrap();
        stmt.query_map([], |r| r.get::<_, String>(1))
            .unwrap()
            .map(|c| c.unwrap())
            .collect()
    }

    fn table_exists(conn: &rusqlite::Connection, table: &str) -> bool {
        conn.query_row(
            "SELECT count(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
            [table],
            |r| r.get::<_, i64>(0),
        )
        .unwrap()
            > 0
    }

    #[test]
    fn fresh_create_stamps_one_and_builds_the_full_co_tenant_schema() {
        let path = unique_db_path("migrations", "fresh-create");
        let handle = CoTenantSqlite::open(&path).unwrap();
        assert_eq!(handle.stamp().unwrap(), 1);
        assert_eq!(CoTenantSqlite::step_count(), 1);

        let db = handle.lock();
        assert_eq!(
            db.query_row("PRAGMA busy_timeout", [], |r| r.get::<_, i64>(0))
                .unwrap(),
            BUSY_TIMEOUT_MS,
            "the pinned busy timeout rides the co-tenant connection"
        );
        assert_eq!(
            db.query_row("PRAGMA journal_mode", [], |r| r.get::<_, String>(0))
                .unwrap(),
            "wal"
        );
        assert_eq!(
            db.query_row("SELECT value FROM registry_seq WHERE id = 1", [], |r| r
                .get::<_, i64>(0))
                .unwrap(),
            0,
            "the watermark singleton seeds at 0 on a fresh store"
        );
        for table in [
            "git_repos",
            "git_refs",
            "registry_seq",
            "repo_mutation_markers",
            "lfs_objects",
        ] {
            assert!(table_exists(&db, table));
        }
        assert_eq!(
            columns_of(&db, "git_repos"),
            vec![
                "repo_id".to_string(),
                "visibility".to_string(),
                "grants".to_string(),
                "seq".to_string()
            ],
            "the record rows carry the RepoRecord serde shape's members plus the mutation seq"
        );
        assert_eq!(
            columns_of(&db, "git_refs"),
            vec![
                "repo_id".to_string(),
                "name".to_string(),
                "oid_hex".to_string(),
                "symref_target".to_string()
            ],
            "HEAD lands as the symref row shape: oid_hex NULL + symref_target set"
        );
        assert_eq!(
            columns_of(&db, "lfs_objects"),
            vec![
                "repo_id".to_string(),
                "oid".to_string(),
                "size".to_string(),
                "refcount".to_string()
            ],
            "the LFS liveness rows carry the day-one refcount column (ADR-024 §5)"
        );
        assert_eq!(
            columns_of(&db, "repo_mutation_markers"),
            vec!["repo_id".to_string(), "seq".to_string()],
        );
        drop(db);

        let reopened = rusqlite::Connection::open(&path).expect("second opener on one file");
        assert_eq!(read_stamp(&reopened).unwrap(), 1, "the stamp is durable");
        assert!(
            table_exists(&reopened, "git_repos"),
            "the schema is durable"
        );
        drop(reopened);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn fresh_create_pins_the_visibility_domain() {
        let path = unique_db_path("migrations", "visibility-check");
        let handle = CoTenantSqlite::open(&path).unwrap();
        let db = handle.lock();
        assert!(
            db.execute(
                "INSERT INTO git_repos (repo_id, visibility, grants, seq) \
                 VALUES ('r', 'weird', '{}', 1)",
                []
            )
            .is_err(),
            "the visibility CHECK pins the record domain at the table"
        );
        drop(db);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn additive_append_applies_step_two_then_is_idempotent() {
        let mut db = rusqlite::Connection::open_in_memory().unwrap();
        let steps = vec![
            step_probe("m_probe_one", "probe one", 1),
            step_probe("m_probe_two", "probe two — additive append", 2),
        ];
        let stamp = run_steps(&mut db, &steps).unwrap();
        assert_eq!(stamp, 2, "both steps applied and stamped");
        assert!(table_exists(&db, "m_probe_one"));
        assert!(table_exists(&db, "m_probe_two"));

        let again = run_steps(&mut db, &steps).unwrap();
        assert_eq!(again, 2, "a re-run is a no-op — steps skip on the stamp");
        assert_eq!(
            db.query_row(
                "SELECT count(*) FROM alkgit_schema_stamp WHERE id = 1",
                [],
                |r| r.get::<_, i64>(0)
            )
            .unwrap(),
            1,
            "one stamp row, always"
        );

        let appended = vec![step_probe(
            "m_probe_three",
            "probe three — appended later",
            3,
        )];
        let stamp = run_steps(&mut db, &appended).unwrap();
        assert_eq!(stamp, 3, "an additive append advances the stamp one step");
        assert!(table_exists(&db, "m_probe_three"));
    }

    #[test]
    fn concurrent_open_race_on_one_file_swallows_the_benign_race() {
        let path = unique_db_path("migrations", "concurrent-open");
        let stamps = std::thread::scope(|scope| {
            let mut joins = Vec::new();
            for _ in 0..4 {
                joins.push(scope.spawn(|| {
                    CoTenantSqlite::open(&path)
                        .map(|handle| handle.stamp().expect("stamped handle reads"))
                }));
            }
            joins
                .into_iter()
                .map(|join| join.join().expect("no opener thread panics"))
                .collect::<Vec<_>>()
        });
        for stamp in &stamps {
            assert_eq!(
                stamp.as_ref().expect("every opener succeeds"),
                &1,
                "the benign race is swallowed — all openers converge on the stamped schema"
            );
        }
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn a_pre_applied_effect_swallows_and_heals_the_stamp() {
        let mut db = rusqlite::Connection::open_in_memory().unwrap();
        db.execute_batch(STAMP_TABLE_BOOTSTRAP).unwrap();
        db.execute("CREATE TABLE m_probe_one (x TEXT, y TEXT)", [])
            .unwrap();
        assert_eq!(
            read_stamp(&db).unwrap(),
            0,
            "the concurrent applier's effect is present, its stamp write not landed"
        );

        let steps = vec![step_probe("m_probe_one", "probe one", 1)];
        run_steps(&mut db, &steps).unwrap();
        assert_eq!(
            read_stamp(&db).unwrap(),
            1,
            "the race swallowed on the presence re-key and healed the stamp"
        );
        assert!(table_exists(&db, "m_probe_one"));

        let again = run_steps(&mut db, &steps).unwrap();
        assert_eq!(again, 1, "the healed store is idempotent on reopen");
    }

    #[test]
    fn a_failing_step_fails_loudly_and_propagates() {
        let path = unique_db_path("migrations", "loud-failure");
        let mut db = rusqlite::Connection::open(&path).unwrap();
        let steps = vec![
            step_probe("m_probe_good", "probe good", 1),
            failing_step_probe("m_probe_bad", "probe bad — fails inside its transaction", 2),
            step_probe("m_probe_after", "probe after — never reached", 3),
        ];
        let err = run_steps(&mut db, &steps).unwrap_err();
        assert!(matches!(err, OpenError::Migration { step: 2, .. }));
        match &err {
            OpenError::Migration { reason, .. } => {
                assert!(reason.contains("already exists"), "{reason}");
            }
            other => panic!("unexpected variant: {other:?}"),
        }
        assert_eq!(
            read_stamp(&db).unwrap(),
            1,
            "step 1 committed, step 2 did not"
        );
        assert!(table_exists(&db, "m_probe_good"));
        assert!(
            !table_exists(&db, "m_probe_bad"),
            "the failed tx rolled back whole"
        );
        assert!(!table_exists(&db, "m_probe_after"), "later steps never ran");
        drop(db);

        let mut retry = rusqlite::Connection::open(&path).unwrap();
        let retry_err = run_steps(&mut retry, &steps).expect_err("the run fails again");
        assert!(
            matches!(retry_err, OpenError::Migration { step: 2, .. }),
            "the half-migrated store refuses service again — loud on every construction"
        );
        drop(retry);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn an_unknown_forward_stamp_is_safe_to_open_against() {
        let path = unique_db_path("migrations", "forward-stamp");
        let handle = CoTenantSqlite::open(&path).unwrap();
        {
            let db = handle.lock();
            db.execute("UPDATE alkgit_schema_stamp SET stamp = 99 WHERE id = 1", [])
                .unwrap();
        }
        let reopened = CoTenantSqlite::open(&path).unwrap();
        assert_eq!(
            reopened.stamp().unwrap(),
            99,
            "no refuse-gate ships in v1 — the forward stamp is left untouched"
        );
        assert!(
            table_exists(&reopened.lock(), "git_repos"),
            "the schema is intact"
        );

        let mut db = rusqlite::Connection::open_in_memory().unwrap();
        let steps = vec![step_probe("m_probe_one", "probe one", 1)];
        run_steps(&mut db, &steps).unwrap();
        db.execute("UPDATE alkgit_schema_stamp SET stamp = 42 WHERE id = 1", [])
            .unwrap();
        let forward_steps = vec![
            step_probe("m_probe_two", "probe two", 2),
            step_probe("m_probe_three", "probe three", 3),
        ];
        let stamp = run_steps(&mut db, &forward_steps).unwrap();
        assert_eq!(stamp, 42, "pending steps skip on the forward stamp");
        assert!(
            !table_exists(&db, "m_probe_two"),
            "nothing applied ahead of the stamp"
        );
        drop(handle);
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn memory_mode_is_rejected_at_the_handle() {
        assert!(
            matches!(CoTenantSqlite::open(":memory:"), Err(OpenError::Invalid(_))),
            ":memory: confers no co-tenancy — the handle rejects the shape (DQ-7)"
        );
    }

    #[tokio::test]
    async fn in_memory_adapter_is_stamp_free_fresh_on_every_construction() {
        let record = alkgit::RepoRecord {
            repo_id: "a".to_string(),
            visibility: alkgit::Visibility::Public,
            grants: Default::default(),
        };

        let first = InMemoryRegistry::new();
        assert_eq!(first.put(&record).await.unwrap(), 1);

        let second = InMemoryRegistry::new();
        assert_eq!(
            second.resolve("a"),
            None,
            "a fresh construction builds the current state empty — nothing durable is versioned"
        );
        assert_eq!(second.watermark().await.unwrap(), 0);

        let (_, store_one) = mem_registry_pair();
        let (_, store_two) = mem_registry_pair();
        store_one.put_repo(&record).await.unwrap();
        assert_eq!(
            store_two.resolve("a").await.unwrap_err(),
            alkgit::RegistryError::NotFound,
            "two in-memory constructions are independent stores, not two views of one"
        );
    }
}