From b28173c2edb3eba16032ca8649a77d849c74f4c1 Mon Sep 17 00:00:00 2001 From: Albert Hui Date: Sun, 9 Aug 2026 14:41:31 +0000 Subject: [PATCH] fix(test): acknowledge sqlite3 statements instead of sleeping on them The two gold-oracle fixtures drove a long-lived `sqlite3` reader process while separate short-lived writers mutated the same database, and synchronised the two with `sleep(400ms)`. That is a guess about how fast the host is, not a guarantee: `CREATE TABLE t` executes asynchronously in the reader process, and when a writer wins the race it fails with sqlite3 writer failed: Error: in prepare, no such table: t which is what CI hit -- a failure with nothing to do with the code under test, appearing only on slower or busier machines. HeldReader::run appends a unique sentinel to each statement and blocks until it appears on the reader's stdout. sqlite3 executes a script strictly in order on one connection, so the sentinel cannot print before the preceding statements have run: the acknowledgement follows from ordering, not from timing. A 30s timeout bounds it and panics with the offending SQL rather than hanging. Verified in this order: 1. reproduced deterministically -- sleeps set to 0 fail 3/3 with the CI error 2. handshake applied, sleeps removed entirely -- green 3. 15 consecutive runs -- 15 pass, 0 fail 4. control A: delete the CREATE TABLE -> FAILS loudly, `no such table: t` 5. control B: sentinel that can never match -> panics in 3.02s, bounded, no hang 6. restored -> green Both controls asserted their target text was present before mutating, so neither could silently no-op and report a green that tested nothing. Applied to wal_snapshot_oracle.rs as well, which carried the identical pattern -- a correction that does not reach its copies is barely a correction. Both now share core/tests/common/mod.rs. Side effect worth noting: row_history_oracle 0.83s -> 0.04s and wal_snapshot_oracle ~0.9s -> 0.07s. The 800ms per fixture was pure waiting. --- core/tests/common/mod.rs | 120 ++++++++++++++++++++++++++++++ core/tests/row_history_oracle.rs | 50 ++++--------- core/tests/wal_snapshot_oracle.rs | 49 ++++-------- 3 files changed, 147 insertions(+), 72 deletions(-) create mode 100644 core/tests/common/mod.rs diff --git a/core/tests/common/mod.rs b/core/tests/common/mod.rs new file mode 100644 index 0000000..ebcd0c8 --- /dev/null +++ b/core/tests/common/mod.rs @@ -0,0 +1,120 @@ +//! Shared helpers for the real-`sqlite3` gold-oracle tests. +//! +//! These fixtures drive a **long-lived `sqlite3` reader process** (held open so +//! the `-wal` sidecar is retained) while separate short-lived writer processes +//! mutate the same database. Statements sent to the reader execute +//! asynchronously in that other process, so the fixture must know when they +//! have actually run before the writers touch the schema they create. +//! +//! The obvious way to "know" is a `sleep`, and it is wrong: it encodes a guess +//! about how fast the host is. Losing that race produces +//! `Error: in prepare, no such table: t` from the writer — a failure with +//! nothing to do with the code under test, appearing only on slower or busier +//! machines. [`HeldReader::run`] replaces the guess with an acknowledgement. + +#![allow(dead_code)] + +use std::io::{BufRead, BufReader, Write}; +use std::path::Path; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::sync::mpsc::{self, Receiver, RecvTimeoutError}; +use std::time::Duration; + +/// Upper bound on how long to wait for one statement to be acknowledged. +/// +/// Generous on purpose: this bounds a hang, it does not pace the test. The +/// happy path returns as soon as the sentinel arrives, typically in +/// microseconds, so a large value costs nothing when things work. +const ACK_TIMEOUT: Duration = Duration::from_secs(30); + +/// A `sqlite3` reader process whose statements are acknowledged before the +/// caller proceeds. +pub struct HeldReader { + child: Child, + stdin: Option, + lines: Receiver, + seq: u32, +} + +impl HeldReader { + /// Spawn `bin` against `db`, with stdout piped so statements can be + /// acknowledged. + pub fn spawn(bin: &str, db: &Path) -> Self { + let mut child = Command::new(bin) + .arg(db) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let stdin = child.stdin.take().unwrap(); + let stdout = child.stdout.take().unwrap(); + + // Drain stdout on a thread: the pipe must never fill, or the reader + // blocks on write and the whole fixture deadlocks. + let (tx, lines) = mpsc::channel(); + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines().map_while(Result::ok) { + if tx.send(line).is_err() { + break; + } + } + }); + + Self { + child, + stdin: Some(stdin), + lines, + seq: 0, + } + } + + /// Send `sql` to the reader and block until it has actually executed. + /// + /// A unique sentinel is appended and awaited on stdout. `sqlite3` executes + /// a script strictly in order on one connection, so the sentinel cannot be + /// printed until every preceding statement has run — the acknowledgement is + /// a consequence of ordering, not of timing. + /// + /// Panics with the offending SQL on timeout rather than hanging. + pub fn run(&mut self, sql: &str) { + self.seq += 1; + let token = format!("__ack_{}__", self.seq); + let stdin = self.stdin.as_mut().expect("reader stdin already released"); + writeln!(stdin, "{sql}\nSELECT '{token}';").unwrap(); + stdin.flush().unwrap(); + + loop { + match self.lines.recv_timeout(ACK_TIMEOUT) { + Ok(line) if line.trim() == token => return, + // Statement output (row counts, pragma results) precedes the + // sentinel; skip it. + Ok(_) => {} + Err(RecvTimeoutError::Timeout) => { + panic!("sqlite3 did not acknowledge within {ACK_TIMEOUT:?}: {sql}") + } + Err(RecvTimeoutError::Disconnected) => { + panic!("sqlite3 reader exited before acknowledging: {sql}") + } + } + } + } + + /// Close the held read transaction and reap the process. + pub fn finish(mut self) { + if let Some(mut stdin) = self.stdin.take() { + let _ = writeln!(stdin, "COMMIT;\n.quit"); + } + let _ = self.child.wait(); + } +} + +/// Run one short-lived writer connection against `db`. +pub fn writer_sql(bin: &str, db: &Path, sql: &str) { + let out = Command::new(bin).arg(db).arg(sql).output().unwrap(); + assert!( + out.status.success(), + "sqlite3 writer failed: {}", + String::from_utf8_lossy(&out.stderr) + ); +} diff --git a/core/tests/row_history_oracle.rs b/core/tests/row_history_oracle.rs index 60dd8bf..812f737 100644 --- a/core/tests/row_history_oracle.rs +++ b/core/tests/row_history_oracle.rs @@ -12,9 +12,8 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] -use std::io::Write; use std::path::{Path, PathBuf}; -use std::process::{Command, Stdio}; +use std::process::Command; use sqlite_core::row_history::ViewState; use sqlite_core::{Database, Value}; @@ -42,15 +41,8 @@ fn scratch(tag: &str) -> PathBuf { p } -/// Run one short-lived writer connection (the held reader blocks checkpoint). -fn writer_sql(bin: &str, db: &Path, sql: &str) { - let out = Command::new(bin).arg(db).arg(sql).output().unwrap(); - assert!( - out.status.success(), - "sqlite3 writer failed: {}", - String::from_utf8_lossy(&out.stderr) - ); -} +mod common; +use common::{writer_sql, HeldReader}; /// Build the WAL fixture with the KNOWN mutation sequence; return the held reader /// (kept alive until dropped) and the `(db, wal)` paths. @@ -59,30 +51,19 @@ fn writer_sql(bin: &str, db: &Path, sql: &str) { /// C1: insert (1,'a'),(2,'b'),(4,'x') /// C2: update 1->'A'; delete 2; delete 4 /// C3: insert 3->'c'; insert 4->'y' (rowid 4 REUSED after its C2 delete) -fn build_fixture(bin: &str, dir: &Path) -> (std::process::Child, PathBuf, PathBuf) { +fn build_fixture(bin: &str, dir: &Path) -> (HeldReader, PathBuf, PathBuf) { let db = dir.join("ev.db"); let wal = dir.join("ev.db-wal"); - let mut reader = Command::new(bin) - .arg(&db) - .stdin(Stdio::piped()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .unwrap(); - let mut rin = reader.stdin.take().unwrap(); - writeln!( - rin, + let mut reader = HeldReader::spawn(bin, &db); + // Each `run` returns only once sqlite3 has executed the statement, so the + // writers below cannot race ahead of the CREATE TABLE. + reader.run( "PRAGMA journal_mode=WAL;\nPRAGMA wal_autocheckpoint=0;\nPRAGMA secure_delete=OFF;\n\ - CREATE TABLE t(id INTEGER PRIMARY KEY, name TEXT);" - ) - .unwrap(); - rin.flush().unwrap(); - std::thread::sleep(std::time::Duration::from_millis(400)); - writeln!(rin, "BEGIN;\nSELECT count(*) FROM t;").unwrap(); - rin.flush().unwrap(); - std::thread::sleep(std::time::Duration::from_millis(400)); - reader.stdin = Some(rin); + CREATE TABLE t(id INTEGER PRIMARY KEY, name TEXT);", + ); + // Open the read transaction that retains the -wal across writer commits. + reader.run("BEGIN;\nSELECT count(*) FROM t;"); writer_sql( bin, @@ -107,11 +88,8 @@ fn build_fixture(bin: &str, dir: &Path) -> (std::process::Child, PathBuf, PathBu } /// Release the held reader and remove the scratch directory. -fn teardown(mut reader: std::process::Child, dir: &Path) { - if let Some(mut rin) = reader.stdin.take() { - let _ = writeln!(rin, "COMMIT;\n.quit"); - } - let _ = reader.wait(); +fn teardown(reader: HeldReader, dir: &Path) { + reader.finish(); let _ = std::fs::remove_dir_all(dir); } diff --git a/core/tests/wal_snapshot_oracle.rs b/core/tests/wal_snapshot_oracle.rs index 115e0c4..c481952 100644 --- a/core/tests/wal_snapshot_oracle.rs +++ b/core/tests/wal_snapshot_oracle.rs @@ -17,9 +17,8 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] -use std::io::Write; use std::path::{Path, PathBuf}; -use std::process::{Command, Stdio}; +use std::process::Command; use sqlite_core::{Database, Value}; @@ -49,14 +48,8 @@ fn scratch(tag: &str) -> PathBuf { /// Run `sqlite3 ""` on a SHORT-LIVED writer connection. With a held /// reader already open (see [`build_incremental_fixture`]) the checkpoint-on-close /// is blocked, so the `-wal` survives. -fn writer_sql(bin: &str, db: &Path, sql: &str) { - let out = Command::new(bin).arg(db).arg(sql).output().unwrap(); - assert!( - out.status.success(), - "sqlite3 writer failed: {}", - String::from_utf8_lossy(&out.stderr) - ); -} +mod common; +use common::{writer_sql, HeldReader}; /// Query a snapshot db with `sqlite3` and return the oracle rows /// `(id, name, quote(big))` in id order. @@ -117,33 +110,20 @@ pub struct Snap { /// The held reader is a `sqlite3` process reading commands from a pipe; it opens /// a read transaction (`BEGIN; SELECT ...`) that blocks checkpoint so the `-wal` /// is retained across the short-lived writer connections. -fn build_incremental_fixture(bin: &str, dir: &Path) -> (std::process::Child, Vec) { +fn build_incremental_fixture(bin: &str, dir: &Path) -> (HeldReader, Vec) { let db = dir.join("ev.db"); let wal = dir.join("ev.db-wal"); // Held reader: keep a connection open with an active read txn to block the // checkpoint-on-close that would otherwise delete the -wal. - let mut reader = Command::new(bin) - .arg(&db) - .stdin(Stdio::piped()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .unwrap(); - let mut rin = reader.stdin.take().unwrap(); - writeln!( - rin, + let mut reader = HeldReader::spawn(bin, &db); + // Acknowledged, not slept on: the writers below cannot race the CREATE TABLE. + reader.run( "PRAGMA journal_mode=WAL;\nPRAGMA wal_autocheckpoint=0;\nPRAGMA secure_delete=OFF;\n\ - CREATE TABLE t(id INTEGER PRIMARY KEY, name TEXT, big BLOB);" - ) - .unwrap(); - rin.flush().unwrap(); - std::thread::sleep(std::time::Duration::from_millis(400)); - writeln!(rin, "BEGIN;\nSELECT count(*) FROM t;").unwrap(); - rin.flush().unwrap(); - std::thread::sleep(std::time::Duration::from_millis(400)); - // Keep the reader's stdin open by handing it back to the child for cleanup. - reader.stdin = Some(rin); + CREATE TABLE t(id INTEGER PRIMARY KEY, name TEXT, big BLOB);", + ); + // Open the read transaction that retains the -wal across writer commits. + reader.run("BEGIN;\nSELECT count(*) FROM t;"); let mut snaps = Vec::new(); let snap = |dir: &Path, n: usize, db: &Path, wal: &Path| -> Snap { @@ -180,11 +160,8 @@ fn build_incremental_fixture(bin: &str, dir: &Path) -> (std::process::Child, Vec } /// Release the held reader and remove the scratch directory. -fn teardown(mut reader: std::process::Child, dir: &Path) { - if let Some(mut rin) = reader.stdin.take() { - let _ = writeln!(rin, "COMMIT;\n.quit"); - } - let _ = reader.wait(); +fn teardown(reader: HeldReader, dir: &Path) { + reader.finish(); let _ = std::fs::remove_dir_all(dir); }