feat(core): delegate db operations to daemon when enabled

packages/nx/src/command-line/report/report.ts

@@ -402,12 +402,14 @@ export async function getReportData(): Promise<ReportData> {
     }
   }
 
-  let cache = dbCacheEnabled()
-    ? {
-        max: resolveMaxCacheSize(nxJson),
-        used: new DbCache({ nxCloudRemoteCache: null }).getUsedCacheSpace(),
-      }
-    : null;
+  let cache = null;
+  if (dbCacheEnabled()) {
+    const cacheInstance = new DbCache({ nxCloudRemoteCache: null });
+    cache = {
+      max: resolveMaxCacheSize(nxJson),
+      used: await cacheInstance.getUsedCacheSpace(),
+    };
+  }
 
   return {
     pm,

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

@@ -33,7 +33,13 @@ import {
   DaemonProjectGraphError,
   ProjectGraphError,
 } from '../../project-graph/error-types';
-import { IS_WASM, NxWorkspaceFiles, TaskRun, TaskTarget } from '../../native';
+import {
+  HashedTask,
+  IS_WASM,
+  NxWorkspaceFiles,
+  TaskRun,
+  TaskTarget,
+} from '../../native';
 import {
   HandleGlobMessage,
   HandleMultiGlobMessage,
@@ -64,6 +70,33 @@ import {
   HandleRecordTaskRunsMessage,
   RECORD_TASK_RUNS,
 } from '../message-types/task-history';
+import {
+  HandleRecordTaskDetailsMessage,
+  RECORD_TASK_DETAILS,
+} from '../message-types/task-details';
+import {
+  HandleGetRunningTasksMessage,
+  HandleAddRunningTaskMessage,
+  HandleRemoveRunningTaskMessage,
+  GET_RUNNING_TASKS,
+  ADD_RUNNING_TASK,
+  REMOVE_RUNNING_TASK,
+} from '../message-types/running-tasks';
+import {
+  CACHE_GET,
+  CACHE_PUT,
+  CACHE_REMOVE_OLD_RECORDS,
+  CACHE_APPLY_REMOTE_RESULTS,
+  CACHE_GET_SIZE,
+  CACHE_CHECK_FS_IN_SYNC,
+  type HandleCacheGetMessage,
+  type HandleCachePutMessage,
+  type HandleCacheRemoveOldRecordsMessage,
+  type HandleCacheApplyRemoteResultsMessage,
+  type HandleCacheGetSizeMessage,
+  type HandleCacheCheckFsInSyncMessage,
+} from '../message-types/cache';
+import type { CachedResult as NativeCacheResult } from '../../native';
 import { FORCE_SHUTDOWN } from '../message-types/force-shutdown';
 import {
   GET_SYNC_GENERATOR_CHANGES,
@@ -511,6 +544,97 @@ export class DaemonClient {
     return this.sendToDaemonViaQueue(message);
   }
 
+  recordTaskDetails(taskDetails: HashedTask[]): Promise<void> {
+    const message: HandleRecordTaskDetailsMessage = {
+      type: RECORD_TASK_DETAILS,
+      taskDetails,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  getRunningTasks(ids: string[]): Promise<string[]> {
+    const message: HandleGetRunningTasksMessage = {
+      type: GET_RUNNING_TASKS,
+      ids,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  addRunningTask(taskId: string): Promise<void> {
+    const message: HandleAddRunningTaskMessage = {
+      type: ADD_RUNNING_TASK,
+      taskId,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  removeRunningTask(taskId: string): Promise<void> {
+    const message: HandleRemoveRunningTaskMessage = {
+      type: REMOVE_RUNNING_TASK,
+      taskId,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  cacheGet(hash: string): Promise<NativeCacheResult | null> {
+    const message: HandleCacheGetMessage = {
+      type: CACHE_GET,
+      hash,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  cachePut(
+    hash: string,
+    terminalOutput: string | null,
+    outputs: string[],
+    code: number
+  ): Promise<void> {
+    const message: HandleCachePutMessage = {
+      type: CACHE_PUT,
+      hash,
+      terminalOutput,
+      outputs,
+      code,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  cacheRemoveOldRecords(): Promise<void> {
+    const message: HandleCacheRemoveOldRecordsMessage = {
+      type: CACHE_REMOVE_OLD_RECORDS,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  cacheApplyRemoteResults(
+    hash: string,
+    result: NativeCacheResult,
+    outputs: string[]
+  ): Promise<void> {
+    const message: HandleCacheApplyRemoteResultsMessage = {
+      type: CACHE_APPLY_REMOTE_RESULTS,
+      hash,
+      result,
+      outputs,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  cacheGetSize(): Promise<number> {
+    const message: HandleCacheGetSizeMessage = {
+      type: CACHE_GET_SIZE,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
+  cacheCheckFsInSync(): Promise<boolean> {
+    const message: HandleCacheCheckFsInSyncMessage = {
+      type: CACHE_CHECK_FS_IN_SYNC,
+    };
+    return this.sendToDaemonViaQueue(message);
+  }
+
   getSyncGeneratorChanges(
     generators: string[]
   ): Promise<SyncGeneratorRunResult[]> {
@@ -776,7 +900,7 @@ export class DaemonClient {
         'result-parse-start-' + this.currentMessage.type,
         'result-parse-end-' + this.currentMessage.type
       );
-      if (parsedResult.error) {
+      if (parsedResult?.error) {
         if (
           'message' in parsedResult.error &&
           (parsedResult.error.message === 'NX_VERSION_CHANGED' ||

packages/nx/src/daemon/is-on-daemon.ts

@@ -1,3 +1,10 @@
 export function isOnDaemon() {
   return !!global.NX_DAEMON;
 }
+
+/**
+ * Feature flag for gradual rollout of delegating database operations to the daemon.
+ */
+export function shouldDelegateDbToDaemon(): boolean {
+  return process.env.NX_DELEGATE_DB_TO_DAEMON === 'true';
+}

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

@@ -0,0 +1,106 @@
+import type { CachedResult as NativeCacheResult } from '../../native';
+
+export const CACHE_GET = 'CACHE_GET' as const;
+export const CACHE_PUT = 'CACHE_PUT' as const;
+export const CACHE_REMOVE_OLD_RECORDS = 'CACHE_REMOVE_OLD_RECORDS' as const;
+export const CACHE_APPLY_REMOTE_RESULTS = 'CACHE_APPLY_REMOTE_RESULTS' as const;
+export const CACHE_GET_SIZE = 'CACHE_GET_SIZE' as const;
+export const CACHE_CHECK_FS_IN_SYNC = 'CACHE_CHECK_FS_IN_SYNC' as const;
+
+export type HandleCacheGetMessage = {
+  type: typeof CACHE_GET;
+  hash: string;
+};
+
+export type HandleCachePutMessage = {
+  type: typeof CACHE_PUT;
+  hash: string;
+  terminalOutput: string | null;
+  outputs: string[];
+  code: number;
+};
+
+export type HandleCacheRemoveOldRecordsMessage = {
+  type: typeof CACHE_REMOVE_OLD_RECORDS;
+};
+
+export type HandleCacheApplyRemoteResultsMessage = {
+  type: typeof CACHE_APPLY_REMOTE_RESULTS;
+  hash: string;
+  result: NativeCacheResult;
+  outputs: string[];
+};
+
+export type HandleCacheGetSizeMessage = {
+  type: typeof CACHE_GET_SIZE;
+};
+
+export type HandleCacheCheckFsInSyncMessage = {
+  type: typeof CACHE_CHECK_FS_IN_SYNC;
+};
+
+export function isHandleCacheGetMessage(
+  message: unknown
+): message is HandleCacheGetMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === CACHE_GET
+  );
+}
+
+export function isHandleCachePutMessage(
+  message: unknown
+): message is HandleCachePutMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === CACHE_PUT
+  );
+}
+
+export function isHandleCacheRemoveOldRecordsMessage(
+  message: unknown
+): message is HandleCacheRemoveOldRecordsMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === CACHE_REMOVE_OLD_RECORDS
+  );
+}
+
+export function isHandleCacheApplyRemoteResultsMessage(
+  message: unknown
+): message is HandleCacheApplyRemoteResultsMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === CACHE_APPLY_REMOTE_RESULTS
+  );
+}
+
+export function isHandleCacheGetSizeMessage(
+  message: unknown
+): message is HandleCacheGetSizeMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === CACHE_GET_SIZE
+  );
+}
+
+export function isHandleCacheCheckFsInSyncMessage(
+  message: unknown
+): message is HandleCacheCheckFsInSyncMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === CACHE_CHECK_FS_IN_SYNC
+  );
+}

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

@@ -0,0 +1,51 @@
+export const GET_RUNNING_TASKS = 'GET_RUNNING_TASKS' as const;
+export const ADD_RUNNING_TASK = 'ADD_RUNNING_TASK' as const;
+export const REMOVE_RUNNING_TASK = 'REMOVE_RUNNING_TASK' as const;
+
+export type HandleGetRunningTasksMessage = {
+  type: typeof GET_RUNNING_TASKS;
+  ids: string[];
+};
+
+export type HandleAddRunningTaskMessage = {
+  type: typeof ADD_RUNNING_TASK;
+  taskId: string;
+};
+
+export type HandleRemoveRunningTaskMessage = {
+  type: typeof REMOVE_RUNNING_TASK;
+  taskId: string;
+};
+
+export function isHandleGetRunningTasksMessage(
+  message: unknown
+): message is HandleGetRunningTasksMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === GET_RUNNING_TASKS
+  );
+}
+
+export function isHandleAddRunningTaskMessage(
+  message: unknown
+): message is HandleAddRunningTaskMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === ADD_RUNNING_TASK
+  );
+}
+
+export function isHandleRemoveRunningTaskMessage(
+  message: unknown
+): message is HandleRemoveRunningTaskMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === REMOVE_RUNNING_TASK
+  );
+}

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

@@ -0,0 +1,19 @@
+import type { HashedTask } from '../../native';
+
+export const RECORD_TASK_DETAILS = 'RECORD_TASK_DETAILS' as const;
+
+export type HandleRecordTaskDetailsMessage = {
+  type: typeof RECORD_TASK_DETAILS;
+  taskDetails: HashedTask[];
+};
+
+export function isHandleRecordTaskDetailsMessage(
+  message: unknown
+): message is HandleRecordTaskDetailsMessage {
+  return (
+    typeof message === 'object' &&
+    message !== null &&
+    'type' in message &&
+    message['type'] === RECORD_TASK_DETAILS
+  );
+}