diff --git a/.changeset/job-retry-timeout-3494.md b/.changeset/job-retry-timeout-3494.md new file mode 100644 index 0000000000..a1d15ba6bd --- /dev/null +++ b/.changeset/job-retry-timeout-3494.md @@ -0,0 +1,32 @@ +--- +"@objectstack/spec": minor +"@objectstack/service-job": minor +"@objectstack/runtime": minor +--- + +feat(job): honor the authored `retryPolicy` / `timeout` in the job scheduler (#3494) + +`JobSchema.retryPolicy` and `JobSchema.timeout` used to be parsed-but-ignored +(the 2026-06 liveness audit's aspirational-config cluster). They are now +enforced end to end — built rather than pruned, since retry/backoff and +per-run time limits are semantics job authors reasonably expect: + +- **spec**: `IJobService.schedule` gains an optional 4th `options` argument + (`JobScheduleOptions` with `retryPolicy` / `timeout`, mirroring the + authorable schema); new `JobRetryPolicy` type. Backward compatible — + existing 3-arg implementations and callers are unaffected. +- **service-job**: new `runWithPolicy` helper (exported, with + `JobTimeoutError`) wraps every handler invocation in `CronJobAdapter` and + `IntervalJobAdapter`; `DbJobAdapter` threads options through to its inner + adapters. Failed attempts (including timeouts) retry with exponential + backoff `backoffMs * backoffMultiplier^(retry-1)` up to `maxRetries`; + an attempt exceeding `timeout` is recorded with execution status + `'timeout'`. No `options` → exactly the legacy single-attempt behavior. +- **runtime**: declarative-jobs registration in AppPlugin forwards the + authored `retryPolicy` / `timeout` to the scheduler. + +Note: JavaScript cannot forcibly cancel an in-flight handler — a timed-out +attempt is abandoned, not killed. The retry delay caps only via the +multiplier arithmetic (no maxDelay knob yet). + +Refs #3494, #1878, #1893. diff --git a/content/docs/references/system/job.mdx b/content/docs/references/system/job.mdx index 4ee5455703..d7798616e5 100644 --- a/content/docs/references/system/job.mdx +++ b/content/docs/references/system/job.mdx @@ -62,8 +62,8 @@ const result = CronSchedule.parse(data); | **description** | `string` | optional | Job description / purpose | | **schedule** | `{ type: 'cron'; expression: string \| { dialect: Enum<'cel' \| 'cron' \| 'template'>; source?: string; ast?: any; meta?: object }; timezone?: string } \| { type: 'interval'; intervalMs: integer } \| { type: 'once'; at: string }` | ✅ | Job schedule configuration | | **handler** | `string` | ✅ | Handler function name (must match a key in `defineStack({ functions })`) | -| **retryPolicy** | `{ maxRetries?: integer; backoffMs?: integer; backoffMultiplier?: number }` | optional | Retry policy configuration. [EXPERIMENTAL — not enforced] The service-job scheduler does not yet read retryPolicy; failed runs are not retried per this config (liveness audit #1878/#1893). | -| **timeout** | `integer` | optional | Timeout in milliseconds. [EXPERIMENTAL — not enforced] The service-job scheduler does not yet enforce a per-run timeout (liveness audit #1878/#1893). | +| **retryPolicy** | `{ maxRetries?: integer; backoffMs?: integer; backoffMultiplier?: number }` | optional | Retry policy: failed runs (including timeouts) are retried with exponential backoff (delay = backoffMs * backoffMultiplier^(retry-1)) up to maxRetries retries after the initial attempt (#3494). Omit for the legacy single-attempt behavior. | +| **timeout** | `integer` | optional | Per-attempt time limit in milliseconds; an over-limit run is recorded with execution status "timeout" (#3494). The in-flight handler is abandoned, not forcibly cancelled. Omit for no time limit. | | **enabled** | `boolean` | optional | Whether the job is enabled | diff --git a/packages/runtime/src/app-plugin.ts b/packages/runtime/src/app-plugin.ts index 9fd5a3cc7e..5a5397bb8c 100644 --- a/packages/runtime/src/app-plugin.ts +++ b/packages/runtime/src/app-plugin.ts @@ -674,6 +674,10 @@ export class AppPlugin implements Plugin { async (jobCtx: any) => { await handler({ ...jobCtx, jobId: jobName, bundle: this.bundle }); }, + // #3494: thread the authored retryPolicy/timeout to the adapter + (job.retryPolicy || job.timeout) + ? { retryPolicy: job.retryPolicy, timeout: job.timeout } + : undefined, ); ok++; } catch (err: any) { diff --git a/packages/services/service-job/src/cron-job-adapter.test.ts b/packages/services/service-job/src/cron-job-adapter.test.ts index 9d4d4a9fdf..187899b82a 100644 --- a/packages/services/service-job/src/cron-job-adapter.test.ts +++ b/packages/services/service-job/src/cron-job-adapter.test.ts @@ -65,3 +65,70 @@ describe('CronJobAdapter', () => { expect(execs[0].error).toBe('boom'); }); }); + +describe('CronJobAdapter retryPolicy / timeout (#3494)', () => { + let adapter: CronJobAdapter; + afterEach(async () => { await adapter?.destroy(); }); + + it('retries a failing handler per retryPolicy and succeeds', async () => { + adapter = new CronJobAdapter(); + let calls = 0; + await adapter.schedule( + 'flaky', + { type: 'cron', expression: '* * * * *' }, + async () => { + calls++; + if (calls < 3) throw new Error(`attempt ${calls} boom`); + }, + { retryPolicy: { maxRetries: 3, backoffMs: 1, backoffMultiplier: 1 } }, + ); + await adapter.trigger('flaky'); + expect(calls).toBe(3); + const execs = await adapter.getExecutions('flaky'); + expect(execs).toHaveLength(1); + expect(execs[0].status).toBe('success'); + }); + + it('exhausts retries and records the last failure', async () => { + adapter = new CronJobAdapter(); + let calls = 0; + await adapter.schedule( + 'doomed', + { type: 'cron', expression: '* * * * *' }, + async () => { calls++; throw new Error('always boom'); }, + { retryPolicy: { maxRetries: 2, backoffMs: 1 } }, + ); + await adapter.trigger('doomed'); + expect(calls).toBe(3); // initial + 2 retries + const execs = await adapter.getExecutions('doomed'); + expect(execs[0].status).toBe('failed'); + expect(execs[0].error).toBe('always boom'); + }); + + it('does not retry when no retryPolicy is given (legacy behavior)', async () => { + adapter = new CronJobAdapter(); + let calls = 0; + await adapter.schedule('legacy', { type: 'cron', expression: '* * * * *' }, async () => { + calls++; + throw new Error('boom'); + }); + await adapter.trigger('legacy'); + expect(calls).toBe(1); + const execs = await adapter.getExecutions('legacy'); + expect(execs[0].status).toBe('failed'); + }); + + it('enforces a per-attempt timeout and records status "timeout"', async () => { + adapter = new CronJobAdapter(); + await adapter.schedule( + 'slow', + { type: 'cron', expression: '* * * * *' }, + async () => { await new Promise((r) => setTimeout(r, 150)); }, + { timeout: 25 }, + ); + await adapter.trigger('slow'); + const execs = await adapter.getExecutions('slow'); + expect(execs[0].status).toBe('timeout'); + expect(execs[0].error).toMatch(/timed out after 25ms/); + }); +}); diff --git a/packages/services/service-job/src/cron-job-adapter.ts b/packages/services/service-job/src/cron-job-adapter.ts index 6d93399c92..cdb0e3e86e 100644 --- a/packages/services/service-job/src/cron-job-adapter.ts +++ b/packages/services/service-job/src/cron-job-adapter.ts @@ -6,7 +6,9 @@ import type { JobSchedule, JobHandler, JobExecution, + JobScheduleOptions, } from '@objectstack/spec/contracts'; +import { runWithPolicy, JobTimeoutError } from './run-with-policy.js'; /** Minimal cluster lock surface for scheduler leader-election (structural — no hard dep on the cluster contract). */ interface SchedulerCluster { @@ -35,6 +37,7 @@ interface CronJobRecord { name: string; schedule: JobSchedule; handler: JobHandler; + options?: JobScheduleOptions; task?: Cron; executions: JobExecution[]; } @@ -60,10 +63,10 @@ export class CronJobAdapter implements IJobService { this.leaseMs = options.leaseMs ?? 60_000; } - async schedule(name: string, schedule: JobSchedule, handler: JobHandler): Promise { + async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { await this.cancel(name); - const record: CronJobRecord = { name, schedule, handler, executions: [] }; + const record: CronJobRecord = { name, schedule, handler, options, executions: [] }; if (schedule.type === 'cron') { if (!schedule.expression) { @@ -152,10 +155,10 @@ export class CronJobAdapter implements IJobService { }; const startMs = Date.now(); try { - await record.handler({ jobId: record.name, data }); + await runWithPolicy(record.name, () => record.handler({ jobId: record.name, data }), record.options); execution.status = 'success'; } catch (err) { - execution.status = 'failed'; + execution.status = err instanceof JobTimeoutError ? 'timeout' : 'failed'; execution.error = err instanceof Error ? err.message : String(err); } finally { execution.completedAt = new Date().toISOString(); diff --git a/packages/services/service-job/src/db-job-adapter.ts b/packages/services/service-job/src/db-job-adapter.ts index 776010e325..05354476d2 100644 --- a/packages/services/service-job/src/db-job-adapter.ts +++ b/packages/services/service-job/src/db-job-adapter.ts @@ -5,6 +5,7 @@ import type { JobSchedule, JobHandler, JobExecution, + JobScheduleOptions, } from '@objectstack/spec/contracts'; import { IntervalJobAdapter } from './interval-job-adapter.js'; @@ -76,18 +77,18 @@ export class DbJobAdapter implements IJobService { // ── IJobService ────────────────────────────────────────────────── - async schedule(name: string, schedule: JobSchedule, handler: JobHandler): Promise { + async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { const wrapped = this.wrap(name, handler, 'schedule'); if (schedule.type === 'cron') { - if (this.cron) await this.cron.schedule(name, schedule, wrapped); + if (this.cron) await this.cron.schedule(name, schedule, wrapped, options); 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); + await this.inner.schedule(name, schedule, wrapped, options); } else { - await this.inner.schedule(name, schedule, wrapped); + await this.inner.schedule(name, schedule, wrapped, options); } await this.upsertJobRow(name, schedule, true); diff --git a/packages/services/service-job/src/index.ts b/packages/services/service-job/src/index.ts index ff5827c894..c35585e6aa 100644 --- a/packages/services/service-job/src/index.ts +++ b/packages/services/service-job/src/index.ts @@ -10,3 +10,4 @@ export { DbJobAdapter } from './db-job-adapter.js'; export type { DbJobAdapterOptions, JobEngineLike, JobLoggerLike } from './db-job-adapter.js'; // JobRunRetention was retired (ADR-0057): sys_job_run declares a `lifecycle` // window and the platform LifecycleService is the one sweeper. +export { runWithPolicy, JobTimeoutError } from './run-with-policy.js'; diff --git a/packages/services/service-job/src/interval-job-adapter.test.ts b/packages/services/service-job/src/interval-job-adapter.test.ts index 6784c0171e..a935f49aae 100644 --- a/packages/services/service-job/src/interval-job-adapter.test.ts +++ b/packages/services/service-job/src/interval-job-adapter.test.ts @@ -118,3 +118,41 @@ describe('IntervalJobAdapter', () => { expect(await adapter.listJobs()).toEqual([]); }); }); + +describe('IntervalJobAdapter retryPolicy / timeout (#3494)', () => { + let adapter: IntervalJobAdapter; + afterEach(async () => { await adapter?.destroy(); }); + + it('retries a failing handler per retryPolicy and succeeds', async () => { + adapter = new IntervalJobAdapter(); + let calls = 0; + await adapter.schedule( + 'flaky', + { type: 'interval', intervalMs: 100000 }, + async () => { + calls++; + if (calls < 2) throw new Error('boom'); + }, + { retryPolicy: { maxRetries: 2, backoffMs: 1 } }, + ); + await adapter.trigger('flaky'); + expect(calls).toBe(2); + const execs = await adapter.getExecutions('flaky'); + expect(execs).toHaveLength(1); + expect(execs[0].status).toBe('success'); + }); + + it('enforces a per-attempt timeout and records status "timeout"', async () => { + adapter = new IntervalJobAdapter(); + await adapter.schedule( + 'slow', + { type: 'interval', intervalMs: 100000 }, + async () => { await new Promise((r) => setTimeout(r, 150)); }, + { timeout: 25 }, + ); + await adapter.trigger('slow'); + const execs = await adapter.getExecutions('slow'); + expect(execs[0].status).toBe('timeout'); + expect(execs[0].error).toMatch(/timed out after 25ms/); + }); +}); diff --git a/packages/services/service-job/src/interval-job-adapter.ts b/packages/services/service-job/src/interval-job-adapter.ts index f20994fdd1..93fb164f32 100644 --- a/packages/services/service-job/src/interval-job-adapter.ts +++ b/packages/services/service-job/src/interval-job-adapter.ts @@ -1,6 +1,7 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. -import type { IJobService, JobSchedule, JobHandler, JobExecution } from '@objectstack/spec/contracts'; +import type { IJobService, JobSchedule, JobHandler, JobExecution, JobScheduleOptions } from '@objectstack/spec/contracts'; +import { runWithPolicy, JobTimeoutError } from './run-with-policy.js'; /** * Internal record for a scheduled job. @@ -9,6 +10,7 @@ interface JobRecord { name: string; schedule: JobSchedule; handler: JobHandler; + options?: JobScheduleOptions; timerId?: ReturnType | ReturnType; executions: JobExecution[]; } @@ -38,11 +40,11 @@ export class IntervalJobAdapter implements IJobService { this.maxExecutions = options.maxExecutions ?? 100; } - async schedule(name: string, schedule: JobSchedule, handler: JobHandler): Promise { + async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { // Cancel any existing job with the same name await this.cancel(name); - const record: JobRecord = { name, schedule, handler, executions: [] }; + const record: JobRecord = { name, schedule, handler, options, executions: [] }; if (schedule.type === 'interval' && schedule.intervalMs) { record.timerId = setInterval(async () => { @@ -111,10 +113,10 @@ export class IntervalJobAdapter implements IJobService { const startMs = Date.now(); try { - await record.handler({ jobId: record.name, data }); + await runWithPolicy(record.name, () => record.handler({ jobId: record.name, data }), record.options); execution.status = 'success'; } catch (err) { - execution.status = 'failed'; + execution.status = err instanceof JobTimeoutError ? 'timeout' : 'failed'; execution.error = err instanceof Error ? err.message : String(err); } finally { execution.completedAt = new Date().toISOString(); diff --git a/packages/services/service-job/src/run-with-policy.ts b/packages/services/service-job/src/run-with-policy.ts new file mode 100644 index 0000000000..07c67c5357 --- /dev/null +++ b/packages/services/service-job/src/run-with-policy.ts @@ -0,0 +1,75 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { JobScheduleOptions } from '@objectstack/spec/contracts'; + +/** + * Error thrown when a job attempt exceeds its configured `timeout`. + * Executors map it to JobExecution status 'timeout' (vs plain 'failed'). + */ +export class JobTimeoutError extends Error { + constructor(jobId: string, timeoutMs: number) { + super(`Job "${jobId}" timed out after ${timeoutMs}ms`); + this.name = 'JobTimeoutError'; + } +} + +const RETRY_DEFAULTS = { maxRetries: 3, backoffMs: 1000, backoffMultiplier: 2 } as const; + +function sleep(ms: number): Promise { + return new Promise((resolve) => { + const t = setTimeout(resolve, ms); + (t as any)?.unref?.(); + }); +} + +function withTimeout(run: () => Promise, jobId: string, timeoutMs?: number): Promise { + if (!timeoutMs || timeoutMs <= 0) return run(); + let timer: ReturnType | undefined; + const guard = new Promise((_, reject) => { + timer = setTimeout(() => reject(new JobTimeoutError(jobId, timeoutMs)), timeoutMs); + (timer as any)?.unref?.(); + }); + return Promise.race([run(), guard]).finally(() => clearTimeout(timer)) as Promise; +} + +/** + * Execute one job run under the authored `retryPolicy` / `timeout` + * (#3494 — JobSchema's retryPolicy/timeout used to be parsed-but-ignored). + * + * - No options → exactly the legacy behavior: one attempt, no time limit. + * - `timeout` applies per attempt; an over-limit attempt rejects with + * {@link JobTimeoutError}. JavaScript cannot forcibly cancel the in-flight + * handler — the attempt is abandoned, not killed. + * - `retryPolicy` re-runs failed attempts (including timeouts) with + * exponential backoff: delay = backoffMs * backoffMultiplier^(retry-1), + * up to maxRetries retries after the initial attempt. The last error is + * rethrown when all attempts fail. + */ +export async function runWithPolicy( + jobId: string, + run: () => Promise, + options?: JobScheduleOptions, +): Promise { + const timeoutMs = options?.timeout; + if (!options?.retryPolicy) { + return withTimeout(run, jobId, timeoutMs); + } + + const maxRetries = options.retryPolicy.maxRetries ?? RETRY_DEFAULTS.maxRetries; + const backoffMs = options.retryPolicy.backoffMs ?? RETRY_DEFAULTS.backoffMs; + const multiplier = options.retryPolicy.backoffMultiplier ?? RETRY_DEFAULTS.backoffMultiplier; + + let lastError: unknown; + for (let attempt = 0; attempt <= maxRetries; attempt++) { + if (attempt > 0) { + await sleep(backoffMs * Math.pow(multiplier, attempt - 1)); + } + try { + await withTimeout(run, jobId, timeoutMs); + return; + } catch (err) { + lastError = err; + } + } + throw lastError; +} diff --git a/packages/spec/api-surface.json b/packages/spec/api-surface.json index 248a95a842..78e8e1d617 100644 --- a/packages/spec/api-surface.json +++ b/packages/spec/api-surface.json @@ -3606,7 +3606,9 @@ "IntrospectedTable (interface)", "JobExecution (interface)", "JobHandler (type)", + "JobRetryPolicy (interface)", "JobSchedule (interface)", + "JobScheduleOptions (interface)", "KNOWLEDGE_SERVICE (const)", "KVEntry (interface)", "KVSetOptions (interface)", diff --git a/packages/spec/src/contracts/job-service.ts b/packages/spec/src/contracts/job-service.ts index 376ef4889b..cfd355cdee 100644 --- a/packages/spec/src/contracts/job-service.ts +++ b/packages/spec/src/contracts/job-service.ts @@ -34,6 +34,36 @@ export interface JobSchedule { */ export type JobHandler = (context: { jobId: string; data?: unknown }) => Promise; +/** + * Retry policy for a scheduled job (mirrors the authorable + * `RetryPolicySchema` in system/job.zod.ts). + */ +export interface JobRetryPolicy { + /** Maximum number of retry attempts after the initial run (default 3) */ + maxRetries?: number; + /** Initial backoff delay in milliseconds (default 1000) */ + backoffMs?: number; + /** Multiplier for exponential backoff (default 2) */ + backoffMultiplier?: number; +} + +/** + * Per-job execution options threaded from the authored JobSchema + * (`retryPolicy` / `timeout`) down to the executing adapter. + * + * Omitted options preserve the legacy behavior: one attempt, no time limit. + */ +export interface JobScheduleOptions { + /** Retry failed runs with exponential backoff */ + retryPolicy?: JobRetryPolicy; + /** + * Per-attempt time limit in milliseconds. A run that exceeds it is + * recorded with status 'timeout'. Note: JavaScript cannot forcibly + * cancel the in-flight handler — the attempt is abandoned, not killed. + */ + timeout?: number; +} + /** * Status of a job execution */ @@ -58,8 +88,9 @@ export interface IJobService { * @param name - Job name (snake_case) * @param schedule - Schedule configuration * @param handler - Job handler function + * @param options - Optional per-job retry policy / timeout */ - schedule(name: string, schedule: JobSchedule, handler: JobHandler): Promise; + schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise; /** * Cancel a scheduled job diff --git a/packages/spec/src/system/job.zod.ts b/packages/spec/src/system/job.zod.ts index 6ff4912410..e8ee30c9f2 100644 --- a/packages/spec/src/system/job.zod.ts +++ b/packages/spec/src/system/job.zod.ts @@ -87,8 +87,8 @@ export const JobSchema = lazySchema(() => z.object({ description: z.string().optional().describe('Job description / purpose'), schedule: ScheduleSchema.describe('Job schedule configuration'), handler: z.string().describe('Handler function name (must match a key in `defineStack({ functions })`)'), - retryPolicy: RetryPolicySchema.optional().describe('Retry policy configuration. [EXPERIMENTAL — not enforced] The service-job scheduler does not yet read retryPolicy; failed runs are not retried per this config (liveness audit #1878/#1893).'), - timeout: z.number().int().positive().optional().describe('Timeout in milliseconds. [EXPERIMENTAL — not enforced] The service-job scheduler does not yet enforce a per-run timeout (liveness audit #1878/#1893).'), + retryPolicy: RetryPolicySchema.optional().describe('Retry policy: failed runs (including timeouts) are retried with exponential backoff (delay = backoffMs * backoffMultiplier^(retry-1)) up to maxRetries retries after the initial attempt (#3494). Omit for the legacy single-attempt behavior.'), + timeout: z.number().int().positive().optional().describe('Per-attempt time limit in milliseconds; an over-limit run is recorded with execution status "timeout" (#3494). The in-flight handler is abandoned, not forcibly cancelled. Omit for no time limit.'), enabled: z.boolean().default(true).describe('Whether the job is enabled'), }));