stream: use BLQS when not in objectMode (fixes nodejs#46347)

BLQS = ByteLengthQueuingStrategy

Fixes: nodejs#46347

lib/internal/webstreams/adapters.js

@@ -33,6 +33,8 @@ const {
   CountQueuingStrategy,
 } = require('internal/webstreams/queuingstrategies');
 
+const { kFresh } = require('internal/webstreams/util');
+
 const {
   Writable,
   Readable,
@@ -489,12 +491,13 @@ function newReadableStreamFromStreamReadable(streamReadable, options = kEmptyObj
   streamReadable.on('data', onData);
 
   return new ReadableStream({
-    start(c) { controller = c; },
+    start(c) { controller = c; controller[kFresh] = true; console.log("start", typeof c); },
 
-    pull() { streamReadable.resume(); },
+    pull() { streamReadable.resume();  },
 
     cancel(reason) {
       destroy(streamReadable, reason);
+      console.log("cancel")
     },
   }, strategy);
 }

lib/internal/webstreams/readablestream.js

@@ -115,6 +115,7 @@ const {
   iteratorNext,
   kType,
   kState,
+  kFresh,
 } = require('internal/webstreams/util');
 
 const {
@@ -2351,6 +2352,10 @@ function readableStreamDefaultControllerGetDesiredSize(controller) {
 }
 
 function readableStreamDefaultControllerShouldCallPull(controller) {
+  if (controller[kFresh]) {
+    controller[kFresh] = false;
+    return false;
+  }
   const {
     stream,
   } = controller[kState];

lib/internal/webstreams/util.js

@@ -47,6 +47,7 @@ const {
   validateFunction,
 } = require('internal/validators');
 
+const kFresh = Symbol('kFresh');
 const kState = Symbol('kState');
 const kType = Symbol('kType');
 
@@ -295,4 +296,5 @@ module.exports = {
   iteratorNext,
   kType,
   kState,
+  kFresh,
 };

t/46347.mjs

@@ -0,0 +1,44 @@
+// since it's ESM, save it as .mjs
+
+import fs from 'node:fs'
+import process from 'node:process'
+import { Readable } from 'node:stream'
+
+// we initialize a stream, but not start consuming it
+const randomNodeStream = fs.createReadStream('/dev/urandom')
+// after 10 seconds, it'll get converted to web stream
+let randomWebStream
+
+// we check memory usage every second
+// since it's a stream, it shouldn't be higher than the chunk size
+const reportMemoryUsage = () => {
+    const { arrayBuffers } = process.memoryUsage()
+    console.log(
+        `Array buffers memory usage is ${Math.round(
+            arrayBuffers / 1024 / 1024
+        )} MiB`
+    )
+    if (arrayBuffers > 256 * 1024 * 1024) {
+        // streaming should not lead to such a memory increase
+        // therefore, if it happens => bail
+        console.log('Over 256 MiB taken, exiting')
+        process.exit(0)
+    }
+}
+setInterval(reportMemoryUsage, 1000)
+
+// after 10 seconds we use Readable.toWeb
+// memory usage should stay pretty much the same since it's still a stream
+setTimeout(() => {
+    console.log('converting node stream to web stream')
+    randomWebStream = Readable.toWeb(randomNodeStream)
+}, 5000)
+
+// after 15 seconds we start consuming the stream
+// memory usage will grow, but the old chunks should be garbage-collected pretty quickly
+setTimeout(async () => {
+    console.log('reading the chunks')
+    for await (const chunk of randomWebStream) {
+        // do nothing, just let the stream flow
+    }
+}, 15000)