mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
feat(managed-agent): verify tool publications asynchronously (#13654)
* feat(managed-agent): verify tool publications asynchronously * codex: fix CI failure on PR #13654 * codex: address PR review feedback (#13654) * codex: address PR review feedback (#13654) * codex: address PR review feedback (#13654) * codex: address PR review feedback (#13654) Preserve the current main actor-key migration and append async verification at V57. Cover both the legacy-to-main and main-to-async upgrade histories. * codex: address PR review feedback (#13654) * fix(managed-agent): recover lapsed activations and failed finishes A renewable activation horizon was recorded as a permanent failure, and a permanently failed finish left the publication stuck in FINISHING. Give the MariaDB job enough time for the new async integration tests. * fix(managed-agent): verify accepted CSI claims after retirement Async verification used PRODUCE, so a settled CSI execution with a retirement row permanently failed work that the claim path would finish. * fix(managed-agent): defer expired grants and drop failed terminal bytes A real activation lapse expires the publication grant on the same clock, so the grant check failed the operation before the activation deferral. A failed finish also refunded its charge while leaving a non-VERIFIED object that blocked collection. * fix(managed-agent): keep object deletes and scan errors off the failure transaction A failed finish deleted its object inside the database transaction, so a rollback could drop the bytes without recording FAILED. A handler exception also aborted the rest of the verification page and retried the same row. * fix(managed-agent): preserve transient retries and writer fences Co-authored-by: Qwen-Coder <[email protected]> * fix(managed-agent): distinguish verification corruption from authority failures Keep accepted verification pending during session recovery and classify corruption only from verified byte mismatches. Preserve permanent writer fences and retry unclassified reads through complete capture validation. Co-authored-by: Qwen-Coder <[email protected]> * fix(managed-agent): fail verification on recovery blocks Restore permanent failure for accepted asynchronous work when Session recovery is blocked. Exercise the production recovery-block entry point without resetting the Session to READY, and document the fencing policy in both languages. Co-authored-by: Qwen-Coder <[email protected]> --------- Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
jinyeandQwen-Coder authored and GitHub committed
1 parent
b675c97461
commit
9f14e7d9bb
23 files changed
+2792
-89
No files matched your search
@@ -0,0 +1,156 @@
|
||||
# Managed tool publication asynchronous verification
|
||||
|
||||
[English](2026-10-07-managed-tool-publication-async-verification.md) | [简体中文](2026-10-07-managed-tool-publication-async-verification.zh-CN.md)
|
||||
|
||||
Status: implementation for issue [#13242](https://github.com/QwenLM/qwen-code/issues/13242).
|
||||
|
||||
## Problem and scope
|
||||
|
||||
Publication uploads read their stored bytes back on the HTTP thread. Seal,
|
||||
prefix, and finish scan captured streams on that same thread. The existing
|
||||
operation journal has leases and idempotency receipts, but cannot reconstruct a
|
||||
seal request or an inline terminal after a process restart. The worker HTTP
|
||||
adapter already polls operations; its callers must continue awaiting verified
|
||||
receipts before declaring a stream sealed or an execution settled.
|
||||
|
||||
Move producer publish, resource publish, seal, prefix, and finish verification
|
||||
into a durable, bounded verifier. Keep writer finished/admission preparation,
|
||||
receipt validation, ACK verification, and actual content-read integrity checks.
|
||||
Do not add a general job system or change Shell execution/recovery semantics.
|
||||
|
||||
## Contract and compatibility
|
||||
|
||||
New clients send `X-Qwen-Tool-Publication-Async: 1`. The server admits new
|
||||
asynchronous operations only with that header and
|
||||
`qwen.managed-agent.tool-publication.async-verification-enabled=true` (default
|
||||
false). The mode is saved when an operation is created. Replays and recovery
|
||||
use that saved mode even after the admission flag is disabled. Legacy callers
|
||||
retain synchronous responses. The header changes neither request digests nor
|
||||
binding, grant, or installation schemas.
|
||||
|
||||
The server returns HTTP 202 with `state: PENDING` only after all inputs and the
|
||||
verification-ready record are committed. Completed responses use HTTP 200.
|
||||
|
||||
| State | Meaning |
|
||||
| --------- | ------------------------------------------------------------------- |
|
||||
| PENDING | Waiting, running, or retrying accepted verification; poll only. |
|
||||
| SUCCEEDED | Verified historical receipt, in `receipt`. |
|
||||
| FAILED | Permanent result, in `error: {status, code}`. |
|
||||
| RETRYABLE | Incomplete upload or legacy claim needs the identical request. |
|
||||
| EXPIRED | Original effective deadline elapsed; explicit recovery is required. |
|
||||
|
||||
Success and failure take precedence over elapsed deadlines. FAILED cannot be
|
||||
reclaimed or recovered. The adapter decodes FAILED outside transport retry
|
||||
handling, including quota status 507 and storage denial. It keeps its serial
|
||||
queue, 250ms polls, 30-minute observation deadline, and three explicit
|
||||
recoveries. Prefix remains unrecoverable after expiry.
|
||||
|
||||
An asynchronous exact success replay returns its historical receipt without
|
||||
object I/O. It still validates identity, request digest, and known quarantine.
|
||||
SUCCEEDED proves verification at completion, not that storage was never changed
|
||||
later. Finish, reads, receipt verification, and ACK keep checking current bytes.
|
||||
An already verified identical slot may return success directly without creating
|
||||
an operation for a new idempotency ID.
|
||||
|
||||
## Persistence and execution
|
||||
|
||||
An additive migration extends the existing operation table with execution mode,
|
||||
verification inputs, readiness, next-attempt time, and safe failure information,
|
||||
plus a due-work index. Existing rows default to synchronous mode. Seal stores its
|
||||
count, byte length, and digest; terminal and inline resource bytes are saved
|
||||
before verification. Object PUT and readiness commit must complete before 202.
|
||||
Unknown PUT results remain retryable and keep quota charged.
|
||||
|
||||
A separate executor defaults to two verifier threads, controlled by
|
||||
`verification-concurrency`. Commit wakes it immediately; it drains work and a
|
||||
one-second database scan recovers missed wakeups and restarts. It must not occupy
|
||||
the GC/retention scheduler. Workers select a bounded unlocked candidate page,
|
||||
then claim each operation using the existing parent/publication/operation lock
|
||||
order. Object reads and hashing happen outside SQL transactions.
|
||||
|
||||
Workers use an internal accepted-operation authorization entry point and never
|
||||
persist plaintext tokens. Claim, heartbeat, and completion retain binding,
|
||||
writer, activation, runtime, and CSI checks using journal-head authorization.
|
||||
The async admission flag requires journal-head authorization to be enabled.
|
||||
Status remains a read-only scope/token query that neither starts nor renews work.
|
||||
|
||||
Reuse claim epochs, leases, and byte-aware deadlines. Queue time counts toward
|
||||
the original deadline. A ready operation with no current lease is PENDING, not
|
||||
RETRYABLE. Temporary I/O failure clears only its lease and schedules a one-second
|
||||
retry; it retains active_operation_id throughout the live attempt. This also
|
||||
keeps prefix input stable. Expired work stops running; recovery schedules the
|
||||
same input with a new bounded recovery deadline and epoch, without changing the
|
||||
original deadline, resource identity, or charged quota.
|
||||
|
||||
Verification keeps existing segment and aggregate SHA-256, ordinal/count,
|
||||
manifest closure, and terminal checks. Only a short fenced transaction installs
|
||||
VERIFIED objects, seal rows, FINISHED, and success receipts. Pending resources
|
||||
cannot enter admission or settlement. Finish preserves its frozen predecessor;
|
||||
a permanently failed predecessor fails finish rather than looping indefinitely.
|
||||
|
||||
Permanent validation/storage-denial/fencing failures persist safe status/code
|
||||
without raw storage messages or credentials.
|
||||
Session recovery blocks permanently fail accepted verification; the product has
|
||||
no transition from a blocked Session back to READY. Authorization failure alone
|
||||
does not certify object-byte corruption.
|
||||
A lapsed activation horizon is retried when the activation id and epoch still
|
||||
match, because renewal keeps that identity. An expired writer lease stays a
|
||||
permanent fence. A permanently failed finish stays FAILED, leaves FINISHING,
|
||||
and retires its terminal candidate so a later finish can be admitted.
|
||||
Accepted tasks whose original CSI binding was released or whose CSI authority
|
||||
was fenced fail permanently and leave the queue. Replica credential-key and SQL
|
||||
failures remain retryable, rather than being mistaken for lost runtime identity.
|
||||
Candidate-local input or authorization errors are isolated: an unclaimed task is
|
||||
failed, deferred for one second, or expired under its unchanged epoch before the
|
||||
scan continues to later candidates. A failed SQL transition aborts that scan.
|
||||
Before the deadline, an expired attempt may only defer under the same epoch; it
|
||||
cannot publish failure or quarantine. Retry times are capped at the deadline.
|
||||
Integrity quarantine and all
|
||||
success/failure writes require the current live epoch. A stale worker cannot
|
||||
overwrite a successor's outcome. Read leases, retirement, unresolved PUTs,
|
||||
candidate/quarantined object protection, and GC quota accounting remain intact.
|
||||
Permanent FAILED operations do not block collection of an accepted complete,
|
||||
referenced publication. Pending operations (including expired ones), unresolved
|
||||
PUTs, candidate or quarantined objects, and all other retention protections still
|
||||
block collection; collection does not revive a failed operation.
|
||||
|
||||
## Rollout and observability
|
||||
|
||||
1. Apply the additive migration and upgrade every server with async admission
|
||||
disabled.
|
||||
2. Upgrade clients; they continue accepting old direct receipts.
|
||||
3. Enable journal-head authorization after all writers maintain its head columns,
|
||||
then enable async admission. Long-lived older clients remain synchronous.
|
||||
4. For rollback, disable new async admission while workers continue draining
|
||||
existing operations. Drain or resolve those operations before downgrading
|
||||
server binaries; keep journal-head authorization enabled during drainage.
|
||||
|
||||
Record queue wait, verification duration, retries, failures, and expiration in
|
||||
safe structured logs. Queue wait uses database timestamps; verification duration
|
||||
uses the monotonic process clock. Retry logs identify the scope and operation
|
||||
with the exception class only. Treat the configured throughput budget and
|
||||
concurrency as operational limits, not a promise that all requests finish faster end to end.
|
||||
|
||||
## Validation and acceptance
|
||||
|
||||
Block object reads and prove async HTTP responses arrive before verification;
|
||||
block PUT and prove acceptance waits for persistence. Restart the verifier
|
||||
without reposting inputs, including seal and both inline/object terminal forms.
|
||||
Exercise two-worker competition, lease takeover, stale success/failure,
|
||||
predecessor failures, stable prefix retries, and expiry/recovery identity/quota.
|
||||
|
||||
Cover missing/extra segments, digest changes, invalid manifests, temporary I/O,
|
||||
fenced owners, retirement, FAILED stability and error classification. Test all
|
||||
new/old client and server combinations, disabling admission while work drains,
|
||||
historical success replay followed by corruption detection on actual reads, and
|
||||
no journal-body reads during healthy-head heartbeat/status polls. Run focused
|
||||
Java/CLI tests, real MySQL recovery/competition tests, build, typecheck, bundle,
|
||||
and full-diff self-audit. E2E plan/results live under `.qwen/e2e-tests/`.
|
||||
|
||||
## Affected components and decisions
|
||||
|
||||
The Java publication catalog, producer controller, authorization store,
|
||||
configuration and migration change together with the CLI publication HTTP
|
||||
adapter and focused tests. No unresolved product decisions remain. Persisted
|
||||
work, bounded execution, strict success fences, rolling compatibility, and
|
||||
historical replay are the chosen defaults.
|
||||
@@ -0,0 +1,118 @@
|
||||
# 工具输出发布异步校验
|
||||
|
||||
[English](2026-10-07-managed-tool-publication-async-verification.md) | [简体中文](2026-10-07-managed-tool-publication-async-verification.zh-CN.md)
|
||||
|
||||
状态：实现 issue [#13242](https://github.com/QwenLM/qwen-code/issues/13242)。
|
||||
|
||||
## 问题与范围
|
||||
|
||||
发布上传会在 HTTP 线程读回存储字节；seal、prefix、finish 也在该线程扫描捕获流。
|
||||
现有 operation 日志具有租约和幂等凭据，但进程重启后无法重建 seal 请求或 inline
|
||||
terminal。worker HTTP adapter 已支持轮询；上层仍须等待验证后的凭据，才能声明
|
||||
流已封存或执行已 settled。
|
||||
|
||||
将 producer publish、resource publish、seal、prefix、finish 的校验移入持久化、
|
||||
有并发上限的 verifier。保留 writer finished/admission prepare、receipt 校验、
|
||||
ACK 校验及实际内容读取的完整性检查。不增加通用任务系统，不改变 Shell 执行和恢复。
|
||||
|
||||
## 契约与兼容
|
||||
|
||||
新客户端发送 `X-Qwen-Tool-Publication-Async: 1`。只有该请求头和服务端
|
||||
`qwen.managed-agent.tool-publication.async-verification-enabled=true` 同时成立
|
||||
才接收新的异步 operation；开关默认 false。首次创建时保存执行模式，重放和恢复
|
||||
沿用该模式，关闭接收开关也不改变已有任务。旧客户端保留同步响应。请求头不参与
|
||||
请求摘要，不改变 binding、grant 或 installation schema。
|
||||
|
||||
仅在全部输入与 verification-ready 记录提交后，返回 HTTP 202、`state: PENDING`。
|
||||
完成结果使用 HTTP 200。
|
||||
|
||||
| 状态 | 含义 |
|
||||
| --------- | -------------------------------------------- |
|
||||
| PENDING | 已接受，等待、运行或重试校验；客户端只轮询。 |
|
||||
| SUCCEEDED | 历史校验成功凭据，位于 `receipt`。 |
|
||||
| FAILED | 永久结果，位于 `error: {status, code}`。 |
|
||||
| RETRYABLE | 上传不完整或旧同步 claim 需要相同请求重试。 |
|
||||
| EXPIRED | 原有效 deadline 已到，需要显式恢复。 |
|
||||
|
||||
成功和失败优先于期限判断。FAILED 不得重新领取或恢复。adapter 在网络重试处理之外
|
||||
解析 FAILED，包括配额状态 507 和存储拒绝；保留串行队列、250ms 轮询、30 分钟
|
||||
观察期限和三次显式恢复。prefix 到期后仍不可恢复。
|
||||
|
||||
异步精确成功重放返回历史凭据，不读取对象，但继续检查身份、请求摘要和已知隔离。
|
||||
SUCCEEDED 证明完成时验证成功，不证明存储之后从未改变。finish、实际读取、receipt
|
||||
校验和 ACK 继续检查当前字节。同一已验证 slot 的相同内容可直接返回成功，不必为
|
||||
新的幂等 ID 创建 operation。
|
||||
|
||||
## 持久化与执行
|
||||
|
||||
增量 migration 为现有 operation 表添加执行模式、校验输入、就绪标记、下次尝试
|
||||
时间及安全失败信息，并添加待执行索引。旧行默认同步模式。seal 保存 count、byte
|
||||
length、digest；terminal 和 inline resource 原始字节在校验前保存。对象 PUT 和
|
||||
就绪事务完成后才能返回 202。未知 PUT 结果仍可重试，且继续占用配额。
|
||||
|
||||
独立执行器默认两个校验线程，由 `verification-concurrency` 控制。提交后立即唤醒，
|
||||
持续处理任务；每秒数据库扫描兜底处理丢失唤醒和重启。不得占用 GC/retention 调度器。
|
||||
worker 先无锁选择有界候选页，再按现有 parent/publication/operation 锁顺序领取。
|
||||
对象读取与散列在 SQL 事务外执行。
|
||||
|
||||
worker 使用仅供已接受 operation 的内部授权入口，不存明文 token。领取、心跳和
|
||||
提交继续通过 journal-head authorization 检查 binding、writer、activation、
|
||||
runtime 与 CSI。开启异步接收要求已开启 journal-head authorization。状态查询
|
||||
仍为只读 scope/token 检查，不启动任务、不续期。
|
||||
|
||||
复用 claim epoch、租约和按字节计算的 deadline。排队时间计入原 deadline。已就绪
|
||||
而无有效租约的任务为 PENDING，不是 RETRYABLE。临时 I/O 失败仅清除租约，一秒后
|
||||
重试，整个有效尝试期间保留 active_operation_id，也保证 prefix 输入稳定。到期
|
||||
停止执行；恢复以新有界 recovery deadline 和 epoch 调度原输入，不改变原 deadline、
|
||||
资源身份或配额。
|
||||
|
||||
保留现有 segment 与聚合 SHA-256、ordinal/count、manifest closure、terminal
|
||||
检查。只有受执行权保护的短事务可以安装 VERIFIED 对象、seal、FINISHED 和成功凭据。
|
||||
待校验资源不能进入 admission 或 settlement。finish 保留已冻结前驱；前驱永久失败
|
||||
时明确失败，不无限等待。
|
||||
|
||||
永久 validation/storage-denial/fencing 失败仅保存安全 status/code，不保存原始
|
||||
存储异常或凭据。会话 recovery block 会使已受理的校验永久失败；产品没有从被阻断
|
||||
会话回到 READY 的路径。授权失败本身不能证明对象字节损坏。
|
||||
activation 的 id 与 epoch 仍匹配时，仅有效期已过会重试，因为续期保留该身份。writer 租约过期仍是永久 fence。finish 永久失败后保持 FAILED，离开FINISHING，并退役其 terminal 候选，之后的 finish 仍可被接受。已接受任务的原始 CSI binding 被释放或 CSI 授权失效时，永久失败并
|
||||
退出队列。副本凭据密钥或 SQL 错误仍可重试，不能误判为运行时身份失效。
|
||||
候选的输入或授权错误逐项隔离：未领取任务在原 epoch 下失败、延后一秒或到期，
|
||||
然后扫描继续处理后面的候选；SQL 状态更新失败则退出本轮扫描。期限内失去租约的
|
||||
尝试只能在同一 epoch 下重新排期，不能提交失败或隔离状态；重试时间不超过操作期限。
|
||||
完整性隔离和所有成功/失败写入都要求当前有效 epoch；旧 worker
|
||||
不能覆盖接管者结果。保留 read lease、retirement、未知 PUT、候选/隔离对象保护和
|
||||
GC 配额核算。
|
||||
永久 FAILED 操作不阻止已完成接收且被引用的 publication 回收。PENDING 操作（包括
|
||||
已超期操作）、未解决 PUT、候选或隔离对象以及其他保留保护仍会阻止回收；回收不会
|
||||
复活失败操作。
|
||||
|
||||
## 发布与可观测性
|
||||
|
||||
1. 应用增量 migration，升级全部服务端，保持异步接收关闭。
|
||||
2. 升级客户端，继续兼容旧服务端的直接凭据。
|
||||
3. 全部 writer 维护 journal head 列后开启 journal-head authorization，再开启异步
|
||||
接收；长寿命旧客户端仍走同步模式。
|
||||
4. 回退先关闭新的异步接收，已有任务继续处理。完成或妥善处理任务后再降级服务端
|
||||
二进制；处理期间保持 journal-head authorization 开启。
|
||||
|
||||
通过安全结构化日志记录排队等待、校验耗时、重试、失败和到期。吞吐预算和并发是
|
||||
运维限制，不承诺全部操作的端到端耗时都会缩短。排队时间使用数据库时间戳，校验
|
||||
耗时使用进程单调时钟；重试日志仅记录 scope、operation 和异常类名。
|
||||
|
||||
## 验证与验收
|
||||
|
||||
阻塞对象读取，证明异步 HTTP 响应先于校验；阻塞 PUT，证明持久化前不会 accepted。
|
||||
不重传输入而重启 verifier，覆盖 seal 与 inline/object terminal。验证双 worker
|
||||
竞争、租约接管、迟到成功/失败、前驱失败、prefix 重试稳定性，以及到期/恢复身份配额。
|
||||
|
||||
覆盖缺段/多段、摘要变化、非法 manifest、临时 I/O、owner 失效、retirement、FAILED
|
||||
稳定性和错误分类。验证新旧客户端/服务端组合、关闭接收后任务继续处理、历史成功
|
||||
重放后实际读取仍发现损坏，以及健康 head 下心跳/状态轮询不读取 journal 正文。
|
||||
运行定向 Java/CLI 测试、真实 MySQL 恢复/竞争测试、build、typecheck、bundle 和完整
|
||||
diff 自审。E2E 计划与结果存放在 `.qwen/e2e-tests/`。
|
||||
|
||||
## 涉及组件与决策
|
||||
|
||||
协调修改 Java publication catalog、producer controller、授权 store、配置和
|
||||
migration，以及 CLI publication HTTP adapter 和定向测试。没有未决产品问题。
|
||||
持久化任务、有界执行、严格成功边界、滚动兼容、历史凭据重放为选定默认行为。
|
||||
@@ -592,7 +592,7 @@ describe('reconcileNeverStartedChildRuns', () => {
|
||||
expect(broker.backgroundProcessStatus).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('refuses a shell admit when the production kind gate is restored', async () => {
|
||||
it('admits a shell when the production kind gate is restored', async () => {
|
||||
await open();
|
||||
await createManagedHarnessHandle(session!).ensureRunnable();
|
||||
enablement.childRun = false;
|
||||
@@ -604,7 +604,11 @@ describe('reconcileNeverStartedChildRuns', () => {
|
||||
executionCallId: 'shell-1',
|
||||
args: { command: 'printf hi', is_background: true },
|
||||
}),
|
||||
).rejects.toThrow('registered but not enabled');
|
||||
).resolves.toMatchObject({ kind: 'managed-tool-args' });
|
||||
expect(committed('shell-1')).toMatchObject({
|
||||
kind: 'shell',
|
||||
run: { state: 'admitted', execution: 'intent' },
|
||||
});
|
||||
});
|
||||
|
||||
it('never scans a child_agent record, the one kind production can commit today', async () => {
|
||||
|
||||
@@ -99,6 +99,154 @@ const request = {
|
||||
afterEach(() => vi.unstubAllGlobals());
|
||||
|
||||
describe('remote Shell result publication', () => {
|
||||
it('waits for asynchronous verification with the same operation and capability header', async () => {
|
||||
let posts = 0;
|
||||
let polls = 0;
|
||||
const operations = new Set<string>();
|
||||
const bytes = Buffer.from('original');
|
||||
vi.stubGlobal(
|
||||
'fetch',
|
||||
vi.fn(async (url: URL, init: RequestInit) => {
|
||||
const headers = new Headers(init.headers);
|
||||
expect(headers.get('X-Qwen-Tool-Publication-Async')).toBe('1');
|
||||
operations.add(headers.get('X-Qwen-Tool-Publication-Operation')!);
|
||||
if (!url.pathname.includes('/operations/')) {
|
||||
posts++;
|
||||
return new Response(JSON.stringify({ state: 'PENDING' }), {
|
||||
status: 202,
|
||||
});
|
||||
}
|
||||
polls++;
|
||||
return new Response(
|
||||
JSON.stringify(
|
||||
polls < 2
|
||||
? { state: 'PENDING' }
|
||||
: {
|
||||
state: 'SUCCEEDED',
|
||||
receipt: {
|
||||
captureId: 'capture-a',
|
||||
streamId: 'stdout',
|
||||
ordinal: 0,
|
||||
byteLength: bytes.length,
|
||||
digest: digest(bytes),
|
||||
},
|
||||
},
|
||||
),
|
||||
);
|
||||
}),
|
||||
);
|
||||
const publisher = new RemoteShellResultPublisher();
|
||||
publisher.install(installation, boot);
|
||||
const { sink } = await publisher.prepare(request);
|
||||
const store = Reflect.get(sink, 'store') as ToolResultSegmentStore;
|
||||
await expect(
|
||||
store.publish({
|
||||
captureId: 'capture-a',
|
||||
streamId: 'stdout',
|
||||
ordinal: 0,
|
||||
bytes,
|
||||
}),
|
||||
).resolves.toMatchObject({ status: 'ok' });
|
||||
expect(posts).toBe(1);
|
||||
expect(polls).toBe(2);
|
||||
expect(operations.size).toBe(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ status: 507, code: 'managed_tool_publication_quota_exhausted' },
|
||||
{ status: 403, code: 'managed_tool_publication_storage_denied' },
|
||||
{ status: 503, code: 'internal_error' },
|
||||
{ status: 400, code: 'managed_tool_result_digest_mismatch' },
|
||||
])(
|
||||
'treats asynchronous FAILED $status as terminal',
|
||||
async ({ status, code }) => {
|
||||
let posts = 0;
|
||||
let polls = 0;
|
||||
vi.stubGlobal(
|
||||
'fetch',
|
||||
vi.fn(async (url: URL) => {
|
||||
expect(url.pathname.endsWith('/recover')).toBe(false);
|
||||
if (!url.pathname.includes('/operations/')) {
|
||||
posts++;
|
||||
return new Response(JSON.stringify({ state: 'PENDING' }), {
|
||||
status: 202,
|
||||
});
|
||||
}
|
||||
polls++;
|
||||
return new Response(
|
||||
JSON.stringify({ state: 'FAILED', error: { status, code } }),
|
||||
);
|
||||
}),
|
||||
);
|
||||
const publisher = new RemoteShellResultPublisher();
|
||||
publisher.install(installation, boot);
|
||||
const { sink } = await publisher.prepare(request);
|
||||
const store = Reflect.get(sink, 'store') as ToolResultSegmentStore;
|
||||
const result = store.publish({
|
||||
captureId: 'capture-a',
|
||||
streamId: 'stdout',
|
||||
ordinal: 0,
|
||||
bytes: Buffer.from('original'),
|
||||
});
|
||||
if (code === 'managed_tool_result_digest_mismatch') {
|
||||
await expect(result).resolves.toMatchObject({
|
||||
status: 'refused',
|
||||
code,
|
||||
});
|
||||
} else {
|
||||
await expect(result).rejects.toThrow(
|
||||
status === 507 ? 'quota_exhausted' : `HTTP ${status}`,
|
||||
);
|
||||
}
|
||||
expect(posts).toBe(1);
|
||||
expect(polls).toBe(1);
|
||||
},
|
||||
);
|
||||
|
||||
it.each([
|
||||
undefined,
|
||||
{},
|
||||
{ code: 'internal_error' },
|
||||
{ status: 503 },
|
||||
{ status: 503, code: '' },
|
||||
{ status: 503.5, code: 'internal_error' },
|
||||
])('rejects malformed FAILED without retrying: %j', async (error) => {
|
||||
let posts = 0;
|
||||
let polls = 0;
|
||||
vi.stubGlobal(
|
||||
'fetch',
|
||||
vi.fn(async (url: URL) => {
|
||||
expect(url.pathname.endsWith('/recover')).toBe(false);
|
||||
if (!url.pathname.includes('/operations/')) {
|
||||
posts++;
|
||||
return new Response(JSON.stringify({ state: 'PENDING' }), {
|
||||
status: 202,
|
||||
});
|
||||
}
|
||||
polls++;
|
||||
return new Response(JSON.stringify({ state: 'FAILED', error }));
|
||||
}),
|
||||
);
|
||||
const publisher = new RemoteShellResultPublisher();
|
||||
publisher.install(installation, boot);
|
||||
const { sink } = await publisher.prepare(request);
|
||||
const store = Reflect.get(sink, 'store') as ToolResultSegmentStore;
|
||||
await expect(
|
||||
store.publish({
|
||||
captureId: 'capture-a',
|
||||
streamId: 'stdout',
|
||||
ordinal: 0,
|
||||
bytes: Buffer.from('original'),
|
||||
}),
|
||||
).rejects.toThrow(
|
||||
error === undefined
|
||||
? 'Publication record is invalid.'
|
||||
: 'Invalid publication failure response.',
|
||||
);
|
||||
expect(posts).toBe(1);
|
||||
expect(polls).toBe(1);
|
||||
});
|
||||
|
||||
it('keeps an installed grant blocked before any capture or tool starts', () => {
|
||||
const publisher = new RemoteShellResultPublisher();
|
||||
const resolver = vi.fn(async () => undefined);
|
||||
|
||||
@@ -172,6 +172,7 @@ class PublicationClient {
|
||||
'X-Qwen-Tenant-Id': String(this.key['tenantId']),
|
||||
'X-Qwen-Tool-Publication-Token': this.installed.publicationToken,
|
||||
'X-Qwen-Tool-Publication-Operation': operationId,
|
||||
'X-Qwen-Tool-Publication-Async': '1',
|
||||
'Cache-Control': 'no-store',
|
||||
...headers,
|
||||
},
|
||||
@@ -279,6 +280,23 @@ class PublicationClient {
|
||||
let status = initial;
|
||||
while (Date.now() < deadline) {
|
||||
if (status['state'] === 'SUCCEEDED') return record(status['receipt']);
|
||||
if (status['state'] === 'FAILED') {
|
||||
const error = record(status['error']);
|
||||
if (
|
||||
!Number.isInteger(error['status']) ||
|
||||
Number(error['status']) < 400 ||
|
||||
Number(error['status']) > 599 ||
|
||||
typeof error['code'] !== 'string' ||
|
||||
!error['code']
|
||||
)
|
||||
throw new Error('Invalid publication failure response.');
|
||||
if (error['code'] === 'managed_tool_publication_quota_exhausted')
|
||||
throw new Error('quota_exhausted');
|
||||
throw new PublicationRejection(
|
||||
Number(error['status']),
|
||||
error['code'],
|
||||
);
|
||||
}
|
||||
if (status['state'] === 'RETRYABLE') break;
|
||||
if (status['state'] === 'EXPIRED' && recoveries < 3) {
|
||||
recoveries++;
|
||||
|
||||
@@ -126,6 +126,32 @@ Design: [English](../../../docs/design/2026-09-29-managed-tool-result-public-pro
|
||||
revalidation window: [English](../../../docs/design/2026-10-02-managed-agent-query-amplification.md) |
|
||||
[简体中文](../../../docs/design/2026-10-02-managed-agent-query-amplification.zh-CN.md).
|
||||
|
||||
## Asynchronous tool publication verification
|
||||
|
||||
Apply migration V68 and upgrade all servers before enabling
|
||||
`qwen.managed-agent.tool-publication.async-verification-enabled` (default
|
||||
`false`). Enable `journal-head-authorization` first; startup rejects async
|
||||
admission without it. `verification-concurrency` defaults to `2` and controls
|
||||
the dedicated verifier pool, independently of retention and GC.
|
||||
|
||||
Updated clients send `X-Qwen-Tool-Publication-Async: 1` and poll the existing
|
||||
operation status after HTTP 202. They continue accepting direct receipts from
|
||||
old or disabled servers. Older clients remain synchronous. A mode is saved
|
||||
when each operation is created, so retries do not change it. `FAILED` carries
|
||||
only a safe `error: {status, code}` and must be handled as a terminal error,
|
||||
outside network retry handling. The adapter keeps awaiting verified success
|
||||
before completing publish, seal, prefix, or finish.
|
||||
|
||||
For rollback, disable new async admission and leave the verifier and head
|
||||
authorization running until accepted work drains. Do not downgrade the server
|
||||
while pending async operations still need verification. Historical async
|
||||
success replay returns the prior receipt; actual reads and final integrity
|
||||
checks still validate current storage bytes.
|
||||
|
||||
Design and client migration contract:
|
||||
[English](../../../docs/design/2026-10-07-managed-tool-publication-async-verification.md) |
|
||||
[简体中文](../../../docs/design/2026-10-07-managed-tool-publication-async-verification.zh-CN.md).
|
||||
|
||||
## Prerequisites
|
||||
|
||||
- Java 21
|
||||
|
||||
+35
-19
@@ -34,12 +34,14 @@ import org.springframework.web.bind.annotation.RestController;
|
||||
@RequestMapping("/internal/managed-tool-publications/v1/sessions/{sessionId}")
|
||||
public class ToolPublicationController {
|
||||
public static final String PUBLICATION_TOKEN_HEADER = "X-Qwen-Tool-Publication-Token";
|
||||
public static final String ASYNC_HEADER = "X-Qwen-Tool-Publication-Async";
|
||||
public static final String OPERATION_ID_HEADER = "X-Qwen-Tool-Publication-Operation";
|
||||
private static final ObjectMapper JSON = new ObjectMapper();
|
||||
private final ToolPublicationStore grants;
|
||||
private final ToolPublicationDataStore data;
|
||||
private final ToolPublicationAdmissionStore admissions;
|
||||
private final Semaphore entries;
|
||||
private final boolean asyncVerificationEnabled;
|
||||
|
||||
public ToolPublicationController(ToolPublicationStore grants,
|
||||
ToolPublicationDataStore data, ToolPublicationAdmissionStore admissions,
|
||||
@@ -47,9 +49,20 @@ public class ToolPublicationController {
|
||||
this.grants = Objects.requireNonNull(grants);
|
||||
this.data = Objects.requireNonNull(data);
|
||||
this.admissions = Objects.requireNonNull(admissions);
|
||||
this.asyncVerificationEnabled = properties.getToolPublication().isAsyncVerificationEnabled();
|
||||
this.entries = new Semaphore(properties.getToolPublication().getEntryConcurrency());
|
||||
}
|
||||
|
||||
private boolean asynchronous(String header) {
|
||||
return asyncVerificationEnabled && "1".equals(header);
|
||||
}
|
||||
|
||||
private ResponseEntity<JsonNode> producerResponse(JsonNode result) {
|
||||
return ResponseEntity.status("PENDING".equals(result.path("state").asText())
|
||||
? HttpStatus.ACCEPTED : HttpStatus.OK)
|
||||
.header(HttpHeaders.CACHE_CONTROL, "no-store").body(result);
|
||||
}
|
||||
|
||||
@PostMapping("/grants")
|
||||
public JsonNode grant(TenantContext tenant, @PathVariable String sessionId,
|
||||
@RequestHeader(ManagedSessionStoreModels.WRITER_TOKEN_HEADER) String writerToken,
|
||||
@@ -63,73 +76,76 @@ public class ToolPublicationController {
|
||||
}
|
||||
|
||||
@PostMapping("/publications/{publicationId}/segments/{streamId}/{ordinal}")
|
||||
public JsonNode segment(TenantContext tenant, @PathVariable String sessionId,
|
||||
public ResponseEntity<JsonNode> segment(TenantContext tenant, @PathVariable String sessionId,
|
||||
@PathVariable String publicationId, @PathVariable String streamId,
|
||||
@PathVariable int ordinal, @RequestParam String workspaceId,
|
||||
@RequestHeader(PUBLICATION_TOKEN_HEADER) String token,
|
||||
@RequestHeader(OPERATION_ID_HEADER) String operationId,
|
||||
@RequestHeader(value = "X-Qwen-Tool-Segment-Digest", required = false) String digest,
|
||||
HttpServletRequest request) throws IOException {
|
||||
return limited(() -> {
|
||||
return producerResponse(limited(() -> {
|
||||
JsonNode key = scope(tenant, workspaceId, sessionId);
|
||||
byte[] bytes = read(request, 16 * 1024 * 1024);
|
||||
return data.publishSegment(key, publicationId, token,
|
||||
operationId, streamId, ordinal, bytes, digest);
|
||||
});
|
||||
operationId, streamId, ordinal, bytes, digest, asynchronous(request.getHeader(ASYNC_HEADER)));
|
||||
}));
|
||||
}
|
||||
|
||||
@PostMapping("/publications/{publicationId}/resources/{kind}/{slot}")
|
||||
public JsonNode resource(TenantContext tenant, @PathVariable String sessionId,
|
||||
public ResponseEntity<JsonNode> resource(TenantContext tenant, @PathVariable String sessionId,
|
||||
@PathVariable String publicationId, @PathVariable String kind,
|
||||
@PathVariable String slot, @RequestParam String workspaceId,
|
||||
@RequestHeader(PUBLICATION_TOKEN_HEADER) String token,
|
||||
@RequestHeader(OPERATION_ID_HEADER) String operationId,
|
||||
HttpServletRequest request) throws IOException {
|
||||
return limited(() -> {
|
||||
return producerResponse(limited(() -> {
|
||||
JsonNode key = scope(tenant, workspaceId, sessionId);
|
||||
byte[] bytes = read(request, 16 * 1024 * 1024);
|
||||
return data.publishResource(key, publicationId, token,
|
||||
operationId, slot, kind, bytes);
|
||||
});
|
||||
operationId, slot, kind, bytes, asynchronous(request.getHeader(ASYNC_HEADER)));
|
||||
}));
|
||||
}
|
||||
|
||||
@PostMapping("/publications/{publicationId}/streams/{streamId}/seal")
|
||||
public JsonNode seal(TenantContext tenant, @PathVariable String sessionId,
|
||||
public ResponseEntity<JsonNode> seal(TenantContext tenant, @PathVariable String sessionId,
|
||||
@PathVariable String publicationId, @PathVariable String streamId,
|
||||
@RequestParam String workspaceId,
|
||||
@RequestHeader(PUBLICATION_TOKEN_HEADER) String token,
|
||||
@RequestHeader(OPERATION_ID_HEADER) String operationId,
|
||||
HttpServletRequest request) throws IOException {
|
||||
return limited(() -> {
|
||||
return producerResponse(limited(() -> {
|
||||
JsonNode body = ToolPublicationContract.readJson(read(request, 64 * 1024));
|
||||
JsonNode key = scope(tenant, workspaceId, sessionId);
|
||||
return data.seal(key, publicationId, token, operationId,
|
||||
streamId, body.path("segmentCount").asInt(-1),
|
||||
body.path("byteLength").asLong(-1), body.path("digest").asText(null));
|
||||
});
|
||||
body.path("byteLength").asLong(-1), body.path("digest").asText(null),
|
||||
asynchronous(request.getHeader(ASYNC_HEADER)));
|
||||
}));
|
||||
}
|
||||
|
||||
@PostMapping("/publications/{publicationId}/streams/{streamId}/prefix")
|
||||
public JsonNode prefix(TenantContext tenant, @PathVariable String sessionId,
|
||||
public ResponseEntity<JsonNode> prefix(TenantContext tenant, @PathVariable String sessionId,
|
||||
@PathVariable String publicationId, @PathVariable String streamId,
|
||||
@RequestParam String workspaceId,
|
||||
@RequestHeader(PUBLICATION_TOKEN_HEADER) String token,
|
||||
@RequestHeader(OPERATION_ID_HEADER) String operationId) {
|
||||
@RequestHeader(OPERATION_ID_HEADER) String operationId,
|
||||
@RequestHeader(value = ASYNC_HEADER, required = false) String async) {
|
||||
JsonNode key = scope(tenant, workspaceId, sessionId);
|
||||
return limited(() -> data.prefix(key, publicationId, token, operationId, streamId));
|
||||
return producerResponse(limited(() -> data.prefix(key, publicationId, token,
|
||||
operationId, streamId, asynchronous(async))));
|
||||
}
|
||||
|
||||
@PostMapping("/publications/{publicationId}/finish")
|
||||
public JsonNode finish(TenantContext tenant, @PathVariable String sessionId,
|
||||
public ResponseEntity<JsonNode> finish(TenantContext tenant, @PathVariable String sessionId,
|
||||
@PathVariable String publicationId, @RequestParam String workspaceId,
|
||||
@RequestHeader(PUBLICATION_TOKEN_HEADER) String token,
|
||||
@RequestHeader(OPERATION_ID_HEADER) String operationId,
|
||||
HttpServletRequest request) throws IOException {
|
||||
return limited(() -> {
|
||||
return producerResponse(limited(() -> {
|
||||
byte[] body = read(request, 2 * 1024 * 1024);
|
||||
JsonNode key = scope(tenant, workspaceId, sessionId);
|
||||
return data.finish(key, publicationId, token, operationId, body);
|
||||
});
|
||||
return data.finish(key, publicationId, token, operationId, body, asynchronous(request.getHeader(ASYNC_HEADER)));
|
||||
}));
|
||||
}
|
||||
|
||||
@GetMapping("/publications/{publicationId}/operations/{operationId}")
|
||||
|
||||
+6
@@ -634,6 +634,8 @@ public class ManagedAgentProperties implements EnvironmentAware {
|
||||
@DurationUnit(ChronoUnit.SECONDS)
|
||||
private Duration maxVerificationTimeout;
|
||||
private boolean gcEnabled;
|
||||
private boolean asyncVerificationEnabled;
|
||||
private int verificationConcurrency = 2;
|
||||
@DurationUnit(ChronoUnit.SECONDS)
|
||||
private Duration deletionGrace = Duration.ofHours(24);
|
||||
// Off by default: the head's activation columns are only trustworthy
|
||||
@@ -642,6 +644,10 @@ public class ManagedAgentProperties implements EnvironmentAware {
|
||||
private boolean journalHeadAuthorization;
|
||||
|
||||
public boolean isJournalHeadAuthorization() { return journalHeadAuthorization; }
|
||||
public boolean isAsyncVerificationEnabled() { return asyncVerificationEnabled; }
|
||||
public void setAsyncVerificationEnabled(boolean value) { asyncVerificationEnabled = value; }
|
||||
public int getVerificationConcurrency() { return verificationConcurrency; }
|
||||
public void setVerificationConcurrency(int value) { verificationConcurrency = value; }
|
||||
public void setJournalHeadAuthorization(boolean value) { journalHeadAuthorization = value; }
|
||||
public boolean isGcEnabled() { return gcEnabled; }
|
||||
public void setGcEnabled(boolean value) { gcEnabled = value; }
|
||||
|
||||
+16
@@ -11,6 +11,7 @@ import com.alibaba.qwen.code.managedagent.store.ToolPublicationStore;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationRetentionStore;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationRetentionObserver;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationCollector;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationVerifier;
|
||||
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRepository;
|
||||
import com.aliyun.oss.ClientBuilderConfiguration;
|
||||
@@ -89,6 +90,9 @@ public class ToolPublicationConfiguration {
|
||||
ManagedSessionStore sessions, ToolPublicationObjectStore objects,
|
||||
ManagedAgentProperties properties) {
|
||||
var settings = properties.getToolPublication();
|
||||
if (settings.isAsyncVerificationEnabled() && !settings.isJournalHeadAuthorization()) {
|
||||
throw new IllegalStateException("Async verification requires journal-head authorization");
|
||||
}
|
||||
if (settings.getOperationTimeout() == null || settings.getClaimTimeout() == null
|
||||
|| settings.getMaxVerificationTimeout() == null) {
|
||||
throw new IllegalStateException("Tool publication operation and verification deadlines are required");
|
||||
@@ -130,6 +134,18 @@ public class ToolPublicationConfiguration {
|
||||
return builder.poolSize(1).threadNamePrefix("managed-tool-output-").build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ThreadPoolTaskScheduler managedToolVerificationScheduler(ThreadPoolTaskSchedulerBuilder builder) {
|
||||
return builder.poolSize(1).threadNamePrefix("managed-tool-verification-scan-").build();
|
||||
}
|
||||
|
||||
// Rollback disables new async admission; accepted operations must still drain.
|
||||
@Bean(destroyMethod = "close")
|
||||
public ToolPublicationVerifier toolPublicationVerifier(ToolPublicationDataStore data,
|
||||
ManagedAgentProperties properties) {
|
||||
return new ToolPublicationVerifier(data, properties.getToolPublication().getVerificationConcurrency());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ToolPublicationCollector toolPublicationCollector(JdbcTemplate jdbc, PlatformTransactionManager manager,
|
||||
ToolPublicationRetentionStore retention, ToolPublicationObjectStore objects, ManagedAgentProperties properties) {
|
||||
|
||||
+13
-3
@@ -55,7 +55,7 @@ public final class ToolPublicationContract {
|
||||
case "binding" -> binding(value);
|
||||
case "request" -> request(value);
|
||||
case "grant" -> grant(value);
|
||||
default -> throw new IllegalArgumentException("Unknown publication record");
|
||||
default -> throw new ContractViolationException("Unknown publication record");
|
||||
}
|
||||
return value.deepCopy();
|
||||
}
|
||||
@@ -240,7 +240,7 @@ public final class ToolPublicationContract {
|
||||
return JSON.reader().with(com.fasterxml.jackson.databind.DeserializationFeature
|
||||
.FAIL_ON_TRAILING_TOKENS).readTree(text);
|
||||
} catch (IOException error) {
|
||||
throw new IllegalArgumentException("Invalid publication JSON", error);
|
||||
throw new ContractViolationException("Invalid publication JSON", error);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -411,9 +411,19 @@ public final class ToolPublicationContract {
|
||||
return node == null ? "" : node.path(key).asText("");
|
||||
}
|
||||
|
||||
static final class ContractViolationException extends IllegalArgumentException {
|
||||
private ContractViolationException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
private ContractViolationException(String message, Throwable cause) {
|
||||
super(message, cause);
|
||||
}
|
||||
}
|
||||
|
||||
static void require(boolean condition, String reason) {
|
||||
if (!condition) {
|
||||
throw new IllegalArgumentException(reason);
|
||||
throw new ContractViolationException(reason);
|
||||
}
|
||||
}
|
||||
}
|
||||
+616
-46
@@ -24,6 +24,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import javax.sql.DataSource;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
@@ -32,9 +33,16 @@ import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
import org.springframework.transaction.support.TransactionSynchronizationManager;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import com.aliyun.oss.OSSException;
|
||||
|
||||
/** O2b's publication catalog. Object I/O always runs outside SQL transactions. */
|
||||
public final class ToolPublicationDataStore {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ToolPublicationDataStore.class);
|
||||
// Nested read helpers also fence quarantine writes against the executing verifier.
|
||||
private final ThreadLocal<VerificationTask> verificationTask = new ThreadLocal<>();
|
||||
private volatile Runnable verificationWakeup = () -> {};
|
||||
private static final ObjectMapper JSON = new ObjectMapper();
|
||||
private static final ObjectMapper ACK_JSON = new ObjectMapper()
|
||||
.enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
|
||||
@@ -256,16 +264,451 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
}
|
||||
|
||||
void setVerificationWakeup(Runnable wakeup) {
|
||||
verificationWakeup = Objects.requireNonNull(wakeup);
|
||||
}
|
||||
|
||||
private JsonNode submitVerification(JsonNode key, String publicationId, String token,
|
||||
String operationId, String slot, String digest, ObjectNode inputs, String kind,
|
||||
byte[] bytes, String payloadDigest, boolean asynchronous) {
|
||||
if (verificationTask.get() != null) {
|
||||
return null;
|
||||
}
|
||||
require(operationId != null && operationId.matches("[a-z0-9_-]{1,128}"),
|
||||
"Invalid publication operation ID");
|
||||
String scope = scope(key);
|
||||
List<Operation> observed = operation(scope, publicationId, operationId, false);
|
||||
if (!asynchronous && (observed.isEmpty() || !observed.getFirst().asynchronous())) {
|
||||
return null;
|
||||
}
|
||||
if (!observed.isEmpty() && observed.getFirst().asynchronous()
|
||||
&& "FAILED".equals(observed.getFirst().state())) {
|
||||
require(slot.equals(observed.getFirst().slot()) && digest.equals(observed.getFirst().digest()),
|
||||
"Publication operation conflicts");
|
||||
return operationStatus(key, publicationId, token, operationId);
|
||||
}
|
||||
Submission submission = transactions.execute(status -> {
|
||||
var authorization = authorizeClaim(key, scope, publicationId, token);
|
||||
List<Operation> prior = operation(scope, publicationId, operationId, true);
|
||||
if (!prior.isEmpty()) {
|
||||
Operation row = prior.getFirst();
|
||||
require(slot.equals(row.slot()) && digest.equals(row.digest()), "Publication operation conflicts");
|
||||
if (!row.asynchronous()) {
|
||||
return null;
|
||||
}
|
||||
if (!"PENDING".equals(row.state()) || row.ready() || !row.deadline().after(now())) {
|
||||
return new Submission(0, null, operationStatus(key, publicationId, token, operationId));
|
||||
}
|
||||
} else if (!asynchronous) {
|
||||
return null;
|
||||
}
|
||||
long epoch;
|
||||
String objectKey = null;
|
||||
JsonNode receipt;
|
||||
if ("terminal".equals(slot)) {
|
||||
FinishClaim claim = beginFinish(key, scope, publicationId, token, operationId,
|
||||
bytes.length, payloadDigest, true);
|
||||
epoch = claim.epoch();
|
||||
objectKey = claim.objectKey();
|
||||
receipt = claim.receipt();
|
||||
} else if (slot.startsWith("seal:") || slot.startsWith("prefix:")) {
|
||||
ScanClaim claim = claimScan(key, scope, publicationId, token, operationId, slot, digest, true);
|
||||
epoch = claim.epoch();
|
||||
receipt = claim.receipt();
|
||||
} else {
|
||||
Candidate claim = claim(key, scope, publicationId, token, operationId, slot, kind,
|
||||
bytes.length, payloadDigest, true);
|
||||
epoch = claim.epoch();
|
||||
objectKey = claim.objectKey();
|
||||
receipt = claim.receipt();
|
||||
}
|
||||
if (receipt != null) {
|
||||
return new Submission(0, null, JSON.createObjectNode().put("state", "SUCCEEDED")
|
||||
.set("receipt", receipt));
|
||||
}
|
||||
inputs.put("bindingDigest", ToolPublicationContract.bindingDigest(authorization.binding()))
|
||||
.put("tokenHash", ToolPublicationContract.tokenHash(token));
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET execution_mode = 'ASYNC',"
|
||||
+ " verification_request_json = ? WHERE scope_key = ? AND publication_id = ?"
|
||||
+ " AND operation_id = ?", inputs.toString(), scope, publicationId, operationId);
|
||||
return new Submission(epoch, objectKey, null);
|
||||
});
|
||||
if (submission == null) {
|
||||
return null;
|
||||
}
|
||||
if (submission.response() != null) {
|
||||
return submission.response();
|
||||
}
|
||||
try {
|
||||
if (submission.objectKey() != null) {
|
||||
retention.put(key, scope, publicationId, submission.objectKey(), bytes, objects);
|
||||
}
|
||||
transactions.executeWithoutResult(status -> {
|
||||
authorize(key, scope, publicationId, token);
|
||||
requireOperationClaim(operation(scope, publicationId, operationId, true), submission.epoch(), now());
|
||||
if (bytes != null && submission.objectKey() == null) {
|
||||
jdbc.update("UPDATE qwen_tool_publication_object SET inline_bytes = ? WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND slot_key = ? AND operation_id = ? AND state = 'CANDIDATE'",
|
||||
bytes, scope, publicationId, slot, operationId);
|
||||
}
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET verification_ready = TRUE,"
|
||||
+ " verification_next_at = ?, claim_owner = NULL, claim_until = NULL"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
now(), scope, publicationId, operationId);
|
||||
});
|
||||
verificationWakeup.run();
|
||||
return JSON.createObjectNode().put("state", "PENDING");
|
||||
} catch (RuntimeException error) {
|
||||
abandonScan(scope, publicationId, operationId, submission.epoch());
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
boolean verifyNextOperation() {
|
||||
var candidates = jdbc.queryForList("SELECT scope_key, publication_id, operation_id, claim_epoch, verification_request_json"
|
||||
+ " FROM qwen_tool_publication_operation WHERE execution_mode = 'ASYNC' AND state = 'PENDING'"
|
||||
+ " AND verification_ready = TRUE AND verification_next_at <= CURRENT_TIMESTAMP(6)"
|
||||
+ " ORDER BY verification_next_at, scope_key, publication_id, operation_id LIMIT 32");
|
||||
boolean processed = false;
|
||||
for (var candidate : candidates) {
|
||||
String scope = (String) candidate.get("scope_key");
|
||||
String publicationId = (String) candidate.get("publication_id");
|
||||
String operationId = (String) candidate.get("operation_id");
|
||||
VerificationTask task;
|
||||
try {
|
||||
Object encoded = candidate.get("verification_request_json");
|
||||
require(encoded instanceof String, "Missing verification inputs");
|
||||
JsonNode inputs = ToolPublicationContract.readJson(((String) encoded).getBytes(StandardCharsets.UTF_8));
|
||||
task = transactions.execute(status -> {
|
||||
var authorization = grants.producerVerificationBindingLocked(scope, publicationId,
|
||||
text(inputs, "bindingDigest"), text(inputs, "tokenHash"));
|
||||
List<Operation> rows = operation(scope, publicationId, operationId, true);
|
||||
Timestamp current = now();
|
||||
if (rows.size() != 1 || !rows.getFirst().asynchronous() || !rows.getFirst().ready()
|
||||
|| !"PENDING".equals(rows.getFirst().state())
|
||||
|| rows.getFirst().claimUntil() != null && rows.getFirst().claimUntil().after(current)) {
|
||||
return null;
|
||||
}
|
||||
Operation row = rows.getFirst();
|
||||
if (!row.deadline().after(current)) {
|
||||
expireVerification(scope, publicationId, operationId);
|
||||
return null;
|
||||
}
|
||||
Timestamp due = jdbc.queryForObject("SELECT verification_next_at FROM qwen_tool_publication_operation"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
Timestamp.class, scope, publicationId, operationId);
|
||||
if (due == null || due.after(current)) {
|
||||
return null;
|
||||
}
|
||||
String active = jdbc.queryForObject("SELECT active_operation_id FROM qwen_tool_publication"
|
||||
+ " WHERE scope_key = ? AND publication_id = ?", String.class, scope, publicationId);
|
||||
requireContract("terminal".equals(row.slot()) || operationId.equals(active), HttpStatus.CONFLICT,
|
||||
"managed_tool_publication_claim_lost", "Publication operation lost its slot");
|
||||
long epoch = row.epoch() + 1;
|
||||
Timestamp until = new Timestamp(Math.min(row.deadline().getTime(),
|
||||
current.getTime() + claimTimeout.toMillis()));
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET claim_owner = ?, claim_epoch = ?,"
|
||||
+ " claim_until = ?, verification_next_at = ? WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND operation_id = ?",
|
||||
UUID.randomUUID().toString(), epoch, until, until, scope, publicationId, operationId);
|
||||
Timestamp created = jdbc.queryForObject("SELECT created_at FROM qwen_tool_publication_operation"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
Timestamp.class, scope, publicationId, operationId);
|
||||
return new VerificationTask(scope, publicationId, operationId, row.slot(), epoch,
|
||||
authorization.binding().path("sessionKey"), inputs, Math.max(0, current.getTime() - created.getTime()));
|
||||
});
|
||||
} catch (RuntimeException error) {
|
||||
try {
|
||||
processed |= deferOrFailUnclaimedVerification(scope, publicationId, operationId,
|
||||
((Number) candidate.get("claim_epoch")).longValue(), error);
|
||||
} catch (RuntimeException handling) {
|
||||
LOG.warn("Tool publication verification handling failed scope={} operation={}",
|
||||
scope, operationId, handling);
|
||||
skipPoisonCandidate(scope, publicationId, operationId);
|
||||
processed = true;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if (task == null) {
|
||||
continue;
|
||||
}
|
||||
long started = System.nanoTime();
|
||||
verificationTask.set(task);
|
||||
try {
|
||||
verifyOperation(task);
|
||||
LOG.info("Tool publication verified operation={} queueMillis={} verificationMillis={}",
|
||||
operationId, task.queueMillis(),
|
||||
TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - started));
|
||||
} catch (RuntimeException error) {
|
||||
try {
|
||||
deferOrFailVerification(task, error);
|
||||
} catch (RuntimeException handling) {
|
||||
LOG.warn("Tool publication verification handling failed scope={} operation={}",
|
||||
task.scope(), task.operationId(), handling);
|
||||
skipPoisonCandidate(task.scope(), task.publicationId(), task.operationId());
|
||||
}
|
||||
} finally {
|
||||
verificationTask.remove();
|
||||
}
|
||||
return true;
|
||||
}
|
||||
return processed;
|
||||
}
|
||||
|
||||
private void verifyOperation(VerificationTask task) {
|
||||
String slot = task.slot();
|
||||
if (slot.startsWith("seal:")) {
|
||||
seal(task.key(), task.publicationId(), null, task.operationId(), slot.substring(5),
|
||||
task.inputs().path("segmentCount").asInt(-1), task.inputs().path("byteLength").asLong(-1),
|
||||
text(task.inputs(), "digest"));
|
||||
} else if (slot.startsWith("prefix:")) {
|
||||
prefix(task.key(), task.publicationId(), null, task.operationId(), slot.substring(7));
|
||||
} else {
|
||||
Stored stored = stored(task.scope(), task.publicationId(), slot).getFirst();
|
||||
require("CANDIDATE".equals(stored.state()) && task.operationId().equals(stored.operationId()),
|
||||
"Verification candidate changed");
|
||||
byte[] inline = stored.objectKey() == null ? jdbc.queryForObject("SELECT inline_bytes FROM qwen_tool_publication_object"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND slot_key = ?", byte[].class,
|
||||
task.scope(), task.publicationId(), slot) : null;
|
||||
if ("terminal".equals(slot)) {
|
||||
byte[] bytes = inline;
|
||||
if (stored.objectKey() != null) {
|
||||
try (InputStream input = retention.open(task.scope(), task.publicationId(), stored.objectKey(),
|
||||
objects, verificationHeartbeat(task))) {
|
||||
bytes = input.readNBytes(MAX_TERMINAL + 1);
|
||||
} catch (IOException error) {
|
||||
throw new IllegalStateException("Terminal read failed", error);
|
||||
}
|
||||
}
|
||||
if (bytes == null || bytes.length != stored.length()
|
||||
|| !ToolPublicationContract.sha256(bytes).equals(stored.digest())) {
|
||||
quarantineCandidate(task.scope(), task.publicationId(), slot, task.operationId());
|
||||
throw new CorruptObjectException("Terminal digest changed");
|
||||
}
|
||||
finish(task.key(), task.publicationId(), null, task.operationId(), bytes);
|
||||
} else {
|
||||
if (stored.objectKey() != null) {
|
||||
try {
|
||||
verify(task.scope(), task.publicationId(), stored.objectKey(), stored.length(), stored.digest());
|
||||
} catch (CorruptObjectException error) {
|
||||
quarantineCandidate(task.scope(), task.publicationId(), slot, task.operationId());
|
||||
throw error;
|
||||
}
|
||||
} else {
|
||||
if (inline == null || inline.length != stored.length()
|
||||
|| !ToolPublicationContract.sha256(inline).equals(stored.digest())) {
|
||||
quarantineCandidate(task.scope(), task.publicationId(), slot, task.operationId());
|
||||
throw new CorruptObjectException("Inline candidate changed");
|
||||
}
|
||||
}
|
||||
String kind = jdbc.queryForObject("SELECT resource_kind FROM qwen_tool_publication_object"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND slot_key = ?", String.class,
|
||||
task.scope(), task.publicationId(), slot);
|
||||
transactions.executeWithoutResult(status -> install(task.key(), task.scope(), task.publicationId(),
|
||||
null, task.operationId(), new Candidate(stored.objectKey(), stored.resourceId(), task.epoch(), null),
|
||||
kind, Math.toIntExact(stored.length()), inline, stored.digest()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private Runnable verificationHeartbeat(VerificationTask task) {
|
||||
return heartbeat(task.key(), task.scope(), task.publicationId(), null, task.operationId(), task.epoch());
|
||||
}
|
||||
|
||||
private ToolPublicationStore.ProducerBinding verificationAuthorization(String scope, String publicationId) {
|
||||
VerificationTask task = verificationTask.get();
|
||||
require(task != null && scope.equals(task.scope()) && publicationId.equals(task.publicationId()),
|
||||
"Verification authorization is unavailable");
|
||||
return grants.producerVerificationBindingLocked(scope, publicationId,
|
||||
text(task.inputs(), "bindingDigest"), text(task.inputs(), "tokenHash"));
|
||||
}
|
||||
|
||||
private JsonNode verificationBinding(String scope, String publicationId) {
|
||||
return verificationAuthorization(scope, publicationId).binding();
|
||||
}
|
||||
|
||||
private void requireVerificationClaim(String scope, String publicationId) {
|
||||
VerificationTask task = verificationTask.get();
|
||||
if (task != null) {
|
||||
require(scope.equals(task.scope()) && publicationId.equals(task.publicationId()), "Verification scope conflicts");
|
||||
requireOperationClaim(operation(scope, publicationId, task.operationId(), true), task.epoch(), now());
|
||||
}
|
||||
}
|
||||
|
||||
private static ApiException permanentVerificationFailure(RuntimeException error) {
|
||||
if (error instanceof ApiException api) {
|
||||
if (List.of("managed_tool_publication_busy", "managed_tool_publication_claim_lost",
|
||||
"managed_tool_publication_claim_expired", "managed_tool_publication_operation_expired")
|
||||
.contains(api.getCode())) {
|
||||
return null;
|
||||
}
|
||||
return api.getStatus().is4xxClientError() || api.getStatus() == HttpStatus.INSUFFICIENT_STORAGE ? api : null;
|
||||
}
|
||||
if (error instanceof CorruptObjectException
|
||||
|| error instanceof ToolPublicationContract.ContractViolationException) {
|
||||
return new ApiException(HttpStatus.BAD_REQUEST, "invalid_request", "Publication verification failed");
|
||||
}
|
||||
if (error instanceof OSSException storage && "AccessDenied".equals(storage.getErrorCode())) {
|
||||
return new ApiException(HttpStatus.FORBIDDEN, "managed_tool_publication_storage_denied",
|
||||
"Publication storage rejected verification");
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private boolean deferOrFailUnclaimedVerification(String scope, String publicationId, String operationId,
|
||||
long epoch, RuntimeException error) {
|
||||
String[] releasedObject = new String[1];
|
||||
boolean handled = Boolean.TRUE.equals(transactions.execute(status -> {
|
||||
retention.lockRetainedPublication(scope, publicationId);
|
||||
var rows = operation(scope, publicationId, operationId, true);
|
||||
Timestamp current = now();
|
||||
if (rows.size() != 1 || !"PENDING".equals(rows.getFirst().state()) || !rows.getFirst().asynchronous()
|
||||
|| rows.getFirst().epoch() != epoch || !rows.getFirst().ready() || rows.getFirst().claimUntil() != null
|
||||
&& rows.getFirst().claimUntil().after(current)) {
|
||||
return false;
|
||||
}
|
||||
if (!rows.getFirst().deadline().after(current)) {
|
||||
expireVerification(scope, publicationId, operationId);
|
||||
return false;
|
||||
}
|
||||
Timestamp due = jdbc.queryForObject("SELECT verification_next_at FROM qwen_tool_publication_operation"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
Timestamp.class, scope, publicationId, operationId);
|
||||
if (due == null || due.after(current)) {
|
||||
return false;
|
||||
}
|
||||
ApiException failure = permanentVerificationFailure(error);
|
||||
if (failure != null) {
|
||||
releasedObject[0] = recordVerificationFailure(scope, publicationId, operationId, failure);
|
||||
} else {
|
||||
deferVerification(scope, publicationId, operationId, current, rows.getFirst().deadline(), error);
|
||||
}
|
||||
return true;
|
||||
}));
|
||||
deleteReleasedObject(releasedObject[0]);
|
||||
return handled;
|
||||
}
|
||||
|
||||
private void deferOrFailVerification(VerificationTask task, RuntimeException error) {
|
||||
String objectKey = transactions.execute(status -> {
|
||||
retention.lockRetainedPublication(task.scope(), task.publicationId());
|
||||
var rows = operation(task.scope(), task.publicationId(), task.operationId(), true);
|
||||
Timestamp current = now();
|
||||
if (rows.size() != 1 || !"PENDING".equals(rows.getFirst().state())
|
||||
|| rows.getFirst().epoch() != task.epoch()) {
|
||||
return null;
|
||||
}
|
||||
if (!rows.getFirst().deadline().after(current)) {
|
||||
expireVerification(task.scope(), task.publicationId(), task.operationId());
|
||||
return null;
|
||||
}
|
||||
boolean liveClaim = rows.getFirst().claimUntil() != null && rows.getFirst().claimUntil().after(current);
|
||||
ApiException failure = permanentVerificationFailure(error);
|
||||
if (liveClaim && failure != null) {
|
||||
return recordVerificationFailure(task.scope(), task.publicationId(), task.operationId(), failure);
|
||||
}
|
||||
deferVerification(task.scope(), task.publicationId(), task.operationId(), current,
|
||||
rows.getFirst().deadline(), error);
|
||||
return null;
|
||||
});
|
||||
deleteReleasedObject(objectKey);
|
||||
}
|
||||
|
||||
private void deferVerification(String scope, String publicationId, String operationId,
|
||||
Timestamp current, Timestamp deadline, RuntimeException error) {
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET claim_owner = NULL, claim_until = NULL,"
|
||||
+ " verification_next_at = ? WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
new Timestamp(Math.min(current.getTime() + 1000, deadline.getTime())), scope, publicationId, operationId);
|
||||
LOG.info("Tool publication verification retry scope={} operation={} cause={}",
|
||||
scope, operationId, error.getClass().getSimpleName());
|
||||
}
|
||||
|
||||
private String recordVerificationFailure(String scope, String publicationId, String operationId, ApiException failure) {
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET state = 'FAILED', failure_status = ?, failure_code = ?,"
|
||||
+ " claim_owner = NULL, claim_until = NULL, verification_next_at = NULL"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
failure.getStatus().value(), failure.getCode(), scope, publicationId, operationId);
|
||||
jdbc.update("UPDATE qwen_tool_publication SET active_operation_id = NULL WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND active_operation_id = ?", scope, publicationId, operationId);
|
||||
String objectKey = releaseFailedFinish(scope, publicationId, operationId);
|
||||
LOG.info("Tool publication verification failed operation={} code={}", operationId, failure.getCode());
|
||||
return objectKey;
|
||||
}
|
||||
|
||||
private void skipPoisonCandidate(String scope, String publicationId, String operationId) {
|
||||
try {
|
||||
transactions.executeWithoutResult(status -> jdbc.update("UPDATE qwen_tool_publication_operation"
|
||||
+ " SET verification_next_at = ? WHERE scope_key = ? AND publication_id = ?"
|
||||
+ " AND operation_id = ? AND state = 'PENDING'",
|
||||
new Timestamp(now().getTime() + 1000), scope, publicationId, operationId));
|
||||
} catch (RuntimeException error) {
|
||||
LOG.warn("Tool publication verification skip failed operation={}", operationId, error);
|
||||
}
|
||||
}
|
||||
|
||||
private void deleteReleasedObject(String objectKey) {
|
||||
if (objectKey != null) {
|
||||
objects.deleteIfPresent(objectKey);
|
||||
}
|
||||
}
|
||||
|
||||
private String releaseFailedFinish(String scope, String publicationId, String operationId) {
|
||||
Integer finishing = jdbc.queryForObject("SELECT COUNT(*) FROM qwen_tool_publication"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND producer_phase = 'FINISHING'"
|
||||
+ " AND finish_operation_id = ?", Integer.class, scope, publicationId, operationId);
|
||||
if (finishing == null || finishing == 0) {
|
||||
return null;
|
||||
}
|
||||
var terminal = jdbc.queryForList("SELECT byte_length, object_key FROM qwen_tool_publication_object"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND slot_key = 'terminal' AND operation_id = ?",
|
||||
scope, publicationId, operationId);
|
||||
long rollback = terminal.isEmpty() ? 0 : ((Number) terminal.get(0).get("byte_length")).longValue();
|
||||
String objectKey = terminal.isEmpty() ? null : (String) terminal.get(0).get("object_key");
|
||||
if (!terminal.isEmpty()) {
|
||||
// A non-VERIFIED row blocks collection forever. Drop it with the
|
||||
// charge so a later successful finish can be reclaimed.
|
||||
jdbc.update("DELETE FROM qwen_tool_publication_object WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND slot_key = 'terminal' AND operation_id = ?",
|
||||
scope, publicationId, operationId);
|
||||
}
|
||||
jdbc.update("UPDATE qwen_tool_publication SET producer_phase = 'OPEN',"
|
||||
+ " finish_operation_id = NULL, finish_predecessor_id = NULL, finish_digest = NULL,"
|
||||
+ " producer_used_bytes = CASE WHEN producer_used_bytes >= ? THEN producer_used_bytes - ?"
|
||||
+ " ELSE 0 END WHERE scope_key = ? AND publication_id = ?"
|
||||
+ " AND producer_phase = 'FINISHING' AND finish_operation_id = ?",
|
||||
rollback, rollback, scope, publicationId, operationId);
|
||||
return objectKey;
|
||||
}
|
||||
|
||||
private void expireVerification(String scope, String publicationId, String operationId) {
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET claim_owner = NULL, claim_until = NULL,"
|
||||
+ " verification_next_at = NULL WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
scope, publicationId, operationId);
|
||||
jdbc.update("UPDATE qwen_tool_publication SET active_operation_id = NULL WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND active_operation_id = ?", scope, publicationId, operationId);
|
||||
LOG.info("Tool publication verification expired operation={}", operationId);
|
||||
}
|
||||
|
||||
public JsonNode publishSegment(JsonNode key, String publicationId, String token,
|
||||
String operationId, String streamId, int ordinal, byte[] input, String expectedDigest) {
|
||||
return publishSegment(key, publicationId, token, operationId, streamId, ordinal, input, expectedDigest, false);
|
||||
}
|
||||
|
||||
public JsonNode publishSegment(JsonNode key, String publicationId, String token,
|
||||
String operationId, String streamId, int ordinal, byte[] input, String expectedDigest, boolean asynchronous) {
|
||||
require("stdout".equals(streamId) || "stderr".equals(streamId), "Invalid stream ID");
|
||||
require(ordinal >= 0 && ordinal < 65536, "Invalid ordinal");
|
||||
return publish(key, publicationId, token, operationId, "segment:" + streamId + ":" + ordinal,
|
||||
null, input, expectedDigest, MAX_SEGMENT);
|
||||
null, input, expectedDigest, MAX_SEGMENT, asynchronous);
|
||||
}
|
||||
|
||||
public JsonNode publishResource(JsonNode key, String publicationId, String token,
|
||||
String operationId, String slot, String kind, byte[] input) {
|
||||
return publishResource(key, publicationId, token, operationId, slot, kind, input, false);
|
||||
}
|
||||
|
||||
public JsonNode publishResource(JsonNode key, String publicationId, String token,
|
||||
String operationId, String slot, String kind, byte[] input, boolean asynchronous) {
|
||||
require(kind != null && (("managed-tool-result-page".equals(kind)
|
||||
&& slot.matches("page:(stdout|stderr):(?:0|[1-9][0-9]{0,4})"))
|
||||
|| ("managed-tool-result-manifest".equals(kind) && "manifest:1".equals(slot))
|
||||
@@ -281,11 +724,16 @@ public final class ToolPublicationDataStore {
|
||||
JsonNode manifest = ToolPublicationContract.parseToolResult("manifest", input, MAX_MANIFEST);
|
||||
require(manifest.path("revision").asInt(-1) == 1, "Publication manifest revision conflicts");
|
||||
}
|
||||
return publish(key, publicationId, token, operationId, slot, kind, input, null, maximum);
|
||||
return publish(key, publicationId, token, operationId, slot, kind, input, null, maximum, asynchronous);
|
||||
}
|
||||
|
||||
public JsonNode seal(JsonNode key, String publicationId, String token, String operationId,
|
||||
String streamId, int segmentCount, long byteLength, String digest) {
|
||||
return seal(key, publicationId, token, operationId, streamId, segmentCount, byteLength, digest, false);
|
||||
}
|
||||
|
||||
public JsonNode seal(JsonNode key, String publicationId, String token, String operationId,
|
||||
String streamId, int segmentCount, long byteLength, String digest, boolean asynchronous) {
|
||||
requireStagedCall(key, publicationId);
|
||||
require("stdout".equals(streamId) || "stderr".equals(streamId), "Invalid stream ID");
|
||||
require(segmentCount >= 0 && segmentCount <= 65536 && byteLength >= 0
|
||||
@@ -294,8 +742,14 @@ public final class ToolPublicationDataStore {
|
||||
String slot = "seal:" + streamId;
|
||||
String requestDigest = hash(JSON.createArrayNode().add(slot).add(segmentCount)
|
||||
.add(byteLength).add(digest).toString());
|
||||
JsonNode accepted = submitVerification(key, publicationId, token, operationId, slot, requestDigest,
|
||||
JSON.createObjectNode().put("segmentCount", segmentCount).put("byteLength", byteLength)
|
||||
.put("digest", digest), null, null, null, asynchronous);
|
||||
if (accepted != null) {
|
||||
return accepted;
|
||||
}
|
||||
ScanClaim claim = transactions.execute(status -> claimScan(key, scope, publicationId, token,
|
||||
operationId, slot, requestDigest));
|
||||
operationId, slot, requestDigest, false));
|
||||
require(claim != null, "Seal operation unavailable");
|
||||
if (claim.receipt() != null) {
|
||||
return claim.receipt();
|
||||
@@ -350,13 +804,23 @@ public final class ToolPublicationDataStore {
|
||||
|
||||
public JsonNode prefix(JsonNode key, String publicationId, String token, String operationId,
|
||||
String streamId) {
|
||||
return prefix(key, publicationId, token, operationId, streamId, false);
|
||||
}
|
||||
|
||||
public JsonNode prefix(JsonNode key, String publicationId, String token, String operationId,
|
||||
String streamId, boolean asynchronous) {
|
||||
requireStagedCall(key, publicationId);
|
||||
require("stdout".equals(streamId) || "stderr".equals(streamId), "Invalid stream ID");
|
||||
String scope = scope(key);
|
||||
String slot = "prefix:" + streamId;
|
||||
String requestDigest = hash(slot);
|
||||
JsonNode accepted = submitVerification(key, publicationId, token, operationId, slot, requestDigest,
|
||||
JSON.createObjectNode(), null, null, null, asynchronous);
|
||||
if (accepted != null) {
|
||||
return accepted;
|
||||
}
|
||||
ScanClaim claim = transactions.execute(status -> claimScan(key, scope, publicationId, token,
|
||||
operationId, slot, requestDigest));
|
||||
operationId, slot, requestDigest, false));
|
||||
require(claim != null, "Prefix operation unavailable");
|
||||
if (claim.receipt() != null) {
|
||||
return claim.receipt();
|
||||
@@ -396,7 +860,8 @@ public final class ToolPublicationDataStore {
|
||||
require(operationId != null && operationId.matches("[a-z0-9_-]{1,128}"),
|
||||
"Invalid publication operation ID");
|
||||
String scope = scope(key);
|
||||
var rows = jdbc.query("SELECT o.state, o.receipt_json,"
|
||||
var rows = jdbc.query("SELECT o.state, o.receipt_json, o.failure_status, o.failure_code,"
|
||||
+ " CASE WHEN o.execution_mode = 'ASYNC' AND o.verification_ready THEN 1 ELSE 0 END AS ready,"
|
||||
+ " COALESCE(o.recovery_deadline, o.deadline) AS deadline,"
|
||||
+ " CASE WHEN o.claim_owner IS NULL OR o.claim_until < CURRENT_TIMESTAMP(6)"
|
||||
+ " THEN 1 ELSE 0 END AS retryable, p.tenant_id, p.token_hash,"
|
||||
@@ -404,32 +869,43 @@ public final class ToolPublicationDataStore {
|
||||
+ " JOIN qwen_tool_publication p ON p.scope_key = o.scope_key"
|
||||
+ " AND p.publication_id = o.publication_id WHERE o.scope_key = ?"
|
||||
+ " AND o.publication_id = ? AND o.operation_id = ?",
|
||||
(r, n) -> Map.<String, Object>of("state", r.getString("state"), "deadline", r.getTimestamp("deadline"),
|
||||
"tenant", r.getString("tenant_id"), "workspace", r.getString("workspace_id"),
|
||||
"session", r.getString("session_id"), "tokenHash", r.getString("token_hash"),
|
||||
"retryable", r.getInt("retryable"), "receipt",
|
||||
r.getString("receipt_json") == null ? "" : r.getString("receipt_json")),
|
||||
(r, n) -> {
|
||||
Map<String, Object> row = new HashMap<>();
|
||||
for (String field : List.of("state", "tenant_id", "workspace_id", "session_id", "token_hash",
|
||||
"receipt_json", "failure_code")) {
|
||||
row.put(field, r.getString(field));
|
||||
}
|
||||
row.put("deadline", r.getTimestamp("deadline"));
|
||||
row.put("retryable", r.getInt("retryable"));
|
||||
row.put("ready", r.getInt("ready"));
|
||||
row.put("failure_status", r.getInt("failure_status"));
|
||||
return row;
|
||||
},
|
||||
scope, publicationId, operationId);
|
||||
if (rows.isEmpty()) {
|
||||
throw new ApiException(HttpStatus.NOT_FOUND,
|
||||
"managed_tool_publication_operation_unknown", "Publication operation is unknown");
|
||||
}
|
||||
Map<String, Object> row = rows.get(0);
|
||||
require(text(key, "tenantId").equals(row.get("tenant"))
|
||||
&& text(key, "workspaceId").equals(row.get("workspace"))
|
||||
&& text(key, "sessionId").equals(row.get("session"))
|
||||
require(text(key, "tenantId").equals(row.get("tenant_id"))
|
||||
&& text(key, "workspaceId").equals(row.get("workspace_id"))
|
||||
&& text(key, "sessionId").equals(row.get("session_id"))
|
||||
&& MessageDigest.isEqual(ToolPublicationContract.tokenHash(token)
|
||||
.getBytes(StandardCharsets.US_ASCII),
|
||||
((String) row.get("tokenHash")).getBytes(StandardCharsets.US_ASCII)),
|
||||
((String) row.get("token_hash")).getBytes(StandardCharsets.US_ASCII)),
|
||||
"Publication scope conflicts");
|
||||
String state = (String) row.get("state");
|
||||
ObjectNode response = JSON.createObjectNode().put("state", state);
|
||||
if ("SUCCEEDED".equals(state)) {
|
||||
response.set("receipt", ToolPublicationContract.readJson(
|
||||
((String) row.get("receipt")).getBytes(StandardCharsets.UTF_8)));
|
||||
((String) row.get("receipt_json")).getBytes(StandardCharsets.UTF_8)));
|
||||
} else if ("FAILED".equals(state)) {
|
||||
response.set("error", JSON.createObjectNode().put("status", ((Number) row.get("failure_status")).intValue())
|
||||
.put("code", (String) row.get("failure_code")));
|
||||
} else if (!((Timestamp) row.get("deadline")).after(now())) {
|
||||
response.put("state", "EXPIRED");
|
||||
} else if ("PENDING".equals(state) && ((Number) row.get("retryable")).intValue() == 1) {
|
||||
} else if ("PENDING".equals(state) && ((Number) row.get("retryable")).intValue() == 1
|
||||
&& ((Number) row.get("ready")).intValue() == 0) {
|
||||
response.put("state", "RETRYABLE");
|
||||
}
|
||||
return response;
|
||||
@@ -451,6 +927,8 @@ public final class ToolPublicationDataStore {
|
||||
List<Operation> rows = operation(scope, publicationId, operationId, true);
|
||||
require(rows.size() == 1, "Publication operation is missing");
|
||||
Operation row = rows.get(0);
|
||||
requireContract(!"FAILED".equals(row.state()), HttpStatus.CONFLICT,
|
||||
"managed_tool_publication_failed", "Failed operation cannot be recovered");
|
||||
if (authorization.execution().getState()
|
||||
== ToolExecutionRecord.State.SETTLED) {
|
||||
exactSucceededReplay(scope, publicationId, operationId, row.slot(), row.digest());
|
||||
@@ -485,11 +963,17 @@ public final class ToolPublicationDataStore {
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
new Timestamp(current.getTime() + verificationWindow(scope, publicationId,
|
||||
row.slot()).toMillis()), scope, publicationId, operationId);
|
||||
if (row.asynchronous() && row.ready()) {
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET verification_next_at = ?"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
current, scope, publicationId, operationId);
|
||||
}
|
||||
if (!"terminal".equals(row.slot())) {
|
||||
jdbc.update("UPDATE qwen_tool_publication SET active_operation_id = ?"
|
||||
+ " WHERE scope_key = ? AND publication_id = ?", operationId, scope, publicationId);
|
||||
}
|
||||
});
|
||||
verificationWakeup.run();
|
||||
return operationStatus(key, publicationId, token, operationId);
|
||||
}
|
||||
|
||||
@@ -702,7 +1186,7 @@ public final class ToolPublicationDataStore {
|
||||
retention.put(key, scope, publicationId, candidate.objectKey(), bytes, objects);
|
||||
try {
|
||||
verify(scope, publicationId, candidate.objectKey(), bytes.length, digest);
|
||||
} catch (IllegalArgumentException error) {
|
||||
} catch (CorruptObjectException error) {
|
||||
try {
|
||||
if (candidate.verified()) {
|
||||
quarantine(scope, publicationId, "admission");
|
||||
@@ -754,6 +1238,11 @@ public final class ToolPublicationDataStore {
|
||||
|
||||
public JsonNode finish(JsonNode key, String publicationId, String token,
|
||||
String operationId, byte[] bytes) {
|
||||
return finish(key, publicationId, token, operationId, bytes, false);
|
||||
}
|
||||
|
||||
public JsonNode finish(JsonNode key, String publicationId, String token,
|
||||
String operationId, byte[] bytes, boolean asynchronous) {
|
||||
requireStagedCall(key, publicationId);
|
||||
require(bytes != null, "Missing terminal result");
|
||||
JsonNode result = ToolPublicationContract.parseToolResult("result", bytes, MAX_TERMINAL);
|
||||
@@ -763,8 +1252,13 @@ public final class ToolPublicationDataStore {
|
||||
"Only a started pending capture can finish");
|
||||
String digest = ToolPublicationContract.sha256(bytes);
|
||||
String scope = scope(key);
|
||||
JsonNode accepted = submitVerification(key, publicationId, token, operationId, "terminal",
|
||||
requestDigest("terminal", bytes.length, digest), JSON.createObjectNode(), null, bytes, digest, asynchronous);
|
||||
if (accepted != null) {
|
||||
return accepted;
|
||||
}
|
||||
FinishClaim claim = transactions.execute(status -> beginFinish(key, scope,
|
||||
publicationId, token, operationId, bytes.length, digest));
|
||||
publicationId, token, operationId, bytes.length, digest, false));
|
||||
require(claim != null, "Finish operation unavailable");
|
||||
if (claim.receipt() != null) {
|
||||
finishedInternal(key, publicationId);
|
||||
@@ -773,16 +1267,20 @@ public final class ToolPublicationDataStore {
|
||||
try {
|
||||
if (claim.predecessor() != null) {
|
||||
List<Operation> previous = operation(scope, publicationId, claim.predecessor(), false);
|
||||
requireContract(previous.isEmpty() || !"FAILED".equals(previous.get(0).state()),
|
||||
HttpStatus.CONFLICT, "managed_tool_publication_predecessor_failed", "Finish predecessor failed");
|
||||
requireContract(previous.size() == 1 && "SUCCEEDED".equals(previous.get(0).state()),
|
||||
HttpStatus.CONFLICT, "managed_tool_publication_busy", "Finish predecessor has not completed");
|
||||
}
|
||||
validateFinished(key, publicationId, claim.binding(), result,
|
||||
heartbeat(key, scope, publicationId, token, operationId, claim.epoch()));
|
||||
if (claim.objectKey() != null) {
|
||||
retention.put(key, scope, publicationId, claim.objectKey(), bytes, objects);
|
||||
if (verificationTask.get() == null) {
|
||||
retention.put(key, scope, publicationId, claim.objectKey(), bytes, objects);
|
||||
}
|
||||
try {
|
||||
verify(scope, publicationId, claim.objectKey(), bytes.length, digest);
|
||||
} catch (IllegalArgumentException error) {
|
||||
} catch (CorruptObjectException error) {
|
||||
quarantineCandidate(scope, publicationId, "terminal", operationId);
|
||||
throw error;
|
||||
}
|
||||
@@ -800,7 +1298,7 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private FinishClaim beginFinish(JsonNode key, String scope, String publicationId,
|
||||
String token, String operationId, int length, String digest) {
|
||||
String token, String operationId, int length, String digest, boolean preparing) {
|
||||
require(operationId != null && operationId.matches("[a-z0-9_-]{1,128}"),
|
||||
"Invalid finish operation ID");
|
||||
var authorization = authorizeClaim(key, scope, publicationId, token);
|
||||
@@ -827,6 +1325,13 @@ public final class ToolPublicationDataStore {
|
||||
: "managed-tool-results/" + scope + "/" + publicationId + "/" + hash("terminal");
|
||||
String requestDigest = requestDigest("terminal", length, digest);
|
||||
List<Operation> prior = operation(scope, publicationId, operationId, true);
|
||||
if (!prior.isEmpty()) {
|
||||
requireContract(!"FAILED".equals(prior.get(0).state()), HttpStatus.CONFLICT,
|
||||
"managed_tool_publication_failed", "Publication operation has failed");
|
||||
requireContract(preparing || verificationTask.get() != null || !prior.get(0).asynchronous(),
|
||||
HttpStatus.CONFLICT, "managed_tool_publication_busy", "Publication operation is asynchronous");
|
||||
}
|
||||
|
||||
if ("FINISHED".equals(phase)) {
|
||||
require(operationId.equals(row.get("finish_operation_id"))
|
||||
&& digest.equals(row.get("finish_digest")) && prior.size() == 1
|
||||
@@ -837,6 +1342,16 @@ public final class ToolPublicationDataStore {
|
||||
return new FinishClaim(0, null, objectKey, resourceId, binding,
|
||||
ToolPublicationContract.readJson(saved.getBytes(StandardCharsets.UTF_8)));
|
||||
}
|
||||
if (verificationTask.get() != null) {
|
||||
VerificationTask task = verificationTask.get();
|
||||
requireOperationClaim(prior, task.epoch(), now());
|
||||
require("FINISHING".equals(phase) && task.operationId().equals(operationId)
|
||||
&& operationId.equals(row.get("finish_operation_id"))
|
||||
&& requestDigest.equals(prior.get(0).digest()) && digest.equals(row.get("finish_digest")),
|
||||
"Accepted finish conflicts");
|
||||
return new FinishClaim(task.epoch(), (String) row.get("finish_predecessor_id"),
|
||||
objectKey, resourceId, binding, null);
|
||||
}
|
||||
Timestamp now = now();
|
||||
if ("FINISHING".equals(phase)) {
|
||||
require(!"QUARANTINED".equals(stored(scope, publicationId, "terminal").get(0).state()),
|
||||
@@ -1021,7 +1536,7 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private ScanClaim claimScan(JsonNode key, String scope, String publicationId, String token,
|
||||
String operationId, String slot, String requestDigest) {
|
||||
String operationId, String slot, String requestDigest, boolean preparing) {
|
||||
require(operationId != null && operationId.matches("[a-z0-9_-]{1,128}"),
|
||||
"Invalid publication operation ID");
|
||||
var authorization = authorizeClaim(key, scope, publicationId, token);
|
||||
@@ -1040,6 +1555,20 @@ public final class ToolPublicationDataStore {
|
||||
&& operationId.equals(phase.get("finish_predecessor_id")),
|
||||
"Publication is finishing");
|
||||
List<Operation> prior = operation(scope, publicationId, operationId, true);
|
||||
if (!prior.isEmpty()) {
|
||||
requireContract(!"FAILED".equals(prior.get(0).state()), HttpStatus.CONFLICT,
|
||||
"managed_tool_publication_failed", "Publication operation has failed");
|
||||
requireContract(preparing || verificationTask.get() != null || !prior.get(0).asynchronous(),
|
||||
HttpStatus.CONFLICT, "managed_tool_publication_busy", "Publication operation is asynchronous");
|
||||
}
|
||||
|
||||
if (verificationTask.get() != null) {
|
||||
VerificationTask task = verificationTask.get();
|
||||
requireOperationClaim(prior, task.epoch(), now());
|
||||
require(task.operationId().equals(operationId) && prior.get(0).slot().equals(slot)
|
||||
&& prior.get(0).digest().equals(requestDigest), "Verification operation conflicts");
|
||||
return new ScanClaim(task.epoch(), null);
|
||||
}
|
||||
Timestamp now = now();
|
||||
if (!prior.isEmpty()) {
|
||||
Operation row = prior.get(0);
|
||||
@@ -1114,6 +1643,9 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private void abandonScan(String scope, String publicationId, String operationId, long epoch) {
|
||||
if (verificationTask.get() != null) {
|
||||
return;
|
||||
}
|
||||
transactions.executeWithoutResult(status -> {
|
||||
retention.lockRetainedPublication(scope, publicationId);
|
||||
List<Operation> rows = operation(scope, publicationId, operationId, true);
|
||||
@@ -1158,7 +1690,7 @@ public final class ToolPublicationDataStore {
|
||||
require(row.objectKey() != null, "Publication segment is missing its object");
|
||||
MessageDigest segment = sha256();
|
||||
long segmentLength = 0;
|
||||
try (InputStream stream = retention.open(scope, publicationId, row.objectKey(), objects)) {
|
||||
try (InputStream stream = retention.open(scope, publicationId, row.objectKey(), objects, heartbeat)) {
|
||||
byte[] buffer = new byte[64 * 1024];
|
||||
for (int bytes; (bytes = stream.read(buffer)) != -1; ) {
|
||||
heartbeat.run();
|
||||
@@ -1199,9 +1731,11 @@ public final class ToolPublicationDataStore {
|
||||
requireOperationClaim(rows, epoch, current);
|
||||
long until = Math.min(rows.get(0).deadline().getTime(),
|
||||
current.getTime() + claimTimeout.toMillis());
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET claim_until = ?"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
new Timestamp(until), scope, publicationId, operationId);
|
||||
jdbc.update("UPDATE qwen_tool_publication_operation SET claim_until = ?,"
|
||||
+ " verification_next_at = CASE WHEN execution_mode = 'ASYNC' AND verification_ready"
|
||||
+ " THEN ? ELSE verification_next_at END"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND operation_id = ?",
|
||||
new Timestamp(until), new Timestamp(until), scope, publicationId, operationId);
|
||||
});
|
||||
next = System.nanoTime() + interval;
|
||||
}
|
||||
@@ -1216,8 +1750,8 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
}
|
||||
|
||||
private IllegalArgumentException corruptPublication(String scope, String publicationId, String slot, String message) {
|
||||
var error = new IllegalArgumentException(message);
|
||||
private CorruptObjectException corruptPublication(String scope, String publicationId, String slot, String message) {
|
||||
var error = new CorruptObjectException(message);
|
||||
try {
|
||||
quarantine(scope, publicationId, slot);
|
||||
} catch (RuntimeException cleanup) {
|
||||
@@ -1229,6 +1763,7 @@ public final class ToolPublicationDataStore {
|
||||
private void quarantine(String scope, String publicationId, String slot) {
|
||||
transactions.executeWithoutResult(status -> {
|
||||
retention.lockRetainedPublication(scope, publicationId);
|
||||
requireVerificationClaim(scope, publicationId);
|
||||
jdbc.update("UPDATE qwen_tool_publication SET quarantined = TRUE"
|
||||
+ " WHERE scope_key = ? AND publication_id = ?", scope, publicationId);
|
||||
jdbc.update("UPDATE qwen_tool_publication_object SET state = 'QUARANTINED'"
|
||||
@@ -1241,6 +1776,7 @@ public final class ToolPublicationDataStore {
|
||||
String slot, String operationId) {
|
||||
transactions.executeWithoutResult(status -> {
|
||||
retention.lockRetainedPublication(scope, publicationId);
|
||||
requireVerificationClaim(scope, publicationId);
|
||||
jdbc.update("UPDATE qwen_tool_publication_object"
|
||||
+ " SET state = 'QUARANTINED' WHERE scope_key = ? AND publication_id = ?"
|
||||
+ " AND slot_key = ? AND operation_id = ? AND state = 'CANDIDATE'",
|
||||
@@ -1249,7 +1785,7 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private JsonNode publish(JsonNode key, String publicationId, String token, String operationId,
|
||||
String slot, String kind, byte[] input, String expectedDigest, int maximum) {
|
||||
String slot, String kind, byte[] input, String expectedDigest, int maximum, boolean asynchronous) {
|
||||
requireStagedCall(key, publicationId);
|
||||
require(input != null && input.length > 0 && input.length <= maximum, "Invalid publication size");
|
||||
require(operationId != null && operationId.matches("[a-z0-9_-]{1,128}"),
|
||||
@@ -1259,14 +1795,19 @@ public final class ToolPublicationDataStore {
|
||||
requireContract(expectedDigest == null || expectedDigest.equals(digest),
|
||||
HttpStatus.BAD_REQUEST, "managed_tool_result_digest_mismatch", "Publication digest mismatch");
|
||||
String scope = scope(key);
|
||||
JsonNode accepted = submitVerification(key, publicationId, token, operationId, slot,
|
||||
requestDigest(slot, bytes.length, digest), JSON.createObjectNode(), kind, bytes, digest, asynchronous);
|
||||
if (accepted != null) {
|
||||
return accepted;
|
||||
}
|
||||
Candidate candidate = transactions.execute(status -> claim(key, scope, publicationId, token,
|
||||
operationId, slot, kind, bytes.length, digest));
|
||||
operationId, slot, kind, bytes.length, digest, false));
|
||||
require(candidate != null, "Publication operation unavailable");
|
||||
if (candidate.receipt() != null) {
|
||||
if (candidate.objectKey() != null) {
|
||||
try {
|
||||
verify(scope, publicationId, candidate.objectKey(), bytes.length, digest);
|
||||
} catch (IllegalArgumentException error) {
|
||||
} catch (CorruptObjectException error) {
|
||||
try {
|
||||
quarantine(scope, publicationId, slot);
|
||||
} catch (RuntimeException cleanup) {
|
||||
@@ -1284,13 +1825,13 @@ public final class ToolPublicationDataStore {
|
||||
retention.put(key, scope, publicationId, candidate.objectKey(), bytes, objects);
|
||||
try {
|
||||
verify(scope, publicationId, candidate.objectKey(), bytes.length, digest);
|
||||
} catch (IllegalArgumentException error) {
|
||||
} catch (CorruptObjectException error) {
|
||||
quarantineCandidate(scope, publicationId, slot, operationId);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return transactions.execute(status -> install(key, scope, publicationId, token,
|
||||
operationId, candidate, kind, bytes, digest));
|
||||
operationId, candidate, kind, bytes.length, bytes, digest));
|
||||
} catch (RuntimeException error) {
|
||||
try {
|
||||
abandonScan(scope, publicationId, operationId, candidate.epoch());
|
||||
@@ -1302,7 +1843,7 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private Candidate claim(JsonNode key, String scope, String publicationId, String token,
|
||||
String operationId, String slot, String kind, int length, String digest) {
|
||||
String operationId, String slot, String kind, int length, String digest, boolean preparing) {
|
||||
var authorization = authorizeClaim(key, scope, publicationId, token);
|
||||
JsonNode binding = authorization.binding();
|
||||
if (authorization.execution().getState()
|
||||
@@ -1326,6 +1867,13 @@ public final class ToolPublicationDataStore {
|
||||
"Publication is finishing");
|
||||
String requestDigest = requestDigest(slot, length, digest);
|
||||
List<Operation> prior = operation(scope, publicationId, operationId, true);
|
||||
if (!prior.isEmpty()) {
|
||||
requireContract(!"FAILED".equals(prior.get(0).state()), HttpStatus.CONFLICT,
|
||||
"managed_tool_publication_failed", "Publication operation has failed");
|
||||
requireContract(preparing || verificationTask.get() != null || !prior.get(0).asynchronous(),
|
||||
HttpStatus.CONFLICT, "managed_tool_publication_busy", "Publication operation is asynchronous");
|
||||
}
|
||||
|
||||
if (!prior.isEmpty()) {
|
||||
require(requestDigest.equals(prior.get(0).digest()) && slot.equals(prior.get(0).slot()),
|
||||
"Publication operation conflicts");
|
||||
@@ -1397,7 +1945,7 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private JsonNode install(JsonNode key, String scope, String publicationId, String token,
|
||||
String operationId, Candidate candidate, String kind, byte[] bytes, String digest) {
|
||||
String operationId, Candidate candidate, String kind, int length, byte[] bytes, String digest) {
|
||||
JsonNode binding = authorize(key, scope, publicationId, token);
|
||||
String active = jdbc.queryForObject("SELECT active_operation_id FROM qwen_tool_publication"
|
||||
+ " WHERE scope_key = ? AND publication_id = ?", String.class, scope, publicationId);
|
||||
@@ -1407,13 +1955,13 @@ public final class ToolPublicationDataStore {
|
||||
"managed_tool_publication_claim_lost", "Publication operation lost its claim");
|
||||
Stored row = stored(scope, publicationId, rows.get(0).slot(), true).get(0);
|
||||
require("CANDIDATE".equals(row.state()) && digest.equals(row.digest())
|
||||
&& row.length() == bytes.length, "Publication candidate changed");
|
||||
&& row.length() == length, "Publication candidate changed");
|
||||
if (candidate.objectKey() == null) {
|
||||
jdbc.update("UPDATE qwen_tool_publication_object SET inline_bytes = ?"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND slot_key = ?",
|
||||
bytes, scope, publicationId, row.slot());
|
||||
}
|
||||
JsonNode receipt = receipt(binding, row.slot(), candidate.resourceId(), kind, bytes.length, digest);
|
||||
JsonNode receipt = receipt(binding, row.slot(), candidate.resourceId(), kind, length, digest);
|
||||
jdbc.update("UPDATE qwen_tool_publication_object SET state = 'VERIFIED'"
|
||||
+ " WHERE scope_key = ? AND publication_id = ? AND slot_key = ?",
|
||||
scope, publicationId, row.slot());
|
||||
@@ -1443,14 +1991,16 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private JsonNode authorize(JsonNode key, String scope, String publicationId, String token) {
|
||||
JsonNode binding = grants.producerMutationBindingLocked(scope, publicationId, token);
|
||||
JsonNode binding = token == null ? verificationBinding(scope, publicationId)
|
||||
: grants.producerMutationBindingLocked(scope, publicationId, token);
|
||||
require(binding.path("sessionKey").equals(key), "Publication scope conflicts");
|
||||
return binding;
|
||||
}
|
||||
|
||||
private ToolPublicationStore.ProducerBinding authorizeClaim(JsonNode key, String scope,
|
||||
String publicationId, String token) {
|
||||
var authorization = grants.producerClaimBindingLocked(scope, publicationId, token);
|
||||
var authorization = token == null ? verificationAuthorization(scope, publicationId)
|
||||
: grants.producerClaimBindingLocked(scope, publicationId, token);
|
||||
require(authorization.binding().path("sessionKey").equals(key), "Publication scope conflicts");
|
||||
return authorization;
|
||||
}
|
||||
@@ -1530,28 +2080,37 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private List<Operation> operation(String scope, String publicationId, String operationId, boolean locked) {
|
||||
return jdbc.query("SELECT request_digest, slot_key, state, claim_epoch, claim_until,"
|
||||
return jdbc.query("SELECT request_digest, slot_key, state, claim_epoch, claim_until, execution_mode,"
|
||||
+ " verification_ready,"
|
||||
+ " COALESCE(recovery_deadline, deadline) AS deadline"
|
||||
+ " FROM qwen_tool_publication_operation WHERE scope_key = ? AND publication_id = ?"
|
||||
+ " AND operation_id = ?" + (locked ? " FOR UPDATE" : ""),
|
||||
(r, n) -> new Operation(r.getString("request_digest"), r.getString("slot_key"),
|
||||
r.getString("state"), r.getLong("claim_epoch"), r.getTimestamp("claim_until"),
|
||||
r.getTimestamp("deadline")), scope, publicationId, operationId);
|
||||
r.getTimestamp("deadline"), "ASYNC".equals(r.getString("execution_mode")),
|
||||
r.getBoolean("verification_ready")), scope, publicationId, operationId);
|
||||
}
|
||||
|
||||
private void verify(String scope, String publicationId, String objectKey, long length, String digest) {
|
||||
VerificationTask task = verificationTask.get();
|
||||
Runnable guard = task == null ? () -> {} : verificationHeartbeat(task);
|
||||
guard.run();
|
||||
objects.requireUnversioned();
|
||||
try (InputStream stream = retention.open(scope, publicationId, objectKey, objects)) {
|
||||
try (InputStream stream = retention.open(scope, publicationId, objectKey, objects, guard)) {
|
||||
var hash = java.security.MessageDigest.getInstance("SHA-256");
|
||||
byte[] buffer = new byte[64 * 1024];
|
||||
long read = 0;
|
||||
for (int count; (count = stream.read(buffer)) != -1; ) {
|
||||
guard.run();
|
||||
hash.update(buffer, 0, count);
|
||||
read += count;
|
||||
require(read <= length, "Publication object length changed");
|
||||
if (read > length) {
|
||||
throw new CorruptObjectException("Publication object length changed");
|
||||
}
|
||||
}
|
||||
if (read != length || !HexFormat.of().formatHex(hash.digest()).equals(digest)) {
|
||||
throw new CorruptObjectException("Publication object digest changed");
|
||||
}
|
||||
require(read == length && HexFormat.of().formatHex(hash.digest()).equals(digest),
|
||||
"Publication object digest changed");
|
||||
} catch (IOException | java.security.NoSuchAlgorithmException error) {
|
||||
throw new IllegalStateException("Publication object verification failed", error);
|
||||
}
|
||||
@@ -2054,6 +2613,17 @@ public final class ToolPublicationDataStore {
|
||||
return jdbc.queryForObject("SELECT CURRENT_TIMESTAMP(6)", Timestamp.class);
|
||||
}
|
||||
|
||||
private static final class CorruptObjectException extends IllegalArgumentException {
|
||||
private CorruptObjectException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
|
||||
private record Submission(long epoch, String objectKey, JsonNode response) {}
|
||||
|
||||
private record VerificationTask(String scope, String publicationId, String operationId, String slot,
|
||||
long epoch, JsonNode key, JsonNode inputs, long queueMillis) {}
|
||||
|
||||
private record Candidate(String objectKey, String resourceId, long epoch, JsonNode receipt) {
|
||||
}
|
||||
|
||||
@@ -2075,7 +2645,7 @@ public final class ToolPublicationDataStore {
|
||||
}
|
||||
|
||||
private record Operation(String digest, String slot, String state, long epoch,
|
||||
Timestamp claimUntil, Timestamp deadline) {
|
||||
Timestamp claimUntil, Timestamp deadline, boolean asynchronous, boolean ready) {
|
||||
}
|
||||
|
||||
private record Resource(String slot, String kind, long length, String digest, String objectKey,
|
||||
|
||||
+1
-1
@@ -318,7 +318,7 @@ public final class ToolPublicationRetentionStore {
|
||||
+ " AND state <> 'RETURNED'", scope, publication) != 0) {
|
||||
blocker = "put_unresolved";
|
||||
} else if (count("SELECT COUNT(*) FROM qwen_tool_publication_operation WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND state <> 'SUCCEEDED'", scope, publication) != 0) {
|
||||
+ " AND publication_id = ? AND state NOT IN ('SUCCEEDED', 'FAILED')", scope, publication) != 0) {
|
||||
blocker = "operation_unresolved";
|
||||
} else if (count("SELECT COUNT(*) FROM qwen_tool_publication_object WHERE scope_key = ?"
|
||||
+ " AND publication_id = ? AND state <> 'VERIFIED'", scope, publication) != 0) {
|
||||
|
||||
+104
-15
@@ -5,6 +5,8 @@ import static com.alibaba.qwen.code.managedagent.store.ToolPublicationContract.t
|
||||
|
||||
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.managedagent.api.ApiException;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRecord;
|
||||
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.JdbcRuntimeBindingRepository;
|
||||
@@ -22,6 +24,7 @@ import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import javax.sql.DataSource;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.jdbc.core.ConnectionCallback;
|
||||
import org.springframework.jdbc.datasource.DataSourceUtils;
|
||||
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
||||
@@ -166,18 +169,68 @@ public final class ToolPublicationStore {
|
||||
private ProducerBinding producerBindingLocked(String scope, String publicationId,
|
||||
String publicationToken, Access access) {
|
||||
String suppliedHash = ToolPublicationContract.tokenHash(publicationToken);
|
||||
return producerBindingWithHashLocked(scope, publicationId, suppliedHash, null, access);
|
||||
}
|
||||
|
||||
ProducerBinding producerVerificationBindingLocked(String scope, String publicationId,
|
||||
String bindingDigest, String tokenHash) {
|
||||
try {
|
||||
require(bindingDigest != null && tokenHash != null, "Accepted verification identity is missing");
|
||||
// Completing an accepted claim. CLAIM still allows a settled CSI execution
|
||||
// after retirement; PRODUCE would fail that claim permanently.
|
||||
return producerBindingWithHashLocked(scope, publicationId, tokenHash, bindingDigest,
|
||||
Access.CLAIM, true);
|
||||
} catch (IllegalArgumentException error) {
|
||||
throw new ApiException(HttpStatus.CONFLICT, "managed_tool_publication_fenced",
|
||||
"Original publication authority is fenced", error);
|
||||
}
|
||||
}
|
||||
|
||||
private ProducerBinding producerBindingWithHashLocked(String scope, String publicationId,
|
||||
String suppliedHash, String expectedBindingDigest, Access access) {
|
||||
return producerBindingWithHashLocked(scope, publicationId, suppliedHash,
|
||||
expectedBindingDigest, access, false);
|
||||
}
|
||||
|
||||
private ProducerBinding producerBindingWithHashLocked(String scope, String publicationId,
|
||||
String suppliedHash, String expectedBindingDigest, Access access,
|
||||
boolean deferActivationLapse) {
|
||||
Row row = publicationRow(scope, publicationId);
|
||||
JsonNode binding = storedBinding(row.binding(), row.digest(), "Publication grant conflicts");
|
||||
require(equalHash(suppliedHash, row.tokenHash()) && "OPEN".equals(row.state()),
|
||||
require((expectedBindingDigest == null || expectedBindingDigest.equals(row.digest()))
|
||||
&& equalHash(suppliedHash, row.tokenHash()) && "OPEN".equals(row.state()),
|
||||
"Publication grant conflicts");
|
||||
Original original = lockOriginal(binding);
|
||||
Original original;
|
||||
try {
|
||||
original = lockOriginal(binding);
|
||||
} catch (RuntimeException error) {
|
||||
// Accepted tasks must leave the queue when their original CSI
|
||||
// authority is gone; credential/SQL failures may recover elsewhere.
|
||||
boolean missingRuntime = error instanceof IllegalStateException && error.getCause() == null
|
||||
&& ("Runtime binding slot is unavailable".equals(error.getMessage())
|
||||
|| "Original Runtime binding is unavailable".equals(error.getMessage()));
|
||||
boolean fencedCsi = error instanceof RuntimeBrokerException broker
|
||||
&& "workspace_csi_unavailable".equals(broker.getCode()) && !broker.isRetryable();
|
||||
if (expectedBindingDigest != null && (missingRuntime || fencedCsi)) {
|
||||
throw new ApiException(HttpStatus.CONFLICT, "managed_tool_publication_runtime_unavailable",
|
||||
"Original publication runtime is unavailable");
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
lockTenant(row.tenant());
|
||||
return producerBindingAfterParentLocked(scope, publicationId, suppliedHash, row,
|
||||
binding, original, access);
|
||||
binding, original, access, deferActivationLapse);
|
||||
}
|
||||
|
||||
private ProducerBinding producerBindingAfterParentLocked(String scope, String publicationId,
|
||||
String suppliedHash, Row row, JsonNode binding, Original original, Access access) {
|
||||
return producerBindingAfterParentLocked(scope, publicationId, suppliedHash, row,
|
||||
binding, original, access, false);
|
||||
}
|
||||
|
||||
private ProducerBinding producerBindingAfterParentLocked(String scope, String publicationId,
|
||||
String suppliedHash, Row row, JsonNode binding, Original original, Access access,
|
||||
boolean deferActivationLapse) {
|
||||
JsonNode key = binding.get("sessionKey");
|
||||
ToolPublicationRetentionStore.requireLive(jdbc, row.tenant(), row.session());
|
||||
require(row.tenant().equals(text(key, "tenantId"))
|
||||
@@ -198,8 +251,7 @@ public final class ToolPublicationStore {
|
||||
require(((Number) current.get("quarantined")).intValue() == 0, "Publication is quarantined");
|
||||
if (access != Access.SETTLE) {
|
||||
require(equalHash(suppliedHash, (String) current.get("token_hash"))
|
||||
&& "OPEN".equals(current.get("state")) && current.get("expires_at") != null
|
||||
&& ((Number) current.get("expires_at")).longValue() > nowEpoch,
|
||||
&& "OPEN".equals(current.get("state")) && current.get("expires_at") != null,
|
||||
"Publication grant changed or expired");
|
||||
}
|
||||
require(row.digest().equals(current.get("binding_digest"))
|
||||
@@ -210,9 +262,19 @@ public final class ToolPublicationStore {
|
||||
require(row.workspace().equals(head.get("workspace_id")) && "ACTIVE".equals(head.get("state"))
|
||||
&& text(binding, "writerId").equals(head.get("writer_id"))
|
||||
&& ((Number) head.get("writer_generation")).longValue() == binding.get("writerGeneration").longValue()
|
||||
&& ((Number) head.get("writer_live")).intValue() == 1 && "READY".equals(head.get("recovery_status"))
|
||||
&& ((Number) head.get("writer_live")).intValue() == 1
|
||||
&& ((Number) head.get("activation_epoch")).longValue() == binding.get("activationEpoch").longValue(),
|
||||
"Original Session owner is fenced");
|
||||
require("READY".equals(head.get("recovery_status")), "Original Session owner is fenced");
|
||||
// The grant expiry is min(writer lease, activation horizon), so a
|
||||
// real activation lapse expires the grant on the same clock and
|
||||
// this check runs before the activation deferral below.
|
||||
if (access != Access.SETTLE && ((Number) current.get("expires_at")).longValue() <= nowEpoch) {
|
||||
if (deferActivationLapse) {
|
||||
throw new ActivationLapsedException();
|
||||
}
|
||||
require(false, "Publication grant changed or expired");
|
||||
}
|
||||
long activationUntil;
|
||||
// The stamp says which journal revision the columns were written
|
||||
// from; a pre-V36 binary's commit bumps journal_revision without
|
||||
@@ -227,20 +289,31 @@ public final class ToolPublicationStore {
|
||||
&& text(binding, "activationId").equals(head.get("activation_id"))
|
||||
&& head.get("activation_event_epoch") instanceof Number epoch
|
||||
&& epoch.longValue() == binding.get("activationEpoch").longValue()
|
||||
&& head.get("activation_expires_at") instanceof Number expires
|
||||
&& expires.longValue() > nowEpoch, "Original activation is fenced");
|
||||
activationUntil = ((Number) head.get("activation_expires_at")).longValue();
|
||||
&& head.get("activation_expires_at") instanceof Number,
|
||||
"Original activation is fenced");
|
||||
activationUntil = requireCurrentActivation(
|
||||
((Number) head.get("activation_expires_at")).longValue(), nowEpoch,
|
||||
deferActivationLapse);
|
||||
} else {
|
||||
activationUntil = requireLegacyActivation(row.tenant(), row.session(), binding,
|
||||
((Number) head.get("journal_revision")).longValue(),
|
||||
nowEpoch, head);
|
||||
nowEpoch, head, deferActivationLapse);
|
||||
}
|
||||
nowEpoch = ToolPublicationRetentionStore.now(jdbc);
|
||||
Integer writerLive = jdbc.queryForObject("SELECT CASE WHEN writer_lease_until > CURRENT_TIMESTAMP(6)"
|
||||
+ " THEN 1 ELSE 0 END FROM qwen_managed_session_journal_head"
|
||||
+ " WHERE tenant_id = ? AND session_id = ? FOR UPDATE", Integer.class, row.tenant(), row.session());
|
||||
require(writerLive != null && writerLive == 1 && activationUntil > nowEpoch,
|
||||
"Original Session or activation expired");
|
||||
if (writerLive == null || writerLive != 1) {
|
||||
require(false, "Original Session or activation expired");
|
||||
}
|
||||
if (activationUntil <= nowEpoch) {
|
||||
// Renewal keeps the activation id and epoch, so a lapsed horizon can
|
||||
// recover. An expired writer lease is not renewable and stays above.
|
||||
if (deferActivationLapse) {
|
||||
throw new ActivationLapsedException();
|
||||
}
|
||||
require(false, "Original Session or activation expired");
|
||||
}
|
||||
if (access != Access.SETTLE) {
|
||||
require(((Number) current.get("expires_at")).longValue() > nowEpoch, "Publication grant expired");
|
||||
}
|
||||
@@ -255,7 +328,7 @@ public final class ToolPublicationStore {
|
||||
*/
|
||||
private long requireLegacyActivation(String tenant, String session,
|
||||
JsonNode binding, long journalRevision, long nowEpoch,
|
||||
java.util.Map<String, Object> head) {
|
||||
java.util.Map<String, Object> head, boolean deferActivationLapse) {
|
||||
JsonNode found = null;
|
||||
for (long revision = journalRevision; revision > 0 && found == null; revision--) {
|
||||
// The locking head read can see a newer revision than this transaction's snapshot.
|
||||
@@ -286,8 +359,8 @@ public final class ToolPublicationStore {
|
||||
require("active".equals(text(found, "phase"))
|
||||
&& text(binding, "activationId").equals(text(found, "activationId"))
|
||||
&& binding.get("activationEpoch").longValue() == found.path("epoch").asLong()
|
||||
&& expiresAt != null && expiresAt > nowEpoch,
|
||||
"Original activation is fenced");
|
||||
&& expiresAt != null, "Original activation is fenced");
|
||||
expiresAt = requireCurrentActivation(expiresAt, nowEpoch, deferActivationLapse);
|
||||
backfillActivation(tenant, session, text(found, "activationId"),
|
||||
text(found, "phase"), found.path("epoch").asLong(),
|
||||
expiresAt, journalRevision,
|
||||
@@ -988,6 +1061,22 @@ public final class ToolPublicationStore {
|
||||
return ToolPublicationContract.parse("grant", grant);
|
||||
}
|
||||
|
||||
private static long requireCurrentActivation(Long expiresAt, long nowEpoch, boolean deferLapse) {
|
||||
if (expiresAt == null || expiresAt <= nowEpoch) {
|
||||
if (deferLapse && expiresAt != null) {
|
||||
throw new ActivationLapsedException();
|
||||
}
|
||||
require(false, "Original activation is fenced");
|
||||
}
|
||||
return expiresAt;
|
||||
}
|
||||
|
||||
private static final class ActivationLapsedException extends RuntimeException {
|
||||
private ActivationLapsedException() {
|
||||
super("Original activation expired");
|
||||
}
|
||||
}
|
||||
|
||||
private static String hash(String text) {
|
||||
return ToolPublicationContract.sha256(text.getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
+68
@@ -0,0 +1,68 @@
|
||||
package com.alibaba.qwen.code.managedagent.store;
|
||||
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.Semaphore;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
|
||||
/** Database-backed verification uses a separate, bounded pool from retention. */
|
||||
public final class ToolPublicationVerifier implements AutoCloseable {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(ToolPublicationVerifier.class);
|
||||
private final ToolPublicationDataStore data;
|
||||
private final ThreadPoolExecutor executor;
|
||||
private final Semaphore slots;
|
||||
private final int concurrency;
|
||||
private volatile boolean closed;
|
||||
|
||||
public ToolPublicationVerifier(ToolPublicationDataStore data, int concurrency) {
|
||||
ToolPublicationContract.require(concurrency > 0, "Invalid verification concurrency");
|
||||
this.data = data;
|
||||
this.concurrency = concurrency;
|
||||
slots = new Semaphore(concurrency);
|
||||
executor = new ThreadPoolExecutor(concurrency, concurrency, 0, TimeUnit.MILLISECONDS,
|
||||
new ArrayBlockingQueue<>(concurrency), Thread.ofPlatform()
|
||||
.name("managed-tool-verification-", 0).factory());
|
||||
data.setVerificationWakeup(this::wake);
|
||||
}
|
||||
|
||||
@Scheduled(fixedDelay = 1000, scheduler = "managedToolVerificationScheduler")
|
||||
public void wake() {
|
||||
for (int i = 0; i < concurrency && !closed && slots.tryAcquire(); i++) {
|
||||
try {
|
||||
executor.execute(() -> {
|
||||
try {
|
||||
while (!closed && runOnce()) {
|
||||
// Drain immediately instead of delaying each sequential upload by a tick.
|
||||
}
|
||||
} finally {
|
||||
slots.release();
|
||||
}
|
||||
});
|
||||
} catch (java.util.concurrent.RejectedExecutionException error) {
|
||||
slots.release();
|
||||
if (!closed) {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public boolean runOnce() {
|
||||
try {
|
||||
return data.verifyNextOperation();
|
||||
} catch (RuntimeException error) {
|
||||
LOG.warn("Tool publication verification will retry failure={}", error.getClass().getSimpleName());
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
closed = true;
|
||||
data.setVerificationWakeup(() -> {});
|
||||
executor.shutdownNow();
|
||||
}
|
||||
}
|
||||
@@ -56,6 +56,8 @@ qwen:
|
||||
gc-enabled: '${QWEN_MANAGED_AGENT_TOOL_PUBLICATION_GC_ENABLED:false}'
|
||||
deletion-grace: '${QWEN_MANAGED_AGENT_TOOL_PUBLICATION_DELETION_GRACE:24h}'
|
||||
journal-head-authorization: '${QWEN_MANAGED_AGENT_TOOL_PUBLICATION_JOURNAL_HEAD_AUTHORIZATION:false}'
|
||||
async-verification-enabled: '${QWEN_MANAGED_AGENT_TOOL_PUBLICATION_ASYNC_VERIFICATION_ENABLED:false}'
|
||||
verification-concurrency: '${QWEN_MANAGED_AGENT_TOOL_PUBLICATION_VERIFICATION_CONCURRENCY:2}'
|
||||
harness:
|
||||
enabled: '${QWEN_MANAGED_AGENT_HARNESS_ENABLED:false}'
|
||||
workspace-files-enabled: '${QWEN_MANAGED_AGENT_WORKSPACE_FILES_ENABLED:false}'
|
||||
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
ALTER TABLE qwen_tool_publication_operation
|
||||
ADD COLUMN execution_mode VARCHAR(16) NOT NULL DEFAULT 'SYNC';
|
||||
ALTER TABLE qwen_tool_publication_operation
|
||||
ADD COLUMN verification_request_json MEDIUMTEXT;
|
||||
ALTER TABLE qwen_tool_publication_operation
|
||||
ADD COLUMN verification_ready BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
ALTER TABLE qwen_tool_publication_operation
|
||||
ADD COLUMN verification_next_at DATETIME(6);
|
||||
ALTER TABLE qwen_tool_publication_operation
|
||||
ADD COLUMN failure_status INT;
|
||||
ALTER TABLE qwen_tool_publication_operation
|
||||
ADD COLUMN failure_code VARCHAR(128);
|
||||
|
||||
CREATE INDEX ix_tool_publication_verification_due
|
||||
ON qwen_tool_publication_operation
|
||||
(execution_mode, state, verification_ready, verification_next_at);
|
||||
+12
@@ -20,10 +20,22 @@ class ToolPublicationConfigurationTest {
|
||||
void disabledPublicationHasNoImplicitVerificationBudget() {
|
||||
var settings = new ManagedAgentProperties().getToolPublication();
|
||||
assertThat(settings.isEnabled()).isFalse();
|
||||
assertThat(settings.isAsyncVerificationEnabled()).isFalse();
|
||||
assertThat(settings.getVerificationConcurrency()).isEqualTo(2);
|
||||
assertThat(settings.getVerificationBytesPerSecond()).isNull();
|
||||
assertThat(settings.getMaxVerificationTimeout()).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void asyncAdmissionRequiresJournalHeadAuthorizationAndASeparateBoundedPool() {
|
||||
var properties = new ManagedAgentProperties();
|
||||
properties.getToolPublication().setAsyncVerificationEnabled(true);
|
||||
assertThatThrownBy(() -> data(properties)).hasMessageContaining("requires journal-head authorization");
|
||||
properties.getToolPublication().setVerificationConcurrency(0);
|
||||
assertThatThrownBy(() -> new ToolPublicationConfiguration().toolPublicationVerifier(
|
||||
mock(ToolPublicationDataStore.class), properties)).hasMessageContaining("Invalid verification concurrency");
|
||||
}
|
||||
|
||||
@Test
|
||||
void requiresExplicitBudgetAndRejectsCapacityThatCannotFit() {
|
||||
var properties = new ManagedAgentProperties();
|
||||
|
||||
+74
@@ -2,8 +2,11 @@ package com.alibaba.qwen.code.managedagent.api;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.verifyNoInteractions;
|
||||
import static org.mockito.Mockito.when;
|
||||
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
|
||||
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
|
||||
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
|
||||
@@ -28,6 +31,7 @@ import org.flywaydb.core.Flyway;
|
||||
import org.h2.jdbcx.JdbcDataSource;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.params.ParameterizedTest;
|
||||
import org.junit.jupiter.params.provider.CsvSource;
|
||||
import org.junit.jupiter.params.provider.ValueSource;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
@@ -39,6 +43,76 @@ import org.springframework.test.web.servlet.setup.MockMvcBuilders;
|
||||
import org.springframework.transaction.support.TransactionTemplate;
|
||||
|
||||
class ToolPublicationControllerTest {
|
||||
@ParameterizedTest
|
||||
@CsvSource({"false,1,false", "true,1,true", "true,0,false", "true,,false"})
|
||||
void asyncAdmissionRequiresBothTheFlagAndCapability(boolean enabled, String header, boolean asynchronous)
|
||||
throws Exception {
|
||||
ManagedAgentProperties properties = new ManagedAgentProperties();
|
||||
properties.getToolPublication().setEntryConcurrency(1);
|
||||
properties.getToolPublication().setAsyncVerificationEnabled(enabled);
|
||||
ToolPublicationDataStore data = mock(ToolPublicationDataStore.class);
|
||||
ToolPublicationController controller = new ToolPublicationController(
|
||||
mock(ToolPublicationStore.class), data, mock(ToolPublicationAdmissionStore.class), properties);
|
||||
var result = new ObjectMapper().createObjectNode();
|
||||
if (asynchronous) result.put("state", "PENDING");
|
||||
else result.put("digest", "legacy-receipt");
|
||||
when(data.publishSegment(any(), eq("publication"), eq("token"), eq("operation"),
|
||||
eq("stdout"), eq(0), any(), eq(null), eq(asynchronous))).thenReturn(result);
|
||||
MockHttpServletRequest request = new MockHttpServletRequest();
|
||||
request.setContent(new byte[] {1});
|
||||
if (header != null) request.addHeader(ToolPublicationController.ASYNC_HEADER, header);
|
||||
var response = controller.segment(new TenantContext("tenant", null), "session", "publication",
|
||||
"stdout", 0, "workspace", "token", "operation", null, request);
|
||||
assertThat(response.getStatusCode().value()).isEqualTo(asynchronous ? 202 : 200);
|
||||
assertThat(response.getBody()).isEqualTo(result);
|
||||
assertThat(response.getHeaders().getCacheControl()).isEqualTo("no-store");
|
||||
if (asynchronous) {
|
||||
result.put("state", "SUCCEEDED");
|
||||
assertThat(controller.segment(new TenantContext("tenant", null), "session", "publication",
|
||||
"stdout", 0, "workspace", "token", "operation", null, request)
|
||||
.getStatusCode().value()).isEqualTo(200);
|
||||
}
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@org.junit.jupiter.params.provider.ValueSource(strings = {"resource", "seal", "prefix", "finish"})
|
||||
void allProducerRoutesForwardAsyncCapabilityAndCompletedReplays(String route) throws Exception {
|
||||
var properties = new ManagedAgentProperties();
|
||||
properties.getToolPublication().setEntryConcurrency(1);
|
||||
properties.getToolPublication().setAsyncVerificationEnabled(true);
|
||||
var data = mock(ToolPublicationDataStore.class);
|
||||
var controller = new ToolPublicationController(mock(ToolPublicationStore.class), data,
|
||||
mock(ToolPublicationAdmissionStore.class), properties);
|
||||
var result = new ObjectMapper().createObjectNode().put("state", "PENDING");
|
||||
when(data.publishResource(any(), eq("publication"), eq("token"), eq("operation"),
|
||||
eq("content:test"), eq("managed-tool-result-content"), any(), eq(true))).thenReturn(result);
|
||||
when(data.seal(any(), eq("publication"), eq("token"), eq("operation"), eq("stdout"),
|
||||
eq(1), eq(1L), eq("digest"), eq(true))).thenReturn(result);
|
||||
when(data.prefix(any(), eq("publication"), eq("token"), eq("operation"), eq("stdout"), eq(true)))
|
||||
.thenReturn(result);
|
||||
when(data.finish(any(), eq("publication"), eq("token"), eq("operation"), any(), eq(true)))
|
||||
.thenReturn(result);
|
||||
for (String state : new String[] {"PENDING", "SUCCEEDED"}) {
|
||||
result.put("state", state);
|
||||
var request = new MockHttpServletRequest();
|
||||
request.addHeader(ToolPublicationController.ASYNC_HEADER, "1");
|
||||
request.setContent("{\"segmentCount\":1,\"byteLength\":1,\"digest\":\"digest\"}"
|
||||
.getBytes(StandardCharsets.UTF_8));
|
||||
var tenant = new TenantContext("tenant", null);
|
||||
var response = switch (route) {
|
||||
case "resource" -> controller.resource(tenant, "session", "publication", "managed-tool-result-content",
|
||||
"content:test", "workspace", "token", "operation", request);
|
||||
case "seal" -> controller.seal(tenant, "session", "publication", "stdout", "workspace", "token", "operation", request);
|
||||
case "prefix" -> controller.prefix(tenant, "session", "publication", "stdout", "workspace", "token", "operation", "1");
|
||||
case "finish" -> controller.finish(tenant, "session", "publication", "workspace", "token", "operation", request);
|
||||
default -> throw new AssertionError(route);
|
||||
};
|
||||
assertThat(response.getStatusCode().value()).isEqualTo(state.equals("PENDING") ? 202 : 200);
|
||||
assertThat(response.getBody()).isSameAs(result);
|
||||
assertThat(response.getHeaders().getCacheControl()).isEqualTo("no-store");
|
||||
}
|
||||
}
|
||||
|
||||
private static final String TENANT = "tenant";
|
||||
private static final String WORKSPACE = "workspace";
|
||||
private static final String SESSION = "session";
|
||||
|
||||
+99
-2
@@ -25,6 +25,7 @@ import com.alibaba.qwen.code.managedagent.store.StoreModels.MaterializationTarge
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationDataStore;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationObjectStore;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationStore;
|
||||
import com.alibaba.qwen.code.managedagent.store.ToolPublicationVerifier;
|
||||
import com.alibaba.qwen.code.runtimebroker.JdbcRuntimeBindingRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.JdbcToolExecutionRepository;
|
||||
import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
|
||||
@@ -78,8 +79,8 @@ import org.springframework.transaction.support.TransactionTemplate;
|
||||
* window, not per delivered event.</li>
|
||||
* <li>listPublicSessions / listWebShellSessions assemble a page from a
|
||||
* fixed number of grouped batch queries.</li>
|
||||
* <li>Tool-publication authorization reads the activation state from the
|
||||
* journal head when journal-head-authorization is enabled (the flag
|
||||
* <li>Tool-publication authorization and async heartbeats
|
||||
* read the activation state from the journal head when journal-head-authorization is enabled (the flag
|
||||
* ships false), rescanning the journal only for pre-migration heads
|
||||
* (and backfilling them).</li>
|
||||
* </ol>
|
||||
@@ -1114,6 +1115,102 @@ class Issue13181QueryBudgetTest {
|
||||
journal.executions, journal.store);
|
||||
}
|
||||
|
||||
@Test
|
||||
void asynchronousVerificationHeartbeatsAndStatusAvoidJournalBodyReads() {
|
||||
Fixture fixture = new Fixture();
|
||||
PublicationFixture publication = publicationFixture(fixture);
|
||||
byte[] bytes = new byte[] {1, 2, 3, 4};
|
||||
ToolPublicationObjectStore bucket = new ToolPublicationObjectStore() {
|
||||
@Override
|
||||
public void putIfAbsent(String key, byte[] input) {}
|
||||
|
||||
@Override
|
||||
public InputStream open(String key) {
|
||||
return open(key, () -> {});
|
||||
}
|
||||
|
||||
@Override
|
||||
public InputStream open(String key, Runnable guard) {
|
||||
return new ByteArrayInputStream(bytes) {
|
||||
@Override
|
||||
public synchronized int read(byte[] buffer, int offset, int length) {
|
||||
try {
|
||||
Thread.sleep(120);
|
||||
} catch (InterruptedException error) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new IllegalStateException(error);
|
||||
}
|
||||
guard.run();
|
||||
return super.read(buffer, offset, Math.min(length, 1));
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public void requireUnversioned() {}
|
||||
};
|
||||
ToolPublicationDataStore data = new ToolPublicationDataStore(fixture.jdbc, fixture.manager,
|
||||
publication.store(), publication.sessions(), bucket, Duration.ofSeconds(10), Duration.ofMillis(300),
|
||||
new ToolPublicationDataStore.VerificationBudget(16 * 1024 * 1024, Duration.ofMinutes(25)));
|
||||
data.publishSegment(binding.path("sessionKey"), "pub-1", PUBLICATION_TOKEN, "async-heartbeat",
|
||||
"stdout", 0, bytes, null, true);
|
||||
fixture.ledger.reset();
|
||||
try (var verifier = new ToolPublicationVerifier(data, 1)) {
|
||||
assertThat(verifier.runOnce()).isTrue();
|
||||
}
|
||||
long renewals = fixture.ledger.count("update qwen_tool_publication_operation set claim_until");
|
||||
assertThat(renewals).isGreaterThanOrEqualTo(2);
|
||||
assertThat(fixture.ledger.count("from qwen_managed_session_journal_tx")).isZero();
|
||||
assertThat(fixture.ledger.count("from qwen_managed_session_journal_head", "for update"))
|
||||
.isGreaterThanOrEqualTo(2 * (renewals + 1));
|
||||
fixture.ledger.reset();
|
||||
for (int i = 0; i < 12; i++) {
|
||||
assertThat(data.operationStatus(binding.path("sessionKey"), "pub-1", PUBLICATION_TOKEN, "async-heartbeat")
|
||||
.path("state").asText()).isEqualTo("SUCCEEDED");
|
||||
}
|
||||
assertThat(fixture.ledger.total()).isEqualTo(12);
|
||||
assertThat(fixture.ledger.count("from qwen_managed_session_journal_tx")).isZero();
|
||||
assertThat(fixture.ledger.count("from qwen_managed_session_journal_head")).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void frequentReadsDoNotRenewBeforeTheHeartbeatIntervalOrFetchInlineBytes() {
|
||||
Fixture fixture = new Fixture();
|
||||
PublicationFixture publication = publicationFixture(fixture);
|
||||
byte[] bytes = new byte[1024];
|
||||
var reads = new java.util.concurrent.atomic.AtomicInteger();
|
||||
ToolPublicationObjectStore bucket = new ToolPublicationObjectStore() {
|
||||
@Override public void putIfAbsent(String key, byte[] input) {}
|
||||
@Override public InputStream open(String key) { return open(key, () -> {}); }
|
||||
@Override public InputStream open(String key, Runnable guard) {
|
||||
return new ByteArrayInputStream(bytes) {
|
||||
@Override public synchronized int read(byte[] buffer, int offset, int length) {
|
||||
guard.run();
|
||||
reads.incrementAndGet();
|
||||
return super.read(buffer, offset, Math.min(length, 1));
|
||||
}
|
||||
};
|
||||
}
|
||||
@Override public void requireUnversioned() {}
|
||||
};
|
||||
// A five-minute lease gives a 100-second interval, well beyond this bounded read.
|
||||
ToolPublicationDataStore data = new ToolPublicationDataStore(fixture.jdbc, fixture.manager,
|
||||
publication.store(), publication.sessions(), bucket, Duration.ofMinutes(10), Duration.ofMinutes(5),
|
||||
new ToolPublicationDataStore.VerificationBudget(16 * 1024 * 1024, Duration.ofMinutes(25)));
|
||||
data.publishSegment(binding.path("sessionKey"), "pub-1", PUBLICATION_TOKEN, "async-fast-read",
|
||||
"stdout", 0, bytes, null, true);
|
||||
fixture.ledger.reset();
|
||||
try (var verifier = new ToolPublicationVerifier(data, 1)) {
|
||||
assertThat(verifier.runOnce()).isTrue();
|
||||
}
|
||||
assertThat(data.operationStatus(binding.path("sessionKey"), "pub-1", PUBLICATION_TOKEN, "async-fast-read")
|
||||
.path("state").asText()).isEqualTo("SUCCEEDED");
|
||||
assertThat(reads).hasValueGreaterThan(1000);
|
||||
assertThat(fixture.ledger.count("update qwen_tool_publication_operation set claim_until")).isZero();
|
||||
assertThat(fixture.ledger.count("select inline_bytes from qwen_tool_publication_object")).isZero();
|
||||
assertThat(fixture.ledger.count("from qwen_managed_session_journal_tx")).isZero();
|
||||
}
|
||||
|
||||
@Test
|
||||
void publicationAuthorizationReadsTheActivationFromTheJournalHead() {
|
||||
Fixture fixture = new Fixture();
|
||||
|
||||
+38
@@ -0,0 +1,38 @@
|
||||
package com.alibaba.qwen.code.managedagent.store;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
import javax.sql.DataSource;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.springframework.jdbc.core.JdbcTemplate;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
|
||||
class ToolPublicationAsyncVerificationMySqlIT extends ToolPublicationAsyncVerificationTest {
|
||||
private JdbcTemplate admin;
|
||||
private final List<String> schemas = new ArrayList<>();
|
||||
|
||||
@Override
|
||||
protected DataSource dataSource() {
|
||||
String url = System.getProperty("mysql.url");
|
||||
if (url == null || !url.matches("jdbc:mysql://[^/]+/[^?]+(?:\\?.*)?")) {
|
||||
throw new IllegalArgumentException("A MySQL test database URL is required");
|
||||
}
|
||||
String user = System.getProperty("mysql.user");
|
||||
String password = System.getProperty("mysql.password", "");
|
||||
admin = new JdbcTemplate(new DriverManagerDataSource(url, user, password));
|
||||
String schema = "publication_async_" + UUID.randomUUID().toString().replace("-", "");
|
||||
admin.execute("CREATE DATABASE " + schema);
|
||||
schemas.add(schema);
|
||||
return new DriverManagerDataSource(url.replaceFirst("/[^/?]+(?=\\?|$)", "/" + schema), user, password);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void removeSchema() {
|
||||
if (admin != null) {
|
||||
for (String schema : schemas) {
|
||||
admin.execute("DROP DATABASE IF EXISTS " + schema);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+1166
File diff suppressed because it is too large.
Load diff
+40
@@ -190,6 +190,46 @@ public class ToolPublicationCollectorTest extends ToolPublicationRetentionStoreT
|
||||
assertThat(objects.deleted).isEmpty();
|
||||
}
|
||||
|
||||
@ParameterizedTest
|
||||
@ValueSource(strings = {"VERIFIED", "CANDIDATE", "QUARANTINED"})
|
||||
void failedPrefixAllowsCollectionOnlyWithVerifiedObjects(String objectState) {
|
||||
jdbc.update("INSERT INTO qwen_tool_publication_operation"
|
||||
+ " (scope_key, publication_id, operation_id, request_digest, slot_key, state, claim_epoch,"
|
||||
+ " deadline, created_at, execution_mode, verification_ready, failure_status, failure_code)"
|
||||
+ " VALUES (?, 'pub-1', 'failed-prefix', ?, 'prefix:stdout', 'FAILED', 1,"
|
||||
+ " CURRENT_TIMESTAMP(6), CURRENT_TIMESTAMP(6), 'ASYNC', TRUE, 403, 'managed_tool_publication_storage_denied')",
|
||||
scope, "b".repeat(64));
|
||||
addObject("one", "failed-prefix/exact-key", null);
|
||||
addObject("inline", null, new byte[] {8});
|
||||
jdbc.update("UPDATE qwen_tool_publication_object SET state = ? WHERE scope_key = ? AND slot_key = 'one'",
|
||||
objectState, scope);
|
||||
var objects = new DeletingObjects();
|
||||
objects.bytes.put("failed-prefix/exact-key", new byte[] {7});
|
||||
retire();
|
||||
if (!"VERIFIED".equals(objectState)) {
|
||||
assertThat(blocker()).isEqualTo("object_unverified");
|
||||
assertThat(collector(objects).runOnce()).isFalse();
|
||||
assertThat(held()).isEqualTo(3000);
|
||||
assertThat(objects.deleted).isEmpty();
|
||||
assertThat(objects.bytes.get("failed-prefix/exact-key")).containsExactly((byte) 7);
|
||||
assertThat(jdbc.queryForObject("SELECT inline_bytes FROM qwen_tool_publication_object"
|
||||
+ " WHERE scope_key = ? AND slot_key = 'inline'", byte[].class, scope)).containsExactly((byte) 8);
|
||||
return;
|
||||
}
|
||||
assertThat(blocker()).isNull();
|
||||
assertThat(collector(objects).runOnce()).isTrue();
|
||||
assertThat(state()).isEqualTo("COLLECTED");
|
||||
assertThat(held()).isZero();
|
||||
assertThat(jdbc.queryForObject("SELECT capture_used_bytes + producer_used_bytes + admission_used_bytes"
|
||||
+ " FROM qwen_tool_publication WHERE scope_key = ?", Long.class, scope)).isZero();
|
||||
assertThat(objects.deleted).containsExactly("failed-prefix/exact-key");
|
||||
assertThat(objects.bytes).isEmpty();
|
||||
assertThat(jdbc.queryForList("SELECT inline_bytes FROM qwen_tool_publication_object WHERE scope_key = ?",
|
||||
byte[].class, scope)).containsOnly((byte[]) null);
|
||||
assertThat(jdbc.queryForObject("SELECT state FROM qwen_tool_publication_operation WHERE scope_key = ?",
|
||||
String.class, scope)).isEqualTo("FAILED");
|
||||
}
|
||||
|
||||
@Test
|
||||
void dueProtectedPublicationRechecksEvidenceInsteadOfTrustingItsStoredBlocker() {
|
||||
addObject("one", "protected/exact-key", null);
|
||||
|
||||
+14
-1
@@ -64,7 +64,7 @@ class WorkspaceMigrationMySqlIT {
|
||||
+ " WHERE installed_rank > ? AND success = TRUE ORDER BY installed_rank",
|
||||
String.class, lastRank)).containsExactly("48", "49", "50",
|
||||
"51", "52", "53", "54", "55", "56", "57", "60", "62", "64",
|
||||
"65", "66", "67");
|
||||
"65", "66", "67", "68");
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM managed_workspace_migration", Integer.class)).isZero();
|
||||
assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM qwen_runtime_storage_fence", Integer.class)).isZero();
|
||||
}
|
||||
@@ -267,6 +267,19 @@ class WorkspaceMigrationMySqlIT {
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void upgradesMain56WithoutChangingAppliedMigrations() {
|
||||
Flyway.configure().dataSource(data).locations("classpath:db/migration").target("56").load().migrate();
|
||||
var applied = jdbc.queryForList("SELECT * FROM flyway_schema_history ORDER BY installed_rank");
|
||||
int lastRank = jdbc.queryForObject("SELECT MAX(installed_rank) FROM flyway_schema_history", Integer.class);
|
||||
Flyway.configure().dataSource(data).locations("classpath:db/migration").load().migrate();
|
||||
assertThat(jdbc.queryForList("SELECT * FROM flyway_schema_history"
|
||||
+ " WHERE installed_rank <= ? ORDER BY installed_rank", lastRank)).isEqualTo(applied);
|
||||
assertThat(jdbc.queryForList("SELECT version FROM flyway_schema_history"
|
||||
+ " WHERE installed_rank > ? AND success = TRUE ORDER BY installed_rank",
|
||||
String.class, lastRank)).containsExactly("57", "60", "62", "64", "65", "66", "67", "68");
|
||||
}
|
||||
|
||||
private void seed(String tenant, String session, String state) {
|
||||
jdbc.update("INSERT INTO managed_agent_session (tenant_id, session_id, agent_id, status, created_at, updated_at,"
|
||||
+ " workspace_id, workspace_generation, workspace_storage_id, cwd_relative, context_config_ref,"
|
||||
|
||||
Reference in new issue
Block a user