alkblobs 0.2.2

//! The `sqlite` kv-tier engine (ADR-003/007): WAL + `synchronous=NORMAL`
//! engaged at open — the shipped durability tier (ADR-003; honker ships
//! the posture and iroh's store measured it) — and `busy_timeout = 5000`
//! pinned at open — the shipped contention posture (review-002 C-3: a
//! pinned constant matching the family's explicit 5000 ms, not the
//! rusqlite-default coincidence; DQ-7's co-tenancy evidence shows no
//! contention observable at POC scale, so the value is contract, not a
//! tuned knob) — `prepare_cached` statement discipline, the
//! materializing read-cursor body, and the degenerate in-memory stage.
//! Feature `sqlite`, **default-on** (ADR-013 §4): the
//! default build's kv engine and the crate's first SQL-backed engine.
//!
//! **Schema (POC #5 finding B5's row shape adapted to sqlite):**
//! `blobs (key BLOB PRIMARY KEY, value BLOB, size INTEGER NOT NULL)`.
//! The dedicated `size` column is what makes `Backend::size`
//! metadata-only — `SELECT size … WHERE key = ?` rides the key's index
//! and never loads the value blob (ADR-008's size discipline;
//! plan-tested below). `has` is an EXISTS probe, never a value load.
//! `put` is CAS first-writer-wins (`INSERT OR IGNORE`): a re-put of an
//! existing key is dedup, never an overwrite — on the whole-value put
//! and the staged commit alike.
//!
//! **Schema-creation timing (recorded design point):** the constructor
//! runs `CREATE TABLE IF NOT EXISTS` at open. The pool table then exists
//! from construction on — `put` needs no lazy ensure-schema path, and
//! the pragma posture is applied exactly once before any statement runs.
//! The virgin-store tolerance itself ("read paths treat a missing table
//! as absent/empty", POC #1 finding 2) is an engine-level property of
//! the read paths independent of that timing — unit-tested against a
//! tableless database file through the test-only constructor below.
//! The tolerance is deliberately read-only: a write against a tableless
//! database surfaces the engine error rather than silently creating
//! schema.
//!
//! **Single-writer posture (recorded design point):** one connection
//! behind a short-held [`parking_lot::Mutex`], shared by the backend and
//! every staging cursor it hands out (a staging cursor sees the same
//! table through the clone — `put_staged`'s out-of-band commit requires
//! the shared handle). A pool is not required solo. rusqlite is
//! synchronous, so every engine I/O operation — constructor, all seven
//! methods, the staged commit — runs in [`tokio::task::spawn_blocking`]
//! (AGENTS convention 3; store-api.md §Concurrency posture); the cursor
//! and staging read bodies are in-memory after materialization and need
//! no blocking. The WAL single-writer ceiling (~1.2k objects/s under
//! contention, POC #5 B3) is a **documented posture, not a defect** —
//! REQ-1's economics put single-node push rates an order below it.
//!
//! The constructor errors when WAL cannot be engaged (in-memory
//! databases included): the durability promise is verified at open, not
//! assumed.
//!
//! The engine is constructed by the store facade —
//! `dual_tier_sqlite_local` / `kv_only_sqlite` (the wave-3 constructor
//! wiring, landed) — which also runs the GC schema's engine-neutral
//! call (`ensure_gc_schema`, creating `gc_window`). The
//! `#![allow(dead_code)]` below still has real scope: the
//! `nanos_to_time` i64-nanos decode helper (the row timestamps are
//! metadata the lib surface never decodes — its only current callers
//! are the test instruments that read `committed_at`-shaped columns),
//! plus the feature-subset shapes the cfg-gates leave without
//! consumers.

#![allow(dead_code)]

use std::path::PathBuf;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use parking_lot::Mutex;
use rusqlite::{params, Connection};

use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
use crate::gc::{EngineWindowState, WindowCandidate, WINDOW_TAKE_BATCH};
use crate::Error;

fn db_err(e: rusqlite::Error) -> Error {
    Error::Io(format!("sqlite: {e}"))
}

/// The virgin-store tolerance's discriminant (POC #1 finding 2): a
/// statement naming a table that does not exist maps on the read paths
/// to absent/empty, never an error. rusqlite carries sqlite's message
/// through its Display, so the check is stringly by nature — the stable
/// surface sqlite itself guarantees.
fn is_no_such_table(e: &rusqlite::Error) -> bool {
    e.to_string().contains("no such table")
}

/// The `size` row's integer, clamped to the domain the trait probes in.
fn size_u64(raw: i64) -> u64 {
    raw.max(0) as u64
}

