From a5802c6550a80cfaeec11e8e68d6a8f500679471 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8C=85=E5=91=A8=E6=B6=9B?= Date: Thu, 30 Jul 2026 09:38:22 -0700 Subject: [PATCH] =?UTF-8?q?fix(service-job):=20migrate=20early=20registrat?= =?UTF-8?q?ions=20across=20the=20placeholder=E2=86=92DbJobAdapter=20upgrad?= =?UTF-8?q?e?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Business plugins start() before kernel:ready, so their schedules all land on the placeholder IntervalJobAdapter — which silently ignores cron schedules — and the upgrade swapped the service without migrating anything. Default config result: cron jobs never ran at all, while interval timers kept running on the orphaned placeholder, invisible to sys_job. The upgrade now snapshots every early registration, stops the placeholder, and re-schedules them on the DbJobAdapter. The IntervalJobAdapter warns per cron registration it cannot execute, and the no-engine path summarizes the stranded cron jobs. Co-Authored-By: Claude Fable 5 --- .../job-placeholder-migrates-to-db-adapter.md | 7 + .../service-job/src/interval-job-adapter.ts | 43 +++++- .../src/job-service-plugin.test.ts | 139 ++++++++++++++++++ .../service-job/src/job-service-plugin.ts | 39 ++++- 4 files changed, 225 insertions(+), 3 deletions(-) create mode 100644 .changeset/job-placeholder-migrates-to-db-adapter.md create mode 100644 packages/services/service-job/src/job-service-plugin.test.ts diff --git a/.changeset/job-placeholder-migrates-to-db-adapter.md b/.changeset/job-placeholder-migrates-to-db-adapter.md new file mode 100644 index 0000000000..e1a5b56d67 --- /dev/null +++ b/.changeset/job-placeholder-migrates-to-db-adapter.md @@ -0,0 +1,7 @@ +--- +'@objectstack/service-job': patch +--- + +Jobs registered before `kernel:ready` now survive the placeholder→DbJobAdapter upgrade. + +Business plugins `start()` before the JobServicePlugin's `kernel:ready` hook, so every schedule they register lands on the placeholder IntervalJobAdapter. That placeholder silently ignores `cron` schedules, and the upgrade used `replaceService` without migrating anything — so in the default configuration a plugin's cron jobs never ran at all, while its interval timers kept running on the orphaned placeholder, invisible to `sys_job`. The upgrade now snapshots every early registration, stops the placeholder, and re-schedules them on the DbJobAdapter (whose croner-backed cron routing makes the cron entries actually fire). The IntervalJobAdapter also warns per cron registration, and the no-engine path summarizes stranded cron jobs instead of staying silent. diff --git a/packages/services/service-job/src/interval-job-adapter.ts b/packages/services/service-job/src/interval-job-adapter.ts index 93fb164f32..c6ea7f2055 100644 --- a/packages/services/service-job/src/interval-job-adapter.ts +++ b/packages/services/service-job/src/interval-job-adapter.ts @@ -21,6 +21,25 @@ interface JobRecord { export interface IntervalJobAdapterOptions { /** Maximum number of execution records to retain per job (default: 100) */ maxExecutions?: number; + /** + * Logger used to surface schedules this adapter cannot execute. A `cron` + * registration is stored but never fired here, and staying silent about it + * meant "background automation off" with zero diagnostics: every + * plugin that registered its jobs before the DbJobAdapter upgrade saw + * interval jobs run and cron jobs do nothing. + */ + logger?: { warn(message: string, ...args: unknown[]): void }; +} + +/** + * A job registration as originally supplied to {@link IntervalJobAdapter.schedule}, + * exposed so an upgraded adapter can re-schedule it (see JobServicePlugin). + */ +export interface JobRegistration { + name: string; + schedule: JobSchedule; + handler: JobHandler; + options?: JobScheduleOptions; } /** @@ -35,9 +54,11 @@ export interface IntervalJobAdapterOptions { export class IntervalJobAdapter implements IJobService { private readonly jobs = new Map(); private readonly maxExecutions: number; + private readonly logger?: IntervalJobAdapterOptions['logger']; constructor(options: IntervalJobAdapterOptions = {}) { this.maxExecutions = options.maxExecutions ?? 100; + this.logger = options.logger; } async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { @@ -57,12 +78,32 @@ export class IntervalJobAdapter implements IJobService { await this.executeJob(record); }, delay); } + } else if (schedule.type === 'cron') { + // Stored but NOT executed — this adapter has no cron engine. Loud on + // purpose: registrations that land here before the DbJobAdapter upgrade + // are re-delivered by JobServicePlugin, but an adapter that KEEPS them + // (explicit `adapter: 'interval'`, or no ObjectQL engine) silently ran + // zero cron jobs. + this.logger?.warn( + `IntervalJobAdapter: cron schedule "${name}" registered but will not run — ` + + 'this adapter has no cron engine. Use the db/cron adapter, or an interval schedule.', + ); } - // 'cron' type: stored but not actively scheduled (needs cron library) this.jobs.set(name, record); } + /** + * Snapshot of every registration as originally supplied, for handing over to + * a replacement adapter. Does not mutate this adapter's state — callers + * migrate with `getRegistrations()` then `destroy()`. + */ + getRegistrations(): JobRegistration[] { + return [...this.jobs.values()].map(({ name, schedule, handler, options }) => ({ + name, schedule, handler, options, + })); + } + async cancel(name: string): Promise { const record = this.jobs.get(name); if (record?.timerId) { diff --git a/packages/services/service-job/src/job-service-plugin.test.ts b/packages/services/service-job/src/job-service-plugin.test.ts new file mode 100644 index 0000000000..9a07d47c68 --- /dev/null +++ b/packages/services/service-job/src/job-service-plugin.test.ts @@ -0,0 +1,139 @@ +// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * Regression: the placeholder→DbJobAdapter upgrade must not strand + * early registrations. + * + * Business plugins `start()` before `kernel:ready`, so every job they register + * lands on the placeholder IntervalJobAdapter. Before this fix: + * 1. the placeholder silently ignored `cron` schedules (no timer, no error); + * 2. the upgrade swapped the service via `replaceService` WITHOUT migrating + * the already-registered jobs — cron entries were lost for good; + * 3. the placeholder's interval timers kept running on the orphaned adapter + * (which is why interval jobs appeared healthy while cron jobs never + * fired, and none of them showed up in sys_job). + */ + +import { describe, it, expect, vi } from 'vitest'; +import { JobServicePlugin } from './job-service-plugin'; + +function makeFakeEngine() { + const tables = new Map(); + return { + tables, + async find(table: string, opts: any = {}) { + const t = tables.get(table) ?? []; + return opts.where + ? t.filter((r) => Object.entries(opts.where).every(([k, v]) => r[k] === v)) + : [...t]; + }, + 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) { + 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; + }, + }; +} + +function makeFakeContext(opts: { engine?: any } = {}) { + const services = new Map(); + const hooks = new Map Promise | void>>(); + if (opts.engine) services.set('objectql', opts.engine); + const logger = { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }; + const ctx = { + logger, + registerService: (name: string, impl: any) => { services.set(name, impl); }, + replaceService: (name: string, impl: any) => { services.set(name, impl); }, + getService: (name: string) => { + if (!services.has(name)) throw new Error(`service "${name}" not registered`); + return services.get(name); + }, + hook: (event: string, cb: () => Promise | void) => { + const list = hooks.get(event) ?? []; + list.push(cb); + hooks.set(event, list); + }, + async fire(event: string) { + for (const cb of hooks.get(event) ?? []) await cb(); + }, + services, + }; + return ctx; +} + +describe('JobServicePlugin placeholder→DbJobAdapter upgrade', () => { + it('migrates jobs registered before kernel:ready onto the DbJobAdapter', async () => { + const engine = makeFakeEngine(); + const ctx = makeFakeContext({ engine }); + const plugin = new JobServicePlugin(); + await plugin.init(ctx as any); + + // Simulate a business plugin registering during its start(): the service + // resolved here is the placeholder IntervalJobAdapter. + const placeholder = ctx.getService('job'); + const cronRuns: string[] = []; + await placeholder.schedule( + 'nightly-report', + { type: 'cron', expression: '0 0 * * *' }, + async () => { cronRuns.push('ran'); }, + ); + await placeholder.schedule( + 'heartbeat', + { type: 'interval', intervalMs: 60_000 }, + async () => {}, + ); + + await ctx.fire('kernel:ready'); + + // Service was swapped… + const upgraded = ctx.getService('job'); + expect(upgraded).not.toBe(placeholder); + // …and BOTH early registrations followed it (the cron one used to vanish). + expect((await upgraded.listJobs()).sort()).toEqual(['heartbeat', 'nightly-report']); + // The cron job is now persisted where operators can see it. + const sysJobs = engine.tables.get('sys_job') ?? []; + expect(sysJobs.map((r: any) => r.name).sort()).toEqual(['heartbeat', 'nightly-report']); + // The orphaned placeholder no longer owns any jobs (its timers are stopped). + expect(await placeholder.listJobs()).toEqual([]); + // The migrated cron handler still works when triggered through the new adapter. + await upgraded.trigger('nightly-report'); + expect(cronRuns).toEqual(['ran']); + + await plugin.destroy(); + }); + + it('warns loudly when cron jobs stay stranded on the interval adapter (no engine)', async () => { + const ctx = makeFakeContext(); + const plugin = new JobServicePlugin(); + await plugin.init(ctx as any); + + const job = ctx.getService('job'); + await job.schedule('never-runs', { type: 'cron', expression: '*/5 * * * *' }, async () => {}); + + // Registration itself already warned (the per-schedule diagnostic). + expect(ctx.logger.warn).toHaveBeenCalledWith(expect.stringContaining('never-runs')); + + await ctx.fire('kernel:ready'); + + // And the no-upgrade path summarizes the stranded cron jobs. + const warned = ctx.logger.warn.mock.calls.some( + (c: any[]) => typeof c[0] === 'string' && c[0].includes('will NOT run') && c[0].includes('never-runs'), + ); + expect(warned).toBe(true); + + await plugin.destroy(); + }); +}); diff --git a/packages/services/service-job/src/job-service-plugin.ts b/packages/services/service-job/src/job-service-plugin.ts index 705b175e27..220b3a0365 100644 --- a/packages/services/service-job/src/job-service-plugin.ts +++ b/packages/services/service-job/src/job-service-plugin.ts @@ -78,7 +78,7 @@ export class JobServicePlugin implements Plugin { const choice = this.options.adapter ?? 'auto'; if (choice === 'interval') { - this.intervalAdapter = new IntervalJobAdapter(this.options.interval); + this.intervalAdapter = new IntervalJobAdapter({ ...this.options.interval, logger: ctx.logger }); ctx.registerService('job', this.intervalAdapter); ctx.logger.info('JobServicePlugin: registered IntervalJobAdapter (in-memory)'); return; @@ -94,7 +94,7 @@ export class JobServicePlugin implements Plugin { // 'auto' or 'db' — register a placeholder Interval adapter synchronously // so callers can `getService('job')` during init, then upgrade in kernel:ready // when the objectql engine is wired. - this.intervalAdapter = new IntervalJobAdapter(this.options.interval); + this.intervalAdapter = new IntervalJobAdapter({ ...this.options.interval, logger: ctx.logger }); ctx.registerService('job', this.intervalAdapter); ctx.hook('kernel:ready', async () => { @@ -108,6 +108,17 @@ export class JobServicePlugin implements Plugin { } else { ctx.logger.info('JobServicePlugin: no ObjectQL engine — staying on IntervalJobAdapter'); } + // Jobs stuck on the placeholder include cron schedules that + // will never fire. The per-registration warning already fired, but the + // summary makes "background automation is off" visible in one line. + const stranded = this.intervalAdapter?.getRegistrations() + .filter((r) => r.schedule.type === 'cron') + .map((r) => r.name) ?? []; + if (stranded.length > 0) { + ctx.logger.warn( + `JobServicePlugin: ${stranded.length} cron job(s) will NOT run on IntervalJobAdapter: ${stranded.join(', ')}`, + ); + } return; } @@ -133,6 +144,30 @@ export class JobServicePlugin implements Plugin { ctx.logger.info('JobServicePlugin: upgraded to DbJobAdapter (sys_job + sys_job_run persistence)'); } catch (err) { ctx.logger.warn('JobServicePlugin: replaceService failed; staying on IntervalJobAdapter', err as any); + return; + } + + // Migrate every registration made against the placeholder. + // Business plugins `start()` before this hook runs, so their schedules + // all landed on the IntervalJobAdapter: its cron entries never fired at + // all, and its interval timers would keep running on the orphaned + // placeholder (invisible to sys_job) after the swap. Stop the placeholder + // FIRST — a brief gap beats a double-fire — then re-schedule everything + // on the DbJobAdapter. + const pending = this.intervalAdapter?.getRegistrations() ?? []; + if (this.intervalAdapter) { + await this.intervalAdapter.destroy(); + this.intervalAdapter = undefined; + } + for (const r of pending) { + try { + await this.dbAdapter.schedule(r.name, r.schedule, r.handler, r.options); + } catch (err) { + ctx.logger.warn(`JobServicePlugin: failed to migrate job "${r.name}" to DbJobAdapter`, err as any); + } + } + if (pending.length > 0) { + ctx.logger.info(`JobServicePlugin: migrated ${pending.length} early job registration(s) to DbJobAdapter`); } // Retention is owned by the platform LifecycleService (ADR-0057):