diff --git a/cmd/server/main.go b/cmd/server/main.go index 7103695..a97ca26 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -30,6 +30,7 @@ import ( "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/plan" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/push" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/recipes" + "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/scheduler" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/secretbox" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/server" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/shopping" @@ -187,17 +188,23 @@ func run(cfg config.Config, logger *slog.Logger) error { return err } + sched, err := scheduler.New(db, settings, logger) + if err != nil { + return err + } + router, err := server.NewRouter(logger, version, cfg.TrustedProxy, authFlow, sessions, settings, search, ai, chatSvc, dinnerSvc, suggestSvc, planSvc, shopSvc, googleSvc, pushSvc, ideasSvc, runner, photos, frontend.Assets()) if err != nil { return err } // Start fails the jobs a previous process left running before any - // worker runs. The job workers and the notification worker stop, and - // are waited for, before the database closes. + // worker runs. The job workers, the notification worker and the + // scheduler stop, and are waited for, before the database closes. workersCtx, stopWorkers := context.WithCancel(ctx) defer func() { stopWorkers() + sched.Wait() runner.Wait() notifier.Wait() }() @@ -213,6 +220,11 @@ func run(cfg config.Config, logger *slog.Logger) error { if _, err := dinnerSvc.FailInterruptedGroundings(ctx); err != nil { return err } + // The scheduler starts last: its catch-up may enqueue jobs and + // notifications, so their workers must already run. + if err := sched.Start(workersCtx); err != nil { + return err + } srv := &http.Server{ Addr: cfg.HTTPAddr, Handler: router, diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 0ebae68..18f0724 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -785,6 +785,24 @@ purge are fixed and not configurable. Runs are recorded in `schedule_runs` and executed through `jobs`. +`internal/scheduler` runs them. A schedule registers a name, a rule read +from the settings (`scheduler.Daily`, `Weekly` or `WeeklyForNextWeek`) and a +run function; settings changes reach it through `SettingsService.Subscribe` +and move the next fire time without a restart. Each fire time has a period +key (`Rule.PeriodKey`): the local date for a daily rule, the ISO week of the +fire date for a weekly one, or of the week after for `WeeklyForNextWeek` +(`plan_draft`). Inserting `schedule_runs(schedule, period_key)` is the claim; +a period already claimed is skipped, so a period runs at most once and a +failed run is not retried. A claim that fails is not skipped: the pass is +repeated every `scheduler.ClaimRetry` until the claim succeeds. A settings +change is recorded with the instant it was saved, and fire times up to that +instant follow the old rule and later ones the new, however late the +scheduler applies it. At boot only each schedule's most recent fire +time is considered: it runs if it passed within 24 hours, and is otherwise +claimed without running (logged as missed) so it never runs later. DST +follows `household.LocalTime`: a skipped time fires later by the gap, a +repeated time fires once, at its first occurrence. + ## Notifications - Events: `plan_ready`, `list_ready`, `day_of_reminder`, `new_suggestions`, diff --git a/internal/db/querier.go b/internal/db/querier.go index c626259..6cd1842 100644 --- a/internal/db/querier.go +++ b/internal/db/querier.go @@ -24,6 +24,9 @@ type Querier interface { // Marks the oldest queued job running and returns it. Returns sql.ErrNoRows // when no job is queued. ClaimNextJob(ctx context.Context, now sql.NullString) (Job, error) + // Claims one period of a schedule. The insert is the claim: 1 row means + // this caller owns the period, 0 rows means it was already claimed. + ClaimScheduleRun(ctx context.Context, arg ClaimScheduleRunParams) (int64, error) // Clears every day's calendar event id: none is on the chosen calendar. ClearCalendarEvents(ctx context.Context, updatedAt string) error // Records that Google refuses the stored grant: clears the refresh token and diff --git a/internal/db/schedule_runs.sql.go b/internal/db/schedule_runs.sql.go new file mode 100644 index 0000000..0370485 --- /dev/null +++ b/internal/db/schedule_runs.sql.go @@ -0,0 +1,38 @@ +// Code generated by sqlc. DO NOT EDIT. +// versions: +// sqlc v1.31.1 +// source: schedule_runs.sql + +package db + +import ( + "context" +) + +const claimScheduleRun = `-- name: ClaimScheduleRun :execrows +INSERT INTO schedule_runs (id, schedule, period_key, created_at) +VALUES (?, ?, ?, ?) +ON CONFLICT (schedule, period_key) DO NOTHING +` + +type ClaimScheduleRunParams struct { + ID string + Schedule string + PeriodKey string + CreatedAt string +} + +// Claims one period of a schedule. The insert is the claim: 1 row means +// this caller owns the period, 0 rows means it was already claimed. +func (q *Queries) ClaimScheduleRun(ctx context.Context, arg ClaimScheduleRunParams) (int64, error) { + result, err := q.db.ExecContext(ctx, claimScheduleRun, + arg.ID, + arg.Schedule, + arg.PeriodKey, + arg.CreatedAt, + ) + if err != nil { + return 0, err + } + return result.RowsAffected() +} diff --git a/internal/scheduler/export_test.go b/internal/scheduler/export_test.go new file mode 100644 index 0000000..38c8057 --- /dev/null +++ b/internal/scheduler/export_test.go @@ -0,0 +1,9 @@ +package scheduler + +import "time" + +// SetClock makes now the clock and after the timer s fires by. +func SetClock(s *Scheduler, now func() time.Time, after func(time.Duration) <-chan time.Time) { + s.now = now + s.after = after +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go new file mode 100644 index 0000000..9f8acee --- /dev/null +++ b/internal/scheduler/scheduler.go @@ -0,0 +1,587 @@ +// 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 looks only at +// each schedule's most recent fire time. If that time passed within +// CatchUpWindow and the period is unclaimed, the period runs once. If it +// passed longer ago, the period 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. Fire times before the most recent +// one 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. +// +// # 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" + "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. Build one with Daily, Weekly +// or WeeklyForNextWeek. +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 +} + +// 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} +} + +// 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") + } + // 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 + // 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 +} + +// 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 changed since the loop last + // looked; pending is the 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.Name) + } + if sch.Run == nil { + return fmt.Errorf("scheduler: register %s: nil run", sch.Name) + } + s.mu.Lock() + defer s.mu.Unlock() + if s.started { + return fmt.Errorf("scheduler: register %s: already started", sch.Name) + } + for _, have := range s.schedules { + if have.Name == sch.Name { + return fmt.Errorf("scheduler: register %s: already registered", sch.Name) + } + } + s.schedules = append(s.schedules, sch) + return nil +} + +// Start reads the settings and starts the scheduler, which first catches +// up on each schedule's most recent period 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() + select { + case s.wake <- struct{}{}: + default: // the loop already has a token and reads pending 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.Name, 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: + if next, ok := p.next(now); 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 period, named by the fire time it belongs to. +type target struct { + it item + loc *time.Location + at time.Time +} + +// fire handles, for every schedule, its most recent fire time at or before +// until if it is after done; now is the current instant. It returns the +// targets whose claim failed, to be claimed 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 + } + at, due := fireBefore(it.rule, p.loc, until) + if !due || !at.After(done) { + continue + } + if t := (target{it: it, loc: p.loc, at: at}); !s.handle(ctx, t, now) { + failed = append(failed, t) + } + } + return failed +} + +// retry handles targets again at now and returns those whose 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 !s.handle(ctx, t, now) { + failed = append(failed, t) + } + } + return failed +} + +// handle claims the period of t and runs it, or records it as missed when +// its fire time 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, t target, now time.Time) bool { + it, at := t.it, t.at + date := household.Today(at, t.loc) + 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) + } + 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 no schedule. +func (p *plan) next(now time.Time) (time.Time, bool) { + var next time.Time + for _, it := range p.items { + at, ok := fireAfter(it.rule, p.loc, now) + if ok && (next.IsZero() || at.Before(next)) { + next = at + } + } + return next, !next.IsZero() +} + +// window is how many days either side of a date fireBefore and fireAfter +// search: more than a week, so a weekly rule always has a fire time on +// each side. +const window = 8 + +// fireBefore returns the latest fire time of r in zone loc at or before t. +func fireBefore(r Rule, loc *time.Location, t time.Time) (time.Time, bool) { + var best time.Time + for _, at := range fireTimes(r, loc, t) { + if !at.After(t) && (best.IsZero() || at.After(best)) { + best = at + } + } + return best, !best.IsZero() +} + +// fireAfter returns the earliest fire time of r in zone loc after t. +func fireAfter(r Rule, loc *time.Location, t time.Time) (time.Time, bool) { + var best time.Time + for _, at := range fireTimes(r, loc, t) { + if at.After(t) && (best.IsZero() || at.Before(best)) { + best = at + } + } + return best, !best.IsZero() +} + +// fireTimes returns r's fire times on the local dates within window days +// of t's. A rule that fails to resolve was validated at compile, so it +// cannot. +func fireTimes(r Rule, loc *time.Location, t time.Time) []time.Time { + today := household.Today(t, loc) + var out []time.Time + for i := -window; i <= window; i++ { + d := addDays(today, i) + if r.Weekly && weekday(d) != r.Weekday { + 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, at) + } + return out +} + +// 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() +} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go new file mode 100644 index 0000000..27af79a --- /dev/null +++ b/internal/scheduler/scheduler_test.go @@ -0,0 +1,757 @@ +package scheduler_test + +import ( + "context" + "errors" + "io" + "log/slog" + "slices" + "sync" + "testing" + "time" + + "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household" + "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/scheduler" + "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store" + "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations" +) + +const zone = "America/Chicago" + +// chicago is the household zone of every test. +var chicago = mustZone(zone) + +func mustZone(name string) *time.Location { + loc, err := household.LoadZone(name) + if err != nil { + panic(err) + } + return loc +} + +// at returns the instant clocks in Chicago read the given time. +func at(year int, month time.Month, day, hour, minute int) time.Time { + return time.Date(year, month, day, hour, minute, 0, 0, chicago) +} + +// fakeClock is a clock whose time moves only when the test moves it. +type fakeClock struct { + mu sync.Mutex + now time.Time + timers []fakeTimer + // waits receives one value per After call, so the test knows the + // scheduler finished a pass and is waiting. + waits chan struct{} +} + +type fakeTimer struct { + deadline time.Time + ch chan time.Time +} + +func newFakeClock(now time.Time) *fakeClock { + return &fakeClock{now: now, waits: make(chan struct{}, 100)} +} + +func (c *fakeClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.now +} + +func (c *fakeClock) After(d time.Duration) <-chan time.Time { + c.mu.Lock() + ch := make(chan time.Time, 1) + c.timers = append(c.timers, fakeTimer{deadline: c.now.Add(d), ch: ch}) + c.mu.Unlock() + c.waits <- struct{}{} + return ch +} + +// set moves the clock to t and fires the timers due by then, reporting +// whether any fired. +func (c *fakeClock) set(t time.Time) bool { + c.mu.Lock() + defer c.mu.Unlock() + c.now = t + fired := false + kept := c.timers[:0] + for _, tm := range c.timers { + if tm.deadline.After(t) { + kept = append(kept, tm) + continue + } + tm.ch <- t + fired = true + } + c.timers = kept + return fired +} + +// nextDeadline returns the earliest pending timer deadline. +func (c *fakeClock) nextDeadline() (time.Time, bool) { + c.mu.Lock() + defer c.mu.Unlock() + var next time.Time + for _, tm := range c.timers { + if next.IsZero() || tm.deadline.Before(next) { + next = tm.deadline + } + } + return next, !next.IsZero() +} + +// recorder records the periods a schedule ran. +type recorder struct { + mu sync.Mutex + runs []scheduler.Period + err error +} + +func (r *recorder) run(_ context.Context, p scheduler.Period) error { + r.mu.Lock() + defer r.mu.Unlock() + r.runs = append(r.runs, p) + return r.err +} + +func (r *recorder) periods() []scheduler.Period { + r.mu.Lock() + defer r.mu.Unlock() + return slices.Clone(r.runs) +} + +func (r *recorder) keys() []string { + var keys []string + for _, p := range r.periods() { + keys = append(keys, p.Key) + } + return keys +} + +// env is a migrated database with household settings in Chicago. +type env struct { + t *testing.T + st *store.Store + settings *household.SettingsService +} + +func newEnv(t *testing.T) *env { + t.Helper() + ctx := context.Background() + st, err := store.Open(ctx, t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := st.Close(); err != nil { + t.Error(err) + } + }) + if err := store.Migrate(ctx, st.Writer, migrations.FS); err != nil { + t.Fatal(err) + } + if err := household.SeedSettings(ctx, st, zone); err != nil { + t.Fatal(err) + } + return &env{t: t, st: st, settings: household.NewSettingsService(st)} +} + +// update changes the household settings. +func (e *env) update(patch household.SettingsPatch) { + e.t.Helper() + if _, err := e.settings.Update(context.Background(), patch, "member"); err != nil { + e.t.Fatal(err) + } +} + +// claimed returns the period keys schedule_runs holds for name. +func (e *env) claimed(name string) []string { + e.t.Helper() + rows, err := e.st.Reader.Query(`SELECT period_key FROM schedule_runs WHERE schedule = ? ORDER BY period_key`, name) + if err != nil { + e.t.Fatal(err) + } + defer rows.Close() + var keys []string + for rows.Next() { + var k string + if err := rows.Scan(&k); err != nil { + e.t.Fatal(err) + } + keys = append(keys, k) + } + if err := rows.Err(); err != nil { + e.t.Fatal(err) + } + return keys +} + +// running is a started scheduler on a fake clock. +type running struct { + t *testing.T + clock *fakeClock + stop func() +} + +// start starts a scheduler on e at now with the schedules, and stops it +// when the test ends. It returns once the first pass, the catch-up, is done. +func (e *env) start(now time.Time, schedules ...scheduler.Schedule) *running { + e.t.Helper() + s, err := scheduler.New(e.st, e.settings, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + e.t.Fatal(err) + } + clock := newFakeClock(now) + scheduler.SetClock(s, clock.Now, clock.After) + for _, sch := range schedules { + if err := s.Register(sch); err != nil { + e.t.Fatal(err) + } + } + ctx, cancel := context.WithCancel(context.Background()) + if err := s.Start(ctx); err != nil { + cancel() + e.t.Fatal(err) + } + var once sync.Once + stop := func() { + once.Do(func() { + cancel() + s.Wait() + }) + } + e.t.Cleanup(stop) + r := &running{t: e.t, clock: clock, stop: stop} + r.idle() + return r +} + +// idle waits until the scheduler finished a pass and is waiting. +func (r *running) idle() { + r.t.Helper() + select { + case <-r.clock.waits: + case <-time.After(10 * time.Second): + r.t.Fatal("scheduler did not finish its pass") + } +} + +// runUntil moves the clock to to, stopping at every instant the scheduler +// asked to wake at on the way, as a real clock would. +func (r *running) runUntil(to time.Time) { + r.t.Helper() + for { + next, ok := r.clock.nextDeadline() + if !ok || next.After(to) { + if r.clock.set(to) { + r.idle() + } + return + } + if r.clock.set(next) { + r.idle() + } + } +} + +// changed waits for the scheduler to apply a settings change. +func (r *running) changed() { + r.t.Helper() + r.idle() +} + +func reminder(rec *recorder) scheduler.Schedule { + return scheduler.Schedule{ + Name: scheduler.DayOfReminder, + Rule: func(s household.Settings) scheduler.Rule { return scheduler.Daily(s.ReminderTime) }, + Run: rec.run, + } +} + +func suggestions(rec *recorder) scheduler.Schedule { + return scheduler.Schedule{ + Name: scheduler.SuggestionsWeekly, + Rule: func(s household.Settings) scheduler.Rule { + return scheduler.Weekly(time.Weekday(s.SuggestionsDow), s.SuggestionsTime) + }, + Run: rec.run, + } +} + +func planDraft(rec *recorder) scheduler.Schedule { + return scheduler.Schedule{ + Name: scheduler.PlanDraft, + Rule: func(s household.Settings) scheduler.Rule { + return scheduler.WeeklyForNextWeek(time.Weekday(s.PlanDraftDow), s.PlanDraftTime) + }, + Run: rec.run, + } +} + +func ptr[T any](v T) *T { return &v } + +func wantKeys(t *testing.T, what string, got, want []string) { + t.Helper() + if !slices.Equal(got, want) { + t.Fatalf("%s = %q, want %q", what, got, want) + } +} + +func TestFiresOncePerPeriod(t *testing.T) { + e := newEnv(t) + var rec recorder + // Seeded reminder time is 16:30; boot just after it on Oct 6 runs it. + r := e.start(at(2026, 10, 6, 16, 31), reminder(&rec)) + wantKeys(t, "after boot", rec.keys(), []string{"2026-10-06"}) + + r.runUntil(at(2026, 10, 7, 16, 29)) + wantKeys(t, "before the fire time", rec.keys(), []string{"2026-10-06"}) + r.runUntil(at(2026, 10, 9, 23, 59)) + wantKeys(t, "after three days", rec.keys(), []string{"2026-10-06", "2026-10-07", "2026-10-08", "2026-10-09"}) + + for i, p := range rec.periods()[1:] { + want := at(2026, 10, 7+i, 16, 30) + if !p.FireAt.Equal(want) { + t.Errorf("run %s fired at %s, want %s", p.Key, p.FireAt, want) + } + if p.FireAt.Location().String() != zone { + t.Errorf("run %s fired in zone %s, want %s", p.Key, p.FireAt.Location(), zone) + } + } + wantKeys(t, "schedule_runs", e.claimed(scheduler.DayOfReminder), rec.keys()) +} + +func TestRestartMidPeriodDoesNotRerun(t *testing.T) { + e := newEnv(t) + var first recorder + r := e.start(at(2026, 10, 7, 12, 0), suggestions(&first)) // Wednesday + // The seeded Sunday 08:00 of Oct 4 is days ago: missed, not run. + wantKeys(t, "first boot", first.keys(), nil) + r.runUntil(at(2026, 10, 11, 9, 0)) + wantKeys(t, "first process", first.keys(), []string{"2026-W41"}) + r.stop() + + // The restart is an hour after the Sunday 08:00 fire, mid-period. + var second recorder + r2 := e.start(at(2026, 10, 11, 9, 0), suggestions(&second)) + wantKeys(t, "after restart", second.keys(), nil) + r2.runUntil(at(2026, 10, 18, 7, 59)) + wantKeys(t, "rest of the period", second.keys(), nil) + r2.runUntil(at(2026, 10, 18, 8, 0)) + wantKeys(t, "next period", second.keys(), []string{"2026-W42"}) + wantKeys(t, "schedule_runs", e.claimed(scheduler.SuggestionsWeekly), []string{"2026-W40", "2026-W41", "2026-W42"}) +} + +func TestCatchUpWithin24Hours(t *testing.T) { + e := newEnv(t) + var rec recorder + // Sunday Oct 11 08:00 passed 23h59m before boot. + e.start(at(2026, 10, 12, 7, 59), suggestions(&rec)) + wantKeys(t, "catch-up", rec.keys(), []string{"2026-W41"}) + if p := rec.periods()[0]; !p.FireAt.Equal(at(2026, 10, 11, 8, 0)) { + t.Errorf("caught-up period fired at %s, want its fire time", p.FireAt) + } +} + +func TestNoCatchUpBeyond24Hours(t *testing.T) { + e := newEnv(t) + var rec recorder + // Sunday Oct 11 08:00 passed 24h01m before boot. + r := e.start(at(2026, 10, 12, 8, 1), suggestions(&rec)) + wantKeys(t, "boot", rec.keys(), nil) + // The missed period is recorded, so it never runs later. + wantKeys(t, "schedule_runs", e.claimed(scheduler.SuggestionsWeekly), []string{"2026-W41"}) + + // Moving the fire time to Sunday 23:00 puts that week's fire time + // within 24 h of now. The change itself never fires a passed time... + e.update(household.SettingsPatch{SuggestionsTime: ptr("23:00")}) + r.changed() + wantKeys(t, "after a settings change", rec.keys(), nil) + r.stop() + + // ...and the restart's catch-up finds the period already recorded. + var again recorder + r2 := e.start(at(2026, 10, 12, 8, 2), suggestions(&again)) + wantKeys(t, "after restart", again.keys(), nil) + r2.runUntil(at(2026, 10, 18, 23, 0)) + wantKeys(t, "next period", again.keys(), []string{"2026-W42"}) +} + +func TestCatchUpOnlyMostRecentPeriod(t *testing.T) { + e := newEnv(t) + var rec recorder + // Days of downtime: only Oct 9 16:30, 30 minutes ago, runs. + e.start(at(2026, 10, 9, 17, 0), reminder(&rec)) + wantKeys(t, "catch-up", rec.keys(), []string{"2026-10-09"}) + wantKeys(t, "schedule_runs", e.claimed(scheduler.DayOfReminder), []string{"2026-10-09"}) +} + +func TestSpringForward(t *testing.T) { + // Clocks in Chicago jump 02:00 CST → 03:00 CDT on Sunday 2026-03-08. + e := newEnv(t) + e.update(household.SettingsPatch{ReminderTime: ptr("08:00"), ReviewTime: ptr("02:30")}) + var daily, skipped recorder + review := scheduler.Schedule{ + Name: scheduler.RotationCheck, + Rule: func(s household.Settings) scheduler.Rule { return scheduler.Daily(s.ReviewTime) }, + Run: skipped.run, + } + r := e.start(at(2026, 3, 7, 9, 0), reminder(&daily), review) + r.runUntil(at(2026, 3, 9, 12, 0)) + + wantKeys(t, "08:00 runs", daily.keys(), []string{"2026-03-07", "2026-03-08", "2026-03-09"}) + wantFired(t, daily.periods(), + time.Date(2026, 3, 7, 14, 0, 0, 0, time.UTC), // 08:00 CST + time.Date(2026, 3, 8, 13, 0, 0, 0, time.UTC), // 08:00 CDT + time.Date(2026, 3, 9, 13, 0, 0, 0, time.UTC), + ) + + // 02:30 does not exist on Mar 8: it fires once, an hour later at + // 03:30 CDT, keyed by Mar 8. + wantKeys(t, "02:30 runs", skipped.keys(), []string{"2026-03-07", "2026-03-08", "2026-03-09"}) + wantFired(t, skipped.periods(), + time.Date(2026, 3, 7, 8, 30, 0, 0, time.UTC), // 02:30 CST + time.Date(2026, 3, 8, 8, 30, 0, 0, time.UTC), // 03:30 CDT + time.Date(2026, 3, 9, 7, 30, 0, 0, time.UTC), // 02:30 CDT + ) + if got := skipped.periods()[1].FireAt.Format("15:04 MST"); got != "03:30 CDT" { + t.Errorf("skipped 02:30 fired at %s local, want 03:30 CDT", got) + } +} + +func TestFallBack(t *testing.T) { + // Clocks in Chicago go back 02:00 CDT → 01:00 CST on Sunday 2026-11-01. + e := newEnv(t) + e.update(household.SettingsPatch{ReminderTime: ptr("08:00"), ReviewTime: ptr("01:30")}) + var daily, repeated recorder + review := scheduler.Schedule{ + Name: scheduler.RotationCheck, + Rule: func(s household.Settings) scheduler.Rule { return scheduler.Daily(s.ReviewTime) }, + Run: repeated.run, + } + r := e.start(at(2026, 10, 31, 9, 0), reminder(&daily), review) + r.runUntil(at(2026, 11, 2, 12, 0)) + + wantKeys(t, "08:00 runs", daily.keys(), []string{"2026-10-31", "2026-11-01", "2026-11-02"}) + wantFired(t, daily.periods(), + time.Date(2026, 10, 31, 13, 0, 0, 0, time.UTC), // 08:00 CDT + time.Date(2026, 11, 1, 14, 0, 0, 0, time.UTC), // 08:00 CST + time.Date(2026, 11, 2, 14, 0, 0, 0, time.UTC), + ) + + // 01:30 happens twice on Nov 1: it fires once, at the first (CDT). + wantKeys(t, "01:30 runs", repeated.keys(), []string{"2026-10-31", "2026-11-01", "2026-11-02"}) + wantFired(t, repeated.periods(), + time.Date(2026, 10, 31, 6, 30, 0, 0, time.UTC), // 01:30 CDT + time.Date(2026, 11, 1, 6, 30, 0, 0, time.UTC), // first 01:30, CDT + time.Date(2026, 11, 2, 7, 30, 0, 0, time.UTC), // 01:30 CST + ) +} + +func TestFallBackSecondOccurrenceDoesNotRerun(t *testing.T) { + e := newEnv(t) + e.update(household.SettingsPatch{ReviewTime: ptr("01:30")}) + var rec recorder + review := scheduler.Schedule{ + Name: scheduler.RotationCheck, + Rule: func(s household.Settings) scheduler.Rule { return scheduler.Daily(s.ReviewTime) }, + Run: rec.run, + } + // Boot between the two 01:30s of Nov 1: the first ran in an earlier + // process, so the second is not a new period. + first := time.Date(2026, 11, 1, 6, 30, 0, 0, time.UTC) + r := e.start(first.Add(-time.Minute), review) + r.runUntil(first) + wantKeys(t, "first 01:30", rec.keys(), []string{"2026-10-31", "2026-11-01"}) + r.stop() + + var again recorder + review.Run = again.run + r2 := e.start(first.Add(30*time.Minute), review) + r2.runUntil(first.Add(2 * time.Hour)) // past the second 01:30, CST + wantKeys(t, "second 01:30", again.keys(), nil) +} + +func wantFired(t *testing.T, got []scheduler.Period, want ...time.Time) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("%d runs, want %d", len(got), len(want)) + } + for i := range want { + if !got[i].FireAt.Equal(want[i]) { + t.Errorf("run %s fired at %s (%s), want %s", got[i].Key, got[i].FireAt.UTC(), got[i].FireAt, want[i]) + } + } +} + +func TestISOWeek53Boundary(t *testing.T) { + e := newEnv(t) + var sugg, draft recorder + // Suggestions Sunday 08:00 keys the ISO week ending that Sunday; the + // plan draft Friday 09:00 keys the week after. + r := e.start(at(2026, 12, 20, 9, 0), suggestions(&sugg), planDraft(&draft)) + r.runUntil(at(2027, 1, 11, 0, 0)) + + wantKeys(t, "suggestions", sugg.keys(), []string{"2026-W51", "2026-W52", "2026-W53", "2027-W01"}) + // Sunday 2027-01-03 is the last day of 2026-W53. + if p := sugg.periods()[2]; p.Date.String() != "2027-01-03" || p.Week.String() != "2026-12-28" { + t.Errorf("2026-W53 suggestions ran on %s for week %s, want 2027-01-03 for 2026-12-28", p.Date, p.Week) + } + // Friday Dec 18, two days before boot, was missed. + wantKeys(t, "plan drafts", draft.keys(), []string{"2026-W53", "2027-W01", "2027-W02"}) + // Friday 2026-12-25 drafts 2026-W53; Friday 2027-01-01 drafts 2027-W01. + if p := draft.periods()[0]; p.Date.String() != "2026-12-25" || p.Week.String() != "2026-12-28" { + t.Errorf("2026-W53 draft ran on %s for week %s, want 2026-12-25 for 2026-12-28", p.Date, p.Week) + } + if p := draft.periods()[1]; p.Date.String() != "2027-01-01" || p.Week.String() != "2027-01-04" { + t.Errorf("2027-W01 draft ran on %s for week %s, want 2027-01-01 for 2027-01-04", p.Date, p.Week) + } +} + +func TestPeriodKey(t *testing.T) { + date := func(s string) store.LocalDate { + d, err := store.ParseLocalDate(s) + if err != nil { + t.Fatal(err) + } + return d + } + tests := []struct { + rule scheduler.Rule + date string + want string + }{ + {scheduler.Daily("16:30"), "2026-12-31", "2026-12-31"}, + {scheduler.Weekly(time.Monday, "00:00"), "2026-12-28", "2026-W53"}, + {scheduler.Weekly(time.Sunday, "08:00"), "2027-01-03", "2026-W53"}, + {scheduler.Weekly(time.Monday, "00:00"), "2027-01-04", "2027-W01"}, + {scheduler.WeeklyForNextWeek(time.Friday, "09:00"), "2026-12-25", "2026-W53"}, + {scheduler.WeeklyForNextWeek(time.Friday, "09:00"), "2027-01-01", "2027-W01"}, + } + for _, tt := range tests { + if got := tt.rule.PeriodKey(date(tt.date)); got != tt.want { + t.Errorf("%+v.PeriodKey(%s) = %s, want %s", tt.rule, tt.date, got, tt.want) + } + } +} + +func TestSettingsChangeMovesNextFire(t *testing.T) { + e := newEnv(t) + var rec recorder + r := e.start(at(2026, 10, 7, 12, 0), reminder(&rec)) + wantKeys(t, "boot", rec.keys(), []string{"2026-10-06"}) + + // Earlier but still ahead: fires at the new time, not the old. + e.update(household.SettingsPatch{ReminderTime: ptr("13:00")}) + r.changed() + if next, _ := r.clock.nextDeadline(); !next.Equal(at(2026, 10, 7, 13, 0)) { + t.Errorf("next wake %s, want the new 13:00", next) + } + r.runUntil(at(2026, 10, 7, 13, 0)) + wantKeys(t, "at the new time", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + r.runUntil(at(2026, 10, 7, 23, 0)) + wantKeys(t, "at the old time", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + + // Moved to a time already passed today: nothing fires now, and + // tomorrow fires at the new time. + r.runUntil(at(2026, 10, 8, 12, 0)) + e.update(household.SettingsPatch{ReminderTime: ptr("11:00")}) + r.changed() + wantKeys(t, "moved into the past", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + r.runUntil(at(2026, 10, 9, 10, 59)) + wantKeys(t, "before the new time", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + r.runUntil(at(2026, 10, 9, 11, 0)) + wantKeys(t, "the new time", rec.keys(), []string{"2026-10-06", "2026-10-07", "2026-10-09"}) +} + +// A change saved at 12:00 moving the 16:30 reminder to 12:01 must fire it +// even when the scheduler only gets to the change at 12:02, after the new +// fire time has passed. +func TestSettingsChangeAppliedLateStillFires(t *testing.T) { + e := newEnv(t) + var rec recorder + // blocker holds the scheduler in a Run at 12:00 on 2026-10-07 while + // the settings change and the clock moves on. + started := make(chan struct{}) + release := make(chan struct{}) + blocker := scheduler.Schedule{ + Name: scheduler.RotationCheck, + Rule: func(household.Settings) scheduler.Rule { return scheduler.Daily("12:00") }, + Run: func(_ context.Context, p scheduler.Period) error { + if p.Key == "2026-10-07" { + close(started) + <-release + } + return nil + }, + } + r := e.start(at(2026, 10, 7, 11, 0), reminder(&rec), blocker) + wantKeys(t, "boot", rec.keys(), []string{"2026-10-06"}) + + r.clock.set(at(2026, 10, 7, 12, 0)) + <-started + e.update(household.SettingsPatch{ReminderTime: ptr("12:01")}) + r.clock.set(at(2026, 10, 7, 12, 2)) + close(release) + r.idle() // the pass the blocker held + r.idle() // the pass that applies the change + wantKeys(t, "after the late change", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + + r.runUntil(at(2026, 10, 7, 23, 0)) + wantKeys(t, "end of day", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + wantKeys(t, "claims", e.claimed(scheduler.DayOfReminder), []string{"2026-10-06", "2026-10-07"}) +} + +// failClaims makes every claim fail until recover is called. +func (e *env) failClaims() (recover func()) { + e.t.Helper() + for _, q := range []string{ + `CREATE TABLE fail_claims (on_ INTEGER NOT NULL)`, + `INSERT INTO fail_claims VALUES (1)`, + `CREATE TRIGGER fail_claim BEFORE INSERT ON schedule_runs + WHEN (SELECT on_ FROM fail_claims) = 1 + BEGIN SELECT RAISE(ABORT, 'injected claim failure'); END`, + } { + if _, err := e.st.Writer.Exec(q); err != nil { + e.t.Fatal(err) + } + } + return func() { + e.t.Helper() + if _, err := e.st.Writer.Exec(`UPDATE fail_claims SET on_ = 0`); err != nil { + e.t.Fatal(err) + } + } +} + +// A claim that fails is retried, not skipped, and the period then runs once. +func TestClaimFailureIsRetried(t *testing.T) { + e := newEnv(t) + recoverClaims := e.failClaims() + var rec recorder + r := e.start(at(2026, 10, 7, 12, 0), reminder(&rec)) + wantKeys(t, "boot with failing claims", rec.keys(), nil) + wantKeys(t, "claims while failing", e.claimed(scheduler.DayOfReminder), nil) + if next, _ := r.clock.nextDeadline(); !next.Equal(at(2026, 10, 7, 12, 0).Add(scheduler.ClaimRetry)) { + t.Errorf("next wake %s, want a retry after %s", next, scheduler.ClaimRetry) + } + + recoverClaims() + r.runUntil(at(2026, 10, 7, 12, 0).Add(scheduler.ClaimRetry)) + wantKeys(t, "after the retry", rec.keys(), []string{"2026-10-06"}) + r.runUntil(at(2026, 10, 7, 23, 0)) + wantKeys(t, "end of day", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + wantKeys(t, "claims", e.claimed(scheduler.DayOfReminder), []string{"2026-10-06", "2026-10-07"}) +} + +// A claim that keeps failing past the next daily fire is retried for its +// own period, not the most recent one: that period, now more than +// CatchUpWindow old, is recorded as missed, and the later period runs once. +func TestClaimFailureRecoversAfterNextFire(t *testing.T) { + e := newEnv(t) + recoverClaims := e.failClaims() + var rec recorder + // The 7 October 16:30 reminder is due and its claim fails, past the + // 8 October 16:30 fire. + r := e.start(at(2026, 10, 7, 16, 45), reminder(&rec)) + r.runUntil(at(2026, 10, 8, 17, 0)) + wantKeys(t, "runs while failing", rec.keys(), nil) + wantKeys(t, "claims while failing", e.claimed(scheduler.DayOfReminder), nil) + + // Claims recover 24h31m after the 7 October fire. + recoverClaims() + r.runUntil(at(2026, 10, 8, 17, 1)) + wantKeys(t, "after recovery", rec.keys(), []string{"2026-10-08"}) + wantKeys(t, "claims after recovery", e.claimed(scheduler.DayOfReminder), []string{"2026-10-07", "2026-10-08"}) + + r.runUntil(at(2026, 10, 9, 16, 0)) + wantKeys(t, "next day", rec.keys(), []string{"2026-10-08"}) +} + +func TestSettingsChangeTimezone(t *testing.T) { + e := newEnv(t) + var rec recorder + r := e.start(at(2026, 10, 7, 12, 0), reminder(&rec)) + e.update(household.SettingsPatch{Timezone: ptr("America/New_York")}) + r.changed() + // 16:30 in New York is 15:30 in Chicago. + r.runUntil(at(2026, 10, 7, 15, 30)) + wantKeys(t, "after the zone change", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + if got := rec.periods()[1].FireAt.Location().String(); got != "America/New_York" { + t.Errorf("fired in zone %s, want America/New_York", got) + } +} + +func TestRunErrorKeepsPeriodClaimed(t *testing.T) { + e := newEnv(t) + rec := recorder{err: errors.New("boom")} + r := e.start(at(2026, 10, 7, 16, 0), reminder(&rec)) + r.runUntil(at(2026, 10, 7, 23, 0)) + wantKeys(t, "runs", rec.keys(), []string{"2026-10-06", "2026-10-07"}) + r.stop() + var again recorder + e.start(at(2026, 10, 7, 23, 0), reminder(&again)) + wantKeys(t, "after restart", again.keys(), nil) +} + +func TestRegister(t *testing.T) { + e := newEnv(t) + s, err := scheduler.New(e.st, e.settings, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatal(err) + } + var rec recorder + bad := []scheduler.Schedule{ + {Name: "hourly", Rule: reminder(&rec).Rule, Run: rec.run}, + {Name: scheduler.DayOfReminder, Run: rec.run}, + {Name: scheduler.DayOfReminder, Rule: reminder(&rec).Rule}, + } + for _, sch := range bad { + if err := s.Register(sch); err == nil { + t.Errorf("Register(%q) accepted an invalid schedule", sch.Name) + } + } + if err := s.Register(reminder(&rec)); err != nil { + t.Fatal(err) + } + if err := s.Register(reminder(&rec)); err == nil { + t.Error("Register accepted a second schedule with the same name") + } +} + +func TestStartRejectsInvalidRule(t *testing.T) { + e := newEnv(t) + s, err := scheduler.New(e.st, e.settings, slog.New(slog.NewTextHandler(io.Discard, nil))) + if err != nil { + t.Fatal(err) + } + var rec recorder + if err := s.Register(scheduler.Schedule{ + Name: scheduler.WeekRollover, + Rule: func(household.Settings) scheduler.Rule { return scheduler.Weekly(7, "00:00") }, + Run: rec.run, + }); err != nil { + t.Fatal(err) + } + if err := s.Start(context.Background()); err == nil { + s.Wait() + t.Fatal("Start accepted weekday 7") + } +} + +func TestStopEndsTheLoop(t *testing.T) { + e := newEnv(t) + var rec recorder + r := e.start(at(2026, 10, 7, 12, 0), reminder(&rec)) + done := make(chan struct{}) + go func() { + r.stop() + close(done) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("scheduler did not stop") + } +} diff --git a/migrations/queries/schedule_runs.sql b/migrations/queries/schedule_runs.sql new file mode 100644 index 0000000..e0b3a93 --- /dev/null +++ b/migrations/queries/schedule_runs.sql @@ -0,0 +1,6 @@ +-- name: ClaimScheduleRun :execrows +-- Claims one period of a schedule. The insert is the claim: 1 row means +-- this caller owns the period, 0 rows means it was already claimed. +INSERT INTO schedule_runs (id, schedule, period_key, created_at) +VALUES (?, ?, ?, ?) +ON CONFLICT (schedule, period_key) DO NOTHING;