606483609

Build Type

github

Committed by GitHub

Commit Message

Merge 69c78ac45 into f02cad31a

Pull Request Pull Request #155: Automatically repair torn writes on opening

Run Details

554 of 590 branches covered (93.9%)

Branch coverage included in aggregate %.

12 of 18 new or added lines in 2 files covered. (66.67%)

3 existing lines in 3 files now uncovered.

1322 of 1340 relevant lines covered (98.66%)

1112.21 hits per line

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

92.17

/src/Partition/ReadablePartition.js

const fs = require('fs');
const path = require('path');
const events = require('events');
const { assert, alignTo } = require('../util');

const DEFAULT_READ_BUFFER_SIZE = 64 * 1024;
const DOCUMENT_HEADER_SIZE = 16;
const DOCUMENT_ALIGNMENT = 4;
const DOCUMENT_SEPARATOR = "\x00\x00\x1E\n";
const DOCUMENT_FOOTER_SIZE = 4 /* additional data size footer */ + DOCUMENT_SEPARATOR.length;

// node-event-store partition V03
const HEADER_MAGIC = "nesprt03";

const NES_EPOCH = new Date('2020-01-01T00:00:00');

class CorruptFileError extends Error {}
class InvalidDataSizeError extends Error {}

/**
 * Method for hashing a string (partition name) to a 32-bit unsigned integer.
 *
 * @param {string} str
 * @returns {number}
 */
function hash(str) {
    /* istanbul ignore if */
    if (str.length === 0) {
        return 0;
    }
    let hash = 5381,
        i    = str.length;

    while(i) {
        hash = ((hash << 5) + hash) ^ str.charCodeAt(--i); // jshint ignore:line
    }

    /* JavaScript does bitwise operations (like XOR, above) on 32-bit signed
     * integers. Since we want the results to be always positive, convert the
     * signed int to an unsigned by doing an unsigned bitshift. */
    return hash >>> 0; // jshint ignore:line
}

/**
 * A partition is a single file where the storage will write documents to depending on some partitioning rules.
 * In the case of an event store, this is most likely the (write) streams.
 */
class ReadablePartition extends events.EventEmitter {

    /**
     * Get the id for a specific partition name.
     *
     * @param {string} name
     * @returns {number}
     */
    static idFor(name) {
        return hash(name);
    }

    /**
     * @param {string} name The name of the partition.
     * @param {object} [config] An object with storage parameters.
     * @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
     * @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
     */
    constructor(name, config = {}) {
        super();
        assert(typeof name === 'string' && name !== '', 'Must specify a partition name.');

        let defaults = {
            dataDirectory: '.',
            readBufferSize: DEFAULT_READ_BUFFER_SIZE
        };
        config = Object.assign(defaults, config);
        this.dataDirectory = path.resolve(config.dataDirectory);

        this.name = name;
        this.id = ReadablePartition.idFor(name);
        this.fileName = path.resolve(this.dataDirectory, this.name);
        this.fileMode = 'r';
        this.headerSize = 0;

        this.readBufferSize = config.readBufferSize >>> 0;  // jshint ignore:line
    }

    /**
     * Check if the partition file is opened.
     *
     * @returns {boolean}
     */
    isOpen() {
        return !!this.fd;
    }

    /**
     * Open the partition storage and create read buffers.
     *
     * @api
     * @returns {boolean} Returns false if the file is not a valid partition.
     */
    open() {
        if (this.fd) {
            return true;
        }

        this.fd = fs.openSync(this.fileName, this.fileMode);

        // allocUnsafeSlow because we don't need buffer pooling for these relatively long-lived buffers
        this.readBuffer = Buffer.allocUnsafeSlow(this.readBufferSize);
        // Where inside the file the read buffer starts
        this.readBufferPos = -1;
        this.readBufferLength = 0;

        this.headerSize = 0;
        this.size = this.readFileSize();
        if (this.size <= 0) {
            this.close();
            return false;
        }

        this.size -= this.readMetadata();

        return true;
    }

