chengmaandClaude Opus 5
7ad82b7795
补齐 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>
375 lines
12 KiB
Go
375 lines
12 KiB
Go
package service
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"cmautobuy/admin/model"
|
|
"cmautobuy/admin/repository"
|
|
)
|
|
|
|
// claimTask 走一遍完整的领取流程，让 task_claims 里留下记录。
|
|
func claimTask(t *testing.T, db *sql.DB, taskID, clientID string) {
|
|
t.Helper()
|
|
if err := RegisterClient(db, model.Client{ClientID: clientID}); err != nil {
|
|
t.Fatalf("注册客户端失败: %v", err)
|
|
}
|
|
task, err := ClaimNextTask(db, clientID, []string{"collect", "purchase"})
|
|
if err != nil {
|
|
t.Fatalf("领取失败: %v", err)
|
|
}
|
|
if task == nil || task.TaskID != taskID {
|
|
t.Fatalf("期望领到 %s，实际 %v", taskID, task)
|
|
}
|
|
}
|
|
|
|
func taskStatus(t *testing.T, db *sql.DB, taskID string) string {
|
|
t.Helper()
|
|
var s string
|
|
if err := db.QueryRow(`SELECT status FROM tasks WHERE task_id = ?`, taskID).Scan(&s); err != nil {
|
|
t.Fatalf("查询任务状态失败: %v", err)
|
|
}
|
|
return s
|
|
}
|
|
|
|
const resultBody = `{"task_version":1,"attempt_id":"a-1","result_type":"purchase",
|
|
"completed_at":"2026-08-06T08:03:00Z","pdd_data":{"schema_version":1,"goods":{"goods_id":"1"}}}`
|
|
|
|
// ── 正常路径 ───────────────────────────────────────────
|
|
|
|
func TestSubmitResult_成功落库并标记succeeded(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
resp, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody))
|
|
if err != nil {
|
|
t.Fatalf("提交失败: %v", err)
|
|
}
|
|
|
|
var got map[string]any
|
|
if err := json.Unmarshal([]byte(resp), &got); err != nil {
|
|
t.Fatalf("响应不是合法 JSON: %v", err)
|
|
}
|
|
if got["accepted"] != true {
|
|
t.Errorf("accepted 应为 true，实际 %v", got["accepted"])
|
|
}
|
|
if got["result_id"] == "" || got["result_id"] == nil {
|
|
t.Error("result_id 不应为空")
|
|
}
|
|
if s := taskStatus(t, db, "TASK-A"); s != "succeeded" {
|
|
t.Errorf("任务状态应为 succeeded，实际 %s", s)
|
|
}
|
|
}
|
|
|
|
// ── 幂等 ───────────────────────────────────────────────
|
|
|
|
func TestSubmitResult_同键同内容返回同一结果(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
first, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody))
|
|
if err != nil {
|
|
t.Fatalf("首次提交失败: %v", err)
|
|
}
|
|
second, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody))
|
|
if err != nil {
|
|
t.Fatalf("重复提交失败: %v", err)
|
|
}
|
|
|
|
if first != second {
|
|
t.Errorf("重复提交应返回完全相同的响应：\n第一次 %s\n第二次 %s", first, second)
|
|
}
|
|
|
|
// 而且不能重复落库
|
|
var n int
|
|
db.QueryRow(`SELECT COUNT(*) FROM idempotency_keys`).Scan(&n)
|
|
if n != 1 {
|
|
t.Errorf("幂等表应只有 1 条记录，实际 %d", n)
|
|
}
|
|
}
|
|
|
|
func TestSubmitResult_同键不同内容返回冲突(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
if _, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody)); err != nil {
|
|
t.Fatalf("首次提交失败: %v", err)
|
|
}
|
|
|
|
other := `{"task_version":1,"attempt_id":"a-1","result_type":"purchase","pdd_data":{"不一样":true}}`
|
|
_, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(other))
|
|
if !errors.Is(err, repository.ErrIdempotencyConflict) {
|
|
t.Errorf("期望幂等冲突，实际 %v", err)
|
|
}
|
|
}
|
|
|
|
// 并发重复提交：只能落库一次。
|
|
func TestSubmitResult_并发同键只处理一次(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
const workers = 6
|
|
var wg sync.WaitGroup
|
|
var mu sync.Mutex
|
|
responses := map[string]int{}
|
|
var lastErr error
|
|
|
|
for i := 0; i < workers; i++ {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
resp, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody))
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
if err != nil {
|
|
lastErr = err
|
|
return
|
|
}
|
|
responses[resp]++
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
|
|
if lastErr != nil {
|
|
t.Fatalf("并发提交出错: %v", lastErr)
|
|
}
|
|
if len(responses) != 1 {
|
|
t.Errorf("并发提交应返回同一个响应，实际出现 %d 种", len(responses))
|
|
}
|
|
var n int
|
|
db.QueryRow(`SELECT COUNT(*) FROM idempotency_keys`).Scan(&n)
|
|
if n != 1 {
|
|
t.Errorf("幂等表应只有 1 条记录，实际 %d", n)
|
|
}
|
|
}
|
|
|
|
// ── 无条件接受（本工单最要紧的三条）───────────────────
|
|
|
|
func TestSubmitResult_任务已取消仍然接受(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
// Admin 侧把任务取消了
|
|
if _, err := db.Exec(`UPDATE tasks SET status='cancelled' WHERE task_id='TASK-A'`); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// 客户端并不知道，照样提交——必须被接受，
|
|
// 因为它可能真的已经在拼多多下过单了，这些数据必须留痕
|
|
if _, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody)); err != nil {
|
|
t.Fatalf("任务已取消时提交被拒绝了，这违反契约 §4.1: %v", err)
|
|
}
|
|
if s := taskStatus(t, db, "TASK-A"); s != "succeeded" {
|
|
t.Errorf("结果应被记录，状态应为 succeeded，实际 %s", s)
|
|
}
|
|
}
|
|
|
|
func TestSubmitResult_任务已重派仍然接受原客户端的结果(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
// 操作员把任务改派给了另一台
|
|
if _, err := db.Exec(
|
|
`UPDATE tasks SET assigned_client='client-002', status='assigned' WHERE task_id='TASK-A'`,
|
|
); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
// 原来那台还是把结果交上来了 —— 必须接受
|
|
if _, err := SubmitResult(db, "TASK-A", "client-001", "key-1", []byte(resultBody)); err != nil {
|
|
t.Fatalf("任务已重派时拒绝了原客户端，这违反契约 §4.1: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestSubmitResult_同一任务接受多个客户端的多份结果(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
// 重派给第二台，让它也领一次
|
|
if _, err := db.Exec(
|
|
`UPDATE tasks SET assigned_client='client-002', status='assigned' WHERE task_id='TASK-A'`,
|
|
); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
claimTask(t, db, "TASK-A", "client-002")
|
|
|
|
// 两台都提交，用各自的幂等键
|
|
if _, err := SubmitResult(db, "TASK-A", "client-001", "key-c1", []byte(resultBody)); err != nil {
|
|
t.Fatalf("client-001 提交被拒: %v", err)
|
|
}
|
|
if _, err := SubmitResult(db, "TASK-A", "client-002", "key-c2", []byte(resultBody)); err != nil {
|
|
t.Fatalf("client-002 提交被拒: %v", err)
|
|
}
|
|
|
|
var n int
|
|
db.QueryRow(`SELECT COUNT(*) FROM idempotency_keys`).Scan(&n)
|
|
if n != 2 {
|
|
t.Errorf("两份结果应各留一条幂等记录，实际 %d", n)
|
|
}
|
|
}
|
|
|
|
// ── 唯一该拒绝的情况 ───────────────────────────────────
|
|
|
|
func TestSubmitResult_从没领过的客户端被拒绝(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
_, err := SubmitResult(db, "TASK-A", "client-999", "key-x", []byte(resultBody))
|
|
if !errors.Is(err, ErrNeverClaimed) {
|
|
t.Errorf("从没领过的客户端应被拒绝，实际 %v", err)
|
|
}
|
|
}
|
|
|
|
func TestSubmitResult_任务不存在(t *testing.T) {
|
|
db := newTestDB(t)
|
|
_, err := SubmitResult(db, "TASK-NOPE", "client-001", "key-x", []byte(resultBody))
|
|
if !errors.Is(err, ErrTaskNotFound) {
|
|
t.Errorf("期望任务不存在错误，实际 %v", err)
|
|
}
|
|
}
|
|
|
|
// ── 采集任务的结果要落到商品级 ─────────────────────────
|
|
|
|
func TestSubmitResult_采集结果写入商品级(t *testing.T) {
|
|
db := newTestDB(t)
|
|
now := model.NowISO()
|
|
|
|
if _, err := db.Exec(`
|
|
INSERT INTO shopee_products (goods_id, title, pdd_goods_url,
|
|
collect_status, created_at, updated_at)
|
|
VALUES ('G-1','测试商品','https://x/1','collecting',?,?)`, now, now); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := db.Exec(`
|
|
INSERT INTO tasks (task_id, task_type, status, assigned_client, goods_id,
|
|
pdd_goods_url, created_at, updated_at)
|
|
VALUES ('TASK-C','collect','assigned','client-001','G-1','https://x/1',?,?)`,
|
|
now, now); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
claimTask(t, db, "TASK-C", "client-001")
|
|
|
|
body := `{"task_version":1,"attempt_id":"a-1","result_type":"collect","pdd_data":{"schema_version":1,"skus":[]}}`
|
|
if _, err := SubmitResult(db, "TASK-C", "client-001", "key-c", []byte(body)); err != nil {
|
|
t.Fatalf("提交采集结果失败: %v", err)
|
|
}
|
|
|
|
var status, data string
|
|
if err := db.QueryRow(
|
|
`SELECT collect_status, COALESCE(pdd_data,'') FROM shopee_products WHERE goods_id='G-1'`,
|
|
).Scan(&status, &data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if status != "collected" {
|
|
t.Errorf("采集状态应为 collected，实际 %s", status)
|
|
}
|
|
if data == "" {
|
|
t.Error("pdd_data 应被写入商品级，实际为空")
|
|
}
|
|
}
|
|
|
|
// ── 提交失败 ───────────────────────────────────────────
|
|
|
|
func TestSubmitFailure_状态映射(t *testing.T) {
|
|
cases := []struct {
|
|
reported string
|
|
want string
|
|
}{
|
|
{"retry_wait", "assigned"}, // 放回去等它再来领
|
|
{"manual_review", "manual_review"},
|
|
{"failed", "failed"},
|
|
{"cancelled", "cancelled"},
|
|
}
|
|
for _, tc := range cases {
|
|
t.Run(tc.reported, func(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
body := `{"task_version":1,"attempt_id":"a-1","status":"` + tc.reported +
|
|
`","error":{"code":"PDD_PAGE_TIMEOUT","message":"页面超时"}}`
|
|
if _, err := SubmitFailure(db, "TASK-A", "client-001", "k", []byte(body)); err != nil {
|
|
t.Fatalf("提交失败结果出错: %v", err)
|
|
}
|
|
if s := taskStatus(t, db, "TASK-A"); s != tc.want {
|
|
t.Errorf("%s 应映射为 %s，实际 %s", tc.reported, tc.want, s)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSubmitFailure_非法状态被拒绝(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
body := `{"attempt_id":"a-1","status":"succeeded"}`
|
|
if _, err := SubmitFailure(db, "TASK-A", "client-001", "k", []byte(body)); err == nil {
|
|
t.Error("失败接口不该接受 succeeded 这种状态")
|
|
}
|
|
}
|
|
|
|
func TestSubmitFailure_采集失败写回商品级(t *testing.T) {
|
|
db := newTestDB(t)
|
|
now := model.NowISO()
|
|
db.Exec(`INSERT INTO shopee_products (goods_id,title,collect_status,created_at,updated_at)
|
|
VALUES ('G-1','测试商品','collecting',?,?)`, now, now)
|
|
db.Exec(`INSERT INTO tasks (task_id,task_type,status,assigned_client,goods_id,
|
|
pdd_goods_url,created_at,updated_at)
|
|
VALUES ('TASK-C','collect','assigned','client-001','G-1','https://x/1',?,?)`, now, now)
|
|
claimTask(t, db, "TASK-C", "client-001")
|
|
|
|
body := `{"attempt_id":"a-1","status":"failed",
|
|
"error":{"code":"PDD_PAGE_TIMEOUT","message":"商品页加载超时"}}`
|
|
if _, err := SubmitFailure(db, "TASK-C", "client-001", "k", []byte(body)); err != nil {
|
|
t.Fatalf("提交失败: %v", err)
|
|
}
|
|
|
|
var status, errMsg string
|
|
db.QueryRow(`SELECT collect_status, COALESCE(collect_error,'')
|
|
FROM shopee_products WHERE goods_id='G-1'`).Scan(&status, &errMsg)
|
|
if status != "failed" {
|
|
t.Errorf("采集状态应为 failed，实际 %s", status)
|
|
}
|
|
if errMsg == "" {
|
|
t.Error("失败原因应写入 collect_error，否则操作员看不到为什么采不到")
|
|
}
|
|
}
|
|
|
|
// ── 活动时间 ───────────────────────────────────────────
|
|
|
|
func TestSubmit_刷新客户端活动时间(t *testing.T) {
|
|
db := newTestDB(t)
|
|
insertTask(t, db, "TASK-A", "client-001")
|
|
claimTask(t, db, "TASK-A", "client-001")
|
|
|
|
// 把客户端改成很久没活动
|
|
db.Exec(`UPDATE clients SET last_seen_at='2020-01-01T00:00:00Z' WHERE client_id='client-001'`)
|
|
views, _ := ListClientViews(db, "", 10*time.Minute)
|
|
if views[0].Status != "离线" {
|
|
t.Fatalf("前置条件不对，应为离线")
|
|
}
|
|
|
|
if _, err := SubmitResult(db, "TASK-A", "client-001", "k", []byte(resultBody)); err != nil {
|
|
t.Fatalf("提交失败: %v", err)
|
|
}
|
|
|
|
views, _ = ListClientViews(db, "", 10*time.Minute)
|
|
if views[0].Status != "在线" {
|
|
t.Error("提交结果后应刷新活动时间——只在 claim 里刷新的话，" +
|
|
"客户端执行长任务期间会被误判成离线")
|
|
}
|
|
}
|