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

albe / node-event-storage / 23376901430

21 Mar 2026 09:40AM UTC coverage: 97.843% (+0.02%) from 97.826%
23376901430

Pull #255

github

web-flow
Merge 4ecb1526a into 28d4e34f0
Pull Request #255: feat: bound index memory usage with a configurable ring buffer cache

612 of 647 branches covered (94.59%)

Branch coverage included in aggregate %.

143 of 143 new or added lines in 7 files covered. (100.0%)

1 existing line in 1 file now uncovered.

1475 of 1486 relevant lines covered (99.26%)

1133.41 hits per line

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

96.32
/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);
56✔
17
    for (let i = items.length - 1; i >= 0; i--) {
56✔
18
        yield items[i];
496✔
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 = {}) {
404!
47
        super();
740✔
48
        if (typeof storageName !== 'string') {
740✔
49
            config = storageName;
52✔
50
            storageName = undefined;
52✔
51
        }
52

53
        this.storageFile = storageName || 'storage';
740✔
54
        const defaults = {
740✔
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);
740✔
63
        this.serializer = config.serializer;
740✔
64

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

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

69
        this.initializeIndexes(config);
740✔
70
        this.scanPartitions(config);
736✔
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);
740✔
104

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

115
    /**
116
     * The amount of documents in the storage.
117
     * @returns {number}
118
     */
119
    get length() {
120
        return this.index.length;
3,240✔
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 = {
736✔
133
            readBufferSize: DEFAULT_READ_BUFFER_SIZE
134
        };
135
        this.partitionConfig = Object.assign(defaults, config);
736✔
136
        this.partitions = Object.create(null);
736✔
137

138
        const files = fs.readdirSync(this.dataDirectory);
736✔
139
        for (let file of files) {
736✔
140
            if (file.substr(-6) === '.index') continue;
944✔
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();
664✔
159

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

162
        this.emit('opened');
664✔
163
        return true;
664✔
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,160✔
175
        this.forEachSecondaryIndex(index => index.close());
1,160✔
176
        for (let index of Object.values(this.readonlyIndexes)) {
1,160✔
177
            index.close();
44✔
178
        }
179
        this.forEachPartition(partition => partition.close());
1,160✔
180
        this.emit('closed');
1,160✔
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,580✔
195

196
        this.partitions[partitionIdentifier].open();
3,580✔
197
        return this.partitions[partitionIdentifier];
3,580✔
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,540✔
210
        const data = partition.readFrom(position, size);
3,540✔
211
        return this.serializer.deserialize(data);
3,540✔
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 full index or in the provided index.
239
     * Returns a generator in order to reduce memory usage and be able to read lots of documents with little latency.
240
     *
241
     * @api
242
     * @param {number} from The 1-based document number (inclusive) to start reading from.
243
     * @param {number} [until] The 1-based document number (inclusive) to read until. Defaults to index.length.
244
     * @param {ReadableIndex} [index] The index to use for finding the documents in the range.
245
     * @returns {Generator<object>} A generator that will read each document in the range one by one.
246
     */
247
    *readRange(from, until = -1, index = null) {
64✔
248
        index = index || this.index;
460✔
249
        index.open();
460✔
250

251
        const readFrom = wrapAndCheck(from, index.length);
460✔
252
        const readUntil = wrapAndCheck(until, index.length);
460✔
253
        assert(readFrom > 0 && readUntil > 0, `Range scan error for range ${from} - ${until}.`);
460✔
254

255
        if (readFrom > readUntil) {
444✔
256
            const batchSize = 10;
40✔
257
            let batchUntil = readFrom;
40✔
258
            while (batchUntil > readUntil) {
40✔
259
                const batchFrom = Math.max(readUntil, batchUntil - batchSize);
56✔
260
                yield* reverse(this.iterateRange(batchFrom, batchUntil, index));
56✔
261
                batchUntil = batchFrom - 1;
56✔
262
            }
263
            return undefined;
40✔
264
        }
265

266
        yield* this.iterateRange(readFrom, readUntil, index);
404✔
267
    }
268

269
    /**
270
     * Iterate all documents in this storage in range from to until inside the index.
271
     * @private
272
     * @param {number} from
273
     * @param {number} until
274
     * @param {ReadableIndex} index
275
     * @returns {Generator<object>}
276
     */
277
    *iterateRange(from, until, index) {
278
        const entries = index.range(from, until);
460✔
279
        for (let entry of entries) {
460✔
280
            const document = this.readFrom(entry.partition, entry.position, entry.size);
1,636✔
281
            yield document;
1,636✔
282
        }
283
    }
284

285
    /**
286
     * Open an existing readonly index for reading, without registering it in the secondary indexes write path.
287
     * Use this for indexes whose files carry a status marker (e.g. `stream-foo.closed.index`).
288
     *
289
     * @api
290
     * @param {string} name The readonly index name (e.g. 'stream-foo.closed').
291
     * @returns {ReadableIndex}
292
     * @throws {Error} if the readonly index does not exist.
293
     */
294
    openReadonlyIndex(name) {
295
        if (name in this.readonlyIndexes) {
44!
UNCOV
296
            return this.readonlyIndexes[name];
×
297
        }
298
        const indexName = this.storageFile + '.' + name + '.index';
44✔
299
        assert(fs.existsSync(path.join(this.indexDirectory, indexName)), `Index "${name}" does not exist.`);
44✔
300
        const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions));
44✔
301
        index.open();
44✔
302
        this.readonlyIndexes[name] = index;
44✔
303
        return index;
44✔
304
    }