    /**
     * @returns {number} -1 if the partition is ok and the sequence number of the broken document if a torn write was detected.
     */
    checkTornWrite() {
        const reader = this.prepareReadBufferBackwards(this.size);
        const separator = reader.buffer.toString('ascii', reader.cursor - DOCUMENT_SEPARATOR.length, reader.cursor);
        if (separator !== DOCUMENT_SEPARATOR) {
            const position = this.findDocumentPositionBefore(this.size);
            const reader = this.prepareReadBuffer(position);
            const { sequenceNumber } = this.readDocumentHeader(reader.buffer, reader.cursor, position);
            return sequenceNumber;
        }
        return -1;
    }

    /**
     * Read the partition metadata from the file.
     *
     * @private
     * @returns {number} The size of the metadata header.
     * @throws {Error} if the file header magic value is invalid.
     * @throws {Error} if the metadata size in the header is invalid.
     */
    readMetadata() {
        assert(this.size >= 16, `Invalid file.`);

        const headerBuffer = Buffer.allocUnsafe(8 + 4);
        fs.readSync(this.fd, headerBuffer, 0, 8 + 4, 0);
        const headerMagic = headerBuffer.toString('utf8', 0, 8);

        assert(headerMagic.substr(0, 6) === HEADER_MAGIC.substr(0, 6), `Invalid file header in partition ${this.name}.`);

        this.header = headerMagic;
        assert(headerMagic === HEADER_MAGIC, `Invalid file version. The partition ${this.name} was created with a different library version (${headerMagic.substr(6)}).`);

        const metadataSize = headerBuffer.readUInt32BE(8);
        assert(metadataSize > 2 && metadataSize <= 4096, 'Invalid metadata size.');

        const metadataBuffer = Buffer.allocUnsafe(metadataSize - 1);
        metadataBuffer.fill(" ");
        fs.readSync(this.fd, metadataBuffer, 0, metadataSize - 1, 8 + 4);
        const metadata = metadataBuffer.toString('utf8').trim();
        try {
            this.metadata = JSON.parse(metadata);
            this.metadata.epoch = this.metadata.epoch /* istanbul ignore next */|| NES_EPOCH.getTime();
        } catch (e) {
            throw new Error('Invalid metadata.');
        }
        this.headerSize = 8 + 4 + metadataSize;
        return this.headerSize;
    }

    /**
     * Get the storage size for a document of a given size.
     *
     * @param {number} dataSize The actual data size of the document.
     * @returns {number} The size of the data including header, padded to 16 bytes alignment and ended with a line break.
     */
    documentWriteSize(dataSize) {
        const padSize = alignTo(dataSize + DOCUMENT_FOOTER_SIZE, DOCUMENT_ALIGNMENT);
        return DOCUMENT_HEADER_SIZE + dataSize + padSize + DOCUMENT_FOOTER_SIZE;
    }

    /**
     * @protected
     * @returns {number} The file size not including the file header.
     */
    readFileSize() {
        const stat = fs.statSync(this.fileName);
        return stat.size - this.headerSize;
    }

    /**
     * Close the partition and frees up all resources.
     *
     * @api
     * @returns void
     */
    close() {
        if (this.fd) {
            fs.closeSync(this.fd);
            this.fd = null;
        }
        if (this.readBuffer) {
            this.readBuffer = null;
            this.readBufferPos = -1;
            this.readBufferLength = 0;
        }
    }

    /**
     * Fill the internal read buffer starting from the given position.
     *
     * @private
     * @param {number} [from] The file position to start filling the read buffer from. Default 0.
     */
    fillBuffer(from = 0) {
        this.readBufferLength = fs.readSync(this.fd, this.readBuffer, 0, this.readBuffer.byteLength, this.headerSize + from);
        this.readBufferPos = from;
    }

    /**
     * @private
     * @param {Buffer} buffer The buffer to read the data length from.
     * @param {number} offset The position inside the buffer to start reading from.
     * @param {number} position The file position to start reading from.
     * @param {number} [size] The expected byte size of the document at the given position.
     * @returns {{ dataSize: number, sequenceNumber: number, time64: number }} The metadata fields of the document
     * @throws {Error} if the storage entry at the given position is corrupted.
     * @throws {InvalidDataSizeError} if the document size at the given position does not match the provided size.
     * @throws {CorruptFileError} if the document at the given position can not be read completely.
     */
    readDocumentHeader(buffer, offset, position, size) {
        const dataSize = buffer.readUInt32BE(offset + 0);
        assert(dataSize > 0 && dataSize <= 64 * 1024 * 1024, `Error reading document size from ${position}, got ${dataSize}.`);

        if (size && dataSize !== size) {
            throw new InvalidDataSizeError(`Invalid document size ${dataSize} at position ${position}, expected ${size}.`);
        }

        const sequenceNumber = buffer.readUInt32BE(offset + 4);
        const time64 = buffer.readDoubleBE(offset + 8);
        return ({ dataSize, sequenceNumber, time64 });
    }

