23711535156

Committed 29 Mar 2026 02:43PM UTC coverage: 97.02% (-1.0%) from 98.004%

Build # 23711535156

Build Type

Pull #277

github

Committed by

web-flow

Commit Message

Merge 1b1c84f1a into de1c33ad1

Pull Request Pull Request #277: Refactor torn write repair: extract `findTornWriteBoundary` and `truncateAfterSequence`, restore index-based `truncatePartitions`

Coverage Stats

701 of 746 branches covered (93.97%)

Branch coverage included in aggregate %.

34 of 48 new or added lines in 3 files covered. (70.83%)

7 existing lines in 1 file now uncovered.

1578 of 1603 relevant lines covered (98.44%)

1316.18 hits per line

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

93.36

/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;
        }
        const result = super.open();
        this.emit('ready');
        return result;
    }

    /**
     * Scan every partition's last document to detect torn writes inline.
     * A document is torn when its expected end exceeds the actual file size:
     *   position + documentWriteSize(dataSize) > partition.size
     *
     * @private
     * @returns {{ lastValidSequenceNumber: number, maxPartitionSequenceNumber: number }}
     */
    findTornWriteBoundary() {
        let lastValidSequenceNumber = Number.MAX_SAFE_INTEGER;
        let maxPartitionSequenceNumber = -1;
        this.forEachPartition(partition => {
            partition.open();
            if (partition.size === 0) return;
            const position = partition.findDocumentPositionBefore(partition.size);
            if (position === false || position < 0) return;
            const reader = partition.prepareReadBufferBackwards(position);
            if (!reader.buffer) return;
            const { sequenceNumber, dataSize } = partition.readDocumentHeader(reader.buffer, reader.cursor, position);
            if (position + partition.documentWriteSize(dataSize) > partition.size) {
                // Torn write: the document extends beyond the end of the file.
                lastValidSequenceNumber = Math.min(lastValidSequenceNumber, sequenceNumber);
            } else {
                maxPartitionSequenceNumber = Math.max(maxPartitionSequenceNumber, sequenceNumber);
            }
        });
        return { lastValidSequenceNumber, maxPartitionSequenceNumber };
    }

    /**
     * Check all partitions for torn writes, physically repair each partition, truncate all indexes
     * to the torn-write boundary, and then reindex to rebuild any missing index entries.
     *
     * A document is torn when the partition file ends before the document's expected end position
     * (i.e. position + documentWriteSize(dataSize) > partition.size).  Detected inline in
     * findTornWriteBoundary(), without any checkTornWrite() call.
     *
     * Repair flow:
     * 1. findTornWriteBoundary() reads the last document of every partition and finds the global
     *    torn-write boundary (minimum torn sequence number across all partitions).
     * 2. If torn writes were found, truncateAfterSequence() removes all documents at or beyond
     *    the boundary from each partition.
     * 3. Truncate all indexes to the torn-write boundary, then reindex to fill any lagging entries.
     * 4. If no torn writes were found but the index is lagging, reindex directly.
     */
    checkTornWrites() {
        const { lastValidSequenceNumber, maxPartitionSequenceNumber } = this.findTornWriteBoundary();

        if (lastValidSequenceNumber < Number.MAX_SAFE_INTEGER) {
            // Phase 2: remove all documents at or beyond the torn-write boundary from each partition.
            this.forEachPartition(partition => {
                partition.open();
                partition.truncateAfterSequence(lastValidSequenceNumber - 1);
            });

            // Truncate all indexes to the torn-write boundary.
            this.index.open();
            this.index.truncate(lastValidSequenceNumber);
            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(lastValidSequenceNumber));
                if (closeIndex) {
                    index.close();
                }
            });

            // Reindex to fill in any missing complete-document entries.
            this.reindex(this.index.length);
        } else if (maxPartitionSequenceNumber >= 0 && maxPartitionSequenceNumber + 1 > this.index.length) {
            // No torn writes, but the index is lagging — repair it.
            this.reindex(this.index.length);
        }

        this.forEachPartition(partition => partition.close());
    }

    /**
     * Rebuild the primary index and all loaded secondary indexes starting from the given sequence
     * number by scanning the partition data directly.
     * This is the building block for both auto-repair (invoked automatically when the primary
     * index is found to be lagging in checkTornWrites()) and for user-driven re-indexing after
     * index corruption.
     *
     * @api
     * @param {number} [fromSequenceNumber=0] The number of primary index entries to keep intact.
     *   All index entries beyond this position will be removed and rebuilt from partition data.
     *   Defaults to 0, which rebuilds all indexes from scratch.
     */
    reindex(fromSequenceNumber = 0) {
        this.index.truncate(fromSequenceNumber);

        // Truncate all loaded secondary indexes to match the new primary length.
        this.forEachSecondaryIndex(index => {
            /* istanbul ignore if */
            if (!(index instanceof WritableIndex)) {
                return;
            }
            // find(0) returns 0, so truncate(0) will remove all entries when fromSequenceNumber===0
            index.truncate(fromSequenceNumber === 0 ? 0 : index.find(fromSequenceNumber));
        });

        // Ensure all partitions are open so iteratePartitionsBySequenceNumber can read them.
        this.forEachPartition(partition => {
            if (!partition.isOpen()) {
                partition.open();
            }
        });

        // Scan partitions in sequence-number order and rebuild index entries.
        for (const { document, partitionId, position, size } of
            this.iteratePartitionsBySequenceNumber(fromSequenceNumber, Number.MAX_SAFE_INTEGER)) {
            const newEntry = new WritableIndex.Entry(this.index.length + 1, position, size, partitionId);
            this.index.add(newEntry);

            this.forEachSecondaryIndex((secIndex, name) => {
                /* istanbul ignore if */
                if (!(secIndex instanceof WritableIndex)) {
                    return;
                }
                /* istanbul ignore if */
                if (!secIndex.isOpen()) {
                    secIndex.open();
                }
                const { matcher } = this.secondaryIndexes[name];
                if (matches(document, matcher)) {
                    secIndex.add(newEntry);
                }
            });
        }

        this.flush();
    }

    /**
     * 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;
    }

    /**
     * Register a handler that is called before a document is written to storage.
     * The handler receives the document and the partition metadata and may throw to abort the write.
     * Multiple handlers can be registered; all run on every write in registration order.
     * Equivalent to `storage.on('preCommit', hook)`.
     *
     * @api
     * @param {function(object, object): void} hook A function receiving (document, partitionMetadata).
     */
    preCommit(hook) {
        this.on('preCommit', hook);
    }

    /**
     * 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 partitionShortName = partitionIdentifier;
            const partitionName = this.storageFile + (partitionIdentifier.length ? '.' + partitionIdentifier : '');
            partitionIdentifier = WritablePartition.idFor(partitionName);
            if (!this.partitions[partitionIdentifier]) {
                const partitionConfig = typeof this.partitionConfig.metadata === 'function'
                    ? { ...this.partitionConfig, metadata: this.partitionConfig.metadata(partitionShortName) }
                    : this.partitionConfig;
                this.partitions[partitionIdentifier] = this.createPartition(partitionName, 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);
        if (this.listenerCount('preCommit') > 0) {
            this.emit('preCommit', document, partition.metadata);
        }
        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.
     *
     * Assumes the primary index is fully consistent with the partition data. Looks up the first
     * index entry after `after` for each affected partition and truncates the partition file
     * at that entry's byte position.
     *
     * @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) {
            // Secondary index is ahead of primary: truncate stale entries.
            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;

albe / node-event-storage / 23711535156

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