diff --git a/cmd/server/main.go b/cmd/server/main.go index 3eed6b6..ca0834e 100644 --- a/cmd/server/main.go +++ b/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, diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index e9605c3..36e46b1 100644 --- a/docs/ARCHITECTURE.md +++ b/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. | diff --git a/internal/db/querier.go b/internal/db/querier.go index 64ecb77..b6094eb 100644 --- a/internal/db/querier.go +++ b/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 diff --git a/internal/db/suggestions.sql.go b/internal/db/suggestions.sql.go index 98fc20d..4106a2a 100644 --- a/internal/db/suggestions.sql.go +++ b/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 { diff --git a/internal/suggest/batch.go b/internal/suggest/batch.go index d3117c8..0492440 100644 --- a/internal/suggest/batch.go +++ b/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 diff --git a/internal/suggest/suggest.go b/internal/suggest/suggest.go index 9ffda32..0b00713 100644 --- a/internal/suggest/suggest.go +++ b/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 diff --git a/internal/suggest/suggest_test.go b/internal/suggest/suggest_test.go index 313920d..43eef70 100644 --- a/internal/suggest/suggest_test.go +++ b/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) } } diff --git a/migrations/queries/suggestions.sql b/migrations/queries/suggestions.sql index c9392e9..3a42470 100644 --- a/migrations/queries/suggestions.sql +++ b/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')