single pub sub lock fix

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
      */

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<void>}
+     */
+    private async retryAcquire(): Promise<void> {
+        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
      *

src/PgPubSub.ts

@@ -447,17 +447,15 @@ export class PgPubSub extends EventEmitter {
      */
     public async listen(channel: string): Promise<void> {
         if (this.options.executionLock) {
-            await this.pgClient.query(`LISTEN ${ident(channel)}`);
-            this.emit('listen', channel);
+            await this.subscribe(channel);
             return;
         }
 
         const lock = await this.lock(channel);
         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<void>}
+     */
+    private async subscribe(channel: string): Promise<void> {
+        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;

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;
 }

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();