fix(core): move plugin worker to socket

.env

@@ -0,0 +1 @@
+NX_ISOLATE_PLUGINS=true

.gitignore

@@ -37,7 +37,7 @@ out
 .angular
 
 # Local dev files
-.env
+.env.local
 .bashrc
 .nx
 

e2e/utils/get-env-info.ts

@@ -160,7 +160,7 @@ export function getStrippedEnvironmentVariables() {
         return true;
       }
 
-      const allowedKeys = ['NX_ADD_PLUGINS'];
+      const allowedKeys = ['NX_ADD_PLUGINS', 'NX_ISOLATE_PLUGINS'];
 
       if (key.startsWith('NX_') && !allowedKeys.includes(key)) {
         return false;

nx.json

@@ -212,6 +212,6 @@
   "nxCloudUrl": "https://staging.nx.app",
   "parallel": 1,
   "cacheDirectory": "/tmp/nx-cache",
-  "bust": 5,
+  "bust": 7,
   "defaultBase": "master"
 }

packages/nx/src/command-line/run/command-object.ts

@@ -37,7 +37,7 @@ export const yargsNxInfixCommand: CommandModule = {
   command: '$0 <target> [project] [_..]',
   describe: 'Run a target for a project',
   handler: async (args) => {
-    await handleErrors(
+    const exitCode = await handleErrors(
       (args.verbose as boolean) ?? process.env.NX_VERBOSE_LOGGING === 'true',
       async () => {
         return (await import('./run-one')).runOne(
@@ -46,5 +46,6 @@ export const yargsNxInfixCommand: CommandModule = {
         );
       }
     );
+    process.exit(exitCode);
   },
 };

packages/nx/src/daemon/socket-utils.ts

@@ -22,6 +22,11 @@ export const getForkedProcessOsSocketPath = (id: string) => {
   return isWindows ? '\\\\.\\pipe\\nx\\' + resolve(path) : resolve(path);
 };
 
+export const getPluginOsSocketPath = (id: string) => {
+  let path = resolve(join(getSocketDir(), 'plugin' + id + '.sock'));
+  return isWindows ? '\\\\.\\pipe\\nx\\' + resolve(path) : resolve(path);
+};
+
 export function killSocketOrPath(): void {
   try {
     unlinkSync(getFullOsSocketPath());

packages/nx/src/plugins/js/project-graph/affected/npm-packages.spec.ts

@@ -296,7 +296,7 @@ describe('getTouchedNpmPackages', () => {
   });
 
   it('should handle and log workspace package.json changes when the changes are not in `npmPackages` (projectGraph.externalNodes)', () => {
-    jest.spyOn(logger, 'warn');
+    jest.spyOn(logger, 'warn').mockImplementation(() => {});
     expect(() => {
       getTouchedNpmPackages(
         [

packages/nx/src/project-graph/plugins/internal-api.ts

@@ -159,7 +159,7 @@ export async function loadNxPlugins(
 
   const cleanupFunctions: Array<() => void> = [];
   for (const plugin of plugins) {
-    const [loadedPluginPromise, cleanup] = loadingMethod(plugin, root);
+    const [loadedPluginPromise, cleanup] = await loadingMethod(plugin, root);
     result.push(loadedPluginPromise);
     cleanupFunctions.push(cleanup);
   }

packages/nx/src/project-graph/plugins/isolation/index.ts

@@ -3,33 +3,15 @@ import { PluginConfiguration } from '../../../config/nx-json';
 import { LoadedNxPlugin } from '../internal-api';
 import { loadRemoteNxPlugin } from './plugin-pool';
 
-/**
- * Used to ensure 1 plugin : 1 worker
- */
-const remotePluginCache = new Map<
-  string,
-  readonly [Promise<LoadedNxPlugin>, () => void]
->();
-
-export function loadNxPluginInIsolation(
+export async function loadNxPluginInIsolation(
   plugin: PluginConfiguration,
   root = workspaceRoot
-): readonly [Promise<LoadedNxPlugin>, () => void] {
-  const cacheKey = JSON.stringify(plugin);
-
-  if (remotePluginCache.has(cacheKey)) {
-    return remotePluginCache.get(cacheKey);
-  }
-
-  const [loadingPlugin, cleanup] = loadRemoteNxPlugin(plugin, root);
-  // We clean up plugin workers when Nx process completes.
-  const val = [
+): Promise<readonly [Promise<LoadedNxPlugin>, () => void]> {
+  const [loadingPlugin, cleanup] = await loadRemoteNxPlugin(plugin, root);
+  return [
     loadingPlugin,
     () => {
       cleanup();
-      remotePluginCache.delete(cacheKey);
     },
   ] as const;
-  remotePluginCache.set(cacheKey, val);
-  return val;
 }

packages/nx/src/project-graph/plugins/isolation/messaging.ts

@@ -7,9 +7,11 @@ import {
   CreateDependenciesContext,
   CreateMetadataContext,
   CreateNodesContext,
+  CreateNodesContextV2,
 } from '../public-api';
 import { LoadedNxPlugin } from '../internal-api';
 import { Serializable } from 'child_process';
+import { Socket } from 'net';
 
 export interface PluginWorkerLoadMessage {
   type: 'load';
@@ -42,7 +44,7 @@ export interface PluginWorkerCreateNodesMessage {
   type: 'createNodes';
   payload: {
     configFiles: string[];
-    context: CreateNodesContext;
+    context: CreateNodesContextV2;
     tx: string;
   };
 }
@@ -159,6 +161,7 @@ export function isPluginWorkerMessage(
       'createNodes',
       'createDependencies',
       'processProjectGraph',
+      'createMetadata',
     ].includes(message.type)
   );
 }
@@ -175,6 +178,7 @@ export function isPluginWorkerResult(
       'createNodesResult',
       'createDependenciesResult',
       'processProjectGraphResult',
+      'createMetadataResult',
     ].includes(message.type)
   );
 }
@@ -192,6 +196,7 @@ type MessageHandlerReturn<T extends PluginWorkerMessage | PluginWorkerResult> =
 export async function consumeMessage<
   T extends PluginWorkerMessage | PluginWorkerResult
 >(
+  socket: Socket,
   raw: T,
   handlers: {
     [K in T['type']]: (
@@ -205,7 +210,14 @@ export async function consumeMessage<
   if (handler) {
     const response = await handler(message.payload);
     if (response) {
-      process.send!(response);
+      sendMessageOverSocket(socket, response);
     }
   }
 }
+
+export function sendMessageOverSocket(
+  socket: Socket,
+  message: PluginWorkerMessage | PluginWorkerResult
+) {
+  socket.write(JSON.stringify(message) + String.fromCodePoint(4));
+}