alkblobs 0.2.2

//! The held-transaction staging cursor's bodies (and the fleet joint
//! form): appends aggregate into [`LOWRITE_CHUNK`] `lowrite` statements;
//! commit publishes (row insert + COMMIT, C6's zero-orphan shape); every
//! other path — explicit discard, a failed commit, the `Drop`
//! cancellation story — converges on ROLLBACK with zero residue. The
//! joint form (feature `postgres`) re-homes the same mechanics onto the
//! store-held joint tx — it opens nothing, terminates nothing, never
//! commits or rolls back (amendment A1's catch-propagate-terminate; the
//! F-2 anchor).

use std::time::SystemTime;

use deadpool_postgres::Object;

use crate::backend::{clamped, ReadCursor, StagingCursor};
use crate::pg_common::{detached_batch_execute, err, pg_err, time_to_nanos};
use crate::Error;

#[cfg(feature = "postgres")]
use crate::postgres::PgJointTx;

use super::windows::read_windows;
#[cfg(any(test, feature = "postgres"))]
use super::PgLoBackend;
#[cfg(feature = "postgres")]
use super::INV_WRITE;
use super::{PgLoStaging, LOWRITE_CHUNK};

/// `lo_create` — a fresh LO oid inside the caller's tx.
pub(super) async fn lo_create(client: &Object) -> Result<u32, Error> {
    let stmt = client
        .prepare_cached("SELECT lo_create(0) AS loid")
        .await
        .map_err(err)?;
    let row = client.query_one(&stmt, &[]).await.map_err(pg_err)?;
    row.try_get(0).map_err(pg_err)
}

/// `lo_open` — the tx-scoped descriptor for in-tx `lowrite` (reads never
/// hold one; the shipped get shape is descriptorless windows).
pub(super) async fn lo_open(client: &Object, loid: u32, mode: i32) -> Result<i32, Error> {
    let stmt = client
        .prepare_cached("SELECT lo_open($1, $2) AS fd")
        .await
        .map_err(err)?;
    let row = client
        .query_one(&stmt, &[&loid, &mode])
        .await
        .map_err(pg_err)?;
    row.try_get(0).map_err(pg_err)
}

/// Chunked `lowrite` — statements of [`LOWRITE_CHUNK`] (the C2 knee);
/// rides the descriptor's implicit sequential position (no repositioning,
/// the 64-bit-variants discipline). Fails on a negative or short count.
pub(super) async fn lowrite_all(client: &Object, fd: i32, mut data: &[u8]) -> Result<(), Error> {
    if data.is_empty() {
        return Ok(());
    }
    let stmt = client
        .prepare_cached("SELECT lowrite($1, $2) AS n")
        .await
        .map_err(err)?;
    while !data.is_empty() {
        let take = data.len().min(LOWRITE_CHUNK);
        let chunk = &data[..take];
        let n: i32 = client
            .query_one(&stmt, &[&fd, &chunk])
            .await
            .map_err(pg_err)?
            .try_get(0)
            .map_err(pg_err)?;
        if n < 0 || (n as usize) != take {
            return Err(Error::Io(format!(
                "postgres: lowrite wrote {n} of {take} staged bytes"
            )));
        }
        data = &data[take..];
    }
    Ok(())
}

/// The fleet joint form's handout (feature `postgres`, ADR-012 §4's
/// one-tx ruling): hand a [`PgLoJointStaging`] cursor against the
/// store-held joint tx — the consolidation posture's one shared
/// instance is what makes the type shareable. `lo_create` +
/// `lo_open(INV_WRITE)` run inside the held checkout: the LO is born
/// inside the store's transaction, and every failure or the driver's
/// termination takes the pages with it (C6's structural zero-orphan
/// story). The cursor owns no checkout and no transaction — the store
/// core holds both from its `begin_joint_tx` to its `pin_commit` (the
/// ownership rule, ADR-010 §2; ADR-012 §4's one-tx ruling) — so the
/// handout opens nothing and terminates nothing: a handout error
/// returns with the held tx exactly as it arrived (open, uncommitted)
/// for the driver to finish.
#[cfg(feature = "postgres")]
impl PgLoBackend {
    /// Hand out the joint staging cursor against the held joint tx.
    pub(crate) async fn put_joint_staged<'a>(
        &self,
        tx: &'a mut PgJointTx,
    ) -> Result<PgLoJointStaging<'a>, Error> {
        let loid = lo_create(tx.client()?).await?;
        let fd = lo_open(tx.client()?, loid, INV_WRITE).await?;
        Ok(PgLoJointStaging {
            tx,
            loid,
            fd,
            table: self.table.clone(),
            flushed_len: 0,
            pending: Vec::new(),
            #[cfg(test)]
            fail_commit: self.fail_commit,
        })
    }
}

impl PgLoStaging {
    /// The live checkout — always `Some` on a live cursor; the `None`
    /// arm is the defensively-typed consumed state, never a panic.
    fn client(&self) -> Result<&Object, Error> {
        self.client
            .as_ref()
            .ok_or_else(|| Error::Io("pg-lo: staging cursor already consumed".to_owned()))
    }

