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..6369854ea 100644 --- a/docs/reference/sdk.md +++ b/docs/reference/sdk.md @@ -224,6 +224,23 @@ 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. 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. + + 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 @@ -303,7 +320,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/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" +} 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..68fd4a494 --- /dev/null +++ b/packages/rust/migrations.d/2026-11-03-fresh-declared-reads.md @@ -0,0 +1,23 @@ +### 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. + +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. 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..778079ede --- /dev/null +++ b/packages/rust/src/freshness.rs @@ -0,0 +1,268 @@ +//! 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)), + } +} + +/// 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; + 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_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)); + } + + #[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..65d1618c5 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 let Some(adopted) = freshness::adopted(known, now, self.inner.built_ns) { + updates.push((key, adopted)); + } + 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/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/foreign_keys.rs b/packages/rust/tests/foreign_keys.rs index 0f97b8734..f2f6cc7f5 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/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_open_handle.rs b/packages/rust/tests/fresh_declared_open_handle.rs new file mode 100644 index 000000000..16d68e879 --- /dev/null +++ b/packages/rust/tests/fresh_declared_open_handle.rs @@ -0,0 +1,98 @@ +//! A write through a handle that is still open is read by the next query, +//! whether or not the writer flushed, even when the file had settled and its +//! last read was trusted. + +#[path = "common/fresh.rs"] +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 = 300; + +/// 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) { + 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, + before: impl Fn(usize) -> String, + after: impl Fn(usize) -> String, +) { + let tree = Tree::new(); + 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 files: Vec = (0..ROUNDS) + .map(|i| { + OpenOptions::new() + .write(true) + .open(tree.path(&rel(i))) + .unwrap() + }) + .collect(); + sleep(SETTLED); + for i in 0..ROUNDS { + 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(files); +} + +fn same_size_before(i: usize) -> String { + format!("a{i:08}") +} + +fn same_size_after(i: usize) -> String { + format!("b{i:08}") +} + +fn short(i: usize) -> String { + format!("{i}") +} + +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_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_before, same_size_after); +} + +#[test] +fn a_resize_through_an_open_handle_is_read_by_the_next_query() { + hammer(true, false, short, long); +} + +#[test] +fn a_flushed_resize_through_an_open_handle_is_read_by_the_next_query() { + hammer(true, true, short, long); +} 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/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/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" +} 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)), }) diff --git a/testing-conventions.toml b/testing-conventions.toml index c4e81d3fa..4d1b28aad 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 = [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` # (the `--lib` scoping keeps the integration tier under packages/rust/tests/ out # of the number -- testing-conventions #269). `branch` runs under `--branch`,