diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 250f11a24..ef795233e 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -5654,6 +5654,16 @@ local function tryServe(ckQueueName, mayRaiseFloor, knownRegistered) if gatedPending == nil then gatedPending = {} end table.insert(gatedPending, ckQueueName) end + -- NEW: report the gate the same way a future head is reported, so pass 1 declines to + -- spend a window slot on a candidate it cannot serve. Previously this fell through + -- returning nil and cost a slot, and because a gated variant's tag stops advancing it + -- also keeps sorting to the front and being revisited first, so enough of them + -- permanently consumed the fair pass and the scheduler ran on pass 2's age order + -- instead. Worse than that in the shape where the gated variants also hold the oldest + -- heads: pass 2's own window fills with them too and servable work behind them is + -- reached by neither pass for as long as the gate holds. Skipping without spending is + -- bounded by scanLimit, which is already the cap on how far pass 1 will read. + return 'notReady' end end diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeGatedOpBudget.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeGatedOpBudget.test.ts new file mode 100644 index 000000000..365d364f3 --- /dev/null +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtimeGatedOpBudget.test.ts @@ -0,0 +1,107 @@ +import { redisTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { Logger } from "@trigger.dev/core/logger"; +import { Decimal } from "@trigger.dev/database"; +import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js"; +import { RunQueue } from "../index.js"; +import { RunQueueFullKeyProducer } from "../keyProducer.js"; +const keys = new RunQueueFullKeyProducer(); +const baseEnv: any = { + id: "e1234", + type: "DEVELOPMENT", + maximumConcurrencyLimit: 100, + concurrencyLimitBurstFactor: new Decimal(1), + project: { id: "p1234" }, + organization: { id: "o1234" }, +}; +const QUEUE = "task/my-task"; +const mk = (o: any) => ({ + runId: "r1", + taskIdentifier: QUEUE, + orgId: "o1234", + projectId: "p1234", + environmentId: "e1234", + environmentType: "DEVELOPMENT", + queue: QUEUE, + timestamp: Date.now(), + attempt: 0, + ...o, +}); +// Declining to spend a window slot on a gated candidate means pass 1 reads further, so a +// fully-gated call costs more than it used to: measured 53 ops before and 80 after, the +// whole difference being SCARDs. That is the price of not silently degrading to age order, +// and it is worth paying because a fully-gated call serves nothing either way. What must +// not happen is the read running away, so this pins it against scanLimit (window * 2) +// rather than against the measured number, which would only be a tripwire. +describe("op count: fully gated dequeue", () => { + redisTest( + "pass 1 reads further when everything is gated, but stays inside scanLimit", + async ({ redisContainer }) => { + const MAX = 10; // window = 30, scanLimit = 60 + const GATED = 80; // more than scanLimit, so both bounds bind + const kp = "rq:opc:"; + const q: any = new RunQueue({ + name: "rq", + tracer: trace.getTracer("rq"), + workers: 1, + defaultEnvConcurrency: 100, + logger: new Logger("RunQueue", "error"), + retryOptions: { + maxAttempts: 5, + factor: 1.1, + minTimeoutInMs: 100, + maxTimeoutInMs: 1000, + randomize: true, + }, + keys, + masterQueueConsumersDisabled: true, + workerOptions: { disabled: true }, + ckVirtualTimeScheduling: { enabled: true, scanWindowMultiplier: 3 }, + queueSelectionStrategy: new FairQueueSelectionStrategy({ + redis: { keyPrefix: kp, host: redisContainer.getHost(), port: redisContainer.getPort() }, + keys, + }), + redis: { keyPrefix: kp, host: redisContainer.getHost(), port: redisContainer.getPort() }, + } as any); + const env = baseEnv; + await q.updateEnvConcurrencyLimits(env); + const shard = keys.masterQueueShardForEnvironment(env.id, 2); + const t0 = Date.now() - 5000000; + for (let i = 0; i < GATED; i++) { + await q.enqueueMessage({ + env, + message: mk({ runId: "g" + i, concurrencyKey: "gated-" + i, timestamp: t0 + i }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + const members = Array.from({ length: 105 }, (_, k) => "busy-" + i + "-" + k); + await q.redis.sadd( + keys.queueKey(env, QUEUE, "gated-" + i) + ":currentConcurrency", + ...members + ); + } + await q.redis.config("RESETSTAT"); + const served = await q.testDequeueFromMasterQueue(shard, env.id, MAX); + const info = await q.redis.info("commandstats"); + let total = 0; + const per: Record = {}; + for (const line of info.split("\n")) { + const m = line.match(/^cmdstat_([a-z|]+):calls=(\d+)/); + if (!m || ["info", "config"].includes(m[1])) continue; + per[m[1]] = parseInt(m[2], 10); + total += parseInt(m[2], 10); + } + // Nothing is servable, so the call is pure scan. + expect(served.length).toBe(0); + // window = MAX * 3 = 30, scanLimit = 60. Pass 1 may read up to scanLimit candidates and + // pass 2 up to its own window, one SCARD each, so that sum is the ceiling. + const scanLimit = MAX * 3 * 2; + const pass2Window = MAX * 3; + expect(per.scard ?? 0).toBeLessThanOrEqual(scanLimit + pass2Window); + // And it must read past the old window bound, or the fix is not in effect. + expect(per.scard ?? 0).toBeGreaterThan(MAX * 3); + await q.quit(); + }, + 120000 + ); +}); diff --git a/internal-packages/run-engine/src/run-queue/tests/ckVtimeGatedWindow.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckVtimeGatedWindow.test.ts new file mode 100644 index 000000000..fa030a786 --- /dev/null +++ b/internal-packages/run-engine/src/run-queue/tests/ckVtimeGatedWindow.test.ts @@ -0,0 +1,153 @@ +import { redisTest } from "@internal/testcontainers"; +import { trace } from "@internal/tracing"; +import { Logger } from "@trigger.dev/core/logger"; +import { Decimal } from "@trigger.dev/database"; +import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js"; +import { RunQueue } from "../index.js"; +import { RunQueueFullKeyProducer } from "../keyProducer.js"; + +// Devin's finding B on #4367: a concurrency-gated candidate returns from tryServe without +// the 'notReady' marker, so it spends one of pass 1's window slots even though it can +// never be served. A gated variant also stops advancing its tag, so it keeps sorting to +// the front and is revisited first on every call. Fill the window with them and pass 1 +// serves nothing, every call, and the scheduler silently degrades to pass 2's age order. +// +// Work conservation survives that, which is why it was originally waved through. What does +// not survive is the feature's whole purpose: fair order. This pins the difference. + +const testOptions = { + name: "rq", + tracer: trace.getTracer("rq"), + workers: 1, + defaultEnvConcurrency: 100, + logger: new Logger("RunQueue", "error"), + retryOptions: { + maxAttempts: 5, + factor: 1.1, + minTimeoutInMs: 100, + maxTimeoutInMs: 1000, + randomize: true, + }, + keys: new RunQueueFullKeyProducer(), +}; +const baseEnv: any = { + id: "e1234", + type: "DEVELOPMENT", + maximumConcurrencyLimit: 100, + concurrencyLimitBurstFactor: new Decimal(1), + project: { id: "p1234" }, + organization: { id: "o1234" }, +}; +const QUEUE = "task/my-task"; +const makeMessage = (o: any) => ({ + runId: "r1", + taskIdentifier: QUEUE, + orgId: "o1234", + projectId: "p1234", + environmentId: "e1234", + environmentType: "DEVELOPMENT", + queue: QUEUE, + timestamp: Date.now(), + attempt: 0, + ...o, +}); +const variantName = (ck: string) => testOptions.keys.queueKey(baseEnv, QUEUE, ck); + +function createQueue(rc: any, keyPrefix: string) { + return new RunQueue({ + ...testOptions, + masterQueueConsumersDisabled: true, + workerOptions: { disabled: true }, + ckVirtualTimeScheduling: { enabled: true, scanWindowMultiplier: 3 }, + queueSelectionStrategy: new FairQueueSelectionStrategy({ + redis: { keyPrefix, host: rc.getHost(), port: rc.getPort() }, + keys: testOptions.keys, + }), + redis: { keyPrefix, host: rc.getHost(), port: rc.getPort() }, + } as any) as any; +} + +describe("CK vtime: gated variants and the pass-1 window", () => { + redisTest( + "gated variants must not spend the fair pass's budget", + async ({ redisContainer }) => { + const MAX = 2; // window = MAX * 3 = 6 + const GATED = 8; // more than the window, all sorting ahead on tag + const queue = createQueue(redisContainer, "runqueue:test:gatedwin:"); + + try { + const env = baseEnv; + await queue.updateEnvConcurrencyLimits(env); + const shard = testOptions.keys.masterQueueShardForEnvironment(env.id, 2); + const t0 = Date.now() - 5_000_000; + + // Gated variants: queued work, but each parked at its per-key ceiling so it can + // never be served. Enqueued first so their heads are oldest too. + for (let i = 0; i < GATED; i++) { + await queue.enqueueMessage({ + env, + message: makeMessage({ + runId: `g-${i}`, + concurrencyKey: `gated-${i}`, + timestamp: t0 + i, + }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + } + + // "owed" is what fair order says to serve next: the lowest tag among servable + // variants. Its head is the NEWEST, so age order would put it last. + await queue.enqueueMessage({ + env, + message: makeMessage({ + runId: "owed-0", + concurrencyKey: "owed", + timestamp: t0 + 900_000, + }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + // "old" has the OLDEST head of the servable pair but a higher tag, so age order + // serves it first and fair order serves it second. + await queue.enqueueMessage({ + env, + message: makeMessage({ runId: "old-0", concurrencyKey: "old", timestamp: t0 + 100 }), + workerQueue: env.id, + skipDequeueProcessing: true, + }); + + // Park every gated variant at its ceiling. + const limit = 100; + for (let i = 0; i < GATED; i++) { + const members = Array.from({ length: limit + 5 }, (_, k) => `busy-${i}-${k}`); + await queue.redis.sadd(`${variantName(`gated-${i}`)}:currentConcurrency`, ...members); + } + + // Tags: gated variants lowest so they lead pass 1, then owed, then old. + const ckv = testOptions.keys.ckVtimeKeyFromQueue(variantName("owed")); + for (let i = 0; i < GATED; i++) await queue.redis.zadd(ckv, 0, variantName(`gated-${i}`)); + await queue.redis.zadd(ckv, 1, variantName("owed")); + await queue.redis.zadd(ckv, 5, variantName("old")); + + const served: string[] = []; + for (let c = 0; c < 4 && served.length < 2; c++) { + for (const m of await queue.testDequeueFromMasterQueue(shard, env.id, MAX)) { + served.push(m.message.concurrencyKey as string); + await queue.acknowledgeMessage(env.organization.id, m.messageId, { + skipDequeueProcessing: true, + }); + } + } + + // Fair order is the point of the feature: lowest tag first. If gated variants have + // eaten the window, pass 1 served nothing and pass 2's age order ran instead, + // which puts "old" first. + expect(served[0]).toBe("owed"); + } finally { + await queue.quit(); + } + }, + 60_000 + ); +}); diff --git a/internal-packages/run-engine/src/run-queue/tests/fairQueueSelectionStrategy.test.ts b/internal-packages/run-engine/src/run-queue/tests/fairQueueSelectionStrategy.test.ts index 5b2bdd8bf..6c388e753 100644 --- a/internal-packages/run-engine/src/run-queue/tests/fairQueueSelectionStrategy.test.ts +++ b/internal-packages/run-engine/src/run-queue/tests/fairQueueSelectionStrategy.test.ts @@ -230,7 +230,26 @@ describe("FairDequeuingStrategy", () => { envId: "env-3", }); - const startDistribute1 = performance.now(); + // Command counts rather than wall clock. What this test is really asserting is that + // the second call reuses the snapshot instead of rebuilding it, and the timing ratio + // it used to assert was a proxy for that: sub-millisecond durations compared as a + // ratio, which flakes the moment anything else is running on the box. Counting the + // commands the strategy issues measures the same thing and cannot be perturbed by + // load. + const counter = createRedisClient(redis); + const commandCount = async () => { + const info = await counter.info("commandstats"); + let total = 0; + for (const line of info.split("\n")) { + const m = line.match(/^cmdstat_([a-z|]+):calls=(\d+)/); + if (!m || ["info", "config"].includes(m[1])) continue; + total += parseInt(m[2], 10); + } + return total; + }; + + await counter.config("RESETSTAT"); + const before1 = await commandCount(); const envResult = await strategy.distributeFairQueuesFromParentQueue( "parent-queue", @@ -238,9 +257,7 @@ describe("FairDequeuingStrategy", () => { ); const result = flattenResults(envResult); - const distribute1Duration = performance.now() - startDistribute1; - - console.log("First distribution took", distribute1Duration, "ms"); + const distribute1Commands = (await commandCount()) - before1; expect(result).toHaveLength(3); // Should only get the two oldest queues @@ -249,33 +266,32 @@ describe("FairDequeuingStrategy", () => { const queue3 = keyProducer.queueKey("org-3", "proj-3", "env-3", "queue-3"); expect(result).toEqual([queue2, queue1, queue3]); - const startDistribute2 = performance.now(); + const before2 = await commandCount(); const _result2 = await strategy.distributeFairQueuesFromParentQueue( "parent-queue", "consumer-1" ); - const distribute2Duration = performance.now() - startDistribute2; + const distribute2Commands = (await commandCount()) - before2; - console.log("Second distribution took", distribute2Duration, "ms"); + // Reused snapshot: the second call does materially less Redis work than the first. + expect(distribute2Commands).toBeLessThan(distribute1Commands / 2); - // Make sure the second call is more than 2 times faster than the first - expect(distribute2Duration).toBeLessThan(distribute1Duration / 2); - - const startDistribute3 = performance.now(); + const before3 = await commandCount(); const _result3 = await strategy.distributeFairQueuesFromParentQueue( "parent-queue", "consumer-1" ); - const distribute3Duration = performance.now() - startDistribute3; + const distribute3Commands = (await commandCount()) - before3; - console.log("Third distribution took", distribute3Duration, "ms"); + // The snapshot has aged out by now, so the third call rebuilds it and pays the full + // cost again rather than the cached one. + expect(distribute3Commands).toBeGreaterThan(distribute2Commands * 2); - // Make sure the third call is more than 4 times the second - expect(distribute3Duration).toBeGreaterThan(distribute2Duration * 2); + await counter.quit(); } );