fix(node): vendor readable-stream from esm.sh (#2584)

This commit replaces the Node.js streams implementation with
an ESM version of the readable-stream package. The goal is to
reduce maintenance burden while allowing Deno to stay up to
date with changes in the Node.js implementation.

node/_stream.d.ts

@@ -3,6 +3,7 @@
 
 // Forked from https://github.com/DefinitelyTyped/DefinitelyTyped/blob/4f538975138678878fed5b2555c0672aa578ab7d/types/node/stream.d.ts
 
+import { Buffer } from "./_buffer.d.ts";
 import { Abortable, EventEmitter } from "./_events.d.ts";
 import {
   Buffered,
@@ -1477,5 +1478,5 @@ interface Pipe {
 }
 
 // These have to be at the bottom of the file to work correctly, for some reason
-export { _uint8ArrayToBuffer } from "./internal/streams/_utils.ts";
-export { isUint8Array as _isUint8Array } from "./internal/util/types.ts";
+export function _uint8ArrayToBuffer(chunk: Uint8Array): Buffer;
+export function _isUint8Array(value: unknown): value is Uint8Array;

node/_tools/README.md

@@ -0,0 +1,17 @@
+# Node.js Compatibility Tooling
+
+This directory contains tooling for implementing Deno's Node.js compatibility
+layer.
+
+## Updating the Streams Implementation
+
+The Node.js streams implementation is based on the `readable-stream` module on
+npm. To update the code, run the following command:
+
+```
+deno run -A --unstable node/_tools/vendor_readable_stream.ts
+```
+
+At the top of this script, there is a variable named `sourceUrl`. This is the
+implementation that is downloaded, modified, and included in `node/_stream.mjs`.
+To change the vendored version of `readable-stream`, update this URL.

node/_tools/config.json

@@ -50,7 +50,13 @@
       "test-path.js",
       "test-querystring.js",
       "test-readline-interface.js",
+      "test-stream-duplex-destroy.js",
+      "test-stream-readable-with-unimplemented-_read.js",
+      "test-stream-transform-destroy.js",
       "test-stream-writable-change-default-encoding.js",
+      "test-stream-writable-destroy.js",
+      "test-stream-writable-end-cb-error.js",
+      "test-stream-write-destroy.js",
       "test-url-urltooptions.js",
       "test-util-format.js",
       "test-util-inspect-namespace.js",

node/_tools/test/parallel/test-stream-duplex-destroy.js

@@ -5,6 +5,10 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
+
 'use strict';
 
 const common = require('../common');
@@ -132,7 +136,7 @@ const assert = require('assert');
   duplex.removeListener('end', fail);
   duplex.removeListener('finish', fail);
   duplex.on('end', common.mustNotCall());
-  duplex.on('finish', common.mustCall());
+  duplex.on('finish', common.mustNotCall());
   assert.strictEqual(duplex.destroyed, true);
 }
 

node/_tools/test/parallel/test-stream-readable-with-unimplemented-_read.js

@@ -5,21 +5,20 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
-'use strict';
-require('../common');
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
 
-const assert = require('assert');
+'use strict';
+const common = require('../common');
 const { Readable } = require('stream');
 
 const readable = new Readable();
 
-assert.throws(
-  () => {
-    readable.read();
-  },
-  {
-    code: 'ERR_METHOD_NOT_IMPLEMENTED',
-    name: 'Error',
-    message: 'The _read() method is not implemented'
-  }
-);
+readable.read();
+readable.on('error', common.expectsError({
+  code: 'ERR_METHOD_NOT_IMPLEMENTED',
+  name: 'Error',
+  message: 'The _read() method is not implemented'
+}));
+readable.on('close', common.mustCall());

node/_tools/test/parallel/test-stream-transform-destroy.js

@@ -5,6 +5,10 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
+
 'use strict';
 
 const common = require('../common');
@@ -124,7 +128,7 @@ const assert = require('assert');
   transform.removeListener('end', fail);
   transform.removeListener('finish', fail);
   transform.on('end', common.mustCall());
-  transform.on('finish', common.mustCall());
+  transform.on('finish', common.mustNotCall());
 }
 
 {

node/_tools/test/parallel/test-stream-writable-change-default-encoding.js

@@ -5,6 +5,10 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
+
 // Copyright Joyent, Inc. and other Node contributors.
 //
 // Permission is hereby granted, free of charge, to any person obtaining a
@@ -72,10 +76,10 @@ assert.throws(() => {
 }, {
   name: 'TypeError',
   code: 'ERR_UNKNOWN_ENCODING',
-  message: 'Unknown encoding: [object Object]'
+  message: 'Unknown encoding: {}'
 });
 
-(function checkVairableCaseEncoding() {
+(function checkVariableCaseEncoding() {
   const m = new MyWritable(function(isBuffer, type, enc) {
     assert.strictEqual(enc, 'ascii');
   }, { decodeStrings: false });

node/_tools/test/parallel/test-stream-writable-destroy.js

@@ -5,6 +5,10 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
+
 'use strict';
 
 const common = require('../common');
@@ -131,8 +135,6 @@ const assert = require('assert');
 
   write.destroy();
 
-  write.removeListener('finish', fail);
-  write.on('finish', common.mustCall());
   assert.strictEqual(write.destroyed, true);
 }
 
@@ -358,33 +360,35 @@ const assert = require('assert');
   const write = new Writable({
     write(chunk, enc, cb) { process.nextTick(cb); }
   });
+  const _err = new Error('asd');
   write.once('error', common.mustCall((err) => {
     assert.strictEqual(err.message, 'asd');
   }));
   write.end('asd', common.mustCall((err) => {
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err, _err);
   }));
-  write.destroy(new Error('asd'));
+  write.destroy(_err);
 }
 
 {
   // Call buffered write callback with error
 
+  const _err = new Error('asd');
   const write = new Writable({
     write(chunk, enc, cb) {
-      process.nextTick(cb, new Error('asd'));
+      process.nextTick(cb, _err);
     },
     autoDestroy: false
   });
   write.cork();
   write.write('asd', common.mustCall((err) => {
-    assert.strictEqual(err.message, 'asd');
+    assert.strictEqual(err, _err);
   }));
   write.write('asd', common.mustCall((err) => {
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err, _err);
   }));
   write.on('error', common.mustCall((err) => {
-    assert.strictEqual(err.message, 'asd');
+    assert.strictEqual(err, _err);
   }));
   write.uncork();
 }
@@ -478,3 +482,19 @@ const assert = require('assert');
   write.destroy();
   write.destroy();
 }