    /**
     * Prepare the read buffer for reading from the specified position.
     *
     * @protected
     * @param {number} position The position in the file to prepare the read buffer for reading from.
     * @returns {{ buffer: Buffer|null, cursor: number, length: number }} A reader object with properties `buffer`, `cursor` and `length`.
     */
    prepareReadBuffer(position) {
        if (position + DOCUMENT_HEADER_SIZE >= this.size) {
            return ({ buffer: null, cursor: 0, length: 0 });
        }
        let bufferCursor = position - this.readBufferPos;
        if (this.readBufferPos < 0 || bufferCursor < 0 || bufferCursor + DOCUMENT_HEADER_SIZE + DOCUMENT_ALIGNMENT > this.readBufferLength) {
            this.fillBuffer(position);
            bufferCursor = 0;
        }
        return ({ buffer: this.readBuffer, cursor: bufferCursor, length: this.readBufferLength });
    }

    /**
     * Prepare the read buffer for reading *before* the specified position. Don't try to reader *after* the returned cursor.
     *
     * @protected
     * @param {number} position The position in the file to prepare the read buffer for reading before.
     * @returns {{ buffer: Buffer|null, cursor: number, length: number }} A reader object with properties `buffer`, `cursor` and `length`.
     */
    prepareReadBufferBackwards(position) {
        if (position < 0) {
            return ({ buffer: null, cursor: 0, length: 0 });
        }
        let bufferCursor = position - this.readBufferPos;
        if (this.readBufferPos < 0 || (this.readBufferPos > 0 && bufferCursor < DOCUMENT_FOOTER_SIZE)) {
            this.fillBuffer(Math.max(position - this.readBuffer.byteLength, 0));
            bufferCursor = position - this.readBufferPos;
        }
        return ({ buffer: this.readBuffer, cursor: bufferCursor, length: this.readBufferLength });
    }

    /**
     * Read the data from the given position.
     *
     * @api
     * @param {number} position The file position to read from.
     * @param {number} [size] The expected byte size of the document at the given position.
     * @returns {string|boolean} The data stored at the given position or false if no data could be read.
     * @throws {Error} if the storage entry at the given position is corrupted.
     * @throws {InvalidDataSizeError} if the document size at the given position does not match the provided size.
     * @throws {CorruptFileError} if the document at the given position can not be read completely.
     */
    readFrom(position, size = 0) {
        assert(this.fd, 'Partition is not opened.');
        assert((position % DOCUMENT_ALIGNMENT) === 0, `Invalid read position ${position}. Needs to be a multiple of ${DOCUMENT_ALIGNMENT}.`);

        const reader = this.prepareReadBuffer(position);
        if (reader.length < size + DOCUMENT_HEADER_SIZE) {
            return false;
        }

        let dataPosition = reader.cursor + DOCUMENT_HEADER_SIZE;
        const { dataSize } = this.readDocumentHeader(reader.buffer, reader.cursor, position, size);

        // TODO: This should only be checked on opening
        const writeSize = this.documentWriteSize(dataSize);
        if (position + writeSize > this.size) {
            throw new CorruptFileError(`Invalid document at position ${position}. This may be caused by an unfinished write.`);
        }

        if (dataSize + DOCUMENT_HEADER_SIZE > reader.buffer.byteLength) {
            //console.log('sync read for large document size', dataLength, 'at position', position);
            const tempReadBuffer = Buffer.allocUnsafe(dataSize);
            fs.readSync(this.fd, tempReadBuffer, 0, dataSize, this.headerSize + position + DOCUMENT_HEADER_SIZE);
            return tempReadBuffer.toString('utf8');
        }

        if (reader.cursor > 0 && dataPosition + dataSize > reader.length) {
            this.fillBuffer(position);
            dataPosition = DOCUMENT_HEADER_SIZE;
        }

        return reader.buffer.toString('utf8', dataPosition, dataPosition + dataSize);
    }

