(MOT-4107) fix(harness): recover interrupted calls after engine restarts - #625
Closed
ytallo wants to merge 1 commit into
Closed
(MOT-4107) fix(harness): recover interrupted calls after engine restarts#625ytallo wants to merge 1 commit into
ytallo wants to merge 1 commit into
Conversation
Detect engine restarts across in-flight queue and harness invocations, redeliver durable turn work after registrations recover, and close interrupted calls from the surviving transcript. Add a deterministic integration scenario that SIGKILLs and restarts the engine while a controlled function is in flight, then verifies exactly-once execution and a completed turn.
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
skill-check — worker0 verified, 49 skipped (no docs/).
Four for four. Nicely done. |
📝 WalkthroughWalkthroughThe changes add engine-epoch restart detection, transient retry handling, lost-turn recovery with synthesized function results, queue reinvocation after connection loss, and a new SIGKILL-based crash-recovery integration scenario. ChangesCrash recovery flow
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant ScenarioRunner
participant EngineStack
participant HarnessTurn
participant SessionTranscript
ScenarioRunner->>EngineStack: SIGKILL engine
EngineStack-->>ScenarioRunner: respawn engine
ScenarioRunner->>HarnessTurn: resume durable turn
HarnessTurn->>SessionTranscript: append synthesized function_result
SessionTranscript-->>HarnessTurn: repaired transcript
HarnessTurn-->>ScenarioRunner: continue recovery
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🧹 Nitpick comments (4)
queue/src/trigger.rs (1)
32-87: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftEpoch probing is duplicated verbatim across crates.
ENGINE_EPOCH_PROBE_*,LAST_KNOWN_ENGINE_EPOCH,seed_engine_epoch, andengine_epoch_msare identical toharness/src/clients/engine.rs(Lines 149-206). The repo already shares code throughiii_helpers(used inqueue/src/runtime.rs); hosting the epoch probe there would keep the two restart detectors from drifting — particularly themin(connected_at_ms)heuristic, which is the part most likely to need tuning.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@queue/src/trigger.rs` around lines 32 - 87, Move the shared epoch-probing constants, LAST_KNOWN_ENGINE_EPOCH, seed_engine_epoch, and engine_epoch_ms implementation from queue/src/trigger.rs into the existing iii_helpers shared module, preserving the current min(connected_at_ms) heuristic and probe behavior. Update queue and harness callers to reuse the shared symbols and remove their duplicated local definitions.harness/src/turn_loop.rs (2)
2886-2908: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUntested branch: results embedded in a user message.
dangling_transcript_callstreatsContentBlock::FunctionResultinside aUsermessage as resolving a call (Lines 2031-2040), but both tests only use message-levelFunctionResult. A regression that drops the embedded-block branch would re-close already-answered calls and duplicate transcript entries — exactly whatexpect_no_duplicate_messagesin INT-010 guards, but only at integration cost. One more fixture case pins it cheaply.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/src/turn_loop.rs` around lines 2886 - 2908, Add a unit-test fixture for dangling_transcript_calls covering a User message whose content includes a ContentBlock::FunctionResult resolving a prior assistant call. Assert the call is not reported as dangling, while preserving the existing message-level result coverage and test expectations.
1991-1999: 🗄️ Data Integrity & Integration | 🔵 Trivial | 🏗️ Heavy liftRecovered turn loses its budget caps.
max_total_tokens,max_cost_usd, andbudget_root_session_idall reset toNone, so the resumed generation runs unbudgeted and charges nothing to the root ledger. For sessions created with a hard cost ceiling, an engine restart becomes a way to exceed it — bounded to the recovery generation, but silently.
functions: Noneis the right fail-closed default; the budget fields deserve the same care. If the per-send options can't be recovered, consider persisting the budget identity alongside the transcript, or at minimum recording the bypass in the recovery audit entry below so it is auditable.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/src/turn_loop.rs` around lines 1991 - 1999, Preserve budget enforcement when reconstructing the recovered turn: restore max_total_tokens, max_cost_usd, and budget_root_session_id from persisted session or transcript state instead of resetting them to None. If these per-send options cannot be recovered, record the bypass in the recovery audit entry so the unbudgeted generation is auditable; keep functions: None as the fail-closed default.harness/tests/integration/src/scenarios/dsl.rs (1)
413-432: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winGuard against silently discarding prior request matchers.
recovery_boundary()unconditionally overwritessystem_prompt/messages/toolsand setsany_turn_step = truewith no check for prior configuration. If a future fixture chains this withturn_request_step(...)or another matcher setter, that configuration is silently dropped —compile()has no way to detect or report the conflict.♻️ Proposed guard
pub(super) fn recovery_boundary(mut self) -> Self { + debug_assert!( + self.turn_step.is_none() && self.system_prompt.is_none(), + "recovery_boundary() must be the only matcher configuration for this request" + ); self.turn_request = true; self.any_turn_step = true;🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@harness/tests/integration/src/scenarios/dsl.rs` around lines 413 - 432, Update MatcherBuilder::recovery_boundary so it validates that no prior system_prompt, messages, tools, or specific turn-step matcher is configured before applying its stable-envelope defaults. Reject the conflicting combination explicitly, using the builder’s existing validation or assertion mechanism, rather than silently overwriting settings; preserve normal recovery_boundary behavior when no conflicting matchers exist.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@harness/src/clients/engine.rs`:
- Around line 155-181: The process-global epoch is being advanced by the first
restart watcher, stranding concurrent calls. In
harness/src/clients/engine.rs:155-181, update engine_link_interrupted to capture
a per-watcher baseline and stop storing the new epoch in its detection path;
advance LAST_KNOWN_ENGINE_EPOCH only through a dedicated monotonically advancing
re-seed path. In queue/src/trigger.rs:221-233, update
IiiInvoker::connection_lost to remove the detection-path store and compare
against each watcher's captured epoch so all concurrent deliveries detect the
transition.
In `@harness/src/functions/turn.rs`:
- Around line 92-101: Update is_transient_step_error to stop treating all
HarnessError::State failures as transient, since state writes such as
state::put_turn can fail after target execution and cause duplicate dispatch on
retry. Restrict retry classification to HarnessError::Dependency, or only state
read errors such as state::get, while preserving the existing
function_not_found, not connected, and enqueue harness::turn checks.
In `@harness/src/turn_loop.rs`:
- Around line 1871-1882: Scope the dangling-call recovery in the current
turn-recovery flow to assistant entries whose origin matches payload.turn_id
before calling dangling_transcript_calls. Use the turn association carried by
each entry, and ensure last_assistant_identity and subsequent
synthesized-result/advance logic operate only on the filtered entries,
preserving no-op behavior when the payload turn has no dangling calls.
- Line 1901: Update recover_lost_turn so its emitted details structure matches
INT-010’s expected /details/error/code path by nesting engine_restart under
error.code, or alternatively change the scenario assertion to the existing flat
error convention; keep the producer and assertion consistent.
In `@queue/src/runtime.rs`:
- Around line 1042-1072: Bound the restart-driven re-invocation loop in
invoke_message_across_engine_restarts with a finite counter or budget, and once
exhausted stop re-invoking so the caller reaches the normal nack/retry and DLQ
handling. Remove or narrow the doc comment’s harness::turn-specific safety
claim, and ensure the limit applies uniformly to all queue consumers.
---
Nitpick comments:
In `@harness/src/turn_loop.rs`:
- Around line 2886-2908: Add a unit-test fixture for dangling_transcript_calls
covering a User message whose content includes a ContentBlock::FunctionResult
resolving a prior assistant call. Assert the call is not reported as dangling,
while preserving the existing message-level result coverage and test
expectations.
- Around line 1991-1999: Preserve budget enforcement when reconstructing the
recovered turn: restore max_total_tokens, max_cost_usd, and
budget_root_session_id from persisted session or transcript state instead of
resetting them to None. If these per-send options cannot be recovered, record
the bypass in the recovery audit entry so the unbudgeted generation is
auditable; keep functions: None as the fail-closed default.
In `@harness/tests/integration/src/scenarios/dsl.rs`:
- Around line 413-432: Update MatcherBuilder::recovery_boundary so it validates
that no prior system_prompt, messages, tools, or specific turn-step matcher is
configured before applying its stable-envelope defaults. Reject the conflicting
combination explicitly, using the builder’s existing validation or assertion
mechanism, rather than silently overwriting settings; preserve normal
recovery_boundary behavior when no conflicting matchers exist.
In `@queue/src/trigger.rs`:
- Around line 32-87: Move the shared epoch-probing constants,
LAST_KNOWN_ENGINE_EPOCH, seed_engine_epoch, and engine_epoch_ms implementation
from queue/src/trigger.rs into the existing iii_helpers shared module,
preserving the current min(connected_at_ms) heuristic and probe behavior. Update
queue and harness callers to reuse the shared symbols and remove their
duplicated local definitions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 639e5af5-0b4b-4b35-9a0c-006aebc4dadf
📒 Files selected for processing (23)
harness/src/clients/engine.rsharness/src/functions/turn.rsharness/src/main.rsharness/src/turn_loop.rsharness/tests/integration/README.mdharness/tests/integration/src/fixtures/loading.rsharness/tests/integration/src/fixtures/tests.rsharness/tests/integration/src/probe.rsharness/tests/integration/src/process/child.rsharness/tests/integration/src/process/supervisor.rsharness/tests/integration/src/process/tests.rsharness/tests/integration/src/runtime.rsharness/tests/integration/src/scenario/phases/execution.rsharness/tests/integration/src/scenario/runner.rsharness/tests/integration/src/scenarios/crash_recovery_507.rsharness/tests/integration/src/scenarios/dsl.rsharness/tests/integration/src/scenarios/mod.rsharness/tests/integration/src/stack/supervisor.rsharness/tests/integration/src/types/probe.rsharness/tests/integration/src/types/scenario/compiled.rsqueue/src/boot.rsqueue/src/runtime.rsqueue/src/trigger.rs
💤 Files with no reviewable changes (1)
- harness/tests/integration/src/process/supervisor.rs
Comment on lines
+155
to
+181
| /// The engine boot epoch this process last observed (0 = not yet sampled). | ||
| /// Seeded at worker boot ([`seed_engine_epoch`]) so a restart is detectable | ||
| /// even when the crash lands before an in-flight dispatch's first own | ||
| /// sample — the exact window the issue-507 fault hits (the crash follows | ||
| /// the dispatch by design). | ||
| static LAST_KNOWN_ENGINE_EPOCH: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); | ||
|
|
||
| /// Sample the engine epoch until it succeeds once and remember it as the | ||
| /// process baseline. Spawned at boot; keeps the baseline warm before any | ||
| /// turn dispatches a call. | ||
| pub async fn seed_engine_epoch(iii: Arc<IIIClient>) { | ||
| loop { | ||
| if let Some(epoch) = engine_epoch_ms(&iii).await { | ||
| let _ = LAST_KNOWN_ENGINE_EPOCH.compare_exchange( | ||
| 0, | ||
| epoch, | ||
| std::sync::atomic::Ordering::SeqCst, | ||
| std::sync::atomic::Ordering::SeqCst, | ||
| ); | ||
| return; | ||
| } | ||
| tokio::time::sleep(std::time::Duration::from_millis( | ||
| ENGINE_EPOCH_PROBE_INTERVAL_MS, | ||
| )) | ||
| .await; | ||
| } | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Process-global epoch baseline is mutated by whichever watcher detects the restart first, so only one in-flight call recovers. Both detectors read LAST_KNOWN_ENGINE_EPOCH once into a local baseline, then store() the new epoch on detection. With multiple calls in flight across a single restart, the first watcher to observe the change advances the global; watchers that read their baseline after that store compare new-vs-new and never resolve, so their calls wait out the full timeout — the stranding #507 exists to eliminate.
harness/src/clients/engine.rs#L155-L181: giveengine_link_interrupteda per-watcher baseline snapshot and stop writing the global from the detection path (Line 249); advance the global only from a dedicated re-seed path, monotonically.queue/src/trigger.rs#L221-L233: apply the same change toIiiInvoker::connection_lost— drop thestore()on Line 228 and compare against the watcher's own captured epoch, so all concurrent deliveries observe the transition.
📍 Affects 2 files
harness/src/clients/engine.rs#L155-L181(this comment)queue/src/trigger.rs#L221-L233
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/clients/engine.rs` around lines 155 - 181, The process-global
epoch is being advanced by the first restart watcher, stranding concurrent
calls. In harness/src/clients/engine.rs:155-181, update engine_link_interrupted
to capture a per-watcher baseline and stop storing the new epoch in its
detection path; advance LAST_KNOWN_ENGINE_EPOCH only through a dedicated
monotonically advancing re-seed path. In queue/src/trigger.rs:221-233, update
IiiInvoker::connection_lost to remove the detection-path store and compare
against each watcher's captured epoch so all concurrent deliveries detect the
transition.
Comment on lines
+92
to
+101
| fn is_transient_step_error(error: &HarnessError) -> bool { | ||
| let message = match error { | ||
| HarnessError::Dependency(m) | HarnessError::State(m) => m.to_ascii_lowercase(), | ||
| _ => return false, | ||
| }; | ||
| if message.contains("enqueue harness::turn") { | ||
| return false; | ||
| } | ||
| message.contains("function_not_found") || message.contains("not connected") | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Retrying HarnessError::State can re-dispatch an already-executed call.
Classifying State failures as transient means a state::put_turn failure is retryable — but run_step writes the Triggered/Done checkpoint via put_turn immediately after invoking the target (harness/src/turn_loop.rs Lines 1079, 1139). If that write is the thing that fails with not connected, the checkpoint is lost and the wholesale re-run re-invokes a target that already ran, breaking the at-most-once guarantee this PR is otherwise careful to preserve.
Dependency errors (context::assemble, session::messages) are the boot race described in the doc comment and are checkpoint-protected; State is the one variant where the checkpoint itself may be the casualty. Narrowing the classifier to Dependency, or to State only on read paths (state::get), would keep the recovery without the duplicate-side-effect window.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/functions/turn.rs` around lines 92 - 101, Update
is_transient_step_error to stop treating all HarnessError::State failures as
transient, since state writes such as state::put_turn can fail after target
execution and cause duplicate dispatch on retry. Restrict retry classification
to HarnessError::Dependency, or only state read errors such as state::get, while
preserving the existing function_not_found, not connected, and enqueue
harness::turn checks.
Comment on lines
+1871
to
+1882
| let entries = session.messages(&payload.session_id, true).await?; | ||
| let dangling = dangling_transcript_calls(&entries); | ||
| if dangling.is_empty() { | ||
| // No interrupted call — indistinguishable from a stale redelivery of | ||
| // an expired turn; keep the historical ack-and-drop. | ||
| return Ok(skipped(&payload.session_id)); | ||
| } | ||
| // The last assistant entry carries the model/provider the lost turn was | ||
| // generating with; a dangling call implies at least one assistant entry. | ||
| let Some((model, provider)) = last_assistant_identity(&entries) else { | ||
| return Ok(skipped(&payload.session_id)); | ||
| }; |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Dangling-call scan is session-wide, not scoped to the payload's turn.
session.messages(&payload.session_id, true) returns the whole active path, so dangling_transcript_calls can pick up an unclosed call left by an earlier abandoned turn. Two consequences:
- The synthesized result is written under
ids::function_result_entry_id(&payload.turn_id, call_id)— an old turn's call gets closed under the new turn's entry id. - A genuinely stale redelivery of an expired turn now stops being a no-op ack: it reconstructs a record and calls
advance(), resurrecting a turn that was intentionally finished.
Filtering the dangling set to calls whose assistant entry belongs to payload.turn_id (the entries carry an origin stamped with the turn id) would keep recovery targeted.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/turn_loop.rs` around lines 1871 - 1882, Scope the dangling-call
recovery in the current turn-recovery flow to assistant entries whose origin
matches payload.turn_id before calling dangling_transcript_calls. Use the turn
association carried by each entry, and ensure last_assistant_identity and
subsequent synthesized-result/advance logic operate only on the filtered
entries, preserving no-op behavior when the payload turn has no dangling calls.
| content: vec![ContentBlock::text( | ||
| crate::clients::engine::ENGINE_RESTART_INTERRUPTED.to_string(), | ||
| )], | ||
| details: json!({ "error": "engine_restart" }), |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# How do other function results shape details.error, and who reads /details/error/code?
rg -nP -C3 '"error"\s*:' --type=rust -g '!**/target/**' | rg -n 'json!|details' -C2
rg -nP -C3 'details/error|details\.error' --type=rustRepository: iii-hq/workers
Length of output: 152
🏁 Script executed:
#!/bin/bash
set -u
echo "== repo files of interest =="
git ls-files | rg '(^harness/src/turn_loop\.rs$|crash_recovery_507|turn_loop)' || true
echo "== turn_loop relevant section =="
if [ -f harness/src/turn_loop.rs ]; then
wc -l harness/src/turn_loop.rs
sed -n '1850,1930p' harness/src/turn_loop.rs | nl -ba -v1850
fi
echo "== scenario file =="
for f in $(git ls-files | rg 'crash_recovery_507'); do
echo "--- $f"
wc -l "$f"
sed -n '1,260p' "$f" | nl -ba -v1
done
echo "== searches for error/code/details in Rust =="
rg -n 'error|details|pointer|code|engine_restart|hook_denied' harness -g '*.rs' | rg -n 'json!|details|pointer|/details/error|details\.error|engine_restart|hook_denied' | head -n 200Repository: iii-hq/workers
Length of output: 13192
🏁 Script executed:
#!/bin/bash
# How do other function results shape details.error, and who reads /details/error/code?
rg -nP -C3 '"error"\s*:' --type=rust -g '!**/target/**' | rg -n 'json!|details' -C2
rg -nP -C3 'details/error|details\.error' --type=rustRepository: iii-hq/workers
Length of output: 152
🏁 Script executed:
#!/bin/bash
set -u
echo "== turn_loop recover_lost_turn section =="
sed -n '1850,1930p' harness/src/turn_loop.rs
echo "== crash_recovery_507 relevant sections =="
sed -n '80,115p' harness/tests/integration/src/scenarios/crash_recovery_507.rs
sed -n '1,240p' harness/tests/integration/src/scenarios/crash_recovery_507.rs
echo "== deterministic JSON pointer probe =="
python3 - <<'PY'
import json
flat = {"details": {"error": "engine_restart"}}
nested = {"details": {"error": {"code": "engine_restart", "message": "interrupted"}}}
def pointer(doc, path):
parts = [] if path == "" else path.split("/")[1:]
for part in parts:
part = part.replace("~1", "/").replace("~0", "~")
if isinstance(doc, dict):
if part not in doc:
return None
doc = doc[part]
elif isinstance(doc, list):
try:
idx = int(part)
except ValueError:
return None
if idx < 0 or idx >= len(doc):
return None
doc = doc[idx]
else:
return None
return doc
print("flat /details/error/code:", pointer(flat, "/details/error/code"))
print("nested /details/error/code:", pointer(nested, "/details/error/code"))
PYRepository: iii-hq/workers
Length of output: 10218
🏁 Script executed:
#!/bin/bash
# How do other function results shape details.error, and who reads /details/error/code?
rg -nP -C3 '"error"\s*:' --type=rust -g '!**/target/**' | rg -n 'json!|details' -C2
rg -nP -C3 'details/error|details\.error' --type=rustRepository: iii-hq/workers
Length of output: 152
Make details.error match INT-010’s assertion.
recover_lost_turn emits details: {"error": "engine_restart"}, but the scenario verifies /details/error/code is "engine_restart". Nest the error shape or update the assertion to match the flat error convention.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@harness/src/turn_loop.rs` at line 1901, Update recover_lost_turn so its
emitted details structure matches INT-010’s expected /details/error/code path by
nesting engine_restart under error.code, or alternatively change the scenario
assertion to the existing flat error convention; keep the producer and assertion
consistent.
Comment on lines
+1042
to
+1072
| /// Invoke one delivery, treating an engine connection loss as a mid-flight | ||
| /// interruption instead of waiting out the full invocation timeout. The | ||
| /// engine's invocation routing dies with it, so the in-flight call's result | ||
| /// can never arrive; before this the crash left the delivery (and its whole | ||
| /// FIFO group) stranded for the 30-minute `harness-turn` budget | ||
| /// (iii-hq/workers#507). On loss, hold until the target function is | ||
| /// registered again — the same gate a freshly restored durable job passes | ||
| /// through — then re-invoke WITHOUT consuming a retry attempt: redelivered | ||
| /// turn steps are checkpointed and tolerate duplicate delivery (MOT-3944). | ||
| async fn invoke_message_across_engine_restarts( | ||
| queue: &str, | ||
| invoker: &Arc<dyn Invoker>, | ||
| message: &QueueMessage, | ||
| attempt: u32, | ||
| timeout_ms: u64, | ||
| poll_interval_ms: u64, | ||
| ) -> Result<Option<Value>, String> { | ||
| loop { | ||
| tokio::select! { | ||
| result = invoke_message(queue, invoker, message, attempt, timeout_ms) => return result, | ||
| () = invoker.connection_lost() => { | ||
| tracing::warn!( | ||
| queue = %queue, | ||
| function_id = %message.function_id, | ||
| "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; | ||
| } | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Unbounded re-invoke applies to every queue consumer, not just harness::turn.
Two things to bound here:
- The safety argument in the doc comment ("redelivered turn steps are checkpointed and tolerate duplicate delivery") is specific to
harness::turn, but this path runs for every function queue. A non-idempotent consumer registered on another queue now gets silently re-invoked after an engine restart, with no attempt consumed and therefore no DLQ escape hatch. - The loop has no restart budget. Combined with an epoch source that can move without a real restart (see the
min(connected_at_ms)note inharness/src/clients/engine.rs), a single flapping signal re-invokes indefinitely.
A cap on restart-driven re-invocations (falling through to the normal nack/retry path once exceeded) would contain both.
🛡️ Bounded restart re-invocation
+/// Restart-driven re-invocations tolerated before falling back to the
+/// ordinary retry/nack path.
+const MAX_RESTART_REINVOCATIONS: u32 = 3;
+
async fn invoke_message_across_engine_restarts(
queue: &str,
invoker: &Arc<dyn Invoker>,
message: &QueueMessage,
attempt: u32,
timeout_ms: u64,
poll_interval_ms: u64,
) -> Result<Option<Value>, String> {
+ let mut restarts = 0u32;
loop {
tokio::select! {
result = invoke_message(queue, invoker, message, attempt, timeout_ms) => return result,
() = invoker.connection_lost() => {
+ restarts += 1;
+ if restarts > MAX_RESTART_REINVOCATIONS {
+ return Err(format!(
+ "engine restarted {restarts} times with the invocation in flight; giving up on re-invoking"
+ ));
+ }
tracing::warn!(
queue = %queue,
function_id = %message.function_id,
+ restarts,
"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;
}
}
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| /// Invoke one delivery, treating an engine connection loss as a mid-flight | |
| /// interruption instead of waiting out the full invocation timeout. The | |
| /// engine's invocation routing dies with it, so the in-flight call's result | |
| /// can never arrive; before this the crash left the delivery (and its whole | |
| /// FIFO group) stranded for the 30-minute `harness-turn` budget | |
| /// (iii-hq/workers#507). On loss, hold until the target function is | |
| /// registered again — the same gate a freshly restored durable job passes | |
| /// through — then re-invoke WITHOUT consuming a retry attempt: redelivered | |
| /// turn steps are checkpointed and tolerate duplicate delivery (MOT-3944). | |
| async fn invoke_message_across_engine_restarts( | |
| queue: &str, | |
| invoker: &Arc<dyn Invoker>, | |
| message: &QueueMessage, | |
| attempt: u32, | |
| timeout_ms: u64, | |
| poll_interval_ms: u64, | |
| ) -> Result<Option<Value>, String> { | |
| loop { | |
| tokio::select! { | |
| result = invoke_message(queue, invoker, message, attempt, timeout_ms) => return result, | |
| () = invoker.connection_lost() => { | |
| tracing::warn!( | |
| queue = %queue, | |
| function_id = %message.function_id, | |
| "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; | |
| } | |
| } | |
| } | |
| } | |
| /// Restart-driven re-invocations tolerated before falling back to the | |
| /// ordinary retry/nack path. | |
| const MAX_RESTART_REINVOCATIONS: u32 = 3; | |
| /// Invoke one delivery, treating an engine connection loss as a mid-flight | |
| /// interruption instead of waiting out the full invocation timeout. The | |
| /// engine's invocation routing dies with it, so the in-flight call's result | |
| /// can never arrive; before this the crash left the delivery (and its whole | |
| /// FIFO group) stranded for the 30-minute `harness-turn` budget | |
| /// (iii-hq/workers#507). On loss, hold until the target function is | |
| /// registered again — the same gate a freshly restored durable job passes | |
| /// through — then re-invoke WITHOUT consuming a retry attempt: redelivered | |
| /// turn steps are checkpointed and tolerate duplicate delivery (MOT-3944). | |
| async fn invoke_message_across_engine_restarts( | |
| queue: &str, | |
| invoker: &Arc<dyn Invoker>, | |
| message: &QueueMessage, | |
| attempt: u32, | |
| timeout_ms: u64, | |
| poll_interval_ms: u64, | |
| ) -> Result<Option<Value>, String> { | |
| let mut restarts = 0u32; | |
| loop { | |
| tokio::select! { | |
| result = invoke_message(queue, invoker, message, attempt, timeout_ms) => return result, | |
| () = invoker.connection_lost() => { | |
| restarts += 1; | |
| if restarts > MAX_RESTART_REINVOCATIONS { | |
| return Err(format!( | |
| "engine restarted {restarts} times with the invocation in flight; giving up on re-invoking" | |
| )); | |
| } | |
| tracing::warn!( | |
| queue = %queue, | |
| function_id = %message.function_id, | |
| restarts, | |
| "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; | |
| } | |
| } | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@queue/src/runtime.rs` around lines 1042 - 1072, Bound the restart-driven
re-invocation loop in invoke_message_across_engine_restarts with a finite
counter or budget, and once exhausted stop re-invoking so the caller reaches the
normal nack/retry and DLQ handling. Remove or narrow the doc comment’s
harness::turn-specific safety claim, and ensure the limit applies uniformly to
all queue consumers.
This was referenced Aug 1, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Refs MOT-4107
Fixes #507.
Supersedes #521.
An engine restart during an in-flight function call no longer strands the durable queue delivery or leaves the conversation with an unclosed call.
What changed
engine_restarterror result, preserving at-most-once behavior when the target result is unknown.Integration coverage
Adds
INT-010to the checked-in deterministic integration suite. The scenario:call-1has one persistedengine_restartresult, the turn completed with nothing pending, lifecycle/traces are complete, and the transcript has no duplicate entries.Verification
INT-010: 6 consecutive passing SIGKILL/restart runs, about 5.3 seconds each.-D warnings.Summary by CodeRabbit
Bug Fixes
Tests