'use strict'; import RootProcessPool from '../root/rootProcessPool'; import log4js from 'log4js'; import should from 'should'; import sinon from 'sinon'; import * as helpers from '../../../../helpers/helpers'; import * as assert from '../../../../helpers/test/assert'; import ControlSignal from '../controlSignal'; import * as errors from '../../errors'; import AsyncProcessPool from './asyncProcessPool'; import RootProcess from '../root/rootProcess'; import RootProcessContext from '../root/rootProcessContext'; /** * @test {AsyncProcessPool} */ describe('AsyncProcessPool', () => { let options: AsyncProcessPool.Options = { processFailoverThrottleDelayInMs: undefined, dependencies: [] }; let sandbox = sinon.createSandbox(); let pool: RootProcessPool; let logger = log4js.getLogger('test'); before(() => { log4js.configure(helpers.assembleLog4jsConfig({ levels: { 'AsyncProcessPool': 'TRACE' } })); }); beforeEach(() => { pool = new RootProcessPool(ProcessMock, options); }); afterEach(async () => { await pool.stop(); sandbox.restore(); }); /** * @test {AsyncProcessPool#scheduleProcess} */ describe('scheduleProcess', () => { /** * @test {AsyncProcessPool#scheduleProcess} */ it('should schedule processes', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test1', {args: [process1]}); pool.scheduleProcess('test2', {args: [process2]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process2.start, 1); sinon.assert.callCount(process1.run, 1); sinon.assert.callCount(process2.run, 1); }); await helpers.delay(25); sinon.assert.callCount(process1.stop, 0); sinon.assert.callCount(process2.stop, 0); }); /** * @test {AsyncProcessPool#scheduleProcess} */ it('should not schedule same process again if it has already scheduled', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test', {args: [process1]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process1.run, 1); }); pool.scheduleProcess('test', {args: [process2]}); await helpers.delay(25); sinon.assert.callCount(process2.start, 0); sinon.assert.callCount(process2.run, 0); }); /** * @test {AsyncProcessPool#scheduleProcess} */ it('should schedule same process again if previous one has canceled', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test', {args: [process1]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process1.run, 1); }); pool.cancelProcess('test'); pool.scheduleProcess('test', {args: [process2]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.stop, 1); sinon.assert.callCount(process2.start, 1); sinon.assert.callCount(process2.run, 1); }); }); /** * @test {AsyncProcessPool#scheduleProcess} */ it('should throw error if stopped (by default)', async () => { let process1 = stubProcess(); pool.stop(); try { pool.scheduleProcess('test', {args: [process1]}); throw new Error('assert'); } catch (err) { logger.info(err); err.message.should.not.equal('assert'); } await helpers.delay(25); sinon.assert.notCalled(process1.start); sinon.assert.notCalled(process1.run); sinon.assert.notCalled(process1.stop); }); /** * @test {AsyncProcessPool#scheduleProcess} */ it('should not throw error if stopped if the throwIfStopped option is disabled', async () => { pool.stop(); pool.scheduleProcess('test', {args: [stubProcess()], throwIfStopped: false}); pool.getScheduledIds().should.deepEqual([]); }); /** * @test {AsyncProcessPool#scheduleProcess} */ it('should cancel process which is failing to be constructed', async () => { let localPool = new RootProcessPool(ProcessFailingToConstruct, {dependencies: []}); localPool.scheduleProcess('test', {args: []}); localPool.getScheduledIds().should.deepEqual([]); }); }); /** * @test {AsyncProcessPool#restartProcess} */ describe('restartProcess', () => { /** * @test {AsyncProcessPool#restartProcess} */ it('should gracefully restart process which is starting', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); pool.restartProcess('test'); await helpers.waitPass(() => { assert.callOrder([ {spy: process.start, call: 0}, {spy: process.stop, call: 0}, {spy: process.start, call: 1}, {spy: process.run, call: 0} ]); }); }); /** * @test {AsyncProcessPool#restartProcess} */ it('should gracefully restart process which is running', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1)); pool.restartProcess('test'); await helpers.waitPass(() => { assert.callOrder([ {spy: process.start, call: 0}, {spy: process.run, call: 0}, {spy: process.stop, call: 0}, {spy: process.start, call: 1}, {spy: process.run, call: 1} ]); }); }); /** * @test {AsyncProcessPool#restartProcess} */ it('should not restart process again which is already going to be restarted', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1)); pool.restartProcess('test'); pool.restartProcess('test'); await helpers.delay(25); sinon.assert.callCount(process.start, 2); }); /** * @test {AsyncProcessPool#restartProcess} */ it('should force restart if process scheduled to be failover', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); let endPromise = helpers.createHandlePromise(); process.run.returns(endPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1), 25, {ignoreSinonClock: true}); process.run.returnsArg(0); endPromise.reject(new Error('test')); await clock.tickAsync(1000 * 5); pool.restartProcess('test'); await helpers.waitPass(() => { sinon.assert.callCount(process.start, 2); sinon.assert.callCount(process.run, 2); sinon.assert.callCount(process.stop, 1); }, 25, {ignoreSinonClock: true}); await clock.tickAsync(1000 * 15); sinon.assert.callCount(process.start, 2); sinon.assert.callCount(process.run, 2); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool#restartProcess} */ it('should restart process by instance', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); let actualProcess = pool.getProcess('test'); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1)); pool.restartProcess(actualProcess); pool.restartProcess(actualProcess); await helpers.waitPass(() => { assert.callOrder([ {spy: process.start, call: 0}, {spy: process.run, call: 0}, {spy: process.stop, call: 0}, {spy: process.start, call: 1}, {spy: process.run, call: 1} ]); }); }); /** * @test {AsyncProcessPool#restartProcess} */ it('should not restart process if given process instance is not actual', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test', {args: [process1]}); let actualProcess1 = pool.getProcess('test'); pool.cancelProcess('test'); pool.scheduleProcess('test', {args: [process2]}); await helpers.waitPass(() => sinon.assert.callCount(process2.run, 1)); pool.restartProcess(actualProcess1); await helpers.delay(25); sinon.assert.notCalled(process2.stop); }); }); /** * @test {AsyncProcessPool#cancelProcess} */ describe('cancelProcess', () => { /** * @test {AsyncProcessPool#cancelProcess} */ it('should cancel not scheduled process', async () => { await pool.cancelProcess('test'); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should cancel starting process', async () => { let process = stubProcess(); process.start.callsFake(stopPromise => stopPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 0); await pool.cancelProcess('test'); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should wait till process stopped at start stage', async () => { let process = stubProcess(); let startPromise = helpers.createHandlePromise(); process.start.callsFake(() => startPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 0); let cancelPromise = pool.cancelProcess('test'); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 0); startPromise.resolve(); await cancelPromise; sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should cancel running process', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 0); await pool.cancelProcess('test'); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should wait till process stopped at run stage', async () => { let process = stubProcess(); let runPromise = helpers.createHandlePromise(); process.run.callsFake(() => runPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 0); let cancelPromise = pool.cancelProcess('test'); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 0); runPromise.resolve(); await cancelPromise; sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should cancel process scheduled to restart', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => { sinon.assert.callCount(process.run, 1); }); pool.restartProcess('test'); await pool.cancelProcess('test'); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should wait till process stopped at stop stage', async () => { let process = stubProcess(); let stopPromise = helpers.createHandlePromise(); process.stop.callsFake(() => stopPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1)); let cancelPromise = helpers.wrapHandlePromise(pool.cancelProcess('test')); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); cancelPromise.completed.should.be.false(); stopPromise.resolve(); await cancelPromise; }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should not cancel process until all usages are canceled', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test1', {args: [process1], usage: 'usage1'}); pool.scheduleProcess('test1', {args: [process1], usage: 'usage2'}); pool.scheduleProcess('test2', {args: [process2], usage: 'usage1'}); await pool.waitProcess('test1'); await pool.waitProcess('test2'); await pool.cancelProcess('test1'); await pool.cancelProcess('test1', {usage: 'usage1'}); sinon.assert.callCount(process1.stop, 0); await pool.cancelProcess('test1', {usage: 'usage2'}); sinon.assert.callCount(process1.stop, 1); }); /** * @test {AsyncProcessPool#cancelProcess} */ it('should cancel process by all usages with one corresponding option', async () => { let process1 = stubProcess(); pool.scheduleProcess('test1', {args: [process1], usage: 'usage1'}); pool.scheduleProcess('test1', {args: [process1], usage: 'usage2'}); await pool.waitProcess('test1'); await pool.cancelProcess('test1', {allUsages: true}); sinon.assert.callCount(process1.stop, 1); }); }); /** * @test {AsyncProcessPool#getScheduledIds} */ describe('getScheduledIds', () => { /** * @test {AsyncProcessPool#getScheduledIds} */ it('should return scheduled process IDs', () => { pool.scheduleProcess('test1', {args: [stubProcess()]}); pool.scheduleProcess('test2', {args: [stubProcess()]}); pool.scheduleProcess('test3', {args: [stubProcess()]}); pool.cancelProcess('test2'); pool.restartProcess('test3'); pool.getScheduledIds().should.deepEqual(['test1', 'test3']); }); }); /** * @test {AsyncProcessPool#hasScheduled} */ describe('hasScheduled', () => { /** * @test {AsyncProcessPool#hasScheduled} */ it('should return whether a process has been sheduled', () => { pool.scheduleProcess('test1', {args: [stubProcess()]}); pool.scheduleProcess('test2', {args: [stubProcess()]}); pool.scheduleProcess('test3', {args: [stubProcess()]}); pool.cancelProcess('test2'); pool.restartProcess('test3'); pool.hasScheduled('test1').should.be.true(); pool.hasScheduled('test2').should.be.false(); pool.hasScheduled('test3').should.be.true(); pool.hasScheduled('test4').should.be.false(); }); }); /** * @test {AsyncProcessPool#hasScheduledBy} */ describe('hasScheduledBy', () => { /** * @test {AsyncProcessPool#hasScheduledBy} */ it('should return whether process has scheduled by specific usage ID', async () => { let process1 = stubProcess(); pool.scheduleProcess('test1', {args: [process1], usage: 'usage1'}); pool.scheduleProcess('test1', {args: [process1], usage: 'usage2'}); pool.hasScheduledBy('wrong', 'usage1').should.be.false(); pool.hasScheduledBy('test1', 'usage1').should.be.true(); pool.hasScheduledBy('test1', 'usage2').should.be.true(); pool.hasScheduledBy('test1', 'usage3').should.be.false(); }); }); /** * @test {AsyncProcessPool#getScheduledBy} */ describe('getScheduledBy', () => { /** * @test {AsyncProcessPool#hasScheduledBy} */ it('should return process IDs scheduled by specific usage ID', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test1', {args: [process1], usage: 'usage1'}); pool.scheduleProcess('test1', {args: [process1], usage: 'usage2'}); pool.scheduleProcess('test2', {args: [process2], usage: 'usage2'}); pool.getScheduledBy('usage1').should.deepEqual(['test1']); pool.getScheduledBy('usage2').should.deepEqual(['test1', 'test2']); pool.getScheduledBy('usage3').should.deepEqual([]); }); }); /** * @test {AsyncProcessPool#waitProcess} */ describe('waitProcess', () => { /** * @test {AsyncProcessPool#waitProcess} */ it('should wait till process will be started', async () => { let process = stubProcess(); let startPromise = helpers.createHandlePromise(); process.start.returns(startPromise); pool.scheduleProcess('test', {args: [process]}); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); startPromise.resolve(); should((await waitPromise).process).equal(process); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should return undefined if not scheduled', async () => { should(await pool.waitProcess('test')).be.undefined(); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should throw if not scheduled if corresponding option enabled', async () => { await pool.waitProcess('test', {throwIfNotScheduled: true}).should.be.rejectedWith(Error); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should return undefined if canceled during waiting', async () => { let process = stubProcess(); process.start.returnsArg(0); pool.scheduleProcess('test', {args: [process]}); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); pool.cancelProcess('test'); should(await waitPromise).be.undefined(); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should throw error if canceled during waiting if corresponding option enabled', async () => { let process = stubProcess(); process.start.returnsArg(0); pool.scheduleProcess('test', {args: [process]}); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test', {throwIfNotScheduled: true})); await helpers.delay(25); waitPromise.completed.should.be.false(); pool.cancelProcess('test'); await waitPromise.should.be.rejectedWith(Error); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should not throw error if process canceled and scheduled again during waiting', async () => { let process1 = stubProcess(); process1.start.callsFake(stopPromise => stopPromise); pool.scheduleProcess('test', {args: [process1]}); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); let process2 = stubProcess(); pool.cancelProcess('test'); pool.scheduleProcess('test', {args: [process2]}); should((await waitPromise).process).equal(process2); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should wait till process will be failovered if current process is in stopped state after error', async () => { let process = stubProcess(); let startPromise = helpers.createHandlePromise(); process.start.returns(startPromise); pool.scheduleProcess('test', {args: [process]}); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); sandbox.stub(options, 'processFailoverThrottleDelayInMs').value(50); startPromise.reject(new Error('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); process.start.resolves(); should((await waitPromise).process).equal(process); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should not return started process which was scheduled to restart immediately right after start', async () => { let process1 = stubProcess(); let startPromise = helpers.createHandlePromise(); process1.start.returns(startPromise); pool.scheduleProcess('test', {args: [process1]}); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); let process2 = stubProcess(); startPromise.resolve(); pool.cancelProcess('test'); pool.scheduleProcess('test', {args: [process2]}); should((await waitPromise).process).equal(process2); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should return with existing process if it is already started successfully', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); await helpers.delay(25); sinon.assert.callCount(process.start, 1); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.true(); waitPromise.result.process.should.equal(process); }); /** * @test {AsyncProcessPool#waitProcess} */ it('should wait if current process is failed to start and waiting to be restarted', async () => { sandbox.stub(options, 'processFailoverThrottleDelayInMs').value(0); let process = stubProcess(); let stopPromise = helpers.createHandlePromise(); process.start .onCall(0).rejects(new Error('test')) .onCall(1).resolves(); process.stop.returns(stopPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.delay(25); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.stop, 1); let waitPromise = helpers.wrapHandlePromise(pool.waitProcess('test')); await helpers.delay(25); waitPromise.completed.should.be.false(); stopPromise.resolve(); await waitPromise; }); /** * @test {AsyncProcessPool#waitProcess} */ it('should wait for new process if current one is still actively running but scheduled to cancel', async () => { let process1 = stubProcess(); let process2 = stubProcess(); pool.scheduleProcess('test', {args: [process1]}); pool.cancelProcess('test'); pool.scheduleProcess('test', {args: [process2]}); (await pool.waitProcess('test')).process.should.equal(process2); }); /** * @test {ChildScheduler#waitProcess} */ it('should time out waiting for the process', async () => { let process = stubProcess(); process.start.returnsArg(0); pool.scheduleProcess('test', {args: [process]}); should(await pool.waitProcess('test', {timeoutInMs: 25})).be.undefined(); }); /** * @test {ChildScheduler#waitProcess} */ it('should throw TimeoutError on timeout if corresponding option enabled', async () => { let process = stubProcess(); process.start.returnsArg(0); pool.scheduleProcess('test', {args: [process]}); await pool.waitProcess('test', { timeoutInMs: 25, throwOnTimeout: true }).should.be.rejectedWith(errors.TimeoutError); }); /** * @test {ChildScheduler#waitProcess} */ it('should wait until given stop promise resolves', async () => { let process = stubProcess(); process.start.returnsArg(0); pool.scheduleProcess('test', {args: [process]}); should(await pool.waitProcess('test', { stopPromise: helpers.delay(25) })).be.undefined(); }); /** * @test {ChildScheduler#waitProcess} */ it('should reject with given stop promise error', async () => { let process = stubProcess(); process.start.returnsArg(0); pool.scheduleProcess('test', {args: [process]}); await pool.waitProcess('test', { stopPromise: Promise.reject(new Error('test')) }).should.be.rejectedWith('test'); }); }); /** * @test {AsyncProcessPool#stop} */ describe('stop', () => { /** * @test {AsyncProcessPool#stop} */ it('should cancel all scheduled processes and wait till they are stopped', async () => { let process1 = stubProcess(); let process2 = stubProcess(); let stopPromise = helpers.createHandlePromise(); process1.stop.returns(stopPromise); process2.stop.returns(stopPromise); pool.scheduleProcess('test1', {args: [process1]}); pool.scheduleProcess('test2', {args: [process2]}); let promise = helpers.wrapHandlePromise(pool.stop()); await helpers.delay(25); sinon.assert.callCount(process1.stop, 1); sinon.assert.callCount(process2.stop, 1); promise.completed.should.be.false(); stopPromise.resolve(); await promise; }); /** * @test {AsyncProcessPool#stop} */ it('should wait for all still running but already canceled processes stopped', async () => { let process1 = stubProcess(); let process2 = stubProcess(); let stopPromise = helpers.createHandlePromise(); process1.stop.returns(stopPromise); process2.stop.returns(stopPromise); pool.scheduleProcess('test1', {args: [process1]}); pool.scheduleProcess('test2', {args: [process2]}); pool.cancelProcess('test1'); pool.cancelProcess('test2'); let promise = helpers.wrapHandlePromise(pool.stop()); await helpers.delay(25); sinon.assert.callCount(process1.stop, 1); sinon.assert.callCount(process2.stop, 1); promise.completed.should.be.false(); stopPromise.resolve(); await promise; }); }); /** * @test {AsyncProcessPool} */ describe('common', () => { /** * @test {AsyncProcessPool} */ it('should not run process if it was canceled when it was starting', async () => { let process = stubProcess(); pool.scheduleProcess('test', {args: [process]}); pool.cancelProcess('test'); await helpers.waitPass(() => { sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 1); }); }); /** * @test {AsyncProcessPool} */ it('should failover process with a delay if it was stopped unexpectedly', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); let endPromise = helpers.createHandlePromise(); process.run.returns(endPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1), 25, {ignoreSinonClock: true}); process.run.returnsArg(0); endPromise.resolve(); await clock.tickAsync(1000 * 9); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); await clock.tickAsync(1000 * 2); sinon.assert.callCount(process.start, 2); sinon.assert.callCount(process.run, 2); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool} */ it('should stop waiting for failover after unexpected stop if canceled', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); let endPromise = helpers.createHandlePromise(); process.run.returns(endPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1), 25, {ignoreSinonClock: true}); process.run.returnsArg(0); endPromise.resolve(); await clock.tickAsync(1000 * 5); await pool.cancelProcess('test'); await clock.tickAsync(1000 * 15); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool} */ it('should failover process with a delay if start failed with an error', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); let startPromise = helpers.createHandlePromise(); process.start.returns(startPromise); pool.scheduleProcess('test', {args: [process]}); sinon.assert.callCount(process.start, 1); process.start.resolves(); startPromise.reject(new Error('test')); await clock.tickAsync(1000 * 9); sinon.assert.callCount(process.run, 0); sinon.assert.callCount(process.stop, 1); await clock.tickAsync(1000 * 2); sinon.assert.callCount(process.start, 2); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool} */ it('should failover process with a delay if run failed with an error', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); let endPromise = helpers.createHandlePromise(); process.run.returns(endPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1), 25, {ignoreSinonClock: true}); process.run.returnsArg(0); endPromise.reject(new Error('test')); await clock.tickAsync(1000 * 9); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); await clock.tickAsync(1000 * 2); sinon.assert.callCount(process.start, 2); sinon.assert.callCount(process.run, 2); sinon.assert.callCount(process.stop, 1); }); /** * @test {AsyncProcessPool} */ it('should stop waiting for failover after error if canceled', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); let endPromise = helpers.createHandlePromise(); process.run.returns(endPromise); pool.scheduleProcess('test', {args: [process]}); await helpers.waitPass(() => sinon.assert.callCount(process.run, 1), 25, {ignoreSinonClock: true}); process.run.returnsArg(0); endPromise.reject(new Error('test')); await clock.tickAsync(1000 * 5); await pool.cancelProcess('test'); await clock.tickAsync(1000 * 15); sinon.assert.callCount(process.start, 1); sinon.assert.callCount(process.run, 1); sinon.assert.callCount(process.stop, 1); }); }); /** * @test {ControlSignal} */ describe('ControlSignal', () => { /** * @test {ControlSignal} */ it('should cancel the process if it throws a cancel signal', async () => { let process1 = stubProcess(); process1.run.throws(new ControlSignal({action: 'cancel'})); pool.scheduleProcess('test1', {args: [process1]}); await helpers.waitTrue(() => !pool.hasScheduled('test1')); sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process1.run, 1); sinon.assert.callCount(process1.stop, 1); }); /** * @test {ControlSignal} */ it('should failover the process if it throws a failover signal', async () => { sandbox.stub(options, 'processFailoverThrottleDelayInMs').value(1000 * 30); let clock = sandbox.useFakeTimers({now: new Date(), shouldAdvanceTime: false}); let process1 = stubProcess(); process1.run .onCall(0).throws(new ControlSignal({action: 'failover'})) .onCall(1).returnsArg(0); pool.scheduleProcess('test1', {args: [process1]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process1.run, 1); sinon.assert.callCount(process1.stop, 1); pool.hasScheduled('test1').should.be.true(); }, 25, {ignoreSinonClock: true}); await clock.tickAsync(1000 * 29); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process1.start, 1); await clock.tickAsync(1000 * 2); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 2); sinon.assert.callCount(process1.run, 2); sinon.assert.callCount(process1.stop, 1); pool.hasScheduled('test1').should.be.true(); }, 25, {ignoreSinonClock: true}); }); /** * @test {ControlSignal} */ it('should not failover the process if it stops being scheduled till the failover', async () => { sandbox.stub(options, 'processFailoverThrottleDelayInMs').value(1000 * 60 * 60 * 24); let process1 = stubProcess(); process1.run.callsFake(async () => { pool.cancelProcess('test1'); throw new ControlSignal({action: 'failover'}); }); pool.scheduleProcess('test1', {args: [process1]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process1.run, 1); sinon.assert.callCount(process1.stop, 1); }); await helpers.delay(25); sinon.assert.callCount(process1.start, 1); pool.hasScheduled('test1').should.be.false(); }); /** * @test {ControlSignal} */ it('should restart the process if it throws a stop signal', async () => { sandbox.stub(options, 'processFailoverThrottleDelayInMs').value(1000 * 30); sandbox.useFakeTimers({now: new Date(), shouldAdvanceTime: false}); let process1 = stubProcess(); process1.run .onCall(0).throws(new ControlSignal({action: 'stop'})) .onCall(1).returnsArg(0); pool.scheduleProcess('test1', {args: [process1]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 2); sinon.assert.callCount(process1.run, 2); sinon.assert.callCount(process1.stop, 1); pool.hasScheduled('test1').should.be.true(); }, 25, {ignoreSinonClock: true}); }); /** * @test {ControlSignal} */ it('should not restart the process if it stops being scheduled till the restart', async () => { let process1 = stubProcess(); process1.run.callsFake(async () => { pool.cancelProcess('test1'); throw new ControlSignal({action: 'stop'}); }); pool.scheduleProcess('test1', {args: [process1]}); await helpers.waitPass(() => { sinon.assert.callCount(process1.start, 1); sinon.assert.callCount(process1.run, 1); sinon.assert.callCount(process1.stop, 1); }); await helpers.delay(25); sinon.assert.callCount(process1.start, 1); pool.hasScheduled('test1').should.be.false(); }); }); /** * @test {AsyncProcessPool} */ describe('AsyncProcess.inject', () => { /** * @test {AsyncProcessPool} */ it('should typize and inject process dependencies', async () => { let injectPool = new RootProcessPool(ProcessWithInject, { dependencies: [1, '2'] }); injectPool.scheduleProcess('test', {args: []}); let process = await injectPool.waitProcess('test'); process.dep1.should.equal(1); process.dep2.should.equal('2'); }); }); /** * @test {AsyncProcessPool} */ describe('failoverThrottleDelay', () => { /** * @test {AsyncProcessPool} */ describe('exponential', () => { /** * @test {AsyncProcessPool} */ it('should throttle failing process with increasing delay', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); process.run.rejects(new Error('test')); pool.scheduleProcess('test', { args: [process], failoverThrottleDelay: { mode: 'exponential', minDelayInMs: 1000, maxDelayInMs: 10000, resetDelayInMs: 5000 } }); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.run, 1); await clock.tickAsync(900); sinon.assert.callCount(process.run, 1); await clock.tickAsync(101); sinon.assert.callCount(process.run, 2); await clock.tickAsync(1900); sinon.assert.callCount(process.run, 2); await clock.tickAsync(101); sinon.assert.callCount(process.run, 3); await clock.tickAsync(3900); sinon.assert.callCount(process.run, 3); await clock.tickAsync(101); sinon.assert.callCount(process.run, 4); await clock.tickAsync(7900); sinon.assert.callCount(process.run, 4); await clock.tickAsync(101); sinon.assert.callCount(process.run, 5); await clock.tickAsync(9900); sinon.assert.callCount(process.run, 5); await clock.tickAsync(101); sinon.assert.callCount(process.run, 6); await clock.tickAsync(9900); sinon.assert.callCount(process.run, 6); await clock.tickAsync(101); sinon.assert.callCount(process.run, 7); }); /** * @test {AsyncProcessPool} */ it('should reset throttling delay if enough time passed since last throttling', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); process.run.rejects(new Error('test')); pool.scheduleProcess('test', { args: [process], failoverThrottleDelay: { mode: 'exponential', minDelayInMs: 1000, maxDelayInMs: 10000, resetDelayInMs: 5000 } }); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.run, 1); await clock.tickAsync(900); sinon.assert.callCount(process.run, 1); await clock.tickAsync(101); sinon.assert.callCount(process.run, 2); await clock.tickAsync(1900); sinon.assert.callCount(process.run, 2); await clock.tickAsync(101); sinon.assert.callCount(process.run, 3); let promise1 = helpers.createHandlePromise(); process.run.returns(promise1); await clock.tickAsync(3900); sinon.assert.callCount(process.run, 3); await clock.tickAsync(101); sinon.assert.callCount(process.run, 4); clock.tick(5100); process.run.rejects(new Error('test')); promise1.reject(new Error('test')); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.run, 4); await clock.tickAsync(900); sinon.assert.callCount(process.run, 4); await clock.tickAsync(101); sinon.assert.callCount(process.run, 5); await clock.tickAsync(1900); sinon.assert.callCount(process.run, 5); await clock.tickAsync(101); sinon.assert.callCount(process.run, 6); }); /** * @test {AsyncProcessPool} */ it('should not reset throttling if reconnecting process on same schedulement', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); process.run.rejects(new Error('test')); pool.scheduleProcess('test', { args: [process], failoverThrottleDelay: { mode: 'exponential', minDelayInMs: 1000, maxDelayInMs: 10000, resetDelayInMs: 5000 } }); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.run, 1); await clock.tickAsync(900); sinon.assert.callCount(process.run, 1); await clock.tickAsync(101); sinon.assert.callCount(process.run, 2); await clock.tickAsync(1900); sinon.assert.callCount(process.run, 2); await clock.tickAsync(101); sinon.assert.callCount(process.run, 3); await clock.tickAsync(3900); sinon.assert.callCount(process.run, 3); await clock.tickAsync(101); sinon.assert.callCount(process.run, 4); pool.restartProcess('test'); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.run, 5); await clock.tickAsync(7900); sinon.assert.callCount(process.run, 5); await clock.tickAsync(101); sinon.assert.callCount(process.run, 6); await clock.tickAsync(9900); sinon.assert.callCount(process.run, 6); await clock.tickAsync(101); sinon.assert.callCount(process.run, 7); await clock.tickAsync(9900); sinon.assert.callCount(process.run, 7); await clock.tickAsync(101); sinon.assert.callCount(process.run, 8); }); /** * @test {AsyncProcessPool} */ it('should start waiting for throttling delay to reset only after next failed start attempt', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); process.run.rejects(new Error('test')); pool.scheduleProcess('test', { args: [process], failoverThrottleDelay: { mode: 'exponential', minDelayInMs: 1000, maxDelayInMs: 10000, resetDelayInMs: 5000 } }); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.start, 1); await clock.tickAsync(900); sinon.assert.callCount(process.start, 1); await clock.tickAsync(101); sinon.assert.callCount(process.start, 2); await clock.tickAsync(1900); sinon.assert.callCount(process.start, 2); await clock.tickAsync(101); sinon.assert.callCount(process.start, 3); let promise1 = helpers.createHandlePromise(); process.start.returns(promise1); await clock.tickAsync(3900); sinon.assert.callCount(process.start, 3); await clock.tickAsync(101); sinon.assert.callCount(process.start, 4); clock.tick(5100); promise1.reject(new Error('test')); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.start, 4); await clock.tickAsync(7900); sinon.assert.callCount(process.start, 4); await clock.tickAsync(101); sinon.assert.callCount(process.start, 5); await clock.tickAsync(9900); sinon.assert.callCount(process.start, 5); await clock.tickAsync(101); sinon.assert.callCount(process.start, 6); }); /** * @test {AsyncProcessPool} */ it('should start waiting for throttling delay to reset only after next successful start attempt', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); process.run.rejects(new Error('test')); pool.scheduleProcess('test', { args: [process], failoverThrottleDelay: { mode: 'exponential', minDelayInMs: 1000, maxDelayInMs: 10000, resetDelayInMs: 5000 } }); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.start, 1); await clock.tickAsync(900); sinon.assert.callCount(process.start, 1); await clock.tickAsync(101); sinon.assert.callCount(process.start, 2); await clock.tickAsync(1900); sinon.assert.callCount(process.start, 2); await clock.tickAsync(101); sinon.assert.callCount(process.start, 3); let startPromise = helpers.createHandlePromise(); let runPromise = helpers.createHandlePromise(); process.start.returns(startPromise); process.run.returns(runPromise); await clock.tickAsync(3900); sinon.assert.callCount(process.start, 3); await clock.tickAsync(101); sinon.assert.callCount(process.start, 4); clock.tick(5100); startPromise.resolve(); runPromise.reject(new Error('test')); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.start, 4); await clock.tickAsync(7900); sinon.assert.callCount(process.start, 4); await clock.tickAsync(101); sinon.assert.callCount(process.start, 5); await clock.tickAsync(9900); sinon.assert.callCount(process.start, 5); await clock.tickAsync(101); sinon.assert.callCount(process.start, 6); }); /** * @test {AsyncProcessPool} */ it('should not reset throttling delay after failed start if reset delay is 0', async () => { let clock = sandbox.useFakeTimers(); let process = stubProcess(); process.start.rejects(new Error('test')); pool.scheduleProcess('test', { args: [process], failoverThrottleDelay: { mode: 'exponential', minDelayInMs: 1000, maxDelayInMs: 10000, resetDelayInMs: 0 } }); await helpers.delay(25, {ignoreSinonClock: true}); sinon.assert.callCount(process.start, 1); await clock.tickAsync(900); sinon.assert.callCount(process.start, 1); await clock.tickAsync(101); sinon.assert.callCount(process.start, 2); await clock.tickAsync(1900); sinon.assert.callCount(process.start, 2); await clock.tickAsync(101); sinon.assert.callCount(process.start, 3); await clock.tickAsync(3900); sinon.assert.callCount(process.start, 3); await clock.tickAsync(101); sinon.assert.callCount(process.start, 4); await clock.tickAsync(7900); sinon.assert.callCount(process.start, 4); await clock.tickAsync(101); sinon.assert.callCount(process.start, 5); await clock.tickAsync(9900); sinon.assert.callCount(process.start, 5); await clock.tickAsync(101); sinon.assert.callCount(process.start, 6); }); }); }); /** * Stubs a process * @returns stubbed async process */ function stubProcess(): sinon.SinonStubbedInstance { let result = sandbox.createStubInstance(ProcessMock); result.start.resolves(); result.run.callsFake(stopPromise => stopPromise); result.stop.resolves(); return result; } }); /** * This mock will just return given process when constructing to simplify testing */ class ProcessMock extends RootProcess { public process: RootProcess; constructor(context: RootProcessContext) { super(context); } initialize(process: RootProcess): void { this.process = process; } async start(stopPromise: helpers.HandlePromise): Promise { return this.process.start(stopPromise); } async run(stopPromise: helpers.HandlePromise): Promise { return this.process.run(stopPromise); } async stop(): Promise { return this.process.stop(); } } class ProcessFailingToConstruct extends RootProcess { constructor(context: RootProcessContext) { super(context); throw new Error('test'); } async start(stopPromise: helpers.HandlePromise): Promise {} async run(stopPromise: helpers.HandlePromise): Promise {} async stop(): Promise {} } class ProcessWithInject extends RootProcess { public dep1: number; public dep2?: string; inject(dep1: number, dep2?: string): void { this.dep1 = dep1; this.dep2 = dep2; } async start(stopPromise: helpers.HandlePromise): Promise {} async run(stopPromise: helpers.HandlePromise): Promise { return stopPromise; } async stop(): Promise {} }