feat: claim 支持领取无主任务 (#17)

PDD 商品页要能「创建采集任务」而不指定客户端——采集只是浏览商品页，没有副作用，哪台设备采都一样。但原来的查询是 WHERE assigned_client = ? AND status = 'assigned' 无主任务（assigned_client 为空 + pending）永远没人能领，建出来就是死的。改动 - ClaimNextTask 同时查两种：指定给本机的 + 无主的 - 排序 ORDER BY (assigned_client IS NULL), priority DESC, created_at ——指定给本机的优先。显式分配是人为决定，应当先兑现 - 原子更新两种情况合成一条语句：对"指定给我的"写 assigned_client 是写同一个值无副作用；对无主的，这一步就是"谁领到就标记谁" - 表结构不用动（assigned_client 本来可空，status 已有 pending）推翻了一条已定案的规则 Client 契约 §5.1 原写「Admin 只把任务分配给指定的 Client」，现改为两种并存并说明各自适用场景： - 采集任务不指定客户端 - 采购任务可指定可留空。涉及钱和账号——不同设备可能登着不同的拼多多账号，需要指定账号时必须显式分配，留空即接受"谁先抢到谁下单" Client 侧对两种没有区别，不需要知道任务原来有没有主。已验证（Go 1.23.0） - 新增 7 个测试，全量 62 个全过 - 并发抢占用例重复 20 次稳定：8 个客户端抢同一条无主任务，正好 1 个拿到，且 assigned_client 记的就是那个赢家 - 既有测试未受影响，"只领分配给自己的"仍然成立一处仍未解决的风险（已记入 #17 风险表） tasks 表没有字段标记"该任务需要真实下单"，所以契约里 "不向 dry_run 客户端分配真实下单任务"实际无法执行。真实下单开关关闭时不出问题，开启前必须补该字段。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-07 10:53:32 +08:00
parent ab988b39de
commit dd387e3ee5
4 changed files with 264 additions and 14 deletions
@@ -16,8 +16,21 @@ const claimCandidateLimit = 10
 //
 // 没有可领的任务时返回 (nil, nil) —— 调用方据此返回 204。
 //
-// 防并发的做法是**条件更新 + 检查影响行数**：先查出候选，
-// 再用 `WHERE task_id = ? AND status = 'assigned'` 去更新，
+// # 两种任务都能领
+//
+//	指定给本机的  assigned_client = 我   且 status = 'assigned'
+//	无主的        assigned_client 为空   且 status = 'pending'
+//
+// 采集任务创建时不指定客户端（浏览商品页没有副作用，哪台设备采都一样），
+// 所以必须支持第二种，否则那些任务永远没人能领。
+// 采购任务可以指定也可以留空——涉及钱和账号时应当显式分配。
+//
+// **指定给本机的优先。** 显式分配是人为决定，应当先兑现；
+// 无主任务谁抢都一样，可以等。
+//
+// # 防并发
+//
+// 做法是**条件更新 + 检查影响行数**：先查出候选，再带着状态条件去更新，
 // 影响行数为 0 就说明被别人抢先了，换下一条。
 // 不用 SELECT ... FOR UPDATE，SQLite 没有那个。
 func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model.Task, error) {
@@ -26,7 +39,8 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model
 	}

 	query := `SELECT task_id FROM tasks
-	          WHERE assigned_client = ? AND status = 'assigned'`
+	          WHERE (   (assigned_client = ?    AND status = 'assigned')
+	                 OR (assigned_client IS NULL AND status = 'pending') )`
 	args := []any{clientID}

 	// 客户端只声明支持某些类型时，不要给它别的类型
@@ -37,7 +51,8 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model
 			args = append(args, t)
 		}
 	}
-	query += ` ORDER BY priority DESC, created_at LIMIT ?`
+	// (assigned_client IS NULL) 为 0/1，0 排前面 —— 指定给本机的优先于无主的
+	query += ` ORDER BY (assigned_client IS NULL), priority DESC, created_at LIMIT ?`
 	args = append(args, claimCandidateLimit)

 	rows, err := db.Query(query, args...)
@@ -60,10 +75,16 @@ func ClaimNextTask(db *sql.DB, clientID string, supportedTypes []string) (*model

 	now := model.NowISO()
 	for _, taskID := range candidates {
+		// 两种情况合成一条语句：对"指定给我的"那种，写 assigned_client
+		// 是写同一个值，无副作用；对无主的，这一步就是"谁领到就标记谁"。
 		res, err := db.Exec(`
