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): register a gated ck variant so it can rejoin the fair order

Browse Source 
tryServe marks a variant attempted before the per-key concurrency gate, and pass
2's discovery step skips anything attempted, so a variant that was both gated and
unregistered fell through every route: pass 1 could not see it without a ckVtime
entry, pass 2's attempt was a no-op behind the gate, and discovery then passed
over it. It stayed invisible to the fair pass on every call for as long as the
gate held, and only a serve would have registered it.

Unregistered only arises where a variant reached ckIndex without a vtime-aware
write, which is the rollout case discovery already exists to repair: a backlog
queued before the flag went on, an enqueue from an instance that still has it
off, or a ckVtime that expired while ckIndex lived. Registration is NX so an
already-registered variant keeps the tag it earned, and the state TTL is only
written when the ZADD actually registered something, which is the path that can
recreate a ckVtime key that expired out from under a live ckIndex.

Adds a regression test that fails without the branch, and a second op-count
budget covering the all-unservable scan. The existing budget only bounds the
servable shape (its fixture acks immediately so nothing is ever gated or
deferred), and its comment read as a general worst case, which it is not.

Reported by Devin on #4367.
This commit is contained in:
Wes Mason
2026-08-16 19:46:34 +01:00
parent 129f8d5b86
commit b05f3f9244
3 changed files with 194 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal-packages/run-engine/src/run-queue/index.ts
+15
View File
@@ -5286,6 +5286,21 @@ local function tryServe(ckQueueName, mayRaiseFloor)
return 'notReady'
end
end
else
-- NEW: gated on the per-key ceiling, so nothing above ran, including the registration
-- that a serve would have done. Pass 2 marks a variant attempted before this gate, and
-- its discovery step skips anything attempted, so an UNREGISTERED variant that is gated
-- was invisible to pass 1 and stayed that way on every call for as long as the gate
-- held. Unregistered here means it reached ckIndex without ever passing through a
-- vtime-aware write: a backlog queued before the flag went on, an enqueue from an
-- instance that still has it off, or a ckVtime that expired while ckIndex lived, which
-- are the same cases pass 2's discovery exists to repair. NX, so a variant that is
-- already registered keeps the tag it earned, which is the usual case and costs one op.
-- The TTL only needs setting when this actually registered something, since that is the
-- path that can recreate a ckVtime key which expired out from under a live ckIndex.
if redis.call('ZADD', ckVtimeKey, 'NX', tostring(floor), ckQueueName) == 1 then
redis.call('EXPIRE', ckVtimeKey, stateTtl)
end
end
end
internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts
+74
View File
@@ -1343,6 +1343,80 @@ describe("CK virtual-time (SFQ) dequeue", () => {
}
});
redisTest(
"a gated variant that was never registered still joins the fair order",
async ({ redisContainer }) => {
// Pass 2 marks a variant attempted before the per-key concurrency gate, and its
// discovery step skips anything attempted, so a variant that is BOTH unregistered and
// gated used to fall through every route: pass 1 cannot see it (no ckVtime entry),
// pass 2 attempts it and the gate makes that a no-op, and discovery then skips it for
// having been attempted. It stayed invisible to the fair pass for as long as the gate
// held, on every call.
//
// Unregistered only happens where a variant reached ckIndex without a vtime-aware
// write, which is the rollout case: a backlog queued before the flag went on, or an
// enqueue from an instance that still has it off. This fixture reproduces that by
// enqueueing through a flag-off queue and then dequeuing through a flag-on one.
const off = createQueue(redisContainer, null);
const on = createQueue(redisContainer);
try {
const t0 = Date.now() - 100_000;
for (let i = 0; i < 2; i++) {
await off.enqueueMessage({
env: authenticatedEnvDev,
message: makeMessage({
runId: `r-gated-${i}`,
concurrencyKey: "gated",
timestamp: t0 + i,
}),
workerQueue: authenticatedEnvDev.id,
skipDequeueProcessing: true,
});
}
const gatedVariant = variantName("gated");
const ckIndexKey = testOptions.keys.ckIndexKeyFromQueue(gatedVariant);
const ckVtimeKey = testOptions.keys.ckVtimeKeyFromQueue(gatedVariant);
// In ckIndex, and with no ckVtime entry, exactly as a pre-flag backlog looks.
expect(await on.redis.zscore(ckIndexKey, gatedVariant)).not.toBeNull();
expect(await on.redis.zscore(ckVtimeKey, gatedVariant)).toBeNull();
// Hold it at its per-key ceiling so every visit hits the gate.
await on.updateQueueConcurrencyLimits(authenticatedEnvDev, QUEUE, 1);
await on.redis.sadd(
testOptions.keys.queueCurrentConcurrencyKeyFromQueue(gatedVariant),
"occupant"
);
const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
const served = await on.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 1);
// Still unservable, so nothing comes back, but the gate no longer costs it its place
// in the fair order: it is registered at the floor and pass 1 can see it from now on.
expect(served.length).toBe(0);
expect(await on.redis.zscore(ckVtimeKey, gatedVariant)).not.toBeNull();
// Registering must not resurrect a ckVtime entry with no ckIndex member, and the key
// it may have just created has to carry the state TTL rather than leaking.
expect(await on.redis.zscore(ckIndexKey, gatedVariant)).not.toBeNull();
expect(await on.redis.ttl(ckVtimeKey)).toBeGreaterThan(0);
// Once the ceiling clears it serves, and from the fair pass rather than by age.
await on.redis.srem(
testOptions.keys.queueCurrentConcurrencyKeyFromQueue(gatedVariant),
"occupant"
);
const after = await on.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 1);
expect(after.map((m) => m.messageId)).toEqual(["r-gated-0"]);
} finally {
await off.quit();
await on.quit();
}
}
);
redisTest(
"a ckVtime entry stranded by an ack is collected by the next scan",
async ({ redisContainer }) => {
internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts
+105 -5
View File
@@ -346,11 +346,18 @@ describe("CK virtual-time concurrency and op-count budget", () => {
expect(off.served).toBe(cks.length * perKey);
expect(on.served).toBe(cks.length * perKey);
// Per dequeue call the vtime path adds at worst 8 fixed ops: GET floor,
// ZRANGE min, ZRANGE window, the pass-2 ZRANGEBYSCORE, the pass-2
// discovery ZADD, SET floor, EXISTS ckVtime, EXPIRE ckVtime, plus per
// serve one ZSCORE and one ZADD. The discovery ZADD is variadic, so it
// stays a single op however many variants one call registers.
// Per dequeue call the vtime path adds 8 fixed ops: GET floor, ZRANGE min,
// ZRANGE window, the pass-2 ZRANGEBYSCORE, the pass-2 discovery ZADD, SET
// floor, EXISTS ckVtime, EXPIRE ckVtime, plus per serve one ZSCORE and one
// ZADD. The discovery ZADD is variadic, so it stays a single op however many
// variants one call registers.
//
// This bounds the SERVABLE shape, which is what this fixture builds: every
// variant has ready work and is acked immediately, so both paths probe the
// same variants and neither pass walks past them. It is not a worst case. A
// call whose candidates are mostly unservable probes further than the flag-off
// command does, because pass 1 spends its window budget only on candidates it
// could serve; that shape is bounded separately by the test below.
const budget = dequeueCalls * (8 + 2 * maxCount);
expect(
on.totalCalls,
@@ -360,6 +367,99 @@ describe("CK virtual-time concurrency and op-count budget", () => {
await statsClient.quit();
}
});
redisTest(
"op-count budget: an all-unservable call stays inside the scan caps",
async ({ redisContainer }) => {
// The budget above measures the servable shape. This one measures the other end:
// every candidate has a future head, so pass 1 serves nothing and spends no window
// budget, which is exactly when it walks its whole scan rather than stopping at
// `window`. That is the shape the scan widening introduced, so it is the one worth
// pinning a ceiling on.
//
// The ceiling is structural rather than tuned: pass 1 probes at most scanLimit
// (2 * window) candidates and pass 2 at most pass2Window more, and the most any
// single unservable probe costs is SCARD + ZRANGEBYSCORE + ZRANGE + ZADD. Nothing
// here asserts the count is small, only that it cannot run past those caps.
const maxCount = 5;
const multiplier = 3;
const window = maxCount * multiplier;
const scanLimit = window * 2;
const pass2Window = Math.max(window, maxCount * 3);
// Far more future-headed variants than either cap can reach, so the caps are what
// stops the scan rather than the fixture running out of candidates.
const blockers = 200;
const statsClient = createRedisClient({
host: redisContainer.getHost(),
port: redisContainer.getPort(),
});
const queue = createQueue(redisContainer, "rq18worst:", true);
try {
const future = Date.now() + 60 * 60 * 1000;
for (let i = 0; i < blockers; i++) {
await queue.enqueueMessage({
env: authenticatedEnvDev,
message: makeMessage({
runId: `r-future-${i}`,
concurrencyKey: `a${String(i).padStart(3, "0")}`,
timestamp: future,
}),
workerQueue: authenticatedEnvDev.id,
skipDequeueProcessing: true,
});
}
// One ready variant, sorting after every blocker so it lands beyond the scan. It is
// what keeps the base queue selectable at all: with nothing ready the master queue
// score is in the future, the consumer never picks the queue, and the script does
// not run, which is its own (useful) form of backpressure but measures nothing.
await queue.enqueueMessage({
env: authenticatedEnvDev,
message: makeMessage({
runId: "r-ready",
concurrencyKey: "zready",
timestamp: Date.now() - 100_000,
}),
workerQueue: authenticatedEnvDev.id,
skipDequeueProcessing: true,
});
const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
await statsClient.call("CONFIG", "RESETSTAT");
const served = await queue.testDequeueFromMasterQueue(
shard,
authenticatedEnvDev.id,
maxCount
);
const used = totalCommandCalls(await statsClient.info("commandstats"));
// Pass 1 spent its whole scan on variants it could not serve and came back with
// nothing; the single ready variant was served by pass 2's age-order fill.
expect(served.map((m) => m.message.concurrencyKey)).toEqual(["zready"]);
const fixedOps = 16;
const perProbe = 4;
const servedOps = 16;
const ceiling = fixedOps + servedOps + perProbe * (scanLimit + pass2Window);
expect(
used,
`unservable-scan call used ${used} ops, ceiling ${ceiling}`
).toBeLessThanOrEqual(ceiling);
// The scan really did run deep rather than bailing early, so the ceiling above is
// measuring something.
expect(used).toBeGreaterThan(perProbe * window);
// And the caps are what bounded it: cost tracks the scan limits, not how many
// concurrency keys the queue happens to have.
expect(used).toBeLessThan(perProbe * blockers);
} finally {
await queue.quit();
await statsClient.quit();
}
}
);
});
// Sums calls= across every cmdstat_ line of INFO commandstats. Includes