fix(filler): close the ship socket before a hard exit

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

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<void>): {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;

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<void> {
+        try {
+            await this.stopFiller();
+        } catch (error) {
+            logger.error('Failed to close the reader before exit', error);
+        }
+    }
+
     async stopFiller(): Promise<void> {
         this.running = false;