fix(emulator): Storage emulator hangs under concurrent requests

CHANGELOG.md

@@ -3,3 +3,6 @@
 - Fixes Storage Emulator to support JSON uploads larger than 100KB without hanging or throwing 413 error (#8355)
 - Add `extdeprecationwarnings` experiment to display phased deprecation notices and guidance across `ext:*` CLI commands.
 - Fixes Data Connect emulator crash when in-flight GraphQL requests are cancelled (#10821)
+- Fixed a Cloud Storage emulator hang under concurrent requests, caused by the rules runtime's stdout being parsed per-chunk instead of per-line so batched responses were dropped (#6194, #6865).
+- Support for specifying that the input for a string or string[] param in Functions must be non-empty (#10678)
+- Removed the warning that Dart functions may not yet be visible in the Firebase Console, since they are now shown.

src/emulator/storage/rules/runtime.spec.ts

@@ -1,6 +1,29 @@
 import { expect } from "chai";
-import { createAuthExpressionValue } from "./runtime";
-import { RulesetOperationMethod } from "./types";
+import { createAuthExpressionValue, StorageRulesRuntime } from "./runtime";
+import { RulesetOperationMethod, RuntimeActionResponse } from "./types";
+
+// Reaches the private stdout handler and pending-request map so we can drive the
+// framing logic directly, without spawning the Java rules runtime.
+type RuntimeInternals = {
+  _requests: Record<number, { request: unknown; handler: (rap: RuntimeActionResponse) => void }>;
+  handleRuntimeStdout(chunk: string): void;
+};
+
+function runtimeWithPendingIds(ids: number[]): {
+  internals: RuntimeInternals;
+  received: number[];
+} {
+  const internals = new StorageRulesRuntime() as unknown as RuntimeInternals;
+  const received: number[] = [];
+  internals._requests = {};
+  for (const id of ids) {
+    internals._requests[id] = {
+      request: { id },
+      handler: (rap) => received.push(rap.id ?? -1),
+    };
+  }
+  return { internals, received };
+}
 
 describe("Storage Rules Runtime", () => {
   describe("createAuthExpressionValue", () => {
@@ -45,4 +68,38 @@ describe("Storage Rules Runtime", () => {
       expect(result.map_value?.fields.token).to.exist;
     });
   });
+
+  describe("handleRuntimeStdout", () => {
+    it("dispatches every response when several arrive in a single chunk", () => {
+      // Regression test for #6194 / #6865. Reverting to a per-chunk JSON.parse
+      // makes this fail: the concatenated responses throw, are swallowed, and
+      // every request in the batch is dropped (and hangs).
+      const { internals, received } = runtimeWithPendingIds([1, 2, 3]);
+
+      const chunk = [1, 2, 3].map((id) => `{"id":${id},"status":"ok"}`).join("\n") + "\n";
+      internals.handleRuntimeStdout(chunk);
+
+      expect(received).to.deep.equal([1, 2, 3]);
+    });
+
+    it("reassembles a response split across two chunks", () => {
+      const { internals, received } = runtimeWithPendingIds([7]);
+
+      internals.handleRuntimeStdout(`{"id":7,"stat`);
+      expect(received).to.deep.equal([]);
+
+      internals.handleRuntimeStdout(`us":"ok"}\n`);
+      expect(received).to.deep.equal([7]);
+    });
+
+    it("ignores blank lines and buffers the trailing partial line", () => {
+      const { internals, received } = runtimeWithPendingIds([1, 2]);
+
+      internals.handleRuntimeStdout(`\n{"id":1,"status":"ok"}\n{"id":2,"stat`);
+      expect(received).to.deep.equal([1]);
+
+      internals.handleRuntimeStdout(`us":"ok"}\n`);
+      expect(received).to.deep.equal([1, 2]);
+    });
+  });
 });

src/emulator/storage/rules/runtime.ts

@@ -62,7 +62,7 @@
     issues: StorageRulesIssues;
   }> {
     if (opts.method === RulesetOperationMethod.LIST && this.rulesVersion < 2) {
       const issues = new StorageRulesIssues();
       issues.warnings.push(
         "Permission denied. List operations are only allowed for rules_version='2'.",
       );
@@ -75,7 +75,7 @@
     return this.runtime.verifyWithRuleset(this.rulesetName, opts, runtimeVariableOverrides);
   }
 
   unload() {
     throw new Error("NOT_IMPLEMENTED");
   }
 }
@@ -86,11 +86,11 @@
     public warnings: string[] = [],
   ) {}
 
   static fromResponse(resp: RuntimeActionResponse) {
     return new StorageRulesIssues(resp.errors || [], resp.warnings || []);
   }
 
   get all() {
     return [...this.errors, ...this.warnings];
   }
 
@@ -109,18 +109,21 @@
   private _requestCount = 0;
   private _requests: {
     [s: number]: {
       handler: (rap: any) => void;
       request: RuntimeActionRequest;
     };
   } = {};
   private _childprocess?: ChildProcess;
   private _alive = false;
+  // Holds the incomplete trailing line of the runtime's stdout between "data"
+  // events. See handleRuntimeStdout().
+  private _stdoutBuffer = "";
 
   get alive() {
     return this._alive;
   }
 
   async start(autoDownload = true) {
     if (this.alive) {
       return;
     }
@@ -167,7 +170,7 @@
     });
 
     // This catches error when spawning the java process
     this._childprocess.on("error", (err: any) => {
       void handleEmulatorProcessError(Emulators.STORAGE, err);
     });
 
