Keep queries out of an open watch batch

packages/python/e2e-attestations/bot-1616-watch-batch-isolation-1104.json

@@ -0,0 +1,7 @@
+{
+  "command": "uv run python -m pytest tests/e2e/fresh_declared_repros_test.py tests/e2e/query_contract_test.py tests/e2e/on_file_query_test.py tests/e2e/watch_speed_test.py tests/e2e/server_speed_test.py -x -q",
+  "ran_at": 1793828589,
+  "exit_code": 0,
+  "commit": "63315e05f4ba70bccf37e4a551cb4f7e9b7322d3",
+  "branch": "bot/1616-watch-batch-isolation-1104"
+}

packages/rust/changelog.d/2026-11-04-queries-wait-for-watch-batch.md

@@ -0,0 +1,3 @@
+**Fixed**
+
+- **A query never reads a watch batch half applied or rolled back.** A query with fresh reads off (as the TypeScript binding runs them) now waits for an open watch batch to end, so it no longer sees one table updated and another not, or rows a foreign-key failure then rolls back.

packages/rust/migrations.d/2026-11-04-queries-wait-for-watch-batch.md

@@ -0,0 +1,21 @@
+### Queries wait for an open watch batch
+
+#### Summary
+
+A query whose instance has fresh reads off now waits for any watch batch in progress to commit or roll back before it reads, as queries with fresh reads on already did. Nothing breaks.
+
+#### Required changes
+
+_None._
+
+#### Deprecations removed
+
+_None._
+
+#### Behavior changes without code changes
+
+A query issued during a watch batch returns the index as it stood before the batch or after it, never in between; it may wait for that batch's hooks to finish.
+
+#### Verification
+
+With two tables `a` and `b` over `d/*.jsonl` whose `on-file` is `sh -c 'sleep 1; cat "$@"' sh`, run `dirsql server`, write a new `d/2.jsonl`, and query `SELECT (SELECT count(*) FROM a) AS na, (SELECT count(*) FROM b) AS nb` repeatedly: every answer has `na` equal to `nb`.

packages/rust/src/lib.rs

@@ -640,15 +640,17 @@ impl DirSQL {
         self.reconcile(&mut state)
     }
 
-    /// Bring every named table in line with the files on disk now, and hold
-    /// the index still until the returned guard drops.
-    fn refresh(&self) -> Result<Option<MutexGuard<'_, FreshState>>> {
-        if self.inner.groups.is_empty() || !self.inner.fresh_reads.load(Ordering::SeqCst) {
-            return Ok(None);
-        }
+    /// Hold the index still until the returned guard drops, first bringing
+    /// every named table in line with the files on disk when fresh reads are
+    /// on. A watch batch releases the database between statements inside its
+    /// transaction, so a query that took only the database could read it
+    /// half applied.
+    fn refresh(&self) -> Result<MutexGuard<'_, FreshState>> {
         let mut state = self.inner.fresh.lock().map_err(DirSqlError::lock)?;
-        self.reconcile(&mut state)?;
-        Ok(Some(state))
+        if !self.inner.groups.is_empty() && self.inner.fresh_reads.load(Ordering::SeqCst) {
+            self.reconcile(&mut state)?;
+        }
+        Ok(state)
     }
 
     /// Walk each anchor, read every file whose stat says it may differ from
@@ -7751,6 +7753,16 @@ mod internal_tests {
         assert_eq!(row_names(&db), vec!["one", "two"]);
     }
 
+    #[test]
+    fn a_query_with_fresh_reads_off_still_holds_the_index_still() {
+        let (_dir, _root, db, _) = fresh_fixture(&[("a.txt", "one")]);
+        db.set_fresh_reads(false);
+        let held = db.refresh().unwrap();
+        assert!(db.inner.fresh.try_lock().is_err());
+        drop(held);
+        assert!(db.inner.fresh.try_lock().is_ok());
+    }
+
     #[test]
     fn read_changes_reads_the_disk_with_fresh_reads_off() {
         let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]);

