diff --git a/.changeset/tall-moons-refuse.md b/.changeset/tall-moons-refuse.md new file mode 100644 index 0000000000..3b7dc5b5f8 --- /dev/null +++ b/.changeset/tall-moons-refuse.md @@ -0,0 +1,30 @@ +--- +'@objectstack/service-automation': patch +--- + +Resume a paused flow run from the shared store, not from the replica's own memory of it + +On a multi-replica deployment over one database, approving a level of a multi-level +approval flow could re-create the level that was just approved instead of opening the +next one — so the same approver had to approve each level twice, and a three-level flow +produced five approval requests. Landing the same stale read on the final level rolled +the run back to the previous one and left it parked forever instead of completing. + +The engine kept paused runs in a per-process map and read that map before the durable +`sys_automation_run` row, so a replica that had handled the run earlier answered from +its own snapshot of the node the run was parked at — a snapshot nothing invalidates. +Whichever replica the next decision reached then traversed forward from a node the run +had already left. A single replica never showed it, because there is only one map and +it is never behind. + +The resume path is now store-authoritative: with a `SuspendedRunStore` configured, the +store answers where a run is parked, and the in-memory map is consulted only for a run +whose durable save failed (the existing degradation, which keeps such a run resumable +in-process and reports the lost durability at `error`). The ordering of the resume +itself is unchanged — the suspension is still consumed before downstream traversal. + +Two consequences worth knowing: every resume now reads the store, so an unreadable +store is reported as `STORE_UNAVAILABLE` for a run this process parked itself rather +than being served a possibly-stale snapshot; and the approvals pre-flight +(`hasSuspendedRun`) is answered from the same authoritative read, so a decision is no +longer recorded against a run that another replica has already advanced or finished. diff --git a/packages/services/service-automation/src/builtin/wait-node.test.ts b/packages/services/service-automation/src/builtin/wait-node.test.ts index b50a35c759..63e51a7f6b 100644 --- a/packages/services/service-automation/src/builtin/wait-node.test.ts +++ b/packages/services/service-automation/src/builtin/wait-node.test.ts @@ -375,13 +375,20 @@ describe('wait timer teardown when the pause ends another way (#5512)', () => { * durable store unreadable, so per #4420 the pause is emphatically NOT gone — * that cancelled the only thing left that would ever wake the run. * - * Reachability is not equal across the two sites, and these tests are built to - * say so rather than to look symmetric: `resumeInternal` reads the durable store - * only on a hot-cache MISS, and a run that paused in this process stays cached - * for the life of its suspension. So the end-to-end specimen below is the - * **re-arm** callback (fresh process, empty cache, store consulted for real); - * the arming callback's branch is latent by construction and is pinned at the - * handler level, with the code injected rather than provoked. + * Reachability was not equal across the two sites when these tests were built, + * and they were shaped to say so rather than to look symmetric: `resumeInternal` + * read the durable store only on a MISS of the engine's in-memory map, and a run + * that paused in this process was answered from memory for the life of its + * suspension. So the end-to-end specimen below is the **re-arm** callback (fresh + * process, empty map, store consulted for real), while the arming callback's + * branch was latent by construction and is pinned at the handler level, with the + * code injected rather than provoked. + * + * [#13617] The asymmetry is gone — a store-backed engine now reads the store on + * every resume — but the arming-path specimen below is unchanged on purpose: it + * runs on an engine with NO store at all, where there is no store read to fail, + * so what it pins is still the handler's branch and the arming site's routing + * through it, not a reachability claim. */ describe('wait timer one-shot vs. a shot that never consumed the pause (#5529)', () => { /** A logger that keeps its `error` lines so the diagnostic can be asserted. */ @@ -520,12 +527,11 @@ describe('wait timer one-shot vs. a shot that never consumed the pause (#5529)', it('arming path: the same handler keeps the job armed on STORE_UNAVAILABLE', async () => { // The arming callback shares one handler with the re-arm callback, so this - // pins the branch on THAT site too. The code is injected, not provoked: a run - // that paused in this process is in the engine's hot cache, so its own resume - // never reads the durable store and cannot produce STORE_UNAVAILABLE here. - // Fabricating a cache miss to "prove" otherwise would pin a scenario the - // engine does not have — what is verified is the handler's branch, and that - // the arming site routes through it rather than keeping its own `finally`. + // pins the branch on THAT site too. The code is injected, not provoked, and + // [#13617] did not change that: this engine is built with NO store, so no + // resume of it can produce STORE_UNAVAILABLE however it reads. What is + // verified is the handler's branch, and that the arming site routes through + // it rather than keeping its own `finally`. const { ctx, scheduled, cancelled } = fakeJobCtx(); const engine = new AutomationEngine(silentLogger()); const ran: string[] = []; diff --git a/packages/services/service-automation/src/builtin/wait-node.ts b/packages/services/service-automation/src/builtin/wait-node.ts index 2742df8b14..50191b7c36 100644 --- a/packages/services/service-automation/src/builtin/wait-node.ts +++ b/packages/services/service-automation/src/builtin/wait-node.ts @@ -82,13 +82,16 @@ interface WaitTimerLogger { * found" and the only remaining path is the next boot's overdue re-arm pass. * Both remedies are named in the log line for that reason. * - * Reachability differs by site, and the honest note is that they are not equal. - * `resumeInternal` reads the durable store only on a hot-cache miss, and a run - * that paused in *this* process is cached for as long as the suspension lives — - * so the **re-arm** callback (a fresh process, empty cache) is where - * `STORE_UNAVAILABLE` is genuinely reachable today, while the arming callback's - * branch is latent by construction. It is shared anyway rather than special-cased: - * a second spelling of "settle the one-shot" is exactly the drift #5512 collapsed. + * Reachability was once unequal by site, and this note used to say so: while + * `resumeInternal` read the durable store only on a miss of the engine's + * in-memory map, a run that paused in *this* process was answered from memory + * for the life of its suspension, so only the **re-arm** callback (a fresh + * process, empty map) could genuinely produce `STORE_UNAVAILABLE`. [#13617] + * ended that: the resume path is store-authoritative whenever a + * `SuspendedRunStore` is configured, so BOTH callbacks read the store and both + * reach this branch — the arming site is no longer latent by construction. The + * handler was shared before that was true and stays shared: a second spelling + * of "settle the one-shot" is exactly the drift #5512 collapsed. */ function makeWaitTimerJobHandler( engine: Pick, diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 910c8074ca..bb6ca93701 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1622,11 +1622,31 @@ export class AutomationEngine implements IAutomationService { private readonly runSummaryLog: RunSummaryLogLevel; private logger: Logger; /** - * Runs paused at a node, keyed by runId (ADR-0019). In-memory hot cache — - * mirrored to {@link store} when one is configured, so a pause survives a - * process restart. See {@link SuspendedRun}. + * Runs paused at a node, keyed by runId (ADR-0019). Process-local copy of + * the pause — mirrored to {@link store} when one is configured, so a pause + * survives a process restart. See {@link SuspendedRun}. + * + * [#13617] NOT a read-through cache sitting in front of the store. When a + * store is configured the STORE is the authority and this map answers only + * for the runs it never accepted ({@link cacheOnlySuspensions}). Reading + * this map first is what made a multi-replica approval flow re-create every + * level — the mechanism is in {@link loadSuspendedRunStrict}. */ private suspendedRuns = new Map(); + /** + * [#13617] Runs whose durable save FAILED, so {@link store} holds no row + * for them and its "no such run" says nothing about them. These are the + * only runs {@link loadSuspendedRunStrict} will answer out of + * {@link suspendedRuns} while a store is configured — which is what keeps + * {@link persistSuspendedRun}'s documented degradation (a save failure + * costs cross-restart durability, not in-process resumability) working. + * + * Written by {@link persistSuspendedRun} — added when a save throws, + * cleared when one lands — and dropped alongside the cache entry by + * {@link forgetSuspendedRun}, the single choke point every consumption + * passes through, so it is bounded by the map it qualifies. + */ + private cacheOnlySuspensions = new Set(); /** * Optional durable backing for {@link suspendedRuns}. When set, suspended * runs are persisted on suspend and rehydrated on resume after a restart; @@ -1788,7 +1808,17 @@ export class AutomationEngine implements IAutomationService { if (this.store) { try { await this.store.save(run); + // [#13617] The store now holds this pause, so it — not this map + // — is the answer for it. Cleared here and not only on the + // failure path: a re-suspend whose save lands after an earlier + // one failed must stop being read out of memory. + this.cacheOnlySuspensions.delete(run.runId); } catch (err) { + // [#13617] The store was never given the row, so its "no such + // run" is silence about this run rather than an answer. This is + // what lets `loadSuspendedRunStrict` keep serving it from the + // map — the in-process resumability the message below promises. + this.cacheOnlySuspensions.add(run.runId); // #6499 — the cause is the datasource DRIVER's own text, so it // goes to the logger's STRUCTURED slot, never spliced into the // message; see `forgetSuspendedRun`'s catch below for the full @@ -1831,6 +1861,10 @@ export class AutomationEngine implements IAutomationService { */ private async forgetSuspendedRun(run: SuspendedRun, reason: SuspensionReleaseReason): Promise { this.suspendedRuns.delete(run.runId); + // [#13617] The qualifier goes with the entry it qualifies — this is the + // one choke point every consumption passes through, so nothing can leave + // a run marked "the store never took this" after its map entry is gone. + this.cacheOnlySuspensions.delete(run.runId); if (this.store) { try { await this.store.delete(run.runId); @@ -4521,12 +4555,52 @@ export class AutomationEngine implements IAutomationService { } /** {@link loadSuspendedRun} without the degradation: a store read failure - * THROWS instead of reading as "no such run". */ + * THROWS instead of reading as "no such run". + * + * [#13617] STORE-AUTHORITATIVE. When a {@link SuspendedRunStore} is + * configured, the store answers and {@link suspendedRuns} answers only for + * a run the store never accepted ({@link cacheOnlySuspensions}). It used + * to be the other way round — this process's map first, the store only on + * a miss — which is a correct read for exactly one deployment shape: a + * single process. Put several replicas behind a load balancer over one + * database and that map is a per-replica snapshot of the node a run was + * parked at THE LAST TIME THIS REPLICA TOUCHED IT, and nothing invalidates + * it, because there is no invalidation channel to it at all. + * + * The measured shape, a multi-level approval flow: replica A parks the run + * at `lv1` and keeps it in its map. The `lv1` decision round-robins to + * replica B, which advances the run to `lv2` in the store and in B's map; + * A's map still says `lv1`. The `lv2` decision lands back on A, which read + * its own map, resumed from `lv1`, and traversed to `lv2` a SECOND time — + * the same level re-created as a fresh pending request tens of + * milliseconds after the first one completed, so one approver approves + * every level twice. Land the same one-beat-stale read on the FINAL level + * and the run rolls back to the previous one instead of terminating. A + * single replica shows zero duplicates because there is one map and it is + * never behind. + * + * Both callers that must not be wrong funnel through here: `resumeInternal` + * (which node does this resume continue from) and {@link hasSuspendedRun} + * (the approvals pre-flight that decides whether to record a decision at + * all), so one seam settles both. + * + * ⛔ 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. */ private async loadSuspendedRunStrict(runId: string): Promise { - const cached = this.suspendedRuns.get(runId); - if (cached) return cached; - if (!this.store) return null; - return await this.store.load(runId); + if (!this.store) return this.suspendedRuns.get(runId) ?? null; + const stored = await this.store.load(runId); + if (stored) return stored; + // The store has no row. For every run it ever accepted that IS the + // answer — including the runs this process advanced past, whose stale + // map entries are the whole defect above. The lone exception is a run + // whose durable save failed here: the store was never handed that row, + // so its silence says nothing about it, and `persistSuspendedRun` + // deliberately keeps such a run resumable in-process (it reports the + // lost durability at `error`). + if (this.cacheOnlySuspensions.has(runId)) return this.suspendedRuns.get(runId) ?? null; + return null; } /** diff --git a/packages/services/service-automation/src/multi-replica-resume-staleness.test.ts b/packages/services/service-automation/src/multi-replica-resume-staleness.test.ts new file mode 100644 index 0000000000..a68fe69a7b --- /dev/null +++ b/packages/services/service-automation/src/multi-replica-resume-staleness.test.ts @@ -0,0 +1,352 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * A resume reads the run's state from the SHARED store, not from this replica's + * own memory of where the run was last parked (#13617). + * + * ## The defect + * + * `AutomationEngine.suspendedRuns` is a per-process map of paused runs, and + * `loadSuspendedRunStrict` used to read it FIRST and consult the durable store + * only on a miss. That is a correct read for exactly one deployment shape: a + * single process. Put three replicas behind a load balancer over one postgres + * and the map becomes a per-replica snapshot of the node a run was parked at + * THE LAST TIME THIS REPLICA TOUCHED IT — and nothing invalidates it, because + * nothing is wired to invalidate it. + * + * Reported from a three-replica deployment as: every approval level except the + * first is created twice, so one approver approves each level twice, and a + * three-level flow produces five `sys_approval_request` rows (lv1 x1, lv2 x2, + * lv3 x2). The duplicate's `created_at` sits ~60ms after the previous node's + * `completed_at` — the resume itself re-created the level it had just left. A + * second observed shape, from the same environment: the FINAL approval rolls + * the run back to the previous level and opens a fourth pending node, and the + * run never terminates. One replica, same database, same flow: zero duplicates. + * + * Both shapes are ONE mechanism at two different points in the flow — a resume + * that read a suspension one beat stale and traversed forward from it — which + * is why both are pinned here off the same helper. + * + * ## What is NOT the mechanism + * + * Not the missing `attachClusterPubSub()` from that deployment's boot log: that + * is the METADATA cache-invalidation channel (`MetadataClusterBridgePlugin` over + * `MetadataManager`), and this package has no cluster or pub/sub wiring of any + * kind — there is no invalidation channel here that could have been disabled. + * Attaching that bridge would not move these tests one bit; only reading the + * shared store does. Nor is it leader election on scheduled jobs: an approval + * resume arrives on the decision-write path, not from a job tick. + * + * ## REVERT-PROOF + * + * Restore the old cache-first order at the top of `loadSuspendedRunStrict` + * (`const cached = this.suspendedRuns.get(runId); if (cached) return cached;`) + * and this file goes 4 red / 4 green — measured, not predicted: + * + * - `THE BUG` shape 1 → `[ 'lv1', 'lv2', 'lv2' ]` where `['lv1','lv2','lv3']` + * is correct: the reported duplicate, reproduced literally. + * - `THE BUG` shape 2 → `expected 'paused' to be undefined`: the run did not + * terminate, which is the reported "stays in approval forever". + * - the `RUN_NOT_FOUND` case → the stale replica resumed a FINISHED run and + * reported success. + * - the `NEW REACH` case → no `STORE_UNAVAILABLE`, because a cache hit meant + * the broken store was never read at all. + * + * The four that stay green under that mutation are the ones that must: the + * single-replica control, the healthy cold-replica control, the no-store + * control, and the failed-durable-save degradation. A fix that moved the defect + * instead of removing it would take one of those with it. + */ + +import { describe, it, expect } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; +import { AutomationEngine } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import type { SuspendedRun, SuspendedRunStore } from './engine.js'; + +function silentLogger(): any { + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } }; +} + +/** A three-level approval chain: start -> lv1 -> lv2 -> lv3 -> end. */ +const APPROVAL_FLOW = { + name: 'expense_approval', + label: 'Expense approval', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'lv1', type: 'approval_level', label: 'Department head' }, + { id: 'lv2', type: 'approval_level', label: 'General manager' }, + { id: 'lv3', type: 'approval_level', label: 'Finance' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'lv1' }, + { id: 'e2', source: 'lv1', target: 'lv2' }, + { id: 'e3', source: 'lv2', target: 'lv3' }, + { id: 'e4', source: 'lv3', target: 'end' }, + ], +} as any; + +/** + * One replica: a fresh engine over the SHARED store, appending the id of every + * level it opens to the shared `opened` ledger. `opened` stands in for + * `sys_approval_request` — the table the report counted rows in — so a level + * created twice appears twice here, in order. + * + * `resumeAuthority: 'service'` mirrors the real `approval` node: the owning + * service authorizes and records the decision, then resumes with the in-process + * marker. Resuming any other way is a different door and a different test. + */ +function replica(store: SuspendedRunStore | undefined, opened: string[]): AutomationEngine { + const engine = new AutomationEngine(silentLogger(), store); + engine.registerNodeExecutor({ + type: 'approval_level', + descriptor: defineActionDescriptor({ + type: 'approval_level', + version: '1.0.0', + name: 'Approval level', + supportsPause: true, + resumeAuthority: 'service', + }), + async execute(node) { + opened.push(node.id); + return { success: true, suspend: true, correlation: `req_${node.id}` }; + }, + }); + engine.registerFlow('expense_approval', APPROVAL_FLOW); + return engine; +} + +/** The approve that the approvals service issues once it has recorded a + * decision — the only door a `resumeAuthority: 'service'` pause opens for. */ +function approve(engine: AutomationEngine, runId: string) { + return engine.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any); +} + +describe('multi-replica approval resume reads the shared store (#13617)', () => { + // ── THE BUG, shape 1: a middle level created twice ────────────────────── + // + // Replica schedule (the load balancer's, not ours): A opens the run and is + // therefore holding `lv1` in memory; B takes the `lv1` decision and moves the + // run to `lv2`; the `lv2` decision comes back to A, whose memory still says + // `lv1`. That third hop is the whole defect — A resumed from `lv1` and opened + // `lv2` a second time instead of opening `lv3`. + + it('THE BUG: a replica that fell one level behind must not re-open that level', async () => { + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const a = replica(store, opened); + const b = replica(store, opened); + + const submitted = await a.execute('expense_approval'); + expect(submitted.status).toBe('paused'); + const runId = submitted.runId!; + expect(opened).toEqual(['lv1']); + + // `lv1` is decided on B. The run advances in the shared store; A's memory + // of it does not move, and nothing tells A that it is now stale. + const first = await approve(b, runId); + expect(first.status).toBe('paused'); + expect(opened).toEqual(['lv1', 'lv2']); + + // `lv2` is decided on A — the replica that is one beat behind. + const second = await approve(a, runId); + expect(second.success).toBe(true); + expect(second.status).toBe('paused'); + + // Exactly one request per level, in order. Cache-first read: a second + // 'lv2' lands here instead of 'lv3'. + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + + // …and BOTH replicas agree the run is parked, at the same level: a + // store-authoritative read is the same read everywhere. Proven by taking + // the last decision on the OTHER replica — it terminates the run and opens + // nothing, which only holds if it resumed from `lv3` too. + expect(await a.hasSuspendedRun(runId)).toBe(true); + expect(await b.hasSuspendedRun(runId)).toBe(true); + const final = await approve(b, runId); + expect(final.success).toBe(true); + expect(final.status).toBeUndefined(); + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + }); + + // ── THE BUG, shape 2: the final approval rolls the run back ───────────── + // + // Same mechanism, landing on the LAST level. Reported as: three levels each + // approved exactly once, then the third approval re-opens a fourth pending + // node and the run stays "in approval" forever instead of completing. + + it('THE BUG: the final approval terminates the run, it does not roll it back', async () => { + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const a = replica(store, opened); + const b = replica(store, opened); + + const runId = (await a.execute('expense_approval')).runId!; + // A takes the first decision too, so A's memory is current at `lv2` — one + // beat behind is what this shape needs, not two. + await approve(a, runId); + // B takes the second: the run moves to `lv3` in the store, A still says `lv2`. + expect((await approve(b, runId)).status).toBe('paused'); + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + + // The final decision lands on the stale replica. + const final = await approve(a, runId); + + // It TERMINATES. `status` is absent on a completed run — a `'paused'` here + // is the reported "single stays in approval and never finishes". + expect(final.success).toBe(true); + expect(final.status).toBeUndefined(); + + // No fourth node was opened, on any level. + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + + // Nothing is left parked, as either replica sees it. + for (const node of [a, b]) { + expect(await node.hasSuspendedRun(runId)).toBe(false); + } + }); + + // ── The store's silence is authoritative, not overridden by stale memory ── + + it('a finished run is RUN_NOT_FOUND on the replica still holding it in memory', async () => { + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const a = replica(store, opened); + const b = replica(store, opened); + + const runId = (await a.execute('expense_approval')).runId!; + // B walks the run all the way out. A's memory still holds the `lv1` pause. + for (let i = 0; i < 3; i++) await approve(b, runId); + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + + const late = await approve(a, runId); + expect(late.success).toBe(false); + // The classified refusal, not a silent no-op: approvals persists a decision + // BEFORE resuming and must be able to tell "gone for good" from "retry". + expect(late.code).toBe('RUN_NOT_FOUND'); + // A refusal carries no run status — nothing dispatched. + expect(late.status).toBeUndefined(); + // And it re-opened nothing on its way to that answer. + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + }); + + // ── Negative controls ─────────────────────────────────────────────────── + + it('NEGATIVE CONTROL: one replica is unchanged — three levels, three requests', async () => { + // The report's own control: same database, same flow definition, a single + // app replica, zero duplicates. It passed before this fix and must keep + // passing after it, or the fix moved the defect rather than removing it. + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const only = replica(store, opened); + + const runId = (await only.execute('expense_approval')).runId!; + expect((await approve(only, runId)).status).toBe('paused'); + expect((await approve(only, runId)).status).toBe('paused'); + const final = await approve(only, runId); + + expect(final.success).toBe(true); + expect(final.status).toBeUndefined(); + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + }); + + it('NEGATIVE CONTROL: the healthy multi-replica path still advances once per level', async () => { + // Every decision lands on a replica that has never seen this run — the + // round-robin case that always worked, because a cold replica had nothing + // stale to read. It must still advance exactly one level per approval. + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const submitter = replica(store, opened); + + const runId = (await submitter.execute('expense_approval')).runId!; + expect((await approve(replica(store, opened), runId)).status).toBe('paused'); + expect((await approve(replica(store, opened), runId)).status).toBe('paused'); + const final = await approve(replica(store, opened), runId); + + expect(final.success).toBe(true); + expect(final.status).toBeUndefined(); + expect(opened).toEqual(['lv1', 'lv2', 'lv3']); + }); + + it('NEGATIVE CONTROL: with no store at all, behaviour is purely in-memory', async () => { + // The historical default (LiteKernel, tests, dev): no durable store, so the + // engine's own map IS the authority and there is nothing to be stale + // against. A run parked here resumes here and nowhere else. + const opened: string[] = []; + const solo = replica(undefined, opened); + + const runId = (await solo.execute('expense_approval')).runId!; + expect((await approve(solo, runId)).status).toBe('paused'); + expect(opened).toEqual(['lv1', 'lv2']); + + // A second engine shares no state with it — nothing was persisted. + const other = replica(undefined, opened); + const elsewhere = await approve(other, runId); + expect(elsewhere.success).toBe(false); + expect(elsewhere.code).toBe('RUN_NOT_FOUND'); + }); + + it('NEGATIVE CONTROL: a run the store never accepted is still resumable in-process', async () => { + // `persistSuspendedRun` documents this degradation and logs it at `error`: + // a failed durable save costs CROSS-RESTART durability, not in-process + // resumability. Making the store authoritative must not quietly convert + // that into an unresumable run — the store's "no row" says nothing about a + // row it was never handed. + const opened: string[] = []; + const saves: string[] = []; + const writeOnlyFailingStore: SuspendedRunStore = { + async save(run: SuspendedRun) { saves.push(run.nodeId); throw new Error('sqlite: disk I/O error'); }, + async load() { return null; }, + async delete() {}, + async list() { return []; }, + }; + const engine = replica(writeOnlyFailingStore, opened); + + const runId = (await engine.execute('expense_approval')).runId!; + expect(saves).toEqual(['lv1']); + + // The pause never reached the store, and this process can still continue it. + const advanced = await approve(engine, runId); + expect(advanced.success).toBe(true); + expect(advanced.status).toBe('paused'); + expect(opened).toEqual(['lv1', 'lv2']); + }); + + it('NEW REACH: an unreadable store is STORE_UNAVAILABLE, never a lost run', async () => { + // Not a control — this case is RED before the fix, and deliberately so. It + // is the reachability change store-authoritative reading buys: a run parked + // in THIS process used to be answered from memory for the life of its + // suspension, so its own resume never touched the store and could not + // produce STORE_UNAVAILABLE at all (`builtin/wait-node.ts` documented that + // asymmetry, and its note is corrected in the same change). Now every + // resume reads the store, so the outage is reported instead of papered + // over with a snapshot that may be stale. + // + // The code matters as much as the refusal (#4420): an outage is "unknown", + // not "gone". A caller that already wrote a decision needs "retry when the + // store is back" to be distinguishable from "this run is gone for good" — + // same failure, opposite remedy. + const opened: string[] = []; + const store = new InMemorySuspendedRunStore(); + const engine = replica(store, opened); + const runId = (await engine.execute('expense_approval')).runId!; + + const brokenStore: SuspendedRunStore = { + async save() {}, + async load() { throw new Error('sqlite: database is locked'); }, + async delete() {}, + async list() { return []; }, + }; + engine.setSuspendedRunStore(brokenStore); + + const refused = await approve(engine, runId); + expect(refused.success).toBe(false); + expect(refused.code).toBe('STORE_UNAVAILABLE'); + expect(refused.status).toBeUndefined(); + // Refused before consuming anything: no level was opened or re-opened. + expect(opened).toEqual(['lv1']); + }); +}); diff --git a/packages/services/service-automation/src/suspended-screen-durability.test.ts b/packages/services/service-automation/src/suspended-screen-durability.test.ts index f2b390b84d..259e0e3243 100644 --- a/packages/services/service-automation/src/suspended-screen-durability.test.ts +++ b/packages/services/service-automation/src/suspended-screen-durability.test.ts @@ -74,8 +74,9 @@ describe('getSuspendedScreen is durable (#4515)', () => { const paused = await engine.execute('onboard'); expect(paused.status).toBe('paused'); - // Same process — served from the in-memory hot cache, no store read - // needed. (Async now, but the answer is identical.) + // Same process. [#13617] made this read store-authoritative, so it does + // consult the store — the point of the case is unchanged: the engine + // that paused the run still renders its screen. const screen = await engine.getSuspendedScreen(paused.runId!); expect(screen).toMatchObject({ nodeId: 'collect', title: 'Your details' }); expect(screen!.fields[0]).toMatchObject({ name: 'full_name', required: true });