test_runner: fix test runner concurrency

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/harness.js

@@ -176,7 +176,7 @@ function setup(root) {
       cancelled: 0,
       skipped: 0,
       todo: 0,
-      planned: 0,
+      topLevel: 0,
       suites: 0,
     },
   };

lib/internal/test_runner/runner.js

@@ -5,14 +5,11 @@ const {
   ArrayPrototypeFilter,
   ArrayPrototypeForEach,
   ArrayPrototypeIncludes,
-  ArrayPrototypeIndexOf,
   ArrayPrototypeMap,
   ArrayPrototypePush,
   ArrayPrototypeSlice,
   ArrayPrototypeSome,
   ArrayPrototypeSort,
-  ArrayPrototypeSplice,
-  Number,
   ObjectAssign,
   PromisePrototypeThen,
   SafePromiseAll,
@@ -207,13 +204,9 @@ class FileTest extends Test {
 
         const diagnostics = YAMLToJs(node.diagnostics);
         const cancelled = kCanceledTests.has(diagnostics.error?.failureType);
-        const testNumber = nesting === 0 ? (Number(node.id) + this.testNumber - 1) : node.id;
+        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,
@@ -240,22 +233,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();
     }
   }
@@ -335,17 +342,7 @@ function runTestFile(path, root, inspectPort, filesWatcher, testNamePatterns) {
       throw err;
     }
   });
-  const promise = subtest.start();
-  if (filesWatcher) {
-    return PromisePrototypeThen(promise, () => {
-      const index = ArrayPrototypeIndexOf(root.subtests, subtest);
-      if (index !== -1) {
-        ArrayPrototypeSplice(root.subtests, index, 1);
-        root.waitingOn--;
-      }
-    });
-  }
-  return promise;
+  return subtest.start();
 }
 
 function watchFiles(testFiles, root, inspectPort, testNamePatterns) {
@@ -361,6 +358,10 @@ function watchFiles(testFiles, root, inspectPort, testNamePatterns) {
         runningProcess.kill();
         await once(runningProcess, 'exit');
       }
+      if (!runningSubtests.size) {
+        // Reset the topLevel counter
+        root.harness.counters.topLevel = 0;
+      }
       await runningSubtests.get(file);
       runningSubtests.set(file, runTestFile(file, root, inspectPort, filesWatcher, testNamePatterns));
     }, undefined, (error) => {

lib/internal/test_runner/test.js

@@ -637,7 +637,7 @@ class Test extends AsyncResource {
       this.parent.processPendingSubtests();
     } else if (!this.reported) {
       this.reported = true;
-      this.reporter.plan(this.nesting, kFilename, this.root.harness.counters.planned);
+      this.reporter.plan(this.nesting, kFilename, this.root.harness.counters.topLevel);
 
       for (let i = 0; i < this.diagnostics.length; i++) {
         this.reporter.diagnostic(this.nesting, kFilename, this.diagnostics[i]);
@@ -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 });

lib/internal/test_runner/utils.js

@@ -227,7 +227,7 @@ function parseCommandLine() {
 
 function countCompletedTest(test, harness = test.root.harness) {
   if (test.nesting === 0) {
-    harness.counters.planned++;
+    harness.counters.topLevel++;
   }
   if (test.reportedType === 'suite') {
     harness.counters.suites++;

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/fixtures/test-runner/output/output_cli.js

@@ -5,5 +5,5 @@ const fixtures = require('../../../common/fixtures');
 const spawn = require('node:child_process').spawn;
 
 spawn(process.execPath,
-      ['--no-warnings', '--test', '--test-reporter', 'tap', fixtures.path('test-runner/output/output.js')],
+      ['--no-warnings', '--test', '--test-reporter', 'tap', fixtures.path('test-runner/output/output.js'), fixtures.path('test-runner/output/single.js')],
       { stdio: 'inherit' });

test/fixtures/test-runner/output/output_cli.snapshot

@@ -672,10 +672,15 @@ not ok 66 - invalid subtest fail
 # Warning: Test "immediate reject - passes but warns" generated asynchronous activity after the test ended. This activity created the error "Error: rejected from immediate reject fail" and would have caused the test to fail, but instead triggered an unhandledRejection event.
 # Warning: Test "callback called twice in different ticks" generated asynchronous activity after the test ended. This activity created the error "Error [ERR_TEST_FAILURE]: callback invoked multiple times" and would have caused the test to fail, but instead triggered an uncaughtException event.
 # Warning: Test "callback async throw after done" generated asynchronous activity after the test ended. This activity created the error "Error: thrown from callback async throw after done" and would have caused the test to fail, but instead triggered an uncaughtException event.
-1..66
-# tests 80
+# Subtest: last test
+ok 67 - last test
+  ---
+  duration_ms: *
+  ...
+1..67
+# tests 81
 # suites 0
-# pass 37
+# pass 38
 # fail 25
 # cancelled 3
 # skipped 10

test/fixtures/test-runner/output/single.js

@@ -0,0 +1,4 @@
+// Flags: --no-warnings
+'use strict';
+const test = require('node:test');
+test('last test', () => {});

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);
+});

test/parallel/test-runner-watch-mode.mjs

@@ -14,13 +14,14 @@ async function testWatch({ files, fileToUpdate }) {
 
   child.stdout.on('data', (data) => {
     stdout += data.toString();
-    const matches = stdout.match(/test has ran/g);
-    if (matches?.length >= 1) ran1.resolve();
-    if (matches?.length >= 2) ran2.resolve();
+    const testRuns = stdout.match(/ - test has ran/g);
+    if (testRuns?.length >= 1) ran1.resolve();
+    if (testRuns?.length >= 2) ran2.resolve();
   });
 
   await ran1.promise;
-  const interval = setInterval(() => writeFileSync(fileToUpdate, readFileSync(fileToUpdate, 'utf8')), 50);
+  const content = readFileSync(fileToUpdate, 'utf8');
+  const interval = setInterval(() => writeFileSync(fileToUpdate, content), 10);
   await ran2.promise;
   clearInterval(interval);
   child.kill();