Expand writable suite: construction, abort matrix, queue math, reentrancy

Author: jasnell

src/tests/streams/writable/abort-matrix.js

@@ -0,0 +1,193 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the Apache 2.0 license found in the LICENSE file or at:
+//     https://opensource.org/licenses/Apache-2.0
+
+// Abort matrix: reason identity across writer promises, sink-hook
+// suppression, in-flight interactions with controller.error(), and the
+// controller.signal surface. WPT writable-streams/aborting.any.js lists
+// eleven C++ expectedFailures ("nitpickiness about the type of error");
+// the scenarios here probe the same territory and pin what each
+// implementation actually does — most of it is parity, with the
+// signal-reason default the notable divergence.
+
+import { strictEqual, ok, rejects } from 'node:assert';
+import { usingTsImpl, pedanticWpt } from 'which-impl';
+
+// Aborting before start() settles rejects ready and closed with the
+// abort reason (by identity); the abort promise fulfills.
+export const abortBeforeStartReasonIdentity = {
+  async test() {
+    const ws = new WritableStream({
+      async start() {
+        await scheduler.wait(5);
+      },
+    });
+    const writer = ws.getWriter();
+    const reason = new Error('r1');
+    const abort = writer.abort(reason);
+
+    await rejects(writer.ready, (e) => e === reason);
+    await rejects(writer.closed, (e) => e === reason);
+    strictEqual(await abort, undefined);
+  },
+};
+
+// After an abort settles, writes reject with the very same reason object.
+export const erroredStateReasonIdentity = {
+  async test() {
+    const ws = new WritableStream();
+    const writer = ws.getWriter();
+    const reason = { custom: true };
+    await writer.abort(reason);
+    await rejects(writer.write('x'), (e) => e === reason);
+  },
+};
+
+// A stream errored by a throwing size() does not run the sink abort hook
+// when abort() arrives later; the abort still fulfills (parity; cf. the
+// WPT "sink abort() should not be called if stream was erroring due to
+// bad strategy" case).
+export const sinkAbortSkippedAfterBadStrategyError = {
+  async test() {
+    let abortCalled = false;
+    const ws = new WritableStream(
+      {
+        abort() {
+          abortCalled = true;
+        },
+      },
+      {
+        size() {
+          throw new Error('bad size');
+        },
+        highWaterMark: 1,
+      }
+    );
+    const writer = ws.getWriter();
+    await rejects(writer.write('x'), { message: 'bad size' });
+    strictEqual(await writer.abort('reason'), undefined);
+    strictEqual(abortCalled, false);
+  },
+};
+
+// writer.abort() while a write is in flight, with the sink write then
+// REJECTING: the write surfaces its own failure, the abort fulfills, and
+// closed carries the abort reason (parity).
+export const inFlightWriteRejectionDuringAbort = {
+  async test() {
+    let rejectWrite;
+    const ws = new WritableStream({
+      write() {
+        return new Promise((res, rej) => (rejectWrite = rej));
+      },
+    });
+    const writer = ws.getWriter();
+    await writer.ready;
+    const write = writer.write('x');
+    await scheduler.wait(1); // the sink write is now in flight on both sides
+    const abortReason = new Error('abort-reason');
+    const abort = writer.abort(abortReason);
+    await scheduler.wait(1);
+    rejectWrite(new Error('write-fail'));
+
+    await rejects(write, { message: 'write-fail' });
+    strictEqual(await abort, undefined);
+    await rejects(writer.closed, (e) => e === abortReason);
+  },
+};
+
+// writer.abort() then controller.error() while a write is in flight,
+// with the write finishing cleanly: the abort request wins, both write
+// and abort fulfill (parity).
+export const abortThenControllerErrorInFlight = {
+  async test() {
+    let resolveWrite;
+    let controller;
+    const ws = new WritableStream({
+      start(c) {
+        controller = c;
+      },
+      write() {
+        return new Promise((res) => (resolveWrite = res));
+      },
+    });
+    const writer = ws.getWriter();
+    await writer.ready;
+    const write = writer.write('x');
+    await scheduler.wait(1);
+    const abort = writer.abort(new Error('abort-reason'));
+    controller.error(new Error('ctrl-error'));
+    await scheduler.wait(1);
+    resolveWrite();
+
+    strictEqual(await write, undefined);
+    strictEqual(await abort, undefined);
+  },
+};
+
+// controller.error() then writer.abort() while a write is in flight: the
+// stream is already erroring, so the abort rejects with the
+// controller's error while the in-flight write still finishes (parity).
+export const controllerErrorThenAbortInFlight = {
+  async test() {
+    let resolveWrite;
+    let controller;
+    const ws = new WritableStream({
+      start(c) {
+        controller = c;
+      },
+      write() {
+        return new Promise((res) => (resolveWrite = res));
+      },
+    });
+    const writer = ws.getWriter();
+    await writer.ready;
+    const write = writer.write('x');
+    await scheduler.wait(1);
+    controller.error(new Error('ctrl-error'));
+    const abort = writer.abort(new Error('abort-reason'));
+    await scheduler.wait(1);
+    resolveWrite();
+
+    strictEqual(await write, undefined);
+    await rejects(abort, { message: 'ctrl-error' });
+  },
+};
+
+// DIVERGENCE: the signal reason for a reasonless abort(). The spec (and
+// TypeScript, and C++ under pedantic_wpt — standard.c++ WritableImpl::
+// abort) synthesizes an AbortError DOMException; C++ otherwise leaves
+// the reason undefined. An explicit reason is passed through verbatim
+// everywhere.
+export const abortSignalReason = {
+  async test() {
+    {
+      let controller;
+      const ws = new WritableStream({
+        start(c) {
+          controller = c;
+        },
+      });
+      ws.abort();
+      await scheduler.wait(1);
+      strictEqual(controller.signal.aborted, true);
+      if (usingTsImpl || pedanticWpt) {
+        ok(controller.signal.reason instanceof DOMException);
+        strictEqual(controller.signal.reason.name, 'AbortError');
+      } else {
+        strictEqual(controller.signal.reason, undefined);
+      }
+    }
+    {
+      let controller;
+      const ws = new WritableStream({
+        start(c) {
+          controller = c;
+        },
+      });
+      ws.abort('why');
+      await scheduler.wait(1);
+      strictEqual(controller.signal.reason, 'why');
+    }
+  },
+};