305

306
    /**
307
     * Open an existing index.
308
     *
309
     * @api
310
     * @param {string} name The index name.
311
     * @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.
312
     * @returns {ReadableIndex}
313
     * @throws {Error} if the index with that name does not exist.
314
     * @throws {Error} if the HMAC for the matcher does not match.
315
     */
316
    openIndex(name, matcher) {
317
        if (name === '_all') {
588✔
318
            return this.index;
8✔
319
        }
320
        if (name in this.secondaryIndexes) {
580✔
321
            return this.secondaryIndexes[name].index;
520✔
322
        }
323

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

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

330
        index.open();
44✔
331
        return index;
44✔
332
    }
333

334
    /**
335
     * Helper method to iterate over all documents.
336
     *
337
     * @protected
338
     * @param {function(object, EntryInterface)} iterationHandler
339
     */
340
    forEachDocument(iterationHandler) {
341
        /* istanbul ignore if  */
342
        if (typeof iterationHandler !== 'function') {
724✔
343
            return;
344
        }
345

346
        const entries = this.index.all();
724✔
347

348
        for (let entry of entries) {
724✔
349
            const document = this.readFrom(entry.partition, entry.position, entry.size);
1,728✔
350
            iterationHandler(document, entry);
1,728✔
351
        }
352
    }
353

354
    /**
355
     * Helper method to iterate over all secondary indexes.
356
     *
357
     * @protected
358
     * @param {function(ReadableIndex, string)} iterationHandler
359
     * @param {object} [matchDocument] If supplied, only indexes the document matches on will be iterated.
360
     */
361
    forEachSecondaryIndex(iterationHandler, matchDocument) {
362
        /* istanbul ignore if  */
363
        if (typeof iterationHandler !== 'function') {
4,164✔
364
            return;
365
        }
366

367
        for (let indexName of Object.keys(this.secondaryIndexes)) {
4,164✔
368
            if (!matchDocument || matches(matchDocument, this.secondaryIndexes[indexName].matcher)) {
4,456✔
369
                iterationHandler(this.secondaryIndexes[indexName].index, indexName);
2,220✔
370
            }
371
        }
372
    }
373

374
    /**
375
     * Helper method to iterate over all partitions.
376
     *
377
     * @protected
378
     * @param {function(ReadablePartition)} iterationHandler
379
     */
380
    forEachPartition(iterationHandler) {
381
        /* istanbul ignore if  */
382
        if (typeof iterationHandler !== 'function') {
1,228✔
383
            return;
384
        }
385

386
        for (let partition of Object.keys(this.partitions)) {
1,228✔
387
            iterationHandler(this.partitions[partition]);
1,160✔
388
        }
389
    }
390

391
}
392

393
module.exports = ReadableStorage;
4✔
394
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