fix(ext/node): fix node:stream.Writable (#21297)

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

closes #20456

Author: kt3k

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();
+  }
+});

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) {