fix(run-engine): bound the ck idle set by rank, not just by floor

The at-or-below-floor reap on :ckVtimeIdle is worth nothing while the floor is
pinned, and a workload that keeps minting fresh concurrency keys pins it
indefinitely: each new key registers at the floor and is served at it, so
minServableTag never rises. A resource benchmark caught the set growing by the
drain count every round and never shrinking, passing ckIndex in size by round
50 and reaching 12000 entries (1.77MB) over 60 rounds, with only the 24h state
TTL bounding it. The mechanism was measured rather than inferred: a probe
sampling the floor found it at 0 on every round while the lowest parked tag was
1, so ZREMRANGEBYSCORE could never match.

Adds a rank cap, keeping the highest idleMaxEntries tags (default 10000,
configurable), which does not depend on the floor moving. Trimming the lowest
tags first drops the entries nearest the floor, whose remembered credit is worth
least. Verified against an explicit cap of 3000: the set rises to it and stays
flat there across 12000 drains with the floor still pinned at 0.

The new ARGV is inserted before the metrics gauge arg, which has to stay last
because the gauge fragment reads ARGV[#ARGV].

Also drops the node:test describe import from the new test file, which shadows
vitest's own under globals:true. That is a wider pattern in this directory and
is left alone elsewhere.

Author: 1stvamp

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

@@ -228,6 +228,13 @@ export type RunQueueOptions = {
     scanWindowMultiplier?: number;
     /** EXPIRE applied to ckVtime/ckVtimeFloor on every write. Default 86400. */
     stateTtlSeconds?: number;
+    /**
+     * Hard cap on remembered tags in ckVtimeIdle, enforced by rank so it holds even
+     * when the floor is pinned and the at-or-below-floor reap cannot fire. The lowest
+     * tags are dropped first: they sit nearest the floor, so they are the entries whose
+     * credit is worth least. Default 10000.
+     */
+    idleMaxEntries?: number;
   };
 };
 
