Files
cmautobuy/admin/repository/idempotency.go
T
chengmaandClaude Opus 5 7ad82b7795 feat: 实现结果提交接口与幂等处理
补齐 submit_result / submit_failure,Admin 侧的三个接口全部可用,
Client 的完整一圈(领取 → 执行 → 提交)现在能走通了。

实现
- 幂等:idempotency_keys 表。同键同内容返回上次的响应且不重复落库,
  同键不同内容返回 409。幂等记录与业务写入在**同一事务**,
  分开写的话业务成功但幂等没记上,重试会被重复处理
- 无条件接受(契约 §4.1,最容易写错的一条):
  任务已取消、已重派给别人,都照样接受结果——客户端中途不查任务状态,
  必然会提交"Admin 这边已经不要了"的结果,而它可能真的已经下过单,
  这些数据必须留痕
- 采集任务的结果落到商品级 shopee_products.pdd_data 并置 collected;
  失败则置 failed 并把原因写进 collect_error,操作员才看得见
- 失败状态映射:retry_wait→assigned,其余同名
- 三个接口都刷新 last_seen_at

新增 task_claims 表(migrations v2)
契约要求"只有从未分配给该客户端的任务才返回 403",但 assigned_client
只记当前归属,重派后就查不出原来那台领过——而契约又要求那种情况必须接受。
没有这张表这条规则根本没法判断。顺带得到一份审计记录。

修复第二个并发 bug:事务必须 BEGIN IMMEDIATE
并发提交报 SQLITE_BUSY。根因是 Go 的 db.Begin() 默认发 BEGIN DEFERRED,
事务开始时不拿写锁,多个事务各自先读再想升级成写就互相卡死,
这种情况 busy_timeout 救不了。DSN 加 _txlock=immediate 后事务一开始
就排队拿锁。实测 6 个并发事务:默认失败 5/6,加参数后 0/6。
已写进 docs/admin/03-data-model.md §2.1。

已验证(Go 1.23.0)
- 30 个单元测试全过,并发用例重复 20 次稳定通过
- 端到端:claim 200 → 提交 200 → 重复提交返回完全相同的响应 →
  同键不同内容 409 → 没领过的客户端 403 → 任务不存在 404 →
  缺 Idempotency-Key 400;库里 task=succeeded、幂等 1 条、领取历史 1 条

说明:Gitea 尚未配置,本次无对应工单号。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 17:04:49 +08:00

63 lines
2.0 KiB
Go

package repository
import (
"crypto/sha256"
"database/sql"
"encoding/hex"
"errors"
"fmt"
"cmautobuy/admin/model"
)
// ErrIdempotencyConflict 表示同一个键提交了不同的内容。
//
// 这说明客户端弄错了——同一个键必须对应同一份内容。
// 内容真的变了,应该用新的 attempt_id 生成新的键。
var ErrIdempotencyConflict = errors.New("相同幂等键提交了不同内容")
// HashRequest 计算请求体的哈希,用来判断"同一个键"配的是不是"同一份内容"。
func HashRequest(body []byte) string {
sum := sha256.Sum256(body)
return hex.EncodeToString(sum[:])
}
// LookupIdempotent 查这个键是不是已经处理过。
//
// 三种结果:
// - 处理过且内容一致 → 返回上次的响应体和 true,**调用方直接原样返回,不要重复落库**
// - 处理过但内容不同 → 返回 ErrIdempotencyConflict,调用方回 409
// - 没处理过 → 返回 ("", false, nil),调用方正常处理
func LookupIdempotent(q Execer, key, requestHash string) (string, bool, error) {
var storedHash, storedBody string
err := q.QueryRow(
`SELECT request_hash, response_body FROM idempotency_keys WHERE key = ?`,
key).Scan(&storedHash, &storedBody)
if err == sql.ErrNoRows {
return "", false, nil
}
if err != nil {
return "", false, fmt.Errorf("查询幂等键失败: %w", err)
}
if storedHash != requestHash {
return "", false, ErrIdempotencyConflict
}
return storedBody, true, nil
}
// SaveIdempotent 记下这个键处理过了,以及当时返回了什么。
//
// `[必须]` 必须和业务写入在**同一个事务**里。
// 分开写的话,业务写成功但幂等记录没写上,客户端重试就会被重复处理。
func SaveIdempotent(q Execer, key, requestHash, responseBody string) error {
_, err := q.Exec(
`INSERT INTO idempotency_keys (key, request_hash, response_body, created_at)
VALUES (?, ?, ?, ?)`,
key, requestHash, responseBody, model.NowISO())
if err != nil {
return fmt.Errorf("保存幂等键失败: %w", err)
}
return nil
}