diff --git a/.changeset/declarative-cron-job-schedule-envelope.md b/.changeset/declarative-cron-job-schedule-envelope.md new file mode 100644 index 0000000000..8194e27594 --- /dev/null +++ b/.changeset/declarative-cron-job-schedule-envelope.md @@ -0,0 +1,38 @@ +--- +"@objectstack/runtime": patch +"@objectstack/observability": patch +--- + +fix(runtime): declarative `defineJob` cron jobs are actually scheduled (#4567) + +Every background job authored as `defineJob({ schedule: { type: 'cron', … } })` +was **silently never scheduled**. `JobSchema.parse` rewrites the cron +`expression` into the canonical expression envelope +(`{ dialect: 'cron', source: '0 1 * * *' }` — the authoring/persistence tier), +but `AppPlugin` handed `job.schedule` verbatim to `IJobService.schedule`, whose +boundary contract documents `expression` as a **bare cron string** because +`CronJobAdapter` passes it straight to croner. croner rejected the object +(`CronPattern: Pattern has to be of type string.`), the throw was swallowed by a +per-job `try/catch` that only `warn`ed, and the author saw a green build and a +green boot with the job never running. `interval` / `once` schedules and +flow `schedule` triggers were unaffected. + +**Fix (contract-first).** The authoring→boundary downgrade now happens at the one +place the two tiers meet — `AppPlugin`'s declarative-job registration, alongside +the existing `retryPolicy` / `timeout` threading — via +`toBoundaryJobSchedule()`. The adapters stay strict: no `typeof === 'object'` +tolerance was added downstream, so the boundary keeps exactly one shape. +A schedule that cannot be reduced to it (unknown type, AST-only or non-`cron` +expression envelope, missing `intervalMs` / `at`) is rejected by name. + +**The failure path is no longer silent.** A job that cannot be scheduled now logs +at **error** level with its own message (`Background job FAILED TO SCHEDULE — it +will never run`), plus a boot summary line when any job failed, and increments +the new `job_schedule_failures_total` counter +(`SEMCONV.jobScheduleFailuresTotal`, labels `app` / `job`) on the observability +metrics registry. "Failed to schedule" no longer shares the quiet `warn` used by +"handler not found in bundle.functions" — the first is an outage of declared +work, the second is a job that was never going to run. + +No authoring change is required: existing `defineJob` cron declarations start +working on upgrade. diff --git a/packages/observability/src/semconv.ts b/packages/observability/src/semconv.ts index 0d7e211cb5..857d342faa 100644 --- a/packages/observability/src/semconv.ts +++ b/packages/observability/src/semconv.ts @@ -46,6 +46,15 @@ export const SEMCONV = { /** Counter, labels: `adapter`, `op`, `errorClass`. */ cacheErrorsTotal: 'cache_errors_total', + // ── Background jobs — emitted by `@objectstack/runtime`'s AppPlugin ── + /** + * Counter, labels: `app`, `job`. Incremented when a DECLARED background + * job could not be handed to the job service — i.e. the app booted green + * but that job will never run (#4567). Any non-zero value is an outage of + * the job, not a warning. + */ + jobScheduleFailuresTotal: 'job_schedule_failures_total', + // ── Package / registry-reader — emitted by `@objectstack/service-package` ── /** Counter, labels: `result` (`ok`|`miss`|`error`). */ registryLookupsTotal: 'registry_lookups_total', diff --git a/packages/runtime/package.json b/packages/runtime/package.json index e48a9a43ec..ca97baf251 100644 --- a/packages/runtime/package.json +++ b/packages/runtime/package.json @@ -47,6 +47,7 @@ "@objectstack/platform-objects": "workspace:*", "@objectstack/plugin-hono-server": "workspace:*", "@objectstack/service-datasource": "workspace:*", + "@objectstack/service-job": "workspace:*", "@objectstack/service-messaging": "workspace:*", "typescript": "^6.0.3", "vitest": "^4.1.10" diff --git a/packages/runtime/src/app-plugin.jobs.test.ts b/packages/runtime/src/app-plugin.jobs.test.ts new file mode 100644 index 0000000000..915277c9c6 --- /dev/null +++ b/packages/runtime/src/app-plugin.jobs.test.ts @@ -0,0 +1,164 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type { PluginContext } from '@objectstack/core'; +import { defineJob } from '@objectstack/spec/system'; +import { InMemoryMetricsRegistry, OBSERVABILITY_METRICS_SERVICE, SEMCONV } from '@objectstack/observability'; +import { CronJobAdapter } from '@objectstack/service-job'; +import { AppPlugin } from './app-plugin.js'; + +/** + * #4567 — declarative cron jobs must actually reach the scheduler. + * + * These run AppPlugin against the REAL `CronJobAdapter` (croner underneath), + * not a recording double: the bug was that the authored schedule was rejected + * *inside* the adapter and the throw was swallowed, so a double that records + * whatever it is handed cannot see it. + */ +describe('AppPlugin — declarative background jobs (#4567)', () => { + let adapter: CronJobAdapter; + let metrics: InMemoryMetricsRegistry; + let ctx: PluginContext; + let readyHooks: Array<() => Promise>; + + beforeEach(() => { + adapter = new CronJobAdapter(); + metrics = new InMemoryMetricsRegistry(); + readyHooks = []; + ctx = { + logger: { info: vi.fn(), error: vi.fn(), warn: vi.fn(), debug: vi.fn() }, + registerService: vi.fn(), + getService: vi.fn((name: string) => { + if (name === 'job') return adapter; + if (name === OBSERVABILITY_METRICS_SERVICE) return metrics; + if (name === 'objectql') return {}; + return undefined; + }), + getServices: vi.fn(() => []), + hook: vi.fn((event: string, cb: () => Promise) => { + if (event === 'kernel:ready') readyHooks.push(cb); + }), + trigger: vi.fn(), + } as unknown as PluginContext; + }); + + afterEach(async () => { + await adapter.destroy(); + }); + + const fireReady = async () => { + for (const cb of readyHooks) await cb(); + }; + + const errorLogs = () => vi.mocked(ctx.logger.error).mock.calls.map(c => String(c[0])); + const warnLogs = () => vi.mocked(ctx.logger.warn).mock.calls.map(c => String(c[0])); + + it('schedules a defineJob cron job end-to-end — the adapter holds a live cron task', async () => { + const sweep = vi.fn(async () => { /* handler body */ }); + const job = defineJob({ + name: 'health_sweep', + schedule: { type: 'cron', expression: '0 1 * * *' }, + handler: 'sweep', + }); + const plugin = new AppPlugin({ + id: 'com.test.jobs', + jobs: [job], + functions: { sweep }, + }); + + await plugin.start!(ctx); + await fireReady(); + + // Scheduler state, not merely "did not throw": the adapter only records + // a job after `new Cron(...)` succeeded. + expect(await adapter.listJobs()).toContain('health_sweep'); + const task = (adapter as unknown as { jobs: Map }) + .jobs.get('health_sweep')?.task; + const next = task?.nextRun(); + expect(next).toBeInstanceOf(Date); + // '0 1 * * *' with the schema-defaulted UTC timezone. + expect(next!.getUTCHours()).toBe(1); + expect(next!.getUTCMinutes()).toBe(0); + + // And the registered task really runs the bundle handler. + await adapter.trigger('health_sweep'); + expect(sweep).toHaveBeenCalledTimes(1); + + expect(errorLogs()).toEqual([]); + expect(metrics.totalCounter(SEMCONV.jobScheduleFailuresTotal)).toBe(0); + }); + + it('REVERT-PROOF: the raw authored schedule still breaks the adapter, exactly as #4567 reported', async () => { + const job = defineJob({ + name: 'health_sweep', + schedule: { type: 'cron', expression: '0 1 * * *' }, + handler: 'sweep', + }); + + // What AppPlugin used to pass verbatim. The adapter's contract is a bare + // string and it stays strict — remove the downgrade in AppPlugin and the + // end-to-end test above fails with precisely this croner error. + await expect( + adapter.schedule('health_sweep', job.schedule as never, async () => { /* noop */ }), + ).rejects.toThrow(/Pattern has to be of type string/i); + expect(await adapter.listJobs()).not.toContain('health_sweep'); + }); + + it('interval jobs keep working (no envelope on that branch)', async () => { + const plugin = new AppPlugin({ + id: 'com.test.jobs', + jobs: [defineJob({ name: 'ping', schedule: { type: 'interval', intervalMs: 60_000 }, handler: 'h' })], + functions: { h: vi.fn(async () => { /* noop */ }) }, + }); + + await plugin.start!(ctx); + await fireReady(); + + expect(await adapter.listJobs()).toContain('ping'); + expect(errorLogs()).toEqual([]); + }); + + it('a job that cannot be scheduled logs ERROR (not a silent warn) and counts', async () => { + const plugin = new AppPlugin({ + id: 'com.test.jobs', + jobs: [{ + name: 'bad_job', + // A CEL envelope where a cron one belongs — unusable at the boundary. + schedule: { type: 'cron', expression: { dialect: 'cel', source: 'now()' } }, + handler: 'h', + enabled: true, + }], + functions: { h: vi.fn(async () => { /* noop */ }) }, + }); + + await plugin.start!(ctx); + await fireReady(); + + expect(await adapter.listJobs()).not.toContain('bad_job'); + + // Loud: error level, its own distinct message, and a counter. + const errors = errorLogs(); + expect(errors.some(m => m.includes('FAILED TO SCHEDULE'))).toBe(true); + expect(errors.some(m => m.includes('declared but NOT scheduled'))).toBe(true); + expect(vi.mocked(ctx.logger.error).mock.calls[0][2]).toMatchObject({ job: 'bad_job' }); + expect(metrics.totalCounter(SEMCONV.jobScheduleFailuresTotal, { job: 'bad_job' })).toBe(1); + + // NOT folded into the warn stream that "handler missing"/"disabled" use. + expect(warnLogs().some(m => /schedule/i.test(m))).toBe(false); + }); + + it('a missing handler stays a warn — the two failures are not one signal', async () => { + const plugin = new AppPlugin({ + id: 'com.test.jobs', + jobs: [defineJob({ name: 'orphan', schedule: { type: 'cron', expression: '0 1 * * *' }, handler: 'nope' })], + functions: {}, + }); + + await plugin.start!(ctx); + await fireReady(); + + expect(warnLogs().some(m => m.includes('job handler not found'))).toBe(true); + expect(errorLogs()).toEqual([]); + expect(metrics.totalCounter(SEMCONV.jobScheduleFailuresTotal)).toBe(0); + }); +}); diff --git a/packages/runtime/src/app-plugin.ts b/packages/runtime/src/app-plugin.ts index 2ba0d0bc66..955e24a9c0 100644 --- a/packages/runtime/src/app-plugin.ts +++ b/packages/runtime/src/app-plugin.ts @@ -13,7 +13,9 @@ import { readServiceSelfInfo } from '@objectstack/spec/api'; import { QuickJSScriptRunner } from './sandbox/quickjs-runner.js'; import { hookBodyRunnerFactory, actionBodyRunnerFactory } from './sandbox/body-runner.js'; import { GLOBAL_ACTION_OBJECT_KEY } from './action-execution.js'; -import { countServerTiming } from '@objectstack/observability'; +import { toBoundaryJobSchedule } from './job-schedule.js'; +import { countServerTiming, SEMCONV } from '@objectstack/observability'; +import { resolveMetrics } from './observability/observability-service-plugin.js'; /** * The write options every seed insert must use — mirrors @@ -810,7 +812,9 @@ export class AppPlugin implements Plugin { return; } const fnMap = collectBundleFunctions(this.bundle); + const metrics = resolveMetrics(ctx); let ok = 0; + let failed = 0; for (const job of jobs) { const jobName: string = job?.name; if (!jobName) { @@ -831,7 +835,13 @@ export class AppPlugin implements Plugin { try { await svc.schedule( jobName, - job.schedule, + // #4567: authoring tier → boundary tier. `job.schedule` + // is the PARSED `Schedule`, whose cron `expression` is + // the ADR expression envelope `{dialect,source}`; + // `IJobService.schedule` (and croner behind it) take a + // bare cron string. Same seam, same place, as the + // retryPolicy/timeout threading just below. + toBoundaryJobSchedule(job.schedule, jobName), async (jobCtx: any) => { await handler({ ...jobCtx, jobId: jobName, bundle: this.bundle }); }, @@ -842,12 +852,29 @@ export class AppPlugin implements Plugin { ); ok++; } catch (err: any) { - ctx.logger.warn('[AppPlugin] Failed to schedule job', { - appId, job: jobName, error: err?.message ?? String(err), - }); + failed++; + // #4567: a job that fails to schedule is a SILENT OUTAGE — + // the app builds and boots green while the work never runs. + // It gets error level plus its own counter, and deliberately + // NOT the `warn` that "handler not found" / "job disabled" + // use: those describe a job that was never going to run, + // this one describes a job the author is owed. + ctx.logger.error( + '[AppPlugin] Background job FAILED TO SCHEDULE — it will never run', + err as Error, + { appId, job: jobName, schedule: job.schedule }, + ); + metrics.counter(SEMCONV.jobScheduleFailuresTotal, { app: appId, job: jobName }); } } - ctx.logger.info('[AppPlugin] Scheduled background jobs', { appId, count: ok }); + ctx.logger.info('[AppPlugin] Scheduled background jobs', { appId, count: ok, failed }); + if (failed > 0) { + ctx.logger.error( + '[AppPlugin] Some background jobs are declared but NOT scheduled', + undefined, + { appId, scheduled: ok, failed }, + ); + } }); } } catch (err: any) { diff --git a/packages/runtime/src/job-schedule.test.ts b/packages/runtime/src/job-schedule.test.ts new file mode 100644 index 0000000000..45e88207f4 --- /dev/null +++ b/packages/runtime/src/job-schedule.test.ts @@ -0,0 +1,94 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect } from 'vitest'; +import { defineJob } from '@objectstack/spec/system'; +import { toBoundaryJobSchedule } from './job-schedule.js'; + +/** + * #4567 — the authoring↔boundary downgrade itself. + * + * `defineJob` parses the cron `expression` into the ADR expression envelope; + * `IJobService.schedule` (and croner behind it) take a bare string. These pin + * the one conversion that bridges them. + */ +describe('toBoundaryJobSchedule', () => { + it('unwraps the cron expression envelope a parsed Schedule carries', () => { + const job = defineJob({ + name: 'health_sweep', + schedule: { type: 'cron', expression: '0 1 * * *' }, + handler: 'sweep', + }); + + // The premise of #4567: parsing produced an envelope, not a string. + expect(job.schedule).toEqual({ + type: 'cron', + expression: { dialect: 'cron', source: '0 1 * * *' }, + timezone: 'UTC', + }); + + expect(toBoundaryJobSchedule(job.schedule, 'health_sweep')).toEqual({ + type: 'cron', + expression: '0 1 * * *', + timezone: 'UTC', + }); + }); + + it('accepts the authoring-input spelling (bare string) unchanged', () => { + expect(toBoundaryJobSchedule({ type: 'cron', expression: '*/5 * * * *' }, 'j')).toEqual({ + type: 'cron', + expression: '*/5 * * * *', + }); + }); + + it('carries a string timezone across and drops a non-string one', () => { + expect( + toBoundaryJobSchedule( + { type: 'cron', expression: { dialect: 'cron', source: '0 2 * * *' }, timezone: 'America/New_York' }, + 'j', + ), + ).toEqual({ type: 'cron', expression: '0 2 * * *', timezone: 'America/New_York' }); + + expect( + toBoundaryJobSchedule({ type: 'cron', expression: '0 2 * * *', timezone: 42 as unknown as string }, 'j'), + ).toEqual({ type: 'cron', expression: '0 2 * * *' }); + }); + + it('passes interval / once schedules through (no envelope on those branches)', () => { + const interval = defineJob({ + name: 'ping', + schedule: { type: 'interval', intervalMs: 5000 }, + handler: 'h', + }); + expect(toBoundaryJobSchedule(interval.schedule, 'ping')).toEqual({ type: 'interval', intervalMs: 5000 }); + + const once = defineJob({ + name: 'kickoff', + schedule: { type: 'once', at: '2030-01-01T00:00:00.000Z' }, + handler: 'h', + }); + expect(toBoundaryJobSchedule(once.schedule, 'kickoff')).toEqual({ + type: 'once', + at: '2030-01-01T00:00:00.000Z', + }); + }); + + it('throws — naming the job — on shapes that cannot reach the boundary', () => { + // Wrong dialect: a CEL expression is not a schedule. + expect(() => + toBoundaryJobSchedule({ type: 'cron', expression: { dialect: 'cel', source: 'now()' } }, 'bad_job'), + ).toThrow(/bad_job.*cel/s); + + // AST-only envelope: nothing for croner to parse. + expect(() => + toBoundaryJobSchedule({ type: 'cron', expression: { dialect: 'cron', ast: {} } }, 'bad_job'), + ).toThrow(/bad_job.*AST-only/s); + + expect(() => toBoundaryJobSchedule({ type: 'cron' }, 'bad_job')).toThrow(/bad_job/); + expect(() => toBoundaryJobSchedule({ type: 'cron', expression: ' ' }, 'bad_job')).toThrow(/bad_job/); + expect(() => toBoundaryJobSchedule({ type: 'interval' }, 'bad_job')).toThrow(/bad_job.*intervalMs/s); + expect(() => toBoundaryJobSchedule({ type: 'once' }, 'bad_job')).toThrow(/bad_job.*at/s); + expect(() => toBoundaryJobSchedule({ type: 'weekly' }, 'bad_job')).toThrow(/bad_job.*weekly/s); + expect(() => toBoundaryJobSchedule(undefined, 'bad_job')).toThrow(/bad_job/); + expect(() => toBoundaryJobSchedule('0 1 * * *', 'bad_job')).toThrow(/bad_job/); + }); +}); diff --git a/packages/runtime/src/job-schedule.ts b/packages/runtime/src/job-schedule.ts new file mode 100644 index 0000000000..038da3d2f0 --- /dev/null +++ b/packages/runtime/src/job-schedule.ts @@ -0,0 +1,103 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +import type { JobSchedule } from '@objectstack/spec/contracts'; + +/** + * Downgrade an AUTHORED `Schedule` — the tier `defineJob` / `JobSchema.parse` + * produces (`@objectstack/spec` `system/job.zod.ts`) — into the BOUNDARY + * `JobSchedule` that `IJobService.schedule` consumes + * (`@objectstack/spec/contracts`). + * + * The two tiers differ in exactly one place: `CronScheduleSchema.expression` + * is a `CronExpressionInputSchema`, so parsing rewrites the authored + * `'0 1 * * *'` into the canonical expression envelope + * `{ dialect: 'cron', source: '0 1 * * *' }` — while the boundary type + * documents `expression` as a **bare cron string**, because `CronJobAdapter` + * hands it straight to croner, which rejects anything else with + * `"CronPattern: Pattern has to be of type string."`. + * + * #4567: nothing bridged the two, so every declarative cron job threw inside + * the adapter and the throw was swallowed by AppPlugin's per-job catch — + * declared, built, booted, never scheduled. This function is the ONE declared + * conversion point, sitting exactly on the authoring↔boundary seam #4538 drew; + * the adapters stay strict (no `typeof === 'object'` tolerance downstream, per + * Prime Directive #12). + * + * A bare string `expression` is still accepted here because that is the + * *authoring input* form (`z.input` of the same schema): bundles assembled + * without a `JobSchema.parse` round-trip legitimately carry it. Both are + * authoring-tier spellings of one value; neither reaches the adapter. + * + * Anything that cannot be reduced to the boundary shape — an unknown schedule + * type, an AST-only or non-cron expression envelope, a cron entry with no + * source — **throws**, naming the job. Silence is precisely what #4567 was. + */ +export function toBoundaryJobSchedule(schedule: unknown, jobName: string): JobSchedule { + if (!schedule || typeof schedule !== 'object') { + throw new Error( + `job "${jobName}": schedule must be an object of type cron | interval | once, got ${typeof schedule}`, + ); + } + const s = schedule as Record; + + if (s.type === 'cron') { + return { type: 'cron', expression: cronSource(s.expression, jobName), ...timezoneOf(s) }; + } + + if (s.type === 'interval') { + if (typeof s.intervalMs !== 'number' || !Number.isFinite(s.intervalMs) || s.intervalMs <= 0) { + throw new Error( + `job "${jobName}": interval schedule needs a positive \`intervalMs\`, got ${JSON.stringify(s.intervalMs)}`, + ); + } + return { type: 'interval', intervalMs: s.intervalMs }; + } + + if (s.type === 'once') { + if (typeof s.at !== 'string' || s.at.trim() === '') { + throw new Error( + `job "${jobName}": once schedule needs an ISO 8601 \`at\`, got ${JSON.stringify(s.at)}`, + ); + } + return { type: 'once', at: s.at }; + } + + throw new Error( + `job "${jobName}": unknown schedule type ${JSON.stringify(s.type)} (expected cron | interval | once)`, + ); +} + +/** Reduce an authored cron `expression` (envelope or bare string) to the bare cron string croner needs. */ +function cronSource(expression: unknown, jobName: string): string { + if (typeof expression === 'string') { + if (expression.trim() === '') { + throw new Error(`job "${jobName}": cron schedule has an empty expression`); + } + return expression; + } + + if (expression && typeof expression === 'object') { + const env = expression as Record; + if (env.dialect !== undefined && env.dialect !== 'cron') { + throw new Error( + `job "${jobName}": cron schedule carries a ${JSON.stringify(env.dialect)} expression; ` + + 'a job schedule must use the `cron` dialect', + ); + } + if (typeof env.source === 'string' && env.source.trim() !== '') return env.source; + if (env.ast !== undefined) { + throw new Error( + `job "${jobName}": cron expression is AST-only; the job scheduler needs the \`source\` string`, + ); + } + } + + throw new Error( + `job "${jobName}": cron schedule missing a usable expression (got ${JSON.stringify(expression)})`, + ); +} + +/** Carry a string `timezone` across; drop anything else rather than handing croner a non-string. */ +function timezoneOf(s: Record): { timezone?: string } { + return typeof s.timezone === 'string' && s.timezone !== '' ? { timezone: s.timezone } : {}; +} diff --git a/packages/runtime/vitest.config.ts b/packages/runtime/vitest.config.ts index 9ce1b9ab6a..a9066e43f4 100644 --- a/packages/runtime/vitest.config.ts +++ b/packages/runtime/vitest.config.ts @@ -32,6 +32,10 @@ export default defineConfig({ '@objectstack/spec/security': path.resolve(__dirname, '../spec/src/security/index.ts'), '@objectstack/spec': path.resolve(__dirname, '../spec/src/index.ts'), '@objectstack/types': path.resolve(__dirname, '../types/src/index.ts'), + // Dev-only: app-plugin.jobs.test.ts drives the REAL CronJobAdapter, so + // the #4567 regression (croner rejecting the expression envelope) is + // reproduced by the actual scheduler rather than by a double. + '@objectstack/service-job': path.resolve(__dirname, '../services/service-job/src/index.ts'), }, }, test: { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 487b1f7134..f948481ac5 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1841,6 +1841,9 @@ importers: '@objectstack/plugin-hono-server': specifier: workspace:* version: link:../plugins/plugin-hono-server + '@objectstack/service-job': + specifier: workspace:* + version: link:../services/service-job '@objectstack/service-messaging': specifier: workspace:* version: link:../services/service-messaging