Fix reconnect logging and memleak

packages/host/src/lib/csi-controller.ts

@@ -169,7 +169,7 @@ export class CSIController extends TypedEmitter<Events> {
     /**
      * Streams connected do API.
      */
-    private downStreams?: DownstreamStreamsConfig;
+    private downStreams: DownstreamStreamsConfig | null = null;
     private upStreams: PassThroughStreamsConfig;
 
     public localEmitter: EventEmitter & { lastEvents: { [evname: string]: any } };
@@ -390,6 +390,15 @@ export class CSIController extends TypedEmitter<Events> {
         return this.instancePromise;
     }
 
+    unhookupStreams() {
+        this.downStreams![CC.STDOUT].unpipe();
+        this.downStreams![CC.STDERR].unpipe();
+        this.downStreams![CC.OUT].unpipe();
+        this.upStreams![CC.STDOUT].unpipe();
+        this.upStreams![CC.STDERR].unpipe();
+        this.upStreams![CC.OUT].unpipe();
+    }
+
     hookupStreams(streams: DownstreamStreamsConfig) {
         this.logger.trace("Hookup streams");
 
@@ -539,6 +548,19 @@ export class CSIController extends TypedEmitter<Events> {
         this.logger.info("Handshake", JSON.stringify(message, undefined));
     }
 
+    async handleInstanceDisconnect() {
+        this.bpmux?.removeAllListeners();
+        if (this.downStreams) this.unhookupStreams();
+
+        this.bpmux = null;
+        this.downStreams = null;
+    }
+
+    async handleInstanceReconnect(streams: DownstreamStreamsConfig) {
+        await this.handleInstanceDisconnect();
+        await this.handleInstanceConnect(streams);
+    }
+
     //@TODO: ! unhookup ! set proper state for reconnecting !
     async handleInstanceConnect(streams: DownstreamStreamsConfig) {
         try {

packages/host/src/lib/host.ts

@@ -1149,11 +1149,17 @@ export class Host implements IComponent {
                     new CommunicationHandler(),
                     this.config,
                     this.instanceProxy);
-            }
 
-            await this.instancesStore[id].handleInstanceConnect(
-                streams
-            );
+                await this.instancesStore[id].handleInstanceConnect(
+                    streams
+                );
+            } else {
+                this.logger.info("Instance already exists", id);
+
+                await this.instancesStore[id].handleInstanceReconnect(
+                    streams
+                );
+            }
         });
     }
 

packages/runner/src/host-client.ts

@@ -8,7 +8,7 @@ import net, { Socket, createConnection } from "net";
 import { PassThrough } from "stream";
 
 type HostOpenConnections = [
-    net.Socket, net.Socket, net.Socket, net.Socket, net.Socket, net.Socket, net.Socket, net.Socket, net.Socket
+    Socket, Socket, Socket, Socket, Socket, Socket, Socket, Socket, Socket
 ]
 
 const BPMux = require("bpmux").BPMux;
