diff --git a/PARITY.md b/PARITY.md index 8cfabb262..812afdf3c 100644 --- a/PARITY.md +++ b/PARITY.md @@ -56,6 +56,15 @@ Python, Rust and TypeScript. No SDK exposes a knob for it: an SDK-level `progress=` argument would be exactly the kind of per-SDK surface this table exists to prevent. +The same reporter covers **query time** (#1001): a `query` whose declared +`[[dirsql.function]]` workers make round trips reports a running count of them +(`dirsql: running 9204 worker calls`). It lives in `Db::query`, below every +SDK's `query`/`await db.query`, so again all three change at once and none +carries a copy. Notably this is the progress `dirsql-plugin-embeddings` gets +for its `embed(content)`-per-row search: the plugin drives the Python SDK, +which drives the same `Db::query`, so no plugin-side reporting exists or is +needed. + | API | Python | Rust | TypeScript | |----------------------------|------------------------------------------------|------------------------------------------------------|---------------------------------------------------------| | Constructor | `DirSQL(root=None, *, tables=None, ignore=None, no_ignore=False, config=None, persist=False, persist_path=None, extensions=None)` | `DirSQL::builder().root(..).tables(..).ignore(..).config(..).persist(Option).extensions(..).build()` (also `DirSQL::new`/`with_ignore` shortcuts) | `new DirSQL(configPath)` or `new DirSQL({ root?, tables?, ignore?, config?, persist?, persistPath?, extensions?, noIgnore? })` + `await db.ready` | diff --git a/docs/reference/cli.md b/docs/reference/cli.md index f8f521242..0377c0f30 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -483,6 +483,24 @@ dirsql: indexed 41231 files in 3m12s stdout is untouched — it carries the query result and nothing else. +A query gets the same treatment when it calls a +[worker-backed function](./config.md#dirsql-function). Those cost one round +trip per row, so a query that calls one over a whole corpus is the slowest +thing dirsql does: + +``` +dirsql: running 9204 worker calls +``` + +and, when the query ends: + +``` +dirsql: ran 41231 worker calls in 2m41s +``` + +A query that calls no worker prints nothing, whatever the setting — the +reporting only ever speaks when there is something to report. + By default this is **terminal-only, and only for work slow enough to wonder about**: a phase that finishes in under half a second prints nothing at all, and a run whose stderr is a pipe or a file prints nothing regardless of how diff --git a/docs/reference/config.md b/docs/reference/config.md index 0f1f58c1c..7bfad0810 100644 --- a/docs/reference/config.md +++ b/docs/reference/config.md @@ -133,6 +133,11 @@ process alive for the rest of the invocation, sending it every subsequent call — one process total, never one per row or per file. The worker is torn down when the invocation ends. +A query paying per-row round trips reports them on stderr while it runs +(`dirsql: running 9204 worker calls`), on the same terms as every other +[progress line](./cli.md#progress-reporting): terminal-only by default, silent +on a pipe, and gone before the result is printed. + The `timeout` bounds each **round-trip call**, not the query: a query that calls the function on 10 000 rows is 10 000 individually timed calls. A call that times out, or a worker that crashes or closes its pipes, fails the query diff --git a/docs/reference/sdk.md b/docs/reference/sdk.md index 05799bdff..b940775a2 100644 --- a/docs/reference/sdk.md +++ b/docs/reference/sdk.md @@ -408,6 +408,10 @@ one summary of what each cost: dirsql: indexed 41231 files in 3m12s ``` +A `query` that calls a worker-backed function is reported the same way, as a +running count of round trips (`dirsql: running 9204 worker calls`) — the shape +`dirsql-plugin-embeddings` produces when it embeds a corpus one row at a time. + This is terminal-only by default, and only for a phase that runs longer than half a second — a program whose stderr is a pipe, a file, or a log collector sees nothing, whatever the scan costs. Set `DIRSQL_PROGRESS=never` to diff --git a/packages/python/e2e-attestations/claude-tackle-957-lrm0z6-worker-calls.json b/packages/python/e2e-attestations/claude-tackle-957-lrm0z6-worker-calls.json new file mode 100644 index 000000000..820184a30 --- /dev/null +++ b/packages/python/e2e-attestations/claude-tackle-957-lrm0z6-worker-calls.json @@ -0,0 +1,7 @@ +{ + "command": "uv run python -m pytest tests/e2e/ -x -q", + "ran_at": 1787263769, + "exit_code": 0, + "commit": "bced5b72c85b9c50073fd6326c1528326a1bd9d3", + "branch": "claude/tackle-957-lrm0z6-worker-calls" +} diff --git a/packages/rust/changelog.d/2026-08-20-worker-call-progress.md b/packages/rust/changelog.d/2026-08-20-worker-call-progress.md new file mode 100644 index 000000000..d8e1d73e5 --- /dev/null +++ b/packages/rust/changelog.d/2026-08-20-worker-call-progress.md @@ -0,0 +1,3 @@ +**Added** + +- **A query reports the worker round trips it is paying for.** A `[[dirsql.function]]` costs one worker round trip per row, so a query calling one over a whole corpus — `dirsql-plugin-embeddings` runs `embed(content)` that way — can run for minutes inside a single `query()` with nothing on the terminal. Core now counts the round trips and draws them with the same reporter the startup scan uses (`dirsql: running 9204 worker calls`, erased at the end of the query and replaced by `dirsql: ran 41231 worker calls in 2m41s`), under the same `DIRSQL_PROGRESS` gate: terminal-only and slow-work-only by default, silent on a pipe. A query that calls no worker prints nothing at any setting. The count is drawn **and erased by dirsql itself** rather than by the worker, which is what keeps the result off the progress line: a worker's own bar cannot be cleaned up, since the worker is SIGKILLed and is torn down only after the result has been printed. (#1001, #957) diff --git a/packages/rust/migrations.d/2026-08-20-worker-call-progress.md b/packages/rust/migrations.d/2026-08-20-worker-call-progress.md new file mode 100644 index 000000000..77b24f7f6 --- /dev/null +++ b/packages/rust/migrations.d/2026-08-20-worker-call-progress.md @@ -0,0 +1,51 @@ +### A query reports its worker round trips on stderr (#1001) + +#### Summary + +`Db::query` now brackets each statement with a progress phase that counts the +round trips its declared `[[dirsql.function]]` workers make, and reports them +on stderr. No API changes; what changes is that a query that used to write +nothing to stderr may now write to it. Gated exactly as the startup scan's +reporting is: terminal-only, and only past a half-second warmup, unless +`DIRSQL_PROGRESS` says otherwise. + +#### Required changes + +_None._ `DIRSQL_PROGRESS=never` is the opt-out for a program that insists on an +empty stderr, and it already covers the scan's reporting too. + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +- **stderr may carry a worker-call counter on a terminal.** A query whose + declared functions make round trips for longer than 500 ms draws + `dirsql: running worker calls`, rewritten in place, then erases it and + prints `dirsql: ran worker calls in `. A query that calls no worker + draws nothing, at any setting. +- **Non-terminal stderr is unchanged**, at any duration. Anything asserting on + captured stderr keeps passing. +- **stdout is unchanged.** The counter is erased before `query()` returns, so + the result is never printed onto a leftover progress line. +- **`functions::register_all` takes a third argument**, the shared + `CallReporter`. It is `pub(crate)`, so no crate consumer can be calling it. + +#### Verification + +```bash +# A worker-backed function over a corpus, forced on so no terminal is needed: +DIRSQL_PROGRESS=always dirsql query "SELECT fake(basename) AS v FROM './*.txt'" -c ./.dirsql.toml +# dirsql: running 13 worker calls (rewritten in place, then erased) +# dirsql: ran 14 worker calls in 0.7s +# [{"v":"f1.txt"}, …] + +# Piped: stderr is empty, and stdout is parseable JSON with nothing glued to it. +dirsql query "SELECT fake(basename) AS v FROM './*.txt'" -c ./.dirsql.toml 2>err.txt | jq . +wc -c err.txt # -> 0 + +# A query that calls no worker reports nothing even when forced on: +DIRSQL_PROGRESS=always dirsql query "SELECT 1" -c ./.dirsql.toml +# -> [{"1":1}] (nothing on stderr) +``` diff --git a/packages/rust/src/db.rs b/packages/rust/src/db.rs index addb5cb60..16d3bea1f 100644 --- a/packages/rust/src/db.rs +++ b/packages/rust/src/db.rs @@ -1,8 +1,10 @@ use rusqlite::Connection; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; +use std::sync::Arc; use thiserror::Error; +use crate::functions::CallReporter; use crate::parsed_vtab; use crate::path_table::{self, PathTable, Resolution}; use crate::scanner; @@ -264,6 +266,10 @@ pub struct Db { /// The persistent cache a parsed path-table reuses across runs. `None` for /// an ephemeral index, which has nowhere to cache to. path_table_cache: Option, + /// Counts the worker round trips a query pays for. Shared with every + /// declared function registered on `conn`; inert until a function is both + /// declared and called. + calls: Arc, } /// The skip rules a fresh `Db` starts with. @@ -299,6 +305,7 @@ impl Db { path_table_gitignore: true, path_table_parser: None, path_table_cache: None, + calls: CallReporter::new(), }) } @@ -321,6 +328,7 @@ impl Db { path_table_gitignore: true, path_table_parser: None, path_table_cache: None, + calls: CallReporter::new(), }) } @@ -365,6 +373,12 @@ impl Db { self.path_table_cache = Some(path); } + /// The reporter every declared function increments. Handed to + /// `functions::register_all` at build time. + pub(crate) fn call_reporter(&self) -> &Arc { + &self.calls + } + /// Borrow the underlying SQLite connection. Internal use only — exposed /// to the `persist` module so it can manage the sidecar tables. #[doc(hidden)] @@ -687,6 +701,12 @@ impl Db { /// internal writes (`insert_row`, delete-by-file, persist), which never go /// through `query()`, are unaffected. pub fn query(&self, sql: &str) -> Result>> { + // Bracket the whole statement, prepare included: a worker-backed + // function called on every row makes this the slowest thing dirsql + // does, and the guard is what erases the counter before the caller + // prints the rows. + let _calls = self.calls.phase(); + // Each iteration must register a table no earlier iteration did; a // repeat means the fallback is not making progress, so the SQLite // error stands. That is what bounds the loop. diff --git a/packages/rust/src/functions.rs b/packages/rust/src/functions.rs index 8f3bb49f5..eb5e427f5 100644 --- a/packages/rust/src/functions.rs +++ b/packages/rust/src/functions.rs @@ -37,6 +37,7 @@ use rusqlite::Connection; use rusqlite::functions::FunctionFlags; use crate::db::Value; +use crate::progress::{CallProgress, Progress}; /// The per-call timeout when a `[[dirsql.function]]` entry declares no /// `timeout` of its own. A round-trip on a persistent worker cannot be @@ -60,14 +61,20 @@ pub struct ResolvedFunction { /// Register every resolved function on `conn`, once per accepted arity. /// Purely registration — no worker is spawned here. +/// +/// Every function shares one `calls` reporter: what a user waiting on a query +/// wants to know is how many round trips it is paying for, not which function +/// made them. pub(crate) fn register_all( conn: &Connection, functions: &[ResolvedFunction], + calls: &Arc, ) -> rusqlite::Result<()> { for function in functions { let worker = Arc::new(Worker::for_process(function)); for &arity in &function.args { let worker = Arc::clone(&worker); + let calls = Arc::clone(calls); conn.create_scalar_function( &function.name, i32::from(arity), @@ -77,6 +84,7 @@ pub(crate) fn register_all( for i in 0..ctx.len() { args.push(Value::from(rusqlite::types::Value::from(ctx.get_raw(i)))); } + calls.record(); worker .call(&args) .map_err(|message| rusqlite::Error::UserFunctionError(message.into())) @@ -87,6 +95,78 @@ pub(crate) fn register_all( Ok(()) } +/// Counts the worker round trips a query pays for, and reports them. +/// +/// A query that calls a declared function on every row spends one round trip +/// per row -- `dirsql-plugin-embeddings` runs `embed(content)` over the whole +/// corpus that way -- so the wall time of a single `query()` can be minutes +/// with nothing on the terminal (dirsql#957). +/// +/// The count has to be drawn *and erased* by the process that owns stdout, or +/// the query result lands on top of a leftover line. A worker cannot do that: +/// it is SIGKILLed on teardown, and teardown happens after the result is +/// printed anyway (dirsql#1001). +/// +/// Behind a mutex because SQLite invokes a scalar function on whatever thread +/// is running the query. +pub(crate) struct CallReporter { + state: Mutex, +} + +struct CallState { + count: u64, + progress: Box, +} + +impl CallReporter { + pub(crate) fn new() -> Arc { + Arc::new(Self::with_progress(Box::new(Progress::worker_calls()))) + } + + fn with_progress(progress: Box) -> Self { + Self { + state: Mutex::new(CallState { count: 0, progress }), + } + } + + /// Open a phase and return the guard that closes it. A guard rather than + /// a paired call so an error mid-query still ends the phase, and so the + /// live line is gone before the caller prints anything. + pub(crate) fn phase(&self) -> CallPhase<'_> { + if let Ok(mut state) = self.state.lock() { + state.count = 0; + state.progress.restart(); + } + CallPhase(self) + } + + /// Record one round trip. A poisoned mutex costs the counter, never the + /// call: reporting is not worth failing a query over. + fn record(&self) { + if let Ok(mut state) = self.state.lock() { + state.count += 1; + let count = state.count; + state.progress.update(count); + } + } + + fn end(&self) { + if let Ok(mut state) = self.state.lock() { + let count = state.count; + state.progress.finish(count); + } + } +} + +/// The open phase. Ends when it drops. +pub(crate) struct CallPhase<'a>(&'a CallReporter); + +impl Drop for CallPhase<'_> { + fn drop(&mut self) { + self.0.end(); + } +} + /// The registration flags for a declared function: UTF-8 always, plus /// `SQLITE_DETERMINISTIC` when the entry opts in. fn function_flags(deterministic: bool) -> FunctionFlags { @@ -438,6 +518,117 @@ mod tests { Value::Text(s.to_string()) } + // --- call reporting ---------------------------------------------------- + + /// A [`CallProgress`] double recording what it was asked to do, so the + /// orchestration is assertable in-module. What the reporter *draws* is + /// `progress.rs`'s business and is unit-tested there. + #[derive(Clone, Default)] + struct Recorder(Arc>>); + + impl Recorder { + fn events(&self) -> Vec { + self.0.lock().unwrap().clone() + } + + fn push(&self, event: String) { + self.0.lock().unwrap().push(event); + } + } + + struct RecordingProgress(Recorder); + + impl CallProgress for RecordingProgress { + fn update(&mut self, done: u64) { + self.0.push(format!("update {done}")); + } + + fn finish(&mut self, done: u64) { + self.0.push(format!("finish {done}")); + } + + fn restart(&mut self) { + self.0.push("restart".to_string()); + } + } + + fn reporter() -> (CallReporter, Recorder) { + let recorder = Recorder::default(); + let reporter = CallReporter::with_progress(Box::new(RecordingProgress(recorder.clone()))); + (reporter, recorder) + } + + #[test] + fn a_recorded_round_trip_is_reported_as_a_running_count() { + let (reporter, recorder) = reporter(); + + reporter.record(); + reporter.record(); + + assert_eq!(recorder.events(), ["update 1", "update 2"]); + } + + /// The phase is what the count belongs to: a second query must not carry + /// the first one's total, and the reporter has to be told to start over. + #[test] + fn opening_a_phase_restarts_the_reporter_and_the_count() { + let (reporter, recorder) = reporter(); + reporter.record(); + reporter.record(); + + { + let _phase = reporter.phase(); + reporter.record(); + } + + assert_eq!( + recorder.events(), + ["update 1", "update 2", "restart", "update 1", "finish 1"], + "the new phase counts from one, not from three" + ); + } + + /// Dropping the guard is what ends the phase — the whole reason it is a + /// guard, since a query can leave by an error path. + #[test] + fn the_phase_guard_finishes_when_it_drops() { + let (reporter, recorder) = reporter(); + + { + let _phase = reporter.phase(); + reporter.record(); + reporter.record(); + assert_eq!( + recorder.events(), + ["restart", "update 1", "update 2"], + "nothing is finished while the phase is open" + ); + } + + assert_eq!( + recorder.events(), + ["restart", "update 1", "update 2", "finish 2"], + "the closed phase reports the count it accumulated" + ); + } + + /// The production constructor wires a real [`Progress`] through the same + /// trait, so a whole phase runs against it here — the double proves the + /// orchestration, this proves the wiring is connected to something real. + /// It draws nothing: the reporter is disabled unless stderr is a terminal. + #[test] + fn the_production_reporter_counts_round_trips() { + let reporter = CallReporter::new(); + + { + let _phase = reporter.phase(); + reporter.record(); + reporter.record(); + } + + assert_eq!(reporter.state.lock().unwrap().count, 2); + } + // --- wire encoding ----------------------------------------------------- #[test] diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 2e2079943..6df900248 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -885,7 +885,8 @@ impl DirSQL { // Register declared functions after extensions so both are available // to queries. Registration is inert: no worker process exists until a // query's first call to the function. - functions::register_all(db.conn(), &functions).map_err(DirSqlError::sqlite)?; + functions::register_all(db.conn(), &functions, db.call_reporter()) + .map_err(DirSqlError::sqlite)?; // Sweep a cache the reconcile rejected. It happens *here*, after the // extensions and functions are on the connection, rather than where diff --git a/packages/rust/src/progress.rs b/packages/rust/src/progress.rs index 909f76092..079b68642 100644 --- a/packages/rust/src/progress.rs +++ b/packages/rust/src/progress.rs @@ -71,7 +71,10 @@ impl Mode { /// The clock [`Progress`] throttles against. A seam so the unit tier can drive /// the warmup and redraw thresholds from both sides instead of sleeping. -pub trait Clock { +/// +/// `Send` because a reporter is shared with the worker-call counter, which +/// SQLite invokes from whatever thread is running the query. +pub trait Clock: Send { fn now(&self) -> Instant; } @@ -96,8 +99,10 @@ pub struct Progress { label: &'static str, /// Past participle for the summary ("indexed 9 files in 4.2s"). summary_label: &'static str, - out: Box, - clock: Box, + /// What is being counted: "files", "worker calls". + noun: &'static str, + out: Box, + clock: Box, mode: Mode, /// Whether the sink is a terminal. Only consulted under [`Mode::Auto`]. terminal: bool, @@ -114,19 +119,27 @@ pub struct Progress { impl Progress { /// Reporter for the directory walk, which counts files as it finds them. pub fn scanning() -> Self { - Self::to_stderr("scanning", "scanned") + Self::to_stderr("scanning", "scanned", "files") } /// Reporter for the ingest pass, which counts files against a known total. pub fn indexing() -> Self { - Self::to_stderr("indexing", "indexed") + Self::to_stderr("indexing", "indexed", "files") + } + + /// Reporter for a query's worker round trips. No total: the query decides + /// how many rows it calls the function on, and SQLite does not say up + /// front. + pub fn worker_calls() -> Self { + Self::to_stderr("running", "ran", "worker calls") } - fn to_stderr(label: &'static str, summary_label: &'static str) -> Self { + fn to_stderr(label: &'static str, summary_label: &'static str, noun: &'static str) -> Self { let terminal = std::io::stderr().is_terminal(); Self::new( label, summary_label, + noun, Box::new(std::io::stderr()), Box::new(SystemClock), Mode::parse(std::env::var(PROGRESS_ENV).ok().as_deref()), @@ -139,8 +152,9 @@ impl Progress { pub fn new( label: &'static str, summary_label: &'static str, - out: Box, - clock: Box, + noun: &'static str, + out: Box, + clock: Box, mode: Mode, terminal: bool, ) -> Self { @@ -148,6 +162,7 @@ impl Progress { Self { label, summary_label, + noun, out, clock, mode, @@ -182,10 +197,21 @@ impl Progress { _ => {} } self.last_draw = Some(now); - let line = render(self.label, done, total); + let line = render(self.label, self.noun, done, total); self.draw(&line); } + /// Reuse this reporter for a fresh phase: erase whatever is on screen and + /// reset the clock and the throttle, keeping the sink, the mode and the + /// wording. One reporter therefore serves every query on a connection -- + /// and, unlike constructing a new one per phase, it keeps whatever sink it + /// was given instead of silently reverting to stderr. + pub fn restart(&mut self) { + self.erase(); + self.started = self.clock.now(); + self.last_draw = None; + } + /// End the phase: erase the live line and leave one summary line behind. /// Silent when nothing was ever drawn, so a fast phase leaves no trace. pub fn finish(&mut self, done: u64) { @@ -196,8 +222,9 @@ impl Progress { let elapsed = self.clock.now().duration_since(self.started); let _ = writeln!( self.out, - "dirsql: {} {done} files in {}", + "dirsql: {} {done} {} in {}", self.summary_label, + self.noun, format_duration(elapsed) ); let _ = self.out.flush(); @@ -231,15 +258,40 @@ impl Drop for Progress { } } +/// The narrow view of a reporter that the worker-call counter needs: a running +/// count with no total, and a phase it can restart. A trait so the counter's +/// unit tests can inject a double without reaching across modules. +pub(crate) trait CallProgress: Send { + /// Report `done` round trips so far. There is no total — SQLite does not + /// say up front how many rows the query will call the function on. + fn update(&mut self, done: u64); + fn finish(&mut self, done: u64); + fn restart(&mut self); +} + +impl CallProgress for Progress { + fn update(&mut self, done: u64) { + Progress::update(self, done, None); + } + + fn finish(&mut self, done: u64) { + Progress::finish(self, done); + } + + fn restart(&mut self) { + Progress::restart(self); + } +} + /// The live line's text. With a total it carries a percentage; the walk has no /// total to divide by until it is over, so it reports a running count. -fn render(label: &str, done: u64, total: Option) -> String { +fn render(label: &str, noun: &str, done: u64, total: Option) -> String { match total { Some(total) => format!( - "dirsql: {label} {done}/{total} files ({}%)", + "dirsql: {label} {done}/{total} {noun} ({}%)", percent(done, total) ), - None => format!("dirsql: {label} {done} files"), + None => format!("dirsql: {label} {done} {noun}"), } } @@ -266,23 +318,23 @@ fn format_duration(elapsed: Duration) -> String { #[cfg(test)] mod tests { use super::*; - use std::cell::{Cell, RefCell}; - use std::rc::Rc; + use std::sync::{Arc, Mutex}; /// A `Write` the test can read back. Shares one buffer with the `Progress` - /// that owns its clone. + /// that owns its clone. `Arc`/`Mutex` rather than `Rc`/`RefCell` because + /// the sink has to satisfy the reporter's `Send` bound. #[derive(Clone, Default)] - struct Sink(Rc>>); + struct Sink(Arc>>); impl Sink { fn text(&self) -> String { - String::from_utf8(self.0.borrow().clone()).unwrap() + String::from_utf8(self.0.lock().unwrap().clone()).unwrap() } } impl Write for Sink { fn write(&mut self, buf: &[u8]) -> std::io::Result { - self.0.borrow_mut().extend_from_slice(buf); + self.0.lock().unwrap().extend_from_slice(buf); Ok(buf.len()) } @@ -296,25 +348,26 @@ mod tests { #[derive(Clone)] struct FakeClock { base: Instant, - offset: Rc>, + offset: Arc>, } impl FakeClock { fn new() -> Self { Self { base: Instant::now(), - offset: Rc::new(Cell::new(Duration::ZERO)), + offset: Arc::new(Mutex::new(Duration::ZERO)), } } fn advance(&self, delta: Duration) { - self.offset.set(self.offset.get() + delta); + let mut offset = self.offset.lock().unwrap(); + *offset += delta; } } impl Clock for FakeClock { fn now(&self) -> Instant { - self.base + self.offset.get() + self.base + *self.offset.lock().unwrap() } } @@ -324,6 +377,7 @@ mod tests { let progress = Progress::new( "indexing", "indexed", + "files", Box::new(sink.clone()), Box::new(clock.clone()), mode, @@ -501,17 +555,30 @@ mod tests { #[test] fn a_phase_with_no_known_total_reports_a_running_count() { - assert_eq!(render("scanning", 42, None), "dirsql: scanning 42 files"); + assert_eq!( + render("scanning", "files", 42, None), + "dirsql: scanning 42 files" + ); } #[test] fn a_phase_with_a_known_total_reports_a_percentage() { assert_eq!( - render("indexing", 3, Some(8)), + render("indexing", "files", 3, Some(8)), "dirsql: indexing 3/8 files (37%)" ); } + /// The noun travels with the phase: worker round trips are not files, and + /// a line that called them files would be lying about what it counted. + #[test] + fn the_counted_thing_is_named_by_the_phase() { + assert_eq!( + render("running", "worker calls", 9204, None), + "dirsql: running 9204 worker calls" + ); + } + #[test] fn a_percentage_floors_rather_than_rounds() { assert_eq!(percent(3, 8), 37); @@ -547,6 +614,60 @@ mod tests { assert_eq!(format_duration(Duration::from_secs(3725)), "62m05s"); } + /// A reporter serves more than one phase, so restarting must forget the + /// previous phase's clock and throttle -- otherwise the second phase + /// inherits the first one's elapsed time and draws immediately. + #[test] + fn restarting_clears_the_line_and_the_clock() { + let (mut progress, sink, clock) = reporter(Mode::Auto, true); + + clock.advance(WARMUP); + progress.update(1, Some(10)); + let drawn = sink.text(); + assert!(!drawn.is_empty(), "the first phase drew"); + + progress.restart(); + progress.update(1, Some(10)); + + let line = "dirsql: indexing 1/10 files (10%)"; + let blanks = " ".repeat(line.len()); + assert_eq!( + sink.text(), + format!("\r{line}\r{blanks}\r"), + "restart erased the line, and the fresh phase is back under its warmup" + ); + + clock.advance(WARMUP); + progress.update(2, Some(10)); + assert!( + sink.text().ends_with("\rdirsql: indexing 2/10 files (20%)"), + "and it draws again once the new phase is old enough: {:?}", + sink.text() + ); + } + + /// The `CallProgress` impl is a delegation, and a delegation that quietly + /// does nothing looks identical to a working one -- until a second query + /// inherits the first one's line. + #[test] + fn restarting_through_call_progress_erases_the_live_line() { + let (mut progress, sink, clock) = reporter(Mode::Auto, true); + + clock.advance(WARMUP); + CallProgress::update(&mut progress, 1); + let line = "dirsql: indexing 1 files"; + assert_eq!(sink.text(), format!("\r{line}"), "the phase drew"); + + CallProgress::restart(&mut progress); + + let blanks = " ".repeat(line.len()); + assert_eq!( + sink.text(), + format!("\r{line}\r{blanks}\r"), + "restarting through the trait erased what it drew" + ); + } + /// An error mid-phase must not leave a half-drawn counter behind for the /// error message to land on top of. #[test] @@ -582,9 +703,16 @@ mod tests { let scanning = Progress::scanning(); assert_eq!(scanning.label, "scanning"); assert_eq!(scanning.summary_label, "scanned"); + assert_eq!(scanning.noun, "files"); let indexing = Progress::indexing(); assert_eq!(indexing.label, "indexing"); assert_eq!(indexing.summary_label, "indexed"); + assert_eq!(indexing.noun, "files"); + + let calls = Progress::worker_calls(); + assert_eq!(calls.label, "running"); + assert_eq!(calls.summary_label, "ran"); + assert_eq!(calls.noun, "worker calls"); } } diff --git a/packages/rust/tests/worker_call_progress_e2e.rs b/packages/rust/tests/worker_call_progress_e2e.rs new file mode 100644 index 000000000..121b9768b --- /dev/null +++ b/packages/rust/tests/worker_call_progress_e2e.rs @@ -0,0 +1,166 @@ +//! End-to-end tests for progress on worker-backed function calls. +//! +//! A query that calls a `[[dirsql.function]]` pays one worker round trip per +//! row (`docs/reference/config.md`, "Worker lifecycle"). On a corpus-sized +//! query — `dirsql-plugin-embeddings` runs `embed(content)` over every matched +//! file — that is tens of thousands of round trips inside a single `query()` +//! call, and dirsql#957 is that they happened in silence. +//! +//! These spawn the real `dirsql` binary against a real python3 worker with +//! nothing mocked, and pin the property that made worker-side reporting +//! unworkable (dirsql#1001): the counter is drawn and **erased by the process +//! that owns stdout**, so the query result is never glued onto a leftover +//! progress line. A worker drawing its own bar cannot do that — it is +//! SIGKILLed, and it is killed after the result is printed anyway. +//! +//! Gated behind `--features cli` (the `dirsql` bin needs it) and Unix (the +//! worker shells out to `python3`); the Rust CI test job runs on Linux. + +#![cfg(all(feature = "cli", unix))] + +use std::fs; +use std::process::Output; + +use assert_cmd::prelude::*; +use serde_json::Value; +use tempfile::TempDir; + +/// A worker that answers every call by echoing its argument back, one round +/// trip per call — the shape every worker-backed function has. +const WORKER: &str = r#" +import json +import sys + +for line in sys.stdin: + req = json.loads(line) + sys.stdout.write(json.dumps({"ok": req["call"][0]}, separators=(",", ":")) + "\n") + sys.stdout.flush() +"#; + +const CONFIG: &str = r#" +[[dirsql.function]] +name = "echo" +args = [1] +command = "python3 worker.py" +"#; + +/// A root with `count` files and a config declaring the worker function, so a +/// query over the path-table makes exactly `count` round trips. +fn fixture(count: usize) -> TempDir { + let root = TempDir::new().unwrap(); + fs::write(root.path().join("worker.py"), WORKER).unwrap(); + fs::write(root.path().join(".dirsql.toml"), CONFIG).unwrap(); + for i in 0..count { + fs::write(root.path().join(format!("file{i}.txt")), "x\n").unwrap(); + } + root +} + +fn run(dir: &std::path::Path, sql: &str, progress: Option<&str>) -> Output { + let mut cmd = std::process::Command::cargo_bin("dirsql").expect("binary must exist"); + cmd.arg("query") + .arg(sql) + .arg("-c") + .arg(".dirsql.toml") + .current_dir(dir) + .env_remove("DIRSQL_PROGRESS"); + if let Some(value) = progress { + cmd.env("DIRSQL_PROGRESS", value); + } + cmd.output().expect("spawning `dirsql query` failed") +} + +fn stderr(out: &Output) -> String { + String::from_utf8_lossy(&out.stderr).into_owned() +} + +/// The headline: a query paying per-row worker round trips says so, rather +/// than sitting silent for however long the worker takes. +#[test] +fn forced_progress_counts_the_worker_round_trips() { + let root = fixture(4); + + let out = run( + root.path(), + "SELECT echo(basename) AS v FROM './*.txt'", + Some("always"), + ); + + let err = stderr(&out); + assert!( + err.contains("worker calls"), + "the round trips are counted on stderr: {err:?}" + ); +} + +/// What survives the query is one line naming what it cost. +#[test] +fn forced_progress_summarizes_the_round_trips() { + let root = fixture(4); + + let out = run( + root.path(), + "SELECT echo(basename) AS v FROM './*.txt'", + Some("always"), + ); + + let err = stderr(&out); + assert!( + err.contains("dirsql: ran 4 worker calls in "), + "the summary carries the count and the elapsed time: {err:?}" + ); +} + +/// The regression this whole design exists for (dirsql#1001): the live line is +/// erased before the result is printed, so stdout is parseable JSON and not +/// `dirsql: running 4 worker calls[{"v":…}]`. +#[test] +fn the_result_is_never_glued_onto_a_leftover_progress_line() { + let root = fixture(4); + + let out = run( + root.path(), + "SELECT echo(basename) AS v FROM './*.txt'", + Some("always"), + ); + + let stdout = String::from_utf8_lossy(&out.stdout); + let rows: Vec = serde_json::from_str(stdout.trim()) + .unwrap_or_else(|e| panic!("stdout is JSON rows: {e}: {stdout:?}")); + assert_eq!(rows.len(), 4, "every file is still returned: {rows:?}"); + assert!( + !stdout.contains("worker calls"), + "progress never reaches stdout: {stdout:?}" + ); +} + +/// A query that calls no worker has nothing to report, even forced on: the +/// reporter speaks only when there is something to say. +#[test] +fn a_query_with_no_worker_calls_reports_nothing() { + let root = fixture(4); + + let out = run(root.path(), "SELECT 1 AS v", Some("always")); + + let err = stderr(&out); + assert!( + !err.contains("worker calls"), + "no round trips means no counter: {err:?}" + ); +} + +/// The default is `auto`, so a piped run stays silent however many round trips +/// it pays for. +#[test] +fn worker_call_progress_is_silent_when_stderr_is_not_a_terminal() { + let root = fixture(4); + + let out = run( + root.path(), + "SELECT echo(basename) AS v FROM './*.txt'", + None, + ); + + let err = stderr(&out); + assert_eq!(err, "", "a piped run writes nothing to stderr: {err:?}"); +} diff --git a/packages/ts/e2e-attestations/claude-tackle-957-lrm0z6-worker-calls.json b/packages/ts/e2e-attestations/claude-tackle-957-lrm0z6-worker-calls.json new file mode 100644 index 000000000..aae4f7900 --- /dev/null +++ b/packages/ts/e2e-attestations/claude-tackle-957-lrm0z6-worker-calls.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1787263810, + "exit_code": 0, + "commit": "0f63d6b6d7a536149a9a727873bbffd46525c68d", + "branch": "claude/tackle-957-lrm0z6-worker-calls" +}