feat(service-job): leader-elect scheduled cron/interval jobs across the cluster (#2219)

CronJobAdapter fired every scheduled tick on EVERY replica, so a cron job ran
N times in an N-node cluster (duplicate side effects). Scheduled fires now
acquire a per-job cluster lock (fail-fast) before running — only the node that
wins runs the handler; peers skip. With no cluster wired or the in-memory
driver (single process) the lock always succeeds, so single-node behaviour is
unchanged. Manual trigger() bypasses the gate (explicit, node-local).
JobServicePlugin injects the cluster service into the adapter.

Co-authored-by: Jack Zhuang <277994282+os-zhuang@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

packages/services/service-job/src/cron-job-adapter.leader.test.ts

@@ -0,0 +1,51 @@
+import { describe, it, expect, vi } from 'vitest';
+import { CronJobAdapter } from './cron-job-adapter.js';
+
+const HOUR = 3_600_000;
+const grants = () => ({ acquire: vi.fn(async () => ({ release: vi.fn(async () => {}) })) });
+const denies = () => ({ acquire: vi.fn(async () => null) });
+
+describe('CronJobAdapter — scheduler leader-election', () => {
+  it('runs the handler when the cluster lock is acquired (leader)', async () => {
+    const handler = vi.fn(async () => {});
+    const adapter = new CronJobAdapter({ cluster: { lock: grants() } });
+    await adapter.schedule('j', { type: 'interval', intervalMs: HOUR }, handler);
+    await (adapter as any).runScheduled('j');
+    expect(handler).toHaveBeenCalledTimes(1);
+    await adapter.destroy();
+  });
+  it('skips the handler when the lock is held by another node', async () => {
+    const handler = vi.fn(async () => {});
+    const adapter = new CronJobAdapter({ cluster: { lock: denies() } });
+    await adapter.schedule('j', { type: 'interval', intervalMs: HOUR }, handler);
+    await (adapter as any).runScheduled('j');
+    expect(handler).not.toHaveBeenCalled();
+    await adapter.destroy();
+  });
+  it('runs without a cluster (single-node, unchanged behaviour)', async () => {
+    const handler = vi.fn(async () => {});
+    const adapter = new CronJobAdapter();
+    await adapter.schedule('j', { type: 'interval', intervalMs: HOUR }, handler);
+    await (adapter as any).runScheduled('j');
+    expect(handler).toHaveBeenCalledTimes(1);
+    await adapter.destroy();
+  });
+  it('manual trigger() bypasses leader-election and always runs', async () => {
+    const handler = vi.fn(async () => {});
+    const adapter = new CronJobAdapter({ cluster: { lock: denies() } });
+    await adapter.schedule('j', { type: 'interval', intervalMs: HOUR }, handler);
+    await adapter.trigger('j');
+    expect(handler).toHaveBeenCalledTimes(1);
+    await adapter.destroy();
+  });
+  it('releases the lock after the scheduled run', async () => {
+    const release = vi.fn(async () => {});
+    const acquire = vi.fn(async () => ({ release }));
+    const adapter = new CronJobAdapter({ cluster: { lock: { acquire } } });
+    await adapter.schedule('j', { type: 'interval', intervalMs: HOUR }, vi.fn(async () => {}));
+    await (adapter as any).runScheduled('j');
+    expect(acquire).toHaveBeenCalledWith('job:j', { ttlMs: 60000, waitMs: 0 });
+    expect(release).toHaveBeenCalledTimes(1);
+    await adapter.destroy();
+  });
+});

packages/services/service-job/src/cron-job-adapter.ts

@@ -8,6 +8,13 @@ import type {
   JobExecution,
 } from '@objectstack/spec/contracts';
 
+/** Minimal cluster lock surface for scheduler leader-election (structural — no hard dep on the cluster contract). */
+interface SchedulerCluster {
+  lock?: {
+    acquire(key: string, opts?: { ttlMs?: number; waitMs?: number }): Promise<{ release(): Promise<void> } | null>;
+  };
+}
+
 /**
  * Configuration for the cron-based job adapter.
  */
@@ -16,6 +23,12 @@ export interface CronJobAdapterOptions {
   timezone?: string;
   /** Maximum execution history per job (default: 100) */
   maxExecutions?: number;
+  /** Cluster service for scheduler leader-election. With a remote driver only ONE
+   * node fires each scheduled job; with the in-memory driver the lock always
+   * succeeds so single-node behaviour is unchanged. */
+  cluster?: SchedulerCluster;
+  /** Lease TTL (ms) held while a scheduled fire runs. Default 60000. */
+  leaseMs?: number;
 }
 
 interface CronJobRecord {
@@ -37,10 +50,14 @@ export class CronJobAdapter implements IJobService {
   private readonly defaultTimezone: string;
   private readonly maxExecutions: number;
   private readonly jobs = new Map<string, CronJobRecord>();
+  private readonly cluster?: SchedulerCluster;
+  private readonly leaseMs: number;
 
   constructor(options: CronJobAdapterOptions = {}) {
     this.defaultTimezone = options.timezone ?? 'UTC';
     this.maxExecutions = options.maxExecutions ?? 100;
+    this.cluster = options.cluster;
+    this.leaseMs = options.leaseMs ?? 60_000;
   }
 
   async schedule(name: string, schedule: JobSchedule, handler: JobHandler): Promise<void> {
@@ -55,18 +72,18 @@ export class CronJobAdapter implements IJobService {
       const task = new Cron(
         schedule.expression,
         { timezone: schedule.timezone ?? this.defaultTimezone, name },
-        async () => { await this.execute(record); },
+        async () => { await this.runScheduled(name); },
       );
       record.task = task;
     } else if (schedule.type === 'interval' && schedule.intervalMs) {
-      const handle = setInterval(() => { void this.execute(record); }, schedule.intervalMs);
+      const handle = setInterval(() => { void this.runScheduled(name); }, schedule.intervalMs);
       (handle as any)?.unref?.();
       // Use a sentinel Cron-like shape with stop() for cancel()
       record.task = { stop: () => clearInterval(handle) } as unknown as Cron;
     } else if (schedule.type === 'once' && schedule.at) {
       const delay = new Date(schedule.at).getTime() - Date.now();
       if (delay > 0) {
-        const handle = setTimeout(() => { void this.execute(record); }, delay);
+        const handle = setTimeout(() => { void this.runScheduled(name); }, delay);
         (handle as any)?.unref?.();
         record.task = { stop: () => clearTimeout(handle) } as unknown as Cron;
       }
@@ -107,6 +124,26 @@ export class CronJobAdapter implements IJobService {
     this.jobs.clear();
   }
 
+  /**
+   * Run a SCHEDULED fire of `name` under cluster leader-election: only the node
+   * that acquires the per-job lock runs the handler; peers skip. No cluster /
+   * in-memory driver => lock always granted => single-node unchanged. Manual
+   * `trigger()` bypasses this.
+   */
+  private async runScheduled(name: string): Promise<void> {
+    const record = this.jobs.get(name);
+    if (!record) return;
+    const lock = this.cluster?.lock;
+    if (!lock) { await this.execute(record); return; }
+    const handle = await lock.acquire(`job:${name}`, { ttlMs: this.leaseMs, waitMs: 0 });
+    if (!handle) return; // another node is the leader for this fire
+    try {
+      await this.execute(record);
+    } finally {
+      try { await handle.release(); } catch { /* ignore */ }
+    }
+  }
+
   private async execute(record: CronJobRecord, data?: unknown): Promise<void> {
     const execution: JobExecution = {
       jobId: record.name,

packages/services/service-job/src/job-service-plugin.ts

@@ -16,6 +16,11 @@ import {
 /**
  * Configuration options for the JobServicePlugin.
  */
+/** Resolve the cluster service if present; undefined on single-node. */
+function getClusterSafe(ctx: any): any {
+  try { return ctx.getService('cluster'); } catch { return undefined; }
+}
+
 export interface JobServicePluginOptions {
   /**
    * Job adapter type.
@@ -99,7 +104,7 @@ export class JobServicePlugin implements Plugin {
     }
 
     if (choice === 'cron') {
-      const cron = new CronJobAdapter({ timezone: 'UTC' });
+      const cron = new CronJobAdapter({ timezone: 'UTC', cluster: getClusterSafe(ctx) });
       ctx.registerService('job', cron);
       ctx.logger.info('JobServicePlugin: registered CronJobAdapter');
       return;
@@ -129,7 +134,7 @@ export class JobServicePlugin implements Plugin {
       let cron: CronJobAdapter | undefined;
       if (this.options.enableCron !== false) {
         try {
-          cron = new CronJobAdapter({ timezone: 'UTC' });
+          cron = new CronJobAdapter({ timezone: 'UTC', cluster: getClusterSafe(ctx) });
         } catch (err) {
           ctx.logger.warn('JobServicePlugin: cron adapter init failed; cron jobs will not auto-run', err as any);
         }