fix(cluster): retry a member that refused a live registration

CHANGELOG.md

@@ -183,6 +183,18 @@ release.
   parked waiting for a usable server wait on that event, so without it they
   would time out against a cluster that had recovered.
 
+- **A cluster member that refused a live registration stayed silent as well.**
+  The retry above covered a joining host only. `subscribe()` on a cluster that
+  already has members — the only route a statically configured cluster ever
+  takes, since its hosts are added without a catch-up — installed the handler on
+  each member, and where one refused it recorded nothing and nothing asked
+  again. The call rejected, but a rejected `subscribe()` cannot be repeated
+  without registering a duplicate, so the application had no repair of its own.
+  A member that refuses a registration is now retried by the same
+  capped-backoff catch-up as a joining host, under the same cancellation rules.
+  The call still rejects, so the caller learns that a host was unreachable when
+  it subscribed; it no longer means that host stays unsubscribed.
+
 - **A subscription handler could be attached twice to the same connection,
   delivering every message twice.** `connect()` binds a connection before it
   awaits it and returns that same object to any concurrent caller, so a

src/ClusteredRedisQueue.ts

@@ -1026,6 +1026,12 @@ export class ClusteredRedisQueue
      * including for future hosts. To rebuild a known registration set, await
      * unsubscribe() and then register the desired handlers again.
      *
+     * A host that refused the registration is not left behind: the cluster
+     * retries its catch-up on its own, with capped backoff, until it succeeds,
+     * the host leaves or the cluster is destroyed. A rejection therefore reports
+     * that a host was unreachable when the call was made, not that it stays
+     * unsubscribed.
+     *
      * The handler receives one invocation per host that delivers the message.
      */
     public async subscribe(
@@ -1062,7 +1068,20 @@ export class ClusteredRedisQueue
                 `${channel}`,
         );
 
