alkblobs 0.2.2

//! The descriptorless read shape and the companion-row `Backend` body:
//! stateless `lo_get` windows (the shipped get shape, C4), the window
//! loop with its unexpected-EOF clean failure, the streaming cursor that
//! closes over the companion row, and the whole `Backend` impl —
//! companion-table `has`/`list`/`size` (the contract authority), the
//! one-tx put (`put_in_tx`), the held-tx stage handout, the one-tx
//! delete's rows-affected discipline. The window-family seam clauses
//! ([`crate::gc::EngineWindowState`]) ride here too — the window-only
//! schema and the durable staged delete window.

use std::time::SystemTime;

use deadpool_postgres::{Object, Pool};

use super::staging::{lo_create, lo_open, lowrite_all};
use super::{gc_window_ddl, PgLoBackend, PgLoStaging, INV_WRITE, READ_CHUNK};
use crate::backend::{clamped, Backend, ReadCursor};
use crate::gc::{EngineWindowState, WindowCandidate, WINDOW_TAKE_BATCH};
use crate::pg_common::{err, is_undefined_table, pg_err, size_u64, time_to_nanos};
use crate::Error;

impl PgLoBackend {
    /// The put's tx body (over the opened transaction): CAS existence
    /// (a re-put is a no-op tx — the COMMIT of an empty transaction —
    /// never an overwrite, never a `lo_create`), then lo_create →
    /// chunked `lowrite` → companion-row insert. Errors bubble to
    /// [`Backend::put`]'s ROLLBACK convergence point.
    async fn put_in_tx(&self, client: &Object, key: &[u8], value: &[u8]) -> Result<(), Error> {
        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(());
        }
        let loid = lo_create(client).await?;
        let fd = lo_open(client, loid, INV_WRITE).await?;
        lowrite_all(client, fd, value).await?;
        let size = value.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, &loid, &size, &committed_at])
            .await
            .map_err(pg_err)?;
        Ok(())
    }
}

/// The window family (ADR-012 §5's pg-lo row): the window-table-only
/// schema and the durable staged delete window. Same semantics as the
/// sqlite and postgres reference implementations — different dialect;
/// nothing of the fleet family exists on this engine.
impl EngineWindowState for PgLoBackend {
    /// Create `gc_window (key bytea PRIMARY KEY, visible_after bigint)`
    /// — this engine's applicability row is the window table only;
    /// `gc_pins` is never created here (ADR-012 §5: fleet pins live on
    /// the kv host) and the embedder's liveness table is never created
    /// in any case (ADR-008's schema-ownership rule). Idempotent, so
    /// the twice-at-open call form is a no-op. The store core calls
    /// this at construction exactly when this engine hosts the window
    /// (the one-window-host rule's large-host arm — kv = mem +
    /// large = pg-lo); on a dual-SQL instance the kv host's call runs
    /// and this stays dormant, never invoked at construction.
    async fn ensure_gc_schema(&self) -> Result<(), Error> {
        let client = self.pool.get().await.map_err(err)?;
        client
            .batch_execute(&gc_window_ddl(&self.schema))
            .await
            .map_err(pg_err)
    }

