diff --git a/.changeset/wait-timer-store-unavailable-keeps-job.md b/.changeset/wait-timer-store-unavailable-keeps-job.md new file mode 100644 index 0000000000..afffca410e --- /dev/null +++ b/.changeset/wait-timer-store-unavailable-keeps-job.md @@ -0,0 +1,46 @@ +--- +"@objectstack/service-automation": patch +--- + +fix(automation): a wait node's timer wake-up no longer disarms itself when the store outage means it never woke the run (#5529) + +A timer `wait` arms one job to wake its run. That job used to disarm itself in an +unconditional `finally` — and `AutomationEngine.resume()` reports failure by +**returning** a code rather than throwing, so "this shot consumed the pause" and +"this shot missed" were indistinguishable to that `finally`. Both were cancelled. + +On `STORE_UNAVAILABLE` that was a durability hole. The durable suspended-run +store being unreadable does **not** mean the run is gone (#4420 draws exactly +that line): the pause was never consumed, the run is still parked at its wait +node, and its row is still there — but the one job that was ever going to wake it +had just retired itself. Nothing then woke that run until the next process start, +where `rearmSuspendedWaitTimers` picks it up as overdue. A store that wobbled for +the one moment the deadline landed, plus no restart, meant a run parked forever. + +The one-shot now settles on the resume's return code: + +- **`STORE_UNAVAILABLE`** — the job stays armed, and the degradation is reported + at `error` (this path was previously silent — the result was discarded without + even a `warn`). The line names the job, the run, and both remedies. +- **everything else** — cancelled exactly as before: success consumed the pause, + `RESUME_IN_PROGRESS` means a concurrent resume is consuming it, a machine-state + failure means there is no pause left to serve, and a thrown error is not a + store outage. + +Keeping the job armed is **not** self-healing, and the log line says so rather +than implying a retry: a `once` schedule is a single `setTimeout`, so it never +re-fires on its own. What survival buys is the two things `cancel` destroys — the +`sys_job` row stays `active` with its deadline (true, here: the run really is +still waiting) instead of flipping to `active: false` and reading as "this +wake-up is done", and the registration stays in the job service, so +`trigger('flow-wait::')` re-fires that wake-up once the store is +back **without a restart**. After a cancel, `trigger` reports the job as not +found and a restart is the only path left. + +Both sites that arm this job — the wait node's own arming path and the cold-boot +re-arm — now share one handler, so they cannot drift, the same reason the job's +name is a single declaration. This is separate from the `onSuspensionReleased` +teardown added in #5512 and does not replace it: that one fires when the **run** +leaves the node, this one when the **job** has had its single shot. + +No authoring surface changes; no flow needs editing. 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 aba4180f90..6132f41ed4 100644 --- a/packages/services/service-automation/src/builtin/wait-node.test.ts +++ b/packages/services/service-automation/src/builtin/wait-node.test.ts @@ -277,6 +277,222 @@ describe('wait timer teardown when the pause ends another way (#5512)', () => { }); }); +/** + * The other half of "when may the one-shot disarm itself?" (#5529). + * + * #5512 gave the RUN-side question a hook (`onSuspensionReleased` — the pause + * ended, so drop the job). The JOB-side question stayed in the timer callback's + * `finally`, and that `finally` read nothing: `engine.resume()` reports failure + * by RETURNING a code, so a shot that consumed the pause and a shot that missed + * it looked identical, and both were cancelled. On `STORE_UNAVAILABLE` — the + * durable store unreadable, so per #4420 the pause is emphatically NOT gone — + * that cancelled the only thing left that would ever wake the run. + * + * Reachability is not equal across the two sites, and these tests are built to + * say so rather than to look symmetric: `resumeInternal` reads the durable store + * only on a hot-cache MISS, and a run that paused in this process stays cached + * for the life of its suspension. So the end-to-end specimen below is the + * **re-arm** callback (fresh process, empty cache, store consulted for real); + * the arming callback's branch is latent by construction and is pinned at the + * handler level, with the code injected rather than provoked. + */ +describe('wait timer one-shot vs. a shot that never consumed the pause (#5529)', () => { + /** A logger that keeps its `error` lines so the diagnostic can be asserted. */ + function capturingLogger() { + const errors: string[] = []; + const logger = { + info() {}, warn() {}, debug() {}, + error(msg: string) { errors.push(msg); }, + child() { return logger; }, + } as any; + return { logger, errors }; + } + + /** + * A durable store that is fully working except that `load` — the read + * `resumeInternal` makes on a cache miss — is unreachable. Everything else + * delegates, so the underlying rows stay inspectable: that is how these tests + * prove the pause SURVIVED the failed shot instead of assuming it. + */ + function storeWithUnreadableLoad(inner: InMemorySuspendedRunStore) { + return { + inner, + async save(run: any) { return inner.save(run); }, + async load(_runId: string): Promise { throw new Error('connection refused'); }, + async delete(runId: string) { return inner.delete(runId); }, + async list() { return inner.list(); }, + }; + } + + const config = { eventType: 'timer', timerDuration: 'P1D' }; + + /** + * Suspend a run in "process 1", then cold-boot "process 2" whose durable + * `load` is broken, and let its re-arm pass re-schedule the wake-up. Returns + * the re-armed job so a test can fire it. + */ + async function coldBootWithBrokenLoad() { + const inner = new InMemorySuspendedRunStore(); + const boot1 = fakeJobCtx(); + const e1 = new AutomationEngine(silentLogger()); + e1.registerNodeExecutor(markerExecutor([])); + registerWaitNode(e1, boot1.ctx); + e1.setSuspendedRunStore(inner); + e1.registerFlow('wait_flow', waitFlow(config)); + const paused = await e1.execute('wait_flow'); + expect(paused.status).toBe('paused'); + + // Process 2: same durable rows, but the resume-time read fails. + const broken = storeWithUnreadableLoad(inner); + const boot2 = fakeJobCtx(); + const ran: string[] = []; + const e2 = new AutomationEngine(silentLogger()); + e2.registerNodeExecutor(markerExecutor(ran)); + registerWaitNode(e2, boot2.ctx); + e2.setSuspendedRunStore(broken as any); + e2.registerFlow('wait_flow', waitFlow(config)); + + const { logger, errors } = capturingLogger(); + const job = boot2.ctx.getService('job') as IJobService; + // The deadline is +24h, so the re-arm re-schedules rather than resuming now. + expect(await rearmSuspendedWaitTimers(e2, broken as any, job, logger)).toBe(1); + expect(boot2.scheduled).toHaveLength(1); + expect(boot2.cancelled).toEqual([]); + + return { paused, inner, boot2, ran, errors, jobName: `flow-wait:${paused.runId}:pause` }; + } + + it('re-arm path: a STORE_UNAVAILABLE shot leaves the one-shot ARMED', async () => { + const { paused, inner, boot2, ran, jobName } = await coldBootWithBrokenLoad(); + + // The deadline arrives and the wake-up fires — into an unreachable store. + await boot2.scheduled[0].handler({ jobId: jobName }); + + // The pause was never consumed: the run is still parked, its row still there. + expect(ran).toEqual([]); + expect((await inner.list()).map((r) => r.runId)).toEqual([paused.runId]); + // …so the job that would wake it MUST survive. This is the regression: the + // unconditional `finally` cancelled here, and nothing would have re-armed + // until the next process start. + expect(boot2.cancelled).toEqual([]); + }); + + it('re-arm path: the failed shot is reported at error, naming the job and the run', async () => { + const { paused, boot2, errors, jobName } = await coldBootWithBrokenLoad(); + await boot2.scheduled[0].handler({ jobId: jobName }); + + // Previously silent: the callback discarded the result without a single line. + expect(errors).toHaveLength(1); + expect(errors[0]).toContain(jobName); + expect(errors[0]).toContain(paused.runId!); + // Both remedies an operator can act on, and the reason the job was kept. + expect(errors[0]).toMatch(/left ARMED on purpose/); + expect(errors[0]).toMatch(new RegExp(`trigger\\('${jobName}'\\)`)); + expect(errors[0]).toMatch(new RegExp(`resume\\('${paused.runId}'\\)`)); + expect(errors[0]).toContain('connection refused'); + }); + + it('re-arm path: a shot that DOES resume still disarms the one-shot (unchanged)', async () => { + // Same cold boot, working store — the branch must not have swallowed the + // ordinary teardown along with the failing one. + const inner = new InMemorySuspendedRunStore(); + const boot1 = fakeJobCtx(); + const e1 = new AutomationEngine(silentLogger()); + e1.registerNodeExecutor(markerExecutor([])); + registerWaitNode(e1, boot1.ctx); + e1.setSuspendedRunStore(inner); + e1.registerFlow('wait_flow', waitFlow(config)); + const paused = await e1.execute('wait_flow'); + + const ran: string[] = []; + const boot2 = fakeJobCtx(); + const e2 = new AutomationEngine(silentLogger()); + e2.registerNodeExecutor(markerExecutor(ran)); + registerWaitNode(e2, boot2.ctx); + e2.setSuspendedRunStore(inner); + e2.registerFlow('wait_flow', waitFlow(config)); + const { logger, errors } = capturingLogger(); + await rearmSuspendedWaitTimers(e2, inner, boot2.ctx.getService('job') as IJobService, logger); + + await boot2.scheduled[0].handler({ jobId: boot2.scheduled[0].name }); + + expect(ran).toEqual(['after']); + expect([...new Set(boot2.cancelled)]).toEqual([`flow-wait:${paused.runId}:pause`]); + expect(errors).toEqual([]); + }); + + it('arming path: the same handler keeps the job armed on STORE_UNAVAILABLE', async () => { + // The arming callback shares one handler with the re-arm callback, so this + // pins the branch on THAT site too. The code is injected, not provoked: a run + // that paused in this process is in the engine's hot cache, so its own resume + // never reads the durable store and cannot produce STORE_UNAVAILABLE here. + // Fabricating a cache miss to "prove" otherwise would pin a scenario the + // engine does not have — what is verified is the handler's branch, and that + // the arming site routes through it rather than keeping its own `finally`. + const { ctx, scheduled, cancelled } = fakeJobCtx(); + const engine = new AutomationEngine(silentLogger()); + const ran: string[] = []; + engine.registerNodeExecutor(markerExecutor(ran)); + registerWaitNode(engine, ctx); + engine.registerFlow('wait_flow', waitFlow(config)); + + const paused = await engine.execute('wait_flow'); + expect(scheduled).toHaveLength(1); + + engine.resume = async () => ({ + success: false, + code: 'STORE_UNAVAILABLE', + error: `Durable suspended-run store unreachable for run '${paused.runId}'`, + }); + await scheduled[0].handler({ jobId: scheduled[0].name }); + + expect(cancelled).toEqual([]); + expect(ran).toEqual([]); + }); + + it('RESUME_IN_PROGRESS still disarms — the other resume owns the pause', async () => { + const { ctx, scheduled, cancelled } = fakeJobCtx(); + const engine = new AutomationEngine(silentLogger()); + const ran: string[] = []; + let release: () => void = () => {}; + const gate = new Promise((r) => { release = r; }); + engine.registerNodeExecutor({ + type: 'mark', + async execute(node) { ran.push(node.id); await gate; return { success: true }; }, + }); + registerWaitNode(engine, ctx); + engine.registerFlow('wait_flow', waitFlow(config)); + const paused = await engine.execute('wait_flow'); + + // A concurrent resume claims the pause and parks inside the next node. + const inFlight = engine.resume(paused.runId!); + await new Promise((r) => setTimeout(r, 10)); + expect(ran).toEqual(['after']); // it really is mid-flight, holding the guard + + // Ignore the release-hook teardown that claim already triggered, so what is + // asserted below can only have come from the timer's own settle step. + cancelled.length = 0; + await scheduled[0].handler({ jobId: scheduled[0].name }); + expect(cancelled).toEqual([`flow-wait:${paused.runId}:pause`]); + + release(); + expect((await inFlight).success).toBe(true); + }); + + it('a thrown resume still disarms — a throw is not a store outage', async () => { + const { ctx, scheduled, cancelled } = fakeJobCtx(); + const engine = new AutomationEngine(silentLogger()); + engine.registerNodeExecutor(markerExecutor([])); + registerWaitNode(engine, ctx); + engine.registerFlow('wait_flow', waitFlow(config)); + await engine.execute('wait_flow'); + + engine.resume = async () => { throw new Error('boom'); }; + await expect(scheduled[0].handler({ jobId: scheduled[0].name })).rejects.toThrow('boom'); + expect(cancelled).toEqual([scheduled[0].name]); + }); +}); + /** * The loose `config.*` back door the executor used to read alongside * `waitEventConfig` graduated into the ADR-0087 D2 conversion layer diff --git a/packages/services/service-automation/src/builtin/wait-node.ts b/packages/services/service-automation/src/builtin/wait-node.ts index 40f3c425f9..e67155c8aa 100644 --- a/packages/services/service-automation/src/builtin/wait-node.ts +++ b/packages/services/service-automation/src/builtin/wait-node.ts @@ -16,6 +16,105 @@ function waitTimerJobName(runId: string, nodeId: string): string { return `flow-wait:${runId}:${nodeId}`; } +/** + * The `error` line {@link makeWaitTimerJobHandler} owes on the one outcome where + * the wake-up fires and the pause survives it (#5529) — and the same level + * {@link RearmLogger} requires for the re-arm path's degradations (#4632), + * which is why that one extends this. + */ +interface WaitTimerLogger { + error(msg: string, ...args: unknown[]): void; +} + +/** + * The one-shot wake-up job's handler. One declaration, shared by the arming path + * and the cold-boot re-arm ({@link rearmSuspendedWaitTimers}) for the same + * reason {@link waitTimerJobName} is one declaration: both sites schedule the + * same job and must settle it the same way. + * + * The job disarms **itself** after its single shot — right for every outcome but + * one. `engine.resume()` reports failure by RETURNING a code rather than + * throwing (`AutomationResult.code`), so "this shot consumed the pause" and + * "this shot missed" are indistinguishable unless the code is read. The bare + * `finally` this replaces read nothing and cancelled both (#5529): + * + * - **`STORE_UNAVAILABLE`** — the durable suspended-run store could not be + * READ, so the pause was **not** consumed: the run is still parked at its + * wait node and its row is still there. #4420 draws exactly this line — an + * unreachable store must never read as "no such run" — and cancelling here + * retires the only thing that was ever going to wake this run. So: keep the + * job, and say so at `error`. A run left parked while the process reports a + * healthy timer is the durability degradation AGENTS.md's log-level rule is + * about (#4632), and this path was previously silent — the result was + * discarded by the callback without so much as a `warn`. + * - **everything else** — cancel, exactly as before. Success consumed the + * pause; `RESUME_IN_PROGRESS` means a concurrent resume is consuming it (and + * #5512's `onSuspensionReleased` drops this job when it does); a machine-state + * failure (`RUN_NOT_FOUND`, a flow/node that no longer exists) means there is + * no pause left for this job to serve; and a *thrown* error is not a store + * outage, so it does not buy an exemption either. + * + * **Keeping the job armed is not self-healing** — measured, not assumed: a + * `once` schedule is a single `setTimeout` in `IntervalJobAdapter` (which + * `DbJobAdapter` delegates all timer mechanics to), so it never re-fires on its + * own. What survival buys is the two things `cancel` destroys. `sys_job` keeps + * an `active` row carrying the deadline — true here, the run really is still + * waiting — instead of flipping to `active: false`, which would read as "this + * wake-up is done" while the run hangs. And the registration stays in the + * adapter, so `IJobService.trigger(jobName)` re-fires this very wake-up once the + * store is back, with **no restart**; after a cancel, `trigger` throws "not + * found" and the only remaining path is the next boot's overdue re-arm pass. + * Both remedies are named in the log line for that reason. + * + * Reachability differs by site, and the honest note is that they are not equal. + * `resumeInternal` reads the durable store only on a hot-cache miss, and a run + * that paused in *this* process is cached for as long as the suspension lives — + * so the **re-arm** callback (a fresh process, empty cache) is where + * `STORE_UNAVAILABLE` is genuinely reachable today, while the arming callback's + * branch is latent by construction. It is shared anyway rather than special-cased: + * a second spelling of "settle the one-shot" is exactly the drift #5512 collapsed. + */ +function makeWaitTimerJobHandler( + engine: Pick, + job: IJobService, + runId: string, + jobName: string, + logger: WaitTimerLogger, +): () => Promise { + return async () => { + // Set only on the one outcome that must NOT disarm the job. A thrown + // `resume` leaves it false, so the `finally` still cancels. + let keepArmed = false; + try { + const result = await engine.resume(runId); + if (result?.code === 'STORE_UNAVAILABLE') { + keepArmed = true; + logger.error( + `[wait] timer wake-up '${jobName}' fired but could NOT resume run '${runId}': the durable suspended-run store was ` + + `unreachable, so the pause was never consumed — the run is STILL parked at its wait node, now past its deadline, ` + + `and this one-shot has already had its single shot, so nothing will wake it on its own. The job is left ARMED on ` + + `purpose (its 'sys_job' row stays active) so the stuck run stays visible and the wake-up re-firable: once the ` + + `store is reachable, re-fire it with the job service's trigger('${jobName}'), or resume the run directly via ` + + `resume('${runId}') — a process restart also picks it up as overdue. Cause: ${result.error ?? 'store unavailable'}`, + ); + } + } finally { + if (!keepArmed) { + // One-shot: drop the job so it never re-fires. Kept alongside the + // `onSuspensionReleased` teardown because the two answer different + // questions: that one fires when the RUN leaves the node, this one when + // the JOB has had its single shot *and* that shot settled the pause one + // way or the other. Both are `cancel`, which is idempotent. + try { + await job.cancel?.(jobName); + } catch { + /* best-effort */ + } + } + } + }; +} + /** * `wait` built-in node — a durable pause (ADR-0019 suspend/resume), the timer / * signal sibling of the human-input `screen` and `approval` nodes. @@ -35,7 +134,11 @@ function waitTimerJobName(runId: string, nodeId: string): string { * * Whatever wakes the run, the one-shot job is dropped when the pause ends — see * `onSuspensionReleased` below (#5512). A timer wait cut short by an external - * `resume` used to leave its wake-up armed for the full duration. + * `resume` used to leave its wake-up armed for the full duration. The timer's own + * callback settles its job separately and on a different question ("has this job + * had its shot, and did that shot settle the pause?") — see + * {@link makeWaitTimerJobHandler}, which keeps the job armed on the one outcome + * that fires without consuming the pause (#5529). * * Reads its own run id from the `$runId` variable the engine injects at start * (same mechanism the approval node uses to map external state back to the run). @@ -96,24 +199,11 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext): if (job && runId != null && at) { const jobName = waitTimerJobName(String(runId), node.id); try { - await job.schedule(jobName, { type: 'once', at }, async () => { - try { - await engine.resume(String(runId)); - } finally { - // One-shot: drop the job so it never re-fires. Kept alongside - // the `onSuspensionReleased` teardown below because the two - // answer different questions: that one fires when the RUN - // leaves the node, this one when the JOB has had its single - // shot — including the shots that did not consume a pause (the - // store was unreachable, another resume was already in - // flight). Both are `cancel`, which is idempotent. - try { - await job.cancel?.(jobName); - } catch { - /* best-effort */ - } - } - }); + await job.schedule( + jobName, + { type: 'once', at }, + makeWaitTimerJobHandler(engine, job, String(runId), jobName, ctx.logger), + ); return { success: true, suspend: true, correlation: jobName, output }; } catch (err) { ctx.logger.warn( @@ -170,17 +260,20 @@ 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 { +/** + * Minimal logger surface for {@link rearmSuspendedWaitTimers}. + * + * `error` comes from {@link WaitTimerLogger} and is required, not optional + * (#4632): every degradation on the re-arm path leaves a run + * persisted-but-unreachable, which is a durability degradation and must be + * reported at `error`; a logger that cannot carry that level could not satisfy + * the contract this function owes its caller. The jobs this pass re-arms carry + * the same obligation into their own callbacks (#5529), which is the other half + * of why the shape is shared. + */ +interface RearmLogger extends WaitTimerLogger { info(msg: string, ...args: unknown[]): void; warn(msg: string, ...args: unknown[]): void; - /** - * #4632 — required, not optional. Every degradation on the re-arm path leaves - * a run persisted-but-unreachable, which is a durability degradation and must - * be reported at `error`; a logger that cannot carry that level could not - * satisfy the contract this function owes its caller. - */ - error(msg: string, ...args: unknown[]): void; } /** @@ -268,17 +361,14 @@ export async function rearmSuspendedWaitTimers( const jobName = waitTimerJobName(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 */ - } - } - }); + await job.schedule( + jobName, + { type: 'once', at: wakeAt }, + // Same settle-the-one-shot rule as the arming path, and the site where + // `STORE_UNAVAILABLE` is actually reachable: this process's hot cache is + // empty, so the resume reads the durable store (#5529). + makeWaitTimerJobHandler(engine, job, run.runId, jobName, logger), + ); rearmed++; } catch (err) { // #4632 — the run is persisted and waiting, but its wake-up job was never