[interpreter] Support Http1 based tests

services/node-services/src/app.ts

@@ -7,7 +7,12 @@
 // directory of this repository or package, or at
 // https://github.com/restatedev/e2e/blob/main/LICENSE
 
+/* eslint-disable @typescript-eslint/no-explicit-any */
+
 import * as restate from "@restatedev/restate-sdk";
+import { endpoint as fetchEndpoint } from "@restatedev/restate-sdk/fetch";
+import { endpoint as lambdaEndpoint } from "@restatedev/restate-sdk/lambda";
+import process from "node:process";
 
 import "./awakeable_holder";
 import "./counter";
@@ -26,19 +31,47 @@ import "./interpreter/entry_point";
 import "./workflow";
 
 import { REGISTRY } from "./services";
+import http1Server from "./h1server";
+
+class EndpointWrapper<K extends "fetch" | "lambda" | "node", T> {
+  static fromEnvironment() {
+    if (process.env.E2E_USE_FETCH) {
+      return new EndpointWrapper("fetch", fetchEndpoint());
+    } else if (process.env.AWS_LAMBDA_FUNCTION_NAME) {
+      return new EndpointWrapper("lambda", lambdaEndpoint());
+    } else {
+      return new EndpointWrapper("node", restate.endpoint());
+    }
+  }
 
-if (!process.env.SERVICES) {
-  throw new Error("Cannot find SERVICES env");
+  constructor(readonly kind: K, readonly endpoint: T) {}
 }
-const fqdns = new Set(process.env.SERVICES.split(","));
-const endpoint = restate.endpoint();
-REGISTRY.register(fqdns, endpoint);
+
+const wrapper = EndpointWrapper.fromEnvironment();
+REGISTRY.registerFromEnvironment(wrapper.endpoint);
 
 if (process.env.E2E_REQUEST_SIGNING) {
-  endpoint.withIdentityV1(...process.env.E2E_REQUEST_SIGNING.split(","));
+  const signing = process.env.E2E_REQUEST_SIGNING;
+  wrapper.endpoint.withIdentityV1(...signing.split(","));
 }
-if (!process.env.AWS_LAMBDA_FUNCTION_NAME) {
-  endpoint.listen();
+
+switch (wrapper.kind) {
+  case "node": {
+    wrapper.endpoint.listen();
+    break;
+  }
+  case "fetch": {
+    http1Server(wrapper.endpoint.handler());
+    break;
+  }
+  case "lambda": {
+    // do nothing, handler is exported
+    break;
+  }
+  default:
+    throw new Error("Unknown endpoint type");
 }
 
-export const handler = endpoint.lambdaHandler();
+// export for lambda. it will be only set for a lambda deployment
+export const handler =
+  wrapper.kind == "lambda" ? wrapper.endpoint.handler() : undefined;

services/node-services/src/h1server.ts

@@ -0,0 +1,87 @@
+// Copyright (c) 2023 - Restate Software, Inc., Restate GmbH
+//
+// This file is part of the Restate e2e tests,
+// which are released under the MIT license.
+//
+// You can find a copy of the license in file LICENSE in the root
+// directory of this repository or package, or at
+// https://github.com/restatedev/e2e/blob/main/LICENSE
+
+import http from "node:http";
+import { Buffer } from "node:buffer";
+
+/* eslint-disable @typescript-eslint/no-explicit-any */
+
+export default function h1server(handler: {
+  fetch: (request: Request) => Promise<Response>;
+}) {
+  let port = 9080;
+  if (process.env.PORT) {
+    try {
+      port = parseInt(process.env.PORT);
+    } catch (e) {
+      console.log(`Failed parsing ${process.env.PORT} with: `, e);
+      throw e;
+    }
+  }
+  const nodeHandler = nodeHandlerFromFetchHandler(handler.fetch);
+  http
+    .createServer(
+      {
+        keepAlive: true,
+      },
+      nodeHandler
+    )
+    .listen({ port, host: "0.0.0.0", backlog: 1024 * 1024 });
+}
+
+function nodeHandlerFromFetchHandler(
+  handler: (request: Request) => Promise<Response>
+): (req: http.IncomingMessage, res: http.ServerResponse) => void {
+  return (r, res) => {
+    toWeb(r)
+      .then((request) => handler(request))
+      .then(async (response: any) => {
+        const body = await response.arrayBuffer();
+        const headers = Object.fromEntries(response.headers.entries());
+        res.writeHead(response.status, headers);
+        res.end(Buffer.from(body));
+      });
+  };
+}
+
+function toWeb(req: http.IncomingMessage): Promise<Request> {
+  const { headers, method, url } = req;
+
+  return new Promise((resolve, reject) => {
+    const body: Uint8Array[] = [];
+
+    req.on("data", (chunk: Uint8Array) => {
+      body.push(chunk);
+    });
+
+    req.on("end", () => {
+      const buf = Buffer.concat(body);
+
+      const h = new Headers();
+      for (const [k, v] of Object.entries(headers)) {
+        // FIXME v can be something other than string.
+        h.append(k, v as string);
+      }
+
+      // Create a new Request object
+      const fullURL = new URL(`http://localhost${url}`);
+      const request = new Request(fullURL, {
+        method,
+        headers: h,
+        body: method !== "GET" && method !== "HEAD" ? buf : undefined,
+      });
+
+      resolve(request);
+    });
+
+    req.on("error", (err) => {
+      reject(err);
+    });
+  });
+}

services/node-services/src/interpreter/test_containers.ts

@@ -10,11 +10,30 @@
 import {
   GenericContainer,
   Network,
-  PullPolicy,
   StartedNetwork,
   StartedTestContainer,
 } from "testcontainers";
 
+export interface EnvironmentSpec {
+  restate: {
+    image: string;
+    env: Record<string, string>;
+    pull?: boolean;
+  };
+
+  interpreters: {
+    image: string;
+    env: Record<string, string>;
+    pull?: boolean;
+  };
+
+  service: {
+    image: string;
+    env: Record<string, string>;
+    pull?: boolean;
+  };
+}
+
 export interface TestEnvironment {
   ingressUrl: string;
   adminUrl: string;
@@ -31,82 +50,91 @@ export interface Containers {
   servicesContainer: StartedTestContainer;
 }
 
-export async function setupContainers(): Promise<TestEnvironment> {
+export async function setupContainers(
+  env: EnvironmentSpec
+): Promise<TestEnvironment> {
+  console.log(env);
+
   const network = await new Network().start();
 
-  const restate = new GenericContainer("ghcr.io/restatedev/restate:main")
+  const restate = new GenericContainer(env.restate.image)
     .withExposedPorts(8080, 9070)
     .withNetwork(network)
     .withNetworkAliases("restate")
-    .withPullPolicy(PullPolicy.alwaysPull())
+    .withPullPolicy({
+      shouldPull() {
+        return env.restate.pull ?? true;
+      },
+    })
     .withEnvironment({
       RESTATE_LOG_FILTER: "restate=warn",
       RESTATE_LOG_FORMAT: "json",
+      ...(env.restate?.env ?? {}),
     })
     .withUlimits({
       nproc: { soft: 65535, hard: 65535 },
       nofile: { soft: 65535, hard: 65535 },
     })
     .start();
 
-  const zero = new GenericContainer("ghcr.io/restatedev/e2e-node-services:main")
-    .withNetwork(network)
-    .withNetworkAliases("interpreter_zero")
-    .withPullPolicy(PullPolicy.alwaysPull())
-    .withEnvironment({
-      PORT: "9000",
-      RESTATE_LOGGING: "ERROR",
-      NODE_ENV: "production",
-      SERVICES: "ObjectInterpreterL0",
-    })
-    .start();
-
-  const one = new GenericContainer("ghcr.io/restatedev/e2e-node-services:main")
-    .withNetwork(network)
-    .withNetworkAliases("interpreter_one")
-    .withPullPolicy(PullPolicy.alwaysPull())
-
-    .withExposedPorts(9001)
-    .withEnvironment({
-      PORT: "9001",
+  const names = ["interpreter_zero", "interpreter_one", "interpreter_two"];
+  const interpreters = [];
+  for (let i = 0; i < 3; i++) {
+    const port = 9000 + i;
+    const auxEnv = {
+      PORT: `${port}`,
       RESTATE_LOGGING: "ERROR",
       NODE_ENV: "production",
-      SERVICES: "ObjectInterpreterL1",
-    })
-    .start();
+      NODE_OPTIONS: "--max-old-space-size=4096",
+      SERVICES: `ObjectInterpreterL${i}`,
+      ...(env.interpreters?.env ?? {}),
+    };
+    const interpreter = new GenericContainer(env.interpreters.image)
+      .withNetwork(network)
+      .withNetworkAliases(names[i])
+      .withExposedPorts(port)
+      .withPullPolicy({
+        shouldPull() {
+          return env.interpreters.pull ?? true;
+        },
+      })
+      .withEnvironment(auxEnv)
+      .withUlimits({
+        nproc: { soft: 65535, hard: 65535 },
+        nofile: { soft: 65535, hard: 65535 },
+      })
+      .start();
 
-  const two = new GenericContainer("ghcr.io/restatedev/e2e-node-services:main")
-    .withNetwork(network)
-    .withNetworkAliases("interpreter_two")
-    .withPullPolicy(PullPolicy.alwaysPull())
-    .withExposedPorts(9002)
-    .withEnvironment({
-      PORT: "9002",
-      RESTATE_LOGGING: "ERROR",
-      NODE_ENV: "production",
-      SERVICES: "ObjectInterpreterL2",
-    })
-    .start();
+    interpreters.push(interpreter);
+  }
 
-  const services = new GenericContainer(
-    "ghcr.io/restatedev/e2e-node-services:main"
-  )
+  const services = new GenericContainer(env.service.image)
     .withNetwork(network)
     .withNetworkAliases("services")
-    .withPullPolicy(PullPolicy.alwaysPull())
+    .withPullPolicy({
+      shouldPull() {
+        return env.service.pull ?? true;
+      },
+    })
     .withExposedPorts(9003)
     .withEnvironment({
       PORT: "9003",
       RESTATE_LOGGING: "ERROR",
       NODE_ENV: "production",
+      NODE_OPTIONS: "--max-old-space-size=4096",
       SERVICES: "ServiceInterpreterHelper",
+      ...(env.service?.env ?? {}),
+    })
+    .withUlimits({
+      nproc: { soft: 65535, hard: 65535 },
+      nofile: { soft: 65535, hard: 65535 },
     })
     .start();
 
   const restateContainer = await restate;
-  const interpreterZeroContainer = await zero;
-  const interpreterOneContainer = await one;
-  const interpreterTwoContainer = await two;
+  const interpreterZeroContainer = await interpreters[0];
+  const interpreterOneContainer = await interpreters[1];
+  const interpreterTwoContainer = await interpreters[2];
   const servicesContainer = await services;
 
   const ingressUrl = `http://${restateContainer.getHost()}:${restateContainer.getMappedPort(