diff --git a/internal-packages/run-engine/src/run-queue/index.test.ts b/internal-packages/run-engine/src/run-queue/index.test.ts index ec731b30686..561b649e7cc 100644 --- a/internal-packages/run-engine/src/run-queue/index.test.ts +++ b/internal-packages/run-engine/src/run-queue/index.test.ts @@ -963,6 +963,7 @@ describe("RunQueue", () => { redisTest("Dead Letter Queue", async ({ redisContainer, redisOptions }) => { const queue = new RunQueue({ ...testOptions, + name: "rq-redrive", retryOptions: { maxAttempts: 1, }, diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 4afa5b2bab9..1a0feb4ec66 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -671,9 +671,8 @@ export class RunQueue { } public async redriveMessage(env: MinimalAuthenticatedEnvironment, messageId: string) { - // Publish redrive message - await this.redis.publish( - "rq:redrive", + const subscriberCount = await this.redis.publish( + this.#redriveChannel, JSON.stringify({ runId: messageId, orgId: env.organization.id, @@ -681,6 +680,19 @@ export class RunQueue { projectId: env.project.id, }) ); + + if (subscriberCount === 0) { + this.logger.error( + "redriveMessage: no subscribers on the redrive channel, message remains in the dead letter queue", + { + channel: this.#redriveChannel, + messageId, + orgId: env.organization.id, + envId: env.id, + projectId: env.project.id, + } + ); + } } public async oldestMessageInQueue( @@ -1460,8 +1472,12 @@ export class RunQueue { ); } + get #redriveChannel() { + return `${this.options.name}:redrive`; + } + async #setupSubscriber() { - const channel = `${this.options.name}:redrive`; + const channel = this.#redriveChannel; this.subscriber.subscribe(channel, (err) => { if (err) { this.logger.error(`Failed to subscribe to ${channel}`, { error: err });