• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

albe / node-event-storage / 23376498872

21 Mar 2026 09:14AM UTC coverage: 97.698% (-0.1%) from 97.826%
23376498872

Pull #254

github

web-flow
Merge 81e06bf3f into 28d4e34f0
Pull Request #254: iterateRange: use k-way partition merge by sequenceNumber when no index specified

615 of 652 branches covered (94.33%)

Branch coverage included in aggregate %.

41 of 42 new or added lines in 2 files covered. (97.62%)

4 existing lines in 2 files now uncovered.

1465 of 1477 relevant lines covered (99.19%)

1145.29 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.07
/src/Storage/ReadableStorage.js
1
const fs = require('fs');
4✔
2
const path = require('path');
4✔
3
const events = require('events');
4✔
4
const Partition = require('../Partition');
4✔
5
const Index = require('../Index');
4✔
6
const { assert, createHmac, matches, wrapAndCheck, buildMetadataForMatcher } = require('../util');
4✔
7

8
const DEFAULT_READ_BUFFER_SIZE = 4 * 1024;
4✔
9

10
/**
11
 * Reverses the items of an iterable
12
 * @param {Generator|Iterable} iterator
13
 * @returns {Generator<*>}
14
 */
15
function *reverse(iterator) {
16
    const items = Array.from(iterator);
60✔
17
    for (let i = items.length - 1; i >= 0; i--) {
60✔
18
        yield items[i];
532✔
19
    }
20
}
21

22
/**
23
 * @typedef {object|function(object):boolean} Matcher
24
 */
25

26
/**
27
 * An append-only storage with highly performant positional range scans.
28
 * It's highly optimized for an event-store and hence does not support compaction or data-rewrite, nor any querying
29
 */
