Skip to content

Commit 02bceb7

Browse files
fix(service-automation): the region refusal walks the child's suspension chain strictly and names the link it cannot read (#29403)
<!-- claude-projects-attribution: {"github_login":"os-steve"} --> _Requested by **Steve** · [project thread](https://claude.ai/code/project/chan_01FdL1XeUyvVNxLreh3F4Qi9?thread=cmsg_01FdL1XeUyvVNxLreh3F4Qi9JeMXnsbKRJxMbW9ZUq9nDf)_ Fixes #29333 Clause-②: no Metadata-protocol: none ## 维护者速读 - 改了什么:区域(loop / parallel / try_catch)内子流程挂起被拒绝时,取消子运行挂起链改用严格读取;链中间某一环读不到时,已读到的环照常取消,读不到的那一环仍交给 `cancelRun`,并记一条 `error` 点名该运行。 - 为什么改:原来用降级读取,中间环读不到就当作"没有挂起",遍历停在那里,它和它下面的运行一直停在存储里,只留一条 warn。 - 风险与回滚:只改取消路径的读取方式;恢复门禁等其它读者仍用原来的降级遍历,未改。拒绝本身(文字、不可路由)不变。回滚即还原 `dag-traversal.ts` 这一段和 `engine.ts` 新增的方法。 - 先看:`packages/services/service-automation/src/dag-traversal.ts`(`runRegion` 的取消段),再看 `engine.ts` 的 `resolveSuspensionChainForCancel` / `walkSuspensionChain`,最后看测试文件末尾新增的两例。 - 两分钟验证:`pnpm --filter @objectstack/service-automation exec vitest run src/subflow-region-suspension.test.ts`,预期 24 例全过。 - 需要你做的:无,席位落地。 ## What changes - `engine.ts`: the chain walk behind `resolveSuspensionChain` moves into `walkSuspensionChain(runId, load)`; only the loader is a parameter, so there is still one walk with the same depth bound and cycle guard. `resolveSuspensionChain` calls it with the degrading `loadSuspendedRun`, exactly as before, so the resume gate and its other readers see no change. New `resolveSuspensionChainForCancel` calls it with the strict `loadSuspendedRunStrict` (the read `cancelRun` already takes) and returns `{ chain, unreadable? }`, where `unreadable` names the link whose read threw, with the cause. A link that is simply not found still ends the chain. - `dag-traversal.ts` `runRegion`: the region refusal's cancel now walks through `resolveSuspensionChainForCancel`. Every link it read is cancelled, deepest first. If a link could not be read, that run is still handed to `cancelRun` first (it reads again, owns the `error` record for that run if the store is still unreadable, and is a no-op if the run is gone), and one `error` record names it: any run it is itself parked on was never reached and may still be parked. The refusal is thrown after, unchanged. - #29329's empty-chain branch is subsumed, not dropped: an unreadable child now comes back as `unreadable` = the child, so it gets the same single `cancelRun`. A child that is truly not found (the strict store answered "no row") gets no cancel, where #29329 sent a `cancelRun` that returned `false`. #29329's two tests pass unedited. - Changeset: `.changeset/29333-region-refusal-deeper-chain-cancel.md`, `@objectstack/service-automation` patch. ## Tests Two cases added at the end of `src/subflow-region-suspension.test.ts` (no existing test edited), on a three-run chain caller → `loop` → `deal_review` → `deal_mid` → `deal_ask` (parks at `wait`): - **middle link unreadable** (the store's `load` always throws for `deal_mid` rows): the refusal text is unchanged; `deal_review` ends `cancelled`; exactly one chain `error` record names `deal_mid`'s run id; `cancelRun` is called for `deal_mid` then `deal_review`, deepest first. `deal_mid` and `deal_ask` are still in the store, asserted as such. - **CONTROL, all readable**: all three runs end `cancelled`, `cancelRun` is called deepest first, and nothing is logged at `error`. ## Acceptance notes - `deal_ask` (below the unreadable link) stays parked. Its id is only recorded in `deal_mid`'s row (`subflow:` correlation), and a `SuspendedRun` carries no parent pointer, so nothing can reach it while that row cannot be read. The `error` record says so and tells the operator to cancel the remaining runs once the store answers. - When the unreadable link's store is still down, two `error` records appear for it: `cancelRun`'s own skipped-cancellation line for that run, and the chain record for the runs below it. They say different things, so both stay. ## Verification <details><summary>Ablation (scripts/ablation-replace.mjs, on-disk proof): the cancel walk read through the degrading loader again</summary> ``` ablation-replace: anchor "return await this.loadSuspendedRunStrict(id);" x1 -> x0 ablation-replace: ok mutation landed: anchor 1 -> 0, blob c937469 -> ebabde87953c × an unreadable store: … the child run gets exactly one cancelRun, … (#29166) × a store that answers again by the cancel: … (#29166) × the middle link cannot be read: the child is cancelled, one `error` record names the middle run, … AssertionError: one record names the unreadable link: expected [] to have a length of 1 but got +0 Tests 3 failed | 21 passed (24) ablation-replace: ok restored: blob == HEAD (c937469) and `git diff HEAD` is empty ``` The CONTROL stays green under the mutation, as it should. The test imports `./engine.js` from `src`, so no `dist` leg applies. </details> <details><summary>Final head a06cb53</summary> ``` head a06cb53 (merge of main 5924e50, which carries #29329, onto e6d2696): pnpm --filter '@objectstack/service-automation...' build, typecheck, test under os-verify-lock, VERDICT command-exit 0 typecheck exit 0 · Test Files 260 passed (260) · Tests 3342 passed (3342) vitest run src/subflow-region-suspension.test.ts Tests 24 passed (24) turbo build ./packages/** (78/78) then pnpm check:dual-build-cjs-loads and pnpm lint VERDICT command-exit 0 dispatch-gates --ran: 62 derived, 62 run, 0 NOT-MEASURED (all exit 0) Two first answered exit 3 (prerequisite, nothing measured) and were re-run green: check-plugin-teardown-shape --self-test (after git fetch origin 621a487…, its pinned fixture commit) check:dual-build-cjs-loads (after the ./packages/** build above) ``` </details> --- _Generated by [Claude Code](https://claude.ai/code/session_01ADHkmS5997viAw9y7Jknnz)_ <!-- signed: domain:server#3 · os-steve · session_01ADHkmS5997viAw9y7Jknnz --> --------- Co-authored-by: Claude <noreply@anthropic.com>
1 parent 8e6ab05 commit 02bceb7

4 files changed

Lines changed: 144 additions & 18 deletions

File tree

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
---
2+
'@objectstack/service-automation': patch
3+
---
4+
5+
fix(service-automation): the region refusal names the deeper suspended run it could not read instead of leaving it parked in silence
6+
7+
Clause-②: no
8+
Metadata-protocol: none
9+
10+
When a `subflow` or `map` call inside a `loop` body, a `parallel` branch or a `try_catch` slot starts a child flow that suspends, the run fails at the region boundary and cancels the child run with every run it is parked on. That walk read each link through a best-effort read, so a link deeper in the chain that the suspended-run store could not read stopped the walk with only a `warn`: that run, and every run below it, stayed parked. The walk now reads through the strict loader that `cancelRun` uses. It cancels every link it read, still cancels the link it could not read through `cancelRun` (which records its own `error` if the store is still unreadable), and logs one `error` naming that run, because any run below it was never reached and may still be parked. The refusal itself, its text and its un-routability are unchanged.

‎packages/services/service-automation/src/dag-traversal.ts‎

Lines changed: 30 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -142,7 +142,7 @@ export interface DagTraversalHost {
142142
readonly persistSuspendedRun: AutomationEngine['persistSuspendedRun'];
143143
readonly flowChecksumOf: AutomationEngine['flowChecksumOf'];
144144
readonly renderFlowTerminalMessage: AutomationEngine['renderFlowTerminalMessage'];
145-
readonly resolveSuspensionChain: AutomationEngine['resolveSuspensionChain'];
145+
readonly resolveSuspensionChainForCancel: AutomationEngine['resolveSuspensionChainForCancel'];
146146
readonly cancelRun: AutomationEngine['cancelRun'];
147147
}
148148

@@ -751,7 +751,7 @@ export async function traverseNext(
751751

752752
/** The body of {@link AutomationEngine.runRegion}; that method's docblock is its contract. */
753753
export async function runRegion(
754-
this: Pick<DagTraversalHost, 'executeNode' | 'resolveSuspensionChain' | 'cancelRun' | 'logger'>,
754+
this: Pick<DagTraversalHost, 'executeNode' | 'resolveSuspensionChainForCancel' | 'cancelRun' | 'logger'>,
755755
region: FlowRegionParsed,
756756
variables: Map<string, unknown>,
757757
context: AutomationContext,
@@ -858,19 +858,35 @@ export async function runRegion(
858858
const childRunId = nestedChildRunId(err.correlation);
859859
if (childRunId !== undefined) {
860860
try {
861-
const chain = await this.resolveSuspensionChain(childRunId);
861+
// [#29333] The chain is read through the STRICT loader,
862+
// the read `cancelRun` itself takes. The degrading one
863+
// answered an unreadable link as "nothing suspended", so
864+
// the walk stopped there and every run below that link
865+
// stayed parked with only a `warn` said (#29166 covered
866+
// the case where the unreadable link is the child itself).
867+
const { chain, unreadable } = await this.resolveSuspensionChainForCancel(childRunId);
862868
const reason = `the run that started it failed at a region boundary: ${refusal.message}`;
863-
if (chain.length === 0) {
864-
// [#29166] The chain is read through the DEGRADING
865-
// loader, so a store it cannot read reads as "nothing
866-
// suspended" and walking it cancelled nothing, with
867-
// only that loader's `warn` said, while the child run
868-
// stayed parked. The child id is known here, so it is
869-
// cancelled once anyway: `cancelRun` reads through the
870-
// STRICT loader and owns the `error` record when the
871-
// store is still unreadable, and it is a no-op when
872-
// there really is no suspension under that id.
873-
await this.cancelRun(childRunId, reason);
869+
if (unreadable) {
870+
// The deepest link the walk knows of. Its id is known
871+
// (the link above names it, or it is the child), so it
872+
// is cancelled anyway, deepest first: `cancelRun`
873+
// reads it again and owns the `error` record for THAT
874+
// run when the store still cannot be read, and it is a
875+
// no-op when the run is in fact gone.
876+
await this.cancelRun(unreadable.runId, reason);
877+
// What `cancelRun` cannot say: the walk never read
878+
// this link, so a run it is itself parked on was never
879+
// reached and may still be parked. #4632 verdict:
880+
// DURABILITY, `error`.
881+
this.logger.error(
882+
`[automation] node '${err.nodeId}' suspended inside region '${refusal.regionNodeId}', and ` +
883+
`the suspended run '${unreadable.runId}' in the chain of the child run '${childRunId}' could not be ` +
884+
`read while cancelling that chain, so any run '${unreadable.runId}' is itself parked on ` +
885+
`was not reached and may still be parked. Once the suspended-run store answers, cancel ` +
886+
`each with cancelRun(runId). The read failure is in this record's meta.`,
887+
undefined,
888+
describeThrownForLog(unreadable.cause),
889+
);
874890
}
875891
for (const run of [...chain].reverse()) await this.cancelRun(run.runId, reason);
876892
} catch (cancelErr) {

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

Lines changed: 43 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -7045,10 +7045,49 @@ export class AutomationEngine implements IAutomationService {
70457045
* walk starts, so a corrupt correlation cycle back to it ends the chain
70467046
* there instead of listing the caller's run again as its own child.
70477047
*/
7048-
private async resolveSuspensionChain(runId: string): Promise<SuspendedRun[]> {
7048+
private resolveSuspensionChain(runId: string): Promise<SuspendedRun[]> {
7049+
return this.walkSuspensionChain(runId, (id) => this.loadSuspendedRun(id));
7050+
}
7051+
7052+
/**
7053+
* [#29333] {@link resolveSuspensionChain} for a caller that is about to
7054+
* CANCEL the chain, read through the STRICT loader
7055+
* ({@link loadSuspendedRunStrict}, the read {@link cancelRun} itself takes).
7056+
*
7057+
* The degrading walk answers a link it cannot read as "nothing suspended
7058+
* there" and stops, which is right for the incidental readers (the resume
7059+
* gate) and wrong for a cancel: the runs below that link are left parked
7060+
* and nothing at `error` says so. This walk stops at the same place but
7061+
* says WHY: `unreadable` names the link whose read threw, with the cause,
7062+
* so the caller can still cancel what it read and record the rest. A link
7063+
* that is simply not found ends the chain as it does in the degrading walk.
7064+
*
7065+
* Same walk, same bounds: only the loader differs.
7066+
*/
7067+
private async resolveSuspensionChainForCancel(
7068+
runId: string,
7069+
): Promise<{ chain: SuspendedRun[]; unreadable?: { runId: string; cause: unknown } }> {
7070+
let unreadable: { runId: string; cause: unknown } | undefined;
7071+
const chain = await this.walkSuspensionChain(runId, async (id) => {
7072+
try {
7073+
return await this.loadSuspendedRunStrict(id);
7074+
} catch (cause) {
7075+
unreadable = { runId: id, cause };
7076+
return null;
7077+
}
7078+
});
7079+
return unreadable ? { chain, unreadable } : { chain };
7080+
}
7081+
7082+
/** The one chain walk behind {@link resolveSuspensionChain} and
7083+
* {@link resolveSuspensionChainForCancel}; `load` answers `null` to stop. */
7084+
private async walkSuspensionChain(
7085+
runId: string,
7086+
load: (runId: string) => Promise<SuspendedRun | null>,
7087+
): Promise<SuspendedRun[]> {
70497088
const seen = new Set<string>([runId]);
70507089
const chain: SuspendedRun[] = [];
7051-
let run = await this.loadSuspendedRun(runId);
7090+
let run = await load(runId);
70527091
for (let depth = 0; run && depth < AutomationEngine.MAX_SUSPENSION_CHAIN_DEPTH; depth++) {
70537092
chain.push(run);
70547093
const correlation = run.correlation;
@@ -7059,7 +7098,7 @@ export class AutomationEngine implements IAutomationService {
70597098
const childRunId = correlation!.slice(prefix.length);
70607099
if (!childRunId || seen.has(childRunId)) return chain;
70617100
seen.add(childRunId);
7062-
run = await this.loadSuspendedRun(childRunId);
7101+
run = await load(childRunId);
70637102
}
70647103
// The depth bound stopped the walk on a run it loaded but never read.
70657104
if (run) chain.push(run);
@@ -9357,7 +9396,7 @@ export class AutomationEngine implements IAutomationService {
93579396
// The body lives in ./dag-traversal.ts (#25222).
93589397
const host: ThisParameterType<typeof runRegion> = {
93599398
executeNode: (...a) => this.executeNode(...a),
9360-
resolveSuspensionChain: (...a) => this.resolveSuspensionChain(...a),
9399+
resolveSuspensionChainForCancel: (...a) => this.resolveSuspensionChainForCancel(...a),
93619400
cancelRun: (...a) => this.cancelRun(...a),
93629401
logger: this.logger,
93639402
};

‎packages/services/service-automation/src/subflow-region-suspension.test.ts‎

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -361,3 +361,64 @@ describe('the region refusal still cancels the child run when its suspension cha
361361
expect(runs.map((r: any) => r.status)).toEqual(['cancelled']);
362362
});
363363
});
364+
365+
describe('the region refusal cancels what it can read of a deeper suspension chain and names the link it cannot (#29333)', () => {
366+
/** caller --loop--> deal_review --subflow--> deal_mid --subflow--> deal_ask (parks at `wait`). */
367+
function threeRunRig(store: InMemorySuspendedRunStore) {
368+
const r = rig();
369+
r.engine.setSuspendedRunStore(store);
370+
r.engine.registerFlow('deal_ask', pausingChild('deal_ask'));
371+
r.engine.registerFlow('deal_mid', chain('deal_mid', [start(false), call('deal_ask', 'subflow', 'ask'), END]));
372+
r.engine.registerFlow('deal_review', chain('deal_review', [start(false), call('deal_mid', 'subflow', 'mid'), END]));
373+
r.engine.registerFlow('deal_submit', regionCaller('deal_submit', 'deal_review', 'loop', 'subflow', false));
374+
const cancel = vi.spyOn(r.engine, 'cancelRun');
375+
return { ...r, cancel };
376+
}
377+
378+
const chainErrors = (logger: any): string[] =>
379+
logger.error.mock.calls
380+
.map((c: unknown[]) => String(c[0]))
381+
.filter((m: string) => m.includes('could not be read while cancelling that chain'));
382+
383+
it('the middle link cannot be read: the child is cancelled, one `error` record names the middle run, and the refusal is still thrown', async () => {
384+
const store = new ChildLoadFailingStore('deal_mid', 'always');
385+
const { engine, logger, cancel } = threeRunRig(store);
386+
387+
const result: any = await engine.execute('deal_submit', {} as any);
388+
// The refusal itself is unchanged.
389+
expect(result.success).toBe(false);
390+
expect(String(result.error)).toContain(
391+
`durable pause inside a structured region: node 'step' (sub-flow 'deal_review') suspended`,
392+
);
393+
394+
expect((await engine.listRuns('deal_review')).map((r: any) => r.status), 'the readable link is cancelled').toEqual(['cancelled']);
395+
const parked = await store.list();
396+
// deal_mid cannot be read, and deal_ask is only reachable through deal_mid's row.
397+
expect(parked.map((r) => r.flowName).sort()).toEqual(['deal_ask', 'deal_mid']);
398+
const midRunId = parked.find((r) => r.flowName === 'deal_mid')!.runId;
399+
400+
const errors = chainErrors(logger);
401+
expect(errors, 'one record names the unreadable link').toHaveLength(1);
402+
expect(errors[0]).toContain(`the suspended run '${midRunId}'`);
403+
404+
// Deepest first: the unreadable link, then the child.
405+
const childRunId = (await engine.listRuns('deal_review'))[0].id;
406+
expect(cancel.mock.calls.map((c) => c[0])).toEqual([midRunId, childRunId]);
407+
});
408+
409+
it('CONTROL: an all-readable three-run chain is cancelled whole, deepest first, with nothing at `error`', async () => {
410+
const { engine, logger, cancel } = threeRunRig(new InMemorySuspendedRunStore());
411+
412+
const result: any = await engine.execute('deal_submit', {} as any);
413+
expect(result.success).toBe(false);
414+
expect(await engine.listSuspendedRunsDurable(), 'no run is left parked').toEqual([]);
415+
const ids: string[] = [];
416+
for (const name of ['deal_ask', 'deal_mid', 'deal_review']) {
417+
const runs = await engine.listRuns(name);
418+
expect(runs.map((r: any) => r.status), name).toEqual(['cancelled']);
419+
ids.push(runs[0].id);
420+
}
421+
expect(cancel.mock.calls.map((c) => c[0])).toEqual(ids);
422+
expect(logger.error).not.toHaveBeenCalled();
423+
});
424+
});

0 commit comments

Comments
 (0)