From 71060515bd724e32e4dc2029ed687ed886f7953a Mon Sep 17 00:00:00 2001 From: dev_agent Date: Wed, 7 Oct 2026 05:13:48 -0400 Subject: [PATCH] Fail orphaned suggestion batches before reporting the deck's active batch Deck returned a pending or running batch's job id without checking that the job was still queued or running, so after a restart failed the job the page polled a dead job as the active batch. Deck and startBatch now share activeBatch, which runs FailOrphanedSuggestionBatches before reading the active batch; Deck therefore uses a write transaction. Co-Authored-By: Claude Opus 5.5 --- internal/suggest/suggest.go | 97 ++++++++++++++++---------------- internal/suggest/suggest_test.go | 37 ++++++++++++ 2 files changed, 87 insertions(+), 47 deletions(-) diff --git a/internal/suggest/suggest.go b/internal/suggest/suggest.go index 81f398d..33d5491 100644 --- a/internal/suggest/suggest.go +++ b/internal/suggest/suggest.go @@ -171,34 +171,47 @@ func (s *Service) StartBatch(ctx context.Context, trigger string) (Batch, error) return b, nil } -// startBatch does StartBatch's work in tx: the batch and the job that runs -// it are committed together, so a batch is never seen without its job. A -// batch still pending or running whose job is no longer queued or running -// is failed first, so a batch whose job ended without recording its end -// (the process stopped, or the write failed) never blocks a later one. The -// caller wakes the job runner once tx has committed and Created is true. -func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (Batch, error) { - q := db.New(tx) +// activeBatch returns the batch pending or running in q's transaction, or +// nil when there is none. A batch still pending or running whose job is no +// longer queued or running is failed first, so a batch whose job ended +// without recording its end (the process stopped, or the write failed) is +// never reported as active and never blocks a later one. +func (s *Service) activeBatch(ctx context.Context, q *db.Queries) (*Batch, error) { n, err := q.FailOrphanedSuggestionBatches(ctx, db.FailOrphanedSuggestionBatchesParams{ UpdatedAt: store.Now().String(), Kind: KindSuggestionBatch, }) if err != nil { - return Batch{}, fmt.Errorf("suggest: fail orphaned batches: %w", err) + return nil, fmt.Errorf("suggest: fail orphaned batches: %w", err) } if n > 0 { s.logger.WarnContext(ctx, "failed suggestion batches whose job ended", "count", n) } cur, err := q.GetActiveSuggestionBatch(ctx) - if err == nil { - jobID, err := q.GetSuggestionBatchJobID(ctx, db.GetSuggestionBatchJobIDParams{Kind: KindSuggestionBatch, BatchID: cur.ID}) - if err != nil { - return Batch{}, fmt.Errorf("suggest: read job of batch %s: %w", cur.ID, err) - } - return Batch{ID: cur.ID, JobID: jobID, Trigger: cur.Trigger, Status: cur.Status}, nil + if errors.Is(err, sql.ErrNoRows) { + return nil, nil } - if !errors.Is(err, sql.ErrNoRows) { - return Batch{}, fmt.Errorf("suggest: read active batch: %w", err) + if err != nil { + return nil, fmt.Errorf("suggest: read active batch: %w", err) + } + jobID, err := q.GetSuggestionBatchJobID(ctx, db.GetSuggestionBatchJobIDParams{Kind: KindSuggestionBatch, BatchID: cur.ID}) + if err != nil { + return nil, fmt.Errorf("suggest: read job of batch %s: %w", cur.ID, err) + } + return &Batch{ID: cur.ID, JobID: jobID, Trigger: cur.Trigger, Status: cur.Status}, nil +} + +// startBatch does StartBatch's work in tx: the batch and the job that runs +// it are committed together, so a batch is never seen without its job. The +// caller wakes the job runner once tx has committed and Created is true. +func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (Batch, error) { + q := db.New(tx) + cur, err := s.activeBatch(ctx, q) + if err != nil { + return Batch{}, err + } + if cur != nil { + return *cur, nil } now := store.Now().String() row, err := q.CreateSuggestionBatch(ctx, db.CreateSuggestionBatchParams{ @@ -269,40 +282,30 @@ type Deck struct { ActiveBatch *Batch `json:"active_batch"` } -// Deck reads the deck and its active batch in one read transaction, so -// both come from the same snapshot. -func (s *Service) Deck(ctx context.Context) (deck Deck, err error) { - tx, err := s.st.Reader.BeginTx(ctx, nil) - if err != nil { - return Deck{}, fmt.Errorf("suggest: begin deck read: %w", err) - } - defer func() { - err = errors.Join(err, tx.Rollback()) - }() - q := db.New(tx) - rows, err := q.ListDeck(ctx) - if err != nil { - return Deck{}, fmt.Errorf("suggest: list deck: %w", err) - } - deck.Suggestions = make([]Suggestion, len(rows)) - for i, row := range rows { - deck.Suggestions[i], err = fromRow(row.Suggestion, row.Dinner) +// Deck reads the deck and its active batch in one transaction, so both come +// from the same snapshot. It is a write transaction because reading the +// active batch first fails any batch whose job has ended (see activeBatch). +func (s *Service) Deck(ctx context.Context) (Deck, error) { + var deck Deck + err := s.st.WithTx(ctx, func(tx *sql.Tx) error { + q := db.New(tx) + rows, err := q.ListDeck(ctx) if err != nil { - return Deck{}, err + return fmt.Errorf("suggest: list deck: %w", err) } - } - cur, err := q.GetActiveSuggestionBatch(ctx) - if errors.Is(err, sql.ErrNoRows) { - return deck, nil - } + deck.Suggestions = make([]Suggestion, len(rows)) + for i, row := range rows { + deck.Suggestions[i], err = fromRow(row.Suggestion, row.Dinner) + if err != nil { + return err + } + } + deck.ActiveBatch, err = s.activeBatch(ctx, q) + return err + }) if err != nil { - return Deck{}, fmt.Errorf("suggest: read active batch: %w", err) + return Deck{}, err } - jobID, err := q.GetSuggestionBatchJobID(ctx, db.GetSuggestionBatchJobIDParams{Kind: KindSuggestionBatch, BatchID: cur.ID}) - if err != nil { - return Deck{}, fmt.Errorf("suggest: read job of batch %s: %w", cur.ID, err) - } - deck.ActiveBatch = &Batch{ID: cur.ID, JobID: jobID, Trigger: cur.Trigger, Status: cur.Status} return deck, nil } diff --git a/internal/suggest/suggest_test.go b/internal/suggest/suggest_test.go index ee850a3..683f046 100644 --- a/internal/suggest/suggest_test.go +++ b/internal/suggest/suggest_test.go @@ -2,6 +2,7 @@ package suggest_test import ( "context" + "database/sql" "encoding/json" "errors" "fmt" @@ -479,6 +480,42 @@ func TestDeckReportsTheActiveBatch(t *testing.T) { } } +func TestDeckFailsBatchInterruptedByRestart(t *testing.T) { + e := newEnv(t) + e.seedDeck(t, "Pho") + b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual) + if err != nil { + t.Fatal(err) + } + // A previous process claimed the job and started the batch, then + // stopped before recording its end. + now := store.Now().String() + if _, err := db.New(e.st.Writer).ClaimNextJob(t.Context(), sql.NullString{String: now, Valid: true}); err != nil { + t.Fatal(err) + } + if n, err := db.New(e.st.Writer).StartSuggestionBatch(t.Context(), db.StartSuggestionBatchParams{UpdatedAt: now, ID: b.ID}); err != nil || n != 1 { + t.Fatalf("start batch: %d rows, %v", n, err) + } + e.start(t) // fails the running job as interrupted + if job := e.waitJob(t, b.JobID); job.Status != jobs.StatusFailed { + t.Fatalf("job status = %s, want failed", job.Status) + } + + deck, err := e.svc.Deck(t.Context()) + if err != nil { + t.Fatal(err) + } + if deck.ActiveBatch != nil { + t.Errorf("active batch = %+v after its job failed, want nil", deck.ActiveBatch) + } + if len(deck.Suggestions) != 1 { + t.Errorf("deck has %d suggestions, want 1", len(deck.Suggestions)) + } + if got := e.batchStatus(t, b.ID); got != suggest.BatchFailed { + t.Errorf("interrupted batch status = %s, want failed", got) + } +} + func TestDeckLeavesOutDinnersNoLongerSuggested(t *testing.T) { e := newEnv(t) ids := e.seedDeck(t, "Pho", "Ramen")