From 89365c8b533e16f4058af456a9609171e4b3f464 Mon Sep 17 00:00:00 2001 From: Mykhailo Stadnyk Date: Fri, 18 Sep 2026 12:48:23 +0200 Subject: [PATCH 1/3] test(cluster): share one refusing subscribe double across the catch-up specs Four specs each carried their own copy of the same mock: a subscribe() that refuses until a flag flips and then records the handler at completion, as RedisQueue.subscribe does. A fifth spec that needs it is about to land, so the copies become one `refusing()` helper next to the fake host it complements. It takes the target, which is a single host or RedisQueue.prototype for every host, and hands back the gate and the mock. No behaviour under test changes: 418 specs before and after. --- test/unit/ClusteredRedisQueue.spec.ts | 96 +++++++++------------------ 1 file changed, 33 insertions(+), 63 deletions(-) diff --git a/test/unit/ClusteredRedisQueue.spec.ts b/test/unit/ClusteredRedisQueue.spec.ts index 9bb14dc..5814f36 100644 --- a/test/unit/ClusteredRedisQueue.spec.ts +++ b/test/unit/ClusteredRedisQueue.spec.ts @@ -629,6 +629,30 @@ describe('ClusteredRedisQueue handler catch-up', () => { return host; }; + // Makes `subscribe` refuse until the gate opens, on one host or, through + // the prototype, on every host. Once open it records the handler at + // completion, just as RedisQueue.subscribe does. + const refusing = (target: any) => { + const gate = { refuse: true }; + const sub = mock.method( + target, + 'subscribe', + async function ( + this: any, + _channel: string, + handler: (data: any) => void, + ) { + if (gate.refuse) { + throw new Error('refused'); + } + + this.subscriptionHandlers.push(handler); + }, + ); + + return { gate, sub }; + }; + it('rejects an empty channel name and a second channel, even with no hosts', async () => { const cq = clusterOf(); await assert.rejects(cq.subscribe('', first), TypeError); @@ -996,28 +1020,13 @@ describe('ClusteredRedisQueue handler catch-up', () => { // a host that joins while its subscription connection is refused: // nothing is recorded, so the connection layer has nothing to replay const host = hostOf(cq); - let refuse = true; - const sub = mock.method( - host, - 'subscribe', - async function ( - this: any, - channel: string, - handler: (data: any) => void, - ) { - if (refuse) { - throw new Error('refused'); - } - - this.subscriptionHandlers.push(handler); - }, - ); + const { gate, sub } = refusing(host); await assert.rejects(cq.syncHost(host), /refused/); assert.deepEqual(host.subscriptionHandlers, []); cq.scheduleSync(host, Promise.resolve(), () => undefined); - refuse = false; + gate.refuse = false; t.mock.timers.tick(1000); await settled(); @@ -1044,18 +1053,7 @@ describe('ClusteredRedisQueue handler catch-up', () => { await cq.subscribe('Events', first); - let refuse = true; - const sub = mock.method( - RedisQueue.prototype, - 'subscribe', - async function (this: any, channel: string, handler: any) { - if (refuse) { - throw new Error('refused'); - } - - this.subscriptionHandlers.push(handler); - }, - ); + const { gate, sub } = refusing(RedisQueue.prototype); // the join path itself has to schedule the retry - nothing in the test // touches scheduleSync, so removing that wiring must fail here @@ -1070,7 +1068,7 @@ describe('ClusteredRedisQueue handler catch-up', () => { 'the first catch-up should have failed', ); - refuse = false; + gate.refuse = false; t.mock.timers.tick(1000); await settled(); await settled(); @@ -1096,18 +1094,7 @@ describe('ClusteredRedisQueue handler catch-up', () => { await cq.subscribe('Events', first); - let refuse = true; - const sub = mock.method( - RedisQueue.prototype, - 'subscribe', - async function (this: any, channel: string, handler: any) { - if (refuse) { - throw new Error('refused'); - } - - this.subscriptionHandlers.push(handler); - }, - ); + const { gate, sub } = refusing(RedisQueue.prototype); const send = mock.method( RedisQueue.prototype, 'send', @@ -1123,7 +1110,7 @@ describe('ClusteredRedisQueue handler catch-up', () => { // the join failed its catch-up, so nothing was announced yet assert.equal(send.mock.callCount(), 0); - refuse = false; + gate.refuse = false; t.mock.timers.tick(1000); await settled(); await settled(); @@ -1154,27 +1141,12 @@ describe('ClusteredRedisQueue handler catch-up', () => { assert.deepEqual(host.subscriptionHandlers, [first]); // the second registration fails on this host, the first is already in - let refuse = true; - const sub = mock.method( - host, - 'subscribe', - async function ( - this: any, - channel: string, - handler: (data: any) => void, - ) { - if (refuse) { - throw new Error('refused'); - } - - this.subscriptionHandlers.push(handler); - }, - ); + const { gate, sub } = refusing(host); await assert.rejects(cq.subscribe('Events', second), /refused/); cq.scheduleSync(host, Promise.resolve(), () => undefined); - refuse = false; + gate.refuse = false; t.mock.timers.tick(1000); await settled(); @@ -1198,9 +1170,7 @@ describe('ClusteredRedisQueue handler catch-up', () => { await cq.subscribe('Events', first); const host = hostOf(cq); - const sub = mock.method(host, 'subscribe', async () => { - throw new Error('refused'); - }); + const { sub } = refusing(host); await assert.rejects(cq.syncHost(host), /refused/); cq.scheduleSync(host, Promise.resolve(), () => undefined); From 29980e34119802c9ceb58b5cb53079ba6423453d Mon Sep 17 00:00:00 2001 From: Mykhailo Stadnyk Date: Fri, 18 Sep 2026 12:48:31 +0200 Subject: [PATCH 2/3] fix(cluster): retry a member that refused a live registration The retry added for #31 was wired to the join only. subscribe() on a cluster that already has members fans syncHost() out to them with nothing behind it, so a member that refused its first subscribe ended exactly where a joining host used to: RedisQueue.subscribe() recorded no handler, the connection layer reconnected and restored a socket subscribed to no channel, and nothing asked again. A statically configured cluster takes no other path, because its hosts are added without a catch-up, so for it the defect in #31 was still fully open. The application could not repair it either. A rejected subscribe() is documented as not retryable: the registration stays remembered, and a second call registers a duplicate on every host that did accept it. Against a real broker behind a port that refuses first and then answers, measured 9 s after the address came up: scenario 3.5.2 after #32 now subscribe, then the host joins silent repaired repaired static member down at subscribe time silent silent repaired host joins, subscribe while it refuses silent silent repaired "silent" is a connected subscription socket, zero subscribers reported by redis, and a handler that never fires. The live path now hands a refusing member to the same scheduleSync() the join uses. Nothing else is new. The per-host timer guard already stops a second schedule while a retry is pending, and that retry installs the whole missing suffix, so a registration arriving meanwhile is covered by it. A join's announcement cannot be displaced: its catch-up is enqueued on the host's chain in the same synchronous step that makes the host a member, so its retry is always scheduled before a live registration's can be. scheduleSync()'s startup and announce arguments become optional, since a member has nothing to announce. subscribe() still rejects, so the caller learns a host was unreachable when it subscribed; it no longer means that host stays unsubscribed. The log line and the doc-blocks say so. One unit spec pins the wiring, and nothing in it touches scheduleSync. The suffix spec loses the manual scheduleSync() call it needed to get the same effect, which is what showed the wiring was missing. One integration spec drives a real broker through a statically configured member that refuses and then answers, and asks redis itself for the subscriber, since the live path announces nothing. Removing the wiring fails both unit specs and times the integration spec out. --- CHANGELOG.md | 12 +++ src/ClusteredRedisQueue.ts | 44 +++++++-- test/integration/clusterSubscription.spec.ts | 99 +++++++++++++++++++- test/unit/ClusteredRedisQueue.spec.ts | 34 ++++++- 4 files changed, 176 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8aa1d62..3b05f58 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/ClusteredRedisQueue.ts b/src/ClusteredRedisQueue.ts index faf6c87..c7b149b 100644 --- a/src/ClusteredRedisQueue.ts +++ b/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, - announce: () => void, + started: Promise = 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; diff --git a/test/integration/clusterSubscription.spec.ts b/test/integration/clusterSubscription.spec.ts index 2f6ef70..f8949cc 100644 --- a/test/integration/clusterSubscription.spec.ts +++ b/test/integration/clusterSubscription.spec.ts @@ -124,6 +124,36 @@ const settle = async (received: unknown[], count: number): Promise => { 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 => { + 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 ( promise: Promise, @@ -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); + } + }, + ); }); diff --git a/test/unit/ClusteredRedisQueue.spec.ts b/test/unit/ClusteredRedisQueue.spec.ts index 5814f36..73c43c7 100644 --- a/test/unit/ClusteredRedisQueue.spec.ts +++ b/test/unit/ClusteredRedisQueue.spec.ts @@ -1084,6 +1084,38 @@ describe('ClusteredRedisQueue handler catch-up', () => { await cq.destroy().catch(() => undefined); }); + it('a member that refuses a live registration is retried by the cluster itself', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + + const cq = clusterOf(); + + // already a member when the registration arrives: the live path, and + // the only one a statically configured cluster ever takes + const host = hostOf(cq); + const { gate, sub } = refusing(host); + + await assert.rejects(cq.subscribe('Events', first), /refused/); + assert.deepEqual(host.subscriptionHandlers, []); + + // nothing in the test touches scheduleSync. A rejected subscribe() + // cannot be repeated without registering a duplicate, so removing the + // cluster's own retry must fail here + gate.refuse = false; + t.mock.timers.tick(1000); + await settled(); + await settled(); + + assert.deepEqual( + host.subscriptionHandlers, + [first], + 'the refused registration should have been installed by the retry', + ); + + sub.mock.restore(); + t.mock.timers.reset(); + await cq.destroy(); + }); + it('releases a parked send once a retried host recovers', async t => { t.mock.timers.enable({ apis: ['setTimeout'] }); @@ -1143,9 +1175,9 @@ describe('ClusteredRedisQueue handler catch-up', () => { // the second registration fails on this host, the first is already in const { gate, sub } = refusing(host); + // the rejection schedules the retry itself: nothing here asks for one await assert.rejects(cq.subscribe('Events', second), /refused/); - cq.scheduleSync(host, Promise.resolve(), () => undefined); gate.refuse = false; t.mock.timers.tick(1000); From c5f425ff8185e250073719b937a3ffa7db6271e0 Mon Sep 17 00:00:00 2001 From: Mykhailo Stadnyk Date: Fri, 18 Sep 2026 12:48:40 +0200 Subject: [PATCH 3/3] test(queue): pin the listener reconcile in restoreSubscription restoreSubscription() clears the connection's `message` listeners before it re-attaches the remembered handlers. That is what stops a subscribe() racing a reconnect from leaving its handler attached twice and every message delivered twice. Removing that one line left every unit spec green. The only spec that saw it was the integration one, and CI has no broker, so there it is skipped and the line was unguarded. The race does not need a broker to be pinned, only its end state: a connection that already carries the handler when the restore runs. The spec subscribes, runs the restore again on the same connection, and asserts one listener and one delivery. With the line removed it fails on a listener count of 2. Against a real broker the race is not rare: without the reconcile, 18 of 20 runs of the join repair spec delivered the payload twice. --- test/unit/RedisQueue.spec.ts | 37 ++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/test/unit/RedisQueue.spec.ts b/test/unit/RedisQueue.spec.ts index 6965949..7b7d81e 100644 --- a/test/unit/RedisQueue.spec.ts +++ b/test/unit/RedisQueue.spec.ts @@ -701,6 +701,43 @@ describe('RedisQueue lifecycle', () => { ); }); + it('restoring a subscription leaves one listener per remembered handler', async t => { + const logger = makeLogger(); + const rq: any = new RedisQueue( + 'SubReconcile', + { logger }, + IMQMode.PUBLISHER, + ); + await rq.start(); + t.after(() => rq.destroy().catch(() => undefined)); + + const received: any[] = []; + await rq.subscribe('SubReconcile', (data: any) => received.push(data)); + + // what a reconnect finds when a subscribe() raced it: connect() hands + // the same, not yet restored connection to both callers, so the handler + // is already attached by the time the restore runs + await rq.restoreSubscription(); + + assert.equal( + rq.subscription.listenerCount('message'), + 1, + 'restore must reconcile the listeners, not append to them', + ); + + rq.subscription.emit( + 'message', + 'imq:SubReconcile', + JSON.stringify({ ok: 1 }), + ); + + assert.deepEqual( + received, + [{ ok: 1 }], + 'a message must reach a remembered handler exactly once', + ); + }); + it('unsubscribe() survives a rejecting quit()', async t => { const logger = makeLogger(); const rq: any = new RedisQueue(