diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 203d754..e4ffb9c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -332,8 +332,12 @@ automatic** (D-021). taking its place before those shown. It fails with `no_ideas_found` (every idea was dropped), `ai_failed` (the skill or `normalize_recipe` failed; no idea is made up instead, D-017), `search_not_configured` or - `search_failed`. Before it fails it publishes its result again with no - ideas, so a failed job shows none, even those kept before the failure. + `search_failed`. A failed job shows no ideas, even those kept before the + failure: the kind registers a `FailedResult` view, so `GET + /api/v1/jobs/{id}` empties the `ideas` of any failed job's result on + read, however it failed (including `interrupted by restart` and an + `internal_error` whose clearing write failed). Before it fails it also + publishes its result again with no ideas, to tidy what is stored. `GET /api/v1/suggestions/deck` returns `{suggestions: [{id, dinner_id, name, photo_url, difficulty, method, minutes_total, cuisine, reason, cited_regular_ids, status, dinner_status, decided_at, created_at}], diff --git a/internal/ideas/ideas.go b/internal/ideas/ideas.go index f6460b4..d55b922 100644 --- a/internal/ideas/ideas.go +++ b/internal/ideas/ideas.go @@ -99,6 +99,9 @@ func NewService(grounder *recipes.Grounder, generate Generator, runner *jobs.Run if err := kind.ServiceOnly(); err != nil { return nil, err } + if err := kind.FailedResult(withoutIdeas); err != nil { + return nil, err + } s.job = kind return s, nil } @@ -192,8 +195,10 @@ func IsQuick(method string, minutes *int) bool { // other grounding failure fails the job, as does ideas_from_ingredients // failing, with no idea made up instead. The result is published at each // stage and again as each idea is kept, so it fills as the ideas ground. -// Before the job fails, its result is published again with no ideas, so a -// failed job never shows ideas kept before the failure. +// Before the job fails, its result is published again with no ideas, to +// tidy the stored result. That write is not what keeps a failed job from +// showing ideas, since it may fail and a job interrupted by restart never +// makes it: withoutIdeas, applied whenever a failed job is read, is. func (s *Service) run(ctx context.Context, p FromIngredientsJob, publish func(Result) error) (Result, error) { res, err := s.fill(ctx, p, publish) if err != nil { @@ -316,6 +321,18 @@ func (s *Service) ground(ctx context.Context, ideas []skills.Idea, kept func(i i return failed } +// withoutIdeas is a failed job's stored result as the job shows it: with no +// ideas, however the job failed, so a failed job never shows ideas kept +// before the failure. +func withoutIdeas(stored json.RawMessage) (json.RawMessage, error) { + var res Result + if err := json.Unmarshal(stored, &res); err != nil { + return nil, fmt.Errorf("ideas: decode failed job's result: %w", err) + } + res.Ideas = []Idea{} + return res.PublicJSON() +} + // ideaOf is idea as grounded by gr. func ideaOf(idea skills.Idea, gr recipes.Grounded) Idea { return Idea{ diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index b98e61a..e6b6561 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -135,12 +135,14 @@ type Job struct { type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (result json.RawMessage, code, reason string) // handler is a registered kind: run runs a job, check reports whether a -// payload decodes into the kind's payload type, and serviceOnly is set when -// Runner.Enqueue refuses the kind. +// payload decodes into the kind's payload type, serviceOnly is set when +// Runner.Enqueue refuses the kind, and failedResult, when set, is what Get +// shows of a failed job's stored result (see Kind.FailedResult). type handler struct { - run runFunc - check func(payload []byte) bool - serviceOnly bool + run runFunc + check func(payload []byte) bool + serviceOnly bool + failedResult func(stored json.RawMessage) (json.RawMessage, error) } // Runner owns the job handlers and the worker pool. @@ -307,6 +309,27 @@ func (k *Kind[P]) ServiceOnly() error { return nil } +// FailedResult makes Get return, for a failed job of the kind with a +// stored result, show's view of that result instead of the result itself. +// It is applied on every read, so it holds however the job failed: by its +// handler, as interrupted by restart, or after a write meant to tidy the +// result failed. A show error fails Get. A kind that never calls it shows a +// failed job's last result as stored. It must be called before Start. +func (k *Kind[P]) FailedResult(show func(stored json.RawMessage) (json.RawMessage, error)) error { + if show == nil { + return fmt.Errorf("jobs: failed result %q: nil func", k.name) + } + k.r.mu.Lock() + defer k.r.mu.Unlock() + if k.r.started { + return fmt.Errorf("jobs: failed result %q: runner already started", k.name) + } + h := k.r.handlers[k.name] + h.failedResult = show + k.r.handlers[k.name] = h + return nil +} + // Wake tells an idle worker to look for a job, as Enqueue does itself: // call it after committing the transaction a job was enqueued in. func (k *Kind[P]) Wake() { k.r.wakeOne() } @@ -402,7 +425,8 @@ func (r *Runner) Wait() { r.wg.Wait() } -// Get returns the job with id, or ErrNotFound. +// Get returns the job with id, or ErrNotFound. A failed job's result is +// shown through its kind's FailedResult, when the kind set one. func (r *Runner) Get(ctx context.Context, id string) (Job, error) { row, err := db.New(r.st.Reader).GetJob(ctx, id) if errors.Is(err, sql.ErrNoRows) { @@ -418,6 +442,21 @@ func (r *Runner) Get(ctx context.Context, id string) (Job, error) { if row.Error.Valid { job.Error = &row.Error.String } + if job.Status == StatusFailed && job.Result != nil { + r.mu.Lock() + show := r.handlers[job.Kind].failedResult + r.mu.Unlock() + if show != nil { + shown, err := show(job.Result) + if err != nil { + return Job{}, fmt.Errorf("jobs: show failed %s result: %w", job.Kind, err) + } + if !json.Valid(shown) { + return Job{}, fmt.Errorf("jobs: show failed %s result: not one JSON value", job.Kind) + } + job.Result = shown + } + } return job, nil } diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index ee7e152..79cb103 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -772,3 +772,47 @@ func TestServiceOnlyKind(t *testing.T) { t.Error("ServiceOnly after Start succeeded, want an error") } } + +func TestFailedResultAppliesOnlyToItsKindsFailedJobs(t *testing.T) { + t.Parallel() + st := migratedStore(t) + r := newRunner(t, st, 1) + h := func(_ context.Context, p countPayload, publish func(countResult) error) (countResult, error) { + if err := publish(countResult{Done: 5}); err != nil { + return countResult{}, err + } + if p.Steps == 0 { + return countResult{}, &Failure{Code: "stopped"} + } + return countResult{Done: p.Steps}, nil + } + hidden := register(t, r, "hidden", h, "stopped") + plain := register(t, r, "plain", h, "stopped") + if err := hidden.FailedResult(func(json.RawMessage) (json.RawMessage, error) { + return json.RawMessage(`{"done":0}`), nil + }); err != nil { + t.Fatal(err) + } + if err := hidden.FailedResult(nil); err == nil { + t.Error("FailedResult(nil) succeeded, want an error") + } + start(t, r) + if err := hidden.FailedResult(func(s json.RawMessage) (json.RawMessage, error) { return s, nil }); err == nil { + t.Error("FailedResult after Start succeeded, want an error") + } + + for _, tc := range []struct { + kind *Kind[countPayload] + steps int + want string + }{ + {hidden, 0, `{"done":0}`}, // failed: shown through FailedResult + {hidden, 3, `{"done":3}`}, // succeeded: as stored + {plain, 0, `{"done":5}`}, // failed, kind without FailedResult: as stored + } { + job := waitFor(t, r, enqueue(t, tc.kind, countPayload{Steps: tc.steps}), finished) + if string(job.Result) != tc.want { + t.Errorf("%s job with %d steps: %s result = %s, want %s", job.Kind, tc.steps, job.Status, job.Result, tc.want) + } + } +} diff --git a/internal/server/ideas_test.go b/internal/server/ideas_test.go index 1a4949d..3227743 100644 --- a/internal/server/ideas_test.go +++ b/internal/server/ideas_test.go @@ -2,6 +2,7 @@ package server import ( "context" + "database/sql" "encoding/json" "errors" "fmt" @@ -16,10 +17,12 @@ import ( "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/harness" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/skills" + "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ideas" "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" ) const ideasPath = "/api/v1/ideas/from-ingredients" @@ -27,8 +30,9 @@ const ideasPath = "/api/v1/ideas/from-ingredients" // newIdeasRouter returns a router whose ideas service runs // ideas_from_ingredients through the real harness with the scripted model // and grounds with fake search, pages and images and the scripted -// normalize. Its job runner is started when start is set. -func newIdeasRouter(t *testing.T, model, normalize *llm.Fake, search *recipestest.Search, pages *recipestest.Pages, images *recipestest.Images, start bool) (http.Handler, *jobs.Runner) { +// normalize, with its job runner and store. The runner is started when +// start is set. +func newIdeasRouter(t *testing.T, model, normalize *llm.Fake, search *recipestest.Search, pages *recipestest.Pages, images *recipestest.Images, start bool) (http.Handler, *jobs.Runner, *store.Store) { t.Helper() st := newTestStore(t) logger := slog.New(slog.NewTextHandler(io.Discard, nil)) @@ -60,22 +64,93 @@ func newIdeasRouter(t *testing.T, model, normalize *llm.Fake, search *recipestes t.Fatal(err) } if start { - ctx, cancel := context.WithCancel(context.Background()) - if err := runner.Start(ctx); err != nil { - cancel() - t.Fatal(err) - } - t.Cleanup(func() { - cancel() - runner.Wait() - }) + startRunner(t, runner) } - return h, runner + return h, runner, st +} + +// startRunner starts runner until the test ends. +func startRunner(t *testing.T, runner *jobs.Runner) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + if err := runner.Start(ctx); err != nil { + cancel() + t.Fatal(err) + } + t.Cleanup(func() { + cancel() + runner.Wait() + }) +} + +// scriptIdeas scripts ideas_from_ingredients to propose names, in order; +// a recipe grounds those in grounds, the others ground none, and search +// fails as fail says. It returns the router's fakes. +func scriptIdeas(t *testing.T, names, grounds []string, fail map[string]error) (*llm.Fake, *llm.Fake, *recipestest.Search, *recipestest.Pages, *recipestest.Images) { + t.Helper() + model, normalize := &llm.Fake{}, &llm.Fake{} + search := &recipestest.Search{Results: map[string][]recipes.Candidate{}, Fail: fail} + pages := &recipestest.Pages{Bodies: map[string][]byte{}} + images := &recipestest.Images{JPEGs: map[string][]byte{}} + for i, name := range grounds { + page, photo := fmt.Sprintf("https://r.example/%d", i), fmt.Sprintf("https://r.example/%d.jpg", i) + search.Results[name] = []recipes.Candidate{{URL: page}} + pages.Bodies[page] = recipestest.RecipePage(name, "R Kitchen", photo, 25) + images.JPEGs[photo] = recipestest.JPEG(t, 400, 320, uint8(i+1)) + normalize.QueueGenerate(recipestest.NormalizedReply("stovetop", 25)) + } + list := make([]string, len(names)) + for i, name := range names { + list[i] = fmt.Sprintf(`{"name":%q,"uses":["rice"],"missing":[],"est_minutes":30,"method":"stovetop","difficulty":"easy","search_query":%q}`, name, name) + } + js := `{"ideas":[` + strings.Join(list, ",") + `]}` + model.QueueGenerate(llm.GenerateResult{Response: llm.Response{Text: js, JSON: json.RawMessage(js)}}) + return model, normalize, search, pages, images +} + +// startIdeasJob posts a request for ideas from rice and returns the job id. +func startIdeasJob(t *testing.T, h http.Handler) string { + t.Helper() + rec, body := callDinners(t, h, http.MethodPost, ideasPath, `{"ingredients": ["rice"], "quick_only": false}`) + if rec.Code != http.StatusAccepted { + t.Fatalf("POST = %d: %s", rec.Code, rec.Body) + } + return body["jobId"].(string) +} + +// assertFailedWithoutIdeas checks that GET /jobs/{id} reports job failed +// with code and a result showing no ideas. +func assertFailedWithoutIdeas(t *testing.T, job map[string]any, code string) { + t.Helper() + if job["status"] != "failed" || job["error"] != code { + t.Fatalf("GET job = %v, want failed with %s", job, code) + } + res, _ := job["result"].(map[string]any) + if res == nil { + t.Fatalf("failed job result = %v, want one showing no ideas", job["result"]) + } + if list, ok := res["ideas"].([]any); !ok || len(list) != 0 { + t.Fatalf("failed job ideas = %v, want none", res["ideas"]) + } +} + +// storedIdeas is how many ideas job id's stored result holds. +func storedIdeas(t *testing.T, st *store.Store, id string) int { + t.Helper() + row, err := db.New(st.Reader).GetJob(t.Context(), id) + if err != nil { + t.Fatal(err) + } + var res ideas.Result + if err := json.Unmarshal([]byte(row.Result.String), &res); err != nil { + t.Fatal(err) + } + return len(res.Ideas) } func TestPostIdeasFromIngredientsRefusesBadRequests(t *testing.T) { t.Parallel() - h, _ := newIdeasRouter(t, &llm.Fake{}, &llm.Fake{}, &recipestest.Search{}, &recipestest.Pages{}, &recipestest.Images{}, false) + h, _, _ := newIdeasRouter(t, &llm.Fake{}, &llm.Fake{}, &recipestest.Search{}, &recipestest.Pages{}, &recipestest.Images{}, false) for _, tc := range []struct{ body, message string }{ {`{"quick_only": false}`, "ingredients: required"}, @@ -108,7 +183,7 @@ func TestPostIdeasFromIngredientsRunsTheJob(t *testing.T) { js := `{"ideas":[{"name":"Chicken Fried Rice","uses":["rice","chicken thighs","peas"],"missing":["2 eggs"],"est_minutes":25,"method":"stovetop","difficulty":"easy","search_query":"chicken fried rice"}]}` model.QueueGenerate(llm.GenerateResult{Response: llm.Response{Text: js, JSON: json.RawMessage(js)}}) normalize.QueueGenerate(recipestest.NormalizedReply("stovetop", 25)) - h, runner := newIdeasRouter(t, model, normalize, search, pages, images, true) + h, runner, _ := newIdeasRouter(t, model, normalize, search, pages, images, true) rec, body := callDinners(t, h, http.MethodPost, ideasPath, `{"ingredients": ["Chicken thighs", "rice", "scallions"], "quick_only": true}`) if rec.Code != http.StatusAccepted { @@ -169,9 +244,6 @@ func TestPostIdeasFromIngredientsRunsTheJob(t *testing.T) { func TestFailedIdeasJobShowsNoIdeas(t *testing.T) { t.Parallel() - idea := func(name string) string { - return fmt.Sprintf(`{"name":%q,"uses":["rice"],"missing":[],"est_minutes":30,"method":"stovetop","difficulty":"easy","search_query":%q}`, name, name) - } for _, tc := range []struct { name string // grounds are the ideas a recipe grounds; the others ground none. @@ -188,43 +260,82 @@ func TestFailedIdeasJobShowsNoIdeas(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { t.Parallel() - model, normalize := &llm.Fake{}, &llm.Fake{} - search := &recipestest.Search{Results: map[string][]recipes.Candidate{}, Fail: tc.fail} - pages := &recipestest.Pages{Bodies: map[string][]byte{}} - images := &recipestest.Images{JPEGs: map[string][]byte{}} - for i, name := range tc.grounds { - page, photo := fmt.Sprintf("https://r.example/%d", i), fmt.Sprintf("https://r.example/%d.jpg", i) - search.Results[name] = []recipes.Candidate{{URL: page}} - pages.Bodies[page] = recipestest.RecipePage(name, "R Kitchen", photo, 25) - images.JPEGs[photo] = recipestest.JPEG(t, 400, 320, uint8(i+1)) - normalize.QueueGenerate(recipestest.NormalizedReply("stovetop", 25)) - } - list := make([]string, len(tc.ideas)) - for i, name := range tc.ideas { - list[i] = idea(name) - } - js := `{"ideas":[` + strings.Join(list, ",") + `]}` - model.QueueGenerate(llm.GenerateResult{Response: llm.Response{Text: js, JSON: json.RawMessage(js)}}) - h, _ := newIdeasRouter(t, model, normalize, search, pages, images, true) - - rec, body := callDinners(t, h, http.MethodPost, ideasPath, `{"ingredients": ["rice"], "quick_only": false}`) - if rec.Code != http.StatusAccepted { - t.Fatalf("POST = %d: %s", rec.Code, rec.Body) - } - job := waitJob(t, h, body["jobId"].(string)) - if job["status"] != "failed" || job["error"] != tc.code { - t.Fatalf("GET job = %v, want failed with %s", job, tc.code) - } - if res, _ := job["result"].(map[string]any); res == nil || len(res["ideas"].([]any)) != 0 { - t.Fatalf("failed job result = %v, want no ideas", job["result"]) - } + model, normalize, search, pages, images := scriptIdeas(t, tc.ideas, tc.grounds, tc.fail) + h, _, _ := newIdeasRouter(t, model, normalize, search, pages, images, true) + assertFailedWithoutIdeas(t, waitJob(t, h, startIdeasJob(t, h)), tc.code) }) } } +// A job left running with ideas shown is failed at the next start as +// interrupted, without its handler running again; it still shows no ideas. +func TestIdeasJobInterruptedByRestartShowsNoIdeas(t *testing.T) { + t.Parallel() + h, runner, st := newIdeasRouter(t, &llm.Fake{}, &llm.Fake{}, &recipestest.Search{}, &recipestest.Pages{}, &recipestest.Images{}, false) + id := startIdeasJob(t, h) + + // The previous process claimed the job and showed an idea, then stopped. + q := db.New(st.Writer) + now := store.Now().String() + if _, err := q.ClaimNextJob(t.Context(), sql.NullString{String: now, Valid: true}); err != nil { + t.Fatal(err) + } + partial, err := ideas.Result{ + Stage: ideas.StageGrounding, + Ingredients: []string{"rice"}, + Ideas: []ideas.Idea{{Name: "Fried Rice", Uses: []string{"rice"}, Missing: []string{}, Method: "stovetop", Difficulty: "easy"}}, + }.PublicJSON() + if err != nil { + t.Fatal(err) + } + if n, err := q.PublishJobResult(t.Context(), db.PublishJobResultParams{ + Result: sql.NullString{String: string(partial), Valid: true}, UpdatedAt: now, ID: id, + }); err != nil || n != 1 { + t.Fatalf("publish partial result = %d, %v", n, err) + } + + startRunner(t, runner) + _, job := callDinners(t, h, http.MethodGet, "/api/v1/jobs/"+id, "") + assertFailedWithoutIdeas(t, job, jobs.CodeInterrupted) + if res := job["result"].(map[string]any); res["stage"] != "grounding" { + t.Errorf("result = %v, want the rest of the partial result kept", res) + } + if n := storedIdeas(t, st, id); n != 1 { + t.Fatalf("stored ideas = %d, want the interrupted job's 1 still stored", n) + } +} + +// When the write clearing a failed job's ideas itself fails, the job fails +// with the runner's internal_error and its stored result keeps the ideas; +// it still shows none. +func TestIdeasJobWhoseClearingWriteFailsShowsNoIdeas(t *testing.T) { + t.Parallel() + model, normalize, search, pages, images := scriptIdeas(t, + []string{"Fried Rice", "Mystery Stew", "Rice Bowls", "Rice Soup", "Rice Salad", "Rice Pilaf"}, + []string{"Fried Rice", "Rice Bowls", "Rice Soup", "Rice Salad"}, + map[string]error{"Rice Pilaf": errors.New("boom")}) + h, runner, st := newIdeasRouter(t, model, normalize, search, pages, images, false) + // Refuse any write that empties a running job's shown ideas. + if _, err := st.Writer.ExecContext(t.Context(), ` + CREATE TRIGGER refuse_clearing_ideas BEFORE UPDATE OF result ON jobs + WHEN NEW.status = 'running' + AND json_array_length(OLD.result, '$.ideas') > 0 + AND json_array_length(NEW.result, '$.ideas') = 0 + BEGIN SELECT RAISE(ABORT, 'clearing refused'); END`); err != nil { + t.Fatal(err) + } + startRunner(t, runner) + + id := startIdeasJob(t, h) + assertFailedWithoutIdeas(t, waitJob(t, h, id), jobs.CodeInternal) + if n := storedIdeas(t, st, id); n != 4 { + t.Fatalf("stored ideas = %d, want the 4 the refused write left", n) + } +} + func TestJobsEndpointRefusesIdeasKind(t *testing.T) { t.Parallel() - h, _ := newIdeasRouter(t, &llm.Fake{}, &llm.Fake{}, &recipestest.Search{}, &recipestest.Pages{}, &recipestest.Images{}, false) + h, _, _ := newIdeasRouter(t, &llm.Fake{}, &llm.Fake{}, &recipestest.Search{}, &recipestest.Pages{}, &recipestest.Images{}, false) rec, body := callDinners(t, h, http.MethodPost, "/api/v1/jobs", `{"kind": "ideas_from_ingredients", "payload": {"ingredients": [], "quick_only": false}}`) if rec.Code != http.StatusBadRequest || errorCode(body) != "service_only_kind" { t.Fatalf("POST /jobs = %d %v, want 400 service_only_kind", rec.Code, body)