alkblobs/tests/conformance/fleet_column.rs

//! The **fleet column** — the suite's third SKIP-posture column: the
//! exact-count fleet sweep outcomes under concurrency, run against
//! live postgres. This is the wave-6 gate (docs/plans/implementation.md
//! §Wave 6): INV-4's **engine-gated fleet form** (store-api.md
//! §Invariants 4; ADR-008 §2; ADR-012 §1/§4; ADR-013 §2) — the
//! architecture's most intricate state machine (the fleet GC protocol:
//! the joint entry+pin tx, the pin rows' TTL/renewal/expiry lives, the
//! sweep lease, the SQL-level arbitration, the maintenance step, the
//! fleet liveness verb) held under two store instances over one shared
//! pool.
//!
//! The column runs under `--features postgres` — fleet mode requires
//! kv = postgres, the composite fleet-validity predicate's kv clause
//! (ADR-012 §3) — and rides the SKIP posture: every runner gates behind
//! the reachability probe (`crate::postgres_column::harness_ready`,
//! the postgres column's probe — the column reuses the posture's
//! instrument), server-less runs skip cleanly, and acceptance is green
//! **against the harness server** (dockerized `pglo-poc`,
//! `postgres:16-alpine`, :15432, db `blobs`).
//!
//! **The fleet factory** (the postgres column's schema-per-open
//! pattern adapted to the fleet shape; README §The factory contract):
//! `open_fleet()` mints **one schema** and vends **two `Store`
//! instances over it**, each `fleet: true` (the `_fleet` constructors,
//! each carrying its own store-instance id — the instances share the
//! pool; the pin rows compose by owner, `PinRow`'s `(key, owner)` row
//! identity, ADR-011/ADR-012 §3's fleet definition). The dual-tier
//! fleet factory gives each instance its own private local root (the
//! large tier rides the operator's [`FleetSharedMedia`::asserted]
//! declaration — the factory owns the isolated-medium form of it); the
//! kv-only fleet factory is the **re-routing posture's client shape**
//! (ADR-008 §3) and the consolidation fleet factory (kv = postgres +
//! large = pg-lo, the same pg instance) is gated on the `postgres` +
//! `pg-lo` feature pair. Teardown drops exactly the minted schemas
//! (`DROP SCHEMA … CASCADE`, the postgres column's registry pattern —
//! idempotent, owned, panicking-row-safe), removes every vended temp
//! root, and — the consolidation shape — reaps the dedicated
//! database's LOs before dropping it (`WITH (FORCE)`, the pg-lo
//! column's LO-isolation corollary).
//!
//! **The test tuning** (the column declares its tuning, the constructor
//! parameters' purpose): window delay `Duration::ZERO`
//! (execute-on-next-sweep — the durable schedule's degenerate legal
//! tuning) and a **short test pin TTL** ([`TEST_TTL`], the fleet
//! constructors' `pin_ttl` parameter — the renewal/expiry rows need
//! observable cadences; the TTL is constructor configuration, never
//! API, ADR-008 §2). Where a row needs ordering the window delay and
//! the TTL tunings provide it; the waits the rows take are the expiry
//! mechanism's own cadence (deterministic by construction — a dropped
//! holder's row expires at its minted TTL) or the independent
//! connection's manual-expiry instrument (the abandonment the
//! maintenance step reaps — the same instrument the fleet engine
//! suite's tests carry). No sleep-based race assertions: the
//! lease/arbitration outcomes are decided by the server, not by timing.
//!
//! The concurrency rows are the acceptance here (the wave's gate):
//! the **two-instance exact-count shape** (instance A pins, instance B
//! sweeps — A's row cancels B's candidate at the execution-time SQL
//! re-arbitration), the **lease contention row** (two instances sweep
//! concurrently — exactly one sweeps, the other aborts `SweeperLock`
//! as report data with nothing deleted), the **cross-instance
//! arbitration row** (a foreign owner's un-expired pin row refuses a
//! direct delete — `GcRefuse` — and stops after expiry), the **race
//! across the window delay in its fleet form** (a put landing during
//! the delay — its row committed before its entry is visible —
//! survives the next sweep's execution-time re-arbitration), the
//! **maintenance-step row** (expired pin rows reaped exactly; un-
//! expired rows survive), and the **fleet liveness row** (a test-owned
//! embedder table's rows protect through the registration verb — the
//! schema-ownership rule: the engine never created the table). The
//! joint-tx rows, the TTL/renewal rows (through the public `put` and
//! the pin guard), the fleet-validity rows (the predicate at the
//! constructor), and the kv-only re-routing row (the reference set's
//! rejection row mirrored verbatim against the fleet kv-only factory)
//! complete the column.

#![allow(dead_code)]

use std::io::Cursor;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use tokio_postgres::NoTls;

use crate::factory::StoreFactory;
use crate::postgres_column::harness_ready;
use crate::rows::{committed_file_count, patterned, LatchProtect, SetSource};
use alkblobs::{
    Error, FleetSharedMedia, GcAbortCause, Key, LivenessRegistration, PgOpts, Store, StoreMode,
};

// ------------------------------------------------------------------
// The column's declared tuning
// ------------------------------------------------------------------

/// The column's declared dual-tier threshold — small, so the rows'
/// contents land sub-threshold (kv tier: the pg schema's pool rows;
/// the pin rows' medium) and the consolidation row's large entry lands
/// over it.
pub(crate) const TUNED_THRESHOLD: u64 = 1024;

/// The column's declared kv-only rejection threshold — small, so every
/// leg of the mirrored rejection row stays trivially materializable.
pub(crate) const KV_ONLY_THRESHOLD: u64 = 1024;

/// The **test tuning of the window delay** (ADR-008 §2's legal
/// degenerate tuning): `0` — execute-on-next-sweep. The fleet sweep's
/// staging sweeps the dead set; the *next* sweep is its executor —
/// exactly the schedule the exact-count rows assert against.
pub(crate) const WINDOW_DELAY: Duration = Duration::ZERO;

/// The **short test pin TTL** — the fleet constructors' `pin_ttl`
/// parameter, the column's declared tuning (ADR-008 §2: fleet
/// constructor parameters, not API). Short enough that the TTL/3
/// cadence (≈300 ms) and the expiry horizon are observable within a
/// test's runtime; long enough that a held guard's renewing row never
/// lapses under a sweep.
pub(crate) const TEST_TTL: Duration = Duration::from_millis(900);
/// One renewal cadence plus margin: any interval this long admits at
/// least one TTL/3 tick (TEST_TTL / 3 ≈ 300 ms).
pub(crate) const ONE_CADENCE: Duration = Duration::from_millis(600);
/// Beyond the short TTL: a renewal stopped at its guard's drop leaves
/// the row expired at every read this long after.
pub(crate) const EXPIRY_HORIZON: Duration = Duration::from_millis(1300);

fn wall_nanos() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as i64)
        .unwrap_or(0)
}

// ------------------------------------------------------------------
// The factories — one minted schema, two (or more) `Store` instances
// over it, each `fleet: true`
// ------------------------------------------------------------------

/// A fresh engine-owned schema name — unique per call, seeded per
/// (tag, pid, nanos, factory-internal sequence) so two factory
/// instances never share a schema, racing opens included (the postgres
/// column's naming policy).
fn fresh_schema_name(tag: &str, seq: &AtomicU64) -> String {
    let seq = seq.fetch_add(1, Ordering::Relaxed);
    format!(
        "alkblobs_fleet_{}_{}_{}_{}",
        tag,
        std::process::id(),
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0),
        seq,
    )
}

fn seq_seed() -> u64 {
    (std::time::SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.subsec_nanos() as u64)
        .unwrap_or(0))
        ^ ((std::process::id() as u64) << 8)
}

/// The factories' engine-opts shape: exactly the minted schema name and
/// the ship defaults for the rest (the alkstore factory's policy — only
/// the schema name is factory policy).
fn opts(schema: String) -> PgOpts {
    PgOpts {
        schema,
        ..PgOpts::default()
    }
}

/// A fresh private temp directory holding one instance's local root.
fn fresh_root(tag: &str, seq: &AtomicU64) -> PathBuf {
    let seq = seq.fetch_add(1, Ordering::Relaxed);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let base = std::env::temp_dir().join(format!(
        "alkblobs-fleet-column-{tag}-{}-{nanos:x}-{seq}",
        std::process::id(),
    ));
    std::fs::create_dir_all(&base).expect("temp site created");
    base
}

fn remove_dir(base: &Path) -> Result<(), Error> {
    match std::fs::remove_dir_all(base) {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(Error::Io(e.to_string())),
    }
}

/// The shared backing site of one open's fleet: the engine-owned
/// schema this open minted (the kv tier's medium, the pin rows', the
/// window's, the sweep lease's, and the advisory lock's — one shared
/// engine over one schema) and the per-instance local roots vended
/// with it.
#[derive(Clone)]
pub(crate) struct FleetSite {
    /// The server's DSN — the oracle's independent connections address
    /// the shared medium through it.
    dsn: String,
    /// The minted schema — the fleet's shared pool (the pin rows
    /// compose by owner inside it).
    schema: String,
    /// The per-instance local roots (the dual-tier fleet's large-tier
    /// media; empty on the kv-only factory).
    roots: Vec<PathBuf>,
}

impl FleetSite {
    pub(crate) fn schema(&self) -> &str {
        &self.schema
    }

    pub(crate) fn dsn(&self) -> &str {
        &self.dsn
    }

    pub(crate) fn root(&self, instance: usize) -> &Path {
        &self.roots[instance]
    }
}

/// The fleet dual-tier factory: `kv = postgres` + `large = local` at
/// the column's tuned threshold, `fleet: true` — one minted schema,
/// two vended instances over it, each with its own store-instance id
/// (the pin rows compose by owner) and its own private local root (the
/// operator's shared-media declaration asserted at each call site).
///
/// **The fleet factory contract** (README §The factory contract, the
/// postgres column's pattern adapted to the fleet shape): one minted
/// schema per `open_fleet()`, tracked per-instance in the schema
/// registry; every vended instance shares the pool (the fleet's
/// definition — two store instances of one store shape over one shared
/// engine); teardown `DROP SCHEMA IF EXISTS … CASCADE` over exactly
/// the minted schemas plus removal of every vended root (idempotent,
/// owned, panicking-row-safe — each runner constructs its own factory
/// instance and owns its lifecycle, so concurrent rows never touch
/// each other's schemas).
pub(crate) struct FleetDualTierFactory {
    dsn: String,
    /// The declared dispatch threshold — the routing boundary the
    /// rows' contents sit against (the shared matrix lands sub-
    /// threshold: kv tier, the pin rows' medium).
    threshold: u64,
    schema_seq: AtomicU64,
    schemas: Mutex<Vec<String>>,
    sites: Mutex<Vec<PathBuf>>,
}

