diff --git a/common/dataset-updater.js b/common/dataset-updater.js index 1e3858f5..bc4a0302 100644 --- a/common/dataset-updater.js +++ b/common/dataset-updater.js @@ -168,12 +168,13 @@ const acquireLock = async (row, { wait = false, signal } = {}) => { // Copy every staged file next to its target, then rename them all into // place. A failed rename rolls back the ones already done: the previous // file restored from its backup, or a target that didn't exist removed. -// `signal` (the lock lost) is checked before every step; once it fires, +// `signal` (a timeout, or the lock lost) is checked before every step and +// rolls back like a failed rename. Only `lockLost` stops the cleanup too: // another process may be publishing into the same paths, so this one -// touches nothing more — no rollback, no cleanup — and leaves the half-done -// set to the next holder (updateDataset marked the state in flux first). -const publish = async (row, staged, signal) => { - const lockLost = () => Boolean(signal?.aborted); +// touches nothing more and leaves the half-done set to the next holder +// (updateDataset marked the state in flux first). +const publish = async (row, staged, { signal, lockLost }) => { + const ownsFiles = () => !lockLost?.aborted; const plan = row.files.map((file) => { const target = path.join(row.dir, file); return { source: staged[file], target, next: `${target}.next`, backup: `${target}.bak`, existed: false, renamed: false }; @@ -196,7 +197,7 @@ const publish = async (row, staged, signal) => { step.renamed = true; } } catch (error) { - if (!lockLost()) { + if (ownsFiles()) { for (const step of plan.filter((s) => s.renamed)) { await (step.existed ? fsp.copyFile(step.backup, step.target) : fsp.rm(step.target, { force: true })) .catch(() => {}); @@ -205,7 +206,7 @@ const publish = async (row, staged, signal) => { throw error; } } finally { - if (!lockLost()) { + if (ownsFiles()) { for (const step of plan) { await fsp.rm(step.next, { force: true }); await fsp.rm(step.backup, { force: true }); @@ -249,8 +250,9 @@ export const updateDataset = async (row, { signal, reason = 'auto update', wait // publishes a whole set. work.throwIfAborted(); await writeState(row, { ...state, identifier: null }); - await publish(row, staged, work); - work.throwIfAborted(); + await publish(row, staged, { signal: work, lockLost: lock.lost }); + // A whole set is in place: record it unless the lock went meanwhile. + lock.lost.throwIfAborted(); await writeState(row, { identifier: remote.identifier, updatedAt: now, checkedAt: now }); logger.info({ dataset: row.id, identifier: remote.identifier }, 'dataset updated'); await reloadRow(row, reason); diff --git a/tests/dataset-updater.test.js b/tests/dataset-updater.test.js index e6a46fbb..d2e01b8a 100644 --- a/tests/dataset-updater.test.js +++ b/tests/dataset-updater.test.js @@ -147,7 +147,7 @@ describe('updateDataset', () => { assert.equal(read(row, 'b.txt'), 'b.txt v2\n'); }); - it('a lock lost between renames touches nothing more; the next holder republishes', async () => { + it('a timeout between renames rolls back and cleans up, as the lock is still held', async () => { const { row, log } = makeRow(); await updateDataset(row); row.version = 'v2'; @@ -155,17 +155,49 @@ describe('updateDataset', () => { const realRename = fsp.rename; fsp.rename = async (...args) => { await realRename(...args); - controller.abort(new Error('lock compromised')); + controller.abort(new Error('update timed out')); }; try { - await assert.rejects(updateDataset(row, { signal: controller.signal }), /lock compromised/); + await assert.rejects(updateDataset(row, { signal: controller.signal }), /update timed out/); } finally { fsp.rename = realRename; } + assert.equal(read(row, 'a.txt'), 'a.txt v1\n'); + assert.equal(read(row, 'b.txt'), 'b.txt v1\n'); + assert.deepEqual(log.reloads, ['auto update']); + assert.deepEqual(fs.readdirSync(row.dir).sort(), [STATE_FILE, 'a.txt', 'b.txt'], 'no .next / .bak left'); + assert.equal(state(row).identifier, null); + assert.equal((await updateDataset(row)).updated, true); + assert.equal(read(row, 'b.txt'), 'b.txt v2\n'); + }); + + it('a lock lost between renames touches nothing more; the next holder republishes', async () => { + const { row, log } = makeRow(); + await updateDataset(row); + row.version = 'v2'; + // Capture the engine's compromise handler, as proper-lockfile would call it. + const realLock = lockfile.lock; + const realRename = fsp.rename; + let compromise = null; + lockfile.lock = async (dir, options) => { + compromise = options.onCompromised; + return realLock(dir, options); + }; + fsp.rename = async (...args) => { + await realRename(...args); + compromise(new Error('lock compromised')); + }; + try { + await assert.rejects(updateDataset(row), /lock compromised/); + } finally { + lockfile.lock = realLock; + fsp.rename = realRename; + } // No rollback: another process may own these paths now. assert.equal(read(row, 'a.txt'), 'a.txt v2\n'); assert.equal(read(row, 'b.txt'), 'b.txt v1\n'); assert.deepEqual(log.reloads, ['auto update']); + assert.ok(fs.existsSync(path.join(row.dir, 'b.txt.next')), 'left to the next holder'); // The state names no version, so the next holder takes nothing for // current and publishes a whole set — not a 'not-modified' over a mix. assert.equal(state(row).identifier, null);