From 6888b2f09930b64c1408c0135ac0d37567222bc8 Mon Sep 17 00:00:00 2001
From: Maha Benzekri <maha.benzekri@scality.com>
Date: Sun, 5 Jul 2026 21:17:42 +0200
Subject: [PATCH 1/6] Add lifecycle scan metric helpers.

Introduce conductor scan lifecycle gauges and bucket-processor scan
counters in LifecycleMetrics:

- s3_lifecycle_latest_batch_start_time is a scheduling heartbeat set
  when a scan starts; s3_lifecycle_latest_batch_end_time and
  s3_lifecycle_latest_batch_bucket_count are only set on successful
  completion, so a scan that fails after starting stays visible as
  stalled instead of looking complete.
- s3_lifecycle_bucket_processor_scan_messages_total counts bucket-task
  messages picked up, labelled by conductor_scan_id to spot concurrent
  scans; s3_lifecycle_bucket_processor_scan_message_age_seconds tracks
  the wall-clock delay between conductor scan start and message pickup
  (a backlog signal).

Per-conductor_scan_id series are removed by per-scan timers after a
fixed retention so the label cardinality stays bounded; messages from an
older conductor without a scan id are ignored so rolling upgrades do not
create synthetic series.

Issue: BB-740
---
 extensions/lifecycle/LifecycleMetrics.js      | 190 +++++++++++++++++-
 tests/unit/lifecycle/LifecycleMetrics.spec.js | 164 ++++++++++++++-
 2 files changed, 348 insertions(+), 6 deletions(-)

diff --git a/extensions/lifecycle/LifecycleMetrics.js b/extensions/lifecycle/LifecycleMetrics.js
index eac0d13467..18d746fa30 100644
--- a/extensions/lifecycle/LifecycleMetrics.js
+++ b/extensions/lifecycle/LifecycleMetrics.js
@@ -1,16 +1,39 @@
 const { ZenkoMetrics } = require('arsenal').metrics;
+const { Logger } = require('werelogs');
 
-const LIFECYCLE_LABEL_ORIGIN =  'origin';
+const lifecycleMetricsLogger = new Logger('Backbeat:LifecycleMetrics');
+
+const LIFECYCLE_LABEL_ORIGIN = 'origin';
 const LIFECYCLE_LABEL_OP = 'op';
 const LIFECYCLE_LABEL_STATUS = 'status';
 const LIFECYCLE_LABEL_LOCATION = 'location';
 const LIFECYCLE_LABEL_TYPE = 'type';
+const LIFECYCLE_LABEL_CONDUCTOR_SCAN_ID = 'conductor_scan_id';
 
 const LIFECYCLE_MARKER_METRICS_LOCATION = '-delete-marker-';
 
+// Keep per-scan series long enough for scraping and debugging recent overlap,
+// but remove them from prom-client after a configurable retention interval.
+// We intentionally do not cap the number of tracked scan IDs: if overlapping
+// scans happen, hiding older IDs would remove the signal this metric provides.
+// Prometheus retains scraped scan-id series until TSDB retention expires.
+const DEFAULT_SCAN_METRIC_RETENTION_S = 24 * 60 * 60;
+const CONDUCTOR_ORIGIN = 'conductor';
+const BUCKET_PROCESSOR_ORIGIN = 'bucket_processor';
+const SCAN_METRIC_RETENTION_MS = DEFAULT_SCAN_METRIC_RETENTION_S * 1000;
+
 const conductorLatestBatchStartTime = ZenkoMetrics.createGauge({
     name: 's3_lifecycle_latest_batch_start_time',
-    help: 'Timestamp of latest lifecycle batch start time',
+    help: 'Conductor scheduling heartbeat: ms-since-epoch timestamp of ' +
+        'the most recent scan start. Use to detect that the conductor is ' +
+        'still scheduling scans (LifecycleLateScan alert).',
+    labelNames: [LIFECYCLE_LABEL_ORIGIN],
+});
+
+const conductorLatestBatchEndTime = ZenkoMetrics.createGauge({
+    name: 's3_lifecycle_latest_batch_end_time',
+    help: 'Timestamp (ms since epoch) of the most recent lifecycle ' +
+        'conductor scan end after the scan reached bucket listing.',
     labelNames: [LIFECYCLE_LABEL_ORIGIN],
 });
 
@@ -50,6 +73,73 @@ const lifecycleLegacyTask = ZenkoMetrics.createCounter({
     labelNames: [LIFECYCLE_LABEL_ORIGIN, LIFECYCLE_LABEL_STATUS],
 });
 
+const conductorLatestBatchBucketCount = ZenkoMetrics.createGauge({
+    name: 's3_lifecycle_latest_batch_bucket_count',
+    help: 'Number of buckets listed in the latest lifecycle conductor batch',
+    labelNames: [LIFECYCLE_LABEL_ORIGIN],
+});
+
+const bucketProcessorScanMessagesReceived = ZenkoMetrics.createCounter({
+    name: 's3_lifecycle_bucket_processor_scan_messages_total',
+    help: 'Total number of bucket-task messages picked up by this bucket ' +
+        'processor, grouped by conductor scan id. Per-scan series are ' +
+        'auto-removed by a local cleanup timer (see setScanMetricTimeout) ' +
+        'to keep label cardinality bounded.',
+    labelNames: [LIFECYCLE_LABEL_ORIGIN, LIFECYCLE_LABEL_CONDUCTOR_SCAN_ID],
+});
+
+const bucketProcessorScanMessageAgeSeconds = ZenkoMetrics.createHistogram({
+    name: 's3_lifecycle_bucket_processor_scan_message_age_seconds',
+    help: 'Elapsed wall-clock time in seconds between conductor scan start ' +
+        'and bucket-tasks topic message pickup by this bucket processor. ' +
+        'This is a dequeue/backlog age signal, not bucket processing duration.',
+    labelNames: [LIFECYCLE_LABEL_ORIGIN],
+    buckets: [60, 300, 600, 1800, 3600, 7200, 14400, 28800, 43200, 86400],
+});
+
+// Map conductor scan ids to the cleanup timer that removes their per-scan
+// prom-client series after they stop receiving bucket-task messages.
+const scanMetricTimers = new Map();
+
+function removeBucketProcessorScanMetrics(conductorScanId) {
+    try {
+        bucketProcessorScanMessagesReceived.remove({
+            [LIFECYCLE_LABEL_ORIGIN]: BUCKET_PROCESSOR_ORIGIN,
+            [LIFECYCLE_LABEL_CONDUCTOR_SCAN_ID]: conductorScanId,
+        });
+    } catch (err) {
+        // Removing a non-existent series should not happen; if it does it
+        // likely signals a bad conductorScanId or a missing metric label.
+        lifecycleMetricsLogger.warn(
+            'failed to remove bucket processor scan metric',
+            { conductorScanId, error: err.message });
+    }
+}
+
+function setScanMetricTimeout(conductorScanId) {
+    const previousTimer = scanMetricTimers.get(conductorScanId);
+    if (previousTimer) {
+        clearTimeout(previousTimer);
+    }
+
+    // Reset retention on every message so an active scan remains observable.
+    // Cleanup starts only after the scan stops producing bucket-task messages.
+    const cleanupTimer = setTimeout(() => {
+        removeBucketProcessorScanMetrics(conductorScanId);
+        scanMetricTimers.delete(conductorScanId);
+    }, SCAN_METRIC_RETENTION_MS);
+    cleanupTimer.unref();
+    scanMetricTimers.set(conductorScanId, cleanupTimer);
+}
+
+function resetLifecycleScanMetricCleanupTimers() {
+    scanMetricTimers.forEach((timer, conductorScanId) => {
+        clearTimeout(timer);
+        removeBucketProcessorScanMetrics(conductorScanId);
+    });
+    scanMetricTimers.clear();
+}
+
 const lifecycleS3Operations = ZenkoMetrics.createCounter({
     name: 's3_lifecycle_s3_operations_total',
     help: 'Total number of S3 operations by the lifecycle processes',
@@ -113,11 +203,23 @@ class LifecycleMetrics {
         }
     }
 
-    static onProcessBuckets(log) {
+    /**
+     * Update the conductor scheduling heartbeat. Called at the start of
+     * every conductor scan; consumed by the LifecycleLateScan alert to
+     * detect that the conductor has stopped scheduling.
+     *
+     * @param {Object} log - logger
+     * @param {number} scanStartTimestamp - scan start timestamp in ms
+     */
+    static onProcessBuckets(log, scanStartTimestamp = Date.now()) {
         try {
-            conductorLatestBatchStartTime.set({ origin: 'conductor' }, Date.now());
+            conductorLatestBatchStartTime.set(
+                { [LIFECYCLE_LABEL_ORIGIN]: CONDUCTOR_ORIGIN },
+                scanStartTimestamp);
         } catch (err) {
-            LifecycleMetrics.handleError(log, err, 'LifecycleMetrics.onProcessBuckets');
+            LifecycleMetrics.handleError(log, err, 'LifecycleMetrics.onProcessBuckets', {
+                scanStartTimestamp,
+            });
         }
     }
 
@@ -172,6 +274,82 @@ class LifecycleMetrics {
         }
     }
 
+    /**
+     * Record metrics at the end of a full conductor scan.
+     * @param {Object} log - logger
+     * @param {number} bucketCount - total buckets listed
+     */
+    static onConductorScanComplete(log, bucketCount) {
+        try {
+            const endTimestamp = Date.now();
+            conductorLatestBatchEndTime.set({
+                [LIFECYCLE_LABEL_ORIGIN]: CONDUCTOR_ORIGIN,
+            }, endTimestamp);
+            conductorLatestBatchBucketCount.set({
+                [LIFECYCLE_LABEL_ORIGIN]: CONDUCTOR_ORIGIN,
+            }, bucketCount);
+        } catch (err) {
+            LifecycleMetrics.handleError(
+                log, err, 'LifecycleMetrics.onConductorScanComplete', {
+                    bucketCount,
+                }
+            );
+        }
+    }
+
+    /**
+     * Increment the count of bucket-tasks topic messages picked up by this
+     * bucket processor for a specific conductor scan. Called before the task
+     * is dispatched to the scheduler, once per Kafka message regardless of how
+     * many objects it covers or whether processing eventually succeeds.
+     *
+     * Note: this counts messages (initial + continuation/listing slices),
+     * not unique buckets. Keep one time series per conductor_scan_id so that
+     * overlapping scans remain visible. Old scan series are removed by a
+     * timer after a fixed retention interval without
+     * update to avoid unbounded prom-client memory growth.
+     *
+     * @param {Object} log - logger
+     * @param {string} conductorScanId - conductor scan id from contextInfo
+     * @param {number} [conductorScanStartTimestamp] - conductor scan start
+     *   timestamp from contextInfo
+     */
+    static onBucketProcessorScanMessageReceived(
+        log, conductorScanId, conductorScanStartTimestamp) {
+        // Old conductor messages produced during rolling upgrades do not have
+        // a scan id. Do not create a synthetic "undefined" scan-id series.
+        if (!conductorScanId) {
+            return;
+        }
+        try {
+            bucketProcessorScanMessagesReceived.inc({
+                [LIFECYCLE_LABEL_ORIGIN]: BUCKET_PROCESSOR_ORIGIN,
+                [LIFECYCLE_LABEL_CONDUCTOR_SCAN_ID]: conductorScanId,
+            });
+            setScanMetricTimeout(conductorScanId);
+            // A negative or non-finite age means the conductor scan-start
+            // timestamp is missing or the producer/consumer clocks are skewed.
+            // Skip the observation and warn instead of recording a misleading
+            // 0 in the first histogram bucket.
+            const ageSeconds = (Date.now() - conductorScanStartTimestamp) / 1000;
+            if (Number.isFinite(ageSeconds) && ageSeconds >= 0) {
+                bucketProcessorScanMessageAgeSeconds.observe({
+                    [LIFECYCLE_LABEL_ORIGIN]: BUCKET_PROCESSOR_ORIGIN,
+                }, ageSeconds);
+            } else {
+                lifecycleMetricsLogger.warn(
+                    'skipping bucket processor scan message age observation',
+                    { conductorScanId, conductorScanStartTimestamp, ageSeconds });
+            }
+        } catch (err) {
+            LifecycleMetrics.handleError(
+                log, err,
+                'LifecycleMetrics.onBucketProcessorScanMessageReceived',
+                { conductorScanId, conductorScanStartTimestamp }
+            );
+        }
+    }
+
     static onLifecycleTriggered(log, process, type, location, latencyMs) {
         try {
             lifecycleTriggerLatency.observe({
@@ -247,6 +425,8 @@ class LifecycleMetrics {
 }
 
 module.exports = {
+    DEFAULT_SCAN_METRIC_RETENTION_S,
     LifecycleMetrics,
     LIFECYCLE_MARKER_METRICS_LOCATION,
+    resetLifecycleScanMetricCleanupTimers,
 };
diff --git a/tests/unit/lifecycle/LifecycleMetrics.spec.js b/tests/unit/lifecycle/LifecycleMetrics.spec.js
index 2acdba2490..2a6a973848 100644
--- a/tests/unit/lifecycle/LifecycleMetrics.spec.js
+++ b/tests/unit/lifecycle/LifecycleMetrics.spec.js
@@ -1,6 +1,9 @@
 const assert = require('assert');
 const sinon = require('sinon');
-const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics');
+const {
+    LifecycleMetrics,
+    resetLifecycleScanMetricCleanupTimers,
+} = require('../../../extensions/lifecycle/LifecycleMetrics');
 const { ZenkoMetrics } = require('arsenal').metrics;
 
 describe('LifecycleMetrics', () => {
@@ -13,6 +16,7 @@ describe('LifecycleMetrics', () => {
     });
 
     afterEach(() => {
+        resetLifecycleScanMetricCleanupTimers();
         sinon.restore();
     });
 
@@ -79,6 +83,133 @@ describe('LifecycleMetrics', () => {
             }));
         });
 
+        it('should catch errors in onConductorScanComplete', () => {
+            const metric = ZenkoMetrics.getMetric('s3_lifecycle_latest_batch_end_time');
+            sinon.stub(metric, 'set').throws(new Error('Metric error'));
+
+            LifecycleMetrics.onConductorScanComplete(log, 10);
+
+            assert(log.error.calledOnce);
+            assert(log.error.calledWithMatch('failed to update prometheus metrics', {
+                method: 'LifecycleMetrics.onConductorScanComplete',
+                bucketCount: 10,
+            }));
+        });
+
+        it('should increment bucket processor messages received counter by scan id', () => {
+            const messagesMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total');
+            const messagesInc = sinon.stub(messagesMetric, 'inc');
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, 'scan-A');
+            assert(messagesInc.calledOnce);
+            assert(messagesInc.calledWithMatch({
+                origin: 'bucket_processor',
+                ['conductor_scan_id']: 'scan-A',
+            }));
+
+            assert(log.error.notCalled);
+        });
+
+        it('should observe bucket processor scan message age from conductor start timestamp', () => {
+            const messageAgeMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_message_age_seconds');
+            const observeStub = sinon.stub(messageAgeMetric, 'observe');
+            sinon.stub(ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total'), 'inc');
+            sinon.useFakeTimers(1700000010000);
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(
+                log, 'scan-A', 1700000000000);
+
+            assert(observeStub.calledOnce);
+            assert(observeStub.calledWithMatch(
+                { origin: 'bucket_processor' }, 10));
+            assert(log.error.notCalled);
+        });
+
+        it('should skip bucket processor scan message age observation on negative age', () => {
+            const messageAgeMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_message_age_seconds');
+            const observeStub = sinon.stub(messageAgeMetric, 'observe');
+            sinon.stub(ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total'), 'inc');
+            sinon.useFakeTimers(1700000000000);
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(
+                log, 'scan-A', 1700000001000);
+
+            assert(observeStub.notCalled);
+            assert(log.error.notCalled);
+        });
+
+        it('should catch errors in onBucketProcessorScanMessageReceived', () => {
+            const messagesMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total');
+            sinon.stub(messagesMetric, 'inc').throws(new Error('Metric error'));
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, 'scan-A');
+
+            assert(log.error.calledOnce);
+            assert(log.error.calledWithMatch('failed to update prometheus metrics', {
+                method: 'LifecycleMetrics.onBucketProcessorScanMessageReceived',
+                conductorScanId: 'scan-A',
+            }));
+        });
+
+        it('should skip per-scan metrics when scan id is missing', () => {
+            const messagesMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total');
+            const incStub = sinon.stub(messagesMetric, 'inc');
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, undefined);
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, '');
+
+            assert(incStub.notCalled);
+            assert(log.error.notCalled);
+        });
+
+        it('should remove stale bucket processor scan series after retention', () => {
+            const messagesMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total');
+            const removeStub = sinon.stub(messagesMetric, 'remove');
+            sinon.stub(messagesMetric, 'inc');
+            const clock = sinon.useFakeTimers(1700000000000);
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, 'scan-A');
+            clock.tick(24 * 60 * 60 * 1000 + 1);
+
+            assert(removeStub.calledOnce);
+            assert(removeStub.calledWithMatch({
+                origin: 'bucket_processor',
+                ['conductor_scan_id']: 'scan-A',
+            }));
+            assert(log.error.notCalled);
+        });
+
+        it('should reset bucket processor scan cleanup timer on update', () => {
+            const messagesMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_bucket_processor_scan_messages_total');
+            const removeStub = sinon.stub(messagesMetric, 'remove');
+            sinon.stub(messagesMetric, 'inc');
+            const clock = sinon.useFakeTimers(1700000000000);
+
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, 'scan-A');
+            clock.tick(12 * 60 * 60 * 1000);
+            LifecycleMetrics.onBucketProcessorScanMessageReceived(log, 'scan-A');
+            clock.tick(12 * 60 * 60 * 1000 + 1);
+
+            assert(removeStub.notCalled);
+            clock.tick(12 * 60 * 60 * 1000);
+
+            assert(removeStub.calledOnce);
+            assert(removeStub.calledWithMatch({
+                origin: 'bucket_processor',
+                ['conductor_scan_id']: 'scan-A',
+            }));
+            assert(log.error.notCalled);
+        });
+
         it('should catch errors in onLifecycleTriggered', () => {
             LifecycleMetrics.onLifecycleTriggered(log, 'conductor', 'expiration', 'us-east-1', NaN);
 
@@ -169,5 +300,36 @@ describe('LifecycleMetrics', () => {
                 process: 'conductor',
             }));
         });
