Add CRR Cascaded capabilities

Dockerfile

@@ -25,7 +25,7 @@ ENV PYTHON=python3
 RUN npm install -g \
     node-gyp \
     typescript@4.9.5
-COPY package.json yarn.lock /usr/src/app/
+COPY package.json yarn.lock scality-cloudserverclient-v1.0.9.tgz /usr/src/app/
 
 RUN yarn install --production --ignore-optional --frozen-lockfile --ignore-engines --network-concurrency 1
 

lib/api/apiUtils/object/versioning.js

@@ -523,7 +523,6 @@ function restoreMetadata(objMD, metadataStoreParams) {
  *  version id of the null version
  */
 function overwritingVersioning(objMD, metadataStoreParams) {
-    metadataStoreParams.updateMicroVersionId = true;
     metadataStoreParams.amzStorageClass = objMD['x-amz-storage-class'];
 
     // set correct originOp
@@ -566,6 +565,21 @@ function overwritingVersioning(objMD, metadataStoreParams) {
     return options;
 }
 
+/**
+ * @param {Object} objectMD - plain object metadata (not an ObjectMD instance)
+*/
+function prepareMetadataForCascadedCrr(objectMD) {
+    // Bump microVersionId so cascade CRR detects the change as a new revision.
+    // eslint-disable-next-line no-param-reassign
+    objectMD.microVersionId = versionIdUtils.generateVersionId(
+        config.instanceId, config.replicationGroupId);
+    if (objectMD.replicationInfo) {
+        // Clear isReplica, as a user modification is no longer purely a replica
+        // eslint-disable-next-line no-param-reassign
+        objectMD.replicationInfo.isReplica = false;
+    }
+}
+
 module.exports = {
     decodeVersionId,
     getVersionIdResHeader,
@@ -577,4 +591,5 @@ module.exports = {
     preprocessingVersioningDelete,
     overwritingVersioning,
     decodeVID,
+    prepareMetadataForCascadedCrr,
 };

lib/api/objectDeleteTagging.js

@@ -1,7 +1,8 @@
 const async = require('async');
 const { errors } = require('arsenal');
 
-const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions }
+const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions,
+    prepareMetadataForCascadedCrr }
     = require('./apiUtils/object/versioning');
 
 const { standardMetadataValidateBucketAndObj } = require('../metadata/metadataUtils');
@@ -85,6 +86,7 @@ function objectDeleteTagging(authInfo, request, log, callback) {
             }
             // eslint-disable-next-line no-param-reassign
             objectMD.originOp = 's3:ObjectTagging:Delete';
+            prepareMetadataForCascadedCrr(objectMD);
             metadata.putObjectMD(bucket.getName(), objectKey, objectMD, params,
             log, err =>
                 next(err, bucket, objectMD));

lib/api/objectPutLegalHold.js

@@ -2,7 +2,8 @@ const async = require('async');
 const { errors, errorInstances, s3middleware } = require('arsenal');
 
 const collectCorsHeaders = require('../utilities/collectCorsHeaders');
-const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions } =
+const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions,
+    prepareMetadataForCascadedCrr } =
   require('./apiUtils/object/versioning');
 const getReplicationInfo = require('./apiUtils/object/getReplicationInfo');
 const metadata = require('../metadata/wrapper');
@@ -96,6 +97,7 @@ function objectPutLegalHold(authInfo, request, log, callback) {
             }
             // eslint-disable-next-line no-param-reassign
             objectMD.originOp = 's3:ObjectLegalHold:Put';
+            prepareMetadataForCascadedCrr(objectMD);
             metadata.putObjectMD(bucket.getName(), objectKey, objectMD, params,
                 log, err => next(err, bucket, objectMD));
         },

lib/api/objectPutRetention.js

@@ -1,7 +1,8 @@
 const async = require('async');
 const { errors, errorInstances, s3middleware } = require('arsenal');
 
-const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions } =
+const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions,
+    prepareMetadataForCascadedCrr } =
   require('./apiUtils/object/versioning');
 const { ObjectLockInfo, hasGovernanceBypassHeader } =
     require('./apiUtils/object/objectLockHelpers');
@@ -119,6 +120,7 @@ function objectPutRetention(authInfo, request, log, callback) {
                     objectMD.replicationInfo, replicationInfo);
             }
             objectMD.originOp = 's3:ObjectRetention:Put';
+            prepareMetadataForCascadedCrr(objectMD);
             /* eslint-enable no-param-reassign */
             metadata.putObjectMD(bucket.getName(), objectKey, objectMD, params,
                 log, err => next(err, bucket, objectMD));

