diff --git a/extensions/replication/tasks/ReplicateObject.js b/extensions/replication/tasks/ReplicateObject.js index 0fb0ebcec..be8343b64 100644 --- a/extensions/replication/tasks/ReplicateObject.js +++ b/extensions/replication/tasks/ReplicateObject.js @@ -1,14 +1,18 @@ 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, @@ -16,6 +20,10 @@ const { GetMetadataCommand, addContentLengthMiddleware, attachReqUids, + attachExpectContinueMiddleware, + VersionIdCollisionException, + StaleMicroVersionIdException, + MicroVersionIdAlreadyStoredException, } = require('@scality/cloudserverclient'); const mapLimitWaitPendingIfError = require('../../../lib/util/mapLimitWaitPendingIfError'); @@ -159,11 +167,11 @@ class ReplicateObject extends BackbeatTask { }, cb); } - _putMetadata(entry, mdOnly, log, cb) { + _putMetadata(entry, mdOnly, conflict, log, cb) { 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 => { @@ -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); }); } @@ -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, @@ -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', @@ -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()), @@ -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(), @@ -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()) : '', }); const writeStartTime = Date.now(); return this.backbeatDest.send(command) @@ -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', @@ -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)); @@ -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(); }); @@ -938,17 +988,37 @@ 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) { + 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 && + (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)); @@ -956,12 +1026,18 @@ class ReplicateObject extends BackbeatTask { _handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry, 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' || @@ -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( diff --git a/package.json b/package.json index 4ee3c304e..25f8fd1ba 100644 --- a/package.json +++ b/package.json @@ -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": { @@ -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", diff --git a/tests/unit/replication/ReplicateObject.spec.js b/tests/unit/replication/ReplicateObject.spec.js index e67aeca9d..03fa02ae1 100644 --- a/tests/unit/replication/ReplicateObject.spec.js +++ b/tests/unit/replication/ReplicateObject.spec.js @@ -1,10 +1,18 @@ const assert = require('assert'); const sinon = require('sinon'); +const { Readable } = require('stream'); const QueueEntry = require('../../../lib/models/QueueEntry'); const ReplicateObject = require('../../../extensions/replication/tasks/ReplicateObject'); const ClientManager = require('../../../lib/clients/ClientManager'); const locations = require('../../../conf/locationConfig.json'); +const { versioning } = require('arsenal'); +const { generateVersionId, encode } = versioning.VersionID; +const { + VersionIdCollisionException, + StaleMicroVersionIdException, + MicroVersionIdAlreadyStoredException, +} = require('@scality/cloudserverclient'); const { replicationEntry } = require('../../utils/kafkaEntries'); const fakeLogger = require('../../utils/fakeLogger'); @@ -72,6 +80,211 @@ describe('ReplicateObject', () => { sinon.restore(); }); + function makeMicroVersionIds() { + const a = generateVersionId('test', 'RG001'); + const b = generateVersionId('test', 'RG001'); + const [older, newer] = a > b ? [a, b] : [b, a]; // In case they are generated at the same millisecond + return { older, newer, olderEncoded: encode(older), newerEncoded: encode(newer) }; + } + + function makeBodyStream() { + const stream = new Readable({ read() {} }); + process.nextTick(() => stream.push(null)); + return stream; + } + + function makeSourceEntry(microVersionId) { + return { + getBucket: () => 'src-bucket', + getObjectKey: () => 'key', + getEncodedVersionId: () => encode(generateVersionId('test', 'RG001')), + getMicroVersionId: () => microVersionId || null, + getOwnerId: () => 'canonical-id', + getLogInfo: () => ({}), + getLocation: () => [{ + key: 'data-key', size: 10, start: 0, + dataStoreName: 'file', dataStoreETag: '1:abc', + }], + getContentLength: () => 10, + }; + } + + function makeDestEntry() { + return { + getBucket: () => 'dest-bucket', + getObjectKey: () => 'key', + getOwnerId: () => 'canonical-id', + getLogInfo: () => ({}), + setAmzServerSideEncryption: () => {}, + setAmzEncryptionCustomerAlgorithm: () => {}, + setAmzEncryptionKeyId: () => {}, + setLocation: () => {}, + }; + } + + describe('putData : cascade VersionIdCollisionException handling', () => { + let part; + + beforeEach(() => { + part = { key: 'data-key', size: 10, start: 0, + dataStoreName: 'file', dataStoreETag: '1:abc' }; + sinon.stub(task, '_publishReadMetrics').returns(); + sinon.stub(task, '_publishDataWriteMetrics').returns(); + }); + + function mockDataTransfer(destinationMicroVersionId) { + task.S3source = { + send: sinon.stub().resolves({ + Body: makeBodyStream(), + ContentLength: 10, + }), + }; + const err = new VersionIdCollisionException({ + message: 'version already at destination', + microVersionId: destinationMicroVersionId, + }); + task.backbeatDest = { + send: sinon.stub().rejects(err), + }; + } + + it('should return collision info on VersionIdCollisionException', done => { + const { older, olderEncoded } = makeMicroVersionIds(); + mockDataTransfer(olderEncoded); + task._getAndPutPartOnce(makeSourceEntry(older), makeDestEntry(), part, fakeLogger, (err, result) => { + assert.ifError(err); + assert.ok(result && result.isCollision, 'should return collision info object'); + assert.ok('microVersionId' in result, + 'collision info should include microVersionId'); + sinon.assert.notCalled(task._publishDataWriteMetrics); + done(); + }); + }); + + it('should set data location and publish metrics when no collision', done => { + task.S3source = { + send: sinon.stub().resolves({ Body: makeBodyStream(), ContentLength: 10 }), + }; + task.backbeatDest = { + send: sinon.stub().resolves({ + Location: [{ key: 'new-key', dataStoreName: 'file' }], + }), + }; + task._getAndPutPartOnce(makeSourceEntry(), makeDestEntry(), part, fakeLogger, (err, result) => { + assert.ifError(err); + assert.deepStrictEqual(result, { + key: 'new-key', + start: 0, + size: 10, + dataStoreName: 'file', + dataStoreETag: '1:abc', + blockId: undefined, + }); + sinon.assert.calledOnce(task._publishDataWriteMetrics); + done(); + }); + }); + }); + + describe('putMetadata : cascade handling', () => { + let entry; + + beforeEach(() => { + entry = QueueEntry.createFromKafkaEntry(replicationEntry); + task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; + }); + + it('should pass through MicroVersionIdAlreadyStoredException and skip metrics', done => { + const metricsStub = sinon.stub(task, '_publishMetadataWriteMetrics').returns(); + const loopErr = new MicroVersionIdAlreadyStoredException({ + message: 'incoming microVersionId already at destination', + }); + task.backbeatDest = { send: sinon.stub().rejects(loopErr) }; + task._putMetadataOnce(entry, false, null, fakeLogger, err => { + assert.ok(err instanceof MicroVersionIdAlreadyStoredException); + sinon.assert.notCalled(metricsStub); + done(); + }); + }); + + it('should pass through StaleMicroVersionIdException', done => { + sinon.stub(task, '_publishMetadataWriteMetrics').returns(); + const staleErr = new StaleMicroVersionIdException({ + message: 'incoming revision is older than destination', + }); + task.backbeatDest = { send: sinon.stub().rejects(staleErr) }; + task._putMetadataOnce(entry, false, null, fakeLogger, err => { + assert.ok(err instanceof StaleMicroVersionIdException); + done(); + }); + }); + + it('should publish metrics and succeed on normal response', done => { + const metricsStub = sinon.stub(task, '_publishMetadataWriteMetrics').returns(); + task.backbeatDest = { send: sinon.stub().resolves({}) }; + task._putMetadataOnce(entry, false, null, fakeLogger, err => { + assert.ifError(err); + sinon.assert.calledOnce(metricsStub); + done(); + }); + }); + }); + + describe('_handleReplicationOutcome : cascade outcomes', () => { + let sourceEntry, destEntry, kafkaEntry; + + beforeEach(() => { + sourceEntry = QueueEntry.createFromKafkaEntry(replicationEntry); + destEntry = makeDestEntry(); + kafkaEntry = {}; + sinon.stub(task, '_publishReplicationStatus').returns(); + + sinon.stub(sourceEntry, 'toCompletedEntry').returns(sourceEntry); + sinon.stub(sourceEntry, 'toFailedEntry').returns(sourceEntry); + sinon.stub(sourceEntry, 'setReplicationSiteDataStoreVersionId').returns(sourceEntry); + sinon.stub(sourceEntry, 'getReplicationSiteDataStoreVersionId').returns('v1'); + }); + + it('should mark COMPLETED for MicroVersionIdAlreadyStoredException', done => { + task._handleReplicationOutcome( + new MicroVersionIdAlreadyStoredException({}), + sourceEntry, destEntry, kafkaEntry, fakeLogger, () => { + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'COMPLETED', sinon.match.any); + done(); + }); + }); + + it('should mark COMPLETED for StaleMicroVersionIdException', done => { + task._handleReplicationOutcome( + new StaleMicroVersionIdException({}), + sourceEntry, destEntry, kafkaEntry, fakeLogger, () => { + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'COMPLETED', sinon.match.any); + done(); + }); + }); + + it('should mark COMPLETED on successful replication', done => { + task._handleReplicationOutcome( + null, sourceEntry, destEntry, kafkaEntry, fakeLogger, () => { + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'COMPLETED', sinon.match.any); + done(); + }); + }); + + it('should mark FAILED for real errors', done => { + const realErr = Object.assign(new Error('network failure'), { origin: 'target' }); + task._handleReplicationOutcome( + realErr, sourceEntry, destEntry, kafkaEntry, fakeLogger, () => { + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'FAILED', sinon.match.any); + done(); + }); + }); + }); + describe('_setTargetAccountMd', () => { it('should skip gettin target account info when auth type is assumeRole', done => { sinon.stub(task, '_setupDestClients').returns(); @@ -107,7 +320,7 @@ describe('ReplicateObject', () => { send: sendStub, }; task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; - task._putMetadataOnce(entry, true, fakeLogger, err => { + task._putMetadataOnce(entry, true, null, fakeLogger, err => { assert.ifError(err); assert(sendStub.calledOnce); assert.deepStrictEqual(sendStub.firstCall.args[0].input.AccountId, '123456789012'); @@ -123,7 +336,7 @@ describe('ReplicateObject', () => { }; task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; sinon.stub(task.destConfig.auth, 'type').value('role'); - task._putMetadataOnce(entry, true, fakeLogger, err => { + task._putMetadataOnce(entry, true, null, fakeLogger, err => { assert.ifError(err); assert(sendStub.calledOnce); assert.strictEqual(sendStub.firstCall.args[0].input.AccountId, undefined); @@ -201,6 +414,53 @@ describe('ReplicateObject', () => { }); }); + describe('_processQueueEntry', () => { + it('should call _putMetadata with mdOnly=false for zero-byte objects', done => { + const destEntry = { + getBucket: () => 'dest-bucket', + getObjectKey: () => 'key', + getLogInfo: () => ({}), + setLocation: () => {}, + getReplicationBackend: () => 'site', + getReplicationSiteStatus: () => 'PENDING', + }; + const sourceEntry = { + getBucket: () => 'src-bucket', + getObjectKey: () => 'key', + getEncodedVersionId: () => encode(generateVersionId('test', 'RG001')), + getMicroVersionId: () => null, + getReplicationContent: () => ['DATA', 'METADATA'], + getContentLength: () => 0, + getLocation: () => [], + getReducedLocations: () => [], + getLastModified: () => new Date().toISOString(), + getIsDeleteMarker: () => false, + getReplicationBackend: () => 'site', + getLogInfo: () => ({}), + toReplicaEntry: () => destEntry, + }; + + task.metricsHandler = { rpo: () => {} }; + task.mProducer = { publishMetrics: () => {} }; + + sinon.stub(task, '_setupRoles').callsFake((e, l, cb) => cb(null, 'srcRole', 'dstRole')); + sinon.stub(task, '_setTargetAccountMd').callsFake((e, r, l, cb) => cb(null)); + sinon.stub(task, '_publishReplicationStatus'); + + const putMetadataStub = sinon.stub(task, '_putMetadata') + .callsFake((e, mdOnly, conflict, l, cb) => { + assert.strictEqual(mdOnly, false, + 'zero-byte objects must use DATA,METADATA (create) mode, not METADATA-only (update) mode'); + cb(null); + }); + + task.processQueueEntry(sourceEntry, {}, () => { + sinon.assert.calledOnce(putMetadataStub); + done(); + }); + }); + }); + describe('_publishReplicationStatus', () => { const ObjectQueueEntry = require('../../../lib/models/ObjectQueueEntry'); @@ -756,6 +1016,94 @@ describe('ReplicateObject', () => { }); }); + describe('_shouldSkipMetadata', () => { + let sourceMvId, olderSourceMvId, encodedNewer, encodedOlder; + + before(() => { + const ids = makeMicroVersionIds(); + olderSourceMvId = ids.older; + sourceMvId = ids.newer; + encodedNewer = ids.newerEncoded; + encodedOlder = ids.olderEncoded; + }); + + it('returns false when conflict is null', () => { + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, null, fakeLogger), false); + }); + + it('returns false when conflict is undefined', () => { + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, undefined, fakeLogger), false); + }); + + it('returns false when conflict has no microVersionId field', () => { + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, { isCollision: true }, fakeLogger), false); + }); + + it('returns false when conflict has null microVersionId', () => { + const conflict = { isCollision: true, microVersionId: null }; + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, conflict, fakeLogger), false); + }); + + it('returns false when conflict has undecodable microVersionId', () => { + const conflict = { isCollision: true, microVersionId: 'tooshort' }; + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, conflict, fakeLogger), false); + }); + + it('returns false when source microVersionId is null (cannot compare)', () => { + const conflict = { isCollision: true, microVersionId: encodedNewer }; + assert.strictEqual(task._shouldSkipMetadata(null, conflict, fakeLogger), false); + }); + + it('returns true when source revision equals destination revision', () => { + const conflict = { isCollision: true, microVersionId: encodedNewer }; + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, conflict, fakeLogger), true); + }); + + it('returns true when source revision is older than destination revision', () => { + const conflict = { isCollision: true, microVersionId: encodedNewer }; + assert.strictEqual(task._shouldSkipMetadata(olderSourceMvId, conflict, fakeLogger), true); + }); + + it('returns false when source revision is newer than destination revision', () => { + const conflict = { isCollision: true, microVersionId: encodedOlder }; + assert.strictEqual(task._shouldSkipMetadata(sourceMvId, conflict, fakeLogger), false); + }); + }); + + describe('_putMetadataOnce with conflict', () => { + it('skips the request when conflict revision is equal to source (already at destination)', done => { + sinon.stub(task, '_publishMetadataWriteMetrics').returns(); + const { newer, newerEncoded } = makeMicroVersionIds(); + const entry = QueueEntry.createFromKafkaEntry(replicationEntry); + sinon.stub(entry, 'getMicroVersionId').returns(newer); + const conflict = { isCollision: true, microVersionId: newerEncoded }; + const sendStub = sinon.stub().resolves({}); + task.backbeatDest = { send: sendStub }; + task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; + task._putMetadataOnce(entry, false, conflict, fakeLogger, err => { + assert.ifError(err); + sinon.assert.notCalled(sendStub); + done(); + }); + }); + + it('proceeds with the request when conflict revision is older than source', done => { + sinon.stub(task, '_publishMetadataWriteMetrics').returns(); + const { olderEncoded, newer } = makeMicroVersionIds(); + const entry = QueueEntry.createFromKafkaEntry(replicationEntry); + sinon.stub(entry, 'getMicroVersionId').returns(newer); + const conflict = { isCollision: true, microVersionId: olderEncoded }; + const sendStub = sinon.stub().resolves({}); + task.backbeatDest = { send: sendStub }; + task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; + task._putMetadataOnce(entry, false, conflict, fakeLogger, err => { + assert.ifError(err); + sinon.assert.calledOnce(sendStub); + done(); + }); + }); + }); + describe('constructor', () => { it('should use retry config of the relevent type', () => { const task = new ReplicateObject({ diff --git a/yarn.lock b/yarn.lock index d78e57c5c..326ae0ef1 100644 --- a/yarn.lock +++ b/yarn.lock @@ -1203,6 +1203,16 @@ "@smithy/types" "^4.12.0" tslib "^2.6.2" +"@aws-sdk/middleware-expect-continue@^3.972.8": + version "3.972.14" + resolved "https://registry.yarnpkg.com/@aws-sdk/middleware-expect-continue/-/middleware-expect-continue-3.972.14.tgz#6bffa547da965dba80ee123ac1f8f1446cc596c2" + integrity sha512-3TNFEVGO4sWZj9TEXOCZLzGEctXHnaO4fk2EQ8KVaboTbwHmEPEQrm17Xb9koImUIXEw0sgi2xtHjg7LuTS3rA== + dependencies: + "@aws-sdk/types" "^3.973.9" + "@smithy/core" "^3.24.5" + "@smithy/types" "^4.14.2" + tslib "^2.6.2" + "@aws-sdk/middleware-expect-continue@^3.972.9": version "3.972.9" resolved "https://registry.yarnpkg.com/@aws-sdk/middleware-expect-continue/-/middleware-expect-continue-3.972.9.tgz#ad62cbc4c5f310a5d104b7fc1150eca13a3c07a4" @@ -1824,6 +1834,14 @@ "@smithy/types" "^4.14.0" tslib "^2.6.2" +"@aws-sdk/types@^3.973.9": + version "3.973.9" + resolved "https://registry.yarnpkg.com/@aws-sdk/types/-/types-3.973.9.tgz#7d1c08cc6e82ec2ac2f2da102a7dd55806592f7f" + integrity sha512-kuBfgQVdcz5Bmapc4A13YbpVw/pXkesfhetcFYwbntqas8sF41OHyd4o28+/TG2ZQdHBsv90Lsu5y6oitvYCdg== + dependencies: + "@smithy/types" "^4.14.2" + tslib "^2.6.2" + "@aws-sdk/util-arn-parser@3.893.0": version "3.893.0" resolved "https://registry.yarnpkg.com/@aws-sdk/util-arn-parser/-/util-arn-parser-3.893.0.tgz#fcc9b792744b9da597662891c2422dda83881d8d" @@ -3215,12 +3233,13 @@ resolved "https://registry.yarnpkg.com/@rtsao/scc/-/scc-1.1.0.tgz#927dd2fae9bc3361403ac2c7a00c32ddce9ad7e8" integrity sha512-zt6OdqaDoOnJ1ZYsCYGt9YmWzDXl4vQdKTyJev62gFhRGKdx7mcT54V9KIjg+d2wi9EXsPvAPKe7i7WjfVWB8g== -"@scality/cloudserverclient@^1.0.8": - version "1.0.8" - resolved "https://registry.yarnpkg.com/@scality/cloudserverclient/-/cloudserverclient-1.0.8.tgz#0b48fa9ae3f280c0f83039029b5527d2a6ba0d57" - integrity sha512-ZjGg91BV6e0pmVVoOu2vo/lA+KVKXX4euHqOpejYvOGY4j/z0lKql8Li2tZ7H5bDitL3jEiXx0p0ZBnXfOdzTw== +"@scality/cloudserverclient@^1.0.9": + version "1.0.9" + resolved "https://registry.yarnpkg.com/@scality/cloudserverclient/-/cloudserverclient-1.0.9.tgz#0a333e39436e1f1e74279c3b5a11645be1fffbc1" + integrity sha512-ZGqH4G535opDAEP2PxdYDFG7kjZ/eBrzP3ZmO49goFVjRhyDCZ1gT0Pask3/wZcyysZ2Au1qylorZSLPiZmB2A== dependencies: "@aws-sdk/client-s3" "^3.1009.0" + "@aws-sdk/middleware-expect-continue" "^3.972.8" JSONStream "^1.3.5" fast-xml-parser "^5.5.7" @@ -3452,6 +3471,15 @@ "@smithy/uuid" "^1.1.2" tslib "^2.6.2" +"@smithy/core@^3.24.5": + version "3.24.5" + resolved "https://registry.yarnpkg.com/@smithy/core/-/core-3.24.5.tgz#396ca5662afc6d83a8f41b7e492e427c48a0924e" + integrity sha512-Kt8phUg45M15EjhYAbZ+fFikYneijLu9Liugz8ZsYz2i8j0hzGv27LWKpEHYRfvj+LyCOSijpcR/2i8RouV+cA== + dependencies: + "@aws-crypto/crc32" "5.2.0" + "@smithy/types" "^4.14.2" + tslib "^2.6.2" + "@smithy/credential-provider-imds@^4.2.13": version "4.2.13" resolved "https://registry.yarnpkg.com/@smithy/credential-provider-imds/-/credential-provider-imds-4.2.13.tgz#c0533f362dec6644f403c7789d8e81233f78c63f" @@ -4415,6 +4443,13 @@ dependencies: tslib "^2.6.2" +"@smithy/types@^4.14.2": + version "4.14.2" + resolved "https://registry.yarnpkg.com/@smithy/types/-/types-4.14.2.tgz#6034ff1e0e52bfb7d744ac371b651a8bf21f30f1" + integrity sha512-P+otAxbV4CqBybp7EkcJCrig63yE2E7PuNVOmilVMRcx/O+QDzGULTrKsq4DV13gSfak9ObPrWaHl/9bL5YcWw== + dependencies: + tslib "^2.6.2" + "@smithy/types@^4.9.0": version "4.9.0" resolved "https://registry.yarnpkg.com/@smithy/types/-/types-4.9.0.tgz#c6636ddfa142e1ddcb6e4cf5f3e1a628d420486f"