@@ -188,42 +191,64 @@
     });
 
     this._childprocess.stdout?.on("data", (buf: Buffer) => {
-      const serializedRuntimeActionResponse = buf.toString("utf-8").trim();
-      if (serializedRuntimeActionResponse !== "") {
-        let rap;
-        try {
-          rap = JSON.parse(serializedRuntimeActionResponse) as RuntimeActionResponse;
-        } catch (err: unknown) {
-          EmulatorLogger.forEmulator(Emulators.STORAGE).log(
-            "INFO",
-            serializedRuntimeActionResponse,
-          );
-          return;
-        }
+      this.handleRuntimeStdout(buf.toString("utf-8"));
+    });
 
-        const id = rap.id ?? rap.server_request_id;
-        if (id === undefined) {
-          console.log(`Received no ID from server response ${serializedRuntimeActionResponse}`);
-          return;
-        }
+    return startPromise;
+  }
 
-        const request = this._requests[id];
+  /**
+   * Dispatches a chunk of the rules runtime's stdout to the awaiting requests.
+   *
+   * The runtime writes one JSON response per line. Node stream "data" events do
+   * not respect message boundaries: under concurrent load several responses
+   * arrive in a single chunk, and a single response can be split across chunks.
+   * So we buffer the incomplete trailing line and only parse complete,
+   * newline-delimited lines. Parsing a raw chunk instead would throw on any
+   * batched responses and silently drop them, hanging every request in the
+   * batch — the root cause of #6194 and #6865.
+   */
+  private handleRuntimeStdout(chunk: string): void {
+    this._stdoutBuffer += chunk;
+    const lines = this._stdoutBuffer.split("\n");
+    // The last element is the incomplete trailing line ("" if the chunk ended
+    // on a newline); keep it buffered until its terminator arrives.
+    this._stdoutBuffer = lines.pop() ?? "";
+
+    for (const rawLine of lines) {
+      const line = rawLine.trim();
+      if (line === "") {
+        continue;
+      }
 
-        if (rap.status !== "ok" && !("action" in rap)) {
-          console.warn(`[RULES] ${rap.status}: ${rap.message}`);
-          rap.errors.forEach(console.warn.bind(console));
-          return;
-        }
+      let rap;
+      try {
+        rap = JSON.parse(line) as RuntimeActionResponse;
+      } catch (err: unknown) {
+        EmulatorLogger.forEmulator(Emulators.STORAGE).log("INFO", line);
+        continue;
+      }
 
-        if (request) {
-          request.handler(rap);
-        } else {
-          console.log(`No handler for event ${serializedRuntimeActionResponse}`);
-        }
+      const id = rap.id ?? rap.server_request_id;
+      if (id === undefined) {
+        console.log(`Received no ID from server response ${line}`);
+        continue;
       }
-    });
 
-    return startPromise;
+      const request = this._requests[id];
+
+      if (rap.status !== "ok" && !("action" in rap)) {
+        console.warn(`[RULES] ${rap.status}: ${rap.message}`);
+        rap.errors.forEach(console.warn.bind(console));
+        continue;
+      }
+
+      if (request) {
+        request.handler(rap);
+      } else {
+        console.log(`No handler for event ${line}`);
+      }
+    }
   }
 
   stop(): Promise<void> {
@@ -263,22 +288,29 @@
     }
 
     return new Promise<RuntimeActionResponse>((resolve) => {
-      this._requests[runtimeActionRequest.id] = {
+      const requestId = runtimeActionRequest.id;
+      this._requests[requestId] = {
         request: runtimeActionRequest,
-        handler: resolve,
+        handler: (rap: RuntimeActionResponse) => {
+          // Free the pending-request entry on completion. Previously entries
+          // were only deleted on the firestore cross-service override path,
+          // leaking one entry per request otherwise.
+          delete this._requests[requestId];
+          resolve(rap);
+        },
       };
 
       const serializedRequest = JSON.stringify(runtimeActionRequest);
 
-      // Added due to https://github.com/firebase/firebase-tools/issues/3915
-      // Without waiting to acquire the lock and allowing the child process enough time
-      // (~15ms) to pipe the output back, the emulator will run into issues with
-      // capturing the output and resolving corresponding promises en masse.
+      // The ~15ms delay that used to sit here (added for #3915) was a workaround
+      // for the stdout framing bug: it slowed request writes so responses were
+      // less likely to batch into a single "data" event. Now that stdout is
+      // framed on newlines (see the handler in start()), the delay is
+      // unnecessary and only slows every request, so release the lock as soon
+      // as the write is queued.
       lock.acquire(synchonizationKey, (done) => {
         this._childprocess?.stdin?.write(serializedRequest + "\n");
-        setTimeout(() => {
-          done();
-        }, 15);
+        done();
       });
     });
   }