Release corrections round 2: offline onboarding lockout and stale candidates after rename (#103)
CI / ci (push) Successful in 2m40s

Co-authored-by: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com>
This commit was merged in pull request #103.
This commit is contained in:
2026-10-07 00:14:32 -04:00
committed by approver_agent
parent cb920714b0
commit 4a209623f5
8 changed files with 368 additions and 50 deletions
+20 -10
View File
@@ -149,7 +149,7 @@ type Service struct {
// id, the candidates last offered by Candidates until they expire, so
// a candidate id is resolved here and its URL never leaves the server.
// offerSearches holds the Candidates searches running, by dinner id,
// so clearing a dinner's offers keeps them from storing theirs.
// so clearing a dinner's offers keeps them from storing stale ones.
offersMu sync.Mutex
offers map[string]*offerSet
offerSearches map[string]map[*offerSearch]struct{}
@@ -305,11 +305,14 @@ func (s *Service) List(ctx context.Context, status string) ([]Dinner, error) {
// Update applies p to dinner id, recording updatedBy. An invalid patch is a
// *ValidationError and a new name repeating another unarchived dinner a
// *DuplicateError; either changes nothing. Renaming a dinner still pending
// clears its recipe_url, raises its grounding_revision and enqueues a fresh
// KindGroundDinner job that grounds the new name: a job queued or running
// for the old name or link carries the old revision, so it can no longer
// write the dinner, and one running is stopped.
// *DuplicateError; either changes nothing. Renaming a dinner raises its
// grounding_revision and drops the candidates offered for its old name:
// each was offered at an earlier revision, so none of them grounds it.
// Renaming a dinner still pending also enqueues a fresh KindGroundDinner
// job that grounds the new name, and clears its recipe_url when the
// normalized name changes: a job queued or running for the old name or
// link carries the old revision, so it can no longer write the dinner, and
// one running is stopped.
func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy string) (Dinner, error) {
if updatedBy == "" {
return Dinner{}, errors.New("dinners: update: no member to record as updated_by")
@@ -320,7 +323,7 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
}
var row db.Dinner
var regrounding bool
var renamed, regrounding bool
err = s.st.WithTx(ctx, func(tx *sql.Tx) error {
q := db.New(tx)
cur, err := q.GetDinner(ctx, id)
@@ -354,11 +357,13 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
}
params.Name = name
params.NormalizedName = harness.Normalize(name)
if name != cur.Name {
params.GroundingRevision++
}
if cur.GroundingStatus == GroundingPending && params.NormalizedName != cur.NormalizedName {
// The link was the old dinner's recipe; the new name is
// grounded by name, and a link can be given by hand.
params.RecipeUrl = sql.NullString{}
params.GroundingRevision++
}
}
if p.Method != nil {
@@ -374,12 +379,17 @@ func (s *Service) Update(ctx context.Context, id string, p Patch, updatedBy stri
if err != nil {
return fmt.Errorf("dinners: update %s: %w", id, err)
}
regrounding = row.GroundingRevision != cur.GroundingRevision
renamed = row.GroundingRevision != cur.GroundingRevision
regrounding = renamed && cur.GroundingStatus == GroundingPending
return nil
})
if err != nil {
return Dinner{}, err
}
if renamed {
updateCommitted()
s.clearOffers(row.ID, row.GroundingRevision)
}
if regrounding {
s.stopGroundings(row.ID, row.GroundingRevision)
jobID, err := s.enqueueGrounding(ctx, row.ID, row.GroundingRevision)
@@ -416,7 +426,7 @@ func (s *Service) Archive(ctx context.Context, id, updatedBy string) (Dinner, er
return Dinner{}, fmt.Errorf("dinners: archive %s: %w", id, err)
}
s.stopGroundings(row.ID, math.MaxInt64)
s.clearOffers(row.ID)
s.clearOffers(row.ID, math.MaxInt64)
return fromRow(row), nil
}
+17
View File
@@ -22,6 +22,15 @@ func SetOfferLifetime(t *testing.T, d time.Duration) {
t.Cleanup(func() { offerLifetime = old })
}
// SetUpdateCommitted runs f in Update once a rename is committed, before
// the dinner's offers are cleared, until t ends. A test calling it must
// not run in parallel.
func SetUpdateCommitted(t *testing.T, f func()) {
old := updateCommitted
updateCommitted = f
t.Cleanup(func() { updateCommitted = old })
}
// AdvanceClock moves the service's clock d ahead of real time until t
// ends, without firing any timer. A test calling it must not run in
// parallel.
@@ -40,6 +49,14 @@ func (s *Service) HasOffers(id string) bool {
return ok
}
// StoreOffers stores urls as dinner id's offers, as a Candidates search
// that read the dinner at grounding_revision revision would on finishing.
func (s *Service) StoreOffers(id string, urls map[string]string, revision int64) {
keep, done := s.trackOffers(id)
defer done()
keep(urls, revision)
}
// OfferCount is the number of dinners whose candidates the service holds.
func (s *Service) OfferCount() int {
s.offersMu.Lock()
+53 -30
View File
@@ -9,6 +9,7 @@ import (
"encoding/json"
"errors"
"fmt"
"math"
"slices"
"strings"
"time"
@@ -303,30 +304,38 @@ var offerLifetime = 30 * time.Minute
// now is the clock offer lifetimes are measured on; tests advance it.
var now = time.Now
// updateCommitted runs in Update once a rename is committed, before the
// dinner's offers are cleared; tests use it to act in between.
var updateCommitted = func() {}
// offerSet is the candidates last offered for a dinner: the page URL of
// each candidate id. They ground the dinner until deadline; expiry drops
// the set from offers once it has passed.
// each candidate id, found by searching the dinner's name at its
// grounding_revision revision. They ground the dinner until deadline, and
// only while it is still at that revision, which every rename raises;
// expiry drops the set from offers once deadline has passed.
type offerSet struct {
urls map[string]string
revision int64
deadline time.Time
expiry *time.Timer
}
// offerSearch is a Candidates search running for a dinner. It is stale
// once the dinner's offers are cleared, and then stores none.
// offerSearch is a Candidates search running for a dinner. It stores no
// offers found at a grounding_revision below below, which clearing the
// dinner's offers raises.
type offerSearch struct {
stale bool
below int64
}
// Candidates searches for dinner id's name and returns up to MaxCandidates
// recipe pages to ground it in, best first; the list may be empty. They
// replace the dinner's earlier candidates: one not offered again no longer
// grounds it. They ground it for offerLifetime, and not at all once it is
// grounded by hand or archived, even while this search runs. A search that
// fails wraps recipes.ErrSearchFailed.
// grounded by hand, renamed or archived, even while this search runs. A
// search that fails wraps recipes.ErrSearchFailed.
func (s *Service) Candidates(ctx context.Context, id string) ([]Candidate, error) {
// Tracked before the dinner is read: grounding or archiving it after
// the read makes this search stale.
// Tracked before the dinner is read: grounding, renaming or archiving
// it after the read keeps this search from storing what it found.
keep, done := s.trackOffers(id)
defer done()
row, err := s.get(ctx, id)
@@ -349,15 +358,17 @@ func (s *Service) Candidates(ctx context.Context, id string) ([]Candidate, error
}
}
if row.Status != StatusArchived {
keep(urls)
keep(urls, row.GroundingRevision)
}
return out, nil
}
// trackOffers registers a Candidates search for dinner id. It returns keep,
// which stores urls as the dinner's offers for offerLifetime unless
// clearOffers has run for it since, and done, to call when the search ends.
func (s *Service) trackOffers(id string) (keep func(urls map[string]string), done func()) {
// which stores urls, found for the dinner at grounding_revision revision,
// as its offers for offerLifetime unless clearOffers has run for it since
// with a revision above that, or offers found at a later revision are
// stored, and done, to call when the search ends.
func (s *Service) trackOffers(id string) (keep func(urls map[string]string, revision int64), done func()) {
search := &offerSearch{}
s.offersMu.Lock()
if s.offerSearches[id] == nil {
@@ -365,14 +376,19 @@ func (s *Service) trackOffers(id string) (keep func(urls map[string]string), don
}
s.offerSearches[id][search] = struct{}{}
s.offersMu.Unlock()
keep = func(urls map[string]string) {
keep = func(urls map[string]string, revision int64) {
s.offersMu.Lock()
defer s.offersMu.Unlock()
if search.stale {
if revision < search.below {
return
}
if set := s.offers[id]; set != nil && set.revision > revision {
// Found at a later revision, so the earlier search's offers
// are already stale.
return
}
s.dropOffers(id)
set := &offerSet{urls: urls, deadline: now().Add(offerLifetime)}
set := &offerSet{urls: urls, revision: revision, deadline: now().Add(offerLifetime)}
// expiry waits for offersMu, so it finds set stored even when it
// fires at once, and drops set only, never offers stored later.
set.expiry = time.AfterFunc(offerLifetime, func() {
@@ -395,15 +411,20 @@ func (s *Service) trackOffers(id string) (keep func(urls map[string]string), don
return keep, done
}
// clearOffers drops dinner id's offers and makes every Candidates search
// running for it stale, so no candidate offered before grounds it. Call
// it once the dinner is grounded by hand or archived.
func (s *Service) clearOffers(id string) {
// clearOffers drops dinner id's offers found at a grounding_revision below
// below and keeps every Candidates search running for it from storing
// such offers, so none of them grounds it. Offers found at below or above
// are kept. Call it with math.MaxInt64 once the dinner is grounded by hand
// or archived, and with its new revision once it is renamed: a search
// that read the rename meanwhile keeps its offers.
func (s *Service) clearOffers(id string, below int64) {
s.offersMu.Lock()
defer s.offersMu.Unlock()
s.dropOffers(id)
if set := s.offers[id]; set != nil && set.revision < below {
s.dropOffers(id)
}
for search := range s.offerSearches[id] {
search.stale = true
search.below = max(search.below, below)
}
}
@@ -425,15 +446,17 @@ func (s *Service) candidateID(id, rawURL string) string {
return base64.RawURLEncoding.EncodeToString(mac.Sum(nil)[:16])
}
// candidateURL is the page URL of candidate cid last offered for dinner id.
// candidateURL is the page URL of candidate cid last offered for dinner d.
// A cid not among the dinner's live offers, or whose page is not a valid
// recipe link, is a *ValidationError. Offers past their deadline are not
// live, even before their expiry drops them.
func (s *Service) candidateURL(id, cid string) (string, error) {
// recipe link, is a *ValidationError. Offers past their deadline, or found
// at a grounding_revision other than d's, are not live, even before they
// are dropped: d was renamed since they were found, if only to its name
// then.
func (s *Service) candidateURL(d db.Dinner, cid string) (string, error) {
var raw string
var ok bool
s.offersMu.Lock()
if set := s.offers[id]; set != nil && now().Before(set.deadline) {
if set := s.offers[d.ID]; set != nil && set.revision == d.GroundingRevision && now().Before(set.deadline) {
raw, ok = set.urls[cid]
}
s.offersMu.Unlock()
@@ -499,7 +522,7 @@ func (s *Service) Ground(ctx context.Context, id string, f Fix, updatedBy string
return Dinner{}, ErrArchived
}
if f.CandidateID != nil {
if link, err = s.candidateURL(cur.ID, *f.CandidateID); err != nil {
if link, err = s.candidateURL(cur, *f.CandidateID); err != nil {
return Dinner{}, err
}
}
@@ -537,7 +560,7 @@ func (s *Service) Ground(ctx context.Context, id string, f Fix, updatedBy string
if err != nil {
return Dinner{}, fmt.Errorf("%w: dinner %s: record: %w", ErrGroundFailed, id, err)
}
s.clearOffers(cur.ID)
s.clearOffers(cur.ID, math.MaxInt64)
return fromRow(row), nil
}
@@ -660,7 +683,7 @@ func (s *Service) groundManual(ctx context.Context, id string, m ManualRecipe, u
if err != nil {
return Dinner{}, fmt.Errorf("%w: dinner %s: %w", ErrGroundFailed, id, err)
}
s.clearOffers(row.ID)
s.clearOffers(row.ID, math.MaxInt64)
return fromRow(row), nil
}
+232 -4
View File
@@ -804,8 +804,8 @@ func wantStaleCandidate(t *testing.T, e *env, id, cid string) {
}
// TestGroundingOrArchivingClearsCandidates: once a dinner is grounded by
// hand — by candidate, link or manual recipe — or archived, the service
// holds none of its candidates' URLs and none of them grounds it.
// hand — by candidate, link or manual recipe — renamed or archived, the
// service holds none of its candidates' URLs and none of them grounds it.
func TestGroundingOrArchivingClearsCandidates(t *testing.T) {
t.Parallel()
tests := []struct {
@@ -834,6 +834,11 @@ func TestGroundingOrArchivingClearsCandidates(t *testing.T) {
t.Fatal(err)
}
}, false},
{"rename", func(t *testing.T, e *env, id, _ string) {
if _, err := e.svc.Update(t.Context(), id, dinners.Patch{Name: ptr("Beef stew")}, member); err != nil {
t.Fatal(err)
}
}, false},
{"archive", func(t *testing.T, e *env, id, _ string) {
if _, err := e.svc.Archive(t.Context(), id, member); err != nil {
t.Fatal(err)
@@ -860,9 +865,227 @@ func TestGroundingOrArchivingClearsCandidates(t *testing.T) {
}
}
// TestRenamePendingRefusesOldCandidates: candidates offered for a pending
// dinner's old name no longer ground it once it is renamed — Ground refuses
// them and leaves the old recipe off it — while candidates for the new name
// do.
func TestRenamePendingRefusesOldCandidates(t *testing.T) {
t.Parallel()
// Unstarted, so the dinner stays pending with its jobs queued.
e := newUnstartedEnv(t)
e.search.Results["Beef stew"] = []recipes.Candidate{{URL: stew}}
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
old := candidateID(t, e, c.ID, "Chicken Curry")
d, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Name: ptr("Beef stew")}, member)
if err != nil {
t.Fatal(err)
}
if d.GroundingStatus != dinners.GroundingPending {
t.Fatalf("renamed = %+v, want still pending", d)
}
if e.svc.HasOffers(c.ID) {
t.Errorf("service still holds the old name's candidates of dinner %s", c.ID)
}
wantStaleCandidate(t, e, c.ID, old)
if row := e.dinner(t, c.ID); row.RecipeID.Valid || row.GroundingStatus != dinners.GroundingPending {
t.Fatalf("dinner = %+v, want pending with no recipe", row)
}
e.model.QueueGenerate(recipestest.NormalizedReply("oven", 150))
g, err := e.svc.Ground(t.Context(), c.ID, dinners.Fix{CandidateID: ptr(candidateID(t, e, c.ID, "Beef Stew"))}, member)
if err != nil {
t.Fatal(err)
}
if g.GroundingStatus != dinners.GroundingGrounded || g.RecipeID == nil {
t.Fatalf("dinner = %+v, want grounded for the new name", g)
}
rec, err := db.New(e.st.Reader).GetRecipe(t.Context(), *g.RecipeID)
if err != nil {
t.Fatal(err)
}
if rec.SourceUrl != stew {
t.Errorf("recipe source = %s, want %s", rec.SourceUrl, stew)
}
}
// TestRenameSameNormalizedNameRefusesOldCandidates: a rename that leaves
// the normalized name as it was still drops the candidates offered before
// it, and none of them grounds the dinner.
func TestRenameSameNormalizedNameRefusesOldCandidates(t *testing.T) {
t.Parallel()
// Unstarted, so the dinner stays pending with its jobs queued.
e := newUnstartedEnv(t)
e.search.Results["Quick Chicken curry"] = e.search.Results["Chicken curry"]
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
old := candidateID(t, e, c.ID, "Chicken Curry")
before := e.dinner(t, c.ID)
d, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Name: ptr("Quick Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
if e.svc.HasOffers(c.ID) {
t.Errorf("service still holds the candidates offered before dinner %s was renamed", c.ID)
}
// A reply is queued, so only refusing the candidate keeps its recipe off.
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
wantStaleCandidate(t, e, c.ID, old)
row := e.dinner(t, c.ID)
if row.RecipeID.Valid {
t.Fatalf("dinner = %+v, want no recipe", row)
}
if row.NormalizedName != before.NormalizedName {
t.Fatalf("normalized name = %q, want %q unchanged", row.NormalizedName, before.NormalizedName)
}
// Still pending, now at the rename's revision, with a job for it.
if d.GroundingStatus != dinners.GroundingPending || row.GroundingRevision != before.GroundingRevision+1 {
t.Fatalf("renamed = %+v, want pending at revision %d", row, before.GroundingRevision+1)
}
var queued int
if err := e.st.Reader.QueryRowContext(t.Context(),
`SELECT count(*) FROM jobs WHERE status = 'queued' AND json_extract(payload, '$.revision') = ?`,
row.GroundingRevision).Scan(&queued); err != nil {
t.Fatal(err)
}
if queued != 1 {
t.Fatalf("queued jobs at revision %d = %d, want 1", row.GroundingRevision, queued)
}
}
// TestCandidatesBetweenRenameAndClearKept: a candidate search that reads
// a dinner after its rename commits, and stores its candidates before the
// rename clears the dinner's offers, keeps them: they ground the renamed
// dinner, while one offered before the rename does not, even when a search
// that read the dinner before the rename finishes after them.
func TestCandidatesBetweenRenameAndClearKept(t *testing.T) {
// Unstarted, so the dinner stays pending with its jobs queued.
e := newUnstartedEnv(t)
e.search.Results["Beef stew"] = []recipes.Candidate{{URL: stew}}
c, err := e.svc.CreateRegular(t.Context(), dinners.NewRegular{Name: ptr("Chicken curry")}, member)
if err != nil {
t.Fatal(err)
}
old := candidateID(t, e, c.ID, "Chicken Curry")
before := e.dinner(t, c.ID)
var fresh string
dinners.SetUpdateCommitted(t, func() {
fresh = candidateID(t, e, c.ID, "Beef Stew")
e.svc.StoreOffers(c.ID, map[string]string{old: curryA}, before.GroundingRevision)
})
if _, err := e.svc.Update(t.Context(), c.ID, dinners.Patch{Name: ptr("Beef stew")}, member); err != nil {
t.Fatal(err)
}
if fresh == "" {
t.Fatal("no candidate search ran between the rename and clearing offers")
}
if !e.svc.HasOffers(c.ID) {
t.Fatalf("clearing offers after the rename dropped the candidates found at its revision for dinner %s", c.ID)
}
wantStaleCandidate(t, e, c.ID, old)
e.model.QueueGenerate(recipestest.NormalizedReply("oven", 150))
g, err := e.svc.Ground(t.Context(), c.ID, dinners.Fix{CandidateID: ptr(fresh)}, member)
if err != nil {
t.Fatal(err)
}
if g.GroundingStatus != dinners.GroundingGrounded || g.RecipeID == nil {
t.Fatalf("dinner = %+v, want grounded for the new name", g)
}
rec, err := db.New(e.st.Reader).GetRecipe(t.Context(), *g.RecipeID)
if err != nil {
t.Fatal(err)
}
if rec.SourceUrl != stew {
t.Errorf("recipe source = %s, want %s", rec.SourceUrl, stew)
}
}
// TestRenameBackDuringCandidatesRefusesThem: a candidate search that read
// a dinner as A and finishes after renames A -> B -> A have committed, but
// before either has cleared the dinner's offers, stores candidates that
// do not ground it, though its name is A again.
func TestRenameBackDuringCandidatesRefusesThem(t *testing.T) {
e := newEnv(t)
id := failedDinner(t, e, "Mystery curry")
e.search.Results["Mystery curry"] = []recipes.Candidate{{URL: curryA}}
asked := len(e.search.Queries())
gate := make(chan struct{})
e.search.Gate = gate
type result struct {
got []dinners.Candidate
err error
}
done := make(chan result, 1)
go func() {
got, err := e.svc.Candidates(t.Context(), id)
done <- result{got, err}
}()
deadline := time.Now().Add(10 * time.Second)
for len(e.search.Queries()) == asked {
if time.Now().After(deadline) {
t.Fatal("candidate search never started")
}
time.Sleep(5 * time.Millisecond)
}
// The first rename's hook makes the second; the second's lets the
// search finish and tries its candidate, before either clears offers.
var hooks int
var stale error
var groundedBy *dinners.Dinner
dinners.SetUpdateCommitted(t, func() {
hooks++
switch hooks {
case 1:
if _, err := e.svc.Update(t.Context(), id, dinners.Patch{Name: ptr("Mystery curry")}, member); err != nil {
t.Error(err)
}
case 2:
close(gate)
r := <-done
if r.err != nil || len(r.got) != 1 {
t.Errorf("candidates = %+v, %v; want curryA", r.got, r.err)
return
}
e.model.QueueGenerate(recipestest.NormalizedReply("stovetop", 40))
d, err := e.svc.Ground(t.Context(), id, dinners.Fix{CandidateID: ptr(r.got[0].CandidateID)}, member)
if err == nil {
groundedBy = &d
}
stale = err
}
})
if _, err := e.svc.Update(t.Context(), id, dinners.Patch{Name: ptr("Beef stew")}, member); err != nil {
t.Fatal(err)
}
if hooks != 2 {
t.Fatalf("renames committed = %d, want 2", hooks)
}
var invalid *dinners.ValidationError
if !errors.As(stale, &invalid) {
t.Fatalf("stale candidate: err = %v, grounded %+v; want *ValidationError", stale, groundedBy)
}
row := e.dinner(t, id)
if row.Name != "Mystery curry" || row.RecipeID.Valid || row.GroundingStatus != dinners.GroundingFailed {
t.Errorf("dinner = %+v, want Mystery curry still failed with no recipe", row)
}
if e.svc.HasOffers(id) {
t.Errorf("service still holds the stale candidates of dinner %s", id)
}
}
// TestCandidatesRunningWhenCleared: a candidate search still running when
// its dinner is grounded by hand or archived stores none of the candidates
// it returns, so none of them grounds the dinner.
// its dinner is grounded by hand, renamed or archived stores none of the
// candidates it returns, so none of them grounds the dinner.
func TestCandidatesRunningWhenCleared(t *testing.T) {
t.Parallel()
tests := []struct {
@@ -877,6 +1100,11 @@ func TestCandidatesRunningWhenCleared(t *testing.T) {
t.Fatal(err)
}
}, false},
{"rename", func(t *testing.T, e *env, id string) {
if _, err := e.svc.Update(t.Context(), id, dinners.Patch{Name: ptr("Beef stew")}, member); err != nil {
t.Fatal(err)
}
}, false},
{"archive", func(t *testing.T, e *env, id string) {
if _, err := e.svc.Archive(t.Context(), id, member); err != nil {
t.Fatal(err)