From ce81ad3cb93b0b56a23a702ead1946b39ce79383 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Tue, 3 Nov 2026 22:29:07 -0500 Subject: [PATCH 01/12] test: declared tables read exactly what is on disk at t+0 Hammers, a fresh-instance oracle and each named edge case, in Rust, Python and TypeScript; adds a macOS Rust test job. --- .github/workflows/dirsql-rust-ci.yml | 24 + .../binding/fresh_declared_test.py | 353 ++++++++++++ packages/rust/tests/common/fresh.rs | 238 ++++++++ packages/rust/tests/fresh_declared_edges.rs | 213 ++++++++ packages/rust/tests/fresh_declared_events.rs | 77 +++ packages/rust/tests/fresh_declared_hammer.rs | 104 ++++ packages/rust/tests/fresh_declared_oracle.rs | 164 ++++++ .../binding/fresh-declared.test.ts | 512 ++++++++++++++++++ 8 files changed, 1685 insertions(+) create mode 100644 packages/python/tests/integration/binding/fresh_declared_test.py create mode 100644 packages/rust/tests/common/fresh.rs create mode 100644 packages/rust/tests/fresh_declared_edges.rs create mode 100644 packages/rust/tests/fresh_declared_events.rs create mode 100644 packages/rust/tests/fresh_declared_hammer.rs create mode 100644 packages/rust/tests/fresh_declared_oracle.rs create mode 100644 packages/ts/tests/integration/binding/fresh-declared.test.ts diff --git a/.github/workflows/dirsql-rust-ci.yml b/.github/workflows/dirsql-rust-ci.yml index 48fa0c246..3f56c1ec8 100644 --- a/.github/workflows/dirsql-rust-ci.yml +++ b/.github/workflows/dirsql-rust-ci.yml @@ -124,6 +124,30 @@ jobs: shell: bash run: 'test ! -e "$LOCALAPPDATA/dirsql" || { echo "::error::a test wrote to the real user cache dir (%LOCALAPPDATA%/dirsql); pass an explicit persist path"; exit 1; }' + macos: + name: macOS tests + runs-on: macos-latest + steps: + - uses: actions/checkout@v6 + + - name: Install Rust + uses: dtolnay/rust-toolchain@stable + + - name: Cache cargo + uses: actions/cache@v4 + with: + path: | + ~/.cargo/registry + ~/.cargo/git + target + key: ${{ runner.os }}-cargo-stable-${{ hashFiles('Cargo.lock') }} + + - name: Run unit and integration tests + run: cargo test --manifest-path packages/rust/Cargo.toml --features cli --tests --no-fail-fast + + - name: Tests left the user cache dir untouched + run: 'test ! -e "$HOME/Library/Caches/dirsql" || { echo "::error::a test wrote to the real user cache dir (~/Library/Caches/dirsql); pass an explicit persist path"; exit 1; }' + coverage: name: Rust unit coverage floor runs-on: ubuntu-latest diff --git a/packages/python/tests/integration/binding/fresh_declared_test.py b/packages/python/tests/integration/binding/fresh_declared_test.py new file mode 100644 index 000000000..d65547e5a --- /dev/null +++ b/packages/python/tests/integration/binding/fresh_declared_test.py @@ -0,0 +1,353 @@ +"""Binding-tier tests: a declared table reads exactly what is on disk. + +Every query runs the moment the write before it returns, and must read what +a ``DirSQL`` built fresh over the same tree reads. +""" + +import asyncio +import os +import random +import shutil +import threading + +import pytest + +from dirsql import DirSQL, Table + +ROUNDS = 2000 + + +def _notes(root): + roots = [os.path.realpath(root), root] + + def on_file(path): + rel = path + for base in roots: + if path.startswith(base + os.sep): + rel = path[len(base) + 1 :] + break + rel = rel.replace(os.sep, "/") + with open(path, encoding="utf-8") as f: + text = f.read() + return [{"file": rel, "line": line} for line in text.splitlines() if line] + + return Table( + name="notes", + ddl="CREATE TABLE notes (file TEXT, line TEXT)", + glob="**/*.txt", + on_file=on_file, + ) + + +async def _open(root): + db = DirSQL(root, tables=[_notes(root)]) + await db.ready() + return db + + +def _write(root, rel, content): + path = os.path.join(root, rel) + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8", newline="") as f: + f.write(content) + + +def _save_atomically(root, rel, content): + _write(root, rel + ".stage", content) + os.replace(os.path.join(root, rel + ".stage"), os.path.join(root, rel)) + + +def _times(root, rel): + st = os.stat(os.path.join(root, rel)) + return (st.st_atime_ns, st.st_mtime_ns) + + +def _restore_times(root, rel, times): + os.utime(os.path.join(root, rel), ns=times) + + +async def _lines(db, rel): + rows = await db.query(f"SELECT line FROM notes WHERE file = '{rel}' ORDER BY line") + return [row["line"] for row in rows] + + +async def _snapshot(db): + rows = await db.query("SELECT file, line FROM notes ORDER BY file, line") + return [(row["file"], row["line"]) for row in rows] + + +async def _assert_matches_fresh_scan(db, root, context): + fresh = await _open(root) + assert await _snapshot(db) == await _snapshot(fresh), ( + f"the long-lived instance disagrees with a fresh scan after {context}" + ) + + +DIRS = ["", "a/", "a/b/", "c/"] +NAMES = ["n0", "n1", "n2", "n3"] +EXTS = ["txt", "json", "stage"] +IGNORES = ["n0.txt\n", "a/\n", "*.txt\n", "c/n1.*\n"] + + +def _content(rng): + return "".join( + f"line-{rng.randrange(10_000):04}\n" for _ in range(rng.randrange(1, 4)) + ) + + +def _random_file(rng): + return f"{rng.choice(DIRS)}{rng.choice(NAMES)}.{rng.choice(EXTS)}" + + +def _existing(root): + return [ + f"{d}{n}.{e}" + for d in DIRS + for n in NAMES + for e in EXTS + if os.path.isfile(os.path.join(root, f"{d}{n}.{e}")) + ] + + +def _step(root, rng): + existing = _existing(root) + op = rng.randrange(9) + if op == 1 and existing: + rel = rng.choice(existing) + times = _times(root, rel) + with open(os.path.join(root, rel), encoding="utf-8") as f: + old = f.read() + flipped = "".join(str((int(c) + 1) % 10) if c.isdigit() else c for c in old) + _write(root, rel, flipped) + _restore_times(root, rel, times) + return f"same-size edit of {rel} with its times restored" + if op == 2: + rel = _random_file(rng) + _save_atomically(root, rel, _content(rng)) + return f"atomic save of {rel}" + if op == 3 and existing: + rel = rng.choice(existing) + os.remove(os.path.join(root, rel)) + return f"delete {rel}" + if op == 4 and existing: + src = rng.choice(existing) + dst = _random_file(rng) + if src == dst: + return "nothing" + os.makedirs(os.path.dirname(os.path.join(root, dst)), exist_ok=True) + os.replace(os.path.join(root, src), os.path.join(root, dst)) + return f"rename {src} to {dst}" + if op == 5: + new = f"{rng.choice(DIRS)}new{rng.randrange(3)}" + for name in NAMES[:2]: + _write(root, f"{new}/{name}.{rng.choice(EXTS)}", _content(rng)) + return f"populate new directory {new}" + if op == 6: + gone = rng.choice(DIRS[1:]).rstrip("/") + shutil.rmtree(os.path.join(root, gone), ignore_errors=True) + return f"remove directory {gone}" + if op == 7: + if rng.randrange(3) == 0: + if os.path.exists(os.path.join(root, ".gitignore")): + os.remove(os.path.join(root, ".gitignore")) + return "remove .gitignore" + rule = rng.choice(IGNORES) + _write(root, ".gitignore", rule) + return f"set .gitignore to {rule!r}" + rel = _random_file(rng) + _write(root, rel, _content(rng)) + return f"write {rel}" + + +def _seed_tree(root, rng): + os.makedirs(os.path.join(root, ".git")) + for _ in range(8): + _write(root, _random_file(rng), _content(rng)) + + +def describe_a_query_right_after_a_write(): + @pytest.mark.asyncio + async def it_reads_every_create(tmp_dir): + db = await _open(tmp_dir) + for i in range(ROUNDS): + _write(tmp_dir, f"h/{i}.txt", f"c{i}") + assert await _lines(db, f"h/{i}.txt") == [f"c{i}"], f"round {i}" + os.remove(os.path.join(tmp_dir, f"h/{i}.txt")) + + @pytest.mark.asyncio + async def it_reads_every_edit(tmp_dir): + _write(tmp_dir, "e.txt", "start") + db = await _open(tmp_dir) + for i in range(ROUNDS): + _write(tmp_dir, "e.txt", f"e{i}") + assert await _lines(db, "e.txt") == [f"e{i}"], f"round {i}" + + @pytest.mark.asyncio + async def it_reads_every_delete(tmp_dir): + db = await _open(tmp_dir) + for i in range(ROUNDS): + _write(tmp_dir, "d.txt", f"d{i}") + assert await _lines(db, "d.txt") == [f"d{i}"], f"round {i}" + os.remove(os.path.join(tmp_dir, "d.txt")) + assert await _lines(db, "d.txt") == [], f"round {i}" + + @pytest.mark.asyncio + async def it_reads_every_rename_into_and_out_of_the_glob(tmp_dir): + db = await _open(tmp_dir) + staged = os.path.join(tmp_dir, "r.stage") + final = os.path.join(tmp_dir, "r.txt") + for i in range(ROUNDS): + _write(tmp_dir, "r.stage", f"r{i}") + os.replace(staged, final) + assert await _lines(db, "r.txt") == [f"r{i}"], f"round {i}" + os.replace(final, staged) + assert await _lines(db, "r.txt") == [], f"round {i}" + + @pytest.mark.asyncio + async def it_reads_every_atomic_save(tmp_dir): + _write(tmp_dir, "a.txt", "start") + db = await _open(tmp_dir) + for i in range(ROUNDS): + _save_atomically(tmp_dir, "a.txt", f"a{i}") + assert await _lines(db, "a.txt") == [f"a{i}"], f"round {i}" + + @pytest.mark.asyncio + async def it_reads_every_same_size_edit(tmp_dir): + _write(tmp_dir, "s.txt", "s------") + db = await _open(tmp_dir) + for i in range(ROUNDS): + _write(tmp_dir, "s.txt", f"s{i:06}") + assert await _lines(db, "s.txt") == [f"s{i:06}"], f"round {i}" + + @pytest.mark.asyncio + async def it_reads_every_same_size_edit_with_its_times_restored(tmp_dir): + _write(tmp_dir, "t.txt", "t------") + times = _times(tmp_dir, "t.txt") + db = await _open(tmp_dir) + for i in range(ROUNDS): + _write(tmp_dir, "t.txt", f"t{i:06}") + _restore_times(tmp_dir, "t.txt", times) + assert await _lines(db, "t.txt") == [f"t{i:06}"], f"round {i}" + + +def describe_a_long_lived_instance(): + @pytest.mark.asyncio + async def it_reads_what_a_fresh_scan_reads_after_random_operations(tmp_dir): + for seed in range(1, 4): + root = os.path.join(tmp_dir, f"seed{seed}") + os.makedirs(root) + rng = random.Random(seed) + _seed_tree(root, rng) + db = await _open(root) + for op in range(100): + done = _step(root, rng) + await _assert_matches_fresh_scan( + db, root, f"seed {seed} op {op}: {done}" + ) + + @pytest.mark.asyncio + async def it_reads_what_a_fresh_scan_reads_across_idle_gaps_and_bursts(tmp_dir): + rng = random.Random(99) + _seed_tree(tmp_dir, rng) + db = await _open(tmp_dir) + for cycle in range(10): + await asyncio.sleep(0.25 * (cycle % 4)) + for op in range(20): + done = _step(tmp_dir, rng) + await _assert_matches_fresh_scan( + db, tmp_dir, f"cycle {cycle} op {op}: {done}" + ) + + +def describe_each_way_a_change_could_be_missed(): + @pytest.mark.asyncio + async def it_reads_an_edit_whose_times_are_put_back(tmp_dir): + _write(tmp_dir, "touched.txt", "before") + times = _times(tmp_dir, "touched.txt") + db = await _open(tmp_dir) + assert await _lines(db, "touched.txt") == ["before"] + _write(tmp_dir, "touched.txt", "after!") + _restore_times(tmp_dir, "touched.txt", times) + assert await _lines(db, "touched.txt") == ["after!"] + + @pytest.mark.asyncio + async def it_reads_a_file_deleted_and_recreated_under_the_same_name(tmp_dir): + _write(tmp_dir, "again.txt", "first") + db = await _open(tmp_dir) + assert await _lines(db, "again.txt") == ["first"] + os.remove(os.path.join(tmp_dir, "again.txt")) + _write(tmp_dir, "again.txt", "secnd") + assert await _lines(db, "again.txt") == ["secnd"] + + @pytest.mark.asyncio + async def it_reads_a_new_and_a_deleted_subdirectory(tmp_dir): + db = await _open(tmp_dir) + _write(tmp_dir, "sub/deep/one.txt", "one") + assert await _lines(db, "sub/deep/one.txt") == ["one"] + shutil.rmtree(os.path.join(tmp_dir, "sub")) + assert await _lines(db, "sub/deep/one.txt") == [] + + @pytest.mark.asyncio + async def it_reads_a_gitignore_change(tmp_dir): + os.makedirs(os.path.join(tmp_dir, ".git")) + _write(tmp_dir, "hidden.txt", "hidden") + db = await _open(tmp_dir) + assert await _lines(db, "hidden.txt") == ["hidden"] + _write(tmp_dir, ".gitignore", "hidden.txt\n") + assert await _lines(db, "hidden.txt") == [] + os.remove(os.path.join(tmp_dir, ".gitignore")) + assert await _lines(db, "hidden.txt") == ["hidden"] + + @pytest.mark.asyncio + async def it_reads_a_retargeted_symlink(tmp_dir): + _write(tmp_dir, "targets/one.dat", "one") + _write(tmp_dir, "targets/two.dat", "two") + link = os.path.join(tmp_dir, "link.txt") + os.symlink(os.path.join(tmp_dir, "targets/one.dat"), link) + db = await _open(tmp_dir) + assert await _lines(db, "link.txt") == ["one"] + os.remove(link) + os.symlink(os.path.join(tmp_dir, "targets/two.dat"), link) + assert await _lines(db, "link.txt") == ["two"] + + @pytest.mark.asyncio + async def it_never_reads_a_file_torn_while_it_is_rewritten_and_ends_exact(tmp_dir): + def version(k): + return f"v{k:06}\n" * 50 + + _write(tmp_dir, "busy.txt", version(0)) + db = await _open(tmp_dir) + done = threading.Event() + + def writer(): + for k in range(1, 301): + _save_atomically(tmp_dir, "busy.txt", version(k)) + done.set() + + thread = threading.Thread(target=writer) + thread.start() + while not done.is_set(): + lines = await _lines(db, "busy.txt") + assert len(lines) == 50 and len(set(lines)) == 1, lines + thread.join() + assert await _lines(db, "busy.txt") == ["v000300"] * 50 + + +def describe_watch_consumers(): + @pytest.mark.asyncio + async def it_streams_each_change_a_query_read_first_once(tmp_dir): + db = await _open(tmp_dir) + stream = db.watch() + files = 100 + for i in range(files): + _write(tmp_dir, f"w/{i}.txt", f"w{i}") + assert await _lines(db, f"w/{i}.txt") == [f"w{i}"], f"file {i}" + counts = {} + while True: + try: + event = await asyncio.wait_for(anext(stream), timeout=2.0) + except TimeoutError: + break + assert event.action == "insert", event + counts[event.row["line"]] = counts.get(event.row["line"], 0) + 1 + assert counts == {f"w{i}": 1 for i in range(files)} diff --git a/packages/rust/tests/common/fresh.rs b/packages/rust/tests/common/fresh.rs new file mode 100644 index 000000000..0efabac92 --- /dev/null +++ b/packages/rust/tests/common/fresh.rs @@ -0,0 +1,238 @@ +//! A tree, the instance over it, and the readings the freshness tests assert: +//! a declared table must read exactly what a fresh scan of the disk reads. + +#![allow(dead_code)] + +use dirsql::{DirSQL, Row, Table, Value}; +use std::collections::HashMap; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::SystemTime; +use tempfile::TempDir; + +/// `recs` is a config table, so its hook runs once over every matched file. +const CONFIG: &str = r#" +[[table]] +name = "recs" +glob = "**/*.json" +on-file = "cat" +ddl = "CREATE TABLE recs (k TEXT, v TEXT)" +"#; + +pub struct Tree { + dir: TempDir, +} + +impl Tree { + pub fn new() -> Self { + let dir = TempDir::new().unwrap(); + fs::write(dir.path().join(".dirsql.toml"), CONFIG).unwrap(); + Self { dir } + } + + pub fn root(&self) -> &Path { + self.dir.path() + } + + pub fn path(&self, rel: &str) -> PathBuf { + self.root().join(rel) + } + + /// The instance under test: `notes`, a per-file table over `**/*.txt`, + /// plus the config's `recs`. + pub fn open(&self) -> DirSQL { + DirSQL::builder() + .root(self.root()) + .config(self.path(".dirsql.toml")) + .table(notes_table(self.root())) + .build() + .unwrap() + } + + pub fn write(&self, rel: &str, content: &str) { + let path = self.path(rel); + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).unwrap(); + } + fs::write(path, content).unwrap(); + } + + /// Write a sibling outside every glob, then rename it over `rel`. + pub fn save_atomically(&self, rel: &str, content: &str) { + let staged = self.path(&format!("{rel}.stage")); + if let Some(parent) = staged.parent() { + fs::create_dir_all(parent).unwrap(); + } + fs::write(&staged, content).unwrap(); + fs::rename(staged, self.path(rel)).unwrap(); + } + + pub fn remove(&self, rel: &str) { + fs::remove_file(self.path(rel)).unwrap(); + } + + pub fn rename(&self, from: &str, to: &str) { + let to = self.path(to); + if let Some(parent) = to.parent() { + fs::create_dir_all(parent).unwrap(); + } + fs::rename(self.path(from), to).unwrap(); + } + + pub fn times(&self, rel: &str) -> (SystemTime, SystemTime) { + let meta = fs::metadata(self.path(rel)).unwrap(); + (meta.accessed().unwrap(), meta.modified().unwrap()) + } + + /// `touch -r`: put back the access and modification times `times` holds. + pub fn restore_times(&self, rel: &str, times: (SystemTime, SystemTime)) { + let file = fs::File::options() + .write(true) + .open(self.path(rel)) + .unwrap(); + file.set_times( + fs::FileTimes::new() + .set_accessed(times.0) + .set_modified(times.1), + ) + .unwrap(); + } + + /// Point the symlink `rel` at `target`, replacing any link already there. + pub fn link(&self, rel: &str, target: &str) { + let link = self.path(rel); + if fs::symlink_metadata(&link).is_ok() { + fs::remove_file(&link).unwrap(); + } + symlink_file(&self.path(target), &link); + } +} + +#[cfg(unix)] +fn symlink_file(target: &Path, link: &Path) { + std::os::unix::fs::symlink(target, link).unwrap(); +} + +#[cfg(windows)] +fn symlink_file(target: &Path, link: &Path) { + std::os::windows::fs::symlink_file(target, link).unwrap(); +} + +/// One row per non-empty line, keyed by the file's root-relative path. +pub fn notes_table(root: &Path) -> Table { + let roots = [ + root.to_path_buf(), + root.canonicalize().unwrap_or_else(|_| root.to_path_buf()), + ]; + Table::new( + "notes", + "CREATE TABLE notes (file TEXT, line TEXT)", + "**/*.txt", + move |path| { + let path = Path::new(path); + let rel = roots + .iter() + .find_map(|root| path.strip_prefix(root).ok()) + .unwrap_or(path) + .to_string_lossy() + .replace('\\', "/"); + let content = fs::read_to_string(path).unwrap_or_default(); + content + .lines() + .filter(|line| !line.is_empty()) + .map(|line| { + HashMap::from([ + ("file".to_string(), Value::Text(rel.clone())), + ("line".to_string(), Value::Text(line.to_string())), + ]) + }) + .collect::>() + }, + ) +} + +/// Every declared table's rows, sorted. +#[derive(Debug, PartialEq, Eq)] +pub struct Snapshot { + pub notes: Vec<(String, String)>, + pub recs: Vec<(String, String)>, +} + +pub fn snapshot(db: &DirSQL) -> Snapshot { + Snapshot { + notes: pairs(db, "SELECT file AS a, line AS b FROM notes ORDER BY a, b"), + recs: pairs(db, "SELECT k AS a, v AS b FROM recs ORDER BY a, b"), + } +} + +fn pairs(db: &DirSQL, sql: &str) -> Vec<(String, String)> { + db.query(sql) + .unwrap() + .iter() + .map(|row| (text(&row["a"]), text(&row["b"]))) + .collect() +} + +pub fn text(value: &Value) -> String { + match value { + Value::Text(s) => s.clone(), + Value::Null => "".to_string(), + other => format!("{other:?}"), + } +} + +/// The `notes` lines read for `file`, in line order. +pub fn lines_of(db: &DirSQL, file: &str) -> Vec { + db.query(&format!( + "SELECT line FROM notes WHERE file = '{file}' ORDER BY line" + )) + .unwrap() + .iter() + .map(|row| text(&row["line"])) + .collect() +} + +/// The `recs` values read for key `k`. +pub fn recs_of(db: &DirSQL, k: &str) -> Vec { + db.query(&format!("SELECT v FROM recs WHERE k = '{k}' ORDER BY v")) + .unwrap() + .iter() + .map(|row| text(&row["v"])) + .collect() +} + +/// The oracle: what a fresh scan of the disk reads right now. +pub fn assert_matches_fresh_scan(db: &DirSQL, tree: &Tree, context: &str) { + let fresh = tree.open(); + assert_eq!( + snapshot(db), + snapshot(&fresh), + "the long-lived instance disagrees with a fresh scan after {context}" + ); +} + +/// A small deterministic generator, so a failing sequence replays. +pub struct Rng(u64); + +impl Rng { + pub fn new(seed: u64) -> Self { + Self(seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1) + } + + pub fn next(&mut self) -> u64 { + let mut x = self.0; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + self.0 = x; + x + } + + pub fn below(&mut self, n: usize) -> usize { + usize::try_from(self.next() % n as u64).unwrap() + } + + pub fn pick<'a, T>(&mut self, items: &'a [T]) -> &'a T { + &items[self.below(items.len())] + } +} diff --git a/packages/rust/tests/fresh_declared_edges.rs b/packages/rust/tests/fresh_declared_edges.rs new file mode 100644 index 000000000..d6dc614f3 --- /dev/null +++ b/packages/rust/tests/fresh_declared_edges.rs @@ -0,0 +1,213 @@ +//! Each way a change could slip past a freshness check, read straight after +//! the change returns. + +#[path = "common/fresh.rs"] +mod fresh; + +use fresh::{Tree, assert_matches_fresh_scan, lines_of, snapshot}; +use std::fs; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +#[test] +fn a_same_size_edit_within_one_mtime_tick_is_read() { + let tree = Tree::new(); + tree.write("tick.txt", "before"); + let db = tree.open(); + assert_eq!(lines_of(&db, "tick.txt"), ["before"]); + let mut same_tick = 0; + for i in 0..2000 { + let (_, mtime) = tree.times("tick.txt"); + let content = format!("x{i:05}"); + tree.write("tick.txt", &content); + if tree.times("tick.txt").1 == mtime { + same_tick += 1; + } + assert_eq!(lines_of(&db, "tick.txt"), [content], "round {i}"); + } + eprintln!("{same_tick} of 2000 edits kept the previous mtime"); +} + +#[test] +fn an_edit_whose_times_are_put_back_is_read() { + let tree = Tree::new(); + tree.write("touched.txt", "before"); + let times = tree.times("touched.txt"); + let db = tree.open(); + assert_eq!(lines_of(&db, "touched.txt"), ["before"]); + tree.write("touched.txt", "after!"); + tree.restore_times("touched.txt", times); + assert_eq!(tree.times("touched.txt").1, times.1); + assert_eq!(lines_of(&db, "touched.txt"), ["after!"]); +} + +#[test] +fn an_atomic_save_over_the_original_is_read() { + let tree = Tree::new(); + tree.write("doc.txt", "one"); + let db = tree.open(); + assert_eq!(lines_of(&db, "doc.txt"), ["one"]); + tree.save_atomically("doc.txt", "two"); + assert_eq!(lines_of(&db, "doc.txt"), ["two"]); +} + +#[test] +fn a_file_deleted_and_recreated_under_the_same_name_is_read() { + let tree = Tree::new(); + tree.write("again.txt", "first"); + let db = tree.open(); + assert_eq!(lines_of(&db, "again.txt"), ["first"]); + tree.remove("again.txt"); + tree.write("again.txt", "secnd"); + assert_eq!(lines_of(&db, "again.txt"), ["secnd"]); + tree.remove("again.txt"); + assert!(lines_of(&db, "again.txt").is_empty()); + tree.write("again.txt", "third"); + assert_eq!(lines_of(&db, "again.txt"), ["third"]); +} + +#[test] +fn a_file_renamed_into_the_glob_is_read() { + let tree = Tree::new(); + tree.write("incoming.stage", "arrived"); + let db = tree.open(); + assert!(lines_of(&db, "incoming.txt").is_empty()); + tree.rename("incoming.stage", "incoming.txt"); + assert_eq!(lines_of(&db, "incoming.txt"), ["arrived"]); +} + +#[test] +fn a_file_renamed_out_of_the_glob_is_dropped() { + let tree = Tree::new(); + tree.write("outgoing.txt", "leaving"); + let db = tree.open(); + assert_eq!(lines_of(&db, "outgoing.txt"), ["leaving"]); + tree.rename("outgoing.txt", "outgoing.stage"); + assert!(lines_of(&db, "outgoing.txt").is_empty()); +} + +#[test] +fn a_new_subdirectory_is_read() { + let tree = Tree::new(); + let db = tree.open(); + tree.write("fresh/deep/one.txt", "one"); + tree.write("fresh/two.txt", "two"); + assert_eq!(lines_of(&db, "fresh/deep/one.txt"), ["one"]); + assert_eq!(lines_of(&db, "fresh/two.txt"), ["two"]); + assert_matches_fresh_scan(&db, &tree, "a new subdirectory"); +} + +#[test] +fn a_deleted_subdirectory_is_dropped() { + let tree = Tree::new(); + tree.write("old/deep/one.txt", "one"); + tree.write("old/two.txt", "two"); + let db = tree.open(); + assert_eq!(lines_of(&db, "old/two.txt"), ["two"]); + fs::remove_dir_all(tree.path("old")).unwrap(); + assert!(lines_of(&db, "old/deep/one.txt").is_empty()); + assert!(lines_of(&db, "old/two.txt").is_empty()); + assert_matches_fresh_scan(&db, &tree, "a deleted subdirectory"); +} + +#[test] +fn a_gitignore_change_changes_which_files_are_read() { + let tree = Tree::new(); + fs::create_dir(tree.path(".git")).unwrap(); + tree.write("kept.txt", "kept"); + tree.write("hidden.txt", "hidden"); + let db = tree.open(); + assert_eq!(lines_of(&db, "hidden.txt"), ["hidden"]); + tree.write(".gitignore", "hidden.txt\n"); + assert!(lines_of(&db, "hidden.txt").is_empty()); + assert_matches_fresh_scan(&db, &tree, "adding a .gitignore rule"); + tree.remove(".gitignore"); + assert_eq!(lines_of(&db, "hidden.txt"), ["hidden"]); + assert_matches_fresh_scan(&db, &tree, "removing a .gitignore rule"); +} + +#[test] +fn a_retargeted_symlink_is_read() { + let tree = Tree::new(); + tree.write("targets/one.dat", "one"); + tree.write("targets/two.dat", "two"); + tree.link("link.txt", "targets/one.dat"); + let db = tree.open(); + assert_eq!(lines_of(&db, "link.txt"), ["one"]); + tree.link("link.txt", "targets/two.dat"); + assert_eq!(lines_of(&db, "link.txt"), ["two"]); + assert_matches_fresh_scan(&db, &tree, "retargeting a symlink"); +} + +#[test] +fn an_edit_to_a_symlink_target_is_read() { + let tree = Tree::new(); + tree.write("targets/one.dat", "one"); + tree.link("link.txt", "targets/one.dat"); + let db = tree.open(); + assert_eq!(lines_of(&db, "link.txt"), ["one"]); + tree.write("targets/one.dat", "uno"); + assert_eq!(lines_of(&db, "link.txt"), ["uno"]); +} + +fn version(k: usize) -> String { + (0..50).map(|_| format!("v{k:06}\n")).collect() +} + +/// A file always holds one whole version on disk, so a query that mixes two +/// versions, or reads part of one, tore it. +#[test] +fn a_file_rewritten_while_queries_run_never_reads_torn_and_ends_exact() { + let tree = Arc::new(Tree::new()); + tree.write("busy.txt", &version(0)); + let db = tree.open(); + let done = Arc::new(AtomicBool::new(false)); + let writer = { + let tree = Arc::clone(&tree); + let done = Arc::clone(&done); + std::thread::spawn(move || { + for k in 1..=500 { + tree.save_atomically("busy.txt", &version(k)); + } + done.store(true, Ordering::SeqCst); + }) + }; + while !done.load(Ordering::SeqCst) { + let lines = lines_of(&db, "busy.txt"); + assert_eq!(lines.len(), 50, "a partial version: {lines:?}"); + assert!( + lines.iter().all(|line| line == &lines[0]), + "mixed versions: {lines:?}" + ); + } + writer.join().unwrap(); + assert_eq!(lines_of(&db, "busy.txt"), vec!["v000500".to_string(); 50]); + assert_matches_fresh_scan(&db, &tree, "the concurrent writer finished"); +} + +#[test] +fn a_file_written_in_place_while_queries_run_ends_exact() { + let tree = Arc::new(Tree::new()); + tree.write("inplace.txt", &version(0)); + let db = tree.open(); + let done = Arc::new(AtomicBool::new(false)); + let writer = { + let tree = Arc::clone(&tree); + let done = Arc::clone(&done); + std::thread::spawn(move || { + for k in 1..=500 { + tree.write("inplace.txt", &version(k)); + } + done.store(true, Ordering::SeqCst); + }) + }; + while !done.load(Ordering::SeqCst) { + snapshot(&db); + } + writer.join().unwrap(); + assert_eq!( + lines_of(&db, "inplace.txt"), + vec!["v000500".to_string(); 50] + ); + assert_matches_fresh_scan(&db, &tree, "the in-place writer finished"); +} diff --git a/packages/rust/tests/fresh_declared_events.rs b/packages/rust/tests/fresh_declared_events.rs new file mode 100644 index 000000000..aa86ea156 --- /dev/null +++ b/packages/rust/tests/fresh_declared_events.rs @@ -0,0 +1,77 @@ +//! A change a query reads first still reaches `watch()` and `poll_events()` +//! consumers, exactly once. + +#[path = "common/fresh.rs"] +mod fresh; + +use dirsql::{DirSQL, RowEvent}; +use fresh::{Tree, lines_of, text}; +use std::collections::BTreeMap; +use std::time::{Duration, Instant}; + +const FILES: usize = 200; + +/// How many inserts each `notes` line got, after events stop arriving. +fn tally(mut next: impl FnMut() -> Vec) -> BTreeMap { + let mut counts = BTreeMap::new(); + let deadline = Instant::now() + Duration::from_secs(20); + let mut quiet_since = Instant::now(); + while Instant::now() < deadline { + let batch = next(); + if batch.is_empty() { + if counts.len() >= FILES && quiet_since.elapsed() > Duration::from_secs(1) { + break; + } + std::thread::sleep(Duration::from_millis(10)); + continue; + } + quiet_since = Instant::now(); + for event in batch { + match event { + RowEvent::Insert { table, row, .. } if table == "notes" => { + *counts.entry(text(&row["line"])).or_insert(0) += 1; + } + other => panic!("unexpected event: {other:?}"), + } + } + } + counts +} + +fn expected() -> BTreeMap { + (0..FILES).map(|i| (format!("w{i}"), 1)).collect() +} + +fn write_and_read_each(tree: &Tree, db: &DirSQL) { + for i in 0..FILES { + let rel = format!("w/{i}.txt"); + tree.write(&rel, &format!("w{i}")); + assert_eq!(lines_of(db, &rel), [format!("w{i}")], "file {i}"); + } +} + +#[test] +fn a_watch_stream_gets_each_change_a_query_read_first_once() { + let tree = Tree::new(); + let db = tree.open(); + let mut stream = db.watch().unwrap(); + write_and_read_each(&tree, &db); + let counts = tally(|| { + let mut batch = Vec::new(); + while let Ok(event) = stream.try_recv() { + batch.push(event); + } + batch + }); + assert_eq!(counts, expected()); +} + +#[test] +fn poll_events_returns_each_change_a_query_read_first_once() { + let tree = Tree::new(); + let db = tree.open(); + assert!(db.poll_events(Duration::ZERO).unwrap().is_empty()); + write_and_read_each(&tree, &db); + let counts = tally(|| db.poll_events(Duration::from_millis(50)).unwrap()); + assert_eq!(counts, expected()); +} diff --git a/packages/rust/tests/fresh_declared_hammer.rs b/packages/rust/tests/fresh_declared_hammer.rs new file mode 100644 index 000000000..bc8a21fda --- /dev/null +++ b/packages/rust/tests/fresh_declared_hammer.rs @@ -0,0 +1,104 @@ +//! A query issued the moment a write returns reads that write, every time. + +#[path = "common/fresh.rs"] +mod fresh; + +use fresh::{Tree, lines_of, recs_of}; + +const ROUNDS: usize = 2000; + +#[test] +fn every_create_is_read_by_the_next_query() { + let tree = Tree::new(); + let db = tree.open(); + for i in 0..ROUNDS { + let rel = format!("h/{i}.txt"); + tree.write(&rel, &format!("c{i}")); + assert_eq!(lines_of(&db, &rel), [format!("c{i}")], "round {i}"); + tree.remove(&rel); + } +} + +#[test] +fn every_edit_is_read_by_the_next_query() { + let tree = Tree::new(); + tree.write("e.txt", "start"); + let db = tree.open(); + for i in 0..ROUNDS { + tree.write("e.txt", &format!("e{i}")); + assert_eq!(lines_of(&db, "e.txt"), [format!("e{i}")], "round {i}"); + } +} + +#[test] +fn every_delete_is_read_by_the_next_query() { + let tree = Tree::new(); + let db = tree.open(); + for i in 0..ROUNDS { + tree.write("d.txt", &format!("d{i}")); + assert_eq!(lines_of(&db, "d.txt"), [format!("d{i}")], "round {i}"); + tree.remove("d.txt"); + assert!(lines_of(&db, "d.txt").is_empty(), "round {i}"); + } +} + +#[test] +fn every_rename_into_and_out_of_the_glob_is_read_by_the_next_query() { + let tree = Tree::new(); + let db = tree.open(); + for i in 0..ROUNDS { + tree.write("r.stage", &format!("r{i}")); + tree.rename("r.stage", "r.txt"); + assert_eq!(lines_of(&db, "r.txt"), [format!("r{i}")], "round {i}"); + tree.rename("r.txt", "r.stage"); + assert!(lines_of(&db, "r.txt").is_empty(), "round {i}"); + } +} + +#[test] +fn every_atomic_save_is_read_by_the_next_query() { + let tree = Tree::new(); + tree.write("a.txt", "start"); + let db = tree.open(); + for i in 0..ROUNDS { + tree.save_atomically("a.txt", &format!("a{i}")); + assert_eq!(lines_of(&db, "a.txt"), [format!("a{i}")], "round {i}"); + } +} + +#[test] +fn every_same_size_edit_is_read_by_the_next_query() { + let tree = Tree::new(); + tree.write("s.txt", "s------"); + let db = tree.open(); + for i in 0..ROUNDS { + let content = format!("s{i:06}"); + tree.write("s.txt", &content); + assert_eq!(lines_of(&db, "s.txt"), [content], "round {i}"); + } +} + +#[test] +fn every_same_size_edit_with_its_times_restored_is_read_by_the_next_query() { + let tree = Tree::new(); + tree.write("t.txt", "t------"); + let times = tree.times("t.txt"); + let db = tree.open(); + for i in 0..ROUNDS { + let content = format!("t{i:06}"); + tree.write("t.txt", &content); + tree.restore_times("t.txt", times); + assert_eq!(lines_of(&db, "t.txt"), [content], "round {i}"); + } +} + +#[test] +fn every_edit_to_a_table_hook_file_is_read_by_the_next_query() { + let tree = Tree::new(); + tree.write("rec.json", "{\"k\":\"rec\",\"v\":\"start\"}\n"); + let db = tree.open(); + for i in 0..ROUNDS { + tree.write("rec.json", &format!("{{\"k\":\"rec\",\"v\":\"{i}\"}}\n")); + assert_eq!(recs_of(&db, "rec"), [i.to_string()], "round {i}"); + } +} diff --git a/packages/rust/tests/fresh_declared_oracle.rs b/packages/rust/tests/fresh_declared_oracle.rs new file mode 100644 index 000000000..6052dcb95 --- /dev/null +++ b/packages/rust/tests/fresh_declared_oracle.rs @@ -0,0 +1,164 @@ +//! After any sequence of file operations, a long-lived instance reads exactly +//! what an instance built fresh over the same tree reads. + +#[path = "common/fresh.rs"] +mod fresh; + +use fresh::{Rng, Tree, assert_matches_fresh_scan}; +use std::fs; +use std::time::Duration; + +const DIRS: &[&str] = &["", "a/", "a/b/", "c/"]; +const NAMES: &[&str] = &["n0", "n1", "n2", "n3"]; +const EXTS: &[&str] = &["txt", "json", "stage"]; +const IGNORES: &[&str] = &["n0.txt\n", "a/\n", "*.json\n", "c/n1.*\n"]; + +/// Every file holds JSON lines, so a rename across globs leaves both tables +/// readable. +fn content(rng: &mut Rng, rel: &str) -> String { + let mut text = String::new(); + for line in 0..=rng.below(3) { + text.push_str(&format!( + "{{\"k\":\"{rel}-{line}\",\"v\":\"{:04}\"}}\n", + rng.below(10_000) + )); + } + text +} + +fn random_file(rng: &mut Rng) -> String { + format!("{}{}.{}", rng.pick(DIRS), rng.pick(NAMES), rng.pick(EXTS)) +} + +fn existing_files(tree: &Tree) -> Vec { + let mut files = Vec::new(); + for dir in DIRS { + for name in NAMES { + for ext in EXTS { + let rel = format!("{dir}{name}.{ext}"); + if tree.path(&rel).is_file() { + files.push(rel); + } + } + } + } + files +} + +/// Apply one random operation and say what it was. +fn step(tree: &Tree, rng: &mut Rng) -> String { + let existing = existing_files(tree); + match rng.below(10) { + 0 | 1 => { + let rel = random_file(rng); + let text = content(rng, &rel); + tree.write(&rel, &text); + format!("write {rel}") + } + 2 if !existing.is_empty() => { + let rel = rng.pick(&existing).clone(); + let old = fs::read_to_string(tree.path(&rel)).unwrap(); + let times = tree.times(&rel); + let flipped: String = old + .chars() + .map(|c| match c.to_digit(10) { + Some(d) => char::from_digit((d + 1) % 10, 10).unwrap(), + None => c, + }) + .collect(); + tree.write(&rel, &flipped); + tree.restore_times(&rel, times); + format!("same-size edit of {rel} with its times restored") + } + 3 => { + let rel = random_file(rng); + let text = content(rng, &rel); + tree.save_atomically(&rel, &text); + format!("atomic save of {rel}") + } + 4 if !existing.is_empty() => { + let rel = rng.pick(&existing).clone(); + tree.remove(&rel); + format!("delete {rel}") + } + 5 if !existing.is_empty() => { + let from = rng.pick(&existing).clone(); + let to = random_file(rng); + if from == to { + return "nothing".to_string(); + } + tree.rename(&from, &to); + format!("rename {from} to {to}") + } + 6 => { + let dir = format!("{}new{}", rng.pick(DIRS), rng.below(3)); + for name in &NAMES[..2] { + let rel = format!("{dir}/{name}.{}", rng.pick(EXTS)); + let text = content(rng, &rel); + tree.write(&rel, &text); + } + format!("populate new directory {dir}") + } + 7 => { + let dir = rng.pick(&DIRS[1..]).trim_end_matches('/').to_string(); + if tree.path(&dir).is_dir() { + fs::remove_dir_all(tree.path(&dir)).unwrap(); + } + format!("remove directory {dir}") + } + 8 => { + if rng.below(3) == 0 { + let _ = fs::remove_file(tree.path(".gitignore")); + "remove .gitignore".to_string() + } else { + let rule = rng.pick(IGNORES); + tree.write(".gitignore", rule); + format!("set .gitignore to {rule:?}") + } + } + _ => { + let rel = random_file(rng); + let text = content(rng, &rel); + tree.write(&rel, &text); + format!("write {rel}") + } + } +} + +fn seeded_tree(rng: &mut Rng) -> Tree { + let tree = Tree::new(); + fs::create_dir(tree.path(".git")).unwrap(); + for _ in 0..8 { + let rel = random_file(rng); + let text = content(rng, &rel); + tree.write(&rel, &text); + } + tree +} + +#[test] +fn random_operations_read_what_a_fresh_scan_reads() { + for seed in 1..=4 { + let mut rng = Rng::new(seed); + let tree = seeded_tree(&mut rng); + let db = tree.open(); + for op in 0..150 { + let done = step(&tree, &mut rng); + assert_matches_fresh_scan(&db, &tree, &format!("seed {seed} op {op}: {done}")); + } + } +} + +#[test] +fn a_long_lived_instance_reads_what_a_fresh_scan_reads_across_idle_gaps_and_bursts() { + let mut rng = Rng::new(99); + let tree = seeded_tree(&mut rng); + let db = tree.open(); + for cycle in 0..12 { + std::thread::sleep(Duration::from_millis(250 * (cycle % 4))); + for op in 0..25 { + let done = step(&tree, &mut rng); + assert_matches_fresh_scan(&db, &tree, &format!("cycle {cycle} op {op}: {done}")); + } + } +} diff --git a/packages/ts/tests/integration/binding/fresh-declared.test.ts b/packages/ts/tests/integration/binding/fresh-declared.test.ts new file mode 100644 index 000000000..0f0231cc6 --- /dev/null +++ b/packages/ts/tests/integration/binding/fresh-declared.test.ts @@ -0,0 +1,512 @@ +import { + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + realpathSync, + renameSync, + rmSync, + statSync, + symlinkSync, + unlinkSync, + utimesSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join, sep } from "node:path"; +import { Worker } from "node:worker_threads"; +import { DirSQL, type RowEvent } from "dirsql"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +const ROUNDS = 2000; +const TEST_TIMEOUT = 120_000; + +function notesDb(root: string): DirSQL { + const bases = [realpathSync(root), root]; + return new DirSQL({ + root, + tables: [ + { + name: "notes", + ddl: "CREATE TABLE notes (file TEXT, line TEXT)", + glob: "**/*.txt", + onFile: (filePath: string) => { + let rel = filePath; + for (const base of bases) { + if (filePath.startsWith(base + sep)) { + rel = filePath.slice(base.length + 1); + break; + } + } + rel = rel.split(sep).join("/"); + return readFileSync(filePath, "utf8") + .split("\n") + .filter((line) => line.length > 0) + .map((line) => ({ file: rel, line })); + }, + }, + ], + }); +} + +async function open(root: string): Promise { + const db = notesDb(root); + await db.ready; + return db; +} + +function write(root: string, rel: string, content: string): void { + const path = join(root, rel); + mkdirSync(dirname(path), { recursive: true }); + writeFileSync(path, content); +} + +function saveAtomically(root: string, rel: string, content: string): void { + write(root, `${rel}.stage`, content); + renameSync(join(root, `${rel}.stage`), join(root, rel)); +} + +function times(root: string, rel: string): [Date, Date] { + const st = statSync(join(root, rel)); + return [st.atime, st.mtime]; +} + +function restoreTimes(root: string, rel: string, [atime, mtime]: [Date, Date]) { + utimesSync(join(root, rel), atime, mtime); +} + +async function lines(db: DirSQL, rel: string): Promise { + const rows = await db.query( + `SELECT line FROM notes WHERE file = '${rel}' ORDER BY line`, + ); + return rows.map((row) => String(row.line)); +} + +async function snapshot(db: DirSQL): Promise<[string, string][]> { + const rows = await db.query( + "SELECT file, line FROM notes ORDER BY file, line", + ); + return rows.map((row) => [String(row.file), String(row.line)]); +} + +async function expectMatchesFreshScan( + db: DirSQL, + root: string, + context: string, +): Promise { + const fresh = await open(root); + try { + expect(await snapshot(db), context).toEqual(await snapshot(fresh)); + } finally { + fresh.close(); + } +} + +function rng(seed: number): (n: number) => number { + let state = seed >>> 0; + return (n) => { + state = (state + 0x6d2b79f5) >>> 0; + let t = state; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return (((t ^ (t >>> 14)) >>> 0) / 4294967296) * n; + }; +} + +const DIRS = ["", "a/", "a/b/", "c/"]; +const NAMES = ["n0", "n1", "n2", "n3"]; +const EXTS = ["txt", "json", "stage"]; +const IGNORES = ["n0.txt\n", "a/\n", "*.txt\n", "c/n1.*\n"]; + +function pick(below: (n: number) => number, items: T[]): T { + return items[Math.floor(below(items.length))]; +} + +function content(below: (n: number) => number): string { + let text = ""; + const count = 1 + Math.floor(below(3)); + for (let i = 0; i < count; i++) { + text += `line-${String(Math.floor(below(10_000))).padStart(4, "0")}\n`; + } + return text; +} + +function randomFile(below: (n: number) => number): string { + return `${pick(below, DIRS)}${pick(below, NAMES)}.${pick(below, EXTS)}`; +} + +function existing(root: string): string[] { + const files: string[] = []; + for (const d of DIRS) { + for (const n of NAMES) { + for (const e of EXTS) { + const rel = `${d}${n}.${e}`; + if (existsSync(join(root, rel)) && statSync(join(root, rel)).isFile()) { + files.push(rel); + } + } + } + } + return files; +} + +function step(root: string, below: (n: number) => number): string { + const files = existing(root); + const op = Math.floor(below(9)); + if (op === 1 && files.length > 0) { + const rel = pick(below, files); + const before = times(root, rel); + const flipped = readFileSync(join(root, rel), "utf8").replace(/\d/g, (d) => + String((Number(d) + 1) % 10), + ); + write(root, rel, flipped); + restoreTimes(root, rel, before); + return `same-size edit of ${rel} with its times restored`; + } + if (op === 2) { + const rel = randomFile(below); + saveAtomically(root, rel, content(below)); + return `atomic save of ${rel}`; + } + if (op === 3 && files.length > 0) { + const rel = pick(below, files); + unlinkSync(join(root, rel)); + return `delete ${rel}`; + } + if (op === 4 && files.length > 0) { + const from = pick(below, files); + const to = randomFile(below); + if (from === to) { + return "nothing"; + } + mkdirSync(dirname(join(root, to)), { recursive: true }); + renameSync(join(root, from), join(root, to)); + return `rename ${from} to ${to}`; + } + if (op === 5) { + const dir = `${pick(below, DIRS)}new${Math.floor(below(3))}`; + for (const name of NAMES.slice(0, 2)) { + write(root, `${dir}/${name}.${pick(below, EXTS)}`, content(below)); + } + return `populate new directory ${dir}`; + } + if (op === 6) { + const dir = pick(below, DIRS.slice(1)).replace(/\/$/, ""); + rmSync(join(root, dir), { recursive: true, force: true }); + return `remove directory ${dir}`; + } + if (op === 7) { + if (Math.floor(below(3)) === 0) { + rmSync(join(root, ".gitignore"), { force: true }); + return "remove .gitignore"; + } + const rule = pick(below, IGNORES); + write(root, ".gitignore", rule); + return `set .gitignore to ${JSON.stringify(rule)}`; + } + const rel = randomFile(below); + write(root, rel, content(below)); + return `write ${rel}`; +} + +function seedTree(root: string, below: (n: number) => number): void { + mkdirSync(join(root, ".git")); + for (let i = 0; i < 8; i++) { + write(root, randomFile(below), content(below)); + } +} + +describe("a declared table reads exactly what is on disk", () => { + let dir: string; + let db: DirSQL | undefined; + + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), "dirsql-fresh-")); + db = undefined; + }); + + afterEach(() => { + db?.close(); + rmSync(dir, { recursive: true, force: true }); + }); + + describe("a query right after a write", () => { + it( + "reads every create", + async () => { + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + write(dir, `h/${i}.txt`, `c${i}`); + expect(await lines(db, `h/${i}.txt`), `round ${i}`).toEqual([ + `c${i}`, + ]); + unlinkSync(join(dir, `h/${i}.txt`)); + } + }, + TEST_TIMEOUT, + ); + + it( + "reads every edit", + async () => { + write(dir, "e.txt", "start"); + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + write(dir, "e.txt", `e${i}`); + expect(await lines(db, "e.txt"), `round ${i}`).toEqual([`e${i}`]); + } + }, + TEST_TIMEOUT, + ); + + it( + "reads every delete", + async () => { + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + write(dir, "d.txt", `d${i}`); + expect(await lines(db, "d.txt"), `round ${i}`).toEqual([`d${i}`]); + unlinkSync(join(dir, "d.txt")); + expect(await lines(db, "d.txt"), `round ${i}`).toEqual([]); + } + }, + TEST_TIMEOUT, + ); + + it( + "reads every rename into and out of the glob", + async () => { + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + write(dir, "r.stage", `r${i}`); + renameSync(join(dir, "r.stage"), join(dir, "r.txt")); + expect(await lines(db, "r.txt"), `round ${i}`).toEqual([`r${i}`]); + renameSync(join(dir, "r.txt"), join(dir, "r.stage")); + expect(await lines(db, "r.txt"), `round ${i}`).toEqual([]); + } + }, + TEST_TIMEOUT, + ); + + it( + "reads every atomic save", + async () => { + write(dir, "a.txt", "start"); + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + saveAtomically(dir, "a.txt", `a${i}`); + expect(await lines(db, "a.txt"), `round ${i}`).toEqual([`a${i}`]); + } + }, + TEST_TIMEOUT, + ); + + it( + "reads every same-size edit", + async () => { + write(dir, "s.txt", "s------"); + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + const text = `s${String(i).padStart(6, "0")}`; + write(dir, "s.txt", text); + expect(await lines(db, "s.txt"), `round ${i}`).toEqual([text]); + } + }, + TEST_TIMEOUT, + ); + + it( + "reads every same-size edit with its times restored", + async () => { + write(dir, "t.txt", "t------"); + const before = times(dir, "t.txt"); + db = await open(dir); + for (let i = 0; i < ROUNDS; i++) { + const text = `t${String(i).padStart(6, "0")}`; + write(dir, "t.txt", text); + restoreTimes(dir, "t.txt", before); + expect(await lines(db, "t.txt"), `round ${i}`).toEqual([text]); + } + }, + TEST_TIMEOUT, + ); + }); + + describe("a long-lived instance", () => { + it( + "reads what a fresh scan reads after random operations", + async () => { + for (let seed = 1; seed <= 3; seed++) { + const root = join(dir, `seed${seed}`); + mkdirSync(root); + const below = rng(seed); + seedTree(root, below); + db?.close(); + db = await open(root); + for (let op = 0; op < 100; op++) { + const done = step(root, below); + await expectMatchesFreshScan( + db, + root, + `seed ${seed} op ${op}: ${done}`, + ); + } + } + }, + TEST_TIMEOUT, + ); + + it( + "reads what a fresh scan reads across idle gaps and bursts", + async () => { + const below = rng(99); + seedTree(dir, below); + db = await open(dir); + for (let cycle = 0; cycle < 10; cycle++) { + await new Promise((resolve) => + setTimeout(resolve, 250 * (cycle % 4)), + ); + for (let op = 0; op < 20; op++) { + const done = step(dir, below); + await expectMatchesFreshScan( + db, + dir, + `cycle ${cycle} op ${op}: ${done}`, + ); + } + } + }, + TEST_TIMEOUT, + ); + }); + + describe("each way a change could be missed", () => { + it("reads an edit whose times are put back", async () => { + write(dir, "touched.txt", "before"); + const before = times(dir, "touched.txt"); + db = await open(dir); + expect(await lines(db, "touched.txt")).toEqual(["before"]); + write(dir, "touched.txt", "after!"); + restoreTimes(dir, "touched.txt", before); + expect(await lines(db, "touched.txt")).toEqual(["after!"]); + }); + + it("reads a file deleted and recreated under the same name", async () => { + write(dir, "again.txt", "first"); + db = await open(dir); + expect(await lines(db, "again.txt")).toEqual(["first"]); + unlinkSync(join(dir, "again.txt")); + write(dir, "again.txt", "secnd"); + expect(await lines(db, "again.txt")).toEqual(["secnd"]); + }); + + it("reads a new and a deleted subdirectory", async () => { + db = await open(dir); + write(dir, "sub/deep/one.txt", "one"); + expect(await lines(db, "sub/deep/one.txt")).toEqual(["one"]); + rmSync(join(dir, "sub"), { recursive: true }); + expect(await lines(db, "sub/deep/one.txt")).toEqual([]); + }); + + it("reads a .gitignore change", async () => { + mkdirSync(join(dir, ".git")); + write(dir, "hidden.txt", "hidden"); + db = await open(dir); + expect(await lines(db, "hidden.txt")).toEqual(["hidden"]); + write(dir, ".gitignore", "hidden.txt\n"); + expect(await lines(db, "hidden.txt")).toEqual([]); + unlinkSync(join(dir, ".gitignore")); + expect(await lines(db, "hidden.txt")).toEqual(["hidden"]); + }); + + it("reads a retargeted symlink", async () => { + write(dir, "targets/one.dat", "one"); + write(dir, "targets/two.dat", "two"); + const link = join(dir, "link.txt"); + symlinkSync(join(dir, "targets/one.dat"), link); + db = await open(dir); + expect(await lines(db, "link.txt")).toEqual(["one"]); + unlinkSync(link); + symlinkSync(join(dir, "targets/two.dat"), link); + expect(await lines(db, "link.txt")).toEqual(["two"]); + }); + + it( + "never reads a file torn while it is rewritten, and ends exact", + async () => { + const version = (k: number) => + `v${String(k).padStart(6, "0")}\n`.repeat(50); + write(dir, "busy.txt", version(0)); + db = await open(dir); + const writer = new Worker( + `const { renameSync, writeFileSync } = require("node:fs"); + const { join } = require("node:path"); + const { workerData } = require("node:worker_threads"); + for (let k = 1; k <= 300; k++) { + const text = ("v" + String(k).padStart(6, "0") + "\\n").repeat(50); + writeFileSync(join(workerData, "busy.txt.stage"), text); + renameSync(join(workerData, "busy.txt.stage"), join(workerData, "busy.txt")); + }`, + { eval: true, workerData: dir }, + ); + let finished = false; + const exited = new Promise((resolve, reject) => { + writer.on("error", reject); + writer.on("exit", () => { + finished = true; + resolve(); + }); + }); + while (!finished) { + const seen = await lines(db, "busy.txt"); + expect(seen.length, "a partial version").toBe(50); + expect(new Set(seen).size, "mixed versions").toBe(1); + } + await exited; + expect(await lines(db, "busy.txt")).toEqual( + version(300) + .split("\n") + .filter((line) => line.length > 0), + ); + }, + TEST_TIMEOUT, + ); + }); + + describe("watch consumers", () => { + it( + "stream each change a query read first, once", + async () => { + db = await open(dir); + const stream = db.watch()[Symbol.asyncIterator](); + const files = 100; + for (let i = 0; i < files; i++) { + write(dir, `w/${i}.txt`, `w${i}`); + expect(await lines(db, `w/${i}.txt`), `file ${i}`).toEqual([`w${i}`]); + } + const counts = new Map(); + for (;;) { + const next = await Promise.race([ + stream.next(), + new Promise<"quiet">((resolve) => + setTimeout(resolve, 2000, "quiet"), + ), + ]); + if (next === "quiet" || next.done) { + break; + } + const event: RowEvent = next.value; + expect(event.action).toBe("insert"); + const line = String(event.row?.line); + counts.set(line, (counts.get(line) ?? 0) + 1); + } + expect(Object.fromEntries(counts)).toEqual( + Object.fromEntries( + Array.from({ length: files }, (_, i) => [`w${i}`, 1]), + ), + ); + }, + TEST_TIMEOUT, + ); + }); +}); From 4acbc043b5db3bf11eb6898b163d01099046ac2b Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Tue, 3 Nov 2026 23:05:05 -0500 Subject: [PATCH 02/12] test: keep the Rust freshness tests; binding tests and macOS move out --- .github/workflows/dirsql-rust-ci.yml | 24 - .../binding/fresh_declared_test.py | 353 ------------ .../binding/fresh-declared.test.ts | 512 ------------------ 3 files changed, 889 deletions(-) delete mode 100644 packages/python/tests/integration/binding/fresh_declared_test.py delete mode 100644 packages/ts/tests/integration/binding/fresh-declared.test.ts diff --git a/.github/workflows/dirsql-rust-ci.yml b/.github/workflows/dirsql-rust-ci.yml index 3f56c1ec8..48fa0c246 100644 --- a/.github/workflows/dirsql-rust-ci.yml +++ b/.github/workflows/dirsql-rust-ci.yml @@ -124,30 +124,6 @@ jobs: shell: bash run: 'test ! -e "$LOCALAPPDATA/dirsql" || { echo "::error::a test wrote to the real user cache dir (%LOCALAPPDATA%/dirsql); pass an explicit persist path"; exit 1; }' - macos: - name: macOS tests - runs-on: macos-latest - steps: - - uses: actions/checkout@v6 - - - name: Install Rust - uses: dtolnay/rust-toolchain@stable - - - name: Cache cargo - uses: actions/cache@v4 - with: - path: | - ~/.cargo/registry - ~/.cargo/git - target - key: ${{ runner.os }}-cargo-stable-${{ hashFiles('Cargo.lock') }} - - - name: Run unit and integration tests - run: cargo test --manifest-path packages/rust/Cargo.toml --features cli --tests --no-fail-fast - - - name: Tests left the user cache dir untouched - run: 'test ! -e "$HOME/Library/Caches/dirsql" || { echo "::error::a test wrote to the real user cache dir (~/Library/Caches/dirsql); pass an explicit persist path"; exit 1; }' - coverage: name: Rust unit coverage floor runs-on: ubuntu-latest diff --git a/packages/python/tests/integration/binding/fresh_declared_test.py b/packages/python/tests/integration/binding/fresh_declared_test.py deleted file mode 100644 index d65547e5a..000000000 --- a/packages/python/tests/integration/binding/fresh_declared_test.py +++ /dev/null @@ -1,353 +0,0 @@ -"""Binding-tier tests: a declared table reads exactly what is on disk. - -Every query runs the moment the write before it returns, and must read what -a ``DirSQL`` built fresh over the same tree reads. -""" - -import asyncio -import os -import random -import shutil -import threading - -import pytest - -from dirsql import DirSQL, Table - -ROUNDS = 2000 - - -def _notes(root): - roots = [os.path.realpath(root), root] - - def on_file(path): - rel = path - for base in roots: - if path.startswith(base + os.sep): - rel = path[len(base) + 1 :] - break - rel = rel.replace(os.sep, "/") - with open(path, encoding="utf-8") as f: - text = f.read() - return [{"file": rel, "line": line} for line in text.splitlines() if line] - - return Table( - name="notes", - ddl="CREATE TABLE notes (file TEXT, line TEXT)", - glob="**/*.txt", - on_file=on_file, - ) - - -async def _open(root): - db = DirSQL(root, tables=[_notes(root)]) - await db.ready() - return db - - -def _write(root, rel, content): - path = os.path.join(root, rel) - os.makedirs(os.path.dirname(path), exist_ok=True) - with open(path, "w", encoding="utf-8", newline="") as f: - f.write(content) - - -def _save_atomically(root, rel, content): - _write(root, rel + ".stage", content) - os.replace(os.path.join(root, rel + ".stage"), os.path.join(root, rel)) - - -def _times(root, rel): - st = os.stat(os.path.join(root, rel)) - return (st.st_atime_ns, st.st_mtime_ns) - - -def _restore_times(root, rel, times): - os.utime(os.path.join(root, rel), ns=times) - - -async def _lines(db, rel): - rows = await db.query(f"SELECT line FROM notes WHERE file = '{rel}' ORDER BY line") - return [row["line"] for row in rows] - - -async def _snapshot(db): - rows = await db.query("SELECT file, line FROM notes ORDER BY file, line") - return [(row["file"], row["line"]) for row in rows] - - -async def _assert_matches_fresh_scan(db, root, context): - fresh = await _open(root) - assert await _snapshot(db) == await _snapshot(fresh), ( - f"the long-lived instance disagrees with a fresh scan after {context}" - ) - - -DIRS = ["", "a/", "a/b/", "c/"] -NAMES = ["n0", "n1", "n2", "n3"] -EXTS = ["txt", "json", "stage"] -IGNORES = ["n0.txt\n", "a/\n", "*.txt\n", "c/n1.*\n"] - - -def _content(rng): - return "".join( - f"line-{rng.randrange(10_000):04}\n" for _ in range(rng.randrange(1, 4)) - ) - - -def _random_file(rng): - return f"{rng.choice(DIRS)}{rng.choice(NAMES)}.{rng.choice(EXTS)}" - - -def _existing(root): - return [ - f"{d}{n}.{e}" - for d in DIRS - for n in NAMES - for e in EXTS - if os.path.isfile(os.path.join(root, f"{d}{n}.{e}")) - ] - - -def _step(root, rng): - existing = _existing(root) - op = rng.randrange(9) - if op == 1 and existing: - rel = rng.choice(existing) - times = _times(root, rel) - with open(os.path.join(root, rel), encoding="utf-8") as f: - old = f.read() - flipped = "".join(str((int(c) + 1) % 10) if c.isdigit() else c for c in old) - _write(root, rel, flipped) - _restore_times(root, rel, times) - return f"same-size edit of {rel} with its times restored" - if op == 2: - rel = _random_file(rng) - _save_atomically(root, rel, _content(rng)) - return f"atomic save of {rel}" - if op == 3 and existing: - rel = rng.choice(existing) - os.remove(os.path.join(root, rel)) - return f"delete {rel}" - if op == 4 and existing: - src = rng.choice(existing) - dst = _random_file(rng) - if src == dst: - return "nothing" - os.makedirs(os.path.dirname(os.path.join(root, dst)), exist_ok=True) - os.replace(os.path.join(root, src), os.path.join(root, dst)) - return f"rename {src} to {dst}" - if op == 5: - new = f"{rng.choice(DIRS)}new{rng.randrange(3)}" - for name in NAMES[:2]: - _write(root, f"{new}/{name}.{rng.choice(EXTS)}", _content(rng)) - return f"populate new directory {new}" - if op == 6: - gone = rng.choice(DIRS[1:]).rstrip("/") - shutil.rmtree(os.path.join(root, gone), ignore_errors=True) - return f"remove directory {gone}" - if op == 7: - if rng.randrange(3) == 0: - if os.path.exists(os.path.join(root, ".gitignore")): - os.remove(os.path.join(root, ".gitignore")) - return "remove .gitignore" - rule = rng.choice(IGNORES) - _write(root, ".gitignore", rule) - return f"set .gitignore to {rule!r}" - rel = _random_file(rng) - _write(root, rel, _content(rng)) - return f"write {rel}" - - -def _seed_tree(root, rng): - os.makedirs(os.path.join(root, ".git")) - for _ in range(8): - _write(root, _random_file(rng), _content(rng)) - - -def describe_a_query_right_after_a_write(): - @pytest.mark.asyncio - async def it_reads_every_create(tmp_dir): - db = await _open(tmp_dir) - for i in range(ROUNDS): - _write(tmp_dir, f"h/{i}.txt", f"c{i}") - assert await _lines(db, f"h/{i}.txt") == [f"c{i}"], f"round {i}" - os.remove(os.path.join(tmp_dir, f"h/{i}.txt")) - - @pytest.mark.asyncio - async def it_reads_every_edit(tmp_dir): - _write(tmp_dir, "e.txt", "start") - db = await _open(tmp_dir) - for i in range(ROUNDS): - _write(tmp_dir, "e.txt", f"e{i}") - assert await _lines(db, "e.txt") == [f"e{i}"], f"round {i}" - - @pytest.mark.asyncio - async def it_reads_every_delete(tmp_dir): - db = await _open(tmp_dir) - for i in range(ROUNDS): - _write(tmp_dir, "d.txt", f"d{i}") - assert await _lines(db, "d.txt") == [f"d{i}"], f"round {i}" - os.remove(os.path.join(tmp_dir, "d.txt")) - assert await _lines(db, "d.txt") == [], f"round {i}" - - @pytest.mark.asyncio - async def it_reads_every_rename_into_and_out_of_the_glob(tmp_dir): - db = await _open(tmp_dir) - staged = os.path.join(tmp_dir, "r.stage") - final = os.path.join(tmp_dir, "r.txt") - for i in range(ROUNDS): - _write(tmp_dir, "r.stage", f"r{i}") - os.replace(staged, final) - assert await _lines(db, "r.txt") == [f"r{i}"], f"round {i}" - os.replace(final, staged) - assert await _lines(db, "r.txt") == [], f"round {i}" - - @pytest.mark.asyncio - async def it_reads_every_atomic_save(tmp_dir): - _write(tmp_dir, "a.txt", "start") - db = await _open(tmp_dir) - for i in range(ROUNDS): - _save_atomically(tmp_dir, "a.txt", f"a{i}") - assert await _lines(db, "a.txt") == [f"a{i}"], f"round {i}" - - @pytest.mark.asyncio - async def it_reads_every_same_size_edit(tmp_dir): - _write(tmp_dir, "s.txt", "s------") - db = await _open(tmp_dir) - for i in range(ROUNDS): - _write(tmp_dir, "s.txt", f"s{i:06}") - assert await _lines(db, "s.txt") == [f"s{i:06}"], f"round {i}" - - @pytest.mark.asyncio - async def it_reads_every_same_size_edit_with_its_times_restored(tmp_dir): - _write(tmp_dir, "t.txt", "t------") - times = _times(tmp_dir, "t.txt") - db = await _open(tmp_dir) - for i in range(ROUNDS): - _write(tmp_dir, "t.txt", f"t{i:06}") - _restore_times(tmp_dir, "t.txt", times) - assert await _lines(db, "t.txt") == [f"t{i:06}"], f"round {i}" - - -def describe_a_long_lived_instance(): - @pytest.mark.asyncio - async def it_reads_what_a_fresh_scan_reads_after_random_operations(tmp_dir): - for seed in range(1, 4): - root = os.path.join(tmp_dir, f"seed{seed}") - os.makedirs(root) - rng = random.Random(seed) - _seed_tree(root, rng) - db = await _open(root) - for op in range(100): - done = _step(root, rng) - await _assert_matches_fresh_scan( - db, root, f"seed {seed} op {op}: {done}" - ) - - @pytest.mark.asyncio - async def it_reads_what_a_fresh_scan_reads_across_idle_gaps_and_bursts(tmp_dir): - rng = random.Random(99) - _seed_tree(tmp_dir, rng) - db = await _open(tmp_dir) - for cycle in range(10): - await asyncio.sleep(0.25 * (cycle % 4)) - for op in range(20): - done = _step(tmp_dir, rng) - await _assert_matches_fresh_scan( - db, tmp_dir, f"cycle {cycle} op {op}: {done}" - ) - - -def describe_each_way_a_change_could_be_missed(): - @pytest.mark.asyncio - async def it_reads_an_edit_whose_times_are_put_back(tmp_dir): - _write(tmp_dir, "touched.txt", "before") - times = _times(tmp_dir, "touched.txt") - db = await _open(tmp_dir) - assert await _lines(db, "touched.txt") == ["before"] - _write(tmp_dir, "touched.txt", "after!") - _restore_times(tmp_dir, "touched.txt", times) - assert await _lines(db, "touched.txt") == ["after!"] - - @pytest.mark.asyncio - async def it_reads_a_file_deleted_and_recreated_under_the_same_name(tmp_dir): - _write(tmp_dir, "again.txt", "first") - db = await _open(tmp_dir) - assert await _lines(db, "again.txt") == ["first"] - os.remove(os.path.join(tmp_dir, "again.txt")) - _write(tmp_dir, "again.txt", "secnd") - assert await _lines(db, "again.txt") == ["secnd"] - - @pytest.mark.asyncio - async def it_reads_a_new_and_a_deleted_subdirectory(tmp_dir): - db = await _open(tmp_dir) - _write(tmp_dir, "sub/deep/one.txt", "one") - assert await _lines(db, "sub/deep/one.txt") == ["one"] - shutil.rmtree(os.path.join(tmp_dir, "sub")) - assert await _lines(db, "sub/deep/one.txt") == [] - - @pytest.mark.asyncio - async def it_reads_a_gitignore_change(tmp_dir): - os.makedirs(os.path.join(tmp_dir, ".git")) - _write(tmp_dir, "hidden.txt", "hidden") - db = await _open(tmp_dir) - assert await _lines(db, "hidden.txt") == ["hidden"] - _write(tmp_dir, ".gitignore", "hidden.txt\n") - assert await _lines(db, "hidden.txt") == [] - os.remove(os.path.join(tmp_dir, ".gitignore")) - assert await _lines(db, "hidden.txt") == ["hidden"] - - @pytest.mark.asyncio - async def it_reads_a_retargeted_symlink(tmp_dir): - _write(tmp_dir, "targets/one.dat", "one") - _write(tmp_dir, "targets/two.dat", "two") - link = os.path.join(tmp_dir, "link.txt") - os.symlink(os.path.join(tmp_dir, "targets/one.dat"), link) - db = await _open(tmp_dir) - assert await _lines(db, "link.txt") == ["one"] - os.remove(link) - os.symlink(os.path.join(tmp_dir, "targets/two.dat"), link) - assert await _lines(db, "link.txt") == ["two"] - - @pytest.mark.asyncio - async def it_never_reads_a_file_torn_while_it_is_rewritten_and_ends_exact(tmp_dir): - def version(k): - return f"v{k:06}\n" * 50 - - _write(tmp_dir, "busy.txt", version(0)) - db = await _open(tmp_dir) - done = threading.Event() - - def writer(): - for k in range(1, 301): - _save_atomically(tmp_dir, "busy.txt", version(k)) - done.set() - - thread = threading.Thread(target=writer) - thread.start() - while not done.is_set(): - lines = await _lines(db, "busy.txt") - assert len(lines) == 50 and len(set(lines)) == 1, lines - thread.join() - assert await _lines(db, "busy.txt") == ["v000300"] * 50 - - -def describe_watch_consumers(): - @pytest.mark.asyncio - async def it_streams_each_change_a_query_read_first_once(tmp_dir): - db = await _open(tmp_dir) - stream = db.watch() - files = 100 - for i in range(files): - _write(tmp_dir, f"w/{i}.txt", f"w{i}") - assert await _lines(db, f"w/{i}.txt") == [f"w{i}"], f"file {i}" - counts = {} - while True: - try: - event = await asyncio.wait_for(anext(stream), timeout=2.0) - except TimeoutError: - break - assert event.action == "insert", event - counts[event.row["line"]] = counts.get(event.row["line"], 0) + 1 - assert counts == {f"w{i}": 1 for i in range(files)} diff --git a/packages/ts/tests/integration/binding/fresh-declared.test.ts b/packages/ts/tests/integration/binding/fresh-declared.test.ts deleted file mode 100644 index 0f0231cc6..000000000 --- a/packages/ts/tests/integration/binding/fresh-declared.test.ts +++ /dev/null @@ -1,512 +0,0 @@ -import { - existsSync, - mkdirSync, - mkdtempSync, - readFileSync, - realpathSync, - renameSync, - rmSync, - statSync, - symlinkSync, - unlinkSync, - utimesSync, - writeFileSync, -} from "node:fs"; -import { tmpdir } from "node:os"; -import { dirname, join, sep } from "node:path"; -import { Worker } from "node:worker_threads"; -import { DirSQL, type RowEvent } from "dirsql"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; - -const ROUNDS = 2000; -const TEST_TIMEOUT = 120_000; - -function notesDb(root: string): DirSQL { - const bases = [realpathSync(root), root]; - return new DirSQL({ - root, - tables: [ - { - name: "notes", - ddl: "CREATE TABLE notes (file TEXT, line TEXT)", - glob: "**/*.txt", - onFile: (filePath: string) => { - let rel = filePath; - for (const base of bases) { - if (filePath.startsWith(base + sep)) { - rel = filePath.slice(base.length + 1); - break; - } - } - rel = rel.split(sep).join("/"); - return readFileSync(filePath, "utf8") - .split("\n") - .filter((line) => line.length > 0) - .map((line) => ({ file: rel, line })); - }, - }, - ], - }); -} - -async function open(root: string): Promise { - const db = notesDb(root); - await db.ready; - return db; -} - -function write(root: string, rel: string, content: string): void { - const path = join(root, rel); - mkdirSync(dirname(path), { recursive: true }); - writeFileSync(path, content); -} - -function saveAtomically(root: string, rel: string, content: string): void { - write(root, `${rel}.stage`, content); - renameSync(join(root, `${rel}.stage`), join(root, rel)); -} - -function times(root: string, rel: string): [Date, Date] { - const st = statSync(join(root, rel)); - return [st.atime, st.mtime]; -} - -function restoreTimes(root: string, rel: string, [atime, mtime]: [Date, Date]) { - utimesSync(join(root, rel), atime, mtime); -} - -async function lines(db: DirSQL, rel: string): Promise { - const rows = await db.query( - `SELECT line FROM notes WHERE file = '${rel}' ORDER BY line`, - ); - return rows.map((row) => String(row.line)); -} - -async function snapshot(db: DirSQL): Promise<[string, string][]> { - const rows = await db.query( - "SELECT file, line FROM notes ORDER BY file, line", - ); - return rows.map((row) => [String(row.file), String(row.line)]); -} - -async function expectMatchesFreshScan( - db: DirSQL, - root: string, - context: string, -): Promise { - const fresh = await open(root); - try { - expect(await snapshot(db), context).toEqual(await snapshot(fresh)); - } finally { - fresh.close(); - } -} - -function rng(seed: number): (n: number) => number { - let state = seed >>> 0; - return (n) => { - state = (state + 0x6d2b79f5) >>> 0; - let t = state; - t = Math.imul(t ^ (t >>> 15), t | 1); - t ^= t + Math.imul(t ^ (t >>> 7), t | 61); - return (((t ^ (t >>> 14)) >>> 0) / 4294967296) * n; - }; -} - -const DIRS = ["", "a/", "a/b/", "c/"]; -const NAMES = ["n0", "n1", "n2", "n3"]; -const EXTS = ["txt", "json", "stage"]; -const IGNORES = ["n0.txt\n", "a/\n", "*.txt\n", "c/n1.*\n"]; - -function pick(below: (n: number) => number, items: T[]): T { - return items[Math.floor(below(items.length))]; -} - -function content(below: (n: number) => number): string { - let text = ""; - const count = 1 + Math.floor(below(3)); - for (let i = 0; i < count; i++) { - text += `line-${String(Math.floor(below(10_000))).padStart(4, "0")}\n`; - } - return text; -} - -function randomFile(below: (n: number) => number): string { - return `${pick(below, DIRS)}${pick(below, NAMES)}.${pick(below, EXTS)}`; -} - -function existing(root: string): string[] { - const files: string[] = []; - for (const d of DIRS) { - for (const n of NAMES) { - for (const e of EXTS) { - const rel = `${d}${n}.${e}`; - if (existsSync(join(root, rel)) && statSync(join(root, rel)).isFile()) { - files.push(rel); - } - } - } - } - return files; -} - -function step(root: string, below: (n: number) => number): string { - const files = existing(root); - const op = Math.floor(below(9)); - if (op === 1 && files.length > 0) { - const rel = pick(below, files); - const before = times(root, rel); - const flipped = readFileSync(join(root, rel), "utf8").replace(/\d/g, (d) => - String((Number(d) + 1) % 10), - ); - write(root, rel, flipped); - restoreTimes(root, rel, before); - return `same-size edit of ${rel} with its times restored`; - } - if (op === 2) { - const rel = randomFile(below); - saveAtomically(root, rel, content(below)); - return `atomic save of ${rel}`; - } - if (op === 3 && files.length > 0) { - const rel = pick(below, files); - unlinkSync(join(root, rel)); - return `delete ${rel}`; - } - if (op === 4 && files.length > 0) { - const from = pick(below, files); - const to = randomFile(below); - if (from === to) { - return "nothing"; - } - mkdirSync(dirname(join(root, to)), { recursive: true }); - renameSync(join(root, from), join(root, to)); - return `rename ${from} to ${to}`; - } - if (op === 5) { - const dir = `${pick(below, DIRS)}new${Math.floor(below(3))}`; - for (const name of NAMES.slice(0, 2)) { - write(root, `${dir}/${name}.${pick(below, EXTS)}`, content(below)); - } - return `populate new directory ${dir}`; - } - if (op === 6) { - const dir = pick(below, DIRS.slice(1)).replace(/\/$/, ""); - rmSync(join(root, dir), { recursive: true, force: true }); - return `remove directory ${dir}`; - } - if (op === 7) { - if (Math.floor(below(3)) === 0) { - rmSync(join(root, ".gitignore"), { force: true }); - return "remove .gitignore"; - } - const rule = pick(below, IGNORES); - write(root, ".gitignore", rule); - return `set .gitignore to ${JSON.stringify(rule)}`; - } - const rel = randomFile(below); - write(root, rel, content(below)); - return `write ${rel}`; -} - -function seedTree(root: string, below: (n: number) => number): void { - mkdirSync(join(root, ".git")); - for (let i = 0; i < 8; i++) { - write(root, randomFile(below), content(below)); - } -} - -describe("a declared table reads exactly what is on disk", () => { - let dir: string; - let db: DirSQL | undefined; - - beforeEach(() => { - dir = mkdtempSync(join(tmpdir(), "dirsql-fresh-")); - db = undefined; - }); - - afterEach(() => { - db?.close(); - rmSync(dir, { recursive: true, force: true }); - }); - - describe("a query right after a write", () => { - it( - "reads every create", - async () => { - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - write(dir, `h/${i}.txt`, `c${i}`); - expect(await lines(db, `h/${i}.txt`), `round ${i}`).toEqual([ - `c${i}`, - ]); - unlinkSync(join(dir, `h/${i}.txt`)); - } - }, - TEST_TIMEOUT, - ); - - it( - "reads every edit", - async () => { - write(dir, "e.txt", "start"); - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - write(dir, "e.txt", `e${i}`); - expect(await lines(db, "e.txt"), `round ${i}`).toEqual([`e${i}`]); - } - }, - TEST_TIMEOUT, - ); - - it( - "reads every delete", - async () => { - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - write(dir, "d.txt", `d${i}`); - expect(await lines(db, "d.txt"), `round ${i}`).toEqual([`d${i}`]); - unlinkSync(join(dir, "d.txt")); - expect(await lines(db, "d.txt"), `round ${i}`).toEqual([]); - } - }, - TEST_TIMEOUT, - ); - - it( - "reads every rename into and out of the glob", - async () => { - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - write(dir, "r.stage", `r${i}`); - renameSync(join(dir, "r.stage"), join(dir, "r.txt")); - expect(await lines(db, "r.txt"), `round ${i}`).toEqual([`r${i}`]); - renameSync(join(dir, "r.txt"), join(dir, "r.stage")); - expect(await lines(db, "r.txt"), `round ${i}`).toEqual([]); - } - }, - TEST_TIMEOUT, - ); - - it( - "reads every atomic save", - async () => { - write(dir, "a.txt", "start"); - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - saveAtomically(dir, "a.txt", `a${i}`); - expect(await lines(db, "a.txt"), `round ${i}`).toEqual([`a${i}`]); - } - }, - TEST_TIMEOUT, - ); - - it( - "reads every same-size edit", - async () => { - write(dir, "s.txt", "s------"); - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - const text = `s${String(i).padStart(6, "0")}`; - write(dir, "s.txt", text); - expect(await lines(db, "s.txt"), `round ${i}`).toEqual([text]); - } - }, - TEST_TIMEOUT, - ); - - it( - "reads every same-size edit with its times restored", - async () => { - write(dir, "t.txt", "t------"); - const before = times(dir, "t.txt"); - db = await open(dir); - for (let i = 0; i < ROUNDS; i++) { - const text = `t${String(i).padStart(6, "0")}`; - write(dir, "t.txt", text); - restoreTimes(dir, "t.txt", before); - expect(await lines(db, "t.txt"), `round ${i}`).toEqual([text]); - } - }, - TEST_TIMEOUT, - ); - }); - - describe("a long-lived instance", () => { - it( - "reads what a fresh scan reads after random operations", - async () => { - for (let seed = 1; seed <= 3; seed++) { - const root = join(dir, `seed${seed}`); - mkdirSync(root); - const below = rng(seed); - seedTree(root, below); - db?.close(); - db = await open(root); - for (let op = 0; op < 100; op++) { - const done = step(root, below); - await expectMatchesFreshScan( - db, - root, - `seed ${seed} op ${op}: ${done}`, - ); - } - } - }, - TEST_TIMEOUT, - ); - - it( - "reads what a fresh scan reads across idle gaps and bursts", - async () => { - const below = rng(99); - seedTree(dir, below); - db = await open(dir); - for (let cycle = 0; cycle < 10; cycle++) { - await new Promise((resolve) => - setTimeout(resolve, 250 * (cycle % 4)), - ); - for (let op = 0; op < 20; op++) { - const done = step(dir, below); - await expectMatchesFreshScan( - db, - dir, - `cycle ${cycle} op ${op}: ${done}`, - ); - } - } - }, - TEST_TIMEOUT, - ); - }); - - describe("each way a change could be missed", () => { - it("reads an edit whose times are put back", async () => { - write(dir, "touched.txt", "before"); - const before = times(dir, "touched.txt"); - db = await open(dir); - expect(await lines(db, "touched.txt")).toEqual(["before"]); - write(dir, "touched.txt", "after!"); - restoreTimes(dir, "touched.txt", before); - expect(await lines(db, "touched.txt")).toEqual(["after!"]); - }); - - it("reads a file deleted and recreated under the same name", async () => { - write(dir, "again.txt", "first"); - db = await open(dir); - expect(await lines(db, "again.txt")).toEqual(["first"]); - unlinkSync(join(dir, "again.txt")); - write(dir, "again.txt", "secnd"); - expect(await lines(db, "again.txt")).toEqual(["secnd"]); - }); - - it("reads a new and a deleted subdirectory", async () => { - db = await open(dir); - write(dir, "sub/deep/one.txt", "one"); - expect(await lines(db, "sub/deep/one.txt")).toEqual(["one"]); - rmSync(join(dir, "sub"), { recursive: true }); - expect(await lines(db, "sub/deep/one.txt")).toEqual([]); - }); - - it("reads a .gitignore change", async () => { - mkdirSync(join(dir, ".git")); - write(dir, "hidden.txt", "hidden"); - db = await open(dir); - expect(await lines(db, "hidden.txt")).toEqual(["hidden"]); - write(dir, ".gitignore", "hidden.txt\n"); - expect(await lines(db, "hidden.txt")).toEqual([]); - unlinkSync(join(dir, ".gitignore")); - expect(await lines(db, "hidden.txt")).toEqual(["hidden"]); - }); - - it("reads a retargeted symlink", async () => { - write(dir, "targets/one.dat", "one"); - write(dir, "targets/two.dat", "two"); - const link = join(dir, "link.txt"); - symlinkSync(join(dir, "targets/one.dat"), link); - db = await open(dir); - expect(await lines(db, "link.txt")).toEqual(["one"]); - unlinkSync(link); - symlinkSync(join(dir, "targets/two.dat"), link); - expect(await lines(db, "link.txt")).toEqual(["two"]); - }); - - it( - "never reads a file torn while it is rewritten, and ends exact", - async () => { - const version = (k: number) => - `v${String(k).padStart(6, "0")}\n`.repeat(50); - write(dir, "busy.txt", version(0)); - db = await open(dir); - const writer = new Worker( - `const { renameSync, writeFileSync } = require("node:fs"); - const { join } = require("node:path"); - const { workerData } = require("node:worker_threads"); - for (let k = 1; k <= 300; k++) { - const text = ("v" + String(k).padStart(6, "0") + "\\n").repeat(50); - writeFileSync(join(workerData, "busy.txt.stage"), text); - renameSync(join(workerData, "busy.txt.stage"), join(workerData, "busy.txt")); - }`, - { eval: true, workerData: dir }, - ); - let finished = false; - const exited = new Promise((resolve, reject) => { - writer.on("error", reject); - writer.on("exit", () => { - finished = true; - resolve(); - }); - }); - while (!finished) { - const seen = await lines(db, "busy.txt"); - expect(seen.length, "a partial version").toBe(50); - expect(new Set(seen).size, "mixed versions").toBe(1); - } - await exited; - expect(await lines(db, "busy.txt")).toEqual( - version(300) - .split("\n") - .filter((line) => line.length > 0), - ); - }, - TEST_TIMEOUT, - ); - }); - - describe("watch consumers", () => { - it( - "stream each change a query read first, once", - async () => { - db = await open(dir); - const stream = db.watch()[Symbol.asyncIterator](); - const files = 100; - for (let i = 0; i < files; i++) { - write(dir, `w/${i}.txt`, `w${i}`); - expect(await lines(db, `w/${i}.txt`), `file ${i}`).toEqual([`w${i}`]); - } - const counts = new Map(); - for (;;) { - const next = await Promise.race([ - stream.next(), - new Promise<"quiet">((resolve) => - setTimeout(resolve, 2000, "quiet"), - ), - ]); - if (next === "quiet" || next.done) { - break; - } - const event: RowEvent = next.value; - expect(event.action).toBe("insert"); - const line = String(event.row?.line); - counts.set(line, (counts.get(line) ?? 0) + 1); - } - expect(Object.fromEntries(counts)).toEqual( - Object.fromEntries( - Array.from({ length: files }, (_, i) => [`w${i}`, 1]), - ), - ); - }, - TEST_TIMEOUT, - ); - }); -}); From e110ec495215fce002104e75bafdeb99fde06851 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:07:44 -0500 Subject: [PATCH 03/12] fix: a query on a declared table reads the files changed since the last read Each query walks the table anchors, fingerprints every match, and re-reads the files whose fingerprint moved or whose last read may have overlapped a write. Rows it changes reach the next poll or watch batch once. --- Cargo.lock | 1 + docs/howto/parse-files-into-columns.md | 5 +- docs/howto/react-to-changes.md | 3 + docs/reference/sdk.md | 12 +- packages/rust/Cargo.toml | 3 + .../2026-11-03-fresh-declared-reads.md | 3 + .../2026-11-03-fresh-declared-reads.md | 21 + packages/rust/src/cli/run.rs | 4 + packages/rust/src/freshness.rs | 236 ++++++++++ packages/rust/src/lib.rs | 445 +++++++++++++++++- packages/rust/tests/foreign_keys.rs | 18 + packages/rust/tests/persist.rs | 6 +- packages/rust/tests/watch_scan_correctness.rs | 8 +- .../2026-11-03-fresh-declared-reads.md | 3 + .../2026-11-03-fresh-declared-reads.md | 21 + packages/ts/napi/src/lib.rs | 2 + 16 files changed, 775 insertions(+), 16 deletions(-) create mode 100644 packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md create mode 100644 packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md create mode 100644 packages/rust/src/freshness.rs create mode 100644 packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md create mode 100644 packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md diff --git a/Cargo.lock b/Cargo.lock index 460b7c9db..6d9169cb5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -610,6 +610,7 @@ dependencies = [ "toml", "tower-http", "typed-path", + "windows-sys 0.61.2", ] [[package]] diff --git a/docs/howto/parse-files-into-columns.md b/docs/howto/parse-files-into-columns.md index 5bc294cef..f55e38f91 100644 --- a/docs/howto/parse-files-into-columns.md +++ b/docs/howto/parse-files-into-columns.md @@ -83,7 +83,7 @@ the fastest way to see whether a parser produces the rows you expect. The inline flag re-scans and re-parses every file on every query, defines one parser for the whole query, and forgets everything when the process exits. When -you want a **watcher** that keeps the rows fresh, **persistence** across +you want **row events** as files change, **persistence** across restarts, or **different parsers for different file sets**, move the parser into a `.dirsql.toml` [`[[table]]`](../reference/config.md#table) — and paste the command in verbatim: @@ -110,7 +110,8 @@ dirsql query "SELECT title, author FROM posts ORDER BY title" -c ./.dirsql.toml ``` Same command, same rows. The difference is what a declared table brings: it is -indexed on build, kept fresh by the watcher, survives restarts with +indexed on build, re-reads only the files that changed on each query, streams +row events as files change, survives restarts with [`--persist`](./persist.md), and a config can declare [many tables](./define-tables.md) — each with its own `on-file` — where the flag gives every path-table one parser. In both spellings the table's columns are diff --git a/docs/howto/react-to-changes.md b/docs/howto/react-to-changes.md index d9889439b..189613c55 100644 --- a/docs/howto/react-to-changes.md +++ b/docs/howto/react-to-changes.md @@ -72,6 +72,9 @@ works — `EventSource` in a browser, an SSE client library, or plain ## Notes +- Querying never waits on this stream: a query reads what is on disk when it + starts ([`query`](../reference/sdk.md#query)). A change a query reads first + still arrives here, once. - Events reflect *row* changes, not raw filesystem events: a file edit that leaves a table's rows identical emits nothing, and one file change can emit several events. The diffing model is part of diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md index 74eedbd7b..dd5f36a4b 100644 --- a/docs/reference/sdk.md +++ b/docs/reference/sdk.md @@ -224,6 +224,14 @@ db.query(sql: &str) -> Result> // Row = HashMap Executes a SQL query and returns rows keyed by column name. +- **Accurate to the filesystem.** A query reads what is on disk when it + starts: a file written, edited, renamed or deleted before the call is + reflected in its rows, with no watcher, wait or retry. Before reading, the + query walks each named table's glob and stats every match; a file whose stat + moved, or that changed within the last two seconds, is parsed again, and a + file that is gone loses its rows. The row events those re-reads produce reach + [`watch`](#watch) streams and the next poll, once each. Rust and Python + today; a TypeScript query reads what the watcher has applied. - **Read-only.** Each statement is classified by SQLite itself (`sqlite3_stmt_readonly`); anything classified as a write — `INSERT`, `UPDATE`, `DELETE`, `DROP`, `CREATE`, `ALTER`, `VACUUM`, … — is rejected @@ -303,7 +311,9 @@ to stop. The watcher is live once both `watch()` and [`ready`](#ready) have returned, in either order (a Rust `DirSQL` is ready when `build()` returns); a file written after that point produces an event and reaches the index. A write -made while the initial scan is still running may be missed. +made while the initial scan is still running may be missed by the stream, but +never by a query (see [`query`](#query)). A change a query read first still +arrives on the stream, once. An instance with no named tables cannot be watched: path-tables emit no watch events, so `watch()` (and the polling primitives) raise an error naming diff --git a/packages/rust/Cargo.toml b/packages/rust/Cargo.toml index 136f73532..6473102b0 100644 --- a/packages/rust/Cargo.toml +++ b/packages/rust/Cargo.toml @@ -95,6 +95,9 @@ typed-path = "0.12" [target.'cfg(unix)'.dependencies] rustix = { version = "1", features = ["fs"] } +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem"] } + [lints.rust] # `deny`, not `forbid`: rusqlite's API imposes five unavoidable `unsafe` # sites (the `load_extension` guard and four `unsafe impl VTab`/`VTabCursor`). diff --git a/packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md b/packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md new file mode 100644 index 000000000..fc48867f5 --- /dev/null +++ b/packages/rust/changelog.d/2026-11-03-fresh-declared-reads.md @@ -0,0 +1,3 @@ +**Fixed** + +A query on a long-lived `DirSQL` (Rust, and Python through it) now reads what is on disk when it starts. Before reading, it walks each named table's glob, re-parses every file whose stat moved or that changed within the last two seconds, and drops the rows of files that are gone, so a write made just before a query is never missed. The row events those re-reads produce reach `watch()` streams and the next `poll_events()`, once each. The one-shot `dirsql query` and the TypeScript binding read the index as built, unchanged. diff --git a/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md new file mode 100644 index 000000000..b5f5682cd --- /dev/null +++ b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md @@ -0,0 +1,21 @@ +### Queries read what is on disk + +#### Summary + +A query on a long-lived Rust or Python `DirSQL` brings its named tables in line with the files on disk before reading. No API or configuration changes are required. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +A query re-parses the files that changed since the index last read them, so code that waited on a watch event or slept before querying no longer needs to. A change a query reads first still arrives on `watch()` and `poll_events()`, once. + +#### Verification + +Build a `DirSQL` over a directory with a declared table, write a matching file, then query the table with no wait; the row for the new file is in the result. diff --git a/packages/rust/src/cli/run.rs b/packages/rust/src/cli/run.rs index e4b700b31..d08ea2c22 100644 --- a/packages/rust/src/cli/run.rs +++ b/packages/rust/src/cli/run.rs @@ -413,6 +413,10 @@ async fn run_query(args: QueryArgs, out: &mut dyn Write) -> u8 { } }; let state = load_state(&args.common, parser); + // The query runs straight after the build's own read of every file. + if let AppState::Ready(db) = &state { + db.set_fresh_reads(false); + } // Unbounded: the process IS the query, so `timeout(1)` expresses any cap // natively; only the long-lived server enforces `query_timeout` (408). diff --git a/packages/rust/src/freshness.rs b/packages/rust/src/freshness.rs new file mode 100644 index 000000000..b5568de5e --- /dev/null +++ b/packages/rust/src/freshness.rs @@ -0,0 +1,236 @@ +//! Which files a query must read again so its rows match the files on disk. +//! +//! A file's change time moves on every write and cannot be set back by +//! `utime`, so once it sits [`SETTLE_NS`] before the moment a read began, the +//! read saw the file's final content and an identical stat later proves the +//! file unchanged. Until then the file is read on every query. + +use std::path::Path; + +/// How far before a read's start a file's change time must sit for that read +/// to have seen its final content. Covers kernels whose timestamps lag the +/// clock by a tick and filesystems that keep whole seconds. +pub(crate) const SETTLE_NS: i64 = 2_000_000_000; + +/// The stat of a file, following symlinks: what a change to it must alter. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct Fingerprint { + pub(crate) dev: u64, + pub(crate) ino: u64, + pub(crate) size: u64, + pub(crate) mtime_ns: i64, + /// `None` where the platform reports no change time: such a file never + /// settles, so every query reads it. + pub(crate) ctime_ns: Option, +} + +/// What the index holds for one `(table, file)`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Known { + /// Indexed by the build and not stat'd since. + Built, + /// Last read with the file at `fp` (`None`: it could not be stat'd). + Read { + fp: Option, + settled: bool, + }, +} + +/// Whether a file a query found must be read for its rows to be current, +/// given what the index holds for it and its stat now. +pub(crate) fn must_read(known: Option<&Known>, now: Option<&Fingerprint>, built_ns: i64) -> bool { + match (known, now) { + (None, _) | (_, None) => true, + (Some(Known::Built), Some(now)) => !changed_before(now, built_ns), + (Some(Known::Read { fp, settled }), Some(now)) => !settled || fp.as_ref() != Some(now), + } +} + +/// Whether `fp` last changed far enough before `start_ns` that a read begun +/// then saw its final content. +pub(crate) fn changed_before(fp: &Fingerprint, start_ns: i64) -> bool { + fp.ctime_ns + .is_some_and(|ctime| ctime < start_ns.saturating_sub(SETTLE_NS)) +} + +/// What the index records for a file read at `start_ns` and stat'd after. +pub(crate) fn after_read(fp: Option, start_ns: i64) -> Known { + Known::Read { + fp, + settled: fp.is_some_and(|fp| changed_before(&fp, start_ns)), + } +} + +#[cfg(unix)] +pub(crate) fn fingerprint(path: &Path) -> std::io::Result { + use std::os::unix::fs::MetadataExt; + let meta = std::fs::metadata(path)?; + Ok(Fingerprint { + dev: meta.dev(), + ino: meta.ino(), + size: meta.size(), + mtime_ns: unix_ns(meta.mtime(), meta.mtime_nsec()), + ctime_ns: Some(unix_ns(meta.ctime(), meta.ctime_nsec())), + }) +} + +#[cfg(unix)] +fn unix_ns(secs: i64, nanos: i64) -> i64 { + secs.saturating_mul(1_000_000_000).saturating_add(nanos) +} + +/// Windows keeps the change time (NTFS `ChangeTime`) out of `std`'s stable +/// metadata, so it is read from the handle, with the volume serial and file +/// index standing in for `dev` and `ino`. +#[cfg(windows)] +#[expect( + unsafe_code, + reason = "GetFileInformationByHandle(Ex) are the only way to read ChangeTime and the file index" +)] +pub(crate) fn fingerprint(path: &Path) -> std::io::Result { + use std::os::windows::fs::OpenOptionsExt; + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::Storage::FileSystem::{ + BY_HANDLE_FILE_INFORMATION, FILE_BASIC_INFO, FILE_READ_ATTRIBUTES, FileBasicInfo, + GetFileInformationByHandle, GetFileInformationByHandleEx, + }; + let file = std::fs::OpenOptions::new() + .access_mode(FILE_READ_ATTRIBUTES) + .open(path)?; + let handle = file.as_raw_handle(); + // SAFETY: both structs are plain data for which all-zero is valid, and + // each call writes at most the size it is given into its own struct + // through a handle `file` keeps open. + let mut info: BY_HANDLE_FILE_INFORMATION = unsafe { std::mem::zeroed() }; + if unsafe { GetFileInformationByHandle(handle, &mut info) } == 0 { + return Err(std::io::Error::last_os_error()); + } + let mut basic: FILE_BASIC_INFO = unsafe { std::mem::zeroed() }; + let size = u32::try_from(std::mem::size_of::()).expect("small struct"); + if unsafe { GetFileInformationByHandleEx(handle, FileBasicInfo, (&raw mut basic).cast(), size) } + == 0 + { + return Err(std::io::Error::last_os_error()); + } + Ok(Fingerprint { + dev: u64::from(info.dwVolumeSerialNumber), + ino: (u64::from(info.nFileIndexHigh) << 32) | u64::from(info.nFileIndexLow), + size: (u64::from(info.nFileSizeHigh) << 32) | u64::from(info.nFileSizeLow), + mtime_ns: windows_ns(basic.LastWriteTime), + ctime_ns: (basic.ChangeTime != 0).then(|| windows_ns(basic.ChangeTime)), + }) +} + +/// A Windows file time (100ns ticks since 1601) as nanoseconds since 1970. +#[cfg(windows)] +fn windows_ns(ticks: i64) -> i64 { + ticks + .saturating_sub(116_444_736_000_000_000) + .saturating_mul(100) +} + +#[cfg(not(any(unix, windows)))] +pub(crate) fn fingerprint(path: &Path) -> std::io::Result { + let meta = std::fs::metadata(path)?; + Ok(Fingerprint { + dev: 0, + ino: 0, + size: meta.len(), + mtime_ns: 0, + ctime_ns: None, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fp(ctime_ns: Option) -> Fingerprint { + Fingerprint { + dev: 1, + ino: 2, + size: 3, + mtime_ns: 4, + ctime_ns, + } + } + + const START: i64 = 10 * SETTLE_NS; + + #[test] + fn a_change_time_settles_only_strictly_before_the_margin() { + assert!(changed_before(&fp(Some(START - SETTLE_NS - 1)), START)); + assert!(!changed_before(&fp(Some(START - SETTLE_NS)), START)); + assert!(!changed_before(&fp(Some(START)), START)); + } + + #[test] + fn a_file_without_a_change_time_never_settles() { + assert!(!changed_before(&fp(None), START)); + } + + #[test] + fn the_margin_does_not_overflow_at_the_earliest_time() { + assert!(!changed_before(&fp(Some(i64::MIN)), i64::MIN)); + } + + #[test] + fn a_file_the_index_has_never_held_is_read() { + assert!(must_read(None, Some(&fp(Some(0))), START)); + } + + #[test] + fn a_file_that_cannot_be_stat_ed_is_read() { + let known = after_read(Some(fp(Some(0))), START); + assert!(must_read(Some(&known), None, START)); + } + + #[test] + fn a_built_file_is_read_unless_it_changed_before_the_build_began() { + let old = fp(Some(START - SETTLE_NS - 1)); + let recent = fp(Some(START - SETTLE_NS)); + assert!(!must_read(Some(&Known::Built), Some(&old), START)); + assert!(must_read(Some(&Known::Built), Some(&recent), START)); + } + + #[test] + fn a_settled_file_is_read_again_only_when_its_stat_moves() { + let old = fp(Some(0)); + let known = after_read(Some(old), START); + assert!(!must_read(Some(&known), Some(&old), START)); + let moved = Fingerprint { size: 9, ..old }; + assert!(must_read(Some(&known), Some(&moved), START)); + } + + #[test] + fn an_unsettled_file_is_read_on_every_query() { + let recent = fp(Some(START)); + let known = after_read(Some(recent), START); + assert!(must_read(Some(&known), Some(&recent), START)); + } + + #[test] + fn a_read_settles_a_file_only_when_its_change_time_is_old() { + assert_eq!( + after_read(Some(fp(Some(0))), START), + Known::Read { + fp: Some(fp(Some(0))), + settled: true + } + ); + assert_eq!( + after_read(Some(fp(Some(START))), START), + Known::Read { + fp: Some(fp(Some(START))), + settled: false + } + ); + assert_eq!( + after_read(None, START), + Known::Read { + fp: None, + settled: false + } + ); + } +} diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 15901a513..fe22dbc29 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -20,6 +20,7 @@ pub mod differ; pub mod extension_resolution; #[doc(hidden)] pub mod foreign_key; +mod freshness; #[doc(hidden)] pub mod functions; mod gitignore_frame; @@ -77,7 +78,7 @@ use std::collections::{HashMap, HashSet}; use std::error::Error as StdError; use std::path::{Path, PathBuf}; use std::sync::{ - Arc, Mutex, + Arc, Mutex, MutexGuard, atomic::{AtomicBool, Ordering}, }; use std::thread; @@ -450,9 +451,30 @@ struct DirSqlInner { /// Locks out [`DirSQL::poll_events`] once [`DirSQL::watch`] has spawned /// its background thread. watch_thread_started: AtomicBool, + /// Set once a watcher is wanted; from then on a query hands the row + /// events of the changes it reads to the watch consumer. + watching: AtomicBool, poll_interval: Duration, /// Filesystem seam: [`RealFs`] in production; unit tests inject a double. fs: Arc, + /// Whether a query first brings the named tables in line with the files + /// on disk. Off where a query cannot run `on_file` on its own thread. + fresh_reads: AtomicBool, + /// When the build's walk began. + built_ns: i64, + /// Held across every change to the index, so a query's reconcile and a + /// watch batch never interleave and their events reach consumers in the + /// order they were applied. + fresh: Mutex, +} + +#[derive(Default)] +struct FreshState { + /// What the index holds for each `(table, file)`. + known: HashMap<(String, String), freshness::Known>, + seeded: bool, + /// Row events a query's reconcile produced, for the next watch batch. + outbox: Vec, } fn watch_scopes(groups: &[AnchorGroup]) -> Vec { @@ -584,10 +606,141 @@ impl DirSQL { /// [`QueryResult::columns`]. A [`Row`] is a `HashMap`, so a caller that /// renders columns in the order the user asked for needs this instead. pub fn query_ordered(&self, sql: &str) -> Result { + let _fresh = self.refresh()?; let db = self.inner.db.lock().map_err(DirSqlError::lock)?; db.query_ordered(sql).map_err(map_db_error) } + /// Turn off reading the named tables' files at query time, for a caller + /// that queries once, straight after the build, or that cannot run + /// `on_file` on the thread a query runs on. + #[doc(hidden)] + pub fn set_fresh_reads(&self, on: bool) { + self.inner.fresh_reads.store(on, Ordering::SeqCst); + } + + /// 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>> { + if self.inner.groups.is_empty() || !self.inner.fresh_reads.load(Ordering::SeqCst) { + return Ok(None); + } + let mut state = self.inner.fresh.lock().map_err(DirSqlError::lock)?; + self.reconcile(&mut state)?; + Ok(Some(state)) + } + + /// Walk each anchor, read every file whose stat says it may differ from + /// what the index holds, and drop the rows of files that are gone, in one + /// transaction. + fn reconcile(&self, state: &mut FreshState) -> Result<()> { + let start_ns = now_ns(); + if !state.seeded { + self.seed(state)?; + } + let mut found: Vec<(String, String, PathBuf)> = Vec::new(); + for group in &self.inner.groups { + for (path, table) in + self.inner + .fs + .scan_subtree(&group.anchor, &group.anchor, &group.matcher) + { + let rel = scanner::to_slash(path.strip_prefix(&group.anchor).unwrap_or(&path)); + found.push((table, rel, path)); + } + } + self.with_db(|db| Ok(foreign_key::begin(db.conn())?))?; + let mut stats: HashMap> = HashMap::new(); + let mut updates: Vec<((String, String), freshness::Known)> = Vec::new(); + let mut pending = PendingRefresh::default(); + let mut events = Vec::new(); + for (table, rel, path) in &found { + let now = *stats + .entry(path.clone()) + .or_insert_with(|| self.inner.fs.fingerprint(path).ok()); + let key = (table.clone(), rel.clone()); + let known = state.known.get(&key); + if !freshness::must_read(known, now.as_ref(), self.inner.built_ns) { + if known == Some(&freshness::Known::Built) { + updates.push((key, freshness::after_read(now, self.inner.built_ns))); + } + continue; + } + if self.is_per_table(table) { + self.track_batch_file(table, path, rel, false); + pending.mark(table, rel); + } else { + events.extend(self.read_file(table, path, rel)); + updates.push(( + key, + freshness::after_read(self.inner.fs.fingerprint(path).ok(), start_ns), + )); + } + } + let present: HashSet<(&str, &str)> = found + .iter() + .map(|(table, rel, _)| (table.as_str(), rel.as_str())) + .collect(); + let gone: Vec<(String, String)> = state + .known + .keys() + .filter(|(table, rel)| !present.contains(&(table.as_str(), rel.as_str()))) + .cloned() + .collect(); + for (table, rel) in &gone { + if self.is_per_table(table) { + self.track_batch_file(table, Path::new(rel), rel, true); + pending.mark(table, rel); + } else { + events.extend(self.handle_delete(table, rel)); + } + } + let rerun: HashSet = pending.0.iter().map(|(table, _)| table.clone()).collect(); + events.extend(self.refresh_pending(pending)); + for (table, rel, path) in found.iter().filter(|(table, _, _)| rerun.contains(table)) { + updates.push(( + (table.clone(), rel.clone()), + freshness::after_read(self.inner.fs.fingerprint(path).ok(), start_ns), + )); + } + self.with_db(|db| foreign_key::commit(db.conn()))?; + for key in gone { + state.known.remove(&key); + } + state.known.extend(updates); + if self.inner.watching.load(Ordering::SeqCst) { + state.outbox.extend(events); + } + Ok(()) + } + + /// Record every file the build indexed as [`freshness::Known::Built`]. + fn seed(&self, state: &mut FreshState) -> Result<()> { + let owned = self.with_db(Db::owned_files)?; + let batch_files = self.inner.batch_files.lock().map_err(DirSqlError::lock)?; + let batched = batch_files + .iter() + .flat_map(|(table, files)| files.iter().map(|rel| (table.clone(), rel.clone()))); + for key in owned.into_iter().chain(batched) { + state.known.insert(key, freshness::Known::Built); + } + state.seeded = true; + Ok(()) + } + + /// Re-read one file of a per-file table. A file whose hook fails keeps no + /// rows, as a fresh build would give it. + fn read_file(&self, table: &str, path: &Path, rel: &str) -> Vec { + let mut events = self.handle_upsert(table, path, rel); + if events + .iter() + .any(|event| matches!(event, RowEvent::Error { .. })) + { + events.extend(self.handle_delete(table, rel)); + } + events + } + /// The directory the index is rooted at: where path-tables resolve. pub fn root(&self) -> &Path { &self.inner.root @@ -609,6 +762,7 @@ impl DirSQL { start: impl FnOnce(&[String]) -> S, on_row: impl FnMut(&mut S, &[Value]), ) -> Result { + let _fresh = self.refresh()?; let db = self.inner.db.lock().map_err(DirSqlError::lock)?; db.query_each(sql, start, on_row).map_err(map_db_error) } @@ -622,6 +776,7 @@ impl DirSQL { start: impl FnOnce(&[String]) -> S, on_row: impl FnMut(&mut S, &[rusqlite::types::ValueRef<'_>]), ) -> Result { + let _fresh = self.refresh()?; let db = self.inner.db.lock().map_err(DirSqlError::lock)?; db.query_each_ref(sql, start, on_row).map_err(map_db_error) } @@ -633,9 +788,15 @@ impl DirSQL { self.require_named_tables()?; let mut guard = self.inner.watcher.lock().map_err(DirSqlError::lock)?; if guard.is_none() { + // Set before the watcher exists so a change a query reads while + // it starts still reaches the consumer. + self.inner.watching.store(true, Ordering::SeqCst); // Watch the canonicalized anchors, never the (possibly relative) // user-supplied ones — `notify` misbehaves on relative paths. - let watcher = Watcher::scoped(self.watch_scopes()).map_err(DirSqlError::watch)?; + let watcher = Watcher::scoped(self.watch_scopes()).map_err(|e| { + self.inner.watching.store(false, Ordering::SeqCst); + DirSqlError::watch(e) + })?; *guard = Some(watcher); } Ok(()) @@ -718,6 +879,21 @@ impl DirSQL { /// before any per-table hook runs. One write reaches the watcher as /// several events; a hook run for the first would run again for the rest. fn apply_file_events_draining( + &self, + events: Vec, + late: impl FnMut() -> Vec, + ) -> Vec { + let mut fresh = self + .inner + .fresh + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut out = std::mem::take(&mut fresh.outbox); + out.extend(self.apply_events(events, late)); + out + } + + fn apply_events( &self, events: Vec, mut late: impl FnMut() -> Vec, @@ -1396,6 +1572,7 @@ impl DirSQL { path_table_parser, no_ignore, } = resolved; + let built_ns = now_ns(); let (groups, anchors, table_names) = compile_groups(&tables, &ignore, &root)?; let groups = groups @@ -1519,6 +1696,7 @@ impl DirSQL { poll_interval, path_table_parser, no_ignore, + built_ns, }) } @@ -1571,6 +1749,7 @@ impl DirSQL { hint_legacy_files_table, path_table_parser, no_ignore, + built_ns, } = prepared; let (mut db, fresh_ddl, ddl_warnings, persist_ready, needs_sweep) = match persist { @@ -1859,8 +2038,12 @@ impl DirSQL { watcher: Mutex::new(None), poll_used: AtomicBool::new(false), watch_thread_started: AtomicBool::new(false), + watching: AtomicBool::new(false), poll_interval, fs, + fresh_reads: AtomicBool::new(true), + built_ns, + fresh: Mutex::new(FreshState::default()), }), }) } @@ -2262,6 +2445,8 @@ pub struct PreparedBuild { hint_legacy_files_table: bool, path_table_parser: Option, no_ignore: bool, + /// When the walk began. + built_ns: i64, } #[doc(hidden)] @@ -2520,6 +2705,7 @@ trait FileSystem: Send + Sync { fn hash(&self, path: &Path) -> std::io::Result<[u8; 32]>; /// Canonicalize the watch root, falling back to the literal path. fn canonical_root(&self, root: &Path) -> String; + fn fingerprint(&self, path: &Path) -> std::io::Result; } struct RealFs; @@ -2550,6 +2736,10 @@ impl FileSystem for RealFs { hash_file(path) } + fn fingerprint(&self, path: &Path) -> std::io::Result { + freshness::fingerprint(path) + } + fn canonical_root(&self, root: &Path) -> String { canonical_root(root) } @@ -3486,6 +3676,16 @@ mod internal_tests { }) } + fn fingerprint(&self, path: &Path) -> std::io::Result { + self.stat(path).map(|stat| freshness::Fingerprint { + dev: stat.dev.unsigned_abs(), + ino: stat.inode.unsigned_abs(), + size: stat.size.unsigned_abs(), + mtime_ns: stat.mtime_ns, + ctime_ns: Some(stat.ctime_ns), + }) + } + fn canonical_root(&self, root: &Path) -> String { self.canonical_roots .get(root) @@ -3897,6 +4097,7 @@ mod internal_tests { hint_legacy_files_table: false, path_table_parser: None, no_ignore: false, + built_ns: 0, }; assert!(DirSQL::finish_build(prepared).is_err()); } @@ -4071,7 +4272,8 @@ mod internal_tests { } /// A one-table db whose fake fs stats every file in `files`, each already - /// upserted so the db holds a row per file. + /// upserted so the db holds a row per file. Queries read the index as + /// applied, since the fake walk lists no files. fn populated_db(root: &Path, glob: &str, files: &[&str]) -> DirSQL { let mut fake = FakeFs::default(); for f in files { @@ -4094,6 +4296,7 @@ mod internal_tests { Arc::new(fake), ) .unwrap(); + db.set_fresh_reads(false); for f in files { let events = db.handle_upsert("items", &root.join(f), f); assert!( @@ -6144,6 +6347,7 @@ mod internal_tests { hint_legacy_files_table: false, path_table_parser: None, no_ignore: false, + built_ns: 0, } } @@ -6666,11 +6870,13 @@ mod internal_tests { DirSQL::with_ignore_and_fs(&root, vec![items], Vec::::new(), Arc::new(RealFs)) .unwrap(); - let calls = calls.lock().unwrap(); - assert_eq!(calls.len(), 1, "{calls:?}"); - let mut seen = calls[0].clone(); - seen.sort(); - assert_eq!(seen, vec![root.join("a.txt"), root.join("b.txt")]); + { + let calls = calls.lock().unwrap(); + assert_eq!(calls.len(), 1, "{calls:?}"); + let mut seen = calls[0].clone(); + seen.sort(); + assert_eq!(seen, vec![root.join("a.txt"), root.join("b.txt")]); + } let mut names = row_names(&db); names.sort(); assert_eq!(names, vec!["a.txt", "b.txt"]); @@ -7219,6 +7425,7 @@ mod internal_tests { fake.clone(), ) .unwrap(); + db.set_fresh_reads(false); let events = db.apply_file_events( files .iter() @@ -7346,6 +7553,228 @@ mod internal_tests { ); assert_eq!(held(&db), Vec::new()); } + + type Reads = Arc>>; + + /// A per-file `items` table over `**/*.txt` whose hook records each file + /// it reads and answers one row holding the file's content, failing for + /// content `bad`. + fn content_table() -> (Table, Reads) { + let reads: Reads = Arc::default(); + let seen = Arc::clone(&reads); + let table = Table::try_new( + "items", + "CREATE TABLE items (name TEXT)", + "**/*.txt", + move |path| { + seen.lock().unwrap().push( + Path::new(path) + .file_name() + .unwrap() + .to_string_lossy() + .into_owned(), + ); + let text = std::fs::read_to_string(path).unwrap_or_default(); + if text == "bad" { + return Err("bad content".into()); + } + Ok(vec![Row::from_iter([( + "name".to_string(), + Value::Text(text), + )])]) + }, + ); + (table, reads) + } + + fn fresh_fixture(files: &[(&str, &str)]) -> (TempDir, PathBuf, DirSQL, Reads) { + let dir = TempDir::new().unwrap(); + let root = dir.path().canonicalize().unwrap(); + for (rel, text) in files { + std::fs::write(root.join(rel), text).unwrap(); + } + let (table, reads) = content_table(); + let db = + DirSQL::with_ignore_and_fs(&root, vec![table], Vec::::new(), Arc::new(RealFs)) + .unwrap(); + (dir, root, db, reads) + } + + #[test] + fn a_query_reads_a_file_written_after_the_build() { + let (_dir, root, db, _) = fresh_fixture(&[]); + std::fs::write(root.join("a.txt"), "one").unwrap(); + assert_eq!(row_names(&db), vec!["one"]); + } + + #[test] + fn a_query_reads_an_edit_to_a_file_the_build_indexed() { + let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]); + std::fs::write(root.join("a.txt"), "two").unwrap(); + assert_eq!(row_names(&db), vec!["two"]); + } + + #[test] + fn a_query_drops_the_rows_of_a_deleted_file() { + let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one"), ("b.txt", "two")]); + std::fs::remove_file(root.join("a.txt")).unwrap(); + assert_eq!(row_names(&db), vec!["two"]); + assert_eq!(row_names(&db), vec!["two"]); + } + + #[test] + fn a_query_with_fresh_reads_off_reads_the_index_as_built() { + let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]); + db.set_fresh_reads(false); + std::fs::write(root.join("b.txt"), "two").unwrap(); + assert_eq!(row_names(&db), vec!["one"]); + db.set_fresh_reads(true); + assert_eq!(row_names(&db), vec!["one", "two"]); + } + + #[test] + fn query_ordered_and_query_each_read_fresh_rows_too() { + let (_dir, root, db, _) = fresh_fixture(&[]); + std::fs::write(root.join("a.txt"), "one").unwrap(); + assert_eq!( + db.query_ordered("SELECT name FROM items") + .unwrap() + .rows + .len(), + 1 + ); + std::fs::write(root.join("b.txt"), "two").unwrap(); + let count = db + .query_each("SELECT name FROM items", |_| 0, |n, _| *n += 1) + .unwrap(); + assert_eq!(count, 2); + } + + #[test] + fn a_file_whose_hook_fails_at_query_time_keeps_no_rows_and_reports_the_error() { + let (_dir, root, db, _) = fresh_fixture(&[("a.txt", "one")]); + db.inner.watching.store(true, Ordering::SeqCst); + std::fs::write(root.join("a.txt"), "bad").unwrap(); + assert_eq!(row_names(&db), Vec::::new()); + let events = db.apply_file_events(Vec::new()); + assert!( + events + .iter() + .any(|event| matches!(event, RowEvent::Error { .. })), + "{events:?}" + ); + assert!( + events + .iter() + .any(|event| matches!(event, RowEvent::Delete { .. })), + "{events:?}" + ); + } + + #[test] + fn a_query_hands_its_row_events_to_the_next_poll_once() { + let (_dir, root, db, _) = fresh_fixture(&[]); + db.inner.watching.store(true, Ordering::SeqCst); + std::fs::write(root.join("a.txt"), "one").unwrap(); + row_names(&db); + assert_eq!( + rescan_summary(&db.apply_file_events(Vec::new())), + vec!["insert one"] + ); + assert_eq!(db.apply_file_events(Vec::new()), vec![]); + } + + #[test] + fn a_query_after_start_watching_hands_its_row_events_to_the_consumer() { + let (_dir, root, db, _) = fresh_fixture(&[]); + db.start_watching().unwrap(); + std::fs::write(root.join("a.txt"), "one").unwrap(); + row_names(&db); + assert_eq!( + rescan_summary(&db.apply_file_events(Vec::new())), + vec!["insert one"] + ); + } + + #[test] + fn a_query_keeps_no_row_events_when_nothing_consumes_them() { + let (_dir, root, db, _) = fresh_fixture(&[]); + std::fs::write(root.join("a.txt"), "one").unwrap(); + row_names(&db); + db.inner.watching.store(true, Ordering::SeqCst); + assert_eq!(db.apply_file_events(Vec::new()), vec![]); + } + + #[test] + fn a_change_a_query_read_reaches_the_next_poll_once_when_the_watcher_reports_it_too() { + let (_dir, root, db, _) = fresh_fixture(&[]); + db.inner.watching.store(true, Ordering::SeqCst); + std::fs::write(root.join("a.txt"), "one").unwrap(); + row_names(&db); + let events = db.apply_file_events(vec![FileEvent::Created(root.join("a.txt"))]); + assert_eq!(rescan_summary(&events), vec!["insert one"]); + } + + /// A built instance whose filesystem double reports `found` under the + /// root, each file last changed long before any query. + fn settled_fixture(built: &[&str], found: &[&str]) -> (TempDir, DirSQL, Reads) { + let dir = TempDir::new().unwrap(); + let root = dir.path().canonicalize().unwrap(); + for rel in built { + std::fs::write(root.join(rel), "x").unwrap(); + } + let fs = FakeFs::default().with_subtree( + &root, + found + .iter() + .map(|rel| (root.join(rel), "items".to_string())) + .collect(), + ); + let (table, reads) = content_table(); + let db = DirSQL::with_ignore_and_fs(&root, vec![table], Vec::::new(), Arc::new(fs)) + .unwrap(); + reads.lock().unwrap().clear(); + (dir, db, reads) + } + + #[test] + fn a_built_file_unchanged_since_before_the_build_is_not_read_again() { + let (_dir, db, reads) = settled_fixture(&["a.txt"], &["a.txt"]); + row_names(&db); + row_names(&db); + assert_eq!(*reads.lock().unwrap(), Vec::::new()); + } + + #[test] + fn a_settled_file_is_read_once_then_not_again() { + let (_dir, db, reads) = settled_fixture(&[], &["a.txt"]); + row_names(&db); + row_names(&db); + assert_eq!(*reads.lock().unwrap(), vec!["a.txt"]); + } + + #[test] + fn a_per_table_hook_reruns_when_one_of_its_files_changes_and_when_one_goes() { + let dir = TempDir::new().unwrap(); + let root = dir.path().canonicalize().unwrap(); + std::fs::write(root.join("a.txt"), "x").unwrap(); + let (items, calls) = recording_per_table("items", "*.txt"); + let db = + DirSQL::with_ignore_and_fs(&root, vec![items], Vec::::new(), Arc::new(RealFs)) + .unwrap(); + std::fs::write(root.join("b.txt"), "x").unwrap(); + assert_eq!(row_names(&db), vec!["a.txt", "b.txt"]); + assert_eq!( + calls.lock().unwrap().last().unwrap(), + &vec![root.join("a.txt"), root.join("b.txt")] + ); + std::fs::remove_file(root.join("a.txt")).unwrap(); + assert_eq!(row_names(&db), vec!["b.txt"]); + assert_eq!( + calls.lock().unwrap().last().unwrap(), + &vec![root.join("b.txt")] + ); + } } #[cfg(test)] diff --git a/packages/rust/tests/foreign_keys.rs b/packages/rust/tests/foreign_keys.rs index 0f97b8734..dc2365b1a 100644 --- a/packages/rust/tests/foreign_keys.rs +++ b/packages/rust/tests/foreign_keys.rs @@ -173,6 +173,7 @@ fn a_parent_and_its_child_arriving_in_one_batch_both_land() { #[test] fn orphaning_a_child_in_watch_reports_it_and_keeps_the_previous_state() { let (_dir, root, db) = family(&[("1", "alpha"), ("2", "beta")], &[("kid", "1")]); + db.set_fresh_reads(false); let parent = root.join("parents").join("1.txt"); fs::remove_file(&parent).unwrap(); @@ -211,6 +212,23 @@ fn orphaning_a_child_in_watch_reports_it_and_keeps_the_previous_state() { ); } +#[test] +fn a_query_over_an_orphaned_child_fails_as_the_build_would() { + let (_dir, root, db) = family(&[("1", "alpha")], &[("kid", "1")]); + fs::remove_file(root.join("parents").join("1.txt")).unwrap(); + + let message = db + .query("SELECT name FROM parent") + .unwrap_err() + .to_string(); + + for needle in ["child", "parent", "children/kid.txt", "parent_id = 1"] { + assert!(message.contains(needle), "{needle:?} missing from: {message}"); + } + fs::write(root.join("parents").join("1.txt"), "alpha again").unwrap(); + assert_eq!(names(&db, "SELECT name FROM parent"), vec!["alpha again"]); +} + #[test] fn a_rejected_batch_lands_with_the_batch_that_resolves_it() { let (_dir, root, db) = family(&[("1", "alpha"), ("2", "beta")], &[("kid", "1")]); diff --git a/packages/rust/tests/persist.rs b/packages/rust/tests/persist.rs index ae4f1ca59..153aca23b 100644 --- a/packages/rust/tests/persist.rs +++ b/packages/rust/tests/persist.rs @@ -378,6 +378,8 @@ fn persist_fans_out_and_composite_key_round_trips() { let cb = Arc::new(AtomicUsize::new(0)); { let db = open_two(root.path(), ca.clone(), cb.clone()); + assert_eq!(ca.swap(0, Ordering::SeqCst), 1, "ta extracted once"); + assert_eq!(cb.swap(0, Ordering::SeqCst), 1, "tb extracted once"); assert_eq!(db.query("SELECT col_a FROM ta").unwrap().len(), 1); assert_eq!( db.query("SELECT col_b FROM tb").unwrap().len(), @@ -385,8 +387,8 @@ fn persist_fans_out_and_composite_key_round_trips() { "second-declared table populated on cold start" ); } - assert_eq!(ca.swap(0, Ordering::SeqCst), 1, "ta extracted once"); - assert_eq!(cb.swap(0, Ordering::SeqCst), 1, "tb extracted once"); + ca.store(0, Ordering::SeqCst); + cb.store(0, Ordering::SeqCst); // Warm start over the same root/cache: both (rel_path, table) entries are // trusted, so neither table re-extracts, yet both still serve the row. diff --git a/packages/rust/tests/watch_scan_correctness.rs b/packages/rust/tests/watch_scan_correctness.rs index c7e158e6a..037844144 100644 --- a/packages/rust/tests/watch_scan_correctness.rs +++ b/packages/rust/tests/watch_scan_correctness.rs @@ -98,7 +98,7 @@ fn rename_out_deletes_rows() { let mut saw_delete = false; let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline && paths(&db).iter().any(|p| p == "a.txt") { + while Instant::now() < deadline && !saw_delete { for e in db.poll_events(Duration::from_millis(200)).unwrap() { if matches!(e, RowEvent::Delete { .. }) { saw_delete = true; @@ -129,6 +129,7 @@ fn dir_moved_into_root_indexes_its_files() { fs::write(stage.path().join("moved/a.txt"), "alpha").unwrap(); let db = DirSQL::new(dir.path(), vec![files_table(dir.path())]).unwrap(); + db.set_fresh_reads(false); db.start_watching().unwrap(); std::thread::sleep(Duration::from_millis(250)); @@ -163,6 +164,7 @@ fn dir_moved_into_root_indexes_nested_files() { fs::write(stage.path().join("moved/one/two/deep.txt"), "deep").unwrap(); let db = DirSQL::new(dir.path(), vec![files_table(dir.path())]).unwrap(); + db.set_fresh_reads(false); db.start_watching().unwrap(); std::thread::sleep(Duration::from_millis(250)); @@ -227,7 +229,7 @@ fn dir_moved_out_of_root_deletes_its_rows() { let mut deleted = Vec::new(); let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline && !under_leaving(&db).is_empty() { + while Instant::now() < deadline && deleted.len() < 2 { for e in db.poll_events(Duration::from_millis(200)).unwrap() { if let RowEvent::Delete { file_path, .. } = e { deleted.push(file_path); @@ -279,7 +281,7 @@ fn dir_renamed_within_root_deletes_rows_under_old_name() { let mut saw_delete = false; let deadline = Instant::now() + Duration::from_secs(5); - while Instant::now() < deadline && paths(&db).iter().any(|p| p.starts_with("a/")) { + while Instant::now() < deadline && !saw_delete { for e in db.poll_events(Duration::from_millis(200)).unwrap() { if matches!(&e, RowEvent::Delete { file_path, .. } if file_path == "a/x.txt") { saw_delete = true; diff --git a/packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md b/packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md new file mode 100644 index 000000000..f10d7f215 --- /dev/null +++ b/packages/ts/changelog.d/2026-11-03-fresh-declared-reads.md @@ -0,0 +1,3 @@ +**Changed** + +The TypeScript binding keeps queries reading the index as the watcher applied it, while the core now re-reads changed files at query time for Rust and Python: a TypeScript query cannot run a JavaScript `onFile` on the thread it runs on. diff --git a/packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md b/packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md new file mode 100644 index 000000000..3a1bbb803 --- /dev/null +++ b/packages/ts/migrations.d/2026-11-03-fresh-declared-reads.md @@ -0,0 +1,21 @@ +### Queries keep reading the index as applied + +#### Summary + +The TypeScript binding opts out of the core's new query-time re-read. Nothing changes for TypeScript callers and nothing breaks. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +_None._ + +#### Verification + +`await db.query("SELECT 1 AS n")` returns `[{ n: 1 }]`. diff --git a/packages/ts/napi/src/lib.rs b/packages/ts/napi/src/lib.rs index 5f5dbd1c7..e1d0b9257 100644 --- a/packages/ts/napi/src/lib.rs +++ b/packages/ts/napi/src/lib.rs @@ -447,6 +447,8 @@ impl Task for OpenTask { fn resolve(&mut self, _env: Env, output: Self::Output) -> Result { let inner = CoreDirSQL::finish_build(output).map_err(to_napi_err)?; + // A query runs on a worker thread, where `on_file` cannot be called. + inner.set_fresh_reads(false); Ok(DirSQL { inner: RefCell::new(Some(inner)), }) From 2b45d6ae988714c87794da65fc3656ac9e683736 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:08:14 -0500 Subject: [PATCH 04/12] docs: a query fails on a foreign-key orphan as the build does --- docs/reference/sdk.md | 6 ++++-- .../rust/migrations.d/2026-11-03-fresh-declared-reads.md | 2 ++ 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md index dd5f36a4b..6df058efb 100644 --- a/docs/reference/sdk.md +++ b/docs/reference/sdk.md @@ -230,8 +230,10 @@ Executes a SQL query and returns rows keyed by column name. query walks each named table's glob and stats every match; a file whose stat moved, or that changed within the last two seconds, is parsed again, and a file that is gone loses its rows. The row events those re-reads produce reach - [`watch`](#watch) streams and the next poll, once each. Rust and Python - today; a TypeScript query reads what the watcher has applied. + [`watch`](#watch) streams and the next poll, once each. If a file on disk + breaks a foreign key, the query fails with the error a build over that tree + returns. Rust and Python today; a TypeScript query reads what the watcher + has applied. - **Read-only.** Each statement is classified by SQLite itself (`sqlite3_stmt_readonly`); anything classified as a write — `INSERT`, `UPDATE`, `DELETE`, `DROP`, `CREATE`, `ALTER`, `VACUUM`, … — is rejected diff --git a/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md index b5f5682cd..68fd4a494 100644 --- a/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md +++ b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md @@ -16,6 +16,8 @@ _None._ A query re-parses the files that changed since the index last read them, so code that waited on a watch event or slept before querying no longer needs to. A change a query reads first still arrives on `watch()` and `poll_events()`, once. +Each query walks its tables' globs and stats every match, so it costs about what the same query on a path-table costs. When a file on disk breaks a foreign key, a query fails with the same `ForeignKeyViolation` a build over that tree returns, until the file is fixed. + #### Verification Build a `DirSQL` over a directory with a declared table, write a matching file, then query the table with no wait; the row for the new file is in the result. From f07ad122fcaa2a78c25f6985320612296ac8d697 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:47:25 -0500 Subject: [PATCH 05/12] e2e attestation for bot/1572-fresh-declared-1103 --- .../e2e-attestations/bot-1572-fresh-declared-1103.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 packages/python/e2e-attestations/bot-1572-fresh-declared-1103.json diff --git a/packages/python/e2e-attestations/bot-1572-fresh-declared-1103.json b/packages/python/e2e-attestations/bot-1572-fresh-declared-1103.json new file mode 100644 index 000000000..ae1fc3dae --- /dev/null +++ b/packages/python/e2e-attestations/bot-1572-fresh-declared-1103.json @@ -0,0 +1,7 @@ +{ + "command": "uv run python -m pytest tests/e2e/declared_table_name_test.py tests/e2e/fanout_test.py tests/e2e/on_file_query_test.py tests/e2e/query_contract_test.py tests/e2e/watch_speed_test.py tests/e2e/server_speed_test.py tests/e2e/repl_interactive_test.py tests/e2e/sigint_interrupt_test.py tests/e2e/use_case_dossier_test.py tests/e2e/use_case_join_test.py tests/e2e/use_case_notebook_test.py tests/e2e/use_case_transcripts_test.py tests/e2e/frontmatter_parse_speed_test.py tests/e2e/jsonl_parse_speed_test.py -q", + "ran_at": 1793771245, + "exit_code": 0, + "commit": "2b45d6ae988714c87794da65fc3656ac9e683736", + "branch": "bot/1572-fresh-declared-1103" +} From ed69641349002e95f5055cd4f92dc0793f38fb4c Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:48:23 -0500 Subject: [PATCH 06/12] e2e attestation for bot/1572-fresh-declared-1103 --- .../ts/e2e-attestations/bot-1572-fresh-declared-1103.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 packages/ts/e2e-attestations/bot-1572-fresh-declared-1103.json diff --git a/packages/ts/e2e-attestations/bot-1572-fresh-declared-1103.json b/packages/ts/e2e-attestations/bot-1572-fresh-declared-1103.json new file mode 100644 index 000000000..eb7f45df0 --- /dev/null +++ b/packages/ts/e2e-attestations/bot-1572-fresh-declared-1103.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1793771303, + "exit_code": 0, + "commit": "f07ad122fcaa2a78c25f6985320612296ac8d697", + "branch": "bot/1572-fresh-declared-1103" +} From df1064b049df059b96675efba4eec60c2cc44855 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 00:59:25 -0500 Subject: [PATCH 07/12] style: rustfmt the foreign-key query test --- packages/rust/tests/foreign_keys.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/rust/tests/foreign_keys.rs b/packages/rust/tests/foreign_keys.rs index dc2365b1a..f2f6cc7f5 100644 --- a/packages/rust/tests/foreign_keys.rs +++ b/packages/rust/tests/foreign_keys.rs @@ -217,13 +217,13 @@ fn a_query_over_an_orphaned_child_fails_as_the_build_would() { let (_dir, root, db) = family(&[("1", "alpha")], &[("kid", "1")]); fs::remove_file(root.join("parents").join("1.txt")).unwrap(); - let message = db - .query("SELECT name FROM parent") - .unwrap_err() - .to_string(); + let message = db.query("SELECT name FROM parent").unwrap_err().to_string(); for needle in ["child", "parent", "children/kid.txt", "parent_id = 1"] { - assert!(message.contains(needle), "{needle:?} missing from: {message}"); + assert!( + message.contains(needle), + "{needle:?} missing from: {message}" + ); } fs::write(root.join("parents").join("1.txt"), "alpha again").unwrap(); assert_eq!(names(&db, "SELECT name FROM parent"), vec!["alpha again"]); From 242095542d64a7f65ceb863cce47dc4bb3a44363 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 01:32:49 -0500 Subject: [PATCH 08/12] test: a built file alone is adopted as read when the build began Moves the adopt decision into freshness so a unit test pins it, and exempts the Windows and fallback fingerprint bodies the Linux mutation runner compiles out. --- packages/rust/src/freshness.rs | 32 ++++++++++++++++++++++++++++++++ packages/rust/src/lib.rs | 4 ++-- testing-conventions.toml | 6 ++++++ 3 files changed, 40 insertions(+), 2 deletions(-) diff --git a/packages/rust/src/freshness.rs b/packages/rust/src/freshness.rs index b5568de5e..778079ede 100644 --- a/packages/rust/src/freshness.rs +++ b/packages/rust/src/freshness.rs @@ -61,6 +61,16 @@ pub(crate) fn after_read(fp: Option, start_ns: i64) -> Known { } } +/// What the index records for a file a query need not read: a built file +/// counts as read when the build began; any other entry stands. +pub(crate) fn adopted( + known: Option<&Known>, + now: Option, + built_ns: i64, +) -> Option { + (known == Some(&Known::Built)).then(|| after_read(now, built_ns)) +} + #[cfg(unix)] pub(crate) fn fingerprint(path: &Path) -> std::io::Result { use std::os::unix::fs::MetadataExt; @@ -164,6 +174,28 @@ mod tests { assert!(!changed_before(&fp(Some(START)), START)); } + #[test] + fn a_built_file_is_adopted_as_read_when_the_build_began() { + let now = fp(Some(START - SETTLE_NS - 1)); + assert_eq!( + adopted(Some(&Known::Built), Some(now), START), + Some(Known::Read { + fp: Some(now), + settled: true + }) + ); + } + + #[test] + fn a_file_already_read_or_unknown_is_not_adopted() { + let read = Known::Read { + fp: Some(fp(Some(0))), + settled: true, + }; + assert_eq!(adopted(Some(&read), Some(fp(Some(0))), START), None); + assert_eq!(adopted(None, Some(fp(Some(0))), START), None); + } + #[test] fn a_file_without_a_change_time_never_settles() { assert!(!changed_before(&fp(None), START)); diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index fe22dbc29..65d1618c5 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -661,8 +661,8 @@ impl DirSQL { let key = (table.clone(), rel.clone()); let known = state.known.get(&key); if !freshness::must_read(known, now.as_ref(), self.inner.built_ns) { - if known == Some(&freshness::Known::Built) { - updates.push((key, freshness::after_read(now, self.inner.built_ns))); + if let Some(adopted) = freshness::adopted(known, now, self.inner.built_ns) { + updates.push((key, adopted)); } continue; } diff --git a/testing-conventions.toml b/testing-conventions.toml index c4e81d3fa..d14ee5bd3 100644 --- a/testing-conventions.toml +++ b/testing-conventions.toml @@ -107,6 +107,12 @@ rules = ["mutation"] lines = [513, 520] reason = "set_terminated_text writes through a raw sqlite3_context, which only a real Connection supplies and unit-lint isolation forbids; tests/vtab.rs (stat_columns_carry_real_values) asserts the full path, basename and ext text" +[[rust.exempt]] +path = "src/freshness.rs" +rules = ["mutation"] +lines = [100, 115, 121, 127, 128, 130, 136, 143] +reason = "the cfg(windows) and non-unix fingerprint bodies are compiled out on the Linux mutation runner, so no Linux test can reach them; the Windows CI job runs tests/fresh_declared_*.rs, whose t+0 asserts fail if the Windows fingerprint misses a change" + # Unit-only coverage floor: measured by `cargo llvm-cov --lib --features cli` # (the `--lib` scoping keeps the integration tier under packages/rust/tests/ out # of the number -- testing-conventions #269). `branch` runs under `--branch`, From 3c5875c2e1f2cca87cc8395f7c0b5bd05e8db122 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 02:18:02 -0500 Subject: [PATCH 09/12] chore: point the fingerprint exemption at the lines the mutants report --- testing-conventions.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/testing-conventions.toml b/testing-conventions.toml index d14ee5bd3..4d1b28aad 100644 --- a/testing-conventions.toml +++ b/testing-conventions.toml @@ -110,7 +110,7 @@ reason = "set_terminated_text writes through a raw sqlite3_context, which only a [[rust.exempt]] path = "src/freshness.rs" rules = ["mutation"] -lines = [100, 115, 121, 127, 128, 130, 136, 143] +lines = [101, 115, 121, 127, 128, 130, 137, 144] reason = "the cfg(windows) and non-unix fingerprint bodies are compiled out on the Linux mutation runner, so no Linux test can reach them; the Windows CI job runs tests/fresh_declared_*.rs, whose t+0 asserts fail if the Windows fingerprint misses a change" # Unit-only coverage floor: measured by `cargo llvm-cov --lib --features cli` From 31c4e540dfd879b7360e31381e120c35a0ad35b3 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:15:50 -0500 Subject: [PATCH 10/12] test: a write through a still-open handle is read by the next query --- .../rust/tests/fresh_declared_open_handle.rs | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 packages/rust/tests/fresh_declared_open_handle.rs diff --git a/packages/rust/tests/fresh_declared_open_handle.rs b/packages/rust/tests/fresh_declared_open_handle.rs new file mode 100644 index 000000000..4962380b8 --- /dev/null +++ b/packages/rust/tests/fresh_declared_open_handle.rs @@ -0,0 +1,68 @@ +//! A write through a handle that is still open is read by the next query, +//! whether or not the writer flushed. + +#[path = "common/fresh.rs"] +mod fresh; + +use fresh::{Tree, lines_of}; +use std::fs::{File, OpenOptions}; +use std::io::{Seek, SeekFrom, Write}; + +const ROUNDS: usize = 500; + +fn open_writer(tree: &Tree, rel: &str) -> File { + OpenOptions::new().write(true).open(tree.path(rel)).unwrap() +} + +/// A same-size rewrite never truncates, so the length never moves. +fn overwrite(file: &mut File, content: &str, resize: bool, flush: bool) { + file.seek(SeekFrom::Start(0)).unwrap(); + file.write_all(content.as_bytes()).unwrap(); + if resize { + file.set_len(content.len() as u64).unwrap(); + } + if flush { + file.sync_all().unwrap(); + } +} + +fn hammer(resize: bool, flush: bool, content: impl Fn(usize) -> String) { + let tree = Tree::new(); + tree.write("o.txt", &content(usize::MAX)); + let db = tree.open(); + let mut file = open_writer(&tree, "o.txt"); + for i in 0..ROUNDS { + let expected = content(i); + overwrite(&mut file, &expected, resize, flush); + assert_eq!(lines_of(&db, "o.txt"), [expected], "round {i}"); + } + drop(file); +} + +fn same_size(i: usize) -> String { + format!("s{:08}", i % 100_000_000) +} + +fn size_change(i: usize) -> String { + "z".repeat(1 + i % 37) +} + +#[test] +fn a_same_size_rewrite_through_an_open_handle_is_read_by_the_next_query() { + hammer(false, false, same_size); +} + +#[test] +fn a_flushed_same_size_rewrite_through_an_open_handle_is_read_by_the_next_query() { + hammer(false, true, same_size); +} + +#[test] +fn a_resize_through_an_open_handle_is_read_by_the_next_query() { + hammer(true, false, size_change); +} + +#[test] +fn a_flushed_resize_through_an_open_handle_is_read_by_the_next_query() { + hammer(true, true, size_change); +} From 43357845207c834a388072499c7907879c0d51cd Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:24:35 -0500 Subject: [PATCH 11/12] test: probe open-handle writes against files whose last read had settled --- .../rust/tests/fresh_declared_open_handle.rs | 70 +++++++++++++------ 1 file changed, 50 insertions(+), 20 deletions(-) diff --git a/packages/rust/tests/fresh_declared_open_handle.rs b/packages/rust/tests/fresh_declared_open_handle.rs index 4962380b8..16d68e879 100644 --- a/packages/rust/tests/fresh_declared_open_handle.rs +++ b/packages/rust/tests/fresh_declared_open_handle.rs @@ -1,5 +1,6 @@ //! A write through a handle that is still open is read by the next query, -//! whether or not the writer flushed. +//! whether or not the writer flushed, even when the file had settled and its +//! last read was trusted. #[path = "common/fresh.rs"] mod fresh; @@ -7,12 +8,14 @@ mod fresh; use fresh::{Tree, lines_of}; use std::fs::{File, OpenOptions}; use std::io::{Seek, SeekFrom, Write}; +use std::thread::sleep; +use std::time::Duration; -const ROUNDS: usize = 500; +const ROUNDS: usize = 300; -fn open_writer(tree: &Tree, rel: &str) -> File { - OpenOptions::new().write(true).open(tree.path(rel)).unwrap() -} +/// Past the settle window, so every file's last read is trusted on its stat +/// alone when the write through the open handle lands. +const SETTLED: Duration = Duration::from_millis(2500); /// A same-size rewrite never truncates, so the length never moves. fn overwrite(file: &mut File, content: &str, resize: bool, flush: bool) { @@ -26,43 +29,70 @@ fn overwrite(file: &mut File, content: &str, resize: bool, flush: bool) { } } -fn hammer(resize: bool, flush: bool, content: impl Fn(usize) -> String) { +fn hammer( + resize: bool, + flush: bool, + before: impl Fn(usize) -> String, + after: impl Fn(usize) -> String, +) { let tree = Tree::new(); - tree.write("o.txt", &content(usize::MAX)); + let rel = |i: usize| format!("o/{i}.txt"); + for i in 0..ROUNDS { + tree.write(&rel(i), &before(i)); + } let db = tree.open(); - let mut file = open_writer(&tree, "o.txt"); + let mut files: Vec = (0..ROUNDS) + .map(|i| { + OpenOptions::new() + .write(true) + .open(tree.path(&rel(i))) + .unwrap() + }) + .collect(); + sleep(SETTLED); for i in 0..ROUNDS { - let expected = content(i); - overwrite(&mut file, &expected, resize, flush); - assert_eq!(lines_of(&db, "o.txt"), [expected], "round {i}"); + assert_eq!(lines_of(&db, &rel(i)), [before(i)], "file {i} before"); + } + for (i, file) in files.iter_mut().enumerate() { + let expected = after(i); + overwrite(file, &expected, resize, flush); + assert_eq!(lines_of(&db, &rel(i)), [expected], "file {i}"); } - drop(file); + drop(files); +} + +fn same_size_before(i: usize) -> String { + format!("a{i:08}") +} + +fn same_size_after(i: usize) -> String { + format!("b{i:08}") } -fn same_size(i: usize) -> String { - format!("s{:08}", i % 100_000_000) +fn short(i: usize) -> String { + format!("{i}") } -fn size_change(i: usize) -> String { - "z".repeat(1 + i % 37) +fn long(i: usize) -> String { + format!("{i}-{}", "z".repeat(1 + i % 37)) } #[test] fn a_same_size_rewrite_through_an_open_handle_is_read_by_the_next_query() { - hammer(false, false, same_size); + hammer(false, false, same_size_before, same_size_after); } #[test] fn a_flushed_same_size_rewrite_through_an_open_handle_is_read_by_the_next_query() { - hammer(false, true, same_size); + hammer(false, true, same_size_before, same_size_after); } #[test] fn a_resize_through_an_open_handle_is_read_by_the_next_query() { - hammer(true, false, size_change); + hammer(true, false, short, long); } #[test] fn a_flushed_resize_through_an_open_handle_is_read_by_the_next_query() { - hammer(true, true, size_change); + hammer(true, true, short, long); } From 97eff272fdcfc5beb6efe47c2cc50948bb3516bc Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Wed, 4 Nov 2026 10:32:17 -0500 Subject: [PATCH 12/12] docs: name the limits a stat-based freshness check shares --- docs/reference/sdk.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md index 6df058efb..6369854ea 100644 --- a/docs/reference/sdk.md +++ b/docs/reference/sdk.md @@ -234,6 +234,13 @@ Executes a SQL query and returns rows keyed by column name. breaks a foreign key, the query fails with the error a build over that tree returns. Rust and Python today; a TypeScript query reads what the watcher has applied. + + The check trusts the file's stat, so it shares the limits of every + stat-based tool (`make`, `rsync`, `git status`): a write through a shared + memory map can land without the file's times moving until the writer + unmaps or syncs it; a network filesystem that caches attributes can report + a stale stat; and a clock stepped backwards can stamp a new write with a + time older than the last read. - **Read-only.** Each statement is classified by SQLite itself (`sqlite3_stmt_readonly`); anything classified as a write — `INSERT`, `UPDATE`, `DELETE`, `DROP`, `CREATE`, `ALTER`, `VACUUM`, … — is rejected