@@ -1229,6 +1229,11 @@ static ACTIVE_FORKS_CVAR: Condvar = Condvar::new();
12291229pub struct CancelToken {
12301230 flag: AtomicBool,
12311231 notify: (Mutex<()>, Condvar),
1232+ /// `WakeSlot` registered by `knot_stm_wait` while this thread is parked
1233+ /// in an STM retry, so `cancel()` can cut that park short too — without
1234+ /// this, a cancelled `race` loser blocked in `retry` would sleep out the
1235+ /// full `STM_WAIT_TIMEOUT` and stall `knot_threads_join` at program exit.
1236+ stm_slot: Mutex<Option<Weak<WakeSlot>>>,
12321237}
12331238
12341239impl CancelToken {
@@ -1236,6 +1241,7 @@ impl CancelToken {
12361241 Self {
12371242 flag: AtomicBool::new(false),
12381243 notify: (Mutex::new(()), Condvar::new()),
1244+ stm_slot: Mutex::new(None),
12391245 }
12401246 }
12411247
@@ -1248,9 +1254,30 @@ impl CancelToken {
12481254 /// just checked the flag and is about to wait cannot miss the
12491255 /// wake-up.
12501256 fn cancel(&self) {
1251− let _g = self.notify.0.lock().unwrap();
1252− self.flag.store(true, Ordering::Release);
1253− self.notify.1.notify_all();
1257+ {
1258+ let _g = self.notify.0.lock().unwrap();
1259+ self.flag.store(true, Ordering::Release);
1260+ self.notify.1.notify_all();
1261+ }
1262+ // Also wake an STM wait slot if the thread is parked in retry.
1263+ // Race-free pairing with `knot_stm_wait`: the waiter registers its
1264+ // slot here BEFORE checking `is_cancelled()` — so either we see the
1265+ // registered slot (and wake it), or the waiter's check sees the flag
1266+ // (the `stm_slot` mutex orders the two paths).
1267+ let slot = self.stm_slot.lock().unwrap().take();
1268+ if let Some(slot) = slot.and_then(|w| w.upgrade()) {
1269+ slot.wake();
1270+ }
1271+ }
1272+
1273+ /// Register the current STM wake slot so `cancel()` can wake it.
1274+ fn set_stm_wake_slot(&self, slot: &Arc<WakeSlot>) {
1275+ *self.stm_slot.lock().unwrap() = Some(Arc::downgrade(slot));
1276+ }
1277+
1278+ /// Clear the registered STM wake slot after the wait completes.
1279+ fn clear_stm_wake_slot(&self) {
1280+ *self.stm_slot.lock().unwrap() = None;
12541281 }
12551282
12561283 /// Sleep for `duration`, returning early as soon as `cancel()` is
@@ -1674,7 +1701,22 @@ fn sql_partial_cmp(a: &SqlVal, b: &SqlVal) -> Option<std::cmp::Ordering> {
16741701 (SqlVal::Real(x), SqlVal::Real(y)) => x.partial_cmp(y),
16751702 (SqlVal::Int(x), SqlVal::Real(y)) => (*x as f64).partial_cmp(y),
16761703 (SqlVal::Real(x), SqlVal::Int(y)) => x.partial_cmp(&(*y as f64)),
1677− (SqlVal::Text(x), SqlVal::Text(y)) => Some(x.cmp(y)),
1704+ (SqlVal::Text(x), SqlVal::Text(y)) => {
1705+ // Int columns are stored as TEXT (`SqlVal::from_knot` serializes
1706+ // Int as Text), so numeric texts must compare numerically —
1707+ // "5" < "10" — not lexicographically. Mirrors `SqlVal::to_key`'s
1708+ // i64-parse collapse: i64 first, then f64; plain string order
1709+ // only when NEITHER side parses as a number. Mixed numeric /
1710+ // non-numeric is unknown → None (callers wake conservatively).
1711+ match (x.parse::<i64>(), y.parse::<i64>()) {
1712+ (Ok(a), Ok(b)) => Some(a.cmp(&b)),
1713+ _ => match (x.parse::<f64>(), y.parse::<f64>()) {
1714+ (Ok(a), Ok(b)) => a.partial_cmp(&b),
1715+ (Err(_), Err(_)) => Some(x.cmp(y)),
1716+ _ => None,
1717+ },
1718+ }
1719+ }
16781720 // BigInts are stored as TEXT in SQLite — handle Int ↔ Text via numeric parse.
16791721 (SqlVal::Int(x), SqlVal::Text(y)) => y.parse::<i64>().ok().map(|n| x.cmp(&n)),
16801722 (SqlVal::Text(x), SqlVal::Int(y)) => x.parse::<i64>().ok().map(|n| n.cmp(y)),
@@ -2402,12 +2444,19 @@ fn wait_for_any_write(timeout: Duration) {
24022444fn bump_table_version(name: &str) {
24032445 if let Some(v) = TABLE_VERSIONS.get(name) {
24042446 v.fetch_add(1, Ordering::Release);
2405− return;
2447+ } else {
2448+ TABLE_VERSIONS
2449+ .entry(name.to_string())
2450+ .or_insert_with(|| Arc::new(AtomicU64::new(0)))
2451+ .fetch_add(1, Ordering::Release);
24062452 }
2407− TABLE_VERSIONS
2408− .entry(name.to_string())
2409− .or_insert_with(|| Arc::new(AtomicU64::new(0)))
2410− .fetch_add(1, Ordering::Release);
2453+ // TOCTOU guard (Dekker pattern), paired with the fence in
2454+ // `knot_stm_wait`: order the version bump before this writer's
2455+ // subsequent reads of watcher-registration state (`table_has_watchers`,
2456+ // `TABLE_FILTER_COLS` in `wake_matching_watchers`, the watcher indices).
2457+ // A watcher registering concurrently either gets seen by those reads or
2458+ // observes this bump in its post-registration re-check.
2459+ std::sync::atomic::fence(Ordering::SeqCst);
24112460}
24122461
24132462/// Walk the broad watcher list for `name`, then for `WriteEvent::Rows` the EQ
@@ -2424,6 +2473,23 @@ fn bump_table_version(name: &str) {
24242473/// Reads only — dead `Weak`s linger here and are reclaimed by `sweep_dead_watchers`
24252474/// on a counter trip via `maybe_sweep_dead_watchers`.
24262475fn wake_matching_watchers(name: &str, event: &WriteEvent) {
2476+ // TOCTOU guard: a `Rows` payload was projected to the filter-column
2477+ // union read at WRITE time (`filter_cols_for`); a watcher that
2478+ // registered a filter on a NEW column between that projection and this
2479+ // notify would never be probed — indexed slots are only discovered via
2480+ // payload columns, and `row_matches_preds_indexed`'s missing-column
2481+ // fallback only helps slots that get probed at all. Re-read the current
2482+ // filter-column set here: if it references a column absent from the
2483+ // payload, fall back to a conservative `Bulk` wake. Paired with the
2484+ // SeqCst fences in `bump_table_version` (runs just before this on every
2485+ // notify path) and `knot_stm_wait` (between registration and the version
2486+ // re-check), at least one side always observes the other.
2487+ let event: &WriteEvent = match event {
2488+ WriteEvent::Rows(rows) if filter_cols_missing_from_payload(name, rows) => {
2489+ &WriteEvent::Bulk
2490+ }
2491+ e => e,
2492+ };
24272493 // 1) Broad watchers — All filters and non-indexable Cols.
24282494 if let Some(slots) = TABLE_WATCHERS.get(name) {
24292495 for weak in slots.iter() {
@@ -2472,6 +2538,17 @@ fn wake_matching_watchers(name: &str, event: &WriteEvent) {
24722538 maybe_sweep_dead_watchers();
24732539}
24742540
2541+/// True if any column currently registered in `TABLE_FILTER_COLS` for
2542+/// `table` is missing from the event payload — i.e. the payload's projection
2543+/// is stale relative to the live watcher set and per-row filtering can no
2544+/// longer be trusted. See the TOCTOU comment in `wake_matching_watchers`.
2545+fn filter_cols_missing_from_payload(table: &str, rows: &EventRows) -> bool {
2546+ match TABLE_FILTER_COLS.get(table) {
2547+ Some(state) => state.cols.keys().any(|c| rows.col_index(c).is_none()),
2548+ None => false,
2549+ }
2550+}
2551+
24752552/// Conservatively wake every EQ- and Range-indexed watcher registered for
24762553/// `table`. Used for `WriteEvent::Bulk` events, which carry no row payload:
24772554/// any value may have changed, so every indexed predicate may now hold.
@@ -3119,8 +3196,10 @@ pub extern "C-unwind" fn knot_override_lookup(
31193196 };
31203197
31213198 if wrap_maybe {
3122− let tag = intern_str("Just");
3123− alloc(Value::Constructor(tag, inner))
3199+ // `Just`'s payload is always the record `{value: ...}` — a bare
3200+ // scalar payload would not match the shape pattern-matching and
3201+ // `unwrap_maybe` expect.
3202+ make_just(inner)
31243203 } else {
31253204 inner
31263205 }
@@ -4789,17 +4868,63 @@ fn infer_temp_schema(rows: &[*mut Value]) -> Option<TempSchema> {
47894868 return Some(TempSchema::Scalar(ColType::Text));
47904869 }
47914870 match unsafe { as_ref(first) } {
4792− Value::Record(fields) => {
4793− let mut cols = Vec::with_capacity(fields.len());
4794− for f in fields {
4795− if !f.value.is_null() {
4871+ Value::Record(first_fields) => {
4872+ // Scan ALL rows, not just the first: a first-row-only guess
4873+ // corrupts data when later rows disagree — e.g. a Maybe field
4874+ // whose first row is `Nothing` (nullary → Tag column) silently
4875+ // drops the payload of a later `Just {value: ...}` row, and a
4876+ // first-row NULL types the column Text, mis-reading later Ints.
4877+ // Any inconsistency (mixed nullary/payload constructors, NULL vs
4878+ // typed, conflicting scalar types, differing field sets) returns
4879+ // None so callers take the safe in-memory fallback.
4880+ let names: Vec<Arc<str>> = first_fields.iter().map(|f| f.name.clone()).collect();
4881+ // Per-field type, refined across rows. `None` = only NULLs so far.
4882+ let mut types: Vec<Option<ColType>> = vec![None; names.len()];
4883+ let mut saw_null: Vec<bool> = vec![false; names.len()];
4884+ for row in rows {
4885+ if row.is_null() {
4886+ return None;
4887+ }
4888+ let fields = match unsafe { as_ref(*row) } {
4889+ Value::Record(fields) => fields,
4890+ _ => return None,
4891+ };
4892+ if fields.len() != names.len() {
4893+ return None;
4894+ }
4895+ for (i, f) in fields.iter().enumerate() {
4896+ if f.name != names[i] {
4897+ return None;
4898+ }
4899+ if f.value.is_null() {
4900+ saw_null[i] = true;
4901+ continue;
4902+ }
47964903 match unsafe { as_ref(f.value) } {
47974904 Value::Relation(_) | Value::Function(_) => return None,
47984905 _ => {}
47994906 }
4907+ let ty = infer_col_type(f.value)?;
4908+ match types[i] {
4909+ None => types[i] = Some(ty),
4910+ Some(prev) if prev == ty => {}
4911+ Some(_) => return None,
4912+ }
48004913 }
4801− let ty = infer_col_type(f.value)?;
4802− cols.push((f.name.to_string(), ty));
4914+ }
4915+ let mut cols = Vec::with_capacity(names.len());
4916+ for (i, name) in names.iter().enumerate() {
4917+ let ty = match types[i] {
4918+ // NULL and typed values mixed in one column: SQLite
4919+ // would conflate "no value" with the column's affinity
4920+ // games — bail to the in-memory path.
4921+ Some(_) if saw_null[i] => return None,
4922+ Some(ty) => ty,
4923+ // Every row NULL: Text (matches `infer_col_type`'s
4924+ // historical null handling).
4925+ None => ColType::Text,
4926+ };
4927+ cols.push((name.to_string(), ty));
48034928 }
48044929 Some(TempSchema::Record(cols))
48054930 }
@@ -5367,16 +5492,20 @@ pub extern "C-unwind" fn knot_relation_singleton(v: *mut Value) -> *mut Value {
53675492}
53685493
53695494/// Unwrap a scalar source relation: extract the `_value` field from the first row.
5370−/// Returns a default (Int 0) if the relation is empty.
5495+/// Panics if the relation is empty — a scalar source must hold exactly one
5496+/// row, so an empty backing table means the database is corrupted (returning
5497+/// a wrong-typed default like Int 0 would silently corrupt downstream data).
53715498#[unsafe(no_mangle)]
53725499pub extern "C-unwind" fn knot_scalar_source_unwrap(rel: *mut Value) -> *mut Value {
53735500 match unsafe { as_ref(rel) } {
53745501 Value::Relation(rows) => {
53755502 if rows.is_empty() {
5376− alloc_int(0)
5377− } else {
5378− knot_record_field(rows[0], "_value".as_ptr(), 6)
5503+ panic!(
5504+ "knot runtime: scalar source has no row — database corrupted? \
5505+ (expected exactly one row holding the scalar value)"
5506+ );
53795507 }
5508+ knot_record_field(rows[0], "_value".as_ptr(), 6)
53805509 }
53815510 _ => rel,
53825511 }
@@ -6160,11 +6289,19 @@ pub extern "C-unwind" fn knot_value_neq_i32(a: *mut Value, b: *mut Value) -> i32
61606289 !values_equal(a, b) as i32
61616290}
61626291
6292+/// Diagnostic for null operands in ordered comparisons. Shared by the
6293+/// `<`/`>` paths (via `compare_lt`/`compare_gt`) and the `<=`/`>=` wrappers,
6294+/// which must short-circuit BEFORE the negated compare call (a silent
6295+/// `!compare_gt(null, ..)` would otherwise return a wrong `true`).
6296+fn warn_null_comparison(a: *mut Value, b: *mut Value) {
6297+ eprintln!("knot runtime: comparison with null value (a={}, b={})",
6298+ if a.is_null() { "null".to_string() } else { brief_value(a) },
6299+ if b.is_null() { "null".to_string() } else { brief_value(b) });
6300+}
6301+
61636302fn compare_lt(a: *mut Value, b: *mut Value) -> bool {
61646303 if a.is_null() || b.is_null() {
6165− eprintln!("knot runtime: comparison with null value (a={}, b={})",
6166− if a.is_null() { "null".to_string() } else { brief_value(a) },
6167− if b.is_null() { "null".to_string() } else { brief_value(b) });
6304+ warn_null_comparison(a, b);
61686305 return false;
61696306 }
61706307 let av = unsafe { as_ref(a) };
@@ -6183,9 +6320,7 @@ fn compare_lt(a: *mut Value, b: *mut Value) -> bool {
61836320
61846321fn compare_gt(a: *mut Value, b: *mut Value) -> bool {
61856322 if a.is_null() || b.is_null() {
6186− eprintln!("knot runtime: comparison with null value (a={}, b={})",
6187− if a.is_null() { "null".to_string() } else { brief_value(a) },
6188− if b.is_null() { "null".to_string() } else { brief_value(b) });
6323+ warn_null_comparison(a, b);
61896324 return false;
61906325 }
61916326 let av = unsafe { as_ref(a) };
@@ -6216,13 +6351,19 @@ pub extern "C-unwind" fn knot_value_gt(a: *mut Value, b: *mut Value) -> *mut Val
62166351
62176352#[unsafe(no_mangle)]
62186353pub extern "C-unwind" fn knot_value_le(a: *mut Value, b: *mut Value) -> *mut Value {
6219− if a.is_null() || b.is_null() { return alloc_bool(false); }
6354+ if a.is_null() || b.is_null() {
6355+ warn_null_comparison(a, b);
6356+ return alloc_bool(false);
6357+ }
62206358 alloc_bool(!compare_gt(a, b))
62216359}
62226360
62236361#[unsafe(no_mangle)]
62246362pub extern "C-unwind" fn knot_value_ge(a: *mut Value, b: *mut Value) -> *mut Value {
6225− if a.is_null() || b.is_null() { return alloc_bool(false); }
6363+ if a.is_null() || b.is_null() {
6364+ warn_null_comparison(a, b);
6365+ return alloc_bool(false);
6366+ }
62266367 alloc_bool(!compare_lt(a, b))
62276368}
62286369
@@ -6240,13 +6381,19 @@ pub extern "C-unwind" fn knot_value_gt_i32(a: *mut Value, b: *mut Value) -> i32
62406381
62416382#[unsafe(no_mangle)]
62426383pub extern "C-unwind" fn knot_value_le_i32(a: *mut Value, b: *mut Value) -> i32 {
6243− if a.is_null() || b.is_null() { return 0; }
6384+ if a.is_null() || b.is_null() {
6385+ warn_null_comparison(a, b);
6386+ return 0;
6387+ }
62446388 !compare_gt(a, b) as i32
62456389}
62466390
62476391#[unsafe(no_mangle)]
62486392pub extern "C-unwind" fn knot_value_ge_i32(a: *mut Value, b: *mut Value) -> i32 {
6249− if a.is_null() || b.is_null() { return 0; }
6393+ if a.is_null() || b.is_null() {
6394+ warn_null_comparison(a, b);
6395+ return 0;
6396+ }
62506397 !compare_lt(a, b) as i32
62516398}
62526399
@@ -6980,8 +7127,17 @@ pub extern "C-unwind" fn knot_fork_io(io_val: *mut Value) -> *mut Value {
69807127 }
69817128 let _guard = CleanupGuard { db, io };
69827129
6983− // Run the IO action
6984− knot_io_run(db, io);
7130+ // Run the IO action under catch_unwind so a panicking body (e.g.
7131+ // inside an `atomic` block, which holds the global write lock
7132+ // across begin/commit) releases the write lock instead of leaking
7133+ // it and deadlocking every other writer in the process. The panic
7134+ // hook has already reported the panic; the thread then exits.
7135+ let run = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
7136+ knot_io_run(db, io);
7137+ }));
7138+ if run.is_err() {
7139+ write_lock_force_release();
7140+ }
69857141 });
69867142
69877143 alloc(Value::Unit)
@@ -7097,6 +7253,16 @@ pub extern "C-unwind" fn knot_race_io(io_a: *mut Value, io_b: *mut Value) -> *mu
70977253 let result = match run {
70987254 Ok(v) => v,
70997255 Err(payload) => {
7256+ // Release any write locks the unwinding body may have
7257+ // held so other threads aren't deadlocked (no-op when
7258+ // none are held). This MUST run for `Cancelled`
7259+ // unwinds too: a loser cancelled inside an `atomic`
7260+ // block still holds the global write lock acquired by
7261+ // `knot_atomic_begin` (the guard was `mem::forget`-en)
7262+ // — skipping the release would deadlock every writer
7263+ // in the process. The connection (and its open
7264+ // savepoints) is discarded by `CleanupGuard` below.
7265+ write_lock_force_release();
71007266 // Cooperative cancellation surfaces as a `Cancelled`
71017267 // panic raised by `knot_io_run`. It is not an error:
71027268 // the peer already produced the outcome, so this side
@@ -7104,10 +7270,6 @@ pub extern "C-unwind" fn knot_race_io(io_a: *mut Value, io_b: *mut Value) -> *mu
71047270 if payload.downcast_ref::<Cancelled>().is_some() {
71057271 return;
71067272 }
7107− // Release any write locks the panicking body may have
7108− // held so other threads aren't deadlocked (no-op when
7109− // none are held).
7110− write_lock_force_release();
71117273 let msg = if let Some(s) = payload.downcast_ref::<&str>() {
71127274 s.to_string()
71137275 } else if let Some(s) = payload.downcast_ref::<String>() {
@@ -7466,13 +7628,37 @@ pub extern "C-unwind" fn knot_stm_wait(_snapshot: i64) {
74667628 bucket.entry(threshold.to_key()).or_default().push(weak);
74677629 }
74687630
7631+ // Cancellation-awareness: register the slot with this thread's
7632+ // CancelToken (if it is a `race` worker) BEFORE checking `is_cancelled`
7633+ // below — a `cancel()` arriving at any point either finds the registered
7634+ // slot (and wakes it) or is observed by the check; no ordering window
7635+ // loses the wakeup.
7636+ let cancel_token = current_cancel_token();
7637+ if let Some(tok) = &cancel_token {
7638+ tok.set_stm_wake_slot(&slot);
7639+ }
7640+
7641+ // TOCTOU guard (Dekker pattern), paired with the fence in
7642+ // `bump_table_version`: order this thread's registration stores
7643+ // (watcher counts, TABLE_FILTER_COLS, watcher indices) before the
7644+ // version re-check load below. Together the fences guarantee that a
7645+ // concurrent writer either sees our registration when it projects /
7646+ // notifies, or we see its version bump here — a wakeup cannot be lost
7647+ // in both directions.
7648+ std::sync::atomic::fence(Ordering::SeqCst);
7649+
74697650 // Re-check after registration to prevent lost wakeups. Uses cached Arcs —
74707651 // one atomic load per table, no DashMap traffic.
74717652 let changed_after_register = read_versions
74727653 .iter()
74737654 .any(|(_, arc, ver)| arc.load(Ordering::Acquire) > *ver);
74747655
7475− if !changed_after_register {
7656+ let cancelled = cancel_token
7657+ .as_ref()
7658+ .map(|t| t.is_cancelled())
7659+ .unwrap_or(false);
7660+
7661+ if !changed_after_register && !cancelled {
74767662 slot.wait(STM_WAIT_TIMEOUT);
74777663 // slot drops → Weak refs become invalid, cleaned up lazily in notify
74787664 } else {
@@ -7481,6 +7667,10 @@ pub extern "C-unwind" fn knot_stm_wait(_snapshot: i64) {
74817667 std::thread::yield_now();
74827668 }
74837669
7670+ if let Some(tok) = &cancel_token {
7671+ tok.clear_stm_wake_slot();
7672+ }
7673+
74847674 // Re-acquire write lock if we held it before waiting
74857675 write_lock_reacquire_at_depth(saved_lock_depth);
74867676}
@@ -9297,10 +9487,13 @@ enum ColType {
92979487 Bytes,
92989488 /// Stored as TEXT, reconstructed as Constructor on read
92999489 Tag,
9300− /// Nested relation stored as JSON text in SQLite. In practice this variant
9301− /// never appears in `RecordSchema::columns`/`NestedField::columns` because
9302− /// `parse_record_schema` routes `[...]` types into the `nested` channel
9303− /// instead — but it remains for ADT field schemas (`parse_adt_schema`).
9490+ /// Value stored as JSON text in SQLite. Used for:
9491+ /// - record-valued fields (`{x: Float, y: Float}` inside a row),
9492+ /// - payload-bearing ADT fields,
9493+ /// - nested relations of *non-record* element type (e.g. `tags: [Text]`,
9494+ /// `shapes: [Shape]`) — the whole relation is serialized as one JSON
9495+ /// column (descriptor `field:json`); only record-element nested
9496+ /// relations take the `field:[child_schema]` child-table form.
93049497 Json,
93059498}
93069499
@@ -9362,6 +9555,10 @@ fn parse_adt_schema(spec: &str) -> AdtSpec {
93629555 "bool" => ColType::Bool,
93639556 "bytes" => ColType::Bytes,
93649557 "tag" => ColType::Tag,
9558+ // Record / payload-ADT constructor fields round-trip
9559+ // through JSON (the compiler's `col_type_str` emits
9560+ // "json" for them).
9561+ "json" => ColType::Json,
93659562 s if s.starts_with('[') => ColType::Json,
93669563 other => panic!("knot runtime: unknown ADT field type '{}'", other),
93679564 };
@@ -9568,7 +9765,13 @@ fn init_record_table(conn: &rusqlite::Connection, table_name: &str, schema: &Rec
95689765
95699766 for c in &schema.columns {
95709767 col_defs.push(format!("{} {}", quote_ident(&c.name), sql_type(c.ty)));
9571− unique_cols.push(quote_ident(&c.name));
9768+ // NULL-coalesced so rows containing NULLs still dedupe under
9769+ // INSERT OR IGNORE (SQLite treats raw NULLs as distinct in UNIQUE
9770+ // indexes) — same sentinel scheme as the ADT path. Existing DBs
9771+ // keep their old raw-column index under the same name (IF NOT
9772+ // EXISTS no-ops), matching how the ADT init path handles upgrades;
9773+ // schema migrations (`auto_apply_record_change`) drop + recreate.
9774+ unique_cols.push(null_safe_coalesce("e_ident(&c.name), c.ty));
95729775 }
95739776
95749777 if col_defs.is_empty() {
@@ -9613,7 +9816,10 @@ fn init_child_table(conn: &rusqlite::Connection, parent_table: &str, nf: &Nested
96139816
96149817 for c in &nf.columns {
96159818 col_defs.push(format!("{} {}", quote_ident(&c.name), sql_type(c.ty)));
9616− unique_cols.push(quote_ident(&c.name));
9819+ // NULL-coalesced for set semantics with NULL-bearing rows — see
9820+ // the matching comment in `init_record_table`. `_parent_id` stays
9821+ // raw (NOT NULL by definition).
9822+ unique_cols.push(null_safe_coalesce("e_ident(&c.name), c.ty));
96179823 }
96189824
96199825 let sql = format!("CREATE TABLE IF NOT EXISTS {} ({});", child_table, col_defs.join(", "));
@@ -9795,7 +10001,8 @@ fn auto_apply_child_change(
979510001
979610002 let mut unique_cols = vec![quote_ident("_parent_id")];
979710003 for c in &new_nf.columns {
9798− unique_cols.push(quote_ident(&c.name));
10004+ // NULL-coalesced — same scheme as `init_child_table`.
10005+ unique_cols.push(null_safe_coalesce("e_ident(&c.name), c.ty));
979910006 }
980010007 if unique_cols.len() > 1 {
980110008 let idx_sql = format!(
@@ -9890,7 +10097,12 @@ fn auto_apply_record_change(
989010097 debug_sql(&drop_idx);
989110098 let _ = conn.execute_batch(&drop_idx);
989210099
9893− let unique_cols: Vec<String> = new_rec.columns.iter().map(|c| quote_ident(&c.name)).collect();
10100+ // NULL-coalesced — same scheme as `init_record_table` / the ADT path.
10101+ let unique_cols: Vec<String> = new_rec
10102+ .columns
10103+ .iter()
10104+ .map(|c| null_safe_coalesce("e_ident(&c.name), c.ty))
10105+ .collect();
989410106 if !unique_cols.is_empty() {
989510107 let idx_sql = format!(
989610108 "CREATE UNIQUE INDEX IF NOT EXISTS {} ON {} ({});",
@@ -11997,8 +12209,12 @@ fn value_to_sqlite(v: *mut Value, ty: ColType) -> rusqlite::types::Value {
1199712209 rusqlite::types::Value::Text(value_to_json(v))
1199812210 }
1199912211 (Value::Constructor(tag, _), _) => rusqlite::types::Value::Text(tag.to_string()),
12000− (Value::Relation(_), ColType::Json) => {
12001− rusqlite::types::Value::Text(value_to_json(v))
12212+ (Value::Relation(rows), ColType::Json) => {
12213+ // Nested relations stored as JSON columns keep set semantics:
12214+ // dedup rows before serializing, mirroring the INSERT OR IGNORE
12215+ // dedup that child-table storage gets from its UNIQUE index.
12216+ let deduped = in_memory_dedup(rows.clone());
12217+ rusqlite::types::Value::Text(value_to_json(alloc(Value::Relation(deduped))))
1200212218 }
1200312219 (Value::Record(_), ColType::Json) => {
1200412220 rusqlite::types::Value::Text(value_to_json(v))
@@ -12435,15 +12651,39 @@ fn check_rate_limit(
1243512651 }
1243612652 ensure_rate_limit_table(db);
1243712653 let now = current_unix_ms();
12438− let _ = db.conn.execute_batch("BEGIN IMMEDIATE;");
12439− let row: Option<(f64, i64)> = db
12440− .conn
12441− .query_row(
12442− "SELECT tokens, last_refill FROM _knot_rate_limits WHERE route = ?1 AND key = ?2",
12443− rusqlite::params![route, key],
12444− |r| Ok((r.get::<_, f64>(0)?, r.get::<_, i64>(1)?)),
12445− )
12446− .ok();
12654+ // BEGIN IMMEDIATE serializes concurrent same-key checks. If it fails
12655+ // (e.g. SQLITE_BUSY under contention), fail OPEN without touching the
12656+ // bucket: allowing a request through under transient DB pressure beats
12657+ // 429ing legitimate traffic, and skipping the accounting write keeps
12658+ // the stored token counts intact.
12659+ if let Err(e) = db.conn.execute_batch("BEGIN IMMEDIATE;") {
12660+ log_warn!(
12661+ "[HTTP] rate limiter: BEGIN failed ({}); allowing request without accounting",
12662+ e
12663+ );
12664+ return Ok(());
12665+ }
12666+ let row: Option<(f64, i64)> = match db.conn.query_row(
12667+ "SELECT tokens, last_refill FROM _knot_rate_limits WHERE route = ?1 AND key = ?2",
12668+ rusqlite::params![route, key],
12669+ |r| Ok((r.get::<_, f64>(0)?, r.get::<_, i64>(1)?)),
12670+ ) {
12671+ Ok(r) => Some(r),
12672+ // Genuinely no bucket yet — start a fresh one at full capacity.
12673+ Err(rusqlite::Error::QueryReturnedNoRows) => None,
12674+ // Real error (I/O, corruption, busy): do NOT conflate with "no
12675+ // bucket" — that would reset the bucket to full capacity and let
12676+ // an attacker leverage DB pressure to bypass the limit. Fail open
12677+ // for this one request and leave the stored bucket untouched.
12678+ Err(e) => {
12679+ log_warn!(
12680+ "[HTTP] rate limiter: bucket read failed ({}); allowing request without accounting",
12681+ e
12682+ );
12683+ let _ = db.conn.execute_batch("ROLLBACK;");
12684+ return Ok(());
12685+ }
12686+ };
1244712687 let capacity = requests as f64;
1244812688 let tokens_before = match row {
1244912689 Some((tokens, last_refill)) => {
@@ -12453,23 +12693,27 @@ fn check_rate_limit(
1245312693 }
1245412694 None => capacity,
1245512695 };
12456− if tokens_before >= 1.0 {
12457− let new_tokens = tokens_before - 1.0;
12458− let _ = db.conn.execute(
12696+ // Persist the updated bucket. Write failures are logged and rolled
12697+ // back; the allow/deny decision (computed from the read above) stands.
12698+ let write_bucket = |tokens: f64| {
12699+ let write = db.conn.execute(
1245912700 "INSERT OR REPLACE INTO _knot_rate_limits (route, key, tokens, last_refill) VALUES (?1, ?2, ?3, ?4)",
12460− rusqlite::params![route, key, new_tokens, now],
12701+ rusqlite::params![route, key, tokens, now],
1246112702 );
12462− let _ = db.conn.execute_batch("COMMIT;");
12703+ let result = write.and_then(|_| db.conn.execute_batch("COMMIT;"));
12704+ if let Err(e) = result {
12705+ log_warn!("[HTTP] rate limiter: bucket write failed ({})", e);
12706+ let _ = db.conn.execute_batch("ROLLBACK;");
12707+ }
12708+ };
12709+ if tokens_before >= 1.0 {
12710+ write_bucket(tokens_before - 1.0);
1246312711 Ok(())
1246412712 } else {
1246512713 let needed = 1.0 - tokens_before;
1246612714 let retry_after_ms =
1246712715 (needed * (window_ms as f64) / capacity).ceil() as i64;
12468− let _ = db.conn.execute(
12469− "INSERT OR REPLACE INTO _knot_rate_limits (route, key, tokens, last_refill) VALUES (?1, ?2, ?3, ?4)",
12470− rusqlite::params![route, key, tokens_before, now],
12471− );
12472− let _ = db.conn.execute_batch("COMMIT;");
12716+ write_bucket(tokens_before);
1247312717 Err(retry_after_ms.max(1))
1247412718 }
1247512719}
@@ -14114,7 +14358,8 @@ fn http_serve_loop(
1411414358 use std::io::Read;
1411514359 let mut buf = Vec::new();
1411614360 let max = http_max_body_bytes();
14117− let mut limited = request.as_reader().take(max + 1);
14361+ // saturating: `max` may be u64::MAX (env override).
14362+ let mut limited = request.as_reader().take(max.saturating_add(1));
1411814363 let _ = limited.read_to_end(&mut buf);
1411914364 if buf.len() as u64 > max {
1412014365 eprintln!(
@@ -14150,13 +14395,20 @@ fn http_serve_loop(
1415014395 _ => None,
1415114396 })
1415214397 .unwrap_or("text");
14153− let value = match try_string_to_value(val, ty) {
14398+ // `?`-typed (Maybe) path params: a matched path
14399+ // segment is always present, so absence (Nothing)
14400+ // can't occur — but the handler's declared type is
14401+ // `Maybe T`, so the value must be Just-wrapped.
14402+ let is_maybe = ty.starts_with('?');
14403+ let inner_ty = if is_maybe { &ty[1..] } else { ty };
14404+ let value = match try_string_to_value(val, inner_ty) {
1415414405 Some(v) => v,
1415514406 None => return Err((400, format!(
1415614407 "invalid path parameter '{}': '{}' is not a valid {}",
14157− name, val, ty
14408+ name, val, inner_ty
1415814409 ), None)),
1415914410 };
14411+ let value = if is_maybe { make_just(value) } else { value };
1416014412 fields.push(RecordField {
1416114413 name: intern_str(name),
1416214414 value,
@@ -14248,7 +14500,15 @@ fn http_serve_loop(
1424814500 } else {
1424914501 match raw_val {
1425014502 Some(v) => coerce_json_field(v, inner_ty),
14251− None => string_to_value("", inner_ty),
14503+ // A missing required field is a
14504+ // client error — defaulting to
14505+ // 0/""/false silently corrupts
14506+ // data (path/query/headers
14507+ // already 400 on bad input).
14508+ None => return Err((400, format!(
14509+ "missing required body field '{}'",
14510+ bname
14511+ ), None)),
1425214512 }
1425314513 };
1425414514 fields.push(RecordField {
@@ -14258,12 +14518,17 @@ fn http_serve_loop(
1425814518 }
1425914519 }
1426014520 _ => {
14521+ // Body parsed but isn't a JSON object: every
14522+ // required (non-Maybe) field is missing.
1426114523 for (bname, bty) in &entry_body_fields {
1426214524 let is_maybe = bty.starts_with('?');
1426314525 let value = if is_maybe {
1426414526 alloc(Value::Constructor("Nothing".into(), alloc(Value::Unit)))
1426514527 } else {
14266− string_to_value("", bty)
14528+ return Err((400, format!(
14529+ "missing required body field '{}' (request body must be a JSON object)",
14530+ bname
14531+ ), None));
1426714532 };
1426814533 fields.push(RecordField {
1426914534 name: intern_str(bname),
@@ -14455,7 +14720,29 @@ fn http_serve_loop(
1445514720 if let Value::Record(hdr_fields) = unsafe { as_ref(hdrs_val) } {
1445614721 for hf in hdr_fields {
1445714722 let http_name = camel_to_header_case(&hf.name);
14458− let hdr_value = fetch_value_to_text(hf.value);
14723+ // This loop runs OUTSIDE the worker's
14724+ // catch_unwind — a conversion panic here
14725+ // would kill the response path entirely.
14726+ // Catch it and drop the offending header
14727+ // (the body still goes out); `Nothing`
14728+ // header values are omitted, mirroring
14729+ // how fetch skips Nothing request headers.
14730+ let converted = std::panic::catch_unwind(
14731+ std::panic::AssertUnwindSafe(|| {
14732+ fetch_value_to_text_opt(hf.value)
14733+ }),
14734+ );
14735+ let hdr_value = match converted {
14736+ Ok(Some(s)) => s,
14737+ Ok(None) => continue, // Nothing → omit header
14738+ Err(_) => {
14739+ log_warn!(
14740+ "[HTTP] dropping response header '{}': value not convertible to text",
14741+ http_name
14742+ );
14743+ continue;
14744+ }
14745+ };
1445914746 if let Ok(header) = format!("{}: {}", http_name, hdr_value)
1446014747 .parse::<tiny_http::Header>()
1446114748 {
@@ -14708,25 +14995,18 @@ pub extern "C-unwind" fn knot_http_fetch_io(
1470814995 if !req_hdrs_str.is_empty() {
1470914996 for field_desc in req_hdrs_str.split(',') {
1471014997 if field_desc.is_empty() { continue; }
14711− let (name, ty) = field_desc.split_once(':').unwrap_or((field_desc, "text"));
14712− let is_maybe = ty.starts_with('?');
14998+ let (name, _ty) = field_desc.split_once(':').unwrap_or((field_desc, "text"));
1471314999 let http_name = camel_to_header_case(name);
1471415000 if http_name.eq_ignore_ascii_case("Content-Type") {
1471515001 has_content_type = true;
1471615002 }
1471715003 let field_val = knot_record_field(payload, name.as_ptr(), name.len());
14718− if is_maybe {
14719− // Maybe type: skip Nothing, extract Just value
14720− if !field_val.is_null() {
14721− if let Value::Constructor(tag, inner) = unsafe { as_ref(field_val) } {
14722− if &**tag == "Just" {
14723− let v = knot_record_field(*inner, "value".as_ptr(), 5);
14724− req_headers.push((http_name, fetch_value_to_text(v)));
14725− }
14726− }
14727− }
14728− } else {
14729− req_headers.push((http_name, fetch_value_to_text(field_val)));
15004+ // `fetch_value_to_text_opt` unwraps Just and returns None
15005+ // for Nothing — covers both Maybe (`?`-typed) and plain
15006+ // header fields (a `None` on a non-Maybe field can't occur
15007+ // by typing, and omitting it is safe regardless).
15008+ if let Some(text) = fetch_value_to_text_opt(field_val) {
15009+ req_headers.push((http_name, text));
1473015010 }
1473115011 }
1473215012 }
@@ -14946,7 +15226,8 @@ fn fetch_read_capped_body(body: &mut ureq::Body) -> Result<String, String> {
1494615226 let max = http_max_body_bytes();
1494715227 let mut buf = Vec::new();
1494815228 body.as_reader()
14949− .take(max + 1)
15229+ // saturating: `max` may be u64::MAX (env override).
15230+ .take(max.saturating_add(1))
1495015231 .read_to_end(&mut buf)
1495115232 .map_err(|e| format!("response read error: {}", e))?;
1495215233 if buf.len() as u64 > max {
@@ -14973,7 +15254,15 @@ fn fetch_build_url(base: &str, path_pattern: &str, payload: *mut Value) -> Strin
1497315254 let param = &remaining[start + 1..end];
1497415255 let (name, _ty) = param.split_once(':').unwrap_or((param, "text"));
1497515256 let field_val = knot_record_field(payload, name.as_ptr(), name.len());
14976− url.push_str(&percent_encode(&fetch_value_to_text(field_val)));
15257+ let text = fetch_value_to_text_opt(field_val).unwrap_or_else(|| {
15258+ // Path segments cannot be omitted — a Nothing here is a caller
15259+ // bug the type checker should have prevented.
15260+ panic!(
15261+ "knot runtime: path parameter '{}' is Nothing — path parameters cannot be omitted",
15262+ name
15263+ )
15264+ });
15265+ url.push_str(&percent_encode(&text));
1497715266 remaining = &remaining[end + 1..];
1497815267 }
1497915268 url.push_str(remaining);
@@ -15013,35 +15302,62 @@ fn fetch_build_query(query_desc: &str, payload: *mut Value) -> String {
1501315302 if field_desc.is_empty() {
1501415303 continue;
1501515304 }
15016− let (name, ty) = field_desc.split_once(':').unwrap_or((field_desc, "text"));
15017− let is_maybe = ty.starts_with('?');
15305+ let (name, _ty) = field_desc.split_once(':').unwrap_or((field_desc, "text"));
1501815306 let field_val = knot_record_field(payload, name.as_ptr(), name.len());
15019− let val = if is_maybe {
15020− match unwrap_maybe(field_val) {
15021− Some(inner) => inner,
15022− None => continue, // Skip Nothing query params
15023− }
15024− } else {
15025− field_val
15307+ // `fetch_value_to_text_opt` unwraps Just and returns None for
15308+ // Nothing — Nothing query params are omitted from the URL.
15309+ let val_str = match fetch_value_to_text_opt(field_val) {
15310+ Some(s) => s,
15311+ None => continue,
1502615312 };
15027− let val_str = fetch_value_to_text(val);
1502815313 parts.push(format!("{}={}", name, percent_encode(&val_str)));
1502915314 }
1503015315 parts.join("&")
1503115316}
1503215317
15033−/// Convert a Knot value to its text representation for URL params.
15034−fn fetch_value_to_text(v: *mut Value) -> String {
15318+/// Convert a Knot value to its text representation for URL params and
15319+/// HTTP headers. Returns `None` for `Nothing` — callers must OMIT the
15320+/// parameter/header entirely (mirroring how fetch skips `Nothing` request
15321+/// headers). `Just {value: v}` unwraps and recurses on `v`; nullary
15322+/// constructors (Unit or empty-record payload) render as their bare tag,
15323+/// matching the server-side `tag` column decode for enum-like ADTs.
15324+fn fetch_value_to_text_opt(v: *mut Value) -> Option<String> {
1503515325 if v.is_null() {
15036− return String::new();
15326+ return Some(String::new());
1503715327 }
1503815328 match unsafe { as_ref(v) } {
15039− Value::Int(n) => n.to_string(),
15040− Value::Float(n) => n.to_string(),
15041− Value::Text(s) => (**s).to_string(),
15042− Value::Bool(b) => b.to_string(),
15329+ Value::Int(n) => Some(n.to_string()),
15330+ Value::Float(n) => Some(n.to_string()),
15331+ Value::Text(s) => Some((**s).to_string()),
15332+ Value::Bool(b) => Some(b.to_string()),
15333+ Value::Constructor(tag, payload) => {
15334+ if &**tag == "Nothing" {
15335+ return None;
15336+ }
15337+ if &**tag == "Just" {
15338+ return fetch_value_to_text_opt(knot_record_field(
15339+ *payload,
15340+ "value".as_ptr(),
15341+ 5,
15342+ ));
15343+ }
15344+ let nullary = (*payload).is_null()
15345+ || match unsafe { as_ref(*payload) } {
15346+ Value::Unit => true,
15347+ Value::Record(fs) => fs.is_empty(),
15348+ _ => false,
15349+ };
15350+ if nullary {
15351+ Some(tag.to_string())
15352+ } else {
15353+ panic!(
15354+ "knot runtime: cannot convert constructor '{}' (with fields) to text for URL parameter or header",
15355+ tag
15356+ )
15357+ }
15358+ }
1504315359 _ => panic!(
15044− "knot runtime: cannot convert {} to text for URL parameter",
15360+ "knot runtime: cannot convert {} to text for URL parameter or header",
1504515361 type_name(v)
1504615362 ),
1504715363 }
@@ -15207,6 +15523,14 @@ fn generate_openapi(name: &str, table: &RouteTable) -> String {
1520715523
1520815524 for (i, (path, entries)) in paths.iter().enumerate() {
1520915525 out.push_str(&format!(" \"{}\": {{\n", json_escape(path)));
15526+ // Methods are JSON keys within a path object — two route entries
15527+ // sharing method+path would emit duplicate keys (invalid JSON).
15528+ // Keep the first entry per method.
15529+ let mut seen_methods: HashSet<String> = HashSet::new();
15530+ let entries: Vec<&&RouteTableEntry> = entries
15531+ .iter()
15532+ .filter(|e| seen_methods.insert(e.method.to_lowercase()))
15533+ .collect();
1521015534 for (j, entry) in entries.iter().enumerate() {
1521115535 let method = entry.method.to_lowercase();
1521215536 out.push_str(&format!(" \"{}\": {{\n", method));
@@ -15227,9 +15551,12 @@ fn generate_openapi(name: &str, table: &RouteTable) -> String {
1522715551 }
1522815552 }
1522915553 for (qname, qty) in &entry.query_fields {
15554+ // `?`-typed (Maybe) query params are optional; all others
15555+ // are required (the server 400s when they're absent).
1523015556 params.push(format!(
15231− "{{ \"name\": \"{}\", \"in\": \"query\", \"required\": false, \"schema\": {} }}",
15557+ "{{ \"name\": \"{}\", \"in\": \"query\", \"required\": {}, \"schema\": {} }}",
1523215558 json_escape(qname),
15559+ !qty.starts_with('?'),
1523315560 type_to_openapi_schema(qty)
1523415561 ));
1523515562 }
@@ -15274,14 +15601,18 @@ fn generate_openapi(name: &str, table: &RouteTable) -> String {
1527415601 out.push('\n');
1527515602 }
1527615603 out.push_str(" },\n");
15277− out.push_str(" \"required\": [");
15278− for (k, (fname, _)) in entry.body_fields.iter().enumerate() {
15279− out.push_str(&format!("\"{}\"", json_escape(fname)));
15280− if k + 1 < entry.body_fields.len() {
15281− out.push_str(", ");
15282− }
15283− }
15284− out.push_str("]\n");
15604+ // `?`-typed (Maybe) body fields are optional — exclude them
15605+ // from the `required` array.
15606+ let required: Vec<String> = entry
15607+ .body_fields
15608+ .iter()
15609+ .filter(|(_, fty)| !fty.starts_with('?'))
15610+ .map(|(fname, _)| format!("\"{}\"", json_escape(fname)))
15611+ .collect();
15612+ out.push_str(&format!(
15613+ " \"required\": [{}]\n",
15614+ required.join(", ")
15615+ ));
1528515616 out.push_str(" }\n");
1528615617 out.push_str(" }\n");
1528715618 out.push_str(" }\n");
@@ -15350,6 +15681,9 @@ fn openapi_path(parts: &[PathPart]) -> String {
1535015681}
1535115682
1535215683fn type_to_openapi_schema(ty: &str) -> &'static str {
15684+ // `?` marks Maybe-typed (optional) fields — optionality is expressed via
15685+ // the `required` flag/array, not the schema, so strip it before mapping.
15686+ let ty = ty.strip_prefix('?').unwrap_or(ty);
1535315687 match ty {
1535415688 "int" => "{ \"type\": \"integer\" }",
1535515689 "float" => "{ \"type\": \"number\" }",
@@ -15507,7 +15841,10 @@ fn serialize_value_for_hash_into(v: *mut Value, buf: &mut Vec<u8>) {
1550715841 }
1550815842 Value::Relation(rows) => {
1550915843 buf.push(8);
15510− buf.extend_from_slice(&(rows.len() as u32).to_le_bytes());
15844+ // Canonical *set* form: sort + dedup, then write the DEDUPED
15845+ // count — must match `value_to_hash_bytes` exactly, otherwise a
15846+ // relation key containing duplicate rows hashes differently
15847+ // across the two paths and equal join keys miss each other.
1551115848 let mut row_bytes: Vec<Vec<u8>> = rows
1551215849 .iter()
1551315850 .map(|r| {
@@ -15517,6 +15854,8 @@ fn serialize_value_for_hash_into(v: *mut Value, buf: &mut Vec<u8>) {
1551715854 })
1551815855 .collect();
1551915856 row_bytes.sort_unstable();
15857+ row_bytes.dedup();
15858+ buf.extend_from_slice(&(row_bytes.len() as u32).to_le_bytes());
1552015859 for rb in &row_bytes {
1552115860 buf.extend_from_slice(&(rb.len() as u32).to_le_bytes());
1552215861 buf.extend_from_slice(rb);
@@ -15636,7 +15975,30 @@ pub extern "C-unwind" fn knot_crypto_generate_signing_key_pair() -> *mut Value {
1563615975 record
1563715976}
1563815977
15639−/// Sealed-box encryption: X25519 ECDH + ChaCha20-Poly1305.
15978+/// Derive the sealed-box AEAD key: BLAKE3 `derive_key` over
15979+/// `shared_secret || ephemeral_pub || recipient_pub` with a fixed,
15980+/// domain-separating context string. Using a KDF instead of the raw ECDH
15981+/// output (a) binds both public keys into the key, preventing cross-keypair
15982+/// ciphertext malleability, and (b) hides any structure of the raw curve
15983+/// point from the cipher.
15984+///
15985+/// NOTE: ciphertexts produced by builds that used the raw shared secret as
15986+/// the key cannot be decrypted by current builds (accepted pre-1.0 format
15987+/// change).
15988+fn derive_sealed_box_key(
15989+ shared: &[u8; 32],
15990+ eph_pub: &[u8; 32],
15991+ recipient_pub: &[u8; 32],
15992+) -> [u8; 32] {
15993+ const CONTEXT: &str = "knot-runtime sealed-box v2 ChaCha20-Poly1305 key";
15994+ let mut material = [0u8; 96];
15995+ material[..32].copy_from_slice(shared);
15996+ material[32..64].copy_from_slice(eph_pub);
15997+ material[64..].copy_from_slice(recipient_pub);
15998+ blake3::derive_key(CONTEXT, &material)
15999+}
16000+
16001+/// Sealed-box encryption: X25519 ECDH + BLAKE3 KDF + ChaCha20-Poly1305.
1564016002/// Takes (publicKey: Bytes, plaintext: Bytes), returns ciphertext Bytes.
1564116003/// Format: [ephemeral_pub: 32][nonce: 12][encrypted + tag: len+16]
1564216004#[unsafe(no_mangle)]
@@ -15663,9 +16025,15 @@ pub extern "C-unwind" fn knot_crypto_encrypt(public_key: *mut Value, plaintext:
1566316025 let eph_secret = x25519_dalek::StaticSecret::from(eph_secret_bytes);
1566416026 let eph_public = x25519_dalek::PublicKey::from(&eph_secret);
1566516027
15666− // ECDH shared secret
16028+ // ECDH shared secret. Reject non-contributory exchanges (low-order /
16029+ // identity points): an attacker-chosen public key would otherwise force
16030+ // a predictable all-zero secret.
1566716031 let shared = eph_secret.diffie_hellman(&recipient_public);
15668− let key = chacha20poly1305::Key::from_slice(shared.as_bytes());
16032+ if !shared.was_contributory() {
16033+ panic!("knot runtime: encrypt publicKey is a low-order point (non-contributory X25519 exchange)");
16034+ }
16035+ let key_bytes = derive_sealed_box_key(shared.as_bytes(), eph_public.as_bytes(), &recipient_pub);
16036+ let key = chacha20poly1305::Key::from_slice(&key_bytes);
1566916037 let cipher = ChaCha20Poly1305::new(key);
1567016038
1567116039 // Random nonce
@@ -15714,8 +16082,19 @@ pub extern "C-unwind" fn knot_crypto_decrypt(private_key: *mut Value, ciphertext
1571416082 let encrypted = &ct[44..];
1571516083
1571616084 let eph_public = x25519_dalek::PublicKey::from(eph_pub_bytes);
16085+ // Reject non-contributory exchanges — a low-order ephemeral point in the
16086+ // ciphertext header would force a predictable key (see encrypt).
1571716087 let shared = secret.diffie_hellman(&eph_public);
15718− let key = chacha20poly1305::Key::from_slice(shared.as_bytes());
16088+ if !shared.was_contributory() {
16089+ panic!("knot runtime: decryption failed (non-contributory X25519 exchange — low-order ephemeral point)");
16090+ }
16091+ let recipient_public = x25519_dalek::PublicKey::from(&secret);
16092+ let key_bytes = derive_sealed_box_key(
16093+ shared.as_bytes(),
16094+ &eph_pub_bytes,
16095+ recipient_public.as_bytes(),
16096+ );
16097+ let key = chacha20poly1305::Key::from_slice(&key_bytes);
1571916098 let cipher = ChaCha20Poly1305::new(key);
1572016099 let nonce = chacha20poly1305::Nonce::from_slice(&nonce_bytes);
1572116100
@@ -16627,3 +17006,361 @@ mod _regress_runtime_fixes_tests {
1662717006 assert!(!values_equal(r1, r2));
1662817007 }
1662917008}
17009+
17010+#[cfg(test)]
17011+mod _bugfix_batch_tests {
17012+ use super::*;
17013+
17014+ fn int(n: i64) -> *mut Value {
17015+ alloc(Value::Int(n))
17016+ }
17017+
17018+ fn text(s: &str) -> *mut Value {
17019+ alloc(Value::Text(Arc::from(s)))
17020+ }
17021+
17022+ fn rec(fields: &[(&str, *mut Value)]) -> *mut Value {
17023+ let mut fs: Vec<RecordField> = fields
17024+ .iter()
17025+ .map(|(n, v)| RecordField {
17026+ name: intern_str(n),
17027+ value: *v,
17028+ })
17029+ .collect();
17030+ fs.sort_by(|a, b| a.name.cmp(&b.name));
17031+ alloc(Value::Record(fs))
17032+ }
17033+
17034+ fn nothing() -> *mut Value {
17035+ // Unit-payload shape, as produced by the DB `tag` column decode —
17036+ // this is the shape `infer_col_type` classifies as a Tag column.
17037+ alloc(Value::Constructor(intern_str("Nothing"), alloc(Value::Unit)))
17038+ }
17039+
17040+ fn just(v: *mut Value) -> *mut Value {
17041+ make_just(v)
17042+ }
17043+
17044+ // ── Fix: sql_partial_cmp must compare numeric texts numerically ──
17045+
17046+ #[test]
17047+ fn sql_partial_cmp_numeric_texts_compare_numerically() {
17048+ use std::cmp::Ordering;
17049+ // "5" < "10" numerically (lexicographic order would say Greater).
17050+ assert_eq!(
17051+ sql_partial_cmp(&SqlVal::Text("5".into()), &SqlVal::Text("10".into())),
17052+ Some(Ordering::Less)
17053+ );
17054+ assert_eq!(
17055+ sql_partial_cmp(&SqlVal::Text("100".into()), &SqlVal::Text("99".into())),
17056+ Some(Ordering::Greater)
17057+ );
17058+ // Float-as-text compares numerically too.
17059+ assert_eq!(
17060+ sql_partial_cmp(&SqlVal::Text("2.5".into()), &SqlVal::Text("10".into())),
17061+ Some(Ordering::Less)
17062+ );
17063+ // Non-numeric texts keep lexicographic order.
17064+ assert_eq!(
17065+ sql_partial_cmp(&SqlVal::Text("apple".into()), &SqlVal::Text("banana".into())),
17066+ Some(Ordering::Less)
17067+ );
17068+ // Mixed numeric / non-numeric is unknown → None (wake conservatively).
17069+ assert_eq!(
17070+ sql_partial_cmp(&SqlVal::Text("5".into()), &SqlVal::Text("apple".into())),
17071+ None
17072+ );
17073+ }
17074+
17075+ #[test]
17076+ fn stm_ordered_pred_on_int_as_text_row() {
17077+ // STM watcher on `qty < 10`: write payload carries Int-as-TEXT "5"
17078+ // (SqlVal::from_knot serializes Int that way) — must match.
17079+ let pred = ColPred::Cmp(CmpOp::Lt, SqlVal::Text("10".into()));
17080+ assert!(col_pred_matches(&SqlVal::Text("5".into()), &pred));
17081+ assert!(!col_pred_matches(&SqlVal::Text("50".into()), &pred));
17082+ }
17083+
17084+ // ── Fix: infer_temp_schema must scan all rows ──
17085+
17086+ #[test]
17087+ fn temp_schema_mixed_maybe_returns_none() {
17088+ // First row `Nothing` (nullary → Tag column), later row carries a
17089+ // `Just {value: 1}` payload — a Tag column would silently drop it.
17090+ let rows = vec![
17091+ rec(&[("m", nothing()), ("x", int(1))]),
17092+ rec(&[("m", just(int(2))), ("x", int(3))]),
17093+ ];
17094+ assert!(infer_temp_schema(&rows).is_none());
17095+ // Same in the opposite order.
17096+ let rows_rev = vec![
17097+ rec(&[("m", just(int(2))), ("x", int(3))]),
17098+ rec(&[("m", nothing()), ("x", int(1))]),
17099+ ];
17100+ assert!(infer_temp_schema(&rows_rev).is_none());
17101+ }
17102+
17103+ #[test]
17104+ fn temp_schema_null_then_typed_returns_none() {
17105+ // First-row SQL NULL used to type the column Text, mis-typing the
17106+ // later Int rows.
17107+ let rows = vec![
17108+ rec(&[("x", std::ptr::null_mut())]),
17109+ rec(&[("x", int(7))]),
17110+ ];
17111+ assert!(infer_temp_schema(&rows).is_none());
17112+ }
17113+
17114+ #[test]
17115+ fn temp_schema_conflicting_scalar_types_returns_none() {
17116+ let rows = vec![rec(&[("x", int(1))]), rec(&[("x", text("a"))])];
17117+ assert!(infer_temp_schema(&rows).is_none());
17118+ }
17119+
17120+ #[test]
17121+ fn temp_schema_consistent_rows_still_infer() {
17122+ let rows = vec![
17123+ rec(&[("name", text("a")), ("x", int(1))]),
17124+ rec(&[("name", text("b")), ("x", int(2))]),
17125+ ];
17126+ match infer_temp_schema(&rows) {
17127+ Some(TempSchema::Record(cols)) => {
17128+ assert_eq!(cols.len(), 2);
17129+ assert!(cols.contains(&("name".to_string(), ColType::Text)));
17130+ assert!(cols.contains(&("x".to_string(), ColType::Int)));
17131+ }
17132+ _ => panic!("expected Record schema"),
17133+ }
17134+ // All-nullary Maybe column stays a Tag column (consistent rows).
17135+ let tags = vec![rec(&[("m", nothing())]), rec(&[("m", nothing())])];
17136+ match infer_temp_schema(&tags) {
17137+ Some(TempSchema::Record(cols)) => {
17138+ assert_eq!(cols, vec![("m".to_string(), ColType::Tag)]);
17139+ }
17140+ _ => panic!("expected Record schema"),
17141+ }
17142+ }
17143+
17144+ #[test]
17145+ fn sql_dedup_falls_back_for_mixed_maybe_rows() {
17146+ // End-to-end through sql_dedup: mixed-arity Maybe rows must return
17147+ // None so the caller's in-memory dedup keeps all three distinct rows
17148+ // (the old Tag-typed column collapsed Just{1} and Just{2}).
17149+ let conn = Connection::open_in_memory().unwrap();
17150+ // `Nothing` FIRST: the old first-row-only inference typed the column
17151+ // Tag and `SELECT DISTINCT` collapsed the two Just rows.
17152+ let rows = vec![
17153+ rec(&[("m", nothing())]),
17154+ rec(&[("m", just(int(1)))]),
17155+ rec(&[("m", just(int(2)))]),
17156+ ];
17157+ assert!(sql_dedup(&conn, &rows).is_none());
17158+ assert_eq!(in_memory_dedup(rows).len(), 3);
17159+ }
17160+
17161+ // ── Fix: hash serializer divergence on duplicate relation rows ──
17162+
17163+ #[test]
17164+ fn hash_serializers_agree_on_relations_with_duplicates() {
17165+ let dup = alloc(Value::Relation(vec![
17166+ rec(&[("id", int(1))]),
17167+ rec(&[("id", int(1))]),
17168+ rec(&[("id", int(2))]),
17169+ ]));
17170+ let nodup = alloc(Value::Relation(vec![
17171+ rec(&[("id", int(2))]),
17172+ rec(&[("id", int(1))]),
17173+ ]));
17174+ assert!(values_equal(dup, nodup));
17175+
17176+ let mut a = Vec::new();
17177+ value_to_hash_bytes(dup, &mut a);
17178+ let mut b = Vec::new();
17179+ value_to_hash_bytes(nodup, &mut b);
17180+ assert_eq!(a, b, "value_to_hash_bytes must canonicalize duplicates");
17181+
17182+ // The join-index serializer must produce the SAME canonical form.
17183+ assert_eq!(
17184+ serialize_value_for_hash(dup),
17185+ serialize_value_for_hash(nodup),
17186+ "serialize_value_for_hash must dedup like value_to_hash_bytes"
17187+ );
17188+ assert_eq!(a, serialize_value_for_hash(dup), "the two paths must match");
17189+ }
17190+
17191+ // ── Fix: fetch_value_to_text constructors / Maybe ──
17192+
17193+ #[test]
17194+ fn fetch_text_opt_handles_maybe_and_enums() {
17195+ // Nothing → None (callers omit the param/header).
17196+ assert_eq!(fetch_value_to_text_opt(nothing()), None);
17197+ // Just {value: v} unwraps.
17198+ assert_eq!(fetch_value_to_text_opt(just(int(7))), Some("7".to_string()));
17199+ assert_eq!(
17200+ fetch_value_to_text_opt(just(text("hi"))),
17201+ Some("hi".to_string())
17202+ );
17203+ // Nullary constructor (enum ADT) renders as its tag — both the
17204+ // Unit-payload shape (DB tag decode) and the empty-record shape
17205+ // (user-written `Red {}`).
17206+ let unit_ctor = alloc(Value::Constructor(intern_str("Red"), alloc(Value::Unit)));
17207+ assert_eq!(fetch_value_to_text_opt(unit_ctor), Some("Red".to_string()));
17208+ let rec_ctor = alloc(Value::Constructor(
17209+ intern_str("Red"),
17210+ alloc(Value::Record(Vec::new())),
17211+ ));
17212+ assert_eq!(fetch_value_to_text_opt(rec_ctor), Some("Red".to_string()));
17213+ }
17214+
17215+ #[test]
17216+ fn fetch_build_query_omits_nothing_params() {
17217+ let payload = rec(&[("limit", just(int(5))), ("q", nothing())]);
17218+ let qs = fetch_build_query("q:?text,limit:?int", payload);
17219+ assert_eq!(qs, "limit=5");
17220+ }
17221+
17222+ // ── Fix: OpenAPI `?` (Maybe) type mapping ──
17223+
17224+ #[test]
17225+ fn openapi_schema_strips_maybe_prefix() {
17226+ assert_eq!(type_to_openapi_schema("?int"), "{ \"type\": \"integer\" }");
17227+ assert_eq!(type_to_openapi_schema("?bool"), "{ \"type\": \"boolean\" }");
17228+ assert_eq!(type_to_openapi_schema("int"), "{ \"type\": \"integer\" }");
17229+ }
17230+
17231+ // ── Fix: scalar source unwrap must not fabricate Int 0 ──
17232+
17233+ #[test]
17234+ #[should_panic(expected = "scalar source has no row")]
17235+ fn scalar_source_unwrap_panics_on_empty_relation() {
17236+ let empty = alloc(Value::Relation(Vec::new()));
17237+ let _ = knot_scalar_source_unwrap(empty);
17238+ }
17239+
17240+ // ── Fix: cancel() must wake a registered STM slot ──
17241+
17242+ #[test]
17243+ fn cancel_wakes_registered_stm_slot() {
17244+ let tok = CancelToken::new();
17245+ let slot = Arc::new(WakeSlot::new());
17246+ tok.set_stm_wake_slot(&slot);
17247+ tok.cancel();
17248+ assert!(tok.is_cancelled());
17249+ assert!(
17250+ *slot.woken.lock().unwrap(),
17251+ "cancel() must wake the slot registered by knot_stm_wait"
17252+ );
17253+ }
17254+
17255+ // ── Fix: stale Rows payloads upgrade to Bulk at notify time ──
17256+
17257+ #[test]
17258+ fn stale_rows_payload_upgrades_to_bulk_wake() {
17259+ let table = "t_toctou_upgrade_unique";
17260+ // Watcher indexed on col "a", registered AFTER a writer projected
17261+ // its payload (which only carries col "b").
17262+ let slot = Arc::new(WakeSlot::new());
17263+ EQ_WATCHERS
17264+ .entry((table.to_string(), "a".to_string(), SqlVal::Int(1).to_key()))
17265+ .or_default()
17266+ .push(Arc::downgrade(&slot));
17267+ TABLE_FILTER_COLS
17268+ .entry(table.to_string())
17269+ .or_default()
17270+ .cols
17271+ .insert("a".to_string(), 1);
17272+
17273+ let cols: Arc<[ColName]> = Arc::from(vec![intern_col("b")]);
17274+ let mut rows = EventRows::new(cols);
17275+ rows.push(vec![SqlVal::Int(9)]);
17276+ // Payload lacks "a": without the staleness re-check the "a" index is
17277+ // never probed and the watcher sleeps until the timeout.
17278+ wake_matching_watchers(table, &WriteEvent::Rows(rows));
17279+ assert!(
17280+ *slot.woken.lock().unwrap(),
17281+ "stale payload must fall back to a conservative Bulk wake"
17282+ );
17283+ TABLE_FILTER_COLS.remove(table);
17284+ }
17285+
17286+ #[test]
17287+ fn fresh_rows_payload_stays_precise() {
17288+ let table = "t_toctou_fresh_unique";
17289+ let slot = Arc::new(WakeSlot::new());
17290+ EQ_WATCHERS
17291+ .entry((table.to_string(), "a".to_string(), SqlVal::Int(1).to_key()))
17292+ .or_default()
17293+ .push(Arc::downgrade(&slot));
17294+ TABLE_FILTER_COLS
17295+ .entry(table.to_string())
17296+ .or_default()
17297+ .cols
17298+ .insert("a".to_string(), 1);
17299+
17300+ let cols: Arc<[ColName]> = Arc::from(vec![intern_col("a")]);
17301+ let mut rows = EventRows::new(cols);
17302+ rows.push(vec![SqlVal::Int(2)]); // ≠ 1 → indexed watcher must NOT wake
17303+ wake_matching_watchers(table, &WriteEvent::Rows(rows));
17304+ assert!(
17305+ !*slot.woken.lock().unwrap(),
17306+ "payload covering all filter cols must keep per-row precision"
17307+ );
17308+ TABLE_FILTER_COLS.remove(table);
17309+ }
17310+
17311+ // ── Fix: sealed-box KDF + contributory check ──
17312+
17313+ #[test]
17314+ fn crypto_sealed_box_round_trip() {
17315+ let pair = knot_crypto_generate_key_pair();
17316+ let private_key = knot_record_field(pair, b"privateKey".as_ptr(), 10);
17317+ let public_key = knot_record_field(pair, b"publicKey".as_ptr(), 9);
17318+
17319+ let plaintext = alloc(Value::Bytes(Arc::from(b"attack at dawn".to_vec())));
17320+ let ct = knot_crypto_encrypt(public_key, plaintext);
17321+ let decrypted = knot_crypto_decrypt(private_key, ct);
17322+ match unsafe { as_ref(decrypted) } {
17323+ Value::Bytes(b) => assert_eq!(&**b, b"attack at dawn" as &[u8]),
17324+ _ => panic!("expected Bytes from decrypt"),
17325+ }
17326+ }
17327+
17328+ #[test]
17329+ #[should_panic(expected = "non-contributory")]
17330+ fn crypto_encrypt_rejects_low_order_public_key() {
17331+ // The identity point (all zeros) is the canonical low-order input:
17332+ // DH with it yields an all-zero shared secret for ANY ephemeral key.
17333+ let zero_pub = alloc(Value::Bytes(Arc::from(vec![0u8; 32])));
17334+ let plaintext = alloc(Value::Bytes(Arc::from(b"x".to_vec())));
17335+ let _ = knot_crypto_encrypt(zero_pub, plaintext);
17336+ }
17337+
17338+ #[test]
17339+ #[should_panic(expected = "non-contributory")]
17340+ fn crypto_decrypt_rejects_low_order_ephemeral() {
17341+ let pair = knot_crypto_generate_key_pair();
17342+ let private_key = knot_record_field(pair, b"privateKey".as_ptr(), 10);
17343+ // Forged ciphertext with an all-zero (low-order) ephemeral point.
17344+ let ct = vec![0u8; 32 + 12 + 16];
17345+ let ct_val = alloc(Value::Bytes(Arc::from(ct)));
17346+ let _ = knot_crypto_decrypt(private_key, ct_val);
17347+ }
17348+
17349+ // ── Fix: HTTP body cap take() overflow ──
17350+
17351+ #[test]
17352+ fn body_cap_take_saturates_at_u64_max() {
17353+ use std::io::Read;
17354+ knot_set_http_max_body_bytes(u64::MAX);
17355+ let data = b"hello".to_vec();
17356+ let mut out = Vec::new();
17357+ // Mirrors the listen/fetch read paths: take(max.saturating_add(1)).
17358+ let max = http_max_body_bytes();
17359+ (&data[..])
17360+ .take(max.saturating_add(1))
17361+ .read_to_end(&mut out)
17362+ .unwrap();
17363+ assert_eq!(out, data);
17364+ knot_set_http_max_body_bytes(0); // restore env-or-default resolution
17365+ }
17366+}