23367085163

Committed 20 Mar 2026 11:47PM UTC coverage: 97.869% (+0.04%) from 97.826%

Build # 23367085163

Build Type

Pull #107

github

Committed by

web-flow

Commit Message

Merge c83ae3b6d into 28d4e34f0

Pull Request Pull Request #107: Start implementing auto-repair

Coverage Stats

613 of 646 branches covered (94.89%)

Branch coverage included in aggregate %.

25 of 27 new or added lines in 5 files covered. (92.59%)

6 existing lines in 2 files now uncovered.

1454 of 1466 relevant lines covered (99.18%)

1237.75 hits per line

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

97.29

/src/Storage/WritableStorage.js

const fs = require('fs');
const path = require('path');
const WritablePartition = require('../Partition/WritablePartition');
const WritableIndex = require('../Index/WritableIndex');
const ReadableStorage = require('./ReadableStorage');
const { assert, matches, buildMetadataForMatcher, buildMatcherFromMetadata, ensureDirectory } = require('../util');

const DEFAULT_WRITE_BUFFER_SIZE = 16 * 1024;

const LOCK_RECLAIM = 0x1;
const LOCK_THROW = 0x2;

class StorageLockedError extends Error {}

/**
 * @typedef {object|function(object):boolean} Matcher
 */

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

    /**
     * @param {string} [storageName] The name of the storage.
     * @param {object} [config] An object with storage parameters.
     * @param {object} [config.serializer] A serializer object with methods serialize(document) and deserialize(data).
     * @param {function(object): string} config.serializer.serialize Default is JSON.stringify.
     * @param {function(string): object} config.serializer.deserialize Default is JSON.parse.
     * @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
     * @param {string} [config.indexDirectory] The path where the indexes should be stored. Defaults to dataDirectory.
     * @param {string} [config.indexFile] The name of the primary index. Default '{storageName}.index'.
     * @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
     * @param {number} [config.writeBufferSize] Size of the write buffer in bytes. Default 16384.
     * @param {number} [config.maxWriteBufferDocuments] How many documents to have in the write buffer at max. 0 means as much as possible. Default 0.
     * @param {boolean} [config.syncOnFlush] If fsync should be called on write buffer flush. Set this if you need strict durability. Defaults to false.
     * @param {boolean} [config.dirtyReads] If dirty reads should be allowed. This means that writes that are in write buffer but not yet flushed can be read. Defaults to true.
     * @param {function(object, number): string} [config.partitioner] A function that takes a document and sequence number and returns a partition name that the document should be stored in. Defaults to write all documents to the primary partition.
     * @param {object} [config.indexOptions] An options object that should be passed to all indexes on construction.
     * @param {string} [config.hmacSecret] A private key that is used to verify matchers retrieved from indexes.
     * @param {number} [config.lock] One of LOCK_* constants that defines how an existing lock should be handled.
     */
    constructor(storageName = 'storage', config = {}) {
        if (typeof storageName !== 'string') {
            config = storageName;
            storageName = undefined;
        }
        const defaults = {
            partitioner: (document, number) => '',
            writeBufferSize: DEFAULT_WRITE_BUFFER_SIZE,
            maxWriteBufferDocuments: 0,
            syncOnFlush: false,
            dirtyReads: true,
            dataDirectory: '.'
        };
        config = Object.assign(defaults, config);
        config.indexOptions = Object.assign({ syncOnFlush: config.syncOnFlush }, config.indexOptions);
        ensureDirectory(config.dataDirectory);
        super(storageName, config);

        this.lockFile = path.resolve(this.dataDirectory, this.storageFile + '.lock');
        if (config.lock === LOCK_RECLAIM) {
            this.unlock();
        }
        this.partitioner = config.partitioner;
    }

    /**
     * @inheritDoc
     * @returns {boolean}
     * @throws {StorageLockedError} If this storage is locked by another process.
     */
    open() {
        if (!this.lock()) {
            return true;
        }
        return super.open();
    }

    /**
     * Check all partitions torn writes and truncate the storage to the position before the first torn write.
     * This might delete correctly written events in partitions, if their sequence number is higher than the
     * torn write in another partition.
     */
    checkTornWrites() {
        let lastValidSequenceNumber = Number.MAX_SAFE_INTEGER;
        this.forEachPartition(partition => {
            partition.open();
            const tornSequenceNumber = partition.checkTornWrite();
            if (tornSequenceNumber >= 0) {
                lastValidSequenceNumber = Math.min(lastValidSequenceNumber, tornSequenceNumber);
            }
        });
        if (lastValidSequenceNumber < Number.MAX_SAFE_INTEGER) {
            this.truncate(lastValidSequenceNumber);
        }
        this.forEachPartition(partition => partition.close());
    }

    /**
     * Attempt to lock this storage by means of a lock directory.
     * @returns {boolean} True if the lock was created or false if the lock is already in place.
     * @throws {StorageLockedError} If this storage is already locked by another process.
     * @throws {Error} If the lock could not be created.
     */
    lock() {
        if (this.locked) {
            return false;
        }
        try {
            fs.mkdirSync(this.lockFile);
            this.locked = true;
        } catch (e) {
            /* istanbul ignore if */
            if (e.code !== 'EEXIST') {
                throw new Error(`Error creating lock for storage ${this.storageFile}: ` + e.message);
            }
            throw new StorageLockedError(`Storage ${this.storageFile} is locked by another process`);
        }
        return true;
    }

    /**
     * Unlock this storage, no matter if it was previously locked by this writer.
     * Only use this if you are sure there is no other process still having a writer open.
     * Current implementation just deletes a lock file that is named like the storage.
     */
    unlock() {
        if (fs.existsSync(this.lockFile)) {
            if (!this.locked) {
                this.checkTornWrites();
            }
            fs.rmdirSync(this.lockFile);
        }
        this.locked = false;
    }

    /**
     * @inheritDoc
     */
    close() {
        if (this.locked) {
            this.unlock();
        }
        super.close();
    }

    /**
     * Add an index entry for the given document at the position and size.
     *
     * @private
     * @param {number} partitionId The partition where the document is stored.
     * @param {number} position The file offset where the document is stored.
     * @param {number} size The size of the stored document.
     * @param {object} document The document to add to the index.
     * @param {function} [callback] The callback to call when the index is written to disk.
     * @returns {EntryInterface} The index entry item.
     */
    addIndex(partitionId, position, size, document, callback) {
        if (!this.index.isOpen()) {
            this.index.open();
        }

        /*if (this.index.lastEntry.position + this.index.lastEntry.size !== position) {
         this.emit('index-corrupted');
         throw new Error('Corrupted index, needs to be rebuilt!');
         }*/

        const entry = new WritableIndex.Entry(this.index.length + 1, position, size, partitionId);
        this.index.add(entry, (indexPosition) => {
            this.emit('wrote', document, entry, indexPosition);
            /* istanbul ignore if  */
            if (typeof callback === 'function') {
                return callback(indexPosition);
            }
        });
        return entry;
    }

    /**
     * Get a partition either by name or by id.
     * If a partition with the given name does not exist, a new one will be created.
     * If a partition with the given id does not exist, an error is thrown.
     *
     * @protected
     * @param {string|number} partitionIdentifier The partition name or the partition Id
     * @returns {ReadablePartition}
     * @throws {Error} If an id is given and no such partition exists.
     */
    getPartition(partitionIdentifier) {
        if (typeof partitionIdentifier === 'string') {
            const partitionName = this.storageFile + (partitionIdentifier.length ? '.' + partitionIdentifier : '');
            partitionIdentifier = WritablePartition.idFor(partitionName);
            if (!this.partitions[partitionIdentifier]) {
                this.partitions[partitionIdentifier] = this.createPartition(partitionName, this.partitionConfig);
                this.emit('partition-created', partitionIdentifier);
            }
            this.partitions[partitionIdentifier].open();
            return this.partitions[partitionIdentifier];
        }
        return super.getPartition(partitionIdentifier);
    }

    /**
     * @api
     * @param {object} document The document to write to storage.
     * @param {function} [callback] A function that will be called when the document is written to disk.
     * @returns {number} The 1-based document sequence number in the storage.
     */
    write(document, callback) {
        const data = this.serializer.serialize(document).toString();
        const dataSize = Buffer.byteLength(data, 'utf8');

        const partitionName = this.partitioner(document, this.index.length + 1);
        const partition = this.getPartition(partitionName);
        const position = partition.write(data, this.length, callback);

        assert(position !== false, 'Error writing document.');

        const indexEntry = this.addIndex(partition.id, position, dataSize, document);
        this.forEachSecondaryIndex((index, name) => {
            if (!index.isOpen()) {
                index.open();
            }
            index.add(indexEntry);
            this.emit('index-add', name, index.length, document);
        }, document);

        return this.index.length;
    }

    /**
     * Ensure that an index with the given name and document matcher exists.
     * Will create the index if it doesn't exist, otherwise return the existing index.
     *
     * @api
     * @param {string} name The index name.
     * @param {Matcher} [matcher] An object that describes the document properties that need to match to add it this index or a function that receives a document and returns true if the document should be indexed.
     * @returns {ReadableIndex} The index containing all documents that match the query.
     * @throws {Error} if the index doesn't exist yet and no matcher was specified.
     */
    ensureIndex(name, matcher) {
        if (name === '_all') {
            return this.index;
        }
        if (name in this.secondaryIndexes) {
            return this.secondaryIndexes[name].index;
        }

        const indexName = this.storageFile + '.' + name + '.index';
        if (fs.existsSync(path.join(this.indexDirectory, indexName))) {
            return this.openIndex(name, matcher);
        }

        assert((typeof matcher === 'object' || typeof matcher === 'function') && matcher !== null, 'Need to specify a matcher.');

        const metadata = buildMetadataForMatcher(matcher, this.hmac);
        const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata }));
        try {
            this.forEachDocument((document, indexEntry) => {
                if (matches(document, matcher)) {
                    index.add(indexEntry);
                }
            });
        } catch (e) {
            index.destroy();
            throw e;
        }

        this.secondaryIndexes[name] = { index, matcher };
        this.emit('index-created', name);
        return index;
    }

    /**
     * Flush all write buffers to disk.
     * This is a sync method and will invoke all previously registered flush callbacks.
     *
     * @api
     * @returns {boolean} Returns true if a flush on any partition or the main index was executed.
     */
    flush() {
        let result = this.index.flush();
        this.forEachPartition(partition => result = result | partition.flush());
        this.forEachSecondaryIndex(index => index.flush());
        return result;
    }

    /**
     * Iterate all distinct partitions in which the given iterable list of entries are stored.
     * @param {Iterable<Index.Entry>} entries
     * @param {function(Index.Entry)} iterationHandler
     */
    forEachDistinctPartitionOf(entries, iterationHandler) {
        const partitions = [];
        const numPartitions = Object.keys(this.partitions).length;
        for (let entry of entries) {
            if (partitions.indexOf(entry.partition) >= 0) {
                continue;
            }
            partitions.push(entry.partition);
            iterationHandler(entry);
            if (partitions.length === numPartitions) {
                break;
            }
        }
    }

    /**
     * Truncate all partitions after the given (global) sequence number.
     *
     * @private
     * @param {number} after The document sequence number to truncate after.
     */
    truncatePartitions(after) {
        if (after === 0) {
            this.forEachPartition(partition => partition.truncate(0));
            return;
        }

        const entries = this.index.range(after + 1);  // We need the first entry that is cut off
        if (entries === false || entries.length === 0) {
            return;
        }

        this.forEachDistinctPartitionOf(entries, entry => this.getPartition(entry.partition).truncate(entry.position));
    }

    /**
     * Truncate the storage after the given sequence number.
     *
     * @param {number} after The document sequence number to truncate after.
     */
    truncate(after) {
        /*
         To truncate the store following steps need to be done:

         1) find all partition positions after which their files should be truncated
         2) truncate all partitions accordingly
         3) truncate/rewrite all indexes
         */
        if (!this.index.isOpen()) {
            this.index.open();
        }
        if (after < 0) {
            after += this.index.length;
        }

        this.truncatePartitions(after);

        this.index.truncate(after);
        this.forEachSecondaryIndex(index => {
            /* istanbul ignore if */
            if (!(index instanceof WritableIndex)) {
                return;
            }
            let closeIndex = false;
            if (!index.isOpen()) {
                index.open();
                closeIndex = true;
            }
            index.truncate(index.find(after));
            if (closeIndex) {
                index.close();
            }
        });
    }

    /**
     * @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) {
            index.truncate(index.find(this.index.length));
        }
        return index;
    }

    /**
     * @protected
     * @param {string} name
     * @param {object} [options]
     * @returns {{ index: WritableIndex, matcher: Matcher }}
     */
    createIndex(name, options = {}) {
        const index = new WritableIndex(name, options);
        let matcher;

        // If the index contains a matcher (possibly a serialized function) we check HMAC
        // to prevent evaluating unknown code.
        if (index.metadata.matcher) {
            try {
                matcher = buildMatcherFromMetadata(index.metadata, this.hmac);
            } catch (e) {
                index.destroy();
                throw e;
            }
        }

        return { index, matcher };
    }

    /**
     * @protected
     * @param {string} name
     * @param {object} [config]
     * @returns {WritablePartition}
     */
    createPartition(name, config = {}) {
        return new WritablePartition(name, config);
    }

}

