Skip to content

Commit 810c401

Browse files
os-zhuangclaude
andauthored
fix(automation): release a wait node's timer job when the run leaves the node (#5512) (#5527)
A timer `wait` arms a one-shot `flow-wait:<runId>:<nodeId>` job on entry, and only that job's own callback ever cancelled it. Every other exit from the pause left it armed: an early resume through the REST resume door (open for `wait` by the #3801 gate), `cancelRun` (ADR-0044), or a terminal failure under a subflow ancestor. The reported symptom was a `sys_job` row still `active: true` with tomorrow's deadline a day after its run had completed, followed by a ghost `resume` at that completed run. `NodeExecutor` gains an optional `onSuspensionReleased(release)` — the mirror of `suspend: true` — dispatched from `forgetSuspendedRun`, the one choke point every consumption of a suspension already passes through, and routed to the executor of the node that paused (recorded `nodeType` first, live flow as the fallback; deprecated ADR-0018 aliases delegate to their canonical). The `wait` node implements it by cancelling the one-shot whose name it recognises as its own, so a pause that armed nothing (signal wait, timer with no parseable duration) cancels nothing. Teardown runs after the suspension is consumed and its failures are logged, never propagated: it must not delay or fail the continuation. Claude-Session: https://claude.ai/code/session_01BWS4heBoAitLmzCLhcYdbK Co-authored-by: Claude <noreply@anthropic.com>
1 parent 152d7fb commit 810c401

6 files changed

Lines changed: 660 additions & 11 deletions

File tree

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
---
2+
"@objectstack/service-automation": minor
3+
---
4+
5+
fix(automation): a `wait` timer's wake-up job is dropped when the run leaves the node, not only when the timer fires (#5512)
6+
7+
A timer `wait` arms a one-shot job on entry (`flow-wait:<runId>:<nodeId>`,
8+
`{ type: 'once', at }`) and, until now, only that job's own callback ever tore it
9+
down. Every other way out of the pause left it armed:
10+
11+
- resumed early through the REST resume endpoint (`POST
12+
/api/v1/automation/:name/runs/:runId/resume` — a door the #3801 resume gate
13+
deliberately leaves open for `screen`/`wait` pauses) or the SDK equivalent;
14+
- cancelled while parked (`cancelRun`, ADR-0044);
15+
- terminally failed under a subflow ancestor.
16+
17+
Reported from 17.0-rc2 acceptance: a `wait P1D` pause resumed early ran to
18+
completion while its one-shot stayed `active: true` in `sys_job` with tomorrow's
19+
deadline. For the next 24h anyone reading `sys_job` saw "a run is still waiting
20+
to be woken" — the row contradicted the run — and when the deadline arrived the
21+
job fired a resume at a run that had completed the day before (harmless: the
22+
engine reports a machine-state error and the callback discards it, then the job
23+
self-cancels). A long-running org accumulated one stale row per early wake-up.
24+
25+
**What changed.** The engine now tells the node its pause is over. `NodeExecutor`
26+
gains an optional `onSuspensionReleased(release)` — the mirror of `suspend: true`
27+
— called from the single choke point every consumption of a suspension already
28+
passes through, with the `runId`, the node, the `correlation` the node minted at
29+
suspend time, and why the pause ended (`resumed` / `cancelled` / `failed`). The
30+
`wait` node implements it by cancelling the one-shot whose name it recognises as
31+
its own, so the `sys_job` row goes inactive the moment the run leaves the node,
32+
whichever route it left by. `SuspensionRelease` / `SuspensionReleaseReason` are
33+
exported for plugin nodes that arm something on entry (a lease, a reminder, a
34+
timeout) and need the same teardown.
35+
36+
Teardown is best-effort and runs after the suspension is consumed: a job service
37+
that is down or throwing can neither delay nor fail the continuation — the engine
38+
logs one warning naming the correlation an operator would cancel by hand. Node
39+
types that arm nothing are unaffected (the hook is optional), and a pause that
40+
armed no job — a signal wait, or a timer with no parseable duration — cancels
41+
nothing, since its correlation is not a job name. Deprecated ADR-0018 node
42+
aliases delegate the hook to their canonical executor, so authoring the old type
43+
name cannot silently lose the teardown.
44+
45+
The timer callback keeps its own `finally` cancel: the two answer different
46+
questions — "the run left the node" versus "this one-shot has had its single
47+
shot", including shots that did not consume a pause. `cancel` is idempotent.

