feat(job): honor the authored retryPolicy / timeout in the job scheduler (#3494)

.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.

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 |
 
 

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) {

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/);
+  });
+});

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<void> {
+  async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise<void> {
     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();

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<void> {
+  async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise<void> {
     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);

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';

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/);
+  });
+});

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<typeof setInterval> | ReturnType<typeof setTimeout>;
   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<void> {
+  async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise<void> {
     // 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();

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<void> {
+  return new Promise((resolve) => {
+    const t = setTimeout(resolve, ms);
+    (t as any)?.unref?.();
+  });
+}
+
+function withTimeout(run: () => Promise<void>, jobId: string, timeoutMs?: number): Promise<void> {
+  if (!timeoutMs || timeoutMs <= 0) return run();
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const guard = new Promise<never>((_, reject) => {
+    timer = setTimeout(() => reject(new JobTimeoutError(jobId, timeoutMs)), timeoutMs);
+    (timer as any)?.unref?.();
+  });
+  return Promise.race([run(), guard]).finally(() => clearTimeout(timer)) as Promise<void>;
+}
+
+/**
+ * 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<void>,
+  options?: JobScheduleOptions,
+): Promise<void> {
+  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;
+}

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)",