Integration fix: Feature 12 feature-review findings F1-F8 (#95)
CI / ci (pull_request) Successful in 2m23s
CI / ci (push) Successful in 2m27s

Co-authored-by: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com>
This commit was merged in pull request #95.
This commit is contained in:
2026-10-06 19:54:47 -04:00
committed by approver_agent
parent 171c9c2d11
commit cc7875c76c
26 changed files with 2402 additions and 257 deletions
+71 -46
View File
@@ -11,9 +11,11 @@ import (
"errors"
"fmt"
"log/slog"
"math"
"net/url"
"slices"
"strings"
"sync"
"unicode"
"unicode/utf8"
@@ -130,6 +132,11 @@ type Service struct {
grounder *recipes.Grounder
logger *slog.Logger
groundJob *jobs.Kind[GroundJob]
// groundingsMu guards groundings: the KindGroundDinner jobs running,
// by dinner id, so archiving or renaming a dinner can stop them.
groundingsMu sync.Mutex
groundings map[string]map[*runningGrounding]struct{}
}
// NewService returns a service over st that grounds dinners with grounder
@@ -146,7 +153,7 @@ func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner
case logger == nil:
return nil, errors.New("dinners: nil logger")
}
s := &Service{st: st, grounder: grounder, logger: logger}
s := &Service{st: st, grounder: grounder, logger: logger, groundings: make(map[string]map[*runningGrounding]struct{})}
kind, err := jobs.Register(runner, KindGroundDinner, s.runGroundJob, Reasons...)
if err != nil {
return nil, err
@@ -176,7 +183,7 @@ func (s *Service) CreateRegular(ctx context.Context, r NewRegular, updatedBy str
if err != nil {
return Created{}, err
}
jobID, err := s.enqueueGrounding(ctx, d.ID, harness.Normalize(d.Name))
jobID, err := s.enqueueGrounding(ctx, d.ID, d.GroundingRevision)
if err != nil {
return Created{}, err
}
@@ -187,60 +194,59 @@ func (s *Service) CreateRegular(ctx context.Context, r NewRegular, updatedBy str
}
return Created{Dinner: fromRow(row)}, nil
}
return Created{Dinner: d, GroundingJobID: jobID}, nil
return Created{Dinner: fromRow(d), GroundingJobID: jobID}, nil
}
// enqueueGrounding enqueues the KindGroundDinner job for pending dinner id,
// whose normalized name is normalizedName, and returns its id. When the job
// cannot be enqueued, the dinner's grounding is failed with
// ReasonGroundingError, so it is not left pending with no job, and the id
// is nil.
func (s *Service) enqueueGrounding(ctx context.Context, id, normalizedName string) (*string, error) {
jobID, err := s.groundJob.Enqueue(ctx, GroundJob{DinnerID: id})
// enqueueGrounding enqueues the KindGroundDinner job for pending dinner id
// at grounding_revision revision and returns its id. When the job cannot be
// enqueued, the dinner's grounding is failed with ReasonGroundingError, so
// it is not left pending with no job, and the id is nil.
func (s *Service) enqueueGrounding(ctx context.Context, id string, revision int64) (*string, error) {
jobID, err := s.groundJob.Enqueue(ctx, GroundJob{DinnerID: id, Revision: revision})
if err == nil {
return &jobID, nil
}
// The error is a database error, never a dependency's text.
s.logger.ErrorContext(ctx, "dinner grounding not enqueued", "dinner_id", id, "error", err)
if _, err := db.New(s.st.Writer).FailPendingDinnerGrounding(context.WithoutCancel(ctx), db.FailPendingDinnerGroundingParams{
GroundingError: sql.NullString{String: ReasonGroundingError, Valid: true},
UpdatedAt: store.Now().String(),
ID: id,
NormalizedName: normalizedName,
GroundingError: sql.NullString{String: ReasonGroundingError, Valid: true},
UpdatedAt: store.Now().String(),
ID: id,
GroundingRevision: revision,
}); err != nil {
return nil, fmt.Errorf("dinners: record failed grounding of %s: %w", id, err)
}
return nil, 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) (db.Dinner, error) {
if updatedBy == "" {
return Dinner{}, errors.New("dinners: create: no member to record as updated_by")
return db.Dinner{}, errors.New("dinners: create: no member to record as updated_by")
}
var name string
var recipeURL sql.NullString
switch {
case r.Name != nil && r.RecipeURL != nil:
return Dinner{}, invalid("name", "give a name or a recipe_url, not both")
return db.Dinner{}, invalid("name", "give a name or a recipe_url, not both")
case r.Name != nil:
n, err := checkName(*r.Name)
if err != nil {
return Dinner{}, err
return db.Dinner{}, err
}
name = n
case r.RecipeURL != nil:
u, err := checkRecipeURL(*r.RecipeURL)
if err != nil {
return Dinner{}, err
return db.Dinner{}, err
}
n, ok := nameFromURL(u)
if !ok {
return Dinner{}, invalid("recipe_url", "the link does not name a dinner; add it by name instead")
return db.Dinner{}, invalid("recipe_url", "the link does not name a dinner; add it by name instead")
}
name = n
recipeURL = sql.NullString{String: u.String(), Valid: true}
default:
return Dinner{}, invalid("name", "give a name or a recipe_url")
return db.Dinner{}, invalid("name", "give a name or a recipe_url")
}
var row db.Dinner
@@ -269,9 +275,9 @@ func (s *Service) createRegular(ctx context.Context, r NewRegular, updatedBy str
return nil
})
if err != nil {
return Dinner{}, err
return db.Dinner{}, err
}
return fromRow(row), nil
return row, nil
}
// List returns the dinners with status, which must be regular or
@@ -294,8 +300,10 @@ func (s *Service) List(ctx context.Context, status string) ([]Dinner, error) {
// Update applies p to dinner id, recording updatedBy. An invalid patch is a
// *ValidationError and a new name repeating another unarchived dinner a
// *DuplicateError; either changes nothing. Renaming a dinner still pending
// enqueues a fresh KindGroundDinner job for the new name: the job already
// running was for the old name, and its result no longer applies.
// clears its recipe_url, raises its grounding_revision and enqueues a fresh
// KindGroundDinner job that grounds the new name: a job queued or running
// for the old name or link carries the old revision, so it can no longer
// write the dinner, and one running is stopped.
func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy string) (Dinner, error) {
if updatedBy == "" {
return Dinner{}, errors.New("dinners: update: no member to record as updated_by")
@@ -317,20 +325,22 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
return fmt.Errorf("dinners: read %s: %w", id, err)
}
params := db.UpdateDinnerParams{
ID: cur.ID,
Name: cur.Name,
NormalizedName: cur.NormalizedName,
Status: cur.Status,
Origin: cur.Origin,
Method: cur.Method,
MinutesTotal: cur.MinutesTotal,
Difficulty: cur.Difficulty,
Cuisine: cur.Cuisine,
RecipeID: cur.RecipeID,
PhotoID: cur.PhotoID,
GroundingStatus: cur.GroundingStatus,
UpdatedBy: sql.NullString{String: updatedBy, Valid: true},
UpdatedAt: store.Now().String(),
ID: cur.ID,
Name: cur.Name,
NormalizedName: cur.NormalizedName,
Status: cur.Status,
Origin: cur.Origin,
Method: cur.Method,
MinutesTotal: cur.MinutesTotal,
Difficulty: cur.Difficulty,
Cuisine: cur.Cuisine,
RecipeID: cur.RecipeID,
RecipeUrl: cur.RecipeUrl,
PhotoID: cur.PhotoID,
GroundingStatus: cur.GroundingStatus,
GroundingRevision: cur.GroundingRevision,
UpdatedBy: sql.NullString{String: updatedBy, Valid: true},
UpdatedAt: store.Now().String(),
}
if p.Name != nil {
if err := checkUnique(ctx, q, cur.ID, name, sql.NullString{}); err != nil {
@@ -338,6 +348,12 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
}
params.Name = name
params.NormalizedName = harness.Normalize(name)
if cur.GroundingStatus == GroundingPending && params.NormalizedName != cur.NormalizedName {
// The link was the old dinner's recipe; the new name is
// grounded by name, and a link can be given by hand.
params.RecipeUrl = sql.NullString{}
params.GroundingRevision++
}
}
if p.Method != nil {
params.Method = sql.NullString{String: *p.Method, Valid: true}
@@ -352,14 +368,15 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
if err != nil {
return fmt.Errorf("dinners: update %s: %w", id, err)
}
regrounding = row.GroundingStatus == GroundingPending && row.NormalizedName != cur.NormalizedName
regrounding = row.GroundingRevision != cur.GroundingRevision
return nil
})
if err != nil {
return Dinner{}, err
}
if regrounding {
jobID, err := s.enqueueGrounding(ctx, row.ID, row.NormalizedName)
s.stopGroundings(row.ID, row.GroundingRevision)
jobID, err := s.enqueueGrounding(ctx, row.ID, row.GroundingRevision)
if err != nil {
return Dinner{}, err
}
@@ -374,7 +391,8 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
}
// Archive sets dinner id's status to archived, recording updatedBy. The row
// is kept, so its name still counts for Side quest dedupe.
// is kept, so its name still counts for Side quest dedupe. A
// KindGroundDinner job running for it is stopped.
func (s *Service) Archive(ctx context.Context, id, updatedBy string) (Dinner, error) {
if updatedBy == "" {
return Dinner{}, errors.New("dinners: archive: no member to record as updated_by")
@@ -390,6 +408,7 @@ func (s *Service) Archive(ctx context.Context, id, updatedBy string) (Dinner, er
if err != nil {
return Dinner{}, fmt.Errorf("dinners: archive %s: %w", id, err)
}
s.stopGroundings(row.ID, math.MaxInt64)
return fromRow(row), nil
}
@@ -507,18 +526,24 @@ func nameFromURL(u *url.URL) (name string, ok bool) {
}
// checkUnique returns a *DuplicateError when name is harness.Similar to an
// unarchived dinner other than self, or recipeURL is one's recipe link.
// unarchived dinner other than self, or recipeURL is one's recipe link: the
// source_url of the recipe it is grounded in, else the recipe_url still
// awaiting grounding.
func checkUnique(ctx context.Context, q *db.Queries, self, name string, recipeURL sql.NullString) error {
rows, err := q.ListUnarchivedDinners(ctx)
if err != nil {
return fmt.Errorf("dinners: list for dedupe: %w", err)
}
for _, row := range rows {
if row.ID == self {
if row.Dinner.ID == self {
continue
}
if harness.Similar(name, row.Name) || (recipeURL.Valid && row.RecipeUrl == recipeURL) {
return &DuplicateError{Existing: fromRow(row)}
link := row.Dinner.RecipeUrl
if row.RecipeSourceUrl.Valid {
link = row.RecipeSourceUrl
}
if harness.Similar(name, row.Dinner.Name) || (recipeURL.Valid && link == recipeURL) {
return &DuplicateError{Existing: fromRow(row.Dinner)}
}
}
return nil
+9
View File
@@ -1,6 +1,7 @@
package dinners
import (
"context"
"testing"
"time"
)
@@ -12,3 +13,11 @@ func SetGroundTimeout(t *testing.T, d time.Duration) {
groundTimeout = d
t.Cleanup(func() { groundTimeout = old })
}
// RunGroundJob runs the KindGroundDinner handler for p in the caller's
// goroutine, as the job runner would, and returns the grounding status it
// reports.
func (s *Service) RunGroundJob(ctx context.Context, p GroundJob) (string, error) {
r, err := s.runGroundJob(ctx, p, nil)
return r.GroundingStatus, err
}
+101 -33
View File
@@ -76,14 +76,17 @@ var groundTimeout = GroundTimeout
// 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: the dinner to ground and its
// grounding_revision when the job was enqueued.
type GroundJob struct {
DinnerID string `json:"dinner_id"`
Revision int64 `json:"revision"`
}
// groundResult is the KindGroundDinner result: the dinner's grounding
// status once the job ends. It is not grounded when the dinner was
// grounded by hand before the job finished.
// status once the job ends. It is not grounded by this job when the dinner
// was grounded by hand before the job finished, renamed (a later job
// grounds it) or archived; the job then reports the status it found.
type groundResult struct {
DinnerID string `json:"dinner_id"`
GroundingStatus string `json:"grounding_status"`
@@ -93,26 +96,38 @@ func (r groundResult) PublicJSON() ([]byte, error) { return json.Marshal(r) }
// runGroundJob grounds a pending dinner as its creation asked: its
// recipe_url with GroundURL, else its name with Ground. Success records
// the recipe, photo, method, minutes and difficulty with grounding_status
// grounded; failure records grounding_status failed with its Reason, which
// is also the job's failure code. Either is recorded only while the dinner
// is still pending under the normalized name read here: a dinner grounded
// by hand meanwhile is left as it is, and one renamed meanwhile is left
// pending for the job its rename enqueued.
// the recipe and photo with grounding_status grounded, and the recipe's
// method, minutes and difficulty where a member has not set them; failure
// records grounding_status failed with its Reason, which is also the job's
// failure code. A dinner archived, no longer pending, or at a
// grounding_revision other than p.Revision is not grounded at all, and
// either result is recorded only while the dinner is still unarchived,
// pending and at p.Revision: a dinner grounded by hand meanwhile is left as
// it is, and one renamed or archived meanwhile is left pending — for the
// job its rename enqueued, or for none. Archiving or renaming the dinner
// stops the grounding where it is, so no further search, fetch, photo
// download or model call is made for it.
func (s *Service) runGroundJob(ctx context.Context, p GroundJob, _ func(groundResult) error) (groundResult, error) {
// Tracked before the dinner is read: an archive or rename committed
// after the read finds this job to stop.
groundCtx, done := s.trackGrounding(ctx, p.DinnerID, p.Revision)
defer done()
row, err := db.New(s.st.Reader).GetDinner(ctx, p.DinnerID)
if err != nil {
return groundResult{}, fmt.Errorf("dinners: ground: read %s: %w", p.DinnerID, err)
}
if row.GroundingStatus != GroundingPending {
if row.Status == StatusArchived || row.GroundingStatus != GroundingPending || row.GroundingRevision != p.Revision {
return groundResult{DinnerID: row.ID, GroundingStatus: row.GroundingStatus}, nil
}
var gr recipes.Grounded
if row.RecipeUrl.Valid {
gr, err = s.grounder.GroundURL(ctx, row.RecipeUrl.String)
gr, err = s.grounder.GroundURL(groundCtx, row.RecipeUrl.String)
} else {
gr, err = s.grounder.Ground(ctx, row.Name, row.Name)
gr, err = s.grounder.Ground(groundCtx, row.Name, row.Name)
}
if err != nil && ctx.Err() == nil && errors.Is(context.Cause(groundCtx), errGroundingStopped) {
return s.currentGrounding(ctx, row.ID)
}
if err != nil {
reason := reasonFor(err, row.RecipeUrl.Valid)
@@ -124,10 +139,10 @@ func (s *Service) runGroundJob(ctx context.Context, p GroundJob, _ func(groundRe
s.logger.InfoContext(ctx, "dinner grounding failed", "dinner_id", row.ID, "reason", reason)
write := context.WithoutCancel(ctx)
if _, failErr := db.New(s.st.Writer).FailPendingDinnerGrounding(write, db.FailPendingDinnerGroundingParams{
GroundingError: sql.NullString{String: reason, Valid: true},
UpdatedAt: store.Now().String(),
ID: row.ID,
NormalizedName: row.NormalizedName,
GroundingError: sql.NullString{String: reason, Valid: true},
UpdatedAt: store.Now().String(),
ID: row.ID,
GroundingRevision: p.Revision,
}); failErr != nil {
return groundResult{}, fmt.Errorf("dinners: ground: record failure of %s: %w", row.ID, failErr)
}
@@ -138,32 +153,85 @@ func (s *Service) runGroundJob(ctx context.Context, p GroundJob, _ func(groundRe
}
n, err := db.New(s.st.Writer).GroundPendingDinner(ctx, db.GroundPendingDinnerParams{
RecipeID: sql.NullString{String: gr.RecipeID, Valid: true},
PhotoID: sql.NullString{String: gr.PhotoID, Valid: true},
Method: sql.NullString{String: gr.Method, Valid: true},
MinutesTotal: nullInt(gr.MinutesTotal),
Difficulty: sql.NullString{String: gr.Difficulty, Valid: true},
UpdatedAt: store.Now().String(),
ID: row.ID,
NormalizedName: row.NormalizedName,
RecipeID: sql.NullString{String: gr.RecipeID, Valid: true},
PhotoID: sql.NullString{String: gr.PhotoID, Valid: true},
Method: sql.NullString{String: gr.Method, Valid: true},
MinutesTotal: nullInt(gr.MinutesTotal),
Difficulty: sql.NullString{String: gr.Difficulty, Valid: true},
UpdatedAt: store.Now().String(),
ID: row.ID,
GroundingRevision: p.Revision,
})
if err != nil {
return groundResult{}, fmt.Errorf("dinners: ground: record %s: %w", row.ID, err)
}
if n == 0 {
cur, err := db.New(s.st.Reader).GetDinner(ctx, row.ID)
if err != nil {
return groundResult{}, fmt.Errorf("dinners: ground: read %s: %w", row.ID, err)
}
return groundResult{DinnerID: cur.ID, GroundingStatus: cur.GroundingStatus}, nil
return s.currentGrounding(ctx, row.ID)
}
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
// currentGrounding is the KindGroundDinner result of a job that did not
// ground dinner id: the grounding status the dinner has now.
func (s *Service) currentGrounding(ctx context.Context, id string) (groundResult, error) {
cur, err := db.New(s.st.Reader).GetDinner(ctx, id)
if err != nil {
return groundResult{}, fmt.Errorf("dinners: ground: read %s: %w", id, err)
}
return groundResult{DinnerID: cur.ID, GroundingStatus: cur.GroundingStatus}, nil
}
// errGroundingStopped is the cause a running KindGroundDinner job's
// grounding is cancelled with when its dinner is archived or renamed.
var errGroundingStopped = errors.New("dinners: grounding stopped: dinner archived or renamed")
// runningGrounding is a KindGroundDinner job running for a dinner at
// revision, with the cancel of its grounding.
type runningGrounding struct {
revision int64
cancel context.CancelCauseFunc
}
// trackGrounding returns a context for grounding dinnerID at revision that
// stopGroundings cancels, and the func to call when the job ends.
func (s *Service) trackGrounding(ctx context.Context, dinnerID string, revision int64) (context.Context, func()) {
groundCtx, cancel := context.WithCancelCause(ctx)
r := &runningGrounding{revision: revision, cancel: cancel}
s.groundingsMu.Lock()
if s.groundings[dinnerID] == nil {
s.groundings[dinnerID] = make(map[*runningGrounding]struct{})
}
s.groundings[dinnerID][r] = struct{}{}
s.groundingsMu.Unlock()
return groundCtx, func() {
s.groundingsMu.Lock()
delete(s.groundings[dinnerID], r)
if len(s.groundings[dinnerID]) == 0 {
delete(s.groundings, dinnerID)
}
s.groundingsMu.Unlock()
cancel(nil)
}
}
// stopGroundings cancels, with errGroundingStopped, the grounding of every
// KindGroundDinner job running for dinnerID at a revision below below. Call
// it once the change that makes those jobs stale is committed.
func (s *Service) stopGroundings(dinnerID string, below int64) {
s.groundingsMu.Lock()
defer s.groundingsMu.Unlock()
for r := range s.groundings[dinnerID] {
if r.revision < below {
r.cancel(errGroundingStopped)
}
}
}
// FailInterruptedGroundings fails, with ReasonInterrupted, every unarchived
// 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. An archived dinner is left pending, since
// archiving, not a restart, stopped its grounding. 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) {
+413 -14
View File
@@ -2,11 +2,15 @@ package dinners_test
import (
"context"
"database/sql"
"encoding/json"
"errors"
"io"
"io/fs"
"log/slog"
"slices"
"testing"
"testing/fstest"
"time"
"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/ai/llm"
@@ -52,6 +56,13 @@ func newEnv(t *testing.T) *env {
func newUnstartedEnv(t *testing.T) *env {
t.Helper()
dataDir := t.TempDir()
return newUnstartedEnvOn(t, openStore(t, dataDir), dataDir)
}
// openStore opens the store in dataDir, closing it when the test ends. It
// is not migrated.
func openStore(t *testing.T, dataDir string) *store.Store {
t.Helper()
st, err := store.Open(t.Context(), dataDir)
if err != nil {
t.Fatal(err)
@@ -61,6 +72,13 @@ func newUnstartedEnv(t *testing.T) *env {
t.Error(err)
}
})
return st
}
// newUnstartedEnvOn is newUnstartedEnv over st, the store in dataDir,
// which it migrates to the latest version.
func newUnstartedEnvOn(t *testing.T, st *store.Store, dataDir string) *env {
t.Helper()
if err := store.Migrate(t.Context(), st.Writer, migrations.FS); err != nil {
t.Fatal(err)
}
@@ -288,19 +306,16 @@ func TestManualFixWhilePendingIsKept(t *testing.T) {
// TestRenameWhilePendingGroundsNewName: a Regular renamed while its job is
// still grounding the old name never gets the old name's result, whether
// that job succeeds or fails; the job the rename enqueues grounds the new
// name.
// the old name would be found or not: the rename stops that job where it
// is, and the job the rename enqueues grounds the new name.
func TestRenameWhilePendingGroundsNewName(t *testing.T) {
t.Parallel()
tests := []struct {
name string
oldName string
replies []llm.GenerateResult
}{
{"old name found", "Chicken curry", []llm.GenerateResult{
recipestest.NormalizedReply("stovetop", 40), recipestest.NormalizedReply("oven", 150),
}},
{"old name not found", "Unknown dish", []llm.GenerateResult{recipestest.NormalizedReply("oven", 150)}},
{"old name found", "Chicken curry"},
{"old name not found", "Unknown dish"},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -308,9 +323,7 @@ func TestRenameWhilePendingGroundsNewName(t *testing.T) {
e := newEnv(t)
e.search.Results["Beef stew"] = []recipes.Candidate{{URL: stew}}
e.search.Gate = make(chan struct{})
for _, r := range tt.replies {
e.model.QueueGenerate(r)
}
e.model.QueueGenerate(recipestest.NormalizedReply("oven", 150))
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr(tt.oldName)}, member)
if err != nil {
@@ -325,12 +338,13 @@ func TestRenameWhilePendingGroundsNewName(t *testing.T) {
if d.GroundingStatus != dinners.GroundingPending {
t.Fatalf("renamed = %+v, want still pending", d)
}
// The old name's job ends while its search is still held.
first := e.waitJob(t, *c.GroundingJobID)
if first.Status != jobs.StatusSucceeded || string(first.Result) != `{"dinner_id":"`+c.ID+`","grounding_status":"pending"}` {
t.Fatalf("old name's job = %+v (result %s), want succeeded leaving the dinner pending", first, first.Result)
}
close(e.search.Gate)
first := e.waitJob(t, *c.GroundingJobID)
if string(first.Result) == `{"dinner_id":"`+c.ID+`","grounding_status":"grounded"}` {
t.Fatalf("old name's job reported grounded: %+v (result %s)", first, first.Result)
}
row := e.waitGrounding(t, c.ID)
if row.GroundingStatus != dinners.GroundingGrounded {
t.Fatalf("dinner = %+v, want grounded for the new name", row)
@@ -345,10 +359,222 @@ func TestRenameWhilePendingGroundsNewName(t *testing.T) {
if got := e.search.Queries(); !slices.Equal(got, []string{tt.oldName, "Beef stew"}) {
t.Errorf("searched %q, want the old name then the new", got)
}
if got := len(e.model.Requests()); got != 1 {
t.Errorf("model called %d times, want once, for the new name only", got)
}
})
}
}
// TestRenameLinkWhilePendingGroundsNewName: a Regular added from a link and
// renamed while still pending drops the link and grounds its new name, not
// the old link's recipe and photo.
func TestRenameLinkWhilePendingGroundsNewName(t *testing.T) {
t.Parallel()
e := newUnstartedEnv(t)
e.search.Results["Beef stew"] = []recipes.Candidate{{URL: stew}}
e.model.QueueGenerate(recipestest.NormalizedReply("oven", 150))
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{RecipeURL: ptr(curryA)}, member)
if err != nil {
t.Fatal(err)
}
d, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Name: ptr("Beef stew")}, member)
if err != nil {
t.Fatal(err)
}
if d.GroundingStatus != dinners.GroundingPending || d.RecipeURL != nil {
t.Fatalf("renamed = %+v, want still pending with no recipe_url", d)
}
e.start(t)
row := e.waitGrounding(t, c.ID)
if row.GroundingStatus != dinners.GroundingGrounded {
t.Fatalf("dinner = %+v, want grounded for the new name", row)
}
rec, err := db.New(e.st.Reader).GetRecipe(t.Context(), row.RecipeID.String)
if err != nil {
t.Fatal(err)
}
if rec.SourceUrl != stew || rec.PhotoID != row.PhotoID || row.Method.String != "oven" || row.MinutesTotal.Int64 != 150 {
t.Errorf("dinner = %+v recipe %s, want grounded in %s, not the old link %s", row, rec.SourceUrl, stew, curryA)
}
if got := e.search.Queries(); !slices.Equal(got, []string{"Beef stew"}) {
t.Errorf("searched %q, want only the new name", got)
}
if row.RecipeUrl.Valid {
t.Errorf("recipe_url = %q, want the old link dropped", row.RecipeUrl.String)
}
if _, err := db.New(e.st.Reader).GetRecipeBySourceURL(t.Context(), curryA); !errors.Is(err, sql.ErrNoRows) {
t.Errorf("old link's recipe lookup err = %v, want it never grounded (sql.ErrNoRows)", err)
}
}
// queuedJob reads the payload of queued KindGroundDinner job id.
func (e *env) queuedJob(t *testing.T, id string) dinners.GroundJob {
t.Helper()
var payload string
if err := e.st.Reader.QueryRowContext(t.Context(), `SELECT payload FROM jobs WHERE id = ?`, id).Scan(&payload); err != nil {
t.Fatal(err)
}
var p dinners.GroundJob
if err := json.Unmarshal([]byte(payload), &p); err != nil {
t.Fatal(err)
}
return p
}
// TestRenameBeforeJobStartsSkipsStaleJob: a Regular renamed before its
// first job starts is grounded only by the job its rename enqueued. The
// first job, though it would read the new name, does no work and writes
// nothing, so its failure cannot leave the dinner "Needs a recipe" while
// the rename's job finds a recipe.
func TestRenameBeforeJobStartsSkipsStaleJob(t *testing.T) {
t.Parallel()
e := newUnstartedEnv(t)
e.search.Results["Beef stew"] = []recipes.Candidate{{URL: stew}}
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
first := e.queuedJob(t, *c.GroundingJobID)
if first.DinnerID != c.ID || first.Revision != e.dinner(t, c.ID).GroundingRevision {
t.Fatalf("first job = %+v, want dinner %s at its revision", first, c.ID)
}
if _, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Name: ptr("Beef stew")}, member); err != nil {
t.Fatal(err)
}
renamed := e.dinner(t, c.ID)
if renamed.GroundingRevision != first.Revision+1 {
t.Fatalf("revision after rename = %d, want %d", renamed.GroundingRevision, first.Revision+1)
}
// The first job would fail if it searched.
e.search.Err = recipes.ErrSearchRateLimited
status, err := e.svc.RunGroundJob(t.Context(), first)
if err != nil || status != dinners.GroundingPending {
t.Fatalf("first job = %q, %v; want pending, no error", status, err)
}
if q := e.search.Queries(); len(q) != 0 {
t.Errorf("first job searched %q, want no work for a stale revision", q)
}
if row := e.dinner(t, c.ID); row != renamed {
t.Fatalf("first job wrote the dinner: %+v, was %+v", row, renamed)
}
e.search.Err = nil
e.model.QueueGenerate(recipestest.NormalizedReply("oven", 150))
status, err = e.svc.RunGroundJob(t.Context(), dinners.GroundJob{DinnerID: c.ID, Revision: renamed.GroundingRevision})
if err != nil || status != dinners.GroundingGrounded {
t.Fatalf("rename's job = %q, %v; want grounded", status, err)
}
row := e.dinner(t, c.ID)
rec, err := db.New(e.st.Reader).GetRecipe(t.Context(), row.RecipeID.String)
if err != nil {
t.Fatal(err)
}
if row.GroundingStatus != dinners.GroundingGrounded || rec.SourceUrl != stew {
t.Errorf("dinner = %+v recipe %s, want grounded in %s", row, rec.SourceUrl, stew)
}
// A stale job finishing last changes nothing either.
if status, err := e.svc.RunGroundJob(t.Context(), first); err != nil || status != dinners.GroundingGrounded {
t.Errorf("first job run last = %q, %v; want grounded reported, no error", status, err)
}
if after := e.dinner(t, c.ID); after != row {
t.Errorf("stale job changed the grounded dinner: %+v, was %+v", after, row)
}
}
// TestEditWhilePendingIsKept: a method or minutes a member saves while a
// Regular is grounding is kept when grounding succeeds; what the member
// did not set comes from the recipe.
func TestEditWhilePendingIsKept(t *testing.T) {
t.Parallel()
e := newEnv(t)
e.search.Gate = make(chan struct{})
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
e.waitQuery(t, "Chicken curry")
if _, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Method: ptr("grill"), MinutesTotal: ptr(15)}, member); err != nil {
t.Fatal(err)
}
close(e.search.Gate)
row := e.waitGrounding(t, c.ID)
if row.GroundingStatus != dinners.GroundingGrounded || !row.RecipeID.Valid || !row.PhotoID.Valid {
t.Fatalf("dinner = %+v, want grounded", row)
}
if row.Method.String != "grill" || row.MinutesTotal.Int64 != 15 || row.Difficulty.String != "easy" {
t.Errorf("dinner = %s, %d minutes, %s; want the member's grill and 15 minutes, the recipe's easy",
row.Method.String, row.MinutesTotal.Int64, row.Difficulty.String)
}
}
// TestArchiveWhilePendingStopsGrounding: a pending Regular archived before
// its job starts is not searched for, and one archived while its job runs
// is stopped there, with no page fetched or model called; neither ends up
// grounded.
func TestArchiveWhilePendingStopsGrounding(t *testing.T) {
t.Parallel()
t.Run("before the job starts", func(t *testing.T) {
t.Parallel()
e := newUnstartedEnv(t)
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
if _, err := e.svc.Archive(t.Context(), c.ID, member); err != nil {
t.Fatal(err)
}
e.start(t)
if job := e.waitJob(t, *c.GroundingJobID); job.Status != jobs.StatusSucceeded {
t.Fatalf("job = %+v, want succeeded with nothing to do", job)
}
if q := e.search.Queries(); len(q) != 0 {
t.Errorf("searched %q for an archived dinner", q)
}
if row := e.dinner(t, c.ID); row.Status != dinners.StatusArchived || row.GroundingStatus == dinners.GroundingGrounded || row.RecipeID.Valid {
t.Errorf("dinner = %+v, want archived and not grounded", row)
}
})
t.Run("while the job runs", func(t *testing.T) {
t.Parallel()
e := newEnv(t)
e.search.Gate = make(chan struct{})
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
e.waitQuery(t, "Chicken curry")
if _, err := e.svc.Archive(t.Context(), c.ID, member); err != nil {
t.Fatal(err)
}
// The job ends while its search is still held: archiving stopped it.
job := e.waitJob(t, *c.GroundingJobID)
close(e.search.Gate)
if job.Status != jobs.StatusSucceeded || string(job.Result) != `{"dinner_id":"`+c.ID+`","grounding_status":"pending"}` {
t.Fatalf("job = %+v (result %s), want succeeded leaving the dinner pending", job, job.Result)
}
if got := e.pages.Fetched(); len(got) != 0 {
t.Errorf("fetched %q for an archived dinner", got)
}
if got := len(e.model.Requests()); got != 0 {
t.Errorf("model called %d times for an archived dinner", got)
}
if row := e.dinner(t, c.ID); row.Status != dinners.StatusArchived || row.GroundingStatus == dinners.GroundingGrounded || row.RecipeID.Valid {
t.Errorf("dinner = %+v, want archived and not grounded", row)
}
})
}
// TestRenameWhilePendingEnqueueFailure: a pending Regular renamed when the
// job for its new name cannot be enqueued is failed, not left pending.
func TestRenameWhilePendingEnqueueFailure(t *testing.T) {
@@ -492,6 +718,59 @@ func TestFixViaCandidate(t *testing.T) {
}
}
// TestFixReplacesPendingLink: fixing a Regular added from a link that held
// no recipe drops that link, so it no longer blocks adding the link again,
// while the link of the recipe chosen does.
func TestFixReplacesPendingLink(t *testing.T) {
t.Parallel()
const lost = "https://gone.example/recipes/lamb-tagine"
tests := []struct {
name string
fix dinners.Fix
// linkTaken is the recipe link the fixed dinner now has, if any.
linkTaken string
}{
{"candidate", dinners.Fix{CandidateID: ptr(curryB)}, curryB},
{"manual", dinners.Fix{Manual: &dinners.ManualRecipe{Ingredients: []string{"1 lamb"}, Minutes: ptr(90), Method: ptr("oven")}}, ""},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
t.Parallel()
e := newEnv(t)
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{RecipeURL: ptr(lost)}, member)
if err != nil {
t.Fatal(err)
}
if job := e.waitJob(t, *c.GroundingJobID); job.Status != jobs.StatusFailed {
t.Fatalf("job = %+v, want failed", job)
}
// Renamed, so only the link could repeat it.
if _, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Name: ptr("Weeknight dinner")}, member); err != nil {
t.Fatal(err)
}
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 25))
d, err := e.svc.Ground(t.Context(), c.ID, tt.fix, member)
if err != nil {
t.Fatal(err)
}
if d.RecipeURL != nil {
t.Errorf("recipe_url = %q after the fix, want the unusable link dropped", *d.RecipeURL)
}
if _, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{RecipeURL: ptr(lost)}, member); err != nil {
t.Errorf("adding the dropped link again: %v, want added", err)
}
if tt.linkTaken != "" {
_, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{RecipeURL: ptr(tt.linkTaken)}, member)
var dup *dinners.DuplicateError
if !errors.As(err, &dup) || dup.Existing.ID != c.ID {
t.Errorf("adding the fixed dinner's recipe link: err = %v, want a duplicate of %s", err, c.ID)
}
}
})
}
}
func TestFixViaRecipeURL(t *testing.T) {
t.Parallel()
e := newEnv(t)
@@ -667,6 +946,126 @@ func TestRestartFailsInterruptedGrounding(t *testing.T) {
}
}
// TestRestartKeepsArchivedPendingDinner: a pending Regular archived while
// its job runs stays pending, its job succeeded; the boot cleanup leaves it
// as it is rather than failing it as interrupted.
func TestRestartKeepsArchivedPendingDinner(t *testing.T) {
t.Parallel()
e := newEnv(t)
e.search.Gate = make(chan struct{})
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
e.waitQuery(t, "Chicken curry")
if _, err := e.svc.Archive(t.Context(), c.ID, member); err != nil {
t.Fatal(err)
}
job := e.waitJob(t, *c.GroundingJobID)
close(e.search.Gate)
if job.Status != jobs.StatusSucceeded {
t.Fatalf("job = %+v, want succeeded", job)
}
// The database is now as the next boot finds it.
n, err := e.svc.FailInterruptedGroundings(t.Context())
if err != nil {
t.Fatal(err)
}
if n != 0 {
t.Errorf("failed %d dinners, want 0", n)
}
if row := e.dinner(t, c.ID); row.Status != dinners.StatusArchived ||
row.GroundingStatus != dinners.GroundingPending || row.GroundingError.Valid {
t.Errorf("dinner = %s, %s (%s), want archived and pending with no error",
row.Status, row.GroundingStatus, row.GroundingError.String)
}
}
// TestUpgradeKeepsLegacyQueuedGrounding: a pending Regular whose
// ground_dinner job was queued before migration 0010, so its payload has
// no revision, is not failed as interrupted at the first boot after the
// upgrade; its job still grounds it. The upgrade starts either from 0009
// or from a database that has already applied 0010.
func TestUpgradeKeepsLegacyQueuedGrounding(t *testing.T) {
t.Parallel()
for name, below := range map[string]string{
"from 0009": "0010",
"from 0010": "0011",
} {
t.Run(name, func(t *testing.T) {
t.Parallel()
upgradeWithLegacyQueuedGrounding(t, below)
})
}
}
// upgradeWithLegacyQueuedGrounding migrates a database to every migration
// named before below, adds a pending Regular with a queued ground_dinner
// job whose payload has no revision, then boots on the latest migrations.
func upgradeWithLegacyQueuedGrounding(t *testing.T, below string) {
ctx := t.Context()
dataDir := t.TempDir()
st := openStore(t, dataDir)
before := fstest.MapFS{}
files, err := fs.Glob(migrations.FS, "*.sql")
if err != nil {
t.Fatal(err)
}
for _, name := range files {
if name >= below {
continue
}
body, err := fs.ReadFile(migrations.FS, name)
if err != nil {
t.Fatal(err)
}
before[name] = &fstest.MapFile{Data: body}
}
if err := store.Migrate(ctx, st.Writer, before); err != nil {
t.Fatal(err)
}
now := store.Now().String()
if _, err := st.Writer.ExecContext(ctx, `INSERT INTO dinners
(id, name, normalized_name, status, origin, is_favorite, grounding_status, updated_by, created_at, updated_at)
VALUES ('dinner-1', 'Chicken curry', 'chicken curry', 'regular', 'regular', 0, 'pending', ?, ?, ?)`,
member, now, now); err != nil {
t.Fatal(err)
}
if _, err := st.Writer.ExecContext(ctx, `INSERT INTO jobs (id, kind, status, payload, created_at, updated_at)
VALUES ('job-1', ?, 'queued', '{"dinner_id":"dinner-1"}', ?, ?)`,
dinners.KindGroundDinner, now, now); err != nil {
t.Fatal(err)
}
// Boot: migrate, start the runner, then fail interrupted groundings. The
// search is held, so the job cannot finish before the cleanup runs.
e := newUnstartedEnvOn(t, st, dataDir)
e.search.Gate = make(chan struct{})
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
if p := e.queuedJob(t, "job-1"); p != (dinners.GroundJob{DinnerID: "dinner-1", Revision: 0}) {
t.Errorf("migrated payload = %+v, want dinner-1 at revision 0", p)
}
e.start(t)
n, err := e.svc.FailInterruptedGroundings(ctx)
if err != nil {
t.Fatal(err)
}
if n != 0 {
t.Errorf("failed %d dinners, want 0", n)
}
if row := e.dinner(t, "dinner-1"); row.GroundingStatus != dinners.GroundingPending {
t.Fatalf("dinner = %s (%s), want still pending", row.GroundingStatus, row.GroundingError.String)
}
close(e.search.Gate)
if job := e.waitJob(t, "job-1"); job.Status != jobs.StatusSucceeded {
t.Fatalf("job = %+v", job)
}
if row := e.dinner(t, "dinner-1"); row.GroundingStatus != dinners.GroundingGrounded {
t.Errorf("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) {