    /// Stage candidates into the durable window — **one transaction per
    /// call** (review-001 N-2): the loop's upserts commit together, and a
    /// failure anywhere inside the loop rolls the session-level
    /// transaction back, leaving zero staged rows (the failure-path
    /// discipline is the engine's one-tx bodies': every error arm closes
    /// the tx with an explicit `ROLLBACK` before surfacing). The upsert
    /// discipline: a re-stage of the same key at the same deadline is
    /// idempotent (one row, value unchanged), a re-stage at a new deadline
    /// updates the row's `visible_after` rather than duplicating it —
    /// **forward-only** (review-001 N-4): an earlier re-stage never moves
    /// `visible_after` backwards (`GREATEST` semantics; the target table
    /// rides the `w` alias, the schema-qualified insert's DO-UPDATE
    /// reference form).
    async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error> {
        if candidates.is_empty() {
            return Ok(());
        }
        #[cfg(test)]
        let fail_window_stage = self.fail_window_stage;
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        let sql = format!(
            "INSERT INTO {} AS w (key, visible_after) VALUES ($1, $2) \
             ON CONFLICT (key) DO UPDATE SET \
             visible_after = GREATEST(w.visible_after, excluded.visible_after)",
            self.window_table
        );
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(pg_err(e));
            }
        };
        for candidate in candidates.iter() {
            #[cfg(test)]
            if fail_window_stage
                && candidates
                    .last()
                    .is_some_and(|last| last.key == candidate.key)
            {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(Error::Io(
                    "pg-lo: window staging failure (mid-loop, injected)".to_owned(),
                ));
            }
            let visible_after = time_to_nanos(candidate.visible_after);
            match client
                .execute(&stmt, &[&candidate.key, &visible_after])
                .await
            {
                Ok(_) => {}
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    return Err(pg_err(e));
                }
            }
        }
        match client.batch_execute("COMMIT").await {
            Ok(_) => Ok(()),
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(pg_err(e))
            }
        }
    }

    /// Take overdue candidates (`visible_after <= now` — the `<=`
    /// boundary is the seam doc's pin): bounded to [`WINDOW_TAKE_BATCH`]
    /// rows (review-001 N-3 — oldest deadlines first, `ORDER BY
    /// visible_after`; the `LIMIT` rides the inner candidate query), the
    /// calling sweep executor consuming the take in a loop until drained.
    /// One atomic step, unchanged: the single `DELETE … RETURNING`
    /// statement selecting and removing the rows together, so a row is
    /// never handed out twice and an aborted body leaves no partial take.
    /// A missing table maps to empty (the postgres twin's virgin take;
    /// the recorded read-only split).
    async fn window_take(&self, now: SystemTime) -> Result<Vec<Vec<u8>>, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let horizon = time_to_nanos(now);
        let sql = format!(
            "DELETE FROM {} WHERE key IN ( \
                 SELECT key FROM {} WHERE visible_after <= $1 \
                 ORDER BY visible_after LIMIT $2 \
             ) RETURNING key",
            self.window_table, self.window_table
        );
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => return Ok(Vec::new()),
            Err(e) => return Err(err(e)),
        };
        let rows = client
            .query(&stmt, &[&horizon, &(WINDOW_TAKE_BATCH as i64)])
            .await
            .map_err(pg_err)?;
        let mut keys = rows
            .iter()
            .map(|row| row.try_get::<_, Vec<u8>>(0).map_err(pg_err))
            .collect::<Result<Vec<Vec<u8>>, _>>()?;
        keys.sort();
        Ok(keys)
    }
}

/// One descriptorless `lo_get` window — the shipped read shape (C4):
/// stateless, pool-friendly, bigint offset. Window length stays within
/// [`READ_CHUNK`] (and the int4 parameter domain).
async fn lo_get_window(
    client: &Object,
    loid: u32,
    offset: u64,
    len: usize,
) -> Result<Vec<u8>, Error> {
    let stmt = client
        .prepare_cached("SELECT lo_get($1, $2, $3) AS b")
        .await
        .map_err(err)?;
    let row = client
        .query_one(&stmt, &[&loid, &(offset as i64), &(len as i32)])
        .await
        .map_err(pg_err)?;
    row.try_get(0).map_err(pg_err)
}

/// The streamed read over [`lo_get_window`] windows: the cursor
/// contract's clamp against `body_len` first, then the window loop; a
/// window coming up empty before the body is complete is the
/// unexpected-EOF clean failure (C4's shape — never surfaced for a
/// healthy store, whose companion row is authoritative).
pub(super) async fn read_windows(
    client: &Object,
    loid: u32,
    start: u64,
    end: u64,
) -> Result<Vec<u8>, Error> {
    let wanted = (end - start) as usize;
    let mut out = Vec::with_capacity(wanted);
    let mut pos = start;
    while pos < end {
        let window = ((end - pos) as usize).min(READ_CHUNK);
        let bytes = lo_get_window(client, loid, pos, window).await?;
        if bytes.is_empty() {
            return Err(Error::Io(format!(
                "pg-lo: large object {loid} ended {} bytes short of the \
                 committed window (unexpected EOF, companion row overclaims)",
                wanted - out.len()
            )));
        }
        out.extend_from_slice(&bytes);
        pos += bytes.len() as u64;
    }
    Ok(out)
}

/// The streaming read cursor (the large body family): descriptorless —
/// it closes over the companion row (oid + length; content immutable,
/// so the row is authoritative), never a tx-scoped descriptor, never a
/// content load at `get` time. `len()` is the companion row's `size` —
/// the same metadata-only fact `Backend::size` probes. Range reads
/// are `lo_get_window` windows.
pub struct PgLoCursor {
    pool: Pool,
    loid: u32,
    len: u64,
}

