Add startup state snapshots for fast storage initialization with lazy filesystem backfill

src/EventStore.js

@@ -3,7 +3,7 @@ import JoinEventStream from './JoinEventStream.js';
 import fs from 'fs';
 import path from 'path';
 import events from 'events';
-import Storage, { ReadOnly as ReadOnlyStorage, LOCK_THROW, LOCK_RECLAIM } from './Storage.js';
+import Storage, { ReadOnly as ReadOnlyStorage, LOCK_THROW, LOCK_RECLAIM, IndexNotFoundError } from './Storage.js';
 import Index from './Index.js';
 import Consumer from './Consumer.js';
 import { assert } from './utils/util.js';
@@ -174,6 +174,9 @@ class EventStore extends events.EventEmitter {
         this.streamsDirectory = resolvePath(storageConfig.indexDirectory);
         this.storeName = storeName;
         this.consumers = new Map();
+        // Read-only stores never enter commit(), so no stream hydration pass is required.
+        // Writable stores defer hydration until the first write path.
+        this.knownStreamsHydrated = storageConfig.readOnly === true;
 
         const storage = storageConfig.readOnly === true
             ? new ReadOnlyStorage(storeName, storageConfig)
@@ -262,9 +265,17 @@ class EventStore extends events.EventEmitter {
             }
             return;
         }
-        const index = isClosed
-            ? this.storage.openReadonlyIndex(name)
-            : this.storage.openIndex(name);
+        let index;
+        try {
+            index = isClosed
+                ? this.storage.openReadonlyIndex(name)
+                : this.storage.openIndex(name);
+        } catch (error) {
+            if (error instanceof IndexNotFoundError) {
+                return;
+            }
+            throw error;
+        }
         // deepcode ignore PrototypePollutionFunctionParams: streams is a Map
         this.streams[streamName] = { index, closed: isClosed };
         this.emit('stream-available', streamName);
@@ -482,6 +493,21 @@ class EventStore extends events.EventEmitter {
         }
     }
 