+
+{
+  // https://github.com/nodejs/node/issues/39356
+  const s = new Writable({
+    final() {}
+  });
+  const _err = new Error('oh no');
+  // Remove `callback` and it works
+  s.end(common.mustCall((err) => {
+    assert.strictEqual(err, _err);
+  }));
+  s.on('error', common.mustCall((err) => {
+    assert.strictEqual(err, _err);
+  }));
+  s.destroy(_err);
+}

node/_tools/test/parallel/test-stream-writable-end-cb-error.js

@@ -5,6 +5,10 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
+
 'use strict';
 
 const common = require('../common');
@@ -15,19 +19,20 @@ const stream = require('stream');
   // Invoke end callback on failure.
   const writable = new stream.Writable();
 
+  const _err = new Error('kaboom');
   writable._write = (chunk, encoding, cb) => {
-    process.nextTick(cb, new Error('kaboom'));
+    process.nextTick(cb, _err);
   };
 
   writable.on('error', common.mustCall((err) => {
-    assert.strictEqual(err.message, 'kaboom');
+    assert.strictEqual(err, _err);
   }));
   writable.write('asd');
   writable.end(common.mustCall((err) => {
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err, _err);
   }));
   writable.end(common.mustCall((err) => {
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err, _err);
   }));
 }
 
@@ -64,18 +69,12 @@ const stream = require('stream');
     }
   });
   w.end('testing ended state', common.mustCall((err) => {
-    // This errors since .destroy(err), which is invoked by errors
-    // in same tick below, will error all pending callbacks.
-    // Does this make sense? Not sure.
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
   }));
   assert.strictEqual(w.destroyed, false);
   assert.strictEqual(w.writableEnded, true);
   w.end(common.mustCall((err) => {
-    // This errors since .destroy(err), which is invoked by errors
-    // in same tick below, will error all pending callbacks.
-    // Does this make sense? Not sure.
-    assert.strictEqual(err.code, 'ERR_STREAM_DESTROYED');
+    assert.strictEqual(err.code, 'ERR_STREAM_WRITE_AFTER_END');
   }));
   assert.strictEqual(w.destroyed, false);
   assert.strictEqual(w.writableEnded, true);

