fix(iii-queue): add per-queue dispatch_timeout_ms to bound lost dispatches#1929
fix(iii-queue): add per-queue dispatch_timeout_ms to bound lost dispatches#1929ytallo wants to merge 2 commits into
Conversation
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (11)
✅ Files skipped from review due to trivial changes (2)
🚧 Files skipped from review as they are similar to previous changes (9)
📝 WalkthroughWalkthroughThe PR adds timeout-aware engine invocation, threads queue dispatch deadlines from configuration into worker execution, and updates tests and docs to cover timeout expiry, cleanup, and the unset-timeout path. ChangesDispatch timeout flow
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning Tools execution failed with the following error: Failed to run tools: 13 INTERNAL: Received RST_STREAM with code 2 (Internal server error) 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 |
ytallo
force-pushed
the
bugfix/mot-3786-fn-queue-dispatch-timeout
branch
from
62ce7b1 to
773d307
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
engine/src/workers/queue/queue.rs (1)
835-846: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winBlock
dispatch_timeout_mson FIFO queues
dispatch_timeout_msreleases the only FIFO permit while the original worker-side invocation can still be running, so a later message in the samemessage_group_fieldcan start before the first attempt finishes. That breaks the strict per-group ordering guarantee; rejectfifo+dispatch_timeout_msin validation or document the limitation clearly.🤖 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 `@engine/src/workers/queue/queue.rs` around lines 835 - 846, The FIFO queue path in queue::worker::queue::dispatch setup allows dispatch_timeout_ms to be used together with fifo, which can release the single permit before the original invocation finishes and violate per-group ordering. Update the validation/config handling around QueueConfig and the prefetch/dispatch_timeout logic to reject or disable dispatch_timeout_ms whenever config.r#type is fifo, and make the constraint explicit near the dispatch_timeout variable setup so the behavior is enforced consistently.
🤖 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.
Outside diff comments:
In `@engine/src/workers/queue/queue.rs`:
- Around line 835-846: The FIFO queue path in queue::worker::queue::dispatch
setup allows dispatch_timeout_ms to be used together with fifo, which can
release the single permit before the original invocation finishes and violate
per-group ordering. Update the validation/config handling around QueueConfig and
the prefetch/dispatch_timeout logic to reject or disable dispatch_timeout_ms
whenever config.r#type is fifo, and make the constraint explicit near the
dispatch_timeout variable setup so the behavior is enforced consistently.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 9775dac0-94b0-42cd-a782-424f3646f1be
📒 Files selected for processing (9)
engine/src/engine/mod.rsengine/src/invocation/mod.rsengine/src/worker_connections/mod.rsengine/src/workers/queue/README.mdengine/src/workers/queue/config.rsengine/src/workers/queue/queue.rsengine/src/workers/queue/skills/SKILL.mdengine/tests/common/queue_helpers.rsengine/tests/queue_integration.rs
✅ Files skipped from review due to trivial changes (1)
- engine/src/workers/queue/skills/SKILL.md
There was a problem hiding this comment.
🧹 Nitpick comments (3)
engine/src/workers/queue/subscriber_config.rs (1)
122-136: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winSilent zero-timeout handling diverges from
FunctionQueueConfig's hard validation.
dispatch_timeout()treatsdispatch_timeout_ms: Some(0)as invalid by logging a warning and silently returningNone(unbounded). By contrast,FunctionQueueConfigvalidation (engine/src/workers/queue/config.rs) rejects the same value outright withanyhow::bail!. This means a subscriber-config user who setsdispatchTimeoutMs: 0gets no dispatch timeout at all with only a log line, whereas a fn-queue user gets a hard config-load failure for the identical mistake. This is called out as intentional in the commit summary, but the asymmetry between the two nearly-identical config surfaces could confuse operators debugging why their subscriber queue "isn't timing out."Consider surfacing this at config-validation time (where the caller has more context to fail loudly) rather than only warning when
dispatch_timeout()happens to be invoked.🤖 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 `@engine/src/workers/queue/subscriber_config.rs` around lines 122 - 136, The zero-value handling in dispatch_timeout() is only warning and silently falling back to unbounded behavior, which is inconsistent with FunctionQueueConfig validation. Move the invalid dispatch_timeout_ms == Some(0) check into subscriber config validation for SubscriberConfig/dispatch_timeout_ms so the config load fails loudly with a clear error, and keep dispatch_timeout() only responsible for converting already-validated values into Duration.engine/src/workers/queue/adapters/builtin/adapter.rs (2)
704-926: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd a
dispatch_timeout_mscoverage case forsubscribe()— The current adapter tests only exercisedispatch_timeout_ms: None; add one topic-subscribe case with a wedged handler to assert the timeout is passed through tocall_with_timeoutand drives nack/DLQ behavior.🤖 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 `@engine/src/workers/queue/adapters/builtin/adapter.rs` around lines 704 - 926, The subscribe tests in adapter.rs do not cover dispatch_timeout_ms being set, so add a new subscribe case with a wedged handler and a non-None dispatch_timeout_ms to verify the timeout is propagated through FunctionHandler/call_with_timeout and results in the expected nack/DLQ behavior. Use the existing builtin adapter test setup around make_adapter, subscribe, and FunctionHandler to locate the change, and assert both the timeout path and the DLQ side effect.
225-242: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valueDocument
dispatchTimeoutMsin subscriberqueue_configexamples.durable:subscriberalready accepts this key, and the SDK builders passqueue_configthrough as a generic payload, so the gap is in the docs/examples (for example,docs/next/creating-workers/queues.mdxanddocs/next/api-reference/sdk-rust.mdx).🤖 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 `@engine/src/workers/queue/adapters/builtin/adapter.rs` around lines 225 - 242, The subscriber queue_config examples are missing documentation for dispatchTimeoutMs even though durable:subscriber supports it and subscribe() already reads it via dispatch_timeout(); update the relevant docs/examples (such as the queue guide and Rust SDK reference) to show dispatchTimeoutMs in queue_config payloads, using the existing subscribe/SubscriberQueueConfig terminology so readers can connect the docs to the runtime behavior.Source: Path instructions
🤖 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.
Nitpick comments:
In `@engine/src/workers/queue/adapters/builtin/adapter.rs`:
- Around line 704-926: The subscribe tests in adapter.rs do not cover
dispatch_timeout_ms being set, so add a new subscribe case with a wedged handler
and a non-None dispatch_timeout_ms to verify the timeout is propagated through
FunctionHandler/call_with_timeout and results in the expected nack/DLQ behavior.
Use the existing builtin adapter test setup around make_adapter, subscribe, and
FunctionHandler to locate the change, and assert both the timeout path and the
DLQ side effect.
- Around line 225-242: The subscriber queue_config examples are missing
documentation for dispatchTimeoutMs even though durable:subscriber supports it
and subscribe() already reads it via dispatch_timeout(); update the relevant
docs/examples (such as the queue guide and Rust SDK reference) to show
dispatchTimeoutMs in queue_config payloads, using the existing
subscribe/SubscriberQueueConfig terminology so readers can connect the docs to
the runtime behavior.
In `@engine/src/workers/queue/subscriber_config.rs`:
- Around line 122-136: The zero-value handling in dispatch_timeout() is only
warning and silently falling back to unbounded behavior, which is inconsistent
with FunctionQueueConfig validation. Move the invalid dispatch_timeout_ms ==
Some(0) check into subscriber config validation for
SubscriberConfig/dispatch_timeout_ms so the config load fails loudly with a
clear error, and keep dispatch_timeout() only responsible for converting
already-validated values into Duration.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 93be002d-d55a-4ad5-b09a-5014d78c5f3b
📒 Files selected for processing (2)
engine/src/workers/queue/adapters/builtin/adapter.rsengine/src/workers/queue/subscriber_config.rs
…tches A queued fn-queue invocation dispatched to a worker could stay unacked until RabbitMQ's consumer_timeout (~30 min) if the dispatch was lost or the worker never returned a result: the message's ack/nack was bound to an unbounded await, and the retry->DLQ path is nack-driven, so a dispatched-but-uncompleted message was never retried and pinned one of the queue's concurrency slots for the whole window (#1927). Add an opt-in per-queue dispatch_timeout_ms. The fn-queue consumer now dispatches via a new Engine::call_with_timeout, which bounds the wait, evicts the abandoned invocation from both tracking structures (the engine invocations map and the owning worker's in-flight set), and returns a dispatch_timeout error that flows through the existing nack->retry->DLQ path and releases the concurrency slot. - 0 is rejected at both the JSON schema and config.yaml seed layers - unset preserves the historical unbounded behaviour - docs warn the timeout does not cancel worker-side work, so bounded handlers must be idempotent Closes #1927
Rename the subscriber config's visibilityTimeout key (parsed but never enforced) to dispatchTimeoutMs and thread it into the builtin adapter's FunctionHandler, which now dispatches via Engine::call_with_timeout so subscriber deliveries get the same bounded-dispatch semantics as the fn-queue consumer path. - 0 is rejected with a warning and treated as unbounded - unset preserves the historical unbounded behaviour
ytallo
force-pushed
the
bugfix/mot-3786-fn-queue-dispatch-timeout
branch
from
2536b7f to
74026c7
Compare
2 participants
What
Adds an opt-in per-queue
dispatch_timeout_mstoqueue_configsentries. When set, a fn-queue invocation that has been dispatched but not completed within the deadline is abandoned and nacked back through the normal retry→DLQ path, releasing its concurrency slot. When unset (the default), behaviour is unchanged.Engine::call_with_timeout(function_id, input, Option<Duration>): bounds the invocation, and on elapse evicts the abandoned invocation from both tracking structures — the engine-global invocations map and the owning worker's in-flight set — then returns adispatch_timeouterror.EngineTrait::callnow delegates to it withNone(no behaviour change for other callers).QueueWorker::spawn_consumer) dispatches through it, so a timeout lands in the existing nack→retry→DLQ arm.dispatch_timeout_ms: 0is rejected at both the JSON schema (range(min = 1)) and theconfig.yamlseed path (QueueModuleConfig::validate), mirroring the existingconcurrency/max_priorityguards.Why
Fixes #1927. When the engine dequeues a fn-queue message and the engine→worker dispatch is lost (or the worker never returns a result), the message's ack/nack was gated behind an unbounded await: the fn-queue consumer awaited the invocation with no deadline, and for a worker-routed (deferred) invocation the engine parked on a oneshot that only resolves on a worker reply or a detected disconnect. A half-open connection resolves neither. Since
max_retries/backoff_ms/DLQ are nack-driven, a dispatched-but-uncompleted message was never retried — it stayedunackeduntil RabbitMQ's brokerconsumer_timeout(~30 min), pinned one of the queue'sconcurrencyslots the whole time, and survived worker restarts because the engine holds the AMQP consumer.Notes
queue.rs). The topic/subscriber (durable:subscriber) delivery path has the same unbounded-await exposure and is tracked separately (MOT-3801), along with centralizing the deadline inInvocationHandlerand active worker-liveness detection.unwrap_or(Uuid::new_v4())inhandle_invocationthat would otherwise draw a discarded UUID per invocation now that the id is generated up front.Test plan
cargo test -p iii --lib— 1900 passed (includes 3 newcall_with_timeoutunit tests: deferred-eviction from the invocations map, worker in-flight set cleanup,Nonestays unbounded)cargo test -p iii --test queue_integration— 15 passed (new: wedged handler dead-letters via timeout; timeout releases the concurrency slot for a queued healthy message; unset timeout leaves a slow handler unbounded)Some(0)-rejection unit testscargo fmt --check,cargo clippyclean on touched filesmessages_unacknowledgeddrains at the configured deadline instead of ~30 minSummary by CodeRabbit
dispatch_timeout_ms(JSON:dispatchTimeoutMs), applied to dispatched work.