Scale startup for 10k+ streams via lazy stream indexes, startup manifest bootstrap, and bounded index FDs

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);
             }
         });
     }

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);
     }
 

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<object>}
+     */
+    *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;

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();
+        }
     }
 
     /**

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;