node/_tools/test/parallel/test-stream-write-destroy.js

@@ -5,6 +5,10 @@
 // Taken from Node 16.13.0
 // This file is automatically generated by "node/_tools/setup.ts". Do not modify this file manually
 
+// TODO(cjihrig): This test was updated in Node.js 18. In order to vendor
+// readable-stream, this must be updated manually. Once std node has been
+// updated to Node.js 18 this can be updated automatically.
+
 'use strict';
 require('../common');
 const assert = require('assert');
@@ -27,9 +31,7 @@ for (const withPendingData of [ false, true ]) {
 
     let chunksWritten = 0;
     let drains = 0;
-    let finished = false;
     w.on('drain', () => drains++);
-    w.on('finish', () => finished = true);
 
     function onWrite(err) {
       if (err) {
@@ -67,9 +69,5 @@ for (const withPendingData of [ false, true ]) {
     assert.strictEqual(chunksWritten, useEnd && !withPendingData ? 1 : 2);
     assert.strictEqual(callbacks.length, 0);
     assert.strictEqual(drains, 1);
-
-    // When we used `.end()`, we see the 'finished' event if and only if
-    // we actually finished processing the write queue.
-    assert.strictEqual(finished, !withPendingData && useEnd);
   }
 }

node/_tools/vendor_readable_stream.ts

@@ -0,0 +1,47 @@
+// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
+// usage: deno run -A --unstable node/_tools/vendor_readable_stream.ts
+const sourceUrl =
+  "https://esm.sh/v92/[email protected]/es2022/readable-stream.js";
+const header =
+  `// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
+// Copyright Joyent and Node contributors. All rights reserved. MIT license.
+// deno-fmt-ignore-file
+// deno-lint-ignore-file
+import { nextTick } from "./_next_tick.ts";
+import { stdio } from "./_process/stdio.mjs";
+
+`;
+const outputFile = new URL("../_stream.mjs", import.meta.url).pathname;
+const endMarker = "/* End esm.sh bundle */";
+
+// Download the readable-stream module.
+const res = await fetch(sourceUrl);
+let src = await res.text();
+
+// Remove the AbortController fallback code since AbortController always
+// exists in Deno.
+src = src.replaceAll(/import { AbortController as.+?;/g, "");
+src = src.replaceAll("||__abort_controller$AbortController", "");
+
+// Replace Node.js core module imports with Deno std modules.
+src = src.replaceAll(/"\/v\d+\/node_buffer.js"/g, '"./buffer.ts"');
+src = src.replaceAll(/"\/v\d+\/string_decoder.+?"/g, '"./string_decoder.ts"');
+src = src.replaceAll(/"\/v\d+\/events@.+?"/g, '"./events.ts"');
+
+// Replace import of the Node.js process object with the APIs that are actually
+// used to avoid issues with circular imports.
+src = src.replaceAll(
+  /import __Process\$ from "\/v\d+\/node_process.js";/g,
+  "const __Process$ = { nextTick, stdio };",
+);
+
+// Get any additional code from the end of the current file.
+const current = Deno.readTextFileSync(outputFile);
+const trailer = current.split(endMarker)[1] ?? "";
+
+// Prepend copyrights, Deno tooling directives, and necessary imports and make
+// sure any code at the end of the file is maintained.
+src = header + src + endMarker + trailer;
+
+// Update the local file.
+Deno.writeTextFileSync(outputFile, src);

node/internal/crypto/cipher.ts

@@ -46,7 +46,7 @@ export interface CipherOCBOptions extends TransformOptions {
   authTagLength: number;
 }
 
-export interface Cipher extends Transform {
+export interface Cipher extends ReturnType<typeof Transform> {
   update(data: BinaryLike): Buffer;
   update(data: string, inputEncoding: Encoding): Buffer;
   update(