Refuse draft_week jobs on the generic jobs endpoint
CI / ci (push) Successful in 3m28s
CI / ci (pull_request) Successful in 3m10s

A kind can now be made service-only: Runner.Enqueue, and so
POST /api/v1/jobs, refuses it with 400 service_only_kind, and only its
typed Kind enqueues it. draft_week is service-only, so StartDraft and
its one-draft-in-flight-per-week rule is the only way to start a draft;
a confirmed draft_week posted to /api/v1/jobs can no longer run beside
the week's unconfirmed draft.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-07 06:57:53 -04:00
co-authored by Claude Opus 5.5
parent 45fcd71d27
commit 1f548c9f1d
5 changed files with 103 additions and 9 deletions
+31 -7
View File
@@ -88,6 +88,10 @@ var ErrNotFound = errors.New("jobs: job not found")
// registered handler. // registered handler.
var ErrUnknownKind = errors.New("jobs: unknown job kind") var ErrUnknownKind = errors.New("jobs: unknown job kind")
// ErrServiceOnly is returned by Runner.Enqueue for a kind made ServiceOnly,
// whose jobs only its own service enqueues.
var ErrServiceOnly = errors.New("jobs: job kind is enqueued only by its service")
// ErrInvalidPayload is returned by Runner.Enqueue for a payload that does // ErrInvalidPayload is returned by Runner.Enqueue for a payload that does
// not decode into the kind's payload type. // not decode into the kind's payload type.
var ErrInvalidPayload = errors.New("jobs: payload does not decode") var ErrInvalidPayload = errors.New("jobs: payload does not decode")
@@ -130,11 +134,13 @@ type Job struct {
// never quotes an error. // never quotes an error.
type runFunc func(ctx context.Context, payload string, publish func(json.RawMessage) error) (result json.RawMessage, code, reason string) 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 // handler is a registered kind: run runs a job, check reports whether a
// a payload decodes into the kind's payload type. // payload decodes into the kind's payload type, and serviceOnly is set when
// Runner.Enqueue refuses the kind.
type handler struct { type handler struct {
run runFunc run runFunc
check func(payload []byte) bool check func(payload []byte) bool
serviceOnly bool
} }
// Runner owns the job handlers and the worker pool. // Runner owns the job handlers and the worker pool.
@@ -286,15 +292,30 @@ func (k *Kind[P]) EnqueueTx(ctx context.Context, tx *sql.Tx, payload P) (string,
return createJob(ctx, db.New(tx), k.name, data) return createJob(ctx, db.New(tx), k.name, data)
} }
// ServiceOnly makes Runner.Enqueue refuse the kind with ErrServiceOnly, so
// its jobs are enqueued only through k, by the service that checks what
// must hold before one starts. It must be called before Start.
func (k *Kind[P]) ServiceOnly() error {
k.r.mu.Lock()
defer k.r.mu.Unlock()
if k.r.started {
return fmt.Errorf("jobs: service-only %q: runner already started", k.name)
}
h := k.r.handlers[k.name]
h.serviceOnly = true
k.r.handlers[k.name] = h
return nil
}
// Wake tells an idle worker to look for a job, as Enqueue does itself: // Wake tells an idle worker to look for a job, as Enqueue does itself:
// call it after committing the transaction a job was enqueued in. // call it after committing the transaction a job was enqueued in.
func (k *Kind[P]) Wake() { k.r.wakeOne() } func (k *Kind[P]) Wake() { k.r.wakeOne() }
// Enqueue stores a queued job of kind with its JSON payload and returns its // 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 // id, as Kind.Enqueue does for a kind known at compile time. It returns
// ErrUnknownKind for a kind with no registered handler and // ErrUnknownKind for a kind with no registered handler, ErrServiceOnly for
// ErrInvalidPayload for a payload that does not decode into the kind's // a ServiceOnly kind and ErrInvalidPayload for a payload that does not
// payload type; neither stores a job. // decode into the kind's payload type; none stores a job.
func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessage) (string, error) { func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessage) (string, error) {
r.mu.Lock() r.mu.Lock()
h, ok := r.handlers[kind] h, ok := r.handlers[kind]
@@ -302,6 +323,9 @@ func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessa
if !ok { if !ok {
return "", ErrUnknownKind return "", ErrUnknownKind
} }
if h.serviceOnly {
return "", ErrServiceOnly
}
if !h.check(payload) { if !h.check(payload) {
return "", ErrInvalidPayload return "", ErrInvalidPayload
} }
+36
View File
@@ -736,3 +736,39 @@ func TestRunnerEnqueueByKindName(t *testing.T) {
t.Errorf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result) t.Errorf("job = %+v (result %s), want succeeded with {\"done\":3}", job, job.Result)
} }
} }
func TestServiceOnlyKind(t *testing.T) {
t.Parallel()
st := migratedStore(t)
r := newRunner(t, st, 1)
k := register(t, r, "count", func(_ context.Context, p countPayload, _ func(countResult) error) (countResult, error) {
return countResult{Done: p.Steps}, nil
})
if err := k.ServiceOnly(); err != nil {
t.Fatal(err)
}
ctx := context.Background()
if _, err := r.Enqueue(ctx, "count", json.RawMessage(`{"steps":3}`)); !errors.Is(err, ErrServiceOnly) {
t.Errorf("Runner.Enqueue error = %v, want ErrServiceOnly", 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 enqueue stored %d jobs, want 0", n)
}
id, err := k.Enqueue(ctx, countPayload{Steps: 3})
if err != nil {
t.Fatal(err)
}
start(t, r)
if job := waitFor(t, r, id, finished); job.Status != StatusSucceeded {
t.Errorf("job enqueued through its Kind = %+v, want succeeded", job)
}
if err := k.ServiceOnly(); err == nil {
t.Error("ServiceOnly after Start succeeded, want an error")
}
}
+5 -1
View File
@@ -129,7 +129,8 @@ type Service struct {
// NewService returns a service over st that drafts with draft and // NewService returns a service over st that drafts with draft and
// registers the KindDraftWeek job on runner, which must not have started // registers the KindDraftWeek job on runner, which must not have started
// yet. // yet. The kind is service-only: StartDraft, which keeps one draft in
// flight per week, is the only way to enqueue it.
func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, logger *slog.Logger) (*Service, error) { func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, logger *slog.Logger) (*Service, error) {
switch { switch {
case st == nil: case st == nil:
@@ -146,6 +147,9 @@ func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, logger *slo
if err != nil { if err != nil {
return nil, err return nil, err
} }
if err := kind.ServiceOnly(); err != nil {
return nil, err
}
s.draftJob = kind s.draftJob = kind
return s, nil return s, nil
} }
+5 -1
View File
@@ -30,7 +30,8 @@ type jobCreated struct {
} }
// postJob serves POST /api/v1/jobs: it enqueues a job of a registered kind // 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. // that is not service-only and returns 202 with its id, which
// GET /api/v1/jobs/{id} reports on.
func postJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc { func postJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
req, err := decodeJobRequest(http.MaxBytesReader(w, r.Body, maxJobBody)) req, err := decodeJobRequest(http.MaxBytesReader(w, r.Body, maxJobBody))
@@ -43,6 +44,9 @@ func postJob(logger *slog.Logger, runner *jobs.Runner) http.HandlerFunc {
case errors.Is(err, jobs.ErrUnknownKind): case errors.Is(err, jobs.ErrUnknownKind):
writeError(w, r, logger, http.StatusBadRequest, "unknown_kind", "unknown job kind") writeError(w, r, logger, http.StatusBadRequest, "unknown_kind", "unknown job kind")
return return
case errors.Is(err, jobs.ErrServiceOnly):
writeError(w, r, logger, http.StatusBadRequest, "service_only_kind", "job kind is started only by its own endpoint")
return
case errors.Is(err, jobs.ErrInvalidPayload): case errors.Is(err, jobs.ErrInvalidPayload):
writeError(w, r, logger, http.StatusBadRequest, "invalid_payload", "payload does not match the job kind") writeError(w, r, logger, http.StatusBadRequest, "invalid_payload", "payload does not match the job kind")
return return
+26
View File
@@ -204,6 +204,32 @@ func TestPostPlanDraft(t *testing.T) {
} }
} }
// TestDraftWeekNotEnqueuedThroughJobs checks POST /api/v1/jobs refuses a
// draft_week job, so a confirmed draft cannot run beside the week's
// unconfirmed one: POST /plans/{weekStart}/draft is the only way to start
// a draft.
func TestDraftWeekNotEnqueuedThroughJobs(t *testing.T) {
t.Parallel()
e := newPlansRouter(t)
rec, _ := callDinners(t, e.h, http.MethodPost, plansPath+"2026-10-12/draft", "{}")
if rec.Code != http.StatusAccepted {
t.Fatalf("POST draft = %d: %s", rec.Code, rec.Body)
}
rec, body := callDinners(t, e.h, http.MethodPost, "/api/v1/jobs",
`{"kind":"`+plan.KindDraftWeek+`","payload":{"week_start":"2026-10-12","confirm":true}}`)
if rec.Code != http.StatusBadRequest || errorCode(body) != "service_only_kind" {
t.Errorf("POST job %s = %d %v, want 400 service_only_kind", plan.KindDraftWeek, rec.Code, body)
}
var queued int
if err := e.st.Reader.QueryRowContext(t.Context(), `SELECT COUNT(*) FROM jobs WHERE kind = ?`, plan.KindDraftWeek).Scan(&queued); err != nil {
t.Fatal(err)
}
if queued != 1 {
t.Errorf("%d draft jobs, want only the one POST draft enqueued", queued)
}
}
func TestSimultaneousPostPlanDraftsShareOneJob(t *testing.T) { func TestSimultaneousPostPlanDraftsShareOneJob(t *testing.T) {
t.Parallel() t.Parallel()
e := newPlansRouter(t) e := newPlansRouter(t)