stream: address remaining webstreams review comments

Merge main so this work sits on the landed promise-churn helpers
(kResolvedPromise / thenAlgorithmResult). Remaining review items:

- Use the shared kResolvedPromise instead of allocating
  PromiseResolve() on the pull/write hot path.
- Drop pipeTo's synchronous FillSync so consecutive pulls stay
  one-per-microtask and the change is no longer semver-major
  for that reason.
- Call readableStreamDefaultControllerError instead of looking
  up controller.error.
- Document and test subclass constructor materialization of the
  deferred empty-stream controller.

Assisted-by: Grok

Co-authored-by: Yagiz Nizipli <anonrig@users.noreply.github.com>

Author: cursoragent

CONTRIBUTING.md

@@ -47,6 +47,8 @@ See [details on our policy on Code of Conduct](./doc/contributing/code-of-conduc
 
 Pull Requests are the way concrete changes are made to the code, documentation,
 dependencies, and tools contained in the `nodejs/node` repository.
+Contributors who are not Collaborators may have no more than 10 pull requests
+open at a time.
 
 * [Dependencies](./doc/contributing/pull-requests.md#dependencies)
 * [Setting up your local environment](./doc/contributing/pull-requests.md#setting-up-your-local-environment)

Makefile

@@ -1449,15 +1449,15 @@ binary: $(BINARYTAR) ## Build release binary tarballs.
 # Note: this is strictly for release builds on release machines only.
 binary-upload: binary
 	ssh $(STAGINGSERVER) "mkdir -p nodejs/$(DISTTYPEDIR)/$(FULLVERSION)"
-	chmod 664 $(TARNAME)-$(OSTYPE)-$(ARCH).tar.gz
-	scp -p $(TARNAME)-$(OSTYPE)-$(ARCH).tar.gz $(STAGINGSERVER):nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.gz
-	ssh $(STAGINGSERVER) "rclone copyto nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.gz $(CLOUDFLARE_BUCKET)/nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.gz"
-	ssh $(STAGINGSERVER) "touch nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.gz.done"
+	chmod 664 $(BINARYNAME).tar.gz
+	scp -p $(BINARYNAME).tar.gz $(STAGINGSERVER):nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.gz
+	ssh $(STAGINGSERVER) "rclone copyto nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.gz $(CLOUDFLARE_BUCKET)/nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.gz"
+	ssh $(STAGINGSERVER) "touch nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.gz.done"
 ifeq ($(XZ), 1)
-	chmod 664 $(TARNAME)-$(OSTYPE)-$(ARCH).tar.xz
-	scp -p $(TARNAME)-$(OSTYPE)-$(ARCH).tar.xz $(STAGINGSERVER):nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.xz
-	ssh $(STAGINGSERVER) "rclone copyto nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.xz $(CLOUDFLARE_BUCKET)/nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.xz"
-	ssh $(STAGINGSERVER) "touch nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(TARNAME)-$(OSTYPE)-$(ARCH).tar.xz.done"
+	chmod 664 $(BINARYNAME).tar.xz
+	scp -p $(BINARYNAME).tar.xz $(STAGINGSERVER):nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.xz
+	ssh $(STAGINGSERVER) "rclone copyto nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.xz $(CLOUDFLARE_BUCKET)/nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.xz"
+	ssh $(STAGINGSERVER) "touch nodejs/$(DISTTYPEDIR)/$(FULLVERSION)/$(BINARYNAME).tar.xz.done"
 endif
 
 .PHONY: bench-all

benchmark/misc/startup-core.js

@@ -64,6 +64,7 @@ function main({ n, script, mode }) {
   const warmup = 3;
   const state = { n, finished: -warmup };
   if (mode === 'worker') {
+    // eslint-disable-next-line no-global-assign
     Worker = require('worker_threads').Worker;
     spawnWorker(script, bench, state);
   } else {

benchmark/repl/completion.js

@@ -0,0 +1,53 @@
+'use strict';
+
+const common = require('../common.js');
+const repl = require('node:repl');
+const { PassThrough, Writable } = require('node:stream');
+
+const bench = common.createBenchmark(main, {
+  n: [5e3],
+  query: [
+    'cons',
+    'console.lo',
+    'Buffer.prototype.wri',
+    "require('f",
+  ],
+  useGlobal: [0, 1],
+});
+
+function main({ n, query, useGlobal }) {
+  const input = new PassThrough();
+  const output = new Writable({ write(c, e, cb) { cb(); } });
+  const server = new repl.REPLServer({
+    input,
+    output,
+    terminal: false,
+    useGlobal: !!useGlobal,
+  });
+
+  // Inspector callbacks do not keep the event loop alive on their own.
+  const keepAlive = setInterval(() => {}, 0x7fffffff);
+  let remaining = n;
+
+  function complete() {
+    server.complete(query, onComplete);
+  }
+
+  function onComplete(err) {
+    if (err) {
+      throw err;
+    }
+
+    if (--remaining === 0) {
+      bench.end(n);
+      clearInterval(keepAlive);
+      server.close();
+      return;
+    }
+
+    setImmediate(complete);
+  }
+
+  bench.start();
+  setImmediate(complete);
+}

benchmark/repl/creation.js

@@ -0,0 +1,39 @@
+'use strict';
+
+const common = require('../common.js');
+const repl = require('node:repl');
+const { PassThrough, Writable } = require('node:stream');
+
+const bench = common.createBenchmark(main, {
+  n: [500],
+  preview: [0, 1],
+  terminal: [0, 1],
+  useGlobal: [0, 1],
+}, {
+  combinationFilter: ({ preview, terminal }) => !!terminal || !preview,
+});
+
+function main({ n, preview, terminal, useGlobal }) {
+  const inputs = Array.from({ length: n }, () => new PassThrough());
+  const outputs = Array.from(
+    { length: n },
+    () => new Writable({ write(c, e, cb) { cb(); } }),
+  );
+  const servers = new Array(n);
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    servers[i] = new repl.REPLServer({
+      input: inputs[i],
+      output: outputs[i],
+      preview: !!preview,
+      terminal: !!terminal,
+      useGlobal: !!useGlobal,
+    });
+  }
+  bench.end(n);
+
+  for (const server of servers) {
+    server.close();
+  }
+}

benchmark/repl/evaluate.js

@@ -0,0 +1,57 @@
+'use strict';
+
+const common = require('../common.js');
+const repl = require('node:repl');
+const { PassThrough, Writable } = require('node:stream');
+
+const bench = common.createBenchmark(main, {
+  n: [2e4],
+  code: [
+    '1 + 1',
+    '({ answer: 42 })',
+    'Promise.resolve(42)',
+    'await Promise.resolve(42)',
+  ],
+  mode: ['sloppy', 'strict'],
+  useGlobal: [0, 1],
+});
+
+function main({ n, code, mode, useGlobal }) {
+  const input = new PassThrough();
+  const output = new Writable({ write(c, e, cb) { cb(); } });
+  const server = new repl.REPLServer({
+    input,
+    output,
+    replMode: mode === 'strict' ?
+      repl.REPL_MODE_STRICT :
+      repl.REPL_MODE_SLOPPY,
+    terminal: false,
+    useGlobal: !!useGlobal,
+  });
+
+  // Inspector callbacks do not keep the event loop alive on their own.
+  const keepAlive = setInterval(() => {}, 0x7fffffff);
+  let remaining = n;
+
+  function evaluate() {
+    server.eval(`${code}\n`, server.context, 'repl', onEvaluate);
+  }
+
+  function onEvaluate(err) {
+    if (err) {
+      throw err;
+    }
+
+    if (--remaining === 0) {
+      bench.end(n);
+      clearInterval(keepAlive);
+      server.close();
+      return;
+    }
+
+    setImmediate(evaluate);
+  }
+
+  bench.start();
+  setImmediate(evaluate);
+}

benchmark/repl/process-lines.js

@@ -0,0 +1,52 @@
+'use strict';
+
+const common = require('../common.js');
+const repl = require('node:repl');
+const { PassThrough, Writable } = require('node:stream');
+
+const bench = common.createBenchmark(main, {
+  n: [1e4],
+  code: [
+    '1 + 1\n',
+    'Promise.resolve(42)\n',
+  ],
+  mode: ['sloppy', 'strict'],
+  terminal: [0, 1],
+  useGlobal: [0, 1],
+});
+
+function main({ n, code: inputCode, mode, terminal, useGlobal }) {
+  const input = new PassThrough();
+  const output = new Writable({ write(c, e, cb) { cb(); } });
+  const server = new repl.REPLServer({
+    input,
+    output,
+    replMode: mode === 'strict' ?
+      repl.REPL_MODE_STRICT :
+      repl.REPL_MODE_SLOPPY,
+    terminal: !!terminal,
+    useGlobal: !!useGlobal,
+  });
+  const originalEval = server.eval;
+  // TTY input dispatch can briefly have no other active event loop handles.
+  const keepAlive = setInterval(() => {}, 0x7fffffff);
+  let remaining = n;
+
+  // eslint-disable-next-line node-core/func-name-matching
+  server.eval = function REPLEval(code, context, file, callback) {
+    originalEval(code, context, file, function onEvaluate() {
+      const result = Reflect.apply(callback, this, arguments);
+      if (--remaining === 0) {
+        bench.end(n);
+        clearInterval(keepAlive);
+        server.close();
+      } else {
+        setImmediate(() => input.write(inputCode));
+      }
+      return result;
+    });
+  };
+
+  bench.start();
+  input.write(inputCode);
+}

benchmark/repl/reset-context.js

@@ -0,0 +1,26 @@
+'use strict';
+
+const common = require('../common.js');
+const repl = require('node:repl');
+const { PassThrough, Writable } = require('node:stream');
+
+const bench = common.createBenchmark(main, {
+  n: [1e3],
+});
+
+function main({ n }) {
+  const input = new PassThrough();
+  const output = new Writable({ write(c, e, cb) { cb(); } });
+  const server = new repl.REPLServer({
+    input,
+    output,
+    terminal: false,
+  });
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    server.resetContext();
+  }
+  bench.end(n);
+  server.close();
+}

benchmark/sqlite/sqlite-diagnostic-channel.js

@@ -0,0 +1,42 @@
+'use strict';
+const common = require('../common.js');
+const sqlite = require('node:sqlite');
+const dc = require('node:diagnostics_channel');
+const assert = require('node:assert');
+
+const bench = common.createBenchmark(main, {
+  n: [1e5],
+  mode: ['none', 'subscribed', 'unsubscribed'],
+});
+
+function main(conf) {
+  const { n, mode } = conf;
+
+  const db = new sqlite.DatabaseSync(':memory:');
+  db.exec('CREATE TABLE t (x INTEGER)');
+  const insert = db.prepare('INSERT INTO t VALUES (?)');
+
+  let subscriber;
+  if (mode === 'subscribed') {
+    subscriber = () => {};
+    dc.subscribe('sqlite.db.query', subscriber);
+  } else if (mode === 'unsubscribed') {
+    subscriber = () => {};
+    dc.subscribe('sqlite.db.query', subscriber);
+    dc.unsubscribe('sqlite.db.query', subscriber);
+  }
+  // mode === 'none': no subscription ever made
+
+  let result;
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    result = insert.run(i);
+  }
+  bench.end(n);
+
+  if (mode === 'subscribed') {
+    dc.unsubscribe('sqlite.db.query', subscriber);
+  }
+
+  assert.ok(result !== undefined);
+}

benchmark/webstreams/pipe-to.js

@@ -7,29 +7,27 @@ const {
 
 const bench = common.createBenchmark(main, {
   n: [5e5],
-  highWaterMarkR: [512, 1024, 2048, 4096],
-  highWaterMarkW: [512, 1024, 2048, 4096],
+  highWaterMarkR: [1, 1024, 4096],
+  highWaterMarkW: [1, 1024, 4096],
 });
 
 
 async function main({ n, highWaterMarkR, highWaterMarkW }) {
   const b = Buffer.alloc(1024);
   let i = 0;
   const rs = new ReadableStream({
-    highWaterMark: highWaterMarkR,
     pull: function(controller) {
       if (i++ < n) {
         controller.enqueue(b);
       } else {
         controller.close();
       }
     },
-  });
+  }, { highWaterMark: highWaterMarkR });
   const ws = new WritableStream({
-    highWaterMark: highWaterMarkW,
     write(chunk, controller) {},
     close() { bench.end(n); },
-  });
+  }, { highWaterMark: highWaterMarkW });
 
   bench.start();
   rs.pipeTo(ws);