impl FleetDualTierFactory {
    pub(crate) fn new(threshold: u64) -> FleetDualTierFactory {
        FleetDualTierFactory {
            dsn: crate::postgres_column::harness_dsn()
                .expect("the harness server is configured for the column"),
            threshold,
            schema_seq: AtomicU64::new(seq_seed()),
            schemas: Mutex::new(Vec::new()),
            sites: Mutex::new(Vec::new()),
        }
    }

    /// The declared dispatch threshold, for the rows that size their
    /// contents against it.
    pub(crate) fn threshold(&self) -> u64 {
        self.threshold
    }

    /// Mint one schema, vend two fleet instances over it (each with
    /// its own store-instance id and its own private local root), and
    /// track the site: the column's rows' instrument.
    pub(crate) async fn open_fleet(&self) -> Result<(Store, Store, FleetSite), Error> {
        let schema = fresh_schema_name("dual-tier", &self.schema_seq);
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(schema.clone());
        let base_a = fresh_root("dual-tier-a", &self.schema_seq);
        let base_b = fresh_root("dual-tier-b", &self.schema_seq);
        let root_a = base_a.join("local-root");
        let root_b = base_b.join("local-root");
        let store_a = Store::dual_tier_postgres_local_fleet(
            &self.dsn,
            opts(schema.clone()),
            root_a.clone(),
            self.threshold,
            WINDOW_DELAY,
            TEST_TTL,
            FleetSharedMedia::asserted(),
        )
        .await?;
        let store_b = Store::dual_tier_postgres_local_fleet(
            &self.dsn,
            opts(schema.clone()),
            root_b.clone(),
            self.threshold,
            WINDOW_DELAY,
            TEST_TTL,
            FleetSharedMedia::asserted(),
        )
        .await?;
        self.sites
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .extend([base_a, base_b]);
        Ok((
            store_a,
            store_b,
            FleetSite {
                dsn: self.dsn.clone(),
                schema,
                roots: vec![root_a, root_b],
            },
        ))
    }

    pub(crate) fn tracked_schemas(&self) -> Vec<String> {
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    /// The DSN the factory opens against — the structural tests'
    /// probe target.
    pub(crate) fn dsn(&self) -> &str {
        &self.dsn
    }

    pub(crate) async fn teardown(&self) -> Result<(), Error> {
        let schemas = self
            .schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        let sites = self
            .sites
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        if !schemas.is_empty() {
            let client = admin_connect(&self.dsn).await?;
            for schema in &schemas {
                let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE");
                client
                    .batch_execute(&sql)
                    .await
                    .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
            }
        }
        for base in &sites {
            remove_dir(base)?;
        }
        Ok(())
    }
}

/// The fleet kv-only factory: no large tier — the **re-routing
/// posture's client shape** (ADR-008 §3), `fleet: true`, the column's
/// kv-only threshold. [`StoreFactory`]'s `open()` vends one isolated
/// fleet store over a freshly minted schema (the mirroring rejection
/// row's factory form); [`Self::open_fleet`] vends **two** instances
/// over one minted schema (the fleet contract's structural test's
/// shape).
pub(crate) struct FleetKvOnlyFactory {
    dsn: String,
    /// The rejection point this column's configuration declares — the
    /// mirrored row discovers it behaviorally (threshold-agnostic by
    /// discovery) and requires it materializable for the boundary legs.
    pub(crate) threshold: u64,
    schema_seq: AtomicU64,
    schemas: Mutex<Vec<String>>,
}

impl FleetKvOnlyFactory {
    pub(crate) fn new(threshold: u64) -> FleetKvOnlyFactory {
        FleetKvOnlyFactory {
            dsn: crate::postgres_column::harness_dsn()
                .expect("the harness server is configured for the column"),
            threshold,
            schema_seq: AtomicU64::new(seq_seed()),
            schemas: Mutex::new(Vec::new()),
        }
    }

    /// One minted schema, two fleet instances over it — the fleet
    /// factory contract's structural shape on this factory too.
    pub(crate) async fn open_fleet(&self) -> Result<(Store, Store, FleetSite), Error> {
        let schema = fresh_schema_name("kv-only", &self.schema_seq);
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(schema.clone());
        let store_a = Store::kv_only_postgres_fleet(
            &self.dsn,
            opts(schema.clone()),
            self.threshold,
            WINDOW_DELAY,
            TEST_TTL,
        )
        .await?;
        let store_b = Store::kv_only_postgres_fleet(
            &self.dsn,
            opts(schema.clone()),
            self.threshold,
            WINDOW_DELAY,
            TEST_TTL,
        )
        .await?;
        Ok((
            store_a,
            store_b,
            FleetSite {
                dsn: self.dsn.clone(),
                schema,
                roots: Vec::new(),
            },
        ))
    }

    pub(crate) fn tracked_schemas(&self) -> Vec<String> {
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    }

    pub(crate) fn dsn(&self) -> &str {
        &self.dsn
    }

    pub(crate) async fn teardown(&self) -> Result<(), Error> {
        let schemas = self
            .schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        if !schemas.is_empty() {
            let client = admin_connect(&self.dsn).await?;
            for schema in &schemas {
                let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE");
                client
                    .batch_execute(&sql)
                    .await
                    .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
            }
        }
        Ok(())
    }
}

impl StoreFactory for FleetKvOnlyFactory {
    fn name(&self) -> &'static str {
        "fleet kv-only postgres"
    }

    async fn open(&self) -> Result<Store, Error> {
        let schema = fresh_schema_name("kv-only", &self.schema_seq);
        self.schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .push(schema.clone());
        Store::kv_only_postgres_fleet(
            &self.dsn,
            opts(schema),
            self.threshold,
            WINDOW_DELAY,
            TEST_TTL,
        )
        .await
    }

    async fn teardown(&self) -> Result<(), Error> {
        self.teardown_impl().await
    }
}

impl FleetKvOnlyFactory {
    async fn teardown_impl(&self) -> Result<(), Error> {
        let schemas = self
            .schemas
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .drain(..)
            .collect::<Vec<_>>();
        if !schemas.is_empty() {
            let client = admin_connect(&self.dsn).await?;
            for schema in &schemas {
                let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE");
                client
                    .batch_execute(&sql)
                    .await
                    .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
            }
        }
        Ok(())
    }
}

// ------------------------------------------------------------------
// The medium oracle — the independent connection (outside every
// store's pool, seeing only durable state): the pool's rows, the pin
// rows' `(key, owner, expiry)` triples, the window's staged
// candidates, the schema/table existence probes, the liveness-table
// instruments, and the manual-expiry instrument (the abandonment the
// maintenance step reaps — the fleet engine suite's recorded shape).
// ------------------------------------------------------------------

async fn admin_connect(dsn: &str) -> Result<tokio_postgres::Client, Error> {
    let (client, connection) = tokio_postgres::connect(dsn, NoTls)
        .await
        .map_err(|e| Error::Io(format!("postgres harness: {e}")))?;
    tokio::spawn(async move {
        let _ = connection.await;
    });
    Ok(client)
}

/// One independent connection's probe — the durable-state instrument.
struct Probe {
    client: tokio_postgres::Client,
    _conn: tokio::task::JoinHandle<()>,
}

async fn connect_probe(dsn: &str) -> Probe {
    let (client, connection) = tokio_postgres::connect(dsn, NoTls)
        .await
        .expect("the independent connection connects");
    Probe {
        client,
        _conn: tokio::spawn(async move {
            let _ = connection.await;
        }),
    }
}

impl Probe {
    async fn scalar(&self, sql: &str) -> i64 {
        let row = self
            .client
            .query_one(sql, &[])
            .await
            .expect("the oracle query succeeds");
        row.get::<_, i64>(0)
    }

    /// The kv tier's pool residency — one row per committed kv-tier
    /// entry (the schema's `blobs` table, the shared pool's).
    async fn pool_rows(&self, schema: &str) -> usize {
        let sql = format!("SELECT COUNT(*) FROM \"{schema}\".blobs");
        self.scalar(&sql).await.max(0) as usize
    }

    /// The durable window's staged-candidate residency — the schedule
    /// observer (staged rows land here; the executing sweep's take
    /// empties them).
    async fn window_rows(&self, schema: &str) -> usize {
        let sql = format!("SELECT COUNT(*) FROM \"{schema}\".gc_window");
        self.scalar(&sql).await.max(0) as usize
    }

    /// The pin rows in their stored encoding, `(key, owner, expiry)`
    /// sorted by `(key, owner)` — the pool's durable protection set
    /// (cross-instance by construction: every instance's rows compose
    /// in this one table).
    async fn pin_rows(&self, schema: &str) -> Vec<(Vec<u8>, Vec<u8>, i64)> {
        let sql =
            format!("SELECT key, owner, expiry FROM \"{schema}\".gc_pins ORDER BY key, owner");
        let rows = self
            .client
            .query(sql.as_str(), &[])
            .await
            .expect("the oracle query succeeds");
        rows.iter()
            .map(|row| {
                (
                    row.get::<_, Vec<u8>>(0),
                    row.get::<_, Vec<u8>>(1),
                    row.get::<_, i64>(2),
                )
            })
            .collect()
    }

    /// Whether the engine-owned schema still exists server-side — the
    /// structural teardown tests' instrument (`pg_namespace`).
    async fn schema_exists(&self, schema: &str) -> bool {
        let count: i64 = self
            .client
            .query_one(
                "SELECT count(*) FROM pg_namespace WHERE nspname = $1",
                &[&schema],
            )
            .await
            .expect("the oracle query succeeds")
            .get(0);
        count > 0
    }

    /// Whether a table exists in a schema — the schema-ownership
    /// rule's instrument (the embedder's liveness table is the
    /// embedder's: the engine never creates, alters, or drops it).
    async fn table_exists(&self, schema: &str, table: &str) -> bool {
        let row = self
            .client
            .query_one(
                "SELECT EXISTS (SELECT 1 FROM pg_tables WHERE schemaname = $1 AND tablename = $2)",
                &[&schema, &table],
            )
            .await
            .expect("the oracle query succeeds");
        row.get::<_, bool>(0)
    }

