Documentation
¶
Overview ¶
Package local 提供纯本地的 cron 调度器：按 cron 表达式触发任务。
零外部中间件、零 Redis 依赖——依赖面只有标准库与 github.com/oekit/pkg/timex/cron（cron 表达式解析复用 oekit 矩阵的实现， 不重复造轮子）。适合纯 Go 嵌入式服务形态（CGO_ENABLED=0 静态链接、 SQLite 落盘、零外部中间件）——这类形态用不了 Redis 后端， 但仍需要「到点自动干活」的能力。
与根包的关系 ¶
本包是 github.com/oekit/timertask 的子包（同模块），import 根包仅限 Clock 别名归一（#78，clock.go 嵌入根包定义——单一事实源）与模块内 单源 internal/gosafe（#168）；根包不 import 本包。需要「延迟队列」 （投递到未来某时刻）而不是「cron 触发」时，用根包的 LocalBackend。
语义（明确约定） ¶
并发度：
- Config.Concurrency 是全局并发度（同时执行的任务数上限），默认 1（严格串行）。
- 同一 spec 的触发始终串行（自身永不重叠）：上一轮没跑完时， 新的触发进入该 spec 的待执行队列。
错过触发点（进程停机、时钟跳变、上一轮未跑完）：
- Config.CatchUp = CatchUpSkip（默认）：只执行最近的一个触发点， 其余错过点全部跳过；待执行队列里的旧触发会被新触发替换 （同一 spec 最多积压 1 个）——不会因为任务慢而无限堆积。
- Config.CatchUp = CatchUpAll：所有错过的触发点按时间顺序逐一补跑。
进程退出：
- Stop 不打断正在执行的任务：任务的 ctx 不会被取消，Stop 等待它们结束； Stop 的等待上限由调用方给的 ctx 决定，超时返回 ctx.Err()， 任务继续跑完（Go 无法安全中断 goroutine）。
- Start 的 ctx 被取消时：调度立刻停止，任务的 ctx 同时被取消 （任务可自行感知并尽快退出），未开始的触发被丢弃。
- 两种情况下「未开始的触发」都会被丢弃——本地调度器不持久化。
重复注册：
- 同一个 spec 重复 Schedule 返回 ErrDuplicateSpec（不静默覆盖： 覆盖会悄悄丢掉原处理函数）。要换处理函数请先 Unschedule。
业务键形态（动态条目管理，#129） ¶
spec 键形态是**静态注册表**：键即表达式、重复注册报错、改表达式要先 Unschedule 旧键——调用方得自己记 spec→业务标识的映射。调度平台形态的 消费方（按任务 ID upsert、任务启停/改表达式时同步调度计划、多实例按 DB 期望集对账）请用业务键形态（与 spec 键形态并存不互通）：
sched.ScheduleFor("task-42", "0 9 * * 1-5", fn) // 按 key upsert
sched.UnscheduleFor("task-42") // 按 key 注销
sched.Reconcile([]local.Entry{ // 收敛到 DB 启用集
{Key: "task-42", Spec: "0 9 * * 1-5", Task: fn},
{Key: "task-43", Spec: "@every 30s", Task: fn2},
})
- 同 key 同 spec 重复 upsert 是**幂等空操作**（不重装、不重置触发基准）
——@every 解析为相对间隔（ConstantDelaySchedule），周期性重装会把
下次触发时间不断后移，间隔长于重装周期的任务永不执行。换任务函数
请先 UnscheduleFor（Task 不可比，指纹只含 spec）；
- Task 与 Reconcile 同口径（#155）：spec 未变的幂等空操作忽略 Task
（ScheduleFor(key, spec, nil) 的纯表达式 upsert 可用），需要安装/
重装（新 key / 改 spec）才要求 Task 非 nil；
- 同 key 改 spec ⇒ 摘旧装新（旧表达式立即停止触发）；
- Reconcile 单条 spec 失败不阻断其余条目的收敛（错误逐条汇总），
期望集内 key 重复 fail-closed（不做任何变更并报错）。
示例 ¶
sched, err := local.New(local.Config{Concurrency: 2})
if err != nil {
return err
}
if err := sched.Schedule("0 9 * * 1-5", func(ctx context.Context, at time.Time) error {
return report(ctx)
}); err != nil {
return err
}
go func() { _ = sched.Start(ctx) }()
defer sched.Stop(context.Background())
Index ¶
- Variables
- type CatchUpPolicy
- type Clock
- type Config
- type Entry
- type FakeClock
- type Scheduler
- func (s *Scheduler) Reconcile(entries []Entry) error
- func (s *Scheduler) Schedule(spec string, fn Task) error
- func (s *Scheduler) ScheduleFor(key, spec string, fn Task) error
- func (s *Scheduler) Start(ctx context.Context) error
- func (s *Scheduler) Stop(ctx context.Context) error
- func (s *Scheduler) Unschedule(spec string) bool
- func (s *Scheduler) UnscheduleFor(key string) bool
- type SystemClock
- type Task
- type Timer
Constants ¶
This section is empty.
Variables ¶
var ErrDuplicateSpec = errors.New("local: cron spec already scheduled")
ErrDuplicateSpec 表示同一个 cron 表达式被重复注册。
var ErrUnsatisfiableSpec = errors.New("local: cron spec can never fire")
ErrUnsatisfiableSpec 表示 cron 表达式语法合法但**永无触发点**（#123）， 例如 `0 0 31 2 *`（2 月永无 31 日）。
pkg/timex/cron 的 SpecSchedule.Next 对这类表达式静默返回零值——解析不报错， 但调度器永远算不出首个触发点：Schedule 返回 nil、Start 正常、日志安静、 任务永不执行，排障时唯一线索是「什么都没发生」。调度器正是 cron 库注释里 说的「消费 Next 前应判 IsZero」的那个消费方，必须把不可满足上升为用户可见 信号（与 ErrDuplicateSpec 的「不静默覆盖」同一防呆哲学）。
判定口径 = 「从注册时刻起 5 年内求不出首个触发点即报错」：这与调度循环 实际消费 Next 的搜索上限一致——超出该上限的表达式对本调度器而言**事实上** 永不触发（如「2 月 29 日且逢周日」这类超长间隔表达式，返回错误是如实 描述调度行为，而非过度拒绝）。
Functions ¶
This section is empty.
Types ¶
type CatchUpPolicy ¶
type CatchUpPolicy int
CatchUpPolicy 是错过触发点的补偿策略。
const ( // CatchUpSkip（默认）：只执行最近的一个触发点。 // // 适用于「状态收敛型」任务（对账、刷新、巡检）——历史触发点已过时， // 补跑没有意义，甚至有害（重复处理旧状态）。 CatchUpSkip CatchUpPolicy = iota // CatchUpAll：所有错过的触发点按时间顺序逐一补跑。 // // 适用于「计数型」任务（每次触发都代表一次必须发生的动作）。 // 注意补跑会积压：任务执行慢于触发周期时，积压会持续增长。 CatchUpAll )
type Clock ¶
type Clock interface {
timertask.Clock
// After 返回一个通道：在 d 之后收到一个时间信号。
// d <= 0 时必须返回「已就绪」的通道（调用方据此立即重算，不空等）。
// 注意：After 形态的 timer 不可停，循环等待 + 可被唤醒打断的场景请用
// NewTimer（#174）。
After(d time.Duration) <-chan time.Time
// NewTimer 返回可停 timer（#174）：C 通道在 d 之后收到信号（d <= 0 时
// 立即就绪），Stop 停止并回收 timer。调度循环醒来即停，不再遗弃挂起的
// timer。
NewTimer(d time.Duration) Timer
}
Clock 是时钟抽象，便于注入假时钟做确定性测试。
调度器的所有时间判断都走本接口：注册时的基准时间、触发点计算、等待下一个 触发点。注入 FakeClock 后，整个调度过程不需要 sleep 即可断言触发序列。 Clock 是时钟抽象（别名归一到根包 timertask.Clock——单一事实源，#78）。 本调度器还需要 After 与 NewTimer（等待下一个触发点），根包接口不含这两个 方法 ⇒ 保留本包扩展接口，嵌入根包定义。
NewTimer（#174）是循环等待场景的必用形态：After 每次调用都遗弃一个不可停 timer——调度循环被 wake（注册/注销任务）打断重算时，旧 timer 全部挂到原 到期点堆积（堆积量 = 打断频率 × 剩余等待）；NewTimer 返回可停面，醒来即停。
type Config ¶
type Config struct {
// Concurrency 全局任务并发度（同时执行的任务数上限），<=0 取 1（严格串行）。
//
// 同一 spec 的触发始终串行（自身不重叠）；不同 spec 之间按此上限并发。
Concurrency int
// CatchUp 错过触发点的补偿策略，零值取 CatchUpSkip。
CatchUp CatchUpPolicy
// Clock 时钟注入，nil 取 SystemClock{}；测试可注入 FakeClock。
Clock Clock
// Logger 日志器，nil 取 slog.Default()。
Logger *slog.Logger
// Location cron 表达式的时区，nil 取 time.Local。
Location *time.Location
// MaxCatchUp 单次补跑的最大触发点数，<=0 取 1000。
MaxCatchUp int
}
Config 是 Scheduler 配置。
type Entry ¶ added in v0.3.0
Entry 是业务键形态的期望条目（Reconcile 的输入，#129）。
Key 是业务侧的任务标识（任务 ID），同一调度器内唯一；Spec 是 cron 表达式 （与 Schedule 同口径：5/6 字段 + @every 等宏）；Task 在安装新 key / 重装 变更的 key 时使用，条目 spec 未变（幂等空操作）时被忽略——对账只需保 spec 与触发行为，任务函数以首次注册为准。
type FakeClock ¶
type FakeClock struct {
// contains filtered or unexported fields
}
FakeClock 是可手动推进的假时钟，用于确定性测试（不依赖 sleep）。
典型用法：
clock := local.NewFakeClock(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))
sched, _ := local.New(local.Config{Clock: clock, Location: time.UTC})
sched.Schedule("*/5 * * * * *", fn)
go sched.Start(ctx)
clock.Advance(5 * time.Second) // 触发点到达，fn 被调用
func (*FakeClock) After ¶
After 返回等待通道：假时钟推进到该等待的截止时间时，通道收到信号。
d <= 0 时立即返回已就绪的通道（不注册等待者）。 经 NewTimer 实现（#174 单源）；After 形态的 timer 不可停是该形态的 固有限制，循环等待场景请直接用 NewTimer。
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler 是基于 cron 表达式的本地调度器。
零外部中间件、零 Redis 依赖：只依赖标准库与 github.com/oekit/pkg/timex/cron （cron 表达式解析复用 oekit 矩阵的实现，不重复造轮子）。
典型用法：
sched, err := local.New(local.Config{Concurrency: 4})
if err != nil {
return err
}
if err := sched.Schedule("*/5 * * * * *", func(ctx context.Context, at time.Time) error {
return doSomething(ctx)
}); err != nil {
return err
}
go func() { _ = sched.Start(ctx) }()
defer sched.Stop(context.Background())
func (*Scheduler) Reconcile ¶ added in v0.3.0
Reconcile 把业务键形态的任务收敛到期望集（#129）：期望集里的条目按 ScheduleFor 语义增量 upsert（spec 变化重装、未变幂等空操作），已注册但 不在期望集里的 key 注销。多实例共用同一 DB 期望集、周期对账的场景下， 期望集未变时 Reconcile 是**空操作**（不重装、不重置 @every 触发基准）。
对账语义（与调度平台消费方对齐）：
- 启停 / 变更的表达式由期望集表达：启用的任务出现在期望集里，停用的 不出现（自动注销）——调用方无须 diff，库侧收敛；
- 单条 spec 解析失败 / 不可满足**不阻断**其余条目的收敛（一个坏任务 不应阻止其余任务对账），失败逐条汇总进返回错误（errors.Join）；
- 期望集内 key 重复 ⇒ 「以哪条为准」存在歧义，fail-closed：本次调用 不做任何变更并返回错误；
- 收敛原子：期间并发的 ScheduleFor/UnscheduleFor 不会看到中间态 （全部变更在单次持锁内完成）。
errors 全部成功时返回 nil；调用方通常应串行化对账（单 goroutine 周期 触发），与多实例并发对账同一期望集是安全的（收敛结果一致）。
func (*Scheduler) Schedule ¶
Schedule 注册一个 cron 任务。
spec 交给 github.com/oekit/pkg/timex/cron 解析：支持 5 字段（分 时 日 月 周） 与 6 字段（秒 分 时 日 月 周）格式，以及 @every/@daily/@hourly 等宏。
同一个 spec 重复注册返回 ErrDuplicateSpec（不静默覆盖：覆盖会悄悄丢掉 原处理函数，且调用方难以察觉）。要换处理函数请先 Unschedule。
语法合法但永无触发点的表达式（如 `0 0 31 2 *`）返回 ErrUnsatisfiableSpec（#123，不静默接受）：此前这类 spec 注册成功后永不 触发且无任何信号。判定方式是注册前用注入时钟试算一次首个触发点—— 求不出（Next 返回零值）即报错，与调度循环实际的消费口径一致。
func (*Scheduler) ScheduleFor ¶ added in v0.3.0
ScheduleFor 按业务键 upsert 一个 cron 任务（#129）。
与 spec 键形态（Schedule/Unschedule，键即表达式、重复注册 ErrDuplicateSpec） 并存不互通，面向「调度平台」的动态形态：按任务 ID 管理条目，任务改表达式 时同步调度计划，调用方无须自己维护 spec→taskID 映射。
语义（指纹幂等，zeus 实证的同款防呆）：
- key 未注册 ⇒ 安装（Task 必填）；
- key 已注册且 spec 未变 ⇒ **幂等空操作**——不重装、不重置触发基准。 重装会把 entry 的 lastFire 重置到此刻：@every 解析为 ConstantDelaySchedule（相对基准的固定间隔），「周期性重装」会把间隔 长于重装周期的任务的下次触发时间不断往后推——任务永不执行。因此 **换任务函数（Task 不可比）请先 UnscheduleFor 再 ScheduleFor**， 同 spec 重复调用本函数不会生效；
- key 已注册且 spec 变化 ⇒ 摘旧装新（Task 必填）：旧表达式立即停止 触发（正在执行的旧任务不打断，与 Unschedule 语义一致），新表达式从 注册时刻起算。
Task 与 Reconcile 同口径（#155）：仅在「需要安装/重装」时使用，spec 未变 的幂等空操作忽略 Task（fn 为 nil 亦放行，任务函数以首次注册为准）——纯 表达式 upsert（ScheduleFor(key, spec, nil)）由此可用；需要安装/重装而 fn 为 nil 时返回错误。
同一 spec 可挂多个业务键（两个任务共用同一表达式，各自触发）；key 键与 spec 键互不可见——同一 spec 在两种形态下各注册一次会各自触发一次，混用 前请确认这正是想要的。
spec 解析失败 / 语法合法但永无触发点（ErrUnsatisfiableSpec）/ 调度器已 停止时返回错误，此时该 key 的既有注册不受影响。Start 之前调用亦可： 条目先入表，Start 时随全部条目一起启动。
func (*Scheduler) Start ¶
Start 启动调度循环并阻塞，直到 ctx 取消或 Stop 被调用。
返回时调度器已停止：不再产生新触发；正在执行的任务不会被强制中断 （任务可通过 ctx 感知取消并自行退出）。重复调用返回错误。
返回值（#99）：Stop 触发的正常停机返回 nil——此前任何路径都返回非 nil （Stop 路径硬编码 context.Canceled），按惯用 `if err := sched.Start(ctx); err != nil { log.Fatal }` 写的调用方会把 正常停机当失败处理。仅 ctx 取消时返回 ctx.Err()。
func (*Scheduler) Stop ¶
Stop 停止调度并等待正在执行的任务结束。
语义：
- 未开始的触发被丢弃；
- 正在执行的任务不被打断（任务的 ctx 不被取消），Stop 最多等待到 ctx 截止；
- ctx 超时返回 ctx.Err()，任务继续跑完（Go 无法安全中断 goroutine）。
可重复调用。
func (*Scheduler) Unschedule ¶
Unschedule 注销一个 cron 任务，返回是否确实移除了该任务。
正在执行的任务不会被中断（任务体可自行感知 ctx 取消）；未开始的触发被丢弃。
func (*Scheduler) UnscheduleFor ¶ added in v0.3.0
UnscheduleFor 按业务键注销一个 cron 任务，返回是否确实移除了该任务。 语义与 Unschedule 一致：正在执行的任务不会被中断（任务体可自行感知 ctx 取消）；未开始的触发被丢弃。
type SystemClock ¶
type SystemClock struct{}
SystemClock 是真实时钟（time.Now + time.Timer）。
func (SystemClock) After ¶
func (SystemClock) After(d time.Duration) <-chan time.Time
After 返回一个在 d 之后收到信号的通道。
Source Files