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
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 !== '') {
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,
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);

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') &&
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',
];
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
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