mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
refactor(managed-agent): Extract runtime execution dispatcher (#16084)
Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
jinyeandQwen-Coder authored and GitHub committed
1 parent
e3c7a6d7ef
commit
87c1918a6a
10 files changed
+960
-480
No files matched your search
@@ -0,0 +1,76 @@
|
||||
# Runtime execution dispatcher ownership
|
||||
|
||||
[English](2026-11-01-runtime-execution-dispatcher.md) | [简体中文](2026-11-01-runtime-execution-dispatcher.zh-CN.md)
|
||||
|
||||
- Status: implemented locally; focused verification passed; full CI acceptance pending.
|
||||
- Date: 2026-11-01
|
||||
- Baseline: `2aa908dc70785e38299a43964e064dbafaf7b878`, after #16029.
|
||||
|
||||
## Problem and current state
|
||||
|
||||
The execution workflows have explicit owners, but `RuntimeBrokerService` still owns dispatch reservations, physical invocations, dispatch-lease renewal and completion. Cancellation receives the invocation set and a per-call service callback. Moving just the transport call would leave the state and cleanup decisions split between owners.
|
||||
|
||||
This increment extracts one package-private `RuntimeExecutionDispatcher` inside the existing runtime-broker module. It preserves the current algorithms, admission boundaries, synchronous exceptions, asynchronous completion and shutdown order. It introduces no public API, HTTP route, SQL/schema, configuration, dependency or product behavior change.
|
||||
|
||||
The earlier full-tier run for #16029 contains an unresolved `WorkspaceCsiHostedFilesTest` cold-recovery failure before its MySQL integration phase. Preserve that evidence separately. Developing this extraction does not establish that the baseline failure is fixed or that the merged refactor has complete integration acceptance.
|
||||
|
||||
## Ownership and consumers
|
||||
|
||||
| Owner | Responsibility | Direct consumers |
|
||||
| ------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------- |
|
||||
| `RuntimeExecutionDispatcher` | Dispatch reservations, physical invocation membership, claim-to-EXECUTING transition, dispatch renewal, transport invocation, v3 result waiting and persisted SSH acknowledgement | Service execution/start and drain paths; cancellation |
|
||||
| `RuntimeExecutionState` | Existing durable execution ownership, fenced settlement, UNKNOWN conversion and background-sibling settlement | Dispatcher and existing execution workflows |
|
||||
| `RuntimeExecutionCancellation` | Durable cancellation intent, physical cancellation and prepared-provider confirmation | Service cancellation facade |
|
||||
| `RuntimeBindingRegistry` | Existing local Session contexts, route identity and Binding invalidation | Service and execution collaborators |
|
||||
| `RuntimeBrokerSupport` | Existing shared rules plus unchanged detached/no-start receipt projections and renewal-delay calculation | Service and dispatcher |
|
||||
| `RuntimeBrokerService` | Public admission and payload/grant validation, provisioning, Session/drain/stop orchestration, shared close flag, both executors and provisioner lifecycle | Existing adapters and assembly |
|
||||
|
||||
The dispatcher borrows the existing repositories, Registry, execution state, provisioner, transport, publication verifier, clock, lease/window settings, executors and close flag. Dependencies remain explicit constructor arguments; no general context container, interface or Spring bean is added.
|
||||
|
||||
The complete direct-consumer audit is the service's creation and two start families, its drain acknowledgement sweep and close sequence, cancellation's two invocation checks and unsent-dispatch branch, and `BrokerServiceTestAccess.pollDelay`. Other service receipt paths retain the shared `sshNoStartReceipt` and `isDetachedCapture` projections through Support. Repository and transport contracts remain unchanged.
|
||||
|
||||
## Dispatch admission and execution
|
||||
|
||||
Move all three `beginDispatch` overloads, `dispatch`, `enterExecuting`, reference/capture projection and `DispatchRenewal` together. Private CSI restrictions, Harness admission and deferred-payload refusal stay in their original order before reservation acquisition. Existing Session-context lock regions remain at callers; the dispatcher adds no lock around admission or dispatch.
|
||||
|
||||
The reservation map uses `putIfAbsent` and conditional `remove(executionId, exactFuture)` on synchronous failure and asynchronous completion. It covers claim/fencing work as well as physical invocation. The distinct invocation set starts only immediately before calling transport, after renewal starts. Cancellation observes that set through `isInvoking(executionId)` rather than receiving mutable state.
|
||||
|
||||
Preserve claim, fenced EXECUTING transition, owner/generation checks, usable-lease validation, renewal start and transport invocation in that order. A cancelled unsent execution still settles through the existing dispatcher path. Do not replace it with a new cancellation settlement algorithm or replay UNKNOWN work.
|
||||
|
||||
On completion, remove the physical invocation marker before persisting either the settled result or UNKNOWN outcome. Preserve background-sibling settlement and best-effort SSH acknowledgement after the durable write. A transport/null-result or ordinary settlement failure retains the existing UNKNOWN mapping. Synchronous `RuntimeBrokerException` values remain unchanged; other synchronous runtime/JVM failures retain `runtime_execution_dispatch_failed` wrapping and reservation cleanup.
|
||||
|
||||
## V3 results and SSH acknowledgement
|
||||
|
||||
V3 result waiting is part of the dispatch call closure and moves unchanged. Keep publication-first evidence reads, dispatch/Binding identity checks, lease/deadline refusal, explicit `not_started`, detached results, non-retryable error handling and 100 ms doubling backoff capped at 2 s. The test-only backoff accessor points to the new owner.
|
||||
|
||||
Move `acknowledgePersistedTools` into the dispatcher because dispatch completion and the service's drain sweep both consume it. Drain and physical Runtime stop remain service orchestration. Move the shared stateless `sshNoStartReceipt`, `isDetachedCapture` and `renewalDelayMillis` helpers to existing Support; do not duplicate their policies or introduce another collaborator.
|
||||
|
||||
## Cancellation and shutdown
|
||||
|
||||
Cancellation receives the concrete dispatcher. Replace the raw invocation-set reads with `isInvoking` and the `BiConsumer` callback with `beginDispatch(context, requested)`. Keep ownership reads, durable intent, admission refusals, control guards and physical cancellation order unchanged.
|
||||
|
||||
The service remains the sole writer of `closed` and the owner of both executors and the provisioner. Its shutdown sequence stays: set the close fence; cancel Registry operations; cancel dispatcher reservation futures; cancel reconciler operations; shut down coordination and renewal executors; close the provisioner.
|
||||
|
||||
`cancelPendingOperations` cancels only the existing reservation-wrapper futures. It does not cancel underlying transport stages, clear physical invocation markers or shut down borrowed resources. Do not add a new close check inside `beginDispatch`; changing the handling of already-admitted operations is outside this extraction.
|
||||
|
||||
## Files and migration
|
||||
|
||||
Add `RuntimeExecutionDispatcher`; update service construction/delegation, cancellation, shared Support and the test-only backoff accessor. Add focused dispatcher tests and retain the existing service, HTTP, SSH and protocol regressions. Link this design from both execution-workflow documents and update their current ownership description together.
|
||||
|
||||
Implement after the baseline dry-run: move state and the complete dispatch closure; wire the dispatcher before cancellation; replace all direct consumers; update tests and both design languages. Keep migration mechanical and review the complete diff for accidental algorithm or exception-boundary changes.
|
||||
|
||||
## Verification and acceptance
|
||||
|
||||
Follow `packages/managed-agent-server/AGENTS.md`, including the shared 10-minute local verification budget, JDK 21, matching source-built SDK and isolated Maven repository. An internal Java extraction has no meaningful global-CLI behavioral probe: record the installed `qwen --version` as environment evidence only, and establish the baseline through the existing Java fixtures. No model call or JavaScript build is needed for those selected fixtures.
|
||||
|
||||
The focused plan is recorded under `.qwen/e2e-tests/runtime-execution-dispatcher.md`. Pin reservation deduplication, synchronous-failure cleanup, invocation removal before settlement/UNKNOWN and shutdown that preserves the underlying invocation and borrowed resources. Reuse `RuntimeBrokerServiceTest`, `RuntimeSshAcknowledgementTest`, `Issue13183RegressionTest` and `RuntimeBrokerHttpServerTest` for cancellation/fencing, v3/private CSI, receipt and HTTP integration behavior. Run compilation, both Checkstyle goals and the selected tests; check actual report counts and skips.
|
||||
|
||||
The baseline passed 280 selected tests. After extraction, the same suites plus five dispatcher cases passed 285 tests with no failures, errors or skips; both Checkstyle goals passed. Verification used JDK 21 on macOS with the source-matching SDK and isolated Maven repository. Setup, the initial unused-import Checkstyle failure and the corrected run used 252.587 seconds of the shared local budget. The cancellation witness proves preservation of the physical invocation and borrowed resources; exact wrapper cancellation is checked by source equivalence rather than a private-state test accessor.
|
||||
|
||||
Acceptance requires one dispatch-state owner; no cancellation callback or raw invocation-set injection; unchanged public contracts/schema and dispatch algorithms; preserved close/completion order; focused regressions passing; synchronized bilingual docs; and two clean self-audit passes. Full MySQL 8.4, bundled Hosted/fault, cross-platform and full SpotBugs gates remain CI work. Local fixtures do not establish those results, and an isolated pass does not clear the earlier wider failure.
|
||||
|
||||
## Risks and open questions
|
||||
|
||||
The material risks are moving the invocation marker relative to settlement, changing reservation cleanup, losing original-generation routing, changing v3 deadline/receipt precedence, or taking over executor/transport cancellation. Preserve method bodies and caller lock envelopes, then exercise the existing witnesses and the new owner-focused tests.
|
||||
|
||||
There is no open product decision. Further provisioning, Session lifecycle or persistence extraction is outside this increment. The unresolved baseline cold-recovery failure remains a separate investigation and acceptance gap.
|
||||
@@ -0,0 +1,76 @@
|
||||
# Runtime 执行分发器状态归属
|
||||
|
||||
[English](2026-11-01-runtime-execution-dispatcher.md) | [简体中文](2026-11-01-runtime-execution-dispatcher.zh-CN.md)
|
||||
|
||||
- 状态：本地已实现，定向验证通过；完整 CI 验收待完成。
|
||||
- 日期：2026-11-01
|
||||
- 基线：`2aa908dc70785e38299a43964e064dbafaf7b878`，位于 #16029 之后。
|
||||
|
||||
## 问题与当前状态
|
||||
|
||||
执行流程已有明确持有者，但 `RuntimeBrokerService` 仍持有分发预留项、实际在途调用、分发租约续期及完成处理。取消流程接收在途集合和逐调用传入的 service 回调。如果只搬运 transport 调用，状态和清理决策仍会分散在多个持有者中。
|
||||
|
||||
本增量在现有 runtime-broker 模块内抽取一个 package-private 的 `RuntimeExecutionDispatcher`。保留当前算法、准入边界、同步异常、异步完成及关闭顺序，不修改公共 API、HTTP 路由、SQL/schema、配置、依赖或产品行为。
|
||||
|
||||
#16029 的早期完整门禁中，`WorkspaceCsiHostedFilesTest` 冷恢复用例在该任务进入 MySQL 集成阶段前失败，尚未完成根因定位。单独保留这份证据。开发本次抽取不代表已修复基线失败，也不代表已合并重构完成了完整集成验收。
|
||||
|
||||
## 归属与消费方
|
||||
|
||||
| 持有者 | 职责 | 直接消费方 |
|
||||
| ------------------------------ | --------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------- |
|
||||
| `RuntimeExecutionDispatcher` | 分发预留项、实际 invocation 集合、claim 到 EXECUTING 的转换、分发续租、transport 调用、v3 结果等待和持久 SSH 回执确认 | service 的 execution/start 与 drain 路径；取消流程 |
|
||||
| `RuntimeExecutionState` | 已有持久执行归属、带 fencing 的结算、UNKNOWN 转换和后台同胞行结算 | 分发器及现有执行流程 |
|
||||
| `RuntimeExecutionCancellation` | 持久取消意图、物理取消和 prepared provider 确认 | service 取消门面 |
|
||||
| `RuntimeBindingRegistry` | 已有本地 Session context、路由身份和 Binding 失效 | service 与执行协作者 |
|
||||
| `RuntimeBrokerSupport` | 已有共享规则，以及不变的 detached/no-start 回执投影和续租间隔计算 | service 与分发器 |
|
||||
| `RuntimeBrokerService` | 公共准入及 payload/grant 校验、供应、Session/drain/stop 编排、共享关闭屏障、两个 executor 及 provisioner 生命周期 | 现有 adapter 与装配 |
|
||||
|
||||
分发器借用现有 repository、Registry、执行状态、provisioner、transport、publication verifier、clock、租约/窗口参数、executor 和关闭屏障。依赖保持为显式构造参数，不增加通用 context 容器、接口或 Spring bean。
|
||||
|
||||
完整直接消费方清单为：service 的创建与两个 start 入口族、drain 确认扫描和关闭顺序；取消流程的两次 invocation 检查与未发送分发分支；测试辅助入口 `BrokerServiceTestAccess.pollDelay`。service 的其他回执路径通过 Support 继续使用共享的 `sshNoStartReceipt` 与 `isDetachedCapture` 投影。Repository 与 transport 契约保持不变。
|
||||
|
||||
## 分发准入与执行
|
||||
|
||||
将三个 `beginDispatch` 重载、`dispatch`、`enterExecuting`、reference/capture 投影及 `DispatchRenewal` 一起迁移。私有 CSI 限制、Harness 准入和 deferred payload 拒绝保留原先顺序，均在取得预留项之前执行。现有 Session context 锁区域继续位于调用方；分发器不为准入或分发新增锁。
|
||||
|
||||
预留 map 使用 `putIfAbsent`，同步失败和异步完成时使用 `remove(executionId, exactFuture)` 条件移除。它覆盖 claim/fencing，也覆盖实际调用。独立 invocation 集合只在续租开始后、调用 transport 前登记。取消通过 `isInvoking(executionId)` 观察该集合，不再接收可变集合。
|
||||
|
||||
保留 claim、带 fencing 的 EXECUTING 转换、owner/代际校验、可用租约校验、启动续租和 transport 调用的顺序。已取消但尚未发送的执行继续经原分发路径结算，不使用新的取消结算算法，也不重放 UNKNOWN 工作。
|
||||
|
||||
完成时先移除实际 invocation 标记，再持久化终态结果或 UNKNOWN。保留后台同胞行结算，以及持久写入之后 best-effort 的 SSH 确认。transport 失败、空结果或普通结算异常继续映射为 UNKNOWN。同步 `RuntimeBrokerException` 原样保留；其他同步运行时/JVM 失败保留 `runtime_execution_dispatch_failed` 包装及预留项清理。
|
||||
|
||||
## V3 结果与 SSH 确认
|
||||
|
||||
v3 结果等待属于分发调用闭包，原样迁移。保留 publication 证据优先读取、dispatch/Binding 身份校验、租约/截止时间拒绝、显式 `not_started`、detached 结果、不可重试错误处理，以及从 100 ms 翻倍、上限 2 s 的退避。仅供测试的退避访问器改为引用新持有者。
|
||||
|
||||
将 `acknowledgePersistedTools` 迁入分发器，因为分发完成和 service 的 drain 扫描都消费它。drain 与物理 Runtime stop 仍由 service 编排。共享且无状态的 `sshNoStartReceipt`、`isDetachedCapture` 和 `renewalDelayMillis` 迁入已有 Support，不复制规则，也不新增协作者。
|
||||
|
||||
## 取消与关闭
|
||||
|
||||
取消接收具体分发器，将直接读取 invocation 集合改为 `isInvoking`，将 `BiConsumer` 回调改为 `beginDispatch(context, requested)`。归属读取、持久意图、准入拒绝、control guard 和物理取消顺序保持不变。
|
||||
|
||||
service 继续是 `closed` 的唯一写入方，并持有两个 executor 和 provisioner。关闭顺序保持为：设置关闭屏障；取消 Registry 操作；取消分发预留 future；取消对账操作；关闭协调与续租 executor；关闭 provisioner。
|
||||
|
||||
`cancelPendingOperations` 只取消已有预留包装 future，不取消底层 transport stage，不清空实际 invocation 标记，也不关闭借用资源。不给 `beginDispatch` 增加新的关闭检查；改变已准入操作的处理方式超出本次抽取范围。
|
||||
|
||||
## 文件与迁移
|
||||
|
||||
新增 `RuntimeExecutionDispatcher`；更新 service 构造与委托、取消流程、共享 Support，以及仅供测试的退避访问器。新增分发器定向测试，保留现有 service、HTTP、SSH 和协议回归。为执行流程文档的两种语言加入本设计链接，同步更新当前状态归属说明。
|
||||
|
||||
基线 dry-run 后开始实现：迁移状态与完整分发闭包；在取消流程之前装配分发器；替换全部直接消费方；更新测试及双语设计。保持机械迁移，审阅完整 diff，检查是否意外改变算法或异常边界。
|
||||
|
||||
## 验证与验收
|
||||
|
||||
遵循 `packages/managed-agent-server/AGENTS.md`，包括共享的 10 分钟本地验证预算、JDK 21、匹配源码构建的 SDK 和隔离 Maven 仓库。内部 Java 抽取没有有效的全局 CLI 行为探针：只记录已安装 `qwen --version` 作为环境证据，通过现有 Java fixture 建立基线。所选 fixture 不需要模型调用或 JavaScript 构建。
|
||||
|
||||
定向计划记录在 `.qwen/e2e-tests/runtime-execution-dispatcher.md`。固定预留去重、同步失败清理、在结算/UNKNOWN 之前移除 invocation，以及关闭时保留底层调用与借用资源。复用 `RuntimeBrokerServiceTest`、`RuntimeSshAcknowledgementTest`、`Issue13183RegressionTest` 和 `RuntimeBrokerHttpServerTest` 覆盖取消/fencing、v3/私有 CSI、回执及 HTTP 集成行为。执行编译、两组 Checkstyle 和选定测试，检查实际报告数量及跳过情况。
|
||||
|
||||
基线选定的 280 个测试通过。抽取后，同一批测试加上五个分发器用例，共 285 个测试通过，无失败、错误或跳过；两组 Checkstyle 均通过。验证在 macOS 上使用 JDK 21、匹配源码的 SDK 和隔离 Maven 仓库。环境准备、首次无用 import 的 Checkstyle 失败及修正后验证共使用本地共享预算 252.587 秒。取消用例证明实际调用及借用资源得到保留；包装 future 的准确取消动作通过源码等价核对确认，不新增访问私有状态的测试入口。
|
||||
|
||||
验收要求：分发状态只有一个持有者；取消没有 service 回调或可变 invocation 集合注入；公共契约/schema 与分发算法不变；关闭/完成顺序不变；定向回归通过；双语设计同步；两轮干净自审。完整 MySQL 8.4、打包 Hosted/fault、跨平台及完整 SpotBugs 门禁继续由 CI 负责。本地 fixture 不构成这些结果，单独一次通过也不能消除此前更广范围的失败。
|
||||
|
||||
## 风险与待定问题
|
||||
|
||||
主要风险包括 invocation 标记与结算顺序改变、预留清理改变、丢失原代际路由、改变 v3 截止时间/回执优先级，以及接管 executor/transport 的取消。保留方法实现与调用方锁区域，再使用现有行为测试和新增持有者测试验证。
|
||||
|
||||
没有待定产品决策。进一步的供应、Session 生命周期或持久化抽取不在本增量内。未解决的基线冷恢复失败继续作为独立排查项和验收缺口保留。
|
||||
@@ -10,7 +10,9 @@
|
||||
|
||||
The Registry gives Binding and Session state an owner, but execution reconciliation, takeover scans and cancellation still live in `RuntimeBrokerService`. Their shared execution reads, fenced settlement and background-hold settlement make a three-way copy of private helpers unsafe. Extract the three workflows with direct module-internal collaborators and one shared execution-state boundary. Preserve current observable behavior, including failure timing, async continuations, writer fencing and cleanup order.
|
||||
|
||||
No public interface, HTTP route, schema, protocol, dependency or Runtime launch policy changes. Provisioning, Binding recovery/drain, dispatch, renewal scheduling and physical Session release stay in the service. This is structural work; full integration acceptance remains separate from implementation completion.
|
||||
No public interface, HTTP route, schema, protocol, dependency or Runtime launch policy changes. Provisioning, Binding recovery/drain, Binding renewal scheduling and physical Session release stay in the service. This is structural work; full integration acceptance remains separate from implementation completion.
|
||||
|
||||
This document records the workflow extraction merged in #16029. The [Dispatcher follow-up](2026-11-01-runtime-execution-dispatcher.md) moves dispatch reservations, invocations and completion together and replaces the original cancellation callback. The ownership table and cancellation wiring below reflect that follow-up.
|
||||
|
||||
## Ownership and dependencies
|
||||
|
||||
@@ -21,8 +23,9 @@ No public interface, HTTP route, schema, protocol, dependency or Runtime launch
|
||||
| `RuntimeExecutionReconciler` | Execution-ID lookup reservations, `ExecutionObservationCooldown`, evidence validation and reconciliation CAS | Service observation endpoints and takeover scanner |
|
||||
| `RuntimeExecutionRecovery` | Bounded takeover scan and background-hold observation/sweep workflows | Acquire/adoption, routine/drain release, background observation and task stop |
|
||||
| `RuntimeExecutionCancellation` | Durable cancellation intent and physical cancellation/confirmation workflow | Service cancellation endpoint |
|
||||
| `RuntimeBrokerSupport` | Unchanged stateless future/error mapping, payload/reference validation and status projection | The extracted collaborators and service |
|
||||
| Service | Public facade, `closed`, dispatch reservations/invocations, schedulers, provisioner lifetime, Binding/Session lifecycle orchestration | Existing adapters and assembly |
|
||||
| `RuntimeExecutionDispatcher` | Dispatch reservations/invocations, claim transition, dispatch renewal, v3 result waiting, completion and persisted SSH acknowledgement | Service execution/start and drain paths; cancellation |
|
||||
| `RuntimeBrokerSupport` | Unchanged stateless future/error mapping, payload/reference validation, status/receipt projection and renewal-delay calculation | The extracted collaborators and service |
|
||||
| Service | Public facade, `closed`, schedulers, provisioner lifetime, Binding/Session lifecycle orchestration | Existing adapters and assembly |
|
||||
|
||||
All classes remain package-private in the existing runtime-broker service package; no new interface or Spring wiring is needed. Repositories remain the sole durable authority. Context locks still belong to the Registry, and every compound read/admit/release decision retains its current lock region. The support class shares existing rules across several real callers; it is not a configurable workflow framework.
|
||||
|
||||
@@ -48,7 +51,7 @@ Task stop keeps tenant/Harness ownership validation and the original Session rou
|
||||
|
||||
## Cancellation design
|
||||
|
||||
Move cancellation admission, prepared-provider confirmation, physical cancel and cancellation-result validation into `RuntimeExecutionCancellation`. It uses Registry, execution state, execution repository, transport, the existing invocation set, coordination scheduler, cancellation duration and close fence. The service facade supplies its existing two-argument dispatch action for the one DISPATCHING branch; this per-call `BiConsumer` is the only service action passed to an extracted workflow. Do not store a service reference or pass maps/private lifecycle helpers as callbacks.
|
||||
Move cancellation admission, prepared-provider confirmation, physical cancel and cancellation-result validation into `RuntimeExecutionCancellation`. It uses Registry, execution state, execution repository, transport, the concrete dispatcher, coordination scheduler, cancellation duration and close fence. Cancellation queries `isInvoking` and calls `beginDispatch` for the unsent DISPATCHING branch. It receives neither a mutable invocation set nor a per-call service callback. Do not store a service reference or pass maps/private lifecycle helpers as callbacks.
|
||||
|
||||
Preserve terminal-receipt ownership reads before requiring a live Session. A prepared provider's cancelled receipt still requires original-worker confirmation when that worker may retain the preparation. Nonterminal cancellation persists intent before sending a worker request. Hold rows and authorized private CSI calls retain their existing refusals. An invocation still running here may receive cancellation after lease lapse; an invocation absent here first follows the repository-clock dispatch fence. UNKNOWN remains UNKNOWN until reconciliation evidence resolves it.
|
||||
|
||||
@@ -68,10 +71,10 @@ Reuse the existing service regressions for explicit versus automatic lookup, tim
|
||||
|
||||
Within the module's shared 10-minute local verification budget, reuse the source-matching SDK and JDK 21/Maven environment, run affected Java suites with Checkstyle and the minimum subprocess path, and inspect the actual selected reports. Full MySQL 8.4, bundled Hosted/fault recovery and SpotBugs remain required CI qualification; fixture/H2 or test-worker subprocess passes do not establish those results. No JavaScript build is required because this change has no TypeScript/shared-contract change.
|
||||
|
||||
Acceptance: all three facade entry families delegate to their named collaborators; lookup reservations and cooldown share one owner; no collaborator reaches back into service internals; the sole cancellation dispatch action is explicit; no SQL/schema/public-contract diff; moved algorithms retain their semantics; the ordering witness and focused regressions pass; bilingual documents and two clean self-audit passes are complete. CI acceptance is pending until its required gates run on the accepted revision.
|
||||
Acceptance: all three facade entry families delegate to their named collaborators; lookup reservations and cooldown share one owner; no collaborator reaches back into service internals; cancellation uses the dispatcher without a service callback or raw invocation set; no SQL/schema/public-contract diff; moved algorithms retain their semantics; the ordering witness and focused regressions pass; bilingual documents and two clean self-audit passes are complete. CI acceptance is pending until its required gates run on the accepted revision.
|
||||
|
||||
## Risks and open questions
|
||||
|
||||
The main risks are altered synchronous exception mapping, completion ordering, guard accounting and cross-generation routing. Mechanical moves and unchanged call-site envelopes reduce these risks, while the existing integration tests exercise them through the public service facade. Shared immutable dependencies are passed directly rather than hidden in a generic context object.
|
||||
|
||||
The callback exception is deliberately limited to cancellation's unsent-dispatch branch; extracting the entire dispatcher would widen this task. No product decisions remain open. A later dispatcher extraction may replace that action with a concrete collaborator without changing cancellation semantics.
|
||||
The original #16029 slice deliberately limited its callback exception to cancellation's unsent-dispatch branch. The Dispatcher follow-up replaces it with a concrete collaborator while preserving cancellation semantics and moving dispatch state and completion together. No product decisions remain open. Further provisioning or Session lifecycle extraction is outside this increment.
|
||||
@@ -10,19 +10,22 @@
|
||||
|
||||
Registry 明确了 Binding 与 Session 状态持有者，但执行对账、接管扫描与取消仍在 `RuntimeBrokerService` 中。它们共享执行读取、带 fencing 的结算和后台 hold 结算，不能安全地把私有 helper 复制三遍。用直接依赖的模块内部协作者拆出三个流程，并建立一个共享执行状态边界。保留当前可观察行为，包括失败时机、异步 continuation、writer fencing 和清理顺序。
|
||||
|
||||
不修改公共接口、HTTP 路由、schema、协议、依赖或 Runtime 启动策略。供应、Binding 恢复/排空、分发、续租调度和物理 Session 释放仍在服务中。这是结构调整；完整集成验收与实现完成分别记录。
|
||||
不修改公共接口、HTTP 路由、schema、协议、依赖或 Runtime 启动策略。供应、Binding 恢复/排空、Binding 续租调度和物理 Session 释放仍在服务中。这是结构调整；完整集成验收与实现完成分别记录。
|
||||
|
||||
本文记录 #16029 已合并的执行流程拆分。[Dispatcher 后续设计](2026-11-01-runtime-execution-dispatcher.zh-CN.md) 将分发预留项、在途调用和完成处理一起迁移，并替换原有取消回调。下方归属表和取消装配说明反映该后续实现。
|
||||
|
||||
## 归属与依赖
|
||||
|
||||
| 组件 | 持有职责 | 调用方 |
|
||||
| ------------------------------ | ------------------------------------------------------------------------------------------------------ | --------------------------------------------------- |
|
||||
| `RuntimeBindingRegistry` | 已有本地 Binding/Session map 与 context；READY/历史 Session 校验 | 服务与全部三个执行流程 |
|
||||
| `RuntimeExecutionState` | 已有 execution 归属读取、带 fencing 的分发结算/UNKNOWN 转换、过期 claim 隔离和后台行结算 | 分发、对账、恢复、取消及现有确认/control 路径 |
|
||||
| `RuntimeExecutionReconciler` | 按 execution ID 预留的查询槽位、`ExecutionObservationCooldown`、证据校验与对账 CAS | 服务观察入口与接管扫描 |
|
||||
| `RuntimeExecutionRecovery` | 有界接管扫描与后台 hold 观察/清理流程 | acquire/接管、普通/排空 release、后台观察与任务停止 |
|
||||
| `RuntimeExecutionCancellation` | 持久取消意图与物理取消/确认流程 | 服务取消入口 |
|
||||
| `RuntimeBrokerSupport` | 不变的无状态 future/错误映射、payload/reference 校验和 status 投影 | 抽出的协作者与服务 |
|
||||
| 服务 | 公共门面、`closed`、分发预留项/在途调用、scheduler、provisioner 生命周期、Binding/Session 生命周期编排 | 现有 adapter 与装配 |
|
||||
| 组件 | 持有职责 | 调用方 |
|
||||
| ------------------------------ | ---------------------------------------------------------------------------------------- | --------------------------------------------------- |
|
||||
| `RuntimeBindingRegistry` | 已有本地 Binding/Session map 与 context；READY/历史 Session 校验 | 服务与全部三个执行流程 |
|
||||
| `RuntimeExecutionState` | 已有 execution 归属读取、带 fencing 的分发结算/UNKNOWN 转换、过期 claim 隔离和后台行结算 | 分发、对账、恢复、取消及现有确认/control 路径 |
|
||||
| `RuntimeExecutionReconciler` | 按 execution ID 预留的查询槽位、`ExecutionObservationCooldown`、证据校验与对账 CAS | 服务观察入口与接管扫描 |
|
||||
| `RuntimeExecutionRecovery` | 有界接管扫描与后台 hold 观察/清理流程 | acquire/接管、普通/排空 release、后台观察与任务停止 |
|
||||
| `RuntimeExecutionCancellation` | 持久取消意图与物理取消/确认流程 | 服务取消入口 |
|
||||
| `RuntimeExecutionDispatcher` | 分发预留项/在途调用、claim 转换、分发续租、v3 结果等待、完成处理和持久 SSH 确认 | 服务的 execution/start 与 drain 路径；取消流程 |
|
||||
| `RuntimeBrokerSupport` | 不变的无状态 future/错误映射、payload/reference 校验、status/回执投影和续租间隔计算 | 抽出的协作者与服务 |
|
||||
| 服务 | 公共门面、`closed`、scheduler、provisioner 生命周期、Binding/Session 生命周期编排 | 现有 adapter 与装配 |
|
||||
|
||||
全部类保留在现有 runtime-broker service package 内，采用 package-private，不新增接口或 Spring 装配。Repository 仍是唯一持久权威。context 的锁继续归 Registry，每个复合读取/准入/释放决策保留当前锁区域。support 类只共享多个真实调用方已有的规则，不引入可配置流程框架。
|
||||
|
||||
@@ -48,7 +51,7 @@ Registry 明确了 Binding 与 Session 状态持有者，但执行对账、接
|
||||
|
||||
## 取消设计
|
||||
|
||||
把取消准入、prepared provider 确认、物理取消与结果校验迁入 `RuntimeExecutionCancellation`。它使用 Registry、执行状态、execution repository、transport、现有 invocation 集合、协调 scheduler、取消时限和关闭屏障。服务门面为唯一的 DISPATCHING 分支传入现有双参数分发动作；这个逐调用传入的 `BiConsumer` 是抽出流程接收的唯一 service 动作。不保存 service 引用，不把 map/私有生命周期 helper 作为回调传递。
|
||||
把取消准入、prepared provider 确认、物理取消与结果校验迁入 `RuntimeExecutionCancellation`。它使用 Registry、执行状态、execution repository、transport、具体分发器、协调 scheduler、取消时限和关闭屏障。取消流程通过 `isInvoking` 查询在途调用，并在未发送的 DISPATCHING 分支调用 `beginDispatch`，不再接收可变 invocation 集合或逐调用传入的 service 回调。不保存 service 引用，不把 map/私有生命周期 helper 作为回调传递。
|
||||
|
||||
保留在要求活跃 Session 前读取终态回执归属的顺序。若原 worker 可能仍保留 preparation，prepared provider 的已取消回执仍需原 worker 确认。非终态取消先持久化意图，再发送 worker 请求。hold 行和已授权的私有 CSI 调用保留原拒绝规则。租约过期后，本进程仍运行的 invocation 可继续接收取消；本进程未运行的 invocation 先按 repository 时钟执行 dispatch fence。UNKNOWN 只在对账证据足够时才能结算。
|
||||
|
||||
@@ -68,10 +71,10 @@ Registry 明确了 Binding 与 Session 状态持有者，但执行对账、接
|
||||
|
||||
在模块共享的 10 分钟本地验证预算内，复用匹配源码的 SDK 和 JDK 21/Maven 环境，运行受影响 Java 测试、Checkstyle 和最小子进程路径，并检查真实选中的报告。完整 MySQL 8.4、打包 Hosted/fault 恢复和 SpotBugs 仍需 CI 验收；fixture/H2 或测试 worker 子进程通过不构成这些结果。本次无 TypeScript/共享契约变化，不需要 JavaScript 构建。
|
||||
|
||||
验收要求：三个公共入口族委托给对应协作者；查询预留项与 cooldown 归同一持有者；协作者不访问 service 内部；唯一的取消分发动作显式传入；SQL/schema/公共契约无 diff；迁移算法语义不变；顺序回归与定向测试通过；双语文档及两轮干净自审完成。在被验收版本运行所需门禁前，CI 验收仍待完成。
|
||||
验收要求：三个公共入口族委托给对应协作者；查询预留项与 cooldown 归同一持有者；协作者不访问 service 内部；取消流程直接使用分发器，不接收 service 回调或可变 invocation 集合；SQL/schema/公共契约无 diff；迁移算法语义不变；顺序回归与定向测试通过；双语文档及两轮干净自审完成。在被验收版本运行所需门禁前，CI 验收仍待完成。
|
||||
|
||||
## 风险与待定问题
|
||||
|
||||
主要风险是同步异常映射、完成顺序、guard 计数和跨代际路由改变。通过机械迁移和保留调用点包络降低风险，现有集成测试继续经公共服务门面验证它们。共享的不可变依赖直接传递，不隐藏在通用 context 对象中。
|
||||
|
||||
回调例外刻意限于取消的未发送分发分支；抽取整个 dispatcher 会扩大本次范围。没有待定产品决策。后续 dispatcher 抽取可在不改变取消语义的前提下，把该动作替换为具体协作者。
|
||||
原 #16029 切片刻意将回调例外限制在取消的未发送分发分支。Dispatcher 后续实现用具体协作者替换它，保持取消语义，并将分发状态与完成处理一起迁移。没有待定产品决策。进一步拆分供应或 Session 生命周期不在本增量范围内。
|
||||
+16
-452
@@ -59,8 +59,6 @@ import java.util.UUID;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
@@ -72,19 +70,21 @@ import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBindingRegistry.evidenceUnavailable;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBindingRegistry.sameLease;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.MAX_CAS_ATTEMPTS;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.conflict;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.failed;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.immutableMap;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.invalid;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.isDetachedCapture;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.mapFailure;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.mapStoreFailure;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.notFound;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.referenceString;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.renewalDelayMillis;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.requireOpen;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.safeStage;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.sshNoStartReceipt;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.unavailable;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.unwrap;
|
||||
|
||||
@@ -135,12 +135,8 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
private final ScheduledExecutorService renewalScheduler;
|
||||
private final AtomicBoolean closed = new AtomicBoolean();
|
||||
private final RuntimeBindingRegistry bindingRegistry;
|
||||
private final ConcurrentMap<String, CompletableFuture<Void>> dispatches =
|
||||
new ConcurrentHashMap<>();
|
||||
// Executions whose transport.execute call is in flight in this process,
|
||||
// as opposed to dispatches, which also covers a claim being fenced.
|
||||
private final Set<String> invocations = ConcurrentHashMap.newKeySet();
|
||||
private final RuntimeExecutionState executionState;
|
||||
private final RuntimeExecutionDispatcher executionDispatcher;
|
||||
private final RuntimeExecutionReconciler executionReconciler;
|
||||
private final RuntimeExecutionRecovery executionRecovery;
|
||||
private final RuntimeExecutionCancellation executionCancellation;
|
||||
@@ -332,13 +328,17 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
sessionRepository, sessionResolver, this.brokerOwnerId, this.operationLeaseDuration, clock);
|
||||
this.executionState = new RuntimeExecutionState(bindingRepository, sessionRepository, executionRepository,
|
||||
this.brokerOwnerId, this.dispatchLeaseDuration, clock);
|
||||
this.executionDispatcher = new RuntimeExecutionDispatcher(bindingRegistry, executionState,
|
||||
bindingRepository, sessionRepository, executionRepository, provisioner, transport,
|
||||
publicationVerifier, this.brokerOwnerId, this.dispatchLeaseDuration, this.v3ResultWindow,
|
||||
clock, scheduler, renewalScheduler, closed);
|
||||
this.executionReconciler = new RuntimeExecutionReconciler(bindingRegistry, executionState, executionRepository,
|
||||
provisioner, transport, publicationVerifier, this.operationLeaseDuration, clock, closed,
|
||||
new ExecutionObservationCooldown(clock, scheduler));
|
||||
this.executionRecovery = new RuntimeExecutionRecovery(bindingRegistry, executionState, executionRepository,
|
||||
sessionRepository, transport, executionReconciler, closed);
|
||||
this.executionCancellation = new RuntimeExecutionCancellation(bindingRegistry, executionState, executionRepository,
|
||||
transport, invocations, scheduler, this.operationLeaseDuration, closed);
|
||||
transport, executionDispatcher, scheduler, this.operationLeaseDuration, closed);
|
||||
}
|
||||
|
||||
public boolean supportsDrainedStop() {
|
||||
@@ -784,7 +784,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
var batch = executionRepository.findByBinding(binding.getBindingId(), binding.getGeneration(), cursor, 100);
|
||||
CompletionStage<Void> work = CompletableFuture.completedFuture(null);
|
||||
for (var execution : batch) {
|
||||
work = work.thenCompose(ignored -> acknowledgePersistedTools(binding, execution));
|
||||
work = work.thenCompose(ignored -> executionDispatcher.acknowledgePersistedTools(binding, execution));
|
||||
}
|
||||
return work.thenCompose(ignored -> batch.size() < 100 ? CompletableFuture.completedFuture(null)
|
||||
: acknowledgeDrainedTools(binding, batch.getLast().getExecutionCallId()));
|
||||
@@ -818,57 +818,6 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
}).thenCompose(stage -> stage);
|
||||
}
|
||||
|
||||
private CompletionStage<Void> acknowledgePersistedTools(RuntimeBindingRecord binding, ToolExecutionRecord execution) {
|
||||
if (!com.alibaba.qwen.code.runtimebroker.protocol.RuntimeProviderKinds.SSH_PROCESS.equals(binding.getRequest().getProvisionerKind())
|
||||
|| !execution.isSettled() || execution.getResult() == null
|
||||
|| ProviderRuntimeProtocol.isReference(execution.getReference())) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
Map<String, Object> noStart = sshNoStartReceipt(binding, execution);
|
||||
if (noStart == null && (execution.getResult().containsKey("capture")
|
||||
|| Integer.valueOf(3).equals(execution.getReference().get("runtimeProtocol"))
|
||||
|| "deferred_v3".equals(execution.getReference().get("dispatchMode")))) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
RuntimeSessionRecord session = sessionRepository.findById(binding.getRequest().getScope(),
|
||||
execution.getRuntimeSessionId());
|
||||
if (binding.getLease() == null || session == null || !execution.belongsTo(session)
|
||||
|| !binding.getBindingId().equals(execution.getBindingId())
|
||||
|| binding.getGeneration() != execution.getRuntimeGeneration()
|
||||
|| !binding.getRequest().getIsolationKey().equals(execution.getHarnessSessionId())) {
|
||||
return failed(conflict("workspace_close_identity_unverified", "Persisted Tool ownership conflicts"));
|
||||
}
|
||||
if (noStart != null) {
|
||||
CompletionStage<Map<String, Object>> acknowledgement = "deferred_v3".equals(execution.getReference().get("dispatchMode"))
|
||||
? transport.acknowledgeV3(binding.getLease(), session.getSession(), execution.getReference(), noStart)
|
||||
: transport.acknowledge(binding.getLease(), session.getSession(), execution.getReference(), noStart);
|
||||
return acknowledgement.thenAccept(answer -> {
|
||||
if (answer == null || !Set.of("settled", "unknown").contains(answer.get("state"))) {
|
||||
throw conflict("runtime_execution_conflict", "Original Runtime did not confirm the no-start acknowledgement");
|
||||
}
|
||||
});
|
||||
}
|
||||
return transport.acknowledgeV2(binding.getLease(), session.getSession(), execution.getReference());
|
||||
}
|
||||
|
||||
private static Map<String, Object> sshNoStartReceipt(RuntimeBindingRecord binding, ToolExecutionRecord execution) {
|
||||
if (!com.alibaba.qwen.code.runtimebroker.protocol.RuntimeProviderKinds.SSH_PROCESS.equals(binding.getRequest().getProvisionerKind())
|
||||
|| !execution.isSettled() || !"not_started".equals(execution.getExecutionStatus())
|
||||
|| execution.getResult() == null || !execution.getResult().containsKey("capture")
|
||||
|| execution.getResult().get("capture") != null
|
||||
|| !(Integer.valueOf(3).equals(execution.getReference().get("runtimeProtocol"))
|
||||
|| "deferred_v3".equals(execution.getReference().get("dispatchMode")))
|
||||
|| ProviderRuntimeProtocol.isReference(execution.getReference())) {
|
||||
return null;
|
||||
}
|
||||
Map<String, Object> receipt = new LinkedHashMap<>();
|
||||
receipt.put("executionCallId", execution.getExecutionCallId());
|
||||
receipt.put("manifest", null);
|
||||
receipt.put("deliveryStatus", "committed");
|
||||
receipt.put("historyRevision", null);
|
||||
return BrokerValues.immutableMap(receipt);
|
||||
}
|
||||
|
||||
private CompletionStage<Void> releaseSavedSession(RuntimeBindingRecord binding, RuntimeSessionRecord record,
|
||||
BindingRenewal renewal) {
|
||||
return releaseSavedSession(binding, record, renewal, true);
|
||||
@@ -1784,7 +1733,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
if (family != null) {
|
||||
admitBackgroundProcess(context, current, family);
|
||||
}
|
||||
beginDispatch(context, current, payload, grant);
|
||||
executionDispatcher.beginDispatch(context, current, payload, grant);
|
||||
ToolExecutionRecord latest = executionRepository
|
||||
.findByExecutionCallId(executionId);
|
||||
return latest == null ? current : latest;
|
||||
@@ -1794,7 +1743,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
});
|
||||
}
|
||||
if (shouldDriveDispatch(record)) {
|
||||
beginDispatch(context, record, payload);
|
||||
executionDispatcher.beginDispatch(context, record, payload);
|
||||
}
|
||||
ToolExecutionRecord latest = executionRepository.findByExecutionCallId(executionId);
|
||||
return CompletableFuture.completedFuture(latest == null ? record : latest);
|
||||
@@ -1822,7 +1771,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
}
|
||||
ProviderRuntimeProtocol.reference(record.getReference(), runtimeSessionId);
|
||||
if (shouldDriveDispatch(record)) {
|
||||
beginDispatch(context, record);
|
||||
executionDispatcher.beginDispatch(context, record);
|
||||
}
|
||||
return executionState.requireExecution(context, executionId);
|
||||
} finally {
|
||||
@@ -2071,7 +2020,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
public CompletionStage<ToolExecutionRecord> cancelExecution(
|
||||
String harnessSessionId, String runtimeSessionId,
|
||||
String executionCallId) {
|
||||
return executionCancellation.cancelExecution(harnessSessionId, runtimeSessionId, executionCallId, this::beginDispatch);
|
||||
return executionCancellation.cancelExecution(harnessSessionId, runtimeSessionId, executionCallId);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -2376,7 +2325,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
context.unlock();
|
||||
}
|
||||
if (dispatch && shouldDriveDispatch(record)) {
|
||||
beginDispatch(context, record);
|
||||
executionDispatcher.beginDispatch(context, record);
|
||||
}
|
||||
ToolExecutionRecord current = executionRepository
|
||||
.findByExecutionCallId(record.getExecutionCallId());
|
||||
@@ -3894,307 +3843,6 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
? bindingRepository.requireSessionAdmission(sessionRepository, record) : record;
|
||||
}
|
||||
|
||||
private void beginDispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared) {
|
||||
beginDispatch(context, prepared, null);
|
||||
}
|
||||
|
||||
private void beginDispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared, Map<String, Object> payload) {
|
||||
beginDispatch(context, prepared, payload, null);
|
||||
}
|
||||
|
||||
private void beginDispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared, Map<String, Object> payload,
|
||||
RuntimePublicationGrant grant) {
|
||||
if (CsiFilesRetirementProfile.isProfile(context.binding().getRequest().getScope())
|
||||
&& (!(bindingRepository instanceof NativeCsiExecutionRepository)
|
||||
|| payload == null || !Set.of("read_file", "write_file", "edit").contains(String.valueOf(payload.get("toolName"))) || grant != null)) {
|
||||
throw new RuntimeBrokerException(501, "csi_file_dispatch_unavailable",
|
||||
"Private CSI dispatch requires original JDBC authorization and a finite file payload.", false);
|
||||
}
|
||||
if (!prepared.isCancelRequested()) {
|
||||
bindingRepository.requireHarnessAdmission(context.binding().getRequest(), context.session().getScope(),
|
||||
context.session().getHarnessSessionId(), null);
|
||||
}
|
||||
if (payload == null && ("deferred".equals(prepared.getReference().get("dispatchMode"))
|
||||
|| "deferred_v3".equals(prepared.getReference().get("dispatchMode")))
|
||||
&& !prepared.isCancelRequested()) {
|
||||
throw conflict("runtime_execution_conflict", "Deferred execution requires its original payload");
|
||||
}
|
||||
CompletableFuture<Void> created = new CompletableFuture<>();
|
||||
CompletableFuture<Void> existing = dispatches.putIfAbsent(
|
||||
prepared.getExecutionCallId(), created);
|
||||
if (existing != null) {
|
||||
return;
|
||||
}
|
||||
CompletionStage<Void> operation;
|
||||
try {
|
||||
operation = dispatch(context, prepared, payload, grant);
|
||||
} catch (RuntimeException | Error exception) {
|
||||
dispatches.remove(prepared.getExecutionCallId(), created);
|
||||
created.completeExceptionally(exception);
|
||||
if (exception instanceof RuntimeBrokerException refused) {
|
||||
throw refused;
|
||||
}
|
||||
throw unavailable("runtime_execution_dispatch_failed",
|
||||
"Runtime execution dispatch failed", exception);
|
||||
}
|
||||
operation.whenComplete((ignored, error) -> {
|
||||
dispatches.remove(prepared.getExecutionCallId(), created);
|
||||
if (error == null) {
|
||||
created.complete(null);
|
||||
} else {
|
||||
created.completeExceptionally(unwrap(error));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private CompletionStage<Void> dispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared, Map<String, Object> payload,
|
||||
RuntimePublicationGrant grant) {
|
||||
ToolExecutionRecord claimed = executionRepository.claimDispatch(
|
||||
prepared.getExecutionCallId(), brokerOwnerId,
|
||||
dispatchLeaseDuration);
|
||||
if (claimed == null
|
||||
|| claimed.getState()
|
||||
!= ToolExecutionRecord.State.DISPATCHING) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
ToolExecutionRecord executing = enterExecuting(claimed);
|
||||
if (executing == null || executing.isTerminal()
|
||||
|| executing.getState()
|
||||
!= ToolExecutionRecord.State.EXECUTING
|
||||
|| !ownsDispatch(executing, claimed)) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
if (!provisioner.isUsable(context.lease())) {
|
||||
bindingRegistry.invalidateBinding(context.binding());
|
||||
executionState.markUnknown(executing);
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
DispatchRenewal renewal = new DispatchRenewal(executing);
|
||||
try {
|
||||
renewal.start();
|
||||
} catch (RuntimeException | Error exception) {
|
||||
executionState.markUnknown(executing);
|
||||
throw exception;
|
||||
}
|
||||
invocations.add(executing.getExecutionCallId());
|
||||
CompletionStage<Map<String, Object>> invocation = grant == null
|
||||
? safeStage(() -> payload == null
|
||||
? transport.execute(context.lease(), context.session(), executing.getReference())
|
||||
: transport.execute(context.lease(), context.session(),
|
||||
dispatchReference(executing, context.binding().getRequest().getScope()), payload))
|
||||
: safeStage(() -> transport.executeV3(context.lease(), context.session(),
|
||||
executing.getReference(), payload, capture(grant)))
|
||||
.handle((answer, error) -> {
|
||||
if (nonRetryableToolV3(error)) {
|
||||
throw new CompletionException(unwrap(error));
|
||||
}
|
||||
if (error == null && answer != null
|
||||
&& "settled".equals(answer.get("state"))
|
||||
&& answer.get("result") instanceof Map<?, ?> result
|
||||
&& "not_started".equals(result.get("executionStatus"))) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> notStarted = (Map<String, Object>) result;
|
||||
return notStarted;
|
||||
}
|
||||
return null;
|
||||
})
|
||||
.thenCompose(notStarted -> notStarted != null
|
||||
? CompletableFuture.completedFuture(notStarted)
|
||||
: awaitV3Result(context, executing,
|
||||
clock.instant().plus(v3ResultWindow)));
|
||||
return invocation
|
||||
.<Void>handle((result, error) -> {
|
||||
// Stop counting as running before the outcome is
|
||||
// written, so a cancel that reads that outcome does not
|
||||
// treat this finished invocation as still running.
|
||||
invocations.remove(executing.getExecutionCallId());
|
||||
if (error != null || result == null) {
|
||||
executionState.markUnknown(executing);
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
executionState.settleExecution(executing, result);
|
||||
executionState.settleUnstartedBackgroundSiblings(
|
||||
executing.getExecutionCallId(), result);
|
||||
} catch (RuntimeException exception) {
|
||||
executionState.markUnknown(executing);
|
||||
}
|
||||
return null;
|
||||
}).thenCompose(ignored -> {
|
||||
if (!com.alibaba.qwen.code.runtimebroker.protocol.RuntimeProviderKinds.SSH_PROCESS.equals(context.binding().getRequest().getProvisionerKind())) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
ToolExecutionRecord settled = executionRepository.findByExecutionCallId(executing.getExecutionCallId());
|
||||
return settled != null && settled.sameIdentity(executing)
|
||||
? acknowledgePersistedTools(context.binding(), settled).exceptionally(error -> null)
|
||||
: CompletableFuture.completedFuture(null);
|
||||
}).whenComplete((ignored, error) -> renewal.close());
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static Map<String, Object> capture(RuntimePublicationGrant grant) {
|
||||
Map<String, Object> binding = grant.binding();
|
||||
Map<String, Object> key = (Map<String, Object>) binding.get("sessionKey");
|
||||
return Map.of("tenantId", key.get("tenantId"), "sessionId", key.get("sessionId"),
|
||||
"turnId", binding.get("turnId"), "executionCallId", binding.get("executionCallId"),
|
||||
"bindingGeneration", binding.get("bindingGeneration"),
|
||||
"capturePolicy", "complete_required");
|
||||
}
|
||||
|
||||
private CompletionStage<Map<String, Object>> awaitV3Result(SessionContext context,
|
||||
ToolExecutionRecord original, Instant deadline) {
|
||||
CompletableFuture<Map<String, Object>> result = new CompletableFuture<>();
|
||||
pollV3Result(context, original, deadline, result, 0);
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* The v3 result-poll backoff: 100ms doubling, capped at 2s. The cap
|
||||
* bounds the scheduled delay between rounds for one execution; rounds
|
||||
* for all executions share the single coordination thread, so pick-up
|
||||
* lateness under concurrency also carries that thread's queue wait.
|
||||
*/
|
||||
static long v3PollDelayMillis(int attempt) {
|
||||
return Math.min(2_000, 100L << Math.min(attempt, 6));
|
||||
}
|
||||
|
||||
private void pollV3Result(SessionContext context, ToolExecutionRecord original,
|
||||
Instant deadline, CompletableFuture<Map<String, Object>> answer,
|
||||
int attempt) {
|
||||
if (answer.isDone()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Map<String, Object> finished = publicationVerifier.finished(original);
|
||||
if (finished != null) {
|
||||
answer.complete(finished);
|
||||
return;
|
||||
}
|
||||
ToolExecutionRecord current = executionRepository.findByExecutionCallId(
|
||||
original.getExecutionCallId());
|
||||
if (current == null || !ownsDispatch(current, original)
|
||||
|| !current.getBindingId().equals(context.binding().getBindingId())
|
||||
|| current.getRuntimeGeneration() != context.binding().getGeneration()
|
||||
|| !provisioner.isUsable(context.lease()) || !clock.instant().isBefore(deadline)) {
|
||||
throw evidenceUnavailable();
|
||||
}
|
||||
safeStage(() -> transport.statusV3(context.lease(), context.session(),
|
||||
original.getReference(), 0)).whenComplete((status, error) -> {
|
||||
if (answer.isDone()) {
|
||||
return;
|
||||
}
|
||||
if (nonRetryableToolV3(error)) {
|
||||
answer.completeExceptionally(unwrap(error));
|
||||
return;
|
||||
}
|
||||
if (error == null && status != null && "settled".equals(status.get("state"))
|
||||
&& status.get("result") instanceof Map<?, ?> result
|
||||
&& "not_started".equals(result.get("executionStatus"))) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> saved = (Map<String, Object>) result;
|
||||
answer.complete(saved);
|
||||
return;
|
||||
}
|
||||
// A background Shell start settles its handle as the
|
||||
// detached family: nothing is ever published for it,
|
||||
// so the status answer is its only durable settle.
|
||||
if (error == null && status != null && "settled".equals(status.get("state"))
|
||||
&& status.get("result") instanceof Map<?, ?> detached
|
||||
&& isDetachedCapture(detached)) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> saved = (Map<String, Object>) detached;
|
||||
answer.complete(saved);
|
||||
return;
|
||||
}
|
||||
if (error == null && status != null && "unknown".equals(status.get("state"))) {
|
||||
answer.completeExceptionally(evidenceUnavailable());
|
||||
return;
|
||||
}
|
||||
// Each round costs two repository reads and one worker
|
||||
// call; back off instead of pinning them at 10/s for
|
||||
// the whole window.
|
||||
long delay = v3PollDelayMillis(attempt);
|
||||
try {
|
||||
scheduler.schedule(() -> pollV3Result(context,
|
||||
original, deadline, answer, attempt + 1),
|
||||
delay, TimeUnit.MILLISECONDS);
|
||||
} catch (RuntimeException failure) {
|
||||
answer.completeExceptionally(failure);
|
||||
}
|
||||
});
|
||||
} catch (RuntimeException failure) {
|
||||
answer.completeExceptionally(failure);
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean nonRetryableToolV3(Throwable error) {
|
||||
return error != null && unwrap(error) instanceof RuntimeBrokerException failure
|
||||
&& !failure.isRetryable();
|
||||
}
|
||||
|
||||
/** The detached capture family: success with a manifest-less capture. */
|
||||
private static boolean isDetachedCapture(Map<?, ?> result) {
|
||||
return result != null && "success".equals(result.get("executionStatus"))
|
||||
&& result.get("capture") instanceof Map<?, ?> capture
|
||||
&& "detached".equals(capture.get("captureStatus"))
|
||||
&& capture.get("manifest") == null;
|
||||
}
|
||||
|
||||
private static Map<String, Object> dispatchReference(ToolExecutionRecord record, RuntimeScope scope) {
|
||||
if (CsiFilesRetirementProfile.isProfile(scope)) {
|
||||
Map<String, Object> reference = new LinkedHashMap<>(record.getReference());
|
||||
reference.put("executionCallId", record.getExecutionCallId());
|
||||
return Map.copyOf(reference);
|
||||
}
|
||||
if (!Integer.valueOf(3).equals(record.getReference().get("runtimeProtocol"))) {
|
||||
return record.getReference();
|
||||
}
|
||||
Map<String, Object> reference = new LinkedHashMap<>(record.getReference());
|
||||
reference.put("executionCallId", record.getExecutionCallId());
|
||||
return Map.copyOf(reference);
|
||||
}
|
||||
|
||||
private ToolExecutionRecord enterExecuting(
|
||||
ToolExecutionRecord claimed) {
|
||||
ToolExecutionRecord current = claimed;
|
||||
for (int attempt = 0; attempt < MAX_CAS_ATTEMPTS; attempt++) {
|
||||
if (current == null || current.isTerminal()
|
||||
|| current.getState()
|
||||
== ToolExecutionRecord.State.UNKNOWN
|
||||
|| !ownsDispatch(current, claimed)) {
|
||||
return current;
|
||||
}
|
||||
if (current.getState() != ToolExecutionRecord.State.DISPATCHING) {
|
||||
return null;
|
||||
}
|
||||
ToolExecutionRecord replacement;
|
||||
if (current.isCancelRequested()) {
|
||||
replacement = current.withResult(
|
||||
current.cancellationBeforeDispatch(),
|
||||
current.getLastSequence(), clock.instant());
|
||||
} else {
|
||||
replacement = current.withState(
|
||||
ToolExecutionRecord.State.EXECUTING, false);
|
||||
}
|
||||
ToolExecutionRecord updated = current.isCancelRequested()
|
||||
? executionRepository.compareAndSet(current, replacement, brokerOwnerId,
|
||||
claimed.getDispatchGeneration())
|
||||
: bindingRepository.authorizeDispatch(sessionRepository, executionRepository, current,
|
||||
brokerOwnerId, claimed.getDispatchGeneration());
|
||||
if (updated != null) {
|
||||
return updated;
|
||||
}
|
||||
current = executionRepository.findByExecutionCallId(
|
||||
claimed.getExecutionCallId());
|
||||
}
|
||||
executionState.markUnknown(claimed);
|
||||
return null;
|
||||
}
|
||||
|
||||
private RuntimeSessionRecord transitionSessionToReleasing(
|
||||
SessionContext context) {
|
||||
RuntimeSessionRecord current = sessionRepository.findById(
|
||||
@@ -4315,7 +3963,7 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
return;
|
||||
}
|
||||
bindingRegistry.cancelPendingOperations();
|
||||
dispatches.values().forEach(future -> future.cancel(false));
|
||||
executionDispatcher.cancelPendingOperations();
|
||||
executionReconciler.cancelPendingOperations();
|
||||
scheduler.shutdownNow();
|
||||
renewalScheduler.shutdownNow();
|
||||
@@ -4374,14 +4022,6 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
&& record.getState() != ToolExecutionRecord.State.UNKNOWN;
|
||||
}
|
||||
|
||||
private static boolean ownsDispatch(ToolExecutionRecord current,
|
||||
ToolExecutionRecord claimed) {
|
||||
return claimed.getDispatchGeneration()
|
||||
== current.getDispatchGeneration()
|
||||
&& claimed.getDispatchOwner().equals(
|
||||
current.getDispatchOwner());
|
||||
}
|
||||
|
||||
private static Duration requireDuration(Duration duration,
|
||||
String name) {
|
||||
if (duration == null || duration.isZero() || duration.isNegative()) {
|
||||
@@ -4418,10 +4058,6 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
return executor;
|
||||
}
|
||||
|
||||
private long renewalDelayMillis(Duration duration) {
|
||||
return Math.max(1, duration.toMillis() / 3);
|
||||
}
|
||||
|
||||
/**
|
||||
* The bound for a step that runs WITHOUT its own renewal (currently only
|
||||
* the maintenance path's attestation leg): it must stay one renewal tick
|
||||
@@ -4631,76 +4267,4 @@ public final class RuntimeBrokerService implements AutoCloseable, com.alibaba.qw
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private final class DispatchRenewal implements AutoCloseable {
|
||||
// Same virtual-thread rule as BindingRenewal: guarded repository
|
||||
// calls must not pin the caller's carrier while they block.
|
||||
private final ReentrantLock monitor = new ReentrantLock();
|
||||
private final ToolExecutionRecord original;
|
||||
private ScheduledFuture<?> task;
|
||||
|
||||
DispatchRenewal(ToolExecutionRecord original) {
|
||||
this.original = original;
|
||||
}
|
||||
|
||||
void start() {
|
||||
monitor.lock();
|
||||
try {
|
||||
long delay = renewalDelayMillis(dispatchLeaseDuration);
|
||||
task = renewalScheduler.scheduleWithFixedDelay(this::renew,
|
||||
delay, delay, TimeUnit.MILLISECONDS);
|
||||
} finally {
|
||||
monitor.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void renew() {
|
||||
monitor.lock();
|
||||
try {
|
||||
if (closed.get()) {
|
||||
closeLocked();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
ToolExecutionRecord renewed =
|
||||
executionRepository.renewDispatch(
|
||||
original.getExecutionCallId(), brokerOwnerId,
|
||||
original.getDispatchGeneration(), dispatchLeaseDuration);
|
||||
if (renewed == null) {
|
||||
executionRepository.claimDispatch(original.getExecutionCallId(),
|
||||
brokerOwnerId, dispatchLeaseDuration);
|
||||
closeLocked();
|
||||
} else if (!renewed.sameIdentity(original) || !renewed.sameAuthorization(original)
|
||||
|| !ownsDispatch(renewed, original)) {
|
||||
closeLocked();
|
||||
}
|
||||
} catch (RuntimeException exception) {
|
||||
if (exception instanceof RuntimeBrokerException failure && !failure.isRetryable()
|
||||
&& (failure.getCode().startsWith("csi_")
|
||||
|| "runtime_admission_closed".equals(failure.getCode()))) {
|
||||
closeLocked();
|
||||
}
|
||||
// A transient repository failure does not prove claim loss.
|
||||
}
|
||||
} finally {
|
||||
monitor.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
monitor.lock();
|
||||
try {
|
||||
closeLocked();
|
||||
} finally {
|
||||
monitor.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void closeLocked() {
|
||||
if (task != null) {
|
||||
task.cancel(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+35
@@ -2,11 +2,16 @@ package com.alibaba.qwen.code.runtimebroker.service;
|
||||
|
||||
import com.alibaba.qwen.code.runtimebroker.BrokerValues;
|
||||
import com.alibaba.qwen.code.runtimebroker.HarnessSessionResolver;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeLifecycleAuthority;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeScope;
|
||||
import com.alibaba.qwen.code.runtimebroker.StoreFailureException;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.protocol.ProviderRuntimeProtocol;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
@@ -203,4 +208,34 @@ final class RuntimeBrokerSupport {
|
||||
throw new IllegalStateException("Runtime Broker is closed");
|
||||
}
|
||||
}
|
||||
|
||||
static Map<String, Object> sshNoStartReceipt(RuntimeBindingRecord binding, ToolExecutionRecord execution) {
|
||||
if (!com.alibaba.qwen.code.runtimebroker.protocol.RuntimeProviderKinds.SSH_PROCESS.equals(binding.getRequest().getProvisionerKind())
|
||||
|| !execution.isSettled() || !"not_started".equals(execution.getExecutionStatus())
|
||||
|| execution.getResult() == null || !execution.getResult().containsKey("capture")
|
||||
|| execution.getResult().get("capture") != null
|
||||
|| !(Integer.valueOf(3).equals(execution.getReference().get("runtimeProtocol"))
|
||||
|| "deferred_v3".equals(execution.getReference().get("dispatchMode")))
|
||||
|| ProviderRuntimeProtocol.isReference(execution.getReference())) {
|
||||
return null;
|
||||
}
|
||||
Map<String, Object> receipt = new LinkedHashMap<>();
|
||||
receipt.put("executionCallId", execution.getExecutionCallId());
|
||||
receipt.put("manifest", null);
|
||||
receipt.put("deliveryStatus", "committed");
|
||||
receipt.put("historyRevision", null);
|
||||
return BrokerValues.immutableMap(receipt);
|
||||
}
|
||||
|
||||
/** The detached capture family: success with a manifest-less capture. */
|
||||
static boolean isDetachedCapture(Map<?, ?> result) {
|
||||
return result != null && "success".equals(result.get("executionStatus"))
|
||||
&& result.get("capture") instanceof Map<?, ?> capture
|
||||
&& "detached".equals(capture.get("captureStatus"))
|
||||
&& capture.get("manifest") == null;
|
||||
}
|
||||
|
||||
static long renewalDelayMillis(Duration duration) {
|
||||
return Math.max(1, duration.toMillis() / 3);
|
||||
}
|
||||
}
|
||||
+7
-8
@@ -21,7 +21,6 @@ import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.function.BiConsumer;
|
||||
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.MAX_CAS_ATTEMPTS;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.conflict;
|
||||
@@ -38,7 +37,7 @@ final class RuntimeExecutionCancellation {
|
||||
private final RuntimeExecutionState executionState;
|
||||
private final ToolExecutionRepository executionRepository;
|
||||
private final RuntimeTransport transport;
|
||||
private final Set<String> invocations;
|
||||
private final RuntimeExecutionDispatcher dispatcher;
|
||||
private final ScheduledExecutorService scheduler;
|
||||
private final Duration operationLeaseDuration;
|
||||
private final AtomicBoolean closed;
|
||||
@@ -47,7 +46,7 @@ final class RuntimeExecutionCancellation {
|
||||
RuntimeExecutionState executionState,
|
||||
ToolExecutionRepository executionRepository,
|
||||
RuntimeTransport transport,
|
||||
Set<String> invocations,
|
||||
RuntimeExecutionDispatcher dispatcher,
|
||||
ScheduledExecutorService scheduler,
|
||||
Duration operationLeaseDuration,
|
||||
AtomicBoolean closed) {
|
||||
@@ -55,7 +54,7 @@ final class RuntimeExecutionCancellation {
|
||||
this.executionState = executionState;
|
||||
this.executionRepository = executionRepository;
|
||||
this.transport = transport;
|
||||
this.invocations = invocations;
|
||||
this.dispatcher = dispatcher;
|
||||
this.scheduler = scheduler;
|
||||
this.operationLeaseDuration = operationLeaseDuration;
|
||||
this.closed = closed;
|
||||
@@ -67,7 +66,7 @@ final class RuntimeExecutionCancellation {
|
||||
|
||||
CompletionStage<ToolExecutionRecord> cancelExecution(
|
||||
String harnessSessionId, String runtimeSessionId,
|
||||
String executionCallId, BiConsumer<SessionContext, ToolExecutionRecord> dispatch) {
|
||||
String executionCallId) {
|
||||
requireOpen(closed);
|
||||
String executionId = BrokerValues.requireId(executionCallId,
|
||||
"executionCallId");
|
||||
@@ -141,7 +140,7 @@ final class RuntimeExecutionCancellation {
|
||||
if (requested.isTerminal()
|
||||
|| (requested.getState()
|
||||
== ToolExecutionRecord.State.UNKNOWN
|
||||
&& !invocations.contains(
|
||||
&& !dispatcher.isInvoking(
|
||||
executionId)
|
||||
&& !publicationV3
|
||||
&& !requested.observableAfterLoss())) {
|
||||
@@ -153,7 +152,7 @@ final class RuntimeExecutionCancellation {
|
||||
}
|
||||
if (requested.getState()
|
||||
== ToolExecutionRecord.State.DISPATCHING) {
|
||||
dispatch.accept(context, requested);
|
||||
dispatcher.beginDispatch(context, requested);
|
||||
ToolExecutionRecord latest = executionRepository
|
||||
.findByExecutionCallId(executionId);
|
||||
return CompletableFuture.completedFuture(
|
||||
@@ -162,7 +161,7 @@ final class RuntimeExecutionCancellation {
|
||||
if (requested.getState()
|
||||
== ToolExecutionRecord.State
|
||||
.CANCEL_REQUESTED
|
||||
&& !invocations.contains(executionId)) {
|
||||
&& !dispatcher.isInvoking(executionId)) {
|
||||
// Fence a lapsed claim nothing here is running. A
|
||||
// claim the repository still holds live falls
|
||||
// through to the physical cancel, as does an
|
||||
|
||||
+511
@@ -0,0 +1,511 @@
|
||||
package com.alibaba.qwen.code.runtimebroker.service;
|
||||
|
||||
import com.alibaba.qwen.code.runtimebroker.CsiFilesRetirementProfile;
|
||||
import com.alibaba.qwen.code.runtimebroker.NativeCsiExecutionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeProvisioner;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimePublicationGrant;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimePublicationVerifier;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeScope;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeSessionRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeSessionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeTransport;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.protocol.ProviderRuntimeProtocol;
|
||||
import com.alibaba.qwen.code.runtimebroker.service.RuntimeBindingRegistry.SessionContext;
|
||||
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CompletionException;
|
||||
import java.util.concurrent.CompletionStage;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBindingRegistry.evidenceUnavailable;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.MAX_CAS_ATTEMPTS;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.conflict;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.failed;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.isDetachedCapture;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.renewalDelayMillis;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.safeStage;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.sshNoStartReceipt;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.unavailable;
|
||||
import static com.alibaba.qwen.code.runtimebroker.service.RuntimeBrokerSupport.unwrap;
|
||||
|
||||
final class RuntimeExecutionDispatcher {
|
||||
private final RuntimeBindingRegistry bindingRegistry;
|
||||
private final RuntimeExecutionState executionState;
|
||||
private final RuntimeBindingRepository bindingRepository;
|
||||
private final RuntimeSessionRepository sessionRepository;
|
||||
private final ToolExecutionRepository executionRepository;
|
||||
private final RuntimeProvisioner provisioner;
|
||||
private final RuntimeTransport transport;
|
||||
private final RuntimePublicationVerifier publicationVerifier;
|
||||
private final String brokerOwnerId;
|
||||
private final Duration dispatchLeaseDuration;
|
||||
private final Duration v3ResultWindow;
|
||||
private final Clock clock;
|
||||
private final ScheduledExecutorService scheduler;
|
||||
private final ScheduledExecutorService renewalScheduler;
|
||||
private final AtomicBoolean closed;
|
||||
private final ConcurrentMap<String, CompletableFuture<Void>> dispatches =
|
||||
new ConcurrentHashMap<>();
|
||||
// Executions whose transport.execute call is in flight in this process,
|
||||
// as opposed to dispatches, which also covers a claim being fenced.
|
||||
private final Set<String> invocations = ConcurrentHashMap.newKeySet();
|
||||
|
||||
RuntimeExecutionDispatcher(RuntimeBindingRegistry bindingRegistry,
|
||||
RuntimeExecutionState executionState,
|
||||
RuntimeBindingRepository bindingRepository,
|
||||
RuntimeSessionRepository sessionRepository,
|
||||
ToolExecutionRepository executionRepository,
|
||||
RuntimeProvisioner provisioner, RuntimeTransport transport,
|
||||
RuntimePublicationVerifier publicationVerifier,
|
||||
String brokerOwnerId, Duration dispatchLeaseDuration,
|
||||
Duration v3ResultWindow, Clock clock,
|
||||
ScheduledExecutorService scheduler,
|
||||
ScheduledExecutorService renewalScheduler, AtomicBoolean closed) {
|
||||
this.bindingRegistry = bindingRegistry;
|
||||
this.executionState = executionState;
|
||||
this.bindingRepository = bindingRepository;
|
||||
this.sessionRepository = sessionRepository;
|
||||
this.executionRepository = executionRepository;
|
||||
this.provisioner = provisioner;
|
||||
this.transport = transport;
|
||||
this.publicationVerifier = publicationVerifier;
|
||||
this.brokerOwnerId = brokerOwnerId;
|
||||
this.dispatchLeaseDuration = dispatchLeaseDuration;
|
||||
this.v3ResultWindow = v3ResultWindow;
|
||||
this.clock = clock;
|
||||
this.scheduler = scheduler;
|
||||
this.renewalScheduler = renewalScheduler;
|
||||
this.closed = closed;
|
||||
}
|
||||
|
||||
boolean isInvoking(String executionCallId) {
|
||||
return invocations.contains(executionCallId);
|
||||
}
|
||||
|
||||
void cancelPendingOperations() {
|
||||
dispatches.values().forEach(future -> future.cancel(false));
|
||||
}
|
||||
|
||||
void beginDispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared) {
|
||||
beginDispatch(context, prepared, null);
|
||||
}
|
||||
|
||||
void beginDispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared, Map<String, Object> payload) {
|
||||
beginDispatch(context, prepared, payload, null);
|
||||
}
|
||||
|
||||
void beginDispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared, Map<String, Object> payload,
|
||||
RuntimePublicationGrant grant) {
|
||||
if (CsiFilesRetirementProfile.isProfile(context.binding().getRequest().getScope())
|
||||
&& (!(bindingRepository instanceof NativeCsiExecutionRepository)
|
||||
|| payload == null || !Set.of("read_file", "write_file", "edit").contains(String.valueOf(payload.get("toolName"))) || grant != null)) {
|
||||
throw new RuntimeBrokerException(501, "csi_file_dispatch_unavailable",
|
||||
"Private CSI dispatch requires original JDBC authorization and a finite file payload.", false);
|
||||
}
|
||||
if (!prepared.isCancelRequested()) {
|
||||
bindingRepository.requireHarnessAdmission(context.binding().getRequest(), context.session().getScope(),
|
||||
context.session().getHarnessSessionId(), null);
|
||||
}
|
||||
if (payload == null && ("deferred".equals(prepared.getReference().get("dispatchMode"))
|
||||
|| "deferred_v3".equals(prepared.getReference().get("dispatchMode")))
|
||||
&& !prepared.isCancelRequested()) {
|
||||
throw conflict("runtime_execution_conflict", "Deferred execution requires its original payload");
|
||||
}
|
||||
CompletableFuture<Void> created = new CompletableFuture<>();
|
||||
CompletableFuture<Void> existing = dispatches.putIfAbsent(
|
||||
prepared.getExecutionCallId(), created);
|
||||
if (existing != null) {
|
||||
return;
|
||||
}
|
||||
CompletionStage<Void> operation;
|
||||
try {
|
||||
operation = dispatch(context, prepared, payload, grant);
|
||||
} catch (RuntimeException | Error exception) {
|
||||
dispatches.remove(prepared.getExecutionCallId(), created);
|
||||
created.completeExceptionally(exception);
|
||||
if (exception instanceof RuntimeBrokerException refused) {
|
||||
throw refused;
|
||||
}
|
||||
throw unavailable("runtime_execution_dispatch_failed",
|
||||
"Runtime execution dispatch failed", exception);
|
||||
}
|
||||
operation.whenComplete((ignored, error) -> {
|
||||
dispatches.remove(prepared.getExecutionCallId(), created);
|
||||
if (error == null) {
|
||||
created.complete(null);
|
||||
} else {
|
||||
created.completeExceptionally(unwrap(error));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private CompletionStage<Void> dispatch(SessionContext context,
|
||||
ToolExecutionRecord prepared, Map<String, Object> payload,
|
||||
RuntimePublicationGrant grant) {
|
||||
ToolExecutionRecord claimed = executionRepository.claimDispatch(
|
||||
prepared.getExecutionCallId(), brokerOwnerId,
|
||||
dispatchLeaseDuration);
|
||||
if (claimed == null
|
||||
|| claimed.getState()
|
||||
!= ToolExecutionRecord.State.DISPATCHING) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
ToolExecutionRecord executing = enterExecuting(claimed);
|
||||
if (executing == null || executing.isTerminal()
|
||||
|| executing.getState()
|
||||
!= ToolExecutionRecord.State.EXECUTING
|
||||
|| !ownsDispatch(executing, claimed)) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
if (!provisioner.isUsable(context.lease())) {
|
||||
bindingRegistry.invalidateBinding(context.binding());
|
||||
executionState.markUnknown(executing);
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
DispatchRenewal renewal = new DispatchRenewal(executing);
|
||||
try {
|
||||
renewal.start();
|
||||
} catch (RuntimeException | Error exception) {
|
||||
executionState.markUnknown(executing);
|
||||
throw exception;
|
||||
}
|
||||
invocations.add(executing.getExecutionCallId());
|
||||
CompletionStage<Map<String, Object>> invocation = grant == null
|
||||
? safeStage(() -> payload == null
|
||||
? transport.execute(context.lease(), context.session(), executing.getReference())
|
||||
: transport.execute(context.lease(), context.session(),
|
||||
dispatchReference(executing, context.binding().getRequest().getScope()), payload))
|
||||
: safeStage(() -> transport.executeV3(context.lease(), context.session(),
|
||||
executing.getReference(), payload, capture(grant)))
|
||||
.handle((answer, error) -> {
|
||||
if (nonRetryableToolV3(error)) {
|
||||
throw new CompletionException(unwrap(error));
|
||||
}
|
||||
if (error == null && answer != null
|
||||
&& "settled".equals(answer.get("state"))
|
||||
&& answer.get("result") instanceof Map<?, ?> result
|
||||
&& "not_started".equals(result.get("executionStatus"))) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> notStarted = (Map<String, Object>) result;
|
||||
return notStarted;
|
||||
}
|
||||
return null;
|
||||
})
|
||||
.thenCompose(notStarted -> notStarted != null
|
||||
? CompletableFuture.completedFuture(notStarted)
|
||||
: awaitV3Result(context, executing,
|
||||
clock.instant().plus(v3ResultWindow)));
|
||||
return invocation
|
||||
.<Void>handle((result, error) -> {
|
||||
// Stop counting as running before the outcome is
|
||||
// written, so a cancel that reads that outcome does not
|
||||
// treat this finished invocation as still running.
|
||||
invocations.remove(executing.getExecutionCallId());
|
||||
if (error != null || result == null) {
|
||||
executionState.markUnknown(executing);
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
executionState.settleExecution(executing, result);
|
||||
executionState.settleUnstartedBackgroundSiblings(
|
||||
executing.getExecutionCallId(), result);
|
||||
} catch (RuntimeException exception) {
|
||||
executionState.markUnknown(executing);
|
||||
}
|
||||
return null;
|
||||
}).thenCompose(ignored -> {
|
||||
if (!com.alibaba.qwen.code.runtimebroker.protocol.RuntimeProviderKinds.SSH_PROCESS.equals(context.binding().getRequest().getProvisionerKind())) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
ToolExecutionRecord settled = executionRepository.findByExecutionCallId(executing.getExecutionCallId());
|
||||
return settled != null && settled.sameIdentity(executing)
|
||||
? acknowledgePersistedTools(context.binding(), settled).exceptionally(error -> null)
|
||||
: CompletableFuture.completedFuture(null);
|
||||
}).whenComplete((ignored, error) -> renewal.close());
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static Map<String, Object> capture(RuntimePublicationGrant grant) {
|
||||
Map<String, Object> binding = grant.binding();
|
||||
Map<String, Object> key = (Map<String, Object>) binding.get("sessionKey");
|
||||
return Map.of("tenantId", key.get("tenantId"), "sessionId", key.get("sessionId"),
|
||||
"turnId", binding.get("turnId"), "executionCallId", binding.get("executionCallId"),
|
||||
"bindingGeneration", binding.get("bindingGeneration"),
|
||||
"capturePolicy", "complete_required");
|
||||
}
|
||||
|
||||
private CompletionStage<Map<String, Object>> awaitV3Result(SessionContext context,
|
||||
ToolExecutionRecord original, Instant deadline) {
|
||||
CompletableFuture<Map<String, Object>> result = new CompletableFuture<>();
|
||||
pollV3Result(context, original, deadline, result, 0);
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* The v3 result-poll backoff: 100ms doubling, capped at 2s. The cap
|
||||
* bounds the scheduled delay between rounds for one execution; rounds
|
||||
* for all executions share the single coordination thread, so pick-up
|
||||
* lateness under concurrency also carries that thread's queue wait.
|
||||
*/
|
||||
static long v3PollDelayMillis(int attempt) {
|
||||
return Math.min(2_000, 100L << Math.min(attempt, 6));
|
||||
}
|
||||
|
||||
private void pollV3Result(SessionContext context, ToolExecutionRecord original,
|
||||
Instant deadline, CompletableFuture<Map<String, Object>> answer,
|
||||
int attempt) {
|
||||
if (answer.isDone()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
Map<String, Object> finished = publicationVerifier.finished(original);
|
||||
if (finished != null) {
|
||||
answer.complete(finished);
|
||||
return;
|
||||
}
|
||||
ToolExecutionRecord current = executionRepository.findByExecutionCallId(
|
||||
original.getExecutionCallId());
|
||||
if (current == null || !ownsDispatch(current, original)
|
||||
|| !current.getBindingId().equals(context.binding().getBindingId())
|
||||
|| current.getRuntimeGeneration() != context.binding().getGeneration()
|
||||
|| !provisioner.isUsable(context.lease()) || !clock.instant().isBefore(deadline)) {
|
||||
throw evidenceUnavailable();
|
||||
}
|
||||
safeStage(() -> transport.statusV3(context.lease(), context.session(),
|
||||
original.getReference(), 0)).whenComplete((status, error) -> {
|
||||
if (answer.isDone()) {
|
||||
return;
|
||||
}
|
||||
if (nonRetryableToolV3(error)) {
|
||||
answer.completeExceptionally(unwrap(error));
|
||||
return;
|
||||
}
|
||||
if (error == null && status != null && "settled".equals(status.get("state"))
|
||||
&& status.get("result") instanceof Map<?, ?> result
|
||||
&& "not_started".equals(result.get("executionStatus"))) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> saved = (Map<String, Object>) result;
|
||||
answer.complete(saved);
|
||||
return;
|
||||
}
|
||||
// A background Shell start settles its handle as the
|
||||
// detached family: nothing is ever published for it,
|
||||
// so the status answer is its only durable settle.
|
||||
if (error == null && status != null && "settled".equals(status.get("state"))
|
||||
&& status.get("result") instanceof Map<?, ?> detached
|
||||
&& isDetachedCapture(detached)) {
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> saved = (Map<String, Object>) detached;
|
||||
answer.complete(saved);
|
||||
return;
|
||||
}
|
||||
if (error == null && status != null && "unknown".equals(status.get("state"))) {
|
||||
answer.completeExceptionally(evidenceUnavailable());
|
||||
return;
|
||||
}
|
||||
// Each round costs two repository reads and one worker
|
||||
// call; back off instead of pinning them at 10/s for
|
||||
// the whole window.
|
||||
long delay = v3PollDelayMillis(attempt);
|
||||
try {
|
||||
scheduler.schedule(() -> pollV3Result(context,
|
||||
original, deadline, answer, attempt + 1),
|
||||
delay, TimeUnit.MILLISECONDS);
|
||||
} catch (RuntimeException failure) {
|
||||
answer.completeExceptionally(failure);
|
||||
}
|
||||
});
|
||||
} catch (RuntimeException failure) {
|
||||
answer.completeExceptionally(failure);
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean nonRetryableToolV3(Throwable error) {
|
||||
return error != null && unwrap(error) instanceof RuntimeBrokerException failure
|
||||
&& !failure.isRetryable();
|
||||
}
|
||||
|
||||
private static Map<String, Object> dispatchReference(ToolExecutionRecord record, RuntimeScope scope) {
|
||||
if (CsiFilesRetirementProfile.isProfile(scope)) {
|
||||
Map<String, Object> reference = new LinkedHashMap<>(record.getReference());
|
||||
reference.put("executionCallId", record.getExecutionCallId());
|
||||
return Map.copyOf(reference);
|
||||
}
|
||||
if (!Integer.valueOf(3).equals(record.getReference().get("runtimeProtocol"))) {
|
||||
return record.getReference();
|
||||
}
|
||||
Map<String, Object> reference = new LinkedHashMap<>(record.getReference());
|
||||
reference.put("executionCallId", record.getExecutionCallId());
|
||||
return Map.copyOf(reference);
|
||||
}
|
||||
|
||||
private ToolExecutionRecord enterExecuting(
|
||||
ToolExecutionRecord claimed) {
|
||||
ToolExecutionRecord current = claimed;
|
||||
for (int attempt = 0; attempt < MAX_CAS_ATTEMPTS; attempt++) {
|
||||
if (current == null || current.isTerminal()
|
||||
|| current.getState()
|
||||
== ToolExecutionRecord.State.UNKNOWN
|
||||
|| !ownsDispatch(current, claimed)) {
|
||||
return current;
|
||||
}
|
||||
if (current.getState() != ToolExecutionRecord.State.DISPATCHING) {
|
||||
return null;
|
||||
}
|
||||
ToolExecutionRecord replacement;
|
||||
if (current.isCancelRequested()) {
|
||||
replacement = current.withResult(
|
||||
current.cancellationBeforeDispatch(),
|
||||
current.getLastSequence(), clock.instant());
|
||||
} else {
|
||||
replacement = current.withState(
|
||||
ToolExecutionRecord.State.EXECUTING, false);
|
||||
}
|
||||
ToolExecutionRecord updated = current.isCancelRequested()
|
||||
? executionRepository.compareAndSet(current, replacement, brokerOwnerId,
|
||||
claimed.getDispatchGeneration())
|
||||
: bindingRepository.authorizeDispatch(sessionRepository, executionRepository, current,
|
||||
brokerOwnerId, claimed.getDispatchGeneration());
|
||||
if (updated != null) {
|
||||
return updated;
|
||||
}
|
||||
current = executionRepository.findByExecutionCallId(
|
||||
claimed.getExecutionCallId());
|
||||
}
|
||||
executionState.markUnknown(claimed);
|
||||
return null;
|
||||
}
|
||||
|
||||
CompletionStage<Void> acknowledgePersistedTools(RuntimeBindingRecord binding, ToolExecutionRecord execution) {
|
||||
if (!com.alibaba.qwen.code.runtimebroker.protocol.RuntimeProviderKinds.SSH_PROCESS.equals(binding.getRequest().getProvisionerKind())
|
||||
|| !execution.isSettled() || execution.getResult() == null
|
||||
|| ProviderRuntimeProtocol.isReference(execution.getReference())) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
Map<String, Object> noStart = sshNoStartReceipt(binding, execution);
|
||||
if (noStart == null && (execution.getResult().containsKey("capture")
|
||||
|| Integer.valueOf(3).equals(execution.getReference().get("runtimeProtocol"))
|
||||
|| "deferred_v3".equals(execution.getReference().get("dispatchMode")))) {
|
||||
return CompletableFuture.completedFuture(null);
|
||||
}
|
||||
RuntimeSessionRecord session = sessionRepository.findById(binding.getRequest().getScope(),
|
||||
execution.getRuntimeSessionId());
|
||||
if (binding.getLease() == null || session == null || !execution.belongsTo(session)
|
||||
|| !binding.getBindingId().equals(execution.getBindingId())
|
||||
|| binding.getGeneration() != execution.getRuntimeGeneration()
|
||||
|| !binding.getRequest().getIsolationKey().equals(execution.getHarnessSessionId())) {
|
||||
return failed(conflict("workspace_close_identity_unverified", "Persisted Tool ownership conflicts"));
|
||||
}
|
||||
if (noStart != null) {
|
||||
CompletionStage<Map<String, Object>> acknowledgement = "deferred_v3".equals(execution.getReference().get("dispatchMode"))
|
||||
? transport.acknowledgeV3(binding.getLease(), session.getSession(), execution.getReference(), noStart)
|
||||
: transport.acknowledge(binding.getLease(), session.getSession(), execution.getReference(), noStart);
|
||||
return acknowledgement.thenAccept(answer -> {
|
||||
if (answer == null || !Set.of("settled", "unknown").contains(answer.get("state"))) {
|
||||
throw conflict("runtime_execution_conflict", "Original Runtime did not confirm the no-start acknowledgement");
|
||||
}
|
||||
});
|
||||
}
|
||||
return transport.acknowledgeV2(binding.getLease(), session.getSession(), execution.getReference());
|
||||
}
|
||||
|
||||
private static boolean ownsDispatch(ToolExecutionRecord current,
|
||||
ToolExecutionRecord claimed) {
|
||||
return claimed.getDispatchGeneration()
|
||||
== current.getDispatchGeneration()
|
||||
&& claimed.getDispatchOwner().equals(
|
||||
current.getDispatchOwner());
|
||||
}
|
||||
|
||||
private final class DispatchRenewal implements AutoCloseable {
|
||||
// Same virtual-thread rule as BindingRenewal: guarded repository
|
||||
// calls must not pin the caller's carrier while they block.
|
||||
private final ReentrantLock monitor = new ReentrantLock();
|
||||
private final ToolExecutionRecord original;
|
||||
private ScheduledFuture<?> task;
|
||||
|
||||
DispatchRenewal(ToolExecutionRecord original) {
|
||||
this.original = original;
|
||||
}
|
||||
|
||||
void start() {
|
||||
monitor.lock();
|
||||
try {
|
||||
long delay = renewalDelayMillis(dispatchLeaseDuration);
|
||||
task = renewalScheduler.scheduleWithFixedDelay(this::renew,
|
||||
delay, delay, TimeUnit.MILLISECONDS);
|
||||
} finally {
|
||||
monitor.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void renew() {
|
||||
monitor.lock();
|
||||
try {
|
||||
if (closed.get()) {
|
||||
closeLocked();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
ToolExecutionRecord renewed =
|
||||
executionRepository.renewDispatch(
|
||||
original.getExecutionCallId(), brokerOwnerId,
|
||||
original.getDispatchGeneration(), dispatchLeaseDuration);
|
||||
if (renewed == null) {
|
||||
executionRepository.claimDispatch(original.getExecutionCallId(),
|
||||
brokerOwnerId, dispatchLeaseDuration);
|
||||
closeLocked();
|
||||
} else if (!renewed.sameIdentity(original) || !renewed.sameAuthorization(original)
|
||||
|| !ownsDispatch(renewed, original)) {
|
||||
closeLocked();
|
||||
}
|
||||
} catch (RuntimeException exception) {
|
||||
if (exception instanceof RuntimeBrokerException failure && !failure.isRetryable()
|
||||
&& (failure.getCode().startsWith("csi_")
|
||||
|| "runtime_admission_closed".equals(failure.getCode()))) {
|
||||
closeLocked();
|
||||
}
|
||||
// A transient repository failure does not prove claim loss.
|
||||
}
|
||||
} finally {
|
||||
monitor.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
monitor.lock();
|
||||
try {
|
||||
closeLocked();
|
||||
} finally {
|
||||
monitor.unlock();
|
||||
}
|
||||
}
|
||||
|
||||
private void closeLocked() {
|
||||
if (task != null) {
|
||||
task.cancel(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -58,6 +58,6 @@ public final class BrokerServiceTestAccess {
|
||||
new RuntimeBindingRegistry.SessionContext(session, binding, lease)));
|
||||
}
|
||||
|
||||
public static long pollDelay(int attempt) { return RuntimeBrokerService.v3PollDelayMillis(attempt); }
|
||||
public static long pollDelay(int attempt) { return RuntimeExecutionDispatcher.v3PollDelayMillis(attempt); }
|
||||
public static final java.time.Duration UNKNOWN_LOOKUP_COOLDOWN = ExecutionObservationCooldown.WINDOW;
|
||||
}
|
||||
+213
@@ -0,0 +1,213 @@
|
||||
package com.alibaba.qwen.code.runtimebroker.service;
|
||||
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeLease;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeProvisionRequest;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeProvisioner;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimePublicationVerifier;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeScope;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeSession;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeSessionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeTransport;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
|
||||
import com.alibaba.qwen.code.runtimebroker.service.RuntimeBindingRegistry.SessionContext;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.Timeout;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
|
||||
import java.time.Clock;
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneOffset;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertThrows;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.doAnswer;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
class RuntimeExecutionDispatcherTest {
|
||||
@Test
|
||||
@Timeout(15)
|
||||
void concurrentDispatchesShareReservationWhileClaimIsPending() throws Exception {
|
||||
var fixture = new Fixture();
|
||||
var entered = new CountDownLatch(1);
|
||||
var release = new CountDownLatch(1);
|
||||
var failure = new AtomicReference<Throwable>();
|
||||
when(fixture.executions.claimDispatch("execution", "broker", Fixture.LEASE)).thenAnswer(call -> {
|
||||
entered.countDown();
|
||||
assertTrue(release.await(5, TimeUnit.SECONDS), "claim was not released");
|
||||
return fixture.prepared;
|
||||
});
|
||||
Thread first = Thread.ofPlatform().start(() -> {
|
||||
try {
|
||||
fixture.dispatcher.beginDispatch(fixture.context, fixture.prepared);
|
||||
} catch (Throwable error) {
|
||||
failure.set(error);
|
||||
}
|
||||
});
|
||||
try {
|
||||
assertTrue(entered.await(5, TimeUnit.SECONDS), "dispatch never reached its claim");
|
||||
assertFalse(fixture.dispatcher.isInvoking("execution"));
|
||||
fixture.dispatcher.beginDispatch(fixture.context, fixture.prepared);
|
||||
verify(fixture.executions).claimDispatch("execution", "broker", Fixture.LEASE);
|
||||
verify(fixture.transport, never()).execute(any(), any(), any());
|
||||
release.countDown();
|
||||
first.join(2000);
|
||||
assertFalse(first.isAlive());
|
||||
assertNull(failure.get());
|
||||
assertTrue(fixture.dispatcher.isInvoking("execution"));
|
||||
verify(fixture.transport).execute(fixture.lease, fixture.session, Fixture.REFERENCE);
|
||||
} finally {
|
||||
release.countDown();
|
||||
first.join(2000);
|
||||
fixture.answer.complete(Fixture.RESULT);
|
||||
}
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(booleans = {false, true})
|
||||
void invocationMembershipIsRemovedBeforeOutcomePersistence(boolean failed) {
|
||||
var fixture = new Fixture();
|
||||
var invokingAtPersistence = new AtomicBoolean(true);
|
||||
if (failed) {
|
||||
doAnswer(call -> {
|
||||
invokingAtPersistence.set(fixture.dispatcher.isInvoking("execution"));
|
||||
return null;
|
||||
}).when(fixture.state).markUnknown(fixture.executing);
|
||||
} else {
|
||||
doAnswer(call -> {
|
||||
invokingAtPersistence.set(fixture.dispatcher.isInvoking("execution"));
|
||||
return null;
|
||||
}).when(fixture.state).settleExecution(fixture.executing, Fixture.RESULT);
|
||||
}
|
||||
fixture.dispatcher.beginDispatch(fixture.context, fixture.prepared);
|
||||
assertTrue(fixture.dispatcher.isInvoking("execution"));
|
||||
if (failed) {
|
||||
fixture.answer.completeExceptionally(new IllegalStateException("worker unavailable"));
|
||||
verify(fixture.state).markUnknown(fixture.executing);
|
||||
verify(fixture.state, never()).settleExecution(any(), any());
|
||||
} else {
|
||||
fixture.answer.complete(Fixture.RESULT);
|
||||
verify(fixture.state).settleExecution(fixture.executing, Fixture.RESULT);
|
||||
verify(fixture.state).settleUnstartedBackgroundSiblings("execution", Fixture.RESULT);
|
||||
verify(fixture.state, never()).markUnknown(any());
|
||||
}
|
||||
assertFalse(invokingAtPersistence.get(), "persistence observed a completed invocation as running");
|
||||
assertFalse(fixture.dispatcher.isInvoking("execution"));
|
||||
verify(fixture.renewal).cancel(false);
|
||||
}
|
||||
|
||||
@Test
|
||||
void synchronousClaimFailureReleasesReservationForRetry() {
|
||||
var fixture = new Fixture();
|
||||
when(fixture.executions.claimDispatch("execution", "broker", Fixture.LEASE))
|
||||
.thenThrow(new IllegalStateException("repository unavailable"))
|
||||
.thenReturn(fixture.prepared);
|
||||
var failure = assertThrows(RuntimeBrokerException.class,
|
||||
() -> fixture.dispatcher.beginDispatch(fixture.context, fixture.prepared));
|
||||
assertEquals("runtime_execution_dispatch_failed", failure.getCode());
|
||||
assertFalse(fixture.dispatcher.isInvoking("execution"));
|
||||
fixture.dispatcher.beginDispatch(fixture.context, fixture.prepared);
|
||||
verify(fixture.executions, times(2)).claimDispatch("execution", "broker", Fixture.LEASE);
|
||||
verify(fixture.transport).execute(fixture.lease, fixture.session, Fixture.REFERENCE);
|
||||
assertTrue(fixture.dispatcher.isInvoking("execution"));
|
||||
fixture.answer.complete(Fixture.RESULT);
|
||||
}
|
||||
|
||||
@Test
|
||||
void cancellingReservationsKeepsPhysicalInvocationAndBorrowedResources() {
|
||||
var fixture = new Fixture();
|
||||
fixture.dispatcher.beginDispatch(fixture.context, fixture.prepared);
|
||||
fixture.dispatcher.cancelPendingOperations();
|
||||
assertFalse(fixture.answer.isCancelled());
|
||||
assertTrue(fixture.dispatcher.isInvoking("execution"));
|
||||
assertFalse(fixture.closed.get());
|
||||
verify(fixture.scheduler, never()).shutdown();
|
||||
verify(fixture.scheduler, never()).shutdownNow();
|
||||
verify(fixture.renewalScheduler, never()).shutdown();
|
||||
verify(fixture.renewalScheduler, never()).shutdownNow();
|
||||
verify(fixture.provisioner, never()).close();
|
||||
fixture.answer.complete(Fixture.RESULT);
|
||||
assertFalse(fixture.dispatcher.isInvoking("execution"));
|
||||
verify(fixture.state).settleExecution(fixture.executing, Fixture.RESULT);
|
||||
verify(fixture.renewal).cancel(false);
|
||||
}
|
||||
|
||||
private static final class Fixture {
|
||||
private static final Duration LEASE = Duration.ofSeconds(60);
|
||||
private static final Map<String, Object> REFERENCE = Map.of("sessionId", "runtime");
|
||||
private static final Map<String, Object> RESULT = Map.of("executionStatus", "success");
|
||||
|
||||
private final RuntimeExecutionState state = mock(RuntimeExecutionState.class);
|
||||
private final ToolExecutionRepository executions = mock(ToolExecutionRepository.class);
|
||||
private final RuntimeProvisioner provisioner = mock(RuntimeProvisioner.class);
|
||||
private final RuntimeTransport transport = mock(RuntimeTransport.class);
|
||||
private final ScheduledExecutorService scheduler = mock(ScheduledExecutorService.class);
|
||||
private final ScheduledExecutorService renewalScheduler = mock(ScheduledExecutorService.class);
|
||||
private final ScheduledFuture<?> renewal = mock(ScheduledFuture.class);
|
||||
private final AtomicBoolean closed = new AtomicBoolean();
|
||||
private final SessionContext context = mock(SessionContext.class);
|
||||
private final RuntimeSession session = mock(RuntimeSession.class);
|
||||
private final RuntimeLease lease = mock(RuntimeLease.class);
|
||||
private final ToolExecutionRecord prepared = mock(ToolExecutionRecord.class);
|
||||
private final ToolExecutionRecord executing = mock(ToolExecutionRecord.class);
|
||||
private final CompletableFuture<Map<String, Object>> answer = new CompletableFuture<>();
|
||||
private final RuntimeExecutionDispatcher dispatcher;
|
||||
|
||||
private Fixture() {
|
||||
var registry = mock(RuntimeBindingRegistry.class);
|
||||
var bindings = mock(RuntimeBindingRepository.class);
|
||||
var sessions = mock(RuntimeSessionRepository.class);
|
||||
var binding = mock(RuntimeBindingRecord.class);
|
||||
var scope = new RuntimeScope("tenant", "workspace", "1", "/workspace",
|
||||
WorkspaceExecutionProfile.CAPABILITY_DIGEST, "session");
|
||||
when(binding.getRequest()).thenReturn(new RuntimeProvisionRequest(scope, "harness", "local"));
|
||||
when(context.binding()).thenReturn(binding);
|
||||
when(context.session()).thenReturn(session);
|
||||
when(context.lease()).thenReturn(lease);
|
||||
when(session.getScope()).thenReturn(scope);
|
||||
when(session.getHarnessSessionId()).thenReturn("harness");
|
||||
for (var record : new ToolExecutionRecord[] {prepared, executing}) {
|
||||
when(record.getExecutionCallId()).thenReturn("execution");
|
||||
when(record.getReference()).thenReturn(REFERENCE);
|
||||
when(record.getDispatchOwner()).thenReturn("broker");
|
||||
when(record.getDispatchGeneration()).thenReturn(7L);
|
||||
}
|
||||
when(prepared.getState()).thenReturn(ToolExecutionRecord.State.DISPATCHING);
|
||||
when(executing.getState()).thenReturn(ToolExecutionRecord.State.EXECUTING);
|
||||
when(executions.claimDispatch("execution", "broker", LEASE)).thenReturn(prepared);
|
||||
when(bindings.authorizeDispatch(sessions, executions, prepared, "broker", 7L)).thenReturn(executing);
|
||||
when(provisioner.isUsable(lease)).thenReturn(true);
|
||||
when(transport.execute(lease, session, REFERENCE)).thenReturn(answer);
|
||||
when(renewalScheduler.scheduleWithFixedDelay(any(Runnable.class), anyLong(), anyLong(),
|
||||
eq(TimeUnit.MILLISECONDS))).thenAnswer(call -> renewal);
|
||||
dispatcher = new RuntimeExecutionDispatcher(registry, state, bindings, sessions, executions,
|
||||
provisioner, transport, mock(RuntimePublicationVerifier.class), "broker", LEASE,
|
||||
Duration.ofMinutes(30), Clock.fixed(Instant.parse("2026-11-01T00:00:00Z"), ZoneOffset.UTC),
|
||||
scheduler, renewalScheduler, closed);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user