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]))
+}