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

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