diff --git a/AGENTS.md b/AGENTS.md index 69aa7a3..2f08421 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. @@ -60,6 +61,8 @@ EventStore → Storage → Partition (append-only data files) - **`initialized` three-state**: `null` = not started, `false` = scan in progress, `true` = scan done. Re-opens after `close()` are synchronous. - **`open(callback)` hook** — fires after `openIndexes()` and before `'opened'`. Used by `WritableStorage` for torn-write repair. - **LOCK_RECLAIM in `open()`** — orphaned lock removal lives in `WritableStorage.open()`, directly before `lock()`; torn-write repair runs via the `open(callback)` hook. +- **Manifest HMAC verification hashes the raw manifest body text** — split the trailing `"hmac"` property off the raw file bytes with a cheap reverse scan, not regexes or full-object re-serialization, so startup can verify first and parse once. +- **Recovery truncates secondary indexes inside the repair run itself** — do not defer torn-write cleanup to later `openIndex()` calls, or a safe close/reopen can preserve stale secondary tails that were never opened during recovery. - **EventStore `initialize()`** — register `storage.on('index-created', ...)` *before* calling `storage.open()`. - **Watcher singleton key includes watch options** — do not share one `DirectoryWatcher` across different `fs.watch` option sets (notably recursive vs non-recursive), or read-only/file watchers can miss events. - **ReadOnlyStorage watcher filenames may contain leading directories** — normalize watched filenames to the storage-relative segment (starting at `...`) so nested index roots (e.g. `streams/`) and hierarchical stream paths both resolve correctly. @@ -101,7 +104,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..136fd61 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. +On that manifest fast path, secondary indexes are registered from the manifest metadata alone and their files stay unopened until the first real index access. -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..d1d95ac 100644 --- a/src/Storage/WritableStorage.js +++ b/src/Storage/WritableStorage.js @@ -70,6 +70,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 +87,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.checkTornWrites(); + callback?.(); + } : callback; return super.open(onOpen); } @@ -164,7 +196,6 @@ class WritableStorage extends ReadableStorage { // Truncate all indexes to the torn-write boundary. this.index.open(); this.index.truncate(lastValidSequenceNumber); - /* c8 ignore next */ this.forEachWritableSecondaryIndex(index => { index.truncate(index.find(lastValidSequenceNumber)); }); @@ -255,12 +286,224 @@ 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 manifest = fs.readFileSync(this.manifestFile, 'utf8'); + const parts = this.splitManifestBodyAndHmac(manifest); + if (!parts) { + return null; + } + const { body, hmac } = parts; + if (hmac !== this.hmac(body)) { + return null; + } + const raw = JSON.parse(body); + if (!raw || raw.version !== 1 || !raw.indexes) { + return null; + } + return { ...raw, hmac }; + } catch { + return null; + } + } + + /** + * Split the raw manifest text into the original JSON body and trailing HMAC property. + * The property is always written last, so a reverse scan can recover the exact body bytes + * without regexes or re-serializing the parsed object. + * + * @private + * @param {string} manifest + * @returns {{ body: string, hmac: string }|null} + */ + splitManifestBodyAndHmac(manifest) { + let cursor = manifest.length - 1; + while (cursor >= 0 && /\s/.test(manifest[cursor])) cursor--; + if (cursor < 0 || manifest[cursor] !== '}') { + return null; + } + const closingBraceIndex = cursor; + + cursor--; + while (cursor >= 0 && /\s/.test(manifest[cursor])) cursor--; + if (cursor < 0 || manifest[cursor] !== '"') { + return null; + } + const valueEnd = cursor; + + const valueStart = manifest.lastIndexOf('"', valueEnd - 1); + if (valueStart < 0) { + return null; + } + const hmac = manifest.slice(valueStart + 1, valueEnd); + + cursor = valueStart - 1; + while (cursor >= 0 && /\s/.test(manifest[cursor])) cursor--; + if (cursor < 0 || manifest[cursor] !== ':') { + return null; + } + + cursor--; + while (cursor >= 0 && /\s/.test(manifest[cursor])) cursor--; + if (cursor < 0 || manifest[cursor] !== '"') { + return null; + } + const keyEnd = cursor; + + const keyStart = manifest.lastIndexOf('"', keyEnd - 1); + if (keyStart < 0 || manifest.slice(keyStart + 1, keyEnd) !== 'hmac') { + return null; + } + + cursor = keyStart - 1; + while (cursor >= 0 && /\s/.test(manifest[cursor])) cursor--; + if (cursor < 0 || manifest[cursor] !== ',') { + return null; + } + + return { + body: manifest.slice(0, cursor) + manifest.slice(closingBraceIndex, closingBraceIndex + 1), + hmac + }; + } + + /** + * 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 } /** @@ -520,22 +763,6 @@ 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 - * index before this secondary index was loaded into memory. - */ - openIndex(name, matcher) { - const index = super.openIndex(name, matcher); - 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; - } - /** * @protected * @param {string} name diff --git a/test/Storage.spec.js b/test/Storage.spec.js index 1afe448..05f4aa2 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() { @@ -1164,6 +1133,43 @@ describe('Storage', function() { crashedStorage.locked = false; }); + it('repairs secondary indexes during LOCK_RECLAIM recovery', function(done) { + const crashedStorage = new Storage({ dataDirectory }); + refs.push(crashedStorage); + crashedStorage.open(); + crashedStorage.ensureIndex('all-docs', { foo: { $gte: 1 } }); + for (let i = 1; i <= 5; i++) { + crashedStorage.write({ foo: i }); + } + crashedStorage.flush(); + + const partitionName = path.join(dataDirectory, 'storage'); + const fd = fs.openSync(partitionName, 'r+'); + const stat = fs.fstatSync(fd); + fs.ftruncateSync(fd, stat.size - 8); + fs.closeSync(fd); + + crashedStorage.index.close(); + crashedStorage.forEachSecondaryIndex(index => index.close()); + crashedStorage.forEachPartition(partition => partition.close()); + + const repairedStorage = new Storage({ dataDirectory, lock: LOCK_RECLAIM }); + refs.push(repairedStorage); + repairedStorage.once('opened', () => { + expect(repairedStorage.secondaryIndexes['all-docs'].index.length).to.be(4); + repairedStorage.close(); + + storage = createStorage(); + storage.once('opened', () => { + expect(storage.openIndex('all-docs').length).to.be(4); + done(); + }); + storage.open(); + }); + repairedStorage.open(); + crashedStorage.locked = false; + }); + }); describe('matches', function() { @@ -1779,6 +1785,117 @@ 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('accepts a valid reformatted manifest without falling back to scanFiles', function(done) { + storage = createStorage(); + storage.open(() => { + storage.ensureIndex('foo', { type: 'foo' }); + storage.write({ type: 'foo', value: 1 }); + storage.flush(); + storage.close(); + + const manifest = JSON.parse(fs.readFileSync(storage.manifestFile, 'utf8')); + const body = JSON.stringify({ + version: manifest.version, + partitions: manifest.partitions, + indexes: manifest.indexes + }, null, 2); + const rewrittenManifest = body.slice(0, -1) + ',\n "hmac": "' + storage.hmac(body) + '"\n}'; + fs.writeFileSync(storage.manifestFile, rewrittenManifest); + + storage = createStorage(); + storage.scanFiles = () => { + throw new Error('scanFiles should not run when the manifest HMAC is valid.'); + }; + storage.open(() => { + expect(storage.secondaryIndexes.foo.index.length).to.be(1); + 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() {