Skip to content

Add CRR Cascaded capabilities #6179

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SylvainSenechal wants to merge 1 commit into
base: development/9.4
Choose a base branch
from
+788 −18
Open

Add CRR Cascaded capabilities #6179

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 12 additions & 2 deletions lib/api/apiUtils/object/getReplicationInfo.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
}
Expand Down
170 changes: 161 additions & 9 deletions lib/routes/routeBackbeat.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
const { constants: { HTTP_STATUS_CONFLICT } } = require('http2');
const url = require('url');
const async = require('async');
const httpProxy = require('http-proxy');
Expand All @@ -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;
Expand All @@ -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');
Expand All @@ -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');
Expand Down Expand Up @@ -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;
Expand All @@ -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();
Expand All @@ -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;
Expand Down Expand Up @@ -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 !== '') {

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we not check incomingVersionIdEncoded against undefined to detect the absence/presence of field ?

@SylvainSenechal SylvainSenechal Jul 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't have a big opinion on this.
VersionID was already always passed from backbeat before, what's sure is right here for the collision check we need it

But I don't really know whether this function should work without a versionID provided.
What I know is In cloudserverClient, version id header field is not mandatory, and in backbeat, the api is called in only one place with the versionId (and I doubt it's used anywhere else)

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) {

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

need to go in this check for incomingVersionIdEncoded === '' as well (i.e. null version object in a versioned bucket)

Suggested change
if (objMd && objMd.versionId === incomingVersionIdDecoded) {
const encodedVersionId = request.headers['x-scal-version-id'];
if (encodedVersionId !== undefined) {
const versionId = encodedVersionId ? decode(encodedVersionId) : '';
if (error) {
...
}
if (objMD && objMd.versionId === versionId) {
...
}

@SylvainSenechal SylvainSenechal Jul 9, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok and actually it's more tricky because I found the null versionID is stored as string "null"
null, "null", undefined, '' 😫

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we need to confirm how objMD is "retrieved", i.e. that it will always return the version we require : for this specific check, we don't care that there is any other version or a maser, but only that this specific version is available...

→ If done, it would allow dropping the parsing of versionID here and comparison with the object. We would only need to check the condition "versionID specified && object found" : if (request.headers['x-scal-version-id']!==undefined && objMD) { //conflict !!!
→ If not done, the conflict detection may not work in some case, for exemple if replication a previous (non-current) version...

// 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,
Expand Down Expand Up @@ -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'];
Comment thread
francoisferrand marked this conversation as resolved.

@francoisferrand francoisferrand Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

when the header is set, maybe we should have a sanity check (later) to verify the metadata contains the same microVersionId as the header

@SylvainSenechal SylvainSenechal Jul 6, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A check that the microVersionId in the header is the same as the microVersionId in the metadata body ? Yes why not, I added it as soon as the request payload is decoded

// 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);
Comment on lines +617 to +618

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

kind of hard to read (and validate that it works), would be best not to see this: i.e. use existing compareVersionId...

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looking at the code of compare versionId in arsenal : it seems that function does not support comparing "null" versionID → should be reworked to support that case

Will need to standardize on what "null" is: null, '' or falsish... Falsish is more flexible (does not require null coalescence after loading document...) but means there may not be an actual equality when the function return 0 (null != undefined)


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,
);
}
}

Comment thread
SylvainSenechal marked this conversation as resolved.
return _getRequestPayload(request, (err, payload) => {
if (err) {
return callback(err);
Expand All @@ -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') &&

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: should we introduce a helper function? or maybe "standardize" omVal.replicationInfo.isReplica after the JSON.parse() call on line 658 ?

        try {
            omVal = JSON.parse(payload);
			omVal.replicationInfo.isReplica ||= omVal.replicationInfo.status === 'REPLICA';
        } catch {
            return callback(errors.MalformedPOSTRequest);
        }

request.serverAccessLog
) {
// eslint-disable-next-line no-param-reassign
Expand All @@ -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
) {
Expand All @@ -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
) {
Expand Down Expand Up @@ -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;
}
Expand Down Expand Up @@ -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',
];
Comment on lines +850 to +853

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should either be in the Config object (if we may need to change) or in constants.js

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: '',
Comment on lines +867 to +874

@francoisferrand francoisferrand Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: do we need to put an empty replicationInfo? can't we keep the field undefined?

};
omVal.replicationInfo.isReplica = true;
}

return async.series(
[
// Zenko's CRR delegates replacing the account
Expand Down
2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Comment thread
SylvainSenechal marked this conversation as resolved.
},
"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",
Expand Down
Loading
Loading