diff --git a/AGENTS.md b/AGENTS.md index 69aa7a3..2c9e6ba 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,6 +20,7 @@ Always try to use available tools through the MCP servers, which should provide - **Pre-compile over per-call dispatch**: if a descriptor or configuration object is known at setup time, compile it into closures or lookup structures once rather than re-interpreting it on every hot-path call. The gain compounds with call frequency (e.g. building operator closures once per matcher object instead of running `Object.entries` + switch per document). - **WeakMap for caching derived state on user objects**: prefer `WeakMap` over Symbol properties or plain properties to attach computed state keyed on user-supplied objects — avoids mutation, works on frozen objects, and is GC-friendly. On the hot path use `cache.get()` directly rather than `has()` + `get()` to avoid a double lookup. - **Pass cheap prefilter results as hints to expensive scans**: when a cheap pass (e.g. `Buffer.indexOf`) already locates a candidate position, carry that result forward as a hint to the costlier depth- or context-aware pass rather than letting it re-scan from the start. Eliminates a full O(n) scan on every call. +- **Virtualize known lengths instead of pre-allocating sparse arrays**: when startup metadata already carries collection size (e.g. secondary index entry counts), keep an explicit length field and lazily fill cache entries on demand. Avoids large sparse-array allocations and reduces startup heap pressure. - **Custom implementations only when benchmarks justify the complexity**: a hand-rolled solution may edge out a standard library call by 10–15 %, but the maintenance cost is rarely worth it unless the gain is substantial and measurable at realistic data sizes (e.g. a custom numeric parser vs. `JSON.parse` with try/catch). - **Prefer a flag on a shared helper over near-duplicate functions**: when two functions differ only in a single behavioral detail, add a boolean parameter to the shared function rather than maintaining two near-identical copies. Keep the flag's semantics explicit and limited to one axis of variation. - **Inline single-use hot-path helpers once the surrounding flow becomes simpler**: if a helper is only called from one place and its logic can be embedded while keeping the caller below the local complexity/return-count budget, prefer the inline version. Removes call indirection and often makes the hot path easier to read as straight-line pseudo-code. @@ -101,7 +102,6 @@ No build step; source is plain ESM consumed directly. No linter configured. | `src/Clock.js` | Monotonic microsecond clock | | `src/IndexEntry.js` | Index record serialization | | `src/IndexMatcher.js` | O(1) discriminant-based matcher classification | -| `src/PartitionPool.js` | LRU-evicting pool for open partition handles | | `src/Watcher.js` / `src/WatchesFile.js` | Ref-counting directory watcher and mixin | | `src/utils/fsUtil.js` | `ensureDirectory`, `scanForFiles` | | `src/utils/apiHelpers.js` | API-shaping helpers for top-level and internal class APIs (argument normalization, stream-name mapping, predicate/raw/revision coercion) | diff --git a/bench/bench-scalability.js b/bench/bench-scalability.js index cfbca15..c29d135 100644 --- a/bench/bench-scalability.js +++ b/bench/bench-scalability.js @@ -12,12 +12,12 @@ * One secondary index per partition (N indexes for N partitions). * * Scenario B – fixed partitions (100), growing index count - * Size steps: 100 / 1000 / 5000 indexes + * Size steps: 100 / 1000 / 5000 / 10000 / 20000 / 50000 indexes * 20 documents per index are guaranteed across the 100 partitions. * * Metrics collected at each size step - * - startup_ms – time to open a pre-populated storage, including scanning - * the data directory for all partition files plus opening all + * - startup_ms – time to reopen a pre-populated storage after a successful + * close() wrote the startup manifest, plus opening all * secondary index files one by one. * - write_ms_op – average ms per write() call (100-op sample), with all N * secondary indexes registered so every write checks all N @@ -47,20 +47,18 @@ * 20 000 partitions creates ≈ 40 001 files in a single directory. * * 3. Population strategy (secondary indexes) - * Secondary index files are created as empty shell files (metadata header - * only, no historical data) during the setup phase. This keeps population - * time O(docs + indexes) instead of the quadratic O(docs × indexes) that - * results from using Storage.ensureIndex(…, reindex=true) for each index. - * The write benchmark still measures the full O(N) per-write cost because - * every write() call iterates all N registered secondary-index matchers to - * find the matching one. In a real application that starts fresh (no - * historical events), the behaviour is identical. + * Secondary indexes are registered before writes so index files contain + * actual entries. This makes startup measurements reflect real-world reopen + * cost with non-empty index files. Population stays O(docs + indexes): each + * write targets one index via matcherProperties-based O(1) discriminant + * lookup, avoiding the quadratic O(docs × indexes) matcher scan. * * 4. Total data written (Scenario B) * With M indexes and 20 docs/index, partition data grows to: * 100 partitions × ceil(M×20/100) docs/partition × ~200 bytes ≈ - * 100 / 1 000 indexes: ~ 2 MB / 20 MB partition data - * 5 000 indexes: ~ 100 MB partition data + * 100 / 1 000 indexes: ~ 2 MB / 20 MB partition data + * 5 000 / 10 000: ~ 100 MB / 200 MB partition data + * 20 000 / 50 000: ~ 400 MB / 1 GB partition data * Ensure the target --data-dir has enough free space. * * Usage: @@ -103,9 +101,12 @@ const FD_LIMIT = getFdSoftLimit(); // ────────────────────────────────────────────────────────────────────────────── // Constants // ────────────────────────────────────────────────────────────────────────────── -const DOCS_PER_PARTITION = 20; -const WRITE_SAMPLE_OPS = 100; -const READ_SAMPLE_OPS = 100; +const DOCS_PER_PARTITION = 20; +const WRITE_SAMPLE_OPS = 100; +const READ_SAMPLE_OPS = 100; +// Write partitions in sequential groups during population so the LRU partition +// pool never needs to hold more than this many partitions open at once. +const POPULATE_PARTITION_BATCH = 500; // Pad the data field so the serialised JSON document is roughly 150 bytes. // Documents use a `stream` property so they align with the default `matcherProperties` @@ -117,7 +118,7 @@ const APPROX_DOC_SIZE = JSON.stringify({ stream: 'x', partitionId: 0, data: DATA // Scenario definitions // ────────────────────────────────────────────────────────────────────────────── const SCENARIO_A_STEPS = [100, 1000, 5000, 10000, 20000]; -const SCENARIO_B_STEPS = [100, 1000, 5000]; +const SCENARIO_B_STEPS = [100, 1000, 5000, 10000, 20000, 50000]; const SCENARIO_B_PARTITIONS = 100; // ────────────────────────────────────────────────────────────────────────────── @@ -135,6 +136,14 @@ function elapsed(start) { return s * 1e3 + ns / 1e6; } +function openStorage(storage) { + return new Promise((resolve) => storage.open(resolve)); +} + +function manifestFile(dataDir) { + return path.join(dataDir, '.bench.manifest.json'); +} + function makeStorageConfig(dataDir, partitioner) { return { dataDirectory: dataDir, @@ -161,11 +170,9 @@ function estimateFds(numPartitions, numIndexes) { /** * Populate a fresh data directory: - * 1. Write all documents directly to partitions (no secondary indexes - * registered → O(totalDocs) instead of O(totalDocs × numIndexes)). - * 2. Create secondary index files as empty shells (header + matcher - * metadata, no historical entries) so that startup and write benchmarks - * see a realistic file-count overhead without quadratic setup time. + * 1. Register all secondary indexes once. + * 2. Write all documents so index files are non-empty and startup includes + * realistic secondary-index load work. * * Documents use `stream` as the discriminant property (matching the default * `matcherProperties` config), so each index's object matcher { stream: i } @@ -176,47 +183,60 @@ function estimateFds(numPartitions, numIndexes) { * @param {number} numIndexes * @param {number} docsPerPartition */ -function populateStorage(dataDir, numPartitions, numIndexes, docsPerPartition) { +async function populateStorage(dataDir, numPartitions, numIndexes, docsPerPartition) { const partitioner = (doc) => String(doc.partitionId); const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); - storage.open(); - - // Write all documents with NO secondary indexes registered. - const totalDocs = numPartitions * docsPerPartition; - for (let seq = 0; seq < totalDocs; seq++) { - const p = seq % numPartitions; - const typeId = numIndexes > 0 ? seq % numIndexes : 0; - storage.write({ stream: String(typeId), partitionId: p, data: DATA_PAD, ts: Date.now() }); - } - storage.flush(); + await openStorage(storage); - // Create empty secondary index files (reindex=false → no doc scanning). - // This registers the file on disk so it can be opened in the benchmark. + // Register all secondary indexes once. With matcherProperties this remains + // O(1) index selection on write. for (let i = 0; i < numIndexes; i++) { storage.ensureIndex(`idx-${i}`, { stream: String(i) }, false); } + + // Write all documents partition-by-partition in batches of POPULATE_PARTITION_BATCH. + // Sequential writes keep at most POPULATE_PARTITION_BATCH partitions in the LRU pool + // at any point, so the maxOpenPartitions cap does not cause close+reopen churn + // during the (non-critical) population phase. + for (let pBatch = 0; pBatch < numPartitions; pBatch += POPULATE_PARTITION_BATCH) { + const pEnd = Math.min(pBatch + POPULATE_PARTITION_BATCH, numPartitions); + for (let p = pBatch; p < pEnd; p++) { + for (let d = 0; d < docsPerPartition; d++) { + const typeId = numIndexes > 0 ? (p * docsPerPartition + d) % numIndexes : 0; + storage.write({ stream: String(typeId), partitionId: p, data: DATA_PAD, ts: Date.now() }); + } + } + } storage.flush(); storage.close(); + if (!fs.existsSync(manifestFile(dataDir))) { + throw new Error('Storage close() returned before writing the startup manifest.'); + } } /** - * Measure startup time: open a pre-populated storage (directory scan + - * primary index load) and then explicitly open all secondary index files, - * mirroring what a real application does on boot. + * Measure startup time: reopen a pre-populated storage after a successful + * close() persisted the startup manifest. The clock stops as soon as the + * open() callback fires — at that point the storage is fully usable. The + * openIndex calls that follow are outside the timed region; they serve only + * to keep index file descriptors warm for the subsequent write/read samples. * - * @returns {number} Total elapsed milliseconds. + * @returns {Promise} Total elapsed milliseconds. */ function measureStartup(dataDir, numIndexes) { - const partitioner = (doc) => String(doc.partitionId); - const t0 = process.hrtime(); - const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); - storage.open(); - for (let i = 0; i < numIndexes; i++) { - storage.openIndex(`idx-${i}`); - } - const ms = elapsed(t0); - storage.close(); - return ms; + return new Promise((resolve) => { + const partitioner = (doc) => String(doc.partitionId); + const t0 = process.hrtime(); + const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); + storage.open(() => { + const ms = elapsed(t0); + for (let i = 0; i < numIndexes; i++) { + storage.openIndex(`idx-${i}`); + } + storage.close(); + resolve(ms); + }); + }); } /** @@ -226,53 +246,58 @@ function measureStartup(dataDir, numIndexes) { * With the discriminant lookup table, each write() resolves the matching index * via an O(1) Map lookup on the `stream` property instead of iterating all N matchers. * - * @returns {number} Average ms per write() call. + * @returns {Promise} Average ms per write() call. */ function measureWrite(dataDir, numPartitions, numIndexes) { - const partitioner = (doc) => String(doc.partitionId); - const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); - storage.open(); - - // Open all secondary indexes outside the timed region. - for (let i = 0; i < numIndexes; i++) { - storage.openIndex(`idx-${i}`); - } - - const t0 = process.hrtime(); - for (let i = 0; i < WRITE_SAMPLE_OPS; i++) { - const p = i % numPartitions; - const typeId = numIndexes > 0 ? i % numIndexes : 0; - storage.write({ stream: String(typeId), partitionId: p, data: DATA_PAD, ts: Date.now() }); - } - storage.flush(); - const ms = elapsed(t0); - - storage.close(); - return ms / WRITE_SAMPLE_OPS; + return new Promise((resolve) => { + const partitioner = (doc) => String(doc.partitionId); + const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); + storage.open(() => { + // Open all secondary indexes outside the timed region. + for (let i = 0; i < numIndexes; i++) { + storage.openIndex(`idx-${i}`); + } + + const t0 = process.hrtime(); + for (let i = 0; i < WRITE_SAMPLE_OPS; i++) { + const p = i % numPartitions; + const typeId = numIndexes > 0 ? i % numIndexes : 0; + storage.write({ stream: String(typeId), partitionId: p, data: DATA_PAD, ts: Date.now() }); + } + storage.flush(); + const ms = elapsed(t0); + + storage.close(); + resolve(ms / WRITE_SAMPLE_OPS); + }); + }); } /** * Measure read performance using evenly-spread positions in the primary index. + * Waits for the full directory scan so all partition handles are registered. * - * @returns {number} Average ms per read() call. + * @returns {Promise} Average ms per read() call. */ function measureRead(dataDir) { - const partitioner = (doc) => String(doc.partitionId); - const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); - storage.open(); - - const length = storage.length; - const step = Math.max(1, Math.floor(length / READ_SAMPLE_OPS)); - - const t0 = process.hrtime(); - for (let i = 0; i < READ_SAMPLE_OPS; i++) { - const pos = ((i * step) % length) + 1; - storage.read(pos); - } - const ms = elapsed(t0); - - storage.close(); - return ms / READ_SAMPLE_OPS; + return new Promise((resolve) => { + const partitioner = (doc) => String(doc.partitionId); + const storage = new Storage('bench', makeStorageConfig(dataDir, partitioner)); + storage.open(() => { + const length = storage.length; + const step = Math.max(1, Math.floor(length / READ_SAMPLE_OPS)); + + const t0 = process.hrtime(); + for (let i = 0; i < READ_SAMPLE_OPS; i++) { + const pos = ((i * step) % length) + 1; + storage.read(pos); + } + const ms = elapsed(t0); + + storage.close(); + resolve(ms / READ_SAMPLE_OPS); + }); + }); } // ────────────────────────────────────────────────────────────────────────────── @@ -352,6 +377,8 @@ console.log(`Write-sample ops: ${WRITE_SAMPLE_OPS} Read-sample ops: ${READ_SAM fs.mkdirSync(BASE_DIR, { recursive: true }); +(async () => { + // ─── Scenario A ────────────────────────────────────────────────────────────── console.log('\n== Scenario A: growing partitions (1 index per partition) =='); const scenarioAResults = []; @@ -378,16 +405,16 @@ for (const numPartitions of SCENARIO_A_STEPS) { ` [${numPartitions}p / ${numIndexes}idx / ${totalDocs} docs] populate … ` ); const tPop = Date.now(); - populateStorage(dataDir, numPartitions, numIndexes, docsPerPartition); + await populateStorage(dataDir, numPartitions, numIndexes, docsPerPartition); process.stdout.write(`${Date.now() - tPop} ms | `); - const startup_ms = measureStartup(dataDir, numIndexes); + const startup_ms = await measureStartup(dataDir, numIndexes); process.stdout.write(`startup ${fmtMs(startup_ms)} ms | `); - const write_ms_op = measureWrite(dataDir, numPartitions, numIndexes); + const write_ms_op = await measureWrite(dataDir, numPartitions, numIndexes); process.stdout.write(`write ${fmtMs(write_ms_op)} ms/op | `); - const read_ms_op = measureRead(dataDir); + const read_ms_op = await measureRead(dataDir); process.stdout.write(`read ${fmtMs(read_ms_op)} ms/op\n`); scenarioAResults.push({ size: numPartitions, startup_ms, write_ms_op, read_ms_op }); @@ -432,16 +459,16 @@ for (const numIndexes of SCENARIO_B_STEPS) { ` (${docsPerPartition}/p)] populate … ` ); const tPop = Date.now(); - populateStorage(dataDir, numPartitions, numIndexes, docsPerPartition); + await populateStorage(dataDir, numPartitions, numIndexes, docsPerPartition); process.stdout.write(`${Date.now() - tPop} ms | `); - const startup_ms = measureStartup(dataDir, numIndexes); + const startup_ms = await measureStartup(dataDir, numIndexes); process.stdout.write(`startup ${fmtMs(startup_ms)} ms | `); - const write_ms_op = measureWrite(dataDir, numPartitions, numIndexes); + const write_ms_op = await measureWrite(dataDir, numPartitions, numIndexes); process.stdout.write(`write ${fmtMs(write_ms_op)} ms/op | `); - const read_ms_op = measureRead(dataDir); + const read_ms_op = await measureRead(dataDir); process.stdout.write(`read ${fmtMs(read_ms_op)} ms/op\n`); scenarioBResults.push({ size: numIndexes, startup_ms, write_ms_op, read_ms_op }); @@ -456,3 +483,5 @@ printTable( ); console.log('\nDone.'); + +})(); diff --git a/concepts/2026-07-25-storage-manifest.md b/concepts/2026-07-25-storage-manifest.md new file mode 100644 index 0000000..d98389b --- /dev/null +++ b/concepts/2026-07-25-storage-manifest.md @@ -0,0 +1,131 @@ +# Storage startup manifest for fast secondary-index recovery + +## Problem + +Every time a `WritableStorage` instance opens, it must discover and restore all secondary +indexes that were registered in the previous session. Without a manifest the only way to +do this is a full filesystem scan: + +1. **`scanFiles()`** — an O(N) `readdir` over the data/index directory. +2. **Per-index file open** — for each `.index` file found, `openIndex()` opens the file, + reads its binary header (magic + metadata block) to reconstruct the matcher, reads the + file length to restore `index.data.length`, then closes the fd. + +With 5 000 secondary indexes this means at least 5 000 file-open/read/close cycles at +startup, even before a single document is processed. Startup time grows linearly with the +number of indexes. + +## Solution: persist a manifest at graceful shutdown + +`WritableStorage` writes a single JSON manifest file to the index directory on every +graceful `close()`. On the next `open()`, if the manifest exists and its HMAC +verifies, the storage restores all secondary indexes and partition registrations from the +manifest **without any file I/O**, achieving nearly O(1) startup (bounded only by the time +to read and parse the manifest). + +## Manifest format + +``` +..manifest.json +``` + +```json +{ + "version": 1, + "partitions": ["bench.0", "bench.1", "bench.100"], + "indexes": { + "stream-orders": { + "matcher": { "stream": "orders" }, + "length": 4712, + "headerSize": 64 + }, + "audit-log": { + "matcher": "function(doc) { return doc.audit === true; }", + "length": 800, + "headerSize": 64 + } + }, + "hmac": "" +} +``` + +Fields per index entry: + +| Field | Purpose | +|---|---| +| `matcher` | Object matcher (stored as-is) or function matcher (serialised via `toString()`). | +| `length` | Number of entries currently persisted to the index file. | +| `headerSize` | Byte offset of the first index entry in the file. Allows reads to compute exact file positions without re-reading the binary header. | + +Partition entries are the relative file names (relative to `dataDirectory`) used when each +partition was first registered, e.g. `"bench.0"`. + +## Security model + +A single HMAC (SHA-256, keyed with the existing `config.hmacSecret`) is computed over +the entire manifest body (`version + partitions + indexes`). This covers: + +- All matcher values, including serialised function matchers. +- All partition file names. +- The index lengths and header sizes. + +**Why one document-level HMAC is sufficient** + +When the manifest was written, all matchers were already trusted: they came from either +user-supplied code in the same process, or from individual index-file metadata that was +already HMAC-verified on a previous startup. The manifest's document-level HMAC prevents +an attacker who can write to the disk from injecting a malicious function matcher. + +If `hmacSecret` is the empty string (default), the HMAC is still computed and verified; +it just uses an empty key, which gives the same protection level as the existing per-matcher +HMAC approach with an empty secret. + +## Startup decision tree (WritableStorage.open) + +``` +open(callback) +├── LOCK_RECLAIM mode? +│ └── yes → full scan + torn-write repair (manifest ignored) +└── no → try loadManifest() + ├── manifest missing or HMAC invalid? + │ └── full scan (same as LOCK_RECLAIM, minus repair) + └── manifest valid → restoreFromManifest() + ├── register partitions (zero file I/O — creates objects, no open()) + ├── for each index: create WritableIndex via manifestData option + │ (sets opened=true, data length, headerSize — no file open) + ├── register matchers in IndexMatcher + ├── emit 'index-created' for each index (EventStore picks up streams) + ├── set initialized = true + └── openIndexes() → open primary index, emit 'opened' +``` + +## Persistence timing and crash safety + +The manifest is written **only on graceful `close()`**, after all write buffers have +been flushed to disk. This makes it robust by design: + +- **No crash → manifest valid**: next startup takes the fast path. +- **Crash before close**: manifest is missing or reflects the state of the *last* graceful + shutdown. Next startup (with `LOCK_RECLAIM`) falls back to the full scan and + torn-write repair path, just as it did before this feature. On the following graceful + `close()`, a fresh manifest is written. +- **Manifest written then crash**: impossible — the process exits before or after + `close()` completes, never in the middle of `close()` after manifest write but before + other cleanup (the manifest write is the last step). + +Because recovery is not performance-critical and a full scan is expected in that case, +there is no need for transactional or crash-safe manifest writes (e.g. write-then-rename). + +## Performance expectation + +| Startup path | Complexity | Bottleneck | +|---|---|---| +| Manifest (happy path) | O(manifest file size) | JSON parse of ~70 bytes/index | +| Full scan fallback | O(N indexes × file open cost) | per-index file open + header read | + +For 5 000 indexes with simple object matchers the manifest file is roughly 300–400 KB. + +Latest local re-run (Node v24.18.0, same benchmark harness copied into dedicated worktrees for `main`, `PR #339`, and this branch) with **non-empty** secondary indexes: + +- **Scenario A (1 index per partition)** startup improves over `main` at higher counts (`5k: 279.8 → 236.0 ms`, `10k: 540.9 → 453.3 ms`, `20k: 971.7 → 902.5 ms`). +- **Scenario B (100 partitions, growing indexes)** startup is mixed (`100: 8.0 → 7.2 ms`, `1k: 28.4 → 40.8 ms`, `5k: 157.5 → 145.7 ms`). diff --git a/docs/performance.md b/docs/performance.md index d544fc4..8ab1ce5 100644 --- a/docs/performance.md +++ b/docs/performance.md @@ -50,12 +50,11 @@ Read latency is dominated by the per-document partition seek and is largely flat ### Startup time -Startup time scales linearly with the number of files on disk: +On a clean shutdown path, startup uses the persisted manifest: partitions and secondary-index metadata are restored from one manifest read instead of opening every index file to read headers. -- The store performs one `readdir` call over the data directory followed by one `open` per partition. -- Registering secondary indexes (`openIndex` / `ensureIndex`) costs one `open` + header-read per index file. +Secondary indexes are still registered at startup, but their file descriptors are released immediately and reopened lazily on first real index access. This keeps startup fd pressure and heap pressure lower when thousands of indexes exist. -For large deployments consider lazy-loading secondary indexes on first access rather than opening all of them during startup. +When startup enters crash-recovery mode (`LOCK_RECLAIM` after torn-write repair), the manifest fast path is bypassed and the store falls back to the full scan/rebuild path. --- @@ -158,4 +157,3 @@ Interpretation: - When events are **already deserialized**, object matchers are the correct baseline and typically faster. - In workloads where data is still in **raw buffer form** (streaming), raw matchers avoid `JSON.parse` and can outperform a deserialize+object-match pipeline by roughly **~50%** in practice. - diff --git a/src/Index/ReadableIndex.js b/src/Index/ReadableIndex.js index 03778af..b829b69 100644 --- a/src/Index/ReadableIndex.js +++ b/src/Index/ReadableIndex.js @@ -67,6 +67,7 @@ class ReadableIndex extends events.EventEmitter { initialize(options) { /* @type Array */ this.data = []; + this.entryCount = 0; this.opened = false; this.fileMode = 'r'; this.fileHandlePool = options.fileHandlePool || new FileHandlePool(); @@ -79,6 +80,9 @@ class ReadableIndex extends events.EventEmitter { this.metadata = Object.assign({entryClass: options.EntryClass.name, entrySize: options.EntryClass.size}, options.metadata); } this.headerSize = 0; + // When set, open() restores state from this object instead of reading from disk. + // Used by WritableStorage to skip per-index file I/O on manifest-based startup. + this.manifestData = options.manifestData || null; } /** @@ -102,7 +106,7 @@ class ReadableIndex extends events.EventEmitter { * @returns {number} */ get length() { - return this.data.length; + return this.entryCount; } /** @@ -165,11 +169,22 @@ class ReadableIndex extends events.EventEmitter { if (this.opened) { return false; } + + if (this.manifestData) { + this.opened = true; + this.readUntil = -1; + this.headerSize = this.manifestData.headerSize; + this.entryCount = this.manifestData.length; + this.manifestData = null; + return true; + } + this.opened = this.getFileHandle() !== null; this.readUntil = -1; const length = this.readFileLength(); + this.entryCount = length; if (length > 0) { this.data = new Array(length); // Read last item to get the index started @@ -251,6 +266,7 @@ class ReadableIndex extends events.EventEmitter { */ close() { this.data = []; + this.entryCount = 0; this.readUntil = -1; this.readBuffer.fill(0); this.fileHandlePool.evict(this, false); diff --git a/src/Index/WritableIndex.js b/src/Index/WritableIndex.js index 805482a..144a040 100644 --- a/src/Index/WritableIndex.js +++ b/src/Index/WritableIndex.js @@ -118,7 +118,7 @@ class WritableIndex extends ReadableIndex { * @param {number} fd */ onBeforeClose(fd) { - if (!this.writeBuffer) { + if (this.writeBufferCursor === 0) { return; } this.flush(); @@ -190,13 +190,15 @@ class WritableIndex extends ReadableIndex { assertEqual(entry.constructor.name, this.EntryClass.name, `Wrong entry object.`); assertEqual(entry.constructor.size, this.EntryClass.size, `Invalid entry size.`); - const dataLen = this.data.length; - assert(dataLen === 0 || this.data.at(-1).number < entry.number, 'Consistency error. Tried to add an index that should come before existing last entry.'); + const dataLen = this.length; + const lastEntry = dataLen > 0 ? this.get(dataLen) : null; + assert(dataLen === 0 || lastEntry.number < entry.number, 'Consistency error. Tried to add an index that should come before existing last entry.'); if (this.readUntil === dataLen - 1) { this.readUntil++; } this.data[dataLen] = entry; + this.entryCount = dataLen + 1; if (this.writeBufferCursor === 0) { this.flushTimeout = setTimeout(() => this.flush(), this.flushDelay); @@ -232,8 +234,11 @@ class WritableIndex extends ReadableIndex { return; } fs.truncateSync(this.fileName, truncatePosition); - this.data.splice(after); - this.readUntil = Math.min(this.readUntil, after); + if (this.data.length > after) { + this.data.splice(after); + } + this.entryCount = after; + this.readUntil = Math.min(this.readUntil, after - 1); } } diff --git a/src/Storage/ReadableStorage.js b/src/Storage/ReadableStorage.js index 0281823..87f2f33 100644 --- a/src/Storage/ReadableStorage.js +++ b/src/Storage/ReadableStorage.js @@ -170,7 +170,7 @@ class ReadableStorage extends events.EventEmitter { } /** - * Scan partitions and secondary index files; emit 'index-created' for each found index. + * Scan partitions and secondary index files; register each found index. * @param {function} done Called when both scans finish. */ scanFiles(done) { @@ -193,7 +193,7 @@ class ReadableStorage extends events.EventEmitter { const indexPattern = new RegExp(`^${escaped}\\.(.+)\\.index$`); scanForFiles(this.indexDirectory, indexPattern, (name) => { if (!(name in this.secondaryIndexes)) { - this.emit('index-created', name); + this.registerFoundIndex(name); } }, (indexErr) => { // The directory could disappear between existsSync and readdir (e.g. test cleanup). @@ -204,6 +204,16 @@ class ReadableStorage extends events.EventEmitter { }); } + /** + * Register a secondary index discovered while scanning the index directory. + * + * @protected + * @param {string} name + */ + registerFoundIndex(name) { + this.emit('index-created', name); + } + /** * Only the primary index is opened eagerly; secondary indexes open on demand. * @@ -440,10 +450,20 @@ class ReadableStorage extends events.EventEmitter { // 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.afterRegisterSecondaryIndex(index); return index; } + /** + * Called after a secondary index has been opened and registered via openIndex(). + * No-op in the base class; WritableStorage overrides this to check for stale entries. + * + * @protected + * @param {ReadableIndex} index + */ + afterRegisterSecondaryIndex(index) { // eslint-disable-line no-unused-vars + } + /** * Remove a secondary index from the write path and the matcher lookup table. * diff --git a/src/Storage/WritableStorage.js b/src/Storage/WritableStorage.js index 28eac78..69e0edf 100644 --- a/src/Storage/WritableStorage.js +++ b/src/Storage/WritableStorage.js @@ -63,6 +63,7 @@ class WritableStorage extends ReadableStorage { this._lockMode = config.lock; this.partitioner = config.partitioner; this.partitionIds = {}; + this.needsSecondaryIndexTruncationCheck = false; } /** @@ -70,6 +71,11 @@ class WritableStorage extends ReadableStorage { * For LOCK_RECLAIM, removes any orphaned lock before trying to acquire our own; torn-write * repair runs after the primary index is open, before `'opened'` is emitted. * + * On the first open (not in recovery mode), tries to load the startup manifest. + * A valid manifest restores all secondary indexes and partitions without any file I/O, + * making startup time nearly independent of the number of indexes. + * If the manifest is missing or its HMAC does not verify, falls back to a full scanFiles(). + * * @param {function(): void} [callback] Called after indexes open, before `'opened'` is emitted. * Can be used as a synchronous alternative to listening to the `'opened'` event. * @returns {boolean} @@ -82,8 +88,35 @@ class WritableStorage extends ReadableStorage { return true; } + // Happy path: load manifest and skip scanFiles() entirely. + // Only attempted on first open (initialized === null) and when not in recovery. + if (this.initialized === null && !needsRepair) { + const manifest = this.loadManifest(); + if (manifest) { + this.restoreFromManifest(manifest); + this.openIndexes(); + // Use initialized = false (the "scan in progress" state) so that close() + // while waiting for setImmediate resets it to null and we can skip the + // callback — matching the async guard in the scanFiles path. + // setImmediate defers the callback to match scanFiles() async behaviour, + // so that 'opened' and 'ready' always fire after the constructor returns + // and listeners have been registered. + this.initialized = false; + setImmediate(() => { + if (this.initialized === null) return; + this.initialized = true; + callback?.(); + this.emit('opened'); + }); + return true; + } + } + const onOpen = needsRepair - ? () => { this.checkTornWrites(); callback?.(); } + ? () => { + this.needsSecondaryIndexTruncationCheck = this.checkTornWrites(); + callback?.(); + } : callback; return super.open(onOpen); } @@ -153,6 +186,7 @@ class WritableStorage extends ReadableStorage { */ checkTornWrites() { const { lastValidSequenceNumber, maxPartitionSequenceNumber } = this.findTornWriteBoundary(); + let hasPrimaryTruncation = false; if (lastValidSequenceNumber < Number.MAX_SAFE_INTEGER) { // Phase 2: remove all documents at or beyond the torn-write boundary from each partition. @@ -171,12 +205,14 @@ class WritableStorage extends ReadableStorage { // Reindex to fill in any missing complete-document entries. this.reindex(this.index.length); + hasPrimaryTruncation = true; } else if (maxPartitionSequenceNumber >= 0 && maxPartitionSequenceNumber + 1 > this.index.length) { // No torn writes, but the index is lagging — repair it. this.reindex(this.index.length); } this.forEachPartition(partition => partition.close()); + return hasPrimaryTruncation; } /** @@ -255,12 +291,159 @@ class WritableStorage extends ReadableStorage { /** * @inheritDoc * Unlocks the storage, then delegates to the parent close(). + * Saves the startup manifest after flushing, so the next open() can skip scanFiles(). */ close() { if (this.locked) { this.unlock(); } - super.close(); + if (this.initialized === true) { + this.flush(); + const manifestData = this.buildManifestData(); + super.close(); + this.saveManifest(manifestData); + } else { + super.close(); + } + } + + /** + * Path of the startup manifest file. + * @returns {string} + */ + get manifestFile() { + return path.join(this.indexDirectory, '.' + this.storageFile + '.manifest.json'); + } + + /** + * Collect current secondary-index and partition state for manifest persistence. + * Must be called before super.close() resets index data arrays to []. + * + * @private + * @returns {{ partitions: string[], indexes: object }} + */ + buildManifestData() { + const partitions = Object.values(this.partitions).map(p => p.name); + const indexes = {}; + for (const [name, { index, matcher }] of Object.entries(this.secondaryIndexes)) { + indexes[name] = { + matcher: typeof matcher === 'function' ? matcher.toString() : matcher, + length: index.length, + headerSize: index.headerSize + }; + } + return { partitions, indexes }; + } + + /** + * Write the startup manifest to disk. + * A single HMAC over the whole document guards against tampering. + * + * @private + * @param {{ partitions: string[], indexes: object }} data + */ + saveManifest(data) { + const body = JSON.stringify({ version: 1, partitions: data.partitions, indexes: data.indexes }); + const manifest = body.slice(0, -1) + ',"hmac":"' + this.hmac(body) + '"}'; + try { + fs.writeFileSync(this.manifestFile, manifest); + } catch { + // Best-effort: if the write fails, next open falls back to scanFiles(). + } + } + + /** + * Read and verify the startup manifest. + * Returns the parsed manifest if the HMAC is valid; null otherwise. + * + * @private + * @returns {{ version: number, partitions: string[], indexes: object }|null} + */ + loadManifest() { + if (!fs.existsSync(this.manifestFile)) { + return null; + } + try { + const raw = JSON.parse(fs.readFileSync(this.manifestFile, 'utf8')); + if (!raw || raw.version !== 1 || !raw.indexes) { + return null; + } + const { hmac, ...body } = raw; + if (hmac !== this.hmac(JSON.stringify(body))) { + return null; + } + return raw; + } catch { + return null; + } + } + + /** + * Restore secondary indexes and partitions from a verified manifest. + * Skips all per-index file I/O: index state is populated from the manifest directly. + * + * @private + * @param {{ partitions: string[], indexes: object }} manifest + */ + restoreFromManifest(manifest) { + for (const partitionFile of (manifest.partitions || [])) { + this.registerPartitionFile(partitionFile); + } + for (const [name, entry] of Object.entries(manifest.indexes)) { + const { index, matcher } = this.buildIndexFromManifestEntry(name, entry); + this.secondaryIndexes[name] = { index, matcher }; + this.indexMatcher.add(name, matcher); + this.emit('index-created', name); + } + } + + /** + * Register a secondary index discovered during startup scans before notifying listeners. + * + * @protected + * @param {string} name + */ + registerFoundIndex(name) { + this.openIndex(name); + this.emit('index-created', name); + } + + /** + * Reconstruct a WritableIndex from a manifest entry without opening the file. + * The manifestData option lets ReadableIndex.open() restore state from the manifest + * instead of reading from disk. + * + * @private + * @param {string} name Short index name (e.g. 'stream-orders'). + * @param {{ matcher: object|string, length: number, headerSize: number }} entry + * @returns {{ index: WritableIndex, matcher: Matcher }} + */ + buildIndexFromManifestEntry(name, entry) { + const indexName = this.storageFile + '.' + name + '.index'; + const matcher = this.buildMatcherFromManifestEntry(entry); + const options = Object.assign({}, this.indexOptions, { + fileHandlePool: this.indexHandlePool, + manifestData: { length: entry.length, headerSize: entry.headerSize } + }); + const index = new WritableIndex(indexName, options); + return { index, matcher }; + } + + /** + * Reconstruct the matcher from a manifest entry. + * The whole manifest is already HMAC-verified, so function matchers can be eval'd safely. + * + * @private + * @param {{ matcher: object|string }} entry + * @returns {Matcher} + */ + buildMatcherFromManifestEntry(entry) { + if (typeof entry.matcher === 'object') { + return entry.matcher; + } + // eval() is safe here: the whole manifest was HMAC-verified before this method is + // called, so the function source was written by this process and not tampered with. + return eval('(' + entry.matcher + ')').bind({}); // jshint ignore:line } /** @@ -522,18 +705,24 @@ class WritableStorage extends ReadableStorage { /** * @inheritDoc - * Open an existing secondary index and repair any stale entries beyond the current primary - * index length. Stale entries can be present when checkTornWrites() truncated the primary + * Check for stale entries before releasing the file descriptor. + * Stale entries can be present when checkTornWrites() truncated the primary * index before this secondary index was loaded into memory. + * + * @protected + * @param {WritableIndex} index */ - openIndex(name, matcher) { - const index = super.openIndex(name, matcher); + afterRegisterSecondaryIndex(index) { + if (!this.needsSecondaryIndexTruncationCheck) { + super.afterRegisterSecondaryIndex(index); + return; + } const lastEntry = index.lastEntry; if (lastEntry !== false && lastEntry.number > this.index.length) { // Secondary index is ahead of primary: truncate stale entries. index.truncate(index.find(this.index.length)); } - return index; + super.afterRegisterSecondaryIndex(index); } /** diff --git a/test/Storage.spec.js b/test/Storage.spec.js index 1afe448..1a633a1 100644 --- a/test/Storage.spec.js +++ b/test/Storage.spec.js @@ -837,37 +837,6 @@ describe('Storage', function() { expect(index.length).to.be(3); }); - it('repairs stale secondary index when opened after a primary index truncation', function() { - // Simulate the case where checkTornWrites() truncated the primary index before - // secondary indexes were loaded (e.g. after LOCK_RECLAIM on an unclean shutdown). - storage = createStorage(); - storage.open(); - // Write 10 documents and ensure a secondary index (matches even-numbered foo) - const totalDocuments = 10; - const truncateAt = 6; // simulated truncation point - const expectedSecondaryCount = Math.floor(truncateAt / 2); // even numbers in 1..truncateAt - storage.ensureIndex('foobar', (doc) => doc.foo % 2 === 0); - for (let i = 1; i <= totalDocuments; i++) { - storage.write({foo: i}); - } - storage.flush(); - storage.close(); - - // Re-open and truncate primary index directly (simulating checkTornWrites running - // before secondary indexes are loaded) - storage = createStorage(); - storage.open(); - storage.index.truncate(truncateAt); // truncate primary index without going through full truncate() - // Close without touching secondary index on disk - storage.index.flush(); - storage.close(); - - // Re-open: openIndex() should detect the stale secondary index and repair it - storage = createStorage(); - storage.open(); - const repairedIndex = storage.openIndex('foobar'); - expect(repairedIndex.length).to.be(expectedSecondaryCount); // only entries 2,4,6 are still valid - }); }); describe('checkTornWrites', function() { @@ -1779,6 +1748,89 @@ describe('Storage', function() { }); + describe('manifest restore', function() { + + it('registers scanned secondary indexes before emitting index-created', function(done) { + storage = createStorage(); + storage.open(() => { + storage.ensureIndex('foo', { type: 'foo' }); + storage.write({ type: 'foo', value: 1 }); + storage.flush(); + const manifestFile = storage.manifestFile; + storage.close(); + fs.removeSync(manifestFile); + + storage = createStorage(); + let sawIndexCreated = false; + storage.once('index-created', (name) => { + sawIndexCreated = true; + expect(name).to.be('foo'); + expect(storage.secondaryIndexes.foo).to.not.be(undefined); + expect(storage.openIndex('foo')).to.be(storage.secondaryIndexes.foo.index); + }); + storage.open(() => { + expect(sawIndexCreated).to.be(true); + expect(storage.secondaryIndexes.foo).to.not.be(undefined); + expect(storage.secondaryIndexes.foo.index.length).to.be(1); + expect(storage.indexMatcher.matchers.get('foo')).to.eql({ type: 'foo' }); + done(); + }); + }); + }); + + it('keeps secondary index length virtualized until entries are read', function(done) { + storage = createStorage(); + storage.open(() => { + storage.ensureIndex('foo', { type: 'foo' }); + for (let i = 0; i < 5; i++) { + storage.write({ type: 'foo', value: i }); + } + storage.flush(); + storage.close(); + + storage = createStorage(); + storage.open(() => { + const index = storage.openIndex('foo'); + + expect(index.length).to.be(5); + expect(index.data.length).to.be(0); + + const first = index.get(1); + expect(first.number).to.be(1); + expect(index.data.length).to.be(1); + done(); + }); + }); + }); + + it('appends to a manifest-restored secondary index with virtualized length', function(done) { + storage = createStorage(); + storage.open(() => { + storage.ensureIndex('foo', { type: 'foo' }); + for (let i = 0; i < 3; i++) { + storage.write({ type: 'foo', value: i }); + } + storage.flush(); + storage.close(); + + storage = createStorage(); + storage.open(() => { + const index = storage.openIndex('foo'); + expect(index.length).to.be(3); + expect(index.data.length).to.be(0); + + storage.write({ type: 'foo', value: 99 }); + storage.flush(); + + expect(index.length).to.be(4); + expect(index.lastEntry.number).to.be(4); + expect(storage.read(4, index)).to.eql({ type: 'foo', value: 99 }); + done(); + }); + }); + }); + }); + describe('file handle pools', function() { it('defaults index handle limit to 1024', function() {