Add mutex for windows

Author: satyakigh

.husky/pre-commit

@@ -1,3 +1,3 @@
 git secrets --register-aws || (echo 'Please install git-secrets https://github.com/awslabs/git-secrets to check for accidentally commited secrets!' && exit 1)
 git secrets --pre_commit_hook -- ""
-
+npm run lint

package-lock.json

@@ -72,6 +72,7 @@
     "@opentelemetry/sdk-trace-node": "2.0.1",
     "@tree-sitter-grammars/tree-sitter-yaml": "0.7.1",
     "archiver": "7.0.1",
+    "async-mutex": "0.5.0",
     "axios": "1.11.0",
     "cfn-guard": "https://gitpkg.now.sh/aws-cloudformation/cloudformation-guard/guard/ts-lib?33d9931",
     "deep-object-diff": "1.1.9",

package.json

@@ -1,26 +1,25 @@
 import { readdirSync, rmSync } from 'fs';
 import { join } from 'path';
+import { Mutex } from 'async-mutex';
 import { open, Database, RootDatabase } from 'lmdb';
 import { Logger } from 'pino';
 import { LoggerFactory } from '../telemetry/LoggerFactory';
 import { ScopedTelemetry } from '../telemetry/ScopedTelemetry';
 import { Telemetry } from '../telemetry/TelemetryDecorator';
 import { TelemetryService } from '../telemetry/TelemetryService';
 import { pathToArtifact } from '../utils/ArtifactsDir';