lib/api/objectPutTagging.js

@@ -1,7 +1,8 @@
 const async = require('async');
 const { errors, s3middleware } = require('arsenal');
 
-const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions } =
+const { decodeVersionId, getVersionIdResHeader, getVersionSpecificMetadataOptions,
+    prepareMetadataForCascadedCrr } =
   require('./apiUtils/object/versioning');
 
 const { standardMetadataValidateBucketAndObj } = require('../metadata/metadataUtils');
@@ -89,6 +90,7 @@ function objectPutTagging(authInfo, request, log, callback) {
             }
             // eslint-disable-next-line no-param-reassign
             objectMD.originOp = 's3:ObjectTagging:Put';
+            prepareMetadataForCascadedCrr(objectMD);
             metadata.putObjectMD(bucket.getName(), objectKey, objectMD, params,
             log, err =>
                 next(err, bucket, objectMD));

lib/metadata/acl.js

@@ -1,6 +1,7 @@
 const { errors } = require('arsenal');
 
 const getReplicationInfo = require('../api/apiUtils/object/getReplicationInfo');
+const { prepareMetadataForCascadedCrr } = require('../api/apiUtils/object/versioning');
 const aclUtils = require('../utilities/aclUtils');
 const constants = require('../../constants');
 const metadata = require('../metadata/wrapper');
@@ -74,7 +75,7 @@ const acl = {
                     backends,
                 };
             }
-
+            prepareMetadataForCascadedCrr(objectMD);
             return metadata.putObjectMD(bucket.getName(), objectKey, objectMD, params, log, cb);
         }
         return cb();

lib/routes/routeBackbeat.js

@@ -7,7 +7,13 @@ const joi = require('@hapi/joi');
 const backbeatProxy = httpProxy.createProxyServer({
     ignorePath: true,
 });
-const { auth, errors, errorInstances, s3middleware, s3routes, models, storage } = require('arsenal');
+const { auth, errors, errorInstances, s3middleware, s3routes, models, storage, versioning } = require('arsenal');
+const { decode, encode, compare: compareMicroVersionId, Ordering } = versioning.VersionID;
+const {
+    VersionIdCollisionException,
+    StaleMicroVersionIdException,
+    CascadeLoopDetectedException,
+} = require('@scality/cloudserverclient');
 
 const { responseJSONBody } = s3routes.routesUtils;
 const { getSubPartIds } = s3middleware.azureHelper.mpuUtils;
