diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 1d12443..9ab4786 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -261,7 +261,9 @@ automatic** (D-021). pending with nothing left to ground it. Photo backfill. `POST /api/v1/settings/photo-backfill` (CSRF-checked) enqueues a `backfill_dinner_photos` job and answers 202 `{id}` without - waiting for it; the kind is refused by `POST /api/v1/jobs`. The job + waiting for it; while one is queued or running it enqueues nothing and + answers with that job's id, so two passes never overlap, across restarts + too. The kind is refused by `POST /api/v1/jobs`. The job grounds, one at a time with `Ground` on its name, every unarchived dinner that is `failed` with no photo and no pasted `recipe_url` — grounded, `manual` and `pending` dinners are never listed, and a pasted link is @@ -271,7 +273,9 @@ automatic** (D-021). the fix-recipe flow. One dinner's failure never stops the others; a dinner renamed, archived or grounded by hand meanwhile is left as it is. Its result is `{dinners: [{dinner_id, outcome, reason}]}`, outcome - `grounded`, `failed` (with its reason) or `skipped`. + `grounded`, `failed` (with its reason) or `skipped`. A dinner whose write + fails is `failed` with `grounding_error` and left as it was; the others + still run, then the job fails with `internal_error`, keeping that result. `GET /api/v1/dinners/{id}/recipe-candidates` searches for the dinner's name and returns `{candidates: [{candidate_id, title, site_name, photo_url, minutes}]}`, at most 3, ranked as `Ground` diff --git a/internal/db/dinners.sql.go b/internal/db/dinners.sql.go index 6e629a4..be82f27 100644 --- a/internal/db/dinners.sql.go +++ b/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 = ? diff --git a/internal/db/querier.go b/internal/db/querier.go index 4ce3309..e19b54e 100644 --- a/internal/db/querier.go +++ b/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) diff --git a/internal/dinners/backfill.go b/internal/dinners/backfill.go index 96ac308..0c06c00 100644 --- a/internal/dinners/backfill.go +++ b/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 } diff --git a/internal/dinners/backfill_test.go b/internal/dinners/backfill_test.go index 3ad08a4..f82e58e 100644 --- a/internal/dinners/backfill_test.go +++ b/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) + } +} diff --git a/internal/dinners/export_test.go b/internal/dinners/export_test.go index ae83d86..dfdf5ee 100644 --- a/internal/dinners/export_test.go +++ b/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 } diff --git a/migrations/queries/dinners.sql b/migrations/queries/dinners.sql index f541136..8bed5a9 100644 --- a/migrations/queries/dinners.sql +++ b/migrations/queries/dinners.sql @@ -196,6 +196,18 @@ WHERE status <> 'archived' AND id = COALESCE(sqlc.narg(id), id) ORDER BY normalized_name, id; +-- name: GetActivePhotoBackfillJob :one +-- 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. +SELECT id +FROM jobs +WHERE kind = sqlc.arg(kind) + AND status IN ('queued', 'running') + AND json_extract(payload, '$.dinner_id') IS NULL +ORDER BY id +LIMIT 1; + -- name: GroundFailedDinner :execrows -- Records a photo backfill's grounding of a dinner it listed, while the -- dinner is still unarchived, failed, photoless, without a recipe_url and at