第35章：全局调度器与暂存清理
业务 Worker（ASR、切片、草稿）各自内建了调度循环，而与任务队列无关的周期性工作——staging 目录清理——交给独立的 internal/scheduler 包。本章解析这个小而完整的调度器，以及它驱动的主要消费者 webroot.CleanupStaging / CleanupASRStaging。
流程
调度器采用"每 Job 一个 goroutine + 独立 ticker"的朴素模型。注册在 Start 之前完成，Start 用 sync.Once 保证幂等；循环开始时立即执行一次再进入周期。两个清理 Job 在组合根 main.go 中注册，间隔共用 cfg.Web.StagingCleanupInterval()，配额分别取 StagingMaxDirs 与 ASRStagingMaxDirs。
flowchart TD
A[main.go 组合根] --> B[scheduler.New]
B --> C[Register cleanup-staging<br/>保留最新 N 个任务目录]
B --> D[Register cleanup-asr-staging<br/>asr 子树下独立配额]
C & D --> E[sched.Start 启动即执行一轮]
E --> F[每 Job 独立 goroutine + ticker]
F --> G[runSafe: recover 包裹]
G --> H{staging 子目录数 > keep?}
H -- 是 --> I[按 mtime 降序<br/>删除第 keep 名之后]
H -- 否 --> J[本轮 no-op]
I --> K[单个删除失败<br/>记 firstErr 继续]
清理算法：os.ReadDir 读一级子目录（跳过 asr 特殊目录与普通文件），mtime 降序排序（相同 mtime 按路径名倒序打破平局），保留前 keep 个、os.RemoveAll 其余。目录不存在视为成功——首次运行无需预建。
实现
调度器细节：Job.Interval <= 0 或 Run == nil 时只记 Warn 跳过，不 panic——配置错误降级为日志而非宕机。runSafe 用 defer recover 兜住 panic，防止一个清理任务崩掉整个进程；循环体内对 ctx.Done 的响应让 SIGTERM 优雅退出路径不被长任务阻塞。
webroot 清理的容错分级：读目录失败且非 NotExist → 返回错误；单个子目录 Info() 失败或 RemoveAll 失败 → 记 firstErr 后继续处理其余目录。聚合错误而非中断，保证一次调度尽可能清出空间——磁盘告急时"删掉 7/10 个"远好于"删 0 个"。
为什么 CleanupStaging 要跳过 asr 子目录：staging/ 下按 task_id 命名的目录是"槽位"，但 staging/asr/ 是整棵 ASR 调试树的根，若不跳过，它会被当成一个普通槽位参与 mtime 排序——一旦它恰好是最旧的，整棵 ASR 调试数据会被一次 RemoveAll 掀翻。ASR 子树因此有自己的入口 CleanupASRStaging 与独立配额，两层清理互不知晓又互不伤害。
返回值 (removed, err) 的消费方式：main.go 的闭包只在 cleanErr != nil 时 Warn、removed > 0 时 Info——正常轮次零日志输出，日志量与"实际发生了清理"的事件数成正比。
📌 设计决策
- 调度器与业务 Worker 分离：任务型 Worker 需要"抢占 + 唤醒"语义（DB 队列），而定时 Job 只需要 ticker——两套循环强行统一会增加抽象税。scheduler 只有 90 行，不引入 cron 表达式、不持久化、不支持动态增删。
- 启动即执行：进程重启后第一轮清理立刻发生，不留下"重启后 30 分钟磁盘无人看管"的窗口。
- 按数量而非按时间清理："保留最新 N 个"比"删除 N 天前的"更可预测——任务量高峰时配额自动收缩时间窗，磁盘上限恒定，适合容量规划。
- 错误聚合不中断：清理是尽力而为的运维动作，部分失败不应让整轮作废。
代码示例
片段 1 展示调度器本体（摘自 internal/scheduler/scheduler.go，节选）：
// Job 周期性任务定义。
type Job struct {
Name string // 任务名，用于日志
Interval time.Duration // 执行间隔；<=0 时该 Job 不会被调度
Run func(ctx context.Context) // 应尽快返回
}
// Start 为每个合法 Job 启动独立循环；可安全重复调用（仅首次生效）。
func (s *Scheduler) Start(ctx context.Context) {
s.startOnce.Do(func() {
for _, job := range s.jobs {
if job.Interval <= 0 || job.Run == nil {
s.logger.Warn("跳过无效定时任务", zap.String("job", job.Name))
continue
}
j := job
go s.loop(ctx, j)
}
})
}
func (s *Scheduler) loop(ctx context.Context, job Job) {
s.runSafe(ctx, job) // 启动即执行一次
ticker := time.NewTicker(job.Interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
s.runSafe(ctx, job)
}
}
}
func (s *Scheduler) runSafe(ctx context.Context, job Job) {
defer func() {
if rec := recover(); rec != nil {
s.logger.Error("定时任务 panic", zap.String("job", job.Name), zap.Any("recover", rec))
}
}()
job.Run(ctx)
}
片段 2 展示清理核心（摘自 internal/pkg/webroot/cleanup.go，节选）：
// cleanupStagingChildren 清理 parentDir 下一级子目录，按 mtime 保留最新 keep 个。
// skipName 非空时跳过该名称的一级子目录（不计入配额、不删除）。
func cleanupStagingChildren(parentDir string, keep int, skipName string) (removed int, err error) {
entries, readErr := os.ReadDir(parentDir)
if readErr != nil {
if os.IsNotExist(readErr) {
return 0, nil // 目录不存在视为成功
}
return 0, fmt.Errorf("读取目录失败 %s: %w", parentDir, readErr)
}
dirs := make([]stagingDir, 0, len(entries))
var firstErr error
for _, entry := range entries {
if !entry.IsDir() {
continue
}
if skipName != "" && entry.Name() == skipName {
continue // 关键：asr 整棵子树不参与本轮配额
}
info, infoErr := entry.Info()
if infoErr != nil {
if firstErr == nil {
firstErr = fmt.Errorf("读取目录信息失败 %s: %w", entry.Name(), infoErr)
}
continue
}
dirs = append(dirs, stagingDir{
path: filepath.Join(parentDir, entry.Name()), modTime: info.ModTime()})
}
if len(dirs) <= keep {
return 0, firstErr
}
// 最新在前：保留前 keep 个，删除其余（mtime 更早的）
sort.Slice(dirs, func(i, j int) bool {
if dirs[i].modTime.Equal(dirs[j].modTime) {
return dirs[i].path > dirs[j].path
}
return dirs[i].modTime.After(dirs[j].modTime)
})
for _, d := range dirs[keep:] {
if rmErr := os.RemoveAll(d.path); rmErr != nil {
if firstErr == nil {
firstErr = fmt.Errorf("删除目录失败 %s: %w", d.path, rmErr)
}
continue // 部分失败继续
}
removed++
}
return removed, firstErr
}
小结
调度器用 90 行换来一个不依赖任何外部组件的定时框架；webroot 清理则示范了"数量配额 + mtime 排序 + 聚合容错"的磁盘治理套路。二者组合，让 staging 调试产物始终处于有界状态。
思考题
- 若清理间隔内新任务目录暴增（如批量拆分成 50 个任务），数量配额会不会误删正在运行任务的 staging？需要与 Worker 生命周期联动吗？如何设计"活跃任务目录豁免"？
Scheduler.Register没有做并发保护且要求在Start前调用——这种"初始化阶段单线程"的隐式契约，在什么情况下会成为陷阱？
项目信息
- GitHub仓库：Chyona/live-mixer
