fix(automation): release the wait node timer job when the run leaves the node (#5512)

.changeset/wait-timer-job-released-with-the-pause.md

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

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

@@ -140,6 +140,143 @@ describe('wait node executor', () => {
   });
 });
 
+/**
+ * #5512 — the one-shot wake-up job is dropped when the run leaves the wait node,
+ * whichever route it leaves by.
+ *
+ * The reported symptom: a `wait P1D` pause resumed early through the REST resume
+ * endpoint (a door the #3801 gate deliberately leaves open for `wait`) ran to
+ * completion while its `flow-wait:<runId>:<nodeId>` one-shot stayed `active` in
+ * `sys_job` with tomorrow's deadline — for 24h it read as "a run is still waiting
+ * to be woken", and then fired a ghost `resume` at a run that had completed the
+ * day before. Only the timer's OWN callback dropped its job.
+ *
+ * `cancelled` here is the fake job service's log of `IJobService.cancel(name)` —
+ * the call the DbJobAdapter turns into `active: false` on the `sys_job` row.
+ */
+describe('wait timer teardown when the pause ends another way (#5512)', () => {
+  let engine: AutomationEngine;
+  let ran: string[];
+
+  beforeEach(() => {
+    engine = new AutomationEngine(silentLogger());
+    ran = [];
+    engine.registerNodeExecutor(markerExecutor(ran));
+  });
+
+  it('cancels the one-shot when an external resume cuts a timer wait short', async () => {
+    const { ctx, scheduled, cancelled } = fakeJobCtx();
+    registerWaitNode(engine, ctx);
+    engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer', timerDuration: 'P1D' }));
+
+    const paused = await engine.execute('wait_flow');
+    expect(paused.status).toBe('paused');
+    expect(scheduled).toHaveLength(1); // armed for +24h
+    expect(cancelled).toEqual([]);
+
+    // The REST resume door: no signal, no job involvement — exactly the repro.
+    const resumed = await engine.resume(paused.runId!);
+    expect(resumed.success).toBe(true);
+    expect(ran).toEqual(['after']); // the run completed
+    expect(engine.listSuspendedRuns()).toEqual([]);
+
+    // …and tomorrow's wake-up is gone with it, instead of lingering `active`.
+    expect(cancelled).toEqual([scheduled[0].name]);
+    expect(scheduled[0].name).toBe(`flow-wait:${paused.runId}:pause`);
+  });
+
+  it('cancels the one-shot when the parked run is cancelled (ADR-0044)', async () => {
+    const { ctx, scheduled, cancelled } = fakeJobCtx();
+    registerWaitNode(engine, ctx);
+    engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer', timerDuration: 'P1D' }));
+
+    const paused = await engine.execute('wait_flow');
+    expect(await engine.cancelRun(paused.runId!, 'window abandoned')).toBe(true);
+
+    expect(cancelled).toEqual([scheduled[0].name]);
+    expect(ran).toEqual([]); // cancelled, not continued
+  });
+
+  it('cancels the re-armed one-shot too (cold boot, then an external resume)', async () => {
+    const store = new InMemorySuspendedRunStore();
+    const config = { eventType: 'timer', timerDuration: 'P1D' };
+
+    // Process 1: suspend at the wait, then "die".
+    const boot1 = fakeJobCtx();
+    const e1 = new AutomationEngine(silentLogger());
+    e1.registerNodeExecutor(markerExecutor([]));
+    registerWaitNode(e1, boot1.ctx);
+    e1.setSuspendedRunStore(store);
+    e1.registerFlow('wait_flow', waitFlow(config));
+    const paused = await e1.execute('wait_flow');
+
+    // Process 2: cold boot + re-arm, then someone resumes the run by hand.
+    const boot2 = fakeJobCtx();
+    registerWaitNode(engine, boot2.ctx);
+    engine.setSuspendedRunStore(store);
+    engine.registerFlow('wait_flow', waitFlow(config));
+    const job = boot2.ctx.getService('job') as IJobService;
+    expect(await rearmSuspendedWaitTimers(engine, store, job, silentLogger())).toBe(1);
+    expect(boot2.scheduled).toHaveLength(1);
+
+    const resumed = await engine.resume(paused.runId!);
+    expect(resumed.success).toBe(true);
+    expect(ran).toEqual(['after']);
+    // The re-armed job carries the same name, so the same teardown reaches it.
+    expect(boot2.cancelled).toEqual([`flow-wait:${paused.runId}:pause`]);
+  });
+
+  it('cancels nothing for a signal wait — it armed no job to cancel', async () => {
+    const { ctx, scheduled, cancelled } = fakeJobCtx();
+    registerWaitNode(engine, ctx);
+    engine.registerFlow('wait_flow', waitFlow({ eventType: 'signal', signalName: 'contract.renewed' }));
+
+    const paused = await engine.execute('wait_flow');
+    expect(scheduled).toEqual([]);
+    const resumed = await engine.resume(paused.runId!);
+
+    expect(resumed.success).toBe(true);
+    expect(ran).toEqual(['after']);
+    // The correlation of a signal wait is the AUTHOR's signal name, not a job
+    // name — the teardown must not hand it to `cancel()`.
+    expect(cancelled).toEqual([]);
+  });
+
+  it('cancels nothing for a timer wait that armed no job (no parseable duration)', async () => {
+    const { ctx, scheduled, cancelled } = fakeJobCtx();
+    registerWaitNode(engine, ctx);
+    // No `timerDuration` ⇒ no deadline ⇒ nothing scheduled; the pause carries
+    // the degraded `timer:<nodeId>` correlation instead of a job name.
+    engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer' }));
+
+    const paused = await engine.execute('wait_flow');
+    expect(scheduled).toEqual([]);
+    expect(engine.listSuspendedRuns()[0]).toMatchObject({ correlation: 'timer:pause' });
+
+    const resumed = await engine.resume(paused.runId!);
+    expect(resumed.success).toBe(true);
+    expect(cancelled).toEqual([]);
+  });
+
+  it('still cancels exactly once when the timer itself fires (idempotent teardown)', async () => {
+    const { ctx, scheduled, cancelled } = fakeJobCtx();
+    registerWaitNode(engine, ctx);
+    engine.registerFlow('wait_flow', waitFlow({ eventType: 'timer', timerDuration: 'PT2H' }));
+
+    const paused = await engine.execute('wait_flow');
+    await scheduled[0].handler({ jobId: scheduled[0].name });
+
+    expect(ran).toEqual(['after']);
+    // Two teardowns now cover this path — the release hook (the run left the
+    // node) and the one-shot's own `finally` (the job had its single shot) — and
+    // they target the same name. `cancel` is idempotent, so what is pinned is
+    // "cancelled, and nothing else cancelled"; the call COUNT is deliberately
+    // not pinned, since which of the two fires is not a behavioural promise.
+    expect(cancelled.length).toBeGreaterThan(0);
+    expect([...new Set(cancelled)]).toEqual([`flow-wait:${paused.runId}:pause`]);
+  });
+});
+
 /**
  * The loose `config.*` back door the executor used to read alongside
  * `waitEventConfig` graduated into the ADR-0087 D2 conversion layer

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

@@ -5,6 +5,17 @@ import { defineActionDescriptor } from '@objectstack/spec/automation';
 import type { IJobService } from '@objectstack/spec/contracts';
 import type { AutomationEngine, SuspendedRunStore } from '../engine.js';
 
+/**
+ * The one-shot wake-up job's name for a timer `wait` pause — and, by
+ * construction, the `correlation` that pause suspends with. One declaration, so
+ * the three sites that must agree on it cannot drift: the arming path, the
+ * cold-boot re-arm ({@link rearmSuspendedWaitTimers}), and the teardown when the
+ * run leaves the node (#5512).
+ */
+function waitTimerJobName(runId: string, nodeId: string): string {
+  return `flow-wait:${runId}:${nodeId}`;
+}
+
 /**
  * `wait` built-in node — a durable pause (ADR-0019 suspend/resume), the timer /
  * signal sibling of the human-input `screen` and `approval` nodes.
@@ -22,6 +33,10 @@ import type { AutomationEngine, SuspendedRunStore } from '../engine.js';
  *    the correlation key; an external producer resumes the run when the event
  *    arrives (`resume(runId)`), exactly like a decision-less approval.
  *
+ * 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.
+ *
  * 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).
  */