-			UPDATE tasks SET status = 'claimed', claimed_at = ?, updated_at = ?
-			WHERE task_id = ? AND status = 'assigned'`,
-			now, now, taskID)
+			UPDATE tasks
+			   SET status = 'claimed', assigned_client = ?,
+			       claimed_at = ?, updated_at = ?
+			 WHERE task_id = ?
+			   AND (   (status = 'assigned' AND assigned_client = ?)
+			        OR (status = 'pending'  AND assigned_client IS NULL) )`,
+			clientID, now, now, taskID, clientID)
 		if err != nil {
 			return nil, fmt.Errorf("领取任务 %s 失败: %w", taskID, err)
 		}
@@ -0,0 +1,198 @@
+package service
+
+import (
+	"database/sql"
+	"sync"
+	"testing"
+
+	"cmautobuy/admin/model"
+)
+
+// insertUnassignedTask 插一条**无主**任务：没有指定客户端，状态是 pending。
+// 采集任务就长这样——创建时不指定客户端，谁领到算谁的。
+func insertUnassignedTask(t *testing.T, db *sql.DB, taskID, taskType string) {
+	t.Helper()
+	now := model.NowISO()
+	_, err := db.Exec(`
+		INSERT INTO tasks (task_id, task_type, status, assigned_client,
+		                   pdd_goods_url, pdd_goods_id, created_at, updated_at)
+		VALUES (?, ?, 'pending', NULL, 'https://x/1', '1', ?, ?)`,
+		taskID, taskType, now, now)
+	if err != nil {
+		t.Fatalf("插入无主任务失败: %v", err)
+	}
+}
+
+func taskAssignee(t *testing.T, db *sql.DB, taskID string) (client, status string) {
+	t.Helper()
+	var c sql.NullString
+	if err := db.QueryRow(
+		`SELECT assigned_client, status FROM tasks WHERE task_id = ?`, taskID,
+	).Scan(&c, &status); err != nil {
+		t.Fatalf("查询任务失败: %v", err)
+	}
+	return c.String, status
+}
+
+// ── 无主任务能被领到 ───────────────────────────────────
+
+func TestClaim_无主任务能被领到并标记领取者(t *testing.T) {
+	db := newTestDB(t)
+	insertUnassignedTask(t, db, "T-FREE", "collect")
+
+	task, err := ClaimNextTask(db, "client-001", []string{"collect"})
+	if err != nil {
+		t.Fatalf("领取失败: %v", err)
+	}
+	if task == nil {
+		t.Fatal("无主任务应该能被领到——不然采集任务建出来就是死的")
+	}
+
+	client, status := taskAssignee(t, db, "T-FREE")
+	if client != "client-001" {
+		t.Errorf("领取后应把领取者写进 assigned_client，实际 %q", client)
+	}
+	if status != "claimed" {
+		t.Errorf("状态应为 claimed，实际 %s", status)
+	}
+	if task.ClaimedAt == "" {
+		t.Error("claimed_at 应该有值")
+	}
+}
+
+func TestClaim_无主任务只能被领一次(t *testing.T) {
+	db := newTestDB(t)
+	insertUnassignedTask(t, db, "T-FREE", "collect")
+
+	if first, _ := ClaimNextTask(db, "client-001", []string{"collect"}); first == nil {
+		t.Fatal("第一次应该领到")
+	}
+	second, err := ClaimNextTask(db, "client-002", []string{"collect"})
+	if err != nil {
+		t.Fatalf("第二次领取报错: %v", err)
+	}
+	if second != nil {
+		t.Errorf("已被领走的任务不该再被领到：%s", second.TaskID)
+	}
+}
+
+// ── 优先级：指定给我的优先于无主的 ─────────────────────
+
+func TestClaim_指定给本机的优先于无主的(t *testing.T) {
+	db := newTestDB(t)
+	// 先插无主的，让它 created_at 更早——如果没有优先级规则，
+	// 按 created_at 排序会先拿到它，测试就能发现问题
+	insertUnassignedTask(t, db, "T-FREE", "purchase")
+	insertTask(t, db, "T-MINE", "client-001")
+
+	task, err := ClaimNextTask(db, "client-001", []string{"purchase"})
+	if err != nil {
+		t.Fatalf("领取失败: %v", err)
+	}
+	if task == nil {
+		t.Fatal("应该领到任务")
+	}
+	if task.TaskID != "T-MINE" {
+		t.Errorf("应优先领取指定给本机的 T-MINE，实际领到 %s —— "+
+			"显式分配是人为决定，应当先兑现", task.TaskID)
+	}
+}
+
+func TestClaim_指定给别人的仍然领不到(t *testing.T) {
+	db := newTestDB(t)
+	insertTask(t, db, "T-OTHER", "client-999")
+
+	task, err := ClaimNextTask(db, "client-001", []string{"purchase"})
+	if err != nil {
+		t.Fatalf("领取失败: %v", err)
+	}
+	if task != nil {
+		t.Errorf("不该领到指定给别的客户端的任务：%s", task.TaskID)
+	}
+}
+
+// ── 类型过滤仍然生效 ───────────────────────────────────
+
+func TestClaim_无主任务也受supported_types约束(t *testing.T) {
+	db := newTestDB(t)
+	insertUnassignedTask(t, db, "T-FREE", "purchase")
+
+	task, err := ClaimNextTask(db, "client-001", []string{"collect"})
+	if err != nil {
+		t.Fatalf("领取失败: %v", err)
+	}
+	if task != nil {
+		t.Errorf("只声明 collect 的客户端不该拿到 purchase 任务：%s", task.TaskID)
+	}
+}
+
+// ── 并发抢占 ───────────────────────────────────────────
+
+// 多个客户端同时抢同一条无主任务，只能有一个拿到。
+// 这是"谁领到算谁的"这个模式的安全底线——两台都拿到就会重复采集/重复下单。
+func TestClaim_并发抢无主任务只有一个拿到(t *testing.T) {
+	db := newTestDB(t)
+	insertUnassignedTask(t, db, "T-ONLY-ONE", "collect")
+
+	const workers = 8
+	var (
+		wg      sync.WaitGroup
+		mu      sync.Mutex
+		winners []string
+		lastErr error
+	)
+	for i := 0; i < workers; i++ {
+		clientID := "client-" + string(rune('A'+i))
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			task, err := ClaimNextTask(db, clientID, []string{"collect"})
+			mu.Lock()
+			defer mu.Unlock()
+			if err != nil {
+				lastErr = err
+				return
+			}
+			if task != nil {
+				winners = append(winners, clientID)
+			}
+		}()
+	}
+	wg.Wait()
+
+	if lastErr != nil {
+		t.Fatalf("并发领取出错: %v", lastErr)
+	}
+	if len(winners) != 1 {
+		t.Fatalf("同一条无主任务被 %d 个客户端领到，期望正好 1 个：%v",
+			len(winners), winners)
+	}
+
+	// 而且库里记的领取者必须就是那个赢家
+	client, _ := taskAssignee(t, db, "T-ONLY-ONE")
+	if client != winners[0] {
+		t.Errorf("assigned_client 记的是 %q，但实际领到的是 %q", client, winners[0])
+	}
+}
+
+// ── 领取历史仍然被记录 ─────────────────────────────────
+
+// 提交结果时的权限判断依赖 task_claims（只有从没领过的才 403），
+// 无主任务这条路径也必须记。
+func TestClaim_无主任务领取后也记领取历史(t *testing.T) {
+	db := newTestDB(t)
+	insertUnassignedTask(t, db, "T-FREE", "collect")
+	RegisterClient(db, model.Client{ClientID: "client-001"}, true)
+
+	if _, err := ClaimNextTask(db, "client-001", []string{"collect"}); err != nil {
+		t.Fatalf("领取失败: %v", err)
+	}
+
+	var n int
+	db.QueryRow(`SELECT COUNT(*) FROM task_claims
+	             WHERE task_id = 'T-FREE' AND client_id = 'client-001'`).Scan(&n)
+	if n != 1 {
+		t.Errorf("领取历史应有 1 条，实际 %d —— "+
+			"没有它的话提交结果会被误判成 403", n)
+	}
+}
@@ -58,8 +58,12 @@ X-Client-Id: client-001
 **处理步骤：**

 1. 先做客户端注册/更新（§3）；
