Merge pull request 'factEnrichmentWorker walks the pending queue twice per tick to write one log line' (#191) from task/647-factenrichmentworker-walks-the-pending-q into master

cmd/mavend/factenrichment.go

@@ -37,8 +37,8 @@ type factEnrichmentWorker struct {
 	nextTry map[int64]time.Time // fact id → earliest retry
 }
 
-// enrichmentScanLimit bounds how deep a single tick (or status report) walks
+// enrichmentScanLimit bounds how deep a single tick walks the pending queue
-// the pending queue looking for facts whose backoff has elapsed. The queue is
+// 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.
@@ -75,8 +75,8 @@ func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval tim
 // 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
+// All three numbers describe the same set of rows, whatever is still pending
-// enrichmentScanLimit pending facts. Counting Pending over a thousand rows
+// out of the first enrichmentScanLimit 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 {
@@ -86,13 +86,22 @@ type enrichmentStatus struct {
 	Scanned     int // rows the other three counts were taken over
 }
 
+// status reads the queue and counts over it. For a caller with no batch in
+// hand — anything asking the worker how it is doing from outside the tick.
 func (w *factEnrichmentWorker) status(ctx context.Context) enrichmentStatus {
-	var st enrichmentStatus
 	pending, err := w.store.PendingFactResolutions(ctx, enrichmentScanLimit)
 	if err != nil {
 		log.Printf("factenrichment: status: %v", err)
-		return st
+		return enrichmentStatus{}
 	}
+	return w.statusOf(pending)
+}
+
+// statusOf counts over a batch the caller already has. The batch is the query
+// the tick already ran, so reporting the backlog costs no second read of the
+// scan limit — up to a thousand rows, on a database that serialises them.
+func (w *factEnrichmentWorker) statusOf(pending []store.Fact) enrichmentStatus {
+	var st enrichmentStatus
 	st.Pending = len(pending)
 	st.Scanned = len(pending)
 	w.mu.Lock()
@@ -144,17 +153,24 @@ func (w *factEnrichmentWorker) tick(ctx context.Context) {
 	}
 	w.forgetDeparted(pending)
 	skipped, failed, attempted := 0, 0, 0
+	// A resolved fact leaves the pending queue, so the batch in hand overstates
+	// the backlog by however many succeeded. Drop them here rather than
+	// re-reading the queue to find out.
+	remaining := make([]store.Fact, 0, len(pending))
 	for _, f := range pending {
 		if attempted >= w.batch {
-			break
+			remaining = append(remaining, f)
+			continue
 		}
 		if !w.due(f.ID) {
 			skipped++
+			remaining = append(remaining, f)
 			continue
 		}
 		attempted++
 		if !w.resolveOne(ctx, f) {
 			failed++
+			remaining = append(remaining, f)
 		}
 	}
 	if failed > 0 {
@@ -164,7 +180,7 @@ func (w *factEnrichmentWorker) tick(ctx context.Context) {
 	// 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 {
+	if st := w.statusOf(remaining); 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)
 	}