packages/rust/tests/watch_batch_isolation.rs

@@ -0,0 +1,180 @@
+//! A query never sees a watch batch half applied, nor one its foreign keys
+//! roll back: it waits for the batch to end. Queries here skip reading the
+//! disk themselves, as the TypeScript binding's do, so only the batch writes.
+
+use std::collections::HashMap;
+use std::fs;
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::mpsc::{Receiver, Sender, channel};
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::Duration;
+
+use dirsql::watcher::FileEvent;
+use dirsql::{DirSQL, Row, Table, Value};
+use tempfile::TempDir;
+
+const HOLD: Duration = Duration::from_secs(1);
+
+/// The hooks of one batch: the second hook to read `target` reports that the
+/// batch is under way, then holds it open for [`HOLD`].
+#[derive(Clone)]
+struct Pacer {
+    target: &'static str,
+    calls: Arc<AtomicUsize>,
+    started: Arc<Mutex<Sender<()>>>,
+}
+
+impl Pacer {
+    fn new(target: &'static str) -> (Self, Receiver<()>) {
+        let (tx, rx) = channel();
+        let pacer = Self {
+            target,
+            calls: Arc::default(),
+            started: Arc::new(Mutex::new(tx)),
+        };
+        (pacer, rx)
+    }
+
+    fn table(&self, name: &str, ddl: &str, glob: &str, row: fn(&str) -> Row) -> Table {
+        let pacer = self.clone();
+        Table::new(name, ddl, glob, move |path: &str| {
+            if Path::new(path).ends_with(pacer.target)
+                && pacer.calls.fetch_add(1, Ordering::SeqCst) == 1
+            {
+                pacer.started.lock().unwrap().send(()).unwrap();
+                thread::sleep(HOLD);
+            }
+            vec![row(path)]
+        })
+    }
+}
+
+fn line_row(path: &str) -> Row {
+    HashMap::from([(
+        "line".into(),
+        Value::Text(fs::read_to_string(path).unwrap().trim().to_string()),
+    )])
+}
+
+fn child_row(path: &str) -> Row {
+    HashMap::from([
+        (
+            "id".into(),
+            Value::Text(
+                Path::new(path)
+                    .file_stem()
+                    .unwrap()
+                    .to_string_lossy()
+                    .into(),
+            ),
+        ),
+        (
+            "parent".into(),
+            Value::Text(fs::read_to_string(path).unwrap().trim().to_string()),
+        ),
+    ])
+}
+
+fn key_row(path: &str) -> Row {
+    HashMap::from([(
+        "k".into(),
+        Value::Text(
+            Path::new(path)
+                .file_stem()
+                .unwrap()
+                .to_string_lossy()
+                .into(),
+        ),
+    )])
+}
+
+fn tree(dirs: &[&str]) -> (TempDir, PathBuf) {
+    let dir = TempDir::new().unwrap();
+    let root = dir.path().canonicalize().unwrap();
+    for d in dirs {
+        fs::create_dir(root.join(d)).unwrap();
+    }
+    (dir, root)
+}
+
+/// Apply `event` on another thread and, once the batch is under way, ask
+/// `sql` from this one.
+fn query_mid_batch(db: &DirSQL, started: &Receiver<()>, event: FileEvent, sql: &str) -> Vec<Row> {
+    let batch = {
+        let db = db.clone();
+        thread::spawn(move || db.apply_file_events(vec![event]))
+    };
+    started.recv_timeout(Duration::from_secs(30)).unwrap();
+    let rows = db.query(sql).unwrap();
+    batch.join().unwrap();
+    rows
+}
+
+#[test]
+fn a_query_during_a_batch_sees_every_table_before_it_or_after_it() {
+    let (_dir, root) = tree(&["d"]);
+    fs::write(root.join("d").join("1.txt"), "one").unwrap();
+    let (pacer, started) = Pacer::new("2.txt");
+    let db = DirSQL::new(
+        &root,
+        vec![
+            pacer.table("a", "CREATE TABLE a (line TEXT)", "d/*.txt", line_row),
+            pacer.table("b", "CREATE TABLE b (line TEXT)", "d/*.txt", line_row),
+        ],
+    )
+    .unwrap();
+    db.set_fresh_reads(false);
+    let added = root.join("d").join("2.txt");
+    fs::write(&added, "two").unwrap();
+
+    let rows = query_mid_batch(
+        &db,
+        &started,
+        FileEvent::Created(added),
+        "SELECT (SELECT count(*) FROM a) AS na, (SELECT count(*) FROM b) AS nb",
+    );
+
+    assert_eq!(rows[0]["na"], rows[0]["nb"], "torn read: {rows:?}");
+}
+
+#[test]
+fn a_query_during_a_batch_never_sees_rows_the_batch_rolls_back() {
+    let (_dir, root) = tree(&["c", "p"]);
+    fs::write(root.join("p").join("p1.txt"), "").unwrap();
+    fs::write(root.join("c").join("c1.txt"), "p1").unwrap();
+    let (pacer, started) = Pacer::new("orphan.txt");
+    let db = DirSQL::new(
+        &root,
+        vec![
+            pacer.table(
+                "child",
+                "CREATE TABLE child (id TEXT PRIMARY KEY, parent TEXT REFERENCES parent(k))",
+                "c/*.txt",
+                child_row,
+            ),
+            pacer.table("pad", "CREATE TABLE pad (line TEXT)", "c/*.txt", line_row),
+            pacer.table(
+                "parent",
+                "CREATE TABLE parent (k TEXT PRIMARY KEY)",
+                "p/*.txt",
+                key_row,
+            ),
+        ],
+    )
+    .unwrap();
+    db.set_fresh_reads(false);
+    let orphan = root.join("c").join("orphan.txt");
+    fs::write(&orphan, "MISSING").unwrap();
+
+    let during = query_mid_batch(
+        &db,
+        &started,
+        FileEvent::Created(orphan),
+        "SELECT id FROM child ORDER BY id",
+    );
+
+    let after = db.query("SELECT id FROM child ORDER BY id").unwrap();
+    assert_eq!(after, during, "read rows the batch rolled back");
+}