-2. 查 `tasks`：`assigned_client = X-Client-Id` 且 `status = 'assigned'`，
-   按 `priority DESC, created_at` 取**一条**；
+2. 查 `tasks`，**两种任务都要查**：
+   - 指定给本机的：`assigned_client = X-Client-Id` 且 `status = 'assigned'`
+   - 无主的：`assigned_client IS NULL` 且 `status = 'pending'`
+
+   排序 `ORDER BY (assigned_client IS NULL), priority DESC, created_at`
+   —— **指定给本机的优先**；
 3. 没有就返回 `204 No Content`；
 4. 有就把 `status` 改成 `claimed`、记 `claimed_at`，返回任务。

@@ -69,8 +73,12 @@ X-Client-Id: client-001
 `[必须]` 第 4 步的查询和更新要在**同一个事务**里，用条件更新防并发：

 ```sql
-UPDATE tasks SET status = 'claimed', claimed_at = ?, updated_at = ?
-WHERE task_id = ? AND status = 'assigned';
+UPDATE tasks
+   SET status = 'claimed', assigned_client = ?, claimed_at = ?, updated_at = ?
+ WHERE task_id = ?
+   AND (   (status = 'assigned' AND assigned_client = ?)
+        OR (status = 'pending'  AND assigned_client IS NULL) );
+-- 两种情况合成一条：对"指定给我的"，写 assigned_client 是写同一个值，无副作用。
 -- 检查影响行数，为 0 说明被别人抢先了，重新取下一条
 ```

