Add Crr Cascade capabilities to backbeat crr replication

Issue: BB-767

Author: SylvainSenechal

extensions/replication/tasks/ReplicateObject.js

@@ -1,21 +1,25 @@
 const async = require('async');
 const { S3Client, GetBucketReplicationCommand, GetObjectCommand } = require('@aws-sdk/client-s3');
 
-const errors = require('arsenal').errors;
-const jsutil = require('arsenal').jsutil;
-const ObjectMDLocation = require('arsenal').models.ObjectMDLocation;
-const ReplicationConfiguration = require('arsenal').models.ReplicationConfiguration;
+const { errors, jsutil, versioning } = require('arsenal');
+const { ObjectMDLocation } = require('arsenal').models;
+const { ReplicationConfiguration } = require('arsenal').models;
+const { encode: encodeMicroVersionId, decode } = versioning.VersionID;
 
 const ClientManager = require('../../../lib/clients/ClientManager');
 const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy');
-const { 
+const {
     BackbeatRoutesClient,
     PutDataCommand,
     BatchDeleteCommand,
     PutMetadataCommand,
     GetMetadataCommand,
     addContentLengthMiddleware,
     attachReqUids,
+    attachExpectContinueMiddleware,
+    VersionIdCollisionException,
+    StaleMicroVersionIdException,
+    MicroVersionIdAlreadyStoredException,
 } = require('@scality/cloudserverclient');
 
 const mapLimitWaitPendingIfError = require('../../../lib/util/mapLimitWaitPendingIfError');
@@ -31,6 +35,7 @@ const ObjectQueueEntry = require('../../../lib/models/ObjectQueueEntry');
 const { authTypeAssumeRole } = require('../../../lib/constants');
 
 const errorAlreadyCompleted = {};
+const partAlreadyAtDest = {};
 
 function _extractAccountIdFromRole(role) {
     return role.split(':')[4];
@@ -452,11 +457,18 @@ class ReplicateObject extends BackbeatTask {
         const mpuConcLimit = this.repConfig.queueProcessor.mpuPartsConcurrency;
         return mapLimitWaitPendingIfError(locations, mpuConcLimit, (part, done) => {
             this._getAndPutPart(sourceEntry, destEntry, part, log, done);
-        }, (err, destLocations) => {
+        }, (err, partResults) => {
+            // partAlreadyAtDest signals data already at dest (cascade putData 409);
+            // exclude it so it is not stored as a new location or orphan-deleted.
+            const destLocations = (partResults || [])
+                .filter(result => result && result !== partAlreadyAtDest);
             if (err) {
                 return this._deleteOrphans(destEntry, destLocations, log, () => cb(err));
             }
-            return cb(null, destLocations);
+            // When no new data locations were written, force metadata-only mode (noNewDataLocations)
+            // so cloudserver preserves the destination's existing location field instead of overwriting it
+            const noNewDataLocations = destLocations.length === 0;
+            return cb(null, destLocations, noNewDataLocations);
         });
     }
 
@@ -570,7 +582,9 @@ class ReplicateObject extends BackbeatTask {
                     // destination bucket has to be versioning enabled.
                     VersioningRequired: true,
                     RequestUids: log.getSerializedUids(),
+                    VersionId: sourceEntry.getEncodedVersionId(),
                 });
+                attachExpectContinueMiddleware(putCommand, this.backbeatDest.config?.requestHandler);
                 addContentLengthMiddleware(
                     putCommand,
                     response.ContentLength,
@@ -600,6 +614,12 @@ class ReplicateObject extends BackbeatTask {
                                 incomingMsg.destroy();
                             }
                         }
+
+                        if (err instanceof VersionIdCollisionException) {
+                            const { err: collisionErr, result } = this._resolveVersionIdCollision(
+                                err, sourceEntry, destEntry, log);
+                            return doneOnce(collisionErr, result);
+                        }
                         // eslint-disable-next-line no-param-reassign
                         err.origin = 'target';
                         log.error('an error occurred on putData to S3',
@@ -642,6 +662,43 @@ class ReplicateObject extends BackbeatTask {
             });
     }
 