@@ -79,13 +94,19 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
 
         const job = getJobService();
         if (job && runId != null && at) {
-          const jobName = `flow-wait:${String(runId)}:${node.id}`;
+          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.
+                // 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 {
@@ -116,6 +137,34 @@ export function registerWaitNode(engine: AutomationEngine, ctx: PluginContext):
       const signal = String(wec.signalName ?? `wait:${node.id}`);
       return { success: true, suspend: true, correlation: signal };
     },
+
+    /**
+     * Disarm the one-shot wake-up when the run leaves this node by ANY route
+     * (#5512). Until this existed only the timer's own callback dropped its job,
+     * so a wait cut short — an external `resume` through the REST door (which
+     * the #3801 gate deliberately allows for `wait`), a `cancelRun`, a subflow
+     * ancestor failing — left the one-shot armed: it stayed `active` in
+     * `sys_job` with tomorrow's `schedule_expression`, read to every operator
+     * and test as "a run is still waiting to be woken", and eventually fired a
+     * ghost `resume` at a run that had completed the day before.
+     *
+     * The pause is already consumed when this runs, so cancelling cannot strand
+     * the run; and `cancel` on a name the job service no longer holds is a
+     * no-op, so a race with the timer's own teardown is harmless.
+     */
+    async onSuspensionReleased({ runId, nodeId, correlation }) {
+      // Only a pause that actually armed a job carries its name as the
+      // correlation. The degraded timer (`timer:<nodeId>`) and every signal wait
+      // (the author's own signal name) armed nothing, so there is nothing to
+      // cancel — and reconstructing the name we mint, rather than prefix-testing
+      // a string we may not own, keeps this from ever cancelling by coincidence.
+      if (correlation !== waitTimerJobName(runId, nodeId)) return;
+      const job = getJobService();
+      if (!job?.cancel) return;
+      // Errors propagate: the engine catches them and logs one line naming this
+      // correlation — which is the job name an operator would cancel by hand.
+      await job.cancel(correlation);
+    },
   });
 
   ctx.logger.info('[Wait Node] 1 built-in node executor registered');
@@ -217,7 +266,7 @@ export async function rearmSuspendedWaitTimers(
       continue;
     }
 
-    const jobName = `flow-wait:${run.runId}:${run.nodeId}`;
+    const jobName = waitTimerJobName(run.runId, run.nodeId);
     try {
       await job.schedule(jobName, { type: 'once', at: wakeAt }, async () => {
         try {