From 5cd8c9a66355fadcb230bf5f2ad49e9d53a82d6a Mon Sep 17 00:00:00 2001 From: Wes Mason Date: Sat, 15 Aug 2026 03:52:57 +0100 Subject: [PATCH] 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. --- .server-changes/ck-wildcard-queue-strand.md | 6 + .../run-engine/src/run-queue/index.ts | 90 +++++++-- .../src/run-queue/tests/ckWildcardKey.test.ts | 190 ++++++++++++++++++ 3 files changed, 266 insertions(+), 20 deletions(-) create mode 100644 .server-changes/ck-wildcard-queue-strand.md create mode 100644 internal-packages/run-engine/src/run-queue/tests/ckWildcardKey.test.ts diff --git a/.server-changes/ck-wildcard-queue-strand.md b/.server-changes/ck-wildcard-queue-strand.md new file mode 100644 index 0000000000..67642e4cb0 --- /dev/null +++ b/.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. diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index cd6a8ce3bd..b5a7eba25a 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/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) diff --git a/internal-packages/run-engine/src/run-queue/tests/ckWildcardKey.test.ts b/internal-packages/run-engine/src/run-queue/tests/ckWildcardKey.test.ts new file mode 100644 index 0000000000..780573f666 --- /dev/null +++ b/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 { + 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::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(); + } + }); +});