Implement CRR Cascaded feature

Issue: CLDSRV-897

Author: SylvainSenechal

lib/api/apiUtils/object/getReplicationInfo.js

@@ -66,23 +66,33 @@ function _canUserReplicate(authInfo) {
  * @param {string} operationType - The type of operation to replicate
  * @param {object} objectMD - The object metadata
  * @param {AuthInfo} [authInfo] - authentication info of object owner
+ * @param {string[]} [blockedSiteTypes=[]] - location types to exclude from the returned backends
  * @return {object|undefined}
  */
-function getReplicationInfo(s3config, objKey, bucketMD, isMD, objSize, operationType, objectMD, authInfo) {
+function getReplicationInfo(
+    s3config, objKey, bucketMD, isMD, objSize,
+    operationType, objectMD, authInfo, blockedSiteTypes = []) {
     const config = bucketMD.getReplicationConfiguration();
     if (!config || !_canUserReplicate(authInfo)) {
         return undefined;
     }
 
     const isCloud = site => !!replicationBackends[s3config.locationConstraints[site]?.type];
     const rules = _withDefaultStorageClass(config.rules || [], s3config);
-    const backends = ReplicationConfiguration.resolveBackends(
+    const allBackends = ReplicationConfiguration.resolveBackends(
         { ...config, rules },
         objKey,
         isCloud,
         objectMD?.replicationInfo?.backends,
     );
 
+    const backends = blockedSiteTypes.length > 0
+        ? allBackends.filter(b => {
+            const loc = s3config.locationConstraints[b.site];
+            return !loc || !blockedSiteTypes.includes(loc.type);
+        })
+        : allBackends;
+
     if (backends.length === 0) {
         return undefined;
     }

lib/routes/routeBackbeat.js

@@ -1,3 +1,4 @@
+const { constants: { HTTP_STATUS_CONFLICT } } = require('http2');
 const url = require('url');
 const async = require('async');
 const httpProxy = require('http-proxy');
@@ -7,7 +8,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 } = versioning.VersionID;
+const {
+    VersionIdCollisionException,
+    StaleMicroVersionIdException,
+    MicroVersionIdAlreadyStoredException,
+} = require('@scality/cloudserverclient');
 
 const { responseJSONBody } = s3routes.routesUtils;
 const { getSubPartIds } = s3middleware.azureHelper.mpuUtils;
@@ -21,6 +28,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 +40,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 +102,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 +124,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();
@@ -129,6 +138,15 @@ function _respond(response, payload, log, callback) {
     _respondWithHeaders(response, payload, {}, log, callback);
 }
 
+function _respondWithHeaderCrrConflict(response, log, callback, code, message, mvId) {
+    return _respondWithHeaders(
+        response,
+        { code, message },
+        { 'x-scal-micro-version-id': mvId ? encode(mvId) : '' },
+        log, callback, HTTP_STATUS_CONFLICT,
+    );
+}
+
 function _getRequestPayload(req, cb) {
     const payload = [];
     let payloadLen = 0;
@@ -414,6 +432,38 @@ 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'];
+    if (incomingVersionIdEncoded != null && incomingVersionIdEncoded !== '') {
+        const incomingVersionIdDecoded = decode(incomingVersionIdEncoded);
+        if (incomingVersionIdDecoded instanceof Error) {
+            log.error('crr putData: failed to decode x-scal-version-id header', {
+                method: 'putData',
+                error: incomingVersionIdDecoded.message,
+            });
+            return callback(errorInstances.BadRequest.customizeDescription(
+                'bad request: invalid x-scal-version-id header'));
+        }
+        if (objMd && objMd.versionId === incomingVersionIdDecoded) {
+            // Data already at destination for this version; return 409 with the existing
+            // microVersionId so backbeat can decide if putMetadata is still needed.
+            log.debug('crr putData: version already at destination', {
+                method: 'putData',
+                bucketName: request.bucketName,
+                objectKey: request.objectKey,
+                hasMicroVersionId: !!objMd.microVersionId,
+            });
+            request.resume();
+            return _respondWithHeaderCrrConflict(
+                response, log, callback,
+                VersionIdCollisionException.name,
+                'version id already at destination',
+                objMd.microVersionId,
+            );
+        }
+    }
+
+    writeContinue(request, response);
     const context = {
         bucketName: request.bucketName,
         owner: canonicalID,
@@ -539,6 +589,64 @@ 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'];
+    // When the header is present this is a conditional write: accept only if
+    // the incoming microVersionId is newer than the one already stored.
+    // '' is a valid value meaning the source has no microVersionId (null revision).
+    // Note: '' is falsy so the presence check must be !== undefined, not truthy.
+    const hasMicroVersionId = encodedMicroVersionId !== undefined;
+    let incomingMicroVersionId = null;
+    if (hasMicroVersionId && objMd) {
+        // '' means source has no microVersionId, treated as older revision
+        incomingMicroVersionId = encodedMicroVersionId === '' ? null : decode(encodedMicroVersionId);
+        if (incomingMicroVersionId instanceof Error) {
+            log.error('putMetadata: failed to decode x-scal-micro-version-id header', {
+                method: 'putMetadata',
+                error: incomingMicroVersionId.message,
+            });
+            return callback(errorInstances.BadRequest.customizeDescription(
+                'bad request: invalid x-scal-micro-version-id header'));
+        }
+
+        // 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 isIncomingOlderThanCurrent = (incoming, current) =>
+            current !== null &&
+            (incoming === null || incoming > current);
+
+        const objectMicroVersionId = objMd.microVersionId || null;
+        if (incomingMicroVersionId === objectMicroVersionId) {
+            log.debug('putMetadata: incoming microVersionId already stored, skipping write', {
+                method: 'putMetadata',
+                bucketName,
+                objectKey,
+            });
+            request.resume();
+            return _respondWithHeaderCrrConflict(
+                response, log, callback,
+                MicroVersionIdAlreadyStoredException.name,
+                'incoming microVersionId already at destination',
+            );
+        }
+        if (isIncomingOlderThanCurrent(incomingMicroVersionId, objectMicroVersionId)) {
+            log.debug('putMetadata: incoming microVersionId is older than stored, rejecting', {
+                method: 'putMetadata',
+                bucketName,
+                objectKey,
+            });
+            request.resume();
+            return _respondWithHeaderCrrConflict(
+                response, log, callback,
+                StaleMicroVersionIdException.name,
+                'incoming revision is older than destination',
+                objMd?.microVersionId,
+            );
+        }
+    }
+
     return _getRequestPayload(request, (err, payload) => {
         if (err) {
             return callback(err);
@@ -552,15 +660,20 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
             return callback(errors.MalformedPOSTRequest);
         }
 
-        const { headers, bucketName, objectKey } = request;
+        if (incomingMicroVersionId !== null && incomingMicroVersionId !== (omVal.microVersionId ?? null)) {
+            return callback(errors.BadRequest.customizeDescription(
+                'bad request: x-scal-micro-version-id header does not match body microVersionId'));
+        }
+
+        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 === true || omVal.replicationInfo.status === 'REPLICA') &&
             request.serverAccessLog
         ) {
             // eslint-disable-next-line no-param-reassign
@@ -576,7 +689,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 +706,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 +785,8 @@ 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;
+            const isReplica = omVal.replicationInfo.isReplica === true
+                || omVal.replicationInfo.status === 'REPLICA';
             versioning = isReplica;
             omVal.replicationInfo.isNFS = !isReplica;
         }
@@ -724,6 +838,44 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
             options.isNull = isNull;
         }
 
+        const isReplicationWrite = !!headers['x-scal-replication-content'];
+        if (isReplicationWrite) {
+            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 cascadeBlockedLocationTypes = [
+                'location-scality-ring-s3-v1',
+                'location-scality-artesca-s3-v1',
+            ];
+            const nextReplInfo = getReplicationInfo(
+                config, objectKey, bucketInfo, isMDOnly, objSize,
+                null, null, null, cascadeBlockedLocationTypes);
+
+            const hasNextHop = nextReplInfo && nextReplInfo.backends.length > 0;
+
+            // Replicating further requires x-scal-micro-version-id for loop detection.
+            if (hasNextHop && !hasMicroVersionId) {
+                return callback(errors.InternalError.customizeDescription(
+                    'x-scal-micro-version-id is required when replication would trigger a next hop'));
+            }
+
+            omVal.replicationInfo = hasNextHop ? nextReplInfo : {
+                status: '',
+                backends: [],
+                content: [],
+                destination: '',
+                storageClass: '',
+                role: '',
+                storageType: '',
+                dataStoreVersionId: '',
+            };
+            omVal.replicationInfo.isReplica = true;
+        }
+
         return async.series(
             [
                 // Zenko's CRR delegates replacing the account

package.json

@@ -65,11 +65,11 @@
         "vaultclient": "scality/vaultclient#8.5.3",
         "werelogs": "scality/werelogs#semver:^8.2.4",
         "ws": "^8.18.0",
+        "@scality/cloudserverclient": "1.0.9",
         "xml2js": "^0.6.2"
     },
     "devDependencies": {
         "@eslint/compat": "^1.2.2",
-        "@scality/cloudserverclient": "1.0.7",
         "@scality/eslint-config-scality": "scality/Guidelines#8.3.1",
         "eslint": "^9.14.0",
         "eslint-plugin-import": "^2.31.0",