fix(lifecycle): debounce bucket processor scan-end to prevent false idle signals

The drain callback on _internalTaskScheduler fires every time the
queue goes from non-empty to empty. Under light load this happens
between Kafka message batches within the same conductor scan,
prematurely resetting the scan start time gauge to 0 and setting
_processorScanMetricsActive to false.

Because subsequent messages carry the same conductorScanId, the
new-scan branch is skipped and onBucketProcessorScanStart is never
called again, leaving the Bucket Processor Scan Progress dashboard
panel hidden for the remainder of the scan.

Fix: replace the immediate reset with a 30-second debounced timeout.
If new messages from the same scan arrive during the debounce window
the timeout is cancelled, keeping the scan active. A new scan also
cancels any pending debounce before performing a full metric reset.

Issue: BB-740

Author: benzekrimaha

extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js

@@ -31,6 +31,11 @@ const locations = require('../../../conf/locationConfig.json');
 
 const PROCESS_OBJECTS_ACTION = 'processObjects';
 
+// Delay before declaring a scan finished after the internal task
+// queue drains.  This prevents premature idle signals when the
+// queue empties between Kafka message batches within the same scan.
+const SCAN_END_DEBOUNCE_DELAY_MS = 30000;
+
 /**
  * @class LifecycleBucketProcessor
  *
@@ -102,6 +107,7 @@ class LifecycleBucketProcessor {
         // Kafka batches from the same scan do not retrigger onBucketProcessorScanStart.
         this._metricsScanId = null;
         this._processorScanMetricsActive = false;
+        this._scanEndTimeout = null;
 
         this.clientManager = new ClientManager({
             id: 'lifecycle',
@@ -151,22 +157,31 @@ class LifecycleBucketProcessor {
         }, this._lcConfig.bucketProcessor.concurrency);
 
         // Listen for errors from any task being processed.
-        // When the internal task queue empties, also reset the
-        // scan-in-progress gauge so Prometheus can detect that
-        // this processor is idle.
+        // When the internal task queue empties, schedule a delayed
+        // reset of the scan-in-progress gauge.  The delay avoids
+        // false idle signals when the queue drains between Kafka
+        // message batches that belong to the same conductor scan.
         this._internalTaskScheduler.drain(err => {
             if (err) {
                 this._log.error('error occurred during task processing', {
                     error: err,
                 });
             }
-            if (this._processorScanMetricsActive) {
-                this._log.info('bucket processor internal task queue drained', {
+            if (this._processorScanMetricsActive && !this._scanEndTimeout) {
+                this._log.info('bucket processor task queue drained, scheduling scan end', {
                     method: 'LifecycleBucketProcessor.drain',
                     conductorScanId: this._metricsScanId,
+                    debounceMs: SCAN_END_DEBOUNCE_DELAY_MS,
                 });
-                LifecycleMetrics.onBucketProcessorScanEnd(this._log);
-                this._processorScanMetricsActive = false;
+                this._scanEndTimeout = setTimeout(() => {
+                    this._scanEndTimeout = null;
+                    this._log.info('bucket processor scan end confirmed', {
+                        method: 'LifecycleBucketProcessor.drain',
+                        conductorScanId: this._metricsScanId,
+                    });
+                    LifecycleMetrics.onBucketProcessorScanEnd(this._log);
+                    this._processorScanMetricsActive = false;
+                }, SCAN_END_DEBOUNCE_DELAY_MS);
             }
         });
 
@@ -296,6 +311,11 @@ class LifecycleBucketProcessor {
         const { bucket, owner, accountId, taskVersion } = result.target;
         const conductorScanId = result.contextInfo && result.contextInfo.conductorScanId;
         if (conductorScanId && conductorScanId !== this._metricsScanId) {
+            // New scan: cancel any pending debounce, full metric reset
+            if (this._scanEndTimeout) {
+                clearTimeout(this._scanEndTimeout);
+                this._scanEndTimeout = null;
+            }
             this._metricsScanId = conductorScanId;
             LifecycleMetrics.onBucketProcessorScanStart(
                 this._log, Date.now()
@@ -306,6 +326,15 @@ class LifecycleBucketProcessor {
                     'LifecycleBucketProcessor._processBucketEntry',
                 conductorScanId,
             });
+        } else if (conductorScanId && this._scanEndTimeout) {
+            // Same scan, but the debounce was pending: cancel it
+            // because more messages are still arriving.
+            clearTimeout(this._scanEndTimeout);
+            this._scanEndTimeout = null;
+            this._log.debug('cancelled scan end debounce, more messages arriving', {
+                method: 'LifecycleBucketProcessor._processBucketEntry',
+                conductorScanId,
+            });
         }
 
         if (!bucket || !owner || (!accountId && this._authConfig.type === authTypeAssumeRole)) {