fix(run-engine): stop concurrency-gated ck variants eating the fair pass

A candidate parked at its per-key concurrency ceiling fell out of tryServe
returning nil, so pass 1 spent one of its window slots on a variant it could not
serve. A gated variant's tag also stops advancing, so it keeps sorting to the
front of ckVtime and is revisited first on every call. Enough of them and pass 1
serves nothing, ever, and the scheduler quietly runs on pass 2's age order
instead.

The original note on this said work conservation still held because pass 2 fills
the batch, and that was the reason it was left alone. It does not hold. Where the
gated variants are also the oldest, which is the ordinary case since a variant
that has been queued longest is likely to be both old and saturated, pass 2's own
window fills with the same variants and servable work behind them is reached by
neither pass. The test added here starts from that shape and serves nothing at all
before the fix, rather than serving in the wrong order.

So a gated candidate now reports 'notReady', exactly as a future-scheduled head
already did, and pass 1 reads past it without spending a slot. The read is bounded
by scanLimit, which is already the cap on how far pass 1 will look.

It is not free. On a fully gated call pass 1 now reads to scanLimit instead of
stopping at the window: measured 53 Redis operations before and 80 after, every
one of the 27 a SCARD, so roughly 10 usec. That is worth paying, because a fully
gated call serves nothing either way, while a partially gated one goes from
serving nothing to serving in fair order. A second test pins the op count against
scanLimit plus the pass-2 window so the read cannot start running away.

Reported by Devin on #4367.

Author: 1stvamp

internal-packages/run-engine/src/run-queue/index.ts

@@ -5632,6 +5632,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
 

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<string, number> = {};
+      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
+  );
+});

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
+  );
+});

internal-packages/run-engine/src/run-queue/tests/fairQueueSelectionStrategy.test.ts

@@ -230,17 +230,34 @@ 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",
         "consumer-1"
       );
       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;
-
-      console.log("Second distribution took", distribute2Duration, "ms");
+      const distribute2Commands = (await commandCount()) - before2;
 
-      // Make sure the second call is more than 2 times faster than the first
-      expect(distribute2Duration).toBeLessThan(distribute1Duration / 2);
+      // Reused snapshot: the second call does materially less Redis work than the first.
+      expect(distribute2Commands).toBeLessThan(distribute1Commands / 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();
     }
   );