From b6f86b326bb32aa5bfddde4fba0356909123cf12 Mon Sep 17 00:00:00 2001 From: os-sales Date: Wed, 2 Sep 2026 16:26:00 +0000 Subject: [PATCH 1/2] fix(service-automation): route the three remaining suspended-run readers through the shared loader `cancelRun`, `failAncestors` and `listSuspendedRunsDurable` still preferred the per-process `suspendedRuns` map over the shared `SuspendedRunStore`, which #13617 had already made authoritative for the resume path. Each now takes one answer to "where is this run parked", with the degrading or strict loader chosen per site so the recorded #4632 / #6299 degradation posture is preserved by choice rather than re-derived. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .changeset/olive-parrots-attend.md | 11 + .../services/service-automation/src/engine.ts | 160 ++++--- ...ulti-replica-suspended-run-readers.test.ts | 433 ++++++++++++++++++ 3 files changed, 551 insertions(+), 53 deletions(-) create mode 100644 .changeset/olive-parrots-attend.md create mode 100644 packages/services/service-automation/src/multi-replica-suspended-run-readers.test.ts 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..7535ddb098 --- /dev/null +++ b/packages/services/service-automation/src/multi-replica-suspended-run-readers.test.ts @@ -0,0 +1,433 @@ +// 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 { AutomationEngine } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import type { AutomationContext, 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' }, + ]); + }); +}); From 89f5f8d15dacac7b65cd1198ff48339d5bd36a18 Mon Sep 17 00:00:00 2001 From: os-sales Date: Wed, 2 Sep 2026 17:04:51 +0000 Subject: [PATCH 2/2] test(service-automation): import AutomationContext from its declaring module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `AutomationContext` is declared in `@objectstack/spec/contracts` and re-imported by `engine.ts` as a type — it is not re-exported from there, so importing it from `./engine.js` added a fourth error (TS2459) to a package whose type-check debt ledger is frozen at 3. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01AUF1NoViznQK32gqpK8wS8 --- .../src/multi-replica-suspended-run-readers.test.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 index 7535ddb098..a0febf6bf8 100644 --- 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 @@ -62,9 +62,10 @@ 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 { AutomationContext, SuspendedRun, SuspendedRunStore } from './engine.js'; +import type { SuspendedRun, SuspendedRunStore } from './engine.js'; function silentLogger(): any { return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } };