diff --git a/internal/jobs/jobs.go b/internal/jobs/jobs.go index 12cb947..b98e61a 100644 --- a/internal/jobs/jobs.go +++ b/internal/jobs/jobs.go @@ -88,6 +88,10 @@ var ErrNotFound = errors.New("jobs: job not found") // registered handler. 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 // not decode into the kind's payload type. var ErrInvalidPayload = errors.New("jobs: payload does not decode") @@ -130,11 +134,13 @@ 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. +// handler is a registered kind: run runs a job, check reports whether a +// payload decodes into the kind's payload type, and serviceOnly is set when +// Runner.Enqueue refuses the kind. type handler struct { - run runFunc - check func(payload []byte) bool + run runFunc + check func(payload []byte) bool + serviceOnly bool } // 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) } +// 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: // call it after committing the transaction a job was enqueued in. func (k *Kind[P]) Wake() { k.r.wakeOne() } // 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. +// ErrUnknownKind for a kind with no registered handler, ErrServiceOnly for +// a ServiceOnly kind and ErrInvalidPayload for a payload that does not +// 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) { r.mu.Lock() h, ok := r.handlers[kind] @@ -302,6 +323,9 @@ func (r *Runner) Enqueue(ctx context.Context, kind string, payload json.RawMessa if !ok { return "", ErrUnknownKind } + if h.serviceOnly { + return "", ErrServiceOnly + } if !h.check(payload) { return "", ErrInvalidPayload } diff --git a/internal/jobs/jobs_test.go b/internal/jobs/jobs_test.go index ece6b13..ee7e152 100644 --- a/internal/jobs/jobs_test.go +++ b/internal/jobs/jobs_test.go @@ -736,3 +736,39 @@ func TestRunnerEnqueueByKindName(t *testing.T) { 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") + } +} diff --git a/internal/plan/plan.go b/internal/plan/plan.go index 0b8d231..e88d74c 100644 --- a/internal/plan/plan.go +++ b/internal/plan/plan.go @@ -129,7 +129,8 @@ type Service struct { // NewService returns a service over st that drafts with draft and // 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) { switch { case st == nil: @@ -146,6 +147,9 @@ func NewService(st *store.Store, draft Drafter, runner *jobs.Runner, logger *slo if err != nil { return nil, err } + if err := kind.ServiceOnly(); err != nil { + return nil, err + } s.draftJob = kind return s, nil } diff --git a/internal/server/jobs.go b/internal/server/jobs.go index 9f33d16..5ca5fba 100644 --- a/internal/server/jobs.go +++ b/internal/server/jobs.go @@ -30,7 +30,8 @@ type jobCreated struct { } // 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 { return func(w http.ResponseWriter, r *http.Request) { 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): writeError(w, r, logger, http.StatusBadRequest, "unknown_kind", "unknown job kind") 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): writeError(w, r, logger, http.StatusBadRequest, "invalid_payload", "payload does not match the job kind") return diff --git a/internal/server/plans_test.go b/internal/server/plans_test.go index 522ef65..86c1a18 100644 --- a/internal/server/plans_test.go +++ b/internal/server/plans_test.go @@ -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) { t.Parallel() e := newPlansRouter(t)