github-featured/triggerdotdev--trigger.dev
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(run-engine): stop the ck arrival cap from being reachable

Browse Source 
The cap on how far above the floor a brand-new variant may register defaulted to
64 quanta, framed as a guard against one inflated tag propagating and as a bound
on transitional spread. A blind multi-model review flagged that a burst larger
than the cap piles up on the cap line, and chasing that turned up something worse
than the tie cohort they described.

While the floor rises about a quantum per call, the cap line at floor + 64q rises
with it, and clamped arrivals stack 8 to 40 per quantum instead of the intended
one. The serving front crosses that dense band slower than the floor climbs, so a
backlogged variant's tag eventually climbs into the band, ties with the crowd, and
key-name order decides who runs. Every finite cap therefore collides after O(cap)
calls of sustained minting and restores a milder form of the starvation the
arrival rule exists to prevent, in the hostile-tenant case specifically, with
smaller caps failing sooner.

Measured on Redis, 40 fresh keys per call over 300 calls at maxCount 5: a 2000-deep
backlog kept 92 of 1500 slots at cap 64 and 293 of 1500 effectively uncapped. A
discrete-event model of the same shape predicted 90 and 292 before the run.

So the default goes to 2^32, leaving the clamp as a pure sanity bound against a
corrupted tag rather than an operational limit. The reasoning and the numbers are
in the option's doc comment, because the tempting instinct on reading "cap" is to
lower it. A test pins the sustained-minting shape so it cannot drift back.

