alkblobs 0.1.0

//! The fleet machinery: [`FleetState`] (the constructor's `fleet:
//! true` declaration's carried state), the store-instance id minting,
//! and the fleet put path's store-core drivers — the joint entry+pin
//! tx (A1's catch-propagate-terminate), the pin guard's TTL/3 renewal
//! duty, and the per-shape drivers (kv joint form, pg-lo one-tx form,
//! local shared-media pin-first form). The two fleet-put test
//! sections ride this file.

use super::engines::AnyStaging;
#[cfg(feature = "postgres")]
use super::engines::KvEngine;
#[cfg(all(test, feature = "postgres"))]
use super::engines::LargeEngine;
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use super::put::PUT_CHUNK;
use super::Store;
#[cfg(all(test, feature = "postgres"))]
use super::{FleetSharedMedia, DEFAULT_DELETE_WINDOW_DELAY, DEFAULT_PIN_TTL};
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use crate::backend::ReadCursor;
#[cfg(feature = "postgres")]
use crate::backend::StagingCursor;
#[cfg(all(test, feature = "postgres"))]
use crate::facade::LivenessSource;
use crate::facade::Pin;
#[cfg(feature = "postgres")]
use crate::facade::PinRenewal;
#[cfg(feature = "postgres")]
use crate::gc::{EngineFleetState, PinRow};
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use crate::hashing::{Algorithm, BlobHasher};
use crate::keys::Key;
#[cfg(all(test, feature = "postgres"))]
use crate::pg_common::PgOpts;
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use crate::pg_lo::PgLoBackend;
#[cfg(feature = "postgres")]
use crate::postgres::{PgJointTx, PostgresBackend};
use crate::Error;
#[cfg(feature = "postgres")]
use sha2::{Digest as _, Sha256};
#[cfg(all(test, feature = "postgres"))]
use std::collections::BTreeSet;
#[cfg(all(test, feature = "postgres"))]
use std::path::PathBuf;
#[cfg(feature = "postgres")]
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(feature = "postgres")]
use std::sync::Arc;
use std::time::Duration;
#[cfg(feature = "postgres")]
use std::time::{SystemTime, UNIX_EPOCH};
#[cfg(all(feature = "postgres", feature = "pg-lo"))]
use tokio::io::{AsyncRead, AsyncReadExt};
/// The store's **fleet state** — what the constructor's `fleet: true`
/// declaration leaves the store carrying (crate-internal, never
/// public API — ADR-008 §3: fleet constructor parameters, not API):
/// the declaration's flag (the put/sweep paths branch on the state's
/// presence, never on re-deriving from the engine set), the pin TTL
/// the constructor took, and the **store-instance id** (`owner`,
/// ADR-011's vocabulary — the pin row's identity half).
// consumer: the three `_fleet` constructors (feature `postgres`) mint
// the state; the fleet put drivers read pin_ttl/owner — in builds
// without the consuming feature the carried state compiles unread.
#[allow(dead_code)]
pub(crate) struct FleetState {
    /// The pin TTL the fleet constructor took (default
    /// [`DEFAULT_PIN_TTL`]) — the store owns the TTL and the TTL/3
    /// renewal cadence; the engine's `pin_renew` successor-time body
    /// reads the TTL its backend was constructed with.
    pub(crate) pin_ttl: Duration,
    /// The store-instance id ([`mint_owner_id`]'s minting) — carried
    /// as the `owner` bytes of every [`crate::gc::PinRow`] the store
    /// mints.
    pub(crate) owner: Vec<u8>,
}

// consumer: the fleet put drivers' guard minting (feature `postgres`)
// and the fleet construction tests' introspection.
#[allow(dead_code)]
impl FleetState {
    /// The pin TTL the fleet constructor took.
    pub(crate) fn pin_ttl(&self) -> Duration {
        self.pin_ttl
    }

    /// The store-instance id — the `owner` bytes of the pin rows this
    /// instance mints.
    pub(crate) fn owner(&self) -> &[u8] {
        &self.owner
    }
}

/// **The store-instance id's minting** (named design point): 32 raw
/// bytes — a SHA-256 fold over the process id, the minting instant's
/// wall-clock nanos, a process-global instance counter, and an
/// address-space-randomized anchor — minted once at construction,
/// stable for the instance's life, carried as the `owner` bytes of
/// every [`crate::gc::PinRow`] the store mints. Uniqueness is the
/// contract, not secrecy: the counter distinguishes instances in one
/// process by construction; time + pid + the ASLR anchor distinguish
/// processes. The id is crate-internal (never public API — ADR-008:
/// fleet constructor parameters, not API).
#[cfg(feature = "postgres")]
pub(crate) fn mint_owner_id() -> Vec<u8> {
    static INSTANCE_SEQ: AtomicU64 = AtomicU64::new(0);
    let seq = INSTANCE_SEQ.fetch_add(1, Ordering::Relaxed);
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let anchor = &seq as *const u64 as usize;
    let mut hasher = Sha256::new();
    hasher.update(u64::from(std::process::id()).to_le_bytes());
    hasher.update(nanos.to_le_bytes());
    hasher.update(seq.to_le_bytes());
    hasher.update((anchor as u64).to_le_bytes());
    hasher.finalize().to_vec()
}

/// **The fleet put path (store-fleet-put; ADR-010 §2, ADR-012 §3/§4)** —
/// the joint entry+pin transaction's store-core driver and the pin
/// guard's fleet form (the TTL/3 renewal duty, ADR-008 §2), behind the
/// public `put`/`put_in_batch` signatures' unchanged routing: the paths
/// branch on [`Store::fleet`]'s presence (the declaration's flag, never
/// re-derived from the engine set), and every non-fleet path is this
/// module's ADR-005 in-process protocol verbatim.
///
/// **Named design point — the termination design (amendment A1, the
/// wave-4 gate's owner landed here): catch-propagate-terminate.** The
/// driver's steps between [`Store::begin` of the joint tx] and
/// [`pin_commit`] run inside one catch block (an async block's `?`
/// surface); any error terminates the held tx (a `ROLLBACK` through
/// [`PgJointTx::client`]) *before* propagating — no mid-flight
/// `PgJointTx` ever returns to the pool (deadpool's `Fast` recycle
/// never resets transaction state, PG-2's hazard). An explicit abort
/// verb was rejected — it would need a new ADR (the seam shape is
/// pinned; `pin_commit`'s terminal-consume reading stays), and the
/// rollback here issues nothing the clause bodies did not already
/// guarantee: a `ROLLBACK` on a session with no active transaction is
/// a completed, warning-level no-op, so the blanket termination after
/// any failure (including the clause bodies' own rollback paths) is
/// always safe; `pin_commit`'s failure paths (the terminal step, which
/// consumes the tx) roll back per the wave-4 clause contract. The pin
/// row's expiry mint (`fleet_pin_row`'s clock-overflow error) is a
/// pre-`pin_commit` step like any other: the kv driver terminates the
/// held tx on it, and the pg-lo drivers resolve the row inside their
/// caught drives. The zero-residue atomicity of the tx itself is the
/// engine clause's
/// contract (wave 4); the driver's duty is only never abandoning the
/// tx mid-flight — its row-only and one-LO drivers below each state
/// their interior `?` surface.
///
/// **The fleet put's per-shape drivers:**
///
/// - **kv tier (every fleet shape's sub-threshold route; the
///   re-routing posture's every put):** the in-process refcount mints
///   (pin-before-publish, unchanged), then `begin_joint_tx` →
///   `joint_put` (the engine's CAS row write inside the held tx; a
///   dedup re-put contributes nothing — the recorded CAS form) →
///   `pin_commit(tx, [row])` — publish and protection in one step.
/// - **large = pg-lo (the consolidation posture):** one tx over the
///   shared instance — the driver begins it and hands the large tier
///   its joint staging cursor; the cursor streams the content inside
///   the held tx (the 512 KiB `lowrite` knee rides), the key resolves
///   at stream end, the cursor's publish is the CAS contribution (a
///   re-put: the empty-contribution form — the staged LO unlinked
///   in-tx), and `pin_commit` lands the row with the entry.
/// - **large = local on declared shared media:** the joint tx is
///   unachievable across media, so the ordering is pin-first,
///   publish-second — a row-only joint tx (no `joint_put`) commits the
///   pin row **before** the staged commit-rename publishes the entry;
///   the invariant survives (a visible entry always has a durable pin
///   row) and the crash window moves to the harmless side (an orphan
///   pin row, reaped by the expiry machinery — never the reverse
///   ordering, which is rejected).
///
/// **Named design points — the renewal lifecycle (the pin guard's
/// fleet form, ADR-008 §2):**
///
/// - **The duty's mechanism: a per-guard tokio task on a TTL/3
///   interval.** The shared-renewal-sweep keyed off a guard registry
///   was the rejected alternative: it needs precisely the lifecycle
///   notifications (mint/drop/convert/batch-contribute) that an abort
///   handle gives for free, wrapped in a second registry keyed by
///   `(key, owner)` — machinery without a behavior difference, with
///   the per-guard task count bounded by the live pin count.
/// - **The spawn point is the pin row's durability, not the guard's
///   mint:** the task is spawned right after the row's `pin_commit`
///   succeeded (a renewal arriving at a nonexistent row would read the
///   not-yet-pinned window), and the first cadence tick is a full
///   TTL/3 after that — the row's initial `now + TTL` expiry is never
///   raced.
/// - **The abort-on-drop discipline:** the task is aborted by
///   [`Pin`]'s drop, by [`Batch`]'s one-step release (a contributed
///   member's duty moved into the batch at `add` time), and by the
///   ordered-step conversion (`RegistryCore::convert` stops the duty
///   before its `ManuallyDrop` suppresses the guard's drop) — no
///   leaked tasks, no renewal past the guard's life. A task aborted
///   mid-renewal is harmless: the clause's expiry update is
///   idempotent, and the row is exactly as it was.
/// - **The failure posture: silent-and-continued on `Err`, ended on
///   `Ok(false)`.** The row's expiry is the protection — a missed
///   renewal degrades to ADR-008's expiry semantics, never an error on
///   the put path (the crate carries no logging dependency; an embedder
///   needing visibility into renewals observes the row). A `Ok(false)`
///   — an expired or absent `(key, owner)` row — ends the duty: the
///   engine's `pin_renew` never resurrects (wave 4's recorded
///   semantics), so continuing would only churn.
impl Store {
    /// The fleet host — the kv-tier postgres backend behind its `Arc`
    /// (the detached renewal tasks share it). Every `fleet: true`
    /// store's kv engine is postgres (the composite fleet-validity
    /// predicate's kv clause), so a `None` here is only the
    /// declaration-absent case; where the fleet put paths read it, the
    /// branch on `Store::fleet` has already admitted the posture.
    #[cfg(feature = "postgres")]
    pub(crate) fn fleet_host(&self) -> Option<&Arc<PostgresBackend>> {
        match &self.kv {
            KvEngine::Postgres(engine) => Some(engine),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => None,
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => None,
        }
    }

    /// The fleet host when the **fleet machinery is active** — `Some`
    /// exactly on a declared fleet store whose kv engine is the
    /// postgres host (the composite fleet-validity predicate makes
    /// that pairing structural). Every fleet-clause consumer branches
    /// through this: under `fleet: false` (including every pre-existing
    /// constructor, even postgres-kv ones) it answers `None` and the
    /// in-process protocol applies verbatim.
    #[cfg(feature = "postgres")]
    pub(crate) fn fleet_sql_host(&self) -> Option<&Arc<PostgresBackend>> {
        if self.fleet.is_some() {
            self.fleet_host()
        } else {
            None
        }
    }