module.exports = WritableStorage;
module.exports.StorageLockedError = StorageLockedError;
module.exports.CorruptFileError = ReadableStorage.CorruptFileError;
module.exports.LOCK_THROW = LOCK_THROW;
module.exports.LOCK_RECLAIM = LOCK_RECLAIM;

1	const fs = require('fs');	4✔
2	const path = require('path');	4✔
3	const WritablePartition = require('../Partition/WritablePartition');	4✔
4	const WritableIndex = require('../Index/WritableIndex');	4✔
5	const ReadableStorage = require('./ReadableStorage');	4✔
6	const { assert, matches, buildMetadataForMatcher, buildMatcherFromMetadata, ensureDirectory } = require('../util');	4✔
7
8	const DEFAULT_WRITE_BUFFER_SIZE = 16 * 1024;	4✔
9
10	const LOCK_RECLAIM = 0x1;	4✔
11	const LOCK_THROW = 0x2;	4✔
12
13	class StorageLockedError extends Error {}
14
15	/**
16	* @typedef {object\|function(object):boolean} Matcher
17	*/
18
19	/**
20	* An append-only storage with highly performant positional range scans.
21	* It's highly optimized for an event-store and hence does not support compaction or data-rewrite, nor any querying
22	*/
23	class WritableStorage extends ReadableStorage {
24
25	/**
26	* @param {string} [storageName] The name of the storage.
27	* @param {object} [config] An object with storage parameters.
28	* @param {object} [config.serializer] A serializer object with methods serialize(document) and deserialize(data).
29	* @param {function(object): string} config.serializer.serialize Default is JSON.stringify.
30	* @param {function(string): object} config.serializer.deserialize Default is JSON.parse.
31	* @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
32	* @param {string} [config.indexDirectory] The path where the indexes should be stored. Defaults to dataDirectory.
33	* @param {string} [config.indexFile] The name of the primary index. Default '{storageName}.index'.
34	* @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
35	* @param {number} [config.writeBufferSize] Size of the write buffer in bytes. Default 16384.
36	* @param {number} [config.maxWriteBufferDocuments] How many documents to have in the write buffer at max. 0 means as much as possible. Default 0.
37	* @param {boolean} [config.syncOnFlush] If fsync should be called on write buffer flush. Set this if you need strict durability. Defaults to false.
38	* @param {boolean} [config.dirtyReads] If dirty reads should be allowed. This means that writes that are in write buffer but not yet flushed can be read. Defaults to true.
39	* @param {function(object, number): string} [config.partitioner] A function that takes a document and sequence number and returns a partition name that the document should be stored in. Defaults to write all documents to the primary partition.
40	* @param {object} [config.indexOptions] An options object that should be passed to all indexes on construction.
41	* @param {string} [config.hmacSecret] A private key that is used to verify matchers retrieved from indexes.
42	* @param {number} [config.lock] One of LOCK_* constants that defines how an existing lock should be handled.
43	*/
44	constructor(storageName = 'storage', config = {}) {	416!
45	if (typeof storageName !== 'string') {	700✔
46	config = storageName;	416✔
47	storageName = undefined;	416✔
48	}
49	const defaults = {	700✔
50	partitioner: (document, number) => '',	1,420✔
51	writeBufferSize: DEFAULT_WRITE_BUFFER_SIZE,
52	maxWriteBufferDocuments: 0,
53	syncOnFlush: false,
54	dirtyReads: true,
55	dataDirectory: '.'
56	};
57	config = Object.assign(defaults, config);	700✔
58	config.indexOptions = Object.assign({ syncOnFlush: config.syncOnFlush }, config.indexOptions);	700✔
59	ensureDirectory(config.dataDirectory);	700✔
60	super(storageName, config);	700✔
61
62	this.lockFile = path.resolve(this.dataDirectory, this.storageFile + '.lock');	700✔
63	if (config.lock === LOCK_RECLAIM) {	700✔
64	this.unlock();	8✔
65	}
66	this.partitioner = config.partitioner;	700✔
67	}
68
69	/**
70	* @inheritDoc
71	* @returns {boolean}
72	* @throws {StorageLockedError} If this storage is locked by another process.
73	*/
74	open() {
75	if (!this.lock()) {	632✔
76	return true;	4✔
77	}
78	return super.open();	624✔
79	}
80
81	/**
82	* Check all partitions torn writes and truncate the storage to the position before the first torn write.
83	* This might delete correctly written events in partitions, if their sequence number is higher than the
84	* torn write in another partition.
85	*/
86	checkTornWrites() {
87	let lastValidSequenceNumber = Number.MAX_SAFE_INTEGER;	8✔
88	this.forEachPartition(partition => {	8✔
89	partition.open();	12✔
90	const tornSequenceNumber = partition.checkTornWrite();	12✔
91	if (tornSequenceNumber >= 0) {	12✔
92	lastValidSequenceNumber = Math.min(lastValidSequenceNumber, tornSequenceNumber);	8✔
93	}
94	});
95	if (lastValidSequenceNumber < Number.MAX_SAFE_INTEGER) {	8!
96	this.truncate(lastValidSequenceNumber);	8✔
97	}
98	this.forEachPartition(partition => partition.close());	12✔
99	}
100
101	/**
102	* Attempt to lock this storage by means of a lock directory.
103	* @returns {boolean} True if the lock was created or false if the lock is already in place.
104	* @throws {StorageLockedError} If this storage is already locked by another process.
105	* @throws {Error} If the lock could not be created.
106	*/
107	lock() {
108	if (this.locked) {	632✔
109	return false;	4✔
110	}
111	try {	628✔
112	fs.mkdirSync(this.lockFile);	628✔
113	this.locked = true;	624✔
114	} catch (e) {
115	/* istanbul ignore if */
116	if (e.code !== 'EEXIST') {	4✔
117	throw new Error(`Error creating lock for storage ${this.storageFile}: ` + e.message);
118	}
119	throw new StorageLockedError(`Storage ${this.storageFile} is locked by another process`);	4✔
120	}
121	return true;	624✔
122	}
123
124	/**
125	* Unlock this storage, no matter if it was previously locked by this writer.
126	* Only use this if you are sure there is no other process still having a writer open.
127	* Current implementation just deletes a lock file that is named like the storage.
128	*/
129	unlock() {
130	if (fs.existsSync(this.lockFile)) {	632✔
131	if (!this.locked) {	624✔
132	this.checkTornWrites();	8✔
133	}
134	fs.rmdirSync(this.lockFile);	624✔
135	}
136	this.locked = false;	632✔
137	}
138
139	/**
140	* @inheritDoc
141	*/
142	close() {
143	if (this.locked) {	1,080✔
144	this.unlock();	624✔
145	}
146	super.close();	1,080✔
147	}
148
149	/**
150	* Add an index entry for the given document at the position and size.
151	*
152	* @private
153	* @param {number} partitionId The partition where the document is stored.
154	* @param {number} position The file offset where the document is stored.
155	* @param {number} size The size of the stored document.
156	* @param {object} document The document to add to the index.
157	* @param {function} [callback] The callback to call when the index is written to disk.
158	* @returns {EntryInterface} The index entry item.
159	*/
160	addIndex(partitionId, position, size, document, callback) {
161	if (!this.index.isOpen()) {	2,344✔
162	this.index.open();	4✔
163	}
164
165	/*if (this.index.lastEntry.position + this.index.lastEntry.size !== position) {
166	this.emit('index-corrupted');
167	throw new Error('Corrupted index, needs to be rebuilt!');
168	}*/
169
170	const entry = new WritableIndex.Entry(this.index.length + 1, position, size, partitionId);	2,344✔
171	this.index.add(entry, (indexPosition) => {	2,344✔
172	this.emit('wrote', document, entry, indexPosition);	2,344✔
173	/* istanbul ignore if */
174	if (typeof callback === 'function') {	2,344✔
175	return callback(indexPosition);
176	}
177	});
178	return entry;	2,344✔
179	}
180
181	/**
182	* Get a partition either by name or by id.
183	* If a partition with the given name does not exist, a new one will be created.
184	* If a partition with the given id does not exist, an error is thrown.
185	*
186	* @protected
187	* @param {string\|number} partitionIdentifier The partition name or the partition Id
188	* @returns {ReadablePartition}
189	* @throws {Error} If an id is given and no such partition exists.
190	*/
191	getPartition(partitionIdentifier) {
192	if (typeof partitionIdentifier === 'string') {	6,156✔
193	const partitionName = this.storageFile + (partitionIdentifier.length ? '.' + partitionIdentifier : '');	2,360✔
194	partitionIdentifier = WritablePartition.idFor(partitionName);	2,360✔
195	if (!this.partitions[partitionIdentifier]) {	2,360✔
196	this.partitions[partitionIdentifier] = this.createPartition(partitionName, this.partitionConfig);	680✔
197	this.emit('partition-created', partitionIdentifier);	680✔
198	}
199	this.partitions[partitionIdentifier].open();	2,360✔
200	return this.partitions[partitionIdentifier];	2,360✔
201	}
202	return super.getPartition(partitionIdentifier);	3,796✔
203	}
204
205	/**
206	* @api
207	* @param {object} document The document to write to storage.
208	* @param {function} [callback] A function that will be called when the document is written to disk.
209	* @returns {number} The 1-based document sequence number in the storage.
210	*/
211	write(document, callback) {
212	const data = this.serializer.serialize(document).toString();	2,344✔
213	const dataSize = Buffer.byteLength(data, 'utf8');	2,344✔
214
215	const partitionName = this.partitioner(document, this.index.length + 1);	2,344✔
216	const partition = this.getPartition(partitionName);	2,344✔
217	const position = partition.write(data, this.length, callback);	2,344✔
218
219	assert(position !== false, 'Error writing document.');	2,344✔
220
221	const indexEntry = this.addIndex(partition.id, position, dataSize, document);	2,344✔
222	this.forEachSecondaryIndex((index, name) => {	2,344✔
223	if (!index.isOpen()) {	1,384✔
224	index.open();	4✔
225	}
226	index.add(indexEntry);	1,384✔
227	this.emit('index-add', name, index.length, document);	1,384✔
228	}, document);
229
230	return this.index.length;	2,344✔
231	}
232
233	/**
234	* Ensure that an index with the given name and document matcher exists.
235	* Will create the index if it doesn't exist, otherwise return the existing index.
236	*
237	* @api
238	* @param {string} name The index name.
239	* @param {Matcher} [matcher] An object that describes the document properties that need to match to add it this index or a function that receives a document and returns true if the document should be indexed.
240	* @returns {ReadableIndex} The index containing all documents that match the query.
241	* @throws {Error} if the index doesn't exist yet and no matcher was specified.
242	*/
243	ensureIndex(name, matcher) {
244	if (name === '_all') {	780✔
245	return this.index;	4✔
246	}
247	if (name in this.secondaryIndexes) {	776✔
248	return this.secondaryIndexes[name].index;	4✔
249	}
250
251	const indexName = this.storageFile + '.' + name + '.index';	772✔
252	if (fs.existsSync(path.join(this.indexDirectory, indexName))) {	772✔
253	return this.openIndex(name, matcher);	16✔
254	}
255
256	assert((typeof matcher === 'object' \|\| typeof matcher === 'function') && matcher !== null, 'Need to specify a matcher.');	756✔
257
258	const metadata = buildMetadataForMatcher(matcher, this.hmac);	752✔
259	const { index } = this.createIndex(indexName, Object.assign({}, this.indexOptions, { metadata }));	752✔
260	try {	752✔
261	this.forEachDocument((document, indexEntry) => {	752✔
262	if (matches(document, matcher)) {	1,748✔
263	index.add(indexEntry);	16✔
264	}
265	});
266	} catch (e) {
267	index.destroy();	4✔
268	throw e;	4✔
269	}
270
271	this.secondaryIndexes[name] = { index, matcher };	748✔
272	this.emit('index-created', name);	748✔
273	return index;	748✔
274	}
275
276	/**
277	* Flush all write buffers to disk.
278	* This is a sync method and will invoke all previously registered flush callbacks.
279	*
280	* @api
281	* @returns {boolean} Returns true if a flush on any partition or the main index was executed.
282	*/
283	flush() {
284	let result = this.index.flush();	60✔
285	this.forEachPartition(partition => result = result \| partition.flush());	60✔
286	this.forEachSecondaryIndex(index => index.flush());	60✔
287	return result;	60✔
288	}
289
290	/**
291	* Iterate all distinct partitions in which the given iterable list of entries are stored.
292	* @param {Iterable<Index.Entry>} entries
293	* @param {function(Index.Entry)} iterationHandler
294	*/
295	forEachDistinctPartitionOf(entries, iterationHandler) {
296	const partitions = [];	36✔
297	const numPartitions = Object.keys(this.partitions).length;	36✔
298	for (let entry of entries) {	36✔
299	if (partitions.indexOf(entry.partition) >= 0) {	56✔
300	continue;	16✔
301	}
302	partitions.push(entry.partition);	40✔
303	iterationHandler(entry);	40✔
304	if (partitions.length === numPartitions) {	40✔
305	break;	20✔
306	}
307	}
308	}
309
310	/**
311	* Truncate all partitions after the given (global) sequence number.
312	*
313	* @private
314	* @param {number} after The document sequence number to truncate after.
315	*/
316	truncatePartitions(after) {
317	if (after === 0) {	52✔
318	this.forEachPartition(partition => partition.truncate(0));	12✔
319	return;	12✔
320	}
321
322	const entries = this.index.range(after + 1); // We need the first entry that is cut off	40✔
323	if (entries === false \|\| entries.length === 0) {	40✔
324	return;	4✔
325	}
326
327	this.forEachDistinctPartitionOf(entries, entry => this.getPartition(entry.partition).truncate(entry.position));	40✔
328	}
329
330	/**
331	* Truncate the storage after the given sequence number.
332	*
333	* @param {number} after The document sequence number to truncate after.
334	*/
335	truncate(after) {
336	/*
337	To truncate the store following steps need to be done:
338
339	1) find all partition positions after which their files should be truncated
340	2) truncate all partitions accordingly
341	3) truncate/rewrite all indexes
342	*/
343	if (!this.index.isOpen()) {	52✔
344	this.index.open();	4✔
345	}
346	if (after < 0) {	52!
NEW UNCOV 347	after += this.index.length;	×
348	}
349
350	this.truncatePartitions(after);	52✔
351
352	this.index.truncate(after);	52✔
353	this.forEachSecondaryIndex(index => {	52✔
354	/* istanbul ignore if */
355	if (!(index instanceof WritableIndex)) {	32✔
356	return;
357	}
358	let closeIndex = false;	32✔
359	if (!index.isOpen()) {	32✔
360	index.open();	4✔
361	closeIndex = true;	4✔
362	}
363	index.truncate(index.find(after));	32✔
364	if (closeIndex) {	32✔
365	index.close();	4✔
366	}
367	});
368	}
369
370	/**
371	* @inheritDoc
372	* Open an existing secondary index and repair any stale entries beyond the current primary
373	* index length. Stale entries can be present when checkTornWrites() truncated the primary
374	* index before this secondary index was loaded into memory.
375	*/
376	openIndex(name, matcher) {
377	const index = super.openIndex(name, matcher);	624✔
378	const lastEntry = index.lastEntry;	608✔
379	if (lastEntry !== false && lastEntry.number > this.index.length) {	608✔
380	index.truncate(index.find(this.index.length));	8✔
381	}
382	return index;	608✔
383	}
384
385	/**
386	* @protected
387	* @param {string} name
388	* @param {object} [options]
389	* @returns {{ index: WritableIndex, matcher: Matcher }}
390	*/
391	createIndex(name, options = {}) {	×
392	const index = new WritableIndex(name, options);	1,556✔
393	let matcher;
394
395	// If the index contains a matcher (possibly a serialized function) we check HMAC
396	// to prevent evaluating unknown code.
397	if (index.metadata.matcher) {	1,548✔
398	try {	848✔
399	matcher = buildMatcherFromMetadata(index.metadata, this.hmac);	848✔
400	} catch (e) {
401	index.destroy();	4✔
402	throw e;	4✔
403	}
404	}
405
406	return { index, matcher };	1,544✔
407	}
408
409	/**
410	* @protected
411	* @param {string} name
412	* @param {object} [config]
413	* @returns {WritablePartition}
414	*/
415	createPartition(name, config = {}) {	×
416	return new WritablePartition(name, config);	760✔
417	}
418
419	}
420
421	module.exports = WritableStorage;	4✔
422	module.exports.StorageLockedError = StorageLockedError;	4✔
423	module.exports.CorruptFileError = ReadableStorage.CorruptFileError;	4✔
424	module.exports.LOCK_THROW = LOCK_THROW;	4✔
425	module.exports.LOCK_RECLAIM = LOCK_RECLAIM;	4✔

albe / node-event-storage / 23367085163

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

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