Reapply "fix(core): prevent phantom connections and dead polling in plugin workers (#34823)"

FrozenPandaz · FrozenPandaz · commit 468621f4c788 · 2026-04-09T17:44:28.000-04:00
This reverts commit 3a264f4.
diff --git a/packages/nx/src/daemon/client/client.ts b/packages/nx/src/daemon/client/client.ts
@@ -28,7 +28,9 @@ import {
 import { preventRecursionInGraphConstruction } from '../../project-graph/project-graph';
 import { ConfigurationSourceMaps } from '../../project-graph/utils/project-configuration/source-maps';
 import { isJsonMessage } from '../../utils/consume-messages-from-socket';
+import { waitForSocketConnection } from '../../utils/wait-for-socket-connection';
 import { DelayedSpinner } from '../../utils/delayed-spinner';
+import { handleImport } from '../../utils/handle-import';
 import { isCI } from '../../utils/is-ci';
 import { isSandbox } from '../../utils/is-sandbox';
 import { output } from '../../utils/output';
@@ -41,6 +43,13 @@ import { workspaceRoot } from '../../utils/workspace-root';
 import { getDaemonProcessIdSync, readDaemonProcessJsonCache } from '../cache';
 import { isNxVersionMismatch } from '../is-nx-version-mismatch';
 import { clientLogger } from '../logger';
+import {
+  type ConfigureAiAgentsStatusResponse,
+  GET_CONFIGURE_AI_AGENTS_STATUS,
+  type HandleGetConfigureAiAgentsStatusMessage,
+  type HandleResetConfigureAiAgentsStatusMessage,
+  RESET_CONFIGURE_AI_AGENTS_STATUS,
+} from '../message-types/configure-ai-agents';
 import {
   FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK,
   type HandleFlushSyncGeneratorChangesToDiskMessage,
@@ -83,13 +92,6 @@ import {
   SET_NX_CONSOLE_PREFERENCE_AND_INSTALL,
   type SetNxConsolePreferenceAndInstallResponse,
 } from '../message-types/nx-console';
-import {
-  GET_CONFIGURE_AI_AGENTS_STATUS,
-  RESET_CONFIGURE_AI_AGENTS_STATUS,
-  type ConfigureAiAgentsStatusResponse,
-  type HandleGetConfigureAiAgentsStatusMessage,
-  type HandleResetConfigureAiAgentsStatusMessage,
-} from '../message-types/configure-ai-agents';
 import { REGISTER_PROJECT_GRAPH_LISTENER } from '../message-types/register-project-graph-listener';
 import {
   HandlePostTasksExecutionMessage,
@@ -1173,35 +1175,34 @@ export class DaemonClient {
   private async waitForServerToBeAvailable(options: {
     ignoreVersionMismatch: boolean;
   }): Promise<boolean> {
-    let attempts = 0;
-
     clientLogger.log(
       `[Client] Waiting for server (max: ${WAIT_FOR_SERVER_CONFIG.maxAttempts} attempts, ${WAIT_FOR_SERVER_CONFIG.delayMs}ms interval)`
     );
 
-    while (attempts < WAIT_FOR_SERVER_CONFIG.maxAttempts) {
-      await new Promise((resolve) =>
-        setTimeout(resolve, WAIT_FOR_SERVER_CONFIG.delayMs)
-      );
-      attempts++;
-
-      try {
-        if (await this.isServerAvailable()) {
-          clientLogger.log(
-            `[Client] Server available after ${attempts} attempts`
-          );
-          return true;
-        }
-      } catch (err) {
-        if (err instanceof VersionMismatchError) {
-          if (!options.ignoreVersionMismatch) {
-            throw err;
+    const socket = await waitForSocketConnection(
+      () => {
+        try {
+          return this.getSocketPath();
+        } catch (err) {
+          if (err instanceof VersionMismatchError) {
+            if (!options.ignoreVersionMismatch) {
+              throw err;
+            }
           }
-          // Keep waiting - old cache file may exist
-        } else {
-          throw err;
+          // Socket path not available yet — keep polling
+          return null;
         }
+      },
+      {
+        maxAttempts: WAIT_FOR_SERVER_CONFIG.maxAttempts,
+        delayMs: WAIT_FOR_SERVER_CONFIG.delayMs,
       }
+    );
+
+    if (socket) {
+      socket.destroy();
+      clientLogger.log(`[Client] Server available`);
+      return true;
     }
 
     clientLogger.log(
@@ -1215,13 +1216,25 @@ export class DaemonClient {
     force?: 'v8' | 'json'
   ): Promise<any> {
     await this.startDaemonIfNecessary();
-    // An open promise isn't enough to keep the event loop
-    // alive, so we set a timeout here and clear it when we hear
-    // back
-    const keepAlive = setTimeout(() => {}, 10 * 60 * 1000);
+
+    let keepAlive: NodeJS.Timeout;
     return new Promise((resolve, reject) => {
       performance.mark('sendMessageToDaemon-start');
 
+      // An open promise isn't enough to keep the event loop
+      // alive, so we set a timeout here and clear it when we hear
+      // back. This **must** be longer than the message timeout used
+      // in the plugin isolation messages, or the daemon will timeout before
+      // a plugin worker would, and that can result in odd exit behavior.
+      keepAlive = setTimeout(
+        () => {
+          reject(
+            new Error('The daemon timed out while processing ' + message.type)
+          );
+        },
+        20 * 60 * 1000
+      );
+
       this.currentMessage = message;
       this.currentResolve = resolve;
       this.currentReject = reject;
diff --git a/packages/nx/src/project-graph/plugins/isolation/isolated-plugin.ts b/packages/nx/src/project-graph/plugins/isolation/isolated-plugin.ts
@@ -1,5 +1,5 @@
 import { ChildProcess, spawn } from 'child_process';
-import { Socket, connect } from 'net';
+import { Socket } from 'net';
 import { Readable, Writable } from 'stream';
 import path = require('path');
 
@@ -9,6 +9,7 @@ import { getPluginOsSocketPath } from '../../../daemon/socket-utils';
 import { consumeMessagesFromSocket } from '../../../utils/consume-messages-from-socket';
 import { getNxRequirePaths } from '../../../utils/installation-directory';
 import { logger } from '../../../utils/logger';
+import { waitForSocketConnection } from '../../../utils/wait-for-socket-connection';
 import type { RawProjectGraphDependency } from '../../project-graph-builder';
 import { LoadedNxPlugin } from '../loaded-nx-plugin';
 import type {
@@ -196,7 +197,12 @@ export class IsolatedPlugin implements LoadedNxPlugin {
 
     if (!this._connectPromise) {
       logger.verbose(`[plugin-client] restarting worker for "${this.name}"`);
-      this._connectPromise = this.spawnAndConnect();
+      this._connectPromise = this.spawnAndConnect().catch((err) => {
+        // Clear the cached promise so subsequent calls can retry
+        // instead of re-awaiting a permanently-rejected promise.
+        this._connectPromise = null;
+        throw err;
+      });
     }
 
     await this._connectPromise;
@@ -552,49 +558,55 @@ async function startPluginWorker(name: string) {
 
   worker.unref();
 
-  let attempts = 0;
-  return new Promise<{ worker: ChildProcess; socket: Socket }>(
-    (resolve, reject) => {
-      const id = setInterval(async () => {
-        const socket = await isServerAvailable(ipcPath);
-        if (socket) {
-          socket.unref();
-          clearInterval(id);
-          logger.verbose(
-            `[isolated-plugin] connected to worker for "${name}" (pid: ${worker.pid}) after ${attempts} attempt(s)`
-          );
-          resolve({ worker, socket });
-        } else if (attempts > 10000) {
-          clearInterval(id);
-          reject(new Error(`Failed to start plugin worker for plugin ${name}`));
-        } else {
-          attempts++;
-        }
-      }, 10);
-    }
-  ).finally(() => {
+  try {
+    const socket = await connectToWorker(worker, ipcPath, name);
+    return { worker, socket };
+  } finally {
     performance.mark(`start-plugin-worker-end:${name}`);
     performance.measure(
       `start-plugin-worker:${name}`,
       `start-plugin-worker:${name}`,
       `start-plugin-worker-end:${name}`
     );
-  });
+  }
 }
 
-function isServerAvailable(ipcPath: string): Promise<Socket | false> {
-  return new Promise((resolve) => {
-    try {
-      const socket = connect(ipcPath, () => {
-        resolve(socket);
-      });
-      socket.once('error', () => {
-        resolve(false);
-      });
-    } catch {
-      resolve(false);
+async function connectToWorker(
+  worker: ChildProcess,
+  ipcPath: string,
+  name: string
+): Promise<Socket> {
+  const abortController = new AbortController();
+  let earlyExitError: Error | null = null;
+
+  // If the worker exits before we connect, abort polling immediately
+  // rather than burning through attempts against a dead socket.
+  worker.once('exit', (code) => {
+    if (!abortController.signal.aborted) {
+      earlyExitError = new Error(
+        `Plugin worker for "${name}" exited with code ${code} before the connection was established.`
+      );
+      abortController.abort();
     }
   });
+
+  const socket = await waitForSocketConnection(ipcPath, {
+    signal: abortController.signal,
+  });
+
+  if (socket) {
+    abortController.abort();
+    socket.unref();
+    logger.verbose(
+      `[isolated-plugin] connected to worker for "${name}" (pid: ${worker.pid})`
+    );
+    return socket;
+  }
+
+  if (earlyExitError) {
+    throw earlyExitError;
+  }
+  throw new Error(`Failed to start plugin worker for plugin ${name}`);
 }
 
 function getTypeName(u: unknown): string {
diff --git a/packages/nx/src/project-graph/plugins/isolation/plugin-worker.ts b/packages/nx/src/project-graph/plugins/isolation/plugin-worker.ts
@@ -36,15 +36,18 @@ let plugin: LoadedNxPlugin;
 const socketPath = process.argv[2];
 const expectedPluginName = process.argv[3];
 
+const CONNECT_TIMEOUT_MS = 30_000;
+
 let connectErrorTimeout = setErrorTimeout(
-  5000,
-  `The plugin worker for ${expectedPluginName} is exiting as it was not connected to within 5 seconds. ` +
+  CONNECT_TIMEOUT_MS,
+  `The plugin worker for ${expectedPluginName} is exiting as it was not connected to within ${CONNECT_TIMEOUT_MS / 1000} seconds. ` +
     'Plugin workers expect to receive a socket connection from the parent process shortly after being started. ' +
     'If you are seeing this issue, please report it to the Nx team at https://github.com/nrwl/nx/issues.'
 );
 
 const server = createServer((socket) => {
   connectErrorTimeout?.clear();
+
   logger.verbose(
     `[plugin-worker] "${expectedPluginName}" (pid: ${process.pid}) connected`
   );
@@ -177,20 +180,13 @@ const server = createServer((socket) => {
     })
   );
 
-  // There should only ever be one host -> worker connection
-  // since the worker is spawned per host process. As such,
-  // we can safely close the worker when the host disconnects.
+  // When the host disconnects, clean up and exit.
   socket.on('end', () => {
-    // Destroys the socket once it's fully closed.
     socket.destroySoon();
-    // Stops accepting new connections, but existing connections are
-    // not closed immediately.
-    server.close(() => {
-      try {
-        unlinkSync(socketPath);
-      } catch (e) {}
-      process.exit(0);
-    });
+    try {
+      unlinkSync(socketPath);
+    } catch (e) {}
+    process.exit(0);
   });
 });
 
diff --git a/packages/nx/src/utils/wait-for-socket-connection.ts b/packages/nx/src/utils/wait-for-socket-connection.ts
@@ -0,0 +1,59 @@
+import { Socket, connect } from 'net';
+
+const DEFAULT_DELAY_MS = 10;
+const DEFAULT_MAX_ATTEMPTS = 10_000;
+
+/**
+ * Polls an IPC socket path until a connection succeeds or the attempt
+ * limit / abort signal is reached.
+ *
+ * @param socketPath - A fixed path string, or a function that resolves
+ *   the path on each attempt (useful when the server hasn't written its
+ *   socket file yet).
+ * @returns The connected socket, or `null` if polling was exhausted or aborted.
+ */
+export async function waitForSocketConnection(
+  socketPath: string | (() => string | null),
+  options?: {
+    signal?: AbortSignal;
+    maxAttempts?: number;
+    delayMs?: number;
+  }
+): Promise<Socket | null> {
+  const maxAttempts = options?.maxAttempts ?? DEFAULT_MAX_ATTEMPTS;
+  const delayMs = options?.delayMs ?? DEFAULT_DELAY_MS;
+  const signal = options?.signal;
+
+  let attempts = 0;
+  while (attempts < maxAttempts) {
+    if (signal?.aborted) return null;
+    await new Promise((r) => setTimeout(r, delayMs));
+    if (signal?.aborted) return null;
+
+    const path = typeof socketPath === 'function' ? socketPath() : socketPath;
+    if (path) {
+      const socket = await tryConnect(path);
+      if (socket) {
+        return socket;
+      }
+    }
+    attempts++;
+  }
+  return null;
+}
+
+function tryConnect(socketPath: string): Promise<Socket | null> {
+  return new Promise((resolve) => {
+    try {
+      const socket = connect(socketPath, () => {
+        resolve(socket);
+      });
+      socket.once('error', () => {
+        socket.destroy();
+        resolve(null);
+      });
+    } catch {
+      resolve(null);
+    }
+  });
+}
