diff --git a/packages/python/e2e-attestations/bot-1562-hook-savepoint-1102.json b/packages/python/e2e-attestations/bot-1562-hook-savepoint-1102.json new file mode 100644 index 00000000..dc835a96 --- /dev/null +++ b/packages/python/e2e-attestations/bot-1562-hook-savepoint-1102.json @@ -0,0 +1,7 @@ +{ + "command": "uv run python -m pytest tests/e2e/on_file_query_test.py tests/e2e/query_contract_test.py tests/e2e/declared_table_name_test.py tests/e2e/fanout_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/watch_speed_test.py -q", + "ran_at": 1793671776, + "exit_code": 0, + "commit": "92277f18890c83a5a800db7c2dec4d6f0e4bb8fb", + "branch": "bot/1562-hook-savepoint-1102" +} diff --git a/packages/rust/changelog.d/2026-11-02-stage-per-table-rows-at-eof.md b/packages/rust/changelog.d/2026-11-02-stage-per-table-rows-at-eof.md new file mode 100644 index 00000000..ce516878 --- /dev/null +++ b/packages/rust/changelog.d/2026-11-02-stage-per-table-rows-at-eof.md @@ -0,0 +1 @@ +**Changed** A watched change to a table filled by a per-table `on-file` command reaches `/events` sooner: the command's rows are written as soon as its stdout closes, while it exits, and kept only once it exits zero. A command that exits non-zero still leaves the table as it was. diff --git a/packages/rust/migrations.d/2026-11-02-stage-per-table-rows-at-eof.md b/packages/rust/migrations.d/2026-11-02-stage-per-table-rows-at-eof.md new file mode 100644 index 00000000..c096e2c9 --- /dev/null +++ b/packages/rust/migrations.d/2026-11-02-stage-per-table-rows-at-eof.md @@ -0,0 +1,30 @@ +### Per-table rows staged at stdout EOF + +#### Summary + +A per-table `on-file` rerun after a watched change writes its rows inside a SQLite savepoint once the command's stdout closes, then keeps them on a zero exit or rolls them back on a failure. Events and failure behavior are unchanged. Nothing breaks. + +#### Required changes + +_None._ + +#### Deprecations removed + +_None._ + +#### Behavior changes without code changes + +_None._ + +#### Verification + +```bash +mkdir -p d +echo '{"v": 0}' > d/f.jl +printf '[[table]]\nname = "rows"\nglob = "**/*.jl"\nddl = "CREATE TABLE rows (v INTEGER)"\non-file = "cat"\n' > .dirsql.toml +dirsql server -c .dirsql.toml --port 7117 & +curl -sN http://localhost:7117/events & +echo '{"v": 1}' > d/f.jl +``` + +Prints a `rows` event carrying `"row":{"v":1}`. diff --git a/packages/rust/src/command.rs b/packages/rust/src/command.rs index 7c77935f..c8628b8e 100644 --- a/packages/rust/src/command.rs +++ b/packages/rust/src/command.rs @@ -286,12 +286,14 @@ const READ_BYTES: usize = 262_144; /// Run an already-built `argv` and hand its stdout to `on_block` as it /// arrives, in blocks that each hold whole lines (the last block may lack its /// trailing newline). The child runs with stdin closed. A non-zero exit is an -/// error after every block has been delivered. +/// error after every block has been delivered. `on_eof` runs once stdout has +/// closed, before the child's exit is known. pub(crate) fn run_argv_blocks( command: &str, argv: &[String], cwd: &Path, on_block: &mut dyn FnMut(&str), + on_eof: &mut dyn FnMut(), ) -> Result<(), CommandError> { let mut cmd = Command::new(&argv[0]); push_args(&mut cmd, &argv[1..]); @@ -310,10 +312,29 @@ pub(crate) fn run_argv_blocks( buf }); let mut stdout_pipe = child.stdout.take().expect("stdout piped"); + let status = + drain_then_reap(&mut stdout_pipe, on_block, on_eof, || child.wait()).map_err(|source| { + CommandError::Io { + command: command.to_string(), + source, + } + })?; + let stderr = stderr_thread.join().unwrap_or_default(); + check_exit(command, status, &stderr) +} + +/// Hand `stdout` to `on_block` in blocks of whole lines until it closes, run +/// `on_eof`, then `reap` the child, so work on the output overlaps its exit. +fn drain_then_reap( + stdout: &mut dyn Read, + on_block: &mut dyn FnMut(&str), + on_eof: &mut dyn FnMut(), + reap: impl FnOnce() -> T, +) -> T { let mut pending: Vec = Vec::new(); let mut buf = vec![0u8; READ_BYTES]; loop { - let n = match stdout_pipe.read(&mut buf) { + let n = match stdout.read(&mut buf) { Ok(0) | Err(_) => break, Ok(n) => n, }; @@ -326,13 +347,8 @@ pub(crate) fn run_argv_blocks( if !pending.is_empty() { on_block(&String::from_utf8_lossy(&pending)); } - - let status = child.wait().map_err(|source| CommandError::Io { - command: command.to_string(), - source, - })?; - let stderr = stderr_thread.join().unwrap_or_default(); - check_exit(command, status, &stderr) + on_eof(); + reap() } // One fn with cfg blocks rather than two cfg'd fns: cargo-mutants mutates @@ -809,10 +825,16 @@ mod tests { ]; let mut seen = String::new(); let mut blocks = 0; - run_argv_blocks("t", &argv, &cwd(), &mut |block| { - seen.push_str(block); - blocks += 1; - }) + run_argv_blocks( + "t", + &argv, + &cwd(), + &mut |block| { + seen.push_str(block); + blocks += 1; + }, + &mut || {}, + ) .unwrap(); assert_eq!(seen, "a\nb\nc"); assert!(blocks >= 1); @@ -827,11 +849,17 @@ mod tests { ]; let mut lines = 0; let mut blocks = 0; - run_argv_blocks("t", &argv, &cwd(), &mut |block| { - blocks += 1; - assert!(block.ends_with('\n'), "block cut mid-line"); - lines += block.lines().count(); - }) + run_argv_blocks( + "t", + &argv, + &cwd(), + &mut |block| { + blocks += 1; + assert!(block.ends_with('\n'), "block cut mid-line"); + lines += block.lines().count(); + }, + &mut || {}, + ) .unwrap(); assert_eq!(lines, 100_000); assert!(blocks > 1, "2.7 MB arrives in more than one read"); @@ -845,8 +873,14 @@ mod tests { "echo partial; echo oops >&2; exit 3".to_string(), ]; let mut seen = String::new(); - let err = - run_argv_blocks("t", &argv, &cwd(), &mut |block| seen.push_str(block)).unwrap_err(); + let err = run_argv_blocks( + "t", + &argv, + &cwd(), + &mut |block| seen.push_str(block), + &mut || {}, + ) + .unwrap_err(); assert_eq!(seen, "partial\n"); match err { CommandError::NonZeroExit { @@ -859,10 +893,26 @@ mod tests { } } + #[test] + fn drain_then_reap_ends_the_output_before_reaping_the_child() { + let order = std::cell::RefCell::new(Vec::new()); + let reaped = drain_then_reap( + &mut &b"a\nb"[..], + &mut |block| order.borrow_mut().push(block.to_string()), + &mut || order.borrow_mut().push("eof".to_string()), + || { + order.borrow_mut().push("reap".to_string()); + 7 + }, + ); + assert_eq!(reaped, 7); + assert_eq!(*order.borrow(), vec!["a\n", "b", "eof", "reap"]); + } + #[test] fn run_argv_blocks_names_a_missing_program() { let argv = vec!["definitely-not-a-real-binary-xyzzy".to_string()]; - let err = run_argv_blocks("t", &argv, &cwd(), &mut |_| {}).unwrap_err(); + let err = run_argv_blocks("t", &argv, &cwd(), &mut |_| {}, &mut || {}).unwrap_err(); assert!( matches!(err, CommandError::NotOnPath { .. }), "got: {err:?}" diff --git a/packages/rust/src/foreign_key.rs b/packages/rust/src/foreign_key.rs index 224ad3dc..02f4fd62 100644 --- a/packages/rust/src/foreign_key.rs +++ b/packages/rust/src/foreign_key.rs @@ -54,18 +54,32 @@ pub(crate) fn commit(conn: &Connection) -> Result<(), DbError> { pub(crate) fn atomically>( conn: &Connection, unit: impl FnOnce() -> Result, +) -> Result { + let value = stage(conn, unit)?; + keep(conn)?; + Ok(value) +} + +/// Apply `unit` but hold it open, for [`keep`] or [`discard`] to settle. A +/// unit that fails is discarded here and leaves nothing open. +pub(crate) fn stage>( + conn: &Connection, + unit: impl FnOnce() -> Result, ) -> Result { conn.execute_batch("SAVEPOINT dirsql_unit")?; - match unit() { - Ok(value) => { - conn.execute_batch("RELEASE dirsql_unit")?; - Ok(value) - } - Err(e) => { - conn.execute_batch("ROLLBACK TO dirsql_unit; RELEASE dirsql_unit")?; - Err(e) - } - } + unit().inspect_err(|_| { + let _ = discard(conn); + }) +} + +/// Keep the unit [`stage`] left open. +pub(crate) fn keep(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch("RELEASE dirsql_unit") +} + +/// Drop the unit [`stage`] left open, as if it had never run. +pub(crate) fn discard(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch("ROLLBACK TO dirsql_unit; RELEASE dirsql_unit") } /// The rows violating a foreign key right now, or `None` when there are none. @@ -402,4 +416,41 @@ mod tests { commit(&conn).unwrap(); assert_eq!(count(&conn, "parent"), 1); } + #[test] + fn a_staged_unit_is_visible_until_discarded() { + let conn = family(CHILD); + let out: rusqlite::Result = stage(&conn, || { + conn.execute_batch("INSERT INTO parent VALUES (1, 'a')")?; + Ok(7) + }); + assert_eq!(out.unwrap(), 7); + assert_eq!(count(&conn, "parent"), 1); + discard(&conn).unwrap(); + assert_eq!(count(&conn, "parent"), 0); + assert!(conn.is_autocommit()); + } + + #[test] + fn a_staged_unit_kept_stays() { + let conn = family(CHILD); + let out: rusqlite::Result<()> = stage(&conn, || { + conn.execute_batch("INSERT INTO parent VALUES (1, 'a')") + }); + out.unwrap(); + keep(&conn).unwrap(); + assert!(conn.is_autocommit()); + assert_eq!(count(&conn, "parent"), 1); + } + + #[test] + fn a_staged_unit_that_fails_leaves_nothing_open() { + let conn = family(CHILD); + let out: rusqlite::Result<()> = stage(&conn, || { + conn.execute_batch("INSERT INTO parent VALUES (1, 'a')")?; + conn.execute_batch("INSERT INTO nowhere VALUES (1)") + }); + assert!(out.is_err()); + assert!(conn.is_autocommit()); + assert_eq!(count(&conn, "parent"), 0); + } } diff --git a/packages/rust/src/lib.rs b/packages/rust/src/lib.rs index 6f8bbb14..76579beb 100644 --- a/packages/rust/src/lib.rs +++ b/packages/rust/src/lib.rs @@ -108,7 +108,10 @@ type OnFileFn = dyn Fn(&str) -> std::result::Result, BoxError> + Send + type RowSink<'a> = dyn FnMut(Vec) + Send + 'a; /// The paths a per-table hook runs over, yielded as the scan finds them. type PathFeed<'a> = dyn Iterator + Send + 'a; -type OnFilesFn = dyn Fn(&mut PathFeed<'_>, &mut RowSink<'_>) -> std::result::Result<(), BoxError> +/// What a per-table hook may call once its last chunk is out, before it knows +/// whether it succeeded, so its rows can be staged while it finishes. +type OutputEnd<'a> = dyn FnMut() + 'a; +type OnFilesFn = dyn Fn(&mut PathFeed<'_>, &mut RowSink<'_>, &mut OutputEnd<'_>) -> std::result::Result<(), BoxError> + Send + Sync + 'static; @@ -378,11 +381,16 @@ impl Table { + Sync + 'static, { - Self::per_table_streaming(name, ddl, glob, move |feed, sink: &mut RowSink<'_>| { - let paths: Vec = feed.collect(); - sink(on_files(&paths)?); - Ok(()) - }) + Self::per_table_streaming( + name, + ddl, + glob, + move |feed, sink: &mut RowSink<'_>, _: &mut OutputEnd<'_>| { + let paths: Vec = feed.collect(); + sink(on_files(&paths)?); + Ok(()) + }, + ) } /// [`per_table`](Self::per_table) for a hook that hands its rows over in @@ -394,7 +402,11 @@ impl Table { on_files: F, ) -> Self where - F: Fn(&mut PathFeed<'_>, &mut RowSink<'_>) -> std::result::Result<(), BoxError> + F: Fn( + &mut PathFeed<'_>, + &mut RowSink<'_>, + &mut OutputEnd<'_>, + ) -> std::result::Result<(), BoxError> + Send + Sync + 'static, @@ -1031,27 +1043,31 @@ impl DirSQL { .unwrap_or_default(), Err(e) => return vec![error_event(Some(table), trigger, e.to_string())], }; - let raw_rows = match collect_table_rows(hook.as_ref(), &abs_paths) { - Ok(rows) => rows, - Err(e) => return vec![error_event(Some(table), trigger, e.to_string())], - }; let strict = *self.inner.strict_map.get(table).unwrap_or(&false); - - let changes = { - let db = match self.inner.db.lock() { - Ok(g) => g, - Err(e) => return vec![error_event(Some(table), trigger, e.to_string())], - }; - let new_rows = match shape_rows(&db, table, raw_rows, strict) { - Ok(rows) => rows, - Err(message) => return vec![error_event(Some(table), trigger, message)], - }; - match foreign_key::atomically(db.conn(), || { - db.reconcile_batch_rows(table, BATCH_OWNER, &new_rows) - }) { - Ok(changes) => changes, - Err(e) => return vec![error_event(Some(table), trigger, e.to_string())], + let rows = Mutex::new(Vec::new()); + let take_rows = || std::mem::take(&mut *rows.lock().expect("hook rows lock")); + let mut staged = None; + let produced = hook( + &mut abs_paths.into_iter(), + &mut |chunk| rows.lock().expect("hook rows lock").extend(chunk), + &mut || staged = Some(self.stage_table_rows(table, take_rows(), strict)), + ); + let staged = match (produced, staged) { + (Err(e), staged) => { + if let Some(Ok(staged)) = staged { + let _ = foreign_key::discard(staged.db.conn()); + } + return vec![error_event(Some(table), trigger, e.to_string())]; } + (Ok(()), Some(staged)) => staged, + (Ok(()), None) => self.stage_table_rows(table, take_rows(), strict), + }; + let changes = match staged { + Ok(Staged { db, changes }) => match foreign_key::keep(db.conn()) { + Ok(()) => changes, + Err(e) => return vec![error_event(Some(table), trigger, e.to_string())], + }, + Err(message) => return vec![error_event(Some(table), trigger, message)], }; if changes.keyed { @@ -1067,6 +1083,23 @@ impl DirSQL { } } + /// Shape a per-table hook's rows and write them as `table`'s, held open + /// under the database lock until the hook's exit keeps or discards them. + fn stage_table_rows( + &self, + table: &str, + raw_rows: Vec, + strict: bool, + ) -> std::result::Result, String> { + let db = self.inner.db.lock().map_err(|e| e.to_string())?; + let rows = shape_rows(&db, table, raw_rows, strict)?; + let changes = foreign_key::stage(db.conn(), || { + db.reconcile_batch_rows(table, BATCH_OWNER, &rows) + }) + .map_err(|e| e.to_string())?; + Ok(Staged { db, changes }) + } + /// Record in the table's file set whether `abs_path` is, now, one of the /// files its per-table hook runs over. The set stays in walk order /// (component-wise byte order), so the hook sees the same argument list a @@ -2567,16 +2600,6 @@ fn shape_rows( .map_err(|e| e.to_string()) } -/// Every row a per-table hook produces over `paths`, gathered into one list. -fn collect_table_rows( - hook: &OnFilesFn, - paths: &[PathBuf], -) -> std::result::Result, BoxError> { - let mut rows = Vec::new(); - hook(&mut paths.iter().cloned(), &mut |chunk| rows.extend(chunk))?; - Ok(rows) -} - /// Replace a per-table hook's rows under [`BATCH_OWNER`], storing each chunk /// the hook emits while the hook, on its own thread, goes on producing the /// next. Chunks are shaped on a thread of their own, in order, so shaping one @@ -2660,6 +2683,12 @@ fn refill_table_rows( Ok(outcome) } +/// A per-table hook's rows written but not yet kept. +struct Staged<'a> { + db: std::sync::MutexGuard<'a, Db>, + changes: db::RowChanges, +} + /// A per-table hook running on its own thread, its rows arriving as chunks. struct Production { chunks: std::sync::mpsc::Receiver>, @@ -2689,9 +2718,13 @@ fn start_hook( if paths.peek().is_none() { return Ok(()); } - hook(&mut paths, &mut |chunk| { - let _ = sender.send(chunk); - }) + hook( + &mut paths, + &mut |chunk| { + let _ = sender.send(chunk); + }, + &mut || {}, + ) .map_err(|e| e.to_string()) }); Production { chunks, handle } @@ -2964,8 +2997,8 @@ fn build_tables_from_config(cfg: &config::Config, config_dir: &Path) -> Result, sink: &mut RowSink<'_>| { - run_on_files(&command, paths, &hook_cwd, &hook_root, sink) + move |paths: &mut PathFeed<'_>, sink: &mut RowSink<'_>, end: &mut OutputEnd<'_>| { + run_on_files(&command, paths, &hook_cwd, &hook_root, sink, end) }, ) .anchored(anchor) @@ -2992,9 +3025,10 @@ fn run_on_files( config_dir: &Path, root: &Path, sink: &mut RowSink<'_>, + end: &mut OutputEnd<'_>, ) -> std::result::Result<(), BoxError> { Ok(on_file::run_streaming( - command, config_dir, root, paths, sink, + command, config_dir, root, paths, sink, end, )?) } @@ -3478,7 +3512,7 @@ mod internal_tests { "items", "CREATE TABLE items (name TEXT)", "**/*.txt", - move |paths: &mut PathFeed<'_>, _: &mut RowSink<'_>| { + move |paths: &mut PathFeed<'_>, _: &mut RowSink<'_>, _: &mut OutputEnd<'_>| { for path in paths { let _ = got.lock().unwrap().send(path); } @@ -4794,7 +4828,7 @@ mod internal_tests { chunks: Vec>, outcome: std::result::Result<(), String>, ) -> Production { - let hook: Arc = Arc::new(move |_paths, sink| { + let hook: Arc = Arc::new(move |_paths, sink, _end| { for chunk in chunks.clone() { sink(chunk); } @@ -5399,12 +5433,14 @@ mod internal_tests { let Hook::PerTable(on_files) = &tables[0].hook else { panic!("a configured on-file command runs once per table"); }; - assert_eq!( - collect_table_rows(on_files.as_ref(), &[dir.path().join("f.a")]) - .unwrap() - .len(), - 1 - ); + let mut rows = Vec::new(); + on_files( + &mut std::iter::once(dir.path().join("f.a")), + &mut |chunk| rows.extend(chunk), + &mut || {}, + ) + .unwrap(); + assert_eq!(rows.len(), 1); assert!(tables[1].strict, "on-file table preserves strict flag"); } @@ -5574,6 +5610,7 @@ mod internal_tests { config_dir, root, &mut |chunk| rows.extend(chunk), + &mut || {}, )?; Ok(rows) } @@ -5605,6 +5642,7 @@ mod internal_tests { dir.path(), dir.path(), &mut |chunk| chunks.push(chunk), + &mut || {}, ) .expect("a well-formed payload parses"); assert!(chunks.len() > 1, "{} invocation(s)", chunks.len()); @@ -5991,7 +6029,7 @@ mod internal_tests { "items", "CREATE TABLE items (name TEXT)", "*.txt", - move |_: &mut PathFeed<'_>, sink: &mut RowSink<'_>| hook(sink), + move |_: &mut PathFeed<'_>, sink: &mut RowSink<'_>, _: &mut OutputEnd<'_>| hook(sink), ) } @@ -6421,6 +6459,69 @@ mod internal_tests { ); } + /// A per-table hook that answers `answer`'s name, ends its output, and + /// then fails when `answer` says so. + fn ending_per_table(answer: Arc>) -> Table { + Table::per_table_streaming( + "items", + "CREATE TABLE items (name TEXT)", + "**/*.txt", + move |_: &mut PathFeed<'_>, sink: &mut RowSink<'_>, end: &mut OutputEnd<'_>| { + let (name, fails) = answer.lock().unwrap().clone(); + sink(vec![name_json_row(&name)]); + end(); + if fails { + return Err("exited 1".into()); + } + Ok(()) + }, + ) + } + + fn ending_hook_db(answer: &Arc>) -> (TempDir, DirSQL, PathBuf) { + let dir = TempDir::new().unwrap(); + let root = dir.path().to_path_buf(); + let a = root.join("a.txt"); + let fake = Arc::new( + FakeFs::default().with_subtree(root.clone(), vec![(a.clone(), "items".into())]), + ); + let items = ending_per_table(Arc::clone(answer)); + let db = + DirSQL::with_ignore_and_fs(&root, vec![items], Vec::::new(), fake).unwrap(); + db.apply_file_events(vec![FileEvent::Created(a.clone())]); + (dir, db, a) + } + + #[test] + fn a_per_table_hook_that_fails_after_ending_its_output_writes_nothing() { + let answer = Arc::new(Mutex::new(("old".to_string(), false))); + let (_dir, db, a) = ending_hook_db(&answer); + *answer.lock().unwrap() = ("new".to_string(), true); + + let events = db.apply_file_events(vec![FileEvent::Modified(a)]); + + assert!( + matches!(&events[..], [RowEvent::Error { error, .. }] if error.contains("exited 1")), + "{events:?}" + ); + assert_eq!(row_names(&db), vec!["old"]); + } + + #[test] + fn a_per_table_hook_that_succeeds_after_ending_its_output_keeps_its_rows() { + let answer = Arc::new(Mutex::new(("old".to_string(), false))); + let (_dir, db, a) = ending_hook_db(&answer); + *answer.lock().unwrap() = ("new".to_string(), false); + + let events = db.apply_file_events(vec![FileEvent::Modified(a)]); + + assert!( + matches!(&events[..], [RowEvent::Update { .. }]), + "{events:?}" + ); + assert_eq!(row_names(&db), vec!["new"]); + } + #[test] fn apply_file_events_writes_only_the_rows_a_per_table_refresh_changed() { let dir = TempDir::new().unwrap(); diff --git a/packages/rust/src/on_file.rs b/packages/rust/src/on_file.rs index dd45cad8..3477c065 100644 --- a/packages/rust/src/on_file.rs +++ b/packages/rust/src/on_file.rs @@ -57,6 +57,7 @@ pub(crate) fn run( root, &mut paths.iter().cloned(), &mut |chunk| rows.extend(chunk), + &mut || {}, )?; Ok(rows) } @@ -90,9 +91,15 @@ impl Pool<'_> { } let mut full = self.argv.to_vec(); full.extend(args); - let outcome = run_invocation(self.command, &full, self.cwd, &mut |chunk| { - let _ = rows.send(chunk); - }); + let outcome = run_invocation( + self.command, + &full, + self.cwd, + &mut |chunk| { + let _ = rows.send(chunk); + }, + &mut || {}, + ); if let Err(message) = outcome { self.aborted.store(true, Ordering::SeqCst); self.failures @@ -110,14 +117,18 @@ impl Pool<'_> { /// /// `paths` is consumed as it yields: an invocation starts as soon as its /// argument list is full, while later paths are still arriving. +/// +/// `end` runs when a lone invocation's stdout closes, after its last rows and +/// before its exit is known. Several invocations never call it. pub(crate) fn run_streaming( command: &str, cwd: &Path, root: &Path, paths: &mut (dyn Iterator + Send), sink: &mut (dyn FnMut(Vec) + Send), + end: &mut dyn FnMut(), ) -> Result<(), String> { - run_streaming_on(command, cwd, root, paths, sink, &cpus) + run_streaming_on(command, cwd, root, paths, sink, end, &cpus) } /// [`run_streaming`] with the machine's CPU count asked of `cpus`. @@ -127,6 +138,7 @@ fn run_streaming_on( root: &Path, paths: &mut (dyn Iterator + Send), sink: &mut (dyn FnMut(Vec) + Send), + end: &mut dyn FnMut(), cpus: &(dyn Fn() -> usize + Sync), ) -> Result<(), String> { let Some(first) = paths.next() else { @@ -144,7 +156,7 @@ fn run_streaming_on( if let Some(args) = lone_run(&known, budget, cpus) { let mut full = argv; full.extend(args); - return run_invocation(command, &full, cwd, sink); + return run_invocation(command, &full, cwd, sink, end); } known_rest = known.into_iter(); let first = known_rest.next().expect("known holds the first path"); @@ -278,27 +290,34 @@ fn run_invocation( full: &[String], cwd: &Path, send: &mut dyn FnMut(Vec), + end: &mut dyn FnMut(), ) -> Result<(), String> { let mut bad_output: Option = None; let mut lines_seen = 0usize; - command::run_argv_blocks(command, full, cwd, &mut |block| { - if bad_output.is_some() { - return; - } - match parse_rows_at(block, lines_seen) { - Ok(rows) => { - lines_seen += block.bytes().filter(|&b| b == b'\n').count(); - if !rows.is_empty() { - send(rows); - } + command::run_argv_blocks( + command, + full, + cwd, + &mut |block| { + if bad_output.is_some() { + return; } - Err(message) => { - bad_output = Some(format!( - "on-file output was not one JSON object per line: {message}" - )); + match parse_rows_at(block, lines_seen) { + Ok(rows) => { + lines_seen += block.bytes().filter(|&b| b == b'\n').count(); + if !rows.is_empty() { + send(rows); + } + } + Err(message) => { + bad_output = Some(format!( + "on-file output was not one JSON object per line: {message}" + )); + } } - } - }) + }, + end, + ) .map_err(spawn_failure)?; bad_output.map_or(Ok(()), Err) } @@ -721,12 +740,31 @@ mod tests { Path::new("."), &mut paths.into_iter(), &mut |chunk| rows.extend(chunk), + &mut || {}, &|| panic!("counted the cpus"), ) .unwrap(); assert_eq!(rows.len(), 1); } + #[test] + fn run_streaming_ends_a_lone_invocations_output_once_after_its_rows() { + let dir = tempfile::tempdir().unwrap(); + let paths = files_of(dir.path(), 2, 100); + let order = Mutex::new(Vec::new()); + run_streaming_on( + r#"sh -c 'echo "{\"n\":$#}"' sh"#, + Path::new("."), + Path::new("."), + &mut paths.into_iter(), + &mut |chunk| order.lock().unwrap().push(chunk.len()), + &mut || order.lock().unwrap().push(0), + &|| 1, + ) + .unwrap(); + assert_eq!(*order.lock().unwrap(), vec![1, 0]); + } + fn two_paths() -> (Vec, Vec, usize) { let paths = vec![PathBuf::from("/a"), PathBuf::from("/b")]; let spawned: Vec = paths diff --git a/packages/ts/e2e-attestations/bot-1562-hook-savepoint-1102.json b/packages/ts/e2e-attestations/bot-1562-hook-savepoint-1102.json new file mode 100644 index 00000000..ae345664 --- /dev/null +++ b/packages/ts/e2e-attestations/bot-1562-hook-savepoint-1102.json @@ -0,0 +1,7 @@ +{ + "command": "pnpm test:e2e", + "ran_at": 1793671855, + "exit_code": 0, + "commit": "db6ce658a452e7868309dbae046797e9cbbab0c4", + "branch": "bot/1562-hook-savepoint-1102" +}