packages/services/service-automation/src/builtin/wait-node.test.ts

Lines changed: 137 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,143 @@ describe('wait node executor', () => {
140140
});
141141
});
142142

143+
/**
144+
* #5512 — the one-shot wake-up job is dropped when the run leaves the wait node,
145+
* whichever route it leaves by.
146+
*
147+
* The reported symptom: a `wait P1D` pause resumed early through the REST resume
148+
* endpoint (a door the #3801 gate deliberately leaves open for `wait`) ran to
149+
* completion while its `flow-wait:<runId>:<nodeId>` one-shot stayed `active` in
150+
* `sys_job` with tomorrow's deadline — for 24h it read as "a run is still waiting
151+
* to be woken", and then fired a ghost `resume` at a run that had completed the
152+
* day before. Only the timer's OWN callback dropped its job.
153+
*
154+
* `cancelled` here is the fake job service's log of `IJobService.cancel(name)` —
155+
* the call the DbJobAdapter turns into `active: false` on the `sys_job` row.
156+
*/
157+
describe('wait timer teardown when the pause ends another way (#5512)', () => {
158+
let engine: AutomationEngine;
159+
let ran: string[];
160+
161+
beforeEach(() => {
162+
engine = new AutomationEngine(silentLogger());
163+
ran = [];
164+
engine.registerNodeExecutor(markerExecutor(ran));
165+
});
166+
167+
it('cancels the one-shot when an external resume cuts a timer wait short', async () => {
168+
const { ctx, scheduled, cancelled } = fakeJobCtx();
169+
registerWaitNode(engine, ctx);
170+
engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer', timerDuration: 'P1D' }));
171+
172+
const paused = await engine.execute('wait_flow');
173+
expect(paused.status).toBe('paused');
174+
expect(scheduled).toHaveLength(1); // armed for +24h
175+
expect(cancelled).toEqual([]);
176+
177+
// The REST resume door: no signal, no job involvement — exactly the repro.
178+
const resumed = await engine.resume(paused.runId!);
179+
expect(resumed.success).toBe(true);
180+
expect(ran).toEqual(['after']); // the run completed
181+
expect(engine.listSuspendedRuns()).toEqual([]);
182+
183+
// …and tomorrow's wake-up is gone with it, instead of lingering `active`.
184+
expect(cancelled).toEqual([scheduled[0].name]);
185+
expect(scheduled[0].name).toBe(`flow-wait:${paused.runId}:pause`);
186+
});
187+
188+
it('cancels the one-shot when the parked run is cancelled (ADR-0044)', async () => {
189+
const { ctx, scheduled, cancelled } = fakeJobCtx();
190+
registerWaitNode(engine, ctx);
191+
engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer', timerDuration: 'P1D' }));
192+
193+
const paused = await engine.execute('wait_flow');
194+
expect(await engine.cancelRun(paused.runId!, 'window abandoned')).toBe(true);
195+
196+
expect(cancelled).toEqual([scheduled[0].name]);
197+
expect(ran).toEqual([]); // cancelled, not continued
198+
});
199+
200+
it('cancels the re-armed one-shot too (cold boot, then an external resume)', async () => {
201+
const store = new InMemorySuspendedRunStore();
202+
const config = { eventType: 'timer', timerDuration: 'P1D' };
203+
204+
// Process 1: suspend at the wait, then "die".
205+
const boot1 = fakeJobCtx();
206+
const e1 = new AutomationEngine(silentLogger());
207+
e1.registerNodeExecutor(markerExecutor([]));
208+
registerWaitNode(e1, boot1.ctx);
209+
e1.setSuspendedRunStore(store);
210+
e1.registerFlow('wait_flow', waitFlow(config));
211+
const paused = await e1.execute('wait_flow');
212+
213+
// Process 2: cold boot + re-arm, then someone resumes the run by hand.
214+
const boot2 = fakeJobCtx();
215+
registerWaitNode(engine, boot2.ctx);
216+
engine.setSuspendedRunStore(store);
217+
engine.registerFlow('wait_flow', waitFlow(config));
218+
const job = boot2.ctx.getService('job') as IJobService;
219+
expect(await rearmSuspendedWaitTimers(engine, store, job, silentLogger())).toBe(1);
220+
expect(boot2.scheduled).toHaveLength(1);
221+
222+
const resumed = await engine.resume(paused.runId!);
223+
expect(resumed.success).toBe(true);
224+
expect(ran).toEqual(['after']);
225+
// The re-armed job carries the same name, so the same teardown reaches it.
226+
expect(boot2.cancelled).toEqual([`flow-wait:${paused.runId}:pause`]);
227+
});
228+
229+
it('cancels nothing for a signal wait — it armed no job to cancel', async () => {
230+
const { ctx, scheduled, cancelled } = fakeJobCtx();
231+
registerWaitNode(engine, ctx);
232+
engine.registerFlow('wait_flow', waitFlow({ eventType: 'signal', signalName: 'contract.renewed' }));
233+
234+
const paused = await engine.execute('wait_flow');
235+
expect(scheduled).toEqual([]);
236+
const resumed = await engine.resume(paused.runId!);
237+
238+
expect(resumed.success).toBe(true);
239+
expect(ran).toEqual(['after']);
240+
// The correlation of a signal wait is the AUTHOR's signal name, not a job
241+
// name — the teardown must not hand it to `cancel()`.
242+
expect(cancelled).toEqual([]);
243+
});
244+
245+
it('cancels nothing for a timer wait that armed no job (no parseable duration)', async () => {
246+
const { ctx, scheduled, cancelled } = fakeJobCtx();
247+
registerWaitNode(engine, ctx);
248+
// No `timerDuration` ⇒ no deadline ⇒ nothing scheduled; the pause carries
249+
// the degraded `timer:<nodeId>` correlation instead of a job name.
250+
engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer' }));
251+
252+
const paused = await engine.execute('wait_flow');
253+
expect(scheduled).toEqual([]);
254+
expect(engine.listSuspendedRuns()[0]).toMatchObject({ correlation: 'timer:pause' });
255+
256+
const resumed = await engine.resume(paused.runId!);
257+
expect(resumed.success).toBe(true);
258+
expect(cancelled).toEqual([]);
259+
});
260+
261+
it('still cancels exactly once when the timer itself fires (idempotent teardown)', async () => {
262+
const { ctx, scheduled, cancelled } = fakeJobCtx();
263+
registerWaitNode(engine, ctx);
264+
engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer', timerDuration: 'PT2H' }));
265+
266+
const paused = await engine.execute('wait_flow');
267+
await scheduled[0].handler({ jobId: scheduled[0].name });
268+
269+
expect(ran).toEqual(['after']);
270+
// Two teardowns now cover this path — the release hook (the run left the
271+
// node) and the one-shot's own `finally` (the job had its single shot) — and
272+
// they target the same name. `cancel` is idempotent, so what is pinned is
273+
// "cancelled, and nothing else cancelled"; the call COUNT is deliberately
274+
// not pinned, since which of the two fires is not a behavioural promise.
275+
expect(cancelled.length).toBeGreaterThan(0);
276+
expect([...new Set(cancelled)]).toEqual([`flow-wait:${paused.runId}:pause`]);
277+
});
278+
});
279+
143280
/**
144281
* The loose `config.*` back door the executor used to read alongside
145282
* `waitEventConfig` graduated into the ADR-0087 D2 conversion layer

packages/services/service-automation/src/builtin/wait-node.ts

Lines changed: 52 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,17 @@ import { defineActionDescriptor } from '@objectstack/spec/automation';
55
import type { IJobService } from '@objectstack/spec/contracts';
66
import type { AutomationEngine, SuspendedRunStore } from '../engine.js';
77

8+
/**
9+
* The one-shot wake-up job's name for a timer `wait` pause — and, by
10+
* construction, the `correlation` that pause suspends with. One declaration, so
11+
* the three sites that must agree on it cannot drift: the arming path, the
12+
* cold-boot re-arm ({@link rearmSuspendedWaitTimers}), and the teardown when the
13+
* run leaves the node (#5512).
14+
*/
15+
function waitTimerJobName(runId: string, nodeId: string): string {
16+
return `flow-wait:${runId}:${nodeId}`;
17+
}
18+
819
/**
920
* `wait` built-in node — a durable pause (ADR-0019 suspend/resume), the timer /
1021
* signal sibling of the human-input `screen` and `approval` nodes.
@@ -22,6 +33,10 @@ import type { AutomationEngine, SuspendedRunStore } from '../engine.js';
2233
* the correlation key; an external producer resumes the run when the event
2334
* arrives (`resume(runId)`), exactly like a decision-less approval.
2435
*
36+
* Whatever wakes the run, the one-shot job is dropped when the pause ends — see
37+
* `onSuspensionReleased` below (#5512). A timer wait cut short by an external
38+
* `resume` used to leave its wake-up armed for the full duration.
39+
*
2540
* Reads its own run id from the `$runId` variable the engine injects at start
2641
* (same mechanism the approval node uses to map external state back to the run).
2742
*/
@@ -79,13 +94,19 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
7994

