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(); + } + }); +});