diff --git a/admin/handler/web/pdd.go b/admin/handler/web/pdd.go index bd1b55b..25a2838 100644 --- a/admin/handler/web/pdd.go +++ b/admin/handler/web/pdd.go @@ -113,19 +113,16 @@ func (h *Handler) PddCollect(c *gin.Context) { return } - created, skipped, err := service.CreatePddCollectTasks(h.db, ids) + result, err := service.CreatePddCollectTasks(h.db, ids) if err != nil { fail(c, http.StatusInternalServerError, "创建采集任务失败："+err.Error()+"。这一批任务整体没有创建，可以直接重试。") return } - msg := fmt.Sprintf("已创建 %d 个采集任务，等待客户端领取", created) - if skipped > 0 { - // 跳过了几个必须说出来，否则操作员会以为都建上了，等半天没动静 - msg += fmt.Sprintf("（跳过 %d 个采集中或已删除的）", skipped) - } - h.pddRedirect(c, msg) + // 跳过原因分类、"还要等多久"的措辞都是业务规则，交给 service 算， + // 这里只负责把结果接过来发给操作员，见 service.FormatCollectTaskMessage。 + h.pddRedirect(c, service.FormatCollectTaskMessage(result)) } // PddDelete 批量软删除。 diff --git a/admin/model/model.go b/admin/model/model.go index 4f29abe..0c0acc3 100644 --- a/admin/model/model.go +++ b/admin/model/model.go @@ -14,10 +14,26 @@ const TimeLayout = time.RFC3339 // NowISO 返回当前 UTC 时间的字符串形式。 // 所有写库的时间戳都要用它，不要各写各的格式。 +// +// `[必须]` 有代码依赖它产出**定宽 UTC 格式**（如 "2026-08-07T06:10:19Z"）： +// repository.MarkCollecting 和 service 里判断"采集是否超时"， +// 靠的是直接用字符串比较 `updated_at < ?`，不解析成时间再比。 +// 定宽 + 同一时区（UTC）+ 补零，字符串的字典序才等于时间先后顺序。 +// 改成带时区偏移的本地时间（比如 "+08:00"）之后，这个比较会**静默失效** +// ——不会报错，但超时判断会全错，见 #24。 func NowISO() string { return time.Now().UTC().Format(TimeLayout) } +// CollectStaleAfter 是采集任务多久没动静就当它死了。 +// +// 采集本身几十秒到两分钟；加上排队等客户端来领，15 分钟足够宽裕。 +// 宁可短也不要长：采集是**只读**操作，多采一次没有任何副作用， +// 而卡死的代价是这个商品永久报废、只能改数据库救，见 #24。 +// +// 定义成有名字的常量，不要把 15 分钟当魔数散落在各处判断里。 +const CollectStaleAfter = 15 * time.Minute + // ParseISO 解析库里存的时间字符串。解析不了返回零值和 false。 func ParseISO(s string) (time.Time, bool) { if s == "" { diff --git a/admin/repository/pdd.go b/admin/repository/pdd.go index e11647a..b496049 100644 --- a/admin/repository/pdd.go +++ b/admin/repository/pdd.go @@ -4,6 +4,7 @@ import ( "database/sql" "fmt" "strings" + "time" "cmautobuy/admin/model" ) @@ -350,17 +351,41 @@ func SetCollectFailed(q Execer, pddGoodsID, msg, artifactRef string) error { // MarkCollecting 把商品置为"采集中"。创建采集任务时调。 // -// 只有 pending / failed 状态才允许发起采集： -// 已经是 collecting 的说明有任务在跑，再建一个就是重复采集，浪费一次。 -// 返回 false 表示当前状态不允许，调用方应跳过并告诉操作员。 +// 允许发起采集的状态： +// - pending / failed —— 没有任务在跑； +// - collecting **且已经超时**（`updated_at` 早于 `now - model.CollectStaleAfter`） +// —— 客户端离线、崩溃或任务被删都会让一个 collecting 卡住不动， +// 这些是常态不是异常，界面上必须有出口，见 #24。 +// +// `[必须]` 超时判定在**读取的这一刻现算**（直接拼进这条 UPDATE 的 WHERE 里）， +// 不是靠后台协程定期把超时的状态改回 pending。本项目已经在并发上栽过两次 +// （PRAGMA 没作用到连接池、`BEGIN DEFERRED` 死锁），能不引入并发就不引入—— +// 现算没有调度、没有窗口期，天然不会有"扫到一半客户端正好提交了结果"这种竞态。 +// +// `[必须]` 用**字符串比较** `updated_at < ?`，不解析成 time.Time 再比。 +// 这是安全的，但依赖 model.NowISO() 永远产出定宽 UTC 格式，见那里的注释。 +// +// `[已知取舍]` 超时后允许重新建任务，但**老任务还留在队列里**， +// 客户端上线后可能把新旧两个任务都领走，同一个商品被采两次。 +// 这是可以接受的：采集是只读操作，没有任何副作用，后一次的结果覆盖前一次， +// 数据仍然正确。**不要看到"可能采两次"就加锁或加租约去"修"它** +// ——租约和心跳是被明确移除过的设计，见 docs/client/00-glossary.md +// 「为什么没有租约和心跳」一节；这条取舍详见 docs/admin/03-data-model.md §4.1。 +// +// 返回 false 表示当前状态不允许发起采集（真的在采集且没超时、已采集、 +// 或商品不存在/已删除），调用方应跳过并把原因告诉操作员。 func MarkCollecting(q Execer, pddGoodsID string) (bool, error) { - now := model.NowISO() + now := time.Now() + nowISO := now.UTC().Format(model.TimeLayout) + staleBefore := now.Add(-model.CollectStaleAfter).UTC().Format(model.TimeLayout) + res, err := q.Exec(` UPDATE pdd_products SET collect_status = 'collecting', updated_at = ? WHERE goods_id = ? AND deleted_at IS NULL - AND collect_status IN ('pending', 'failed')`, - now, pddGoodsID) + AND ( collect_status IN ('pending', 'failed') + OR (collect_status = 'collecting' AND updated_at < ?) )`, + nowISO, pddGoodsID, staleBefore) if err != nil { return false, fmt.Errorf("标记 PDD 商品 %s 采集中失败: %w", pddGoodsID, err) } diff --git a/admin/service/pdd.go b/admin/service/pdd.go index 9ca4620..89cb3a6 100644 --- a/admin/service/pdd.go +++ b/admin/service/pdd.go @@ -12,9 +12,11 @@ import ( "database/sql" "errors" "fmt" + "math" "net/url" "sort" "strings" + "time" "cmautobuy/admin/model" "cmautobuy/admin/repository" @@ -152,6 +154,37 @@ func collectStatusText(s model.CollectStatus) string { return string(s) } +// statusTextFor 返回一条 PDD 商品该显示的状态文字，比 collectStatusText +// 多做一件事：区分"采集中"和"采集中（超时）"。 +// +// `[必须]` 超时的必须显示不一样的文字——操作员盯着「采集中」不知道它已经死了， +// 会一直傻等；显示超时他才知道可以重新采，见 #24。 +// `[必须]` 只加文字，不新增筛选状态：超时不是数据库里真实存在的一种 +// collect_status 取值，只是"采集中"在读取那一刻的一种呈现。 +func statusTextFor(p model.PddProduct) string { + if isCollectingStale(p) { + return "采集中（超时）" + } + return collectStatusText(p.CollectStatus) +} + +// isCollectingStale 判断一条"采集中"记录是不是已经超时。 +// +// 现算，不依赖任何后台任务——本项目已经在并发上栽过两次， +// 能不引入并发就不引入，见 repository.MarkCollecting 的注释。 +// updated_at 解析不出来时保守当作"没超时"：数据本身已经有问题， +// 不该顺带触发"可以重新采集"，让问题被掩盖。 +func isCollectingStale(p model.PddProduct) bool { + if p.CollectStatus != model.CollectCollecting { + return false + } + t, ok := model.ParseISO(p.UpdatedAt) + if !ok { + return false + } + return time.Since(t) > model.CollectStaleAfter +} + // formatLocalTime 把库里的 ISO 8601 转成本地时区的可读写法。 // 空值或坏值都显示占位符，不显示 0001-01-01。 func formatLocalTime(iso string) string { @@ -236,7 +269,7 @@ func ListPddProducts(db *sql.DB, keyword string, status model.CollectStatus) (*P GoodsID: r.GoodsID, URL: r.URL, Title: r.Title, - StatusText: collectStatusText(r.CollectStatus), + StatusText: statusTextFor(r.PddProduct), CollectedAt: formatLocalTime(r.CollectedAt), UpdatedAt: formatLocalTime(r.UpdatedAt), CollectMsg: r.CollectMsg, @@ -323,7 +356,7 @@ func GetPddProductDetail(db *sql.DB, id int64) (*PddProductDetail, error) { GoodsID: p.GoodsID, URL: p.URL, Title: p.Title, - StatusText: collectStatusText(p.CollectStatus), + StatusText: statusTextFor(*p), CollectMsg: p.CollectMsg, CollectedAt: formatLocalTime(p.CollectedAt), ArtifactRef: p.ArtifactRef, @@ -453,6 +486,28 @@ func DeletePddProducts(db *sql.DB, goodsIDs []string) (int64, error) { // ---------- 创建采集任务 ---------- +// CollectTaskResult 是创建采集任务的结果，供 handler 组装状态条提示。 +// +// `[必须]` 跳过原因要分开计数，不能只给一个笼统的"跳过 N 个"—— +// 那种提示对操作员没有信息量：不知道该等还是该去处理别的，见 #24。 +type CollectTaskResult struct { + Created int + + SkippedCollecting int // 正在采集且**没超时**，得等客户端来领/提交 + SkippedCollected int // 已经采集完成，本来就不需要重新采集 + SkippedDeleted int // 商品不存在或已被软删除 + + // RetryWaitText 是 SkippedCollecting 里最快能重试的还需等待时长， + // 形如 "12 分钟"；SkippedCollecting 为 0 时是空串。 + RetryWaitText string +} + +// Skipped 是跳过总数，供测试和粗粒度统计用； +// 界面提示要按原因分开说，见 FormatCollectTaskMessage，不要直接显示这个数。 +func (r CollectTaskResult) Skipped() int { + return r.SkippedCollecting + r.SkippedCollected + r.SkippedDeleted +} + // CreatePddCollectTasks 为勾选的 PDD 商品创建采集任务。 // // `[必须]` **不指定客户端**（assigned_client 为 NULL，status 为 pending）， @@ -461,56 +516,148 @@ func DeletePddProducts(db *sql.DB, goodsIDs []string) (int64, error) { // // 规则： // - 按 goods_id 去重，勾了重复的只建一个； -// - collect_status 已经是 collecting 的**跳过**——有任务在跑了， +// - collect_status 是 collecting 且**没超时**的跳过——有任务在跑了， // 再建一个就是让两台机器采同一个商品，白费一趟； +// collecting 但**已超时**的（见 model.CollectStaleAfter）当作可以重建， +// 不再跳过，这是 #24 要修的死锁； +// - 已采集的跳过（本来就不需要重采）； // - 建成功后把状态置为 collecting。 // -// 返回的 skipped 必须显示给操作员。静默跳过的话， +// 返回的跳过分类必须显示给操作员。静默跳过的话， // 操作员会以为任务建好了，等半天没动静也不知道为什么。 -func CreatePddCollectTasks(db *sql.DB, goodsIDs []string) (created, skipped int, err error) { +func CreatePddCollectTasks(db *sql.DB, goodsIDs []string) (CollectTaskResult, error) { + var result CollectTaskResult + goodsIDs = dedupe(goodsIDs) if len(goodsIDs) == 0 { - return 0, 0, nil + return result, nil } tx, err := db.Begin() if err != nil { - return 0, 0, fmt.Errorf("开始事务失败: %w", err) + return CollectTaskResult{}, fmt.Errorf("开始事务失败: %w", err) } defer tx.Rollback() // 已提交的事务再 Rollback 是空操作，安全 + // 跳过原因里"正在采集"的那些，各自还要等多久才超时可重试； + // 取其中最快的一个，给操作员一个"下一步该等多久"的具体数字。 + var ( + minRetryWait time.Duration + hasMinRetryWait bool + ) + for _, goodsID := range goodsIDs { p, err := repository.GetPddProductByGoodsID(tx, goodsID) if err != nil { - return 0, 0, err + return CollectTaskResult{}, err } if p == nil || p.IsDeleted() { - skipped++ + result.SkippedDeleted++ continue } - // 先占状态再建任务：MarkCollecting 只在 pending/failed 时成功， - // 它同时起到"这个商品有没有人已经在采"的判断作用。 + // 先占状态再建任务：MarkCollecting 决定"这个商品现在允不允许发起采集" + // （pending/failed，或 collecting 但已超时），它同时起到原子抢占的作用—— + // 同一时刻只有一个并发请求能把它从"可采"改成"collecting"。 ok, err := repository.MarkCollecting(tx, goodsID) if err != nil { - return 0, 0, err + return CollectTaskResult{}, err } if !ok { - skipped++ + // MarkCollecting 内部已经把"pending/failed"和"collecting 但已超时" + // 都判成允许，所以这里失败时 p 只可能是两种情况： + // - collecting 且没超时 —— 真的有任务在跑； + // - collected —— 采集已经完成，不需要重采。 + // （p 是这个事务里刚读到的，事务全程持有写锁，中途不会被别人改。） + if p.CollectStatus == model.CollectCollecting { + result.SkippedCollecting++ + if wait, ok := retryWaitFor(p.UpdatedAt); ok { + if !hasMinRetryWait || wait < minRetryWait { + minRetryWait, hasMinRetryWait = wait, true + } + } + } else { + result.SkippedCollected++ + } continue } if err := repository.InsertCollectTask( tx, newCollectTaskID(), p.GoodsID, p.URL); err != nil { - return 0, 0, err + return CollectTaskResult{}, err } - created++ + result.Created++ } if err := tx.Commit(); err != nil { - return 0, 0, fmt.Errorf("提交事务失败: %w", err) + return CollectTaskResult{}, fmt.Errorf("提交事务失败: %w", err) } - return created, skipped, nil + if hasMinRetryWait { + result.RetryWaitText = formatRetryWait(minRetryWait) + } + return result, nil +} + +// retryWaitFor 算一条"采集中"记录还要多久才会被判定超时、可以重新发起采集。 +// updated_at 解析不出来时返回 (0, false)——不该拿一个解析不出的时间瞎猜等待时长。 +func retryWaitFor(updatedAt string) (time.Duration, bool) { + t, ok := model.ParseISO(updatedAt) + if !ok { + return 0, false + } + remain := model.CollectStaleAfter - time.Since(t) + if remain < 0 { + remain = 0 + } + return remain, true +} + +// formatRetryWait 把还需等待的时长翻成"12 分钟"这样的人话。 +// 向上取整、且至少显示 1 分钟——不然临界值会显示"0 分钟才可重试"， +// 操作员会以为可以立刻点，实际上还没到点。 +func formatRetryWait(remain time.Duration) string { + minutes := int(math.Ceil(remain.Minutes())) + if minutes < 1 { + minutes = 1 + } + return fmt.Sprintf("%d 分钟", minutes) +} + +// FormatCollectTaskMessage 把 CreatePddCollectTasks 的结果组装成状态条提示。 +// +// `[必须]` 一个都没建成时不能只说"已创建 0 个"，要让操作员知道下一步该干嘛 +// （比如还要等多久），见 #24。 +func FormatCollectTaskMessage(r CollectTaskResult) string { + var reasons []string + if r.SkippedCollecting > 0 { + reason := fmt.Sprintf("%d 个正在采集中", r.SkippedCollecting) + if r.Created > 0 { + // 已经建成了一批时，措辞换成"正在采集的"，配合"跳过…"这句话通顺 + reason = fmt.Sprintf("%d 个正在采集的", r.SkippedCollecting) + } else if r.RetryWaitText != "" { + reason += fmt.Sprintf("（还需等待约 %s才可重试）", r.RetryWaitText) + } + reasons = append(reasons, reason) + } + if r.SkippedCollected > 0 { + reasons = append(reasons, fmt.Sprintf("%d 个已采集的", r.SkippedCollected)) + } + if r.SkippedDeleted > 0 { + reasons = append(reasons, fmt.Sprintf("%d 个已删除的", r.SkippedDeleted)) + } + + if r.Created == 0 { + if len(reasons) == 0 { + return "没有创建任何任务" + } + return "没有创建任何任务：" + strings.Join(reasons, "、") + } + + msg := fmt.Sprintf("已创建 %d 个采集任务，等待客户端领取", r.Created) + if len(reasons) > 0 { + msg += fmt.Sprintf("（跳过 %s）", strings.Join(reasons, "、")) + } + return msg } // newCollectTaskID 生成采集任务编号。 diff --git a/admin/service/pdd_page_test.go b/admin/service/pdd_page_test.go index 9abd13d..f7a0358 100644 --- a/admin/service/pdd_page_test.go +++ b/admin/service/pdd_page_test.go @@ -4,7 +4,9 @@ import ( "database/sql" "errors" "strings" + "sync" "testing" + "time" "cmautobuy/admin/model" "cmautobuy/admin/repository" @@ -530,12 +532,12 @@ func TestCreatePddCollectTasks_不指定客户端(t *testing.T) { db := newTestDB(t) createProduct(t, db, "737116531267") - created, skipped, err := CreatePddCollectTasks(db, []string{"737116531267"}) + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) if err != nil { t.Fatalf("建任务失败: %v", err) } - if created != 1 || skipped != 0 { - t.Fatalf("created=%d skipped=%d，期望 1/0", created, skipped) + if result.Created != 1 || result.Skipped() != 0 { + t.Fatalf("Created=%d Skipped=%d，期望 1/0", result.Created, result.Skipped()) } status, assigned, url, ok := collectTaskOf(t, db, "737116531267") @@ -582,13 +584,13 @@ func TestCreatePddCollectTasks_按商品去重(t *testing.T) { db := newTestDB(t) createProduct(t, db, "737116531267") - created, _, err := CreatePddCollectTasks(db, + result, err := CreatePddCollectTasks(db, []string{"737116531267", "737116531267", " 737116531267 ", ""}) if err != nil { t.Fatalf("建任务失败: %v", err) } - if created != 1 { - t.Errorf("重复勾选只该建 1 个任务，实际 %d", created) + if result.Created != 1 { + t.Errorf("重复勾选只该建 1 个任务，实际 %d", result.Created) } var n int @@ -598,28 +600,31 @@ func TestCreatePddCollectTasks_按商品去重(t *testing.T) { } } -// 已经在采的跳过：再建一个就是让两台机器采同一个商品，白费一趟。 +// 已经在采、还没超时的跳过：再建一个就是让两台机器采同一个商品，白费一趟。 // 而且跳过了几个必须报出来，静默跳过会让操作员等半天不知道为什么没动静。 -func TestCreatePddCollectTasks_采集中的跳过并报数(t *testing.T) { +func TestCreatePddCollectTasks_采集中未超时的跳过并报数(t *testing.T) { db := newTestDB(t) createProduct(t, db, "100000000001") createProduct(t, db, "100000000002") - // 第一个先建一次，让它进入 collecting - if _, _, err := CreatePddCollectTasks(db, []string{"100000000001"}); err != nil { + // 第一个先建一次，让它进入 collecting，updated_at 是刚才，远没到 15 分钟 + if _, err := CreatePddCollectTasks(db, []string{"100000000001"}); err != nil { t.Fatalf("第一次建任务失败: %v", err) } - created, skipped, err := CreatePddCollectTasks(db, + result, err := CreatePddCollectTasks(db, []string{"100000000001", "100000000002"}) if err != nil { t.Fatalf("第二次建任务失败: %v", err) } - if created != 1 { - t.Errorf("只该给没在采的那个建任务，实际建了 %d 个", created) + if result.Created != 1 { + t.Errorf("只该给没在采的那个建任务，实际建了 %d 个", result.Created) } - if skipped != 1 { - t.Errorf("采集中的那个应计入 skipped，实际 %d", skipped) + if result.SkippedCollecting != 1 { + t.Errorf("采集中且没超时的那个应计入 SkippedCollecting，实际 %d", result.SkippedCollecting) + } + if result.RetryWaitText == "" { + t.Error("没超时的应该给出还要等多久，RetryWaitText 不该为空") } } @@ -629,12 +634,12 @@ func TestCreatePddCollectTasks_失败的可以重新采集(t *testing.T) { createProduct(t, db, "737116531267") repository.SetCollectFailed(db, "737116531267", "页面打不开", "") - created, skipped, err := CreatePddCollectTasks(db, []string{"737116531267"}) + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) if err != nil { t.Fatalf("建任务失败: %v", err) } - if created != 1 || skipped != 0 { - t.Errorf("失败的应该能重采，created=%d skipped=%d", created, skipped) + if result.Created != 1 || result.Skipped() != 0 { + t.Errorf("失败的应该能重采，Created=%d Skipped=%d", result.Created, result.Skipped()) } } @@ -643,19 +648,195 @@ func TestCreatePddCollectTasks_已删除的跳过(t *testing.T) { createProduct(t, db, "737116531267") DeletePddProducts(db, []string{"737116531267"}) - created, skipped, err := CreatePddCollectTasks(db, []string{"737116531267"}) + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) if err != nil { t.Fatalf("不该报错: %v", err) } - if created != 0 || skipped != 1 { - t.Errorf("已删除的应被跳过，created=%d skipped=%d", created, skipped) + if result.Created != 0 || result.SkippedDeleted != 1 { + t.Errorf("已删除的应被跳过，Created=%d SkippedDeleted=%d", result.Created, result.SkippedDeleted) + } +} + +// 已经采集完成的跳过：本来就不需要重新采集，不该算进"正在采集"那一类。 +func TestCreatePddCollectTasks_已采集的跳过并单独计数(t *testing.T) { + db := newTestDB(t) + createProduct(t, db, "737116531267") + repository.SetCollectResult(db, "737116531267", "已采完的商品", sampleSkusJSON) + + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) + if err != nil { + t.Fatalf("不该报错: %v", err) + } + if result.Created != 0 || result.SkippedCollected != 1 || result.SkippedCollecting != 0 { + t.Errorf("已采集的应单独计入 SkippedCollected，实际 %+v", result) } } func TestCreatePddCollectTasks_什么都没勾时不建任务(t *testing.T) { db := newTestDB(t) - created, skipped, err := CreatePddCollectTasks(db, nil) - if err != nil || created != 0 || skipped != 0 { - t.Errorf("空输入应安静返回，得到 %d/%d/%v", created, skipped, err) + result, err := CreatePddCollectTasks(db, nil) + if err != nil || result.Created != 0 || result.Skipped() != 0 { + t.Errorf("空输入应安静返回，得到 %+v/%v", result, err) + } +} + +// ── 超时：#24 要修的死锁 ───────────────────────────────── + +// insertStaleCollectingProduct 直接建一条"20 分钟前就进入 collecting"的商品， +// 并配一条同样卡在 20 分钟前的采集任务，模拟"客户端领了任务却再也没提交结果" +// （离线/崩溃/任务被删）的常见场景——这正是 #24 复现步骤里卡死的那种状态。 +// +// `[必须]` 用固定时间戳而不是让代码在测试运行时调用 time.Now() 现算 20 分钟前， +// 否则测试结果会跟着执行的那一刻的系统时钟漂移，CLAUDE.md §5 明确要求这样写。 +func insertStaleCollectingProduct(t *testing.T, db *sql.DB, goodsID string) { + t.Helper() + createProduct(t, db, goodsID) + staleAt := time.Now().Add(-20 * time.Minute).UTC().Format(model.TimeLayout) + if _, err := db.Exec( + `UPDATE pdd_products SET collect_status = 'collecting', updated_at = ? WHERE goods_id = ?`, + staleAt, goodsID); err != nil { + t.Fatalf("构造超时数据失败: %v", err) + } + // 卡死的这一条旧任务：客户端已经 claimed，但再也没提交结果， + // updated_at 和商品一样停在 20 分钟前。 + if _, err := db.Exec(` + INSERT INTO tasks (task_id, task_type, status, assigned_client, + pdd_goods_url, pdd_goods_id, created_at, updated_at) + VALUES (?, 'collect', 'claimed', 'client-已离线', + 'https://mobile.yangkeduo.com/goods.html?goods_id=' || ?, ?, ?, ?)`, + "COL-STALE-"+goodsID, goodsID, goodsID, staleAt, staleAt); err != nil { + t.Fatalf("构造卡死任务失败: %v", err) + } +} + +// 验收标准：「collecting 且 updated_at 超过 15 分钟 → 能重新创建采集任务」。 +// 老任务不会被撤销——它还在队列里，见 MarkCollecting 注释里"可能采两次"的取舍。 +func TestCreatePddCollectTasks_超时后可以重新建任务(t *testing.T) { + db := newTestDB(t) + insertStaleCollectingProduct(t, db, "737116531267") + + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) + if err != nil { + t.Fatalf("建任务失败: %v", err) + } + if result.Created != 1 || result.SkippedCollecting != 0 { + t.Errorf("超时的应该能重建，Created=%d SkippedCollecting=%d", + result.Created, result.SkippedCollecting) + } + + // 商品状态回到 collecting（这次是"新一轮"采集），updated_at 也刷新了 + p, _ := repository.GetPddProductByGoodsID(db, "737116531267") + if p.CollectStatus != model.CollectCollecting { + t.Errorf("商品状态应回到 collecting，实际 %s", p.CollectStatus) + } + + // [已知取舍] 旧的那条卡死任务还在队列里没被撤销，加上新建的这条，应该有 2 条。 + var n int + db.QueryRow(`SELECT COUNT(*) FROM tasks WHERE task_type = 'collect' AND pdd_goods_id = ?`, + "737116531267").Scan(&n) + if n != 2 { + t.Errorf("应该是 1 条旧的卡死任务 + 1 条新建的 = 2 条，实际 %d", n) + } +} + +// 验收标准：「collecting 但没到 15 分钟 → 仍然跳过，不重复建」。 +// 用 5 分钟前（明显没到 15 分钟阈值）而不是踩着边界，测试更稳。 +func TestCreatePddCollectTasks_collecting未到十五分钟仍然跳过(t *testing.T) { + db := newTestDB(t) + createProduct(t, db, "737116531267") + notStaleAt := time.Now().Add(-5 * time.Minute).UTC().Format(model.TimeLayout) + if _, err := db.Exec( + `UPDATE pdd_products SET collect_status = 'collecting', updated_at = ? WHERE goods_id = ?`, + notStaleAt, "737116531267"); err != nil { + t.Fatalf("构造数据失败: %v", err) + } + + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) + if err != nil { + t.Fatalf("不该报错: %v", err) + } + if result.Created != 0 || result.SkippedCollecting != 1 { + t.Errorf("没到 15 分钟的应该继续跳过，Created=%d SkippedCollecting=%d", + result.Created, result.SkippedCollecting) + } +} + +// 并发：多个请求同时对同一个"已超时"商品建采集任务，只能产生 1 条新任务。 +// 加了超时分支之后，原有的原子抢占（#18 已验证过的性质）不能因此失效—— +// 否则多台客户端会同时去采同一个商品，白费好几趟。 +func TestCreatePddCollectTasks_并发对同一个已超时商品建任务只产生一条(t *testing.T) { + db := newTestDB(t) + insertStaleCollectingProduct(t, db, "737116531267") + + const workers = 8 + var ( + wg sync.WaitGroup + mu sync.Mutex + totalCreated int + lastErr error + ) + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + result, err := CreatePddCollectTasks(db, []string{"737116531267"}) + mu.Lock() + defer mu.Unlock() + if err != nil { + lastErr = err + return + } + totalCreated += result.Created + }() + } + wg.Wait() + + if lastErr != nil { + t.Fatalf("并发建任务出错: %v", lastErr) + } + if totalCreated != 1 { + t.Fatalf("同一个已超时商品被并发建了 %d 次任务，期望正好 1 次", totalCreated) + } + + // 旧的卡死任务 1 条 + 新建的 1 条 = 2 条，不会因为并发多建出第 3 条 + var n int + db.QueryRow(`SELECT COUNT(*) FROM tasks WHERE task_type = 'collect' AND pdd_goods_id = ?`, + "737116531267").Scan(&n) + if n != 2 { + t.Errorf("库里采集任务数应为 2，实际 %d", n) + } +} + +// ── 提示文案 ─────────────────────────────────────────── + +func TestFormatCollectTaskMessage_建成且有跳过时按原因分类(t *testing.T) { + msg := FormatCollectTaskMessage(CollectTaskResult{ + Created: 2, + SkippedCollecting: 1, + SkippedDeleted: 1, + }) + want := "已创建 2 个采集任务，等待客户端领取（跳过 1 个正在采集的、1 个已删除的）" + if msg != want { + t.Errorf("消息 = %q，期望 %q", msg, want) + } +} + +// 一个都没建成时不能只说"已创建 0 个"，要告诉操作员还要等多久。 +func TestFormatCollectTaskMessage_一个都没建成时说明还要等多久(t *testing.T) { + msg := FormatCollectTaskMessage(CollectTaskResult{ + SkippedCollecting: 1, + RetryWaitText: "12 分钟", + }) + want := "没有创建任何任务：1 个正在采集中（还需等待约 12 分钟才可重试）" + if msg != want { + t.Errorf("消息 = %q，期望 %q", msg, want) + } +} + +func TestFormatCollectTaskMessage_全部建成不带跳过语句(t *testing.T) { + msg := FormatCollectTaskMessage(CollectTaskResult{Created: 3}) + want := "已创建 3 个采集任务，等待客户端领取" + if msg != want { + t.Errorf("消息 = %q，期望 %q", msg, want) } } diff --git a/docs/admin/03-data-model.md b/docs/admin/03-data-model.md index 5da5a4e..8ccfe4f 100644 --- a/docs/admin/03-data-model.md +++ b/docs/admin/03-data-model.md @@ -299,6 +299,56 @@ CREATE INDEX idx_pdd_products_status ON pdd_products(collect_status); 所以存 `artifact_ref`（形如 `client-001:artifacts/PDD-0001/attempt-xxx/`）， 告诉操作员去哪台机器的哪个目录捞。 +**采集超时：`collecting` 怎么才不会永久卡死** + +客户端离线、崩溃、任务被删——这些都是常态，不是异常。原来的规则是 +"只有 `pending` / `failed` 才允许建采集任务"，而**只有客户端提交结果**才会 +把状态从 `collecting` 改走。于是上面任何一种情况发生后，这个商品的 +`collect_status` 就永远卡在 `collecting`，界面上没有任何入口能救回来， +只能改数据库（见 #24）。 + +修复：`collecting` 状态超过 `model.CollectStaleAfter`（15 分钟）**也允许** +重新建采集任务： + +```sql +-- repository.MarkCollecting，简化版 +UPDATE pdd_products + SET collect_status = 'collecting', updated_at = ? + WHERE goods_id = ? AND deleted_at IS NULL + AND ( collect_status IN ('pending', 'failed') + OR (collect_status = 'collecting' AND updated_at < ?) ) +``` + +`[必须]` 15 分钟为什么是这个数：采集本身几十秒到两分钟，加上排队等客户端来领， +15 分钟足够宽裕。宁可短也不要长——采集是**只读**操作，多采一次没有任何副作用， +而卡死的代价是这个商品永久报废、只能改数据库救。定义成常量 `model.CollectStaleAfter`， +不要把 15 分钟当魔数散落在各处判断里。 + +`[必须]` 超时判定在**读取的这一刻现算**（拼进上面这条 UPDATE 的 WHERE 里）， +**不是**后台协程定期扫描把超时的状态改回 `pending`。本项目已经在并发上栽过两次 +（PRAGMA 没作用到连接池、`BEGIN DEFERRED` 死锁），能不引入并发就不引入； +现算没有调度、没有窗口期，天然不会有"扫到一半客户端正好提交了结果"这种竞态。 +同理，**没有引入租约（lease）或心跳**——那是被明确移除的设计， +见 [Client 术语表](../client/00-glossary.md)「为什么没有租约和心跳」一节。 + +`[必须]` 时间比较用**字符串比较** `updated_at < ?`，不解析成 `time.Time` 再比。 +这依赖 `model.NowISO()` 永远产出定宽 UTC 格式（如 `2026-08-07T06:10:19Z`）： +定宽 + 同一时区 + 补零，字符串的字典序才等于时间先后顺序。 +谁把它改成带时区偏移的本地时间（比如 `+08:00`），这个比较会**静默失效** +——不会报错，但超时判断全错，见 `model.NowISO` 的注释。 + +**已知取舍：可能采两次** + +超时后允许重新建任务，但**老任务还留在队列里**，客户端上线后可能把新旧两个 +任务都领走，同一个商品被采两次。这是可以接受的： + +- 采集是只读操作，没有任何副作用； +- 后一次的结果覆盖前一次（`SetCollectResult` 按 `goods_id` 整体覆盖），数据仍然正确。 + +`[必须]` **不要看到"可能采两次"就去加锁或加租约"修"它**——那正是被明确移除的设计， +加回来会重新引入本项目已经吃过两次亏的并发复杂度，而这里换来的收益（避免极少数情况下 +多采一次）远小于代价。 + ### 4.2 `skus_json` 的结构 由 Client 采集后原样提交，Admin **不做转换**： diff --git a/docs/admin/05-ui-specification.md b/docs/admin/05-ui-specification.md index 9e01519..5e94306 100644 --- a/docs/admin/05-ui-specification.md +++ b/docs/admin/05-ui-specification.md @@ -166,7 +166,7 @@ | 商品 ID | `goods_id` | | 标题 | 采集回来的；未采集时显示"（未采集，采集后自动回填）" | | PDD 链接 | 截断显示，可点开（`target="_blank"` 要带 `rel="noopener noreferrer"`） | -| 采集状态 | 中文文字，**不能只靠颜色**；失败时在下面补一行原因 | +| 采集状态 | 中文文字，**不能只靠颜色**；失败时在下面补一行原因；`collecting` 超过 15 分钟没动静显示「**采集中（超时）**」，见下 | | 规格数 | 从 `skus_json` 算 | | 采集时间 | 本地时区；未采集显示 `—` | | 更新时间 | 本地时区 | @@ -184,6 +184,19 @@ `[必须]` 空状态分两种文案：从没创建过 → 引导去点「创建」；筛选无结果 → 给"查看全部"的入口。 +`[必须]` **`collecting` 超过 15 分钟没有更新，显示「采集中（超时）」，不是「采集中」。** +客户端离线、崩溃、任务被删都是常态，不是异常——一旦发生，操作员盯着「采集中」 +不知道它已经死了，会一直傻等；显示超时他才知道可以重新点「创建采集任务」。 +判定在**读取列表的这一刻现算**，不依赖任何后台任务，见 +[03 数据模型](03-data-model.md) §4「采集超时」。 + +`[必须]` 状态照旧**不能只靠颜色区分**，必须有文字——超时也是文字的一部分， +不能只是把这一行的底色改深。 + +`[建议]` 筛选下拉里**不新增**「已超时」这个选项。它不是数据库里真实存在的 +一种 `collect_status` 取值，只是「采集中」在读取那一刻的一种呈现； +加进筛选会让下拉框的取值和 `collect_status` 对不上。 + ### 5.3 创建弹窗 `[必须]` **只填 PDD 链接**，其余字段全靠采集回填。 @@ -251,7 +264,10 @@ Go 的 map 是无序的，不靠它定顺序的话，同一个商品每次刷新 指定了反而会在那台机器关着的时候干等。 - 勾选多行 → 按 `goods_id` 去重 → 建任务； -- `collect_status` 已是 `collecting` 的**跳过**； +- `collect_status` 是 `collecting` 且**没超时**（15 分钟内）的**跳过**； + 已超时的当作可以重建，不再跳过——这是 #24 修的死锁， + 见 [03 数据模型](03-data-model.md) §4「采集超时」； +- 已经是 `collected` 的**跳过**（本来就不需要重新采集）； - 建成功后把状态置为 `collecting`； - 任务的 `pdd_goods_url` 和 `pdd_goods_id` 必须填（Client 契约要求 `goods_url` 必填）。 @@ -259,10 +275,18 @@ Go 的 map 是无序的，不靠它定顺序的话，同一个商品每次刷新 不是立刻去采——真正的采集要等 Client 来领、去手机上跑，可能几秒也可能几分钟。 点完页面上只有状态从"未采集"变成"采集中"，文案不说清楚操作员会以为没生效。 -`[必须]` 结果在状态条明确提示，**跳过了几个也要说**： +`[必须]` 结果在状态条明确提示，**跳过原因要分开说**，不能只给一句笼统的 +"跳过 N 个"——那样操作员不知道该等还是该做点什么： ```text -已创建 3 个采集任务，等待客户端领取（跳过 1 个采集中或已删除的） +已创建 2 个采集任务，等待客户端领取（跳过 1 个正在采集的、1 个已删除的） +``` + +`[必须]` **一个都没建成时不要只说「已创建 0 个」**，要让操作员知道下一步该干嘛， +比如还要等多久： + +```text +没有创建任何任务：1 个正在采集中（还需等待约 12 分钟才可重试） ``` 静默跳过的话，操作员会以为任务都建上了，等半天没动静也不知道为什么。