[teraslice-messaging, teraslice] Add abortController for use with sen…

…dSliceComplete() (#3838)

This PR makes the following changes:

- The execution-controller `Client` now creates an `abortController`.
Every call to `Client.sendSliceComplete()` now includes a reference to
`abortController.signal` allowing for the `pEvent` in the messenger
`Core.onceWithTimeout()` to be aborted.
- A new listener is added for the `server:shutdown` event that calls
`abortController.abort()`. This prevents a worker from still waiting for
a response to `sendSliceComplete()` after the server is shutdown.
- bump teraslice-messaging from 1.7.0 to 1.7.1

ref: #2106

packages/teraslice-messaging/package.json

@@ -1,7 +1,7 @@
 {
     "name": "@terascope/teraslice-messaging",
     "displayName": "Teraslice Messaging",
-    "version": "1.7.0",
+    "version": "1.7.1",
     "description": "An internal teraslice messaging library using socket.io",
     "homepage": "https://github.com/terascope/teraslice/tree/master/packages/teraslice-messaging#readme",
     "bugs": {

packages/teraslice-messaging/src/execution-controller/client.ts

@@ -7,6 +7,7 @@ const ONE_MIN = 60 * 1000;
 
 export class Client extends core.Client {
     public workerId: string;
+    private abortController: AbortController;
 
     constructor(opts: i.ClientOptions) {
         const {
@@ -47,6 +48,7 @@ export class Client extends core.Client {
 
         this.workerId = workerId;
         this.available = false;
+        this.abortController = new AbortController();
     }
 
     async start() {
@@ -75,6 +77,13 @@ export class Client extends core.Client {
                 payload: msg.payload,
             });
         });
+
+        this.on('server:shutdown', () => {
+            // this will send an abort signal to the Core.onceWithTimeout pEvent
+            // allowing the worker to shutdown without receiving a response to
+            // a sendSliceComplete() message
+            this.abortController.abort();
+        });
     }
 
     onExecutionFinished(fn: () => void) {
@@ -85,6 +94,7 @@ export class Client extends core.Client {
         return this.send('worker:slice:complete', withoutNil(payload), {
             response: true,
             volatile: false,
+            signal: this.abortController.signal
         });
     }
 

packages/teraslice-messaging/src/messenger/client.ts

@@ -278,7 +278,7 @@ export class Client extends Core {
             respondBy,
         };
 
-        const responseMsg = this.handleSendResponse(message);
+        const responseMsg = this.handleSendResponse(message, options.signal);
         this.socket.emit(eventName, message);
         return responseMsg;
     }

packages/teraslice-messaging/src/messenger/core.ts

@@ -2,7 +2,8 @@ import ms from 'ms';
 import { pEvent } from 'p-event';
 import { EventEmitter } from 'node:events';
 import {
-    toString, isInteger, debugLogger, Logger
+    toString, isInteger, debugLogger,
+    Logger, TSError
 } from '@terascope/utils';
 import * as i from './interfaces.js';
 
@@ -40,11 +41,22 @@ export class Core extends EventEmitter {
         this.removeAllListeners();
     }
 
-    protected async handleSendResponse(sent: i.Message): Promise<i.Message | null> {
+    protected async handleSendResponse(
+        sent: i.Message,
+        signal?: AbortSignal
+    ): Promise<i.Message | null> {
         if (!sent.response) return null;
 
         const remaining = sent.respondBy - Date.now();
-        const response = await this.onceWithTimeout(sent.id, remaining);
+        const response = await this.onceWithTimeout(sent.id, remaining, signal);
+
+        // server shutdown
+        if (signal?.aborted) {
+            const msg = sent.eventName === 'worker:slice:complete'
+                ? `Execution controller shutdown before receiving worker slice analytics. Event: "${sent.eventName}"`
+                : `Execution controller shutdown before receiving "${sent.eventName}" event`;
+            throw new TSError(msg, { retryable: false });
+        }
 
         // it is a timeout
         if (response == null) {
@@ -136,12 +148,17 @@ export class Core extends EventEmitter {
         }
     }
 
-    async onceWithTimeout(eventName: string, timeout?: number): Promise<any> {
+    async onceWithTimeout(
+        eventName: string,
+        timeout?: number,
+        abortSignal?: AbortSignal
+    ): Promise<any> {
         const timeoutMs: number = this.getTimeout(timeout);
         try {
             const { payload } = (await pEvent(this, eventName, {
                 rejectionEvents: [],
                 timeout: timeoutMs,
+                signal: abortSignal
             })) as i.EventMessage;
             return payload;
         } catch (err) {

packages/teraslice-messaging/src/messenger/interfaces.ts

@@ -57,6 +57,7 @@ export interface SendOptions {
     volatile?: boolean;
     response?: boolean;
     timeout?: number;
+    signal?: AbortSignal;
 }
 
 export interface ConnectedClient {

packages/teraslice/package.json

@@ -41,7 +41,7 @@
         "@kubernetes/client-node": "~0.22.0",
         "@terascope/elasticsearch-api": "~4.4.0",
         "@terascope/job-components": "~1.6.0",
-        "@terascope/teraslice-messaging": "~1.7.0",
+        "@terascope/teraslice-messaging": "~1.7.1",
         "@terascope/types": "~1.3.0",
         "@terascope/utils": "~1.4.0",
         "async-mutex": "~0.5.0",

packages/teraslice/src/lib/workers/worker/index.ts

@@ -349,6 +349,9 @@ export class Worker {
             } catch (err) {
                 if (this.isShuttingDown) {
                     throw err;
+                } else if (err.retryable === false) {
+                    this.logger.warn(`${err}, will not retry.`);
+                    return true;
                 } else {
                     this.logger.warn(err);
                 }