diff --git a/cmd/server/main.go b/cmd/server/main.go index b8bd3ff..3935c0a 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -139,6 +139,12 @@ func run(cfg config.Config, logger *slog.Logger) error { if err := runner.Start(jobsCtx); err != nil { return err } + // With those jobs failed, a dinner still pending with no grounding job + // queued or running is failed as interrupted, before any request can + // create one. + if _, err := dinnerSvc.FailInterruptedGroundings(ctx); err != nil { + return err + } srv := &http.Server{ Addr: cfg.HTTPAddr, Handler: router, diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c83239c..2b64f31 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -202,7 +202,9 @@ automatic** (D-021). the session's subject as `updated_by`; an unknown id is 404. - Dinner grounding. `POST /api/v1/dinners` also enqueues a `ground_dinner` job and answers with its `grounding_job_id` without waiting for it - (`GET /api/v1/jobs/{id}` reports it). The job grounds the dinner's + (`GET /api/v1/jobs/{id}` reports it); if the job cannot be enqueued it + still answers 201, with grounding_status `failed` (`grounding_error`) + and `grounding_job_id` null. The job grounds the dinner's `recipe_url` directly with `GroundURL`, else its name with `Ground` (see Recipe grounding), and sets grounding_status `grounded` with the recipe's recipe_id, photo_id, method, minutes_total and difficulty, or @@ -210,7 +212,11 @@ automatic** (D-021). `no_recipe_found`, `recipe_unusable` (the pasted link), `search_not_configured`, `search_failed`, `ai_failed`, `interrupted` (the server stopped) or `grounding_error`. It only writes a dinner still `pending`, so a fix-up - made meanwhile wins. `GET /api/v1/dinners/{id}/recipe-candidates` + made meanwhile wins. At boot, after the job runner fails the jobs a + previous process left running, every dinner still `pending` with no + `ground_dinner` job queued or running is set `failed` with `interrupted`, + so no dinner stays pending with nothing left to ground it. + `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` ranks them; it fetches pages but downloads no photo and runs no model, @@ -226,8 +232,11 @@ automatic** (D-021). photo and the dinner's difficulty unchanged. It answers 200 with the dinner; an invalid body is 400, an unknown dinner 404, a page that cannot ground it 422 `ungroundable` (the dinner is unchanged), recipe search not - set up 409 `not_configured`, and a search or model failure 502 - `search_failed` / `ai_failed`. + set up 409 `not_configured`, a search or model failure 502 + `search_failed` / `ai_failed` (only the reason code is logged, never the + provider's text), and grounding a candidate or link that runs past 90 + seconds (fetch, photo and `normalize_recipe`, all in the request) 504 + `ground_timeout`, leaving the dinner unchanged. - `/health` — liveness. - Everything else serves the embedded PWA with SPA fallback. diff --git a/internal/db/dinners.sql.go b/internal/db/dinners.sql.go index ee0fcb5..9e867ce 100644 --- a/internal/db/dinners.sql.go +++ b/internal/db/dinners.sql.go @@ -129,6 +129,38 @@ func (q *Queries) CreateDinner(ctx context.Context, arg CreateDinnerParams) (Din return i, err } +const failOrphanedPendingDinnerGroundings = `-- name: FailOrphanedPendingDinnerGroundings :execrows +UPDATE dinners +SET grounding_status = 'failed', + grounding_error = ?1, + updated_at = ?2 +WHERE grounding_status = 'pending' + AND NOT EXISTS ( + SELECT 1 FROM jobs + WHERE jobs.kind = ?3 + AND jobs.status IN ('queued', 'running') + AND json_extract(jobs.payload, '$.dinner_id') = dinners.id + ) +` + +type FailOrphanedPendingDinnerGroundingsParams struct { + GroundingError sql.NullString + UpdatedAt string + Kind string +} + +// Fails every dinner still pending whose grounding job is no longer queued +// or running: the process stopped before the job, or before its handler +// recorded a result. Run at boot, after the job runner fails the jobs a +// previous process left running. +func (q *Queries) FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error) { + result, err := q.db.ExecContext(ctx, failOrphanedPendingDinnerGroundings, arg.GroundingError, arg.UpdatedAt, arg.Kind) + if err != nil { + return 0, err + } + return result.RowsAffected() +} + const failPendingDinnerGrounding = `-- name: FailPendingDinnerGrounding :execrows UPDATE dinners SET grounding_status = 'failed', diff --git a/internal/db/querier.go b/internal/db/querier.go index 2c2c4a6..3c804d8 100644 --- a/internal/db/querier.go +++ b/internal/db/querier.go @@ -34,6 +34,11 @@ type Querier interface { DeleteRecipeIngredients(ctx context.Context, recipeID string) error DeleteRecipeSteps(ctx context.Context, recipeID string) error FailJob(ctx context.Context, arg FailJobParams) (int64, error) + // Fails every dinner still pending whose grounding job is no longer queued + // or running: the process stopped before the job, or before its handler + // recorded a result. Run at boot, after the job runner fails the jobs a + // previous process left running. + FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error) // Records why the background grounding of a dinner still pending failed; a // dinner grounded some other way meanwhile is left as it is (0 rows). FailPendingDinnerGrounding(ctx context.Context, arg FailPendingDinnerGroundingParams) (int64, error) diff --git a/internal/dinners/dinners.go b/internal/dinners/dinners.go index 760c27a..2af0e2f 100644 --- a/internal/dinners/dinners.go +++ b/internal/dinners/dinners.go @@ -155,10 +155,12 @@ func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner return s, nil } -// Created is a new Regular with the id of the job grounding it. +// Created is a new Regular with the id of the job grounding it, which is +// nil when the job could not be enqueued and the dinner's grounding is +// failed instead. type Created struct { Dinner - GroundingJobID string `json:"grounding_job_id"` + GroundingJobID *string `json:"grounding_job_id"` } // CreateRegular adds a Regular named by r.Name, or by the last path segment @@ -166,24 +168,35 @@ type Created struct { // origin regular and grounding_status pending, recording updatedBy, and // enqueues the KindGroundDinner job that grounds it. It returns without // waiting for the job. A name repeating an unarchived dinner is a -// *DuplicateError. +// *DuplicateError. When the job cannot be enqueued, the stored dinner is +// returned with grounding_status failed (ReasonGroundingError) and no job +// id, so the household can fix its recipe by hand. func (s *Service) CreateRegular(ctx context.Context, r NewRegular, updatedBy string) (Created, error) { d, err := s.createRegular(ctx, r, updatedBy) if err != nil { return Created{}, err } jobID, err := s.groundJob.Enqueue(ctx, GroundJob{DinnerID: d.ID}) - if err != nil { - // The dinner is stored; record that it will not be grounded so it - // is not left pending with no job, and can be fixed by hand. - _, failErr := db.New(s.st.Writer).FailPendingDinnerGrounding(context.WithoutCancel(ctx), db.FailPendingDinnerGroundingParams{ - GroundingError: sql.NullString{String: ReasonGroundingError, Valid: true}, - UpdatedAt: store.Now().String(), - ID: d.ID, - }) - return Created{}, errors.Join(fmt.Errorf("dinners: enqueue grounding of %s: %w", d.ID, err), failErr) + if err == nil { + return Created{Dinner: d, GroundingJobID: &jobID}, nil } - return Created{Dinner: d, GroundingJobID: jobID}, nil + // The dinner is stored; record that it will not be grounded so it is + // not left pending with no job. The error is a database error, never a + // dependency's text. + s.logger.ErrorContext(ctx, "dinner grounding not enqueued", "dinner_id", d.ID, "error", err) + write := context.WithoutCancel(ctx) + if _, err := db.New(s.st.Writer).FailPendingDinnerGrounding(write, db.FailPendingDinnerGroundingParams{ + GroundingError: sql.NullString{String: ReasonGroundingError, Valid: true}, + UpdatedAt: store.Now().String(), + ID: d.ID, + }); err != nil { + return Created{}, fmt.Errorf("dinners: record failed grounding of %s: %w", d.ID, err) + } + row, err := db.New(s.st.Reader).GetDinner(write, d.ID) + if err != nil { + return Created{}, fmt.Errorf("dinners: read %s: %w", d.ID, err) + } + return Created{Dinner: fromRow(row)}, nil } func (s *Service) createRegular(ctx context.Context, r NewRegular, updatedBy string) (Dinner, error) { diff --git a/internal/dinners/export_test.go b/internal/dinners/export_test.go new file mode 100644 index 0000000..9eb3511 --- /dev/null +++ b/internal/dinners/export_test.go @@ -0,0 +1,14 @@ +package dinners + +import ( + "testing" + "time" +) + +// SetGroundTimeout sets the deadline of grounding by hand until t ends. A +// test calling it must not run in parallel. +func SetGroundTimeout(t *testing.T, d time.Duration) { + old := groundTimeout + groundTimeout = d + t.Cleanup(func() { groundTimeout = old }) +} diff --git a/internal/dinners/grounding.go b/internal/dinners/grounding.go index 1cf204c..9f8335f 100644 --- a/internal/dinners/grounding.go +++ b/internal/dinners/grounding.go @@ -8,6 +8,7 @@ import ( "fmt" "slices" "strings" + "time" "unicode/utf8" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db" @@ -63,6 +64,18 @@ const ( manualSourcePrefix = "mealquest:manual:" ) +// GroundTimeout bounds grounding a candidate or link by hand, which runs in +// the request: the page fetch (recipes.Timeout), its photo download and +// normalize_recipe (harness.DefaultTimeout), with some slack. +const GroundTimeout = 90 * time.Second + +// groundTimeout is GroundTimeout; tests shorten it. +var groundTimeout = GroundTimeout + +// ErrGroundTimeout is returned by Ground when grounding a candidate or link +// runs past GroundTimeout; the dinner is left as it was. +var ErrGroundTimeout = errors.New("dinners: grounding the recipe took too long") + // GroundJob is the KindGroundDinner payload. type GroundJob struct { DinnerID string `json:"dinner_id"` @@ -143,6 +156,27 @@ func (s *Service) runGroundJob(ctx context.Context, p GroundJob, _ func(groundRe return groundResult{DinnerID: row.ID, GroundingStatus: GroundingGrounded}, nil } +// FailInterruptedGroundings fails, with ReasonInterrupted, every dinner +// still pending whose KindGroundDinner job is no longer queued or running: +// a previous process stopped before enqueuing the job or before its +// handler recorded a result. Call it once at boot, after the job runner's +// Start has failed the jobs that process left running, so no dinner stays +// pending with nothing left to ground it. It returns how many it failed. +func (s *Service) FailInterruptedGroundings(ctx context.Context) (int64, error) { + n, err := db.New(s.st.Writer).FailOrphanedPendingDinnerGroundings(ctx, db.FailOrphanedPendingDinnerGroundingsParams{ + GroundingError: sql.NullString{String: ReasonInterrupted, Valid: true}, + UpdatedAt: store.Now().String(), + Kind: KindGroundDinner, + }) + if err != nil { + return 0, fmt.Errorf("dinners: fail interrupted groundings: %w", err) + } + if n > 0 { + s.logger.WarnContext(ctx, "failed dinner groundings interrupted by restart", "count", n) + } + return n, nil +} + // reasonFor is the Reason for a grounding error; fromURL says whether the // dinner's recipe_url was grounded rather than its name. func reasonFor(err error, fromURL bool) string { @@ -236,8 +270,13 @@ func (s *Service) Ground(ctx context.Context, id string, f Fix, updatedBy string if _, err := s.get(ctx, id); err != nil { return Dinner{}, err } - gr, err := s.grounder.GroundURL(ctx, link) + groundCtx, cancel := context.WithTimeout(ctx, groundTimeout) + defer cancel() + gr, err := s.grounder.GroundURL(groundCtx, link) if err != nil { + if ctx.Err() == nil && errors.Is(groundCtx.Err(), context.DeadlineExceeded) { + return Dinner{}, ErrGroundTimeout + } return Dinner{}, fmt.Errorf("dinners: ground %s: %w", id, err) } row, err := db.New(s.st.Writer).GroundDinner(ctx, db.GroundDinnerParams{ diff --git a/internal/dinners/grounding_test.go b/internal/dinners/grounding_test.go index 39bf232..1d7e70e 100644 --- a/internal/dinners/grounding_test.go +++ b/internal/dinners/grounding_test.go @@ -33,6 +33,7 @@ type env struct { st *store.Store runner *jobs.Runner search *recipestest.Search + pages *recipestest.Pages model *llm.Fake } @@ -40,6 +41,15 @@ type env struct { // with fake search, pages, images and model. Searching "Chicken curry" // finds curryA, curryB and a missing page. func newEnv(t *testing.T) *env { + t.Helper() + e := newUnstartedEnv(t) + e.start(t) + return e +} + +// newUnstartedEnv is newEnv with its job runner not yet started, so jobs +// enqueued stay queued until start. +func newUnstartedEnv(t *testing.T) *env { t.Helper() dataDir := t.TempDir() st, err := store.Open(t.Context(), dataDir) @@ -79,16 +89,21 @@ func newEnv(t *testing.T) *env { if err != nil { t.Fatal(err) } + return &env{svc: svc, st: st, runner: runner, search: search, pages: pages, model: model} +} + +// start starts the env's job runner, stopping it when the test ends. +func (e *env) start(t *testing.T) { + t.Helper() ctx, cancel := context.WithCancel(context.Background()) - if err := runner.Start(ctx); err != nil { + if err := e.runner.Start(ctx); err != nil { cancel() t.Fatal(err) } t.Cleanup(func() { cancel() - runner.Wait() + e.runner.Wait() }) - return &env{svc: svc, st: st, runner: runner, search: search, model: model} } func ptr[T any](v T) *T { return &v } @@ -130,10 +145,10 @@ func TestCreateGroundsInBackground(t *testing.T) { if err != nil { t.Fatal(err) } - if c.GroundingStatus != dinners.GroundingPending || c.GroundingJobID == "" { + if c.GroundingStatus != dinners.GroundingPending || c.GroundingJobID == nil { t.Fatalf("created = %+v, want pending with a job id", c) } - job := e.waitJob(t, c.GroundingJobID) + job := e.waitJob(t, *c.GroundingJobID) if job.Kind != dinners.KindGroundDinner || job.Status != jobs.StatusSucceeded || string(job.Result) != `{"dinner_id":"`+c.ID+`","grounding_status":"grounded"}` { t.Fatalf("job = %+v (result %s)", job, job.Result) @@ -165,7 +180,7 @@ func TestCreateByRecipeURLGroundsThatURL(t *testing.T) { if err != nil { t.Fatal(err) } - if job := e.waitJob(t, c.GroundingJobID); job.Status != jobs.StatusSucceeded { + if job := e.waitJob(t, *c.GroundingJobID); job.Status != jobs.StatusSucceeded { t.Fatalf("job = %+v", job) } row := e.dinner(t, c.ID) @@ -208,7 +223,7 @@ func TestCreateGroundingFailureRecordsReason(t *testing.T) { if err != nil { t.Fatal(err) } - job := e.waitJob(t, c.GroundingJobID) + job := e.waitJob(t, *c.GroundingJobID) if job.Status != jobs.StatusFailed || job.Error == nil || *job.Error != tt.reason { t.Errorf("job = %+v, want failed with %s", job, tt.reason) } @@ -234,11 +249,11 @@ func TestCreateReturnsBeforeGroundingCompletes(t *testing.T) { if row := e.dinner(t, c.ID); row.GroundingStatus != "pending" { t.Fatalf("grounding_status = %s right after create, want pending", row.GroundingStatus) } - if job, err := e.runner.Get(t.Context(), c.GroundingJobID); err != nil || (job.Status != jobs.StatusQueued && job.Status != jobs.StatusRunning) { + if job, err := e.runner.Get(t.Context(), *c.GroundingJobID); err != nil || (job.Status != jobs.StatusQueued && job.Status != jobs.StatusRunning) { t.Fatalf("job = %+v, %v; want queued or running", job, err) } close(e.search.Gate) - if job := e.waitJob(t, c.GroundingJobID); job.Status != jobs.StatusSucceeded { + if job := e.waitJob(t, *c.GroundingJobID); job.Status != jobs.StatusSucceeded { t.Fatalf("job = %+v", job) } if row := e.dinner(t, c.ID); row.GroundingStatus != "grounded" { @@ -262,7 +277,7 @@ func TestManualFixWhilePendingIsKept(t *testing.T) { t.Fatal(err) } close(e.search.Gate) - job := e.waitJob(t, c.GroundingJobID) + job := e.waitJob(t, *c.GroundingJobID) if job.Status != jobs.StatusSucceeded || string(job.Result) != `{"dinner_id":"`+c.ID+`","grounding_status":"manual"}` { t.Errorf("job = %+v (result %s), want succeeded reporting manual", job, job.Result) } @@ -278,7 +293,7 @@ func failedDinner(t *testing.T, e *env, name string) string { if err != nil { t.Fatal(err) } - if job := e.waitJob(t, c.GroundingJobID); job.Status != jobs.StatusFailed { + if job := e.waitJob(t, *c.GroundingJobID); job.Status != jobs.StatusFailed { t.Fatalf("job = %+v, want failed", job) } return c.ID @@ -293,7 +308,7 @@ func TestCandidates(t *testing.T) { if err != nil { t.Fatal(err) } - if job := e.waitJob(t, c.GroundingJobID); job.Status != jobs.StatusSucceeded { + if job := e.waitJob(t, *c.GroundingJobID); job.Status != jobs.StatusSucceeded { t.Fatalf("job = %+v", job) } @@ -476,3 +491,105 @@ func TestFixValidation(t *testing.T) { t.Errorf("unknown dinner, link: err = %v, want ErrNotFound", err) } } + +// TestRestartFailsInterruptedGrounding simulates a crash: one dinner's job +// was left running and another dinner was saved before its job was +// enqueued. At the next boot both end failed as interrupted, while a +// dinner whose job is still queued stays pending and is grounded. +func TestRestartFailsInterruptedGrounding(t *testing.T) { + t.Parallel() + e := newUnstartedEnv(t) + e.search.Gate = make(chan struct{}) + e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40)) + + running, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Fish pie")}, member) + if err != nil { + t.Fatal(err) + } + if _, err := e.st.Writer.ExecContext(t.Context(), + `UPDATE jobs SET status = 'running', started_at = updated_at WHERE id = ?`, *running.GroundingJobID); err != nil { + t.Fatal(err) + } + unqueued, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Lentil soup")}, member) + if err != nil { + t.Fatal(err) + } + if _, err := e.st.Writer.ExecContext(t.Context(), `DELETE FROM jobs WHERE id = ?`, *unqueued.GroundingJobID); err != nil { + t.Fatal(err) + } + queued, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member) + if err != nil { + t.Fatal(err) + } + + // Boot: the runner fails the running job, then the service fails the + // dinners nothing is left to ground. The search is held, so the queued + // dinner's job cannot finish meanwhile. + e.start(t) + n, err := e.svc.FailInterruptedGroundings(t.Context()) + if err != nil { + t.Fatal(err) + } + if n != 2 { + t.Errorf("failed %d dinners, want 2", n) + } + if job, err := e.runner.Get(t.Context(), *running.GroundingJobID); err != nil || job.Status != jobs.StatusFailed { + t.Errorf("running job = %+v, %v; want failed", job, err) + } + for _, id := range []string{running.ID, unqueued.ID} { + if row := e.dinner(t, id); row.GroundingStatus != dinners.GroundingFailed || row.GroundingError.String != dinners.ReasonInterrupted { + t.Errorf("dinner %s = %s (%s), want failed (interrupted)", id, row.GroundingStatus, row.GroundingError.String) + } + } + if row := e.dinner(t, queued.ID); row.GroundingStatus != dinners.GroundingPending { + t.Errorf("queued dinner = %s, want pending", row.GroundingStatus) + } + close(e.search.Gate) + if job := e.waitJob(t, *queued.GroundingJobID); job.Status != jobs.StatusSucceeded { + t.Fatalf("queued job = %+v", job) + } + if row := e.dinner(t, queued.ID); row.GroundingStatus != dinners.GroundingGrounded { + t.Errorf("queued dinner = %s, want grounded", row.GroundingStatus) + } +} + +// TestCreateEnqueueFailureReturnsFailedDinner: a dinner saved when its job +// cannot be enqueued is returned, with its id, failed and no job id. +func TestCreateEnqueueFailureReturnsFailedDinner(t *testing.T) { + t.Parallel() + e := newEnv(t) + if _, err := e.st.Writer.ExecContext(t.Context(), + `CREATE TRIGGER refuse_jobs BEFORE INSERT ON jobs BEGIN SELECT RAISE(ABORT, 'no jobs'); END`); err != nil { + t.Fatal(err) + } + c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member) + if err != nil { + t.Fatal(err) + } + if c.ID == "" || c.GroundingJobID != nil || c.GroundingStatus != dinners.GroundingFailed || + c.GroundingError == nil || *c.GroundingError != dinners.ReasonGroundingError { + t.Fatalf("created = %+v, want its id, failed (grounding_error) and no job", c) + } + if row := e.dinner(t, c.ID); row.GroundingStatus != dinners.GroundingFailed { + t.Errorf("stored grounding_status = %s, want failed", row.GroundingStatus) + } +} + +// TestGroundTimesOut: grounding a link by hand that runs past its deadline +// is ErrGroundTimeout and leaves the dinner as it was. It shortens the +// package deadline, so it does not run in parallel. +func TestGroundTimesOut(t *testing.T) { + dinners.SetGroundTimeout(t, 50*time.Millisecond) + e := newEnv(t) + id := failedDinner(t, e, "Beef stew") + e.pages.Gate = make(chan struct{}) + defer close(e.pages.Gate) + + _, err := e.svc.Ground(t.Context(), id, dinners.Fix{RecipeURL: ptr(stew)}, member) + if !errors.Is(err, dinners.ErrGroundTimeout) { + t.Fatalf("err = %v, want ErrGroundTimeout", err) + } + if row := e.dinner(t, id); row.GroundingStatus != dinners.GroundingFailed { + t.Errorf("grounding_status = %s, want failed as before", row.GroundingStatus) + } +} diff --git a/internal/recipes/recipestest/recipestest.go b/internal/recipes/recipestest/recipestest.go index ea66af1..46aa72a 100644 --- a/internal/recipes/recipestest/recipestest.go +++ b/internal/recipes/recipestest/recipestest.go @@ -62,12 +62,20 @@ func (s *Search) Queries() []string { } // Pages is a recipes.PageFetcher serving Bodies by URL; any other URL is a -// 404. +// 404. Gate, when set, makes every fetch wait for it to close. type Pages struct { Bodies map[string][]byte + Gate chan struct{} } -func (p *Pages) Fetch(_ context.Context, rawURL string) (recipes.Page, error) { +func (p *Pages) Fetch(ctx context.Context, rawURL string) (recipes.Page, error) { + if p.Gate != nil { + select { + case <-p.Gate: + case <-ctx.Done(): + return recipes.Page{}, ctx.Err() + } + } body, ok := p.Bodies[rawURL] if !ok { return recipes.Page{}, &recipes.StatusError{URL: rawURL, Status: 404} diff --git a/internal/server/dinners.go b/internal/server/dinners.go index ae571d9..185b823 100644 --- a/internal/server/dinners.go +++ b/internal/server/dinners.go @@ -71,7 +71,8 @@ func getDinners(logger *slog.Logger, svc *dinners.Service) http.HandlerFunc { // postDinner adds a Regular, recording the session's actor as updated_by, // and answers 201 with it and its grounding_job_id without waiting for the -// job. +// job. When the job cannot be enqueued it still answers 201, with the dinner +// grounding_status failed and grounding_job_id null. func postDinner(logger *slog.Logger, svc *dinners.Service) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { actor, ok := actorOrFail(w, r, logger) @@ -198,7 +199,8 @@ func actorOrFail(w http.ResponseWriter, r *http.Request, logger *slog.Logger) (a // writeDinnerError maps a dinners.Service error to its API error. A page // that cannot ground the dinner is 422 ungroundable; recipe search not set -// up is 409 not_configured; a search or model failure is 502. +// up is 409 not_configured; a search or model failure is 502; grounding by +// hand that runs past dinners.GroundTimeout is 504 ground_timeout. func writeDinnerError(w http.ResponseWriter, r *http.Request, logger *slog.Logger, action string, err error) { var invalid *dinners.ValidationError var dup *dinners.DuplicateError @@ -211,11 +213,16 @@ func writeDinnerError(w http.ResponseWriter, r *http.Request, logger *slog.Logge case errors.Is(err, recipes.ErrSearchNotConfigured): writeError(w, r, logger, http.StatusConflict, "not_configured", "recipe search is not configured") case errors.Is(err, recipes.ErrSearchFailed): - logger.WarnContext(r.Context(), action, "error", err) + // The error may quote the search provider's response, so only the + // reason is logged. + logger.WarnContext(r.Context(), action, "reason", "search_failed") writeError(w, r, logger, http.StatusBadGateway, "search_failed", "recipe search failed") case errors.Is(err, recipes.ErrNormalizeFailed): logger.WarnContext(r.Context(), action, "reason", "normalize_recipe failed") writeError(w, r, logger, http.StatusBadGateway, "ai_failed", "the AI provider failed to read the recipe") + case errors.Is(err, dinners.ErrGroundTimeout): + logger.WarnContext(r.Context(), action, "reason", "ground_timeout") + writeError(w, r, logger, http.StatusGatewayTimeout, "ground_timeout", "grounding the recipe took too long; try again") case errors.As(err, &invalid): writeError(w, r, logger, http.StatusBadRequest, "invalid_dinner", invalid.Error()) case errors.As(err, &dup): diff --git a/internal/server/dinners_test.go b/internal/server/dinners_test.go index 3de2f56..402b90d 100644 --- a/internal/server/dinners_test.go +++ b/internal/server/dinners_test.go @@ -1,12 +1,14 @@ package server import ( + "bytes" "context" "io" "log/slog" "net/http" "net/http/httptest" "strings" + "sync" "testing" "testing/fstest" @@ -39,6 +41,26 @@ type dinnersEnv struct { pages *recipestest.Pages images *recipestest.Images model *llm.Fake + // logs holds everything the router and services log. + logs *logBuffer +} + +// logBuffer is a bytes.Buffer safe for concurrent log writes. +type logBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *logBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *logBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() } func newDinnersEnv(t *testing.T) *dinnersEnv { @@ -50,8 +72,9 @@ func newDinnersEnv(t *testing.T) *dinnersEnv { pages: &recipestest.Pages{Bodies: map[string][]byte{}}, images: &recipestest.Images{JPEGs: map[string][]byte{}}, model: &llm.Fake{}, + logs: &logBuffer{}, } - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + logger := slog.New(slog.NewTextHandler(e.logs, nil)) runner, err := jobs.NewRunner(st, logger, 1) if err != nil { t.Fatal(err) diff --git a/internal/server/grounding_test.go b/internal/server/grounding_test.go index b041c9a..1365f9e 100644 --- a/internal/server/grounding_test.go +++ b/internal/server/grounding_test.go @@ -1,6 +1,7 @@ package server import ( + "fmt" "net/http" "strings" "testing" @@ -162,12 +163,21 @@ func TestGetRecipeCandidates(t *testing.T) { func TestGetRecipeCandidatesSearchFails(t *testing.T) { t.Parallel() e := newDinnersEnv(t) - e.search.Err = recipes.ErrSearchRefused + // The provider's text must reach neither the response nor the logs. + const leak = "provider-said-token-abc123" + e.search.Err = fmt.Errorf("%w: %s", recipes.ErrSearchRefused, leak) id := createDinner(t, e.h, `{"name":"Chicken curry"}`) rec, body := callDinners(t, e.h, http.MethodGet, dinnersPath+"/"+id+"/recipe-candidates", "") if rec.Code != http.StatusBadGateway || errorCode(body) != "search_failed" { t.Errorf("GET = %d %v, want 502 search_failed", rec.Code, body) } + if strings.Contains(rec.Body.String(), leak) { + t.Errorf("response quotes the provider: %s", rec.Body) + } + logs := e.logs.String() + if strings.Contains(logs, leak) || !strings.Contains(logs, "reason=search_failed") { + t.Errorf("logs = %q, want reason=search_failed and no provider text", logs) + } } func TestPostGroundViaCandidate(t *testing.T) { @@ -271,3 +281,23 @@ func TestPostGroundRejects(t *testing.T) { t.Errorf("unknown dinner = %d %v, want 404", rec.Code, body) } } + +func TestPostDinnerEnqueueFailureAnswersFailedDinner(t *testing.T) { + t.Parallel() + e := newDinnersEnv(t) + if _, err := e.st.Writer.ExecContext(t.Context(), + `CREATE TRIGGER refuse_jobs BEFORE INSERT ON jobs BEGIN SELECT RAISE(ABORT, 'no jobs'); END`); err != nil { + t.Fatal(err) + } + rec, created := callDinners(t, e.h, http.MethodPost, dinnersPath, `{"name":"Chicken curry"}`) + if rec.Code != http.StatusCreated { + t.Fatalf("POST = %d: %s", rec.Code, rec.Body) + } + id, _ := created["id"].(string) + if id == "" || created["grounding_status"] != "failed" || created["grounding_error"] != "grounding_error" || created["grounding_job_id"] != nil { + t.Fatalf("created = %v, want its id, failed (grounding_error) and a null grounding_job_id", created) + } + if got := listedDinner(t, e.h, id); got["grounding_status"] != "failed" { + t.Errorf("listed = %v, want failed", got) + } +} diff --git a/migrations/queries/dinners.sql b/migrations/queries/dinners.sql index 362e11f..54eaa3d 100644 --- a/migrations/queries/dinners.sql +++ b/migrations/queries/dinners.sql @@ -109,3 +109,20 @@ SET grounding_status = 'failed', grounding_error = ?, updated_at = ? WHERE id = ? AND grounding_status = 'pending'; + +-- name: FailOrphanedPendingDinnerGroundings :execrows +-- Fails every dinner still pending whose grounding job is no longer queued +-- or running: the process stopped before the job, or before its handler +-- recorded a result. Run at boot, after the job runner fails the jobs a +-- previous process left running. +UPDATE dinners +SET grounding_status = 'failed', + grounding_error = sqlc.arg(grounding_error), + updated_at = sqlc.arg(updated_at) +WHERE grounding_status = 'pending' + AND NOT EXISTS ( + SELECT 1 FROM jobs + WHERE jobs.kind = sqlc.arg(kind) + AND jobs.status IN ('queued', 'running') + AND json_extract(jobs.payload, '$.dinner_id') = dinners.id + );