    /// The pin row a fleet put mints: the store-instance id as the
    /// `owner` half, `expiry = now + pin_ttl` (the store resolves the
    /// expiry; the TTL lives with the store's renewal schedule — ADR-012
    /// §2's rename).
    #[cfg(feature = "postgres")]
    fn fleet_pin_row(&self, fleet: &FleetState, key: &Key) -> Result<PinRow, Error> {
        let expiry = SystemTime::now()
            .checked_add(fleet.pin_ttl())
            .ok_or_else(|| {
                Error::Io("fleet: the pin expiry overflows the system clock".to_owned())
            })?;
        Ok(PinRow {
            key: key.as_bytes(),
            owner: fleet.owner().to_vec(),
            expiry,
        })
    }

    /// Mint the pin guard's renewal duty: the per-guard tokio task on
    /// the TTL/3 cadence (the lifecycle's named design points above).
    ///
    /// The cadence floor (1 ms) is the degenerate-tuning guard: a TTL
    /// below 3 ms would floor the interval to zero and churn the engine
    /// — the TTL/3 cadence must remain a *cadence*.
    #[cfg(feature = "postgres")]
    fn spawn_pin_renewal(
        &self,
        host: &Arc<PostgresBackend>,
        fleet: &FleetState,
        key: &Key,
    ) -> PinRenewal {
        let cadence = (fleet.pin_ttl() / 3).max(Duration::from_millis(1));
        let host = Arc::clone(host);
        let key = key.as_bytes();
        let owner = fleet.owner().to_vec();
        let handle = tokio::spawn(async move {
            loop {
                tokio::time::sleep(cadence).await;
                match EngineFleetState::pin_renew(&*host, &key, &owner).await {
                    Ok(true) => {}
                    Ok(false) => break,
                    Err(_) => {}
                }
            }
        });
        PinRenewal::spawned(handle)
    }

    /// Amendment A1's termination (the design's one verb): a step's
    /// failure between `begin_joint_tx` and `pin_commit` lands here —
    /// terminate the held tx before propagating. A `ROLLBACK` on a
    /// session with no active transaction (the clause bodies' own
    /// rollback paths having already closed it) is a completed
    /// warning-level no-op, never an error.
    #[cfg(feature = "postgres")]
    async fn terminate_joint_tx(tx: &mut PgJointTx) {
        if let Ok(client) = tx.client() {
            let _ = client.batch_execute("ROLLBACK").await;
        }
    }

    /// The kv tier's fleet put: the joint entry+pin tx with the CAS
    /// row write inside it (`joint_put` — a dedup re-put contributes
    /// nothing, the recorded CAS form) and the pin row's commit —
    /// publish and protection in one step. The returned guard carries
    /// the in-process refcount's release duty (unchanged) and the
    /// fleet row's renewal duty. The body's error paths terminate the
    /// held tx before propagating (A1, above) — the put paths' route
    /// into this is only ever a declared fleet store, which the
    /// `_fleet` constructors make impossible without the postgres
    /// host; the non-postgres build's arm is the dead shape, surfaced
    /// as an error never a panic.
    pub(crate) async fn fleet_put_kv(
        &self,
        fleet: &FleetState,
        key: Key,
        value: &[u8],
    ) -> Result<(Key, Pin), Error> {
        #[cfg(feature = "postgres")]
        {
            let host = match self.fleet_host() {
                Some(host) => host,
                None => {
                    return Err(Error::Io(
                        "fleet put without the postgres kv host".to_owned(),
                    ))
                }
            };
            let pin = self.gc().acquire(&key).await;
            let mut tx = match host.begin_joint_tx().await {
                Ok(tx) => tx,
                Err(e) => return Err(e),
            };
            match EngineFleetState::joint_put(&**host, &mut tx, &key.as_bytes(), value).await {
                Ok(()) => {}
                Err(e) => {
                    Self::terminate_joint_tx(&mut tx).await;
                    return Err(e);
                }
            }
            let row = match self.fleet_pin_row(fleet, &key) {
                Ok(row) => row,
                Err(e) => {
                    Self::terminate_joint_tx(&mut tx).await;
                    return Err(e);
                }
            };
            match EngineFleetState::pin_commit(&**host, tx, std::slice::from_ref(&row)).await {
                Ok(()) => {}
                Err(e) => return Err(e),
            }
            let mut pin = pin;
            pin.attach_renewal(self.spawn_pin_renewal(host, fleet, &key));
            Ok((key, pin))
        }
        #[cfg(not(feature = "postgres"))]
        {
            let _ = (fleet, key, value);
            Err(Error::Io(
                "fleet put without the postgres kv host: no fleet constructor exists \
                 without it"
                    .to_owned(),
            ))
        }
    }

    /// The row-only joint tx (pin row alone, no `joint_put`) — the
    /// large=local shared-media ordering's pin half (ADR-012 §4) and
    /// the terminal-consume seam's edge: between the tx's begin and
    /// its `pin_commit` there are no interior steps, so the A1
    /// discipline here is the clause's own failure rollback.
    #[cfg(feature = "postgres")]
    async fn fleet_pin_only_commit(
        &self,
        host: &Arc<PostgresBackend>,
        row: PinRow,
    ) -> Result<(), Error> {
        let tx = match host.begin_joint_tx().await {
            Ok(tx) => tx,
            Err(e) => return Err(e),
        };
        match EngineFleetState::pin_commit(&**host, tx, std::slice::from_ref(&row)).await {
            Ok(()) => Ok(()),
            Err(e) => Err(e),
        }
    }

    /// The shared-media ordering's staged commit (ADR-012 §4: pin
    /// row first, publish-second): the pin row's joint tx commits
    /// **before** the staged commit-rename publishes the entry; the
    /// crash window is the harmless side (an orphan pin row whose
    /// renewal stops at the guard's drop — the put's failure return —
    /// and whose expiry reclaims it). A pin-commit failure discards
    /// the stage and propagates with zero published residue.
    pub(crate) async fn fleet_staged_commit(
        &self,
        fleet: &FleetState,
        key: Key,
        stage: AnyStaging,
    ) -> Result<(Key, Pin), Error> {
        #[cfg(feature = "postgres")]
        {
            let host = match self.fleet_host() {
                Some(host) => host,
                None => {
                    stage.discard().await;
                    return Err(Error::Io(
                        "fleet put without the postgres kv host".to_owned(),
                    ));
                }
            };
            let mut pin = self.gc().acquire(&key).await;
            let row = self.fleet_pin_row(fleet, &key)?;
            match self.fleet_pin_only_commit(host, row).await {
                Ok(()) => {
                    pin.attach_renewal(self.spawn_pin_renewal(host, fleet, &key));
                }
                Err(e) => {
                    stage.discard().await;
                    return Err(e);
                }
            }
            stage.commit(&key.as_bytes()).await?;
            Ok((key, pin))
        }
        #[cfg(not(feature = "postgres"))]
        {
            let _ = (fleet, key, stage);
            Err(Error::Io(
                "fleet put without the postgres kv host: no fleet constructor exists \
                 without it"
                    .to_owned(),
            ))
        }
    }

    /// The joint publish's tail (shared by both pg-lo drivers): the
    /// row's `pin_commit` (the entry + pin row in one step — the CAS
    /// contribution, fresh or empty, is already inside the tx), then
    /// the in-process pin mint and the renewal's spawn onto the durable
    /// row. The publish-time protection is the row (durable with the
    /// entry in the one committed tx — ADR-012 §4); the in-process
    /// refcount is the instance-local bookkeeping minted after it.
    ///
    /// Every fallible step is resolved before the held tx can be
    /// abandoned: the host and the pin row arrive precomputed (the
    /// callers resolve both inside their caught drives — the row's
    /// expiry-mint failure is a pre-`pin_commit` path terminating the
    /// tx there), and `pin_commit` is the terminal-consume step whose
    /// own failure paths roll back per the wave-4 clause contract.
    #[cfg(feature = "postgres")]
    // consumer: the consolidation posture's one-tx publish (the pg-lo
    // drivers — features `postgres` + `pg-lo`; builds without `pg-lo`
    // compile the kv/shared-medium drivers only, which publish inline
    // through `pin_commit`).
    #[allow(dead_code)]
    async fn fleet_joint_publish(
        &self,
        host: &Arc<PostgresBackend>,
        fleet: &FleetState,
        tx: PgJointTx,
        key: Key,
        row: PinRow,
    ) -> Result<(Key, Pin), Error> {
        match EngineFleetState::pin_commit(&**host, tx, std::slice::from_ref(&row)).await {
            Ok(()) => {}
            Err(e) => return Err(e),
        }
        let mut pin = self.gc().acquire(&key).await;
        pin.attach_renewal(self.spawn_pin_renewal(host, fleet, &key));
        Ok((key, pin))
    }

