diff --git a/cmd/server/jobs_test.go b/cmd/server/jobs_test.go new file mode 100644 index 0000000..8d02103 --- /dev/null +++ b/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) + } +} diff --git a/cmd/server/main.go b/cmd/server/main.go index ac18300..4f46d29 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -67,8 +67,7 @@ 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) + runner, _, err := newJobRunner(db, logger) if err != nil { return err } @@ -186,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) { diff --git a/internal/jobs/builtin.go b/internal/jobs/builtin.go new file mode 100644 index 0000000..754babe --- /dev/null +++ b/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 +} diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index ddceb10..16f3b5f 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -4,11 +4,12 @@ // 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. +// 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 ( @@ -19,6 +20,7 @@ import ( "errors" "fmt" "log/slog" + "regexp" "sync" "code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/db" @@ -36,31 +38,46 @@ const ( StatusFailed Status = "failed" ) -// Stored error messages the runner writes itself. +// Failure codes the runner stores itself. A kind cannot declare them. 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" + // CodeInterrupted is the error of a job a previous process left running. + CodeInterrupted = "interrupted" + // 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" ) +// 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 whose Message is safe to show the household and -// is stored as the job's error. Every other handler error is stored as -// MessageInternal. +// 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 { - Message string + Code string } -func (f *Failure) Error() string { return f.Message } +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, R any] func(ctx context.Context, payload P, publish func(partial R) error) (R, error) +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. @@ -73,8 +90,10 @@ type Job struct { 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) +// 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 { @@ -119,14 +138,25 @@ type Kind[P any] struct { } // 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) { +// 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 == CodeInterrupted || 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 { @@ -135,32 +165,49 @@ func Register[P, R any](r *Runner, kind string, h Handler[P, R]) (*Kind[P], erro 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) { + 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, fmt.Errorf("decode payload: %w", err) + return nil, CodeInternal, "payload does not decode" } result, err := h(ctx, p, func(partial R) error { - data, err := json.Marshal(partial) + data, err := publicJSON(partial) if err != nil { - return fmt.Errorf("jobs: encode partial result: %w", err) + return err } return publish(data) }) if err != nil { - return nil, err + 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 := json.Marshal(result) + data, err := publicJSON(result) if err != nil { - return nil, fmt.Errorf("encode result: %w", err) + return nil, CodeInternal, "result does not encode" } - return data, nil + 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) { @@ -187,7 +234,7 @@ func (k *Kind[P]) Enqueue(ctx context.Context, payload P) (string, error) { } // Start fails every job a previous process left running with -// MessageInterrupted, then starts the workers. They stop when ctx is cancelled; +// 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 { @@ -198,7 +245,7 @@ func (r *Runner) Start(ctx context.Context) error { } now := store.Now().String() n, err := db.New(r.st.Writer).FailRunningJobs(ctx, db.FailRunningJobsParams{ - Error: sql.NullString{String: MessageInterrupted, Valid: true}, + Error: sql.NullString{String: CodeInterrupted, Valid: true}, Now: sql.NullString{String: now, Valid: true}, }) if err != nil { @@ -270,8 +317,8 @@ func (r *Runner) run(ctx context.Context, job db.Job) { 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) + logger.ErrorContext(ctx, "job failed", "code", CodeUnknownKind, "reason", "no handler registered for the kind") + r.fail(write, logger, job.ID, CodeUnknownKind) return } @@ -290,20 +337,18 @@ func (r *Runner) run(ctx context.Context, job db.Job) { return nil } - result, err := h(ctx, job.Payload, publish) - if err != 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", "error", err) + logger.WarnContext(write, "job interrupted by shutdown; it fails at the next start") return } - var failure *Failure - message := MessageInternal - if errors.As(err, &failure) { - message = failure.Message - } else { - logger.ErrorContext(ctx, "job failed", "error", err) + level := slog.LevelInfo + if code == CodeInternal { + level = slog.LevelError } - r.fail(write, logger, job.ID, message) + logger.Log(ctx, level, "job failed", "code", code, "reason", reason) + r.fail(write, logger, job.ID, code) return } @@ -320,9 +365,11 @@ func (r *Runner) run(ctx context.Context, job db.Job) { } } -func (r *Runner) fail(ctx context.Context, logger *slog.Logger, id, message string) { +// 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: message, Valid: true}, + Error: sql.NullString{String: code, Valid: true}, Now: sql.NullString{String: store.Now().String(), Valid: true}, ID: id, }) diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index 37c5fdf..6ea3239 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -1,12 +1,15 @@ package jobs import ( + "bytes" "context" + "encoding/json" "errors" "fmt" "io" "log/slog" "strings" + "sync" "sync/atomic" "testing" "time" @@ -26,6 +29,8 @@ 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() @@ -46,13 +51,36 @@ func migratedStore(t *testing.T) *store.Store { 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) + 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()) { @@ -70,9 +98,9 @@ func start(t *testing.T, r *Runner) (stop func()) { return stop } -func register[P, R any](t *testing.T, r *Runner, kind string, h Handler[P, R]) *Kind[P] { +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) + k, err := Register(r, kind, h, failureCodes...) if err != nil { t.Fatal(err) } @@ -157,21 +185,43 @@ func TestJobSucceeds(t *testing.T) { } } -func TestJobFailsWithFailureMessage(t *testing.T) { +func TestJobFailsWithFailureCode(t *testing.T) { t.Parallel() - r := newRunner(t, migratedStore(t), 1) + 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{Message: "add some Regulars first"}) - }) + 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 != "add some Regulars first" { - t.Fatalf("job = %+v, want failed with the Failure message", job) + 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 @@ -180,44 +230,100 @@ 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 `json:"dinners"` - Key credential `json:"key"` - Token string `json:"-"` + Dinners []string + Key credential + Token string 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) { +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) - r := newRunner(t, st, 1) + var logs logBuffer + r := newRunnerLogging(t, st, 1, &logs) 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) { + 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 }) - failing := register(t, r, "leaky-error", func(context.Context, countPayload, func(leakyResult) error) (leakyResult, error) { + 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) }) - start(t, r) + 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) - 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) + 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}, } - 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) + 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) + } } - for _, id := range []string{typed.ID, failed.ID} { + 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) { @@ -285,8 +391,8 @@ func TestRestartFailsInterruptedJobsAndRunsQueuedOnes(t *testing.T) { 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 job.Status != StatusFailed || job.Error == nil || *job.Error != CodeInterrupted { + t.Errorf("interrupted job = %+v, want failed with %q", job, CodeInterrupted) } if string(job.Result) != `{"done":1}` { t.Errorf("interrupted job result = %s, want the last partial result kept", job.Result) @@ -349,7 +455,7 @@ func TestJobWithoutHandlerFails(t *testing.T) { 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" { + if job.Status != StatusFailed || job.Error == nil || *job.Error != CodeUnknownKind { t.Errorf("job = %+v, want failed for a missing handler", job) } } @@ -379,6 +485,11 @@ func TestRegisterRefusesBadRegistrations(t *testing.T) { 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") @@ -402,3 +513,18 @@ func TestNewRunnerRefusesBadConfig(t *testing.T) { 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) + } +} diff --git a/internal/server/jobs_test.go b/internal/server/jobs_test.go index 3994daf..6898242 100644 --- a/internal/server/jobs_test.go +++ b/internal/server/jobs_test.go @@ -20,6 +20,8 @@ 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. @@ -34,11 +36,11 @@ func startedRunner(t *testing.T, release, fail <-chan struct{}) (*jobs.Runner, * case <-release: return draftResult{Days: 7}, nil case <-fail: - return draftResult{}, &jobs.Failure{Message: "no Regulars to plan from"} + return draftResult{}, &jobs.Failure{Code: "no_regulars"} case <-ctx.Done(): return draftResult{}, ctx.Err() } - }) + }, "no_regulars") if err != nil { t.Fatal(err) } @@ -117,8 +119,8 @@ func TestGetJobShowsFailure(t *testing.T) { } 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"]) + if body["error"] != "no_regulars" { + t.Errorf("error = %v, want the Failure code", body["error"]) } }