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>
This commit is contained in:
chengma
@@ -49,9 +49,9 @@ func UpsertClient(db *sql.DB, c model.Client) error {
|
||||
//
|
||||
// claim / result / failure 三个接口都要调。只在 claim 里调的话，
|
||||
// 客户端执行长任务期间不调 claim，会被误判成离线。
|
||||
func TouchClient(db *sql.DB, clientID string) error {
|
||||
func TouchClient(q Execer, clientID string) error {
|
||||
now := model.NowISO()
|
||||
_, err := db.Exec(
|
||||
_, err := q.Exec(
|
||||
`UPDATE clients SET last_seen_at = ?, updated_at = ? WHERE client_id = ?`,
|
||||
now, now, clientID)
|
||||
if err != nil {
|
||||
|
||||
+49
-1
@@ -20,6 +20,17 @@ import (
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
// Execer 让同一个 repository 函数既能直接用 *sql.DB，
|
||||
// 也能在事务里用 *sql.Tx。
|
||||
//
|
||||
// 需要"多张表要么一起改、要么都不改"时，service 开一个事务，
|
||||
// 把 *sql.Tx 传进来即可，不用为事务再写一套函数。
|
||||
type Execer interface {
|
||||
Exec(query string, args ...any) (sql.Result, error)
|
||||
Query(query string, args ...any) (*sql.Rows, error)
|
||||
QueryRow(query string, args ...any) *sql.Row
|
||||
}
|
||||
|
||||
// Open 打开 data/admin.db。
|
||||
//
|
||||
// **PRAGMA 必须写在 DSN 里，不能用 db.Exec("PRAGMA ...") 设置。**
|
||||
@@ -37,10 +48,13 @@ func Open(dataDir string) (*sql.DB, error) {
|
||||
// busy_timeout 拿不到锁时最多等 5 秒，而不是立刻报错
|
||||
// journal_mode WAL 模式，读和写可以同时进行
|
||||
// foreign_keys 打开外键约束（SQLite 默认是关的）
|
||||
//
|
||||
// _txlock=immediate 是另一个**必须加**的参数，原因见下。
|
||||
dsn := "file:" + path +
|
||||
"?_pragma=busy_timeout(5000)" +
|
||||
"&_pragma=journal_mode(WAL)" +
|
||||
"&_pragma=foreign_keys(1)"
|
||||
"&_pragma=foreign_keys(1)" +
|
||||
"&_txlock=immediate"
|
||||
|
||||
db, err := sql.Open("sqlite", dsn)
|
||||
if err != nil {
|
||||
@@ -55,6 +69,21 @@ func Open(dataDir string) (*sql.DB, error) {
|
||||
db.SetMaxOpenConns(4)
|
||||
db.SetMaxIdleConns(4)
|
||||
|
||||
// 关于 _txlock=immediate：
|
||||
//
|
||||
// Go 的 db.Begin() 默认发的是 BEGIN DEFERRED——事务开始时**不拿写锁**，
|
||||
// 等到第一次写才去拿。于是多个事务可以同时开始、各自先读，
|
||||
// 然后同时想升级成写，互相卡死，直接报 SQLITE_BUSY。
|
||||
// 这种情况 busy_timeout **救不了**：等下去也不可能有结果，
|
||||
// 只能让某个事务整个重来。
|
||||
//
|
||||
// 加上 _txlock=immediate 后，事务一开始就拿写锁，
|
||||
// 拿不到就按 busy_timeout 排队等——这才是我们要的行为。
|
||||
//
|
||||
// 实测（6 个并发事务，每个先读后写）：
|
||||
// 默认 deferred 失败 5/6
|
||||
// _txlock=immediate 失败 0/6
|
||||
|
||||
// sql.Open 是懒加载的，这里主动连一次，好让配置错误立刻暴露
|
||||
if err := db.Ping(); err != nil {
|
||||
db.Close()
|
||||
@@ -192,6 +221,25 @@ var migrations = [][]string{
|
||||
created_at TEXT NOT NULL
|
||||
);`,
|
||||
},
|
||||
|
||||
// v2: 领取历史。
|
||||
//
|
||||
// 为什么需要它：契约要求"只有**从未分配给该客户端**的任务才返回 403"
|
||||
// （docs/admin/04-client-api.md §4.1）。但 tasks.assigned_client 只记
|
||||
// **当前**归属，任务一旦重派给别人，就查不出原来那台领过——
|
||||
// 而契约又明确要求"已重派仍要接受原客户端提交的结果"。
|
||||
// 没有这张表，那条规则根本没法判断。
|
||||
//
|
||||
// 顺带得到一份审计记录：这个任务被哪几台客户端领过。
|
||||
{
|
||||
`CREATE TABLE task_claims (
|
||||
task_id TEXT NOT NULL,
|
||||
client_id TEXT NOT NULL,
|
||||
claimed_at TEXT NOT NULL,
|
||||
PRIMARY KEY (task_id, client_id)
|
||||
);`,
|
||||
`CREATE INDEX idx_task_claims_client ON task_claims(client_id);`,
|
||||
},
|
||||
}
|
||||
|
||||
// Migrate 把数据库升到最新版本。
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
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
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package repository
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"cmautobuy/admin/model"
|
||||
)
|
||||
|
||||
// SetCollectResult 把采集回来的 PDD 商品数据存到商品级。
|
||||
//
|
||||
// `[必须]` pdd_data 存在**商品级**（shopee_products），不是订单级——
|
||||
// 一个 PDD 商品采一次，所有相关订单共用这份结果。
|
||||
// 见 docs/admin/03-data-model.md §3.1。
|
||||
func SetCollectResult(q Execer, goodsID, pddData string) error {
|
||||
if goodsID == "" {
|
||||
return nil // 任务没关联蝦皮商品（比如手工造的测试任务），跳过
|
||||
}
|
||||
now := model.NowISO()
|
||||
_, err := q.Exec(`
|
||||
UPDATE shopee_products
|
||||
SET pdd_data = ?, collect_status = 'collected',
|
||||
collect_error = NULL, collected_at = ?, updated_at = ?
|
||||
WHERE goods_id = ?`,
|
||||
pddData, now, now, goodsID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("保存商品 %s 的采集结果失败: %w", goodsID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetCollectFailed 标记采集失败，并记下原因。
|
||||
//
|
||||
// 错误信息要能在界面上看见，否则操作员不知道为什么采不到。
|
||||
func SetCollectFailed(q Execer, goodsID, errMsg string) error {
|
||||
if goodsID == "" {
|
||||
return nil
|
||||
}
|
||||
now := model.NowISO()
|
||||
_, err := q.Exec(`
|
||||
UPDATE shopee_products
|
||||
SET collect_status = 'failed', collect_error = ?, updated_at = ?
|
||||
WHERE goods_id = ?`,
|
||||
errMsg, now, goodsID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("标记商品 %s 采集失败出错: %w", goodsID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -74,6 +74,11 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model
|
||||
if n == 0 {
|
||||
continue // 被别的客户端抢先了，换下一条
|
||||
}
|
||||
// 记一笔领取历史。提交结果时要靠它判断这台客户端有没有领过——
|
||||
// 任务重派后 assigned_client 会变，只看它就查不出来了。
|
||||
if err := RecordClaim(db, taskID, clientID, now); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return GetTask(db, taskID)
|
||||
}
|
||||
return nil, nil // 没有可领的任务
|
||||
@@ -125,3 +130,101 @@ func GetTask(db *sql.DB, taskID string) (*model.Task, error) {
|
||||
t.FinishedAt = finishedAt.String
|
||||
return &t, nil
|
||||
}
|
||||
|
||||
// RecordClaim 记一笔"某客户端领过某任务"。
|
||||
//
|
||||
// 同一台客户端重复领同一个任务时只更新时间，不报错。
|
||||
func RecordClaim(q Execer, taskID, clientID, claimedAt string) error {
|
||||
_, err := q.Exec(`
|
||||
INSERT INTO task_claims (task_id, client_id, claimed_at)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(task_id, client_id) DO UPDATE SET claimed_at = excluded.claimed_at`,
|
||||
taskID, clientID, claimedAt)
|
||||
if err != nil {
|
||||
return fmt.Errorf("记录领取历史失败 task=%s client=%s: %w", taskID, clientID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// HasEverClaimed 判断这台客户端**曾经**领过这个任务。
|
||||
//
|
||||
// 用于提交结果时的权限判断：只有从没领过的才拒绝（403）。
|
||||
// 任务已取消、已重派给别人，都**不影响**这个判断——
|
||||
// 契约要求那些情况仍然要接受结果，见 docs/admin/04-client-api.md §4.1。
|
||||
func HasEverClaimed(q Execer, taskID, clientID string) (bool, error) {
|
||||
var one int
|
||||
err := q.QueryRow(
|
||||
`SELECT 1 FROM task_claims WHERE task_id = ? AND client_id = ?`,
|
||||
taskID, clientID).Scan(&one)
|
||||
if err == sql.ErrNoRows {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("查询领取历史失败: %w", err)
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// TaskExists 判断任务在不在，并返回它的类型和关联的蝦皮商品编号。
|
||||
func TaskExists(q Execer, taskID string) (exists bool, taskType model.TaskType, goodsID string, err error) {
|
||||
var gid sql.NullString
|
||||
e := q.QueryRow(
|
||||
`SELECT task_type, goods_id FROM tasks WHERE task_id = ?`,
|
||||
taskID).Scan(&taskType, &gid)
|
||||
if e == sql.ErrNoRows {
|
||||
return false, "", "", nil
|
||||
}
|
||||
if e != nil {
|
||||
return false, "", "", fmt.Errorf("查询任务 %s 失败: %w", taskID, e)
|
||||
}
|
||||
return true, taskType, gid.String, nil
|
||||
}
|
||||
|
||||
// MarkTaskSucceeded 记录成功结果。
|
||||
//
|
||||
// `[必须]` **不检查任务当前状态**。任务已取消、已重派给别的客户端，
|
||||
// 都照样接受——客户端可能真的已经下单了，这些数据必须能留痕。
|
||||
// 理由见 docs/admin/04-client-api.md §4.1。
|
||||
func MarkTaskSucceeded(q Execer, taskID, resultData string) error {
|
||||
now := model.NowISO()
|
||||
_, err := q.Exec(`
|
||||
UPDATE tasks
|
||||
SET status = 'succeeded', result_data = ?, finished_at = ?, updated_at = ?
|
||||
WHERE task_id = ?`,
|
||||
resultData, now, now, taskID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("标记任务 %s 成功失败: %w", taskID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// MarkTaskFailure 记录失败/需人工处理的结果。
|
||||
//
|
||||
// status 由客户端报告，映射规则见 docs/admin/04-client-api.md §5：
|
||||
//
|
||||
// retry_wait -> assigned（放回去等它再来领）
|
||||
// manual_review -> manual_review
|
||||
// failed -> failed
|
||||
// cancelled -> cancelled
|
||||
//
|
||||
// 同样**不检查任务当前状态**，理由同 MarkTaskSucceeded。
|
||||
func MarkTaskFailure(q Execer, taskID string, newStatus model.TaskStatus, errCode, errMsg string) error {
|
||||
now := model.NowISO()
|
||||
|
||||
// 放回待领取的话不算结束，finished_at 保持为空
|
||||
finishedAt := any(now)
|
||||
if newStatus == model.TaskAssigned {
|
||||
finishedAt = nil
|
||||
}
|
||||
|
||||
_, err := q.Exec(`
|
||||
UPDATE tasks
|
||||
SET status = ?, error_code = ?, error_message = ?,
|
||||
finished_at = ?, updated_at = ?
|
||||
WHERE task_id = ?`,
|
||||
newStatus, errCode, errMsg, finishedAt, now, taskID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("标记任务 %s 失败状态出错: %w", taskID, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user