Repository: nodejs/node
Pull request: cluster: fix port reuse between cluster

Fixes: https://github.com/nodejs/node/issues/60086

If the `message.index` passed to [`queryServer`](https://github.com/nodejs/node/pull/60141/files#diff-6af1a5122ce960c2853fe60a767019d74ab5339a8e161f3ef1a13317d1fbfb57R274-R278) becomes out of sync due to errors or other issues, it may prevent a cluster from reusing a port that was acquired by another cluster.
To address this, I modified the code so that the handle key no longer includes the index (except when the port number is 0), and only the latest handle corresponding to a successfully acquired port is stored.


<!--
Before submitting a pull request, please read:

- the CONTRIBUTING guide at https://github.com/nodejs/node/blob/HEAD/CONTRIBUTING.md
- the commit message formatting guidelines at
  https://github.com/nodejs/node/blob/HEAD/doc/contributing/pull-requests.md#commit-message-guidelines

For code changes:
1. Include tests for any bug fixes or new features.
2. Update documentation if relevant.
3. Ensure that `make -j4 test` (UNIX), or `vcbuild test` (Windows) passes.

If you believe this PR should be highlighted in the Node.js CHANGELOG
please add the `notable-change` label.

Developer's Certificate of Origin 1.1

By making a contribution to this project, I certify that:

(a) The contribution was created in whole or in part by me and I
    have the right to submit it under the open source license
    indicated in the file; or

(b) The contribution is based upon previous work that, to the best
    of my knowledge, is covered under an appropriate open source
    license and I have the right under that license to submit that
    work with modifications, whether created in whole or in part
    by me, under the same open source license (unless I am
    permitted to submit under a different license), as indicated
    in the file; or

(c) The contribution was provided directly to me by some other
    person who certified (a), (b) or (c) and I have not modified
    it.

(d) I understand and agree that this project and the contribution
    are public and that a record of the contribution (including all
    personal information I submit with it, including my sign-off) is
    maintained indefinitely and may be redistributed consistent with
    this project or the open source license(s) involved.
-->


Changed files:
filename: lib/internal/cluster/primary.js
patch: @@ -271,8 +271,12 @@ function queryServer(worker, message) {
     return;
 
   const key = `${message.address}:${message.port}:${message.addressType}:` +
-              `${message.fd}:${message.index}`;
-  let handle = handles.get(key);
+              `${message.fd}` + (message.port === 0 ? `:${message.index}` : '');
+  const cachedHandle = handles.get(key);
+  let handle;
+  if (cachedHandle && !cachedHandle.has(worker)) {
+    handle = cachedHandle;
+  }
 
   if (handle === undefined) {
     let address = message.address;
@@ -298,25 +302,30 @@ function queryServer(worker, message) {
       handle = new RoundRobinHandle(key, address, message);
     }
 
-    handles.set(key, handle);
+    if (!cachedHandle) {
+      handles.set(key, handle);
+    }
   }
 
   handle.data ||= message.data;
 
   // Set custom server data
-  handle.add(worker, (errno, reply, handle) => {
+  handle.add(worker, (errno, reply, serverHandle) => {
+    if (!errno) {
+      handles.set(key, handle);  // Update in case it was replaced.
+    }
     const { data } = handles.get(key);
-
-    if (errno)
-      handles.delete(key);  // Gives other workers a chance to retry.
+    if (!cachedHandle && errno) {
+      handles.delete(key);
+    }
 
     send(worker, {
       errno,
       key,
       ack: message.seq,
       data,
       ...reply,
-    }, handle);
+    }, serverHandle);
   });
 }
 

filename: lib/internal/cluster/round_robin_handle.js
patch: @@ -137,3 +137,7 @@ RoundRobinHandle.prototype.handoff = function(worker) {
     this.handoff(worker);
   });
 };
+
+RoundRobinHandle.prototype.has = function(worker) {
+  return this.all.has(worker.id);
+};

filename: lib/internal/cluster/shared_handle.js
patch: @@ -47,3 +47,7 @@ SharedHandle.prototype.remove = function(worker) {
   this.handle = null;
   return true;
 };
+
+SharedHandle.prototype.has = function(worker) {
+  return this.workers.has(worker.id);
+};

filename: test/sequential/test-cluster-port-reuse-between-workers.js
patch: @@ -0,0 +1,93 @@
+'use strict';
+
+const common = require('../common');
+const cluster = require('cluster');
+const assert = require('assert');
+
+const acts = {
+  WORKER1_SERVER1_CLOSED: { cmd: 'WORKER1_SERVER1_CLOSED' },
+  WORKER2_SERVER1_STARTED: { cmd: 'WORKER2_SERVER1_STARTED' },
+  WORKER1_SERVER2_CLOSED: { cmd: 'WORKER1_SERVER2_CLOSED' },
+};
+
+if (cluster.isMaster) {
+  const currentHost = '::';
+  const worker1 = cluster.fork({
+    WORKER_ID: 'worker1',
+    HOST: currentHost,
+  });
+  let worker2;
+  worker1.on('error', common.mustNotCall());
+  worker1.on('message', onMessage);
+
+  function createWorker2() {
+    worker2 = cluster.fork({
+      WORKER_ID: 'worker2',
+      HOST: currentHost,
+    });
+    worker2.on('error', common.mustNotCall());
+    worker2.on('message', onMessage);
+  }
+
+  function onMessage(msg) {
+    switch (msg.cmd) {
+      case acts.WORKER1_SERVER1_CLOSED.cmd:
+        createWorker2();
+        break;
+      case acts.WORKER2_SERVER1_STARTED.cmd:
+        worker1.send(acts.WORKER2_SERVER1_STARTED);
+        break;
+      case acts.WORKER1_SERVER2_CLOSED.cmd:
+        worker1.kill();
+        worker2.kill();
+        break;
+      default:
+        assert.fail(`Unexpected message ${msg.cmd}`);
+    }
+  }
+} else {
+  const WORKER_ID = process.env.WORKER_ID;
+  function createServer() {
+    return new Promise((resolve, reject) => {
+      const net = require('net');
+      const PORT = 8000;
+      const server = net
+        .createServer((socket) => {
+          socket.end(
+            `Handled by worker ${process.env.WORKER_ID} (${process.pid})\n`
+          );
+        })
+        .on('error', (e) => {
+          reject(e);
+        });
+
+      server.listen(
+        {
+          port: PORT,
+          host: process.env.HOST,
+        },
+        () => resolve(server)
+      );
+    });
+  }
+  (async () => {
+    const server1 = await createServer();
+    if (WORKER_ID === 'worker2') {
+      process.send(acts.WORKER2_SERVER1_STARTED);
+    } else {
+      await createServer().catch(common.mustCall());
+      await new Promise((r) => server1.close(r));
+      process.send(acts.WORKER1_SERVER1_CLOSED);
+
+      process.on('message', async (msg) => {
+        if (msg.cmd === acts.WORKER2_SERVER1_STARTED.cmd) {
+          const server2 = await createServer();
+          await new Promise((r) => server2.close(r));
+          process.send(acts.WORKER1_SERVER2_CLOSED);
+        } else {
+          assert.fail(`Unexpected message ${msg.cmd}`);
+        }
+      });
+    }
+  })().then(common.mustCall());
+}