diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 36e46b1..2e54f05 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -283,10 +283,13 @@ automatic** (D-021). `grounding`, `done`). It fails with `no_regulars`, `ai_failed` (the skill or `normalize_recipe` failed; no Side quest is made up instead, D-017), `search_not_configured` or `search_failed`, and the batch is set - `failed`; Side quests stored before the failure stay in the deck. At - boot, after the job runner fails the jobs a previous process left - running, every batch still `pending` or `running` with no job queued or - running is set `failed`, so it does not block the next. + `failed`; Side quests stored before the failure stay in the deck. If + recording the batch's end fails, its job fails; the job runner retries + a job's final status write with backoff until it lands. Each time a + batch is started, before the active batch is selected, every batch + still `pending` or `running` with no job queued or running (the process + stopped, or recording its end failed) is set `failed`, so it does not + block the next. `GET /api/v1/suggestions/deck` returns `{suggestions: [{id, dinner_id, name, photo_url, difficulty, method, minutes_total, cuisine, reason, cited_regular_ids, status, dinner_status, decided_at, created_at}]}`: diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index 2d2c865..b3eb2a6 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -54,12 +54,13 @@ const ( CodeUnknownKind = "unknown_kind" ) -// Claim retry backoff. A worker whose claim fails with a database error -// retries after claimRetryMin, doubling up to claimRetryMax, so a job queued -// when the error clears is claimed without waiting for a wake token. +// Retry backoff. A worker whose claim or final job write fails with a +// database error retries after retryMin, doubling up to retryMax, so a job +// queued when the error clears is claimed without waiting for a wake token, +// and a job the worker has finished is never left running. const ( - claimRetryMin = 50 * time.Millisecond - claimRetryMax = 5 * time.Second + retryMin = 50 * time.Millisecond + retryMax = 5 * time.Second ) // codePattern is the form of every failure code: lower-case snake case. @@ -387,16 +388,16 @@ func (r *Runner) Get(ctx context.Context, id string) (Job, error) { // already queued is not stranded by a transient error. func (r *Runner) work(ctx context.Context) { defer r.wg.Done() - backoff := claimRetryMin + backoff := retryMin for ctx.Err() == nil { job, err := r.claim(ctx, sql.NullString{String: store.Now().String(), Valid: true}) switch { case err == nil: - backoff = claimRetryMin + backoff = retryMin r.run(ctx, job) continue case errors.Is(err, sql.ErrNoRows): - backoff = claimRetryMin + backoff = retryMin select { case <-r.wake: case <-ctx.Done(): @@ -412,7 +413,7 @@ func (r *Runner) work(ctx context.Context) { case <-ctx.Done(): timer.Stop() } - backoff = min(2*backoff, claimRetryMax) + backoff = min(2*backoff, retryMax) } } @@ -420,14 +421,13 @@ func (r *Runner) work(ctx context.Context) { // ctx has been cancelled, so a handler that finishes during shutdown is // recorded. func (r *Runner) run(ctx context.Context, job db.Job) { - write := context.WithoutCancel(ctx) logger := r.logger.With("job_id", job.ID, "kind", job.Kind) q := db.New(r.st.Writer) h, ok := r.handlers[job.Kind] // read-only once started if !ok { logger.ErrorContext(ctx, "job failed", "code", CodeUnknownKind, "reason", "no handler registered for the kind") - r.fail(write, logger, job.ID, CodeUnknownKind) + r.fail(ctx, logger, job.ID, CodeUnknownKind) return } @@ -449,7 +449,7 @@ func (r *Runner) run(ctx context.Context, job db.Job) { result, code, reason := h.run(ctx, job.Payload, publish) if code != "" { if ctx.Err() != nil { - logger.WarnContext(write, "job interrupted by shutdown; it fails at the next start") + logger.WarnContext(context.WithoutCancel(ctx), "job interrupted by shutdown; it fails at the next start") return } level := slog.LevelInfo @@ -457,35 +457,58 @@ func (r *Runner) run(ctx context.Context, job db.Job) { level = slog.LevelError } logger.Log(ctx, level, "job failed", "code", code, "reason", reason) - r.fail(write, logger, job.ID, code) + r.fail(ctx, logger, job.ID, code) return } - n, err := q.SucceedJob(write, db.SucceedJobParams{ - Result: sql.NullString{String: string(result), Valid: true}, - Now: sql.NullString{String: store.Now().String(), Valid: true}, - ID: job.ID, + r.record(ctx, logger, "record job success", func(write context.Context) (int64, error) { + return q.SucceedJob(write, db.SucceedJobParams{ + Result: sql.NullString{String: string(result), Valid: true}, + Now: sql.NullString{String: store.Now().String(), Valid: true}, + ID: job.ID, + }) }) - if err == nil && n != 1 { - err = errors.New("job is not running") - } - if err != nil { - logger.ErrorContext(write, "record job success", "error", err) - } } // fail records the job as failed with code. The record errors it logs come // from the database, never from a handler. func (r *Runner) fail(ctx context.Context, logger *slog.Logger, id, code string) { - n, err := db.New(r.st.Writer).FailJob(ctx, db.FailJobParams{ - Error: sql.NullString{String: code, Valid: true}, - Now: sql.NullString{String: store.Now().String(), Valid: true}, - ID: id, + r.record(ctx, logger, "record job failure", func(write context.Context) (int64, error) { + return db.New(r.st.Writer).FailJob(write, db.FailJobParams{ + Error: sql.NullString{String: code, Valid: true}, + Now: sql.NullString{String: store.Now().String(), Valid: true}, + ID: id, + }) }) - if err == nil && n != 1 { - err = errors.New("job is not running") - } - if err != nil { - logger.ErrorContext(ctx, "record job failure", "error", err) +} + +// record makes a job's final write, which must change the one running job. +// A database error is retried with backoff until the write lands, so a +// worker never moves on from a job it has finished while leaving it +// running. Each attempt is made even when ctx has been cancelled; once it +// is, a failed attempt is not retried and the job stays running, so the +// next Start fails it as interrupted. +func (r *Runner) record(ctx context.Context, logger *slog.Logger, what string, write func(context.Context) (int64, error)) { + backoff := retryMin + for { + n, err := write(context.WithoutCancel(ctx)) + if err == nil { + if n != 1 { + logger.ErrorContext(ctx, what, "error", "job is not running") + } + return + } + if ctx.Err() != nil { + logger.ErrorContext(context.WithoutCancel(ctx), what+" failed during shutdown; it fails at the next start", "error", err) + return + } + logger.ErrorContext(ctx, what+" failed; retrying", "error", err, "retry_in", backoff) + timer := time.NewTimer(backoff) + select { + case <-timer.C: + case <-ctx.Done(): + timer.Stop() + } + backoff = min(2*backoff, retryMax) } } diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index b1b6752..1727d18 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -609,6 +609,39 @@ func TestClaimErrorIsRetried(t *testing.T) { } } +func TestFinalWriteErrorIsRetried(t *testing.T) { + st := migratedStore(t) + logs := &logBuffer{} + r := newRunnerLogging(t, st, 1, logs) + k := register(t, r, "count", func(ctx context.Context, p countPayload, publish func(countResult) error) (countResult, error) { + return countResult{Done: p.Steps}, nil + }) + if _, err := st.Writer.ExecContext(t.Context(), `CREATE TRIGGER refuse_end BEFORE UPDATE OF status ON jobs WHEN NEW.status IN ('succeeded', 'failed') BEGIN SELECT RAISE(ABORT, 'end refused'); END`); err != nil { + t.Fatal(err) + } + start(t, r) + id := enqueue(t, k, countPayload{Steps: 3}) + + deadline := time.Now().Add(5 * time.Second) + for !strings.Contains(logs.String(), "record job success") { + if time.Now().After(deadline) { + t.Fatal("job success never recorded") + } + time.Sleep(5 * time.Millisecond) + } + if _, err := st.Writer.ExecContext(t.Context(), `DROP TRIGGER refuse_end`); err != nil { + t.Fatal(err) + } + + job := waitFor(t, r, id, finished) + if job.Status != StatusSucceeded || string(job.Result) != `{"done":3}` { + t.Fatalf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result) + } + if out := logs.String(); !strings.Contains(out, "record job success failed; retrying") { + t.Fatalf("final write error not logged as retried: %s", out) + } +} + func TestRunnerEnqueueByKindName(t *testing.T) { t.Parallel() st := migratedStore(t) diff --git a/internal/suggest/suggest_test.go b/internal/suggest/suggest_test.go index 43eef70..7f89c88 100644 --- a/internal/suggest/suggest_test.go +++ b/internal/suggest/suggest_test.go @@ -44,6 +44,9 @@ type env struct { regulars map[string]string grounder *recipes.Grounder + // logs collects the service's and job runner's log output. + logs *logBuffer + mu sync.Mutex inputs []skills.SideQuestsRequest // stub, when set, answers in place of suggest_side_quests. @@ -82,6 +85,7 @@ func newEnv(t *testing.T) *env { model: &llm.Fake{}, side: &llm.Fake{}, regulars: map[string]string{}, + logs: &logBuffer{}, } for _, name := range []string{"Beef Chili", "Chicken Fajitas"} { e.regulars[name] = e.dinner(t, name, "regular") @@ -108,7 +112,7 @@ func newEnv(t *testing.T) *env { } return skills.SuggestSideQuests(ctx, cfg, scope, in) } - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + logger := slog.New(slog.NewTextHandler(e.logs, nil)) e.runner, err = jobs.NewRunner(st, logger, 1) if err != nil { t.Fatal(err) @@ -120,6 +124,24 @@ func newEnv(t *testing.T) *env { return e } +// logBuffer collects log output written by concurrent goroutines. +type logBuffer struct { + mu sync.Mutex + buf strings.Builder +} + +func (b *logBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *logBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + // start starts the env's job runner, stopping it when the test ends. func (e *env) start(t *testing.T) { t.Helper() @@ -951,6 +973,62 @@ func TestBatchJobFailsWhenRecordingItsEndFails(t *testing.T) { } } +func TestBatchRecoversWhenRecordingItsEndAndItsJobsEndFail(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")) + for _, trigger := range []string{ + `CREATE TRIGGER refuse_done BEFORE UPDATE OF status ON suggestion_batches WHEN NEW.status = 'done' BEGIN SELECT RAISE(ABORT, 'done refused'); END`, + `CREATE TRIGGER refuse_job_end BEFORE UPDATE OF status ON jobs WHEN NEW.status IN ('succeeded', 'failed') BEGIN SELECT RAISE(ABORT, 'job end refused'); END`, + } { + if _, err := e.st.Writer.ExecContext(t.Context(), trigger); err != nil { + t.Fatal(err) + } + } + + b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(10 * time.Second) + for !strings.Contains(e.logs.String(), "record job failure") { + if time.Now().After(deadline) { + t.Fatalf("job end never recorded; batch %s", e.batchStatus(t, b.ID)) + } + time.Sleep(5 * time.Millisecond) + } + // While neither end is recorded the batch is still the active one. + again, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual) + if err != nil { + t.Fatal(err) + } + if again.Created || again.ID != b.ID { + t.Fatalf("batch = %+v while batch %s is running, want it", again, b.ID) + } + + // The database recovers; the service is not restarted. + for _, trigger := range []string{"refuse_done", "refuse_job_end"} { + if _, err := e.st.Writer.ExecContext(t.Context(), "DROP TRIGGER "+trigger); err != nil { + t.Fatal(err) + } + } + job := e.waitJob(t, b.JobID) + if job.Status != jobs.StatusFailed { + t.Fatalf("job %s, want failed", job.Status) + } + 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) + } +} + func TestStartBatchRefusesUnknownTrigger(t *testing.T) { e := newEnv(t) if _, err := e.svc.StartBatch(t.Context(), "whenever"); err == nil {