From 2da0b9a830d7f257ec701b78594e9b3cec035139 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 18 Jul 2026 16:10:05 +0000 Subject: [PATCH 1/2] feat: speed up startup with lazy stream indexes and startup state --- src/EventStore.js | 56 ++++++++++--- src/FileHandlePool.js | 136 ++++++++++++++++++++++++++++++ src/Index/ReadableIndex.js | 4 +- src/IndexPool.js | 8 ++ src/PartitionPool.js | 149 +-------------------------------- src/Storage/ReadOnlyStorage.js | 4 + src/Storage/ReadableStorage.js | 134 ++++++++++++++++++++++++++++- src/Storage/StartupState.js | 101 ++++++++++++++++++++++ src/Storage/WritableStorage.js | 22 ++++- test/EventStore.spec.js | 46 ++++++++++ test/Storage.spec.js | 74 ++++++++++++++++ 11 files changed, 571 insertions(+), 163 deletions(-) create mode 100644 src/FileHandlePool.js create mode 100644 src/IndexPool.js create mode 100644 src/Storage/StartupState.js diff --git a/src/EventStore.js b/src/EventStore.js index 429c411e..3d7bd7f8 100644 --- a/src/EventStore.js +++ b/src/EventStore.js @@ -202,6 +202,32 @@ 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) { + const entry = { closed: isClosed, _index: null }; + Object.defineProperty(entry, 'index', { + enumerable: true, + configurable: true, + get: () => { + if (entry._index) { + return entry._index; + } + entry._index = isClosed + ? this.storage.openReadonlyIndex('stream-' + streamName + '.closed') + : this.storage.openIndex('stream-' + streamName); + return entry._index; + } + }); + return entry; + } + /** * 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 +278,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 +793,19 @@ class EventStore extends events.EventEmitter { if (!(streamName in this.streams)) { return; } - this.streams[streamName].index.destroy(); + this.storage.markStartupStateDirty(); + const streamEntry = this.streams[streamName]; + const index = streamEntry._index || null; + this.storage.removeSecondaryIndex('stream-' + streamName); + if (index) { + index.destroy(); + } else { + const fileName = path.join(this.storage.indexDirectory, `${this.storeName}.stream-${streamName}.index`); + if (fs.existsSync(fileName)) { + fs.unlinkSync(fileName); + } + } + this.storage.persistStartupState(); delete this.streams[streamName]; this.emit('stream-deleted', streamName); } @@ -799,6 +830,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 +847,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 00000000..19b544bc --- /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 4ef9c990..474e09af 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 00000000..f68ca96a --- /dev/null +++ b/src/IndexPool.js @@ -0,0 +1,8 @@ +import FileHandlePool from './FileHandlePool.js'; + +/** + * LRU file-handle pool for secondary index files. + */ +class IndexPool extends FileHandlePool {} + +export default IndexPool; diff --git a/src/PartitionPool.js b/src/PartitionPool.js index 55bfc2fa..452548a6 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 cde5e5a2..a615549c 100644 --- a/src/Storage/ReadOnlyStorage.js +++ b/src/Storage/ReadOnlyStorage.js @@ -66,6 +66,7 @@ class ReadOnlyStorage extends ReadableStorage { this.scanSchedule = this.scanSchedule || setTimeout(() => this.scanFiles(() => { this.scanSchedule = null; + this.persistStartupState(); const callbacks = this.onScanFinished || []; this.onScanFinished = []; callbacks.forEach(callback => callback()); @@ -82,12 +83,15 @@ class ReadOnlyStorage extends ReadableStorage { } if (filename.endsWith('.index')) { const indexName = filename.substring(this.storageFile.length + 1, filename.length - 6); + this.discoveredIndexFiles.add(filename); // New indexes are not automatically opened in the reader this.emit('index-created', indexName); + this.persistStartupState(); return; } this.registerPartitionFile(filename); + this.persistStartupState(); } /** diff --git a/src/Storage/ReadableStorage.js b/src/Storage/ReadableStorage.js index c92b7ff8..b9a88326 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,12 @@ 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. + * @param {string} [config.startupState.fileName] Optional manifest file name. */ constructor(storageName = 'storage', config = {}) { super(); @@ -72,7 +81,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 +95,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; @@ -131,6 +144,7 @@ class ReadableStorage extends events.EventEmitter { this.index = index; this.secondaryIndexes = {}; this.readonlyIndexes = {}; + this.discoveredIndexFiles = new Set(); /** Fast secondary-index lookup — classifies matchers for O(1) candidate resolution on write. */ this.indexMatcher = new IndexMatcher(config.matcherProperties); @@ -162,6 +176,61 @@ class ReadableStorage extends events.EventEmitter { return partitionId; } + /** + * Build a startup-state snapshot from current in-memory discovery state. + * + * @protected + * @returns {{ clean: boolean, primaryLength: number, partitions: string[], indexes: string[] }} + */ + buildStartupSnapshot() { + const indexes = Array.from(this.discoveredIndexFiles).sort(); + const partitions = Array.from(this.partitions.values()) + .map(partition => partition.name) + .sort(); + return { + clean: true, + primaryLength: this.index.length, + 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(); + } + + /** + * 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 || []) { + this.discoveredIndexFiles.add(indexFile); + const prefix = this.storageFile + '.'; + if (indexFile.startsWith(prefix) && indexFile.endsWith('.index')) { + const name = indexFile.slice(prefix.length, -'.index'.length); + 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. @@ -185,6 +254,8 @@ class ReadableStorage extends events.EventEmitter { } const indexPattern = new RegExp(`^${escaped}\\.(.+)\\.index$`); scanForFiles(this.indexDirectory, indexPattern, (name) => { + const indexFile = this.storageFile + '.' + name + '.index'; + this.discoveredIndexFiles.add(indexFile); if (!(name in this.secondaryIndexes)) { this.emit('index-created', name); } @@ -227,17 +298,67 @@ class ReadableStorage extends events.EventEmitter { return true; } this.initialized = false; + const state = this.startupState.load(); + if (state && state.clean) { + this.openFromStartupState(state, callback); + 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; } + /** + * Bootstrap from startup state and trigger a scan fallback when manifest length mismatches. + * + * @private + * @param {object} state + * @param {function(): void} [callback] + */ + openFromStartupState(state, callback) { + this.applyStartupSnapshot(state); + this.openIndexes(); + if (typeof state.primaryLength === 'number' && state.primaryLength !== this.index.length) { + this.scanFiles(() => { + if (this.initialized === null) return; + this.initialized = true; + this.persistStartupState(); + callback?.(); + this.emit('opened'); + }); + return; + } + this.initialized = true; + callback?.(); + this.emit('opened'); + setImmediate(() => { + if (this.initialized === true) { + this.scheduleReconciliationScan(); + } + }); + } + + /** + * Reconcile in-memory discovery state with filesystem and refresh startup state. + * + * @protected + */ + scheduleReconciliationScan() { + this.scanFiles(() => { + if (this.initialized === null) { + return; + } + this.persistStartupState(); + }); + } + /** * Close the storage and free up all resources. * Will emit a 'closed' event when finished. @@ -255,6 +376,7 @@ class ReadableStorage extends events.EventEmitter { index.close(); } this.forEachPartition(partition => partition.close()); + this.secondaryIndexHandles.clearOpenHandles(); this.emit('closed'); } @@ -398,7 +520,9 @@ class ReadableStorage extends events.EventEmitter { 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.discoveredIndexFiles.add(indexName); + this.secondaryIndexHandles.add(name, index); + this.secondaryIndexHandles.open(name); this.readonlyIndexes[name] = index; return index; } @@ -418,6 +542,7 @@ class ReadableStorage extends events.EventEmitter { return this.index; } if (name in this.secondaryIndexes) { + this.secondaryIndexHandles.open(name); return this.secondaryIndexes[name].index; } @@ -426,11 +551,13 @@ class ReadableStorage extends events.EventEmitter { const metadata = buildMetadataForMatcher(matcher, this.hmac); let { index } = this.secondaryIndexes[name] = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata })); + this.discoveredIndexFiles.add(indexName); + this.secondaryIndexHandles.add(name, index); // Register the actual stored matcher (may have been reconstructed from metadata by WritableStorage.createIndex). this.indexMatcher.add(name, this.secondaryIndexes[name].matcher); - index.open(); + this.secondaryIndexHandles.open(name); return index; } @@ -443,6 +570,7 @@ class ReadableStorage extends events.EventEmitter { removeSecondaryIndex(name) { const entry = this.secondaryIndexes[name]; if (entry) { + this.secondaryIndexHandles.remove(name); this.indexMatcher.remove(name); delete this.secondaryIndexes[name]; } diff --git a/src/Storage/StartupState.js b/src/Storage/StartupState.js new file mode 100644 index 00000000..73cf45a0 --- /dev/null +++ b/src/Storage/StartupState.js @@ -0,0 +1,101 @@ +import fs from 'fs'; +import path from 'path'; +import { hash } from '../utils/util.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] + * @param {string} [config.fileName] + */ + constructor(storageFile, directory, config = {}) { + this.enabled = !!config.enabled; + this.storageFile = storageFile; + this.fileName = path.resolve(directory, config.fileName || `${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) + }; + const tmpFile = `${this.fileName}.${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}.tmp`; + fs.writeFileSync(tmpFile, JSON.stringify(envelope)); + fs.renameSync(tmpFile, this.fileName); + this.lastState = payload; + } + + /** + * Mark the state as dirty before mutating on-disk layout. + */ + markDirty() { + if (!this.enabled) { + return; + } + const base = this.lastState || { + clean: true, + primaryLength: 0, + partitions: [], + indexes: [] + }; + this.save({ ...base, clean: false }); + } +} + +export default StartupState; diff --git a/src/Storage/WritableStorage.js b/src/Storage/WritableStorage.js index cb1958c9..792e2cb9 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); } @@ -152,6 +155,7 @@ class WritableStorage extends ReadableStorage { * 4. If no torn writes were found but the index is lagging, reindex directly. */ checkTornWrites() { + this.markStartupStateDirty(); const { lastValidSequenceNumber, maxPartitionSequenceNumber } = this.findTornWriteBoundary(); if (lastValidSequenceNumber < Number.MAX_SAFE_INTEGER) { @@ -179,6 +183,7 @@ class WritableStorage extends ReadableStorage { this.forEachPartition(partition => partition.close()); // Partitions were closed directly (bypassing the pool), so reset the open-handle tracking. this.partitions.clearOpenHandles(); + this.persistStartupState(); } /** @@ -194,6 +199,7 @@ class WritableStorage extends ReadableStorage { * Defaults to 0, which rebuilds all indexes from scratch. */ reindex(fromSequenceNumber = 0) { + this.markStartupStateDirty(); this.index.truncate(fromSequenceNumber); // Truncate all loaded secondary indexes to match the new primary length. @@ -214,6 +220,7 @@ class WritableStorage extends ReadableStorage { } this.flush(); + this.persistStartupState(); } /** @@ -334,10 +341,12 @@ class WritableStorage extends ReadableStorage { if (this.partitions.has(partitionId)) { return; } + this.markStartupStateDirty(); const partitionConfig = this.buildPartitionConfig(partitionShortName); this.ensurePartitionDirectory(partitionName); this.partitions.add(partitionId, this.createPartition(partitionName, partitionConfig)); this.emit('partition-created', partitionId); + this.persistStartupState(); } /** @@ -382,7 +391,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); @@ -415,6 +424,7 @@ class WritableStorage extends ReadableStorage { } assert((typeof matcher === 'object' || typeof matcher === 'function') && matcher !== null, 'Need to specify a matcher.'); + this.markStartupStateDirty(); const metadata = buildMetadataForMatcher(matcher, this.hmac); const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata })); @@ -432,8 +442,12 @@ class WritableStorage extends ReadableStorage { } this.secondaryIndexes[name] = { index, matcher }; + this.secondaryIndexHandles.add(name, index); + this.secondaryIndexHandles.open(name); + this.discoveredIndexFiles.add(indexName); this.indexMatcher.add(name, matcher); this.emit('index-created', name); + this.persistStartupState(); return index; } @@ -501,6 +515,7 @@ class WritableStorage extends ReadableStorage { * @param {number} after The document sequence number to truncate after. */ truncate(after) { + this.markStartupStateDirty(); /* To truncate the store following steps need to be done: @@ -520,6 +535,7 @@ class WritableStorage extends ReadableStorage { this.forEachWritableSecondaryIndex(index => { index.truncate(index.find(after)); }); + this.persistStartupState(); } /** diff --git a/test/EventStore.spec.js b/test/EventStore.spec.js index f0127d91..b610957e 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 6daee2e7..063b9cc5 100644 --- a/test/Storage.spec.js +++ b/test/Storage.spec.js @@ -1817,6 +1817,80 @@ 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, fileName: 'startup-state-test.json' }; + + storage = createStorage({ startupState }); + storage.open(); + storage.write({ type: 'one', foo: 1 }); + storage.ensureIndex('one', { type: 'one' }); + storage.close(); + + storage = createStorage({ startupState }); + storage.scheduleReconciliationScan = () => null; + 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, fileName: 'startup-state-test.json' }; + + 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(); + }); + }); + }); }); From 309342775e0b6fc2a4bfcac104a380d5e3104f45 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 18 Jul 2026 18:45:09 +0000 Subject: [PATCH 2/2] refactor startup-state/index lifecycle per review feedback --- src/Consumer.js | 11 +- src/EventStore.js | 38 ++---- src/IndexPool.js | 52 +++++++- src/Storage/ReadOnlyStorage.js | 5 +- src/Storage/ReadableStorage.js | 237 +++++++++++++++++++++++---------- src/Storage/StartupState.js | 9 +- src/Storage/WritableStorage.js | 75 +++++------ src/utils/apiHelpers.js | 34 ++++- src/utils/fsUtil.js | 24 ++++ test/Storage.spec.js | 9 +- 10 files changed, 327 insertions(+), 167 deletions(-) diff --git a/src/Consumer.js b/src/Consumer.js index 58411b57..4e513839 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 3d7bd7f8..2333a51c 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"; @@ -211,21 +211,13 @@ class EventStore extends events.EventEmitter { * @returns {{ closed: boolean, _index: object|null, index: object }} */ createLazyStreamEntry(streamName, isClosed) { - const entry = { closed: isClosed, _index: null }; - Object.defineProperty(entry, 'index', { - enumerable: true, - configurable: true, - get: () => { - if (entry._index) { - return entry._index; - } - entry._index = isClosed - ? this.storage.openReadonlyIndex('stream-' + streamName + '.closed') - : this.storage.openIndex('stream-' + streamName); - return entry._index; - } - }); - return entry; + return createLazyPropertyHolder( + { closed: isClosed, _index: null }, + 'index', + () => isClosed + ? this.storage.openReadonlyIndex('stream-' + streamName + '.closed') + : this.storage.openIndex('stream-' + streamName) + ); } /** @@ -793,19 +785,7 @@ class EventStore extends events.EventEmitter { if (!(streamName in this.streams)) { return; } - this.storage.markStartupStateDirty(); - const streamEntry = this.streams[streamName]; - const index = streamEntry._index || null; - this.storage.removeSecondaryIndex('stream-' + streamName); - if (index) { - index.destroy(); - } else { - const fileName = path.join(this.storage.indexDirectory, `${this.storeName}.stream-${streamName}.index`); - if (fs.existsSync(fileName)) { - fs.unlinkSync(fileName); - } - } - this.storage.persistStartupState(); + this.storage.deleteSecondaryIndex('stream-' + streamName); delete this.streams[streamName]; this.emit('stream-deleted', streamName); } diff --git a/src/IndexPool.js b/src/IndexPool.js index f68ca96a..11a9b5f4 100644 --- a/src/IndexPool.js +++ b/src/IndexPool.js @@ -3,6 +3,56 @@ import FileHandlePool from './FileHandlePool.js'; /** * LRU file-handle pool for secondary index files. */ -class IndexPool extends FileHandlePool {} +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/Storage/ReadOnlyStorage.js b/src/Storage/ReadOnlyStorage.js index a615549c..149d1f3f 100644 --- a/src/Storage/ReadOnlyStorage.js +++ b/src/Storage/ReadOnlyStorage.js @@ -66,7 +66,6 @@ class ReadOnlyStorage extends ReadableStorage { this.scanSchedule = this.scanSchedule || setTimeout(() => this.scanFiles(() => { this.scanSchedule = null; - this.persistStartupState(); const callbacks = this.onScanFinished || []; this.onScanFinished = []; callbacks.forEach(callback => callback()); @@ -83,15 +82,13 @@ class ReadOnlyStorage extends ReadableStorage { } if (filename.endsWith('.index')) { const indexName = filename.substring(this.storageFile.length + 1, filename.length - 6); - this.discoveredIndexFiles.add(filename); + this.registerSecondaryIndexDescriptor(indexName); // New indexes are not automatically opened in the reader this.emit('index-created', indexName); - this.persistStartupState(); return; } this.registerPartitionFile(filename); - this.persistStartupState(); } /** diff --git a/src/Storage/ReadableStorage.js b/src/Storage/ReadableStorage.js index b9a88326..482e0188 100644 --- a/src/Storage/ReadableStorage.js +++ b/src/Storage/ReadableStorage.js @@ -66,7 +66,6 @@ class ReadableStorage extends events.EventEmitter { * Default: 1024. * @param {object} [config.startupState] Startup manifest options. * @param {boolean} [config.startupState.enabled=false] Enable startup state manifest fast-path. - * @param {string} [config.startupState.fileName] Optional manifest file name. */ constructor(storageName = 'storage', config = {}) { super(); @@ -142,9 +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.discoveredIndexFiles = new Set(); + this.secondaryIndexes = this.secondaryIndexHandles.registry; /** Fast secondary-index lookup — classifies matchers for O(1) candidate resolution on write. */ this.indexMatcher = new IndexMatcher(config.matcherProperties); @@ -176,20 +173,76 @@ 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, primaryLength: number, partitions: string[], indexes: string[] }} + * @returns {{ clean: boolean, partitions: string[], indexes: string[] }} */ buildStartupSnapshot() { - const indexes = Array.from(this.discoveredIndexFiles).sort(); + 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, - primaryLength: this.index.length, partitions, indexes }; @@ -211,6 +264,23 @@ class ReadableStorage extends events.EventEmitter { 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. * @@ -222,10 +292,10 @@ class ReadableStorage extends events.EventEmitter { this.registerPartitionFile(partitionName); } for (const indexFile of state.indexes || []) { - this.discoveredIndexFiles.add(indexFile); 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); } } @@ -254,9 +324,8 @@ class ReadableStorage extends events.EventEmitter { } const indexPattern = new RegExp(`^${escaped}\\.(.+)\\.index$`); scanForFiles(this.indexDirectory, indexPattern, (name) => { - const indexFile = this.storageFile + '.' + name + '.index'; - this.discoveredIndexFiles.add(indexFile); if (!(name in this.secondaryIndexes)) { + this.registerSecondaryIndexDescriptor(name); this.emit('index-created', name); } }, (indexErr) => { @@ -299,8 +368,12 @@ class ReadableStorage extends events.EventEmitter { } this.initialized = false; const state = this.startupState.load(); - if (state && state.clean) { - this.openFromStartupState(state, callback); + if (state && state.clean && this.isStartupStateSafe(state)) { + this.applyStartupSnapshot(state); + this.initialized = true; + this.openIndexes(); + callback?.(); + this.emit('opened'); return true; } this.scanFiles(() => { @@ -316,47 +389,27 @@ class ReadableStorage extends events.EventEmitter { } /** - * Bootstrap from startup state and trigger a scan fallback when manifest length mismatches. - * - * @private * @param {object} state - * @param {function(): void} [callback] - */ - openFromStartupState(state, callback) { - this.applyStartupSnapshot(state); - this.openIndexes(); - if (typeof state.primaryLength === 'number' && state.primaryLength !== this.index.length) { - this.scanFiles(() => { - if (this.initialized === null) return; - this.initialized = true; - this.persistStartupState(); - callback?.(); - this.emit('opened'); - }); - return; + * @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; + } } - this.initialized = true; - callback?.(); - this.emit('opened'); - setImmediate(() => { - if (this.initialized === true) { - this.scheduleReconciliationScan(); + for (const indexFile of state.indexes || []) { + if (!indexFile.startsWith(indexPathPrefix) || !indexFile.endsWith('.index')) { + return false; } - }); - } - - /** - * Reconcile in-memory discovery state with filesystem and refresh startup state. - * - * @protected - */ - scheduleReconciliationScan() { - this.scanFiles(() => { - if (this.initialized === null) { - return; + if (!fs.existsSync(path.join(this.indexDirectory, indexFile))) { + return false; } - this.persistStartupState(); - }); + } + return true; } /** @@ -371,9 +424,8 @@ 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(); @@ -514,17 +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)); - this.discoveredIndexFiles.add(indexName); - this.secondaryIndexHandles.add(name, index); this.secondaryIndexHandles.open(name); - this.readonlyIndexes[name] = index; - return index; + return descriptor.index; } /** @@ -541,21 +590,25 @@ class ReadableStorage extends events.EventEmitter { if (name === '_all') { return this.index; } - if (name in this.secondaryIndexes) { + const descriptor = this.registerSecondaryIndexDescriptor(name, { closed: false }); + if (descriptor.index) { this.secondaryIndexHandles.open(name); - return this.secondaryIndexes[name].index; + 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 })); - this.discoveredIndexFiles.add(indexName); - this.secondaryIndexHandles.add(name, index); - - // 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); + } this.secondaryIndexHandles.open(name); return index; @@ -571,11 +624,41 @@ class ReadableStorage extends events.EventEmitter { const entry = this.secondaryIndexes[name]; if (entry) { this.secondaryIndexHandles.remove(name); - this.indexMatcher.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 @@ -665,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 index 73cf45a0..5ecbbbb0 100644 --- a/src/Storage/StartupState.js +++ b/src/Storage/StartupState.js @@ -1,6 +1,7 @@ import fs from 'fs'; import path from 'path'; import { hash } from '../utils/util.js'; +import { writeFileAtomic } from '../utils/fsUtil.js'; const STATE_VERSION = 1; @@ -14,12 +15,11 @@ class StartupState { * @param {string} directory * @param {object} [config] * @param {boolean} [config.enabled=false] - * @param {string} [config.fileName] */ constructor(storageFile, directory, config = {}) { this.enabled = !!config.enabled; this.storageFile = storageFile; - this.fileName = path.resolve(directory, config.fileName || `${storageFile}.startup-state.json`); + this.fileName = path.resolve(directory, `${storageFile}.startup-state.json`); this.lastState = null; } @@ -75,9 +75,7 @@ class StartupState { payload, checksum: this.checksum(payload) }; - const tmpFile = `${this.fileName}.${process.pid}.${Date.now()}.${Math.random().toString(16).slice(2)}.tmp`; - fs.writeFileSync(tmpFile, JSON.stringify(envelope)); - fs.renameSync(tmpFile, this.fileName); + writeFileAtomic(this.fileName, JSON.stringify(envelope)); this.lastState = payload; } @@ -90,7 +88,6 @@ class StartupState { } const base = this.lastState || { clean: true, - primaryLength: 0, partitions: [], indexes: [] }; diff --git a/src/Storage/WritableStorage.js b/src/Storage/WritableStorage.js index 792e2cb9..8c5957e1 100644 --- a/src/Storage/WritableStorage.js +++ b/src/Storage/WritableStorage.js @@ -155,7 +155,6 @@ class WritableStorage extends ReadableStorage { * 4. If no torn writes were found but the index is lagging, reindex directly. */ checkTornWrites() { - this.markStartupStateDirty(); const { lastValidSequenceNumber, maxPartitionSequenceNumber } = this.findTornWriteBoundary(); if (lastValidSequenceNumber < Number.MAX_SAFE_INTEGER) { @@ -183,7 +182,6 @@ class WritableStorage extends ReadableStorage { this.forEachPartition(partition => partition.close()); // Partitions were closed directly (bypassing the pool), so reset the open-handle tracking. this.partitions.clearOpenHandles(); - this.persistStartupState(); } /** @@ -199,7 +197,6 @@ class WritableStorage extends ReadableStorage { * Defaults to 0, which rebuilds all indexes from scratch. */ reindex(fromSequenceNumber = 0) { - this.markStartupStateDirty(); this.index.truncate(fromSequenceNumber); // Truncate all loaded secondary indexes to match the new primary length. @@ -220,7 +217,6 @@ class WritableStorage extends ReadableStorage { } this.flush(); - this.persistStartupState(); } /** @@ -341,12 +337,12 @@ class WritableStorage extends ReadableStorage { if (this.partitions.has(partitionId)) { return; } - this.markStartupStateDirty(); - const partitionConfig = this.buildPartitionConfig(partitionShortName); - this.ensurePartitionDirectory(partitionName); - this.partitions.add(partitionId, this.createPartition(partitionName, partitionConfig)); - this.emit('partition-created', partitionId); - this.persistStartupState(); + this.withStartupStateMutation(() => { + const partitionConfig = this.buildPartitionConfig(partitionShortName); + this.ensurePartitionDirectory(partitionName); + this.partitions.add(partitionId, this.createPartition(partitionName, partitionConfig)); + this.emit('partition-created', partitionId); + }); } /** @@ -414,41 +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.'); - this.markStartupStateDirty(); - - 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.secondaryIndexHandles.add(name, index); - this.secondaryIndexHandles.open(name); - this.discoveredIndexFiles.add(indexName); - this.indexMatcher.add(name, matcher); - this.emit('index-created', name); - this.persistStartupState(); - 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; + }); } /** @@ -515,7 +509,6 @@ class WritableStorage extends ReadableStorage { * @param {number} after The document sequence number to truncate after. */ truncate(after) { - this.markStartupStateDirty(); /* To truncate the store following steps need to be done: @@ -535,7 +528,6 @@ class WritableStorage extends ReadableStorage { this.forEachWritableSecondaryIndex(index => { index.truncate(index.find(after)); }); - this.persistStartupState(); } /** @@ -554,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 d6f18aa3..7802e13c 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 deac4721..6e20e1a8 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/Storage.spec.js b/test/Storage.spec.js index 063b9cc5..f89cf2a2 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(); }); @@ -1849,7 +1849,7 @@ describe('Storage', function() { describe('startupState', function() { it('uses a clean manifest fast-path without calling scanFiles on open', function() { - const startupState = { enabled: true, fileName: 'startup-state-test.json' }; + const startupState = { enabled: true }; storage = createStorage({ startupState }); storage.open(); @@ -1858,7 +1858,6 @@ describe('Storage', function() { storage.close(); storage = createStorage({ startupState }); - storage.scheduleReconciliationScan = () => null; storage.scanFiles = () => { throw new Error('scanFiles should not be called for clean startup-state fast-path'); }; @@ -1868,7 +1867,7 @@ describe('Storage', function() { }); it('falls back to scan when manifest is dirty', function(done) { - const startupState = { enabled: true, fileName: 'startup-state-test.json' }; + const startupState = { enabled: true }; storage = createStorage({ startupState }); storage.open();