impl ReadCursor for PgLoCursor {
    fn len(&self) -> u64 {
        self.len
    }

    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.pool.get().await.map_err(err)?;
        read_windows(&client, self.loid, start as u64, end as u64).await
    }
}

impl Backend for PgLoBackend {
    type Cursor = PgLoCursor;
    type Staging = PgLoStaging;

    fn name(&self) -> &'static str {
        "pg-lo"
    }

    /// The companion-table EXISTS probe — never a value load, never LO
    /// content. Virgin store: `false`.
    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let sql = format!("SELECT EXISTS(SELECT 1 FROM {} WHERE key = $1)", self.table);
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => return Ok(false),
            Err(e) => return Err(err(e)),
        };
        let present = client
            .query_one(&stmt, &[&key])
            .await
            .map_err(pg_err)?
            .try_get::<_, bool>(0)
            .map_err(pg_err)?;
        Ok(present)
    }

    /// The descriptorless cursor body: the companion row is read once
    /// (oid + size — metadata only); the bytes stream through
    /// [`lo_get_window`] windows afterwards. Virgin store: `None`.
    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let sql = format!("SELECT loid, size FROM {} WHERE key = $1", self.table);
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => return Ok(None),
            Err(e) => return Err(err(e)),
        };
        match client.query_opt(&stmt, &[&key]).await.map_err(pg_err)? {
            Some(row) => {
                let loid = row.try_get::<_, u32>(0).map_err(pg_err)?;
                let len = size_u64(row.try_get::<_, i64>(1).map_err(pg_err)?);
                Ok(Some(PgLoCursor {
                    pool: self.pool.clone(),
                    loid,
                    len,
                }))
            }
            None => Ok(None),
        }
    }

    /// The whole-value put — one tx (BEGIN → CAS existence → `lo_create`
    /// → chunked `lowrite` → companion-row insert → COMMIT), CAS
    /// first-writer-wins: an existing key is a no-op tx (nothing is
    /// staged, no LO is born), never an overwrite.
    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        match self.put_in_tx(&client, key, value).await {
            Ok(()) => match client.batch_execute("COMMIT").await {
                Ok(()) => Ok(()),
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    Err(pg_err(e))
                }
            },
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(e)
            }
        }
    }

    /// The held-tx stage's handout: checkout + BEGIN + `lo_create` +
    /// `lo_open(INV_WRITE)`. Every failure inside the handout converges
    /// on ROLLBACK before returning the checkout.
    async fn put_staged(&self) -> Result<Self::Staging, Error> {
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        let loid = match lo_create(&client).await {
            Ok(loid) => loid,
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(e);
            }
        };
        let fd = match lo_open(&client, loid, INV_WRITE).await {
            Ok(fd) => fd,
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(e);
            }
        };
        Ok(PgLoStaging {
            client: Some(client),
            loid,
            fd,
            table: self.table.clone(),
            flushed_len: 0,
            pending: Vec::new(),
            #[cfg(test)]
            fail_commit: self.fail_commit,
        })
    }

    /// Unconditional delete — one tx (`lo_unlink` + companion-row
    /// delete), rows-affected discipline. Virgin store (42P01):
    /// `Err(Error::Missing)`.
    async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        let client = self.pool.get().await.map_err(err)?;
        client.batch_execute("BEGIN").await.map_err(pg_err)?;
        let loid_sql = format!("SELECT loid FROM {} WHERE key = $1", self.table);
        let loid_stmt = match client.prepare_cached(&loid_sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(Error::Missing);
            }
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(err(e));
            }
        };
        let row = match client.query_opt(&loid_stmt, &[&key]).await {
            Ok(row) => row,
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(pg_err(e));
            }
        };
        let Some(row) = row else {
            let _ = client.batch_execute("ROLLBACK").await;
            return Err(Error::Missing);
        };
        let loid: u32 = match row.try_get(0) {
            Ok(loid) => loid,
            Err(e) => {
                let e = pg_err(e);
                let _ = client.batch_execute("ROLLBACK").await;
                return Err(e);
            }
        };
        let result = async {
            let unlink_stmt = client
                .prepare_cached("SELECT lo_unlink($1)")
                .await
                .map_err(pg_err)?;
            client
                .query_one(&unlink_stmt, &[&loid])
                .await
                .map_err(pg_err)?;
            let delete_sql = format!("DELETE FROM {} WHERE key = $1", self.table);
            let delete_stmt = client.prepare_cached(&delete_sql).await.map_err(err)?;
            client
                .execute(&delete_stmt, &[&key])
                .await
                .map_err(pg_err)?;
            Ok(())
        }
        .await;
        match result {
            Ok(()) => match client.batch_execute("COMMIT").await {
                Ok(()) => Ok(()),
                Err(e) => {
                    let _ = client.batch_execute("ROLLBACK").await;
                    Err(pg_err(e))
                }
            },
            Err(e) => {
                let _ = client.batch_execute("ROLLBACK").await;
                Err(e)
            }
        }
    }

    /// Whole-engine key enumeration — the single-statement MVCC-snapshot
    /// cursor over the companion table (the contract authority), ordered
    /// by key. Virgin store: empty.
    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let sql = format!("SELECT key FROM {} ORDER BY key", self.table);
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => return Ok(Vec::new()),
            Err(e) => return Err(err(e)),
        };
        let rows = client.query(&stmt, &[]).await.map_err(pg_err)?;
        rows.iter()
            .map(|row| row.try_get::<_, Vec<u8>>(0).map_err(pg_err))
            .collect()
    }

    /// The metadata-only length probe: the companion row's `size`
    /// column — never LO content, never a value load. Virgin store:
    /// `None`.
    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        let client = self.pool.get().await.map_err(err)?;
        let sql = format!("SELECT size FROM {} WHERE key = $1", self.table);
        let stmt = match client.prepare_cached(&sql).await {
            Ok(stmt) => stmt,
            Err(e) if is_undefined_table(&e) => return Ok(None),
            Err(e) => return Err(err(e)),
        };
        match client.query_opt(&stmt, &[&key]).await.map_err(pg_err)? {
            Some(row) => Ok(Some(size_u64(row.try_get::<_, i64>(0).map_err(pg_err)?))),
            None => Ok(None),
        }
    }
}

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

    use crate::pg_lo::testing::{
        at, catalog_loids, catalog_rows, companion_loids_sorted, harness, harness_gc,
        harness_schemaless, pages_for, pattern, schema_tables, staged, staged_in_64k_windows,
        stages, stored_value, window_rows, CONTENT,
    };

    /// The Backend method contract: the name, the seven methods'
    /// absence/value/error disciplines, CAS first-writer-wins, and
    /// virgin-store reads as no-ops — mirroring the postgres engine's
    /// suite.
    #[tokio::test]
    async fn backend_method_contract() {
        let Some(backend) = harness("engine").await else {
            return;
        };
        let key = b"\x01opaque-key-bytes".to_vec();
        assert_eq!(backend.name(), "pg-lo", "reporting identity");

        assert!(
            !backend.has(&key).await.unwrap(),
            "virgin store: has is a no-op read"
        );
        assert!(
            backend.get(&key).await.unwrap().is_none(),
            "virgin store: get is a no-op read"
        );
        assert_eq!(
            backend.size(&key).await.unwrap(),
            None,
            "virgin store: size is a no-op read"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "virgin store: list is empty"
        );
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "virgin store: delete reports Missing"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            0,
            "virgin operations touched no large object pages"
        );

        backend.put(&key, CONTENT).await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "put lands");
        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(CONTENT.len() as u64),
            "size is metadata-only and exact"
        );
        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "CAS first-writer wins: re-put is a no-op"
        );
        assert_eq!(
            backend.list().await.unwrap(),
            vec![key.clone()],
            "list is complete, keys not values"
        );
        assert_eq!(catalog_loids(&backend).await.len(), 1, "one LO exactly");
        assert_eq!(
            catalog_rows(&backend).await,
            pages_for(CONTENT.len()),
            "the LO catalog holds ceil(len/2048) page rows (C2's catalog math)"
        );

        staged_in_64k_windows(&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"
        );

        backend.delete(&key).await.unwrap();
        assert!(!backend.has(&key).await.unwrap(), "delete lands");
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "delete reports Missing when absent"
        );
        assert_eq!(
            catalog_rows(&backend).await,
            0,
            "delete freed the LO's pages at commit (C5: space is reused)"
        );
    }

    /// The `list` contract is complete AND well-formed after mixed ops
    /// (POC #7's exact-count gate: list == companion table), and in a
    /// healthy store the companion table's oids agree exactly with the
    /// LO catalog's distinct oids — the C1 shape (the catalog read here
    /// is test instrumentation; the engine never addresses through it).
    #[tokio::test]
    async fn list_is_complete_after_mixed_ops() {
        let Some(backend) = harness("list").await else {
            return;
        };
        let mut keys: Vec<Vec<u8>> = vec![
            b"\x02b-key".to_vec(),
            b"\x02a-key".to_vec(),
            b"\x02c-key".to_vec(),
        ];
        for key in &keys[..2] {
            backend.put(key, CONTENT).await.unwrap();
        }
        staged(&backend, &keys[2], b"staged-content")
            .await
            .expect("staged commit lands");
        backend.delete(&keys[1]).await.unwrap();

        keys.remove(1);
        let mut listed = keys.clone();
        listed.sort();
        assert_eq!(
            backend.list().await.unwrap(),
            listed,
            "list is complete after put/staged-commit/delete mixed ops"
        );
        for key in &listed {
            assert!(backend.has(key).await.unwrap(), "no spurious keys");
        }

        let client = backend.pool.get().await.unwrap();
        let sql = format!(
            "SELECT count(*) FROM (\
             SELECT loid FROM {t} EXCEPT SELECT DISTINCT loid FROM pg_largeobject\
             ) missing, (\
             SELECT DISTINCT loid FROM pg_largeobject EXCEPT SELECT loid FROM {t}\
             ) orphaned",
            t = backend.table
        );
        let disagreements: i64 = client.query_one(sql.as_str(), &[]).await.unwrap().get(0);
        assert_eq!(
            disagreements, 0,
            "companion table and LO catalog agree exactly (no missing pages, no orphans)"
        );
        assert_eq!(
            companion_loids_sorted(&backend).await,
            catalog_loids(&backend).await,
            "the companion table's loids are the catalog's oids"
        );
    }

    /// The streaming window cursor body's read contract: `len()` exact
    /// (the companion row's size), EOF-clamped reads (over-read, tail,
    /// at-EOF empty, zero-length), window-boundary crossings (reads that
    /// straddle the 512 KiB window edge), and the ADR-006 range seam's
    /// digest over exactly the returned bytes.
    #[tokio::test]
    async fn cursor_contract_streams_windows() {
        let Some(backend) = harness("cursor").await else {
            return;
        };
        let content = pattern(2 * READ_CHUNK + 100);
        let key = b"\x01cursor-key".to_vec();
        backend.put(&key, &content).await.unwrap();

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let len = content.len() as u64;
        assert_eq!(cursor.len(), len, "len is the companion row's size");

        assert_eq!(
            cursor.read_at(0, len).await.unwrap(),
            content,
            "full multi-window read assembles byte-exact"
        );
        assert_eq!(
            cursor.read_at(0, len + 4097).await.unwrap(),
            content,
            "over-read clamps at EOF, pread-style"
        );
        let mid = (content.len() / 2) as u64;
        assert_eq!(
            cursor.read_at(mid, len).await.unwrap(),
            content[(content.len() / 2)..],
            "tail read clamps"
        );
        assert_eq!(
            cursor.read_at(len, 5).await.unwrap(),
            Vec::<u8>::new(),
            "read at EOF is empty, never an error"
        );
        assert_eq!(
            cursor.read_at(u64::MAX, 5).await.unwrap(),
            Vec::<u8>::new(),
            "an overflowing u64 offset clamps in u64 arithmetic — the C7 trap not inherited"
        );
        assert_eq!(
            cursor.read_at(0, 0).await.unwrap(),
            Vec::<u8>::new(),
            "zero-length read"
        );

        let boundary = READ_CHUNK as u64;
        let windowed = cursor.read_at(boundary - 100, 300).await.unwrap();
        assert_eq!(
            windowed,
            content[(READ_CHUNK - 100)..(READ_CHUNK + 200)],
            "a read straddling the window edge crosses two lo_get statements byte-exact"
        );
        let (slice, digest) = cursor.read_range(boundary, 7).await.unwrap();
        assert_eq!(slice, content[READ_CHUNK..READ_CHUNK + 7]);
        assert_eq!(digest, crate::backend::slice_digest(&slice));
    }

    /// The degenerate case: a zero-byte entry round-trips through put,
    /// the staged form, and the committed cursor (size exact-zero, every
    /// read empty).
    #[tokio::test]
    async fn empty_entry_round_trips() {
        let Some(backend) = harness("empty").await else {
            return;
        };
        let key = b"\x01empty-key".to_vec();

        backend.put(&key, b"").await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "empty value still lands");
        assert_eq!(backend.size(&key).await.unwrap(), Some(0), "exact zero");
        let cursor = backend.get(&key).await.unwrap().expect("entry present");
        assert_eq!(cursor.len(), 0);
        assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::<u8>::new());

        let staged_key = b"\x01empty-staged-key".to_vec();
        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"").await.unwrap();
        assert_eq!(staging.len(), 0, "empty append stays empty");
        staging
            .commit(&staged_key)
            .await
            .expect("empty commit publishes");
        assert!(
            backend.has(&staged_key).await.unwrap(),
            "empty entry landed"
        );
        assert_eq!(backend.size(&staged_key).await.unwrap(), Some(0));
        let committed = backend
            .get(&staged_key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(
            committed.read_at(0, 8).await.unwrap(),
            Vec::<u8>::new(),
            "committed empty body reads empty"
        );
    }

    /// `size` reads the companion row's `size` column only (metadata
    /// only, the C1 table-level authority) — exact for a large value,
    /// riding the PK index (EXPLAIN proof), and never touching LO
    /// content.
    #[tokio::test]
    async fn size_probe_is_metadata_only_and_exact() {
        let Some(backend) = harness("size").await else {
            return;
        };
        let key = b"\x01size-probe-key".to_vec();
        let large_value = pattern(1024 * 1024);
        backend.put(&key, &large_value).await.unwrap();

        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(large_value.len() as u64),
            "size is metadata-only and exact"
        );
        assert_eq!(backend.size(b"\x01absent").await.unwrap(), None);

        let client = backend.pool.get().await.unwrap();
        let explain = format!(
            "EXPLAIN SELECT size FROM {} WHERE key = '\\x{}'",
            backend.table,
            hex::encode(&key)
        );
        let rows = client.query(explain.as_str(), &[]).await.unwrap();
        let plan: String = rows
            .iter()
            .map(|row| row.get::<_, String>(0))
            .collect::<Vec<_>>()
            .join(" ");
        assert!(
            plan.contains("Index Scan") && plan.contains("lo_entries_pkey"),
            "the size probe searches the companion table's PK index: {plan}"
        );
    }

    /// Schema contract (the pg-lo row): `ensure_gc_schema` creates
    /// `gc_window` only — `gc_pins` is never created by pg-lo (fleet
    /// pins live on the kv host, ADR-012 §5) and the embedder's
    /// liveness table is never created in any case (the
    /// schema-ownership rule: exactly the engine's two tables exist) —
    /// the call is idempotent, and the gc_window row shape is
    /// bytea/bigint.
    #[tokio::test]
    async fn gc_schema_creates_window_table_only() {
        let Some(backend) = harness("window-schema").await else {
            return;
        };
        backend.ensure_gc_schema().await.unwrap();
        backend.ensure_gc_schema().await.unwrap();
        assert_eq!(
            schema_tables(&backend).await,
            vec!["gc_window", "lo_entries"],
            "the engine's two tables exactly — no gc_pins, no embedder liveness table"
        );
        assert!(
            !backend.table_exists("gc_pins").await.unwrap(),
            "gc_pins is never created by pg-lo (fleet pins live on the kv host)"
        );

        let client = backend.pool.get().await.unwrap();
        let columns: Vec<String> = client
            .query(
                "SELECT data_type::text FROM information_schema.columns \
                 WHERE table_schema = $1 AND table_name = 'gc_window' \
                 ORDER BY ordinal_position",
                &[&backend.schema],
            )
            .await
            .unwrap()
            .iter()
            .map(|row| row.get::<_, String>(0))
            .collect();
        assert_eq!(
            columns,
            ["bytea", "bigint"],
            "the gc_window row shape: key/visible_after"
        );
    }

    /// Staging contract: idempotent per `(key, deadline)` — the same
    /// candidate staged twice leaves exactly one row — and the
    /// re-stage's deadline conflict update is **forward-only** (review-001
    /// N-4): a re-stage at a later deadline moves the row forward, a
    /// re-stage at an earlier deadline never moves `visible_after`
    /// backwards.
    #[tokio::test]
    async fn window_stage_is_idempotent_and_restage_moves_the_deadline_forward_only() {
        let Some(backend) = harness_gc("window-stage").await else {
            return;
        };
        let key = b"\x01window-key".to_vec();
        let deadline = at(1_000_000_000);

        backend
            .window_stage(&[stages(&key, deadline)])
            .await
            .unwrap();
        backend
            .window_stage(&[stages(&key, deadline)])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await,
            vec![(key.clone(), 1_000_000_000)],
            "re-stage at the same deadline is one idempotent row"
        );

        backend
            .window_stage(&[stages(&key, at(2_000_000_000))])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await,
            vec![(key.clone(), 2_000_000_000)],
            "a re-stage at a later deadline moves the row forward"
        );

        backend
            .window_stage(&[stages(&key, at(1_000_000_000))])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await,
            vec![(key.clone(), 2_000_000_000)],
            "a re-stage at an earlier deadline never moves visible_after backwards"
        );

        let other = b"\x01window-other".to_vec();
        backend
            .window_stage(&[stages(&other, deadline)])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await.len(),
            2,
            "distinct keys are distinct rows"
        );
    }

    /// The staging is transactional (review-001 N-2): a mid-loop failure
    /// (injected after the first candidate's upsert has run inside the
    /// open transaction) rolls the whole call back — zero staged rows,
    /// prior rows untouched — and the backend stages cleanly again once
    /// the failure clears.
    #[tokio::test]
    async fn window_stage_mid_loop_failure_stages_nothing() {
        let Some(mut backend) = harness_gc("window-stage-abort").await else {
            return;
        };
        let prior = b"\x01window-prior".to_vec();
        backend
            .window_stage(&[stages(&prior, at(1_000_000_000))])
            .await
            .unwrap();

        backend.fail_window_stage = true;
        let aborted_a = b"\x01window-aborted-a".to_vec();
        let aborted_b = b"\x01window-aborted-b".to_vec();
        let err = backend
            .window_stage(&[
                stages(&aborted_a, at(2_000_000_000)),
                stages(&aborted_b, at(2_000_000_000)),
            ])
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::Io(ref m) if m.contains("mid-loop")),
            "the injected stage failure surfaces as Io: {err}"
        );
        assert_eq!(
            window_rows(&backend).await,
            vec![(prior.clone(), 1_000_000_000)],
            "the aborted call staged nothing — the prior row is the only residue"
        );

        backend.fail_window_stage = false;
        backend
            .window_stage(&[
                stages(&aborted_a, at(2_000_000_000)),
                stages(&aborted_b, at(2_000_000_000)),
            ])
            .await
            .unwrap();
        assert_eq!(
            window_rows(&backend).await.len(),
            3,
            "the cleared backend stages cleanly again"
        );
    }

    /// The take is bounded (review-001 N-3): a backlog larger than
    /// [`WINDOW_TAKE_BATCH`] hands out at most that many overdue rows per
    /// call — oldest deadlines first — draining across repeated calls,
    /// and the drained window takes empty.
    #[tokio::test]
    async fn window_take_is_bounded_across_the_overdue_backlog() {
        let Some(backend) = harness_gc("window-take-bounded").await else {
            return;
        };
        let total = WINDOW_TAKE_BATCH + 7;
        let candidates: Vec<WindowCandidate> = (0..total)
            .map(|i| {
                stages(
                    format!("\x01win-backlog-{i:05}").as_bytes(),
                    at(i as i64 * 1_000_000),
                )
            })
            .collect();
        backend.window_stage(&candidates).await.unwrap();

        let first = backend
            .window_take(at(total as i64 * 1_000_000))
            .await
            .unwrap();
        let mut expected_first: Vec<Vec<u8>> = (0..WINDOW_TAKE_BATCH)
            .map(|i| format!("\x01win-backlog-{i:05}").into_bytes())
            .collect();
        expected_first.sort();
        assert_eq!(
            first, expected_first,
            "the first take hands out exactly the batch, oldest deadlines first"
        );

        let second = backend
            .window_take(at(total as i64 * 1_000_000))
            .await
            .unwrap();
        let mut expected_rest: Vec<Vec<u8>> = (WINDOW_TAKE_BATCH..total)
            .map(|i| format!("\x01win-backlog-{i:05}").into_bytes())
            .collect();
        expected_rest.sort();
        assert_eq!(second, expected_rest, "the next take drains the remainder");

        assert_eq!(
            backend
                .window_take(at(total as i64 * 1_000_000))
                .await
                .unwrap(),
            Vec::<Vec<u8>>::new(),
            "the drained window takes empty"
        );
    }

    /// Take contract: exactly the overdue set (`visible_after <= now`),
    /// removed atomically — a second take returns nothing — with future
    /// rows left in place, and the key round-trips through the raw
    /// encoding (signed nanos, so a pre-epoch deadline rides along).
    #[tokio::test]
    async fn window_take_returns_the_overdue_set_and_leaves_future_rows() {
        let Some(backend) = harness_gc("window-take").await else {
            return;
        };
        let past_pre_epoch = b"\x01win-past-pre-epoch".to_vec();
        let past = b"\x01win-past".to_vec();
        let future = b"\x01win-future".to_vec();
        let far_future = b"\x01win-far-future".to_vec();

        backend
            .window_stage(&[
                stages(&past_pre_epoch, at(-1_000_000_000)),
                stages(&past, at(1_000_000_000)),
                stages(&future, at(2_000_000_000)),
                stages(&far_future, at(3_600_000_000_000)),
            ])
            .await
            .unwrap();

        let taken = backend.window_take(at(1_000_000_001)).await.unwrap();
        let mut expected = vec![past_pre_epoch.clone(), past.clone()];
        expected.sort();
        assert_eq!(taken, expected, "exactly the overdue set");
        assert_eq!(
            backend.window_take(at(1_000_000_001)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "the rows left the table: never handed out twice"
        );
        assert_eq!(
            window_rows(&backend).await,
            vec![
                (far_future.clone(), 3_600_000_000_000),
                (future.clone(), 2_000_000_000)
            ],
            "future rows left in place"
        );
    }

    /// The take boundary is `visible_after == now` (the seam doc's
    /// `<=`): a candidate visible exactly at the horizon is overdue;
    /// one nanosecond later it is not.
    #[tokio::test]
    async fn window_take_boundary_visible_after_equal_now_is_overdue() {
        let Some(backend) = harness_gc("window-boundary").await else {
            return;
        };
        let equal = b"\x01win-equal-boundary".to_vec();
        let just_later = b"\x01win-nanosecond-later".to_vec();

        backend
            .window_stage(&[
                stages(&equal, at(1_500_000_000)),
                stages(&just_later, at(1_500_000_001)),
            ])
            .await
            .unwrap();

        assert_eq!(
            backend.window_take(at(1_500_000_000)).await.unwrap(),
            vec![equal.clone()],
            "visible_after == now is overdue (the `<=` boundary)"
        );
        assert_eq!(
            backend.window_take(at(1_500_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "the not-yet-overdue row stays (one nanosecond of slack)"
        );
        assert_eq!(
            backend.window_take(at(1_500_000_001)).await.unwrap(),
            vec![just_later.clone()],
            "the horizon itself moved, so the row is now overdue"
        );
    }

    /// Virgin-store shapes (the recorded read-only split): on a
    /// schema-less database `window_take` maps a missing table to empty
    /// (the read-tolerance parity with the postgres twin), while
    /// `window_stage` is a write and surfaces the engine error without
    /// silently creating schema. A constructed backend with an empty
    /// window table also takes empty.
    #[tokio::test]
    async fn window_virgin_store_shapes() {
        let Some(tableless) = harness_schemaless("window-virgin").await else {
            return;
        };
        let key = b"\x01virgin-window-key".to_vec();

        assert_eq!(
            tableless.window_take(at(1_000_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "schema-less take is a no-op read"
        );
        assert_eq!(
            tableless.window_take(at(1_000_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "still empty on a repeat"
        );
        match tableless.window_stage(&[stages(&key, at(1))]).await {
            Err(Error::Io(message)) => {
                assert!(
                    message.contains("does not exist"),
                    "the write surfaces the engine error, not a silent schema create: {message}"
                )
            }
            other => panic!("schema-less stage is the engine error, got {other:?}"),
        }

        let Some(backend) = harness_gc("window-empty").await else {
            return;
        };
        assert_eq!(
            backend.window_take(SystemTime::now()).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "empty window table takes empty"
        );
    }
}