From e63ff26e8558360bdc8e74ab630ea95aaf013a70 Mon Sep 17 00:00:00 2001 From: Rob Konsdorf Date: Tue, 25 Aug 2026 11:13:40 -0400 Subject: [PATCH] fix(filler): close the ship socket before a hard exit The filler leaves on process.exit(1) from an unhandled rejection, an uncaught exception, and both watchdog paths, and none of them closed the state_history websocket first. Abandoning a socket the client could have closed leaves the node holding a half-open session until it works out the peer is gone, so a filler restarted straight after a crash opens a second session beside the first. A repeating foreign-key crash showed the shape of it in the field: an immediate restart crawled at the live chain rate while the same restart with a short pause ran at full speed. The node half of that is inference rather than something measured, but the abandoned socket is wrong on its own terms. Both watchdog paths already pause before exiting, so they take the full graceful stop and the close frame leaves while the process is still alive. The crash handlers cannot await anything safely, since a wedged reader is exactly what they exist to escape, so they send the frame and exit on a short fixed timer instead. Signed-off-by: Rob Konsdorf --- src/bin/filler.ts | 35 ++++++++++++++++++++++++++++------- src/filler/filler.test.ts | 28 ++++++++++++++++++++++++++++ src/filler/filler.ts | 17 +++++++++++++++++ 3 files changed, 73 insertions(+), 7 deletions(-) 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;