From 32453d552b7bc1aa3b6db8cb4ba8b95c5cc01019 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Sun, 11 Oct 2026 21:34:09 -0400 Subject: [PATCH] Test photo backfill single-flight under simultaneous triggers TestStartPhotoBackfillIsSingleFlight now releases eight StartPhotoBackfill calls together behind a start gate, both while the pass is queued and while it runs, and asserts they all return the one job id with a single backfill job stored. The restart case closes the store and reopens it from the data directory before triggering again. Maestro task 36c40130-a69a-46ed-a047-2cdc5ee2d2e0 (review finding F23-1-T). Co-Authored-By: Claude Opus 5.5 --- internal/dinners/backfill_test.go | 78 ++++++++++++++++++++++++------- 1 file changed, 60 insertions(+), 18 deletions(-) diff --git a/internal/dinners/backfill_test.go b/internal/dinners/backfill_test.go index f82e58e..6d22243 100644 --- a/internal/dinners/backfill_test.go +++ b/internal/dinners/backfill_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "slices" + "sync" "testing" "time" @@ -241,51 +242,92 @@ func TestBackfillLeavesDinnerChangedWhileGrounding(t *testing.T) { } } +// triggerAtOnce calls svc.StartPhotoBackfill from n goroutines released +// together and returns the job ids they got. +func triggerAtOnce(t *testing.T, svc *dinners.Service, n int) []string { + t.Helper() + ids := make([]string, n) + errs := make([]error, n) + release := make(chan struct{}) + var wg sync.WaitGroup + for i := range n { + wg.Add(1) + go func() { + defer wg.Done() + <-release + ids[i], errs[i] = svc.StartPhotoBackfill(t.Context()) + }() + } + close(release) + wg.Wait() + if err := errors.Join(errs...); err != nil { + t.Fatal(err) + } + return ids +} + +// backfillJobs counts the KindBackfillPhotos jobs in st. +func backfillJobs(t *testing.T, st *store.Store) int { + t.Helper() + var n int + if err := st.Reader.QueryRowContext(t.Context(), `SELECT count(*) FROM jobs WHERE kind = ?`, dinners.KindBackfillPhotos).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} + func TestStartPhotoBackfillIsSingleFlight(t *testing.T) { t.Parallel() + const triggers = 8 dataDir := t.TempDir() st := openStore(t, dataDir) e := newUnstartedEnvOn(t, st, dataDir) e.failedDinner(t, "Chicken curry") - e.search.Gate = make(chan struct{}) - e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40)) + gate := make(chan struct{}) + e.search.Gate = gate - // Queued: a second trigger returns the queued pass. - first, err := e.svc.StartPhotoBackfill(t.Context()) - if err != nil { - t.Fatal(err) + // Queued: simultaneous triggers all get the one pass they enqueued. + ids := triggerAtOnce(t, e.svc, triggers) + first := ids[0] + for _, id := range ids { + if id != first { + t.Fatalf("simultaneous triggers = %q, want one job id", ids) + } } - if again, err := e.svc.StartPhotoBackfill(t.Context()); err != nil || again != first { - t.Fatalf("second trigger = %q, %v; want the queued %q", again, err, first) + if n := backfillJobs(t, st); n != 1 { + t.Fatalf("%d backfill jobs after simultaneous triggers, want 1", n) } // After a restart the queued pass is still the one: the claim is the // stored job, not the process's memory. + if err := st.Close(); err != nil { + t.Fatal(err) + } + st = openStore(t, dataDir) restarted := newUnstartedEnvOn(t, st, dataDir) - restarted.search.Gate = e.search.Gate + restarted.search.Gate = gate restarted.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40)) if again, err := restarted.svc.StartPhotoBackfill(t.Context()); err != nil || again != first { t.Fatalf("trigger after restart = %q, %v; want the queued %q", again, err, first) } - // Running: the pass waits at its search, and a trigger returns it. + // Running: the pass waits at its search, and simultaneous triggers all + // return it. restarted.start(t) for len(restarted.search.Queries()) == 0 { time.Sleep(time.Millisecond) } - if again, err := restarted.svc.StartPhotoBackfill(t.Context()); err != nil || again != first { - t.Fatalf("trigger while running = %q, %v; want the running %q", again, err, first) + for _, id := range triggerAtOnce(t, restarted.svc, triggers) { + if id != first { + t.Fatalf("trigger while running = %q, want the running %q", id, first) + } } - var n int - if err := st.Reader.QueryRowContext(t.Context(), `SELECT count(*) FROM jobs WHERE kind = ?`, dinners.KindBackfillPhotos).Scan(&n); err != nil { - t.Fatal(err) - } - if n != 1 { + if n := backfillJobs(t, st); n != 1 { t.Fatalf("%d backfill jobs, want 1", n) } // Finished: the next trigger starts a new pass. - close(e.search.Gate) + close(gate) if job := restarted.waitJob(t, first); job.Status != jobs.StatusSucceeded { t.Fatalf("job = %+v (result %s), want succeeded", job, job.Result) }