test following code review

SylvainSenechal · SylvainSenechal · commit e214df08f89a · 2026-07-09T16:39:27.000+02:00
diff --git a/extensions/replication/tasks/ReplicateObject.js b/extensions/replication/tasks/ReplicateObject.js
@@ -4,7 +4,12 @@ const { S3Client, GetBucketReplicationCommand, GetObjectCommand } = require('@aw
 const { errors, jsutil, versioning } = require('arsenal');
 const { ObjectMDLocation } = require('arsenal').models;
 const { ReplicationConfiguration } = require('arsenal').models;
-const { encode: encodeMicroVersionId, decode: decodeMicroVersionId } = versioning.VersionID;
+const {
+    encode: encodeMicroVersionId,
+    decode: decodeMicroVersionId,
+    compare: compareMicroVersionId,
+    Ordering,
+} = versioning.VersionID;
 
 const ClientManager = require('../../../lib/clients/ClientManager');
 const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy');
@@ -457,28 +462,31 @@ class ReplicateObject extends BackbeatTask {
         return mapLimitWaitPendingIfError(locations, mpuConcLimit, (part, done) => {
             this._getAndPutPart(sourceEntry, destEntry, part, log, done);
         }, (err, partResults) => {
-            const collisionResult = (partResults || []).find(r => r && r.isCollision);
+            let collisionResult;
+            const nonCollisionParts = [];
+            for (const result of (partResults || [])) {
+                if (!result) {
+                    continue;
+                }
+                if (result.isCollision) {
+                    collisionResult = collisionResult || result;
+                } else {
+                    nonCollisionParts.push(result);
+                }
+            }
             const hasPutDataConflict = collisionResult !== undefined;
             if (err || hasPutDataConflict) {
                 // On error or conflict, drop all parts written
-                return this._deleteOrphans(destEntry,
-                    (partResults || []).filter(r => r && !r.isCollision), log, () => {
-                        if (hasPutDataConflict) {
-                            // Another replicant already stored the object data
-                            // Compare microVersionIds to decide what to do with metadata:
-                            // - source is newer : proceed in metadata-only mode
-                            // - same or older : skip putMetadata
-                            const destMvId = collisionResult.destMvIdRaw;
-                            const srcMvId = sourceEntry.getMicroVersionId() || null;
-                            const isLoop = srcMvId === destMvId;
-                            const isStale = destMvId !== null &&
-                                (srcMvId === null || srcMvId > destMvId);
-                            return cb(null, [], true, isLoop || isStale);
-                        }
-                        return cb(err);
-                    });
+                return this._deleteOrphans(destEntry, nonCollisionParts, log, () => {
+                    if (hasPutDataConflict) {
+                        return cb(null, [], {
+                            microVersionId: collisionResult.destinationMicroVersionId,
+                        });
+                    }
+                    return cb(err);
+                });
             }
-            return cb(null, partResults, false, false);
+            return cb(null, partResults, undefined);
         });
     }
 
@@ -626,22 +634,14 @@ class ReplicateObject extends BackbeatTask {
                         }
 
                         if (err instanceof VersionIdCollisionException) {
-                            const destMvIdRaw = err.microVersionId
-                                ? decodeMicroVersionId(err.microVersionId) : null;
-                            if (destMvIdRaw instanceof Error) {
-                                log.error('failed to decode microVersionId from putData 409',
-                                    {
-                                        method: 'ReplicateObject._getAndPutPartOnce',
-                                        entry: destEntry.getLogInfo(),
-                                        error: destMvIdRaw.message,
-                                    });
-                                return doneOnce(destMvIdRaw);
-                            }
                             log.info('cascade putData: data already at destination', {
                                 method: 'ReplicateObject._getAndPutPartOnce',
                                 entry: destEntry.getLogInfo(),
                             });
-                            return doneOnce(null, { isCollision: true, destMvIdRaw });
+                            return doneOnce(null, {
+                                isCollision: true,
+                                destinationMicroVersionId: err.microVersionId,
+                            });
                         }
                         // eslint-disable-next-line no-param-reassign
                         err.origin = 'target';
@@ -726,10 +726,8 @@ class ReplicateObject extends BackbeatTask {
             .catch(err => {
                 // eslint-disable-next-line no-param-reassign
                 err.origin = 'target';
-                if (err.ObjNotFound || err.name === 'ObjNotFound') {
-                    return cbOnce(err);
-                }
-                if (err instanceof MicroVersionIdAlreadyStoredException ||
+                if (err.ObjNotFound || err.name === 'ObjNotFound' ||
+                    err instanceof MicroVersionIdAlreadyStoredException ||
                     err instanceof StaleMicroVersionIdException) {
                     return cbOnce(err);
                 }
@@ -969,30 +967,27 @@ class ReplicateObject extends BackbeatTask {
                     return this._getAndPutData(sourceEntry, destEntry, log,
                                                next);
                 }
-                return next(null, [], false, false);
+                return next(null, [], undefined);
             },
             // update location, replication status and put metadata in
             // target bucket
-            (destLocations, noNewDataLocations, skipMetadata, next) => {
-                if (skipMetadata) {
+            (destLocations, conflict, next) => {
+                if (this._shouldSkipMetadata(sourceEntry.getMicroVersionId(), conflict, log)) {
                     log.info('skipping putMetadata: destination already has same or newer revision', {
                         entry: destEntry.getLogInfo(),
                     });
                     return next();
                 }
-                // METADATA mode tells cloudserver to preserve the destination's existing
-                // data location instead of overwriting it. It applies when either:
+                // METADATA tells cloudserver to UPDATE the existing metadata document
+                // (preserving the stored location). DATA,METADATA tells cloudserver to
+                // CREATE a new document. Use METADATA when:
                 //   - this is a metadata-only replication entry (mdOnly), or
-                //   - data was already at the destination from another path (noNewDataLocations,
-                //     i.e. cascade putData 409).
-                // Zero-byte objects have no location to preserve, so METADATA mode is
-                // always wrong for them regardless of the above flags.
-                const localMdOnly = (mdOnly || noNewDataLocations) && sourceEntry.getContentLength() > 0;
+                //   - data was already written to the destination from another path (conflict).
+                const localMdOnly = mdOnly || conflict !== undefined;
                 destEntry.setLocation(destLocations);
                 return this._putMetadata(destEntry, localMdOnly, log, err => {
                     if (err) {
-                        return this._deleteOrphans(
-                            destEntry, destLocations, log, () => next(err));
+                        return this._deleteOrphans(destEntry, destLocations, log, () => next(err));
                     }
                     return next();
                 });
@@ -1001,47 +996,55 @@ class ReplicateObject extends BackbeatTask {
             err, sourceEntry, destEntry, kafkaEntry, log, done));
     }
 
+    // Returns true if putMetadata can be skipped because the destination already
+    // holds this revision or a newer one. Returns false when there is no conflict,
+    // when the destination microVersionId is absent or can't be parsed (proceed
+    // conservatively), or when the source holds a newer revision.
+    _shouldSkipMetadata(sourceMicroVersionId, conflict, log) {
+        if (!conflict) {
+            return false;
+        }
+        let destMvId = null;
+        if (conflict.microVersionId) {
+            const decoded = decodeMicroVersionId(conflict.microVersionId);
+            if (decoded instanceof Error) {
+                log.warn('could not decode microVersionId from putData 409, ' +
+                    'proceeding to putMetadata without skip optimisation', {
+                    error: decoded.message,
+                });
+            } else {
+                destMvId = decoded;
+            }
+        }
+        const comparison = compareMicroVersionId(sourceMicroVersionId, destMvId);
+        return destMvId !== null &&
+            (comparison === Ordering.OLDER || comparison === Ordering.EQUAL);
+    }
+
     _processQueueEntryRetryFull(sourceEntry, destEntry, kafkaEntry, log, done) {
         return async.waterfall([
             next => this._getAndPutData(sourceEntry, destEntry, log, next),
-            // update location, replication status and put metadata in
-            // target bucket
-            (destLocations, noNewDataLocations, skipMetadata, next) => {
-                if (skipMetadata) {
+            (destLocations, conflict, next) => {
+                if (this._shouldSkipMetadata(sourceEntry.getMicroVersionId(), conflict, log)) {
                     log.info('skipping putMetadata: destination already has same or newer revision', {
                         entry: destEntry.getLogInfo(),
                     });
                     return next();
                 }
-                // On retry, only keep METADATA mode when data was already at the destination
-                // from another path (cascade 409) and the object is non-zero-byte.
-                // For zero-byte there is no location to preserve; always write metadata directly.
-                const retryMdOnly = noNewDataLocations && sourceEntry.getContentLength() > 0;
+                const localMdOnly = conflict !== undefined;
                 destEntry.setLocation(destLocations);
-                return this._putMetadata(destEntry, retryMdOnly, log, err => {
-                    if (err) {
-                        return next(err);
-                    }
-                    return next();
-                });
+                return this._putMetadata(destEntry, localMdOnly, log, next);
             },
         ], err => this._handleReplicationOutcome(
             err, sourceEntry, destEntry, kafkaEntry, log, done));
     }
 
     _handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry,
         log, done) {
-        if (err instanceof MicroVersionIdAlreadyStoredException) {
-            log.info('replication completed via cascade loop: ' +
-                'object already at destination with the same revision',
-                { entry: sourceEntry.getLogInfo() });
-            this._publishReplicationStatus(sourceEntry, 'COMPLETED', { kafkaEntry, log });
-            return done(null, { committable: false });
-        }
-        if (err instanceof StaleMicroVersionIdException) {
-            log.info('replication completed: destination already holds ' +
-                'this version with a newer revision',
-                { entry: sourceEntry.getLogInfo() });
+        if (err instanceof MicroVersionIdAlreadyStoredException ||
+            err instanceof StaleMicroVersionIdException) {
+            log.info('replication completed: object already at destination',
+                { entry: sourceEntry.getLogInfo(), reason: err.name });
             this._publishReplicationStatus(sourceEntry, 'COMPLETED', { kafkaEntry, log });
             return done(null, { committable: false });
         }
diff --git a/tests/unit/replication/ReplicateObject.spec.js b/tests/unit/replication/ReplicateObject.spec.js
@@ -154,7 +154,7 @@ describe('ReplicateObject', () => {
             task._getAndPutPartOnce(makeSourceEntry(older), makeDestEntry(), part, fakeLogger, (err, result) => {
                 assert.ifError(err);
                 assert.ok(result && result.isCollision, 'should return collision info object');
-                assert.ok('destMvIdRaw' in result, 'collision info should include destMvIdRaw');
+                assert.ok('destinationMicroVersionId' in result, 'collision info should include destinationMicroVersionId');
                 sinon.assert.notCalled(task._publishDataWriteMetrics);
                 done();
             });
@@ -428,9 +428,10 @@ describe('ReplicateObject', () => {
                 getObjectKey: () => 'key',
                 getEncodedVersionId: () => encode(generateVersionId('test', 'RG001')),
                 getMicroVersionId: () => null,
-                getReplicationContent: () => ['METADATA'],
+                getReplicationContent: () => ['DATA', 'METADATA'],
                 getContentLength: () => 0,
                 getLocation: () => [],
+                getReducedLocations: () => [],
                 getLastModified: () => new Date().toISOString(),
                 getIsDeleteMarker: () => false,
                 getReplicationBackend: () => 'site',
@@ -439,6 +440,7 @@ describe('ReplicateObject', () => {
             };
 
             task.metricsHandler = { rpo: () => {} };
+            task.mProducer = { publishMetrics: () => {} };
 
             sinon.stub(task, '_setupRoles').callsFake((e, l, cb) => cb(null, 'srcRole', 'dstRole'));
             sinon.stub(task, '_setTargetAccountMd').callsFake((e, r, l, cb) => cb(null));
@@ -447,7 +449,7 @@ describe('ReplicateObject', () => {
             const putMetadataStub = sinon.stub(task, '_putMetadata')
                 .callsFake((e, mdOnly, l, cb) => {
                     assert.strictEqual(mdOnly, false,
-                        'zero-byte objects must not use METADATA mode even when content is METADATA-only');
+                        'zero-byte objects must use DATA,METADATA (create) mode, not METADATA-only (update) mode');
                     cb(null);
                 });
 
