fix(core): ensure daemon socket messenger is set on first message

.circleci/config.yml

@@ -82,7 +82,7 @@ jobs:
       NX_E2E_CI_CACHE_KEY: e2e-circleci-linux
       NX_DAEMON: 'true'
       NX_PERF_LOGGING: 'false'
-      NX_VERBOSE_LOGGING: 'false'
+      NX_VERBOSE_LOGGING: 'true'
       NX_NATIVE_LOGGING: 'false'
       NX_E2E_RUN_E2E: 'true'
       NX_CI_EXECUTION_ENV: 'linux'

packages/nx/src/daemon/client/client.ts

@@ -81,12 +81,38 @@ import {
   DelayedSpinner,
   SHOULD_SHOW_SPINNERS,
 } from '../../utils/delayed-spinner';
+import { PendingPromise, registerPendingPromise } from '../../utils/messaging';
+import {
+  HandleRecordOutputsHashMessage,
+  RECORD_OUTPUTS_HASH,
+} from '../message-types/record-outputs-hash';
 
 const DAEMON_ENV_SETTINGS = {
   NX_PROJECT_GLOB_CACHE: 'false',
   NX_CACHE_PROJECTS_CONFIG: 'false',
 };
 
+const DAEMON_TIMEOUT_HINT_TEXT =
+  'As a last resort, you can set NX_DAEMON_NO_TIMEOUTS=true to bypass this timeout.';
+
+const MINUTES = 15;
+
+const MAX_MESSAGE_WAIT =
+  process.env.NX_DAEMON_NO_TIMEOUTS === 'true'
+    ? // Registering a timeout prevents the process from exiting
+      // if the call to a plugin happens to be the only thing
+      // keeping the process alive. As such, even if timeouts are disabled
+      // we need to register one. 2147483647 is the max timeout
+      // that Node.js allows, and is equivalent to 24.8 days....
+      // This does mean that the NX_PLUGIN_NO_TIMEOUTS env var
+      // would still timeout after 24.8 days, but that seems
+      // like a reasonable compromise.
+      2147483647
+    : 1000 * 60 * MINUTES; // 10 minutes
+
+const getTimeoutErrorMessage = (messageType: string) => () =>
+  `The daemon process has not responded to the ${messageType} message within ${MINUTES} minutes. ${DAEMON_TIMEOUT_HINT_TEXT}`;
+
 export type UnregisterCallback = () => void;
 export type ChangedFile = {
   path: string;
@@ -114,6 +140,9 @@ export class DaemonClient {
   private queue: PromisedBasedQueue;
   private socketMessenger: DaemonSocketMessenger;
 
+  private pendingPromises: Map<string, PendingPromise> = new Map();
+  private pendingMessages: Map<string, Message> = new Map();
+
   private currentMessage;
   private currentResolve;
   private currentReject;
@@ -124,6 +153,11 @@ export class DaemonClient {
   private _daemonReady: () => void | null = null;
   private _out: FileHandle = null;
   private _err: FileHandle = null;
+  private txId: number = 0;
+
+  getNextTxId(type: string) {
+    return `${process.pid}:${type}:${this.txId++}`;
+  }
 
   enabled() {
     if (this._enabled === undefined) {
@@ -177,6 +211,7 @@ export class DaemonClient {
     this.currentMessage = null;
     this.currentResolve = null;
     this.currentReject = null;
+    this.pendingPromises.clear();
     this._enabled = undefined;
 
     this._out?.close();
@@ -190,8 +225,44 @@ export class DaemonClient {
     );
   }
 
+  private sendMessageToDaemonViaTx<
+    T extends { type: string } = { type: string }
+  >(message: T, socket: DaemonSocketMessenger) {
+    const tx = this.getNextTxId(message.type);
+    this.pendingMessages.set(tx, message);
+    return registerPendingPromise(
+      tx,
+      this.pendingPromises,
+      () => {
+        this.sendMessageToDaemon({ ...message, tx }, socket);
+      },
+      getTimeoutErrorMessage(message.type),
+      MAX_MESSAGE_WAIT
+    );
+  }
+
+  private async sendToDaemonViaQueue(
+    messageToDaemon: Message,
+    socket: DaemonSocketMessenger
+  ): Promise<any> {
+    return this.queue.sendToQueue(() =>
+      this.sendMessageToDaemon(messageToDaemon, socket)
+    );
+  }
+
+  private async sendMessageToDaemonViaQueueOrTx<T extends Message>(
+    message: T,
+    socket = this.socketMessenger
+  ) {
+    if (process.env.NX_DAEMON_USE_QUEUE === 'true') {
+      return this.sendToDaemonViaQueue(message, socket);
+    } else {
+      return this.sendMessageToDaemonViaTx(message, socket);
+    }
+  }
+
   async requestShutdown(): Promise<void> {
-    return this.sendToDaemonViaQueue({ type: 'REQUEST_SHUTDOWN' });
+    return this.sendMessageToDaemonViaQueueOrTx({ type: 'REQUEST_SHUTDOWN' });
   }
 
   async getProjectGraphAndSourceMaps(): Promise<{
@@ -210,7 +281,7 @@ export class DaemonClient {
       );
     }
     try {
-      const response = await this.sendToDaemonViaQueue({
+      const response = await this.sendMessageToDaemonViaQueueOrTx({
         type: 'REQUEST_PROJECT_GRAPH',
       });
       return {
@@ -229,7 +300,9 @@ export class DaemonClient {
   }
 
   async getAllFileData(): Promise<FileData[]> {
-    return await this.sendToDaemonViaQueue({ type: 'REQUEST_FILE_DATA' });
+    return this.sendMessageToDaemonViaQueueOrTx({
+      type: 'REQUEST_ALL_FILE_DATA',
+    });
   }
 
   hashTasks(
@@ -238,7 +311,7 @@ export class DaemonClient {
     taskGraph: TaskGraph,
     env: NodeJS.ProcessEnv
   ): Promise<Hash[]> {
-    return this.sendToDaemonViaQueue({
+    return this.sendMessageToDaemonViaQueueOrTx({
       type: 'HASH_TASKS',
       runnerOptions,
       env,
@@ -271,12 +344,9 @@ export class DaemonClient {
         throw e;
       }
     }
-    let messenger: DaemonSocketMessenger | undefined;
 
-    await this.queue.sendToQueue(() => {
-      messenger = new DaemonSocketMessenger(
-        connect(getFullOsSocketPath())
-      ).listen(
+    let messenger: DaemonSocketMessenger | undefined =
+      new DaemonSocketMessenger(connect(getFullOsSocketPath())).listen(
         (message) => {
           try {
             const parsedMessage = JSON.parse(message);
@@ -290,8 +360,14 @@ export class DaemonClient {
         },
         (err) => callback(err, null)
       );
-      return messenger.sendMessage({ type: 'REGISTER_FILE_WATCHER', config });
-    });
+
+    await this.sendMessageToDaemonViaQueueOrTx(
+      {
+        type: 'REGISTER_FILE_WATCHER',
+        config,
+      },
+      messenger
+    );
 
     return () => {
       messenger?.close();
@@ -299,7 +375,7 @@ export class DaemonClient {
   }
 
   processInBackground(requirePath: string, data: any): Promise<any> {
-    return this.sendToDaemonViaQueue({
+    return this.sendMessageToDaemonViaQueueOrTx({
       type: 'PROCESS_IN_BACKGROUND',
       requirePath,
       data,
@@ -307,17 +383,19 @@ export class DaemonClient {
   }
 
   recordOutputsHash(outputs: string[], hash: string): Promise<any> {
-    return this.sendToDaemonViaQueue({
-      type: 'RECORD_OUTPUTS_HASH',
-      data: {
-        outputs,
-        hash,
-      },
-    });
+    return this.sendMessageToDaemonViaQueueOrTx<HandleRecordOutputsHashMessage>(
+      {
+        type: RECORD_OUTPUTS_HASH,
+        data: {
+          outputs,
+          hash,
+        },
+      }
+    );
   }
 
   outputsHashesMatch(outputs: string[], hash: string): Promise<any> {
-    return this.sendToDaemonViaQueue({
+    return this.sendMessageToDaemonViaQueueOrTx({
       type: 'OUTPUTS_HASHES_MATCH',
       data: {
         outputs,
@@ -327,104 +405,99 @@ export class DaemonClient {
   }
 
   glob(globs: string[], exclude?: string[]): Promise<string[]> {
-    const message: HandleGlobMessage = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleGlobMessage>({
       type: 'GLOB',
       globs,
       exclude,
-    };
-    return this.sendToDaemonViaQueue(message);
+    });
   }
 
   getWorkspaceContextFileData(): Promise<FileData[]> {
-    const message: HandleContextFileDataMessage = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleContextFileDataMessage>({
       type: GET_CONTEXT_FILE_DATA,
-    };
-    return this.sendToDaemonViaQueue(message);
+    });
   }
 
   getWorkspaceFiles(
     projectRootMap: Record<string, string>
   ): Promise<NxWorkspaceFiles> {
-    const message: HandleNxWorkspaceFilesMessage = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleNxWorkspaceFilesMessage>({
       type: GET_NX_WORKSPACE_FILES,
       projectRootMap,
-    };
-    return this.sendToDaemonViaQueue(message);
+    });
   }
 
   getFilesInDirectory(dir: string): Promise<string[]> {
-    const message: HandleGetFilesInDirectoryMessage = {
-      type: GET_FILES_IN_DIRECTORY,
-      dir,
-    };
-    return this.sendToDaemonViaQueue(message);
+    return this.sendMessageToDaemonViaQueueOrTx<HandleGetFilesInDirectoryMessage>(
+      {
+        type: GET_FILES_IN_DIRECTORY,
+        dir,
+      }
+    );
   }
 
   hashGlob(globs: string[], exclude?: string[]): Promise<string> {
-    const message: HandleHashGlobMessage = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleHashGlobMessage>({
       type: HASH_GLOB,
       globs,
       exclude,
-    };
-    return this.sendToDaemonViaQueue(message);
+    });
   }
 
   getFlakyTasks(hashes: string[]): Promise<string[]> {
-    const message: HandleGetFlakyTasks = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleGetFlakyTasks>({
       type: GET_FLAKY_TASKS,
       hashes,
-    };
-
-    return this.sendToDaemonViaQueue(message);
+    });
   }
 
   async getEstimatedTaskTimings(
     targets: TaskTarget[]
   ): Promise<Record<string, number>> {
-    const message: HandleGetEstimatedTaskTimings = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleGetEstimatedTaskTimings>({
       type: GET_ESTIMATED_TASK_TIMINGS,
       targets,
-    };
-
-    return this.sendToDaemonViaQueue(message);
+    });
   }
 
   recordTaskRuns(taskRuns: TaskRun[]): Promise<void> {
-    const message: HandleRecordTaskRunsMessage = {
+    return this.sendMessageToDaemonViaQueueOrTx<HandleRecordTaskRunsMessage>({
       type: RECORD_TASK_RUNS,
       taskRuns,
-    };
-    return this.sendMessageToDaemon(message);
+    });
   }
 
   getSyncGeneratorChanges(
     generators: string[]
   ): Promise<SyncGeneratorRunResult[]> {
-    const message: HandleGetSyncGeneratorChangesMessage = {
-      type: GET_SYNC_GENERATOR_CHANGES,
-      generators,
-    };
-    return this.sendToDaemonViaQueue(message);
+    return this.sendMessageToDaemonViaQueueOrTx<HandleGetSyncGeneratorChangesMessage>(
+      {
+        type: GET_SYNC_GENERATOR_CHANGES,
+        generators,
+      }
+    );
   }
 
   flushSyncGeneratorChangesToDisk(
     generators: string[]
   ): Promise<FlushSyncGeneratorChangesResult> {
-    const message: HandleFlushSyncGeneratorChangesToDiskMessage = {
-      type: FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK,
-      generators,
-    };
-    return this.sendToDaemonViaQueue(message);
+    return this.sendMessageToDaemonViaQueueOrTx<HandleFlushSyncGeneratorChangesToDiskMessage>(
+      {
+        type: FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK,
+        generators,
+      }
+    );
   }
 
   getRegisteredSyncGenerators(): Promise<{
     globalGenerators: string[];
     taskGenerators: string[];
   }> {
-    const message: HandleGetRegisteredSyncGeneratorsMessage = {
-      type: GET_REGISTERED_SYNC_GENERATORS,
-    };
-    return this.sendToDaemonViaQueue(message);
+    return this.sendMessageToDaemonViaQueueOrTx<HandleGetRegisteredSyncGeneratorsMessage>(
+      {
+        type: GET_REGISTERED_SYNC_GENERATORS,
+      }
+    );
   }
 
   updateWorkspaceContext(
@@ -432,13 +505,14 @@ export class DaemonClient {
     updatedFiles: string[],
     deletedFiles: string[]
   ): Promise<void> {
-    const message: HandleUpdateWorkspaceContextMessage = {
-      type: UPDATE_WORKSPACE_CONTEXT,
-      createdFiles,
-      updatedFiles,
-      deletedFiles,
-    };
-    return this.sendToDaemonViaQueue(message);
+    return this.sendMessageToDaemonViaQueueOrTx<HandleUpdateWorkspaceContextMessage>(
+      {
+        type: UPDATE_WORKSPACE_CONTEXT,
+        createdFiles,
+        updatedFiles,
+        deletedFiles,
+      }
+    );
   }
 
   async isServerAvailable(): Promise<boolean> {
@@ -457,10 +531,11 @@ export class DaemonClient {
     });
   }
 
-  private async sendToDaemonViaQueue(messageToDaemon: Message): Promise<any> {
-    return this.queue.sendToQueue(() =>
-      this.sendMessageToDaemon(messageToDaemon)
-    );
+  private rejectAllPendingPromises(err) {
+    this.pendingPromises.forEach((pending) => {
+      pending.rejector(err);
+    });
+    this.pendingPromises.clear();
   }
 
   private setUpConnection() {
@@ -471,7 +546,7 @@ export class DaemonClient {
       () => {
         // it's ok for the daemon to terminate if the client doesn't wait on
         // any messages from the daemon
-        if (this.queue.isEmpty()) {
+        if (this.pendingPromises.size === 0) {
           this.reset();
         } else {
           output.error({
@@ -482,9 +557,11 @@ export class DaemonClient {
             ],
           });
           this._daemonStatus = DaemonStatus.DISCONNECTED;
-          this.currentReject?.(
-            daemonProcessException(
-              'Daemon process terminated and closed the connection'
+          this.pendingPromises.forEach((pending) =>
+            pending.rejector?.(
+              daemonProcessException(
+                'Daemon process terminated and closed the connection'
+              )
             )
           );
           process.exit(1);
@@ -492,19 +569,23 @@ export class DaemonClient {
       },
       (err) => {
         if (!err.message) {
-          return this.currentReject(daemonProcessException(err.toString()));
-        }
-
-        if (err.message.startsWith('LOCK-FILES-CHANGED')) {
-          // retry the current message
-          // we cannot send it via the queue because we are in the middle of processing
-          // a message from the queue
-          return this.sendMessageToDaemon(this.currentMessage).then(
-            this.currentResolve,
-            this.currentReject
+          return this.rejectAllPendingPromises(
+            daemonProcessException(err.toString())
           );
         }
 
+        // TODO: @AgentEnder - figure out how to handle this... Not sure this is
+        // actually even being used.
+        // if (err.message.startsWith('LOCK-FILES-CHANGED')) {
+        //   // retry the current message
+        //   // we cannot send it via the queue because we are in the middle of processing
+        //   // a message from the queue
+        //   return this.sendMessageToDaemon(this.currentMessage).then(
+        //     this.currentResolve,
+        //     this.currentReject
+        //   );
+        // }
+
         let error: any;
         if (err.message.startsWith('connect ENOENT')) {
           error = daemonProcessException('The Daemon Server is not running');
@@ -520,12 +601,15 @@ export class DaemonClient {
         } else {
           error = daemonProcessException(err.toString());
         }
-        return this.currentReject(error);
+        return this.rejectAllPendingPromises(error);
       }
     );
   }
 
-  private async sendMessageToDaemon(message: Message): Promise<any> {
+  private async sendMessageToDaemon(
+    message: Message,
+    socket: DaemonSocketMessenger = this.socketMessenger
+  ): Promise<any> {
     if (this._daemonStatus == DaemonStatus.DISCONNECTED) {
       this._daemonStatus = DaemonStatus.CONNECTING;
 
@@ -533,65 +617,77 @@ export class DaemonClient {
         await this.startInBackground();
       }
       this.setUpConnection();
+      socket ??= this.socketMessenger;
       this._daemonStatus = DaemonStatus.CONNECTED;
       this._daemonReady();
     } else if (this._daemonStatus == DaemonStatus.CONNECTING) {
       await this._waitForDaemonReady;
     }
-    // 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');
-
-      this.currentMessage = message;
-      this.currentResolve = resolve;
-      this.currentReject = reject;
-
-      this.socketMessenger.sendMessage(message);
-    }).finally(() => {
-      clearTimeout(keepAlive);
-    });
+    socket.sendMessage(message);
   }
 
   private handleMessage(serializedResult: string) {
     try {
       performance.mark('json-parse-start');
-      const parsedResult = JSON.parse(serializedResult);
+      const { response: parsedResult, tx } = JSON.parse(serializedResult);
       performance.mark('json-parse-end');
       performance.measure(
         'deserialize daemon response',
         'json-parse-start',
         'json-parse-end'
       );
-      if (parsedResult.error) {
-        this.currentReject(parsedResult.error);
+      if (tx) {
+        const pending = this.pendingPromises.get(tx);
+        if (pending) {
+          if (parsedResult.error) {
+            pending.rejector(parsedResult.error);
+          } else {
+            pending.resolver(parsedResult);
+          }
+          this.pendingPromises.delete(tx);
+        } else {
+        }
+      } else if (process.env.NX_DAEMON_USE_QUEUE === 'true') {
+        if (parsedResult.error) {
+          this.currentReject(parsedResult.error);
+        } else {
+          performance.measure(
+            'total for sendMessageToDaemon()',
+            'sendMessageToDaemon-start',
+            'json-parse-end'
+          );
+          return this.currentResolve(parsedResult);
+        }
       } else {
-        performance.measure(
-          'total for sendMessageToDaemon()',
-          'sendMessageToDaemon-start',
-          'json-parse-end'
-        );
-        return this.currentResolve(parsedResult);
+        output.error({
+          title: `Received a response without a transaction ID: ${JSON.stringify(
+            parsedResult
+          )}`,
+          bodyLines: [
+            'This is likely a bug in Nx. Please report it.',
+            'The response will be ignored.',
+          ],
+        });
       }
     } catch (e) {
       const endOfResponse =
         serializedResult.length > 300
           ? serializedResult.substring(serializedResult.length - 300)
           : serializedResult;
-      this.currentReject(
-        daemonProcessException(
-          [
-            'Could not deserialize response from Nx daemon.',
-            `Message: ${e.message}`,
-            '\n',
-            `Received:`,
-            endOfResponse,
-            '\n',
-          ].join('\n')
-        )
-      );
+      this.pendingPromises.forEach((pending) => {
+        pending.rejector(
+          daemonProcessException(
+            [
+              'Could not deserialize response from Nx daemon.',
+              `Message: ${e.message}`,
+              '\n',
+              `Received:`,
+              endOfResponse,
+              '\n',
+            ].join('\n')
+          )
+        );
+      });
     }
   }
 
@@ -652,18 +748,17 @@ export class DaemonClient {
   }
 
   async stop(): Promise<void> {
-    try {
-      await this.sendMessageToDaemon({ type: FORCE_SHUTDOWN });
-      await waitForDaemonToExitAndCleanupProcessJson();
-    } catch (err) {
-      output.error({
-        title:
-          err?.message ||
-          'Something unexpected went wrong when stopping the daemon server',
+    return this.sendMessageToDaemonViaQueueOrTx({ type: 'FORCE_SHUTDOWN' })
+      .catch((e) => {
+        output.error({
+          title:
+            e?.message ||
+            'Something unexpected went wrong when stopping the daemon server',
+        });
+      })
+      .then(() => {
+        removeSocketDir();
       });
-    }
-
-    removeSocketDir();
   }
 }
 

packages/nx/src/daemon/client/daemon-socket-messenger.ts

@@ -1,11 +1,12 @@
 import { randomUUID } from 'crypto';
 import { Socket } from 'net';
 import { performance } from 'perf_hooks';
-import { consumeMessagesFromSocket } from '../../utils/consume-messages-from-socket';
+import { consumeMessagesFromSocket } from '../../utils/messaging';
 
 export interface Message extends Record<string, any> {
   type: string;
   data?: any;
+  tx?: string;
 }
 
 export class DaemonSocketMessenger {

packages/nx/src/daemon/message-types/flush-sync-generator-changes-to-disk.ts

@@ -1,7 +1,9 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK =
   'CLEAR_CACHED_SYNC_GENERATOR_CHANGES' as const;
 
-export type HandleFlushSyncGeneratorChangesToDiskMessage = {
+export type HandleFlushSyncGeneratorChangesToDiskMessage = Message & {
   type: typeof FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK;
   generators: string[];
 };

packages/nx/src/daemon/message-types/force-shutdown.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const FORCE_SHUTDOWN = 'FORCE_SHUTDOWN' as const;
 
-export type HandleForceShutdownMessage = {
+export type HandleForceShutdownMessage = Message & {
   type: typeof FORCE_SHUTDOWN;
 };
 

packages/nx/src/daemon/message-types/get-context-file-data.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const GET_CONTEXT_FILE_DATA = 'GET_CONTEXT_FILE_DATA' as const;
 
-export type HandleContextFileDataMessage = {
+export type HandleContextFileDataMessage = Message & {
   type: typeof GET_CONTEXT_FILE_DATA;
 };
 

packages/nx/src/daemon/message-types/get-files-in-directory.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const GET_FILES_IN_DIRECTORY = 'GET_FILES_IN_DIRECTORY' as const;
 
-export type HandleGetFilesInDirectoryMessage = {
+export type HandleGetFilesInDirectoryMessage = Message & {
   type: typeof GET_FILES_IN_DIRECTORY;
   dir: string;
 };

packages/nx/src/daemon/message-types/get-nx-workspace-files.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const GET_NX_WORKSPACE_FILES = 'GET_NX_WORKSPACE_FILES' as const;
 
-export type HandleNxWorkspaceFilesMessage = {
+export type HandleNxWorkspaceFilesMessage = Message & {
   type: typeof GET_NX_WORKSPACE_FILES;
   projectRootMap: Record<string, string>;
 };

packages/nx/src/daemon/message-types/get-registered-sync-generators.ts

@@ -1,7 +1,9 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const GET_REGISTERED_SYNC_GENERATORS =
   'GET_REGISTERED_SYNC_GENERATORS' as const;
 
-export type HandleGetRegisteredSyncGeneratorsMessage = {
+export type HandleGetRegisteredSyncGeneratorsMessage = Message & {
   type: typeof GET_REGISTERED_SYNC_GENERATORS;
 };
 

packages/nx/src/daemon/message-types/get-sync-generator-changes.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const GET_SYNC_GENERATOR_CHANGES = 'GET_SYNC_GENERATOR_CHANGES' as const;
 
-export type HandleGetSyncGeneratorChangesMessage = {
+export type HandleGetSyncGeneratorChangesMessage = Message & {
   type: typeof GET_SYNC_GENERATOR_CHANGES;
   generators: string[];
 };

packages/nx/src/daemon/message-types/glob.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const GLOB = 'GLOB' as const;
 
-export type HandleGlobMessage = {
+export type HandleGlobMessage = Message & {
   type: typeof GLOB;
   globs: string[];
   exclude?: string[];

packages/nx/src/daemon/message-types/hash-glob.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const HASH_GLOB = 'HASH_GLOB' as const;
 
-export type HandleHashGlobMessage = {
+export type HandleHashGlobMessage = Message & {
   type: typeof HASH_GLOB;
   globs: string[];
   exclude?: string[];

packages/nx/src/daemon/message-types/hash-tasks.ts

@@ -0,0 +1,23 @@
+import { Task, TaskGraph } from '../../config/task-graph';
+import { Message } from '../client/daemon-socket-messenger';
+
+export const HASH_TASKS = 'HASH_TASKS' as const;
+
+export type HandleHashTasksMessage = Message & {
+  type: typeof HASH_TASKS;
+  runnerOptions: any;
+  env: any;
+  tasks: Task[];
+  taskGraph: TaskGraph;
+};
+
+export function isHandleHashTasksMessage(
+  message: unknown
+): message is HandleHashTasksMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === HASH_TASKS
+  );
+}

packages/nx/src/daemon/message-types/output-hashes-match.ts

@@ -0,0 +1,20 @@
+import { Task, TaskGraph } from '../../config/task-graph';
+import { Message } from '../client/daemon-socket-messenger';
+
+export const OUTPUTS_HASHES_MATCH = 'OUTPUTS_HASHES_MATCH' as const;
+
+export type HandleOutputHashesMatchMessage = Message & {
+  type: typeof OUTPUTS_HASHES_MATCH;
+  data: { outputs: string[]; hash: string };
+};
+
+export function isHandleOutputHashesMatchMessage(
+  message: unknown
+): message is HandleOutputHashesMatchMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === OUTPUTS_HASHES_MATCH
+  );
+}

packages/nx/src/daemon/message-types/process-in-background.ts

@@ -0,0 +1,21 @@
+import { Task, TaskGraph } from '../../config/task-graph';
+import { Message } from '../client/daemon-socket-messenger';
+
+export const PROCESS_IN_BACKGROUND = 'PROCESS_IN_BACKGROUND' as const;
+
+export type HandleProcessInBackgroundMessage = Message & {
+  type: typeof PROCESS_IN_BACKGROUND;
+  requirePath: string;
+  data: any;
+};
+
+export function isHandleProcessInBackgroundMessageMessage(
+  message: unknown
+): message is HandleProcessInBackgroundMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === PROCESS_IN_BACKGROUND
+  );
+}

packages/nx/src/daemon/message-types/record-outputs-hash.ts

@@ -0,0 +1,20 @@
+import { Task, TaskGraph } from '../../config/task-graph';
+import { Message } from '../client/daemon-socket-messenger';
+
+export const RECORD_OUTPUTS_HASH = 'RECORD_OUTPUTS_HASH' as const;
+
+export type HandleRecordOutputsHashMessage = Message & {
+  type: typeof RECORD_OUTPUTS_HASH;
+  data: { outputs: string[]; hash: string };
+};
+
+export function isHandleRecordOutputsHashMessage(
+  message: unknown
+): message is HandleRecordOutputsHashMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === RECORD_OUTPUTS_HASH
+  );
+}

packages/nx/src/daemon/message-types/task-history.ts

@@ -1,20 +1,21 @@
 import type { TaskRun, TaskTarget } from '../../native';
+import { Message } from '../client/daemon-socket-messenger';
 
 export const GET_FLAKY_TASKS = 'GET_FLAKY_TASKS' as const;
 export const GET_ESTIMATED_TASK_TIMINGS = 'GET_ESTIMATED_TASK_TIMINGS' as const;
 export const RECORD_TASK_RUNS = 'RECORD_TASK_RUNS' as const;
 
-export type HandleGetFlakyTasks = {
+export type HandleGetFlakyTasks = Message & {
   type: typeof GET_FLAKY_TASKS;
   hashes: string[];
 };
 
-export type HandleGetEstimatedTaskTimings = {
+export type HandleGetEstimatedTaskTimings = Message & {
   type: typeof GET_ESTIMATED_TASK_TIMINGS;
   targets: TaskTarget[];
 };
 
-export type HandleRecordTaskRunsMessage = {
+export type HandleRecordTaskRunsMessage = Message & {
   type: typeof RECORD_TASK_RUNS;
   taskRuns: TaskRun[];
 };

packages/nx/src/daemon/message-types/update-workspace-context.ts

@@ -1,6 +1,8 @@
+import { Message } from '../client/daemon-socket-messenger';
+
 export const UPDATE_WORKSPACE_CONTEXT = 'UPDATE_WORKSPACE_CONTEXT' as const;
 
-export type HandleUpdateWorkspaceContextMessage = {
+export type HandleUpdateWorkspaceContextMessage = Message & {
   type: typeof UPDATE_WORKSPACE_CONTEXT;
   createdFiles: string[];
   updatedFiles: string[];

packages/nx/src/daemon/server/file-watching/file-watcher-sockets.ts

@@ -89,7 +89,7 @@ export function notifyFileWatcherSockets(
         }
 
         if (changedProjects.length > 0 || changedFiles.length > 0) {
-          return handleResult(socket, 'FILE-WATCH-CHANGED', () =>
+          return handleResult(socket, 'FILE-WATCH-CHANGED', null, () =>
             Promise.resolve({
               description: 'File watch changed',
               response: JSON.stringify({

packages/nx/src/daemon/server/handle-hash-tasks.ts

@@ -3,6 +3,7 @@ import { getCachedSerializedProjectGraphPromise } from './project-graph-incremen
 import { InProcessTaskHasher } from '../../hasher/task-hasher';
 import { readNxJson } from '../../config/configuration';
 import { DaemonProjectGraphError } from '../../project-graph/error-types';
+import { HandleHashTasksMessage } from '../message-types/hash-tasks';
 
 /**
  * We use this not to recreated hasher for every hash operation
@@ -11,12 +12,7 @@ import { DaemonProjectGraphError } from '../../project-graph/error-types';
 let storedProjectGraph: any = null;
 let storedHasher: InProcessTaskHasher | null = null;
 
-export async function handleHashTasks(payload: {
-  runnerOptions: any;
-  env: any;
-  tasks: Task[];
-  taskGraph: TaskGraph;
-}) {
+export async function handleHashTasks(payload: HandleHashTasksMessage) {
   const {
     error,
     projectGraph: _graph,
@@ -45,9 +41,14 @@ export async function handleHashTasks(payload: {
       payload.runnerOptions
     );
   }
-  const response = JSON.stringify(
-    await storedHasher.hashTasks(payload.tasks, payload.taskGraph, payload.env)
-  );
+  const response = JSON.stringify({
+    ...(await storedHasher.hashTasks(
+      payload.tasks,
+      payload.taskGraph,
+      payload.env
+    )),
+    tx: payload.tx,
+  });
   return {
     response,
     description: 'handleHashTasks',

packages/nx/src/daemon/server/handle-outputs-tracking.ts

@@ -1,10 +1,11 @@
 import { HandlerResult } from './server';
 import { outputsHashesMatch, recordOutputsHash } from './outputs-tracking';
+import { HandleRecordOutputsHashMessage } from '../message-types/record-outputs-hash';
+import { HandleOutputHashesMatchMessage } from '../message-types/output-hashes-match';
 
-export async function handleRecordOutputsHash(payload: {
-  type: string;
-  data: { outputs: string[]; hash: string };
-}): Promise<HandlerResult> {
+export async function handleRecordOutputsHash(
+  payload: HandleRecordOutputsHashMessage
+): Promise<HandlerResult> {
   try {
     await recordOutputsHash(payload.data.outputs, payload.data.hash);
     return {
@@ -21,10 +22,9 @@ export async function handleRecordOutputsHash(payload: {
   }
 }
 
-export async function handleOutputsHashesMatch(payload: {
-  type: string;
-  data: { outputs: string[]; hash: string };
-}): Promise<HandlerResult> {
+export async function handleOutputsHashesMatch(
+  payload: HandleOutputHashesMatchMessage
+): Promise<HandlerResult> {
   try {
     const res = await outputsHashesMatch(
       payload.data.outputs,

packages/nx/src/daemon/server/handle-process-in-background.ts

@@ -1,12 +1,11 @@
 import { HandlerResult } from './server';
 import { serverLogger } from './logger';
 import { getNxRequirePaths } from '../../utils/installation-directory';
+import { HandleProcessInBackgroundMessage } from '../message-types/process-in-background';
 
-export async function handleProcessInBackground(payload: {
-  type: string;
-  requirePath: string;
-  data: any;
-}): Promise<HandlerResult> {
+export async function handleProcessInBackground(
+  payload: HandleProcessInBackgroundMessage
+): Promise<HandlerResult> {
   let fn;
   try {
     fn = require(require.resolve(payload.requirePath, {

packages/nx/src/daemon/server/server.ts

@@ -4,7 +4,7 @@ import { join } from 'path';
 import { PerformanceObserver } from 'perf_hooks';
 import { hashArray } from '../../hasher/file-hasher';
 import { hashFile } from '../../native';
-import { consumeMessagesFromSocket } from '../../utils/consume-messages-from-socket';
+import { consumeMessagesFromSocket } from '../../utils/messaging';
 import { readJsonFile } from '../../utils/fileutils';
 import { PackageJson } from '../../utils/package-json';
 import { nxVersion } from '../../utils/versions';
@@ -43,6 +43,7 @@ import {
   handleServerProcessTermination,
   resetInactivityTimeout,
   respondToClient,
+  respondWithError,
   respondWithErrorAndExit,
   SERVER_INACTIVITY_TIMEOUT_MS,
   storeOutputWatcherInstance,
@@ -109,6 +110,20 @@ import {
   isHandleFlushSyncGeneratorChangesToDiskMessage,
 } from '../message-types/flush-sync-generator-changes-to-disk';
 import { handleFlushSyncGeneratorChangesToDisk } from './handle-flush-sync-generator-changes-to-disk';
+import { Message } from '../client/daemon-socket-messenger';
+import {
+  HASH_TASKS,
+  isHandleHashTasksMessage,
+} from '../message-types/hash-tasks';
+import {
+  isHandleProcessInBackgroundMessageMessage,
+  PROCESS_IN_BACKGROUND,
+} from '../message-types/process-in-background';
+import { isHandleRecordOutputsHashMessage } from '../message-types/record-outputs-hash';
+import {
+  isHandleOutputHashesMatchMessage,
+  OUTPUTS_HASHES_MATCH,
+} from '../message-types/output-hashes-match';
 
 let performanceObserver: PerformanceObserver | undefined;
 let workspaceWatcherError: Error | undefined;
@@ -185,7 +200,7 @@ async function handleMessage(socket, data: string) {
   resetInactivityTimeout(handleInactivityTimeout);
 
   const unparsedPayload = data;
-  let payload;
+  let payload: Message;
   try {
     payload = JSON.parse(unparsedPayload);
   } catch (e) {
@@ -197,91 +212,178 @@ async function handleMessage(socket, data: string) {
   }
 
   if (payload.type === 'PING') {
-    await handleResult(socket, 'PING', () =>
+    handleResult(socket, 'PING', payload.tx, () =>
       Promise.resolve({ response: JSON.stringify(true), description: 'ping' })
     );
   } else if (payload.type === 'REQUEST_PROJECT_GRAPH') {
-    await handleResult(socket, 'REQUEST_PROJECT_GRAPH', () =>
+    handleResult(socket, 'REQUEST_PROJECT_GRAPH', payload.tx, () =>
       handleRequestProjectGraph()
     );
-  } else if (payload.type === 'HASH_TASKS') {
-    await handleResult(socket, 'HASH_TASKS', () => handleHashTasks(payload));
-  } else if (payload.type === 'PROCESS_IN_BACKGROUND') {
-    await handleResult(socket, 'PROCESS_IN_BACKGROUND', () =>
-      handleProcessInBackground(payload)
+  } else if (isHandleHashTasksMessage(payload)) {
+    const promise = handleResult(socket, HASH_TASKS, payload.tx, () =>
+      handleHashTasks(payload)
     );
-  } else if (payload.type === 'RECORD_OUTPUTS_HASH') {
-    await handleResult(socket, 'RECORD_OUTPUTS_HASH', () =>
-      handleRecordOutputsHash(payload)
+    if (!payload.tx) {
+      await promise;
+    }
+  } else if (isHandleProcessInBackgroundMessageMessage(payload)) {
+    const promise = handleResult(
+      socket,
+      PROCESS_IN_BACKGROUND,
+      payload.tx,
+      () => handleProcessInBackground(payload)
     );
-  } else if (payload.type === 'OUTPUTS_HASHES_MATCH') {
-    await handleResult(socket, 'OUTPUTS_HASHES_MATCH', () =>
+    if (!payload.tx) {
+      await promise;
+    }
+  } else if (isHandleRecordOutputsHashMessage(payload)) {
+    const promise = handleResult(
+      socket,
+      'RECORD_OUTPUTS_HASH',
+      payload.tx,
+      () => handleRecordOutputsHash(payload)
+    );
+    if (!payload.tx) {
+      await promise;
+    }
+  } else if (isHandleOutputHashesMatchMessage(payload)) {
+    const promise = handleResult(socket, OUTPUTS_HASHES_MATCH, payload.tx, () =>
       handleOutputsHashesMatch(payload)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (payload.type === 'REQUEST_SHUTDOWN') {
-    await handleResult(socket, 'REQUEST_SHUTDOWN', () =>
+    const promise = handleResult(socket, 'REQUEST_SHUTDOWN', payload.tx, () =>
       handleRequestShutdown(server, numberOfOpenConnections)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (payload.type === 'REGISTER_FILE_WATCHER') {
     registeredFileWatcherSockets.push({ socket, config: payload.config });
   } else if (isHandleGlobMessage(payload)) {
-    await handleResult(socket, GLOB, () =>
+    const promise = handleResult(socket, GLOB, payload.tx, () =>
       handleGlob(payload.globs, payload.exclude)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleNxWorkspaceFilesMessage(payload)) {
-    await handleResult(socket, GET_NX_WORKSPACE_FILES, () =>
-      handleNxWorkspaceFiles(payload.projectRootMap)
+    const promise = handleResult(
+      socket,
+      GET_NX_WORKSPACE_FILES,
+      payload.tx,
+      () => handleNxWorkspaceFiles(payload.projectRootMap)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleGetFilesInDirectoryMessage(payload)) {
-    await handleResult(socket, GET_FILES_IN_DIRECTORY, () =>
-      handleGetFilesInDirectory(payload.dir)
+    const promise = handleResult(
+      socket,
+      GET_FILES_IN_DIRECTORY,
+      payload.tx,
+      () => handleGetFilesInDirectory(payload.dir)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleContextFileDataMessage(payload)) {
-    await handleResult(socket, GET_CONTEXT_FILE_DATA, () =>
-      handleContextFileData()
+    const promise = handleResult(
+      socket,
+      GET_CONTEXT_FILE_DATA,
+      payload.tx,
+      () => handleContextFileData()
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleHashGlobMessage(payload)) {
-    await handleResult(socket, HASH_GLOB, () =>
+    const promise = handleResult(socket, HASH_GLOB, payload.tx, () =>
       handleHashGlob(payload.globs, payload.exclude)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleGetFlakyTasksMessage(payload)) {
-    await handleResult(socket, GET_FLAKY_TASKS, () =>
+    const promise = handleResult(socket, GET_FLAKY_TASKS, payload.tx, () =>
       handleGetFlakyTasks(payload.hashes)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleGetEstimatedTaskTimings(payload)) {
-    await handleResult(socket, GET_ESTIMATED_TASK_TIMINGS, () =>
-      handleGetEstimatedTaskTimings(payload.targets)
+    const promise = handleResult(
+      socket,
+      GET_ESTIMATED_TASK_TIMINGS,
+      payload.tx,
+      () => handleGetEstimatedTaskTimings(payload.targets)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleWriteTaskRunsToHistoryMessage(payload)) {
-    await handleResult(socket, RECORD_TASK_RUNS, () =>
+    const promise = handleResult(socket, RECORD_TASK_RUNS, payload.tx, () =>
       handleRecordTaskRuns(payload.taskRuns)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleForceShutdownMessage(payload)) {
-    await handleResult(socket, 'FORCE_SHUTDOWN', () =>
+    const promise = handleResult(socket, 'FORCE_SHUTDOWN', payload.tx, () =>
       handleForceShutdown(server)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleGetSyncGeneratorChangesMessage(payload)) {
-    await handleResult(socket, GET_SYNC_GENERATOR_CHANGES, () =>
-      handleGetSyncGeneratorChanges(payload.generators)
+    const promise = handleResult(
+      socket,
+      GET_SYNC_GENERATOR_CHANGES,
+      payload.tx,
+      () => handleGetSyncGeneratorChanges(payload.generators)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleFlushSyncGeneratorChangesToDiskMessage(payload)) {
-    await handleResult(socket, FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK, () =>
-      handleFlushSyncGeneratorChangesToDisk(payload.generators)
+    const promise = handleResult(
+      socket,
+      FLUSH_SYNC_GENERATOR_CHANGES_TO_DISK,
+      payload.tx,
+      () => handleFlushSyncGeneratorChangesToDisk(payload.generators)
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleGetRegisteredSyncGeneratorsMessage(payload)) {
-    await handleResult(socket, GET_REGISTERED_SYNC_GENERATORS, () =>
-      handleGetRegisteredSyncGenerators()
+    const promise = handleResult(
+      socket,
+      GET_REGISTERED_SYNC_GENERATORS,
+      payload.tx,
+      () => handleGetRegisteredSyncGenerators()
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else if (isHandleUpdateWorkspaceContextMessage(payload)) {
-    await handleResult(socket, UPDATE_WORKSPACE_CONTEXT, () =>
-      handleUpdateWorkspaceContext(
-        payload.createdFiles,
-        payload.updatedFiles,
-        payload.deletedFiles
-      )
+    const promise = handleResult(
+      socket,
+      UPDATE_WORKSPACE_CONTEXT,
+      payload.tx,
+      () =>
+        handleUpdateWorkspaceContext(
+          payload.createdFiles,
+          payload.updatedFiles,
+          payload.deletedFiles
+        )
     );
+    if (!payload.tx) {
+      await promise;
+    }
   } else {
-    await respondWithErrorAndExit(
+    respondWithErrorAndExit(
       socket,
       `Invalid payload from the client`,
       new Error(`Unsupported payload sent to daemon server: ${unparsedPayload}`)
@@ -292,19 +394,20 @@ async function handleMessage(socket, data: string) {
 export async function handleResult(
   socket: Socket,
   type: string,
+  tx: string,
   hrFn: () => Promise<HandlerResult>
 ) {
   const startMark = new Date();
   const hr = await hrFn();
   const doneHandlingMark = new Date();
   if (hr.error) {
-    await respondWithErrorAndExit(socket, hr.description, hr.error);
+    await respondWithError(socket, hr.description, hr.error, tx);
   } else {
-    await respondToClient(socket, hr.response, hr.description);
+    await respondToClient(socket, hr.response, hr.description, tx);
   }
   const endMark = new Date();
   serverLogger.log(
-    `Handled ${type}. Handling time: ${
+    `Handled ${type}${tx ? ` (${tx})` : ''}. Handling time: ${
       doneHandlingMark.getTime() - startMark.getTime()
     }. Response time: ${endMark.getTime() - doneHandlingMark.getTime()}.`
   );

packages/nx/src/daemon/server/shutdown-utils.ts

@@ -92,26 +92,33 @@ export function resetInactivityTimeout(cb: () => void): void {
 export function respondToClient(
   socket: Socket,
   response: string,
-  description: string
+  description: string,
+  tx: string
 ) {
   return new Promise(async (res) => {
     if (description) {
       serverLogger.requestLog(`Responding to the client.`, description);
     }
-    socket.write(`${response}${String.fromCodePoint(4)}`, (err) => {
-      if (err) {
-        console.error(err);
+    socket.write(
+      tx
+        ? `{ "tx": "${tx}", "response": ${response}}${String.fromCodePoint(4)}`
+        : `{ "response": ${response}}${String.fromCodePoint(4)}`,
+      (err) => {
+        if (err) {
+          console.error(err);
+        }
+        serverLogger.log(`Done responding to the client`, description);
+        res(null);
       }
-      serverLogger.log(`Done responding to the client`, description);
-      res(null);
-    });
+    );
   });
 }
 
-export async function respondWithErrorAndExit(
+export async function respondWithError(
   socket: Socket,
   description: string,
-  error: Error
+  error: Error,
+  tx: string
 ) {
   const normalizedError =
     error instanceof DaemonProjectGraphError
@@ -127,5 +134,30 @@ export async function respondWithErrorAndExit(
   console.error(normalizedError.stack);
 
   // Respond with the original error
-  await respondToClient(socket, serializeResult(error, null, null), null);
+  await respondToClient(socket, serializeResult(error, null, null), null, tx);
+}
+
+export async function respondWithErrorAndExit(
+  socket: Socket,
+  description: string,
+  error: Error,
+  tx?: string
+) {
+  const normalizedError =
+    error instanceof DaemonProjectGraphError
+      ? ProjectGraphError.fromDaemonProjectGraphError(error)
+      : error;
+
+  // print some extra stuff in the error message
+  serverLogger.requestLog(
+    `Responding to the client with an error.`,
+    description,
+    normalizedError.message
+  );
+  console.error(normalizedError.stack);
+
+  // Respond with the original error
+  await respondToClient(socket, serializeResult(error, null, null), null, tx);
+
+  process.exit(1);
 }

packages/nx/src/project-graph/plugins/isolation/plugin-pool.ts

@@ -9,7 +9,11 @@ import { PluginConfiguration } from '../../../config/nx-json';
 
 import { LoadedNxPlugin } from '../internal-api';
 import { getPluginOsSocketPath } from '../../../daemon/socket-utils';
-import { consumeMessagesFromSocket } from '../../../utils/consume-messages-from-socket';
+import {
+  consumeMessagesFromSocket,
+  PendingPromise,
+  registerPendingPromise,
+} from '../../../utils/messaging';
 
 import {
   consumeMessage,
@@ -41,12 +45,6 @@ const MAX_MESSAGE_WAIT =
       2147483647
     : 1000 * 60 * MINUTES; // 10 minutes
 
-interface PendingPromise {
-  promise: Promise<unknown>;
-  resolver: (result: any) => void;
-  rejector: (err: any) => void;
-}
-
 type NxPluginWorkerCache = Map<string, Promise<LoadedNxPlugin>>;
 
 const nxPluginWorkerCache: NxPluginWorkerCache = (global[
@@ -119,6 +117,10 @@ export async function loadRemoteNxPlugin(
   return [pluginPromise, cleanupFunction];
 }
 
+const getTimeoutErrorText =
+  (context: { plugin: string; operation: string }) => (): string =>
+    `${context.plugin} timed out after ${MINUTES} minutes during ${context.operation}. ${PLUGIN_TIMEOUT_HINT_TEXT}`;
+
 /**
  * Creates a message handler for the given worker.
  * @param worker Instance of plugin-worker
@@ -169,10 +171,11 @@ function createWorkerHandler(
                           payload: { configFiles, context: ctx, tx },
                         });
                       },
-                      {
+                      getTimeoutErrorText({
                         plugin: pluginName,
                         operation: 'createNodes',
-                      }
+                      }),
+                      MAX_MESSAGE_WAIT
                     );
                   },
                 ]
@@ -190,10 +193,11 @@ function createWorkerHandler(
                         payload: { context: ctx, tx },
                       });
                     },
-                    {
+                    getTimeoutErrorText({
                       plugin: pluginName,
                       operation: 'createDependencies',
-                    }
+                    }),
+                    MAX_MESSAGE_WAIT
                   );
                 }
               : undefined,
@@ -210,10 +214,11 @@ function createWorkerHandler(
                         payload: { graph, context: ctx, tx },
                       });
                     },
-                    {
+                    getTimeoutErrorText({
                       plugin: pluginName,
                       operation: 'createMetadata',
-                    }
+                    }),
+                    MAX_MESSAGE_WAIT
                   );
                 }
               : undefined,
@@ -267,46 +272,6 @@ function createWorkerExitHandler(
   };
 }
 
-function registerPendingPromise(
-  tx: string,
-  pending: Map<string, PendingPromise>,
-  callback: () => void,
-  context: {
-    plugin: string;
-    operation: string;
-  }
-): Promise<any> {
-  let resolver: (x: unknown) => void,
-    rejector: (e: Error | unknown) => void,
-    timeout: NodeJS.Timeout;
-
-  const promise = new Promise((res, rej) => {
-    rejector = rej;
-    resolver = res;
-
-    timeout = setTimeout(() => {
-      rej(
-        new Error(
-          `${context.plugin} timed out after ${MINUTES} minutes during ${context.operation}. ${PLUGIN_TIMEOUT_HINT_TEXT}`
-        )
-      );
-    }, MAX_MESSAGE_WAIT);
-
-    callback();
-  }).finally(() => {
-    pending.delete(tx);
-    if (timeout) clearTimeout(timeout);
-  });
-
-  pending.set(tx, {
-    promise,
-    resolver,
-    rejector,
-  });
-
-  return promise;
-}
-
 global.nxPluginWorkerCount ??= 0;
 async function startPluginWorker() {
   // this should only really be true when running unit tests within

packages/nx/src/project-graph/plugins/isolation/plugin-worker.ts

@@ -1,6 +1,6 @@
 import { consumeMessage, isPluginWorkerMessage } from './messaging';
 import { createSerializableError } from '../../../utils/serializable-error';
-import { consumeMessagesFromSocket } from '../../../utils/consume-messages-from-socket';
+import { consumeMessagesFromSocket } from '../../../utils/messaging';
 
 import { createServer } from 'net';
 import { unlinkSync } from 'fs';

packages/nx/src/tasks-runner/pseudo-ipc.ts

@@ -18,7 +18,7 @@
  */
 
 import { connect, Server, Socket } from 'net';
-import { consumeMessagesFromSocket } from '../utils/consume-messages-from-socket';
+import { consumeMessagesFromSocket } from '../utils/messaging';
 import { Serializable } from 'child_process';
 
 export interface PseudoIPCMessage {

packages/nx/src/utils/consume-messages-from-socket.ts

@@ -1,19 +0,0 @@
-export function consumeMessagesFromSocket(callback: (message: string) => void) {
-  let message = '';
-  return (data) => {
-    const chunk = data.toString();
-    if (chunk.codePointAt(chunk.length - 1) === 4) {
-      message += chunk.substring(0, chunk.length - 1);
-
-      // Server may send multiple messages in one chunk, so splitting by 0x4
-      const messages = message.split('');
-      for (const splitMessage of messages) {
-        callback(splitMessage);
-      }
-
-      message = '';
-    } else {
-      message += chunk;
-    }
-  };
-}

packages/nx/src/utils/messaging.spec.ts

@@ -1,4 +1,8 @@
-import { consumeMessagesFromSocket } from './consume-messages-from-socket';
+import {
+  consumeMessagesFromSocket,
+  PendingPromise,
+  registerPendingPromise,
+} from './messaging';
 
 describe('consumeMessagesFromSocket', () => {
   it('should handle messages where every messages is in its own chunk', () => {
@@ -44,3 +48,28 @@ describe('consumeMessagesFromSocket', () => {
   //   expect(messages).toEqual([{ one: 1 }, { two: 2 }, { three: 3 }]);
   // });
 });
+
+describe('registerPendingPromise', () => {
+  it('should store a pending promise', async () => {
+    const pending = new Map<string, PendingPromise>();
+    const p = registerPendingPromise('1', pending, jest.fn(), () => 'foo', 25);
+    setTimeout(() => pending.get('1')!.resolver('bar'), 15);
+    expect(await p).toEqual('bar');
+    expect(pending.size).toBe(0);
+  });
+
+  it('should reject the promise if the callback takes too long', async () => {
+    const pending = new Map<string, PendingPromise>();
+    const p = registerPendingPromise('1', pending, jest.fn(), () => 'foo', 10);
+    await expect(p).rejects.toThrow('foo');
+    expect(pending.size).toBe(0);
+  });
+
+  it('should call the callback', async () => {
+    const pending = new Map<string, PendingPromise>();
+    const callback = jest.fn();
+    registerPendingPromise('1', pending, callback, () => 'foo', 10);
+    pending.get('1')!.resolver('bar');
+    expect(callback).toHaveBeenCalled();
+  });
+});

packages/nx/src/utils/messaging.ts

@@ -0,0 +1,59 @@
+export function consumeMessagesFromSocket(callback: (message: string) => void) {
+  let message = '';
+  return (data) => {
+    const chunk = data.toString();
+    if (chunk.codePointAt(chunk.length - 1) === 4) {
+      message += chunk.substring(0, chunk.length - 1);
+
+      // Server may send multiple messages in one chunk, so splitting by 0x4
+      const messages = message.split('');
+      for (const splitMessage of messages) {
+        callback(splitMessage);
+      }
+
+      message = '';
+    } else {
+      message += chunk;
+    }
+  };
+}
+
+export interface PendingPromise {
+  promise: Promise<unknown>;
+  resolver: (result: any) => void;
+  rejector: (err: any) => void;
+}
+
+export function registerPendingPromise(
+  tx: string,
+  pending: Map<string, PendingPromise>,
+  callback: () => void,
+  timeoutErrorText: () => string,
+  timeoutMs: number
+): Promise<any> {
+  let resolver: (x: unknown) => void,
+    rejector: (e: Error | unknown) => void,
+    timeout: NodeJS.Timeout;
+
+  const promise = new Promise((res, rej) => {
+    rejector = rej;
+    resolver = res;
+
+    timeout = setTimeout(() => {
+      rej(new Error(timeoutErrorText()));
+    }, timeoutMs);
+
+    callback();
+  }).finally(() => {
+    pending.delete(tx);
+    if (timeout) clearTimeout(timeout);
+  });
+
+  pending.set(tx, {
+    promise,
+    resolver,
+    rejector,
+  });
+
+  return promise;
+}