fix(runtime): 声明式 cron job 的表达式信封在 job 边界处降解 (#4567)

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

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

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"

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<void>>;
+
+    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<void>) => {
+                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<string, { task?: { nextRun(): Date | null } }> })
+            .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);
+    });
+});

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

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