fix(managed-agent): lock the head revision read in AgentDefinition update

update() decided "content equals the latest revision, so store nothing"
from a non-locking latest() read. The command lookup before it opens the
transaction's snapshot, so a revision committed in between is invisible:
a write whose content matched the stale head inserted nothing, neither
DuplicateKeyException branch could fire, and the route answered 202 with
a revision that was no longer the head. The command row then pinned that
Idempotency-Key to the stale revision, so every retry replayed it and the
write could never land.

Take the head read FOR UPDATE so the digest comparison and the revision
allocation both see the committed head, and concurrent updates of one
definition serialize. get() keeps the non-locking latest(), so reads do
not contend with writers. The SQL shape is the module's own:
ManagedToolResultStore.java:116 issues the same ORDER BY ... LIMIT 1 FOR
UPDATE, exercised on H2 by ToolPublicationStoreTest.backfillOnePage.

The published updateAgent description and the design doc's concurrency
bullet both described the revision-number collision this replaces, so
both now state the serialized behavior. The 409 branch stays as a
fallback; idempotency_conflict is unchanged.
This commit is contained in:
易良 committed 2026-10-02 19:22:54 +08:00
1 parent b2d568dc21
commit 7d7cc0e414
4 files changed
+29 -12

No files matched your search

@@ -80,12 +80,12 @@ request digest also covers the operation and, for an update, the agent ID.
including a create key reused for an update, answers
`409 idempotency_conflict`.
- **Concurrency.** Revision rows and the command commit in one transaction.
When a concurrent request commits first, the losing write rolls back and
reads the command committed under its key: the same request replays that
result. Otherwise a concurrent update that took the revision number answers
`409 agent_revision_conflict` without recording the command, so its key can
be retried, and a different request under the same key answers
`409 idempotency_conflict`.
An update reads the head revision under a row lock, so concurrent updates of
one definition serialize and each stores its own revision, and the digest
comparison sees the committed head rather than a stale snapshot. When a
concurrent request commits first, the losing write rolls back and reads the
command committed under its key: the same request replays that result, and a
different request under the same key answers `409 idempotency_conflict`.
- **Get** reads the latest revision, or the one named by `revision` (a decimal
number from `1`). Anything else answers `404 agent_not_found`.
- The response carries `id`, `object: "agent"`, `revision`, `digest`,
@@ -67,10 +67,11 @@ Workspace 路由;而某个方法没有字面兄弟路由的保留名(如 `PO
- **回放。** 相同键与相同请求 digest 应答已记录的 revision,并带
`X-Qwen-Idempotent-Replay: true`。相同键配不同请求(包括把创建用过的键用于
更新)应答 `409 idempotency_conflict`。
- **并发。** revision 行与命令在同一事务中提交。并发请求先提交时,落败的写入
回滚,并读取该键下已提交的命令:相同请求回放其结果。否则,若并发更新先占用了
revision 号,应答 `409 agent_revision_conflict` 且不记录命令,可以用原键重试;
同一键下的不同请求应答 `409 idempotency_conflict`。
- **并发。** revision 行与命令在同一事务中提交。更新以行锁读取头部 revision,
因此同一定义的并发更新串行执行、各自存储自己的 revision,摘要比较读到的是
已提交的头部而不是过期快照。并发请求先提交时,落败的写入回滚,并读取该键
下已提交的命令:相同请求回放其结果,同一键下的不同请求应答
`409 idempotency_conflict`。
- **读取** 读取最新 revision,或 `revision` 指定的那个(从 `1` 开始的十进制数)。
其他值一律应答 `404 agent_not_found`。
- 响应包含 `id`、`object: "agent"`、`revision`、`digest`、`created_at` 以及请求的
@@ -79,7 +79,7 @@ public class ManagedAgentDefinitionStore {
if (replay.isPresent()) {
return replay.get();
}
DefinitionRevision latest = latest(tenantId, agentId)
DefinitionRevision latest = latestForUpdate(tenantId, agentId)
.orElseThrow(ManagedAgentDefinitionStore::notFound);
DefinitionRevision result = latest;
if (!latest.digest().equals(digest)) {
@@ -107,6 +107,22 @@ public class ManagedAgentDefinitionStore {
ManagedAgentDefinitionStore::row, tenantId, agentId));
}
/**
* The head read of {@link #update}, under a row lock. The digest
* comparison must see the committed head, not this transaction's
* snapshot: on a plain read a concurrent update can commit in between,
* so content that matches the stale head stores nothing and still
* answers 202 with a revision that is no longer the head.
*/
private Optional<DefinitionRevision> latestForUpdate(String tenantId,
String agentId) {
return first(jdbc.query("SELECT agent_id, revision, digest,"
+ " definition_json, created_at FROM managed_agent_definition"
+ " WHERE tenant_id = ? AND agent_id = ?"
+ " ORDER BY revision DESC LIMIT 1 FOR UPDATE",
ManagedAgentDefinitionStore::row, tenantId, agentId));
}
public Optional<DefinitionRevision> find(String tenantId, String agentId,
long revision) {
return first(jdbc.query("SELECT agent_id, revision, digest,"
@@ -128,7 +128,7 @@
"post": {
"tags": ["Public Resources"],
"operationId": "updateAgent",
"description": "Stores the next immutable revision. Content whose digest equals the latest revision stores nothing and answers with that revision. An identical concurrent request replays the committed result; a different concurrent update answers 409 agent_revision_conflict, and its Idempotency-Key can be retried.",
"description": "Stores the next immutable revision. Content whose digest equals the latest revision stores nothing and answers with that revision. Concurrent updates of one definition serialize on its head revision, so each stores its own revision. An identical concurrent request replays the committed result; the same Idempotency-Key with a different request answers 409 idempotency_conflict.",
"x-qwen-implementation-status": "implemented",
"parameters": [
{