fix(run-engine): remember a drained ck variant's virtual time

A concurrency-key variant whose message zset drained was removed from ckVtime, throwing away the virtual-time tag it had accumulated. Its next enqueue re-registered it at the floor, so it came back with full credit. A variant holding a persistent backlog keeps advancing its tag instead, so it lost every pass-1 slot to variants that drain and reset each call, and pass 2 could not help once the batch was already full. Measured on a backlogged variant against five trickle variants: 1 serve out of 600 with the flag on, against 120 with it off, inverting the fairness the feature exists to provide. A parked tag now survives the drain in a sibling :ckVtimeIdle zset, and every registration path (enqueue, enqueue-with-ttl, nack, and the gated-variant branch) starts the variant at max(floor, idleTag) rather than at the floor. The out-of-band drains (ack, dead-letter, TTL expiry) park the tag too. Entries at or below the floor confer nothing, so a single ZREMRANGEBYSCORE per serving call reaps them and bounds the set. Deriving the floor differently was tried first and rejected by measurement: any variant with a tag that never advances, which includes any concurrency-gated key, pins the minimum and defeats it. Backlogged variant now lands on its round-robin share in every shape, including with a pinned-low gated or future-headed variant present, and with more trickle variants than batch slots. Op-count overhead goes from 587 to 641 against a budget of 900. The flag-off Lua is still byte-identical: 7 vtime command variants changed, the other 31 commands hash the same as HEAD.
2026-08-18 13:01:37 +01:00
parent 47c267dbbd
commit d25ccccc5e
6 changed files with 562 additions and 38 deletions
@@ -1701,20 +1701,28 @@ export class RunQueue {
    const visibilityTimeoutMs = (ttlSystem.visibilityTimeoutMs ?? 30_000).toString();

    // Atomically get and remove expired runs from TTL set, ack them from normal queues, and enqueue to TTL worker
-    const expireTtlRuns = this.#ckVtimeEnabled
-      ? this.redis.expireTtlRunsVtimeTracked.bind(this.redis)
-      : this.redis.expireTtlRunsTracked.bind(this.redis);
-
-    const results = await expireTtlRuns(
-      ttlQueueKey,
-      keyPrefix,
-      now.toString(),
-      batchSize.toString(),
-      shardCount.toString(),
-      workerQueueKey,
-      workerItemsKey,
-      visibilityTimeoutMs
-    );
+    const results = this.#ckVtimeEnabled
+      ? await this.redis.expireTtlRunsVtimeTracked(
+          ttlQueueKey,
+          keyPrefix,
+          now.toString(),
+          batchSize.toString(),
+          shardCount.toString(),
+          workerQueueKey,
+          workerItemsKey,
+          visibilityTimeoutMs,
+          String(this.#ckVtimeStateTtl)
+        )
+      : await this.redis.expireTtlRunsTracked(
+          ttlQueueKey,
+          keyPrefix,
+          now.toString(),
+          batchSize.toString(),
+          shardCount.toString(),
+          workerQueueKey,
+          workerItemsKey,
+          visibilityTimeoutMs
+        );

    if (!results || results.length === 0) {
      return [];
@@ -2240,6 +2248,7 @@ export class RunQueue {
              baseQueueKey,
              this.keys.ckVtimeKeyFromQueue(message.queue),
              this.keys.ckVtimeFloorKeyFromQueue(message.queue),
+              this.keys.ckVtimeIdleKeyFromQueue(message.queue),
              // args
              queueName,
              messageId,
@@ -2315,6 +2324,7 @@ export class RunQueue {
              baseQueueKey,
              this.keys.ckVtimeKeyFromQueue(message.queue),
              this.keys.ckVtimeFloorKeyFromQueue(message.queue),
+              this.keys.ckVtimeIdleKeyFromQueue(message.queue),
              // args
              queueName,
              messageId,
@@ -2634,6 +2644,7 @@ export class RunQueue {
            runningCounterKey,
            this.keys.ckVtimeKeyFromQueue(ckWildcardQueue),
            this.keys.ckVtimeFloorKeyFromQueue(ckWildcardQueue),
+            this.keys.ckVtimeIdleKeyFromQueue(ckWildcardQueue),
            //args
            ckWildcardQueue,
            String(Date.now()),
@@ -2918,11 +2929,13 @@ export class RunQueue {
          lengthCounterKey,
          runningCounterKey,
          this.keys.ckVtimeKeyFromQueue(message.queue),
+          this.keys.ckVtimeIdleKeyFromQueue(message.queue),
          messageId,
          messageQueue,
          messageKeyValue,
          removeFromWorkerQueue ? "1" : "0",
-          ckWildcardName
+          ckWildcardName,
+          String(this.#ckVtimeStateTtl)
        );
      }

@@ -3071,6 +3084,7 @@ export class RunQueue {
          runningCounterKey,
          this.keys.ckVtimeKeyFromQueue(message.queue),
          this.keys.ckVtimeFloorKeyFromQueue(message.queue),
+          this.keys.ckVtimeIdleKeyFromQueue(message.queue),
          //args
          messageId,
          messageQueue,
@@ -3161,9 +3175,11 @@ export class RunQueue {
          lengthCounterKey,
          runningCounterKey,
          this.keys.ckVtimeKeyFromQueue(message.queue),
+          this.keys.ckVtimeIdleKeyFromQueue(message.queue),
          messageId,
          messageQueue,
-          ckWildcardName
+          ckWildcardName,
+          String(this.#ckVtimeStateTtl)
        );
      } else {
        await this.redis.moveToDeadLetterQueueCkTracked(
@@ -4223,7 +4239,7 @@ return __qmret(0)
    // brand-new key is present in the fair order from its first enqueue. The
    // fast path (direct-to-worker-queue) neither registers nor advances.
    this.redis.defineCommand("enqueueMessageCkVtimeTracked", {
-      numberOfKeys: 17,
+      numberOfKeys: 18,
      lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -4242,9 +4258,10 @@ local envConcurrencyLimitBurstFactorKey = KEYS[13]
 -- Counter keys (KEYS 14-15)
 local lengthCounterKey = KEYS[14]
 local baseQueueKey = KEYS[15]
-- Virtual-time keys (KEYS 16-17)
+-- Virtual-time keys (KEYS 16-18)
 local ckVtimeKey = KEYS[16]
 local ckVtimeFloorKey = KEYS[17]
+local ckVtimeIdleKey = KEYS[18]

 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -4329,12 +4346,20 @@ if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], queueName)
 end

-- Register this variant in the virtual-time index at the floor. NX means an
-- already-advanced tag is never rewound.
+-- Register this variant in the virtual-time index. NX means an already-advanced tag is
+-- never rewound. The start is max(floor, remembered idle tag): a variant that drained and
+-- came back would otherwise be handed full credit at the floor on every re-enqueue, which
+-- starves any variant carrying a persistent backlog.
 local vfloor = redis.call('GET', ckVtimeFloorKey) or '0'
-redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName)
+local vstart = tonumber(vfloor)
+local vidle = redis.call('ZSCORE', ckVtimeIdleKey, queueName)
+if vidle and tonumber(vidle) > vstart then
+  vstart = tonumber(vidle)
+end
+redis.call('ZADD', ckVtimeKey, 'NX', tostring(vstart), queueName)
 redis.call('EXPIRE', ckVtimeKey, stateTtl)
 redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
+redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)

 -- Rebalance master queue with ck:* member
 local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
@@ -4365,7 +4390,7 @@ return __qmret(0)
    // Vtime variant of enqueueMessageWithTtlCkTracked. Same slow-path-only
    // registration as enqueueMessageCkVtimeTracked above.
    this.redis.defineCommand("enqueueMessageWithTtlCkVtimeTracked", {
-      numberOfKeys: 18,
+      numberOfKeys: 19,
      lua: `
 local masterQueueKey = KEYS[1]
 local queueKey = KEYS[2]
@@ -4385,9 +4410,10 @@ local envConcurrencyLimitBurstFactorKey = KEYS[14]
 -- Counter keys (KEYS 15-16)
 local lengthCounterKey = KEYS[15]
 local baseQueueKey = KEYS[16]
-- Virtual-time keys (KEYS 17-18)
+-- Virtual-time keys (KEYS 17-19)
 local ckVtimeKey = KEYS[17]
 local ckVtimeFloorKey = KEYS[18]
+local ckVtimeIdleKey = KEYS[19]

 local queueName = ARGV[1]
 local messageId = ARGV[2]
@@ -4467,12 +4493,20 @@ if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], queueName)
 end

-- Register this variant in the virtual-time index at the floor. NX means an
-- already-advanced tag is never rewound.
+-- Register this variant in the virtual-time index. NX means an already-advanced tag is
+-- never rewound. The start is max(floor, remembered idle tag): a variant that drained and
+-- came back would otherwise be handed full credit at the floor on every re-enqueue, which
+-- starves any variant carrying a persistent backlog.
 local vfloor = redis.call('GET', ckVtimeFloorKey) or '0'
-redis.call('ZADD', ckVtimeKey, 'NX', vfloor, queueName)
+local vstart = tonumber(vfloor)
+local vidle = redis.call('ZSCORE', ckVtimeIdleKey, queueName)
+if vidle and tonumber(vidle) > vstart then
+  vstart = tonumber(vidle)
+end
+redis.call('ZADD', ckVtimeKey, 'NX', tostring(vstart), queueName)
 redis.call('EXPIRE', ckVtimeKey, stateTtl)
 redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
+redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)

 -- Rebalance master queue with ck:* member
 local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
@@ -4732,6 +4766,7 @@ local shardCount = tonumber(ARGV[4])
 local workerQueueKey = ARGV[5]
 local workerItemsKey = ARGV[6]
 local visibilityTimeoutMs = tonumber(ARGV[7])
+local stateTtl = tonumber(ARGV[8] or '86400')

 local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
@@ -4809,7 +4844,16 @@ for i, member in ipairs(expiredMembers) do
          redis.call('ZREM', ckIndexKey, rawQueueKey)
          -- NEW: derived rather than passed in, because this sweep discovers the queues it
          -- touches inside the script, exactly as ckIndexKey above is derived.
-          redis.call('ZREM', keyPrefix .. ckMatch .. ":ckVtime", rawQueueKey)
+          -- NEW: park the tag first so the variant's next enqueue re-registers with the
+          -- credit it earned rather than at the floor.
+          local ckVtimeKey = keyPrefix .. ckMatch .. ":ckVtime"
+          local ckVtimeIdleKey = keyPrefix .. ckMatch .. ":ckVtimeIdle"
+          local idleTag = redis.call('ZSCORE', ckVtimeKey, rawQueueKey)
+          if idleTag then
+            redis.call('ZADD', ckVtimeIdleKey, idleTag, rawQueueKey)
+            redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)
+          end
+          redis.call('ZREM', ckVtimeKey, rawQueueKey)
        else
          redis.call('ZADD', ckIndexKey, earliest[2], rawQueueKey)
        end
@@ -5282,7 +5326,7 @@ return __qmret(results)
    // pass 1 could not see rather than serving them, which is what keeps a backlog
    // queued before the flag went on from being unreachable.
    this.redis.defineCommand("dequeueMessagesFromCkQueueVtimeTracked", {
-      numberOfKeys: 13,
+      numberOfKeys: 14,
      lua: `
 local ckIndexKey = KEYS[1]
 local queueConcurrencyLimitKey = KEYS[2]
@@ -5297,6 +5341,7 @@ local lengthCounterKey = KEYS[10]
 local runningCounterKey = KEYS[11]
 local ckVtimeKey = KEYS[12]
 local ckVtimeFloorKey = KEYS[13]
+local ckVtimeIdleKey = KEYS[14]

 local ckWildcardName = ARGV[1]
 local currentTime = tonumber(ARGV[2])
@@ -5364,6 +5409,9 @@ local attempted = {}
 local function tryServe(ckQueueName, mayRaiseFloor)
  attempted[ckQueueName] = true
  local fullQueueKey = keyPrefix .. ckQueueName
+  -- NEW: the tag this call wrote back, if it served. Site A below reuses it rather than
+  -- re-reading the score it just wrote.
+  local servedTag = nil

  local ckConcurrencyKey = fullQueueKey .. ':currentConcurrency'
  local ckCurrentConcurrency = tonumber(redis.call('SCARD', ckConcurrencyKey) or '0')
@@ -5418,7 +5466,8 @@ local function tryServe(ckQueueName, mayRaiseFloor)
          if mayRaiseFloor and (minServableTag == nil or tag < minServableTag) then
            minServableTag = tag
          end
-          redis.call('ZADD', ckVtimeKey, tostring(tag + (quantum / weight)), ckQueueName)
+          servedTag = tag + (quantum / weight)
+          redis.call('ZADD', ckVtimeKey, tostring(servedTag), ckQueueName)
        end
      else
        redis.call('ZREM', fullQueueKey, messageId)
@@ -5429,6 +5478,24 @@ local function tryServe(ckQueueName, mayRaiseFloor)
      local earliest = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
      if #earliest == 0 then
        redis.call('ZREM', ckIndexKey, ckQueueName)
+        -- NEW: park the tag in the idle set so the next enqueue re-registers with the
+        -- credit this variant earned instead of full credit at the floor. Only above the
+        -- floor is worth keeping: registration takes max(floor, idleTag), so an entry at
+        -- or below the floor confers nothing and just grows the set.
+        -- servedTag is nil when the branch above dropped an expired or payload-less message
+        -- rather than serving one; the tag is still on record, so read it back before the
+        -- ZREM discards it. Only that rare path pays the extra read.
+        local parkTag = servedTag
+        if parkTag == nil then
+          local storedTag = redis.call('ZSCORE', ckVtimeKey, ckQueueName)
+          if storedTag then
+            parkTag = tonumber(storedTag)
+          end
+        end
+        if parkTag ~= nil and parkTag > floor then
+          redis.call('ZADD', ckVtimeIdleKey, tostring(parkTag), ckQueueName)
+          redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)
+        end
        redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW
      else
        redis.call('ZADD', ckIndexKey, earliest[2], ckQueueName)
@@ -5437,6 +5504,13 @@ local function tryServe(ckQueueName, mayRaiseFloor)
      local any = redis.call('ZRANGE', fullQueueKey, 0, 0, 'WITHSCORES')
      if #any == 0 then
        redis.call('ZREM', ckIndexKey, ckQueueName)
+        -- NEW: nothing was served, so no tag is in hand. Read it before the ZREM discards
+        -- it, and keep it only if it is above the floor (see Site A).
+        local idleTag = redis.call('ZSCORE', ckVtimeKey, ckQueueName)
+        if idleTag and tonumber(idleTag) > floor then
+          redis.call('ZADD', ckVtimeIdleKey, idleTag, ckQueueName)
+          redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)
+        end
        redis.call('ZREM', ckVtimeKey, ckQueueName) -- NEW
      else
        redis.call('ZADD', ckIndexKey, any[2], ckQueueName)
@@ -5458,7 +5532,14 @@ local function tryServe(ckQueueName, mayRaiseFloor)
    -- 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
+    -- NEW: registers at max(floor, remembered idle tag), same rule as the enqueue path, so
+    -- a variant that drained under the gate does not come back with full credit.
+    local gateStart = floor
+    local gateIdle = redis.call('ZSCORE', ckVtimeIdleKey, ckQueueName)
+    if gateIdle and tonumber(gateIdle) > gateStart then
+      gateStart = tonumber(gateIdle)
+    end
+    if redis.call('ZADD', ckVtimeKey, 'NX', tostring(gateStart), ckQueueName) == 1 then
      redis.call('EXPIRE', ckVtimeKey, stateTtl)
    end
  end
@@ -5531,6 +5612,9 @@ if dequeuedCount > 0 then
    floor = minServableTag
  end
  redis.call('SET', ckVtimeFloorKey, tostring(floor), 'EX', stateTtl)
+  -- 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))
  if redis.call('EXISTS', ckVtimeKey) == 1 then
    redis.call('EXPIRE', ckVtimeKey, stateTtl)
  end
@@ -6074,7 +6158,7 @@ end
    });

    this.redis.defineCommand("acknowledgeMessageCkVtimeTracked", {
-      numberOfKeys: 13,
+      numberOfKeys: 14,
      lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -6090,6 +6174,7 @@ local ckIndexKey = KEYS[10]
 local lengthCounterKey = KEYS[11]
 local runningCounterKey = KEYS[12]
 local ckVtimeKey = KEYS[13]
+local ckVtimeIdleKey = KEYS[14]

 -- Args:
 local messageId = ARGV[1]
@@ -6097,6 +6182,7 @@ local messageQueueName = ARGV[2]
 local messageKeyValue = ARGV[3]
 local removeFromWorkerQueue = ARGV[4]
 local ckWildcardName = ARGV[5]
+local stateTtl = tonumber(ARGV[6] or '86400')

 local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
@@ -6123,6 +6209,14 @@ if #earliestInCkQueue == 0 then
  -- NEW: the variant has drained, so it leaves the fair order too. Previously only the
  -- vtime dequeue removed a ckVtime entry, so an ack left one behind whose tag had
  -- stopped advancing until some later scan happened to visit and collect it.
+  -- NEW: park the tag first, so the variant's next enqueue re-registers with the credit it
+  -- earned rather than at the floor. No floor key here, so this saves unconditionally; the
+  -- dequeue command reaps everything at or below the floor on its next serving call.
+  local idleTag = redis.call('ZSCORE', ckVtimeKey, messageQueueName)
+  if idleTag then
+    redis.call('ZADD', ckVtimeIdleKey, idleTag, messageQueueName)
+    redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)
+  end
  redis.call('ZREM', ckVtimeKey, messageQueueName)
 else
  redis.call('ZADD', ckIndexKey, earliestInCkQueue[2], messageQueueName)
@@ -6262,7 +6356,7 @@ end
    // of the variant into the :ckVtime ZSET at the floor (NX), so a GC'd variant
    // that a nack revives rejoins the fair order.
    this.redis.defineCommand("nackMessageCkVtimeTracked", {
-      numberOfKeys: 13,
+      numberOfKeys: 14,
      lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -6279,6 +6373,7 @@ local runningCounterKey = KEYS[11]
 -- Virtual-time keys (KEYS 12-13)
 local ckVtimeKey = KEYS[12]
 local ckVtimeFloorKey = KEYS[13]
+local ckVtimeIdleKey = KEYS[14]

 -- Args:
 local messageId = ARGV[1]
@@ -6341,12 +6436,20 @@ if #earliest > 0 then
  redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName)
 end

-- Register this variant in the virtual-time index at the floor. NX means an
-- already-advanced tag is never rewound.
+-- Register this variant in the virtual-time index. NX means an already-advanced tag is
+-- never rewound. The start is max(floor, remembered idle tag): a nack after the variant
+-- drained would otherwise hand back full credit at the floor, which is the same starvation
+-- the enqueue path guards against.
 local vfloor = redis.call('GET', ckVtimeFloorKey) or '0'
-redis.call('ZADD', ckVtimeKey, 'NX', vfloor, messageQueueName)
+local vstart = tonumber(vfloor)
+local vidle = redis.call('ZSCORE', ckVtimeIdleKey, messageQueueName)
+if vidle and tonumber(vidle) > vstart then
+  vstart = tonumber(vidle)
+end
+redis.call('ZADD', ckVtimeKey, 'NX', tostring(vstart), messageQueueName)
 redis.call('EXPIRE', ckVtimeKey, stateTtl)
 redis.call('EXPIRE', ckVtimeFloorKey, stateTtl)
+redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)

 -- Rebalance master queue with ck:* member
 local earliestIdx = redis.call('ZRANGE', ckIndexKey, 0, 0, 'WITHSCORES')
@@ -6445,7 +6548,7 @@ end
    });

    this.redis.defineCommand("moveToDeadLetterQueueCkVtimeTracked", {
-      numberOfKeys: 13,
+      numberOfKeys: 14,
      lua: `
 -- Keys:
 local masterQueueKey = KEYS[1]
@@ -6461,11 +6564,13 @@ local ckIndexKey = KEYS[10]
 local lengthCounterKey = KEYS[11]
 local runningCounterKey = KEYS[12]
 local ckVtimeKey = KEYS[13]
+local ckVtimeIdleKey = KEYS[14]

 -- Args:
 local messageId = ARGV[1]
 local messageQueueName = ARGV[2]
 local ckWildcardName = ARGV[3]
+local stateTtl = tonumber(ARGV[4] or '86400')

 local function decrFloored(key)
  if tonumber(redis.call('GET', key) or '0') > 0 then
@@ -6489,6 +6594,12 @@ if #earliest == 0 then
  -- NEW: the variant has drained, so it leaves the fair order too. Previously only the
  -- vtime dequeue removed a ckVtime entry, so an ack left one behind whose tag had
  -- stopped advancing until some later scan happened to visit and collect it.
+  -- NEW: park the tag first, same rule as the ack path.
+  local idleTag = redis.call('ZSCORE', ckVtimeKey, messageQueueName)
+  if idleTag then
+    redis.call('ZADD', ckVtimeIdleKey, idleTag, messageQueueName)
+    redis.call('EXPIRE', ckVtimeIdleKey, stateTtl)
+  end
  redis.call('ZREM', ckVtimeKey, messageQueueName)
 else
  redis.call('ZADD', ckIndexKey, earliest[2], messageQueueName)
@@ -7184,6 +7295,7 @@ declare module "@internal/redis" {
      baseQueueKey: string,
      ckVtimeKey: string,
      ckVtimeFloorKey: string,
+      ckVtimeIdleKey: string,
      queueName: string,
      messageId: string,
      messageData: string,
@@ -7220,6 +7332,7 @@ declare module "@internal/redis" {
      baseQueueKey: string,
      ckVtimeKey: string,
      ckVtimeFloorKey: string,
+      ckVtimeIdleKey: string,
      queueName: string,
      messageId: string,
      messageData: string,
@@ -7275,6 +7388,7 @@ declare module "@internal/redis" {
      runningCounterKey: string,
      ckVtimeKey: string,
      ckVtimeFloorKey: string,
+      ckVtimeIdleKey: string,
      ckWildcardName: string,
      currentTime: string,
      defaultEnvConcurrencyLimit: string,
@@ -7330,11 +7444,13 @@ declare module "@internal/redis" {
      lengthCounterKey: string,
      runningCounterKey: string,
      ckVtimeKey: string,
+      ckVtimeIdleKey: string,
      messageId: string,
      messageQueueName: string,
      messageKeyValue: string,
      removeFromWorkerQueue: string,
      ckWildcardName: string,
+      stateTtl: string,
      callback?: Callback<void>
    ): Result<void, Context>;

@@ -7374,6 +7490,7 @@ declare module "@internal/redis" {
      runningCounterKey: string,
      ckVtimeKey: string,
      ckVtimeFloorKey: string,
+      ckVtimeIdleKey: string,
      messageId: string,
      messageQueueName: string,
      messageData: string,
@@ -7418,9 +7535,11 @@ declare module "@internal/redis" {
      lengthCounterKey: string,
      runningCounterKey: string,
      ckVtimeKey: string,
+      ckVtimeIdleKey: string,
      messageId: string,
      messageQueueName: string,
      ckWildcardName: string,
+      stateTtl: string,
      callback?: Callback<void>
    ): Result<void, Context>;

@@ -7445,6 +7564,7 @@ declare module "@internal/redis" {
      workerQueueKey: string,
      workerItemsKey: string,
      visibilityTimeoutMs: string,
+      stateTtl: string,
      callback?: Callback<string[]>
    ): Result<string[], Context>;

@@ -24,6 +24,7 @@ const constants = {
  CK_INDEX_PART: "ckIndex",
  CK_VTIME_PART: "ckVtime",
  CK_VTIME_FLOOR_PART: "ckVtimeFloor",
+  CK_VTIME_IDLE_PART: "ckVtimeIdle",
  LENGTH_COUNTER_PART: "lengthCounter",
  RUNNING_COUNTER_PART: "runningCounter",
 } as const;
@@ -325,6 +326,10 @@ export class RunQueueFullKeyProducer implements RunQueueKeyProducer {
    return `${this.baseQueueKeyFromQueue(queue)}:${constants.CK_VTIME_FLOOR_PART}`;
  }

+  ckVtimeIdleKeyFromQueue(queue: string): string {
+    return `${this.baseQueueKeyFromQueue(queue)}:${constants.CK_VTIME_IDLE_PART}`;
+  }
+
  // indexOf instead of /:ck:.+$/ (queue names are user-controlled; polynomial regex).
  // Only strips when at least one character follows ":ck:", matching the old semantics.
  baseQueueKeyFromQueue(queue: string): string {
@@ -1575,9 +1575,11 @@ describe("CK virtual-time (SFQ) dequeue", () => {
          testOptions.keys.queueLengthCounterKeyFromQueue(v),
          testOptions.keys.queueRunningCounterKeyFromQueue(v),
          ckVtimeKey,
+          testOptions.keys.ckVtimeIdleKeyFromQueue(v),
          "r-dlq",
          v,
-          testOptions.keys.toCkWildcard(v)
+          testOptions.keys.toCkWildcard(v),
+          "86400"
        );

        expect(await queue.redis.zscore(ckIndexKey, v)).toBeNull();
@@ -1629,7 +1631,8 @@ describe("CK virtual-time (SFQ) dequeue", () => {
        "2",
        "ttlworker",
        "ttlworkeritems",
-        "30000"
+        "30000",
+        "86400"
      );

      expect(await queue.redis.zscore(ckIndexKey, v)).toBeNull();
@@ -0,0 +1,389 @@
+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";
+import type { InputPayload } from "../types.js";
+
+// Starvation of a persistently-backlogged CK variant by variants that drain on every call.
+//
+// A variant that empties is GC'd out of ckVtime, and before the idle zset its next enqueue
+// re-registered it at the floor: full credit, every call. A variant carrying a backlog
+// keeps advancing its tag and never gets it back, so it loses every comparison. Measured
+// on the plain shape below: 1 serve out of 600 with the flag on, against 120 with it off.
+//
+// ckVtimeIdle remembers the tag across a drain, so re-registration takes
+// max(floor, idleTag) and a drain buys nothing. Each shape here asserts the backlogged
+// variant lands near its round-robin share of 1/(1 + competitors).
+//
+// Shapes covered: the plain trickle shape, more trickles than batch slots, a
+// concurrency-gated variant pinned at a low tag, a future-headed variant pinned at a low
+// tag, and a deep-queue control whose competitors never drain (so they were never affected
+// by the bug, and it shows what the fair share is).
+
+const testOptions = {
+  name: "rq",
+  tracer: trace.getTracer("rq"),
+  workers: 1,
+  defaultEnvConcurrency: 25,
+  logger: new Logger("RunQueue", "warn"),
+  retryOptions: {
+    maxAttempts: 5,
+    factor: 1.1,
+    minTimeoutInMs: 100,
+    maxTimeoutInMs: 1_000,
+    randomize: true,
+  },
+  keys: new RunQueueFullKeyProducer(),
+};
+
+const baseEnv = {
+  id: "e1234",
+  type: "DEVELOPMENT" as const,
+  maximumConcurrencyLimit: 20,
+  concurrencyLimitBurstFactor: new Decimal(1),
+  project: { id: "p1234" },
+  organization: { id: "o1234" },
+};
+
+const QUEUE = "task/my-task";
+
+function createQueue(redisContainer: any, keyPrefix: string, vtimeEnabled: boolean) {
+  return new RunQueue({
+    ...testOptions,
+    masterQueueConsumersDisabled: true,
+    workerOptions: { disabled: true },
+    ...(vtimeEnabled ? { ckVirtualTimeScheduling: { enabled: true } } : {}),
+    queueSelectionStrategy: new FairQueueSelectionStrategy({
+      redis: { keyPrefix, host: redisContainer.getHost(), port: redisContainer.getPort() },
+      keys: testOptions.keys,
+    }),
+    redis: { keyPrefix, host: redisContainer.getHost(), port: redisContainer.getPort() },
+  });
+}
+
+function makeMessage(overrides: Partial<InputPayload> = {}): InputPayload {
+  return {
+    runId: "r1",
+    taskIdentifier: "task/my-task",
+    orgId: "o1234",
+    projectId: "p1234",
+    environmentId: "e1234",
+    environmentType: "DEVELOPMENT",
+    queue: QUEUE,
+    timestamp: Date.now(),
+    attempt: 0,
+    ...overrides,
+  };
+}
+
+function variantName(ck: string): string {
+  return testOptions.keys.queueKey(baseEnv, QUEUE, ck);
+}
+
+type RunOpts = {
+  trickleCount: number;
+  maxCount: number;
+  calls: number;
+  backlogSize: number;
+  envLimit: number;
+  // Control: give each competitor a deep backlog instead of one message, so it never
+  // drains and is never GC'd out of ckVtime / re-registered.
+  steadyCompetitors?: boolean;
+  // A variant parked at the per-key concurrency ceiling for the whole run. It stays in
+  // ckVtime holding a permanently low tag, which is what defeated the earlier floor-only
+  // prototype of this fix.
+  gatedVariant?: boolean;
+  // A variant whose whole backlog is scheduled an hour out. Also stays registered at a
+  // permanently low tag, by the 'notReady' route rather than the concurrency gate.
+  futureVariant?: boolean;
+};
+
+type RunResult = {
+  backlogServed: number;
+  trickleServed: number;
+  totalServed: number;
+  lastBacklogServeCall: number;
+  finalFloor: string | null;
+  finalTags: Record<string, number>;
+  idleSize: number;
+};
+
+async function runShape(
+  redisContainer: any,
+  label: string,
+  vtimeEnabled: boolean,
+  opts: RunOpts
+): Promise<RunResult> {
+  const keyPrefix = `runqueue:test:${label}:`;
+  const queue = createQueue(redisContainer, keyPrefix, vtimeEnabled);
+
+  try {
+    const env = { ...baseEnv, maximumConcurrencyLimit: opts.envLimit };
+    await queue.updateEnvConcurrencyLimits(env);
+
+    const t0 = Date.now() - 10_000_000;
+
+    const enqueue = async (runId: string, ck: string, timestamp: number) => {
+      await queue.enqueueMessage({
+        env,
+        message: makeMessage({ runId, concurrencyKey: ck, timestamp }),
+        workerQueue: env.id,
+        skipDequeueProcessing: true,
+      });
+    };
+
+    // Backlogged variant FIRST so its heads are the oldest in ckIndex.
+    for (let i = 0; i < opts.backlogSize; i++) {
+      await enqueue(`b${i}`, "backlog", t0 + i);
+    }
+
+    if (opts.gatedVariant) {
+      // Ready, old work so it is a live ckIndex/ckVtime member every call...
+      for (let i = 0; i < 50; i++) {
+        await enqueue(`g${i}`, "gated", t0 + 100 + i);
+      }
+      // ...but parked at the per-key ceiling, so tryServe never serves it. The ceiling is
+      // min(queue limit, env limit) and the queue limit is unset here, so it is envLimit.
+      const members = Array.from({ length: opts.envLimit + 10 }, (_, i) => `held-${i}`);
+      await queue.redis.sadd(`${variantName("gated")}:currentConcurrency`, ...members);
+    }
+
+    if (opts.futureVariant) {
+      const future = Date.now() + 60 * 60 * 1000;
+      for (let i = 0; i < 50; i++) {
+        await enqueue(`f${i}`, "future", future + i);
+      }
+    }
+
+    // One ready message per trickle variant, newer than the whole backlog, so age order
+    // alone would never prefer them. Under steadyCompetitors each gets a deep queue.
+    const seedPerCompetitor = opts.steadyCompetitors ? opts.calls + 10 : 1;
+    for (let t = 0; t < opts.trickleCount; t++) {
+      for (let i = 0; i < seedPerCompetitor; i++) {
+        await enqueue(`t${t}-seed-${i}`, `trickle-${t}`, t0 + 5_000_000 + i);
+      }
+    }
+
+    const shard = testOptions.keys.masterQueueShardForEnvironment(env.id, 2);
+
+    let backlogServed = 0;
+    let trickleServed = 0;
+    let totalServed = 0;
+    let lastBacklogServeCall = -1;
+    let refeed = 0;
+
+    for (let call = 0; call < opts.calls; call++) {
+      const messages = await queue.testDequeueFromMasterQueue(shard, env.id, opts.maxCount);
+
+      const drainedTrickles = new Set<string>();
+      for (const m of messages) {
+        const ck = m.message.concurrencyKey ?? "";
+        totalServed++;
+        if (ck === "backlog") {
+          backlogServed++;
+          lastBacklogServeCall = call;
+        } else if (ck.startsWith("trickle-")) {
+          trickleServed++;
+          drainedTrickles.add(ck);
+        }
+        // Ack immediately: concurrency is never the limiting factor here.
+        await queue.acknowledgeMessage(env.organization.id, m.messageId, {
+          skipDequeueProcessing: true,
+        });
+      }
+
+      if (opts.steadyCompetitors) continue;
+
+      // Re-feed every trickle variant that drained, so it is ready again next call with a
+      // strictly newer head than the backlog. This is the re-registration under test.
+      for (const ck of drainedTrickles) {
+        refeed++;
+        await enqueue(`t-${ck}-${refeed}`, ck, t0 + 5_000_000 + refeed);
+      }
+    }
+
+    const backlogVariant = variantName("backlog");
+    const finalFloor = await queue.redis.get(
+      testOptions.keys.ckVtimeFloorKeyFromQueue(backlogVariant)
+    );
+    const raw = await queue.redis.zrange(
+      testOptions.keys.ckVtimeKeyFromQueue(backlogVariant),
+      0,
+      -1,
+      "WITHSCORES"
+    );
+    const finalTags: Record<string, number> = {};
+    for (let i = 0; i < raw.length; i += 2) {
+      const member = raw[i]!;
+      const short = member.includes(":ck:") ? member.slice(member.indexOf(":ck:") + 4) : member;
+      finalTags[short] = Number(raw[i + 1]);
+    }
+    const idleSize = await queue.redis.zcard(
+      testOptions.keys.ckVtimeIdleKeyFromQueue(backlogVariant)
+    );
+
+    return {
+      backlogServed,
+      trickleServed,
+      totalServed,
+      lastBacklogServeCall,
+      finalFloor,
+      finalTags,
+      idleSize,
+    };
+  } finally {
+    await queue.quit();
+  }
+}
+
+// Round-robin share of the served batch for one variant among 1 + trickleCount claimants,
+// which is what the deep-queue control measures out at.
+function fairShare(opts: RunOpts, served: number): number {
+  return served / (1 + opts.trickleCount);
+}
+
+// 0.7 rather than 1.0: pass 1 walks in tag order and pass 2 fills by age, so a variant can
+// lose a slot to rounding at the batch boundary. The gap being defended against is two
+// orders of magnitude (1 vs 100), so this has plenty of room and is not tuned to a number.
+function expectFairish(label: string, opts: RunOpts, r: RunResult) {
+  const target = fairShare(opts, r.totalServed);
+  expect(
+    r.backlogServed,
+    `${label}: backlog served ${r.backlogServed} of ${r.totalServed}, fair share ${target.toFixed(
+      1
+    )}`
+  ).toBeGreaterThanOrEqual(target * 0.7);
+}
+
+vi.setConfig({ testTimeout: 300_000 });
+
+describe("CK vtime starvation by drain-and-re-register", () => {
+  redisTest(
+    "backlogged variant keeps its share against trickle variants (flag on vs off)",
+    async ({ redisContainer }) => {
+      const opts: RunOpts = {
+        trickleCount: 5,
+        maxCount: 5,
+        calls: 120,
+        backlogSize: 400,
+        envLimit: 20,
+      };
+
+      const on = await runShape(redisContainer, "starve-on", true, opts);
+      const off = await runShape(redisContainer, "starve-off", false, opts);
+
+      expect(on.totalServed).toBe(600);
+      expect(off.totalServed).toBe(600);
+
+      // The comparison arm: flag off is pure age order, so the always-oldest backlog wins
+      // a slot on every call. That is the number the flag-on path regressed against.
+      expect(off.backlogServed).toBeGreaterThanOrEqual(100);
+
+      expectFairish("flag on", opts, on);
+
+      // It was served throughout, not just drained early and then starved.
+      expect(on.lastBacklogServeCall).toBeGreaterThanOrEqual(opts.calls - 10);
+
+      // The idle zset is reaped at or below the floor on every serving call, so it holds
+      // at most the variants that drained since the last one.
+      expect(on.idleSize).toBeLessThanOrEqual(opts.trickleCount + 2);
+    }
+  );
+
+  redisTest(
+    "holds when there are more trickle variants than batch slots",
+    async ({ redisContainer }) => {
+      for (const trickleCount of [6, 8]) {
+        const opts: RunOpts = {
+          trickleCount,
+          maxCount: 5,
+          calls: 60,
+          backlogSize: 400,
+          envLimit: 20,
+        };
+        const on = await runShape(redisContainer, `over-on-${trickleCount}`, true, opts);
+        expect(on.totalServed).toBe(300);
+        expectFairish(`trickle=${trickleCount}`, opts, on);
+      }
+    }
+  );
+
+  redisTest(
+    "a concurrency-gated variant pinned at a low tag does not defeat it",
+    async ({ redisContainer }) => {
+      const opts: RunOpts = {
+        trickleCount: 5,
+        maxCount: 5,
+        calls: 120,
+        backlogSize: 400,
+        envLimit: 20,
+        gatedVariant: true,
+      };
+      const on = await runShape(redisContainer, "pin-gated-on", true, opts);
+
+      // The gated variant is still registered and still holding a tag below the floor, which
+      // is the state that defeated the earlier floor-only prototype.
+      expect(on.finalTags["gated"]).toBeLessThan(Number(on.finalFloor));
+      expectFairish("gated", opts, on);
+    }
+  );
+
+  redisTest(
+    "a future-headed variant pinned at a low tag does not defeat it",
+    async ({ redisContainer }) => {
+      const opts: RunOpts = {
+        trickleCount: 5,
+        maxCount: 5,
+        calls: 120,
+        backlogSize: 400,
+        envLimit: 20,
+        futureVariant: true,
+      };
+      const on = await runShape(redisContainer, "pin-future-on", true, opts);
+
+      expect(on.finalTags["future"]).toBeLessThan(Number(on.finalFloor));
+      expectFairish("future", opts, on);
+    }
+  );
+
+  redisTest("both pinned-low variants at once", async ({ redisContainer }) => {
+    const opts: RunOpts = {
+      trickleCount: 5,
+      maxCount: 5,
+      calls: 120,
+      backlogSize: 400,
+      envLimit: 20,
+      gatedVariant: true,
+      futureVariant: true,
+    };
+    const on = await runShape(redisContainer, "pin-both-on", true, opts);
+    expectFairish("gated+future", opts, on);
+  });
+
+  redisTest(
+    "control: competitors with deep queues never drain, so they never re-registered",
+    async ({ redisContainer }) => {
+      const opts: RunOpts = {
+        trickleCount: 5,
+        maxCount: 5,
+        calls: 120,
+        backlogSize: 400,
+        envLimit: 20,
+        steadyCompetitors: true,
+      };
+      const on = await runShape(redisContainer, "steady-on", true, opts);
+
+      // This shape never triggered the bug, so it measures what fair looks like: the
+      // trickle shapes above are held to the same standard.
+      expect(on.totalServed).toBe(600);
+      expectFairish("steady", opts, on);
+
+      // Nothing drained, so nothing was ever parked.
+      expect(on.idleSize).toBe(0);
+    }
+  );
+});
@@ -442,9 +442,15 @@ describe("KeyProducer", () => {
    expect(keyProducer.ckVtimeFloorKeyFromQueue(q)).toBe(
      "{org:o1}:proj:p1:env:e1:queue:task/my-task:ckVtimeFloor"
    );
+    expect(keyProducer.ckVtimeIdleKeyFromQueue(q)).toBe(
+      "{org:o1}:proj:p1:env:e1:queue:task/my-task:ckVtimeIdle"
+    );
    // ck wildcard and base-queue inputs normalise the same way
    expect(keyProducer.ckVtimeKeyFromQueue(q.replace(":ck:tenant-a", ":ck:*"))).toBe(
      keyProducer.ckVtimeKeyFromQueue(q)
    );
+    expect(keyProducer.ckVtimeIdleKeyFromQueue(q.replace(":ck:tenant-a", ":ck:*"))).toBe(
+      keyProducer.ckVtimeIdleKeyFromQueue(q)
+    );
  });
 });
@@ -134,6 +134,7 @@ export interface RunQueueKeyProducer {
  ckIndexKeyFromQueue(queue: string): string;
  ckVtimeKeyFromQueue(queue: string): string;
  ckVtimeFloorKeyFromQueue(queue: string): string;
+  ckVtimeIdleKeyFromQueue(queue: string): string;
  baseQueueKeyFromQueue(queue: string): string;
  isCkWildcard(queue: string): boolean;
  toCkWildcard(queue: string): string;