Improve error handling (#552)

---------

Signed-off-by: Nik Nasr <[email protected]>

Author: nikrooz

packages/restate-e2e-services/src/app.ts

@@ -62,6 +62,8 @@ if (process.env.E2E_REQUEST_SIGNING) {
 }
 
 let INFLIGHT_REQUESTS = 0;
+let ACTIVE_SESSIONS = 0;
+const sessions = new Map();
 
 const handler = endpoint.http2Handler();
 const server = http2.createServer((req, res) => {
@@ -72,11 +74,43 @@ const server = http2.createServer((req, res) => {
   handler(req, res);
 });
 
+server.on("session", (session) => {
+  const sessionId = ACTIVE_SESSIONS++;
+  const streams = new Set();
+  sessions.set(sessionId, streams);
+
+  const handleCloseSession = () => {
+    sessions.delete(sessionId);
+  };
+
+  session.on("close", handleCloseSession);
+  session.on("error", handleCloseSession);
+
+  session.on("stream", (stream) => {
+    streams.add(`${sessionId}_${stream.id}`);
+
+    const handleCloseStream = () => {
+      streams.delete(`${sessionId}_${stream.id}`);
+    };
+
+    stream.on("close", handleCloseStream);
+    stream.on("error", handleCloseStream);
+  });
+
+  return undefined;
+});
+
 setInterval(() => {
   // eslint-disable-next-line no-console
   console.log(
     `${new Date().toISOString()}: Inflight requests: ${INFLIGHT_REQUESTS}`
   );
+  // eslint-disable-next-line no-console
+  console.table(
+    Array.from(sessions.values()).map((set: Set<string>) => ({
+      "#streams": set.size,
+    }))
+  );
 }, 30 * 1000);
 
 server.updateSettings(settings);

packages/restate-sdk/src/endpoint/handlers/generic.ts

@@ -271,6 +271,13 @@ export class GenericHandler implements RestateHandler {
       );
 
       const inputReader = body.getReader();
+      abortSignal.addEventListener(
+        "abort",
+        () => {
+          void inputReader.cancel();
+        },
+        { once: true }
+      );
 
       // Now buffer input entries
       while (!coreVm.is_ready_to_execute()) {

packages/restate-sdk/src/endpoint/node_endpoint.ts

@@ -74,6 +74,7 @@ export class NodeEndpoint implements RestateEndpoint {
     return (request, response) => {
       (async () => {
         const abortController = new AbortController();
+
         request.once("aborted", () => {
           abortController.abort();
         });
@@ -83,21 +84,33 @@ export class NodeEndpoint implements RestateEndpoint {
         request.once("error", () => {
           abortController.abort();
         });
+
+        if (request.destroyed || request.aborted) {
+          this.builder.rlog.error("Client disconnected");
+          abortController.abort();
+        }
+
         try {
           const url = request.url;
+          const webRequestBody = Readable.toWeb(request);
+
           const resp = await handler.handle({
             url,
             headers: request.headers,
-            body: Readable.toWeb(request),
+            body: webRequestBody,
             extraArgs: [],
             abortSignal: abortController.signal,
           });
+
+          if (response.destroyed) {
+            return;
+          }
+
           response.writeHead(resp.statusCode, resp.headers);
           const responseWeb = Writable.toWeb(
             response
           ) as WritableStream<Uint8Array>;
           await resp.body.pipeTo(responseWeb);
-          await new Promise<void>((resolve) => response.end(resolve));
         } catch (e) {
           const error = ensureError(e);
           this.builder.rlog.error(