+    /**
+     * Delay hydrating known stream indexes until a write path is entered to keep startup fast,
+     * while still guaranteeing existing stream versions/matchers are loaded before commit logic runs.
+     * @private
+     */
+    ensureKnownStreamsHydratedForWrite() {
+        if (this.knownStreamsHydrated) {
+            return;
+        }
+        this.knownStreamsHydrated = true;
+        for (const indexName of this.storage.knownIndexes) {
+            this.registerStream(indexName);
+        }
+    }
+
     /**
      * Commit a list of events for the given stream name, which is expected to be at the given version.
      * Note that the events committed may still appear in other streams too - the given stream name is only
@@ -501,6 +527,7 @@ class EventStore extends events.EventEmitter {
         assert(!(this.storage instanceof ReadOnlyStorage), 'The storage was opened in read-only mode. Can not commit to it.');
         assert(typeof streamName === 'string' && streamName !== '', 'Must specify a stream name for commit.');
         assert(typeof events !== 'undefined' && events !== null, 'No events specified for commit.');
+        this.ensureKnownStreamsHydratedForWrite();
 
         ({ events, expectedVersion, metadata, callback } = fixCommitArgumentTypes(
             events,

src/Storage.js

@@ -1,10 +1,11 @@
 import WritableStorage, { StorageLockedError, LOCK_THROW, LOCK_RECLAIM } from './Storage/WritableStorage.js';
 import ReadOnlyStorage from './Storage/ReadOnlyStorage.js';
+import { IndexNotFoundError } from './Storage/ReadableStorage.js';
 
 WritableStorage.ReadOnly = ReadOnlyStorage;
 WritableStorage.StorageLockedError = StorageLockedError;
 WritableStorage.LOCK_THROW = LOCK_THROW;
 WritableStorage.LOCK_RECLAIM = LOCK_RECLAIM;
 
 export default WritableStorage;
-export { ReadOnlyStorage as ReadOnly, StorageLockedError, LOCK_THROW, LOCK_RECLAIM };
+export { ReadOnlyStorage as ReadOnly, StorageLockedError, LOCK_THROW, LOCK_RECLAIM, IndexNotFoundError };

src/Storage/ReadableStorage.js

@@ -12,6 +12,15 @@ import PartitionPool from '../PartitionPool.js';
 import ReadablePartition from "../Partition/ReadablePartition.js";
 
 const DEFAULT_READ_BUFFER_SIZE = 4 * 1024;
+const STARTUP_STATE_FILE_SUFFIX = '.startup-state.json';
+
+class IndexNotFoundError extends Error {
+    constructor(indexName) {
+        super(`Index "${indexName}" does not exist.`);
+        this.name = 'IndexNotFoundError';
+        this.code = 'INDEX_NOT_FOUND';
+    }
+}
 
 /**
  * Default ordered list of document property paths used as discriminant keys when
@@ -122,6 +131,7 @@ class ReadableStorage extends events.EventEmitter {
      */
     initializeIndexes(config) {
         this.indexDirectory = resolvePath(config.indexDirectory || this.dataDirectory);
+        this.startupStateFile = path.join(this.indexDirectory, '.' + this.storageFile + STARTUP_STATE_FILE_SUFFIX);
 
         this.indexOptions = config.indexOptions;
         this.indexOptions.dataDirectory = this.indexDirectory;
@@ -131,6 +141,7 @@ class ReadableStorage extends events.EventEmitter {
         this.index = index;
         this.secondaryIndexes = {};
         this.readonlyIndexes = {};
+        this.knownIndexes = new Set();
 
         /** Fast secondary-index lookup — classifies matchers for O(1) candidate resolution on write. */
         this.indexMatcher = new IndexMatcher(config.matcherProperties);
@@ -158,10 +169,96 @@ class ReadableStorage extends events.EventEmitter {
             const partition = this.createPartition(filename, this.partitionConfig);
             this.partitions.add(partition.id, partition);
             this.emit('partition-created', partition.id);
+            this.onKnownStateChanged();
         }
         return partitionId;
     }
 
+    /**
+     * Track a known secondary index name and optionally emit `index-created` for first discovery.
+     *
+     * @protected
+     * @param {string} name
+     * @param {boolean} [emitEvent=true]
+     * @returns {boolean} True when the name was newly tracked.
+     */
+    registerKnownIndex(name, emitEvent = true) {
+        if (this.knownIndexes.has(name)) {
+            return false;
+        }
+        this.knownIndexes.add(name);
+        if (emitEvent) {
+            this.emit('index-created', name);
+        }
+        this.onKnownStateChanged();
+        return true;
+    }
+
+    /**
+     * @private
+     * @param {string} name
+     */
+    throwIndexNotFoundError(name) {
+        throw new IndexNotFoundError(name);
+    }
+
+    /**
+     * Build a startup-state snapshot from currently known partitions and indexes.
+     *
+     * @protected
+     * @returns {{version: number, partitions: string[], indexes: string[]}}
+     */
+    buildStartupStateSnapshot() {
+        const partitions = [];
+        this.forEachPartition(partition => {
+            partitions.push(partition.name);
+        });
+        return {
+            version: 1,
+            partitions: partitions.sort(),
+            indexes: Array.from(this.knownIndexes).sort()
+        };
+    }
+
+    /**
+     * Load known partition/index names from the persisted startup-state snapshot.
+     * Missing files are ignored and discovered later by the background scan.
+     *
+     * @protected
+     * @returns {boolean} True when a valid snapshot file was consumed.
+     */
+    loadStartupStateSnapshot() {
+        if (!fs.existsSync(this.startupStateFile)) {
+            return false;
+        }
+        let snapshot;
+        try {
+            snapshot = JSON.parse(fs.readFileSync(this.startupStateFile, 'utf8'));
+        } catch (e) {
+            return false;
+        }
+        if (!snapshot || snapshot.version !== 1) {
+            return false;
+        }
+
+        const partitions = Array.isArray(snapshot.partitions) ? snapshot.partitions : [];
+        for (const partitionName of partitions) {
+            if (typeof partitionName !== 'string' || partitionName === '') {
+                continue;
+            }
+            this.registerPartitionFile(partitionName);
+        }
+
+        const indexes = Array.isArray(snapshot.indexes) ? snapshot.indexes : [];
+        for (const indexName of indexes) {
+            if (typeof indexName !== 'string' || indexName === '' || indexName === '_all') {
+                continue;
+            }
+            this.registerKnownIndex(indexName);
+        }
+        return true;
+    }
+
     /**
      * Scan partitions and secondary index files; emit 'index-created' for each found index.
      * @param {function} done Called when both scans finish.
@@ -170,7 +267,12 @@ class ReadableStorage extends events.EventEmitter {
         const escaped = this.storageFile.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
         const partitionPattern = new RegExp(`^(${escaped}.*)$`);
         scanForFiles(this.dataDirectory, partitionPattern, (file) => {
-            if (file.endsWith('.index') || file.endsWith('.branch') || file.endsWith('.lock')) return;
+            if (this.initialized === null) {
+                return;
+            }
+            if (file.endsWith('.index') || file.endsWith('.branch') || file.endsWith('.lock')) {
+                return;
+            }
             this.registerPartitionFile(file);
         }, (partErr) => {
             /* c8 ignore next */
@@ -185,9 +287,10 @@ class ReadableStorage extends events.EventEmitter {
             }
             const indexPattern = new RegExp(`^${escaped}\\.(.+)\\.index$`);
             scanForFiles(this.indexDirectory, indexPattern, (name) => {
-                if (!(name in this.secondaryIndexes)) {
-                    this.emit('index-created', name);
+                if (this.initialized === null) {
+                    return;
                 }
+                this.registerKnownIndex(name);
             }, (indexErr) => {
                 // The directory could disappear between existsSync and readdir (e.g. test cleanup).
                 /* c8 ignore next */
@@ -227,13 +330,26 @@ class ReadableStorage extends events.EventEmitter {
             return true;
         }
         this.initialized = false;
-        this.scanFiles(() => {
-            // Guard: close() while scanning resets initialized to null.
+        const finishOpen = () => {
             if (this.initialized === null) return;
             this.initialized = true;
             this.openIndexes();
             callback?.();
             this.emit('opened');
+        };
+        if (this.loadStartupStateSnapshot()) {
+            this.scanFiles(() => {
+                // Guard: close() while scanning resets initialized to null.
+                if (this.initialized === null) return;
+                this.onKnownStateChanged();
+            });
+            setImmediate(finishOpen);
+            return true;
+        }
+        this.scanFiles(() => {
+            // Guard: close() while scanning resets initialized to null.
+            finishOpen();
+            this.onKnownStateChanged();
         });
         return true;
     }
@@ -396,10 +512,13 @@ class ReadableStorage extends events.EventEmitter {
             return this.readonlyIndexes[name];
         }
         const indexName = this.storageFile + '.' + name + '.index';
-        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
+        if (!fs.existsSync(path.join(this.indexDirectory, indexName))) {
+            this.throwIndexNotFoundError(name);
+        }
         const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions));
         index.open();
         this.readonlyIndexes[name] = index;
+        this.registerKnownIndex(name, false);
         return index;
     }
 
@@ -422,13 +541,16 @@ class ReadableStorage extends events.EventEmitter {
         }
 
         const indexName = this.storageFile + '.' + name + '.index';
-        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
+        if (!fs.existsSync(path.join(this.indexDirectory, indexName))) {
+            this.throwIndexNotFoundError(name);
+        }
 
         const metadata = buildMetadataForMatcher(matcher, this.hmac);
         let { index } = this.secondaryIndexes[name] = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata }));
 
         // Register the actual stored matcher (may have been reconstructed from metadata by WritableStorage.createIndex).
         this.indexMatcher.add(name, this.secondaryIndexes[name].matcher);
+        this.registerKnownIndex(name, false);
 
         index.open();
         return index;
@@ -562,7 +684,15 @@ class ReadableStorage extends events.EventEmitter {
         this.partitions.forEach(iterationHandler);
     }
 
+    /**
+     * Hook called when the set of known partitions/indexes changes.
+     * WritableStorage overrides this to persist startup-state snapshots.
+     *
+     * @protected
+     */
+    onKnownStateChanged() {}
+
 }
 
 export default ReadableStorage;
-export { matches };
+export { matches, IndexNotFoundError };