on master help sign in
Share caller db with compiled snippets so compile results are transactional in atomic

`base.compile` used to open its OWN SQLite connection to the host's db file. A compiled value used inside the host's `atomic` block therefore ran outside the atomic's SAVEPOINT: it couldn't see the atomic's uncommitted writes, and its own writes escaped the transaction (invisible to rollback, re-applied on retry). Run the snippet on the CALLER's live db handle instead. A new CURRENT_DB thread-local records the db of the knot code currently executing on this thread, set at each foreign-boundary crossing and restored on return (knot_run_user_main, knot_value_call, knot_io_run). compile_run uses CURRENT_DB when set — the normal case, since the host calls base.compile from running code — and only falls back to opening its own connection when there is no caller (the compiler's const-eval thread). Compiled closures/IO already receive the caller's db per-call, so a compiled value inside an atomic block now shares the host's transaction: it reads the atomic's uncommitted writes, and its own writes roll back if the atomic fails. Tests (persistence e2e): a compiled closure called inside atomic sees the atomic's uncommitted writes; a compiled IO's write rolls back when the atomic body panics; the ordinary (non-atomic) compile path is unchanged.

commit553ca191663c4f2910a081b09844ffed159ace0b authoriko <ilyakooo0@gmail.com> dateFri 4 Sep 2026, 09:10 (1 month ago) parent8faff2f
2 files changed+197−10
crates/knot-jit/tests/persistence.rs+126−1 crates/knot-runtime/src/lib.rs+71−9
crates/knot-jit/tests/persistence.rs+126−1
@@ -3,7 +3,18 @@
33//! process-level effect). Builds and runs real binaries.
44
55mod e2e;
6−use e2e::assert_stdout;
6+use e2e::{assert_stdout, run_program};
7+
8+/// Line-set comparison (order-independent): forked output interleaves.
9+fn assert_stdout_unordered(name: &str, src: &str, expected: &[&str]) {
10+ let (out, err, code) = run_program(name, src);
11+ assert_eq!(code, 0, "program {name} exited {code}\nstderr:\n{err}");
12+ let mut got: Vec<&str> = out.trim_end().lines().collect();
13+ let mut want: Vec<&str> = expected.to_vec();
14+ got.sort_unstable();
15+ want.sort_unstable();
16+ assert_eq!(got, want, "stdout line-set mismatch for {name}");
17+}
718
819#[test]
920fn persisted_relation_groupby() {
@@ -709,3 +720,117 @@ Rel E *emps
709720 // behavior, not something this fix changes; the fix is that the *message*
710721 // is the clean overflow report, not the raw SQLite error.
711722}
723+// ── base.compile results are transactional inside `atomic` ─────────────────
724+//
725+// A compiled snippet used to open its OWN SQLite connection to the host's db
726+// file, so its writes were invisible to — and unrolled-back by — the host's
727+// `atomic` SAVEPOINT. `base.compile` now runs the snippet on the caller's live
728+// db handle (the runtime's CURRENT_DB, set at every knot_value_call/knot_io_run
729+// boundary), so a compiled value inside an `atomic` block shares the host's
730+// transaction: it reads the atomic's uncommitted writes, and its own writes
731+// roll back if the atomic fails.
732+
733+#[test]
734+fn compile_closure_in_atomic_sees_uncommitted_writes() {
735+ // The compiled closure reads *accounts at CALL time (a `\_ ->` function),
736+ // so invoking it inside the atomic — after the atomic's own write — must
737+ // see all 3 rows, not the 1 committed before the atomic began.
738+ assert_stdout(
739+ "compile_atomic_read",
740+ r#"with {
741+Account {name Text balance (Int 1)}
742+Rel Account *accounts
743+}
744+(|
745+ *accounts = [{name "a" balance 1}]
746+ (? (base.compile "with {\nAccount {name Text balance (Int 1)}\nRel Account *accounts\n}\n(\\_ -> base.count *accounts)")
747+ Result.Ok {value countNow} (|
748+ _ <- atomic |
749+ *accounts = [{name "a" balance 1} {name "b" balance 2} {name "c" balance 3}]
750+ base.println (base.show (countNow 0))
751+ yield {}
752+ yield {})
753+ Result.Err {error e} base.println ("compile err: " ++ e))
754+ yield {})"#,
755+ // Sees the atomic's uncommitted write: 3 rows, not 1.
756+ "\"[3]\"\n{}",
757+ );
758+}
759+
760+#[test]
761+fn compile_io_write_in_atomic_rolls_back_on_panic() {
762+ // The compiled IO writes a row inside the atomic; the body then panics
763+ // (integer overflow) so the atomic rolls back. If the snippet shared the
764+ // host's connection, its write is undone and the final count stays [1].
765+ // (On the old separate-connection behavior the row would persist.)
766+ assert_stdout_unordered(
767+ "compile_atomic_rollback",
768+ r#"with {
769+Account {name Text balance (Int 1)}
770+Rel Account *accounts
771+}
772+(|
773+ *accounts = [{name "a" balance 1}]
774+ (? (base.compile "with {\nAccount {name Text balance (Int 1)}\nRel Account *accounts\n}\n(| *accounts = [{name \"sneak\" balance 999}] yield 7)")
775+ Result.Ok {value io} (|
776+ base.fork (atomic |
777+ r <- io
778+ base.println (base.show (9223372036854775807 + 1))
779+ yield {})
780+ base.sleep 300
781+ yield {})
782+ Result.Err {error e} base.println ("compile err: " ++ e))
783+ base.println ("final: " ++ (base.show (base.count *accounts)))
784+ yield {})"#,
785+ &["\"final: [1]\"", "{}"],
786+ );
787+}
788+
789+#[test]
790+fn compile_outside_atomic_still_works() {
791+ // The shared-connection change must not break the ordinary path: a
792+ // compiled pure value used outside any atomic behaves exactly as before.
793+ assert_stdout(
794+ "compile_plain",
795+ r#"(? (base.the (Result Text (Int 1)) (base.compile "6 * 7"))
796+ Result.Ok {value v} base.println (base.show v)
797+ Result.Err {error e} base.println ("err: " ++ e))"#,
798+ "\"42\"\n{}",
799+ );
800+}
801+
802+/// A compiled reader gated by `base.retry`: the snippet reads `*flag` on the
803+/// host's shared connection, so its read registers with the STM change-tracker
804+/// and the atomic is woken when the forked writer flips the flag. On the old
805+/// separate-connection behavior the snippet's read was invisible to the host's
806+/// STM, so the retry would not be driven by the snippet's read. The compiled
807+/// closure runs fresh on each retry iteration and eventually sees the flag.
808+#[test]
809+fn compile_reader_in_atomic_wakes_on_retry() {
810+ assert_stdout(
811+ "compile_atomic_retry",
812+ r#"with {
813+Flag {ready (Int 1)}
814+Rel Flag *flag
815+}
816+(|
817+ *flag = [{ready 0}]
818+ base.fork (|
819+ base.sleep 200
820+ *flag = [{ready 1}]
821+ yield {})
822+ (? (base.compile "with {\nFlag {ready (Int 1)}\nRel Flag *flag\n}\n(\\_ -> base.count *flag)")
823+ Result.Ok {value countNow} (|
824+ got <- atomic |
825+ (? (base.count (countNow 0))
826+ [1] (yield "set")
827+ _ base.retry)
828+ base.println ("compiled reader: " ++ (base.show got))
829+ yield {})
830+ Result.Err {error e} base.println ("compile err: " ++ e))
831+ yield {})"#,
832+ // Blocked in retry until the fork set the flag; the compiled read then
833+ // saw it and the atomic committed.
834+ "\"compiled reader: set\"\n{}",
835+ );
836+}
crates/knot-runtime/src/lib.rs+71−9
@@ -8804,6 +8804,11 @@ pub extern "C-unwind" fn knot_value_call(
88048804) -> *mut Value {
88058805 match unsafe { as_ref(func) } {
88068806 Value::Function(f) => {
8807+ // Establish the callee's db as this thread's current-db for the
8808+ // duration of the call, so a `base.compile` inside it (or inside a
8809+ // deeper call) runs its snippet on this handle — sharing any
8810+ // enclosing `atomic` SAVEPOINT — then restore the caller's.
8811+ let _db_scope = RestoreDb::new(db);
88078812 let fun: extern "C-unwind" fn(*mut c_void, *mut Value, *mut Value) -> *mut Value =
88088813 unsafe { std::mem::transmute(f.fn_ptr) };
88098814 fun(db, f.env, arg)
@@ -9472,6 +9477,9 @@ pub extern "C-unwind" fn knot_io_pure(val: *mut Value) -> *mut Value {
94729477/// Execute an IO thunk. If the value is not IO, return it as-is.
94739478#[unsafe(no_mangle)]
94749479pub extern "C-unwind" fn knot_io_run(db: *mut c_void, val: *mut Value) -> *mut Value {
9480+ // Current-db for the whole IO chain: any `base.compile` reached while
9481+ // running these thunks runs on this handle (sharing an enclosing atomic).
9482+ let _db_scope = RestoreDb::new(db);
94759483 let mut current = val;
94769484 loop {
94779485 // Cooperative cancellation: a `race` loser exits its IO chain at the
@@ -11857,16 +11865,28 @@ fn compile_run(
1185711865 Some("no JIT linked (knot-compile-rt not registered)".into()),
1185811866 );
1185911867 };
11860− // Open a db handle for the snippet. knot_db_open records the path on first
11861− // open (DB_PATH), so reuse it; a pure program (no store) gets :memory:.
11862− // A snippet reading the host's persisted relations thus sees real data.
11863− let db_path = DB_PATH.lock().unwrap_or_else(|e| e.into_inner()).clone();
11864− let (pstr, plen) = if db_path.is_empty() {
11865− (":memory:".as_bytes().as_ptr(), ":memory:".len())
11866− } else {
11867− (db_path.as_ptr(), db_path.len())
11868+ // Run the snippet on the CALLER's live db handle when one is active
11869+ // (the normal case: the host program called `base.compile` from running
11870+ // code, so CURRENT_DB holds its handle). This makes the compiled value
11871+ // share the host's connection — and any enclosing `atomic` SAVEPOINT — so
11872+ // its reads see uncommitted writes and its writes roll back with the
11873+ // transaction. When there is no caller (the compiler's const-eval thread),
11874+ // fall back to opening a connection: reuse the recorded host path so a
11875+ // snippet reading persisted relations sees real data, else `:memory:`.
11876+ let db = {
11877+ let cur = current_db();
11878+ if !cur.is_null() {
11879+ cur
11880+ } else {
11881+ let db_path = DB_PATH.lock().unwrap_or_else(|e| e.into_inner()).clone();
11882+ let (pstr, plen) = if db_path.is_empty() {
11883+ (":memory:".as_bytes().as_ptr(), ":memory:".len())
11884+ } else {
11885+ (db_path.as_ptr(), db_path.len())
11886+ };
11887+ knot_db_open(pstr, plen)
11888+ }
1186811889 };
11869− let db = knot_db_open(pstr, plen);
1187011890
1187111891 // The host's expected type (source-annotation string) forwarded to the JIT,
1187211892 // which runs the real-type subsumption. Empty → unpinned.
@@ -14272,6 +14292,9 @@ pub extern "C-unwind" fn knot_run_user_main(
1427214292) -> *mut c_void {
1427314293 let f: extern "C-unwind" fn(*mut c_void) -> *mut c_void =
1427414294 unsafe { std::mem::transmute(main_fn) };
14295+ // Establish the program's db as current-db so a top-level `base.compile`
14296+ // (before any knot_value_call/knot_io_run) runs its snippet on this handle.
14297+ let _db_scope = RestoreDb::new(db);
1427514298 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| f(db))) {
1427614299 Ok(v) => v,
1427714300 Err(payload) => {
@@ -19694,6 +19717,45 @@ pub extern "C-unwind" fn knot_constraint_register(
1969419717
1969519718// ── Atomic (transactions) ─────────────────────────────────────────
1969619719
19720+thread_local! {
19721+ /// The db handle of the knot code currently executing on this thread —
19722+ /// the handle the innermost active `knot_user_main`/`knot_value_call`/
19723+ /// `knot_io_run` was invoked with. `base.compile` runs its snippet on this
19724+ /// handle (rather than opening a fresh connection) so a compiled value used
19725+ /// inside the host's `atomic` block shares the host's SAVEPOINT: its reads
19726+ /// see uncommitted writes and its own writes roll back with the transaction.
19727+ ///
19728+ /// Set on entry to the foreign-boundary crossings (`knot_run_user_main`,
19729+ /// `knot_value_call`, `knot_io_run`) and restored on return, so a nested
19730+ /// `compile`'s snippet doesn't clobber the host's current-db after the
19731+ /// snippet returns. Null outside any knot execution (e.g. the compiler's
19732+ /// const-eval thread), where `compile` falls back to its own connection.
19733+ static CURRENT_DB: std::cell::Cell<*mut c_void> = const { std::cell::Cell::new(std::ptr::null_mut()) };
19734+}
19735+
19736+/// Set the current-db for this thread, returning the previous value so the
19737+/// caller can restore it on scope exit (see [`RestoreDb`]).
19738+fn set_current_db(db: *mut c_void) -> *mut c_void {
19739+ CURRENT_DB.with(|c| c.replace(db))
19740+}
19741+
19742+/// RAII guard restoring the previous current-db on drop.
19743+struct RestoreDb(*mut c_void);
19744+impl RestoreDb {
19745+ fn new(db: *mut c_void) -> Self {
19746+ RestoreDb(set_current_db(db))
19747+ }
19748+}
19749+impl Drop for RestoreDb {
19750+ fn drop(&mut self) {
19751+ set_current_db(self.0);
19752+ }
19753+}
19754+
19755+/// The current thread's db handle, or null when outside knot execution.
19756+fn current_db() -> *mut c_void {
19757+ CURRENT_DB.with(|c| c.get())
19758+}
1969719759
1969819760#[unsafe(no_mangle)]
1969919761pub extern "C-unwind" fn knot_atomic_begin(db: *mut c_void) {