diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 57675e1fc1..21892e2381 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -2034,6 +2034,22 @@ export class AutomationEngine implements IAutomationService { * nobody read, and every in-flight approval zombified by the next restart. */ private async persistSuspendedRun(run: SuspendedRun): Promise { + // [#16129] THE MAP WRITE IS FIRST, and `cacheOnlySuspensions` is written + // only after the save below settles => for the whole duration of that + // await this entry is in the map and is NOT yet qualified. A concurrent + // per-id `loadSuspendedRunStrict` therefore reads a store that + // truthfully has no row yet, finds no qualifier, and takes + // `evictConsumedSuspension` on a run being parked RIGHT NOW. Reachable + // without out-of-band knowledge of the id, because the map write is what + // publishes it to `listSuspendedRuns`. + // + // Bounded, measured, and pinned in + // `suspended-run-mid-park-eviction-window.test.ts`: the store-first + // strict load keeps the run resumable once the save lands, and the cost + // falls entirely on `listSuspendedRuns`, which merely OMITS the run -- + // inside that listing's declared latitude. Do not widen the marking, add + // a lock, or move the save above this line without reading that pin's + // header: it also records the ONE compound case that escapes the bounds. this.suspendedRuns.set(run.runId, run); if (this.store) { try { @@ -2220,6 +2236,15 @@ export class AutomationEngine implements IAutomationService { * costs cross-restart durability, not in-process resumability — into a * run that vanishes from its own process. * + * [#16129] Neither guard covers the MID-PARK WINDOW: `persistSuspendedRun` + * writes its map entry BEFORE it awaits the durable save, so an entry can be + * live here while the store legitimately has no row for it and the + * cache-only qualifier is not yet set. Evicting it is bounded -- the run + * stays resumable through the store-first strict load and only the + * cache-only listing under-reports -- and + * `suspended-run-mid-park-eviction-window.test.ts` pins both the window and + * the one compound case that escapes those bounds. + * * 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. diff --git a/packages/services/service-automation/src/suspended-run-mid-park-eviction-window.test.ts b/packages/services/service-automation/src/suspended-run-mid-park-eviction-window.test.ts new file mode 100644 index 0000000000..6b8e5f3919 --- /dev/null +++ b/packages/services/service-automation/src/suspended-run-mid-park-eviction-window.test.ts @@ -0,0 +1,336 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#16129] The MID-PARK WINDOW in {@link AutomationEngine.persistSuspendedRun}: + * a concurrent per-id read can evict a LIVE map entry while the durable save is + * still in flight. + * + * ## Why this file exists at all + * + * This is not a contract defect and must not be read as one. It is a real but + * BOUNDED limit that was, until this file, undocumented and unpinned — and an + * unpinned limit becomes folklore: the next reader cannot tell a deliberate + * boundary from an oversight. Pinning it makes the boundary EXECUTABLE. A + * comment saying the same thing is a claim that drifts away from the code; a + * test that drives the interleaving cannot. + * + * ## The window, as measured on this head (not as inherited from the card) + * + * `persistSuspendedRun` writes the map entry FIRST and marks the run cache-only + * LAST, and only on the failure path: + * + * 1. `this.suspendedRuns.set(run.runId, run)` + * 2. `await this.store.save(run)` <- the window is this await + * 3. on success: `cacheOnlySuspensions.delete(runId)` + * on failure: `cacheOnlySuspensions.add(runId)` + * + * Between 1 and the resolution of 2 the entry is in the map and is NOT yet in + * {@link AutomationEngine.cacheOnlySuspensions}. A concurrent + * `loadSuspendedRunStrict` for that same id therefore reads a store that + * truthfully answers "no row" (the save has not landed), finds no cache-only + * qualifier, and takes the eviction path #16031 added — deleting an entry for a + * run that is being parked right now. + * + * ## Reachability — measured, because the card recorded a reading and not a + * measurement + * + * REACHABLE, on an ordinary single-process composition, and with no + * out-of-band knowledge of the run id. The map write happens first, so + * `listSuspendedRuns()` PUBLISHES the id during the window: the very + * list-then-open consumer #16031 was written for can obtain the id and issue + * the per-id read without the run having been handed to anyone yet. The second + * test drives the same window on the RE-suspend path, where the id has been + * public since the first park, so the reachability does not rest on the listing + * either. Both need only a store whose `save` is asynchronous — that is every + * real store. + * + * ## The two bounds this pin exists to keep standing + * + * - {@link AutomationEngine.loadSuspendedRunStrict} is STORE-FIRST while a + * store is attached, so once the save lands the run is resumable from the + * store. The evicted entry was a cache, not the authority. + * - {@link AutomationEngine.listSuspendedRuns} merely OMITS the run. + * Under-reporting is already inside that method's declared latitude (its own + * docblock says it omits runs parked in a previous process lifetime); + * over-reporting never was, which is the asymmetry #16031 rests on. + * + * ⛔ So this file does NOT widen the cache-only marking, add a lock, or move the + * save before the map write. It pins the outcome those two bounds promise, and + * a future change that alters the trade — in either direction — has to come + * through here and say so. + * + * ## One measured case that does NOT stay inside those bounds + * + * `FINDING` below. Bound 1 holds only because the save eventually LANDS. Let the + * save FAIL after an evicting read has already run, and the compound outcome is + * a run with no durable row and no map entry: unresumable, and the engine's own + * `error` record for the failed save promises the opposite ("it is kept in + * memory only"). It is narrower than the base window — it needs a store that + * rejects the write while still answering reads with "no row" rather than + * throwing (a healthy read replica behind a broken write path, a missing INSERT + * grant, a full disk) — but it is not hypothetical, and it escapes the bound. + * + * ⛔ It is deliberately NOT fixed here. Widening the cache-only marking is + * exactly the move this card forbids taking unilaterally, and the choice + * between that, a lock, and reordering the save is a decision above it. The + * case is pinned at its MEASURED behaviour so the cost is visible and so any + * future fix has a red test to turn green. + */ + +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 levels, so a re-suspend has somewhere to go. */ +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; + +function engineOver(store: SuspendedRunStore | undefined): 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) { + 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); + +/** + * A store whose `save` PARKS INSIDE THE WINDOW. `entered` resolves with the run + * being saved the first time `save` is called — that is the instant between the + * map write and the save landing — and nothing proceeds until `release()`. + * + * Every later `save` passes straight through the already-resolved gate, so a + * re-suspend after the window is an ordinary park. + */ +function gatedSaveStore( + inner: SuspendedRunStore, + opts: { failSave?: boolean; loadThrows?: boolean } = {}, +): { store: SuspendedRunStore; entered: Promise; release: () => void } { + let announce!: (run: SuspendedRun) => void; + const entered = new Promise(r => { announce = r; }); + let open!: () => void; + const gate = new Promise(r => { open = r; }); + const store: SuspendedRunStore = { + async save(run: SuspendedRun) { + announce(run); + await gate; + if (opts.failSave) throw new Error('sqlite: attempt to write a readonly database'); + return inner.save(run); + }, + async load(id: string) { + if (opts.loadThrows) throw new Error('sqlite: database is locked'); + return inner.load(id); + }, + delete: (id: string) => inner.delete(id), + list: () => inner.list(), + }; + return { store, entered, release: () => open() }; +} + +// -- the window, and the bounds it stays inside ------------------------------- + +describe('#16129 — the mid-park window between the map write and the durable save', () => { + it('THE WINDOW: a per-id read taken mid-park evicts a LIVE entry, and the listing hands out the id to do it with', async () => { + const inner = new InMemorySuspendedRunStore(); + const { store, entered, release } = gatedSaveStore(inner); + const engine = engineOver(store); + + const parking = engine.execute('expense_approval'); // deliberately not awaited + const parked = await entered; // now INSIDE the window + const runId = parked.runId; + + // Reachability without out-of-band knowledge of the id: the map write is + // first, so the cache-only listing publishes the run mid-park... + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + // ...while the store truthfully has no row for it yet. + expect(await inner.load(runId)).toBeNull(); + + // The per-id read. The store answers "no row", the run is not cache-only, + // so #16031's eviction path deletes an entry for a run being parked NOW. + expect(await engine.hasSuspendedRun(runId)).toBe(false); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual([]); + + release(); + expect((await parking).runId).toBe(runId); + + // BOUND 1 — store-first: the run is resumable. The evicted entry was a + // cache; the authority is the row that has now landed. + expect(await inner.load(runId)).not.toBeNull(); + expect(await engine.hasSuspendedRun(runId)).toBe(true); + expect(await engine.getSuspendedScreen(runId)).not.toBeUndefined(); + + // BOUND 2 — the cost is confined to the cache-only listing, which OMITS the + // run. Under-reporting is inside its declared latitude. + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual([]); + // The durable listing is unaffected: it reads the store. + expect(listedNodes(await engine.listSuspendedRunsDurable(), runId)).toEqual(['lv1']); + + // Resumable END TO END, not merely answering `true` — and the next park + // re-seeds the map, so the omission lasts one park, not forever. + expect((await approve(engine, runId)).status).toBe('paused'); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv2']); + expect((await approve(engine, runId)).success).toBe(true); + }); + + it('THE WINDOW (re-suspend): the same eviction on a run whose id has been public since the first park', async () => { + // The reachability here rests on nothing at all: an operator holding the id + // from the first park issues an ordinary read while the SECOND park's save + // is in flight. `claimAdvance` has already removed the durable row, so the + // store's "no row" is again truthful and again not the whole truth. + const inner = new InMemorySuspendedRunStore(); + const first = new InMemorySuspendedRunStore(); + const engine = engineOver({ + save: (run: SuspendedRun) => first.save(run), + load: (id: string) => first.load(id), + delete: (id: string) => first.delete(id), + list: () => first.list(), + }); + const runId = (await engine.execute('expense_approval')).runId!; + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + + // Swap in the gated store over the same rows, then resume: lv1 -> lv2 parks + // again, and THAT save is the one that waits. + for (const r of await first.list()) await inner.save(r); + const { store, entered, release } = gatedSaveStore(inner); + engine.setSuspendedRunStore(store); + + const resuming = approve(engine, runId); + const reparked = await entered; + expect(reparked.nodeId).toBe('lv2'); + + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv2']); + expect(await inner.load(runId)).toBeNull(); + expect(await engine.hasSuspendedRun(runId)).toBe(false); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual([]); + + release(); + expect((await resuming).status).toBe('paused'); + + // Same two bounds. + expect(await engine.hasSuspendedRun(runId)).toBe(true); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual([]); + expect(listedNodes(await engine.listSuspendedRunsDurable(), runId)).toEqual(['lv2']); + expect((await approve(engine, runId)).success).toBe(true); + }); +}); + +// -- the one case that escapes the bounds, pinned at its measured behaviour --- + +describe('#16129 — the window compounded with a FAILING save', () => { + it('FINDING: an evicting read inside the window of a save that then fails leaves the run unresumable', async () => { + // ⛔ Deliberately NOT fixed here — see this file's header. Pinned so the + // cost is visible and so a future fix has a red test to turn green. + const inner = new InMemorySuspendedRunStore(); + const { store, entered, release } = gatedSaveStore(inner, { failSave: true }); + const engine = engineOver(store); + + const parking = engine.execute('expense_approval'); + const runId = (await entered).runId; + + // Same window, same evicting read. + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + expect(await engine.hasSuspendedRun(runId)).toBe(false); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual([]); + + // The save now fails. `persistSuspendedRun` marks the run cache-only — but + // the map entry it qualifies is already gone, so the qualifier qualifies + // nothing and the strict loader has nothing left to serve. + release(); + expect((await parking).runId).toBe(runId); + + // ESCAPES BOUND 1. The store never took the row and the cache no longer + // holds it, so the run is unresumable — while the engine's `error` record + // for the failed save says it "is kept in memory only". + expect(await inner.load(runId)).toBeNull(); + expect(await engine.hasSuspendedRun(runId)).toBe(false); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual([]); + const resumed = await approve(engine, runId); + expect(resumed.success).toBe(false); + expect(resumed.code).toBe('RUN_NOT_FOUND'); + + // The control that isolates the window as the cause: WITHOUT the mid-park + // read, the identical failing save is the documented degradation — the run + // stays resumable in-process, which is exactly what the promise says. + const solo = engineOver({ + async save() { throw new Error('sqlite: attempt to write a readonly database'); }, + async load() { return null; }, + async delete() {}, + async list() { return []; }, + }); + const soloRun = (await solo.execute('expense_approval')).runId!; + expect(await solo.hasSuspendedRun(soloRun)).toBe(true); + expect(listedNodes(solo.listSuspendedRuns(), soloRun)).toEqual(['lv1']); + }); +}); + +// -- controls: the shapes in which the window cannot bite --------------------- + +describe('#16129 — where the window does not exist', () => { + it('CONTROL: with no store attached there is no window and nothing is evicted', async () => { + // `persistSuspendedRun` awaits nothing, and `evictConsumedSuspension` + // refuses to act because the map IS the authority. + const engine = engineOver(undefined); + const runId = (await engine.execute('expense_approval')).runId!; + + expect(await engine.hasSuspendedRun(runId)).toBe(true); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + expect((await approve(engine, runId)).status).toBe('paused'); + }); + + it('CONTROL: a read that THROWS inside the window evicts nothing — unknown is not "gone"', async () => { + // The store-outage guard covers the window too: an unreadable store makes + // the run's existence UNKNOWN, and the strict read throws rather than + // reaching the eviction. + const inner = new InMemorySuspendedRunStore(); + const { store, entered, release } = gatedSaveStore(inner, { loadThrows: true }); + const engine = engineOver(store); + + const parking = engine.execute('expense_approval'); + const runId = (await entered).runId; + + await expect(engine.hasSuspendedRun(runId)).rejects.toThrow(/database is locked/); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + + release(); + expect((await parking).runId).toBe(runId); + expect(listedNodes(engine.listSuspendedRuns(), runId)).toEqual(['lv1']); + }); +});