Fail the batch job when recording its end fails, and reconcile orphans

finishBatch now returns its error, and treats a batch that is no longer
running as one, so a batch job never ends succeeded with its batch still
running. A batch left pending or running with its job ended (process
stopped, or the final status write failed) is failed inside startBatch,
before the active batch is read, so it never blocks a manual or
replenish batch. This replaces the boot-only FailInterruptedBatches.

Document suggestions.name in the architecture schema summary.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

cmd/server/main.go

@@ -150,11 +150,6 @@ func run(cfg config.Config, logger *slog.Logger) error {
 	if _, err := dinnerSvc.FailInterruptedGroundings(ctx); err != nil {
 		return err
 	}
-	// Likewise a suggestion batch with no job left to run it, so it does
-	// not block every later batch.
-	if _, err := suggestSvc.FailInterruptedBatches(ctx); err != nil {
-		return err
-	}
 	srv := &http.Server{
 		Addr:              cfg.HTTPAddr,
 		Handler:           router,

docs/ARCHITECTURE.md

@@ -98,7 +98,7 @@ Columns ending in `_ct` hold AES-256-GCM ciphertext (via `secretbox`).
 | `photos` | source_url, local_path, width, height, sha256, fetched_at. |
 | `dinners` | The meal record and its lifecycle: name, normalized_name, status (`suggested` / `quest_log` / `tried` / `regular` / `archived`), origin, method, minutes_total, difficulty, cuisine, is_favorite, recipe_id, recipe_url (a pasted recipe link awaiting grounding), photo_id, grounding_status (`pending` / `grounded` / `failed` / `manual`), grounding_error (why grounding failed), grounding_revision (raised when a pending dinner is renamed, so only the latest `ground_dinner` job may write it), updated_by. |
 | `suggestion_batches` | trigger, status, ai_run_id. |
-| `suggestions` | dinner_id, batch_id, reason, cited_regular_ids (JSON), status, decided_at. |
+| `suggestions` | dinner_id, batch_id, name (the name the dish was suggested under, kept whatever its dinner is later renamed, so a Pass stays an immutable record every later batch dedupes against), reason, cited_regular_ids (JSON), status, decided_at. |
 | `week_plans` | week_start (unique), status (`draft` / `synced` / `changed`), ai_note, ai_run_id, calendar_synced_at. |
 | `plan_days` | plan_id, date (unique per plan), dinner_id, is_new, calendar_event_id. |
 | `rotation_reviews` | dinner_id + plan_date (unique), status, decision (`keep` / `drop` / `not_made`), decided_at, updated_by. |

internal/db/querier.go

@@ -52,9 +52,8 @@ type Querier interface {
 	// boot, after the job runner fails the jobs a previous process left running.
 	FailOrphanedPendingDinnerGroundings(ctx context.Context, arg FailOrphanedPendingDinnerGroundingsParams) (int64, error)
 	// Fails every batch still pending or running whose job is no longer queued
-	// or running: the process stopped before its handler recorded a result.
-	// Run at boot, after the job runner fails the jobs a previous process left
-	// running.
+	// or running: the process stopped, or recording the batch's end failed,
+	// before its handler recorded a result. Run before reading the active batch.
 	FailOrphanedSuggestionBatches(ctx context.Context, arg FailOrphanedSuggestionBatchesParams) (int64, error)
 	// Records why the background grounding of an unarchived dinner still
 	// pending at the grounding_revision it was grounded for failed; a dinner

internal/db/suggestions.sql.go

@@ -132,9 +132,8 @@ type FailOrphanedSuggestionBatchesParams struct {
 }
 
 // Fails every batch still pending or running whose job is no longer queued
-// or running: the process stopped before its handler recorded a result.
-// Run at boot, after the job runner fails the jobs a previous process left
-// running.
+// or running: the process stopped, or recording the batch's end failed,
+// before its handler recorded a result. Run before reading the active batch.
 func (q *Queries) FailOrphanedSuggestionBatches(ctx context.Context, arg FailOrphanedSuggestionBatchesParams) (int64, error) {
 	result, err := q.db.ExecContext(ctx, failOrphanedSuggestionBatches, arg.UpdatedAt, arg.Kind)
 	if err != nil {

internal/suggest/batch.go

@@ -50,7 +50,8 @@ func (r batchResult) PublicJSON() ([]byte, error) { return json.Marshal(r) }
 // batch, as does suggest_side_quests failing, with no Side quest made up
 // instead. Each round's grounded Side quests are stored before the next
 // round starts, so the deck fills as the batch goes. The batch ends done
-// or failed; Side quests stored before a failure stay in the deck.
+// or failed; Side quests stored before a failure stay in the deck. When
+// recording the batch's end fails, so does the job.
 func (s *Service) runBatch(ctx context.Context, p BatchJob, publish func(batchResult) error) (batchResult, error) {
 	n, err := db.New(s.st.Writer).StartSuggestionBatch(ctx, db.StartSuggestionBatchParams{
 		UpdatedAt: store.Now().String(),
@@ -65,12 +66,14 @@ func (s *Service) runBatch(ctx context.Context, p BatchJob, publish func(batchRe
 	res, err := s.fill(ctx, p.BatchID, publish)
 	write := context.WithoutCancel(ctx)
 	if err == nil {
-		s.finishBatch(write, p.BatchID, BatchDone)
+		if err := s.finishBatch(write, p.BatchID, BatchDone); err != nil {
+			return batchResult{}, err
+		}
 		return res, nil
 	}
-	s.finishBatch(write, p.BatchID, BatchFailed)
+	finishErr := s.finishBatch(write, p.BatchID, BatchFailed)
 	if ctx.Err() != nil {
-		return batchResult{}, ctx.Err()
+		return batchResult{}, errors.Join(ctx.Err(), finishErr)
 	}
 	var failure *jobs.Failure
 	if errors.As(err, &failure) {
@@ -78,7 +81,7 @@ func (s *Service) runBatch(ctx context.Context, p BatchJob, publish func(batchRe
 		// search provider's response.
 		s.logger.InfoContext(ctx, "suggestion batch failed", "batch_id", p.BatchID, "reason", failure.Code)
 	}
-	return batchResult{}, err
+	return batchResult{}, errors.Join(err, finishErr)
 }
 
 // fill does runBatch's work, returning a *jobs.Failure for a failure the

internal/suggest/suggest.go

@@ -172,10 +172,23 @@ func (s *Service) StartBatch(ctx context.Context, trigger string) (Batch, error)
 }
 
 // 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
+// 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)
+	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)
+	}
+	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})
@@ -204,34 +217,23 @@ func (s *Service) startBatch(ctx context.Context, tx *sql.Tx, trigger string) (B
 	return Batch{ID: row.ID, JobID: jobID, Trigger: row.Trigger, Status: row.Status, Created: true}, nil
 }
 
-// finishBatch records batch id as status, logging a failure to do so.
-func (s *Service) finishBatch(ctx context.Context, id, status string) {
-	if _, err := db.New(s.st.Writer).FinishSuggestionBatch(ctx, db.FinishSuggestionBatchParams{
+// finishBatch records running batch id as status. Its job's handler
+// returns any error, so the job never ends succeeded with the batch still
+// running; the runner keeps only a code, so the error is logged too.
+func (s *Service) finishBatch(ctx context.Context, id, status string) error {
+	n, err := db.New(s.st.Writer).FinishSuggestionBatch(ctx, db.FinishSuggestionBatchParams{
 		Status:    status,
 		UpdatedAt: store.Now().String(),
 		ID:        id,
-	}); err != nil {
-		s.logger.ErrorContext(ctx, "record suggestion batch status", "batch_id", id, "status", status, "error", err)
-	}
-}
-
-// FailInterruptedBatches fails every batch still pending or running whose
-// KindSuggestionBatch job is no longer queued or running, so a batch a
-// previous process left unfinished does not block every later one. Call it
-// once at boot, after the job runner's Start has failed the jobs that
-// process left running. It returns how many it failed.
-func (s *Service) FailInterruptedBatches(ctx context.Context) (int64, error) {
-	n, err := db.New(s.st.Writer).FailOrphanedSuggestionBatches(ctx, db.FailOrphanedSuggestionBatchesParams{
-		UpdatedAt: store.Now().String(),
-		Kind:      KindSuggestionBatch,
 	})
+	if err == nil && n != 1 {
+		err = errors.New("batch is not running")
+	}
 	if err != nil {
-		return 0, fmt.Errorf("suggest: fail interrupted batches: %w", err)
+		s.logger.ErrorContext(ctx, "record suggestion batch status", "batch_id", id, "status", status, "error", err)
+		return fmt.Errorf("suggest: record batch %s %s: %w", id, status, err)
 	}
-	if n > 0 {
-		s.logger.WarnContext(ctx, "failed suggestion batches interrupted by restart", "count", n)
-	}
-	return n, nil
+	return nil
 }
 
 // Suggestion is a Side quest suggestion as the API shows it, with its

internal/suggest/suggest_test.go

@@ -891,23 +891,63 @@ func TestBatchFailsWithNoRegulars(t *testing.T) {
 	}
 }
 
-func TestFailInterruptedBatches(t *testing.T) {
+func TestStartBatchFailsOrphanedBatch(t *testing.T) {
 	e := newEnv(t)
-	queued, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual) // its job is still queued
-	if err != nil {
-		t.Fatal(err)
-	}
 	now := store.Now().String()
 	orphan, err := db.New(e.st.Writer).CreateSuggestionBatch(t.Context(), db.CreateSuggestionBatchParams{ID: store.NewID(), Trigger: "manual", CreatedAt: now, UpdatedAt: now})
 	if err != nil {
 		t.Fatal(err)
 	}
-	n, err := e.svc.FailInterruptedBatches(t.Context())
+	b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if n != 1 || e.batchStatus(t, orphan.ID) != suggest.BatchFailed || e.batchStatus(t, queued.ID) != suggest.BatchPending {
-		t.Errorf("failed %d; orphan %s, queued %s; want only the orphan failed", n, e.batchStatus(t, orphan.ID), e.batchStatus(t, queued.ID))
+	if !b.Created || b.ID == orphan.ID {
+		t.Fatalf("batch = %+v, want a new one past the orphan", b)
+	}
+	if got := e.batchStatus(t, orphan.ID); got != suggest.BatchFailed {
+		t.Errorf("orphan status = %s, want failed", got)
+	}
+	// A batch whose job is still queued is not an orphan.
+	again, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if again.Created || again.ID != b.ID || e.batchStatus(t, b.ID) != suggest.BatchPending {
+		t.Errorf("second batch = %+v, status %s; want batch %s still pending", again, e.batchStatus(t, b.ID), b.ID)
+	}
+}
+
+func TestBatchJobFailsWhenRecordingItsEndFails(t *testing.T) {
+	e := newEnv(t)
+	e.start(t)
+	e.recipe(t, "Chili Verde")
+	e.reply(sideQuest("Chili Verde", "Your Beef Chili gone green.", "Beef Chili"))
+	if _, err := e.st.Writer.ExecContext(t.Context(), `CREATE TRIGGER refuse_done BEFORE UPDATE OF status ON suggestion_batches WHEN NEW.status = 'done' BEGIN SELECT RAISE(ABORT, 'done refused'); END`); err != nil {
+		t.Fatal(err)
+	}
+
+	b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
+	if err != nil {
+		t.Fatal(err)
+	}
+	job := e.waitJob(t, b.JobID)
+	if job.Status != jobs.StatusFailed {
+		t.Fatalf("job %s with its batch %s, want failed", job.Status, e.batchStatus(t, b.ID))
+	}
+
+	if _, err := e.st.Writer.ExecContext(t.Context(), `DROP TRIGGER refuse_done`); err != nil {
+		t.Fatal(err)
+	}
+	next, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !next.Created || next.ID == b.ID {
+		t.Errorf("next batch = %+v, want a new one", next)
+	}
+	if got := e.batchStatus(t, b.ID); got != suggest.BatchFailed {
+		t.Errorf("batch status = %s with its job failed, want failed", got)
 	}
 }
 

migrations/queries/suggestions.sql

@@ -35,9 +35,8 @@ WHERE id = ? AND status IN ('pending', 'running');
 
 -- name: FailOrphanedSuggestionBatches :execrows
 -- Fails every batch still pending or running whose job is no longer queued
--- or running: the process stopped before its handler recorded a result.
--- Run at boot, after the job runner fails the jobs a previous process left
--- running.
+-- or running: the process stopped, or recording the batch's end failed,
+-- before its handler recorded a result. Run before reading the active batch.
 UPDATE suggestion_batches
 SET status = 'failed', updated_at = sqlc.arg(updated_at)
 WHERE status IN ('pending', 'running')