diff --git a/.changeset/cron-rebind-after-kernel-rebuild.md b/.changeset/cron-rebind-after-kernel-rebuild.md new file mode 100644 index 0000000000..f08b1e1cfa --- /dev/null +++ b/.changeset/cron-rebind-after-kernel-rebuild.md @@ -0,0 +1,26 @@ +--- +"@objectstack/service-job": patch +"@objectstack/trigger-schedule": patch +--- + +Fix scheduled and time-relative flows permanently failing to re-bind after a kernel rebuild. + +`DbJobAdapter.destroy()` destroyed only its interval adapter, never the cron adapter it +was handed — so every evicted kernel left its croner timers running, holding their names +in croner's process-global registry for the life of the process. Because kernel eviction +is routine in the cloud runtime, the normal path was: a scheduled automation binds once, +the next metadata edit evicts the kernel, and the flow never binds again ("name already +taken") while Studio, the metadata API and `verify_build` all keep reporting it healthy. + +Four changes close it: + +- `DbJobAdapter.destroy()` now also destroys the cron adapter, and `JobServicePlugin` + releases the cron adapter it owns on the `adapter: 'cron'` path. +- `CronJobAdapter` scopes its entry in croner's process-global registry to the adapter + INSTANCE (`CronJobAdapter.cronRegistryName()` exposes the key). This also fixes a + second defect with no eviction involved: two environments in one container binding the + same flow name no longer collide. +- Registering a name something else still holds now REPLACES it — the previous job is + stopped, never left running alongside the new one. +- A flow that fails to bind to the job service is now reported at `error` with the + consequence and the remedy, instead of a `warn` nobody reads. diff --git a/packages/services/service-job/src/cron-job-adapter.test.ts b/packages/services/service-job/src/cron-job-adapter.test.ts index c073efb9d4..ce943229e3 100644 --- a/packages/services/service-job/src/cron-job-adapter.test.ts +++ b/packages/services/service-job/src/cron-job-adapter.test.ts @@ -1,6 +1,7 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import { describe, it, expect, afterEach } from 'vitest'; +import { Cron, scheduledJobs } from 'croner'; import { CronJobAdapter } from './cron-job-adapter.js'; describe('CronJobAdapter', () => { @@ -132,3 +133,95 @@ describe('CronJobAdapter retryPolicy / timeout (#3494)', () => { expect(execs[0].error).toMatch(/timed out after 25ms/); }); }); + +// ─── #8362 — croner's PROCESS-GLOBAL named registry ───────────────────────── +// +// `new Cron(expr, { name }, fn)` pushes into a module-level array inside croner +// and throws `name already taken` when that name is live. That array is scoped +// to the PROCESS, not to this adapter, not to a kernel and not to an +// environment — so two adapter instances that are each perfectly consistent +// with themselves can still collide, and a stopped-but-never-destroyed instance +// keeps its names forever. +// +// Two live-fire consequences these cases pin, both reproduced on a real rig +// before the fix: +// 1. two environments in one container, same AI-generated flow name, NO +// eviction involved — the second environment's automation never binds; +// 2. an evicted kernel whose cron adapter was never destroyed holds the name +// forever, so every later rebind of that flow fails permanently. +// +// The pins deliberately go through the `cron` path: `interval` schedules use +// `setInterval` and never enter croner's named registry at all, so an +// interval-shaped fixture would pass on a completely unfixed tree. +describe('CronJobAdapter — process-global croner name registry (#8362)', () => { + const live: CronJobAdapter[] = []; + const make = (options?: ConstructorParameters[0]) => { + const a = new CronJobAdapter(options); + live.push(a); + return a; + }; + afterEach(async () => { + while (live.length) await live.pop()!.destroy(); + }); + + /** Croner's process-global registry, narrowed to one PUBLIC job name. */ + const registeredFor = (jobName: string) => + scheduledJobs.filter((j) => (j.name ?? '').endsWith(jobName)); + + const DAILY = { type: 'cron', expression: '0 8 * * *' } as const; + + it('lets two live adapters hold the SAME job name — two environments, one container', async () => { + const NAME = 'flow-time-relative:contract_expiry_reminder_flow'; + const fired: string[] = []; + + const envA = make(); + await envA.schedule(NAME, DAILY, async () => { fired.push('A'); }); + // The FIRST bind must really have entered the named registry: a rebind pin + // whose first bind registered nothing passes for the wrong reason. + expect(registeredFor(NAME)).toHaveLength(1); + + const envB = make(); + await envB.schedule(NAME, DAILY, async () => { fired.push('B'); }); + + expect(registeredFor(NAME)).toHaveLength(2); + + // Each environment's timer drives its OWN handler. + for (const job of registeredFor(NAME)) await job.trigger(); + expect([...fired].sort()).toEqual(['A', 'B']); + }); + + it('frees the process-global name on destroy() — the job is STOPPED, not renamed around', async () => { + const NAME = 'flow-schedule:nightly_rollup'; + const adapterA = make(); + await adapterA.schedule(NAME, DAILY, async () => {}); + + const [job] = registeredFor(NAME); + expect(job).toBeDefined(); + expect(job.isStopped()).toBe(false); + + await adapterA.destroy(); + + expect(job.isStopped()).toBe(true); + expect(registeredFor(NAME)).toHaveLength(0); + }); + + it('reclaims its registry name from a foreign holder instead of warning and giving up', async () => { + const NAME = 'flow-schedule:reclaim_me'; + const adapterA = make(); + let calls = 0; + + // Somebody else already holds the exact name this adapter will register + // under — the residual shape once per-instance namespacing rules out our + // own collisions. Replace semantics: the holder is stopped, not tolerated. + const squatter = new Cron(DAILY.expression, { name: adapterA.cronRegistryName(NAME) }, () => {}); + expect(registeredFor(NAME)).toHaveLength(1); + + await adapterA.schedule(NAME, DAILY, async () => { calls++; }); + + expect(squatter.isStopped()).toBe(true); + const held = registeredFor(NAME); + expect(held).toHaveLength(1); + await held[0].trigger(); + expect(calls).toBe(1); + }); +}); diff --git a/packages/services/service-job/src/cron-job-adapter.ts b/packages/services/service-job/src/cron-job-adapter.ts index 983bea150f..acb6ea1ba8 100644 --- a/packages/services/service-job/src/cron-job-adapter.ts +++ b/packages/services/service-job/src/cron-job-adapter.ts @@ -1,6 +1,6 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. -import { Cron } from 'croner'; +import { Cron, scheduledJobs } from 'croner'; import type { IJobService, JobSchedule, @@ -10,6 +10,21 @@ import type { } from '@objectstack/spec/contracts'; import { runWithPolicy, JobTimeoutError } from './run-with-policy.js'; +/** + * Monotonic counter that makes every adapter instance's registry prefix unique + * within the process. Uniqueness has to be PER INSTANCE, not per environment: + * a kernel rebuild produces a new adapter for the *same* environment id, which + * is exactly the collision an environment-scoped namespace would fail to + * prevent (#8362). + */ +let ADAPTER_SEQUENCE = 0; + +/** Namespace labels ride in a croner job name — keep them boring. */ +function sanitizeNamespaceLabel(label: string | undefined): string { + const trimmed = (label ?? '').trim().replace(/[^A-Za-z0-9._-]+/g, '-'); + return trimmed.length > 0 ? trimmed.slice(0, 48) : 'kernel'; +} + /** Minimal cluster lock surface for scheduler leader-election (structural — no hard dep on the cluster contract). */ interface SchedulerCluster { lock?: { @@ -31,6 +46,16 @@ export interface CronJobAdapterOptions { cluster?: SchedulerCluster; /** Lease TTL (ms) held while a scheduled fire runs. Default 60000. */ leaseMs?: number; + /** + * Human-readable label folded into this adapter's entry in croner's + * process-global name registry — an environment id, a kernel id, anything + * that makes `scheduledJobs` readable while debugging a multi-tenant + * container. Purely cosmetic: uniqueness is guaranteed by the per-instance + * discriminator and NEVER depends on this value being supplied or distinct. + */ + namespace?: string; + /** Surface for registry-level anomalies (a reclaimed job name). */ + logger?: { warn(msg: string, meta?: unknown): void }; } interface CronJobRecord { @@ -55,12 +80,40 @@ export class CronJobAdapter implements IJobService { private readonly jobs = new Map(); private readonly cluster?: SchedulerCluster; private readonly leaseMs: number; + private readonly logger?: { warn(msg: string, meta?: unknown): void }; + + /** + * This instance's prefix in croner's PROCESS-GLOBAL name registry. + * + * croner keys named jobs in a module-level array shared by everything in the + * process, so a bare job name is a process-wide claim — which is why two + * environments in one container used to collide on the same AI-generated + * flow name with no kernel eviction involved at all, and why an evicted + * kernel's leftovers used to block every later rebind (#8362). Scoping the + * registry key to the adapter INSTANCE makes both collisions unreachable: + * one kernel builds one adapter, and a rebuilt kernel builds a new one. + */ + readonly registryNamespace: string; constructor(options: CronJobAdapterOptions = {}) { this.defaultTimezone = options.timezone ?? 'UTC'; this.maxExecutions = options.maxExecutions ?? 100; this.cluster = options.cluster; this.leaseMs = options.leaseMs ?? 60_000; + this.logger = options.logger; + this.registryNamespace = `${sanitizeNamespaceLabel(options.namespace)}#${++ADAPTER_SEQUENCE}.${Math.random() + .toString(36) + .slice(2, 8)}`; + } + + /** + * The name `jobName` is registered under in croner's process-global + * registry. Public because that registry is shared with everything else in + * the process: this is the only way an operator (or a test) can tell which + * entry of `scheduledJobs` belongs to which kernel. + */ + cronRegistryName(jobName: string): string { + return `${this.registryNamespace}::${jobName}`; } async schedule(name: string, schedule: JobSchedule, handler: JobHandler, options?: JobScheduleOptions): Promise { @@ -72,9 +125,11 @@ export class CronJobAdapter implements IJobService { if (!schedule.expression) { throw new Error(`CronJobAdapter: cron schedule for "${name}" missing expression`); } + const registryName = this.cronRegistryName(name); + this.reclaimRegistryName(registryName); const task = new Cron( schedule.expression, - { timezone: schedule.timezone ?? this.defaultTimezone, name }, + { timezone: schedule.timezone ?? this.defaultTimezone, name: registryName }, async () => { await this.runScheduled(name); }, ); record.task = task; @@ -119,7 +174,38 @@ export class CronJobAdapter implements IJobService { return [...this.jobs.keys()]; } - /** Stop all timers — call from plugin destroy. */ + /** + * Replace semantics for the process-global registry: if anything still holds + * the name we are about to claim, STOP it and take the name — never warn and + * give up, which is how a failed rebind used to end (#8362). + * + * Stopping is the whole point and not a detail. A leaked croner job is not + * merely holding a string: it is a live timer whose closure still references + * the kernel that created it. Taking the name while leaving that timer + * running would turn a silent death into a zombie double-write — two live + * jobs for one flow, one of them driving a shut-down kernel — which is + * strictly worse than the bug being fixed. `stop()` both kills the timer and + * splices the entry out of croner's registry, so the reclaim is complete. + * + * With per-instance namespacing our own adapters can no longer collide, so + * reaching this at all means a foreign holder — worth a line in the log. + */ + private reclaimRegistryName(registryName: string): void { + const holder = scheduledJobs.find((job) => job.name === registryName); + if (!holder) return; + try { holder.stop(); } catch { /* ignore — the retake below is what matters */ } + this.logger?.warn( + `CronJobAdapter: reclaimed croner job name "${registryName}" from a job this adapter did not schedule; ` + + 'the previous job was STOPPED and replaced.', + ); + } + + /** + * Stop all timers and release every process-global croner name this adapter + * holds. Called from `DbJobAdapter.destroy()` and `JobServicePlugin.destroy()` + * — i.e. from the kernel eviction chain, which until #8362 stopped one level + * above this method and left every evicted kernel's timers running forever. + */ async destroy(): Promise { for (const rec of this.jobs.values()) { try { rec.task?.stop(); } catch { /* ignore */ } diff --git a/packages/services/service-job/src/db-job-adapter.test.ts b/packages/services/service-job/src/db-job-adapter.test.ts index 2c11e19478..1cd0b9f6c1 100644 --- a/packages/services/service-job/src/db-job-adapter.test.ts +++ b/packages/services/service-job/src/db-job-adapter.test.ts @@ -1,7 +1,9 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { scheduledJobs } from 'croner'; import { DbJobAdapter } from './db-job-adapter.js'; +import { CronJobAdapter } from './cron-job-adapter.js'; function makeFakeEngine() { const tables = new Map(); @@ -142,3 +144,76 @@ describe('DbJobAdapter', () => { expect(triggers).toEqual(['replay', 'schedule']); }); }); + +// ─── #8362 — the destroy chain, and what an evicted kernel leaves behind ───── +// +// Kernel eviction is ROUTINE in the cloud runtime: a freshness probe runs every +// few seconds and every auto-publish bumps freshness, so the eviction chain +// `KernelManager.evict() -> kernel.shutdown() -> plugin.destroy() -> +// JobServicePlugin.destroy() -> dbAdapter.destroy()` runs constantly. It used +// to stop one level short — `destroy()` destroyed `inner` and never `cron` — so +// every evicted kernel left its croner timers running and holding their +// PROCESS-GLOBAL names, and the rebuilt kernel could never re-bind that flow +// again. The only signal was one WARN. +// +// Why the ordering of the two fixes matters, pinned by the second case below: +// the leaked job is not merely holding a name, it is still ALIVE with a closure +// over the shut-down kernel's engine. Namespacing the names WITHOUT closing the +// destroy chain would therefore convert a silent death into a zombie +// double-write — two live jobs, one driving a dead kernel. Hence the assertion +// is `oldJob.isStopped()`, not "a new job exists somewhere". +describe('DbJobAdapter — kernel rebuild (#8362)', () => { + /** Croner's process-global registry, narrowed to one PUBLIC job name. */ + const registeredFor = (jobName: string) => + scheduledJobs.filter((j) => (j.name ?? '').endsWith(jobName)); + + const DAILY = { type: 'cron', expression: '0 8 * * *' } as const; + + /** One kernel's job-service wiring: the pair JobServicePlugin builds. */ + function kernel() { + const cron = new CronJobAdapter(); + return { cron, db: new DbJobAdapter({ engine: makeFakeEngine(), cron }) }; + } + + it('destroy() destroys the CRON adapter too, freeing the process-global name', async () => { + const NAME = 'flow-time-relative:xqao_contract_expiry_reminder_flow'; + const k = kernel(); + await k.db.schedule(NAME, DAILY, async () => {}); + + const [job] = registeredFor(NAME); + expect(job, 'the first bind must register a REAL croner named job').toBeDefined(); + expect(job.isStopped()).toBe(false); + + // Exactly what the eviction chain reaches, one call short of which was the + // whole defect. + await k.db.destroy(); + + expect(job.isStopped()).toBe(true); + expect(registeredFor(NAME)).toHaveLength(0); + }); + + it('a rebuilt kernel re-binds the same flow: scheduled exactly once, and it fires', async () => { + const NAME = 'flow-time-relative:xqao_contract_expiry_reminder_flow'; + const fired: string[] = []; + + const old = kernel(); + await old.db.schedule(NAME, DAILY, async () => { fired.push('old-kernel'); }); + // Assert the FIRST bind landed before asserting anything about the second. + expect(registeredFor(NAME)).toHaveLength(1); + const oldJob = registeredFor(NAME)[0]; + + await old.db.destroy(); // kernel evicted by the freshness probe + + const rebuilt = kernel(); + await rebuilt.db.schedule(NAME, DAILY, async () => { fired.push('new-kernel'); }); + + const held = registeredFor(NAME); + expect(held).toHaveLength(1); // exactly once — not one live + one zombie + expect(oldJob.isStopped()).toBe(true); // the old job is STOPPED, not merely renamed around + + await held[0].trigger(); + expect(fired).toEqual(['new-kernel']); // the dead kernel's closure never runs + + await rebuilt.db.destroy(); + }); +}); diff --git a/packages/services/service-job/src/db-job-adapter.ts b/packages/services/service-job/src/db-job-adapter.ts index 454eec1996..110e4df6c5 100644 --- a/packages/services/service-job/src/db-job-adapter.ts +++ b/packages/services/service-job/src/db-job-adapter.ts @@ -191,8 +191,41 @@ export class DbJobAdapter implements IJobService { })); } + /** + * Release every timer this adapter owns — BOTH halves of it. + * + * This is the far end of the kernel eviction chain + * (`KernelManager.evict()` -> `kernel.shutdown()` -> `plugin.destroy()` -> + * `JobServicePlugin.destroy()` -> here), and eviction is routine rather than + * exceptional in the cloud runtime: a freshness probe runs every few seconds + * and every auto-publish bumps freshness. Until #8362 this method destroyed + * `inner` and never `cron`, so each evicted kernel left its croner timers + * running and holding their PROCESS-GLOBAL names for the life of the + * process. The rebuilt kernel then failed to bind that flow — permanently, + * reproduced across four consecutive rebuilds — and the only signal was one + * WARN from the trigger. + * + * `IJobService` does not declare `destroy()`, so the call is structural, + * exactly like the `cancel` forwarding above. + */ async destroy(): Promise { await this.inner.destroy(); + const cron = this.cron as (IJobService & { destroy?: () => Promise }) | undefined; + if (!cron || typeof cron.destroy !== 'function') return; + try { + await cron.destroy(); + } catch (err) { + // Runtime state now disagrees with every other surface: the kernel is + // gone, its cron timers are not, and nothing else in the system looks + // wrong — so this is the durability class, not the functional one. + const report = this.logger?.error?.bind(this.logger) ?? this.logger?.warn?.bind(this.logger); + report?.( + 'DbJobAdapter: the cron adapter failed to shut down — its croner jobs stay ALIVE holding their ' + + 'process-global names, so scheduled flows will silently fail to re-bind after this kernel is ' + + 'rebuilt, while every other surface keeps reporting them healthy. Restart the process to clear them.', + err as any, + ); + } } // ── Internals ──────────────────────────────────────────────────── diff --git a/packages/services/service-job/src/job-service-plugin.ts b/packages/services/service-job/src/job-service-plugin.ts index 19023e1bc1..3c1a5fd16d 100644 --- a/packages/services/service-job/src/job-service-plugin.ts +++ b/packages/services/service-job/src/job-service-plugin.ts @@ -16,6 +16,18 @@ function getClusterSafe(ctx: any): any { try { return ctx.getService('cluster'); } catch { return undefined; } } +/** + * Best-effort environment label for the cron adapter's entry in croner's + * process-global name registry — it makes `scheduledJobs` readable when several + * environments share one container. Cosmetic only: per-instance uniqueness is + * the adapter's own guarantee and does not depend on this resolving to + * anything (#8362). + */ +function environmentLabel(): string | undefined { + const raw = process.env.OS_ENVIRONMENT_ID?.trim(); + return raw ? raw : undefined; +} + export interface JobServicePluginOptions { /** * Job adapter type. @@ -62,6 +74,9 @@ export class JobServicePlugin implements Plugin { private readonly options: JobServicePluginOptions; private dbAdapter?: DbJobAdapter; private intervalAdapter?: IntervalJobAdapter; + /** Only set on the `adapter: 'cron'` path — otherwise the cron adapter is + * owned (and destroyed) by {@link DbJobAdapter}. */ + private cronAdapter?: CronJobAdapter; constructor(options: JobServicePluginOptions = {}) { this.options = { @@ -98,8 +113,16 @@ export class JobServicePlugin implements Plugin { } if (choice === 'cron') { - const cron = new CronJobAdapter({ timezone: 'UTC', cluster: getClusterSafe(ctx) }); - ctx.registerService('job', cron); + // Held on the instance so `destroy()` can reach it: this adapter owns + // process-global croner names, and a kernel evicted without releasing + // them blocks every later rebind of those jobs (#8362). + this.cronAdapter = new CronJobAdapter({ + timezone: 'UTC', + cluster: getClusterSafe(ctx), + namespace: environmentLabel(), + logger: ctx.logger, + }); + ctx.registerService('job', this.cronAdapter); ctx.logger.info('JobServicePlugin: registered CronJobAdapter'); return; } @@ -139,7 +162,12 @@ export class JobServicePlugin implements Plugin { let cron: CronJobAdapter | undefined; if (this.options.enableCron !== false) { try { - cron = new CronJobAdapter({ timezone: 'UTC', cluster: getClusterSafe(ctx) }); + cron = new CronJobAdapter({ + timezone: 'UTC', + cluster: getClusterSafe(ctx), + namespace: environmentLabel(), + logger: ctx.logger, + }); } catch (err) { ctx.logger.warn('JobServicePlugin: cron adapter init failed; cron jobs will not auto-run', err as any); } @@ -192,8 +220,17 @@ export class JobServicePlugin implements Plugin { }); } + /** + * Kernel eviction lands here. Every adapter this plugin built must be + * released, cron included: croner's named registry is process-global, so a + * timer that outlives its kernel keeps its name for the life of the process + * and blocks the rebuilt kernel from ever re-binding that job (#8362). + * `dbAdapter.destroy()` covers the cron adapter it owns; `cronAdapter` is + * the `adapter: 'cron'` path, where nothing else would. + */ async destroy(): Promise { await this.dbAdapter?.destroy(); await this.intervalAdapter?.destroy(); + await this.cronAdapter?.destroy(); } } diff --git a/packages/triggers/trigger-schedule/package.json b/packages/triggers/trigger-schedule/package.json index 1c18b5c86f..2ae8b746bc 100644 --- a/packages/triggers/trigger-schedule/package.json +++ b/packages/triggers/trigger-schedule/package.json @@ -24,6 +24,7 @@ "devDependencies": { "@objectstack/service-automation": "workspace:*", "@types/node": "^26.1.2", + "croner": "^10.0.1", "typescript": "^6.0.3", "vitest": "^4.1.10" }, diff --git a/packages/triggers/trigger-schedule/src/kernel-rebuild-rebind.test.ts b/packages/triggers/trigger-schedule/src/kernel-rebuild-rebind.test.ts new file mode 100644 index 0000000000..8ce6bc7d9c --- /dev/null +++ b/packages/triggers/trigger-schedule/src/kernel-rebuild-rebind.test.ts @@ -0,0 +1,228 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// #8362 — BOTH schedule triggers must survive a kernel rebuild. +// +// Kernel eviction is routine in the cloud runtime (a freshness probe every few +// seconds; every AI auto-publish bumps freshness), so "AI builds a scheduled +// automation -> the user edits one piece of metadata -> the automation is +// silently dead, permanently" was the NORMAL path, not an edge case. The +// four-layer chain behind it was: a job name with no kernel scope, an +// instance-local `bound` map that makes the rebuilt trigger's pre-bind cleanup +// a no-op, croner's process-global named registry, and — the single point — +// `DbJobAdapter.destroy()` never destroying the cron adapter. +// +// WHAT THIS FILE IS, AND IS NOT. The destroy chain and the registry mechanics +// are pinned where they live, against the real adapters and the real croner +// registry: `service-job/src/db-job-adapter.test.ts` and +// `cron-job-adapter.test.ts`. This file pins the TRIGGER half of the same +// scenario — that both triggers name their job deterministically enough to be +// re-bindable at all, that a rebuilt kernel ends up with exactly one live job +// which fires the NEW kernel's callback, and that a bind failure is reported +// where an operator will see it. +// +// The job service double is backed by REAL croner rather than a Map, on +// purpose: the card's own control experiment showed that an `interval` fixture +// bypasses croner's named registry entirely and would pass against a +// completely unfixed tree. Every case here goes through the `cron` path. + +import { describe, it, expect } from 'vitest'; +import { Cron, scheduledJobs } from 'croner'; +import type { AutomationContext, JobSchedule, JobHandler } from '@objectstack/spec/contracts'; +import { ScheduleTrigger, type FlowTriggerBinding, type JobServiceSurface, type TriggerLogger } from './schedule-trigger.js'; +import { TimeRelativeTrigger, type TimeRelativeDataEngine } from './time-relative-trigger.js'; + +const flush = () => new Promise((r) => setTimeout(r, 0)); + +let KERNEL_SEQ = 0; + +/** + * One kernel's job service: croner-backed, with the two properties the real + * `CronJobAdapter` guarantees — a registry key scoped to this instance, and a + * `destroy()` that STOPS every job it holds (which is what frees the name). + */ +function cronBackedJobService() { + const kernelId = `test-kernel-${++KERNEL_SEQ}`; + const jobs = new Map(); + const service: JobServiceSurface = { + async schedule(name, schedule: JobSchedule, handler: JobHandler) { + if (schedule.type !== 'cron' || !schedule.expression) { + throw new Error(`this fixture only exercises the cron path, got ${schedule.type}`); + } + jobs.get(name)?.stop(); + const job = new Cron( + schedule.expression, + { name: `${kernelId}::${name}` }, + async () => { await handler({ jobId: name }); }, + ); + jobs.set(name, job); + }, + async cancel(name) { + jobs.get(name)?.stop(); + jobs.delete(name); + }, + }; + return { + service, + /** What kernel eviction reaches: every timer stopped, every name freed. */ + async destroy() { + for (const job of jobs.values()) job.stop(); + jobs.clear(); + }, + }; +} + +/** croner's PROCESS-GLOBAL registry, narrowed to one job name. */ +const registeredFor = (jobName: string) => + scheduledJobs.filter((j) => (j.name ?? '').endsWith(jobName)); + +function recordingLogger(): TriggerLogger & { errors: string[]; warns: string[] } { + const errors: string[] = []; + const warns: string[] = []; + return { + errors, + warns, + info: () => {}, + debug: () => {}, + warn: (msg: string) => { warns.push(msg); }, + error: (msg: string) => { errors.push(msg); }, + }; +} + +const DAILY = { type: 'cron', expression: '0 8 * * *' } as const; + +/** A data engine with exactly one row in every window, so a sweep launches once. */ +function oneRowEngine(): TimeRelativeDataEngine { + return { + async find() { return [{ id: 'rec_1' }]; }, + getObject: (name: string) => ({ name }), + }; +} + +describe('#8362 — a rebuilt kernel re-binds scheduled flows (both triggers)', () => { + it('ScheduleTrigger: bind -> evict -> re-bind is scheduled exactly once and fires the NEW kernel', async () => { + const FLOW = 'nightly_contract_rollup'; + const JOB = `flow-schedule:${FLOW}`; + const fired: string[] = []; + const binding: FlowTriggerBinding = { flowName: FLOW, schedule: DAILY }; + + // ── kernel 1 ────────────────────────────────────────────────────── + const k1 = cronBackedJobService(); + const trigger1 = new ScheduleTrigger(() => k1.service, recordingLogger()); + trigger1.start(binding, async () => { fired.push('kernel-1'); }); + await flush(); + + // The FIRST bind must really have registered: a rebind pin whose first + // bind registered nothing passes for the wrong reason. + expect(registeredFor(JOB)).toHaveLength(1); + const oldJob = registeredFor(JOB)[0]; + + // ── eviction ────────────────────────────────────────────────────── + await k1.destroy(); + expect(oldJob.isStopped()).toBe(true); + expect(registeredFor(JOB)).toHaveLength(0); + + // ── kernel 2: fresh plugin instance, so the trigger's `bound` map is + // empty and its pre-bind `stop()` is a no-op — the shape that used to + // make the rebind unrecoverable. + const k2 = cronBackedJobService(); + const trigger2 = new ScheduleTrigger(() => k2.service, recordingLogger()); + trigger2.start(binding, async () => { fired.push('kernel-2'); }); + await flush(); + + const live = registeredFor(JOB); + expect(live).toHaveLength(1); // exactly once — not one live + one zombie + await live[0].trigger(); + expect(fired).toEqual(['kernel-2']); // the evicted kernel's closure never runs + + await k2.destroy(); + }); + + it('TimeRelativeTrigger: bind -> evict -> re-bind is scheduled exactly once and sweeps for the NEW kernel', async () => { + const FLOW = 'xqao_contract_expiry_reminder_flow'; + const JOB = `flow-time-relative:${FLOW}`; + const swept: string[] = []; + const binding: FlowTriggerBinding = { + flowName: FLOW, + schedule: DAILY, + config: { + timeRelative: { object: 'xqao_contract', dateField: 'expiry_date', offsetDays: [3] }, + }, + }; + + const k1 = cronBackedJobService(); + const trigger1 = new TimeRelativeTrigger(() => k1.service, oneRowEngine, recordingLogger()); + trigger1.start(binding, async (_ctx: AutomationContext) => { swept.push('kernel-1'); }); + await flush(); + + expect(registeredFor(JOB)).toHaveLength(1); + const oldJob = registeredFor(JOB)[0]; + + await k1.destroy(); + expect(oldJob.isStopped()).toBe(true); + + const k2 = cronBackedJobService(); + const trigger2 = new TimeRelativeTrigger(() => k2.service, oneRowEngine, recordingLogger()); + trigger2.start(binding, async (_ctx: AutomationContext) => { swept.push('kernel-2'); }); + await flush(); + + const live = registeredFor(JOB); + expect(live).toHaveLength(1); + await live[0].trigger(); + expect(swept).toEqual(['kernel-2']); + + await k2.destroy(); + }); +}); + +describe('#8362 — a failed bind is reported where an operator sees it', () => { + /** A job service whose schedule() always rejects, as croner did on a taken name. */ + const rejectingService = (): JobServiceSurface => ({ + async schedule(name) { + throw new Error(`Cron: Tried to initialize new named job '${name}', but name already taken.`); + }, + async cancel() {}, + }); + + it('ScheduleTrigger reports at ERROR, naming the consequence and the remedy', async () => { + const logger = recordingLogger(); + const trigger = new ScheduleTrigger(() => rejectingService(), logger); + trigger.start({ flowName: 'nightly_rollup', schedule: DAILY }, async () => {}); + await flush(); + + expect(logger.errors).toHaveLength(1); + const [line] = logger.errors; + // The failure itself, and the flow it belongs to. + expect(line).toContain("flow 'nightly_rollup'"); + expect(line).toContain('name already taken'); + // The consequence: everything else keeps looking healthy. + expect(line).toMatch(/stays published and active/); + // The remedy. + expect(line).toMatch(/Re-publish the flow/); + // A silent WARN was the whole problem — it must not degrade back to one. + expect(logger.warns).toHaveLength(0); + }); + + it('TimeRelativeTrigger reports at ERROR, naming the consequence and the remedy', async () => { + const logger = recordingLogger(); + const trigger = new TimeRelativeTrigger(() => rejectingService(), oneRowEngine, logger); + trigger.start( + { + flowName: 'xqao_contract_expiry_reminder_flow', + schedule: DAILY, + config: { + timeRelative: { object: 'xqao_contract', dateField: 'expiry_date', offsetDays: [3] }, + }, + }, + async () => {}, + ); + await flush(); + + expect(logger.errors).toHaveLength(1); + const [line] = logger.errors; + expect(line).toContain("flow 'xqao_contract_expiry_reminder_flow'"); + expect(line).toContain('name already taken'); + expect(line).toMatch(/stays published and active/); + expect(line).toMatch(/Re-publish the flow/); + expect(logger.warns).toHaveLength(0); + }); +}); diff --git a/packages/triggers/trigger-schedule/src/schedule-trigger.ts b/packages/triggers/trigger-schedule/src/schedule-trigger.ts index c17e8dcd21..ff9eb4d03d 100644 --- a/packages/triggers/trigger-schedule/src/schedule-trigger.ts +++ b/packages/triggers/trigger-schedule/src/schedule-trigger.ts @@ -58,6 +58,41 @@ export interface TriggerLogger { const JOB_PREFIX = 'flow-schedule'; +/** + * Report a scheduled flow that failed to bind to the job service. + * + * **Why this is `error` and not `warn`** — the repo's degradation-log-level + * rule (AGENTS.md) decides the level with one question: after the degradation, + * does the system still look normal from the outside while something it claims + * is in place has not landed? Here it does, completely: the flow stays + * published and active in `sys_metadata`, Studio lists it, the metadata API + * serves it and `verify_build` passes — while nothing will ever fire it. That + * is persisted state and runtime state disagreeing, which the rule puts in the + * `error` class, not the functional-degradation class. + * + * The neighbouring composition branch — "no job service is registered at all" — + * deliberately stays at `warn`: the system is *visibly* smaller and the rule + * names that exact message as correctly a `warn`. The distinction is not the + * severity of the outcome, it is whether the outside can see it. + * + * An `error` here owes two things, both in the first line it prints: the + * concrete consequence (including that everything else keeps looking healthy) + * and the remedy. Kept in one helper so both triggers say it the same way. + */ +export function reportBindFailure( + logger: TriggerLogger, + tag: 'schedule' | 'time-relative', + flowName: string, + err: unknown, +): void { + const report = logger.error?.bind(logger) ?? logger.warn.bind(logger); + report( + `[${tag}] flow '${flowName}' FAILED to bind to the job service: ${(err as Error)?.message ?? String(err)}. ` + + 'The flow stays published and active — Studio, the metadata API and verify_build all keep reporting it ' + + 'healthy — but nothing will fire it until it binds. Re-publish the flow (or restart the environment) to retry.', + ); +} + /** * Normalize a flow's raw `schedule` descriptor into a {@link JobSchedule}, or * `null` if it can't be understood. Accepts the canonical @@ -193,9 +228,7 @@ export class ScheduleTrigger implements FlowTrigger { }) .catch((err) => { this.bound.delete(binding.flowName); - this.logger.warn( - `[schedule] failed to schedule flow '${binding.flowName}': ${(err as Error)?.message ?? String(err)}`, - ); + reportBindFailure(this.logger, 'schedule', binding.flowName, err); }); } diff --git a/packages/triggers/trigger-schedule/src/time-relative-trigger.ts b/packages/triggers/trigger-schedule/src/time-relative-trigger.ts index 8cc6b3fd3d..d58d1ef081 100644 --- a/packages/triggers/trigger-schedule/src/time-relative-trigger.ts +++ b/packages/triggers/trigger-schedule/src/time-relative-trigger.ts @@ -7,7 +7,7 @@ import { TIME_RELATIVE_DEFAULT_MAX_RECORDS, } from '@objectstack/spec/automation'; import type { TimeRelativeTrigger as TimeRelativeDescriptor } from '@objectstack/spec/automation'; -import { normalizeSchedule } from './schedule-trigger.js'; +import { normalizeSchedule, reportBindFailure } from './schedule-trigger.js'; import type { FlowTrigger, FlowTriggerBinding, JobServiceSurface, TriggerLogger } from './schedule-trigger.js'; /** @@ -242,9 +242,7 @@ export class TimeRelativeTrigger implements FlowTrigger { }) .catch((err) => { this.bound.delete(binding.flowName); - this.logger.warn( - `[time-relative] failed to schedule flow '${binding.flowName}': ${errMessage(err)}`, - ); + reportBindFailure(this.logger, 'time-relative', binding.flowName, err); }); } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4e6827155c..d4267094cd 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2647,6 +2647,9 @@ importers: '@types/node': specifier: ^26.1.2 version: 26.1.2 + croner: + specifier: ^10.0.1 + version: 10.0.1 typescript: specifier: ^6.0.3 version: 6.0.3