diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 190022c..18f0724 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -793,7 +793,11 @@ 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 +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 diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 77c5ca5..38d3b57 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -23,6 +23,13 @@ // 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 does not advance past it, but repeats the +// whole pass every ClaimRetry until the claim succeeds. Periods the pass +// already claimed are claimed again and skipped, so each still runs once. +// // # Time zone and DST // // Fire times are wall-clock times in the household zone, never the process @@ -35,10 +42,12 @@ // // # 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. +// 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 ( @@ -80,6 +89,10 @@ var names = map[string]bool{ // 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 @@ -183,10 +196,10 @@ type Scheduler struct { 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 + // looked; pending is the changes to apply, oldest first. + wake chan struct{} + pendingMu sync.Mutex + pending []change mu sync.Mutex schedules []Schedule @@ -281,18 +294,33 @@ 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 settings and wakes the loop. +// it only records the change and wakes the loop. func (s *Scheduler) changed(settings household.Settings) { - s.latestMu.Lock() - s.latest = settings - s.latestMu.Unlock() + 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 latest then + 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 @@ -323,21 +351,46 @@ func compile(schedules []Schedule, settings household.Settings) (*plan, error) { // 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. +// first pass catches up. Each pass first 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. A +// pass with a failed claim keeps done and its unapplied changes where they +// were and is repeated after ClaimRetry. func (s *Scheduler) loop(ctx context.Context, p *plan) { var done time.Time + var pending []change for { now := s.now() - if p != nil { - s.fire(ctx, p, done, now) + pending = append(pending, s.takeChanges()...) + failed := false + for len(pending) > 0 { + c := pending[0] + if c.at.After(done) { + if p != nil && !s.fire(ctx, p, done, c.at, now) { + failed = true + break + } + done = c.at + } + p = s.compileChange(ctx, c.settings) + pending = pending[1:] + } + if !failed { + if p == nil || s.fire(ctx, p, done, now, now) { + done = now + } else { + failed = true + } } - done = now if ctx.Err() != nil { return } var timeout <-chan time.Time - if p != nil { + switch { + case failed: + timeout = s.after(ClaimRetry) + case p != nil: if next, ok := p.next(now); ok { timeout = s.after(min(next.Sub(now), maxWait)) } @@ -347,45 +400,48 @@ func (s *Scheduler) loop(ctx context.Context, p *plan) { 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 } } } +// 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 +} + // 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) { +// until if it is after done; now is the current instant. It reports false +// when a claim failed, so the fire times it covered must be handled again. +func (s *Scheduler) fire(ctx context.Context, p *plan, done, until, now time.Time) bool { + ok := true for _, it := range p.items { if ctx.Err() != nil { - return + return ok } - at, ok := fireBefore(it.rule, p.loc, now) - if !ok || !at.After(done) { + at, due := fireBefore(it.rule, p.loc, until) + if !due || !at.After(done) { continue } - s.handle(ctx, it, p.loc, at, now) + if !s.handle(ctx, it, p.loc, at, now) { + ok = false + } } + return ok } // 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) { +// missed when at 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, loc *time.Location, at, now time.Time) bool { date := household.Today(at, loc) period := Period{Key: it.rule.PeriodKey(date), FireAt: at, Date: date} if it.rule.NextWeek { @@ -397,21 +453,22 @@ func (s *Scheduler) handle(ctx context.Context, it item, loc *time.Location, at, 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 + log.ErrorContext(ctx, "scheduler: claim period failed; retrying", "error", err, "retry_in", ClaimRetry.String()) + return false } if !claimed { - return + return true } if late := now.Sub(at); late > CatchUpWindow { log.WarnContext(ctx, "scheduler: period missed; recorded without running", "late", late.String()) - return + return true } if err := it.sch.Run(ctx, period); err != nil { log.ErrorContext(ctx, "scheduler: run failed", "error", err) - return + return true } log.InfoContext(ctx, "scheduler: ran") + return true } // claim inserts schedule_runs(name, key) and reports whether this call diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 2bb6183..51b30d1 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -565,6 +565,76 @@ func TestSettingsChangeMovesNextFire(t *testing.T) { 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"}) +} + +// A claim that fails is retried, not skipped, and the period then runs once. +func TestClaimFailureIsRetried(t *testing.T) { + e := newEnv(t) + 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 { + t.Fatal(err) + } + } + 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) + } + + if _, err := e.st.Writer.Exec(`UPDATE fail_claims SET on_ = 0`); err != nil { + t.Fatal(err) + } + 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"}) +} + func TestSettingsChangeTimezone(t *testing.T) { e := newEnv(t) var rec recorder