    /**
     * Find the start position of the document that precedes the given position.
     *
     * @protected
     * @param {number} position The file position to read backwards from.
     * @returns {number|boolean} The start position of the first document before the given position or false if no header could be found.
     */
    findDocumentPositionBefore(position) {
        assert(this.fd, 'Partition is not opened.');
        position -= (position % DOCUMENT_ALIGNMENT);
        if (position <= 0) {
            return false;
        }

        const separatorSize = DOCUMENT_SEPARATOR.length;
        // Optimization if we are at an exact document boundary, where we can just read the document size
        let reader = this.prepareReadBufferBackwards(position);
        const block = reader.buffer.toString('ascii', reader.cursor - separatorSize, reader.cursor);
        if (block === DOCUMENT_SEPARATOR) {
            const dataSize = reader.buffer.readUInt32BE(reader.cursor - separatorSize - 4);
            return position - this.documentWriteSize(dataSize);
        }

        do {
            reader = this.prepareReadBufferBackwards(position - separatorSize);

            const bufferSeparatorPosition = reader.buffer.lastIndexOf(DOCUMENT_SEPARATOR, reader.cursor - separatorSize, 'ascii');
            if (bufferSeparatorPosition >= 0) {
                position = this.readBufferPos + bufferSeparatorPosition + separatorSize;
                break;
            }
            position -= this.readBufferLength;
        } while (position > 0);
        return position;
    }

    /**
     * @api
     * @param {number} [after] The document position to start reading from.
     * @returns {Generator<string>} A generator that returns all documents in this partition.
     */
    *readAll(after = 0) {
        let position = after < 0 ? this.size + after + 1 : after;
        let data;
        while ((data = this.readFrom(position)) !== false) {
            yield data;
            position += this.documentWriteSize(Buffer.byteLength(data, 'utf8'));
        }
    }

    /**
     * @api
     * @param {number} [before] The document position to start reading backward from.
     * @returns {Generator<string>} A generator that returns all documents in this partition in reverse order.
     */
    *readAllBackwards(before = -1) {
        let position = before < 0 ? this.size + before + 1 : before;
        while ((position = this.findDocumentPositionBefore(position)) !== false) {
            const data = this.readFrom(position);
            yield data;
        }
    }
}

module.exports = ReadablePartition;
module.exports.CorruptFileError = CorruptFileError;
module.exports.InvalidDataSizeError = InvalidDataSizeError;
module.exports.HEADER_MAGIC = HEADER_MAGIC;
module.exports.DOCUMENT_SEPARATOR = DOCUMENT_SEPARATOR;
module.exports.DOCUMENT_ALIGNMENT = DOCUMENT_ALIGNMENT;
module.exports.DOCUMENT_HEADER_SIZE = DOCUMENT_HEADER_SIZE;
module.exports.DOCUMENT_FOOTER_SIZE = DOCUMENT_FOOTER_SIZE;

