diff --git a/src/bin/filler.ts b/src/bin/filler.ts index 0c6c4c9d..45c444e7 100644 --- a/src/bin/filler.ts +++ b/src/bin/filler.ts @@ -51,16 +51,38 @@ if (!readerConfigs || readerConfigs.length === 0) { // during AtomicAssetsHandler.init on a cold cache) - the reader never // started but Kubernetes had no crashed process to restart. Node's default // (crash) is what gets the pod restarted; matching it here is the fix. +// Published for the crash handlers below, which are registered before the worker +// branch builds its filler and so cannot reach a local reference. +let activeFiller: Filler | null = null; + +// How long a crashing worker waits for the ship close frame to leave. Short and +// unconditional: a crashed worker must not stay alive waiting on a socket. +const SHIP_CLOSE_GRACE_MS = 250; + +// process.exit() abandons the ship websocket rather than closing it, which +// leaves the node holding a half-open state_history session until it works out +// the peer is gone. Send the close frame first, then exit on a short timer so +// the frame has a chance to go out. +function exitAfterClosingShip(code: number): void { + try { + activeFiller?.reader.ship.stopProcessing(); + } catch (closeError) { + logger.error('Failed to close the ship socket before exit', closeError); + } + + setTimeout(() => process.exit(code), SHIP_CLOSE_GRACE_MS); +} + process.on('unhandledRejection', error => { logger.error('Unhandled Rejection', error); - process.exit(1); + exitAfterClosingShip(1); }); process.on('uncaughtException', error => { logger.error('Uncaught Exception', error); - process.exit(1); + exitAfterClosingShip(1); }); // @ts-ignore @@ -272,13 +294,12 @@ if (cluster.isPrimary || cluster.isMaster) { new AggregatorRegistry(); const index = parseInt(process.env.config_index, 10); - let filler: Filler | null = null; process.on('SIGTERM', async () => { logger.info(`Worker ${process.pid} received SIGTERM - stopping filler`); - if (filler) { - await filler.stopFiller(); + if (activeFiller) { + await activeFiller.stopFiller(); } process.exit(0); @@ -287,8 +308,8 @@ if (cluster.isPrimary || cluster.isMaster) { // delay startup for each reader to avoid startup transaction conflicts setTimeout(async () => { const connection = new ConnectionManager(connectionConfig); - filler = new Filler(readerConfigs[index], connection); + activeFiller = new Filler(readerConfigs[index], connection); - await filler.startFiller(5); + await activeFiller.startFiller(5); }, index * 1000); } diff --git a/src/filler/filler.test.ts b/src/filler/filler.test.ts index e0fed4d6..79a43937 100644 --- a/src/filler/filler.test.ts +++ b/src/filler/filler.test.ts @@ -1,5 +1,6 @@ import 'mocha'; import {expect} from 'chai'; +import * as sinon from 'sinon'; import Filler from './filler'; @@ -40,6 +41,33 @@ describe('Filler', () => { }); }); + describe('closeReaderBeforeExit', () => { + const stub = (stopProcessing: () => Promise): {filler: Filler; jobsStopped: () => boolean} => { + const filler = Object.create(Filler.prototype) as Filler; + let jobsStopped = false; + (filler as any).jobs = {stop: (): void => { jobsStopped = true; }}; + (filler as any).reader = {stopProcessing}; + return {filler, jobsStopped: (): boolean => jobsStopped}; + }; + + it('stops the reader, so the ship socket is closed before the process exits', async () => { + const closed = sinon.stub().resolves(); + const {filler, jobsStopped} = stub(closed); + + await filler.closeReaderBeforeExit(); + + expect(closed.calledOnce).to.equal(true); + expect(jobsStopped()).to.equal(true); + expect((filler as any).running).to.equal(false); + }); + + it('swallows a close failure, so a wedged reader cannot hold up the exit', async () => { + const {filler} = stub(() => Promise.reject(new Error('socket already gone'))); + + await filler.closeReaderBeforeExit(); + }); + }); + describe('shouldDeferDrain (hysteresis)', () => { const stub = (): { filler: Filler; setBehind: (n: number) => void } => { const filler = Object.create(Filler.prototype) as Filler; diff --git a/src/filler/filler.ts b/src/filler/filler.ts index c445ba74..cefa18f0 100644 --- a/src/filler/filler.ts +++ b/src/filler/filler.ts @@ -205,6 +205,8 @@ export default class Filler { if (this.reader.queueStopped) { logger.error('Reader ' + this.config.name + ' - consumer queue is dead, exiting immediately for restart'); + await this.closeReaderBeforeExit(); + process.send({msg: 'failure'}); await new Promise(resolve => setTimeout(resolve, logInterval / 2 * 1000)); @@ -237,6 +239,8 @@ export default class Filler { const staleTime = Date.now() - lastBlockTime; if (staleTime > timeout) { + await this.closeReaderBeforeExit(); + process.send({msg: 'failure'}); await new Promise(resolve => setTimeout(resolve, logInterval / 2 * 1000)); @@ -293,6 +297,19 @@ export default class Filler { this.running = true; } + // process.exit() abandons the ship websocket rather than closing it, so the + // node keeps a half-open state_history session until it works out the peer + // is gone. Both exit paths pause before exiting, which gives the close frame + // time to leave. A failure here must never block the exit, because a wedged + // reader is the case those paths exist to handle. + async closeReaderBeforeExit(): Promise { + try { + await this.stopFiller(); + } catch (error) { + logger.error('Failed to close the reader before exit', error); + } + } + async stopFiller(): Promise { this.running = false;