From 0910a4a18691852684723e2a5947a7456ae7542f Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Thu, 27 Aug 2026 14:26:47 +0200
Subject: [PATCH 1/6] Add a helper telling whether a location is remote

Data on an `isCRR` location belongs to a remote site: it may be read, but
never deleted, and a version whose data still lives there has not been
localized yet. Several places need to ask that question.

Lifted verbatim from BB-813 (#2828), which introduces the same helper and
carries its unit test and the location fixture. Kept byte-identical so that
whichever branch lands second has this commit dropped as already applied,
rather than conflicting.

Issue: BB-811
---
 lib/util/locations.js | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)
 create mode 100644 lib/util/locations.js

diff --git a/lib/util/locations.js b/lib/util/locations.js
new file mode 100644
index 000000000..bb7c114e2
--- /dev/null
+++ b/lib/util/locations.js
@@ -0,0 +1,18 @@
+const locationsConfig = require('../../conf/locationConfig.json') || {};
+
+/**
+ * Tell whether a location holds data owned by a remote site.
+ *
+ * Data stored on such a location is remote production data: we may read it
+ * (e.g. to copy it locally), but we must never delete it.
+ *
+ * @param {String} dataStoreName - location name
+ * @return {Boolean} true if the location is a CRR (remote) location
+ */
+function isCRRLocation(dataStoreName) {
+    return Boolean(locationsConfig[dataStoreName]?.isCRR);
+}
+
+module.exports = {
+    isCRRLocation,
+};

From 2f21619a9e145202b25bb2310508e748ee4eb2e4 Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Wed, 26 Aug 2026 14:16:59 +0200
Subject: [PATCH 2/6] Read the mongo client config from the extension

mongoProcessorTask takes the mongo client config from
config.queuePopulator.mongo, which works when a queue populator is deployed
beside the processor. The D/R metadata sink deploys none, so its
configuration would have to carry a queuePopulator block it never runs.

Accept extensions.mongoProcessor.mongodb, validated with the shared
mongoJoi, and fall back to the queue populator's config when absent. The
fallback is guarded: a config omitting queuePopulator entirely would
otherwise throw before validation could report anything useful.

Issue: BB-811
---
 .../MongoProcessorConfigValidator.js          |  4 ++-
 .../mongoProcessor/mongoProcessorTask.js      |  7 ++--
 .../MongoProcessorConfigValidator.spec.js     | 35 +++++++++++++++++++
 3 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/extensions/mongoProcessor/MongoProcessorConfigValidator.js b/extensions/mongoProcessor/MongoProcessorConfigValidator.js
index b5ae49183..5d8154d8d 100644
--- a/extensions/mongoProcessor/MongoProcessorConfigValidator.js
+++ b/extensions/mongoProcessor/MongoProcessorConfigValidator.js
@@ -1,5 +1,6 @@
 const joi = require('joi');
-const { retryParamsJoi, probeServerJoi, logJoiOptional } = require('../../lib/config/configItems.joi');
+const { retryParamsJoi, probeServerJoi, logJoiOptional, mongoJoi } =
+    require('../../lib/config/configItems.joi');
 const { extensionConfigValidator } = require('../../lib/config/extensionConfigValidator');
 
 const { MAX_QUEUED_DEFAULT }  = require('../../lib/constants').backbeatConsumer;
@@ -7,6 +8,7 @@ const { MAX_QUEUED_DEFAULT }  = require('../../lib/constants').backbeatConsumer;
 const joiSchema = joi.object({
     topic: joi.string().required(),
     groupId: joi.string().required(),
+    mongodb: mongoJoi,
     retry: retryParamsJoi,
     concurrency: joi.number().greater(0).default(1),
     maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),
diff --git a/extensions/mongoProcessor/mongoProcessorTask.js b/extensions/mongoProcessor/mongoProcessorTask.js
index e0ccf0258..b29535282 100644
--- a/extensions/mongoProcessor/mongoProcessorTask.js
+++ b/extensions/mongoProcessor/mongoProcessorTask.js
@@ -18,9 +18,10 @@ const { startProbeServer } = require('../../lib/util/probe');
 const kafkaConfig = config.kafka;
 const mConfig = config.metrics;
 const mongoProcessorConfig = config.extensions.mongoProcessor;
-// TODO: consider whether we would want a separate mongo config
-// for the consumer side
-const mongoClientConfig = config.queuePopulator.mongo;
+// the queue populator's mongo config is the fallback for a deployment that
+// runs one beside this process; a D/R sink runs none, so it configures its own
+const mongoClientConfig =
+    mongoProcessorConfig.mongodb ?? config.queuePopulator?.mongo;
 
 const log = new werelogs.Logger('Backbeat:MongoProcessor:task');
 const mongoProcessorLogConfig = mongoProcessorConfig.log ?? config.log;
diff --git a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
index 9723c20a4..1ba4cf9dd 100644
--- a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
+++ b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
@@ -41,3 +41,38 @@ describe('MongoProcessorConfigValidator log override', () => {
         }
     });
 });
