Scheduler core with period keys and catch-up (Feature 20) #171

8 changed files with 1239 additions and 2 deletions
Showing only changes of commit b46192a43d - Show all commits
+14 -2
View File
@@ -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,
+14
View File
@@ -785,6 +785,20 @@ 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. 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`,
+3
View File
@@ -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
+38
View File
@@ -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()
}
+9
View File
@@ -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
}
+502
View File
@@ -0,0 +1,502 @@
// 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.
//
// # 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. A change takes effect
// at once without a restart: fire times up to the change follow the old
// settings and later ones the new, so moving a fire time to earlier the
// same day does not fire the passed time.
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
// 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; latest is the settings to apply.
wake chan struct{}
latestMu sync.Mutex
latest household.Settings
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()
}
// changed is the settings subscriber. It runs on the updating goroutine, so
// it only records the settings and wakes the loop.
func (s *Scheduler) changed(settings household.Settings) {
s.latestMu.Lock()
s.latest = settings
s.latestMu.Unlock()
select {
case s.wake <- struct{}{}:
default: // the loop already has a token and reads latest then
}
}
// 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.
func (s *Scheduler) loop(ctx context.Context, p *plan) {
var done time.Time
for {
now := s.now()
if p != nil {
s.fire(ctx, p, done, now)
}
done = now
if ctx.Err() != nil {
return
}
var timeout <-chan time.Time
if 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:
// Fire times up to the change follow the old settings.
now := s.now()
if p != nil {
s.fire(ctx, p, done, now)
}
done = now
s.latestMu.Lock()
settings := s.latest
s.latestMu.Unlock()
np, 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)
}
p = np
}
}
}
// fire handles, for every schedule, its most recent fire time at or before
// now if it is after done.
func (s *Scheduler) fire(ctx context.Context, p *plan, done, now time.Time) {
for _, it := range p.items {
if ctx.Err() != nil {
return
}
at, ok := fireBefore(it.rule, p.loc, now)
if !ok || !at.After(done) {
continue
}
s.handle(ctx, it, p.loc, at, now)
}
}
// handle claims the period of it fired at at and runs it, or records it as
// missed when at is more than CatchUpWindow before now.
func (s *Scheduler) handle(ctx context.Context, it item, loc *time.Location, at, now time.Time) {
date := household.Today(at, 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; period not run", "error", err)
return
}
if !claimed {
return
}
if late := now.Sub(at); late > CatchUpWindow {
log.WarnContext(ctx, "scheduler: period missed; recorded without running", "late", late.String())
return
}
if err := it.sch.Run(ctx, period); err != nil {
log.ErrorContext(ctx, "scheduler: run failed", "error", err)
return
}
log.InfoContext(ctx, "scheduler: ran")
}
// 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()
}
+653
View File
@@ -0,0 +1,653 @@
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"})
}
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")
}
}
+6
View File
@@ -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;