Fix input & not-serialized output on reconnect

Author: patuwwy

packages/adapters/src/process-instance-adapter.ts

@@ -254,7 +254,9 @@ class ProcessInstanceAdapter implements
                         process.kill(this.processPID, 0);
                     } catch (e) {
                         this.logger.error("Runner process not exists", e);
-                        /** process not exists */
+
+                        clearInterval(interval);
+
                         reject("pid not exists");
                     }
                 }

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

@@ -116,6 +116,7 @@ export class CSIController extends TypedEmitter<Events> {
     apiInputEnabled = true;
 
     executionTime: number = -1;
+    inputHeadersSent = false;
 
     /**
      * Topic to which the output stream should be routed
@@ -222,7 +223,8 @@ export class CSIController extends TypedEmitter<Events> {
             this.logger.info("Instance status: errored", e);
 
             this.status ||= InstanceStatus.ERRORED;
-            this.executionTime = (Date.now() - this.info.created!.getTime()) / 1000;
+
+            this.executionTime = this.info.created ? (Date.now() - this.info.created!.getTime()) / 1000 : -1;
 
             this.setExitInfo(e.exitcode, e.message);
 
@@ -403,9 +405,10 @@ export class CSIController extends TypedEmitter<Events> {
             .pipe(this.upStreams[CC.CONTROL]);
 
         this.communicationHandler.addMonitoringHandler(RunnerMessageCode.PING, async (message) => {
-            const { status, payload } = message[1];
+            const { status, payload, inputHeadersSent } = message[1];
 
             this.status = status || InstanceStatus.RUNNING;
+            this.inputHeadersSent = inputHeadersSent;
 
             if (!payload) {
                 this.emit("error", "No payload in ping!");
@@ -485,6 +488,10 @@ export class CSIController extends TypedEmitter<Events> {
             this.logger.trace("Received a PING message with ports config");
         }
 
+        this.inputHeadersSent = !!message[1].inputHeadersSent;
+
+        this.logger.info("Headers already sent for input?", this.inputHeadersSent);
+
         if (this.instanceAdapter.setRunner) {
             await this.instanceAdapter.setRunner({
                 ...message[1].payload.system,
@@ -537,8 +544,6 @@ export class CSIController extends TypedEmitter<Events> {
     }
 
     createInstanceAPIRouter() {
-        let inputHeadersSent = false;
-
         if (!this.upStreams) {
             throw new AppError("UNATTACHED_STREAMS");
         }
@@ -551,11 +556,11 @@ export class CSIController extends TypedEmitter<Events> {
          * @experimental
          */
         this.router.duplex("/inout", (duplex, _headers) => {
-            if (!inputHeadersSent) {
+            if (!this.inputHeadersSent) {
                 this.downStreams![CC.IN].write(`Content-Type: ${_headers["content-type"]}\r\n`);
                 this.downStreams![CC.IN].write("\r\n");
 
-                inputHeadersSent = true;
+                this.inputHeadersSent = true;
             }
 
             (duplex as unknown as DuplexStream).input.pipe(this.downStreams![CC.IN], { end: false });
@@ -597,15 +602,15 @@ export class CSIController extends TypedEmitter<Events> {
                 const contentType = req.headers["content-type"];
 
                 // @TODO: Check if subsequent requests have the same content-type.
-                if (!inputHeadersSent) {
+                if (!this.inputHeadersSent) {
                     if (contentType === undefined) {
                         return { opStatus: ReasonPhrases.NOT_ACCEPTABLE, error: "Content-Type must be defined" };
                     }
 
                     stream.write(`Content-Type: ${contentType}\r\n`);
                     stream.write("\r\n");
 
-                    inputHeadersSent = true;
+                    this.inputHeadersSent = true;
                 }
 
                 return stream;

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

@@ -66,12 +66,14 @@ export class CSIDispatcher extends TypedEmitter<Events> {
             id,
             sequenceInfo,
             payload,
-            status: InstanceStatus.INITIALIZING
+            status: InstanceStatus.INITIALIZING,
+            inputHeadersSent: false
         }, communicationHandler, config, instanceProxy, this.STHConfig.runtimeAdapter);
 
         this.logger.trace("CSIController created", id, sequenceInfo);
 
         csiController.logger.pipe(this.logger, { end: false });
+
         communicationHandler.logger.pipe(this.logger, { end: false });
 
         csiController
@@ -99,15 +101,15 @@ export class CSIDispatcher extends TypedEmitter<Events> {
                     this.logger.warn("Missing topic content-type");
                 }
 
-                if (data.requires && !csiController.inputRouted && data.contentType) {
-                    this.logger.trace("Routing topic to Sequence input", data.requires);
+                if (data.requires && data.contentType) {
+                    this.logger.trace("Routing topic to Instance input", data.requires);
 
                     await this.serviceDiscovery.routeTopicToStream(
                         { topic: new TopicId(data.requires), contentType: data.contentType as ContentType },
                         csiController.getInputStream()
                     );
 
-                    csiController.inputRouted = true;
+                    csiController.inputHeadersSent = true;
 
                     await this.serviceDiscovery.update({
                         requires: data.requires, contentType: data.contentType, topicName: data.requires, status: "add"

packages/host/src/lib/host.ts

@@ -340,7 +340,7 @@ export class Host implements IComponent {
 
         const seq = this.sequenceStore.getById(instance.sequence.id);
 
-        if (!seq) {
+        if (!seq && this.cpmConnector?.connected) {
             this.logger.info("Sequence not found. Checking Store...");
 
             try {

packages/host/src/lib/serviceDiscovery/sd-adapter.ts

@@ -179,9 +179,8 @@ export class ServiceDiscovery {
     }
 
     async update(data: STHTopicEventData) {
-        this.logger.trace("Topic update. Send topic info to CPM", data);
-
         if (this.cpmConnector?.connected) {
+            this.logger.trace("Topic update. Send topic info to CPM", data);
             await this.cpmConnector?.sendTopicInfo(data);
         }
     }

packages/host/src/lib/socket-server.ts

@@ -77,6 +77,7 @@ export class SocketServer extends TypedEmitter<Events> implements IComponent {
             this.server!
                 .listen(this.port, this.hostname, () => {
                     this.logger.info("SocketServer on", this.server?.address());
+
                     res();
                 })
                 .on("error", rej);

packages/runner/src/host-client.ts

@@ -2,8 +2,10 @@
 import { ObjLogger } from "@scramjet/obj-logger";
 import { CommunicationChannel as CC } from "@scramjet/symbols";
 import { IHostClient, IObjectLogger, UpstreamStreamsConfig, } from "@scramjet/types";
+import { defer } from "@scramjet/utility";
 import { Agent } from "http";
 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
@@ -42,20 +44,24 @@ class HostClient implements IHostClient {
     async init(id: string): Promise<void> {
         const openConnections = await Promise.all(
             Array.from(Array(9))
-                .map(() => {
+                .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", () => {});
+                        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));
+                        connection.on("connect", () => {
+                            res(connection);
+                        });
                     });
                 })
                 .map((connPromised, index) => {
@@ -74,6 +80,26 @@ class HostClient implements IHostClient {
 
         this._streams = openConnections as HostOpenConnections;
 
+        const input = this._streams[CC.IN];
+
+        const inputTarget = new PassThrough({ emitClose: false });
+
+        input.on("end", async () => {
+            await defer(500);
+
+            if ((this._streams![CC.CONTROL] as net.Socket).readableEnded) {
+                this.logger.info("Input end. Control is also ended... We are disconnected.");
+            } else {
+                this.logger.info("Input end. Control not ended. We are online. Desired input end.");
+                inputTarget.end();
+            }
+        });
+
+        input.pipe(inputTarget, { end: false });
+
+        this._streams[CC.IN] = inputTarget;
+        //this._streams[CC.STDIN] = this._streams[CC.STDIN].pipe(new PassThrough({ emitClose: false }), { end: false });
+
         try {
             this.bpmux = new BPMux(this._streams[CC.PACKAGE]);
         } catch (e) {
@@ -118,6 +144,11 @@ class HostClient implements IHostClient {
         const streamsExitedPromised: Promise<void>[] = this.streams.map((stream, i) =>
             new Promise(
                 (res) => {
+                    if ([CC.IN, CC.STDIN, CC.CONTROL].includes(i)) {
+                        res();
+                        return;
+                    }
+
                     if (!hard && "writable" in stream!) {
                         stream
                             .on("error", (e) => {