fix(replay): match replayed calls by position, not by value (audit theme 8)

.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

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,

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,

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(())

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(())

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<String>,
+    },
+}
+
+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<Option<ActionLogEntry>, 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<ActionLogEntry> {
         std::mem::take(&mut self.pending)
     }

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<ResourceInstance>) -> ActionLogEntry {
+        ActionLogEntry {
+            call_index: index,
+            call_kind: kind,
+            resources,
+            barrier: None,
+            extra: Some("SIGHUP".to_owned()),
+        }
+    }
+
+    fn ctx_with(committed: Vec<ActionLogEntry>) -> 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}");
+    }
+}

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::<EvalAltResult>::from(mismatch.to_string())),
             }
         }