refactor(process): own worker frame wait lifecycle

src/browser/WebBundleAnalysisWorker.ts

@@ -1,3 +1,4 @@
+import { WorkerFrameSignal } from "../process/WorkerFrameSignal.js";
 import { createHash, randomUUID } from "node:crypto";
 import { open, rm } from "node:fs/promises";
 import { join } from "node:path";
@@ -139,7 +140,7 @@ export class WebBundleAnalysisWorker {
   #activeDeadline: ProviderStartupDeadline | undefined;
   #activeJob: WebBundleActiveJob | undefined;
   #activeTerminalResponseObserved = false;
-  readonly #frameWaiters = new Set<(outcome: "frame" | "failure") => void>();
+  readonly #frameSignal = new WorkerFrameSignal();
   #retainedResultBytes = 0;
   #ready = false;
   readonly #stdout = (chunk: Buffer | string): void => {
@@ -172,12 +173,12 @@ export class WebBundleAnalysisWorker {
         this.#observeFrame(this.#frames.at(-1)!);
         this.#pending = this.#pending.slice(end + 1);
       }
-      this.#wakeFrameWaiters("frame");
+      this.#frameSignal.wake("frame");
     } catch (cause) {
       this.#pending = "";
       this.#protocolFailure = cause;
       this.#latchFailure({ kind: "protocol", cause });
-      this.#wakeFrameWaiters("failure");
+      this.#frameSignal.wake("failure");
     }
   };
 