@@ -317,6 +324,7 @@ export class RunQueue {
   readonly #ckVtimeQuantum: number;
   readonly #ckVtimeWindowMultiplier: number;
   readonly #ckVtimeStateTtl: number;
+  readonly #ckVtimeIdleMaxEntries: number;
 
   constructor(public readonly options: RunQueueOptions) {
     this.shardCount = options.shardCount ?? 2;
@@ -335,6 +343,10 @@ export class RunQueue {
       1,
       Math.floor(options.ckVirtualTimeScheduling?.stateTtlSeconds ?? 86400)
     );
+    this.#ckVtimeIdleMaxEntries = Math.max(
+      1,
+      Math.floor(options.ckVirtualTimeScheduling?.idleMaxEntries ?? 10000)
+    );
     this.retryOptions = options.retryOptions ?? defaultRetrySettings;
     this.redis = createRedisClient(options.redis, {
       onError: (error) => {
@@ -2647,6 +2659,7 @@ export class RunQueue {
             String(this.#ckVtimeQuantum),
             String(this.#ckVtimeWindowMultiplier),
             String(this.#ckVtimeStateTtl),
+            String(this.#ckVtimeIdleMaxEntries),
             // Must stay last: the gauge fragment reads ARGV[#ARGV].
             metricsGaugeArg
           )
@@ -5330,6 +5343,7 @@ local maxCount = tonumber(ARGV[6] or '1')
 local quantum = tonumber(ARGV[7] or '1')
 local windowMultiplier = tonumber(ARGV[8] or '3')
 local stateTtl = tonumber(ARGV[9] or '86400')
+local idleMaxEntries = tonumber(ARGV[10] or '10000')
 ${QUEUE_METRICS_GAUGE_PRELUDE}
 ${QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA}
 
@@ -5593,6 +5607,13 @@ if dequeuedCount > 0 then
   -- NEW: an idle entry at or below the floor confers no credit, because registration takes
   -- max(floor, idleTag). Dropping it bounds the idle set at one op per serving call.
   redis.call('ZREMRANGEBYSCORE', ckVtimeIdleKey, '-inf', tostring(floor))
+  -- NEW: the reap above is worth nothing while the floor is pinned, which a workload that
+  -- keeps minting fresh concurrency keys does indefinitely (each one registers at the floor
+  -- and is served at it, so minServableTag never rises). Measured: the set grew by the drain
+  -- count every round and never shrank. Cap by rank as well, which does not depend on the
+  -- floor moving. Trimming the lowest tags first drops the entries nearest the floor, whose
+  -- remembered credit is worth least.
+  redis.call('ZREMRANGEBYRANK', ckVtimeIdleKey, 0, -(idleMaxEntries + 1))
   if redis.call('EXISTS', ckVtimeKey) == 1 then
     redis.call('EXPIRE', ckVtimeKey, stateTtl)
   end
@@ -7376,6 +7397,7 @@ declare module "@internal/redis" {
       quantum: string,
       windowMultiplier: string,
       stateTtlSeconds: string,
+      idleMaxEntries: string,
       metricsEnabled: string,
       callback?: Callback<[string[] | null, number[] | null]>
     ): Result<[string[] | null, number[] | null], Context>;

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

@@ -2,7 +2,6 @@ import { redisTest } from "@internal/testcontainers";
 import { trace } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import { Decimal } from "@trigger.dev/database";
-import { describe } from "node:test";
 import { FairQueueSelectionStrategy } from "../fairQueueSelectionStrategy.js";
 import { RunQueue } from "../index.js";
 import { RunQueueFullKeyProducer } from "../keyProducer.js";
@@ -51,12 +50,19 @@ const baseEnv = {
 
 const QUEUE = "task/my-task";
 
-function createQueue(redisContainer: any, keyPrefix: string, vtimeEnabled: boolean) {
+function createQueue(
+  redisContainer: any,
+  keyPrefix: string,
+  vtimeEnabled: boolean,
+  idleMaxEntries?: number
+) {
   return new RunQueue({
     ...testOptions,
     masterQueueConsumersDisabled: true,
     workerOptions: { disabled: true },
-    ...(vtimeEnabled ? { ckVirtualTimeScheduling: { enabled: true } } : {}),
+    ...(vtimeEnabled
+      ? { ckVirtualTimeScheduling: { enabled: true, ...(idleMaxEntries ? { idleMaxEntries } : {}) } }
+      : {}),
     queueSelectionStrategy: new FairQueueSelectionStrategy({
       redis: { keyPrefix, host: redisContainer.getHost(), port: redisContainer.getPort() },
       keys: testOptions.keys,
@@ -386,4 +392,58 @@ describe("CK vtime starvation by drain-and-re-register", () => {
       expect(on.idleSize).toBe(0);
     }
   );
+
+  // The idle set is trimmed two ways. The at-or-below-floor reap is the cheap one, but it
+  // is worth nothing while the floor is pinned, and a workload that keeps minting fresh
+  // concurrency keys pins it indefinitely: each new key registers at the floor and is
+  // served at it, so minServableTag never rises. A resource benchmark caught the set
+  // growing by the drain count every round and never shrinking. The rank cap is the bound
+  // that does not depend on the floor moving.
+  redisTest(
+    "the idle set stays capped when fresh keys keep the floor pinned",
+    async ({ redisContainer }) => {
+      const CAP = 25;
+      const DRAINS = 400;
+      const keyPrefix = `runqueue:test:idlecap:`;
+      const queue = createQueue(redisContainer, keyPrefix, true, CAP);
+
+      try {
+        const env = { ...baseEnv, maximumConcurrencyLimit: 20 };
+        await queue.updateEnvConcurrencyLimits(env);
+        const shard = testOptions.keys.masterQueueShardForEnvironment(env.id, 2);
+
+        // Every iteration uses a concurrency key never seen before, drains it, and never
+        // brings it back: the worst case for a set that remembers drained variants.
+        for (let i = 0; i < DRAINS; i++) {
+          await queue.enqueueMessage({
+            env,
+            message: makeMessage({ runId: `f-${i}`, concurrencyKey: `fresh-${i}` }),
+            workerQueue: env.id,
+            skipDequeueProcessing: true,
+          });
+          const msgs = await queue.testDequeueFromMasterQueue(shard, env.id, 1);
+          for (const m of msgs) {
+            await queue.acknowledgeMessage(env.organization.id, m.messageId, {
+              skipDequeueProcessing: true,
+            });
+          }
+        }
+
+        const idleKey = testOptions.keys.ckVtimeIdleKeyFromQueue(variantName("fresh-0"));
+        const idleSize = await queue.redis.zcard(idleKey);
+        const floor = await queue.redis.get(
+          testOptions.keys.ckVtimeFloorKeyFromQueue(variantName("fresh-0"))
+        );
+
+        // The floor really is pinned, so the score reap cannot be what bounded this.
+        expect(Number(floor ?? 0)).toBe(0);
+        // One call can park past the cap before the next trim, hence the small margin.
+        expect(idleSize).toBeLessThanOrEqual(CAP + 10);
+        // And it is the cap doing the work, not an empty set.
+        expect(idleSize).toBeGreaterThan(0);
+      } finally {
+        await queue.quit();
+      }
+    }
+  );
 });