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()
|
|
}
|