diff --git a/.changeset/job-once-schedule-leader-election.md b/.changeset/job-once-schedule-leader-election.md new file mode 100644 index 0000000000..4272953c5c --- /dev/null +++ b/.changeset/job-once-schedule-leader-election.md @@ -0,0 +1,35 @@ +--- +"@objectstack/service-job": patch +--- + +fix(service-job): leader-elect `type: 'once'` schedules on `DbJobAdapter` (#13918) + +`DbJobAdapter.schedule()` decides which adapter owns a scheduled fire, and only +`CronJobAdapter` takes the cluster lock (`runScheduled()` -> `lock.acquire('job:' ++ name, { waitMs: 0 })`). `cron` schedules were routed there from the start and +`interval` schedules since #13686 — but `once` schedules still went to the inner +`IntervalJobAdapter`, a bare `setTimeout` with no lock. On a multi-replica +deployment a one-shot job therefore ran **once per replica**, not once per +cluster, and a one-shot is the worst-shaped of the three: there is no later tick +during which a business-level de-duplication marker could win, so every replica's +copy lands inside the same short window. + +`once` now takes the same leader-elected path as `cron` and `interval` whenever a +cron adapter is assembled, and stays registered on the inner adapter via +`register()` (stored, not armed), so `trigger()`, `replay()`, `getExecutions()` +and `listJobs()` answer for it exactly as before. Affected paths in this repo: +the automation wait-node's timer resume and its cold-boot re-arm +(`@objectstack/service-automation`), schedule-triggered flows with an `at` +(`@objectstack/trigger-schedule`), and app-declared jobs with a `once` schedule +(`@objectstack/runtime`). + +**Behaviour change, on multi-replica assemblies only.** A `once` job now fires on +one replica per cluster instead of on every replica. Single-node behaviour is +unchanged in both assemblies: with a cron adapter and no cluster driver the lock +is always granted, and with no cron adapter at all the job still fires on the +inner timer exactly as before. The semantics are **at-most-once per cluster** +(maintainer ruling 2026-09-01): election decides who fires, not that the fire +survives — a leader that dies mid-fire loses it, and nothing re-arms it. That +takes nothing away, because the previous unelected `setTimeout` was not persisted +either and the same crash lost it on every replica at once. No re-arm, retry or +persistence mechanism is added. diff --git a/packages/services/service-job/src/db-job-adapter.once-leader.test.ts b/packages/services/service-job/src/db-job-adapter.once-leader.test.ts new file mode 100644 index 0000000000..16f7d361d1 --- /dev/null +++ b/packages/services/service-job/src/db-job-adapter.once-leader.test.ts @@ -0,0 +1,369 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #13918 — a `once` schedule registered on `DbJobAdapter` must reach the SAME + * leader-elected fire path `cron` (#2219) and `interval` (#13686) reach. + * + * ⚠️ What this file can and cannot measure — identical to its `interval` + * sibling, and for the same reason. The defect is a concurrency defect across + * OS processes and this harness is one process, so nothing here is a cluster + * test. What IS pinned deterministically: the ROUTING (which adapter received + * the registration, and that exactly one timer exists per job in one process), + * and the LOCK SEMANTICS at the adapter seam (two adapter instances contending + * for one fire against one shared lock ⇒ one execution, the loser SKIPPING + * rather than throwing, waiting or retrying). + * + * The one-shot's own asymmetry, ruled at-most-once per cluster (maintainer + * 2026-09-01): there is no later tick, so a leader that dies mid-fire loses the + * fire outright. Nothing re-arms it and nothing here pretends otherwise — the + * pins below assert the fire happens ONCE, never that it is redelivered. + */ + +import { describe, it, expect, vi, afterEach } from 'vitest'; +import type { IJobService, JobSchedule } from '@objectstack/spec/contracts'; +import { assertEngineUpdateDispatch } from '@objectstack/metadata-core'; +import { DbJobAdapter } from './db-job-adapter.js'; +import { CronJobAdapter } from './cron-job-adapter.js'; + +const TICK = 60_000; + +/** + * A `once` schedule due `ms` from NOW — built inside the test so it is read off + * the fake clock vitest installed, never off the wall clock. + */ +function onceIn(ms: number): JobSchedule { + return { type: 'once', at: new Date(Date.now() + ms).toISOString() }; +} + +function makeFakeEngine() { + const tables = new Map(); + return { + tables, + rows(table: string) { return tables.get(table) ?? []; }, + async find(table: string, opts: any = {}) { + const t = tables.get(table) ?? []; + const matched = opts.where + ? t.filter((r) => Object.entries(opts.where).every(([k, v]) => { + // REFUSE what this double does not implement, rather than reading a + // combinator as if it were a column name. + if (k.startsWith('$')) throw new Error(`fake engine: unsupported operator ${k}`); + return r[k] === v; + })) + : [...t]; + // The caller's bound, applied AFTER the filter and by PRESENCE: a `limit` + // of zero is a bound of zero rows, not an absent bound. + return typeof opts?.limit === 'number' ? matched.slice(0, opts.limit) : matched; + }, + async insert(table: string, data: any) { + const t = tables.get(table) ?? []; + t.push({ ...data }); + tables.set(table, t); + return { id: data.id }; + }, + async update(table: string, data: any, options?: Record) { + // Hold this double to ObjectQL.update's own dispatch rule, so it cannot be + // looser than the engine `DbJobAdapter` really writes through. + assertEngineUpdateDispatch(data, options); + const r = (tables.get(table) ?? []).find((x) => x.id === data.id); + if (r) Object.assign(r, data); + return r; + }, + }; +} + +/** + * A cron adapter that RECORDS what it was handed and owns no clock of its own. + * It is the routing probe: with it in place, anything that still fires came + * from a timer `DbJobAdapter` armed somewhere else. + */ +function recordingCron() { + const calls: Array<{ name: string; schedule: JobSchedule }> = []; + const svc: IJobService & { calls: typeof calls } = { + calls, + async schedule(name: string, schedule: JobSchedule) { calls.push({ name, schedule }); }, + async cancel() {}, + async trigger() {}, + async getExecutions() { return []; }, + async listJobs() { return []; }, + }; + return svc; +} + +/** One lock shared by every simulated replica — the redis fence's stand-in. */ +function sharedLock() { + const held = new Set(); + const acquire = vi.fn(async (key: string) => { + if (held.has(key)) return null; // another node is the leader for this fire + held.add(key); + return { release: vi.fn(async () => { held.delete(key); }) }; + }); + return { lock: { acquire }, acquire, held }; +} + +const denies = () => ({ acquire: vi.fn(async () => null) }); + +const built: Array<{ destroy(): Promise }> = []; +function track }>(a: T): T { built.push(a); return a; } + +afterEach(async () => { + while (built.length) await built.pop()!.destroy(); + vi.useRealTimers(); +}); + +describe('DbJobAdapter — once schedules are leader-elected (#13918)', () => { + it('routes a once registration to the cron (leader-electing) adapter, and arms no timer of its own', async () => { + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const cron = recordingCron(); + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, cron })); + const schedule = onceIn(TICK); + + await db.schedule('kickoff', schedule, handler); + + // The routing itself — asserted at the seam, not against the wall clock. + expect(cron.calls).toEqual([{ name: 'kickoff', schedule }]); + + // …and NOTHING else armed a timer. The probe owns no clock, so ten ticks + // must produce zero runs; a second, unelected `setTimeout` inside `inner` + // would show up here as one. + await vi.advanceTimersByTimeAsync(TICK * 10); + expect(handler).not.toHaveBeenCalled(); + + // The registration is still reachable through the adapter's own surface. + expect(await db.listJobs()).toEqual(['kickoff']); + }); + + it('one process holds exactly ONE timer for a delegated once job: the deadline runs the handler once', async () => { + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const acquire = vi.fn(async () => ({ release: vi.fn(async () => {}) })); + const cron = new CronJobAdapter({ cluster: { lock: { acquire } } }); + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, cron })); + + await db.schedule('wait_timer', onceIn(TICK), handler); + + await vi.advanceTimersByTimeAsync(TICK); + expect(handler).toHaveBeenCalledTimes(1); + // A one-shot stays a one-shot: no later deadline, and no re-arm (ruled + // at-most-once per cluster, so nothing redelivers it either). + await vi.advanceTimersByTimeAsync(TICK * 5); + expect(handler).toHaveBeenCalledTimes(1); + expect(acquire).toHaveBeenCalledTimes(1); + expect(acquire).toHaveBeenCalledWith('job:wait_timer', { ttlMs: 60000, waitMs: 0 }); + }); + + it('THE CARD PIN — two simulated replicas, ONE deadline: exactly one execution and ONE run row', async () => { + // One engine and one lock, two adapter stacks — the shared postgres + redis + // of a 3-replica deployment, minus the process boundary this harness has no + // way to cross. Today (unrouted) this executes twice and writes two rows. + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const fence = sharedLock(); + // The winner HOLDS its lease until this opens, so the loser's acquire is + // guaranteed to land while the lock is held rather than after it is + // released — which is what makes the count below a fact about the lock and + // not about how many microtasks the timer flush happened to run. + let open!: () => void; + const lease = new Promise((resolve) => { open = resolve; }); + const handler = vi.fn(async () => { await lease; }); + const schedule = onceIn(TICK); + + const replica = () => { + const cron = new CronJobAdapter({ cluster: { lock: fence.lock } }); + return { cron, db: track(new DbJobAdapter({ engine, cron })) }; + }; + const a = replica(); + const b = replica(); + await a.db.schedule('flow_wake', schedule, handler); + await b.db.schedule('flow_wake', schedule, handler); + + // One deadline on the wall clock reaches BOTH replicas. + await vi.advanceTimersByTimeAsync(TICK); + + // SOFT on purpose, and only in this test: THIS is the card's pin, so when it + // is red its job is to state the whole of today's picture in one run — the + // execution count, the fence that was never consulted, AND the run-row count + // below ("two `sys_job_run` rows today, one after"). A hard throw on the + // first of those hides the other two. Every other assertion in this file, + // this test's own row assertions included, is hard. + expect.soft(handler, 'one deadline must execute the job once across the cluster, not once per replica').toHaveBeenCalledTimes(1); + expect.soft(fence.acquire, 'each replica must ASK the fence — an unrouted fire never consults it at all').toHaveBeenCalledTimes(2); + // waitMs:0 is the "skip", spelled structurally — a waiting acquire would let + // the loser run the same one-shot a moment later. + expect.soft(fence.acquire).toHaveBeenCalledWith('job:flow_wake', { ttlMs: 60000, waitMs: 0 }); + + open(); + await vi.advanceTimersByTimeAsync(0); + + // The durable record agrees: one deadline, ONE run row, run_count +1. + // Unrouted, this shared engine took one row per replica — and for a + // one-shot there is no later tick during which a business-level marker + // could win the race, so both land inside the same short window. + const runs = engine.rows('sys_job_run').filter((r) => r.job_name === 'flow_wake'); + expect(runs).toHaveLength(1); + expect(runs[0].status).toBe('success'); + expect(engine.rows('sys_job')[0].run_count).toBe(1); + }); + + it('the replica that loses the lock SKIPS: it resolves, it does not throw and it does not retry', async () => { + const engine = makeFakeEngine(); + const fence = sharedLock(); + const handler = vi.fn(async () => {}); + + const replica = () => { + const cron = new CronJobAdapter({ cluster: { lock: fence.lock } }); + return { cron, db: track(new DbJobAdapter({ engine, cron })) }; + }; + const a = replica(); + const b = replica(); + const schedule = onceIn(TICK); + await a.db.schedule('flow_wake', schedule, handler); + await b.db.schedule('flow_wake', schedule, handler); + + // Driven at the fire seam so both promises are observable: `b` calls acquire + // while `a` still holds the lease. + const fireA = (a.cron as any).runScheduled('flow_wake'); + const fireB = (b.cron as any).runScheduled('flow_wake'); + await expect(Promise.all([fireA, fireB])).resolves.toHaveLength(2); + + expect(handler).toHaveBeenCalledTimes(1); + expect(fence.acquire).toHaveBeenCalledTimes(2); + expect(fence.held.has('job:flow_wake'), 'the winner must release its lease when the fire ends').toBe(false); + }); + + it('single-replica, no cluster driver: the once job still fires (the regression that would be worse than the defect)', async () => { + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const cron = new CronJobAdapter(); // no `cluster` — nothing to elect against + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, cron })); + + await db.schedule('kickoff', onceIn(TICK), handler); + + await vi.advanceTimersByTimeAsync(TICK * 3); + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('no cron adapter assembled: the once job still fires on the inner timer, exactly as before', async () => { + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const logger = { info: vi.fn(), warn: vi.fn(), error: vi.fn() }; + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, logger })); + + await db.schedule('kickoff', onceIn(TICK), handler); + + await vi.advanceTimersByTimeAsync(TICK * 2); + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('a once deadline already in the past arms nothing, with or without a cron adapter', async () => { + vi.useFakeTimers(); + const past = onceIn(-TICK); + + const withCron = (() => { + const engine = makeFakeEngine(); + const acquire = vi.fn(async () => ({ release: vi.fn(async () => {}) })); + const cron = new CronJobAdapter({ cluster: { lock: { acquire } } }); + const handler = vi.fn(async () => {}); + return { acquire, handler, db: track(new DbJobAdapter({ engine, cron })) }; + })(); + const withoutCron = (() => { + const engine = makeFakeEngine(); + const handler = vi.fn(async () => {}); + return { handler, db: track(new DbJobAdapter({ engine })) }; + })(); + + await withCron.db.schedule('overdue', past, withCron.handler); + await withoutCron.db.schedule('overdue', past, withoutCron.handler); + + await vi.advanceTimersByTimeAsync(TICK * 5); + expect(withCron.handler).not.toHaveBeenCalled(); + expect(withCron.acquire).not.toHaveBeenCalled(); + expect(withoutCron.handler).not.toHaveBeenCalled(); + // …and it is still registered, so an operator can still trigger it by hand. + expect(await withCron.db.listJobs()).toEqual(['overdue']); + }); + + it('manual trigger() still runs on THIS node while another replica holds the lock', async () => { + const engine = makeFakeEngine(); + const cron = new CronJobAdapter({ cluster: { lock: denies() } }); + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, cron })); + + await db.schedule('flow_wake', onceIn(TICK), handler); + // Reaches `inner`, which still holds the registration: a delegated job that + // vanished from `inner` would throw `Job "…" not found` right here. + await db.trigger('flow_wake'); + + expect(handler).toHaveBeenCalledTimes(1); + expect(await db.listJobs()).toEqual(['flow_wake']); + }); + + it('replay() and getExecutions() still work for a delegated once job', async () => { + const engine = makeFakeEngine(); + const cron = new CronJobAdapter({ cluster: { lock: denies() } }); + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, cron })); + + await db.schedule('flow_wake', onceIn(TICK), handler); + await db.replay('flow_wake'); + + expect(handler).toHaveBeenCalledTimes(1); + expect((await db.getExecutions('flow_wake')).map((e) => e.status)).toEqual(['success']); + expect(engine.rows('sys_job_run').some((r) => r.trigger === 'replay')).toBe(true); + }); + + it('cancel() before the deadline stops the delegated once job on BOTH adapters', async () => { + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const acquire = vi.fn(async () => ({ release: vi.fn(async () => {}) })); + const cron = new CronJobAdapter({ cluster: { lock: { acquire } } }); + const handler = vi.fn(async () => {}); + const db = track(new DbJobAdapter({ engine, cron })); + + await db.schedule('flow_wake', onceIn(TICK), handler); + await db.cancel('flow_wake'); + + await vi.advanceTimersByTimeAsync(TICK * 5); + expect(handler).not.toHaveBeenCalled(); + expect(acquire).not.toHaveBeenCalled(); + expect(await db.listJobs()).toEqual([]); + expect(engine.rows('sys_job')[0]).toMatchObject({ name: 'flow_wake', active: false }); + }); + + it('still upserts the sys_job row for a delegated once schedule', async () => { + vi.useFakeTimers(); + const engine = makeFakeEngine(); + const db = track(new DbJobAdapter({ engine, cron: recordingCron() })); + const schedule = onceIn(TICK); + + await db.schedule('kickoff', schedule, async () => {}); + + expect(engine.rows('sys_job')[0]).toMatchObject({ + name: 'kickoff', + schedule_type: 'once', + schedule_expression: (schedule as { at: string }).at, + active: true, + }); + }); + + it('DECLARED CONTROL — cron and interval routing are unchanged', async () => { + const engine = makeFakeEngine(); + const cron = recordingCron(); + const db = track(new DbJobAdapter({ engine, cron })); + const cronSchedule: JobSchedule = { type: 'cron', expression: '0 0 30 2 *' }; + const intervalSchedule: JobSchedule = { type: 'interval', intervalMs: TICK }; + + await db.schedule('nightly_report', cronSchedule, async () => {}); + await db.schedule('heartbeat', intervalSchedule, async () => {}); + + expect(cron.calls).toEqual([ + { name: 'nightly_report', schedule: cronSchedule }, + { name: 'heartbeat', schedule: intervalSchedule }, + ]); + expect(await db.listJobs()).toEqual(['nightly_report', 'heartbeat']); + }); +}); diff --git a/packages/services/service-job/src/db-job-adapter.ts b/packages/services/service-job/src/db-job-adapter.ts index f5e5c48d81..4d58eb1fa6 100644 --- a/packages/services/service-job/src/db-job-adapter.ts +++ b/packages/services/service-job/src/db-job-adapter.ts @@ -75,11 +75,11 @@ function uid(prefix: string): string { * DbJobAdapter — IJobService that persists job registry and execution * history to ObjectQL while delegating timer mechanics downwards. * - * Every SCHEDULED fire goes to the `cron` adapter callers supply — both - * `cron` and `interval` schedules — because that adapter is the one that - * leader-elects each fire against the cluster lock (#13686); `inner` + * Every SCHEDULED fire goes to the `cron` adapter callers supply — `cron`, + * `interval` and `once` schedules alike — because that adapter is the one that + * leader-elects each fire against the cluster lock (#13686, #13918); `inner` * (`IntervalJobAdapter`) keeps the registration for `trigger()` / `replay()` - * and owns the timer only for the schedule types nothing else can run. See + * and owns the timer only when no `cron` adapter was assembled at all. See * {@link DbJobAdapter.schedule} for the routing and its no-cron fallbacks. * * Persisted side effects: @@ -121,20 +121,32 @@ export class DbJobAdapter implements IJobService { /** * Register `name`, and decide WHICH adapter owns its scheduled fire — which - * is the same thing as deciding whether that fire is leader-elected (#13686). + * is the same thing as deciding whether that fire is leader-elected (#13686, + * #13918). * * `CronJobAdapter` is the only adapter here that holds a cluster lock, and it - * takes that lock in `runScheduled()` — the single path BOTH its cron limb and - * its interval limb fire through. `IntervalJobAdapter` has no lock at all. So on + * takes that lock in `runScheduled()` — the single path ALL THREE of its limbs + * fire through, `once` included. `IntervalJobAdapter` has no lock at all. So on * a multi-replica deployment the routing below *is* the leader election: every * schedule type this adapter hands to `inner` runs on every replica at once. * `interval` used to be one of them, which made #2219's declared "leader-elect * scheduled cron/interval jobs across the cluster" true of only its cron half — * measured in the field as one tick executing N times, its duplicate writes * visible wherever per-handler business de-duplication did not happen to cover - * them (three recipients, six notification rows, 54 ms apart). + * them (three recipients, six notification rows, 54 ms apart). `once` was the + * last one left (#13918), and the worst-shaped of the three: a one-shot has no + * later tick during which a business-level de-duplication marker could win, so + * every replica's copy lands inside the same short window. * - * Both delegated types are ALSO registered in `inner` — via + * **`once` is AT-MOST-ONCE per cluster, and deliberately so** (maintainer + * ruling 2026-09-01). Election decides *who* fires, never *that* the fire + * survives: there is no second deadline, so a leader that dies mid-fire loses + * the fire outright and nothing re-arms it. That takes nothing away — today's + * unelected `setTimeout` is not persisted either and the same crash loses it on + * every replica at once — so no re-arm, no "release the lease only on success", + * and no persistence is built here for a scenario no measured consumer has. + * + * All three delegated types are ALSO registered in `inner` — via * {@link IntervalJobAdapter.register}, which stores without arming a timer, so * one process never ends up holding an elected timer and an unelected one for * the same job. That registration is what keeps `trigger()`, `replay()`, @@ -143,12 +155,18 @@ export class DbJobAdapter implements IJobService { * an operator asking THIS node to run the job now. * * **Without a `cron` adapter** (`enableCron: false`, or its construction threw) - * the two types part company, because their fallbacks are not the same choice: a + * the types part company, because their fallbacks are not the same choice: a * `cron` schedule cannot run here at all, so it is warned about and left to - * manual triggering, whereas an `interval` schedule still fires on `inner`'s own - * timer exactly as it did before this routing existed. Unelected, so it is warned - * about too — but silently dropping a job an assembly CAN run is not an - * improvement on running it more often than intended. + * manual triggering, whereas `interval` and `once` schedules still fire on + * `inner`'s own timer exactly as they did before this routing existed — + * silently dropping a job an assembly CAN run is not an improvement on running + * it more often than intended. `interval` says so in a `warn`; `once` + * deliberately does NOT, and the asymmetry is about FREQUENCY, not about one + * being less worth saying: interval registrations are per-plugin-startup and + * countable, while `once` registrations are per-occurrence — the wait-node arms + * one per suspended flow run (`service-automation` + * `builtin/wait-node.ts`) — so the same line there is a per-run log flood, and + * the way to make everyone skim `warn` is to write one on a hot path. */ async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { const wrapped = this.wrap(name, handler, 'schedule', options); @@ -162,7 +180,7 @@ export class DbJobAdapter implements IJobService { ); // Still record in inner so trigger() works await this.inner.register(name, schedule, wrapped, downstream); - } else if (schedule.type === 'interval' && this.cron) { + } else if ((schedule.type === 'interval' || schedule.type === 'once') && this.cron) { // The leader-elected path — same one cron takes, for the same reason. await this.cron.schedule(name, schedule, wrapped, downstream); await this.inner.register(name, schedule, wrapped, downstream); diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index 3ca0671909..89a5672181 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3221,6 +3221,11 @@ "verb": "update", "pinned": 1 }, + { + "file": "packages/services/service-job/src/db-job-adapter.once-leader.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/services/service-job/src/db-job-adapter.timeout.test.ts", "verb": "update",