From 79b2998edfb843a17d156b377e485a4bfb899ec1 Mon Sep 17 00:00:00 2001 From: cool_machine Date: Mon, 14 Sep 2026 19:55:59 +0200 Subject: [PATCH] fix(cluster): replay every subscription handler to a joining server --- CHANGELOG.md | 88 ++ src/ClusteredRedisQueue.ts | 532 +++++++++--- test/integration/clusterSubscription.spec.ts | 384 +++++++++ test/unit/ClusteredRedisQueue.spec.ts | 805 ++++++++++++++++++- 4 files changed, 1689 insertions(+), 120 deletions(-) create mode 100644 test/integration/clusterSubscription.spec.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a707bfc..1f152a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,11 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [Unreleased] +The clustered subscription changes below are **patch-level bug fixes**: they +restore additive subscription replay and close lifecycle races without adding a +public API. Package versions are updated separately; other unreleased features may require a minor +release. + ### Added - **TLS on connections to the redis broker.** A new `tls` option encrypts every @@ -40,6 +45,14 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm This covers the queue's own connections. `UDPClusterManager` announcements remain unauthenticated UDP broadcast and are unaffected. +- **Integration specs covering clustered subscription against a real broker**, + in `test/integration/clusterSubscription.spec.ts`. Unit mocks verify call + ordering and simulated delivery; these additionally verify Redis subscription + acknowledgments and actual deliveries. Unlike the TLS specs they do not stand + up their own server: they use an ambient one at `REDIS_HOST`/`REDIS_PORT` (default + `127.0.0.1:6379`) and skip, with a reason, when none answers. `npm test` does + not run them. + - **Integration specs covering TLS against a real broker**, in `test/integration/`, run by `npm run test-integration`. They stand up a throwaway TLS-only redis and assert what a mocked `ioredis` cannot: that the @@ -52,6 +65,39 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Changed +- **Clustered subscribe()/unsubscribe() now serialise per host.** A call can + block behind an earlier subscription operation on that host that never + settles. Different hosts proceed independently. Rejected registrations remain + remembered and may have succeeded on some hosts; **subscribe() is not + retryable**. Repeating it adds another registration, including on future + hosts. To rebuild a known set, await unsubscribe() and register that set again. + Function-identity deduplication would break deliberate additive registrations; + rolling back a partial fan-out would require tearing down working subscribers. + +- **Clustered destroy() clears routing membership and subscription state + synchronously.** Queued work cannot reopen a destroyed host. Teardown bypasses + subscription chains so a wedged operation cannot hold it up. Concurrent callers + await the same teardown (including manager removal) and observe the same + AggregateError if it fails. Every independent host teardown and manager removal + is attempted even if others fail. A later destroy() retries only failed tasks; + successful cleanup is not repeated. Discovery admission closes synchronously + and stays closed during and after teardown, including retries. A `send()` + already parked waiting for the first server is rejected as teardown begins, + rather than waiting out `IMQ_SEND_INIT_TIMEOUT` on a timer that keeps the + process alive; `send()` and `subscribe()` on a destroyed instance are + rejected outright instead of stalling or repopulating the cleared state. + Destroyed instances must not be reused. + +- **`ClusteredRedisQueue.subscribe()` now rejects a bad channel on an empty + cluster, where it used to resolve.** Validation lived in the underlying + queues, so with no servers yet there was nothing to raise it: a second + channel name, or an empty one, resolved and left the remembered subscription + naming a channel nothing was subscribed to. The cluster now applies the same + two checks itself, with the same messages the underlying queue uses, before + touching any state. An empty cluster is the normal starting point for + membership discovered at runtime, so a caller that registers before the first + server arrives will see an error it previously did not. + - **Nothing changes for a queue that does not use TLS.** The option is absent from `options` unless it was configured, the connection pool key stays the plain `host:port`, the redis client is handed no `tls` option, and no new @@ -111,6 +157,48 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Fixed +- **A clustered queue gave a server that joined later only the last-registered + subscription handler, silencing every other handler on that host.** + `ClusteredRedisQueue` remembered one `{ channel, handler }` pair, so each + `subscribe()` overwrote the previous one. Handlers were still forwarded to the + hosts already known, which is why this stayed invisible while cluster + membership was stable — but a host added afterwards (a broker replaced, or + discovery completing after the calls at start-up) was subscribed with the + **last** handler alone. With one handler carrying business events and another + carrying cache invalidation, a host could serve the second and silently drop + the first: the socket stays subscribed, the publisher still sees a subscriber + and RPC is unaffected, so nothing surfaces the loss. + + Every registration is now remembered and installed, in registration order, + including the same function registered twice — `subscribe()` remains additive + as documented. + + Live registration and joining-host catch-up share a serialised operation per + host. Each run reads a cluster-owned per-host installation count and installs + the remaining registrations, advancing only after success. Direct host + registrations do not count towards cluster progress. Cluster unsubscribe + resets that count inside the same chain, including after a rejected teardown. + Overlapping runs do not duplicate a cluster registration within an + uninterrupted subscription. This is not an unconditional exactly-once delivery + guarantee: deliberate repeated registrations still invoke twice, publication + across hosts can deliver more than once, and unsubscribe/replacement can cause + a temporary installation that teardown removes before catch-up reinstalls it. + + A host dropped mid-catch-up receives no further installations and is not + announced as initialized. Two kinds of `info` line report the ordinal and + channel of each registration, and how many handlers a catch-up run installed. The verbose + `Initializing queue with state` line has been removed. + +- **A server that failed to start while joining could take the process down.** + Its background initialization now observes rejection, preventing an unhandled + rejection and process exit. Startup and subscription catch-up run independently: + a failed or stalled start does not block pub/sub, preserving the documented + ability to subscribe without start(). Joining hosts still start automatically + in a started cluster; later subscribe() calls do not retry startup. Deferred + joining-host startup and cluster start() share pending startup per host, so + they cannot start that host concurrently. Settled startup is released so an + explicit later start() can retry. + - **A failed connection could crash the process as it was being torn down.** The redis client guards its socket with a one-shot `error` listener, which the failure that brings the connection down spends. A socket that goes on to diff --git a/src/ClusteredRedisQueue.ts b/src/ClusteredRedisQueue.ts index cdee876..152a3be 100644 --- a/src/ClusteredRedisQueue.ts +++ b/src/ClusteredRedisQueue.ts @@ -55,16 +55,36 @@ export interface ClusterServer extends IMessageQueueConnection { /** * Queue instance created for this host. Present once the server has been * registered; the queue may still be starting. + * + * @remarks + * Exposed to inspect or address one specific host. Subscribing through it + * directly is safe only on the channel the cluster itself uses: a queue + * accepts one channel, so a direct subscription to another name makes every + * later cluster registration on that host fail. `unsubscribe()` and + * `destroy()` on it are not + * supported: the cluster tracks how many of its own registrations a host + * has taken, and it cannot see a handler removed behind its back, so a + * registration made afterwards would be installed while an earlier one + * stayed missing. Use {@link ClusteredRedisQueue.unsubscribe} and + * {@link ClusteredRedisQueue.removeServer} instead. */ imq?: RedisQueue; } interface ClusterState { started: boolean; - subscription: { - channel: string; - handler: (data: JsonObject) => void; - } | null; + channel: string | null; + handlers: Array<(data: JsonObject) => void>; +} + +/** + * Serialises subscription changes for one host. + */ +interface HostProgress { + /** Repaired tail; callers receive the original operation's rejection. */ + chain: Promise; + /** Registrations successfully installed by this cluster since teardown. */ + installed: number; } /** @@ -80,8 +100,8 @@ interface ClusterState { * `clear`, `destroy`, `publish`, `subscribe`, `unsubscribe` and `queueLength` — * fans out to every server. * - * Every fan-out uses `Promise.all`, so one failing host fails the whole call with - * no partial-failure reporting and no rollback. + * Fan-out normally uses `Promise.all`, with no rollback on failure. Destroy + * attempts every cleanup task and reports failures together in an AggregateError. * * The class only `implements` the `EventEmitter` interface rather than extending * it, so `instanceof EventEmitter` is false and every emitter method is a @@ -168,14 +188,41 @@ export class ClusteredRedisQueue * @remarks * Cluster membership changes at runtime, so a per-host queue may be created * long after `start()` and `subscribe()` were called on the cluster. This - * records those calls; {@link ClusteredRedisQueue.initializeQueue} replays - * them onto each new queue. + * records those calls; joining hosts replay startup and subscription + * catch-up independently. */ private state: ClusterState = { started: false, - subscription: null, + channel: null, + handlers: [], }; + /** + * Tracks this cluster's successful installations, independently of handlers + * registered directly on a host. Teardown resets the installation count. + */ + private readonly progress = new WeakMap(); + + /** + * Sends parked in {@link ClusteredRedisQueue.sendWhenInitialized}, waiting + * for a server to appear. Kept so {@link ClusteredRedisQueue.destroy} can + * settle them at once rather than leaving each to time out against a + * cluster that can no longer admit a server. + */ + private readonly waitingSends = new Set<(reason: Error) => void>(); + + /** Pending startup shared by batch startup and joining-host initialization. */ + private readonly starting = new WeakMap>(); + + /** Membership admission stays closed after the first destroy() call. */ + private closed = false; + + /** Failed cleanup tasks remain here for an explicit destroy() retry. */ + private readonly cleanup = new Set<() => Promise>(); + + /** Teardown shared by concurrent destroy() callers. */ + private destroying?: Promise; + /** * Handles for the cluster managers this queue is registered with, kept so * that {@link ClusteredRedisQueue.destroy} can deregister from each. @@ -328,6 +375,16 @@ export class ClusteredRedisQueue delay?: number, errorHandler?: (err: Error) => void, ): Promise { + if (this.closed) { + // admission is closed, so no server can ever arrive: waiting out + // the initialisation timeout would stall a shutdown path for + // IMQ_SEND_INIT_TIMEOUT on a timer that is not unref()'d + throw new TypeError( + 'ClusteredRedisQueue: the queue was destroyed and cannot be ' + + 'reused, so this message has nowhere to go!', + ); + } + if (!this.imqLength) { return this.sendWhenInitialized( toQueue, @@ -396,24 +453,38 @@ export class ClusteredRedisQueue return new Promise((resolve, reject) => { const onInitialized = ({ imq }: { imq: RedisQueue }): void => { clearTimeout(timer); + this.waitingSends.delete(giveUp); imq.send(toQueue, message, delay, errorHandler).then( resolve, reject, ); }; - const timer = setTimeout(() => { + const giveUp = (reason: Error): void => { + clearTimeout(timer); this.clusterEmitter.removeListener( 'initialized', onInitialized, ); - reject( - new Error( - 'ClusteredRedisQueue: no cluster server became ' + - 'available to send the message', + this.waitingSends.delete(giveUp); + reject(reason); + }; + + const timer = setTimeout( + () => + giveUp( + new Error( + 'ClusteredRedisQueue: no cluster server became ' + + 'available to send the message', + ), ), - ); - }, this.sendInitTimeout); + this.sendInitTimeout, + ); + + // registered so destroy() can settle this immediately: once + // admission is closed no server can ever initialise, and waiting + // out the timeout would hold the event loop open on a shutdown + this.waitingSends.add(giveUp); this.clusterEmitter.once('initialized', onInitialized); }); @@ -421,34 +492,91 @@ export class ClusteredRedisQueue /** * Destroys every server's queue — closing their connections and removing - * their event listeners — then unregisters this cluster from all configured + * their event listeners — and unregisters this cluster from all configured * cluster managers. * * @remarks * Unregistering shuts a manager down entirely once it has no clusters left, * which for {@link UDPClusterManager} also terminates its shared UDP worker. * - * The instance must not be reused afterwards: internal routing state is not - * cleared, so a subsequent {@link ClusteredRedisQueue.send} would silently - * re-open a connection. + * Routing membership and remembered subscriptions are cleared synchronously, + * so queued subscription work cannot reopen a destroyed host. Teardown does + * not wait for the subscription chain; concurrent callers await the same + * teardown, including manager removal. Every independent cleanup is attempted; + * failures are reported together in an AggregateError. A later destroy() + * retries only failed tasks; successful tasks are not repeated. Membership + * admission stays closed, including during retries. The instance must not + * be reused. */ public async destroy(): Promise { - this.state.started = false; + if (this.destroying) { + return this.destroying; + } - await this.batch( - 'destroy', - 'Destroying clustered redis message queue...', - ); + if (!this.closed) { + this.closed = true; + this.state.started = false; - if (!this.options.clusterManagers?.length) { - return; - } + for (const imq of this.imqs) { + this.cleanup.add(() => imq.destroy()); + } - for (const manager of this.options.clusterManagers) { - for (const cluster of this.initializedClusters) { - await manager.remove(cluster); + for (const manager of this.options.clusterManagers || []) { + for (const cluster of this.initializedClusters) { + this.cleanup.add(() => manager.remove(cluster)); + } + } + + // Close routing before queued subscription or startup work can run. + // Teardown bypasses those chains so a stalled install cannot block it. + this.imqs = []; + this.servers = []; + this.imqLength = 0; + this.state.channel = null; + this.state.handlers = []; + + // a send parked waiting for a server can never be satisfied once + // admission is closed, and its timer is referenced, so leaving it + // to expire holds the event loop open for the whole timeout + const parked = Array.from(this.waitingSends); + + this.waitingSends.clear(); + + for (const giveUp of parked) { + giveUp( + new Error( + 'ClusteredRedisQueue: the queue was destroyed before ' + + 'a cluster server became available', + ), + ); } } + + this.destroying = Promise.resolve() + .then(async () => { + this.logLine( + 'info', + 'Destroying clustered redis message queue...', + ); + const results = await Promise.allSettled( + [...this.cleanup].map(async operation => { + await operation(); + this.cleanup.delete(operation); + }), + ); + const errors = results + .filter(result => result.status === 'rejected') + .map(result => result.reason); + + if (errors.length) { + throw new AggregateError(errors, 'Cluster teardown failed'); + } + }) + .finally(() => { + this.destroying = undefined; + }); + + return this.destroying; } /** @@ -530,7 +658,7 @@ export class ClusteredRedisQueue * @param message - */ private async batch( - action: 'start' | 'stop' | 'destroy' | 'clear', + action: 'start' | 'stop' | 'clear', message: string, ): Promise { this.logger.info(message); @@ -538,9 +666,9 @@ export class ClusteredRedisQueue const promises: Promise[] = []; for (const imq of this.imqs) { - const run = imq[action] as () => Promise; - - promises.push(run.call(imq)); + promises.push( + action === 'start' ? this.startHost(imq) : imq[action](), + ); } await Promise.all(promises); @@ -857,15 +985,30 @@ export class ClusteredRedisQueue * * @param channel - channel name within the queue's prefix namespace * @param handler - invoked with the parsed payload of each published message - * @throws TypeError when a different channel name is supplied while a - * subscription is already open on the underlying queues + * @throws TypeError when no channel name is given, or when a different + * channel name is supplied while this instance already remembers + * one - both are raised here, so they fire on an empty cluster too, + * where there is no underlying queue to raise them * * @remarks * Only one channel per instance is supported. Calling this again with the - * same channel registers the handler a second time; calling it with a - * different channel rejects — and the remembered subscription is left - * pointing at the rejected name, which is what newly joining servers would - * then use. + * same channel registers an additional handler — every registration is + * remembered and all of them are invoked, including the same function + * registered twice. Calling it with a different channel throws before any + * state is touched, so the remembered channel keeps naming the channel that + * is actually subscribed. + * + * Servers joining later are given every handler registered before they + * joined, in registration order. + * + * Subscription uses its own connection and does not require start(), even + * when a host's startup fails or stalls. Subscription changes serialise per + * host, so a call can wait behind an earlier operation that never settles. + * + * A rejected call is not retryable: its registration remains remembered and + * may already be installed on some hosts. Calling again adds another copy, + * including for future hosts. To rebuild a known registration set, await + * unsubscribe() and then register the desired handlers again. * * The handler receives one invocation per host that delivers the message. */ @@ -873,35 +1016,68 @@ export class ClusteredRedisQueue channel: string, handler: (data: JsonObject) => void, ): Promise { - this.state.subscription = { channel, handler }; + if (this.closed) { + throw new TypeError( + 'ClusteredRedisQueue: the queue was destroyed and cannot be ' + + 'reused, so this subscription would never reach a server!', + ); + } - const promises: Array> = []; + if (!channel) { + throw new TypeError( + `${channel}: No subscription channel name provided!`, + ); + } - for (const imq of this.imqs) { - promises.push(imq.subscribe(channel, handler)); + if (this.state.channel && this.state.channel !== channel) { + throw new TypeError( + `Invalid channel name provided: expected "${ + this.state.channel + }", but "${channel}" given instead!`, + ); } - await Promise.all(promises); + this.state.channel = channel; + this.state.handlers.push(handler); + + this.logLine( + 'info', + `registered handler #${this.state.handlers.length} for channel ` + + `${channel}`, + ); + + await Promise.all(this.imqs.map(imq => this.syncHost(imq))); } /** - * Unsubscribes from the channel on every redis host and forgets the - * remembered subscription, so servers joining later are no longer subscribed + * Unsubscribes from the channel on every redis host and forgets every + * remembered handler, so servers joining later are no longer subscribed * automatically. * * @remarks * Resolves without effect on an empty cluster. + * + * Clears the remembered channel and handlers immediately, then queues each + * host's teardown behind its current subscription work. A stalled operation + * on that host therefore also stalls unsubscribe(). Later catch-up reads the + * cluster's installation count after teardown, even if an earlier run + * temporarily installed handlers from the replacement list. */ public async unsubscribe(): Promise { - this.state.subscription = null; - - const promises: Array> = []; - - for (const imq of this.imqs) { - promises.push(imq.unsubscribe()); - } - - await Promise.all(promises); + this.state.channel = null; + this.state.handlers = []; + + await Promise.all( + this.imqs.map(imq => + this.enqueue(imq, async () => { + try { + await imq.unsubscribe(); + } finally { + this.progressOf(imq).installed = 0; + } + }), + ), + ); } /** @@ -921,7 +1097,8 @@ export class ClusteredRedisQueue * * For a genuinely new server this returns as soon as the record is created — * starting the queue and re-applying any active subscription happen - * asynchronously afterwards. + * asynchronously afterwards. Once destroy() begins, discovery is ignored: + * the returned address has no queue and is not admitted to membership. */ protected addServer(server: IServerInput): ClusterServer { this.verbose(`Adding new server: ${JSON.stringify(server)}`); @@ -955,9 +1132,16 @@ export class ClusteredRedisQueue const imqToRemove = remove.imq; if (imqToRemove) { + // dropped from routing first: a catch-up run in progress tests + // membership between handlers and stops as soon as it sees this this.imqs = this.imqs.filter( imq => imqToRemove.redisKey !== imq.redisKey, ); + + // not queued behind this host's other work: a queue wedged on a + // connection that never answers would then never be torn down at + // all, and teardown is the one operation that has to happen + // regardless of what the host is doing imqToRemove .destroy() .catch((err: unknown) => @@ -990,6 +1174,10 @@ export class ClusteredRedisQueue server: ClusterServer, initializeQueue: boolean = true, ): ClusterServer { + if (this.closed) { + return { ...server, imq: undefined }; + } + const existingServer = this.findServer(server); if (existingServer) { @@ -1015,19 +1203,39 @@ export class ClusteredRedisQueue copyEventEmitter(this.templateEmitter, imq); - if (initializeQueue) { - this.initializeQueue(imq).then(() => { - this.clusterEmitter.emit('initialized', { - server: newServer, - imq, - }); - }); - } - newServer.imq = imq; + // registered before the catch-up run starts, so that run can test + // membership against `imqs` and see the host it is working on. Nothing + // can observe the order: there is no await between here and the call this.imqs.push(imq); this.servers.push(newServer); + + if (initializeQueue) { + // Lifecycle and subscription use separate connections: a stalled + // start must not hold up the host's subscription chain. + Promise.all([this.startHost(imq), this.syncHost(imq)]).then( + () => { + // a host dropped while it was being brought up to date + // never became a member, and announcing it would release a + // send that is waiting for a usable server onto a queue + // that is being destroyed + if (!this.imqs.includes(imq)) { + return; + } + + this.clusterEmitter.emit('initialized', { + server: newServer, + imq, + }); + }, + // reported inside the run; without a handler here a host that + // simply refuses a connection - routine - becomes an unhandled + // rejection, which is fatal on current node defaults + () => undefined, + ); + } + this.clusterEmitter.emit('add', { server: newServer, imq }); this.imqLength = this.imqs.length; @@ -1048,58 +1256,176 @@ export class ClusteredRedisQueue } /** - * Brings a newly created per-host queue up to the cluster's current state. + * Appends one operation to a host's serialised queue. * - * @param imq - the queue to initialize + * @param imq - the queue the operation belongs to + * @param operation - the work to run once everything before it has finished + * @returns a promise for this operation alone, which rejects if it fails * * @remarks - * Replays whatever {@link ClusteredRedisQueue.state} records, so a server - * that joins after the cluster started is started and subscribed too rather - * than sitting idle. + * Every `subscribe`, `unsubscribe` and catch-up run for a host goes through + * here, so operations on one host never overlap, while different hosts stay + * independent. The tail kept for the next operation is deliberately + * repaired with a `catch`: chaining onto a rejected tail would make one + * failed operation reject every operation the host is ever given again, + * which is precisely the "host that silently stopped working" this class + * has to avoid. The caller still receives the failure, through the returned + * promise. */ - private async initializeQueue(imq: RedisQueue): Promise { - this.verbose( - `Initializing queue with state: ${JSON.stringify(this.state)}`, - ); + private enqueue( + imq: RedisQueue, + operation: () => Promise, + ): Promise { + const progress = this.progressOf(imq); + const run = progress.chain.then(operation); - // both failures are reported here, inside the function, and the - // value is re-thrown as it was: the caller starts this without - // awaiting it, so a new .catch() would either swallow the failure or - // add a second unhandled rejection - if (this.state.started) { - try { - await imq.start(); - } catch (err) { - this.logLine( - 'error', - `server ${imq.redisKey} failed to start, code ${errorCode( - err, - )}: the node is not ready to serve queues`, - ); + progress.chain = run.catch(() => undefined); - throw err; - } + return run; + } + + /** + * Returns the progress record for a host, creating it on first use. + * + * @param imq - the queue to look up + * @returns that host's progress record + */ + private progressOf(imq: RedisQueue): HostProgress { + let progress = this.progress.get(imq); + + if (!progress) { + progress = { + chain: Promise.resolve(), + installed: 0, + }; + + this.progress.set(imq, progress); + } + + return progress; + } + + /** + * Starts a joining host if it still belongs to a started cluster. + * Subscription catch-up proceeds independently of this lifecycle operation. + */ + private startHost(imq: RedisQueue): Promise { + // initial eligibility is decided here, synchronously, and rechecked + // again before starting: the + // promise is cached the moment it is created, so a body that decides + // later hands a caller arriving afterwards a settled promise whose + // decision was made against state this caller has since changed. That + // is how `cluster.start()` could join a run created while the cluster + // was still stopped and never start the host at all + if (!this.state.started || !this.imqs.includes(imq)) { + return Promise.resolve(); } - if (this.state.subscription) { + const pending = this.starting.get(imq); + + if (pending) { + return pending; + } + + const run = Promise.resolve() + .then(async () => { + // rechecked here as well as synchronously above: a stop() can + // land between creating this run and the microtask that runs + // it, and starting a host the cluster has just stopped leaves + // the cluster stopped with a host running + if (this.state.started && this.imqs.includes(imq)) { + try { + await imq.start(); + } catch (err) { + this.logLine( + 'error', + `server ${imq.redisKey} failed to start, code ` + + `${errorCode(err)}: the node is not ready to serve queues`, + ); + + throw err; + } + } + }) + .finally(() => { + this.starting.delete(imq); + }); + + this.starting.set(imq, run); + + return run; + } + + /** + * Installs registrations this cluster has not yet installed on a host. + * + * @param imq - the queue to bring up to date + * @returns this run's completion, rejecting on a failed installation + * + * @remarks + * Both live registrations and joining hosts use the same serialised path. + * Each run reads the cluster-owned count inside the chain, so repeating it adds + * nothing. A queued teardown may erase a temporary installation; the run + * behind that teardown will then see and install the missing suffix again. + */ + private syncHost(imq: RedisQueue): Promise { + return this.enqueue(imq, async () => { + const progress = this.progressOf(imq); + const channel = this.state.channel; + + if (!channel) { + return; + } + + let installed = 0; + try { - await imq.subscribe( - this.state.subscription.channel, - this.state.subscription.handler, - ); + for ( + let i = progress.installed; + i < this.state.handlers.length; + i++ + ) { + if (!this.imqs.includes(imq)) { + return; + } + + if (this.state.channel !== channel) { + return; + } + + await imq.subscribe(channel, this.state.handlers[i]); + + // rechecked after the await: teardown bypasses this chain, + // so the host can have been removed or destroyed while the + // subscribe was in flight. Recording it would leave the + // counter claiming an install on a queue that is gone, and + // the handler it just attached is torn down with the host + if (!this.imqs.includes(imq)) { + return; + } + + progress.installed = i + 1; + installed++; + } } catch (err) { this.logLine( 'error', - `server ${imq.redisKey} failed to subscribe to channel ${ - this.state.subscription.channel - }, code ${errorCode( - err, - )}: events from this node will never arrive`, + `server ${imq.redisKey} failed to subscribe to channel ` + + `${channel}, code ${errorCode(err)}: some handlers remain ` + + 'uninstalled until a later registration triggers another catch-up', ); throw err; + } finally { + if (installed) { + this.logLine( + 'info', + `server ${imq.redisKey} installed ${installed} handler(s) ` + + `for channel ${channel}`, + ); + } } - } + }); } /** diff --git a/test/integration/clusterSubscription.spec.ts b/test/integration/clusterSubscription.spec.ts new file mode 100644 index 0000000..342645a --- /dev/null +++ b/test/integration/clusterSubscription.spec.ts @@ -0,0 +1,384 @@ +/*! + * ClusteredRedisQueue subscription integration specs + * + * I'm Queue Software Project + * Copyright (C) 2025 imqueue.com + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + * + * If you want to use this code in a closed source (commercial) project, you can + * purchase a proprietary commercial license. Please contact us at + * to get commercial licensing options. + * + * @remarks + * The unit specs replace `ioredis` wholesale, so they can prove which calls a + * cluster makes and can exercise delivery through simulated events. These + * specs additionally check actual Redis acknowledgments and message delivery + * when a server joins after multiple subscribe() calls. + * + * These specs skip - never fail - where no redis is reachable, matching the + * contract the TLS specs follow, so a checkout without redis stays green. + */ +import assert from 'node:assert/strict'; +import { randomUUID as uuid } from 'node:crypto'; +import { once } from 'node:events'; +import { after, describe, it, mock } from 'node:test'; +import { Redis } from 'ioredis'; +import { ClusteredRedisQueue, RedisQueue } from '../../src/index.js'; + +process.setMaxListeners(100); + +const HOST = process.env.REDIS_HOST || '127.0.0.1'; +const PORT = +(process.env.REDIS_PORT || 6379); + +/** Silences the queue; a failing assertion says more than its log would */ +const quiet = { log() {}, info() {}, warn() {}, error() {} }; + +/** + * Confirms a broker answers, returning a skip reason instead of throwing when + * it does not — a machine without redis must report these as skipped. Both + * connection and PING are bounded, and reconnects cannot keep the probe alive. + */ +const brokerReason = async (): Promise => { + const probe = new Redis({ + host: HOST, + port: PORT, + lazyConnect: true, + connectTimeout: 1000, + commandTimeout: 1000, + retryStrategy: null, + }); + + let connectionError: unknown; + probe.on('error', error => { + connectionError = error; + }); + + try { + await probe.connect(); + await probe.ping(); + + return undefined; + } catch (err) { + return `redis at ${HOST}:${PORT} is not reachable: ${String(connectionError || err)}`; + } finally { + probe.disconnect(); + } +}; + +const skip = await brokerReason(); + +/** + * Adds the broker and waits for catch-up to finish, including subscription + * acknowledgments. Listen before adding it so initialization cannot be missed. + */ +const join = async (queue: ClusteredRedisQueue): Promise => { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 5000); + const ready = once((queue as any).clusterEmitter, 'initialized', { + signal: controller.signal, + }); + + try { + (queue as any).addServer({ host: HOST, port: PORT }); + await ready; + } finally { + clearTimeout(timeout); + controller.abort(); + } +}; + +/** Resolves once `received` holds `count` items, or rejects on timeout */ +const settle = async (received: unknown[], count: number): Promise => { + const deadline = Date.now() + 5000; + + while (received.length < count) { + if (Date.now() > deadline) { + throw new Error( + `timed out waiting for ${count} deliveries, got ` + + `${received.length}`, + ); + } + + await new Promise(resolve => setTimeout(resolve, 25)); + } + + // a duplicate would arrive on the same tick as the last expected one + await new Promise(resolve => setTimeout(resolve, 150)); +}; + +/** Bounds a test gate without leaving a timer behind on success or failure. */ +const bounded = async (promise: Promise, label: string): Promise => { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout( + () => reject(new Error(`timed out: ${label}`)), + 5000, + ); + }), + ]); + } finally { + clearTimeout(timer); + } +}; + +describe('ClusteredRedisQueue subscription over a real broker', () => { + const queues: ClusteredRedisQueue[] = []; + + const cluster = (name: string): ClusteredRedisQueue => { + // starts EMPTY: the server is added after subscribe(), which is the + // path where handlers used to be lost + const queue = new ClusteredRedisQueue(name, { + cluster: [], + logger: quiet, + }); + + queues.push(queue); + + return queue; + }; + + after(async () => { + for (const queue of queues) { + await queue.destroy().catch(() => undefined); + } + }); + + it( + 'preserves one subscription without starting the queue', + { skip }, + async () => { + const channel = `chan-${uuid()}`; + const queue = cluster(`single-${uuid()}`); + const received: unknown[] = []; + await join(queue); + await queue.subscribe(channel, data => { + received.push(data); + }); + const publisher = new Redis({ + host: HOST, + port: PORT, + retryStrategy: null, + }); + publisher.on('error', quiet.error); + try { + assert.equal( + await publisher.publish( + `${(queue as any).options.prefix}:${channel}`, + JSON.stringify({ mark: channel }), + ), + 1, + ); + await settle(received, 1); + assert.deepEqual(received, [{ mark: channel }]); + } finally { + publisher.disconnect(); + } + }, + ); + + it( + 'delivers to every handler on a server that joined after subscribe()', + { + skip, + }, + async () => { + const channel = `chan-${uuid()}`; + const queue = cluster(`join-${uuid()}`); + const first: unknown[] = []; + const second: unknown[] = []; + + await queue.start(); + await queue.subscribe(channel, data => first.push(data)); + await queue.subscribe(channel, data => second.push(data)); + + // the server arrives only now, so both handlers reach it through the + // catch-up run rather than through the subscribe() calls themselves + await join(queue); + await queue.publish({ mark: channel }, channel); + + await settle(first, 1); + await settle(second, 1); + + assert.deepEqual(first, [{ mark: channel }], 'first handler'); + assert.deepEqual(second, [{ mark: channel }], 'second handler'); + }, + ); + + it( + 'delivers exactly once per handler, with no duplicates', + { + skip, + }, + async () => { + const channel = `chan-${uuid()}`; + const queue = cluster(`once-${uuid()}`); + const received: unknown[] = []; + + await queue.start(); + await queue.subscribe(channel, data => received.push(data)); + + await join(queue); + await queue.publish({ mark: channel }, channel); + + await settle(received, 1); + + // This registration reaches the host only through catch-up; a + // duplicate installation by that run would deliver twice. + assert.equal( + received.length, + 1, + 'exactly one delivery per handler', + ); + }, + ); + + it( + 'delivers exactly once when subscribe lands during host catch-up', + { skip }, + async () => { + const channel = `chan-${uuid()}`; + const queue = cluster(`overlap-${uuid()}`); + const early: unknown[] = []; + const live: unknown[] = []; + const entered = Promise.withResolvers(); + const gate = Promise.withResolvers(); + + await queue.start(); + await queue.subscribe(channel, data => { + early.push(data); + }); + + const controller = new AbortController(); + const ready = once((queue as any).clusterEmitter, 'initialized', { + signal: controller.signal, + }); + // Observe rejection even when the first-registration gate fails first. + void ready.catch(() => undefined); + const original = RedisQueue.prototype.subscribe; + let first = true; + // Intercept BEFORE discovery, including a mutant that starts eagerly. + const subscription = mock.method( + RedisQueue.prototype, + 'subscribe', + async function ( + this: RedisQueue, + name: string, + handler: (data: any) => void, + ) { + if (first) { + first = false; + entered.resolve(); + await bounded( + gate.promise, + 'first registration release', + ); + } + await original.call(this, name, handler); + }, + ); + + try { + (queue as any).addServer({ host: HOST, port: PORT }); + await bounded( + entered.promise, + 'first registration interception', + ); + const registering = queue.subscribe(channel, data => { + live.push(data); + }); + // Preserve deferral: an unserialised run must get a turn while + // the first registration is still held before ACK/installation. + await new Promise(resolve => setImmediate(resolve)); + gate.resolve(); + await bounded( + Promise.all([ready, registering]), + 'host initialization', + ); + await queue.publish({ mark: channel }, channel); + await settle(early, 1); + await settle(live, 1); + assert.deepEqual( + early, + [{ mark: channel }], + 'catch-up handler', + ); + assert.deepEqual(live, [{ mark: channel }], 'live handler'); + } finally { + gate.resolve(); + controller.abort(); + subscription.mock.restore(); + } + }, + ); + + it( + 'keeps two deliberate registrations of the same function', + { + skip, + }, + async () => { + const channel = `chan-${uuid()}`; + const queue = cluster(`twice-${uuid()}`); + const received: unknown[] = []; + const handler = (data: unknown): void => { + received.push(data); + }; + + await queue.start(); + await queue.subscribe(channel, handler); + await queue.subscribe(channel, handler); + + await join(queue); + await queue.publish({ mark: channel }, channel); + + await settle(received, 2); + + // the interface documents repeated registration as additive + assert.equal(received.length, 2, 'both registrations must fire'); + }, + ); + + it( + 'delivers to a handler registered after the server joined', + { + skip, + }, + async () => { + const channel = `chan-${uuid()}`; + const queue = cluster(`late-${uuid()}`); + const early: unknown[] = []; + const late: unknown[] = []; + + await queue.start(); + await queue.subscribe(channel, data => early.push(data)); + + await join(queue); + + // registered once the host is already up to date, so this one goes + // through the live path while the first went through the catch-up run + await queue.subscribe(channel, data => late.push(data)); + await queue.publish({ mark: channel }, channel); + + await settle(early, 1); + await settle(late, 1); + + assert.equal(early.length, 1, 'handler from the catch-up run'); + assert.equal(late.length, 1, 'handler from the live path'); + }, + ); +}); diff --git a/test/unit/ClusteredRedisQueue.spec.ts b/test/unit/ClusteredRedisQueue.spec.ts index d8c3435..cc0943c 100644 --- a/test/unit/ClusteredRedisQueue.spec.ts +++ b/test/unit/ClusteredRedisQueue.spec.ts @@ -1,6 +1,6 @@ /*! * ClusteredRedisQueue Unit Tests (core behavior + EventEmitter proxy methods, - * addServerWithQueueInitializing, initializeQueue, and matchServers) + * addServerWithQueueInitializing, syncHost, and matchServers) * * I'm Queue Software Project * Copyright (C) 2025 imqueue.com @@ -258,7 +258,7 @@ describe('ClusteredRedisQueue', () => { }); describe('subscribe()', () => { - it('should subscribe after queue initialization', () => { + it('should subscribe after queue initialization', async () => { const clusterManager = new (ClusterManager as any)(); const cq: any = new ClusteredRedisQueue('TestClusteredQueue', { clusterManagers: [clusterManager], @@ -269,6 +269,10 @@ describe('ClusteredRedisQueue', () => { cq.subscribe(channel, () => {}); cq.addServer(clusterConfig.cluster[0]); + // addServer() documents the subscription as re-applied + // asynchronously afterwards, so let that run settle + await new Promise(resolve => setImmediate(resolve)); + assert.equal(cq.imqs[0].subscriptionName, channel); }); }); @@ -498,7 +502,8 @@ describe('ClusteredRedisQueue.addServerWithQueueInitializing() default param', ( }); // prevent any actual start/subscription side-effects (cq as any).state.started = false; - (cq as any).state.subscription = null; + (cq as any).state.channel = null; + (cq as any).state.handlers = []; const server = { host: '192.168.0.1', port: 6380 }; const initializedSpy = new Promise(resolve => { @@ -550,8 +555,8 @@ describe('ClusteredRedisQueue.addServerWithQueueInitializing(false)', () => { }); }); -describe('ClusteredRedisQueue.initializeQueue()', () => { - it('should call imq.start when started and imq.subscribe when subscription is set', async () => { +describe('ClusteredRedisQueue.syncHost()', () => { + it('should call imq.start when started and imq.subscribe for each handler', async () => { const startStub: Mock = mock.method( RedisQueue.prototype as any, 'start', @@ -574,7 +579,7 @@ describe('ClusteredRedisQueue.initializeQueue()', () => { const handler = () => undefined; await cq.subscribe(channel, handler); - // adding a server triggers initializeQueue which should call start and subscribe + // adding a server starts its lifecycle and subscription catch-up cq.addServer({ host: '127.0.0.1', port: 6453 }); // allow promises to resolve @@ -588,6 +593,763 @@ describe('ClusteredRedisQueue.initializeQueue()', () => { }); }); +describe('ClusteredRedisQueue handler catch-up', () => { + afterEach(() => mock.restoreAll()); + + const clusterOf = (): any => + new ClusteredRedisQueue('CatchUp', { + cluster: [], + logger, + }); + const settled = (): Promise => + new Promise(resolve => setImmediate(resolve)); + const deferred = () => Promise.withResolvers(); + const first = (): void => undefined; + const second = (): void => undefined; + const third = (): void => undefined; + + // The fake records installed handlers at completion, just as RedisQueue.subscribe does. + const hostOf = (cq: any): any => { + const host = { + redisKey: 'fake', + subscriptionHandlers: [] as Array<(data: any) => void>, + async subscribe(_channel: string, handler: (data: any) => void) { + this.subscriptionHandlers.push(handler); + }, + async unsubscribe() { + this.subscriptionHandlers = []; + }, + async destroy() { + await this.unsubscribe(); + }, + async start() {}, + }; + cq.imqs.push(host); + cq.imqLength = cq.imqs.length; + return host; + }; + + 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); + await cq.subscribe('Events', first); + await assert.rejects(cq.subscribe('Other', second), TypeError); + assert.equal(cq.state.channel, 'Events'); + assert.deepEqual(cq.state.handlers, [first]); + await cq.destroy(); + }); + + it('gives a later-joining server every handler, each exactly once', async () => { + const cq = clusterOf(); + await cq.subscribe('Events', first); + await cq.subscribe('Events', second); + const host = cq.addServer({ host: '127.0.0.1', port: 6601 }).imq; + await settled(); + assert.deepEqual(host.subscriptionHandlers, [first, second]); + await cq.destroy(); + }); + + it('keeps two deliberate registrations of the same function', async () => { + const cq = clusterOf(); + await cq.subscribe('Events', first); + await cq.subscribe('Events', first); + const host = cq.addServer({ host: '127.0.0.1', port: 6602 }).imq; + await settled(); + assert.deepEqual(host.subscriptionHandlers, [first, first]); + await cq.destroy(); + }); + + it('counts only cluster installations when callers subscribe directly on a host', async () => { + const cq = clusterOf(); + const host = cq.addServerWithQueueInitializing( + { host: '127.0.0.1', port: 6610 }, + false, + ).imq; + await cq.subscribe('Events', first); + await host.subscribe('Events', third); + await cq.subscribe('Events', second); + assert.deepEqual(host.subscriptionHandlers, [first, third, second]); + await cq.syncHost(host); + assert.deepEqual(host.subscriptionHandlers, [first, third, second]); + await cq.destroy(); + }); + + it('resets the cluster cursor when unsubscribe completes before replacement', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + await cq.subscribe('Events', first); + await cq.unsubscribe(); + await cq.subscribe('Events', second); + assert.deepEqual(host.subscriptionHandlers, [second]); + await cq.destroy(); + }); + + it('shares a pending start between deferred discovery and concurrent cluster starts', async () => { + const cq = clusterOf(); + const gate = deferred(); + const start = mock.method( + RedisQueue.prototype, + 'start', + async function (this: RedisQueue) { + await gate.promise; + return this; + }, + ); + cq.addServer({ host: '127.0.0.1', port: 6611 }); + const one = cq.start(); + const two = cq.start(); + try { + await settled(); + assert.equal(start.mock.callCount(), 1); + } finally { + gate.resolve(); + await Promise.all([one, two]); + await cq.destroy(); + } + }); + + for (const fails of [false, true]) { + it(`releases pending startup after ${fails ? 'failure' : 'success'} so start can run again`, async () => { + const cq = clusterOf(); + const host = hostOf(cq); + let calls = 0; + mock.method(host, 'start', async () => { + if (++calls === 1 && fails) throw new Error('start refused'); + }); + if (fails) await assert.rejects(cq.start(), /start refused/); + else await cq.start(); + await cq.start(); + assert.equal(calls, 2); + await cq.destroy(); + }); + } + + it('starts a host added before start(), whatever the await depth between them', async () => { + const started: any[] = []; + + mock.method( + RedisQueue.prototype as any, + 'start', + async function (this: any): Promise { + started.push(this); + }, + ); + mock.method( + RedisQueue.prototype as any, + 'subscribe', + async () => undefined, + ); + + const cq = clusterOf(); + const joined = cq.addServer({ host: '127.0.0.1', port: 6801 }).imq; + + // the decision to start is taken when the run is created, but the run + // is cached at once - so a start() arriving a microtask later used to + // join a settled promise that had already decided not to start + await Promise.resolve(); + + await cq.start(); + await settled(); + + assert.equal( + started.includes(joined), + true, + 'a host added before start() must still be started', + ); + + await cq.destroy(); + }); + + it('clears the remembered subscription when the cluster is destroyed', async () => { + const cq = clusterOf(); + + await cq.subscribe('Events', () => undefined); + await cq.destroy(); + + assert.equal((cq as any).state.channel, null); + assert.deepEqual((cq as any).state.handlers, []); + }); + + it('installs onto a joining host while its start is still stalled', async () => { + let releaseStart: () => void = () => undefined; + const starting = new Promise(resolve => { + releaseStart = resolve; + }); + + mock.method( + RedisQueue.prototype as any, + 'start', + async () => await starting, + ); + + const subscribeStub: Mock = mock.method( + RedisQueue.prototype as any, + 'subscribe', + async () => undefined, + ); + + const cq = clusterOf(); + + await cq.start(); + await cq.subscribe('Events', first); + + const joined = cq.addServer({ host: '127.0.0.1', port: 6802 }).imq; + + await settled(); + + // no later subscribe() here: the handler must arrive through the + // catch-up alone, which must not be sequenced behind a stalled start + const onJoined = subscribeStub.mock.calls + .filter((call: any) => call.this === joined) + .map((call: any) => call.arguments); + + assert.deepEqual(onJoined, [['Events', first]]); + + releaseStart(); + await cq.destroy(); + }); + + it('repeating catch-up installs nothing on an up-to-date host', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const sub = mock.method(host, 'subscribe'); + await cq.subscribe('Events', first); + await cq.syncHost(host); + await cq.syncHost(host); + assert.equal(sub.mock.callCount(), 1); + assert.deepEqual(host.subscriptionHandlers, [first]); + await cq.destroy(); + }); + + it('preserves a single live subscription without starting its host', async () => { + const cq = clusterOf(); + const host = cq.addServerWithQueueInitializing( + { host: '127.0.0.1', port: 6603 }, + false, + ).imq; + const start = mock.method(host, 'start', async () => { + throw new Error('unavailable'); + }); + cq.state.started = true; + const received: unknown[] = []; + const handler = (data: unknown) => { + received.push(data); + }; + assert.equal(await cq.subscribe('Events', handler), undefined); + assert.equal(start.mock.callCount(), 0); + assert.deepEqual(host.subscriptionHandlers, [handler]); + host.subscription.emit( + 'message', + `${host.options.prefix}:Events`, + '{"ok":true}', + ); + assert.deepEqual(received, [{ ok: true }]); + await cq.unsubscribe(); + assert.deepEqual(host.subscriptionHandlers, []); + await cq.destroy(); + }); + + for (const stalled of [false, true]) { + it(`installs handlers independently of a ${stalled ? 'stalled' : 'failed'} joining start`, async () => { + const cq = clusterOf(); + const gate = deferred(); + mock.method( + RedisQueue.prototype, + 'start', + async function (this: RedisQueue) { + if (stalled) { + await gate.promise; + return this; + } + throw new Error('unavailable'); + }, + ); + await cq.start(); + await cq.subscribe('Events', first); + const host = cq.addServer({ host: '127.0.0.1', port: 6604 }).imq; + try { + // A stall must not even delay the live registration. + await cq.subscribe('Events', second); + assert.deepEqual(host.subscriptionHandlers, [first, second]); + } finally { + gate.resolve(); + await settled(); + await cq.destroy(); + } + }); + } + + it('starts a joining host even when unsubscribe lands before catch-up', async () => { + const cq = clusterOf(); + const start = mock.method( + RedisQueue.prototype, + 'start', + async function (this: RedisQueue) { + return this; + }, + ); + await cq.start(); + await cq.subscribe('Events', first); + const host = cq.addServer({ host: '127.0.0.1', port: 6605 }).imq; + await cq.unsubscribe(); + await settled(); + assert.ok(start.mock.calls.some(call => call.this === host)); + await cq.destroy(); + }); + + it('does not start a joining host in a stopped cluster', async () => { + const cq = clusterOf(); + const start = mock.method( + RedisQueue.prototype, + 'start', + async function (this: RedisQueue) { + return this; + }, + ); + cq.addServer({ host: '127.0.0.1', port: 6606 }); + await settled(); + assert.equal(start.mock.callCount(), 0); + await cq.destroy(); + }); + + it('does not start a host removed before its lifecycle run', async () => { + const cq = clusterOf(); + const start = mock.method( + RedisQueue.prototype, + 'start', + async function (this: RedisQueue) { + return this; + }, + ); + await cq.start(); + const address = { host: '127.0.0.1', port: 6607 }; + cq.addServer(address); + cq.removeServer(address); + await settled(); + assert.equal(start.mock.callCount(), 0); + await cq.destroy(); + }); + + it('does not announce a removed host even with no subscriptions', async () => { + const cq = clusterOf(); + const initialized = mock.fn(); + cq.clusterEmitter.on('initialized', initialized); + const address = { host: '127.0.0.1', port: 6608 }; + cq.addServer(address); + cq.removeServer(address); + await settled(); + assert.equal(initialized.mock.callCount(), 0); + await cq.destroy(); + }); + + it('picks up a handler registered while a host is catching up', async () => { + const cq = clusterOf(); + await cq.subscribe('Events', first); + const host = hostOf(cq); + const gate = deferred(); + const entered = deferred(); + const original = host.subscribe; + mock.method( + host, + 'subscribe', + async function (this: any, channel: string, handler: typeof first) { + if (handler === first) { + entered.resolve(); + await gate.promise; + } + await original.call(this, channel, handler); + }, + ); + const catchUp = cq.syncHost(host); + await entered.promise; + const live = cq.subscribe('Events', second); + gate.resolve(); + await Promise.all([catchUp, live]); + assert.deepEqual(host.subscriptionHandlers, [first, second]); + await cq.destroy(); + }); + + it('repairs a rejected chain and retries only the missing suffix', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + await cq.subscribe('Events', first); + const boom = new Error('refused'); + const sub = mock.method(host, 'subscribe', async () => { + throw boom; + }); + await assert.rejects( + cq.subscribe('Events', second), + err => err === boom, + ); + sub.mock.restore(); + await cq.subscribe('Events', third); + assert.deepEqual(host.subscriptionHandlers, [first, second, third]); + await cq.destroy(); + }); + + it('rebuilds a known registration set after partial fan-out failure', async () => { + const cq = clusterOf(); + const healthy = hostOf(cq); + const failed = hostOf(cq); + const sub = mock.method(failed, 'subscribe', async () => { + throw new Error('refused'); + }); + await assert.rejects(cq.subscribe('Events', first), /refused/); + assert.deepEqual(healthy.subscriptionHandlers, [first]); + assert.deepEqual( + cq.state.handlers, + [first], + 'rejection does not roll back', + ); + sub.mock.restore(); + await cq.unsubscribe(); + await cq.subscribe('Events', first); + const joined = cq.addServer({ host: '127.0.0.1', port: 6609 }).imq; + await settled(); + for (const host of [healthy, failed, joined]) { + assert.deepEqual(host.subscriptionHandlers, [first]); + } + await cq.destroy(); + }); + + it('stops installing on a removed host while the channel stays unchanged', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const gate = deferred(); + const entered = deferred(); + mock.method( + host, + 'subscribe', + async (_channel: string, handler: typeof first) => { + if (handler === first) { + entered.resolve(); + await gate.promise; + } + host.subscriptionHandlers.push(handler); + }, + ); + cq.state.channel = 'Events'; + cq.state.handlers = [first, second]; + const run = cq.syncHost(host); + await entered.promise; + cq.imqs = []; + gate.resolve(); + await run; + assert.deepEqual(host.subscriptionHandlers, [first]); + await cq.destroy(); + }); + + it('stops an in-flight run when replacement handlers use a different channel', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const gate = deferred(); + const entered = deferred(); + const channels: string[] = []; + const original = host.subscribe; + mock.method( + host, + 'subscribe', + async function (this: any, channel: string, handler: typeof first) { + channels.push(channel); + if (handler === first) { + entered.resolve(); + await gate.promise; + } + await original.call(this, channel, handler); + }, + ); + const old = cq.subscribe('Old', first); + await entered.promise; + const clear = cq.unsubscribe(); + const newFirst = cq.subscribe('New', second); + const newSecond = cq.subscribe('New', third); + gate.resolve(); + await Promise.all([old, clear, newFirst, newSecond]); + assert.deepEqual(channels, ['Old', 'New', 'New']); + assert.deepEqual(host.subscriptionHandlers, [second, third]); + await cq.destroy(); + }); + + for (const inFlight of [false, true]) { + it(`keeps replacement handlers after teardown with an ${inFlight ? 'in-flight' : 'unstarted'} run`, async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const gate = deferred(); + const entered = deferred(); + const original = host.subscribe; + mock.method( + host, + 'subscribe', + async function ( + this: any, + channel: string, + handler: typeof first, + ) { + if (handler === first) { + entered.resolve(); + await gate.promise; + } + await original.call(this, channel, handler); + }, + ); + const old = cq.subscribe('Events', first); + if (inFlight) { + await entered.promise; + } + const clear = cq.unsubscribe(); + const replacement = cq.subscribe('Events', second); + gate.resolve(); + await Promise.all([old, clear, replacement]); + // A temporary installation ahead of teardown is harmless: progress is + // reset by teardown, so the replacement is restored exactly once. + assert.deepEqual(host.subscriptionHandlers, [second]); + await cq.destroy(); + }); + } + + it('resets progress after a teardown that clears handlers then rejects', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + await cq.subscribe('Events', first); + const unsub = mock.method(host, 'unsubscribe', async () => { + host.subscriptionHandlers = []; + throw new Error('teardown refused'); + }); + const clear = assert.rejects(cq.unsubscribe(), /teardown refused/); + const replacement = cq.subscribe('Events', second); + await Promise.all([clear, replacement]); + assert.deepEqual(host.subscriptionHandlers, [second]); + unsub.mock.restore(); + await cq.destroy(); + }); + + it('leaves the host unsubscribed when teardown lands mid-install', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const gate = deferred(); + const entered = deferred(); + mock.method( + host, + 'subscribe', + async (_channel: string, handler: typeof first) => { + entered.resolve(); + await gate.promise; + host.subscriptionHandlers.push(handler); + }, + ); + const install = cq.subscribe('Events', first); + await entered.promise; + let cleared = false; + const clear = cq.unsubscribe().then(() => { + cleared = true; + }); + await settled(); + assert.equal(cleared, false, 'teardown waits for the host chain'); + gate.resolve(); + await Promise.all([install, clear]); + assert.deepEqual(host.subscriptionHandlers, []); + await cq.destroy(); + }); + + it('does nothing when catch-up executes without a remembered channel', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const subscribe = mock.method(host, 'subscribe'); + // Exercise the empty-channel boundary independently of an empty list. + cq.state.handlers = [first]; + await cq.syncHost(host); + assert.equal(subscribe.mock.callCount(), 0); + await cq.destroy(); + }); + + it('keeps healthy hosts progressing while another subscription is stalled', async () => { + const cq = clusterOf(); + const stalled = hostOf(cq); + const healthy = hostOf(cq); + const gate = deferred(); + mock.method(stalled, 'subscribe', () => gate.promise); + const registration = cq.subscribe('Events', first); + try { + await settled(); + assert.deepEqual(healthy.subscriptionHandlers, [first]); + } finally { + gate.resolve(); + await registration; + await cq.destroy(); + } + }); + + it('reports registration order and successful installs before a partial failure', async () => { + const cq = clusterOf(); + const cap = capturing(); + cq.logger = cap.logger; + await cq.subscribe('Events', first); + await cq.subscribe('Events', second); + const host = hostOf(cq); + const original = host.subscribe; + mock.method( + host, + 'subscribe', + async function (this: any, channel: string, handler: typeof first) { + if (handler === second) throw new Error('refused'); + await original.call(this, channel, handler); + }, + ); + await assert.rejects(cq.syncHost(host), /refused/); + assert.equal( + matching(cap.info, /registered handler #1 for channel Events/) + .length, + 1, + ); + assert.equal( + matching(cap.info, /registered handler #2 for channel Events/) + .length, + 1, + ); + assert.equal( + matching(cap.info, /installed 1 handler\(s\) for channel Events/) + .length, + 1, + ); + assert.equal( + matching(cap.error, /some handlers remain uninstalled/).length, + 1, + ); + await cq.destroy(); + }); + + it('drops membership synchronously so queued work cannot reopen destroyed hosts', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const gate = deferred(); + const blocked = cq.enqueue(host, () => gate.promise); + const queued = cq.subscribe('Events', first); + await cq.destroy(); + + // a destroyed cluster refuses new registrations rather than resolving + // with nowhere to put them and repopulating the state destroy() cleared + await assert.rejects(() => cq.subscribe('Events', second), TypeError); + + gate.resolve(); + await Promise.all([blocked, queued]); + assert.deepEqual(host.subscriptionHandlers, []); + assert.equal(cq.imqLength, 0); + assert.equal(cq.state.channel, null); + assert.deepEqual(cq.state.handlers, []); + }); + + it('closes discovery admission synchronously and keeps it closed after destroy', async () => { + const cq = clusterOf(); + const address = { host: '127.0.0.1', port: 6612 }; + const host = cq.addServerWithQueueInitializing(address, false).imq; + const gate = deferred(); + mock.method(host, 'destroy', () => gate.promise); + const closing = cq.destroy(); + try { + assert.equal( + cq.addServer({ host: '127.0.0.1', port: 6613 }).imq, + undefined, + ); + assert.equal(cq.findServer(address), undefined); + assert.equal(cq.imqLength, 0); + } finally { + gate.resolve(); + await closing; + } + assert.equal(cq.addServer(address).imq, undefined); + assert.equal(cq.imqs.length, 0); + await cq.destroy(); + }); + + it('attempts every cleanup despite failures and retries only failed tasks', async () => { + const cq = clusterOf(); + const healthy = hostOf(cq); + const failed = hostOf(cq); + const hostError = new Error('host refused'); + const managerError = new Error('manager refused'); + const healthyDestroy = mock.method(healthy, 'destroy'); + const failedDestroy = mock.method(failed, 'destroy', async () => { + throw hostError; + }); + const failedCluster = {}; + const healthyCluster = {}; + const remove = mock.fn((cluster: object) => { + if (cluster === failedCluster) throw managerError; + return Promise.resolve(); + }); + const otherRemove = mock.fn(async () => undefined); + cq.options.clusterManagers = [{ remove }, { remove: otherRemove }]; + cq.initializedClusters = [failedCluster, healthyCluster]; + const one = cq.destroy(); + const two = cq.destroy(); + const results = await Promise.allSettled([one, two]); + assert.ok(results[0].status === 'rejected'); + assert.ok(results[1].status === 'rejected'); + assert.equal(results[0].reason, results[1].reason); + assert.deepEqual(results[0].reason.errors, [hostError, managerError]); + assert.equal(healthyDestroy.mock.callCount(), 1); + assert.equal(failedDestroy.mock.callCount(), 1); + assert.equal(remove.mock.callCount(), 2); + assert.equal(otherRemove.mock.callCount(), 2); + assert.equal( + cq.addServer({ host: '127.0.0.1', port: 6614 }).imq, + undefined, + ); + failedDestroy.mock.mockImplementation(async () => undefined); + remove.mock.mockImplementation(async () => undefined); + await cq.destroy(); + assert.equal(healthyDestroy.mock.callCount(), 1); + assert.equal(failedDestroy.mock.callCount(), 2); + assert.equal(remove.mock.callCount(), 3); + assert.equal(remove.mock.calls[2].arguments[0], failedCluster); + assert.equal(otherRemove.mock.callCount(), 2); + await cq.destroy(); + assert.equal(failedDestroy.mock.callCount(), 2); + assert.equal(remove.mock.callCount(), 3); + }); + + it('concurrent destroy callers await host teardown and manager removal', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const hostsDone = deferred(); + const managersDone = deferred(); + const enteredManager = deferred(); + const destroy = mock.method(host, 'destroy', () => hostsDone.promise); + const remove = mock.fn(async () => { + enteredManager.resolve(); + await managersDone.promise; + }); + cq.options.clusterManagers = [{ remove }]; + cq.initializedClusters = [{}]; + const one = cq.destroy(); + let finished = false; + const two = cq.destroy().then(() => { + finished = true; + }); + await settled(); + assert.equal(finished, false); + await enteredManager.promise; + hostsDone.resolve(); + await settled(); + assert.equal(finished, false); + managersDone.resolve(); + await Promise.all([one, two]); + assert.equal(destroy.mock.callCount(), 1); + assert.equal(remove.mock.callCount(), 1); + }); + + it('concurrent destroy callers receive the teardown failure', async () => { + const cq = clusterOf(); + const host = hostOf(cq); + const gate = deferred(); + mock.method(host, 'destroy', () => gate.promise); + const one = assert.rejects(cq.destroy(), AggregateError); + const two = assert.rejects(cq.destroy(), AggregateError); + gate.reject(new Error('teardown refused')); + await Promise.all([one, two]); + }); +}); + describe('ClusteredRedisQueue.matchServers()', () => { it('should return sameAddress when no ids provided', () => { assert.equal( @@ -884,13 +1646,18 @@ describe('ClusteredRedisQueue joining-server failure logging', () => { cq.state.started = true; + const fake: any = { + redisKey: '127.0.0.1:9999', + start: () => Promise.reject(boom), + destroy: () => Promise.resolve(), + }; + + cq.imqs.push(fake); + let thrown: any; try { - await cq.initializeQueue({ - redisKey: '127.0.0.1:9999', - start: () => Promise.reject(boom), - }); + await cq.startHost(fake); } catch (err) { thrown = err; } @@ -915,18 +1682,22 @@ describe('ClusteredRedisQueue joining-server failure logging', () => { const boom = new Error('NOPERM no permissions'); cq.state.started = false; - cq.state.subscription = { - channel: 'FlowEvents', - handler: () => undefined, + cq.state.channel = 'FlowEvents'; + cq.state.handlers = [() => undefined]; + + const fake: any = { + redisKey: '127.0.0.1:9999', + subscriptionHandlers: [], + subscribe: () => Promise.reject(boom), + destroy: () => Promise.resolve(), }; + cq.imqs.push(fake); + let thrown: any; try { - await cq.initializeQueue({ - redisKey: '127.0.0.1:9999', - subscribe: () => Promise.reject(boom), - }); + await cq.syncHost(fake); } catch (err) { thrown = err; }