fix(test): acknowledge sqlite3 statements instead of sleeping on them

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<ChildStdin>,
+    lines: Receiver<String>,
+    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)
+    );
+}

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

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 <db> "<sql>"` 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<Snap>) {
+fn build_incremental_fixture(bin: &str, dir: &Path) -> (HeldReader, Vec<Snap>) {
     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);
 }