Fail interrupted groundings at boot and bound hand grounding
CI / ci (push) Successful in 2m25s
CI / ci (pull_request) Successful in 2m20s

At boot, after the job runner fails the jobs a previous process left
running, FailInterruptedGroundings sets every dinner still pending with
no ground_dinner job queued or running to failed (interrupted). A crash
mid-job, or between saving a dinner and enqueuing its job, no longer
leaves the dinner pending forever.

A search failure on recipe-candidates or ground now logs only the reason
code search_failed, never the wrapped error, which may quote the
provider; the response stays redacted.

When the grounding job cannot be enqueued, POST /dinners answers 201
with the saved dinner failed (grounding_error) and grounding_job_id
null, instead of a 500 that hid the dinner it had created.

Grounding a candidate or link by hand runs under a 90-second deadline
(fetch, photo, normalize_recipe); past it the request answers 504
ground_timeout and the dinner is unchanged.

Maestro task: 0cbaa5bf-8be4-4cbd-aab2-891e7f1bc2b5

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-06 16:30:53 -04:00
co-authored by Claude Opus 5.5
parent d6d82e8368
commit f4bed5d82d
13 changed files with 357 additions and 37 deletions
+6
View File
@@ -139,6 +139,12 @@ func run(cfg config.Config, logger *slog.Logger) error {
if err := runner.Start(jobsCtx); err != nil { if err := runner.Start(jobsCtx); err != nil {
return err 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{ srv := &http.Server{
Addr: cfg.HTTPAddr, Addr: cfg.HTTPAddr,
Handler: router, Handler: router,
+13 -4
View File
@@ -202,7 +202,9 @@ automatic** (D-021).
the session's subject as `updated_by`; an unknown id is 404. the session's subject as `updated_by`; an unknown id is 404.
- Dinner grounding. `POST /api/v1/dinners` also enqueues a `ground_dinner` - Dinner grounding. `POST /api/v1/dinners` also enqueues a `ground_dinner`
job and answers with its `grounding_job_id` without waiting for it 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` `recipe_url` directly with `GroundURL`, else its name with `Ground`
(see Recipe grounding), and sets grounding_status `grounded` with the (see Recipe grounding), and sets grounding_status `grounded` with the
recipe's recipe_id, photo_id, method, minutes_total and difficulty, or 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`, `no_recipe_found`, `recipe_unusable` (the pasted link), `search_not_configured`,
`search_failed`, `ai_failed`, `interrupted` (the server stopped) or `search_failed`, `ai_failed`, `interrupted` (the server stopped) or
`grounding_error`. It only writes a dinner still `pending`, so a fix-up `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, searches for the dinner's name and returns `{candidates: [{candidate_id,
title, site_name, photo_url, minutes}]}`, at most 3, ranked as `Ground` 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, 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 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 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 ground it 422 `ungroundable` (the dinner is unchanged), recipe search not
set up 409 `not_configured`, and a search or model failure 502 set up 409 `not_configured`, a search or model failure 502
`search_failed` / `ai_failed`. `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. - `/health` — liveness.
- Everything else serves the embedded PWA with SPA fallback. - Everything else serves the embedded PWA with SPA fallback.
+32
View File
@@ -129,6 +129,38 @@ func (q *Queries) CreateDinner(ctx context.Context, arg CreateDinnerParams) (Din
return i, err 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 const failPendingDinnerGrounding = `-- name: FailPendingDinnerGrounding :execrows
UPDATE dinners UPDATE dinners
SET grounding_status = 'failed', SET grounding_status = 'failed',
+5
View File
@@ -34,6 +34,11 @@ type Querier interface {
DeleteRecipeIngredients(ctx context.Context, recipeID string) error DeleteRecipeIngredients(ctx context.Context, recipeID string) error
DeleteRecipeSteps(ctx context.Context, recipeID string) error DeleteRecipeSteps(ctx context.Context, recipeID string) error
FailJob(ctx context.Context, arg FailJobParams) (int64, 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 // 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). // dinner grounded some other way meanwhile is left as it is (0 rows).
FailPendingDinnerGrounding(ctx context.Context, arg FailPendingDinnerGroundingParams) (int64, error) FailPendingDinnerGrounding(ctx context.Context, arg FailPendingDinnerGroundingParams) (int64, error)
+26 -13
View File
@@ -155,10 +155,12 @@ func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner
return s, nil 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 { type Created struct {
Dinner 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 // 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 // origin regular and grounding_status pending, recording updatedBy, and
// enqueues the KindGroundDinner job that grounds it. It returns without // enqueues the KindGroundDinner job that grounds it. It returns without
// waiting for the job. A name repeating an unarchived dinner is a // 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) { func (s *Service) CreateRegular(ctx context.Context, r NewRegular, updatedBy string) (Created, error) {
d, err := s.createRegular(ctx, r, updatedBy) d, err := s.createRegular(ctx, r, updatedBy)
if err != nil { if err != nil {
return Created{}, err return Created{}, err
} }
jobID, err := s.groundJob.Enqueue(ctx, GroundJob{DinnerID: d.ID}) jobID, err := s.groundJob.Enqueue(ctx, GroundJob{DinnerID: d.ID})
if err != nil { if err == nil {
// The dinner is stored; record that it will not be grounded so it return Created{Dinner: d, GroundingJobID: &jobID}, nil
// 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)
} }
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) { func (s *Service) createRegular(ctx context.Context, r NewRegular, updatedBy string) (Dinner, error) {
+14
View File
@@ -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 })
}
+40 -1
View File
@@ -8,6 +8,7 @@ import (
"fmt" "fmt"
"slices" "slices"
"strings" "strings"
"time"
"unicode/utf8" "unicode/utf8"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
@@ -63,6 +64,18 @@ const (
manualSourcePrefix = "mealquest:manual:" 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. // GroundJob is the KindGroundDinner payload.
type GroundJob struct { type GroundJob struct {
DinnerID string `json:"dinner_id"` 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 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 // reasonFor is the Reason for a grounding error; fromURL says whether the
// dinner's recipe_url was grounded rather than its name. // dinner's recipe_url was grounded rather than its name.
func reasonFor(err error, fromURL bool) string { 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 { if _, err := s.get(ctx, id); err != nil {
return Dinner{}, err 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 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) return Dinner{}, fmt.Errorf("dinners: ground %s: %w", id, err)
} }
row, err := db.New(s.st.Writer).GroundDinner(ctx, db.GroundDinnerParams{ row, err := db.New(s.st.Writer).GroundDinner(ctx, db.GroundDinnerParams{
+129 -12
View File
@@ -33,6 +33,7 @@ type env struct {
st *store.Store st *store.Store
runner *jobs.Runner runner *jobs.Runner
search *recipestest.Search search *recipestest.Search
pages *recipestest.Pages
model *llm.Fake model *llm.Fake
} }
@@ -40,6 +41,15 @@ type env struct {
// with fake search, pages, images and model. Searching "Chicken curry" // with fake search, pages, images and model. Searching "Chicken curry"
// finds curryA, curryB and a missing page. // finds curryA, curryB and a missing page.
func newEnv(t *testing.T) *env { 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() t.Helper()
dataDir := t.TempDir() dataDir := t.TempDir()
st, err := store.Open(t.Context(), dataDir) st, err := store.Open(t.Context(), dataDir)
@@ -79,16 +89,21 @@ func newEnv(t *testing.T) *env {
if err != nil { if err != nil {
t.Fatal(err) 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()) ctx, cancel := context.WithCancel(context.Background())
if err := runner.Start(ctx); err != nil { if err := e.runner.Start(ctx); err != nil {
cancel() cancel()
t.Fatal(err) t.Fatal(err)
} }
t.Cleanup(func() { t.Cleanup(func() {
cancel() 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 } func ptr[T any](v T) *T { return &v }
@@ -130,10 +145,10 @@ func TestCreateGroundsInBackground(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) 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) 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 || if job.Kind != dinners.KindGroundDinner || job.Status != jobs.StatusSucceeded ||
string(job.Result) != `{"dinner_id":"`+c.ID+`","grounding_status":"grounded"}` { string(job.Result) != `{"dinner_id":"`+c.ID+`","grounding_status":"grounded"}` {
t.Fatalf("job = %+v (result %s)", job, job.Result) t.Fatalf("job = %+v (result %s)", job, job.Result)
@@ -165,7 +180,7 @@ func TestCreateByRecipeURLGroundsThatURL(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) 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) t.Fatalf("job = %+v", job)
} }
row := e.dinner(t, c.ID) row := e.dinner(t, c.ID)
@@ -208,7 +223,7 @@ func TestCreateGroundingFailureRecordsReason(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) 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 { if job.Status != jobs.StatusFailed || job.Error == nil || *job.Error != tt.reason {
t.Errorf("job = %+v, want failed with %s", job, 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" { if row := e.dinner(t, c.ID); row.GroundingStatus != "pending" {
t.Fatalf("grounding_status = %s right after create, want pending", row.GroundingStatus) 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) t.Fatalf("job = %+v, %v; want queued or running", job, err)
} }
close(e.search.Gate) 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) t.Fatalf("job = %+v", job)
} }
if row := e.dinner(t, c.ID); row.GroundingStatus != "grounded" { if row := e.dinner(t, c.ID); row.GroundingStatus != "grounded" {
@@ -262,7 +277,7 @@ func TestManualFixWhilePendingIsKept(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
close(e.search.Gate) 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"}` { 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) 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 { if err != nil {
t.Fatal(err) 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) t.Fatalf("job = %+v, want failed", job)
} }
return c.ID return c.ID
@@ -293,7 +308,7 @@ func TestCandidates(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) 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) t.Fatalf("job = %+v", job)
} }
@@ -476,3 +491,105 @@ func TestFixValidation(t *testing.T) {
t.Errorf("unknown dinner, link: err = %v, want ErrNotFound", err) 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)
}
}
+10 -2
View File
@@ -62,12 +62,20 @@ func (s *Search) Queries() []string {
} }
// Pages is a recipes.PageFetcher serving Bodies by URL; any other URL is a // 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 { type Pages struct {
Bodies map[string][]byte 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] body, ok := p.Bodies[rawURL]
if !ok { if !ok {
return recipes.Page{}, &recipes.StatusError{URL: rawURL, Status: 404} return recipes.Page{}, &recipes.StatusError{URL: rawURL, Status: 404}
+10 -3
View File
@@ -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, // 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 // 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 { func postDinner(logger *slog.Logger, svc *dinners.Service) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
actor, ok := actorOrFail(w, r, logger) 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 // writeDinnerError maps a dinners.Service error to its API error. A page
// that cannot ground the dinner is 422 ungroundable; recipe search not set // 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) { func writeDinnerError(w http.ResponseWriter, r *http.Request, logger *slog.Logger, action string, err error) {
var invalid *dinners.ValidationError var invalid *dinners.ValidationError
var dup *dinners.DuplicateError 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): case errors.Is(err, recipes.ErrSearchNotConfigured):
writeError(w, r, logger, http.StatusConflict, "not_configured", "recipe search is not configured") writeError(w, r, logger, http.StatusConflict, "not_configured", "recipe search is not configured")
case errors.Is(err, recipes.ErrSearchFailed): 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") writeError(w, r, logger, http.StatusBadGateway, "search_failed", "recipe search failed")
case errors.Is(err, recipes.ErrNormalizeFailed): case errors.Is(err, recipes.ErrNormalizeFailed):
logger.WarnContext(r.Context(), action, "reason", "normalize_recipe failed") 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") 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): case errors.As(err, &invalid):
writeError(w, r, logger, http.StatusBadRequest, "invalid_dinner", invalid.Error()) writeError(w, r, logger, http.StatusBadRequest, "invalid_dinner", invalid.Error())
case errors.As(err, &dup): case errors.As(err, &dup):
+24 -1
View File
@@ -1,12 +1,14 @@
package server package server
import ( import (
"bytes"
"context" "context"
"io" "io"
"log/slog" "log/slog"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"strings" "strings"
"sync"
"testing" "testing"
"testing/fstest" "testing/fstest"
@@ -39,6 +41,26 @@ type dinnersEnv struct {
pages *recipestest.Pages pages *recipestest.Pages
images *recipestest.Images images *recipestest.Images
model *llm.Fake 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 { func newDinnersEnv(t *testing.T) *dinnersEnv {
@@ -50,8 +72,9 @@ func newDinnersEnv(t *testing.T) *dinnersEnv {
pages: &recipestest.Pages{Bodies: map[string][]byte{}}, pages: &recipestest.Pages{Bodies: map[string][]byte{}},
images: &recipestest.Images{JPEGs: map[string][]byte{}}, images: &recipestest.Images{JPEGs: map[string][]byte{}},
model: &llm.Fake{}, 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) runner, err := jobs.NewRunner(st, logger, 1)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
+31 -1
View File
@@ -1,6 +1,7 @@
package server package server
import ( import (
"fmt"
"net/http" "net/http"
"strings" "strings"
"testing" "testing"
@@ -162,12 +163,21 @@ func TestGetRecipeCandidates(t *testing.T) {
func TestGetRecipeCandidatesSearchFails(t *testing.T) { func TestGetRecipeCandidatesSearchFails(t *testing.T) {
t.Parallel() t.Parallel()
e := newDinnersEnv(t) 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"}`) id := createDinner(t, e.h, `{"name":"Chicken curry"}`)
rec, body := callDinners(t, e.h, http.MethodGet, dinnersPath+"/"+id+"/recipe-candidates", "") rec, body := callDinners(t, e.h, http.MethodGet, dinnersPath+"/"+id+"/recipe-candidates", "")
if rec.Code != http.StatusBadGateway || errorCode(body) != "search_failed" { if rec.Code != http.StatusBadGateway || errorCode(body) != "search_failed" {
t.Errorf("GET = %d %v, want 502 search_failed", rec.Code, body) 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) { 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) 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)
}
}
+17
View File
@@ -109,3 +109,20 @@ SET grounding_status = 'failed',
grounding_error = ?, grounding_error = ?,
updated_at = ? updated_at = ?
WHERE id = ? AND grounding_status = 'pending'; 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
);