diff --git a/.changeset/olive-parrots-attend.md b/.changeset/olive-parrots-attend.md new file mode 100644 index 0000000000..8348c6a7fa --- /dev/null +++ b/.changeset/olive-parrots-attend.md @@ -0,0 +1,11 @@ +--- +'@objectstack/service-automation': patch +--- + +**The last three readers of suspended-run state read the shared store, not this replica's memory of it.** #13617 made the resume path store-authoritative; `cancelRun`, `failAncestors` and `listSuspendedRunsDurable` still preferred the per-process `suspendedRuns` map, so on a replica holding a stale entry each acted on the node a run was parked at the last time THIS replica touched it. All three now take one answer to "where is this run parked", through the existing `loadSuspendedRun` / `loadSuspendedRunStrict` pair, with the degrading or strict loader chosen per site so each recorded degradation posture is preserved by choice rather than re-derived. + +- **`cancelRun` — the strict loader.** Before: a stale replica cancelled from its own snapshot; the row deletion is by id and was right either way, but `forgetSuspendedRun` told the executor of the node in the SNAPSHOT that its pause was over, so the live node's pause stayed armed and a node the run had already left was released a second time. Now the shared row decides which pause is torn down. The strict loader is deliberate: "not found" still returns `false` (already terminal / unknown) exactly as before, and an unreadable store still lands on this seam's own #4632 DURABILITY record at `error` — the degrading loader would have answered `null` under its best-effort `warn` and silently downgraded that verdict. ⚠️ One consequence, stated: while a store is configured this process's map is no longer an answer, so a store outage now reaches that `error` record even for a run this replica is holding, where the old cache-first read cancelled from the local snapshot. +- **`failAncestors` — the degrading loader.** Before: a stale parent was failed at a node it had already left (#13617's own harm shape, one level up). The degrading loader is deliberate: this walk runs inside the catch arm already handling a run's failure, so it must not throw, and "a store failure reads as no ancestor here and stops the walk" is exactly the posture the bare `.catch(() => null)` had. The one thing gained beyond the fix: that silent swallow is now recorded, at the loader's declared best-effort `warn` — no new `error` seam. +- **`listSuspendedRunsDurable` — the merge direction, and the comment.** The durable row now wins an id collision; the comment claiming "In-memory entries win — they are the freshest copy" is corrected, since it is true of exactly one deployment shape. Map entries the durable listing does not carry are still included, deliberately: `store.list()` is a capped, best-effort enumeration (at most 1000 `paused` rows) and the same merge is reached on the degraded path, so "absent from the list" is not the per-id "the store answered and has no row" the strict loader rests on. + +No signature, export or return-shape change on any of the three. diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 111b718544..f2c33d8b57 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -5330,54 +5330,73 @@ export class AutomationEngine implements IAutomationService { * indistinguishable to the caller, so the run may still be parked and * resumable. That path is reported at `error` (#4632/#6299) precisely * because nothing above it can tell the difference; see the catch below. + * + * [#14332] WHERE THE RUN IS READ FROM: {@link loadSuspendedRunStrict} — + * the same store-authoritative read `resumeInternal` takes, and the STRICT + * loader by deliberate choice rather than the degrading + * {@link loadSuspendedRun}. NOT FOUND (the store answered and holds no row + * for this id) reads as "already terminal / unknown" and returns `false`, + * exactly as before; a store that cannot be READ throws out of the loader + * into the catch below, which keeps this site's own #4632 DURABILITY record + * at `error`. The degrading loader would have answered `null` under its own + * best-effort `warn` and silently downgraded that verdict — the posture is + * preserved here by picking the loader that preserves it. + * + * ⚠️ The consequence of a store-authoritative read, stated rather than left + * to be discovered: while a store is configured this process's map is no + * longer an answer, so a store outage reaches the `error` record above even + * for a run THIS replica is holding — where the old cache-first read + * cancelled from the local snapshot instead. That snapshot is the defect: + * the row delete is by id and is therefore right either way, but + * {@link forgetSuspendedRun} notifies the executor of the node recorded on + * the SNAPSHOT, so a stale replica tore down the pause of a node the run had + * already left and left the live one's armed. */ async cancelRun(runId: string, reason?: string): Promise { - let run = this.suspendedRuns.get(runId) ?? null; - if (!run && this.store) { - try { - run = await this.store.load(runId); - } catch (err) { - // #6299 — same family, same mechanism as `forgetSuspendedRun` - // above: the driver's uncontrolled text goes to the structured - // slot so the record stays one physical line. - // - // #4632 verdict: DURABILITY — raised from `warn` to `error`. The - // failed read is silently turned into "no such suspended run" - // and this method returns `false`, which its own contract - // documents as idempotent success (already terminal / unknown), - // so the cancellation is SKIPPED while the call reads clean. The - // only in-repo caller measures the cost: plugin-approvals' - // revise-window recall - // (`packages/plugins/plugin-approvals/src/approval-service.ts`) - // never reads the boolean at all — it only catches a THROW, and - // grades that throw `error` with "the run may be stranded" - // (#4420). A store-read failure produces precisely that stranded - // run WITHOUT firing that alarm: the request is marked - // `recalled`, the record lock is released, `resumeError` stays - // undefined — and the run stays parked in the store, to be - // re-armed and resumed by the next restart, inside a flow whose - // approval has already been withdrawn. - // - // This is why #6230's verdict must not be copied here. - // `loadSuspendedRun` is a DECLARED best-effort reader for - // incidental callers (a gate lookup, a screen fetch), and - // `resumeInternal` takes the strict form exactly where the - // difference matters. `cancelRun` has no strict alternative, and - // its degradation decides a WRITE. - // - // THIRD argument (`error(message, error?, meta?)`), `Error` slot - // deliberately empty (#5575). - this.logger.error( - `[automation] cancelRun('${runId}') could not read the durable suspended-run store, so the ` + - `cancellation was SKIPPED and reported as idempotent success — this call returns false, which ` + - `its callers read as "no such suspended run". The run is NOT cancelled: if it is parked in the ` + - `store it stays parked, and the next restart re-arms and resumes it while the caller has ` + - `already recorded the cancellation. Fix the store failure in this record's meta, then re-issue ` + - `cancelRun('${runId}').`, - undefined, - describeThrownForLog(err), - ); - } + let run: SuspendedRun | null = null; + try { + run = await this.loadSuspendedRunStrict(runId); + } catch (err) { + // #6299 — same family, same mechanism as `forgetSuspendedRun` + // above: the driver's uncontrolled text goes to the structured + // slot so the record stays one physical line. + // + // #4632 verdict: DURABILITY — raised from `warn` to `error`. The + // failed read is silently turned into "no such suspended run" + // and this method returns `false`, which its own contract + // documents as idempotent success (already terminal / unknown), + // so the cancellation is SKIPPED while the call reads clean. The + // only in-repo caller measures the cost: plugin-approvals' + // revise-window recall + // (`packages/plugins/plugin-approvals/src/approval-service.ts`) + // never reads the boolean at all — it only catches a THROW, and + // grades that throw `error` with "the run may be stranded" + // (#4420). A store-read failure produces precisely that stranded + // run WITHOUT firing that alarm: the request is marked + // `recalled`, the record lock is released, `resumeError` stays + // undefined — and the run stays parked in the store, to be + // re-armed and resumed by the next restart, inside a flow whose + // approval has already been withdrawn. + // + // This is why #6230's verdict must not be copied here. + // `loadSuspendedRun` is a DECLARED best-effort reader for + // incidental callers (a gate lookup, a screen fetch), and + // `resumeInternal` takes the strict form exactly where the + // difference matters. `cancelRun` has no strict alternative, and + // its degradation decides a WRITE. + // + // THIRD argument (`error(message, error?, meta?)`), `Error` slot + // deliberately empty (#5575). + this.logger.error( + `[automation] cancelRun('${runId}') could not read the durable suspended-run store, so the ` + + `cancellation was SKIPPED and reported as idempotent success — this call returns false, which ` + + `its callers read as "no such suspended run". The run is NOT cancelled: if it is parked in the ` + + `store it stays parked, and the next restart re-arms and resumes it while the caller has ` + + `already recorded the cancellation. Fix the store failure in this record's meta, then re-issue ` + + `cancelRun('${runId}').`, + undefined, + describeThrownForLog(err), + ); } if (!run) return false; await this.forgetSuspendedRun(run, 'cancelled'); @@ -5747,9 +5766,19 @@ export class AutomationEngine implements IAutomationService { let parentId = (context as Record | undefined)?.$parentRunId; let hops = 0; while (typeof parentId === 'string' && parentId && hops++ < 32) { - const parent = - this.suspendedRuns.get(parentId) ?? - (this.store ? await this.store.load(parentId).catch(() => null) : null); + // [#14332] The DEGRADING loader, by deliberate choice: this walk runs + // inside the catch arm that is already handling a run's failure, so it + // must not throw, and its recorded posture is exactly + // `loadSuspendedRun`'s — a store failure reads as "no ancestor here" + // and stops the walk. What changes is only WHICH suspension is read: + // the shared store's, not this replica's memory of where the parent + // was last parked. The old `??` chain had #13617's own harm shape — + // a stale parent failed at a node it had already left, so + // `forgetSuspendedRun` released the wrong node's pause. The one thing + // gained beyond that: the bare `.catch(() => null)` swallowed a store + // failure in total silence, and the loader records it (at `warn`, + // its declared best-effort level — no new `error` seam here). + const parent = await this.loadSuspendedRun(parentId); if (!parent) return; await this.failSuspendedRun(parent, `subflow descendant failed: ${error}`); parentId = (parent.context as Record | undefined)?.$parentRunId; @@ -5776,9 +5805,14 @@ export class AutomationEngine implements IAutomationService { /** * Like {@link listSuspendedRuns} but includes runs held only in the durable - * {@link SuspendedRunStore} (e.g. suspended before a restart). The in-memory - * cache takes precedence on id collisions. Falls back to the in-memory list - * when no store is configured. + * {@link SuspendedRunStore} (e.g. suspended before a restart). Falls back to + * the in-memory list when no store is configured. + * + * [#14332] The DURABLE row wins an id collision — the store is the shared + * answer to "where is this run parked" and this process's map is only its + * own memory of it. A run present in the map but absent from the durable + * listing is still included, because a capped or failed enumeration is not + * the per-id "no row" the strict loader rests on; see the merge below. */ async listSuspendedRunsDurable(): Promise> { const byId = new Map(); @@ -5831,8 +5865,28 @@ export class AutomationEngine implements IAutomationService { ); } } - // In-memory entries win — they are the freshest copy. + // [#14332] The DURABLE row wins a collision. The comment this replaces + // said the opposite — "In-memory entries win — they are the freshest + // copy" — which is true of exactly one deployment shape, a single + // process. Put several replicas over one store and this map is a + // per-replica snapshot of the node a run was parked at THE LAST TIME + // THIS REPLICA TOUCHED IT, with no invalidation channel to it at all + // (the mechanism is in {@link loadSuspendedRunStrict}), so preferring it + // reported a run at a node it had already left. + // + // Map entries the durable list does not carry are still appended, and + // that is NOT the strict loader's rule being softened: a LIST is not a + // per-id answer. `store.list()` is a capped, best-effort enumeration + // (`ObjectStoreSuspendedRunStore` reads at most 1000 `paused` rows) and + // this line is also reached on the DEGRADED path above, where the + // enumeration failed outright and `byId` is empty. So "absent from the + // list" is not the evidence "the store answered and has no row" is, + // which is what {@link loadSuspendedRunStrict} rests on when it lets + // only {@link cacheOnlySuspensions} answer out of the map. Applying that + // qualifier here would let a truncated or failed enumeration silently + // drop live runs from an operability listing. for (const r of this.suspendedRuns.values()) { + if (byId.has(r.runId)) continue; byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation }); } return [...byId.values()]; diff --git a/packages/services/service-automation/src/multi-replica-suspended-run-readers.test.ts b/packages/services/service-automation/src/multi-replica-suspended-run-readers.test.ts new file mode 100644 index 0000000000..a0febf6bf8 --- /dev/null +++ b/packages/services/service-automation/src/multi-replica-suspended-run-readers.test.ts @@ -0,0 +1,434 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * The three OTHER readers of suspended-run state read the SHARED store, not + * this replica's memory of where the run was last parked (#14332). + * + * ## The defect + * + * #13617 made the RESUME path store-authoritative: `loadSuspendedRunStrict` + * reads the shared row and consults the per-process map only for a run the + * store never accepted. Three other readers of the same state still preferred + * the map, each with the same mechanism and a different consequence: + * + * 1. `cancelRun` — the row deletion is by id and is therefore right either + * way, but `forgetSuspendedRun` notifies the executor of the node recorded + * on the object it was handed. Cancel from a stale snapshot and the WRONG + * node executor is told its pause is over: whatever the live node armed is + * not torn down, and a node the run left long ago is released a second + * time. So every pin below asserts on the NOTIFICATION, never on the row. + * 2. `failAncestors` — #13617's own harm shape one level up: a stale parent is + * failed at a node it has already left. + * 3. `listSuspendedRunsDurable` — merged the durable list under the in-process + * map, so the listing reported the node this replica last saw rather than + * the node the run is parked at. + * + * ## Why a sibling file rather than more cases in the #13617 harness + * + * `multi-replica-resume-staleness.test.ts` carries a REVERT-PROOF ledger in its + * module docblock: one named mutation (restoring the cache-first read at the + * top of `loadSuspendedRunStrict`) and the measured 4 red / 4 green split it + * produces. Adding these cases to that file would falsify those counts and mix + * two different mutations' revert-proofs into one statement. This file gets its + * own, below, and shares only the two-engines-over-one-store SHAPE. + * + * ## REVERT-PROOF + * + * Restore ONE site's map-first read at a time and this file splits as follows — + * measured per site on the committed tree, not predicted: + * + * - site 1, `run = this.suspendedRuns.get(runId) ?? await this.loadSuspendedRunStrict(runId)` + * → **2 red / 9 green**. `THE BUG` reports the teardown naming `lv1` while + * the run is parked at `lv2` — the wrong node executor told its pause is + * over. `NEW REACH` goes red with it, and that is the same fact seen from + * the other side: a map hit means the unreadable store is never read at all, + * so the cancel proceeds from the local snapshot instead of degrading. + * - site 2, `const parent = this.suspendedRuns.get(parentId) ?? await this.loadSuspendedRun(parentId)` + * → **2 red / 9 green**. The ancestor is failed at `lv1`, a node it has + * already left; the unreadable-ancestor control goes with it for site 1's + * reason — a map hit never reaches the store, so the walk that should have + * stopped instead tears down a stale parent. + * - site 3, drop the `byId.has(r.runId)` guard so the map overwrites again + * → **1 red / 10 green**. The listing reports `lv1`, one level stale. + * + * What stays green under all three is what must: the no-store cases where the + * map IS the authority, the failed-durable-save cases the store's silence says + * nothing about, and the unlistable store's documented short list. Each + * mutation was proven on disk by anchored occurrence counts before its run and + * restored with `git checkout HEAD --`, the restore proven by `git hash-object` + * against the HEAD blob. + */ + +import { describe, it, expect } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; +import type { AutomationContext } 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(); } }; +} + +/** What `NodeExecutor.onSuspensionReleased` was told — the fact under test. */ +interface Release { + runId: string; + flowName: string; + nodeId: string; + correlation?: string; + reason: string; +} + +/** 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; + +/** + * A subflow child that pauses and then fails downstream — the only door to + * `failAncestors`, which runs in `resumeInternal`'s catch arm. The child's own + * pause is deliberately a different node type with no release hook, so the + * `released` ledger holds ancestor teardowns alone. + */ +const CHILD_FLOW = { + name: 'child_task', + label: 'Child task', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'hold', type: 'child_pause', label: 'Hold' }, + { id: 'write_back', type: 'exploding_writer', label: 'Write back' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'hold' }, + { id: 'e2', source: 'hold', target: 'write_back' }, + { id: 'e3', source: 'write_back', target: 'end' }, + ], +} as any; + +/** + * One replica: a fresh engine over the SHARED store. `opened` records every + * level opened (a level re-opened appears twice); `released` records every + * suspension teardown the approval executor is notified of — the node id and + * the correlation IT minted, which is what says whose pause was torn down. + */ +function replica( + store: SuspendedRunStore | undefined, + opened: string[], + released: Release[], +): 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}` }; + }, + async onSuspensionReleased(release) { + released.push({ + runId: release.runId, + flowName: release.flowName, + nodeId: release.nodeId, + correlation: release.correlation, + reason: release.reason, + }); + }, + }); + engine.registerNodeExecutor({ + type: 'child_pause', + descriptor: defineActionDescriptor({ + type: 'child_pause', + version: '1.0.0', + name: 'Child pause', + supportsPause: true, + resumeAuthority: 'any', + }), + async execute() { + return { success: true, suspend: true, correlation: 'child:hold' }; + }, + }); + engine.registerNodeExecutor({ + type: 'exploding_writer', + async execute() { + throw new Error('update_record(expense_claim) failed: Record not found in expense_claim'); + }, + }); + engine.registerFlow('expense_approval', APPROVAL_FLOW); + engine.registerFlow('child_task', CHILD_FLOW); + return engine; +} + +/** The approve the approvals service issues once it has recorded a decision. */ +function approve(engine: AutomationEngine, runId: string) { + return engine.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any); +} + +/** A store that saves and lists fine but cannot be READ for one id. */ +function unreadableFor(inner: SuspendedRunStore, blindId: () => string): SuspendedRunStore { + return { + save: (run) => inner.save(run), + async load(runId) { + if (runId === blindId()) throw new Error('sqlite: database is locked'); + return inner.load(runId); + }, + delete: (runId) => inner.delete(runId), + list: () => inner.list(), + }; +} + +/** Park a run on `a` at `lv1`, then let `b` advance it to `lv2` in the store. */ +async function parkedOnOneReplicaAdvancedOnAnother( + store: SuspendedRunStore, + opened: string[], + released: Release[], +): Promise<{ a: AutomationEngine; b: AutomationEngine; runId: string }> { + const a = replica(store, opened, released); + const b = replica(store, opened, released); + const runId = (await a.execute('expense_approval')).runId!; + expect(opened).toEqual(['lv1']); + // The lv1 decision round-robins to B: the shared row advances to lv2 while + // A's map still says lv1, and nothing tells A that it is now stale. + expect((await approve(b, runId)).status).toBe('paused'); + expect(opened).toEqual(['lv1', 'lv2']); + return { a, b, runId }; +} + +// ── site 1: cancelRun ─────────────────────────────────────────────────────── + +describe('#14332 site 1 — cancelRun tears down the pause the run is ACTUALLY parked at', () => { + it('THE BUG: a stale replica must not tell the wrong node executor its pause is over', async () => { + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const released: Release[] = []; + const { a, b, runId } = await parkedOnOneReplicaAdvancedOnAnother(store, opened, released); + + // The submitter withdraws, and the recall lands on the STALE replica. + expect(await a.cancelRun(runId, 'submitter withdrew the claim')).toBe(true); + + // The live pause is lv2's. Cache-first read: `lv1` / `req_lv1` lands here — + // lv2's executor never learns its pause ended and whatever it armed stays + // armed, while lv1's is released a second time. + expect(released.filter(r => r.reason === 'cancelled')).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv2', correlation: 'req_lv2', reason: 'cancelled' }, + ]); + + // The row deletion is by id and is correct either way — which is exactly + // why the assertion above is on the notification and not on the row. + expect(await store.load(runId)).toBeNull(); + expect(await b.hasSuspendedRun(runId)).toBe(false); + }); + + it('NEGATIVE CONTROL: with no store the map IS the authority and the cancel still tears down', async () => { + const opened: string[] = []; + const released: Release[] = []; + const solo = replica(undefined, opened, released); + + const runId = (await solo.execute('expense_approval')).runId!; + expect(await solo.cancelRun(runId, 'withdrawn')).toBe(true); + expect(released.filter(r => r.reason === 'cancelled')).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv1', correlation: 'req_lv1', reason: 'cancelled' }, + ]); + }); + + it('NEGATIVE CONTROL: a run the store never accepted is still cancellable in-process', async () => { + // `persistSuspendedRun`'s documented degradation: a failed durable save + // costs cross-restart durability, not in-process resumability — and not + // cancellability either. `loadSuspendedRunStrict`'s `cacheOnlySuspensions` + // branch is the whole reason this stays true through a store-authoritative + // read: the store was never handed the row, so its silence says nothing. + const opened: string[] = []; + const released: Release[] = []; + const saves: string[] = []; + const writeFailingStore: 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(writeFailingStore, opened, released); + + const runId = (await engine.execute('expense_approval')).runId!; + expect(saves).toEqual(['lv1']); + expect(await engine.cancelRun(runId, 'withdrawn')).toBe(true); + expect(released.filter(r => r.reason === 'cancelled')).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv1', correlation: 'req_lv1', reason: 'cancelled' }, + ]); + }); + + it('NEW REACH: an unreadable store is "unknown", not a licence to tear down the local snapshot', async () => { + // The reachability this fix buys, and the one behaviour change it makes on + // this site: a run parked in THIS process used to be answered from memory, + // so its own cancel never touched the store. Now the read is + // store-authoritative, so an outage lands on the #4632 DURABILITY record + // this seam already carries (pinned byte-for-byte in + // `suspended-run-store-consume-log-cause.test.ts`) instead of tearing down + // a snapshot that may name a node the run has left. The cancellation is + // skipped, and the run stays parked and resumable — which is what that + // record tells the operator. + const opened: string[] = []; + const released: Release[] = []; + const store = new InMemorySuspendedRunStore(); + const engine = replica(store, opened, released); + const runId = (await engine.execute('expense_approval')).runId!; + + engine.setSuspendedRunStore(unreadableFor(store, () => runId)); + + expect(await engine.cancelRun(runId, 'withdrawn')).toBe(false); + expect(released.filter(r => r.reason === 'cancelled'), 'nothing was torn down').toEqual([]); + expect(await store.load(runId), 'the run is still parked').not.toBeNull(); + }); +}); + +// ── site 2: failAncestors ─────────────────────────────────────────────────── + +describe('#14332 site 2 — failAncestors fails the ancestor at the node it is ACTUALLY parked at', () => { + it('THE BUG: a stale parent must not be failed at a node it has already left', async () => { + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const released: Release[] = []; + const { a, runId: parentRunId } = await parkedOnOneReplicaAdvancedOnAnother(store, opened, released); + + // A subflow child of that parent, launched and then resumed on the STALE + // replica, whose downstream node throws — `resumeInternal`'s catch arm is + // the only caller of `failAncestors`. + const child = await a.execute('child_task', { $parentRunId: parentRunId } as unknown as AutomationContext); + expect(child.status).toBe('paused'); + const failed = await a.resume(child.runId!); + expect(failed.success).toBe(false); + expect(failed.error).toContain('Record not found in expense_claim'); + + // The ancestor is failed where it is parked — lv2. Cache-first read: lv1, + // so lv2's approval sits armed under a run that has just been failed. + // Filtered to the TERMINAL teardown: the setup's own `lv1` release (reason + // `resumed`, when B took the lv1 decision) is a different, correct fact. + expect(released.filter(r => r.runId === parentRunId && r.reason === 'failed')).toEqual([ + { runId: parentRunId, flowName: 'expense_approval', nodeId: 'lv2', correlation: 'req_lv2', reason: 'failed' }, + ]); + expect(await store.load(parentRunId), 'the ancestor is terminal').toBeNull(); + }); + + it('NEGATIVE CONTROL: with no store, the in-process ancestor is still failed', async () => { + const opened: string[] = []; + const released: Release[] = []; + const solo = replica(undefined, opened, released); + + const parentRunId = (await solo.execute('expense_approval')).runId!; + const child = await solo.execute('child_task', { $parentRunId: parentRunId } as unknown as AutomationContext); + expect((await solo.resume(child.runId!)).success).toBe(false); + + expect(released.filter(r => r.runId === parentRunId && r.reason === 'failed')).toEqual([ + { runId: parentRunId, flowName: 'expense_approval', nodeId: 'lv1', correlation: 'req_lv1', reason: 'failed' }, + ]); + }); + + it('NEGATIVE CONTROL: an unreadable ancestor stops the walk, it does not throw out of the failure path', async () => { + // The posture the bare `.catch(() => null)` had and the degrading loader + // keeps: this walk runs inside the catch arm that is already handling a + // failure, so a store outage must read as "no ancestor here" and end the + // walk — never propagate. The child's own failure is what the caller sees. + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const released: Release[] = []; + let parentRunId = ''; + const a = replica(unreadableFor(store, () => parentRunId), opened, released); + + parentRunId = (await a.execute('expense_approval')).runId!; + const child = await a.execute('child_task', { $parentRunId: parentRunId } as unknown as AutomationContext); + const failed = await a.resume(child.runId!); + + expect(failed.success).toBe(false); + expect(failed.error, "the child's own failure, not the store's").toContain('Record not found in expense_claim'); + expect(released.filter(r => r.runId === parentRunId), 'no ancestor was released').toEqual([]); + expect(await store.load(parentRunId), 'the ancestor is untouched and still parked').not.toBeNull(); + }); +}); + +// ── site 3: listSuspendedRunsDurable ──────────────────────────────────────── + +describe('#14332 site 3 — listSuspendedRunsDurable prefers the durable row over this process\'s memory', () => { + it('THE BUG: the durable row wins an id collision, the stale in-process copy does not', async () => { + const store = new InMemorySuspendedRunStore(); + const opened: string[] = []; + const released: Release[] = []; + const { a, runId } = await parkedOnOneReplicaAdvancedOnAnother(store, opened, released); + + // Cache-first merge: `lv1` / `req_lv1` — the node this replica last saw. + expect(await a.listSuspendedRunsDurable()).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv2', correlation: 'req_lv2' }, + ]); + }); + + it('a run the store never accepted is still listed — the cache is its only copy', async () => { + const opened: string[] = []; + const released: Release[] = []; + const writeFailingStore: SuspendedRunStore = { + async save() { throw new Error('sqlite: disk I/O error'); }, + async load() { return null; }, + async delete() {}, + async list() { return []; }, + }; + const engine = replica(writeFailingStore, opened, released); + + const runId = (await engine.execute('expense_approval')).runId!; + expect(await engine.listSuspendedRunsDurable()).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv1', correlation: 'req_lv1' }, + ]); + }); + + it('NEGATIVE CONTROL: an unlistable store still degrades to the in-memory cache alone', async () => { + // #4632 verdict FUNCTIONAL, unchanged: the enumeration failed, the rows are + // intact, and the listing says so at `warn` while answering from the cache. + // The merge direction must not turn that degradation into an empty answer. + const opened: string[] = []; + const released: Release[] = []; + const inner = new InMemorySuspendedRunStore(); + const unlistable: SuspendedRunStore = { + save: (run) => inner.save(run), + load: (runId) => inner.load(runId), + delete: (runId) => inner.delete(runId), + async list(): Promise { throw new Error('sqlite: database is locked'); }, + }; + const engine = replica(unlistable, opened, released); + + const runId = (await engine.execute('expense_approval')).runId!; + expect(await engine.listSuspendedRunsDurable()).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv1', correlation: 'req_lv1' }, + ]); + }); + + it('NEGATIVE CONTROL: with no store the durable listing is the in-memory listing', async () => { + const opened: string[] = []; + const released: Release[] = []; + const solo = replica(undefined, opened, released); + + const runId = (await solo.execute('expense_approval')).runId!; + expect(await solo.listSuspendedRunsDurable()).toEqual(solo.listSuspendedRuns()); + expect(await solo.listSuspendedRunsDurable()).toEqual([ + { runId, flowName: 'expense_approval', nodeId: 'lv1', correlation: 'req_lv1' }, + ]); + }); +});