@@ -200,7 +201,7 @@ export class WebBundleAnalysisWorker {
       onInputFailure: (cause) => {
         if (this.#activeDeadline?.interruption !== undefined) return;
         this.#latchFailure({ kind: "input", cause });
-        this.#wakeFrameWaiters("failure");
+        this.#frameSignal.wake("failure");
       },
       supervisorOptions: {
         captureStdout: false,
@@ -232,7 +233,7 @@ export class WebBundleAnalysisWorker {
                 !this.#activeTerminalResponseObserved)
             )
               this.#latchFailure({ kind: "incomplete-output" });
-            this.#wakeFrameWaiters("failure");
+            this.#frameSignal.wake("failure");
           } else if (event.type === "output") {
             const failure = this.#observedFailure;
             if (failure?.kind !== "exit") return;
@@ -780,41 +781,12 @@ export class WebBundleAnalysisWorker {
     };
   }
 
-  #wakeFrameWaiters(outcome: "frame" | "failure"): void {
-    for (const wake of this.#frameWaiters) wake(outcome);
-    this.#frameWaiters.clear();
-  }
-
   async #waitForFrameOrOutputClose(
     supervisor: NonNullable<OwnedWorkerRuntime["supervisor"]>,
     deadline: ProviderStartupDeadline,
   ): Promise<"frame" | "failure" | "closed" | "interrupted"> {
     if (this.#frames.length > 0) return "frame";
-    let wakeFrame!: (outcome: "frame" | "failure") => void;
-    const frameAvailable = new Promise<"frame" | "failure">((resolve) => {
-      wakeFrame = resolve;
-      this.#frameWaiters.add(wakeFrame);
-    });
-    const outputCloseController = new AbortController();
-    const interruptOutputClose = () =>
-      outputCloseController.abort(deadline.signal.reason);
-    if (deadline.signal.aborted) interruptOutputClose();
-    else
-      deadline.signal.addEventListener("abort", interruptOutputClose, {
-        once: true,
-      });
-    try {
-      const outputClose = supervisor
-        .waitForOutputClose(undefined, outputCloseController.signal)
-        .then((closed): "closed" | "interrupted" =>
-          closed ? "closed" : "interrupted",
-        );
-      return await Promise.race([frameAvailable, outputClose]);
-    } finally {
-      outputCloseController.abort();
-      deadline.signal.removeEventListener("abort", interruptOutputClose);
-      this.#frameWaiters.delete(wakeFrame);
-    }
+    return this.#frameSignal.waitForOutputClose(supervisor, deadline.signal);
   }
 
   #completedSnapshot(cause: unknown): StableArtifactSnapshot | undefined {
@@ -875,7 +847,7 @@ export class WebBundleAnalysisWorker {
     this.#activeDeadline = undefined;
     this.#activeJob = undefined;
     this.#activeTerminalResponseObserved = false;
-    this.#wakeFrameWaiters("failure");
+    this.#frameSignal.wake("failure");
   }
 
   #cleanupFailure(cause: unknown, primary?: AnalysisError): AnalysisError {

src/javascript/analysis/JavaScriptAnalysisWorker.ts

@@ -1,3 +1,4 @@
+import { WorkerFrameSignal } from "../../process/WorkerFrameSignal.js";
 import type { JavaScriptSemanticProjectionCapacity } from "../../domain/javascript/javascriptSemanticFileProjection.js";
 import { createHash, randomUUID } from "node:crypto";
 import { readFile, rm, stat, writeFile } from "node:fs/promises";
@@ -100,7 +101,7 @@ export class JavaScriptAnalysisWorker {
   #ready = false;
   #heapLimitBytes = 0;
   #frames: JavaScriptWorkerMessage[] = [];
-  #frameWaiters = new Set<(outcome: "frame" | "failure") => void>();
+  readonly #frameSignal = new WorkerFrameSignal();
   #pending = "";
   #decoder = new StringDecoder("utf8");
   #primaryFailure: WorkerPrimaryFailure | undefined;
@@ -141,7 +142,7 @@ export class JavaScriptAnalysisWorker {
         const frame = javaScriptWorkerMessageSchema.parse(raw);
         this.#frames.push(frame);
         this.#observeJobFrame(frame);
-        this.#wakeFrameWaiters("frame");
+        this.#frameSignal.wake("frame");
       }
       if (Buffer.byteLength(this.#pending) > DEFAULTS.protocolBytes)
         this.#latchOutputFailure(
@@ -837,7 +838,7 @@ export class JavaScriptAnalysisWorker {
       )
     ) {
       this.#primaryFailure = { ...this.#primaryFailure, heapExhaustion: true };
-      this.#wakeFrameWaiters("failure");
+      this.#frameSignal.wake("failure");
     }
     if (event.type === "exit" && (event.code !== 0 || event.signal !== null)) {
       if (this.#primaryFailure === undefined)
@@ -860,7 +861,7 @@ export class JavaScriptAnalysisWorker {
           ...this.#primaryFailure,
           heapExhaustion: true,
         };
-        this.#wakeFrameWaiters("failure");
+        this.#frameSignal.wake("failure");
       }
     }
   }
@@ -880,7 +881,7 @@ export class JavaScriptAnalysisWorker {
         this.#primaryFailure?.kind === "memory" ||
         this.#primaryFailure?.kind === "exit"
       )
-        this.#wakeFrameWaiters("failure");
+        this.#frameSignal.wake("failure");
     }
   }
   #observeJobFrame(frame: JavaScriptWorkerMessage): void {
@@ -906,7 +907,7 @@ export class JavaScriptAnalysisWorker {
             kind: "output",
             reason: "JavaScript worker completed before a static checkpoint",
           };
-        this.#wakeFrameWaiters("failure");
+        this.#frameSignal.wake("failure");
         return;
       }
       this.#terminalResponseId = frame.id;
