diff --git a/cmd/mavend/factenrichment.go b/cmd/mavend/factenrichment.go index a19ea55..835e630 100644 --- a/cmd/mavend/factenrichment.go +++ b/cmd/mavend/factenrichment.go @@ -35,9 +35,15 @@ type factEnrichmentWorker struct { mu sync.Mutex attempt map[int64]int // fact id → consecutive failures nextTry map[int64]time.Time // fact id → earliest retry - skipped int // facts held back by backoff on the last tick } +// enrichmentScanLimit bounds how deep a single tick (or status report) walks +// the pending queue looking for facts whose backoff has elapsed. The queue is +// ordered by id, so without a scan the oldest facts hold every batch slot +// whether or not they are eligible, and one permanently failing fact stalls +// every younger one behind it. +const enrichmentScanLimit = 1000 + // enrichmentBackoff is the wait before retrying a fact after n consecutive // failures, capped so a long Nexus outage still retries about hourly. func enrichmentBackoff(n int) time.Duration { @@ -64,26 +70,39 @@ func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval tim } // enrichmentStatus is what the worker reports about its own health: how many -// facts are waiting, how many are currently in backoff, and the worst retry -// count seen. Degradation is reported, never hidden — a Nexus that has been -// down all day must be visible as a backlog, not as facts that silently -// never got tagged. +// facts are waiting, how many of those are currently in backoff, and the worst +// retry count among them. Degradation is reported, never hidden — a Nexus that +// has been down all day must be visible as a backlog, not as facts that +// silently never got tagged. +// +// All three numbers describe the same set of rows, the first +// enrichmentScanLimit pending facts. Counting Pending over a thousand rows +// while counting InBackoff over the twenty that reached the head of a batch +// described two different populations under one struct. type enrichmentStatus struct { Pending int InBackoff int MaxAttempts int + Scanned int // rows the other three counts were taken over } func (w *factEnrichmentWorker) status(ctx context.Context) enrichmentStatus { var st enrichmentStatus - if pending, err := w.store.PendingFactResolutions(ctx, 1000); err == nil { - st.Pending = len(pending) + pending, err := w.store.PendingFactResolutions(ctx, enrichmentScanLimit) + if err != nil { + log.Printf("factenrichment: status: %v", err) + return st } + st.Pending = len(pending) + st.Scanned = len(pending) w.mu.Lock() defer w.mu.Unlock() - st.InBackoff = w.skipped - for _, n := range w.attempt { - if n > st.MaxAttempts { + now := w.now() + for _, f := range pending { + if next, ok := w.nextTry[f.ID]; ok && now.Before(next) { + st.InBackoff++ + } + if n := w.attempt[f.ID]; n > st.MaxAttempts { st.MaxAttempts = n } } @@ -112,27 +131,65 @@ func (w *factEnrichmentWorker) run(ctx context.Context) { } func (w *factEnrichmentWorker) tick(ctx context.Context) { - pending, err := w.store.PendingFactResolutions(ctx, w.batch) + // Scan past the facts that are still in backoff instead of letting them + // occupy the batch. The queue is ordered by id, so the oldest facts are + // pulled first whether or not they are eligible: twenty facts Nexus keeps + // rejecting would otherwise hold every slot forever and enrichment would + // stop with no error and no log line, because a tick that skips everything + // fails nothing. + pending, err := w.store.PendingFactResolutions(ctx, enrichmentScanLimit) if err != nil { log.Printf("factenrichment: list pending: %v", err) return } - skipped, failed := 0, 0 + w.forgetDeparted(pending) + skipped, failed, attempted := 0, 0, 0 for _, f := range pending { + if attempted >= w.batch { + break + } if !w.due(f.ID) { skipped++ continue } + attempted++ if !w.resolveOne(ctx, f) { failed++ } } - w.mu.Lock() - w.skipped = skipped - w.mu.Unlock() if failed > 0 { log.Printf("factenrichment: %d/%d resolutions failed this tick, %d held in backoff", - failed, len(pending), skipped) + failed, attempted, skipped) + } + // Report the backlog every tick, not only when something failed: the + // stalled state worth seeing is the one where nothing failed because + // nothing was attempted. + if st := w.status(ctx); st.Pending > 0 { + log.Printf("factenrichment: %d facts pending entity resolution, %d in backoff, worst attempt %d (scanned %d)", + st.Pending, st.InBackoff, st.MaxAttempts, st.Scanned) + } +} + +// forgetDeparted drops retry state for facts that are no longer pending. A +// fact can leave the queue without ever resolving here — voided, or resolved +// by a later write — and its entries would otherwise live as long as the +// process does. +func (w *factEnrichmentWorker) forgetDeparted(pending []store.Fact) { + live := make(map[int64]struct{}, len(pending)) + for _, f := range pending { + live[f.ID] = struct{}{} + } + w.mu.Lock() + defer w.mu.Unlock() + for id := range w.attempt { + if _, ok := live[id]; !ok { + delete(w.attempt, id) + } + } + for id := range w.nextTry { + if _, ok := live[id]; !ok { + delete(w.nextTry, id) + } } } @@ -150,17 +207,11 @@ func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) boo entityID, _, ambiguous, err := w.eco.resolveEntityReference(ctx, f.Subject, nil) if err != nil { // Transient (Nexus unreachable) — leave pending, back off, retry later. - log.Printf("factenrichment: resolve fact %d subject %q: %v", f.ID, f.Subject, err) - w.mu.Lock() - w.attempt[f.ID]++ - w.nextTry[f.ID] = w.now().Add(enrichmentBackoff(w.attempt[f.ID])) - w.mu.Unlock() + // The subject is his words: log its length, the way the trace does. + log.Printf("factenrichment: resolve fact %d subject %s: %v", f.ID, redactSubject(f.Subject), err) + w.backOff(f.ID) return false } - w.mu.Lock() - delete(w.attempt, f.ID) - delete(w.nextTry, f.ID) - w.mu.Unlock() state := store.ResolutionNotFound switch { case entityID != "": @@ -169,8 +220,25 @@ func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) boo state = store.ResolutionAmbiguous } if err := w.store.ResolveFactEntity(ctx, f.ID, entityID, state); err != nil { + // A failed write leaves the fact pending exactly like a failed resolve + // does, so it gets the same pacing. Clearing the counters first meant + // this one path retried every tick, at full rate, with no ceiling. log.Printf("factenrichment: record resolution for fact %d: %v", f.ID, err) + w.backOff(f.ID) return false } + w.mu.Lock() + delete(w.attempt, f.ID) + delete(w.nextTry, f.ID) + w.mu.Unlock() return true } + +// backOff records one more consecutive failure for a fact and pushes its next +// attempt out accordingly. +func (w *factEnrichmentWorker) backOff(id int64) { + w.mu.Lock() + defer w.mu.Unlock() + w.attempt[id]++ + w.nextTry[id] = w.now().Add(enrichmentBackoff(w.attempt[id])) +}