30
class ReadableStorage extends events.EventEmitter {
31

32
    /**
33
     * @param {string} [storageName] The name of the storage.
34
     * @param {object} [config] An object with storage parameters.
35
     * @param {object} [config.serializer] A serializer object with methods serialize(document) and deserialize(data).
36
     * @param {function(object): string} config.serializer.serialize Default is JSON.stringify.
37
     * @param {function(string): object} config.serializer.deserialize Default is JSON.parse.
38
     * @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
39
     * @param {string} [config.indexDirectory] The path where the indexes should be stored. Defaults to dataDirectory.
40
     * @param {string} [config.indexFile] The name of the primary index. Default '{storageName}.index'.
41
     * @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
42
     * @param {object} [config.indexOptions] An options object that should be passed to all indexes on construction.
43
     * @param {string} [config.hmacSecret] A private key that is used to verify matchers retrieved from indexes.
44
     * @param {object} [config.metadata] A metadata object to be stored in all partitions belonging to this storage.
45
     */
46
    constructor(storageName = 'storage', config = {}) {
416!
47
        super();
752✔
48
        if (typeof storageName !== 'string') {
752✔
49
            config = storageName;
52✔
50
            storageName = undefined;
52✔
51
        }
52

53
        this.storageFile = storageName || 'storage';
752✔
54
        const defaults = {
752✔
55
            serializer: { serialize: JSON.stringify, deserialize: JSON.parse },
56
            dataDirectory: '.',
57
            indexFile: this.storageFile + '.index',
58
            indexOptions: {},
59
            hmacSecret: '',
60
            metadata: {}
61
        };
62
        config = Object.assign(defaults, config);
752✔
63
        this.serializer = config.serializer;
752✔
64

65
        this.hmac = createHmac(config.hmacSecret);
752✔
66

67
        this.dataDirectory = path.resolve(config.dataDirectory);
752✔
68

69
        this.initializeIndexes(config);
752✔
70
        this.scanPartitions(config);
748✔
71
    }
72

73
    /**
74
     * @protected
75
     * @param {string} name
76
     * @param {object} [options]
77
     * @returns {{ index: ReadableIndex, matcher?: Matcher }}
78
     */
79
    createIndex(name, options = {}) {
×
80
        /** @type ReadableIndex */
81
        const index = new Index.ReadOnly(name, options);
112✔
82
        return { index };
108✔
83
    }
84

85
    /**
86
     * @protected
87
     * @param {string} name
88
     * @param {object} [options]
89
     * @returns {ReadablePartition}
90
     */
91
    createPartition(name, options = {}) {
×
92
        return new Partition.ReadOnly(name, options);
40✔
93
    }
94

95
    /**
96
     * Create/open the primary index and build the base configuration for all secondary indexes.
97
     *
98
     * @private
99
     * @param {object} config The configuration object
100
     * @returns void
101
     */
102
    initializeIndexes(config) {
103
        this.indexDirectory = path.resolve(config.indexDirectory || this.dataDirectory);
752✔
104

105
        this.indexOptions = config.indexOptions;
752✔
106
        this.indexOptions.dataDirectory = this.indexDirectory;
752✔
107
        // Safety precaution to prevent accidentally restricting main index
108
        delete this.indexOptions.matcher;
752✔
109
        const { index } = this.createIndex(config.indexFile, this.indexOptions);
752✔
110
        this.index = index;
748✔
111
        this.secondaryIndexes = {};
748✔
112
        this.readonlyIndexes = {};
748✔
113
    }
114

115
    /**
116
     * The amount of documents in the storage.
117
     * @returns {number}
118
     */
119
    get length() {
120
        return this.index.length;
3,348✔
121
    }
122

123
    /**
124
     * Scan the data directory for all existing partitions.
125
     * Every file beginning with the storageFile name is considered a partition.
126
     *
127
     * @private
128
     * @param {object} config The configuration object containing options for the partitions.
129
     * @returns void
130
     */
131
    scanPartitions(config) {
132
        const defaults = {
748✔
133
            readBufferSize: DEFAULT_READ_BUFFER_SIZE
134
        };
135
        this.partitionConfig = Object.assign(defaults, config);
748✔
136
        this.partitions = Object.create(null);
748✔
137

138
        const files = fs.readdirSync(this.dataDirectory);
748✔
139
        for (let file of files) {
748✔
140
            if (file.substr(-6) === '.index') continue;
956✔
141
            if (file.substr(-7) === '.branch') continue;
440!
142
            if (file.substr(-5) === '.lock') continue;
440✔
143
            if (file.substr(0, this.storageFile.length) !== this.storageFile) continue;
356✔
144

145
            const partition = this.createPartition(file, this.partitionConfig);
76✔
146
            this.partitions[partition.id] = partition;
76✔
147
        }
148
    }
149

150
    /**
151
     * Open the storage and indexes and create read and write buffers eagerly.
152
     * Will emit an 'opened' event if finished.
153
     *
154
     * @api
155
     * @returns {boolean}
156
     */
157
    open() {
158
        this.index.open();
688✔
159

160
        this.forEachSecondaryIndex(index => index.open());
688✔
161

162
        this.emit('opened');
688✔
163
        return true;
688✔
164
    }
165

166
    /**
167
     * Close the storage and frees up all resources.
168
     * Will emit a 'closed' event when finished.
169
     *
170
     * @api
171
     * @returns void
172
     */
173
    close() {
174
        this.index.close();
1,196✔
175
        this.forEachSecondaryIndex(index => index.close());
1,196✔
176
        for (let index of Object.values(this.readonlyIndexes)) {
1,196✔
177
            index.close();
44✔
178
        }
179
        this.forEachPartition(partition => partition.close());
1,208✔
180
        this.emit('closed');
1,196✔
181
    }
182

183
    /**
184
     * Get a partition either by name or by id.
185
     * If a partition with the given name does not exist, a new one will be created.
186
     * If a partition with the given id does not exist, an error is thrown.
187
     *
188
     * @protected
189
     * @param {string|number} partitionIdentifier The partition name or the partition Id
190
     * @returns {ReadablePartition}
191
     * @throws {Error} If an id is given and no such partition exists.
192
     */
193
    getPartition(partitionIdentifier) {
194
        assert(partitionIdentifier in this.partitions, `Partition #${partitionIdentifier} does not exist.`);
3,356✔
195

196
        this.partitions[partitionIdentifier].open();
3,356✔
197
        return this.partitions[partitionIdentifier];
3,356✔
198
    }
199

200
    /**
201
     * @protected
202
     * @param {number} partitionId The partition to read from.
203
     * @param {number} position The file position to read from.
204
     * @param {number} [size] The expected byte size of the document at the given position.
205
     * @returns {object} The document stored at the given position.
206
     * @throws {Error} if the document at the given position can not be deserialized.
207
     */
208
    readFrom(partitionId, position, size) {
209
        const partition = this.getPartition(partitionId);
3,316✔
210
        const data = partition.readFrom(position, size);
3,316✔
211
        return this.serializer.deserialize(data);
3,316✔
212
    }
213

214
    /**
215
     * Read a single document from the given position, in the full index or in the provided index.
216
     *
217
     * @api
218
     * @param {number} number The 1-based document number (inside the given index) to read.
219
     * @param {ReadableIndex} [index] The index to use for finding the document position.
220
     * @returns {object} The document at the given position inside the index.
221
     */
222
    read(number, index) {
223
        index = index || this.index;
168✔
224

225
        if (!index.isOpen()) {
168✔
226
            index.open();
4✔
227
        }
228

229
        const entry = index.get(number);
168✔
230
        if (entry === false) {
168✔
231
            return false;
4✔
232
        }
233

234
        return this.readFrom(entry.partition, entry.position, entry.size);
164✔
235
    }
236

237
    /**
238
     * Read a range of documents from the given position range, in the provided index or across all partitions.
239
     * When no index is given, iterates all partitions directly and returns documents ordered by their
240
     * sequenceNumber, allowing reconstruction of insertion order without a global index.
241
     * Returns a generator in order to reduce memory usage and be able to read lots of documents with little latency.
242
     *
243
     * @api
244
     * @param {number} from The 1-based document number (inclusive) to start reading from.
245
     * @param {number} [until] The 1-based document number (inclusive) to read until. Defaults to index.length.
246
     * @param {ReadableIndex} [index] The index to use for finding the documents in the range. If not given, iterates all partitions ordered by sequenceNumber.
247
     * @returns {Generator<object>} A generator that will read each document in the range one by one.
248
     */
249
    *readRange(from, until = -1, index = null) {
80✔
250
        const lengthSource = index || this.index;
472✔
251
        if (!lengthSource.isOpen()) {
472✔
252
            lengthSource.open();
4✔
253
        }
254

255
        const readFrom = wrapAndCheck(from, lengthSource.length);
472✔
256
        const readUntil = wrapAndCheck(until, lengthSource.length);
472✔
257
        assert(readFrom > 0 && readUntil > 0, `Range scan error for range ${from} - ${until}.`);
472✔
258

259
        if (readFrom > readUntil) {
456✔
260
            const batchSize = 10;
44✔
261
            let batchUntil = readFrom;
44✔
262
            while (batchUntil > readUntil) {
44✔
263
                const batchFrom = Math.max(readUntil, batchUntil - batchSize);
60✔
264
                yield* reverse(this.iterateRange(batchFrom, batchUntil, index));
60✔
265
                batchUntil = batchFrom - 1;
60✔
266
            }
267
            return undefined;
44✔
268
        }
269

270
        yield* this.iterateRange(readFrom, readUntil, index);
412✔
271
    }
272

273
    /**
274
     * Iterate all documents in this storage in range from to until.
275
     * If an index is provided, uses it to look up document positions; otherwise iterates all partitions
276
     * directly and merges them in sequenceNumber order.
277
     * @private
278
     * @param {number} from
279
     * @param {number} until
280
     * @param {ReadableIndex|null} index
281
     * @returns {Generator<object>}
282
     */
283
    *iterateRange(from, until, index) {
284
        if (index !== null) {
472✔
285
            const entries = index.range(from, until);
428✔
286
            for (let entry of entries) {
428✔
287
                const document = this.readFrom(entry.partition, entry.position, entry.size);
1,412✔
288
                yield document;
1,412✔
289
            }
290
            return;
420✔
291
        }
292

293
        // No index: iterate all partitions and merge documents by sequenceNumber.
294
        // Document header sequenceNumber is 0-based; from/until are 1-based index positions.
295
        yield* this.iteratePartitionsBySequenceNumber(from - 1, until - 1);
44✔
296
    }
297

298
    /**
299
     * Iterate documents across all partitions in sequenceNumber order using a k-way merge.
300
     * SequenceNumbers stored in document headers are 0-based.
301
     * @private
302
     * @param {number} fromSeq The 0-based sequenceNumber to start from (inclusive).
303
     * @param {number} untilSeq The 0-based sequenceNumber to read until (inclusive).
304
     * @returns {Generator<object>}
305
     */
306
    *iteratePartitionsBySequenceNumber(fromSeq, untilSeq) {
307
        const iterators = [];
44✔
308

309
        for (const partition of Object.values(this.partitions)) {
44✔
310
            if (!partition.isOpen()) {
76✔
311
                partition.open();
60✔
312
            }
313
            const gen = partition.readAllWithHeaders();
76✔
314

315
            // Advance to the first document with sequenceNumber >= fromSeq
316
            let result = gen.next();
76✔
317
            while (!result.done && result.value.sequenceNumber < fromSeq) {
76✔
318
                result = gen.next();
80✔
319
            }
320

321
            if (!result.done && result.value.sequenceNumber <= untilSeq) {
76✔
322
                iterators.push({ gen, current: result.value });
72✔
323
            }
324
        }
325

326
        // K-way merge: at each step, yield the document with the smallest sequenceNumber
327
        while (iterators.length > 0) {
44✔
328
            let minIdx = 0;
316✔
329
            for (let i = 1; i < iterators.length; i++) {
316✔
330
                if (iterators[i].current.sequenceNumber < iterators[minIdx].current.sequenceNumber) {
164✔
331
                    minIdx = i;
72✔
332
                }
333
            }
334

335
            yield this.serializer.deserialize(iterators[minIdx].current.data);
316✔
336

337
            const next = iterators[minIdx].gen.next();
316✔
338
            if (!next.done && next.value.sequenceNumber <= untilSeq) {
316✔
339
                iterators[minIdx].current = next.value;
244✔
340
            } else {
341
                iterators.splice(minIdx, 1);
72✔
342
            }
343
        }
344
    }
345

346
    /**
347
     * Open an existing readonly index for reading, without registering it in the secondary indexes write path.
348
     * Use this for indexes whose files carry a status marker (e.g. `stream-foo.closed.index`).
349
     *
350
     * @api
351
     * @param {string} name The readonly index name (e.g. 'stream-foo.closed').
352
     * @returns {ReadableIndex}
353
     * @throws {Error} if the readonly index does not exist.
354
     */
355
    openReadonlyIndex(name) {
356
        if (name in this.readonlyIndexes) {
44!
UNCOV
357
            return this.readonlyIndexes[name];
×
358
        }
359
        const indexName = this.storageFile + '.' + name + '.index';
44✔
360
        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
44✔
361
        const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions));
