From 43a8d574582a396574fcab9f147a21457c1d9ee3 Mon Sep 17 00:00:00 2001 From: Andrii Glushko Date: Wed, 29 Jul 2026 17:34:30 +0200 Subject: [PATCH] single pub sub lock fix --- src/NoLock.ts | 6 +++++ src/PgIpLock.ts | 53 ++++++++++++++++++++++++++++++++++++++++++- src/PgPubSub.ts | 27 ++++++++++++++++++---- src/types/AnyLock.ts | 10 ++++++++ test/PgIpLock.spec.ts | 36 +++++++++++++++++++++++++++++ 5 files changed, 127 insertions(+), 5 deletions(-) diff --git a/src/NoLock.ts b/src/NoLock.ts index 67624b1..2111a91 100644 --- a/src/NoLock.ts +++ b/src/NoLock.ts @@ -37,6 +37,12 @@ export class NoLock implements AnyLock { return; } + // no lock is never taken over: acquire() always succeeds on the spot + // eslint-disable-next-line @typescript-eslint/no-unused-vars + public onAcquire(handler: (channel: string) => void): void { + return; + } + /** * Frees no resources: no lock holds none */ diff --git a/src/PgIpLock.ts b/src/PgIpLock.ts index b63b0ea..45b4045 100644 --- a/src/PgIpLock.ts +++ b/src/PgIpLock.ts @@ -102,6 +102,7 @@ export class PgIpLock implements AnyLock { private acquired = false; private lockedReported = false; private notifyHandler?: (message: Notification) => void; + private acquireHandler?: (channel: string) => void; private acquireTimer?: Timeout; /** @@ -146,7 +147,7 @@ export class PgIpLock implements AnyLock { // noinspection TypeScriptValidateTypes if (!this.acquireTimer) { this.acquireTimer = setInterval( - () => !this.acquired && this.acquire(), + () => this.retryAcquire(), this.options.acquireInterval, ); } @@ -178,6 +179,31 @@ export class PgIpLock implements AnyLock { this.options.pgClient.on('notification', this.notifyHandler); } + /** + * This would provide a late acquire handler which will be called once the + * lock is taken over by the retry timer, with the channel name bypassed + * to a given handler. + * + * Acquiring the lock is only ever half of the caller's job - the other + * half (issuing `LISTEN`, in the PgPubSub case) happens right after + * acquire() returns true. When the first attempt loses to another + * process, that caller is long gone by the time the timer wins the lock, + * so without this handler the second half never runs and the lock is held + * by a process that does nothing with it - forever, since a held lock + * keeps every other process away too. + * + * @param {(channel: string) => void} handler + */ + public onAcquire(handler: (channel: string) => void): void { + if (this.acquireHandler) { + throw new TypeError( + 'Acquire handler for IPC lock has been already set up!', + ); + } + + this.acquireHandler = handler; + } + /** * Acquires a lock on the current channel. Returns true on success, * false - otherwise @@ -229,6 +255,31 @@ export class PgIpLock implements AnyLock { return this.acquired; } + /** + * Timer-driven acquire attempt. Notifies the acquire handler on success, + * which is what makes a late takeover complete instead of leaving the + * lock held but unused. + * + * The notification lives here rather than inside acquire() on purpose: + * acquire() is also called directly by whoever set the handler up, and + * notifying from there would call that caller back into itself. + * + * @return {Promise} + */ + private async retryAcquire(): Promise { + if (this.acquired) { + return; + } + + try { + if (await this.acquire()) { + this.acquireHandler?.(this.publicChannel); + } + } catch (err) { + this.options.logger.error(err); + } + } + /** * Channel name without the internal lock prefix, for log messages * diff --git a/src/PgPubSub.ts b/src/PgPubSub.ts index be41fef..3879cc7 100644 --- a/src/PgPubSub.ts +++ b/src/PgPubSub.ts @@ -447,8 +447,7 @@ export class PgPubSub extends EventEmitter { */ public async listen(channel: string): Promise { if (this.options.executionLock) { - await this.pgClient.query(`LISTEN ${ident(channel)}`); - this.emit('listen', channel); + await this.subscribe(channel); return; } @@ -456,8 +455,7 @@ export class PgPubSub extends EventEmitter { const acquired = await lock.acquire(); if (acquired) { - await this.pgClient.query(`LISTEN ${ident(channel)}`); - this.emit('listen', channel); + await this.subscribe(channel); return; } @@ -837,6 +835,24 @@ export class PgPubSub extends EventEmitter { this.reListenChannels = channels; } + /** + * Issues the actual `LISTEN` on a channel this process is already + * entitled to listen, and announces it. + * + * Kept apart from listen() because a late lock takeover must not run + * listen() again: re-acquiring a lock we already hold fails the deadlock + * check (it sees our own live connection as the current holder) and would + * skip the subscription it was meant to complete. + * + * @access private + * @param {string} channel + * @return {Promise} + */ + private async subscribe(channel: string): Promise { + await this.pgClient.query(`LISTEN ${ident(channel)}`); + this.emit('listen', channel); + } + /** * Instantiates and returns process lock for a given channel or returns * existing one @@ -889,6 +905,9 @@ export class PgPubSub extends EventEmitter { if (!uniqueKey) { lock.onRelease(chan => this.listen(chan)); + lock.onAcquire(chan => + this.subscribe(chan).catch(err => this.logger.error(err)), + ); } return lock; diff --git a/src/types/AnyLock.ts b/src/types/AnyLock.ts index 3d5f566..b0a8601 100644 --- a/src/types/AnyLock.ts +++ b/src/types/AnyLock.ts @@ -61,4 +61,14 @@ export interface AnyLock { * @param {(channel: string) => void} handler */ onRelease(handler: (channel: string) => void): void; + + /** + * Implements late lock acquire handler upset. The handler is called when + * the lock gets taken over by a retry, long after the acquire() caller + * has given up - which makes it the only chance for that caller to finish + * whatever the lock was being acquired for. + * + * @param {(channel: string) => void} handler + */ + onAcquire(handler: (channel: string) => void): void; } diff --git a/test/PgIpLock.spec.ts b/test/PgIpLock.spec.ts index 486027d..5152bf2 100644 --- a/test/PgIpLock.spec.ts +++ b/test/PgIpLock.spec.ts @@ -163,6 +163,42 @@ describe('IPCLock', () => { assert.equal(spy.calledOnce, true); }); }); + describe('onAcquire()', () => { + it('should not allow set handler twice', () => { + lock.onAcquire(() => { + /**/ + }); + assert.throws(() => + lock.onAcquire(() => { + /**/ + }), + ); + }); + it('should call the handler when the retry timer takes over', async () => { + const spy = makeSpy(); + lock.onAcquire(spy); + await lock.init(); + await new Promise(res => setTimeout(res, ACQUIRE_INTERVAL * 2 + 5)); + + assert.equal(spy.called, true); + assert.deepEqual(spy.getCalls()[0].args, ['LockTest']); + }); + it('should not call the handler on a direct acquire', async () => { + const spy = makeSpy(); + lock.onAcquire(spy); + + assert.equal(await lock.acquire(), true); + assert.equal(spy.called, false); + }); + it('should call the handler only once per takeover', async () => { + const spy = makeSpy(); + lock.onAcquire(spy); + await lock.init(); + await new Promise(res => setTimeout(res, ACQUIRE_INTERVAL * 4 + 5)); + + assert.equal(spy.calledOnce, true); + }); + }); describe('Shutdown', () => { // signal handling is opt-in since 3.0.0 enableGracefulShutdown();