feat(core): add fallback cache for db cache

packages/nx/src/native/cache/cache.rs

@@ -4,12 +4,12 @@ use std::time::Instant;
 
 use fs_extra::remove_items;
 use napi::bindgen_prelude::*;
+use regex::Regex;
 use rusqlite::{params, Connection, OptionalExtension};
 use tracing::trace;
 
 use crate::native::cache::expand_outputs::_expand_outputs;
 use crate::native::cache::file_ops::_copy;
-use crate::native::machine_id::get_machine_id;
 use crate::native::utils::Normalize;
 
 #[napi(object)]
@@ -36,8 +36,7 @@ impl NxCache {
         cache_path: String,
         db_connection: External<Connection>,
     ) -> anyhow::Result<Self> {
-        let machine_id = get_machine_id();
-        let cache_path = PathBuf::from(&cache_path).join(machine_id);
+        let cache_path = PathBuf::from(&cache_path);
 
         create_dir_all(&cache_path)?;
         create_dir_all(cache_path.join("terminalOutputs"))?;
@@ -143,7 +142,11 @@ impl NxCache {
     }
 
     #[napi]
-    pub fn apply_remote_cache_results(&self, hash: String, result: CachedResult) -> anyhow::Result<()> {
+    pub fn apply_remote_cache_results(
+        &self,
+        hash: String,
+        result: CachedResult,
+    ) -> anyhow::Result<()> {
         let terminal_output = result.terminal_output;
         write(self.get_task_outputs_path(hash.clone()), terminal_output)?;
 
@@ -153,14 +156,13 @@ impl NxCache {
     }
 
     fn get_task_outputs_path_internal(&self, hash: &str) -> PathBuf {
-        self.cache_path
-            .join("terminalOutputs")
-            .join(hash)
+        self.cache_path.join("terminalOutputs").join(hash)
     }
 
     #[napi]
     pub fn get_task_outputs_path(&self, hash: String) -> String {
-        self.get_task_outputs_path_internal(&hash).to_normalized_string()
+        self.get_task_outputs_path_internal(&hash)
+            .to_normalized_string()
     }
 
     fn record_to_cache(&self, hash: String, code: i16) -> anyhow::Result<()> {
@@ -192,11 +194,12 @@ impl NxCache {
                 .as_slice(),
         )?;
 
-        trace!("Copying Files from Cache {:?} -> {:?}", &outputs_path, &self.workspace_root);
-        _copy(
-            outputs_path,
-            &self.workspace_root,
-        )?;
+        trace!(
+            "Copying Files from Cache {:?} -> {:?}",
+            &outputs_path,
+            &self.workspace_root
+        );
+        _copy(outputs_path, &self.workspace_root)?;
 
         Ok(())
     }
@@ -224,4 +227,37 @@ impl NxCache {
 
         Ok(())
     }
+
+    #[napi]
+    pub fn check_cache_fs_in_sync(&self) -> anyhow::Result<bool> {
+        // Checks that the number of cache records in the database
+        // matches the number of cache directories on the filesystem.
+        // If they don't match, it means that the cache is out of sync.
+        let cache_records = self
+            .db
+            .query_row("SELECT COUNT(*) FROM cache_outputs", [], |row| {
+                let count: i64 = row.get(0)?;
+                Ok(count)
+            })?;
+        let hash_regex = Regex::new(r"^\d+$").expect("Hash regex is invalid");
+        let fs_entries = std::fs::read_dir(&self.cache_path)
+            .map_err(anyhow::Error::from)?
+            // Cache entries are directories, that name is a hash (numerical string)
+            .filter(|entry| {
+                entry
+                    .as_ref()
+                    .map(|entry| {
+                        entry.file_type().map(|ft| ft.is_dir()).unwrap_or(false)
+                            && entry
+                                .file_name()
+                                .to_str()
+                                .map(|name| hash_regex.is_match(name))
+                                .unwrap_or(false)
+                    })
+                    .unwrap_or(false)
+            })
+            .count() as i64;
+
+        Ok(cache_records == fs_entries)
+    }
 }

packages/nx/src/native/db/mod.rs

@@ -10,10 +10,13 @@ use crate::native::machine_id::get_machine_id;
 pub fn connect_to_nx_db(
     cache_dir: String,
     nx_version: String,
+    db_name: Option<String>,
 ) -> anyhow::Result<External<Connection>> {
-    let machine_id = get_machine_id();
     let cache_dir_buf = PathBuf::from(cache_dir);
-    let db_path = cache_dir_buf.join(format!("{}.db", machine_id));
+    let db_path = cache_dir_buf.join(format!(
+        "{}.db",
+        db_name.unwrap_or_else(get_machine_id)
+    ));
     create_dir_all(cache_dir_buf)?;
 
     let c = create_connection(&db_path)?;

packages/nx/src/native/index.d.ts

@@ -35,6 +35,7 @@ export declare class NxCache {
   getTaskOutputsPath(hash: string): string
   copyFilesFromCache(cachedResult: CachedResult, outputs: Array<string>): void
   removeOldCacheRecords(): void
+  checkCacheFsInSync(): boolean
 }
 
 export declare class NxTaskHistory {
@@ -96,7 +97,7 @@ export interface CachedResult {
   outputsPath: string
 }
 
-export declare export function connectToNxDb(cacheDir: string, nxVersion: string): ExternalObject<Connection>
+export declare export function connectToNxDb(cacheDir: string, nxVersion: string, dbName?: string | undefined | null): ExternalObject<Connection>
 
 export declare export function copy(src: string, dest: string): void
 

packages/nx/src/native/tests/cache.spec.ts

@@ -16,7 +16,9 @@ describe('Cache', () => {
       force: true,
     });
 
-    const dbConnection = getDbConnection(join(__dirname, 'temp-db'));
+    const dbConnection = getDbConnection({
+      directory: join(__dirname, 'temp-db'),
+    });
 
     taskDetails = new TaskDetails(dbConnection);
 

packages/nx/src/native/tests/task_history.spec.ts

@@ -17,7 +17,9 @@ describe('NxTaskHistory', () => {
       force: true,
     });
 
-    const dbConnection = getDbConnection(join(__dirname, 'temp-db'));
+    const dbConnection = getDbConnection({
+      directory: join(__dirname, 'temp-db'),
+    });
     taskHistory = new NxTaskHistory(dbConnection);
     taskDetails = new TaskDetails(dbConnection);
 

packages/nx/src/tasks-runner/cache.ts

@@ -8,7 +8,11 @@ import {
   RemoteCacheV2,
 } from './default-tasks-runner';
 import { spawn } from 'child_process';
-import { cacheDir } from '../utils/cache-directory';
+import {
+  cacheDir,
+  defaultCacheDir,
+  workspaceDataDirectory,
+} from '../utils/cache-directory';
 import { Task } from '../config/task-graph';
 import { machineId } from 'node-machine-id';
 import { NxCache, CachedResult as NativeCacheResult } from '../native';
@@ -39,6 +43,8 @@ export function getCache(options: DefaultTasksRunnerOptions) {
 
 export class DbCache {
   private cache = new NxCache(workspaceRoot, cacheDir, getDbConnection());
+  private fallbackDbCache: NxCache | null = null;
+
   private remoteCache: RemoteCacheV2 | null;
   private remoteCachePromise: Promise<RemoteCacheV2>;
 
@@ -46,7 +52,28 @@ export class DbCache {
     this.remoteCache = await this.getRemoteCache();
   }
 
-  constructor(private readonly options: { nxCloudRemoteCache: RemoteCache }) {}
+  constructor(private readonly options: { nxCloudRemoteCache: RemoteCache }) {
+    // User has customized the cache directory - this could be because they
+    // are using a shared cache in the custom directory. The db cache is not
+    // stored in the cache directory, and is keyed by machine ID so they would
+    // hit issues. If we detect this, we can create a fallback db cache in the
+    // custom directory, and check if the entries are there when the main db
+    // cache misses.
+    if (cacheDir !== defaultCacheDir && !this.cache.checkCacheFsInSync()) {
+      if (/* { TODO: INSERT HAS POWERPACK CHECK HERE } */ true) {
+        this.fallbackDbCache = new NxCache(
+          workspaceRoot,
+          cacheDir,
+          getDbConnection({
+            dbName: 'shared',
+            directory: cacheDir,
+          })
+        );
+      } else {
+        throw new Error(/* TODO: Add nx.dev link explaining cache + powerpack */);
+      }
+    }
+  }
 
   async get(task: Task): Promise<CachedResult | null> {
     const res = this.cache.get(task.hash);
@@ -56,6 +83,14 @@ export class DbCache {
         ...res,
         remote: false,
       };
+    } else if (this.fallbackDbCache) {
+      const sharedRes = this.fallbackDbCache.get(task.hash);
+      if (sharedRes) {
+        return {
+          ...sharedRes,
+          remote: false,
+        };
+      }
     }
     await this.setup();
     if (this.remoteCache) {
@@ -94,6 +129,10 @@ export class DbCache {
     return tryAndRetry(async () => {
       this.cache.put(task.hash, terminalOutput, outputs, code);
 
+      if (this.fallbackDbCache) {
+        this.fallbackDbCache.put(task.hash, terminalOutput, outputs, code);
+      }
+
       await this.setup();
       if (this.remoteCache) {
         await this.remoteCache.store(

packages/nx/src/utils/cache-directory.ts

@@ -71,6 +71,7 @@ export const cacheDir = cacheDirectory(
   workspaceRoot,
   readCacheDirectoryProperty(workspaceRoot)
 );
+export const defaultCacheDir = defaultCacheDirectory(workspaceRoot);
 
 export function cacheDirectoryForWorkspace(workspaceRoot: string) {
   return cacheDirectory(

packages/nx/src/utils/db-connection.ts

@@ -2,9 +2,32 @@ import { connectToNxDb, ExternalObject } from '../native';
 import { workspaceDataDirectory } from './cache-directory';
 import { version as NX_VERSION } from '../../package.json';
 
-let dbConnection: ExternalObject<any>;
+const dbConnectionMap = new Map<string, ExternalObject<any>>();
 
-export function getDbConnection(directory = workspaceDataDirectory) {
-  dbConnection ??= connectToNxDb(directory, NX_VERSION);
-  return dbConnection;
+export function getDbConnection(
+  opts: {
+    directory?: string;
+    dbName?: string;
+  } = {}
+) {
+  opts.directory ??= workspaceDataDirectory;
+  const key = `${opts.directory}:${opts.dbName ?? 'default'}`;
+  const connection = getEntryOrSet(dbConnectionMap, key, () =>
+    connectToNxDb(opts.directory, NX_VERSION, opts.dbName)
+  );
+  return connection;
+}
+
+function getEntryOrSet<TKey, TVal>(
+  map: Map<TKey, TVal>,
+  key: TKey,
+  defaultValue: () => TVal
+) {
+  const existing = map.get(key);
+  if (existing) {
+    return existing;
+  }
+  const val = defaultValue();
+  map.set(key, val);
+  return val;
 }