🐛 Make piped xmd syntax output complete, and report a closed sink (#715)

.github/workflows/ci.yml

@@ -99,7 +99,7 @@ jobs:
       # spawn a literal `bun`. The runner image ships Node but not bun.
       - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2
         with:
-          bun-version: 1.3.14
+          bun-version: 1.4.0
 
       - name: Typecheck
         run: deno task check
@@ -445,7 +445,7 @@ jobs:
 
       - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2
         with:
-          bun-version: 1.3.14
+          bun-version: 1.4.0
 
       # `deno install` rather than `deno task deps`: the task also caches the
       # graphs a browser build and a release compile walk, and it reaches them
@@ -534,7 +534,7 @@ jobs:
       # empty spool.
       - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2
         with:
-          bun-version: 1.3.14
+          bun-version: 1.4.0
 
       - uses: denoland/setup-deno@e95548e56dfa95d4e1a28d6f422fafe75c4c26fb # v2.0.3
         with:
@@ -629,7 +629,7 @@ jobs:
 
       - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2
         with:
-          bun-version: 1.3.14
+          bun-version: 1.4.0
 
       - run: bun install
 

.github/workflows/measure-test-weights.yml

@@ -38,7 +38,7 @@ jobs:
 
       - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2
         with:
-          bun-version: 1.3.14
+          bun-version: 1.4.0
 
       # One preparation, in the repository's load-bearing order: `deno install`,
       # then `pnpm install` beside it, then the browser bundle. The measured

packages/cli/src/cli.ts

@@ -130,6 +130,7 @@ import { runUpgrade } from "./upgrade.ts";
 import type { UpgradeAssembly } from "./upgrade.ts";
 import { componentSearchPath, resolveTestTarget } from "./test-target.ts";
 import { renderSyntaxJson, renderSyntaxMarkdown, syntaxCatalog } from "./syntax.ts";
+import { deliverWhole } from "./stdout-delivery.ts";
 import { testingExecutionHost } from "./testing-host.ts";
 import { unsupportedRepositories } from "./run-repositories.ts";
 import type { RepositoryInstaller } from "./run-repositories.ts";
@@ -2429,7 +2430,16 @@ function* dispatch(
         yield* exit(1);
         break;
       }
-      yield* writeStdoutWhole(rendered);
+      // Only the catalog goes through delivery today, because it is the one
+      // output this command writes in a single call and the only one already
+      // past a pipe buffer.
+      const written = yield* deliverWhole(rendered, process.stdout);
+      if (!written.ok) {
+        console.error(
+          `xmd syntax: stdout did not accept the whole catalog: ${describeError(written.error)}`,
+        );
+        yield* exit(1);
+      }
       break;
     }
     case "test-agent":
@@ -2530,27 +2540,6 @@ function* dispatch(
   }
 }
 