/// The window row's timestamp encoding (the recorded design point):
/// signed nanoseconds since the Unix epoch — sub-second precision for
/// the test tunings, negative for pre-epoch instants, i64 across the
/// full range sqlite stores an INTEGER (±292 years around the epoch).
fn time_to_nanos(t: SystemTime) -> i64 {
    match t.duration_since(UNIX_EPOCH) {
        Ok(d) => d.as_nanos() as i64,
        Err(e) => -(e.duration().as_nanos() as i64),
    }
}

fn nanos_to_time(n: i64) -> SystemTime {
    if n >= 0 {
        UNIX_EPOCH + std::time::Duration::from_nanos(n as u64)
    } else {
        UNIX_EPOCH - std::time::Duration::from_nanos((-n) as u64)
    }
}

/// Run a synchronous rusqlite body off the executor — the engine's one
/// blocking discipline (rusqlite is synchronous; store-api.md §Concurrency
/// posture). Every engine I/O operation below routes through this; the
/// in-memory cursor/staging read bodies do not.
async fn blocking<F, T>(body: F) -> Result<T, Error>
where
    F: FnOnce() -> Result<T, Error> + Send + 'static,
    T: Send + 'static,
{
    match tokio::task::spawn_blocking(body).await {
        Ok(result) => result,
        Err(e) => Err(Error::Io(format!("blocking task joint failure: {e}"))),
    }
}

/// The `sqlite` engine — one connection behind a short-held
/// [`parking_lot::Mutex`] (parking_lot does not poison, so there is no
/// `unwrap_or_else(|e| e.into_inner())` recovery posture to apply).
#[derive(Clone)]
pub(crate) struct SqliteBackend {
    /// The single connection (the recorded single-writer posture),
    /// shared with every staging cursor handout through the Arc.
    conn: Arc<Mutex<Connection>>,
    /// Test-only commit-failure injection, standing in for the media
    /// failure the sqlite tier cannot reliably produce in tests; mirrors
    /// the mem engine's knob so the stage-hygiene failure form is
    /// exercised against the real engine.
    #[cfg(test)]
    fail_commit: bool,
    /// Test-only mid-loop stage-failure injection (the transactional
    /// staging test's instrument — the failure lands after prior
    /// candidates' upserts have run inside the open transaction).
    #[cfg(test)]
    fail_window_stage: bool,
}

impl SqliteBackend {
    /// Open (or create) the pool database file and set the shipped
    /// durability posture: `journal_mode=WAL` + `synchronous=NORMAL`,
    /// and the shipped contention posture: `busy_timeout=5000` (the
    /// pinned constant — the family's explicit 5000 ms, alkstore-sqlite
    /// pins the same; not a knob — DQ-7's co-tenancy evidence shows no
    /// contention observable at POC scale, so there is no tuning story
    /// yet and the value is contract, not a tuning surface). Then
    /// `CREATE TABLE IF NOT EXISTS` for the pool table. WAL engagement
    /// is verified at open (the pragma returns the mode it actually
    /// set); a database that cannot enter WAL is a construction error,
    /// not a silent posture downgrade.
    pub(crate) async fn new(path: impl Into<PathBuf>) -> Result<Self, Error> {
        let path: Arc<PathBuf> = Arc::new(path.into());
        let conn = blocking(move || {
            let conn = Connection::open(&*path).map_err(db_err)?;
            let mode: String = conn
                .query_row("PRAGMA journal_mode = WAL", [], |row| row.get(0))
                .map_err(db_err)?;
            if !mode.eq_ignore_ascii_case("wal") {
                return Err(Error::Io(format!(
                    "sqlite: journal_mode is {mode}, WAL unavailable"
                )));
            }
            conn.execute_batch(
                "PRAGMA synchronous = NORMAL;
                 PRAGMA busy_timeout = 5000;",
            )
            .map_err(db_err)?;
            conn.execute_batch(
                "CREATE TABLE IF NOT EXISTS blobs (
                    key BLOB PRIMARY KEY,
                    value BLOB,
                    size INTEGER NOT NULL
                );",
            )
            .map_err(db_err)?;
            Ok(conn)
        })
        .await?;
        Ok(Self {
            conn: Arc::new(Mutex::new(conn)),
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_window_stage: false,
        })
    }

    /// Open a database file without schema creation — the virgin-store
    /// tolerance test's instrument (constructor-vs-lazy is a recorded
    /// design point; the read paths' no-tables-yet semantics are tested
    /// against a genuinely tableless database file either way).
    #[cfg(test)]
    async fn new_tableless(path: impl Into<PathBuf>) -> Result<Self, Error> {
        let path: Arc<PathBuf> = Arc::new(path.into());
        let conn = blocking(move || Connection::open(&*path).map_err(db_err)).await?;
        Ok(Self {
            conn: Arc::new(Mutex::new(conn)),
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_window_stage: false,
        })
    }

    /// The sqlite_master table-existence probe — test-only: the store
    /// construction test's instrument (covers every schema object,
    /// not just probed reads) and the local tests' helper body.
    #[cfg(test)]
    pub(crate) async fn table_exists(&self, name: &str) -> Result<bool, Error> {
        let conn = Arc::clone(&self.conn);
        let name = name.to_owned();
        blocking(move || {
            let guard = conn.lock();
            let probed: rusqlite::Result<bool> = guard
                .query_row(
                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
                    params![name],
                    |row| row.get::<_, i64>(0),
                )
                .map(|count| count != 0);
            probed.map_err(db_err)
        })
        .await
    }

    /// The connection's live pragma value — the posture test's probe
    /// (`synchronous` is per-connection, so the backend's own connection
    /// must be asked).
    #[cfg(test)]
    async fn pragma(&self, name: &str) -> Result<rusqlite::types::Value, Error> {
        let conn = Arc::clone(&self.conn);
        let name = name.to_owned();
        blocking(move || {
            let guard = conn.lock();
            let value: rusqlite::types::Value = guard
                .query_row(&format!("PRAGMA {name}"), [], |row| row.get(0))
                .map_err(db_err)?;
            Ok(value)
        })
        .await
    }
}

