Fix(datasets): roll back a publish a timeout cuts short; skip only on a lost lock

publish took the combined signal (caller timeout + lock loss) and treated
any abort as a lost lock, so a scheduled or boot timeout firing between
renames skipped the rollback and cleanup while this process still held the
lock, leaving a mixed set and .next / .bak files. The combined signal still
cancels, but only the lock's own signal now stops rollback and cleanup.
Once every rename is done, only a lost lock keeps the version from being
recorded, so a timeout landing after a complete publish no longer forces a
re-download.

The lock-loss test now trips the engine's real onCompromised handler; a
new test covers the timeout path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CauPFca6bYL9qpuYEmsU9H

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);

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);