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

## The bug

A concurrency key is an unrestricted client string
(`ConcurrencyKeySchema` is `z.union([z.string(),
z.number()]).transform(String)`), and `concurrencyKeySection` does no
escaping, so `*` reaches the queue raw. `queueKey` then renders it as
`...:queue:<q>:ck:*`, which is byte-identical to the wildcard member the
CK scripts keep in the master queue to mean "this base queue has
concurrency-key work".

Every CK script ends with the same pair:

```lua
-- Rebalance master queue with ck:* member
redis.call('ZADD', masterQueueKey, earliestIdx[2], ckWildcardName)

-- Remove old-format entry from master queue (transition cleanup)
redis.call('ZREM', masterQueueKey, queueName)
```

`ckWildcardName` is `toCkWildcard(message.queue)`, and for a `*`-keyed
run that returns the identical string, so the cleanup on the second line
deletes what the rebalance on the first line just wrote.

The master queue then has no entry for that base queue, while `ckIndex`
and the variant queues still hold the work. **Every concurrency key on
the queue stops being dequeued**, not just the `*` one. It is silent,
and it only recovers if some later write happens to re-add the member.

Reproduced before the fix:

```
master queue AFTER normal ck enqueue: ["{org:...}:queue:task/my-task:ck:*"]
master queue AFTER ck='*' enqueue:    []
ckIndex members (work still queued):  [":ck:user-1", ":ck:*"]
dequeued:                             []
```

Blast radius is bounded to the environment that triggers it, so it is
self-inflicted rather than cross-tenant, but a single trigger stalls the
queue for everything on it.

## The fix

Guard the cleanup so it never removes the wildcard member:

```lua
if queueName ~= ckWildcardName then
  redis.call('ZREM', masterQueueKey, queueName)
end
```

Applied to all 10 CK scripts (4 enqueue, 6 ack/nack/dead-letter). No
key-format change and no migration: a queue already stranded in Redis is
repaired by its next write.

I considered rejecting `*` at the API boundary instead and rejected it.
Existing Redis state and `TaskRun.concurrencyKey` rows already hold raw
`:`-bearing and `*` keys, so changing key construction would orphan
in-flight messages and split concurrency accounting mid-deploy. Boundary
validation would still be reasonable as belt-and-braces later, but the
Lua guard alone fixes it including for state already out there.

## Testing

`ckWildcardKey.test.ts` covers the enqueue, ack and nack paths. All
three pass with the guard and **all three fail without it**, verified by
reverting. Full `src/run-queue/` suite is green (166 tests).

## Note for #4367

The virtual-time branch adds three more CK scripts with the same pattern
(`enqueueMessageCkVtimeTracked`, `enqueueMessageWithTtlCkVtimeTracked`,
`nackMessageCkVtimeTracked`). They do not exist on main so they are not
in this PR; the same guard needs applying there, and I will do that on
that branch.

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,190 @@
+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";
+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();
+    }
+  });
+});