fix(service-job): migrate early registrations across the placeholder→DbJobAdapter upgrade — cron jobs finally run

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

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<string, JobRecord>();
   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<void> {
@@ -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<void> {
     const record = this.jobs.get(name);
     if (record?.timerId) {

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<string, any[]>();
+  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<string, any>();
+  const hooks = new Map<string, Array<() => Promise<void> | 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> | 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();
+  });
+});

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