Daily jobs run once per installation: notification cleanup and the file integrity audit.

The notification cleanup ran at every start and daily on every server, reading
and rewriting the same users 24 times at a rollout; a notification added by
another server between a read and its write was lost. A run now claims a
document per job and day in dailyJobRuns, and the timer looks hourly for an
unclaimed day. The file integrity audit read 'lastRun', checked it, then wrote
it, so two servers could both scan; one atomic update now claims the run.

Continuous backup has no leader: that is documented as single-server, and the
lease it needs is not built, so #6861 item 3 stays open and this commit does
not close it.

Thanks to Raymond-Voogt and xet7 !

docs/Backup/Continuous-Backup.md

@@ -20,6 +20,18 @@ only what changes: documents, files and log lines. It writes them seconds after
 they happen, to a directory that can be another disk, an NFS share or an
 [rclone mount](Rclone/Rclone.md).
 
+## One server only, for now
+
+Continuous backup is for installations with **one WeKan server**. When it is
+switched on in the Admin Panel, every WeKan server that shares the database
+applies the setting, so with several servers (for example the replicas of a
+Kubernetes Deployment) each of them tails the oplog and writes its own archives
+into the same target directory: the same backup stream, many times, into one
+place. It has no leader: one server should run the stream and another take it
+over when that one stops (a lease), and that is not built yet
+([#6861](https://github.com/wekan/wekan/issues/6861)). With several servers, use
+the database's own backup instead.
+
 ## Licences: only copyfree code by default
 
 The built-in engines are part of WeKan and use the MIT licence. They depend

models/lib/dailyClaim.js

@@ -0,0 +1,58 @@
+'use strict';
+
+// Work that every WeKan server schedules for itself but that should happen once
+// for the whole installation (#6861). With several servers on one database - a
+// Kubernetes Deployment of 24 replicas - each of them ran its own copy: the
+// notification cleanup read and rewrote the same users 24 times at every rollout
+// (and a notification added by another server between a read and its write was
+// lost), and the file integrity audit could start on two servers at once. The
+// first server to claim the run does it; the others skip. Same pattern as
+// models/lib/dueScanClaim.js.
+//
+// Two shapes of claim, both ONE database operation, so there is no gap between
+// "is it due" and "it is mine" for a second server to slip through:
+// - claimDailyJob: a document per job and local day; a duplicate key says another
+//   server has today's run;
+// - claimIntervalRun: one document per job, claimed by an update that matches only
+//   while the last run is older than the interval.
+
+const COLLECTION = 'dailyJobRuns';
+
+function dayKey(date) {
+  const pad = n => String(n).padStart(2, '0');
+  return `${date.getFullYear()}-${pad(date.getMonth() + 1)}-${pad(date.getDate())}`;
+}
+
+const isDuplicateKey = error => !!error && (error.code === 11000 || /E11000/.test(String(error.message || '')));
+
+// db: a mongodb driver Db. Resolves true when this server should run `job` today.
+async function claimDailyJob(db, job, now = new Date()) {
+  try {
+    await db.collection(COLLECTION).insertOne({ _id: `${job}:${dayKey(now)}`, job, at: now });
+    return true;
+  } catch (error) {
+    if (isDuplicateKey(error)) return false;
+    throw error;
+  }
+}
+
+// collection: a driver Collection holding the run's document under `id`, with the
+// time of the last run in `at`. Resolves true when this server's update was the
+// one that moved the run forward: the filter matches only a missing or old `at`,
+// and with upsert a document that exists but does not match is a duplicate key.
+async function claimIntervalRun(collection, id, intervalMs, now = new Date()) {
+  const due = new Date(now.getTime() - intervalMs);
+  try {
+    const result = await collection.updateOne(
+      { _id: id, $or: [{ at: { $lt: due } }, { at: { $exists: false } }] },
+      { $set: { at: now } },
+      { upsert: true },
+    );
+    return !!result && (result.modifiedCount > 0 || result.upsertedCount > 0);
+  } catch (error) {
+    if (isDuplicateKey(error)) return false;
+    throw error;
+  }
+}
+
+module.exports = { claimDailyJob, claimIntervalRun, dayKey, COLLECTION };

server/lib/fileIntegrityScan.js

@@ -381,7 +381,12 @@ if (Meteor.isServer && process.env.WEKAN_INTEGRITY_SCAN !== 'false') {
         if (!isDue(last && last.at && new Date(last.at).getTime(), Date.now())) return;
         if (Number(cpuPercent()) >= PACING.maxCpuPercent) return;  // try again next hour
 
-        await IntegrityKeys.upsertAsync({ _id: 'lastRun' }, { $set: { at: new Date() } });
+        // One atomic update claims the run: the read above only avoids a write
+        // when it is plainly not due. Two servers whose hourly check falls at the
+        // same moment both read "due"; only the one whose update matches scans
+        // (#6861).
+        const { claimIntervalRun } = require('/models/lib/dailyClaim');
+        if (!(await claimIntervalRun(IntegrityKeys.rawCollection(), 'lastRun', PACING.intervalMs))) return;
         const { runWhenCpuLow } = require('/server/lib/cpuMonitor');
         await runWhenCpuLow('filesystem integrity audit',
           () => runIntegrityScan(), { maxWaitMs: 55 * 60 * 1000 });

server/models/users.js

@@ -2372,19 +2372,28 @@ const startNotificationCleanup = debounce(
       return;
     }
 
-    const oneDayMs = 24 * 60 * 60 * 1000;
+    // Once a day for the whole installation, not once per server and per
+    // rollout (#6861): the first server to claim today's run does it, the rest
+    // skip. The timer looks hourly whether today's run is still unclaimed, so
+    // the cleanup happens daily even when no server restarts.
+    const oneHourMs = 60 * 60 * 1000;
+    const runClaimed = async () => {
+      const { claimDailyJob } = require('/models/lib/dailyClaim');
+      if (!(await claimDailyJob(Users.rawDatabase(), 'notification-cleanup'))) return;
+      await runNotificationCleanup();
+    };
 
     Meteor.defer(() => {
-      runNotificationCleanup().catch(error => {
+      runClaimed().catch(error => {
         console.error('Notification cleanup failed:', error);
       });
     });
 
     notificationCleanupIntervalId = Meteor.setInterval(() => {
-      runNotificationCleanup().catch(error => {
+      runClaimed().catch(error => {
         console.error('Notification cleanup failed:', error);
       });
-    }, oneDayMs);
+    }, oneHourMs);
   },
   500,
 );

tests/dailyClaim.test.cjs

@@ -0,0 +1,115 @@
+'use strict';
+
+// #6861: work every server scheduled for itself runs once per installation.
+// Run: node tests/dailyClaim.test.cjs
+
+const assert = require('assert');
+const fs = require('fs');
+const path = require('path');
+const { claimDailyJob, claimIntervalRun, dayKey, COLLECTION } = require('../models/lib/dailyClaim');
+
+const read = rel => fs.readFileSync(path.join(__dirname, '..', rel), 'utf8');
+let passed = 0;
+async function test(name, fn) { await fn(); passed += 1; console.log('  ok -', name); }
+const dup = () => Object.assign(new Error('E11000 duplicate key error'), { code: 11000 });
+
+// A database whose inserts are unique on _id, like MongoDB's.
+function fakeDb() {
+  const docs = new Map();
+  return { docs, collection: () => ({ async insertOne(doc) { await null; if (docs.has(doc._id)) throw dup(); docs.set(doc._id, doc); } }) };
+}
+// One document per job; updateOne is atomic like MongoDB's, and a document that
+// exists but does not match the filter is a duplicate key under upsert.
+function fakeIntervalCollection(initial) {
+  const store = new Map(initial ? [[initial._id, { ...initial }]] : []);
+  return {
+    store,
+    async updateOne(filter, update, options = {}) {
+      await null;
+      const doc = store.get(filter._id);
+      const matches = doc && filter.$or.some(branch => ('at' in branch && branch.at.$lt
+        ? doc.at && doc.at < branch.at.$lt : branch.at && branch.at.$exists === false ? !('at' in doc) : false));
+      if (matches) { Object.assign(doc, update.$set); return { modifiedCount: 1, upsertedCount: 0 }; }
+      if (doc) { if (options.upsert) throw dup(); return { modifiedCount: 0, upsertedCount: 0 }; }
+      if (!options.upsert) return { modifiedCount: 0, upsertedCount: 0 };
+      store.set(filter._id, { _id: filter._id, ...update.$set });
+      return { modifiedCount: 0, upsertedCount: 1 };
+    },
+  };
+}
+
+(async () => {
+  await test('24 servers claiming the same day: exactly one runs the job', async () => {
+    const db = fakeDb();
+    const now = new Date(2026, 10, 2, 8, 0, 3);
+    const results = await Promise.all(Array.from({ length: 24 }, () => claimDailyJob(db, 'notification-cleanup', now)));
+    assert.strictEqual(results.filter(Boolean).length, 1);
+    assert.deepStrictEqual([...db.docs.keys()], ['notification-cleanup:2026-11-02']);
+    assert.strictEqual(COLLECTION, 'dailyJobRuns');
+  });
+
+  await test('jobs and days are independent', async () => {
+    const db = fakeDb();
+    assert.strictEqual(await claimDailyJob(db, 'a', new Date(2026, 10, 2, 8)), true);
+    assert.strictEqual(await claimDailyJob(db, 'b', new Date(2026, 10, 2, 8)), true, 'another job, same day');
+    assert.strictEqual(await claimDailyJob(db, 'a', new Date(2026, 10, 3, 8)), true, 'the next day');
+    assert.strictEqual(await claimDailyJob(db, 'a', new Date(2026, 10, 3, 23)), false, 'the same day again');
+    assert.strictEqual(dayKey(new Date(2026, 0, 5, 23, 59)), '2026-01-05');
+  });
+
+  await test('negative: any other database error is not swallowed', async () => {
+    const down = { collection: () => ({ insertOne: async () => { throw Object.assign(new Error('down'), { code: 6 }); } }) };
+    await assert.rejects(() => claimDailyJob(down, 'a'), /down/);
+    const broken = { updateOne: async () => { throw new Error('down'); } };
+    await assert.rejects(() => claimIntervalRun(broken, 'lastRun', 1000), /down/);
+  });
+
+  await test('an interval run: 24 servers at the same moment, exactly one scans', async () => {
+    const day = 24 * 3600 * 1000;
+    const now = new Date(2026, 10, 2, 8);
+    const coll = fakeIntervalCollection({ _id: 'lastRun', at: new Date(now.getTime() - day - 1000) });
+    const results = await Promise.all(Array.from({ length: 24 }, () => claimIntervalRun(coll, 'lastRun', day, now)));
+    assert.strictEqual(results.filter(Boolean).length, 1);
+    assert.strictEqual(coll.store.get('lastRun').at.getTime(), now.getTime());
+  });
+
+  await test('an interval run: not due yet is not claimed; never run before is', async () => {
+    const day = 24 * 3600 * 1000;
+    const now = new Date(2026, 10, 2, 8);
+    const recent = fakeIntervalCollection({ _id: 'lastRun', at: new Date(now.getTime() - 3600 * 1000) });
+    assert.strictEqual(await claimIntervalRun(recent, 'lastRun', day, now), false);
+    const fresh = fakeIntervalCollection(null);
+    const many = await Promise.all(Array.from({ length: 24 }, () => claimIntervalRun(fresh, 'lastRun', day, now)));
+    assert.strictEqual(many.filter(Boolean).length, 1, 'with no document at all, still exactly one');
+    const noAt = fakeIntervalCollection({ _id: 'lastRun' });
+    assert.strictEqual(await claimIntervalRun(noAt, 'lastRun', day, now), true, 'a document without a time is due');
+  });
+
+  await test('the notification cleanup goes through the claim, at start and on its timer', async () => {
+    const src = read('server/models/users.js');
+    const start = src.slice(src.indexOf('const startNotificationCleanup = debounce('), src.indexOf('// Repair avatar URLs'));
+    assert.ok(start.includes("claimDailyJob(Users.rawDatabase(), 'notification-cleanup')"), 'claims before it runs');
+    assert.ok(/Meteor\.defer\(\(\) => \{\s+runClaimed\(\)/.test(start), 'at start');
+    assert.ok(/Meteor\.setInterval\(\(\) => \{\s+runClaimed\(\)[\s\S]*?\}, oneHourMs\)/.test(start), 'on the hourly timer');
+    // negative: nothing in the scheduler runs the cleanup unclaimed, or on a bare 24-hour timer
+    assert.ok(!/runNotificationCleanup\(\)\.catch/.test(start), 'no unclaimed run');
+    assert.ok(!/oneDayMs/.test(start), 'no 24-hour timer per server');
+  });
+
+  await test('the integrity audit claims with one atomic update before it scans', async () => {
+    const src = read('server/lib/fileIntegrityScan.js');
+    const job = src.slice(src.indexOf('const CHECK_EVERY_MS'));
+    const claim = job.indexOf("claimIntervalRun(IntegrityKeys.rawCollection(), 'lastRun', PACING.intervalMs)");
+    assert.ok(claim > 0 && claim < job.indexOf('runWhenCpuLow('), 'claim, then scan');
+    assert.ok(!/IntegrityKeys\.upsertAsync\(\{ _id: 'lastRun' \}/.test(src), 'the read-then-write claim is gone');
+  });
+
+  await test('negative, anywhere in the server: no other 24-hour timer re-runs a job on every server unclaimed', async () => {
+    // The two jobs of this issue; a third found later belongs in the claim too.
+    for (const file of ['server/models/users.js', 'server/lib/fileIntegrityScan.js']) {
+      assert.ok(!/setInterval\([^]*?24 \* 60 \* 60 \* 1000\)/.test(read(file).replace(/\/\/.*$/gm, '')), file);
+    }
+  });
+
+  console.log(`dailyClaim.test.cjs: ${passed} tests passed`);
+})().catch(e => { console.error(e); process.exit(1); });