From 270c6355e73f662694fe42eadea0bd27dc8c0080 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 12:41:03 +0000 Subject: [PATCH 1/4] fix(service-job): leader-elect `type: 'once'` schedules on DbJobAdapter MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `DbJobAdapter.schedule()` routed `once` registrations to `inner` (`IntervalJobAdapter`), a bare `setTimeout` with no cluster lock anywhere in that file, so a one-shot job ran once per replica instead of once per cluster — the last limb left after #13686 did the same for `interval`, and the worst-shaped of the three: a one-shot has no later tick during which a business-level de-duplication marker could win. Route `once` to `this.cron` (`CronJobAdapter`, whose own `once` branch already fires through the leader-electing `runScheduled()`) when a cron adapter is assembled, and keep the registration in `inner` via `register()` so `trigger()`, `replay()`, `getExecutions()` and `listJobs()` are unaffected. No cron adapter assembled => unchanged: `inner.schedule()`, as before. Crash semantics are at-most-once per cluster, per the maintainer ruling of 2026-09-01, and stated in the docblock: no re-arm and no persistence is added. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .../src/db-job-adapter.once-leader.test.ts | 367 ++++++++++++++++++ .../service-job/src/db-job-adapter.ts | 48 ++- 2 files changed, 400 insertions(+), 15 deletions(-) create mode 100644 packages/services/service-job/src/db-job-adapter.once-leader.test.ts 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..34abbd9ba1 --- /dev/null +++ b/packages/services/service-job/src/db-job-adapter.once-leader.test.ts @@ -0,0 +1,367 @@ +// 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 here: when this pin is red the row count below + // is the other half of the card's evidence ("two `sys_job_run` rows today, + // one after"), and a hard throw here would hide it. Every other assertion + // in this file is hard. + expect.soft(handler, 'one deadline must execute the job once across the cluster, not once per replica').toHaveBeenCalledTimes(1); + expect(fence.acquire).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(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); From aea6e9a118fae28433ec4936388dfbffc4973122 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 12:45:37 +0000 Subject: [PATCH 2/4] chore(service-job): changeset for the `once` leader-election fix Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .../job-once-schedule-leader-election.md | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 .changeset/job-once-schedule-leader-election.md 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. From a6946375cc46beded70d4fa9adb9c7c6236a11ec Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 13:05:28 +0000 Subject: [PATCH 3/4] chore(scripts): record the new service-job engine double in the pinned ledger MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `node scripts/check-engine-double-contract.mjs --write` — 1 row added, 0 lost: the `update` double in the new `db-job-adapter.once-leader.test.ts`, which is already routed through `assertEngineUpdateDispatch`. Coverage growth only. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- scripts/engine-double-contract.pinned.json | 5 +++++ 1 file changed, 5 insertions(+) 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", From 51fb2de6fe2aa829e299fded5c192deceb27b3fd Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 14:26:13 +0000 Subject: [PATCH 4/4] test(service-job): let the #13918 card pin report all of today's evidence The pin exists to state the card's repro: two replicas, one deadline, two `sys_job_run` rows today and one after. A hard throw on the execution count stops the run before the fence count and the row count are ever reported, so the ablation that proves the pin can fail printed only the first of the three. The three pre-row assertions are now soft; the row assertions stay hard. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .../src/db-job-adapter.once-leader.test.ts | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) 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 index 34abbd9ba1..16f7d361d1 100644 --- 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 @@ -182,15 +182,17 @@ describe('DbJobAdapter — once schedules are leader-elected (#13918)', () => { // One deadline on the wall clock reaches BOTH replicas. await vi.advanceTimersByTimeAsync(TICK); - // SOFT on purpose, and only here: when this pin is red the row count below - // is the other half of the card's evidence ("two `sys_job_run` rows today, - // one after"), and a hard throw here would hide it. Every other assertion - // in this file is hard. + // 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(fence.acquire).toHaveBeenCalledTimes(2); + 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(fence.acquire).toHaveBeenCalledWith('job:flow_wake', { ttlMs: 60000, waitMs: 0 }); + expect.soft(fence.acquire).toHaveBeenCalledWith('job:flow_wake', { ttlMs: 60000, waitMs: 0 }); open(); await vi.advanceTimersByTimeAsync(0);