From: Justin Wind Date: Mon, 16 Aug 2021 18:38:48 +0000 (-0700) Subject: ignore invocation of worker process if already running X-Git-Tag: v1.1.4^2~4 X-Git-Url: https://git.squeep.com/?a=commitdiff_plain;h=fe3a4d96d559edcc30c2047109cd7b027e6258eb;p=websub-hub ignore invocation of worker process if already running also catch any errors thrown when invoking by admin api --- diff --git a/CHANGELOG.md b/CHANGELOG.md index 0378b17..05ed31d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ Releases and notable changes to this project are documented here. ## [Unreleased] +### Fixed + +- Prevent task processor from being re-invoked if it is already running. + ## [v1.1.3] - 2021-08-13 ### Fixed diff --git a/src/manager.js b/src/manager.js index d7a07f1..fddc8e4 100644 --- a/src/manager.js +++ b/src/manager.js @@ -666,7 +666,9 @@ class Manager { this.logger.debug(_scope, 'called', { ctx }); // N.B. no await on this - this.communication.worker.process(); + this.communication.worker.process().catch((e) => { + this.logger.error(_scope, 'failed', { error: e, ctx }); + }); res.end(); this.logger.info(_scope, 'invoked worker process', { ctx }); diff --git a/src/worker.js b/src/worker.js index 9b6fd7a..4f566ad 100644 --- a/src/worker.js +++ b/src/worker.js @@ -42,6 +42,7 @@ class Worker { this.inFlight = []; // Our work heap of Promises this.nextTimeout = undefined; // Allow clearTimeout() to reset waiting period. this.running = false; + this.isProcessing = false; // Only let one process() method execute on the work heap at a time } /** @@ -177,7 +178,13 @@ class Worker { async process() { const _scope = _fileScope('process'); - this.logger.debug(_scope, 'called', {}); + this.logger.debug(_scope, 'called', { isProcessing: this.isProcessing }); + + + if (this.isProcessing) { + return; + } + this.isProcessing = true; // Interrupt any pending sleep, if we were called out of timeout-cycle. clearTimeout(this.nextTimeout); @@ -218,6 +225,8 @@ class Worker { // No more work, wait a while and retry this._recurr(); + + this.isProcessing = false; } } diff --git a/test/src/manager.js b/test/src/manager.js index d8c0921..7870a55 100644 --- a/test/src/manager.js +++ b/test/src/manager.js @@ -609,7 +609,13 @@ describe('Manager', function () { describe('processTasks', function () { it('covers', async function () { - sinon.stub(manager.communication.worker, 'process'); + sinon.stub(manager.communication.worker, 'process').resolves(); + await manager.processTasks(res, ctx); + assert(manager.communication.worker.process.called); + assert(res.end.called); + }); + it('covers error', async function () { + sinon.stub(manager.communication.worker, 'process').rejects(); await manager.processTasks(res, ctx); assert(manager.communication.worker.process.called); assert(res.end.called); diff --git a/test/src/worker.js b/test/src/worker.js index 32bd065..5c97c0b 100644 --- a/test/src/worker.js +++ b/test/src/worker.js @@ -204,6 +204,18 @@ describe('Worker', function () { assert.strictEqual(worker._getWork.callCount, 0); assert.strictEqual(worker._recurr.callCount, 1); }); + it('covers double invocation', async function () { + const snooze = async (ms) => new Promise((resolve) => setTimeout(resolve, ms)); + + this.slow(300); + worker.inFlight = [ + Worker.watchedPromise(snooze(100)), + ]; + + await Promise.all([worker.process(), worker.process()]); + assert.strictEqual(worker._getWork.callCount, 2); + assert.strictEqual(worker._recurr.callCount, 1); + }); }); // process }); // Worker