136 lines
4.9 KiB
Go
136 lines
4.9 KiB
Go
package schedule
|
||
|
||
import (
|
||
"errors"
|
||
"time"
|
||
)
|
||
|
||
// Policy 显式规定漏跑与重叠；窗口、批量和持久化扫描间隔共同限制补跑。
|
||
type Policy struct {
|
||
Misfire string `json:"misfire"` // Misfire SKIP、COALESCE 或 CATCH_UP，必须明确提供。
|
||
Overlap string `json:"overlap"` // Overlap PARALLEL、SERIAL 或 SKIP，必须明确提供。
|
||
CatchupLimit uint32 `json:"catchup_limit"` // CatchupLimit 每批最多生成的任务数，1–100；串行最多一个。
|
||
CatchupWindowMS uint64 `json:"catchup_window_ms"` // CatchupWindowMS 最多保留多久的未处理触发点，1000 毫秒–30 天。
|
||
GraceMS uint64 `json:"grace_ms"` // GraceMS 允许正常派发延迟的毫秒数，1000–60000，不超过窗口。
|
||
}
|
||
|
||
func (p Policy) Validate() error {
|
||
if p.Misfire != "SKIP" && p.Misfire != "COALESCE" && p.Misfire != "CATCH_UP" {
|
||
return errors.New("explicit misfire policy required")
|
||
}
|
||
if p.Overlap != "PARALLEL" && p.Overlap != "SERIAL" && p.Overlap != "SKIP" {
|
||
return errors.New("explicit overlap policy required")
|
||
}
|
||
if p.CatchupLimit < 1 || p.CatchupLimit > 100 || p.CatchupWindowMS < 1000 || p.CatchupWindowMS > 2592000000 || p.GraceMS < 1000 || p.GraceMS > 60000 || p.GraceMS > p.CatchupWindowMS {
|
||
return errors.New("invalid bounded catchup policy")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Decision 描述单点或连续范围的处理；范围为 [ScheduledAt,CoveredUntil)，结束不包含。
|
||
type Decision struct {
|
||
ScheduledAt time.Time // ScheduledAt 范围中的首个原触发点，作为唯一去重身份。
|
||
CoveredUntil *time.Time // CoveredUntil 合并/跳过范围的排除边界；单点为空。
|
||
Kind string // Kind FIRED、COALESCED 或 SKIPPED。
|
||
Reason string // Reason 稳定处理分类，不包含业务参数。
|
||
}
|
||
|
||
// Batch 是同一个不可变版本的一批决定，必须与任务及进度原子持久化。
|
||
type Batch struct {
|
||
Decisions []Decision // Decisions 有界决策，最多 CatchupLimit+2 条。
|
||
Next *time.Time // Next 当前版本下一触发点；nil 表示版本区间结束。
|
||
Blocked bool // Blocked SERIAL 正在等待已有非终态任务，不能跨越该点。
|
||
}
|
||
|
||
// Plan 不读取墙上时钟。now/active 由资源锁定后的数据库事实提供。
|
||
// before 是下一版本的生效排除边界，旧积压仍按旧版本计算。
|
||
func Plan(c *Calendar, p Policy, due, now time.Time, before *time.Time, active bool) (Batch, error) {
|
||
out := Batch{Decisions: []Decision{}}
|
||
if err := p.Validate(); err != nil {
|
||
return out, err
|
||
}
|
||
if due.IsZero() || now.IsZero() {
|
||
return out, errors.New("schedule facts required")
|
||
}
|
||
actual, err := c.AtOrAfter(due)
|
||
if err != nil {
|
||
return out, err
|
||
}
|
||
if actual == nil || !actual.Equal(due) {
|
||
return out, errors.New("cursor is not an occurrence")
|
||
}
|
||
if before != nil && !due.Before(*before) {
|
||
return out, nil
|
||
}
|
||
out.Next = &due
|
||
upper := now.Add(time.Microsecond)
|
||
if before != nil && before.Before(upper) {
|
||
upper = *before
|
||
}
|
||
advance := func(at time.Time) error {
|
||
next, e := c.AtOrAfter(at)
|
||
if e != nil {
|
||
return e
|
||
}
|
||
if next != nil && before != nil && !next.Before(*before) {
|
||
next = nil
|
||
}
|
||
out.Next = next
|
||
return nil
|
||
}
|
||
skip := func(until time.Time, reason string) error {
|
||
out.Decisions = append(out.Decisions, Decision{ScheduledAt: *out.Next, CoveredUntil: &until, Kind: "SKIPPED", Reason: reason})
|
||
return advance(until)
|
||
}
|
||
cutoff := now.Add(-time.Duration(p.CatchupWindowMS) * time.Millisecond)
|
||
if upper.Before(cutoff) {
|
||
cutoff = upper
|
||
}
|
||
if out.Next.Before(cutoff) {
|
||
if err := skip(cutoff, "CATCHUP_WINDOW_EXCEEDED"); err != nil {
|
||
return out, err
|
||
}
|
||
}
|
||
if out.Next == nil || !out.Next.Before(upper) {
|
||
return out, nil
|
||
}
|
||
if p.Misfire == "SKIP" {
|
||
grace := now.Add(-time.Duration(p.GraceMS) * time.Millisecond)
|
||
if upper.Before(grace) {
|
||
grace = upper
|
||
}
|
||
if out.Next.Before(grace) {
|
||
if err := skip(grace, "MISFIRE_SKIPPED"); err != nil {
|
||
return out, err
|
||
}
|
||
}
|
||
}
|
||
if out.Next == nil || !out.Next.Before(upper) {
|
||
return out, nil
|
||
}
|
||
if active && p.Overlap == "SERIAL" {
|
||
out.Blocked = true
|
||
return out, nil
|
||
}
|
||
if active && p.Overlap == "SKIP" {
|
||
err := skip(upper, "OVERLAP_SKIPPED")
|
||
return out, err
|
||
}
|
||
if p.Misfire == "COALESCE" && out.Next.Before(now.Add(-time.Duration(p.GraceMS)*time.Millisecond)) {
|
||
out.Decisions = append(out.Decisions, Decision{ScheduledAt: *out.Next, CoveredUntil: &upper, Kind: "COALESCED", Reason: "MISFIRE_COALESCED"})
|
||
err := advance(upper)
|
||
return out, err
|
||
}
|
||
for n := uint32(0); n < p.CatchupLimit && out.Next != nil && out.Next.Before(upper); n++ {
|
||
at := *out.Next
|
||
out.Decisions = append(out.Decisions, Decision{ScheduledAt: at, Kind: "FIRED", Reason: "SCHEDULED"})
|
||
if err := advance(at.Add(time.Microsecond)); err != nil {
|
||
return out, err
|
||
}
|
||
if p.Overlap == "SERIAL" || p.Overlap == "SKIP" {
|
||
break
|
||
}
|
||
}
|
||
return out, nil
|
||
}
|