    /// Create the embedder's liveness table — **test-owned DDL**, the
    /// instrument that keeps the schema-ownership rule honest (the
    /// table exists because this probe made it, never because an
    /// engine did).
    async fn create_liveness_table(&self, schema: &str, table: &str) {
        let sql = format!("CREATE TABLE \"{schema}\".\"{table}\" (key bytea PRIMARY KEY)");
        self.client
            .execute(sql.as_str(), &[])
            .await
            .expect("the embedder DDL succeeds");
    }

    async fn insert_live_row(&self, schema: &str, table: &str, key: &[u8]) {
        let sql = format!("INSERT INTO \"{schema}\".\"{table}\" (key) VALUES ($1)");
        self.client
            .execute(sql.as_str(), &[&key])
            .await
            .expect("the live row inserts");
    }

    /// The manual-expiry instrument: move every pin row for `key` to an
    /// already-past expiry — the abandonment shape (a holder whose
    /// renewal stopped), immediately durable, no timing reliance. The
    /// row stops protecting at once (`arbitrate` requires
    /// `expiry > now`); the fleet sweep's maintenance step reaps it.
    async fn expire_pin_rows(&self, schema: &str, key: &[u8]) {
        let expired = wall_nanos() - 1_000_000_000;
        let sql = format!("UPDATE \"{schema}\".gc_pins SET expiry = $1 WHERE key = $2");
        self.client
            .execute(sql.as_str(), &[&expired, &key])
            .await
            .expect("the manual expiry succeeds");
    }

    /// The one pin row's expiry (by key), when exactly one exists.
    async fn pin_expiry(&self, schema: &str, key: &[u8]) -> Option<i64> {
        let sql = format!("SELECT expiry FROM \"{schema}\".gc_pins WHERE key = $1");
        let rows = self
            .client
            .query(sql.as_str(), &[&key])
            .await
            .expect("the oracle query succeeds");
        rows.into_iter().next().map(|row| row.get::<_, i64>(0))
    }
}

// ------------------------------------------------------------------
// The fleet-validity rows — the composite fleet-validity predicate at
// the constructor, its clause-per-row form through the public
// constructors.
// ------------------------------------------------------------------

/// The composite fleet-validity predicate at the constructor: every
/// fleet-valid configuration constructs — the kv-only fleet client
/// (large = none, the re-routing posture's arm), the dual-tier fleet
/// (large = local on the declared shared media), and — gated — the
/// consolidation fleet (large = pg-lo, the same pg instance). Each
/// construct carries its mode, the minted schema is the fleet's one
/// shared pool, and a second vended instance constructs over the same
/// pool with its own store-instance id (the two instances' pin rows
/// compose by owner — the fleet's definition, exercised).
///
/// Contract stamp: store-api.md §Invariants 7 (facade-contract row);
/// ADR-011 §4 (the constructor-mode vocabulary); ADR-012 §3 (the
/// declaration — the composite fleet-validity predicate at the
/// constructor); ADR-008 §3 (the declared-shared-media arm; the
/// re-routing arm); ADR-012 §4 (the same-instance arm, ADR-012 §2's
/// TTL parameter).
pub async fn every_fleet_valid_configuration_constructs(
    kv_only: &FleetKvOnlyFactory,
    dual_tier: &FleetDualTierFactory,
) {
    let (a, b, site) = kv_only
        .open_fleet()
        .await
        .expect("the kv-only fleet constructs (large = none, the re-routing arm)");
    assert_eq!(a.mode(), StoreMode::KvOnly);
    assert_eq!(b.mode(), StoreMode::KvOnly);
    let content = b"the kv-only fleet client's entry".to_vec();
    let (_, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the kv-only fleet put publishes");
    drop(pin);
    let (_, re_pin) = b
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the second instance's fleet re-put publishes");
    drop(re_pin);
    let probe = connect_probe(site.dsn()).await;
    assert_eq!(
        probe.pool_rows(&site.schema).await,
        1,
        "the fleet re-put dedups — one pool entry across the two instances"
    );
    let rows = probe.pin_rows(&site.schema).await;
    assert_eq!(rows.len(), 2, "two instances, two owner rows on one pool");
    assert_ne!(
        rows[0].1, rows[1].1,
        "the two instances carry distinct store-instance ids — the pin \
         rows compose by owner"
    );
    kv_only
        .teardown()
        .await
        .expect("teardown drops exactly the minted schemas");
    assert!(
        !probe.schema_exists(&site.schema).await,
        "the fleet factory's schema is gone after teardown"
    );

    let (a, b, site) = dual_tier
        .open_fleet()
        .await
        .expect("the dual-tier fleet constructs (large = local on the declared shared media)");
    assert_eq!(a.mode(), StoreMode::DualTier);
    assert_eq!(b.mode(), StoreMode::DualTier);
    let content = patterned((dual_tier.threshold() - 1) as usize, 191);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the dual-tier fleet put publishes");
    drop(pin);
    let (_, re_pin) = b
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the second instance's fleet re-put publishes");
    drop(re_pin);
    assert!(
        b.has(&key.as_bytes()).await.unwrap(),
        "the second instance sees the first's entry — one fleet, one pool"
    );
    let probe = connect_probe(site.dsn()).await;
    assert_eq!(probe.pool_rows(&site.schema).await, 1, "one entry row");
    let rows = probe.pin_rows(&site.schema).await;
    assert_eq!(rows.len(), 2, "two owner rows");
    assert_ne!(rows[0].1, rows[1].1, "distinct store-instance ids");
    dual_tier
        .teardown()
        .await
        .expect("teardown drops exactly the minted schemas");
}

/// The consolidation fleet constructs — `kv = postgres` + `large =
/// pg-lo`, **the same pg instance** (the composite fleet-validity
/// predicate's runtime-checkable arm: the `system_identifier` probe,
/// server-side, immune to DSN aliases). Two instances over the one
/// shared instance, each with its own store-instance id; the pin rows
/// live on the kv host, the companion rows on the pg-lo host.
///
/// Contract stamp: ADR-012 §4 (the consolidation posture — the
/// same-instance clause the constructor probes); ADR-009 §2 (the
/// companion table); ADR-012 §3 (the declaration); ADR-012 §2 (the TTL
/// parameter).
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
pub async fn the_consolidation_fleet_constructs_over_one_shared_instance(
    consolidation: &FleetConsolidationFactory,
) {
    let (a, b, site) = consolidation
        .open_fleet()
        .await
        .expect("the consolidation fleet constructs over one shared instance");
    assert_eq!(a.mode(), StoreMode::DualTier);
    assert_eq!(b.mode(), StoreMode::DualTier);
    let content = patterned(2 * TUNED_THRESHOLD as usize + 7, 173);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the consolidation fleet put publishes");
    drop(pin);
    let (_, re_pin) = b
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the second instance's fleet re-put publishes");
    drop(re_pin);
    assert!(
        b.has(&key.as_bytes()).await.unwrap(),
        "the second instance sees the first's entry — one fleet, one \
         shared instance"
    );
    let probe = connect_probe(site.dsn()).await;
    assert_eq!(
        probe.schema_scalar(site.kv_schema(), "blobs").await,
        0,
        "the over-threshold entry rides the large tier — the kv pool \
         holds no entry row"
    );
    let rows = probe.pin_rows(site.kv_schema()).await;
    assert_eq!(rows.len(), 2, "two instances' pin rows on the kv host");
    assert_ne!(rows[0].1, rows[1].1, "distinct store-instance ids");
    consolidation
        .teardown()
        .await
        .expect("teardown reaps the LOs, drops the schemas and the database");
}

/// The constructor's entry guards — the rejects that error **closed,
/// before any connection attempt**: a `Duration::ZERO` pin TTL
/// (`Duration::ZERO` would degenerate the TTL/3 renewal cadence and
/// expire every pin at once — validated at entry on every `_fleet`
/// constructor) and a single-connection pool (the sweep lease holds
/// its checkout for the sweep's duration and the body's other clauses
/// ride further checkouts under it — the fleet pool-sizing guard,
/// ADR-013 §2). The predicate's structural non-fleet-valid shapes —
/// kv ≠ postgres, mem anywhere, `local` without the shared-media
/// declaration — carry **no fleet constructor to reach through** (the
/// recorded enforcement map: the `_fleet` constructors are named only
/// for the shapes the predicate accepts), so those clauses have no
/// runtime form here beyond the constructors' very existence.
///
/// Contract stamp: ADR-012 §3 (the declaration at the constructor);
/// ADR-008 §2 (the TTL's expiry semantics — zero degenerates the
/// cadence); ADR-013 §2 (the lease's hold across the sweep body — the
/// pool-sizing guard); ADR-008 §3 (the shared-media declaration — the
/// sealed type the call site names; `local` undeclared has no
/// constructor to reach it through).
pub async fn the_constructor_entry_guards_reject_the_degenerate_tunings() {
    let Some(dsn) = crate::postgres_column::harness_dsn() else {
        return;
    };
    let schema = "no_such_schema_zero_ttl";
    let err = Store::kv_only_postgres_fleet(
        &dsn,
        opts(schema.to_owned()),
        KV_ONLY_THRESHOLD,
        WINDOW_DELAY,
        Duration::ZERO,
    )
    .await
    .err()
    .expect("a zero pin TTL rejects at the kv-only fleet constructor");
    assert!(
        err.to_string().contains("must be positive"),
        "the zero-TTL rejection is the constructor's, spelled for the \
         embedder: {err}"
    );
    let err = Store::dual_tier_postgres_local_fleet(
        &dsn,
        opts(schema.to_owned()),
        std::env::temp_dir().join("alkblobs-fleet-column-zero-ttl-root"),
        TUNED_THRESHOLD,
        WINDOW_DELAY,
        Duration::ZERO,
        FleetSharedMedia::asserted(),
    )
    .await
    .err()
    .expect("a zero pin TTL rejects at the dual-tier fleet constructor");
    assert!(
        err.to_string().contains("must be positive"),
        "the zero-TTL rejection is the constructor's: {err}"
    );
    let err = Store::kv_only_postgres_fleet(
        &dsn,
        PgOpts {
            schema: schema.to_owned(),
            max_size: 1,
            ..PgOpts::default()
        },
        KV_ONLY_THRESHOLD,
        WINDOW_DELAY,
        TEST_TTL,
    )
    .await
    .err()
    .expect("a single-connection pool rejects at the fleet constructor");
    assert!(
        err.to_string().contains("at least 2 connections"),
        "the pool-sizing guard is the rejection: {err}"
    );
}

/// The consolidation fleet's same-instance probe **errors closed** (the
/// runtime-checkable arm's failure posture): pointing the pg-lo large
/// engine at an unreachable instance is a constructor error — any
/// probe failure rejects, never a silent pass. The probe's
/// *mismatching-instances* leg itself needs two postgres instances;
/// the harness server is one instance, so that leg is unexercisable
/// here (recorded — the engine suite's tests carry the same-instance
/// posture; the cross-instance mismatch is the constructor's recorded
/// shape).
///
/// Contract stamp: ADR-012 §4 (the same-instance clause — the probe
/// reads `pg_control_system()` and any failure, including an
/// unreachable instance, is a constructor error); ADR-012 §3.
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
pub async fn the_consolidation_same_instance_probe_errors_closed(dsn: &str) {
    let unreachable = "host=no-such-host.invalid port=1 user=x password=x dbname=x";
    let err = Store::dual_tier_postgres_pglo_fleet(
        dsn,
        opts("no_such_schema_probe_closed".to_owned()),
        unreachable,
        opts("no_such_schema_probe_closed".to_owned()),
        TUNED_THRESHOLD,
        WINDOW_DELAY,
        TEST_TTL,
    )
    .await
    .err()
    .expect("the unreachable large instance rejects at the constructor");
    assert!(
        matches!(err, Error::Io(_)),
        "the probe failure surfaces as the operational constructor error \
         it is (the probe errors closed): {err}"
    );
}

// ------------------------------------------------------------------
// The joint entry+pin tx rows — through the public `put`, the fleet
// put path's atomicity visible to an independent connection.
// ------------------------------------------------------------------

/// The fleet put path's joint transaction, through the public `put`:
/// the entry row and its pin row commit **together** — an independent
/// connection sees both or neither. The both-halves: instance a's put
/// publishes one pool row and one pin row (the pin row's expiry ahead
/// of `now` — the TTL the store resolved), and the second instance
/// sees the entry at once (one fleet, one pool). The neither-half: a
/// rejecting put (declared-length mismatch, the staged-put contract's
/// failure mode) leaves zero pool rows and zero pin rows.
///
/// Contract stamp: store-api.md §Invariants 4 (a visible pool entry is
/// never unprotected — the joint commit is the mechanism; an in-flight
/// put is never deleted by a sweep started before it committed);
/// store-api.md §Invariants 3 (zero residue on failed puts); ADR-008
/// §2 (the joint entry+pin tx, pin-before-publish); ADR-010 §2 (the
/// ownership statement — the store core holds the tx end to end);
/// ADR-012 §3/§4 (the fleet declaration; the put path's wiring).
pub async fn the_joint_entry_and_pin_tx_commits_both_or_neither(factory: &FleetDualTierFactory) {
    let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&a);

    let content = patterned((factory.threshold() - 1) as usize, 191);
    let canonical = Key::derive(&content, content.len() as u64);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the fleet put publishes");

    assert_eq!(key, canonical, "the canonical derivation addresses the put");
    let probe = connect_probe(site.dsn()).await;
    assert_eq!(
        (
            probe.pool_rows(&site.schema).await,
            probe.pin_rows(&site.schema).await.len()
        ),
        (1, 1),
        "the independent connection sees both — the entry row and its \
         pin row committed together"
    );
    assert!(
        probe
            .pin_expiry(&site.schema, &key.as_bytes())
            .await
            .unwrap()
            > wall_nanos(),
        "the pin row's expiry is ahead of now — the TTL the store resolved"
    );
    assert!(
        b.has(&key.as_bytes()).await.unwrap(),
        "the second instance sees the entry — one fleet, one pool"
    );

    // The neither-half: the declared-length mismatch rejects before
    // any entry or pin row exists (a failed put leaves nothing).
    let err = a
        .put(
            Some(content.len() as u64 + 400),
            Cursor::new(content.as_slice()),
        )
        .await
        .expect_err("the over-fed mismatch rejects");
    assert_eq!(err, Error::Verification, "declared-len mismatch rejects");
    assert_eq!(
        (
            probe.pool_rows(&site.schema).await,
            probe.pin_rows(&site.schema).await.len()
        ),
        (1, 1),
        "the failed put left neither — the pool and the pin set are \
         exactly the successful put's"
    );

    drop(pin);
    factory.teardown().await.expect("teardown is idempotent");
}