@@ -268,7 +276,10 @@ Client 契约里散落的 Admin 侧硬要求，汇总在这里，**可以直接
 - [x] claim 隐式登记**不更新名称**
 - [x] 登记和 claim **共用同一套字段校验**，非法内容返回 `422 INVALID_CLIENT_PROFILE`
 - [x] 校验不通过时不写库
- [ ] `claim` 一次只返回一个任务，且只返回分配给该客户端的
+- [ ] `claim` 一次只返回一个任务
+- [ ] 能领到无主任务（`assigned_client IS NULL` + `pending`），领取时写入领取者
+- [ ] 指定给本机的任务优先于无主任务
+- [ ] 不会返回指定给别的客户端的任务
 - [ ] `claim` 原子完成，并发下不会把同一任务发给两个客户端
 - [ ] `claim` 无任务时返回 `204`，不是 `200` 加空对象
 - [ ] 响应不含租约、不含 Admin 侧状态
@@ -216,7 +216,26 @@ POST /api/v1/client/tasks/claim

 ### 5.1 Admin 侧的分配语义

-已确认：**Admin 只把任务分配给指定的 Client**，`claim` 只会返回分配给本 `X-Client-Id` 的任务。
+任务分两种，`claim` **两种都会返回**：
+
+| 类型 | `assigned_client` | Admin 侧状态 | 谁能领 |
+|---|---|---|---|
+| **指定分配** | 某个 Client 编号 | `assigned` | 只有那个 Client |
+| **无主** | 空 | `pending` | **谁先抢到算谁的**，领取时才记下领取者 |
+
+`[必须]` **指定给本机的优先于无主的。** 显式分配是人为决定，应当先兑现；
+无主任务谁抢都一样，可以等。
+
+哪种任务用哪种方式：
+
+- **采集任务不指定客户端。** 采集只是浏览商品页，没有副作用，哪台设备采都一样，
+  没必要每次都挑一台。
+- **采购任务可以指定，也允许留空。** 涉及钱和账号——不同设备可能登着不同的
+  拼多多账号，买到谁头上是有区别的。**需要指定账号时必须显式分配**，
+  留空就意味着接受"谁先抢到谁去下单"。
+
+Client 侧对这两种没有任何区别：调 `claim`，拿到任务就做，做完提交。
+**不需要知道这个任务原来有没有主。**

 因此 Client 完全不需要看到全局任务池，本地也不缓存未领取的任务
 （见 [03 数据模型](03-data-model.md) §3.1）。操作人员想知道队列里还有多少活，
@@ -353,7 +372,8 @@ Admin 还没做完，下面这些要和 Admin 一起定。**不许因此停工**

 **已定案（不再是待确认项）：**

- Admin 只把任务分配给指定 Client，Client 不读全局任务池。见 §5.1。
+- 任务分「指定分配」和「无主」两种，`claim` 两种都返回，指定的优先。
+  采集任务不指定，采购任务可指定可留空。Client 不读全局任务池。见 §5.1。
 - **没有租约、没有心跳、没有状态回查。** 任务流程只有 §5 / §6 / §7 三个调用；设置页另有 §4.1 的幂等登记调用。
 - Admin 必须无条件接受已派发过的 Client 提交的结果。见 §6.1。