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

jaysoo · jaysoo · commit 3a264f46924e · 2026-04-09T17:40:19.000-04:00
This reverts commit 1a8438f.
diff --git a/packages/nx/src/daemon/client/client.ts b/packages/nx/src/daemon/client/client.ts
@@ -28,9 +28,7 @@ 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';
@@ -43,13 +41,6 @@ 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,
@@ -92,6 +83,13 @@ 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,
@@ -1175,34 +1173,35 @@ 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)`
     );
 
-    const socket = await waitForSocketConnection(
-      () => {
-        try {
-          return this.getSocketPath();
-        } catch (err) {
-          if (err instanceof VersionMismatchError) {
-            if (!options.ignoreVersionMismatch) {
-              throw err;
-            }
+    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;
           }
-          // Socket path not available yet — keep polling
-          return null;
+          // Keep waiting - old cache file may exist
+        } else {
+          throw err;
         }
-      },
-      {
-        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(
@@ -1216,25 +1215,13 @@ export class DaemonClient {
     force?: 'v8' | 'json'
   ): Promise<any> {
     await this.startDaemonIfNecessary();
-
-    let keepAlive: NodeJS.Timeout;
+    // 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);
     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 } from 'net';
+import { Socket, connect } from 'net';
 import { Readable, Writable } from 'stream';
 import path = require('path');
 
@@ -9,7 +9,6 @@ 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 {
@@ -197,12 +196,7 @@ export class IsolatedPlugin implements LoadedNxPlugin {
 
     if (!this._connectPromise) {
       logger.verbose(`[plugin-client] restarting worker for "${this.name}"`);
-      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;
-      });
+      this._connectPromise = this.spawnAndConnect();
     }
 
     await this._connectPromise;
@@ -558,55 +552,49 @@ async function startPluginWorker(name: string) {
 
   worker.unref();
 
-  try {
-    const socket = await connectToWorker(worker, ipcPath, name);
-    return { worker, socket };
-  } finally {
+  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(() => {
     performance.mark(`start-plugin-worker-end:${name}`);
     performance.measure(
       `start-plugin-worker:${name}`,
       `start-plugin-worker:${name}`,
       `start-plugin-worker-end:${name}`
     );
-  }
+  });
 }
 
-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();
+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);
     }
   });
-
-  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,18 +36,15 @@ let plugin: LoadedNxPlugin;
 const socketPath = process.argv[2];
 const expectedPluginName = process.argv[3];
 
-const CONNECT_TIMEOUT_MS = 30_000;
-
 let connectErrorTimeout = setErrorTimeout(
-  CONNECT_TIMEOUT_MS,
-  `The plugin worker for ${expectedPluginName} is exiting as it was not connected to within ${CONNECT_TIMEOUT_MS / 1000} seconds. ` +
+  5000,
+  `The plugin worker for ${expectedPluginName} is exiting as it was not connected to within 5 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`
   );
@@ -180,13 +177,20 @@ const server = createServer((socket) => {
     })
   );
 
-  // When the host disconnects, clean up and exit.
+  // 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.
   socket.on('end', () => {
+    // Destroys the socket once it's fully closed.
     socket.destroySoon();
-    try {
-      unlinkSync(socketPath);
-    } catch (e) {}
-    process.exit(0);
+    // Stops accepting new connections, but existing connections are
+    // not closed immediately.
+    server.close(() => {
+      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