/// The dedup re-put under fleet, through the public `put`: the same
/// instance's re-put commits the pin row alone (the joint tx's row
/// upsert-renewing `(key, owner)` — the wave-4 recorded shape; no new
/// entry row), and a second instance's re-put composes a **second**
/// pin row — protection by row (ADR-010 §2's refcount-in-rows
/// reading), one pool entry throughout.
///
/// Contract stamp: store-api.md §Invariants 2 (dedup: identical
/// content twice is one pool entry — the fleet's cross-instance leg);
/// ADR-003 (whole-value put is CAS first-writer-wins); ADR-005 (a
/// re-put refcounts the existing entry); ADR-010 §2 (the pin rows
/// compose by owner); ADR-008 §2 (the joint tx's row upsert; the
/// `(key, owner)` row identity).
pub async fn a_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row(
    factory: &FleetDualTierFactory,
) {
    let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&a);

    let content = patterned((factory.threshold() - 1) as usize, 167);
    let canonical = Key::derive(&content, content.len() as u64);

    let (_, pin_one) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the first put publishes");
    let probe = connect_probe(site.dsn()).await;
    assert_eq!(probe.pool_rows(&site.schema).await, 1, "one entry row");
    let first_rows = probe.pin_rows(&site.schema).await;
    assert_eq!(first_rows.len(), 1, "instance a's one pin row");
    drop(pin_one);

    let (key_again, pin_two) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the dedup re-put publishes");
    assert_eq!(key_again, canonical, "one derivation, one address");
    assert_eq!(
        probe.pool_rows(&site.schema).await,
        1,
        "the dedup re-put committed the pin row alone — no new entry row"
    );
    let renewed_rows = probe.pin_rows(&site.schema).await;
    assert_eq!(
        renewed_rows.len(),
        1,
        "the same instance's row, upsert-renewed"
    );
    assert!(
        renewed_rows[0].2 > first_rows[0].2,
        "the row upsert-renews (key, owner): the expiry moved forward"
    );
    drop(pin_two);

    // The second instance's re-put: a different pin row, the same one
    // pool entry.
    let (key_third, pin_three) = b
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the second instance's re-put publishes");
    assert_eq!(key_third, canonical);
    assert_eq!(
        probe.pool_rows(&site.schema).await,
        1,
        "still one pool entry across instances"
    );
    let rows = probe.pin_rows(&site.schema).await;
    assert_eq!(rows.len(), 2, "protection composes by row — two owners");
    assert_ne!(rows[0].1, rows[1].1, "two store-instance ids, two rows");
    drop(pin_three);

    factory.teardown().await.expect("teardown is idempotent");
}

/// A guard's TTL/3 renewal duty, through the facade: while the guard
/// lives, the pin row's expiry moves forward every TTL/3 (the short
/// test TTL makes the cadence observable over an independent
/// connection); the dropped guard's renewal is stopped with its life —
/// no leaked tasks, no renewal past the guard's life — and the row
/// freezes.
///
/// Contract stamp: ADR-008 §2 (the pin guard renews its row every
/// TTL/3; a dropped guard stops its duty); ADR-012 §2 (the TTL lives
/// with the store's renewal schedule — the constructor parameter this
/// column tunes).
pub async fn the_pin_guards_ttl_third_renewal_moves_the_rows_expiry(
    factory: &FleetDualTierFactory,
) {
    let (a, _b, site) = factory.open_fleet().await.expect("open vends the fleet");

    let content = patterned((factory.threshold() - 1) as usize, 149);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the fleet put publishes");
    let probe = connect_probe(site.dsn()).await;
    let initial = probe
        .pin_expiry(&site.schema, &key.as_bytes())
        .await
        .expect("the put's pin row");
    tokio::time::sleep(ONE_CADENCE).await;
    let renewed = probe
        .pin_expiry(&site.schema, &key.as_bytes())
        .await
        .expect("the row");
    assert!(
        renewed > initial,
        "the guard's TTL/3 duty moved the row's expiry forward \
         (initial {initial}, renewed {renewed})"
    );
    drop(pin);
    let at_drop = probe
        .pin_expiry(&site.schema, &key.as_bytes())
        .await
        .expect("the row");
    assert!(
        at_drop >= renewed,
        "the row's expiry moved forward until the drop and never \
         backwards (one TTL/3 tick may land between the read and the \
         drop)"
    );
    tokio::time::sleep(EXPIRY_HORIZON).await;
    assert_eq!(
        at_drop,
        probe
            .pin_expiry(&site.schema, &key.as_bytes())
            .await
            .expect("the row"),
        "the dropped guard's renewal is stopped — no leaked tasks, no \
         renewal past the guard's life"
    );

    factory.teardown().await.expect("teardown is idempotent");
}

/// The pin row's **never-resurrect semantics, through the facade**: an
/// expired row is never renewed back to life — even under a *live*
/// guard whose renewal duty is still running. The independent
/// connection's manual-expiry instrument creates the abandoned shape
/// (a row past its expiry under a live holder); after at least one
/// TTL/3 tick the row holds exactly the expired value — the guard's
/// renewal observed the expired slot and the never-resurrect predicate
/// (`pin_renew`'s `expiry > now` clause) kept it dead. The row's
/// protection is over; the delete-then-recover backstop's territory.
///
/// Contract stamp: ADR-008 §2 (the never-resurrect semantics — an
/// expired row is never revived; the expiry semantics verbatim);
/// ADR-012 §2 (the TTL/renewal schedule's expiry reading).
pub async fn an_expired_row_is_never_renewed_back_to_life(factory: &FleetDualTierFactory) {
    let (a, _b, site) = factory.open_fleet().await.expect("open vends the fleet");

    let content = patterned((factory.threshold() - 1) as usize, 151);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the fleet put publishes");

    let probe = connect_probe(site.dsn()).await;
    probe.expire_pin_rows(&site.schema, &key.as_bytes()).await;
    let expired = probe
        .pin_expiry(&site.schema, &key.as_bytes())
        .await
        .expect("the abandoned row");
    assert!(
        expired <= wall_nanos(),
        "the instrument left the row expired (a live guard holds it)"
    );

    tokio::time::sleep(ONE_CADENCE).await;
    assert_eq!(
        expired,
        probe
            .pin_expiry(&site.schema, &key.as_bytes())
            .await
            .expect("the row"),
        "the expired row was never renewed back to life — the live \
         guard's renewal never resurrects"
    );

    drop(pin);
    factory.teardown().await.expect("teardown is idempotent");
}

