Wake the scheduler on plan and dinner changes (MQ-177-01)
CI / ci (push) Successful in 7m58s
CI / ci (pull_request) Successful in 7m20s

A swap, stored draft, or dinner method/minutes change (edit or grounding)
now calls Scheduler.Wake once committed, so a crock-pot dinner swapped in
for tonight is reminded at start_by minus 15 minutes rather than at the
next hourly pass. Adds a fake-clock test that swaps in a crock-pot
Quest-log dinner at 09:01 and asserts the start reminder at 09:15, once.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-10 14:55:07 -04:00
co-authored by Claude Opus 5.5
parent a7a56623e5
commit 765a36fa97
18 changed files with 172 additions and 39 deletions
+7 -6
View File
@@ -130,7 +130,12 @@ func run(cfg config.Config, logger *slog.Logger) error {
return err
}
dinnerSvc, err := dinners.NewService(db, grounder, runner, logger)
// The scheduler is made before the services whose changes wake it.
sched, err := scheduler.New(db, settings, logger)
if err != nil {
return err
}
dinnerSvc, err := dinners.NewService(db, grounder, runner, sched.Wake, logger)
if err != nil {
return err
}
@@ -152,7 +157,7 @@ func run(cfg config.Config, logger *slog.Logger) error {
return err
}
planSvc, err := plan.NewService(db, weekDrafter(ai, guard), runner, googleSvc, notifier, cfg.PublicURL, logger)
planSvc, err := plan.NewService(db, weekDrafter(ai, guard), runner, googleSvc, notifier, sched.Wake, cfg.PublicURL, logger)
if err != nil {
return err
}
@@ -188,10 +193,6 @@ func run(cfg config.Config, logger *slog.Logger) error {
return err
}
sched, err := scheduler.New(db, settings, logger)
if err != nil {
return err
}
if err := sched.Register(suggestSvc.WeeklySchedule()); err != nil {
return err
}
+8 -3
View File
@@ -832,9 +832,14 @@ computation the plan API shows) when the dinner is a crock-pot dinner
whose `start_by` is earlier than that day's `reminder_time`; its run
calls `EmitStartReminder(period.Date)`, which checks the same again. A
date whose week has no plan emits neither and logs it. A `DailyAt` fire
time is looked up from the data on every pass, so a plan change moves it
within an hour (`maxWait`); a lookup that fails is logged and repeated
every `ClaimRetry`.
time is looked up from the data on every pass. A committed change that
can move a `start_by` calls `Scheduler.Wake` (passed to `plan.NewService`
and `dinners.NewService` as `wake`): a plan swap or stored draft, and a
dinner's method or minutes changed by edit or grounding. The next pass
comes at once, so a crock-pot dinner swapped in for tonight is still
reminded 15 minutes before its `start_by`, not at the next hourly timer
(`maxWait`); a lookup that fails is logged and repeated every
`ClaimRetry`.
Fake-clock tests of a schedule live in `internal/scheduler`, where
`SetClock` is visible to tests.
+14 -5
View File
@@ -138,6 +138,9 @@ type Service struct {
grounder *recipes.Grounder
logger *slog.Logger
groundJob *jobs.Kind[GroundJob]
// wake is called after every committed change to a dinner's method or
// minutes, which can move its crock-pot start reminder.
wake func()
// groundingsMu guards groundings: the KindGroundDinner jobs running,
// by dinner id, so archiving or renaming a dinner can stop them.
@@ -157,10 +160,11 @@ type Service struct {
offerSearches map[string]map[*offerSearch]struct{}
}
// NewService returns a service over st that grounds dinners with grounder
// and registers the KindGroundDinner job on runner, which must not have
// started yet.
func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner, logger *slog.Logger) (*Service, error) {
// NewService returns a service over st that grounds dinners with grounder,
// calls wake (the scheduler's Wake) after every committed change to a
// dinner's method or minutes, and registers the KindGroundDinner job on
// runner, which must not have started yet.
func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner, wake func(), logger *slog.Logger) (*Service, error) {
switch {
case st == nil:
return nil, errors.New("dinners: nil store")
@@ -168,10 +172,12 @@ func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner
return nil, errors.New("dinners: nil grounder")
case runner == nil:
return nil, errors.New("dinners: nil job runner")
case wake == nil:
return nil, errors.New("dinners: nil wake")
case logger == nil:
return nil, errors.New("dinners: nil logger")
}
s := &Service{st: st, grounder: grounder, logger: logger, groundings: make(map[string]map[*runningGrounding]struct{}), offerKey: make([]byte, 32), offers: make(map[string]*offerSet), offerSearches: make(map[string]map[*offerSearch]struct{})}
s := &Service{st: st, grounder: grounder, wake: wake, logger: logger, groundings: make(map[string]map[*runningGrounding]struct{}), offerKey: make([]byte, 32), offers: make(map[string]*offerSet), offerSearches: make(map[string]map[*offerSearch]struct{})}
rand.Read(s.offerKey) // never fails: it crashes the program instead
kind, err := jobs.Register(runner, KindGroundDinner, s.runGroundJob, Reasons...)
if err != nil {
@@ -502,6 +508,9 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
if err != nil {
return Dinner{}, err
}
if p.Method != nil || p.MinutesTotal != nil {
s.wake()
}
if renamed {
updateCommitted()
s.clearOffers(row.ID, row.GroundingRevision)
+3
View File
@@ -196,6 +196,7 @@ func (s *Service) runGroundJob(ctx context.Context, p GroundJob, _ func(groundRe
if n == 0 {
return s.currentGrounding(ctx, row.ID)
}
s.wake()
return groundResult{DinnerID: row.ID, GroundingStatus: GroundingGrounded}, nil
}
@@ -577,6 +578,7 @@ func (s *Service) Ground(ctx context.Context, id string, f Fix, updatedBy string
if err != nil {
return Dinner{}, fmt.Errorf("%w: dinner %s: record: %w", ErrGroundFailed, id, err)
}
s.wake()
s.clearOffers(cur.ID, math.MaxInt64)
return fromRow(row), nil
}
@@ -700,6 +702,7 @@ func (s *Service) groundManual(ctx context.Context, id string, m ManualRecipe, u
if err != nil {
return Dinner{}, fmt.Errorf("%w: dinner %s: %w", ErrGroundFailed, id, err)
}
s.wake()
s.clearOffers(row.ID, math.MaxInt64)
return fromRow(row), nil
}
+1 -1
View File
@@ -104,7 +104,7 @@ func newUnstartedEnvOn(t *testing.T, st *store.Store, dataDir string) *env {
if err != nil {
t.Fatal(err)
}
svc, err := dinners.NewService(st, g, runner, logger)
svc, err := dinners.NewService(st, g, runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
+1
View File
@@ -234,6 +234,7 @@ func (s *Service) storeDraft(ctx context.Context, weekStart, thisWeek store.Loca
return "", err
}
s.notifier.Wake()
s.wake()
return planID, nil
}
+10 -3
View File
@@ -141,6 +141,9 @@ type Service struct {
draft Drafter
cal Calendar
notifier *notify.Service
// wake is called after every committed change to a plan's days, which
// can move a crock-pot dinner's start reminder.
wake func()
logger *slog.Logger
draftJob *jobs.Kind[DraftJob]
// publicURL is MEALQUEST_PUBLIC_URL, under which a pushed event links
@@ -150,11 +153,12 @@ type Service struct {
// NewService returns a service over st that drafts with draft, pushes to
// cal with links under publicURL, tells the household through notifier,
// which must be over st, and registers the KindDraftWeek job on
// which must be over st, calls wake (the scheduler's Wake) after every
// committed change to a plan's days, and registers the KindDraftWeek job on
// runner, which must not have started yet. The kind is service-only:
// StartDraft and StartScheduledDraft, which keep one draft in flight per
// week, are the only way to enqueue it.
func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, cal Calendar, notifier *notify.Service, publicURL *url.URL, logger *slog.Logger) (*Service, error) {
func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, cal Calendar, notifier *notify.Service, wake func(), publicURL *url.URL, logger *slog.Logger) (*Service, error) {
switch {
case st == nil:
return nil, errors.New("plan: nil store")
@@ -166,12 +170,14 @@ func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, cal Calenda
return nil, errors.New("plan: nil calendar")
case notifier == nil:
return nil, errors.New("plan: nil notifier")
case wake == nil:
return nil, errors.New("plan: nil wake")
case publicURL == nil:
return nil, errors.New("plan: nil public URL")
case logger == nil:
return nil, errors.New("plan: nil logger")
}
s := &Service{st: st, draft: draft, cal: cal, notifier: notifier, logger: logger, publicURL: publicURL}
s := &Service{st: st, draft: draft, cal: cal, notifier: notifier, wake: wake, logger: logger, publicURL: publicURL}
kind, err := jobs.Register(runner, KindDraftWeek, s.runDraft, Reasons...)
if err != nil {
return nil, err
@@ -469,6 +475,7 @@ func (s *Service) Swap(ctx context.Context, weekStart store.LocalDate, date, din
if err != nil {
return Plan{}, err
}
s.wake()
return p, nil
}
+1 -1
View File
@@ -146,7 +146,7 @@ func newEnv(t *testing.T) *env {
if err != nil {
t.Fatal(err)
}
e.svc, err = plan.NewService(st, drafter, e.runner, e.gcal, notifytest.New(t, st), publicURL, logger)
e.svc, err = plan.NewService(st, drafter, e.runner, e.gcal, notifytest.New(t, st), func() {}, publicURL, logger)
if err != nil {
t.Fatal(err)
}
@@ -8,6 +8,8 @@ import (
"testing"
"time"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/harness"
"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/notify"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/notify/notifytest"
@@ -265,3 +267,62 @@ func TestDailyAtLookupFailureIsRetried(t *testing.T) {
wantKeys(t, "next day", rec.keys(), []string{"start:2026-10-11", "start:2026-10-12", "start:2026-10-13"})
wantKeys(t, "schedule_runs", e.claimed(scheduler.DayOfReminder), rec.keys())
}
// questLog stores name as a Quest-log dinner, a crock-pot dinner of
// minutes, and returns its id. It is not a Regular.
func (p *planEnv) questLog(name string, minutes int) string {
p.e.t.Helper()
now := store.Now().String()
d, err := db.New(p.e.st.Writer).CreateDinner(context.Background(), db.CreateDinnerParams{
ID: store.NewID(),
Name: name,
NormalizedName: harness.Normalize(name),
Status: "quest_log",
Origin: "side_quest",
GroundingStatus: "manual",
CreatedAt: now,
UpdatedAt: now,
})
if err != nil {
p.e.t.Fatal(err)
}
p.ids[name] = d.ID
p.crockPot(name, minutes)
return d.ID
}
// TestStartReminderAfterSwapFiresOnTime: swapping a crock-pot dinner in
// for tonight wakes the scheduler, so its start reminder fires at start_by
// minus 15 minutes, 09:15, and not at the next hourly check, 10:00. At
// 09:00 tonight is Beef Chili, not a crock-pot dinner, and the scheduler's
// next timer is 10:00; at 09:01 the household swaps in an eight-hour
// crock-pot Quest-log dinner, start by 09:30. Each reminder fires once.
func TestStartReminderAfterSwapFiresOnTime(t *testing.T) {
e := newEnv(t)
p := e.wakingPlans(e.pool())
stew := p.questLog("Crock Pot Stew", 480) // start by 09:30
p.planWeek(w42)
r := e.start(at(2026, 10, 12, 8, 0), p.svc.ReminderSchedules()...)
r.runUntil(at(2026, 10, 12, 9, 0))
if next, _ := r.clock.nextDeadline(); !next.Equal(at(2026, 10, 12, 10, 0)) {
t.Fatalf("next wake at 09:00 = %s, want the hourly 10:00", next)
}
r.runUntil(at(2026, 10, 12, 9, 1))
if _, err := p.svc.Swap(context.Background(), w42, "2026-10-12", stew); err != nil {
t.Fatal(err)
}
r.changed()
r.runUntil(at(2026, 10, 12, 9, 14))
wantKeys(t, "before 09:15", e.reminderKeys(), nil)
r.runUntil(at(2026, 10, 12, 9, 15))
wantKeys(t, "at 09:15", e.reminderKeys(), []string{"day_of_reminder:start:2026-10-12"})
p.wantReminder("day_of_reminder:start:2026-10-12", "Start the crock pot by 9:30 am", "Crock Pot Stew", "Crock Pot Stew")
r.runUntil(at(2026, 10, 13, 23, 59))
wantKeys(t, "Tuesday", e.reminderKeys(), []string{"day_of_reminder:start:2026-10-12", "day_of_reminder:2026-10-12", "day_of_reminder:2026-10-13"})
p.wantReminder("day_of_reminder:2026-10-12", "Tonight: Crock Pot Stew · Recipe →", "", "Crock Pot Stew")
wantKeys(t, "schedule_runs", e.claimed(scheduler.DayOfReminder), []string{"2026-10-11", "2026-10-12", "2026-10-13", "start:2026-10-12"})
}
+17 -1
View File
@@ -81,7 +81,23 @@ func (c noCalendar) PushEvents(context.Context, []gcal.Event, func() error) erro
// plans builds a plan service drafting from the dinners ids over e with a
// started job runner, which stop stops (and the test's end, if not before).
// Its changes do not wake the scheduler: a draft job's wake would add a
// pass the fake clock's waits do not expect.
func (e *env) plans(ids map[string]string) *planEnv {
e.t.Helper()
return e.newPlans(ids, func() {})
}
// wakingPlans is plans whose changes wake the scheduler, as in the server.
// A test using it changes plans only from its own goroutine and waits for
// the wake's pass with changed.
func (e *env) wakingPlans(ids map[string]string) *planEnv {
e.t.Helper()
return e.newPlans(ids, e.wake)
}
// newPlans is plans whose service calls wake after a plan change.
func (e *env) newPlans(ids map[string]string, wake func()) *planEnv {
e.t.Helper()
p := &planEnv{e: e, model: &llm.Fake{}, ids: ids, logs: &syncBuffer{}}
guard, err := harness.NewGuard(harness.DefaultMaxCalls)
@@ -102,7 +118,7 @@ func (e *env) plans(ids map[string]string) *planEnv {
if err != nil {
e.t.Fatal(err)
}
p.svc, err = plan.NewService(e.st, drafter, runner, noCalendar{e.t}, notifytest.New(e.t, e.st), publicURL, logger)
p.svc, err = plan.NewService(e.st, drafter, runner, noCalendar{e.t}, notifytest.New(e.t, e.st), wake, publicURL, logger)
if err != nil {
e.t.Fatal(err)
}
+21 -7
View File
@@ -50,10 +50,11 @@
// than the settings: its function gives, for each household-local date,
// the instant the date's period fires, or none. The period is that date
// even when the instant falls on another day. The scheduler asks again on
// every pass, so a change to the data moves the fire time by the next pass,
// at most maxWait after the change; the claim still runs a period at most
// once. A lookup that fails is logged and repeated every ClaimRetry, as a
// failed claim is.
// every pass, and whoever changes that data calls Wake once the change is
// committed, so the next pass comes at once and a fire time the change
// moved fires on time rather than up to maxWait late; the claim still runs
// a period at most once. A lookup that fails is logged and repeated every
// ClaimRetry, as a failed claim is.
//
// # Settings changes
//
@@ -242,8 +243,9 @@ type Scheduler struct {
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 holds one token when the settings or the data a DailyAt rule
// reads changed since the loop last looked; pending is the settings
// changes to apply, oldest first.
wake chan struct{}
pendingMu sync.Mutex
pending []change
@@ -356,9 +358,21 @@ func (s *Scheduler) changed(settings household.Settings) {
s.pendingMu.Lock()
s.pending = append(s.pending, change{at: s.now(), settings: settings})
s.pendingMu.Unlock()
s.signal()
}
// Wake makes the scheduler look up its fire times again now rather than at
// its next timer. Call it after committing a change to the data a DailyAt
// rule reads, such as tonight's planned dinner. It never blocks.
func (s *Scheduler) Wake() {
s.signal()
}
// signal leaves one token in wake for the loop.
func (s *Scheduler) signal() {
select {
case s.wake <- struct{}{}:
default: // the loop already has a token and reads pending then
default: // the loop already has a token and looks again then
}
}
+18 -2
View File
@@ -7,6 +7,7 @@ import (
"log/slog"
"slices"
"sync"
"sync/atomic"
"testing"
"time"
@@ -59,10 +60,13 @@ func (c *fakeClock) Now() time.Time {
return c.now
}
// After returns a timer that fires d from now. The scheduler waits on only
// its latest timer, so this one replaces any still pending: a timer the
// scheduler gave up on when a wake came first never fires.
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.timers = []fakeTimer{{deadline: c.now.Add(d), ch: ch}}
c.mu.Unlock()
c.waits <- struct{}{}
return ch
@@ -134,6 +138,16 @@ type env struct {
t *testing.T
st *store.Store
settings *household.SettingsService
// sched is the scheduler start last started, which wake wakes.
sched atomic.Pointer[scheduler.Scheduler]
}
// wake wakes the scheduler start last started, as the services' wake does
// in the server; before the first start it does nothing.
func (e *env) wake() {
if s := e.sched.Load(); s != nil {
s.Wake()
}
}
func newEnv(t *testing.T) *env {
@@ -214,6 +228,7 @@ func (e *env) start(now time.Time, schedules ...scheduler.Schedule) *running {
cancel()
e.t.Fatal(err)
}
e.sched.Store(s)
var once sync.Once
stop := func() {
once.Do(func() {
@@ -255,7 +270,8 @@ func (r *running) runUntil(to time.Time) {
}
}
// changed waits for the scheduler to apply a settings change.
// changed waits for the scheduler to apply a settings change, or to look
// again after a wake.
func (r *running) changed() {
r.t.Helper()
r.idle()
+1 -1
View File
@@ -93,7 +93,7 @@ func newDinnersEnv(t *testing.T) *dinnersEnv {
if err != nil {
t.Fatal(err)
}
svc, err := dinners.NewService(st, g, runner, logger)
svc, err := dinners.NewService(st, g, runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
+2 -2
View File
@@ -65,7 +65,7 @@ func newIdeaActionsRouter(t *testing.T, model, normalize *llm.Fake, search *reci
if err != nil {
t.Fatal(err)
}
dinnerSvc, err := dinners.NewService(st, g, runner, logger)
dinnerSvc, err := dinners.NewService(st, g, runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
@@ -73,7 +73,7 @@ func newIdeaActionsRouter(t *testing.T, model, normalize *llm.Fake, search *reci
t.Error("draft_week ran: an idea action must never draft a plan")
return skills.Week{}, context.Canceled
}
planSvc, err := plan.NewService(st, draft, runner, newTestGoogle(t), notifytest.New(t, st), &url.URL{Scheme: "https", Host: "mealquest.example"}, logger)
planSvc, err := plan.NewService(st, draft, runner, newTestGoogle(t), notifytest.New(t, st), func() {}, &url.URL{Scheme: "https", Host: "mealquest.example"}, logger)
if err != nil {
t.Fatal(err)
}
+1 -1
View File
@@ -93,7 +93,7 @@ func newPlansRouterTitled(t *testing.T, title shopping.Titler) plansEnv {
t.Fatal(err)
}
notifier := notifytest.New(t, st)
svc, err := plan.NewService(st, fail, runner, googleSvc, notifier, pub, logger)
svc, err := plan.NewService(st, fail, runner, googleSvc, notifier, func() {}, pub, logger)
if err != nil {
t.Fatal(err)
}
+2 -2
View File
@@ -166,7 +166,7 @@ func newTestDinners(t *testing.T) *dinners.Service {
t.Fatal(err)
}
g := recipestest.NewGrounder(t, st, t.TempDir(), &recipestest.Search{}, &recipestest.Pages{}, &recipestest.Images{}, &llm.Fake{})
svc, err := dinners.NewService(st, g, runner, logger)
svc, err := dinners.NewService(st, g, runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
@@ -232,7 +232,7 @@ func newTestPlan(t *testing.T) *plan.Service {
fail := func(context.Context, string, skills.WeekRequest) (skills.Week, error) {
return skills.Week{}, errors.New("no model in this test")
}
svc, err := plan.NewService(st, fail, runner, newTestGoogle(t), notifytest.New(t, st), &url.URL{Scheme: "https", Host: "mealquest.example"}, logger)
svc, err := plan.NewService(st, fail, runner, newTestGoogle(t), notifytest.New(t, st), func() {}, &url.URL{Scheme: "https", Host: "mealquest.example"}, logger)
if err != nil {
t.Fatal(err)
}
+1 -1
View File
@@ -516,7 +516,7 @@ func TestGenerateManualRecipe(t *testing.T) {
if err != nil {
t.Fatal(err)
}
dsvc, err := dinners.NewService(e.st, grounder, runner, logger)
dsvc, err := dinners.NewService(e.st, grounder, runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
+3 -3
View File
@@ -1102,7 +1102,7 @@ func TestBatchNeverStoresDishPassedWhileRunning(t *testing.T) {
func TestLaterBatchNeverResuggestsRenamedPass(t *testing.T) {
e := newEnv(t) // runner not started: the replenish batch waits for the rename
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
dinnerSvc, err := dinners.NewService(e.st, e.grounder, e.runner, logger)
dinnerSvc, err := dinners.NewService(e.st, e.grounder, e.runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
@@ -1150,14 +1150,14 @@ func TestLaterBatchNeverResuggestsRenamedDrop(t *testing.T) {
e := newEnv(t)
ctx := t.Context()
logger := slog.New(slog.NewTextHandler(io.Discard, nil))
dinnerSvc, err := dinners.NewService(e.st, e.grounder, e.runner, logger)
dinnerSvc, err := dinners.NewService(e.st, e.grounder, e.runner, func() {}, logger)
if err != nil {
t.Fatal(err)
}
fail := func(context.Context, string, skills.WeekRequest) (skills.Week, error) {
return skills.Week{}, errors.New("no draft in this test")
}
planSvc, err := plan.NewService(e.st, fail, e.runner, noCalendar{}, notifytest.New(t, e.st), &url.URL{Scheme: "https", Host: "meals.example.com"}, logger)
planSvc, err := plan.NewService(e.st, fail, e.runner, noCalendar{}, notifytest.New(t, e.st), func() {}, &url.URL{Scheme: "https", Host: "meals.example.com"}, logger)
if err != nil {
t.Fatal(err)
}