Make photo backfill single-flight and fail on unrecorded writes

Task 36c40130-a69a-46ed-a047-2cdc5ee2d2e0, remediation of review
findings F23-1 and F23-2 (PR #205).

F23-1: StartPhotoBackfill now checks, in the same write transaction
that enqueues, for a full backfill_dinner_photos job already queued or
running (new GetActivePhotoBackfillJob query) and returns its id instead
of enqueuing another, as plan.StartDraft does with GetActiveDraftJob.
The claim is the stored job row, so it holds across restarts; a job left
running by a stopped process is failed at boot and no longer blocks.

F23-2: a dinner whose outcome write fails is still reported failed with
grounding_error and the remaining dinners still run, but the job then
fails (internal_error, as a ground_dinner job's failed write does) with
the per-dinner outcomes published as its result, so it never reports
success for state it did not store.

Tests: overlapping triggers while queued, after a restart and while
running return the one job; a refused dinner write fails the job.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

internal/db/dinners.sql.go

@@ -214,6 +214,26 @@ func (q *Queries) FailPendingDinnerGrounding(ctx context.Context, arg FailPendin
 	return result.RowsAffected()
 }
 
+const getActivePhotoBackfillJob = `-- name: GetActivePhotoBackfillJob :one
+SELECT id
+FROM jobs
+WHERE kind = ?1
+	AND status IN ('queued', 'running')
+	AND json_extract(payload, '$.dinner_id') IS NULL
+ORDER BY id
+LIMIT 1
+`
+
+// The id of the full photo backfill job (no dinner_id) of kind queued or
+// running. StartPhotoBackfill enqueues one only when there is none, so
+// there is at most one. Returns sql.ErrNoRows when there is none.
+func (q *Queries) GetActivePhotoBackfillJob(ctx context.Context, kind string) (string, error) {
+	row := q.db.QueryRowContext(ctx, getActivePhotoBackfillJob, kind)
+	var id string
+	err := row.Scan(&id)
+	return id, err
+}
+
 const getDinner = `-- name: GetDinner :one
 SELECT id, name, normalized_name, status, origin, method, minutes_total, difficulty, cuisine, is_favorite, recipe_id, photo_id, grounding_status, updated_by, created_at, updated_at, recipe_url, grounding_error, grounding_revision FROM dinners
 WHERE id = ?

internal/db/querier.go

@@ -117,6 +117,10 @@ type Querier interface {
 	// when there is none, so there is at most one. Returns sql.ErrNoRows when
 	// there is none.
 	GetActiveDraftJob(ctx context.Context, arg GetActiveDraftJobParams) (GetActiveDraftJobRow, error)
+	// The id of the full photo backfill job (no dinner_id) of kind queued or
+	// running. StartPhotoBackfill enqueues one only when there is none, so
+	// there is at most one. Returns sql.ErrNoRows when there is none.
+	GetActivePhotoBackfillJob(ctx context.Context, kind string) (string, error)
 	// The batch still pending or running, oldest first. Returns sql.ErrNoRows
 	// when there is none.
 	GetActiveSuggestionBatch(ctx context.Context) (SuggestionBatch, error)

internal/dinners/backfill.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
@@ -47,15 +48,43 @@ type backfillResult struct {
 func (r backfillResult) PublicJSON() ([]byte, error) { return json.Marshal(r) }
 
 // StartPhotoBackfill enqueues a KindBackfillPhotos job over every dinner
-// without a photo and returns its id, without waiting for it.
+// without a photo and returns its id, without waiting for it. While a full
+// backfill is queued or running it enqueues nothing and returns that job's
+// id instead, so two passes never ground the same dinner at once; the jobs
+// table holds the claim, so it holds across restarts too.
 func (s *Service) StartPhotoBackfill(ctx context.Context) (string, error) {
-	id, err := s.backfillJob.Enqueue(ctx, BackfillJob{})
+	var id string
+	created := false
+	err := s.st.WithTx(ctx, func(tx *sql.Tx) error {
+		active, err := db.New(tx).GetActivePhotoBackfillJob(ctx, KindBackfillPhotos)
+		switch {
+		case errors.Is(err, sql.ErrNoRows):
+		case err != nil:
+			return fmt.Errorf("dinners: read active photo backfill: %w", err)
+		default:
+			id = active
+			return nil
+		}
+		id, err = s.backfillJob.EnqueueTx(ctx, tx, BackfillJob{})
+		if err != nil {
+			return fmt.Errorf("dinners: enqueue photo backfill: %w", err)
+		}
+		created = true
+		return nil
+	})
 	if err != nil {
-		return "", fmt.Errorf("dinners: enqueue photo backfill: %w", err)
+		return "", err
+	}
+	if created {
+		s.backfillJob.Wake()
 	}
 	return id, nil
 }
 
+// errBackfillNotRecorded fails a photo backfill that could not record the
+// outcome of one or more of its dinners.
+var errBackfillNotRecorded = errors.New("dinners: photo backfill: dinner outcomes not recorded")
+
 // runBackfillJob grounds each dinner ListPhotoBackfillDinners lists for p,
 // one at a time, with recipes.Grounder.Ground on its name: the first ranked
 // candidate whose photo caches and whose ingredients normalize grounds it,
@@ -64,38 +93,49 @@ func (s *Service) StartPhotoBackfill(ctx context.Context) (string, error) {
 // failure leaves the dinner failed with the new Reason as its
 // grounding_error, for the household to fix its recipe by hand. One
 // dinner's failure, its search's, model's or database write's, never stops
-// the others. Only a failure to list the dinners, or the job's context
-// ending, fails the job.
-func (s *Service) runBackfillJob(ctx context.Context, p BackfillJob, _ func(backfillResult) error) (backfillResult, error) {
+// the others. A failure to list the dinners, or the job's context ending,
+// fails the job at once. A write that failed fails the job once every
+// dinner has been tried, as a grounding job's failed write does, with the
+// outcomes published as its result, so the job never reports done for a
+// dinner whose outcome is not stored.
+func (s *Service) runBackfillJob(ctx context.Context, p BackfillJob, publish func(backfillResult) error) (backfillResult, error) {
 	q := db.New(s.st.Reader)
 	rows, err := q.ListPhotoBackfillDinners(ctx, sql.NullString{String: p.DinnerID, Valid: p.DinnerID != ""})
 	if err != nil {
 		return backfillResult{}, fmt.Errorf("dinners: photo backfill: list: %w", err)
 	}
 	res := backfillResult{Dinners: make([]BackfillOutcome, 0, len(rows))}
-	grounded := false
+	grounded, unrecorded := false, false
 	for _, row := range rows {
-		out, err := s.backfillDinner(ctx, row)
+		out, recorded, err := s.backfillDinner(ctx, row)
 		if err != nil {
 			return backfillResult{}, err
 		}
 		grounded = grounded || out.Outcome == BackfillGrounded
+		unrecorded = unrecorded || !recorded
 		res.Dinners = append(res.Dinners, out)
 	}
 	if grounded {
 		s.wake()
 	}
+	if unrecorded {
+		if err := publish(res); err != nil {
+			return backfillResult{}, errors.Join(errBackfillNotRecorded, err)
+		}
+		return backfillResult{}, errBackfillNotRecorded
+	}
 	return res, nil
 }
 
 // backfillDinner grounds row, a dinner ListPhotoBackfillDinners listed, as
-// runBackfillJob describes. Its error is only the job's context ending.
-func (s *Service) backfillDinner(ctx context.Context, row db.Dinner) (BackfillOutcome, error) {
+// runBackfillJob describes, and reports whether its outcome was recorded.
+// Its error is only the job's context ending.
+func (s *Service) backfillDinner(ctx context.Context, row db.Dinner) (BackfillOutcome, bool, error) {
 	out := BackfillOutcome{DinnerID: row.ID}
 	w := db.New(s.st.Writer)
 	gr, err := s.grounder.Ground(ctx, row.Name, row.Name)
 	if ctxErr := ctx.Err(); ctxErr != nil {
-		return BackfillOutcome{}, ctxErr
+		return BackfillOutcome{}, false, ctxErr
 	}
 	if err != nil {
 		reason := reasonFor(err, false)
@@ -108,7 +148,8 @@ func (s *Service) backfillDinner(ctx context.Context, row db.Dinner) (BackfillOu
 			ID:                row.ID,
 			GroundingRevision: row.GroundingRevision,
 		})
-		return s.backfillRecorded(ctx, out, BackfillFailed, reason, n, err), nil
+		out, recorded := s.backfillRecorded(ctx, out, BackfillFailed, reason, n, err)
+		return out, recorded, nil
 	}
 	n, err := w.GroundFailedDinner(ctx, db.GroundFailedDinnerParams{
 		RecipeID:          sql.NullString{String: gr.RecipeID, Valid: true},
@@ -120,14 +161,16 @@ func (s *Service) backfillDinner(ctx context.Context, row db.Dinner) (BackfillOu
 		ID:                row.ID,
 		GroundingRevision: row.GroundingRevision,
 	})
-	return s.backfillRecorded(ctx, out, BackfillGrounded, "", n, err), nil
+	out, recorded := s.backfillRecorded(ctx, out, BackfillGrounded, "", n, err)
+	return out, recorded, nil
 }
 
 // backfillRecorded is out once the backfill recorded outcome (with reason,
 // for BackfillFailed) on the dinner, writing n rows with err: a dinner that
 // changed meanwhile (no row) is BackfillSkipped, and one whose write failed
-// is BackfillFailed with ReasonGroundingError, the dinner left as it was.
-func (s *Service) backfillRecorded(ctx context.Context, out BackfillOutcome, outcome, reason string, n int64, err error) BackfillOutcome {
+// is BackfillFailed with ReasonGroundingError, the dinner left as it was,
+// and not recorded.
+func (s *Service) backfillRecorded(ctx context.Context, out BackfillOutcome, outcome, reason string, n int64, err error) (BackfillOutcome, bool) {
 	switch {
 	case err != nil:
 		// The error is a database error, never a dependency's text.
@@ -141,5 +184,5 @@ func (s *Service) backfillRecorded(ctx context.Context, out BackfillOutcome, out
 	if reason != "" {
 		out.Reason = &reason
 	}
-	return out
+	return out, err == nil
 }

internal/dinners/backfill_test.go

@@ -11,6 +11,7 @@ import (
 	"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/dinners"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/recipes"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/recipes/recipestest"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
@@ -239,3 +240,97 @@ func TestBackfillLeavesDinnerChangedWhileGrounding(t *testing.T) {
 		t.Errorf("dinner = %+v, want left failed without a photo", row)
 	}
 }
+
+func TestStartPhotoBackfillIsSingleFlight(t *testing.T) {
+	t.Parallel()
+	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))
+
+	// Queued: a second trigger returns the queued pass.
+	first, err := e.svc.StartPhotoBackfill(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	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)
+	}
+
+	// After a restart the queued pass is still the one: the claim is the
+	// stored job, not the process's memory.
+	restarted := newUnstartedEnvOn(t, st, dataDir)
+	restarted.search.Gate = e.search.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.
+	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)
+	}
+	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 {
+		t.Fatalf("%d backfill jobs, want 1", n)
+	}
+
+	// Finished: the next trigger starts a new pass.
+	close(e.search.Gate)
+	if job := restarted.waitJob(t, first); job.Status != jobs.StatusSucceeded {
+		t.Fatalf("job = %+v (result %s), want succeeded", job, job.Result)
+	}
+	next, err := restarted.svc.StartPhotoBackfill(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if next == first {
+		t.Fatalf("trigger after the pass finished = %q, want a new job", next)
+	}
+	restarted.waitJob(t, next)
+}
+
+func TestBackfillFailsJobWhenAWriteFails(t *testing.T) {
+	t.Parallel()
+	e := newEnv(t)
+	e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
+	absent := e.failedDinner(t, "Abalone surprise")
+	curry := e.failedDinner(t, "Chicken curry")
+	if _, err := e.st.Writer.ExecContext(t.Context(), `CREATE TRIGGER refuse_grounded BEFORE UPDATE OF grounding_status ON dinners
+		WHEN NEW.id = '`+curry+`' AND NEW.grounding_status = 'grounded'
+		BEGIN SELECT RAISE(ABORT, 'grounded refused'); END`); err != nil {
+		t.Fatal(err)
+	}
+
+	id, err := e.svc.StartPhotoBackfill(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	job := e.waitJob(t, id)
+	if job.Status != jobs.StatusFailed || job.Error == nil || *job.Error != jobs.CodeInternal {
+		t.Fatalf("job = %+v, want failed with %s", job, jobs.CodeInternal)
+	}
+	var got struct {
+		Dinners []dinners.BackfillOutcome `json:"dinners"`
+	}
+	if err := json.Unmarshal(job.Result, &got); err != nil {
+		t.Fatalf("result %s: %v", job.Result, err)
+	}
+	// The other dinner was still tried and recorded.
+	checkOutcomes(t, got.Dinners, []dinners.BackfillOutcome{
+		outcome(absent, dinners.BackfillFailed, ptr(dinners.ReasonNoRecipe)),
+		outcome(curry, dinners.BackfillFailed, ptr(dinners.ReasonGroundingError)),
+	})
+	if row := e.dinner(t, curry); row.GroundingStatus != dinners.GroundingFailed || row.PhotoID.Valid {
+		t.Errorf("curry = %+v, want left failed without a photo", row)
+	}
+}

internal/dinners/export_test.go

@@ -76,8 +76,16 @@ func (s *Service) RunGroundJob(ctx context.Context, p GroundJob) (string, error)
 var SafeSourceURL = safeSourceURL
 
 // RunBackfillJob runs the KindBackfillPhotos handler for p in the caller's
-// goroutine, as the job runner would, and returns its outcomes.
+// goroutine, as the job runner would, and returns its outcomes: on an
+// error, those it last published.
 func (s *Service) RunBackfillJob(ctx context.Context, p BackfillJob) ([]BackfillOutcome, error) {
-	r, err := s.runBackfillJob(ctx, p, nil)
-	return r.Dinners, err
+	var published backfillResult
+	r, err := s.runBackfillJob(ctx, p, func(partial backfillResult) error {
+		published = partial
+		return nil
+	})
+	if err != nil {
+		return published.Dinners, err
+	}
+	return r.Dinners, nil
 }