/// A paused holder (renewal stopped) loses protection exactly at
/// expiry, and the delete-then-recover backstop recovers: the dropped
/// guard's row freezes at its remaining TTL; while un-expired, the
/// (foreign-instance) delete is refused — `GcRefuse`, the row's
/// once-protection — and nothing is deleted; past expiry the delete
/// proceeds; and a byte-identical re-put recovers the entry (one new
/// pin row, the pool's own protection restored).
///
/// Contract stamp: ADR-008 §2 (the expiry semantics verbatim: a pin
/// whose holder paused past TTL abandons the row; the entry becomes
/// deleteable; recovery is the byte-identical re-put); ADR-012 §2
/// (the TTL schedule's reading); store-api.md §Invariants 4 (correct
/// liveness: the refuse/proceed boundary is exactly the expiry).
pub async fn a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers(
    factory: &FleetDualTierFactory,
) {
    let (a, b, _site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&a);

    let content = patterned((factory.threshold() - 1) as usize, 137);
    let canonical = Key::derive(&content, content.len() as u64);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the fleet put publishes");

    // Renewal stopped: the guard's drop pauses the holder (the row's
    // remaining TTL is its last protection).
    drop(pin);
    let err = b
        .delete(&key.as_bytes())
        .await
        .expect_err("the un-expired row still protects");
    assert_eq!(
        err,
        Error::GcRefuse,
        "the row's once-protection holds while un-expired (a foreign \
         instance's read of the pool's pin set)"
    );
    assert!(a.has(&key.as_bytes()).await.unwrap(), "nothing was deleted");

    tokio::time::sleep(EXPIRY_HORIZON).await;
    b.delete(&key.as_bytes())
        .await
        .expect("past expiry, the paused holder's protection is gone");
    assert!(!a.has(&key.as_bytes()).await.unwrap());

    // The delete-then-recover backstop: a byte-identical re-put
    // recovers the entry.
    let (recovered, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the recovering re-put publishes");
    assert_eq!(recovered, canonical, "the same canonical address");
    let (got_len, blob) = a
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the entry recovered");
    assert_eq!(got_len, content.len() as u64);
    assert_eq!(
        blob.read_at(0, got_len).await.expect("the window reads"),
        content.as_slice(),
        "the recovered entry is byte-identical"
    );
    drop(pin);

    factory.teardown().await.expect("teardown is idempotent");
}

// ------------------------------------------------------------------
// The exact-count fleet sweep rows — INV-4's engine-gated fleet form,
// the wave's gate. Every row readsits counts against the durable
// schedule (staged this sweep, executed by the next — the window
// delay `0` tuning), through the pool/window/pin-row oracle.
// ------------------------------------------------------------------

/// The wave's gate — INV-4's **two-instance exact-count shape**:
/// instance a pins an entry (its pin row — the joint tx's row, its
/// guard held), instance b sweeps. The mark phases stage exactly the
/// dead set (the SQL liveness is never enumerated into the mark set —
/// a's row is not b's in-process state), the executing sweep deletes
/// the unpinned twins exactly and cancels exactly the pinned entry at
/// the execution-time SQL re-arbitration (a's un-expired row, a
/// foreign owner from b's seat), and the pinned entry stays byte-
/// identical. Counts exact: staged 3, executed 2+1, re-staged 1.
///
/// Contract stamp: store-api.md §Invariants 4 (sweep safety's
/// engine-gated fleet form — DB-backed pins + the advisory-locked
/// sweep, the exact-count sweep-outcome shape against dockerized
/// postgres) and §Invariants 5 (list correctness through outcomes);
/// ADR-008 §2 (the fleet sweep body; the SQL arbitration composes
/// behind the in-process domain); ADR-012 §1 (abort is report data;
/// the exact report fields); ADR-012 §6.7 (the executor is the next
/// sweep); ADR-013 §2 (the sweep lease's single-flight); ADR-007 §5
/// (postgres, the fleet host — the exact-count gate against
/// dockerized postgres).
pub async fn two_instance_exact_counts_instance_a_pins_instance_b_sweeps(
    factory: &FleetDualTierFactory,
) {
    let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&b);

    let pinned = patterned((factory.threshold() - 1) as usize, 179);
    let (key_pinned, pin) = a
        .put(Some(pinned.len() as u64), Cursor::new(pinned.as_slice()))
        .await
        .expect("instance a's fleet put publishes — its pin row with it");

    let dead = patterned((factory.threshold() - 3) as usize, 173);
    let (key_dead, pin_dead) = b
        .put(Some(dead.len() as u64), Cursor::new(dead.as_slice()))
        .await
        .expect("the twin publishes");
    drop(pin_dead);
    let dead_too = patterned((factory.threshold() - 5) as usize, 181);
    let (key_dead_too, pin_dead_too) = b
        .put(
            Some(dead_too.len() as u64),
            Cursor::new(dead_too.as_slice()),
        )
        .await
        .expect("the other twin publishes");
    drop(pin_dead_too);

    // The twins' holders abandoned their rows: the expired shape (the
    // manual-expiry instrument — the abandonment the maintenance step
    // reaps; no timing reliance).
    let probe = connect_probe(site.dsn()).await;
    probe
        .expire_pin_rows(&site.schema, &key_dead.as_bytes())
        .await;
    probe
        .expire_pin_rows(&site.schema, &key_dead_too.as_bytes())
        .await;

    assert_eq!(
        (
            probe.pool_rows(&site.schema).await,
            probe.pin_rows(&site.schema).await.len()
        ),
        (3, 3),
        "three entries; three pin rows on the one shared pool (one per \
         instance's put, a's renewing)"
    );

    let first = b.sweep().await.expect("sweeps are never errors");
    assert_eq!(first.aborted, None);
    assert_eq!(
        (
            first.candidates_staged,
            first.deleted,
            first.cancelled_by_arbitration
        ),
        (3, 0, 0),
        "b's mark staged exactly the three dead-by-b's-seat candidates \
         (a's row is not b's enumerable live state) and deleted nothing"
    );
    assert_eq!(
        probe.window_rows(&site.schema).await,
        3,
        "the staged set waits in the durable window"
    );
    assert_eq!(
        probe.pin_rows(&site.schema).await.len(),
        1,
        "the maintenance step reaped exactly the twins' expired rows — \
         a's renewing row survives"
    );

    let second = b.sweep().await.expect("sweeps are never errors");
    assert_eq!(second.aborted, None);
    assert_eq!(
        second.candidates_staged, 1,
        "the survivor re-staged (a's row is still not enumerable live \
         state)"
    );
    assert_eq!(
        (second.deleted, second.cancelled_by_arbitration),
        (2, 1),
        "the executing sweep deleted exactly the two unpinned twins and \
         cancelled exactly the pinned entry — a's un-expired row at the \
         SQL re-arbitration"
    );
    assert!(!b.has(&key_dead.as_bytes()).await.unwrap());
    assert!(!b.has(&key_dead_too.as_bytes()).await.unwrap());
    assert!(a.has(&key_pinned.as_bytes()).await.unwrap());
    assert!(
        b.has(&key_pinned.as_bytes()).await.unwrap(),
        "the fleet's shared storage"
    );
    let (got_len, blob) = b
        .get(&key_pinned.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the pinned entry survived");
    assert_eq!(got_len, pinned.len() as u64);
    assert_eq!(
        blob.read_at(0, got_len).await.expect("the window reads"),
        pinned.as_slice(),
        "the surviving pinned entry is byte-identical"
    );

    drop(pin);
    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, the lease contention row: two instances sweep **concurrently**
/// — exactly one sweeps; the contended second aborts `SweeperLock` as
/// report data with nothing staged, nothing executed, nothing deleted
/// (abort before staging: form-independent), while the holder — whose
/// protect consult latched mid-body — completes by staging only (its
/// staged set waits for the next sweep to execute). The SQL lease is
/// the fleet host's advisory-locked single-flight across instances;
/// the in-process flag composes beside it.
///
/// Contract stamp: store-api.md §Invariants 4 (aborting sources delete
/// nothing); ADR-013 §2 (the named exclusive sweep lease — one sweeper
/// per pool across instances, atomic test-and-set, holdership ends at
/// release; the fleet pool-sizing guard's semantics); ADR-012 §1 (the
/// contended sweep's abort is report data); ADR-012 §6.7 (the holder
/// stages only — the executor is the next sweep).
pub async fn two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data(
    factory: &FleetDualTierFactory,
) {
    let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&a);
    SetSource::register(&b);

    let dead = patterned((factory.threshold() - 1) as usize, 191);
    let (key, pin) = a
        .put(Some(dead.len() as u64), Cursor::new(dead.as_slice()))
        .await
        .expect("the dead witness publishes");
    drop(pin);
    let probe = connect_probe(site.dsn()).await;
    probe.expire_pin_rows(&site.schema, &key.as_bytes()).await;

    let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>();
    let (release_tx, release_rx) = std::sync::mpsc::channel::<()>();
    a.register_protect_callback(Arc::new(LatchProtect {
        entered: entered_tx,
        release: parking_lot::Mutex::new(Some(release_rx)),
    }));

    let a = Arc::new(a);
    let holder = tokio::spawn({
        let a = Arc::clone(&a);
        async move { a.sweep().await }
    });
    while entered_rx.recv_timeout(Duration::from_secs(5)).is_err() {
        tokio::task::yield_now().await;
    }

    let contended = b.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        contended.aborted,
        Some(GcAbortCause::SweeperLock),
        "the contended second instance aborts as report data"
    );
    assert_eq!(
        (
            contended.candidates_staged,
            contended.deleted,
            contended.cancelled_by_arbitration
        ),
        (0, 0, 0),
        "the contended sweep staged, executed, and deleted nothing"
    );
    assert!(
        b.has(&key.as_bytes()).await.unwrap(),
        "the contended sweep deleted nothing"
    );
    assert_eq!(
        probe.window_rows(&site.schema).await,
        0,
        "the aborted sweep staged nothing into the window"
    );
    release_tx.send(()).ok();

    let held = holder
        .await
        .expect("the holder task joined")
        .expect("sweeps are never errors");
    assert_eq!(held.aborted, None, "the lease holder completes");
    assert_eq!(
        (held.candidates_staged, held.deleted),
        (1, 0),
        "the holder stages exactly the dead set — it does not execute"
    );
    assert_eq!(
        probe.window_rows(&site.schema).await,
        1,
        "the holder's staged set waits in the durable window"
    );
    assert!(
        b.has(&key.as_bytes()).await.unwrap(),
        "the holder staged, did not execute"
    );

    let executed = b.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        executed.deleted, 1,
        "the released lease's next sweep executes the staged set"
    );
    assert!(!b.has(&key.as_bytes()).await.unwrap());

    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, the cross-instance arbitration row: a foreign owner's
