From 51d8f6525fd9a87a66378c7672b4d9fd15497283 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Mon, 12 Oct 2026 04:39:45 +0000 Subject: [PATCH] =?UTF-8?q?feat(gc):=20the=20sqlite=20engine-state=20windo?= =?UTF-8?q?w=20family=20=E2=80=94=20the=20durable=20single-node=20delete?= =?UTF-8?q?=20window?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 | 378 ++++++++++++++++++++++++++++++++++ tasks/engine-sqlite-window.md | 72 ++++++- 2 files changed, 443 insertions(+), 7 deletions(-) diff --git a/src/sqlite.rs b/src/sqlite.rs index 3e5833a..5a23df2 100644 --- a/src/sqlite.rs +++ b/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, 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>, Error> { + let conn = Arc::clone(&self.conn); + let horizon = time_to_nanos(now); + blocking(move || { + let guard = conn.lock(); + let taken: rusqlite::Result>> = (|| { + 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>(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> { @@ -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, i64)> { + let conn = Arc::clone(&backend.conn); + blocking(move || { + let guard = conn.lock(); + let listed: rusqlite::Result, 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>(0)?, row.get::<_, i64>(1)?)) + })?; + let mut out: Vec<(Vec, i64)> = rows.collect::>()?; + 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::>::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::>::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::>::new(), + "tableless take is a no-op read" + ); + assert_eq!( + tableless.window_take(at(1_000_000_000)).await.unwrap(), + Vec::>::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::>::new(), + "empty window table takes empty" + ); + } } diff --git a/tasks/engine-sqlite-window.md b/tasks/engine-sqlite-window.md index 9b23c8a..058bc18 100644 --- a/tasks/engine-sqlite-window.md +++ b/tasks/engine-sqlite-window.md @@ -1,7 +1,7 @@ --- id: engine-sqlite-window name: sqlite engine-state window family — the durable single-node delete window -status: pending +status: completed depends_on: [engine-sqlite] scope: narrow risk: low @@ -57,17 +57,17 @@ this task ships the clauses and their engine-level contract tests. ## Acceptance Criteria -- [ ] The sqlite backend implements `EngineWindowState` — the window +- [x] The sqlite backend implements `EngineWindowState` — the window family only; no pin-clause or sweep-lock body exists anywhere in the sqlite engine -- [ ] `ensure_gc_schema` creates `gc_window` only; `gc_pins` is never +- [x] `ensure_gc_schema` creates `gc_window` only; `gc_pins` is never created by sqlite (contract-tested) -- [ ] `window_stage` is idempotent per `(key, deadline)`; a re-stage +- [x] `window_stage` is idempotent per `(key, deadline)`; a re-stage at a new deadline updates the row -- [ ] `window_take` returns exactly the overdue set (`<= now` +- [x] `window_take` returns exactly the overdue set (`<= now` boundary), removes it atomically, leaves future rows; empty on none/virgin -- [ ] Clause contract tests green; `cargo test` (default) green; +- [x] Clause contract tests green; `cargo test` (default) green; `cargo clippy --all-targets -- -D warnings` and `cargo fmt --check` green @@ -83,6 +83,64 @@ this task ships the clauses and their engine-level contract tests. > To be filled by implementation agent +- The `EngineWindowState` impl lives in `src/sqlite.rs` after the + `Backend` impl, as `impl EngineWindowState for SqliteBackend`. The + pin clauses and the sweep-lock body are absent by construction — + nothing of `EngineFleetState` is implementable or referenced on this + engine (ADR-012 §5's sqlite row: the single-flighter is the + in-process sweep lease on `GcHome`, a sqlite fleet is a recorded + non-option). +- **Timestamp encoding (recorded design point):** `visible_after` + travels as a signed i64 nanosecond count from the Unix epoch — + sub-second precision preserved for the test tunings, negatives for + pre-epoch instants, full i64 range. Helpers `time_to_nanos` / + `nanos_to_time` sit beside `size_u64` in `src/sqlite.rs`. +- **Staging discipline:** one `prepare_cached` upsert per row — + `INSERT … ON CONFLICT(key) DO UPDATE SET visible_after = + excluded.visible_after`. The row's primary key is the window + candidate's key, so a same-deadline re-stage is idempotent (one row, + unchanged value) and a new-deadline re-stage updates the row rather + than duplicating it. `INSERT OR IGNORE` alone would have made the + new-deadline update a no-op, so the upsert carries both clauses + of the idempotence contract. +- **Take atomicity:** `DELETE FROM gc_window WHERE visible_after <= ? + RETURNING key` — one statement on the single pooled connection, so + selection and removal are one atomic step (a row is never handed out + twice); `RETURNING` order is unspecified, so the returned keys are + sorted for determinism. +- **Virgin-store tolerance split per the sqlite engine's recorded + posture:** `window_take` is a read — a no-tables-yet database maps + to empty via the existing `is_no_such_table` tolerance; `window_stage` + is a write — a tableless database surfaces the engine error rather + than silently creating schema (the constructor-vs-lazy design point: + store construction always wires `ensure_gc_schema`, so the table + exists whenever the engine is in store service; the lazily staged + option was rejected as contradicting that recorded point). +- Schema timing unchanged from the pool table's recorded design point: + the backend constructor never creates `gc_window` itself — the store + construction call (`store-sqlite-construction`) wires + `ensure_gc_schema`, which is idempotent and creates `gc_window` + only. The clause-level tests use a `new_window_backend()` fixture + that mirrors that wiring. +- The boundary contract (`visible_after <= now`, inclusive) is + contract-tested, including the nanosecond slack row (visible at + `now + 1ns` stays). + ## Summary -> To be filled on completion \ No newline at end of file +> To be filled on completion + +The sqlite engine-state seam's window family landed: `impl +EngineWindowState for SqliteBackend` in `src/sqlite.rs` — +`ensure_gc_schema` (creates `gc_window` only, never `gc_pins`, +contract-tested through `sqlite_master`), `window_stage` (key-keyed +upsert; idempotent per `(key, deadline)`, new-deadline re-stage +updates the row), and `window_take` (`DELETE … RETURNING` — one atomic +overdue take with the `<= now` boundary, future rows left in place). +Signed-nano timestamp encoding recorded; virgin-store tolerance split +read/write per the sqlite engine's posture. Six clause contract tests +added (schema window-only, staging idempotence + deadline update, +overdue-exact take + future rows left, the `<=` boundary with +nanosecond slack, and both virgin shapes). Verification: `cargo test` +green (58 lib + 5 integration), `cargo clippy --all-targets -- -D +warnings` green, `cargo fmt --check` green. \ No newline at end of file