diff --git a/packages/services/service-automation/src/builtin/index.ts b/packages/services/service-automation/src/builtin/index.ts index 64e42e80a9..b516a1bbcb 100644 --- a/packages/services/service-automation/src/builtin/index.ts +++ b/packages/services/service-automation/src/builtin/index.ts @@ -51,7 +51,7 @@ export { registerScreenNodes } from './screen-nodes.js'; export { registerHttpNodes } from './http-nodes.js'; export { registerConnectorNodes } from './connector-nodes.js'; export { registerNotifyNode } from './notify-node.js'; -export { registerWaitNode, parseIsoDuration } from './wait-node.js'; +export { registerWaitNode, parseIsoDuration, rearmSuspendedWaitTimers } from './wait-node.js'; export { registerSubflowNode } from './subflow-node.js'; /** diff --git a/packages/services/service-automation/src/builtin/wait-node.test.ts b/packages/services/service-automation/src/builtin/wait-node.test.ts index 1bf9988ce7..add4490824 100644 --- a/packages/services/service-automation/src/builtin/wait-node.test.ts +++ b/packages/services/service-automation/src/builtin/wait-node.test.ts @@ -3,7 +3,8 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { AutomationEngine } from '../engine.js'; import type { NodeExecutor } from '../engine.js'; -import { registerWaitNode, parseIsoDuration } from './wait-node.js'; +import { InMemorySuspendedRunStore } from '../suspended-run-store.js'; +import { registerWaitNode, parseIsoDuration, rearmSuspendedWaitTimers } from './wait-node.js'; import type { IJobService, JobHandler, JobSchedule } from '@objectstack/spec/contracts'; function silentLogger() { @@ -138,3 +139,93 @@ describe('wait node executor', () => { expect(ran).toEqual(['after']); }); }); + +describe('rearmSuspendedWaitTimers (cold-boot timer re-arm)', () => { + /** Boot a fresh engine wired to `store` with the wait flow registered — one "process". */ + function bootEngine(store: InMemorySuspendedRunStore, ctx: any, waitConfig: Record) { + const engine = new AutomationEngine(silentLogger()); + const ran: string[] = []; + engine.registerNodeExecutor(markerExecutor(ran)); + registerWaitNode(engine, ctx); + engine.setSuspendedRunStore(store); + engine.registerFlow('wait_flow', waitFlow(waitConfig)); + return { engine, ran }; + } + + it('re-schedules a future timer on a new engine and resumes when it fires', async () => { + const store = new InMemorySuspendedRunStore(); + const config = { eventType: 'timer', timerDuration: 'PT2H' }; + + // Process 1: suspend at the wait, then "die" (engine discarded). + const boot1 = fakeJobCtx(); + const a = bootEngine(store, boot1.ctx, config); + const paused = await a.engine.execute('wait_flow'); + expect(paused.status).toBe('paused'); + const storedAt = boot1.scheduled[0]?.schedule.at; + expect(storedAt).toBeTruthy(); + + // Process 2: cold boot — fresh engine + fresh (empty) job service. + const boot2 = fakeJobCtx(); + const b = bootEngine(store, boot2.ctx, config); + const job = boot2.ctx.getService('job') as IJobService; + const rearmed = await rearmSuspendedWaitTimers(b.engine, store, job, silentLogger()); + expect(rearmed).toBe(1); + + // Same one-shot job name + the persisted deadline (not a fresh now+2h). + expect(boot2.scheduled).toHaveLength(1); + expect(boot2.scheduled[0].name).toBe(`flow-wait:${paused.runId}:pause`); + expect(boot2.scheduled[0].schedule).toMatchObject({ type: 'once', at: storedAt }); + + // Firing the re-armed job resumes the run on the new engine. + await boot2.scheduled[0].handler({ jobId: boot2.scheduled[0].name }); + expect(b.ran).toEqual(['after']); + expect(await store.list()).toHaveLength(0); // consumed + }); + + it('resumes an overdue timer immediately (deadline elapsed while down)', async () => { + const store = new InMemorySuspendedRunStore(); + const config = { eventType: 'timer', timeoutMs: 1 }; + + const a = bootEngine(store, ctxNoJob(), config); // degraded: no job service + const paused = await a.engine.execute('wait_flow'); + expect(paused.status).toBe('paused'); + await new Promise((r) => setTimeout(r, 10)); // let the 1ms deadline lapse + + const boot2 = fakeJobCtx(); + const b = bootEngine(store, boot2.ctx, config); + const rearmed = await rearmSuspendedWaitTimers(b.engine, store, undefined, silentLogger()); + expect(rearmed).toBe(1); + expect(b.ran).toEqual(['after']); // resumed inline, no job needed + expect(boot2.scheduled).toHaveLength(0); + }); + + it('skips non-timer pauses (signal waits have no persisted deadline)', async () => { + const store = new InMemorySuspendedRunStore(); + const config = { eventType: 'signal', signalName: 'contract.renewed' }; + + const a = bootEngine(store, ctxNoJob(), config); + await a.engine.execute('wait_flow'); + + const boot2 = fakeJobCtx(); + const b = bootEngine(store, boot2.ctx, config); + const job = boot2.ctx.getService('job') as IJobService; + const rearmed = await rearmSuspendedWaitTimers(b.engine, store, job, silentLogger()); + expect(rearmed).toBe(0); + expect(boot2.scheduled).toHaveLength(0); + expect(await store.list()).toHaveLength(1); // still suspended, untouched + }); + + it('leaves a future timer suspended (with a warning) when no job service exists', async () => { + const store = new InMemorySuspendedRunStore(); + const config = { eventType: 'timer', timerDuration: 'PT2H' }; + + const a = bootEngine(store, ctxNoJob(), config); + await a.engine.execute('wait_flow'); + + const b = bootEngine(store, ctxNoJob(), config); + const rearmed = await rearmSuspendedWaitTimers(b.engine, store, undefined, silentLogger()); + expect(rearmed).toBe(0); + expect(b.ran).toEqual([]); + expect(await store.list()).toHaveLength(1); // resumable externally later + }); +}); diff --git a/packages/services/service-automation/src/builtin/wait-node.ts b/packages/services/service-automation/src/builtin/wait-node.ts index bd42121d14..47ecb29822 100644 --- a/packages/services/service-automation/src/builtin/wait-node.ts +++ b/packages/services/service-automation/src/builtin/wait-node.ts @@ -3,7 +3,7 @@ import type { PluginContext } from '@objectstack/core'; import { defineActionDescriptor } from '@objectstack/spec/automation'; import type { IJobService } from '@objectstack/spec/contracts'; -import type { AutomationEngine } from '../engine.js'; +import type { AutomationEngine, SuspendedRunStore } from '../engine.js'; /** * `wait` built-in node — a durable pause (ADR-0019 suspend/resume), the timer / @@ -62,10 +62,16 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext): (typeof wec.timeoutMs === 'number' ? wec.timeoutMs : undefined) ?? (typeof loose.timeoutMs === 'number' ? (loose.timeoutMs as number) : undefined); + // Persist the wake deadline as node output: the engine writes output + // to variables (`.waitUntil`) *before* snapshotting the + // suspended run, so a cold-booted kernel can re-arm the timer from the + // durable store ({@link rearmSuspendedWaitTimers}). + const at = durationMs && durationMs > 0 ? new Date(Date.now() + durationMs).toISOString() : undefined; + const output = at ? { waitUntil: at } : undefined; + const job = getJobService(); - if (job && runId != null && durationMs && durationMs > 0) { + if (job && runId != null && at) { const jobName = `flow-wait:${String(runId)}:${node.id}`; - const at = new Date(Date.now() + durationMs).toISOString(); try { await job.schedule(jobName, { type: 'once', at }, async () => { try { @@ -79,7 +85,7 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext): } } }); - return { success: true, suspend: true, correlation: jobName }; + return { success: true, suspend: true, correlation: jobName, output }; } catch (err) { ctx.logger.warn( `[wait] node '${node.id}': failed to schedule timer resume (${(err as Error)?.message ?? err}); ` + @@ -92,8 +98,9 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext): `(resume it via resume(runId), or install the job service for durable timers)`, ); } - // Degrade: still suspend; resumption comes from an external resume(). - return { success: true, suspend: true, correlation: `timer:${node.id}` }; + // Degrade: still suspend; resumption comes from an external resume() + // (or a later boot's re-arm pass, when the deadline was persisted). + return { success: true, suspend: true, correlation: `timer:${node.id}`, output }; } // signal / webhook / manual / condition — suspend; an external producer @@ -106,6 +113,93 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext): ctx.logger.info('[Wait Node] 1 built-in node executor registered'); } +/** Minimal logger surface for {@link rearmSuspendedWaitTimers}. */ +interface RearmLogger { + info(msg: string, ...args: unknown[]): void; + warn(msg: string, ...args: unknown[]): void; +} + +/** + * Re-arm auto-resume timers for suspended timer-`wait` runs after a cold boot + * (ADR-0019 follow-up). The one-shot job a `wait` node schedules lives in the + * job service's process memory unless that service is itself durable — so a + * restart loses the wake-up while the suspended run survives in + * `sys_automation_run`. This pass walks the durable store and: + * + * - **overdue** deadlines (`.waitUntil` in the past) → `resume()` now; + * - **future** deadlines → re-schedule the same `flow-wait::` + * one-shot job (no job service → warn and leave it for an external resume); + * - runs without a persisted `waitUntil` (approval / screen / signal pauses, + * or pre-deadline-persistence rows) → skipped untouched. + * + * Double-fire safe: if the original job *did* survive (durable job store), the + * second `resume(runId)` finds no suspended run — the engine's resume + * idempotency absorbs it. Returns how many runs were resumed or re-armed. + * + * Called by `AutomationServicePlugin.start()` *after* the flow pull, because + * `resume()` needs the flow definitions registered. + */ +export async function rearmSuspendedWaitTimers( + engine: Pick, + store: SuspendedRunStore, + job: IJobService | undefined, + logger: RearmLogger, +): Promise { + let runs; + try { + runs = await store.list(); + } catch (err) { + logger.warn(`[wait] timer re-arm: failed to list suspended runs: ${(err as Error)?.message ?? err}`); + return 0; + } + + let rearmed = 0; + for (const run of runs) { + const wakeAt = run.variables?.[`${run.nodeId}.waitUntil`]; + if (typeof wakeAt !== 'string' || !wakeAt) continue; // not a timer wait + const atMs = Date.parse(wakeAt); + if (Number.isNaN(atMs)) continue; + + if (atMs <= Date.now()) { + // Deadline elapsed while the process was down — resume immediately. + try { + await engine.resume(run.runId); + rearmed++; + } catch (err) { + logger.warn(`[wait] timer re-arm: resume of overdue run '${run.runId}' failed: ${(err as Error)?.message ?? err}`); + } + continue; + } + + if (!job) { + logger.warn( + `[wait] timer re-arm: run '${run.runId}' waits until ${wakeAt} but no job service is registered — ` + + `resume it externally via resume(runId)`, + ); + continue; + } + + const jobName = `flow-wait:${run.runId}:${run.nodeId}`; + try { + await job.schedule(jobName, { type: 'once', at: wakeAt }, async () => { + try { + await engine.resume(run.runId); + } finally { + try { + await job.cancel?.(jobName); + } catch { + /* best-effort */ + } + } + }); + rearmed++; + } catch (err) { + logger.warn(`[wait] timer re-arm: failed to re-schedule run '${run.runId}': ${(err as Error)?.message ?? err}`); + } + } + return rearmed; +} + /** * Parse an ISO-8601 duration (the subset flows use — weeks/days + a time part * of hours/minutes/seconds, e.g. `PT1H`, `P3D`, `PT90M`, `P1DT12H`) into diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index a3cf6a9d03..ed172ce48c 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -1,8 +1,9 @@ // Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license. import type { Plugin, PluginContext } from '@objectstack/core'; +import type { IJobService } from '@objectstack/spec/contracts'; import { AutomationEngine } from './engine.js'; -import { installBuiltinNodes } from './builtin/index.js'; +import { installBuiltinNodes, rearmSuspendedWaitTimers } from './builtin/index.js'; import { SysAutomationRun } from './sys-automation-run.object.js'; import { ObjectStoreSuspendedRunStore, type SuspendedRunStoreEngine } from './suspended-run-store.js'; @@ -112,9 +113,8 @@ export class AutomationServicePlugin implements Plugin { } async start(ctx: PluginContext): Promise { - console.warn('[Automation:start] entering start()'); if (!this.engine) { - console.warn('[Automation:start] engine missing, bailing'); + ctx.logger.warn('[Automation] start() called before init() — engine missing, skipping'); return; } @@ -131,12 +131,14 @@ export class AutomationServicePlugin implements Plugin { // services were wired, so we attach the DB-backed store here. Without an // engine (or with `suspendedRunStore: 'memory'`) the in-memory default // stands — suspended runs simply don't survive a restart. + let durableStore: ObjectStoreSuspendedRunStore | null = null; if ((this.options.suspendedRunStore ?? 'auto') !== 'memory') { let dataEngine: SuspendedRunStoreEngine | null = null; try { dataEngine = ctx.getService('objectql'); } catch { try { dataEngine = ctx.getService('data'); } catch { /* none */ } } if (dataEngine && typeof dataEngine.find === 'function' && typeof dataEngine.insert === 'function') { - this.engine.setSuspendedRunStore(new ObjectStoreSuspendedRunStore(dataEngine, ctx.logger)); + durableStore = new ObjectStoreSuspendedRunStore(dataEngine, ctx.logger); + this.engine.setSuspendedRunStore(durableStore); ctx.logger.info('[Automation] Suspended-run persistence enabled (sys_automation_run)'); } else { ctx.logger.info('[Automation] No ObjectQL engine — suspended runs kept in-memory only'); @@ -152,14 +154,14 @@ export class AutomationServicePlugin implements Plugin { registry?: { listItems?: (type: string) => unknown[] }; }>('objectql'); if (!ql) { - console.warn('[Automation] objectql service not found at start()'); + ctx.logger.debug('[Automation] objectql service not found at start()'); } else if (!ql.registry) { - console.warn('[Automation] objectql.registry is undefined at start()'); + ctx.logger.debug('[Automation] objectql.registry is undefined at start()'); } else if (typeof ql.registry.listItems !== 'function') { - console.warn('[Automation] objectql.registry.listItems is not a function'); + ctx.logger.debug('[Automation] objectql.registry.listItems is not a function'); } const flows = ql?.registry?.listItems?.('flow') ?? []; - console.warn(`[Automation] flow pull: registry returned ${flows.length} flow(s)`); + ctx.logger.debug(`[Automation] flow pull: registry returned ${flows.length} flow(s)`); let registered = 0; for (const f of flows) { const def = f as { name?: string }; @@ -179,6 +181,25 @@ export class AutomationServicePlugin implements Plugin { const msg = err instanceof Error ? err.message : String(err); ctx.logger.warn(`[Automation] flow pull from ObjectQL registry failed: ${msg}`); } + + // ADR-0019 follow-up: re-arm auto-resume timers for runs that were + // suspended at a timer-`wait` node when the process went down. Must run + // *after* the flow pull above — resume() needs the flow definitions + // registered. Overdue deadlines resume immediately; future ones get + // their one-shot job re-scheduled. Best-effort: a failure here only + // means those runs wait for an external resume(runId). + if (durableStore) { + let job: IJobService | undefined; + try { job = ctx.getService('job'); } catch { /* none */ } + try { + const rearmed = await rearmSuspendedWaitTimers(this.engine, durableStore, job, ctx.logger); + if (rearmed > 0) { + ctx.logger.info(`[Automation] Re-armed ${rearmed} suspended wait timer(s) after restart`); + } + } catch (err) { + ctx.logger.warn(`[Automation] wait-timer re-arm failed: ${(err as Error).message}`); + } + } } async destroy(): Promise {