@@ -21,6 +27,7 @@ const locationStorageCheck = require('../api/apiUtils/object/locationStorageChec
 const { dataStore } = require('../api/apiUtils/object/storeObject');
 const prepareRequestContexts = require('../api/apiUtils/authorization/prepareRequestContexts');
 const { decodeVersionId } = require('../api/apiUtils/object/versioning');
+const getReplicationInfo = require('../api/apiUtils/object/getReplicationInfo');
 const locationKeysHaveChanged = require('../api/apiUtils/object/locationKeysHaveChanged');
 const { standardMetadataValidateBucketAndObj, metadataGetObject } = require('../metadata/metadataUtils');
 const { config } = require('../Config');
@@ -32,6 +39,7 @@ const {
 } = require('../api/apiUtils/integrity/validateChecksums');
 const { BackendInfo } = models;
 const { pushReplicationMetric } = require('./utilities/pushReplicationMetric');
+const writeContinue = require('../utilities/writeContinue');
 const kms = require('../kms/wrapper');
 const { listLifecycleCurrents } = require('../api/backbeat/listLifecycleCurrents');
 const { listLifecycleNonCurrents } = require('../api/backbeat/listLifecycleNonCurrents');
@@ -93,7 +101,7 @@ function _isObjectRequest(req) {
     return ['data', 'metadata', 'multiplebackenddata', 'multiplebackendmetadata'].includes(req.resourceType);
 }
 
-function _respondWithHeaders(response, payload, extraHeaders, log, callback) {
+function _respondWithHeaders(response, payload, extraHeaders, log, callback, statusCode = 200) {
     let body = '';
     if (typeof payload === 'string') {
         body = payload;
@@ -115,10 +123,10 @@ function _respondWithHeaders(response, payload, extraHeaders, log, callback) {
         // eslint-disable-next-line no-param-reassign
         response.serverAccessLog.endTurnAroundTime = process.hrtime.bigint();
     }
-    response.writeHead(200, httpHeaders);
+    response.writeHead(statusCode, httpHeaders);
     response.end(body, 'utf8', () => {
         log.end().info('responded with payload', {
-            httpCode: 200,
+            httpCode: statusCode,
             contentLength: Buffer.byteLength(body),
         });
         callback();
@@ -414,6 +422,32 @@ function putData(request, response, bucketInfo, objMd, log, callback) {
         log.error(errMessage);
         return callback(errorInstances.BadRequest.customizeDescription(errMessage));
     }
+
+    const incomingVersionIdEncoded = request.headers['x-scal-version-id'];
+    const decoded = incomingVersionIdEncoded ? decode(incomingVersionIdEncoded) : null;
+    const incomingVersionIdDecoded = decoded instanceof Error ? null : decoded;
+    if (incomingVersionIdDecoded && objMd && objMd.versionId === incomingVersionIdDecoded) {
+        // Skip the write if data is already at destination for this version id
+        // Return 409 with the existing microVersionId so backbeat can
+        // decide if putMetadata is still needed
+        log.debug('crr cascade putData: version already at destination', {
+            method: 'putData',
+            bucketName: request.bucketName,
+            objectKey: request.objectKey,
+            hasMicroVersionId: !!objMd.microVersionId,
+        });
+        request.resume();
+        return _respondWithHeaders(
+            response,
+            { code: VersionIdCollisionException.name, message: 'version id already at destination' },
+            { 'x-scal-micro-version-id': objMd.microVersionId ? encode(objMd.microVersionId) : '' },
+            log,
+            callback,
+            409,
+        );
+    }
+
+    writeContinue(request, response);
     const context = {
         bucketName: request.bucketName,
         owner: canonicalID,
@@ -539,6 +573,47 @@ function getCanonicalIdsByAccountId(accountId, log, cb) {
 }
 
 function putMetadata(request, response, bucketInfo, objMd, log, callback) {
+    const { bucketName, objectKey } = request;
+
+    const encodedMicroVersionId = request.headers['x-scal-micro-version-id'];
+    const decoded = encodedMicroVersionId ? decode(encodedMicroVersionId) : null;
+    const incomingMicroVersionId = decoded instanceof Error ? null : decoded;
+    if (incomingMicroVersionId) {
+        const cmp = compareMicroVersionId(incomingMicroVersionId, objMd?.microVersionId);
+        if (cmp === Ordering.EQUAL) {
+            log.debug('crr cascade putMetadata: loop detected, skipping write', {
+                method: 'putMetadata',
+                bucketName,
+                objectKey,
+            });
+            request.resume();
+            return _respondWithHeaders(
+                response,
+                { code: CascadeLoopDetectedException.name, message: 'incoming microVersionId already at destination' },
+                {},
+                log,
+                callback,
+                409,
+            );
+        }
+        if (cmp === Ordering.OLDER) {
+            log.debug('crr cascade putMetadata: stale event, rejecting', {
+                method: 'putMetadata',
+                bucketName,
+                objectKey,
+            });
+            request.resume();
+            return _respondWithHeaders(
+                response,
+                { code: StaleMicroVersionIdException.name, message: 'incoming revision is older than destination' },
+                { 'x-scal-micro-version-id': objMd?.microVersionId ? encode(objMd.microVersionId) : '' },
+                log,
+                callback,
+                409,
+            );
+        }
+    }
+
     return _getRequestPayload(request, (err, payload) => {
         if (err) {
             return callback(err);
@@ -552,15 +627,15 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
             return callback(errors.MalformedPOSTRequest);
         }
 
-        const { headers, bucketName, objectKey } = request;
+        const { headers } = request;
 
         // Destination-side delete-marker replication.
         // We need the REPLICA status to distinguish from
         // source-side replication status updates that also carry isDeleteMarker=true.
         if (
             omVal.isDeleteMarker &&
             omVal.replicationInfo &&
-            omVal.replicationInfo.status === 'REPLICA' &&
+            (omVal.replicationInfo.isReplica || omVal.replicationInfo.status === 'REPLICA') &&
             request.serverAccessLog
         ) {
             // eslint-disable-next-line no-param-reassign
@@ -576,7 +651,7 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
         // The REPLICA status excludes source-side replication-status updates.
         if (
             omVal.replicationInfo &&
-            omVal.replicationInfo.status === 'REPLICA' &&
+            (omVal.replicationInfo.isReplica === true || omVal.replicationInfo.status === 'REPLICA') &&
             (omVal.originOp === 's3:ObjectTagging:Put' || omVal.originOp === 's3:ObjectTagging:Delete') &&
             request.serverAccessLog
         ) {
@@ -593,7 +668,7 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
         // The REPLICA status excludes source-side replication-status updates.
         if (
             omVal.replicationInfo &&
-            omVal.replicationInfo.status === 'REPLICA' &&
+            (omVal.replicationInfo.isReplica === true || omVal.replicationInfo.status === 'REPLICA') &&
             omVal.originOp === 's3:ObjectAcl:Put' &&
             request.serverAccessLog
         ) {
@@ -672,7 +747,7 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
         // then we want to create a version for the replica object even though
         // none was provided in the object metadata value.
         if (omVal.replicationInfo.isNFS) {
-            const isReplica = omVal.replicationInfo.status === 'REPLICA';
+            const isReplica = omVal.replicationInfo.isReplica === true || omVal.replicationInfo.status === 'REPLICA';
             versioning = isReplica;
             omVal.replicationInfo.isNFS = !isReplica;
         }
@@ -724,6 +799,48 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
             options.isNull = isNull;
         }
 
+        // Cascade triggering
+        // If the bucket receiving this replica has its own CRR rules, set
+        // status to PENDING so the queue populator here picks it up for the
+        // next hop. If not, clear the source-side replicationInfo fields
+        // Always mark isReplica=true.
+        if (incomingMicroVersionId) {
+            const isMDOnly = headers['x-scal-replication-content'] === 'METADATA';
+            const objSize = omVal['content-length'] || 0;
+
+            // These S3-compatible Scality locations are excluded
+            // as cascade targets because they use the MultiBackend S3 path which
+            // bypasses the putData/putMetadata routes, so loop detection cannot fire
+            // on those destinations.
+            const BLOCKED_LOCATION_TYPES = ['location-scality-ring-s3-v1', 'location-scality-artesca-s3-v1'];
+
+            const nextReplInfo = getReplicationInfo(config, objectKey, bucketInfo, isMDOnly, objSize, null, null, null);
+
+            if (nextReplInfo) {
+                nextReplInfo.backends = nextReplInfo.backends.filter(b => {
+                    const loc = config.locationConstraints[b.site];
+                    return !loc || !BLOCKED_LOCATION_TYPES.includes(loc.type);
+                });
+            }
+
+            if (nextReplInfo && nextReplInfo.backends.length > 0) {
+                omVal.replicationInfo = nextReplInfo;
+            } else {
+                omVal.replicationInfo = {
+                    status: '',
+                    backends: [],
+                    content: [],
+                    destination: '',
+                    storageClass: '',
+                    role: '',
+                    storageType: '',
+                    dataStoreVersionId: '',
+                };
+            }
+
+            omVal.replicationInfo.isReplica = true;
+        }
+
         return async.series(
             [
                 // Zenko's CRR delegates replacing the account

lib/routes/utilities/pushReplicationMetric.js

@@ -4,7 +4,7 @@ const { pushMetric } = require('../../utapi/utilities');
 
 function getMetricToPush(prevObjectMD, newObjectMD) {
     // We only want to update metrics for a destination bucket.
-    if (newObjectMD.getReplicationStatus() !== 'REPLICA') {
+    if (!newObjectMD.getReplicationIsReplica()) {
         return null;
     }
 

lib/services.js

@@ -109,9 +109,9 @@ const services = {
             lastModifiedDate, versioning, versionId, uploadId,
             tagging, taggingCopy, replicationInfo, defaultRetention,
             dataStoreName, creationTime, retentionMode, retentionDate,
-            legalHold, originOp, updateMicroVersionId, archive, oldReplayId,
-            deleteNullKey, amzStorageClass, overheadField, needOplogUpdate,
-            restoredEtag, bucketOwnerId } = params;
+            legalHold, originOp, archive,
+            oldReplayId, deleteNullKey, amzStorageClass, overheadField,
+            needOplogUpdate, restoredEtag, bucketOwnerId } = params;
         log.trace('storing object in metadata');
         assert.strictEqual(typeof bucketName, 'string');
         const md = new ObjectMD();
@@ -189,10 +189,7 @@ const services = {
             md.setUploadId(uploadId);
             options.replayId = uploadId;
         }
-        // update microVersionId when overwriting metadata.
-        if (updateMicroVersionId) {
-            md.updateMicroVersionId();
-        }
+        md.updateMicroVersionId(config.instanceId, config.replicationGroupId);
         // update restore
         if (archive) {
             md.setAmzStorageClass(amzStorageClass);