fix(run-engine): stop a '*' concurrency key stranding its whole base queue

A concurrency key is an unrestricted client string, so '*' reaches the queue
unescaped, and queueKey renders it as a variant name byte-identical to the
wildcard member the CK scripts keep in the master queue for that base queue.

Each CK script rebalances the master queue with that wildcard member and then
removes the 'old-format' entry for the variant it just touched. Where the variant
IS the wildcard those two calls name the same member, so the cleanup undid the
rebalance and took the base queue's only master-queue entry with it. Nothing then
pointed at the queue, so every concurrency key on it silently stopped being
dequeued until some later write happened to re-add the member.

Guards the cleanup in all 10 CK scripts (4 enqueue, 6 ack/nack/dead-letter). No
key-format change, so state already in Redis is repaired by the next write rather
than needing a migration.

Tests cover the enqueue, ack and nack paths, and fail without the guard.

Author: 1stvamp

.server-changes/ck-wildcard-queue-strand.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Using `*` as a concurrency key no longer stops a queue from being processed. Triggering a single run with that key could leave the whole queue stalled, including runs using other concurrency keys on it, until something else was triggered on the same queue.

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

@@ -3603,8 +3603,13 @@ if #earliestIdx > 0 then
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, queueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if queueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, queueName)
+end
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
@@ -3708,8 +3713,13 @@ if #earliestIdx > 0 then
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, queueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if queueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, queueName)
+end
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
@@ -3838,8 +3848,13 @@ if #earliestIdx > 0 then
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, queueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if queueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, queueName)
+end
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
@@ -3956,8 +3971,13 @@ if #earliestIdx > 0 then
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, queueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if queueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, queueName)
+end
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
@@ -4908,8 +4928,13 @@ else
   redis.call('ZADD', masterQueueKey, earliestInCkIndex[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, messageQueueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if messageQueueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, messageQueueName)
+end
 
 -- Update the concurrency keys
 redis.call('SREM', queueCurrentConcurrencyKey, messageId)
@@ -4973,8 +4998,13 @@ else
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, messageQueueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if messageQueueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, messageQueueName)
+end
 `,
     });
 
@@ -5019,8 +5049,13 @@ else
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, messageQueueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if messageQueueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, messageQueueName)
+end
 
 -- Add the message to the dead letter queue
 redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId)
@@ -5095,8 +5130,13 @@ else
   redis.call('ZADD', masterQueueKey, earliestInCkIndex[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, messageQueueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if messageQueueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, messageQueueName)
+end
 
 -- Update the concurrency keys. DECR runningCounter only when SREM
 -- currentDequeued actually removed an entry (the message was in flight).
@@ -5201,8 +5241,13 @@ else
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, messageQueueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if messageQueueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, messageQueueName)
+end
 `,
     });
 
@@ -5261,8 +5306,13 @@ else
   redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)
 end
 
--- Remove old-format entry from master queue (transition cleanup)
-redis.call('ZREM', masterQueueKey, messageQueueName)
+-- Remove old-format entry from master queue (transition cleanup). Skipped when the
+-- variant name IS the wildcard: a concurrency key of '*' produces a queue key identical
+-- to the wildcard member, so an unguarded ZREM here deletes the entry the rebalance just
+-- wrote and strands every concurrency key on this base queue.
+if messageQueueName ~= ckWildcardName then
+  redis.call('ZREM', masterQueueKey, messageQueueName)
+end
 
 -- Add the message to the dead letter queue
 redis.call('ZADD', deadLetterQueueKey, tonumber(redis.call('TIME')[1]), messageId)

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

