Documentation
¶
Overview ¶
Package cron 提供多实例动态 cron 装配面（#159）：运行期增删改 cron 任务 （指纹 diff 增量重装，不重置未变化项触发点）+ 多实例执行互斥（租约型， distlock 可选注入）+ 运行台账（执行历史/结果可观测）。
触发引擎复用同模块 local.Scheduler（表达式解析/catch-up/并发度/假时钟 都是既有基建），本包在其上补三块缺口：DB 计划的期望集收敛、跨实例 触发互斥、运行台账——此前下游 3 处各自手搓调度器，锁 TTL 预算公式这类 易错口径散落各处（见 RetryBudget）。
装配示例（调度平台形态，多实例共用 DB 期望集）：
sched := cron.New(exec,
cron.WithMutex(distAdapter{locker}), // redisx/distlock 等，见 Mutex
cron.WithInstance("task-1"), // 台账/互斥归因
)
// 库替你管循环：启动清扫 → 首轮对账 → 按周期收敛，ctx 取消停机
if err := sched.StartReconcile(ctx, 30*time.Second, listFromDB); err != nil {
return err
}
defer sched.Close(context.Background())
也可自带循环周期调 Reconcile（同样自动拉起触发引擎）：
sched.Reconcile(ctx, []cron.Plan{{ID: "t1", Spec: "@every 1m", Enabled: true}})
互斥语义（触发点恰好一次） ¶
互斥键 = `timertask/cron/exec/<planID>/<trigger>`（trigger 为触发点 RFC3339 标识，秒精度）——同一触发点跨实例同键裁决，只跑一次；让位方 记台账（Skipped）。持有即租约（TTL = RetryBudget.LockTTL()，到期自动 接管，崩溃实例不尸占），认领租约不随执行完成释放——兼作「slot 已认领」 标记，迟到重复触发在租约有效期内一律让位；互斥设施故障 fail-closed 跳过 （宁可漏跑一轮可补发，不可双跑）。
与 local 包的关系 ¶
本包是 github.com/oekit/timertask 的同模块子包，import local 复用触发 引擎；需要裸 cron 触发（无计划表/无互斥/无台账）直接用 local.Scheduler。
Index ¶
- Constants
- Variables
- type CronOption
- func WithCatchUp(p local.CatchUpPolicy) CronOption
- func WithClock(clock local.Clock) CronOption
- func WithConcurrency(n int) CronOption
- func WithInstance(id string) CronOption
- func WithLedger(l Ledger) CronOption
- func WithLocation(loc *time.Location) CronOption
- func WithLogger(l *slog.Logger) CronOption
- func WithMutex(m Mutex) CronOption
- func WithRetryBudget(b RetryBudget) CronOption
- func WithRunCapacity(n int) CronOption
- func WithStaleSweep(fn StaleSweepFunc) CronOption
- type Executor
- type Ledger
- type LocalMutex
- type MemLedger
- type Mutex
- type Plan
- type RetryBudget
- type RunRecord
- type Scheduler
- func (s *Scheduler) Close(ctx context.Context) error
- func (s *Scheduler) Reconcile(ctx context.Context, plans []Plan) error
- func (s *Scheduler) Remove(_ context.Context, id string) bool
- func (s *Scheduler) RunHistory(planID string) []RunRecord
- func (s *Scheduler) StartReconcile(ctx context.Context, interval time.Duration, listFn func() ([]Plan, error)) error
- func (s *Scheduler) Upsert(ctx context.Context, p Plan) error
- type StaleSweepFunc
Constants ¶
const ( // DefaultBudgetAttempts 默认总尝试次数（含首次）。 DefaultBudgetAttempts = 3 // DefaultBudgetTimeout 默认单次尝试上限。 DefaultBudgetTimeout = 60 * time.Second // DefaultBudgetBackoff 默认相邻尝试间退避。 DefaultBudgetBackoff = 5 * time.Second // DefaultBudgetSlack 默认余量（时钟漂移/调度延迟）。 DefaultBudgetSlack = 30 * time.Second )
默认重试预算：对齐根包延迟任务的默认量级（WithRetry 默认 3 次 / 5s 间隔、 WithVisibilityTimeout 默认 60s 租约），沿用同一族默认值让「锁 TTL 预算」 与「任务执行预算」在同一量纲上起步。
const DefaultRunCapacity = 256
DefaultRunCapacity 是内置运行台账（MemLedger）的默认容量（有界环形， 超容最旧记录被覆盖——台账是可观测面不是持久化，持久化走 WithLedger）。
Variables ¶
var ErrLockNotAcquired = errors.New("cron: lock not acquired")
ErrLockNotAcquired 表示锁被其他实例持有（非阻塞争抢失败）。 Mutex.TryLock 的占用哨兵：errors.Is 判定；与「互斥设施故障」（其它 error） 区分——占用是常态（跳过本次触发即可），故障要进台账告警。
适配 redisx/distlock 时把 distlock.ErrLockNotAcquired 映射到本哨兵 （见 Mutex 的 godoc 适配示例）。
var ErrSchedulerStopped = errors.New("cron: scheduler stopped")
ErrSchedulerStopped 表示实例已停机（Close 已调用或 StartReconcile 的 ctx 已取消），进入终态——对齐根包 ErrBackendStopped 的实例一次性纪律（#124）： 停机后 Reconcile/Upsert/StartReconcile 返回本错误，不静默哑火。需要重启 请重建实例。
Functions ¶
This section is empty.
Types ¶
type CronOption ¶
type CronOption func(*cronOptions)
CronOption 配置 Scheduler。命名遵循仓 AGENTS 的 XxxOption 纪律 （禁止裸 Option，多包 co-import 冲突）——issue 建议形态的 `...Option` 相应加前缀。
func WithCatchUp ¶
func WithCatchUp(p local.CatchUpPolicy) CronOption
WithCatchUp 设置错过触发点的补偿策略（透传 local.CatchUpPolicy）。
func WithClock ¶
func WithClock(clock local.Clock) CronOption
WithClock 注入时钟（触发点计算/租约到期/台账时刻同源），nil 取 local.SystemClock{}；测试注入 local.FakeClock 确定性驱动（与 local.Scheduler 的 Clock 注入同一形态，同一假时钟两层共用）。
func WithConcurrency ¶
func WithConcurrency(n int) CronOption
WithConcurrency 设置全局任务并发度（透传 local.Config.Concurrency， <=0 取 1 严格串行）。
func WithLedger ¶
func WithLedger(l Ledger) CronOption
WithLedger 注入外置运行台账（DB 落库/审计流等），与内置 MemLedger 并联 记录；nil = 只用内置（显式回退默认值，不算非法）。
func WithLocation ¶
func WithLocation(loc *time.Location) CronOption
WithLocation 设置 cron 表达式时区（透传 local.Config.Location，nil 取 time.Local）。
func WithLogger ¶
func WithLogger(l *slog.Logger) CronOption
WithLogger 注入日志器，nil 取 slog.Default()（根模块纯 stdlib，与 local.Scheduler 同口径）。
func WithMutex ¶
func WithMutex(m Mutex) CronOption
WithMutex 注入跨实例执行互斥（distlock 等，见 Mutex 的适配示例）。 不设置时只有进程内 slot 互斥（单实例部署即够；多实例部署不注入会双跑—— 这是显式接线项，不给静默默认）。
func WithRetryBudget ¶
func WithRetryBudget(b RetryBudget) CronOption
WithRetryBudget 设置执行重试预算（互斥锁 TTL 预算公式的唯一出处， 见 RetryBudget）。
func WithRunCapacity ¶
func WithRunCapacity(n int) CronOption
WithRunCapacity 设置内置运行台账容量（有界环形，<=0 取 DefaultRunCapacity）。
func WithStaleSweep ¶
func WithStaleSweep(fn StaleSweepFunc) CronOption
WithStaleSweep 设置启动清扫钩子（见 StaleSweepFunc）。
type Executor ¶
Executor 是计划触发的执行函数（#159 建议形态原样交付）。
planID 为计划 ID；trigger 为触发点标识（RFC3339，秒精度）——幂等键/ 互斥键/台账键三同源（execLockKey/RunRecord 同值），业务可直接拿去做 幂等去重。返回错误只记台账与日志（local.Scheduler「任务失败不重试」 同口径——需要重试语义请在任务内部实现）。
type Ledger ¶
Ledger 是运行台账的最小契约（外置落库缝：DB 表/审计流等）。
Record 不返回错误：台账写失败不得影响主链路（外置实现自行兜底重试）； Scheduler 侧对 panic 也做收编（台账故障不得打死触发）。每次触发一条 记录（成功/失败/让位都记——「谁在什么时候跑了哪一档、结果如何」是排障 与「恰好一次」核对面）。
type LocalMutex ¶
type LocalMutex struct {
// contains filtered or unexported fields
}
LocalMutex 是进程内 Mutex 实现（租约表 + TTL 惰性回收，零外部依赖）：
- 单实例部署的默认互斥（Scheduler 内建一份守触发 slot——摘旧装新的在飞 旧触发与新条目补跑可撞同一 slot，进程内先裁决，不劳分布式锁）；
- 多实例测试的共享假件：两个 Scheduler 共享同一个 LocalMutex 即可复现 跨实例仲裁（见 cron_test 的互斥并发场景）。
多实例生产部署请注入分布式实现（redisx/distlock 等，经 Mutex 缝）。
func NewLocalMutex ¶
func NewLocalMutex(clock local.Clock) *LocalMutex
NewLocalMutex 创建进程内互斥。clock 为租约到期判定的时钟，nil 取 local.SystemClock{}（测试注入 FakeClock 可确定性推进 TTL）。
type MemLedger ¶
type MemLedger struct {
// contains filtered or unexported fields
}
MemLedger 是内置有界运行台账（环形覆盖，New 缺省挂载）：RunHistory 的 读取面，永不无界增长。
func NewMemLedger ¶
NewMemLedger 创建容量为 capacity 的有界台账（<=0 取 DefaultRunCapacity）。
type Mutex ¶
type Mutex interface {
// TryLock 非阻塞争抢 key 的执行租约，ttl 为租约时长。
//
// - 成功 ⇒ (ownerID, nil)，ownerID 非空（后续 Unlock 凭它释放）；
// - 被占用 ⇒ ("", ErrLockNotAcquired)（errors.Is 判定）；
// - 其它 error ⇒ 互斥设施故障（调度面 fail-closed 跳过本次执行，
// 台账记错——宁可漏跑一轮（可补发）不可多实例双跑）。
TryLock(ctx context.Context, key string, ttl time.Duration) (ownerID string, err error)
// Unlock 释放 key 的租约，ownerID 必须匹配。锁不存在或已易主（TTL 过期
// 后被接管）按各自后端语义返回错误——调度面只记日志不改执行结果
// （执行已结束，租约易主是租约模型的正常演进）。
Unlock(ctx context.Context, key string, ownerID string) error
}
Mutex 是多实例执行互斥的最小契约（#159）：**可选依赖 distlock 的缝**—— 仓内不引 distlock/redisx 依赖（根模块 require 只允许 pkg），分布式锁 实现经本接口注入。语义与仓内 at-least-once 租约同型（visibility timeout 租约集）：持有即租约，TTL 到期自动可被他实例接管（崩溃实例不尸占）， Unlock 提前释放。
执行互斥按「slot 认领」使用（fire 的消费形态）：触发点的租约**不**在 执行完成时提前释放，认领标记保留到 TTL 到期——同触发点的迟到重复触发 在租约有效期内一律让位（「触发点恰好一次」），执行完成即放锁会让迟到 方重跑同一 slot。确需提前释放的消费方（长租约省存储）仍可用 Unlock， 但那会重新打开「迟到触发重跑」的口子。
TryLock/Unlock 与 redisx/distlock.Locker 同形（其 Lock 阻塞形态调度面 用不上，不进本契约）——*distlock.RedisLocker 结构上即满足本接口的 方法集，但占用哨兵不同名，语义适配包一层即可：
type distAdapter struct{ l distlock.Locker }
func (a distAdapter) TryLock(ctx context.Context, key string, ttl time.Duration) (string, error) {
id, err := a.l.TryLock(ctx, key, ttl)
if errors.Is(err, distlock.ErrLockNotAcquired) {
return "", cron.ErrLockNotAcquired
}
return id, err
}
func (a distAdapter) Unlock(ctx context.Context, key, owner string) error {
return a.l.Unlock(ctx, key, owner)
}
type Plan ¶
type Plan struct {
// ID 计划唯一 ID（业务任务 ID）。空 ID 使 Reconcile 整体 fail-closed。
ID string
// Spec cron 表达式（交 oekit/pkg/timex/cron 解析：5/6 字段 + @every
// 等宏，与 local.Scheduler 同口径）。
Spec string
// Fingerprint 变更指纹——增量重装的唯一 diff 单元（对齐 zeus/task 的
// taskFingerprint：表达式+状态+updatedAt 的哈希）：
//
// - 指纹未变 ⇒ 幂等空操作（不重装、不重置触发基准——防 @every 被
// 周期重装不断后移）；
// - 指纹变化 ⇒ 摘旧装新（计划被编辑过，从新基准起算）；
// - 留空 ⇒ 派生指纹 = Spec（库侧可见的行为面只有表达式）。
//
// 注意指纹构成是调用方的契约选择：含 updatedAt 时「改任务元数据」
// 也会重装、@every 触发基准随之重置（刻意语义——编辑过就该从新基准
// 起算）；只想「行为变更才重装」就把指纹定为 表达式+状态 的哈希。
Fingerprint string
// Enabled false = 停用：出现在期望集里时按注销收敛（不出现在期望集里
// 同效——两种停用表达都收敛为不触发）。
Enabled bool
}
Plan 是一条期望计划（DB 计划表单条的装配面镜像，#159 建议形态原样交付）。
type RetryBudget ¶
type RetryBudget struct {
// Attempts 总尝试次数（含首次），<1 归一为 1。
Attempts int
// Timeout 单次尝试（执行 + 尝试内等待）上限，<=0 归 0。
Timeout time.Duration
// Backoff 相邻尝试间的固定退避，<=0 归 0。
Backoff time.Duration
// Slack 余量（时钟漂移/调度延迟），<=0 取 DefaultBudgetSlack。
Slack time.Duration
}
RetryBudget 是任务执行的重试预算，也是**互斥锁 TTL 预算公式的唯一出处** （#159：与 zeus/task 的 execLockTTL 同口径收编）：
锁 TTL = 全部尝试 + 全部退避 + 余量
= Attempts*Timeout + TotalBackoff() + Slack
此前该预算散在下游各自手算，口径漂移的后果是互斥提前过期——同 slot 被 他实例接管、双跑破坏「触发点恰好一次」。公式必须单点：改预算只改这里。
零值可用（Normalized 归一：Attempts < 1 取 1、负时长归 0、Slack <= 0 取 DefaultBudgetSlack——零值透传的 Slack 才是缺省，不是「不要余量」；确需 零余量显式给 1ns 这类正值）。
func (RetryBudget) LockTTL ¶
func (b RetryBudget) LockTTL() time.Duration
LockTTL 返回互斥锁的租约时长（唯一出处，见类型注释）： Attempts*Timeout + TotalBackoff() + Slack。
func (RetryBudget) Normalized ¶
func (b RetryBudget) Normalized() RetryBudget
Normalized 返回归一后的预算（负值钳制 min/max 内建；Slack 零值/负值取 DefaultBudgetSlack——显式更小的正值照用）。
func (RetryBudget) TotalBackoff ¶
func (b RetryBudget) TotalBackoff() time.Duration
TotalBackoff 返回全部退避之和 = (Attempts-1)*Backoff（首次与末次之间 共 Attempts-1 个退避间隔）。
type RunRecord ¶
type RunRecord struct {
// PlanID 计划 ID。
PlanID string
// Trigger 触发点标识（RFC3339 秒精度，与互斥键同源——幂等键/互斥键/
// 台账键三同源）。
Trigger string
// Instance 执行实例标识（WithInstance，缺省主机名）——多实例归因。
Instance string
// StartedAt 执行开始时刻（跳过记录为跳过判定时刻）。
StartedAt time.Time
// EndedAt 执行结束时刻（跳过记录同 StartedAt）。
EndedAt time.Time
// Err 执行结果（nil=成功），语义见类型注释。
Err error
// Skipped 本次触发是否未执行（互斥让位/故障 fail-closed）。
Skipped bool
}
RunRecord 是一次触发的运行台账记录（#159「运行台账：执行历史/结果可观测」）。
状态读法（Skipped/Err 组合语义）：
- Skipped=false、Err=nil ⇒ 执行成功；
- Skipped=false、Err!=nil ⇒ 执行失败（含 executor panic 收编）；
- Skipped=true、Err=nil ⇒ 互斥让位（同触发点由他实例执行，跨实例 恰好一次的正常形态）；
- Skipped=true、Err!=nil ⇒ 互斥设施故障，fail-closed 未执行（告警面）。
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler 是多实例动态 cron 装配面（#159）：local.Scheduler 触发引擎 + 指纹 diff 增量装配 + 触发互斥 + 运行台账。零值不可用，New 构造。
func New ¶
func New(exec Executor, opts ...CronOption) *Scheduler
New 创建调度器。构造期 fail-fast（#97 同哲学）：nil Executor、非法 CatchUp 等配置错误在启动期炸响，不带病运行。
func (*Scheduler) Close ¶
Close 停止调度并等待在飞触发收尾（组件生命周期统一 Close(ctx) error + sync.Once 幂等）：对账循环与触发引擎同停，正在执行的任务不打断（等待到 ctx 截止，local.Stop 同语义）。ctx 超时返回 ctx.Err()，任务继续跑完。 可重复调用；未启动过触发引擎时立即返回。
func (*Scheduler) Reconcile ¶
Reconcile 把期望计划集收敛到触发引擎（#159「指纹 diff 增量重装」）： 期望集里的启用条目按指纹增量 upsert（指纹未变幂等空操作、变化摘旧装新）， 停用条目与已装但不在期望集里的条目注销。多实例共用同一 DB 期望集、 周期对账的场景下，期望集未变时 Reconcile 是**空操作**（不重装、不重置 @every 触发基准）。
对账语义（与 local.Reconcile 同骨架）：
- 启停/变更由期望集表达：启用的条目出现在期望集里（Enabled=true）， 停用的不出现或 Enabled=false（同效注销）；
- 单条失败（坏 spec 等）**不阻断**其余条目收敛，失败逐条汇总进返回 错误（errors.Join）；
- 期望集内 ID 重复 / 空 ID ⇒ 「以哪条为准」存在歧义，fail-closed： 本次调用不做任何变更并返回错误；
- 首次调用自动拉起触发引擎（StartReconcile 形态亦然）。
errors 全部成功时返回 nil；调用方通常应串行化对账（单 goroutine 周期 触发）——并发调用无数据竞争，但收敛次序不保证。停机后返回 ErrSchedulerStopped（实例一次性，#124）。
func (*Scheduler) Remove ¶
Remove 按计划 ID 注销（运行期删），返回是否确实移除了触发条目。正在 执行的触发不打断（执行已进台账），未开始的触发丢弃（local.UnscheduleFor 同语义）。
func (*Scheduler) RunHistory ¶
RunHistory 返回内置台账记录（时间序旧→新）；planID 非空时只返回该计划的。
func (*Scheduler) StartReconcile ¶
func (s *Scheduler) StartReconcile(ctx context.Context, interval time.Duration, listFn func() ([]Plan, error)) error
StartReconcile 启动「启动清扫 + 周期对账 + 触发引擎」并阻塞（#159 `StartReconcile`；返回 error 为生命周期纪律，退出语义与 local.Start 同款——#99）：
- 启动清扫（WithStaleSweep）先于首次对账——zeus/task 的 FailStaleRunning 启动清扫 + 清扫后补发事件同序，补发在钩子内完成；
- 立即 listFn() → Reconcile 首轮对账，随后按 interval 周期收敛 （listFn 失败/单条收敛失败只记日志不中断循环——多实例对账要韧性）；
- ctx 取消 ⇒ 停机（触发引擎同停）返回 ctx.Err()；Close 触发的正常 停机返回 nil。
周期等待走注入时钟（clock.NewTimer 可停 timer，#174）——FakeClock 测试下 对账节奏确定性可控，不真睡。自带循环的调用方直接周期调 Reconcile 即可。
Source Files