1	const fs = require('fs');	4✔
2	const path = require('path');	4✔
3	const events = require('events');	4✔
4	const { assert, alignTo } = require('../util');	4✔
5
6	const DEFAULT_READ_BUFFER_SIZE = 64 * 1024;	4✔
7	const DOCUMENT_HEADER_SIZE = 16;	4✔
8	const DOCUMENT_ALIGNMENT = 4;	4✔
9	const DOCUMENT_SEPARATOR = "\x00\x00\x1E\n";	4✔
10	const DOCUMENT_FOOTER_SIZE = 4 /* additional data size footer */ + DOCUMENT_SEPARATOR.length;	4✔
11
12	// node-event-store partition V03
13	const HEADER_MAGIC = "nesprt03";	4✔
14
15	const NES_EPOCH = new Date('2020-01-01T00:00:00');	4✔
16
17	class CorruptFileError extends Error {}
18	class InvalidDataSizeError extends Error {}
19
20	/**
21	* Method for hashing a string (partition name) to a 32-bit unsigned integer.
22	*
23	* @param {string} str
24	* @returns {number}
25	*/
26	function hash(str) {
27	/* istanbul ignore if */
28	if (str.length === 0) {	3,024✔
29	return 0;
30	}
31	let hash = 5381,	3,024✔
32	i = str.length;	3,024✔
33
34	while(i) {	3,024✔
35	hash = ((hash << 5) + hash) ^ str.charCodeAt(--i); // jshint ignore:line	31,700✔
36	}
37
38	/* JavaScript does bitwise operations (like XOR, above) on 32-bit signed
39	* integers. Since we want the results to be always positive, convert the
40	* signed int to an unsigned by doing an unsigned bitshift. */
41	return hash >>> 0; // jshint ignore:line	3,024✔
42	}
43
44	/**
45	* A partition is a single file where the storage will write documents to depending on some partitioning rules.
46	* In the case of an event store, this is most likely the (write) streams.
47	*/
48	class ReadablePartition extends events.EventEmitter {
49
50	/**
51	* Get the id for a specific partition name.
52	*
53	* @param {string} name
54	* @returns {number}
55	*/
56	static idFor(name) {
57	return hash(name);	3,024✔
58	}
59
60	/**
61	* @param {string} name The name of the partition.
62	* @param {object} [config] An object with storage parameters.
63	* @param {string} [config.dataDirectory] The path where the storage data should reside. Default '.'.
64	* @param {number} [config.readBufferSize] Size of the read buffer in bytes. Default 4096.
65	*/
66	constructor(name, config = {}) {	×
67	super();	984✔
68	assert(typeof name === 'string' && name !== '', 'Must specify a partition name.');	984✔
69
70	let defaults = {	976✔
71	dataDirectory: '.',
72	readBufferSize: DEFAULT_READ_BUFFER_SIZE
73	};
74	config = Object.assign(defaults, config);	976✔
75	this.dataDirectory = path.resolve(config.dataDirectory);	976✔
76
77	this.name = name;	976✔
78	this.id = ReadablePartition.idFor(name);	976✔
79	this.fileName = path.resolve(this.dataDirectory, this.name);	976✔
80	this.fileMode = 'r';	976✔
81	this.headerSize = 0;	976✔
82
83	this.readBufferSize = config.readBufferSize >>> 0; // jshint ignore:line	976✔
84	}
85
86	/**
87	* Check if the partition file is opened.
88	*
89	* @returns {boolean}
90	*/
91	isOpen() {
92	return !!this.fd;	12✔
93	}
94
95	/**
96	* Open the partition storage and create read buffers.
97	*
98	* @api
99	* @returns {boolean} Returns false if the file is not a valid partition.
100	*/
101	open() {
102	if (this.fd) {	1,928!
103	return true;	×
104	}
105
106	this.fd = fs.openSync(this.fileName, this.fileMode);	1,928✔
107
108	// allocUnsafeSlow because we don't need buffer pooling for these relatively long-lived buffers
109	this.readBuffer = Buffer.allocUnsafeSlow(this.readBufferSize);	1,928✔
110	// Where inside the file the read buffer starts
111	this.readBufferPos = -1;	1,928✔
112	this.readBufferLength = 0;	1,928✔
113
114	this.headerSize = 0;	1,928✔
115	this.size = this.readFileSize();	1,928✔
116	if (this.size <= 0) {	1,928✔
117	this.close();	776✔
118	return false;	776✔
119	}
120
121	this.size -= this.readMetadata();	1,152✔
122
123	return true;	1,144✔
124	}
125
126	/**
127	* @returns {number} -1 if the partition is ok and the sequence number of the broken document if a torn write was detected.
128	*/
129	checkTornWrite() {
130	const reader = this.prepareReadBufferBackwards(this.size);	124✔
131	const separator = reader.buffer.toString('ascii', reader.cursor - DOCUMENT_SEPARATOR.length, reader.cursor);	124✔
132	if (separator !== DOCUMENT_SEPARATOR) {	124!
NEW 133	const position = this.findDocumentPositionBefore(this.size);	×
NEW 134	const reader = this.prepareReadBuffer(position);	×
NEW 135	const { sequenceNumber } = this.readDocumentHeader(reader.buffer, reader.cursor, position);	×
NEW 136	return sequenceNumber;	×
137	}
138	return -1;	124✔
139	}
140
141	/**
142	* Read the partition metadata from the file.
143	*
144	* @private
145	* @returns {number} The size of the metadata header.
146	* @throws {Error} if the file header magic value is invalid.
147	* @throws {Error} if the metadata size in the header is invalid.
148	*/
149	readMetadata() {
150	assert(this.size >= 16, `Invalid file.`);	1,152✔
151
152	const headerBuffer = Buffer.allocUnsafe(8 + 4);	1,144✔
153	fs.readSync(this.fd, headerBuffer, 0, 8 + 4, 0);	1,144✔
154	const headerMagic = headerBuffer.toString('utf8', 0, 8);	1,144✔
155
156	assert(headerMagic.substr(0, 6) === HEADER_MAGIC.substr(0, 6), `Invalid file header in partition ${this.name}.`);	1,144✔
157
158	this.header = headerMagic;	1,144✔
159	assert(headerMagic === HEADER_MAGIC, `Invalid file version. The partition ${this.name} was created with a different library version (${headerMagic.substr(6)}).`);	1,144✔
160
161	const metadataSize = headerBuffer.readUInt32BE(8);	1,144✔
162	assert(metadataSize > 2 && metadataSize <= 4096, 'Invalid metadata size.');	1,144✔
163
164	const metadataBuffer = Buffer.allocUnsafe(metadataSize - 1);	1,144✔
165	metadataBuffer.fill(" ");	1,144✔
166	fs.readSync(this.fd, metadataBuffer, 0, metadataSize - 1, 8 + 4);	1,144✔
167	const metadata = metadataBuffer.toString('utf8').trim();	1,144✔
168	try {	1,144✔
169	this.metadata = JSON.parse(metadata);	1,144✔
170	this.metadata.epoch = this.metadata.epoch /* istanbul ignore next */\|\| NES_EPOCH.getTime();	1,144✔
171	} catch (e) {
172	throw new Error('Invalid metadata.');	×
173	}
174	this.headerSize = 8 + 4 + metadataSize;	1,144✔
175	return this.headerSize;	1,144✔
176	}
177
178	/**
179	* Get the storage size for a document of a given size.
180	*
181	* @param {number} dataSize The actual data size of the document.
182	* @returns {number} The size of the data including header, padded to 16 bytes alignment and ended with a line break.
183	*/
184	documentWriteSize(dataSize) {
185	const padSize = alignTo(dataSize + DOCUMENT_FOOTER_SIZE, DOCUMENT_ALIGNMENT);	21,240✔
186	return DOCUMENT_HEADER_SIZE + dataSize + padSize + DOCUMENT_FOOTER_SIZE;	21,240✔
187	}
188
189	/**
190	* @protected
191	* @returns {number} The file size not including the file header.
192	*/
193	readFileSize() {
194	const stat = fs.statSync(this.fileName);	1,940✔
195	return stat.size - this.headerSize;	1,940✔
196	}
197
198	/**
199	* Close the partition and frees up all resources.
200	*
201	* @api
202	* @returns void
203	*/
204	close() {
205	if (this.fd) {	2,400✔
206	fs.closeSync(this.fd);	1,928✔
207	this.fd = null;	1,928✔
208	}
209	if (this.readBuffer) {	2,400✔
210	this.readBuffer = null;	1,928✔
211	this.readBufferPos = -1;	1,928✔
212	this.readBufferLength = 0;	1,928✔
213	}
214	}
215
216	/**
217	* Fill the internal read buffer starting from the given position.
218	*
219	* @private
220	* @param {number} [from] The file position to start filling the read buffer from. Default 0.
221	*/
222	fillBuffer(from = 0) {	×
223	this.readBufferLength = fs.readSync(this.fd, this.readBuffer, 0, this.readBuffer.byteLength, this.headerSize + from);	528✔
224	this.readBufferPos = from;	528✔
225	}
226
227	/**
228	* @private
229	* @param {Buffer} buffer The buffer to read the data length from.
230	* @param {number} offset The position inside the buffer to start reading from.
231	* @param {number} position The file position to start reading from.
232	* @param {number} [size] The expected byte size of the document at the given position.
233	* @returns {{ dataSize: number, sequenceNumber: number, time64: number }} The metadata fields of the document
234	* @throws {Error} if the storage entry at the given position is corrupted.
235	* @throws {InvalidDataSizeError} if the document size at the given position does not match the provided size.
236	* @throws {CorruptFileError} if the document at the given position can not be read completely.
237	*/
238	readDocumentHeader(buffer, offset, position, size) {
239	const dataSize = buffer.readUInt32BE(offset + 0);	8,192✔
240	assert(dataSize > 0 && dataSize <= 64 * 1024 * 1024, `Error reading document size from ${position}, got ${dataSize}.`);	8,192✔
241
242	if (size && dataSize !== size) {	8,192✔
243	throw new InvalidDataSizeError(`Invalid document size ${dataSize} at position ${position}, expected ${size}.`);	8✔
244	}
245
246	const sequenceNumber = buffer.readUInt32BE(offset + 4);	8,184✔
247	const time64 = buffer.readDoubleBE(offset + 8);	8,184✔
248	return ({ dataSize, sequenceNumber, time64 });	8,184✔
249	}
250
251	/**
252	* Prepare the read buffer for reading from the specified position.
253	*
254	* @protected
255	* @param {number} position The position in the file to prepare the read buffer for reading from.
256	* @returns {{ buffer: Buffer\|null, cursor: number, length: number }} A reader object with properties `buffer`, `cursor` and `length`.
257	*/
258	prepareReadBuffer(position) {
259	if (position + DOCUMENT_HEADER_SIZE >= this.size) {	6,180!
260	return ({ buffer: null, cursor: 0, length: 0 });	×
261	}
262	let bufferCursor = position - this.readBufferPos;	6,180✔
263	if (this.readBufferPos < 0 \|\| bufferCursor < 0 \|\| bufferCursor + DOCUMENT_HEADER_SIZE + DOCUMENT_ALIGNMENT > this.readBufferLength) {	6,180✔
264	this.fillBuffer(position);	376✔
265	bufferCursor = 0;	376✔
266	}
267	return ({ buffer: this.readBuffer, cursor: bufferCursor, length: this.readBufferLength });	6,180✔
268	}
269
270	/**
271	* Prepare the read buffer for reading before the specified position. Don't try to reader after the returned cursor.
272	*
273	* @protected
274	* @param {number} position The position in the file to prepare the read buffer for reading before.
275	* @returns {{ buffer: Buffer\|null, cursor: number, length: number }} A reader object with properties `buffer`, `cursor` and `length`.
276	*/
277	prepareReadBufferBackwards(position) {
278	if (position < 0) {	756!
279	return ({ buffer: null, cursor: 0, length: 0 });	×
280	}
281	let bufferCursor = position - this.readBufferPos;	756✔
282	if (this.readBufferPos < 0 \|\| (this.readBufferPos > 0 && bufferCursor < DOCUMENT_FOOTER_SIZE)) {	756✔
283	this.fillBuffer(Math.max(position - this.readBuffer.byteLength, 0));	152✔
284	bufferCursor = position - this.readBufferPos;	152✔
285	}
286	return ({ buffer: this.readBuffer, cursor: bufferCursor, length: this.readBufferLength });	756✔
287	}
288
289	/**
290	* Read the data from the given position.
291	*
292	* @api
293	* @param {number} position The file position to read from.
294	* @param {number} [size] The expected byte size of the document at the given position.
295	* @returns {string\|boolean} The data stored at the given position or false if no data could be read.
296	* @throws {Error} if the storage entry at the given position is corrupted.
297	* @throws {InvalidDataSizeError} if the document size at the given position does not match the provided size.
298	* @throws {CorruptFileError} if the document at the given position can not be read completely.
299	*/
300	readFrom(position, size = 0) {	5,360✔
301	assert(this.fd, 'Partition is not opened.');	8,232✔
302	assert((position % DOCUMENT_ALIGNMENT) === 0, `Invalid read position ${position}. Needs to be a multiple of ${DOCUMENT_ALIGNMENT}.`);	8,228✔
303
304	const reader = this.prepareReadBuffer(position);	8,220✔
305	if (reader.length < size + DOCUMENT_HEADER_SIZE) {	8,220✔
306	return false;	28✔
307	}
308
309	let dataPosition = reader.cursor + DOCUMENT_HEADER_SIZE;	8,192✔
310	const { dataSize } = this.readDocumentHeader(reader.buffer, reader.cursor, position, size);	8,192✔
311
312	// TODO: This should only be checked on opening
313	const writeSize = this.documentWriteSize(dataSize);	8,184✔
314	if (position + writeSize > this.size) {	8,184✔
315	throw new CorruptFileError(`Invalid document at position ${position}. This may be caused by an unfinished write.`);	8✔
316	}
317
318	if (dataSize + DOCUMENT_HEADER_SIZE > reader.buffer.byteLength) {	8,176✔
319	//console.log('sync read for large document size', dataLength, 'at position', position);
320	const tempReadBuffer = Buffer.allocUnsafe(dataSize);	4✔
321	fs.readSync(this.fd, tempReadBuffer, 0, dataSize, this.headerSize + position + DOCUMENT_HEADER_SIZE);	4✔
322	return tempReadBuffer.toString('utf8');	4✔
323	}
324
325	if (reader.cursor > 0 && dataPosition + dataSize > reader.length) {	8,172!
326	this.fillBuffer(position);	×
327	dataPosition = DOCUMENT_HEADER_SIZE;	×
328	}
329
330	return reader.buffer.toString('utf8', dataPosition, dataPosition + dataSize);	8,172✔
331	}
332
333	/**
334	* Find the start position of the document that precedes the given position.
335	*
336	* @protected
337	* @param {number} position The file position to read backwards from.
338	* @returns {number\|boolean} The start position of the first document before the given position or false if no header could be found.
339	*/
340	findDocumentPositionBefore(position) {
341	assert(this.fd, 'Partition is not opened.');	616✔
342	position -= (position % DOCUMENT_ALIGNMENT);	616✔
343	if (position <= 0) {	616✔
344	return false;	12✔
345	}
346
347	const separatorSize = DOCUMENT_SEPARATOR.length;	604✔
348	// Optimization if we are at an exact document boundary, where we can just read the document size
349	let reader = this.prepareReadBufferBackwards(position);	604✔
350	const block = reader.buffer.toString('ascii', reader.cursor - separatorSize, reader.cursor);	604✔
351	if (block === DOCUMENT_SEPARATOR) {	604✔
352	const dataSize = reader.buffer.readUInt32BE(reader.cursor - separatorSize - 4);	592✔
353	return position - this.documentWriteSize(dataSize);	592✔
354	}
355
356	do {	12✔
357	reader = this.prepareReadBufferBackwards(position - separatorSize);	28✔
358
359	const bufferSeparatorPosition = reader.buffer.lastIndexOf(DOCUMENT_SEPARATOR, reader.cursor - separatorSize, 'ascii');	28✔
360	if (bufferSeparatorPosition >= 0) {	28✔
361	position = this.readBufferPos + bufferSeparatorPosition + separatorSize;	12✔
362	break;	12✔
363	}
364	position -= this.readBufferLength;	16✔
365	} while (position > 0);
366	return position;	12✔
367	}
368
369	/**
370	* @api
371	* @param {number} [after] The document position to start reading from.
372	* @returns {Generator<string>} A generator that returns all documents in this partition.
373	*/
374	*readAll(after = 0) {	12✔
375	let position = after < 0 ? this.size + after + 1 : after;	16✔
376	let data;
377	while ((data = this.readFrom(position)) !== false) {	16✔
378	yield data;	604✔
379	position += this.documentWriteSize(Buffer.byteLength(data, 'utf8'));	604✔
380	}
381	}
382
383	/**
384	* @api
385	* @param {number} [before] The document position to start reading backward from.
386	* @returns {Generator<string>} A generator that returns all documents in this partition in reverse order.
387	*/
388	*readAllBackwards(before = -1) {	4✔
389	let position = before < 0 ? this.size + before + 1 : before;	12✔
390	while ((position = this.findDocumentPositionBefore(position)) !== false) {	12✔
391	const data = this.readFrom(position);	600✔
392	yield data;	600✔
393	}
394	}
395	}
396
397	module.exports = ReadablePartition;	4✔
398	module.exports.CorruptFileError = CorruptFileError;	4✔
399	module.exports.InvalidDataSizeError = InvalidDataSizeError;	4✔
400	module.exports.HEADER_MAGIC = HEADER_MAGIC;	4✔
401	module.exports.DOCUMENT_SEPARATOR = DOCUMENT_SEPARATOR;	4✔
402	module.exports.DOCUMENT_ALIGNMENT = DOCUMENT_ALIGNMENT;	4✔
403	module.exports.DOCUMENT_HEADER_SIZE = DOCUMENT_HEADER_SIZE;	4✔
404	module.exports.DOCUMENT_FOOTER_SIZE = DOCUMENT_FOOTER_SIZE;	4✔

albe / node-event-storage / 606483609

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