+
+        it('should set latest batch bucket count metric', () => {
+            const endMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_latest_batch_end_time');
+            const countMetric = ZenkoMetrics.getMetric(
+                's3_lifecycle_latest_batch_bucket_count');
+            const endSet = sinon.stub(endMetric, 'set');
+            const countSet = sinon.stub(countMetric, 'set');
+
+            LifecycleMetrics.onConductorScanComplete(log, 7);
+
+            assert(endSet.calledOnce);
+            assert(endSet.calledWithMatch(
+                { origin: 'conductor' }));
+            assert.strictEqual(countSet.callCount, 1);
+            assert(countSet.getCall(0).calledWithMatch(
+                { origin: 'conductor' }, 7));
+            assert(log.error.notCalled);
+        });
+
+        it('should set latest batch start time', () => {
+            const latestStartSet = sinon.stub(ZenkoMetrics.getMetric(
+                's3_lifecycle_latest_batch_start_time'), 'set');
+
+            LifecycleMetrics.onProcessBuckets(log, 1700000000000);
+
+            assert(latestStartSet.calledWithMatch(
+                { origin: 'conductor' }, 1700000000000));
+            assert(log.error.notCalled);
+        });
+
     });
 });

From f6029a25e887aa50dc71ed0557520b8a755bb78c Mon Sep 17 00:00:00 2001
From: Maha Benzekri <maha.benzekri@scality.com>
Date: Sun, 5 Jul 2026 21:17:57 +0200
Subject: [PATCH 2/6] Track conductor scans with a scan id and lifecycle
 metrics.

Each processBuckets run gets a uuid v7 scan id, timestamped right after
the backlog-control gate (the only step that skips a scan), so a
Throttling error always means no scan was started. The id and start
timestamp are added to the request-logger default fields, published in
the scan-start metric, and carried in bucket-task messages under
contextInfo so downstream components can attribute work to the scan.

Completion metrics (end time, buckets listed) are only published on
success: a failed scan resets the local conductor state but leaves the
completion metrics behind, which is what lets a stalled-scan alert fire
and auto-resolve on the next good scan. Batch logs now report the scan
duration as fullScanElapsedMs and the count of unknown canonical ids
instead of the full list.

Issue: BB-740
---
 .../lifecycle/conductor/LifecycleConductor.js |  92 +++++++++---
 .../lifecycle/LifecycleConductor.spec.js      |  36 ++++-
 .../unit/lifecycle/LifecycleConductor.spec.js | 136 ++++++++++++++++++
 tests/utils/BackbeatTestConsumer.js           |  32 +++--
 tests/utils/fakeLogger.js                     |   1 +
 5 files changed, 263 insertions(+), 34 deletions(-)

diff --git a/extensions/lifecycle/conductor/LifecycleConductor.js b/extensions/lifecycle/conductor/LifecycleConductor.js
index 89203b7ee2..7c72a41033 100644
--- a/extensions/lifecycle/conductor/LifecycleConductor.js
+++ b/extensions/lifecycle/conductor/LifecycleConductor.js
@@ -3,6 +3,7 @@
 const async = require('async');
 const schedule = require('node-schedule');
 const zookeeper = require('node-zookeeper-client');
+const { v7: uuid } = require('uuid');
 
 const { constants, errors } = require('arsenal');
 const Logger = require('werelogs').Logger;
