fix(run-engine): publish dead-letter redrives to the configured queue channel

redriveMessage published to the hardcoded rq:redrive channel while the
subscriber listens on the configured queue name, so redrive silently
no-oped for any RunQueue not named rq. Both sides now derive the channel
from the queue name, and a redrive that reaches zero subscribers logs an
error instead of reporting nothing.

Fixes #4854

Author: matt-aitken

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,
       },

internal-packages/run-engine/src/run-queue/index.ts

@@ -671,16 +671,28 @@ 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,
         envId: env.id,
         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 });