44✔
362
        index.open();
44✔
363
        this.readonlyIndexes[name] = index;
44✔
364
        return index;
44✔
365
    }
366

367
    /**
368
     * Open an existing index.
369
     *
370
     * @api
371
     * @param {string} name The index name.
372
     * @param {Matcher} [matcher] The matcher object or function that the index needs to have been defined with. If not given it will not be validated.
373
     * @returns {ReadableIndex}
374
     * @throws {Error} if the index with that name does not exist.
375
     * @throws {Error} if the HMAC for the matcher does not match.
376
     */
377
    openIndex(name, matcher) {
378
        if (name === '_all') {
588✔
379
            return this.index;
8✔
380
        }
381
        if (name in this.secondaryIndexes) {
580✔
382
            return this.secondaryIndexes[name].index;
520✔
383
        }
384

385
        const indexName = this.storageFile + '.' + name + '.index';
60✔
386
        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
60✔
387

388
        const metadata = buildMetadataForMatcher(matcher, this.hmac);
56✔
389
        let { index } = this.secondaryIndexes[name] = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata }));
56✔
390

391
        index.open();
44✔
392
        return index;
44✔
393
    }
394

395
    /**
396
     * Helper method to iterate over all documents.
397
     *
398
     * @protected
399
     * @param {function(object, EntryInterface)} iterationHandler
400
     */