/// The materializing read cursor (ADR-010 §1's kv body family): the whole
/// stored value in memory — `get` loads the row once (the kv tier holds
/// only ≤ threshold content); `len()` reads the row's `size` column, so
/// the cursor's length claim is the same indexed fact [`Backend::size`]
/// probes; range reads are in-memory slices with the cursor contract's
/// EOF clamping. This body is also what the facade's verification-on-get
/// re-derives over (no engine change needed).
pub(crate) struct SqliteCursor {
    value: Vec<u8>,
    len: u64,
}

impl ReadCursor for SqliteCursor {
    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.value.len());
        Ok(self.value[start..end].to_vec())
    }
}

/// The degenerate in-memory staging cursor (ADR-010 §1): appends extend
/// the buffer; `commit(key)` publishes the whole staged bytes through the
/// same `INSERT OR IGNORE` clause as [`Backend::put`] — under the
/// store-core-resolved canonical key (the staged-put seam's key-at-commit
/// design point), first writer wins, losing the race is dedup (`Ok`),
/// never a failure — and every non-commit path, including the failure
/// form, converges on zero residue. `commit` takes the buffer via
/// `std::mem::take` before the connection lock, so a failed publish
/// cannot strand the staged bytes anywhere.
pub(crate) struct SqliteStaging {
    staged: Vec<u8>,
    conn: Arc<Mutex<Connection>>,
    /// Test-only injection, shared with the issuing [`SqliteBackend`] —
    /// see its field note.
    #[cfg(test)]
    fail_commit: bool,
}

impl ReadCursor for SqliteStaging {
    fn len(&self) -> u64 {
        self.staged.len() as u64
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(offset, len, self.staged.len());
        Ok(self.staged[start..end].to_vec())
    }
}

impl StagingCursor for SqliteStaging {
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        self.staged.extend_from_slice(chunk);
        Ok(())
    }

    async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
        let value = std::mem::take(&mut self.staged);
        let conn = Arc::clone(&self.conn);
        let key = key.to_vec();
        #[cfg(test)]
        let fail_commit = self.fail_commit;
        blocking(move || {
            #[cfg(test)]
            if fail_commit {
                return Err(Error::Io(
                    "engine commit failure (sqlite, injected)".to_owned(),
                ));
            }
            let guard = conn.lock();
            let size = value.len() as i64;
            let written: rusqlite::Result<()> = (|| {
                let mut stmt = guard.prepare_cached(
                    "INSERT OR IGNORE INTO blobs (key, value, size) VALUES (?1, ?2, ?3)",
                )?;
                stmt.execute(params![key, value, size])?;
                Ok(())
            })();
            written.map_err(db_err)
        })
        .await
    }

    async fn discard(mut self) {
        self.staged.clear();
    }
}