@@ -0,0 +1,191 @@
+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";
+
+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 authenticatedEnvDev = {
+  id: "e1234",
+  type: "DEVELOPMENT" as const,
+  maximumConcurrencyLimit: 10,
+  concurrencyLimitBurstFactor: new Decimal(2.0),
+  project: { id: "p1234" },
+  organization: { id: "o1234" },
+};
+
+function createQueue(redisContainer: any) {
+  return new RunQueue({
+    ...testOptions,
+    masterQueueConsumersDisabled: true,
+    workerOptions: { disabled: true },
+    queueSelectionStrategy: new FairQueueSelectionStrategy({
+      redis: {
+        keyPrefix: "runqueue:test:",
+        host: redisContainer.getHost(),
+        port: redisContainer.getPort(),
+      },
+      keys: testOptions.keys,
+    }),
+    redis: {
+      keyPrefix: "runqueue:test:",
+      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: "task/my-task",
+    timestamp: Date.now(),
+    attempt: 0,
+    ...overrides,
+  };
+}
+
+const QUEUE = "task/my-task";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+// A concurrency key is an unrestricted client string, so `*` is reachable from the public
+// API, and `queueKey` renders it as `...:queue:<q>:ck:*`, which is byte-identical to the
+// wildcard member the CK scripts keep in the master queue. Each of those scripts rebalances
+// the master queue with that wildcard member and then removes the "old-format" entry for the
+// variant it just touched. When the variant IS the wildcard, the second call undid the
+// first, taking the whole base queue's master-queue entry with it: nothing pointed at the
+// queue any more, so every concurrency key on it stopped being dequeued, silently, until
+// some later write happened to re-add the member.
+describe("concurrency key of '*'", () => {
+  redisTest("enqueueing it leaves the base queue reachable", async ({ redisContainer }) => {
+    const queue = createQueue(redisContainer);
+    try {
+      const t0 = Date.now() - 100_000;
+      const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+      const masterQueueKey = testOptions.keys.masterQueueKeyForShard(shard);
+
+      // An ordinary key with real queued work: the bystander that used to be taken down.
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r-victim", concurrencyKey: "user-1", timestamp: t0 }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+      expect(await queue.redis.zcard(masterQueueKey)).toBe(1);
+
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r-star", concurrencyKey: "*", timestamp: t0 + 1 }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+
+      // The master queue still points at this base queue.
+      expect(await queue.redis.zcard(masterQueueKey)).toBe(1);
+
+      // Both variants are registered, and both runs come back out.
+      const ckIndexKey = testOptions.keys.ckIndexKeyFromQueue(
+        testOptions.keys.queueKey(authenticatedEnvDev, QUEUE, "user-1")
+      );
+      expect((await queue.redis.zrange(ckIndexKey, 0, -1)).length).toBe(2);
+
+      const served = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 10);
+      expect(served.map((m) => m.messageId).sort()).toEqual(["r-star", "r-victim"]);
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("acking it leaves the base queue reachable", async ({ redisContainer }) => {
+    const queue = createQueue(redisContainer);
+    try {
+      const t0 = Date.now() - 100_000;
+      const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+      const masterQueueKey = testOptions.keys.masterQueueKeyForShard(shard);
+
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r-star", concurrencyKey: "*", timestamp: t0 }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r-victim", concurrencyKey: "user-1", timestamp: t0 + 1 }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+
+      // Ack the '*' run while the other key still has work queued: the ack script runs the
+      // same rebalance-then-cleanup pair as the enqueue one.
+      await queue.acknowledgeMessage(authenticatedEnvDev.organization.id, "r-star", {
+        skipDequeueProcessing: true,
+      });
+
+      expect(await queue.redis.zcard(masterQueueKey)).toBe(1);
+
+      const served = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 10);
+      expect(served.map((m) => m.messageId)).toEqual(["r-victim"]);
+    } finally {
+      await queue.quit();
+    }
+  });
+
+  redisTest("nacking it leaves the base queue reachable", async ({ redisContainer }) => {
+    const queue = createQueue(redisContainer);
+    try {
+      const t0 = Date.now() - 100_000;
+      const shard = testOptions.keys.masterQueueShardForEnvironment(authenticatedEnvDev.id, 2);
+      const masterQueueKey = testOptions.keys.masterQueueKeyForShard(shard);
+
+      await queue.enqueueMessage({
+        env: authenticatedEnvDev,
+        message: makeMessage({ runId: "r-star", concurrencyKey: "*", timestamp: t0 }),
+        workerQueue: authenticatedEnvDev.id,
+        skipDequeueProcessing: true,
+      });
+
+      const [dequeued] = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 1);
+      expect(dequeued?.messageId).toBe("r-star");
+
+      await queue.nackMessage({
+        orgId: authenticatedEnvDev.organization.id,
+        messageId: "r-star",
+        retryAt: Date.now() - 1,
+        skipDequeueProcessing: true,
+      });
+
+      expect(await queue.redis.zcard(masterQueueKey)).toBe(1);
+
+      const served = await queue.testDequeueFromMasterQueue(shard, authenticatedEnvDev.id, 1);
+      expect(served.map((m) => m.messageId)).toEqual(["r-star"]);
+    } finally {
+      await queue.quit();
+    }
+  });
+});