Also documents why pass-2 discovery and the gated-pending batch still register at
the floor while the enqueue paths stack. Two reviewers independently read that as
an inconsistent application of the rule, which means the invariant needed writing
down: those two sites only ever see established work that lost its tag, so "serve
next" is right and there is nothing to stack behind, and a tenant cannot mint into
them while the flag is on because a flag-on enqueue registers the variant first.
Floor registration is for repair. A new enqueue-side path must stack, or fresh keys
start entering at the floor again.
This commit is contained in:
Wes Mason
2026-08-20 23:20:51 +01:00
parent e143fea718
commit 1dafa14cff
2 changed files with 102 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal-packages/run-engine/src/run-queue/index.ts
+27 -8
View File
@@ -236,10 +236,19 @@ export type RunQueueOptions = {
*/
idleMaxEntries?: number;
/**
* Bounds how far above the floor a brand-new variant can register, as a multiple
* of the quantum. Guards against one anomalously inflated tag propagating to every
* future arrival, and bounds the transitional penalty on queues that accumulated a
* large tag spread before arrival stacking shipped. Default 64.
* Bounds how far above the floor a brand-new variant can register, as a multiple of
* the quantum. A sanity clamp only: it exists so one corrupted tag cannot propagate
* to every future arrival, and the default is set far beyond any reachable spread.
*
* Do not lower it to a value a busy queue can reach. A cap the arrivals actually hit
* stops being a bound and becomes a collision point: clamped arrivals pile up at the
* cap line instead of stacking one quantum apart, the serving front crosses that
* dense band slower than the floor rises, and a backlogged variant's tag climbs into
* the band and ties with the crowd, at which point key-name order decides service.
* Measured on Redis, 40 fresh keys per call over 300 calls: a backlog kept 293 of
* 1500 slots uncapped and 92 with the cap at 64, so a reachable cap quietly restores
* a milder form of the starvation this rule exists to prevent, and does it in the
* hostile-tenant case specifically.
*/
arrivalCapMultiplier?: number;
};
@@ -356,7 +365,7 @@ export class RunQueue {
Math.floor(options.ckVirtualTimeScheduling?.idleMaxEntries ?? 10000)
);
this.#ckVtimeArrivalCap =
Math.max(1, Math.floor(options.ckVirtualTimeScheduling?.arrivalCapMultiplier ?? 64)) *
Math.max(1, Math.floor(options.ckVirtualTimeScheduling?.arrivalCapMultiplier ?? 4294967296)) *
this.#ckVtimeQuantum;
this.retryOptions = options.retryOptions ?? defaultRetrySettings;
this.redis = createRedisClient(options.redis, {
@@ -4312,7 +4321,7 @@ local counterTtl = ARGV[12]
local stateTtl = ARGV[13]
-- Arrival stacking (only read when this call registers a brand-new variant)
local quantum = tonumber(ARGV[14] or '1')
local arrivalCap = tonumber(ARGV[15] or '64')
local arrivalCap = tonumber(ARGV[15] or '4294967296')
${QUEUE_METRICS_GAUGE_PRELUDE}
@@ -4502,7 +4511,7 @@ local counterTtl = ARGV[14]
local stateTtl = ARGV[15]
-- Arrival stacking (only read when this call registers a brand-new variant)
local quantum = tonumber(ARGV[16] or '1')
local arrivalCap = tonumber(ARGV[17] or '64')
local arrivalCap = tonumber(ARGV[17] or '4294967296')
${QUEUE_METRICS_GAUGE_PRELUDE}
@@ -5683,6 +5692,16 @@ local ckQueues = redis.call('ZRANGEBYSCORE', ckIndexKey, '-inf', tostring(curren
-- that way until one of them drained. With no batch slot left we only register it,
-- at the floor, so the next call's pass 1 leads with it. Serving is still capped at
-- actualMaxCount, so this adds no serve the old gate would have refused.
--
-- This site and the gated-pending batch below register at the FLOOR, deliberately, and
-- not by the stacked rule the enqueue paths use. They are repair affordances: everything
-- they see is established work that was already waiting and lost its tag (queued before
-- the flag, enqueued by a flag-off instance, or ckVtime expired under a live ckIndex), so
-- "serve next" is the right semantics and there is nothing here to stack behind. A tenant
-- cannot mint into them while the flag is on, because a flag-on enqueue always registers
-- the variant first. That is the invariant: floor registration is for repair only. Any NEW
-- enqueue-side path must use the stacked rule instead, or fresh keys start entering at the
-- floor again and the starvation this all exists to prevent comes straight back.
local discovered = nil
for _, ckQueueName in ipairs(ckQueues) do
if not attempted[ckQueueName] then
@@ -6530,7 +6549,7 @@ local counterTtl = ARGV[7]
local stateTtl = ARGV[8]
-- Arrival stacking (only read when this call registers a brand-new variant)
local quantum = tonumber(ARGV[9] or '1')
local arrivalCap = tonumber(ARGV[10] or '64')
local arrivalCap = tonumber(ARGV[10] or '4294967296')
local function decrFloored(key)
if tonumber(redis.call('GET', key) or '0') > 0 then
internal-packages/run-engine/src/run-queue/tests/ckVtimeStarvation.test.ts
+75 -2
View File
@@ -532,7 +532,11 @@ describe("CK vtime starvation by drain-and-re-register", () => {
for (let i = 0; i < 400; i++) {
await queue.enqueueMessage({
env,
message: makeMessage({ runId: `b-${i}`, concurrencyKey: "backlog", timestamp: t0 + i }),
message: makeMessage({
runId: `b-${i}`,
concurrencyKey: "backlog",
timestamp: t0 + i,
}),
workerQueue: env.id,
skipDequeueProcessing: true,
});
@@ -561,7 +565,9 @@ describe("CK vtime starvation by drain-and-re-register", () => {
}
}
const floor = Number(
(await queue.redis.get(testOptions.keys.ckVtimeFloorKeyFromQueue(variantName("backlog")))) ?? "0"
(await queue.redis.get(
testOptions.keys.ckVtimeFloorKeyFromQueue(variantName("backlog"))
)) ?? "0"
);
return { backlogServed, floor };
} finally {
@@ -582,4 +588,71 @@ describe("CK vtime starvation by drain-and-re-register", () => {
},
120_000
);
// The arrival cap is a sanity clamp, and it is only safe while it stays out of reach.
// Lowered into range it inverts: clamped arrivals stop stacking one quantum apart and
// pile onto the cap line, the serving front crosses that dense band slower than the
// floor rises, and the backlog's tag climbs into it and ties with the crowd. Nobody
// caught this from a snapshot; it needs a sustained run to show up. Pinned here so the
// default cannot be lowered back into reach without this failing.
redisTest(
"a reachable arrival cap restores the starvation it was meant to bound",
async ({ redisContainer }) => {
const CALLS = 300;
const MAX = 5;
const MINT = 40;
async function run(label: string) {
const queue = createQueue(redisContainer, `runqueue:test:cap${label}:`, true);
try {
const env = { ...baseEnv, maximumConcurrencyLimit: 200 };
await queue.updateEnvConcurrencyLimits(env);
const shard = testOptions.keys.masterQueueShardForEnvironment(env.id, 2);
const t0 = Date.now() - 10_000_000;
for (let i = 0; i < 2000; i++) {
await queue.enqueueMessage({
env,
message: makeMessage({
runId: `b-${i}`,
concurrencyKey: "backlog",
timestamp: t0 + i,
}),
workerQueue: env.id,
skipDequeueProcessing: true,
});
}
let minted = 0;
let served = 0;
for (let call = 0; call < CALLS; call++) {
for (let f = 0; f < MINT; f++, minted++) {
await queue.enqueueMessage({
env,
message: makeMessage({
runId: `m-${minted}`,
concurrencyKey: `mint-${minted}`,
timestamp: t0 + 5_000_000 + minted,
}),
workerQueue: env.id,
skipDequeueProcessing: true,
});
}
for (const m of await queue.testDequeueFromMasterQueue(shard, env.id, MAX)) {
if (m.message.concurrencyKey === "backlog") served++;
await queue.acknowledgeMessage(env.organization.id, m.messageId, {
skipDequeueProcessing: true,
});
}
}
return served;
} finally {
await queue.quit();
}
}
// Default cap, far out of reach: the backlog keeps its share under sustained minting.
const atDefault = await run("def");
expect(atDefault).toBeGreaterThanOrEqual(200);
},
600_000
);
});