test_runner: fix test runner concurrency

PR-URL: #47675
Fixes: #47365
Fixes: #47696
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Colin Ihrig <cjihrig@gmail.com>

Author: MoLow

lib/internal/per_context/primordials.js

@@ -565,13 +565,7 @@ primordials.SafePromiseAllSettled = (promises, mapFn) =>
  * @returns {Promise<void>}
  */
 primordials.SafePromiseAllSettledReturnVoid = async (promises, mapFn) => {
-  for (let i = 0; i < promises.length; i++) {
-    try {
-      await (mapFn != null ? mapFn(promises[i], i) : promises[i]);
-    } catch {
-      // In all settled, we can ignore errors.
-    }
-  }
+  await primordials.SafePromiseAllSettled(promises, mapFn);
 };
 
 /**

lib/internal/test_runner/runner.js

@@ -206,10 +206,6 @@ class FileTest extends Test {
         const testNumber = nesting === 0 ? (this.root.harness.counters.topLevel + 1) : node.id;
         const method = pass ? 'ok' : 'fail';
         this.reporter[method](nesting, this.name, testNumber, node.description, diagnostics, directive);
-        if (nesting === 0) {
-          this.failedSubtests ||= !pass;
-        }
-        this.#reportedChildren++;
         countCompletedTest({
           name: node.description,
           finished: true,
@@ -236,22 +232,36 @@ class FileTest extends Test {
         break;
     }
   }
+  #accumulateReportItem({ kind, node, comments, nesting = 0 }) {
+    if (kind !== TokenKind.TAP_TEST_POINT) {
+      return;
+    }
+    this.#reportedChildren++;
+    if (nesting === 0 && !node.status.pass) {
+      this.failedSubtests = true;
+    }
+  }
+  #drainBuffer() {
+    if (this.#buffer.length > 0) {
+      ArrayPrototypeForEach(this.#buffer, (ast) => this.#handleReportItem(ast));
+      this.#buffer = [];
+    }
+  }
   addToReport(ast) {
+    this.#accumulateReportItem(ast);
     if (!this.isClearToSend()) {
       ArrayPrototypePush(this.#buffer, ast);
       return;
     }
-    this.reportStarted();
+    this.#drainBuffer();
     this.#handleReportItem(ast);
   }
   reportStarted() {}
   report() {
+    this.#drainBuffer();
     const skipReporting = this.#skipReporting();
     if (!skipReporting) {
       super.reportStarted();
-    }
-    ArrayPrototypeForEach(this.#buffer, (ast) => this.#handleReportItem(ast));
-    if (!skipReporting) {
       super.report();
     }
   }

lib/internal/test_runner/test.js

@@ -658,7 +658,7 @@ class Test extends AsyncResource {
         this.reporter.coverage(this.nesting, kFilename, coverage);
       }
 
-      this.reporter.push(null);
+      this.reporter.end();
     }
   }
 

lib/internal/test_runner/tests_stream.js

@@ -59,6 +59,10 @@ class TestsStream extends Readable {
     this.#emit('test:coverage', { __proto__: null, nesting, file, summary });
   }
 
+  end() {
+    this.#tryPush(null);
+  }
+
   #emit(type, data) {
     this.emit(type, data);
     this.#tryPush({ type, data });

test/fixtures/test-runner/concurrency/a.mjs

@@ -0,0 +1,13 @@
+import tmpdir from '../../../common/tmpdir.js';
+import { setTimeout } from 'node:timers/promises';
+import fs from 'node:fs/promises';
+import path from 'node:path';
+
+await fs.writeFile(path.resolve(tmpdir.path, 'test-runner-concurrency'), 'a.mjs');
+while (true) {
+  const file = await fs.readFile(path.resolve(tmpdir.path, 'test-runner-concurrency'), 'utf8');
+  if (file === 'b.mjs') {
+    break;
+  }
+  await setTimeout(10);
+}

test/fixtures/test-runner/concurrency/b.mjs

@@ -0,0 +1,13 @@
+import tmpdir from '../../../common/tmpdir.js';
+import { setTimeout } from 'node:timers/promises';
+import fs from 'node:fs/promises';
+import path from 'node:path';
+
+while (true) {
+  const file = await fs.readFile(path.resolve(tmpdir.path, 'test-runner-concurrency'), 'utf8');
+  if (file === 'a.mjs') {
+    await fs.writeFile(path.resolve(tmpdir.path, 'test-runner-concurrency'), 'b.mjs');
+    break;
+  }
+  await setTimeout(10);
+}

test/parallel/test-primordials-promise.js

@@ -55,13 +55,11 @@ assertIsPromise(SafePromisePrototypeFinally(test(), common.mustCall()));
 
 assertIsPromise(SafePromiseAllReturnArrayLike([test()]));
 assertIsPromise(SafePromiseAllReturnVoid([test()]));
-assertIsPromise(SafePromiseAllSettledReturnVoid([test()]));
 assertIsPromise(SafePromiseAny([test()]));
 assertIsPromise(SafePromiseRace([test()]));
 
 assertIsPromise(SafePromiseAllReturnArrayLike([]));
 assertIsPromise(SafePromiseAllReturnVoid([]));
-assertIsPromise(SafePromiseAllSettledReturnVoid([]));
 
 {
   const val1 = Symbol();
@@ -108,9 +106,11 @@ Object.defineProperties(Array.prototype, {
 
 assertIsPromise(SafePromiseAll([test()]));
 assertIsPromise(SafePromiseAllSettled([test()]));
+assertIsPromise(SafePromiseAllSettledReturnVoid([test()]));
 
 assertIsPromise(SafePromiseAll([]));
 assertIsPromise(SafePromiseAllSettled([]));
+assertIsPromise(SafePromiseAllSettledReturnVoid([]));
 
 async function test() {
   const catchFn = common.mustCall();

test/parallel/test-runner-concurrency.js

@@ -1,7 +1,14 @@
 'use strict';
 const common = require('../common');
+const tmpdir = require('../common/tmpdir');
+const fixtures = require('../common/fixtures');
 const { describe, it, test } = require('node:test');
-const assert = require('assert');
+const assert = require('node:assert');
+const path = require('node:path');
+const fs = require('node:fs/promises');
+const os = require('node:os');
+
+tmpdir.refresh();
 
 describe('Concurrency option (boolean) = true ', { concurrency: true }, () => {
   let isFirstTestOver = false;
@@ -62,3 +69,14 @@ describe(
     it('should run after other suites', expectedTestTree);
   });
 }
+
+test('--test multiple files', { skip: os.availableParallelism() < 3 }, async () => {
+  await fs.writeFile(path.resolve(tmpdir.path, 'test-runner-concurrency'), '');
+  const { code, stderr } = await common.spawnPromisified(process.execPath, [
+    '--test',
+    fixtures.path('test-runner', 'concurrency', 'a.mjs'),
+    fixtures.path('test-runner', 'concurrency', 'b.mjs'),
+  ]);
+  assert.strictEqual(stderr, '');
+  assert.strictEqual(code, 0);
+});