/// un-expired pin row refuses instance b's direct delete
/// (`GcRefuse` — nothing deleted); the row's holder stops renewing
/// (the guard's drop — the paused holder), the short test TTL expires
/// the row, and the delete proceeds. The direct path's SQL-level
/// arbitration is the fleet-authoritative half of the two-domain
/// composition: b's own in-process domain has none of a's state, and
/// a's row alone decides the refusal.
///
/// Contract stamp: store-api.md §Invariants 4 (with correct liveness
/// the delete is refused — the refusal boundary is exactly the row's
/// expiry); ADR-008 §2 (the SQL-level arbitration on the direct
/// delete path folds un-expired pin rows of *any* instance into one
/// answer); ADR-012 §2 (the expiry semantics — `expiry > now` is
/// live); ADR-005 (the facade owns the refusal — `GcRefuse`,
/// nothing deleted).
pub async fn a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry(
    factory: &FleetDualTierFactory,
) {
    let (a, b, _site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&b);

    let content = patterned((factory.threshold() - 1) as usize, 163);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("instance a pins via its fleet put");

    let err = b.delete(&key.as_bytes()).await.expect_err(
        "the foreign owner's un-expired pin row refuses instance b's \
         delete",
    );
    assert_eq!(err, Error::GcRefuse, "the refusal is the facade's, as data");
    assert!(a.has(&key.as_bytes()).await.unwrap(), "nothing was deleted");
    assert!(
        b.has(&key.as_bytes()).await.unwrap(),
        "the shared storage is untouched"
    );

    drop(pin);
    tokio::time::sleep(EXPIRY_HORIZON).await;
    b.delete(&key.as_bytes()).await.expect(
        "after the row expires (renewal stopped, short TTL), the \
                 delete proceeds",
    );
    assert!(!a.has(&key.as_bytes()).await.unwrap());

    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, the race across the window delay in its fleet form: an
/// entry staged by the first sweep, a put landing **during the
/// window** — instance b's fleet re-put dedups into the staged entry
/// and its pin row commits before anything further could resolve the
/// candidate — and the executing sweep's SQL re-arbitration cancels
/// the candidate, never the entry. The surviving entry is byte-
/// identical; once the foreign guard's row expires (renewal stopped,
/// short TTL), the next sweep deletes exactly it.
///
/// Contract stamp: store-api.md §Invariants 4 (an in-flight/just-
/// landed put is never deleted by a sweep started before it
/// committed — the race's fleet form is closed by the row); ADR-008
/// §2 (pin-before-publish; the window delay separates staging from
/// execution); ADR-012 §6.7 (the execution-time re-arbitration is the
/// executor sweep's posture); ADR-012 §2 (the expiry reading the
/// final delete rides).
pub async fn a_put_landing_during_the_fleet_window_survives_the_executing_sweep(
    factory: &FleetDualTierFactory,
) {
    let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&a);

    let content = patterned((factory.threshold() - 1) as usize, 191);
    let canonical = Key::derive(&content, content.len() as u64);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the twin publishes");
    drop(pin);
    let probe = connect_probe(site.dsn()).await;
    probe.expire_pin_rows(&site.schema, &key.as_bytes()).await;

    let staged = a.sweep().await.expect("sweeps are never errors");
    assert_eq!(staged.aborted, None);
    assert_eq!(
        (staged.candidates_staged, staged.deleted),
        (1, 0),
        "the staging sweep stages exactly the abandoned twin and \
         deletes nothing"
    );

    // The put lands during the window: instance b's fleet re-put
    // dedups into the staged entry — its pin row commits (the joint
    // tx, row-alone on the dedup leg) before anything resolves the
    // candidate.
    let (_, foreign_pin) = b
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("instance b's fleet put lands during the window");

    let executed = a.sweep().await.expect("sweeps are never errors");
    assert_eq!(executed.aborted, None);
    assert_eq!(
        (executed.deleted, executed.cancelled_by_arbitration),
        (0, 1),
        "the executing sweep's SQL re-arbitration cancelled the \
         candidate — b's un-expired row"
    );
    assert_eq!(
        executed.candidates_staged, 1,
        "the survivor re-staged (b's row is not a's enumerable live state)"
    );
    let (got_len, blob) = a
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("the put landing during the window survived");
    assert_eq!(got_len, content.len() as u64);
    assert_eq!(
        blob.read_at(0, got_len).await.expect("the window reads"),
        content.as_slice(),
        "the survivor is byte-identical"
    );

    drop(foreign_pin);
    tokio::time::sleep(EXPIRY_HORIZON).await;
    let reaped = a.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (reaped.deleted, reaped.candidates_staged),
        (1, 0),
        "after the foreign row expired (renewal stopped, short TTL) \
         and the maintenance step reaped it, the survivor executed \
         exactly once"
    );
    assert!(
        !a.has(&key.as_bytes()).await.unwrap(),
        "the address is the canonical one"
    );
    assert_eq!(canonical.as_bytes(), key.as_bytes());

    factory.teardown().await.expect("teardown is idempotent");
}

/// INV-4, the maintenance-step row: the fleet sweep's pin-reap (the
/// GC-of-GC cleanup) deletes **exactly** the expired pin rows — the
/// abandoned shape (renewal stopped past expiry) is reaped, the
/// un-expired row survives — and the reaped row's candidate executes
/// while the surviving row's candidate is cancelled at the
/// execution-time SQL re-arbitration (the reap runs before the
/// overdue execution: every later arbitration sees the post-reap
/// protection state).
///
/// Contract stamp: store-api.md §Invariants 4 (exact counts with
/// correct liveness — the maintenance step is the pin-set's sweep);
/// ADR-008 §2 (the sweeper's maintenance step: an expired pin is
/// abandoned and reaped; the GC-of-GC cleanup); ADR-012 §6.7 (the
/// executor is the next sweep — the counts' schedule);
/// ADR-012 §1 (the exact report fields).
pub async fn the_maintenance_step_reaps_exactly_the_expired_rows(factory: &FleetDualTierFactory) {
    let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
    SetSource::register(&b);

    let doomed = b"whose pin is abandoned".to_vec();
    let (key_doomed, pin_one) = a
        .put(Some(doomed.len() as u64), Cursor::new(doomed.as_slice()))
        .await
        .expect("the doomed entry pins");
    let survivor = b"whose pin renews".to_vec();
    let (key_survivor, pin_two) = a
        .put(
            Some(survivor.len() as u64),
            Cursor::new(survivor.as_slice()),
        )
        .await
        .expect("the survivor entry pins");

    let probe = connect_probe(site.dsn()).await;
    probe
        .expire_pin_rows(&site.schema, &key_doomed.as_bytes())
        .await;

    let first = b.sweep().await.expect("sweeps are never errors");
    assert_eq!(first.aborted, None);
    assert_eq!(
        (
            first.candidates_staged,
            first.deleted,
            first.cancelled_by_arbitration
        ),
        (2, 0, 0),
        "both entries marked (the in-process live set is b's, empty); \
         nothing overdue yet"
    );
    let rows = probe.pin_rows(&site.schema).await;
    assert_eq!(
        rows.len(),
        1,
        "the maintenance step reaped exactly the expired row — one \
         row survives"
    );
    assert_eq!(
        rows[0].0,
        key_survivor.as_bytes(),
        "the un-expired row survives"
    );

    let second = b.sweep().await.expect("sweeps are never errors");
    assert_eq!(second.aborted, None);
    assert_eq!(
        (second.deleted, second.cancelled_by_arbitration),
        (1, 1),
        "the reaped row's candidate executed; the survivor's cancelled \
         at the SQL re-arbitration"
    );
    assert_eq!(second.candidates_staged, 1, "the survivor re-staged");
    assert!(a.has(&key_survivor.as_bytes()).await.unwrap());
    assert!(!a.has(&key_doomed.as_bytes()).await.unwrap());

    drop(pin_one);
    drop(pin_two);
    factory.teardown().await.expect("teardown is idempotent");
}

