diff --git a/.changeset/suspended-run-cache-consumed-elsewhere.md b/.changeset/suspended-run-cache-consumed-elsewhere.md new file mode 100644 index 0000000000..1475441cb7 --- /dev/null +++ b/.changeset/suspended-run-cache-consumed-elsewhere.md @@ -0,0 +1,64 @@ +--- +"@objectstack/service-automation": patch +--- + +fix(service-automation): evict a suspension consumed by another replica, so the run listings stop reporting phantoms (#15832) + +`AutomationEngine` had exactly one eviction site for its `suspendedRuns` +map, inside `forgetSuspendedRun` — and that runs in whichever process +**consumes** the suspension. In a multi-replica deployment that is routinely +not the process that parked it: replica A parks a run, replica B resumes it, +and nothing ever removes A's entry. There is no invalidation channel from B +to A. + +The card that found this located the leak on `resumeInternal`'s +`claim.kind === 'lost'` branch, which returns before that choke point. That +branch does leak, but it is not the common shape: the **no-race** variant +leaks identically — A parks, only B ever resumes, A never attempts a claim +and there is no `'lost'` anywhere in the sequence — so an eviction hung on +`'lost'` alone would have left the ordinary deployment untouched. + +The retained snapshot was **not only memory**. Two readers handed it back: +`listSuspendedRuns()` (synchronous, cache-only, and the one listing on the +`AutomationService` spec contract) and `listSuspendedRunsDurable()` (which +deliberately appends map entries the durable list lacks). Once the other +replica **completed** the run, both reported a phantom — a finished run +listed as suspended, whose `getSuspendedScreen()` answers `null`, so a +consumer that listed and then opened got an entry it could not act on. + +An entry is now dropped whenever this process holds a store-authoritative, +per-id "no row" answer for it: the strict loader's store miss (which reaches +`resume`, `hasSuspendedRun`, `cancelRun` and `getSuspendedScreen`), a lost +advance claim, and a bounded per-id reconcile for the map-only entries of +`listSuspendedRunsDurable()`. + +**Nothing here moves the cache-only listing's contract.** The fix only ever +*removes* entries. The spec says `listSuspendedRuns()` lists "the currently +suspended (paused) runs awaiting a resume"; the engine's own docblock adds +only that it may OMIT runs (those parked in a previous process lifetime), +because it reads the cache alone. Under-reporting is therefore already +inside the declared latitude, and over-reporting was never inside the +promise. Neither listing becomes store-backed, and `listSuspendedRuns()` +stays synchronous. + +Three shapes are deliberately **never** evicted, each pinned by a control: +no store attached (the map IS the authority); a run whose durable save +failed (`cacheOnlySuspensions` — the store was never handed the row, so its +silence says nothing about it); and a store read that THROWS (an outage +means the run's existence is unknown, not gone). A failed `list()` +enumeration likewise triggers no per-id reconcile — during an outage that +would ask about every live run in the process. + +**Residual, stated rather than implied.** Eviction is demand-driven: a +phantom is cleared when this process next obtains the per-id answer for that +run — any `resume` / `hasSuspendedRun` / `getSuspendedScreen`, or a +`listSuspendedRunsDurable()` reconcile. A process that never looks at the +run again keeps the entry until it does. With no invalidation channel +between replicas, closing that last gap needs either a background sweep or a +store-backed listing, and both are decisions above this change; the boundary +is pinned by a `RESIDUAL` test rather than left to be discovered. + +Note 2 of the same card — the `'unsupported'` branch deciding on the shape of +a value the conditional delete has **already** been issued to obtain — is +**not** addressed here: its honest fix is a declared return contract for the +engine's multi-row delete, which lands in another package. diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 9c749b02f5..57675e1fc1 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -2164,6 +2164,72 @@ export class AutomationEngine implements IAutomationService { await this.releaseSuspension(run, reason); } + /** + * [#15832] Drop a map entry for a run this process has a store-authoritative + * per-id "no row" answer for — a run parked HERE and consumed by another + * replica. + * + * ## Why this exists at all, and why not on the `'lost'` branch + * + * {@link forgetSuspendedRun} is the one eviction site, and it runs in + * whichever process CONSUMES the suspension. Put two replicas over one + * store and the parking process is routinely not that one: A parks a run + * and B resumes it, so A's entry is never removed by anything. The card + * that found this located the leak on `resumeInternal`'s `'lost'` branch, + * which returns before that choke point — but the NO-RACE shape leaks + * identically (A parks, only B ever resumes, A never attempts a claim and + * there is no `'lost'` at all), so an eviction hung on `'lost'` alone would + * leave the ordinary multi-replica deployment untouched. + * + * The retained snapshot is not only memory. Two readers hand it back: + * {@link listSuspendedRuns} — synchronous, cache-only, and the one listing + * on the `AutomationService` spec contract — and + * {@link listSuspendedRunsDurable}, which deliberately appends map entries + * the durable list lacks. After the other replica COMPLETES the run both + * report a phantom: a finished run listed as suspended, whose + * {@link getSuspendedScreen} answers `null`, so a consumer that lists and + * then opens gets an entry it cannot act on. + * + * ## What this does NOT do + * + * ⛔ It does not touch the cache-only listing's contract. The spec says + * `listSuspendedRuns()` lists "the currently suspended (paused) runs + * awaiting a resume"; this engine's own docblock adds only that it may + * OMIT runs (those parked in a previous process lifetime) because it reads + * the cache alone. Removing an entry therefore moves nothing: under- + * reporting is already inside that declared latitude, and over-reporting + * was never inside the promise. Nothing here makes either listing + * store-backed. + * + * ⛔ It does not notify the paused node's executor + * ({@link NodeExecutor.onSuspensionReleased}). That notification belongs to + * {@link forgetSuspendedRun} because it is the choke point every + * CONSUMPTION passes through, and an eviction is not a consumption — this + * process consumed nothing, the replica that did fired its own. Firing one + * here would tear down a pause twice, once per replica. + * + * ## The two guards, and why each is load-bearing + * + * - **no store** — the map IS the authority (`loadSuspendedRunStrict` + * returns from it directly), so there is no second reader to be wrong + * about and nothing may be dropped. + * - **{@link cacheOnlySuspensions}** — a run whose durable save failed was + * never handed to the store, so the store's "no row" is SILENCE about it + * rather than an answer (#13617). Evicting on that would convert + * {@link persistSuspendedRun}'s documented degradation — a failed save + * costs cross-restart durability, not in-process resumability — into a + * run that vanishes from its own process. + * + * A store read that THROWS must never reach here: an outage means the + * run's existence is UNKNOWN, not "gone". Every caller below is on a path + * where the store answered. + */ + private evictConsumedSuspension(runId: string): void { + if (!this.store) return; + if (this.cacheOnlySuspensions.has(runId)) return; + this.suspendedRuns.delete(runId); + } + /** * [#14333] Claim the right to advance this run past the node it is parked * at — the CROSS-REPLICA half of the resume idempotency guard. @@ -4996,6 +5062,13 @@ export class AutomationEngine implements IAutomationService { // 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; + // [#15832] The store ANSWERED, and the answer is "no row". Any entry + // this process still holds for that run is a run it parked and another + // replica consumed — the phantom the two listings hand back. This is + // the definitive per-id evidence the paragraph above already rests on, + // so the same reading that refuses to serve it here stops publishing it + // there. A store read that threw never reaches this line. + this.evictConsumedSuspension(runId); return null; } @@ -5374,6 +5447,14 @@ export class AutomationEngine implements IAutomationService { // already maps it to 409. A distinct code would be vocabulary // nothing reads — add one the day a caller needs the // difference. + // [#15832] The store's compare-and-set ANSWERED: no row is parked + // where this replica read it. Whatever snapshot this process + // still holds for the run is stale by construction — it names a + // node the run has left — so it stops being published by the two + // listings. ⛔ NOT `forgetSuspendedRun`: nothing was consumed + // here, and firing that choke point would tear the pause down a + // second time in this process on top of the winner's own. + this.evictConsumedSuspension(runId); return { success: false, code: 'RESUME_IN_PROGRESS', @@ -6536,11 +6617,16 @@ export class AutomationEngine implements IAutomationService { */ async listSuspendedRunsDurable(): Promise> { const byId = new Map(); + // [#15832] Did the ENUMERATION answer? The reconcile below is allowed + // only when it did — see the merge comment for why a failed listing is + // silence rather than evidence. + let enumerated = false; if (this.store) { try { for (const r of await this.store.list()) { byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation }); } + enumerated = true; } catch (err) { // #6299 — driver text to the structured slot, message one line, // same as the two seams above. The SLOT differs: the `Logger` @@ -6605,8 +6691,35 @@ export class AutomationEngine implements IAutomationService { // 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()) { + // + // [#15832] What that reasoning leaves open is a run this process parked + // and ANOTHER replica has since consumed: absent from the durable list + // because it is finished, appended here, and published as suspended by + // a listing that also backs the cache-only one. The paragraph above is + // right that list-absence is not evidence — so this asks for the + // evidence instead. `store.load` is the same definitive per-id read + // {@link loadSuspendedRunStrict} rests on, and it is bought only for the + // entries that look suspicious: a healthy process, whose map entries all + // appear in the durable list, buys none. A read that THROWS leaves the + // entry standing (unknown is not gone), and a store that could not be + // enumerated at all is not probed row by row — an outage would answer + // for every live run in the process. + for (const r of [...this.suspendedRuns.values()]) { if (byId.has(r.runId)) continue; + if (enumerated && !this.cacheOnlySuspensions.has(r.runId)) { + let stored: SuspendedRun | null; + try { + stored = await this.store!.load(r.runId); + } catch { + // Unknown, not gone — keep the entry and publish it, exactly + // as this method did before the reconcile existed. + stored = r; + } + if (stored === null) { + this.evictConsumedSuspension(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/suspended-run-cache-eviction.test.ts b/packages/services/service-automation/src/suspended-run-cache-eviction.test.ts new file mode 100644 index 0000000000..711b48d9f6 --- /dev/null +++ b/packages/services/service-automation/src/suspended-run-cache-eviction.test.ts @@ -0,0 +1,287 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * A run parked by one process and consumed by another leaves the parking + * process's `suspendedRuns` map holding it forever (#15832 note 1). + * + * ## The defect, located where the correction measured it and not where the + * card first pointed + * + * The card put the leak on `resumeInternal`'s `claim.kind === 'lost'` branch, + * which returns before the one `suspendedRuns.delete` in `forgetSuspendedRun`. + * That branch does leak — but it is not where the common shape lives. The + * NO-RACE variant leaks identically: replica A parks a run, replica B is the + * only one that ever resumes it, A never attempts a claim and there is no + * `'lost'` at all. So the leak is a property of ANY run parked by one process + * and consumed by another — the ordinary multi-replica deployment — and an + * eviction hung on `'lost'` would leave that shape untouched. + * + * ## Why it is not only a memory leak + * + * The card's severity note said "the stale entry is not read back", which is + * false: two readers hand it back. + * + * - {@link AutomationEngine.listSuspendedRuns} — synchronous, cache-only, and + * the one listing on the `AutomationService` SPEC contract, where it is + * documented as "the currently suspended (paused) runs awaiting a resume". + * - {@link AutomationEngine.listSuspendedRunsDurable} — which deliberately + * APPENDS map entries the durable list lacks. + * + * After B **completes** the run the durable row is gone, and both listings + * still report it: a phantom whose `getSuspendedScreen()` answers `null`, so a + * consumer that lists and then opens gets an entry it cannot act on. + * + * ## The promise this file's fix relies on — stated, because it is the whole + * question + * + * The fix only ever REMOVES map entries, and only ones the process has a + * store-authoritative, per-id "no row" answer for. It relies on exactly one + * promise, in the direction that promise already runs: + * + * - the SPEC contract says `listSuspendedRuns()` lists "the currently + * suspended (paused) runs awaiting a resume" — a completed run is not one; + * - the engine's own docblock adds only that it may OMIT runs (those parked in + * a previous process lifetime), because it reads the cache alone. + * + * So under-reporting is already inside the method's declared latitude and + * over-reporting was never inside its promise. ⛔ Nothing here makes the + * synchronous listing store-backed, and nothing widens what it returns. + * + * ## The residual, pinned rather than described + * + * A phantom is evicted when this process obtains the per-id answer — any + * `hasSuspendedRun` / `resume` / `getSuspendedScreen` on that run, or a + * `listSuspendedRunsDurable()` reconcile. A process that never looks at the run + * again keeps the entry: with no invalidation channel from B to A, the only + * remedies for THAT are a background sweep or a store-backed listing, and both + * are decisions above this card. `RESIDUAL` below pins the boundary so it + * cannot be mistaken for a fix. + */ + +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(); } }; +} + +/** start -> lv1 -> lv2 -> end. Two approval levels is all the shape needs. */ +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: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'lv1' }, + { id: 'e2', source: 'lv1', target: 'lv2' }, + { id: 'e3', source: 'lv2', target: 'end' }, + ], +} as any; + +/** One replica: a fresh engine over the SHARED store. */ +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: any) { + opened.push(node.id); + return { success: true, suspend: true, correlation: `req_${node.id}` }; + }, + } as any); + engine.registerFlow('expense_approval', APPROVAL_FLOW); + return engine; +} + +const approve = (engine: AutomationEngine, runId: string) => + engine.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any); + +/** Node ids the listing reports for `runId`, in call order. */ +const listedNodes = (rows: Array<{ runId: string; nodeId: string }>, runId: string) => + rows.filter(r => r.runId === runId).map(r => r.nodeId); + +// ── the leak, both variants ───────────────────────────────────────────────── + +describe('#15832 note 1 — a run parked here and consumed elsewhere leaves no phantom', () => { + it('THE BUG (no race): A parks, B alone runs it to completion, A must list nothing', async () => { + // The ordinary multi-replica shape: A never attempts a claim, so there is + // no `'lost'` anywhere in this sequence. + 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!; + expect(opened).toEqual(['lv1']); + expect((await approve(b, runId)).status).toBe('paused'); // lv1 -> lv2, on B + expect((await approve(b, runId)).success).toBe(true); // lv2 -> end, run COMPLETES + expect(await store.load(runId)).toBeNull(); // durable row is gone + + // Reader 1: the durable listing must not append a completed run. + expect(listedNodes(await a.listSuspendedRunsDurable(), runId)).toEqual([]); + // Reader 2: the spec-contract listing, once the reconcile above has run. + expect(listedNodes(a.listSuspendedRuns(), runId)).toEqual([]); + }); + + it('THE BUG (race): the loser of the advance claim drops its stale entry', async () => { + // A reads the run parked at lv2, then blocks INSIDE claimSuspension while B + // consumes it — so A's compare-and-set runs against a row that is gone and + // answers `'lost'`. That is the branch the card named. + 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!; + expect((await approve(b, runId)).status).toBe('paused'); // parked at lv2 + + let releaseClaim: () => void = () => {}; + const gate = new Promise(r => { releaseClaim = r; }); + a.setSuspendedRunStore({ + save: (run: SuspendedRun) => store.save(run), + load: (id: string) => store.load(id), + delete: (id: string) => store.delete(id), + list: () => store.list(), + async claimSuspension(id: string, at: any) { + await gate; + return store.claimSuspension(id, at); + }, + }); + + const losing = approve(a, runId); + expect((await approve(b, runId)).success).toBe(true); // B finishes the run + releaseClaim(); + const lost = await losing; + expect(lost.success).toBe(false); + expect(lost.code).toBe('RESUME_IN_PROGRESS'); + + expect(listedNodes(a.listSuspendedRuns(), runId)).toEqual([]); + }); + + it('list-then-open: the per-id read a consumer makes next evicts the phantom', async () => { + // The harm the correction named — list, then open — is also the channel + // that heals it: `hasSuspendedRun` is store-authoritative per id. + const store = new InMemorySuspendedRunStore(); + const a = replica(store); + const b = replica(store); + + const runId = (await a.execute('expense_approval')).runId!; + expect((await approve(b, runId)).status).toBe('paused'); + expect((await approve(b, runId)).success).toBe(true); + + expect(await a.hasSuspendedRun(runId)).toBe(false); + expect(listedNodes(a.listSuspendedRuns(), runId)).toEqual([]); + expect(await a.getSuspendedScreen(runId)).toBeNull(); + }); + + it('RESIDUAL: with no store-authoritative read of its own, A still holds the entry', async () => { + // The declared boundary, pinned so it is not mistaken for a fix: eviction + // is demand-driven. Closing THIS requires a background sweep or a + // store-backed listing — both above this card. + const store = new InMemorySuspendedRunStore(); + const a = replica(store); + const b = replica(store); + + const runId = (await a.execute('expense_approval')).runId!; + expect((await approve(b, runId)).status).toBe('paused'); + expect((await approve(b, runId)).success).toBe(true); + + // No `listSuspendedRunsDurable()`, no `hasSuspendedRun`, no resume on A. + expect(listedNodes(a.listSuspendedRuns(), runId)).toEqual(['lv1']); + }); +}); + +// ── controls: the three shapes eviction must NEVER touch ──────────────────── + +describe('#15832 note 1 — eviction fires only on a store-authoritative per-id miss', () => { + it('CONTROL: with no store attached the map IS the authority and nothing is evicted', async () => { + const solo = replica(undefined); + const runId = (await solo.execute('expense_approval')).runId!; + + expect(await solo.hasSuspendedRun(runId)).toBe(true); + expect(listedNodes(await solo.listSuspendedRunsDurable(), runId)).toEqual(['lv1']); + expect(listedNodes(solo.listSuspendedRuns(), runId)).toEqual(['lv1']); + }); + + it('CONTROL: a run the store never accepted (cache-only) is never evicted', async () => { + // `persistSuspendedRun`'s documented degradation: a failed durable save + // costs cross-restart durability, not in-process resumability. The store's + // "no row" is SILENCE about this run, not an answer. + 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); + const runId = (await engine.execute('expense_approval')).runId!; + expect(saves).toEqual(['lv1']); + + expect(await engine.hasSuspendedRun(runId)).toBe(true); + expect(listedNodes(await engine.listSuspendedRunsDurable(), runId)).toEqual(['lv1']); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + }); + + it('CONTROL: an unreadable store is "unknown", not a licence to evict', async () => { + const store = new InMemorySuspendedRunStore(); + const engine = replica(store); + const runId = (await engine.execute('expense_approval')).runId!; + + engine.setSuspendedRunStore({ + save: (run: SuspendedRun) => store.save(run), + async load() { throw new Error('sqlite: database is locked'); }, + delete: (id: string) => store.delete(id), + list: () => store.list(), + }); + + // The strict read THROWS rather than answering "no row" — the entry stands. + await expect(engine.hasSuspendedRun(runId)).rejects.toThrow(/database is locked/); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + }); + + it('CONTROL: an unlistable store degrades the listing and evicts nothing', async () => { + // `listSuspendedRunsDurable`'s documented degraded path: `byId` is empty + // because the ENUMERATION failed, not because the rows are gone. A + // reconcile here would evict every live run in the process. + const store = new InMemorySuspendedRunStore(); + const engine = replica(store); + const runId = (await engine.execute('expense_approval')).runId!; + + engine.setSuspendedRunStore({ + save: (run: SuspendedRun) => store.save(run), + load: (id: string) => store.load(id), + delete: (id: string) => store.delete(id), + async list() { throw new Error('sqlite: database is locked'); }, + }); + + expect(listedNodes(await engine.listSuspendedRunsDurable(), runId)).toEqual(['lv1']); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + }); + + it('CONTROL: a live run parked in THIS process survives every reconcile', async () => { + const store = new InMemorySuspendedRunStore(); + const engine = replica(store); + const runId = (await engine.execute('expense_approval')).runId!; + + expect(listedNodes(await engine.listSuspendedRunsDurable(), runId)).toEqual(['lv1']); + expect(await engine.hasSuspendedRun(runId)).toBe(true); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + expect((await approve(engine, runId)).status).toBe('paused'); + }); +});