-/**
- * Write to stdout and wait for it to reach the operating system.
- *
- * `process.stdout` is asynchronous when it is a pipe and synchronous when it is
- * a file or a terminal. A large document handed to `write` is therefore still
- * sitting in a buffer when the run ends, and the process exits without it:
- * `xmd syntax --json > file` is whole, `xmd syntax --json | jq` stops at about
- * 64 KiB, in the middle of a token. Waiting for the callback is what makes the
- * write finish before anything can exit.
- *
- * Only the catalog goes through this today, because it is the one output this
- * command writes in a single call and the only one already past the buffer.
- */
-function* writeStdoutWhole(text: string): Operation<void> {
-  yield* until(
-    new Promise<void>((resolve, reject) => {
-      process.stdout.write(text, (error) => (error ? reject(error) : resolve()));
-    }),
-  );
-}
-
 export function* runXmd(
   args: string[],
   installService: HostServiceInstaller,

packages/cli/src/stdout-delivery.ts

@@ -0,0 +1,120 @@
+/**
+ * Handing one rendered result to a stream, and waiting for the stream to take
+ * all of it.
+ *
+ * A stream is asynchronous when it is a pipe and synchronous when it is a file
+ * or a terminal. A large text handed to `write` is therefore still sitting in a
+ * buffer when a run ends, and the process exits without it: `xmd syntax --json
+ * > file` is whole, `xmd syntax --json | jq` stops at about 64 KiB, in the
+ * middle of a token. Waiting for the write callback is what makes the write
+ * finish before anything can exit.
+ */
+
+import { Err, Ok, resource, scoped, withResolvers } from "effection";
+import type { Operation, Result } from "effection";
+
+/**
+ * What delivery asks of a stream.
+ *
+ * Narrower than a writable stream so a suite can supply the arrival orders a
+ * real pipe produces on one runtime and not another, and can watch the
+ * listener come and go.
+ */
+export interface DeliverySink {
+  write(text: string, callback: (error?: Error | null) => void): unknown;
+  on(event: "error", listener: (error: Error) => void): unknown;
+  off(event: "error", listener: (error: Error) => void): unknown;
+  /** The error the stream was destroyed with and has not emitted yet. */
+  readonly errored?: Error | null;
+}
+
+/**
+ * The sink's `error` events, observed for exactly as long as the enclosing
+ * scope lives.
+ *
+ * The listener's lifetime is the scope's and nothing else's: it is not removed
+ * by the event firing, and it does not survive the scope whether the event
+ * fired, never fired, or the scope was halted first. A stream is a
+ * process-global object, so a listener bound to anything looser would
+ * accumulate across deliveries and absorb a later, unrelated failure.
+ */
+function useErrorObserver(sink: DeliverySink, observe: (error: Error) => void): Operation<void> {
+  return resource(function* (provide) {
+    sink.on("error", observe);
+    try {
+      yield* provide();
+    } finally {
+      sink.off("error", observe);
+    }
+  });
+}
+
+/**
+ * Deliver `text`, and report whether the stream took all of it.
+ *
+ * A broken pipe can arrive twice: at the write callback, and again as an
+ * `error` event. An `error` event nobody is listening for ends the process with
+ * a stack trace, which is why this listens — and the first arrival is the
+ * verdict, whichever it was, so the duplicate is absorbed rather than reported
+ * a second time.
+ *
+ * **The listener lives for one delivery and no longer.** It survives a first
+ * failed arrival only until the paired one lands, and `errored` is what says
+ * whether one is still owed: Node and Deno destroy the stream with the very
+ * error they are about to emit, so finding this delivery's own failure there
+ * means the event is still coming, while Bun reports the failure once and holds
+ * nothing. Every other path — success, cancellation, a `write` that refuses
+ * outright — ends the scope, and the observer with it, without waiting at all.
+ *
+ * The outcome is bridged with `withResolvers()` rather than `action()`: a file
+ * or a terminal calls the write callback synchronously, inside `write`, and an
+ * `action()` resolved before its executor has returned never runs the cleanup
+ * that executor returns (effection 4.1.0).
+ */
+export function deliverWhole(text: string, sink: DeliverySink): Operation<Result<void>> {
+  return scoped(function* () {
+    const outcome = withResolvers<Result<void>>("deliverWhole");
+    let calledBack = false;
+    let observed = false;
+    let failure: Error | undefined;
+
+    const settle = () => {
+      if (!calledBack) {
+        return;
+      }
+      // Identity, not presence: an error the stream was already holding before
+      // this delivery has been emitted already, and waiting for it would wait
+      // forever.
+      if (!observed && failure !== undefined && sink.errored === failure) {
+        return;
+      }
+      outcome.resolve(failure === undefined ? Ok() : Err(failure));
+    };
+
+    const record = (error?: Error | null) => {
+      if (error && failure === undefined) {
+        failure = error;
+      }
+    };
+
+    yield* useErrorObserver(sink, (error) => {
+      observed = true;
+      record(error);
+      settle();
+    });
+
+    try {
+      sink.write(text, (error) => {
+        calledBack = true;
+        record(error);
+        settle();
+      });
+    } catch (error) {
+      // A stream that refuses the call outright never calls back, so nothing
+      // else would settle this.
+      return Err(error instanceof Error ? error : new Error(String(error)));
+    }
+
+    return yield* outcome.operation;
+  });
+}