diff --git a/src/Consumer.js b/src/Consumer.js index 58411b5..4e51383 100644 --- a/src/Consumer.js +++ b/src/Consumer.js @@ -2,7 +2,7 @@ import stream from 'stream'; import fs from 'fs'; import path from 'path'; import { assert } from './utils/util.js'; -import { ensureDirectory } from './utils/fsUtil.js'; +import { ensureDirectory, writeFileAtomic } from './utils/fsUtil.js'; import { normalizeConsumerStateArgs } from './utils/apiHelpers.js'; import Storage from './Storage/ReadableStorage.js'; const MAX_CATCHUP_BATCH = 10; @@ -166,20 +166,13 @@ class Consumer extends stream.Readable { const consumerData = Buffer.allocUnsafe(4 + consumerState.length); consumerData.writeInt32LE(this.position, 0); consumerData.write(consumerState, 4, consumerState.length, 'utf-8'); - const tmpFile = this.fileName + '.' + this.position; this.persisting = null; - /* c8 ignore next 3 */ - if (fs.existsSync(tmpFile)) { - throw new Error(`Trying to update consumer ${this.name} concurrently. Keep each single consumer within a single process.`); - } try { - fs.writeFileSync(tmpFile, consumerData); // If the write fails (half-way), the consumer state file will not be corrupted - fs.renameSync(tmpFile, this.fileName); + writeFileAtomic(this.fileName, consumerData); this.emit('persisted', consumerState); } catch (e) { /* c8 ignore next */ - safeUnlink(tmpFile); } }); } diff --git a/src/EventStore.js b/src/EventStore.js index 429c411..2333a51 100644 --- a/src/EventStore.js +++ b/src/EventStore.js @@ -8,7 +8,7 @@ import Index from './Index.js'; import Consumer from './Consumer.js'; import { assert } from './utils/util.js'; import { ensureDirectory, resolvePath, scanForFiles } from './utils/fsUtil.js'; -import { fixCommitArgumentTypes, parseStreamFromIndexName, normalizePredicateRaw } from './utils/apiHelpers.js'; +import { fixCommitArgumentTypes, parseStreamFromIndexName, normalizePredicateRaw, createLazyPropertyHolder } from './utils/apiHelpers.js'; import { normalizeSelector, buildStreamSource } from "./utils/indexUtil.js"; import { isDcbQuery, compileDcbQuery } from "./utils/dcbUtil.js"; @@ -202,6 +202,24 @@ class EventStore extends events.EventEmitter { this.storage.open(onOpened); } + /** + * Create a lazy stream entry that opens the backing index on first access. + * + * @private + * @param {string} streamName + * @param {boolean} isClosed + * @returns {{ closed: boolean, _index: object|null, index: object }} + */ + createLazyStreamEntry(streamName, isClosed) { + return createLazyPropertyHolder( + { closed: isClosed, _index: null }, + 'index', + () => isClosed + ? this.storage.openReadonlyIndex('stream-' + streamName + '.closed') + : this.storage.openIndex('stream-' + streamName) + ); + } + /** * Check if the last commit in the store was unfinished, which is the case if not all events of the commit have been written. * Torn writes are handled at the storage level, so this method only deals with unfinished commits. @@ -252,21 +270,14 @@ class EventStore extends events.EventEmitter { } if (streamName in this.streams) { if (isClosed && !this.streams[streamName].closed) { - // The stream was renamed to .closed while this instance had it open. - // The old ReadOnlyIndex was already closed via onRename, so we open the new one. - const closedIndexName = 'stream-' + streamName + '.closed'; - const closedIndex = this.storage.openReadonlyIndex(closedIndexName); // deepcode ignore PrototypePollutionFunctionParams: streams is a Map - this.streams[streamName] = { index: closedIndex, closed: true }; + this.streams[streamName] = this.createLazyStreamEntry(streamName, true); this.emit('stream-closed', streamName); } return; } - const index = isClosed - ? this.storage.openReadonlyIndex(name) - : this.storage.openIndex(name); // deepcode ignore PrototypePollutionFunctionParams: streams is a Map - this.streams[streamName] = { index, closed: isClosed }; + this.streams[streamName] = this.createLazyStreamEntry(streamName, isClosed); this.emit('stream-available', streamName); } @@ -774,7 +785,7 @@ class EventStore extends events.EventEmitter { if (!(streamName in this.streams)) { return; } - this.streams[streamName].index.destroy(); + this.storage.deleteSecondaryIndex('stream-' + streamName); delete this.streams[streamName]; this.emit('stream-deleted', streamName); } @@ -799,6 +810,7 @@ class EventStore extends events.EventEmitter { const indexName = 'stream-' + streamName; const { index } = this.streams[streamName]; + this.storage.markStartupStateDirty(); // Flush and close the index before renaming the file index.close(); @@ -815,7 +827,9 @@ class EventStore extends events.EventEmitter { const closedIndex = this.storage.openReadonlyIndex(closedIndexName); // deepcode ignore PrototypePollutionFunctionParams: streams is a Map - this.streams[streamName] = { index: closedIndex, closed: true }; + this.streams[streamName] = this.createLazyStreamEntry(streamName, true); + this.streams[streamName]._index = closedIndex; + this.storage.persistStartupState(); this.emit('stream-closed', streamName); } diff --git a/src/FileHandlePool.js b/src/FileHandlePool.js new file mode 100644 index 0000000..19b544b --- /dev/null +++ b/src/FileHandlePool.js @@ -0,0 +1,136 @@ +/** + * A fixed-capacity registry of resources with LRU eviction of open file handles. + * + * Resources are keyed by an identifier and are expected to implement + * `open()`, `close()`, and `isOpen()`. + * + * Setting `maxOpen` to 0 disables eviction (unbounded open handles). + */ +class FileHandlePool { + + /** + * @param {number} [maxOpen=0] Maximum number of simultaneously open file + * handles. 0 disables the limit. + */ + constructor(maxOpen = 0) { + this.maxOpen = maxOpen; + this.registry = Object.create(null); + this.handles = new Map(); + } + + /** + * @param {number|string} id + * @param {object} resource + */ + add(id, resource) { + this.registry[id] = resource; + } + + /** + * @param {number|string} id + * @returns {object|undefined} + */ + get(id) { + return this.registry[id]; + } + + /** + * @param {number|string} id + * @returns {boolean} + */ + has(id) { + return id in this.registry; + } + + /** + * Remove a resource from the pool and close it if open. + * + * @param {number|string} id + */ + remove(id) { + const resource = this.registry[id]; + if (resource && resource.isOpen()) { + resource.close(); + } + delete this.registry[id]; + this.handles.delete(id); + } + + /** + * Drop only the open-handle tracking entry for a resource. + * + * @param {number|string} id + */ + forgetOpenHandle(id) { + this.handles.delete(id); + } + + /** + * Open the resource with LRU eviction if needed. + * + * @param {number|string} id + * @returns {object} + */ + open(id) { + const resource = this.registry[id]; + + if (this.maxOpen > 0) { + this.handles.delete(id); + if (this.handles.size >= this.maxOpen) { + for (const [lruId] of this.handles) { + this.handles.delete(lruId); + const lruResource = this.registry[lruId]; + if (lruResource && lruResource.isOpen()) { + lruResource.close(); + break; + } + } + } + this.handles.set(id, true); + } + + resource.open(); + return resource; + } + + /** + * @param {function(object): void} callback + */ + forEach(callback) { + for (const id of Object.keys(this.registry)) { + callback(this.registry[id]); + } + } + + /** + * @returns {Generator} + */ + *values() { + for (const id of Object.keys(this.registry)) { + yield this.registry[id]; + } + } + + /** + * @returns {number} + */ + get count() { + return Object.keys(this.registry).length; + } + + /** + * @returns {number} + */ + get openCount() { + return this.handles.size; + } + + /** + * Reset open-handle tracking without closing resources. + */ + clearOpenHandles() { + this.handles.clear(); + } +} + +export default FileHandlePool; diff --git a/src/Index/ReadableIndex.js b/src/Index/ReadableIndex.js index 4ef9c99..474e09a 100644 --- a/src/Index/ReadableIndex.js +++ b/src/Index/ReadableIndex.js @@ -55,7 +55,9 @@ class ReadableIndex extends events.EventEmitter { this.name = name; this.initialize(options); - this.open(); + if (options.autoOpen !== false) { + this.open(); + } } /** diff --git a/src/IndexPool.js b/src/IndexPool.js new file mode 100644 index 0000000..11a9b5f --- /dev/null +++ b/src/IndexPool.js @@ -0,0 +1,58 @@ +import FileHandlePool from './FileHandlePool.js'; + +/** + * LRU file-handle pool for secondary index files. + */ +class IndexPool extends FileHandlePool { + + /** + * @param {string} id + * @param {{ index: object|null, matcher?: object|function, closed?: boolean }} descriptor + */ + add(id, descriptor) { + super.add(id, descriptor); + } + + /** + * @param {string} id + * @returns {{ index: object|null, matcher?: object|function, closed?: boolean }|undefined} + */ + open(id) { + const descriptor = this.registry[id]; + if (!descriptor || !descriptor.index) { + return descriptor; + } + + if (this.maxOpen > 0) { + this.handles.delete(id); + if (this.handles.size >= this.maxOpen) { + for (const [lruId] of this.handles) { + this.handles.delete(lruId); + const lruDescriptor = this.registry[lruId]; + if (lruDescriptor?.index?.isOpen()) { + lruDescriptor.index.close(); + break; + } + } + } + this.handles.set(id, true); + } + + descriptor.index.open(); + return descriptor; + } + + /** + * @param {string} id + */ + remove(id) { + const descriptor = this.registry[id]; + if (descriptor?.index?.isOpen()) { + descriptor.index.close(); + } + delete this.registry[id]; + this.handles.delete(id); + } +} + +export default IndexPool; diff --git a/src/PartitionPool.js b/src/PartitionPool.js index 55bfc2f..452548a 100644 --- a/src/PartitionPool.js +++ b/src/PartitionPool.js @@ -1,149 +1,8 @@ +import FileHandlePool from './FileHandlePool.js'; + /** - * A fixed-capacity registry of partitions with LRU eviction of open file handles. - * - * All partitions are stored by their numeric id and may be queried at any time. - * The pool additionally tracks which partitions currently have an open file descriptor - * in LRU (least-recently-used) order. When the pool is asked to open a partition and - * doing so would exceed the configured cap, the least-recently-used open partition is - * closed first to stay within the limit. - * - * Setting the cap to 0 disables eviction: all partitions are allowed to remain open - * simultaneously, which matches the uncapped behaviour of the original implementation. + * LRU file-handle pool for partitions. */ -class PartitionPool { - - /** - * @param {number} [maxOpen=0] Maximum number of simultaneously open partition file - * handles. 0 disables the limit (no eviction). - */ - constructor(maxOpen = 0) { - this.maxOpen = maxOpen; - /** Registry of all known partitions keyed by id. */ - this.registry = Object.create(null); - /** - * Insertion-order map used for LRU tracking of open file handles. - * Key = partition id, value = true. - * Oldest (least-recently-used) entry is first; newest (most-recently-used) is last. - */ - this.handles = new Map(); - } - - /** - * Register a partition under the given id. - * - * @param {number|string} id - * @param {object} partition - */ - add(id, partition) { - this.registry[id] = partition; - } - - /** - * Retrieve a registered partition without opening it. - * - * @param {number|string} id - * @returns {object|undefined} - */ - get(id) { - return this.registry[id]; - } - - /** - * Check whether a partition with the given id is registered in the pool. - * - * @param {number|string} id - * @returns {boolean} - */ - has(id) { - return id in this.registry; - } - - /** - * Open the partition with the given id, applying LRU eviction if necessary. - * - * If the partition is not yet open and adding it would exceed `maxOpen`, the - * least-recently-used open partition is closed first. Stale entries (partitions - * that were closed externally) are discarded from the LRU map as they are - * encountered; if all tracked entries turn out to be stale the loop exits without - * closing any partition — the handle count stays temporarily inflated (bounded by - * the number of external closes since the last `open()` call) but correctness is - * preserved. - * - * @param {number|string} id - * @returns {object} The opened partition. - */ - open(id) { - const partition = this.registry[id]; - - if (this.maxOpen > 0) { - // Remove id first — this may already bring the handle count below the cap. - this.handles.delete(id); - if (this.handles.size >= this.maxOpen) { - for (const [lruId] of this.handles) { - this.handles.delete(lruId); - const lruPartition = this.registry[lruId]; - if (lruPartition && lruPartition.isOpen()) { - lruPartition.close(); - break; - } - } - } - // (Re-)add id at the MRU end of the map. - this.handles.set(id, true); - } - - partition.open(); - return partition; - } - - /** - * Invoke `callback` for every registered partition. - * - * @param {function(object): void} callback - */ - forEach(callback) { - for (const id of Object.keys(this.registry)) { - callback(this.registry[id]); - } - } - - /** - * Yield every registered partition object. - * - * @returns {Generator} - */ - *values() { - for (const id of Object.keys(this.registry)) { - yield this.registry[id]; - } - } - - /** - * The total number of registered partitions. - * @returns {number} - */ - get count() { - return Object.keys(this.registry).length; - } - - /** - * The number of open partition file handles currently tracked by the pool. - * @returns {number} - */ - get openCount() { - return this.handles.size; - } - - /** - * Reset the open-handle tracking without closing any partitions. - * - * Call this after externally closing all partitions (e.g. after - * `checkTornWrites`) to keep the pool's LRU state consistent with reality. - */ - clearOpenHandles() { - this.handles.clear(); - } - -} +class PartitionPool extends FileHandlePool {} export default PartitionPool; diff --git a/src/Storage/ReadOnlyStorage.js b/src/Storage/ReadOnlyStorage.js index cde5e5a..149d1f3 100644 --- a/src/Storage/ReadOnlyStorage.js +++ b/src/Storage/ReadOnlyStorage.js @@ -82,6 +82,7 @@ class ReadOnlyStorage extends ReadableStorage { } if (filename.endsWith('.index')) { const indexName = filename.substring(this.storageFile.length + 1, filename.length - 6); + this.registerSecondaryIndexDescriptor(indexName); // New indexes are not automatically opened in the reader this.emit('index-created', indexName); return; diff --git a/src/Storage/ReadableStorage.js b/src/Storage/ReadableStorage.js index c92b7ff..482e018 100644 --- a/src/Storage/ReadableStorage.js +++ b/src/Storage/ReadableStorage.js @@ -9,7 +9,9 @@ import { createHmac, matches, buildMetadataForMatcher } from '../utils/metadataU import { normalizeNamedCtorArgs } from '../utils/apiHelpers.js'; import IndexMatcher from '../IndexMatcher.js'; import PartitionPool from '../PartitionPool.js'; +import IndexPool from '../IndexPool.js'; import ReadablePartition from "../Partition/ReadablePartition.js"; +import StartupState from './StartupState.js'; const DEFAULT_READ_BUFFER_SIZE = 4 * 1024; @@ -27,6 +29,7 @@ const DEFAULT_MATCHER_PROPERTIES = ['stream', 'payload.type']; * Partitions beyond this limit are evicted using LRU order. 0 disables the limit. */ const DEFAULT_MAX_OPEN_PARTITIONS = 1024; +const DEFAULT_MAX_OPEN_INDEXES = 1024; /** * @typedef {object|function(object):boolean} Matcher @@ -58,6 +61,11 @@ class ReadableStorage extends events.EventEmitter { * @param {number} [config.maxOpenPartitions] Maximum number of partition file descriptors kept open at one time. * When the limit is reached the least-recently-used partition is closed to make room. 0 disables the limit. * Default: 1024. + * @param {number} [config.maxOpenIndexes] Maximum number of secondary index file descriptors kept open at one time. + * When the limit is reached the least-recently-used index is closed to make room. 0 disables the limit. + * Default: 1024. + * @param {object} [config.startupState] Startup manifest options. + * @param {boolean} [config.startupState.enabled=false] Enable startup state manifest fast-path. */ constructor(storageName = 'storage', config = {}) { super(); @@ -72,7 +80,9 @@ class ReadableStorage extends events.EventEmitter { hmacSecret: '', metadata: {}, matcherProperties: DEFAULT_MATCHER_PROPERTIES, - maxOpenPartitions: DEFAULT_MAX_OPEN_PARTITIONS + maxOpenPartitions: DEFAULT_MAX_OPEN_PARTITIONS, + maxOpenIndexes: DEFAULT_MAX_OPEN_INDEXES, + startupState: { enabled: false } }; config = Object.assign(defaults, config); this.serializer = config.serializer; @@ -84,6 +94,8 @@ class ReadableStorage extends events.EventEmitter { const partitionDefaults = { readBufferSize: DEFAULT_READ_BUFFER_SIZE }; this.partitionConfig = Object.assign(partitionDefaults, config); this.partitions = new PartitionPool(config.maxOpenPartitions); + this.secondaryIndexHandles = new IndexPool(config.maxOpenIndexes); + this.startupState = new StartupState(this.storageFile, resolvePath(config.indexDirectory || config.dataDirectory), config.startupState || {}); // initialized: null = not started (or scan cancelled), false = in progress, true = done this.initialized = null; @@ -129,8 +141,7 @@ class ReadableStorage extends events.EventEmitter { delete this.indexOptions.matcher; const { index } = this.createIndex(config.indexFile, Object.assign({}, this.indexOptions, { syncOnMissingWatchFilename: true })); this.index = index; - this.secondaryIndexes = {}; - this.readonlyIndexes = {}; + this.secondaryIndexes = this.secondaryIndexHandles.registry; /** Fast secondary-index lookup — classifies matchers for O(1) candidate resolution on write. */ this.indexMatcher = new IndexMatcher(config.matcherProperties); @@ -162,6 +173,134 @@ class ReadableStorage extends events.EventEmitter { return partitionId; } + /** + * @protected + * @param {string} name + * @returns {string} + */ + getIndexFileName(name) { + return this.storageFile + '.' + name + '.index'; + } + + /** + * @protected + * @param {string} name + * @returns {boolean} + */ + isClosedSecondaryIndex(name) { + return name.endsWith('.closed'); + } + + /** + * @protected + * @param {string} name + * @param {object} [entry] + * @returns {{ index: ReadableIndex|null, matcher?: Matcher, closed: boolean }} + */ + registerSecondaryIndexDescriptor(name, entry = {}) { + let descriptor = this.secondaryIndexes[name]; + if (!descriptor) { + descriptor = this.createSecondaryIndexDescriptor(name); + this.secondaryIndexHandles.add(name, descriptor); + } + if ('index' in entry) { + descriptor.index = entry.index; + } + if ('matcher' in entry) { + descriptor.matcher = entry.matcher; + } + if ('closed' in entry) { + descriptor.closed = entry.closed; + } + return descriptor; + } + + /** + * @private + * @param {string} name + * @returns {{ index: ReadableIndex|null, matcher?: Matcher, closed: boolean }} + */ + createSecondaryIndexDescriptor(name) { + return { + index: null, + matcher: undefined, + closed: this.isClosedSecondaryIndex(name) + }; + } + + /** + * Build a startup-state snapshot from current in-memory discovery state. + * + * @protected + * @returns {{ clean: boolean, partitions: string[], indexes: string[] }} + */ + buildStartupSnapshot() { + const indexes = Object.keys(this.secondaryIndexes) + .map(name => this.getIndexFileName(name)) + .sort(); + const partitions = Array.from(this.partitions.values()) + .map(partition => partition.name) + .sort(); + return { + clean: true, + partitions, + indexes + }; + } + + /** + * Persist startup-state snapshot if enabled. + * + * @protected + */ + persistStartupState() { + this.startupState.save(this.buildStartupSnapshot()); + } + + /** + * Mark startup state dirty before on-disk structure changes. + */ + markStartupStateDirty() { + this.startupState.markDirty(); + } + + /** + * Run a structural on-disk mutation with startup-state dirty/clean transitions. + * @protected + * @template T + * @param {function(): T} operation + * @returns {T} + */ + withStartupStateMutation(operation) { + if (!this.startupState.enabled) { + return operation(); + } + this.markStartupStateDirty(); + const result = operation(); + this.persistStartupState(); + return result; + } + + /** + * Apply a loaded startup-state snapshot into in-memory discovery state. + * + * @protected + * @param {object} state + */ + applyStartupSnapshot(state) { + for (const partitionName of state.partitions || []) { + this.registerPartitionFile(partitionName); + } + for (const indexFile of state.indexes || []) { + const prefix = this.storageFile + '.'; + if (indexFile.startsWith(prefix) && indexFile.endsWith('.index')) { + const name = indexFile.slice(prefix.length, -'.index'.length); + this.registerSecondaryIndexDescriptor(name); + this.emit('index-created', name); + } + } + } + /** * Scan partitions and secondary index files; emit 'index-created' for each found index. * @param {function} done Called when both scans finish. @@ -186,6 +325,7 @@ class ReadableStorage extends events.EventEmitter { const indexPattern = new RegExp(`^${escaped}\\.(.+)\\.index$`); scanForFiles(this.indexDirectory, indexPattern, (name) => { if (!(name in this.secondaryIndexes)) { + this.registerSecondaryIndexDescriptor(name); this.emit('index-created', name); } }, (indexErr) => { @@ -227,17 +367,51 @@ class ReadableStorage extends events.EventEmitter { return true; } this.initialized = false; + const state = this.startupState.load(); + if (state && state.clean && this.isStartupStateSafe(state)) { + this.applyStartupSnapshot(state); + this.initialized = true; + this.openIndexes(); + callback?.(); + this.emit('opened'); + return true; + } this.scanFiles(() => { // Guard: close() while scanning resets initialized to null. if (this.initialized === null) return; this.initialized = true; this.openIndexes(); + this.persistStartupState(); callback?.(); this.emit('opened'); }); return true; } + /** + * @param {object} state + * @returns {boolean} + */ + isStartupStateSafe(state) { + const indexPathPrefix = this.storageFile + '.'; + // This guards only against missing files. Corruption checks still happen when the + // corresponding partition/index is actually opened via the normal read paths. + for (const partitionName of state.partitions || []) { + if (!fs.existsSync(path.join(this.dataDirectory, partitionName))) { + return false; + } + } + for (const indexFile of state.indexes || []) { + if (!indexFile.startsWith(indexPathPrefix) || !indexFile.endsWith('.index')) { + return false; + } + if (!fs.existsSync(path.join(this.indexDirectory, indexFile))) { + return false; + } + } + return true; + } + /** * Close the storage and free up all resources. * Will emit a 'closed' event when finished. @@ -250,11 +424,11 @@ class ReadableStorage extends events.EventEmitter { this.initialized = null; } this.index.close(); - this.forEachSecondaryIndex(index => index.close()); - for (let index of Object.values(this.readonlyIndexes)) { - index.close(); + for (const descriptor of Object.values(this.secondaryIndexes)) { + descriptor.index?.close(); } this.forEachPartition(partition => partition.close()); + this.secondaryIndexHandles.clearOpenHandles(); this.emit('closed'); } @@ -392,15 +566,14 @@ class ReadableStorage extends events.EventEmitter { * @throws {Error} if the readonly index does not exist. */ openReadonlyIndex(name) { - if (name in this.readonlyIndexes) { - return this.readonlyIndexes[name]; + const descriptor = this.registerSecondaryIndexDescriptor(name, { closed: true }); + if (!descriptor.index) { + const indexName = this.getIndexFileName(name); + assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`); + descriptor.index = this.createIndex(indexName, Object.assign({}, this.indexOptions)).index; } - const indexName = this.storageFile + '.' + name + '.index'; - assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`); - const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions)); - index.open(); - this.readonlyIndexes[name] = index; - return index; + this.secondaryIndexHandles.open(name); + return descriptor.index; } /** @@ -417,20 +590,27 @@ class ReadableStorage extends events.EventEmitter { if (name === '_all') { return this.index; } - if (name in this.secondaryIndexes) { - return this.secondaryIndexes[name].index; + const descriptor = this.registerSecondaryIndexDescriptor(name, { closed: false }); + if (descriptor.index) { + this.secondaryIndexHandles.open(name); + return descriptor.index; } - const indexName = this.storageFile + '.' + name + '.index'; + const indexName = this.getIndexFileName(name); assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`); 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); + const { index, matcher: storedMatcher } = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata })); + descriptor.index = index; + descriptor.matcher = storedMatcher; + descriptor.closed = false; + + // Read-only open can omit matcher metadata; only register when a matcher is known. + if (descriptor.matcher) { + this.indexMatcher.add(name, descriptor.matcher); + } - index.open(); + this.secondaryIndexHandles.open(name); return index; } @@ -443,11 +623,42 @@ class ReadableStorage extends events.EventEmitter { removeSecondaryIndex(name) { const entry = this.secondaryIndexes[name]; if (entry) { - this.indexMatcher.remove(name); + this.secondaryIndexHandles.remove(name); + if (!entry.closed) { + this.indexMatcher.remove(name); + } delete this.secondaryIndexes[name]; } } + /** + * Remove a secondary index file and descriptor. + * + * @api + * @param {string} name + */ + deleteSecondaryIndex(name) { + const entry = this.secondaryIndexes[name]; + if (entry?.index && typeof entry.index.destroy === 'function') { + entry.index.destroy(); + } else { + const fileName = path.join(this.indexDirectory, this.getIndexFileName(name)); + if (fs.existsSync(fileName)) { + fs.unlinkSync(fileName); + } + } + this.removeSecondaryIndex(name); + } + + /** + * @private + * @param {{ index: ReadableIndex|null, matcher?: Matcher, closed: boolean }} descriptor + * @returns {boolean} + */ + hasOpenIndexDescriptor(descriptor) { + return !!descriptor && !descriptor.closed && descriptor.index !== null; + } + /** * Build the standard document result entry from a readRange yield. * @private @@ -537,13 +748,21 @@ class ReadableStorage extends events.EventEmitter { if (!matchDocument) { // No document filter: iterate all secondary indexes unconditionally. for (const indexName of Object.keys(this.secondaryIndexes)) { - iterationHandler(this.secondaryIndexes[indexName].index, indexName); + const descriptor = this.secondaryIndexes[indexName]; + if (!this.hasOpenIndexDescriptor(descriptor)) { + continue; + } + iterationHandler(descriptor.index, indexName); } return; } this.indexMatcher.forEachMatch(matchDocument, indexName => { - iterationHandler(this.secondaryIndexes[indexName].index, indexName); + const descriptor = this.secondaryIndexes[indexName]; + if (!this.hasOpenIndexDescriptor(descriptor)) { + return; + } + iterationHandler(descriptor.index, indexName); }); } diff --git a/src/Storage/StartupState.js b/src/Storage/StartupState.js new file mode 100644 index 0000000..5ecbbbb --- /dev/null +++ b/src/Storage/StartupState.js @@ -0,0 +1,98 @@ +import fs from 'fs'; +import path from 'path'; +import { hash } from '../utils/util.js'; +import { writeFileAtomic } from '../utils/fsUtil.js'; + +const STATE_VERSION = 1; + +/** + * Persistent startup manifest used to speed up storage bootstrap. + */ +class StartupState { + + /** + * @param {string} storageFile + * @param {string} directory + * @param {object} [config] + * @param {boolean} [config.enabled=false] + */ + constructor(storageFile, directory, config = {}) { + this.enabled = !!config.enabled; + this.storageFile = storageFile; + this.fileName = path.resolve(directory, `${storageFile}.startup-state.json`); + this.lastState = null; + } + + /** + * @param {object} payload + * @returns {string} + */ + checksum(payload) { + return String(hash(JSON.stringify(payload))); + } + + /** + * @param {object} raw + * @returns {boolean} + */ + isValid(raw) { + if (!raw || typeof raw !== 'object') return false; + if (raw.version !== STATE_VERSION) return false; + if (raw.storageFile !== this.storageFile) return false; + if (!raw.payload || typeof raw.payload !== 'object') return false; + return raw.checksum === this.checksum(raw.payload); + } + + /** + * @returns {object|null} + */ + load() { + if (!this.enabled || !fs.existsSync(this.fileName)) { + return null; + } + try { + const raw = JSON.parse(fs.readFileSync(this.fileName, 'utf8')); + if (!this.isValid(raw)) { + return null; + } + this.lastState = raw.payload; + return raw.payload; + } catch (e) { + return null; + } + } + + /** + * @param {object} payload + */ + save(payload) { + if (!this.enabled) { + return; + } + const envelope = { + version: STATE_VERSION, + storageFile: this.storageFile, + payload, + checksum: this.checksum(payload) + }; + writeFileAtomic(this.fileName, JSON.stringify(envelope)); + this.lastState = payload; + } + + /** + * Mark the state as dirty before mutating on-disk layout. + */ + markDirty() { + if (!this.enabled) { + return; + } + const base = this.lastState || { + clean: true, + partitions: [], + indexes: [] + }; + this.save({ ...base, clean: false }); + } +} + +export default StartupState; diff --git a/src/Storage/WritableStorage.js b/src/Storage/WritableStorage.js index cb1958c..8c5957e 100644 --- a/src/Storage/WritableStorage.js +++ b/src/Storage/WritableStorage.js @@ -102,9 +102,12 @@ class WritableStorage extends ReadableStorage { /* c8 ignore next */ if (!(index instanceof WritableIndex)) return; const wasOpen = index.isOpen(); - if (!wasOpen) index.open(); + this.secondaryIndexHandles.open(name); iterationHandler(index, name, wasOpen); - if (!wasOpen) index.close(); + if (!wasOpen) { + index.close(); + this.secondaryIndexHandles.forgetOpenHandle(name); + } }, matchDocument); } @@ -334,10 +337,12 @@ class WritableStorage extends ReadableStorage { if (this.partitions.has(partitionId)) { return; } - const partitionConfig = this.buildPartitionConfig(partitionShortName); - this.ensurePartitionDirectory(partitionName); - this.partitions.add(partitionId, this.createPartition(partitionName, partitionConfig)); - this.emit('partition-created', partitionId); + this.withStartupStateMutation(() => { + const partitionConfig = this.buildPartitionConfig(partitionShortName); + this.ensurePartitionDirectory(partitionName); + this.partitions.add(partitionId, this.createPartition(partitionName, partitionConfig)); + this.emit('partition-created', partitionId); + }); } /** @@ -382,7 +387,7 @@ class WritableStorage extends ReadableStorage { const indexEntry = this.addIndex(partition.id, position, dataSize, document); this.forEachSecondaryIndex((index, name) => { - index.open(); + this.secondaryIndexHandles.open(name); index.add(indexEntry); this.emit('index-add', name, index.length, document); }, document); @@ -405,36 +410,39 @@ class WritableStorage extends ReadableStorage { if (name === '_all') { return this.index; } - if (name in this.secondaryIndexes) { - return this.secondaryIndexes[name].index; + const existing = this.secondaryIndexes[name]; + if (existing?.index) { + return existing.index; } - const indexName = this.storageFile + '.' + name + '.index'; + const indexName = this.getIndexFileName(name); if (fs.existsSync(path.join(this.indexDirectory, indexName))) { return this.openIndex(name, matcher); } assert((typeof matcher === 'object' || typeof matcher === 'function') && matcher !== null, 'Need to specify a matcher.'); - - const metadata = buildMetadataForMatcher(matcher, this.hmac); - const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata })); - if (reindex) { - try { - this.forEachDocument((document, indexEntry) => { - if (matches(document, matcher)) { - index.add(indexEntry); - } - }); - } catch (e) { - index.destroy(); - throw e; + return this.withStartupStateMutation(() => { + const metadata = buildMetadataForMatcher(matcher, this.hmac); + const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata })); + if (reindex) { + try { + this.forEachDocument((document, indexEntry) => { + if (matches(document, matcher)) { + index.add(indexEntry); + } + }); + } catch (e) { + index.destroy(); + throw e; + } } - } - this.secondaryIndexes[name] = { index, matcher }; - this.indexMatcher.add(name, matcher); - this.emit('index-created', name); - return index; + this.registerSecondaryIndexDescriptor(name, { index, matcher, closed: false }); + this.secondaryIndexHandles.open(name); + this.indexMatcher.add(name, matcher); + this.emit('index-created', name); + return index; + }); } /** @@ -538,6 +546,13 @@ class WritableStorage extends ReadableStorage { return index; } + /** + * @inheritDoc + */ + deleteSecondaryIndex(name) { + return this.withStartupStateMutation(() => super.deleteSecondaryIndex(name)); + } + /** * @protected * @param {string} name diff --git a/src/utils/apiHelpers.js b/src/utils/apiHelpers.js index d6f18aa..7802e13 100644 --- a/src/utils/apiHelpers.js +++ b/src/utils/apiHelpers.js @@ -109,6 +109,36 @@ function normalizeConsumerStateArgs(initialState, startFrom) { return { initialState, startFrom }; } +/** + * Create an object with a lazily-resolved property. + * + * @param {object} initialValues Eager properties to copy to the object. + * @param {string} propertyName Property to resolve lazily. + * @param {function(): *} resolver Function returning the property value on first access. + * @returns {object} + */ +function createLazyPropertyHolder(initialValues, propertyName, resolver) { + const holder = { ...initialValues }; + let resolved = false; + let value = null; + Object.defineProperty(holder, propertyName, { + enumerable: true, + configurable: true, + get: () => { + if (!resolved) { + value = resolver(); + resolved = true; + } + return value; + }, + set: (newValue) => { + value = newValue; + resolved = true; + } + }); + return holder; +} + export { fixCommitArgumentTypes, parseStreamFromIndexName, @@ -116,8 +146,8 @@ export { normalizeNamedCtorArgs, normalizeRevision, normalizeMaxRevision, - normalizeConsumerStateArgs + normalizeConsumerStateArgs, + createLazyPropertyHolder }; - diff --git a/src/utils/fsUtil.js b/src/utils/fsUtil.js index deac472..6e20e1a 100644 --- a/src/utils/fsUtil.js +++ b/src/utils/fsUtil.js @@ -1,6 +1,7 @@ import fs from 'fs'; import os from "os"; import path from 'path'; +import crypto from 'crypto'; import { mkdirpSync } from 'mkdirp'; /** @@ -133,6 +134,28 @@ function scanForFiles(directory, regexPattern, onEach, onDone) { scanDir(directory, '', true, regexPattern, onEach, onDone); } +/** + * Atomically replace a file by writing to a unique temp file in the same + * directory and then renaming it over the target path. + * + * @param {string} fileName Target file path. + * @param {string|Buffer} data File contents. + */ +function writeFileAtomic(fileName, data) { + const tmpFile = `${fileName}.${process.pid}.${Date.now()}.${crypto.randomBytes(4).toString('hex')}.tmp`; + try { + fs.writeFileSync(tmpFile, data); + fs.renameSync(tmpFile, fileName); + } catch (e) { + /* c8 ignore next 7 */ + try { + fs.unlinkSync(tmpFile); + } catch { + } + throw e; + } +} + /** * Return true when both paths are equal or `child` is nested inside `parent`. * @@ -150,5 +173,6 @@ export { resolvePath, ensureDirectory, scanForFiles, + writeFileAtomic, isSameOrParentDirectory }; diff --git a/test/EventStore.spec.js b/test/EventStore.spec.js index f0127d9..b610957 100644 --- a/test/EventStore.spec.js +++ b/test/EventStore.spec.js @@ -304,6 +304,52 @@ describe('EventStore', function() { }); }); }); + + it('registers streams lazily on startup without eagerly opening stream indexes', function(done) { + eventstore = new EventStore({ storageDirectory }); + eventstore.on('ready', () => { + eventstore.commit('stream-a', [{ foo: 1 }], () => { + eventstore.commit('stream-b', [{ foo: 2 }], () => { + eventstore.close(); + + const originalOpenIndex = EventStoreBase.Storage.prototype.openIndex; + let openIndexCalls = 0; + try { + EventStoreBase.Storage.prototype.openIndex = function(...args) { + openIndexCalls++; + return originalOpenIndex.apply(this, args); + }; + + let reopened; + try { + reopened = new EventStore({ storageDirectory }); + } catch (error) { + EventStoreBase.Storage.prototype.openIndex = originalOpenIndex; + done(error); + return; + } + reopened.on('ready', () => { + let assertionError = null; + try { + expect(openIndexCalls).to.be(0); + expect(reopened.getStreamVersion('stream-a')).to.be(1); + expect(openIndexCalls).to.be(1); + } catch (error) { + assertionError = error; + } finally { + reopened.close(); + EventStoreBase.Storage.prototype.openIndex = originalOpenIndex; + done(assertionError); + } + }); + } catch (error) { + EventStoreBase.Storage.prototype.openIndex = originalOpenIndex; + done(error); + } + }); + }); + }); + }); }); describe('commit', function() { diff --git a/test/Storage.spec.js b/test/Storage.spec.js index 6daee2e..f89cf2a 100644 --- a/test/Storage.spec.js +++ b/test/Storage.spec.js @@ -1390,7 +1390,7 @@ describe('Storage', function() { reader.open(); reader.once('index-created', (name) => { expect(name).to.be('one'); - expect(reader.secondaryIndexes[name]).to.be(undefined); + expect(reader.secondaryIndexes[name]).to.eql({ index: null, matcher: undefined, closed: false }); reader.close(); done(); }); @@ -1407,7 +1407,7 @@ describe('Storage', function() { reader.open(); reader.once('index-created', (name) => { expect(name).to.be('one'); - expect(reader.secondaryIndexes[name]).to.be(undefined); + expect(reader.secondaryIndexes[name]).to.eql({ index: null, matcher: undefined, closed: false }); reader.close(); done(); }); @@ -1817,6 +1817,79 @@ describe('Storage', function() { expect(openCount).to.be(5); }); + describe('maxOpenIndexes (LRU index pool)', function() { + + it('closes the LRU secondary index when the limit is reached', function() { + storage = createStorage({ maxOpenIndexes: 2 }); + storage.open(); + + storage.ensureIndex('one', { type: 'one' }); + storage.ensureIndex('two', { type: 'two' }); + storage.ensureIndex('three', { type: 'three' }); + + const openCount = Object.values(storage.secondaryIndexes) + .filter(({ index }) => index.isOpen()).length; + expect(openCount).to.be(2); + }); + + it('setting maxOpenIndexes to 0 disables the limit', function() { + storage = createStorage({ maxOpenIndexes: 0 }); + storage.open(); + + storage.ensureIndex('one', { type: 'one' }); + storage.ensureIndex('two', { type: 'two' }); + storage.ensureIndex('three', { type: 'three' }); + + const openCount = Object.values(storage.secondaryIndexes) + .filter(({ index }) => index.isOpen()).length; + expect(openCount).to.be(3); + }); + }); + + describe('startupState', function() { + + it('uses a clean manifest fast-path without calling scanFiles on open', function() { + const startupState = { enabled: true }; + + storage = createStorage({ startupState }); + storage.open(); + storage.write({ type: 'one', foo: 1 }); + storage.ensureIndex('one', { type: 'one' }); + storage.close(); + + storage = createStorage({ startupState }); + storage.scanFiles = () => { + throw new Error('scanFiles should not be called for clean startup-state fast-path'); + }; + + expect(() => storage.open()).to.not.throwError(); + expect(storage.length).to.be(1); + }); + + it('falls back to scan when manifest is dirty', function(done) { + const startupState = { enabled: true }; + + storage = createStorage({ startupState }); + storage.open(); + storage.write({ foo: 1 }); + storage.markStartupStateDirty(); + storage.close(); + + storage = createStorage({ startupState }); + const originalScanFiles = storage.scanFiles.bind(storage); + let scanCalled = false; + storage.scanFiles = (callback) => { + scanCalled = true; + originalScanFiles(callback); + }; + storage.once('opened', () => { + expect(scanCalled).to.be(true); + done(); + }); + storage.open(); + }); + }); + }); });