diff --git a/index.js b/index.js index b3fa954..132b08c 100644 --- a/index.js +++ b/index.js @@ -52,15 +52,17 @@ const write = (filepath, data, options, callback) => { throw new Error('File already exists: ' + destpath); } - const promise = mkdir(path.dirname(destpath), { recursive: true, ...options }) - .then(() => { - return new Promise((resolve, reject) => { - fs.createWriteStream(destpath, opts) - .on('error', err => reject(err)) - .on('close', resolve) - .end(ensureNewline(data, opts)); + const promise = enqueueWrite(destpath, () => { + return mkdir(path.dirname(destpath), { recursive: true, ...options }) + .then(() => { + return new Promise((resolve, reject) => { + fs.createWriteStream(destpath, opts) + .on('error', err => reject(err)) + .on('close', resolve) + .end(ensureNewline(data, opts)); + }); }); - }); + }); if (typeof callback === 'function') { promise.then(() => callback(null, result)).catch(callback); @@ -140,6 +142,28 @@ write.stream = (filepath, options) => { return fs.createWriteStream(destpath, opts); }; +/** + * Serialize async writes to the same absolute path so concurrent + * write() calls cannot interleave. write.sync uses writeFileSync and + * completes before returning, so two sync writes cannot interleave. + */ + +const pendingWrites = new Map(); + +const enqueueWrite = (filepath, task) => { + const key = path.resolve(filepath); + const previous = pendingWrites.get(key) || Promise.resolve(); + const current = previous.then(() => task(), () => task()); + pendingWrites.set(key, current); + const cleanup = () => { + if (pendingWrites.get(key) === current) { + pendingWrites.delete(key); + } + }; + current.then(cleanup, cleanup); + return current; +}; + /** * Increment the filename if the file already exists and enabled by the user */ diff --git a/test/test.js b/test/test.js index aab6af2..c5fa6ec 100644 --- a/test/test.js +++ b/test/test.js @@ -91,6 +91,63 @@ describe('write', () => { assert.equal('Hello!\n', contents.toString()); } }); + + it('should serialize concurrent writes to the same filepath', async() => { + const fp = tmp('race.txt'); + // A larger first write and a smaller second write reproduces #13 + // without 100MB buffers: without a queue the smaller write finishes + // first, then the large write overwrites (or tears) the file. + const first = Buffer.alloc(2 * 1024 * 1024, 0x61); + const second = Buffer.alloc(64 * 1024, 0x62); + let contentsAfterFirst; + + const started = write(fp, first, { overwrite: true }).then(result => { + contentsAfterFirst = fs.readFileSync(fp); + return result; + }); + const queuedLast = write(fp, second, { overwrite: true }); + + await Promise.all([started, queuedLast]); + + assert.ok(contentsAfterFirst.equals(first), 'each write finishes completely before the next starts'); + assert.ok(fs.readFileSync(fp).equals(second), 'last-queued write wins after serialization'); + }); + + it('should serialize concurrent writes that use different path strings', async() => { + const fp = tmp('race-alias.txt'); + const alias = path.relative(process.cwd(), fp); + const first = Buffer.from('first-alias-write'); + const second = Buffer.from('second-alias-write'); + + await Promise.all([ + write(fp, first, { overwrite: true }), + write(alias, second, { overwrite: true }) + ]); + + assert.equal(fs.readFileSync(fp, 'utf8'), second.toString()); + }); + + it('should run the next write if the previous write fails', async() => { + const fp = tmp('after-error.txt'); + const orig = fs.createWriteStream; + let count = 0; + + fs.createWriteStream = function(dest, options) { + const stream = orig.call(this, dest, options); + if (++count === 1) { + process.nextTick(() => stream.destroy(new Error('boom'))); + } + return stream; + }; + + try { + await assert.rejects(() => write(fp, 'nope')); + await write(fp, 'recovered'); + assert.equal(fs.readFileSync(fp, 'utf8'), 'recovered'); + } finally { + fs.createWriteStream = orig; + } + }); }); describe('sync', () => {