(MOT-4107) fix(harness): recover calls safely after engine restarts - #630
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
skill-check — worker0 verified, 50 skipped (no docs/).
Four for four. Nicely done. |
|
Warning Review limit reached
Next review available in: 33 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
📝 WalkthroughWalkthroughThe change adds engine-epoch restart detection, bounded transient retries, configurable queue redelivery, engine fault injection, and INT-010 integration coverage for recovering interrupted turns. ChangesEngine restart recovery
Estimated code review effort: 5 (Critical) | ~90 minutes Possibly related PRs
Sequence Diagram(s)sequenceDiagram
participant ScenarioRunner
participant ScenarioProbe
participant Stack
participant QueueRuntime
participant EngineClient
ScenarioRunner->>ScenarioProbe: wait for target call
ScenarioProbe-->>ScenarioRunner: target call recorded
ScenarioRunner->>Stack: kill and respawn engine
Stack-->>QueueRuntime: engine available
QueueRuntime->>EngineClient: redeliver interrupted turn
EngineClient-->>QueueRuntime: engine_restart closure or recovered result
QueueRuntime-->>ScenarioRunner: completed recovery result
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 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: 1
🧹 Nitpick comments (10)
queue/src/trigger.rs (1)
164-166: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
connection_epochcan never returnErr.
engine_epoch_msmaps every failure toNone, soIiiInvoker::connection_epochalways returnsOk. Ininvoke_checkpointed_message_across_engine_restartsthe?branch is therefore unreachable, and a probe failure is indistinguishable from "restart watching disabled".That conflation is acceptable for the current callers, because both cases fall back to a plain invoke. If you want the distinction later, propagate the trigger error instead of discarding it.
🤖 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 164 - 166, Update IiiInvoker::connection_epoch to use a non-Result return type matching engine_epoch_ms, removing the unreachable error branch and adjusting invoke_checkpointed_message_across_engine_restarts to handle the optional epoch directly. Preserve the existing fallback to plain invocation for both a missing epoch and a failed probe.queue/src/runtime.rs (4)
1098-1100: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winMake the
select!biased so a completed invocation wins over a simultaneous restart signal.
tokio::select!polls its branches in random order. Ifinvoke_messageandconnection_lost_sinceboth become ready in the same poll, the loop can discard a completed invocation result and re-invoke the target. That duplicates the side effect even though the original call finished.
harness::turntolerates a duplicate step, so the impact is bounded today. Addingbiased;makes the preference deterministic: an available result always wins.♻️ Proposed change
tokio::select! { + biased; result = invoke_message(queue, invoker, message, attempt, timeout_ms) => return result, () = invoker.connection_lost_since(baseline) => {🤖 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 1098 - 1100, Update the tokio::select! in the invocation loop around invoke_message and connection_lost_since to use biased polling, placing the bias directive before the branches so a completed invoke_message result is selected over a simultaneous restart signal.
954-963: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider grouping the scalar consumer parameters into a struct.
process_standard_messagenow takes five scalars andprocess_fifo_messagetakes six, ending with a positionalbool. Call sites pass3, 1, 1, 1_800_000, truewith no field names. A swappedpoll_interval_msandbackoff_ms, or an invertedbool, compiles cleanly.The values all originate from
FunctionQueueConfig. Passing a small borrowed settings struct instead of loose scalars would remove that class of mistake.Also applies to: 992-1003
🤖 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 954 - 963, Group the scalar consumer settings currently passed to process_standard_message and process_fifo_message into a small borrowed configuration struct, using the corresponding fields from FunctionQueueConfig such as retry, polling, timeout, backoff, and redelivery settings. Update both function signatures and all call sites to pass the struct, preserving the existing behavior while eliminating positional scalar and boolean arguments.
1789-1814: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThis test does not cover the opt-in race it appears to model.
RestartSignalButSuccessfulInvokerresolvesconnection_lost_sinceafter 2 ms whilecallneeds 20 ms. The test passesfalseforredeliver_on_engine_restart, so neither hook is consulted, and the invoker's restart signal is never observed. The test proves the opt-out path, which matches its name.No test covers
redeliver_on_engine_restart: truewith an invocation that succeeds while a restart signal is already pending. That is exactly theselect!race noted oninvoke_checkpointed_message_across_engine_restarts. Add a case that reuses this invoker withtrueand assertscalls == 1.🤖 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 1789 - 1814, Update the test around standard delivery to cover the opt-in restart behavior: pass true for redeliver_on_engine_restart while reusing RestartSignalButSuccessfulInvoker, and add a case asserting invoker.calls remains 1 after successful completion. Preserve the existing false/normal-semantics test and its acknowledgment assertions.
1079-1090: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument the multiplicative invocation bound for FIFO queues.
process_fifo_messagetreats an exhausted restart budget as an ordinary invocation failure. It then retries in place, and each in-place retry enters this loop with a fresh budget. Forharness-turnwithmax_retries: 3andMAX_RESTART_REDELIVERIES: 3, one message can reach(3 + 1) * (3 + 1) = 16target invocations before the DLQ.The total is bounded, so a FIFO group cannot be pinned forever. The current doc comment does not state the interaction with the queue retry budget. Add that to the comment so the effective ceiling is clear to the next reader.
🤖 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 1079 - 1090, Update the doc comment for invoke_checkpointed_message_across_engine_restarts to document that exhausting the restart budget becomes an ordinary invocation failure, which can trigger the queue’s in-place retry budget and re-enter this loop with a fresh restart budget; state the resulting multiplicative ceiling as (max_retries + 1) × (MAX_RESTART_REDELIVERIES + 1), while preserving the bounded FIFO-group behavior.harness/src/clients/engine.rs (3)
61-100: 🚀 Performance & Scalability | 🔵 Trivial | ⚖️ Poor tradeoffConsider caching the engine epoch instead of probing it on every dispatch.
dispatchnow performs a blockingengine::workers::listround trip before every invocation, andengine_link_interruptedthen issues one more probe per second for the whole dispatch lifetime. Every tool call therefore pays one extra RPC of added latency, and each in-flight call adds sustained 1 Hz load onengine::workers::list. With many concurrent calls and long tool timeouts this cost grows linearly with concurrency.A single shared epoch watcher per client, or a short-TTL cached epoch, would give the same restart signal at constant cost. The baseline can still be read from the cache because the epoch only changes on restart.
This is a refactor suggestion, not a blocker; the current behavior is correct.
🤖 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 61 - 100, Refactor EngineClient dispatch restart detection to reuse a shared cached engine epoch instead of calling engine_epoch_ms before every invocation. Add a single watcher or short-TTL cache associated with the client, have dispatch read the cached baseline, and make engine_link_interrupted consume the same cached state while preserving restart interruption behavior and the existing fallback when no epoch is available.
165-198: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated engine-epoch contract in two crates. Both files define
ENGINE_EPOCH_PROBE_INTERVAL_MS,ENGINE_EPOCH_PROBE_TIMEOUT_MS, and anengine_epoch_msthat queriesengine::workers::listand takes the minimumconnected_at_msover workers withruntime == "engine". The restart signal is one shared contract with the engine, implemented twice. If the engine changes the runtime label, the field name, or the response shape, one copy can be updated and the other left behind, and restart detection then differs between the harness and the queue consumer. Only the harness copy has a unit test.
harness/src/clients/engine.rs#L165-L198: extract the constants,engine_epoch_ms, andparse_engine_epochinto a shared helper that both crates depend on, and keepengine_epoch_uses_the_oldest_in_process_engine_workeras its test.queue/src/trigger.rs#L32-L60: replace the local constants and the inlined parse with a call into that shared helper, so the queue invoker andEngineClient::dispatchcannot diverge.If no crate is shared by both today, add a unit test for the parse in
queue/src/trigger.rsas the smaller step.🤖 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 165 - 198, The engine-epoch contract is duplicated between harness/src/clients/engine.rs#L165-L198 and queue/src/trigger.rs#L32-L60. Extract ENGINE_EPOCH_PROBE_INTERVAL_MS, ENGINE_EPOCH_PROBE_TIMEOUT_MS, engine_epoch_ms, and parse_engine_epoch into a shared helper used by both crates, preserving engine_epoch_uses_the_oldest_in_process_engine_worker as its test; update queue/src/trigger.rs#L32-L60 to call that helper and remove its local constants and parsing. If no shared crate is available, instead add equivalent parse coverage in queue/src/trigger.rs.
190-198: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winEpoch detection silently disables when the worker shape changes.
parse_engine_epochreturnsNonewhen no worker reportsruntime == "engine"or whenconnected_at_msis missing.dispatchthen falls back to the plain SDK timeout, andengine_link_interruptednever trips. That is a safe fallback, but the failure is invisible: a future engine that renames the runtime label or the field disables restart recovery with no signal.Consider logging once at
warnlevel when the preflight epoch cannot be read, so the loss of restart detection is observable.🤖 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 190 - 198, Update the preflight flow around parse_engine_epoch and dispatch to emit a warn-level log once whenever the engine epoch cannot be determined because no matching worker or connected_at_ms value is available. Preserve the existing timeout fallback and engine_link_interrupted behavior, while avoiding repeated warnings for the same preflight failure.harness/tests/integration/src/fixtures/loading.rs (1)
147-166: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueOptional: also validate the restart delay against the scenario deadline.
This block catches fault authoring mistakes before the stack boots. One related mistake is still deferred to run time: a
restart_delay_msat or abovedeadlines.scenario_msalways burns the deadline inside the fault phase. The failure then surfaces as"fault restart delay exceeded scenario deadline"after a full stack boot.♻️ Proposed additional check
anyhow::ensure!( target.hold_response, "engine fault target must hold its response until SIGKILL" ); + anyhow::ensure!( + fault.restart_delay_ms < self.scenario.deadlines.scenario_ms, + "fault restart delay {}ms must be shorter than the scenario deadline {}ms", + fault.restart_delay_ms, + self.scenario.deadlines.scenario_ms + ); }🤖 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/fixtures/loading.rs` around lines 147 - 166, In the fault validation block for the scenario loading flow, add validation that fault.restart_delay_ms is strictly less than deadlines.scenario_ms, rejecting values at or above the scenario deadline with a clear authoring error. Keep the existing target, call-count, function, and hold-response checks unchanged.harness/tests/integration/src/stack/supervisor.rs (1)
177-185: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winConsider waiting for the engine listener after respawn.
boot_oncewaits for the TCP listener and checksearly_exitbefore it continues.respawn_enginereturns as soon as the child is spawned. If the respawned engine fails to bind the port, the fault phase blocks inbind_observersuntil the scenario deadline (120s for INT-010) instead of failing fast with the real cause.Reuse the boot readiness probe, or poll
early_exit()for a short bounded period after the respawn.🤖 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/stack/supervisor.rs` around lines 177 - 185, Update respawn_engine to wait for the respawned engine’s listener readiness before returning, reusing the boot_once readiness probe and early_exit check where possible. Ensure startup failures are surfaced promptly while keeping the wait bounded rather than blocking until the scenario deadline.
🤖 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/functions/turn.rs`:
- Around line 92-107: Narrow the Dependency handling in is_transient_step_error
so “not connected” is considered transient only for the intended boot-race
prefixes, such as context::assemble or session::messages. Preserve the existing
exclusion for “enqueue harness::turn” and the function_not_found behavior, while
preventing unrelated dependency transport errors from triggering a whole-step
retry.
---
Nitpick comments:
In `@harness/src/clients/engine.rs`:
- Around line 61-100: Refactor EngineClient dispatch restart detection to reuse
a shared cached engine epoch instead of calling engine_epoch_ms before every
invocation. Add a single watcher or short-TTL cache associated with the client,
have dispatch read the cached baseline, and make engine_link_interrupted consume
the same cached state while preserving restart interruption behavior and the
existing fallback when no epoch is available.
- Around line 165-198: The engine-epoch contract is duplicated between
harness/src/clients/engine.rs#L165-L198 and queue/src/trigger.rs#L32-L60.
Extract ENGINE_EPOCH_PROBE_INTERVAL_MS, ENGINE_EPOCH_PROBE_TIMEOUT_MS,
engine_epoch_ms, and parse_engine_epoch into a shared helper used by both
crates, preserving engine_epoch_uses_the_oldest_in_process_engine_worker as its
test; update queue/src/trigger.rs#L32-L60 to call that helper and remove its
local constants and parsing. If no shared crate is available, instead add
equivalent parse coverage in queue/src/trigger.rs.
- Around line 190-198: Update the preflight flow around parse_engine_epoch and
dispatch to emit a warn-level log once whenever the engine epoch cannot be
determined because no matching worker or connected_at_ms value is available.
Preserve the existing timeout fallback and engine_link_interrupted behavior,
while avoiding repeated warnings for the same preflight failure.
In `@harness/tests/integration/src/fixtures/loading.rs`:
- Around line 147-166: In the fault validation block for the scenario loading
flow, add validation that fault.restart_delay_ms is strictly less than
deadlines.scenario_ms, rejecting values at or above the scenario deadline with a
clear authoring error. Keep the existing target, call-count, function, and
hold-response checks unchanged.
In `@harness/tests/integration/src/stack/supervisor.rs`:
- Around line 177-185: Update respawn_engine to wait for the respawned engine’s
listener readiness before returning, reusing the boot_once readiness probe and
early_exit check where possible. Ensure startup failures are surfaced promptly
while keeping the wait bounded rather than blocking until the scenario deadline.
In `@queue/src/runtime.rs`:
- Around line 1098-1100: Update the tokio::select! in the invocation loop around
invoke_message and connection_lost_since to use biased polling, placing the bias
directive before the branches so a completed invoke_message result is selected
over a simultaneous restart signal.
- Around line 954-963: Group the scalar consumer settings currently passed to
process_standard_message and process_fifo_message into a small borrowed
configuration struct, using the corresponding fields from FunctionQueueConfig
such as retry, polling, timeout, backoff, and redelivery settings. Update both
function signatures and all call sites to pass the struct, preserving the
existing behavior while eliminating positional scalar and boolean arguments.
- Around line 1789-1814: Update the test around standard delivery to cover the
opt-in restart behavior: pass true for redeliver_on_engine_restart while reusing
RestartSignalButSuccessfulInvoker, and add a case asserting invoker.calls
remains 1 after successful completion. Preserve the existing
false/normal-semantics test and its acknowledgment assertions.
- Around line 1079-1090: Update the doc comment for
invoke_checkpointed_message_across_engine_restarts to document that exhausting
the restart budget becomes an ordinary invocation failure, which can trigger the
queue’s in-place retry budget and re-enter this loop with a fresh restart
budget; state the resulting multiplicative ceiling as (max_retries + 1) ×
(MAX_RESTART_REDELIVERIES + 1), while preserving the bounded FIFO-group
behavior.
In `@queue/src/trigger.rs`:
- Around line 164-166: Update IiiInvoker::connection_epoch to use a non-Result
return type matching engine_epoch_ms, removing the unreachable error branch and
adjusting invoke_checkpointed_message_across_engine_restarts to handle the
optional epoch directly. Preserve the existing fallback to plain invocation for
both a missing epoch and a failed probe.
🪄 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: dca77451-449f-4127-845c-8e3721f625a4
📒 Files selected for processing (23)
harness/src/clients/engine.rsharness/src/functions/turn.rsharness/src/queue.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/dsl.rsharness/tests/integration/src/scenarios/engine_restart_recovery.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/adapter.rsqueue/src/runtime.rsqueue/src/trigger.rs
💤 Files with no reviewable changes (1)
- harness/tests/integration/src/process/supervisor.rs
1 participant
Refs MOT-4107
Fixes #507.
Supersedes #625.
An engine restart during an in-flight Harness function call now closes the interrupted call without replaying arbitrary queue consumers or reconstructing incomplete turn state.
What changed
harness-turnqueue.engine_restartresult, no duplicate messages, and a completed session.Verification
cargo test --libinharness(295 passed)cargo test --libinqueue(125 passed)context::assembleis held out while the restored turn retriescargo test --manifest-path tests/integration/Cargo.tomlcargo clippy --all-targets -- -D warningsinharnessandqueuecargo clippy --manifest-path tests/integration/Cargo.toml --all-targets -- -D warnings15dc993ebfdbfcabe5d299cf4cae4dd676db4c06(passed)Summary by CodeRabbit
New Features
Bug Fixes
Tests