Evict expired candidate offers on a timer
Each dinner's offered candidates now carry a time.AfterFunc expiry that drops them from the service once offerLifetime passes, so token-bearing candidate URLs are no longer held until some later search sweeps them. Storing new offers or clearing a dinner's offers stops the old timer; the timer drops only the set it was armed for. The expiry runs no goroutine until it fires and touches only the in-memory map. TestCandidatesExpire now waits for the eviction with no further search and checks the candidate no longer grounds; TestExpiredCandidatesNotHeld checks no dinner's expired offers remain held. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -151,7 +151,7 @@ type Service struct {
|
|||||||
// offerSearches holds the Candidates searches running, by dinner id,
|
// 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 theirs.
|
||||||
offersMu sync.Mutex
|
offersMu sync.Mutex
|
||||||
offers map[string]offerSet
|
offers map[string]*offerSet
|
||||||
offerSearches map[string]map[*offerSearch]struct{}
|
offerSearches map[string]map[*offerSearch]struct{}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -169,7 +169,7 @@ func NewService(st *store.Store, grounder *recipes.Grounder, runner *jobs.Runner
|
|||||||
case logger == nil:
|
case logger == nil:
|
||||||
return nil, errors.New("dinners: nil logger")
|
return nil, errors.New("dinners: nil logger")
|
||||||
}
|
}
|
||||||
s := &Service{st: st, grounder: grounder, logger: logger, groundings: make(map[string]map[*runningGrounding]struct{}), offerKey: make([]byte, 32), offers: make(map[string]offerSet), offerSearches: make(map[string]map[*offerSearch]struct{})}
|
s := &Service{st: st, grounder: grounder, logger: logger, groundings: make(map[string]map[*runningGrounding]struct{}), offerKey: make([]byte, 32), offers: make(map[string]*offerSet), offerSearches: make(map[string]map[*offerSearch]struct{})}
|
||||||
rand.Read(s.offerKey) // never fails: it crashes the program instead
|
rand.Read(s.offerKey) // never fails: it crashes the program instead
|
||||||
kind, err := jobs.Register(runner, KindGroundDinner, s.runGroundJob, Reasons...)
|
kind, err := jobs.Register(runner, KindGroundDinner, s.runGroundJob, Reasons...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -31,6 +31,13 @@ func (s *Service) HasOffers(id string) bool {
|
|||||||
return ok
|
return ok
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// OfferCount is the number of dinners whose candidates the service holds.
|
||||||
|
func (s *Service) OfferCount() int {
|
||||||
|
s.offersMu.Lock()
|
||||||
|
defer s.offersMu.Unlock()
|
||||||
|
return len(s.offers)
|
||||||
|
}
|
||||||
|
|
||||||
// RunGroundJob runs the KindGroundDinner handler for p in the caller's
|
// RunGroundJob runs the KindGroundDinner handler for p in the caller's
|
||||||
// goroutine, as the job runner would, and returns the grounding status it
|
// goroutine, as the job runner would, and returns the grounding status it
|
||||||
// reports.
|
// reports.
|
||||||
|
|||||||
@@ -301,10 +301,11 @@ type Candidate struct {
|
|||||||
var offerLifetime = 30 * time.Minute
|
var offerLifetime = 30 * time.Minute
|
||||||
|
|
||||||
// offerSet is the candidates last offered for a dinner: the page URL of
|
// offerSet is the candidates last offered for a dinner: the page URL of
|
||||||
// each candidate id, until expires.
|
// each candidate id. expiry drops the set from offers once offerLifetime
|
||||||
|
// has passed.
|
||||||
type offerSet struct {
|
type offerSet struct {
|
||||||
urls map[string]string
|
urls map[string]string
|
||||||
expires time.Time
|
expiry *time.Timer
|
||||||
}
|
}
|
||||||
|
|
||||||
// offerSearch is a Candidates search running for a dinner. It is stale
|
// offerSearch is a Candidates search running for a dinner. It is stale
|
||||||
@@ -350,9 +351,8 @@ func (s *Service) Candidates(ctx context.Context, id string) ([]Candidate, error
|
|||||||
}
|
}
|
||||||
|
|
||||||
// trackOffers registers a Candidates search for dinner id. It returns keep,
|
// trackOffers registers a Candidates search for dinner id. It returns keep,
|
||||||
// which stores urls as the dinner's offers unless clearOffers has run for
|
// which stores urls as the dinner's offers for offerLifetime unless
|
||||||
// it since, and done, to call when the search ends. keep also drops every
|
// clearOffers has run for it since, and done, to call when the search ends.
|
||||||
// dinner's expired offers, so offers holds only those still live.
|
|
||||||
func (s *Service) trackOffers(id string) (keep func(urls map[string]string), done func()) {
|
func (s *Service) trackOffers(id string) (keep func(urls map[string]string), done func()) {
|
||||||
search := &offerSearch{}
|
search := &offerSearch{}
|
||||||
s.offersMu.Lock()
|
s.offersMu.Lock()
|
||||||
@@ -362,17 +362,23 @@ func (s *Service) trackOffers(id string) (keep func(urls map[string]string), don
|
|||||||
s.offerSearches[id][search] = struct{}{}
|
s.offerSearches[id][search] = struct{}{}
|
||||||
s.offersMu.Unlock()
|
s.offersMu.Unlock()
|
||||||
keep = func(urls map[string]string) {
|
keep = func(urls map[string]string) {
|
||||||
now := time.Now()
|
|
||||||
s.offersMu.Lock()
|
s.offersMu.Lock()
|
||||||
defer s.offersMu.Unlock()
|
defer s.offersMu.Unlock()
|
||||||
for dinnerID, set := range s.offers {
|
if search.stale {
|
||||||
if !now.Before(set.expires) {
|
return
|
||||||
delete(s.offers, dinnerID)
|
}
|
||||||
|
s.dropOffers(id)
|
||||||
|
set := &offerSet{urls: urls}
|
||||||
|
// 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() {
|
||||||
|
s.offersMu.Lock()
|
||||||
|
defer s.offersMu.Unlock()
|
||||||
|
if s.offers[id] == set {
|
||||||
|
delete(s.offers, id)
|
||||||
}
|
}
|
||||||
}
|
})
|
||||||
if !search.stale {
|
s.offers[id] = set
|
||||||
s.offers[id] = offerSet{urls: urls, expires: now.Add(offerLifetime)}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
done = func() {
|
done = func() {
|
||||||
s.offersMu.Lock()
|
s.offersMu.Lock()
|
||||||
@@ -391,12 +397,21 @@ func (s *Service) trackOffers(id string) (keep func(urls map[string]string), don
|
|||||||
func (s *Service) clearOffers(id string) {
|
func (s *Service) clearOffers(id string) {
|
||||||
s.offersMu.Lock()
|
s.offersMu.Lock()
|
||||||
defer s.offersMu.Unlock()
|
defer s.offersMu.Unlock()
|
||||||
delete(s.offers, id)
|
s.dropOffers(id)
|
||||||
for search := range s.offerSearches[id] {
|
for search := range s.offerSearches[id] {
|
||||||
search.stale = true
|
search.stale = true
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// dropOffers drops dinner id's offers and stops their expiry. The caller
|
||||||
|
// holds offersMu.
|
||||||
|
func (s *Service) dropOffers(id string) {
|
||||||
|
if set := s.offers[id]; set != nil {
|
||||||
|
set.expiry.Stop()
|
||||||
|
delete(s.offers, id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// candidateID is the opaque id of the page at rawURL as a candidate for
|
// candidateID is the opaque id of the page at rawURL as a candidate for
|
||||||
// dinner id: the same page is offered under the same id until the service
|
// dinner id: the same page is offered under the same id until the service
|
||||||
// restarts, and the id reveals nothing of the URL.
|
// restarts, and the id reveals nothing of the URL.
|
||||||
@@ -410,11 +425,14 @@ func (s *Service) candidateID(id, rawURL string) string {
|
|||||||
// A cid not among the dinner's live offers, or whose page is not a valid
|
// A cid not among the dinner's live offers, or whose page is not a valid
|
||||||
// recipe link, is a *ValidationError.
|
// recipe link, is a *ValidationError.
|
||||||
func (s *Service) candidateURL(id, cid string) (string, error) {
|
func (s *Service) candidateURL(id, cid string) (string, error) {
|
||||||
|
var raw string
|
||||||
|
var ok bool
|
||||||
s.offersMu.Lock()
|
s.offersMu.Lock()
|
||||||
set := s.offers[id]
|
if set := s.offers[id]; set != nil {
|
||||||
raw, ok := set.urls[cid]
|
raw, ok = set.urls[cid]
|
||||||
|
}
|
||||||
s.offersMu.Unlock()
|
s.offersMu.Unlock()
|
||||||
if ok && time.Now().Before(set.expires) {
|
if ok {
|
||||||
if u, err := checkRecipeURL(raw); err == nil {
|
if u, err := checkRecipeURL(raw); err == nil {
|
||||||
return u.String(), nil
|
return u.String(), nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -928,24 +928,47 @@ func TestCandidatesRunningWhenCleared(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// TestCandidatesExpire: candidates ground a dinner only for their
|
// TestCandidatesExpire: once their lifetime passes, a dinner's candidates
|
||||||
// lifetime; storing another dinner's candidates drops the expired ones.
|
// are dropped and no longer ground it, with no other search running.
|
||||||
func TestCandidatesExpire(t *testing.T) {
|
func TestCandidatesExpire(t *testing.T) {
|
||||||
dinners.SetOfferLifetime(t, 0)
|
dinners.SetOfferLifetime(t, 0)
|
||||||
e := newEnv(t)
|
e := newEnv(t)
|
||||||
id := failedDinner(t, e, "Mystery curry")
|
id := failedDinner(t, e, "Mystery curry")
|
||||||
other := failedDinner(t, e, "Other curry")
|
|
||||||
e.search.Results["Mystery curry"] = []recipes.Candidate{{URL: curryA}}
|
e.search.Results["Mystery curry"] = []recipes.Candidate{{URL: curryA}}
|
||||||
e.search.Results["Other curry"] = []recipes.Candidate{{URL: curryB}}
|
|
||||||
cid := candidateID(t, e, id, "Chicken Curry")
|
cid := candidateID(t, e, id, "Chicken Curry")
|
||||||
|
|
||||||
|
waitNoOffers(t, e)
|
||||||
wantStaleCandidate(t, e, id, cid)
|
wantStaleCandidate(t, e, id, cid)
|
||||||
if !e.svc.HasOffers(id) {
|
}
|
||||||
t.Fatalf("candidates of %s not held before another search", id)
|
|
||||||
|
// TestExpiredCandidatesNotHeld: the service holds no dinner's candidates
|
||||||
|
// past their lifetime, however many dinners were offered some.
|
||||||
|
func TestExpiredCandidatesNotHeld(t *testing.T) {
|
||||||
|
dinners.SetOfferLifetime(t, 50*time.Millisecond)
|
||||||
|
e := newEnv(t)
|
||||||
|
names := map[string]string{"Mystery curry": curryA, "Other curry": curryB}
|
||||||
|
ids := make([]string, 0, len(names))
|
||||||
|
for name, u := range names {
|
||||||
|
ids = append(ids, failedDinner(t, e, name))
|
||||||
|
e.search.Results[name] = []recipes.Candidate{{URL: u}}
|
||||||
}
|
}
|
||||||
candidateID(t, e, other, "Easy Weeknight Curry")
|
for _, id := range ids {
|
||||||
if e.svc.HasOffers(id) {
|
if _, err := e.svc.Candidates(t.Context(), id); err != nil {
|
||||||
t.Errorf("expired candidates of %s still held after another dinner's search", id)
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
waitNoOffers(t, e)
|
||||||
|
}
|
||||||
|
|
||||||
|
// waitNoOffers waits for the service to hold no dinner's candidates.
|
||||||
|
func waitNoOffers(t *testing.T, e *env) {
|
||||||
|
t.Helper()
|
||||||
|
deadline := time.Now().Add(5 * time.Second)
|
||||||
|
for n := e.svc.OfferCount(); n > 0; n = e.svc.OfferCount() {
|
||||||
|
if time.Now().After(deadline) {
|
||||||
|
t.Fatalf("service still holds the expired candidates of %d dinners", n)
|
||||||
|
}
|
||||||
|
time.Sleep(5 * time.Millisecond)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user