From b05f3f92449f313af99be5a417cc6aa40035d6c0 Mon Sep 17 00:00:00 2001 From: Wes Mason Date: Sun, 16 Aug 2026 19:46:34 +0100 Subject: [PATCH] fix(run-engine): register a gated ck variant so it can rejoin the fair order 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. --- .../run-engine/src/run-queue/index.ts | 15 +++ .../src/run-queue/tests/ckVtime.test.ts | 74 ++++++++++++ .../tests/ckVtimeConcurrency.test.ts | 110 +++++++++++++++++- 3 files changed, 194 insertions(+), 5 deletions(-) diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 26ec1dd99..366e90a0e 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -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 diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts index 8ac6f0a5b..af20d4ce1 100644 --- a/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtime.test.ts @@ -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 }) => { diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts index 9dc0d31f7..788a2b45e 100644 --- a/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtimeConcurrency.test.ts @@ -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