    /// Flush full [`LOWRITE_CHUNK`] blocks out of the aggregation buffer
    /// (the store core's 64 KiB put-loop windows aggregate here into the
    /// C2 knee's statement size). On failure the buffer is intact and
    /// `flushed_len` unclaimed — the caller converges on ROLLBACK.
    async fn flush_blocks(&mut self) -> Result<(), Error> {
        while self.pending.len() >= LOWRITE_CHUNK {
            let block = self.pending[..LOWRITE_CHUNK].to_vec();
            lowrite_all(self.client()?, self.fd, &block).await?;
            self.pending.drain(..LOWRITE_CHUNK);
            self.flushed_len += LOWRITE_CHUNK as u64;
        }
        Ok(())
    }

    /// Flush the aggregation tail (called by publish over the taken
    /// checkout; the tail is < [`LOWRITE_CHUNK`]).
    async fn flush_tail(&mut self, client: &Object) -> Result<(), Error> {
        if self.pending.is_empty() {
            return Ok(());
        }
        let tail = std::mem::take(&mut self.pending);
        lowrite_all(client, self.fd, &tail).await?;
        self.flushed_len += tail.len() as u64;
        Ok(())
    }

    /// The one ROLLBACK convergence point: every non-publish path runs
    /// this — the staged LO's pages vanish with the tx, zero residue.
    async fn rollback(mut self) {
        if let Some(client) = self.client.take() {
            let _ = client.batch_execute("ROLLBACK").await;
        }
    }
}

impl ReadCursor for PgLoStaging {
    fn len(&self) -> u64 {
        self.flushed_len + self.pending.len() as u64
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(offset, len, self.len() as usize);
        if start == end {
            return Ok(Vec::new());
        }
        let client = self.client()?;
        let lo_end = self.flushed_len.min(end as u64) as usize;
        let mut out = Vec::with_capacity(end - start);
        if (start as u64) < self.flushed_len {
            out.extend(read_windows(client, self.loid, start as u64, lo_end as u64).await?);
        }
        if end > lo_end {
            let pending_start = (start.max(lo_end)) - lo_end;
            out.extend_from_slice(&self.pending[pending_start..(end - lo_end)]);
        }
        Ok(out)
    }
}

impl StagingCursor for PgLoStaging {
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        if chunk.is_empty() {
            return Ok(());
        }
        self.pending.extend_from_slice(chunk);
        self.flush_blocks().await
    }

    /// Publish: flush the tail, CAS existence against the companion
    /// table (a re-put of an existing key ROLLBACKs the staged LO away
    /// and returns `Ok` — first-writer-wins dedup, never an overwrite),
    /// insert the companion row, COMMIT. Every failure path — including
    /// the injected one — converges on `rollback` (self-discard; the
    /// consuming call took the cursor).
    async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
        #[cfg(test)]
        if self.fail_commit {
            self.rollback().await;
            return Err(Error::Io(
                "engine commit failure (pg-lo, injected)".to_owned(),
            ));
        }
        let client = match self.client.take() {
            Some(client) => client,
            None => {
                return Err(Error::Io(
                    "pg-lo: staging cursor already consumed".to_owned(),
                ))
            }
        };
        let result = self.publish(&client, key).await;
        match result {
            Ok(true) => match client.batch_execute("COMMIT").await {
                Ok(()) => Ok(()),
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    Err(pg_err(e))
                }
            },
            Ok(false) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Ok(())
            }
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(e)
            }
        }
    }

    /// Discard the stage: ROLLBACK — the staged LO's pages vanish with
    /// the tx, no row existed to remove. Zero residue.
    async fn discard(self) {
        self.rollback().await;
    }
}

impl PgLoStaging {
    /// The publish body (over the taken checkout): tail flush, CAS
    /// existence against the companion table (a re-put returns
    /// `Ok(false)` — first-writer-wins dedup, commit ROLLBACKs the
    /// staged LO away, never an overwrite), then the companion-row
    /// insert (`Ok(true)`: commit proceeds). Errors bubble to
    /// [`StagingCursor::commit`]'s ROLLBACK convergence point.
    async fn publish(&mut self, client: &Object, key: &[u8]) -> Result<bool, Error> {
        self.flush_tail(client).await?;
        let exists_sql = format!("SELECT 1 FROM {} WHERE key = $1", self.table);
        let exists_stmt = client.prepare_cached(&exists_sql).await.map_err(err)?;
        let present = client
            .query_opt(&exists_stmt, &[&key])
            .await
            .map_err(pg_err)?
            .is_some();
        if present {
            return Ok(false);
        }
        let size = self.flushed_len as i64;
        let committed_at = time_to_nanos(SystemTime::now());
        let insert_sql = format!(
            "INSERT INTO {} (key, loid, size, committed_at) VALUES ($1, $2, $3, $4)",
            self.table
        );
        let insert_stmt = client.prepare_cached(&insert_sql).await.map_err(err)?;
        client
            .execute(&insert_stmt, &[&key, &self.loid, &size, &committed_at])
            .await
            .map_err(pg_err)?;
        Ok(true)
    }
}

