diff --git a/.changeset/stranded-run-status-stamp.md b/.changeset/stranded-run-status-stamp.md new file mode 100644 index 0000000000..167063169e --- /dev/null +++ b/.changeset/stranded-run-status-stamp.md @@ -0,0 +1,63 @@ +--- +"@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, 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 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 + 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/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..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; } /** @@ -1282,12 +1320,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 { /** @@ -1302,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'; } /** @@ -1355,9 +1407,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 +4972,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 +5323,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 +5344,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 +5496,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 +5530,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, }; @@ -5820,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 @@ -5834,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, @@ -5858,13 +5970,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 +6005,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 * @@ -5906,6 +6020,15 @@ 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: 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 * ordinary {@link resume} afterwards, through the same authority gate, the @@ -5991,12 +6114,43 @@ 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; + // [#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; + if (this.store?.loadTerminal) { try { terminal = await this.store.loadTerminal(runId); } catch (err) { @@ -6014,19 +6168,65 @@ 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 ?? '', - }; + } + + 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); } 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 + // 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`); @@ -6078,9 +6278,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, @@ -6397,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 @@ -6407,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/stranded-run-object-store.test.ts b/packages/services/service-automation/src/stranded-run-object-store.test.ts new file mode 100644 index 0000000000..75ec8d30db --- /dev/null +++ b/packages/services/service-automation/src/stranded-run-object-store.test.ts @@ -0,0 +1,449 @@ +// 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, assertEngineUpdateDispatch } 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` and `update` route through + * the real engine's dispatch predicates (`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) { + // 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 }); + 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 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', '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', '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 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', '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); + }); +}); 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'); + }); +}); 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 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', }), 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",