feat(server): take a database lease before a cron tick runs

waitForCompletion only kept a tick from overlapping itself inside one
process. Two processes on the same database (a rolling update's overlap,
a second replica, a dev container pointed at the same file) both ran
every job: two auto-backups, double syncs and retention sweeps, and
reminder runs racing each other.

A new migration adds scheduler_leases, and CronRegistrarService now runs
every registered tick through SchedulerLeasesRepository.acquire: one
conditional UPDATE moves the job's row to this process when it is free
(no holder, a lapsed holder, or this process already), so exactly one
process wins. A running tick renews its lease every 20 seconds, a crashed
holder frees the job after a minute, and a finished tick keeps the lease
for 30 seconds so a peer whose timer fires late does not run the same
tick again. The per-row reminder claims stay as they are; jobs need no
change of their own.

server/src/db/entities/SchedulerLeases.entity.ts

@@ -0,0 +1,20 @@
+import { EntityRepositoryType, PrimaryKeyProp, defineEntity, p } from '@mikro-orm/core';
+import { SchedulerLeasesRepository } from '../repositories/SchedulerLeases.repository';
+
+export class SchedulerLeases {
+  [EntityRepositoryType]?: SchedulerLeasesRepository;
+  [PrimaryKeyProp]?: 'name';
+  name!: string;
+  owner!: string;
+  expires_at!: number;
+}
+
+export const SchedulerLeasesSchema = defineEntity({
+  class: SchedulerLeases,
+  repository: () => SchedulerLeasesRepository,
+  properties: {
+    name: p.text().primary(),
+    owner: p.text(),
+    expires_at: p.integer(),
+  },
+});

server/src/db/entities/index.ts

@@ -92,6 +92,7 @@ import { RoadtripDayTracksSchema } from './RoadtripDayTracks.entity';
 import { RoadtripPreferencesSchema } from './RoadtripPreferences.entity';
 import { RoadtripViasSchema } from './RoadtripVias.entity';
 import { RouteUsageDailySchema } from './RouteUsageDaily.entity';
+import { SchedulerLeasesSchema } from './SchedulerLeases.entity';
 import { SchemaVersionSchema } from './SchemaVersion.entity';
 import { SchoolHolidayCountriesSchema } from './SchoolHolidayCountries.entity';
 import { SchoolHolidayPeriodsSchema } from './SchoolHolidayPeriods.entity';
