feat(gc): the sqlite engine-state window family — the durable single-node delete window

- impl EngineWindowState for SqliteBackend, the window family only per
  ADR-012 §5's sqlite row; no pin-clause or sweep-lock body exists —
  EngineFleetState is never implemented on this engine
- ensure_gc_schema creates gc_window only (never gc_pins; the embedder's
  liveness table is never created — ADR-008 schema ownership)
- window_stage is a key-keyed upsert: idempotent per (key, deadline),
  a re-stage at a new deadline updates the row
- window_take is one atomic DELETE … RETURNING with the <= now
  boundary; future rows left in place; tableless take is a no-op read,
  tableless stage surfaces the engine error (the recorded read/write
  tolerance split)
- visible_after encodes as signed i64 nanoseconds from the epoch
  (recorded design point; sub-second precision for test tunings)
- six clause contract tests per the ADR-012 §5 sqlite row

Verification: cargo test green (58 lib + 5 integration);
cargo clippy --all-targets -- -D warnings green; cargo fmt --check green.

src/sqlite.rs

@@ -51,11 +51,13 @@
 
 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};
 use crate::Error;
 
 fn db_err(e: rusqlite::Error) -> Error {
@@ -76,6 +78,25 @@ 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
@@ -418,6 +439,110 @@ impl Backend for SqliteBackend {
     }
 }
 
+/// 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. 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.
+    async fn window_stage(&self, candidates: &[WindowCandidate]) -> Result<(), Error> {
+        if candidates.is_empty() {
+            return Ok(());
+        }
+        let conn = Arc::clone(&self.conn);
+        let rows: Vec<(Vec<u8>, i64)> = candidates
+            .iter()
+            .map(|c| (c.key.clone(), time_to_nanos(c.visible_after)))
+            .collect();
+        blocking(move || {
+            let guard = conn.lock();
+            let written: rusqlite::Result<()> = (|| {
+                let mut stmt = guard.prepare_cached(
+                    "INSERT INTO gc_window (key, visible_after) VALUES (?1, ?2) \
+                     ON CONFLICT(key) DO UPDATE SET visible_after = excluded.visible_after",
+                )?;
+                for (key, visible_after) in &rows {
+                    stmt.execute(params![key, visible_after])?;
+                }
+                Ok(())
+            })();
+            written.map_err(db_err)
+        })
+        .await
+    }
+
+    /// Take overdue candidates (`visible_after <= now`) — the boundary
+    /// is `<=`: a candidate visible exactly at `now` is overdue. One
+    /// atomic step: `DELETE … RETURNING` selects and removes the rows in
+    /// a single statement on the single connection, 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 visible_after <= ?1 RETURNING key",
+                )?;
+                let rows = stmt.query_map(params![horizon], |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::*;
@@ -470,6 +595,16 @@ mod tests {
         (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>> {
@@ -887,4 +1022,247 @@ mod tests {
             "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).
+    async fn table_exists(backend: &SqliteBackend, name: &str) -> bool {
+        let conn = Arc::clone(&backend.conn);
+        let name = name.to_owned();
+        blocking(move || {
+            let guard = conn.lock();
+            Ok(guard
+                .query_row(
+                    "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?1",
+                    params![name],
+                    |row| row.get::<_, i64>(0),
+                )
+                .unwrap()
+                != 0)
+        })
+        .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 a re-stage
+    /// at a new deadline updates the row (upsert on the key) rather
+    /// than duplicating it.
+    #[tokio::test]
+    async fn window_stage_is_idempotent_and_restage_updates_the_deadline() {
+        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 new deadline updates the row"
+        );
+
+        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"
+        );
+    }
+
+    /// 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"
+        );
+    }
 }