analyzer/item: Do not exit immediately on empty queue

analyzer/item/item.go

@@ -32,11 +32,11 @@ const (
 var ErrEmptyBatch = errors.New("no items in batch")
 
 type itemBasedAnalyzer[Item any] struct {
-	maxBatchSize     uint64
-	stopOnEmptyQueue bool
-	fixedInterval    time.Duration
-	interItemDelay   time.Duration
-	analyzerName     string
+	maxBatchSize        uint64
+	stopIfQueueEmptyFor time.Duration
+	fixedInterval       time.Duration
+	interItemDelay      time.Duration
+	analyzerName        string
 
 	processor ItemProcessor[Item]
 
@@ -61,8 +61,8 @@ type ItemProcessor[Item any] interface {
 
 // NewAnalyzer returns a new item based analyzer using the provided item processor.
 //
-// If stopOnEmptyQueue is true, the analyzer will process batches of items until its
-// work queue is empty, at which point it will terminate and return. Likely to
+// If stopIfQueueEmptyFor is a non-zero duration, the analyzer will process batches of items until its
+// work queue is empty for `stopIfQueueEmptyFor`, at which point it will terminate and return. Likely to
 // be used in the regression tests.
 //
 // If fixedInterval is provided, the analyzer will process one batch every fixedInterval.
@@ -79,15 +79,15 @@ func NewAnalyzer[Item any](
 		cfg.BatchSize = defaultBatchSize
 	}
 	a := &itemBasedAnalyzer[Item]{
-		cfg.BatchSize,
-		cfg.StopOnEmptyQueue,
-		cfg.Interval,
-		cfg.InterItemDelay,
-		name,
-		processor,
-		target,
-		logger,
-		metrics.NewDefaultAnalysisMetrics(name),
+		maxBatchSize:        cfg.BatchSize,
+		stopIfQueueEmptyFor: cfg.StopIfQueueEmptyFor,
+		fixedInterval:       cfg.Interval,
+		interItemDelay:      cfg.InterItemDelay,
+		analyzerName:        name,
+		processor:           processor,
+		target:              target,
+		logger:              logger,
+		metrics:             metrics.NewDefaultAnalysisMetrics(name),
 	}
 
 	return a, nil
@@ -195,6 +195,7 @@ func (a *itemBasedAnalyzer[Item]) Start(ctx context.Context) {
 		)
 		return
 	}
+	mostRecentTask := time.Now()
 
 	for firstIter := true; ; firstIter = false {
 		delay := backoff.Timeout()
@@ -213,8 +214,12 @@ func (a *itemBasedAnalyzer[Item]) Start(ctx context.Context) {
 		}
 		// Update queueLength
 		queueLength, err := a.sendQueueLengthMetric(ctx)
-		if err == nil && queueLength == 0 && a.stopOnEmptyQueue {
-			a.logger.Warn("item analyzer work queue is empty; shutting down")
+		// Stop if queue has been empty for a while, and configured to do so.
+		if err == nil && queueLength == 0 && a.stopIfQueueEmptyFor != 0 && time.Since(mostRecentTask) > a.stopIfQueueEmptyFor {
+			a.logger.Warn("item analyzer work queue has been empty for a while; shutting down",
+				"queue_empty_since", mostRecentTask,
+				"queue_empty_for", time.Since(mostRecentTask),
+				"stop_if_queue_empty_for", a.stopIfQueueEmptyFor)
 			return
 		}
 		a.logger.Info("work queue length", "num_items", queueLength)
@@ -231,6 +236,7 @@ func (a *itemBasedAnalyzer[Item]) Start(ctx context.Context) {
 			backoff.Failure()
 			continue
 		}
+		mostRecentTask = time.Now()
 
 		backoff.Success()
 	}

analyzer/item/item_test.go

@@ -50,10 +50,10 @@ const (
 
 // Default item based config.
 var testItemBasedConfig = &config.ItemBasedAnalyzerConfig{
-	BatchSize:        3,
-	StopOnEmptyQueue: true,
-	Interval:         0, // use backoff
-	InterItemDelay:   0,
+	BatchSize:           3,
+	StopIfQueueEmptyFor: time.Second,
+	Interval:            0, // use backoff
+	InterItemDelay:      0,
 }
 
 type mockItem struct {

config/config.go

@@ -385,13 +385,14 @@ type ItemBasedAnalyzerConfig struct {
 	// Uses default value of 20 if unset/set to 0.
 	BatchSize uint64 `koanf:"batch_size"`
 
-	// If StopOnEmptyQueue is true, the analyzer will exit the main processing loop when
-	// there are no items left in the work queue. This is useful during testing when
-	// 1) The number of items in the queue is determinate
+	// If StopIfQueueEmptyFor is a non-zero duration, the analyzer will terminate when
+	// there are no items left in the work queue for this specified amount of time.
+	// This is useful during testing when
+	// 1) The number of items in the queue is determinate.
 	// 2) We want the analyzer to naturally terminate after processing all available work items.
 	//
-	// Defaults to false.
-	StopOnEmptyQueue bool `koanf:"stop_on_empty_queue"`
+	// Defaults to 0, i.e. the analyzer never terminates.
+	StopIfQueueEmptyFor time.Duration `koanf:"stop_if_queue_empty_for"`
 
 	// If Interval is set, the analyzer will process batches at a fixed cadence specified by Interval.
 	// Otherwise, the analyzer will use an adaptive backoff to determine the delay between

tests/e2e_regression/e2e_config_2.yml

@@ -22,7 +22,7 @@ analysis:
     # Some non-block analyzers are not tested in e2e regressions.
     # They are largely not worth the trouble as they do not interact with rest of the system much.
     # metadata_registry: {} # Awkward to inject mock registry responses.
-    # node_stats: {}        # Awkward to inject mock node response using the current paradigm (= reseponse caching).
+    # node_stats: {}        # Awkward to inject mock node response using the current paradigm (= response caching).
     # aggregate_stats: {}   # Awkward to make stop after a single run.
   storage:
     backend: postgres