src/tests/streams/writable/backpressure.js

@@ -3,9 +3,22 @@
 //     https://opensource.org/licenses/Apache-2.0
 
 // desiredSize accounting and ready-promise behavior under backpressure.
-// Migrated from streams-js-test.js and streams-backpressure-test.js.
-
-import { strictEqual, ok } from 'node:assert';
+// Migrated from streams-js-test.js and streams-backpressure-test.js, plus
+// the floating-point queue-total pins mirroring WPT
+// writable-streams/floating-point-total-queue-size.any.js.
+//
+// DIVERGENCE (queue arithmetic): the spec tracks [[queueTotalSize]] as a
+// double. The C++ implementation converts each size() result to uint64
+// through the jsg boundary (fractions truncate toward zero; negatives,
+// NaN and infinities throw TypeError — common.h StreamQueuingStrategy)
+// and then narrows the ssize_t desiredSize through `int`, so totals past
+// 2^31 wrap (WritableStreamJsController::getDesiredSize). These are the
+// three WPT floating-point expectedFailures. TypeScript follows the spec
+// exactly, including RangeError("Invalid chunk size") for invalid size
+// returns.
+
+import { strictEqual, ok, rejects } from 'node:assert';
+import { usingTsImpl, pedanticWpt } from 'which-impl';
 
 // desiredSize decrements per queued write and recovers; ready is replaced
 // once the queue drains.
@@ -143,3 +156,146 @@ export const backpressureWritableSlowSink = {
     await writer.close();
   },
 };