@@ -925,48 +926,20 @@ export class JavaScriptAnalysisWorker {
     if (this.#deadline?.interruption !== undefined) return;
     if (this.#primaryFailure === undefined)
       this.#primaryFailure = { kind: "output", reason };
-    this.#wakeFrameWaiters("failure");
+    this.#frameSignal.wake("failure");
   }
   #latchInputFailure(cause: Error): void {
     if (this.#deadline?.interruption !== undefined) return;
     if (this.#primaryFailure === undefined)
       this.#primaryFailure = { kind: "input", cause };
-    this.#wakeFrameWaiters("failure");
+    this.#frameSignal.wake("failure");
   }
   async #waitForFrameOrOutputClose(
     supervisor: ProviderProcessSupervisor,
     deadline: ProviderStartupDeadline,
   ): Promise<"frame" | "failure" | "closed" | "interrupted"> {
     if (this.#frames.length !== 0) return "frame";
-    let wakeFrame!: (outcome: "frame" | "failure") => void;
-    const frameArrival = new Promise<"frame" | "failure">((resolve) => {
-      wakeFrame = resolve;
-      this.#frameWaiters.add(wakeFrame);
-    });
-    const outputCloseController = new AbortController();
-    const interruptOutputClose = () =>
-      outputCloseController.abort(deadline.signal.reason);
-    if (deadline.signal.aborted) interruptOutputClose();
-    else
-      deadline.signal.addEventListener("abort", interruptOutputClose, {
-        once: true,
-      });
-    try {
-      const outputClose = supervisor
-        .waitForOutputClose(undefined, outputCloseController.signal)
-        .then((closed): "closed" | "interrupted" =>
-          closed ? "closed" : "interrupted",
-        );
-      return await Promise.race([frameArrival, outputClose]);
-    } finally {
-      outputCloseController.abort();
-      deadline.signal.removeEventListener("abort", interruptOutputClose);
-      this.#frameWaiters.delete(wakeFrame);
-    }
-  }
-  #wakeFrameWaiters(outcome: "frame" | "failure"): void {
-    for (const wake of this.#frameWaiters) wake(outcome);
-    this.#frameWaiters.clear();
+    return this.#frameSignal.waitForOutputClose(supervisor, deadline.signal);
   }
   #selectedWorkerFailure(path: string): AnalysisError {
     if (this.#primaryFailure !== undefined)
@@ -1231,7 +1204,7 @@ export class JavaScriptAnalysisWorker {
   #resetState(): void {
     this.#ready = false;
     this.#frames = [];
-    this.#frameWaiters.clear();
+    this.#frameSignal.clear();
     this.#pending = "";
     this.#decoder = new StringDecoder("utf8");
     this.#primaryFailure = undefined;

src/process/WorkerFrameSignal.ts

@@ -0,0 +1,45 @@
+import type { ProviderProcessSupervisor } from "./ProviderProcess.js";
+
+type FrameOutcome = "frame" | "failure";
+
+/** Own frame waiters and release the losing output-close wait after each race. */
+export class WorkerFrameSignal {
+  readonly #waiters = new Set<(outcome: FrameOutcome) => void>();
+
+  wake(outcome: FrameOutcome): void {
+    for (const wake of this.#waiters) wake(outcome);
+    this.clear();
+  }
+
+  clear(): void {
+    this.#waiters.clear();
+  }
+
+  async waitForOutputClose(
+    supervisor: ProviderProcessSupervisor,
+    signal: AbortSignal,
+  ): Promise<FrameOutcome | "closed" | "interrupted"> {
+    let wakeFrame!: (outcome: FrameOutcome) => void;
+    const frameAvailable = new Promise<FrameOutcome>((resolve) => {
+      wakeFrame = resolve;
+      this.#waiters.add(wakeFrame);
+    });
+    const outputCloseController = new AbortController();
+    const interruptOutputClose = () =>
+      outputCloseController.abort(signal.reason);
+    if (signal.aborted) interruptOutputClose();
+    else signal.addEventListener("abort", interruptOutputClose, { once: true });
+    try {
+      const outputClose = supervisor
+        .waitForOutputClose(undefined, outputCloseController.signal)
+        .then((closed): "closed" | "interrupted" =>
+          closed ? "closed" : "interrupted",
+        );
+      return await Promise.race([frameAvailable, outputClose]);
+    } finally {
+      outputCloseController.abort();
+      signal.removeEventListener("abort", interruptOutputClose);
+      this.#waiters.delete(wakeFrame);
+    }
+  }
+}