Goodbye invokeTrigger. Everything is http (unless in debug mode) (#4886)

This is it! It was all for this!

Now that all functions, regardless of signature, is invoked via an HTTP endpoint, we complete the refactor to remove all traces of IPC comm between the functions runtime and the functions emulator.

Instead of relying on inter-process messages from the runtime process to monitor the state of the function ("is it idle? is it processing request? is it done processing the request?") we simply let the Functions Emulator submit a request to the a worker associated with the trigger and infer status based on the state of the request.

The only way IPC is being used is when Functions Emulator is running in DEBUG mode. In debug mode, a single process is used to invoke all function triggers, and we use IPC to let the runtime process know which trigger should be called in the subsequent request. We'll tackle fixing this feature later - we can just let runtime for other language not support debug mode until we have a concrete plan on how we can support debug mode in the SDK itself.

Author: taeold

CHANGELOG.md

@@ -1 +1,2 @@
 - Fix Storage Emulator crashing with NoClassDefFoundError in some cases (#3481).
+- Refactor mechanism for invoking function triggers (#4886).

npm-shrinkwrap.json

@@ -214,6 +214,7 @@
     "google-discovery-to-swagger": "^2.1.0",
     "mocha": "^9.1.3",
     "nock": "^13.0.5",
+    "node-mocks-http": "^1.11.0",
     "nyc": "^15.1.0",
     "openapi-merge": "^1.0.23",
     "prettier": "^2.5.1",

package.json

@@ -179,12 +179,8 @@ async function sendReq(runtime: Runtime, opts: ReqOpts = {}): Promise<string> {
 }
 
 async function sendDebugBundle(runtime: Runtime, debug: FunctionsRuntimeBundle["debug"]) {
-  const frb: FunctionsRuntimeBundle = {
-    proto: {},
-    debug,
-  };
   return new Promise((resolve) => {
-    runtime.proc.send(JSON.stringify({ frb }), resolve);
+    runtime.proc.send(JSON.stringify(debug), resolve);
   });
 }
 

scripts/emulator-tests/functionsEmulatorRuntime.spec.ts

@@ -12,13 +12,7 @@ import { Account } from "../auth";
 import { logger } from "../logger";
 import { track, trackEmulator } from "../track";
 import { Constants } from "./constants";
-import {
-  EmulatorInfo,
-  EmulatorInstance,
-  EmulatorLog,
-  Emulators,
-  FunctionsExecutionMode,
-} from "./types";
+import { EmulatorInfo, EmulatorInstance, Emulators, FunctionsExecutionMode } from "./types";
 import * as chokidar from "chokidar";
 
 import * as spawn from "cross-spawn";
@@ -29,7 +23,6 @@ import {
   EventSchedule,
   EventTrigger,
   formatHost,
-  FunctionsRuntimeBundle,
   FunctionsRuntimeFeatures,
   getFunctionService,
   getSignatureType,
@@ -331,7 +324,13 @@ export class FunctionsEmulator implements EmulatorInstance {
     return hub;
   }
 
-  async sendRequest(worker: RuntimeWorker, body?: any) {
+  async sendRequest(trigger: EmulatedTriggerDefinition, body?: any) {
+    const record = this.getTriggerRecordByKey(this.getTriggerKey(trigger));
+    const pool = this.workerPools[record.backend.codebase];
+    if (!pool.readyForWork(trigger.id)) {
+      await this.startRuntime(record.backend, trigger);
+    }
+    const worker = pool.getIdleWorker(trigger.id)!;
     const reqBody = JSON.stringify(body);
     const headers = {
       "Content-Type": "application/json",
@@ -352,35 +351,6 @@ export class FunctionsEmulator implements EmulatorInstance {
     });
   }
 
-  async invokeTrigger(
-    trigger: EmulatedTriggerDefinition,
-    proto?: any,
-    runtimeOpts?: InvokeRuntimeOpts
-  ): Promise<RuntimeWorker> {
-    const record = this.getTriggerRecordByKey(this.getTriggerKey(trigger));
-    const backend = record.backend;
-    const bundleTemplate = this.getBaseBundle();
-    const runtimeBundle: FunctionsRuntimeBundle = {
-      ...bundleTemplate,
-      proto,
-    };
-    if (this.args.debugPort) {
-      runtimeBundle.debug = {
-        functionTarget: trigger.entryPoint,
-        functionSignature: getSignatureType(trigger),
-      };
-    }
-    if (!backend.nodeBinary) {
-      throw new FirebaseError(`No node binary for ${trigger.id}. This should never happen.`);
-    }
-    const opts = runtimeOpts || {
-      nodeBinary: backend.nodeBinary,
-      extensionTriggers: backend.predefinedTriggers,
-    };
-    const worker = await this.invokeRuntime(backend, trigger, runtimeBundle, opts);
-    return worker;
-  }
-
   async start(): Promise<void> {
     for (const backend of this.args.emulatableBackends) {
       backend.nodeBinary = this.getNodeBinary(backend);
@@ -680,7 +650,7 @@ export class FunctionsEmulator implements EmulatorInstance {
     // In debug mode, we eagerly start a runtime process to allow debuggers to attach
     // before invoking a function.
     if (this.args.debugPort) {
-      await this.startRuntime(emulatableBackend, { nodeBinary: emulatableBackend.nodeBinary });
+      await this.startRuntime(emulatableBackend);
     }
   }
 
@@ -998,13 +968,6 @@ export class FunctionsEmulator implements EmulatorInstance {
     triggers.forEach((def) => this.addTriggerRecord(def, { backend, ignored: false }));
   }
 
-  getBaseBundle(): FunctionsRuntimeBundle {
-    return {
-      proto: {},
-      disabled_features: this.args.disabledRuntimeFeatures,
-    };
-  }
-
   getNodeBinary(backend: EmulatableBackend): string {
     const pkg = require(path.join(backend.functionsDir, "package.json"));
     // If the developer hasn't specified a Node to use, inform them that it's an option and use default
@@ -1275,37 +1238,18 @@ export class FunctionsEmulator implements EmulatorInstance {
     return secretEnvs;
   }
 
-  async invokeRuntime(
-    backend: EmulatableBackend,
-    trigger: EmulatedTriggerDefinition,
-    frb: FunctionsRuntimeBundle,
-    opts: InvokeRuntimeOpts
-  ): Promise<RuntimeWorker> {
-    const pool = this.workerPools[backend.codebase];
-    if (!pool.readyForWork(trigger.id)) {
-      await this.startRuntime(backend, opts, trigger);
-    }
-    return pool.submitWork(trigger.id, frb, opts);
-  }
-
   async startRuntime(
     backend: EmulatableBackend,
-    opts: InvokeRuntimeOpts,
     trigger?: EmulatedTriggerDefinition
-  ) {
+  ): Promise<RuntimeWorker> {
     const emitter = new EventEmitter();
     const args = [path.join(__dirname, "functionsEmulatorRuntime")];
 
-    if (opts.ignore_warnings) {
-      args.unshift("--no-warnings");
-    }
-
     if (this.args.debugPort) {
-      if (process.env.FIREPIT_VERSION && process.execPath === opts.nodeBinary) {
-        const requestedMajorNodeVersion = this.getNodeBinary(backend);
+      if (process.env.FIREPIT_VERSION && process.execPath === backend.nodeBinary) {
         this.logger.log(
           "WARN",
-          `To enable function inspection, please run "${process.execPath} is:npm i node@${requestedMajorNodeVersion} --save-dev" in your functions directory`
+          `To enable function inspection, please run "${process.execPath} is:npm i node@${backend.nodeMajorVersion} --save-dev" in your functions directory`
         );
       } else {
         const { host } = this.getInfo();
@@ -1332,10 +1276,10 @@ export class FunctionsEmulator implements EmulatorInstance {
     const secretEnvs = await this.resolveSecretEnvs(backend, trigger);
     const socketPath = getTemporarySocketPath();
 
-    const childProcess = spawn(opts.nodeBinary, args, {
+    const childProcess = spawn(backend.nodeBinary!, args, {
       cwd: backend.functionsDir,
       env: {
-        node: opts.nodeBinary,
+        node: backend.nodeBinary,
         ...process.env,
         ...runtimeEnv,
         ...secretEnvs,
@@ -1357,7 +1301,7 @@ export class FunctionsEmulator implements EmulatorInstance {
     const pool = this.workerPools[backend.codebase];
     const worker = pool.addWorker(trigger?.id, runtime, extensionLogInfo);
     await worker.waitForSocketReady();
-    return;
+    return worker;
   }
 
   async disableBackgroundTriggers() {
@@ -1480,20 +1424,12 @@ export class FunctionsEmulator implements EmulatorInstance {
         );
       }
     }
-    const worker = await this.invokeTrigger(trigger);
-
     // For analytics, track the invoked service
     void track(EVENT_INVOKE, getFunctionService(trigger));
     void trackEmulator(EVENT_INVOKE_GA4, {
       function_service: getFunctionService(trigger),
     });
 
-    worker.onLogs((el: EmulatorLog) => {
-      if (el.level === "FATAL") {
-        res.status(500).send(el.text);
-      }
-    });
-
     this.logger.log("DEBUG", `[functions] Runtime ready! Sending request!`);
 
     // To match production behavior we need to drop the path prefix
@@ -1508,59 +1444,27 @@ export class FunctionsEmulator implements EmulatorInstance {
     // cause unexpected situations - not to mention CORS troubles and this enables us to use
     // a socketPath (IPC socket) instead of consuming yet another port which is probably faster as well.
     this.logger.log("DEBUG", `[functions] Got req.url=${req.url}, mapping to path=${path}`);
-    const runtimeReq = http.request(
+
+    const pool = this.workerPools[record.backend.codebase];
+    if (!pool.readyForWork(trigger.id)) {
+      await this.startRuntime(record.backend, trigger);
+    }
+    const debugBundle = this.args.debugPort
+      ? {
+          functionTarget: trigger.entryPoint,
+          functionSignature: getSignatureType(trigger),
+        }
+      : undefined;
+    await pool.submitRequest(
+      trigger.id,
       {
         method,
         path,
         headers: req.headers,
-        socketPath: worker.runtime.socketPath,
       },
-      (runtimeRes: http.IncomingMessage) => {
-        function forwardStatusAndHeaders(): void {
-          res.status(runtimeRes.statusCode || 200);
-          if (!res.headersSent) {
-            Object.keys(runtimeRes.headers).forEach((key) => {
-              const val = runtimeRes.headers[key];
-              if (val) {
-                res.setHeader(key, val);
-              }
-            });
-          }
-        }
-
-        runtimeRes.on("data", (buf) => {
-          forwardStatusAndHeaders();
-          res.write(buf);
-        });
-
-        runtimeRes.on("close", () => {
-          forwardStatusAndHeaders();
-          res.end();
-        });
-
-        runtimeRes.on("end", () => {
-          forwardStatusAndHeaders();
-          res.end();
-        });
-      }
+      res,
+      reqBody,
+      debugBundle
     );
-
-    runtimeReq.on("error", () => {
-      res.end();
-    });
-
-    // If the original request had a body, forward that over the connection.
-    // TODO: Why is this not handled by the pipe?
-    if (reqBody) {
-      runtimeReq.write(reqBody);
-      runtimeReq.end();
-    }
-
-    // Pipe the incoming request over the socket.
-    req.pipe(runtimeReq, { end: true }).on("error", () => {
-      res.end();
-    });
-
-    await worker.waitForDone();
   }
 }