impl Drop for PgLoStaging {
    /// Best-effort residue cleanup for the cancellation story: a cursor
    /// dropped without reaching `commit` or `discard` holds an open tx
    /// on a pooled session — hand it to
    /// `crate::pg_common::detached_batch_execute` (the shared
    /// spawned-or-thread shape; `PgJointTx`'s cancellation guard
    /// aliases it) so the checkout never returns dirty; the recorded
    /// pitfall stands (deadpool's `Fast` recycle never resets
    /// transaction state). Cursors reaching `commit`/`discard` have
    /// already taken the checkout; this is a no-op on those paths.
    fn drop(&mut self) {
        if let Some(client) = self.client.take() {
            detached_batch_execute(client, "ROLLBACK".to_owned());
        }
    }
}

/// The joint staging cursor — the staged put re-homed onto the
/// store-held joint tx (feature `postgres`, ADR-012 §4's one-tx
/// ruling; the named design point recorded in the module doc: the
/// joint staging-cursor variant, the large tier's put contract carried
/// into the consolidation posture). It borrows the held checkout —
/// everything it issues (`lowrite` statements, `lo_get` windows, the
/// CAS probe, the companion insert or the re-put `lo_unlink`) rides
/// the store's transaction; it holds no checkout of its own, opens
/// nothing, and terminates nothing: its error paths return the tx the
/// way they found it (open, uncommitted) for the driver to finish
/// (amendment A1's catch-propagate-terminate — the contribution never
/// commits, never rolls back the store's tx unilaterally). Dropping
/// the cursor without [`PgLoJointStaging::publish`] is likewise a
/// no-op on the tx — the driver owns termination on every path, and
/// the dropped cursor's staged pages die with that termination.
#[cfg(feature = "postgres")]
pub(crate) struct PgLoJointStaging<'a> {
    /// The store-held joint tx this contribution rides (borrowed, never
    /// owned — the ownership rule).
    tx: &'a mut PgJointTx,
    loid: u32,
    fd: i32,
    table: String,
    /// Bytes flushed to the LO via `lowrite` — the read seam's LO part
    /// boundary; the aggregation buffer holds everything above it.
    flushed_len: u64,
    pending: Vec<u8>,
    /// Test-only mid-contribution failure injection (after the tail
    /// flush, before the CAS probe), standing in for the media failure;
    /// mirrors the engine-level knob so the joint form's failure
    /// discipline is exercised against the real engine.
    #[cfg(test)]
    fail_commit: bool,
}

#[cfg(feature = "postgres")]
impl PgLoJointStaging<'_> {
    /// Flush full [`LOWRITE_CHUNK`] blocks out of the aggregation buffer
    /// (the store core's 64 KiB put-loop windows aggregate here into the
    /// C2 knee's statement size — the engine-level staged put's shape).
    /// On failure the buffer is intact and `flushed_len` unclaimed —
    /// the tx stays as the contribution found it.
    async fn flush_blocks(&mut self) -> Result<(), Error> {
        while self.pending.len() >= LOWRITE_CHUNK {
            let block = self.pending[..LOWRITE_CHUNK].to_vec();
            lowrite_all(self.tx.client()?, self.fd, &block).await?;
            self.pending.drain(..LOWRITE_CHUNK);
            self.flushed_len += LOWRITE_CHUNK as u64;
        }
        Ok(())
    }

    /// Flush the aggregation tail (< [`LOWRITE_CHUNK`]) — publish's
    /// first mechanic, over the held checkout.
    async fn flush_tail(&mut self) -> Result<(), Error> {
        if self.pending.is_empty() {
            return Ok(());
        }
        let tail = std::mem::take(&mut self.pending);
        lowrite_all(self.tx.client()?, self.fd, &tail).await?;
        self.flushed_len += tail.len() as u64;
        Ok(())
    }

    /// Append a chunk to the contribution — the staged put's
    /// aggregation shape (the store core's put loop hands 64 KiB
    /// windows; they aggregate into [`LOWRITE_CHUNK`] `lowrite`
    /// statements inside the held tx).
    pub(crate) async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        if chunk.is_empty() {
            return Ok(());
        }
        self.pending.extend_from_slice(chunk);
        self.flush_blocks().await
    }

    /// Publish the contribution into the held tx: the tail's
    /// aggregation flush, then the CAS existence check against the
    /// companion table (the key resolves at publish — the staged put's
    /// named design point: the derivation completes at stream end), and
    /// either
    ///
    /// - the companion-row insert — `Ok(true)`, the fresh form — or
    /// - the **re-put form**: the staged LO unlinked *inside the
    ///   store-held tx*, `Ok(false)` — an **empty contribution**: the
    ///   committed tx carries nothing of the losing put (zero residue),
    ///   and the driver's `pin_commit` then upsert-renews the pin row.
    ///
    /// A concurrent CAS miss surfaces the companion key's unique
    /// violation as an error (the landed puts' posture: the check is
    /// the fast path, the primary key is the hard dedup guarantee).
    /// Every failure path — including the injected one — returns with
    /// the tx exactly as the contribution found it: open, uncommitted,
    /// un-rolled-back (the driver owns termination; no partial LO state
    /// outlives its rollback — transactional LO creation, C6).
    pub(crate) async fn publish(mut self, key: &[u8]) -> Result<bool, Error> {
        #[cfg(test)]
        if self.fail_commit {
            return Err(Error::Io(
                "engine joint commit failure (pg-lo, injected)".to_owned(),
            ));
        }
        self.flush_tail().await?;
        let client = self.tx.client()?;
        let exists_sql = format!("SELECT 1 FROM {} WHERE key = $1", self.table);
        let exists_stmt = client.prepare_cached(&exists_sql).await.map_err(err)?;
        let present = client
            .query_opt(&exists_stmt, &[&key])
            .await
            .map_err(pg_err)?
            .is_some();
        if present {
            let unlink_stmt = client
                .prepare_cached("SELECT lo_unlink($1)")
                .await
                .map_err(err)?;
            client
                .query_one(&unlink_stmt, &[&self.loid])
                .await
                .map_err(pg_err)?;
            return Ok(false);
        }
        let size = self.flushed_len as i64;
        let committed_at = time_to_nanos(SystemTime::now());
        let insert_sql = format!(
            "INSERT INTO {} (key, loid, size, committed_at) VALUES ($1, $2, $3, $4)",
            self.table
        );
        let insert_stmt = client.prepare_cached(&insert_sql).await.map_err(err)?;
        client
            .execute(&insert_stmt, &[&key, &self.loid, &size, &committed_at])
            .await
            .map_err(pg_err)?;
        Ok(true)
    }
}

