diff --git a/internal/shopping/archive.go b/internal/shopping/archive.go index ba4eb17..c87ad63 100644 --- a/internal/shopping/archive.go +++ b/internal/shopping/archive.go @@ -32,10 +32,12 @@ const ( // Generate titles it, so its post in the Supply archive gets a title when // the AI can write one. Titling never fails the freeze: a title that // cannot be written is logged and the list is frozen untitled, when its -// week label stands in for the title for good. A title job still in -// flight for the same plan revision and dinners shares Freeze's -// archive_title run (see harness.Run), and whichever stores the title -// first wins; a job whose write comes after the freeze stores nothing. +// week label stands in for the title for good. Freeze first waits for the +// title jobs in flight to store their titles or fail, and holds off new +// ones until the list is frozen, so a title a job's archive_title run +// wrote is stored before the freeze, and the freeze runs archive_title +// only when no job is running it. A job that runs after the freeze finds +// the list titled, or frozen untitled, and stores nothing. func (s *Service) Freeze(ctx context.Context, weekStart store.LocalDate) error { if _, err := household.WeekDays(weekStart); err != nil { return fmt.Errorf("shopping: freeze: %w", err) @@ -48,6 +50,8 @@ func (s *Service) Freeze(ctx context.Context, weekStart store.LocalDate) error { if weekStart.String() >= thisWeek.String() { return fmt.Errorf("shopping: freeze %s: %w", weekStart, ErrWeekNotEnded) } + s.titling.Lock() + defer s.titling.Unlock() list, err := q.GetShoppingList(ctx, weekStart.String()) if errors.Is(err, sql.ErrNoRows) { return ErrListNotFound @@ -63,7 +67,7 @@ func (s *Service) Freeze(ctx context.Context, weekStart store.LocalDate) error { if err != nil { return err } - if _, err := s.writeTitle(ctx, p); err != nil { + if _, err := s.storeTitle(ctx, p); err != nil { s.logger.WarnContext(ctx, "Supply run frozen untitled; the Supply archive shows its week label", "week", list.WeekStart, "list", list.ID, "error", err) } diff --git a/internal/shopping/archive_test.go b/internal/shopping/archive_test.go index 43f7a87..00fff92 100644 --- a/internal/shopping/archive_test.go +++ b/internal/shopping/archive_test.go @@ -1,11 +1,14 @@ package shopping_test import ( + "context" "database/sql" "errors" "reflect" "strconv" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -210,6 +213,121 @@ func TestFreezeTitlesUntitledList(t *testing.T) { wantTitle(t, "stored title after the job", e.storedTitle(t), "Tacos then Chili") } +// pauseTitle makes the env's first archive_title run pause once it has +// returned, before its title is stored, until the returned release is +// called or the test ends; paused is closed as it pauses. +func (e *env) pauseTitle(t *testing.T) (paused <-chan struct{}, release func()) { + t.Helper() + p, r := make(chan struct{}), make(chan struct{}) + var first atomic.Bool + e.afterTitle = func() { + if first.CompareAndSwap(false, true) { + close(p) + select { + case <-r: + case <-t.Context().Done(): + } + } + } + var released sync.Once + return p, func() { released.Do(func() { close(r) }) } +} + +// waitClosed fails t unless c is closed within 10 seconds. +func waitClosed(t *testing.T, what string, c <-chan struct{}) { + t.Helper() + select { + case <-c: + case <-time.After(10 * time.Second): + t.Fatalf("timed out waiting for %s", what) + } +} + +// TestFreezeWaitsForTitleJob: a freeze that starts while a title job's +// archive_title run has returned but its title is not yet stored waits for +// the job to store it, freezes the list with that title, and asks the +// model nothing itself. +func TestFreezeWaitsForTitleJob(t *testing.T) { + e := newEnv(t) + e.zoned(t, "America/New_York", "2026-10-19T12:00:00Z") + e.dinner(t, "Beef Chili", ing{"ground beef", "1", "lb"}) + e.plan(t, "Beef Chili") + e.model.QueueGenerate(titled("Chili Night")) + paused, release := e.pauseTitle(t) + e.start(t) + g := e.generate(t) + waitClosed(t, "the title job's archive_title run", paused) + + froze := make(chan error, 1) + go func() { froze <- e.svc.Freeze(context.Background(), week) }() + var err error + select { + case err = <-froze: + t.Errorf("Freeze = %v before the title job stored its title, want it to wait", err) + release() + case <-time.After(time.Second): + release() + err = <-froze + } + if err != nil { + t.Fatal(err) + } + + wantJob(t, e.waitTitle(t, g), jobs.StatusSucceeded, "") + wantTitle(t, "stored title", e.storedTitle(t), "Chili Night") + if status, _ := e.frozenState(t, week); status != shopping.StatusArchived { + t.Errorf("status = %q, want archived", status) + } + if n := len(e.model.Requests()); n != 1 { + t.Errorf("%d model calls, want 1: the freeze waits for the job's title", n) + } +} + +// TestTitleJobWaitsForFreeze: a title job that starts while a freeze's +// archive_title run has returned but its title is not yet stored waits for +// the freeze, then finds the list titled and asks the model nothing. +func TestTitleJobWaitsForFreeze(t *testing.T) { + e := newEnv(t) + e.zoned(t, "America/New_York", "2026-10-19T12:00:00Z") + e.dinner(t, "Beef Chili", ing{"ground beef", "1", "lb"}) + e.plan(t, "Beef Chili") + e.model.QueueGenerate(titled("Chili Night")) + g := e.generate(t) // the runner is not started: its title job waits + paused, release := e.pauseTitle(t) + + froze := make(chan error, 1) + go func() { froze <- e.svc.Freeze(context.Background(), week) }() + waitClosed(t, "the freeze's archive_title run", paused) + e.start(t) + deadline := time.Now().Add(10 * time.Second) + for { + job, err := e.runner.Get(t.Context(), *g.TitleJobID) + if err != nil { + t.Fatal(err) + } + if job.Status == jobs.StatusRunning { + break + } + if job.Status != jobs.StatusQueued || time.Now().After(deadline) { + t.Fatalf("title job %s, want it running while the freeze is paused", job.Status) + } + time.Sleep(5 * time.Millisecond) + } + release() + if err := <-froze; err != nil { + t.Fatal(err) + } + + wantJob(t, e.waitTitle(t, g), jobs.StatusSucceeded, "") + wantTitle(t, "stored title", e.storedTitle(t), "Chili Night") + if status, _ := e.frozenState(t, week); status != shopping.StatusArchived { + t.Errorf("status = %q, want archived", status) + } + if n := len(e.model.Requests()); n != 1 { + t.Errorf("%d model calls, want 1: the job waits for the freeze's title", n) + } +} + // TestFreezeUntitledWhenTitleFails: when archive_title cannot title the // list, it is frozen untitled all the same, the failure is logged, and the // Supply archive shows its week label for good. diff --git a/internal/shopping/shopping.go b/internal/shopping/shopping.go index 789c832..5a5451b 100644 --- a/internal/shopping/shopping.go +++ b/internal/shopping/shopping.go @@ -23,6 +23,7 @@ import ( "fmt" "log/slog" "strings" + "sync" "time" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/skills" @@ -97,6 +98,12 @@ type Service struct { logger *slog.Logger // now is the clock weeks are ended and current on; tests replace it. now func() time.Time + // titling orders the title writes against Freeze. A title job holds it + // shared from reading the list to storing its title, and Freeze holds + // it exclusively from reading the list to freezing it, so a freeze + // waits for the title jobs in flight to store their titles or fail, + // and no title job runs archive_title while a freeze is under way. + titling sync.RWMutex } // NewService returns a service over st that titles Supply runs with title @@ -360,10 +367,19 @@ func (s *Service) runTitleJob(ctx context.Context, p TitleJob, _ func(titleResul return titleResult{ListID: p.ListID, Title: title}, nil } -// writeTitle runs archive_title for p's Supply run and stores its title. -// When the list was titled by another job, before this one ran or -// meanwhile, that title stands and is returned. +// writeTitle runs archive_title for p's Supply run and stores its title, +// as storeTitle does, while no freeze is under way: it waits for one to +// end first, and a freeze waits for it to end. func (s *Service) writeTitle(ctx context.Context, p TitleJob) (string, error) { + s.titling.RLock() + defer s.titling.RUnlock() + return s.storeTitle(ctx, p) +} + +// storeTitle runs archive_title for p's Supply run and stores its title. +// When the list was titled by another job, before this one ran or +// meanwhile, that title stands and is returned. The caller holds titling. +func (s *Service) storeTitle(ctx context.Context, p TitleJob) (string, error) { q := db.New(s.st.Writer) title, err := s.currentTitle(ctx, q, p) if err != nil || title != "" { diff --git a/internal/shopping/shopping_test.go b/internal/shopping/shopping_test.go index a746343..a38f37e 100644 --- a/internal/shopping/shopping_test.go +++ b/internal/shopping/shopping_test.go @@ -54,6 +54,9 @@ type env struct { model *llm.Fake // beforeTitle, when set, runs as archive_title is asked for a title. beforeTitle func() + // afterTitle, when set, runs once archive_title has returned, before + // its title is stored. + afterTitle func() // beforeModel, when set, runs as the model is sent a request. beforeModel func(llm.Request) // log holds what the service logged. @@ -108,7 +111,11 @@ func newEnv(t *testing.T) *env { if e.beforeTitle != nil { e.beforeTitle() } - return skills.TitleArchive(ctx, cfg, scope, in) + out, err := skills.TitleArchive(ctx, cfg, scope, in) + if e.afterTitle != nil { + e.afterTitle() + } + return out, err } logger := slog.New(slog.NewTextHandler(e.log, nil)) e.runner, err = jobs.NewRunner(st, logger, 2)