From f37f240b26c533890eb80fab6005b5fccecadd89 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Sat, 3 Oct 2026 15:05:23 -0400 Subject: [PATCH] Add async job runner and GET /api/v1/jobs/{id} internal/jobs persists jobs in the jobs table, which is also the queue: a bounded pool of workers claims queued jobs oldest first. Handlers are registered per kind before Start with typed payloads and results, and can publish partial results to the row while they run. Only typed results and *jobs.Failure messages are stored; any other handler error is logged and stored as "internal error", so dependency error text cannot surface a secret. Start fails jobs a previous process left running with "interrupted by restart" before any worker runs. GET /api/v1/jobs/{id} returns {id, kind, status, result, error} inside the protected /api/v1 group; an unknown id is a 404 not_found. Migration 0003_jobs_status rebuilds jobs with status queued, running, succeeded, failed, mapping legacy pending and done rows. Co-Authored-By: Claude Opus 5.5 --- cmd/server/main.go | 23 +- internal/db/jobs.sql.go | 166 +++++++++++++ internal/db/querier.go | 12 + internal/jobs/jobs.go | 335 +++++++++++++++++++++++++ internal/jobs/jobs_test.go | 404 +++++++++++++++++++++++++++++++ internal/server/jobs.go | 29 +++ internal/server/jobs_test.go | 151 ++++++++++++ internal/server/login_test.go | 2 +- internal/server/origin_test.go | 2 +- internal/server/server.go | 10 +- internal/server/server_test.go | 44 +++- internal/server/settings_test.go | 6 +- internal/store/migrate_test.go | 55 ++++- internal/store/schema_test.go | 4 +- migrations/0003_jobs_status.sql | 32 +++ migrations/queries/jobs.sql | 37 +++ 16 files changed, 1290 insertions(+), 22 deletions(-) create mode 100644 internal/db/jobs.sql.go create mode 100644 internal/jobs/jobs.go create mode 100644 internal/jobs/jobs_test.go create mode 100644 internal/server/jobs.go create mode 100644 internal/server/jobs_test.go create mode 100644 migrations/0003_jobs_status.sql create mode 100644 migrations/queries/jobs.sql diff --git a/cmd/server/main.go b/cmd/server/main.go index e532ea8..ac18300 100644 --- a/cmd/server/main.go +++ b/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=". 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,11 @@ func run(cfg config.Config, logger *slog.Logger) error { return err } settings := household.NewSettingsService(db) + // Job handlers are registered on the runner here, before Start. + runner, err := jobs.NewRunner(db, logger, jobWorkers) + if err != nil { + return err + } sessions, err := auth.NewSessionManager(db, cfg.SessionIdle, cfg.SessionMax) if err != nil { @@ -80,10 +89,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, diff --git a/internal/db/jobs.sql.go b/internal/db/jobs.sql.go new file mode 100644 index 0000000..7835ceb --- /dev/null +++ b/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() +} diff --git a/internal/db/querier.go b/internal/db/querier.go index bc63ad5..f283584 100644 --- a/internal/db/querier.go +++ b/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 diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go new file mode 100644 index 0000000..ddceb10 --- /dev/null +++ b/internal/jobs/jobs.go @@ -0,0 +1,335 @@ +// 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 typed result, which the runner stores as JSON; +// it may publish partial results while it runs. Only typed data and +// *Failure messages are stored, so no handler error text, which may quote a +// URL, token or response body from a dependency, can reach the table or the +// API. +package jobs + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log/slog" + "sync" + + "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" +) + +// Stored error messages the runner writes itself. +const ( + // MessageInterrupted is the error of a job a previous process left running. + MessageInterrupted = "interrupted by restart" + // MessageInternal is the error of a job whose handler failed with an error + // that is not a *Failure. The handler's error is logged, not stored. + MessageInternal = "internal error" +) + +// ErrNotFound is returned by Get for an unknown job id. +var ErrNotFound = errors.New("jobs: job not found") + +// Failure is a handler error whose Message is safe to show the household and +// is stored as the job's error. Every other handler error is stored as +// MessageInternal. +type Failure struct { + Message string +} + +func (f *Failure) Error() string { return f.Message } + +// 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, R any] 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. +type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (json.RawMessage, error) + +// 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{} + + 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), + 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. It must be called before Start, once per kind. +func Register[P, R any](r *Runner, kind string, h Handler[P, R]) (*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) + } + 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, error) { + var p P + dec := json.NewDecoder(bytes.NewReader([]byte(payload))) + dec.DisallowUnknownFields() + if err := dec.Decode(&p); err != nil { + return nil, fmt.Errorf("decode payload: %w", err) + } + result, err := h(ctx, p, func(partial R) error { + data, err := json.Marshal(partial) + if err != nil { + return fmt.Errorf("jobs: encode partial result: %w", err) + } + return publish(data) + }) + if err != nil { + return nil, err + } + data, err := json.Marshal(result) + if err != nil { + return nil, fmt.Errorf("encode result: %w", err) + } + return data, nil + } + return &Kind[P]{r: r, name: kind}, 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 +// MessageInterrupted, 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: MessageInterrupted, 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. +func (r *Runner) work(ctx context.Context) { + defer r.wg.Done() + q := db.New(r.st.Writer) + for ctx.Err() == nil { + job, err := q.ClaimNextJob(ctx, sql.NullString{String: store.Now().String(), Valid: true}) + if err == nil { + r.run(ctx, job) + continue + } + if !errors.Is(err, sql.ErrNoRows) && ctx.Err() == nil { + r.logger.ErrorContext(ctx, "claim job", "error", err) + } + select { + case <-r.wake: + case <-ctx.Done(): + } + } +} + +// 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, "no handler for job kind") + r.fail(write, logger, job.ID, "no handler for job kind "+job.Kind) + 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, err := h(ctx, job.Payload, publish) + if err != nil { + if ctx.Err() != nil { + logger.WarnContext(write, "job interrupted by shutdown; it fails at the next start", "error", err) + return + } + var failure *Failure + message := MessageInternal + if errors.As(err, &failure) { + message = failure.Message + } else { + logger.ErrorContext(ctx, "job failed", "error", err) + } + r.fail(write, logger, job.ID, message) + 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) + } +} + +func (r *Runner) fail(ctx context.Context, logger *slog.Logger, id, message string) { + n, err := db.New(r.st.Writer).FailJob(ctx, db.FailJobParams{ + Error: sql.NullString{String: message, 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) + } +} diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go new file mode 100644 index 0000000..37c5fdf --- /dev/null +++ b/internal/jobs/jobs_test.go @@ -0,0 +1,404 @@ +package jobs + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "strings" + "sync/atomic" + "testing" + "time" + + "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 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() + r, err := NewRunner(st, slog.New(slog.NewTextHandler(io.Discard, nil)), workers) + if err != nil { + t.Fatal(err) + } + return r +} + +// 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, R any](t *testing.T, r *Runner, kind string, h Handler[P, R]) *Kind[P] { + t.Helper() + k, err := Register(r, kind, h) + 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 TestJobFailsWithFailureMessage(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{}, fmt.Errorf("suggest: %w", &Failure{Message: "add some Regulars first"}) + }) + start(t, r) + + job := waitFor(t, r, enqueue(t, k, countPayload{}), finished) + if job.Status != StatusFailed || job.Error == nil || *job.Error != "add some Regulars first" { + t.Fatalf("job = %+v, want failed with the Failure message", job) + } + if job.Result != nil { + t.Errorf("result = %s, want none", job.Result) + } +} + +// 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]" } + +type leakyResult struct { + Dinners []string `json:"dinners"` + Key credential `json:"key"` + Token string `json:"-"` + apiKey string +} + +// TestSecretsNeverStored proves a handler holding a secret cannot surface it +// through its typed result, its partial results or its error text. +func TestSecretsNeverStored(t *testing.T) { + t.Parallel() + st := migratedStore(t) + r := newRunner(t, st, 1) + secretResult := leakyResult{Dinners: []string{"Tacos"}, Key: credential{sentinelSecret}, Token: sentinelSecret, apiKey: sentinelSecret} + ok := 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 + }) + failing := 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) + }) + start(t, r) + + typed := waitFor(t, r, enqueue(t, ok, countPayload{}), finished) + if typed.Status != StatusSucceeded || string(typed.Result) != `{"dinners":["Tacos"],"key":{}}` { + t.Errorf("typed job = %+v (result %s), want succeeded with only the declared data", typed, typed.Result) + } + failed := waitFor(t, r, enqueue(t, failing, countPayload{}), finished) + if failed.Status != StatusFailed || failed.Error == nil || *failed.Error != MessageInternal { + t.Errorf("leaky-error job = %+v, want failed with %q", failed, MessageInternal) + } + for _, id := range []string{typed.ID, failed.ID} { + if row := rawRow(t, st, id); strings.Contains(row, sentinelSecret) { + t.Errorf("job %s row contains the secret: %s", id, row) + } + } +} + +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 != MessageInterrupted { + t.Errorf("interrupted job = %+v, want failed with %q", job, MessageInterrupted) + } + 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 != "no handler for job kind retired" { + 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") + } + 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") + } +} diff --git a/internal/server/jobs.go b/internal/server/jobs.go new file mode 100644 index 0000000..2fdd15f --- /dev/null +++ b/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) + } +} diff --git a/internal/server/jobs_test.go b/internal/server/jobs_test.go new file mode 100644 index 0000000..3994daf --- /dev/null +++ b/internal/server/jobs_test.go @@ -0,0 +1,151 @@ +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"` +} + +// 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{Message: "no Regulars to plan from"} + case <-ctx.Done(): + return draftResult{}, ctx.Err() + } + }) + 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 to plan from" { + t.Errorf("error = %v, want the Failure message", 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) + } +} diff --git a/internal/server/login_test.go b/internal/server/login_test.go index 0b8ada9..88b60a5 100644 --- a/internal/server/login_test.go +++ b/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) } diff --git a/internal/server/origin_test.go b/internal/server/origin_test.go index e4d0030..b686164 100644 --- a/internal/server/origin_test.go +++ b/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) } diff --git a/internal/server/server.go b/internal/server/server.go index c296893..a01adf3 100644 --- a/internal/server/server.go +++ b/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) diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 5b59d2d..136d9f8 100644 --- a/internal/server/server_test.go +++ b/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() diff --git a/internal/server/settings_test.go b/internal/server/settings_test.go index 2fdfbca..e9da996 100644 --- a/internal/server/settings_test.go +++ b/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) } diff --git a/internal/store/migrate_test.go b/internal/store/migrate_test.go index cc4f356..e104996 100644 --- a/internal/store/migrate_test.go +++ b/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") + } +} diff --git a/internal/store/schema_test.go b/internal/store/schema_test.go index ed61047..2b36ad1 100644 --- a/internal/store/schema_test.go +++ b/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) diff --git a/migrations/0003_jobs_status.sql b/migrations/0003_jobs_status.sql new file mode 100644 index 0000000..c5adef8 --- /dev/null +++ b/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; diff --git a/migrations/queries/jobs.sql b/migrations/queries/jobs.sql new file mode 100644 index 0000000..6d2bad9 --- /dev/null +++ b/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 = ?;