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>

internal/jobs/jobs.go

@@ -68,14 +68,6 @@ 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")
 
-// ErrUnknownKind is returned by Runner.Enqueue for a kind with no
-// registered handler.
-var ErrUnknownKind = errors.New("jobs: unknown job kind")
-
-// ErrInvalidPayload is returned by Runner.Enqueue for a payload that does
-// not decode into the kind's payload type.
-var ErrInvalidPayload = errors.New("jobs: payload does not decode")
-
 // 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
@@ -114,13 +106,6 @@ type Job struct {
 // never quotes an error.
 type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (result json.RawMessage, code, reason string)
 
-// handler is a registered kind: run runs a job, and check reports whether
-// a payload decodes into the kind's payload type.
-type handler struct {
-	run   runFunc
-	check func(payload []byte) bool
-}
-
 // Runner owns the job handlers and the worker pool.
 type Runner struct {
 	st      *store.Store
@@ -133,7 +118,7 @@ type Runner struct {
 	claim func(ctx context.Context, now sql.NullString) (db.Job, error)
 
 	mu       sync.Mutex
-	handlers map[string]handler
+	handlers map[string]runFunc
 	started  bool
 
 	wg sync.WaitGroup
@@ -156,7 +141,7 @@ func NewRunner(st *store.Store, logger *slog.Logger, workers int) (*Runner, erro
 		workers:  workers,
 		wake:     make(chan struct{}, workers),
 		claim:    db.New(st.Writer).ClaimNextJob,
-		handlers: make(map[string]handler),
+		handlers: make(map[string]runFunc),
 	}, nil
 }
 
@@ -194,9 +179,11 @@ func Register[P any, R Public](r *Runner, kind string, h Handler[P, R], failureC
 	if _, ok := r.handlers[kind]; ok {
 		return nil, fmt.Errorf("jobs: register %q: kind already registered", kind)
 	}
-	run := func(ctx context.Context, payload string, publish func(json.RawMessage) error) (json.RawMessage, string, string) {
-		p, err := decodePayload[P]([]byte(payload))
-		if err != nil {
+	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 {
@@ -222,23 +209,9 @@ func Register[P any, R Public](r *Runner, kind string, h Handler[P, R], failureC
 		}
 		return data, "", ""
 	}
-	check := func(payload []byte) bool {
-		_, err := decodePayload[P](payload)
-		return err == nil
-	}
-	r.handlers[kind] = handler{run: run, check: check}
 	return &Kind[P]{r: r, name: kind}, nil
 }
 
-// decodePayload decodes payload into P, refusing unknown fields.
-func decodePayload[P any](payload []byte) (P, error) {
-	var p P
-	dec := json.NewDecoder(bytes.NewReader(payload))
-	dec.DisallowUnknownFields()
-	err := dec.Decode(&p)
-	return p, err
-}
-
 // 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) {
@@ -256,43 +229,19 @@ func (k *Kind[P]) Enqueue(ctx context.Context, payload P) (string, error) {
 	if err != nil {
 		return "", fmt.Errorf("jobs: enqueue %s: encode payload: %w", k.name, err)
 	}
-	return k.r.enqueue(ctx, k.name, data)
-}
-
-// Enqueue stores a queued job of kind with its JSON payload and returns its
-// id, as Kind.Enqueue does for a kind known at compile time. It returns
-// ErrUnknownKind for a kind with no registered handler and
-// ErrInvalidPayload for a payload that does not decode into the kind's
-// payload type; neither stores a job.
-func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessage) (string, error) {
-	r.mu.Lock()
-	h, ok := r.handlers[kind]
-	r.mu.Unlock()
-	if !ok {
-		return "", ErrUnknownKind
-	}
-	if !h.check(payload) {
-		return "", ErrInvalidPayload
-	}
-	return r.enqueue(ctx, kind, payload)
-}
-
-// enqueue stores a queued job of kind with its encoded payload and wakes a
-// worker.
-func (r *Runner) enqueue(ctx context.Context, kind string, payload []byte) (string, error) {
 	id := store.NewID()
 	now := store.Now().String()
-	if err := db.New(r.st.Writer).CreateJob(ctx, db.CreateJobParams{
+	if err := db.New(k.r.st.Writer).CreateJob(ctx, db.CreateJobParams{
 		ID:        id,
-		Kind:      kind,
-		Payload:   string(payload),
+		Kind:      k.name,
+		Payload:   string(data),
 		CreatedAt: now,
 		UpdatedAt: now,
 	}); err != nil {
-		return "", fmt.Errorf("jobs: enqueue %s: %w", kind, err)
+		return "", fmt.Errorf("jobs: enqueue %s: %w", k.name, err)
 	}
 	select {
-	case r.wake <- struct{}{}:
+	case k.r.wake <- struct{}{}:
 	default: // every worker already has a token and will look again
 	}
 	return id, nil
@@ -416,7 +365,7 @@ func (r *Runner) run(ctx context.Context, job db.Job) {
 		return nil
 	}
 
-	result, code, reason := h.run(ctx, job.Payload, publish)
+	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")

internal/jobs/jobs_test.go

@@ -568,39 +568,3 @@ func TestClaimErrorIsRetried(t *testing.T) {
 		t.Fatalf("claim error text logged: %s", out)
 	}
 }
-
-func TestRunnerEnqueueByKindName(t *testing.T) {
-	t.Parallel()
-	st := migratedStore(t)
-	r := newRunner(t, st, 1)
-	register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
-		return countResult{Done: p.Steps}, nil
-	})
-	ctx := context.Background()
-
-	if _, err := r.Enqueue(ctx, "no_such_kind", json.RawMessage(`{}`)); !errors.Is(err, ErrUnknownKind) {
-		t.Errorf("unknown kind error = %v, want ErrUnknownKind", err)
-	}
-	for _, payload := range []string{`{"steps":"three"}`, `{"other":1}`, `[]`, ``} {
-		if _, err := r.Enqueue(ctx, "count", json.RawMessage(payload)); !errors.Is(err, ErrInvalidPayload) {
-			t.Errorf("payload %q error = %v, want ErrInvalidPayload", payload, err)
-		}
-	}
-	var n int
-	if err := st.Reader.QueryRow("SELECT COUNT(*) FROM jobs").Scan(&n); err != nil {
-		t.Fatal(err)
-	}
-	if n != 0 {
-		t.Errorf("refused enqueues stored %d jobs, want 0", n)
-	}
-
-	id, err := r.Enqueue(ctx, "count", json.RawMessage(`{"steps":3}`))
-	if err != nil {
-		t.Fatal(err)
-	}
-	start(t, r)
-	job := waitFor(t, r, id, finished)
-	if job.Kind != "count" || job.Status != StatusSucceeded || string(job.Result) != `{"done":3}` {
-		t.Errorf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result)
-	}
-}

internal/server/jobs.go

@@ -1,11 +1,7 @@
 package server
 
 import (
-	"bytes"
-	"encoding/json"
 	"errors"
-	"fmt"
-	"io"
 	"log/slog"
 	"net/http"
 
@@ -14,80 +10,6 @@ import (
 	"code.cubecraftlabs.com/Maniacs_Mansion/Meal-Quest/internal/jobs"
 )
 
-// maxJobBody bounds a POST /jobs request body.
-const maxJobBody = 64 << 10
-
-// jobRequest is the POST /jobs body. Payload is required; a job kind that
-// takes no input is sent the empty object {}.
-type jobRequest struct {
-	Kind    string          `json:"kind"`
-	Payload json.RawMessage `json:"payload"`
-}
-
-// jobCreated is the POST /jobs response.
-type jobCreated struct {
-	ID string `json:"id"`
-}
-
-// postJob serves POST /api/v1/jobs: it enqueues a job of a registered kind
-// and returns 202 with its id, which GET /api/v1/jobs/{id} reports on.
-func postJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
-	return func(w http.ResponseWriter, r *http.Request) {
-		req, err := decodeJobRequest(http.MaxBytesReader(w, r.Body, maxJobBody))
-		if err != nil {
-			writeError(w, r, logger, http.StatusBadRequest, "invalid_request", err.Error())
-			return
-		}
-		id, err := runner.Enqueue(r.Context(), req.Kind, req.Payload)
-		switch {
-		case errors.Is(err, jobs.ErrUnknownKind):
-			writeError(w, r, logger, http.StatusBadRequest, "unknown_kind", "unknown job kind")
-			return
-		case errors.Is(err, jobs.ErrInvalidPayload):
-			writeError(w, r, logger, http.StatusBadRequest, "invalid_payload", "payload does not match the job kind")
-			return
-		case err != nil:
-			logger.ErrorContext(r.Context(), "enqueue job", "kind", req.Kind, "error", err)
-			writeError(w, r, logger, http.StatusInternalServerError, "internal", "internal error")
-			return
-		}
-		writeJSON(w, r, logger, http.StatusAccepted, jobCreated{ID: id})
-	}
-}
-
-// decodeJobRequest reads a single JSON object with a non-empty kind and a
-// payload, refusing unknown fields, an omitted or null payload and trailing
-// data.
-func decodeJobRequest(body io.Reader) (jobRequest, error) {
-	data, err := io.ReadAll(body)
-	if err != nil {
-		var tooLarge *http.MaxBytesError
-		if errors.As(err, &tooLarge) {
-			return jobRequest{}, fmt.Errorf("request body exceeds %d bytes", tooLarge.Limit)
-		}
-		return jobRequest{}, errors.New("cannot read request body")
-	}
-	var req jobRequest
-	dec := json.NewDecoder(bytes.NewReader(data))
-	dec.DisallowUnknownFields()
-	if err := dec.Decode(&req); err != nil {
-		return jobRequest{}, errors.New("request body must be a JSON object with kind and payload")
-	}
-	if _, err := dec.Token(); err != io.EOF {
-		return jobRequest{}, errors.New("request body must be a single JSON object")
-	}
-	if req.Kind == "" {
-		return jobRequest{}, errors.New("kind: required")
-	}
-	if req.Payload == nil {
-		return jobRequest{}, errors.New("payload: required")
-	}
-	if bytes.Equal(req.Payload, []byte("null")) {
-		return jobRequest{}, errors.New("payload: must not be null")
-	}
-	return req, nil
-}
-
 // 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 {

internal/server/jobs_test.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"slices"
-	"strings"
 	"testing"
 	"time"
 
@@ -152,95 +151,3 @@ func TestGetJobRequiresSession(t *testing.T) {
 		t.Errorf("status = %d, want 401", rec.Code)
 	}
 }
-
-// TestPostJobThenGetProgressAndResult drives the whole flow over the
-// router: POST enqueues a progressive "draft" job, GET reads its partial
-// result while the handler is held mid-run, then its final result.
-func TestPostJobThenGetProgressAndResult(t *testing.T) {
-	t.Parallel()
-	release := make(chan struct{})
-	runner, _ := startedRunner(t, release, nil)
-	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
-
-	rec, body := serve(t, router, signedIn(http.MethodPost, "/api/v1/jobs", strings.NewReader(`{"kind":"draft","payload":{}}`)))
-	if rec.Code != http.StatusAccepted {
-		t.Fatalf("status = %d, want 202: %s", rec.Code, rec.Body)
-	}
-	id, _ := body["id"].(string)
-	if keys := slices.Sorted(maps.Keys(body)); !slices.Equal(keys, []string{"id"}) || id == "" {
-		t.Fatalf("body = %v, want only a job id", body)
-	}
-
-	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 done["id"] != id || resultDays(done) != 7 || done["error"] != nil {
-		t.Errorf("finished body = %v, want result {\"days\":7} and no error", done)
-	}
-}
-
-func TestPostJobRefusesBadRequests(t *testing.T) {
-	t.Parallel()
-	runner, _ := startedRunner(t, nil, nil)
-	router := newTestRouterWith(t, "dev", newTestSettings(t), runner)
-
-	for _, tc := range []struct {
-		name, body, code, message string
-	}{
-		{"unknown kind", `{"kind":"no_such_kind","payload":{}}`, "unknown_kind", "unknown job kind"},
-		{"payload for another type", `{"kind":"draft","payload":{"days":3}}`, "invalid_payload", "payload does not match the job kind"},
-		{"payload not an object", `{"kind":"draft","payload":[1]}`, "invalid_payload", "payload does not match the job kind"},
-		{"missing payload", `{"kind":"draft"}`, "invalid_request", "payload: required"},
-		{"null payload", `{"kind":"draft","payload":null}`, "invalid_request", "payload: must not be null"},
-		{"missing kind", `{"payload":{}}`, "invalid_request", "kind: required"},
-		{"empty body", ``, "invalid_request", "request body must be a JSON object with kind and payload"},
-		{"not an object", `["draft"]`, "invalid_request", "request body must be a JSON object with kind and payload"},
-		{"unknown field", `{"kind":"draft","payload":{},"extra":1}`, "invalid_request", "request body must be a JSON object with kind and payload"},
-		{"trailing data", `{"kind":"draft","payload":{}} {}`, "invalid_request", "request body must be a single JSON object"},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			t.Parallel()
-			rec := httptest.NewRecorder()
-			router.ServeHTTP(rec, signedIn(http.MethodPost, "/api/v1/jobs", strings.NewReader(tc.body)))
-			if rec.Code != http.StatusBadRequest {
-				t.Fatalf("status = %d, want 400: %s", rec.Code, rec.Body)
-			}
-			var body errorBody
-			if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
-				t.Fatal(err)
-			}
-			if want := (errorBody{Error: apiError{Code: tc.code, Message: tc.message}}); body != want {
-				t.Errorf("body = %+v, want %+v", body, want)
-			}
-		})
-	}
-}
-
-func TestPostJobNeedsSessionAndCSRF(t *testing.T) {
-	t.Parallel()
-	router := newTestRouter(t, "dev")
-	const body = `{"kind":"noop","payload":{}}`
-
-	rec := httptest.NewRecorder()
-	router.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(body)))
-	if rec.Code != http.StatusUnauthorized || rec.Body.String() != unauthenticatedJSON {
-		t.Errorf("no session: %d %s, want 401 %s", rec.Code, rec.Body, unauthenticatedJSON)
-	}
-
-	for _, header := range []string{"", "wrong-token"} {
-		req := httptest.NewRequest(http.MethodPost, "/api/v1/jobs", strings.NewReader(body))
-		req.AddCookie(validSessionCookie)
-		if header != "" {
-			req.Header.Set(CSRFHeader, header)
-		}
-		rec := httptest.NewRecorder()
-		router.ServeHTTP(rec, req)
-		if rec.Code != http.StatusForbidden || rec.Body.String() != csrfJSON {
-			t.Errorf("CSRF header %q: %d %s, want 403 %s", header, rec.Code, rec.Body, csrfJSON)
-		}
-	}
-}

internal/server/server.go

@@ -30,7 +30,7 @@ type RouteMounter interface {
 // 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; runner serves
-// POST /api/v1/jobs and GET /api/v1/jobs/{id}; assets is the web build
+// /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, runner *jobs.Runner, assets fs.FS) (http.Handler, error) {
@@ -68,7 +68,6 @@ 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.Post("/jobs", postJob(logger, runner))
 	apiV1.Get("/jobs/{id}", getJob(logger, runner))
 	r.Route("/api", func(r chi.Router) {
 		r.MethodNotAllowed(methodNotAllowed)