From d918beebef6f6e2409c1e41d3df7f261f6a0f8e8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 06:13:50 +0000 Subject: [PATCH 1/7] feat(automation): stamp status 'stranded' on the resume catch arm and pin the re-armed run's exactly-once (the #13937 services half) The shape-4 ruling keeps resumeInternal's consumption order; the state it leaves behind when a downstream node throws now carries the contract's name (AutomationResult.status: 'stranded', #14384) on the one exit that journals a consumed suspension. Stale "unruled" comments cite the ruling. The existing verb (restoreConsumedSuspension) is the exit; its double-run pins are added, including the cross-replica stale-journal case, which is RED at this commit by design (the precedence fix follows in the next commit). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .changeset/stranded-run-status-stamp.md | 40 ++ .../src/consumed-suspension-restore.test.ts | 34 +- .../services/service-automation/src/engine.ts | 111 ++++-- .../src/stranded-run-status.test.ts | 370 ++++++++++++++++++ 4 files changed, 507 insertions(+), 48 deletions(-) create mode 100644 .changeset/stranded-run-status-stamp.md create mode 100644 packages/services/service-automation/src/stranded-run-status.test.ts diff --git a/.changeset/stranded-run-status-stamp.md b/.changeset/stranded-run-status-stamp.md new file mode 100644 index 0000000000..33dd1377f9 --- /dev/null +++ b/.changeset/stranded-run-status-stamp.md @@ -0,0 +1,40 @@ +--- +"@objectstack/service-automation": minor +--- + +feat(automation): a resume that consumed the pause and then failed downstream answers `status: 'stranded'` (#13937) + +The services half of the #13937 shape-4 ruling (maintainer 2026-09-01): +`resumeInternal`'s consumption order is kept — the suspension is consumed +before downstream nodes run, which is what buys exactly-once across a crash — +and the state that order leaves behind when a downstream node throws now +carries the platform-level name #14384 put on the contract. + +`AutomationEngine.resume()` (and every engine continuation that reaches the +same catch arm) returns `{ success: false, status: 'stranded', … }` where it +returned no `status` at all. Stamped on that one exit only: the pause a +durable decision was waiting on is gone, the run is recorded `failed`, and it +can be re-armed only by the explicit operator verb +`restoreConsumedSuspension` (#13909 slice 2, already published) — never by +`resume` (which answers `RUN_NOT_FOUND`) and never automatically. Distinct +from `'failed'` on purpose: that one says the run ran and was rejected; this +one says a recorded continuation stopped mid-flight and an operator has +something to repair. The result's verdict and the restore verb are held to +agree by test: a stranded result is exactly a restorable run. + +Not changed: the run's RECORDED status (the run log, `getRun`, `listRuns`, the +durable `sys_automation_run` history row) stays `failed` — that vocabulary is +`ExecutionStatus` in `@objectstack/spec`, which the ruling did not widen; the +durable discriminator for the condition remains the snapshot the terminal row +carries. No resume semantics move for any pausing node type; shapes 2 and 3 +of the decision stay excluded. + +Also in this change, under the same ruling's exactly-once guarantee: +`restoreConsumedSuspension` now reads the run's durable terminal row BEFORE its +own per-process journal. Read the other way round, the replica that stranded a +run kept a hot copy after another replica restored, resumed and finished it, +and a repeated restore on the first replica re-armed the COMPLETED run — whose +next resume re-ran every node after the pause. Such a restore is now refused +(`RUN_COMPLETED`, or `NO_CONSUMED_SUSPENSION` when the durable row holds no +snapshot for any other reason). Single-process and store-less deployments +observe no difference. diff --git a/packages/services/service-automation/src/consumed-suspension-restore.test.ts b/packages/services/service-automation/src/consumed-suspension-restore.test.ts index 183141f9ef..8f3fb5770f 100644 --- a/packages/services/service-automation/src/consumed-suspension-restore.test.ts +++ b/packages/services/service-automation/src/consumed-suspension-restore.test.ts @@ -14,10 +14,13 @@ * the REST run surface has no cancel or retry route, and none of the engine's * public methods moved such a run out. * - * ⛔ **This file changes nothing about that ordering.** Which ordering is right - * is #13937, unruled and in the maintainer's hands. What is pinned here is the - * EXIT: `restoreConsumedSuspension` puts the consumed suspension back so the - * run is resumable again, under the ordering exactly as it is. + * ⛔ **This file changes nothing about that ordering.** #13937 ruled it (shape + * 4, maintainer 2026-09-01): the order stays, for exactly-once across a crash, + * and this verb is the operator exit. What is pinned here is that EXIT: + * `restoreConsumedSuspension` puts the consumed suspension back so the run is + * resumable again, under the ordering exactly as it is. The NAME the ruling + * gave the state (`AutomationResult.status: 'stranded'`) is pinned beside it + * in `stranded-run-status.test.ts`. * * ## What is pinned, and why each one is here * @@ -554,17 +557,19 @@ describe('#13909 — across a restart: the deployment shape this exists for', () }); }); -describe('#13909 — what this slice deliberately does NOT do', () => { - it('leaves AutomationResult.status and the run\'s recorded status alone', async () => { +describe('#13909 — what this verb deliberately does NOT do', () => { + it('names the condition on the RESULT only — the run\'s recorded status stays failed', async () => { const { engine } = newEngine(new InMemorySuspendedRunStore()); const started = await engine.execute('strand_flow', ctx); const runId = started.runId as string; const failed = await engine.resume(runId); - // ⛔ No new platform status is minted for the condition — naming it is - // an explicit same-batch sub-item of #13937, because what it should be - // called depends on which resume-ordering shape is ruled. - expect(failed.status).toBeUndefined(); + // The #13937 shape-4 ruling put the name on `AutomationResult.status` + // (#14384) and the catch arm now stamps it (the producer half, pinned + // in full in `stranded-run-status.test.ts`). It did NOT widen + // `ExecutionStatus`: the run's recorded lifecycle stays `failed`, in + // the log and in the durable history row. + expect(failed.status).toBe('stranded'); expect((await engine.getRun(runId))?.status).toBe('failed'); }); @@ -574,10 +579,11 @@ describe('#13909 — what this slice deliberately does NOT do', () => { const runId = started.runId as string; // The consumption still precedes the traversal: measured from inside - // the downstream node, the suspension is already gone. If a future - // change made the pause survive a downstream throw (#13937 shape 2), - // THIS is the assertion that should be reconsidered — deliberately, not - // by accident. + // the downstream node, the suspension is already gone. #13937 ruled + // this order KEPT (shape 4); shape 2 — the pause surviving a downstream + // throw — reopens only together with a durable claim/lease that keeps + // exactly-once, and THIS is the assertion such a change must flip + // deliberately, in the same batch, not by accident. let suspendedDuringTraversal: boolean | undefined; registerDownstream(engine, { throws: true, diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index e0dc235910..487afe03b4 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1282,12 +1282,14 @@ export interface RunRecord { * durable paused row is deleted at consumption and the terminal history row * carries no `variables` / `context` / `screen` at all. * - * ⛔ **This is not a run state and not a status.** The run is `failed`, - * {@link AutomationResult.status} is unchanged, and nothing here names the - * condition — naming it is an explicit same-batch sub-item of #13937, because - * what it should be called depends on which resume-ordering shape is ruled. - * This is a SNAPSHOT of a deleted row, kept so a deliberate operator action can - * restore it. + * ⛔ **This is not a run state.** The run's RECORDED status is `failed` + * (`ExecutionStatus`, the run-row vocabulary, is unchanged) and this is a + * SNAPSHOT of a deleted row, kept so a deliberate operator action can restore + * it. What the condition is CALLED was settled by the #13937 shape-4 ruling + * (maintainer 2026-09-01): `'stranded'`, on {@link AutomationResult.status} + * (#14384) — and the catch arm that writes this journal is the one exit that + * stamps it, so "a snapshot exists" and "the result said stranded" are one + * fact stated twice. */ export interface ConsumedSuspension { /** @@ -1355,9 +1357,10 @@ export type SuspensionRestoreRefusal = * * Deliberately NOT an {@link AutomationResult}: this verb does not execute a * flow, and its refusal vocabulary is its own. Folding it into the platform - * result type would put eight new codes into a contract every transport reads - * — and would mint platform vocabulary for a condition #13937 has not yet - * ruled the shape of. + * result type would put eight new codes into a contract every transport reads. + * The condition this verb exits from IS platform vocabulary since the #13937 + * shape-4 ruling — `AutomationResult.status: 'stranded'`, stamped by the + * resume that stranded the run — but the repair's own outcome stays here. */ export interface SuspensionRestoreResult { /** `true` only when a suspension was actually put back by THIS call. */ @@ -4919,8 +4922,9 @@ export class AutomationEngine implements IAutomationService { * * ⛔ NOT a re-ordering of the resume path. The suspension is still consumed * before `traverseNext` and {@link forgetSuspendedRun} is untouched — - * which ordering is right is #13937's question, and unruled. This changes - * only WHICH suspension is read, never when it is consumed. */ + * the #13937 shape-4 ruling (maintainer 2026-09-01) keeps that order, for + * exactly-once across a crash. This changes only WHICH suspension is + * read, never when it is consumed. */ private async loadSuspendedRunStrict(runId: string): Promise { if (!this.store) return this.suspendedRuns.get(runId) ?? null; const stored = await this.store.load(runId); @@ -5269,8 +5273,13 @@ export class AutomationEngine implements IAutomationService { // below still precedes the traversal, `forgetSuspendedRun` is // untouched, and `hasSuspendedRun` still answers false for the // whole traversal window (pinned in - // `consumed-suspension-restore.test.ts`). Which ordering is right - // is #13937's, and unruled. + // `consumed-suspension-restore.test.ts`). That order is RULED, + // not inherited: #13937 shape 4 (maintainer 2026-09-01) keeps it + // for exactly-once across a crash, excludes the flip (shape 2) + // unless a durable claim/lease lands with it, and excludes + // consume-then-re-arm (shape 3) outright. The exit from the state + // it leaves behind is `restoreConsumedSuspension`, and the state + // itself is named on the result (`status: 'stranded'`, below). const stepCountAtPause = run.steps.length; // Consume the suspension *before* running downstream work — a run @@ -5285,7 +5294,7 @@ export class AutomationEngine implements IAutomationService { // this answers "did any process get here first", against the shared // store, atomically. Placed exactly where the unconditional consume // was: every refusal above it still refuses without consuming, and - // the ordering #13937 has not ruled on is untouched. + // the ordering #13937 ruled to keep is untouched. const claim = await this.claimAdvance(run); if (claim.kind === 'lost') { // An ORDINARY outcome, not a degradation: this is the guard @@ -5437,15 +5446,15 @@ export class AutomationEngine implements IAutomationService { // above, this node merely threw, and the record below makes the // run terminal — so from here on nothing in the engine can move // it: `resume` answers RUN_NOT_FOUND, `cancelRun` is a no-op on - // it, and none of the other public methods takes it anywhere. - // Journal the suspension that was consumed so a DELIBERATE - // operator action can put it back - // ({@link restoreConsumedSuspension}). + // it, and no other public method takes it anywhere EXCEPT the + // deliberate operator exit, {@link restoreConsumedSuspension}. + // Journal the suspension that was consumed so that exit has + // something to put back. // // ⛔ Not a repair, and nothing here re-arms anything: the run - // stays `failed`, no suspension exists after this line, and no - // caller's result changes. It is the evidence a repair needs, - // written at the only moment it still exists. + // stays `failed`, no suspension exists after this line, and the + // ordering above is untouched. It is the evidence a repair + // needs, written at the only moment it still exists. const consumed = this.journalConsumedSuspension(run, stepCountAtPause, errorMessage); const logged = this.recordLog({ id: runId, @@ -5471,6 +5480,35 @@ export class AutomationEngine implements IAutomationService { success: false, error: errorMessage, durationMs, + // [#13937] STRANDED, said by the producer that knows — the + // services half of the shape-4 ruling (maintainer + // 2026-09-01): the ordering stays, and the condition it + // leaves behind carries the platform-level name #14384 put + // on the contract. Stamped HERE and on no other exit, + // because this is the one exit that journalled a snapshot + // two statements up: the pause a durable decision was + // waiting on is gone, the run is recorded `failed`, and it + // can be re-armed only by {@link restoreConsumedSuspension} + // — never by `resume` (which now answers RUN_NOT_FOUND), + // never automatically. Terminal exactly like `'failed'` + // and distinct from it on purpose: `'failed'` says the run + // ran and was rejected; this says a recorded continuation + // stopped mid-flight and an operator has something to + // repair. The verdict and the journal are one fact stated + // twice — `stranded-run-status.test.ts` holds them equal. + // + // ⚠️ The run's RECORDED status — the log entry above and + // the durable history row — stays `'failed'`: that is + // `ExecutionStatus` (`@objectstack/spec`, + // `automation/execution.zod.ts`), a vocabulary the ruling + // did not widen; the durable discriminator for this + // condition is the snapshot the terminal row carries. + // + // ⛔ The cascade-failed ancestors above are NOT stranded: + // `failSuspendedRun` consumes their pauses and journals + // nothing, so no verb can re-arm them. A different (and + // worse) condition, which this stamp must not claim. + status: 'stranded', errorMessage: flow.errorMessage, summary: logged.summary, }; @@ -5858,13 +5896,14 @@ export class AutomationEngine implements IAutomationService { * deployment could enter that state and never leave it. This verb is the * leaving. * - * ⚠️ It is a REPAIR, not a prevention: whether the pause should survive a - * downstream throw at all is #13937, unruled and in the maintainer's hands. - * This method changes no resume semantics for any pausing node type — the - * ordering, {@link forgetSuspendedRun} and `traverseNext` are all exactly - * as they were — and it stays useful whichever way that card is ruled, - * because the runs already stuck today are not released by changing what - * FUTURE resumes do. + * ⚠️ It is a REPAIR, not a prevention — and that is the ruled shape. + * #13937 (maintainer 2026-09-01, shape 4) keeps the consumption order for + * exactly-once across a crash and makes THIS verb the way out: the state + * it exits from is `AutomationResult.status: 'stranded'`, stamped by the + * resume that consumed the pause and then failed downstream. This method + * changes no resume semantics for any pausing node type — the ordering, + * {@link forgetSuspendedRun} and `traverseNext` are all exactly as they + * were. * * ## Deliberate — never automatic * @@ -5892,9 +5931,10 @@ export class AutomationEngine implements IAutomationService { * roll a transaction back. That is why re-deciding is the right default * and why this is an operator's call rather than the engine's. * - * Restoring the pause exactly as it stood is also the shape that does not - * pre-empt #13937: it is precisely the state a resume-ordering change would - * have left behind, so this composes with that ruling instead of racing it. + * Restoring the pause exactly as it stood is also the shape that composes + * with a later resume-ordering change (#13937 point 3 reopens shape 2 only + * with a durable claim/lease): it is precisely the state such a change + * would leave behind, so nothing here would need to move. * * ## Idempotence — carried by the paused row, not by a flag * @@ -6078,9 +6118,12 @@ export class AutomationEngine implements IAutomationService { // restart the DURABLE surfaces still read this run `failed`, because // `getRun` / `listRuns` deliberately let a terminal row win over a // paused one (a paused row can outlive the run it describes). The - // in-process log entry and the `warn` above are the trace this slice - // ships; a durable status that survives the restart is naming work, - // and naming is #13937's same-batch sub-item. + // in-process log entry and the `warn` above are the trace this + // ships. The #13937 ruling named the condition on the RESULT + // (`status: 'stranded'`, the resume's catch arm), not on the run + // row: `ExecutionStatus` carries no such member, so a durable + // status that survives the restart is a spec-vocabulary decision + // nobody has taken — the row's discriminator stays the snapshot. this.recordLog({ id: runId, flowName: consumed.run.flowName, diff --git a/packages/services/service-automation/src/stranded-run-status.test.ts b/packages/services/service-automation/src/stranded-run-status.test.ts new file mode 100644 index 0000000000..072fc2b2c1 --- /dev/null +++ b/packages/services/service-automation/src/stranded-run-status.test.ts @@ -0,0 +1,370 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #13937 — the services half of the shape-4 ruling (maintainer 2026-09-01). + * + * The ruling keeps `resumeInternal`'s consumption order (the suspension is + * consumed BEFORE the downstream nodes run, which is what buys exactly-once + * across a crash), makes an explicit operator verb the way out of the state + * that order leaves behind when a downstream node throws, and names that state + * on the contract: `AutomationResult.status: 'stranded'` (#14384). The verb + * already exists — `restoreConsumedSuspension`, #13909 slice 2, pinned in full + * in `consumed-suspension-restore.test.ts` — so what this file pins is the + * NAME, and the one property the ruling's whole point rests on: a re-armed run + * must never become double-runnable. + * + * 1. **The stamp lands on exactly one exit** — the catch arm that journalled + * a consumed suspension — and `'stranded'` and "restorable" are one fact: + * a result that says stranded is a run the verb accepts, a `'failed'` + * result is one it refuses. + * 2. **Controls**, so the stamp is a reading and not a constant: a run + * rejected at trigger time is `'failed'`; a resume refused BEFORE the + * consumption point carries no status and leaves the pause live; a resume + * that completes carries no status. + * 3. ⭐ **A re-armed run resumes exactly once.** Two concurrent resumes of a + * restored run — in one process, and on two replicas over one store — run + * the downstream node once more, not twice, and the loser is told. + * 4. ⭐ **A stale hot journal cannot re-arm a run that finished elsewhere.** + * The journal is a per-process cache; the durable terminal row is the + * record. The replica that stranded a run keeps a hot copy after another + * replica restored, resumed and finished it. Honouring that copy would + * re-arm a COMPLETED run, whose next resume re-runs everything after the + * pause — shape 2's silent double-run, through the repair verb's side + * door. The verb reads the durable row first, the same way + * `loadSuspendedRunStrict` reads the store first (#13617). Measured RED on + * the tree that read the hot copy first (both cases below), green after. + * 5. **Round trip** — a restored run that strands again is stranded again and + * restorable again: one strand per resume, each an operator's own call. + * 6. **The recorded `ExecutionStatus` stays `failed`** — the ruling widened + * the RESULT vocabulary; the run-row vocabulary is `@objectstack/spec`'s + * (`automation/execution.zod.ts`) and is untouched, in the log and in the + * durable history row. + */ + +import { describe, it, expect } from 'vitest'; + +import { AutomationEngine } from './engine.js'; +import type { SuspendedRunStore } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; + +const silent = { info() {}, warn() {}, error() {}, debug() {} } as never; + +/** + * `resumeAuthority: 'any'` because every continuation here goes through the + * public `resume` door (the gate itself is `resume-authority-gate.test.ts`'s). + */ +const holdDescriptor = defineActionDescriptor({ + type: 'hold', version: '1.0.0', name: 'hold', + supportsPause: true, resumeAuthority: 'any', +}); +const plain = (type: string) => defineActionDescriptor({ type, version: '1.0.0', name: type }); + +/** start → hold (pauses) → tail (the node that throws, or not) → end. */ +const STRAND_FLOW = { + name: 'strand_flow', label: 'Strand', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'hold', type: 'hold', label: 'Hold' }, + { id: 'tail', type: 'tail', label: 'Tail' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'hold' }, + { id: 'e2', source: 'hold', target: 'tail' }, + { id: 'e3', source: 'tail', target: 'end' }, + ], +}; + +/** No pause at all — the trigger-time control. */ +const NO_PAUSE_FLOW = { + name: 'no_pause_flow', label: 'No pause', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'boom', type: 'boom', label: 'Boom' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'boom' }, + { id: 'e2', source: 'boom', target: 'end' }, + ], +}; + +/** How many times the downstream node RAN — the observable a double-run moves. */ +interface Ledger { tail: number } +/** Whether the downstream node throws on its next run. Shared across replicas. */ +interface Knobs { throws: boolean } + +/** + * One engine over `store`, writing to the SHARED ledger and reading the SHARED + * knobs, so two of them stand in for two app replicas over one database. + */ +function replica(store: SuspendedRunStore | undefined, led: Ledger, knobs: Knobs): AutomationEngine { + const engine = new AutomationEngine(silent, store); + engine.registerNodeExecutor({ + type: 'hold', + descriptor: holdDescriptor, + async execute() { + return { success: true, suspend: true, correlation: 'approval:req_1' }; + }, + } as never); + engine.registerNodeExecutor({ + type: 'tail', + descriptor: plain('tail'), + async execute() { + led.tail++; + if (knobs.throws) throw new Error('tail blew up'); + return { success: true, output: { done: true } }; + }, + } as never); + engine.registerNodeExecutor({ + type: 'boom', + descriptor: plain('boom'), + async execute() { throw new Error('never paused, just broke'); }, + } as never); + engine.registerFlow('strand_flow', STRAND_FLOW as never); + engine.registerFlow('no_pause_flow', NO_PAUSE_FLOW as never); + return engine; +} + +const ctx = { event: 'test', record: { id: 'rec_1' } } as unknown as AutomationContext; + +function harness(store?: SuspendedRunStore) { + const led: Ledger = { tail: 0 }; + const knobs: Knobs = { throws: true }; + return { led, knobs, engine: replica(store, led, knobs) }; +} + +/** Park a run, then resume it into the downstream throw. Returns id + the result. */ +async function strand(engine: AutomationEngine, knobs: Knobs) { + knobs.throws = true; + const started = await engine.execute('strand_flow', ctx); + expect(started.status).toBe('paused'); + const runId = started.runId as string; + const failed = await engine.resume(runId); + expect(failed.success).toBe(false); + return { runId, failed }; +} + +describe('#13937 — the resume that consumed the pause and failed downstream says so: status stranded', () => { + it('stamps stranded on that exit, and that run is exactly the one the operator verb accepts', async () => { + const store = new InMemorySuspendedRunStore(); + const { engine, knobs, led } = harness(store); + const { runId, failed } = await strand(engine, knobs); + + // The verdict, by name — not inferred from `success` plus a missing + // suspension, which is what every consumer had to do before. + expect(failed.status).toBe('stranded'); + expect(failed.error).toContain('tail blew up'); + // …and it is TERMINAL exactly like `failed`: the pause is gone and + // `resume` cannot continue it. + expect(await engine.hasSuspendedRun(runId)).toBe(false); + expect((await engine.resume(runId)).code).toBe('RUN_NOT_FOUND'); + expect(led.tail).toBe(1); + + // One fact stated twice: "the result said stranded" ⇔ "a snapshot was + // journalled for the verb to put back". + const restored = await engine.restoreConsumedSuspension(runId, { requestedBy: 'ops' }); + expect(restored.restored).toBe(true); + expect(restored.refusal).toBeUndefined(); + expect(await engine.hasSuspendedRun(runId)).toBe(true); + }); + + it('CONTROL — a run rejected at trigger time is failed, never stranded, and the verb refuses it', async () => { + const { engine } = harness(new InMemorySuspendedRunStore()); + + const rejected = await engine.execute('no_pause_flow', ctx); + expect(rejected.success).toBe(false); + // The producer's existing verdict for "ran and was rejected" (#9378). + expect(rejected.status).toBe('failed'); + + const runId = (await engine.listRuns('no_pause_flow'))[0]?.id as string; + expect(runId).toBeTruthy(); + const res = await engine.restoreConsumedSuspension(runId); + expect(res.restored).toBe(false); + expect(res.refusal).toBe('NO_CONSUMED_SUSPENSION'); + }); + + it('CONTROL — a resume refused BEFORE the consumption point carries no status and leaves the pause live', async () => { + const { engine, knobs, led } = harness(new InMemorySuspendedRunStore()); + const started = await engine.execute('strand_flow', ctx); + const runId = started.runId as string; + + // Raised while folding the signal — above the consumption point. + const refused = await engine.resume(runId, { variables: { $internal: 1 } } as never); + expect(refused.success).toBe(false); + expect(refused.code).toBe('INVALID_SIGNAL'); + expect(refused.status).toBeUndefined(); + expect(led.tail).toBe(0); + // Nothing to repair: the pause survived, and the verb says exactly that. + expect(await engine.hasSuspendedRun(runId)).toBe(true); + expect((await engine.restoreConsumedSuspension(runId)).refusal).toBe('RUN_SUSPENDED'); + + // And the SAME run, resumed cleanly, carries no status on completion. + knobs.throws = false; + const done = await engine.resume(runId); + expect(done.success).toBe(true); + expect(done.status).toBeUndefined(); + expect(led.tail).toBe(1); + }); + + it('the recorded ExecutionStatus stays failed — in the log and in the durable history row', async () => { + const store = new InMemorySuspendedRunStore(); + const { engine, knobs } = harness(store); + const { runId, failed } = await strand(engine, knobs); + expect(failed.status).toBe('stranded'); + + // The ruling named the condition on the RESULT; `ExecutionStatus` (the + // run-row vocabulary, `@objectstack/spec`) carries no such member, and + // this seat does not mint one. The durable discriminator for the + // condition is the snapshot the terminal row carries. + expect((await engine.getRun(runId))?.status).toBe('failed'); + const row = await store.loadTerminal(runId); + expect(row?.status).toBe('failed'); + expect(row?.consumedSuspension).toBeDefined(); + expect((await engine.listRuns('strand_flow', { status: 'failed' })).map(r => r.id)).toContain(runId); + }); +}); + +describe('#13937 — a re-armed run is not double-runnable', () => { + it('⭐ two concurrent resumes of a restored run, one process: the tail runs once more and the loser is told', async () => { + const store = new InMemorySuspendedRunStore(); + const { engine, knobs, led } = harness(store); + const { runId } = await strand(engine, knobs); + expect(led.tail).toBe(1); + + expect((await engine.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throws = false; + + const [x, y] = await Promise.all([engine.resume(runId), engine.resume(runId)]); + const winners = [x, y].filter(r => r.success); + const losers = [x, y].filter(r => !r.success); + expect(winners).toHaveLength(1); + expect(losers).toHaveLength(1); + expect(losers[0]!.code).toBe('RESUME_IN_PROGRESS'); + expect(losers[0]!.status).toBeUndefined(); + + // THE OBSERVABLE: one strand, one continuation. Not three. + expect(led.tail).toBe(2); + expect((await engine.getRun(runId))?.status).toBe('completed'); + + // And the run is finished for good — neither verb re-opens it. + expect((await engine.restoreConsumedSuspension(runId)).refusal).toBe('RUN_COMPLETED'); + expect((await engine.resume(runId)).code).toBe('RUN_NOT_FOUND'); + expect(led.tail).toBe(2); + }); + + it('⭐ two concurrent resumes of a restored run, two replicas over one store: the tail runs once more', async () => { + const store = new InMemorySuspendedRunStore(); + const led: Ledger = { tail: 0 }; + const knobs: Knobs = { throws: true }; + const a = replica(store, led, knobs); + const b = replica(store, led, knobs); + + const { runId } = await strand(a, knobs); + expect(led.tail).toBe(1); + + // Restored on the OTHER replica, from the durable row alone. + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throws = false; + + // Each replica's own `resuming` set is empty; only the store's + // conditional consume (#14333) separates them. + const [byA, byB] = await Promise.all([a.resume(runId), b.resume(runId)]); + expect([byA, byB].filter(r => r.success)).toHaveLength(1); + expect([byA, byB].filter(r => r.code === 'RESUME_IN_PROGRESS')).toHaveLength(1); + + expect(led.tail).toBe(2); + expect(await store.list()).toHaveLength(0); + const row = await store.loadTerminal(runId); + expect(row?.status).toBe('completed'); + expect(row?.consumedSuspension).toBeUndefined(); + }); + + it('⭐ a stale hot journal on the replica that stranded the run cannot re-arm it after another replica restored and finished it', async () => { + const store = new InMemorySuspendedRunStore(); + const led: Ledger = { tail: 0 }; + const knobs: Knobs = { throws: true }; + const a = replica(store, led, knobs); + const b = replica(store, led, knobs); + + // A strands the run: A's hot journal now holds the snapshot, and so + // does the durable terminal row. + const { runId } = await strand(a, knobs); + expect(led.tail).toBe(1); + + // B — a support engineer's replica, hours later — restores and + // finishes it. The durable row now says `completed`, snapshot cleared. + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throws = false; + expect((await b.resume(runId)).success).toBe(true); + expect(led.tail).toBe(2); + expect((await store.loadTerminal(runId))?.consumedSuspension).toBeUndefined(); + + // The SAME restore request lands on A (an at-least-once delivery, a + // second operator, a retry). A still holds its hot copy. Honouring it + // would re-arm a completed run. + const stale = await a.restoreConsumedSuspension(runId, { requestedBy: 'ops-retry' }); + expect(stale.restored).toBe(false); + expect(stale.refusal).toBe('RUN_COMPLETED'); + + // Nothing was minted, on either replica's reading… + expect(await store.list()).toHaveLength(0); + expect(await a.hasSuspendedRun(runId)).toBe(false); + expect(await b.hasSuspendedRun(runId)).toBe(false); + // …so the tail cannot run a third time through any door. + expect((await a.resume(runId)).code).toBe('RUN_NOT_FOUND'); + expect((await b.resume(runId)).code).toBe('RUN_NOT_FOUND'); + expect(led.tail).toBe(2); + }); + + it('⭐ the same stale copy is refused when the run was CANCELLED elsewhere after the restore', async () => { + const store = new InMemorySuspendedRunStore(); + const led: Ledger = { tail: 0 }; + const knobs: Knobs = { throws: true }; + const a = replica(store, led, knobs); + const b = replica(store, led, knobs); + + const { runId } = await strand(a, knobs); + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + // Somebody decided, on purpose, that this run ends here. + expect(await b.cancelRun(runId, 'submitter withdrew')).toBe(true); + expect((await store.loadTerminal(runId))?.consumedSuspension).toBeUndefined(); + + const stale = await a.restoreConsumedSuspension(runId); + expect(stale.restored).toBe(false); + // A's own log still says `failed` for this run and the durable row + // records a cancelled run as `failed` too, so A cannot name the + // cancellation — what it CAN say, honestly, is that no snapshot is + // held any more. ⛔ Never `RUN_SUSPENDED`, and never `restored: true`. + expect(stale.refusal).toBe('NO_CONSUMED_SUSPENSION'); + expect(await store.list()).toHaveLength(0); + expect((await a.resume(runId)).code).toBe('RUN_NOT_FOUND'); + expect(led.tail).toBe(1); + }); + + it('a restored run that strands AGAIN is stranded again and restorable again — one strand per resume', async () => { + const store = new InMemorySuspendedRunStore(); + const { engine, knobs, led } = harness(store); + const { runId } = await strand(engine, knobs); + expect(led.tail).toBe(1); + + expect((await engine.restoreConsumedSuspension(runId)).restored).toBe(true); + // The operator's retry hits the same broken dependency. + const again = await engine.resume(runId); + expect(again.success).toBe(false); + expect(again.status).toBe('stranded'); + expect(led.tail).toBe(2); + expect(await engine.hasSuspendedRun(runId)).toBe(false); + + // Nothing sweeps, nothing retries by itself — and the exit still works. + expect((await engine.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throws = false; + const done = await engine.resume(runId); + expect(done.success).toBe(true); + expect(done.status).toBeUndefined(); + expect(led.tail).toBe(3); + expect((await engine.getRun(runId))?.status).toBe('completed'); + }); +}); From 8c13b3cbd2f2b61b2f9c8a8461d4f3018212c616 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 06:30:36 +0000 Subject: [PATCH 2/7] fix(automation): restoreConsumedSuspension reads the durable terminal row before its per-process journal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A hot copy left on the replica that stranded a run could re-arm the run after another replica restored, resumed and finished it — the next resume then re-ran every node after the pause. The terminal row is the record; a row that exists without a snapshot is the last word and the hot copy is dropped. The two stale-journal pins added in the previous commit go red -> green here. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .../services/service-automation/src/engine.ts | 54 ++++++++++++++++--- 1 file changed, 48 insertions(+), 6 deletions(-) diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 487afe03b4..0f5915c44f 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -5946,6 +5946,14 @@ export class AutomationEngine implements IAutomationService { * `restored: true` and one `RESTORE_IN_PROGRESS`, rather than two calls * both claiming the restore. * + * [#13937] The third half, across replicas and across TIME: the snapshot + * is read from the durable terminal row before this process's own + * journal, so a hot copy left behind on the replica that stranded the run + * cannot re-arm it after another replica restored and finished it — the + * row that no longer carries a snapshot is the record, and the copy is + * dropped. (The journal still answers alone where there is no row to + * ask: no store, no run history, or a history write that never landed.) + * * And it cannot produce two traversals, by construction: **this verb does * not resume.** It re-arms the pause and stops. The continuation is an * ordinary {@link resume} afterwards, through the same authority gate, the @@ -6031,12 +6039,27 @@ export class AutomationEngine implements IAutomationService { ); } - // The journal: this process's hot copy first, then the durable - // copy on the run's own terminal history row. One reader, two - // sources — the same pairing `resume` itself uses for suspensions. - let consumed = this.consumedSuspensions.get(runId); - if (!consumed && this.store?.loadTerminal) { - let terminal: RunRecord | null; + // The journal: the DURABLE copy on the run's own terminal history + // row FIRST, then this process's hot copy. [#13937] The order is + // the point, and it is the order `loadSuspendedRunStrict` reads + // suspensions in (#13617): the hot journal is a per-process cache, + // the terminal row is the record. Read hot-first, the replica that + // stranded a run keeps a copy that outlives the run — another + // replica restores, resumes and finishes it, and a repeated restore + // here would re-arm a COMPLETED run, whose next resume re-runs + // every node after the pause: shape 2's silent double-run, through + // this verb's side door (pinned in `stranded-run-status.test.ts`, + // measured red on the hot-first tree). So a terminal row that + // exists and carries NO snapshot is the last word — a later + // terminal record wrote it (`recordLog` writes explicit NULLs) — + // and the hot copy is DROPPED rather than honoured. The hot copy + // answers only when there is no row to ask: no store, a store + // without run history, or a history write that never landed + // (reported at `error` where it failed). + let consumed: ConsumedSuspension | undefined; + let terminal: RunRecord | null = null; + let durableSaysNoSnapshot = false; + if (this.store?.loadTerminal) { try { terminal = await this.store.loadTerminal(runId); } catch (err) { @@ -6060,13 +6083,32 @@ export class AutomationEngine implements IAutomationService { consumedAt: terminal.finishedAt ?? terminal.startedAt, error: terminal.error ?? '', }; + } else if (terminal) { + durableSaysNoSnapshot = true; } } + if (durableSaysNoSnapshot) { + // Stale by definition — the record moved on without it. + this.consumedSuspensions.delete(runId); + } else if (!consumed) { + consumed = this.consumedSuspensions.get(runId); + } if (!consumed) { // Nothing to restore — say WHICH nothing. The remedy differs for // every one of these and a single "bad run" refusal would send an // operator looking for the wrong thing. + // + // The durable row, when there is one, is the later word on how + // the run ended than this process's own log — which may still + // say `failed` for a run another replica finished. + if (terminal?.status === 'completed') { + return this.refuseRestore( + runId, + 'RUN_COMPLETED', + `Run '${runId}' completed — there is no unresumable state to exit`, + ); + } const logged = await this.getRun(runId); if (!logged) { return this.refuseRestore(runId, 'RUN_NOT_FOUND', `No run '${runId}' is known`); From 749c496900bc2f0c09fd6d1dd082ee6b1aea632e Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 07:43:28 +0000 Subject: [PATCH 3/7] =?UTF-8?q?test(automation):=20pin=20the=20operator=20?= =?UTF-8?q?exit=20on=20ObjectStoreSuspendedRunStore=20=E2=80=94=20pause-no?= =?UTF-8?q?de=20provenance,=20over-budget=20drop,=20two-witness=20read?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Eight pins over the production store class and the repo's fake ObjectQL engine; all eight are RED at this commit by design (the durable-first read of the previous commit re-arms at the node that threw, and a dropped snapshot reads as the run having moved on). The changeset's "no difference" claim is replaced with what is now known. The fix follows in the next commit. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .changeset/stranded-run-status-stamp.md | 36 +- .../src/stranded-run-object-store.test.ts | 446 ++++++++++++++++++ 2 files changed, 473 insertions(+), 9 deletions(-) create mode 100644 packages/services/service-automation/src/stranded-run-object-store.test.ts diff --git a/.changeset/stranded-run-status-stamp.md b/.changeset/stranded-run-status-stamp.md index 33dd1377f9..47359fef32 100644 --- a/.changeset/stranded-run-status-stamp.md +++ b/.changeset/stranded-run-status-stamp.md @@ -29,12 +29,30 @@ durable discriminator for the condition remains the snapshot the terminal row carries. No resume semantics move for any pausing node type; shapes 2 and 3 of the decision stay excluded. -Also in this change, under the same ruling's exactly-once guarantee: -`restoreConsumedSuspension` now reads the run's durable terminal row BEFORE its -own per-process journal. Read the other way round, the replica that stranded a -run kept a hot copy after another replica restored, resumed and finished it, -and a repeated restore on the first replica re-armed the COMPLETED run — whose -next resume re-ran every node after the pause. Such a restore is now refused -(`RUN_COMPLETED`, or `NO_CONSUMED_SUSPENSION` when the durable row holds no -snapshot for any other reason). Single-process and store-less deployments -observe no difference. +Also in this change, under the same ruling's exactly-once guarantee, two +repairs to how `restoreConsumedSuspension` finds a stranded run's snapshot: + +- The durable run-history row of a stranded run now records the PAUSE node in + `node_id`. It recorded the node that threw — the run's last step — and the + object store read that column back as the snapshot's node, so a restore + from the row (after a restart, or on another replica) re-armed the run at + the failed node and the next resume skipped it while reporting the run + completed. The throwing node stays where the Runs surface reads it: the + row's step log and `error`. +- The verb reads the durable row and its own per-process journal as two + witnesses of one strand instead of trusting either alone. The hot copy is + preferred when both describe the same pause (it is the verbatim object the + failure was journalled from). A row that carries no snapshot is read as + "the run moved on" only when this process's own history write landed — + the replica that stranded a run used to keep a hot copy that could re-arm + the run after another replica had restored, resumed and finished it, and + the next resume re-ran every node after the pause. A snapshot the object + store could not persist (over its 256 KiB row budget) is now recorded in + the row as dropped, with the pause it belonged to, so the replica holding + the hot copy still restores and any other replica is refused with a reason + that names the budget and the remedy. + +In-memory and store-less deployments observe no behaviour difference. On the +object store, same-replica restores re-arm the pause node on every path, and +restores from the row alone do too; restores across replicas of a run that +finished elsewhere are refused. diff --git a/packages/services/service-automation/src/stranded-run-object-store.test.ts b/packages/services/service-automation/src/stranded-run-object-store.test.ts new file mode 100644 index 0000000000..5e35bc7215 --- /dev/null +++ b/packages/services/service-automation/src/stranded-run-object-store.test.ts @@ -0,0 +1,446 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #13937 — the operator exit on the PRODUCTION store class, + * `ObjectStoreSuspendedRunStore`, over the repo's own fake ObjectQL engine. + * + * `stranded-run-status.test.ts` pins the ruling on `InMemorySuspendedRunStore`, + * whose `recordTerminal` is synchronous, uncapped, and keeps the nested + * snapshot verbatim. The object store differs in exactly the ways that decide + * whether the verb re-arms the RIGHT pause: it flattens the snapshot into + * `sys_automation_run` columns, drops it over a byte budget, and is written + * fire-and-forget. A suite green on the in-memory store said nothing about + * any of that — the contract review of the first cut measured the defect one + * store class over, and this file is that measurement kept. + * + * What is pinned, and why each one is here: + * + * 1. ⭐ **The re-armed snapshot carries the PAUSE node — same replica.** The + * terminal row's `node_id` used to be the LAST STEP, i.e. the node that + * threw; read back as the snapshot's node, a restore re-armed the run at + * the failed node and the next resume SKIPPED it and reported the run + * completed. Measured red before the provenance fix. + * 2. ⭐ **…and across a restart / on another replica**, where the durable + * row is the only copy. Same provenance, other reader. + * 3. ⭐ **An over-budget snapshot does not destroy recoverability.** The + * store cannot persist a 300 KiB snapshot; it says so IN THE ROW, and the + * replica that holds the hot copy still restores. A replica without the + * hot copy is refused with a reason that names the budget. Nothing is + * deleted on that reading. + * 4. ⭐ **A stale hot copy still cannot re-arm a run that finished + * elsewhere** — the cross-replica pin from the in-memory file, on the + * object store. + * 5. ⭐ **Two witnesses, one truth.** When the hot copy and the durable row + * describe different pauses, the newest strand wins: a hot copy whose own + * terminal write never landed beats an older row (the #13617 exception — + * a row the store was never handed says nothing), and a landed hot copy + * yields to a later strand another replica recorded. + * 6. **The verdict itself** (`status: 'stranded'`) on this store, and the + * row's discriminator column semantics. + */ + +import { describe, it, expect } from 'vitest'; +import { assertEngineDeleteDispatch } from '@objectstack/metadata-core'; +import type { AutomationContext } from '@objectstack/spec/contracts'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; + +import { AutomationEngine } from './engine.js'; +import type { SuspendedRunStore } from './engine.js'; +import { ObjectStoreSuspendedRunStore } from './suspended-run-store.js'; +import type { SuspendedRunStoreEngine } from './suspended-run-store.js'; + +function createTestLogger() { + return { info: () => {}, warn: () => {}, error: () => {}, debug: () => {}, child: () => createTestLogger() } as any; +} + +/** + * Minimal in-memory ObjectQL-like engine: rows keyed by id, with `where` + * equality filtering — the same double `suspended-run-store.test.ts` drives + * {@link ObjectStoreSuspendedRunStore} with, copied rather than shared so this + * file registers no tests it does not own. `delete` routes through the real + * engine's dispatch predicate (`check:engine-double-contract`). + */ +function createFakeEngine(): SuspendedRunStoreEngine & { rows: Map } { + const rows = new Map(); + const matches = (row: any, where: any) => + !where || Object.entries(where).every(([k, v]) => { + if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`); + return v && typeof v === 'object' && '$lt' in (v as any) + ? row[k] < (v as any).$lt + : row[k] === v; + }); + return { + rows, + async find(_object, options) { + const where = options?.where; + const out = [...rows.values()].filter(r => matches(r, where)); + return typeof options?.limit === 'number' ? out.slice(0, options.limit) : out; + }, + async insert(_object, data) { + rows.set(String(data.id), { ...data }); + return data; + }, + async update(_object, data, options) { + const id = options?.where?.id ?? data.id; + const existing = rows.get(String(id)) ?? { id }; + rows.set(String(id), { ...existing, ...data }); + return rows.get(String(id)); + }, + async delete(_object, options) { + const dispatch = assertEngineDeleteDispatch(options as any); + if (dispatch.kind === 'multi') { + const doomed = [...rows.values()].filter(r => matches(r, options?.where)); + for (const r of doomed) rows.delete(String(r.id)); + return doomed.length; + } + rows.delete(String(dispatch.id)); + return true; + }, + }; +} + +const holdDescriptor = defineActionDescriptor({ + type: 'hold', version: '1.0.0', name: 'hold', + supportsPause: true, resumeAuthority: 'any', +}); +const tailDescriptor = defineActionDescriptor({ type: 'tail', version: '1.0.0', name: 'tail' }); + +/** start → hold (pauses) → tail (throws while `knobs.throwAt` names it) → end. */ +const STRAND_FLOW = { + name: 'strand_flow', label: 'Strand', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'hold', type: 'hold', label: 'Hold' }, + { id: 'tail', type: 'tail', label: 'Tail' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'hold' }, + { id: 'e2', source: 'hold', target: 'tail' }, + { id: 'e3', source: 'tail', target: 'end' }, + ], +}; + +/** Two pauses, so a run can strand at two DIFFERENT pauses in its life. */ +const TWO_PAUSE_FLOW = { + name: 'two_pause_flow', label: 'Two pauses', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'hold', type: 'hold', label: 'Hold' }, + { id: 'tail', type: 'tail', label: 'Tail' }, + { id: 'hold2', type: 'hold', label: 'Hold 2' }, + { id: 'tail2', type: 'tail', label: 'Tail 2' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'hold' }, + { id: 'e2', source: 'hold', target: 'tail' }, + { id: 'e3', source: 'tail', target: 'hold2' }, + { id: 'e4', source: 'hold2', target: 'tail2' }, + { id: 'e5', source: 'tail2', target: 'end' }, + ], +}; + +/** The node ids that RAN, in order — what a wrong re-arm or a double-run moves. */ +interface Ledger { ran: string[] } +/** Which tail nodes throw on their next run. Shared across replicas. */ +interface Knobs { throwAt: Set } + +/** One engine over `store`, writing to the SHARED ledger, reading the SHARED knobs. */ +function replica(store: SuspendedRunStore | undefined, led: Ledger, knobs: Knobs, logger = createTestLogger()): AutomationEngine { + const engine = new AutomationEngine(logger, store); + engine.registerNodeExecutor({ + type: 'hold', + descriptor: holdDescriptor, + async execute(node: { id: string }) { + return { success: true, suspend: true, correlation: `approval:${node.id}` }; + }, + } as never); + engine.registerNodeExecutor({ + type: 'tail', + descriptor: tailDescriptor, + async execute(node: { id: string }) { + led.ran.push(node.id); + if (knobs.throwAt.has(node.id)) throw new Error(`${node.id} blew up`); + return { success: true }; + }, + } as never); + engine.registerFlow('strand_flow', STRAND_FLOW as never); + engine.registerFlow('two_pause_flow', TWO_PAUSE_FLOW as never); + return engine; +} + +const ctx = { event: 'test', record: { id: 'rec_1' } } as unknown as AutomationContext; + +function objectStore(rows?: SuspendedRunStoreEngine & { rows: Map }) { + const engine = rows ?? createFakeEngine(); + return { engine, store: new ObjectStoreSuspendedRunStore(engine, createTestLogger()) }; +} + +function harness(store: SuspendedRunStore) { + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + return { led, knobs, engine: replica(store, led, knobs) }; +} + +/** Park a run at `hold`, then resume it into the `tail` throw. */ +async function strand(engine: AutomationEngine, flow = 'strand_flow', context: AutomationContext = ctx) { + const started = await engine.execute(flow, context); + expect(started.status).toBe('paused'); + const runId = started.runId as string; + const failed = await engine.resume(runId); + expect(failed.success).toBe(false); + expect(failed.status).toBe('stranded'); + return { runId, failed }; +} + +/** + * The terminal write is fire-and-forget (`void store.recordTerminal(...)`), so + * a cross-replica reader has to wait for the row the way a real one would: + * by observing it land. + */ +async function untilTerminalRow(store: SuspendedRunStore, runId: string) { + for (let i = 0; i < 200; i++) { + const row = await store.loadTerminal!(runId); + if (row) return row; + await new Promise(r => setTimeout(r, 1)); + } + throw new Error(`terminal row for ${runId} never landed`); +} + +/** A store whose history write can be made to fail, everything else delegated. */ +function withFailingHistory(inner: ObjectStoreSuspendedRunStore, state: { fail: boolean }): SuspendedRunStore { + return { + save: (r) => inner.save(r), + load: (id) => inner.load(id), + delete: (id) => inner.delete(id), + list: () => inner.list(), + claimSuspension: (id, at) => inner.claimSuspension(id, at), + recordTerminal: async (r) => { + if (state.fail) throw new Error('history table unreachable'); + return inner.recordTerminal(r); + }, + listHistory: (f, n) => inner.listHistory(f, n), + loadTerminal: (id) => inner.loadTerminal(id), + }; +} + +describe('#13937 — the re-armed snapshot carries the PAUSE node (ObjectStoreSuspendedRunStore)', () => { + it('⭐ same replica: strand → restore → resume re-runs the node that threw, at the pause it left', async () => { + const { store } = objectStore(); + const { engine, knobs, led } = harness(store); + const { runId } = await strand(engine); + expect(led.ran).toEqual(['tail']); + await untilTerminalRow(store, runId); + + const restored = await engine.restoreConsumedSuspension(runId, { requestedBy: 'ops' }); + expect(restored.restored).toBe(true); + // The re-armed pause is the PAUSE node — never the node that threw. + expect(restored.nodeId).toBe('hold'); + expect((await store.load(runId))?.nodeId).toBe('hold'); + expect((await store.load(runId))?.correlation).toBe('approval:hold'); + + knobs.throwAt.clear(); + const done = await engine.resume(runId); + expect(done.success).toBe(true); + // THE OBSERVABLE: the failed node RAN AGAIN. A re-arm at `tail` would + // have skipped it and still reported the run completed. + expect(led.ran).toEqual(['tail', 'tail']); + expect((await engine.getRun(runId))?.status).toBe('completed'); + }); + + it('⭐ the durable row records the pause node on a stranded run, so a restore from the row alone re-arms the pause', async () => { + const { engine: fake, store } = objectStore(); + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + const a = replica(store, led, knobs); + const { runId } = await strand(a); + const row = await untilTerminalRow(store, runId); + + // The row itself: its snapshot names the pause, its step log still + // ends with the node that threw (the Runs surface reads that half). + expect(row.status).toBe('failed'); + expect(row.consumedSuspension?.nodeId).toBe('hold'); + expect(row.consumedSuspension?.correlation).toBe('approval:hold'); + expect(row.steps?.[row.steps.length - 1]?.nodeId).toBe('tail'); + expect(fake.rows.get(`run_${runId}`)?.node_id).toBe('hold'); + + // A fresh replica — no hot copy at all — restores from the row. + const b = replica(store, led, knobs); + const restored = await b.restoreConsumedSuspension(runId); + expect(restored.restored).toBe(true); + expect(restored.nodeId).toBe('hold'); + knobs.throwAt.clear(); + expect((await b.resume(runId)).success).toBe(true); + expect(led.ran).toEqual(['tail', 'tail']); + }); +}); + +describe('#13937 — an over-budget snapshot does not destroy recoverability', () => { + // A context far over the 256 KiB row budget the store applies to the + // snapshot's JSON columns. + const hugeCtx = { ...ctx, payload: 'x'.repeat(300 * 1024) } as unknown as AutomationContext; + + it('⭐ the store says IN THE ROW that a snapshot existed and was not persisted, and the replica holding the hot copy still restores', async () => { + const { engine: fake, store } = objectStore(); + const { engine, knobs, led } = harness(store); + const { runId, failed } = await strand(engine, 'strand_flow', hugeCtx); + expect(failed.status).toBe('stranded'); + const row = await untilTerminalRow(store, runId); + + // No snapshot — and NOT nothing: the row records the drop, with the + // pause it belonged to, so no reader mistakes it for a run that moved + // on or never paused. + expect(row.consumedSuspension).toBeUndefined(); + expect(row.consumedSuspensionDropped).toBeDefined(); + expect(row.consumedSuspensionDropped?.bytes).toBeGreaterThan(row.consumedSuspensionDropped?.budget ?? Infinity); + expect(row.consumedSuspensionDropped?.nodeId).toBe('hold'); + expect(row.consumedSuspensionDropped?.correlation).toBe('approval:hold'); + expect(fake.rows.get(`run_${runId}`)?.context_json).toBeNull(); + + // The hot copy is the only copy — and it is honoured, not deleted. + const restored = await engine.restoreConsumedSuspension(runId); + expect(restored.restored).toBe(true); + expect(restored.nodeId).toBe('hold'); + expect((await engine.restoreConsumedSuspension(runId)).refusal).toBe('RUN_SUSPENDED'); + knobs.throwAt.clear(); + expect((await engine.resume(runId)).success).toBe(true); + expect(led.ran).toEqual(['tail', 'tail']); + }); + + it('a replica WITHOUT the hot copy is refused with a reason that names the budget, and deletes nothing', async () => { + const { store } = objectStore(); + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + const a = replica(store, led, knobs); + const { runId } = await strand(a, 'strand_flow', hugeCtx); + await untilTerminalRow(store, runId); + + const b = replica(store, led, knobs); + const refused = await b.restoreConsumedSuspension(runId); + expect(refused.restored).toBe(false); + expect(refused.refusal).toBe('NO_CONSUMED_SUSPENSION'); + expect(refused.reason).toMatch(/budget/); + expect(refused.reason).toMatch(/process that stranded it/); + + // …and A's hot copy is untouched by B's reading: A still restores. + expect((await a.restoreConsumedSuspension(runId)).restored).toBe(true); + expect(await b.hasSuspendedRun(runId)).toBe(true); + }); +}); + +describe('#13937 — a re-armed run is not double-runnable (ObjectStoreSuspendedRunStore)', () => { + it('⭐ a stale hot copy cannot re-arm a run another replica restored and finished', async () => { + const { store } = objectStore(); + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + const a = replica(store, led, knobs); + const b = replica(store, led, knobs); + + const { runId } = await strand(a); + await untilTerminalRow(store, runId); + + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throwAt.clear(); + expect((await b.resume(runId)).success).toBe(true); + expect(led.ran).toEqual(['tail', 'tail']); + // The completing upsert is fire-and-forget too: wait for THAT row. + for (let i = 0; i < 200 && (await store.loadTerminal!(runId))?.status !== 'completed'; i++) { + await new Promise(r => setTimeout(r, 1)); + } + expect((await store.loadTerminal!(runId))?.status).toBe('completed'); + + const stale = await a.restoreConsumedSuspension(runId, { requestedBy: 'ops-retry' }); + expect(stale.restored).toBe(false); + expect(stale.refusal).toBe('RUN_COMPLETED'); + expect(await store.list()).toHaveLength(0); + expect((await a.resume(runId)).code).toBe('RUN_NOT_FOUND'); + expect(led.ran).toEqual(['tail', 'tail']); + }); + + it('⭐ two witnesses, different pauses: a hot copy whose own write never landed beats the older row', async () => { + const { store: inner } = objectStore(); + const history = { fail: false }; + const store = withFailingHistory(inner, history); + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + const a = replica(store, led, knobs); + + // Strand at `hold` (write lands), restore, run on to `hold2`. + const { runId } = await strand(a, 'two_pause_flow'); + await untilTerminalRow(store, runId); + expect((await a.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throwAt = new Set(['tail2']); + const parkedAgain = await a.resume(runId); + expect(parkedAgain.status).toBe('paused'); + expect((await store.load(runId))?.nodeId).toBe('hold2'); + + // Strand at `hold2` with the history write FAILING: the row still + // describes the `hold` strand; the hot copy is the newest witness. + history.fail = true; + const again = await a.resume(runId); + expect(again.status).toBe('stranded'); + expect((await store.loadTerminal!(runId))?.consumedSuspension?.nodeId).toBe('hold'); + history.fail = false; + + const restored = await a.restoreConsumedSuspension(runId); + expect(restored.restored).toBe(true); + // Re-armed at the pause the run actually left — never back at `hold`, + // which would re-run `tail` a second time on the way to `hold2`. + expect(restored.nodeId).toBe('hold2'); + knobs.throwAt.clear(); + expect((await a.resume(runId)).success).toBe(true); + expect(led.ran).toEqual(['tail', 'tail2', 'tail2']); + }); + + it('⭐ two witnesses, different pauses: a landed hot copy yields to the later strand another replica recorded', async () => { + const { store } = objectStore(); + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + const a = replica(store, led, knobs); + const b = replica(store, led, knobs); + + // A strands at `hold` and keeps its hot copy of that pause. + const { runId } = await strand(a, 'two_pause_flow'); + await untilTerminalRow(store, runId); + + // B restores, runs on to `hold2`, and strands THERE. + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throwAt = new Set(['tail2']); + expect((await b.resume(runId)).status).toBe('paused'); + expect((await b.resume(runId)).status).toBe('stranded'); + expect(led.ran).toEqual(['tail', 'tail2']); + for (let i = 0; i < 200 && (await store.loadTerminal!(runId))?.consumedSuspension?.nodeId !== 'hold2'; i++) { + await new Promise(r => setTimeout(r, 1)); + } + expect((await store.loadTerminal!(runId))?.consumedSuspension?.nodeId).toBe('hold2'); + + // A's copy is of a pause the run has LEFT. Re-arming it would send the + // run back through `tail` a second time. + const restored = await a.restoreConsumedSuspension(runId); + expect(restored.restored).toBe(true); + expect(restored.nodeId).toBe('hold2'); + knobs.throwAt.clear(); + expect((await a.resume(runId)).success).toBe(true); + expect(led.ran).toEqual(['tail', 'tail2', 'tail2']); + }); + + it('two concurrent resumes of a restored run on two replicas run the tail once more', async () => { + const { store } = objectStore(); + const led: Ledger = { ran: [] }; + const knobs: Knobs = { throwAt: new Set(['tail']) }; + const a = replica(store, led, knobs); + const b = replica(store, led, knobs); + + const { runId } = await strand(a); + await untilTerminalRow(store, runId); + expect((await b.restoreConsumedSuspension(runId)).restored).toBe(true); + knobs.throwAt.clear(); + + const [byA, byB] = await Promise.all([a.resume(runId), b.resume(runId)]); + expect([byA, byB].filter(r => r.success)).toHaveLength(1); + expect([byA, byB].filter(r => r.code === 'RESUME_IN_PROGRESS')).toHaveLength(1); + expect(led.ran).toEqual(['tail', 'tail']); + expect(await store.list()).toHaveLength(0); + }); +}); From 1ad59b971976aa5643b52e54fe21c7fb490edbcf Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 07:43:28 +0000 Subject: [PATCH 4/7] fix(automation): restoreConsumedSuspension reads its journal and the durable row as two witnesses; the stranded row records the pause node and any dropped snapshot - The stranded run's terminal row carries the PAUSE node in node_id (it carried the last step, the node that threw, which the object store read back as the snapshot's node: a restore from the row re-armed the failed node and the next resume skipped it). - The object store records an over-budget snapshot drop IN the row, with the pause it belonged to (RunRecord.consumedSuspensionDropped), instead of a bare NULL that read as the run having moved on. - The engine prefers its hot copy when it and the row describe the same pause; between different pauses the newest strand wins, judged by whether this process's own history write landed (ConsumedSuspension.persisted); a snapshot-less, notice-less row discards the hot copy only when that write did land. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .../services/service-automation/src/engine.ts | 220 ++++++++++++++---- .../services/service-automation/src/index.ts | 5 + .../src/suspended-run-store.ts | 74 +++++- 3 files changed, 251 insertions(+), 48 deletions(-) diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 0f5915c44f..923b60f8be 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1189,6 +1189,16 @@ export interface RunRecord { durationMs?: number; /** Failure reason for a `failed` run — what a designer needs to fix it. */ error?: string; + /** + * The node this record is ABOUT. On an ordinary terminal record: the run's + * last step. On a stranded run's record — `consumedSuspension` present, or + * `consumedSuspensionDropped` — the PAUSE node, the node the consumed + * snapshot re-arms, because the object store rebuilds the snapshot's node + * from this one column and there is no other (#13937: written as the last + * step, a restore from the row re-armed the run at the node that threw and + * the next resume skipped it). The node that threw is still where the Runs + * surface reads it: the last entry of `steps`, and `error`. + */ nodeId?: string; /** * [#10101] The ACTING context's tenant (`AutomationContext.tenantId`), @@ -1269,6 +1279,34 @@ export interface RunRecord { * ⛔ Not a run state. See {@link ConsumedSuspension}. */ consumedSuspension?: SuspendedRun; + /** + * [#13937] The store could not persist the snapshot this run had — the + * object store DROPS one over its row byte budget rather than truncating + * it — and says so, with the pause it belonged to. Present exactly when a + * snapshot existed and `consumedSuspension` is absent for THAT reason, so + * a reader can tell "dropped" from "the run moved on": a terminal record + * with neither field is a run that reached a terminal state which was not + * a strand (completed, cancelled, cascade-failed), or never paused at all. + * {@link AutomationEngine.restoreConsumedSuspension} honours a hot copy of + * the same pause on this reading, and without one refuses naming the + * budget. A store that cannot persist a snapshot MUST record it this way: + * writing nothing reads as the run having moved on. + */ + consumedSuspensionDropped?: ConsumedSuspensionDropNotice; +} + +/** + * [#13937] What a store records in place of a consumed-suspension snapshot it + * could not persist — see {@link RunRecord.consumedSuspensionDropped}. + */ +export interface ConsumedSuspensionDropNotice { + /** Size of the snapshot's JSON columns together, in bytes. */ + bytes: number; + /** The store's budget the snapshot was over, in bytes. */ + budget: number; + /** The pause the dropped snapshot belonged to — its identity, not its state. */ + nodeId?: string; + correlation?: string; } /** @@ -1304,6 +1342,18 @@ export interface ConsumedSuspension { consumedAt: string; /** The downstream failure that left the run terminal and unresumable. */ error: string; + /** + * [#13937] Whether THIS process's own history write for the strand reached + * the store: `'pending'` while the fire-and-forget write is in flight, + * `'landed'` once the store accepted it (the row now describes this strand + * — as a snapshot, or as a drop notice), `'failed'` when it threw, absent + * when no store writes history at all. The engine's own memory of what the + * store was handed — the #13617 exception, for this journal: a row the + * store never received says nothing about this copy, so + * {@link AutomationEngine.restoreConsumedSuspension} reads a snapshot-less + * row as "the run moved on" only when this is `'landed'`. + */ + persisted?: 'pending' | 'landed' | 'failed'; } /** @@ -5858,6 +5908,8 @@ export class AutomationEngine implements IAutomationService { run: { ...run, steps: run.steps.slice(0, stepCountAtPause) }, consumedAt: new Date().toISOString(), error, + // [#13937] `recordLog` settles this when its write settles. + ...(this.store?.recordTerminal ? { persisted: 'pending' as const } : {}), }; this.consumedSuspensions.set(run.runId, consumed); // Oldest first — `Map` iterates in insertion order, and re-`set`ting an @@ -5872,6 +5924,28 @@ export class AutomationEngine implements IAutomationService { return consumed; } + /** + * [#13937] Does the durable row describe a LATER strand than this + * process's hot copy? True only when the two name different pauses AND the + * hot copy's own history write landed — then the row can only be a later + * strand another replica recorded. Same pause: never (the hot copy is the + * more faithful witness of it). Different pause but the hot copy's write + * never reached the store (`persisted` not `'landed'`): never — the row + * predates this strand and says nothing about it (the #13617 exception). + * The pause identity is the pair {@link SuspendedRunStore.claimSuspension} + * compares: the node, and the correlation when one was minted. + */ + private rowSupersedesJournal( + hot: ConsumedSuspension, + rowPause: { nodeId?: string; correlation?: string }, + ): boolean { + const samePause = + hot.run.nodeId === rowPause.nodeId && + (hot.run.correlation ?? undefined) === (rowPause.correlation ?? undefined); + if (samePause) return false; + return hot.persisted === 'landed'; + } + /** Build a refusal from {@link restoreConsumedSuspension}. */ private refuseRestore( runId: string, @@ -5946,13 +6020,14 @@ export class AutomationEngine implements IAutomationService { * `restored: true` and one `RESTORE_IN_PROGRESS`, rather than two calls * both claiming the restore. * - * [#13937] The third half, across replicas and across TIME: the snapshot - * is read from the durable terminal row before this process's own - * journal, so a hot copy left behind on the replica that stranded the run - * cannot re-arm it after another replica restored and finished it — the - * row that no longer carries a snapshot is the record, and the copy is - * dropped. (The journal still answers alone where there is no row to - * ask: no store, no run history, or a history write that never landed.) + * [#13937] The third half, across replicas and across TIME: this + * process's journal and the durable terminal row are read as two + * witnesses of one strand (see the read itself, below). A hot copy left + * behind on the replica that stranded the run cannot re-arm it after + * another replica restored and finished it, or restranded it at a later + * pause; a snapshot the store could not persist is still restorable from + * the copy the stranding process holds; and a row the store was never + * handed says nothing about that copy. * * And it cannot produce two traversals, by construction: **this verb does * not resume.** It re-arms the pause and stops. The continuation is an @@ -6039,26 +6114,42 @@ export class AutomationEngine implements IAutomationService { ); } - // The journal: the DURABLE copy on the run's own terminal history - // row FIRST, then this process's hot copy. [#13937] The order is - // the point, and it is the order `loadSuspendedRunStrict` reads - // suspensions in (#13617): the hot journal is a per-process cache, - // the terminal row is the record. Read hot-first, the replica that - // stranded a run keeps a copy that outlives the run — another - // replica restores, resumes and finishes it, and a repeated restore - // here would re-arm a COMPLETED run, whose next resume re-runs - // every node after the pause: shape 2's silent double-run, through - // this verb's side door (pinned in `stranded-run-status.test.ts`, - // measured red on the hot-first tree). So a terminal row that - // exists and carries NO snapshot is the last word — a later - // terminal record wrote it (`recordLog` writes explicit NULLs) — - // and the hot copy is DROPPED rather than honoured. The hot copy - // answers only when there is no row to ask: no store, a store - // without run history, or a history write that never landed - // (reported at `error` where it failed). - let consumed: ConsumedSuspension | undefined; + // [#13937] Two witnesses of one strand, and neither is trusted + // alone — the contract review of this ruling's services half + // measured both single-witness readings wrong, one store class + // apart: + // + // - This process's HOT copy is the verbatim object the failure + // was journalled from: the pause's own node, variables, step + // log as of the pause. It is a per-process cache. The replica + // that stranded a run keeps it after another replica restored, + // resumed and FINISHED the run, and re-arming it then re-runs + // every node after the pause — shape 2's silent double-run, + // through this verb's side door (pinned in + // `stranded-run-status.test.ts`, red on the hot-only tree). + // - The DURABLE row is the record every replica can read, and a + // flattened, column-bounded copy of the same snapshot: the + // object store rebuilds it from columns, DROPS it over a byte + // budget — and says so in the row, `consumedSuspensionDropped` + // — and receives it fire-and-forget. Read alone, a snapshot-less + // row sent a run the store could not persist into + // NO_CONSUMED_SUSPENSION on the very replica holding its copy + // (pinned in `stranded-run-object-store.test.ts`, red on the + // durable-first tree). + // + // So: the hot copy is preferred whenever both describe the SAME + // pause (`rowSupersedesJournal`). When they describe different + // pauses, the newest strand wins — a hot copy whose own write never + // landed (`persisted` is not `'landed'`: the #13617 exception, a + // row the store was never handed says nothing) beats the older + // row, and a landed hot copy yields to the later strand another + // replica recorded. A row with neither a snapshot nor a drop notice + // is "the run moved on" only for a hot copy whose write did land — + // that copy is then DROPPED rather than honoured. A hot copy + // answers alone where there is no row to ask: no store, no run + // history, a write that never landed or is still in flight. + const hot = this.consumedSuspensions.get(runId); let terminal: RunRecord | null = null; - let durableSaysNoSnapshot = false; if (this.store?.loadTerminal) { try { terminal = await this.store.loadTerminal(runId); @@ -6077,27 +6168,54 @@ export class AutomationEngine implements IAutomationService { `Durable run-history unreachable for run '${runId}' — whether a consumed suspension survives is unknown`, ); } - if (terminal?.consumedSuspension) { - consumed = { - run: terminal.consumedSuspension, - consumedAt: terminal.finishedAt ?? terminal.startedAt, - error: terminal.error ?? '', - }; - } else if (terminal) { - durableSaysNoSnapshot = true; - } } - if (durableSaysNoSnapshot) { - // Stale by definition — the record moved on without it. + + let consumed: ConsumedSuspension | undefined; + let dropped: ConsumedSuspensionDropNotice | undefined; + if (!terminal) { + consumed = hot; + } else if (terminal.consumedSuspension) { + const durable: ConsumedSuspension = { + run: terminal.consumedSuspension, + consumedAt: terminal.finishedAt ?? terminal.startedAt, + error: terminal.error ?? '', + }; + consumed = hot && !this.rowSupersedesJournal(hot, durable.run) ? hot : durable; + // A hot copy of a pause the run has since LEFT — re-arming it + // would send the run back through work it already did. + if (hot && consumed !== hot) this.consumedSuspensions.delete(runId); + } else if (terminal.consumedSuspensionDropped) { + dropped = terminal.consumedSuspensionDropped; + if (hot && !this.rowSupersedesJournal(hot, dropped)) { + consumed = hot; + } else if (hot) { + this.consumedSuspensions.delete(runId); + } + } else if (hot && hot.persisted !== 'landed') { + // The row predates this strand — this process's own write for + // it never reached the store (in flight, or failed and + // reported at `error`). The store's silence says nothing. + consumed = hot; + } else if (hot) { + // A later terminal record with no snapshot and no drop notice: + // completed, cancelled or cascade-failed after this copy was + // taken. Stale by definition. this.consumedSuspensions.delete(runId); - } else if (!consumed) { - consumed = this.consumedSuspensions.get(runId); } if (!consumed) { // Nothing to restore — say WHICH nothing. The remedy differs for // every one of these and a single "bad run" refusal would send an // operator looking for the wrong thing. + if (dropped) { + return this.refuseRestore( + runId, + 'NO_CONSUMED_SUSPENSION', + `Run '${runId}' is recorded 'failed' and its consumed suspension (${dropped.bytes} bytes) was over ` + + `the store's ${dropped.budget}-byte row budget, so it was not persisted — it can be restored ` + + `only by the process that stranded it, while that process is still running`, + ); + } // // The durable row, when there is one, is the later word on how // the run ended than this process's own log — which may still @@ -6482,7 +6600,13 @@ export class AutomationEngine implements IAutomationService { triggerType: entry.trigger?.type || undefined, triggerObject: entry.trigger?.object, triggerRecordId: entry.trigger?.recordId, - nodeId: lastStep?.nodeId, + // [#13937] On a stranded run's record the PAUSE node, not the + // node that threw: the object store rebuilds the snapshot's + // node from this one column, and a restore from the row used + // to re-arm the run at the failed node — whose next resume + // then SKIPPED it and reported the run completed. The throwing + // node is the last entry of `steps` below, and `error`. + nodeId: consumedSuspension?.nodeId ?? lastStep?.nodeId, steps: this.compactStepsForHistory(entry.steps), summary: entry.summary, // [#13909] Present only on the resume-consumed-then-failed @@ -6492,7 +6616,21 @@ export class AutomationEngine implements IAutomationService { // leaving a stale one an operator could restore a second time. consumedSuspension, }; - void this.store.recordTerminal(record).catch((err) => { + // [#13937] The journal entry this record describes, when it is a + // strand's: `journalConsumedSuspension` ran just above in the + // same arm, so the entry under this run id is the one built from + // `consumedSuspension` — checked by identity, because a later + // strand of the same run replaces the entry and must not inherit + // this write's outcome. + const journal = consumedSuspension ? this.consumedSuspensions.get(entry.id) : undefined; + const write = this.store.recordTerminal(record); + if (journal && journal.run === consumedSuspension) { + void write.then( + () => { journal.persisted = 'landed'; }, + () => { journal.persisted = 'failed'; }, + ); + } + void write.catch((err) => { // #6499 — driver text to the structured slot; see // `forgetSuspendedRun`'s catch above for the full mechanism // (#6299). diff --git a/packages/services/service-automation/src/index.ts b/packages/services/service-automation/src/index.ts index e9efd24379..e03ee19697 100644 --- a/packages/services/service-automation/src/index.ts +++ b/packages/services/service-automation/src/index.ts @@ -51,6 +51,11 @@ export type { // no barrel-reachable signature. SuspensionRestoreResult, SuspensionRestoreRefusal, + // [#13937] What a store records in place of a snapshot it could not + // persist — the type of `RunRecord.consumedSuspensionDropped`, which a + // host store implementing `recordTerminal` / `loadTerminal` writes and + // reads; unnameable, the field would be writable only by structural luck. + ConsumedSuspensionDropNotice, } from './engine.js'; // [#11997] ADR-0005 overlay precedence for same-named flow definitions. The boot diff --git a/packages/services/service-automation/src/suspended-run-store.ts b/packages/services/service-automation/src/suspended-run-store.ts index 0ad54b2841..78f65ceddb 100644 --- a/packages/services/service-automation/src/suspended-run-store.ts +++ b/packages/services/service-automation/src/suspended-run-store.ts @@ -10,6 +10,7 @@ import type { Logger } from '@objectstack/spec/contracts'; // be a third answer to a question the codebase already answered two ways. import { createRecordOrganizationResolver, type RecordOrganizationResolver } from '@objectstack/metadata-core'; import type { + ConsumedSuspensionDropNotice, RunRecord, SuspendedRun, SuspendedRunStore, @@ -71,12 +72,26 @@ const MAX_STEPS_JSON_BYTES = 64 * 1024; * ⛔ Over budget the snapshot is DROPPED, never truncated — and that asymmetry * with `steps_json` above is the whole point. A halved step tail is still an * honest, smaller observation; half a variable map is a run that would resume - * from state it was never in. `restoreConsumedSuspension` then answers - * `NO_CONSUMED_SUSPENSION` for the run, which is true, instead of restoring a - * corrupt pause that looks perfectly healthy. + * from state it was never in. The drop is RECORDED in the row (#13937, + * {@link CONSUMED_SUSPENSION_DROPPED_KEY}): `restoreConsumedSuspension` then + * restores from the hot copy in the process that stranded the run, and from + * any other replica answers `NO_CONSUMED_SUSPENSION` naming this budget — + * both true — instead of restoring a corrupt pause that looks perfectly + * healthy, or reading the missing snapshot as a run that moved on. */ const MAX_CONSUMED_SUSPENSION_JSON_BYTES = 256 * 1024; +/** + * [#13937] The ONE key a terminal row's `variables_json` holds when the + * consumed-suspension snapshot was dropped over + * {@link MAX_CONSUMED_SUSPENSION_JSON_BYTES}. It rides the discriminator + * column on purpose — "this failed run had a pause and no longer has one" + * stays true of the row — and it can never collide with a real variable map: + * `$`-prefixed names are the engine's own, and a map holding exactly this one + * key is not a state any flow was ever in. + */ +const CONSUMED_SUSPENSION_DROPPED_KEY = '$consumedSuspensionDropped'; + /** Byte cap for a terminal row's persisted `summary_json` (#4354). Generous * relative to the shape it holds — one entry per node that ran, one per gate * that closed — so only a pathological flow ever trips it. */ @@ -580,6 +595,10 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { // `restoreConsumedSuspension` reports honestly rather than as a run that // never suspended. consumedSuspension: deserializeConsumedSuspension(row), + // [#13937] …and when the column holds the drop notice instead of the + // snapshot, the pause existed and was too large to keep: said as such, + // so the engine does not read this row as the run having moved on. + consumedSuspensionDropped: readConsumedSuspensionDropNotice(row), }; } @@ -707,11 +726,28 @@ function serializeConsumedSuspension( if (bytes > MAX_CONSUMED_SUSPENSION_JSON_BYTES) { logger?.warn?.( `[automation] run '${run.runId}': the suspension its resume consumed is ${bytes} bytes, over the ` + - `${MAX_CONSUMED_SUSPENSION_JSON_BYTES}-byte row budget, so it was NOT persisted and this run cannot be ` + - `restored after a restart. It was dropped rather than truncated on purpose — half a variable map would ` + - `restore the run into a state it was never in.`, + `${MAX_CONSUMED_SUSPENSION_JSON_BYTES}-byte row budget, so it was NOT persisted — this run can be ` + + `restored only by the process that stranded it, while that process is running, never after a restart ` + + `or from another replica. It was dropped rather than truncated on purpose — half a variable map would ` + + `restore the run into a state it was never in. The row records the drop.`, ); - return empty; + // [#13937] Say so IN THE ROW, with the pause the snapshot belonged to: + // a bare NULL here is indistinguishable from a run that completed or was + // cancelled after a restore, and the engine — rightly — reads that as + // "moved on" and discards its own hot copy. This notice is what keeps + // the hot copy honoured on the replica that has it, and the refusal + // honest everywhere else. + return { + ...empty, + variables_json: JSON.stringify({ + [CONSUMED_SUSPENSION_DROPPED_KEY]: { + bytes, + budget: MAX_CONSUMED_SUSPENSION_JSON_BYTES, + nodeId: run.nodeId, + correlation: run.correlation, + }, + }), + }; } // `correlation` is part of the resumable state, not decoration: a run parked // at a `subflow:`/`map:` node resumes down a DIFFERENT path on it, and a @@ -739,6 +775,9 @@ function serializeConsumedSuspension( */ function deserializeConsumedSuspension(row: any): SuspendedRun | undefined { if (row.variables_json == null || row.variables_json === '') return undefined; + // [#13937] The column holds the drop notice, not a snapshot: nothing to + // rebuild — `readConsumedSuspensionDropNotice` reports it separately. + if (readConsumedSuspensionDropNotice(row)) return undefined; const startedAt = row.started_at ?? row.created_at ?? ''; const rawId = String(row.id ?? ''); return { @@ -757,6 +796,27 @@ function deserializeConsumedSuspension(row: any): SuspendedRun | undefined { }; } +/** + * [#13937] The drop notice a terminal row's `variables_json` carries when the + * snapshot was over budget, or `undefined` when the column holds a snapshot + * (or nothing). Strict about the shape — exactly one key, the reserved one — + * so a real variable map can never be read as a notice. + */ +function readConsumedSuspensionDropNotice(row: any): ConsumedSuspensionDropNotice | undefined { + const parsed = parseJson(row.variables_json, undefined); + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return undefined; + const keys = Object.keys(parsed as object); + if (keys.length !== 1 || keys[0] !== CONSUMED_SUSPENSION_DROPPED_KEY) return undefined; + const notice = (parsed as Record)[CONSUMED_SUSPENSION_DROPPED_KEY] as Record | null; + if (!notice || typeof notice !== 'object') return undefined; + return { + bytes: typeof notice.bytes === 'number' ? notice.bytes : 0, + budget: typeof notice.budget === 'number' ? notice.budget : MAX_CONSUMED_SUSPENSION_JSON_BYTES, + nodeId: typeof notice.nodeId === 'string' ? notice.nodeId : undefined, + correlation: typeof notice.correlation === 'string' ? notice.correlation : undefined, + }; +} + /** * JSON-encode a terminal run's step log under the {@link MAX_STEPS_JSON_BYTES} * cap. The engine already bounds step COUNT (and strips stacks); this bounds From ff0e13f6e7474fa20f4279b25991407998149bb3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 07:46:07 +0000 Subject: [PATCH 5/7] test(automation): count the successful re-run of the failed node in the two-pause ledgers The two-witness pins expected the ledger without the tail re-run that a restore + resume legitimately performs; the engine was right, the count was not. Pause-node assertions were already green. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .../src/stranded-run-object-store.test.ts | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/packages/services/service-automation/src/stranded-run-object-store.test.ts b/packages/services/service-automation/src/stranded-run-object-store.test.ts index 5e35bc7215..8dc1fb0d3f 100644 --- a/packages/services/service-automation/src/stranded-run-object-store.test.ts +++ b/packages/services/service-automation/src/stranded-run-object-store.test.ts @@ -386,11 +386,12 @@ describe('#13937 — a re-armed run is not double-runnable (ObjectStoreSuspended const restored = await a.restoreConsumedSuspension(runId); expect(restored.restored).toBe(true); // Re-armed at the pause the run actually left — never back at `hold`, - // which would re-run `tail` a second time on the way to `hold2`. + // which would run `tail` a THIRD time on the way to `hold2`. (The + // second `tail` is the successful re-run after the first restore.) expect(restored.nodeId).toBe('hold2'); knobs.throwAt.clear(); expect((await a.resume(runId)).success).toBe(true); - expect(led.ran).toEqual(['tail', 'tail2', 'tail2']); + expect(led.ran).toEqual(['tail', 'tail', 'tail2', 'tail2']); }); it('⭐ two witnesses, different pauses: a landed hot copy yields to the later strand another replica recorded', async () => { @@ -409,20 +410,20 @@ describe('#13937 — a re-armed run is not double-runnable (ObjectStoreSuspended knobs.throwAt = new Set(['tail2']); expect((await b.resume(runId)).status).toBe('paused'); expect((await b.resume(runId)).status).toBe('stranded'); - expect(led.ran).toEqual(['tail', 'tail2']); + expect(led.ran).toEqual(['tail', 'tail', 'tail2']); for (let i = 0; i < 200 && (await store.loadTerminal!(runId))?.consumedSuspension?.nodeId !== 'hold2'; i++) { await new Promise(r => setTimeout(r, 1)); } expect((await store.loadTerminal!(runId))?.consumedSuspension?.nodeId).toBe('hold2'); // A's copy is of a pause the run has LEFT. Re-arming it would send the - // run back through `tail` a second time. + // run back through `tail` a third time. const restored = await a.restoreConsumedSuspension(runId); expect(restored.restored).toBe(true); expect(restored.nodeId).toBe('hold2'); knobs.throwAt.clear(); expect((await a.resume(runId)).success).toBe(true); - expect(led.ran).toEqual(['tail', 'tail2', 'tail2']); + expect(led.ran).toEqual(['tail', 'tail', 'tail2', 'tail2']); }); it('two concurrent resumes of a restored run on two replicas run the tail once more', async () => { From c5025d04644a130775fd158154fb8972c016d197 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 08:01:15 +0000 Subject: [PATCH 6/7] test(automation): pin the object-store fake's update to the engine's dispatch predicate; record the new pinned double check:engine-double-contract's own two prescriptions: open the fake's update() with assertEngineUpdateDispatch(data, options), and let the pinned ledger learn the file (--write). The store's own update calls pass the predicate: the object-store file stays 8/8. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .../src/stranded-run-object-store.test.ts | 8 +++++--- scripts/engine-double-contract.pinned.json | 10 ++++++++++ 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/packages/services/service-automation/src/stranded-run-object-store.test.ts b/packages/services/service-automation/src/stranded-run-object-store.test.ts index 8dc1fb0d3f..75ec8d30db 100644 --- a/packages/services/service-automation/src/stranded-run-object-store.test.ts +++ b/packages/services/service-automation/src/stranded-run-object-store.test.ts @@ -40,7 +40,7 @@ */ import { describe, it, expect } from 'vitest'; -import { assertEngineDeleteDispatch } from '@objectstack/metadata-core'; +import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/metadata-core'; import type { AutomationContext } from '@objectstack/spec/contracts'; import { defineActionDescriptor } from '@objectstack/spec/automation'; @@ -57,8 +57,8 @@ function createTestLogger() { * Minimal in-memory ObjectQL-like engine: rows keyed by id, with `where` * equality filtering — the same double `suspended-run-store.test.ts` drives * {@link ObjectStoreSuspendedRunStore} with, copied rather than shared so this - * file registers no tests it does not own. `delete` routes through the real - * engine's dispatch predicate (`check:engine-double-contract`). + * file registers no tests it does not own. `delete` and `update` route through + * the real engine's dispatch predicates (`check:engine-double-contract`). */ function createFakeEngine(): SuspendedRunStoreEngine & { rows: Map } { const rows = new Map(); @@ -81,6 +81,8 @@ function createFakeEngine(): SuspendedRunStoreEngine & { rows: Map return data; }, async update(_object, data, options) { + // Refuses what a real server refuses (`check:engine-double-contract`). + assertEngineUpdateDispatch(data, options as any); const id = options?.where?.id ?? data.id; const existing = rows.get(String(id)) ?? { id }; rows.set(String(id), { ...existing, ...data }); diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index ae302177dd..ba0ef11772 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3256,6 +3256,16 @@ "verb": "findOne", "pinned": 1 }, + { + "file": "packages/services/service-automation/src/stranded-run-object-store.test.ts", + "verb": "delete", + "pinned": 1 + }, + { + "file": "packages/services/service-automation/src/stranded-run-object-store.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/services/service-automation/src/suspended-run-store.test.ts", "verb": "delete", From 6a768ff80cfd7ca95b390fe4de1edc12f9f748b5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 4 Sep 2026 09:10:39 +0000 Subject: [PATCH 7/7] docs(automation): declare the stranded-row carve-out on sys_automation_run's node_id and variables_json The durable row's own declarations now agree with their writer: on the one terminal-row class that carries a consumed suspension, node_id is the PAUSED node (the Runs surface titles and highlights the row with it), and variables_json is either the restorable snapshot or the store's drop notice, which is not one. Same carve-out pattern node_type's description already carried. The changeset names the visible Runs-surface change. Source-only. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .changeset/stranded-run-status-stamp.md | 9 +++++++-- .../src/sys-automation-run.object.ts | 20 ++++++++++++++++--- 2 files changed, 24 insertions(+), 5 deletions(-) diff --git a/.changeset/stranded-run-status-stamp.md b/.changeset/stranded-run-status-stamp.md index 47359fef32..167063169e 100644 --- a/.changeset/stranded-run-status-stamp.md +++ b/.changeset/stranded-run-status-stamp.md @@ -37,8 +37,13 @@ repairs to how `restoreConsumedSuspension` finds a stranded run's snapshot: object store read that column back as the snapshot's node, so a restore from the row (after a restart, or on another replica) re-armed the run at the failed node and the next resume skipped it while reporting the run - completed. The throwing node stays where the Runs surface reads it: the - row's step log and `error`. + completed. The throwing node stays in the row's step log and `error`. + Visible on the Runs surface: `sys_automation_run`'s row title and highlight + set are built from `node_id` (`titleFormat '{flow_name} · {node_id}'`), so a + stranded run's row now names the PAUSED node — the one an operator can + re-arm — where it named the node that threw; ordinary completed / failed + rows are unchanged. The `node_id` and `variables_json` field descriptions + carry this carve-out, the way `node_type`'s already did. - The verb reads the durable row and its own per-process journal as two witnesses of one strand instead of trusting either alone. The hot copy is preferred when both describe the same pause (it is the verbatim object the diff --git a/packages/services/service-automation/src/sys-automation-run.object.ts b/packages/services/service-automation/src/sys-automation-run.object.ts index 69d085b960..bbc7a552d3 100644 --- a/packages/services/service-automation/src/sys-automation-run.object.ts +++ b/packages/services/service-automation/src/sys-automation-run.object.ts @@ -113,11 +113,20 @@ export const SysAutomationRun = ObjectSchema.create({ flow_version: Field.number({ label: 'Flow Version', required: false, group: 'Identity' }), + // [#13937] The stranded-class carve-out in this description mirrors + // `node_type`'s below, for the same reason: the restore verb re-arms the + // pause from this column (`ObjectStoreSuspendedRunStore` rebuilds the + // consumed snapshot's node from it — there is no other column), so on that + // one terminal-row class the writer (`AutomationEngine.recordLog`) puts + // the PAUSED node here, not the last step. This column is also what the + // Runs surface titles and highlights a row with (`titleFormat` / + // `highlightFields` above), so the carve-out is visible there too, by + // design: the row names the pause an operator can re-arm. node_id: Field.text({ label: 'Node', required: false, maxLength: 255, - description: 'For a suspended run, the node it is paused at (resume continues from its out-edges); for a terminal run, the last node reached.', + description: 'For a suspended run, the node it is paused at (resume continues from its out-edges); for a terminal run, the last node reached — except the one class of terminal row that carries a consumed suspension (a run whose resume consumed its pause and then failed downstream, answered `status: \'stranded\'`), which keeps the PAUSED node so a restore re-arms the pause rather than the node that threw. On that row the last node reached is the final entry of steps_json and the failure is in error; the Runs surface titles the row with this column, so a stranded run reads as its pause node there.', group: 'State', }), @@ -235,11 +244,16 @@ export const SysAutomationRun = ObjectSchema.create({ // [#13909] The presence-discriminator named in this description lives in // ObjectStoreSuspendedRunStore.deserializeConsumedSuspension — one writer, - // one reader, this column is the key for both. + // one reader, this column is the key for both. [#13937] The same column + // carries the store's drop notice when the snapshot was over its row + // budget (`$consumedSuspensionDropped`, read back as + // `RunRecord.consumedSuspensionDropped`), so "present" now has two shapes + // and only one of them is a restorable snapshot — stated in the + // description rather than left to the reader of the column. variables_json: Field.textarea({ label: 'Variables', required: false, - description: 'JSON snapshot of the flow variable map at suspend time. On a terminal row its PRESENCE is the discriminator: nothing but the consumed-suspension path writes it there, so variables_json present on a completed/failed row ⇔ the row carries a restorable suspension — the store\'s deserializer keys off exactly this.', + description: 'JSON snapshot of the flow variable map at suspend time. On a terminal row its PRESENCE is the discriminator: nothing but the consumed-suspension path writes it there, so variables_json present on a completed/failed row ⇔ the row\'s run had a pause that its resume consumed before a downstream node failed — the store\'s deserializer keys off exactly this. Two shapes on such a row: the snapshot itself (a restorable suspension), or a one-key notice `{"$consumedSuspensionDropped": …}` recording that the snapshot existed and was NOT persisted (over the store\'s row budget) — the notice is not a restorable snapshot; such a run can be restored only by the process that stranded it, while it runs.', group: 'State', }),