feat(run-engine): queue gates enforced from the message payload - #4826
feat(run-engine): queue gates enforced from the message payload#4826matt-aitken wants to merge 8 commits into
Conversation
|
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughAdds optional queue gates to run payloads, with up to three requested gate definitions plus the inline-limit gate. Adds configuration to enable gate enforcement. Updates enqueue and dequeue Lua scripts to check capacity and acquire gate slots. Updates completion, retry, dead-letter, expiry, and concurrency cleanup paths to release gate slots. Adds stale-member reconciliation and Redis-backed tests for gate and total-concurrency behavior. Merge Risk: 🔵 Low · up to Queue gates add optional cross-queue concurrency control. When enabled, gate holders may be charged against a queue total-concurrency budget even when that budget is otherwise disabled, which can cause unnecessary run queuing; the feature remains disabled by default. 🚥 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 |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
1c73107 to
d1de69a
Compare
d1de69a to
209641d
Compare
209641d to
f3b6b77
Compare
@trigger.dev/build
trigger.dev
@trigger.dev/core
@trigger.dev/python
@trigger.dev/react-hooks
@trigger.dev/redis-worker
@trigger.dev/rsc
@trigger.dev/schema-to-json
@trigger.dev/sdk
commit: |
cc8bc17 to
2a26cc5
Compare
8ae0f80 to
2d5a654
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
eb98fd1 to
da7dcb8
Compare
A gate is another declared queue a run must also hold a concurrency slot in
while it executes, carried in the message payload as gates =
[{queue, concurrencyKey?}] (at most two). Admit paths (both dequeue scripts
and every enqueue fast path) check each gate's per-key limit and, for keyed
entries, its total limit, then acquire a slot in each gate. Every release
path removes the run from its gate sets, driven by the payload itself with a
cheap substring probe before decoding, so gateless messages pay near zero
and releases stay correct across flag toggles.
Enforcement sits behind RUN_ENGINE_QUEUE_GATES_ENABLED (default off).
Saturated gates self-heal with the same bounded message-key reconciliation
as the total cap. The SDK and trigger surface that populates gates follows
on top of this.
…r own slot A gate without a concurrencyKey now resolves to the run's own key inside the scripts, matching the documented semantics for every caller and for payloads written before a fix at the contract layer. Admission also passes a gate or total cap whose set already contains the message: an unmirrored release from an older build leaves the run's own membership behind, and blocking on it would deadlock the run against itself; since acquiring is an idempotent set add, an existing member simply keeps the slot it holds. Gate names and keys in the payload schema are bounded to 128 characters.
A member whose message still exists but is sitting in a queue zset holds no legitimate slot: queued and in-flight are mutually exclusive on every path. The bounded reconcile now prunes those members too (the stored payload's queue field is the full variant key, so one ZSCORE answers it), which frees a saturated cap even when the self-holding run's variant lies beyond the dequeue candidate window and can never exempt itself. Covered by a test where the leaked member belongs to a different queue entirely.
…ncurrency set A member left behind by a release path without the mirror (an older build during a rolling upgrade) was only pruned if its message key was deleted or the run was re-queued. Dead-lettered runs keep their message key and sit in no queue, so they held a combined-cap slot forever; suspended runs held one until their terminal ack. Group and gate sets are strict mirrors of the home queue's currentConcurrency set, so the reconcile now prunes any member absent from it, healing every mirror-less path within one pass.
da7dcb8 to
8d9cc1e
Compare
1 participant
Summary
Stacked on #4823. Adds queue gates to the run queue: a run can name up to three additional declared queues (in practice a task's anonymous inline-limit gate plus two named limits) it must also hold a concurrency slot in while it executes. The run waits in its own queue as today; each gate contributes an extra admit condition and an extra slot held until release. This is the engine mechanism behind two long-requested shapes:
tenantqueue keyed by tenant id, so one tenant's total across all tasks is bounded), andproviderqueue).Gates ride the message payload as
gates: [{ queue, concurrencyKey? }]. At a gate, the gate queue'sconcurrencyLimitapplies per key value and itstotalConcurrencyLimitbounds the sum across keys, exactly the same semantics a run meets in its own queue. Enforcement is gated behindRUN_ENGINE_QUEUE_GATES_ENABLED(default off). The SDK and trigger API surface that populatesgatesfollows in a PR stacked on this one; nothing produces gated payloads yet.Design
Admit: both dequeue scripts already decode each message's payload, so the gate check reads it directly and skips a message whose gates are full (it stays queued in place). The enqueue fast paths probe the raw payload for
"gates"before decoding, so ungated messages pay a substring scan and nothing more. On admit the run is added to each gate's per-key concurrency set, and to the gate's group set when the entry is keyed, reusing the exact structures the base queue and total cap already use. Gate keys are built in Lua from the payload's own org/project/environment, so they always live in the message's hash slot.Release: every release path (ack, nack, dead-letter, concurrency release, sweeper clear, TTL expiry, re-enqueue cleanup) removes the run from its gate sets, driven by the payload rather than the flag, so slots acquired while the flag was on always drain after it is turned off. A saturated gate self-heals with the same bounded cursor reconciliation as the total cap in #4823: a member whose message key no longer exists was terminally released by a path without the removal and is pruned, one bounded batch per interval.
Head-of-line note: a message whose gate is full blocks the messages behind it in the same (sub-)queue. That is the intended trade and the contract PR enforces the rule that makes it benign: a gate's key must be a function of the run's own concurrency key, or a constant.
Covered by tests for unkeyed and keyed gates, flag-off behavior, nack release, fast-path fallback, and leaked-member reconciliation, plus the existing run-queue suite (129 tests).