impl Backend for SqliteBackend {
    type Cursor = SqliteCursor;
    type Staging = SqliteStaging;

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

    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        let conn = Arc::clone(&self.conn);
        let key = key.to_vec();
        blocking(move || {
            let guard = conn.lock();
            let probed: rusqlite::Result<bool> = (|| {
                let mut stmt =
                    guard.prepare_cached("SELECT EXISTS(SELECT 1 FROM blobs WHERE key = ?1)")?;
                Ok(stmt.query_row(params![key], |row| row.get::<_, i64>(0))? != 0)
            })();
            match probed {
                Ok(present) => Ok(present),
                Err(e) if is_no_such_table(&e) => Ok(false),
                Err(e) => Err(db_err(e)),
            }
        })
        .await
    }

    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
        let conn = Arc::clone(&self.conn);
        let key = key.to_vec();
        blocking(move || {
            let guard = conn.lock();
            let loaded: rusqlite::Result<Option<SqliteCursor>> = (|| {
                let mut stmt =
                    guard.prepare_cached("SELECT value, size FROM blobs WHERE key = ?1")?;
                let row = stmt.query_row(params![key], |row| {
                    Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?))
                });
                match row {
                    Ok((value, size)) => Ok(Some(SqliteCursor {
                        value,
                        len: size_u64(size),
                    })),
                    Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
                    Err(e) => Err(e),
                }
            })();
            match loaded {
                Ok(cursor) => Ok(cursor),
                Err(e) if is_no_such_table(&e) => Ok(None),
                Err(e) => Err(db_err(e)),
            }
        })
        .await
    }

    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
        let conn = Arc::clone(&self.conn);
        let key = key.to_vec();
        let value = value.to_vec();
        let size = value.len() as i64;
        blocking(move || {
            let guard = conn.lock();
            let written: rusqlite::Result<()> = (|| {
                let mut stmt = guard.prepare_cached(
                    "INSERT OR IGNORE INTO blobs (key, value, size) VALUES (?1, ?2, ?3)",
                )?;
                stmt.execute(params![key, value, size])?;
                Ok(())
            })();
            written.map_err(db_err)
        })
        .await
    }

    async fn put_staged(&self) -> Result<Self::Staging, Error> {
        Ok(SqliteStaging {
            staged: Vec::new(),
            conn: Arc::clone(&self.conn),
            #[cfg(test)]
            fail_commit: self.fail_commit,
        })
    }

    async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        let conn = Arc::clone(&self.conn);
        let key = key.to_vec();
        blocking(move || {
            let guard = conn.lock();
            let removed: rusqlite::Result<usize> = (|| {
                let mut stmt = guard.prepare_cached("DELETE FROM blobs WHERE key = ?1")?;
                stmt.execute(params![key])
            })();
            match removed {
                Ok(0) => Err(Error::Missing),
                Ok(_) => Ok(()),
                Err(e) if is_no_such_table(&e) => Err(Error::Missing),
                Err(e) => Err(db_err(e)),
            }
        })
        .await
    }

    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        let conn = Arc::clone(&self.conn);
        blocking(move || {
            let guard = conn.lock();
            let listed: rusqlite::Result<Vec<Vec<u8>>> = (|| {
                let mut stmt = guard.prepare_cached("SELECT key FROM blobs")?;
                let rows = stmt.query_map([], |row| row.get::<_, Vec<u8>>(0))?;
                let mut keys = Vec::new();
                for row in rows {
                    keys.push(row?);
                }
                Ok(keys)
            })();
            match listed {
                Ok(keys) => Ok(keys),
                Err(e) if is_no_such_table(&e) => Ok(Vec::new()),
                Err(e) => Err(db_err(e)),
            }
        })
        .await
    }

    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        let conn = Arc::clone(&self.conn);
        let key = key.to_vec();
        blocking(move || {
            let guard = conn.lock();
            let probed: rusqlite::Result<Option<i64>> = (|| {
                let mut stmt = guard.prepare_cached("SELECT size FROM blobs WHERE key = ?1")?;
                let row = stmt.query_row(params![key], |row| row.get::<_, i64>(0));
                match row {
                    Ok(size) => Ok(Some(size)),
                    Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
                    Err(e) => Err(e),
                }
            })();
            match probed {
                Ok(size) => Ok(size.map(size_u64)),
                Err(e) if is_no_such_table(&e) => Ok(None),
                Err(e) => Err(db_err(e)),
            }
        })
        .await
    }
}

