From 45fcd71d27fa8512b6e8a07e976523ccb2df03f4 Mon Sep 17 00:00:00 2001 From: dev_agent Date: Wed, 7 Oct 2026 06:46:13 -0400 Subject: [PATCH] Keep one draft in flight per week A confirmed draft asked for while the week's draft job was asked for without confirm no longer enqueues a second job. With two job workers the two drafts ran at once, and whichever stored second failed plan_changed, so the household's confirmed draft could fail. StartDraft now refuses it with ErrDraftRunning, 409 draft_running, and enqueues nothing; the household asks again once the running draft finishes. Every other draft joins the week's queued or running job, so a week has at most one. The plan tests run the job runner with two workers, as cmd/server does, and a confirmed draft during a running unconfirmed one is refused while only the running job drafts. Maestro task ea48e650-d510-49cc-8485-0f07174ee731. Co-Authored-By: Claude Opus 5.5 --- docs/ARCHITECTURE.md | 5 +- internal/db/plans.sql.go | 29 ++++---- internal/db/querier.go | 9 +-- internal/plan/plan.go | 34 +++++---- internal/plan/plan_test.go | 130 +++++++++++++++++----------------- internal/server/plans.go | 12 ++-- internal/server/plans_test.go | 18 ++--- migrations/queries/plans.sql | 13 ++-- 8 files changed, 133 insertions(+), 117 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 985f135..f2cb34d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -345,8 +345,9 @@ automatic** (D-021). crock-pot dinner with known minutes, else null. `POST /api/v1/plans/{weekStart}/draft` takes `{confirm?}` and answers 202 `{jobId}` for a `draft_week` job, or for the one already queued or - running for the week — a confirmed one when the draft is confirmed, so a - confirmed draft never joins an unconfirmed job. The job runs the `draft_week` skill on every Regular + running for the week, so a week has at most one draft in flight. A + confirmed draft while the week's job was asked for without confirm is + 409 `draft_running` and enqueues nothing. The job runs the `draft_week` skill on every Regular and every Quest log dinner no other plan for the current or a later week has, with the household context and last week's dinners, and stores the week: a new plan is `draft`; a redraft keeps each day's diff --git a/internal/db/plans.sql.go b/internal/db/plans.sql.go index 71fb67a..bf97ec5 100644 --- a/internal/db/plans.sql.go +++ b/internal/db/plans.sql.go @@ -34,30 +34,35 @@ func (q *Queries) CreateWeekPlan(ctx context.Context, arg CreateWeekPlanParams) return err } -const getActiveDraftJobID = `-- name: GetActiveDraftJobID :one -SELECT id FROM jobs +const getActiveDraftJob = `-- name: GetActiveDraftJob :one +SELECT id, CAST(json_extract(payload, '$.confirm') = 1 AS INTEGER) AS confirmed +FROM jobs WHERE kind = ?1 AND status IN ('queued', 'running') AND json_extract(payload, '$.week_start') = ?2 - AND (CAST(?3 AS INTEGER) = 0 OR json_extract(payload, '$.confirm') = 1) ORDER BY id LIMIT 1 ` -type GetActiveDraftJobIDParams struct { +type GetActiveDraftJobParams struct { Kind string WeekStart string +} + +type GetActiveDraftJobRow struct { + ID string Confirmed int64 } -// The id of the job of kind queued or running for the week starting -// week_start, one whose payload has confirm when confirmed is 1. Returns -// sql.ErrNoRows when there is none. -func (q *Queries) GetActiveDraftJobID(ctx context.Context, arg GetActiveDraftJobIDParams) (string, error) { - row := q.db.QueryRowContext(ctx, getActiveDraftJobID, arg.Kind, arg.WeekStart, arg.Confirmed) - var id string - err := row.Scan(&id) - return id, err +// The job of kind queued or running for the week starting week_start, +// with whether its payload has confirm. StartDraft enqueues a draft only +// when there is none, so there is at most one. Returns sql.ErrNoRows when +// there is none. +func (q *Queries) GetActiveDraftJob(ctx context.Context, arg GetActiveDraftJobParams) (GetActiveDraftJobRow, error) { + row := q.db.QueryRowContext(ctx, getActiveDraftJob, arg.Kind, arg.WeekStart) + var i GetActiveDraftJobRow + err := row.Scan(&i.ID, &i.Confirmed) + return i, err } const getWeekPlan = `-- name: GetWeekPlan :one diff --git a/internal/db/querier.go b/internal/db/querier.go index b5abfe1..55b84c4 100644 --- a/internal/db/querier.go +++ b/internal/db/querier.go @@ -68,10 +68,11 @@ type Querier interface { FinishSuggestionBatch(ctx context.Context, arg FinishSuggestionBatchParams) (int64, error) // Returns sql.ErrNoRows until the row exists. GetAISettings(ctx context.Context) (AiSetting, error) - // The id of the job of kind queued or running for the week starting - // week_start, one whose payload has confirm when confirmed is 1. Returns - // sql.ErrNoRows when there is none. - GetActiveDraftJobID(ctx context.Context, arg GetActiveDraftJobIDParams) (string, error) + // The job of kind queued or running for the week starting week_start, + // with whether its payload has confirm. StartDraft enqueues a draft only + // when there is none, so there is at most one. Returns sql.ErrNoRows when + // there is none. + GetActiveDraftJob(ctx context.Context, arg GetActiveDraftJobParams) (GetActiveDraftJobRow, error) // The batch still pending or running, oldest first. Returns sql.ErrNoRows // when there is none. GetActiveSuggestionBatch(ctx context.Context) (SuggestionBatch, error) diff --git a/internal/plan/plan.go b/internal/plan/plan.go index d0a154c..0b8d231 100644 --- a/internal/plan/plan.go +++ b/internal/plan/plan.go @@ -71,6 +71,10 @@ var ErrDinnerNotFound = errors.New("dinner not found") // without confirm. var ErrConfirmRequired = errors.New("the plan was pushed to Calendar; drafting over it needs confirm") +// ErrDraftRunning is a confirmed draft asked for while a draft of the week +// asked for without confirm is queued or running. +var ErrDraftRunning = errors.New("a draft of the week without confirm is already running; ask again when it finishes") + // InvalidWeekError is a week start that is not a Monday date. type InvalidWeekError struct { Value string @@ -275,10 +279,11 @@ type Draft struct { // StartDraft enqueues the KindDraftWeek job drafting the week starting // weekStart, unless one is already queued or running for it, which it -// returns instead with Created false. A confirmed draft joins only a -// confirmed job, so a job asked for without confirm cannot fail it with -// ReasonConfirmRequired. Drafting over a synced or changed plan needs -// confirm, else it fails with ErrConfirmRequired and enqueues nothing. +// returns instead with Created false, so a week has at most one draft in +// flight. A confirmed draft joins only a confirmed job: while the week's +// job was asked for without confirm, it fails with ErrDraftRunning and +// enqueues nothing. Drafting over a synced or changed plan needs confirm, +// else it fails with ErrConfirmRequired and enqueues nothing. func (s *Service) StartDraft(ctx context.Context, weekStart store.LocalDate, confirm bool) (Draft, error) { if _, err := household.WeekDays(weekStart); err != nil { return Draft{}, &InvalidWeekError{Value: weekStart.String()} @@ -294,19 +299,18 @@ func (s *Service) StartDraft(ctx context.Context, weekStart store.LocalDate, con case cur.Status != StatusDraft && !confirm: return ErrConfirmRequired } - confirmed := int64(0) - if confirm { - confirmed = 1 - } - id, err := q.GetActiveDraftJobID(ctx, db.GetActiveDraftJobIDParams{Kind: KindDraftWeek, WeekStart: weekStart.String(), Confirmed: confirmed}) - if err == nil { - d = Draft{JobID: id} + active, err := q.GetActiveDraftJob(ctx, db.GetActiveDraftJobParams{Kind: KindDraftWeek, WeekStart: weekStart.String()}) + switch { + case errors.Is(err, sql.ErrNoRows): + case err != nil: + return fmt.Errorf("plan: read draft job of %s: %w", weekStart, err) + case confirm && active.Confirmed != 1: + return ErrDraftRunning + default: + d = Draft{JobID: active.ID} return nil } - if !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("plan: read draft job of %s: %w", weekStart, err) - } - id, err = s.draftJob.EnqueueTx(ctx, tx, DraftJob{WeekStart: weekStart.String(), Confirm: confirm}) + id, err := s.draftJob.EnqueueTx(ctx, tx, DraftJob{WeekStart: weekStart.String(), Confirm: confirm}) if err != nil { return fmt.Errorf("plan: enqueue draft of %s: %w", weekStart, err) } diff --git a/internal/plan/plan_test.go b/internal/plan/plan_test.go index c29641f..a437bb6 100644 --- a/internal/plan/plan_test.go +++ b/internal/plan/plan_test.go @@ -53,9 +53,9 @@ type env struct { // newEnv returns a plan service over a migrated store in zone // America/New_York with dinner time 17:30, the Regulars below and Quest -// log dinners Pad Thai and Shawarma Bowls, and an unstarted job runner. It -// drafts through the real harness and draft_week with the scripted -// e.model. +// log dinners Pad Thai and Shawarma Bowls, and an unstarted job runner +// with two workers. It drafts through the real harness and draft_week with +// the scripted e.model. func newEnv(t *testing.T) *env { t.Helper() ctx := t.Context() @@ -106,7 +106,8 @@ func newEnv(t *testing.T) *env { return skills.DraftWeek(ctx, cfg, scope, in) } logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - e.runner, err = jobs.NewRunner(st, logger, 1) + // Two workers, as cmd/server runs, so drafts of a week could overlap. + e.runner, err = jobs.NewRunner(st, logger, 2) if err != nil { t.Fatal(err) } @@ -518,100 +519,97 @@ func TestStartDraftIsSingleFlight(t *testing.T) { } } -func TestSimultaneousStartDraftsShareOneJobPerConfirm(t *testing.T) { - e := newEnv(t) - first, err := e.svc.StartDraft(t.Context(), week, false) - if err != nil { - t.Fatal(err) - } - // Twenty simultaneous drafts, half of them confirmed: the unconfirmed - // ones join the queued job, and the confirmed ones share one confirmed - // job of their own. - const n = 20 - drafts := make([]plan.Draft, n) - errs := make([]error, n) - var wg sync.WaitGroup - for i := range n { - wg.Go(func() { - drafts[i], errs[i] = e.svc.StartDraft(context.Background(), week, i%2 == 1) - }) - } - wg.Wait() - confirmedJob := "" - created := 0 - for i, d := range drafts { - if errs[i] != nil { - t.Fatalf("draft %d: %v", i, errs[i]) - } - if d.Created { - created++ - } - if i%2 == 0 { - if d.JobID != first.JobID || d.Created { - t.Errorf("unconfirmed draft %d = %+v, want it to join %s", i, d, first.JobID) +func TestSimultaneousStartDraftsShareOneJob(t *testing.T) { + for _, confirmFirst := range []bool{false, true} { + t.Run(fmt.Sprintf("first confirm %v", confirmFirst), func(t *testing.T) { + e := newEnv(t) + first, err := e.svc.StartDraft(t.Context(), week, confirmFirst) + if err != nil { + t.Fatal(err) } - continue - } - if confirmedJob == "" { - confirmedJob = d.JobID - } - if d.JobID != confirmedJob || d.JobID == first.JobID { - t.Errorf("confirmed draft %d = %+v, want the one confirmed job %s, not %s", i, d, confirmedJob, first.JobID) - } - } - if created != 1 { - t.Errorf("%d drafts enqueued a job, want 1", created) - } - var payload string - if err := e.st.Reader.QueryRowContext(t.Context(), `SELECT payload FROM jobs WHERE id = ?`, confirmedJob).Scan(&payload); err != nil { - t.Fatal(err) - } - if !strings.Contains(payload, `"confirm":true`) { - t.Errorf("confirmed job payload = %s, want confirm", payload) + // Twenty simultaneous drafts, half of them confirmed: every + // one joins the queued job, except a confirmed one while the + // job is unconfirmed, which is refused. + const n = 20 + drafts := make([]plan.Draft, n) + errs := make([]error, n) + var wg sync.WaitGroup + for i := range n { + wg.Go(func() { + drafts[i], errs[i] = e.svc.StartDraft(context.Background(), week, i%2 == 1) + }) + } + wg.Wait() + for i, d := range drafts { + if i%2 == 1 && !confirmFirst { + if !errors.Is(errs[i], plan.ErrDraftRunning) { + t.Errorf("confirmed draft %d = %+v, %v, want ErrDraftRunning", i, d, errs[i]) + } + continue + } + if errs[i] != nil || d.JobID != first.JobID || d.Created { + t.Errorf("draft %d = %+v, %v, want it to join %s", i, d, errs[i], first.JobID) + } + } + 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 1", queued) + } + }) } } -func TestConfirmedDraftDoesNotJoinUnconfirmedJob(t *testing.T) { +func TestConfirmedDraftWhileUnconfirmedDraftRuns(t *testing.T) { e := newEnv(t) e.start(t) e.reply("first", fullWeek...) e.mustDraft(t, false) // An unconfirmed redraft is running when the plan is synced; the - // household then confirms a redraft, which must not join the job that - // will fail confirm_required. + // household then confirms a redraft. With a second worker idle, a + // second job would draft alongside the first, so the confirmed draft + // is refused and only the first job runs draft_week. reversed := slices.Clone(fullWeek) slices.Reverse(reversed) started, release := make(chan struct{}), make(chan struct{}) r := e.weekReply("unconfirmed", fullWeek...) r.Started, r.Release = started, release e.model.QueueGenerate(r) - e.reply("confirmed", reversed...) + e.reply("concurrent", reversed...) unconfirmed, err := e.svc.StartDraft(t.Context(), week, false) if err != nil { t.Fatal(err) } <-started e.setStatus(t, plan.StatusSynced) - confirmed, err := e.svc.StartDraft(t.Context(), week, true) - if err != nil { - t.Fatal(err) - } - if !confirmed.Created || confirmed.JobID == unconfirmed.JobID { - t.Fatalf("confirmed draft = %+v, want its own job, not %s", confirmed, unconfirmed.JobID) + if d, err := e.svc.StartDraft(t.Context(), week, true); !errors.Is(err, plan.ErrDraftRunning) { + t.Fatalf("confirmed draft = %+v, %v, want ErrDraftRunning", d, err) } close(release) if job := e.waitJob(t, unconfirmed.JobID); job.Status != jobs.StatusFailed || *job.Error != plan.ReasonConfirmRequired { t.Errorf("unconfirmed job %s %v, want failed %s", job.Status, job.Error, plan.ReasonConfirmRequired) } - if job := e.waitJob(t, confirmed.JobID); job.Status != jobs.StatusSucceeded { - t.Fatalf("confirmed job %s %v, want succeeded", job.Status, job.Error) + 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 != 2 { + t.Errorf("%d draft jobs, want the first and the unconfirmed redraft", queued) } p, err := e.svc.Get(t.Context(), week) if err != nil { t.Fatal(err) } - if p.Status != plan.StatusChanged || *p.AINote != "confirmed" || !slices.Equal(dayNames(p), reversed) { + if p.Status != plan.StatusSynced || *p.AINote != "first" || !slices.Equal(dayNames(p), fullWeek) { + t.Errorf("plan = %s %q %v, want the synced first draft", p.Status, *p.AINote, dayNames(p)) + } + + // Asked again once the unconfirmed job has finished, the confirmed + // draft replaces the plan. + if p := e.mustDraft(t, true); p.Status != plan.StatusChanged || *p.AINote != "concurrent" || !slices.Equal(dayNames(p), reversed) { t.Errorf("plan = %s %q %v, want the confirmed draft, changed", p.Status, *p.AINote, dayNames(p)) } } diff --git a/internal/server/plans.go b/internal/server/plans.go index e6ed165..949f1a5 100644 --- a/internal/server/plans.go +++ b/internal/server/plans.go @@ -44,8 +44,9 @@ func getPlan(logger *slog.Logger, svc *plan.Service) http.HandlerFunc { // postPlanDraft enqueues the draft of the week and answers 202 with the // jobId that GET /jobs/{id} reports on, without waiting for the job. When -// a draft of the week is already queued or running, confirmed if this one -// is, it answers 202 with that job instead. +// a draft of the week is already queued or running, it answers 202 with +// that job instead, or 409 when this draft is confirmed and that one is +// not. func postPlanDraft(logger *slog.Logger, svc *plan.Service) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { week, ok := weekStartOrFail(w, r, logger) @@ -105,8 +106,9 @@ func weekStartOrFail(w http.ResponseWriter, r *http.Request, logger *slog.Logger // writePlanError maps a plan.Service error to its API error: a week with // no plan or an unknown dinner is 404, a date outside the week is 400, and -// drafting over a synced plan without confirm, or planning a dinner that -// is neither a Regular nor in the Quest log, is 409. +// drafting over a synced plan without confirm, a confirmed draft while an +// unconfirmed one runs, or planning a dinner that is neither a Regular nor +// in the Quest log, is 409. func writePlanError(w http.ResponseWriter, r *http.Request, logger *slog.Logger, action string, err error) { var day *plan.InvalidDayError var week *plan.InvalidWeekError @@ -120,6 +122,8 @@ func writePlanError(w http.ResponseWriter, r *http.Request, logger *slog.Logger, writeError(w, r, logger, http.StatusConflict, "dinner_not_plannable", notPlannable.Error()) case errors.Is(err, plan.ErrConfirmRequired): writeError(w, r, logger, http.StatusConflict, "confirm_required", err.Error()) + case errors.Is(err, plan.ErrDraftRunning): + writeError(w, r, logger, http.StatusConflict, "draft_running", err.Error()) case errors.Is(err, plan.ErrNotFound): writeError(w, r, logger, http.StatusNotFound, "not_found", "plan not found") case errors.Is(err, plan.ErrDinnerNotFound): diff --git a/internal/server/plans_test.go b/internal/server/plans_test.go index 7b314a1..522ef65 100644 --- a/internal/server/plans_test.go +++ b/internal/server/plans_test.go @@ -184,16 +184,18 @@ func TestPostPlanDraft(t *testing.T) { t.Errorf("POST draft of a new week = %d %v, want 202 with its own job", rec.Code, body) } - // A confirmed draft does not join the unconfirmed one, but a second - // confirmed draft joins the first. + // A confirmed draft does not join the unconfirmed one and enqueues + // nothing. rec, body = callDinners(t, e.h, http.MethodPost, plansPath+"2026-10-12/draft", `{"confirm":true}`) - confirmedID, _ := body["jobId"].(string) - if rec.Code != http.StatusAccepted || confirmedID == "" || confirmedID == jobID { - t.Errorf("confirmed POST draft = %d %v, want 202 with its own job", rec.Code, body) + if rec.Code != http.StatusConflict || errorCode(body) != "draft_running" { + t.Errorf("confirmed POST draft = %d %v, want 409 draft_running", rec.Code, body) } - rec, body = callDinners(t, e.h, http.MethodPost, plansPath+"2026-10-12/draft", `{"confirm":true}`) - if rec.Code != http.StatusAccepted || body["jobId"] != confirmedID { - t.Errorf("second confirmed POST draft = %d %v, want 202 with job %s", rec.Code, body, confirmedID) + 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 != 2 { + t.Errorf("%d draft jobs, want one per week", queued) } rec, body = callDinners(t, e.h, http.MethodPost, plansPath+"2026-10-12/draft", `{"confirm":"yes"}`) diff --git a/migrations/queries/plans.sql b/migrations/queries/plans.sql index ef18a9c..c64d7f0 100644 --- a/migrations/queries/plans.sql +++ b/migrations/queries/plans.sql @@ -72,14 +72,15 @@ JOIN week_plans ON week_plans.id = plan_days.plan_id WHERE week_plans.week_start = ? AND plan_days.dinner_id IS NOT NULL ORDER BY plan_days.date; --- name: GetActiveDraftJobID :one --- The id of the job of kind queued or running for the week starting --- week_start, one whose payload has confirm when confirmed is 1. Returns --- sql.ErrNoRows when there is none. -SELECT id FROM jobs +-- name: GetActiveDraftJob :one +-- The job of kind queued or running for the week starting week_start, +-- with whether its payload has confirm. StartDraft enqueues a draft only +-- when there is none, so there is at most one. Returns sql.ErrNoRows when +-- there is none. +SELECT id, CAST(json_extract(payload, '$.confirm') = 1 AS INTEGER) AS confirmed +FROM jobs WHERE kind = sqlc.arg(kind) AND status IN ('queued', 'running') AND json_extract(payload, '$.week_start') = sqlc.arg(week_start) - AND (CAST(sqlc.arg(confirmed) AS INTEGER) = 0 OR json_extract(payload, '$.confirm') = 1) ORDER BY id LIMIT 1;