@@ -224,6 +225,7 @@ export const ALL_ENTITIES: readonly EntitySchema[] = [
   RoadtripPreferencesSchema,
   RoadtripViasSchema,
   RouteUsageDailySchema,
+  SchedulerLeasesSchema,
   SchemaVersionSchema,
   SchoolHolidayCountriesSchema,
   SchoolHolidayPeriodsSchema,

server/src/db/migrations/Migration20200101042500_a_cron_tick_runs_in_one_process.ts

@@ -0,0 +1,27 @@
+import { Migration } from '@mikro-orm/migrations';
+
+/**
+ * A cron tick runs in one process. Every job registered through
+ * CronRegistrarService takes a row here before its tick body runs: the row
+ * names the process holding the job and until when, and a process only runs
+ * the tick when one conditional UPDATE moved the row to itself. Two processes
+ * on one database (a rolling update's overlap, a second replica, a stray dev
+ * container) therefore no longer both send the reminders, take the backup or
+ * run the sync.
+ *
+ * `expires_at` is in milliseconds since the epoch. A row is only ever written
+ * by the scheduler, so there is nothing to backfill.
+ */
+export class Migration20200101042500_a_cron_tick_runs_in_one_process extends Migration {
+  override name = 'Migration20200101042500_a_cron_tick_runs_in_one_process';
+
+  override async up(): Promise<void> {
+    this.addSql(`
+      CREATE TABLE IF NOT EXISTS scheduler_leases (
+        name TEXT NOT NULL PRIMARY KEY,
+        owner TEXT NOT NULL,
+        expires_at INTEGER NOT NULL
+      )
+    `);
+  }
+}

server/src/db/repositories/SchedulerLeases.repository.ts

@@ -0,0 +1,40 @@
+import type { SchedulerLeases } from '../entities/SchedulerLeases.entity';
+import { TrekRepository } from './_shared/trek-repository';
+
+/**
+ * `scheduler_leases`: which process may run a cron job's tick, and until when
+ * (`CronRegistrarService`). A lease is free when no row exists yet, when its
+ * `expires_at` has passed, or when the asking process already holds it.
+ */
+export class SchedulerLeasesRepository extends TrekRepository<SchedulerLeases> {
+  /**
+   * Take (or renew) the lease on `name` for `owner` until `until`, if it is
+   * free at `now`. The decision is one conditional UPDATE, so of two
+   * processes asking at the same moment exactly one gets a changed row; the
+   * insert before it only makes sure there is a row to race on. True when
+   * `owner` holds the lease afterwards.
+   */
+  async acquire(name: string, owner: string, now: number, until: number): Promise<boolean> {
+    await this.upsert(
+      { name, owner, expires_at: 0 },
+      { onConflictFields: ['name'], onConflictAction: 'ignore' },
+    );
+    const changed = await this.nativeUpdate(
+      { name, $or: [{ expires_at: { $lte: now } }, { owner }] },
+      { owner, expires_at: until },
+    );
+    return changed > 0;
+  }
+
+  /** Move the expiry of a lease `owner` still holds; false when somebody else has it now. */
+  async extend(name: string, owner: string, until: number): Promise<boolean> {
+    const changed = await this.nativeUpdate({ name, owner }, { expires_at: until });
+    return changed > 0;
+  }
+
+  /** The current holder and expiry, for diagnostics and tests. */
+  async holder(name: string): Promise<{ owner: string; expires_at: number } | null> {
+    const row = await this.findOne({ name }, { fields: ['owner', 'expires_at'] });
+    return row ? { owner: row.owner, expires_at: row.expires_at } : null;
+  }
+}

server/src/nest/scheduling/cron-registrar.service.ts

@@ -1,3 +1,5 @@
+import { hostname } from 'node:os';
+import { randomUUID } from 'node:crypto';
 import { Injectable, Optional, type OnApplicationShutdown } from '@nestjs/common';
 import { SchedulerRegistry } from '@nestjs/schedule';
 import { CronJob } from 'cron';
@@ -5,8 +7,28 @@ import { MikroORM } from '@mikro-orm/core';
 import { readEnv } from '../../app-config';
 import { RuntimeEnvService } from '../app-config/runtime-env.service';
 import { withRequestContext } from '../database/request-context';
-import { logError } from '../audit/audit-log.logger';
+import { logDebug, logError } from '../audit/audit-log.logger';
 import { traceEntry, wasTraced } from '../audit/entry-trace.logger';
+import { SchedulerLeases } from '../../db/entities/SchedulerLeases.entity';
+
+/**
+ * Who holds a lease: this process. Every registrar in one process shares it,
+ * so a job re-registered (or a second app built in the same process, as the
+ * test harness does) keeps the lease it already had, while another process on
+ * the same database is a different owner.
+ */
+export const LEASE_OWNER = `${hostname()}:${process.pid}:${randomUUID()}`;
+
+/** How long a lease lasts without a heartbeat; a crashed holder frees its jobs after this. */
+export const LEASE_TTL_MS = 60_000;
+/** How often a running tick renews its lease. */
+export const LEASE_HEARTBEAT_MS = 20_000;
+/**
+ * How long a finished tick keeps its lease. Long enough that another process
+ * whose timer fires a little late for the same tick finds it taken, short
+ * enough that the next tick of a minutely job is free for whoever fires it.
+ */
+export const LEASE_SETTLE_MS = 30_000;
 
 /**
  * The one way TREK code schedules a cron. Job providers register here from
@@ -93,7 +115,7 @@ export class CronRegistrarService implements OnApplicationShutdown {
       }
       // Each tick is its own unit of work: one correlation id for every line
       // the job writes, and one line for the tick itself.
-      return traceEntry('cron', name, () => withRequestContext(orm, () => onTick()), {
+      return traceEntry('cron', name, () => withRequestContext(orm, () => this.runLeased(orm, name, onTick)), {
         failureLevel: 'error',
         failureMessage: failed,
       });
@@ -118,6 +140,38 @@ export class CronRegistrarService implements OnApplicationShutdown {
     return true;
   }
 
+  /**
+   * Run one tick only if this process holds the job's lease.
+   *
+   * waitForCompletion keeps a tick from overlapping itself inside one process;
+   * the lease does the same across processes sharing the database (a rolling
+   * update's overlap, a second replica, a dev container pointed at the same
+   * file). The lease is renewed while the tick runs, so a long backup keeps it,
+   * and held for LEASE_SETTLE_MS afterwards, so a peer whose timer fires late
+   * for the same tick does not run it again. A process that dies holding it
+   * frees the job after LEASE_TTL_MS.
+   */
+  private async runLeased(orm: Pick<MikroORM, 'em'>, name: string, onTick: () => void | Promise<void>): Promise<void> {
+    const leases = orm.em.getRepository(SchedulerLeases);
+    if (!(await leases.acquire(name, LEASE_OWNER, Date.now(), Date.now() + LEASE_TTL_MS))) {
+      logDebug(`Cron job "${name}" skipped: another process holds this tick`);
+      return;
+    }
+    const leaseFailed = (err: unknown) =>
+      logError(`Cron job "${name}": lease update failed: ${err instanceof Error ? err.message : String(err)}`);
+    const heartbeat = setInterval(() => {
+      withRequestContext(orm, () => leases.extend(name, LEASE_OWNER, Date.now() + LEASE_TTL_MS)).catch(leaseFailed);
+    }, LEASE_HEARTBEAT_MS);
+    heartbeat.unref();
+    try {
+      await onTick();
+    } finally {
+      clearInterval(heartbeat);
+      // Never let the bookkeeping replace the tick's own error.
+      await leases.extend(name, LEASE_OWNER, Date.now() + LEASE_SETTLE_MS).catch(leaseFailed);
+    }
+  }
+
   /**
    * Runs a boot-time (`onApplicationBootstrap`) one-off sweep inside the same
    * request context a registered tick gets from `register()` above — the one