@@ -105,6 +106,8 @@ class LifecycleConductor {
         this._vaultClientCache = null;
         this._initialized = false;
         this._batchInProgress = false;
+        this._currentScanId = null;
+        this._currentScanStartTimestamp = null;
 
         // this cache only needs to be the size of one listing.
         // worst case scenario is 1 account per bucket:
@@ -353,6 +356,8 @@ class LifecycleConductor {
                 action: 'processObjects',
                 contextInfo: {
                     reqId: log.getSerializedUids(),
+                    conductorScanId: this._currentScanId,
+                    conductorScanStartTimestamp: this._currentScanStartTimestamp,
                 },
                 target: {
                     bucket: task.bucketName,
@@ -395,7 +400,8 @@ class LifecycleConductor {
 
     _createBucketTaskMessages(tasks, log, cb) {
         if (this.lcConfig.forceLegacyListing) {
-            return process.nextTick(cb, null, tasks.map(t => this._taskToMessage(t, lifecycleTaskVersions.v1, log)));
+            return process.nextTick(cb, null, tasks.map(t =>
+                this._taskToMessage(t, lifecycleTaskVersions.v1, log)));
         }
 
         return async.mapLimit(tasks, 10, (t, taskDone) =>
@@ -409,10 +415,24 @@ class LifecycleConductor {
             }), cb);
     }
 
+    _completeCurrentScan(log, nBucketsListed) {
+        LifecycleMetrics.onConductorScanComplete(
+            log,
+            nBucketsListed || 0,
+        );
+        this._resetCurrentScan();
+    }
+
+    _resetCurrentScan() {
+        this._currentScanId = null;
+        this._currentScanStartTimestamp = null;
+    }
+
     processBuckets(cb) {
         const log = this.logger.newRequestLogger();
-        const start = new Date();
+        const scanId = uuid();
         let nBucketsQueued = 0;
+        let nBucketsListed = 0;
         let messageDeliveryReports = 0;
 
         const messageSendQueue = async.cargo((tasks, done) => {
@@ -439,7 +459,8 @@ class LifecycleConductor {
                         }
                         return true;
                     }))),
-                (tasksWithAccountId, next) => this._createBucketTaskMessages(tasksWithAccountId, log, next),
+                (tasksWithAccountId, next) => this._createBucketTaskMessages(
+                    tasksWithAccountId, log, next),
             ],
             (err, messages) => {
                 nBucketsQueued += tasks.length;
@@ -448,7 +469,8 @@ class LifecycleConductor {
                     nBucketsQueued,
                     bucketsInCargo: tasks.length,
                     kafkaBucketMessagesDeliveryReports: messageDeliveryReports,
-                    kafkaEnqueueRateHz: Math.round(nBucketsQueued * 1000 / (new Date() - start)),
+                    kafkaEnqueueRateHz: Math.round(
+                        nBucketsQueued * 1000 / (Date.now() - this._currentScanStartTimestamp)),
                 });
 
                 this._accountIdCache.expireOldest();
@@ -463,18 +485,30 @@ class LifecycleConductor {
 
         async.waterfall([
             next => this._controlBacklog(next),
-            // error retrieving in progress jobs should not stop the current batch
-            // fallback to V1 listings
-            next => this._indexesGetInProgressJobs(log, () => next(null)),
             next => {
+                // The scan has started: index handling and listing below are
+                // part of it, so timestamp the scan start here (right after
+                // the backlog-control gate, the only step that skips a scan).
+                this._currentScanId = scanId;
+                this._currentScanStartTimestamp = Date.now();
+                log.addDefaultFields({
+                    conductorScanId: scanId,
+                    conductorScanStartTimestamp: this._currentScanStartTimestamp,
+                });
+                LifecycleMetrics.onProcessBuckets(log, this._currentScanStartTimestamp);
                 this._batchInProgress = true;
                 log.info('starting new lifecycle batch', { bucketSource: this._bucketSource });
-                this.listBuckets(messageSendQueue, log, (err, nBucketsListed) => {
-                    LifecycleMetrics.onBucketListing(log, err);
-                    return next(err, nBucketsListed);
-                });
+                return next();
             },
-            (nBucketsListed, next) => {
+            // error retrieving in progress jobs should not stop the current batch
+            // fallback to V1 listings
+            next => this._indexesGetInProgressJobs(log, () => next(null)),
+            next => this.listBuckets(messageSendQueue, log, (err, listedBucketsCount) => {
+                LifecycleMetrics.onBucketListing(log, err);
+                nBucketsListed = listedBucketsCount;
+                return next(err);
+            }),
+            next => {
                 async.until(
                     () => nBucketsQueued === nBucketsListed,
                     unext => setTimeout(unext, 1000),
@@ -482,7 +516,12 @@ class LifecycleConductor {
             },
         ], err => {
             if (err && err.Throttling) {
-                log.info('not starting new lifecycle batch', { reason: err });
+                // Throttling only originates from the backlog-control gate,
+                // before the scan id is set, so no scan was started and there
+                // is no scan duration to report here.
+                log.info('not starting new lifecycle batch', {
+                    reason: err,
+                });
                 if (cb) {
                     cb(err);
                 }
@@ -493,17 +532,30 @@ class LifecycleConductor {
             this.activeIndexingJobs = [];
             this._batchInProgress = false;
             const unknownCanonicalIds = this._accountIdCache.getMisses();
+            const fullScanElapsedMs = Date.now() - this._currentScanStartTimestamp;
 
             if (err) {
-                log.error('lifecycle batch failed', { error: err, unknownCanonicalIds });
+                log.error('lifecycle batch failed', {
+                    error: err,
+                    unknownCanonicalIdCount: unknownCanonicalIds.length,
+                    fullScanElapsedMs,
+                    nBucketsListed,
+                    nBucketsQueued,
+                });
+                this._resetCurrentScan();
                 if (cb) {
                     cb(err);
                 }
                 return;
             }
 
-            log.info('finished pushing lifecycle batch', { nBucketsQueued, unknownCanonicalIds });
-            LifecycleMetrics.onProcessBuckets(log);
+            log.info('finished pushing lifecycle batch', {
+                nBucketsQueued,
+                unknownCanonicalIdCount: unknownCanonicalIds.length,
+                fullScanElapsedMs,
+                nBucketsListed,
+            });
+            this._completeCurrentScan(log, nBucketsListed);
             if (cb) {
                 cb(null, nBucketsQueued);
             }
@@ -615,7 +667,7 @@ class LifecycleConductor {
         let isTruncated = true;
         let marker = initMarker;
         let nEnqueued = 0;
-        const start = new Date();
+        const start = Date.now();
         const retryWrapper = new BackbeatTask(this.lcConfig.conductor.retry);
 
         this.lastSentId = null;
@@ -628,7 +680,7 @@ class LifecycleConductor {
                     nEnqueuedToDownstream: nEnqueued,
                     inFlight: queue.length(),
                     maxInFlight: this._maxInFlightBatchSize,
-                    bucketListingPushRateHz: Math.round(nEnqueued * 1000 / (new Date() - start)),
+                    bucketListingPushRateHz: Math.round(nEnqueued * 1000 / (Date.now() - start)),
                     breakerState,
                 };
 
@@ -713,7 +765,7 @@ class LifecycleConductor {
 
     listMongodbBuckets(queue, log, cb) {
         let nEnqueued = 0;
-        const start = new Date();
+        const start = Date.now();
 
         const lastEntryPath = this.getBucketProgressZkPath();
         let lastSentId = null;
@@ -788,7 +840,7 @@ class LifecycleConductor {
                             nEnqueuedToDownstream: nEnqueued,
                             inFlight: queue.length(),
                             maxInFlight: this._maxInFlightBatchSize,
-                            enqueueRateHz: Math.round(nEnqueued * 1000 / (new Date() - start)),
+                            enqueueRateHz: Math.round(nEnqueued * 1000 / (Date.now() - start)),
                             breakerState,
                         };
 
diff --git a/tests/functional/lifecycle/LifecycleConductor.spec.js b/tests/functional/lifecycle/LifecycleConductor.spec.js
index 9943ab4c26..fc0b22bb2e 100644
--- a/tests/functional/lifecycle/LifecycleConductor.spec.js
+++ b/tests/functional/lifecycle/LifecycleConductor.spec.js
@@ -41,7 +41,11 @@ const expected2Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: {
+                reqId: 'test-request-id',
+                conductorScanId: 'test-scan-id',
+                conductorScanStartTimestamp: 0,
+            },
             target: { bucket: 'bucket1', owner: 'owner1', taskVersion: version },
             details: {},
         },
@@ -49,7 +53,11 @@ const expected2Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: {
+                reqId: 'test-request-id',
+                conductorScanId: 'test-scan-id',
+                conductorScanStartTimestamp: 0,
+            },
             target: { bucket: 'bucket1-2', owner: 'owner1', taskVersion: version },
             details: {},
         },
@@ -60,7 +68,11 @@ const expected4Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: {
+                reqId: 'test-request-id',
+                conductorScanId: 'test-scan-id',
+                conductorScanStartTimestamp: 0,
+            },
             target: { bucket: 'bucket1', owner: 'owner1', taskVersion: version },
             details: {},
         },
@@ -68,7 +80,11 @@ const expected4Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: {
+                reqId: 'test-request-id',
+                conductorScanId: 'test-scan-id',
+                conductorScanStartTimestamp: 0,
+            },
             target: { bucket: 'bucket1-2', owner: 'owner1', taskVersion: version },
             details: {},
         },
@@ -76,7 +92,11 @@ const expected4Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: {
+                reqId: 'test-request-id',
+                conductorScanId: 'test-scan-id',
+                conductorScanStartTimestamp: 0,
+            },
             target: { bucket: 'bucket3', owner: 'owner3', taskVersion: version },
             details: {},
         },
@@ -84,7 +104,11 @@ const expected4Messages = (version='v2') => ([
     {
         value: {
             action: 'processObjects',
-            contextInfo: { reqId: 'test-request-id' },
+            contextInfo: {
+                reqId: 'test-request-id',
+                conductorScanId: 'test-scan-id',
+                conductorScanStartTimestamp: 0,
+            },
             target: { bucket: 'bucket4', owner: 'owner4', taskVersion: version },
             details: {},
         },
diff --git a/tests/unit/lifecycle/LifecycleConductor.spec.js b/tests/unit/lifecycle/LifecycleConductor.spec.js
index 787b7cf0f0..834146b7fc 100644
--- a/tests/unit/lifecycle/LifecycleConductor.spec.js
+++ b/tests/unit/lifecycle/LifecycleConductor.spec.js
@@ -3,6 +3,7 @@
 const assert = require('assert');
 const sinon = require('sinon');
 const fakeLogger = require('../../utils/fakeLogger');
+const { errors } = require('arsenal');
 const { splitter } = require('arsenal').constants;
 
 const LifecycleConductor = require(
@@ -180,9 +181,14 @@ describe('Lifecycle Conductor', () => {
 
             sinon.stub(conductor, '_controlBacklog')
                 .callsFake(cb => cb(null));
+            const metricStub = sinon.stub(LifecycleMetrics, 'onConductorScanComplete');
 
             conductor.processBuckets(err => {
                 assert.strictEqual(err.message, 'error');
+                assert(metricStub.notCalled);
+                assert.strictEqual(conductor._batchInProgress, false);
+                assert.strictEqual(conductor._currentScanId, null);
+                assert.strictEqual(conductor._currentScanStartTimestamp, null);
                 done();
             });
         });
@@ -211,6 +217,124 @@ describe('Lifecycle Conductor', () => {
             conductor.processBuckets(done);
         });
 
+        it('should publish full scan metrics at end of scan', done => {
+            conductor._mongodbClient = {
+                getIndexingJobs: (_, cb) => cb(null, ['job1']),
+                getCollection: () => ({
+                    find: () => ({
+                        project: () => ({
+                            hasNext: () => Promise.resolve(false),
+                        }),
+                    }),
+                }),
+            };
+            conductor._zkClient = {
+                getData: (_, cb) => cb(null, null, null),
+                setData: (path, data, version, cb) => cb(null, { version: 1 }),
+            };
+
+            sinon.stub(conductor, '_controlBacklog').callsFake(cb => cb(null));
+            const metricStub = sinon.stub(LifecycleMetrics, 'onConductorScanComplete');
+
+            conductor.processBuckets(err => {
+                assert.ifError(err);
+                assert(metricStub.calledOnce);
+                const [, bucketCount] = metricStub.firstCall.args;
+                assert.strictEqual(bucketCount, 0);
+                done();
+            });
+        });
+
+        it('should keep the in-flight scan id when backlog control throttles a new scan', done => {
+            const inFlightScanId = 'in-flight-scan-id';
+            conductor._currentScanId = inFlightScanId;
+            conductor._batchInProgress = true;
+
+            sinon.stub(conductor, '_controlBacklog')
+                .callsFake(cb => cb(errors.Throttling.customizeDescription('Batch in progress')));
+            const onProcessBucketsStub = sinon.stub(LifecycleMetrics, 'onProcessBuckets');
+
+            conductor.processBuckets(err => {
+                assert(err && err.Throttling);
+                assert.strictEqual(conductor._currentScanId, inFlightScanId);
+                assert(onProcessBucketsStub.notCalled);
+                done();
+            });
+        });
+
+        it('should reset scan state without completion metrics when the scan fails after start', done => {
+            conductor._mongodbClient = {
+                getIndexingJobs: (_, cb) => cb(null, []),
+            };
+
+            // Throttling can only originate from _controlBacklog (before the
+            // scan starts); a failure after start is a generic error and must
+            // reset the scan state.
+            sinon.stub(conductor, '_controlBacklog').callsFake(cb => cb(null));
+            sinon.stub(conductor, 'listBuckets')
+                .callsFake((queue, log, cb) => cb(errors.InternalError));
+            const metricStub = sinon.stub(LifecycleMetrics, 'onConductorScanComplete');
+
+            conductor.processBuckets(err => {
+                assert(err && err.InternalError);
+                assert(metricStub.notCalled);
+                assert.strictEqual(conductor._batchInProgress, false);
+                assert.strictEqual(conductor._currentScanId, null);
+                assert.strictEqual(conductor._currentScanStartTimestamp, null);
+                done();
+            });
+        });
+
+        it('should generate a conductorScanId', done => {
+            conductor._mongodbClient = {
+                getIndexingJobs: (_, cb) => cb(null, []),
+                getCollection: () => ({
+                    find: () => ({
+                        project: () => ({
+                            hasNext: () => Promise.resolve(false),
+                        }),
+                    }),
+                }),
+            };
+            conductor._zkClient = {
+                getData: (_, cb) => cb(null, null, null),
+                setData: (path, data, version, cb) => cb(null, { version: 1 }),
+            };
+            conductor._producer = { send: (msg, cb) => cb(null, {}) };
+            const addDefaultFieldsStub = sinon.stub();
+            const testLog = conductor.logger.newRequestLogger();
+            testLog.addDefaultFields = addDefaultFieldsStub;
+            sinon.stub(conductor.logger, 'newRequestLogger').returns(testLog);
+
+            sinon.stub(conductor, '_controlBacklog').callsFake(cb => cb(null));
+            let capturedScanId = null;
+            sinon.stub(conductor, 'listBuckets')
+                .callsFake((queue, log, cb) => {
+                    const scanId = conductor._currentScanId;
+                    // Verify scanId is a valid UUID
+                    const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
+                    assert(uuidRegex.test(scanId),
+                        `conductorScanId should be a valid UUID v7, got: ${scanId}`);
+                    capturedScanId = scanId;
+                    assert(addDefaultFieldsStub.calledOnce);
+                    assert.strictEqual(addDefaultFieldsStub.firstCall.args[0].conductorScanId, scanId);
+                    cb(null, 0);
+                });
+            const metricStub = sinon.stub(LifecycleMetrics, 'onConductorScanComplete');
+
+            conductor.processBuckets(err => {
+                assert.ifError(err);
+                assert(metricStub.calledOnce);
+                const logPassedToMetric = metricStub.firstCall.args[0];
+                assert.strictEqual(logPassedToMetric, testLog);
+                assert(capturedScanId);
+                const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
+                assert(uuidRegex.test(capturedScanId),
+                    `captured conductorScanId should be a valid UUID v7, got: ${capturedScanId}`);
+                done();
+            });
+        });
+
         // tests that `activeIndexingJobRetrieved` is not reset until the e
         it('should not reset `activeIndexingJobsRetrieved` while async operations are in progress', done => {
             const order = [];
@@ -264,6 +388,18 @@ describe('Lifecycle Conductor', () => {
         });
     });
 
+    describe('_taskToMessage', () => {
+        it('should include conductor scan id in task context', () => {
+            conductor._currentScanId = 'scan-id-test';
+            conductor._currentScanStartTimestamp = 1700000000000;
+            const taskMessage = conductor._taskToMessage(
+                getTask(true), lifecycleTaskVersions.v2, log);
+            const parsed = JSON.parse(taskMessage.message);
+            assert.strictEqual(parsed.contextInfo.conductorScanId, 'scan-id-test');
+            assert.strictEqual(parsed.contextInfo.conductorScanStartTimestamp, 1700000000000);
+        });
+    });
+
     describe('_indexesGetOrCreate', () => {
         it('should return v2 for bucketd bucket sources', done => {
             conductor._bucketSource = 'bucketd';
diff --git a/tests/utils/BackbeatTestConsumer.js b/tests/utils/BackbeatTestConsumer.js
index 00ed404dd1..6c64b662d3 100644
--- a/tests/utils/BackbeatTestConsumer.js
+++ b/tests/utils/BackbeatTestConsumer.js
@@ -23,20 +23,36 @@ class BackbeatTestConsumer extends BackbeatConsumer {
                         `expected ${expectedMsg.key}`);
             }
             if (expectedMsg.value !== undefined) {
-                const parsedMsg = typeof expectedMsg.value === 'object' ?
-                          JSON.parse(message.value) :
-                          message.value.toString();
-                if (typeof expectedMsg.value === 'object' &&
-                    expectedMsg.value.contextInfo?.reqId === 'test-request-id') {
+                const isExpectedObject = typeof expectedMsg.value === 'object';
+                const parsedMsg = isExpectedObject ?
+                    JSON.parse(message.value) :
+                    message.value.toString();
+                const expectedValue = isExpectedObject ?
+                    JSON.parse(JSON.stringify(expectedMsg.value)) :
+                    expectedMsg.value;
+                if (isExpectedObject &&
+                    expectedValue.contextInfo?.reqId === 'test-request-id') {
                     // RequestId is generated randomly, we can't compare it: just check that it is
                     // present
                     assert(parsedMsg.contextInfo?.reqId, 'expected contextInfo.reqId field');
-                    parsedMsg.contextInfo.reqId = expectedMsg.value.contextInfo?.reqId;
+                    expectedValue.contextInfo.reqId = parsedMsg.contextInfo.reqId;
+                    if (expectedValue.contextInfo?.conductorScanId === 'test-scan-id') {
+                        assert(parsedMsg.contextInfo?.conductorScanId,
+                            'expected contextInfo.conductorScanId field');
+                        expectedValue.contextInfo.conductorScanId =
+                            parsedMsg.contextInfo.conductorScanId;
+                    }
+                    if (expectedValue.contextInfo?.conductorScanStartTimestamp === 0) {
+                        assert(parsedMsg.contextInfo?.conductorScanStartTimestamp,
+                            'expected contextInfo.conductorScanStartTimestamp field');
+                        expectedValue.contextInfo.conductorScanStartTimestamp =
+                            parsedMsg.contextInfo.conductorScanStartTimestamp;
+                    }
                 }
                 assert.deepStrictEqual(
-                    parsedMsg, expectedMsg.value,
+                    parsedMsg, expectedValue,
                     `unexpected message value ${parsedMsg}, ` +
-                        `expected ${expectedMsg.value}`);
+                        `expected ${expectedValue}`);
             }
         }
 
diff --git a/tests/utils/fakeLogger.js b/tests/utils/fakeLogger.js
index 2e241d2a97..7fd57e2d65 100644
--- a/tests/utils/fakeLogger.js
+++ b/tests/utils/fakeLogger.js
@@ -5,6 +5,7 @@ const fakeLogger = {
     debug: console.debug, // eslint-disable-line no-console
     warn: console.warn,    // eslint-disable-line no-console
     getSerializedUids: () => {},
+    addDefaultFields: () => {},
     end: () => fakeLogger,
     newRequestLogger: () => fakeLogger,
 };

From e6e85f2b728b3f6f1ff79b4c37dd6678d639d332 Mon Sep 17 00:00:00 2001
From: Maha Benzekri <maha.benzekri@scality.com>
Date: Sun, 5 Jul 2026 21:19:13 +0200
Subject: [PATCH 3/6] Record scan pickup metrics in the lifecycle bucket
 processor.

The bucket processor parses the conductor scan context from each
bucket-task message, attaches it (with bucket, owner and accountId) to a
per-message request logger so all logs for that message carry the scan
id, and reports the message to the per-scan received counter and
message-age histogram. Messages without a scan id (from an older
conductor during a rolling upgrade) or with a skewed timestamp are
counted without an age observation instead of recording a misleading
value.

Issue: BB-740
---
 .../LifecycleBucketProcessor.js               | 47 +++++++++++++------
 .../LifecycleBucketProcessor.spec.js          | 39 +++++++++++++++
 2 files changed, 71 insertions(+), 15 deletions(-)

diff --git a/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js b/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js
index 02de93e6ac..09eeff0411 100644
--- a/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js
+++ b/extensions/lifecycle/bucketProcessor/LifecycleBucketProcessor.js
@@ -26,6 +26,9 @@ const {
     extractBucketProcessorCircuitBreakerConfigs,
 } = require('../CircuitBreakerGroup');
 const { lifecycleTaskVersions } = require('../../../lib/constants');
+const {
+    LifecycleMetrics,
+} = require('../LifecycleMetrics');
 const locations = require('../../../conf/locationConfig.json');
 
 const PROCESS_OBJECTS_ACTION = 'processObjects';
@@ -277,21 +280,30 @@ class LifecycleBucketProcessor {
             return process.nextTick(() => cb(errors.InternalError));
         }
         const { bucket, owner, accountId, taskVersion } = result.target;
+        const conductorScanId = result.contextInfo?.conductorScanId;
+        const conductorScanStartTimestamp = result.contextInfo?.conductorScanStartTimestamp;
+        // Use one request logger per bucket-task message so scan context stays
+        // attached to all logs emitted while that message is processed.
+        const log = this._log.newRequestLogger();
+        log.addDefaultFields({
+            conductorScanId,
+            conductorScanStartTimestamp,
+            bucket,
+            owner,
+            accountId,
+        });
+
         if (!bucket || !owner || (!accountId && this._authConfig.type === authTypeAssumeRole)) {
-            this._log.error('kafka bucket entry missing required fields', {
+            log.error('kafka bucket entry missing required fields', {
                 method: 'LifecycleBucketProcessor._processBucketEntry',
-                bucket,
-                owner,
-                accountId,
             });
             return process.nextTick(() => cb(errors.InternalError));
         }
-        this._log.debug('processing bucket entry', {
+        log.debug('processing bucket entry', {
             method: 'LifecycleBucketProcessor._processBucketEntry',
-            bucket,
-            owner,
-            accountId,
         });
+        LifecycleMetrics.onBucketProcessorScanMessageReceived(
+            log, conductorScanId, conductorScanStartTimestamp);
 
         const s3Client = this.clientManager.getS3Client(accountId);
         if (!s3Client) {
@@ -307,19 +319,23 @@ class LifecycleBucketProcessor {
         }
 
         const params = { Bucket: bucket };
-        return this._getBucketLifecycleConfiguration(s3Client, params, (err, config) => {
+        return this._getBucketLifecycleConfiguration(s3Client, params, log, (err, config) => {
             if (err) {
                 if (err.name === 'NoSuchLifecycleConfiguration') {
-                    this._log.debug('skipping non-lifecycled bucket', { bucket });
+                    log.debug('skipping non-lifecycled bucket', {
+                        bucket,
+                    });
                     return cb();
                 }
 
                 if (err.name === 'NoSuchBucket') {
-                    this._log.error('skipping non-existent bucket', { bucket });
+                    log.error('skipping non-existent bucket', {
+                        bucket,
+                    });
                     return cb();
                 }
 
-                this._log.error('error getting bucket lifecycle config', {
+                log.error('error getting bucket lifecycle config', {
                     method: 'LifecycleBucketProcessor._processBucketEntry',
                     bucket,
                     owner,
@@ -339,7 +355,7 @@ class LifecycleBucketProcessor {
                 task = new LifecycleTaskV2(this);
             }
 
-            this._log.info('scheduling new task for bucket lifecycle', {
+            log.info('scheduling new task for bucket lifecycle', {
                 method: 'LifecycleBucketProcessor._processBucketEntry',
                 bucket,
                 owner,
@@ -360,10 +376,11 @@ class LifecycleBucketProcessor {
      * Call AWS.S3.GetBucketLifecycleConfiguration in a retry wrapper.
      * @param {S3Client} s3Client - the s3 client
      * @param {object} params - the parameters to pass to getBucketLifecycleConfiguration
+     * @param {Logger} log - The request logger to use
      * @param {Function} cb - The callback to call
      * @return {undefined}
      */
-    _getBucketLifecycleConfiguration(s3Client, params, cb) {
+    _getBucketLifecycleConfiguration(s3Client, params, log, cb) {
         return this.retryWrapper.retry({
             actionDesc: 'get bucket lifecycle',
             logFields: { params },
@@ -374,7 +391,7 @@ class LifecycleBucketProcessor {
                     .catch(done);
             },
             shouldRetryFunc: err => err.retryable,
-            log: this._log,
+            log,
         }, cb);
     }
 
diff --git a/tests/unit/lifecycle/LifecycleBucketProcessor.spec.js b/tests/unit/lifecycle/LifecycleBucketProcessor.spec.js
index ef7c88beb5..f892695a14 100644
--- a/tests/unit/lifecycle/LifecycleBucketProcessor.spec.js
+++ b/tests/unit/lifecycle/LifecycleBucketProcessor.spec.js
@@ -28,6 +28,7 @@ const {
 } = require('../../utils/kafkaEntries');
 const LifecycleTask = require('../../../extensions/lifecycle/tasks/LifecycleTask');
 const LifecycleTaskV2 = require('../../../extensions/lifecycle/tasks/LifecycleTaskV2');
+const { LifecycleMetrics } = require('../../../extensions/lifecycle/LifecycleMetrics');
 
 
 describe('Lifecycle Bucket Processor', () => {
@@ -262,6 +263,44 @@ describe('Lifecycle Bucket Processor', () => {
                 });
             });
         });
+
+        it('should pass conductor scan start timestamp to lifecycle metrics', done => {
+            const kafkaMessage = {
+                ...bucketProcessorV2Entry,
+                value: JSON.stringify({
+                    ...JSON.parse(bucketProcessorV2Entry.value),
+                    contextInfo: {
+                        conductorScanId: 'scan-A',
+                        conductorScanStartTimestamp: 1700000000000,
+                    },
+                }),
+            };
+            const metricStub = sinon.stub(
+                LifecycleMetrics, 'onBucketProcessorScanMessageReceived');
+            lbp.clientManager = {
+                getS3Client: () => ({}),
+                getBackbeatMetadataProxy: () => ({}),
+            };
+            sinon.stub(lbp, '_getBucketLifecycleConfiguration').yields(null, {
+                Rules: [{
+                    Status: 'Enabled',
+                    Transitions: [{
+                        Days: 10,
+                        StorageClass: 'azure'
+                    }],
+                    ID: 'dac36d89-0005-4c78-8e00-7e9ace06a9c4'
+                }]
+            });
+            lbp._internalTaskScheduler = async.queue((ctx, cb) => cb(), 1);
+
+            lbp._processBucketEntry(kafkaMessage, err => {
+                assert.ifError(err);
+                assert(metricStub.calledOnce);
+                assert.strictEqual(metricStub.firstCall.args[1], 'scan-A');
+                assert.strictEqual(metricStub.firstCall.args[2], 1700000000000);
+                done();
+            });
+        });
     });
 
     describe('start', () => {

From 66f58b2e004ebdab304cbc37ae3553719948bf08 Mon Sep 17 00:00:00 2001
From: Maha Benzekri <maha.benzekri@scality.com>
Date: Sun, 5 Jul 2026 21:19:13 +0200
Subject: [PATCH 4/6] Propagate the conductor scan context through lifecycle
 tasks.

Lifecycle tasks derive their action and continuation entries from the
bucket-task data, so the conductorScanId and conductorScanStartTimestamp
carried in contextInfo follow the work through continuation entries,
action queue entries and task logs. ActionQueueEntry exposes both fields
in its log info, which ties object-level task logs back to the scan that
scheduled them without putting bucket names in metric labels.

Issue: BB-740
---
 .../tasks/LifecycleDeleteObjectTask.js        |   1 +
 extensions/lifecycle/tasks/LifecycleTask.js   | 108 +++++++++---------
 extensions/lifecycle/tasks/LifecycleTaskV2.js |  51 ++++-----
 .../tasks/LifecycleUpdateExpirationTask.js    |   1 +
 .../tasks/LifecycleUpdateTransitionTask.js    |   1 +
 lib/models/ActionQueueEntry.js                |   2 +
 tests/unit/lifecycle/LifecycleTask.spec.js    | 101 ++++++++++++++++
 7 files changed, 188 insertions(+), 77 deletions(-)

diff --git a/extensions/lifecycle/tasks/LifecycleDeleteObjectTask.js b/extensions/lifecycle/tasks/LifecycleDeleteObjectTask.js
index 7139d8724d..c6d93dec30 100644
--- a/extensions/lifecycle/tasks/LifecycleDeleteObjectTask.js
+++ b/extensions/lifecycle/tasks/LifecycleDeleteObjectTask.js
@@ -309,6 +309,7 @@ class LifecycleDeleteObjectTask extends BackbeatTask {
             objectKey: 'target.key',
             versionId: 'target.version',
         });
+        log.addDefaultFields(entry.getLogInfo());
 
         return async.series([
             next => this._checkDate(entry, log, next),
diff --git a/extensions/lifecycle/tasks/LifecycleTask.js b/extensions/lifecycle/tasks/LifecycleTask.js
index 591ce63e81..0464f9556a 100644
--- a/extensions/lifecycle/tasks/LifecycleTask.js
+++ b/extensions/lifecycle/tasks/LifecycleTask.js
@@ -128,6 +128,45 @@ class LifecycleTask extends BackbeatTask {
         });
     }
 
+    _getScanContext(bucketData) {
+        return {
+            conductorScanId: bucketData?.contextInfo?.conductorScanId,
+            conductorScanStartTimestamp: bucketData?.contextInfo?.conductorScanStartTimestamp,
+        };
+    }
+
+    _getBucketEntryContext(bucketData, log) {
+        return {
+            ...this._getBucketEntryRequestIdContext(log),
+            ...this._getScanContext(bucketData),
+        };
+    }
+
+    _getBucketEntryRequestIdContext(log) {
+        return {
+            reqId: log.getSerializedUids(),
+        };
+    }
+
+    _getActionContext(bucketData, log, ruleType) {
+        return {
+            origin: 'lifecycle',
+            ruleType,
+            reqId: log.getSerializedUids(),
+            ...this._getScanContext(bucketData),
+        };
+    }
+
+    _makeContinuationEntry(bucketData, log, details) {
+        return {
+            ...bucketData,
+            contextInfo: this._getBucketEntryContext(bucketData, log),
+            // Preserve the previous Object.assign-style behavior: callers pass
+            // freshly-built details and ownership moves to the queued entry.
+            details,
+        };
+    }
+
     /**
      * This function forces syncing the latest data mover topic
      * offsets to the 'lifecycle' metrics snapshot. It is called when
@@ -235,10 +274,7 @@ class LifecycleTask extends BackbeatTask {
                         marker = contents[contents.length - 1].Key;
                     }
 
-                    const entry = Object.assign({}, bucketData, {
-                        contextInfo: { reqId: log.getSerializedUids() },
-                        details: { marker },
-                    });
+                    const entry = this._makeContinuationEntry(bucketData, log, { marker });
                     this._sendBucketEntry(entry, err => {
                         if (!err) {
                             log.debug(
@@ -408,15 +444,10 @@ class LifecycleTask extends BackbeatTask {
             if (data.IsTruncated && allVersions.length > 0 && nbRetries === 0) {
                 // Uses last version whether Version or DeleteMarker
                 const last = allVersions[allVersions.length - 1];
-                const entry = Object.assign({}, bucketData, {
-                    contextInfo: {
-                        reqId: log.getSerializedUids(),
-                    },
-                    details: {
-                        keyMarker: data.NextKeyMarker,
-                        versionIdMarker: data.NextVersionIdMarker,
-                        prevDate: last.LastModified,
-                    },
+                const entry = this._makeContinuationEntry(bucketData, log, {
+                    keyMarker: data.NextKeyMarker,
+                    versionIdMarker: data.NextVersionIdMarker,
+                    prevDate: last.LastModified,
                 });
                 this._sendBucketEntry(entry, err => {
                     if (!err) {
@@ -494,14 +525,9 @@ class LifecycleTask extends BackbeatTask {
                 if (data.IsTruncated && nbRetries === 0) {
                     // re-queue to kafka with `NextUploadIdMarker` &
                     // `NextKeyMarker` only once.
-                    const entry = Object.assign({}, bucketData, {
-                        contextInfo: {
-                            reqId: log.getSerializedUids(),
-                        },
-                        details: {
-                            keyMarker: data.NextKeyMarker,
-                            uploadIdMarker: data.NextUploadIdMarker,
-                        },
+                    const entry = this._makeContinuationEntry(bucketData, log, {
+                        keyMarker: data.NextKeyMarker,
+                        uploadIdMarker: data.NextUploadIdMarker,
                     });
                     return this._sendBucketEntry(entry, err => {
                         if (!err) {
@@ -1036,11 +1062,7 @@ class LifecycleTask extends BackbeatTask {
             rules.Expiration.Date < currentDate) {
             // expiration date passed for this object
             const entry = ActionQueueEntry.create('deleteObject')
-                .addContext({
-                    origin: 'lifecycle',
-                    ruleType: 'expiration',
-                    reqId: log.getSerializedUids(),
-                })
+                .addContext(this._getActionContext(bucketData, log, 'expiration'))
                 .setAttribute('target.owner', bucketData.target.owner)
                 .setAttribute('target.bucket', bucketData.target.bucket)
                 .setAttribute('target.accountId', bucketData.target.accountId)
@@ -1064,11 +1086,7 @@ class LifecycleTask extends BackbeatTask {
         if (rules.Expiration.Days !== undefined &&
             daysSinceInitiated >= rules.Expiration.Days) {
             const entry = ActionQueueEntry.create('deleteObject')
-                .addContext({
-                    origin: 'lifecycle',
-                    ruleType: 'expiration',
-                    reqId: log.getSerializedUids(),
-                })
+                .addContext(this._getActionContext(bucketData, log, 'expiration'))
                 .setAttribute('target.owner', bucketData.target.owner)
                 .setAttribute('target.bucket', bucketData.target.bucket)
                 .setAttribute('target.accountId', bucketData.target.accountId)
@@ -1170,11 +1188,7 @@ class LifecycleTask extends BackbeatTask {
             transitionTime: new Date(params.transitionTime).toISOString(),
             attempt,
         });
-        entry.addContext({
-            origin: 'lifecycle',
-            ruleType: 'transition',
-            reqId: log.getSerializedUids(),
-        });
+        entry.addContext(this._getActionContext(params.bucketData, log, 'transition'));
         entry.setAttribute('source', {
             bucket: params.bucket,
             objectKey: params.objectKey,
@@ -1374,6 +1388,7 @@ class LifecycleTask extends BackbeatTask {
                 site: rules[ncvt].StorageClass,
                 transitionTime: this._lifecycleDateTime.getTransitionTimestamp(
                     { Days: rules[ncvt][ncd] }, staleDate),
+                bucketData,
             }, log, cb);
             return;
         }
@@ -1446,11 +1461,7 @@ class LifecycleTask extends BackbeatTask {
                 // if a valid Expiration rule exists, apply and permanently delete this DM
                 if (matchingNoncurrentKeys.length === 0 && applicableExpRule) {
                     const entry = ActionQueueEntry.create('deleteObject')
-                        .addContext({
-                            origin: 'lifecycle',
-                            ruleType: 'expiration',
-                            reqId: log.getSerializedUids(),
-                        })
+                        .addContext(this._getActionContext(bucketData, log, 'expiration'))
                         .setAttribute('target.owner', bucketData.target.owner)
                         .setAttribute('target.bucket', bucketData.target.bucket)
                         .setAttribute('target.key', deleteMarker.Key)
@@ -1515,11 +1526,7 @@ class LifecycleTask extends BackbeatTask {
             const storageClass = this._isDeleteMarker(verToExpire) ?
                 LIFECYCLE_MARKER_METRICS_LOCATION : verToExpire.StorageClass;
             const entry = ActionQueueEntry.create('deleteObject')
-                .addContext({
-                    origin: 'lifecycle',
-                    ruleType: 'expiration',
-                    reqId: log.getSerializedUids(),
-                })
+                .addContext(this._getActionContext(bucketData, log, 'expiration'))
                 .setAttribute('target.owner', bucketData.target.owner)
                 .setAttribute('target.bucket', bucketData.target.bucket)
                 .setAttribute('target.accountId', bucketData.target.accountId)
@@ -1606,6 +1613,7 @@ class LifecycleTask extends BackbeatTask {
                         site: rules.Transition.StorageClass,
                         transitionTime: this._lifecycleDateTime.getTransitionTimestamp(
                             rules.Transition, object.LastModified),
+                        bucketData,
                     }, log, done);
                 }
                 return done();
@@ -1709,6 +1717,7 @@ class LifecycleTask extends BackbeatTask {
                 encodedVersionId: undefined,
                 transitionTime: this._lifecycleDateTime.getTransitionTimestamp(
                     rules.Transition, version.LastModified),
+                bucketData,
             }, log, done);
         }
 
@@ -1754,11 +1763,7 @@ class LifecycleTask extends BackbeatTask {
                     uploadId: upload.UploadId,
                 });
                 const entry = ActionQueueEntry.create('deleteMPU')
-                    .addContext({
-                        origin: 'lifecycle',
-                        ruleType: 'expiration',
-                        reqId: log.getSerializedUids(),
-                    })
+                    .addContext(this._getActionContext(bucketData, log, 'expiration'))
                     .setAttribute('target.owner', bucketData.target.owner)
                     .setAttribute('target.bucket', bucketData.target.bucket)
                     .setAttribute('target.accountId', bucketData.target.accountId)
@@ -1827,6 +1832,7 @@ class LifecycleTask extends BackbeatTask {
     processBucketEntry(bucketLCRules, bucketData, s3target,
     backbeatMetadataProxy, nbRetries, done) {
         const log = this.log.newRequestLogger();
+        log.addDefaultFields(this._getScanContext(bucketData));
         this.s3target = s3target;
         this.backbeatMetadataProxy = backbeatMetadataProxy;
         if (!this.backbeatMetadataProxy) {
diff --git a/extensions/lifecycle/tasks/LifecycleTaskV2.js b/extensions/lifecycle/tasks/LifecycleTaskV2.js
index f7ca26652f..e2d66c68a2 100644
--- a/extensions/lifecycle/tasks/LifecycleTaskV2.js
+++ b/extensions/lifecycle/tasks/LifecycleTaskV2.js
@@ -15,6 +15,12 @@ class LifecycleTaskV2 extends LifecycleTask {
      * NOTE: Among other methods, LifecycleTaskV2 inherits the processBucketEntry method from LifecycleTask.
      */
 
+    _getBucketEntryRequestIdContext(log) {
+        return {
+            requestId: log.getSerializedUids(),
+        };
+    }
+
     /**
      * Skips kafka entry
      * @param {object} bucketData - bucket data from bucketTasksTopic
@@ -51,9 +57,11 @@ class LifecycleTaskV2 extends LifecycleTask {
                     storageClass,
                 } = l;
 
-                const entry = Object.assign({}, bucketData, {
-                    contextInfo: { requestId: log.getSerializedUids() },
-                    details: { beforeDate, prefix, listType, storageClass },
+                const entry = this._makeContinuationEntry(bucketData, log, {
+                    beforeDate,
+                    prefix,
+                    listType,
+                    storageClass,
                 });
 
                 this._sendBucketEntry(entry, err => {
@@ -114,15 +122,12 @@ class LifecycleTaskV2 extends LifecycleTask {
 
             // re-queue truncated listing only once.
             if (isTruncated && nbRetries === 0) {
-                const entry = Object.assign({}, bucketData, {
-                    contextInfo: { requestId: log.getSerializedUids() },
-                    details: {
-                        beforeDate: params.BeforeDate,
-                        prefix: params.Prefix,
-                        storageClass: params.ExcludedDataStoreName,
-                        listType,
-                        ...markerInfo
-                    },
+                const entry = this._makeContinuationEntry(bucketData, log, {
+                    beforeDate: params.BeforeDate,
+                    prefix: params.Prefix,
+                    storageClass: params.ExcludedDataStoreName,
+                    listType,
+                    ...markerInfo,
                 });
 
                 this._sendBucketEntry(entry, err => {
@@ -198,15 +203,12 @@ class LifecycleTaskV2 extends LifecycleTask {
 
             // re-queue truncated listing only once.
             if (isTruncated && nbRetries === 0) {
-                const entry = Object.assign({}, bucketData, {
-                    contextInfo: { requestId: log.getSerializedUids() },
-                    details: {
-                        beforeDate: params.BeforeDate,
-                        prefix: params.Prefix,
-                        storageClass: params.ExcludedDataStoreName,
-                        listType,
-                        ...markerInfo,
-                    },
+                const entry = this._makeContinuationEntry(bucketData, log, {
+                    beforeDate: params.BeforeDate,
+                    prefix: params.Prefix,
+                    storageClass: params.ExcludedDataStoreName,
+                    listType,
+                    ...markerInfo,
                 });
 
                 this._sendBucketEntry(entry, err => {
@@ -350,6 +352,7 @@ class LifecycleTaskV2 extends LifecycleTask {
                 site: rules.Transition.StorageClass,
                 transitionTime: this._lifecycleDateTime.getTransitionTimestamp(
                     rules.Transition, obj.LastModified),
+                bucketData,
             }, log, cb);
         }
 
@@ -422,11 +425,7 @@ class LifecycleTaskV2 extends LifecycleTask {
 
         if (applicableExpRule) {
             const entry = ActionQueueEntry.create('deleteObject')
-                .addContext({
-                    origin: 'lifecycle',
-                    ruleType: 'expiration',
-                    reqId: log.getSerializedUids(),
-                })
+                .addContext(this._getActionContext(bucketData, log, 'expiration'))
                 .setAttribute('target.owner', bucketData.target.owner)
                 .setAttribute('target.bucket', bucketData.target.bucket)
                 .setAttribute('target.key', deleteMarker.Key)
diff --git a/extensions/lifecycle/tasks/LifecycleUpdateExpirationTask.js b/extensions/lifecycle/tasks/LifecycleUpdateExpirationTask.js
index a94928ab91..d0e1badce9 100644
--- a/extensions/lifecycle/tasks/LifecycleUpdateExpirationTask.js
+++ b/extensions/lifecycle/tasks/LifecycleUpdateExpirationTask.js
@@ -155,6 +155,7 @@ class LifecycleUpdateExpirationTask extends BackbeatTask {
             objectKey: 'target.key',
             versionId: 'target.version',
         });
+        log.addDefaultFields(entry.getLogInfo());
 
         async.waterfall([
             next => {
diff --git a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js
index 55820bad8e..4f57c33c7f 100644
--- a/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js
+++ b/extensions/lifecycle/tasks/LifecycleUpdateTransitionTask.js
@@ -267,6 +267,7 @@ class LifecycleUpdateTransitionTask extends BackbeatTask {
             eTag: 'target.eTag',
             lastModified: 'target.lastModified',
         });
+        log.addDefaultFields(entry.getLogInfo());
         if (entry.getStatus() === 'success') {
             return this.handleSuccessfullTransition(entry, log, done);
         }
diff --git a/lib/models/ActionQueueEntry.js b/lib/models/ActionQueueEntry.js
index eaab70aff3..18c79474bd 100644
--- a/lib/models/ActionQueueEntry.js
+++ b/lib/models/ActionQueueEntry.js
@@ -107,6 +107,8 @@ class ActionQueueEntry {
             actionContext: 'contextInfo',
             actionStatus: 'status',
             actionError: 'error',
+            conductorScanId: 'contextInfo.conductorScanId',
+            conductorScanStartTimestamp: 'contextInfo.conductorScanStartTimestamp',
         };
     }
 
diff --git a/tests/unit/lifecycle/LifecycleTask.spec.js b/tests/unit/lifecycle/LifecycleTask.spec.js
index 3c8b93d780..b80859ab8e 100644
--- a/tests/unit/lifecycle/LifecycleTask.spec.js
+++ b/tests/unit/lifecycle/LifecycleTask.spec.js
@@ -8,6 +8,8 @@ const { ValidLifecycleRules } = require('arsenal').models;
 
 const LifecycleTask = require(
     '../../../extensions/lifecycle/tasks/LifecycleTask');
+const LifecycleTaskV2 = require(
+    '../../../extensions/lifecycle/tasks/LifecycleTaskV2');
 const fakeLogger = require('../../utils/fakeLogger');
 const { timeOptions } = require('../../functional/lifecycle/configObjects');
 
@@ -654,6 +656,94 @@ describe('lifecycle task helper methods', () => {
         });
     });
 
+    describe('LifecycleTaskV2 scan context propagation', () => {
+        let lifecycleTaskV2;
+
+        const bucketData = {
+            target: {
+                owner: 'test-user',
+                accountId: 'test-account',
+                bucket: 'test-bucket',
+            },
+            contextInfo: {
+                conductorScanId: 'scan-A',
+                conductorScanStartTimestamp: 1700000000000,
+            },
+        };
+
+        beforeEach(() => {
+            lifecycleTaskV2 = new LifecycleTaskV2(lp);
+        });
+
+        afterEach(() => {
+            sinon.restore();
+        });
+
+        it('should propagate scan context to current-version transition actions', done => {
+            sinon.stub(lifecycleTaskV2, '_checkAndApplyExpirationRule').returns(false);
+            sinon.stub(lifecycleTaskV2, '_isDeleteMarker').returns(false);
+            sinon.stub(lifecycleTaskV2, '_applyTransitionRule')
+                .callsFake((params, log, cb) => {
+                    assert.strictEqual(params.bucketData.contextInfo.conductorScanId, 'scan-A');
+                    assert.strictEqual(
+                        params.bucketData.contextInfo.conductorScanStartTimestamp, 1700000000000);
+                    cb();
+                });
+
+            lifecycleTaskV2._compareCurrent(bucketData, {
+                Key: 'test-key',
+                VersionId: 'test-version',
+                ETag: '"test-etag"',
+                LastModified: '2023-01-01T00:00:00.000Z',
+            }, {
+                Transition: {
+                    Days: 1,
+                    StorageClass: 'test-site',
+                },
+            }, fakeLogger, done);
+        });
+
+        it('should propagate scan context to expired delete-marker actions', () => {
+            class LifecycleTaskV2Mock extends LifecycleTaskV2 {
+                _sendObjectAction(entry, cb) {
+                    this.latestEntry = entry;
+                    return cb();
+                }
+            }
+            lifecycleTaskV2 = new LifecycleTaskV2Mock(lp);
+
+            lifecycleTaskV2._checkAndApplyEODMRule(bucketData, {
+                Key: 'test-key',
+                VersionId: 'test-version',
+                LastModified: new Date(Date.now() - (2 * DAY)).toISOString(),
+            }, {
+                Expiration: { Days: 1 },
+            }, fakeLogger);
+
+            const context = lifecycleTaskV2.latestEntry.getContext();
+            assert.strictEqual(context.conductorScanId, 'scan-A');
+            assert.strictEqual(context.conductorScanStartTimestamp, 1700000000000);
+        });
+
+        it('should preserve V2 continuation requestId context', () => {
+            const log = {
+                ...fakeLogger,
+                getSerializedUids: () => 'test-request-id',
+            };
+            const entry = lifecycleTaskV2._makeContinuationEntry(bucketData, log, {
+                beforeDate: '2023-01-01T00:00:00.000Z',
+                prefix: '',
+                listType: 'current',
+            });
+
+            assert.strictEqual(entry.contextInfo.requestId, 'test-request-id');
+            assert.strictEqual(entry.contextInfo.reqId, undefined);
+            assert.strictEqual(entry.contextInfo.conductorScanId, 'scan-A');
+            assert.strictEqual(
+                entry.contextInfo.conductorScanStartTimestamp, 1700000000000);
+        });
+    });
+
     describe('_rulesHaveTag', () => {
         it('should return true if rule has a prefix and tag', () => {
             const rules = [
@@ -2230,6 +2320,17 @@ describe('lifecycle task helper methods', () => {
             site: 'test-site',
             accountId: 'test-account-id',
             transitionTime: Date.now(),
+            bucketData: {
+                target: {
+                    bucket: 'test-bucket',
+                    owner: 'test-owner',
+                    accountId: 'test-account-id',
+                },
+                contextInfo: {
+                    conductorScanId: 'scan-A',
+                    conductorScanStartTimestamp: 1700000000000,
+                },
+            },
         };
 
         before(() => {

From d9a23de4389a2d55c1dcad0b6fe71eaaf54f5eda Mon Sep 17 00:00:00 2001
From: Maha Benzekri <maha.benzekri@scality.com>
Date: Sun, 5 Jul 2026 21:19:14 +0200
Subject: [PATCH 5/6] Add conductor scan lifecycle alerts.

LifecycleLateScan fires when the conductor stops scheduling scans, based
on the scan-start heartbeat. LifecycleStalledScan fires when a scan
started but its completion metric did not catch up (start time above end
time), i.e. it is stalled or failed after starting; it uses dedicated
conductor-scan thresholds (wired into the alert rendering workflow) and
auto-resolves on the next successful scan.
LifecycleBucketProcessorMultipleParallelScans reports bucket processors
seeing messages from several conductor scans at once.

Issue: BB-740
---
 .github/workflows/alerts.yaml         |   2 +
 monitoring/lifecycle/alerts.test.yaml | 135 +++++++++++++++++++-------
 monitoring/lifecycle/alerts.yaml      |  91 ++++++++++++++++-
 3 files changed, 188 insertions(+), 40 deletions(-)

diff --git a/.github/workflows/alerts.yaml b/.github/workflows/alerts.yaml
index f4e17f10d5..4ebbb7445c 100644
--- a/.github/workflows/alerts.yaml
+++ b/.github/workflows/alerts.yaml
@@ -32,6 +32,8 @@ jobs:
             lifecycle_transition_replicas=3
             lifecycle_latency_warning_threshold=120
             lifecycle_latency_critical_threshold=180
+            lifecycle_conductor_scan_warning_threshold=120
+            lifecycle_conductor_scan_critical_threshold=180
           github_token: ${{ secrets.GIT_ACCESS_TOKEN }}
 
       - name: Render and test replication 
diff --git a/monitoring/lifecycle/alerts.test.yaml b/monitoring/lifecycle/alerts.test.yaml
index f796a51bc9..9977a5fe05 100644
--- a/monitoring/lifecycle/alerts.test.yaml
+++ b/monitoring/lifecycle/alerts.test.yaml
@@ -328,8 +328,8 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 4m
         exp_alerts:
@@ -337,14 +337,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 5m
         exp_alerts:
@@ -352,14 +352,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 6m
         exp_alerts:
@@ -367,8 +367,8 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 7m
         exp_alerts:
@@ -376,8 +376,8 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 8m
         exp_alerts:
@@ -385,14 +385,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 9m
         exp_alerts:
@@ -400,14 +400,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 10m
         exp_alerts:
@@ -415,14 +415,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 11m
         exp_alerts:
@@ -430,14 +430,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 12m
         exp_alerts:
@@ -445,14 +445,14 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
           - exp_labels:
               severity: critical
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 3m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 3m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 13m
         exp_alerts:
@@ -460,8 +460,8 @@ tests:
               severity: warning
             exp_annotations:
               zenko_service: backbeat-lifecycle-producer
-              description: "Last lifecycle scan was performed more than 2m 0s ago."
-              summary: "Lifecycle scan not executed in time"
+              description: "Last lifecycle scan was scheduled more than 2m 0s ago."
+              summary: "Lifecycle scan not scheduled in time"
       - alertname: LifecycleLateScan
         eval_time: 14m
         exp_alerts: []
@@ -509,6 +509,69 @@ tests:
         eval_time: 18h
         exp_alerts: []
 
+  - name: LifecycleStalledScan
+    interval: 1m
+    input_series:
+      - series: s3_lifecycle_latest_batch_start_time{namespace="zenko",job="artesca-data-backbeat-lifecycle-producer-headless",pod="foo"}
+        values: 60000+0x5 360000
+      - series: s3_lifecycle_latest_batch_end_time{namespace="zenko",job="artesca-data-backbeat-lifecycle-producer-headless",pod="foo"}
+        values: 0+0x5 360000
+    alert_rule_test:
+      - alertname: LifecycleStalledScan
+        eval_time: 2m
+        exp_alerts: []
+      - alertname: LifecycleStalledScan
+        eval_time: 4m
+        exp_alerts:
+          - exp_labels:
+              severity: warning
+            exp_annotations:
+              zenko_service: backbeat-lifecycle-producer
+              description: "Last lifecycle scan did not complete within 2m 0s of being scheduled (it may be stalled or have failed)."
+              summary: "Lifecycle scan did not complete in time"
+      - alertname: LifecycleStalledScan
+        eval_time: 5m
+        exp_alerts:
+          - exp_labels:
+              severity: critical
+            exp_annotations:
+              zenko_service: backbeat-lifecycle-producer
+              description: "Last lifecycle scan did not complete within 3m 0s of being scheduled (it may be stalled or have failed)."
+              summary: "Lifecycle scan did not complete in time"
+          - exp_labels:
+              severity: warning
+            exp_annotations:
+              zenko_service: backbeat-lifecycle-producer
+              description: "Last lifecycle scan did not complete within 2m 0s of being scheduled (it may be stalled or have failed)."
+              summary: "Lifecycle scan did not complete in time"
+      - alertname: LifecycleStalledScan
+        eval_time: 6m
+        exp_alerts: []
+
+  - name: LifecycleBucketProcessorMultipleParallelScans
+    interval: 1m
+    input_series:
+      - series: s3_lifecycle_bucket_processor_scan_messages_total{namespace="zenko",job="artesca-data-backbeat-lifecycle-bucket-processor-headless",conductor_scan_id="scan-a"}
+        values: 0+10x20
+      - series: s3_lifecycle_bucket_processor_scan_messages_total{namespace="zenko",job="artesca-data-backbeat-lifecycle-bucket-processor-headless",conductor_scan_id="scan-b"}
+        values: _ _ _ 0+10x17
+    alert_rule_test:
+      - alertname: LifecycleBucketProcessorMultipleParallelScans
+        eval_time: 2m
+        exp_alerts: []
+      - alertname: LifecycleBucketProcessorMultipleParallelScans
+        eval_time: 10m
+        exp_alerts: []
+      - alertname: LifecycleBucketProcessorMultipleParallelScans
+        eval_time: 15m
+        exp_alerts:
+          - exp_labels:
+              severity: warning
+            exp_annotations:
+              zenko_service: backbeat-lifecycle-bucket-processor
+              description: "Bucket processors have processed bucket-task messages from multiple lifecycle conductor scans in parallel for more than 10 minutes."
+              summary: "Lifecycle bucket processors are receiving multiple scans"
+
   - name: LifecycleLatency
     interval: 10m
     input_series:
diff --git a/monitoring/lifecycle/alerts.yaml b/monitoring/lifecycle/alerts.yaml
index 3498bc302e..7e5176e4bd 100644
--- a/monitoring/lifecycle/alerts.yaml
+++ b/monitoring/lifecycle/alerts.yaml
@@ -25,6 +25,12 @@ x-inputs:
 - name: lifecycle_latency_critical_threshold
   type: config
   value: 36*60*60 # 36h, in seconds
+- name: lifecycle_conductor_scan_warning_threshold
+  type: config
+  value: 10*60 # 10m, in seconds
+- name: lifecycle_conductor_scan_critical_threshold
+  type: config
+  value: 30*60 # 30m, in seconds
 
 groups:
 - name: LifecycleProducer
@@ -57,9 +63,9 @@ groups:
     Annotations:
       zenko_service: backbeat-lifecycle-producer
       description: >-
-        Last lifecycle scan was performed more than
+        Last lifecycle scan was scheduled more than
         {{ ${lifecycle_latency_warning_threshold} | humanizeDuration }} ago.
-      summary: "Lifecycle scan not executed in time"
+      summary: "Lifecycle scan not scheduled in time"
 
   - alert: LifecycleLateScan
     Expr: |
@@ -78,9 +84,67 @@ groups:
     Annotations:
       zenko_service: backbeat-lifecycle-producer
       description: >-
-        Last lifecycle scan was performed more than
+        Last lifecycle scan was scheduled more than
         {{ ${lifecycle_latency_critical_threshold} | humanizeDuration }} ago.
-      summary: "Lifecycle scan not executed in time"
+      summary: "Lifecycle scan not scheduled in time"
+
+  - alert: LifecycleStalledScan
+    Expr: |
+      # Keep end_time=0 meaningful: if the first scan starts but never
+      # completes, or errors after starting without updating end_time,
+      # this alert should fire after the configured threshold.
+      time() - (
+        (
+          max(s3_lifecycle_latest_batch_start_time{
+            namespace="${namespace}", job="${job_lifecycle_producer}"
+          })
+          >
+          (
+            max(s3_lifecycle_latest_batch_end_time{
+              namespace="${namespace}", job="${job_lifecycle_producer}"
+            })
+            or vector(0)
+          )
+        ) / 1000
+      ) > ${lifecycle_conductor_scan_warning_threshold}
+    Labels:
+      severity: warning
+    Annotations:
+      zenko_service: backbeat-lifecycle-producer
+      description: >-
+        Last lifecycle scan did not complete within
+        {{ ${lifecycle_conductor_scan_warning_threshold} | humanizeDuration }} of being scheduled
+        (it may be stalled or have failed).
+      summary: "Lifecycle scan did not complete in time"
+
+  - alert: LifecycleStalledScan
+    Expr: |
+      # Keep end_time=0 meaningful: if the first scan starts but never
+      # completes, or errors after starting without updating end_time,
+      # this alert should fire after the configured threshold.
+      time() - (
+        (
+          max(s3_lifecycle_latest_batch_start_time{
+            namespace="${namespace}", job="${job_lifecycle_producer}"
+          })
+          >
+          (
+            max(s3_lifecycle_latest_batch_end_time{
+              namespace="${namespace}", job="${job_lifecycle_producer}"
+            })
+            or vector(0)
+          )
+        ) / 1000
+      ) > ${lifecycle_conductor_scan_critical_threshold}
+    Labels:
+      severity: critical
+    Annotations:
+      zenko_service: backbeat-lifecycle-producer
+      description: >-
+        Last lifecycle scan did not complete within
+        {{ ${lifecycle_conductor_scan_critical_threshold} | humanizeDuration }} of being scheduled
+        (it may be stalled or have failed).
+      summary: "Lifecycle scan did not complete in time"
 
   - alert: LifecycleIndexCreationFailed
     Expr: |
@@ -143,6 +207,25 @@ groups:
       description: "More than 5% of S3 requests by bucket processors resulting in errors"
       summary: "Very high rate of S3 request errors"
 
+  - alert: LifecycleBucketProcessorMultipleParallelScans
+    Expr: |
+      count(
+        sum by(conductor_scan_id) (
+          rate(s3_lifecycle_bucket_processor_scan_messages_total{
+            namespace="${namespace}", job="${job_lifecycle_bucket_processor}"
+          }[1m])
+        ) > 0
+      ) > 1
+    For: "10m"
+    Labels:
+      severity: warning
+    Annotations:
+      zenko_service: backbeat-lifecycle-bucket-processor
+      description: >-
+        Bucket processors have processed bucket-task messages from multiple
+        lifecycle conductor scans in parallel for more than 10 minutes.
+      summary: "Lifecycle bucket processors are receiving multiple scans"
+
 - name: LifecycleObjectProcessor
   rules:
 

From df55485c48a8e4ee485ac9394526d998ab3508fc Mon Sep 17 00:00:00 2001
From: Maha Benzekri <maha.benzekri@scality.com>
Date: Sun, 5 Jul 2026 21:19:14 +0200
Subject: [PATCH 6/6] Add conductor scan panels to the lifecycle dashboard.

A new Lifecycle Conductor section shows the scheduling heartbeat, the
batch duration as a time series with min/mean/max legend values, the
buckets listed in the latest scan, and the stalled state. Bucket
processor panels show the pickup rate by pod (unfiltered so a pod stuck
at zero stays visible), the total messages per conductor scan plotted
with increase() so each scan draws a rising line that plateaus on
completion, and the message-age heatmap.

Issue: BB-740
---
 monitoring/lifecycle/dashboard.json | 431 +++++++++++++++++++++++++++-
 monitoring/lifecycle/dashboard.py   | 116 ++++++++
 2 files changed, 541 insertions(+), 6 deletions(-)

diff --git a/monitoring/lifecycle/dashboard.json b/monitoring/lifecycle/dashboard.json
index a714de95e7..f8b016ee6d 100644
--- a/monitoring/lifecycle/dashboard.json
+++ b/monitoring/lifecycle/dashboard.json
@@ -4345,6 +4345,425 @@
       "transparent": false,
       "type": "timeseries"
     },
+    {
+      "datasource": "${DS_PROMETHEUS}",
+      "description": "Duration of completed conductor scans over time. Gaps appear while a scan is in progress (its end time has not caught up with its start time yet).",
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 0,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "lineInterpolation": "smooth",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "log": 2,
+              "type": "linear"
+            },
+            "showPoints": "auto",
+            "spanNulls": false,
+            "stacking": {},
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": []
+          },
+          "unit": "s"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 20,
+        "x": 0,
+        "y": 111
+      },
+      "hideTimeOverride": false,
+      "id": 50,
+      "links": [],
+      "maxDataPoints": 100,
+      "options": {
+        "legend": {
+          "calcs": [
+            "min",
+            "mean",
+            "max"
+          ],
+          "displayMode": "table",
+          "placement": "right"
+        },
+        "tooltip": {
+          "mode": "single"
+        }
+      },
+      "targets": [
+        {
+          "datasource": null,
+          "expr": "((max(s3_lifecycle_latest_batch_end_time{job=\"${job_lifecycle_producer}\", namespace=\"${namespace}\"}) - max(s3_lifecycle_latest_batch_start_time{job=\"${job_lifecycle_producer}\", namespace=\"${namespace}\"})) > 0) / 1000",
+          "format": "time_series",
+          "hide": false,
+          "instant": false,
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "scan duration",
+          "metric": "",
+          "refId": "",
+          "step": 10,
+          "target": ""
+        }
+      ],
+      "title": "Conductor Batch Duration",
+      "transformations": [],
+      "transparent": false,
+      "type": "timeseries"
+    },
+    {
+      "datasource": "${DS_PROMETHEUS}",
+      "description": "Number of buckets listed during the latest conductor scan.",
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "custom": {},
+          "decimals": 0,
+          "mappings": [],
+          "noValue": "0",
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "semi-dark-blue",
+                "index": 0,
+                "line": true,
+                "op": "gt",
+                "value": "null",
+                "yaxis": "left"
+              }
+            ]
+          },
+          "unit": "none"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 4,
+        "x": 20,
+        "y": 111
+      },
+      "hideTimeOverride": false,
+      "id": 51,
+      "links": [],
+      "maxDataPoints": 100,
+      "options": {
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": [
+            "last"
+          ],
+          "fields": "",
+          "values": false
+        },
+        "textMode": "auto"
+      },
+      "targets": [
+        {
+          "datasource": null,
+          "expr": "s3_lifecycle_latest_batch_bucket_count{job=\"${job_lifecycle_producer}\", namespace=\"${namespace}\"}",
+          "format": "time_series",
+          "hide": false,
+          "instant": false,
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "",
+          "metric": "",
+          "refId": "",
+          "step": 10,
+          "target": ""
+        }
+      ],
+      "title": "Buckets Listed",
+      "transformations": [],
+      "transparent": false,
+      "type": "stat"
+    },
+    {
+      "datasource": "${DS_PROMETHEUS}",
+      "description": "Rate of messages received by bucket processor, by pod",
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 0,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "lineInterpolation": "smooth",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "log": 2,
+              "type": "linear"
+            },
+            "showPoints": "auto",
+            "spanNulls": false,
+            "stacking": {},
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": []
+          },
+          "unit": ""
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 8,
+        "x": 0,
+        "y": 118
+      },
+      "hideTimeOverride": false,
+      "id": 52,
+      "links": [],
+      "maxDataPoints": 100,
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom"
+        },
+        "tooltip": {
+          "mode": "single"
+        }
+      },
+      "targets": [
+        {
+          "datasource": null,
+          "expr": "sum(rate(s3_lifecycle_bucket_processor_scan_messages_total{job=\"${job_lifecycle_bucket_processor}\", namespace=\"${namespace}\"}[$__rate_interval])) by (pod)",
+          "format": "time_series",
+          "hide": false,
+          "instant": false,
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "{{pod}}",
+          "metric": "",
+          "refId": "",
+          "step": 10,
+          "target": ""
+        }
+      ],
+      "title": "Bucket Tasks Picked Up by Pod",
+      "transformations": [],
+      "transparent": false,
+      "type": "timeseries"
+    },
+    {
+      "datasource": "${DS_PROMETHEUS}",
+      "description": "Total messages received by bucket-processor, by scan: each scan draws a rising line that plateaus when the scan completes (until its per-scan series expires). This panel is used to verify that there are no concurrent scans",
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 0,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "lineInterpolation": "smooth",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "log": 2,
+              "type": "linear"
+            },
+            "showPoints": "auto",
+            "spanNulls": false,
+            "stacking": {},
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": []
+          },
+          "unit": ""
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 8,
+        "x": 8,
+        "y": 118
+      },
+      "hideTimeOverride": false,
+      "id": 53,
+      "links": [],
+      "maxDataPoints": 100,
+      "options": {
+        "legend": {
+          "calcs": [],
+          "displayMode": "hidden",
+          "placement": "bottom"
+        },
+        "tooltip": {
+          "mode": "single"
+        }
+      },
+      "targets": [
+        {
+          "datasource": null,
+          "expr": "sum(increase(s3_lifecycle_bucket_processor_scan_messages_total{job=\"${job_lifecycle_bucket_processor}\", namespace=\"${namespace}\"}[$__rate_interval])) by (conductor_scan_id)",
+          "format": "time_series",
+          "hide": false,
+          "instant": false,
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "{{conductor_scan_id}}",
+          "metric": "",
+          "refId": "",
+          "step": 10,
+          "target": ""
+        }
+      ],
+      "title": "Bucket Processor Recent Conductor Scans",
+      "transformations": [],
+      "transparent": false,
+      "type": "timeseries"
+    },
+    {
+      "cards": {
+        "cardPadding": 1,
+        "cardRound": 2
+      },
+      "color": {
+        "cardColor": "#b4ff00",
+        "colorScale": "sqrt",
+        "colorScheme": "interpolateOranges",
+        "exponent": 0.5,
+        "max": null,
+        "min": null,
+        "mode": "opacity"
+      },
+      "dataFormat": "tsbuckets",
+      "datasource": "${DS_PROMETHEUS}",
+      "description": "Elapsed wall-clock time between conductor scan start and bucket-task message pickup by bucket processors.",
+      "editable": true,
+      "error": false,
+      "fieldConfig": {
+        "defaults": {
+          "thresholds": {
+            "mode": "absolute",
+            "steps": []
+          }
+        }
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 8,
+        "x": 16,
+        "y": 118
+      },
+      "heatmap": {},
+      "hideTimeOverride": false,
+      "hideZeroBuckets": true,
+      "highlightCards": true,
+      "id": 54,
+      "legend": {
+        "show": false
+      },
+      "links": [],
+      "maxDataPoints": 25,
+      "reverseYBuckets": false,
+      "targets": [
+        {
+          "datasource": null,
+          "expr": "sum(increase(s3_lifecycle_bucket_processor_scan_message_age_seconds_bucket{job=\"${job_lifecycle_bucket_processor}\", namespace=\"${namespace}\"}[$__rate_interval])) by(le)",
+          "format": "heatmap",
+          "hide": false,
+          "instant": false,
+          "interval": "",
+          "intervalFactor": 1,
+          "legendFormat": "{{le}}",
+          "metric": "",
+          "refId": "",
+          "step": 10,
+          "target": ""
+        }
+      ],
+      "title": "Bucket Processor Scan Message Age",
+      "tooltip": {
+        "show": true,
+        "showHistogram": true
+      },
+      "transformations": [],
+      "transparent": false,
+      "type": "heatmap",
+      "xAxis": {
+        "mode": "time",
+        "name": null,
+        "show": true,
+        "values": []
+      },
+      "yAxis": {
+        "decimals": 0,
+        "format": "s",
+        "label": null,
+        "logBase": 1,
+        "max": null,
+        "min": null,
+        "show": true
+      }
+    },
     {
       "datasource": "${DS_PROMETHEUS}",
       "editable": true,
@@ -4442,10 +4861,10 @@
         "h": 7,
         "w": 8,
         "x": 0,
-        "y": 111
+        "y": 125
       },
       "hideTimeOverride": false,
-      "id": 50,
+      "id": 55,
       "links": [],
       "maxDataPoints": 100,
       "options": {
@@ -4555,10 +4974,10 @@
         "h": 7,
         "w": 8,
         "x": 8,
-        "y": 111
+        "y": 125
       },
       "hideTimeOverride": false,
-      "id": 51,
+      "id": 56,
       "links": [],
       "maxDataPoints": 100,
       "options": {
@@ -4640,10 +5059,10 @@
         "h": 7,
         "w": 8,
         "x": 16,
-        "y": 111
+        "y": 125
       },
       "hideTimeOverride": false,
-      "id": 52,
+      "id": 57,
       "links": [],
       "maxDataPoints": 100,
       "options": {
diff --git a/monitoring/lifecycle/dashboard.py b/monitoring/lifecycle/dashboard.py
index bf11220acc..d2ae9cfab5 100644
--- a/monitoring/lifecycle/dashboard.py
+++ b/monitoring/lifecycle/dashboard.py
@@ -52,6 +52,11 @@ class Metrics:
         job='${job_lifecycle_producer}', namespace='${namespace}',
     )
 
+    LATEST_BATCH_END_TIME = metrics.Metric(
+        's3_lifecycle_latest_batch_end_time',
+        job='${job_lifecycle_producer}', namespace='${namespace}',
+    )
+
     BUCKET_LISTING_SUCCESS, BUCKET_LISTING_ERROR, BUCKET_LISTING_THROTTLING = [
         metrics.CounterMetric(
             name, job='${job_lifecycle_producer}', namespace='${namespace}',
@@ -73,6 +78,21 @@ class Metrics:
         'status', job='${job_lifecycle_producer}', namespace='${namespace}',
     )
 
+    LATEST_BATCH_BUCKET_COUNT = metrics.Metric(
+        's3_lifecycle_latest_batch_bucket_count',
+        job='${job_lifecycle_producer}', namespace='${namespace}',
+    )
+
+    BUCKET_PROCESSOR_SCAN_MESSAGES_RECEIVED = metrics.CounterMetric(
+        's3_lifecycle_bucket_processor_scan_messages_total',
+        'conductor_scan_id', job='${job_lifecycle_bucket_processor}', namespace='${namespace}',
+    )
+
+    BUCKET_PROCESSOR_SCAN_MESSAGE_AGE = metrics.BucketMetric(
+        's3_lifecycle_bucket_processor_scan_message_age_seconds',
+        job='${job_lifecycle_bucket_processor}', namespace='${namespace}',
+    )
+
     S3_OPS = metrics.CounterMetric(
        's3_lifecycle_s3_operations_total',
        'origin', 'op', 'status', job=['$jobs'], namespace='${namespace}',
@@ -730,6 +750,96 @@ def color_override(name, color):
     ]
 )
 
+lifecycle_full_scan_duration = TimeSeries(
+    title="Conductor Batch Duration",
+    description="Duration of completed conductor scans over time. "
+        "Gaps appear while a scan is in progress (its end time has not "
+        "caught up with its start time yet).",
+    dataSource="${DS_PROMETHEUS}",
+    unit=UNITS.SECONDS,
+    legendCalcs=['min', 'mean', 'max'],
+    legendDisplayMode='table',
+    legendPlacement='right',
+    lineInterpolation='smooth',
+    targets=[
+        Target(
+            expr='((max(' + Metrics.LATEST_BATCH_END_TIME() + ') - '
+                'max(' + Metrics.LATEST_BATCH_START_TIME() + ')) > 0) / 1000',
+            legendFormat='scan duration',
+        ),
+    ],
+)
+
+lifecycle_scan_count_buckets = Stat(
+    title="Buckets Listed",
+    description="Number of buckets listed during the latest conductor scan.",
+    dataSource="${DS_PROMETHEUS}",
+    reduceCalc="last",
+    decimals=0,
+    noValue='0',
+    targets=[
+        Target(
+            expr=Metrics.LATEST_BATCH_BUCKET_COUNT(),
+        ),
+    ],
+    thresholds=[
+        Threshold('semi-dark-blue', 0, 0.0),
+    ],
+)
+
+bucket_processor_messages_received = TimeSeries(
+    title="Bucket Tasks Picked Up by Pod",
+    description="Rate of messages received by bucket processor, by pod",
+    dataSource="${DS_PROMETHEUS}",
+    lineInterpolation="smooth",
+    decimals=0,
+    targets=[
+        Target(
+            expr='sum(rate(' + Metrics.BUCKET_PROCESSOR_SCAN_MESSAGES_RECEIVED() + ')) '
+                'by (pod)',
+            legendFormat='{{pod}}',
+        ),
+    ],
+)
+
+bucket_processor_active_scan_ids = TimeSeries(
+    title="Bucket Processor Recent Conductor Scans",
+    description="Total messages received by bucket-processor, by scan: each "
+        "scan draws a rising line that plateaus when the scan completes "
+        "(until its per-scan series expires). "
+        "This panel is used to verify that there are no concurrent scans",
+    dataSource="${DS_PROMETHEUS}",
+    lineInterpolation="smooth",
+    decimals=0,
+    legendDisplayMode="hidden",
+    targets=[
+        Target(
+            expr='sum(increase(' + Metrics.BUCKET_PROCESSOR_SCAN_MESSAGES_RECEIVED() + ')) '
+                'by (conductor_scan_id)',
+            legendFormat='{{conductor_scan_id}}',
+        ),
+    ],
+)
+
+bucket_processor_scan_message_age = Heatmap(
+    title="Bucket Processor Scan Message Age",
+    description="Elapsed wall-clock time between conductor scan start and "
+        "bucket-task message pickup by bucket processors.",
+    dataSource="${DS_PROMETHEUS}",
+    dataFormat='tsbuckets',
+    hideZeroBuckets=True,
+    maxDataPoints=25,
+    tooltip=Tooltip(show=True, showHistogram=True),
+    yAxis=YAxis(format=UNITS.SECONDS, decimals=0),
+    cards={'cardPadding': 1, 'cardRound': 2},
+    color=HeatmapColor(mode='opacity'),
+    targets=[Target(
+        expr='sum(increase(' + Metrics.BUCKET_PROCESSOR_SCAN_MESSAGE_AGE.bucket() + ')) by(le)',
+        format="heatmap",
+        legendFormat="{{le}}",
+    )],
+)
+
 active_indexing_jobs = TimeSeries(
     title="Active Indexing jobs",
     dataSource="${DS_PROMETHEUS}",
@@ -897,6 +1007,12 @@ def color_override(name, color):
             layout.row([s3_delete_object_ops, s3_delete_mpu_ops], height=8),
             RowPanel(title="Lifecycle Conductor"),
             layout.row([lifecycle_scans, trigger_latency], height=7),
+            layout.row([lifecycle_full_scan_duration, *layout.resize([lifecycle_scan_count_buckets], width=4)], height=7),
+            layout.row([
+                bucket_processor_messages_received,
+                bucket_processor_active_scan_ids,
+                bucket_processor_scan_message_age,
+            ], height=7),
             layout.row([lifecycle_scan_rate, active_indexing_jobs, legacy_tasks], height=7),
         ]),
     )