401
    forEachDocument(iterationHandler) {
402
        /* istanbul ignore if  */
403
        if (typeof iterationHandler !== 'function') {
724✔
404
            return;
405
        }
406

407
        const entries = this.index.all();
724✔
408

409
        for (let entry of entries) {
724✔
410
            const document = this.readFrom(entry.partition, entry.position, entry.size);
1,728✔
411
            iterationHandler(document, entry);
1,728✔
412
        }
413
    }
414

415
    /**
416
     * Helper method to iterate over all secondary indexes.
417
     *
418
     * @protected
419
     * @param {function(ReadableIndex, string)} iterationHandler
420
     * @param {object} [matchDocument] If supplied, only indexes the document matches on will be iterated.
421
     */
422
    forEachSecondaryIndex(iterationHandler, matchDocument) {
423
        /* istanbul ignore if  */
424
        if (typeof iterationHandler !== 'function') {
4,332✔
425
            return;
426
        }
427

428
        for (let indexName of Object.keys(this.secondaryIndexes)) {
4,332✔
429
            if (!matchDocument || matches(matchDocument, this.secondaryIndexes[indexName].matcher)) {
4,456✔
430
                iterationHandler(this.secondaryIndexes[indexName].index, indexName);
2,220✔
431
            }
432
        }
433
    }
434

435
    /**
436
     * Helper method to iterate over all partitions.
437
     *
438
     * @protected
439
     * @param {function(ReadablePartition)} iterationHandler
440
     */
441
    forEachPartition(iterationHandler) {
442
        /* istanbul ignore if  */
443
        if (typeof iterationHandler !== 'function') {
1,264✔
444
            return;
445
        }
446

447
        for (let partition of Object.keys(this.partitions)) {
1,264✔
448
            iterationHandler(this.partitions[partition]);
1,268✔
449
        }
450
    }
451

452
}
453

454
module.exports = ReadableStorage;
4✔
455
module.exports.matches = matches;
4✔
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc