From c4e208a4554b6c9a1075ef6f50b4c9b77eb0e33b Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 31 Oct 2026 03:14:51 +0000 Subject: [PATCH] fix(plugin-approvals): re-issuing a restored continuation proves the pause past the first 1000 parked runs GUARD 3 of continueRestoredRun read where the run is parked from listSuspendedRunsDurable(), one 1000-row page of a DB-backed store plus what this process parked itself. A run another process parked behind more than 1000 others was on no page, so every re-issue was refused with "parked at no node this engine can see". It now asks the engine for that one run: the new AutomationEngine.suspensionOf(runId) is one keyed, store-authoritative read, declared on ApprovalResumeSurface as an optional, fail-closed member. Co-authored-by: Claude Claude-Session: https://claude.ai/code/session_016WV4czGci1wfPnGZhkvrcF --- ...1-restored-continuation-past-first-page.md | 14 ++ .../plugin-approvals/src/approval-service.ts | 60 ++++--- .../restored-approval-continuation.test.ts | 156 +++++++++++++++++- .../services/service-automation/src/engine.ts | 20 +++ .../src/suspension-of.test.ts | 95 +++++++++++ 5 files changed, 322 insertions(+), 23 deletions(-) create mode 100644 .changeset/28271-restored-continuation-past-first-page.md create mode 100644 packages/services/service-automation/src/suspension-of.test.ts diff --git a/.changeset/28271-restored-continuation-past-first-page.md b/.changeset/28271-restored-continuation-past-first-page.md new file mode 100644 index 00000000000..8279e8c09a4 --- /dev/null +++ b/.changeset/28271-restored-continuation-past-first-page.md @@ -0,0 +1,14 @@ +--- +'@objectstack/plugin-approvals': patch +'@objectstack/service-automation': patch +--- + +fix(plugin-approvals): re-issuing a restored approval continuation proves the pause past the first 1000 parked runs + +Clause-②: no (a re-issue on a store with more than 1000 parked runs now proves the pause as it already does below that; no schema, accept set, error code or result field changes) +Metadata-protocol: none + +`continueRestoredRun` proves the pause it continues is the one the request's outcome was issued at, by reading where the run is parked. It read that from `listSuspendedRunsDurable()`. On a database-backed store that listing reads one page of 1000 `paused` rows, and a process adds only the runs it parked itself. So when another replica (or this process before a restart) parked the run and more than 1000 runs were parked ahead of it, the run was on no page. Every re-issue was refused with `INVALID_STATE` "parked at no node this engine can see", every retry read the same page, and the recorded outcome never took effect. + +- **Now.** The proof asks the engine for that one run. The new `AutomationEngine.suspensionOf(runId)` reads the run's suspension by id, through the same strict, store-authoritative read `hasSuspendedRun` uses, and answers its flow, node and correlation, or `undefined` when it is not suspended. It rejects when the store cannot be read. +- **Unchanged.** Below 1000 parked runs the re-issue proceeds as before. A run that is not parked, or is parked at another node, is still refused, and the refusal still names the node. An engine without the read is still refused; the refusal now names `suspensionOf` as the missing read. diff --git a/packages/plugins/plugin-approvals/src/approval-service.ts b/packages/plugins/plugin-approvals/src/approval-service.ts index 0cdf9f83890..3e15f7668af 100644 --- a/packages/plugins/plugin-approvals/src/approval-service.ts +++ b/packages/plugins/plugin-approvals/src/approval-service.ts @@ -270,24 +270,36 @@ export interface ApprovalResumeSurface { */ unresumableSuspensionCause?(runId: string): Promise; /** - * [#15389] Where each suspended run is currently parked. Read by - * {@link ApprovalService.continueRestoredRun} to prove the pause it was asked - * to continue is the one THIS request's recorded outcome was issued at, and - * not merely some live pause on the same run — `hasSuspendedRun` answers a - * boolean and cannot tell the two apart. Which node that is depends on the signal, not - * only on the row: see {@link ApprovalService.expectedPauseNode}. + * [#15389] Where each suspended run is currently parked, durable-first. * * ⚠️ Declares a method `AutomationEngine` ALREADY implements publicly - * (`listSuspendedRunsDurable`); it widens no engine surface. Durable-first, - * so it sees a pause parked by another replica or before a restart. + * (`listSuspendedRunsDurable`); it widens no engine surface. [#28271] + * {@link ApprovalService.continueRestoredRun} no longer reads it: on a + * DB-backed store it is one capped page, and it now asks + * {@link suspensionOf} for the one run instead. + */ + listSuspendedRunsDurable?(): Promise>; + /** + * [#15389, #28271] Where the run `runId` is parked, or `undefined` when it + * is not suspended. Read by {@link ApprovalService.continueRestoredRun} to + * prove the pause it was asked to continue is the one THIS request's + * recorded outcome was issued at, and not merely some live pause on the + * same run — `hasSuspendedRun` answers a boolean and cannot tell the two + * apart. Which node that is depends on the signal, not only on the row: + * see {@link ApprovalService.expectedPauseNode}. + * + * ⚠️ Declares the method `AutomationEngine` implements publicly + * (`suspensionOf`): one keyed read of the store, so it sees a pause parked + * by another replica or before a restart however many runs are parked + * ahead of it — where {@link listSuspendedRunsDurable}, which this proof + * read until #28271, is one capped page of a DB-backed store. Read-only; + * rejects when the store cannot be read. * * Optional, and its absence is FAIL-CLOSED: a caller that cannot prove node * identity refuses the continuation rather than proceeding on the weaker - * check. The engine degrades a store outage to its in-memory list rather - * than throwing, and that degradation is safe HERE for the same reason — a - * pause it cannot see is a pause this verb will not act on. + * check. */ - listSuspendedRunsDurable?(): Promise>; + suspensionOf?(runId: string): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined>; /** * [#28230] The run parked at a `map` node on the item run `itemRunId`, or * `undefined` when there is none. Read by @@ -7381,23 +7393,27 @@ export class ApprovalService implements IApprovalService { // node refused the one population the journal was built to serve, with a // message that named a cause the code had not taken. // - // FAIL-CLOSED in every direction: no reader, no listing, no matching entry, - // an entry at another node, or an expected node this service cannot derive - // — all refuse. The engine degrades a store outage to its in-memory list - // instead of throwing, so a pause this cannot see is a pause this verb - // declines to act on. A false refusal costs the operator a retry; a false - // admission advances a flow nobody decided. + // FAIL-CLOSED in every direction: no reader, no pause, a pause at another + // node, or an expected node this service cannot derive — all refuse. A + // store that cannot be read rejects the read, as GUARD 2's does. A false + // refusal costs the operator a retry; a false admission advances a flow + // nobody decided. + // + // [#28271] The read is KEYED by the run id. It used to be the run's entry + // in `listSuspendedRunsDurable()`, which on a DB-backed store is one page + // of 1000 parked runs plus what this process parked itself: a run another + // process parked behind more than 1000 others was on no page, so every + // retry was refused the same way and the recorded outcome never took effect. const expected = await this.expectedPauseNode(raw, signal, requestId, runId); - if (typeof this.automation?.listSuspendedRunsDurable !== 'function') { + if (typeof this.automation?.suspensionOf !== 'function') { throw new Error( `INVALID_STATE: this automation engine cannot report WHERE run '${runId}' is parked ` + - `(no listSuspendedRunsDurable), so the pause cannot be proved to be the one ${signal.what} on ` + + `(no suspensionOf), so the pause cannot be proved to be the one ${signal.what} on ` + `request ${requestId} was refused on — refusing, because continuing the wrong pause advances a ` + `flow with no decision behind it`, ); } - const parkedAt = (await this.automation.listSuspendedRunsDurable()) - .find(r => String(r.runId) === String(runId))?.nodeId; + const parkedAt = (await this.automation.suspensionOf(runId))?.nodeId; if (parkedAt !== expected.nodeId) { throw new Error( `INVALID_STATE: run '${runId}' is parked at ` + diff --git a/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts b/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts index fa9d7183132..fd4690d5a9e 100644 --- a/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts +++ b/packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts @@ -39,7 +39,7 @@ */ import { describe, it, expect, beforeEach } from 'vitest'; -import { AutomationEngine, InMemorySuspendedRunStore, resolveRunDataContext } from '@objectstack/service-automation'; +import { AutomationEngine, InMemorySuspendedRunStore, ObjectStoreSuspendedRunStore, resolveRunDataContext } from '@objectstack/service-automation'; // [#4550] The engine double below routes its write verbs through ObjectQL's OWN // dispatch predicates rather than a hand-mirrored copy. import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql'; @@ -1035,3 +1035,157 @@ describe('ruling A — a decision resumes the run as the person who decided', () expect(seen.after_reject, 'the replay fell back to the trigger user').toEqual({ userId: 'u1', permissions: ['deal_manager'] }); }); }); + +/** + * GUARD 3 proves where the run is parked past the first page of parked runs + * (#28271). + * + * It used to find the run in `listSuspendedRunsDurable()`. A DB-backed store + * lists one 1000-row page of `paused` rows in driver order, and a process adds + * only what it parked itself. So when another process (a replica, or this one + * before a restart) parked the run behind more than 1000 others, the run was + * not on the listing, and every re-issue was refused with "parked at no node + * this engine can see" — the retry the refusal promises read the same page. + * Both replicas below share one store and one table set: `a` strands and + * restores the run, `b` re-issues it. + */ +describe('#28271 — GUARD 3 reads where the run is parked by its id, not from one page', () => { + let data: ReturnType; + let marks: string[]; + let rejectBranchThrows: string | undefined; + let a: ReturnType; + let b: ReturnType; + + /** One process: a real engine over `store`, the approval node, and its own approvals service. */ + function wire(store: ObjectStoreSuspendedRunStore) { + const service = new ApprovalService({ engine: data as any, logger: noopLogger }); + const automation = new AutomationEngine(noopLogger as any, store); + registerApprovalNode(automation, service, noopLogger as any); + automation.registerNodeExecutor({ + type: 'mark', + async execute(node: any) { + if (node.id === 'mark_rejected' && rejectBranchThrows) throw new Error(rejectBranchThrows); + marks.push(node.id); + return { success: true }; + }, + } as never); + automation.registerFlow('deal_approval', DEAL_APPROVAL as never); + automation.registerFlow('revise_flow', REVISE_FLOW as never); + service.attachAutomation(automation); + return { automation, service }; + } + + /** `count` parked runs of some other flow, stored ahead of everything this test parks. */ + const parkOthers = (count: number) => { + const rows = data.tables.get('sys_automation_run') ?? (data.tables.set('sys_automation_run', []), data.tables.get('sys_automation_run')!); + for (let i = 0; i < count; i++) { + rows.push({ id: `other-${String(i).padStart(4, '0')}`, flow_name: 'other_flow', node_id: 'wait', status: 'paused' }); + } + }; + + /** `b`'s engine with one member taken away, everything else delegated. */ + const without = (member: string) => new Proxy(b.automation, { + get(target, prop) { + if (prop === member) return undefined; + const value = Reflect.get(target, prop, target); + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + + beforeEach(() => { + marks = []; rejectBranchThrows = undefined; + data = makeFakeEngine(); + const store = new ObjectStoreSuspendedRunStore(data as any, noopLogger as any); + a = wire(store); + b = wire(store); + }); + + const rowOf = async (id: string) => + (await data.find('sys_approval_request', { where: { id } }))[0]; + + /** On `a`: park, strand a reject decision, and (unless told not to) re-arm its pause. */ + async function strandOnA({ restore = true } = {}) { + await a.automation.execute('deal_approval', { + object: 'crm_deal', record: { id: 'd1', amount: 100 }, userId: 'submitter', + } as never); + const req = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + rejectBranchThrows = DOWNSTREAM_FAILURE; + const err = await a.service + .decide(req.id, { decision: 'reject', actorId: 'u1', comment: 'no' }, SYSTEM_CTX) + .then(() => null, (e: Error) => e); + expect(err?.message).toMatch(/^RESUME_FAILED/); + if (restore) { + expect((await a.automation.restoreConsumedSuspension(req.flow_run_id, { requestedBy: 'ops' })).restored).toBe(true); + } + rejectBranchThrows = undefined; + return req; + } + + it('past the first page, the pause another process re-armed is proved and the decision is re-issued', async () => { + parkOthers(1000); + const req = await strandOnA(); + // The premise: `b` holds nothing in memory, and the run is not on the page. + expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the pause is back, in the store').toBe(true); + expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === req.flow_run_id), + 'and not on the listing\'s one page').toBe(false); + + const out = await b.service.continueRestoredRun(req.id, { requestedBy: 'ops', reason: 'record recreated' }); + expect(out.resumed, JSON.stringify(out)).toBe(true); + expect(out.decision).toBe('reject'); + expect(out.source).toBe('journal'); + expect(marks, 'the reject branch ran').toEqual(['mark_rejected']); + expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the continuation consumed the pause').toBe(false); + }); + + it('CONTROL: within the first page it is re-issued as before', async () => { + parkOthers(998); + const req = await strandOnA(); + expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === req.flow_run_id)).toBe(true); + + const out = await b.service.continueRestoredRun(req.id, { requestedBy: 'ops' }); + expect(out.resumed, JSON.stringify(out)).toBe(true); + expect(marks).toEqual(['mark_rejected']); + }); + + it('CONTROL: past the first page, a run parked at another node is still refused, and the refusal names that node', async () => { + parkOthers(1000); + await a.automation.execute('revise_flow', { + object: 'crm_deal', record: { id: 'd7', amount: 3 }, userId: 'submitter', + } as never); + const rev = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0]; + await a.service.sendBack(rev.id, { actorId: 'u1', comment: 'fix it' } as any, SYSTEM_CTX); + expect((await rowOf(rev.id)).status).toBe('returned'); + + const refused = await b.service.continueRestoredRun(rev.id).then(() => null, (e: Error) => e); + expect(refused?.message).toMatch( + /^INVALID_STATE: run '.*' is parked at node 'wait_revision', but the send-back on request .* was issued from its own approval node 'review'/, + ); + expect(await data.find('sys_approval_request', { where: { flow_run_id: rev.flow_run_id } }), + 'no new pending round was opened').toHaveLength(1); + }); + + it('CONTROL: past the first page, a run that is not parked is still refused — by GUARD 3 too, where GUARD 2 cannot ask', async () => { + parkOthers(1000); + const req = await strandOnA({ restore: false }); + + const byGuard2 = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e); + expect(byGuard2?.message).toMatch(/is not suspended/); + + // An engine that cannot answer `hasSuspendedRun` skips GUARD 2; GUARD 3's + // keyed read finds no pause and refuses. + b.service.attachAutomation(without('hasSuspendedRun') as any); + const byGuard3 = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e); + expect(byGuard3?.message).toMatch(/^INVALID_STATE: run '.*' is parked at no node this engine can see/); + expect(marks, 'nothing ran').toEqual([]); + }); + + it('CONTROL: an engine that cannot say where a run is parked is refused', async () => { + const req = await strandOnA(); + b.service.attachAutomation(without('suspensionOf') as any); + + const refused = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e); + expect(refused?.message).toMatch(/^INVALID_STATE: this automation engine cannot report WHERE run '.*' is parked \(no suspensionOf\)/); + expect(marks, 'nothing ran').toEqual([]); + expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the pause is left as it was').toBe(true); + }); +}); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index a63f1be5343..099bc293118 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -6905,6 +6905,26 @@ export class AutomationEngine implements IAutomationService { return { runId: parent.runId, flowName: parent.flowName, nodeId: parent.nodeId, correlation: parent.correlation }; } + /** + * [#28271] Where `runId` is parked — its flow, node and correlation — or + * `undefined` when it is not suspended. One keyed read through the strict, + * store-authoritative loader {@link hasSuspendedRun} uses, never a listing: + * {@link listSuspendedRunsDurable} reads one capped page of a DB-backed + * store, so past 1000 parked runs a run another process parked may not be + * on it. + * + * Read-only. THROWS when the store cannot be read, as + * {@link hasSuspendedRun} does: an outage means "unknown", never "not + * parked". + */ + async suspensionOf( + runId: string, + ): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined> { + const run = await this.loadSuspendedRunStrict(runId); + if (!run) return undefined; + return { runId: run.runId, flowName: run.flowName, nodeId: run.nodeId, correlation: run.correlation }; + } + /** * Credit a completed child run's totals to the parent step waiting on it * (#4354). diff --git a/packages/services/service-automation/src/suspension-of.test.ts b/packages/services/service-automation/src/suspension-of.test.ts new file mode 100644 index 00000000000..e8ea050eca8 --- /dev/null +++ b/packages/services/service-automation/src/suspension-of.test.ts @@ -0,0 +1,95 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * `suspensionOf` answers where one run is parked, with a keyed read by its id + * (#28271). + * + * The approvals re-issue of a restored continuation reads it to prove the + * pause it continues is the one its outcome was issued at. It used to find the + * run in `listSuspendedRunsDurable`, which reads one capped page of a + * DB-backed store; the end-to-end pin past that page is + * `restored-approval-continuation.test.ts` (plugin-approvals). + * + * The pins: a parked run answers its node, also on an engine that did not park + * it; the answer follows the run to its next pause; a completed run and an + * unknown id answer `undefined`; a store that cannot be read rejects. + */ + +import { describe, it, expect, beforeEach } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { AutomationEngine } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; + +function silentLogger(): any { + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } }; +} + +/** start → `p1` → `p2` → end, both pauses raw-resumable. */ +const TWO_PAUSES = { + name: 'two_pauses', label: 'two_pauses', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'p1', type: 'open_pause', label: 'First' }, + { id: 'p2', type: 'open_pause', label: 'Second' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'p1' }, + { id: 'e2', source: 'p1', target: 'p2' }, + { id: 'e3', source: 'p2', target: 'end' }, + ], +}; + +function makeEngine(store: InMemorySuspendedRunStore): AutomationEngine { + const engine = new AutomationEngine(silentLogger(), store); + engine.registerNodeExecutor({ + type: 'open_pause', + descriptor: defineActionDescriptor({ + type: 'open_pause', version: '1.0.0', name: 'Open Pause', supportsPause: true, resumeAuthority: 'any', + }), + async execute() { return { success: true, suspend: true }; }, + }); + engine.registerFlow('two_pauses', TWO_PAUSES as never); + return engine; +} + +describe('suspensionOf answers where one run is parked, by its id (#28271)', () => { + let store: InMemorySuspendedRunStore; + let engine: AutomationEngine; + + beforeEach(() => { + store = new InMemorySuspendedRunStore(); + engine = makeEngine(store); + }); + + it('a parked run answers its node, also on an engine that did not park it', async () => { + const run = await engine.execute('two_pauses', {}); + expect(run.status).toBe('paused'); + + const expected = { runId: run.runId, flowName: 'two_pauses', nodeId: 'p1' }; + expect(await engine.suspensionOf(run.runId!)).toMatchObject(expected); + expect(await makeEngine(store).suspensionOf(run.runId!)).toMatchObject(expected); + }); + + it('the answer follows the run to its next pause, and is undefined once it completes', async () => { + const run = await engine.execute('two_pauses', {}); + const other = makeEngine(store); + + expect((await engine.resume(run.runId!, {})).status).toBe('paused'); + expect((await other.suspensionOf(run.runId!))?.nodeId, 'another engine reads the store, not a memory').toBe('p2'); + + expect((await other.resume(run.runId!, {})).success).toBe(true); + expect(await engine.suspensionOf(run.runId!), 'nor a stale entry of its own').toBeUndefined(); + }); + + it('an unknown id answers undefined', async () => { + expect(await engine.suspensionOf('no-such-run')).toBeUndefined(); + }); + + it('a store that cannot be read rejects: an outage is "unknown", never "not parked"', async () => { + const run = await engine.execute('two_pauses', {}); + store.load = async () => { throw new Error('store down'); }; + + await expect(engine.suspensionOf(run.runId!)).rejects.toThrow('store down'); + }); +});