Replace in-process mutex with cross-process file locking for FileStore

package.json

@@ -88,8 +88,8 @@
     "partial-json": "0.1.7",
     "pino": "9.9.0",
     "pino-pretty": "13.1.1",
+    "proper-lockfile": "4.1.2",
     "pyodide": "0.28.2",
-    "stack-utils": "2.0.6",
     "tree-sitter": "0.22.4",
     "tree-sitter-json": "0.24.8",
     "ts-essentials": "10.1.1",
@@ -106,8 +106,8 @@
     "@types/archiver": "7.0.0",
     "@types/js-yaml": "4.0.9",
     "@types/luxon": "3.7.1",
+    "@types/proper-lockfile": "4.1.4",
     "@types/semver": "7.7.1",
-    "@types/stack-utils": "2.0.3",
     "@types/yargs": "17.0.33",
     "@types/yauzl": "2.10.3",
     "@vitest/coverage-v8": "3.2.4",

src/datastore/DataStore.ts

@@ -16,6 +16,8 @@ export enum StoreName {
     private_schemas = 'private_schemas',
 }
 
+export const PersistedStores = [StoreName.public_schemas, StoreName.sam_schemas];
+
 export interface DataStore {
     get<T>(key: string): T | undefined;
 

src/datastore/FileStore.ts

@@ -5,7 +5,7 @@ import { LoggerFactory } from '../telemetry/LoggerFactory';
 import { ScopedTelemetry } from '../telemetry/ScopedTelemetry';
 import { Telemetry } from '../telemetry/TelemetryDecorator';
 import { formatNumber } from '../utils/String';
-import { DataStore, DataStoreFactory, StoreName } from './DataStore';
+import { DataStore, DataStoreFactory, PersistedStores, StoreName } from './DataStore';
 import { EncryptedFileStore } from './file/EncryptedFileStore';
 import { encryptionKey } from './file/Encryption';
 
@@ -20,7 +20,7 @@ export class FileStoreFactory implements DataStoreFactory {
     private readonly metricsInterval: NodeJS.Timeout;
     private readonly timeout: NodeJS.Timeout;
 
-    constructor(rootDir: string) {
+    constructor(rootDir: string, storeNames: StoreName[] = PersistedStores) {
         this.log = LoggerFactory.getLogger('FileStore.Global');
 
         this.fileDbRoot = join(rootDir, 'filedb');
@@ -30,6 +30,10 @@ export class FileStoreFactory implements DataStoreFactory {
             mkdirSync(this.fileDbDir, { recursive: true });
         }
 
+        for (const store of storeNames) {
+            this.stores.set(store, new EncryptedFileStore(encryptionKey(VersionNumber), store, this.fileDbDir));
+        }
+
         this.metricsInterval = setInterval(() => {
             this.emitMetrics();
         }, 60 * 1000);
@@ -45,10 +49,9 @@ export class FileStoreFactory implements DataStoreFactory {
     }
 
     get(store: StoreName): DataStore {
-        let val = this.stores.get(store);
-        if (!val) {
-            val = new EncryptedFileStore(encryptionKey(VersionNumber), store, this.fileDbDir);
-            this.stores.set(store, val);
+        const val = this.stores.get(store);
+        if (val === undefined) {
+            throw new Error(`Store ${store} not found. Available stores: ${[...this.stores.keys()].join(', ')}`);
         }
         return val;
     }

src/datastore/LMDB.ts

@@ -6,7 +6,7 @@ import { ScopedTelemetry } from '../telemetry/ScopedTelemetry';
 import { Telemetry } from '../telemetry/TelemetryDecorator';
 import { isWindows } from '../utils/Environment';
 import { formatNumber, toString } from '../utils/String';
-import { DataStore, DataStoreFactory, StoreName } from './DataStore';
+import { DataStore, DataStoreFactory, PersistedStores, StoreName } from './DataStore';
 import { LMDBStore } from './lmdb/LMDBStore';
 import { stats } from './lmdb/Stats';
 import { encryptionStrategy } from './lmdb/Utils';
@@ -22,7 +22,7 @@ export class LMDBStoreFactory implements DataStoreFactory {
 
     private readonly stores = new Map<StoreName, LMDBStore>();
 
-    constructor(rootDir: string, storeNames: StoreName[] = [StoreName.public_schemas, StoreName.sam_schemas]) {
+    constructor(rootDir: string, storeNames: StoreName[] = PersistedStores) {
         this.lmdbDir = join(rootDir, 'lmdb');
 
         const config: RootDatabaseOptionsWithPath = {

src/datastore/file/EncryptedFileStore.ts

@@ -1,122 +1,116 @@
-import { existsSync, readFileSync, statSync, unlinkSync } from 'fs'; // eslint-disable-line no-restricted-imports -- files being checked
+import { existsSync, readFileSync, statSync, writeFileSync } from 'fs'; // eslint-disable-line no-restricted-imports -- files being checked
 import { writeFile } from 'fs/promises';
 import { join } from 'path';
-import { Mutex } from 'async-mutex';
 import { Logger } from 'pino';
+import { lock, LockOptions, lockSync } from 'proper-lockfile';
 import { LoggerFactory } from '../../telemetry/LoggerFactory';
 import { ScopedTelemetry } from '../../telemetry/ScopedTelemetry';
 import { TelemetryService } from '../../telemetry/TelemetryService';
 import { DataStore } from '../DataStore';
 import { decrypt, encrypt } from './Encryption';
 
+const LOCK_OPTIONS_SYNC: LockOptions = { stale: 10_000 };
+const LOCK_OPTIONS: LockOptions = { ...LOCK_OPTIONS_SYNC, retries: { retries: 15, minTimeout: 10, maxTimeout: 500 } };
+
 export class EncryptedFileStore implements DataStore {
     private readonly log: Logger;
-
     private readonly file: string;
-    private content?: Record<string, unknown>;
+    private content: Record<string, unknown> = {};
     private readonly telemetry: ScopedTelemetry;
-    private readonly lock = new Mutex();
 
     constructor(
         private readonly KEY: Buffer,
-        private readonly name: string,
+        name: string,
         fileDbDir: string,
     ) {
         this.log = LoggerFactory.getLogger(`FileStore.${name}`);
         this.file = join(fileDbDir, `${name}.enc`);
-
         this.telemetry = TelemetryService.instance.get(`FileStore.${name}`);
-    }
-
-    get<T>(key: string): T | undefined {
-        return this.telemetry.countExecution('get', () => {
-            if (this.content) {
-                return this.content[key] as T | undefined;
-            }
 
-            if (!existsSync(this.file)) {
-                return;
-            }
-
-            if (this.lock.isLocked()) {
-                return this.content?.[key];
+        if (existsSync(this.file)) {
+            try {
+                this.content = this.readFile();
+            } catch (error) {
+                this.log.error(error, 'Failed to decrypt file store, recreating store');
+                this.telemetry.count('filestore.recreate', 1);
+
+                const release = lockSync(this.file, LOCK_OPTIONS_SYNC);
+                try {
+                    this.saveSync();
+                } finally {
+                    release();
+                }
             }
+        } else {
+            this.saveSync();
+        }
+    }
 
-            const decrypted = decrypt(this.KEY, readFileSync(this.file));
-            this.content = JSON.parse(decrypted) as Record<string, unknown>;
-            return this.content[key] as T | undefined;
-        });
+    get<T>(key: string): T | undefined {
+        return this.telemetry.countExecution('get', () => this.content[key] as T | undefined);
     }
 
     put<T>(key: string, value: T): Promise<boolean> {
-        return this.lock.runExclusive(() =>
-            this.telemetry.measureAsync('put', async () => {
-                if (!this.content) {
-                    this.get(key);
-                }
-
-                this.content = {
-                    ...this.content,
-                    [key]: value,
-                };
-                const encrypted = encrypt(this.KEY, JSON.stringify(this.content));
-                await writeFile(this.file, encrypted);
-                return true;
-            }),
-        );
+        return this.withLock('put', async () => {
+            this.content[key] = value;
+            await this.save();
+            return true;
+        });
     }
 
     remove(key: string): Promise<boolean> {
-        return this.lock.runExclusive(() => {
-            return this.telemetry.measureAsync('remove', async () => {
-                if (!this.content) {
-                    this.get(key);
-                }
-
-                if (!this.content || !(key in this.content)) {
-                    return false;
-                }
+        return this.withLock('remove', async () => {
+            if (!(key in this.content)) {
+                return false;
+            }
 
-                delete this.content[key];
-                const encrypted = encrypt(this.KEY, JSON.stringify(this.content));
-                await writeFile(this.file, encrypted);
-                return true;
-            });
+            delete this.content[key];
+            await this.save();
+            return true;
         });
     }
 
     clear(): Promise<void> {
-        return this.lock.runExclusive(() => {
-            return this.telemetry.countExecutionAsync('clear', () => {
-                if (existsSync(this.file)) {
-                    unlinkSync(this.file);
-                }
-                this.content = undefined;
-                return Promise.resolve();
-            });
+        return this.withLock('clear', async () => {
+            this.content = {};
+            await this.save();
         });
     }
 
     keys(limit: number): ReadonlyArray<string> {
-        return this.telemetry.countExecution('keys', () => {
-            if (!this.content) {
-                this.get('ANY_KEY');
-            }
-
-            return Object.keys(this.content ?? {}).slice(0, limit);
-        });
+        return this.telemetry.countExecution('keys', () => Object.keys(this.content).slice(0, limit));
     }
 
     stats(): FileStoreStats {
-        if (!this.content) {
-            this.get('ANY_KEY');
-        }
-
         return {
-            entries: Object.keys(this.content ?? {}).length,
+            entries: Object.keys(this.content).length,
             totalSize: existsSync(this.file) ? statSync(this.file).size : 0,
         };
     }
+
+    private async withLock<T>(operation: string, fn: () => Promise<T>): Promise<T> {
+        return await this.telemetry.measureAsync(operation, async () => {
+            const release = await lock(this.file, LOCK_OPTIONS);
+            try {
+                this.content = this.readFile();
+                return await fn();
+            } finally {
+                await release();
+            }
+        });
+    }
+
+    private readFile(): Record<string, unknown> {
+        return JSON.parse(decrypt(this.KEY, readFileSync(this.file))) as Record<string, unknown>;
+    }
+
+    private saveSync() {
+        writeFileSync(this.file, encrypt(this.KEY, JSON.stringify(this.content)));
+    }
+
+    private async save() {
+        await writeFile(this.file, encrypt(this.KEY, JSON.stringify(this.content)));
+    }
 }
 
 export type FileStoreStats = {