diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 00e82e4..3227230 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -287,7 +287,10 @@ automatic** (D-021). 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 attempt bounded at 5 seconds; at shutdown an attempt that fails or times out is - not retried, and the next start fails the job as interrupted. Each time a + not retried, and the next start fails the job as interrupted. Recording + the batch's end is bounded the same way (one attempt, 5 seconds, made + even during shutdown), so a busy writer cannot hold a stopping worker; + a batch whose end timed out is left `running` with its job. 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 diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index db57527..12cb947 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -63,10 +63,20 @@ const ( retryMax = 5 * time.Second ) -// recordTimeout bounds each attempt at a job's final write, so a writer -// connection that stays busy cannot hold a stopping worker, and with it -// shutdown, open. -const recordTimeout = 5 * time.Second +// FinalWriteTimeout bounds each attempt at a job's final write, and at a +// handler's own last write, so a writer connection that stays busy cannot +// hold a stopping worker, and with it shutdown, open. +const FinalWriteTimeout = 5 * time.Second + +// FinalWriteContext returns the context for one attempt at a final write +// of a job: it is not cancelled with ctx, so a job that finishes during +// shutdown is still recorded, but it ends after FinalWriteTimeout. A +// handler makes its own last write, the one recording what the job did, +// with it too, and returns the error when the attempt times out, so its +// worker can stop; the next Start fails the job as interrupted. +func FinalWriteContext(ctx context.Context) (context.Context, context.CancelFunc) { + return context.WithTimeout(context.WithoutCancel(ctx), FinalWriteTimeout) +} // codePattern is the form of every failure code: lower-case snake case. var codePattern = regexp.MustCompile(`^[a-z][a-z0-9_]{0,63}$`) @@ -137,8 +147,6 @@ type Runner struct { wake chan struct{} // claim claims the oldest queued job, marking it running at now. claim func(ctx context.Context, now sql.NullString) (db.Job, error) - // recordTimeout bounds each attempt at a job's final write. - recordTimeout time.Duration mu sync.Mutex handlers map[string]handler @@ -159,13 +167,12 @@ func NewRunner(st *store.Store, logger *slog.Logger, workers int) (*Runner, erro return nil, fmt.Errorf("jobs: %d workers, want at least 1", workers) } return &Runner{ - st: st, - logger: logger, - workers: workers, - wake: make(chan struct{}, workers), - claim: db.New(st.Writer).ClaimNextJob, - recordTimeout: recordTimeout, - handlers: make(map[string]handler), + st: st, + logger: logger, + workers: workers, + wake: make(chan struct{}, workers), + claim: db.New(st.Writer).ClaimNextJob, + handlers: make(map[string]handler), }, nil } @@ -493,14 +500,14 @@ func (r *Runner) fail(ctx context.Context, logger *slog.Logger, id, code string) // 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, and is -// bounded by r.recordTimeout; once ctx is cancelled, a failed or timed-out -// attempt is not retried and the job stays running, so the next Start -// fails it as interrupted. +// running. Each attempt is made with FinalWriteContext, so even when ctx +// has been cancelled, and is bounded by FinalWriteTimeout; once ctx is +// cancelled, a failed or timed-out 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 { - attempt, cancel := context.WithTimeout(context.WithoutCancel(ctx), r.recordTimeout) + attempt, cancel := FinalWriteContext(ctx) n, err := write(attempt) cancel() if err == nil { diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index 02e05a3..ece6b13 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -645,7 +645,6 @@ func TestFinalWriteErrorIsRetried(t *testing.T) { func TestBlockedFinalWriteDoesNotHoldShutdown(t *testing.T) { st := migratedStore(t) r := newRunner(t, st, 1) - r.recordTimeout = 100 * time.Millisecond started, finish := make(chan struct{}), make(chan struct{}) k := register(t, r, "count", func(ctx context.Context, p countPayload, publish func(countResult) error) (countResult, error) { close(started) @@ -676,9 +675,11 @@ func TestBlockedFinalWriteDoesNotHoldShutdown(t *testing.T) { stop() close(stopped) }() + // The final write's attempt ends after 5 seconds; well past that, the + // worker is waiting for the writer with no bound. select { case <-stopped: - case <-time.After(5 * time.Second): + case <-time.After(15 * time.Second): conn.Close() <-stopped t.Fatal("worker waited for a held writer through shutdown") diff --git a/internal/suggest/batch.go b/internal/suggest/batch.go index 0492440..00aeccb 100644 --- a/internal/suggest/batch.go +++ b/internal/suggest/batch.go @@ -51,7 +51,9 @@ func (r batchResult) PublicJSON() ([]byte, error) { return json.Marshal(r) } // 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. When -// recording the batch's end fails, so does the job. +// recording the batch's end fails, so does the job; when it times out at +// shutdown, the job is left running and the batch with it, for the next +// start to fail. 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(), @@ -64,14 +66,13 @@ func (s *Service) runBatch(ctx context.Context, p BatchJob, publish func(batchRe return batchResult{}, fmt.Errorf("suggest: batch %s is not pending", p.BatchID) } res, err := s.fill(ctx, p.BatchID, publish) - write := context.WithoutCancel(ctx) if err == nil { - if err := s.finishBatch(write, p.BatchID, BatchDone); err != nil { + if err := s.finishBatch(ctx, p.BatchID, BatchDone); err != nil { return batchResult{}, err } return res, nil } - finishErr := s.finishBatch(write, p.BatchID, BatchFailed) + finishErr := s.finishBatch(ctx, p.BatchID, BatchFailed) if ctx.Err() != nil { return batchResult{}, errors.Join(ctx.Err(), finishErr) } diff --git a/internal/suggest/suggest.go b/internal/suggest/suggest.go index 0b00713..efed277 100644 --- a/internal/suggest/suggest.go +++ b/internal/suggest/suggest.go @@ -217,11 +217,16 @@ 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 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. +// finishBatch records running batch id as status, with +// jobs.FinalWriteContext: the write is made even when ctx has been +// cancelled, but a writer that stays busy cannot hold its worker past +// jobs.FinalWriteTimeout. 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{ + write, cancel := jobs.FinalWriteContext(ctx) + defer cancel() + n, err := db.New(s.st.Writer).FinishSuggestionBatch(write, db.FinishSuggestionBatchParams{ Status: status, UpdatedAt: store.Now().String(), ID: id, diff --git a/internal/suggest/suggest_test.go b/internal/suggest/suggest_test.go index 7f89c88..157f1a5 100644 --- a/internal/suggest/suggest_test.go +++ b/internal/suggest/suggest_test.go @@ -1029,6 +1029,113 @@ func TestBatchRecoversWhenRecordingItsEndAndItsJobsEndFail(t *testing.T) { } } +func TestBlockedBatchEndDoesNotHoldShutdown(t *testing.T) { + e := newEnv(t) + ctx, cancel := context.WithCancel(context.Background()) + if err := e.runner.Start(ctx); err != nil { + cancel() + t.Fatal(err) + } + stopped := make(chan struct{}) + t.Cleanup(func() { + cancel() + <-stopped + }) + e.search.Gate = make(chan struct{}) + e.search.Err = errors.New("search down") + e.reply(sideQuest("Chili Verde", "Your Beef Chili gone green.", "Beef Chili")) + + b, err := e.svc.StartBatch(t.Context(), suggest.TriggerManual) + if err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(10 * time.Second) + for len(e.search.Queries()) == 0 { + if time.Now().After(deadline) { + t.Fatal("batch never searched") + } + time.Sleep(5 * time.Millisecond) + } + + // Hold the only writer connection, so recording the batch's end + // waits for it once the search fails. + conn, err := e.st.Writer.Conn(t.Context()) + if err != nil { + t.Fatal(err) + } + waits := e.st.Writer.Stats().WaitCount + close(e.search.Gate) + for e.st.Writer.Stats().WaitCount == waits { + if time.Now().After(deadline) { + conn.Close() + t.Fatal("batch end never waited for the writer") + } + time.Sleep(5 * time.Millisecond) + } + + go func() { + e.runner.Wait() + close(stopped) + }() + cancel() + // Each final write's attempt ends after jobs.FinalWriteTimeout; well + // past that, the worker is waiting for the writer with no bound. + select { + case <-stopped: + case <-time.After(jobs.FinalWriteTimeout + 10*time.Second): + conn.Close() + <-stopped + t.Fatal("worker waited for a held writer through shutdown") + } + if err := conn.Close(); err != nil { + t.Fatal(err) + } + if got := e.batchStatus(t, b.ID); got != suggest.BatchRunning { + t.Fatalf("after shutdown batch status = %s, want left running", got) + } + + // The next process fails the job as interrupted, and the batch no + // longer blocks the next one. + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + next, err := jobs.NewRunner(e.st, logger, 1) + if err != nil { + t.Fatal(err) + } + refuse := func(context.Context, string, skills.SideQuestsRequest) (skills.SideQuests, error) { + return skills.SideQuests{}, errors.New("not wanted") + } + svc, err := suggest.NewService(e.st, e.grounder, refuse, next, logger) + if err != nil { + t.Fatal(err) + } + nextCtx, nextCancel := context.WithCancel(context.Background()) + if err := next.Start(nextCtx); err != nil { + nextCancel() + t.Fatal(err) + } + t.Cleanup(func() { + nextCancel() + next.Wait() + }) + job, err := next.Get(t.Context(), b.JobID) + if err != nil { + t.Fatal(err) + } + if job.Status != jobs.StatusFailed || job.Error == nil || *job.Error != jobs.CodeInterrupted { + t.Fatalf("interrupted job = %+v, want failed with %q", job, jobs.CodeInterrupted) + } + again, err := svc.StartBatch(t.Context(), suggest.TriggerManual) + if err != nil { + t.Fatal(err) + } + if !again.Created || again.ID == b.ID { + t.Errorf("next batch = %+v, want a new one", again) + } + if got := e.batchStatus(t, b.ID); got != suggest.BatchFailed { + t.Errorf("interrupted batch status = %s, want failed", got) + } +} + func TestStartBatchRefusesUnknownTrigger(t *testing.T) { e := newEnv(t) if _, err := e.svc.StartBatch(t.Context(), "whenever"); err == nil {