From 61803e5a2e8b5e3f144ec31ebbe58a34481a5d98 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Sat, 29 Aug 2026 04:12:09 -0400 Subject: [PATCH 1/2] drive: cloud run e1d7225d Work produced by cloud run e1d7225d-2e0a-4919-be38-b9dde5727315 in a workflow sandbox and delivered from this host, because a sandbox has no remote and no GitHub token. Verification and adversarial review ran in-run; see ops/reviews/ in the diff. --- kernel/relayflowd/src/server.rs | 13 ++-- kernel/relayflowd/src/server/tests.rs | 102 +++++++++++++++++++++----- 2 files changed, 93 insertions(+), 22 deletions(-) diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index 6a44012ba..98ba5accd 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -413,9 +413,10 @@ fn handle_request( /// Register the watcher BEFORE replaying, then replay and hand the hub the /// sequence cursor the replay covered. Entries appended concurrently are /// buffered by the hub and flushed deduped against that cursor, so nothing -/// appended between snapshot and registration can be missed — that ordering -/// gap no longer exists. `after_register` is a test seam pinning the race -/// window between registration and the snapshot read. +/// appended between snapshot and registration can be missed. The run lock +/// also keeps an append's journal commit and hub notification atomic with +/// respect to registration, preventing a replayed entry from later arriving +/// as live. `after_ready` is a test seam pinning that notification ordering. #[cfg(unix)] fn watch_with_replay( engine: &Engine, @@ -423,10 +424,11 @@ fn watch_with_replay( connection_id: u64, run_id: &str, writer: &SharedWriter, - after_register: impl FnOnce(), + after_ready: impl FnOnce(), ) -> ProtocolResult { + let lock = hub.run_lock(run_id); + let _guard = lock.lock().expect("run lock"); hub.watch(connection_id, run_id.to_owned(), writer.clone()); - after_register(); let replay = (|| -> Result<()> { let entries = engine.journal_entries(run_id, 1, usize::MAX)?; let replayed_through_seq = entries.last().map(|entry| entry.seq).unwrap_or(0); @@ -434,6 +436,7 @@ fn watch_with_replay( write_frame(writer, &json!({"event": "entry", "data": entry}))?; } hub.watch_ready(connection_id, run_id, replayed_through_seq); + after_ready(); Ok(()) })(); if let Err(error) = replay { diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index 05d9dfd71..01b714d91 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -2,7 +2,8 @@ use std::{ io::{BufRead, BufReader}, os::unix::net::UnixStream, path::Path, - sync::{Arc, Mutex}, + sync::{Arc, Condvar, Mutex, mpsc}, + thread, time::Duration, }; @@ -11,7 +12,7 @@ use serde_json::json; use tempfile::tempdir; use super::*; -use crate::worker::LeaseProbe; +use crate::worker::{JournalObserver, LeaseProbe}; mod agent; @@ -86,6 +87,41 @@ fn step_completions(data_dir: &Path, run_id: &str) -> Vec .collect() } +struct PausingObserver { + hub: Arc, + committed: Arc<(Mutex, Condvar)>, + resume: Arc<(Mutex, Condvar)>, +} + +impl JournalObserver for PausingObserver { + fn appended(&self, entry: &relayflowd_core::JournalEntry) { + let (committed, committed_signal) = &*self.committed; + *committed.lock().unwrap() = true; + committed_signal.notify_one(); + + let (resume, resume_signal) = &*self.resume; + let mut ready = resume.lock().unwrap(); + while !*ready { + ready = resume_signal.wait(ready).unwrap(); + } + self.hub.appended(entry); + } +} + +fn wait_for_signal(signal: &Arc<(Mutex, Condvar)>) { + let (ready, condition) = &**signal; + let mut ready = ready.lock().unwrap(); + while !*ready { + ready = condition.wait(ready).unwrap(); + } +} + +fn send_signal(signal: &Arc<(Mutex, Condvar)>) { + let (ready, condition) = &**signal; + *ready.lock().unwrap() = true; + condition.notify_one(); +} + #[test] fn hello_enforces_protocol_version() { let directory = tempdir().unwrap(); @@ -202,9 +238,9 @@ fn stopped_heartbeats_past_the_deadline_journal_lease_expired_and_release_the_st assert_eq!(redispatch["data"]["attempt"], 2); } -/// Finding 4: an entry appended concurrently with `run.watch` registration is -/// delivered exactly once — the watcher registers with a cursor before the -/// replay, and buffered live entries are deduped against it. +/// Finding 4: an entry committed before `run.watch` registration but whose +/// hub notification is delayed is delivered exactly once. The run lock makes +/// the journal commit and notification indivisible from watch registration. #[test] fn an_entry_appended_during_watch_registration_is_delivered_exactly_once() { let directory = tempdir().unwrap(); @@ -224,26 +260,58 @@ fn an_entry_appended_during_watch_registration_is_delivered_exactly_once() { .to_owned(); let dispatcher: Arc = hub.clone(); - let observer: Arc = hub.clone(); + let observer: Arc = hub.clone(); let engine = Engine::with_runtime(data_dir, dispatcher, observer); - let interleaver = { - let dispatcher: Arc = hub.clone(); - let observer: Arc = hub.clone(); - Engine::with_runtime(data_dir, dispatcher, observer) - }; - let (watch_writer, watch_peer) = shared_writer(); - let run = run_id.clone(); - let result = watch_with_replay(&engine, &hub, 3, &run_id, &watch_writer, || { - // The historical race window: an append interleaved with watch setup. + let committed = Arc::new((Mutex::new(false), Condvar::new())); + let resume = Arc::new((Mutex::new(false), Condvar::new())); + let interleaver = Engine::with_runtime( + data_dir, + hub.clone(), + Arc::new(PausingObserver { + hub: hub.clone(), + committed: committed.clone(), + resume: resume.clone(), + }), + ); + + let append_run = run_id.clone(); + let append_hub = hub.clone(); + let append = thread::spawn(move || { + let lock = append_hub.run_lock(&append_run); + let _guard = lock.lock().unwrap(); interleaver - .append_stream(&run, "results", "test", json!({"interleaved": true})) + .append_stream( + &append_run, + "results", + "test", + json!({"interleaved": true}), + ) .unwrap(); }); + wait_for_signal(&committed); + + let (watch_writer, watch_peer) = shared_writer(); + let watch_hub = hub.clone(); + let watch_run = run_id.clone(); + let (watch_ready, watch_is_ready) = mpsc::channel(); + let watch = thread::spawn(move || { + watch_with_replay(&engine, &watch_hub, 3, &watch_run, &watch_writer, || { + watch_ready.send(()).unwrap(); + }) + }); + // Without the run lock, registration reaches Live while the committed + // append's hub notification is still paused. With the lock, this times out + // because registration correctly waits for that notification to finish. + let _ = watch_is_ready.recv_timeout(Duration::from_millis(100)); + send_signal(&resume); + append.join().unwrap(); + let result = watch.join().unwrap(); assert!(result.is_ok(), "run.watch failed: {result:?}"); // A post-registration append must flow through live delivery, once. - engine + let live_engine = Engine::with_runtime(data_dir, hub.clone(), hub.clone()); + live_engine .append_stream(&run_id, "results", "test", json!({"live": true})) .unwrap(); From 05cb7d93e2ac7da42fc4fa70a3432a9355f59dfb Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Sat, 29 Aug 2026 06:12:17 -0400 Subject: [PATCH 2/2] fix: release the run lock before writing replay frames (PR #18 P1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review was right. The first version of this fix held the run lock for the whole of watch_with_replay, including the loop that writes every historical entry through a blocking UnixStream. A slow or stalled reader would have pinned that lock for the duration, blocking every other operation on the run — a fix for a race that introduced a head-of-line block. The lock now covers registration and the snapshot read and nothing else. That is all it needs to cover: the guarantee required is that registration and the snapshot are atomic with respect to an append, so no entry can slip between them. Once both have happened the set is fixed, and live entries arriving during the writes are buffered by the hub and flushed deduped against replayed_through_seq. The error path drops the guard before unwatching rather than holding it across that call too. Verified: kernel 19+19+1+1+26+5+6 passed, 0 failed, including an_entry_appended_during_watch_registration_is_delivered_exactly_once. Co-Authored-By: Claude Fable 5 --- kernel/relayflowd/src/server.rs | 29 +++++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/kernel/relayflowd/src/server.rs b/kernel/relayflowd/src/server.rs index 98ba5accd..dc17615ed 100644 --- a/kernel/relayflowd/src/server.rs +++ b/kernel/relayflowd/src/server.rs @@ -426,11 +426,32 @@ fn watch_with_replay( writer: &SharedWriter, after_ready: impl FnOnce(), ) -> ProtocolResult { - let lock = hub.run_lock(run_id); - let _guard = lock.lock().expect("run lock"); - hub.watch(connection_id, run_id.to_owned(), writer.clone()); + // The lock covers registration and the snapshot read, and NOTHING ELSE. + // Holding it across the replay writes would pin it for the duration of a + // blocking UnixStream write per historical entry — a slow or stalled + // reader would then block every other operation on this run. Review caught + // that (PR #18, P1) on the first version of this fix. + // + // Releasing before the writes is safe: what the lock must guarantee is + // that registration and the snapshot are atomic with respect to an + // append, so no entry can slip between them. Once both have happened the + // set is fixed. Live entries arriving during the writes are buffered by + // the hub and flushed deduped against `replayed_through_seq`. + let entries = { + let lock = hub.run_lock(run_id); + let _guard = lock.lock().expect("run lock"); + hub.watch(connection_id, run_id.to_owned(), writer.clone()); + match engine.journal_entries(run_id, 1, usize::MAX) { + Ok(entries) => entries, + Err(error) => { + drop(_guard); + hub.unwatch(connection_id, run_id); + return Err(internal_error(error)); + } + } + }; + let replay = (|| -> Result<()> { - let entries = engine.journal_entries(run_id, 1, usize::MAX)?; let replayed_through_seq = entries.last().map(|entry| entry.seq).unwrap_or(0); for entry in entries { write_frame(writer, &json!({"event": "entry", "data": entry}))?;