/// The **fleet liveness row**: a test-owned embedder table's rows
/// protect fleet-wide through the fleet registration verb — the SQL
/// arbitration folds the table into the answer on the sweep's
/// execution-time re-arbitration and on the direct-delete path alike —
/// and the schema-ownership rule holds: the table exists because the
/// test made it (the engine never creates, alters, or drops it), the
/// table is never enumerated into the mark set (consulted at
/// arbitration time), a malformed reference errors at registration,
/// re-registration is idempotent, and the deregistration composes
/// (the correction flow: the token's removal lets the delete proceed).
///
/// Contract stamp: store-api.md §Invariants 4 (correct liveness: a
/// registered fleet liveness table's rows protect); ADR-012 §6.6 (the
/// fleet liveness verb — the same verb with a fleet-shaped argument;
/// registration records the reference); ADR-008 §2 (the SQL protocol
/// sees tables, never callbacks — the schema-ownership rule: the
/// store reads, never owns).
pub async fn an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned(
    factory: &FleetDualTierFactory,
) {
    let (a, _b, site) = factory.open_fleet().await.expect("open vends the fleet");

    let probe = connect_probe(site.dsn()).await;
    probe
        .create_liveness_table(&site.schema, "embedder_live")
        .await;
    let reference = format!("{}.embedder_live", site.schema);

    let registration = a
        .register_liveness_source(LivenessRegistration::FleetTable(reference.clone()))
        .expect("the embedder table reference registers");
    let re_registered = a
        .register_liveness_source(LivenessRegistration::FleetTable(reference.clone()))
        .expect("re-registration is idempotent");
    assert_eq!(
        registration, re_registered,
        "re-registering the same reference returns the original token"
    );
    let malformed = a
        .register_liveness_source(LivenessRegistration::FleetTable(
            "bad..reference".to_owned(),
        ))
        .expect_err("a malformed reference errors at registration");
    assert!(
        malformed
            .to_string()
            .contains("invalid liveness table reference"),
        "the malformed reference surfaces the engine error: {malformed}"
    );
    assert!(
        probe.table_exists(&site.schema, "embedder_live").await
            && !probe.table_exists(&site.schema, "no_such_table").await,
        "the schema-ownership rule: the embedder's table is the \
         embedder's, and no engine ever created one at registration \
         time"
    );

    let protected = b"live in the embedder's table".to_vec();
    let (key_protected, pin_one) = a
        .put(
            Some(protected.len() as u64),
            Cursor::new(protected.as_slice()),
        )
        .await
        .expect("the protected entry pins");
    drop(pin_one);
    let dead = b"dead beside the table's row".to_vec();
    let (key_dead, pin_two) = a
        .put(Some(dead.len() as u64), Cursor::new(dead.as_slice()))
        .await
        .expect("the dead entry pins");
    drop(pin_two);
    probe
        .expire_pin_rows(&site.schema, &key_protected.as_bytes())
        .await;
    probe
        .expire_pin_rows(&site.schema, &key_dead.as_bytes())
        .await;
    probe
        .insert_live_row(&site.schema, "embedder_live", &key_protected.as_bytes())
        .await;

    let first = a.sweep().await.expect("sweeps are never errors");
    assert_eq!(first.aborted, None);
    assert_eq!(
        (first.candidates_staged, first.deleted),
        (2, 0),
        "both marked — the table is never enumerated into the mark set"
    );
    let second = a.sweep().await.expect("sweeps are never errors");
    assert_eq!(
        (second.deleted, second.cancelled_by_arbitration),
        (1, 1),
        "the dead entry executed; the table-protected entry cancelled \
         at the execution-time SQL re-arbitration"
    );
    assert!(a.has(&key_protected.as_bytes()).await.unwrap());
    assert!(!a.has(&key_dead.as_bytes()).await.unwrap());

    assert_eq!(
        a.delete(&key_protected.as_bytes()).await.unwrap_err(),
        Error::GcRefuse,
        "the liveness table protects the direct delete path too"
    );

    assert!(
        a.unregister_liveness_source(registration),
        "the fleet token deregisters once"
    );
    assert!(
        !a.unregister_liveness_source(re_registered),
        "the idempotent re-registration consumed one token only"
    );
    a.delete(&key_protected.as_bytes())
        .await
        .expect("the correction flow: deregistered, the delete proceeds");

    factory.teardown().await.expect("teardown is idempotent");
}

// ------------------------------------------------------------------
// The structural factory tests — the fleet factory contract's own
// wiring-level assertions (not conformance rows: no backlog entry).
// ------------------------------------------------------------------

/// The fleet factory contract, structurally: one minted schema, two
/// instances over it, distinct store-instance ids observable in the
/// pin rows' owner bytes, distinct vended roots, teardown dropping
/// exactly the minted schema and removing every vend, idempotent.
/// Run on both factories: the dual-tier fleet (kv = postgres + large
/// = local) and the kv-only fleet (the re-routing client shape).
#[tokio::test]
async fn fleet_factory_one_schema_vends_instances_with_distinct_owner_ids() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    {
        let (a, b, site) = factory.open_fleet().await.expect("open vends the fleet");
        assert_eq!(
            factory.tracked_schemas().len(),
            1,
            "one minted schema per open"
        );
        assert_ne!(
            site.root(0),
            site.root(1),
            "two instances, two private roots"
        );
        let one = b"instance a's entry".to_vec();
        let two = b"instance b's entry".to_vec();
        let (_, pin_a) = a
            .put(Some(one.len() as u64), Cursor::new(one.as_slice()))
            .await
            .expect("a's put publishes");
        let (_, pin_b) = b
            .put(Some(two.len() as u64), Cursor::new(two.as_slice()))
            .await
            .expect("b's put publishes");
        drop(pin_a);
        drop(pin_b);
        let probe = connect_probe(site.dsn()).await;
        let rows = probe.pin_rows(&site.schema).await;
        assert_eq!(rows.len(), 2, "two instances, two pin rows");
        assert_ne!(
            rows[0].1, rows[1].1,
            "two store-instance ids — the pin rows compose by owner"
        );
        assert_eq!(
            committed_file_count(site.root(0)),
            0,
            "the kv-tier entries left no large-tier files"
        );
    }
    factory.teardown().await.expect("teardown drops the schema");
    factory.teardown().await.expect("teardown repeats cleanly");
    let probe = connect_probe(factory.dsn()).await;
    for schema in factory.tracked_schemas() {
        assert!(
            !probe.schema_exists(&schema).await,
            "teardown dropped exactly the minted schema: {schema}"
        );
    }

    let kv_only = FleetKvOnlyFactory::new(KV_ONLY_THRESHOLD);
    {
        let (a, b, site) = kv_only
            .open_fleet()
            .await
            .expect("open vends the kv-only fleet");
        assert_eq!(
            kv_only.tracked_schemas().len(),
            1,
            "one minted schema per open"
        );
        assert_eq!(a.mode(), StoreMode::KvOnly);
        assert_eq!(b.mode(), StoreMode::KvOnly);
        let one = b"instance a's kv entry".to_vec();
        let two = b"instance b's kv entry".to_vec();
        let (_, pin_a) = a
            .put(Some(one.len() as u64), Cursor::new(one.as_slice()))
            .await
            .expect("a's put publishes");
        let (_, pin_b) = b
            .put(Some(two.len() as u64), Cursor::new(two.as_slice()))
            .await
            .expect("b's put publishes");
        drop(pin_a);
        drop(pin_b);
        let probe = connect_probe(site.dsn()).await;
        let rows = probe.pin_rows(&site.schema).await;
        assert_eq!(rows.len(), 2, "two instances, two pin rows");
        assert_ne!(rows[0].1, rows[1].1, "distinct store-instance ids");
    }
    kv_only.teardown().await.expect("teardown drops the schema");
    let probe = connect_probe(kv_only.dsn()).await;
    for schema in kv_only.tracked_schemas() {
        assert!(
            !probe.schema_exists(&schema).await,
            "teardown dropped the kv-only fleet's schema: {schema}"
        );
    }
}

/// A failing assertion panics the row before teardown runs — and the
/// fleet factory's teardown must make sense over the abandoned fleet
/// anyway: the minted schema is really gone server-side (`DROP SCHEMA
/// … CASCADE` over exactly the registry's list — CASCADE removes the
/// dependents while the abandoned instances' pooled connections fail
/// closed on next use), and the next open mints a fresh schema on a
/// fresh (empty) pool (README §The factory contract).
#[tokio::test]
async fn fleet_factory_teardown_is_safe_against_a_panicking_row() {
    if !harness_ready().await {
        return;
    }
    let content = patterned(TUNED_THRESHOLD as usize - 7, 173);
    let abandoned = Key::derive(&content, content.len() as u64);
    let factory = Arc::new(FleetDualTierFactory::new(TUNED_THRESHOLD));

    let task = {
        let factory = Arc::clone(&factory);
        tokio::spawn(async move {
            let (store, _, _) = factory.open_fleet().await.expect("open vends the fleet");
            let (_, pin) = store
                .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
                .await
                .expect("the put publishes");
            drop(pin);
            panic!("a failing assertion panics the row (the factory contract's early exit)");
        })
    };
    assert!(
        task.await.is_err(),
        "the row's panic surfaced as its failure"
    );

    let schemas = factory.tracked_schemas();
    assert!(!schemas.is_empty(), "the panicked row minted a schema");
    factory
        .teardown()
        .await
        .expect("teardown makes sense over the abandoned fleet");
    factory.teardown().await.expect("teardown is idempotent");
    let probe = connect_probe(factory.dsn()).await;
    for schema in &schemas {
        assert!(
            !probe.schema_exists(schema).await,
            "teardown dropped the panicked row's schema: {schema}"
        );
    }

    let (store, _, site) = factory
        .open_fleet()
        .await
        .expect("the next open still vends a fleet");
    assert!(
        !store.has(&abandoned.as_bytes()).await.unwrap(),
        "no residue from the panicked run — isolation is factory-owned"
    );
    assert_eq!(
        probe.pool_rows(&site.schema).await,
        0,
        "the fresh pool is empty"
    );
    factory
        .teardown()
        .await
        .expect("teardown drops the last schema");
}

// ------------------------------------------------------------------
// The consolidation fleet factory — kv = postgres + large = pg-lo on
// one shared pg instance (the composite fleet-validity predicate's
// same-instance arm), gated on the `postgres` + `pg-lo` feature pair.
// ------------------------------------------------------------------
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
mod consolidation {
    use super::*;
    use crate::pg_lo_column::{
        admin_dsn, create_db, db_dsn, drop_db, fresh_db_name, fresh_schema_name, reap_lo_schema,
        TrackedDb,
    };

    /// One open's tracked consolidation site: the dedicated database
    /// (LOs are database-level — the C7 lesson), the kv schema (the
    /// fleet host: pin rows, window, lease) and the pg-lo schema (the
    /// companion table) minted inside it.
    #[derive(Clone)]
    pub(crate) struct ConsolidationSite {
        dsn: String,
        db: String,
        kv_schema: String,
        lo_schema: String,
    }

    impl ConsolidationSite {
        pub(crate) fn dsn(&self) -> &str {
            &self.dsn
        }

        pub(crate) fn kv_schema(&self) -> &str {
            &self.kv_schema
        }

        pub(crate) fn lo_schema(&self) -> &str {
            &self.lo_schema
        }

        pub(crate) fn db(&self) -> &str {
            &self.db
        }
    }

    /// The consolidation fleet factory: one dedicated database per
    /// `open_fleet()`, two engine schemas minted inside it, two fleet
    /// instances (`dual_tier_postgres_pglo_fleet` — the same-instance
    /// clause's probe passing by construction: one instance, two
    /// schemas), teardown reaping the LOs (via the companion table —
    /// the contract authority), dropping exactly the minted schemas,
    /// dropping the database (`WITH (FORCE)`), with a drop-guard
    /// janitor per open backing the abandoned paths (the pg-lo
    /// column's LO-isolation corollary, adapted to the fleet shape).
    pub(crate) struct FleetConsolidationFactory {
        dsn: String,
        seq: AtomicU64,
        dbs: Mutex<Vec<TrackedDb>>,
    }

    impl FleetConsolidationFactory {
        pub(crate) fn new() -> FleetConsolidationFactory {
            FleetConsolidationFactory {
                dsn: crate::postgres_column::harness_dsn()
                    .expect("the harness server is configured for the column"),
                seq: AtomicU64::new(seq_seed()),
                dbs: Mutex::new(Vec::new()),
            }
        }

