Retry job claims after a database error instead of waiting for a wake

A claim that failed with anything but sql.ErrNoRows parked the worker
on the wake channel, so a job queued across a restart (which has no
wake token) stayed queued until another enqueue or restart. The worker
now retries such a claim with backoff from 50ms doubling to 5s, reset
on any successful or empty claim, and stops promptly on shutdown. The
claim error is logged by category only.

The claim query is a Runner field so a test can inject one failing
claim; TestClaimErrorIsRetried drops the wake token and asserts the
queued job still runs.

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

cmd/server/jobs_test.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"io"
+	"log/slog"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+)
+
+// TestNewJobRunnerRunsNoop proves the runner run builds can execute a job:
+// a noop job enqueued before Start, as at boot, succeeds with its result.
+func TestNewJobRunnerRunsNoop(t *testing.T) {
+	ctx := context.Background()
+	db, err := openStore(ctx, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := db.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	runner, kinds, err := newJobRunner(db, slog.New(slog.NewTextHandler(io.Discard, nil)))
+	if err != nil {
+		t.Fatal(err)
+	}
+	id, err := kinds.builtins.Noop.Enqueue(ctx, jobs.NoopPayload{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	jobsCtx, stopJobs := context.WithCancel(ctx)
+	if err := runner.Start(jobsCtx); err != nil {
+		stopJobs()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		stopJobs()
+		runner.Wait()
+	})
+
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		job, err := runner.Get(ctx, id)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if job.Status == jobs.StatusSucceeded || job.Status == jobs.StatusFailed {
+			if job.Kind != jobs.KindNoop || job.Status != jobs.StatusSucceeded || string(job.Result) != `{"ok":true}` || job.Error != nil {
+				t.Fatalf("noop job = %+v (result %s), want succeeded with {\"ok\":true}", job, job.Result)
+			}
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("noop job never finished; last %+v", job)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}

cmd/server/main.go

@@ -18,6 +18,7 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/forwarded"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/frontend"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/secretbox"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/server"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
@@ -27,6 +28,9 @@ import (
 // version is set at build time with -ldflags "-X main.version=<version>".
 var version = "dev"
 
+// jobWorkers is the number of background jobs run at once.
+const jobWorkers = 2
+
 func main() {
 	cfg, err := config.Load()
 	if err != nil {
@@ -63,6 +67,10 @@ func run(cfg config.Config, logger *slog.Logger) error {
 		return err
 	}
 	settings := household.NewSettingsService(db)
+	runner, _, err := newJobRunner(db, logger)
+	if err != nil {
+		return err
+	}
 
 	sessions, err := auth.NewSessionManager(db, cfg.SessionIdle, cfg.SessionMax)
 	if err != nil {
@@ -80,10 +88,22 @@ func run(cfg config.Config, logger *slog.Logger) error {
 		return err
 	}
 
-	router, err := server.NewRouter(logger, version, cfg.TrustedProxy, authFlow, sessions, settings, frontend.Assets())
+	router, err := server.NewRouter(logger, version, cfg.TrustedProxy, authFlow, sessions, settings, runner, frontend.Assets())
 	if err != nil {
 		return err
 	}
+
+	// Start fails the jobs a previous process left running before any
+	// worker runs. The workers stop, and are waited for, before the
+	// database closes.
+	jobsCtx, stopJobs := context.WithCancel(ctx)
+	defer func() {
+		stopJobs()
+		runner.Wait()
+	}()
+	if err := runner.Start(jobsCtx); err != nil {
+		return err
+	}
 	srv := &http.Server{
 		Addr:              cfg.HTTPAddr,
 		Handler:           router,
@@ -165,6 +185,35 @@ func seedDefaults(ctx context.Context, db *store.Store, defaultTZ string) error
 	return household.SeedNotificationPrefs(ctx, db)
 }
 
+// jobKinds enqueues every job kind the server runs.
+type jobKinds struct {
+	builtins *jobs.Builtins
+}
+
+// newJobRunner returns the job runner with every job kind registered, ready
+// to Start.
+func newJobRunner(db *store.Store, logger *slog.Logger) (*jobs.Runner, *jobKinds, error) {
+	runner, err := jobs.NewRunner(db, logger, jobWorkers)
+	if err != nil {
+		return nil, nil, err
+	}
+	kinds, err := registerJobKinds(runner)
+	if err != nil {
+		return nil, nil, err
+	}
+	return runner, kinds, nil
+}
+
+// registerJobKinds registers every job kind's handler on runner. Each
+// feature's job kinds are registered here.
+func registerJobKinds(runner *jobs.Runner) (*jobKinds, error) {
+	builtins, err := jobs.RegisterBuiltins(runner)
+	if err != nil {
+		return nil, err
+	}
+	return &jobKinds{builtins: builtins}, nil
+}
+
 // logListening logs the startup line. Only the public URL's origin is logged
 // so no credential-bearing URL component can reach the log.
 func logListening(logger *slog.Logger, cfg config.Config) {

internal/db/jobs.sql.go

@@ -0,0 +1,166 @@
+// Code generated by sqlc. DO NOT EDIT.
+// versions:
+//   sqlc v1.31.1
+// source: jobs.sql
+
+package db
+
+import (
+	"context"
+	"database/sql"
+)
+
+const claimNextJob = `-- name: ClaimNextJob :one
+UPDATE jobs
+SET status = 'running', started_at = ?1, updated_at = ?1
+WHERE id = (SELECT id FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 1)
+RETURNING id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at
+`
+
+// Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
+// when no job is queued.
+func (q *Queries) ClaimNextJob(ctx context.Context, now sql.NullString) (Job, error) {
+	row := q.db.QueryRowContext(ctx, claimNextJob, now)
+	var i Job
+	err := row.Scan(
+		&i.ID,
+		&i.Kind,
+		&i.Status,
+		&i.Payload,
+		&i.Result,
+		&i.Error,
+		&i.StartedAt,
+		&i.FinishedAt,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
+const createJob = `-- name: CreateJob :exec
+INSERT INTO jobs (id, kind, status, payload, created_at, updated_at)
+VALUES (?, ?, 'queued', ?, ?, ?)
+`
+
+type CreateJobParams struct {
+	ID        string
+	Kind      string
+	Payload   string
+	CreatedAt string
+	UpdatedAt string
+}
+
+func (q *Queries) CreateJob(ctx context.Context, arg CreateJobParams) error {
+	_, err := q.db.ExecContext(ctx, createJob,
+		arg.ID,
+		arg.Kind,
+		arg.Payload,
+		arg.CreatedAt,
+		arg.UpdatedAt,
+	)
+	return err
+}
+
+const failJob = `-- name: FailJob :execrows
+UPDATE jobs
+SET status = 'failed', error = ?1, finished_at = ?2, updated_at = ?2
+WHERE id = ?3 AND status = 'running'
+`
+
+type FailJobParams struct {
+	Error sql.NullString
+	Now   sql.NullString
+	ID    string
+}
+
+func (q *Queries) FailJob(ctx context.Context, arg FailJobParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, failJob, arg.Error, arg.Now, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const failRunningJobs = `-- name: FailRunningJobs :execrows
+UPDATE jobs
+SET status = 'failed', error = ?1, finished_at = ?2, updated_at = ?2
+WHERE status = 'running'
+`
+
+type FailRunningJobsParams struct {
+	Error sql.NullString
+	Now   sql.NullString
+}
+
+// Fails every job left running, at boot before any worker starts.
+func (q *Queries) FailRunningJobs(ctx context.Context, arg FailRunningJobsParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, failRunningJobs, arg.Error, arg.Now)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const getJob = `-- name: GetJob :one
+SELECT id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at FROM jobs
+WHERE id = ?
+`
+
+func (q *Queries) GetJob(ctx context.Context, id string) (Job, error) {
+	row := q.db.QueryRowContext(ctx, getJob, id)
+	var i Job
+	err := row.Scan(
+		&i.ID,
+		&i.Kind,
+		&i.Status,
+		&i.Payload,
+		&i.Result,
+		&i.Error,
+		&i.StartedAt,
+		&i.FinishedAt,
+		&i.CreatedAt,
+		&i.UpdatedAt,
+	)
+	return i, err
+}
+
+const publishJobResult = `-- name: PublishJobResult :execrows
+UPDATE jobs
+SET result = ?, updated_at = ?
+WHERE id = ? AND status = 'running'
+`
+
+type PublishJobResultParams struct {
+	Result    sql.NullString
+	UpdatedAt string
+	ID        string
+}
+
+// Stores a partial result of a running job.
+func (q *Queries) PublishJobResult(ctx context.Context, arg PublishJobResultParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, publishJobResult, arg.Result, arg.UpdatedAt, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}
+
+const succeedJob = `-- name: SucceedJob :execrows
+UPDATE jobs
+SET status = 'succeeded', result = ?1, finished_at = ?2, updated_at = ?2
+WHERE id = ?3 AND status = 'running'
+`
+
+type SucceedJobParams struct {
+	Result sql.NullString
+	Now    sql.NullString
+	ID     string
+}
+
+func (q *Queries) SucceedJob(ctx context.Context, arg SucceedJobParams) (int64, error) {
+	result, err := q.db.ExecContext(ctx, succeedJob, arg.Result, arg.Now, arg.ID)
+	if err != nil {
+		return 0, err
+	}
+	return result.RowsAffected()
+}

internal/db/querier.go

@@ -6,17 +6,25 @@ package db
 
 import (
 	"context"
+	"database/sql"
 )
 
 type Querier interface {
 	// An unknown id returns sql.ErrNoRows.
 	ArchiveDinner(ctx context.Context, arg ArchiveDinnerParams) (Dinner, error)
+	// Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
+	// when no job is queued.
+	ClaimNextJob(ctx context.Context, now sql.NullString) (Job, error)
 	CreateDinner(ctx context.Context, arg CreateDinnerParams) (Dinner, error)
+	CreateJob(ctx context.Context, arg CreateJobParams) error
 	CreatePhoto(ctx context.Context, arg CreatePhotoParams) (Photo, error)
 	CreateRecipeIngredient(ctx context.Context, arg CreateRecipeIngredientParams) error
 	CreateRecipeStep(ctx context.Context, arg CreateRecipeStepParams) error
 	DeleteRecipeIngredients(ctx context.Context, recipeID string) error
 	DeleteRecipeSteps(ctx context.Context, recipeID string) error
+	FailJob(ctx context.Context, arg FailJobParams) (int64, error)
+	// Fails every job left running, at boot before any worker starts.
+	FailRunningJobs(ctx context.Context, arg FailRunningJobsParams) (int64, error)
 	// Returns sql.ErrNoRows until the row exists.
 	GetAISettings(ctx context.Context) (AiSetting, error)
 	// Returns sql.ErrNoRows until the row exists.
@@ -26,6 +34,7 @@ type Querier interface {
 	GetHouseholdSettings(ctx context.Context) (HouseholdSetting, error)
 	// Returns sql.ErrNoRows until the row exists.
 	GetIntegrationsGoogle(ctx context.Context) (IntegrationsGoogle, error)
+	GetJob(ctx context.Context, id string) (Job, error)
 	// Finds stored image bytes by content hash. The schema does not make sha256
 	// unique, so the earliest-created (lowest id) match is returned.
 	GetPhotoBySHA256(ctx context.Context, sha256 string) (Photo, error)
@@ -36,8 +45,11 @@ type Querier interface {
 	ListDinnersByStatus(ctx context.Context, status string) ([]Dinner, error)
 	ListRecipeIngredients(ctx context.Context, recipeID string) ([]RecipeIngredient, error)
 	ListRecipeSteps(ctx context.Context, recipeID string) ([]RecipeStep, error)
+	// Stores a partial result of a running job.
+	PublishJobResult(ctx context.Context, arg PublishJobResultParams) (int64, error)
 	// An unknown id returns sql.ErrNoRows.
 	SetDinnerFavorite(ctx context.Context, arg SetDinnerFavoriteParams) (Dinner, error)
+	SucceedJob(ctx context.Context, arg SucceedJobParams) (int64, error)
 	// An unknown id returns sql.ErrNoRows.
 	UpdateDinner(ctx context.Context, arg UpdateDinnerParams) (Dinner, error)
 	// Creates the single row (id 1) or replaces every column except id and

internal/jobs/builtin.go

@@ -0,0 +1,44 @@
+package jobs
+
+import (
+	"context"
+	"encoding/json"
+)
+
+// KindNoop is the framework's built-in job kind: it does no work and
+// succeeds with {"ok":true}. It is the runner's own trivial kind, as Feature
+// 09's trivial test skill is the harness's, and the startup and readiness
+// job: a noop job reaching succeeded proves the runner claims, runs and
+// records jobs.
+const KindNoop = "noop"
+
+// NoopPayload is the payload of a noop job.
+type NoopPayload struct{}
+
+// NoopResult is the result of a noop job.
+type NoopResult struct{}
+
+// PublicJSON returns {"ok":true}.
+func (NoopResult) PublicJSON() ([]byte, error) {
+	return json.Marshal(struct {
+		OK bool `json:"ok"`
+	}{OK: true})
+}
+
+func runNoop(context.Context, NoopPayload, func(NoopResult) error) (NoopResult, error) {
+	return NoopResult{}, nil
+}
+
+// Builtins enqueues the framework's built-in job kinds.
+type Builtins struct {
+	Noop *Kind[NoopPayload]
+}
+
+// RegisterBuiltins registers the framework's built-in job kinds on r.
+func RegisterBuiltins(r *Runner) (*Builtins, error) {
+	noop, err := Register(r, KindNoop, runNoop)
+	if err != nil {
+		return nil, err
+	}
+	return &Builtins{Noop: noop}, nil
+}

internal/jobs/jobs.go

@@ -0,0 +1,410 @@
+// Package jobs runs long work in the background. A job is a row in the jobs
+// table: Enqueue inserts it queued, and a fixed pool of workers claims queued
+// jobs oldest first, so the table itself is the queue and a job enqueued
+// before a restart still runs after it.
+//
+// Each job kind has one handler, registered before Start. A handler takes a
+// typed payload and returns a result implementing Public; it may publish
+// partial results while it runs. The runner stores only what a result's
+// PublicJSON returns and, for a failed job, only a failure code: either a
+// *Failure code the kind declared at registration or one of the runner's own
+// codes. Handler error text, which may quote a URL, token or response body
+// from a dependency, is never stored, returned by the API or logged.
+package jobs
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"regexp"
+	"sync"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+)
+
+// Status is the state of a job.
+type Status string
+
+// Job states. A job moves queued → running → succeeded or failed.
+const (
+	StatusQueued    Status = "queued"
+	StatusRunning   Status = "running"
+	StatusSucceeded Status = "succeeded"
+	StatusFailed    Status = "failed"
+)
+
+// Failure codes the runner stores itself. A kind cannot declare them.
+const (
+	// CodeInterrupted is the error of a job a previous process left running.
+	// It is fixed framework text, not a snake-case code, so no kind can
+	// declare it.
+	CodeInterrupted = "interrupted by restart"
+	// CodeInternal is the error of a job whose handler failed with anything
+	// but a *Failure with a code its kind declared, or whose payload or
+	// result could not be encoded. The underlying error is neither stored
+	// nor logged.
+	CodeInternal = "internal_error"
+	// CodeUnknownKind is the error of a job whose kind has no handler.
+	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.
+const (
+	claimRetryMin = 50 * time.Millisecond
+	claimRetryMax = 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}$`)
+
+// ErrNotFound is returned by Get for an unknown job id.
+var ErrNotFound = errors.New("jobs: job not found")
+
+// Failure is a handler error reporting a failure code the household may
+// see. Code must be one the job's kind declared at registration; the runner
+// stores it as the job's error. Any other handler error, including a Failure
+// with an undeclared code, is stored as CodeInternal.
+type Failure struct {
+	Code string
+}
+
+func (f *Failure) Error() string { return "jobs: failure " + f.Code }
+
+// Public is a job result. PublicJSON returns the JSON the runner stores and
+// GET /api/v1/jobs/{id} returns, so a result type states exactly which of its
+// data is public; nothing else of the value is ever encoded.
+type Public interface {
+	PublicJSON() ([]byte, error)
+}
+
+// Handler runs one job with its payload. It may call publish any number of
+// times with a partial result, which Get returns while the job runs; its
+// return value is the final result. ctx is cancelled when the runner stops.
+type Handler[P any, R Public] func(ctx context.Context, payload P, publish func(partial R) error) (R, error)
+
+// Job is a job as reported by GET /api/v1/jobs/{id}. Result is the latest
+// partial or final result, and is null until the handler publishes one.
+// Error is set only on a failed job.
+type Job struct {
+	ID     string          `json:"id"`
+	Kind   string          `json:"kind"`
+	Status Status          `json:"status"`
+	Result json.RawMessage `json:"result"`
+	Error  *string         `json:"error"`
+}
+
+// runFunc is a registered Handler with its payload and results as JSON. It
+// returns the failure code to store, or "" on success; the logged reason
+// never quotes an error.
+type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (result json.RawMessage, code, reason string)
+
+// Runner owns the job handlers and the worker pool.
+type Runner struct {
+	st      *store.Store
+	logger  *slog.Logger
+	workers int
+	// wake holds up to one token per worker; Enqueue adds one so an idle
+	// worker looks for the new job.
+	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)
+
+	mu       sync.Mutex
+	handlers map[string]runFunc
+	started  bool
+
+	wg sync.WaitGroup
+}
+
+// NewRunner returns a runner over st with workers concurrent workers.
+func NewRunner(st *store.Store, logger *slog.Logger, workers int) (*Runner, error) {
+	if st == nil {
+		return nil, errors.New("jobs: nil store")
+	}
+	if logger == nil {
+		return nil, errors.New("jobs: nil logger")
+	}
+	if workers < 1 {
+		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]runFunc),
+	}, nil
+}
+
+// Kind enqueues jobs of one registered kind.
+type Kind[P any] struct {
+	r    *Runner
+	name string
+}
+
+// Register makes h the handler for kind and returns the Kind that enqueues
+// its jobs. failureCodes are the *Failure codes h may report. It must be
+// called before Start, once per kind.
+func Register[P any, R Public](r *Runner, kind string, h Handler[P, R], failureCodes ...string) (*Kind[P], error) {
+	if kind == "" {
+		return nil, errors.New("jobs: register: empty kind")
+	}
+	if h == nil {
+		return nil, fmt.Errorf("jobs: register %q: nil handler", kind)
+	}
+	declared := make(map[string]bool, len(failureCodes))
+	for _, code := range failureCodes {
+		if !codePattern.MatchString(code) {
+			return nil, fmt.Errorf("jobs: register %q: failure code %q is not lower-case snake case", kind, code)
+		}
+		if code == CodeInternal || code == CodeUnknownKind {
+			return nil, fmt.Errorf("jobs: register %q: failure code %q is reserved for the runner", kind, code)
+		}
+		declared[code] = true
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.started {
+		return nil, fmt.Errorf("jobs: register %q: runner already started", kind)
+	}
+	if _, ok := r.handlers[kind]; ok {
+		return nil, fmt.Errorf("jobs: register %q: kind already registered", kind)
+	}
+	r.handlers[kind] = func(ctx context.Context, payload string, publish func(json.RawMessage) error) (json.RawMessage, string, string) {
+		var p P
+		dec := json.NewDecoder(bytes.NewReader([]byte(payload)))
+		dec.DisallowUnknownFields()
+		if err := dec.Decode(&p); err != nil {
+			return nil, CodeInternal, "payload does not decode"
+		}
+		result, err := h(ctx, p, func(partial R) error {
+			data, err := publicJSON(partial)
+			if err != nil {
+				return err
+			}
+			return publish(data)
+		})
+		if err != nil {
+			var failure *Failure
+			if !errors.As(err, &failure) {
+				return nil, CodeInternal, "handler error"
+			}
+			if !declared[failure.Code] {
+				return nil, CodeInternal, "undeclared failure code"
+			}
+			return nil, failure.Code, "handler failure"
+		}
+		data, err := publicJSON(result)
+		if err != nil {
+			return nil, CodeInternal, "result does not encode"
+		}
+		return data, "", ""
+	}
+	return &Kind[P]{r: r, name: kind}, nil
+}
+
+// publicJSON returns v's PublicJSON, refusing anything that is not one JSON
+// value. Its error never quotes PublicJSON's error.
+func publicJSON(v Public) (json.RawMessage, error) {
+	data, err := v.PublicJSON()
+	if err != nil || !json.Valid(data) {
+		return nil, errors.New("jobs: result has no valid public JSON")
+	}
+	return data, nil
+}
+
+// Enqueue stores a queued job with payload and returns its id. A job
+// enqueued before Start waits for it.
+func (k *Kind[P]) Enqueue(ctx context.Context, payload P) (string, error) {
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return "", fmt.Errorf("jobs: enqueue %s: encode payload: %w", k.name, err)
+	}
+	id := store.NewID()
+	now := store.Now().String()
+	if err := db.New(k.r.st.Writer).CreateJob(ctx, db.CreateJobParams{
+		ID:        id,
+		Kind:      k.name,
+		Payload:   string(data),
+		CreatedAt: now,
+		UpdatedAt: now,
+	}); err != nil {
+		return "", fmt.Errorf("jobs: enqueue %s: %w", k.name, err)
+	}
+	select {
+	case k.r.wake <- struct{}{}:
+	default: // every worker already has a token and will look again
+	}
+	return id, nil
+}
+
+// Start fails every job a previous process left running with
+// CodeInterrupted, then starts the workers. They stop when ctx is cancelled;
+// Wait waits for them. A job whose handler returns an error after ctx is
+// cancelled is left running, so the next Start fails it as interrupted.
+func (r *Runner) Start(ctx context.Context) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.started {
+		return errors.New("jobs: runner already started")
+	}
+	now := store.Now().String()
+	n, err := db.New(r.st.Writer).FailRunningJobs(ctx, db.FailRunningJobsParams{
+		Error: sql.NullString{String: CodeInterrupted, Valid: true},
+		Now:   sql.NullString{String: now, Valid: true},
+	})
+	if err != nil {
+		return fmt.Errorf("jobs: fail interrupted jobs: %w", err)
+	}
+	if n > 0 {
+		r.logger.WarnContext(ctx, "failed jobs interrupted by restart", "count", n)
+	}
+	r.started = true
+	for range r.workers {
+		r.wg.Add(1)
+		go r.work(ctx)
+	}
+	return nil
+}
+
+// Wait blocks until every worker has stopped.
+func (r *Runner) Wait() {
+	r.wg.Wait()
+}
+
+// Get returns the job with id, or ErrNotFound.
+func (r *Runner) Get(ctx context.Context, id string) (Job, error) {
+	row, err := db.New(r.st.Reader).GetJob(ctx, id)
+	if errors.Is(err, sql.ErrNoRows) {
+		return Job{}, ErrNotFound
+	}
+	if err != nil {
+		return Job{}, fmt.Errorf("jobs: read job: %w", err)
+	}
+	job := Job{ID: row.ID, Kind: row.Kind, Status: Status(row.Status)}
+	if row.Result.Valid {
+		job.Result = json.RawMessage(row.Result.String)
+	}
+	if row.Error.Valid {
+		job.Error = &row.Error.String
+	}
+	return job, nil
+}
+
+// work claims and runs queued jobs until ctx is cancelled, waiting for a
+// wake token whenever none is queued. A claim that fails with a database
+// error is retried with backoff rather than waiting for a token, so a job
+// already queued is not stranded by a transient error.
+func (r *Runner) work(ctx context.Context) {
+	defer r.wg.Done()
+	backoff := claimRetryMin
+	for ctx.Err() == nil {
+		job, err := r.claim(ctx, sql.NullString{String: store.Now().String(), Valid: true})
+		switch {
+		case err == nil:
+			backoff = claimRetryMin
+			r.run(ctx, job)
+			continue
+		case errors.Is(err, sql.ErrNoRows):
+			backoff = claimRetryMin
+			select {
+			case <-r.wake:
+			case <-ctx.Done():
+			}
+			continue
+		case ctx.Err() != nil:
+			return
+		}
+		r.logger.ErrorContext(ctx, "claim job failed; retrying", "reason", "database error", "retry_in", backoff)
+		timer := time.NewTimer(backoff)
+		select {
+		case <-timer.C:
+		case <-ctx.Done():
+			timer.Stop()
+		}
+		backoff = min(2*backoff, claimRetryMax)
+	}
+}
+
+// run runs a claimed job to completion. The final write is made even when
+// 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)
+		return
+	}
+
+	publish := func(partial json.RawMessage) error {
+		n, err := q.PublishJobResult(ctx, db.PublishJobResultParams{
+			Result:    sql.NullString{String: string(partial), Valid: true},
+			UpdatedAt: store.Now().String(),
+			ID:        job.ID,
+		})
+		if err != nil {
+			return fmt.Errorf("jobs: publish result: %w", err)
+		}
+		if n != 1 {
+			return fmt.Errorf("jobs: publish result: job %s is not running", job.ID)
+		}
+		return nil
+	}
+
+	result, code, reason := h(ctx, job.Payload, publish)
+	if code != "" {
+		if ctx.Err() != nil {
+			logger.WarnContext(write, "job interrupted by shutdown; it fails at the next start")
+			return
+		}
+		level := slog.LevelInfo
+		if code == CodeInternal {
+			level = slog.LevelError
+		}
+		logger.Log(ctx, level, "job failed", "code", code, "reason", reason)
+		r.fail(write, 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,
+	})
+	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,
+	})
+	if err == nil && n != 1 {
+		err = errors.New("job is not running")
+	}
+	if err != nil {
+		logger.ErrorContext(ctx, "record job failure", "error", err)
+	}
+}

internal/jobs/jobs_test.go

@@ -0,0 +1,570 @@
+package jobs
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
+)
+
+// sentinelSecret stands in for a credential a handler holds while it runs.
+const sentinelSecret = "sk-sentinel-5f2c9a"
+
+type countPayload struct {
+	Steps int `json:"steps"`
+}
+
+type countResult struct {
+	Done int `json:"done"`
+}
+
+func (c countResult) PublicJSON() ([]byte, error) { return json.Marshal(c) }
+
+func migratedStore(t *testing.T) *store.Store {
+	t.Helper()
+	ctx := context.Background()
+	st, err := store.Open(ctx, t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := st.Close(); err != nil {
+			t.Error(err)
+		}
+	})
+	if err := store.Migrate(ctx, st.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	return st
+}
+
+func newRunner(t *testing.T, st *store.Store, workers int) *Runner {
+	t.Helper()
+	return newRunnerLogging(t, st, workers, io.Discard)
+}
+
+func newRunnerLogging(t *testing.T, st *store.Store, workers int, w io.Writer) *Runner {
+	t.Helper()
+	r, err := NewRunner(st, slog.New(slog.NewTextHandler(w, nil)), workers)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
+// logBuffer collects log output written by concurrent workers.
+type logBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+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 r and stops it when the test ends. The returned function
+// stops it early and waits for its workers.
+func start(t *testing.T, r *Runner) (stop func()) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(context.Background())
+	if err := r.Start(ctx); err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	stop = func() {
+		cancel()
+		r.Wait()
+	}
+	t.Cleanup(stop)
+	return stop
+}
+
+func register[P any, R Public](t *testing.T, r *Runner, kind string, h Handler[P, R], failureCodes ...string) *Kind[P] {
+	t.Helper()
+	k, err := Register(r, kind, h, failureCodes...)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return k
+}
+
+func enqueue[P any](t *testing.T, k *Kind[P], payload P) string {
+	t.Helper()
+	id, err := k.Enqueue(context.Background(), payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return id
+}
+
+// waitFor polls Get until done accepts the job, failing after five seconds.
+func waitFor(t *testing.T, r *Runner, id string, done func(Job) bool) Job {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		job, err := r.Get(context.Background(), id)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if done(job) {
+			return job
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("job %s never reached the expected state; last %+v (result %s)", id, job, job.Result)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func finished(j Job) bool { return j.Status == StatusSucceeded || j.Status == StatusFailed }
+
+// rawRow returns every column of the job's row joined, for leak checks.
+func rawRow(t *testing.T, st *store.Store, id string) string {
+	t.Helper()
+	var cols [10]any
+	ptrs := make([]any, len(cols))
+	for i := range cols {
+		ptrs[i] = &cols[i]
+	}
+	if err := st.Reader.QueryRow("SELECT * FROM jobs WHERE id = ?", id).Scan(ptrs...); err != nil {
+		t.Fatal(err)
+	}
+	return fmt.Sprint(cols[:]...)
+}
+
+func TestJobSucceeds(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	r := newRunner(t, st, 2)
+	count := register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
+		return countResult{Done: p.Steps}, nil
+	})
+	start(t, r)
+
+	id := enqueue(t, count, countPayload{Steps: 3})
+	job := waitFor(t, r, id, finished)
+	if job.ID != id || job.Kind != "count" || job.Status != StatusSucceeded || job.Error != nil {
+		t.Fatalf("job = %+v, want succeeded count job %s with no error", job, id)
+	}
+	if string(job.Result) != `{"done":3}` {
+		t.Errorf("result = %s, want {\"done\":3}", job.Result)
+	}
+
+	var startedAt, finishedAt, createdAt, updatedAt, payload string
+	if err := st.Reader.QueryRow("SELECT started_at, finished_at, created_at, updated_at, payload FROM jobs WHERE id = ?", id).
+		Scan(&startedAt, &finishedAt, &createdAt, &updatedAt, &payload); err != nil {
+		t.Fatal(err)
+	}
+	for name, v := range map[string]string{"started_at": startedAt, "finished_at": finishedAt, "created_at": createdAt, "updated_at": updatedAt} {
+		var ts store.Timestamp
+		if err := ts.Scan(v); err != nil {
+			t.Errorf("%s = %q: %v", name, v, err)
+		}
+	}
+	if payload != `{"steps":3}` {
+		t.Errorf("payload = %s, want {\"steps\":3}", payload)
+	}
+}
+
+func TestJobFailsWithFailureCode(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	r := newRunner(t, st, 1)
+	k := register(t, r, "suggest", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, fmt.Errorf("suggest: %w", &Failure{Code: "no_regulars"})
+	}, "no_regulars")
+	start(t, r)
+
+	job := waitFor(t, r, enqueue(t, k, countPayload{}), finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != "no_regulars" {
+		t.Fatalf("job = %+v, want failed with code no_regulars", job)
+	}
+	if job.Result != nil {
+		t.Errorf("result = %s, want none", job.Result)
+	}
+	var stored string
+	if err := st.Reader.QueryRow("SELECT error FROM jobs WHERE id = ?", job.ID).Scan(&stored); err != nil {
+		t.Fatal(err)
+	}
+	if stored != "no_regulars" {
+		t.Errorf("stored error = %q, want only the code", stored)
+	}
+}
+
+func TestUndeclaredFailureCodeIsInternal(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	k := register(t, r, "suggest", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, &Failure{Code: "not_declared"}
+	}, "no_regulars")
+	start(t, r)
+
+	job := waitFor(t, r, enqueue(t, k, countPayload{}), finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != CodeInternal {
+		t.Fatalf("job = %+v, want failed with %q", job, CodeInternal)
+	}
+}
+
+// credential holds a secret the way config.Secret does: unexported, so
+// encoding/json cannot see it.
+type credential struct{ value string }
+
+func (credential) String() string { return "[REDACTED]" }
+
+// leakyResult holds a secret in exported and unexported fields; only
+// Dinners is public.
+type leakyResult struct {
+	Dinners []string
+	Key     credential
+	Token   string
+	apiKey  string
+}
+
+func (l leakyResult) PublicJSON() ([]byte, error) {
+	return json.Marshal(struct {
+		Dinners []string `json:"dinners"`
+	}{l.Dinners})
+}
+
+// badResult has no valid public JSON; its encoding error quotes a secret.
+type badResult struct{}
+
+func (badResult) PublicJSON() ([]byte, error) {
+	return nil, fmt.Errorf("encode with key %s", sentinelSecret)
+}
+
+// TestSecretsNeverStoredOrLogged proves a handler holding a secret cannot
+// surface it in the jobs row, Get's result or the log through its result,
+// partial results, *Failure, error text or result encoding error.
+func TestSecretsNeverStoredOrLogged(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	var logs logBuffer
+	r := newRunnerLogging(t, st, 1, &logs)
+	secretResult := leakyResult{Dinners: []string{"Tacos"}, Key: credential{sentinelSecret}, Token: sentinelSecret, apiKey: sentinelSecret}
+	typed := register(t, r, "typed", func(_ context.Context, _ countPayload, publish func(leakyResult) error) (leakyResult, error) {
+		if err := publish(secretResult); err != nil {
+			return leakyResult{}, err
+		}
+		return secretResult, nil
+	})
+	leakyError := register(t, r, "leaky-error", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) {
+		return leakyResult{}, fmt.Errorf("GET https://api.example.com/v1?key=%s: 401 %s", sentinelSecret, sentinelSecret)
+	})
+	leakyFailure := register(t, r, "leaky-failure", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) {
+		return leakyResult{}, &Failure{Code: sentinelSecret}
+	}, "quota_exceeded")
+	failure := register(t, r, "failure", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) {
+		return leakyResult{}, fmt.Errorf("call with key %s: %w", sentinelSecret, &Failure{Code: "quota_exceeded"})
+	}, "quota_exceeded")
+	badEncoding := register(t, r, "bad-encoding", func(context.Context, countPayload, func(badResult) error) (badResult, error) {
+		return badResult{}, nil
+	})
+	stop := start(t, r)
+
+	cases := []struct {
+		kind       *Kind[countPayload]
+		wantStatus Status
+		wantResult string
+		wantError  string
+	}{
+		{typed, StatusSucceeded, `{"dinners":["Tacos"]}`, ""},
+		{leakyError, StatusFailed, "", CodeInternal},
+		{leakyFailure, StatusFailed, "", CodeInternal},
+		{failure, StatusFailed, "", "quota_exceeded"},
+		{badEncoding, StatusFailed, "", CodeInternal},
+	}
+	var ids []string
+	for _, c := range cases {
+		job := waitFor(t, r, enqueue(t, c.kind, countPayload{}), finished)
+		ids = append(ids, job.ID)
+		gotError := ""
+		if job.Error != nil {
+			gotError = *job.Error
+		}
+		if job.Status != c.wantStatus || string(job.Result) != c.wantResult || gotError != c.wantError {
+			t.Errorf("%s job = %+v (result %s), want %s with result %q and error %q", job.Kind, job, job.Result, c.wantStatus, c.wantResult, c.wantError)
+		}
+		body, err := json.Marshal(job)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if strings.Contains(string(body), sentinelSecret) {
+			t.Errorf("%s job as served contains the secret: %s", job.Kind, body)
+		}
+	}
+	stop()
+	for _, id := range ids {
+		if row := rawRow(t, st, id); strings.Contains(row, sentinelSecret) {
+			t.Errorf("job %s row contains the secret: %s", id, row)
+		}
+	}
+	if strings.Contains(logs.String(), sentinelSecret) {
+		t.Errorf("log contains the secret:\n%s", logs.String())
+	}
+	if !strings.Contains(logs.String(), "code="+CodeInternal) {
+		t.Errorf("log does not record the internal_error code:\n%s", logs.String())
+	}
+}
+
+func TestProgressiveResultsVisibleMidRun(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	release := make(chan struct{})
+	k := register(t, r, "draft", func(ctx context.Context, p countPayload, publish func(countResult) error) (countResult, error) {
+		for i := 1; i < p.Steps; i++ {
+			if err := publish(countResult{Done: i}); err != nil {
+				return countResult{}, err
+			}
+			select {
+			case <-release:
+			case <-ctx.Done():
+				return countResult{}, ctx.Err()
+			}
+		}
+		return countResult{Done: p.Steps}, nil
+	})
+	start(t, r)
+
+	id := enqueue(t, k, countPayload{Steps: 3})
+	for step := 1; step < 3; step++ {
+		want := fmt.Sprintf(`{"done":%d}`, step)
+		job := waitFor(t, r, id, func(j Job) bool { return string(j.Result) == want })
+		if job.Status != StatusRunning || job.Error != nil {
+			t.Fatalf("mid-run job = %+v, want running with no error", job)
+		}
+		release <- struct{}{}
+	}
+	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)
+	}
+}
+
+func TestRestartFailsInterruptedJobsAndRunsQueuedOnes(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+
+	first := newRunner(t, st, 1)
+	started := make(chan struct{})
+	k := register(t, first, "slow", func(ctx context.Context, _ countPayload, publish func(countResult) error) (countResult, error) {
+		if err := publish(countResult{Done: 1}); err != nil {
+			return countResult{}, err
+		}
+		close(started)
+		<-ctx.Done()
+		return countResult{}, ctx.Err()
+	})
+	stop := start(t, first)
+	interrupted := enqueue(t, k, countPayload{})
+	<-started
+	waiting := enqueue(t, k, countPayload{}) // the only worker is busy
+	stop()
+
+	if job := waitFor(t, first, interrupted, 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, "slow", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{Done: 9}, nil
+	})
+	start(t, second)
+
+	job := waitFor(t, second, interrupted, finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != "interrupted by restart" {
+		t.Errorf("interrupted job = %+v, want failed with %q", job, "interrupted by restart")
+	}
+	if string(job.Result) != `{"done":1}` {
+		t.Errorf("interrupted job result = %s, want the last partial result kept", job.Result)
+	}
+	if job := waitFor(t, second, waiting, finished); job.Status != StatusSucceeded {
+		t.Errorf("queued job = %+v, want run by the new runner", job)
+	}
+}
+
+func TestWorkerPoolIsBounded(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 2)
+	var running, peak atomic.Int32
+	release := make(chan struct{})
+	k := register(t, r, "busy", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		n := running.Add(1)
+		for {
+			p := peak.Load()
+			if n <= p || peak.CompareAndSwap(p, n) {
+				break
+			}
+		}
+		<-release
+		running.Add(-1)
+		return countResult{}, nil
+	})
+	start(t, r)
+
+	ids := make([]string, 5)
+	for i := range ids {
+		ids[i] = enqueue(t, k, countPayload{})
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for running.Load() < 2 {
+		if time.Now().After(deadline) {
+			t.Fatal("two workers never ran at once")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	close(release)
+	for _, id := range ids {
+		if job := waitFor(t, r, id, finished); job.Status != StatusSucceeded {
+			t.Errorf("job = %+v, want succeeded", job)
+		}
+	}
+	if got := peak.Load(); got != 2 {
+		t.Errorf("peak concurrent jobs = %d, want 2", got)
+	}
+}
+
+func TestJobWithoutHandlerFails(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	old := newRunner(t, st, 1)
+	k := register(t, old, "retired", func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, nil
+	})
+	id := enqueue(t, k, countPayload{}) // never started
+
+	r := newRunner(t, st, 1)
+	start(t, r)
+	job := waitFor(t, r, id, finished)
+	if job.Status != StatusFailed || job.Error == nil || *job.Error != CodeUnknownKind {
+		t.Errorf("job = %+v, want failed for a missing handler", job)
+	}
+}
+
+func TestGetUnknownID(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	if _, err := r.Get(context.Background(), store.NewID()); !errors.Is(err, ErrNotFound) {
+		t.Errorf("Get unknown id error = %v, want ErrNotFound", err)
+	}
+}
+
+func TestRegisterRefusesBadRegistrations(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	h := func(context.Context, countPayload, func(countResult) error) (countResult, error) {
+		return countResult{}, nil
+	}
+	register(t, r, "count", h)
+
+	if _, err := Register(r, "count", h); err == nil {
+		t.Error("duplicate kind registered")
+	}
+	if _, err := Register(r, "", h); err == nil {
+		t.Error("empty kind registered")
+	}
+	if _, err := Register[countPayload, countResult](r, "nil", nil); err == nil {
+		t.Error("nil handler registered")
+	}
+	for _, code := range []string{"", "Bad Code", "sk-sentinel", CodeInternal, CodeInterrupted, CodeUnknownKind} {
+		if _, err := Register(r, "code", h, code); err == nil {
+			t.Errorf("failure code %q accepted", code)
+		}
+	}
+	start(t, r)
+	if _, err := Register(r, "late", h); err == nil {
+		t.Error("kind registered after Start")
+	}
+	if err := r.Start(context.Background()); err == nil {
+		t.Error("second Start succeeded")
+	}
+}
+
+func TestNewRunnerRefusesBadConfig(t *testing.T) {
+	t.Parallel()
+	st := migratedStore(t)
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	if _, err := NewRunner(st, logger, 0); err == nil {
+		t.Error("zero workers accepted")
+	}
+	if _, err := NewRunner(nil, logger, 1); err == nil {
+		t.Error("nil store accepted")
+	}
+	if _, err := NewRunner(st, nil, 1); err == nil {
+		t.Error("nil logger accepted")
+	}
+}
+
+func TestNoopSucceeds(t *testing.T) {
+	t.Parallel()
+	r := newRunner(t, migratedStore(t), 1)
+	builtins, err := RegisterBuiltins(r)
+	if err != nil {
+		t.Fatal(err)
+	}
+	start(t, r)
+
+	job := waitFor(t, r, enqueue(t, builtins.Noop, NoopPayload{}), finished)
+	if job.Kind != KindNoop || job.Status != StatusSucceeded || string(job.Result) != `{"ok":true}` || job.Error != nil {
+		t.Errorf("noop job = %+v (result %s), want succeeded with {\"ok\":true}", job, job.Result)
+	}
+}
+
+func TestClaimErrorIsRetried(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
+	})
+	id := enqueue(t, k, countPayload{Steps: 3})
+	// Drop the wake token, as for a job queued before a restart, so only
+	// the retry can find the job.
+	<-r.wake
+
+	var claims atomic.Int32
+	claim := r.claim
+	r.claim = func(ctx context.Context, now sql.NullString) (db.Job, error) {
+		if claims.Add(1) == 1 {
+			return db.Job{}, errors.New("database is locked " + sentinelSecret)
+		}
+		return claim(ctx, now)
+	}
+	start(t, r)
+
+	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 n := claims.Load(); n < 2 {
+		t.Fatalf("claims = %d, want a retry after the error", n)
+	}
+	out := logs.String()
+	if !strings.Contains(out, "claim job failed; retrying") {
+		t.Fatalf("claim error not logged: %s", out)
+	}
+	if strings.Contains(out, sentinelSecret) {
+		t.Fatalf("claim error text logged: %s", out)
+	}
+}

internal/server/jobs.go

@@ -0,0 +1,29 @@
+package server
+
+import (
+	"errors"
+	"log/slog"
+	"net/http"
+
+	"github.com/go-chi/chi/v5"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+)
+
+// getJob serves GET /api/v1/jobs/{id}: the job's id, kind, status, latest
+// result and error.
+func getJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		job, err := runner.Get(r.Context(), chi.URLParam(r, "id"))
+		if errors.Is(err, jobs.ErrNotFound) {
+			writeError(w, r, logger, http.StatusNotFound, "not_found", "job not found")
+			return
+		}
+		if err != nil {
+			logger.ErrorContext(r.Context(), "get job", "error", err)
+			writeError(w, r, logger, http.StatusInternalServerError, "internal", "internal error")
+			return
+		}
+		writeJSON(w, r, logger, http.StatusOK, job)
+	}
+}

internal/server/jobs_test.go

@@ -0,0 +1,153 @@
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"maps"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"testing"
+	"time"
+
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
+)
+
+type draftPayload struct{}
+
+type draftResult struct {
+	Days int `json:"days"`
+}
+
+func (d draftResult) PublicJSON() ([]byte, error) { return json.Marshal(d) }
+
+// startedRunner registers a "draft" kind that publishes {"days":1}, waits
+// for release and returns {"days":7}, or fails with a *jobs.Failure when
+// fail is closed first. It starts the runner and stops it when the test ends.
+func startedRunner(t *testing.T, release, fail <-chan struct{}) (*jobs.Runner, *jobs.Kind[draftPayload]) {
+	t.Helper()
+	runner := newTestRunner(t)
+	kind, err := jobs.Register(runner, "draft", func(ctx context.Context, _ draftPayload, publish func(draftResult) error) (draftResult, error) {
+		if err := publish(draftResult{Days: 1}); err != nil {
+			return draftResult{}, err
+		}
+		select {
+		case <-release:
+			return draftResult{Days: 7}, nil
+		case <-fail:
+			return draftResult{}, &jobs.Failure{Code: "no_regulars"}
+		case <-ctx.Done():
+			return draftResult{}, ctx.Err()
+		}
+	}, "no_regulars")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	if err := runner.Start(ctx); err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		cancel()
+		runner.Wait()
+	})
+	return runner, kind
+}
+
+// pollJob GETs the job until want accepts the body, failing after five
+// seconds.
+func pollJob(t *testing.T, h http.Handler, id string, want func(map[string]any) bool) map[string]any {
+	t.Helper()
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		rec, body := serve(t, h, signedIn(http.MethodGet, "/api/v1/jobs/"+id, nil))
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status = %d, want 200: %s", rec.Code, rec.Body)
+		}
+		if want(body) {
+			if keys := slices.Sorted(maps.Keys(body)); !slices.Equal(keys, []string{"error", "id", "kind", "result", "status"}) {
+				t.Errorf("body fields = %v, want id, kind, status, result, error", keys)
+			}
+			return body
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("job %s never reached the expected state; last %v", id, body)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+func resultDays(body map[string]any) float64 {
+	result, _ := body["result"].(map[string]any)
+	days, _ := result["days"].(float64)
+	return days
+}
+
+func TestGetJobShowsProgressThenResult(t *testing.T) {
+	t.Parallel()
+	release := make(chan struct{})
+	runner, kind := startedRunner(t, release, nil)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+	id, err := kind.Enqueue(context.Background(), draftPayload{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	mid := pollJob(t, router, id, func(b map[string]any) bool { return resultDays(b) == 1 })
+	if mid["id"] != id || mid["kind"] != "draft" || mid["status"] != "running" || mid["error"] != nil {
+		t.Errorf("mid-run body = %v, want running draft job %s with no error", mid, id)
+	}
+
+	close(release)
+	done := pollJob(t, router, id, func(b map[string]any) bool { return b["status"] == "succeeded" })
+	if resultDays(done) != 7 || done["error"] != nil {
+		t.Errorf("finished body = %v, want result {\"days\":7} and no error", done)
+	}
+}
+
+func TestGetJobShowsFailure(t *testing.T) {
+	t.Parallel()
+	fail := make(chan struct{})
+	close(fail)
+	runner, kind := startedRunner(t, nil, fail)
+	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
+	id, err := kind.Enqueue(context.Background(), draftPayload{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	body := pollJob(t, router, id, func(b map[string]any) bool { return b["status"] == "failed" })
+	if body["error"] != "no_regulars" {
+		t.Errorf("error = %v, want the Failure code", body["error"])
+	}
+}
+
+func TestGetJobUnknownIDIsNotFound(t *testing.T) {
+	t.Parallel()
+	router := newTestRouter(t, "dev")
+
+	rec, _ := serve(t, router, signedIn(http.MethodGet, "/api/v1/jobs/"+store.NewID(), nil))
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("status = %d, want 404", rec.Code)
+	}
+	var body errorBody
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatal(err)
+	}
+	if want := (errorBody{Error: apiError{Code: "not_found", Message: "job not found"}}); body != want {
+		t.Errorf("body = %+v, want %+v", body, want)
+	}
+}
+
+func TestGetJobRequiresSession(t *testing.T) {
+	t.Parallel()
+	router := newTestRouter(t, "dev")
+
+	rec := httptest.NewRecorder()
+	router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/v1/jobs/"+store.NewID(), nil))
+	if rec.Code != http.StatusUnauthorized {
+		t.Errorf("status = %d, want 401", rec.Code)
+	}
+}

internal/server/login_test.go

@@ -83,7 +83,7 @@ func newLoginRouter(t *testing.T, issuer string) http.Handler {
 		t.Fatal(err)
 	}
 	r, err := NewRouter(logger, "dev", trusted, flow, sessions, newTestSettings(t),
-		fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
+		newTestRunner(t), fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/server/origin_test.go

@@ -45,7 +45,7 @@ func newOriginRouter(t *testing.T, logs io.Writer) http.Handler {
 	}
 	logger := slog.New(forwarded.LogHandler(slog.NewJSONHandler(logs, nil)))
 	r, err := NewRouter(logger, "dev", trusted, probeAuth{}, stubSessions{}, newTestSettings(t),
-		fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
+		newTestRunner(t), fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/server/server.go

@@ -13,6 +13,7 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/forwarded"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/frontend"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 )
 
 // reservedPrefixes are owned by the API router and are never answered with the
@@ -28,13 +29,17 @@ type RouteMounter interface {
 // has its client address and scheme resolved, honouring forwarded headers
 // only from a peer in trusted (see resolveOrigin). version is reported by
 // GET /health; auth mounts the /auth routes; sessions guards /api/v1 and
-// /media; settings serves /api/v1/settings/household; assets is the web build
+// /media; settings serves /api/v1/settings/household; runner serves
+// /api/v1/jobs/{id}; assets is the web build
 // served for every other GET. /health, /auth and the web build are public.
 // Requests to /api/v1 other than GET also need the session's CSRF token in CSRFHeader.
-func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, auth RouteMounter, sessions SessionLookup, settings *household.SettingsService, assets fs.FS) (http.Handler, error) {
+func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, auth RouteMounter, sessions SessionLookup, settings *household.SettingsService, runner *jobs.Runner, assets fs.FS) (http.Handler, error) {
 	if settings == nil {
 		return nil, errors.New("server: nil household settings service")
 	}
+	if runner == nil {
+		return nil, errors.New("server: nil job runner")
+	}
 	notFound := apiErrorHandler(logger, http.StatusNotFound, "not_found", "not found")
 	methodNotAllowed := apiErrorHandler(logger, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
 
@@ -63,6 +68,7 @@ func NewRouter(logger *slog.Logger, version string, trusted forwarded.Trusted, a
 	apiV1.Get("/me", me(logger))
 	apiV1.Get("/settings/household", getHouseholdSettings(logger, settings))
 	apiV1.Patch("/settings/household", patchHouseholdSettings(logger, settings))
+	apiV1.Get("/jobs/{id}", getJob(logger, runner))
 	r.Route("/api", func(r chi.Router) {
 		r.MethodNotAllowed(methodNotAllowed)
 		r.NotFound(notFound)

internal/server/server_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/go-chi/chi/v5"
 
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/household"
+	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/store"
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/migrations"
 )
@@ -27,17 +28,17 @@ const (
 
 func newTestRouter(t *testing.T, version string) http.Handler {
 	t.Helper()
-	return newTestRouterWith(t, version, newTestSettings(t))
+	return newTestRouterWith(t, version, newTestSettings(t), newTestRunner(t))
 }
 
-func newTestRouterWith(t *testing.T, version string, settings *household.SettingsService) http.Handler {
+func newTestRouterWith(t *testing.T, version string, settings *household.SettingsService, runner *jobs.Runner) http.Handler {
 	t.Helper()
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 	assets := fstest.MapFS{
 		"index.html":             {Data: []byte(fixtureIndex)},
 		"assets/index-AbC123.js": {Data: []byte(fixtureAsset)},
 	}
-	r, err := NewRouter(logger, version, nil, stubAuth{}, stubSessions{}, settings, assets)
+	r, err := NewRouter(logger, version, nil, stubAuth{}, stubSessions{}, settings, runner, assets)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -70,9 +71,8 @@ func TestAuthRoutesMounted(t *testing.T) {
 	}
 }
 
-// newTestSettings returns a settings service over a migrated temp database
-// seeded with zone America/New_York.
-func newTestSettings(t *testing.T) *household.SettingsService {
+// newTestStore returns a migrated temp database, closed when the test ends.
+func newTestStore(t *testing.T) *store.Store {
 	t.Helper()
 	ctx := context.Background()
 	st, err := store.Open(ctx, t.TempDir())
@@ -87,21 +87,49 @@ func newTestSettings(t *testing.T) *household.SettingsService {
 	if err := store.Migrate(ctx, st.Writer, migrations.FS); err != nil {
 		t.Fatal(err)
 	}
-	if err := household.SeedSettings(ctx, st, "America/New_York"); err != nil {
+	return st
+}
+
+// newTestSettings returns a settings service over a migrated temp database
+// seeded with zone America/New_York.
+func newTestSettings(t *testing.T) *household.SettingsService {
+	t.Helper()
+	st := newTestStore(t)
+	if err := household.SeedSettings(context.Background(), st, "America/New_York"); err != nil {
 		t.Fatal(err)
 	}
 	return household.NewSettingsService(st)
 }
 
+// newTestRunner returns an unstarted job runner over a migrated temp
+// database.
+func newTestRunner(t *testing.T) *jobs.Runner {
+	t.Helper()
+	runner, err := jobs.NewRunner(newTestStore(t), slog.New(slog.NewTextHandler(io.Discard, nil)), 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return runner
+}
+
 func TestNewRouterRequiresSettings(t *testing.T) {
 	t.Parallel()
 
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
-	if _, err := NewRouter(logger, "dev", nil, stubAuth{}, stubSessions{}, nil, fstest.MapFS{"index.html": {}}); err == nil {
+	if _, err := NewRouter(logger, "dev", nil, stubAuth{}, stubSessions{}, nil, newTestRunner(t), fstest.MapFS{"index.html": {}}); err == nil {
 		t.Fatal("NewRouter accepted a nil settings service")
 	}
 }
 
+func TestNewRouterRequiresJobRunner(t *testing.T) {
+	t.Parallel()
+
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	if _, err := NewRouter(logger, "dev", nil, stubAuth{}, stubSessions{}, newTestSettings(t), nil, fstest.MapFS{"index.html": {}}); err == nil {
+		t.Fatal("NewRouter accepted a nil job runner")
+	}
+}
+
 func TestHealth(t *testing.T) {
 	t.Parallel()
 

internal/server/settings_test.go

@@ -84,7 +84,7 @@ func TestGetHouseholdSettings(t *testing.T) {
 func TestPatchHouseholdSettings(t *testing.T) {
 	t.Parallel()
 	settings := newTestSettings(t)
-	router := newTestRouterWith(t, "dev", settings)
+	router := newTestRouterWith(t, "dev", settings, newTestRunner(t))
 
 	var notified []household.Settings
 	settings.Subscribe(func(s household.Settings) { notified = append(notified, s) })
@@ -146,7 +146,7 @@ func TestPatchHouseholdSettingsRejected(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
 			settings := newTestSettings(t)
-			router := newTestRouterWith(t, "dev", settings)
+			router := newTestRouterWith(t, "dev", settings, newTestRunner(t))
 			notified := 0
 			settings.Subscribe(func(household.Settings) { notified++ })
 
@@ -216,7 +216,7 @@ func TestHouseholdSettingsWithRealSession(t *testing.T) {
 	}
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
 	router, err := NewRouter(logger, "dev", nil, stubAuth{}, sessions, household.NewSettingsService(st),
-		fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
+		newTestRunner(t), fstest.MapFS{"index.html": {Data: []byte(fixtureIndex)}})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/store/migrate_test.go

@@ -175,8 +175,8 @@ func TestMigrateEmbeddedMigrations(t *testing.T) {
 	if err := Migrate(context.Background(), s.Writer, migrations.FS); err != nil {
 		t.Fatal(err)
 	}
-	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by" {
-		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by", got)
+	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status" {
+		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status", got)
 	}
 }
 
@@ -196,8 +196,8 @@ func TestMigrateBootstrapThenEmbeddedInitApplies(t *testing.T) {
 	if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
 		t.Fatal(err)
 	}
-	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by" {
-		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by", got)
+	if got := versions(migrationRows(t, s)); got != "0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status" {
+		t.Errorf("versions = %q, want 0000_bootstrap,0001_init,0002_household_settings_updated_by,0003_jobs_status", got)
 	}
 	if !tableExists(t, s, "dinners") {
 		t.Error("table from 0001_init was not created")
@@ -262,3 +262,50 @@ func TestMigrateReopenSameDataDirIsNoOp(t *testing.T) {
 		t.Errorf("schema changed on reopen:\n got %v\nwant %v", afterSchema, beforeSchema)
 	}
 }
+
+// TestMigrateJobsStatusMapsLegacyRows upgrades a database holding jobs in
+// every 0001 status and checks 0003 carries each row over with its status
+// mapped and every other column unchanged.
+func TestMigrateJobsStatusMapsLegacyRows(t *testing.T) {
+	ctx := context.Background()
+	s := openTestStore(t)
+	before := fstest.MapFS{}
+	for _, name := range []string{"0000_bootstrap.sql", "0001_init.sql", "0002_household_settings_updated_by.sql"} {
+		body, err := fs.ReadFile(migrations.FS, name)
+		if err != nil {
+			t.Fatal(err)
+		}
+		before[name] = &fstest.MapFile{Data: body}
+	}
+	if err := Migrate(ctx, s.Writer, before); err != nil {
+		t.Fatal(err)
+	}
+	legacy := map[string]string{"pending": "queued", "running": "running", "done": "succeeded", "failed": "failed"}
+	for old := range legacy {
+		if _, err := s.Writer.Exec(`INSERT INTO jobs (id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at)
+			VALUES (?, 'suggest', ?, '{"n":1}', '{"r":2}', 'boom', 's', 'f', 'c', 'u')`, "job-"+old, old); err != nil {
+			t.Fatalf("insert %s job: %v", old, err)
+		}
+	}
+
+	if err := Migrate(ctx, s.Writer, migrations.FS); err != nil {
+		t.Fatal(err)
+	}
+	for old, want := range legacy {
+		var kind, status, payload, result, errText, startedAt, finishedAt, createdAt, updatedAt string
+		if err := s.Reader.QueryRow(`SELECT kind, status, payload, result, error, started_at, finished_at, created_at, updated_at FROM jobs WHERE id = ?`, "job-"+old).
+			Scan(&kind, &status, &payload, &result, &errText, &startedAt, &finishedAt, &createdAt, &updatedAt); err != nil {
+			t.Fatalf("read %s job: %v", old, err)
+		}
+		if status != want {
+			t.Errorf("%s job status = %q, want %q", old, status, want)
+		}
+		got := []string{kind, payload, result, errText, startedAt, finishedAt, createdAt, updatedAt}
+		if wantCols := []string{"suggest", `{"n":1}`, `{"r":2}`, "boom", "s", "f", "c", "u"}; !slices.Equal(got, wantCols) {
+			t.Errorf("%s job columns = %v, want %v", old, got, wantCols)
+		}
+	}
+	if tableExists(t, s, "jobs_new") {
+		t.Error("jobs_new left behind")
+	}
+}

internal/store/schema_test.go

@@ -165,7 +165,7 @@ func testRow(table string, i int) map[string]any {
 		r["attempts"] = 0
 	case "jobs":
 		r["kind"] = "suggest"
-		r["status"] = "pending"
+		r["status"] = "queued"
 		r["payload"] = "{}"
 	case "schedule_runs":
 		r["schedule"] = "plan_draft"
@@ -330,7 +330,7 @@ func TestSchemaEnumChecksRejectInvalidValues(t *testing.T) {
 		{"notification_prefs", "chat", 2},
 		{"notification_outbox", "channel", "email"},
 		{"notification_outbox", "status", "queued"},
-		{"jobs", "status", "queued"},
+		{"jobs", "status", "bogus"},
 		{"schedule_runs", "schedule", "hourly"},
 	}
 	s := seededStore(t)

migrations/0003_jobs_status.sql

@@ -0,0 +1,32 @@
+-- 0003_jobs_status: jobs.status becomes queued, running, succeeded, failed,
+-- the states the job runner (internal/jobs) reports. SQLite cannot change a
+-- CHECK in place, so the table is rebuilt: create the new table, copy every
+-- row, drop the old one and rename. Nothing references jobs and it has no
+-- indexes or triggers, so the rebuild touches no other schema object. Rows
+-- written under the 0001 values are mapped (pending -> queued,
+-- done -> succeeded) so the copy cannot fail on an existing database.
+
+CREATE TABLE jobs_new (
+	id          TEXT NOT NULL PRIMARY KEY,
+	kind        TEXT NOT NULL,
+	status      TEXT NOT NULL CHECK (status IN ('queued', 'running', 'succeeded', 'failed')),
+	payload     TEXT NOT NULL, -- JSON
+	result      TEXT,          -- JSON, written progressively
+	error       TEXT,
+	started_at  TEXT,
+	finished_at TEXT,
+	created_at  TEXT NOT NULL,
+	updated_at  TEXT NOT NULL
+);
+
+INSERT INTO jobs_new (id, kind, status, payload, result, error, started_at, finished_at, created_at, updated_at)
+SELECT
+	id,
+	kind,
+	CASE status WHEN 'pending' THEN 'queued' WHEN 'done' THEN 'succeeded' ELSE status END,
+	payload, result, error, started_at, finished_at, created_at, updated_at
+FROM jobs;
+
+DROP TABLE jobs;
+
+ALTER TABLE jobs_new RENAME TO jobs;

migrations/queries/jobs.sql

@@ -0,0 +1,37 @@
+-- name: CreateJob :exec
+INSERT INTO jobs (id, kind, status, payload, created_at, updated_at)
+VALUES (?, ?, 'queued', ?, ?, ?);
+
+-- name: ClaimNextJob :one
+-- Marks the oldest queued job running and returns it. Returns sql.ErrNoRows
+-- when no job is queued.
+UPDATE jobs
+SET status = 'running', started_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE id = (SELECT id FROM jobs WHERE status = 'queued' ORDER BY id LIMIT 1)
+RETURNING *;
+
+-- name: PublishJobResult :execrows
+-- Stores a partial result of a running job.
+UPDATE jobs
+SET result = ?, updated_at = ?
+WHERE id = ? AND status = 'running';
+
+-- name: SucceedJob :execrows
+UPDATE jobs
+SET status = 'succeeded', result = sqlc.arg(result), finished_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'running';
+
+-- name: FailJob :execrows
+UPDATE jobs
+SET status = 'failed', error = sqlc.arg(error), finished_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE id = sqlc.arg(id) AND status = 'running';
+
+-- name: FailRunningJobs :execrows
+-- Fails every job left running, at boot before any worker starts.
+UPDATE jobs
+SET status = 'failed', error = sqlc.arg(error), finished_at = sqlc.arg(now), updated_at = sqlc.arg(now)
+WHERE status = 'running';
+
+-- name: GetJob :one
+SELECT * FROM jobs
+WHERE id = ?;