From 43ecf15f80277c0eb931c08ddd89de55179cc795 Mon Sep 17 00:00:00 2001
From: Matt Aitken <matt@mattaitken.com>
Date: Mon, 31 Aug 2026 18:47:52 +0100
Subject: [PATCH] fix(run-engine): publish dead-letter redrives to the
 configured queue channel (#4855)

## Summary

Redriving a run out of the dead letter queue published to the hardcoded
Redis channel `rq:redrive`, while the subscriber listens on
`${options.name}:redrive`. For any `RunQueue` not named `rq` the publish
reached zero subscribers, Redis reported success, and the run stayed in
the dead letter queue with no log, error, or metric.

Both sides now derive the channel from the same expression, and
`redriveMessage` logs an error when a redrive publish reaches zero
subscribers instead of failing silently.

The existing "Dead Letter Queue" test now constructs its queue as
`rq-redrive`, so it fails against the old code (verified red before the
fix, green after) and stops the channel names from silently re-locking
to a single magic value.

Fixes #4854
---
 .../run-engine/src/run-queue/index.test.ts    |  1 +
 .../run-engine/src/run-queue/index.ts         | 24 +++++++++++++++----
 2 files changed, 21 insertions(+), 4 deletions(-)

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 ec731b306..561b649e7 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 4afa5b2ba..1a0feb4ec 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 });