8095
const job = getJobService();
8196
if (job && runId != null && at) {
82-
const jobName = `flow-wait:${String(runId)}:${node.id}`;
97+
const jobName = waitTimerJobName(String(runId), node.id);
8398
try {
8499
await job.schedule(jobName, { type: 'once', at }, async () => {
85100
try {
86101
await engine.resume(String(runId));
87102
} finally {
88-
// One-shot: drop the job so it never re-fires.
103+
// One-shot: drop the job so it never re-fires. Kept alongside
104+
// the `onSuspensionReleased` teardown below because the two
105+
// answer different questions: that one fires when the RUN
106+
// leaves the node, this one when the JOB has had its single
107+
// shot — including the shots that did not consume a pause (the
108+
// store was unreachable, another resume was already in
109+
// flight). Both are `cancel`, which is idempotent.
89110
try {
90111
await job.cancel?.(jobName);
91112
} catch {
@@ -116,6 +137,34 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
116137
const signal = String(wec.signalName ?? `wait:${node.id}`);
117138
return { success: true, suspend: true, correlation: signal };
118139
},
140+
141+
/**
142+
* Disarm the one-shot wake-up when the run leaves this node by ANY route
143+
* (#5512). Until this existed only the timer's own callback dropped its job,
144+
* so a wait cut short — an external `resume` through the REST door (which
145+
* the #3801 gate deliberately allows for `wait`), a `cancelRun`, a subflow
146+
* ancestor failing — left the one-shot armed: it stayed `active` in
147+
* `sys_job` with tomorrow's `schedule_expression`, read to every operator
148+
* and test as "a run is still waiting to be woken", and eventually fired a
149+
* ghost `resume` at a run that had completed the day before.
150+
*
151+
* The pause is already consumed when this runs, so cancelling cannot strand
152+
* the run; and `cancel` on a name the job service no longer holds is a
153+
* no-op, so a race with the timer's own teardown is harmless.
154+
*/
155+
async onSuspensionReleased({ runId, nodeId, correlation }) {
156+
// Only a pause that actually armed a job carries its name as the
157+
// correlation. The degraded timer (`timer:<nodeId>`) and every signal wait
158+
// (the author's own signal name) armed nothing, so there is nothing to
159+
// cancel — and reconstructing the name we mint, rather than prefix-testing
160+
// a string we may not own, keeps this from ever cancelling by coincidence.
161+
if (correlation !== waitTimerJobName(runId, nodeId)) return;
162+
const job = getJobService();
163+
if (!job?.cancel) return;
164+
// Errors propagate: the engine catches them and logs one line naming this
165+
// correlation — which is the job name an operator would cancel by hand.
166+
await job.cancel(correlation);
167+
},
119168
});
120169

121170
ctx.logger.info('[Wait Node] 1 built-in node executor registered');
@@ -217,7 +266,7 @@ export async function rearmSuspendedWaitTimers(
217266
continue;
218267
}
219268

220-
const jobName = `flow-wait:${run.runId}:${run.nodeId}`;
269+
const jobName = waitTimerJobName(run.runId, run.nodeId);
221270
try {
222271
await job.schedule(jobName, { type: 'once', at: wakeAt }, async () => {
223272
try {

0 commit comments

Comments
 (0)