    /// The pg-lo consolidation put's known-length driver: the joint tx
    /// begins first, the content streams into the tier's joint staging
    /// cursor while hashing (the declared-len `Verification` checks
    /// unchanged), the key resolves at stream end, the cursor's
    /// publish is the CAS contribution (a re-put takes the empty-
    /// contribution form — the staged LO unlinked in-tx), and
    /// `pin_commit` lands the row with the entry.
    ///
    /// Amendment A1's catch-propagate-terminate wraps exactly the
    /// steps between the tx's begin and `pin_commit`: any error
    /// (stream, staged append, the contribution's publish) is caught,
    /// terminates the tx (the staged LO's pages die with it —
    /// transactional LO creation, C6), and propagates.
    #[cfg(all(feature = "postgres", feature = "pg-lo"))]
    pub(crate) async fn fleet_put_pglo_known<R: AsyncRead + Unpin + Send>(
        &self,
        fleet: &FleetState,
        large: &PgLoBackend,
        algorithm: Algorithm,
        len: u64,
        mut stream: R,
    ) -> Result<(Key, Pin), Error> {
        let host = match self.fleet_host() {
            Some(host) => host,
            None => {
                return Err(Error::Io(
                    "fleet put without the postgres kv host".to_owned(),
                ))
            }
        };
        let mut tx = match host.begin_joint_tx().await {
            Ok(tx) => tx,
            Err(e) => return Err(e),
        };
        let driven = async {
            let mut stage = large.put_joint_staged(&mut tx).await?;
            let mut hasher = BlobHasher::new(algorithm, len);
            let mut chunk = vec![0u8; PUT_CHUNK];
            let mut received = 0u64;
            let fed: Result<(), Error> = 'stream: loop {
                let n = match stream.read(&mut chunk).await {
                    Ok(n) => n,
                    Err(e) => break 'stream Err(Error::Io(e.to_string())),
                };
                if n == 0 {
                    break 'stream Ok(());
                }
                received = received.saturating_add(n as u64);
                if received > len {
                    break 'stream Err(Error::Verification);
                }
                hasher.update(&chunk[..n]);
                if let Err(e) = stage.append(&chunk[..n]).await {
                    break 'stream Err(e);
                }
            };
            fed?;
            if received != len {
                return Err(Error::Verification);
            }
            let key = Key::from(hasher.finish());
            let _fresh = stage.publish(&key.as_bytes()).await?;
            let row = self.fleet_pin_row(fleet, &key)?;
            Ok((key, row))
        }
        .await;
        match driven {
            Ok((key, row)) => self.fleet_joint_publish(host, fleet, tx, key, row).await,
            Err(e) => {
                Self::terminate_joint_tx(&mut tx).await;
                Err(e)
            }
        }
    }

    /// The pg-lo consolidation put's unknown-length driver: the
    /// buffer-to-threshold phase first, exactly `put_unknown`'s rule 1
    /// (a stream that ends sub-threshold stays on the kv tier's joint
    /// buffer form); the tx begins only when the threshold overflows,
    /// the buffered prefix flushes into the joint cursor and the
    /// stream continues inside the held tx; the derivation pass
    /// re-runs over the staged artifact through the joint cursor's
    /// read seam (rule 2 — read-your-own-writes inside the held tx);
    /// the CAS publish and `pin_commit` close the flow in the one tx.
    #[cfg(all(feature = "postgres", feature = "pg-lo"))]
    pub(crate) async fn fleet_put_pglo_unknown<R: AsyncRead + Unpin + Send>(
        &self,
        fleet: &FleetState,
        large: &PgLoBackend,
        algorithm: Algorithm,
        mut stream: R,
    ) -> Result<(Key, Pin), Error> {
        let host = match self.fleet_host() {
            Some(host) => host,
            None => {
                return Err(Error::Io(
                    "fleet put without the postgres kv host".to_owned(),
                ))
            }
        };
        let Some(threshold) = self.threshold() else {
            return Err(Error::Io(
                "no dispatch threshold in effect; large-tier routing cannot decide".to_owned(),
            ));
        };
        let mut buffered = Vec::with_capacity(PUT_CHUNK);
        let mut streamed_total = 0u64;
        let mut chunk = vec![0u8; PUT_CHUNK];
        let overflow_n;

        'feed: loop {
            let n = match stream.read(&mut chunk).await {
                Ok(n) => n,
                Err(e) => return Err(Error::Io(e.to_string())),
            };
            if n == 0 {
                let mut hasher = BlobHasher::new(algorithm, streamed_total);
                hasher.update(&buffered);
                let key = Key::from(hasher.finish());
                return self.fleet_put_kv(fleet, key, &buffered).await;
            }
            streamed_total = streamed_total.saturating_add(n as u64);
            let overflow = (buffered.len() as u64).saturating_add(n as u64) > threshold;
            if !overflow {
                buffered.extend_from_slice(&chunk[..n]);
                continue;
            }
            overflow_n = n;
            break 'feed;
        }

        let mut tx = match host.begin_joint_tx().await {
            Ok(tx) => tx,
            Err(e) => return Err(e),
        };
        let driven = async {
            let mut stage = large.put_joint_staged(&mut tx).await?;
            stage.append(&buffered).await?;
            stage.append(&chunk[..overflow_n]).await?;
            let fed: Result<(), Error> = 'stream: loop {
                let n = match stream.read(&mut chunk).await {
                    Ok(n) => n,
                    Err(e) => break 'stream Err(Error::Io(e.to_string())),
                };
                if n == 0 {
                    break 'stream Ok(());
                }
                if let Err(e) = stage.append(&chunk[..n]).await {
                    break 'stream Err(e);
                }
            };
            fed?;
            let mut hasher = BlobHasher::new(algorithm, streamed_total);
            let mut offset = 0u64;
            let derived: Result<(), Error> = 'pass: loop {
                if offset >= streamed_total {
                    break 'pass Ok(());
                }
                let want = (streamed_total - offset).min(PUT_CHUNK as u64);
                let bytes = match stage.read_at(offset, want).await {
                    Ok(bytes) => bytes,
                    Err(e) => break 'pass Err(e),
                };
                if bytes.is_empty() {
                    break 'pass Err(Error::Io(
                        "staged artifact under-read during the derivation pass".to_owned(),
                    ));
                }
                hasher.update(&bytes);
                offset += bytes.len() as u64;
            };
            derived?;
            let key = Key::from(hasher.finish());
            let _fresh = stage.publish(&key.as_bytes()).await?;
            let row = self.fleet_pin_row(fleet, &key)?;
            Ok((key, row))
        }
        .await;
        match driven {
            Ok((key, row)) => self.fleet_joint_publish(host, fleet, tx, key, row).await,
            Err(e) => {
                Self::terminate_joint_tx(&mut tx).await;
                Err(e)
            }
        }
    }
}

/// The wave-6 fleet put path's store-core tests (feature `postgres`)
/// for the kv tier and the local shared-media tier: the joint
/// entry+pin tx driver exercised through the public `put` /
/// `put_in_batch` — the kv tier's joint form (entry + pin row in one
/// step, seen by an independent connection), the A1 termination
/// discipline's forced kv failure form, the pin guard's TTL/3 renewal
/// duty (cadence, drop/conversion/batch stop, the never-resurrect
/// form), the dedup re-put's row-alone upsert-renewal and the by-row
/// protection composition, the shared-media pin-first /
/// publish-second ordering, and the `fleet: false` non-event (no pin
/// rows at all). Env-driven (`ALKBLOBS_PG_*`), skipping cleanly
/// server-less; every test takes a fresh engine-owned schema on the
/// harness database (no LOs involved here — the LO-bearing shapes' own
/// module is `postgres_pglo_fleet_put_tests`) with a janitor removing
/// it in every exit path.
#[cfg(all(test, feature = "postgres"))]
mod postgres_fleet_put_tests {
    use super::*;
    use crate::pg_common::qualified_table;
    use crate::postgres::test_support::SchemaJanitor;
    use std::io::Cursor;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio_postgres::NoTls;