+
+// Builds the WPT floating-point scenario: identity size(), highWaterMark
+// 0, and a gated sink so queued chunks stay queued while desiredSize is
+// read.
+function gatedFpStream() {
+  let release;
+  const gate = new Promise((res) => (release = res));
+  const ws = new WritableStream(
+    {
+      async write() {
+        await gate;
+      },
+    },
+    {
+      size(x) {
+        return x;
+      },
+      highWaterMark: 0,
+    }
+  );
+  return { writer: ws.getWriter(), release };
+}
+
+// The four WPT floating-point queue-total scenarios, with each side's
+// exact arithmetic pinned.
+export const floatingPointQueueTotals = {
+  async test() {
+    // Chunk sizes, expected desiredSize while queued (ts: double math;
+    // cpp: uint64-truncated sizes, so 1e-16 and 2e-16 count as 0), and
+    // expected desiredSize after the queue drains.
+    const cases = [
+      {
+        sizes: [2, Number.MAX_SAFE_INTEGER],
+        // ts: -(2 + MAX_SAFE_INTEGER) rounded in double arithmetic; cpp:
+        // the true total exceeds 2^31 and the int narrowing wraps to -1.
+        during: usingTsImpl ? 0 - 2 - Number.MAX_SAFE_INTEGER : -1,
+        after: 0,
+      },
+      {
+        sizes: [1e-16, 1],
+        during: usingTsImpl ? 0 - 1e-16 - 1 : -1,
+        after: 0,
+      },
+      {
+        sizes: [1e-16, 1, 2e-16],
+        during: usingTsImpl ? 0 - 1e-16 - 1 - 2e-16 : -1,
+        // ts: the spec's incremental subtraction leaves a residue that is
+        // not clamped (the WPT "positive, and not clamped" case).
+        after: usingTsImpl ? 0 - 1e-16 - 1 - 2e-16 + 1e-16 + 1 + 2e-16 : 0,
+      },
+      {
+        sizes: [2e-16, 1],
+        during: usingTsImpl ? 0 - 2e-16 - 1 : -1,
+        after: 0,
+      },
+    ];
+
+    for (const { sizes, during, after } of cases) {
+      const { writer, release } = gatedFpStream();
+      const writes = sizes.map((v) => writer.write(v));
+      await scheduler.wait(1); // let the TypeScript side start the sink
+      strictEqual(writer.desiredSize, during);
+      release();
+      await Promise.all(writes);
+      strictEqual(writer.desiredSize, after);
+    }
+  },
+};
+
+// A lone fractional chunk size: counted faithfully by TypeScript,
+// truncated to zero by the C++ uint64 conversion.
+export const fractionalSizeTruncation = {
+  async test() {
+    const { writer, release } = gatedFpStream();
+    const write = writer.write(0.5);
+    await scheduler.wait(1);
+    strictEqual(writer.desiredSize, usingTsImpl ? -0.5 : 0);
+    release();
+    await write;
+    strictEqual(writer.desiredSize, 0);
+  },
+};
+
+// Invalid size() return values reject the write and error the stream;
+// the error type diverges (C++ TypeError from the uint64 conversion with
+// value-specific messages, TypeScript the spec's RangeError).
+export const invalidSizeReturnRejects = {
+  async test() {
+    const cases = [
+      { ret: NaN, cppMessage: /not an integer/ },
+      { ret: -1, cppMessage: /negative/ },
+      { ret: Infinity, cppMessage: /not an integer/ },
+    ];
+    for (const { ret, cppMessage } of cases) {
+      const ws = new WritableStream(
+        {},
+        {
+          size() {
+            return ret;
+          },
+          highWaterMark: 5,
+        }
+      );
+      const writer = ws.getWriter();
+      const expected = usingTsImpl
+        ? { name: 'RangeError', message: 'Invalid chunk size' }
+        : { name: 'TypeError', message: cppMessage };
+      await rejects(writer.write('x'), expected);
+      // The bad size does not just doom the write; the stream errors.
+      await rejects(writer.closed, expected);
+    }
+  },
+};
+
+// DIVERGENCE: while the stream is ERRORING (in-flight write outstanding,
+// controller.error() already called) the spec reports desiredSize null;
+// TypeScript and C++ with pedantic_wpt do that, while C++ otherwise still
+// reports the queue accounting value. Once fully errored, everyone
+// reports null.
+export const desiredSizeWhileErroring = {
+  async test() {
+    let resolveWrite;
+    let controller;
+    const ws = new WritableStream({
+      start(c) {
+        controller = c;
+      },
+      write() {
+        return new Promise((res) => (resolveWrite = res));
+      },
+    });
+    const writer = ws.getWriter();
+    await writer.ready;
+    const write = writer.write('x');
+    await scheduler.wait(1); // the write is in flight; erroring can begin
+    controller.error(new Error('e'));
+    strictEqual(writer.desiredSize, usingTsImpl || pedanticWpt ? null : 0);
+    resolveWrite();
+    await write;
+    strictEqual(writer.desiredSize, null);
+    await writer.closed.catch(() => {});
+  },
+};