+
+describe('MongoProcessorConfigValidator mongodb', () => {
+    const mongodb = {
+        replicaSetHosts: 'mongo:27017',
+        database: 'datadb',
+        authCredentials: { username: 'u', password: 'p' },
+    };
+
+    it('should accept a mongodb client config', () => {
+        const validated = configValidator(globalConfig, { ...baseExtConfig, mongodb });
+        assert.strictEqual(validated.mongodb.replicaSetHosts, 'mongo:27017');
+        assert.strictEqual(validated.mongodb.database, 'datadb');
+        assert.deepStrictEqual(validated.mongodb.authCredentials,
+            { username: 'u', password: 'p' });
+    });
+
+    it('should leave mongodb undefined when not set, deferring to the ' +
+    'queue populator config', () => {
+        const validated = configValidator(globalConfig, baseExtConfig);
+        assert.strictEqual(validated.mongodb, undefined);
+    });
+
+    it('should reject credentials missing a password', () => {
+        let err;
+        try {
+            configValidator(globalConfig, {
+                ...baseExtConfig,
+                mongodb: { ...mongodb, authCredentials: { username: 'u' } },
+            });
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on partial credentials');
+    });
+});

From 39f8e9e04695de2edca122594641e6fe02334698 Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Thu, 27 Aug 2026 17:51:57 +0200
Subject: [PATCH 3/6] Whitelist health check addresses only when an API is
 served

The `server` section is optional, but the whitelist of addresses allowed to
reach the health checks was extended unconditionally, so a process serving no
API exited before starting: a D/R sink runs the mongo-processor alone, and
configuring a section it never reads to get past this is no answer.

Issue: BB-811
---
 lib/Config.js                        | 8 +++++---
 tests/unit/lib/config/Config.spec.js | 6 ++++++
 2 files changed, 11 insertions(+), 3 deletions(-)

diff --git a/lib/Config.js b/lib/Config.js
index 2a67c9398..9fdd2a655 100644
--- a/lib/Config.js
+++ b/lib/Config.js
@@ -119,9 +119,11 @@ class Config extends EventEmitter {
 
         // whitelist IP, CIDR for health checks
         const defaultHealthChecks = ['127.0.0.1/8', '::1'];
-        const healthChecks = parsedConfig.server.healthChecks;
-        healthChecks.allowFrom =
-            healthChecks.allowFrom.concat(defaultHealthChecks);
+        const healthChecks = parsedConfig.server?.healthChecks;
+        if (healthChecks) {
+            healthChecks.allowFrom =
+                healthChecks.allowFrom.concat(defaultHealthChecks);
+        }
 
         // additional certs checks
         if (parsedConfig.certFilePaths) {
diff --git a/tests/unit/lib/config/Config.spec.js b/tests/unit/lib/config/Config.spec.js
index 67d6ba047..f8cbc6af2 100644
--- a/tests/unit/lib/config/Config.spec.js
+++ b/tests/unit/lib/config/Config.spec.js
@@ -31,6 +31,12 @@ describe('Config', () => {
         assert.doesNotThrow(() => config._parseConfig(testConfig));
     });
 
+    it('should accept a config serving no API, as a D/R sink does', () => {
+        delete testConfig.server;
+        testConfig.extensions = { mongoProcessor: testConfig.extensions.mongoProcessor };
+        assert.doesNotThrow(() => config._parseConfig(testConfig));
+    });
+
     it('should throw an error when dataMoverTopic is not provided and transition is supported', () => {
         delete testConfig.extensions.replication.dataMoverTopic;
         testConfig.extensions.lifecycle.supportedLifecycleRules = [

From d9516c4820ccfd20e6c22c105051792238bbd26a Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Wed, 26 Aug 2026 14:23:20 +0200
Subject: [PATCH 4/6] Introduce a processor mode, defaulting to ingestion

The mongo-processor was written as the out-of-band ingestion consumer, and
the D/R metadata sink now reuses it. The two disagree about most of what it
does to an object, so put those decisions behind a mode: an abstract
ProcessorMode whose methods assert, an implementation per mode, and an index
mapping the configured name to the class, as the notification extension does
for its destinations.

IngestionMode carries today's behaviour verbatim, so this commit changes
nothing. The default lives beside the mode map, so a processor built
programmatically gets the same mode as one built from a config file.

Issue: BB-811
---
 .../MongoProcessorConfigValidator.js          |   2 +
 .../mongoProcessor/MongoQueueProcessor.js     | 127 ++-----------
 .../mongoProcessor/modes/IngestionMode.js     | 172 ++++++++++++++++++
 .../mongoProcessor/modes/ProcessorMode.js     | 107 +++++++++++
 extensions/mongoProcessor/modes/index.js      |  10 +
 .../MongoProcessorConfigValidator.spec.js     |  18 ++
 .../unit/mongoProcessor/ProcessorMode.spec.js |  24 +++
 7 files changed, 349 insertions(+), 111 deletions(-)
 create mode 100644 extensions/mongoProcessor/modes/IngestionMode.js
 create mode 100644 extensions/mongoProcessor/modes/ProcessorMode.js
 create mode 100644 extensions/mongoProcessor/modes/index.js
 create mode 100644 tests/unit/mongoProcessor/ProcessorMode.spec.js

diff --git a/extensions/mongoProcessor/MongoProcessorConfigValidator.js b/extensions/mongoProcessor/MongoProcessorConfigValidator.js
index 5d8154d8d..e4be3a26b 100644
--- a/extensions/mongoProcessor/MongoProcessorConfigValidator.js
+++ b/extensions/mongoProcessor/MongoProcessorConfigValidator.js
@@ -1,4 +1,5 @@
 const joi = require('joi');
+const { modes, defaultMode } = require('./modes');
 const { retryParamsJoi, probeServerJoi, logJoiOptional, mongoJoi } =
     require('../../lib/config/configItems.joi');
 const { extensionConfigValidator } = require('../../lib/config/extensionConfigValidator');
@@ -8,6 +9,7 @@ const { MAX_QUEUED_DEFAULT }  = require('../../lib/constants').backbeatConsumer;
 const joiSchema = joi.object({
     topic: joi.string().required(),
     groupId: joi.string().required(),
+    mode: joi.string().valid(...Object.keys(modes)).default(defaultMode),
     mongodb: mongoJoi,
     retry: retryParamsJoi,
     concurrency: joi.number().greater(0).default(1),
diff --git a/extensions/mongoProcessor/MongoQueueProcessor.js b/extensions/mongoProcessor/MongoQueueProcessor.js
index d0b659354..c67370a58 100644
--- a/extensions/mongoProcessor/MongoQueueProcessor.js
+++ b/extensions/mongoProcessor/MongoQueueProcessor.js
@@ -4,11 +4,10 @@ const async = require('async');
 
 const Logger = require('werelogs').Logger;
 const errors = require('arsenal').errors;
-const { replicationBackends, emptyFileMd5 } = require('arsenal').constants;
+const { replicationBackends } = require('arsenal').constants;
 const MongoClient = require('arsenal').storage
     .metadata.mongoclient.MongoClientInterface;
 const { ObjectMD, ReplicationConfiguration } = require('arsenal').models;
-const { VersionID } = require('arsenal').versioning;
 const { extractVersionId } = require('../../lib/util/versioning');
 
 const Config = require('../../lib/Config');
@@ -19,7 +18,7 @@ const ObjectQueueEntry = require('../../lib/models/ObjectQueueEntry');
 const MetricsProducer = require('../../lib/MetricsProducer');
 const { metricsExtension, metricsTypeCompleted, metricsTypePendingOnly } =
     require('../ingestion/constants');
-const getContentType = require('./utils/contentTypeHelper');
+const { modes, defaultMode } = require('./modes');
 const BucketMemState = require('./utils/BucketMemState');
 const MongoProcessorMetrics = require('./MongoProcessorMetrics');
 
@@ -75,6 +74,8 @@ class MongoQueueProcessor {
         this._bootstrapList = null;
         this.logger = new Logger('Backbeat:Ingestion:MongoProcessor');
         this.mongoClientConfig.logger = this.logger;
+        this._mode =
+            new modes[mongoProcessorConfig.mode ?? defaultMode]();
         this._mongoClient = new MongoClient(this.mongoClientConfig);
         this._bucketMemState = new BucketMemState(Config);
 
@@ -236,78 +237,6 @@ class MongoQueueProcessor {
         });
     }
 
-    /**
-     * Update ingested entry metadata fields: owner-id, owner-display-name
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @param {BucketInfo} bucketInfo - bucket info object
-     * @return {undefined}
-     */
-    _updateOwnerMD(entry, bucketInfo) {
-        // zenko bucket owner information is being set on ingested md
-        entry.setOwnerDisplayName(bucketInfo.getOwnerDisplayName());
-        entry.setOwnerId(bucketInfo.getOwner());
-    }
-
-    /**
-     * Update ingested entry metadata fields: dataStoreName
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @param {string} location - owner details
-     * @return {undefined}
-     */
-    _updateObjectDataStoreName(entry, location) {
-        entry.setDataStoreName(location);
-    }
-
-    /**
-     * Update ingested entry metadata location field. Each location change
-     * includes: key, dataStoreName, dataStoreType, dataStoreVersionId
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @param {string} zenkoLocation - zenko storage location name
-     * @return {undefined}
-     */
-    _updateLocations(entry, zenkoLocation) {
-        const locations = entry.getLocation();
-        // if version id is undefined, we have a single null object.
-        // To hold reference to this null object, we need to encode "null"
-        // as its dataStoreVersionId
-        const dataStoreVersionId = entry.getVersionId() ?
-            entry.getEncodedVersionId() : 'null';
-        let zenkoDataLocations;
-        if (!locations || locations.length === 0) {
-            zenkoDataLocations = [{
-                key: entry.getObjectKey(),
-                size: 0,
-                start: 0,
-                dataStoreName: zenkoLocation,
-                dataStoreType: 'aws_s3',
-                dataStoreETag: `1:${emptyFileMd5}`,
-                dataStoreVersionId,
-            }];
-        } else {
-            zenkoDataLocations = [{
-                key: entry.getObjectKey(),
-                size: entry.getContentLength(),
-                start: 0,
-                dataStoreName: zenkoLocation,
-                dataStoreType: 'aws_s3',
-                dataStoreETag: `1:${entry.getContentMd5()}`,
-                dataStoreVersionId,
-            }];
-        }
-        entry.setLocation(zenkoDataLocations);
-    }
-
-    /**
-     * Update acl info on ingested object MD
-     * @param {ObjectQueueEntry} entry - object queue entry object
-     * @return {undefined}
-     */
-    _updateAcl(entry) {
-        // reset acl info
-        const objectMDModel = new ObjectMD();
-        entry.setAcl(objectMDModel.getAcl());
-    }
-
     /**
      * Update replication info on ingested object MD to match Zenko defined
      * replication info.
@@ -367,29 +296,17 @@ class MongoQueueProcessor {
         const key = sourceEntry.getObjectKey();
         const entryVersionId = extractVersionId(sourceEntry.getObjectVersionedKey());
 
-        // Use x-amz-meta-scal-version-id if provided, instead of the actual versionId of the object.
-        // This should happen only for restored objects : in all other situations, both the source
-        // and ingested objects should have the same version id (and no x-amz-meta-scal-version-id
-        // metadata).
         const scalVersionId = sourceEntry.getOverheadField('x-amz-meta-scal-version-id');
-        const versionId = scalVersionId ? VersionID.decode(scalVersionId) : entryVersionId;
+        const versionId =
+            this._mode.resolveVersionId(scalVersionId, entryVersionId);
 
         this.logger.debug('processing object delete', { bucket, key, versionId });
 
         async.waterfall([
             cb => this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb),
             (zenkoObjMd, cb) => {
-                // Skip if the object is in a different location, i.e. when the delete was caused
-                // by restored-object expiration or transition. It works because the dataStoreName
-                // is updated before actually sending the object to GC to effectively delete the
-                // data.
-                const encode = versionId => (versionId ? VersionID.encode(versionId) : 'null');
-                if (zenkoObjMd.dataStoreName !== location ||
-                    zenkoObjMd.location?.length !== 1 ||
-                    zenkoObjMd.location[0].dataStoreName !== location ||
-                    zenkoObjMd.location[0].key !== key ||
-                    (zenkoObjMd.location[0].dataStoreVersionId || 'null') !== encode(entryVersionId)
-                ) {
+                if (!this._mode.shouldProcessDelete(zenkoObjMd, location, key,
+                                                   entryVersionId)) {
                     log.end().info('ignore delete entry, transitioned to another location', {
                         entry: sourceEntry.getLogInfo(),
                         location,
@@ -467,19 +384,12 @@ class MongoQueueProcessor {
         this.logger.debug('processing object metadata', { bucket, key, scalVersionId });
 
         const maybeGetZenkoObjectMetadata = cb => {
-            // NOTE: ZenkoObjMD is used for updating replication info, as well as validating the
-            // `x-amz-meta-scal-version-id` header of restored objects. If the Zenko bucket does
-            // not have repInfo set and the header is not set, then we can skip fetching.
-            const bucketRepInfo = bucketInfo.getReplicationConfiguration();
-            if (!scalVersionId && !bucketRepInfo?.rules?.some(r => r.enabled)) {
+            if (!this._mode.needsExistingMetadata(sourceEntry, bucketInfo)) {
                 return cb();
             }
 
-            // Use x-amz-meta-scal-version-id if provided, instead of the actual versionId of the object.
-            // This should happen only for restored objects : in all other situations, both the source
-            // and ingested objects should have the same version id (and not x-amz-meta-scal-version-id
-            // metadata).
-            const versionId = scalVersionId ? VersionID.decode(scalVersionId) : sourceEntry.getVersionId();
+            const versionId = this._mode.resolveVersionId(scalVersionId,
+                sourceEntry.getVersionId());
             return this._getZenkoObjectMetadata(log, sourceEntry, versionId, cb);
         };
 
@@ -494,7 +404,8 @@ class MongoQueueProcessor {
                 return done(err);
             }
 
-            const content = getContentType(sourceEntry, zenkoObjMd);
+            const content =
+                this._mode.getChangedContent(sourceEntry, zenkoObjMd);
             if (content.length === 0) {
                 this._normalizePendingMetric(location);
                 log.end().debug('skipping duplicate entry', {
@@ -507,16 +418,10 @@ class MongoQueueProcessor {
             }
 
             if (zenkoObjMd) {
-                // Keep existing metadata fields, only need to update the tags
-                const tags = sourceEntry.getTags();
-                sourceEntry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign
-                sourceEntry.setTags(tags);
+                this._mode.mergeExistingMetadata(sourceEntry, zenkoObjMd);
             } else {
-                // Update necessary metadata fields before saving to Zenko MongoDB
-                this._updateOwnerMD(sourceEntry, bucketInfo);
-                this._updateObjectDataStoreName(sourceEntry, location);
-                this._updateLocations(sourceEntry, location);
-                this._updateAcl(sourceEntry);
+                this._mode.applyNewObjectMetadata(sourceEntry, location,
+                                                 bucketInfo);
             }
 
             // Try to update replication info, if applicable
diff --git a/extensions/mongoProcessor/modes/IngestionMode.js b/extensions/mongoProcessor/modes/IngestionMode.js
new file mode 100644
index 000000000..dfce04034
--- /dev/null
+++ b/extensions/mongoProcessor/modes/IngestionMode.js
@@ -0,0 +1,172 @@
+'use strict';
+
+const { emptyFileMd5 } = require('arsenal').constants;
+const { ObjectMD } = require('arsenal').models;
+const { VersionID } = require('arsenal').versioning;
+
+const ProcessorMode = require('./ProcessorMode');
+const getContentType = require('../utils/contentTypeHelper');
+
+class IngestionMode extends ProcessorMode {
+    /**
+     * ZenkoObjMD is used for updating replication info, as well as validating
+     * the `x-amz-meta-scal-version-id` header of restored objects. If the Zenko
+     * bucket does not have repInfo set and the header is not set, then we can
+     * skip fetching.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {boolean} true if the stored document is needed
+     */
+    needsExistingMetadata(entry, bucketInfo) {
+        const scalVersionId = entry.getValue()['x-amz-meta-scal-version-id'];
+        const bucketRepInfo = bucketInfo.getReplicationConfiguration();
+
+        return !!scalVersionId || !!bucketRepInfo?.rules?.some(r => r.enabled);
+    }
+
+    /**
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo
+     * @return {Array} array of ReplicationInfo Content Type
+     */
+    getChangedContent(entry, zenkoObjMd) {
+        return getContentType(entry, zenkoObjMd);
+    }
+
+    /**
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {string} location - zenko storage location name
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {undefined}
+     */
+    applyNewObjectMetadata(entry, location, bucketInfo) {
+        this._updateOwnerMD(entry, bucketInfo);
+        this._updateObjectDataStoreName(entry, location);
+        this._updateLocations(entry, location);
+        this._updateAcl(entry);
+    }
+
+    /**
+     * Update ingested entry metadata fields: owner-id, owner-display-name
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {undefined}
+     */
+    _updateOwnerMD(entry, bucketInfo) {
+        // zenko bucket owner information is being set on ingested md
+        entry.setOwnerDisplayName(bucketInfo.getOwnerDisplayName());
+        entry.setOwnerId(bucketInfo.getOwner());
+    }
+
+    /**
+     * Update ingested entry metadata fields: dataStoreName
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {string} location - owner details
+     * @return {undefined}
+     */
+    _updateObjectDataStoreName(entry, location) {
+        entry.setDataStoreName(location);
+    }
+
+    /**
+     * Update ingested entry metadata location field. Each location change
+     * includes: key, dataStoreName, dataStoreType, dataStoreVersionId
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {string} zenkoLocation - zenko storage location name
+     * @return {undefined}
+     */
+    _updateLocations(entry, zenkoLocation) {
+        const locations = entry.getLocation();
+        // if version id is undefined, we have a single null object.
+        // To hold reference to this null object, we need to encode "null"
+        // as its dataStoreVersionId
+        const dataStoreVersionId = entry.getVersionId() ?
+            entry.getEncodedVersionId() : 'null';
+        let zenkoDataLocations;
+        if (!locations || locations.length === 0) {
+            zenkoDataLocations = [{
+                key: entry.getObjectKey(),
+                size: 0,
+                start: 0,
+                dataStoreName: zenkoLocation,
+                dataStoreType: 'aws_s3',
+                dataStoreETag: `1:${emptyFileMd5}`,
+                dataStoreVersionId,
+            }];
+        } else {
+            zenkoDataLocations = [{
+                key: entry.getObjectKey(),
+                size: entry.getContentLength(),
+                start: 0,
+                dataStoreName: zenkoLocation,
+                dataStoreType: 'aws_s3',
+                dataStoreETag: `1:${entry.getContentMd5()}`,
+                dataStoreVersionId,
+            }];
+        }
+        entry.setLocation(zenkoDataLocations);
+    }
+
+    /**
+     * Update acl info on ingested object MD
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @return {undefined}
+     */
+    _updateAcl(entry) {
+        // reset acl info
+        const objectMDModel = new ObjectMD();
+        entry.setAcl(objectMDModel.getAcl());
+    }
+
+    /**
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {undefined}
+     */
+    mergeExistingMetadata(entry, zenkoObjMd) {
+        // Keep existing metadata fields, only need to update the tags
+        const tags = entry.getTags();
+        entry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign
+        entry.setTags(tags);
+    }
+
+    /**
+     * Use x-amz-meta-scal-version-id if provided, instead of the actual
+     * versionId of the object. This should happen only for restored objects:
+     * in all other situations, both the source and ingested objects should have
+     * the same version id (and no x-amz-meta-scal-version-id metadata).
+     *
+     * @param {string|undefined} scalVersionId - encoded scal version id
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {string|undefined} version id to act on
+     */
+    resolveVersionId(scalVersionId, versionId) {
+        return scalVersionId ? VersionID.decode(scalVersionId) : versionId;
+    }
+
+    /**
+     * Skip if the object is in a different location, i.e. when the delete was
+     * caused by restored-object expiration or transition. It works because the
+     * dataStoreName is updated before actually sending the object to GC to
+     * effectively delete the data.
+     *
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @param {string} location - zenko storage location name
+     * @param {string} key - object key
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {boolean} true if the object should be deleted
+     */
+    shouldProcessDelete(zenkoObjMd, location, key, versionId) {
+        const encode = vid => (vid ? VersionID.encode(vid) : 'null');
+
+        return zenkoObjMd.dataStoreName === location &&
+            zenkoObjMd.location?.length === 1 &&
+            zenkoObjMd.location[0].dataStoreName === location &&
+            zenkoObjMd.location[0].key === key &&
+            (zenkoObjMd.location[0].dataStoreVersionId || 'null') ===
+                encode(versionId);
+    }
+}
+
+module.exports = IngestionMode;
diff --git a/extensions/mongoProcessor/modes/ProcessorMode.js b/extensions/mongoProcessor/modes/ProcessorMode.js
new file mode 100644
index 000000000..d17ea7570
--- /dev/null
+++ b/extensions/mongoProcessor/modes/ProcessorMode.js
@@ -0,0 +1,107 @@
+'use strict';
+
+const assert = require('assert');
+
+/**
+ * A processor mode holds the decisions that differ between the streams the
+ * mongo-processor writes: out-of-band ingestion, where identity and placement
+ * are rewritten to local values because the source system's accounts and
+ * locations do not exist here, and D/R, where they are replicated and so are
+ * applied as they arrive.
+ *
+ * Everything else about processing an entry is shared.
+ */
+class ProcessorMode {
+    /**
+     * Whether the entry's existing metadata has to be read before processing.
+     *
+     * This method must be implemented by subclasses of ProcessorMode
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {boolean} true if the stored document is needed
+     */
+    needsExistingMetadata(entry, bucketInfo) { // eslint-disable-line no-unused-vars
+        assert(false,
+            'sub-classes of ProcessorMode must implement ' +
+            'the needsExistingMetadata() method');
+    }
+
+    /**
+     * What changed between the entry and the object already stored, as
+     * replicationInfo content values. An empty list means the entry carries no
+     * change and is not written.
+     *
+     * This method must be implemented by subclasses of ProcessorMode
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo
+     * @return {Array} array of ReplicationInfo Content Type
+     */
+    getChangedContent(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars
+        assert(false,
+            'sub-classes of ProcessorMode must implement ' +
+            'the getChangedContent() method');
+    }
+
+    /**
+     * Apply the metadata fields an object gets when it is first written here.
+     *
+     * This method must be implemented by subclasses of ProcessorMode
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {string} location - zenko storage location name
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {undefined}
+     */
+    applyNewObjectMetadata(entry, location, bucketInfo) { // eslint-disable-line no-unused-vars
+        assert(false,
+            'sub-classes of ProcessorMode must implement ' +
+            'the applyNewObjectMetadata() method');
+    }
+
+    /**
+     * Merge the entry into the object already stored, deciding which fields the
+     * entry brings and which the stored document keeps.
+     *
+     * This method must be implemented by subclasses of ProcessorMode
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {undefined}
+     */
+    mergeExistingMetadata(entry, zenkoObjMd) { // eslint-disable-line no-unused-vars
+        assert(false,
+            'sub-classes of ProcessorMode must implement ' +
+            'the mergeExistingMetadata() method');
+    }
+
+    /**
+     * Which version of the object the entry acts on, given the version id it
+     * carries and the `x-amz-meta-scal-version-id` it may carry alongside.
+     *
+     * This method must be implemented by subclasses of ProcessorMode
+     * @param {string|undefined} scalVersionId - encoded scal version id
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {string|undefined} version id to act on
+     */
+    resolveVersionId(scalVersionId, versionId) { // eslint-disable-line no-unused-vars
+        assert(false,
+            'sub-classes of ProcessorMode must implement ' +
+            'the resolveVersionId() method');
+    }
+
+    /**
+     * Whether a delete entry still applies to the object as stored.
+     *
+     * This method must be implemented by subclasses of ProcessorMode
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @param {string} location - zenko storage location name
+     * @param {string} key - object key
+     * @param {string|undefined} versionId - decoded version id of the entry
+     * @return {boolean} true if the object should be deleted
+     */
+    shouldProcessDelete(zenkoObjMd, location, key, versionId) { // eslint-disable-line no-unused-vars
+        assert(false,
+            'sub-classes of ProcessorMode must implement ' +
+            'the shouldProcessDelete() method');
+    }
+}
+
+module.exports = ProcessorMode;
diff --git a/extensions/mongoProcessor/modes/index.js b/extensions/mongoProcessor/modes/index.js
new file mode 100644
index 000000000..b534acbb8
--- /dev/null
+++ b/extensions/mongoProcessor/modes/index.js
@@ -0,0 +1,10 @@
+const IngestionMode = require('./IngestionMode');
+
+const modes = {
+    ingestion: IngestionMode,
+};
+
+module.exports = {
+    modes,
+    defaultMode: 'ingestion',
+};
diff --git a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
index 1ba4cf9dd..505c70665 100644
--- a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
+++ b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
@@ -76,3 +76,21 @@ describe('MongoProcessorConfigValidator mongodb', () => {
         assert(err, 'expected configValidator to throw on partial credentials');
     });
 });
+
+describe('MongoProcessorConfigValidator mode', () => {
+    it('should default to ingestion so an existing config is unchanged', () => {
+        const validated = configValidator(globalConfig, baseExtConfig);
+        assert.strictEqual(validated.mode, 'ingestion');
+    });
+
+    it('should reject a mode with no implementation', () => {
+        let err;
+        try {
+            configValidator(globalConfig, { ...baseExtConfig, mode: 'sideways' });
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on an unknown mode');
+        assert.match(err.message, /mode/);
+    });
+});
diff --git a/tests/unit/mongoProcessor/ProcessorMode.spec.js b/tests/unit/mongoProcessor/ProcessorMode.spec.js
new file mode 100644
index 000000000..5e385260a
--- /dev/null
+++ b/tests/unit/mongoProcessor/ProcessorMode.spec.js
@@ -0,0 +1,24 @@
+'use strict';
+
+const assert = require('assert');
+
+const ProcessorMode =
+    require('../../../extensions/mongoProcessor/modes/ProcessorMode');
+
+describe('ProcessorMode', () => {
+    const mode = new ProcessorMode();
+
+    [
+        'needsExistingMetadata',
+        'getChangedContent',
+        'applyNewObjectMetadata',
+        'mergeExistingMetadata',
+        'resolveVersionId',
+        'shouldProcessDelete',
+    ].forEach(method => it(`should refuse to ${method}() without an ` +
+    'implementation', () => {
+        assert.throws(() => mode[method](),
+            new RegExp('sub-classes of ProcessorMode must implement the ' +
+                `${method}\\(\\) method`));
+    }));
+});

From eb60f94da56b5b33da026570a00f193b762229e7 Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Wed, 26 Aug 2026 14:56:00 +0200
Subject: [PATCH 5/6] Add the D/R mode

The D/R metadata sink replicates production's objects, accounts included, so
it applies what the source-side pipeline sends rather than rewriting it into
something local: a new object is written as it arrives, and an update takes
the entry's tags, object-lock state and ACLs while keeping the placement
already stored.

Keeping the stored placement is the point. The copy engine rewrites location
and dataStoreName to a local location after the first write, and applying the
entry's would send reads back to the source and leak a local copy that is
never garbage-collected. Cleared values are applied like any other: removing
a legal hold is an update.

Three things follow from that and were unreachable before it:

- the stored document is always read, because it is what distinguishes a
  first write from an update, and the entry cannot -- an insert is redelivered
  on replay and overlaps the bootstrap dump, so it is no promise that the
  object is absent here;
- object-lock and ACL changes count as changes, where the ingestion diff
  looks only at tags and dropped them as duplicates;
- a delete always applies, where the ingestion guard skips one whose object
  has moved location, which for a replicated object it always has.

Issue: BB-811
---
 extensions/mongoProcessor/modes/DRMode.js     | 133 ++++++++++
 extensions/mongoProcessor/modes/index.js      |   2 +
 .../ingestion/MongoQueueProcessor.js          | 237 ++++++++++++++++++
 .../MongoProcessorConfigValidator.spec.js     |  15 ++
 4 files changed, 387 insertions(+)
 create mode 100644 extensions/mongoProcessor/modes/DRMode.js

diff --git a/extensions/mongoProcessor/modes/DRMode.js b/extensions/mongoProcessor/modes/DRMode.js
new file mode 100644
index 000000000..1c3d0742d
--- /dev/null
+++ b/extensions/mongoProcessor/modes/DRMode.js
@@ -0,0 +1,133 @@
+'use strict';
+
+const { ObjectMD } = require('arsenal').models;
+
+const locations = require('../../../lib/util/locations');
+const ProcessorMode = require('./ProcessorMode');
+const getContentType = require('../utils/contentTypeHelper');
+
+class DRMode extends ProcessorMode {
+    /**
+     * The stored document tells a first write from an update: a source insert
+     * is redelivered on replay and overlaps the bootstrap dump.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {boolean} true if the stored document is needed
+     */
+    needsExistingMetadata(entry, bucketInfo) { // eslint-disable-line no-unused-vars
+        return true;
+    }
+
+    /**
+     * The ingestion diff only covers tags; a replicated object can also change
+     * its object-lock state and, until localized, its placement.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object|undefined} zenkoObjMd - metadata fetched from mongo
+     * @return {Array} array of ReplicationInfo Content Type
+     */
+    getChangedContent(entry, zenkoObjMd) {
+        const content = getContentType(entry, zenkoObjMd);
+        if (!zenkoObjMd || content.length !== 0) {
+            return content;
+        }
+
+        return this._hasMutableChange(entry, zenkoObjMd) ? ['METADATA'] : [];
+    }
+
+    /**
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {boolean} true if the entry changes mutable metadata
+     */
+    _hasMutableChange(entry, zenkoObjMd) {
+        return entry.getRetentionMode() !== zenkoObjMd.retentionMode ||
+            entry.getRetentionDate() !== zenkoObjMd.retentionDate ||
+            entry.getLegalHold() !== !!zenkoObjMd.legalHold ||
+            (this._isNotLocalized(zenkoObjMd) &&
+                entry.getDataStoreName() !== zenkoObjMd.dataStoreName);
+    }
+
+    /**
+     * A version whose data still lives on the remote site.
+     *
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {boolean} true if the stored version is not localized
+     */
+    _isNotLocalized(zenkoObjMd) {
+        return locations.isCRRLocation(zenkoObjMd.dataStoreName);
+    }
+
+    /**
+     * The source-side pipeline shaped everything this object keeps. ACLs are
+     * not replicated, so they are reset as they are for an ingested object.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {string} location - zenko storage location name
+     * @param {BucketInfo} bucketInfo - bucket info object
+     * @return {undefined}
+     */
+    applyNewObjectMetadata(entry, location, bucketInfo) { // eslint-disable-line no-unused-vars
+        entry.setAcl(new ObjectMD().getAcl());
+    }
+
+    /**
+     * An update brings tags and object-lock state, cleared values included. A
+     * localized version keeps the placement the copy engine gave it, one still
+     * on the remote site takes the entry's.
+     *
+     * @param {ObjectQueueEntry} entry - object queue entry object
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @return {undefined}
+     */
+    mergeExistingMetadata(entry, zenkoObjMd) {
+        const tags = entry.getTags();
+        const retentionMode = entry.getRetentionMode();
+        const retentionDate = entry.getRetentionDate();
+        const legalHold = entry.getLegalHold();
+        const notLocalized = this._isNotLocalized(zenkoObjMd);
+        const dataStoreName = entry.getDataStoreName();
+        const location = entry.getLocation();
+
+        entry._data = { ...zenkoObjMd }; // eslint-disable-line no-param-reassign
+
+        entry.setTags(tags);
+        entry.setRetentionMode(retentionMode);
+        entry.setRetentionDate(retentionDate);
+        entry.setLegalHold(legalHold);
+
+        if (notLocalized) {
+            entry.setDataStoreName(dataStoreName);
+            entry.setLocation(location);
+        }
+    }
+
+    /**
+     * Version ids are identical on both sides, so the entry's is authoritative:
+     * a scal version id names a version of another system entirely.
+     *
+     * @param {string|undefined} scalVersionId - encoded scal version id
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {string|undefined} version id to act on
+     */
+    resolveVersionId(scalVersionId, versionId) {
+        return versionId;
+    }
+
+    /**
+     * A replicated object's location legitimately differs, so the ingestion
+     * guard would ignore every deletion.
+     *
+     * @param {Object} zenkoObjMd - metadata fetched from mongo
+     * @param {string} location - zenko storage location name
+     * @param {string} key - object key
+     * @param {string|undefined} versionId - version id the entry carries
+     * @return {boolean} true if the object should be deleted
+     */
+    shouldProcessDelete(zenkoObjMd, location, key, versionId) { // eslint-disable-line no-unused-vars
+        return true;
+    }
+}
+
+module.exports = DRMode;
diff --git a/extensions/mongoProcessor/modes/index.js b/extensions/mongoProcessor/modes/index.js
index b534acbb8..31def379c 100644
--- a/extensions/mongoProcessor/modes/index.js
+++ b/extensions/mongoProcessor/modes/index.js
@@ -1,7 +1,9 @@
+const DRMode = require('./DRMode');
 const IngestionMode = require('./IngestionMode');
 
 const modes = {
     ingestion: IngestionMode,
+    dr: DRMode,
 };
 
 module.exports = {
diff --git a/tests/functional/ingestion/MongoQueueProcessor.js b/tests/functional/ingestion/MongoQueueProcessor.js
index 626b825a8..427ae529a 100644
--- a/tests/functional/ingestion/MongoQueueProcessor.js
+++ b/tests/functional/ingestion/MongoQueueProcessor.js
@@ -16,6 +16,7 @@ const authdata = require('../../../conf/authdata.json');
 const ObjectQueueEntry = require('../../../lib/models/ObjectQueueEntry');
 const DeleteOpQueueEntry = require('../../../lib/models/DeleteOpQueueEntry');
 const fakeLogger = require('../../utils/fakeLogger');
+const locations = require('../../../lib/util/locations');
 const { ObjectMDArchive, LifecycleConfiguration, NotificationConfiguration } = require('arsenal/build/lib/models');
 
 const kafkaConfig = config.kafka;
@@ -1068,3 +1069,239 @@ describe('MongoQueueProcessor', function mqp() {
         });
     });
 });
+
+describe('MongoQueueProcessor in dr mode', function drMode() {
+    this.timeout(5000);
+
+    let mqp;
+    let mongoClient;
+
+    before(() => {
+        mqp = new MongoQueueProcessorMock(kafkaConfig,
+            { ...mongoProcessorConfig, mode: 'dr' }, mongoClientConfig, mConfig);
+        mqp.start();
+
+        mongoClient = mqp._mongoClient;
+    });
+
+    afterEach(() => {
+        mqp.reset();
+        sinon.restore();
+    });
+
+    function processEntry(entry, next) {
+        return async.waterfall([
+            cb => mongoClient.getBucketAttributes(BUCKET, fakeLogger, cb),
+            (bucketInfo, cb) => mqp._processObjectQueueEntry(fakeLogger, entry,
+                LOCATION, bucketInfo, cb),
+        ], next);
+    }
+
+    it('should apply a new object as the source describes it', done => {
+        const key = 'dr-new-key';
+        const versionKey = `${key}${VID_SEP}${VERSION_ID}`;
+        const objmd = new ObjectMD()
+            .setKey(key)
+            .setVersionId(VERSION_ID)
+            .setOwnerId('source-owner-id')
+            .setOwnerDisplayName('source-owner')
+            .setDataStoreName('cold-location')
+            // ACLs are not replicated: the matrix in the design resets them
+            .setAcl({ Canned: '', FULL_CONTROL: ['source-grantee'],
+                WRITE_ACP: [], READ: [], READ_ACP: [] });
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+
+        processEntry(entry, err => {
+            assert.ifError(err);
+
+            const added = mqp.getAdded();
+            assert.strictEqual(added.length, 1);
+            const { objVal } = added[0];
+            assert.strictEqual(objVal['owner-id'], 'source-owner-id');
+            assert.strictEqual(objVal['owner-display-name'], 'source-owner');
+            assert.strictEqual(objVal.dataStoreName, 'cold-location');
+            assert.deepStrictEqual(objVal.acl, new ObjectMD().getAcl());
+            done();
+        });
+    });
+
+    it('should keep the stored placement when updating a localized object',
+    done => {
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        // the copy engine has since moved the object to a local location, so
+        // the entry's own placement must not be written back
+        const objmd = new ObjectMD()
+            .setKey(KEY)
+            .setVersionId(VERSION_ID)
+            .setTags({ mytag: 'mytags-value' })
+            .setDataStoreName('source-site')
+            .setLocation([{ key: KEY, dataStoreName: 'source-site' }])
+            .setLegalHold(true);
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+
+        processEntry(entry, err => {
+            assert.ifError(err);
+
+            const added = mqp.getAdded();
+            assert.strictEqual(added.length, 1);
+            const { objVal } = added[0];
+            // placement from the stored document
+            assert.strictEqual(objVal.dataStoreName, LOCATION);
+            assert.strictEqual(objVal.location[0].dataStoreName, LOCATION);
+            // mutable metadata from the entry
+            assert.strictEqual(objVal.legalHold, true);
+            done();
+        });
+    });
+
+    it('should skip a replayed entry that changes nothing', done => {
+        // at-least-once delivery means the same entry arrives twice; the second
+        // must be a no-op rather than another write
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        const objmd = new ObjectMD()
+            .setKey(KEY)
+            .setVersionId(VERSION_ID)
+            .setTags({ mytag: 'mytags-value' })
+            .setDataStoreName(LOCATION);
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+
+        processEntry(entry, err => {
+            assert.ifError(err);
+            assert.strictEqual(mqp.getAdded().length, 0);
+            done();
+        });
+    });
+
+    it('should take the entry placement for a version still on the source',
+    done => {
+        // not localized yet, so the source still describes where the data is:
+        // this is the production-side archive of an already-replicated version
+        sinon.stub(locations, 'isCRRLocation').returns(true);
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        const objmd = new ObjectMD()
+            .setKey(KEY)
+            .setVersionId(VERSION_ID)
+            .setTags({ mytag: 'mytags-value' })
+            .setDataStoreName('cold-location');
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+
+        processEntry(entry, err => {
+            assert.ifError(err);
+
+            const added = mqp.getAdded();
+            assert.strictEqual(added.length, 1);
+            assert.strictEqual(added[0].objVal.dataStoreName, 'cold-location');
+            done();
+        });
+    });
+
+    it('should not skip an update that only changes the object lock', done => {
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        // same tags as the stored object: the ingestion diff would call this a
+        // duplicate and drop the retention with it
+        const objmd = new ObjectMD()
+            .setKey(KEY)
+            .setVersionId(VERSION_ID)
+            .setTags({ mytag: 'mytags-value' })
+            .setRetentionMode('GOVERNANCE')
+            .setRetentionDate('2099-01-01T00:00:00.000Z');
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+
+        processEntry(entry, err => {
+            assert.ifError(err);
+
+            const added = mqp.getAdded();
+            assert.strictEqual(added.length, 1);
+            assert.strictEqual(added[0].objVal.retentionMode, 'GOVERNANCE');
+            assert.strictEqual(added[0].objVal.retentionDate,
+                '2099-01-01T00:00:00.000Z');
+            done();
+        });
+    });
+
+    it('should read the stored object even when the bucket has no ' +
+    'replication configuration', done => {
+        const getObject = sinon.spy(mongoClient, 'getObject');
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        const objmd = new ObjectMD()
+            .setKey(KEY)
+            .setVersionId(VERSION_ID)
+            .setTags({ mytag: 'mytags-value' })
+            .setLegalHold(true);
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+
+        async.waterfall([
+            cb => mongoClient.getBucketAttributes(BUCKET, fakeLogger, cb),
+            (bucketInfo, cb) => {
+                sinon.stub(bucketInfo, 'getReplicationConfiguration')
+                    .returns(null);
+                return mqp._processObjectQueueEntry(fakeLogger, entry, LOCATION,
+                    bucketInfo, cb);
+            },
+        ], err => {
+            assert.ifError(err);
+
+            sinon.assert.called(getObject);
+            // the merge happened, so the entry was recognised as an update
+            assert.strictEqual(mqp.getAdded()[0].objVal.dataStoreName,
+                LOCATION);
+            done();
+        });
+    });
+
+    it('should ignore a scal version id the source object carries', done => {
+        // production may itself have ingested or cold-restored this object, so
+        // it can carry a scal version id of its own; it names a version of the
+        // system production ingested from, not anything here
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        const objmd = new ObjectMD()
+            .setKey(KEY)
+            .setVersionId(VERSION_ID)
+            .setTags({ mytag: 'mytags-value' })
+            .setLegalHold(true)
+            .setUserMetadata({
+                'x-amz-meta-scal-version-id': encode(NEW_VERSION_ID),
+            });
+        const entry = new ObjectQueueEntry(BUCKET, versionKey, objmd);
+        const getObject = sinon.spy(mongoClient, 'getObject');
+
+        processEntry(entry, err => {
+            assert.ifError(err);
+
+            // read, and written back, on the version the entry names
+            assert.strictEqual(getObject.getCall(0).args[2].versionId,
+                VERSION_ID);
+            assert.strictEqual(mqp.getAdded()[0].key,
+                `${KEY}${VID_SEP}${VERSION_ID}`);
+            done();
+        });
+    });
+
+    it('should delete an object whose location differs from the bucket\'s',
+    done => {
+        const versionKey = `${KEY}${VID_SEP}${VERSION_ID}`;
+        const entry = new DeleteOpQueueEntry(BUCKET, versionKey);
+        sinon.stub(mongoClient, 'getObject').callsFake((b, k, p, l, cb) =>
+            cb(null, new ObjectMD()
+                .setKey(KEY)
+                .setVersionId(VERSION_ID)
+                // cold, or still pointing at the source: never the bucket's
+                // own location constraint
+                .setDataStoreName('cold-location')
+                .setLocation(null)
+                ._data));
+
+        async.waterfall([
+            cb => mongoClient.getBucketAttributes(BUCKET, fakeLogger, cb),
+            (bucketInfo, cb) => mqp._processDeleteOpQueueEntry(fakeLogger,
+                entry, LOCATION, bucketInfo, cb),
+        ], err => {
+            assert.ifError(err);
+
+            const deleted = mqp.getDeleted();
+            assert.strictEqual(deleted.length, 1);
+            assert.strictEqual(deleted[0].versionId, VERSION_ID);
+            done();
+        });
+    });
+});
diff --git a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
index 505c70665..a21946e6a 100644
--- a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
+++ b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
@@ -83,6 +83,21 @@ describe('MongoProcessorConfigValidator mode', () => {
         assert.strictEqual(validated.mode, 'ingestion');
     });
 
+    it('should accept the dr mode', () => {
+        const validated = configValidator(globalConfig, { ...baseExtConfig, mode: 'dr' });
+        assert.strictEqual(validated.mode, 'dr');
+    });
+
+    it('should accept the mode from the environment', () => {
+        process.env.EXTENSIONS_MONGO_PROCESSOR_MODE = 'dr';
+        try {
+            const validated = configValidator(globalConfig, baseExtConfig);
+            assert.strictEqual(validated.mode, 'dr');
+        } finally {
+            delete process.env.EXTENSIONS_MONGO_PROCESSOR_MODE;
+        }
+    });
+
     it('should reject a mode with no implementation', () => {
         let err;
         try {

From b1be27479d8dfacb75a00d18c07ef9dac284713b Mon Sep 17 00:00:00 2001
From: Thomas Flament <thomas.flament@scality.com>
Date: Thu, 27 Aug 2026 16:40:09 +0200
Subject: [PATCH 6/6] Read the service account from the extension

Backbeat validates every extension listed under `extensions` against that
extension's own schema, so naming the service account in a partial
`extensions.ingestion` block makes the whole configuration invalid: the
ingestion schema also requires a topic, a zookeeper path and a source list.

A D/R sink runs no extension but this one, so it now carries the account in
its own configuration, falling back to the ingestion extension for the
deployments that enable both.

Issue: BB-811
---
 .../MongoProcessorConfigValidator.js          |  5 +-
 .../mongoProcessor/mongoProcessorTask.js      |  7 +-
 .../MongoProcessorConfigValidator.spec.js     | 80 ++++++++++++++++---
 3 files changed, 77 insertions(+), 15 deletions(-)

diff --git a/extensions/mongoProcessor/MongoProcessorConfigValidator.js b/extensions/mongoProcessor/MongoProcessorConfigValidator.js
index e4be3a26b..e9b56aa11 100644
--- a/extensions/mongoProcessor/MongoProcessorConfigValidator.js
+++ b/extensions/mongoProcessor/MongoProcessorConfigValidator.js
@@ -1,6 +1,6 @@
 const joi = require('joi');
 const { modes, defaultMode } = require('./modes');
-const { retryParamsJoi, probeServerJoi, logJoiOptional, mongoJoi } =
+const { authJoi, retryParamsJoi, probeServerJoi, logJoiOptional, mongoJoi } =
     require('../../lib/config/configItems.joi');
 const { extensionConfigValidator } = require('../../lib/config/extensionConfigValidator');
 
@@ -10,7 +10,8 @@ const joiSchema = joi.object({
     topic: joi.string().required(),
     groupId: joi.string().required(),
     mode: joi.string().valid(...Object.keys(modes)).default(defaultMode),
-    mongodb: mongoJoi,
+    mongodb: mongoJoi.when('mode', { is: 'dr', then: joi.required() }),
+    auth: authJoi.when('mode', { is: 'dr', then: joi.required() }),
     retry: retryParamsJoi,
     concurrency: joi.number().greater(0).default(1),
     maxQueued: joi.number().greater(0).default(MAX_QUEUED_DEFAULT),
diff --git a/extensions/mongoProcessor/mongoProcessorTask.js b/extensions/mongoProcessor/mongoProcessorTask.js
index b29535282..da5d32e03 100644
--- a/extensions/mongoProcessor/mongoProcessorTask.js
+++ b/extensions/mongoProcessor/mongoProcessorTask.js
@@ -64,10 +64,13 @@ async function handleMetrics(res, log) {
 }
 
 function loadManagementDatabase() {
-    const ingestionServiceAuth = config.extensions.ingestion.auth;
+    // the ingestion extension's auth is the fallback for a deployment that
+    // configures it; a D/R sink enables no extension but this one
+    const serviceAuth =
+        mongoProcessorConfig.auth ?? config.extensions.ingestion?.auth;
     initManagement({
         serviceName: 'md-ingestion',
-        serviceAccount: ingestionServiceAuth.account,
+        serviceAccount: serviceAuth.account,
     }, error => {
         if (error) {
             log.error('could not load management db', { error });
diff --git a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
index a21946e6a..f48a80067 100644
--- a/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
+++ b/tests/unit/mongoProcessor/MongoProcessorConfigValidator.spec.js
@@ -9,6 +9,17 @@ const baseExtConfig = {
     probeServer: { port: 4000 },
 };
 
+const serviceAuth = { type: 'service', account: 'service-md-ingestion' };
+
+const mongodb = {
+    replicaSetHosts: 'mongo:27017',
+    database: 'datadb',
+    authCredentials: { username: 'u', password: 'p' },
+};
+
+// what a D/R sink configures: no other extension supplies these
+const drExtConfig = { ...baseExtConfig, mode: 'dr', auth: serviceAuth, mongodb };
+
 // the validated backbeat config, passed to every extension validator
 const globalConfig = { log: { logLevel: 'info', dumpLevel: 'trace' } };
 
@@ -22,7 +33,7 @@ describe('MongoProcessorConfigValidator log override', () => {
     });
 
     it('should leave log undefined when not set, deferring to global config.log', () => {
-        const validated = configValidator(globalConfig, baseExtConfig);
+        const validated = configValidator(globalConfig, { ...baseExtConfig });
         assert.strictEqual(validated.log, undefined);
     });
 
@@ -43,12 +54,6 @@ describe('MongoProcessorConfigValidator log override', () => {
 });
 
 describe('MongoProcessorConfigValidator mongodb', () => {
-    const mongodb = {
-        replicaSetHosts: 'mongo:27017',
-        database: 'datadb',
-        authCredentials: { username: 'u', password: 'p' },
-    };
-
     it('should accept a mongodb client config', () => {
         const validated = configValidator(globalConfig, { ...baseExtConfig, mongodb });
         assert.strictEqual(validated.mongodb.replicaSetHosts, 'mongo:27017');
@@ -59,10 +64,23 @@ describe('MongoProcessorConfigValidator mongodb', () => {
 
     it('should leave mongodb undefined when not set, deferring to the ' +
     'queue populator config', () => {
-        const validated = configValidator(globalConfig, baseExtConfig);
+        const validated = configValidator(globalConfig, { ...baseExtConfig });
         assert.strictEqual(validated.mongodb, undefined);
     });
 
+    it('should require mongodb in dr mode, which runs no queue populator', () => {
+        let err;
+        try {
+            const noMongo = { ...drExtConfig };
+            delete noMongo.mongodb;
+            configValidator(globalConfig, noMongo);
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on dr mode with no mongodb');
+        assert.match(err.message, /mongodb/);
+    });
+
     it('should reject credentials missing a password', () => {
         let err;
         try {
@@ -79,19 +97,20 @@ describe('MongoProcessorConfigValidator mongodb', () => {
 
 describe('MongoProcessorConfigValidator mode', () => {
     it('should default to ingestion so an existing config is unchanged', () => {
-        const validated = configValidator(globalConfig, baseExtConfig);
+        const validated = configValidator(globalConfig, { ...baseExtConfig });
         assert.strictEqual(validated.mode, 'ingestion');
     });
 
     it('should accept the dr mode', () => {
-        const validated = configValidator(globalConfig, { ...baseExtConfig, mode: 'dr' });
+        const validated = configValidator(globalConfig, { ...drExtConfig });
         assert.strictEqual(validated.mode, 'dr');
     });
 
     it('should accept the mode from the environment', () => {
         process.env.EXTENSIONS_MONGO_PROCESSOR_MODE = 'dr';
         try {
-            const validated = configValidator(globalConfig, baseExtConfig);
+            const validated = configValidator(globalConfig,
+                { ...baseExtConfig, auth: serviceAuth, mongodb });
             assert.strictEqual(validated.mode, 'dr');
         } finally {
             delete process.env.EXTENSIONS_MONGO_PROCESSOR_MODE;
@@ -109,3 +128,42 @@ describe('MongoProcessorConfigValidator mode', () => {
         assert.match(err.message, /mode/);
     });
 });
+
+describe('MongoProcessorConfigValidator auth', () => {
+    it('should accept a service account', () => {
+        const validated = configValidator(globalConfig,
+            { ...baseExtConfig, auth: serviceAuth });
+        assert.strictEqual(validated.auth.type, 'service');
+        assert.strictEqual(validated.auth.account, 'service-md-ingestion');
+    });
+
+    it('should leave auth undefined when not set, deferring to the ' +
+    'ingestion extension config', () => {
+        const validated = configValidator(globalConfig, { ...baseExtConfig });
+        assert.strictEqual(validated.auth, undefined);
+    });
+
+    it('should require auth in dr mode, which enables no other extension', () => {
+        let err;
+        try {
+            const noAuth = { ...drExtConfig };
+            delete noAuth.auth;
+            configValidator(globalConfig, noAuth);
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on dr mode with no auth');
+        assert.match(err.message, /auth/);
+    });
+
+    it('should reject a service auth missing the account', () => {
+        let err;
+        try {
+            configValidator(globalConfig, { ...baseExtConfig, auth: { type: 'service' } });
+        } catch (e) {
+            err = e;
+        }
+        assert(err, 'expected configValidator to throw on a service auth with no account');
+        assert.match(err.message, /account/);
+    });
+});