From 128a6ef52cc8be3d1834fb19f8f292bed52e68b4 Mon Sep 17 00:00:00 2001 From: kjgbot Date: Sat, 5 Sep 2026 16:08:15 +0200 Subject: [PATCH] fix(kernel): stop the claim repair from stealing an in-flight claim Two racing deliveries of one event each started a run (#160). CI caught it as `left: 2`; it is an exactly-once violation, not a flake. `claim_event` repaired a claim whose run never materialised, testing for that with "no row in `runs` for the claimed run_id". That predicate is true of two situations it could not tell apart: a run that CRASHED before registering, which nobody will ever spawn, and a run that is IN FLIGHT and has not registered yet. `wake.rs` makes the window explicit -- the claim is written, and only several statements later does `SqliteJournal::create` bring the run into existence -- so a second delivery landing in that gap judged the first abandoned and spawned its own run. The claim now carries a boot id. A claim from THIS boot is in flight and dedupes; one from a previous boot with no registered run is wreckage and is repaired, preserving the crash recovery the original comment defends. **The boot id identifies the PROCESS, not an `Engine`.** This is the whole of the fix and an earlier revision got it wrong: it generated the id per `Engine`, which made the change inert under the only topology that matters. The server builds a fresh `Engine::with_runtime` inside `handle_request`, so two concurrent `event.submit` calls hold two different `Engine`s over one data dir; with per-`Engine` ids the second delivery would still treat the first's live claim as wreckage and spawn a duplicate. A review lens caught it, and it also caught that the test shared one `Engine` and therefore passed for the wrong reason. Both are fixed. The racing test now builds a separate `Engine` per racer, which is what production does. And because that test is probabilistic -- against a per-`Engine` id it caught the bug only 2 times in 20, since the winner usually registers before any loser reads -- the property is also asserted deterministically in `every_engine_in_this_process_shares_one_boot_id`. The same-boot rule creates one obligation: a claim this boot takes but cannot turn into a run would strand the event. The span from claim to `register` is therefore a named method, `spawn_claimed_run`, whose failure path releases the claim, scoped to the claiming run_id. Concurrency also had to be made to work at all, which is a separate question from who owns a claim: * `Registry::open` had no busy timeout, so concurrent deliveries failed with `database is locked` rather than waiting. * `PRAGMA journal_mode = WAL` does NOT honour the busy handler, so the timeout cannot cover it however early it is set. It retries explicitly and then VERIFIES the mode, returning `RegistryNotWal` otherwise rather than running in rollback-journal mode silently. * The timeout is armed AFTER the switch: the retry is bounded in attempts, not wall-clock, so arming it first let one open block 50 x 5s. Rejected: a time-based abandonment threshold, which swaps a correctness bound for a timing guess. Deferred and tracked: a panic between claim and `register` strands the event for the life of the boot (#173). Mutation witnesses, which are cited instead of test totals because totals drift with the base and a witness does not: * `new_boot_id` returning a fresh id per call fails `every_engine_in_this_process_shares_one_boot_id` deterministically: "two Engines in one process must share a boot id, or concurrent deliveries repair each other's live claims" * `if false && existing_boot == boot_id` fails `a_same_boot_claim_with_no_run_yet_is_a_duplicate_not_wreckage` with `left: None, right: Some("run-a")` Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FtQSAcGDta5VH9xiZFT4sR Session-Id: c228933d-4f94-4d83-9a9a-daf3c83b94f1 --- kernel/relayflowd-journal/src/lib.rs | 5 + kernel/relayflowd-journal/src/registry.rs | 323 +++++++++++++++++- .../relayflowd-journal/src/subscriptions.rs | 56 ++- kernel/relayflowd/src/engine.rs | 77 +++++ kernel/relayflowd/src/engine/wake.rs | 119 +++++-- kernel/relayflowd/tests/event_wake.rs | 97 ++++++ 6 files changed, 628 insertions(+), 49 deletions(-) diff --git a/kernel/relayflowd-journal/src/lib.rs b/kernel/relayflowd-journal/src/lib.rs index fd0fe81c3..59dc976d3 100644 --- a/kernel/relayflowd-journal/src/lib.rs +++ b/kernel/relayflowd-journal/src/lib.rs @@ -246,6 +246,11 @@ fn to_core_error(error: JournalStoreError) -> JournalError { pub enum JournalStoreError { #[error("journal file already exists: {0}")] AlreadyExists(PathBuf), + #[error( + "run registry at {path} is in journal mode {actual}, not WAL; \ + concurrent deliveries would not be durable" + )] + RegistryNotWal { path: PathBuf, actual: String }, #[error("journal I/O failed: {0}")] Io(#[from] std::io::Error), #[error("SQLite journal failed: {0}")] diff --git a/kernel/relayflowd-journal/src/registry.rs b/kernel/relayflowd-journal/src/registry.rs index f3f884458..c60a47fce 100644 --- a/kernel/relayflowd-journal/src/registry.rs +++ b/kernel/relayflowd-journal/src/registry.rs @@ -4,6 +4,12 @@ use rusqlite::{Connection, OptionalExtension, params}; use crate::JournalStoreError; +/// Bounded retry for the WAL switch; see `Registry::open`. Named for the +/// switch specifically -- these are not a general retry policy, and a search +/// for "retry" elsewhere in the crate should not land on them by accident. +const WAL_SWITCH_RETRY_ATTEMPTS: usize = 50; +const WAL_SWITCH_RETRY_SLEEP_MS: u64 = 2; + pub struct Registry { connection: Connection, } @@ -33,9 +39,62 @@ impl Registry { std::fs::create_dir_all(parent)?; } let connection = Connection::open(path)?; + // The WAL switch: `busy_timeout` does NOT cover it. SQLite takes an + // exclusive lock to change journal mode and does not invoke the busy + // handler for it, so a concurrent open fails immediately with + // `database is locked` however long the timeout is. Measured directly: + // with the timeout set first, the racing test still failed 36 times in + // 50, and instrumenting the batch statement-by-statement named + // `journal_mode` every time. + // + // So retry it explicitly. WAL is a persistent property of the file, so + // a loser here is racing a winner that is setting the same mode. + let mut mode = None; + for attempt in 0..WAL_SWITCH_RETRY_ATTEMPTS { + match connection.query_row("PRAGMA journal_mode = WAL", [], |row| { + row.get::<_, String>(0) + }) { + Ok(observed) => { + mode = Some(observed); + break; + } + Err(error) => { + if attempt + 1 == WAL_SWITCH_RETRY_ATTEMPTS { + return Err(error.into()); + } + std::thread::sleep(std::time::Duration::from_millis(WAL_SWITCH_RETRY_SLEEP_MS)); + } + } + } + // Fail closed. Swallowing the error and continuing would leave the + // registry in rollback-journal mode silently, which is exactly the + // silent fallback this codebase refuses -- a reader would see a healthy + // open and lose the concurrency guarantee the mode is there to provide. + let mode = mode.unwrap_or_default(); + if !mode.eq_ignore_ascii_case("wal") { + return Err(JournalStoreError::RegistryNotWal { + path: path.to_path_buf(), + actual: mode, + }); + } + + // busy_timeout goes on AFTER the WAL switch, deliberately. + // + // It makes ordinary statements wait instead of failing instantly, which + // is what two concurrent deliveries need. But the retry loop above is + // bounded in ATTEMPTS, not in wall-clock, so with a 5s timeout already + // armed a single open could block 50 x 5s. `crash_resume`'s sigkill + // sweep opens the registry many times over, and on a GitHub runner that + // compounded into a step that ran past 25 minutes and was cancelled + // (run 33960456965) while passing locally in 1.5s. + // + // Ordering is safe to choose freely here: measured against the seeded + // race, timeout-first and timeout-last behaved the same (36/50 and + // 34/50 lock failures). The WAL retry is what fixed that, not this. connection.execute_batch( - "PRAGMA journal_mode = WAL; + "PRAGMA busy_timeout = 5000; PRAGMA synchronous = FULL; + CREATE TABLE IF NOT EXISTS runs ( run_id TEXT PRIMARY KEY, file TEXT NOT NULL, @@ -47,11 +106,19 @@ impl Registry { -- that derive the same key -- both using a template like -- `test.ping:hello` -- would collide, and the second flow's event -- would be reported deduped without ever spawning a run. + -- `boot_id` names the engine boot that wrote the claim. It exists + -- to separate two states the repair path below could not tell + -- apart: a run that CRASHED before registering, and a run that is + -- concurrently in flight and has not registered YET. Both present + -- as a claim row with no matching `runs` row, and treating the + -- second as the first let two racing deliveries of one event each + -- start a run (#160). CREATE TABLE IF NOT EXISTS event_dedupe ( flow_key TEXT NOT NULL, subscription_id TEXT NOT NULL, dedupe_key TEXT NOT NULL, run_id TEXT NOT NULL, + boot_id TEXT NOT NULL DEFAULT '', PRIMARY KEY (flow_key, subscription_id, dedupe_key) ) WITHOUT ROWID;", )?; @@ -60,6 +127,28 @@ impl Registry { // connection during initialization. See `subscriptions.rs` for // the LIVENESS_SCHEMA_SQL contents and its rationale. connection.execute_batch(crate::subscriptions::LIVENESS_SCHEMA_SQL)?; + // `CREATE TABLE IF NOT EXISTS` above is inert against a database that + // predates `boot_id`, so an existing registry needs the column added. + // A duplicate-column error means a prior open already did it; anything + // else is a real failure and must surface. + // Ask the table what it has rather than adding the column and reading + // the failure message. Matching on "duplicate column name" would couple + // this to SQLite's wording, which is not part of any contract. + let has_boot_id = connection + .prepare("PRAGMA table_info(event_dedupe)")? + .query_map([], |row| row.get::<_, String>(1))? + .collect::, _>>()? + .iter() + .any(|column| column == "boot_id"); + if !has_boot_id { + connection.execute( + "ALTER TABLE event_dedupe ADD COLUMN boot_id TEXT NOT NULL DEFAULT ''", + [], + )?; + } + // Rows written before this column existed carry '', which matches no + // live boot id, so they are treated as belonging to a previous boot -- + // which is exactly right: the process that wrote them is gone. Ok(Self { connection }) } @@ -146,21 +235,22 @@ impl Registry { subscription_id: &str, dedupe_key: &str, run_id: &str, + boot_id: &str, ) -> Result, JournalStoreError> { let changed = self.connection.execute( - "INSERT OR IGNORE INTO event_dedupe(flow_key, subscription_id, dedupe_key, run_id) - VALUES (?1, ?2, ?3, ?4)", - params![flow_key, subscription_id, dedupe_key, run_id], + "INSERT OR IGNORE INTO event_dedupe(flow_key, subscription_id, dedupe_key, run_id, boot_id) + VALUES (?1, ?2, ?3, ?4, ?5)", + params![flow_key, subscription_id, dedupe_key, run_id, boot_id], )?; if changed == 1 { return Ok(None); } - let existing: String = self.connection.query_row( - "SELECT run_id FROM event_dedupe + let (existing, existing_boot): (String, String) = self.connection.query_row( + "SELECT run_id, boot_id FROM event_dedupe WHERE flow_key = ?1 AND subscription_id = ?2 AND dedupe_key = ?3", params![flow_key, subscription_id, dedupe_key], - |row| row.get(0), + |row| Ok((row.get(0)?, row.get(1)?)), )?; // Is the claimed run real? `runs` is written when the run is @@ -174,16 +264,57 @@ impl Registry { return Ok(Some(existing)); } - // The prior claim never became a run. Repair it by taking it over, so - // the retry spawns rather than being told it is a duplicate. + // No run row yet. That is TWO different situations, and the difference + // is the whole of #160: + // + // * a claim from a PREVIOUS boot -- the process that took it is gone, + // so no one will ever spawn that run. Repair it. + // * a claim from THIS boot -- another delivery is between its claim + // and its registration right now. Repairing that starts a second + // run for one event, which is an exactly-once violation. + // + // Before `boot_id` these were indistinguishable and both were repaired, + // so two racing deliveries each spawned a run. A same-boot claim is + // therefore a duplicate, not wreckage. + // + // The in-flight caller releases its own claim if the run fails to + // materialise (see `release_claim`), so a same-boot claim that will + // never become a run does not linger and strand the event. + if existing_boot == boot_id { + return Ok(Some(existing)); + } + + // The prior boot's claim never became a run. Repair it by taking it + // over, so the retry spawns rather than being told it is a duplicate. self.connection.execute( - "UPDATE event_dedupe SET run_id = ?4 + "UPDATE event_dedupe SET run_id = ?4, boot_id = ?5 WHERE flow_key = ?1 AND subscription_id = ?2 AND dedupe_key = ?3", - params![flow_key, subscription_id, dedupe_key, run_id], + params![flow_key, subscription_id, dedupe_key, run_id, boot_id], )?; Ok(None) } + /// Release a claim this boot took but could not turn into a run. + /// + /// Without this, `claim_event`'s same-boot rule would strand the event: the + /// claim stays, every retry inside this process is told "duplicate", and no + /// run exists to carry it. Scoped to `run_id` so a release cannot delete a + /// claim that some other delivery has since legitimately taken over. + pub fn release_claim( + &self, + flow_key: &str, + subscription_id: &str, + dedupe_key: &str, + run_id: &str, + ) -> Result<(), JournalStoreError> { + self.connection.execute( + "DELETE FROM event_dedupe + WHERE flow_key = ?1 AND subscription_id = ?2 AND dedupe_key = ?3 + AND run_id = ?4", + params![flow_key, subscription_id, dedupe_key, run_id], + )?; + Ok(()) + } } #[cfg(test)] @@ -192,6 +323,175 @@ mod tests { use super::*; + /// #160, stated without concurrency. The racing integration test in + /// `relayflowd/tests/event_wake.rs` exercises the real path but is + /// probabilistic: it only catches the bug when a loser reads inside the + /// winner's claim window, which measured 3 runs in 30 against a + /// deliberately broken guard. The rule it depends on is deterministic, so + /// it is also asserted directly here, where no scheduling is involved -- + /// and this is what actually gates the fix. + #[test] + fn a_same_boot_claim_with_no_run_yet_is_a_duplicate_not_wreckage() { + let directory = tempdir().unwrap(); + let registry = Registry::open(directory.path().join("r.sqlite3")).unwrap(); + + // First delivery claims and, like the real engine, has not registered + // its run yet. + assert_eq!( + registry + .claim_event("f", "s", "k", "run-a", "boot-1") + .unwrap(), + None + ); + + // Second delivery, same boot, arrives inside that window. Before the + // fix this repaired the claim and returned None, spawning a second run. + assert_eq!( + registry + .claim_event("f", "s", "k", "run-b", "boot-1") + .unwrap(), + Some("run-a".to_string()), + "a claim held by this boot is in flight, not abandoned" + ); + } + + /// The crash case the repair path was written for must still work: a claim + /// left by a process that died names a run nobody will ever spawn. + #[test] + fn a_previous_boots_claim_with_no_run_is_repaired() { + let directory = tempdir().unwrap(); + let registry = Registry::open(directory.path().join("r.sqlite3")).unwrap(); + + assert_eq!( + registry + .claim_event("f", "s", "k", "run-a", "boot-1") + .unwrap(), + None + ); + // A new boot finds the orphaned claim and must take it over, otherwise + // the event is lost silently -- the exactly-once violation the original + // repair existed to prevent. + assert_eq!( + registry + .claim_event("f", "s", "k", "run-b", "boot-2") + .unwrap(), + None, + "a claim from a dead boot with no run must be repaired" + ); + } + + /// The ALTER path: a registry written before `boot_id` existed must open, + /// gain the column, and read its pre-existing rows as a previous boot's. + /// This branch only runs against an old database, so nothing else in the + /// suite covers it and a refactor could drop it silently. + #[test] + fn a_pre_migration_registry_gains_boot_id_and_its_claims_are_repairable() { + let directory = tempdir().unwrap(); + let path = directory.path().join("r.sqlite3"); + + // Exactly the pre-#160 schema, with a claim recorded under it. + { + let legacy = rusqlite::Connection::open(&path).unwrap(); + legacy + .execute_batch( + "CREATE TABLE runs ( + run_id TEXT PRIMARY KEY, file TEXT NOT NULL, + status TEXT NOT NULL, next_wake_at_ms INTEGER + ) WITHOUT ROWID; + CREATE TABLE event_dedupe ( + flow_key TEXT NOT NULL, subscription_id TEXT NOT NULL, + dedupe_key TEXT NOT NULL, run_id TEXT NOT NULL, + PRIMARY KEY (flow_key, subscription_id, dedupe_key) + ) WITHOUT ROWID;", + ) + .unwrap(); + legacy + .execute( + "INSERT INTO event_dedupe(flow_key, subscription_id, dedupe_key, run_id) + VALUES ('f', 's', 'k', 'run-legacy')", + [], + ) + .unwrap(); + } + + let registry = Registry::open(&path).unwrap(); + // The legacy row carries boot_id '' -- no live boot -- so it is a dead + // process's claim and must be repaired, not treated as in flight. + assert_eq!( + registry.claim_event("f", "s", "k", "run-new", "boot-1").unwrap(), + None, + "a pre-migration claim with no run belongs to no live boot" + ); + } + + /// A registered run dedupes regardless of which boot claimed it. + #[test] + fn a_registered_run_dedupes_across_boots() { + let directory = tempdir().unwrap(); + let registry = Registry::open(directory.path().join("r.sqlite3")).unwrap(); + + assert_eq!( + registry + .claim_event("f", "s", "k", "run-a", "boot-1") + .unwrap(), + None + ); + registry + .register("run-a", std::path::Path::new("/tmp/run-a.sqlite3")) + .unwrap(); + assert_eq!( + registry + .claim_event("f", "s", "k", "run-b", "boot-2") + .unwrap(), + Some("run-a".to_string()) + ); + } + + /// Releasing a claim this boot could not turn into a run must free the + /// event, or the same-boot rule above would strand it inside the process. + #[test] + fn releasing_a_claim_lets_the_same_boot_retry() { + let directory = tempdir().unwrap(); + let registry = Registry::open(directory.path().join("r.sqlite3")).unwrap(); + + assert_eq!( + registry + .claim_event("f", "s", "k", "run-a", "boot-1") + .unwrap(), + None + ); + registry.release_claim("f", "s", "k", "run-a").unwrap(); + assert_eq!( + registry + .claim_event("f", "s", "k", "run-b", "boot-1") + .unwrap(), + None, + "a released claim must not keep dedupe-ing the event" + ); + } + + /// A release must not steal a claim some other delivery legitimately holds. + #[test] + fn releasing_is_scoped_to_the_claiming_run() { + let directory = tempdir().unwrap(); + let registry = Registry::open(directory.path().join("r.sqlite3")).unwrap(); + + assert_eq!( + registry + .claim_event("f", "s", "k", "run-a", "boot-1") + .unwrap(), + None + ); + registry.release_claim("f", "s", "k", "run-stale").unwrap(); + assert_eq!( + registry + .claim_event("f", "s", "k", "run-b", "boot-1") + .unwrap(), + Some("run-a".to_string()), + "releasing a run that does not hold the claim must be a no-op" + ); + } + fn open_registry() -> (tempfile::TempDir, Registry) { let directory = tempdir().unwrap(); let registry = Registry::open(directory.path().join("relayflowd.sqlite3")).unwrap(); @@ -208,5 +508,4 @@ mod tests { assert_eq!(record.file, run_file); assert_eq!(record.status, "completed"); } - } diff --git a/kernel/relayflowd-journal/src/subscriptions.rs b/kernel/relayflowd-journal/src/subscriptions.rs index 22dcfe493..856dd009d 100644 --- a/kernel/relayflowd-journal/src/subscriptions.rs +++ b/kernel/relayflowd-journal/src/subscriptions.rs @@ -148,7 +148,13 @@ impl Registry { stale_after_ms = excluded.stale_after_ms, last_event_at_ms = excluded.last_event_at_ms, stale_at_ms = NULL", - params![flow_key, subscription_id, event_type, stale_after_ms, now_ms], + params![ + flow_key, + subscription_id, + event_type, + stale_after_ms, + now_ms + ], )?; Ok(()) } @@ -232,7 +238,12 @@ impl Registry { WHERE flow_key = ?1 AND subscription_id = ?2 AND last_event_at_ms = ?3", - params![flow_key, subscription_id, detected_last_event_at_ms, stale_at_ms], + params![ + flow_key, + subscription_id, + detected_last_event_at_ms, + stale_at_ms + ], )?; Ok(changed > 0) } @@ -276,7 +287,12 @@ mod tests { let rows = registry.detect_stale(sweep_id, worker, now_ms).unwrap(); for row in &rows { registry - .latch_stale(&row.flow_key, &row.subscription_id, row.last_event_at_ms, now_ms) + .latch_stale( + &row.flow_key, + &row.subscription_id, + row.last_event_at_ms, + now_ms, + ) .unwrap(); } rows @@ -305,7 +321,10 @@ mod tests { let first = sweep_and_latch(®istry, "bucket-1", "w", 1_030_500); assert_eq!(first.len(), 1); let second = sweep_and_latch(®istry, "bucket-2", "w", 1_060_500); - assert!(second.is_empty(), "sweep re-emitted a latched row: {second:?}"); + assert!( + second.is_empty(), + "sweep re-emitted a latched row: {second:?}" + ); } #[test] @@ -377,7 +396,10 @@ mod tests { let winner = sweep_and_latch(®istry, "bucket-42", "w1", 1_040_000); assert_eq!(winner.len(), 1); let loser = sweep_and_latch(®istry, "bucket-42", "w2", 1_040_100); - assert!(loser.is_empty(), "second caller in same bucket won: {loser:?}"); + assert!( + loser.is_empty(), + "second caller in same bucket won: {loser:?}" + ); } #[test] @@ -437,12 +459,24 @@ mod tests { // regressions (rowid ASC or DESC) or an omitted ORDER BY all // fail — only `ORDER BY r.run_id DESC` passes. let (dir, registry) = open_registry(); - registry.claim_event("flow", "sub", "k1", "01AAAAAA").unwrap(); - registry.register("01AAAAAA", &dir.path().join("a.sqlite3")).unwrap(); - registry.claim_event("flow", "sub", "k2", "01ZZZZZZ").unwrap(); - registry.register("01ZZZZZZ", &dir.path().join("z.sqlite3")).unwrap(); - registry.claim_event("flow", "sub", "k3", "01MMMMMM").unwrap(); - registry.register("01MMMMMM", &dir.path().join("m.sqlite3")).unwrap(); + registry + .claim_event("flow", "sub", "k1", "01AAAAAA", "boot") + .unwrap(); + registry + .register("01AAAAAA", &dir.path().join("a.sqlite3")) + .unwrap(); + registry + .claim_event("flow", "sub", "k2", "01ZZZZZZ", "boot") + .unwrap(); + registry + .register("01ZZZZZZ", &dir.path().join("z.sqlite3")) + .unwrap(); + registry + .claim_event("flow", "sub", "k3", "01MMMMMM", "boot") + .unwrap(); + registry + .register("01MMMMMM", &dir.path().join("m.sqlite3")) + .unwrap(); let record = registry .last_run_for_subscription("flow", "sub") .unwrap() diff --git a/kernel/relayflowd/src/engine.rs b/kernel/relayflowd/src/engine.rs index 91d6796d7..cf9e441ec 100644 --- a/kernel/relayflowd/src/engine.rs +++ b/kernel/relayflowd/src/engine.rs @@ -60,17 +60,50 @@ pub struct CancelOptions { pub pause_after_request: bool, } +/// The identity of this PROCESS, not of an `Engine`. +/// +/// This distinction is the whole of the dedupe rule. `claim_event` treats a +/// claim carrying this id as a run that is in flight, and any other id as +/// wreckage from a dead process. Generating it per `Engine` would make that +/// rule inert in production: the server constructs a fresh +/// `Engine::with_runtime` inside `handle_request` (server.rs), so two +/// concurrent `event.submit` calls would hold two different ids, and the second +/// would "repair" the first's live claim and spawn a duplicate run -- exactly +/// the bug this is supposed to close. +/// +/// Process-wide is also the correct semantics on its own terms: a claim is +/// abandoned when the process that took it is gone, and nothing smaller than a +/// process can die. +fn new_boot_id() -> String { + static BOOT_ID: std::sync::OnceLock = std::sync::OnceLock::new(); + BOOT_ID.get_or_init(|| Ulid::new().to_string()).clone() +} + pub struct Engine { data_dir: PathBuf, + /// Identifies this PROCESS. Written onto every event claim so the dedupe + /// repair can tell a claim abandoned by a dead process from one held by a + /// delivery that is in flight right now (#160). See `new_boot_id`. + boot_id: String, clock: C, dispatcher: Option>, observer: Option>, } impl Engine { + /// Open an engine over `data_dir`. + /// + /// Any number of `Engine`s may exist over one `data_dir` in one process, + /// which is what production does -- the server builds one per protocol + /// request. They share a `boot_id` because it identifies the process, so + /// concurrent deliveries see each other's claims as in flight rather than + /// as wreckage. An earlier revision generated it per `Engine` and stated + /// the opposite invariant here; that was wrong, and it made the #160 fix + /// inert under the only topology that matters. pub fn new(data_dir: impl Into) -> Self { Self { data_dir: data_dir.into(), + boot_id: new_boot_id(), clock: WallClock, dispatcher: None, observer: None, @@ -84,6 +117,7 @@ impl Engine { ) -> Self { Self { data_dir: data_dir.into(), + boot_id: new_boot_id(), clock: WallClock, dispatcher: Some(dispatcher), observer: Some(observer), @@ -95,6 +129,7 @@ impl Engine { pub fn with_clock(data_dir: impl Into, clock: C) -> Self { Self { data_dir: data_dir.into(), + boot_id: new_boot_id(), clock, dispatcher: None, observer: None, @@ -438,6 +473,10 @@ impl Engine { Registry::open(self.data_dir.join("relayflowd.sqlite3")).context("open run registry") } + pub(super) fn boot_id(&self) -> &str { + &self.boot_id + } + /// The conventional location of a run's journal. Single source of truth: /// the resume-repair path in `server.rs` opens by this too, so a change to /// the layout cannot leave the two disagreeing. @@ -503,3 +542,41 @@ pub fn read_spec(path: &Path) -> Result { RunSpec::parse(&value).with_context(|| format!("parse run spec {}", path.display()))?; Ok(spec) } + +#[cfg(test)] +mod boot_identity_tests { + use super::*; + + /// The boot id must identify the PROCESS, not an `Engine`. + /// + /// Production builds a fresh `Engine::with_runtime` inside + /// `handle_request`, so two concurrent `event.submit` calls hold two + /// different `Engine`s over one data dir. `claim_event` treats a claim from + /// a different boot as wreckage to be repaired, so a per-`Engine` id would + /// let the second delivery take over the first's LIVE claim and spawn a + /// duplicate run -- leaving #160 fixed only in tests that happen to share + /// an Engine. + /// + /// This is asserted here rather than left to the racing integration test, + /// which measured only 2 catches in 20 against a per-`Engine` id: the + /// property is deterministic, so its gate should be too. + #[test] + fn every_engine_in_this_process_shares_one_boot_id() { + let a = tempfile::tempdir().unwrap(); + let b = tempfile::tempdir().unwrap(); + + // Different directories, different constructors -- still one process. + let first = Engine::new(a.path()); + let second = Engine::new(b.path()); + let third = Engine::with_clock(a.path(), WallClock); + + assert_eq!( + first.boot_id(), + second.boot_id(), + "two Engines in one process must share a boot id, or concurrent \ + deliveries repair each other's live claims" + ); + assert_eq!(first.boot_id(), third.boot_id()); + assert!(!first.boot_id().is_empty()); + } +} diff --git a/kernel/relayflowd/src/engine/wake.rs b/kernel/relayflowd/src/engine/wake.rs index df7053173..b8d9e777a 100644 --- a/kernel/relayflowd/src/engine/wake.rs +++ b/kernel/relayflowd/src/engine/wake.rs @@ -1,5 +1,7 @@ use anyhow::{Context, Result}; -use relayflowd_core::{Clock, EntryType, Event, JournalEntry, RunSpawnedPayload, RunSpec}; +use relayflowd_core::{ + Clock, EntryType, Event, JournalEntry, RunSpawnedPayload, RunSpec, TriggerSpec, +}; use relayflowd_journal::SqliteJournal; use serde::{Deserialize, Serialize}; use serde_json::json; @@ -60,9 +62,15 @@ impl Engine { let run_id = Ulid::new().to_string(); // Scope the claim to this flow and subscription. A bare key is // globally unique, so two flows deriving the same key would suppress - // one another; and the registry repairs a claim whose run was never - // registered, so a crash between claim and journal creation no longer - // loses the event. + // one another. + // + // A claim whose run was never registered is recovered two ways, and + // which one applies depends on whether the claiming process is still + // alive: a claim left by a PREVIOUS boot is repaired by the registry + // (see `claim_event`), while one this boot cannot turn into a run is + // handed back explicitly on the failure path below. Before #160 the + // registry did both, which is what let a concurrent delivery mistake + // an in-flight claim for wreckage. // The flow's stable identity is the canonical hash of its spec — the // same value the engine already journals as `spec_hash`. RunSpec has // no id, and `name` is optional, so two unnamed flows would collide on @@ -99,7 +107,7 @@ impl Engine { )?; if self .registry()? - .claim_event(&flow_key, &trigger.id, &event_key, &run_id)? + .claim_event(&flow_key, &trigger.id, &event_key, &run_id, self.boot_id())? .is_some() { return Ok(EventSubmitOutcome { @@ -109,16 +117,81 @@ impl Engine { run: None, }); } - let path = self.run_path(&run_id); + // The claim is now held by this boot, and `claim_event` will tell any + // concurrent delivery that the event is a duplicate on the strength of + // it. That obliges us to either turn it into a registered run or hand + // it back: a claim kept without a run strands the event inside this + // process. + // + // The span that carries that obligation is a named method rather than + // an inline block, so its boundary is unmistakable. Work added to + // `spawn_claimed_run` is covered by the release below; work added after + // the call is not, and that distinction is the whole invariant. + let spawned = self.spawn_claimed_run( + &spec, + &event, + &trigger, + &event_key, + &run_id, + stale_after_ms, + created_by, + ); + // Released only on the failure path. Once `register` has succeeded the + // run is discoverable, so a later failure is a run that exists and + // needs resuming -- not a claim to hand back. + // + // TODO(#173): a PANIC between the claim and `register` unwinds past + // this match without releasing, stranding the event for the life of the + // boot (a restart clears it -- the claim is then a previous boot's and + // gets repaired). A Drop guard closes it, and is deliberately a + // separate change: a Drop that opens a database and cannot report its + // own failure is a mechanism with its own failure modes, and bundling + // it here would ship it unreviewed under a fix for something else. + let journal = match spawned { + Ok(journal) => journal, + Err(error) => { + self.registry()? + .release_claim(&flow_key, &trigger.id, &event_key, &run_id)?; + return Err(error); + } + }; + let run = self.drive(journal, spec, DriveOptions::default())?; + Ok(EventSubmitOutcome { + matched: true, + deduped: false, + subscription_id: Some(trigger.id), + run: Some(run), + }) + } + + /// Turn a claim this boot already holds into a registered run. + /// + /// Everything here is covered by the caller's release-on-failure path: if + /// this returns `Err`, the claim is handed back. Once `register` has + /// succeeded the run is discoverable and the claim is permanent, so + /// `register` is deliberately the LAST thing this does. Adding work after + /// it would put that work outside the release guarantee. + #[allow(clippy::too_many_arguments)] + fn spawn_claimed_run( + &self, + spec: &RunSpec, + event: &Event, + trigger: &TriggerSpec, + event_key: &str, + run_id: &str, + stale_after_ms: i64, + created_by: &str, + ) -> Result { + let path = self.run_path(run_id); let now_ms = self.clock.now_ms(); let mut journal = - SqliteJournal::create(&path, &run_id, now_ms).context("create event run journal")?; - let spec_value = serde_json::to_value(&spec)?; + SqliteJournal::create(&path, run_id, now_ms).context("create event run journal")?; + let spec_value = serde_json::to_value(spec)?; self.append( &mut journal, &JournalEntry::new( EntryType::RunSpawned, - &run_id, + run_id, None, None, now_ms, @@ -137,24 +210,18 @@ impl Engine { // the engine default. Without this, `stale_after_ms=None` in // the spec was silently indistinguishable from `stale_after_ms: // DEFAULT_STALE_AFTER_MS` at the alert boundary. - self.append(&mut journal, &JournalEntry::new(EntryType::SubscriptionRegistered, &run_id, None, None, now_ms, - json!({"subscription_id": trigger.id, "event_type": event.event_type, "executor": trigger.executor, - "effective_stale_after_ms": stale_after_ms})))?; - self.append(&mut journal, &JournalEntry::new(EntryType::EventReceived, &run_id, None, None, now_ms, - json!({"event": event, "event_key": event_key, "matched_subscription_id": trigger.id})))?; - self.append(&mut journal, &JournalEntry::new(EntryType::SubscriptionMatched, &run_id, None, None, now_ms, - json!({"subscription_id": trigger.id, "event_key": event_key, - "wake_context": {"epoch_summary": {"open_steps": spec.steps.iter().map(|step| &step.id).collect::>()}, - "triggering_event": event}})))?; + self.append(&mut journal, &JournalEntry::new(EntryType::SubscriptionRegistered, run_id, None, None, now_ms, + json!({"subscription_id": trigger.id, "event_type": event.event_type, "executor": trigger.executor, + "effective_stale_after_ms": stale_after_ms})))?; + self.append(&mut journal, &JournalEntry::new(EntryType::EventReceived, run_id, None, None, now_ms, + json!({"event": event, "event_key": event_key, "matched_subscription_id": trigger.id})))?; + self.append(&mut journal, &JournalEntry::new(EntryType::SubscriptionMatched, run_id, None, None, now_ms, + json!({"subscription_id": trigger.id, "event_key": event_key, + "wake_context": {"epoch_summary": {"open_steps": spec.steps.iter().map(|step| &step.id).collect::>()}, + "triggering_event": event}})))?; self.registry()? - .register(&run_id, &path) + .register(run_id, &path) .context("register event run")?; - let run = self.drive(journal, spec, DriveOptions::default())?; - Ok(EventSubmitOutcome { - matched: true, - deduped: false, - subscription_id: Some(trigger.id), - run: Some(run), - }) + Ok(journal) } } diff --git a/kernel/relayflowd/tests/event_wake.rs b/kernel/relayflowd/tests/event_wake.rs index a33ea2b1d..c24b62b77 100644 --- a/kernel/relayflowd/tests/event_wake.rs +++ b/kernel/relayflowd/tests/event_wake.rs @@ -48,3 +48,100 @@ fn matching_event_wakes_once_with_fresh_context() { assert!(second.matched && second.deduped); assert!(second.run.is_none()); } + +/// #160. Two deliveries of ONE event, racing, must produce exactly one run. +/// +/// This is the exactly-once invariant at its sharpest, and it failed before the +/// `boot_id` fix: `claim_event` repaired any claim with no matching `runs` row, +/// which is true both of a run that crashed and of a run that is mid-flight and +/// has not registered yet. The second delivery saw the first inside that window, +/// judged its claim abandoned, took it over and spawned a run of its own. +/// +/// Topology matters here. An earlier version of this test used two `Engine`s +/// over one directory and failed with `database is locked` -- registry-open +/// contention, which is a different defect and would have passed the dedupe path +/// without ever exercising it. One shared `Engine` and two threads is what puts +/// two deliveries into the claim window at the same time. +#[test] +fn two_racing_deliveries_of_one_event_produce_exactly_one_run() { + let directory = tempfile::tempdir().unwrap(); + let engine = Engine::new(directory.path()); + let value = serde_json::from_str(include_str!(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../testdata/event-triggered-flow.spec.canonical.json" + ))) + .unwrap(); + let spec = RunSpec::parse(&value).unwrap(); + let event = Event { + event_type: "test.ping".into(), + payload: json!({"message":"hello"}), + key: None, + }; + + // Each racer builds its OWN Engine, because that is what production does: + // the server constructs a fresh `Engine::with_runtime` inside + // `handle_request`, so two concurrent `event.submit` calls never share one. + // An earlier version of this test shared a single Engine and passed while + // the fix was inert under the real topology -- the boot id was per-Engine, + // so two production deliveries held different ids and the second treated + // the first's live claim as wreckage. Sharing an Engine here hid exactly + // the bug the test exists to catch. + // + // A barrier, not just two spawns, so both threads enter the claim window + // together rather than whenever the scheduler reaches them. + // + // Be honest about what this test is worth. Against a deliberately broken + // guard it caught the bug 3 times in 30 -- the winner usually registers + // before any loser reads, and no amount of threads changes that because + // SQLite serialises the writes. It exercises the real path and cannot + // false-positive, which is why it is here, but it is NOT the gate. + // `registry::tests` asserts the same rule deterministically. + const RACERS: usize = 2; + let gate = std::sync::Barrier::new(RACERS); + let outcomes: Vec<_> = std::thread::scope(|scope| { + let handles: Vec<_> = (0..RACERS) + .map(|_| { + let gate = &gate; + let directory = directory.path(); + let spec = spec.clone(); + let event = event.clone(); + scope.spawn(move || { + let engine = Engine::new(directory); + gate.wait(); + engine.submit_event(spec, event, "test").unwrap() + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + + assert!( + outcomes.iter().all(|outcome| outcome.matched), + "both deliveries match the same trigger" + ); + let spawned = outcomes.iter().filter(|o| !o.deduped).count(); + assert_eq!( + spawned, 1, + "exactly one racing delivery may start a run, got {spawned}" + ); + assert_eq!( + outcomes.iter().filter(|o| o.deduped).count(), + RACERS - 1, + "every losing delivery must be reported deduped, not dropped" + ); + // A deduped outcome carries no run, and the winner's run must be real -- + // the pre-fix bug produced two runs, so counting outcomes alone could pass + // while the registry held two. + let winner = outcomes.iter().find(|o| !o.deduped).unwrap(); + let run_id = &winner.run.as_ref().unwrap().run_id; + let received = engine + .journal_entries(run_id, 1, 100) + .unwrap() + .into_iter() + .filter(|entry| entry.entry_type == EntryType::EventReceived) + .count(); + assert_eq!( + received, 1, + "the surviving run records the event exactly once" + ); +}