alkblobs 0.1.0

//! The constructor family (the constructor modes' named constructors,
//! feature-gated per engine), the introspection accessors the store's
//! paths and the ops layer program through, [`StoreMode`], the
//! constructor-tunable defaults ([`DEFAULT_DELETE_WINDOW_DELAY`],
//! [`DEFAULT_PIN_TTL`]), the shared-media declaration type
//! ([`FleetSharedMedia`]), the composite fleet-validity predicate and
//! its descriptors, the constructor-entry pool-sizing guard, and the
//! construction-time GC-schema dispatch under the one-window-host
//! rule. The engine/feature-gated construction test sections ride
//! this file.

use self::StoreMode::{DualTier, KvOnly, MemOnly};
use super::engines::{KvEngine, LargeEngine};
#[cfg(feature = "postgres")]
use super::fleet::mint_owner_id;
use super::fleet::FleetState;
use super::Store;
#[cfg(all(test, feature = "postgres", feature = "pg-lo"))]
use crate::backend::Backend;
use crate::dispatch::{route, Tier};
#[cfg(all(
    test,
    any(feature = "postgres", all(feature = "pg-lo", feature = "mem"))
))]
use crate::facade::LivenessRegistration;
#[cfg(all(
    test,
    any(feature = "postgres", all(feature = "pg-lo", feature = "mem"))
))]
use crate::facade::LivenessSource;
#[cfg(all(feature = "ops", feature = "postgres"))]
use crate::gc::EngineFleetState;
use crate::gc::GcHome;
#[cfg(all(
    test,
    any(
        feature = "sqlite",
        feature = "postgres",
        all(feature = "pg-lo", feature = "mem")
    )
))]
use crate::keys::Key;
#[cfg(any(feature = "mem", feature = "sqlite", feature = "postgres"))]
use crate::local::LocalBackend;
#[cfg(feature = "mem")]
use crate::mem::MemBackend;
#[cfg(any(feature = "postgres", all(feature = "pg-lo", feature = "mem")))]
use crate::pg_common::PgOpts;
#[cfg(all(feature = "pg-lo", any(feature = "mem", feature = "postgres")))]
use crate::pg_lo::PgLoBackend;
#[cfg(feature = "postgres")]
use crate::postgres::PostgresBackend;
#[cfg(feature = "sqlite")]
use crate::sqlite::SqliteBackend;
use crate::Error;
#[cfg(all(
    test,
    any(feature = "postgres", all(feature = "pg-lo", feature = "mem"))
))]
use std::collections::BTreeSet;
#[cfg(any(feature = "mem", feature = "postgres", feature = "sqlite"))]
use std::path::PathBuf;
#[cfg(any(
    feature = "postgres",
    all(
        test,
        any(feature = "postgres", all(feature = "pg-lo", feature = "mem"))
    )
))]
use std::sync::Arc;
use std::time::Duration;

/// The shipped default **delete-window delay** (ADR-008 §2): the
/// staged candidates in the durable window become visible for
/// execution this long after the sweep staged them — 5 minutes,
/// constructor-tunable per deployment. `Duration::ZERO` is a legal
/// degenerate tuning (execute-on-next-sweep; the test tuning). The
/// sqlite constructors take the delay explicitly and their docs point
/// here as the shipped default; the mem/local inline form has no
/// delay in effect and the constant is irrelevant to it.
pub const DEFAULT_DELETE_WINDOW_DELAY: Duration = Duration::from_secs(5 * 60);

/// The shipped default **pin TTL** (ADR-008 §2): a fleet mode's pin
/// rows renew to `now + pin_ttl` on the guard's TTL/3 cadence — 60
/// seconds, the fleet constructors' default tuning (`FleetState`'s
/// carried value; the engine's `pin_renew` successor-time body reads
/// the TTL its backend was constructed with). Constructor-tunable
/// per deployment, never API (ADR-008 §2: fleet constructor
/// parameters, not API). `Duration::ZERO` is a constructor error —
/// the TTL/3 renewal cadence degenerates at zero and a zero TTL
/// expires every pin at once.
#[cfg(feature = "postgres")]
pub const DEFAULT_PIN_TTL: Duration = Duration::from_secs(crate::postgres::DEFAULT_PIN_TTL_SECS);

/// The operator's explicit **shared-media assertion** (ADR-008 §3's
/// enforceable-seam bullet): the declaration a `local` large tier
/// needs under `fleet: true` — the operator's assertion that the
/// constructor's `local` root is fleet-shared media. **Named design
/// point — the declaration's type shape:** a dedicated sealed type
/// with a private seal field, not a bare flag — the caller cannot
/// construct it accidentally and must name [`FleetSharedMedia::asserted`]
/// at the call site (the ADR's "explicit declaration" wording's
/// reading: a named type the call site cannot misread), while the
/// predicate-level alternative (`LocalUndeclared`) has no constructor
/// to reach it from. The assertion asserts — cross-node truth ("is
/// the declared media actually shared?") remains the deployment's
/// verified invariant, the detection symptom documented (ADR-008 §3's
/// seam; ADR-012 §3's misconfiguration reading): the constructor
/// cannot prove cross-node fact, only admit the declared posture.
#[cfg(feature = "postgres")]
pub struct FleetSharedMedia {
    _seal: (),
}

#[cfg(feature = "postgres")]
impl FleetSharedMedia {
    /// Mint the shared-media assertion — the operator's explicit
    /// declaration at the fleet constructor's call site (the type's
    /// sealed shape makes this the only minting path).
    pub fn asserted() -> Self {
        Self { _seal: () }
    }
}

/// The constructor modes (ADR-011 §4's pinned set; ADR-012 §6.3's
/// dual-tier ruling). Three variants, ADR-pinned — a new mode is a new
/// ADR.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StoreMode {
    /// kv + large, size-threshold dispatch between them (the default).
    DualTier,
    /// kv alone; over-threshold puts reject at put time (ADR-012 §6.2).
    KvOnly,
    /// The mem engine alone, no dispatch threshold in effect — never
    /// production (ADR-011 §4).
    MemOnly,
}

/// The composite fleet-validity predicate's **kv descriptor** — the
/// engine-set shape the predicate reasons over, derived off the
/// engine representation ([`FleetKvShape::of`]) like every other
/// dispatch in this store, never a flag.
// consumer: the `_fleet` constructors' admission gate — the dispatch's
// callers compile only with feature `postgres`; builds without it carry
// the closed vocabulary only.
#[allow(dead_code)]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum FleetKvShape {
    /// The fleet kv engine (ADR-011 §4's constructor table's only
    /// fleet-valid kv row).
    Postgres,
    /// The durable single-node kv engine — never fleet-valid (a
    /// sqlite fleet is a recorded non-option, ADR-010 §3).
    Sqlite,
    /// The contract-reference ephemeral engine — never fleet-valid
    /// (ADR-011 §4).
    Mem,
}

/// The composite fleet-validity predicate's **large descriptor**.
// consumer: the `_fleet` constructors' admission gate — the dispatch's
// callers compile only with feature `postgres`; builds without it carry
// the closed vocabulary only.
#[allow(dead_code)]
#[derive(Debug, Clone, Copy)]
enum FleetLargeShape {
    /// No large tier — the kv-only re-routing posture (ADR-008 §3):
    /// pool large content rides the ops surface.
    None,
    /// The `local` engine with the operator's shared-media
    /// declaration ([`FleetSharedMedia`]).
    LocalSharedMedia,
    /// The `local` engine undeclared — the predicate's rejected arm:
    /// a fleet-mode constructor requires the explicit shared-media
    /// assertion (ADR-008 §3).
    LocalUndeclared,
    /// The pg-lo engine — fleet-valid under ADR-012 §4's
    /// same-instance clause (the constructor-level probe).
    PgLo,
}

/// **The fleet sweep lease's pool-sizing guard** (the sweep-lease
/// design point's constructor-level enforcement, ADR-013 §2): the SQL
/// lease holds one pooled checkout for the sweep's duration, and the
/// fleet sweep body's other engine clauses ride further checkouts
/// under the held lease — a single-connection pool would make the
/// sweep acquire-wait against itself, a deadlock no sweep may
/// tolerate. A sequential fleet sweep body holds exactly two
/// concurrent checkouts (the lease's plus the current clause's), so
/// `max_size >= 2` is the floor; the shipped default (8) provides
/// headroom. Enforced at every `_fleet` constructor's entry — errors
/// closed, before any connection attempt.
#[cfg(feature = "postgres")]
fn fleet_pool_size_guard(opts: &PgOpts) -> Result<(), Error> {
    if opts.max_size < 2 {
        return Err(Error::Io(
            "fleet: the kv pool must size at least 2 connections — the sweep lease holds \
             its checkout for the sweep's duration (ADR-013 §2) and the sweep body's other \
             clauses ride further checkouts under it; a single-connection pool would \
             deadlock the fleet sweep"
                .to_owned(),
        ));
    }
    Ok(())
}

/// **The composite fleet-validity predicate** — backends-and-dispatch.md's
/// one-sentence rule, both tiers, as code: a configuration is
/// fleet-valid iff **kv = postgres AND (large = pg-lo on the same pg
/// instance as the kv engine OR large = `local` on declared shared
/// media OR large = none)**. `None` = fleet-valid; `Some` = the
/// constructor error `fleet: true` carries (ADR-012 §3). The
/// enforcement map: the `_fleet` constructors are named only for the
/// shapes the predicate accepts — the structural clauses (kv ≠
/// postgres, mem anywhere, `local` without the shared-media
/// declaration) have no fleet constructor to reach them through (the
/// recorded "not fleet-anything" postures carry no fleet parameter)
/// and are enforced by that constructor-set shape plus this
/// predicate's rejection arms (unit-tested per clause, the
/// predicate's error being the constructor error a would-be
/// construction carries); the pg-lo shape's runtime-checkable clause
/// — the same-instance probe (ADR-012 §4) — runs at its constructor
/// alongside this call. Derived off the engine representation;
/// crate-internal (fleet parameters, not API).
// consumer: the `_fleet` constructors' admission gate (feature
// `postgres`) and the predicate-level test section that walks the
// rejection arms.
#[allow(dead_code)]
fn fleet_validity_error(kv: FleetKvShape, large: FleetLargeShape) -> Option<Error> {
    match (kv, large) {
        (FleetKvShape::Postgres, FleetLargeShape::None) => None,
        (FleetKvShape::Postgres, FleetLargeShape::LocalSharedMedia) => None,
        (FleetKvShape::Postgres, FleetLargeShape::PgLo) => None,
        (FleetKvShape::Postgres, FleetLargeShape::LocalUndeclared) => Some(Error::Io(
            "fleet: the local large tier rejects without the shared-media declaration \
             (ADR-008 \u{a7}3): a fleet-mode constructor requires the explicit \
             FleetSharedMedia assertion"
                .to_owned(),
        )),
        (kv, _) => Some(Error::Io(format!(
            "fleet: the kv engine {kv:?} rejects the fleet declaration (ADR-012 \u{a7}3): \
             kv = postgres is the composite fleet-validity rule's kv clause — \
             {kv:?} is never fleet-valid"
        ))),
    }
}

impl FleetKvShape {
    /// The predicate kv descriptor derived off the engine
    /// representation.
    #[cfg(any(feature = "mem", feature = "postgres", feature = "sqlite"))]
    // consumer: the `_fleet` constructors' engine-shape derivation
    // (feature `postgres`).
    #[allow(dead_code)]
    fn of(engine: &KvEngine) -> FleetKvShape {
        match engine {
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => FleetKvShape::Mem,
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(_) => FleetKvShape::Postgres,
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => FleetKvShape::Sqlite,
        }
    }
}

/// **The GC schema's engine-neutral construction call under the
/// one-window-host rule (ADR-012 §5)** — the rule's store-core
/// dispatch, run at construction exactly once per instance: the kv
/// engine's arm when the kv engine is SQL-backed (its
/// `ensure_gc_schema` creates its applicability row's tables; the
/// dual-SQL instance's pg-lo window clauses stay dormant — never
/// consulted here), the large tier's arm only when the kv engine is
/// not SQL-backed (kv = mem + large = pg-lo — the pg-lo host's call
/// creating `gc_window` in the pg-lo schema). The in-process
/// combinations (mem alone, local large) are structurally nothing: no
/// GC schema exists on their side of the rule. The dispatch keys off
/// the engine representation, never a flag, and consults the engine
/// enums' own `ensure_gc_schema` methods — one dispatch. (Review-001
/// D-3's resolution: the construction call here previously
/// re-implemented the method bodies' dispatch verbatim with the
/// zero-caller twins' docs claiming this fn consulted them — routing
/// through the methods made the claim true and left one dispatch.)
// consumer: the SQL-backed kv constructors and the large-host arm's
// pg-lo constructor — a build constructible only in-process (mem-only)
// runs the call structurally nothing and compiles it unread.
#[allow(dead_code)]
async fn ensure_gc_schema_at_construction(
    kv: &KvEngine,
    large: Option<&LargeEngine>,
) -> Result<(), Error> {
    // `large` is consulted only by the mem arm (the one-window-host
    // rule's large-host arm); absent feature `mem` no arm reads it.
    #[cfg(not(feature = "mem"))]
    let _ = large;
    match kv {
        #[cfg(feature = "mem")]
        KvEngine::Mem(_) => match large {
            Some(engine) => engine.ensure_gc_schema().await,
            None => Ok(()),
        },
        #[cfg(feature = "postgres")]
        KvEngine::Postgres(_) => kv.ensure_gc_schema().await,
        #[cfg(feature = "sqlite")]
        KvEngine::Sqlite(_) => kv.ensure_gc_schema().await,
        #[cfg(not(any(feature = "mem", feature = "sqlite", feature = "postgres")))]
        _ => Err(super::engines::no_kv_engine()),
    }
}

impl Store {
    /// **mem-only mode** (ADR-011 §4): the mem engine alone, no
    /// dispatch threshold in effect — a testing/embedder-ephemeral
    /// posture, never production. Every length routes to the (sole)
    /// mem engine; there is no large tier and nothing to reject.
    #[cfg(feature = "mem")]
    pub fn mem_only() -> Store {
        Store {
            kv: KvEngine::Mem(MemBackend::default()),
            large: None,
            threshold: crate::dispatch::DEFAULT_DISPATCH_THRESHOLD,
            window_delay: DEFAULT_DELETE_WINDOW_DELAY,
            mode: MemOnly,
            gc: GcHome::default(),
            fleet: None,
        }
    }

