diff --git a/.changeset/job-timeout-durable-record.md b/.changeset/job-timeout-durable-record.md new file mode 100644 index 0000000000..1748e52720 --- /dev/null +++ b/.changeset/job-timeout-durable-record.md @@ -0,0 +1,44 @@ +--- +"@objectstack/service-job": patch +--- + +fix(service-job): a timed-out job run is recorded as `timeout`, not `success` (#7734) + +A job declared with `timeout: 2000` whose handler ran for 10 s persisted +`sys_job_run.status: 'success'` with `duration_ms` ≈ 10000 — five times the +declared limit — and left `sys_job.last_status: success`, `failure_count: 0`. +The scheduler did the right thing at runtime (it abandoned the attempt and +retried it); only the record an operator reads was wrong, which is the worse +half: the `timeout` verdict existed solely in the in-memory `JobExecution` +history that `sys_job_run` never reads. + +**The race.** `DbJobAdapter.schedule()` wrapped the handler in its recorder and +handed the *wrapper* to the timer adapter, which applied the `runWithPolicy` +timeout guard around it. So the guard raced the recorder rather than the +handler: when the guard won, the recorder's own `await handler(ctx)` was still +pending on a handler JavaScript cannot cancel, and whenever that finally +resolved it wrote `success` over the run. The same seam is why every +`sys_job_run.attempt` read `1` — the recorder ran once per attempt but had no +way to know which attempt it was. + +**The fix.** The party that observes the timeout is now the party that records +it. `runWithPolicy` takes an optional per-attempt `JobAttemptRecorder` +(`onAttemptStart` / `onAttemptSettled`, reporting `timedOut` at the instant the +guard fires), and `DbJobAdapter` runs the policy itself and records from those +callbacks. An abandoned attempt's late value loses the race and reaches no +observer, so it has no path to the row at all; a per-run latch keeps that +one-terminal-write-per-row invariant explicit. The timer adapters receive a +registration with `retryPolicy`/`timeout` removed, since the wrapper above them +now applies both. + +- `sys_job_run.status` is `timeout` for a run that blew its limit, with the + guard's message in `error` and `duration_ms` measuring the abandoned attempt. +- `sys_job.last_status` is `timeout` and `failure_count` increments: a run that + never finished is a failure, and alerting keys on that count. +- `sys_job_run.attempt` carries the real attempt number, so a retry lands `2`. +- `replay()`'s synthetic row now mirrors any terminal status of the run it + replayed (it already did this for `degraded`), instead of pairing an honest + `timeout` row with a `success` one. + +Additive: a handler that finishes inside its timeout, or that carries no +`timeout`/`retryPolicy` at all, records exactly what it recorded before. diff --git a/packages/services/service-job/src/db-job-adapter.timeout.test.ts b/packages/services/service-job/src/db-job-adapter.timeout.test.ts new file mode 100644 index 0000000000..955452c217 --- /dev/null +++ b/packages/services/service-job/src/db-job-adapter.timeout.test.ts @@ -0,0 +1,229 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { assertEngineUpdateDispatch } from '@objectstack/metadata-core'; +import { DbJobAdapter } from './db-job-adapter.js'; +import { CronJobAdapter } from './cron-job-adapter.js'; + +/** + * #7734 — a job that blows its `timeout` must say so in the DURABLE record. + * + * Every assertion here reads a `sys_job_run` / `sys_job` cell, never the + * in-memory `JobExecution` history. That is the whole point of the card: the + * in-memory "records status 'timeout'" assertion in `interval-job-adapter.test.ts` + * stayed green throughout the defect, because the timeout was computed in a + * place `sys_job_run` never reads. An operator reading the run log saw + * `status: 'success'` with a `duration_ms` five times the declared `timeout`. + */ + +function makeFakeEngine() { + const tables = new Map(); + return { + tables, + async find(table: string, opts: any = {}) { + const t = tables.get(table) ?? []; + let out = opts.where + ? t.filter((r) => Object.entries(opts.where).every(([k, v]) => r[k] === v)) + : [...t]; + if (opts.limit) out = out.slice(0, opts.limit); + return out; + }, + 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, patch: any, options?: any) { + assertEngineUpdateDispatch(patch, options); + const t = tables.get(table) ?? []; + const r = t.find((x) => x.id === patch.id); + if (!r) throw new Error(`row ${patch.id} not in ${table}`); + Object.assign(r, patch); + return r; + }, + }; +} + +const CRON = { type: 'cron', expression: '* * * * *' } as const; +const TIMEOUT_MS = 20; +const HANDLER_MS = 300; + +/** A handler that outlives its timeout, then resolves — the reported symptom. */ +function slowHandler() { + const state = { calls: 0, resolved: 0 }; + const handler = async () => { + state.calls++; + await new Promise((resolve) => { + const t = setTimeout(() => { state.resolved++; resolve(); }, HANDLER_MS); + (t as any)?.unref?.(); + }); + }; + return { state, handler }; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => { + const t = setTimeout(resolve, ms); + (t as any)?.unref?.(); + }); +} + +describe('DbJobAdapter — a timed-out run is recorded as one (#7734)', () => { + let engine: ReturnType; + let adapter: DbJobAdapter; + + beforeEach(() => { + engine = makeFakeEngine(); + adapter = new DbJobAdapter({ engine }); + }); + afterEach(async () => { await adapter.destroy(); }); + + const runRows = () => engine.tables.get('sys_job_run') ?? []; + const jobRow = () => (engine.tables.get('sys_job') ?? [])[0]; + + it('persists sys_job_run.status = "timeout", not "success"', async () => { + const { handler } = slowHandler(); + await adapter.schedule('slow', CRON, handler, { timeout: TIMEOUT_MS }); + await adapter.trigger('slow'); + + expect(runRows()).toHaveLength(1); + // The cell an operator reads. Before #7734 this said 'success'. + expect(runRows()[0].status).toBe('timeout'); + expect(runRows()[0].error).toMatch(/timed out after 20ms/); + expect(runRows()[0].completed_at).toBeTruthy(); + }); + + it('counts the timeout as a failure on sys_job', async () => { + const { handler } = slowHandler(); + await adapter.schedule('slow', CRON, handler, { timeout: TIMEOUT_MS }); + await adapter.trigger('slow'); + + expect(jobRow().last_status).toBe('timeout'); + expect(jobRow().last_error).toMatch(/timed out after 20ms/); + // A run abandoned mid-flight is a failure — alerting keys on this count. + expect(jobRow().failure_count).toBe(1); + expect(jobRow().run_count).toBe(1); + }); + + it('records the ABANDONED duration, not how long the handler kept running', async () => { + const { handler } = slowHandler(); + await adapter.schedule('slow', CRON, handler, { timeout: TIMEOUT_MS }); + await adapter.trigger('slow'); + + // The symptom row carried duration_ms ≈ the handler's full runtime, which + // is only possible if the recorder waited for the abandoned handler. + expect(runRows()[0].duration_ms).toBeLessThan(HANDLER_MS); + }); + + // ── the overwrite race, head-on ────────────────────────────────────────── + + it('a handler that resolves AFTER the guard fired cannot overwrite the timeout row', async () => { + const { state, handler } = slowHandler(); + await adapter.schedule('late', CRON, handler, { timeout: TIMEOUT_MS }); + await adapter.trigger('late'); + + expect(runRows()[0].status).toBe('timeout'); + expect(state.resolved).toBe(0); // the handler is still running right now + + // Let the abandoned handler run to completion — this is the window in + // which the old wrapper wrote `finishRun(runId, 'success')` over the row. + await sleep(HANDLER_MS * 2); + expect(state.resolved).toBe(1); + + expect(runRows()).toHaveLength(1); + expect(runRows()[0].status).toBe('timeout'); + expect(jobRow().last_status).toBe('timeout'); + expect(jobRow().run_count).toBe(1); + expect(jobRow().failure_count).toBe(1); + }); + + // ── attempt numbering ──────────────────────────────────────────────────── + + it('a retried timeout persists attempt 2 on its second row', async () => { + const { state, handler } = slowHandler(); + await adapter.schedule('retried', CRON, handler, { + timeout: TIMEOUT_MS, + retryPolicy: { maxRetries: 1, backoffMs: 1 }, + }); + await adapter.trigger('retried'); + + expect(state.calls).toBe(2); // initial + one retry + expect(runRows()).toHaveLength(2); + // Every row used to read `attempt: 1` — the number was hardcoded. + expect(runRows().map((r) => r.attempt)).toEqual([1, 2]); + expect(runRows().map((r) => r.status)).toEqual(['timeout', 'timeout']); + expect(jobRow().failure_count).toBe(2); + }); + + it('a retried FAILURE numbers its attempts too', async () => { + let calls = 0; + await adapter.schedule('flaky', CRON, async () => { + calls++; + if (calls < 3) throw new Error('boom'); + }, { retryPolicy: { maxRetries: 3, backoffMs: 1 } }); + await adapter.trigger('flaky'); + + expect(runRows().map((r) => r.attempt)).toEqual([1, 2, 3]); + expect(runRows().map((r) => r.status)).toEqual(['failed', 'failed', 'success']); + }); + + // ── additivity ─────────────────────────────────────────────────────────── + + it('a handler that finishes inside its timeout is unchanged: success, attempt 1', async () => { + await adapter.schedule('quick', CRON, async () => {}, { timeout: 60_000 }); + await adapter.trigger('quick'); + + expect(runRows()[0].status).toBe('success'); + expect(runRows()[0].attempt).toBe(1); + expect(runRows()[0].error).toBeNull(); + expect(jobRow().last_status).toBe('success'); + expect(jobRow().failure_count).toBe(0); + }); + + it('the in-memory execution and the persisted row report the SAME verdict', async () => { + const { handler } = slowHandler(); + await adapter.schedule('agree', CRON, handler, { timeout: TIMEOUT_MS }); + await adapter.trigger('agree'); + + const [exec] = await adapter.getExecutions('agree'); + expect(exec.status).toBe('timeout'); + expect(runRows()[0].status).toBe('timeout'); + expect(await adapter.listExecutionsByStatus('timeout')).toHaveLength(1); + expect(await adapter.listExecutionsByStatus('success')).toEqual([]); + }); + + it('replay of a timing-out job writes NO success row', async () => { + const { handler } = slowHandler(); + await adapter.schedule('rp', CRON, handler, { timeout: TIMEOUT_MS }); + await adapter.replay('rp'); + + // One synthetic `replay` row + one wrapped row, and they must agree. + expect(runRows().map((r) => r.trigger).sort()).toEqual(['replay', 'schedule']); + expect(runRows().map((r) => r.status)).toEqual(['timeout', 'timeout']); + }); +}); + +describe('the timeout policy still applies through an injected cron adapter (#7734)', () => { + it('a cron-scheduled run lands a timeout row even though the adapter no longer sees the policy', async () => { + // DbJobAdapter now runs `retryPolicy`/`timeout` itself and hands the timer + // adapter a policy-free registration. If that stripping ever outran the + // wrapper that replaces it, this run would record `success`. + const engine = makeFakeEngine(); + const cron = new CronJobAdapter(); + const adapter = new DbJobAdapter({ engine, cron }); + const { handler } = slowHandler(); + + await adapter.schedule('cronic', CRON, handler, { timeout: TIMEOUT_MS }); + await cron.trigger('cronic'); // fire the copy the cron adapter holds + + const runs = engine.tables.get('sys_job_run') ?? []; + expect(runs).toHaveLength(1); + expect(runs[0].status).toBe('timeout'); + expect((engine.tables.get('sys_job') ?? [])[0].failure_count).toBe(1); + expect((await cron.getExecutions('cronic'))[0].status).toBe('timeout'); + + await adapter.destroy(); + await cron.destroy(); + }); +}); diff --git a/packages/services/service-job/src/db-job-adapter.ts b/packages/services/service-job/src/db-job-adapter.ts index 0889a7fcea..454eec1996 100644 --- a/packages/services/service-job/src/db-job-adapter.ts +++ b/packages/services/service-job/src/db-job-adapter.ts @@ -5,9 +5,11 @@ import type { JobSchedule, JobHandler, JobExecution, + JobRunOutcome, JobScheduleOptions, } from '@objectstack/spec/contracts'; import { IntervalJobAdapter } from './interval-job-adapter.js'; +import { runWithPolicy } from './run-with-policy.js'; const JOB_TABLE = 'sys_job'; const RUN_TABLE = 'sys_job_run'; @@ -33,6 +35,25 @@ export interface DbJobAdapterOptions { recordRuns?: boolean; } +/** Terminal statuses a finished run can land in — everything except `running`. */ +type TerminalStatus = 'success' | 'failed' | 'degraded' | 'timeout'; + +/** + * The options handed DOWN to the timer adapter. + * + * `retryPolicy` and `timeout` are deliberately NOT forwarded (#7734): this + * adapter now runs the policy itself, inside {@link DbJobAdapter.wrap}, which + * is what lets the recorder observe a timeout at the instant it happens. A + * second `runWithPolicy` downstream would race that whole retry sequence + * against one more timeout budget and abandon the retries mid-flight. + * Every other option keeps flowing through untouched. + */ +function withoutPolicy(options?: JobScheduleOptions): JobScheduleOptions | undefined { + if (!options) return options; + const { retryPolicy: _retryPolicy, timeout: _timeout, ...rest } = options; + return Object.keys(rest).length > 0 ? (rest as JobScheduleOptions) : undefined; +} + function uid(prefix: string): string { const g: any = globalThis as any; if (g.crypto?.randomUUID) return `${prefix}_${g.crypto.randomUUID()}`; @@ -78,17 +99,19 @@ export class DbJobAdapter implements IJobService { // ── IJobService ────────────────────────────────────────────────── async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { - const wrapped = this.wrap(name, handler, 'schedule'); + const wrapped = this.wrap(name, handler, 'schedule', options); + // The wrapper OWNS `retryPolicy`/`timeout` from here down — see withoutPolicy. + const downstream = withoutPolicy(options); if (schedule.type === 'cron') { - if (this.cron) await this.cron.schedule(name, schedule, wrapped, options); + if (this.cron) await this.cron.schedule(name, schedule, wrapped, downstream); else this.logger?.warn?.( `DbJobAdapter: cron schedule registered for "${name}" without CronJobAdapter — job will only run via manual trigger`, ); // Still record in inner so trigger() works - await this.inner.schedule(name, schedule, wrapped, options); + await this.inner.schedule(name, schedule, wrapped, downstream); } else { - await this.inner.schedule(name, schedule, wrapped, options); + await this.inner.schedule(name, schedule, wrapped, downstream); } await this.upsertJobRow(name, schedule, true); @@ -131,9 +154,17 @@ export class DbJobAdapter implements IJobService { // replayed degraded run would land TWO rows disagreeing with each other, // one `degraded` and one `success`, which is the very defect this card // fixes wearing a `trigger: 'replay'` tag. + // #7734 widened this from `degraded` to every terminal status for the + // same reason #5548 introduced it: a replayed run that timed out (or + // threw — `executeJob` swallows the error, so the catch below never sees + // it) used to land a `success` row next to the wrapper's honest one. const [last] = await this.inner.getExecutions(name, 1); - if (last?.status === 'degraded') await this.finishRun(runId, 'degraded', last.error); - else await this.finishRun(runId, 'success'); + const status = last?.status; + if (status === 'degraded' || status === 'timeout' || status === 'failed') { + await this.finishRun(runId, status, last.error); + } else { + await this.finishRun(runId, 'success'); + } } catch (err) { await this.finishRun(runId, 'failed', err instanceof Error ? err.message : String(err)); throw err; @@ -180,6 +211,29 @@ export class DbJobAdapter implements IJobService { * | resolves `undefined` | `success` | `success` | no | * | resolves `{ outcome: 'completed' }` | `success` | `success` | no | * | resolves `{ outcome: 'degraded', reason? }` | `degraded` (reason in `error`) | `degraded`, `failure_count` **flat** | no | + * | exceeds its `timeout` | `timeout` (guard message in `error`) | `timeout`, `failure_count` +1 | yes, by the retry policy | + * + * **Who runs the policy, and why it has to be this one** (#7734). The + * wrapper does not merely record around the handler — it runs the handler + * *under* `runWithPolicy` and records from that policy's per-attempt + * {@link JobAttemptRecorder}. Until this wiring, the wrapper was handed to + * the timer adapter, which applied `withTimeout` around the WRAPPER: the + * guard rejected, the adapter noted `timeout` in its in-memory history, and + * the wrapper's own `await handler(ctx)` — still pending on a handler + * JavaScript cannot cancel — resolved minutes later and wrote + * `finishRun(runId, 'success')` over the run an operator was looking at. + * A row saying `success` with a `duration_ms` five times the declared + * `timeout` was the visible symptom. + * + * Recording from inside the policy closes that race by construction: the + * abandoned attempt's eventual value loses `Promise.race` and reaches no + * observer at all. The per-run `settled` latch below is the second lock on + * the same door — one terminal write per `sys_job_run` row, so no path that + * anyone adds later can move a run off a terminal status. + * + * It also makes the attempt number REAL: `runWithPolicy` counts the attempts, + * so a retry lands `sys_job_run.attempt: 2` instead of the `1` every row used + * to carry. * * Until this wiring, "the handler did not throw" WAS the success criterion, * so a handler that degraded internally — #5529's wait-wake shot into an @@ -192,35 +246,59 @@ export class DbJobAdapter implements IJobService { * Retry is untouched — it keys on a *rejected* promise only, so a `degraded` * run never re-runs (`spec/contracts/job-service.ts`, the JobHandler TSDoc). */ - private wrap(name: string, handler: JobHandler, defaultTrigger: 'schedule' | 'manual' | 'replay'): JobHandler { + private wrap( + name: string, + handler: JobHandler, + defaultTrigger: 'schedule' | 'manual' | 'replay', + options?: JobScheduleOptions, + ): JobHandler { return async (ctx) => { - const runId = this.recordRuns ? await this.startRun(name, defaultTrigger) : undefined; - const startMs = Date.now(); - try { - const outcome = await handler(ctx); - if (outcome && outcome.outcome === 'degraded') { - // Not a failure: the reason rides the existing `error` column and - // `failure_count` stays flat (decided on #7072, recorded in the - // `JobExecutionStatus` TSDoc). A reader must gate on `status` - // before reading that column as a failure. - const reason = outcome.reason; - if (runId) await this.finishRun(runId, 'degraded', reason, Date.now() - startMs); - await this.bumpJob(name, 'degraded', reason); - return outcome; - } - if (runId) await this.finishRun(runId, 'success', undefined, Date.now() - startMs); - await this.bumpJob(name, 'success'); - return outcome; - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - if (runId) await this.finishRun(runId, 'failed', msg, Date.now() - startMs); - await this.bumpJob(name, 'failed', msg); - throw err; - } + // Per-FIRE state: one of these closures exists per invocation, so two + // overlapping fires of the same job never share a run row. + let current: { id?: string; settled: boolean } | undefined; + + const settle = async (status: TerminalStatus, error?: string, durationMs?: number) => { + const run = current; + // One terminal write per row. Nothing reaches here twice today; the + // latch is what keeps that true as this seam grows (#7734). + if (!run || run.settled) return; + run.settled = true; + if (run.id) await this.finishRun(run.id, status, error, durationMs); + await this.bumpJob(name, status, error); + }; + + return await runWithPolicy(name, () => handler(ctx), options, { + onAttemptStart: async (attempt) => { + current = { id: this.recordRuns ? await this.startRun(name, defaultTrigger, attempt) : undefined, settled: false }; + }, + onAttemptSettled: async (_attempt, result) => { + if (result.ok) { + const outcome = result.value; + if (outcome && outcome.outcome === 'degraded') { + // Not a failure: the reason rides the existing `error` column and + // `failure_count` stays flat (decided on #7072, recorded in the + // `JobExecutionStatus` TSDoc). A reader must gate on `status` + // before reading that column as a failure. + await settle('degraded', outcome.reason, result.durationMs); + } else { + await settle('success', undefined, result.durationMs); + } + return; + } + const msg = result.error instanceof Error ? result.error.message : String(result.error); + // `timeout` vs `failed` is the distinction `JobExecutionStatus` has + // always declared and the durable record never carried. + await settle(result.timedOut ? 'timeout' : 'failed', msg, result.durationMs); + }, + }); }; } - private async startRun(jobName: string, trigger: 'schedule' | 'manual' | 'replay'): Promise { + private async startRun( + jobName: string, + trigger: 'schedule' | 'manual' | 'replay', + attempt = 1, + ): Promise { const id = uid('run'); const now = new Date().toISOString(); try { @@ -230,7 +308,7 @@ export class DbJobAdapter implements IJobService { status: 'running', started_at: now, trigger, - attempt: 1, + attempt, created_at: now, }, { context: SYSTEM_CTX }); return id; @@ -335,8 +413,13 @@ export class DbJobAdapter implements IJobService { * `last_error` carries the degraded `reason`, per the column decision * recorded in the `JobExecutionStatus` TSDoc (#7072): no new column, and the * "Error" label only reads correctly alongside `last_status`. + * + * `timeout` (#7734) is the mirror image of `degraded` here: it IS a failure — + * the run did not finish, the policy retries it — so it bumps `failure_count` + * alongside `failed`. Alerting that keys on that count is the reason a job + * stuck at five times its declared `timeout` must not read as a quiet success. */ - private async bumpJob(name: string, last_status: 'success' | 'failed' | 'degraded', last_error?: string): Promise { + private async bumpJob(name: string, last_status: TerminalStatus, last_error?: string): Promise { try { const existing = await this.engine.find(JOB_TABLE, { where: { name }, @@ -352,7 +435,8 @@ export class DbJobAdapter implements IJobService { last_status, last_error: last_status === 'success' ? null : (last_error ?? null), run_count: (row.run_count ?? 0) + 1, - failure_count: (row.failure_count ?? 0) + (last_status === 'failed' ? 1 : 0), + failure_count: + (row.failure_count ?? 0) + (last_status === 'failed' || last_status === 'timeout' ? 1 : 0), updated_at: now, }, { context: SYSTEM_CTX }); } catch (err) { diff --git a/packages/services/service-job/src/run-with-policy.ts b/packages/services/service-job/src/run-with-policy.ts index 0bc0802c5a..191bd1efd1 100644 --- a/packages/services/service-job/src/run-with-policy.ts +++ b/packages/services/service-job/src/run-with-policy.ts @@ -33,6 +33,40 @@ const RETRY_DEFAULTS = { jitter: false, } as const; +/** + * How ONE attempt settled, as observed by {@link runWithPolicy} itself. + * + * `timedOut` is the whole point of the shape: it is reported at the instant the + * timeout guard wins the race, while the abandoned handler is still running. + * A recorder driven by these results therefore cannot be overtaken by a late + * handler resolution — the value an already-timed-out attempt eventually + * resolves to loses the race and never reaches an observer (#7734). + */ +export type JobAttemptResult = + | { ok: true; value: T; durationMs: number } + | { ok: false; error: unknown; timedOut: boolean; durationMs: number }; + +/** + * Per-attempt observer for {@link runWithPolicy}. + * + * Exists so the party that DURABLY RECORDS a run (`DbJobAdapter`) can be the + * same party that OBSERVES its timeout and its attempt number. Recording from + * outside the policy — wrapping the handler in a recorder and letting the guard + * race the *recorder* — records whatever the abandoned handler eventually did, + * which is how a timed-out run came to persist `status: 'success'` (#7734). + * + * Both callbacks are awaited: `onAttemptStart` before the attempt's clock + * starts (so a recorder's own I/O never eats the timeout budget), and + * `onAttemptSettled` before any retry backoff. Both are optional, and a caller + * that passes no recorder at all gets exactly the previous behaviour. + */ +export interface JobAttemptRecorder { + /** Called once per attempt, before it starts. `attempt` is 1-based. */ + onAttemptStart?(attempt: number): void | Promise; + /** Called once per attempt, the moment it settles — timeout included. */ + onAttemptSettled?(attempt: number, result: JobAttemptResult): void | Promise; +} + function sleep(ms: number): Promise { return new Promise((resolve) => { const t = setTimeout(resolve, ms); @@ -68,6 +102,12 @@ function withTimeout(run: () => Promise, jobId: string, timeoutMs?: number * what stops a fleet of jobs that failed on the same outage from retrying in * lockstep. * + * An optional {@link JobAttemptRecorder} observes each attempt from INSIDE the + * policy (#7734). It is the only way to learn that an attempt timed out at the + * moment it did: `withTimeout` abandons the attempt but cannot cancel the + * handler, so anything watching from outside eventually sees the abandoned + * handler's own result and mistakes it for the run's. + * * Generic in the run's resolved value, defaulting to `void` (#6617). Every * existing caller passes a `() => Promise` and still infers `T = void`, * so this is purely additive — what changed is that the wrapper no longer @@ -82,29 +122,42 @@ export async function runWithPolicy( jobId: string, run: () => Promise, options?: JobScheduleOptions, + recorder?: JobAttemptRecorder, ): Promise { const timeoutMs = options?.timeout; - if (!options?.retryPolicy) { - return withTimeout(run, jobId, timeoutMs); - } + const policy = options?.retryPolicy; - const maxRetries = options.retryPolicy.maxRetries ?? RETRY_DEFAULTS.maxRetries; - const backoffMs = options.retryPolicy.backoffMs ?? RETRY_DEFAULTS.backoffMs; - const multiplier = options.retryPolicy.backoffMultiplier ?? RETRY_DEFAULTS.backoffMultiplier; - const maxRetryDelayMs = options.retryPolicy.maxRetryDelayMs ?? RETRY_DEFAULTS.maxRetryDelayMs; - const jitter = options.retryPolicy.jitter ?? RETRY_DEFAULTS.jitter; + // No policy ⇒ maxRetries 0 ⇒ the loop below runs exactly one attempt and + // rethrows its error: the legacy single-shot path, expressed once so the + // recorder is driven identically with and without a retry policy. + const maxRetries = policy ? (policy.maxRetries ?? RETRY_DEFAULTS.maxRetries) : 0; + const backoffMs = policy?.backoffMs ?? RETRY_DEFAULTS.backoffMs; + const multiplier = policy?.backoffMultiplier ?? RETRY_DEFAULTS.backoffMultiplier; + const maxRetryDelayMs = policy?.maxRetryDelayMs ?? RETRY_DEFAULTS.maxRetryDelayMs; + const jitter = policy?.jitter ?? RETRY_DEFAULTS.jitter; let lastError: unknown; - for (let attempt = 0; attempt <= maxRetries; attempt++) { - if (attempt > 0) { + for (let retry = 0; retry <= maxRetries; retry++) { + if (retry > 0) { // Same formula the try_catch executor runs — one policy, one backoff. - let delay = Math.min(backoffMs * Math.pow(multiplier, attempt - 1), maxRetryDelayMs); + let delay = Math.min(backoffMs * Math.pow(multiplier, retry - 1), maxRetryDelayMs); if (jitter) delay = delay * (0.5 + Math.random() * 0.5); await sleep(delay); } + const attempt = retry + 1; // 1-based, as `sys_job_run.attempt` records it + await recorder?.onAttemptStart?.(attempt); + const startMs = Date.now(); try { - return await withTimeout(run, jobId, timeoutMs); + const value = await withTimeout(run, jobId, timeoutMs); + await recorder?.onAttemptSettled?.(attempt, { ok: true, value, durationMs: Date.now() - startMs }); + return value; } catch (err) { + await recorder?.onAttemptSettled?.(attempt, { + ok: false, + error: err, + timedOut: err instanceof JobTimeoutError, + durationMs: Date.now() - startMs, + }); lastError = err; } }