fix: 商品卡在「采集中」无法恢复 (#24)
MarkCollecting 原本只在 pending/failed 时成功，而全库只有客户端提交结果 才会把 collecting 改走。客户端离线、崩溃、任务被删都是常态——一旦发生， 这个商品就永久报废，界面上没有任何入口能救，只能改数据库。 改成 collecting 超过 15 分钟视为已超时，允许重新创建采集任务。 判定在读取那一刻现算，仍是同一条原子 UPDATE，不加后台清理协程： 后台扫要处理"扫到一半客户端正好提交了"的竞态，本项目已经在并发上 栽过两次，能不引入并发就不引入。 15 分钟写成有名字的常量并注明理由：采集本身几十秒到两分钟，加上排队 等客户端来领。宁可短也不要长——采集是只读的，多采一次没有副作用， 而卡死的代价是商品永久报废。 已知取舍：超时后老任务还在队列里，客户端上线可能两个都领走、采两次。 可接受（只读，后一次覆盖前一次），已写进代码注释和 03-data-model.md， 免得后来人当成 bug 去"修"成加锁或加租约——租约和心跳是被明确移除的设计。 时间比较用字符串，依赖 NowISO 产出定宽 UTC。已在 NowISO 上加注释： 改成带时区偏移的本地时间会让这个比较静默失效，不报错但判断全错。 界面两处：超时的显示「采集中（超时）」，否则操作员盯着「采集中」 不知道它已经死了；跳过原因按正在采集/已采集/已删除分类计数， 一个都没建成时告诉操作员还要等多久。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
chengma
+164
-17
@@ -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 生成采集任务编号。
|
||||
|
||||
Reference in New Issue
Block a user