    /// **kv-only mode** (ADR-011 §4; ADR-012 §6.2) with the mem kv
    /// engine: no large tier; `threshold` is the put-time rejection
    /// point — known-length puts over it reject immediately,
    /// unknown-length puts at mid-stream threshold overflow (the put
    /// seam's behavior; zero residue on either rejection). Routing in
    /// this mode is kv for every length.
    #[cfg(feature = "mem")]
    pub fn kv_only_mem(threshold: u64) -> Store {
        Store {
            kv: KvEngine::Mem(MemBackend::default()),
            large: None,
            threshold,
            window_delay: DEFAULT_DELETE_WINDOW_DELAY,
            mode: KvOnly,
            gc: GcHome::default(),
            fleet: None,
        }
    }

    /// **dual-tier mode** (the default, ADR-011 §4) in the wave-2
    /// constructible shape — `kv = mem` + `large = local`, the
    /// dispatch-coverage test shape ADR-012 §6.3 names: both tiers
    /// load-bearing, routed by size-threshold dispatch at `threshold`
    /// (constructor-tunable; [`crate::dispatch::DEFAULT_DISPATCH_THRESHOLD`]
    /// is the shipped default).
    ///
    /// Constructing the large engine creates the root directory —
    /// fallible I/O, hence a fallible constructor. The mem kv engine
    /// cannot fail.
    ///
    /// Not fleet-anything: mem is never fleet-valid (ADR-011 §4).
    #[cfg(feature = "mem")]
    pub async fn dual_tier_mem_local(
        root: impl Into<PathBuf>,
        threshold: u64,
    ) -> Result<Store, Error> {
        Ok(Store {
            kv: KvEngine::Mem(MemBackend::default()),
            large: Some(LargeEngine::Local(LocalBackend::new(root).await?)),
            threshold,
            window_delay: DEFAULT_DELETE_WINDOW_DELAY,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// **kv-only mode** (ADR-011 §4; ADR-012 §6.2) with the sqlite kv
    /// engine — the durable single-node kv-only shape: no large tier;
    /// `threshold` is the put-time rejection point (known-length puts
    /// over it reject immediately, unknown-length puts at mid-stream
    /// threshold overflow — the put seam's behavior, identical to the
    /// mem kv-only wiring; zero residue on either rejection). Routing
    /// in this mode is kv for every length.
    ///
    /// Opening (or creating) the database file is fallible I/O; the
    /// constructor also runs the GC schema's engine-neutral call
    /// (`ensure_gc_schema` — sqlite's body creates `gc_window` only,
    /// ADR-012 §5), so a construction failure is `Err(Error::Io)` per
    /// the fallible-constructor pattern `dual_tier_mem_local` set.
    ///
    /// `window_delay` (ADR-008 §2) is the durable window's cadence
    /// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
    /// `Duration::ZERO` the legal degenerate tuning. Not fleet-
    /// anything: a sqlite fleet is a recorded non-option (ADR-010 §3).
    #[cfg(feature = "sqlite")]
    pub async fn kv_only_sqlite(
        database: impl Into<PathBuf>,
        threshold: u64,
        window_delay: Duration,
    ) -> Result<Store, Error> {
        let kv = KvEngine::Sqlite(SqliteBackend::new(database).await?);
        ensure_gc_schema_at_construction(&kv, None).await?;
        Ok(Store {
            kv,
            large: None,
            threshold,
            window_delay,
            mode: KvOnly,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// **dual-tier mode** (the default, ADR-011 §4) in the default-build
    /// production shape — `kv = sqlite` + `large = local` (feature
    /// `sqlite` is default-on, ADR-013 §4): size-threshold dispatch at
    /// `threshold` (constructor-tunable;
    /// [`crate::dispatch::DEFAULT_DISPATCH_THRESHOLD`] is the shipped
    /// default), at-or-below on the kv tier, over-threshold staged into
    /// the large tier.
    ///
    /// Fallible on two accounts (file I/O): opening/creating the pool
    /// database — which also runs the GC schema's engine-neutral call
    /// (`ensure_gc_schema`; sqlite's body creates `gc_window` only,
    /// ADR-012 §5's one-window-host rule: the durable window is the kv
    /// engine's when the kv engine is SQL-backed) — and creating the
    /// local root. A construction failure is `Err(Error::Io)` per the
    /// fallible-constructor pattern `dual_tier_mem_local` set.
    ///
    /// `window_delay` (ADR-008 §2) is the durable window's cadence
    /// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
    /// `Duration::ZERO` the legal degenerate tuning. Not fleet-
    /// anything: a sqlite fleet is a recorded non-option (ADR-010 §3).
    #[cfg(feature = "sqlite")]
    pub async fn dual_tier_sqlite_local(
        database: impl Into<PathBuf>,
        root: impl Into<PathBuf>,
        threshold: u64,
        window_delay: Duration,
    ) -> Result<Store, Error> {
        let kv = KvEngine::Sqlite(SqliteBackend::new(database).await?);
        let large = LargeEngine::Local(LocalBackend::new(root).await?);
        ensure_gc_schema_at_construction(&kv, Some(&large)).await?;
        Ok(Store {
            kv,
            large: Some(large),
            threshold,
            window_delay,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// **kv-only mode** (ADR-011 §4; ADR-012 §6.2) with the postgres kv
    /// engine — the fleet host's kv-only shape: no large tier;
    /// `threshold` is the put-time rejection point (known-length puts
    /// over it reject immediately, unknown-length puts at mid-stream
    /// threshold overflow — the put seam's behavior, identical to the
    /// mem and sqlite kv-only wiring; zero residue on either
    /// rejection). Routing in this mode is kv for every length.
    ///
    /// `config` is the connection settings — the tokio-postgres
    /// [`tokio_postgres::Config`]-parseable DSN form; `opts` carries
    /// the engine-configuration knobs (schema name, pool size,
    /// `synchronous_commit`), riding the constructor parameters, never
    /// invented defaults. Opening the engine (config parse, pool
    /// build, connect, the bootstrap DDL) is fallible I/O; the
    /// constructor also runs the GC schema's engine-neutral call
    /// (`ensure_gc_schema` — the postgres row's body creates
    /// `gc_pins` + `gc_window`, the fleet host's both-tables row,
    /// ADR-012 §5), so a construction failure is `Err(Error::Io)` per
    /// the fallible-constructor pattern `dual_tier_mem_local` set.
    ///
    /// `window_delay` (ADR-008 §2) is the durable window's cadence
    /// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
    /// `Duration::ZERO` the legal degenerate tuning.
    ///
    /// **Non-fleet (ADR-012 §3):** this constructor is implicitly
    /// `fleet: false` — the ADR-005 in-process protocol applies
    /// verbatim regardless of engine (pins are in-process guards, the
    /// sweep lease is the in-process lock, none of the
    /// `EngineFleetState` clauses run), behavior byte-unchanged. The
    /// `fleet: true` declaration is this constructor's [`Store::kv_only_postgres_fleet`]
    /// twin (wave 6's landed declaration — the composite
    /// fleet-validity predicate plus the fleet parameters), slotted in
    /// without reshaping; this signature carries no fleet parameter.
    #[cfg(feature = "postgres")]
    pub async fn kv_only_postgres(
        config: &str,
        opts: PgOpts,
        threshold: u64,
        window_delay: Duration,
    ) -> Result<Store, Error> {
        let kv = KvEngine::Postgres(Arc::new(PostgresBackend::new(config, opts).await?));
        ensure_gc_schema_at_construction(&kv, None).await?;
        Ok(Store {
            kv,
            large: None,
            threshold,
            window_delay,
            mode: KvOnly,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// **dual-tier mode** (the default, ADR-011 §4) in the
    /// fleet-capable-kv production shape — `kv = postgres` +
    /// `large = local`: size-threshold dispatch at `threshold`
    /// (constructor-tunable;
    /// [`crate::dispatch::DEFAULT_DISPATCH_THRESHOLD`] is the shipped
    /// default), at-or-below on the kv tier, over-threshold staged
    /// into the large tier.
    ///
    /// `config` is the connection settings — the tokio-postgres
    /// [`tokio_postgres::Config`]-parseable DSN form; `opts` carries
    /// the engine-configuration knobs (schema name, pool size,
    /// `synchronous_commit`), riding the constructor parameters, never
    /// invented defaults.
    ///
    /// Fallible on two accounts (I/O): opening the engine (config
    /// parse, pool build, connect, the bootstrap DDL) — which also
    /// runs the GC schema's engine-neutral call (`ensure_gc_schema`;
    /// the postgres row's body creates `gc_pins` + `gc_window`,
    /// ADR-012 §5's one-window-host rule: the durable window is the
    /// kv engine's when the kv engine is SQL-backed, so the postgres
    /// host creates both GC tables per its applicability row) — and
    /// creating the local root. A construction failure is
    /// `Err(Error::Io)` per the fallible-constructor pattern
    /// `dual_tier_mem_local` set.
    ///
    /// `window_delay` (ADR-008 §2) is the durable window's cadence
    /// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
    /// `Duration::ZERO` the legal degenerate tuning.
    ///
    /// **Non-fleet (ADR-012 §3):** this constructor is implicitly
    /// `fleet: false` — the ADR-005 in-process protocol applies
    /// verbatim regardless of engine (pins are in-process guards, the
    /// sweep lease is the in-process lock, none of the
    /// `EngineFleetState` clauses run), behavior byte-unchanged. The
    /// `fleet: true` declaration is this constructor's [`Store::dual_tier_postgres_local_fleet`]
    /// twin (wave 6's landed declaration — the composite
    /// fleet-validity predicate plus the fleet parameters, the
    /// shared-media assertion included), slotted in without
    /// reshaping; this signature carries no fleet parameter.
    #[cfg(feature = "postgres")]
    pub async fn dual_tier_postgres_local(
        config: &str,
        opts: PgOpts,
        root: impl Into<PathBuf>,
        threshold: u64,
        window_delay: Duration,
    ) -> Result<Store, Error> {
        let kv = KvEngine::Postgres(Arc::new(PostgresBackend::new(config, opts).await?));
        let large = LargeEngine::Local(LocalBackend::new(root).await?);
        ensure_gc_schema_at_construction(&kv, Some(&large)).await?;
        Ok(Store {
            kv,
            large: Some(large),
            threshold,
            window_delay,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// **dual-tier mode** (the default, ADR-011 §4) in the
    /// **consolidation posture** (ADR-009: "one engine to operate") —
    /// `kv = postgres` + `large = pg-lo`: size-threshold dispatch at
    /// `threshold` (constructor-tunable;
    /// [`crate::dispatch::DEFAULT_DISPATCH_THRESHOLD`] is the shipped
    /// default), at-or-below on the kv tier, over-threshold staged
    /// into the large tier's held-transaction staging cursor.
    ///
    /// **Connection-settings design point (two pairs):** this
    /// constructor takes a DSN + [`PgOpts`] pair per engine (kv first,
    /// large second). Cross-instance kv=postgres + large=pg-lo is a
    /// valid non-fleet construction (ADR-012 §4); the consolidation
    /// shape passes **the same DSN and opts to both** — the pool
    /// content lives in one pg instance the kv tier rides (the
    /// recorded corollary for a future conformance-suite factory:
    /// minting one engine-owned schema and passing it to both engines
    /// is equally valid — both engines' table names are distinct, so
    /// the one-schema form holds `blobs`, `lo_entries`, and the GC
    /// tables together; the two-schema form splits them — either is a
    /// legitimate mint for this constructor's shape).
    ///
    /// Fallible on I/O — engine opens (config parse, pool build,
    /// connect, the bootstrap DDL) and the GC schema call
    /// (the construction dispatch `ensure_gc_schema_at_construction`: the kv host runs per the
    /// one-window-host rule, creating `gc_pins` + `gc_window` in the
    /// kv schema — the pg-lo window clauses are **dormant** on this
    /// dual-SQL instance, so the pg-lo schema carries no `gc_window`).
    /// A construction failure is `Err(Error::Io)` per the
    /// fallible-constructor pattern `dual_tier_mem_local` set.
    ///
    /// `window_delay` (ADR-008 §2) is the durable window's cadence
    /// knob — [`DEFAULT_DELETE_WINDOW_DELAY`] is the shipped default,
    /// `Duration::ZERO` the legal degenerate tuning.
    ///
    /// **Non-fleet (ADR-012 §3):** this constructor is implicitly
    /// `fleet: false` — the ADR-005 in-process protocol applies
    /// verbatim regardless of engine (pins are in-process guards, the
    /// sweep lease is the in-process lock, none of the
    /// `EngineFleetState` clauses run), behavior byte-unchanged. The
    /// `fleet: true` declaration is this constructor's [`Store::dual_tier_postgres_pglo_fleet`]
    /// twin (wave 6's landed declaration — the composite fleet-validity
    /// predicate plus the fleet parameters, the same-instance probe
    /// included), slotted in without reshaping; this signature carries
    /// no fleet parameter. This constructor completed
    /// the fleet-valid engine set — the module doc's recorded forward
    /// reference, resolved: with pg-lo's landing, every fleet-valid
    /// engine had its named constructor before the declaration's
    /// predicate and parameters followed.
    #[cfg(all(feature = "postgres", feature = "pg-lo"))]
    pub async fn dual_tier_postgres_pglo(
        config: &str,
        opts: PgOpts,
        large_config: &str,
        large_opts: PgOpts,
        threshold: u64,
        window_delay: Duration,
    ) -> Result<Store, Error> {
        let kv = KvEngine::Postgres(Arc::new(PostgresBackend::new(config, opts).await?));
        let large = LargeEngine::PgLo(PgLoBackend::new(large_config, large_opts).await?);
        ensure_gc_schema_at_construction(&kv, Some(&large)).await?;
        Ok(Store {
            kv,
            large: Some(large),
            threshold,
            window_delay,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// **kv-only mode, `fleet: true`** (ADR-012 §3's declaration;
    /// ADR-011 §4; ADR-012 §6.2) — the fleet re-routing client posture
    /// (ADR-008 §3): the postgres kv engine alone, no large tier — the
    /// composite fleet-validity predicate's `large = none` arm, the
    /// client node's shape whose over-threshold pool content rides the
    /// ops surface (the re-routing composition, the ops module's
    /// landed wiring — wave 7; the
    /// over-threshold put-time rejection, ADR-012 §6.2, unchanged
    /// under fleet).
    ///
    /// `config`/`opts`/`threshold`/`window_delay` carry exactly the
    /// [`Store::kv_only_postgres`] twin's meaning (the non-fleet
    /// posture stays byte-unchanged there). `pin_ttl` (ADR-008 §2) is
    /// the fleet constructor parameter — the pin rows' renewal window
    /// the store owns and renews on the TTL/3 cadence, the engine's
    /// `pin_renew` successor-time body reading the TTL this
    /// constructor passed to its backend; [`DEFAULT_PIN_TTL`] is the
    /// shipped default (60 s). `Duration::ZERO` is a constructor
    /// error at entry (the TTL/3 cadence degenerates at zero; a zero
    /// TTL expires every pin at once) — validated before any I/O.
    ///
    /// **The composite fleet-validity predicate runs at the
    /// constructor** (`fleet_validity_error`, keyed off the engine
    /// representation like every dispatch in this store): kv = postgres
    /// (structural — the only kv engine a fleet constructor names;
    /// the predicate's mem/sqlite rejection arms are the clause's
    /// testing home) and `large = none` — the re-routing arm. The
    /// store carries the fleet state on success (`Store::fleet`):
    /// the declaration's flag, the pin TTL, and the store-instance id
    /// (`mint_owner_id`).
    #[cfg(feature = "postgres")]
    pub async fn kv_only_postgres_fleet(
        config: &str,
        opts: PgOpts,
        threshold: u64,
        window_delay: Duration,
        pin_ttl: Duration,
    ) -> Result<Store, Error> {
        if pin_ttl.is_zero() {
            return Err(Error::Io(
                "fleet: pin_ttl must be positive — the TTL/3 renewal cadence degenerates \
                 at zero, and a zero-TTL pin expires at once"
                    .to_owned(),
            ));
        }
        fleet_pool_size_guard(&opts)?;
        let kv = KvEngine::Postgres(Arc::new(
            PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?,
        ));
        if let Some(err) = fleet_validity_error(FleetKvShape::of(&kv), FleetLargeShape::None) {
            return Err(err);
        }
        ensure_gc_schema_at_construction(&kv, None).await?;
        Ok(Store {
            kv,
            large: None,
            threshold,
            window_delay,
            mode: KvOnly,
            gc: GcHome::default(),
            fleet: Some(FleetState {
                pin_ttl,
                owner: mint_owner_id(),
            }),
        })
    }

    /// **dual-tier mode, `fleet: true`** (ADR-012 §3's declaration;
    /// ADR-011 §4) — the fleet shared-media posture: `kv = postgres` +
    /// `large = local`, the composite fleet-validity predicate's
    /// declared-shared-media arm (ADR-008 §3's enforceable seam). The
    /// joint entry+pin transaction is unachievable across media, so
    /// the put path's ordering under fleet is pin-row-first,
    /// publish-second (ADR-012 §4; the put path's wiring is
    /// `store-fleet-put`'s).
    ///
    /// `config`/`opts`/`root`/`threshold`/`window_delay` carry exactly
    /// the [`Store::dual_tier_postgres_local`] twin's meaning (the
    /// non-fleet posture stays byte-unchanged there). `pin_ttl` is the
    /// fleet constructor parameter ([`DEFAULT_PIN_TTL`] the shipped
    /// default; `Duration::ZERO` a constructor error at entry). **The
    /// large-tier declaration** rides [`FleetSharedMedia`] — the
    /// operator's explicit shared-media assertion, a named type the
    /// call site cannot construct accidentally; `local` without it
    /// never reaches a fleet constructor (the predicate's
    /// `LocalUndeclared` arm is the rule's textual form). The
    /// composite fleet-validity predicate runs at the constructor
    /// (kv = postgres and the declared shared media); the cross-node
    /// truth behind the assertion stays the deployment's verified
    /// invariant (ADR-008 §3's seam). The store carries the fleet
    /// state on success (`Store::fleet`).
    #[cfg(feature = "postgres")]
    pub async fn dual_tier_postgres_local_fleet(
        config: &str,
        opts: PgOpts,
        root: impl Into<PathBuf>,
        threshold: u64,
        window_delay: Duration,
        pin_ttl: Duration,
        shared_media: FleetSharedMedia,
    ) -> Result<Store, Error> {
        if pin_ttl.is_zero() {
            return Err(Error::Io(
                "fleet: pin_ttl must be positive — the TTL/3 renewal cadence degenerates \
                 at zero, and a zero-TTL pin expires at once"
                    .to_owned(),
            ));
        }
        fleet_pool_size_guard(&opts)?;
        let kv = KvEngine::Postgres(Arc::new(
            PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?,
        ));
        let large = LargeEngine::Local(LocalBackend::new(root).await?);
        let _ = shared_media;
        if let Some(err) =
            fleet_validity_error(FleetKvShape::of(&kv), FleetLargeShape::LocalSharedMedia)
        {
            return Err(err);
        }
        ensure_gc_schema_at_construction(&kv, Some(&large)).await?;
        Ok(Store {
            kv,
            large: Some(large),
            threshold,
            window_delay,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: Some(FleetState {
                pin_ttl,
                owner: mint_owner_id(),
            }),
        })
    }

    /// **dual-tier mode, `fleet: true`** (ADR-012 §3's declaration;
    /// ADR-009's consolidation posture) — `kv = postgres` + `large =
    /// pg-lo`, **the same pg instance** (ADR-012 §4's checkable
    /// clause, the predicate's runtime arm): the joint entry+pin tx
    /// the fleet put path drives (`store-fleet-put`) is expressible
    /// over the shared instance.
    ///
    /// `config`/`opts`/`large_config`/`large_opts`/`threshold`/
    /// `window_delay` carry exactly the [`Store::dual_tier_postgres_pglo`]
    /// twin's meaning (cross-instance kv=postgres + large=pg-lo stays
    /// valid only as the non-fleet twin). `pin_ttl` is the fleet
    /// constructor parameter ([`DEFAULT_PIN_TTL`] the shipped
    /// default; `Duration::ZERO` a constructor error at entry) — it
    /// configures the **kv engine's** backend (`pin_renew` is the kv
    /// host's clause; the pg-lo engine hosts no fleet state, ADR-012
    /// §5).
    ///
    /// **The pg-lo same-instance probe (named design point):** the
    /// instance's `system_identifier` (via `pg_control_system()`,
    /// PG14+) read across one pooled connection per engine and
    /// compared — server-side, immune to DSN aliases (textual DSN
    /// comparison was rejected: hostname vs IP vs alias cannot decide
    /// instance identity). A mismatch, **or any probe failure** (a
    /// pre-PG14 server, a denied function call, an unreachable
    /// instance), is a constructor error — the probe errors closed,
    /// never a silent pass. The composite fleet-validity predicate
    /// runs at the constructor alongside the probe; the store carries
    /// the fleet state on success (`Store::fleet`).
    #[cfg(all(feature = "postgres", feature = "pg-lo"))]
    pub async fn dual_tier_postgres_pglo_fleet(
        config: &str,
        opts: PgOpts,
        large_config: &str,
        large_opts: PgOpts,
        threshold: u64,
        window_delay: Duration,
        pin_ttl: Duration,
    ) -> Result<Store, Error> {
        if pin_ttl.is_zero() {
            return Err(Error::Io(
                "fleet: pin_ttl must be positive — the TTL/3 renewal cadence degenerates \
                 at zero, and a zero-TTL pin expires at once"
                    .to_owned(),
            ));
        }
        fleet_pool_size_guard(&opts)?;
        let kv_backend = Arc::new(PostgresBackend::new_with_pin_ttl(config, opts, pin_ttl).await?);
        let large_backend = PgLoBackend::new(large_config, large_opts).await?;
        let kv_id = kv_backend.system_identifier().await?;
        let large_id = large_backend.system_identifier().await?;
        let kv = KvEngine::Postgres(kv_backend);
        let large = LargeEngine::PgLo(large_backend);
        if let Some(err) = fleet_validity_error(FleetKvShape::of(&kv), FleetLargeShape::PgLo) {
            return Err(err);
        }
        if kv_id != large_id {
            return Err(Error::Io(format!(
                "fleet: the kv engine and the pg-lo large engine resolve to different pg \
                 instances (system_identifier {kv_id} vs {large_id}) — ADR-012 \u{a7}4's \
                 same-instance clause; cross-instance pg-lo is valid only as fleet: false"
            )));
        }
        ensure_gc_schema_at_construction(&kv, Some(&large)).await?;
        Ok(Store {
            kv,
            large: Some(large),
            threshold,
            window_delay,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: Some(FleetState {
                pin_ttl,
                owner: mint_owner_id(),
            }),
        })
    }

    /// **dual-tier mode** (the default, ADR-011 §4) in the
    /// one-window-host rule's **named test topology** (ADR-012 §5) —
    /// `kv = mem` + `large = pg-lo`: the durable delete window hosts
    /// on the **large engine** when the kv engine is not SQL-backed;
    /// the constructor runs the pg-lo host's `ensure_gc_schema` call
    /// (the construction dispatch `ensure_gc_schema_at_construction`'s large-host arm —
    /// `gc_window` in the pg-lo schema) and its sweeps run the
    /// durable form on the pg-lo host, with the inline form nowhere in
    /// this shape's story (the mem kv arm of the form branch consults
    /// the large engine precisely when it hosts the window;
    /// ADR-012 §5's non-SQL inline form exists for mem/local *without*
    /// a large SQL host).
    ///
    /// `config` is the connection settings — the tokio-postgres
    /// [`tokio_postgres::Config`]-parseable DSN form; `opts` carries
    /// the engine-configuration knobs (schema name, pool size,
    /// `synchronous_commit`), riding the constructor parameters, never
    /// invented defaults. The kv tier is the mem engine — fallible
    /// only on the pg-lo engine's I/O (config parse, pool build,
    /// connect, the bootstrap DDL, the GC schema call): a construction
    /// failure is `Err(Error::Io)` per the fallible-constructor
    /// pattern `dual_tier_mem_local` set.
    ///
    /// `window_delay` (ADR-008 §2) is the durable window's cadence
    /// knob on the pg-lo host — [`DEFAULT_DELETE_WINDOW_DELAY`] is the
    /// shipped default, `Duration::ZERO` the legal degenerate tuning.
    ///
    /// Not fleet-anything: mem is never fleet-valid (ADR-011 §4).
    #[cfg(all(feature = "mem", feature = "pg-lo"))]
    pub async fn dual_tier_mem_pglo(
        config: &str,
        opts: PgOpts,
        threshold: u64,
        window_delay: Duration,
    ) -> Result<Store, Error> {
        let kv = KvEngine::Mem(MemBackend::default());
        let large = LargeEngine::PgLo(PgLoBackend::new(config, opts).await?);
        ensure_gc_schema_at_construction(&kv, Some(&large)).await?;
        Ok(Store {
            kv,
            large: Some(large),
            threshold,
            window_delay,
            mode: DualTier,
            gc: GcHome::default(),
            fleet: None,
        })
    }

    /// The constructor mode (ADR-011 §4's vocabulary) — introspection
    /// for embedders and tests; the dispatch/put/read paths branch on
    /// it crate-internally.
    pub fn mode(&self) -> StoreMode {
        self.mode
    }

    /// The active dispatch threshold — `Some` in dual-tier (the routing
    /// boundary) and kv-only (the put-time rejection point); `None` in
    /// mem-only mode ("no dispatch threshold in effect", ADR-011 §4).
    pub(crate) fn threshold(&self) -> Option<u64> {
        match self.mode {
            DualTier | KvOnly => Some(self.threshold),
            MemOnly => None,
        }
    }

    /// The delete-window delay the constructor took (ADR-008 §2) —
    /// crate-internal: the sweep's SQL-backed form consumes it when
    /// staging candidates (`visible_after = mark_end + window`);
    /// introspection for the construction tests. `Duration::ZERO`
    /// (a constructor-passed value) is the legal degenerate tuning —
    /// execute-on-next-sweep.
    // consumer: the construction test sections' introspection
    // instrument (the window-delay tunings); no lib build consults it.
    #[allow(dead_code)]
    pub(crate) fn window_delay(&self) -> Duration {
        self.window_delay
    }

    /// The store's fleet state — `Some` exactly when the constructor
    /// declared `fleet: true` (the declaration is explicit, never
    /// inferred, ADR-012 §3): the put/sweep paths branch on the
    /// state's presence, never on re-deriving from the engine set.
    /// Crate-internal introspection (`FleetState`): the fleet
    /// parameters are constructor parameters, not API (ADR-008 §3).
    pub(crate) fn fleet(&self) -> Option<&FleetState> {
        self.fleet.as_ref()
    }

    /// **Write-side tier selection for a given content length** — the
    /// dispatch seam's routing half (the put seam programs through
    /// this; read paths do not consult it). Dual-tier:
    /// [`crate::dispatch::route`]'s pure function at the instance's
    /// threshold (exactly-at-threshold stays kv — the pinned boundary
    /// reading). kv-only: kv for every length — the over-threshold
    /// *rejection* is the put seam's behavior keyed off
    /// [`Store::threshold`] (ADR-012 §6.2), not a routing answer.
    /// Mem-only: kv — there is no other tier and no threshold in
    /// effect.
    pub(crate) fn tier_for(&self, len: u64) -> Tier {
        match self.mode {
            DualTier => route(len, self.threshold),
            KvOnly | MemOnly => Tier::Kv,
        }
    }

    /// **The fall-through consult order — kv first** (ADR-012 §6.1's
    /// general rule): every key-addressed op consults the kv tier
    /// first; large comes after ([`Store::large`]). A miss on both
    /// tiers resolves per op's shape (`None` on get/stat,
    /// `Err(Missing)` on read_range/delete).
    pub(crate) fn kv(&self) -> &KvEngine {
        &self.kv
    }

    /// **…then large** — the second consult; `None` in single-tier
    /// modes (mem-only, kv-only), where large-tier content cannot
    /// exist.
    pub(crate) fn large(&self) -> Option<&LargeEngine> {
        self.large.as_ref()
    }

    /// The GC state home (ADR-010 §2: the store core owns all GC
    /// state) — the gc-protocol task's pin registry, liveness sources,
    /// protect callback, and in-process delete window land here and
    /// drive the sweep and direct delete from this facade.
    pub(crate) fn gc(&self) -> &GcHome {
        &self.gc
    }

    /// **The ops module's crate-internal renewal access — `pin_renew`**
    /// (ADR-010 §4: "the ops module is in-crate and crosses the seam by
    /// construction, the same way the put handler holds the `Pin`
    /// guard"): the have handler's renewal step's store-core crossing,
    /// routed through the fleet host. This verb is the crossing's named
    /// shape — the store core owns the seam's reachability, so a
    /// *consumer* never reaches the `EngineFleetState` clause, and no
    /// public API widens for this (ADR-001 §4's facade-consumer rule:
    /// this and [`Store::gc`] are the ops module's only store-core
    /// access crate-internally — the facade plus exactly this seam).
    ///
    /// Non-fleet stores answer `Err`: the pin row's renewal is the
    /// fleet form's act (under `fleet: false` the token is the
    /// in-process guard's custody — nothing expires — and the renewal
    /// probe answers from custody, never through this verb; the
    /// ops-layer custody branch is upstream of this call).
    #[cfg(all(feature = "ops", feature = "postgres"))]
    pub(crate) async fn ops_pin_renew(&self, key: &[u8], owner: &[u8]) -> Result<bool, Error> {
        match self.fleet_sql_host() {
            Some(host) => EngineFleetState::pin_renew(&**host, key, owner).await,
            None => Err(Error::Io(
                "pin renewal outside a declared fleet store: under fleet: false the token \
                 is the in-process guard's custody (nothing expires) — the renewal probe \
                 answers from the ops layer's custody, never through this verb"
                    .to_owned(),
            )),
        }
    }

    /// **The ops module's crate-internal renewal access — `pin_check`**
    /// (ADR-010 §2's engine-state seam; the ADR-008 row classifier):
    /// the have handler's renewal-verification step's store-core
    /// crossing (the clause's named consumer — the wave-6 review's
    /// W6-3 note), routed through the fleet host under the same
    /// ownership rule as [`Store::ops_pin_renew`]: store-core-owned,
    /// never public API, never a backend side-door.
    #[cfg(all(feature = "ops", feature = "postgres"))]
    pub(crate) async fn ops_pin_check(&self, key: &[u8]) -> Result<crate::facade::PinState, Error> {
        match self.fleet_sql_host() {
            Some(host) => EngineFleetState::pin_check(&**host, key).await,
            None => Err(Error::Io(
                "pin check outside a declared fleet store: the pin-row classification \
                 is the fleet form's act (ADR-010 §2) — under fleet: false the in-process \
                 guard's custody is the classification, and the ops layer holds it"
                    .to_owned(),
            )),
        }
    }
}

/// The wave-3 constructor wiring's tests (feature `sqlite`): the
/// sqlite modes construct and introspect correctly, the GC schema ran
/// at construction, the window-delay default is the pinned constant
/// and is constructor-tunable, the kv-only rejection wiring is the
/// mode's put seam unchanged, and the default-build production shape
/// (kv = sqlite + large = local) round-trips end-to-end through both
/// tiers. All-features builds share this module (mem on alongside):
/// the engine accessors carry cfg arms for that.
#[cfg(all(test, feature = "sqlite"))]
mod sqlite_construction_tests {
    use super::*;
    use std::fs;
    use std::io::Cursor;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};

    /// A fresh temp database file (and sibling stage/root dir) per
    /// test; Drop removes the file tree and the WAL sidecars
    /// idempotently (the sqlite engine's test helper's shape).
    struct TempSite {
        database: PathBuf,
        root: PathBuf,
    }

    impl TempSite {
        fn path(&self) -> &Path {
            &self.root
        }

        fn drop_sidecar(&self, suffix: &str) {
            let base = self.database.to_string_lossy().into_owned();
            let _ = fs::remove_file(format!("{base}{suffix}"));
        }
    }

    impl Drop for TempSite {
        fn drop(&mut self) {
            self.drop_sidecar("-wal");
            self.drop_sidecar("-shm");
            let _ = fs::remove_file(&self.database);
            let _ = fs::remove_dir_all(&self.root);
        }
    }

    fn temp_site(tag: &str) -> TempSite {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        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-store-sqlite-{tag}-{}-{nanos:x}-{seq}",
            std::process::id()
        ));
        fs::create_dir_all(&base).expect("temp site created");
        TempSite {
            database: base.join("pool.db"),
            root: base.join("local-root"),
        }
    }

    fn key_for(content: &[u8]) -> Key {
        Key::derive(content, content.len() as u64)
    }

    /// The sqlite kv engine behind the constructed store — the schema
    /// observations' instrument.
    fn sqlite_engine(store: &Store) -> &SqliteBackend {
        match store.kv() {
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(engine) => engine,
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("these tests construct sqlite stores"),
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(_) => panic!("these tests construct sqlite stores"),
        }
    }

    /// The engine-neutral GC-schema call ran at construction: the pool
    /// table exists (the engine's own open), `gc_window` exists (the
    /// store-core call), `gc_pins` never (sqlite's applicability row,
    /// ADR-012 §5's one-window-host rule — the kv engine hosts the
    /// window and nothing else).
    async fn assert_gc_schema_ran(store: &Store) {
        let engine = sqlite_engine(store);
        assert!(
            engine.table_exists("blobs").await.unwrap(),
            "the pool table exists at construction"
        );
        assert!(
            engine.table_exists("gc_window").await.unwrap(),
            "ensure_gc_schema ran at construction: the window table exists in the fresh database"
        );
        assert!(
            !engine.table_exists("gc_pins").await.unwrap(),
            "gc_pins is never created on sqlite (ADR-012 §5)"
        );
    }

    /// The dual-tier construction: both tiers present, mode/threshold
    /// introspection exact, boundary routing deterministic, and the
    /// GC schema ran.
    #[tokio::test]
    async fn dual_tier_sqlite_local_constructs_and_routes() {
        let site = temp_site("dual-tier");
        let store = Store::dual_tier_sqlite_local(
            &site.database,
            site.path(),
            1024,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("dual-tier sqlite+local constructs");
        assert_eq!(store.mode(), DualTier);
        assert_eq!(store.threshold(), Some(1024));
        assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
        assert!(store.large().is_some(), "both tiers present");

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "length {len} routes deterministically"
            );
        }

        assert_gc_schema_ran(&store).await;
    }

    /// The kv-only construction: no large tier, the threshold is the
    /// rejection point's carrier, and the rejection wiring is the put
    /// seam's unchanged — known-length over-threshold rejects
    /// immediately (zero residue), unknown-length at mid-stream
    /// overflow, and exactly-at-threshold puts land.
    #[tokio::test]
    async fn kv_only_sqlite_constructs_and_rejects_over_threshold_puts() {
        let site = temp_site("kv-only");
        let store = Store::kv_only_sqlite(&site.database, 64, DEFAULT_DELETE_WINDOW_DELAY)
            .await
            .expect("kv-only sqlite constructs");
        assert_eq!(store.mode(), KvOnly);
        assert_eq!(store.threshold(), Some(64));
        assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
        assert!(store.large().is_none(), "kv-only has no large tier");
        assert_gc_schema_ran(&store).await;

        for len in [0u64, 63, 64, 65, u64::MAX] {
            assert_eq!(
                store.tier_for(len),
                Tier::Kv,
                "kv-only routes every length kv"
            );
        }

        let err = store
            .put(Some(65), Cursor::new(&b"never read"[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("kv-only mode rejects"),
            "the immediate rejection is the mode's: {err}"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "zero residue"
        );

        let streamed = [3u8; 100];
        let err = store
            .put(None, Cursor::new(&streamed[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("kv-only mode rejects an over-threshold put"),
            "the overflow rejection is the mode's, not the stream's: {err}"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "zero residue"
        );

        let at_threshold = vec![9u8; 64];
        let (key, pin) = store
            .put(Some(64), Cursor::new(&at_threshold[..]))
            .await
            .expect("exactly-at-threshold known put lands");
        assert_eq!(key, key_for(&at_threshold));
        assert!(store.has(&key.as_bytes()).await.unwrap());
        drop(pin);
    }

    /// The window-delay parameter: the shipped default is the pinned
    /// 5-minute constant, the constructor's value passes through the
    /// store's introspection, and `Duration::ZERO` is a legal
    /// degenerate tuning (execute-on-next-sweep).
    #[tokio::test]
    async fn window_delay_default_is_pinned_and_constructor_tunable() {
        assert_eq!(
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::from_secs(5 * 60),
            "the shipped default is 5 minutes (ADR-008 §2)"
        );

        let site = temp_site("delay-default");
        let store = Store::dual_tier_sqlite_local(
            &site.database,
            site.path(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("constructs at the shipped default");
        assert_eq!(store.window_delay(), Duration::from_secs(300));

        let site = temp_site("delay-tuned");
        let store = Store::kv_only_sqlite(&site.database, 64, Duration::from_secs(90))
            .await
            .expect("constructs at a custom tuning");
        assert_eq!(store.window_delay(), Duration::from_secs(90));

        let site = temp_site("delay-zero");
        let store = Store::dual_tier_sqlite_local(&site.database, site.path(), 64, Duration::ZERO)
            .await
            .expect("the degenerate zero tuning is legal");
        assert_eq!(store.window_delay(), Duration::ZERO);

        let site = temp_site("delay-long");
        Store::kv_only_sqlite(&site.database, 64, Duration::from_secs(6000))
            .await
            .expect("a long delay is a legal tuning too");
    }

    /// The default-build production shape round-trips end-to-end: a
    /// sub-threshold put lands on the kv tier (sqlite), an
    /// over-threshold put stages into the large tier (local), both
    /// get back byte-identical, and the pool list is the sorted union
    /// of both tiers.
    #[tokio::test]
    async fn dual_tier_sqlite_shape_round_trips_end_to_end_through_both_tiers() {
        let site = temp_site("round-trip");
        let store = Store::dual_tier_sqlite_local(&site.database, site.path(), 64, Duration::ZERO)
            .await
            .expect("dual-tier constructs");

        let small = b"a sub-threshold blob";
        let (key, pin) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("kv put publishes");
        assert_eq!(key, key_for(small));
        assert!(
            store.kv().has(&key.as_bytes()).await.unwrap(),
            "landed on the kv tier (sqlite)"
        );
        assert!(
            !store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
            "no large-tier copy on the kv route"
        );
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("entry present");
        assert_eq!(len, small.len() as u64);
        assert_eq!(blob.read_at(0, len).await.unwrap(), &small[..]);
        drop(pin);

        let big = vec![42u8; 200];
        let (key, pin) = store
            .put(Some(200), Cursor::new(&big[..]))
            .await
            .expect("large staged put publishes");
        assert_eq!(key, key_for(&big));
        assert!(
            store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
            "landed on the large tier (commit-rename)"
        );
        assert!(
            !store.kv().has(&key.as_bytes()).await.unwrap(),
            "no kv copy on the large route"
        );
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("entry present");
        assert_eq!(len, 200);
        assert_eq!(blob.read_at(0, len).await.unwrap(), big);
        drop(pin);

        let listed = store.list().await.expect("list succeeds");
        assert_eq!(listed.len(), 2, "one entry per tier, both enumerated");
        for window in listed.windows(2) {
            assert!(window[0] < window[1], "list is sorted");
        }

        // A re-put dedups into the existing entry (CAS first-writer-
        // wins, through the facade's put).
        let (_, pin) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("dedup re-put");
        assert_eq!(store.list().await.unwrap().len(), 2, "still two entries");
        drop(pin);
    }
}

/// The wave-4 constructor wiring's tests (feature `postgres`): the
/// postgres modes construct and introspect correctly, the GC schema
/// ran at construction (both GC tables in the fresh schema — the
/// fleet host's both-tables row), the window-delay default is the
/// pinned constant and is constructor-tunable, the kv-only rejection
/// wiring is the mode's put seam unchanged, the fleet-capable-kv
/// shape (kv = postgres + large = local) round-trips end-to-end
/// through both tiers, and the sweep's form branch runs the durable
/// form on the postgres host. Env-driven — the engine tests'
/// convention — skipping cleanly server-less.
#[cfg(all(test, feature = "postgres"))]
mod postgres_construction_tests {
    use super::*;
    use crate::postgres::test_support::SchemaJanitor;
    use std::fs;
    use std::io::Cursor;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};

    /// A fresh temp root per test (mirrors the mem construction
    /// tests' helper); Drop removes the tree idempotently.
    struct TempRoot(PathBuf);

    fn temp_root(tag: &'static str) -> TempRoot {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        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 dir = std::env::temp_dir().join(format!(
            "alkblobs-store-pg-{tag}-{}-{:x}-{seq}",
            std::process::id(),
            nanos
        ));
        fs::create_dir_all(&dir).expect("temp root created");
        TempRoot(dir)
    }

    impl TempRoot {
        fn path(&self) -> &Path {
            &self.0
        }
    }

    impl Drop for TempRoot {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.0);
        }
    }

    /// A fresh engine-owned schema per test — the postgres suite
    /// factory's fresh-schema-per-open convention — so the GC-schema
    /// assertions observe genuine creation.
    static SCHEMA_SEQ: AtomicU64 = AtomicU64::new(0);

    fn fresh_schema(tag: &str) -> String {
        format!(
            "abt_sctor_{}_{}_{}",
            tag,
            std::process::id(),
            SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed)
        )
    }

    fn env_nonempty(name: &str) -> Option<String> {
        std::env::var(name).ok().filter(|v| !v.is_empty())
    }

    /// The harness connection settings ride the environment
    /// (`ALKBLOBS_PG_HOST` required; port/user/password/db default to
    /// the harness server's values). Absent → the caller skips.
    fn harness_dsn() -> Option<String> {
        let host = env_nonempty("ALKBLOBS_PG_HOST")?;
        Some(format!(
            "host={host} port={} user={} password={} dbname={}",
            env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
            env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
            env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
            env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
        ))
    }

    /// The skip note — printed exactly once per skipped test.
    fn skip_note() {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
    }

    /// A fresh-schema opts for the harness server — with the schema
    /// janitor that drops it at test end (success, early return, or
    /// unwind) — or `None` (skip note printed) server-less.
    fn fresh_opts(tag: &str) -> Option<(PgOpts, SchemaJanitor)> {
        let dsn = harness_dsn()?;
        let schema = fresh_schema(tag);
        Some((
            PgOpts {
                schema: schema.clone(),
                ..Default::default()
            },
            SchemaJanitor::new(&dsn, &schema),
        ))
    }

    fn key_for(content: &[u8]) -> Key {
        Key::derive(content, content.len() as u64)
    }

    /// The postgres kv engine behind the constructed store — the
    /// schema observations' instrument.
    fn postgres_engine(store: &Store) -> &PostgresBackend {
        match store.kv() {
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine,
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("these tests construct postgres stores"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("these tests construct postgres stores"),
        }
    }

    /// The engine-neutral GC-schema call ran at construction on the
    /// fleet host's applicability row: the pool table exists (the
    /// engine's own bootstrap) and *both* GC tables exist in the
    /// fresh schema (postgres's both-tables row, ADR-012 §5).
    async fn assert_gc_schema_ran(store: &Store) {
        let engine = postgres_engine(store);
        assert!(
            engine.table_exists("blobs").await.unwrap(),
            "the pool table exists at construction"
        );
        assert!(
            engine.table_exists("gc_window").await.unwrap(),
            "ensure_gc_schema ran at construction: the window table exists in the fresh schema"
        );
        assert!(
            engine.table_exists("gc_pins").await.unwrap(),
            "ensure_gc_schema ran at construction: the pins table exists (the fleet host's both-tables row)"
        );
    }

    /// The dual-tier construction: both tiers present, mode/threshold/
    /// window-delay introspection exact, boundary routing
    /// deterministic, and the GC schema ran (both GC tables in the
    /// fresh schema).
    #[tokio::test]
    async fn dual_tier_postgres_local_constructs_and_routes() {
        let Some((opts, _janitor)) = fresh_opts("dual-tier") else {
            skip_note();
            return;
        };
        let root = temp_root("dual-tier");
        let dsn = harness_dsn().expect("harness dsn re-read");
        let store = Store::dual_tier_postgres_local(
            &dsn,
            opts,
            root.path(),
            1024,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("dual-tier postgres+local constructs");
        assert_eq!(store.mode(), DualTier);
        assert_eq!(store.threshold(), Some(1024));
        assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
        assert!(store.large().is_some(), "both tiers present");

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "length {len} routes deterministically"
            );
        }

        assert_gc_schema_ran(&store).await;
    }

    /// The kv-only construction: no large tier, the threshold is the
    /// rejection point's carrier, and the rejection wiring is the put
    /// seam's unchanged — known-length over-threshold rejects
    /// immediately (zero residue), unknown-length at mid-stream
    /// overflow, and exactly-at-threshold puts land.
    #[tokio::test]
    async fn kv_only_postgres_constructs_and_rejects_over_threshold_puts() {
        let Some((opts, _janitor)) = fresh_opts("kv-only") else {
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let store = Store::kv_only_postgres(&dsn, opts, 64, DEFAULT_DELETE_WINDOW_DELAY)
            .await
            .expect("kv-only postgres constructs");
        assert_eq!(store.mode(), KvOnly);
        assert_eq!(store.threshold(), Some(64));
        assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
        assert!(store.large().is_none(), "kv-only has no large tier");
        assert_gc_schema_ran(&store).await;

        for len in [0u64, 63, 64, 65, u64::MAX] {
            assert_eq!(
                store.tier_for(len),
                Tier::Kv,
                "kv-only routes every length kv"
            );
        }

        let err = store
            .put(Some(65), Cursor::new(&b"never read"[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("kv-only mode rejects"),
            "the immediate rejection is the mode's: {err}"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "zero residue"
        );

        let streamed = [3u8; 100];
        let err = store
            .put(None, Cursor::new(&streamed[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string()
                .contains("kv-only mode rejects an over-threshold put"),
            "the overflow rejection is the mode's, not the stream's: {err}"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "zero residue"
        );

        let at_threshold = vec![9u8; 64];
        let (key, pin) = store
            .put(Some(64), Cursor::new(&at_threshold[..]))
            .await
            .expect("exactly-at-threshold known put lands");
        assert_eq!(key, key_for(&at_threshold));
        assert!(store.has(&key.as_bytes()).await.unwrap());
        drop(pin);
    }

    /// The window-delay parameter: the shipped default is the pinned
    /// 5-minute constant, the constructor's value passes through the
    /// store's introspection, and `Duration::ZERO` is a legal
    /// degenerate tuning (execute-on-next-sweep).
    #[tokio::test]
    async fn window_delay_default_is_pinned_and_constructor_tunable() {
        assert_eq!(
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::from_secs(5 * 60),
            "the shipped default is 5 minutes (ADR-008 §2)"
        );

        let Some((opts, _janitor)) = fresh_opts("delay-default") else {
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let root = temp_root("delay-default");
        let store = Store::dual_tier_postgres_local(
            &dsn,
            opts,
            root.path(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("constructs at the shipped default");
        assert_eq!(store.window_delay(), Duration::from_secs(300));

        let Some((opts, _janitor)) = fresh_opts("delay-tuned") else {
            return;
        };
        let store = Store::kv_only_postgres(&dsn, opts, 64, Duration::from_secs(90))
            .await
            .expect("constructs at a custom tuning");
        assert_eq!(store.window_delay(), Duration::from_secs(90));

        let Some((opts, _janitor)) = fresh_opts("delay-zero") else {
            return;
        };
        let root = temp_root("delay-zero");
        let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO)
            .await
            .expect("the degenerate zero tuning is legal");
        assert_eq!(store.window_delay(), Duration::ZERO);

        let Some((opts, _janitor)) = fresh_opts("delay-long") else {
            return;
        };
        Store::kv_only_postgres(&dsn, opts, 64, Duration::from_secs(6000))
            .await
            .expect("a long delay is a legal tuning too");
    }

    /// The fleet-capable-kv shape round-trips end-to-end: a
    /// sub-threshold put lands on the kv tier (postgres), an
    /// over-threshold put stages into the large tier (local), both
    /// get back byte-identical, and the pool list is the sorted union
    /// of both tiers.
    #[tokio::test]
    async fn dual_tier_postgres_shape_round_trips_end_to_end_through_both_tiers() {
        let Some((opts, _janitor)) = fresh_opts("round-trip") else {
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let root = temp_root("round-trip");
        let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO)
            .await
            .expect("dual-tier constructs");

        let small = b"a sub-threshold blob";
        let (key, pin) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("kv put publishes");
        assert_eq!(key, key_for(small));
        assert!(
            store.kv().has(&key.as_bytes()).await.unwrap(),
            "landed on the kv tier (postgres)"
        );
        assert!(
            !store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
            "no large-tier copy on the kv route"
        );
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("kv entry present");
        assert_eq!(len, small.len() as u64, "whole-blob length exact");
        assert_eq!(
            blob.read_at(0, len).await.expect("full window"),
            small,
            "kv-tier round-trip byte-identical"
        );
        drop(pin);

        let large = vec![7u8; 128];
        let (key, pin) = store
            .put(Some(large.len() as u64), Cursor::new(&large[..]))
            .await
            .expect("staged put publishes");
        assert_eq!(key, key_for(&large));
        assert!(
            store.large().unwrap().has(&key.as_bytes()).await.unwrap(),
            "landed on the large tier (local)"
        );
        assert!(
            !store.kv().has(&key.as_bytes()).await.unwrap(),
            "no kv-tier copy on the large route"
        );
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .expect("get succeeds")
            .expect("large entry present");
        assert_eq!(len, large.len() as u64);
        assert_eq!(
            blob.read_at(0, len).await.expect("full window"),
            large,
            "large-tier round-trip byte-identical"
        );
        drop(pin);

        let mut expected: Vec<Key> = [key_for(small), key_for(&large)].to_vec();
        expected.sort();
        assert_eq!(
            store.list().await.expect("list succeeds"),
            expected,
            "the pool list is the sorted union of both tiers"
        );
    }

    /// A set-backed liveness source (the construction tests' shared
    /// sweep instrument).
    #[derive(Default)]
    struct SetSource {
        live: parking_lot::Mutex<BTreeSet<Key>>,
    }

    impl LivenessSource for SetSource {
        fn live_keys(&self) -> Vec<Key> {
            self.live.lock().iter().cloned().collect()
        }

        fn contains(&self, key: &Key) -> bool {
            self.live.lock().contains(key)
        }
    }

    /// The sweep's form branch runs the **durable form** on the
    /// postgres host (the branch keyed off the kv-engine enum,
    /// ADR-012 §5's one-window-host rule): a zero-delay sweep stages
    /// this sweep's dead candidates and executes none of them (the
    /// inline form would delete in the same sweep), and the *next*
    /// sweep executes the overdue set through time-of-delete
    /// re-arbitration.
    #[tokio::test]
    async fn sweep_runs_the_durable_form_on_the_postgres_host() {
        let source = Arc::new(SetSource::default());

        let Some((opts, _janitor)) = fresh_opts("durable-sweep") else {
            drop(source);
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let root = temp_root("durable-sweep");
        let store = Store::dual_tier_postgres_local(&dsn, opts, root.path(), 64, Duration::ZERO)
            .await
            .expect("dual-tier constructs");
        let _ = store.register_liveness_source(LivenessRegistration::InProcess(source.clone()));

        let dead = key_for(b"a dead candidate under the durable form");
        store
            .put(
                Some(dead_content().len() as u64),
                Cursor::new(&dead_content()[..]),
            )
            .await
            .expect("the candidate's put publishes");
        assert_eq!(store.list().await.unwrap(), vec![dead.clone()]);

        let first = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(first.aborted, None);
        assert_eq!(
            first.candidates_staged, 1,
            "the dead candidate staged into the durable window"
        );
        assert_eq!(
            first.deleted, 0,
            "the durable form executes nothing in its own staging sweep"
        );
        assert!(
            store.has(&dead.as_bytes()).await.unwrap(),
            "the staged candidate still waits out its window"
        );

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.aborted, None);
        assert_eq!(
            second.deleted, 1,
            "the next sweep executed the overdue candidate"
        );
        assert_eq!(
            second.candidates_staged, 0,
            "the pool holds nothing dead after the execution"
        );
        assert!(
            !store.has(&dead.as_bytes()).await.unwrap(),
            "the expired candidate is gone"
        );
    }

    /// The content the durable-sweep test's candidate carries — the
    /// key derives over it, the entry is genuinely dead (nothing
    /// liveness-protects it; the registered source set stays empty).
    fn dead_content() -> Vec<u8> {
        b"a dead candidate under the durable form".to_vec()
    }
}

/// The wave-5 constructor wiring's tests (features `pg-lo` + `mem`):
/// the one-window-host rule's named test topology — the durable
/// window's large host. The mode constructs and introspects correctly,
/// the GC schema ran on the **pg-lo host** at construction (`gc_window`
/// in the pg-lo schema; `gc_pins` never), the window-delay default is
/// the pinned constant and is constructor-tunable (`0` legal), the
/// shape round-trips end-to-end through both tiers with LO placement
/// asserted (kv entries never in the pg database), and the sweep's
/// form branch runs the durable form on the pg-lo host. Env-driven
/// (`ALKBLOBS_PG_*`), skipping cleanly server-less.
#[cfg(all(test, feature = "pg-lo", feature = "mem"))]
mod mem_pglo_construction_tests {
    use super::*;
    use crate::backend::Backend;
    use crate::pg_common::test_support::DatabaseJanitor;
    use crate::pg_lo::PgLoBackend;
    use std::io::Cursor;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio_postgres::NoTls;

    static DB_SEQ: AtomicU64 = AtomicU64::new(0);

    fn env_nonempty(name: &str) -> Option<String> {
        std::env::var(name).ok().filter(|v| !v.is_empty())
    }

    /// The harness connection settings ride the environment
    /// (`ALKBLOBS_PG_HOST` required; port/user/password/db default to
    /// the harness server's values). Absent → the caller skips.
    fn harness_dsn() -> Option<String> {
        let host = env_nonempty("ALKBLOBS_PG_HOST")?;
        Some(format!(
            "host={host} port={} user={} password={} dbname={}",
            env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
            env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
            env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
            env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
        ))
    }

    /// The skip note — printed exactly once per skipped test.
    fn skip_note() {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
    }

    /// A fresh dedicated database per test (LOs are database-level —
    /// the C7 lesson): the test's constructor opens its engine inside
    /// it, and a `DatabaseJanitor` drop-guard (`WITH (FORCE)`) removes
    /// the whole database at test end — success, early return, or
    /// unwind — zero residue on the shared harness database.
    struct TestDatabase {
        /// The DSN into the dedicated database (the constructor's
        /// connection settings).
        dsn: String,
        _janitor: DatabaseJanitor,
    }

    fn admin_dsn(dsn: &str) -> String {
        let base = dsn.rsplit_once("dbname=").expect("dbname").0;
        format!("{base} dbname=postgres")
    }

    fn db_dsn(dsn: &str, db: &str) -> String {
        let base = dsn.rsplit_once("dbname=").expect("dbname").0;
        format!("{base} dbname={db}")
    }

    async fn fresh_db(tag: &str) -> Option<TestDatabase> {
        let dsn = harness_dsn()?;
        let db = format!(
            "abtmplo_{}_{}_{:x}",
            tag,
            std::process::id(),
            DB_SEQ.fetch_add(1, Ordering::Relaxed)
        );
        let admin = admin_dsn(&dsn);
        {
            let (client, conn) = tokio_postgres::connect(&admin, NoTls)
                .await
                .expect("admin connect");
            tokio::spawn(async move {
                let _ = conn.await;
            });
            client
                .execute(
                    &format!(
                        "CREATE DATABASE {}",
                        crate::pg_common::quote_identifier(&db)
                    ),
                    &[],
                )
                .await
                .expect("create test database");
        }
        Some(TestDatabase {
            dsn: db_dsn(&dsn, &db),
            _janitor: DatabaseJanitor::new(&admin, &db),
        })
    }

    fn pglo_engine(store: &Store) -> &PgLoBackend {
        match store.large() {
            Some(LargeEngine::PgLo(engine)) => engine,
            Some(LargeEngine::Local(_)) | None => {
                panic!("these tests construct pg-lo-tier stores")
            }
        }
    }

    fn key_for(content: &[u8]) -> Key {
        Key::derive(content, content.len() as u64)
    }

    /// The one-window-host rule's named test topology constructs and
    /// introspects: both tiers present, the dispatch threshold active,
    /// and `ensure_gc_schema` ran at construction on the **pg-lo host**
    /// (`gc_window` in the pg-lo schema; `gc_pins` never — the pg-lo
    /// row of ADR-012 §5's applicability table).
    #[tokio::test]
    async fn dual_tier_mem_pglo_constructs_and_validates() {
        let Some(site) = fresh_db("mem-pglo").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_mem_pglo(
            &site.dsn,
            PgOpts::default(),
            1024,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("mem+pglo constructs");

        assert_eq!(store.mode(), DualTier);
        assert_eq!(store.threshold(), Some(1024));
        assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
        assert!(store.large().is_some(), "both tiers present");

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "length {len} routes deterministically"
            );
        }

        let lo = pglo_engine(&store);
        assert!(
            lo.table_exists("lo_entries").await.unwrap(),
            "the pg-lo engine's companion table bootstrapped at construction"
        );
        assert!(
            lo.table_exists("gc_window").await.unwrap(),
            "ensure_gc_schema ran at construction on the pg-lo host: \
             gc_window in the pg-lo schema (the one-window-host rule's \
             large-host arm)"
        );
        assert!(
            !lo.table_exists("gc_pins").await.unwrap(),
            "gc_pins is never created by pg-lo (fleet pins live on the kv host)"
        );
    }

    /// The mem-pglo shape round-trips end-to-end: a sub-threshold
    /// put lands on the kv tier (mem) — never in the pg database —
    /// an over-threshold put stages into the pg-lo tier (the
    /// companion row + LO present), both get back byte-identical,
    /// and the pool list is the sorted union of both tiers.
    #[tokio::test]
    async fn dual_tier_mem_pglo_shape_round_trips_with_lo_placement() {
        let Some(site) = fresh_db("mem-pglo-rt").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_mem_pglo(&site.dsn, PgOpts::default(), 64, Duration::ZERO)
            .await
            .expect("mem+pglo constructs");

        let small = b"a sub-threshold blob";
        let (key, pin) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("kv put publishes");
        assert_eq!(key, key_for(small));
        assert!(
            store.kv().has(&key.as_bytes()).await.unwrap(),
            "landed on the kv tier (mem)"
        );
        let lo = pglo_engine(&store);
        assert!(
            !lo.has(&key.as_bytes()).await.unwrap(),
            "kv entries never land in the pg database"
        );
        drop(pin);

        let big = vec![7u8; 200];
        let (key, pin) = store
            .put(Some(200), Cursor::new(&big[..]))
            .await
            .expect("staged put publishes");
        assert_eq!(key, key_for(&big));
        assert!(
            lo.has(&key.as_bytes()).await.unwrap(),
            "landed on the pg-lo tier: the companion row is present"
        );
        assert_eq!(
            lo.size(&key.as_bytes()).await.unwrap(),
            Some(200),
            "the companion row's size is exact"
        );
        assert!(
            !store.kv().has(&key.as_bytes()).await.unwrap(),
            "no kv copy on the large route"
        );
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("entry present");
        assert_eq!(len, 200);
        assert_eq!(
            blob.read_at(0, len).await.unwrap(),
            big,
            "the LO's content round-trips byte-identical (large placement)"
        );
        drop(pin);

        let mut expected: Vec<Key> = [key_for(small), key_for(&big)].to_vec();
        expected.sort();
        assert_eq!(
            store.list().await.expect("list succeeds"),
            expected,
            "the pool list is the sorted union of both tiers"
        );
    }

    /// The window-delay parameter on the pg-lo host: the shipped
    /// default is the pinned 5-minute constant, the constructor's
    /// value passes through the store's introspection, and
    /// `Duration::ZERO` is a legal degenerate tuning (execute-on-
    /// next-sweep).
    #[tokio::test]
    async fn window_delay_default_is_pinned_and_constructor_tunable() {
        assert_eq!(
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::from_secs(5 * 60),
            "the shipped default is 5 minutes (ADR-008 §2)"
        );

        let Some(site) = fresh_db("mem-pglo-delay").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_mem_pglo(
            &site.dsn,
            PgOpts::default(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("constructs at the shipped default");
        assert_eq!(store.window_delay(), Duration::from_secs(300));

        let store =
            Store::dual_tier_mem_pglo(&site.dsn, PgOpts::default(), 64, Duration::from_secs(90))
                .await
                .expect("constructs at a custom tuning");
        assert_eq!(store.window_delay(), Duration::from_secs(90));

        let store = Store::dual_tier_mem_pglo(&site.dsn, PgOpts::default(), 64, Duration::ZERO)
            .await
            .expect("the degenerate zero tuning is legal");
        assert_eq!(store.window_delay(), Duration::ZERO);

        Store::dual_tier_mem_pglo(&site.dsn, PgOpts::default(), 64, Duration::from_secs(6000))
            .await
            .expect("a long delay is a legal tuning too");
    }

    /// A set-backed liveness source — the construction tests' shared
    /// sweep instrument.
    #[derive(Default)]
    struct SetSource {
        live: parking_lot::Mutex<BTreeSet<Key>>,
    }

    impl LivenessSource for SetSource {
        fn live_keys(&self) -> Vec<Key> {
            self.live.lock().iter().cloned().collect()
        }

        fn contains(&self, key: &Key) -> bool {
            self.live.lock().contains(key)
        }
    }

    /// The sweep's form branch runs the **durable form on the pg-lo
    /// host** (the branch keyed off the engine representation — the mem
    /// arm consulting the large engine, never a flag, ADR-012 §5): a
    /// zero-delay sweep stages this sweep's dead candidate into the
    /// pg-lo schema's `gc_window` and executes none of them (the
    /// inline form — mem kv, no large host — would delete in the same
    /// sweep, so this is the form-branch-is-large-host proof), and the
    /// *next* sweep executes the overdue set re-arbitrated at
    /// execution time.
    #[tokio::test]
    async fn sweep_runs_the_durable_form_on_the_pg_lo_host() {
        let Some(site) = fresh_db("mem-pglo-sweep").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_mem_pglo(&site.dsn, PgOpts::default(), 64, Duration::ZERO)
            .await
            .expect("mem+pglo constructs");
        let _ = store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(
            SetSource::default(),
        )));

        let content = b"a dead candidate under the pg-lo host's window";
        let (dead, _) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the candidate's put publishes");
        assert_eq!(store.list().await.unwrap(), vec![dead.clone()]);

        let first = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(first.aborted, None);
        assert_eq!(
            first.candidates_staged, 1,
            "the dead candidate staged into the pg-lo schema's gc_window"
        );
        assert_eq!(
            first.deleted, 0,
            "the durable form executes nothing in its own staging sweep \
             (the form-branch-is-large-host proof)"
        );
        assert!(
            store.has(&dead.as_bytes()).await.unwrap(),
            "the staged candidate still waits out its window"
        );

        let second = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(second.aborted, None);
        assert_eq!(
            second.candidates_staged, 0,
            "nothing newly dead after the execution"
        );
        assert_eq!(
            second.deleted, 1,
            "the next sweep executed the overdue set, re-arbitrated at \
             execution time"
        );
        assert!(!store.has(&dead.as_bytes()).await.unwrap());
    }
}

/// The wave-5 constructor wiring's tests (features `pg-lo` +
/// `postgres`): the **consolidation posture** (ADR-009) — the
/// constructor completes the fleet-valid engine set. The mode
/// constructs and introspects correctly, the GC schema ran on the
/// **kv host** (both GC tables in the kv schema; the pg-lo window
/// clauses dormant — no `gc_window` in the pg-lo schema), the
/// window-delay default is the pinned constant and is
/// constructor-tunable (`0` legal), and the consolidation shape
/// round-trips end-to-end through both tiers with LO placement
/// asserted. Env-driven (`ALKBLOBS_PG_*`), skipping cleanly
/// server-less.
#[cfg(all(test, feature = "pg-lo", feature = "postgres"))]
mod postgres_pglo_construction_tests {
    use super::*;
    use crate::pg_common::test_support::DatabaseJanitor;
    use std::io::Cursor;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio_postgres::NoTls;

    static DB_SEQ: AtomicU64 = AtomicU64::new(0);

    fn env_nonempty(name: &str) -> Option<String> {
        std::env::var(name).ok().filter(|v| !v.is_empty())
    }

    /// The harness connection settings ride the environment
    /// (`ALKBLOBS_PG_HOST` required; port/user/password/db default to
    /// the harness server's values). Absent → the caller skips.
    fn harness_dsn() -> Option<String> {
        let host = env_nonempty("ALKBLOBS_PG_HOST")?;
        Some(format!(
            "host={host} port={} user={} password={} dbname={}",
            env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
            env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
            env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
            env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
        ))
    }

    /// The skip note — printed exactly once per skipped test.
    fn skip_note() {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
    }

    /// A fresh dedicated database per test (LOs are database-level —
    /// the C7 lesson): the drop-guard removes the whole database —
    /// both engines' schemas with it — at test end in every exit path.
    struct TestDatabase {
        /// The DSN into the dedicated database (the constructor's
        /// connection settings for both engines).
        dsn: String,
        _janitor: DatabaseJanitor,
    }

    fn admin_dsn(dsn: &str) -> String {
        let base = dsn.rsplit_once("dbname=").expect("dbname").0;
        format!("{base} dbname=postgres")
    }

    fn db_dsn(dsn: &str, db: &str) -> String {
        let base = dsn.rsplit_once("dbname=").expect("dbname").0;
        format!("{base} dbname={db}")
    }

    async fn fresh_db(tag: &str) -> Option<TestDatabase> {
        let dsn = harness_dsn()?;
        let db = format!(
            "abtmpglo_{}_{}_{:x}",
            tag,
            std::process::id(),
            DB_SEQ.fetch_add(1, Ordering::Relaxed)
        );
        let admin = admin_dsn(&dsn);
        {
            let (client, conn) = tokio_postgres::connect(&admin, NoTls)
                .await
                .expect("admin connect");
            tokio::spawn(async move {
                let _ = conn.await;
            });
            client
                .execute(
                    &format!(
                        "CREATE DATABASE {}",
                        crate::pg_common::quote_identifier(&db)
                    ),
                    &[],
                )
                .await
                .expect("create test database");
        }
        Some(TestDatabase {
            dsn: db_dsn(&dsn, &db),
            _janitor: DatabaseJanitor::new(&admin, &db),
        })
    }

    /// A fresh engine-owned schema per test (the host-split test's
    /// instrument: two distinct schemas so the dormancy assertion
    /// observes the pg-lo schema genuinely without `gc_window`).
    fn fresh_schema(tag: &str, seq: u64) -> String {
        format!("abtmpglo_{tag}_{}_{seq:x}", std::process::id())
    }

    fn pglo_engine(store: &Store) -> &crate::pg_lo::PgLoBackend {
        match store.large() {
            Some(LargeEngine::PgLo(engine)) => engine,
            Some(LargeEngine::Local(_)) | None => {
                panic!("these tests construct pg-lo-tier stores")
            }
        }
    }

    fn postgres_engine(store: &Store) -> &PostgresBackend {
        match store.kv() {
            #[cfg(feature = "postgres")]
            KvEngine::Postgres(engine) => engine,
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("these tests construct postgres stores"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("these tests construct postgres stores"),
        }
    }

    fn key_for(content: &[u8]) -> Key {
        Key::derive(content, content.len() as u64)
    }

    /// The consolidation constructor constructs and introspects; the
    /// GC schema ran on the **kv host** (both GC tables in the kv
    /// schema) and the pg-lo window clauses are **dormant** (no
    /// `gc_window` in the pg-lo schema — two distinct schemas here so
    /// the dormancy assertion observes genuine absence, ADR-012 §5's
    /// dual-SQL shape).
    #[tokio::test]
    async fn dual_tier_postgres_pglo_constructs_with_the_kv_host_winning_the_window() {
        let Some(site) = fresh_db("consolidation").await else {
            skip_note();
            return;
        };
        let seq = DB_SEQ.fetch_add(1, Ordering::Relaxed);
        let kv_schema = fresh_schema("kv", seq);
        let lo_schema = fresh_schema("lo", seq);
        let store = Store::dual_tier_postgres_pglo(
            &site.dsn,
            PgOpts {
                schema: kv_schema,
                ..Default::default()
            },
            &site.dsn,
            PgOpts {
                schema: lo_schema,
                ..Default::default()
            },
            1024,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("consolidation constructs");

        assert_eq!(store.mode(), DualTier);
        assert_eq!(store.threshold(), Some(1024));
        assert_eq!(store.window_delay(), DEFAULT_DELETE_WINDOW_DELAY);
        assert!(store.large().is_some(), "both tiers present");

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "length {len} routes deterministically"
            );
        }

        let kv = postgres_engine(&store);
        assert!(
            kv.table_exists("blobs").await.unwrap(),
            "the pool table exists at construction"
        );
        assert!(
            kv.table_exists("gc_window").await.unwrap(),
            "the kv host's call ran at construction: gc_window in the kv schema"
        );
        assert!(
            kv.table_exists("gc_pins").await.unwrap(),
            "the fleet host's both-tables row: gc_pins in the kv schema"
        );

        let lo = pglo_engine(&store);
        assert!(
            lo.table_exists("lo_entries").await.unwrap(),
            "the pg-lo engine's companion table bootstrapped at construction"
        );
        assert!(
            !lo.table_exists("gc_window").await.unwrap(),
            "the pg-lo window clauses are dormant on the dual-SQL instance \
             (ADR-012 §5): the pg-lo schema carries no gc_window"
        );
        assert!(
            !lo.table_exists("gc_pins").await.unwrap(),
            "gc_pins is never created by pg-lo (fleet pins live on the kv host)"
        );
    }

    /// The consolidation shape round-trips end-to-end in the recorded
    /// corollary form — the same DSN and opts to both engines (one
    /// minted schema holding `blobs`, `lo_entries`, and the GC tables
    ///; ADR-012 §4's co-tenancy): a sub-threshold put lands on the kv
    /// tier and never in the pg-lo companion table; an over-threshold
    /// put stages into the pg-lo tier (the companion row + LO
    /// present); both get back byte-identical; the pool list is the
    /// sorted union.
    #[tokio::test]
    async fn consolidation_shape_round_trips_end_to_end_through_both_tiers() {
        let Some(site) = fresh_db("consolidation-rt").await else {
            skip_note();
            return;
        };
        let opts = PgOpts::default();
        let store = Store::dual_tier_postgres_pglo(
            &site.dsn,
            opts.clone(),
            &site.dsn,
            opts,
            64,
            Duration::ZERO,
        )
        .await
        .expect("consolidation constructs");

        let small = b"a sub-threshold blob";
        let (key, pin) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("kv put publishes");
        assert_eq!(key, key_for(small));
        assert!(
            store.kv().has(&key.as_bytes()).await.unwrap(),
            "landed on the kv tier (postgres)"
        );
        let lo = pglo_engine(&store);
        assert!(
            !lo.has(&key.as_bytes()).await.unwrap(),
            "the sub-threshold key never entered the pg-lo companion table"
        );
        drop(pin);

        let big = vec![42u8; 200];
        let (key, pin) = store
            .put(Some(200), Cursor::new(&big[..]))
            .await
            .expect("staged put publishes");
        assert_eq!(key, key_for(&big));
        assert!(
            lo.has(&key.as_bytes()).await.unwrap(),
            "landed on the pg-lo tier: the companion row is present"
        );
        assert_eq!(
            lo.size(&key.as_bytes()).await.unwrap(),
            Some(200),
            "the companion row's size is exact"
        );
        assert!(
            !store.kv().has(&key.as_bytes()).await.unwrap(),
            "no kv copy on the large route"
        );
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("entry present");
        assert_eq!(len, 200);
        assert_eq!(
            blob.read_at(0, len).await.unwrap(),
            big,
            "the LO's content round-trips byte-identical (LO placement)"
        );
        drop(pin);

        let mut expected: Vec<Key> = [key_for(small), key_for(&big)].to_vec();
        expected.sort();
        assert_eq!(
            store.list().await.expect("list succeeds"),
            expected,
            "the pool list is the sorted union of both tiers"
        );
    }

    /// The window-delay parameter on the consolidation constructor:
    /// the shipped default is the pinned 5-minute constant, tunable,
    /// `0` legal.
    #[tokio::test]
    async fn window_delay_default_is_pinned_and_constructor_tunable() {
        assert_eq!(
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::from_secs(5 * 60),
            "the shipped default is 5 minutes (ADR-008 §2)"
        );

        let Some(site) = fresh_db("consolidation-delay").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_postgres_pglo(
            &site.dsn,
            PgOpts::default(),
            &site.dsn,
            PgOpts::default(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("constructs at the shipped default");
        assert_eq!(store.window_delay(), Duration::from_secs(300));

        let store = Store::dual_tier_postgres_pglo(
            &site.dsn,
            PgOpts::default(),
            &site.dsn,
            PgOpts::default(),
            64,
            Duration::from_secs(90),
        )
        .await
        .expect("constructs at a custom tuning");
        assert_eq!(store.window_delay(), Duration::from_secs(90));

        let store = Store::dual_tier_postgres_pglo(
            &site.dsn,
            PgOpts::default(),
            &site.dsn,
            PgOpts::default(),
            64,
            Duration::ZERO,
        )
        .await
        .expect("the degenerate zero tuning is legal");
        assert_eq!(store.window_delay(), Duration::ZERO);

        Store::dual_tier_postgres_pglo(
            &site.dsn,
            PgOpts::default(),
            &site.dsn,
            PgOpts::default(),
            64,
            Duration::from_secs(6000),
        )
        .await
        .expect("a long delay is a legal tuning too");
    }
}

/// The composite fleet-validity predicate's per-clause tests (the
/// predicate is the fleet constructors' admission gate; the mem and
/// sqlite clauses have no fleet constructor to reach them through —
/// the recorded "not fleet-anything" postures — so the rejection arms
/// are exercised at the predicate level, the same error the would-be
/// construction carries). Runs in every build: pure vocabulary, no
/// engine I/O.
#[cfg(test)]
mod fleet_validity_predicate_tests {
    use super::*;

    /// Every fleet-valid shape admits: kv = postgres and (large = none
    /// — the re-routing arm, OR `local` on the declared shared media,
    /// OR pg-lo — the same-instance clause's runtime probe rides its
    /// constructor). (backends-and-dispatch.md, "The composite
    /// fleet-validity rule", one sentence, both tiers.)
    #[test]
    fn every_fleet_valid_shape_admits() {
        assert_eq!(
            fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::None),
            None,
            "the re-routing posture (large = none) is fleet-valid with kv = postgres"
        );
        assert_eq!(
            fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::LocalSharedMedia),
            None,
            "the declared shared-media arm is fleet-valid"
        );
        assert_eq!(
            fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::PgLo),
            None,
            "the pg-lo arm is fleet-valid (its same-instance clause probes at the constructor)"
        );
    }

    /// `mem` anywhere rejects (ADR-012 §3): mem is never fleet-valid
    /// (ADR-011 §4) — every large shape over a mem kv engine is the
    /// constructor error.
    #[test]
    fn mem_anywhere_rejects() {
        for large in [
            FleetLargeShape::None,
            FleetLargeShape::LocalSharedMedia,
            FleetLargeShape::LocalUndeclared,
            FleetLargeShape::PgLo,
        ] {
            let err = fleet_validity_error(FleetKvShape::Mem, large)
                .expect("mem is never fleet-valid, any large shape");
            assert!(
                err.to_string().contains("Mem is never fleet-valid"),
                "mem is the kv engine, large = {large:?}: {err}"
            );
        }
    }

    /// The sqlite kv engine rejects (ADR-012 §3's "any kv engine other
    /// than postgres"): a sqlite fleet is a recorded non-option (ADR-010
    /// §3).
    #[test]
    fn sqlite_rejects() {
        for large in [
            FleetLargeShape::None,
            FleetLargeShape::LocalSharedMedia,
            FleetLargeShape::LocalUndeclared,
            FleetLargeShape::PgLo,
        ] {
            assert!(
                fleet_validity_error(FleetKvShape::Sqlite, large).is_some(),
                "the sqlite kv engine, large = {large:?}, rejects"
            );
        }
    }

    /// `local` without the shared-media declaration rejects (ADR-008
    /// §3): a fleet-mode constructor requires the explicit
    /// [`FleetSharedMedia`] assertion — the sealed declaration type
    /// keeps every real call site on the declared arm, and this is the
    /// rule's rejected arm.
    #[test]
    fn local_without_the_shared_media_declaration_rejects() {
        let err = fleet_validity_error(FleetKvShape::Postgres, FleetLargeShape::LocalUndeclared)
            .expect("the undeclared arm rejects");
        assert!(
            err.to_string().contains("shared-media declaration"),
            "the error names the missing declaration: {err}"
        );
    }
}

/// The wave-6 fleet declaration's constructor tests (feature
/// `postgres`): the `_fleet` constructors' fleet state introspection
/// (flag, TTL, owner id), the TTL's engine wiring, the admission gate
/// (a zero TTL errors closed at entry), and the declaration's behavioral
/// nothing-else posture — put/sweep on a fleet store run exactly the
/// non-fleet store's machinery (the fleet mechanism turns on with the
/// later put/sweep driver tasks, not with the declaration).
/// Env-driven (`ALKBLOBS_PG_*`), skipping cleanly server-less.
#[cfg(all(test, feature = "postgres"))]
mod postgres_fleet_construction_tests {
    use super::*;
    use crate::postgres::test_support::SchemaJanitor;
    use std::fs;
    use std::io::Cursor;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU64, Ordering};

    /// A fresh temp root per test; Drop removes the tree idempotently.
    struct TempRoot(PathBuf);

    fn temp_root(tag: &'static str) -> TempRoot {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
        let dir = std::env::temp_dir().join(format!(
            "alkblobs-store-fleet-{tag}-{}-{:x}-{seq}",
            std::process::id(),
            seq,
        ));
        fs::create_dir_all(&dir).expect("temp root created");
        TempRoot(dir)
    }

    impl TempRoot {
        fn path(&self) -> &Path {
            &self.0
        }
    }

    impl Drop for TempRoot {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.0);
        }
    }

    static SCHEMA_SEQ: AtomicU64 = AtomicU64::new(0);

    fn fresh_schema(tag: &str) -> String {
        format!(
            "abtfleet_{}_{}_{:x}",
            tag,
            std::process::id(),
            SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed)
        )
    }

    fn env_nonempty(name: &str) -> Option<String> {
        std::env::var(name).ok().filter(|v| !v.is_empty())
    }

    fn harness_dsn() -> Option<String> {
        let host = env_nonempty("ALKBLOBS_PG_HOST")?;
        Some(format!(
            "host={host} port={} user={} password={} dbname={}",
            env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
            env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
            env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
            env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
        ))
    }

    fn skip_note() {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
    }

    fn fresh_opts(tag: &str) -> Option<(PgOpts, SchemaJanitor)> {
        let dsn = harness_dsn()?;
        let schema = fresh_schema(tag);
        Some((
            PgOpts {
                schema: schema.clone(),
                ..Default::default()
            },
            SchemaJanitor::new(&dsn, &schema),
        ))
    }

    fn key_for(content: &[u8]) -> Key {
        Key::derive(content, content.len() as u64)
    }

    fn postgres_engine(store: &Store) -> &PostgresBackend {
        match store.kv() {
            KvEngine::Postgres(engine) => engine,
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("these tests construct postgres stores"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("these tests construct postgres stores"),
        }
    }

    async fn assert_gc_schema_ran(store: &Store) {
        let engine = postgres_engine(store);
        assert!(
            engine.table_exists("blobs").await.unwrap(),
            "the pool table exists at construction"
        );
        assert!(
            engine.table_exists("gc_window").await.unwrap(),
            "ensure_gc_schema ran at construction"
        );
        assert!(
            engine.table_exists("gc_pins").await.unwrap(),
            "the fleet host's both-tables row: gc_pins exists"
        );
    }

    /// The shipped default pin TTL is the pinned 60-second constant
    /// (ADR-008 §2) — the fleet constructors' default tuning, the
    /// constant the engine's `pin_renew` successor-time body renews
    /// against.
    #[test]
    fn default_pin_ttl_is_the_pinned_sixty_seconds() {
        assert_eq!(
            DEFAULT_PIN_TTL,
            Duration::from_secs(crate::postgres::DEFAULT_PIN_TTL_SECS),
            "the store's shipped default is the engine constant's seconds"
        );
        assert_eq!(
            DEFAULT_PIN_TTL,
            Duration::from_secs(60),
            "the shipped default is 60 s (ADR-008 §2)"
        );
    }

    /// The store-instance id's minting: 32 raw bytes, unique per
    /// instance (the process-global counter distinguishes instances in
    /// one process; time + pid + the ASLR anchor distinguish
    /// processes) — minted at construction, carried as the pin rows'
    /// `owner` half, crate-internal.
    #[test]
    fn owner_id_mints_unique_and_fixed_width() {
        let first = mint_owner_id();
        let second = mint_owner_id();
        assert_eq!(first.len(), 32, "the SHA-256 fold's width");
        assert_eq!(second.len(), 32);
        assert_ne!(
            first, second,
            "two instances in one process mint distinct ids"
        );
    }

    /// A zero pin TTL errors closed at entry — before any I/O — never
    /// a silent pass (the TTL/3 renewal cadence degenerates at zero
    /// and a zero-TTL pin expires at once).
    #[tokio::test]
    async fn zero_pin_ttl_is_a_constructor_error_at_entry() {
        let err = Store::kv_only_postgres_fleet(
            "host=no-such-host.invalid port=1 user=x password=x dbname=x",
            PgOpts::default(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::ZERO,
        )
        .await
        .err()
        .expect("the entry validation rejects before any connection attempt");
        assert!(
            err.to_string().contains("pin_ttl must be positive"),
            "the zero TTL rejects before any connection attempt: {err}"
        );

        let err = Store::dual_tier_postgres_local_fleet(
            "host=no-such-host.invalid port=1 user=x password=x dbname=x",
            PgOpts::default(),
            std::env::temp_dir().join("alkblobs-fleet-zero-ttl"),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::ZERO,
            FleetSharedMedia::asserted(),
        )
        .await
        .err()
        .expect("the entry validation rejects on the shared-media shape too");
        assert!(
            err.to_string().contains("pin_ttl must be positive"),
            "the zero TTL rejects at entry on the shared-media shape too: {err}"
        );
    }

    /// The fleet re-routing client posture constructs and introspects:
    /// the fleet state carries the declaration's flag, the pin TTL
    /// (default and constructor-tuned), and the store-instance id —
    /// distinct across two instances — and the GC schema ran (the
    /// fleet host's both-tables row). The over-threshold put-time
    /// rejection (ADR-012 §6.2) is unchanged under fleet.
    #[tokio::test]
    async fn kv_only_postgres_fleet_constructs_and_introspects_fleet_state() {
        let Some((opts, _janitor)) = fresh_opts("fleet-kv-only") else {
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let store = Store::kv_only_postgres_fleet(
            &dsn,
            opts,
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
        )
        .await
        .expect("the fleet kv-only posture constructs");
        assert_eq!(store.mode(), KvOnly);
        assert!(store.large().is_none(), "no large tier (large = none)");
        let fleet = store
            .fleet()
            .expect("fleet: true declares: the state carries");
        assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL);
        assert_eq!(fleet.owner().len(), 32);
        assert_gc_schema_ran(&store).await;

        for (len, tier) in [
            (0u64, Tier::Kv),
            (63, Tier::Kv),
            (64, Tier::Kv),
            (65, Tier::Kv),
            (u64::MAX, Tier::Kv),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "kv-only routes every length kv, fleet or not"
            );
        }

        let err = store
            .put(Some(65), Cursor::new(&b"never read"[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("kv-only mode rejects"),
            "the put-time rejection is unchanged under the declaration: {err}"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "zero residue"
        );
        drop(store);

        let Some((opts, _janitor)) = fresh_opts("fleet-kv-ttl") else {
            skip_note();
            return;
        };
        let store = Store::kv_only_postgres_fleet(
            &dsn,
            opts,
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::from_secs(90),
        )
        .await
        .expect("constructs at a custom pin TTL");
        let fleet = store.fleet().expect("fleet state carries");
        assert_eq!(
            fleet.pin_ttl(),
            Duration::from_secs(90),
            "the TTL is a fleet constructor parameter"
        );
        let same_instance_owner = fleet.owner().to_vec();
        drop(store);

        let Some((opts, _janitor)) = fresh_opts("fleet-kv-owner2") else {
            skip_note();
            return;
        };
        let other = Store::kv_only_postgres_fleet(
            &dsn,
            opts,
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
        )
        .await
        .expect("a second instance constructs");
        assert_ne!(
            other.fleet().expect("fleet state carries").owner(),
            same_instance_owner,
            "two instances never share the store-instance id"
        );
    }

    /// The fleet declaration's scope after the sweep driver landed:
    /// the kv put routes through the joint entry+pin tx (the pin row
    /// lands with the entry — asserted against the pin check), and the
    /// sweep runs the **fleet body** on the kv host — the form branch
    /// keys off the declaration, the durable window stages there, and
    /// the instance's own un-expired pin row protects the candidate at
    /// the execution-time SQL re-arbitration (guard dropped, the row
    /// still lives its TTL) until it expires and the maintenance step
    /// reaps it — then the candidate executes.
    #[tokio::test]
    async fn fleet_declaration_runs_the_sweep_machinery_on_the_kv_host() {
        struct SetSource;
        impl LivenessSource for SetSource {
            fn live_keys(&self) -> Vec<Key> {
                Vec::new()
            }
            fn contains(&self, _key: &Key) -> bool {
                false
            }
        }

        let Some((opts, _janitor)) = fresh_opts("fleet-inert") else {
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let store =
            Store::kv_only_postgres_fleet(&dsn, opts, 64, Duration::ZERO, Duration::from_secs(2))
                .await
                .expect("constructs");
        let _ =
            store.register_liveness_source(LivenessRegistration::InProcess(Arc::new(SetSource)));

        let content = b"a dead candidate on a declared fleet store";
        let (key, pin) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the fleet put publishes through the joint tx");
        assert_eq!(key, key_for(content));
        assert!(store.has(&key.as_bytes()).await.unwrap());
        drop(pin);

        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(staged.aborted, None);
        assert_eq!(
            staged.candidates_staged, 1,
            "the fleet body staged on the kv host (the one-window-host rule's fleet reading)"
        );
        assert_eq!(staged.deleted, 0);

        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            (executed.deleted, executed.cancelled_by_arbitration),
            (0, 1),
            "the instance's own un-expired pin row protects at the execution-time SQL \
             re-arbitration (guard dropped, the row lives its TTL)"
        );
        assert!(store.has(&key.as_bytes()).await.unwrap());

        tokio::time::sleep(Duration::from_secs(3)).await;
        let reaped = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(
            reaped.deleted, 1,
            "the expired row reaped by the maintenance step, the re-arbitration passes"
        );
        assert!(!store.has(&key.as_bytes()).await.unwrap());
    }

    /// The fleet shared-media posture constructs and introspects; the
    /// GC schema ran on the kv host (the one-window-host rule intact
    /// under fleet — kv = postgres hosts the window; there is no
    /// second SQL engine to host anything else).
    #[tokio::test]
    async fn dual_tier_postgres_local_fleet_constructs_and_introspects() {
        let Some((opts, _janitor)) = fresh_opts("fleet-shared-media") else {
            skip_note();
            return;
        };
        let dsn = harness_dsn().expect("harness dsn re-read");
        let root = temp_root("fleet-shared-media");
        let store = Store::dual_tier_postgres_local_fleet(
            &dsn,
            opts,
            root.path(),
            1024,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
            FleetSharedMedia::asserted(),
        )
        .await
        .expect("the fleet shared-media posture constructs");
        assert_eq!(store.mode(), DualTier);
        assert!(store.large().is_some(), "both tiers present");
        let fleet = store
            .fleet()
            .expect("fleet: true declares: the state carries");
        assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL);
        assert_eq!(fleet.owner().len(), 32);
        assert_gc_schema_ran(&store).await;

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "the declaration does not disturb the dispatch"
            );
        }
    }
}

/// The wave-6 fleet declaration's pg-lo-side constructor tests
/// (features `postgres` + `pg-lo`): the consolidation posture's fleet
/// twin — the pg-lo same-instance probe (ADR-012 §4) green on the
/// same instance (including the instance-level shape: two distinct
/// *databases* on one instance are one probe-matching instance), the
/// probe's error-closed negative (a DSN that cannot resolve to the kv
/// engine's instance is a constructor error, never a silent pass), and
/// the one-window-host posture (the window hosts on the kv engine; the
/// pg-lo window clauses stay dormant). Env-driven (`ALKBLOBS_PG_*`),
/// skipping cleanly server-less.
#[cfg(all(test, feature = "pg-lo", feature = "postgres"))]
mod postgres_pglo_fleet_construction_tests {
    use super::*;
    use crate::pg_common::test_support::DatabaseJanitor;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio_postgres::NoTls;

    static DB_SEQ: AtomicU64 = AtomicU64::new(0);

    fn env_nonempty(name: &str) -> Option<String> {
        std::env::var(name).ok().filter(|v| !v.is_empty())
    }

    fn harness_dsn() -> Option<String> {
        let host = env_nonempty("ALKBLOBS_PG_HOST")?;
        Some(format!(
            "host={host} port={} user={} password={} dbname={}",
            env_nonempty("ALKBLOBS_PG_PORT").unwrap_or_else(|| "15432".to_owned()),
            env_nonempty("ALKBLOBS_PG_USER").unwrap_or_else(|| "postgres".to_owned()),
            env_nonempty("ALKBLOBS_PG_PASSWORD").unwrap_or_else(|| "poc".to_owned()),
            env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
        ))
    }

    fn skip_note() {
        eprintln!(
            "skipping (no postgres harness server): set ALKBLOBS_PG_HOST/PORT/USER/PASSWORD/DB"
        );
    }

    /// A fresh dedicated database per test (LOs are database-level —
    /// the C7 lesson); the drop-guard removes it in every exit path.
    struct TestDatabase {
        dsn: String,
        _janitor: DatabaseJanitor,
    }

    fn admin_dsn(dsn: &str) -> String {
        let base = dsn.rsplit_once("dbname=").expect("dbname").0;
        format!("{base} dbname=postgres")
    }

    fn db_dsn(dsn: &str, db: &str) -> String {
        let base = dsn.rsplit_once("dbname=").expect("dbname").0;
        format!("{base} dbname={db}")
    }

    async fn fresh_db(tag: &str) -> Option<TestDatabase> {
        let dsn = harness_dsn()?;
        let db = format!(
            "abtfleetlo_{}_{}_{:x}",
            tag,
            std::process::id(),
            DB_SEQ.fetch_add(1, Ordering::Relaxed)
        );
        let admin = admin_dsn(&dsn);
        {
            let (client, conn) = tokio_postgres::connect(&admin, NoTls)
                .await
                .expect("admin connect");
            tokio::spawn(async move {
                let _ = conn.await;
            });
            client
                .execute(
                    &format!(
                        "CREATE DATABASE {}",
                        crate::pg_common::quote_identifier(&db)
                    ),
                    &[],
                )
                .await
                .expect("create test database");
        }
        Some(TestDatabase {
            dsn: db_dsn(&dsn, &db),
            _janitor: DatabaseJanitor::new(&admin, &db),
        })
    }

    fn fresh_schema(tag: &str) -> String {
        format!(
            "abtfleetlo_{}_{}_{:x}",
            tag,
            std::process::id(),
            DB_SEQ.fetch_add(1, Ordering::Relaxed)
        )
    }

    fn pglo_engine(store: &Store) -> &crate::pg_lo::PgLoBackend {
        match store.large() {
            Some(LargeEngine::PgLo(engine)) => engine,
            Some(LargeEngine::Local(_)) | None => panic!("these tests construct pg-lo-tier stores"),
        }
    }

    fn postgres_engine(store: &Store) -> &PostgresBackend {
        match store.kv() {
            KvEngine::Postgres(engine) => engine,
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("these tests construct postgres stores"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("these tests construct postgres stores"),
        }
    }

    /// The consolidation posture's fleet twin constructs on the same
    /// pg instance: the fleet state carries (flag, TTL, owner id), the
    /// kv host's GC schema ran (the fleet host's both-tables row), and
    /// the one-window-host posture composes — the pg-lo window clauses
    /// stay dormant (no `gc_window` in the pg-lo schema; two distinct
    /// schemas so the dormancy assertion observes genuine absence),
    /// `gc_pins` never in the pg-lo schema.
    #[tokio::test]
    async fn dual_tier_postgres_pglo_fleet_constructs_same_instance() {
        let Some(site) = fresh_db("fleet-consolidation").await else {
            skip_note();
            return;
        };
        let kv_schema = fresh_schema("kv");
        let lo_schema = fresh_schema("lo");
        let store = Store::dual_tier_postgres_pglo_fleet(
            &site.dsn,
            PgOpts {
                schema: kv_schema.clone(),
                ..Default::default()
            },
            &site.dsn,
            PgOpts {
                schema: lo_schema.clone(),
                ..Default::default()
            },
            1024,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
        )
        .await
        .expect("the same-instance consolidation posture constructs");

        assert_eq!(store.mode(), DualTier);
        assert!(store.large().is_some());
        let fleet = store
            .fleet()
            .expect("fleet: true declares: the state carries");
        assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL);
        assert_eq!(fleet.owner().len(), 32);

        for (len, tier) in [
            (0u64, Tier::Kv),
            (1023, Tier::Kv),
            (1024, Tier::Kv),
            (1025, Tier::Large),
            (u64::MAX, Tier::Large),
        ] {
            assert_eq!(
                store.tier_for(len),
                tier,
                "the declaration does not disturb the dispatch"
            );
        }

        let kv = postgres_engine(&store);
        assert!(kv.table_exists("blobs").await.unwrap());
        assert!(
            kv.table_exists("gc_window").await.unwrap(),
            "the window hosts on the kv engine (the one-window-host rule under fleet)"
        );
        assert!(
            kv.table_exists("gc_pins").await.unwrap(),
            "the fleet host's both-tables row: gc_pins in the kv schema"
        );

        let lo = pglo_engine(&store);
        assert!(lo.table_exists("lo_entries").await.unwrap());
        assert!(
            !lo.table_exists("gc_window").await.unwrap(),
            "the pg-lo window clauses stay dormant on the dual-SQL instance"
        );
        assert!(
            !lo.table_exists("gc_pins").await.unwrap(),
            "gc_pins is never created by pg-lo (fleet pins live on the kv host)"
        );
        drop(store);
        let _ = (kv_schema, lo_schema);
    }

    /// The probe is instance-level, not database-level: two distinct
    /// *databases* on one pg instance carry one `system_identifier`
    /// (the data directory's), so the fleet twin accepts the kv engine
    /// and the pg-lo engine living in different databases of the same
    /// instance — the textual-DSN-comparison alternative could not
    /// have answered this shape.
    #[tokio::test]
    async fn dual_tier_postgres_pglo_fleet_two_databases_one_instance() {
        let Some(kv_site) = fresh_db("fleet-two-db-kv").await else {
            skip_note();
            return;
        };
        let Some(lo_site) = fresh_db("fleet-two-db-lo").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_postgres_pglo_fleet(
            &kv_site.dsn,
            PgOpts::default(),
            &lo_site.dsn,
            PgOpts::default(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
        )
        .await
        .expect("two databases on one instance are one fleet-valid instance");

        let fleet = store
            .fleet()
            .expect("fleet: true declares: the state carries");
        assert_eq!(fleet.pin_ttl(), DEFAULT_PIN_TTL);
    }

    /// The probe errors closed: a large-side DSN that cannot resolve
    /// to the kv engine's instance (a database that does not exist on
    /// the dev harness — the recorded negative, one instance only) is
    /// a constructor error, never a silent pass. The failing point
    /// rides the engine open or the probe — either way the
    /// constructor rejects.
    #[tokio::test]
    async fn dual_tier_postgres_pglo_fleet_unresolvable_instance_errors_closed() {
        let Some(kv_site) = fresh_db("fleet-closed-kv").await else {
            skip_note();
            return;
        };
        let nonexistent = db_dsn(&kv_site.dsn, "abtnosuchdb_fleet_negative");
        let err = Store::dual_tier_postgres_pglo_fleet(
            &kv_site.dsn,
            PgOpts::default(),
            &nonexistent,
            PgOpts::default(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
        )
        .await
        .err()
        .expect("the cross-instance (unresolvable) negative errors closed");
        assert!(
            err.to_string().contains("postgres"),
            "the rejection is a pg-side failure surfaced as the constructor error: {err}"
        );
        assert!(
            !matches!(
                err,
                Error::Missing | Error::Verification | Error::KeyInvalid | Error::GcRefuse
            ),
            "the failure never masquerades as a non-Io member"
        );
    }

    /// A zero pin TTL errors closed at entry — before any I/O. The
    /// entry check runs server-less.
    #[tokio::test]
    async fn zero_pin_ttl_is_a_constructor_error_at_entry() {
        let err = Store::dual_tier_postgres_pglo_fleet(
            "host=no-such-host.invalid port=1 user=x password=x dbname=x",
            PgOpts::default(),
            "host=no-such-host.invalid port=1 user=x password=x dbname=x",
            PgOpts::default(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            Duration::ZERO,
        )
        .await
        .err()
        .expect("the entry validation rejects before any connection attempt");
        assert!(
            err.to_string().contains("pin_ttl must be positive"),
            "the zero TTL rejects at entry on the consolidation shape too: {err}"
        );
    }
}