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 <noreply@anthropic.com>

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
+// activeBatch returns the batch pending or running in q's transaction, or
-// it are committed together, so a batch is never seen without its job. A
+// nil when there is none. A batch still pending or running whose job is no
-// batch still pending or running whose job is no longer queued or running
+// longer queued or running is failed first, so a batch whose job ended
-// is failed first, so a batch whose job ended without recording its end
+// without recording its end (the process stopped, or the write failed) is
-// (the process stopped, or the write failed) never blocks a later one. The
+// never reported as active and never blocks a later one.
-// caller wakes the job runner once tx has committed and Created is true.
+func (s *Service) activeBatch(ctx context.Context, q *db.Queries) (*Batch, error) {
-func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (Batch, error) {
-	q := db.New(tx)
 	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 {
+	if errors.Is(err, sql.ErrNoRows) {
-		jobID, err := q.GetSuggestionBatchJobID(ctx, db.GetSuggestionBatchJobIDParams{Kind: KindSuggestionBatch, BatchID: cur.ID})
+		return nil, nil
-		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) {
+	if err != nil {
-		return Batch{}, fmt.Errorf("suggest: read active batch: %w", err)
+		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
+// Deck reads the deck and its active batch in one transaction, so both come
-// both come from the same snapshot.
+// from the same snapshot. It is a write transaction because reading the
-func (s *Service) Deck(ctx context.Context) (deck Deck, err error) {
+// active batch first fails any batch whose job has ended (see activeBatch).
-	tx, err := s.st.Reader.BeginTx(ctx, nil)
+func (s *Service) Deck(ctx context.Context) (Deck, error) {
-	if err != nil {
+	var deck Deck
-		return Deck{}, fmt.Errorf("suggest: begin deck read: %w", err)
+	err := s.st.WithTx(ctx, func(tx *sql.Tx) error {
-	}
+		q := db.New(tx)
-	defer func() {
+		rows, err := q.ListDeck(ctx)
-		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)
 		if err != nil {
-			return Deck{}, err
+			return fmt.Errorf("suggest: list deck: %w", err)
 		}
-	}
+		deck.Suggestions = make([]Suggestion, len(rows))
-	cur, err := q.GetActiveSuggestionBatch(ctx)
+		for i, row := range rows {
-	if errors.Is(err, sql.ErrNoRows) {
+			deck.Suggestions[i], err = fromRow(row.Suggestion, row.Dinner)
-		return deck, nil
+			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
 }
 

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")