fix(windows): keep polling owned process output after a failed poll (#8163)

* fix(windows): keep polling owned process output after a failed poll

* test(windows): pin the failed-poll retry delay

src/windows/WindowsOwnedProcess.test.ts

@@ -509,3 +509,55 @@ describe("Windows output backpressure and cleanup", () => {
     }
   });
 });
+
+describe("Windows output poll failure", () => {
+  it("keeps polling after failed native polls and reports the run once", async () => {
+    vi.useFakeTimers();
+    try {
+      const { native, calls } = authority({
+        polls: [
+          { malformed: true },
+          { malformed: true },
+          {
+            stdout: Buffer.from("frame\n"),
+            stderr: Buffer.alloc(0),
+            stdoutEnded: false,
+            stderrEnded: false,
+            exitCode: null,
+            activeProcesses: 1,
+          },
+          { ...settledPoll, stderr: Buffer.alloc(0), exitCode: 0 },
+        ],
+      });
+      const child = launch(native);
+      const errors: unknown[] = [];
+      child.on("error", (error) => errors.push(error));
+      const output: Buffer[] = [];
+      child.stdout.on("data", (chunk: Buffer) => output.push(chunk));
+      let closed = false;
+      child.once("close", () => (closed = true));
+      const polls = () =>
+        calls.filter((call) => call.operation === "process_poll").length;
+      // Failed polls back off 250 ms instead of retrying at the idle cadence.
+      await vi.advanceTimersByTimeAsync(25);
+      expect(polls()).toBe(1);
+      await vi.advanceTimersByTimeAsync(249);
+      expect(polls()).toBe(1);
+      await vi.advanceTimersByTimeAsync(1);
+      expect(polls()).toBe(2);
+      await vi.advanceTimersByTimeAsync(249);
+      expect(polls()).toBe(2);
+      await vi.advanceTimersByTimeAsync(1);
+      expect(polls()).toBe(3);
+      // Output resumes the immediate cadence (Node clamps a 0 ms timer to 1 ms).
+      await vi.advanceTimersByTimeAsync(1);
+      expect(polls()).toBe(4);
+      expect(errors).toHaveLength(1);
+      expect(Buffer.concat(output).toString()).toBe("frame\n");
+      expect(child.exitCode).toBe(0);
+      expect(closed).toBe(true);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});

src/windows/WindowsOwnedProcess.ts

@@ -73,6 +73,9 @@ export const quoteWindowsProcessArgument = (value: string): string => {
   return `"${value.replace(/(\\*)"/gu, '$1$1\\"').replace(/(\\+)$/u, "$1$1")}"`;
 };
 
+// Back off after a failed native poll instead of retrying at the idle cadence.
+const POLL_RETRY_MS = 250;
+
 /** Process events and streams backed by a retained native Job Object. */
 export class WindowsOwnedProcess extends EventEmitter {
   readonly stdout = new PassThrough();
@@ -399,12 +402,18 @@ export class WindowsOwnedProcess extends EventEmitter {
       } else this.#schedule(idle ? 25 : 0);
     } catch (cause: unknown) {
       const failure = cause instanceof Error ? cause : new Error(String(cause));
+      const repeated = this.#failure !== undefined;
       this.#failure = failure.message;
       // Native lifecycle failures belong to the error/cleanup channels, never
-      // the captured stderr observation produced by the child itself.
-      if (this.listenerCount("error") > 0) this.emit("error", failure);
-      // Poll failures keep the job for another observation. A process_close
-      // failure leaves #releasePending so cleanup retries that close alone.
+      // the captured stderr observation produced by the child itself. A run of
+      // consecutive failures is reported once.
+      if (!repeated && this.listenerCount("error") > 0)
+        this.emit("error", failure);
+      // Poll failures keep the job for another observation. Without one, a
+      // single failed poll would stop output and exit delivery for the rest of
+      // the process lifetime. A process_close failure leaves #releasePending so
+      // cleanup retries that close alone.
+      this.#schedule(POLL_RETRY_MS);
     }
   }