From ddf48063f0fd4900f5f7ec4dcbc05ed73c45feb8 Mon Sep 17 00:00:00 2001 From: Jack Zhuang <277994282+os-zhuang@users.noreply.github.com> Date: Mon, 22 Jun 2026 23:16:50 +0800 Subject: [PATCH] feat(service-job): leader-elect scheduled cron/interval jobs across the cluster MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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: Claude Opus 4.8 --- .../src/cron-job-adapter.leader.test.ts | 51 +++++++++++++++++++ .../service-job/src/cron-job-adapter.ts | 43 ++++++++++++++-- .../service-job/src/job-service-plugin.ts | 9 +++- 3 files changed, 98 insertions(+), 5 deletions(-) create mode 100644 packages/services/service-job/src/cron-job-adapter.leader.test.ts diff --git a/packages/services/service-job/src/cron-job-adapter.leader.test.ts b/packages/services/service-job/src/cron-job-adapter.leader.test.ts new file mode 100644 index 0000000000..9a8a2bfd87 --- /dev/null +++ b/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(); + }); +}); diff --git a/packages/services/service-job/src/cron-job-adapter.ts b/packages/services/service-job/src/cron-job-adapter.ts index 06b196c69d..6d93399c92 100644 --- a/packages/services/service-job/src/cron-job-adapter.ts +++ b/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 } | 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(); + 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 { @@ -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 { + 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 { const execution: JobExecution = { jobId: record.name, diff --git a/packages/services/service-job/src/job-service-plugin.ts b/packages/services/service-job/src/job-service-plugin.ts index 2ef6f8d618..28a40d2ced 100644 --- a/packages/services/service-job/src/job-service-plugin.ts +++ b/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); }