+import { isWindows } from '../utils/Environment';
 import { toString } from '../utils/String';
 import { DataStore, DataStoreFactory, StoreName } from './DataStore';
 import { encryptionStrategy } from './lmdb/Utils';
 
 export class LMDBStore implements DataStore {
-    private readonly log: Logger;
-    private readonly telemetry: ScopedTelemetry;
-
     constructor(
         public readonly name: StoreName,
-        private readonly store: Database<unknown, string>,
+        protected readonly store: Database<unknown, string>,
+        private readonly telemetry: ScopedTelemetry = TelemetryService.instance.get(`LMDB.${name}`),
+        private readonly log: Logger = LoggerFactory.getLogger(`LMDB.${name}`),
     ) {
-        this.telemetry = TelemetryService.instance.get(`LMDB.${name}`);
-        this.log = LoggerFactory.getLogger(`LMDB.${name}`);
         this.log.info('Initialized');
     }
 
@@ -29,21 +28,21 @@ export class LMDBStore implements DataStore {
         return this.store.get(key) as T | undefined;
     }
 
-    put<T>(key: string, value: T): Promise<boolean> {
+    async put<T>(key: string, value: T): Promise<boolean> {
         this.log.info(`Put ${key}`);
-        return this.telemetry.measureAsync('put', () => {
+        return await this.telemetry.measureAsync('put', () => {
             return this.store.put(key, value);
         });
     }
 
-    remove(key: string): Promise<boolean> {
+    async remove(key: string): Promise<boolean> {
         this.log.info(`Remove ${key}`);
-        return this.store.remove(key);
+        return await this.store.remove(key);
     }
 
-    clear(): Promise<void> {
+    async clear(): Promise<void> {
         this.log.info(`Clear ${this.name}`);
-        return this.store.clearAsync();
+        return await this.store.clearAsync();
     }
 
     keys(limit: number): ReadonlyArray<string> {
@@ -56,6 +55,52 @@ export class LMDBStore implements DataStore {
     }
 }
 
+export class LockedLMDBStore extends LMDBStore {
+    private readonly lock = new Mutex();
+
+    constructor(name: StoreName, store: Database<unknown, string>) {
+        super(
+            name,
+            store,
+            TelemetryService.instance.get(`LockedLMDBStore.${name}`),
+            LoggerFactory.getLogger(`LockedLMDBStore.${name}`),
+        );
+    }
+
+    override get<T>(key: string): T | undefined {
+        if (this.lock.isLocked()) {
+            return undefined;
+        }
+
+        return super.get(key);
+    }
+
+    override async put<T>(key: string, value: T): Promise<boolean> {
+        return await this.lock.runExclusive(async () => {
+            return await super.put(key, value);
+        });
+    }
+
+    override async remove(key: string): Promise<boolean> {
+        return await this.lock.runExclusive(async () => {
+            return await super.remove(key);
+        });
+    }
+
+    override async clear(): Promise<void> {
+        return await this.lock.runExclusive(async () => {
+            return await super.clear();
+        });
+    }
+
+    override keys(limit: number): ReadonlyArray<string> {
+        if (this.lock.isLocked()) {
+            return [];
+        }
+        return super.keys(limit);
+    }
+}
+
 export class LMDBStoreFactory implements DataStoreFactory {
     private readonly log: Logger;
     @Telemetry({ scope: 'LMDB.Global' }) private readonly telemetry!: ScopedTelemetry;
@@ -90,7 +135,14 @@ export class LMDBStoreFactory implements DataStoreFactory {
                 name: store,
                 encoding: Encoding,
             });
-            this.stores.set(store, new LMDBStore(store, database));
+
+            let lmdbStore: LMDBStore;
+            if (isWindows) {
+                lmdbStore = new LockedLMDBStore(store, database);
+            } else {
+                lmdbStore = new LMDBStore(store, database);
+            }
+            this.stores.set(store, lmdbStore);
         }
         this.registerLMDBGauges();
 
@@ -177,7 +229,7 @@ export class LMDBStoreFactory implements DataStoreFactory {
 
 const Version = 4;
 const VersionFileName = `.v${Version}.lmdb`;
-const Encoding: 'msgpack' | 'json' | 'string' | 'binary' | 'ordered-binary' = 'msgpack';
+const Encoding = 'msgpack';
 const TotalMaxDbSize = 5 * 250 * 1024 * 1024; // 250MB max size
 
 function stats(store: RootDatabase | Database): StoreStatsType {

src/datastore/LMDB.ts

@@ -36,13 +36,21 @@ export class MemoryStore implements DataStore {
     keys(limit: number): ReadonlyArray<string> {
         return [...this.store.keys()].slice(0, limit);
     }
+
+    size() {
+        return this.store.size;
+    }
 }
 
 export class MemoryStoreFactory implements DataStoreFactory {
     @Telemetry({ scope: 'MemoryStore.Global' }) private readonly telemetry!: ScopedTelemetry;
 
     private readonly stores = new Map<StoreName, MemoryStore>();
 
+    constructor() {
+        this.registerMemoryStoreGauges();
+    }
+
     get(store: StoreName): DataStore {
         let val = this.stores.get(store);
         if (val === undefined) {
@@ -62,13 +70,10 @@ export class MemoryStoreFactory implements DataStoreFactory {
     }
 
     private registerMemoryStoreGauges(): void {
-        this.telemetry.registerGaugeProvider('stores.count', () => this.stores.size);
-        this.telemetry.registerGaugeProvider('global.entries', () => {
-            let total = 0;
-            for (const store of this.stores.values()) {
-                total += store.keys(1000).length;
-            }
-            return total;
-        });
+        this.telemetry.registerGaugeProvider('env.entries', () => this.stores.size);
+
+        for (const [name, store] of this.stores.entries()) {
+            this.telemetry.registerGaugeProvider(`store.${name}.entries`, () => store.size());
+        }
     }
 }

src/datastore/MemoryStore.ts

@@ -78,6 +78,7 @@ export class SchemaRetriever implements SettingsConfigurable, Closeable {
     initialize() {
         this.getRegionalSchemasIfStale();
         this.getSamSchemasIfMissingOrStale();
+        this.schemaTaskManager.runPrivateTask();
     }
 
     getDefault(): CombinedSchemas {