@@ -21,6 +21,7 @@ class HostClient implements IHostClient {
     public agent?: Agent;
     logger: IObjectLogger;
     bpmux: any;
+
     constructor(private instancesServerPort: number, private instancesServerHost: string) {
         this.logger = new ObjLogger(this);
     }
@@ -41,29 +42,30 @@ class HostClient implements IHostClient {
         throw new Error("No HTTP Agent set");
     }
 
-    async init(id: string): Promise<void> {
-        const openConnections = await Promise.all(
-            Array.from(Array(9))
-                .map((_e: any, i: number) => {
-                    // Error handling for each connection is process crash for now
-                    let connection: Socket;
-
-                    try {
-                        connection = net.createConnection(this.instancesServerPort, this.instancesServerHost);
-                        connection.on("error", () => {
-                            this.logger.warn(`${i} Stream error`);
-                        });
-                        connection.setNoDelay(true);
-                    } catch (e) {
-                        return Promise.reject(e);
-                    }
-
-                    return new Promise<net.Socket>(res => {
-                        connection.on("connect", () => {
-                            res(connection);
-                        });
+    private async connectOne(i: number): Promise<Socket> {
+        return new Promise<Socket>((res, rej) => {
+            try {
+                const connection = net.createConnection(this.instancesServerPort, this.instancesServerHost);
+
+                connection.setNoDelay(true);
+                connection.on("error", rej);
+                connection.on("connect", () => {
+                    res(connection);
+                    connection.removeAllListeners("error");
+                    connection.on("error", () => {
+                        this.logger.warn(`${i} Stream error`);
                     });
-                })
+                });
+            } catch (e) {
+                rej(e);
+            }
+        });
+    }
+
+    private async connect(id: string): Promise<HostOpenConnections> {
+        return Promise.all(
+            Array.from(Array(9))
+                .map((_e: void, i: number) => this.connectOne(i))
                 .map((connPromised, index) => {
                     return connPromised.then((connection) => {
                         // Assuming id is exactly 36 bytes
@@ -74,11 +76,12 @@ class HostClient implements IHostClient {
                         return connection;
                     });
                 })
-        ).catch((_e) => {
-            //@TODO: handle error.
-        });
+        ) as Promise<unknown> as Promise<HostOpenConnections>;
+    }
+
+    async init(id: string): Promise<void> {
+        this._streams = await this.connect(id);
 
-        this._streams = openConnections as HostOpenConnections;
         this._streams[CC.OUT].on("end", () => {
             this.logger.info("Total data written to instance output", (this.streams[CC.OUT] as net.Socket).bytesWritten);
         });

packages/runner/src/runner.ts

@@ -746,7 +746,7 @@ export class Runner<X extends AppConfig> implements IComponent {
         }
 
         // eslint-disable-next-line complexity
-        await new Promise<void>((res) => {
+        await new Promise<void>((res, rej) => {
             /**
              * @analyze-how-to-pass-in-out-streams
              * We need to make sure to close input and output streams
@@ -780,6 +780,12 @@ export class Runner<X extends AppConfig> implements IComponent {
                 this.logger.info("Stream encoding is", this.instanceOutput.readableEncoding);
 
                 this.instanceOutput
+                    .on("error", (e) => {
+                        this.logger.error("Sequence output stream error", e);
+                        this.status = InstanceStatus.ERRORED;
+
+                        rej(new RunnerError("SEQUENCE_RUNTIME_ERROR", e));
+                    })
                     .once("end", () => {
                         this.logger.debug("Sequence stream ended");
                         res();

packages/sth/src/bin/hub.ts

@@ -9,6 +9,7 @@ import { HostError } from "@scramjet/model";
 import { inspect } from "util";
 import { Host } from "@scramjet/host";
 import { FileBuilder, processCommanderRunnerEnvs } from "@scramjet/utility";
+import { constants } from "os";
 
 const stringToIntSanitizer = (str : string) => {
     const parsedValue = parseInt(str, 10);
@@ -31,6 +32,7 @@ const options: OptionValues & STHCommandOptions = program
     .option("-H, --hostname <IP>", "API IP")
     .option("-E, --identify-existing", "Index existing volumes as sequences")
     .option("-C, --cpm-url <host:ip>")
+    .option("-K, --kill-on-exit", "Kills all instances on exit")
     .option("--platform-api <url>", "Platform API url, ie. https://api.scramjet.org/api/v1")
     .option("--platform-api-version <version>", "Platform API version", "v1")
     .option("--platform-api-key <string>", "Platform API Key")
@@ -218,6 +220,29 @@ const options: OptionValues & STHCommandOptions = program
             if (config.telemetry.status) {
                 host.logger.info("Telemetry is active. If you don't want to send anonymous telemetry data use '--no-telemetry' when starting STH or set it in the config file.");
             }
+
+            if (options.killOnExit) {
+                let killing = false;
+                const kill = (signal: NodeJS.Signals) => {
+                    if (killing) return process.exit(constants.signals[signal]);
+                    killing = true;
+
+                    host.logger.warn("Kill on exit is enabled. Killing all instances and exiting.");
+                    host.stop()
+                        .then(() => {
+                            host.logger.info("All instances killed. Exiting.");
+                            process.exit(constants.signals[signal]);
+                        }, (e) => {
+                            host.logger.error("Error killing instances", e);
+                            process.exit(constants.signals[signal]);
+                        });
+
+                    return undefined;
+                };
+
+                process.on("SIGINT", kill);
+                process.on("SIGTERM", kill);
+            }
         });
 })()
     .catch((e: (Error | HostError) & { exitCode?: number }) => {
@@ -234,3 +259,4 @@ const options: OptionValues & STHCommandOptions = program
         process.exitCode = e.exitCode || 1;
         process.exit();
     });
+