test: deflake stream-readable-to-web test

Co-authored-by: Luigi Pinca <luigipinca@gmail.com>
PR-URL: https://github.com/nodejs/node/pull/58948
Fixes: https://github.com/nodejs/node/issues/58949
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: Juan José Arboleda <soyjuanarbol@gmail.com>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Zeyu "Alex" Yang <himself65@outlook.com>
Reviewed-By: Daeyeon Jeong <daeyeon.dev@gmail.com>

test/parallel/test-stream-readable-to-web.js

@@ -1,62 +0,0 @@
-'use strict';
-const common = require('../common');
-if (!common.hasCrypto) { common.skip('missing crypto'); }
-
-const { Readable } = require('stream');
-const process = require('process');
-const { randomBytes } = require('crypto');
-const assert = require('assert');
-
-// Based on: https://github.com/nodejs/node/issues/46347#issuecomment-1413886707
-// edit: make it cross-platform as /dev/urandom is not available on Windows
-{
-  let currentMemoryUsage = process.memoryUsage().arrayBuffers;
-
-  // We initialize a stream, but not start consuming it
-  const randomNodeStream = new Readable({
-    read(size) {
-      randomBytes(size, (err, buffer) => {
-        if (err) {
-          // If an error occurs, emit an 'error' event
-          this.emit('error', err);
-          return;
-        }
-
-        // Push the random bytes to the stream
-        this.push(buffer);
-      });
-    }
-  });
-  // after 2 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();
-    currentMemoryUsage = arrayBuffers;
-
-    assert(currentMemoryUsage <= 256 * 1024 * 1024);
-  };
-  setInterval(reportMemoryUsage, 1000);
-
-  // after 1 second we use Readable.toWeb
-  // memory usage should stay pretty much the same since it's still a stream
-  setTimeout(() => {
-    randomWebStream = Readable.toWeb(randomNodeStream);
-  }, 1000);
-
-  // after 2 seconds we start consuming the stream
-  // memory usage will grow, but the old chunks should be garbage-collected pretty quickly
-  setTimeout(async () => {
-    // eslint-disable-next-line no-unused-vars
-    for await (const _ of randomWebStream) {
-      // Do nothing, just let the stream flow
-    }
-  }, 2000);
-
-  setTimeout(() => {
-    // Test considered passed if we don't crash
-    process.exit(0);
-  }, 5000);
-}

test/parallel/test-stream-readable-to-web.mjs

@@ -0,0 +1,64 @@
+import { mustCall } from '../common/index.mjs';
+import { Readable } from 'node:stream';
+import { memoryUsage } from 'node:process';
+import assert from 'node:assert';
+import { setImmediate } from 'node:timers/promises';
+
+// Based on: https://github.com/nodejs/node/issues/46347#issuecomment-1413886707
+// edit: make it cross-platform as /dev/urandom is not available on Windows
+
+const MAX_MEM = 256 * 1024 * 1024; // 256 MiB
+
+function checkMemoryUsage() {
+  assert(memoryUsage().arrayBuffers < MAX_MEM);
+}
+
+const MAX_BUFFERS = 1000;
+let buffersCreated = 0;
+
+const randomNodeStream = new Readable({
+  read(size) {
+    if (buffersCreated >= MAX_BUFFERS) {
+      this.push(null);
+      return;
+    }
+
+    this.push(Buffer.alloc(size));
+    buffersCreated++;
+  }
+});
+
+randomNodeStream.on('error', (err) => {
+  assert.fail(err);
+});
+
+// Before doing anything, make sure memory usage is okay
+checkMemoryUsage();
+
+// Create stream and check memory usage remains okay
+
+const randomWebStream = Readable.toWeb(randomNodeStream);
+
+checkMemoryUsage();
+
+let timeout;
+try {
+  // Wait two seconds before consuming the stream to see if memory usage increases
+  timeout = setTimeout(mustCall(async () => {
+    // Did the stream leak memory?
+    checkMemoryUsage();
+    // eslint-disable-next-line no-unused-vars
+    for await (const _ of randomWebStream) {
+      // Yield event loop to allow garbage collection
+      await setImmediate();
+      // consume the stream
+      // check memory usage remains okay
+      checkMemoryUsage();
+    }
+  }), 2000);
+} catch (err) {
+  if (timeout) {
+    clearTimeout(timeout);
+  }
+  assert.fail(err);
+}