diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 023944d3..3e2e5dfb 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -39,6 +39,8 @@ jobs: run: cargo clippy --workspace --all-targets -- -D warnings - name: No hand-rolled subscription handshakes run: etc/ci/check-accept-uni.sh + - name: No unscoped row overwrites + run: etc/ci/check-insert-or-replace.sh test: name: Test diff --git a/crates/core/src/oi/handler/apps.rs b/crates/core/src/oi/handler/apps.rs index 8330198a..4df5c999 100644 --- a/crates/core/src/oi/handler/apps.rs +++ b/crates/core/src/oi/handler/apps.rs @@ -124,9 +124,15 @@ pub(crate) fn persist_app_fields( installing: bool, ) -> rusqlite::Result<()> { db.conn.execute( - "INSERT OR REPLACE INTO registered_apps \ + // r[impl history.persist.partial-update] + "INSERT INTO registered_apps \ (name, installed, uninstalling, installing, current_generation) \ - VALUES (?1, ?2, ?3, ?4, ?5)", + VALUES (?1, ?2, ?3, ?4, ?5) \ + ON CONFLICT(name) DO UPDATE SET \ + installed = excluded.installed, \ + uninstalling = excluded.uninstalling, \ + installing = excluded.installing, \ + current_generation = excluded.current_generation", rusqlite::params![ name, installed as i64, diff --git a/crates/core/src/runtime/apps.rs b/crates/core/src/runtime/apps.rs index 67b9b2a8..4495f77c 100644 --- a/crates/core/src/runtime/apps.rs +++ b/crates/core/src/runtime/apps.rs @@ -352,9 +352,15 @@ impl AppRegistry { pub fn persist_app(db: &Db, entry: &AppEntry) -> rusqlite::Result<()> { let (installed, uninstalling, installing) = encode_phase(&entry.phase.lock()); db.conn.execute( - "INSERT OR REPLACE INTO registered_apps \ + // r[impl history.persist.partial-update] + "INSERT INTO registered_apps \ (name, installed, uninstalling, installing, current_generation) \ - VALUES (?1, ?2, ?3, ?4, ?5)", + VALUES (?1, ?2, ?3, ?4, ?5) \ + ON CONFLICT(name) DO UPDATE SET \ + installed = excluded.installed, \ + uninstalling = excluded.uninstalling, \ + installing = excluded.installing, \ + current_generation = excluded.current_generation", rusqlite::params![ entry.name, installed as i64, diff --git a/crates/core/src/runtime/apps/params.rs b/crates/core/src/runtime/apps/params.rs index 506a588d..72c87b05 100644 --- a/crates/core/src/runtime/apps/params.rs +++ b/crates/core/src/runtime/apps/params.rs @@ -33,7 +33,9 @@ pub fn upsert_param( value: &str, ) -> rusqlite::Result<()> { db.conn.execute( - "INSERT OR REPLACE INTO params (app_name, param_name, value) VALUES (?1, ?2, ?3)", + // r[impl history.persist.partial-update] + "INSERT INTO params (app_name, param_name, value) VALUES (?1, ?2, ?3) \ + ON CONFLICT(app_name, param_name) DO UPDATE SET value = excluded.value", rusqlite::params![app_name, param_name, value], )?; Ok(()) diff --git a/crates/core/src/runtime/apps/secret_params.rs b/crates/core/src/runtime/apps/secret_params.rs index 93b947c8..faa3c44f 100644 --- a/crates/core/src/runtime/apps/secret_params.rs +++ b/crates/core/src/runtime/apps/secret_params.rs @@ -47,7 +47,9 @@ pub fn upsert_secret_param( .encrypt(value) .map_err(|e| rusqlite::Error::ToSqlConversionFailure(Box::new(e)))?; db.conn.execute( - "INSERT OR REPLACE INTO secret_params (app_name, param_name, ciphertext) VALUES (?1, ?2, ?3)", + // r[impl history.persist.partial-update] + "INSERT INTO secret_params (app_name, param_name, ciphertext) VALUES (?1, ?2, ?3) \ + ON CONFLICT(app_name, param_name) DO UPDATE SET ciphertext = excluded.ciphertext", rusqlite::params![app_name, param_name, ct], )?; Ok(()) diff --git a/crates/core/src/runtime/barrier.rs b/crates/core/src/runtime/barrier.rs index 22a8cc7a..a6f7e9ca 100644 --- a/crates/core/src/runtime/barrier.rs +++ b/crates/core/src/runtime/barrier.rs @@ -111,6 +111,82 @@ pub struct BarrierRecord { } // r[impl barrier.replay] +/// The committed log does not describe the calls the closure is making. +/// +/// Means the script changed between the crash and the replay, or the engine +/// took a different branch. Either way the recorded results cannot be +/// attributed to the calls now being made. +// r[impl barrier.replay.positional] +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ReplayMismatch { + /// A different kind of call is being made at this position. + Kind { + call_index: usize, + expected: CallKind, + found: CallKind, + }, + /// The right kind of call, but its recorded argument differs. + /// + /// Only for arguments that are stable across passes by construction. The + /// resolved instance set is *not* one of those — a replica may be added + /// or retired between passes and it is still the same call — but a + /// literal like a signal name is: it comes from the script text, so a + /// change means the script changed under the log. + Extra { + call_index: usize, + kind: CallKind, + expected: String, + found: Option, + }, +} + +impl ReplayMismatch { + pub fn call_index(&self) -> usize { + match self { + Self::Kind { call_index, .. } | Self::Extra { call_index, .. } => *call_index, + } + } +} + +impl std::fmt::Display for ReplayMismatch { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Kind { + call_index, + expected, + found, + } => write!( + f, + "replay diverged at call {call_index}: the log records a {found:?} but the \ + script is making a {expected:?}" + ), + // This message is the whole explanation an operator gets for why + // an operation refused to resume, so it prints the recorded + // argument itself rather than its `Option` wrapper. + Self::Extra { + call_index, + kind, + expected, + found: Some(found), + } => write!( + f, + "replay diverged at call {call_index}: the log records a {kind:?} of `{found}` \ + but the script is making one of `{expected}`" + ), + Self::Extra { + call_index, + kind, + expected, + found: None, + } => write!( + f, + "replay diverged at call {call_index}: the log records a {kind:?} with no \ + recorded argument but the script is making one of `{expected}`" + ), + } + } +} + pub struct ReplayContext { pub operation_id: OperationId, pub call_index: usize, @@ -292,6 +368,56 @@ impl ReplayContext { self.committed.get(self.call_index) } + /// Consume the committed entry for the call being made *at this position*, + /// or `None` when this call is running for the first time. + /// + /// The action log is positional: `call_index` walks `committed` in the + /// order the closure makes its calls. `do_exec` always understood that; + /// `do_signal` did not, and scanned the whole log for any entry with the + /// same resources and signal. Both halves of that are wrong. A second, + /// identical `rt.signal` later in the same closure matched the first + /// entry and was swallowed — the signal was never delivered. And when the + /// resolved instance set changed between passes, no entry matched, so a + /// signal already delivered before the crash was delivered again. + /// + /// Advancing the index is part of consuming the entry, so a caller cannot + /// check without advancing or advance without checking. + // r[impl barrier.replay.positional] + /// `expect_extra` is checked when the caller's argument is stable across + /// passes by construction; pass `None` to skip the check. Positional + /// matching alone would treat a script edit that changes the argument at + /// this position — `SIGHUP` to `SIGTERM`, say — as already replayed, and + /// silently never deliver the new one. + pub fn replay_step( + &mut self, + expect: CallKind, + expect_extra: Option<&str>, + ) -> Result, ReplayMismatch> { + if !self.is_replaying() { + return Ok(None); + } + let entry = self.committed[self.call_index].clone(); + if entry.call_kind != expect { + return Err(ReplayMismatch::Kind { + call_index: self.call_index, + expected: expect, + found: entry.call_kind, + }); + } + if let Some(expected) = expect_extra + && entry.extra.as_deref() != Some(expected) + { + return Err(ReplayMismatch::Extra { + call_index: self.call_index, + kind: expect, + expected: expected.to_owned(), + found: entry.extra.clone(), + }); + } + self.call_index += 1; + Ok(Some(entry)) + } + pub fn take_pending(&mut self) -> Vec { std::mem::take(&mut self.pending) } diff --git a/crates/core/src/runtime/barrier/replay/tests.rs b/crates/core/src/runtime/barrier/replay/tests.rs index 89550798..b91db858 100644 --- a/crates/core/src/runtime/barrier/replay/tests.rs +++ b/crates/core/src/runtime/barrier/replay/tests.rs @@ -273,3 +273,156 @@ fn db_action_log_sequential_barriers() { assert_eq!(entries[0].call_index, 0); assert_eq!(entries[1].call_index, 1); } + +// r[verify barrier.replay.positional] +// Positional matching, not value matching. Both halves of the old +// value-scan were wrong: a second identical call matched the first entry +// and was swallowed, and a call whose arguments resolved differently +// between passes matched nothing and re-ran. +mod positional { + use crate::runtime::barrier::{ActionLogEntry, CallKind, ReplayContext}; + + use super::*; + + fn instance(name: &str) -> ResourceInstance { + ResourceInstance { + id: crate::runtime::identity::InstanceId::generate(), + app: app_name(), + kind: ResourceKind::Deployment, + name: Some(name.to_owned()), + variant: crate::runtime::identity::InstanceVariant::Singleton, + display_name: format!("test-app-{name}"), + } + } + + fn entry(index: usize, kind: CallKind, resources: Vec) -> ActionLogEntry { + ActionLogEntry { + call_index: index, + call_kind: kind, + resources, + barrier: None, + extra: Some("SIGHUP".to_owned()), + } + } + + fn ctx_with(committed: Vec) -> ReplayContext { + ReplayContext::new( + OperationId("op-positional".into()), + committed, + Arc::new(TestWorldOracle::default()), + Arc::new(crate::runtime::barrier::CancelToken::default()), + ) + } + + // r[verify barrier.replay.positional] + #[test] + fn two_identical_calls_consume_two_entries() { + let db = instance("db"); + let mut ctx = ctx_with(vec![ + entry(0, CallKind::Signal, vec![db.clone()]), + entry(1, CallKind::Signal, vec![db.clone()]), + ]); + + // Both are replays of their own position, not one matching twice. + assert!(ctx.replay_step(CallKind::Signal, None).unwrap().is_some()); + assert!(ctx.replay_step(CallKind::Signal, None).unwrap().is_some()); + // A third identical call at a position the log does not cover is new + // and must actually run — which is what the value scan swallowed. + assert!(ctx.replay_step(CallKind::Signal, None).unwrap().is_none()); + } + + // r[verify barrier.replay.positional] + // The instance set can differ between passes — a replica added or + // retired — and that does not make it a different call. Matching by value + // found nothing here and re-delivered a signal already delivered. + #[test] + fn a_changed_instance_set_is_still_the_same_call() { + let before = vec![instance("db")]; + let after = vec![instance("db"), instance("db")]; + let mut ctx = ctx_with(vec![entry(0, CallKind::Signal, before)]); + + let replayed = ctx.replay_step(CallKind::Signal, None).unwrap(); + assert!( + replayed.is_some(), + "the call at this position was already made, whatever it resolved to" + ); + assert_ne!(replayed.unwrap().resources, after); + } + + // r[verify barrier.replay.positional] + #[test] + fn a_diverged_log_fails_rather_than_guessing() { + let mut ctx = ctx_with(vec![entry(0, CallKind::Exec, vec![instance("db")])]); + let err = ctx.replay_step(CallKind::Signal, None).unwrap_err(); + assert_eq!(err.call_index(), 0); + assert!( + matches!( + err, + crate::runtime::barrier::ReplayMismatch::Kind { + expected: CallKind::Signal, + found: CallKind::Exec, + .. + } + ), + "{err}" + ); + // The message is the operator's only account of why an operation + // refused to resume, so it has to name the position and both kinds. + let message = err.to_string(); + assert!(message.contains("call 0"), "{message}"); + assert!(message.contains("Exec"), "{message}"); + assert!(message.contains("Signal"), "{message}"); + } + + // r[verify barrier.replay.positional] + // Position alone is not enough for an argument that comes from the script + // text. A script edited from SIGHUP to SIGTERM at the same position would + // otherwise be treated as already replayed, and the new signal never + // delivered. The resolved instance set is deliberately *not* checked this + // way — it legitimately varies between passes. + #[test] + fn a_changed_signal_name_is_a_divergence() { + let mut ctx = ctx_with(vec![entry(0, CallKind::Signal, vec![instance("db")])]); + let err = ctx + .replay_step(CallKind::Signal, Some("SIGTERM")) + .unwrap_err(); + assert!( + matches!( + err, + crate::runtime::barrier::ReplayMismatch::Extra { ref expected, .. } + if expected == "SIGTERM" + ), + "{err}" + ); + let message = err.to_string(); + assert!(message.contains("SIGTERM"), "{message}"); + assert!(message.contains("SIGHUP"), "{message}"); + assert!( + !message.contains("Some("), + "the recorded argument is shown, not its Option wrapper: {message}" + ); + + // The recorded name replays cleanly. + let mut ctx = ctx_with(vec![entry(0, CallKind::Signal, vec![instance("db")])]); + assert!( + ctx.replay_step(CallKind::Signal, Some("SIGHUP")) + .unwrap() + .is_some() + ); + } + + // r[verify barrier.replay.positional] + #[test] + fn a_log_entry_with_no_recorded_argument_says_so() { + let mut committed = vec![entry(0, CallKind::Signal, vec![instance("db")])]; + committed[0].extra = None; + let mut ctx = ctx_with(committed); + let err = ctx + .replay_step(CallKind::Signal, Some("SIGTERM")) + .unwrap_err(); + let message = err.to_string(); + assert!(message.contains("no recorded argument"), "{message}"); + assert!(message.contains("SIGTERM"), "{message}"); + assert!(!message.contains("None"), "{message}"); + } +} diff --git a/crates/core/src/runtime/barrier/runtime.rs b/crates/core/src/runtime/barrier/runtime.rs index 33be2ae6..cf69e3fc 100644 --- a/crates/core/src/runtime/barrier/runtime.rs +++ b/crates/core/src/runtime/barrier/runtime.rs @@ -1064,18 +1064,22 @@ impl RuntimeInstance { Some(c) => Arc::clone(c), }; + // l[impl rt.signal] r[impl barrier.replay.positional] + // At-most-once is per *call site*, not per (resources, signal) value. + // Scanning the whole committed log for a matching value got both + // halves wrong: a second, identical rt.signal later in the same + // closure matched the first entry and was swallowed, so the signal was + // never delivered; and when the resolved instance set changed between + // passes — a replica added or retired — nothing matched, so a signal + // already delivered before the crash was delivered again. { let mut g = ctx.lock(); - let already = g.committed.iter().any(|e| { - matches!(e.call_kind, CallKind::Signal) - && e.resources == expanded - && e.extra.as_deref() == Some(canonical.as_str()) - }); - if already { - if g.is_replaying() { - g.call_index += 1; - } - return Ok(()); + // The signal name is a literal in the script, so unlike the + // resolved instance set it must match the log. + match g.replay_step(CallKind::Signal, Some(canonical.as_str())) { + Ok(Some(_)) => return Ok(()), + Ok(None) => {} + Err(mismatch) => return Err(Box::::from(mismatch.to_string())), } } diff --git a/crates/core/src/runtime/desired.rs b/crates/core/src/runtime/desired.rs index 1dd2cc91..3ca6697a 100644 --- a/crates/core/src/runtime/desired.rs +++ b/crates/core/src/runtime/desired.rs @@ -309,9 +309,17 @@ pub fn insert_dynamic_resource( description: Option<&str>, ) -> rusqlite::Result<()> { db.conn.execute( - "INSERT OR REPLACE INTO dynamic_resources + // r[impl history.persist.partial-update] + "INSERT INTO dynamic_resources (instance_id, app, operation_id, kind, display_name, resource_name, description) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) + ON CONFLICT(instance_id) DO UPDATE SET + app = excluded.app, + operation_id = excluded.operation_id, + kind = excluded.kind, + display_name = excluded.display_name, + resource_name = excluded.resource_name, + description = excluded.description", rusqlite::params![ instance.id.to_hex(), instance.app, diff --git a/crates/core/src/runtime/history.rs b/crates/core/src/runtime/history.rs index da9ef2b0..edf4702d 100644 --- a/crates/core/src/runtime/history.rs +++ b/crates/core/src/runtime/history.rs @@ -504,10 +504,26 @@ pub fn save_current_operation( .map_err(OperationPersistError::Cipher)?; db.conn .execute( - "INSERT OR REPLACE INTO current_operation + // r[impl history.persist.partial-update] + // Not INSERT OR REPLACE. That is delete-then-insert, so every + // column this statement does not name reverts to its default — + // including `cancel_requested`, which a *different* writer owns. + // The replay path re-saves this row immediately after the daemon + // has read the cancel flag, so a cancellation was silently + // dropped while the operation it cancelled was still in flight. + // The column was added by a later migration than the statement, + // which is exactly how this kind of bug arrives. + "INSERT INTO current_operation (singleton, operation_id, app, action_name, source_generation, target_generation, params_ciphertext) - VALUES (1, ?1, ?2, ?3, ?4, ?5, ?6)", + VALUES (1, ?1, ?2, ?3, ?4, ?5, ?6) + ON CONFLICT(singleton) DO UPDATE SET + operation_id = excluded.operation_id, + app = excluded.app, + action_name = excluded.action_name, + source_generation = excluded.source_generation, + target_generation = excluded.target_generation, + params_ciphertext = excluded.params_ciphertext", params![ op.operation_id.0, op.app, diff --git a/crates/core/src/runtime/history/tests.rs b/crates/core/src/runtime/history/tests.rs index de6720a6..019d4747 100644 --- a/crates/core/src/runtime/history/tests.rs +++ b/crates/core/src/runtime/history/tests.rs @@ -709,3 +709,36 @@ fn delete_instance_clears_observations_and_faults_atomically() { assert_eq!(obs_count, 0, "observations deleted"); assert_eq!(fault_count, 0, "faults deleted"); } + +// r[verify history.persist.partial-update] +// The replay path re-saves the operation row immediately after the daemon has +// read the cancel flag. Under INSERT OR REPLACE — delete-then-insert — that +// reverted `cancel_requested` to its default, silently dropping a +// cancellation while the operation it cancelled was still in flight. The +// column is owned by `set_cancel_requested`, not by this writer. +#[test] +fn re_saving_the_operation_preserves_a_cancel_request() { + use crate::runtime::secrets::Cipher; + + let db = Db::open_in_memory().unwrap(); + let cipher = Cipher::for_tests(); + let op = CurrentOperation { + operation_id: OperationId("op-cancel".into()), + app: app_name("app"), + action_name: action_name("migrate"), + source_generation: 1, + target_generation: 1, + }; + + save_current_operation(&db, &cipher, &op, &serde_json::Map::new()).unwrap(); + assert!(set_cancel_requested(&db, &OperationId("op-cancel".into())).unwrap()); + assert!(load_cancel_requested(&db).unwrap()); + + // The same operation is re-saved, as the replay path does. + save_current_operation(&db, &cipher, &op, &serde_json::Map::new()).unwrap(); + + assert!( + load_cancel_requested(&db).unwrap(), + "a writer must not reset a column another writer owns" + ); +} diff --git a/crates/core/src/runtime/scaling.rs b/crates/core/src/runtime/scaling.rs index e72cac4f..09ab7782 100644 --- a/crates/core/src/runtime/scaling.rs +++ b/crates/core/src/runtime/scaling.rs @@ -34,8 +34,12 @@ pub fn save_scaling_decision( ) -> rusqlite::Result<()> { let now = jiff::Timestamp::now().to_string(); db.conn.execute( - "INSERT OR REPLACE INTO scaling_decisions (app, deployment, scale, updated_at) - VALUES (?1, ?2, ?3, ?4)", + // r[impl history.persist.partial-update] + "INSERT INTO scaling_decisions (app, deployment, scale, updated_at) + VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(app, deployment) DO UPDATE SET + scale = excluded.scale, + updated_at = excluded.updated_at", params![app, deployment, scale as i64, now], )?; Ok(()) diff --git a/crates/core/src/runtime/schedules.rs b/crates/core/src/runtime/schedules.rs index 750e5305..6198c831 100644 --- a/crates/core/src/runtime/schedules.rs +++ b/crates/core/src/runtime/schedules.rs @@ -105,19 +105,32 @@ pub fn check_due_schedules( let accepted = matches!(result, ScheduleResult::Accepted); match result { ScheduleResult::Accepted | ScheduleResult::Queued => { - let fired_at = now.to_string(); - if let Err(e) = db::upsert_schedule_fired( - db, - &row.app, - &row.action, - &row.cronexpr, - &fired_at, - ) { - tracing::error!( - app = %row.app, - action = %row.action, - "failed to update last_fired_at: {e}" - ); + // r[impl schedule.catch-up] + // Stamp the fire when it happens, not when it is intended. + // A queued fire exists only in the scheduler's in-memory + // VecDeque, so stamping it here lost the operation on a + // restart while the database said it had fired — and + // r[schedule.catch-up] then had nothing to catch up on. + // While it stays queued the next tick's re-fire is + // rejected as SameAppAlreadyQueued and silently dropped, + // so leaving it unstamped cannot double-fire; after a + // restart the unstamped schedule fires again, which is + // the point. + if accepted { + let fired_at = now.to_string(); + if let Err(e) = db::upsert_schedule_fired( + db, + &row.app, + &row.action, + &row.cronexpr, + &fired_at, + ) { + tracing::error!( + app = %row.app, + action = %row.action, + "failed to update last_fired_at: {e}" + ); + } } let op_id = if accepted { scheduler.active().map(|a| a.operation_id.clone()) diff --git a/docs/spec/runtime.md b/docs/spec/runtime.md index bb00aa1e..643f30c2 100644 --- a/docs/spec/runtime.md +++ b/docs/spec/runtime.md @@ -27,6 +27,15 @@ Absent specification bugs, anything that is not defined here is either defined i > All operations performed by the reconciler must be idempotent. > Performing the same operation twice must not cause errors or duplicate side effects. +> r[barrier.replay.positional] +> The action execution log is positional: a replayed call is matched to the committed entry at its own position in the closure's call sequence, never to any entry elsewhere in the log that happens to look similar. +> "At most once" for a call is therefore per call site: two identical calls at different points in a closure are two calls, and a call whose arguments resolve differently between passes is still the same call. +> A replay whose committed entry does not correspond to the call being made must fail rather than guess. + +> r[history.persist.partial-update] +> A writer that updates a row shared with another writer must not reset columns it does not own. +> This holds as columns are added: a write that names its own columns explicitly must continue to leave the rest untouched when the row gains a new one. + > r[reconciliation.liveness] > Individual reconciliation operations must not block the loop for an unbounded or long duration. > When an operation requires waiting for an external condition (e.g. a process to terminate), the reconciler must release control and re-evaluate the condition on a subsequent iteration rather than polling inline. diff --git a/etc/ci/check-insert-or-replace.sh b/etc/ci/check-insert-or-replace.sh new file mode 100755 index 00000000..10a83e00 --- /dev/null +++ b/etc/ci/check-insert-or-replace.sh @@ -0,0 +1,65 @@ +#!/usr/bin/env bash +# Fail on new `INSERT OR REPLACE` outside the allowlist. +# +# INSERT OR REPLACE is delete-then-insert: every column the statement does not +# name reverts to its default. That is safe only while one writer owns the +# whole row — and it stops being safe the moment a migration adds a column +# another writer owns, which is precisely how `current_operation` came to +# silently drop a persisted cancellation. The failure arrives with the +# migration, not with the statement, so it cannot be caught by reviewing the +# statement alone. +# +# Prefer `INSERT ... ON CONFLICT(key) DO UPDATE SET col = excluded.col, ...`, +# which names what it writes and leaves the rest alone. +# +# Migrations are exempt: they run once against a known schema. + +set -euo pipefail + +# Each entry is a file that may use it, with the reason. +# history.rs — action_log's positional overwrite keyed on +# (operation_id, call_index) IS the replay contract: +# rewriting a call's entry in place is the intended +# semantics. +# history/tests.rs — an assertion message naming the statement it is +# asserting about. Tests are not writers of production +# rows, and the alternative is a prose filter loose +# enough to skip real code. +allowed=( + 'crates/core/src/runtime/history.rs' + 'crates/core/src/runtime/history/tests.rs' +) + +violations=() +while IFS= read -r hit; do + file="${hit%%:*}" + [[ "$file" == *"/migrations/"* ]] && continue + # A comment mentioning the statement is not a use of it. Only a genuine + # comment line is skipped: an earlier version skipped any line containing a + # quote unless `INTO` was on that same line too, which let a statement + # whose `INTO` wrapped to the next line through unseen. Everything that is + # not a comment must reach the allowlist to be excused. + line="${hit#*:}" + line="${line#*:}" + trimmed="${line#"${line%%[![:space:]]*}"}" + [[ "$trimmed" == //* || "$trimmed" == --* ]] && continue + skip=false + for ok in "${allowed[@]}"; do + [[ "$file" == "$ok" ]] && skip=true && break + done + $skip || violations+=("$hit") +done < <(grep -rn --include='*.rs' 'INSERT OR REPLACE' crates/ || true) + +if ((${#violations[@]})); then + echo "error: INSERT OR REPLACE outside the allowlist:" >&2 + printf ' %s\n' "${violations[@]}" >&2 + cat >&2 <<'EOF' + +INSERT OR REPLACE resets every column the statement does not name, so it is +only safe while a single writer owns the whole row — and an ALTER TABLE ADD +COLUMN can end that at any time, silently. Use +`INSERT ... ON CONFLICT(key) DO UPDATE SET ...` naming the columns this writer +owns. See docs/logic-bug-audit-2026-07/theme-8-restart-replay.md. +EOF + exit 1 +fi