[e2e] Support unlimited concurrent streams (#546)

Summary:
- Also enable heapdump with USR2 signal
- Show number of inflight invocation periodically

Author: muhamadazmy

package-lock.json

@@ -7,6 +7,7 @@ COPY . .
 RUN npm install
 
 RUN npm run build
+RUN npm rebuild heapdump
 
 # Install Tini
 RUN apt-get update && apt-get -y install tini

packages/restate-e2e-services/Dockerfile

@@ -32,10 +32,12 @@
   "dependencies": {
     "@restatedev/restate-sdk": "^1.6.0",
     "@restatedev/restate-sdk-clients": "^1.6.0",
+    "heapdump": "^0.3.15",
     "uuid": "^9.0.0"
   },
   "devDependencies": {
-    "@types/node": "*",
+    "@types/heapdump": "^0.3.4",
+    "@types/node": "^20.17.4",
     "@types/uuid": "^10.0.0",
     "tsx": "^4.15.7"
   },

packages/restate-e2e-services/package.json

@@ -25,6 +25,23 @@ import "./test_utils.js";
 import "./kill.js";
 import "./virtual_object_command_interpreter.js";
 import * as http2 from "http2";
+import * as heapdump from "heapdump";
+import path from "path";
+
+// Optional: trigger a heap snapshot on signal
+process.on("SIGUSR2", () => {
+  const filename = path.join("/opt", `heap-${Date.now()}.heapsnapshot`);
+  // eslint-disable-next-line no-console
+  console.log(`Writing snapshot to ${filename}...`);
+  heapdump.writeSnapshot(filename, (err, filename) => {
+    if (err) {
+      // eslint-disable-next-line no-console
+      console.error(err);
+    }
+    // eslint-disable-next-line no-console
+    else console.log(`Heap snapshot written to ${filename}`);
+  });
+});
 
 import { REGISTRY } from "./services.js";
 
@@ -35,14 +52,33 @@ const fqdns = new Set(process.env.SERVICES.split(","));
 const endpoint = restate.endpoint();
 REGISTRY.register(fqdns, endpoint);
 
+const settings: http2.Settings = {};
+if (process.env.MAX_CONCURRENT_STREAMS) {
+  settings.maxConcurrentStreams = parseInt(process.env.MAX_CONCURRENT_STREAMS);
+}
+
 if (process.env.E2E_REQUEST_SIGNING) {
   endpoint.withIdentityV1(...process.env.E2E_REQUEST_SIGNING.split(","));
 }
 
-const server = http2.createServer(endpoint.http2Handler());
-const maxConcurrentStreams = parseInt(
-  process.env.MAX_CONCURRENT_STREAMS || "256"
-);
-server.updateSettings({ maxConcurrentStreams: maxConcurrentStreams });
+let INFLIGHT_REQUESTS = 0;
+
+const handler = endpoint.http2Handler();
+const server = http2.createServer((req, res) => {
+  INFLIGHT_REQUESTS++;
+  res.once("close", () => {
+    INFLIGHT_REQUESTS--;
+  });
+  handler(req, res);
+});
+
+setInterval(() => {
+  // eslint-disable-next-line no-console
+  console.log(
+    `${new Date().toISOString()}: Inflight requests: ${INFLIGHT_REQUESTS}`
+  );
+}, 30 * 1000);
+
+server.updateSettings(settings);
 
 server.listen(parseInt(process.env.PORT || "9080"));