Skip to content

Commit 00a350f

Browse files
fix(plugin-approvals): re-issuing a restored approval continuation proves the pause past the first 1000 parked runs (#28314)
<!-- claude-projects-attribution: {"github_login":"os-elon-musk"} --> _Requested by **Elon** · [project thread](https://claude.ai/code/project/chan_01WedDzV26nniwxv7jC88B9D?thread=cmsg_01WedDzV26nniwxv7jC88B9DX6HgHjCbUTACRbNAr419qF)_ domain:server#3 · os-elon-musk · session_016WV4czGci1wfPnGZhkvrcF Fixes #28271 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`) 的 GUARD 3 改为按 run id 键读这个 run 停在哪个节点(新增引擎成员 `AutomationEngine.suspensionOf(runId)`),不再在 `listSuspendedRunsDurable()` 的 1000 行首页里找。 - 为什么改:实测 main(`69ce01d7`):两个副本共用一个 `ObjectStoreSuspendedRunStore`,前面已挂起 1000 条别的 run,副本 `a` 搁浅并恢复一个驳回决定,副本 `b` 重发:每次都以 `INVALID_STATE` "parked at no node this engine can see" 拒绝,连试 3 次一样,决定永远落不了地;998 条时正常。 - 风险与回滚:只换了「run 停在哪」这一步读法;期望节点的推导、不在暂停或停在别的节点照旧拒绝、重发本身都没变。新成员只读,一次按 id 读,用的是 `hasSuspendedRun` 同一个严格读法,存储读不出时抛错。回滚即 revert 本 PR。 - 先看:`packages/services/service-automation/src/engine.ts` 的 `suspensionOf`,与 `approval-service.ts` 里 `continueRestoredRun` 的 GUARD 3。 - 两分钟验证:先 build `@objectstack/service-automation`,再 `pnpm --filter @objectstack/plugin-approvals exec vitest run src/restored-approval-continuation.test.ts`(预期 20 passed)与 `pnpm --filter @objectstack/service-automation exec vitest run src/suspension-of.test.ts`(预期 4 passed)。 - 需要你做的:无,席位在 CI 绿后按落地授权落地。 ## What changes - **`packages/services/service-automation/src/engine.ts`: one new public read, `suspensionOf(runId)`.** It loads the run's suspension by id through `loadSuspendedRunStrict`, the store-authoritative loader `hasSuspendedRun` uses, and answers `{ runId, flowName, nodeId, correlation }`, or `undefined` when the run is not suspended. One keyed read, never a listing. It rejects when the store cannot be read, as `hasSuspendedRun` does. It sits beside `suspendedMapParentOf` (#28268). No other engine hunk. - **`packages/plugins/plugin-approvals/src/approval-service.ts`.** `ApprovalResumeSurface` declares `suspensionOf?` (optional, fail-closed: an engine without it is refused, as an engine without the listing was). GUARD 3 of `continueRestoredRun` reads `suspensionOf(runId)?.nodeId` instead of the run's entry in `listSuspendedRunsDurable()`. The refusal texts are unchanged except the missing-reader one, which now names `suspensionOf`. `listSuspendedRunsDurable?` stays declared (the interface is exported); its docblock now says this service no longer reads it. - **Tests.** `suspension-of.test.ts` (new, service-automation) pins the engine member. `restored-approval-continuation.test.ts` (plugin-approvals) gains a `#28271` block over two replicas and one `ObjectStoreSuspendedRunStore`, reusing the file's existing engine double, so `scripts/engine-double-contract.pinned.json` is untouched. - **`.changeset/28271-restored-continuation-past-first-page.md`.** `@objectstack/plugin-approvals` and `@objectstack/service-automation`, patch. ## Why ### Before (measured at `origin/main` `69ce01d7`) Replicas `a` and `b` share one `ObjectStoreSuspendedRunStore` and one table set. N `paused` rows of another flow are stored first. On `a`, a `deal_approval` run parks, its `reject` decision strands (the reject branch throws), and `restoreConsumedSuspension` re-arms the pause. On `b`, `continueRestoredRun` re-issues it. | | N = 1000, before | N = 998, before | N = 1000, after | |:---|:---|:---|:---| | `b.hasSuspendedRun(run)` | `true` | `true` | `true` | | run on `b.listSuspendedRunsDurable()` | no | yes | no (no longer read) | | `continueRestoredRun`, 3 tries | refused each time: `INVALID_STATE: run '…' is parked at no node this engine can see, …` | `resumed: true` | `resumed: true`, the reject branch runs | The listing reads one 1000-row page of `paused` rows in driver order, and a process adds only what it parked itself. So the run is missed exactly when another process (a replica, or this one before a restart) parked it behind more than 1000 runs, and a retry reads the same page. ### Why this path Triage 6535230031: a keyed read by run id, the same kind as `suspendedMapParentOf`, still fail-closed when the run is not parked. The strict loader is the one GUARD 2 (`hasSuspendedRun`) already reads, so GUARD 2 and GUARD 3 now ask the same store the same question; the listing's own merge also let the durable row win (#14332), so below the page nothing changes. On an outage GUARD 2 already throws before GUARD 3 is reached. ### Other `listSuspendedRunsDurable()` callers Grepped (`git grep -n "listSuspendedRunsDurable\|readSuspendedRuns"`, non-test). GUARD 3 was the only caller that looks up a single run. The rest: - `engine.ts` `listSuspendedRunsDurable()` itself — the operability listing; a page is its contract. - `subflow-guards.ts` `parkedRunsOf` → `readSuspendedRuns('throw', flowNames)` — per-flow, read to its end with `listByFlow` (#20725); not a page. - `runtime/src/domains/automation.ts:1628` reads the synchronous in-memory `listSuspendedRuns()`, not the durable listing, for a listing, not a single-run lookup. - The revise-window recall's batch lookup moved off the listing in #28268. ## Tests `suspension-of.test.ts` (4 cells): a parked run answers its node, also on a second engine over the same store with nothing in memory; the answer follows the run to its next pause on another engine and is `undefined` once that engine completes it (no stale memory answer); an unknown id answers `undefined`; a store whose `load` throws rejects. `restored-approval-continuation.test.ts`, `#28271` block (5 cells): 1. past the first page, the pause `a` re-armed is proved on `b` and the decision is re-issued; the reject branch runs (red on `main`: refused "parked at no node this engine can see"); 2. CONTROL: within the first page (998) it is re-issued as before; 3. CONTROL: past the first page, a run parked at another node (the revise window after a send-back) is still refused, and the refusal names `wait_revision` (red on `main`: it named no node); 4. CONTROL: past the first page, a run that is not parked is still refused — by GUARD 2, and by GUARD 3 on an engine without `hasSuspendedRun`; 5. CONTROL: an engine without `suspensionOf` is refused with `INVALID_STATE`, and the pause is left in place. Ablations through `scripts/ablation-replace.mjs`, each restored to the HEAD blob with `git diff HEAD` empty: | Ablation | Red | |:---|:---| | A1: GUARD 3 reads the listing again | `#28271` cells 1 and 3 | | A2: GUARD 3 drops the node comparison | PIN 7, `#28271` cells 3 and 4 | | A3: GUARD 3 drops the missing-reader refusal | `#28271` cell 5 | | A4: `suspensionOf` answers from this process's memory, not the store | engine cells 1 and 2; `#28271` cells 1, 2 and 3 | ## Verification Head `c4e208a4`, after a whole-tree `turbo run build`, `pnpm install --frozen-lockfile` and an unshallow fetch; `@objectstack/service-automation` rebuilt after the engine edit (plugin-approvals tests import its dist). - `pnpm --filter @objectstack/service-automation run test`: 236 files, 2973 passed. `pnpm --filter @objectstack/plugin-approvals run test`: 106 files, 1366 passed. Both packages' `typecheck` (including `check:test-typecheck`): exit 0. - `dispatch-gates --commands`: 64 families, all exit 0. `--ran`: 64 derived, 64 run, 0 NOT-MEASURED, 0 UNRUN. `check:engine-double-contract` and `check:objectql-double-limit` are green with no ledger edit. - `pnpm lint`: exit 0. `check-commit-card-trailers --range origin/main..HEAD`: exit 0. Skills-docs: none. `git grep -n "listSuspendedRunsDurable\|parked at no node\|continueRestoredRun\|suspendedMapParentOf" -- skills content docs` has no hit; `git grep -il "restoreConsumedSuspension\|re-issue the continuation" -- skills content` hits only `content/docs/automation/flows.mdx`, whose restore paragraph does not describe the listing or a page cap. ## Acceptance notes - `run-history.ts:463-465` (`readSuspendedRuns`' degrade-path comment) still says "plugin-approvals' restored-continuation proof refuses on an absent entry, so a short list costs it a retry". After this PR that proof no longer reads the listing, so the sentence is stale; the comment's conclusion (keep the level at `warn`) still holds. Outside this card's file surface; not edited. ## Human Minimum - [x] 1. Schemas start as Zod; `packages/spec` holds no runtime logic. (Not touched.) - [x] 2. Every name I added uses the casing for its kind. - [x] 3. No formatter ran; `pnpm lint` passes. - [x] 4. One issue, one branch off an up-to-date `main`, one pull request. - [x] 5. A change that publishes carries a changeset, and both declaration lines are filled in here and in the changeset. - [x] 6. `packages/spec` touched: its generated artifacts are regenerated and committed. (Not touched.) - [x] 7. No hand edits to `content/docs/references/`, `packages/*/CHANGELOG.md` or `content/docs/releases/`. - [x] 8. Title, body and comments are in English (the 维护者速读 block is the lane's maintainer summary). - [x] The *Before Submitting* list in CONTRIBUTING.md is done. 🤖 Generated with [Claude Code](https://claude.com/claude-code) https://claude.ai/code/session_016WV4czGci1wfPnGZhkvrcF Co-authored-by: Claude <noreply@anthropic.com>
1 parent 69c39b1 commit 00a350f

5 files changed

Lines changed: 322 additions & 23 deletions

File tree

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
---
2+
'@objectstack/plugin-approvals': patch
3+
'@objectstack/service-automation': patch
4+
---
5+
6+
fix(plugin-approvals): re-issuing a restored approval continuation proves the pause past the first 1000 parked runs
7+
8+
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)
9+
Metadata-protocol: none
10+
11+
`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.
12+
13+
- **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.
14+
- **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.

‎packages/plugins/plugin-approvals/src/approval-service.ts‎

Lines changed: 38 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -270,24 +270,36 @@ export interface ApprovalResumeSurface {
270270
*/
271271
unresumableSuspensionCause?(runId: string): Promise<string | undefined>;
272272
/**
273-
* [#15389] Where each suspended run is currently parked. Read by
274-
* {@link ApprovalService.continueRestoredRun} to prove the pause it was asked
275-
* to continue is the one THIS request's recorded outcome was issued at, and
276-
* not merely some live pause on the same run — `hasSuspendedRun` answers a
277-
* boolean and cannot tell the two apart. Which node that is depends on the signal, not
278-
* only on the row: see {@link ApprovalService.expectedPauseNode}.
273+
* [#15389] Where each suspended run is currently parked, durable-first.
279274
*
280275
* ⚠️ Declares a method `AutomationEngine` ALREADY implements publicly
281-
* (`listSuspendedRunsDurable`); it widens no engine surface. Durable-first,
282-
* so it sees a pause parked by another replica or before a restart.
276+
* (`listSuspendedRunsDurable`); it widens no engine surface. [#28271]
277+
* {@link ApprovalService.continueRestoredRun} no longer reads it: on a
278+
* DB-backed store it is one capped page, and it now asks
279+
* {@link suspensionOf} for the one run instead.
280+
*/
281+
listSuspendedRunsDurable?(): Promise<Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }>>;
282+
/**
283+
* [#15389, #28271] Where the run `runId` is parked, or `undefined` when it
284+
* is not suspended. Read by {@link ApprovalService.continueRestoredRun} to
285+
* prove the pause it was asked to continue is the one THIS request's
286+
* recorded outcome was issued at, and not merely some live pause on the
287+
* same run — `hasSuspendedRun` answers a boolean and cannot tell the two
288+
* apart. Which node that is depends on the signal, not only on the row:
289+
* see {@link ApprovalService.expectedPauseNode}.
290+
*
291+
* ⚠️ Declares the method `AutomationEngine` implements publicly
292+
* (`suspensionOf`): one keyed read of the store, so it sees a pause parked
293+
* by another replica or before a restart however many runs are parked
294+
* ahead of it — where {@link listSuspendedRunsDurable}, which this proof
295+
* read until #28271, is one capped page of a DB-backed store. Read-only;
296+
* rejects when the store cannot be read.
283297
*
284298
* Optional, and its absence is FAIL-CLOSED: a caller that cannot prove node
285299
* identity refuses the continuation rather than proceeding on the weaker
286-
* check. The engine degrades a store outage to its in-memory list rather
287-
* than throwing, and that degradation is safe HERE for the same reason — a
288-
* pause it cannot see is a pause this verb will not act on.
300+
* check.
289301
*/
290-
listSuspendedRunsDurable?(): Promise<Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }>>;
302+
suspensionOf?(runId: string): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined>;
291303
/**
292304
* [#28230] The run parked at a `map` node on the item run `itemRunId`, or
293305
* `undefined` when there is none. Read by
@@ -7381,23 +7393,27 @@ export class ApprovalService implements IApprovalService {
73817393
// node refused the one population the journal was built to serve, with a
73827394
// message that named a cause the code had not taken.
73837395
//
7384-
// FAIL-CLOSED in every direction: no reader, no listing, no matching entry,
7385-
// an entry at another node, or an expected node this service cannot derive
7386-
// — all refuse. The engine degrades a store outage to its in-memory list
7387-
// instead of throwing, so a pause this cannot see is a pause this verb
7388-
// declines to act on. A false refusal costs the operator a retry; a false
7389-
// admission advances a flow nobody decided.
7396+
// FAIL-CLOSED in every direction: no reader, no pause, a pause at another
7397+
// node, or an expected node this service cannot derive — all refuse. A
7398+
// store that cannot be read rejects the read, as GUARD 2's does. A false
7399+
// refusal costs the operator a retry; a false admission advances a flow
7400+
// nobody decided.
7401+
//
7402+
// [#28271] The read is KEYED by the run id. It used to be the run's entry
7403+
// in `listSuspendedRunsDurable()`, which on a DB-backed store is one page
7404+
// of 1000 parked runs plus what this process parked itself: a run another
7405+
// process parked behind more than 1000 others was on no page, so every
7406+
// retry was refused the same way and the recorded outcome never took effect.
73907407
const expected = await this.expectedPauseNode(raw, signal, requestId, runId);
7391-
if (typeof this.automation?.listSuspendedRunsDurable !== 'function') {
7408+
if (typeof this.automation?.suspensionOf !== 'function') {
73927409
throw new Error(
73937410
`INVALID_STATE: this automation engine cannot report WHERE run '${runId}' is parked ` +
7394-
`(no listSuspendedRunsDurable), so the pause cannot be proved to be the one ${signal.what} on ` +
7411+
`(no suspensionOf), so the pause cannot be proved to be the one ${signal.what} on ` +
73957412
`request ${requestId} was refused on — refusing, because continuing the wrong pause advances a ` +
73967413
`flow with no decision behind it`,
73977414
);
73987415
}
7399-
const parkedAt = (await this.automation.listSuspendedRunsDurable())
7400-
.find(r => String(r.runId) === String(runId))?.nodeId;
7416+
const parkedAt = (await this.automation.suspensionOf(runId))?.nodeId;
74017417
if (parkedAt !== expected.nodeId) {
74027418
throw new Error(
74037419
`INVALID_STATE: run '${runId}' is parked at ` +

‎packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts‎

Lines changed: 155 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,7 @@
3939
*/
4040

4141
import { describe, it, expect, beforeEach } from 'vitest';
42-
import { AutomationEngine, InMemorySuspendedRunStore, resolveRunDataContext } from '@objectstack/service-automation';
42+
import { AutomationEngine, InMemorySuspendedRunStore, ObjectStoreSuspendedRunStore, resolveRunDataContext } from '@objectstack/service-automation';
4343
// [#4550] The engine double below routes its write verbs through ObjectQL's OWN
4444
// dispatch predicates rather than a hand-mirrored copy.
4545
import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql';
@@ -1035,3 +1035,157 @@ describe('ruling A — a decision resumes the run as the person who decided', ()
10351035
expect(seen.after_reject, 'the replay fell back to the trigger user').toEqual({ userId: 'u1', permissions: ['deal_manager'] });
10361036
});
10371037
});
1038+
1039+
/**
1040+
* GUARD 3 proves where the run is parked past the first page of parked runs
1041+
* (#28271).
1042+
*
1043+
* It used to find the run in `listSuspendedRunsDurable()`. A DB-backed store
1044+
* lists one 1000-row page of `paused` rows in driver order, and a process adds
1045+
* only what it parked itself. So when another process (a replica, or this one
1046+
* before a restart) parked the run behind more than 1000 others, the run was
1047+
* not on the listing, and every re-issue was refused with "parked at no node
1048+
* this engine can see" — the retry the refusal promises read the same page.
1049+
* Both replicas below share one store and one table set: `a` strands and
1050+
* restores the run, `b` re-issues it.
1051+
*/
1052+
describe('#28271 — GUARD 3 reads where the run is parked by its id, not from one page', () => {
1053+
let data: ReturnType<typeof makeFakeEngine>;
1054+
let marks: string[];
1055+
let rejectBranchThrows: string | undefined;
1056+
let a: ReturnType<typeof wire>;
1057+
let b: ReturnType<typeof wire>;
1058+
1059+
/** One process: a real engine over `store`, the approval node, and its own approvals service. */
1060+
function wire(store: ObjectStoreSuspendedRunStore) {
1061+
const service = new ApprovalService({ engine: data as any, logger: noopLogger });
1062+
const automation = new AutomationEngine(noopLogger as any, store);
1063+
registerApprovalNode(automation, service, noopLogger as any);
1064+
automation.registerNodeExecutor({
1065+
type: 'mark',
1066+
async execute(node: any) {
1067+
if (node.id === 'mark_rejected' && rejectBranchThrows) throw new Error(rejectBranchThrows);
1068+
marks.push(node.id);
1069+
return { success: true };
1070+
},
1071+
} as never);
1072+
automation.registerFlow('deal_approval', DEAL_APPROVAL as never);
1073+
automation.registerFlow('revise_flow', REVISE_FLOW as never);
1074+
service.attachAutomation(automation);
1075+
return { automation, service };
1076+
}
1077+
1078+
/** `count` parked runs of some other flow, stored ahead of everything this test parks. */
1079+
const parkOthers = (count: number) => {
1080+
const rows = data.tables.get('sys_automation_run') ?? (data.tables.set('sys_automation_run', []), data.tables.get('sys_automation_run')!);
1081+
for (let i = 0; i < count; i++) {
1082+
rows.push({ id: `other-${String(i).padStart(4, '0')}`, flow_name: 'other_flow', node_id: 'wait', status: 'paused' });
1083+
}
1084+
};
1085+
1086+
/** `b`'s engine with one member taken away, everything else delegated. */
1087+
const without = (member: string) => new Proxy(b.automation, {
1088+
get(target, prop) {
1089+
if (prop === member) return undefined;
1090+
const value = Reflect.get(target, prop, target);
1091+
return typeof value === 'function' ? value.bind(target) : value;
1092+
},
1093+
});
1094+
1095+
beforeEach(() => {
1096+
marks = []; rejectBranchThrows = undefined;
1097+
data = makeFakeEngine();
1098+
const store = new ObjectStoreSuspendedRunStore(data as any, noopLogger as any);
1099+
a = wire(store);
1100+
b = wire(store);
1101+
});
1102+
1103+
const rowOf = async (id: string) =>
1104+
(await data.find('sys_approval_request', { where: { id } }))[0];
1105+
1106+
/** On `a`: park, strand a reject decision, and (unless told not to) re-arm its pause. */
1107+
async function strandOnA({ restore = true } = {}) {
1108+
await a.automation.execute('deal_approval', {
1109+
object: 'crm_deal', record: { id: 'd1', amount: 100 }, userId: 'submitter',
1110+
} as never);
1111+
const req = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0];
1112+
rejectBranchThrows = DOWNSTREAM_FAILURE;
1113+
const err = await a.service
1114+
.decide(req.id, { decision: 'reject', actorId: 'u1', comment: 'no' }, SYSTEM_CTX)
1115+
.then(() => null, (e: Error) => e);
1116+
expect(err?.message).toMatch(/^RESUME_FAILED/);
1117+
if (restore) {
1118+
expect((await a.automation.restoreConsumedSuspension(req.flow_run_id, { requestedBy: 'ops' })).restored).toBe(true);
1119+
}
1120+
rejectBranchThrows = undefined;
1121+
return req;
1122+
}
1123+
1124+
it('past the first page, the pause another process re-armed is proved and the decision is re-issued', async () => {
1125+
parkOthers(1000);
1126+
const req = await strandOnA();
1127+
// The premise: `b` holds nothing in memory, and the run is not on the page.
1128+
expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the pause is back, in the store').toBe(true);
1129+
expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === req.flow_run_id),
1130+
'and not on the listing\'s one page').toBe(false);
1131+
1132+
const out = await b.service.continueRestoredRun(req.id, { requestedBy: 'ops', reason: 'record recreated' });
1133+
expect(out.resumed, JSON.stringify(out)).toBe(true);
1134+
expect(out.decision).toBe('reject');
1135+
expect(out.source).toBe('journal');
1136+
expect(marks, 'the reject branch ran').toEqual(['mark_rejected']);
1137+
expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the continuation consumed the pause').toBe(false);
1138+
});
1139+
1140+
it('CONTROL: within the first page it is re-issued as before', async () => {
1141+
parkOthers(998);
1142+
const req = await strandOnA();
1143+
expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === req.flow_run_id)).toBe(true);
1144+
1145+
const out = await b.service.continueRestoredRun(req.id, { requestedBy: 'ops' });
1146+
expect(out.resumed, JSON.stringify(out)).toBe(true);
1147+
expect(marks).toEqual(['mark_rejected']);
1148+
});
1149+
1150+
it('CONTROL: past the first page, a run parked at another node is still refused, and the refusal names that node', async () => {
1151+
parkOthers(1000);
1152+
await a.automation.execute('revise_flow', {
1153+
object: 'crm_deal', record: { id: 'd7', amount: 3 }, userId: 'submitter',
1154+
} as never);
1155+
const rev = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0];
1156+
await a.service.sendBack(rev.id, { actorId: 'u1', comment: 'fix it' } as any, SYSTEM_CTX);
1157+
expect((await rowOf(rev.id)).status).toBe('returned');
1158+
1159+
const refused = await b.service.continueRestoredRun(rev.id).then(() => null, (e: Error) => e);
1160+
expect(refused?.message).toMatch(
1161+
/^INVALID_STATE: run '.*' is parked at node 'wait_revision', but the send-back on request .* was issued from its own approval node 'review'/,
1162+
);
1163+
expect(await data.find('sys_approval_request', { where: { flow_run_id: rev.flow_run_id } }),
1164+
'no new pending round was opened').toHaveLength(1);
1165+
});
1166+
1167+
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 () => {
1168+
parkOthers(1000);
1169+
const req = await strandOnA({ restore: false });
1170+
1171+
const byGuard2 = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e);
1172+
expect(byGuard2?.message).toMatch(/is not suspended/);
1173+
1174+
// An engine that cannot answer `hasSuspendedRun` skips GUARD 2; GUARD 3's
1175+
// keyed read finds no pause and refuses.
1176+
b.service.attachAutomation(without('hasSuspendedRun') as any);
1177+
const byGuard3 = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e);
1178+
expect(byGuard3?.message).toMatch(/^INVALID_STATE: run '.*' is parked at no node this engine can see/);
1179+
expect(marks, 'nothing ran').toEqual([]);
1180+
});
1181+
1182+
it('CONTROL: an engine that cannot say where a run is parked is refused', async () => {
1183+
const req = await strandOnA();
1184+
b.service.attachAutomation(without('suspensionOf') as any);
1185+
1186+
const refused = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e);
1187+
expect(refused?.message).toMatch(/^INVALID_STATE: this automation engine cannot report WHERE run '.*' is parked \(no suspensionOf\)/);
1188+
expect(marks, 'nothing ran').toEqual([]);
1189+
expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the pause is left as it was').toBe(true);
1190+
});
1191+
});

‎packages/services/service-automation/src/engine.ts‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6905,6 +6905,26 @@ export class AutomationEngine implements IAutomationService {
69056905
return { runId: parent.runId, flowName: parent.flowName, nodeId: parent.nodeId, correlation: parent.correlation };
69066906
}
69076907

6908+
/**
6909+
* [#28271] Where `runId` is parked — its flow, node and correlation — or
6910+
* `undefined` when it is not suspended. One keyed read through the strict,
6911+
* store-authoritative loader {@link hasSuspendedRun} uses, never a listing:
6912+
* {@link listSuspendedRunsDurable} reads one capped page of a DB-backed
6913+
* store, so past 1000 parked runs a run another process parked may not be
6914+
* on it.
6915+
*
6916+
* Read-only. THROWS when the store cannot be read, as
6917+
* {@link hasSuspendedRun} does: an outage means "unknown", never "not
6918+
* parked".
6919+
*/
6920+
async suspensionOf(
6921+
runId: string,
6922+
): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined> {
6923+
const run = await this.loadSuspendedRunStrict(runId);
6924+
if (!run) return undefined;
6925+
return { runId: run.runId, flowName: run.flowName, nodeId: run.nodeId, correlation: run.correlation };
6926+
}
6927+
69086928
/**
69096929
* Credit a completed child run's totals to the parent step waiting on it
69106930
* (#4354).

0 commit comments

Comments
 (0)