diff --git a/harness/src/clients/engine.rs b/harness/src/clients/engine.rs index dc096a75b..095c85014 100644 --- a/harness/src/clients/engine.rs +++ b/harness/src/clients/engine.rs @@ -9,6 +9,12 @@ use iii_sdk::protocol::TriggerRequest; use iii_sdk::IIIClient; use serde_json::{json, Value}; +/// The message a dispatch interrupted by an engine restart resolves with — +/// it becomes the synthesized error `function_result` that closes the call +/// in the transcript (iii-hq/workers#507). +pub const ENGINE_RESTART_INTERRUPTED: &str = + "execution interrupted by engine restart; the call ran at most once and its result is unknown"; + /// A registry function descriptor (id + optional schemas/description). Only /// the fields the harness reads are typed; the rest pass through as `extra`. #[derive(Debug, Clone)] @@ -46,25 +52,51 @@ impl EngineClient { /// Dispatch an arbitrary iii function and return its raw result. This is /// the target invocation of an unwrapped `agent_trigger` call. + /// + /// An engine restart while the call is in flight fails the dispatch with + /// `engine_restart` instead of waiting out the full timeout: the engine's + /// in-memory invocation routing died with it, so the result can never be + /// delivered, and the caller must close the interrupted call promptly to + /// keep the session usable (iii-hq/workers#507). pub async fn dispatch( &self, function_id: &str, payload: Value, ) -> Result { - self.iii - .trigger(TriggerRequest { - function_id: function_id.to_string(), - payload, - action: None, - timeout_ms: Some(self.timeout_ms), - }) - .await - .map_err(|e| { - let raw = e.to_string(); - let mut parsed = parse_dispatch_error_message(&raw); - parsed.message = format!("{function_id}: {}", parsed.message); - parsed - }) + // Capture the epoch immediately before the invocation instead of + // comparing against process-global state. A global baseline becomes + // stale when the engine restarts while this worker is idle and would + // falsely interrupt the first slow call made afterwards. + let baseline = engine_epoch_ms(&self.iii).await; + let call = self.iii.trigger(TriggerRequest { + function_id: function_id.to_string(), + payload, + action: None, + timeout_ms: Some(self.timeout_ms), + }); + let result = match baseline { + Some(baseline) => { + tokio::select! { + result = call => result, + () = engine_link_interrupted(&self.iii, baseline) => { + return Err(DispatchError { + code: Some("engine_restart".to_string()), + message: format!("{function_id}: {ENGINE_RESTART_INTERRUPTED}"), + }); + }, + } + } + // If the preflight epoch cannot be read, do not guess. The normal + // SDK timeout/error is safer than declaring an invocation + // interrupted from an untrusted baseline and re-running it. + None => call.await, + }; + result.map_err(|e| { + let raw = e.to_string(); + let mut parsed = parse_dispatch_error_message(&raw); + parsed.message = format!("{function_id}: {}", parsed.message); + parsed + }) } /// List registry function descriptors (best-effort; empty on failure). @@ -130,6 +162,66 @@ impl EngineClient { } } +/// How often an in-flight dispatch samples the engine epoch, and how long +/// one sample may take. Sampling is cheap relative to the dispatched calls +/// it guards; dispatches shorter than the first interval never even probe. +const ENGINE_EPOCH_PROBE_INTERVAL_MS: u64 = 1_000; +const ENGINE_EPOCH_PROBE_TIMEOUT_MS: u64 = 3_000; + +/// The engine's boot identity: the earliest `connected_at_ms` among its +/// in-process (`runtime == "engine"`) workers, which attach once at engine +/// startup. A restarted engine reports a new epoch — the signal that every +/// invocation in flight across the restart lost its result routing. `None` +/// when the engine cannot be reached (outage in progress) or the response +/// shape is unrecognized. +async fn engine_epoch_ms(iii: &IIIClient) -> Option { + let response = iii + .trigger(TriggerRequest { + function_id: "engine::workers::list".to_string(), + payload: json!({}), + action: None, + timeout_ms: Some(ENGINE_EPOCH_PROBE_TIMEOUT_MS), + }) + .await + .ok()?; + parse_engine_epoch(&response) +} + +fn parse_engine_epoch(response: &Value) -> Option { + response + .get("workers")? + .as_array()? + .iter() + .filter(|worker| worker.get("runtime").and_then(Value::as_str) == Some("engine")) + .filter_map(|worker| worker.get("connected_at_ms").and_then(Value::as_u64)) + .min() +} + +/// Pend until the engine under an in-flight dispatch is observed to have +/// RESTARTED (its boot epoch changed). Neither the SDK connection state nor +/// plain liveness probes can see a fast restart: the reconnect loop reports +/// `Connected` through its silent 2s retry sleep, and outbound messages +/// buffered during the outage are answered by the NEW engine as if nothing +/// happened (both verified against a SIGKILLed-and-respawned engine). An +/// epoch read that succeeds with a changed value doubles as proof the +/// engine is answering again, so the caller can immediately persist the +/// synthesized "interrupted" result without tripping over the same outage. +async fn engine_link_interrupted(iii: &IIIClient, baseline: u64) { + loop { + tokio::time::sleep(std::time::Duration::from_millis( + ENGINE_EPOCH_PROBE_INTERVAL_MS, + )) + .await; + // An unreadable epoch (outage in progress) never trips by itself: + // the first successful sample afterwards decides. + if let Some(epoch) = engine_epoch_ms(iii).await { + if epoch != baseline { + return; + } + } + } +} + fn parse_dispatch_error_message(raw: &str) -> DispatchError { if let Some(parsed) = parse_dispatch_error_value(raw) { return parsed; @@ -253,6 +345,20 @@ fn descriptor_of(v: &Value) -> Option { mod tests { use super::*; + #[test] + fn engine_epoch_uses_the_oldest_in_process_engine_worker() { + let response = json!({ + "workers": [ + { "runtime": "rust", "connected_at_ms": 1 }, + { "runtime": "engine", "connected_at_ms": 30 }, + { "runtime": "engine", "connected_at_ms": 20 }, + { "runtime": "engine", "connected_at_ms": "invalid" } + ] + }); + assert_eq!(parse_engine_epoch(&response), Some(20)); + assert_eq!(parse_engine_epoch(&json!({ "workers": [] })), None); + } + #[test] fn dispatch_error_parser_extracts_plain_json_code_and_message() { let err = parse_dispatch_error_message( diff --git a/harness/src/functions/turn.rs b/harness/src/functions/turn.rs index 8391694ed..1f4344869 100644 --- a/harness/src/functions/turn.rs +++ b/harness/src/functions/turn.rs @@ -74,13 +74,62 @@ pub async fn handle(deps: &Deps, payload: TurnStepPayload) -> Result bool { + let message = match error { + HarnessError::Dependency(m) => m.to_ascii_lowercase(), + // Retrying a read from the top of the step is safe. Do not classify + // state writes broadly: a lost acknowledgement after a successful + // write must not cause the whole step to repeat side effects. + HarnessError::State(m) if m.to_ascii_lowercase().starts_with("state::get ") => { + m.to_ascii_lowercase() + } + _ => return false, + }; + if message.contains("enqueue harness::turn") { + return false; + } + message.contains("function_not_found") || message.contains("not connected") +} + async fn run(deps: &Deps, payload: TurnStepPayload) -> Result { let (session_id, turn_id) = (payload.session_id.clone(), payload.turn_id.clone()); - let result = match turn_loop::run_step(deps, payload).await { - Ok(result) => result, - Err(e) => { - tracing::error!(session_id = %session_id, turn_id = %turn_id, error = %e, "turn step failed; finalising turn as failed"); - turn_loop::fail_turn(deps, &session_id, &turn_id, &e.to_string()).await + let mut transient_attempts = 0u32; + let result = loop { + match turn_loop::run_step(deps, payload.clone()).await { + Ok(result) => break result, + Err(e) + if transient_attempts < TRANSIENT_STEP_RETRIES && is_transient_step_error(&e) => + { + transient_attempts += 1; + tracing::warn!( + session_id = %session_id, + turn_id = %turn_id, + error = %e, + attempt = transient_attempts, + "turn step hit a transient dependency outage (engine restart boot race); retrying in place" + ); + tokio::time::sleep(std::time::Duration::from_millis(TRANSIENT_STEP_BACKOFF_MS)) + .await; + } + Err(e) => { + tracing::error!(session_id = %session_id, turn_id = %turn_id, error = %e, "turn step failed; finalising turn as failed"); + break turn_loop::fail_turn(deps, &session_id, &turn_id, &e.to_string()).await; + } } }; record_step_status(&result); @@ -106,3 +155,39 @@ fn record_step_status(result: &TurnStepResult) { span.set_status(Status::error("harness turn failed")); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn boot_race_errors_are_transient_and_real_failures_are_not() { + // The issue-507 boot race: a restored step reaching a dependency the + // restarted engine has not re-registered yet. + assert!(is_transient_step_error(&HarnessError::Dependency( + "context::assemble: remote error (function_not_found): Function not found".into() + ))); + assert!(is_transient_step_error(&HarnessError::State( + "state::get harness_turn/s_1: iii is not connected".into() + ))); + assert!(!is_transient_step_error(&HarnessError::State( + "state::set harness_turn/s_1: iii is not connected".into() + ))); + // Case-insensitive on the engine's code spelling. + assert!(is_transient_step_error(&HarnessError::Dependency( + "session::messages: remote error (FUNCTION_NOT_FOUND): unknown".into() + ))); + // Enqueue exhaustion must fail the turn, not retry into a stale-ack + // wedge (enqueue_step already retried internally). + assert!(!is_transient_step_error(&HarnessError::Dependency( + "enqueue harness::turn: remote error (function_not_found): Function not found".into() + ))); + // Ordinary failures stay terminal. + assert!(!is_transient_step_error(&HarnessError::Dependency( + "router::chat: provider rejected the request".into() + ))); + assert!(!is_transient_step_error(&HarnessError::Internal( + "function_not_found mentioned in an internal error".into() + ))); + } +} diff --git a/harness/src/queue.rs b/harness/src/queue.rs index 21ed50604..a45caa639 100644 --- a/harness/src/queue.rs +++ b/harness/src/queue.rs @@ -69,7 +69,8 @@ fn turn_queue_definition() -> Value { "concurrency": 10, "max_retries": 3, "backoff_ms": 1_000, - "poll_interval_ms": 100 + "poll_interval_ms": 100, + "redeliver_on_engine_restart": true } }) } @@ -90,7 +91,8 @@ mod tests { "concurrency": 10, "max_retries": 3, "backoff_ms": 1_000, - "poll_interval_ms": 100 + "poll_interval_ms": 100, + "redeliver_on_engine_restart": true } }) ); diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs index 1f0842ae7..22523279f 100644 --- a/harness/src/turn_loop.rs +++ b/harness/src/turn_loop.rs @@ -117,6 +117,15 @@ pub struct TurnStepResult { pub skipped: bool, } +/// Bounded in-place retry for the enqueue that follows a persisted step +/// advance: `queue::enqueue` can be briefly unregistered while the queue +/// worker replays its registrations against a restarted engine, and failing +/// the turn in that window (or worse, wedging it Running with the step +/// persisted but never enqueued) turns a recoverable restart into a dead +/// session (iii-hq/workers#507). +const ENQUEUE_ATTEMPTS: u32 = 5; +const ENQUEUE_RETRY_BACKOFF_MS: u64 = 500; + /// Enqueue the next durable loop step onto the dedicated `harness-turn` queue. pub async fn enqueue_step( iii: &IIIClient, @@ -131,17 +140,40 @@ pub async fn enqueue_step( if let Some(preview) = message_preview { payload["message_preview"] = json!(preview); } - iii.trigger(TriggerRequest { - function_id: "harness::turn".to_string(), - payload, - action: Some(TriggerAction::Enqueue { - queue: TURN_QUEUE.to_string(), - }), - timeout_ms: None, - }) - .await - .map(|_| ()) - .map_err(|e| HarnessError::Dependency(format!("enqueue harness::turn: {e}"))) + let mut last_error = String::new(); + for attempt in 1..=ENQUEUE_ATTEMPTS { + match iii + .trigger(TriggerRequest { + function_id: "harness::turn".to_string(), + payload: payload.clone(), + action: Some(TriggerAction::Enqueue { + queue: TURN_QUEUE.to_string(), + }), + timeout_ms: None, + }) + .await + { + Ok(_) => return Ok(()), + Err(e) => { + last_error = e.to_string(); + if attempt < ENQUEUE_ATTEMPTS { + tracing::warn!( + session_id = %session_id, + turn_id = %turn_id, + step, + attempt, + error = %last_error, + "enqueue harness::turn failed; retrying" + ); + tokio::time::sleep(std::time::Duration::from_millis(ENQUEUE_RETRY_BACKOFF_MS)) + .await; + } + } + } + } + Err(HarnessError::Dependency(format!( + "enqueue harness::turn: {last_error}" + ))) } fn origin(turn_id: &str) -> Value { @@ -190,6 +222,10 @@ pub async fn run_step( match crate::state::get_turn(&deps.iii, &payload.session_id, cfg.session_timeout_ms).await? { Some(r) => r, + // The turn record is the authoritative recovery snapshot. A + // transcript alone cannot recover budgets, parent linkage, output + // contracts, or dispatch policy safely, so an absent record stays + // a stale delivery and is acknowledged without fabricating state. None => return Ok(skipped(&payload.session_id)), }; diff --git a/harness/tests/integration/README.md b/harness/tests/integration/README.md index 9f0ce26b3..d7d1751bd 100644 --- a/harness/tests/integration/README.md +++ b/harness/tests/integration/README.md @@ -20,6 +20,7 @@ No provider key or network access is required. | INT-007 | `coalesced-fire` | direct | a burst past the fire-rate cap coalesces: cap + 1 dispatches, the trailing one stamped `__coalesced_fires` (shrunken gate via harness env; probe-side whole-run call evidence) | | INT-008 | `reaction-unregisters-run` | direct | a reaction session in the registrant's lineage unregisters the registrant's subscription (serve-time capture of the runtime sub id; probe-side call await) | | INT-009 | `late-join-predecessor-replay` | direct | a join predecessor registered after its watched session completed receives a catch-up completion fire (level-triggered join barrier; `probe_after_calls` gating) | +| INT-010 | `crash-recovery-507` | direct | SIGKILL and restart the engine while a controlled function is in flight, with `context::assemble` held out during boot; the side effect runs once, the interrupted call closes, and the turn completes | | UI-001 | `console-streamed-text` | playground | a message sent by the Console streams to durable completion | | UI-002 | `multi-turn-traces` | playground | a native function turn and a Console turn expose distinct traces and function-call events | @@ -101,8 +102,8 @@ cargo clippy --manifest-path harness/Cargo.toml \ ``` `validate --scenario all` checks every fixture. `run --scenario all` executes -the direct scenarios (INT-001, INT-002, INT-003); UI-001 and UI-002 must use -`playground`. INT-003 produces two terminal turns from one send: generation 1 +all direct scenarios; UI-001 and UI-002 must use `playground`. INT-003 produces +two terminal turns from one send: generation 1 steers a message into the running session (it parks durably) and then fails, so the harness's failed finalize drains the parked row and reseeds a turn to react to it. The failed route is deliberate — a park during a *completing* @@ -118,6 +119,7 @@ The fixture tests pin: - the streamed frame sequence and terminal response agreement; - function-call and function-result history for INT-002; +- fault timing, held-response wiring, and restart deadlines for INT-010; - the Console-specific system-prompt and `agent_trigger` tool matchers; - serialization round trips and the authoritative `harness::send` schema. diff --git a/harness/tests/integration/src/fixtures/loading.rs b/harness/tests/integration/src/fixtures/loading.rs index 373b51251..c4174b042 100644 --- a/harness/tests/integration/src/fixtures/loading.rs +++ b/harness/tests/integration/src/fixtures/loading.rs @@ -144,6 +144,26 @@ impl ScenarioFixture { target.function_id ); } + if let Some(fault) = &self.scenario.fault { + let target = + self.scenario.target.as_ref().ok_or_else(|| { + anyhow::anyhow!("fault injection needs a controlled function") + })?; + anyhow::ensure!( + fault.after_target_calls > 0, + "fault after_target_calls must be positive" + ); + anyhow::ensure!( + fault.function_id == target.function_id, + "fault function {:?} does not match controlled target {:?}", + fault.function_id, + target.function_id + ); + anyhow::ensure!( + target.hold_response, + "engine fault target must hold its response until SIGKILL" + ); + } anyhow::ensure!( !self.script.generations.is_empty(), "router script has no generations" diff --git a/harness/tests/integration/src/fixtures/tests.rs b/harness/tests/integration/src/fixtures/tests.rs index 19294dd99..12956bfcf 100644 --- a/harness/tests/integration/src/fixtures/tests.rs +++ b/harness/tests/integration/src/fixtures/tests.rs @@ -11,7 +11,7 @@ fn all_selection_returns_the_checked_in_fixtures() { ids, std::collections::BTreeSet::from([ "INT-001", "INT-002", "INT-003", "INT-004", "INT-005", "INT-006", "INT-007", "INT-008", - "INT-009", "UI-001", "UI-002" + "INT-009", "INT-010", "UI-001", "UI-002" ]) ); assert_eq!( @@ -19,7 +19,7 @@ fn all_selection_returns_the_checked_in_fixtures() { .iter() .filter(|fixture| fixture.driver == crate::scenarios::ScenarioDriver::Direct) .count(), - 9 + 10 ); } diff --git a/harness/tests/integration/src/probe.rs b/harness/tests/integration/src/probe.rs index adc57c489..2800cc3f7 100644 --- a/harness/tests/integration/src/probe.rs +++ b/harness/tests/integration/src/probe.rs @@ -56,6 +56,8 @@ pub struct ScenarioProbe { /// collection cannot. target_calls: Arc>>, target_notify: Arc, + target_response_released: Arc, + target_response_notify: Arc, } impl ScenarioProbe { @@ -71,6 +73,8 @@ impl ScenarioProbe { bindings: Mutex::new(Vec::new()), target_calls: Arc::new(Mutex::new(Vec::new())), target_notify: Arc::new(tokio::sync::Notify::new()), + target_response_released: Arc::new(AtomicBool::new(false)), + target_response_notify: Arc::new(tokio::sync::Notify::new()), }; probe.register_sinks(); Ok(probe) @@ -168,6 +172,8 @@ impl ScenarioProbe { target, Arc::clone(&self.target_calls), Arc::clone(&self.target_notify), + Arc::clone(&self.target_response_released), + Arc::clone(&self.target_response_notify), ); Ok(()) } @@ -203,6 +209,23 @@ impl ScenarioProbe { } } + /// Release a controlled target held at the response boundary. Idempotent + /// so fault-error cleanup and normal cleanup can both call it safely. + pub fn release_target_response(&self) { + self.target_response_released.store(true, Ordering::Release); + self.target_response_notify.notify_waiters(); + } + + /// Trigger ids are engine-local. Once that process restarts, discard the + /// stale ids before registering fresh observer bindings. + pub fn forget_observer_bindings(&self) -> anyhow::Result<()> { + self.bindings + .lock() + .map_err(|_| anyhow::anyhow!("integration/bindings_lock_poisoned"))? + .clear(); + Ok(()) + } + /// Bind every observer through acknowledged engine RPCs. Because these /// calls use the probe connection, each response is also a barrier for the /// function registrations queued before it. @@ -388,6 +411,7 @@ impl ScenarioProbe { } pub async fn shutdown(&self) { + self.release_target_response(); let bindings = self .bindings .lock() @@ -480,14 +504,19 @@ fn register_controlled_function( target: &ControlledTargetV1, calls: Arc>>, notify: Arc, + response_released: Arc, + response_notify: Arc, ) { let response = target.response.clone(); + let hold_response = target.hold_response; iii.register_function( &target.function_id, RegisterFunction::new_async(move |mut payload: Value| { let response = response.clone(); let calls = Arc::clone(&calls); let notify = Arc::clone(¬ify); + let response_released = Arc::clone(&response_released); + let response_notify = Arc::clone(&response_notify); async move { strip_engine_fields(&mut payload); calls @@ -495,6 +524,13 @@ fn register_controlled_function( .map_err(|_| Error::Handler("integration/target_calls_lock_poisoned".into()))? .push(payload); notify.notify_waiters(); + while hold_response && !response_released.load(Ordering::Acquire) { + let released = response_notify.notified(); + if response_released.load(Ordering::Acquire) { + break; + } + released.await; + } Ok::(response) } }) diff --git a/harness/tests/integration/src/process/child.rs b/harness/tests/integration/src/process/child.rs index 49e018a6e..2a877e6b1 100644 --- a/harness/tests/integration/src/process/child.rs +++ b/harness/tests/integration/src/process/child.rs @@ -9,6 +9,7 @@ use nix::sys::wait::waitpid; use nix::unistd::Pid; pub(super) const REAP_INTERVAL: Duration = Duration::from_millis(25); +const IMMEDIATE_KILL_BUDGET: Duration = Duration::from_secs(2); /// Tracks who owns the final wait for the direct child. /// @@ -59,6 +60,32 @@ impl SupervisedChild { self.child.try_wait() } + pub async fn kill_now(&mut self) -> anyhow::Result<()> { + self.signal_tree(Signal::SIGKILL)?; + let deadline = tokio::time::Instant::now() + IMMEDIATE_KILL_BUDGET; + loop { + match self.poll_reap() { + Ok(true) => return Ok(()), + Ok(false) if tokio::time::Instant::now() < deadline => { + tokio::time::sleep(REAP_INTERVAL).await; + } + Ok(false) => { + let _ = self.child.kill(); + self.defer_reap(); + anyhow::bail!( + "{} did not reap within {}ms after SIGKILL", + self.name, + IMMEDIATE_KILL_BUDGET.as_millis() + ); + } + Err(error) => { + self.defer_reap(); + return Err(error).with_context(|| format!("reaping {}", self.name)); + } + } + } + } + pub(super) fn signal_tree(&self, signal: Signal) -> anyhow::Result<()> { let group = self.signal_group(signal); let direct = match kill(Pid::from_raw(self.child.id() as i32), signal) { diff --git a/harness/tests/integration/src/process/supervisor.rs b/harness/tests/integration/src/process/supervisor.rs index a9f4e827a..bbd288907 100644 --- a/harness/tests/integration/src/process/supervisor.rs +++ b/harness/tests/integration/src/process/supervisor.rs @@ -62,7 +62,6 @@ impl ProcessSupervisor { Ok(pid) } - #[cfg(test)] pub fn remove(&mut self, name: &str) -> Option { let index = self .children diff --git a/harness/tests/integration/src/process/tests.rs b/harness/tests/integration/src/process/tests.rs index c3b5921e1..98598f9fa 100644 --- a/harness/tests/integration/src/process/tests.rs +++ b/harness/tests/integration/src/process/tests.rs @@ -42,6 +42,26 @@ fn drop_cleans_up_a_partially_started_supervisor() { ); } +#[tokio::test] +async fn kill_now_is_idempotent_after_a_synchronous_reap() { + let dir = tempfile::tempdir().unwrap(); + let mut supervisor = ProcessSupervisor::new(Duration::from_millis(100)); + let running = ProcessSpec::new( + "running", + "/bin/sh", + dir.path(), + dir.path().join("running.out"), + dir.path().join("running.err"), + ) + .args(["-c", "exec sleep 60"]); + let pid = supervisor.spawn(running).unwrap(); + let mut child = supervisor.remove("running").unwrap(); + + child.kill_now().await.unwrap(); + assert!(!process_is_running(pid)); + child.kill_now().await.unwrap(); +} + #[tokio::test] async fn observing_a_leader_exit_does_not_abandon_its_descendants() { let dir = tempfile::tempdir().unwrap(); diff --git a/harness/tests/integration/src/runtime.rs b/harness/tests/integration/src/runtime.rs index 2fbd1fa79..821469a20 100644 --- a/harness/tests/integration/src/runtime.rs +++ b/harness/tests/integration/src/runtime.rs @@ -10,6 +10,7 @@ pub enum RunPhase { Boot, Arm, Send, + Fault, Await, Collect, Grade, @@ -24,6 +25,7 @@ impl std::fmt::Display for RunPhase { RunPhase::Boot => "boot", RunPhase::Arm => "arm", RunPhase::Send => "send", + RunPhase::Fault => "fault", RunPhase::Await => "await", RunPhase::Collect => "collect", RunPhase::Grade => "grade", diff --git a/harness/tests/integration/src/scenario/phases/execution.rs b/harness/tests/integration/src/scenario/phases/execution.rs index 3c9776468..32a13fcb0 100644 --- a/harness/tests/integration/src/scenario/phases/execution.rs +++ b/harness/tests/integration/src/scenario/phases/execution.rs @@ -3,14 +3,18 @@ use std::time::Duration; use serde_json::{json, Value}; use crate::deadline::Deadline; +use crate::discovery; use crate::runtime::{RunError, RunErrorKind, RunPhase}; use crate::services::RunServices; +use crate::stack::Stack; +use crate::types::scenario::FaultKind; use super::super::report::rpc_failure; use super::super::runner::ScenarioRunner; use super::super::state::{ActiveTurn, PreparedRun}; const SEND_TIMEOUT_MS: u64 = 30_000; +const BOOT_RACE_DEPENDENCY_GAP: Duration = Duration::from_secs(3); impl ScenarioRunner<'_> { pub(in crate::scenario) async fn send( @@ -66,4 +70,149 @@ impl ScenarioRunner<'_> { } } } + + pub(in crate::scenario) async fn fault( + &mut self, + stack: &mut Stack, + services: &RunServices, + prepared: &PreparedRun, + active: &ActiveTurn, + ) -> Result<(), RunError> { + let Some(fault) = &prepared.scenario.fault else { + return Ok(()); + }; + let phase = RunPhase::Fault; + let deadline = active.deadline; + + let FaultKind::EngineSigkill = fault.kind; + let expected_calls = usize::try_from(fault.after_target_calls) + .map_err(|error| RunError::runner(phase, "convert fault target call count", error))?; + if let Err(error) = services + .probe() + .wait_for_target_calls(expected_calls, deadline) + .await + { + services.probe().release_target_response(); + let kind = if deadline.is_expired() { + RunErrorKind::Contract + } else { + RunErrorKind::Runner + }; + return Err(RunError::with_source( + phase, + kind, + format!( + "fewer than {} controlled-function calls observed before fault", + fault.after_target_calls + ), + error, + )); + } + + // Keep context::assemble unavailable after the engine comes back so + // the restored harness::turn must exercise its boot-race retry path. + stack + .kill_worker("context-manager") + .await + .map_err(|error| { + RunError::runner(phase, "stop context manager for boot race", error) + })?; + let kill_result = stack.kill_engine().await; + services.probe().release_target_response(); + kill_result + .map_err(|error| RunError::runner(phase, "kill engine for fault injection", error))?; + services + .probe() + .forget_observer_bindings() + .map_err(|error| { + RunError::runner(phase, "discard pre-restart observer bindings", error) + })?; + + deadline + .timeout( + "fault restart delay", + tokio::time::sleep(Duration::from_millis(fault.restart_delay_ms)), + ) + .await + .map_err(|error| { + RunError::runner( + phase, + "fault restart delay exceeded scenario deadline", + error, + ) + })?; + stack + .respawn_engine() + .map_err(|error| RunError::runner(phase, "respawn engine after fault", error))?; + + // Bind through the probe connection first: its replayed function + // registrations are ordered before this acknowledged RPC, and + // recoverable lifecycle bindings park until the harness trigger type + // returns. This minimizes the post-restart completion race. + services + .probe() + .bind_observers(&self.session_id, deadline) + .await + .map_err(|error| { + RunError::runner(phase, "restore observers after engine restart", error) + })?; + + let mut required = vec![ + "harness::turn", + "harness::send", + "session::messages", + "router::chat", + ]; + if let Some(target) = &prepared.scenario.target { + required.push(target.function_id.as_str()); + } + discovery::wait_for_functions(services.client(), &required, deadline) + .await + .map_err(|error| { + RunError::runner( + phase, + "wait for turn dependencies after engine restart", + error, + ) + })?; + + deadline + .timeout( + "hold context manager out of the boot-race window", + tokio::time::sleep(BOOT_RACE_DEPENDENCY_GAP), + ) + .await + .map_err(|error| { + RunError::runner( + phase, + "boot-race dependency gap exceeded scenario deadline", + error, + ) + })?; + stack + .respawn_worker(self.bins, "context-manager") + .map_err(|error| RunError::runner(phase, "respawn context manager", error))?; + required.push("context::assemble"); + discovery::wait_for_functions(services.client(), &required, deadline) + .await + .map_err(|error| { + RunError::runner( + phase, + "wait for turn dependencies after boot-race gap", + error, + ) + })?; + services + .probe() + .confirm_completion_binding(&self.session_id, deadline) + .await + .map_err(|error| { + RunError::runner( + phase, + "confirm completion observer after engine restart", + error, + ) + })?; + Ok(()) + } } diff --git a/harness/tests/integration/src/scenario/runner.rs b/harness/tests/integration/src/scenario/runner.rs index eacfada1f..9cfe4d399 100644 --- a/harness/tests/integration/src/scenario/runner.rs +++ b/harness/tests/integration/src/scenario/runner.rs @@ -264,7 +264,7 @@ impl<'a> ScenarioRunner<'a> { let outcome = async { self.arm_booted(&mut booted).await?; - self.run_phases_after_arm(&booted.services, &booted.prepared) + self.run_phases_after_arm(&mut booted.stack, &booted.services, &booted.prepared) .await } .await; @@ -279,10 +279,12 @@ impl<'a> ScenarioRunner<'a> { async fn run_phases_after_arm( &mut self, + stack: &mut Stack, services: &RunServices, prepared: &PreparedRun, ) -> Result<(), RunError> { let mut active = self.send(services, prepared).await?; + self.fault(stack, services, prepared, &active).await?; self.r#await(services, &mut active).await?; self.collect(services, prepared, &mut active).await?; let evidence = self.build_evidence(services, &active, Some(active.send_response.clone())); diff --git a/harness/tests/integration/src/scenarios/dsl.rs b/harness/tests/integration/src/scenarios/dsl.rs index 078a54e23..c6b71b73e 100644 --- a/harness/tests/integration/src/scenarios/dsl.rs +++ b/harness/tests/integration/src/scenarios/dsl.rs @@ -14,8 +14,8 @@ use crate::types::frames::{ }; use crate::types::probe::ControlledTargetV1; use crate::types::scenario::{ - CompiledFunctionExposureV1, CompiledFunctionPolicyV1, CompiledScenarioV1, - CompiledSendOptionsV1, CompiledSendV1, DeadlinesV1, + CompiledFaultV1, CompiledFunctionExposureV1, CompiledFunctionPolicyV1, CompiledScenarioV1, + CompiledSendOptionsV1, CompiledSendV1, DeadlinesV1, FaultKind, }; use crate::types::script::{ GenerationMatchV1, JsonMatcherV1, JsonNormalizerV1, ModelFixtureV1, NormalizerOperation, @@ -52,6 +52,8 @@ pub(super) struct Scenario { model: ModelFixtureV1, send: Option, target: Option, + fault: Option, + deadlines: DeadlinesV1, generations: Vec, expected_turn_statuses: Vec, verify: Option, @@ -77,6 +79,8 @@ impl Scenario { model, send: None, target: None, + fault: None, + deadlines: DeadlinesV1::default(), generations: Vec::new(), expected_turn_statuses: vec!["completed".to_string()], verify: None, @@ -97,6 +101,31 @@ impl Scenario { self } + pub(super) fn engine_sigkill( + mut self, + function: &ControlledFunction, + after_target_calls: u64, + restart_delay_ms: u64, + ) -> Self { + assert!( + after_target_calls > 0, + "fault target call count must be positive" + ); + self.fault = Some(CompiledFaultV1 { + kind: FaultKind::EngineSigkill, + function_id: function.id().to_string(), + after_target_calls, + restart_delay_ms, + }); + self + } + + pub(super) fn scenario_timeout_ms(mut self, timeout_ms: u64) -> Self { + assert!(timeout_ms > 0, "scenario timeout must be positive"); + self.deadlines.scenario_ms = timeout_ms; + self + } + pub(super) fn generation(mut self, generation: Generation) -> Self { self.generations.push(generation); self @@ -212,7 +241,8 @@ impl Scenario { description: self.description, send: compiled_send, target: self.target, - deadlines: DeadlinesV1::default(), + fault: self.fault, + deadlines: self.deadlines, }, script: RouterScriptV1 { schema_version: SchemaVersion1::V1, @@ -308,6 +338,7 @@ impl ControlledFunction { description: description.to_string(), request_schema: serde_json::Map::new(), response: Value::Null, + hold_response: false, }, } } @@ -328,6 +359,11 @@ impl ControlledFunction { self } + pub(super) fn hold_response(mut self) -> Self { + self.target.hold_response = true; + self + } + pub(super) fn id(&self) -> &str { &self.target.function_id } diff --git a/harness/tests/integration/src/scenarios/engine_restart_recovery.rs b/harness/tests/integration/src/scenarios/engine_restart_recovery.rs new file mode 100644 index 000000000..03b4a5e4c --- /dev/null +++ b/harness/tests/integration/src/scenarios/engine_restart_recovery.rs @@ -0,0 +1,161 @@ +//! INT-010 — an engine restart cannot strand an interrupted function call or +//! fail a turn while its dependencies re-register. +//! +//! Deterministic regression coverage for +//! . + +use serde_json::json; + +use super::dsl::{ControlledFunction, Generation, Model, Request, Response, Scenario, Send}; +use super::ScenarioDriver; +use crate::evidence_data::message_text; +use crate::fixtures::ScenarioFixture; + +pub(super) fn scenario() -> ScenarioFixture { + const ID: &str = "INT-010"; + const MESSAGE: &str = "Call the recorder once."; + const CALL_ID: &str = "call-1"; + + let model = Model::scripted("fixture-model"); + let record = ControlledFunction::new( + "{{run_id}}::record", + "Record one integration fixture value.", + ) + .request_schema(json!({ + "type": "object", + "additionalProperties": false, + "properties": { "value": { "type": "string" } }, + "required": ["value"] + })) + .returns_text("recorded") + .hold_response(); + let arguments = json!({ "value": "expected" }); + + Scenario::new( + ID, + "crash-recovery-507", + "A conversation stays usable when the engine restarts during a function call and dependencies re-register at different times.", + ScenarioDriver::Direct, + model.clone(), + ) + .send( + Send::message(MESSAGE) + .idempotency_key("{{run_id}}:integration-010") + .allow_function(&record), + ) + .function(record.clone()) + .generation( + Generation::new(1) + .expect( + Request::new() + .turn_request() + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_exact([super::dsl::Message::user(MESSAGE)]) + .tools_exact([record.tool()]), + ) + .respond(Response::function_call( + CALL_ID, + &record, + arguments.clone(), + 8, + 4, + )), + ) + // Recovery must continue from the authoritative durable turn record: the + // original prompt, tool policy, and exact turn step survive the engine. + .generation( + Generation::new(2) + .expect( + Request::new() + .turn_request_step(1) + .system_prompt_sha256("{{system_prompt_sha256}}") + .messages_subset([ + json!({ "role": "user" }), + json!({ "role": "assistant", "content": [{ + "type": "function_call", + "id": CALL_ID, + "function_id": "{{run_id}}::record" + }]}), + json!({ + "role": "function_result", + "function_call_id": CALL_ID, + "function_id": "{{run_id}}::record", + "is_error": true, + "details": { "error": { "code": "engine_restart" } } + }), + ]) + .tools_exact([record.tool()]), + ) + .respond(Response::text("recovered", 20, 2)), + ) + .engine_sigkill(&record, 1, 1_500) + .scenario_timeout_ms(120_000) + .verify(|run| { + run.expect_assistant_texts(["recovered"])?; + run.expect_message_counts(1, 2, 1)?; + run.expect_target_calls(1)?; + anyhow::ensure!( + run.target_calls == [json!({ "value": "expected" })], + "controlled function payloads {:?} != expected", + run.target_calls + ); + + let function_id = format!("{}::record", run.run_id); + let results = run.function_results(&function_id); + anyhow::ensure!( + results.len() == 1, + "interrupted call has {} closing function results, expected 1", + results.len() + ); + let result = results[0]; + anyhow::ensure!( + result + .get("function_call_id") + .and_then(serde_json::Value::as_str) + == Some(CALL_ID), + "function result does not close {CALL_ID}: {result}" + ); + anyhow::ensure!( + result.get("is_error").and_then(serde_json::Value::as_bool) == Some(true), + "interrupted function result must be an error: {result}" + ); + anyhow::ensure!( + result + .pointer("/details/error/code") + .and_then(serde_json::Value::as_str) + == Some("engine_restart"), + "interrupted function result must identify engine_restart: {result}" + ); + anyhow::ensure!( + message_text(result).contains("execution interrupted by engine restart"), + "interrupted function result lacks the recovery explanation: {result}" + ); + run.expect_no_duplicate_messages() + }) + .build() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::types::scenario::FaultKind; + use crate::types::script::JsonMatcherV1; + + #[test] + fn fixture_holds_the_first_call_and_requires_a_real_engine_restart() { + let fixture = scenario(); + let target = fixture.scenario.target.as_ref().unwrap(); + let fault = fixture.scenario.fault.as_ref().unwrap(); + + assert!(target.hold_response); + assert_eq!(fault.kind, FaultKind::EngineSigkill); + assert_eq!(fault.function_id, target.function_id); + assert_eq!(fault.after_target_calls, 1); + assert_eq!(fault.restart_delay_ms, 1_500); + assert_eq!(fixture.scenario.deadlines.scenario_ms, 120_000); + assert!(matches!( + fixture.script.generations[1].match_.request_id, + JsonMatcherV1::Regex { .. } + )); + } +} diff --git a/harness/tests/integration/src/scenarios/mod.rs b/harness/tests/integration/src/scenarios/mod.rs index 5aead54da..806c8695d 100644 --- a/harness/tests/integration/src/scenarios/mod.rs +++ b/harness/tests/integration/src/scenarios/mod.rs @@ -3,6 +3,7 @@ mod coalesced_fire; mod console_streamed_text; mod dsl; +mod engine_restart_recovery; mod exactly_once_function; mod join_spec_mismatch; mod late_join_replay; @@ -30,6 +31,7 @@ pub fn all() -> Vec { vec![ coalesced_fire::scenario(), console_streamed_text::scenario(), + engine_restart_recovery::scenario(), exactly_once_function::scenario(), join_spec_mismatch::scenario(), late_join_replay::scenario(), @@ -49,7 +51,7 @@ mod tests { #[test] fn every_fixture_is_unique_and_valid() { let fixtures = all(); - assert_eq!(fixtures.len(), 11); + assert_eq!(fixtures.len(), 12); let mut slugs = std::collections::BTreeSet::new(); let mut ids = std::collections::BTreeSet::new(); for fixture in fixtures { diff --git a/harness/tests/integration/src/stack/supervisor.rs b/harness/tests/integration/src/stack/supervisor.rs index c55f2d24b..9c3667339 100644 --- a/harness/tests/integration/src/stack/supervisor.rs +++ b/harness/tests/integration/src/stack/supervisor.rs @@ -1,4 +1,4 @@ -use std::path::Path; +use std::path::{Path, PathBuf}; use std::time::Duration; use crate::process::{ProcessSpec, ProcessSupervisor, TeardownReport, DEFAULT_TEARDOWN_BUDGET}; @@ -14,6 +14,10 @@ pub struct Stack { pub ws_url: String, pub paths: RunLayout, processes: ProcessSupervisor, + /// Engine spawn recipe retained for deterministic fault-injection + /// restarts: binary, arguments, and working directory. + engine_recipe: Option<(PathBuf, Vec, PathBuf)>, + engine_restarts: u32, } #[derive(Debug)] @@ -68,6 +72,12 @@ impl Stack { ws_url, paths: paths.clone(), processes: ProcessSupervisor::default(), + engine_recipe: Some(( + bins.engine.clone(), + engine_args.clone(), + paths.engine_dir.clone(), + )), + engine_restarts: 0, }; if let Err(error) = @@ -151,6 +161,50 @@ impl Stack { Ok(format!("http://127.0.0.1:{port}")) } + pub async fn kill_engine(&mut self) -> anyhow::Result<()> { + let mut engine = self + .processes + .remove("engine") + .ok_or_else(|| anyhow::anyhow!("no live engine child to kill"))?; + engine.kill_now().await?; + tracing::info!( + target: "harness_integration::stack", + "engine SIGKILLed for fault injection" + ); + Ok(()) + } + + pub async fn kill_worker(&mut self, worker: &str) -> anyhow::Result<()> { + let mut child = self + .processes + .remove(worker) + .ok_or_else(|| anyhow::anyhow!("no live {worker} child to kill"))?; + child.kill_now().await?; + tracing::info!( + target: "harness_integration::stack", + worker, + "worker stopped for fault injection" + ); + Ok(()) + } + + pub fn respawn_engine(&mut self) -> anyhow::Result<()> { + let (bin, args, cwd) = self + .engine_recipe + .clone() + .ok_or_else(|| anyhow::anyhow!("stack has no engine recipe"))?; + self.engine_restarts += 1; + let log_name = format!("engine.restart{}", self.engine_restarts); + self.spawn_child_logged_with_env("engine", &log_name, &bin, &args, &cwd, &[]) + } + + pub fn respawn_worker(&mut self, bins: &StackBins, worker: &str) -> anyhow::Result<()> { + let bin = bins + .resolve(worker) + .ok_or_else(|| anyhow::anyhow!("no binary configured for {worker}"))?; + self.spawn_worker(worker, bin) + } + fn spawn_worker(&mut self, worker: &str, bin: &Path) -> anyhow::Result<()> { self.spawn_worker_with_env(worker, bin, &[]) } @@ -176,6 +230,8 @@ impl Stack { ws_url: "ws://127.0.0.1:0".to_string(), paths, processes: ProcessSupervisor::new(DEFAULT_TEARDOWN_BUDGET), + engine_recipe: None, + engine_restarts: 0, } } @@ -209,8 +265,20 @@ impl Stack { cwd: &Path, extra_env: &[(String, String)], ) -> anyhow::Result<()> { - let stdout_log = self.paths.log_path(name, "out")?; - let stderr_log = self.paths.log_path(name, "err")?; + self.spawn_child_logged_with_env(name, name, bin, args, cwd, extra_env) + } + + fn spawn_child_logged_with_env( + &mut self, + name: &str, + log_name: &str, + bin: &Path, + args: &[String], + cwd: &Path, + extra_env: &[(String, String)], + ) -> anyhow::Result<()> { + let stdout_log = self.paths.log_path(log_name, "out")?; + let stderr_log = self.paths.log_path(log_name, "err")?; let mut spec = ProcessSpec::new(name, bin, cwd, stdout_log, stderr_log).args(args.iter().cloned()); for key in ENV_ALLOWLIST { diff --git a/harness/tests/integration/src/types/probe.rs b/harness/tests/integration/src/types/probe.rs index f01d343fb..d17def7ea 100644 --- a/harness/tests/integration/src/types/probe.rs +++ b/harness/tests/integration/src/types/probe.rs @@ -11,6 +11,11 @@ pub struct ControlledTargetV1 { pub description: String, pub request_schema: serde_json::Map, pub response: serde_json::Value, + /// Keep the handler pending until the fault injector releases it. This + /// makes the engine crash land after the side effect starts but before its + /// result can close the function call. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub hold_response: bool, } /// Exact wire shape accepted by the private lifecycle sink. diff --git a/harness/tests/integration/src/types/scenario/compiled.rs b/harness/tests/integration/src/types/scenario/compiled.rs index a3c1d0bf6..c7b591c3a 100644 --- a/harness/tests/integration/src/types/scenario/compiled.rs +++ b/harness/tests/integration/src/types/scenario/compiled.rs @@ -28,6 +28,8 @@ pub struct CompiledScenarioV1 { pub send: CompiledSendV1, #[serde(default, skip_serializing_if = "Option::is_none")] pub target: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fault: Option, pub deadlines: DeadlinesV1, } @@ -65,6 +67,22 @@ pub enum CompiledFunctionExposureV1 { Native, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct CompiledFaultV1 { + pub kind: FaultKind, + pub function_id: String, + #[schemars(range(min = 1))] + pub after_target_calls: u64, + pub restart_delay_ms: u64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum FaultKind { + EngineSigkill, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(deny_unknown_fields)] pub struct DeadlinesV1 { diff --git a/queue/src/adapter.rs b/queue/src/adapter.rs index 505992f51..56f65eb9a 100644 --- a/queue/src/adapter.rs +++ b/queue/src/adapter.rs @@ -78,6 +78,10 @@ pub struct FunctionQueueConfig { pub backoff_ms: u64, /// Delay between polls for adapters backed by a local store. pub poll_interval_ms: u64, + /// Re-deliver an invocation inside the same queue attempt when the engine + /// restarts while it is in flight. This is only safe for consumers whose + /// own durable checkpoints make duplicate delivery idempotent. + pub redeliver_on_engine_restart: bool, /// Declares the queue as a RabbitMQ priority queue with this many /// priority levels (`x-max-priority`). `None` means not a priority /// queue. RabbitMQ-only; other adapters ignore it. Added (rather than @@ -102,6 +106,7 @@ impl Default for FunctionQueueConfig { message_group_field: None, backoff_ms: 1_000, poll_interval_ms: 100, + redeliver_on_engine_restart: false, max_priority: None, priority_field: None, } @@ -630,14 +635,17 @@ mod tests { let defaults: FunctionQueueConfig = serde_json::from_value(serde_json::json!({})).unwrap(); assert_eq!(defaults, FunctionQueueConfig::default()); assert_eq!(defaults.timeout_ms, 1_800_000); + assert!(!defaults.redeliver_on_engine_restart); assert!(defaults.validate("turns").is_ok()); let fifo: FunctionQueueConfig = serde_json::from_value(serde_json::json!({ "type": "fifo", - "message_group_field": "session_id" + "message_group_field": "session_id", + "redeliver_on_engine_restart": true })) .unwrap(); assert!(fifo.validate("harness-turn").is_ok()); + assert!(fifo.redeliver_on_engine_restart); let invalid = FunctionQueueConfig { r#type: "fifo".to_string(), diff --git a/queue/src/runtime.rs b/queue/src/runtime.rs index 86f6b252c..124fd8c56 100644 --- a/queue/src/runtime.rs +++ b/queue/src/runtime.rs @@ -17,6 +17,8 @@ use crate::boot::{ApplyLock, ConfigCell}; use crate::config::QueueConfig; use crate::trigger::{Invoker, QueueTriggerHandler}; +const MAX_RESTART_REDELIVERIES: u32 = 3; + #[derive(Debug, Clone, Deserialize, JsonSchema)] #[serde(deny_unknown_fields)] pub struct DefineQueueInput { @@ -820,12 +822,14 @@ async fn run_concurrent( let max_retries = config.max_retries; let poll_interval_ms = config.poll_interval_ms; let timeout_ms = config.timeout_ms; + let redeliver_on_engine_restart = config.redeliver_on_engine_restart; tasks.spawn(async move { process_standard_message( &queue_name, max_retries, poll_interval_ms, timeout_ms, + redeliver_on_engine_restart, active, adapter, invoker, @@ -898,6 +902,7 @@ async fn run_grouped_fifo( let backoff_ms = config.backoff_ms; let poll_interval_ms = config.poll_interval_ms; let timeout_ms = config.timeout_ms; + let redeliver_on_engine_restart = config.redeliver_on_engine_restart; tasks.spawn(async move { while let Ok(Some(message)) = tokio::time::timeout(Duration::from_secs(60), group_rx.recv()).await @@ -908,6 +913,7 @@ async fn run_grouped_fifo( backoff_ms, poll_interval_ms, timeout_ms, + redeliver_on_engine_restart, active.clone(), adapter.clone(), invoker.clone(), @@ -950,6 +956,7 @@ async fn process_standard_message( max_retries: u32, poll_interval_ms: u64, timeout_ms: u64, + redeliver_on_engine_restart: bool, active: Arc, adapter: Arc, invoker: Arc, @@ -959,7 +966,16 @@ async fn process_standard_message( let Ok(_permit) = active.acquire_owned().await else { return; }; - let result = invoke_message(queue, &invoker, &message, message.attempt, timeout_ms).await; + let result = invoke_message_with_restart_policy( + queue, + &invoker, + &message, + message.attempt, + timeout_ms, + poll_interval_ms, + redeliver_on_engine_restart, + ) + .await; let operation = if result.is_ok() { adapter.ack_function_queue(queue, message.delivery_id).await } else { @@ -979,6 +995,7 @@ async fn process_fifo_message( backoff_ms: u64, poll_interval_ms: u64, timeout_ms: u64, + redeliver_on_engine_restart: bool, active: Arc, adapter: Arc, invoker: Arc, @@ -990,9 +1007,17 @@ async fn process_fifo_message( let Ok(permit) = active.clone().acquire_owned().await else { return; }; - let succeeded = invoke_message(queue, &invoker, &message, attempt, timeout_ms) - .await - .is_ok(); + let succeeded = invoke_message_with_restart_policy( + queue, + &invoker, + &message, + attempt, + timeout_ms, + poll_interval_ms, + redeliver_on_engine_restart, + ) + .await + .is_ok(); drop(permit); if succeeded { if let Err(err) = adapter.ack_function_queue(queue, message.delivery_id).await { @@ -1024,6 +1049,74 @@ async fn process_fifo_message( } } +/// Only queues that explicitly declare their consumer checkpoint-safe may +/// re-deliver inside one queue attempt after an engine restart. Other queues +/// retain their normal timeout, retry, and DLQ semantics. +async fn invoke_message_with_restart_policy( + queue: &str, + invoker: &Arc, + message: &QueueMessage, + attempt: u32, + timeout_ms: u64, + poll_interval_ms: u64, + redeliver_on_engine_restart: bool, +) -> Result, String> { + if !redeliver_on_engine_restart { + return invoke_message(queue, invoker, message, attempt, timeout_ms).await; + } + + invoke_checkpointed_message_across_engine_restarts( + queue, + invoker, + message, + attempt, + timeout_ms, + poll_interval_ms, + ) + .await +} + +/// Replace a stranded checkpoint-safe invocation after an engine restart. +/// The loop is deliberately bounded; repeated restarts fall back to the +/// queue's ordinary nack/retry budget instead of pinning a FIFO group forever. +async fn invoke_checkpointed_message_across_engine_restarts( + queue: &str, + invoker: &Arc, + message: &QueueMessage, + attempt: u32, + timeout_ms: u64, + poll_interval_ms: u64, +) -> Result, String> { + let mut redeliveries = 0u32; + loop { + // Capture the current epoch before starting this invocation. A + // process-global epoch is unsafe here: after an idle engine restart it + // would make the first post-restart delivery look interrupted. + let Some(baseline) = invoker.connection_epoch().await? else { + return invoke_message(queue, invoker, message, attempt, timeout_ms).await; + }; + tokio::select! { + result = invoke_message(queue, invoker, message, attempt, timeout_ms) => return result, + () = invoker.connection_lost_since(baseline) => { + if redeliveries >= MAX_RESTART_REDELIVERIES { + return Err(format!( + "{} crossed more than {MAX_RESTART_REDELIVERIES} engine restarts in one queue attempt", + message.function_id + )); + } + redeliveries += 1; + tracing::warn!( + queue = %queue, + function_id = %message.function_id, + redeliveries, + "engine connection lost with the invocation in flight; holding until the target re-registers, then re-invoking" + ); + wait_for_function(invoker, queue, &message.function_id, poll_interval_ms).await; + } + } + } +} + async fn invoke_message( queue: &str, invoker: &Arc, @@ -1336,6 +1429,91 @@ mod tests { } } + /// First call pends forever (the invocation stranded by an engine crash); + /// `connection_lost` resolves exactly once, after that call has started; + /// every later call succeeds. Mirrors the issue-507 SIGKILL-and-respawn + /// window as seen by the consumer loop. + #[derive(Default)] + struct CrashOnceInvoker { + calls: AtomicUsize, + lost_fired: std::sync::atomic::AtomicBool, + } + + #[async_trait] + impl Invoker for CrashOnceInvoker { + async fn call(&self, _function_id: &str, _payload: Value) -> Result, String> { + if self.calls.fetch_add(1, Ordering::SeqCst) == 0 { + std::future::pending::<()>().await; + } + Ok(None) + } + + async fn connection_epoch(&self) -> Result, String> { + Ok(Some(if self.lost_fired.load(Ordering::SeqCst) { + 2 + } else { + 1 + })) + } + + async fn connection_lost_since(&self, baseline: u64) { + loop { + if baseline == 1 + && self.calls.load(Ordering::SeqCst) == 1 + && !self.lost_fired.swap(true, Ordering::SeqCst) + { + return; + } + tokio::time::sleep(Duration::from_millis(2)).await; + } + } + } + + #[derive(Default)] + struct RestartSignalButSuccessfulInvoker { + calls: AtomicUsize, + } + + #[async_trait] + impl Invoker for RestartSignalButSuccessfulInvoker { + async fn call(&self, _function_id: &str, _payload: Value) -> Result, String> { + self.calls.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(20)).await; + Ok(None) + } + + async fn connection_epoch(&self) -> Result, String> { + Ok(Some(1)) + } + + async fn connection_lost_since(&self, _baseline: u64) { + tokio::time::sleep(Duration::from_millis(2)).await; + } + } + + #[derive(Default)] + struct AlwaysRestartingInvoker { + calls: AtomicUsize, + } + + #[async_trait] + impl Invoker for AlwaysRestartingInvoker { + async fn call(&self, _function_id: &str, _payload: Value) -> Result, String> { + self.calls.fetch_add(1, Ordering::SeqCst); + std::future::pending::, String>>().await + } + + async fn connection_epoch(&self) -> Result, String> { + Ok(Some(self.calls.load(Ordering::SeqCst) as u64 + 1)) + } + + async fn connection_lost_since(&self, baseline: u64) { + while self.calls.load(Ordering::SeqCst) < baseline as usize { + tokio::task::yield_now().await; + } + } + } + #[derive(Default)] struct TimeoutRecordingInvoker { timeout_ms: AtomicU64, @@ -1358,10 +1536,16 @@ mod tests { } } - fn message(delivery_id: u64, session: &str, sequence: u64, delay_ms: u64) -> QueueMessage { + fn message_for( + delivery_id: u64, + function_id: &str, + session: &str, + sequence: u64, + delay_ms: u64, + ) -> QueueMessage { QueueMessage { delivery_id, - function_id: "harness::turn".to_string(), + function_id: function_id.to_string(), data: json!({ "session_id": session, "sequence": sequence, @@ -1374,6 +1558,10 @@ mod tests { } } + fn message(delivery_id: u64, session: &str, sequence: u64, delay_ms: u64) -> QueueMessage { + message_for(delivery_id, "harness::turn", session, sequence, delay_ms) + } + #[test] fn fifo_group_key_accepts_string_and_number() { assert_eq!( @@ -1568,6 +1756,86 @@ mod tests { assert_eq!(*adapter.acked.lock().unwrap(), vec![7]); } + #[tokio::test] + async fn fifo_reinvokes_after_engine_connection_loss_without_burning_attempts() { + let adapter = Arc::new(RecordingAdapter::default()); + let invoker = Arc::new(CrashOnceInvoker::default()); + + tokio::time::timeout( + Duration::from_secs(2), + process_fifo_message( + "harness-turn", + 3, + 1, + 1, + 1_800_000, + true, + Arc::new(Semaphore::new(1)), + adapter.clone(), + invoker.clone(), + message(13, "s1", 1, 0), + ), + ) + .await + .expect("interrupted delivery must be re-invoked, not stranded"); + + // The stranded first invocation is replaced by exactly one re-invoke; + // it succeeds and acks with no nack (no retry attempt consumed). + assert_eq!(invoker.calls.load(Ordering::SeqCst), 2); + assert_eq!(*adapter.acked.lock().unwrap(), vec![13]); + assert!(adapter.nacked.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn standard_delivery_without_checkpoint_opt_in_uses_normal_semantics() { + let adapter = Arc::new(RecordingAdapter::default()); + let invoker = Arc::new(RestartSignalButSuccessfulInvoker::default()); + + tokio::time::timeout( + Duration::from_secs(2), + process_standard_message( + "turns", + 3, + 1, + 1_800_000, + false, + Arc::new(Semaphore::new(1)), + adapter.clone(), + invoker.clone(), + message_for(17, "jobs::run", "s1", 1, 0), + ), + ) + .await + .expect("generic delivery should complete through normal queue semantics"); + + assert_eq!(invoker.calls.load(Ordering::SeqCst), 1); + assert_eq!(*adapter.acked.lock().unwrap(), vec![17]); + assert!(adapter.nacked.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn checkpointed_restart_redelivery_is_bounded() { + let concrete = Arc::new(AlwaysRestartingInvoker::default()); + let invoker: Arc = concrete.clone(); + let message = message(19, "s1", 1, 0); + let error = invoke_checkpointed_message_across_engine_restarts( + "harness-turn", + &invoker, + &message, + 0, + 1_800_000, + 1, + ) + .await + .unwrap_err(); + + assert!(error.contains("more than 3 engine restarts")); + assert_eq!( + concrete.calls.load(Ordering::SeqCst), + MAX_RESTART_REDELIVERIES as usize + 1 + ); + } + #[tokio::test] async fn fifo_retries_in_place_then_acks_without_requeueing() { let adapter = Arc::new(RecordingAdapter::default()); @@ -1582,6 +1850,7 @@ mod tests { 1, 1, 1_800_000, + false, Arc::new(Semaphore::new(1)), adapter.clone(), invoker.clone(), @@ -1608,6 +1877,7 @@ mod tests { 1, 1, 1_800_000, + false, Arc::new(Semaphore::new(1)), adapter.clone(), invoker.clone(), diff --git a/queue/src/trigger.rs b/queue/src/trigger.rs index 21f50f7df..321ec8e77 100644 --- a/queue/src/trigger.rs +++ b/queue/src/trigger.rs @@ -8,6 +8,7 @@ use std::collections::HashMap; use std::sync::Arc; +use std::time::Duration; use async_trait::async_trait; use iii_sdk::errors::Error; @@ -28,6 +29,36 @@ use crate::subscriber_config::SubscriberQueueConfig; /// is still running. 30 minutes covers the longest intended job budget. const FUNCTION_QUEUE_INVOCATION_TIMEOUT_MS: u64 = 30 * 60 * 1_000; +/// How often [`IiiInvoker::connection_lost_since`] samples the engine epoch +/// while an invocation is in flight, and how long one sample may take. +const ENGINE_EPOCH_PROBE_INTERVAL_MS: u64 = 1_000; +const ENGINE_EPOCH_PROBE_TIMEOUT_MS: u64 = 3_000; + +/// The engine's boot identity: the earliest `connected_at_ms` among its +/// in-process (`runtime == "engine"`) workers, which attach once at engine +/// startup. A restarted engine reports a new epoch — the signal that every +/// invocation in flight across the restart lost its result routing. `None` +/// when the engine cannot be reached (outage in progress) or the response +/// shape is unrecognized. +async fn engine_epoch_ms(iii: &IIIClient) -> Option { + let response = iii + .trigger(TriggerRequest { + function_id: "engine::workers::list".to_string(), + payload: json!({}), + action: None, + timeout_ms: Some(ENGINE_EPOCH_PROBE_TIMEOUT_MS), + }) + .await + .ok()?; + response + .get("workers")? + .as_array()? + .iter() + .filter(|worker| worker.get("runtime").and_then(Value::as_str) == Some("engine")) + .filter_map(|worker| worker.get("connected_at_ms").and_then(Value::as_u64)) + .min() +} + #[async_trait] pub trait Invoker: Send + Sync + 'static { async fn call(&self, function_id: &str, payload: Value) -> Result, String>; @@ -50,6 +81,17 @@ pub trait Invoker: Send + Sync + 'static { async fn function_available(&self, _function_id: &str) -> Result { Ok(true) } + + /// Capture the engine epoch immediately before a restart-sensitive + /// invocation. `None` disables restart watching for this invocation. + async fn connection_epoch(&self) -> Result, String> { + Ok(None) + } + + /// Resolve when the engine moves away from the captured epoch. + async fn connection_lost_since(&self, _baseline: u64) { + std::future::pending::<()>().await + } } #[derive(Clone)] @@ -118,6 +160,31 @@ impl Invoker for IiiInvoker { } } } + + async fn connection_epoch(&self) -> Result, String> { + Ok(engine_epoch_ms(&self.iii).await) + } + + async fn connection_lost_since(&self, baseline: u64) { + // Neither the SDK connection state nor plain liveness probes can see + // a fast restart: the reconnect loop reports `Connected` through its + // silent 2s retry sleep, and outbound messages buffered during the + // outage are answered by the NEW engine as if nothing happened + // (both verified against a SIGKILLed-and-respawned engine). The + // engine's boot epoch is the reliable signal — a buffered probe + // answered by a restarted engine reveals the changed epoch. + loop { + tokio::time::sleep(Duration::from_millis(ENGINE_EPOCH_PROBE_INTERVAL_MS)).await; + // An unreadable epoch (outage in progress) never trips by + // itself: nothing can progress until the engine is back, and + // the first successful sample afterwards decides. + if let Some(epoch) = engine_epoch_ms(&self.iii).await { + if epoch != baseline { + return; + } + } + } + } } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]