+    _resolveVersionIdCollision(collisionErr, sourceEntry, destEntry, log) {
+        const logMeta = {
+            method: 'ReplicateObject._resolveVersionIdCollision',
+            entry: destEntry.getLogInfo(),
+        };
+        const decodedDestinationMicroVersionId = collisionErr.microVersionId
+            ? decode(collisionErr.microVersionId) : null;
+        if (decodedDestinationMicroVersionId instanceof Error) {
+            log.error('failed to decode microVersionId from putData 409',
+                { ...logMeta, error: decodedDestinationMicroVersionId.message });
+            return { err: decodedDestinationMicroVersionId };
+        }
+        // null = oldest revision (original putObject, before any metadata update).
+        // VersionID strings are in reverse chronological order: a lexicographically
+        // larger string is an older revision.
+        const isSourceOlderThanDestination = (sourceMicroVersionId, destinationMicroVersionId) =>
+            destinationMicroVersionId !== null &&
+            (sourceMicroVersionId === null || sourceMicroVersionId > destinationMicroVersionId);
+
+        // Treat '' and undefined as null (no microVersionId set).
+        const sourceMicroVersionId = sourceEntry.getMicroVersionId() || null;
+        const destinationMicroVersionId = decodedDestinationMicroVersionId || null;
+        if (sourceMicroVersionId === destinationMicroVersionId) {
+            log.info('cascade loop detected on putData: ' +
+                'destination already has this exact revision, skipping', logMeta);
+            return { err: new MicroVersionIdAlreadyStoredException({}) };
+        }
+        if (isSourceOlderThanDestination(sourceMicroVersionId, destinationMicroVersionId)) {
+            log.info('cascade stale on putData: ' +
+                'destination already has a newer revision', logMeta);
+            return { err: new StaleMicroVersionIdException({}) };
+        }
+        log.info('cascade putData: data already at destination, ' +
+            'proceeding with metadata update', logMeta);
+        return { err: null, result: partAlreadyAtDest };
+    }
+
     _putMetadataOnce(entry, mdOnly, log, cb) {
         log.debug('putting metadata', {
             where: 'target', entry: entry.getLogInfo(),
@@ -671,6 +728,8 @@ class ReplicateObject extends BackbeatTask {
             // destination bucket has to be versioning enabled.
             VersioningRequired: true,
             RequestUids: log.getSerializedUids(),
+            MicroVersionId: entry.getMicroVersionId()
+                ? encodeMicroVersionId(entry.getMicroVersionId()) : '',
         });
         const writeStartTime = Date.now();
         return this.backbeatDest.send(command)
@@ -685,14 +744,14 @@ class ReplicateObject extends BackbeatTask {
                     return cbOnce(err);
                 }
                 log.error('an error occurred when putting metadata to S3',
-                    {
-                        method: 'ReplicateObject._putMetadataOnce',
-                        entry: entry.getLogInfo(),
-                        origin: 'target',
-                        peer: this.destBackbeatHost,
-                        error: err.message,
-                        err,
-                    });
+                {
+                    method: 'ReplicateObject._putMetadataOnce',
+                    entry: entry.getLogInfo(),
+                    origin: 'target',
+                    peer: this.destBackbeatHost,
+                    error: err.message,
+                    err,
+                });
                 return cbOnce(err);
             });
     }
@@ -920,13 +979,14 @@ class ReplicateObject extends BackbeatTask {
                     return this._getAndPutData(sourceEntry, destEntry, log,
                                                next);
                 }
-                return next(null, []);
+                return next(null, [], false);
             },
             // update location, replication status and put metadata in
             // target bucket
-            (destLocations, next) => {
+            (destLocations, noNewDataLocations, next) => {
+                const localMdOnly = mdOnly || noNewDataLocations;
                 destEntry.setLocation(destLocations);
-                this._putMetadata(destEntry, mdOnly, log, err => {
+                this._putMetadata(destEntry, localMdOnly, log, err => {
                     if (err) {
                         return this._deleteOrphans(
                             destEntry, destLocations, log, () => next(err));
@@ -946,22 +1006,35 @@ class ReplicateObject extends BackbeatTask {
             next => this._getAndPutData(sourceEntry, destEntry, log, next),
             // update location, replication status and put metadata in
             // target bucket
-            (location, next) => {
-                destEntry.setLocation(location);
-                this._putMetadata(destEntry, false, log, next);
+            (destLocations, noNewDataLocations, next) => {
+                destEntry.setLocation(destLocations);
+                this._putMetadata(destEntry, noNewDataLocations, 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() });
+            this._publishReplicationStatus(sourceEntry, 'COMPLETED', { kafkaEntry, log });
+            return done(null, { committable: false });
+        }
         if (!err) {
             log.debug('replication succeeded for object, publishing ' +
                 'replication status as COMPLETED',
                 { entry: sourceEntry.getLogInfo() });
-            this._publishReplicationStatus(
-                sourceEntry, 'COMPLETED', { kafkaEntry, log });
+            this._publishReplicationStatus(sourceEntry, 'COMPLETED', { kafkaEntry, log });
             return done(null, { committable: false });
         }
         if (err.BadRole || err.name === 'BadRole' ||
@@ -1020,3 +1093,4 @@ class ReplicateObject extends BackbeatTask {
 }
 
 module.exports = ReplicateObject;
+module.exports._cascadeSignals = { partAlreadyAtDest };

package.json

@@ -56,7 +56,7 @@
     "@aws-sdk/client-s3": "^3.921.0",
     "@aws-sdk/client-sts": "^3.921.0",
     "@aws-sdk/credential-providers": "^3.921.0",
-    "@scality/cloudserverclient": "^1.0.8",
+    "@scality/cloudserverclient": "^1.0.9",
     "@smithy/node-http-handler": "^3.3.3",
     "JSONStream": "^1.3.5",
     "arsenal": "git+https://github.com/scality/arsenal#8.5.2",