perf: apply a per-table hook's rows at stdout EOF, before its exit is reaped

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"
+}

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.

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}`.

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<T>(
+    stdout: &mut dyn Read,
+    on_block: &mut dyn FnMut(&str),
+    on_eof: &mut dyn FnMut(),
+    reap: impl FnOnce() -> T,
+) -> T {
     let mut pending: Vec<u8> = 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:?}"

packages/rust/src/foreign_key.rs

@@ -54,18 +54,32 @@ pub(crate) fn commit(conn: &Connection) -> Result<(), DbError> {
 pub(crate) fn atomically<T, E: From<rusqlite::Error>>(
     conn: &Connection,
     unit: impl FnOnce() -> Result<T, E>,
+) -> Result<T, E> {
+    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<T, E: From<rusqlite::Error>>(
+    conn: &Connection,
+    unit: impl FnOnce() -> Result<T, E>,
 ) -> Result<T, E> {
     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<i32> = 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);
+    }
 }