From 9f5a2a9304b53ac8055d410243ab1e363b438c59 Mon Sep 17 00:00:00 2001 From: cool_machine Date: Thu, 17 Sep 2026 12:55:47 +0200 Subject: [PATCH 1/2] fix(cluster): retry a joining host whose subscription catch-up failed A host whose first subscribe() rejected was left without subscription handlers for the life of the process, while every probe it answered reported health. RedisQueue.subscribe() records a handler only after chan.subscribe() resolves, so a first failure leaves subscriptionHandlers empty. The connection layer then reconnects normally, but restoreSubscription() returns early on an empty list - and returns before re-subscribing - so the replacement socket is connected, ready, and subscribed to nothing. Nothing brought that host back: the joining catch-up rejection was swallowed and left progress.installed at 0, start() fans out startHost only, a re-announced address is recognised as a known server and skipped, and no event reports the failure. A service that subscribes once at start-up never registers again, so the message promising repair "until a later registration triggers another catch-up" never came true. send() is unaffected throughout, which is what makes it hard to spot. The join site now retries the subscription leg with capped exponential backoff, cancelled when the host leaves the cluster or the cluster is destroyed, on an unref()ed timer so a host that never returns cannot hold the process open. Only that leg is retried: a failed start() is already retryable through an explicit start(), and retrying it here would poll a connection object the reconnect path owns. Retrying is safe because catch-up is idempotent - syncHost() reads the cluster-owned installed count inside the per-host chain and installs only the missing suffix, so a retry cannot reinstall a handler that already landed. The backoff mirrors the connection layer's own policy, which this has to outlive. A host that becomes usable only through a retry is announced as initialized once its startup has also succeeded and while it is still a member. Sends parked in sendWhenInitialized() wait on that event, so without it they would time out against a cluster that had recovered. The end-to-end test for this found a second defect. connect() binds a connection before awaiting it and returns that same object to a concurrent caller, so a subscribe() racing a reconnect attached its handler to a connection whose restore had not yet run; restoreSubscription() then re-attached every remembered handler to the same connection and every message was delivered twice. The race predates this change - any live subscribe() during a reconnect could hit it - but the retry and the reconnect share a 1s base delay, which made it land in 3 of 30 runs. restoreSubscription() now reconciles the connection's message listeners to exactly the remembered handlers instead of appending to them. Both orders of the race end with one listener per handler, and a deliberate double registration of the same function still fires twice. syncHost()'s failure message is updated accordingly: a joining host now retries on its own, while a failure during a live registration is still repaired by the next one. Five unit tests cover the join-site wiring, the retried install, suffix-only installation, cancellation on removal, and the parked send; each was calibrated by mutation. One integration test drives a real broker behind a proxy that refuses the first connection and asserts exactly one delivery after recovery; it fails 3/3 on the unmodified source, delivered twice in 3/30 runs with the retry alone, and is green 20/20 with both changes. 418 unit tests pass. Closes #31 --- CHANGELOG.md | 36 +++ src/ClusteredRedisQueue.ts | 148 ++++++++++-- src/RedisQueue.ts | 7 + test/integration/clusterSubscription.spec.ts | 190 ++++++++++++++- test/unit/ClusteredRedisQueue.spec.ts | 236 +++++++++++++++++++ 5 files changed, 598 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f152a4..8aa1d62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -157,6 +157,42 @@ release. ### Fixed +- **A joining host whose first subscription attempt failed stayed silent for the + life of the process.** `RedisQueue.subscribe()` records a handler only after + the subscribe resolves, so a first failure left that host's handler list + empty — and the connection layer's own reconnect replays from that list, so it + restored a socket that was connected, `ready` and subscribed to nothing. + Nothing brought the host back: the joining catch-up's rejection was swallowed, + `start()` fans out startup only, a re-announced address is recognised as a + known server and skipped, and no event reported the failure. A service that + subscribes once at start-up never registers again, so the promise of repair + "on a later registration" never came due. `send()` was unaffected throughout, + which is what made it hard to see. + + The join now retries the subscription leg with capped exponential backoff, + cancelled when the host leaves the cluster or the cluster is destroyed, on an + unref'd timer so a host that never returns cannot hold the process open. Only + that leg is retried: a failed `start()` is already retryable through an + explicit `start()`, and retrying it here would poll a connection object the + reconnect path owns. Retrying is safe because catch-up reads the cluster-owned + installed count inside the per-host chain and installs only the missing + suffix, so it cannot reinstall a handler that already landed. + + A host that becomes usable only through a retry is announced as `initialized` + once its startup has also succeeded and while it is still a member. Sends + parked waiting for a usable server wait on that event, so without it they + would time out against a cluster that had recovered. + +- **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 + `subscribe()` racing a reconnect attached its handler to a connection whose + restore had not yet run; the restore then re-attached every remembered + handler, including that one. Reachable before this release by a live + `subscribe()` during a reconnect; the retry above made it likely, which is how + it was found. `restoreSubscription()` now reconciles the connection's listeners + to exactly the remembered handlers instead of appending to them. + - **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 diff --git a/src/ClusteredRedisQueue.ts b/src/ClusteredRedisQueue.ts index 152a3be..faf6c87 100644 --- a/src/ClusteredRedisQueue.ts +++ b/src/ClusteredRedisQueue.ts @@ -44,6 +44,17 @@ import { */ const SEND_INIT_TIMEOUT = +(process.env.IMQ_SEND_INIT_TIMEOUT || 0) || 30000; +/** + * Delay (ms) before the first retry of a joining host's subscription catch-up, + * doubling up to {@link SYNC_RETRY_MAX_DELAY}. Mirrors the connection layer's + * own reconnect policy, which this retry has to outlive: the socket may take + * several attempts to come back, and the handlers have to go on afterwards. + */ +const SYNC_RETRY_BASE_DELAY = 1000; + +/** Longest delay (ms) between subscription catch-up retries. */ +const SYNC_RETRY_MAX_DELAY = 30000; + /** * A server registered in a {@link ClusteredRedisQueue}: its address, plus the * {@link RedisQueue} instance serving that host. @@ -85,6 +96,10 @@ interface HostProgress { chain: Promise; /** Registrations successfully installed by this cluster since teardown. */ installed: number; + /** Pending catch-up retry, so teardown can cancel it. */ + retryTimer?: ReturnType; + /** Consecutive failed catch-up attempts, for the backoff. */ + retryAttempts: number; } /** @@ -518,6 +533,7 @@ export class ClusteredRedisQueue this.state.started = false; for (const imq of this.imqs) { + this.cancelSync(imq); this.cleanup.add(() => imq.destroy()); } @@ -1132,6 +1148,9 @@ export class ClusteredRedisQueue const imqToRemove = remove.imq; if (imqToRemove) { + // a retry scheduled for this host must not outlive it + this.cancelSync(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( @@ -1214,21 +1233,36 @@ export class ClusteredRedisQueue 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; - } + const started = this.startHost(imq); + const synced = this.syncHost(imq); + + // Sends park on this event, so a host that only becomes usable + // after a retry has to reach it too, or they wait out their whole + // budget against a cluster that recovered. A host dropped while it + // was being brought up to date never became a member, and + // announcing it would release those sends onto a queue that is + // being destroyed. Exactly one of the two paths below reaches this: + // a join whose legs both succeeded, or a retry repairing a catch-up + // whose failure already rejected that join + const announce = (): void => { + if (!this.imqs.includes(imq)) { + return; + } - this.clusterEmitter.emit('initialized', { - server: newServer, - imq, - }); - }, + this.clusterEmitter.emit('initialized', { + server: newServer, + imq, + }); + }; + + // only the subscription leg is retried. A failed start is already + // retryable through an explicit start(), and retrying it here would + // poll a connection object the reconnect path owns; a failed + // catch-up has no other route back + synced.catch(() => this.scheduleSync(imq, started, announce)); + + Promise.all([started, synced]).then( + announce, // 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 @@ -1297,6 +1331,7 @@ export class ClusteredRedisQueue progress = { chain: Promise.resolve(), installed: 0, + retryAttempts: 0, }; this.progress.set(imq, progress); @@ -1356,6 +1391,88 @@ export class ClusteredRedisQueue return run; } + /** + * Retries a joining 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 + * + * @remarks + * A joining 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 — + * `start()` fans out startup only, a re-announced address is recognised as a + * known server and skipped, and a service that subscribes once at boot never + * registers again. The host stays a silent member for the life of the + * process while every probe it answers reports health. + * + * Retrying is safe because catch-up is idempotent: {@link + * ClusteredRedisQueue.syncHost} reads the cluster-owned installed count + * inside the per-host chain and installs only the missing suffix, so a retry + * cannot duplicate a handler that did land. The backoff mirrors the + * connection layer's, which this has to outlive. + */ + private scheduleSync( + imq: RedisQueue, + started: Promise, + announce: () => void, + ): void { + if (this.closed || !this.imqs.includes(imq)) { + return; + } + + const progress = this.progressOf(imq); + + if (progress.retryTimer) { + return; + } + + const attempts = progress.retryAttempts + 1; + const delay = Math.min( + SYNC_RETRY_MAX_DELAY, + SYNC_RETRY_BASE_DELAY * 2 ** (attempts - 1), + ); + + progress.retryAttempts = attempts; + + const timer = setTimeout(() => { + progress.retryTimer = undefined; + + if (this.closed || !this.imqs.includes(imq)) { + return; + } + + this.syncHost(imq).then( + () => { + progress.retryAttempts = 0; + + // the host is only usable once its lifecycle came up too + started.then(announce, () => undefined); + }, + () => this.scheduleSync(imq, started, announce), + ); + }, delay); + + // a host that never comes back must not hold the process open + timer.unref?.(); + progress.retryTimer = timer; + } + + /** + * Cancels a pending catch-up retry for a host being torn down. + * + * @param imq - the queue leaving the cluster + */ + private cancelSync(imq: RedisQueue): void { + const progress = this.progress.get(imq); + + if (progress?.retryTimer) { + clearTimeout(progress.retryTimer); + progress.retryTimer = undefined; + } + } + /** * Installs registrations this cluster has not yet installed on a host. * @@ -1412,7 +1529,8 @@ export class ClusteredRedisQueue 'error', `server ${imq.redisKey} failed to subscribe to channel ` + `${channel}, code ${errorCode(err)}: some handlers remain ` + - 'uninstalled until a later registration triggers another catch-up', + 'uninstalled. A joining host retries this on its own; a ' + + 'failure during a live registration is repaired by the next one', ); throw err; diff --git a/src/RedisQueue.ts b/src/RedisQueue.ts index d58f3e1..31a667a 100644 --- a/src/RedisQueue.ts +++ b/src/RedisQueue.ts @@ -708,6 +708,13 @@ export class RedisQueue await chan.subscribe(fcn); + // Reconcile rather than append. connect() binds a connection before it + // awaits it and hands that same object to any concurrent caller, so a + // subscribe() racing a reconnect can attach a handler to this very + // connection before this runs. Appending there would leave the socket + // carrying the handler twice and deliver every message twice. + chan.removeAllListeners('message'); + for (const handler of this.subscriptionHandlers) { this.attachSubscriptionHandler(chan, handler); } diff --git a/test/integration/clusterSubscription.spec.ts b/test/integration/clusterSubscription.spec.ts index 342645a..2f6ef70 100644 --- a/test/integration/clusterSubscription.spec.ts +++ b/test/integration/clusterSubscription.spec.ts @@ -33,6 +33,12 @@ import assert from 'node:assert/strict'; import { randomUUID as uuid } from 'node:crypto'; import { once } from 'node:events'; +import { + createConnection, + createServer, + type Server, + type Socket, +} from 'node:net'; import { after, describe, it, mock } from 'node:test'; import { Redis } from 'ioredis'; import { ClusteredRedisQueue, RedisQueue } from '../../src/index.js'; @@ -119,7 +125,11 @@ const settle = async (received: unknown[], count: number): Promise => { }; /** Bounds a test gate without leaving a timer behind on success or failure. */ -const bounded = async (promise: Promise, label: string): Promise => { +const bounded = async ( + promise: Promise, + label: string, + timeoutMs: number = 5000, +): Promise => { let timer: NodeJS.Timeout | undefined; try { return await Promise.race([ @@ -127,7 +137,7 @@ const bounded = async (promise: Promise, label: string): Promise => { new Promise((_resolve, reject) => { timer = setTimeout( () => reject(new Error(`timed out: ${label}`)), - 5000, + timeoutMs, ); }), ]); @@ -136,15 +146,98 @@ const bounded = async (promise: Promise, label: string): Promise => { } }; +/** + * Takes an ephemeral port out of circulation, then releases it. Connecting to + * the returned loopback port before another listener is started gets the real + * TCP ECONNREFUSED that a broker which is not up yet would produce. + */ +const closedPort = async (): Promise => { + const reservation = createServer(); + + await once(reservation.listen({ host: '127.0.0.1', port: 0 }), 'listening'); + + const address = reservation.address(); + + if (!address || typeof address === 'string') { + throw new Error('could not reserve an IPv4 test port'); + } + + await new Promise((resolve, reject) => { + reservation.close(error => (error ? reject(error) : resolve())); + }); + + return address.port; +}; + +/** + * A deliberately transparent TCP bridge. Redis protocol bytes are neither + * parsed nor fabricated: after start(), the joining host talks to the broker + * through normal TCP sockets. + */ +class RedisProxy { + private server: Server | undefined; + private readonly sockets = new Set(); + + public constructor(private readonly port: number) {} + + public async start(): Promise { + if (this.server) { + return; + } + + const server = createServer(client => { + const upstream = createConnection({ host: HOST, port: PORT }); + const closeBoth = (): void => { + client.destroy(); + upstream.destroy(); + }; + + this.sockets.add(client); + this.sockets.add(upstream); + client.once('close', () => this.sockets.delete(client)); + upstream.once('close', () => this.sockets.delete(upstream)); + client.on('error', closeBoth); + upstream.on('error', closeBoth); + client.pipe(upstream).pipe(client); + }); + + // Keep a durable error listener after the listen() await has settled. + server.on('error', quiet.error); + await once( + server.listen({ host: '127.0.0.1', port: this.port }), + 'listening', + ); + this.server = server; + } + + public async close(): Promise { + for (const socket of this.sockets) { + socket.destroy(); + } + + this.sockets.clear(); + + if (!this.server?.listening) { + return; + } + + const server = this.server; + this.server = undefined; + await new Promise((resolve, reject) => { + server.close(error => (error ? reject(error) : resolve())); + }); + } +} + describe('ClusteredRedisQueue subscription over a real broker', () => { const queues: ClusteredRedisQueue[] = []; - const cluster = (name: string): 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 queue = new ClusteredRedisQueue(name, { cluster: [], - logger: quiet, + logger, }); queues.push(queue); @@ -381,4 +474,93 @@ describe('ClusteredRedisQueue subscription over a real broker', () => { assert.equal(late.length, 1, 'handler from the live path'); }, ); + + it( + 'repairs a join whose first real subscription connection is refused', + { skip }, + async () => { + const channel = `refused-${uuid()}`; + const port = await closedPort(); + const proxy = new RedisProxy(port); + const received: unknown[] = []; + const refused = Promise.withResolvers(); + const logger = { + ...quiet, + error(...args: unknown[]) { + const error = args.at(-1) as + | (Error & { code?: string }) + | undefined; + + if (error?.code === 'ECONNREFUSED') { + refused.resolve(); + } + }, + }; + const queue = cluster(`refused-${uuid()}`, logger); + let publisher: Redis | undefined; + + try { + // Register on the cluster before it has a host, so the new + // host's catch-up is the only attempt to install this handler. + await queue.subscribe(channel, data => received.push(data)); + + // This has no listener at this point. The logger gate observes + // the actual error event emitted by ioredis before the proxy is + // allowed to forward anything to Redis. + (queue as any).addServer({ host: '127.0.0.1', port }); + await bounded( + refused.promise, + 'the joining subscription to be refused', + ); + + const controller = new AbortController(); + const initialized = once( + (queue as any).clusterEmitter, + 'initialized', + { signal: controller.signal }, + ); + + try { + await proxy.start(); + // First reconnect and catch-up are each scheduled after + // one second. If their order needs a second catch-up, + // exponential backoff makes that three seconds total; + // ten seconds leaves ample scheduler and broker slack. + await bounded( + initialized, + 'the repaired host to initialize', + 10000, + ); + } finally { + controller.abort(); + } + + publisher = new Redis({ + host: HOST, + port: PORT, + lazyConnect: true, + retryStrategy: null, + }); + publisher.on('error', quiet.error); + await publisher.connect(); + assert.equal( + await publisher.publish( + `${(queue as any).options.prefix}:${channel}`, + JSON.stringify({ mark: channel }), + ), + 1, + 'the recovered Redis connection has one subscriber', + ); + await settle(received, 1); + assert.deepEqual(received, [{ mark: channel }]); + } finally { + // Stop producers first. Keep the bridge live while the queue + // closes its subscription, then tear down both ends of every + // bridged socket before closing the listening socket. + 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 cc0943c..394bab7 100644 --- a/test/unit/ClusteredRedisQueue.spec.ts +++ b/test/unit/ClusteredRedisQueue.spec.ts @@ -987,6 +987,242 @@ describe('ClusteredRedisQueue handler catch-up', () => { await cq.destroy(); }); + it('retries a joining host whose first catch-up failed', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + + const cq = clusterOf(); + await cq.subscribe('Events', first); + + // 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); + }); + + await assert.rejects(cq.syncHost(host), /refused/); + assert.deepEqual(host.subscriptionHandlers, []); + + cq.scheduleSync(host, Promise.resolve(), () => undefined); + refuse = false; + + t.mock.timers.tick(1000); + await settled(); + await settled(); + + assert.deepEqual( + host.subscriptionHandlers, + [first], + 'the missing registration should have been installed by the retry', + ); + + sub.mock.restore(); + t.mock.timers.reset(); + await cq.destroy(); + }); + + it('a joining host whose catch-up fails is retried by the join itself', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + + const cq: any = new ClusteredRedisQueue('JoinRetry', { + cluster: [], + logger, + }); + + 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); + }, + ); + + // the join path itself has to schedule the retry - nothing in the test + // touches scheduleSync, so removing that wiring must fail here + cq.addServer({ host: '127.0.0.1', port: 6379 }); + await settled(); + + const host = cq.imqs[0]; + + assert.deepEqual( + host.subscriptionHandlers, + [], + 'the first catch-up should have failed', + ); + + refuse = false; + t.mock.timers.tick(1000); + await settled(); + await settled(); + + assert.deepEqual( + host.subscriptionHandlers, + [first], + 'the join should have retried the failed catch-up', + ); + + sub.mock.restore(); + t.mock.timers.reset(); + await cq.destroy().catch(() => undefined); + }); + + it('releases a parked send once a retried host recovers', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + + const cq: any = new ClusteredRedisQueue('ParkedSend', { + cluster: [], + logger, + }); + + 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 send = mock.method( + RedisQueue.prototype, + 'send', + async () => 'sent', + ); + + // parked: the cluster is empty, so this waits for 'initialized' + const parked = cq.send('Somewhere', { a: 1 }); + + cq.addServer({ host: '127.0.0.1', port: 6379 }); + await settled(); + + // the join failed its catch-up, so nothing was announced yet + assert.equal(send.mock.callCount(), 0); + + refuse = false; + t.mock.timers.tick(1000); + await settled(); + await settled(); + + assert.equal( + await parked, + 'sent', + 'the retry must release a send parked on initialized', + ); + assert.equal( + send.mock.callCount(), + 1, + 'the parked send must be released exactly once', + ); + + sub.mock.restore(); + send.mock.restore(); + t.mock.timers.reset(); + await cq.destroy().catch(() => undefined); + }); + + it('retrying a catch-up installs only the missing suffix', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + + const cq = clusterOf(); + const host = hostOf(cq); + await cq.subscribe('Events', first); + 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); + }); + + await assert.rejects(cq.subscribe('Events', second), /refused/); + + cq.scheduleSync(host, Promise.resolve(), () => undefined); + refuse = false; + + t.mock.timers.tick(1000); + await settled(); + await settled(); + + assert.deepEqual( + host.subscriptionHandlers, + [first, second], + 'the retry must not reinstall a handler that already landed', + ); + + sub.mock.restore(); + t.mock.timers.reset(); + await cq.destroy(); + }); + + it('stops retrying a host that left the cluster', async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + + const cq = clusterOf(); + await cq.subscribe('Events', first); + + const host = hostOf(cq); + const sub = mock.method(host, 'subscribe', async () => { + throw new Error('refused'); + }); + + await assert.rejects(cq.syncHost(host), /refused/); + cq.scheduleSync(host, Promise.resolve(), () => undefined); + + // the host goes away before the retry is due + cq.imqs = cq.imqs.filter((each: any) => each !== host); + sub.mock.restore(); + + const sync = mock.method(cq, 'syncHost'); + + t.mock.timers.tick(60000); + await settled(); + + assert.equal( + sync.mock.callCount(), + 0, + 'a pending retry must not run catch-up for a host that left', + ); + assert.deepEqual( + host.subscriptionHandlers, + [], + 'a host that left must not be installed on by a pending retry', + ); + + sync.mock.restore(); + + t.mock.timers.reset(); + await cq.destroy(); + }); + it('rebuilds a known registration set after partial fan-out failure', async () => { const cq = clusterOf(); const healthy = hostOf(cq); From b52eacec24551040c9d2f99d1221b697703d2296 Mon Sep 17 00:00:00 2001 From: cool_machine Date: Thu, 17 Sep 2026 13:02:38 +0200 Subject: [PATCH 2/2] style(test): format the catch-up specs with oxfmt --- test/unit/ClusteredRedisQueue.spec.ts | 48 ++++++++++++++++----------- 1 file changed, 28 insertions(+), 20 deletions(-) diff --git a/test/unit/ClusteredRedisQueue.spec.ts b/test/unit/ClusteredRedisQueue.spec.ts index 394bab7..9bb14dc 100644 --- a/test/unit/ClusteredRedisQueue.spec.ts +++ b/test/unit/ClusteredRedisQueue.spec.ts @@ -997,17 +997,21 @@ describe('ClusteredRedisQueue handler catch-up', () => { // 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'); - } + 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); - }); + this.subscriptionHandlers.push(handler); + }, + ); await assert.rejects(cq.syncHost(host), /refused/); assert.deepEqual(host.subscriptionHandlers, []); @@ -1151,17 +1155,21 @@ describe('ClusteredRedisQueue handler catch-up', () => { // 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'); - } + 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); - }); + this.subscriptionHandlers.push(handler); + }, + ); await assert.rejects(cq.subscribe('Events', second), /refused/);