Commit 8a6dbc8c5d93379e9c56da95a9df4dbc4829032d
1 parent
aecf5c9f
feat(graph-engine): advanceBatch 接进 runAuto/advanceTree,修 ok:false 语义 bug(27-P4 T4/T6 收尾)
自动模式的驱动原语 advanceTree 里 advanceOne 换成 advanceBatch:树上「最深 active run」的选取逻辑不变,但推进这一步现在能一次并发 claim 一批 ready 节点, 而不是每 tick 只推一个。核实过 advanceTree/runAuto 的返回值全仓库只被读 .ok, 不依赖 .node/.results 的具体形状,不需要兼容层。 顺带抓到并修了一个真 bug:advanceBatch 的 no-ready-node / taken-by-other 分支 原来写的是 ok:true,跟 advanceOne 用 result() helper 的惯例(这两种情形一律 ok:false)不一致。不修的话接进 runAuto 后,一个已经收敛的 run 会被自动循环 一路空转到 maxSteps 才停——旧测试从没对已收敛的 run 连续调 advanceBatch, 所以这条一直没被撞出来。 新增两条测试:ok:false 语义与 advanceOne 对齐(直接绕开引擎构造边界状态); runAuto/advanceTree 端到端确实吃到并行(用真实 setTimeout 延迟证明并发窗口 重叠,不是同步假象)。全量 vitest 147 文件 / 1350 用例,零回归。 Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Showing
3 changed files
with
92 additions
and
8 deletions
docs/todo/system-design/27-p4-parallel-and-artifacts.md
| @@ -605,6 +605,27 @@ R-19 之后这两件事有实体(`graph_edge_contracts` × `graph_files`), | @@ -605,6 +605,27 @@ R-19 之后这两件事有实体(`graph_edge_contracts` × `graph_files`), | ||
| 605 | > - **没做的**:`advanceBatch` 还没接进 `runAuto`/路由——自动模式今天仍然只调 `advanceOne`, | 605 | > - **没做的**:`advanceBatch` 还没接进 `runAuto`/路由——自动模式今天仍然只调 `advanceOne`, |
| 606 | > 实际生产流量拿不到并行,这是接下来的线(大概率是 T10 之前、或与它同批)。T5(欠费门 + 自动恢复 | 606 | > 实际生产流量拿不到并行,这是接下来的线(大概率是 T10 之前、或与它同批)。T5(欠费门 + 自动恢复 |
| 607 | > 触发器)仍卡在计费系统接口确认,T10(控制台可视化)未动。 | 607 | > 触发器)仍卡在计费系统接口确认,T10(控制台可视化)未动。 |
| 608 | +> | ||
| 609 | +> **2026-08-21 追加:`advanceBatch` 接进 `advanceTree`/`runAuto`**(同一天,紧接上面那批): | ||
| 610 | +> - `advanceTree`(`runAuto` 每一步都调它)里 `advanceOne(r.id, opts)` 换成 `advanceBatch(r.id, opts)`—— | ||
| 611 | +> 「推最深的一个 active run」这个 tree-walk 逻辑不变,变的是「推」这一步本身现在能一次 claim 一批。 | ||
| 612 | +> `.driver` 的调用方(生产路由、测试)一律只读 `.ok`,不依赖 `.node`/`.results` 的具体形状—— | ||
| 613 | +> 核实过全仓库唯一消费方是 `advanceTree` 自己的循环判定 + 两个 blind-spot 测试文件,两者都只用 | ||
| 614 | +> `.ok`,这次改动不需要做返回形状的兼容层。 | ||
| 615 | +> - **审这段时抓到一个真 bug 并顺手修了**:`advanceBatch` 的 `no-ready-node` / `taken-by-other` | ||
| 616 | +> 分支原来写的是 `ok:true`,跟 `advanceOne` 用 `result()` helper 的惯例(这两种情形一律 `ok:false`) | ||
| 617 | +> 不一致。不修的话接进 `runAuto` 后,一个已经收敛的 run 会被自动循环一路空转到 `maxSteps`(1000) | ||
| 618 | +> 才停——不是崩溃,但是真实的效率/语义回归,而且正好是旧测试从没试过「对已收敛的 run 连续调 | ||
| 619 | +> `advanceBatch`」才漏掉的那种 bug。已修好并补了回归用例。 | ||
| 620 | +> - 新增 2 条测试(同一文件):`ok:false` 语义与 `advanceOne` 对齐(直接绕开引擎构造 | ||
| 621 | +> 「`run_status` 还是 running 但没有 ready/running 节点」这个状态,同款手法 `graph-engine.test.js` | ||
| 622 | +> 已经在用);`runAuto`/`advanceTree` 端到端确实吃到并行——用真实 `setTimeout` 延迟(不是同步 | ||
| 623 | +> `deferred`)证明 a/b/c 三个 executor 调用有时间窗口重叠,不是「反正都同步排到一起」的假象。 | ||
| 624 | +> - 全量 `npx vitest run`:147 文件 / 1350 用例全绿,零回归。 | ||
| 625 | +> - **仍然没做的**:`advanceTree` 一次只推「最深的一个」active run,**兄弟子 run**(比如两个并行 | ||
| 626 | +> subgraph 节点各自起的子 run)不会同时被推——它们靠 `runAuto` 外层循环一轮一轮轮流服务, | ||
| 627 | +> 不是真并发(已在 `advanceTree` 加了 ponytail 注释记这笔)。真并发目前只发生在**同一个 run | ||
| 628 | +> 内部**。这个跨 run 兄弟并发是否值得做,留给 T10 之后再看有没有真实场景撞上。 | ||
| 608 | 629 | ||
| 609 | --- | 630 | --- |
| 610 | 631 |
server/services/graph-engine.js
| @@ -1599,7 +1599,11 @@ async function advanceBatch(runId, opts = {}) { | @@ -1599,7 +1599,11 @@ async function advanceBatch(runId, opts = {}) { | ||
| 1599 | .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT ?`) | 1599 | .prepare(`SELECT * FROM graph_run_nodes WHERE run_id = ? AND status = 'ready' ORDER BY created_at, id LIMIT ?`) |
| 1600 | .all(runId, maxConcurrency); | 1600 | .all(runId, maxConcurrency); |
| 1601 | if (!candidates.length) { | 1601 | if (!candidates.length) { |
| 1602 | - return { ok: true, runId, runStatus: settleRun(runId), reason: 'no-ready-node', results: [], activated: [] }; | 1602 | + // ok:false 跟 advanceOne 的 result() 惯例对齐——「这一次调用没有推进任何节点」, |
| 1603 | + // 不是「出错了」。runAuto 的循环靠 !res.ok 判断要不要停,这里如果写成 true, | ||
| 1604 | + // 一个已经收敛的 run 会被 runAuto 一路空转到 maxSteps 才停(§4.3 决策审这段代码时发现, | ||
| 1605 | + // 上线前修,不留到接 runAuto 那一步才炸)。 | ||
| 1606 | + return { ok: false, runId, runStatus: settleRun(runId), reason: 'no-ready-node', results: [], activated: [] }; | ||
| 1603 | } | 1607 | } |
| 1604 | 1608 | ||
| 1605 | const claimed = []; | 1609 | const claimed = []; |
| @@ -1617,8 +1621,9 @@ async function advanceBatch(runId, opts = {}) { | @@ -1617,8 +1621,9 @@ async function advanceBatch(runId, opts = {}) { | ||
| 1617 | } | 1621 | } |
| 1618 | } | 1622 | } |
| 1619 | if (!claimed.length) { | 1623 | if (!claimed.length) { |
| 1620 | - // 候选全被别的 advanceBatch/advanceOne 抢走:不是错误,下一轮再拾取 | ||
| 1621 | - return { ok: true, runId, runStatus: run.run_status, reason: 'taken-by-other', results: [], activated: [] }; | 1624 | + // 候选全被别的 advanceBatch/advanceOne 抢走:不是出错,但同样是「这一次没推进任何东西」, |
| 1625 | + // ok:false 理由同上——taken-by-other 在 advanceOne 里也是 ok:false(见其 result() 用法)。 | ||
| 1626 | + return { ok: false, runId, runStatus: run.run_status, reason: 'taken-by-other', results: [], activated: [] }; | ||
| 1622 | } | 1627 | } |
| 1623 | 1628 | ||
| 1624 | const settled = await Promise.allSettled( | 1629 | const settled = await Promise.allSettled( |
| @@ -3519,18 +3524,25 @@ function hashExport(value) { | @@ -3519,18 +3524,25 @@ function hashExport(value) { | ||
| 3519 | } | 3524 | } |
| 3520 | 3525 | ||
| 3521 | // ──────────────────────────────────────────────────────────────── | 3526 | // ──────────────────────────────────────────────────────────────── |
| 3522 | -// 自动模式 = 不停调 advanceOne 的循环(不是另一套实现) | 3527 | +// 自动模式 = 不停调 advanceBatch 的循环(不是另一套实现) |
| 3523 | // ──────────────────────────────────────────────────────────────── | 3528 | // ──────────────────────────────────────────────────────────────── |
| 3524 | 3529 | ||
| 3525 | /** | 3530 | /** |
| 3526 | * 在一棵 run **子树**里推进一步:优先推最深的 active 子 run(父节点正卡在它上面)。 | 3531 | * 在一棵 run **子树**里推进一步:优先推最深的 active 子 run(父节点正卡在它上面)。 |
| 3527 | - * 单张 run 时退化成 advanceOne(runId)。 | 3532 | + * 单张 run 时退化成 advanceBatch(runId)(27-P4 T4/T6 接入:以前这里调的是 advanceOne, |
| 3533 | + * 一次只推一个节点;现在同一个「最深 active run」内部能一次并发跑多个 ready 节点)。 | ||
| 3528 | * | 3534 | * |
| 3529 | * **形参是子树根,不要求是 root run**(R6-F3 问题 2)。控制台「进入子图」后那一层的五个 | 3535 | * **形参是子树根,不要求是 root run**(R6-F3 问题 2)。控制台「进入子图」后那一层的五个 |
| 3530 | * 控制按钮作用在子 run 上(§六 验收 15),对子 run 点自动启动时 runAuto 调的就是 | 3536 | * 控制按钮作用在子 run 上(§六 验收 15),对子 run 点自动启动时 runAuto 调的就是 |
| 3531 | * advanceTree(子 runId)。旧实现直接 `listRunTree(传入值)` —— 那是按 `root_run_id = ?` 查的, | 3537 | * advanceTree(子 runId)。旧实现直接 `listRunTree(传入值)` —— 那是按 `root_run_id = ?` 查的, |
| 3532 | * 传子 run 拿到**零行**,循环第 0 步就 break:HTTP 回 `ok:true / auto:true`, | 3538 | * 传子 run 拿到**零行**,循环第 0 步就 break:HTTP 回 `ok:true / auto:true`, |
| 3533 | * 而那个 run 里的 ready 节点纹丝不动(返回 ok 却什么都不做,是最坏的形态)。 | 3539 | * 而那个 run 里的 ready 节点纹丝不动(返回 ok 却什么都不做,是最坏的形态)。 |
| 3540 | + * | ||
| 3541 | + * ponytail: 仍然只推**一个**「最深 active run」,同一 tick 里的**兄弟**子 run(比如两个并行 | ||
| 3542 | + * subgraph 节点各自起的子 run)不会同时被推——它们靠 runAuto 的外层循环一轮一轮轮流服务, | ||
| 3543 | + * 不是真并发。真并发只发生在**同一个 run 内部**(advanceBatch 的 fan-out)。上限是「兄弟子 run | ||
| 3544 | + * 之间」这一维的吞吐,不影响正确性(不会有子 run 被饿死,只是不同时跑);升级路径是把这个 | ||
| 3545 | + * for 循环也从「找到第一个 ok 就 return」改成对多个 active run 并发调 advanceBatch。 | ||
| 3534 | */ | 3546 | */ |
| 3535 | async function advanceTree(runId, opts = {}) { | 3547 | async function advanceTree(runId, opts = {}) { |
| 3536 | const self = getRun(runId); | 3548 | const self = getRun(runId); |
| @@ -3543,7 +3555,7 @@ async function advanceTree(runId, opts = {}) { | @@ -3543,7 +3555,7 @@ async function advanceTree(runId, opts = {}) { | ||
| 3543 | // 后建的 run 一定更深(子 run 总是在父节点 running 之后建的),倒序即「最深优先」 | 3555 | // 后建的 run 一定更深(子 run 总是在父节点 running 之后建的),倒序即「最深优先」 |
| 3544 | let informative = null; | 3556 | let informative = null; |
| 3545 | for (const r of tree.reverse()) { | 3557 | for (const r of tree.reverse()) { |
| 3546 | - const res = await advanceOne(r.id, opts); | 3558 | + const res = await advanceBatch(r.id, opts); |
| 3547 | if (res.ok) return res; | 3559 | if (res.ok) return res; |
| 3548 | // minor 3:ok:false 时**不能**把所有原因都塌缩成 'no-ready-node'。子层返回 | 3560 | // minor 3:ok:false 时**不能**把所有原因都塌缩成 'no-ready-node'。子层返回 |
| 3549 | // 'budget-exhausted'(还带着 budget.exhausted 明细)而树层报「没活可干」, | 3561 | // 'budget-exhausted'(还带着 budget.exhausted 明细)而树层报「没活可干」, |
| @@ -3564,8 +3576,8 @@ async function advanceTree(runId, opts = {}) { | @@ -3564,8 +3576,8 @@ async function advanceTree(runId, opts = {}) { | ||
| 3564 | * 自动驱动:只要 pause_requested / stop_requested 都是假就在上一次 resolve 后立刻发起下一次。 | 3576 | * 自动驱动:只要 pause_requested / stop_requested 都是假就在上一次 resolve 后立刻发起下一次。 |
| 3565 | * 与 advanceTree 同款语义 —— runId 是**子树根**,对子 run 起自动就只驱动它这棵子树。 | 3577 | * 与 advanceTree 同款语义 —— runId 是**子树根**,对子 run 起自动就只驱动它这棵子树。 |
| 3566 | * maxSteps 只是防跑飞的安全阀,真正的终止条件是预算(§四.A.3)与熔断(§四.D.2.1)。 | 3578 | * maxSteps 只是防跑飞的安全阀,真正的终止条件是预算(§四.A.3)与熔断(§四.D.2.1)。 |
| 3567 | - * ponytail: 串行单步循环,一次只跑一个节点,并行分支不会并发执行;上限是吞吐, | ||
| 3568 | - * 升级路径是按 ready 集合并发 advanceOne(节点状态机已经是幂等条件更新,能直接支撑)。 | 3579 | + * 27-P4 T4/T6 接入后,「同一个 run 内部」已经能并发;跨兄弟子 run 的并发仍是 advanceTree |
| 3580 | + * 那条 ponytail(一次只推最深的一个),不在本函数里。 | ||
| 3569 | */ | 3581 | */ |
| 3570 | async function runAuto(runId, opts = {}) { | 3582 | async function runAuto(runId, opts = {}) { |
| 3571 | const maxSteps = opts.maxSteps || 1000; | 3583 | const maxSteps = opts.maxSteps || 1000; |
server/tests/unit/graph-engine-advance-batch.test.js
| @@ -231,4 +231,55 @@ describe('advanceBatch(27-P4 T4/T6)', () => { | @@ -231,4 +231,55 @@ describe('advanceBatch(27-P4 T4/T6)', () => { | ||
| 231 | for (const g of Object.values(gates)) g.resolve(); | 231 | for (const g of Object.values(gates)) g.resolve(); |
| 232 | await batchPromise; | 232 | await batchPromise; |
| 233 | }); | 233 | }); |
| 234 | + | ||
| 235 | + it('ok:false 语义与 advanceOne 对齐:no-ready-node 不算「推进了」', async () => { | ||
| 236 | + // no-ready-node:直接绕开引擎把节点写成 success(不经过 settleNode,run_status 就不会 | ||
| 237 | + // 被顺带收敛),构造出「run_status 还是 running,但已经没有 ready/running 节点」这个 | ||
| 238 | + // advanceOne 的 !picked 分支本来就要处理的状态。真实场景里这一般是重入/竞态的产物—— | ||
| 239 | + // 这里直接摆出结果状态,不追究怎么走到的(同款手法 graph-engine.test.js 234 行已经在用: | ||
| 240 | + // `wdb.updateNodeStatus(a.id, 'ready', 'running')` 绕开引擎构造中间状态)。 | ||
| 241 | + const companyId1 = nextCompany(); | ||
| 242 | + ensureAgents(['solo']); | ||
| 243 | + const g1 = wdb.createWorkGraph({ companyId: companyId1, nodes: ['solo'], edges: [] }); | ||
| 244 | + const run1 = wdb.createRun({ graphKind: 'work', graphId: g1.id, companyId: companyId1, executionMode: 'manual' }); | ||
| 245 | + engine.markRunReady(run1.id); | ||
| 246 | + engine.startRun(run1.id); | ||
| 247 | + const solo1 = nodeByName(run1.id, 'solo'); | ||
| 248 | + expect(wdb.updateNodeStatus(solo1.id, 'ready', 'running')).toBe(1); | ||
| 249 | + expect(wdb.updateNodeStatus(solo1.id, 'running', 'success')).toBe(1); | ||
| 250 | + | ||
| 251 | + // advanceOne 的 result() 惯例是 ok:false——advanceBatch 必须跟它一致,否则接进 runAuto | ||
| 252 | + // 之后,一个已经收敛的 run 会被循环一路空转到 maxSteps 才停(2026-08-21 审 §4.3 时抓到的 | ||
| 253 | + // 真 bug,见 advanceBatch 里 no-ready-node 分支的注释)。 | ||
| 254 | + const drained = await engine.advanceBatch(run1.id, { maxConcurrency: 3 }); | ||
| 255 | + expect(drained.ok).toBe(false); | ||
| 256 | + expect(drained.reason).toBe('no-ready-node'); | ||
| 257 | + // 顺带验证:这一次调用自己把 run 收敛掉了(settleRun 就挂在这个分支里) | ||
| 258 | + expect(wdb.getRun(run1.id).run_status).toBe('completed'); | ||
| 259 | + | ||
| 260 | + // taken-by-other(!claimed.length 分支)分支上同一惯例——ok:false,不额外起用例: | ||
| 261 | + // 单进程 + 同步拾取循环下这个分支在真实调用序列里不可达(§4.1 决策的论证本身就是 | ||
| 262 | + // 「单行条件更新在单线程下不会输给同进程的另一次调用」),要触发它得在 SELECT 和 | ||
| 263 | + // claim 之间人为插入另一次写,等价于对内部实现打桩,价值不大,不为了覆盖率硬凑。 | ||
| 264 | + }); | ||
| 265 | + | ||
| 266 | + it('runAuto/advanceTree 端到端也吃到并行(不是只有直接调 advanceBatch 才有效)', async () => { | ||
| 267 | + const { runId } = buildDiamond(); // max_concurrency=null → 默认 DEFAULT_MAX_CONCURRENCY=5 | ||
| 268 | + let active = 0; | ||
| 269 | + let maxActive = 0; | ||
| 270 | + engine.setAgentExecutor(async () => { | ||
| 271 | + active += 1; | ||
| 272 | + maxActive = Math.max(maxActive, active); | ||
| 273 | + // 真实的异步延迟(不是同步卡在 deferred 上):要证明的是 runAuto 的循环驱动下 | ||
| 274 | + // a/b/c 三个 executor 调用有时间窗口上的重叠,不是「反正都同步排到一起」的假象。 | ||
| 275 | + await new Promise((r) => { setTimeout(r, 15); }); | ||
| 276 | + active -= 1; | ||
| 277 | + return { status: 'success' }; | ||
| 278 | + }); | ||
| 279 | + | ||
| 280 | + await engine.startRun(runId, { mode: 'auto' }).driver; | ||
| 281 | + | ||
| 282 | + expect(wdb.getRun(runId).run_status).toBe('completed'); | ||
| 283 | + expect(maxActive).toBeGreaterThanOrEqual(3); // entry 之后 a/b/c 应该同时在跑,而不是依次 | ||
| 284 | + }); | ||
| 234 | }); | 285 | }); |