/// The engine-state companion seam's window family (ADR-010 §2 as
/// reduced by ADR-012 §5's table — sqlite row: **window clauses only**).
/// The pin clauses and the sweep-lock body are dropped dead machinery
/// here: the single-flighter is the in-process lock (GcHome's sweep
/// lease) and a sqlite fleet is a recorded non-option (ADR-010 §3), so
/// `EngineFleetState` is never implemented on this engine.
///
/// **Virgin-store tolerance, per the sqlite engine's recorded posture:**
/// `window_take` is a read — a no-tables-yet database maps to empty,
/// like every read path. `window_stage` is a write — a no-tables-yet
/// database surfaces the engine error rather than silently creating
/// schema (the constructor-vs-lazy design point above; a tableless
/// database only exists through the test instrument, since store
/// construction always calls [`EngineWindowState::ensure_gc_schema`]).
impl EngineWindowState for SqliteBackend {
    /// Create `gc_window (key BLOB PRIMARY KEY, visible_after INTEGER)`
    /// — this engine's applicability row is the window table only;
    /// `gc_pins` is never created here (ADR-012 §5). The embedder's
    /// liveness table is never created in any case (ADR-008's
    /// schema-ownership rule — the sweep reads it and never owns it).
    /// Idempotent, so the twice-at-open call form is a no-op.
    async fn ensure_gc_schema(&self) -> Result<(), Error> {
        let conn = Arc::clone(&self.conn);
        blocking(move || {
            let guard = conn.lock();
            guard
                .execute_batch(
                    "CREATE TABLE IF NOT EXISTS gc_window (
                        key BLOB PRIMARY KEY,
                        visible_after INTEGER
                    );",
                )
                .map_err(db_err)
        })
        .await
    }

    /// 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 transaction back, leaving
    /// zero staged rows. The body's discipline: an upsert keyed on the
    /// row's primary key — a re-stage of the same key at the same deadline
    /// is idempotent (one row, value unchanged), and 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 (sqlite's `max` scalar carries the
    /// seam's `GREATEST` semantics).
    async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error> {
        if candidates.is_empty() {
            return Ok(());
        }
        let conn = Arc::clone(&self.conn);
        #[cfg(test)]
        let fail_window_stage = self.fail_window_stage;
        let rows: Vec<(Vec<u8>, i64)> = candidates
            .iter()
            .map(|c| (c.key.clone(), time_to_nanos(c.visible_after)))
            .collect();
        blocking(move || {
            let mut guard = conn.lock();
            let staged: Result<(), Error> = (|| {
                let tx = guard.transaction().map_err(db_err)?;
                {
                    let mut stmt = tx
                        .prepare_cached(
                            "INSERT INTO gc_window (key, visible_after) VALUES (?1, ?2) \
                             ON CONFLICT(key) DO UPDATE SET \
                             visible_after = max(visible_after, excluded.visible_after)",
                        )
                        .map_err(db_err)?;
                    for (key, visible_after) in rows.iter() {
                        #[cfg(test)]
                        if fail_window_stage && rows.last().is_some_and(|(last, _)| last == key) {
                            return Err(Error::Io(
                                "sqlite: window staging failure (mid-loop, injected)".to_owned(),
                            ));
                        }
                        stmt.execute(params![key, visible_after]).map_err(db_err)?;
                    }
                }
                tx.commit().map_err(db_err)
            })();
            staged
        })
        .await
    }

    /// Take overdue candidates (`visible_after <= now`) — the boundary
    /// is `<=`: a candidate visible exactly at `now` is overdue. The take
    /// is **bounded** to [`WINDOW_TAKE_BATCH`] rows (review-001 N-3):
    /// oldest deadlines first (`ORDER BY visible_after`), so an enormous
    /// overdue backlog never materializes entirely in memory in one call —
    /// the calling sweep executor consumes the take in a loop until
    /// drained. One atomic step, unchanged: the single `DELETE … RETURNING`
    /// statement selects and removes its rows together on the single
    /// connection (the `LIMIT` rides the inner candidate query), so a row
    /// is never handed out twice and an aborted body leaves no partial
    /// take. The returned keys go back to the calling sweep to
    /// re-arbitrate and delete (ADR-012 §6.7 — the executor is the next
    /// sweep).
    async fn window_take(&self, now: SystemTime) -> Result<Vec<Vec<u8>>, Error> {
        let conn = Arc::clone(&self.conn);
        let horizon = time_to_nanos(now);
        blocking(move || {
            let guard = conn.lock();
            let taken: rusqlite::Result<Vec<Vec<u8>>> = (|| {
                let mut stmt = guard.prepare_cached(
                    "DELETE FROM gc_window WHERE key IN ( \
                         SELECT key FROM gc_window WHERE visible_after <= ?1 \
                         ORDER BY visible_after LIMIT ?2 \
                     ) RETURNING key",
                )?;
                let rows = stmt.query_map(params![horizon, WINDOW_TAKE_BATCH as i64], |row| {
                    row.get::<_, Vec<u8>>(0)
                })?;
                let mut keys = Vec::new();
                for row in rows {
                    keys.push(row?);
                }
                Ok(keys)
            })();
            match taken {
                Ok(mut keys) => {
                    keys.sort();
                    Ok(keys)
                }
                Err(e) if is_no_such_table(&e) => Ok(Vec::new()),
                Err(e) => Err(db_err(e)),
            }
        })
        .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::Path;
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};

    const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";

    /// A fresh temp database file per test; Drop removes the file and
    /// its WAL sidecars idempotently.
    struct TempDb(PathBuf);

    fn temp_db(tag: &str) -> TempDb {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let path = std::env::temp_dir().join(format!(
            "alkblobs-sqlite-{tag}-{}-{nanos:x}-{seq}.db",
            std::process::id()
        ));
        TempDb(path)
    }

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

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

    impl Drop for TempDb {
        fn drop(&mut self) {
            self.drop_sidecar("-wal");
            self.drop_sidecar("-shm");
            let _ = std::fs::remove_file(&self.0);
        }
    }

    async fn new_backend() -> (TempDb, SqliteBackend) {
        let db = temp_db("engine");
        let backend = SqliteBackend::new(db.path()).await.unwrap();
        (db, backend)
    }

    /// A constructed backend with the GC window table ensured — the
    /// clause tests' fixture (store construction always wires the
    /// ensure call; the backend constructor itself never creates the
    /// window table).
    async fn new_window_backend() -> (TempDb, SqliteBackend) {
        let (db, backend) = new_backend().await;
        backend.ensure_gc_schema().await.unwrap();
        (db, backend)
    }

    /// The real stored value, observed through the read seam — the
    /// mirror of the mem engine's helper.
    async fn stored_value(backend: &SqliteBackend, key: &[u8]) -> Option<Vec<u8>> {
        let cursor = backend.get(key).await.expect("get succeeds")?;
        let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read");
        Some(bytes)
    }

    async fn staged(backend: &SqliteBackend, key: &[u8], content: &[u8]) -> Result<(), Error> {
        let mut staging = backend.put_staged().await?;
        staging.append(content).await?;
        staging.commit(key).await
    }

    /// 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 mem engine's suite
    /// (every other engine's conformance column mirrors the reference).
    #[tokio::test]
    async fn backend_method_contract() {
        let (_db, backend) = new_backend().await;
        let key = b"\x01opaque-key-bytes".to_vec();
        assert_eq!(backend.name(), "sqlite", "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"
        );

        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"
        );

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

    /// The `list` contract is complete AND well-formed: every key in the
    /// table, no values leaked, asserted order-insensitively against the
    /// put set plus a round-trip through `has`.
    #[tokio::test]
    async fn list_is_complete_and_well_formed() {
        let (_db, backend) = new_backend().await;
        let keys = [
            b"\x02b-key".to_vec(),
            b"\x02a-key".to_vec(),
            b"\x02c-key".to_vec(),
        ];
        for key in &keys {
            backend.put(key, b"value").await.unwrap();
        }
        let mut listed = backend.list().await.unwrap();
        let mut expected = keys.to_vec();
        listed.sort();
        expected.sort();
        assert_eq!(listed, expected, "list is complete and well-formed");
        for key in &expected {
            assert!(
                backend.has(key).await.unwrap(),
                "no spurious or mangled keys"
            );
        }
    }

    /// The materializing cursor body's read contract: `len()` exact (the
    /// row's size column), EOF-clamped reads (over-read, tail, at-EOF
    /// empty, zero-length), and the ADR-006 range seam's digest over
    /// exactly the returned bytes.
    #[tokio::test]
    async fn cursor_contract_materialized_body() {
        let (_db, backend) = new_backend().await;
        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 entry length");

        assert_eq!(cursor.read_at(0, len).await.unwrap(), CONTENT, "full read");
        assert_eq!(
            cursor.read_at(0, len + 17).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[mid as usize..],
            "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(0, 0).await.unwrap(),
            Vec::<u8>::new(),
            "zero-length read"
        );

        let (slice, digest) = cursor.read_range(mid, 7).await.unwrap();
        assert_eq!(slice, &CONTENT[mid as usize..(mid as usize + 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). A value of zero bytes is an empty BLOB, never a NULL.
    #[tokio::test]
    async fn empty_entry_round_trips() {
        let (_db, backend) = new_backend().await;
        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"
        );
    }

    /// The staged put's happy path: handout, appends, the verification
    /// pass reading the staged artifact back as a whole through the read
    /// seam, commit publishing through the same CAS clause, and the
    /// committed entry round-tripping through get's cursor — with both
    /// re-put forms (whole-value and staged) dedup no-ops against the
    /// live entry.
    #[tokio::test]
    async fn staged_put_round_trips_through_the_read_seam() {
        let (_db, backend) = new_backend().await;
        let key = b"\x01staged-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(&CONTENT[..8]).await.unwrap();
        staging.append(&CONTENT[8..16]).await.unwrap();
        staging.append(&CONTENT[16..]).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));

        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.unwrap();
        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"
        );
    }

    /// Stage hygiene, discard form: every non-commit path converges on
    /// discard with zero residue — nothing was staged into the table.
    #[tokio::test]
    async fn staged_discard_leaves_zero_residue() {
        let (_db, backend) = new_backend().await;
        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 table holds exactly zero entries after discard"
        );
    }

    /// Stage hygiene, failure form: a failed commit is already
    /// self-discarded — the buffer was taken before the connection lock,
    /// so 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 (_db, backend) = new_backend().await;
        let failing = SqliteBackend {
            conn: Arc::clone(&backend.conn),
            #[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 table itself"
        );
    }

    /// The pragma posture: the shipped durability tier engaged at open —
    /// journal_mode WAL (persistent) and synchronous NORMAL (1,
    /// per-connection) — and the shipped contention posture —
    /// busy_timeout 5000 ms (the pinned constant, review-002 C-3) —
    /// verified on the backend's own single connection.
    #[tokio::test]
    async fn opens_with_wal_synchronous_normal_and_busy_timeout() {
        let (_db, backend) = new_backend().await;
        match backend.pragma("journal_mode").await.unwrap() {
            rusqlite::types::Value::Text(mode) => {
                assert_eq!(mode.to_lowercase(), "wal", "journal_mode is WAL");
            }
            other => panic!("journal_mode probe returned {other:?}"),
        }
        match backend.pragma("synchronous").await.unwrap() {
            rusqlite::types::Value::Integer(sync) => {
                assert_eq!(sync, 1, "synchronous is NORMAL (1)");
            }
            other => panic!("synchronous probe returned {other:?}"),
        }
        match backend.pragma("busy_timeout").await.unwrap() {
            rusqlite::types::Value::Integer(timeout) => {
                assert_eq!(
                    timeout, 5000,
                    "busy_timeout is the pinned 5000 ms contention posture"
                );
            }
            other => panic!("busy_timeout probe returned {other:?}"),
        }
    }

    /// A database that cannot enter WAL is a construction error — the
    /// durability promise is verified at open, not assumed (in-memory
    /// databases cannot hold the WAL posture).
    #[tokio::test]
    async fn construction_rejects_a_database_that_cannot_hold_wal() {
        let opened = SqliteBackend::new(":memory:").await;
        assert!(
            opened.is_err(),
            "in-memory databases cannot hold the WAL posture"
        );
    }

    /// `size` reads the dedicated size column via the key's index, never
    /// the value blob (metadata-only, B5's row shape): the probe is exact
    /// for a large value and absent keys, and `EXPLAIN QUERY PLAN` proves
    /// the probe searches the key's index rather than scanning/loading.
    #[tokio::test]
    async fn size_probe_is_exact_and_rides_the_key_index() {
        let (_db, backend) = new_backend().await;
        let key = b"\x01size-probe-key".to_vec();
        let large_value = vec![0xAB_u8; 256 * 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 conn = Arc::clone(&backend.conn);
        let plan = blocking(move || {
            let guard = conn.lock();
            let planned: rusqlite::Result<Vec<String>> = (|| {
                let mut stmt = guard
                    .prepare_cached("EXPLAIN QUERY PLAN SELECT size FROM blobs WHERE key = ?1")?;
                let rows = stmt.query_map(params![key], |row| row.get::<_, String>(3))?;
                let mut details = Vec::new();
                for row in rows {
                    details.push(row?);
                }
                Ok(details)
            })();
            planned.map_err(db_err)
        })
        .await
        .unwrap();
        let joined = plan.join(" ");
        assert!(
            joined.contains("USING INDEX"),
            "the size probe searches the key's index: {joined}"
        );
    }

    /// Virgin-store no-tables-yet tolerance (POC #1 finding 2): every
    /// read path against a genuinely tableless database file is a
    /// no-op — absent/empty, never "table does not exist" — and delete
    /// reports Missing. The tolerance is read-only: a write surfaces the
    /// engine error and leaves zero residue.
    #[tokio::test]
    async fn virgin_store_no_tables_yet_reads_are_noops() {
        let db = temp_db("tableless");
        let backend = SqliteBackend::new_tableless(db.path()).await.unwrap();
        let key = b"\x01opaque-key".to_vec();

        assert!(
            !backend.has(&key).await.unwrap(),
            "has on a tableless db is a no-op read"
        );
        assert!(
            backend.get(&key).await.unwrap().is_none(),
            "get is a no-op read"
        );
        assert_eq!(
            backend.size(&key).await.unwrap(),
            None,
            "size is a no-op read"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "list is empty on a tableless db"
        );
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "delete on a tableless db reports Missing"
        );

        assert!(
            backend.put(&key, CONTENT).await.is_err(),
            "the tolerance is read-only: writes surface the engine error"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the failed write left zero residue"
        );
    }

    /// A fixed-instant deadline family for the window tests (epoch-based
    /// nanos, so pre-epoch coverage rides the same helpers): -10s (past,
    /// pre-epoch), +1s (past), +2s (future), +1h (far future).
    fn at(nanos: i64) -> SystemTime {
        nanos_to_time(nanos)
    }

    /// The gc_window rows as the engine stores them — the raw encoding,
    /// read back through the sqlite_master-visible table itself.
    async fn window_rows(backend: &SqliteBackend) -> Vec<(Vec<u8>, i64)> {
        let conn = Arc::clone(&backend.conn);
        blocking(move || {
            let guard = conn.lock();
            let listed: rusqlite::Result<Vec<(Vec<u8>, i64)>> = (|| {
                let mut stmt = guard
                    .prepare_cached("SELECT key, visible_after FROM gc_window")
                    .unwrap();
                let rows = stmt.query_map([], |row| {
                    Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, i64>(1)?))
                })?;
                let mut out: Vec<(Vec<u8>, i64)> = rows.collect::<Result<_, _>>()?;
                out.sort();
                Ok(out)
            })();
            listed.map_err(db_err)
        })
        .await
        .unwrap()
    }

    /// Table existence probed through sqlite_master — the schema test's
    /// instrument (covers every schema object, not just probed reads); a
    /// thin unwrap-wrap over the backend's helper.
    async fn table_exists(backend: &SqliteBackend, name: &str) -> bool {
        backend.table_exists(name).await.unwrap()
    }

    fn stages(key: &[u8], visible_after: SystemTime) -> WindowCandidate {
        WindowCandidate {
            key: key.to_vec(),
            visible_after,
        }
    }

    /// Schema contract: `ensure_gc_schema` creates `gc_window` only —
    /// `gc_pins` is never created by sqlite (ADR-012 §5's sqlite row),
    /// on the constructed backend and on a tableless database alike,
    /// and the call is idempotent (twice at open is a no-op).
    #[tokio::test]
    async fn gc_schema_creates_window_table_only() {
        let (_db, backend) = new_backend().await;
        backend.ensure_gc_schema().await.unwrap();
        backend.ensure_gc_schema().await.unwrap();
        assert!(table_exists(&backend, "gc_window").await, "window table");
        assert!(
            !table_exists(&backend, "gc_pins").await,
            "gc_pins is never created by sqlite (dead pin machinery dropped)"
        );

        let db = temp_db("tableless-window");
        let tableless = SqliteBackend::new_tableless(db.path()).await.unwrap();
        assert!(!table_exists(&tableless, "gc_window").await);
        tableless.ensure_gc_schema().await.unwrap();
        assert!(table_exists(&tableless, "gc_window").await);
        assert!(!table_exists(&tableless, "gc_pins").await);
    }

    /// 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** (the
    /// seam doc's corrected semantics, 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 (_db, backend) = new_window_backend().await;
        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 (_db, mut backend) = new_window_backend().await;
        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 (_db, backend) = new_window_backend().await;
        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 (_db, backend) = new_window_backend().await;
        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 (_db, backend) = new_window_backend().await;
        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: on a no-tables-yet database `window_take`
    /// is a no-op read (empty, never an error — the sqlite read
    /// tolerance carries), while `window_stage` is a write and surfaces
    /// the engine error without silently creating schema (the
    /// constructor-vs-lazy design point; store construction always runs
    /// `ensure_gc_schema`). A freshly constructed backend with an empty
    /// window table also takes empty.
    #[tokio::test]
    async fn window_virgin_store_shapes() {
        let db = temp_db("tableless-window-take");
        let tableless = SqliteBackend::new_tableless(db.path()).await.unwrap();
        let key = b"\x01virgin-window-key".to_vec();

        assert_eq!(
            tableless.window_take(at(1_000_000_000)).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "tableless 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"
        );
        let err = tableless.window_stage(&[stages(&key, at(1))]).await;
        match err {
            Err(Error::Io(message)) => {
                assert!(
                    message.contains("no such table"),
                    "the write surfaces the engine error, not a silent schema create"
                )
            }
            other => panic!("tableless stage is the engine error, got {other:?}"),
        }

        let (_db, backend) = new_window_backend().await;
        assert_eq!(
            backend.window_take(SystemTime::now()).await.unwrap(),
            Vec::<Vec<u8>>::new(),
            "empty window table takes empty"
        );
    }
}