        /// One dedicated database, two schemas, two fleet instances
        /// over them.
        pub(crate) async fn open_fleet(&self) -> Result<(Store, Store, ConsolidationSite), Error> {
            let maintenance = admin_dsn(&self.dsn);
            let db = fresh_db_name("fleet-consolidation", &self.seq);
            create_db(&maintenance, &db).await?;
            let dsn = db_dsn(&self.dsn, &db);
            let kv_schema = fresh_schema_name("kv", &self.seq);
            let lo_schema = fresh_schema_name("lo", &self.seq);
            self.dbs
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .push(TrackedDb::new(
                    db.clone(),
                    dsn.clone(),
                    vec![kv_schema.clone(), lo_schema.clone()],
                    vec![lo_schema.clone()],
                    &maintenance,
                ));
            let store_a = Store::dual_tier_postgres_pglo_fleet(
                &dsn,
                opts(kv_schema.clone()),
                &dsn,
                opts(lo_schema.clone()),
                TUNED_THRESHOLD,
                WINDOW_DELAY,
                TEST_TTL,
            )
            .await?;
            let store_b = Store::dual_tier_postgres_pglo_fleet(
                &dsn,
                opts(kv_schema.clone()),
                &dsn,
                opts(lo_schema.clone()),
                TUNED_THRESHOLD,
                WINDOW_DELAY,
                TEST_TTL,
            )
            .await?;
            Ok((
                store_a,
                store_b,
                ConsolidationSite {
                    dsn,
                    db,
                    kv_schema,
                    lo_schema,
                },
            ))
        }

        pub(crate) async fn teardown(&self) -> Result<(), Error> {
            let mut dbs = self
                .dbs
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .drain(..)
                .collect::<Vec<_>>();
            let maintenance = admin_dsn(&self.dsn);
            for tracked in dbs.iter_mut() {
                if let Ok(client) = admin_connect(&tracked.dsn).await {
                    for schema in &tracked.lo_schemas {
                        reap_lo_schema(&client, schema).await;
                    }
                    for schema in &tracked.schemas {
                        let sql = format!("DROP SCHEMA IF EXISTS \"{schema}\" CASCADE");
                        let _ = client.batch_execute(&sql).await;
                    }
                }
                drop_db(&maintenance, &tracked.name).await?;
                TrackedDb::disarm(tracked);
            }
            Ok(())
        }
    }
}

#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use consolidation::FleetConsolidationFactory;

impl Probe {
    /// The schema-qualified pool table's count via the probe (the
    /// consolidation rows' kv-side oracle: the kv schema's `blobs`).
    async fn schema_scalar(&self, schema: &str, table: &str) -> usize {
        let sql = format!("SELECT COUNT(*) FROM \"{schema}\".\"{table}\"");
        self.scalar(&sql).await.max(0) as usize
    }
}

// ------------------------------------------------------------------
// The consolidation round-trip row — the joint tx over the shared
// instance, byte-exact large-entry form (the feature pair's home).
// ------------------------------------------------------------------

/// The pg-lo consolidation round-trip **over the shared instance, two
/// instances**: instance a's fleet put routes over the threshold to
/// the large tier — the joint tx carries the LO writes, the companion
/// row insert, and the pin row in **one transaction over the shared
/// instance** (the consolidation posture's whole point), and instance
/// b reads the entry whole: byte-exact, stat-correct. The mediums'
/// oracle legs: one companion row + its LO on the pg-lo medium, one
/// pin row per put on the kv host, zero pool rows (the large tier
/// holds it).
///
/// Contract stamp: store-api.md §Invariants 1 (the round-trip's
/// consolidation leg — byte-identical) and §Invariants 4 (the joint
/// tx holds on every fleet-valid configuration); ADR-009 §2 (the
/// companion table — the addressing authority; delete = lo_unlink +
/// row delete); ADR-008 §2 (the joint entry+pin tx); ADR-012 §4 (the
/// consolidation posture: one tx over the shared instance; the
/// same-instance clause); ADR-012 §3 (the declaration).
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
pub async fn the_pg_lo_consolidation_round_trip_is_byte_exact_over_the_shared_instance(
    factory: &FleetConsolidationFactory,
) {
    let (a, b, site) = factory
        .open_fleet()
        .await
        .expect("open vends the consolidation fleet");
    SetSource::register(&a);

    let content = patterned(2 * 1024 + 7, 167);
    let canonical = Key::derive(&content, content.len() as u64);
    let (key, pin) = a
        .put(Some(content.len() as u64), Cursor::new(content.as_slice()))
        .await
        .expect("the fleet put routes to the consolidation joint tx");

    assert_eq!(key, canonical, "the canonical derivation addresses the put");
    let probe = connect_probe(site.dsn()).await;
    assert_eq!(
        probe.schema_scalar(site.kv_schema(), "blobs").await,
        0,
        "the over-threshold entry rides the large tier — the kv pool \
         holds none"
    );
    assert_eq!(
        probe.schema_scalar(site.lo_schema(), "lo_entries").await,
        1,
        "the joint tx's companion row is on the LO medium — one row"
    );
    assert_eq!(
        probe.pin_rows(site.kv_schema()).await.len(),
        1,
        "the joint tx's pin row is on the kv host — both halves of the \
         one tx, over the one shared instance"
    );

    let (got_len, blob) = b
        .get(&key.as_bytes())
        .await
        .expect("get succeeds")
        .expect("instance b reads the entry from the shared instance");
    assert_eq!(got_len, content.len() as u64);
    assert_eq!(
        blob.read_at(0, got_len).await.expect("the windows read"),
        content.as_slice(),
        "instance b's read is byte-exact — the large-entry form \
         round-trips over the shared instance"
    );
    assert_eq!(
        b.stat(&key.as_bytes()).await.expect("stat succeeds"),
        Some(alkblobs::EntryMeta {
            len: content.len() as u64,
            algorithm: key.algorithm(),
        }),
        "stat carries the length plus algorithm on the consolidation \
         shape"
    );

    drop(pin);
    factory.teardown().await.expect("teardown is idempotent");
}

// ------------------------------------------------------------------
// The runners — the column's tests. Every one gates behind the
// reachability probe (the SKIP posture, shared with the postgres
// column): server-less runs skip cleanly, a skip is not a failure;
// the acceptance run is green against the harness server.
// ------------------------------------------------------------------

fn harness_dsn() -> Option<String> {
    crate::postgres_column::harness_dsn()
}

#[tokio::test]
async fn fleet_validity_every_fleet_valid_configuration_constructs() {
    if !harness_ready().await {
        return;
    }
    let kv_only = FleetKvOnlyFactory::new(KV_ONLY_THRESHOLD);
    let dual_tier = FleetDualTierFactory::new(TUNED_THRESHOLD);
    every_fleet_valid_configuration_constructs(&kv_only, &dual_tier).await;
}

#[tokio::test]
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
async fn fleet_validity_the_consolidation_fleet_constructs() {
    if !harness_ready().await {
        return;
    }
    let consolidation = FleetConsolidationFactory::new();
    the_consolidation_fleet_constructs_over_one_shared_instance(&consolidation).await;
}

#[tokio::test]
async fn fleet_validity_the_constructor_entry_guards_reject_the_degenerate_tunings() {
    if !harness_ready().await {
        return;
    }
    the_constructor_entry_guards_reject_the_degenerate_tunings().await;
}

#[tokio::test]
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
async fn fleet_validity_the_consolidation_same_instance_probe_errors_closed() {
    if !harness_ready().await {
        return;
    }
    let Some(dsn) = harness_dsn() else {
        return;
    };
    the_consolidation_same_instance_probe_errors_closed(&dsn).await;
}

#[tokio::test]
async fn fleet_put_the_joint_entry_and_pin_tx_commits_both_or_neither() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    the_joint_entry_and_pin_tx_commits_both_or_neither(&factory).await;
}

#[tokio::test]
async fn fleet_put_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    a_dedup_reput_upsert_renews_and_a_fleet_reput_composes_a_second_row(&factory).await;
}

#[tokio::test]
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
async fn fleet_put_the_consolidation_round_trip_is_byte_exact_over_the_shared_instance() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetConsolidationFactory::new();
    the_pg_lo_consolidation_round_trip_is_byte_exact_over_the_shared_instance(&factory).await;
}

#[tokio::test]
async fn fleet_ttl_the_pin_guards_ttl_third_renewal_moves_the_rows_expiry() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    the_pin_guards_ttl_third_renewal_moves_the_rows_expiry(&factory).await;
}

#[tokio::test]
async fn fleet_ttl_an_expired_row_is_never_renewed_back_to_life() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    an_expired_row_is_never_renewed_back_to_life(&factory).await;
}

#[tokio::test]
async fn fleet_ttl_a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    a_paused_holder_loses_protection_at_expiry_and_a_reput_recovers(&factory).await;
}

#[tokio::test]
async fn fleet_sweep_two_instance_exact_counts_instance_a_pins_instance_b_sweeps() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    two_instance_exact_counts_instance_a_pins_instance_b_sweeps(&factory).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn fleet_sweep_two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    two_instances_sweep_concurrently_one_leases_the_other_aborts_as_data(&factory).await;
}

#[tokio::test]
async fn fleet_sweep_a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    a_foreign_owners_unexpired_pin_row_refuses_the_direct_delete_until_expiry(&factory).await;
}

#[tokio::test]
async fn fleet_sweep_a_put_landing_during_the_fleet_window_survives_the_executing_sweep() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    a_put_landing_during_the_fleet_window_survives_the_executing_sweep(&factory).await;
}

#[tokio::test]
async fn fleet_sweep_the_maintenance_step_reaps_exactly_the_expired_rows() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    the_maintenance_step_reaps_exactly_the_expired_rows(&factory).await;
}

#[tokio::test]
async fn fleet_sweep_an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetDualTierFactory::new(TUNED_THRESHOLD);
    an_embedder_liveness_table_protects_fleet_wide_and_is_never_owned(&factory).await;
}

#[tokio::test]
async fn fleet_kv_only_mode_rejects_over_threshold_puts_at_put_time() {
    if !harness_ready().await {
        return;
    }
    let factory = FleetKvOnlyFactory::new(KV_ONLY_THRESHOLD);
    crate::rows::kv_only_mode_rejects_over_threshold_puts_at_put_time(&factory).await;
}