lib: fix benchmark record acknowledgement race

The child can receive an acknowledgement and send its next record before
the parent's send callback runs. Clear recordPending before sending the
acknowledgement to avoid rejecting that record as an invalid sequence.

Add a regression test that delays send callbacks until the child sends
its next record or closes.

Signed-off-by: Filip Skokan <panva.ip@gmail.com>
Assisted-by: Codex

Author: panva

lib/internal/bench_runner/cli.js

@@ -954,10 +954,11 @@ async function runChild(path, options, scope, onRecord) {
         }
         const pending = handleRecord(record);
         if (protocolError === undefined) {
-          const acknowledged = PromisePrototypeThen(
-            PromiseResolve(pending), () => sendAck(child, message.id));
-          trackPending(PromisePrototypeThen(acknowledged, () => {
+          trackPending(PromisePrototypeThen(PromiseResolve(pending), () => {
+            // The child can receive the acknowledgement and send its next
+            // record before the send callback runs.
             recordPending = false;
+            return sendAck(child, message.id);
           }));
         }
       } catch (error) {

test/fixtures/bench-runner/delayed-ack.cjs

@@ -0,0 +1,31 @@
+'use strict';
+
+const common = require('../../common');
+const childProcess = require('child_process');
+const { spawn } = childProcess;
+
+childProcess.spawn = (...args) => {
+  const child = spawn(...args);
+  const { send } = child;
+  child.send = function(message, handle, options, callback) {
+    if (message?.type === 'node:bench:ack') {
+      // Report send completion only after the child has responded to the ack.
+      const onComplete = common.mustCall(() => {
+        child.removeListener('message', onComplete);
+        child.removeListener('close', onComplete);
+        callback(null);
+      });
+      child.once('message', onComplete);
+      child.once('close', onComplete);
+      return send.call(this, message, handle, options, (error) => {
+        if (error) {
+          child.removeListener('message', onComplete);
+          child.removeListener('close', onComplete);
+          callback(error);
+        }
+      });
+    }
+    return send.call(this, message, handle, options, callback);
+  };
+  return child;
+};

test/parallel/test-bench-cli-ack.js

@@ -0,0 +1,23 @@
+'use strict';
+
+const common = require('../common');
+const assert = require('assert');
+const { spawnSync } = require('child_process');
+const fixtures = require('../common/fixtures');
+
+const result = spawnSync(process.execPath, [
+  '--no-warnings', '--experimental-bench', '--bench',
+  '--require', fixtures.path('bench-runner/delayed-ack.cjs'),
+  '--bench-reporter=json',
+  fixtures.path('bench-runner/acknowledged-records.mjs'),
+], { encoding: 'utf8', timeout: common.platformTimeout(30_000) });
+
+assert.ifError(result.error);
+assert.strictEqual(result.signal, null);
+assert.strictEqual(result.status, 0, result.stdout + result.stderr);
+assert.strictEqual(result.stderr, '');
+const records = result.stdout.trim().split('\n').map((line) => JSON.parse(line));
+const diagnostics = records.filter(({ type }) => type === 'bench:diagnostic');
+assert.strictEqual(diagnostics.length, 32);
+assert(diagnostics.every(({ data }) => /^acknowledged 10\d{3}$/.test(data.message)));
+assert.strictEqual(records.at(-1).data.success, true);