Compare commits

...
Author SHA1 Message Date
dev_agentandClaude Opus 5.5 0942e65392 Replace a held Side quest made ineligible before a batch's final store
CI / ci (push) Successful in 9m38s
CI / ci (pull_request) Successful in 8m21s
A notifying (scheduled or manual) suggestion batch held each round's
grounded Side quests and only checked them at its single final store, so
a household change during grounding could drop a held card after the
fill loop had stopped, leaving the batch short of BatchSize with spare
candidates unused (MQ-172-3). The batch now rechecks every held Side
quest after each round with the same eligibility check the store uses
(eligible, extracted from store), so a later round grounds a spare in
its place. The single-transaction store and once-only new_suggestions
are unchanged.

Narrow ARCHITECTURE.md's "a failed scheduled or manual batch has stored
nothing" to failures before the final store, and state that a job that
fails recording the batch's end keeps its Side quests and
new_suggestions (MQ-172-4).

Maestro task 244f9834-faa8-4bb8-b5e6-43677dec0cc0

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-11 08:53:58 -04:00
orchestrator-agent 791b321b71 Expose the Supply run title job and cover the replacement race (#194)
CI / ci (push) Successful in 8m9s
Co-authored-by: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com>
2026-10-11 08:24:51 -04:00
orchestrator-agent be41b56690 Make login OAuth state single-use (#193)
CI / ci (push) Successful in 8m6s
Co-authored-by: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com>
2026-10-11 02:13:34 -04:00
11 changed files with 474 additions and 86 deletions
+20 -6
View File
@@ -319,15 +319,22 @@ automatic** (D-021).
suggestion with the name it is suggested under, the reason and
`cited_regular_ids` — unless, since the job read its input, a dinner
it repeats was added, a suggestion it repeats passed, or a Regular it
cites renamed or archived (checked in the storing transaction). The job
cites renamed or archived (checked in the storing transaction). A
notifying batch also rechecks the Side quests it holds after each round
and lets go of those no longer eligible, so a later round grounds a
spare in place of each; only a change committed between the last
recheck and the store can leave it short of 5 with spares unused. The job
publishes `{batch_id, stage, suggestion_ids, dropped}` as it goes (`suggesting`,
`grounding`, `done`). It fails with `no_regulars`, `ai_failed` (the
skill or `normalize_recipe` failed; no Side quest is made up instead,
D-017), `search_not_configured` or `search_failed`, and the batch is set
`failed`. A failed `scheduled` or `manual` batch has stored nothing; a
`failed`. A `scheduled` or `manual` batch that fails before its final
store (the AI, grounding, or the store itself) has stored nothing; a
failed `replenish` batch's Side quests stored before the failure stay
in the deck. If
recording the batch's end fails, its job fails; the job runner retries
recording the batch's end fails, its job fails, even after the final
store committed: a notifying batch's Side quests and their
`new_suggestions` then stay; the job runner retries
a job's final status write with backoff until it lands, each attempt
bounded at 5 seconds; at shutdown an attempt that fails or times out is
not retried, and the next start fails the job as interrupted. Recording
@@ -1120,7 +1127,10 @@ Fake-clock tests of a schedule live in `internal/scheduler`, where
- The week's Supply run (Feature 17) is made from a pushed plan: after
a successful `Push`, the handler runs `shopping.GeneratePushed` for
the week at the plan revision `Push` pushed, and the 200 answer
carries `supply_run: {status, error}`. `status` is `generated`;
carries `supply_run: {status, error, title_job_id}`. `title_job_id` is
the `title_supply_run` job titling a generated list, as for `POST
/shopping/{weekStart}/generate`; null when the list already has its
title or was not generated. `status` is `generated`;
`archived` when the week's Supply run is in the Supply archive and was
left as it was; or `failed` — error `plan_changed` (retryable, logged
as a warning) when the plan was edited between the push and
@@ -1143,8 +1153,12 @@ Fake-clock tests of a schedule live in `internal/scheduler`, where
when the week has no Supply run, 400 `invalid_week`.
- `POST /api/v1/shopping/{weekStart}/generate` makes or remakes the list
through `shopping.Generate` and answers with the same shape plus
`skipped: [{date, dinner_id, dinner_name, reason}]`. 404 when the week
has no plan, 409 `archived` when its list is in the Supply archive.
`skipped: [{date, dinner_id, dinner_name, reason}]` and
`title_job_id`, the `title_supply_run` job titling the list, which
`GET /api/v1/jobs/{id}` reports on (failed `title_failed` when the title
could not be written); null when the list already has its title. 404
when the week has no plan, 409 `archived` when its list is in the Supply
archive.
- `POST /api/v1/shopping/{weekStart}/send-chat` (CSRF-checked) takes
`{}` or `{"resend": true}` and posts the Supply run to Google Chat
through `chat.PostSupplyRun`, in the week's thread: "Supply run · " and
+17 -2
View File
@@ -79,6 +79,7 @@ type Auth struct {
provider *oidc.Provider
client *http.Client
cookie loginCookie
pending *PendingStates
sessions Sessions
allowed []string
// endSessionURL is the provider's end_session_endpoint; logout ends
@@ -157,6 +158,7 @@ func New(ctx context.Context, cfg Config, sessions Sessions, logger *slog.Logger
client: client,
allowed: slices.Clone(cfg.AllowedGroups),
cookie: newLoginCookie(cfg.SecretKey),
pending: NewPendingStates(loginCookieTTL),
sessions: sessions,
endSessionURL: discovered.EndSessionURI,
logger: logger,
@@ -250,7 +252,13 @@ func (a *Auth) login(w http.ResponseWriter, r *http.Request) {
Verifier: oauth2.GenerateVerifier(),
Return: returnTo,
}
if err := a.cookie.set(w, state, time.Now()); err != nil {
now := time.Now()
if err := a.pending.Add(state.State, now); err != nil {
a.logger.WarnContext(r.Context(), "login refused", "error", err)
http.Error(w, "too many sign-ins in progress; try again later", http.StatusServiceUnavailable)
return
}
if err := a.cookie.set(w, state, now); err != nil {
a.logger.ErrorContext(r.Context(), "set login cookie", "error", err)
http.Error(w, "internal error", http.StatusInternalServerError)
return
@@ -262,7 +270,7 @@ func (a *Auth) login(w http.ResponseWriter, r *http.Request) {
}
// callback completes the flow: it checks state against the login cookie,
// exchanges the code, verifies the ID token and nonce, then starts a session
// consumes it, exchanges the code, verifies the ID token and nonce, then starts a session
// if the account is in an allowed group. Membership is checked on every
// login, so an account removed from the group is refused at its next one.
func (a *Auth) callback(w http.ResponseWriter, r *http.Request) {
@@ -283,6 +291,13 @@ func (a *Auth) callback(w http.ResponseWriter, r *http.Request) {
http.Error(w, "sign-in state mismatch; start again", http.StatusBadRequest)
return
}
// A copy of the cookie verifies until it expires, so the state is
// single-use only because it is consumed here.
if !a.pending.Consume(state.State) {
a.logger.WarnContext(ctx, "login callback state already used")
http.Error(w, "sign-in state mismatch; start again", http.StatusBadRequest)
return
}
if q.Get("error") != "" {
// error and error_description are provider-supplied text; only the
// fixed category is logged.
+91
View File
@@ -497,6 +497,59 @@ func TestCallbackDefaultReturnPath(t *testing.T) {
}
}
// A saved copy of the login cookie still verifies until it expires, so a
// replay with a second valid code is refused only because the server
// consumed the state on first use.
func TestCallbackRefusesReplayedLoginState(t *testing.T) {
t.Parallel()
h := newHarness(t)
l := h.login(t, LoginPath)
idToken := h.provider.signRS256(h.provider.claims(l.query.Get("nonce")))
if resp := h.callback(l, l.query.Get("state"), idToken); resp.StatusCode != http.StatusSeeOther {
t.Fatalf("first callback status = %d, want 303", resp.StatusCode)
}
resp := h.callback(l, l.query.Get("state"), idToken)
if resp.StatusCode != http.StatusBadRequest {
t.Errorf("replayed callback status = %d, want 400", resp.StatusCode)
}
if n := len(h.sessions.started()); n != 1 {
t.Errorf("%d sessions started, want 1", n)
}
assertLoginCookieCleared(t, resp)
}
// Pending login states live in memory, so a login started before a restart
// fails closed after it even though its cookie still verifies.
func TestCallbackRefusesLoginStartedBeforeRestart(t *testing.T) {
t.Parallel()
h := newHarness(t)
l := h.login(t, LoginPath)
restarted, err := New(context.Background(), Config{
Issuer: h.provider.issuer(),
ClientID: testClientID,
ClientSecret: testClientSecret,
PublicURL: testPublicURL,
SecretKey: []byte("server-secret-key"),
AllowedGroups: testAllowedGroups,
}, h.sessions, slog.New(slog.DiscardHandler))
if err != nil {
t.Fatal(err)
}
r := chi.NewRouter()
restarted.Mount(r)
h.router = r
resp := h.callback(l, l.query.Get("state"), h.provider.signRS256(h.provider.claims(l.query.Get("nonce"))))
if resp.StatusCode != http.StatusBadRequest {
t.Errorf("callback after restart status = %d, want 400", resp.StatusCode)
}
if n := len(h.sessions.started()); n != 0 {
t.Errorf("%d sessions started, want none", n)
}
}
func TestCallbackRejected(t *testing.T) {
t.Parallel()
@@ -819,6 +872,44 @@ func TestLoginCookieExpires(t *testing.T) {
}
}
func TestPendingStates(t *testing.T) {
t.Parallel()
p := NewPendingStates(time.Minute)
now := time.Now()
if err := p.Add("first", now); err != nil {
t.Fatal(err)
}
if !p.Consume("first") {
t.Error("Consume of a pending state = false")
}
if p.Consume("first") {
t.Error("second Consume of a state = true")
}
if p.Consume("never-added") {
t.Error("Consume of an unknown state = true")
}
for i := range maxPendingStates - 1 {
if err := p.Add(fmt.Sprint("s", i), now); err != nil {
t.Fatalf("Add %d: %v", i, err)
}
}
// "first" was consumed but has not expired, so it still counts.
if err := p.Add("over", now); !errors.Is(err, ErrTooManyPending) {
t.Errorf("Add past the bound = %v, want ErrTooManyPending", err)
}
// Once the earlier states expire, they are forgotten and a new one fits.
if err := p.Add("later", now.Add(time.Minute)); err != nil {
t.Errorf("Add after expiry: %v", err)
}
if p.Consume("s0") {
t.Error("an expired state was still pending")
}
if !p.Consume("later") {
t.Error("the state added after expiry is not pending")
}
}
func TestLoginRejectsOffOriginReturn(t *testing.T) {
t.Parallel()
h := newHarness(t)
+68
View File
@@ -10,6 +10,7 @@ import (
"net/http"
"net/url"
"strings"
"sync"
"time"
)
@@ -153,6 +154,73 @@ func (c FlowCookie) Read(r *http.Request, v flowValue, now time.Time) error {
return nil
}
// maxPendingStates bounds each flow's pending states. The login route is
// unauthenticated, so without a bound anyone could grow the set without
// limit; one household never has close to this many flows in progress.
const maxPendingStates = 10000
// ErrTooManyPending means a flow's pending states are at maxPendingStates,
// so a new flow is refused until older ones expire.
var ErrTooManyPending = errors.New("too many sign-in flows in progress")
// PendingStates holds the state of every redirect flow started and not yet
// completed, until its flow cookie expires. A copy of a flow cookie verifies
// until it expires, so a flow is single-use only because its callback
// consumes the state here. The states live in memory — MealQuest runs as one
// process — so a flow started before a restart fails closed after it.
type PendingStates struct {
ttl time.Duration
mu sync.Mutex
states map[string]time.Time
// order is every added state in the order added. Every state is pending
// for the same ttl, so it is also expiry order and Add forgets expired
// states from its front.
order []pendingState
}
type pendingState struct {
state string
expires time.Time
}
// NewPendingStates returns an empty set whose states are pending for ttl,
// the flow cookie's lifetime.
func NewPendingStates(ttl time.Duration) *PendingStates {
return &PendingStates{ttl: ttl, states: map[string]time.Time{}}
}
// Add records state, started at now, as pending and forgets expired ones.
// It fails with ErrTooManyPending when maxPendingStates are pending.
func (p *PendingStates) Add(state string, now time.Time) error {
p.mu.Lock()
defer p.mu.Unlock()
expired := 0
for expired < len(p.order) && !now.Before(p.order[expired].expires) {
delete(p.states, p.order[expired].state)
expired++
}
p.order = p.order[expired:]
// The bound counts every unexpired state, consumed or not: a consumed
// state leaves states at once but order only when it expires.
if len(p.order) >= maxPendingStates {
return ErrTooManyPending
}
expires := now.Add(p.ttl)
p.states[state] = expires
p.order = append(p.order, pendingState{state: state, expires: expires})
return nil
}
// Consume removes state from the pending states, reporting whether it was
// there.
func (p *PendingStates) Consume(state string) bool {
p.mu.Lock()
defer p.mu.Unlock()
_, ok := p.states[state]
delete(p.states, state)
return ok
}
// returnPath validates the optional post-login return path. Empty means "/".
// Only same-origin absolute paths are allowed: no scheme, host, protocol-
// relative "//" prefix, backslashes or control characters.
+8 -30
View File
@@ -137,10 +137,9 @@ type Service struct {
account sync.Mutex
// pending holds the state of every connect started and not yet
// completed, with when it expires. Complete consumes a state once, so a
// replayed cookie finds it gone.
pendingMu sync.Mutex
pending map[string]time.Time
// completed. Complete consumes a state once, so a replayed cookie finds
// it gone.
pending *auth.PendingStates
}
// NewService returns the service over st, sealing refresh tokens with box.
@@ -175,7 +174,7 @@ func NewService(st *store.Store, box *secretbox.Box, cfg Config) (*Service, erro
cookie: auth.NewFlowCookie(connectCookieName, connectCookieKeyLabel, connectCookieTTL, cfg.SecretKey),
retryDelay: firstRetryDelay,
now: time.Now,
pending: map[string]time.Time{},
pending: auth.NewPendingStates(connectCookieTTL),
}, nil
}
@@ -210,10 +209,12 @@ func (s *Service) Connect(w http.ResponseWriter, session auth.Session) (string,
Session: sessionBinding(session),
}
now := s.now()
if err := s.pending.Add(state.State, now); err != nil {
return "", fmt.Errorf("gcal: %w", err)
}
if err := s.cookie.Set(w, &state, now); err != nil {
return "", fmt.Errorf("gcal: set connect cookie: %w", err)
}
s.addPending(state.State, now)
return s.oauth.AuthCodeURL(state.State,
oauth2.AccessTypeOffline,
oauth2.ApprovalForce,
@@ -243,7 +244,7 @@ func (s *Service) Complete(w http.ResponseWriter, r *http.Request, session auth.
}
// A copy of the cookie verifies until it expires, so the state is
// single-use only because it is consumed here.
if !s.consumePending(state.State) {
if !s.pending.Consume(state.State) {
return fmt.Errorf("%w: already used", ErrBadState)
}
if q.Get("error") != "" {
@@ -276,29 +277,6 @@ func (s *Service) Complete(w http.ResponseWriter, r *http.Request, session auth.
return s.store(ctx, sealed)
}
// addPending records state, started at now, as pending until the connect
// cookie expires, and forgets expired ones.
func (s *Service) addPending(state string, now time.Time) {
s.pendingMu.Lock()
defer s.pendingMu.Unlock()
for k, expires := range s.pending {
if !now.Before(expires) {
delete(s.pending, k)
}
}
s.pending[state] = now.Add(connectCookieTTL)
}
// consumePending removes state from the pending states, reporting whether
// it was there.
func (s *Service) consumePending(state string) bool {
s.pendingMu.Lock()
defer s.pendingMu.Unlock()
_, ok := s.pending[state]
delete(s.pending, state)
return ok
}
// checkScopes fails with ErrScopes unless token's granted scopes include
// every scope in Scopes. Google lets the household untick a permission on
// the consent screen and reports what was granted in "scope".
+10 -5
View File
@@ -111,10 +111,13 @@ type pushedPlan struct {
// plan Calendar does not show and pushing again makes it; or
// supply_run_failed when it could not be made, and POST
// /shopping/{weekStart}/generate makes it again. error is null when status
// is generated.
// is generated. title_job_id is the job titling a generated Supply run,
// which GET /jobs/{id} reports on; null when it already has its title or
// was not generated.
type pushedSupplyRun struct {
Status string `json:"status"`
Error *apiError `json:"error"`
Status string `json:"status"`
Error *apiError `json:"error"`
TitleJobID *string `json:"title_job_id"`
}
// postPlanPush pushes the week's plan to the chosen Google calendar
@@ -143,7 +146,9 @@ func postPlanPush(logger *slog.Logger, svc *plan.Service, shop *shopping.Service
// gone.
ctx := context.WithoutCancel(r.Context())
run := pushedSupplyRun{Status: "generated"}
if _, err := shop.GeneratePushed(ctx, week, revision); errors.Is(err, shopping.ErrFrozen) {
if g, err := shop.GeneratePushed(ctx, week, revision); err == nil {
run.TitleJobID = g.TitleJobID
} else if errors.Is(err, shopping.ErrFrozen) {
logger.InfoContext(ctx, "Supply run in the Supply archive left as it was after Push to Calendar", "week", week.String())
run = pushedSupplyRun{Status: "archived", Error: &apiError{Code: "archived", Message: err.Error()}}
} else if errors.Is(err, shopping.ErrPlanChanged) {
@@ -153,7 +158,7 @@ func postPlanPush(logger *slog.Logger, svc *plan.Service, shop *shopping.Service
Message: "the plan is on Calendar as pushed, but it was edited before its Supply run was made, so no Supply run was made from it; push again",
Retryable: true,
}}
} else if err != nil {
} else {
logger.ErrorContext(ctx, "generate Supply run after Push to Calendar", "week", week.String(), "error", err)
run = pushedSupplyRun{Status: "failed", Error: &apiError{
Code: "supply_run_failed",
+7 -4
View File
@@ -60,10 +60,13 @@ type supplyRunItem struct {
}
// generatedSupplyRun is the POST /shopping/{weekStart}/generate answer:
// the Supply run as made, and the plan days with nothing to shop for.
// the Supply run as made, the plan days with nothing to shop for, and the
// title_job_id of the job titling it, which GET /jobs/{id} reports on;
// null when the Supply run already has its title.
type generatedSupplyRun struct {
supplyRun
Skipped []shopping.SkippedDay `json:"skipped"`
Skipped []shopping.SkippedDay `json:"skipped"`
TitleJobID *string `json:"title_job_id"`
}
// sendChatRequest is the POST /shopping/{weekStart}/send-chat body:
@@ -108,7 +111,7 @@ func getSupplyRun(logger *slog.Logger, svc *shopping.Service) http.HandlerFunc {
}
// postSupplyRunGenerate makes, or remakes, the week's Supply run from its
// plan and answers with it and the plan days it skipped.
// plan and answers with it, the plan days it skipped and its title job.
func postSupplyRunGenerate(logger *slog.Logger, svc *shopping.Service) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
week, ok := weekStartOrFail(w, r, logger)
@@ -129,7 +132,7 @@ func postSupplyRunGenerate(logger *slog.Logger, svc *shopping.Service) http.Hand
if skipped == nil {
skipped = []shopping.SkippedDay{}
}
writeJSON(w, r, logger, http.StatusOK, generatedSupplyRun{supplyRun: out, Skipped: skipped})
writeJSON(w, r, logger, http.StatusOK, generatedSupplyRun{supplyRun: out, Skipped: skipped, TitleJobID: g.TitleJobID})
}
}
+72 -10
View File
@@ -186,7 +186,7 @@ func TestSupplyRunShape(t *testing.T) {
if status != http.StatusOK {
t.Fatalf("generate = %d %v", status, raw)
}
wantTop := []string{"chat_send_unconfirmed_at", "groups", "label", "needed_count", "sent_to_chat_at", "skipped", "status", "title", "total_count", "weekStart"}
wantTop := []string{"chat_send_unconfirmed_at", "groups", "label", "needed_count", "sent_to_chat_at", "skipped", "status", "title", "title_job_id", "total_count", "weekStart"}
if got := keys(raw); !reflect.DeepEqual(got, wantTop) {
t.Errorf("generate fields = %v, want %v", got, wantTop)
}
@@ -199,7 +199,7 @@ func TestSupplyRunShape(t *testing.T) {
if status != http.StatusOK {
t.Fatalf("GET = %d %v", status, raw)
}
if k := keys(raw); !reflect.DeepEqual(k, slices.DeleteFunc(slices.Clone(wantTop), func(s string) bool { return s == "skipped" })) {
if k := keys(raw); !reflect.DeepEqual(k, slices.DeleteFunc(slices.Clone(wantTop), func(s string) bool { return s == "skipped" || s == "title_job_id" })) {
t.Errorf("GET fields = %v", k)
}
group := raw["groups"].([]any)[0].(map[string]any)
@@ -376,8 +376,12 @@ func TestPushPlanMakesSupplyRun(t *testing.T) {
if rec.Code != http.StatusOK || body["status"] != "synced" {
t.Fatalf("push = %d %s", rec.Code, rec.Body)
}
if want := map[string]any{"status": "generated", "error": nil}; !reflect.DeepEqual(body["supply_run"], want) {
t.Errorf("supply_run = %v, want %v", body["supply_run"], want)
run, _ := body["supply_run"].(map[string]any)
if k := keys(run); run["status"] != "generated" || run["error"] != nil || !reflect.DeepEqual(k, []string{"error", "status", "title_job_id"}) {
t.Errorf("supply_run = %v, want generated with no error", run)
}
if job := titleJob(t, e, run["title_job_id"]); job["status"] != "queued" {
t.Errorf("title job = %v, want queued: no runner in this test", job)
}
var l supplyRunJSON
if status, got := callShopping(t, e, http.MethodGet, shoppingPath+"2026-10-12", "", &l); status != http.StatusOK {
@@ -467,9 +471,11 @@ func TestPushPlanDoesNotWaitForTitle(t *testing.T) {
if rec.Code != http.StatusOK || body["status"] != "synced" {
t.Fatalf("push = %d %s", rec.Code, rec.Body)
}
if want := map[string]any{"status": "generated", "error": nil}; !reflect.DeepEqual(body["supply_run"], want) {
t.Errorf("supply_run = %v, want %v", body["supply_run"], want)
run, _ := body["supply_run"].(map[string]any)
if run["status"] != "generated" || run["error"] != nil {
t.Errorf("supply_run = %v, want generated with no error", run)
}
titleJobID, _ := run["title_job_id"].(string)
held.waitEntered(t)
var l supplyRunJSON
callShopping(t, e, http.MethodGet, shoppingPath+"2026-10-12", "", &l)
@@ -478,6 +484,7 @@ func TestPushPlanDoesNotWaitForTitle(t *testing.T) {
}
close(held.release)
wantTitleJobSucceeded(t, e, titleJobID)
if got := e.waitTitled(t); got != "Chili and Friends" {
t.Errorf("title = %q, want the title job's", got)
}
@@ -493,7 +500,8 @@ func TestSupplyRunGenerateDoesNotWaitForTitle(t *testing.T) {
e.startRunner(t)
var l supplyRunJSON
if status, body := callShopping(t, e, http.MethodPost, shoppingPath+"2026-10-12/generate", "", &l); status != http.StatusOK {
status, body := callShopping(t, e, http.MethodPost, shoppingPath+"2026-10-12/generate", "", &l)
if status != http.StatusOK {
t.Fatalf("generate = %d %v", status, body)
}
if l.Title != nil {
@@ -502,12 +510,66 @@ func TestSupplyRunGenerateDoesNotWaitForTitle(t *testing.T) {
if !reflect.DeepEqual(l.render(), weekLines) {
t.Errorf("generated = %v, want %v", l.render(), weekLines)
}
titleJobID, _ := body["title_job_id"].(string)
held.waitEntered(t)
close(held.release)
wantTitleJobSucceeded(t, e, titleJobID)
if got := e.waitTitled(t); got != "Chili and Friends" {
t.Errorf("title = %q, want the title job's", got)
}
if _, body := callShopping(t, e, http.MethodPost, shoppingPath+"2026-10-12/generate", "", nil); body["title_job_id"] != nil {
t.Errorf("remake title_job_id = %v, want null: the Supply run has its title", body["title_job_id"])
}
}
// TestSupplyRunGenerateTitleJobFails: a title job that cannot title the
// Supply run is reported failed with title_failed by GET /jobs/{id}, and
// the week's next generation answers with a new title job.
func TestSupplyRunGenerateTitleJobFails(t *testing.T) {
t.Parallel()
e := newPlansRouter(t)
e.groundWeek(t)
e.startRunner(t)
_, body := callShopping(t, e, http.MethodPost, shoppingPath+"2026-10-12/generate", "", nil)
first := titleJob(t, e, body["title_job_id"])
job := waitJob(t, e.h, first["id"].(string))
if job["status"] != "failed" || job["error"] != shopping.ReasonTitleFailed {
t.Errorf("title job = %v, want failed %s", job, shopping.ReasonTitleFailed)
}
_, body = callShopping(t, e, http.MethodPost, shoppingPath+"2026-10-12/generate", "", nil)
if second := titleJob(t, e, body["title_job_id"]); second["id"] == first["id"] {
t.Errorf("retry title_job_id = %v, want a new job", second["id"])
}
}
// titleJob returns the title_supply_run job id, as GET /jobs/{id} reports
// it; id must be a job id string.
func titleJob(t *testing.T, e plansEnv, id any) map[string]any {
t.Helper()
s, ok := id.(string)
if !ok || s == "" {
t.Fatalf("title_job_id = %v, want a job id", id)
}
rec, job := callDinners(t, e.h, http.MethodGet, "/api/v1/jobs/"+s, "")
if rec.Code != http.StatusOK || job["kind"] != shopping.KindTitleSupplyRun {
t.Fatalf("GET title job = %d %s, want a %s job", rec.Code, rec.Body, shopping.KindTitleSupplyRun)
}
return job
}
// wantTitleJobSucceeded waits for title job id to finish and fails t
// unless GET /jobs/{id} reports it succeeded with the title written.
func wantTitleJobSucceeded(t *testing.T, e plansEnv, id string) {
t.Helper()
titleJob(t, e, id)
job := waitJob(t, e.h, id)
result, _ := job["result"].(map[string]any)
if job["status"] != "succeeded" || result["title"] != "Chili and Friends" {
t.Errorf("title job = %v, want succeeded with the title", job)
}
}
// TestFailedPushMakesNoSupplyRun: a push Calendar refused makes no Supply
@@ -546,7 +608,7 @@ func TestPushPlanSupplyRunFails(t *testing.T) {
"code": "supply_run_failed",
"message": "the plan is on Calendar, but its Supply run could not be made; make it again from the Supply run",
"retryable": true,
}}
}, "title_job_id": nil}
if !reflect.DeepEqual(body["supply_run"], want) {
t.Errorf("supply_run = %v, want %v", body["supply_run"], want)
}
@@ -599,7 +661,7 @@ func TestPushPlanEditedBeforeSupplyRun(t *testing.T) {
"code": "plan_changed",
"message": "the plan is on Calendar as pushed, but it was edited before its Supply run was made, so no Supply run was made from it; push again",
"retryable": true,
}}
}, "title_job_id": nil}
if !reflect.DeepEqual(body["supply_run"], want) {
t.Errorf("supply_run = %v, want %v", body["supply_run"], want)
}
@@ -632,7 +694,7 @@ func TestPushPlanLeavesArchivedSupplyRun(t *testing.T) {
t.Fatalf("push = %d %s", rec.Code, rec.Body)
}
run, _ := body["supply_run"].(map[string]any)
if apiErr, _ := run["error"].(map[string]any); run["status"] != "archived" || apiErr["code"] != "archived" {
if apiErr, _ := run["error"].(map[string]any); run["status"] != "archived" || apiErr["code"] != "archived" || run["title_job_id"] != nil {
t.Errorf("supply_run = %v, want archived", body["supply_run"])
}
var l supplyRunJSON
+35
View File
@@ -311,3 +311,38 @@ func TestGenerateTitleOverlapAcrossRename(t *testing.T) {
t.Errorf("stored title = %v, want one of the two titles", deref(got))
}
}
// TestGenerateTitleNotStoredOnReplacedList: a list replaced by a new
// untitled list of the same week once archive_title has answered, before
// its title is stored, keeps the replacement untitled; the old list's
// title job fails, and the failure is logged.
func TestGenerateTitleNotStoredOnReplacedList(t *testing.T) {
e := titleWeek(t)
e.model.QueueGenerate(titled("Noodles, Tacos and Chili"))
e.afterTitle = func() {
for _, stmt := range []string{
`DELETE FROM shopping_items WHERE group_id IN (SELECT id FROM shopping_groups WHERE list_id IN (SELECT id FROM shopping_lists WHERE week_start = ?1))`,
`DELETE FROM shopping_groups WHERE list_id IN (SELECT id FROM shopping_lists WHERE week_start = ?1)`,
`UPDATE shopping_lists SET id = 'replacement' WHERE week_start = ?1`,
} {
if _, err := e.st.Writer.Exec(stmt, week.String()); err != nil {
t.Error(err)
}
}
}
g := e.generate(t)
wantJob(t, e.waitTitle(t, g), jobs.StatusFailed, shopping.ReasonTitleFailed)
var id string
if err := e.st.Reader.QueryRow(`SELECT id FROM shopping_lists WHERE week_start = ?`, week.String()).Scan(&id); err != nil {
t.Fatal(err)
}
if id != "replacement" || id == g.List.ID {
t.Fatalf("week's list = %s, want the replacement of %s", id, g.List.ID)
}
wantUntitled(t, "replacement's title", e.storedTitle(t))
if !strings.Contains(e.log.String(), "the Supply run was frozen or replaced before its title was stored") {
t.Errorf("log lacks the failure:\n%s", e.log.String())
}
}
+85 -29
View File
@@ -54,7 +54,8 @@ func (r batchResult) PublicJSON() ([]byte, error) { return json.Marshal(r) }
// no recipe grounds is dropped; any other grounding failure fails the
// batch, as does suggest_side_quests failing, with no Side quest made up
// instead. The batch ends done or failed. When recording the batch's end
// fails, so does the job; when it times out at shutdown, the job is left
// fails, so does the job, with what the batch stored, and any
// new_suggestions it emitted, kept; when it times out at shutdown, the job is left
// running and the batch with it, for the next start to fail.
//
// When a batch stores its Side quests depends on whether it notifies (see
@@ -64,7 +65,12 @@ func (r batchResult) PublicJSON() ([]byte, error) { return json.Marshal(r) }
// Side quest stored. So the household is told of exactly what the batch
// added, once; and a batch that fails at any round, including a later
// round's normalize_recipe, stores nothing and sends nothing, leaving no
// Side quest in the deck that was never announced. A replenish batch,
// Side quest in the deck that was never announced. After each round it
// rechecks every Side quest it holds (see eligible) and lets go of those
// the household's changes made ineligible, so a later round grounds a
// spare Side quest in place of each; only a change committed between the
// last recheck and the store can still leave the batch short of
// BatchSize while spare Side quests remain. A replenish batch,
// which tops up the deck as the household swipes and never notifies,
// stores each round's Side quests before the next round starts, so the
// deck fills as it goes; those stored before a failure stay in the deck.
@@ -146,7 +152,13 @@ func (s *Service) fill(ctx context.Context, trigger string, job BatchJob, publis
}
res.Dropped += len(round) - len(grounded)
if note != nil {
held = append(held, grounded...)
// Recheck every held Side quest, not just this round's, so
// one the household's changes made ineligible while this
// round grounded is replaced by a later round, not dropped by
// the store once no round is left to replace it.
if held, err = s.recheck(ctx, &res, append(held, grounded...)); err != nil {
return batchResult{}, err
}
} else if err := s.storeInto(ctx, &res, nil, grounded); err != nil {
return batchResult{}, err
}
@@ -163,6 +175,22 @@ func (s *Service) fill(ctx context.Context, trigger string, job BatchJob, publis
return res, nil
}
// recheck returns, in order, those of a notifying batch's held Side
// quests that store would still keep (see eligible), adding those left
// out to res.Dropped.
func (s *Service) recheck(ctx context.Context, res *batchResult, held []groundedQuest) ([]groundedQuest, error) {
kept, err := s.eligible(ctx, db.New(s.st.Reader), res.BatchID, held)
if err != nil {
return nil, err
}
res.Dropped += len(held) - len(kept)
out := make([]groundedQuest, len(kept))
for i, k := range kept {
out[i] = k.groundedQuest
}
return out, nil
}
// storeInto stores quests for res's batch with note (see store) and adds
// the suggestions stored, and the Side quests left out, to res.
func (s *Service) storeInto(ctx context.Context, res *batchResult, note *notice, quests []groundedQuest) error {
@@ -277,11 +305,11 @@ func (s *Service) ground(ctx context.Context, quests []skills.SideQuest) ([]grou
// store stores each grounded Side quest of batchID as a dinner with status
// suggested and origin side_quest, grounded in its recipe, and a pending
// suggestion citing its Regulars by id, all in one transaction, and
// returns the suggestion ids. A Side quest is left out when, since
// suggest_side_quests ran, a dinner it repeats was added, a suggestion it
// repeats passed or a Rotation check dropped, or a Regular it cites renamed
// or archived. Passed names are read in the same transaction as the store,
// so a Pass or drop committed before it is never missed.
// returns the suggestion ids. A Side quest eligible leaves out is not
// stored: since suggest_side_quests ran, a dinner it repeats was added, a
// suggestion it repeats passed or a Rotation check dropped, or a Regular
// it cites renamed or archived. eligible reads in the same transaction as
// the store, so a Pass or drop committed before it is never missed.
//
// When note is not nil and a Side quest is stored, new_suggestions keyed by
// note, counting and naming those stored, is emitted in the same
@@ -295,28 +323,13 @@ func (s *Service) store(ctx context.Context, batchID string, note *notice, quest
var stored []string
err := s.st.WithTx(ctx, func(tx *sql.Tx) error {
q := db.New(tx)
existing, err := q.ListUnarchivedDinners(ctx)
kept, err := s.eligible(ctx, q, batchID, quests)
if err != nil {
return fmt.Errorf("suggest: list dinners for dedupe: %w", err)
return err
}
names, err := q.ListPassedNames(ctx)
if err != nil {
return fmt.Errorf("suggest: list passed names for dedupe: %w", err)
}
regulars := map[string][]string{}
for _, row := range existing {
names = append(names, row.Dinner.Name)
if row.Dinner.Status == dinners.StatusRegular {
regulars[row.Dinner.Name] = append(regulars[row.Dinner.Name], row.Dinner.ID)
}
}
for _, gq := range quests {
cited, ok := citedIDs(gq.quest.CitedRegulars, regulars)
if !ok || repeats(gq.quest.Name, names) {
s.logger.InfoContext(ctx, "suggested dinner dropped: changed while grounding", "batch_id", batchID)
continue
}
citedJSON, err := json.Marshal(cited)
for _, k := range kept {
gq := k.groundedQuest
citedJSON, err := json.Marshal(k.cited)
if err != nil {
return fmt.Errorf("suggest: encode cited Regulars: %w", err)
}
@@ -354,7 +367,6 @@ func (s *Service) store(ctx context.Context, batchID string, note *notice, quest
}); err != nil {
return fmt.Errorf("suggest: create suggestion: %w", err)
}
names = append(names, d.Name)
ids = append(ids, id)
stored = append(stored, d.Name)
}
@@ -372,6 +384,50 @@ func (s *Service) store(ctx context.Context, batchID string, note *notice, quest
return ids, nil
}
// eligibleQuest is a Side quest store would keep, with the ids of the
// Regulars it cites.
type eligibleQuest struct {
groundedQuest
cited []string
}
// eligible returns, in order, those of batchID's quests that q's dinners
// and passed names still allow, each with the ids of the Regulars it
// cites. A Side quest is left out when a dinner it repeats is unarchived,
// a suggestion it repeats passed or a Rotation check dropped, a Regular it
// cites is renamed or archived, or an earlier Side quest of quests kept
// repeats it. store checks with this in its transaction; a notifying
// batch checks the Side quests it holds with it after each round (see
// fill).
func (s *Service) eligible(ctx context.Context, q *db.Queries, batchID string, quests []groundedQuest) ([]eligibleQuest, error) {
existing, err := q.ListUnarchivedDinners(ctx)
if err != nil {
return nil, fmt.Errorf("suggest: list dinners for dedupe: %w", err)
}
names, err := q.ListPassedNames(ctx)
if err != nil {
return nil, fmt.Errorf("suggest: list passed names for dedupe: %w", err)
}
regulars := map[string][]string{}
for _, row := range existing {
names = append(names, row.Dinner.Name)
if row.Dinner.Status == dinners.StatusRegular {
regulars[row.Dinner.Name] = append(regulars[row.Dinner.Name], row.Dinner.ID)
}
}
var kept []eligibleQuest
for _, gq := range quests {
cited, ok := citedIDs(gq.quest.CitedRegulars, regulars)
if !ok || repeats(gq.quest.Name, names) {
s.logger.InfoContext(ctx, "suggested dinner dropped: changed while grounding", "batch_id", batchID)
continue
}
names = append(names, gq.quest.Name)
kept = append(kept, eligibleQuest{groundedQuest: gq, cited: cited})
}
return kept, nil
}
// citedIDs returns the ids of the Regulars named cited, false when one
// names no Regular or more than one.
func citedIDs(cited []string, regulars map[string][]string) ([]string, bool) {
+61
View File
@@ -660,6 +660,67 @@ func TestBatchKeepsAtMostBatchSize(t *testing.T) {
}
}
// TestBatchReplacesSideQuestMadeIneligibleWhileHeld: a notifying batch
// holds its first round's Side quests until its last round is done; when
// the household adds a dinner one of them repeats while that round grounds,
// the batch lets it go and grounds a spare Side quest in its place, so it
// still stores BatchSize and announces them once.
func TestBatchReplacesSideQuestMadeIneligibleWhileHeld(t *testing.T) {
for _, trigger := range []string{suggest.TriggerScheduled, suggest.TriggerManual} {
t.Run(trigger, func(t *testing.T) {
e := newEnv(t)
e.start(t)
var quests, want []string
for i := range suggest.BatchSize + 1 {
name := fmt.Sprintf("Fajita Dish %c", 'A'+i)
e.recipe(t, name)
quests = append(quests, sideQuest(name, "Like your Chicken Fajitas.", "Chicken Fajitas"))
if i != 0 {
want = append(want, name)
}
}
e.reply(quests...)
// The first round's last search waits, so the household adds a
// dinner repeating Fajita Dish A while that round grounds.
last := fmt.Sprintf("Fajita Dish %c", 'A'+suggest.BatchSize-1)
gate := make(chan struct{})
e.search.Gates = map[string]chan struct{}{last: gate}
b, key := e.startBatch(t, trigger)
deadline := time.Now().Add(10 * time.Second)
for !slices.Contains(e.search.Queries(), last) {
if time.Now().After(deadline) {
close(gate)
t.Fatalf("search for %s never ran", last)
}
time.Sleep(5 * time.Millisecond)
}
e.dinner(t, "Fajita Dish A", "regular")
close(gate)
job := e.waitJob(t, b.JobID)
if job.Status != jobs.StatusSucceeded {
t.Fatalf("batch job %s, error %v", job.Status, job.Error)
}
var res batchResult
if err := json.Unmarshal(job.Result, &res); err != nil {
t.Fatal(err)
}
if len(res.SuggestionIDs) != suggest.BatchSize || res.Dropped != 1 {
t.Errorf("batch result = %+v, want %d stored and 1 dropped", res, suggest.BatchSize)
}
if got := e.deckNames(t); !slices.Equal(got, want) {
t.Errorf("deck = %v, want %v", got, want)
}
rows := notifytest.Rows(t, e.st, notify.NewSuggestions)
title := fmt.Sprintf("%d new Side quests are in", suggest.BatchSize)
if len(rows) != 1 || rows[0].Key != key || rows[0].Payload.Title != title {
t.Errorf("new_suggestions rows = %+v, want one keyed %s titled %q", rows, key, title)
}
})
}
}
func TestDeckIsOldestFirst(t *testing.T) {
e := newEnv(t)
e.start(t)