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

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