#[cfg(feature = "postgres")]
impl ReadCursor for PgLoJointStaging<'_> {
    fn len(&self) -> u64 {
        self.flushed_len + self.pending.len() as u64
    }

    /// The staged artifact's read seam, inside the store-held tx
    /// (read-your-own-writes: the LO's pages are visible to `lo_get`
    /// from the tx that wrote them; reads above the flushed boundary
    /// serve from the aggregation buffer) — the derivation pass's
    /// instrument, the staged put's shape carried.
    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(offset, len, self.len() as usize);
        if start == end {
            return Ok(Vec::new());
        }
        let client = self.tx.client()?;
        let lo_end = self.flushed_len.min(end as u64) as usize;
        let mut out = Vec::with_capacity(end - start);
        if (start as u64) < self.flushed_len {
            out.extend(read_windows(client, self.loid, start as u64, lo_end as u64).await?);
        }
        if end > lo_end {
            let pending_start = (start.max(lo_end)) - lo_end;
            out.extend_from_slice(&self.pending[pending_start..(end - lo_end)]);
        }
        Ok(out)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::backend::Backend;

    use crate::pg_lo::testing::{
        catalog_rows, companion_loids_sorted, harness, pages_for, pattern, staged, stored_value,
        CONTENT,
    };

    /// The staged put's happy path: 64 KiB appends (the store core's
    /// put-loop window size) aggregate into 512 KiB `lowrite` statements
    /// (the C2 knee), the verification pass reads the staged artifact
    /// back as a whole through the read seam — including reads crossing
    /// the flushed/pending aggregation boundary — commit publishes, and
    /// the committed entry round-trips through get's cursor.
    #[tokio::test]
    async fn staged_put_round_trips_through_the_read_seam() {
        let Some(backend) = harness("staged").await else {
            return;
        };
        let content = pattern(1_500_000);
        let key = b"\x01staged-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        for chunk in content.chunks(64 * 1024) {
            staging.append(chunk).await.unwrap();
        }

        assert_eq!(
            staging.len() as usize,
            content.len(),
            "staged length is the appended length"
        );
        let (bytes, digest) = staging
            .read_range(0, staging.len())
            .await
            .expect("staged bytes readable");
        assert_eq!(bytes, content, "staged artifact readable as a whole");
        assert_eq!(digest, crate::backend::slice_digest(&content));

        let boundary = LOWRITE_CHUNK as u64;
        assert!(
            staging.len() > boundary,
            "the staged content crossed the first aggregation flush"
        );
        let crossed = staging
            .read_at(boundary - 64, 256)
            .await
            .expect("boundary-crossing staged read");
        assert_eq!(
            crossed,
            content[(LOWRITE_CHUNK - 64)..(LOWRITE_CHUNK + 192)],
            "a staged read crossing the flushed/pending boundary is byte-exact \
             (LO windows over committed-into-tx pages + the in-memory tail)"
        );

        staging.commit(&key).await.expect("commit publishes");
        assert!(backend.has(&key).await.unwrap(), "entry published");

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let (bytes, digest) = cursor
            .read_range(0, cursor.len())
            .await
            .expect("committed read");
        assert_eq!(bytes, content);
        assert_eq!(digest, crate::backend::slice_digest(&content));

        backend
            .put(&key, b"overwritten")
            .await
            .expect("whole-value re-put lands as dedup");
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(&content[..]),
            "whole-value re-put does not overwrite"
        );
        staged(&backend, &key, b"overwritten-by-stage")
            .await
            .expect("staged re-commit is dedup, not failure");
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(&content[..]),
            "staged re-commit does not overwrite"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            pages_for(content.len()),
            "the dedup re-puts created no orphan LO pages \
             (a re-put never births an LO)"
        );
    }

    /// Stage hygiene, discard form: every non-commit path converges on
    /// discard with zero residue — no companion row, no LO pages.
    #[tokio::test]
    async fn staged_discard_leaves_zero_residue() {
        let Some(backend) = harness("discard").await else {
            return;
        };
        let key = b"\x01discard-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"residue").await.unwrap();
        staging.discard().await;

        assert!(
            !backend.has(&key).await.unwrap(),
            "discarded stage published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the pool holds exactly zero entries after discard"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            0,
            "discard left zero LO pages (the tx rolled back)"
        );
    }

    /// Stage hygiene, failure form: a failed commit is already
    /// self-discarded — zero residue, entry unpublished, no post-failure
    /// discard left to drive (the consuming call took the cursor). The
    /// injection stands in for the media failure.
    #[tokio::test]
    async fn commit_failure_self_discards() {
        let Some(backend) = harness("fail").await else {
            return;
        };
        let failing = PgLoBackend {
            pool: backend.pool.clone(),
            table: backend.table.clone(),
            window_table: backend.window_table.clone(),
            schema: backend.schema.clone(),
            orphan_reap_permitted: false,
            #[cfg(test)]
            fail_commit: true,
            #[cfg(test)]
            fail_window_stage: false,
        };
        let key = b"\x01failure-key".to_vec();

        let mut staging = failing.put_staged().await.unwrap();
        staging.append(CONTENT).await.unwrap();
        let err = staging.commit(&key).await.unwrap_err();
        assert!(matches!(err, Error::Io(_)), "engine media failure");

        assert!(
            !backend.has(&key).await.unwrap(),
            "failed commit published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "zero residue in the companion table itself"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            0,
            "zero residue in the LO catalog itself"
        );
    }

    /// The engine's own cancellation story: a staging cursor dropped
    /// without commit or discard leaves no residue (the Drop body's
    /// spawned ROLLBACK — the pool never receives a dirty checkout).
    #[tokio::test]
    async fn drop_cancellation_leaves_zero_residue() {
        let Some(backend) = harness("drop").await else {
            return;
        };
        let key = b"\x01cancelled-key".to_vec();

        {
            let mut staging = backend.put_staged().await.unwrap();
            staging.append(b"never published").await.unwrap();
            // Dropped without commit or discard.
        }
        tokio::time::sleep(std::time::Duration::from_millis(300)).await;

        assert!(
            !backend.has(&key).await.unwrap(),
            "cancelled stage published nothing"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the pool holds exactly zero entries after cancellation"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            0,
            "the spawned ROLLBACK removed the staged LO's pages"
        );
    }

    /// Exact-count stage hygiene (store-api.md invariant 3): N failed
    /// stages — over all three forms, explicit discard, failed commit,
    /// and Drop cancellation — leave zero companion rows and zero LO
    /// pages, and N successful stages leave exactly N entries.
    #[tokio::test]
    async fn exact_count_hygiene_forms_leave_zero() {
        const N: usize = 3;
        let Some(backend) = harness("exact").await else {
            return;
        };
        for i in 0..N {
            let content = format!("discard-{i}");
            let mut staging = backend.put_staged().await.unwrap();
            staging.append(content.as_bytes()).await.unwrap();
            staging.discard().await;

            let content = format!("commit-failure-{i}");
            let key = format!("failure-key-{i}").into_bytes();
            let failing = PgLoBackend {
                pool: backend.pool.clone(),
                table: backend.table.clone(),
                window_table: backend.window_table.clone(),
                schema: backend.schema.clone(),
                orphan_reap_permitted: false,
                #[cfg(test)]
                fail_commit: true,
                #[cfg(test)]
                fail_window_stage: false,
            };
            let mut staging = failing.put_staged().await.unwrap();
            staging.append(content.as_bytes()).await.unwrap();
            staging.commit(&key).await.unwrap_err();

            let content = format!("cancelled-{i}");
            {
                let mut staging = backend.put_staged().await.unwrap();
                staging.append(content.as_bytes()).await.unwrap();
                drop(staging);
            }
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
        }

        assert!(
            backend.list().await.unwrap().is_empty(),
            "{} failed stages left zero companion rows",
            3 * N
        );
        assert_eq!(
            catalog_rows(&backend).await,
            0,
            "{} failed stages left zero LO pages",
            3 * N
        );

        for i in 0..N {
            let content = format!("success-{i}");
            let key = format!("success-key-{i}").into_bytes();
            staged(&backend, &key, content.as_bytes())
                .await
                .expect("commit publishes");
        }
        let listed = backend.list().await.unwrap();
        assert_eq!(
            listed.len(),
            N,
            "N successful stages leave exactly N entries"
        );
        for key in &listed {
            assert!(backend.has(key).await.unwrap(), "every entry is real");
        }
        assert_eq!(
            companion_loids_sorted(&backend).await.len(),
            N,
            "every entry owns an LO"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            pages_for(CONTENT.len()) * (N as i64),
            "success leaves exactly the live entries' LO pages"
        );
    }

    /// The fleet joint contribution's contract tests (feature
    /// `postgres`; ADR-012 §4's one-tx ruling): the consolidation
    /// posture — kv = postgres + large = pg-lo on one shared instance,
    /// both engines opened through the default engine schema on the
    /// same dedicated database — with the pg-lo engine's staged-put
    /// mechanics riding the store-held `PgJointTx` beside the kv
    /// host's `joint_put` row write, and `pin_commit` as the joint
    /// terminal step. Failure forms play the store-core driver: the
    /// contribution's error paths leave the held tx terminable, and
    /// the driver-side rollback here is amendment A1's
    /// catch-propagate-terminate exercised at the instrumentation
    /// level the driver task designs on.
    #[cfg(feature = "postgres")]
    mod joint {
        use super::*;
        use crate::gc::{EngineFleetState, EngineWindowState, PinRow};
        use crate::pg_common::test_support::DatabaseJanitor;
        use crate::pg_common::{qualified_table, time_to_nanos, PgOpts, DEFAULT_SCHEMA};
        use crate::pg_lo::testing::{
            catalog_loids, catalog_rows, companion_loids_sorted, create_db, dsn_for_db, fresh_db,
            harness_admin_dsn, pages_for, pattern, stored_value,
        };
        use crate::postgres::PostgresBackend;
        use crate::store::DEFAULT_PIN_TTL;
        use crate::PinState;
        use std::time::Duration;

        /// The consolidation shape's harness: both engines on one
        /// dedicated database through the default engine schema (the
        /// consolidation posture's shared-schema shape), the kv host's
        /// GC schema ensured (`gc_pins` — the pin rows' home; fleet
        /// pins live on the kv host, ADR-012 §5), one janitor holding
        /// the whole database for both engines.
        struct JointHarness {
            lo: PgLoBackend,
            kv: PostgresBackend,
            _janitor: DatabaseJanitor,
        }

        async fn harness_joint(tag: &str) -> Option<JointHarness> {
            let admin = harness_admin_dsn()?;
            let db = fresh_db(tag);
            create_db(&admin, &db).await;
            let dsn = dsn_for_db(&db);
            let lo = PgLoBackend::new(&dsn, PgOpts::default())
                .await
                .expect("open the pg-lo engine");
            let kv = PostgresBackend::new(&dsn, PgOpts::default())
                .await
                .expect("open the postgres engine");
            kv.ensure_gc_schema().await.expect("kv host gc schema");
            Some(JointHarness {
                lo,
                kv,
                _janitor: DatabaseJanitor::new(&admin, &db),
            })
        }

        fn pin_row(key: &[u8], owner: &[u8], expiry: SystemTime) -> PinRow {
            PinRow {
                key: key.to_vec(),
                owner: owner.to_vec(),
                expiry,
            }
        }

        /// The gc_pins rows as the engine stores them — the raw
        /// encoding, read back through the shared instance (the probes
        /// ride the pg-lo engine's pool over the kv host's table: one
        /// instance, one schema).
        async fn pin_rows(h: &JointHarness) -> Vec<(Vec<u8>, Vec<u8>, i64)> {
            let client = h.lo.pool.get().await.expect("checkout");
            let sql = format!(
                "SELECT key, owner, expiry FROM {} ORDER BY key, owner",
                qualified_table(DEFAULT_SCHEMA, "gc_pins")
            );
            let rows = 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-tier entry's round-trip — byte-exact through the
        /// postgres engine's cursor (the large-tier twin of the
        /// parent's `stored_value`).
        async fn kv_stored_value(h: &JointHarness, key: &[u8]) -> Option<Vec<u8>> {
            let cursor = h.kv.get(key).await.expect("kv get succeeds")?;
            Some(cursor.read_at(0, cursor.len()).await.expect("kv read"))
        }

        /// The one-tx publish (the consolidation put's engine-level
        /// form, store-fleet-put's driver shape): the store core opens
        /// the joint tx on the kv host, the kv tier contributes its row
        /// write, the large tier contributes LO + companion row through
        /// the joint staging cursor, and `pin_commit` publishes entry +
        /// pin row in one step — byte-exact round-trips after it, pin
        /// row landed with the entry (pin-before-publish, one tx), and
        /// exactly one LO with exactly its page count.
        #[tokio::test]
        async fn joint_one_tx_publishes_kv_entry_lo_entry_and_pin_row() {
            let Some(h) = harness_joint("joint-one-tx").await else {
                return;
            };
            let lo_key = b"\x01joint-lo-key".to_vec();
            let kv_key = b"\x01joint-kv-key".to_vec();
            let owner = b"\x01instance-id".to_vec();

            let mut tx = h.kv.begin_joint_tx().await.unwrap();
            EngineFleetState::joint_put(&h.kv, &mut tx, &kv_key, CONTENT)
                .await
                .unwrap();
            let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap();
            staged.append(CONTENT).await.unwrap();
            assert_eq!(staged.len() as usize, CONTENT.len(), "staged length");
            assert!(
                staged.publish(&lo_key).await.unwrap(),
                "the fresh contribution publishes"
            );
            let expiry = SystemTime::now() + DEFAULT_PIN_TTL;
            h.kv.pin_commit(tx, &[pin_row(&lo_key, &owner, expiry)])
                .await
                .unwrap();

            let cursor =
                h.lo.get(&lo_key)
                    .await
                    .unwrap()
                    .expect("the LO entry published");
            assert_eq!(cursor.len(), CONTENT.len() as u64);
            let (bytes, digest) = cursor.read_range(0, cursor.len()).await.unwrap();
            assert_eq!(bytes, CONTENT, "byte-exact LO round-trip after pin_commit");
            assert_eq!(digest, crate::backend::slice_digest(CONTENT));

            assert!(h.kv.has(&kv_key).await.unwrap(), "the kv entry landed");
            assert_eq!(
                kv_stored_value(&h, &kv_key).await.as_deref(),
                Some(CONTENT),
                "byte-exact kv round-trip after pin_commit"
            );

            assert_eq!(
                EngineFleetState::pin_check(&h.kv, &lo_key).await.unwrap(),
                PinState::Protected,
                "the pin row landed with the entry in the one tx"
            );
            assert_eq!(
                pin_rows(&h).await,
                vec![(lo_key, owner, time_to_nanos(expiry))],
                "exactly the joint tx's pin row"
            );
            assert_eq!(
                catalog_loids(&h.lo).await.len(),
                1,
                "exactly one LO for the large entry"
            );
            assert_eq!(
                catalog_rows(&h.lo).await,
                pages_for(CONTENT.len()),
                "exactly the entry's LO pages"
            );
        }

        /// The joint form's failure discipline (the stage-hygiene
        /// invariant's tx form, the wave-4 joint-atomicity test's
        /// pg-lo edition): a forced failure mid-contribution — after
        /// the tail flush, so LO pages are staged into the held tx —
        /// returns with the tx terminable (never committed, never
        /// rolled back by the contribution); the driver-side
        /// termination (amendment A1's catch-propagate-terminate, the
        /// ROLLBACK the driver owns on failure paths) aborts the whole
        /// tx with zero residue: no LO pages, no companion row, no pin
        /// row, no kv row.
        #[tokio::test]
        async fn joint_contribution_failure_terminates_to_zero_residue() {
            let Some(h) = harness_joint("joint-fail").await else {
                return;
            };
            let failing_lo = PgLoBackend {
                pool: h.lo.pool.clone(),
                table: h.lo.table.clone(),
                window_table: h.lo.window_table.clone(),
                schema: h.lo.schema.clone(),
                orphan_reap_permitted: false,
                #[cfg(test)]
                fail_commit: true,
                #[cfg(test)]
                fail_window_stage: false,
            };
            let lo_key = b"\x01joint-failure-key".to_vec();
            let kv_key = b"\x01joint-failure-kv-key".to_vec();

            let mut tx = h.kv.begin_joint_tx().await.unwrap();
            EngineFleetState::joint_put(&h.kv, &mut tx, &kv_key, CONTENT)
                .await
                .unwrap();
            let mut staged = failing_lo
                .put_joint_staged(&mut tx)
                .await
                .expect("the handout holds the held checkout");
            staged
                .append(&pattern(2 * LOWRITE_CHUNK + 128))
                .await
                .unwrap();
            let err = staged.publish(&lo_key).await.unwrap_err();
            assert!(
                matches!(err, Error::Io(_)),
                "the injected media failure surfaces as Io"
            );

            tx.client()
                .unwrap()
                .batch_execute("ROLLBACK")
                .await
                .unwrap();

            assert!(
                !h.lo.has(&lo_key).await.unwrap(),
                "the aborted tx published no LO entry"
            );
            assert!(
                !h.kv.has(&kv_key).await.unwrap(),
                "the aborted tx published no kv row"
            );
            assert!(
                h.lo.list().await.unwrap().is_empty(),
                "zero residue in the companion table"
            );
            assert_eq!(
                catalog_rows(&h.lo).await,
                0,
                "zero residue in the LO catalog — the staged pages died with the tx"
            );
            assert_eq!(
                EngineFleetState::pin_check(&h.kv, &lo_key).await.unwrap(),
                PinState::Unprotected,
                "no pin row survived"
            );
            assert_eq!(pin_rows(&h).await, vec![], "zero residue in gc_pins");
            assert!(
                h.kv.list().await.unwrap().is_empty(),
                "zero residue in the blobs table itself"
            );
        }

        /// The CAS re-put form: a second contribution of the same key
        /// publishes an **empty contribution** (`Ok(false)` — the
        /// staged LO unlinked inside the store-held tx), the tx commits
        /// the pin row alone, and the pin row upsert-renews — with the
        /// losing contribution's staged LO leaving zero residue.
        #[tokio::test]
        async fn joint_contribution_reput_commits_empty_and_upsert_renews() {
            let Some(h) = harness_joint("joint-reput").await else {
                return;
            };
            let key = b"\x01joint-reput-key".to_vec();
            let owner = b"\x01instance-id".to_vec();

            let mut tx = h.kv.begin_joint_tx().await.unwrap();
            let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap();
            staged.append(CONTENT).await.unwrap();
            assert!(staged.publish(&key).await.unwrap(), "fresh contribution");
            let expiry = SystemTime::now() + DEFAULT_PIN_TTL;
            h.kv.pin_commit(tx, &[pin_row(&key, &owner, expiry)])
                .await
                .unwrap();

            let renewed = SystemTime::now() + DEFAULT_PIN_TTL + Duration::from_secs(7);
            let mut tx = h.kv.begin_joint_tx().await.unwrap();
            let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap();
            staged.append(b"overwritten-by-joint-reput").await.unwrap();
            assert!(
                !staged.publish(&key).await.unwrap(),
                "the re-put publishes an empty contribution, never an overwrite"
            );
            h.kv.pin_commit(tx, &[pin_row(&key, &owner, renewed)])
                .await
                .unwrap();

            assert_eq!(
                stored_value(&h.lo, &key).await.as_deref(),
                Some(CONTENT),
                "first-writer-wins: the re-put does not overwrite"
            );
            assert_eq!(
                pin_rows(&h).await,
                vec![(key, owner, time_to_nanos(renewed))],
                "the re-put's pin row upsert renews the same instance's row"
            );
            assert_eq!(
                catalog_rows(&h.lo).await,
                pages_for(CONTENT.len()),
                "the losing contribution's staged LO left zero residue (unlinked in-tx)"
            );
            assert_eq!(
                catalog_loids(&h.lo).await.len(),
                1,
                "exactly the first writer's LO remains"
            );
            assert_eq!(
                companion_loids_sorted(&h.lo).await.len(),
                1,
                "the companion table holds exactly one row"
            );
        }

        /// The 512 KiB knee rides: a multi-chunk contribution — 64 KiB
        /// appends (the store core's put-loop window size) aggregating
        /// into `LOWRITE_CHUNK` `lowrite` statements inside the held
        /// tx — with the derivation pass reading the staged artifact
        /// back through the read seam (including a read crossing the
        /// flushed/pending aggregation boundary), the one-tx publish,
        /// and the byte-exact committed round-trip.
        #[tokio::test]
        async fn joint_contribution_rides_the_512kib_knee_multichunk() {
            let Some(h) = harness_joint("joint-knee").await else {
                return;
            };
            let content = pattern(1_500_000);
            let key = b"\x01joint-knee-key".to_vec();
            let owner = b"\x01instance-id".to_vec();

            let mut tx = h.kv.begin_joint_tx().await.unwrap();
            let mut staged = h.lo.put_joint_staged(&mut tx).await.unwrap();
            for chunk in content.chunks(64 * 1024) {
                staged.append(chunk).await.unwrap();
            }
            assert_eq!(staged.len() as usize, content.len(), "staged length");

            let (bytes, digest) = staged.read_range(0, staged.len()).await.unwrap();
            assert_eq!(bytes, content, "staged artifact readable as a whole");
            assert_eq!(digest, crate::backend::slice_digest(&content));
            let boundary = LOWRITE_CHUNK as u64;
            assert!(
                staged.len() > boundary,
                "the staged content crossed the first aggregation flush"
            );
            let crossed = staged.read_at(boundary - 64, 256).await.unwrap();
            assert_eq!(
                crossed,
                content[(LOWRITE_CHUNK - 64)..(LOWRITE_CHUNK + 192)],
                "a staged read crossing the flushed/pending boundary is byte-exact \
                 (LO windows over committed-into-tx pages + the in-memory tail)"
            );

            assert!(
                staged.publish(&key).await.unwrap(),
                "the multi-chunk contribution publishes"
            );
            let expiry = SystemTime::now() + DEFAULT_PIN_TTL;
            h.kv.pin_commit(tx, &[pin_row(&key, &owner, expiry)])
                .await
                .unwrap();

            let cursor =
                h.lo.get(&key)
                    .await
                    .unwrap()
                    .expect("the LO entry published");
            let (bytes, digest) = cursor.read_range(0, cursor.len()).await.unwrap();
            assert_eq!(bytes, content, "byte-exact multi-chunk round-trip");
            assert_eq!(digest, crate::backend::slice_digest(&content));
            assert_eq!(
                catalog_rows(&h.lo).await,
                pages_for(content.len()),
                "exactly the entry's LO pages (C2's catalog math)"
            );
        }
    }
}