Skip to content

Add Crr Cascade capabilities to backbeat crr replication #2747

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

Merged
bert-e merged 2 commits into from
Jul 16, 2026
Merged

Add Crr Cascade capabilities to backbeat crr replication #2747

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8c10723
Add Crr Cascade capabilities to backbeat crr replication
SylvainSenechal Jun 3, 2026
d7f3303
bump to 9.5.0-preview.5
SylvainSenechal Jul 16, 2026
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
143 changes: 109 additions & 34 deletions extensions/replication/tasks/ReplicateObject.js
View file
Open in desktop
Comment thread
SylvainSenechal marked this conversation as resolved.
Original file line number Diff line number Diff line change
@@ -1,21 +1,29 @@
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, ReplicationConfiguration } = require('arsenal').models;
const {
encode: encodeMicroVersionId,
decode: decodeMicroVersionId,
compare: compareMicroVersionId,
Ordering,
} = 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');
Expand Down Expand Up @@ -159,11 +167,11 @@ class ReplicateObject extends BackbeatTask {
}, cb);
}

_putMetadata(entry, mdOnly, log, cb) {
_putMetadata(entry, mdOnly, conflict, log, cb) {
Comment thread
SylvainSenechal marked this conversation as resolved.
this.retry({
actionDesc: 'update metadata on target',
logFields: { entry: entry.getLogInfo() },
actionFunc: done => this._putMetadataOnce(entry, mdOnly,
actionFunc: done => this._putMetadataOnce(entry, mdOnly, conflict,
log, done),
shouldRetryFunc: err => err.retryable,
onRetryFunc: err => {
Expand Down Expand Up @@ -452,11 +460,30 @@ 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) => {
if (err) {
return this._deleteOrphans(destEntry, destLocations, log, () => cb(err));
}, (err, partResults) => {
let collisionResult;
const uploadedParts = [];
for (const result of (partResults || [])) {
if (!result) {
continue;
}
if (result.isCollision) {
collisionResult = collisionResult || result;
} else {
uploadedParts.push(result);
}
}
const hasPutDataConflict = collisionResult !== undefined;
if (err || hasPutDataConflict) {
// On error or conflict, drop all parts written
return this._deleteOrphans(destEntry, uploadedParts, log, () => {
if (hasPutDataConflict) {
return cb(null, [], collisionResult);
}
return cb(err);
});
}
return cb(null, destLocations);
return cb(null, partResults, undefined);
});
}

Expand Down Expand Up @@ -570,7 +597,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,
Expand Down Expand Up @@ -600,6 +629,17 @@ class ReplicateObject extends BackbeatTask {
incomingMsg.destroy();
}
}

if (err instanceof VersionIdCollisionException) {
log.info('cascade putData: data already at destination', {
method: 'ReplicateObject._getAndPutPartOnce',
entry: destEntry.getLogInfo(),
});
return doneOnce(null, {
isCollision: true,
microVersionId: err.microVersionId,
});
}
// eslint-disable-next-line no-param-reassign
err.origin = 'target';
log.error('an error occurred on putData to S3',
Expand Down Expand Up @@ -642,7 +682,13 @@ class ReplicateObject extends BackbeatTask {
});
}

_putMetadataOnce(entry, mdOnly, log, cb) {
_putMetadataOnce(entry, mdOnly, conflict, log, cb) {
if (this._shouldSkipMetadata(entry.getMicroVersionId(), conflict, log)) {
log.info('skipping putMetadata: destination already has same or newer revision', {
entry: entry.getLogInfo(),
});
return cb();
}
log.debug('putting metadata', {
where: 'target', entry: entry.getLogInfo(),
replicationStatus: entry.getReplicationSiteStatus(entry.getReplicationBackend()),
Expand All @@ -657,9 +703,10 @@ class ReplicateObject extends BackbeatTask {
accountId = _extractAccountIdFromRole(this.targetRole);
}

// sends extra header x-scal-replication-content to the target
// if it's a metadata operation only
const replicationContent = (mdOnly ? 'METADATA' : undefined);
// METADATA: update existing document (preserve stored location).
// DATA,METADATA: create a new document.
const localMdOnly = mdOnly || !!conflict;
const replicationContent = (localMdOnly ? 'METADATA' : 'DATA,METADATA');
const mdBlob = entry.getSerialized();
const command = new PutMetadataCommand({
Bucket: entry.getBucket(),
Expand All @@ -671,6 +718,8 @@ class ReplicateObject extends BackbeatTask {
// destination bucket has to be versioning enabled.
VersioningRequired: true,
RequestUids: log.getSerializedUids(),
MicroVersionId: entry.getMicroVersionId()
? encodeMicroVersionId(entry.getMicroVersionId()) : '',
Comment on lines +721 to +722

@benzekrimaha benzekrimaha Jul 7, 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.

the empty-string convention is one of the open topics on scality/cloudserver#6179. Is the shape settled now?

@SylvainSenechal SylvainSenechal Jul 7, 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.

yep this is very intentional, using undefined would cause the sdk to not even send the header and would break the cascade replication. A bit fragile if you ask me but this is the current design

});
const writeStartTime = Date.now();
return this.backbeatDest.send(command)
Expand All @@ -681,7 +730,9 @@ class ReplicateObject extends BackbeatTask {
.catch(err => {
// eslint-disable-next-line no-param-reassign
err.origin = 'target';
if (err.ObjNotFound || err.name === 'ObjNotFound') {
if (err.ObjNotFound || err.name === 'ObjNotFound' ||
err instanceof MicroVersionIdAlreadyStoredException ||
err instanceof StaleMicroVersionIdException) {
return cbOnce(err);
}
log.error('an error occurred when putting metadata to S3',
Comment thread
SylvainSenechal marked this conversation as resolved.
Expand Down Expand Up @@ -868,7 +919,7 @@ class ReplicateObject extends BackbeatTask {
// put metadata in target bucket
next => {
// TODO check that bucket role matches role in metadata
this._putMetadata(destEntry, false, log, next);
this._putMetadata(destEntry, false, null, log, next);
},
], err => this._handleReplicationOutcome(
err, sourceEntry, destEntry, kafkaEntry, log, done));
Expand Down Expand Up @@ -920,16 +971,15 @@ class ReplicateObject extends BackbeatTask {
return this._getAndPutData(sourceEntry, destEntry, log,
next);
}
return next(null, []);
return next(null, [], undefined);
},
// update location, replication status and put metadata in
// target bucket
(destLocations, next) => {
(destLocations, conflict, next) => {
destEntry.setLocation(destLocations);
this._putMetadata(destEntry, mdOnly, log, err => {
return this._putMetadata(destEntry, mdOnly, conflict, log, err => {
if (err) {
return this._deleteOrphans(
destEntry, destLocations, log, () => next(err));
return this._deleteOrphans(destEntry, destLocations, log, () => next(err));
}
return next();
});
Expand All @@ -938,30 +988,56 @@ class ReplicateObject extends BackbeatTask {
err, sourceEntry, destEntry, kafkaEntry, log, done));
}

_processQueueEntryRetryFull(sourceEntry, destEntry, kafkaEntry, log, done) {
log.debug('reprocessing entry as full replication',
{ entry: sourceEntry.getLogInfo() });
// Returns true if putMetadata can be skipped because the destination already
// holds this revision or a newer one. Returns false when there is no conflict,
// when the destination microVersionId is absent or can't be parsed (proceed
// conservatively), or when the source holds a newer revision.
_shouldSkipMetadata(sourceMicroVersionId, conflict, log) {

@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.

Discuss : Could be moved into putMetada so that we don't have to call this from 2 places, and instead inline it.

But imo it's a weird pattern to add an extra param to a function, then call that function and use that extra param at the top to decide to potentially not run this function

@francoisferrand francoisferrand Jul 10, 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.

either are fine with me.

  • while I agree on the param (especially the skip(bool skip) anti-pattern), in this case it seems fine to me : the parameter is not just an indication to skip, but really an input allowing to make the decision ("details of the conflict").
  • in my mind, putting it is putMetadata makes it a further from the anti-pattern : to me it is not skipping the metadata write, but just raising the abstraction level. PutMetadata's goal is to ensure the metadata is in the system, and it would gain the ability to reach the goal without actually making a write (kind of magic!) - similar in a way to how a getter could either read from the actual DB or quickly return the data from a cache.

(but either way may be better to keep _shouldSkipMetadata separate -just decide if it is called from within _putMetadata or before it- so each path can be demonstrated and tested easily)

if (!conflict) {
return false;
}
let destMvId = null;
if (conflict.microVersionId) {
const decoded = decodeMicroVersionId(conflict.microVersionId);
if (decoded instanceof Error) {
log.warn('could not decode microVersionId from putData 409, ' +
'proceeding to putMetadata without skip optimisation', {
error: decoded.message,
});
} else {
destMvId = decoded;
}
}
const comparison = compareMicroVersionId(sourceMicroVersionId, destMvId);
return destMvId !== null &&

@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.

May completely remove the null check if I end up update the compareMicroVersionId function.
Still reviewing cloudserver, haven't done it/decided yet

(comparison === Ordering.OLDER || comparison === Ordering.EQUAL);
}

_processQueueEntryRetryFull(sourceEntry, destEntry, kafkaEntry, log, done) {
return async.waterfall([
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, conflict, next) => {
destEntry.setLocation(destLocations);
return this._putMetadata(destEntry, false, conflict, log, next);
},
], err => this._handleReplicationOutcome(
err, sourceEntry, destEntry, kafkaEntry, log, done));
}

_handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry,

@maeldonn maeldonn Jun 23, 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.

This section has quite a bit of nested conditional logic and duplicate checks that make it hard to read and maintain.

Could we flatten this using guard clauses (early returns) and abstract the err.XYZ || err.name === 'XYZ' checks into a helper function? It would drastically reduce the cognitive load of this function. Let me know if you want to pair on it!

@SylvainSenechal SylvainSenechal Jun 25, 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 agree its trash code and the diff is hard to read with the last else

I just changed it and tried something that i didn't want to do first but i think it's fine : for each condition, directly publish/return, without having to do it at the end of the function. I believe its quite readable this way

@SylvainSenechal SylvainSenechal Jun 25, 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.

Could still use some refactor maybe, although all the erro don't have the same form, I think the diff is reasonnable here

log, done) {
if (err instanceof MicroVersionIdAlreadyStoredException ||
err instanceof StaleMicroVersionIdException) {
log.info('replication completed: metadata revision already at destination',
{ entry: sourceEntry.getLogInfo(), reason: err.name });
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' ||
Expand Down Expand Up @@ -991,8 +1067,7 @@ class ReplicateObject extends BackbeatTask {
{ entry: sourceEntry.getLogInfo() });
return done();
}
log.info('target object version does not exist, retrying ' +
'a full replication',
log.info('replication target object not found, retrying with full data write',
{ entry: sourceEntry.getLogInfo() });
// TODO: Is this the right place to capture retry metrics?
return this._processQueueEntryRetryFull(
Expand Down
4 changes: 2 additions & 2 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "backbeat",
"version": "9.5.0-preview.4",
"version": "9.5.0-preview.5",
"description": "Asynchronous queue and job manager",
"main": "index.js",
"scripts": {
Expand Down Expand Up @@ -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.6",
Expand Down
Loading
Loading