Run each large-sprint rollover and undo job on one server at a time, by lease.

The background rollover and the background undo or redo of a large batch
were resumed by every server that found them after a restart; two of them
on one job wrote alike, and the card writes' compare-and-set stopped one
with a conflict the Scrum view showed as a failure. Each step now first
takes or renews a lease on the sprint or the job, one minute long, that a
server which went away loses when it runs out. A live lease of another
server leaves the job to it: a retried close says the rollover is running
there, and undo and redo on the board wait. The lease is not History
content and is not sent to the browser.

Tests: a foreign live lease leaves both the rollover and the undo job
untouched, and an expired one is taken over.

Thanks to xet7 !

models/lib/scrumHistory.js

@@ -11,7 +11,8 @@ function historyDocument(type, doc) {
   // into a restored record, which gets a fresh one.
   // The rollover's progress is the background job's, not the sprint's
   // content: an undo never restores it.
-  const { revision, updatedAt, updatedBy, rolloverPending, rolloverTotal, rolloverDone, rolloverError, incarnation, ...content } = doc;
+  const { revision, updatedAt, updatedBy, rolloverPending, rolloverTotal, rolloverDone, rolloverError, rolloverLease, incarnation,
+    ...content } = doc;
   return structuredClone(content);
 }
 function historyRecords(changes) {

server/lib/scrumBatchJobs.js

@@ -9,11 +9,24 @@
 import { Meteor } from 'meteor/meteor';
 import { Mongo } from 'meteor/mongo';
 import { DDP } from 'meteor/ddp';
+import { Random } from 'meteor/random';
 
 export const ScrumBatchJobs = new Mongo.Collection('scrumBatchJobs');
 ScrumBatchJobs.deny({ insert: () => true, update: () => true, remove: () => true });
 
 const running = new Map();
+// One server at a time runs a board's job: each step takes or renews a lease
+// on its document, which a server that went away loses when it runs out;
+// another server's live lease means the job is running there.
+const SERVER_ID = Random.id();
+const LEASE_MS = 60000;
+async function claim(boardId) {
+  const now = new Date();
+  const { matchedCount } = await ScrumBatchJobs.rawCollection().updateOne({ _id: boardId,
+    $or: [{ lease: { $exists: false } }, { 'lease.owner': SERVER_ID }, { 'lease.until': { $lt: now } }] },
+  { $set: { lease: { owner: SERVER_ID, until: new Date(now.getTime() + LEASE_MS) } } });
+  return matchedCount === 1;
+}
 let stepper = null;
 // server/models/changeHistory.js registers how one row is applied.
 export function setBatchStepper(step) { stepper = step; }
@@ -26,7 +39,7 @@ const asUser = (userId, work) => {
 async function run(boardId) {
   for (;;) {
     const job = await ScrumBatchJobs.findOneAsync(boardId);
-    if (!job) return;
+    if (!job || !await claim(boardId)) return;
     let more;
     try { more = await asUser(job.userId, context => stepper(context, job)); }
     catch (error) {

server/lib/tests/scrumSnapshotRows.tests.js

@@ -209,6 +209,14 @@ describe('Scrum snapshot rows', function () {
       const batch = (await ChangeHistory.findOneAsync({ boardId, entityType: 'scrum', batchId: /^scrum-close-/, undone: false, isCheckpoint: { $ne: true } })).batchId;
       await ScrumBatchJobs.rawCollection().insertOne({ _id: boardId, boardId, userId: actor, direction: 'undo', batchId: batch,
         requestId: null, index: 1, done: 0, total: 99, state: 'failed', error: 'stopped' });
+      // NEGATIVE: while another server holds its lease, the job is left to it.
+      await ScrumBatchJobs.rawCollection().updateOne({ _id: boardId },
+        { $set: { lease: { owner: 'another-server', until: new Date(Date.now() + 60000) } } });
+      await assert.rejects(call('changeHistory.undoLast', boardId), /scrum-history-running/);
+      await batchJob(boardId);
+      assert.equal((await ScrumBatchJobs.findOneAsync(boardId)).done, 0, 'nothing ran here');
+      // Once the lease runs out, the next press takes it over.
+      await ScrumBatchJobs.rawCollection().updateOne({ _id: boardId }, { $set: { 'lease.until': new Date(Date.now() - 1000) } });
       await assert.rejects(call('changeHistory.undoLast', boardId), /scrum-history-running/);
       await batchJob(boardId);
       assert.equal(await ScrumBatchJobs.findOneAsync(boardId), undefined);
@@ -260,6 +268,16 @@ describe('Scrum snapshot rows', function () {
       await assert.rejects(call('scrum.saveSprint', boardId, null, { name: 'X', plannedStart: '2026-09-01', plannedEnd: '2026-12-30' }, null),
         /scrum-rollover-pending/);
       assert.equal((await call('scrum.getBoardData', boardId)).sprints.find(row => row._id === sprint._id).rolloverPending, true);
+      // NEGATIVE: while another server holds a live lease on the rollover,
+      // a retry here leaves it alone and changes nothing.
+      const foreign = { owner: 'another-server', until: new Date(Date.now() + 60000) };
+      await ScrumSprints.rawCollection().updateOne({ _id: sprint._id }, { $set: { rolloverLease: foreign } });
+      await assert.rejects(call('scrum.closeSprint', boardId, sprint._id, active.revision, null), /scrum-rollover-running/);
+      assert.equal((await ScrumSprints.findOneAsync(sprint._id)).rolloverPending, true);
+      assert.equal((await call('scrum.getBoardData', boardId)).sprints.find(row => row._id === sprint._id).rolloverLease, undefined,
+        'the lease is not sent to the browser');
+      // Once that lease runs out (the server went away), a retry takes over.
+      await ScrumSprints.rawCollection().updateOne({ _id: sprint._id }, { $set: { 'rolloverLease.until': new Date(Date.now() - 1000) } });
       // The retry of the same close finishes it.
       await call('scrum.closeSprint', boardId, sprint._id, active.revision, null);
       const finished = await ScrumSprints.findOneAsync(sprint._id);

server/scrum.js

@@ -122,6 +122,7 @@ export async function getScrumBoardData(userId, boardId) {
     sprint.report = sprint.startSnapshot || sprint.closeSnapshot ? sprintReport(sprint) : null;
     for (const name of ['startSnapshot', 'closeSnapshot']) if (sprint[name]) sprint[name] = withoutRows(sprint[name]);
     delete sprint.reportTotals;
+    delete sprint.rolloverLease;
     sprints[index] = sprint;
   }
   for (const event of events) event.followUpCardIds = (event.followUpCardIds || []).filter(id => visible.has(id));
@@ -299,7 +300,7 @@ async function rolloverStep(userId, sprint) {
     return false;
   }
   await ScrumSprints.rawCollection().updateOne({ _id: sprint._id, revision: sprint.revision, rolloverPending: true },
-    { $unset: { rolloverPending: '', rolloverTotal: '', rolloverDone: '', rolloverError: '' } });
+    { $unset: { rolloverPending: '', rolloverTotal: '', rolloverDone: '', rolloverError: '', rolloverLease: '' } });
   return true;
 }
 
@@ -326,11 +327,27 @@ export function startRollover(sprintId) {
 const INLINE_ROLLOVER = SNAPSHOT_CHUNK;
 // For tests and callers that must wait: the running job, if any.
 export function rolloverJob(sprintId) { return (rollovers.get(sprintId) || Promise.resolve()).catch(() => {}); }
+// One server at a time runs a rollover: each step first takes or renews a
+// lease on the sprint, which a server that went away loses when it runs out.
+// Another server's live lease means the job is running there, and this one
+// leaves it alone. Card writes stay compare-and-set, so a server paused past
+// its lease cannot write over the one that took over.
+const SERVER_ID = Random.id();
+const ROLLOVER_LEASE_MS = 60000;
+async function claimRollover(sprintId) {
+  const now = new Date();
+  const { matchedCount } = await ScrumSprints.rawCollection().updateOne({ _id: sprintId,
+    $or: [{ rolloverLease: { $exists: false } }, { 'rolloverLease.owner': SERVER_ID }, { 'rolloverLease.until': { $lt: now } }] },
+  { $set: { rolloverLease: { owner: SERVER_ID, until: new Date(now.getTime() + ROLLOVER_LEASE_MS) } } });
+  return matchedCount === 1;
+}
 async function runRollover(sprintId) {
+  if (!await claimRollover(sprintId)) return;
   await ScrumSprints.rawCollection().updateOne({ _id: sprintId }, { $unset: { rolloverError: '' } });
   for (;;) {
     const sprint = await ScrumSprints.findOneAsync(sprintId);
     if (!sprint || !(sprint.rolloverPending === true || Array.isArray(sprint.rolloverPending))) return;
+    if (!await claimRollover(sprintId)) return;
     const userId = sprint.updatedBy;
     const finished = await withScrumBoardLock(sprint.boardId, () => historyBatchRunner(sprint.boardId, userId,
       () => rolloverStep(userId, sprint), { batchId: closeBatchId(sprint._id, sprint.closedFromRevision) }));
@@ -473,6 +490,8 @@ const methods = {
     // History row, as always; a larger one names its batch, and its rollover
     // continues in the background under that name.
     const inline = async closed => {
+      // Another server running this rollover (its lease is live) is left to it.
+      if (!await claimRollover(closed._id)) throw new Meteor.Error('scrum-rollover-running', 'The rollover is running on another server.');
       try { while (!(await rolloverStep(userId, await ScrumSprints.findOneAsync(closed._id)))); }
       catch (error) {
         await ScrumSprints.rawCollection().updateOne({ _id: closed._id, rolloverPending: { $exists: true } },