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 9bb14dc..73c43c7 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(); @@ -1086,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'] }); @@ -1096,18 +1126,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 +1142,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 +1173,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); + // 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); - refuse = false; + gate.refuse = false; t.mock.timers.tick(1000); await settled(); @@ -1198,9 +1202,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); 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(