-        await Promise.all(this.imqs.map(imq => this.syncHost(imq)));
+        await Promise.all(
+            this.imqs.map(imq =>
+                this.syncHost(imq).catch(err => {
+                    // a member that refuses a live registration has no other
+                    // route back: the caller cannot retry, because calling
+                    // again registers a second copy, and a service that
+                    // subscribes once at start-up never registers again. So the
+                    // cluster retries on its own, exactly as it does for a join
+                    this.scheduleSync(imq);
+
+                    throw err;
+                }),
+            ),
+        );
     }
 
     /**
@@ -1392,13 +1411,22 @@ export class ClusteredRedisQueue
     }
 
     /**
-     * Retries a joining host's subscription catch-up until it succeeds, the
-     * host leaves the cluster, or the cluster is destroyed.
+     * Retries a host's subscription catch-up until it succeeds, the host leaves
+     * the cluster, or the cluster is destroyed.
      *
      * @param imq - the queue whose catch-up failed
+     * @param started - the joining host's startup, which the announcement waits
+     *        for. A live registration has nothing to announce and omits it
+     * @param announce - emits `initialized` for a joining host that only became
+     *        usable through this retry. Omitted by a live registration, whose
+     *        host is already a member that sends are routed to
      *
      * @remarks
-     * A joining host whose first {@link RedisQueue.subscribe} rejects records
+     * Both routes to a first subscribe end here: a joining host's catch-up, and
+     * a live registration on a host that is already a member, which is the only
+     * route a statically configured cluster ever takes.
+     *
+     * A host whose first {@link RedisQueue.subscribe} rejects records
      * nothing: `subscriptionHandlers` stays empty, so the connection layer's
      * own reconnect has nothing to replay and restores a socket subscribed to
      * no channel. Without this, that host never receives another installation —
@@ -1415,8 +1443,8 @@ export class ClusteredRedisQueue
      */
     private scheduleSync(
         imq: RedisQueue,
-        started: Promise<void>,
-        announce: () => void,
+        started: Promise<void> = Promise.resolve(),
+        announce: () => void = () => undefined,
     ): void {
         if (this.closed || !this.imqs.includes(imq)) {
             return;
@@ -1529,8 +1557,8 @@ export class ClusteredRedisQueue
                     'error',
                     `server ${imq.redisKey} failed to subscribe to channel ` +
                         `${channel}, code ${errorCode(err)}: some handlers remain ` +
-                        'uninstalled. A joining host retries this on its own; a ' +
-                        'failure during a live registration is repaired by the next one',
+                        'uninstalled until the retry the cluster schedules for ' +
+                        'this host succeeds',
                 );
 
                 throw err;

test/integration/clusterSubscription.spec.ts

@@ -124,6 +124,36 @@ const settle = async (received: unknown[], count: number): Promise<void> => {
     await new Promise(resolve => setTimeout(resolve, 150));
 };
 
+/**
+ * Resolves once the broker reports a subscriber on `channel`, or rejects on
+ * timeout. The live registration path announces nothing, so the broker is the
+ * only witness that a refused member was subscribed after all.
+ */
+const subscribed = async (
+    publisher: Redis,
+    channel: string,
+    timeoutMs: number,
+): Promise<void> => {
+    const deadline = Date.now() + timeoutMs;
+
+    for (;;) {
+        const [, count] = (await publisher.pubsub('NUMSUB', channel)) as [
+            string,
+            number,
+        ];
+
+        if (+count > 0) {
+            return;
+        }
+
+        if (Date.now() > deadline) {
+            throw new Error(`timed out waiting for a subscriber on ${channel}`);
+        }
+
+        await new Promise(resolve => setTimeout(resolve, 25));
+    }
+};
+
 /** Bounds a test gate without leaving a timer behind on success or failure. */
 const bounded = async <T>(
     promise: Promise<T>,
@@ -232,11 +262,15 @@ class RedisProxy {
 describe('ClusteredRedisQueue subscription over a real broker', () => {
     const queues: ClusteredRedisQueue[] = [];
 
-    const cluster = (name: string, logger = quiet): ClusteredRedisQueue => {
-        // starts EMPTY: the server is added after subscribe(), which is the
-        // path where handlers used to be lost
+    const cluster = (
+        name: string,
+        logger = quiet,
+        servers: Array<{ host: string; port: number }> = [],
+    ): ClusteredRedisQueue => {
+        // starts EMPTY unless told otherwise: the server is added after
+        // subscribe(), which is the path where handlers used to be lost
         const queue = new ClusteredRedisQueue(name, {
-            cluster: [],
+            cluster: servers,
             logger,
         });
 
@@ -563,4 +597,61 @@ describe('ClusteredRedisQueue subscription over a real broker', () => {
             }
         },
     );
+
+    it(
+        'repairs a member whose first live subscription connection is refused',
+        { skip },
+        async () => {
+            const channel = `member-${uuid()}`;
+            const port = await closedPort();
+            const proxy = new RedisProxy(port);
+            const received: unknown[] = [];
+            // a statically configured cluster: the host is a member before any
+            // registration, so the live path is the only one that reaches it
+            const queue = cluster(`member-${uuid()}`, quiet, [
+                { host: '127.0.0.1', port },
+            ]);
+            let publisher: Redis | undefined;
+
+            try {
+                // nothing listens on the port yet, so this is the real refusal
+                await assert.rejects(
+                    queue.subscribe(channel, data => received.push(data)),
+                );
+
+                await proxy.start();
+
+                publisher = new Redis({
+                    host: HOST,
+                    port: PORT,
+                    lazyConnect: true,
+                    retryStrategy: null,
+                });
+                publisher.on('error', quiet.error);
+                await publisher.connect();
+
+                const target = `${(queue as any).options.prefix}:${channel}`;
+
+                // Reconnect and catch-up are each due after one second, and a
+                // catch-up that loses that race is due again two seconds
+                // later; ten seconds leaves ample scheduler and broker slack.
+                await subscribed(publisher, target, 10000);
+
+                assert.equal(
+                    await publisher.publish(
+                        target,
+                        JSON.stringify({ mark: channel }),
+                    ),
+                    1,
+                    'the recovered Redis connection has one subscriber',
+                );
+                await settle(received, 1);
+                assert.deepEqual(received, [{ mark: channel }]);
+            } finally {
+                publisher?.disconnect();
+                await queue.destroy().catch(() => undefined);
+                await proxy.close().catch(() => undefined);
+            }
+        },
+    );
 });