test_runner: do not reuse a worker ID held by a running file

Worker IDs were handed out round-robin and never released, so a file
that started after another finished could get an ID still held by a
live process. Track the IDs in use, hand out the lowest free one, and
release it in a finally block once the child process exits.

Refs: #61394
Signed-off-by: Vasiliy Serpokryl <vasiliy.serpokryl@mail.ru>
PR-URL: #65739
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Pietro Marchini <pietro.marchini94@gmail.com>

Author: vserpokryl

doc/api/test.md

@@ -4541,7 +4541,9 @@ The unique identifier of the worker running the current test file. This value is
 derived from the `NODE_TEST_WORKER_ID` environment variable. When running tests
 with `--test-isolation=process` (the default), each test file runs in a separate
 child process and is assigned a worker ID from 1 to N, where N is the number of
-concurrent workers. When running with `--test-isolation=none`, all tests run in
+concurrent workers. A worker ID is never shared by two test files running at the
+same time. Once a test file finishes, its worker ID is reused by the next test
+file that starts. When running with `--test-isolation=none`, all tests run in
 the same process and the worker ID is always 1. This value is `undefined` when
 not running in a test context.
 

lib/internal/test_runner/runner.js

@@ -15,7 +15,6 @@ const {
   ArrayPrototypeSlice,
   ArrayPrototypeSome,
   ArrayPrototypeSort,
-  MathMax,
   ObjectAssign,
   PromisePrototypeThen,
   PromiseWithResolvers,
@@ -36,7 +35,6 @@ const {
 
 const { spawn } = require('child_process');
 const { finished } = require('internal/streams/end-of-stream');
-const { availableParallelism } = require('os');
 const { resolve, sep, isAbsolute } = require('path');
 const { DefaultDeserializer, DefaultSerializer } = require('v8');
 const { getOptionValue, getOptionsAsFlagsFromBinding } = require('internal/options');
@@ -137,17 +135,22 @@ let kResistStopPropagation;
 
 // Worker ID pool management for concurrent test execution
 class WorkerIdPool {
-  #nextId = 0;
-  #maxConcurrency;
-
-  constructor(maxConcurrency) {
-    this.#maxConcurrency = maxConcurrency;
-  }
+  #acquiredIds = new SafeSet();
 
   acquire() {
-    const id = (this.#nextId++ % this.#maxConcurrency) + 1;
+    let id = 1;
+
+    while (this.#acquiredIds.has(id)) {
+      id++;
+    }
+
+    this.#acquiredIds.add(id);
     return id;
   }
+
+  release(id) {
+    this.#acquiredIds.delete(id);
+  }
 }
 
 function createTestFileList(patterns, cwd) {
@@ -519,94 +522,102 @@ function runTestFile(path, filesWatcher, opts) {
       debug('Assigned worker ID %d to test file: %s', workerId, path);
     }
 
-    if (watchMode) {
-      stdio.push('ipc');
-      env.WATCH_REPORT_DEPENDENCIES = '1';
-    }
-    if (opts.root.harness.shouldColorizeTestFiles) {
-      env.FORCE_COLOR = '1';
-    }
-
-    const child = spawn(
-      process.execPath, args,
-      {
-        __proto__: null,
-        signal: t.signal,
-        encoding: 'utf8',
-        env,
-        stdio,
-        cwd: opts.cwd,
-      },
-    );
-    if (watchMode) {
-      filesWatcher.runningProcesses.set(path, child);
-      filesWatcher.watcher.watchChildProcessModules(child, path);
-    }
-
-    let err;
+    try {
+      if (watchMode) {
+        stdio.push('ipc');
+        env.WATCH_REPORT_DEPENDENCIES = '1';
+      }
+      if (opts.root.harness.shouldColorizeTestFiles) {
+        env.FORCE_COLOR = '1';
+      }
 
-    child.on('error', (error) => {
-      err = error;
-    });
+      const child = spawn(
+        process.execPath, args,
+        {
+          __proto__: null,
+          signal: t.signal,
+          encoding: 'utf8',
+          env,
+          stdio,
+          cwd: opts.cwd,
+        },
+      );
+      if (watchMode) {
+        filesWatcher.runningProcesses.set(path, child);
+        filesWatcher.watcher.watchChildProcessModules(child, path);
+      }
 
-    child.stdout.on('data', (data) => {
-      subtest.parseMessage(data);
-    });
+      let err;
 
-    const rl = new Interface({ __proto__: null, input: child.stderr });
-    rl.on('line', (line) => {
-      if (isInspectorMessage(line)) {
-        process.stderr.write(line + '\n');
-        return;
-      }
+      child.on('error', (error) => {
+        err = error;
+      });
 
-      // stderr cannot be treated as TAP, per the spec. However, we want to
-      // surface stderr lines to improve the DX. Inject each line into the
-      // test output as an unknown token as if it came from the TAP parser.
-      subtest.addToReport({
-        __proto__: null,
-        type: 'test:stderr',
-        data: { __proto__: null, file: path, message: line + '\n' },
+      child.stdout.on('data', (data) => {
+        subtest.parseMessage(data);
       });
-    });
 
-    const { 0: { 0: code, 1: signal } } = await SafePromiseAll([
-      once(child, 'exit', { __proto__: null, signal: t.signal }),
-      finished(child.stdout, { __proto__: null, signal: t.signal }),
-    ]);
-
-    // Close readline interface to prevent memory leak
-    rl.close();
-
-    if (watchMode) {
-      filesWatcher.runningProcesses.delete(path);
-      filesWatcher.runningSubtests.delete(path);
-      (async () => {
-        try {
-          await subTestEnded;
-        } finally {
-          if (filesWatcher.runningSubtests.size === 0) {
-            opts.root.reporter[kEmitMessage]('test:watch:drained');
-            opts.root.postRun();
-          }
+      const rl = new Interface({ __proto__: null, input: child.stderr });
+      rl.on('line', (line) => {
+        if (isInspectorMessage(line)) {
+          process.stderr.write(line + '\n');
+          return;
         }
-      })();
-    }
 
-    if (code !== 0 || signal !== null) {
-      if (!err) {
-        const failureType = subtest.failedSubtests ? kSubtestsFailed : kTestCodeFailure;
-        err = ObjectAssign(new ERR_TEST_FAILURE('test failed', failureType), {
+        // stderr cannot be treated as TAP, per the spec. However, we want to
+        // surface stderr lines to improve the DX. Inject each line into the
+        // test output as an unknown token as if it came from the TAP parser.
+        subtest.addToReport({
           __proto__: null,
-          exitCode: code,
-          signal: signal,
-          // The stack will not be useful since the failures came from tests
-          // in a child process.
-          stack: undefined,
+          type: 'test:stderr',
+          data: { __proto__: null, file: path, message: line + '\n' },
         });
+      });
+
+      const { 0: { 0: code, 1: signal } } = await SafePromiseAll([
+        once(child, 'exit', { __proto__: null, signal: t.signal }),
+        finished(child.stdout, { __proto__: null, signal: t.signal }),
+      ]);
+
+      // Close readline interface to prevent memory leak
+      rl.close();
+
+      if (watchMode) {
+        filesWatcher.runningProcesses.delete(path);
+        filesWatcher.runningSubtests.delete(path);
+        (async () => {
+          try {
+            await subTestEnded;
+          } finally {
+            if (filesWatcher.runningSubtests.size === 0) {
+              opts.root.reporter[kEmitMessage]('test:watch:drained');
+              opts.root.postRun();
+            }
+          }
+        })();
       }
 
-      throw err;
+      if (code !== 0 || signal !== null) {
+        if (!err) {
+          const failureType = subtest.failedSubtests ? kSubtestsFailed : kTestCodeFailure;
+          err = ObjectAssign(new ERR_TEST_FAILURE('test failed', failureType), {
+            __proto__: null,
+            exitCode: code,
+            signal: signal,
+            // The stack will not be useful since the failures came from tests
+            // in a child process.
+            stack: undefined,
+          });
+        }
+
+        throw err;
+      }
+    } finally {
+      // Every exit path must return the ID, including abort and spawn failure.
+      if (opts.workerIdPool && workerId !== undefined) {
+        opts.workerIdPool.release(workerId);
+        debug('Released worker ID %d from test file: %s', workerId, path);
+      }
     }
   });
   const subTestEnded = subtest.start();
@@ -992,23 +1003,10 @@ function run(options = kEmptyObject) {
   let filesWatcher;
   let runFiles;
 
-  // Create worker ID pool for concurrent test execution.
-  // Use concurrency from globalOptions which has been processed by parseCommandLine().
-  const effectiveConcurrency = globalOptions.concurrency ?? concurrency;
-  let maxConcurrency = 1;
-  if (effectiveConcurrency === true) {
-    maxConcurrency = MathMax(availableParallelism() - 1, 1);
-  } else if (typeof effectiveConcurrency === 'number') {
-    maxConcurrency = effectiveConcurrency;
-  }
-  const workerIdPool = new WorkerIdPool(maxConcurrency);
-  debug(
-    'Created worker ID pool with max concurrency: %d, ' +
-    'effectiveConcurrency: %s, testFiles: %d',
-    maxConcurrency,
-    effectiveConcurrency,
-    testFiles.length,
-  );
+  // The pool tracks the IDs actually in use, so they stay exclusive and never
+  // exceed the number of files running concurrently.
+  const workerIdPool = new WorkerIdPool();
+  debug('Created worker ID pool, testFiles: %d', testFiles.length);
 
   const opts = {
     __proto__: null,