Fix bug where event triggered functions failed in debug mode (#5211)

A classic case of deadlock!

Internally, the Functions Emulator maintains a work queue which processes unit of work called "tasks" with configured parallelism.

When Functions Emulator runs in debug mode, the work queue runs in `SEQUENTIAL` mode where tasks are processed one by one in FIFO order.

When event functions are triggered via the multicast route (e.g. storage and auth triggers), Functions Emulator submits a task per function associated with the event to the work queue where each task makes call to the event function via its HTTP endpoint. Let's call this task an "invocation task".

When the Function Emulator receives an HTTP request for function invocation, it submits a new task to the work queue to handle the http request. Let's call this task a "http task".

So, a call to the multicast route begets zero or more "invocation task" (one per trigger). Each "invocation task" begets exactly one "http task".

The code today is written such that an "invocation task" will wait for the corresponding "http task" to complete.

In debug mode this causes the invocation task to wait forever - the "http task" is stuck behind its "invocation task" in the queue. It never has chance to be processed.

This PR proposes that "invocation task" doesn't wait for the corresponding "http task" to complete, effectively breaking the deadlock.

The result is that "invocation task" now fire-and-forget an "http task". AFAIK, this doesn't have any negative effect because "invocation task" didn't do anything with the result of the "http task" anyway.

Fixes #5008, #5050

CHANGELOG.md

@@ -1,3 +1,4 @@
 - Fixes gzipped file handling in Storage Emulator.
 - Add support for object list using certain Admin SDKs (#5208)
 - Fixes source token expiration issue by acquiring new source token upon expiration.
+- Fix bug where emulated event triggered function broke in debug mode (#5211)

scripts/triggers-end-to-end-tests/tests.inspect.ts

@@ -10,6 +10,7 @@ const FIREBASE_PROJECT = process.env.FBTOOLS_TARGET_PROJECT || "";
  * parallel emulator subprocesses.
  */
 const TEST_SETUP_TIMEOUT = 80000;
+const EMULATORS_WRITE_DELAY_MS = 5000;
 const EMULATORS_SHUTDOWN_DELAY_MS = 5000;
 
 function readConfig(): FrameworkOptions {
@@ -28,45 +29,68 @@ describe("function triggers with inspect flag", () => {
 
     const config = readConfig();
     test = new TriggerEndToEndTest(FIREBASE_PROJECT, __dirname, config);
-    await test.startEmulators(["--only", "functions", "--inspect-functions"]);
+    await test.startEmulators(["--only", "functions,auth,storage", "--inspect-functions"]);
   });
 
   after(async function (this) {
     this.timeout(EMULATORS_SHUTDOWN_DELAY_MS);
     await test.stopEmulators();
   });
 
-  it("should invoke correct function in the same codebase", async function (this) {
-    this.timeout(TEST_SETUP_TIMEOUT);
-    const v1response = await test.invokeHttpFunction("onreqv2b");
-    expect(v1response.status).to.equal(200);
-    const v1body = await v1response.text();
-    expect(v1body).to.deep.equal("onreqv2b");
-
-    const v2response = await test.invokeHttpFunction("onreqv2a");
-    expect(v2response.status).to.equal(200);
-    const v2body = await v2response.text();
-    expect(v2body).to.deep.equal("onreqv2a");
-  });
+  describe("http functions", () => {
+    it("should invoke correct function in the same codebase", async function (this) {
+      this.timeout(TEST_SETUP_TIMEOUT);
+      const v1response = await test.invokeHttpFunction("onreqv2b");
+      expect(v1response.status).to.equal(200);
+      const v1body = await v1response.text();
+      expect(v1body).to.deep.equal("onreqv2b");
 
-  it("should invoke correct function across codebases", async function (this) {
-    this.timeout(TEST_SETUP_TIMEOUT);
-    const v1response = await test.invokeHttpFunction("onReq");
-    expect(v1response.status).to.equal(200);
-    const v1body = await v1response.text();
-    expect(v1body).to.deep.equal("onReq");
-
-    const v2response = await test.invokeHttpFunction("onreqv2a");
-    expect(v2response.status).to.equal(200);
-    const v2body = await v2response.text();
-    expect(v2body).to.deep.equal("onreqv2a");
+      const v2response = await test.invokeHttpFunction("onreqv2a");
+      expect(v2response.status).to.equal(200);
+      const v2body = await v2response.text();
+      expect(v2body).to.deep.equal("onreqv2a");
+    });
+
+    it("should invoke correct function across codebases", async function (this) {
+      this.timeout(TEST_SETUP_TIMEOUT);
+      const v1response = await test.invokeHttpFunction("onReq");
+      expect(v1response.status).to.equal(200);
+      const v1body = await v1response.text();
+      expect(v1body).to.deep.equal("onReq");
+
+      const v2response = await test.invokeHttpFunction("onreqv2a");
+      expect(v2response.status).to.equal(200);
+      const v2body = await v2response.text();
+      expect(v2body).to.deep.equal("onreqv2a");
+    });
+
+    it("should disable timeout", async function (this) {
+      this.timeout(TEST_SETUP_TIMEOUT);
+      const v2response = await test.invokeHttpFunction("onreqv2timeout");
+      expect(v2response.status).to.equal(200);
+      const v2body = await v2response.text();
+      expect(v2body).to.deep.equal("onreqv2timeout");
+    });
   });
 
-  it("should disable timeout", async function (this) {
-    this.timeout(TEST_SETUP_TIMEOUT);
-    const v2response = await test.invokeHttpFunction("onreqv2timeout");
-    expect(v2response.status).to.equal(200);
-    const v2body = await v2response.text();
-    expect(v2body).to.deep.equal("onreqv2timeout");
+  describe("event triggered (multicast) functions", () => {
+    it("should trigger auth triggered functions in response to auth events", async function (this) {
+      this.timeout(TEST_SETUP_TIMEOUT);
+      const response = await test.writeToAuth();
+      expect(response.status).to.equal(200);
+      await new Promise((resolve) => setTimeout(resolve, EMULATORS_WRITE_DELAY_MS));
+      expect(test.authTriggerCount).to.equal(1);
+    });
+
+    it("should trigger storage triggered functions in response to storage events across codebases", async function (this) {
+      this.timeout(TEST_SETUP_TIMEOUT);
+
+      const response = await test.writeToDefaultStorage();
+      expect(response.status).to.equal(200);
+      await new Promise((resolve) => setTimeout(resolve, EMULATORS_WRITE_DELAY_MS));
+
+      expect(test.storageFinalizedTriggerCount).to.equal(1);
+      expect(test.storageV2FinalizedTriggerCount).to.equal(1);
+    });
   });
 });

src/emulator/functionsEmulator.ts

@@ -292,20 +292,18 @@ export class FunctionsEmulator implements EmulatorInstance {
       const { host, port } = this.getInfo();
       triggers.forEach((triggerId) => {
         this.workQueue.submit(() => {
-          return new Promise((resolve, reject) => {
-            const trigReq = http.request(
-              {
-                host: connectableHostname(host),
-                port,
-                method: req.method,
-                path: `/functions/projects/${projectId}/triggers/${triggerId}`,
-                headers: req.headers,
-              },
-              resolve
-            );
+          return new Promise<void>((resolve, reject) => {
+            const trigReq = http.request({
+              host: connectableHostname(host),
+              port,
+              method: req.method,
+              path: `/functions/projects/${projectId}/triggers/${triggerId}`,
+              headers: req.headers,
+            });
             trigReq.on("error", reject);
             trigReq.write(rawBody);
             trigReq.end();
+            resolve();
           });
         });
       });