diff --git a/src/EventStore.js b/src/EventStore.js index 429c411..37cf59e 100644 --- a/src/EventStore.js +++ b/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, diff --git a/src/Storage.js b/src/Storage.js index 5bec197..68125f1 100644 --- a/src/Storage.js +++ b/src/Storage.js @@ -1,5 +1,6 @@ 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; @@ -7,4 +8,4 @@ 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 }; diff --git a/src/Storage/ReadableStorage.js b/src/Storage/ReadableStorage.js index c92b7ff..d0e00a8 100644 --- a/src/Storage/ReadableStorage.js +++ b/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 }; diff --git a/src/Storage/WritableStorage.js b/src/Storage/WritableStorage.js index cb1958c..23267a1 100644 --- a/src/Storage/WritableStorage.js +++ b/src/Storage/WritableStorage.js @@ -9,6 +9,7 @@ import { matches, buildMetadataForMatcher, buildMatcherFromMetadata } from '../u import { normalizeNamedCtorArgs } from '../utils/apiHelpers.js'; const DEFAULT_WRITE_BUFFER_SIZE = 16 * 1024; +const STARTUP_STATE_PERSIST_DELAY_MS = 10; const LOCK_RECLAIM = 0x1; const LOCK_THROW = 0x2; @@ -63,6 +64,7 @@ class WritableStorage extends ReadableStorage { this._lockMode = config.lock; this.partitioner = config.partitioner; this.partitionIds = {}; + this.startupStatePersistTimer = null; } /** @@ -82,9 +84,13 @@ class WritableStorage extends ReadableStorage { return true; } + const finishOpen = () => { + callback?.(); + this.scheduleStartupStatePersist(); + }; const onOpen = needsRepair - ? () => { this.checkTornWrites(); callback?.(); } - : callback; + ? () => { this.checkTornWrites(); finishOpen(); } + : finishOpen; return super.open(onOpen); } @@ -259,6 +265,11 @@ class WritableStorage extends ReadableStorage { * Unlocks the storage, then delegates to the parent close(). */ close() { + this.persistStartupStateSnapshot(); + if (this.startupStatePersistTimer) { + clearTimeout(this.startupStatePersistTimer); + this.startupStatePersistTimer = null; + } if (this.locked) { this.unlock(); } @@ -338,6 +349,7 @@ class WritableStorage extends ReadableStorage { this.ensurePartitionDirectory(partitionName); this.partitions.add(partitionId, this.createPartition(partitionName, partitionConfig)); this.emit('partition-created', partitionId); + this.onKnownStateChanged(); } /** @@ -433,7 +445,7 @@ class WritableStorage extends ReadableStorage { this.secondaryIndexes[name] = { index, matcher }; this.indexMatcher.add(name, matcher); - this.emit('index-created', name); + this.registerKnownIndex(name); return index; } @@ -572,6 +584,43 @@ class WritableStorage extends ReadableStorage { return new WritablePartition(name, config); } + /** + * Persist the known partitions/indexes snapshot asynchronously, debounced. + * + * @protected + */ + onKnownStateChanged() { + this.scheduleStartupStatePersist(); + } + + /** + * Schedule snapshot persistence for startup-state optimization. + * @private + */ + scheduleStartupStatePersist() { + if (this.startupStatePersistTimer || this.initialized !== true) { + return; + } + this.startupStatePersistTimer = setTimeout(() => { + this.startupStatePersistTimer = null; + this.persistStartupStateSnapshot(); + }, STARTUP_STATE_PERSIST_DELAY_MS); + } + + /** + * Persist currently known partitions/indexes. + * Best-effort optimization: failures are intentionally ignored. + * + * @private + */ + persistStartupStateSnapshot() { + try { + fs.writeFileSync(this.startupStateFile, JSON.stringify(this.buildStartupStateSnapshot())); + } catch (e) { + // Startup snapshots are a best-effort performance optimization. + } + } + } export default WritableStorage; diff --git a/test/EventStore.spec.js b/test/EventStore.spec.js index f0127d9..70b559b 100644 --- a/test/EventStore.spec.js +++ b/test/EventStore.spec.js @@ -394,6 +394,32 @@ describe('EventStore', function() { }); }); + it('hydrates known streams only once on first commit', function(done) { + eventstore = new EventStore({ + storageDirectory + }); + + eventstore.on('ready', () => { + eventstore.storage.registerKnownIndex('stream-missing', false); + + let knownStreamOpenAttempts = 0; + const originalOpenIndex = eventstore.storage.openIndex.bind(eventstore.storage); + eventstore.storage.openIndex = (name, matcher) => { + if (name === 'stream-missing') { + knownStreamOpenAttempts++; + } + return originalOpenIndex(name, matcher); + }; + + eventstore.commit('foo', [{ value: 1 }], () => { + eventstore.commit('foo', [{ value: 2 }], () => { + expect(knownStreamOpenAttempts).to.be(1); + done(); + }); + }); + }); + }); + it('invokes callback when finished with optimistic concurrency check', function(done) { eventstore = new EventStore({ storageDirectory diff --git a/test/Storage.spec.js b/test/Storage.spec.js index 6daee2e..d198386 100644 --- a/test/Storage.spec.js +++ b/test/Storage.spec.js @@ -1527,6 +1527,50 @@ describe('Storage', function() { storage.getPartition(''); }); + + it('loads startup state snapshot and opens before the lazy scan finishes', function(done) { + storage = createStorage({ partitioner: (document) => document.type }); + storage.open(); + storage.write({ foo: 1, type: 'one' }); + storage.write({ foo: 2, type: 'two' }); + storage.ensureIndex('one', doc => doc.type === 'one'); + storage.flush(); + storage.close(); + + const startupStateFile = path.join(dataDirectory, '.storage.startup-state.json'); + expect(fs.existsSync(startupStateFile)).to.be(true); + + const reader = createReader(); + let scanCalled = false; + reader.scanFiles = () => { + scanCalled = true; + }; + reader.once('opened', () => { + expect(scanCalled).to.be(true); + expect(reader.read(1)).to.eql({ foo: 1, type: 'one' }); + expect(reader.read(2)).to.eql({ foo: 2, type: 'two' }); + reader.close(); + done(); + }); + reader.open(); + }); + + it('does not persist startup state in read-only mode', function() { + storage = createStorage(); + storage.open(); + storage.write({ foo: 1 }); + storage.flush(); + storage.close(); + + const startupStateFile = path.join(dataDirectory, '.storage.startup-state.json'); + fs.removeSync(startupStateFile); + + const reader = createReader(); + reader.open(); + reader.close(); + + expect(fs.existsSync(startupStateFile)).to.be(false); + }); }); describe('preCommit', function() {