fix(service-job): destroy the cron adapter on kernel eviction so scheduled flows re-bind (#8362)

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

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<typeof CronJobAdapter>[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);
+ });
+});

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<string, CronJobRecord>();
  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<void> {
@@ -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<void> {
  for (const rec of this.jobs.values()) {
  try { rec.task?.stop(); } catch { /* ignore */ }

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<string, any[]>();
@@ -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();
+ });
+});