第15章：仓储层设计与任务抢占
本章总论 internal/repository：仓储接口的书写规范（注释即契约）、防误更新的 Select 限定、关键词筛选的统一抽象 KeywordGroups，以及全系统最核心的机制——ClaimPendingByType 乐观锁抢占。素材表与任务表的两套抢占实现同构，本章以任务仓储为主线。
流程
仓储层向上暴露的接口是一组具名业务操作而非通用 CRUD：TaskRepository 有 ClaimPendingByType、MarkCompleted、UpdateDraftURL；LiveMaterialRepository 有 ClaimPendingASR、UpdateASRProgress、ResetASRToPending。每个方法的注释写明前置条件与生效条件（如「仅当 asr_status=processing 且 asr_version 匹配时生效」）——仓储把状态机的守卫条件内化到 SQL 的 WHERE 子句里。
抢占流程（多实例安全的核心）：
flowchart TD
A[Worker: SELECT 最早一条 pending] --> B[UPDATE status=processing<br/>WHERE id=? AND status=pending AND version=?]
B -->|RowsAffected=1| C[抢占成功 返回任务 执行]
B -->|RowsAffected=0| D[被别人抢了 重试下一条<br/>最多 64 次]
D --> A
E[另一个 Worker] --> B
执行期的每次状态写回都带 version 条件：UpdateProgress 仅对 processing 生效；UpdateASRProgress 匹配 asr_version。这意味着即使孤儿回收把任务重置回 pending 且被新 Worker 抢走（version 已递增），旧 Worker 的迟到写入会被 WHERE 拦下，不会污染新一轮执行——注释里「避免超时回收后旧 Worker 继续写回」说的正是这个场景。
实现
任务抢占的实现要点（task_repo.go）：
- 查询选「最早创建」的 pending（隐含 FIFO 公平），
LIMIT 1； - 抢占用单条
UPDATE ... SET status='processing', version=version+1, started_at=now() WHERE id=? AND status='pending' AND version=?，靠 RowsAffected 判定胜负——数据库行锁天然序列化并发 UPDATE； claimOptimisticMaxAttempts = 64：注释解释「多 Worker 同时选中同一 pending 行时，仅一个 CAS 成功，其余需改抢下一条；上限避免极端竞争下空转」。
其余设计规范贯穿全部仓储文件：
- Select 限定更新列：
UpdateNameRemark用.Select("name", "remark")并注释「确保 live_url、ASR 等字段不会被意外覆盖」——GORM 默认 Updates 会写全部非零字段，Select 是防半吊子更新事故的第一道闸。 - 列表查询瘦身：LiveMaterial 的 List 注释「不含 live_asr 字段」（原始 JSON 可能数 MB）；Task 的 List 直接读快照字段零 JOIN。
- KeywordGroups 统一抽象（keyword_filter.go）：
组内 AND、组间 OR的关键词表达式（如 "利率 债券 | 股票" 表示 (利率 AND 债券) OR 股票），任务列表匹配 video_project_name、素材列表匹配 name/remark、ASR 关键词匹配 asr_paragraphs 文本（PG 端asr_paragraphs::text ILIKE）。一处抽象三处复用，前端搜一下统一语义。 - Delete 的级联：LiveMaterial.Delete 注释「物理删除素材，并级联删除关联的 video_project 记录」——删除在一个事务里完成逻辑级联（弱关联的善后责任归仓储）。
- 每个仓储都有接口定义 + 私有结构体 +
New*构造器，service 与 handler 全部面向接口（task_repo_test.go等直接测 SQL 行为）。
📌 设计决策
- 具名操作而非通用 Save/Find：状态机守卫进 SQL，调用方无法绕过；代价是方法数量多，但每个都可独立测试。
- 版本条件写回（而非无脑 UPDATE）：用「乐观读-条件写」替代长事务，Worker 崩溃恢复的正确性由此保证。
- 抢占在仓储而非 Worker 实现：多实例正确性是数据层不变式，任何未来新增的 Worker 类型直接复用。
代码示例
仓储接口的注释密度——契约全在方法签名旁：
// internal/repository/task_repo.go（节选）
type TaskRepository interface {
// ClaimPendingByType 多实例安全地抢占一条指定类型的 pending 任务。
// 使用乐观锁（version CAS）：将状态改为 processing 并返回；无待处理任务时返回 nil。
ClaimPendingByType(ctx context.Context, taskType string) (*model.Task, error)
// UpdateProgress 更新任务进度（0-100），仅 processing 状态时生效。
UpdateProgress(ctx context.Context, id string, progress int16) error
// MarkCompleted 标记任务成功完成，并写入最终进度与扩展字段。
MarkCompleted(ctx context.Context, id string, progress int16, ext string) error
// MarkFailed 标记任务失败，写入错误信息与当前进度。
MarkFailed(ctx context.Context, id string, progress int16, errorMsg string) error
// RequeueStaleProcessingByType 将指定类型下超时未更新的 processing 任务
// 重置为 pending，供崩溃/重启恢复。以 updated_at 为心跳。
RequeueStaleProcessingByType(ctx context.Context, taskType string, olderThan time.Duration) (int64, error)
}
Select 防误更新的模式（素材仓储实拍）：
// internal/repository/live_material_repo.go（节选）
// UpdateNameRemark 仅更新素材名称与备注，防止误改其它字段。
func (r *liveMaterialRepository) UpdateNameRemark(ctx context.Context, material *model.LiveMaterial) error {
// Select 限定更新列，确保 live_url、ASR 等字段不会被意外覆盖。
return r.db.WithContext(ctx).
Model(material).
Select("name", "remark").
Updates(material).Error
}
小结
- 仓储 = 具名业务操作 + SQL 级状态守卫 + 瘦身列表 + 统一关键词抽象。
- 乐观锁抢占三要素：FIFO 选择、CAS UPDATE 判 RowsAffected、版本条件写回防迟到写入。
- 崩溃恢复链条：updated_at 心跳 → RequeueStale → 新 Worker 重抢（version 递增）→ 旧 Worker 写入被拦。
思考题
- 抢占重试上限 64 次后返回「无任务」，在高竞争场景下会否造成任务饿死？如何验证与改进？
- 若把乐观锁换成
SELECT ... FOR UPDATE SKIP LOCKED，吞吐与实现复杂度分别怎么变？
项目信息
- GitHub仓库：live-mixer
