Catch-up handled only each schedule's most recent fire time, so an earlier unclaimed period still within CatchUpWindow was skipped for good. Across the Chicago spring-forward change, 01:30 on 8 March and 01:30 on 9 March 2026 are 23 hours apart; booting at 9 March 01:35 ran only 9 March. The scheduler now handles, in time order, every fire time after done that passed within CatchUpWindow, plus the most recent one even when it is older (recorded as missed, as before). Each period keeps its own claim; a failed claim stops that schedule's queue and is retried in order. Older fire times are still never looked at. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
747 lines
24 KiB
Go
747 lines
24 KiB
Go
// Package scheduler runs the household's timed work: each registered
|
|
// schedule fires once per period at a household-local time read from the
|
|
// household settings.
|
|
//
|
|
// # Periods and the claim
|
|
//
|
|
// Every fire time belongs to one period, named by a period key derived in
|
|
// the household zone from the schedule's cadence (see Rule.PeriodKey): the
|
|
// local date "2026-10-09" for a daily rule, and the ISO week "2026-W41" of
|
|
// the fire date, or of the week after it, for a weekly rule. Before running,
|
|
// the scheduler inserts schedule_runs(schedule, period_key); the insert is
|
|
// the claim, and a period already claimed, by an earlier run, an earlier
|
|
// process or another instance, is skipped. A period therefore runs at most
|
|
// once: a Run that fails is logged and its period is not retried.
|
|
//
|
|
// # Catch-up
|
|
//
|
|
// When the scheduler starts, and whenever it wakes late, it handles, oldest
|
|
// first, every fire time of each schedule that passed within CatchUpWindow,
|
|
// and the most recent fire time even when it passed longer ago. Each
|
|
// unclaimed period within CatchUpWindow runs once. A most recent one that
|
|
// passed longer ago is claimed without running and logged as missed, so it
|
|
// can never run later, not after a restart and not after a settings change
|
|
// moves the fire time. Other fire times older than CatchUpWindow are never
|
|
// looked at.
|
|
//
|
|
// # Claim failures
|
|
//
|
|
// A claim that fails (the database refused the insert) is never skipped:
|
|
// the scheduler logs it and keeps that period's fire time, and every
|
|
// ClaimRetry claims that same period again until the claim succeeds. Only
|
|
// then does it apply the catch-up rule to it, running it if its fire time
|
|
// is within CatchUpWindow of the successful claim and recording it as
|
|
// missed otherwise, and only then does it go on to later fire times, even
|
|
// when one has passed meanwhile. Each period is claimed by its own fire
|
|
// time, so the later period runs once as well.
|
|
//
|
|
// # Time zone and DST
|
|
//
|
|
// Fire times are wall-clock times in the household zone, never the process
|
|
// zone, turned into instants by household.LocalTime. Its rule is the
|
|
// scheduler's DST rule: a time that does not exist on a spring-forward day
|
|
// fires later by the length of the gap (02:30 fires at 03:30 local), and a
|
|
// time that occurs twice on a fall-back day fires once, at its first
|
|
// occurrence. Either way the period key is that local date's, so the
|
|
// period still runs exactly once.
|
|
//
|
|
// # Fire times from data
|
|
//
|
|
// A DailyAt rule fires at a time read from the household's data rather
|
|
// than the settings: its function gives, for each household-local date,
|
|
// the instant the date's period fires, or none. The period is that date
|
|
// even when the instant falls on another day. The scheduler asks again on
|
|
// every pass, and whoever changes that data calls Wake once the change is
|
|
// committed, so the next pass comes at once and a fire time the change
|
|
// moved fires on time rather than up to maxWait late; the claim still runs
|
|
// a period at most once. A lookup that fails is logged and repeated every
|
|
// ClaimRetry, as a failed claim is.
|
|
//
|
|
// # Settings changes
|
|
//
|
|
// The scheduler subscribes to the household settings and records each change
|
|
// with the instant it was saved. A change takes effect at once without a
|
|
// restart: fire times up to that instant follow the old settings and later
|
|
// ones the new, even when the scheduler gets to the change late. So moving
|
|
// a fire time to earlier the same day does not fire the passed time, and a
|
|
// new fire time that was still ahead when the change was saved fires.
|
|
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
|
|
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
|
|
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
|
|
)
|
|
|
|
// Schedule names, as the schedule_runs.schedule CHECK constraint allows.
|
|
const (
|
|
SuggestionsWeekly = "suggestions_weekly"
|
|
PlanDraft = "plan_draft"
|
|
ShoppingChatAutosend = "shopping_chat_autosend"
|
|
DayOfReminder = "day_of_reminder"
|
|
RotationCheck = "rotation_check"
|
|
WeekRollover = "week_rollover"
|
|
AIRunsPurge = "ai_runs_purge"
|
|
)
|
|
|
|
// names are the schedule names Register accepts.
|
|
var names = map[string]bool{
|
|
SuggestionsWeekly: true,
|
|
PlanDraft: true,
|
|
ShoppingChatAutosend: true,
|
|
DayOfReminder: true,
|
|
RotationCheck: true,
|
|
WeekRollover: true,
|
|
AIRunsPurge: true,
|
|
}
|
|
|
|
// CatchUpWindow is how long after its fire time a period that has not run
|
|
// still runs. A period found later than that is recorded as missed.
|
|
const CatchUpWindow = 24 * time.Hour
|
|
|
|
// ClaimRetry is how long the scheduler waits before repeating a pass in
|
|
// which a claim failed.
|
|
const ClaimRetry = time.Minute
|
|
|
|
// maxWait bounds one sleep, so the scheduler checks the wall clock at least
|
|
// this often even if the host was suspended or its clock stepped.
|
|
const maxWait = time.Hour
|
|
|
|
// Rule is when a schedule fires: at household-local Time ("HH:MM") every
|
|
// day, or on Weekday only when Weekly is set, or every day at the instant
|
|
// At gives. Build one with Daily, Weekly, WeeklyForNextWeek or DailyAt.
|
|
type Rule struct {
|
|
Weekly bool
|
|
Weekday time.Weekday
|
|
Time string
|
|
// NextWeek keys a weekly period by the ISO week after the fire date,
|
|
// for work done ahead for the coming week.
|
|
NextWeek bool
|
|
// At, instead of Time, gives the instant the period of a
|
|
// household-local date fires.
|
|
At FireAt
|
|
}
|
|
|
|
// FireAt gives the instant the period of household-local date fires, or
|
|
// false when that date's period does not fire.
|
|
type FireAt func(ctx context.Context, date store.LocalDate) (time.Time, bool, error)
|
|
|
|
// Daily fires every day at hhmm; its period is the local date.
|
|
func Daily(hhmm string) Rule {
|
|
return Rule{Time: hhmm}
|
|
}
|
|
|
|
// Weekly fires on day at hhmm; its period is the ISO week of the fire date.
|
|
func Weekly(day time.Weekday, hhmm string) Rule {
|
|
return Rule{Weekly: true, Weekday: day, Time: hhmm}
|
|
}
|
|
|
|
// WeeklyForNextWeek fires on day at hhmm; its period is the ISO week after
|
|
// the fire date's.
|
|
func WeeklyForNextWeek(day time.Weekday, hhmm string) Rule {
|
|
return Rule{Weekly: true, Weekday: day, Time: hhmm, NextWeek: true}
|
|
}
|
|
|
|
// DailyAt fires at most once a day, at the instant at gives for the date;
|
|
// its period is that date, wherever the instant falls.
|
|
func DailyAt(at FireAt) Rule {
|
|
return Rule{At: at}
|
|
}
|
|
|
|
// PeriodKey returns the period key of a fire on household-local date
|
|
// fireDate: the date ("2026-10-09") for a daily rule, the ISO week
|
|
// ("2026-W53") for a weekly one, and the ISO week after that ("2027-W01")
|
|
// when NextWeek is set. It panics if fireDate is the zero LocalDate.
|
|
func (r Rule) PeriodKey(fireDate store.LocalDate) string {
|
|
if !r.Weekly {
|
|
return fireDate.String()
|
|
}
|
|
if r.NextWeek {
|
|
return household.PeriodKey(addDays(fireDate, 7))
|
|
}
|
|
return household.PeriodKey(fireDate)
|
|
}
|
|
|
|
// validate reports whether r can fire.
|
|
func (r Rule) validate() error {
|
|
if r.Weekly && (r.Weekday < time.Sunday || r.Weekday > time.Saturday) {
|
|
return fmt.Errorf("invalid weekday %d (want 0-6, Sunday = 0)", r.Weekday)
|
|
}
|
|
if !r.Weekly && (r.Weekday != time.Sunday || r.NextWeek) {
|
|
return errors.New("a daily rule has no weekday or next week")
|
|
}
|
|
if r.At != nil {
|
|
if r.Weekly || r.Time != "" {
|
|
return errors.New("a DailyAt rule has no weekday or time")
|
|
}
|
|
return nil
|
|
}
|
|
// Any date checks the time's form.
|
|
_, err := household.LocalTime(time.UTC, store.LocalDateOf(time.Unix(0, 0).UTC()), r.Time)
|
|
return err
|
|
}
|
|
|
|
// Period is one period of a schedule, as given to its Run.
|
|
type Period struct {
|
|
// Key is the period key claimed in schedule_runs.
|
|
Key string
|
|
// FireAt is the instant the period fired, in the household zone.
|
|
FireAt time.Time
|
|
// Date is the household-local date of FireAt.
|
|
Date store.LocalDate
|
|
// Week is the ISO Monday of the week Key names; for a daily rule, of
|
|
// the week containing Date.
|
|
Week store.LocalDate
|
|
}
|
|
|
|
// Schedule is registered work that runs once per period.
|
|
type Schedule struct {
|
|
// Name is one of the schedule name constants.
|
|
Name string
|
|
// Key, when set, prefixes the schedule's period keys with Key and a
|
|
// colon ("start:2026-10-14"), so two schedules of one Name claim
|
|
// different periods. Name and Key together are unique.
|
|
Key string
|
|
// PeriodKey, when set, gives the period key claimed for p in place of
|
|
// the rule's, such as "list_ready:chat:2026-10-19" for the week p.Week;
|
|
// p.Key is the rule's key. It must give each period its own key. A
|
|
// schedule sets Key or PeriodKey, not both.
|
|
PeriodKey func(p Period) string
|
|
// Rule returns when the schedule fires under settings. It is called
|
|
// again after every settings change.
|
|
Rule func(settings household.Settings) Rule
|
|
// Run does the period's work. It runs on the scheduler's goroutine,
|
|
// one schedule at a time, so it must return promptly: long work is
|
|
// enqueued as a job. ctx is cancelled when the scheduler stops. An
|
|
// error is logged; the period stays claimed and is not retried.
|
|
Run func(ctx context.Context, p Period) error
|
|
}
|
|
|
|
// label names sch in errors and logs: its Name, and its Key if set.
|
|
func (sch Schedule) label() string {
|
|
if sch.Key == "" {
|
|
return sch.Name
|
|
}
|
|
return sch.Name + ":" + sch.Key
|
|
}
|
|
|
|
// Settings is the household settings source: *household.SettingsService.
|
|
type Settings interface {
|
|
Get(ctx context.Context) (household.Settings, error)
|
|
Subscribe(fn func(household.Settings)) (unsubscribe func())
|
|
}
|
|
|
|
// Scheduler runs the registered schedules.
|
|
type Scheduler struct {
|
|
st *store.Store
|
|
settings Settings
|
|
logger *slog.Logger
|
|
|
|
// now is the clock and after the timer; tests replace both.
|
|
now func() time.Time
|
|
after func(d time.Duration) <-chan time.Time
|
|
|
|
// wake holds one token when the settings or the data a DailyAt rule
|
|
// reads changed since the loop last looked; pending is the settings
|
|
// changes to apply, oldest first.
|
|
wake chan struct{}
|
|
pendingMu sync.Mutex
|
|
pending []change
|
|
|
|
mu sync.Mutex
|
|
schedules []Schedule
|
|
started bool
|
|
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
// New returns a scheduler over st reading the household settings from
|
|
// settings.
|
|
func New(st *store.Store, settings Settings, logger *slog.Logger) (*Scheduler, error) {
|
|
if st == nil {
|
|
return nil, errors.New("scheduler: nil store")
|
|
}
|
|
if settings == nil {
|
|
return nil, errors.New("scheduler: nil settings")
|
|
}
|
|
if logger == nil {
|
|
return nil, errors.New("scheduler: nil logger")
|
|
}
|
|
return &Scheduler{
|
|
st: st,
|
|
settings: settings,
|
|
logger: logger,
|
|
now: time.Now,
|
|
after: time.After,
|
|
wake: make(chan struct{}, 1),
|
|
}, nil
|
|
}
|
|
|
|
// Register adds sch. It must be called before Start, once per name.
|
|
func (s *Scheduler) Register(sch Schedule) error {
|
|
if !names[sch.Name] {
|
|
return fmt.Errorf("scheduler: register: unknown schedule %q", sch.Name)
|
|
}
|
|
if sch.Rule == nil {
|
|
return fmt.Errorf("scheduler: register %s: nil rule", sch.label())
|
|
}
|
|
if sch.Run == nil {
|
|
return fmt.Errorf("scheduler: register %s: nil run", sch.label())
|
|
}
|
|
if strings.Contains(sch.Key, ":") {
|
|
return fmt.Errorf("scheduler: register %s: key has a colon", sch.label())
|
|
}
|
|
if sch.Key != "" && sch.PeriodKey != nil {
|
|
return fmt.Errorf("scheduler: register %s: both key and period key", sch.label())
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.started {
|
|
return fmt.Errorf("scheduler: register %s: already started", sch.label())
|
|
}
|
|
for _, have := range s.schedules {
|
|
if have.Name == sch.Name && have.Key == sch.Key {
|
|
return fmt.Errorf("scheduler: register %s: already registered", sch.label())
|
|
}
|
|
}
|
|
s.schedules = append(s.schedules, sch)
|
|
return nil
|
|
}
|
|
|
|
// Start reads the settings and starts the scheduler, which first catches
|
|
// up on each schedule's recent periods and then fires schedules as
|
|
// their times come. Settings whose zone or rules are invalid are an error
|
|
// and nothing starts. The scheduler stops when ctx is cancelled; Wait waits
|
|
// for it.
|
|
func (s *Scheduler) Start(ctx context.Context) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.started {
|
|
return errors.New("scheduler: already started")
|
|
}
|
|
// Subscribing first means a change saved after the read below still
|
|
// wakes the loop.
|
|
unsubscribe := s.settings.Subscribe(s.changed)
|
|
settings, err := s.settings.Get(ctx)
|
|
if err != nil {
|
|
unsubscribe()
|
|
return fmt.Errorf("scheduler: %w", err)
|
|
}
|
|
p, err := compile(s.schedules, settings)
|
|
if err != nil {
|
|
unsubscribe()
|
|
return err
|
|
}
|
|
s.started = true
|
|
s.wg.Add(1)
|
|
go func() {
|
|
defer s.wg.Done()
|
|
defer unsubscribe()
|
|
s.loop(ctx, p)
|
|
}()
|
|
return nil
|
|
}
|
|
|
|
// Wait blocks until the scheduler has stopped.
|
|
func (s *Scheduler) Wait() {
|
|
s.wg.Wait()
|
|
}
|
|
|
|
// change is a settings change and the instant it was saved.
|
|
type change struct {
|
|
at time.Time
|
|
settings household.Settings
|
|
}
|
|
|
|
// changed is the settings subscriber. It runs on the updating goroutine, so
|
|
// it only records the change and wakes the loop.
|
|
func (s *Scheduler) changed(settings household.Settings) {
|
|
s.pendingMu.Lock()
|
|
s.pending = append(s.pending, change{at: s.now(), settings: settings})
|
|
s.pendingMu.Unlock()
|
|
s.signal()
|
|
}
|
|
|
|
// Wake makes the scheduler look up its fire times again now rather than at
|
|
// its next timer. Call it after committing a change to the data a DailyAt
|
|
// rule reads, such as tonight's planned dinner. It never blocks.
|
|
func (s *Scheduler) Wake() {
|
|
s.signal()
|
|
}
|
|
|
|
// signal leaves one token in wake for the loop.
|
|
func (s *Scheduler) signal() {
|
|
select {
|
|
case s.wake <- struct{}{}:
|
|
default: // the loop already has a token and looks again then
|
|
}
|
|
}
|
|
|
|
// takeChanges returns and clears the pending changes.
|
|
func (s *Scheduler) takeChanges() []change {
|
|
s.pendingMu.Lock()
|
|
defer s.pendingMu.Unlock()
|
|
out := s.pending
|
|
s.pending = nil
|
|
return out
|
|
}
|
|
|
|
// plan is the schedules compiled against one version of the settings.
|
|
type plan struct {
|
|
loc *time.Location
|
|
items []item
|
|
}
|
|
|
|
type item struct {
|
|
sch Schedule
|
|
rule Rule
|
|
}
|
|
|
|
// compile builds the plan for settings.
|
|
func compile(schedules []Schedule, settings household.Settings) (*plan, error) {
|
|
loc, err := household.LoadZone(settings.Timezone)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("scheduler: household zone: %w", err)
|
|
}
|
|
p := &plan{loc: loc, items: make([]item, len(schedules))}
|
|
for i, sch := range schedules {
|
|
rule := sch.Rule(settings)
|
|
if err := rule.validate(); err != nil {
|
|
return nil, fmt.Errorf("scheduler: %s: %w", sch.label(), err)
|
|
}
|
|
p.items[i] = item{sch: sch, rule: rule}
|
|
}
|
|
return p, nil
|
|
}
|
|
|
|
// loop fires schedules until ctx is cancelled. done is the instant up to
|
|
// which fire times have been handled; it starts at the zero time so the
|
|
// first pass catches up. retry is the periods whose claim failed, each
|
|
// with its own fire time. Each pass first claims retry again; while any
|
|
// claim there still fails, the pass stops and is repeated after
|
|
// ClaimRetry. Otherwise it applies the pending settings changes in order,
|
|
// handling fire times up to a change under the plan it replaces, then
|
|
// handles fire times up to now under the current plan, stopping after the
|
|
// first stretch in which a claim failed.
|
|
func (s *Scheduler) loop(ctx context.Context, p *plan) {
|
|
var done time.Time
|
|
var pending []change
|
|
var retry []target
|
|
for {
|
|
now := s.now()
|
|
pending = append(pending, s.takeChanges()...)
|
|
retry = s.retry(ctx, retry, now)
|
|
for len(retry) == 0 && len(pending) > 0 {
|
|
c := pending[0]
|
|
if c.at.After(done) {
|
|
if p != nil {
|
|
retry = s.fire(ctx, p, done, c.at, now)
|
|
}
|
|
done = c.at
|
|
}
|
|
p = s.compileChange(ctx, c.settings)
|
|
pending = pending[1:]
|
|
}
|
|
if len(retry) == 0 {
|
|
if p != nil {
|
|
retry = s.fire(ctx, p, done, now, now)
|
|
}
|
|
done = now
|
|
}
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
|
|
var timeout <-chan time.Time
|
|
switch {
|
|
case len(retry) > 0:
|
|
timeout = s.after(ClaimRetry)
|
|
case p != nil:
|
|
next, ok, err := p.next(ctx, now)
|
|
switch {
|
|
case err != nil:
|
|
s.logger.ErrorContext(ctx, "scheduler: look up next fire time failed; retrying", "error", err, "retry_in", ClaimRetry.String())
|
|
timeout = s.after(ClaimRetry)
|
|
case ok:
|
|
timeout = s.after(min(next.Sub(now), maxWait))
|
|
}
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timeout:
|
|
case <-s.wake:
|
|
}
|
|
}
|
|
}
|
|
|
|
// compileChange returns the plan for changed settings, or nil when they
|
|
// are unusable.
|
|
func (s *Scheduler) compileChange(ctx context.Context, settings household.Settings) *plan {
|
|
p, err := compile(s.schedules, settings)
|
|
if err != nil {
|
|
// Settings are validated before they are saved, so this is a
|
|
// bug; firing at a guessed time would be worse than not firing
|
|
// until the settings are fixed.
|
|
s.logger.ErrorContext(ctx, "scheduler: settings unusable; no schedule fires until they change", "error", err)
|
|
return nil
|
|
}
|
|
return p
|
|
}
|
|
|
|
// target is one schedule's periods still to handle, oldest first, each
|
|
// named by its fire time and the household-local date of its period. A
|
|
// target whose fire times could not be looked up has lookup set instead,
|
|
// with the stretch (done, until] to look them up in again.
|
|
type target struct {
|
|
it item
|
|
loc *time.Location
|
|
fires []fireTime
|
|
|
|
lookup bool
|
|
done, until time.Time
|
|
}
|
|
|
|
// fire handles, for every schedule, its catch-up fire times after done and
|
|
// at or before until; now is the current instant. It returns the
|
|
// targets whose lookup or claim failed, to be handled again.
|
|
func (s *Scheduler) fire(ctx context.Context, p *plan, done, until, now time.Time) []target {
|
|
var failed []target
|
|
for _, it := range p.items {
|
|
if ctx.Err() != nil {
|
|
return failed
|
|
}
|
|
if t, ok := s.fireItem(ctx, it, p.loc, done, until, now); !ok {
|
|
failed = append(failed, t)
|
|
}
|
|
}
|
|
return failed
|
|
}
|
|
|
|
// fireItem handles, oldest first, it's fire times after done and at or
|
|
// before until that passed within CatchUpWindow of now, and the most recent
|
|
// of them even when it passed longer ago. It reports false, with the target
|
|
// to handle again, when the fire times could not be looked up or a period's
|
|
// claim failed.
|
|
func (s *Scheduler) fireItem(ctx context.Context, it item, loc *time.Location, done, until, now time.Time) (target, bool) {
|
|
fs, err := fireBetween(ctx, it.rule, loc, done, until)
|
|
if err != nil {
|
|
s.logger.ErrorContext(ctx, "scheduler: look up fire time failed; retrying",
|
|
"schedule", it.sch.label(), "error", err, "retry_in", ClaimRetry.String())
|
|
return target{it: it, loc: loc, lookup: true, done: done, until: until}, false
|
|
}
|
|
i := 0
|
|
for i < len(fs)-1 && now.Sub(fs[i].at) > CatchUpWindow {
|
|
i++
|
|
}
|
|
return s.handleAll(ctx, target{it: it, loc: loc, fires: fs[i:]}, now)
|
|
}
|
|
|
|
// handleAll handles the periods of t in order, stopping at the first whose
|
|
// claim failed. It reports false, with that period and those after it,
|
|
// when one did.
|
|
func (s *Scheduler) handleAll(ctx context.Context, t target, now time.Time) (target, bool) {
|
|
for i, f := range t.fires {
|
|
if !s.handle(ctx, t.it, f, now) {
|
|
t.fires = t.fires[i:]
|
|
return t, false
|
|
}
|
|
}
|
|
return target{}, true
|
|
}
|
|
|
|
// retry handles targets again at now and returns those whose lookup or
|
|
// claim still failed.
|
|
func (s *Scheduler) retry(ctx context.Context, targets []target, now time.Time) []target {
|
|
var failed []target
|
|
for i, t := range targets {
|
|
if ctx.Err() != nil {
|
|
return append(failed, targets[i:]...)
|
|
}
|
|
if t.lookup {
|
|
if again, ok := s.fireItem(ctx, t.it, t.loc, t.done, t.until, now); !ok {
|
|
failed = append(failed, again)
|
|
}
|
|
continue
|
|
}
|
|
if again, ok := s.handleAll(ctx, t, now); !ok {
|
|
failed = append(failed, again)
|
|
}
|
|
}
|
|
return failed
|
|
}
|
|
|
|
// handle claims the period of it's fire time f and runs it, or records it
|
|
// as missed when f is more than CatchUpWindow before now. It reports false
|
|
// when the claim failed and the period is still unclaimed.
|
|
func (s *Scheduler) handle(ctx context.Context, it item, f fireTime, now time.Time) bool {
|
|
at, date := f.at, f.date
|
|
period := Period{Key: it.rule.PeriodKey(date), FireAt: at, Date: date}
|
|
if it.rule.NextWeek {
|
|
period.Week = household.WeekStart(addDays(date, 7))
|
|
} else {
|
|
period.Week = household.WeekStart(date)
|
|
}
|
|
switch {
|
|
case it.sch.Key != "":
|
|
period.Key = it.sch.Key + ":" + period.Key
|
|
case it.sch.PeriodKey != nil:
|
|
period.Key = it.sch.PeriodKey(period)
|
|
}
|
|
log := s.logger.With("schedule", it.sch.Name, "period", period.Key, "fire_at", at.Format(time.RFC3339))
|
|
|
|
claimed, err := s.claim(ctx, it.sch.Name, period.Key, now)
|
|
if err != nil {
|
|
log.ErrorContext(ctx, "scheduler: claim period failed; retrying", "error", err, "retry_in", ClaimRetry.String())
|
|
return false
|
|
}
|
|
if !claimed {
|
|
return true
|
|
}
|
|
if late := now.Sub(at); late > CatchUpWindow {
|
|
log.WarnContext(ctx, "scheduler: period missed; recorded without running", "late", late.String())
|
|
return true
|
|
}
|
|
if err := it.sch.Run(ctx, period); err != nil {
|
|
log.ErrorContext(ctx, "scheduler: run failed", "error", err)
|
|
return true
|
|
}
|
|
log.InfoContext(ctx, "scheduler: ran")
|
|
return true
|
|
}
|
|
|
|
// claim inserts schedule_runs(name, key) and reports whether this call
|
|
// inserted it.
|
|
func (s *Scheduler) claim(ctx context.Context, name, key string, now time.Time) (bool, error) {
|
|
n, err := db.New(s.st.Writer).ClaimScheduleRun(ctx, db.ClaimScheduleRunParams{
|
|
ID: store.NewID(),
|
|
Schedule: name,
|
|
PeriodKey: key,
|
|
CreatedAt: store.Timestamp{Time: now.UTC().Truncate(time.Millisecond)}.String(),
|
|
})
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
return n == 1, nil
|
|
}
|
|
|
|
// next returns the earliest fire time of any schedule after now, or false
|
|
// when there is none.
|
|
func (p *plan) next(ctx context.Context, now time.Time) (time.Time, bool, error) {
|
|
var next time.Time
|
|
for _, it := range p.items {
|
|
f, ok, err := fireAfter(ctx, it.rule, p.loc, now)
|
|
if err != nil {
|
|
return time.Time{}, false, fmt.Errorf("%s: %w", it.sch.label(), err)
|
|
}
|
|
if ok && (next.IsZero() || f.at.Before(next)) {
|
|
next = f.at
|
|
}
|
|
}
|
|
return next, !next.IsZero(), nil
|
|
}
|
|
|
|
// window is how many days either side of a date fireBetween and fireAfter
|
|
// search: more than a week, so a weekly rule always has a fire time on
|
|
// each side.
|
|
const window = 8
|
|
|
|
// atWindow is how many days either side of a date fireBetween and fireAfter
|
|
// look up a DailyAt rule's fire times. A period firing further off is
|
|
// looked up on a later pass, which comes within maxWait.
|
|
const atWindow = 2
|
|
|
|
// fireTime is a fire time and the household-local date of its period.
|
|
type fireTime struct {
|
|
at time.Time
|
|
date store.LocalDate
|
|
}
|
|
|
|
// fireBetween returns the fire times of r in zone loc after from and at or
|
|
// before t, oldest first.
|
|
func fireBetween(ctx context.Context, r Rule, loc *time.Location, from, t time.Time) ([]fireTime, error) {
|
|
fs, err := fireTimes(ctx, r, loc, t)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out []fireTime
|
|
for _, f := range fs {
|
|
if f.at.After(from) && !f.at.After(t) {
|
|
out = append(out, f)
|
|
}
|
|
}
|
|
slices.SortFunc(out, func(a, b fireTime) int { return a.at.Compare(b.at) })
|
|
return out, nil
|
|
}
|
|
|
|
// fireAfter returns the earliest fire time of r in zone loc after t.
|
|
func fireAfter(ctx context.Context, r Rule, loc *time.Location, t time.Time) (fireTime, bool, error) {
|
|
fs, err := fireTimes(ctx, r, loc, t)
|
|
if err != nil {
|
|
return fireTime{}, false, err
|
|
}
|
|
var best fireTime
|
|
for _, f := range fs {
|
|
if f.at.After(t) && (best.at.IsZero() || f.at.Before(best.at)) {
|
|
best = f
|
|
}
|
|
}
|
|
return best, !best.at.IsZero(), nil
|
|
}
|
|
|
|
// fireTimes returns r's fire times for the local dates within window days
|
|
// of t's, or atWindow days for a DailyAt rule. A Time that fails to
|
|
// resolve was validated at compile, so it cannot; a DailyAt lookup that
|
|
// fails is an error.
|
|
func fireTimes(ctx context.Context, r Rule, loc *time.Location, t time.Time) ([]fireTime, error) {
|
|
today := household.Today(t, loc)
|
|
span := window
|
|
if r.At != nil {
|
|
span = atWindow
|
|
}
|
|
var out []fireTime
|
|
for i := -span; i <= span; i++ {
|
|
d := addDays(today, i)
|
|
if r.Weekly && weekday(d) != r.Weekday {
|
|
continue
|
|
}
|
|
if r.At != nil {
|
|
at, ok, err := r.At(ctx, d)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("fire time of %s: %w", d, err)
|
|
}
|
|
if ok {
|
|
out = append(out, fireTime{at: at, date: d})
|
|
}
|
|
continue
|
|
}
|
|
at, err := household.LocalTime(loc, d, r.Time)
|
|
if err != nil {
|
|
panic(fmt.Sprintf("scheduler: validated rule %+v: %v", r, err))
|
|
}
|
|
out = append(out, fireTime{at: at, date: d})
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// addDays returns d plus n calendar days.
|
|
func addDays(d store.LocalDate, n int) store.LocalDate {
|
|
y, m, day := d.Date()
|
|
return store.LocalDateOf(time.Date(y, m, day+n, 0, 0, 0, 0, time.UTC))
|
|
}
|
|
|
|
// weekday returns the day of the week of d.
|
|
func weekday(d store.LocalDate) time.Weekday {
|
|
y, m, day := d.Date()
|
|
return time.Date(y, m, day, 0, 0, 0, 0, time.UTC).Weekday()
|
|
}
|