From bf42467e215b20b36ec6b4bf30212e4beb2dd01f Mon Sep 17 00:00:00 2001
From: Yoshiya Hinosawa <stibium121@gmail.com>
Date: Thu, 23 Nov 2023 09:57:05 +0900
Subject: [PATCH] fix(ext/node): fix node:stream.Writable (#21297)

This change applies the same fix as
https://github.com/nodejs/node/pull/46818, and the original example
given in #20456 works as expected.

closes #20456
---
 cli/tests/unit_node/stream_test.ts | 37 +++++++++++++++++++++++++++++-
 ext/node/polyfills/_stream.mjs     |  6 +++--
 2 files changed, 40 insertions(+), 3 deletions(-)

diff --git a/cli/tests/unit_node/stream_test.ts b/cli/tests/unit_node/stream_test.ts
index 058d3ca7c6e412..e63171df7b6ab1 100644
--- a/cli/tests/unit_node/stream_test.ts
+++ b/cli/tests/unit_node/stream_test.ts
@@ -1,8 +1,9 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
 
-import { assert } from "../../../test_util/std/testing/asserts.ts";
+import { assert, fail } from "../../../test_util/std/testing/asserts.ts";
 import { fromFileUrl, relative } from "../../../test_util/std/path/mod.ts";
 import { pipeline } from "node:stream/promises";
+import { Writable } from "node:stream";
 import { createReadStream, createWriteStream } from "node:fs";
 
 Deno.test("stream/promises pipeline", async () => {
@@ -23,3 +24,37 @@ Deno.test("stream/promises pipeline", async () => {
     // pass
   }
 });
+
+// TODO(kt3k): Remove this test case when the node compat test suite is
+// updated to version 18.16.0 or above.
+// The last case in parallel/test-stream2-transform.js covers this case.
+// See https://github.com/nodejs/node/pull/46818
+Deno.test("stream.Writable does not change the order of items", async () => {
+  async function test() {
+    const chunks: Uint8Array[] = [];
+    const writable = new Writable({
+      construct(cb) {
+        setTimeout(cb, 10);
+      },
+      write(chunk, _, cb) {
+        chunks.push(chunk);
+        cb();
+      },
+    });
+
+    for (const i of Array(20).keys()) {
+      writable.write(Uint8Array.from([i]));
+      await new Promise((resolve) => setTimeout(resolve, 1));
+    }
+
+    if (chunks[0][0] !== 0) {
+      // The first chunk is swapped with the later chunk.
+      fail("The first chunk is swapped");
+    }
+  }
+
+  for (const _ of Array(10)) {
+    // Run it multiple times to avoid flaky false negative.
+    await test();
+  }
+});
diff --git a/ext/node/polyfills/_stream.mjs b/ext/node/polyfills/_stream.mjs
index 26d5fd30a32c43..23df11ab3dd11f 100644
--- a/ext/node/polyfills/_stream.mjs
+++ b/ext/node/polyfills/_stream.mjs
@@ -1669,9 +1669,11 @@ var require_destroy = __commonJS({
         }
       }
       try {
-        stream._construct(onConstruct);
+        stream._construct((err) => {
+          nextTick(onConstruct, err);
+        });
       } catch (err) {
-        onConstruct(err);
+        nextTick(onConstruct, err);
       }
     }
     function emitConstructNT(stream) {