packages/rust/tests/watch_batch_isolation_e2e.rs

@@ -0,0 +1,131 @@
+//! A server query issued while a watch batch runs never sees it half applied,
+//! nor a row the batch's foreign keys roll back. Real binary, real
+//! filesystem, nothing mocked.
+
+#![cfg(feature = "cli")]
+
+#[path = "common/server.rs"]
+mod server;
+
+use reqwest::blocking::Client;
+use serde_json::{Value, json};
+use std::fs;
+use std::io::{BufRead, BufReader};
+use std::net::TcpListener;
+use std::path::Path;
+use std::thread;
+use std::time::Duration;
+use tempfile::TempDir;
+
+const SLOW_CAT: &str = r#"on-file = """sh -c 'sleep 1; cat "$@"' sh""""#;
+
+fn serve(root: &Path) -> (server::ServerGuard, String) {
+    let port = TcpListener::bind("localhost:0")
+        .unwrap()
+        .local_addr()
+        .unwrap()
+        .port();
+    let mut server = server::spawn_server(root, port, &["-c", ".dirsql.toml"]);
+    let mut banner = String::new();
+    BufReader::new(server.stdout.take().unwrap())
+        .read_line(&mut banner)
+        .unwrap();
+    assert!(banner.starts_with("Running at"), "{banner:?}");
+    (server, format!("http://localhost:{port}/query"))
+}
+
+fn ask(client: &Client, url: &str, sql: &str) -> Value {
+    client
+        .post(url)
+        .json(&json!({ "sql": sql }))
+        .send()
+        .unwrap()
+        .json()
+        .unwrap()
+}
+
+#[test]
+fn every_query_sees_both_tables_fed_by_one_batch_at_the_same_point() {
+    let dir = TempDir::new().unwrap();
+    let root = dir.path();
+    fs::create_dir(root.join("d")).unwrap();
+    let table = |name: &str| {
+        format!(
+            "[[table]]\nname = \"{name}\"\nddl = \"CREATE TABLE {name} (line TEXT)\"\nglob = \"d/*.jsonl\"\n{SLOW_CAT}\n"
+        )
+    };
+    fs::write(
+        root.join(".dirsql.toml"),
+        format!("{}\n{}", table("a"), table("b")),
+    )
+    .unwrap();
+    fs::write(root.join("d").join("1.jsonl"), "{\"line\":\"one\"}\n").unwrap();
+    let (_server, url) = serve(root);
+    let client = Client::new();
+
+    fs::write(root.join("d").join("2.jsonl"), "{\"line\":\"two\"}\n").unwrap();
+    let mut seen = Vec::new();
+    for _ in 0..25 {
+        seen.push(ask(
+            &client,
+            &url,
+            "SELECT (SELECT count(*) FROM a) AS na, (SELECT count(*) FROM b) AS nb",
+        ));
+        thread::sleep(Duration::from_millis(100));
+    }
+
+    for rows in &seen {
+        assert_eq!(rows[0]["na"], rows[0]["nb"], "torn read in {seen:?}");
+    }
+    assert_eq!(seen.last().unwrap()[0]["na"], 2, "{seen:?}");
+}
+
+#[test]
+fn no_query_sees_a_row_from_a_batch_its_foreign_keys_reject() {
+    let dir = TempDir::new().unwrap();
+    let root = dir.path();
+    fs::create_dir(root.join("c")).unwrap();
+    fs::write(
+        root.join(".dirsql.toml"),
+        r#"[[table]]
+name = "child"
+ddl = "CREATE TABLE child (id TEXT PRIMARY KEY, parent TEXT REFERENCES parent(k))"
+glob = "c/*.jsonl"
+on-file = "cat"
+
+[[table]]
+name = "parent"
+ddl = "CREATE TABLE parent (k TEXT PRIMARY KEY)"
+glob = "c/*.jsonl"
+on-file = """sh -c 'sleep 2; cat "$0"' p.jsonl"""
+"#,
+    )
+    .unwrap();
+    fs::write(root.join("p.jsonl"), "{\"k\":\"p1\"}\n").unwrap();
+    fs::write(
+        root.join("c").join("a.jsonl"),
+        "{\"id\":\"c1\",\"parent\":\"p1\"}\n",
+    )
+    .unwrap();
+    let (_server, url) = serve(root);
+    let client = Client::new();
+
+    fs::write(
+        root.join("c").join("a.jsonl"),
+        "{\"id\":\"c1\",\"parent\":\"p1\"}\n{\"id\":\"c2\",\"parent\":\"MISSING\"}\n",
+    )
+    .unwrap();
+    let mut seen = Vec::new();
+    for _ in 0..10 {
+        thread::sleep(Duration::from_millis(400));
+        seen.push(ask(&client, &url, "SELECT id FROM child ORDER BY id"));
+    }
+
+    for answer in &seen {
+        let ids: Vec<&Value> = answer
+            .as_array()
+            .map(|rows| rows.iter().map(|row| &row["id"]).collect())
+            .unwrap_or_default();
+        assert!(!ids.contains(&&json!("c2")), "dirty read in {seen:?}");
+    }
+}

packages/ts/e2e-attestations/bot-1616-watch-batch-isolation-1104.json

@@ -0,0 +1,7 @@
+{
+  "command": "pnpm exec vitest run --dir tests/e2e fresh-declared-repros query-contract fanout",
+  "ran_at": 1793828673,
+  "exit_code": 0,
+  "commit": "f701c789b97e1a9a2aa8fe0778c21ebf23cb0661",
+  "branch": "bot/1616-watch-batch-isolation-1104"
+}