    static SCHEMA_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"
        );
    }

    struct TestSite {
        dsn: String,
        db: String,
        schema: String,
        _janitor: SchemaJanitor,
    }

    impl TestSite {
        /// The test's own session tag — the fresh schema's name, unique
        /// across concurrently-running test instances by its pid +
        /// sequence parts. Every store this test constructs carries it
        /// as its pools' `application_name`, so server-global
        /// observability probes scoped to the tag count exactly this
        /// test's sessions.
        fn session_tag(&self) -> String {
            self.schema.clone()
        }
    }

    async fn fresh_site(tag: &str) -> Option<TestSite> {
        let dsn = harness_dsn()?;
        let schema = format!(
            "abtfput_{}_{}_{:x}",
            tag,
            std::process::id(),
            SCHEMA_SEQ.fetch_add(1, Ordering::Relaxed)
        );
        Some(TestSite {
            dsn: dsn.clone(),
            db: env_nonempty("ALKBLOBS_PG_DB").unwrap_or_else(|| "blobs".to_owned()),
            schema: schema.clone(),
            _janitor: SchemaJanitor::new(&dsn, &schema),
        })
    }

    /// A fresh temp root per test (the local tier's medium); Drop
    /// removes the whole 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-fput-{tag}-{}-{:x}-{seq}",
            std::process::id(),
            seq,
        ));
        std::fs::create_dir_all(&dir).expect("temp root created");
        TempRoot(dir)
    }

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

    /// A short test pin TTL: the TTL/3 cadence observable within a
    /// test's runtime (≈1.3 s renewals, ≈4 s expiry horizon). Widened
    /// past the original 2 s shape on the load-tolerance posture: a
    /// starved-but-landed tick must still find the row alive.
    const TEST_TTL: Duration = Duration::from_secs(4);
    /// The bounded window a renewal may land in — the positive cadence
    /// arms poll until the expiry moves instead of sampling once at a
    /// fixed wall time, so a loaded-but-unsaturated server cannot miss
    /// the window (the tolerance gap the wall-clock sampling had).
    /// 3x the nominal ≈1.3 s cadence, still comfortably inside the
    /// widened TTL's unexpired life.
    const RENEW_WINDOW: Duration = Duration::from_millis(3000);
    /// The poll granularity of the renewal arms.
    const POLL_STEP: Duration = Duration::from_millis(100);
    /// More than a cadence beyond a stopped renewal: the frozen row
    /// proves the task ended (no leaked renewals).
    const QUIET_WINDOW: Duration = Duration::from_millis(1800);

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

    /// The independent connection — the "sees both or neither"
    /// acceptance's instrument: every probe below runs outside the
    /// store's pool, over its own connection, seeing only durable
    /// state.
    struct Independent {
        client: tokio_postgres::Client,
        _conn: tokio::task::JoinHandle<()>,
    }

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

    impl Independent {
        /// The gc_pins rows in their stored encoding, in the
        /// `(key, owner)` sort order — the pool's durable protection
        /// set as a second process sees it.
        async fn pin_rows(&self, schema: &str) -> Vec<(Vec<u8>, Vec<u8>, i64)> {
            let sql = format!(
                "SELECT key, owner, expiry FROM {} ORDER BY key, owner",
                qualified_table(schema, "gc_pins")
            );
            let rows = self
                .client
                .query(sql.as_str(), &[])
                .await
                .expect("pin rows");
            rows.iter()
                .map(|row| {
                    (
                        row.get::<_, Vec<u8>>(0),
                        row.get::<_, Vec<u8>>(1),
                        row.get::<_, i64>(2),
                    )
                })
                .collect()
        }

        /// The kv engine's entry rows `(key, size)`.
        async fn kv_rows(&self, schema: &str) -> Vec<(Vec<u8>, i64)> {
            let sql = format!(
                "SELECT key, size FROM {} ORDER BY key",
                qualified_table(schema, "blobs")
            );
            let rows = self.client.query(sql.as_str(), &[]).await.expect("kv rows");
            rows.iter()
                .map(|row| (row.get::<_, Vec<u8>>(0), row.get::<_, i64>(1)))
                .collect()
        }

        /// The manual expiry instrument (the abandonment shape the
        /// maintenance step reaps — no timing reliance).
        async fn expire_pin_rows(&self, schema: &str, key: &[u8]) {
            let expired = crate::pg_common::time_to_nanos(SystemTime::now()) - 1_000_000_000;
            let sql = format!(
                "UPDATE {} SET expiry = $1 WHERE key = $2",
                qualified_table(schema, "gc_pins")
            );
            self.client
                .execute(sql.as_str(), &[&expired, &key])
                .await
                .expect("the manual expiry succeeds");
        }

        /// The pooled sessions sitting in a not-yet-committed
        /// transaction, scoped to this store family's own sessions by
        /// the instances' `application_name` tag — the dirty-session
        /// observable the A1 termination design's test asserts to zero
        /// (deadpool's `Fast` recycle never resets transaction state,
        /// PG-2). The `datname`-only scope is cross-test by
        /// construction (every concurrently-running test's transient
        /// idle-in-transaction session lands in whole-database counts);
        /// the tag makes the zero-asserts exact.
        async fn idle_in_transaction_sessions(&self, db: &str, tag: &str) -> i64 {
            self.client
                .query_one(
                    "SELECT count(*) FROM pg_stat_activity \
                     WHERE datname = $1 AND application_name = $2 \
                     AND state IS NOT NULL \
                     AND state LIKE 'idle in transaction%'",
                    &[&db, &tag],
                )
                .await
                .expect("pg_stat_activity probe")
                .get::<_, i64>(0)
        }
    }

    /// The store's pin rows for its own instance id — the filters
    /// keeping multi-instance tests' assertions per-instance
    /// (protection composes by row).
    fn instance_rows(rows: &[(Vec<u8>, Vec<u8>, i64)], owner: &[u8]) -> Vec<(Vec<u8>, i64)> {
        rows.iter()
            .filter(|(_, row_owner, _)| row_owner == owner)
            .map(|(key, _, expiry)| (key.clone(), *expiry))
            .collect()
    }

    /// A store instance's sole row's expiry (the single-row shape's
    /// tests); `None` when the row is not there.
    fn expiry_of_row(rows: &[(Vec<u8>, Vec<u8>, i64)], owner: &[u8]) -> Option<i64> {
        instance_rows(rows, owner)
            .into_iter()
            .next()
            .map(|(_, expiry)| expiry)
    }

    /// A particular `(key, owner)` row's expiry.
    fn row_expiry(rows: &[(Vec<u8>, Vec<u8>, i64)], owner: &[u8], key: &Key) -> Option<i64> {
        rows.iter()
            .find(|(k, o, _)| k == &key.as_bytes() && o == owner)
            .map(|(_, _, expiry)| *expiry)
    }

    async fn fleet_kv_only_store(site: &TestSite, ttl: Duration) -> (Store, Vec<u8>) {
        let store = Store::kv_only_postgres_fleet(
            &site.dsn,
            PgOpts {
                schema: site.schema.clone(),
                application_name: site.session_tag(),
                ..Default::default()
            },
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            ttl,
        )
        .await
        .expect("the fleet kv-only store constructs");
        let owner = store.fleet().expect("fleet state").owner().to_vec();
        (store, owner)
    }

    async fn sleep_of(d: Duration) {
        tokio::time::sleep(d).await;
    }

    /// The kv tier's joint form (the re-routing client posture): the
    /// put returns byte-exact, and an independent connection sees
    /// **both** the entry row and its pin row — one committed step.
    /// The pin row carries the store-instance id as its owner and an
    /// expiry at `now + TTL`.
    #[tokio::test]
    async fn fleet_kv_put_publishes_entry_and_pin_row_together() {
        let Some(site) = fresh_site("fput-kv-joint").await else {
            skip_note();
            return;
        };
        let (store, owner) = fleet_kv_only_store(&site, DEFAULT_PIN_TTL).await;

        let content = b"a sub-threshold blob under the fleet joint tx".to_vec();
        let (key, pin) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the fleet kv put publishes");
        assert_eq!(key, key_for(&content));

        // The in-process refcount machinery continues to function
        // (instance-local bookkeeping; unchanged shape).
        assert_eq!(store.gc().pin_count(&key), 1);
        drop(pin);
        assert_eq!(store.gc().pin_count(&key), 0);

        let probe = independent(&site.dsn).await;
        assert_eq!(
            probe.kv_rows(&site.schema).await,
            vec![(key.as_bytes(), content.len() as i64)],
            "the entry row is durable"
        );
        let now = SystemTime::now();
        let rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        assert_eq!(rows.len(), 1, "exactly this instance's pin row");
        assert_eq!(rows[0].0, key.as_bytes());
        assert!(
            rows[0].1
                > crate::pg_common::time_to_nanos(now + DEFAULT_PIN_TTL - Duration::from_secs(5)),
            "expiry = now + TTL at the joint tx's commit"
        );

        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry reads back");
        assert_eq!(len, content.len() as u64);
        assert_eq!(blob.read_at(0, len).await.unwrap(), content);
    }

    /// The kv joint tx's zero-residue failure form (A1's propagation
    /// through the public `put`; the clause's own rollback is part of
    /// the wave-4 contract): a forced `pin_commit` failure — publish +
    /// protection landing in one step, so both miss together — leaves
    /// the pool's sessions committed-transaction-free and the store's
    /// tables empty.
    #[tokio::test]
    async fn fleet_kv_joint_commit_failure_leaves_neither_entry_nor_pin() {
        let Some(site) = fresh_site("fput-kv-fail").await else {
            skip_note();
            return;
        };
        let (mut store, _owner) = fleet_kv_only_store(&site, DEFAULT_PIN_TTL).await;
        match &mut store.kv {
            KvEngine::Postgres(engine) => match Arc::get_mut(engine) {
                Some(backend) => backend.inject_fail_joint_commit(),
                None => panic!("the store uniquely holds its own engine"),
            },
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("a postgres store is required"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("a postgres store is required"),
        }

        let content = b"content whose joint tx is forced to fail".to_vec();
        let err = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("injected"),
            "the engine's forced failure surfaces: {err}"
        );

        let probe = independent(&site.dsn).await;
        assert_eq!(
            probe.kv_rows(&site.schema).await,
            vec![],
            "no published entry"
        );
        assert_eq!(probe.pin_rows(&site.schema).await, vec![], "no pin rows");
        assert_eq!(
            probe
                .idle_in_transaction_sessions(&site.db, &site.session_tag())
                .await,
            0,
            "no session returned to the pool holding a transaction (A1)"
        );
        assert_eq!(
            store.list().await.unwrap(),
            Vec::<Key>::new(),
            "zero residue"
        );
    }

    /// A session tagged with an explicit `application_name`, outside
    /// the stores' pools — the scoped probe's red/green instrument:
    /// the noise vector (another concurrently-running test's transient
    /// idle-in-transaction session) and the deliberate residue (a
    /// stranded session this test owns) are both manufactured this
    /// way.
    struct Tagged {
        client: tokio_postgres::Client,
        _conn: tokio::task::JoinHandle<()>,
    }

    async fn tagged_connection(dsn: &str, tag: &str) -> Tagged {
        let mut cfg: tokio_postgres::Config = dsn.parse().expect("harness dsn parsed");
        cfg.application_name(tag);
        let (client, conn) = cfg.connect(NoTls).await.expect("tagged connection");
        Tagged {
            client,
            _conn: tokio::spawn(async move {
                let _ = conn.await;
            }),
        }
    }

    /// The dirty-session probe's exact scoping (the isolation
    /// hardening's own acceptance): the tag is per-test-instance, a
    /// concurrently-running test's dirty session cannot poison the
    /// zero-assert, and — the red/green check, a mis-scoped tag would
    /// make the assert silently vacuous — a deliberately stranded
    /// session in the test's own tag still fires. Tag uniqueness
    /// across instances is this test's explicit first act (two fresh
    /// sites' tags are distinct; cross-test uniqueness rides the
    /// pid + sequence parts the schema names carry).
    #[tokio::test]
    async fn the_session_tag_scopes_the_dirty_session_probe_exactly() {
        let Some(site) = fresh_site("fput-scoped-a").await else {
            skip_note();
            return;
        };
        let Some(other) = fresh_site("fput-scoped-b").await else {
            skip_note();
            return;
        };
        assert_ne!(
            site.session_tag(),
            other.session_tag(),
            "two concurrently-running instances' tags cannot collide"
        );

        let probe = independent(&site.dsn).await;
        assert_eq!(
            probe
                .idle_in_transaction_sessions(&site.db, &site.session_tag())
                .await,
            0,
            "the clean starting state"
        );

        // The noise vector, materialized: another instance's dirty
        // session — same database, foreign tag. The probe must not
        // see it.
        let noise = tagged_connection(&site.dsn, &other.session_tag()).await;
        noise
            .client
            .batch_execute("BEGIN")
            .await
            .expect("the noise session's transaction");
        assert_eq!(
            probe
                .idle_in_transaction_sessions(&site.db, &site.session_tag())
                .await,
            0,
            "a concurrent test's dirty session cannot poison the zero-assert"
        );

        // The red half: the test's own stranded session — the tag
        // carries it and the assert fires (scoping is never vacuous).
        let stranded = tagged_connection(&site.dsn, &site.session_tag()).await;
        stranded
            .client
            .batch_execute("BEGIN")
            .await
            .expect("the stranded session's transaction");
        assert_eq!(
            probe
                .idle_in_transaction_sessions(&site.db, &site.session_tag())
                .await,
            1,
            "real residue under the test's own tag still fires the assert"
        );

        // The green return: terminating the residue restores zero.
        let _ = noise.client.batch_execute("ROLLBACK").await;
        let _ = stranded.client.batch_execute("ROLLBACK").await;
        drop(stranded);
        drop(noise);
        assert_eq!(
            probe
                .idle_in_transaction_sessions(&site.db, &site.session_tag())
                .await,
            0,
            "the scoped probe returns to zero once the residue is cleaned"
        );
    }

    /// The pin guard's renewal duty (a short test TTL; the cadence
    /// observed): while the guard lives, its row's expiry moves
    /// forward every TTL/3; the guard's drop stops the renewal; the
    /// ordered-step conversion stops it likewise — no leaked tasks, no
    /// renewal past the guard's life.
    #[tokio::test]
    async fn fleet_pin_row_renews_until_the_guard_drops_or_converts() {
        let Some(site) = fresh_site("fput-renewal").await else {
            skip_note();
            return;
        };
        let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await;
        let probe = independent(&site.dsn).await;

        // While the guard lives, the cadence moves the expiry; the
        // drop stops the duty and the row freezes.
        let content = b"a guard whose row renews while it lives".to_vec();
        let (_, pin) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the fleet kv put publishes");
        let initial = expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row");
        let deadline = std::time::Instant::now() + RENEW_WINDOW;
        let mut renewed =
            expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row");
        while renewed <= initial {
            assert!(
                std::time::Instant::now() < deadline,
                "the guard's TTL/3 duty moved the row's expiry forward within the window"
            );
            sleep_of(POLL_STEP).await;
            renewed = expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row");
        }
        assert!(
            renewed > initial,
            "the guard's TTL/3 duty moved the row's expiry forward"
        );
        drop(pin);
        let at_drop = expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row");
        assert_eq!(
            renewed, at_drop,
            "no renewal between the last tick and the drop"
        );
        sleep_of(QUIET_WINDOW).await;
        assert_eq!(
            at_drop,
            expiry_of_row(&probe.pin_rows(&site.schema).await, &owner).expect("the row"),
            "the dropped guard's renewal is stopped — no leaked tasks, \
             no renewal past the guard's life"
        );

        // The conversion boundary: a converted guard stops its duty —
        // the ordered-step conversion's pre-stop (RegistryCore::
        // convert's ManuallyDrop would otherwise suppress the guard's
        // own stop).
        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)
            }
        }
        let source = Arc::new(SetSource {
            live: parking_lot::Mutex::new(BTreeSet::new()),
        });
        let content_two = b"a guard converted into a registered source".to_vec();
        let (second_key, guard) = store
            .put(
                Some(content_two.len() as u64),
                Cursor::new(&content_two[..]),
            )
            .await
            .expect("the second put publishes");
        let at_mint = row_expiry(&probe.pin_rows(&site.schema).await, &owner, &second_key)
            .expect("the second row");
        let registered = second_key.clone();
        let source_clone = Arc::clone(&source);
        store.gc().convert(guard, move || {
            source_clone.live.lock().insert(registered);
        });
        assert_eq!(
            store.gc().pin_count(&second_key),
            0,
            "the conversion released the refcount in the ordered step"
        );
        sleep_of(QUIET_WINDOW).await;
        assert_eq!(
            at_mint,
            row_expiry(&probe.pin_rows(&site.schema).await, &owner, &second_key).expect("the row"),
            "the converted guard's duty is stopped with its life"
        );
    }

    /// The never-resurrect form: an expired row is never renewed back
    /// to life — the guard's renewal observes `pin_renew`'s `false`
    /// and the duty ends (the delete-then-recover backstop's
    /// territory, ADR-008's expiry semantics verbatim).
    #[tokio::test]
    async fn fleet_pin_renewal_never_resurrects_an_expired_row() {
        let Some(site) = fresh_site("fput-expired").await else {
            skip_note();
            return;
        };
        let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await;
        let content = b"a guard whose row a manual expiry abandons".to_vec();
        let (key, _pin) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the fleet kv put publishes");

        // Expire the row out from under the guard: the abandonment the
        // sweeper's maintenance step would reap.
        let expired = crate::pg_common::time_to_nanos(SystemTime::now() - Duration::from_secs(1));
        let probe = independent(&site.dsn).await;
        probe
            .client
            .execute(
                &format!(
                    "UPDATE {} SET expiry = $1 WHERE key = $2 AND owner = $3",
                    qualified_table(site.schema.as_str(), "gc_pins")
                ),
                &[&expired, &key.as_bytes(), &owner],
            )
            .await
            .expect("the manual expiry");

        // One cadence beyond the manual expiry: a tick would have
        // landed and the row would have moved if the guard renewed —
        // it stays dead (the abandonment the sweeper's maintenance
        // step would reap).
        sleep_of(QUIET_WINDOW).await;
        let rows = probe.pin_rows(&site.schema).await;
        let row = rows
            .iter()
            .find(|(k, o, _)| k == &key.as_bytes() && o == &owner)
            .expect("the row still holds its slot");
        assert_eq!(
            row.2, expired,
            "the expired row was never renewed back to life"
        );
    }

    /// The batch form: each member's joint tx commits its row; while
    /// the batch lives the members' rows renew; the batch's one-step
    /// release ends the members' renewals (the rows expire and the
    /// sweeper's maintenance step reaps them — `store-fleet-sweep`'s).
    #[tokio::test]
    async fn fleet_batch_release_ends_the_members_renewals() {
        let Some(site) = fresh_site("fput-batch").await else {
            skip_note();
            return;
        };
        let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await;
        let first = b"batch member one".to_vec();
        let second = b"batch member two".to_vec();

        let mut batch = store.batch();
        let k1 = store
            .put_in_batch(
                &mut batch,
                Some(first.len() as u64),
                Cursor::new(&first[..]),
            )
            .await
            .expect("member one publishes");
        let k2 = store
            .put_in_batch(
                &mut batch,
                Some(second.len() as u64),
                Cursor::new(&second[..]),
            )
            .await
            .expect("member two publishes");

        let probe = independent(&site.dsn).await;
        let initial = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        assert_eq!(initial.len(), 2, "each member's joint tx committed its row");

        let deadline = std::time::Instant::now() + RENEW_WINDOW;
        let mut renewed = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        while !renewed
            .iter()
            .zip(initial.iter())
            .all(|((_, r), (_, i))| r > i)
        {
            assert!(
                std::time::Instant::now() < deadline,
                "the members' rows renew while the batch lives"
            );
            sleep_of(POLL_STEP).await;
            renewed = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        }
        assert!(
            renewed
                .iter()
                .zip(initial.iter())
                .all(|((_, r), (_, i))| r > i),
            "the members' rows renew while the batch lives"
        );

        batch.commit();
        assert_eq!(store.gc().pin_count(&k1), 0, "the batch's one-step release");
        assert_eq!(store.gc().pin_count(&k2), 0);
        let at_release = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        sleep_of(QUIET_WINDOW).await;
        let frozen = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        assert_eq!(
            at_release, frozen,
            "the batch's release ended the members' renewals"
        );
    }

    /// The dedup re-put under fleet: CAS dedup means no new entry —
    /// the same store's re-put commits the pin row alone (the joint
    /// tx's row upsert renewing `(key, owner)`, the wave-4 recorded
    /// shape), and a different instance's re-put is a different pin
    /// row — protection composes by row (ADR-010 §2).
    #[tokio::test]
    async fn fleet_dedup_reput_commits_the_pin_row_alone_and_composes_by_row() {
        let Some(site) = fresh_site("fput-dedup").await else {
            skip_note();
            return;
        };
        let (store, owner) = fleet_kv_only_store(&site, TEST_TTL).await;
        let content = b"put twice; addressed once".to_vec();

        let (key, pin_one) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the first put publishes");
        let probe = independent(&site.dsn).await;
        let first_rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        assert_eq!(probe.kv_rows(&site.schema).await.len(), 1, "one entry row");
        drop(pin_one);

        let (key_again, pin_two) = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the dedup re-put publishes");
        assert_eq!(key_again, key, "CAS dedup under the same canonical key");
        assert_eq!(
            probe.kv_rows(&site.schema).await.len(),
            1,
            "the dedup re-put committed the pin row alone — no new entry row"
        );
        let renewed_rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        assert_eq!(
            renewed_rows.len(),
            1,
            "the same store's row, upsert-renewed"
        );
        assert!(
            renewed_rows[0].1 >= first_rows[0].1,
            "the row upsert-renews (key, owner): expiry moves forward"
        );

        // A different instance's re-put: a different pin row.
        let other_store = Store::kv_only_postgres_fleet(
            &site.dsn,
            PgOpts {
                schema: site.schema.clone(),
                application_name: site.session_tag(),
                ..Default::default()
            },
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            TEST_TTL,
        )
        .await
        .expect("the second instance constructs");
        let other_owner = other_store.fleet().expect("fleet state").owner().to_vec();
        assert_ne!(owner, other_owner, "two instances, two store-instance ids");
        other_store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .expect("the foreign instance's re-put publishes");
        assert_eq!(
            probe.kv_rows(&site.schema).await.len(),
            1,
            "still one entry row across instances"
        );
        let rows = probe.pin_rows(&site.schema).await;
        assert_eq!(
            rows.len(),
            2,
            "a different instance's re-put is a different pin row — \
             protection composes by row"
        );
        assert!(
            row_expiry(&rows, &owner, &key).is_some()
                && row_expiry(&rows, &other_owner, &key).is_some(),
            "both instances' rows present, keyed by their own owner"
        );
        let _ = pin_two;
    }

    /// A liveness source answering nothing — the sweep gate's
    /// registration (the fleet pin rows compose behind it; ADR-012
    /// §1's mark composition).
    struct NoLive;

    impl LivenessSource for NoLive {
        fn live_keys(&self) -> Vec<Key> {
            Vec::new()
        }
        fn contains(&self, _key: &Key) -> bool {
            false
        }
    }

    /// The git-object family put under a fleet declaration (ADR-018
    /// §Decision 7): `put_git_object` routes the kv tier's joint
    /// entry+pin tx — the entry row and its pin row land in one
    /// committed step, the derivation flows from the caller's family
    /// choice (no blob hasher anywhere on the path) — the loose-object
    /// dedup re-put commits the pin row alone, and the family key
    /// rides the sweep machinery with exact counts (held pins protect;
    /// the expired shape stages and deletes exactly the one entry).
    #[tokio::test]
    async fn fleet_kv_put_git_object_lands_through_the_joint_tx() {
        let Some(site) = fresh_site("fput-kv-gitobj").await else {
            skip_note();
            return;
        };
        let store = Store::kv_only_postgres_fleet(
            &site.dsn,
            PgOpts {
                schema: site.schema.clone(),
                application_name: site.session_tag(),
                ..Default::default()
            },
            64,
            Duration::ZERO,
            TEST_TTL,
        )
        .await
        .expect("the fleet kv-only store constructs");
        let _ = store.register_liveness_source(crate::facade::LivenessRegistration::InProcess(
            Arc::new(NoLive) as Arc<dyn LivenessSource>,
        ));

        // git's empty-tree loose form: the digest is the empty tree's
        // oid (git's own well-known constant, CLI-validated in
        // tests/git_interop.rs).
        let loose: &[u8] = b"tree 0\0";
        let (key, pin) = store
            .put_git_object(
                Algorithm::GitObjectSha1,
                loose.len() as u64,
                Cursor::new(loose),
            )
            .await
            .expect("the fleet git-object put publishes");
        assert_eq!(key.as_bytes()[0], 0x03, "the family tag");
        assert_eq!(
            hex::encode(key.digest().bytes()),
            "4b825dc642cb6eb9a060e54bf8d69288fbee4904",
            "the digest is git's oid under the family's derivation"
        );
        assert_eq!(store.gc().pin_count(&key), 1);

        let probe = independent(&site.dsn).await;
        assert_eq!(
            probe.kv_rows(&site.schema).await,
            vec![(key.as_bytes(), loose.len() as i64)],
            "the entry row is durable, joint with its pin"
        );
        assert_eq!(
            probe.pin_rows(&site.schema).await.len(),
            1,
            "the pin row landed in the same committed step"
        );
        assert_eq!(
            hex::encode(&probe.kv_rows(&site.schema).await[0].0),
            hex::encode(key.as_bytes())
        );

        // The dedup re-put: CAS by oid, the pin row alone.
        let (key_again, pin_two) = store
            .put_git_object(
                Algorithm::GitObjectSha1,
                loose.len() as u64,
                Cursor::new(loose),
            )
            .await
            .expect("the dedup re-put publishes");
        assert_eq!(key_again, key);
        assert_eq!(
            probe.kv_rows(&site.schema).await.len(),
            1,
            "one entry row all along"
        );

        // Held pins: the sweep stages and deletes nothing — the
        // in-process marks are the staging-time live set.
        let report = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(report.aborted, None);
        assert_eq!(
            (
                report.candidates_staged,
                report.deleted,
                report.cancelled_by_arbitration
            ),
            (0, 0, 0),
            "the held pins protect the family entry exactly as a blob's do"
        );

        // Unprotected and expired: the family entry stages and deletes
        // exactly once — the GC machinery is family-blind by tag only.
        drop(pin);
        drop(pin_two);
        probe.expire_pin_rows(&site.schema, &key.as_bytes()).await;
        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(staged.aborted, None);
        assert_eq!(
            staged.candidates_staged, 1,
            "the unprotected family entry staged exactly once"
        );
        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(executed.aborted, None);
        assert_eq!(
            (executed.deleted, executed.cancelled_by_arbitration),
            (1, 0),
            "the executing sweep deleted exactly the one family entry"
        );
        assert!(
            !store.has(&key.as_bytes()).await.unwrap(),
            "the family entry is gone"
        );
    }

    /// The shared-media ordering (ADR-012 §4): pin row first,
    /// publish-second. The success form lands both; the forced publish
    /// failure is the ordering's proof — the pin row **existed and was
    /// durable before the committed file came to be**, and the failed
    /// put leaves no committed entry, no stage file, and no in-process
    /// refcount behind (the orphan pin row the expiry machinery reaps;
    /// the harmless crash window).
    #[tokio::test]
    async fn fleet_local_shared_media_put_is_pin_first_publish_second() {
        let Some(site) = fresh_site("fput-local").await else {
            skip_note();
            return;
        };
        let (store, owner) = {
            let store = Store::dual_tier_postgres_local_fleet(
                &site.dsn,
                PgOpts {
                    schema: site.schema.clone(),
                    application_name: site.session_tag(),
                    ..Default::default()
                },
                {
                    let root = temp_root("fput-local-root");
                    root.0.clone()
                },
                64,
                DEFAULT_DELETE_WINDOW_DELAY,
                DEFAULT_PIN_TTL,
                FleetSharedMedia::asserted(),
            )
            .await
            .expect("the fleet shared-media store constructs");
            let owner = store.fleet().expect("fleet state").owner().to_vec();
            (store, owner)
        };

        // Success form: the pin row is durable, then the commit-rename
        // publishes — after the put, both are there, byte-exact.
        let content = vec![7u8; 200];
        let (key, pin) = store
            .put(Some(200), Cursor::new(&content[..]))
            .await
            .expect("the shared-media put publishes");
        let probe = independent(&site.dsn).await;
        let rows = instance_rows(&probe.pin_rows(&site.schema).await, &owner);
        assert_eq!(rows.len(), 1, "the pin row is durable");
        assert_eq!(rows[0].0, key.as_bytes());
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry is published");
        assert_eq!(len, 200);
        assert_eq!(blob.read_at(0, len).await.unwrap(), content);
        drop(pin);

        // The forced publish failure: pin-first means the row landed
        // before the publish could.
        let (mut store2, owner2) = {
            let store = Store::dual_tier_postgres_local_fleet(
                &site.dsn,
                PgOpts {
                    schema: site.schema.clone(),
                    application_name: site.session_tag(),
                    ..Default::default()
                },
                {
                    let root = temp_root("fput-local-fail");
                    root.0.clone()
                },
                64,
                DEFAULT_DELETE_WINDOW_DELAY,
                DEFAULT_PIN_TTL,
                FleetSharedMedia::asserted(),
            )
            .await
            .expect("second instance");
            let owner = store.fleet().expect("fleet state").owner().to_vec();
            (store, owner)
        };
        match &mut store2.large {
            Some(LargeEngine::Local(engine)) => engine.inject_fail_commit(),
            _ => panic!("a local large tier is required"),
        }
        let failed = b"content whose commit-rename is forced to fail, over the threshold".to_vec();
        assert!(
            failed.len() > 64,
            "the failed put must route to the large tier"
        );
        let err = store2
            .put(Some(failed.len() as u64), Cursor::new(&failed[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("injected"),
            "the engine's forced publish failure surfaces: {err}"
        );
        let key = key_for(&failed);
        let probe = independent(&site.dsn).await;
        let orphan = instance_rows(&probe.pin_rows(&site.schema).await, &owner2);
        assert_eq!(
            orphan.len(),
            1,
            "pin-first: the pin row committed before the failed publish \
             (the orphan the expiry machinery reaps)"
        );
        assert_eq!(orphan[0].0, key.as_bytes());
        assert!(
            !store2.has(&key.as_bytes()).await.unwrap(),
            "no committed file exists — the publish never happened"
        );
        assert_eq!(
            store2.gc().pin_count(&key),
            0,
            "the failed put's in-process refcount released with the guard"
        );
    }

    /// The shared-media ordering's pin-commit failure form (A1's
    /// catch-propagate-terminate on the row-only joint tx): a forced
    /// `pin_commit` failure (the kv engine's injection) precedes the
    /// publish — pin-first means **both** miss together, the stage
    /// discards, and nothing lands anywhere: no pin row, no kv row, no
    /// committed file, no stage file, no dirty pooled session (the
    /// clause's own rollback; a follow-up put round-trips clean).
    #[tokio::test]
    async fn fleet_shared_media_pin_commit_failure_leaves_neither_row_nor_publish() {
        let Some(site) = fresh_site("fput-local-pinfail").await else {
            skip_note();
            return;
        };
        let (store, _owner) = {
            let store = Store::dual_tier_postgres_local_fleet(
                &site.dsn,
                PgOpts {
                    schema: site.schema.clone(),
                    application_name: site.session_tag(),
                    ..Default::default()
                },
                temp_root("fput-local-pinfail-root").0.clone(),
                64,
                DEFAULT_DELETE_WINDOW_DELAY,
                DEFAULT_PIN_TTL,
                FleetSharedMedia::asserted(),
            )
            .await
            .expect("the fleet shared-media store constructs");
            let owner = store.fleet().expect("fleet state").owner().to_vec();
            (store, owner)
        };

        // The known- and unknown-length happy forms first: the
        // unknown-length staged form routes the fleet staged commit —
        // the same pin-first ordering, both entries byte-exact. (The
        // failing form runs on a second instance of the same schema:
        // the first instance's renewal guards hold its engine Arc, so
        // the injection needs a store whose engine is uniquely held.)
        let known = [5u8; 200];
        let (known_key, known_pin) = store
            .put(Some(known.len() as u64), Cursor::new(&known[..]))
            .await
            .expect("the shared-media known put publishes");
        let staged_form = [6u8; 1_500];
        let (staged_key, staged_pin) = store
            .put(None, Cursor::new(&staged_form[..]))
            .await
            .expect("the shared-media unknown put publishes through the fleet staged commit");
        for (key, want) in [(&known_key, &known[..]), (&staged_key, &staged_form[..])] {
            let (len, blob) = store
                .get(&key.as_bytes())
                .await
                .unwrap()
                .expect("the entry reads back");
            assert_eq!(blob.read_at(0, len).await.unwrap(), want);
        }
        drop((known_pin, staged_pin));

        let failing_store = Store::dual_tier_postgres_local_fleet(
            &site.dsn,
            PgOpts {
                schema: site.schema.clone(),
                application_name: site.session_tag(),
                ..Default::default()
            },
            temp_root("fput-local-pinfail-fail-root").0.clone(),
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
            DEFAULT_PIN_TTL,
            FleetSharedMedia::asserted(),
        )
        .await
        .expect("the second shared-media instance constructs");
        let mut store2 = failing_store;
        match &mut store2.kv {
            KvEngine::Postgres(engine) => match Arc::get_mut(engine) {
                Some(backend) => backend.inject_fail_joint_commit(),
                None => panic!("the store uniquely holds its own engine"),
            },
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("a postgres store is required"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("a postgres store is required"),
        }

        let failed =
            b"content whose pin-only commit is forced to fail, past the threshold".to_vec();
        assert!(failed.len() > 64, "the failed put routes to the large tier");
        let err = store2
            .put(Some(failed.len() as u64), Cursor::new(&failed[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("injected"),
            "the engine's forced pin-commit failure surfaces: {err}"
        );

        let key = key_for(&failed);
        let probe = independent(&site.dsn).await;
        assert_eq!(probe.kv_rows(&site.schema).await, vec![], "no kv row");
        let rows = probe.pin_rows(&site.schema).await;
        assert!(
            rows.iter()
                .all(|(row_key, _, _)| *row_key != key.as_bytes()),
            "the failed put committed no pin row of its own: {rows:?}"
        );
        assert_eq!(
            probe
                .idle_in_transaction_sessions(&site.db, &site.session_tag())
                .await,
            0,
            "no session returned to the pool holding a transaction (A1)"
        );
        assert!(
            !store2.has(&key.as_bytes()).await.unwrap(),
            "no committed file — the publish never happened"
        );
        assert_eq!(
            store2.gc().pin_count(&key),
            0,
            "the failed put's in-process refcount released with the guard"
        );
        assert!(store2.list().await.unwrap().is_empty(), "zero residue");

        match &mut store2.kv {
            KvEngine::Postgres(engine) => match Arc::get_mut(engine) {
                Some(backend) => backend.clear_fail_joint_commit(),
                None => panic!("the store uniquely holds its own engine"),
            },
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("a postgres store is required"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("a postgres store is required"),
        }
        let fresh = b"content after the failed put, clean".to_vec();
        let (key, pin) = store2
            .put(Some(fresh.len() as u64), Cursor::new(&fresh[..]))
            .await
            .expect("the follow-up put publishes");
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry reads back");
        assert_eq!(blob.read_at(0, len).await.unwrap(), fresh);
        drop(pin);
    }

    /// Under `fleet: false` the put paths are byte-unchanged: the
    /// ADR-005 in-process protocol verbatim — no pin rows are ever
    /// committed (the pin machinery is the in-process guard only), and
    /// the puts/gets behave exactly as before this task.
    #[tokio::test]
    async fn non_fleet_put_commits_no_pin_rows() {
        let Some(site) = fresh_site("fput-nonfleet").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_postgres_local(
            &site.dsn,
            PgOpts {
                schema: site.schema.clone(),
                application_name: site.session_tag(),
                ..Default::default()
            },
            {
                let root = temp_root("fput-nonfleet-root");
                root.0.clone()
            },
            64,
            DEFAULT_DELETE_WINDOW_DELAY,
        )
        .await
        .expect("the non-fleet store constructs");
        assert!(store.fleet().is_none(), "implicitly fleet: false");

        let small = b"a non-fleet kv put".to_vec();
        let (k_small, pin_one) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("the put publishes");
        let big = vec![5u8; 300];
        let (k_big, pin_two) = store
            .put(Some(300), Cursor::new(&big[..]))
            .await
            .expect("the staged put publishes, unchanged");

        for (key, want) in [(&k_small, &small[..]), (&k_big, &big[..])] {
            let (len, blob) = store
                .get(&key.as_bytes())
                .await
                .unwrap()
                .expect("the entry reads back");
            assert_eq!(blob.read_at(0, len).await.unwrap(), want);
        }

        let probe = independent(&site.dsn).await;
        assert_eq!(
            probe.pin_rows(&site.schema).await,
            vec![],
            "fleet: false pins nothing durable — the in-process guard only"
        );
        drop(pin_one);
        drop(pin_two);
    }
}

/// The wave-6 fleet put path's store-core tests for the consolidation
/// posture (features `postgres` + `pg-lo`): the A1 termination
/// discipline's mid-driver failure form through the public `put` (the
/// contribution returns with the held tx; the driver terminates — zero
/// residue, no dirty pooled session), and the one-tx round-trips
/// (both tiers, known- and unknown-length, byte-exact; the pool's pin
/// rows exactly one per member). Every test takes a fresh dedicated
/// database (large objects are database-level — the C7 lesson) with a
/// janitor removing it in every exit path. Env-driven
/// (`ALKBLOBS_PG_*`), skipping cleanly server-less.
#[cfg(all(test, feature = "postgres", feature = "pg-lo"))]
mod postgres_pglo_fleet_put_tests {
    use super::*;
    use crate::backend::Backend;
    use crate::pg_common::{qualified_table, test_support::DatabaseJanitor, DEFAULT_SCHEMA};
    use std::io::Cursor;
    use std::sync::atomic::{AtomicU64, Ordering};
    use tokio_postgres::NoTls;

    /// A liveness source answering nothing — the sweep gate's
    /// registration (the fleet pin rows compose behind it; ADR-012
    /// §1's mark composition).
    struct NoLive;

    impl LivenessSource for NoLive {
        fn live_keys(&self) -> Vec<Key> {
            Vec::new()
        }
        fn contains(&self, _key: &Key) -> bool {
            false
        }
    }

    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,
        db: 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!(
            "abtfputpglo_{}_{}_{: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),
            db: db.clone(),
            _janitor: DatabaseJanitor::new(&admin, &db),
        })
    }

    /// A short test pin TTL: the TTL/3 cadence observable within a
    /// test's runtime (≈1.3 s renewals, ≈4 s expiry horizon) — the
    /// widened load-tolerance posture (the kv fleet tests' constant
    /// family, mirrored here for the consolidation row's renewal arm).
    const TEST_TTL: Duration = Duration::from_secs(4);

    fn pages_for(len: usize) -> i64 {
        (len as i64 + 2047) / 2048
    }

    async fn sleep_of(d: Duration) {
        tokio::time::sleep(d).await;
    }

    async fn fleet_pglo_store(
        site: &TestDatabase,
        threshold: u64,
        ttl: Duration,
    ) -> (Store, Vec<u8>) {
        let store = Store::dual_tier_postgres_pglo_fleet(
            &site.dsn,
            PgOpts {
                application_name: site.db.clone(),
                ..Default::default()
            },
            &site.dsn,
            PgOpts {
                application_name: site.db.clone(),
                ..Default::default()
            },
            threshold,
            DEFAULT_DELETE_WINDOW_DELAY,
            ttl,
        )
        .await
        .expect("the fleet consolidation store constructs");
        let owner = store.fleet().expect("fleet state").owner().to_vec();
        (store, owner)
    }

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

    /// The independent connection — every probe runs outside the
    /// store's pool, over its own connection, seeing only durable
    /// state.
    struct Independent {
        client: tokio_postgres::Client,
        _conn: tokio::task::JoinHandle<()>,
    }

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

    impl Independent {
        /// The gc_pins rows in their stored encoding, in the
        /// `(key, owner)` sort order.
        async fn pin_rows(&self) -> Vec<(Vec<u8>, Vec<u8>, i64)> {
            let sql = format!(
                "SELECT key, owner, expiry FROM {} ORDER BY key, owner",
                qualified_table(DEFAULT_SCHEMA, "gc_pins")
            );
            let rows = self
                .client
                .query(sql.as_str(), &[])
                .await
                .expect("pin rows");
            rows.iter()
                .map(|row| {
                    (
                        row.get::<_, Vec<u8>>(0),
                        row.get::<_, Vec<u8>>(1),
                        row.get::<_, i64>(2),
                    )
                })
                .collect()
        }

        /// The kv engine's entry rows `(key, size)`.
        async fn kv_rows(&self) -> Vec<(Vec<u8>, i64)> {
            let sql = format!(
                "SELECT key, size FROM {} ORDER BY key",
                qualified_table(DEFAULT_SCHEMA, "blobs")
            );
            let rows = self.client.query(sql.as_str(), &[]).await.expect("kv rows");
            rows.iter()
                .map(|row| (row.get::<_, Vec<u8>>(0), row.get::<_, i64>(1)))
                .collect()
        }

        /// The LO catalog's page count (LOs are database-level, C7 —
        /// the pages every failure form's zero-residue assertion
        /// counts).
        async fn lo_pages(&self) -> i64 {
            self.client
                .query_one("SELECT count(*) FROM pg_largeobject", &[])
                .await
                .expect("LO catalog probe")
                .get::<_, i64>(0)
        }

        /// The manual expiry instrument (the abandonment shape the
        /// maintenance step reaps — no timing reliance).
        async fn expire_pin_rows(&self, key: &[u8]) {
            let expired = crate::pg_common::time_to_nanos(SystemTime::now()) - 1_000_000_000;
            let sql = format!(
                "UPDATE {} SET expiry = $1 WHERE key = $2",
                qualified_table(DEFAULT_SCHEMA, "gc_pins")
            );
            self.client
                .execute(sql.as_str(), &[&expired, &key])
                .await
                .expect("the manual expiry succeeds");
        }

        /// The pooled sessions sitting in a not-yet-committed
        /// transaction, scoped to this store family's own sessions by
        /// the instances' `application_name` tag — the dirty-session
        /// observable the A1 termination design's test asserts to zero
        /// (deadpool's `Fast` recycle never resets transaction state,
        /// PG-2). The `datname`-only scope is cross-test by
        /// construction (every concurrently-running test's transient
        /// idle-in-transaction session lands in whole-database counts);
        /// the tag makes the zero-asserts exact.
        async fn idle_in_transaction_sessions(&self, db: &str, tag: &str) -> i64 {
            self.client
                .query_one(
                    "SELECT count(*) FROM pg_stat_activity \
                     WHERE datname = $1 AND application_name = $2 \
                     AND state IS NOT NULL \
                     AND state LIKE 'idle in transaction%'",
                    &[&db, &tag],
                )
                .await
                .expect("pg_stat_activity probe")
                .get::<_, i64>(0)
        }
    }

    /// The store's pin rows for its own instance id — the filter
    /// keeping multi-instance assertions per-instance.
    fn instance_rows(rows: &[(Vec<u8>, Vec<u8>, i64)], owner: &[u8]) -> Vec<(Vec<u8>, i64)> {
        rows.iter()
            .filter(|(_, row_owner, _)| row_owner == owner)
            .map(|(key, _, expiry)| (key.clone(), *expiry))
            .collect()
    }

    /// The A1 termination discipline's mid-driver failure form through
    /// the public `put` (the consolidation posture): the tier's
    /// injected contribution failure — LO pages already staged into
    /// the held tx — is caught by the driver, which terminates the
    /// tx (the contribution never does) before propagating: zero
    /// residue everywhere (no kv row, no pin row, no companion row, no
    /// LO pages) and no dirty session returned to the pool (deadpool's
    /// `Fast` recycle never resets transaction state, PG-2; a follow-up
    /// put on the pool round-trips clean).
    #[tokio::test]
    async fn fleet_pglo_mid_driver_failure_terminates_with_zero_residue() {
        let Some(site) = fresh_db("fput-pglo-fail").await else {
            skip_note();
            return;
        };
        let (mut store, _owner) = fleet_pglo_store(&site, 64, DEFAULT_PIN_TTL).await;
        match &mut store.large {
            Some(LargeEngine::PgLo(engine)) => engine.inject_fail_commit(),
            _ => panic!("the consolidation posture's large tier is required"),
        }

        // ≥ two 512 KiB blocks flushed into the held tx before the
        // injected publish failure — real staged LO state for the
        // termination to take away.
        let content = vec![9u8; 1_500_000];
        let err = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("injected"),
            "the contribution's forced failure surfaces: {err}"
        );

        let probe = independent(&site.dsn).await;
        assert_eq!(probe.kv_rows().await, vec![], "no kv row");
        assert_eq!(probe.pin_rows().await, vec![], "no pin row");
        assert_eq!(
            probe.lo_pages().await,
            0,
            "the staged LO's pages died with the driver's termination"
        );
        assert_eq!(
            probe.idle_in_transaction_sessions(&site.db, &site.db).await,
            0,
            "no dirty pooled session (A1's catch-propagate-terminate)"
        );

        // The session the pool hands back is workable: a follow-up
        // put/get round-trips.
        let fresh = b"the pool's session came back clean".to_vec();
        let (key, pin) = store
            .put(Some(fresh.len() as u64), Cursor::new(&fresh[..]))
            .await
            .expect("the follow-up put publishes");
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry reads back");
        assert_eq!(blob.read_at(0, len).await.unwrap(), fresh);
        drop(pin);
    }

    /// The consolidation posture's round-trip through the public
    /// `put` (both tiers, one tx each): the sub-threshold puts ride
    /// the kv tier's joint form (known- and unknown-length alike),
    /// the over-threshold puts ride the pg-lo tier's joint form (the
    /// staged LO + companion row + pin row + commit inside the
    /// store-held tx), every entry reads back byte-exact, and the
    /// pool's pin rows are exactly the four members' — same owner,
    /// row-per-key.
    #[tokio::test]
    async fn fleet_pglo_consolidation_round_trips_through_the_public_put() {
        let Some(site) = fresh_db("fput-pglo-rt").await else {
            skip_note();
            return;
        };
        let (store, owner) = fleet_pglo_store(&site, 64, TEST_TTL).await;

        let small = b"a sub-threshold consolidation put".to_vec();
        let (k_small, guard_one) = store
            .put(Some(small.len() as u64), Cursor::new(&small[..]))
            .await
            .expect("the kv joint put publishes");
        let small_streamed = b"an unknown-length sub-threshold put".to_vec();
        let (k_kv_streamed, guard_two) = store
            .put(None, Cursor::new(&small_streamed[..]))
            .await
            .expect("the unknown-length kv joint put publishes");

        let big = vec![7u8; 1_500_000];
        let (k_big, guard_three) = store
            .put(Some(big.len() as u64), Cursor::new(&big[..]))
            .await
            .expect("the large joint put publishes");
        let big_streamed = vec![11u8; 300_000];
        let (k_large_streamed, guard_four) = store
            .put(None, Cursor::new(&big_streamed[..]))
            .await
            .expect("the unknown-length large joint put publishes");

        for (key, want) in [
            (&k_small, &small[..]),
            (&k_kv_streamed, &small_streamed[..]),
            (&k_big, &big[..]),
            (&k_large_streamed, &big_streamed[..]),
        ] {
            let (len, blob) = store
                .get(&key.as_bytes())
                .await
                .unwrap()
                .expect("the entry reads back");
            assert_eq!(
                blob.read_at(0, len).await.unwrap(),
                want,
                "byte-exact round-trip (the tier's one-tx form)"
            );
        }

        let probe = independent(&site.dsn).await;
        let kv_rows = probe.kv_rows().await;
        assert_eq!(
            kv_rows.len(),
            2,
            "exactly the two kv-tier entries (the large tier's one-tx form \
             carries no kv row)"
        );
        let rows = probe.pin_rows().await;
        let mine = instance_rows(&rows, &owner);
        assert_eq!(
            mine.len(),
            4,
            "exactly the four members' pin rows, one tx apiece"
        );
        for key in [&k_small, &k_kv_streamed, &k_big, &k_large_streamed] {
            assert!(
                mine.iter().any(|(k, _)| k == &key.as_bytes()),
                "each member's row carries its key"
            );
        }

        let lo = pglo_engine(&store);
        assert_eq!(lo.size(&k_big.as_bytes()).await.unwrap(), Some(1_500_000));
        assert_eq!(
            lo.size(&k_large_streamed.as_bytes()).await.unwrap(),
            Some(300_000),
            "the unknown-length staged form's derivation resolved exactly"
        );
        assert_eq!(
            probe.lo_pages().await,
            pages_for(1_500_000) + pages_for(300_000),
            "exactly the two entries' LO pages"
        );

        // The renewal duty rides the durable rows: the members'
        // expiries move within the bounded renewal window (the
        // load-tolerance posture — the arms poll instead of sampling
        // once at a fixed wall time), the guards still held by this
        // test's drops-never bindings.
        let deadline = std::time::Instant::now() + Duration::from_millis(3000);
        loop {
            let moved = instance_rows(&probe.pin_rows().await, &owner);
            if moved
                .iter()
                .zip(mine.iter())
                .all(|((_, expiry), (_, initial))| expiry > initial)
            {
                break;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "the guard's TTL/3 duty keeps the row moving forward"
            );
            sleep_of(Duration::from_millis(100)).await;
        }
        drop((guard_one, guard_two, guard_three, guard_four));
    }

    /// The git-object family put under a fleet declaration, on the
    /// consolidation form (ADR-018 §Decision 7): a loose form over the
    /// declared threshold routes the pg-lo tier's one-tx joint form —
    /// staged LO + companion row + pin row + commit inside the held tx
    /// — under the family's derivation (no blob hasher anywhere on the
    /// path), and the family key rides the sweep machinery with exact
    /// counts.
    #[tokio::test]
    async fn fleet_pglo_put_git_object_lands_through_the_consolidation_tx() {
        let Some(site) = fresh_db("fput-pglo-gitobj").await else {
            skip_note();
            return;
        };
        let store = Store::dual_tier_postgres_pglo_fleet(
            &site.dsn,
            PgOpts {
                application_name: site.db.clone(),
                ..Default::default()
            },
            &site.dsn,
            PgOpts {
                application_name: site.db.clone(),
                ..Default::default()
            },
            64,
            Duration::ZERO,
            TEST_TTL,
        )
        .await
        .expect("the fleet consolidation store constructs");
        let owner = store.fleet().expect("fleet state").owner().to_vec();
        let _ = store.register_liveness_source(crate::facade::LivenessRegistration::InProcess(
            Arc::new(NoLive) as Arc<dyn LivenessSource>,
        ));

        // A tree entry padded past the declared threshold (68 bytes of
        // body; the loose form 76 — the large tier's routing is the
        // pure length function, family puts ride it unchanged).
        let tree_body: Vec<u8> = {
            let mut t = format!("100644 {}\0", "0".repeat(40)).into_bytes();
            t.extend_from_slice(&[0u8; 20]);
            t
        };
        let loose = {
            let mut v = format!("tree {}\0", tree_body.len()).into_bytes();
            v.extend_from_slice(&tree_body);
            v
        };
        assert!(loose.len() > 64, "the body routes the large tier");
        let (key, pin) = store
            .put_git_object(
                Algorithm::GitObjectSha1,
                loose.len() as u64,
                Cursor::new(loose.clone()),
            )
            .await
            .expect("the fleet git-object consolidation put publishes");
        assert_eq!(
            key.digest().bytes().to_vec(),
            Algorithm::GitObjectSha1
                .derive(&loose, loose.len() as u64)
                .bytes()
                .to_vec(),
            "the key's digest is git's oid of the crafted tree (the family's derivation)"
        );

        // The one-tx consolidation: the LO, the pin row; no kv row.
        let probe = independent(&site.dsn).await;
        assert_eq!(
            probe.kv_rows().await.len(),
            0,
            "the large tier carries no kv row"
        );
        assert_eq!(
            probe.lo_pages().await,
            pages_for(loose.len()),
            "exactly the entry's LO pages"
        );
        let mine = instance_rows(&probe.pin_rows().await, &owner);
        assert_eq!(mine.len(), 1, "exactly this instance's pin row");
        assert_eq!(mine[0].0, key.as_bytes());

        // The stat semantic (ADR-018 §Decision 3) on the LO medium,
        // and the byte-exact round-trip through the family's
        // verification-on-get.
        let meta = store
            .stat(&key.as_bytes())
            .await
            .unwrap()
            .expect("stat sees the family entry");
        assert_eq!(meta.len, loose.len() as u64);
        assert_eq!(meta.algorithm, Algorithm::GitObjectSha1);
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry reads back");
        assert_eq!(len, loose.len() as u64);
        assert_eq!(blob.read_at(0, len).await.unwrap(), loose);

        // The exact-count sweep outcome: held pins protect; the
        // expired shape stages and deletes exactly once (the LO pages
        // die with the entry).
        let report = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(report.aborted, None);
        assert_eq!(
            (
                report.candidates_staged,
                report.deleted,
                report.cancelled_by_arbitration
            ),
            (0, 0, 0),
            "the held pin protects the family entry exactly as a blob's does"
        );
        drop(pin);
        probe.expire_pin_rows(&key.as_bytes()).await;
        let staged = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(staged.aborted, None);
        assert_eq!(
            staged.candidates_staged, 1,
            "the unprotected family entry staged exactly once"
        );
        let executed = store.sweep().await.expect("sweeps are never errors");
        assert_eq!(executed.aborted, None);
        assert_eq!(
            (executed.deleted, executed.cancelled_by_arbitration),
            (1, 0),
            "the executing sweep deleted exactly the one family entry"
        );
        assert_eq!(
            probe.lo_pages().await,
            0,
            "the LO's pages died with the entry"
        );
    }

    /// A stream that delivers `fail_after` bytes then errors
    /// mid-stream — the fleet path's failing reader (the same double
    /// the mem suite's rule-3 tests run, local to this section).
    struct Failing {
        data: Vec<u8>,
        cursor: usize,
        fail_after: usize,
    }

    impl Failing {
        fn new(data: &[u8], fail_after: usize) -> Failing {
            Failing {
                data: data.to_vec(),
                cursor: 0,
                fail_after,
            }
        }
    }

    impl AsyncRead for Failing {
        fn poll_read(
            mut self: std::pin::Pin<&mut Self>,
            _cx: &mut Context<'_>,
            buf: &mut ReadBuf<'_>,
        ) -> Poll<std::io::Result<()>> {
            let message = "stream failure (test fixture)";
            if self.cursor >= self.fail_after {
                return Poll::Ready(Err(std::io::Error::other(message)));
            }
            let wanted = (self.data.len() - self.cursor)
                .min(self.fail_after - self.cursor)
                .min(buf.remaining());
            if wanted == 0 {
                return Poll::Ready(Err(std::io::Error::other(message)));
            }
            buf.put_slice(&self.data[self.cursor..self.cursor + wanted]);
            self.cursor += wanted;
            Poll::Ready(Ok(()))
        }
    }

    use std::task::{Context, Poll};
    use tokio::io::ReadBuf;

    async fn assert_zero_residue(site: &TestDatabase, err: &Error, what: &str) {
        assert!(
            err.to_string().contains(what),
            "the {what} form surfaces: {err}"
        );
        let probe = independent(&site.dsn).await;
        assert_eq!(probe.kv_rows().await, vec![], "no kv row ({what})");
        assert_eq!(probe.pin_rows().await, vec![], "no pin row ({what})");
        assert_eq!(probe.lo_pages().await, 0, "no staged LO pages ({what})");
        assert_eq!(
            probe.idle_in_transaction_sessions(&site.db, &site.db).await,
            0,
            "no dirty pooled session ({what})"
        );
    }

    /// The failing reader through the fleet pg-lo put (the D-2 sweep's
    /// second region — the put paths' mid-read error arms on the
    /// consolidation posture, driven as five forms): a stream nobody
    /// may read, a stream dying mid-staging inside the held tx, an
    /// under-received clean EOF, an over-fed stream, an unknown-length
    /// stream dying before the threshold (pre-tx), and one dying after
    /// the overflow (mid-tx) — every one surfaces its typed rejection,
    /// terminates the held tx, and leaves zero residue everywhere (no
    /// kv row, no pin row, no LO pages, no dirty session). The pool's
    /// handed-back session round-trips a happy put at the end.
    #[tokio::test]
    async fn the_failing_reader_through_the_fleet_pglo_put_terminates_with_zero_residue() {
        let Some(site) = fresh_db("fput-pglo-reader").await else {
            skip_note();
            return;
        };
        let (store, _owner) = fleet_pglo_store(&site, 64, DEFAULT_PIN_TTL).await;
        let body = vec![9u8; 1_500];

        // The unreadable form (the read nobody may take): the known-
        // length put's stream errs before the first byte flows.
        let err = store
            .put(Some(1_500_000), Failing::new(&body, 0))
            .await
            .unwrap_err();
        assert_zero_residue(&site, &err, "stream failure").await;

        // The mid-staging failure: bytes already staged into the held
        // tx die with its termination.
        let err = store.put(None, Failing::new(&body, 200)).await.unwrap_err();
        assert_zero_residue(&site, &err, "stream failure").await;

        // The unknown-length form's pre-threshold death: the stream
        // errs before any joint tx exists.
        let err = store.put(None, Failing::new(&body, 10)).await.unwrap_err();
        assert_zero_residue(&site, &err, "stream failure").await;

        // The under-received clean EOF: the declared length the stream
        // never reaches — the derivation's Verification rejection.
        let short: &[u8] = &body[..37];
        let err = store
            .put(Some(1_500_000), Cursor::new(short))
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::Verification),
            "the under-received put is the Verification rejection: {err}"
        );
        assert_zero_residue(&site, &err, "verification failed").await;

        // The over-fed stream: the length routes large (over the
        // threshold) and the first read already overshoots it — the
        // over-length check rejects before anything lands.
        let err = store
            .put(
                Some((body.len() - body.len() / 10) as u64),
                Cursor::new(&body[..]),
            )
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::Verification),
            "the over-fed put is the Verification rejection: {err}"
        );
        assert_zero_residue(&site, &err, "verification failed").await;

        // The pool's handed-back session round-trips a happy put.
        let fresh = b"the pool came back clean after the failure round".to_vec();
        let (key, pin) = store
            .put(Some(fresh.len() as u64), Cursor::new(&fresh[..]))
            .await
            .expect("the follow-up put publishes");
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry reads back");
        assert_eq!(blob.read_at(0, len).await.unwrap(), fresh);
        drop(pin);
    }

    /// The fleet joint publish's pin-commit failure form through the
    /// public put (the consolidation posture): the driver completes —
    /// the LO staged inside the held tx with the CAS contribution —
    /// then the forced `pin_commit` failure (the kv engine's
    /// injection) propagates: the tx rolls back whole (the entry, the
    /// pin row, and the LO pages all die together; the joint form's
    /// one-tx protection), and no session returns to the pool dirty.
    #[tokio::test]
    async fn the_fleet_joint_publish_pin_commit_failure_rolls_everything_back() {
        let Some(site) = fresh_db("fput-pglo-pinfail").await else {
            skip_note();
            return;
        };
        let (mut store, _owner) = fleet_pglo_store(&site, 64, DEFAULT_PIN_TTL).await;
        match &mut store.kv {
            KvEngine::Postgres(engine) => match Arc::get_mut(engine) {
                Some(backend) => backend.inject_fail_joint_commit(),
                None => panic!("the store uniquely holds its own engine"),
            },
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("a postgres store is required"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("a postgres store is required"),
        }

        let content = vec![7u8; 1_500_000];
        let err = store
            .put(Some(content.len() as u64), Cursor::new(&content[..]))
            .await
            .unwrap_err();
        assert!(
            err.to_string().contains("injected"),
            "the kv engine's forced joint-commit failure surfaces: {err}"
        );

        let probe = independent(&site.dsn).await;
        assert_eq!(probe.kv_rows().await, vec![], "no kv row");
        assert_eq!(probe.pin_rows().await, vec![], "no pin row");
        assert_eq!(
            probe.lo_pages().await,
            0,
            "the staged LO's pages rolled back with the tx (the joint form)"
        );
        assert_eq!(
            probe.idle_in_transaction_sessions(&site.db, &site.db).await,
            0,
            "no dirty pooled session"
        );
        match &mut store.kv {
            KvEngine::Postgres(engine) => match Arc::get_mut(engine) {
                Some(backend) => backend.clear_fail_joint_commit(),
                None => panic!("the store uniquely holds its own engine"),
            },
            #[cfg(feature = "mem")]
            KvEngine::Mem(_) => panic!("a postgres store is required"),
            #[cfg(feature = "sqlite")]
            KvEngine::Sqlite(_) => panic!("a postgres store is required"),
        }
        let fresh = b"the pool came back clean after the publish failure".to_vec();
        let (key, pin) = store
            .put(Some(fresh.len() as u64), Cursor::new(&fresh[..]))
            .await
            .expect("the follow-up put publishes");
        let (len, blob) = store
            .get(&key.as_bytes())
            .await
            .unwrap()
            .expect("the entry reads back");
        assert_eq!(blob.read_at(0, len).await.unwrap(), fresh);
        drop(pin);
    }
}