diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 2e54f05..00e82e4 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -285,7 +285,9 @@ automatic** (D-021). D-017), `search_not_configured` or `search_failed`, and the batch is set `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 + 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 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 b3eb2a6..db57527 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -63,6 +63,11 @@ 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 + // codePattern is the form of every failure code: lower-case snake case. var codePattern = regexp.MustCompile(`^[a-z][a-z0-9_]{0,63}$`) @@ -132,6 +137,8 @@ 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 @@ -152,12 +159,13 @@ 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, - handlers: make(map[string]handler), + st: st, + logger: logger, + workers: workers, + wake: make(chan struct{}, workers), + claim: db.New(st.Writer).ClaimNextJob, + recordTimeout: recordTimeout, + handlers: make(map[string]handler), }, nil } @@ -485,13 +493,16 @@ 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; once it -// is, a failed attempt is not retried and the job stays running, so the -// next Start fails it as interrupted. +// 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. 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)) + attempt, cancel := context.WithTimeout(context.WithoutCancel(ctx), r.recordTimeout) + n, err := write(attempt) + cancel() if err == nil { if n != 1 { logger.ErrorContext(ctx, what, "error", "job is not running") diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index 1727d18..02e05a3 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -642,6 +642,64 @@ 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) + <-finish + return countResult{Done: p.Steps}, nil + }) + stop := start(t, r) + id := enqueue(t, k, countPayload{Steps: 3}) + <-started + + // Hold the only writer connection, so the job's final write waits for it. + conn, err := st.Writer.Conn(t.Context()) + if err != nil { + t.Fatal(err) + } + waits := st.Writer.Stats().WaitCount + close(finish) + deadline := time.Now().Add(5 * time.Second) + for st.Writer.Stats().WaitCount == waits { + if time.Now().After(deadline) { + t.Fatal("final write never waited for the writer") + } + time.Sleep(5 * time.Millisecond) + } + + stopped := make(chan struct{}) + go func() { + stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(5 * 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 job := waitFor(t, r, id, func(Job) bool { return true }); job.Status != StatusRunning { + t.Fatalf("after shutdown job = %+v, want left running", job) + } + second := newRunner(t, st, 1) + register(t, second, "count", func(context.Context, countPayload, func(countResult) error) (countResult, error) { + return countResult{}, nil + }) + start(t, second) + if job := waitFor(t, second, id, finished); job.Status != StatusFailed || job.Error == nil || *job.Error != CodeInterrupted { + t.Errorf("interrupted job = %+v, want failed with %q", job, CodeInterrupted) + } +} + func TestRunnerEnqueueByKindName(t *testing.T) { t.Parallel() st := migratedStore(t)