diff --git a/extensions/replication/tasks/MultipleBackendTask.js b/extensions/replication/tasks/MultipleBackendTask.js index 3afa707bcd..144834370c 100644 --- a/extensions/replication/tasks/MultipleBackendTask.js +++ b/extensions/replication/tasks/MultipleBackendTask.js @@ -1,3 +1,4 @@ +const { promisify } = require('util'); const async = require('async'); const { v4: uuid } = require('uuid'); const { GetBucketReplicationCommand } = require('@aws-sdk/client-s3'); @@ -54,7 +55,7 @@ class MultipleBackendTask extends ReplicateObject { return this.destConfig.replicationEndpoint.type; } - _setupRolesOnce(entry, log, cb) { + async _setupRolesOnce(entry, log) { log.debug('getting bucket replication', { entry: entry.getLogInfo() }); const entryRolesString = entry.getReplicationRoles(entry.getReplicationBackend()); let errMessage; @@ -71,7 +72,7 @@ class MultipleBackendTask extends ReplicateObject { entry: entry.getLogInfo(), roles: entryRolesString, }); - return cb(errors.BadRole.customizeDescription(errMessage)); + throw errors.BadRole.customizeDescription(errMessage); } this.sourceRole = entryRoles[0]; @@ -81,89 +82,88 @@ class MultipleBackendTask extends ReplicateObject { Bucket: entry.getBucket(), }); attachReqUids(command, log.getSerializedUids()); - return this.S3source.send(command) - .then(data => { - const replicationEnabled = data.ReplicationConfiguration.Rules - .some(rule => rule.Status === 'Enabled' && - entry.getObjectKey().startsWith( - rule.Filter?.Prefix ?? rule.Prefix ?? '')); - if (!replicationEnabled) { - errMessage = 'replication disabled for object'; - log.debug(errMessage, { - method: 'MultipleBackendTask._setupRolesOnce', - entry: entry.getLogInfo(), - }); - return cb(errors.PreconditionFailed.customizeDescription( - errMessage)); - } - const roles = data.ReplicationConfiguration.Role.split(','); - if (roles.length > 2) { - errMessage = 'expecting no more than two roles in bucket ' + - 'replication configuration when replicating to an ' + - 'external location'; - log.error(errMessage, { - method: 'MultipleBackendTask._setupRolesOnce', - entry: entry.getLogInfo(), - roles, - }); - return cb(errors.BadRole.customizeDescription(errMessage)); - } - if (roles[0] !== entryRoles[0]) { - log.error('role in replication entry for source does not ' + - 'match role in bucket replication configuration', { - method: 'MultipleBackendTask._setupRolesOnce', - entry: entry.getLogInfo(), - entryRole: entryRoles[0], - bucketRole: roles[0], - }); - return cb(errors.BadRole); - } - return cb(); - }) - .catch(err => { - log.error('error getting replication configuration from S3', { - method: 'MultipleBackendTask._setupRolesOnce', - entry: entry.getLogInfo(), - origin: 'source', - peer: this.sourceConfig.s3, - error: err.message, - httpStatus: err.$metadata?.httpStatusCode, - }); - // eslint-disable-next-line no-param-reassign - err.origin = 'source'; - return cb(err); + + let data; + try { + data = await this.S3source.send(command); + } catch (err) { + log.error('error getting replication configuration from S3', { + method: 'MultipleBackendTask._setupRolesOnce', + entry: entry.getLogInfo(), + origin: 'source', + peer: this.sourceConfig.s3, + error: err.message, + httpStatus: err.$metadata?.httpStatusCode, + }); + err.origin = 'source'; + throw err; + } + + const replicationEnabled = data.ReplicationConfiguration.Rules + .some(rule => rule.Status === 'Enabled' && + entry.getObjectKey().startsWith( + rule.Filter?.Prefix ?? rule.Prefix ?? '')); + if (!replicationEnabled) { + errMessage = 'replication disabled for object'; + log.debug(errMessage, { + method: 'MultipleBackendTask._setupRolesOnce', + entry: entry.getLogInfo(), }); + throw errors.PreconditionFailed.customizeDescription(errMessage); + } + const roles = data.ReplicationConfiguration.Role.split(','); + if (roles.length > 2) { + errMessage = 'expecting no more than two roles in bucket ' + + 'replication configuration when replicating to an ' + + 'external location'; + log.error(errMessage, { + method: 'MultipleBackendTask._setupRolesOnce', + entry: entry.getLogInfo(), + roles, + }); + throw errors.BadRole.customizeDescription(errMessage); + } + if (roles[0] !== entryRoles[0]) { + log.error('role in replication entry for source does not ' + + 'match role in bucket replication configuration', { + method: 'MultipleBackendTask._setupRolesOnce', + entry: entry.getLogInfo(), + entryRole: entryRoles[0], + bucketRole: roles[0], + }); + throw errors.BadRole; + } } - _refreshSourceEntry(sourceEntry, log, cb) { + async _refreshSourceEntry(sourceEntry, log) { const params = { bucket: sourceEntry.getBucket(), objectKey: sourceEntry.getObjectKey(), versionId: sourceEntry.getEncodedVersionId() || 'null', }; - return this.backbeatSourceProxy.getMetadata( - params, log, (err, blob) => { - if (err) { - log.error('error getting metadata blob from S3', { - method: 'MultipleBackendTask._refreshSourceEntry', - error: err, - }); - return cb(err); - } - const parsedEntry = ObjectQueueEntry.createFromBlob(blob.Body); - if (parsedEntry.error) { - log.error('error parsing metadata blob', { - error: parsedEntry.error, - method: 'MultipleBackendTask._refreshSourceEntry', - }); - return cb(errors.InternalError. - customizeDescription('error parsing metadata blob')); - } - const refreshedEntry = new ObjectQueueEntry(sourceEntry.getBucket(), - sourceEntry.getObjectVersionedKey(), parsedEntry.result) - .setReplicationBackend(sourceEntry.getReplicationBackend()); - return cb(null, refreshedEntry); - }); + const getMetadata = promisify( + this.backbeatSourceProxy.getMetadata.bind(this.backbeatSourceProxy)); + let blob; + try { + blob = await getMetadata(params, log); + } catch (err) { + log.error('error getting metadata blob from S3', { + method: 'MultipleBackendTask._refreshSourceEntry', + error: err, + }); + throw err; + } + const parsedEntry = ObjectQueueEntry.createFromBlob(blob.Body); + if (parsedEntry.error) { + log.error('error parsing metadata blob', { + error: parsedEntry.error, + method: 'MultipleBackendTask._refreshSourceEntry', + }); + throw errors.InternalError.customizeDescription('error parsing metadata blob'); + } + return new ObjectQueueEntry(sourceEntry.getBucket(), + sourceEntry.getObjectVersionedKey(), parsedEntry.result) + .setReplicationBackend(sourceEntry.getReplicationBackend()); } /** @@ -1186,7 +1186,7 @@ class MultipleBackendTask extends ReplicateObject { _setupClients(entry, log, cb) { // Sets up source clients using the role from the replication // configuration if the authentication type is as such. - return this._setupRoles(entry, log, cb); + return this._setupRoles(entry, log).then(() => cb(), cb); } processQueueEntry(sourceEntry, kafkaEntry, done) { @@ -1196,18 +1196,16 @@ class MultipleBackendTask extends ReplicateObject { return async.waterfall([ next => this._setupClients(sourceEntry, log, next), - next => this._refreshSourceEntry(sourceEntry, log, (err, res) => { - if (err && err.name === 'ObjNotFound' && - sourceEntry.getReplicationIsNFS() && !sourceEntry.getIsDeleteMarker()) { + next => this._refreshSourceEntry(sourceEntry, log) + .then(res => next(null, res), err => { + if (err.name === 'ObjNotFound' && + sourceEntry.getReplicationIsNFS() && !sourceEntry.getIsDeleteMarker()) { // The object was deleted before entry is processed, we // can safely skip this entry. return next(errors.InvalidObjectState); - } - if (err) { + } return next(err); - } - return next(null, res); - }), + }), (refreshedEntry, next) => { const lastModified = new Date(refreshedEntry.getLastModified()); this.metricsHandler.rpo({ @@ -1267,18 +1265,18 @@ class MultipleBackendTask extends ReplicateObject { } return this._getAndPutObject(sourceEntry, log, next); }, - ], err => this._handleReplicationOutcome( - err, sourceEntry, kafkaEntry, log, done)); + ], err => this._handleReplicationOutcome(err, sourceEntry, null, kafkaEntry, log) + .then(result => (result === null ? done() : done(null, result)), done)); } - _handleReplicationOutcome(err, sourceEntry, kafkaEntry, log, done) { + async _handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry, log) { if (!err) { log.debug('replication succeeded for object, publishing ' + 'replication status as COMPLETED', { entry: sourceEntry.getLogInfo() }); this._publishReplicationStatus( sourceEntry, 'COMPLETED', { kafkaEntry, log }); - return done(null, { committable: false }); + return { committable: false }; } if (err.BadRole || err.name === 'BadRole' || (err.origin === 'source' && @@ -1291,18 +1289,18 @@ class MultipleBackendTask extends ReplicateObject { entry: sourceEntry.getLogInfo(), origin: err.origin, error: err.description }); - return done(); + return null; } if (err.ObjNotFound || err.name === 'ObjNotFound') { log.info('replication skipped: ' + 'source object version does not exist', { entry: sourceEntry.getLogInfo() }); - return done(); + return null; } if (err.InvalidObjectState || err.name === 'InvalidObjectState') { log.info('replication skipped: invalid object state', { entry: sourceEntry.getLogInfo() }); - return done(); + return null; } log.debug('replication failed permanently for object, ' + 'publishing replication status as FAILED', @@ -1315,7 +1313,7 @@ class MultipleBackendTask extends ReplicateObject { reason: err.description, kafkaEntry, }); - return done(null, { committable: false }); + return { committable: false }; } } diff --git a/extensions/replication/tasks/ReplicateObject.js b/extensions/replication/tasks/ReplicateObject.js index 0995eafb19..62404e1d6c 100644 --- a/extensions/replication/tasks/ReplicateObject.js +++ b/extensions/replication/tasks/ReplicateObject.js @@ -1,7 +1,7 @@ -const async = require('async'); +const { promisify } = require('util'); const { S3Client, GetBucketReplicationCommand, GetObjectCommand } = require('@aws-sdk/client-s3'); -const { errors, jsutil, versioning } = require('arsenal'); +const { errors, versioning } = require('arsenal'); const { ObjectMDLocation, ReplicationConfiguration } = require('arsenal').models; const { encode: encodeMicroVersionId, @@ -26,7 +26,7 @@ const { MicroVersionIdAlreadyStoredException, } = require('@scality/cloudserverclient'); -const mapLimitWaitPendingIfError = require('../../../lib/util/mapLimitWaitPendingIfError'); +const runTasksWithConcurrency = require('../../../lib/util/runTasksWithConcurrency'); const { isRetryableMiddleware, TIMEOUT_MS } = require('../../../lib/clients/utils'); const { isAccessDeniedError, getAccessDeniedLogFields } = require('../../../lib/util/replicationPermissionError'); const getExtMetrics = require('../utils/getExtMetrics'); @@ -103,24 +103,25 @@ class ReplicateObject extends BackbeatTask { return new RoleCredentials(vaultclient, 'replication', roleArn, log); } - _setupRoles(entry, log, cb) { - this.retry({ + async _setupRoles(entry, log) { + return await this.retry({ actionDesc: 'get bucket replication configuration', logFields: { entry: entry.getLogInfo() }, - actionFunc: done => this._setupRolesOnce(entry, log, done), + actionFunc: done => this._setupRolesOnce(entry, log) + .then(roles => done(null, roles), done), // Rely on AWS SDK notion of retryable error to decide if // we should set the entry replication status to FAILED // (non retryable) or retry later. shouldRetryFunc: err => err.retryable, log, - }, cb); + }); } - _setTargetAccountMd(destEntry, targetRole, log, cb) { + async _setTargetAccountMd(destEntry, targetRole, log) { if (!this.destHosts) { log.warn('cannot process entry: no target site configured', { entry: destEntry.getLogInfo() }); - return cb(errors.InternalError); + throw errors.InternalError; } this._setupDestClients(this.targetRole, log); @@ -128,14 +129,14 @@ class ReplicateObject extends BackbeatTask { // using assumeRole i.e when targeting an Zenko // We delegate this task to the destination's Cloudserver if (this.destConfig.auth.type === authTypeAssumeRole) { - return process.nextTick(cb); + return; } - return this.retry({ + await this.retry({ actionDesc: 'lookup target account attributes', logFields: { entry: destEntry.getLogInfo() }, - actionFunc: done => this._setTargetAccountMdOnce( - destEntry, targetRole, log, done), + actionFunc: done => this._setTargetAccountMdOnce(destEntry, targetRole, log) + .then(() => done(), done), // this call uses our own Vault client which does not set // the 'retryable' field shouldRetryFunc: err => @@ -146,16 +147,16 @@ class ReplicateObject extends BackbeatTask { this._setupDestClients(this.targetRole, log); }, log, - }, cb); + }); } - _getAndPutPart(sourceEntry, destEntry, part, log, cb) { + async _getAndPutPart(sourceEntry, destEntry, part, log) { const partLogger = this.logger.newRequestLogger(log.getUids()); - this.retry({ + return await this.retry({ actionDesc: 'stream part data', logFields: { entry: sourceEntry.getLogInfo(), part }, - actionFunc: done => this._getAndPutPartOnce( - sourceEntry, destEntry, part, partLogger, done), + actionFunc: done => this._getAndPutPartOnce(sourceEntry, destEntry, part, partLogger) + .then(r => done(null, r), done), shouldRetryFunc: err => err.retryable, onRetryFunc: err => { if (err.origin === 'target') { @@ -164,15 +165,15 @@ class ReplicateObject extends BackbeatTask { } }, log: partLogger, - }, cb); + }); } - _putMetadata(entry, mdOnly, conflict, log, cb) { - this.retry({ + async _putMetadata(entry, mdOnly, conflict, log) { + return await this.retry({ actionDesc: 'update metadata on target', logFields: { entry: entry.getLogInfo() }, - actionFunc: done => this._putMetadataOnce(entry, mdOnly, conflict, - log, done), + actionFunc: done => this._putMetadataOnce(entry, mdOnly, conflict, log) + .then(data => done(null, data), done), shouldRetryFunc: err => err.retryable, onRetryFunc: err => { if (err.origin === 'target') { @@ -181,7 +182,7 @@ class ReplicateObject extends BackbeatTask { } }, log, - }, cb); + }); } _getUpdatedSourceEntry(params) { @@ -233,7 +234,7 @@ class ReplicateObject extends BackbeatTask { }); } - _setupRolesOnce(entry, log, cb) { + async _setupRolesOnce(entry, log) { log.debug('getting bucket replication', { entry: entry.getLogInfo() }); const entryRolesString = entry.getReplicationRoles(entry.getReplicationBackend()); @@ -249,7 +250,7 @@ class ReplicateObject extends BackbeatTask { entry: entry.getLogInfo(), roles: entryRolesString, }); - return cb(errors.BadRole); + throw errors.BadRole; } this.sourceRole = entryRoles[0]; this.targetRole = entryRoles[1]; @@ -259,89 +260,13 @@ class ReplicateObject extends BackbeatTask { const command = new GetBucketReplicationCommand( { Bucket: entry.getBucket() }); attachReqUids(command, log.getSerializedUids()); - return this.S3source.send(command) - .then(data => { - const replicationEnabled = ( - data.ReplicationConfiguration.Rules.some( - rule => entry.getObjectKey().startsWith( - rule.Filter?.Prefix ?? rule.Prefix ?? '') - && rule.Status === 'Enabled')); - if (!replicationEnabled) { - log.debug('replication disabled for object', - { - method: 'ReplicateObject._setupRolesOnce', - entry: entry.getLogInfo(), - }); - return cb(errors.PreconditionFailed.customizeDescription( - 'replication disabled for object')); - } - const roles = data.ReplicationConfiguration.Role.split(','); - if (roles.length > 2) { - log.error('expecting one or two roles in bucket ' + - 'replication configuration', - { - method: 'ReplicateObject._setupRolesOnce', - entry: entry.getLogInfo(), - roles, - }); - return cb(errors.BadRole); - } - if (roles[0] !== entryRoles[0]) { - log.error('role in replication entry for source does ' + - 'not match role in bucket replication configuration ', - { - method: 'ReplicateObject._setupRolesOnce', - entry: entry.getLogInfo(), - entryRole: entryRoles[0], - bucketRole: roles[0], - }); - return cb(errors.BadRole); - } - // Pick the rule for this specific backend (multi-destination - // configs may share a StorageClass with distinct Bucket / - // Account), then derive the expected destination role from - // its Account; fall back to literal role[1] for legacy - // configs without Account. - const entryDestination = entry.getDestination(); - const entryRoleAccount = entry.getRole()?.split(':')[4]; - const matchingRule = data.ReplicationConfiguration.Rules.find(rule => { - const prefix = rule.Filter?.Prefix ?? rule.Prefix ?? ''; - return rule.Status === 'Enabled' && - rule.Destination?.StorageClass === this.site && - entry.getObjectKey().startsWith(prefix) && - (!entryDestination || rule.Destination?.Bucket === entryDestination) && - (!entryRoleAccount || !rule.Destination?.Account - || rule.Destination.Account === entryRoleAccount); - }); - let expectedDestRole; - if (matchingRule && matchingRule.Destination.Account) { - expectedDestRole = ReplicationConfiguration - .resolveDestinationRole( - data.ReplicationConfiguration.Role, - matchingRule.Destination.Account); - } else if (roles.length === 2) { - expectedDestRole = roles[1]; - } else { - expectedDestRole = roles[0]; - } - if (expectedDestRole !== entryRoles[1]) { - log.error('role in replication entry for target does ' + - 'not match role in bucket replication configuration ', - { - method: 'ReplicateObject._setupRolesOnce', - entry: entry.getLogInfo(), - entryRole: entryRoles[1], - bucketRole: expectedDestRole, - }); - return cb(errors.BadRole); - } - return cb(null, entryRoles[0], entryRoles[1]); - }) - .catch(err => { - // eslint-disable-next-line no-param-reassign + + let data; + try { + data = await this.S3source.send(command); + } catch (err) { err.origin = 'source'; - log.error('error getting replication ' + - 'configuration from S3', + log.error('error getting replication configuration from S3', { method: 'ReplicateObject._setupRolesOnce', entry: entry.getLogInfo(), @@ -351,140 +276,209 @@ class ReplicateObject extends BackbeatTask { err, httpStatus: err.$metadata?.httpStatusCode, }); - return cb(err); + throw err; + } + + const replicationEnabled = data.ReplicationConfiguration.Rules.some( + rule => entry.getObjectKey().startsWith( + rule.Filter?.Prefix ?? rule.Prefix ?? '') + && rule.Status === 'Enabled'); + if (!replicationEnabled) { + log.debug('replication disabled for object', + { + method: 'ReplicateObject._setupRolesOnce', + entry: entry.getLogInfo(), + }); + throw errors.PreconditionFailed.customizeDescription( + 'replication disabled for object'); + } + const roles = data.ReplicationConfiguration.Role.split(','); + if (roles.length > 2) { + log.error('expecting one or two roles in bucket ' + + 'replication configuration', + { + method: 'ReplicateObject._setupRolesOnce', + entry: entry.getLogInfo(), + roles, + }); + throw errors.BadRole; + } + if (roles[0] !== entryRoles[0]) { + log.error('role in replication entry for source does ' + + 'not match role in bucket replication configuration ', + { + method: 'ReplicateObject._setupRolesOnce', + entry: entry.getLogInfo(), + entryRole: entryRoles[0], + bucketRole: roles[0], + }); + throw errors.BadRole; + } + // Pick the rule for this specific backend (multi-destination + // configs may share a StorageClass with distinct Bucket / + // Account), then derive the expected destination role from + // its Account; fall back to literal role[1] for legacy + // configs without Account. + const entryDestination = entry.getDestination(); + const entryRoleAccount = entry.getRole()?.split(':')[4]; + const matchingRule = data.ReplicationConfiguration.Rules.find(rule => { + const prefix = rule.Filter?.Prefix ?? rule.Prefix ?? ''; + return rule.Status === 'Enabled' && + rule.Destination?.StorageClass === this.site && + entry.getObjectKey().startsWith(prefix) && + (!entryDestination || rule.Destination?.Bucket === entryDestination) && + (!entryRoleAccount || !rule.Destination?.Account + || rule.Destination.Account === entryRoleAccount); }); + let expectedDestRole; + if (matchingRule && matchingRule.Destination.Account) { + expectedDestRole = ReplicationConfiguration + .resolveDestinationRole( + data.ReplicationConfiguration.Role, + matchingRule.Destination.Account); + } else if (roles.length === 2) { + expectedDestRole = roles[1]; + } else { + expectedDestRole = roles[0]; + } + if (expectedDestRole !== entryRoles[1]) { + log.error('role in replication entry for target does ' + + 'not match role in bucket replication configuration ', + { + method: 'ReplicateObject._setupRolesOnce', + entry: entry.getLogInfo(), + entryRole: entryRoles[1], + bucketRole: expectedDestRole, + }); + throw errors.BadRole; + } + return [entryRoles[0], entryRoles[1]]; } - _setTargetAccountMdOnce(destEntry, targetRole, log, cb) { + async _setTargetAccountMdOnce(destEntry, targetRole, log) { log.debug('changing target account owner', { entry: destEntry.getLogInfo() }); const targetAccountId = _extractAccountIdFromRole(targetRole); - this.s3destCredentials.lookupAccountAttributes( - targetAccountId, (err, accountAttr) => { - if (err) { - // eslint-disable-next-line no-param-reassign - err.origin = 'target'; - let peer; - if (this.destConfig.auth.type === 'role') { - peer = this.destBackbeatHost; - if (this.destConfig.auth.vault) { - const { host, port } = this.destConfig.auth.vault; - if (host) { - // no proxy is used, log the vault host/port - peer = { host, port }; - } - } + const lookupAccountAttributes = promisify( + this.s3destCredentials.lookupAccountAttributes.bind(this.s3destCredentials)); + let accountAttr; + try { + accountAttr = await lookupAccountAttributes(targetAccountId); + } catch (err) { + err.origin = 'target'; + let peer; + if (this.destConfig.auth.type === 'role') { + peer = this.destBackbeatHost; + if (this.destConfig.auth.vault) { + const { host, port } = this.destConfig.auth.vault; + if (host) { + peer = { host, port }; } - log.error('an error occurred when looking up target ' + - 'account attributes', - { - method: 'ReplicateObject._setTargetAccountMdOnce', - entry: destEntry.getLogInfo(), - origin: 'target', - peer, - error: err.message, - err, - }); - return cb(err); } - log.debug('setting owner info in target metadata', - { - entry: destEntry.getLogInfo(), - accountAttr, - }); - destEntry.setOwnerId(accountAttr.canonicalID); - destEntry.setOwnerDisplayName(accountAttr.displayName); - return cb(); + } + log.error('an error occurred when looking up target ' + + 'account attributes', + { + method: 'ReplicateObject._setTargetAccountMdOnce', + entry: destEntry.getLogInfo(), + origin: 'target', + peer, + error: err.message, + err, + }); + throw err; + } + log.debug('setting owner info in target metadata', + { + entry: destEntry.getLogInfo(), + accountAttr, }); + destEntry.setOwnerId(accountAttr.canonicalID); + destEntry.setOwnerDisplayName(accountAttr.displayName); } - _refreshSourceEntry(sourceEntry, log, cb) { + async _refreshSourceEntry(sourceEntry, log) { const params = { Bucket: sourceEntry.getBucket(), Key: sourceEntry.getObjectKey(), VersionId: sourceEntry.getEncodedVersionId(), RequestUids: log.getSerializedUids(), }; - return this.backbeatSource.send(new GetMetadataCommand(params)) - .then(data => { - const parsedEntry = ObjectQueueEntry.createFromBlob(data.Body); - if (parsedEntry.error) { - log.error('error parsing metadata blob', { - error: parsedEntry.error, - method: 'ReplicateObject._refreshSourceEntry', - }); - return cb(errors.InternalError. - customizeDescription('error parsing metadata blob')); - } - const refreshedEntry = new ObjectQueueEntry(sourceEntry.getBucket(), - sourceEntry.getObjectVersionedKey(), parsedEntry.result) - .setReplicationBackend(sourceEntry.getReplicationBackend()); - return cb(null, refreshedEntry); - }) - .catch(err => { - err.origin = 'source'; // eslint-disable-line no-param-reassign - const logFields = { - method: 'ReplicateObject._refreshSourceEntry', - error: err, - }; - if (isAccessDeniedError(err)) { - Object.assign(logFields, getAccessDeniedLogFields( - sourceEntry.getBucket(), this.sourceRole)); - } - log.error('error getting metadata blob from S3', logFields); - return cb(err); + let data; + try { + data = await this.backbeatSource.send(new GetMetadataCommand(params)); + } catch (err) { + err.origin = 'source'; + const logFields = { + method: 'ReplicateObject._refreshSourceEntry', + error: err, + }; + if (isAccessDeniedError(err)) { + Object.assign(logFields, getAccessDeniedLogFields( + sourceEntry.getBucket(), this.sourceRole)); + } + log.error('error getting metadata blob from S3', logFields); + throw err; + } + const parsedEntry = ObjectQueueEntry.createFromBlob(data.Body); + if (parsedEntry.error) { + log.error('error parsing metadata blob', { + error: parsedEntry.error, + method: 'ReplicateObject._refreshSourceEntry', }); + throw errors.InternalError.customizeDescription('error parsing metadata blob'); + } + return new ObjectQueueEntry(sourceEntry.getBucket(), + sourceEntry.getObjectVersionedKey(), parsedEntry.result) + .setReplicationBackend(sourceEntry.getReplicationBackend()); } - _getAndPutData(sourceEntry, destEntry, log, cb) { + async _getAndPutData(sourceEntry, destEntry, log) { log.debug('replicating data', { entry: sourceEntry.getLogInfo() }); - if (sourceEntry.getLocation().some(part => { - const partObj = new ObjectMDLocation(part); - return partObj.getDataStoreETag() === undefined; - })) { - const errMessage = - 'cannot replicate object without dataStoreETag property'; + const missingETag = sourceEntry.getLocation().some(part => + new ObjectMDLocation(part).getDataStoreETag() === undefined); + if (missingETag) { + const errMessage = 'cannot replicate object without dataStoreETag property'; log.error(errMessage, { method: 'ReplicateObject._getAndPutData', entry: sourceEntry.getLogInfo(), }); - return cb(errors.InternalError.customizeDescription(errMessage)); + throw errors.InternalError.customizeDescription(errMessage); } // For Replication Replay testing, set the BACKBEAT_INJECT_REPLICATION_ERROR_RATE variable if (BACKBEAT_INJECT_REPLICATION_ERROR_RATE) { if (Math.random() < BACKBEAT_INJECT_REPLICATION_ERROR_RATE) { - return process.nextTick(() => cb(new Error('Replication error'))); + throw new Error('Replication error'); } } const locations = sourceEntry.getReducedLocations(); const mpuConcLimit = this.repConfig.queueProcessor.mpuPartsConcurrency; - return mapLimitWaitPendingIfError(locations, mpuConcLimit, (part, done) => { - this._getAndPutPart(sourceEntry, destEntry, part, log, done); - }, (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 [mapErr, partResults] = await runTasksWithConcurrency( + part => this._getAndPutPart(sourceEntry, destEntry, part, log), + mpuConcLimit, locations); + + let collisionResult; + const uploadedParts = []; + for (const result of (partResults || [])) { + if (!result) { + continue; } - 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); - }); + if (result.isCollision) { + collisionResult = collisionResult || result; + } else { + uploadedParts.push(result); } - return cb(null, partResults, undefined); - }); + } + const hasPutDataConflict = collisionResult !== undefined; + // On error or conflict, drop all parts written + if (mapErr || hasPutDataConflict) { + await this._deleteOrphans(destEntry, uploadedParts, log); + if (hasPutDataConflict) { + return [[], collisionResult]; + } + throw mapErr; + } + return [partResults, undefined]; } _publishReadMetrics(size, readStartTime) { @@ -534,16 +528,15 @@ class ReplicateObject extends BackbeatTask { }); } - _getAndPutPartOnce(sourceEntry, destEntry, part, log, done) { - const doneOnce = jsutil.once(done); + async _getAndPutPartOnce(sourceEntry, destEntry, part, log) { const partObj = new ObjectMDLocation(part); const partNumber = partObj.getPartNumber(); const partSize = partObj.getPartSize(); - + const abortController = new AbortController(); let sourceStreamAborted = false; let destRequestAborted = false; - + const command = new GetObjectCommand({ Bucket: sourceEntry.getBucket(), Key: sourceEntry.getObjectKey(), @@ -552,152 +545,149 @@ class ReplicateObject extends BackbeatTask { }); attachReqUids(command, log.getSerializedUids()); const readStartTime = Date.now(); - - this.S3source.send(command, { abortSignal: abortController.signal }) - .then(response => { - const incomingMsg = response.Body; - incomingMsg.on('error', err => { - if (!sourceStreamAborted && !destRequestAborted) { - abortController.abort(); - destRequestAborted = true; - } - if (err.$metadata?.httpStatusCode === 404) { - return doneOnce(errors.ObjNotFound); - } - if (!sourceStreamAborted) { - // eslint-disable-next-line no-param-reassign - err.origin = 'source'; - // eslint-disable-next-line no-param-reassign - err.retryable = true; - log.error('an error occurred when streaming data from S3', - { - method: 'ReplicateObject._getAndPutPartOnce', - entry: destEntry.getLogInfo(), - part, - origin: 'source', - peer: this.sourceConfig.s3, - error: err.message, - err, - }); - } - return doneOnce(err); - }); - - incomingMsg.on('end', () => { - this._publishReadMetrics(partSize, readStartTime); - }); - - log.debug('putting data', { entry: destEntry.getLogInfo(), part }); - const putCommand = new PutDataCommand({ - Bucket: destEntry.getBucket(), - Key: destEntry.getObjectKey(), - CanonicalID: destEntry.getOwnerId(), - ContentMD5: partObj.getPartETag(), - Body: incomingMsg, - // destination bucket has to be versioning enabled. - VersioningRequired: true, - RequestUids: log.getSerializedUids(), - VersionId: sourceEntry.getEncodedVersionId(), + + let response; + try { + response = await this.S3source.send(command, { abortSignal: abortController.signal }); + } catch (err) { + if (!sourceStreamAborted) { + abortController.abort(); + destRequestAborted = true; + } + err.origin = 'source'; + if (err.$metadata?.httpStatusCode !== 404) { + log.error('an error occurred on getObject from S3', { + method: 'ReplicateObject._getAndPutPartOnce', + entry: sourceEntry.getLogInfo(), + part, + origin: 'source', + peer: this.sourceConfig.s3, + error: err.message, + err, + httpStatus: err.$metadata?.httpStatusCode, }); - addContentLengthMiddleware( - putCommand, - response.ContentLength, - ); - attachExpectContinueMiddleware( - putCommand, - this.backbeatDest.config?.requestHandler, - replicationExpectContinueThreshold, - ); - const writeStartTime = Date.now(); - return this.backbeatDest.send(putCommand, { abortSignal: abortController.signal }) - .then(data => { - partObj.setDataLocation(data.Location[0]); - - // Set encryption parameters that were used to encrypt the - // target data in the object metadata, or reset them if - // there was no encryption - const { ServerSideEncryption, SSECustomerAlgorithm, SSEKMSKeyId } = data; - destEntry.setAmzServerSideEncryption(ServerSideEncryption || ''); - destEntry.setAmzEncryptionCustomerAlgorithm(SSECustomerAlgorithm || ''); - destEntry.setAmzEncryptionKeyId(SSEKMSKeyId || ''); - - this._publishDataWriteMetrics(partSize, sourceEntry, writeStartTime); - return doneOnce(null, partObj.getValue()); - }) - .catch(err => { - if (!destRequestAborted) { - // Abort the source stream - abortController.abort(); - sourceStreamAborted = true; - if (incomingMsg.destroy) { - incomingMsg.destroy(); - } - } + } + throw err; + } - 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', - { - method: 'ReplicateObject._getAndPutPartOnce', - entry: destEntry.getLogInfo(), - part, - origin: 'target', - peer: this.destBackbeatHost, - error: err.message, - httpStatus: err.$metadata?.httpStatusCode, - err, - }); - return doneOnce(err); - }); - }) - .catch(err => { - if (!sourceStreamAborted) { - // Abort controller in case the destination request is still pending + const incomingMsg = response.Body; + const putCommand = new PutDataCommand({ + Bucket: destEntry.getBucket(), + Key: destEntry.getObjectKey(), + CanonicalID: destEntry.getOwnerId(), + ContentMD5: partObj.getPartETag(), + Body: incomingMsg, + // destination bucket has to be versioning enabled. + VersioningRequired: true, + RequestUids: log.getSerializedUids(), + VersionId: sourceEntry.getEncodedVersionId(), + }); + addContentLengthMiddleware(putCommand, response.ContentLength); + attachExpectContinueMiddleware( + putCommand, + this.backbeatDest.config?.requestHandler, + replicationExpectContinueThreshold, + ); + log.debug('putting data', { entry: destEntry.getLogInfo(), part }); + const writeStartTime = Date.now(); + + return await new Promise((resolve, reject) => { + incomingMsg.on('error', err => { + if (!sourceStreamAborted && !destRequestAborted) { abortController.abort(); destRequestAborted = true; } - // eslint-disable-next-line no-param-reassign - err.origin = 'source'; if (err.$metadata?.httpStatusCode === 404) { - return doneOnce(err); + // eslint-disable-next-line no-param-reassign + err.origin = 'source'; + // eslint-disable-next-line no-param-reassign + err.ObjNotFound = true; + // eslint-disable-next-line no-param-reassign + err.name = 'ObjNotFound'; + return reject(err); } - log.error('an error occurred on getObject from S3', - { - method: 'ReplicateObject._getAndPutPartOnce', - entry: sourceEntry.getLogInfo(), - part, - origin: 'source', - peer: this.sourceConfig.s3, - error: err.message, - err, - httpStatus: err.$metadata?.httpStatusCode, - }); - return doneOnce(err); + if (!sourceStreamAborted) { + // eslint-disable-next-line no-param-reassign + err.origin = 'source'; + // eslint-disable-next-line no-param-reassign + err.retryable = true; + log.error('an error occurred when streaming data from S3', + { + method: 'ReplicateObject._getAndPutPartOnce', + entry: destEntry.getLogInfo(), + part, + origin: 'source', + peer: this.sourceConfig.s3, + error: err.message, + err, + }); + } + return reject(err); }); + + incomingMsg.on('end', () => { + this._publishReadMetrics(partSize, readStartTime); + }); + + this.backbeatDest.send(putCommand, { abortSignal: abortController.signal }) + .then(data => { + partObj.setDataLocation(data.Location[0]); + + // Set encryption parameters that were used to encrypt the + // target data in the object metadata, or reset them if + // there was no encryption + const { ServerSideEncryption, SSECustomerAlgorithm, SSEKMSKeyId } = data; + destEntry.setAmzServerSideEncryption(ServerSideEncryption || ''); + destEntry.setAmzEncryptionCustomerAlgorithm(SSECustomerAlgorithm || ''); + destEntry.setAmzEncryptionKeyId(SSEKMSKeyId || ''); + + this._publishDataWriteMetrics(partSize, sourceEntry, writeStartTime); + resolve(partObj.getValue()); + }) + .catch(err => { + if (!destRequestAborted) { + abortController.abort(); + sourceStreamAborted = true; + if (incomingMsg.destroy) { + incomingMsg.destroy(); + } + } + if (err instanceof VersionIdCollisionException) { + log.info('cascade putData: data already at destination', { + method: 'ReplicateObject._getAndPutPartOnce', + entry: destEntry.getLogInfo(), + }); + return resolve({ isCollision: true, microVersionId: err.microVersionId }); + } + // eslint-disable-next-line no-param-reassign + err.origin = 'target'; + log.error('an error occurred on putData to S3', + { + method: 'ReplicateObject._getAndPutPartOnce', + entry: destEntry.getLogInfo(), + part, + origin: 'target', + peer: this.destBackbeatHost, + error: err.message, + httpStatus: err.$metadata?.httpStatusCode, + err, + }); + return reject(err); + }); + }); } - _putMetadataOnce(entry, mdOnly, conflict, log, cb) { + async _putMetadataOnce(entry, mdOnly, conflict, log) { if (this._shouldSkipMetadata(entry.getMicroVersionId(), conflict, log)) { log.info('skipping putMetadata: destination already has same or newer revision', { entry: entry.getLogInfo(), }); - return cb(); + return; } log.debug('putting metadata', { where: 'target', entry: entry.getLogInfo(), replicationStatus: entry.getReplicationSiteStatus(entry.getReplicationBackend()), }); - const cbOnce = jsutil.once(cb); // accountid is only needed when using assumeRole auth // to delegate the task of updating metadata with @@ -726,38 +716,35 @@ class ReplicateObject extends BackbeatTask { ? encodeMicroVersionId(entry.getMicroVersionId()) : '', }); const writeStartTime = Date.now(); - return this.backbeatDest.send(command) - .then(data => { - this._publishMetadataWriteMetrics(mdBlob, writeStartTime); - return cbOnce(null, data); - }) - .catch(err => { - // eslint-disable-next-line no-param-reassign - err.origin = 'target'; - 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', - { - method: 'ReplicateObject._putMetadataOnce', - entry: entry.getLogInfo(), - origin: 'target', - peer: this.destBackbeatHost, - error: err.message, - err, - }); - return cbOnce(err); - }); + try { + await this.backbeatDest.send(command); + } catch (err) { + err.origin = 'target'; + if (err.ObjNotFound || err.name === 'ObjNotFound' || + err instanceof MicroVersionIdAlreadyStoredException || + err instanceof StaleMicroVersionIdException) { + throw err; + } + log.error('an error occurred when putting metadata to S3', + { + method: 'ReplicateObject._putMetadataOnce', + entry: entry.getLogInfo(), + origin: 'target', + peer: this.destBackbeatHost, + error: err.message, + err, + }); + throw err; + } + this._publishMetadataWriteMetrics(mdBlob, writeStartTime); } - _deleteOrphans(entry, locations, log, cb) { + async _deleteOrphans(entry, locations, log) { const writtenLocations = locations .filter(loc => loc) .map(loc => ({ key: loc.key, dataStoreName: loc.dataStoreName })); if (writtenLocations.length === 0) { - return process.nextTick(cb); + return; } log.info('deleting orphan data after replication failure', { @@ -771,30 +758,28 @@ class ReplicateObject extends BackbeatTask { Locations: writtenLocations, RequestUids: log.getSerializedUids(), }); - - return this.backbeatDest.send(command) - .then(() => cb()) - .catch(err => { - log.error('an error occurred during batch delete of orphan data', - { - method: 'ReplicateObject._deleteOrphans', - entry: entry.getLogInfo(), - origin: 'target', - peer: this.destBackbeatHost, - error: err.message, - httpStatus: err.$metadata?.httpStatusCode, - err, - }); - writtenLocations.forEach(location => { - log.error('orphan data location was not deleted', { - method: 'ReplicateObject._deleteOrphans', - entry: entry.getLogInfo(), - location, - }); + try { + await this.backbeatDest.send(command); + } catch (err) { + log.error('an error occurred during batch delete of orphan data', + { + method: 'ReplicateObject._deleteOrphans', + entry: entry.getLogInfo(), + origin: 'target', + peer: this.destBackbeatHost, + error: err.message, + httpStatus: err.$metadata?.httpStatusCode, + err, + }); + writtenLocations.forEach(location => { + log.error('orphan data location was not deleted', { + method: 'ReplicateObject._deleteOrphans', + entry: entry.getLogInfo(), + location, }); - // do not return the batch delete error, only log it - return cb(); }); + // do not propagate the batch delete error, only log it + } } _setupSourceClients(sourceRole, log) { @@ -897,7 +882,7 @@ class ReplicateObject extends BackbeatTask { }); } - processQueueEntry(_sourceEntry, kafkaEntry, done) { + async _processQueueEntry(_sourceEntry, kafkaEntry) { let sourceEntry = _sourceEntry; const log = this.logger.newRequestLogger(); const destEntry = sourceEntry.toReplicaEntry(sourceEntry.getReplicationBackend()); @@ -911,85 +896,63 @@ class ReplicateObject extends BackbeatTask { location: this.site, }, (Date.now() - lastModified) / 1000); - if (sourceEntry.getIsDeleteMarker()) { - return async.waterfall([ - next => { - this._setupRoles(sourceEntry, log, next); - }, - (sourceRole, targetRole, next) => { - this._setTargetAccountMd(destEntry, targetRole, log, - next); - }, - // put metadata in target bucket - next => { - // TODO check that bucket role matches role in metadata - this._putMetadata(destEntry, false, null, log, next); - }, - ], err => this._handleReplicationOutcome( - err, sourceEntry, destEntry, kafkaEntry, log, done)); - } - + const isDeleteMarker = sourceEntry.getIsDeleteMarker(); const mdOnly = !sourceEntry.getReplicationContent().includes('DATA'); - return async.waterfall([ - // get data stream from source bucket - next => { - this._setupRoles(sourceEntry, log, next); - }, - (sourceRole, targetRole, next) => { - this._setTargetAccountMd(destEntry, targetRole, log, next); - }, - next => { - if (mdOnly) { - return next(); - } - const isLargeObject = sourceEntry.getContentLength() / 1000000 >= - this.repConfig.queueProcessor.sourceCheckIfSizeGreaterThanMB; - const isLocationStripped = sourceEntry.getContentLength() > 0 && sourceEntry.getLocation().length === 0; - if (!isLargeObject && !isLocationStripped) { - return next(); - } - return this._refreshSourceEntry(sourceEntry, log, (err, refreshedEntry) => { - if (err) { - return next(err); - } - const status = refreshedEntry.getReplicationSiteStatus( - sourceEntry.getReplicationBackend()); - if (status === 'COMPLETED') { - log.info('replication already completed, skipping', { - entry: sourceEntry.getLogInfo(), - }); - return next(errorAlreadyCompleted); - } - // Reassign sourceEntry to use fresh metadata - sourceEntry = refreshedEntry; - return next(); - }); - }, - // Get data from source bucket and put it on the target bucket - next => { + + try { + const [, targetRole] = await this._setupRoles(sourceEntry, log); + await this._setTargetAccountMd(destEntry, targetRole, log); + + if (isDeleteMarker) { + // TODO check that bucket role matches role in metadata + await this._putMetadata(destEntry, false, null, log); + } else { if (!mdOnly) { + const isLargeObject = sourceEntry.getContentLength() / 1000000 >= + this.repConfig.queueProcessor.sourceCheckIfSizeGreaterThanMB; + const isLocationStripped = sourceEntry.getContentLength() > 0 && + sourceEntry.getLocation().length === 0; + if (isLargeObject || isLocationStripped) { + const refreshedEntry = await this._refreshSourceEntry(sourceEntry, log); + const status = refreshedEntry.getReplicationSiteStatus( + sourceEntry.getReplicationBackend()); + if (status === 'COMPLETED') { + log.info('replication already completed, skipping', { + entry: sourceEntry.getLogInfo(), + }); + throw errorAlreadyCompleted; + } + sourceEntry = refreshedEntry; + } const extMetrics = getExtMetrics(this.site, sourceEntry.getContentLength(), sourceEntry); this.mProducer.publishMetrics(extMetrics, metricsTypeQueued, metricsExtension, () => {}); - return this._getAndPutData(sourceEntry, destEntry, log, - next); } - return next(null, [], undefined); - }, - // update location, replication status and put metadata in - // target bucket - (destLocations, conflict, next) => { + + const [destLocations, conflict] = !mdOnly + ? await this._getAndPutData(sourceEntry, destEntry, log) + : [[], undefined]; + destEntry.setLocation(destLocations); - return this._putMetadata(destEntry, mdOnly, conflict, log, err => { - if (err) { - return this._deleteOrphans(destEntry, destLocations, log, () => next(err)); - } - return next(); - }); - }, - ], err => this._handleReplicationOutcome( - err, sourceEntry, destEntry, kafkaEntry, log, done)); + try { + await this._putMetadata(destEntry, mdOnly, conflict, log); + } catch (err) { + await this._deleteOrphans(destEntry, destLocations, log); + throw err; + } + } + } catch (err) { + return this._handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry, log); + } + return this._handleReplicationOutcome(null, sourceEntry, destEntry, kafkaEntry, log); + } + + processQueueEntry(_sourceEntry, kafkaEntry, done) { + this._processQueueEntry(_sourceEntry, kafkaEntry).then( + result => result === null ? done() : done(null, result), + err => done(err), + ); } // Returns true if putMetadata can be skipped because the destination already @@ -1017,42 +980,41 @@ class ReplicateObject extends BackbeatTask { (comparison === Ordering.OLDER || comparison === Ordering.EQUAL); } - _processQueueEntryRetryFull(sourceEntry, destEntry, kafkaEntry, log, done) { - return async.waterfall([ - next => this._getAndPutData(sourceEntry, destEntry, log, next), - (destLocations, conflict, next) => { - destEntry.setLocation(destLocations); - return this._putMetadata(destEntry, false, conflict, log, err => { - if (err) { - log.warn('putMetadata failed during full retry, cleaning up orphan data', { - method: 'ReplicateObject._processQueueEntryRetryFull', - entry: destEntry.getLogInfo(), - error: err.message, - }); - return this._deleteOrphans(destEntry, destLocations, log, () => next(err)); - } - return next(); + async _processQueueEntryRetryFull(sourceEntry, destEntry, kafkaEntry, log) { + let destLocations = null; + try { + let conflict; + [destLocations, conflict] = await this._getAndPutData(sourceEntry, destEntry, log); + destEntry.setLocation(destLocations); + await this._putMetadata(destEntry, false, conflict, log); + } catch (err) { + if (destLocations !== null) { + log.warn('putMetadata failed during full retry, cleaning up orphan data', { + method: 'ReplicateObject._processQueueEntryRetryFull', + entry: destEntry.getLogInfo(), + error: err.message, }); - }, - ], err => this._handleReplicationOutcome( - err, sourceEntry, destEntry, kafkaEntry, log, done)); + await this._deleteOrphans(destEntry, destLocations, log); + } + return this._handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry, log); + } + return this._handleReplicationOutcome(null, sourceEntry, destEntry, kafkaEntry, log); } - _handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry, - log, done) { + async _handleReplicationOutcome(err, sourceEntry, destEntry, kafkaEntry, log) { 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 }); + return { committable: false }; } if (!err) { log.debug('replication succeeded for object, publishing ' + 'replication status as COMPLETED', { entry: sourceEntry.getLogInfo() }); this._publishReplicationStatus(sourceEntry, 'COMPLETED', { kafkaEntry, log }); - return done(null, { committable: false }); + return { committable: false }; } if (err.BadRole || err.name === 'BadRole' || (err.origin === 'source' && @@ -1066,31 +1028,31 @@ class ReplicateObject extends BackbeatTask { origin: err.origin, error: err.description, }); - return done(); + return null; } if (err === errorAlreadyCompleted) { log.warn('replication skipped: ' + 'source object version already COMPLETED', { entry: sourceEntry.getLogInfo() }); - return done(); + return null; } if (err.ObjNotFound || err.name === 'ObjNotFound') { if (err.origin === 'source') { log.info('replication skipped: ' + 'source object version does not exist', { entry: sourceEntry.getLogInfo() }); - return done(); + return null; } 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( - sourceEntry, destEntry, kafkaEntry, log, done); + sourceEntry, destEntry, kafkaEntry, log); } if (err.InvalidObjectState || err.name === 'InvalidObjectState') { log.info('replication skipped: invalid object state', { entry: sourceEntry.getLogInfo() }); - return done(); + return null; } log.debug('replication failed permanently for object, ' + 'publishing replication status as FAILED', @@ -1104,7 +1066,7 @@ class ReplicateObject extends BackbeatTask { reason: err.description, kafkaEntry, }); - return done(null, { committable: false }); + return { committable: false }; } } diff --git a/lib/tasks/BackbeatTask.js b/lib/tasks/BackbeatTask.js index 7c70d9488f..e8f525f8b6 100644 --- a/lib/tasks/BackbeatTask.js +++ b/lib/tasks/BackbeatTask.js @@ -30,6 +30,15 @@ class BackbeatTask { } retry(args, done) { + if (!done) { + return new Promise((resolve, reject) => + this.retry(args, (err, ...rest) => { + if (err) { + return reject(err); + } + return resolve(rest.length <= 1 ? rest[0] : rest); + })); + } const { actionDesc, logFields, noTimeout, actionFunc, shouldRetryFunc, onRetryFunc, log } = args; const backoffCtx = new BackOff(this.retryParams.backoff); @@ -119,8 +128,9 @@ class BackbeatTask { } return doneOnce(...args); }; - actionFunc(_handleRes, nbRetries); + return actionFunc(_handleRes, nbRetries); } + } module.exports = BackbeatTask; diff --git a/lib/util/mapLimitWaitPendingIfError.js b/lib/util/mapLimitWaitPendingIfError.js deleted file mode 100644 index 10bb43bfc1..0000000000 --- a/lib/util/mapLimitWaitPendingIfError.js +++ /dev/null @@ -1,68 +0,0 @@ -/** - * Custom variant of async.mapLimit, where a failure does not call the - * callback immediately, but waits until the pending requests complete - * before returning the error and the latest array of results. It does - * not trigger the remaining non-started requests after an error - * occurs. - * - * The initial motivation to write this function is for cleaning up - * orphan data after a failed replication: in this case we must wait - * for all pending requests if an error occurs, but not trigger new - * requests, so that we have the full list of orphans to delete - * afterwards without missing the in-progress requests at the time the - * error occurs. - * - * @param {Array} coll - collection to iterate over - * - * @param {number} limit - The maximum number of async operations at a - * time - * - * @param {AsyncFunction} iteratee - An async function to apply to - * each item in coll. The iteratee should complete with the - * transformed item. Invoked with (item, callback). - * - * @param {function} callback - A callback which is called when all - * iteratee functions have finished, or an error occurs. Results is an - * array of the transformed items from the coll. Invoked with (err, - * results). - * - * @return {undefined} - */ -function mapLimitWaitPendingIfError(coll, limit, iteratee, callback) { - if (coll.length === 0) { - return callback(null, []); - } - const results = []; - let nPendingRequests = 0; - let nextIdx = 0; - let pendingError = null; - const processNext = () => { - const idx = nextIdx; - nextIdx += 1; - nPendingRequests += 1; - iteratee(coll[idx], (err, res) => { - nPendingRequests -= 1; - results[idx] = res; - if (err) { - // don't trigger any new request - nextIdx = coll.length; - if (!pendingError) { - pendingError = err; - } - if (nPendingRequests === 0) { - callback(pendingError, results); - } - } else if (nextIdx < coll.length) { - processNext(); - } else if (nPendingRequests === 0) { - callback(pendingError, results); - } - }); - }; - while (nextIdx < Math.min(coll.length, limit)) { - processNext(); - } - return undefined; -} - -module.exports = mapLimitWaitPendingIfError; diff --git a/lib/util/runTasksWithConcurrency.js b/lib/util/runTasksWithConcurrency.js new file mode 100644 index 0000000000..6cf6f2e236 --- /dev/null +++ b/lib/util/runTasksWithConcurrency.js @@ -0,0 +1,49 @@ +/** + * Runs an async task for each item in a collection, up to `limit` at a time. + * On error, no new tasks are started, but already-running tasks are awaited + * before the function resolves. Returns [firstError, results] so that callers + * can inspect partial results (e.g. to clean up orphan data). + * + * The initial motivation to write this function is for cleaning up + * orphan data after a failed replication: in this case we must wait + * for all pending requests if an error occurs, but not trigger new + * requests, so that we have the full list of orphans to delete + * afterwards without missing the in-progress requests at the time the + * error occurs. + * + * @param {Function} task - async function (item) => result + * @param {number} limit - maximum number of concurrent tasks + * @param {Array} coll - collection to iterate over + * @return {Promise<[Error|null, Array]>} - always resolves, never rejects + */ +async function runTasksWithConcurrency(task, limit, coll) { + if (coll.length === 0) { + return [null, []]; + } + const results = new Array(coll.length); + let nextIdx = 0; + const errors = []; + + const worker = async () => { + while (nextIdx < coll.length) { + const idx = nextIdx++; + try { + results[idx] = await task(coll[idx]); + } catch (err) { + errors.push(err); + nextIdx = coll.length; + return; + } + } + }; + + await Promise.all( + Array.from({ length: Math.min(limit, coll.length) }, () => worker()) + ); + + // Return only the first error. + // Subsequent errors from still-pending concurrent tasks are dropped. + return [errors[0] ?? null, results]; +} + +module.exports = runTasksWithConcurrency; diff --git a/tests/functional/replication/streamedCopy.spec.js b/tests/functional/replication/streamedCopy.spec.js index 0e240eaee4..394843cfcf 100644 --- a/tests/functional/replication/streamedCopy.spec.js +++ b/tests/functional/replication/streamedCopy.spec.js @@ -222,7 +222,9 @@ describe('streamed copy functional tests', () => { repTask._setupDestClients('dummyrole', log); repTask._getAndPutPartOnce( mockSourceEntry, mockSourceEntry, mockPartInfo, - log.newRequestLogger(), cb); + log.newRequestLogger()) + .then(r => cb(null, r)) + .catch(cb); }, }, { name: 'MultipleBackendTask::_getAndPutObjectOnce', diff --git a/tests/unit/lib/tasks/BackbeatTask.spec.js b/tests/unit/lib/tasks/BackbeatTask.spec.js index 100dd85720..f82e6724f9 100644 --- a/tests/unit/lib/tasks/BackbeatTask.spec.js +++ b/tests/unit/lib/tasks/BackbeatTask.spec.js @@ -79,6 +79,27 @@ describe('BackbeatTask', () => { }); }); + describe('retry (Promise mode)', () => { + it('should retry on retryable errors and resolve with the final result', async () => { + const task = new BackbeatTask({ timeoutS: 10, backoff: { min: 1, max: 10, jitter: 0, factor: 1 } }); + let attempts = 0; + const result = await task.retry({ + actionDesc: 'test', + actionFunc: done => { + attempts += 1; + if (attempts < 3) { + return done(Object.assign(new Error('retry me'), { retryable: true })); + } + return done(null, 'ok'); + }, + shouldRetryFunc: err => err.retryable, + log: logger, + }); + assert.strictEqual(result, 'ok'); + assert.strictEqual(attempts, 3); + }); + }); + describe('retry method with sinon fake timers', () => { let clock; diff --git a/tests/unit/lib/util/mapLimitWaitPendingIfError.spec.js b/tests/unit/lib/util/mapLimitWaitPendingIfError.spec.js deleted file mode 100644 index 662b0b53b7..0000000000 --- a/tests/unit/lib/util/mapLimitWaitPendingIfError.spec.js +++ /dev/null @@ -1,140 +0,0 @@ -const assert = require('assert'); - -const mapLimitWaitPendingIfError = - require('../../../../lib/util/mapLimitWaitPendingIfError'); - -describe('mapLimitWaitPendingIfError', () => { - it('should process an empty array', done => { - mapLimitWaitPendingIfError([], 10, (item, itemCb) => { - setTimeout(() => itemCb(null, item * 2), Math.random() * 10); - }, (err, results) => { - assert.ifError(err); - assert.deepStrictEqual(results, []); - done(); - }); - }); - [ - { - arrayDesc: 'smaller than the concurrency limit', - limit: 10, - }, - { - arrayDesc: 'equal to the concurrency limit', - limit: 5, - }, - { - arrayDesc: 'larger than the concurrency limit', - limit: 3, - }, - { - arrayDesc: 'with a concurrency limit of 1', - limit: 1, - }, - ].forEach(testCase => { - it(`should process an array ${testCase.arrayDesc}`, done => { - let concurrency = 0; - mapLimitWaitPendingIfError([1, 2, 3, 4, 5], testCase.limit, (item, itemCb) => { - concurrency += 1; - assert(concurrency <= testCase.limit); - setTimeout(() => { - concurrency -= 1; - itemCb(null, item * 2); - }, Math.random() * 10); - }, (err, results) => { - assert.ifError(err); - assert.deepStrictEqual(results, [2, 4, 6, 8, 10]); - done(); - }); - }); - }); - it('should launch tasks in parallel up to the concurrency limit', done => { - let concurrency = 0; - const cbs = []; - let testDone = false; - mapLimitWaitPendingIfError([1, 2, 3, 4, 5], 3, (item, itemCb) => { - concurrency += 1; - assert(concurrency <= 3); - const itemDone = () => { - concurrency -= 1; - process.nextTick(() => itemCb(null, item * 2)); - }; - if (testDone) { - itemDone(); - } else { - cbs.push(itemDone); - if (concurrency === 3) { - setTimeout(() => { - cbs.forEach(cb => cb()); - }, 10); - testDone = true; - } - } - }, (err, results) => { - assert.ifError(err); - assert.deepStrictEqual(results, [2, 4, 6, 8, 10]); - done(); - }); - }); - - it('should stop processing new requests on error', done => { - mapLimitWaitPendingIfError([1, 2, 3, 4, 5], 1, (item, itemCb) => { - // check that no more item is processed after an error - // occurs (limit is 1 so item are processed in order) - assert(item <= 3); - if (item === 3) { - process.nextTick(() => itemCb(new Error('OOPS'), 'error item')); - } else { - process.nextTick(() => itemCb(null, item * 2)); - } - }, (err, results) => { - assert(err); - assert.deepStrictEqual(results, [2, 4, 'error item']); - done(); - }); - }); - - it('should finish all pending requests on error', done => { - let concurrency = 0; - const cbs = []; - mapLimitWaitPendingIfError([1, 2, 3, 4, 5], 5, (item, itemCb) => { - concurrency += 1; - if (concurrency === 5) { - process.nextTick(() => { - itemCb(new Error('OOPS'), 'error item'); - setTimeout(() => { - cbs.forEach(cb => cb()); - }, 10); - }); - } else { - cbs.push(() => itemCb(null, item * 2)); - } - }, (err, results) => { - assert(err); - assert.deepStrictEqual(results, [2, 4, 6, 8, 'error item']); - done(); - }); - }); - - it('should return the first error', done => { - mapLimitWaitPendingIfError([1, 2, 3, 4, 5], 5, (item, itemCb) => { - const errorCb = () => itemCb( - new Error(`OOPS ${item}`), `error item ${item}`); - if (item === 3) { - process.nextTick(errorCb); - } else { - setTimeout(errorCb, 10); - } - }, (err, results) => { - assert(err); - assert.strictEqual(err.message, 'OOPS 3'); - assert.deepStrictEqual(results, [ - 'error item 1', - 'error item 2', - 'error item 3', - 'error item 4', - 'error item 5', - ]); - done(); - }); - }); -}); diff --git a/tests/unit/lib/util/runTasksWithConcurrency.spec.js b/tests/unit/lib/util/runTasksWithConcurrency.spec.js new file mode 100644 index 0000000000..37833f962e --- /dev/null +++ b/tests/unit/lib/util/runTasksWithConcurrency.spec.js @@ -0,0 +1,123 @@ +const assert = require('assert'); + +const runTasksWithConcurrency = + require('../../../../lib/util/runTasksWithConcurrency'); + +const delay = ms => new Promise(resolve => setTimeout(resolve, ms)); + +describe('runTasksWithConcurrency', () => { + it('should process an empty array', async () => { + const [err, results] = await runTasksWithConcurrency( + async item => item * 2, 10, []); + assert.ifError(err); + assert.deepStrictEqual(results, []); + }); + + [ + { arrayDesc: 'smaller than the concurrency limit', limit: 10 }, + { arrayDesc: 'equal to the concurrency limit', limit: 5 }, + { arrayDesc: 'larger than the concurrency limit', limit: 3 }, + { arrayDesc: 'with a concurrency limit of 1', limit: 1 }, + ].forEach(testCase => { + it(`should process an array ${testCase.arrayDesc}`, async () => { + let concurrency = 0; + const [err, results] = await runTasksWithConcurrency( + async item => { + concurrency++; + assert(concurrency <= testCase.limit); + await delay(Math.random() * 10); + concurrency--; + return item * 2; + }, testCase.limit, [1, 2, 3, 4, 5]); + assert.ifError(err); + assert.deepStrictEqual(results, [2, 4, 6, 8, 10]); + }); + }); + + it('should launch tasks in parallel up to the concurrency limit', async () => { + let concurrency = 0; + const pendingResolvers = []; + let scheduled = false; + + const [err, results] = await runTasksWithConcurrency( + item => { + concurrency++; + assert(concurrency <= 3); + return new Promise(resolve => { + const settle = () => { + concurrency--; + resolve(item * 2); + }; + if (!scheduled) { + pendingResolvers.push(settle); + if (concurrency === 3) { + scheduled = true; + setTimeout(() => pendingResolvers.splice(0).forEach(fn => fn()), 10); + } + } else { + settle(); + } + }); + }, 3, [1, 2, 3, 4, 5]); + assert.ifError(err); + assert.deepStrictEqual(results, [2, 4, 6, 8, 10]); + }); + + it('should stop processing new tasks on error', async () => { + const processed = []; + const [err, results] = await runTasksWithConcurrency( + async item => { + processed.push(item); + if (item === 3) { + throw new Error('OOPS'); + } + return item * 2; + }, 1, [1, 2, 3, 4, 5]); + assert(err); + assert.strictEqual(err.message, 'OOPS'); + // items 4 and 5 must not have been started + assert.deepStrictEqual(processed, [1, 2, 3]); + assert.strictEqual(results[0], 2); + assert.strictEqual(results[1], 4); + assert.strictEqual(results[2], undefined); // errored item has no result + }); + + it('should finish all pending tasks on error', async () => { + let concurrency = 0; + const pendingResolvers = []; + + const [err, results] = await runTasksWithConcurrency( + item => { + concurrency++; + if (concurrency === 5) { + setTimeout(() => pendingResolvers.splice(0).forEach(fn => fn()), 10); + return Promise.reject(new Error('OOPS')); + } + return new Promise(resolve => + pendingResolvers.push(() => resolve(item * 2))); + }, 5, [1, 2, 3, 4, 5]); + assert(err); + assert.strictEqual(err.message, 'OOPS'); + // items 1-4 completed despite item 5 erroring first + assert.strictEqual(results[0], 2); + assert.strictEqual(results[1], 4); + assert.strictEqual(results[2], 6); + assert.strictEqual(results[3], 8); + assert.strictEqual(results[4], undefined); // errored item has no result + }); + + it('should return the first error (consistent with original behavior)', async () => { + const [err, results] = await runTasksWithConcurrency( + item => { + if (item === 3) { + return Promise.reject(new Error(`OOPS ${item}`)); + } + return delay(10).then(() => { throw new Error(`OOPS ${item}`); }); + }, 5, [1, 2, 3, 4, 5]); + assert(err); + // item 3 rejects first (no delay), all 5 were already in-flight + assert.strictEqual(err.message, 'OOPS 3'); + assert.strictEqual(results.length, 5); + results.forEach(r => assert.strictEqual(r, undefined)); + }); +}); diff --git a/tests/unit/replication/MultipleBackendTask.js b/tests/unit/replication/MultipleBackendTask.js index 47ef7a26e2..0ababc2f83 100644 --- a/tests/unit/replication/MultipleBackendTask.js +++ b/tests/unit/replication/MultipleBackendTask.js @@ -101,7 +101,7 @@ describe('MultipleBackendTask', function test() { }).setSite('test-site-2'); } - it('matches V2 rules by Filter.Prefix', done => { + it('matches V2 rules by Filter.Prefix', () => { sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { send: () => Promise.resolve({ @@ -119,13 +119,10 @@ describe('MultipleBackendTask', function test() { }), }; - task._setupRolesOnce(makeEntry(), fakeLogger, err => { - assert.ifError(err); - done(); - }); + return task._setupRolesOnce(makeEntry(), fakeLogger); }); - it('rejects with PreconditionFailed when no rule matches the object key', done => { + it('rejects with PreconditionFailed when no rule matches the object key', () => { sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { send: () => Promise.resolve({ @@ -143,14 +140,15 @@ describe('MultipleBackendTask', function test() { }), }; - task._setupRolesOnce(makeEntry(), fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.PreconditionFailed, true); - done(); - }); + return task._setupRolesOnce(makeEntry(), fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.PreconditionFailed, true); + }); }); - it('accepts V1 rules with top-level Prefix', done => { + it('accepts V1 rules with top-level Prefix', () => { sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { send: () => Promise.resolve({ @@ -168,10 +166,7 @@ describe('MultipleBackendTask', function test() { }), }; - task._setupRolesOnce(makeEntry(), fakeLogger, err => { - assert.ifError(err); - done(); - }); + return task._setupRolesOnce(makeEntry(), fakeLogger); }); }); @@ -401,9 +396,9 @@ describe('MultipleBackendTask', function test() { fakeLogger.newRequestLogger = () => fakeLogger; queueEntry = QueueEntry.createFromKafkaEntry(replicationEntry); sinon.stub(task, '_setupClients').yields(null); - sinon.stub(task, '_refreshSourceEntry').yields(null, queueEntry); + sinon.stub(task, '_refreshSourceEntry').resolves(queueEntry); sinon.stub(task, '_handleReplicationOutcome').callsFake( - (err, sourceEntry, kafkaEntry, log, done) => done(err, null)); + err => (err ? Promise.reject(err) : Promise.resolve(null))); }); afterEach(() => { diff --git a/tests/unit/replication/ReplicateObject.spec.js b/tests/unit/replication/ReplicateObject.spec.js index beb26c45a4..ed50a6c234 100644 --- a/tests/unit/replication/ReplicateObject.spec.js +++ b/tests/unit/replication/ReplicateObject.spec.js @@ -150,20 +150,19 @@ describe('ReplicateObject', () => { }; } - it('should return collision info on VersionIdCollisionException', done => { + it('should return collision info on VersionIdCollisionException', () => { 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(); - }); + return task._getAndPutPartOnce(makeSourceEntry(older), makeDestEntry(), part, fakeLogger) + .then(result => { + 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); + }); }); - it('should set data location and publish metrics when no collision', done => { + it('should set data location and publish metrics when no collision', () => { task.S3source = { send: sinon.stub().resolves({ Body: makeBodyStream(), ContentLength: 10 }), }; @@ -172,19 +171,18 @@ describe('ReplicateObject', () => { 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, + return task._getAndPutPartOnce(makeSourceEntry(), makeDestEntry(), part, fakeLogger) + .then(result => { + assert.deepStrictEqual(result, { + key: 'new-key', + start: 0, + size: 10, + dataStoreName: 'file', + dataStoreETag: '1:abc', + blockId: undefined, + }); + sinon.assert.calledOnce(task._publishDataWriteMetrics); }); - sinon.assert.calledOnce(task._publishDataWriteMetrics); - done(); - }); }); }); @@ -211,30 +209,24 @@ describe('ReplicateObject', () => { }; } - it('should not set Expect header for objects below the threshold', done => { + it('should not set Expect header for objects below the threshold', () => { task.S3source = { send: sinon.stub().resolves({ Body: makeBodyStream(), ContentLength: replicationExpectContinueThreshold - 1, }) }; const { dest, getExpect } = makeDestWithExpectCapture(); task.backbeatDest = dest; - task._getAndPutPartOnce(makeSourceEntry(), makeDestEntry(), part, fakeLogger, err => { - assert.ifError(err); - assert.strictEqual(getExpect(), undefined); - done(); - }); + return task._getAndPutPartOnce(makeSourceEntry(), makeDestEntry(), part, fakeLogger) + .then(() => assert.strictEqual(getExpect(), undefined)); }); - it('should set Expect header for objects at or above the threshold', done => { + it('should set Expect header for objects at or above the threshold', () => { task.S3source = { send: sinon.stub().resolves({ Body: makeBodyStream(), ContentLength: replicationExpectContinueThreshold, }) }; const { dest, getExpect } = makeDestWithExpectCapture(); task.backbeatDest = dest; - task._getAndPutPartOnce(makeSourceEntry(), makeDestEntry(), part, fakeLogger, err => { - assert.ifError(err); - assert.strictEqual(getExpect(), '100-continue'); - done(); - }); + return task._getAndPutPartOnce(makeSourceEntry(), makeDestEntry(), part, fakeLogger) + .then(() => assert.strictEqual(getExpect(), '100-continue')); }); }); @@ -246,39 +238,36 @@ describe('ReplicateObject', () => { task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; }); - it('should pass through MicroVersionIdAlreadyStoredException and skip metrics', done => { + it('should pass through MicroVersionIdAlreadyStoredException and skip metrics', () => { 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(); - }); + return task._putMetadataOnce(entry, false, null, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert.ok(err instanceof MicroVersionIdAlreadyStoredException); + sinon.assert.notCalled(metricsStub); + }); }); - it('should pass through StaleMicroVersionIdException', done => { + it('should pass through StaleMicroVersionIdException', () => { 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(); - }); + return task._putMetadataOnce(entry, false, null, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => assert.ok(err instanceof StaleMicroVersionIdException)); }); - it('should publish metrics and succeed on normal response', done => { + it('should publish metrics and succeed on normal response', () => { 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(); - }); + return task._putMetadataOnce(entry, false, null, fakeLogger) + .then(() => sinon.assert.calledOnce(metricsStub)); }); }); @@ -297,103 +286,83 @@ describe('ReplicateObject', () => { sinon.stub(sourceEntry, 'getReplicationSiteDataStoreVersionId').returns('v1'); }); - it('should mark COMPLETED for MicroVersionIdAlreadyStoredException', done => { - task._handleReplicationOutcome( + it('should mark COMPLETED for MicroVersionIdAlreadyStoredException', async () => { + await task._handleReplicationOutcome( new MicroVersionIdAlreadyStoredException({}), - sourceEntry, destEntry, kafkaEntry, fakeLogger, () => { - sinon.assert.calledWith(task._publishReplicationStatus, - sourceEntry, 'COMPLETED', sinon.match.any); - done(); - }); + sourceEntry, destEntry, kafkaEntry, fakeLogger); + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'COMPLETED', sinon.match.any); }); - it('should mark COMPLETED for StaleMicroVersionIdException', done => { - task._handleReplicationOutcome( + it('should mark COMPLETED for StaleMicroVersionIdException', async () => { + await task._handleReplicationOutcome( new StaleMicroVersionIdException({}), - sourceEntry, destEntry, kafkaEntry, fakeLogger, () => { - sinon.assert.calledWith(task._publishReplicationStatus, - sourceEntry, 'COMPLETED', sinon.match.any); - done(); - }); + sourceEntry, destEntry, kafkaEntry, fakeLogger); + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'COMPLETED', sinon.match.any); }); - 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 COMPLETED on successful replication', async () => { + await task._handleReplicationOutcome( + null, sourceEntry, destEntry, kafkaEntry, fakeLogger); + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'COMPLETED', sinon.match.any); }); - it('should mark FAILED for real errors', done => { + it('should mark FAILED for real errors', async () => { 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(); - }); + await task._handleReplicationOutcome( + realErr, sourceEntry, destEntry, kafkaEntry, fakeLogger); + sinon.assert.calledWith(task._publishReplicationStatus, + sourceEntry, 'FAILED', sinon.match.any); }); }); describe('_setTargetAccountMd', () => { - it('should skip gettin target account info when auth type is assumeRole', done => { + it('should skip gettin target account info when auth type is assumeRole', () => { sinon.stub(task, '_setupDestClients').returns(); - const setTargetAccountStub = sinon.stub(task, '_setTargetAccountMdOnce').yields(); - task._setTargetAccountMd({}, '', fakeLogger, err => { - assert.ifError(err); - assert(setTargetAccountStub.notCalled); - done(); - }); + const setTargetAccountStub = sinon.stub(task, '_setTargetAccountMdOnce').resolves(); + return task._setTargetAccountMd({}, '', fakeLogger) + .then(() => assert(setTargetAccountStub.notCalled)); }); - it('should get target account info', done => { + it('should get target account info', () => { sinon.stub(task, '_setupDestClients').returns(); - const setTargetAccountStub = sinon.stub(task, '_setTargetAccountMdOnce').yields(); + const setTargetAccountStub = sinon.stub(task, '_setTargetAccountMdOnce').resolves(); task.destConfig.auth = { type: 'service', account: 'replication-service', }; - task._setTargetAccountMd({ getLogInfo: () => {} }, '', fakeLogger, err => { - assert.ifError(err); - assert(setTargetAccountStub.calledOnce); - done(); - }); + return task._setTargetAccountMd({ getLogInfo: () => {} }, '', fakeLogger) + .then(() => assert(setTargetAccountStub.calledOnce)); }); }); describe('_putMetadataOnce', () => { - it('should pass extract accountId from role and pass it when using AssumeRole auth', done => { + it('should pass extract accountId from role and pass it when using AssumeRole auth', () => { sinon.stub(task, '_publishMetadataWriteMetrics').returns(); const entry = QueueEntry.createFromKafkaEntry(replicationEntry); const sendStub = sinon.stub().resolves({}); - task.backbeatDest = { - send: sendStub, - }; + task.backbeatDest = { send: sendStub }; task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; - task._putMetadataOnce(entry, true, null, fakeLogger, err => { - assert.ifError(err); - assert(sendStub.calledOnce); - assert.deepStrictEqual(sendStub.firstCall.args[0].input.AccountId, '123456789012'); - done(); - }); + return task._putMetadataOnce(entry, true, null, fakeLogger) + .then(() => { + assert(sendStub.calledOnce); + assert.deepStrictEqual(sendStub.firstCall.args[0].input.AccountId, '123456789012'); + }); }); - it('should not pass accountId when not in assumeRole', done => { + it('should not pass accountId when not in assumeRole', () => { sinon.stub(task, '_publishMetadataWriteMetrics').returns(); const entry = QueueEntry.createFromKafkaEntry(replicationEntry); const sendStub = sinon.stub().resolves({}); - task.backbeatDest = { - send: sendStub, - }; + task.backbeatDest = { send: sendStub }; task.targetRole = 'arn:aws:iam::123456789012:role/crr-role'; sinon.stub(task.destConfig.auth, 'type').value('role'); - task._putMetadataOnce(entry, true, null, fakeLogger, err => { - assert.ifError(err); - assert(sendStub.calledOnce); - assert.strictEqual(sendStub.firstCall.args[0].input.AccountId, undefined); - done(); - }); + return task._putMetadataOnce(entry, true, null, fakeLogger) + .then(() => { + assert(sendStub.calledOnce); + assert.strictEqual(sendStub.firstCall.args[0].input.AccountId, undefined); + }); }); }); @@ -495,15 +464,15 @@ describe('ReplicateObject', () => { 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, '_setupRoles').resolves(['srcRole', 'dstRole']); + sinon.stub(task, '_setTargetAccountMd').resolves(); sinon.stub(task, '_publishReplicationStatus'); const putMetadataStub = sinon.stub(task, '_putMetadata') - .callsFake((e, mdOnly, conflict, l, cb) => { + .callsFake((e, mdOnly) => { assert.strictEqual(mdOnly, false, 'zero-byte objects must use DATA,METADATA (create) mode, not METADATA-only (update) mode'); - cb(null); + return Promise.resolve(); }); task.processQueueEntry(sourceEntry, {}, () => { @@ -518,19 +487,19 @@ describe('ReplicateObject', () => { const sourceEntry = QueueEntry.createFromKafkaEntry(replicationEntry); const destEntry = makeDestEntry(); sinon.stub(task, '_publishReplicationStatus').returns(); - sinon.stub(task, '_deleteOrphans').callsFake((entry, locations, log, cb) => cb()); - sinon.stub(task, '_getAndPutData').callsFake((src, dest, log, cb) => - cb(null, writtenLocations, undefined)); - sinon.stub(task, '_putMetadata').callsFake((entry, mdOnly, conflict, log, cb) => - cb(new MicroVersionIdAlreadyStoredException({ message: 'collision' }))); - - task._processQueueEntryRetryFull(sourceEntry, destEntry, {}, fakeLogger, err => { - assert.ifError(err); - sinon.assert.calledOnce(task._deleteOrphans); - sinon.assert.calledWith(task._deleteOrphans, - destEntry, writtenLocations, sinon.match.any, sinon.match.any); - done(); - }); + sinon.stub(task, '_deleteOrphans').resolves(); + sinon.stub(task, '_getAndPutData').resolves([writtenLocations, undefined]); + sinon.stub(task, '_putMetadata').rejects( + new MicroVersionIdAlreadyStoredException({ message: 'collision' })); + + task._processQueueEntryRetryFull(sourceEntry, destEntry, {}, fakeLogger) + .then(() => { + sinon.assert.calledOnce(task._deleteOrphans); + sinon.assert.calledWith(task._deleteOrphans, + destEntry, writtenLocations, sinon.match.any); + done(); + }) + .catch(done); }); describe('_publishReplicationStatus', () => { @@ -606,7 +575,7 @@ describe('ReplicateObject', () => { return entry.setReplicationBackend(backends[0]); } - it('validates per-backend role via account substitution', done => { + it('validates per-backend role via account substitution', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -630,15 +599,14 @@ describe('ReplicateObject', () => { destination: 'arn:aws:s3:::bucket-a', role: 'arn:aws:iam::222:role/repRule', }]); - task._setupRolesOnce(entry, fakeLogger, (err, src, dst) => { - assert.ifError(err); - assert.strictEqual(src, 'arn:aws:iam::111:role/src'); - assert.strictEqual(dst, 'arn:aws:iam::222:role/repRule'); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(([src, dst]) => { + assert.strictEqual(src, 'arn:aws:iam::111:role/src'); + assert.strictEqual(dst, 'arn:aws:iam::222:role/repRule'); + }); }); - it('rejects when per-backend role does not match substituted role', done => { + it('rejects when per-backend role does not match substituted role', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -662,14 +630,15 @@ describe('ReplicateObject', () => { destination: 'arn:aws:s3:::bucket-a', role: 'arn:aws:iam::999:role/repRule', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.BadRole, true); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.BadRole, true); + }); }); - it('matches V2 rules by Filter.Prefix', done => { + it('matches V2 rules by Filter.Prefix', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -712,13 +681,10 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::222:role/dst', }); - task._setupRolesOnce(entry, fakeLogger, err => { - assert.ifError(err); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger); }); - it('rejects with PreconditionFailed when V1 prefix does not match the object key', done => { + it('rejects with PreconditionFailed when V1 prefix does not match the object key', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -745,14 +711,15 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::222:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.PreconditionFailed, true); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.PreconditionFailed, true); + }); }); - it('rejects with PreconditionFailed when the only matching rule is Disabled', done => { + it('rejects with PreconditionFailed when the only matching rule is Disabled', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -778,14 +745,15 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::222:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.PreconditionFailed, true); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.PreconditionFailed, true); + }); }); - it('accepts when at least one enabled rule matches among several', done => { + it('accepts when at least one enabled rule matches among several', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -822,13 +790,10 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::222:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert.ifError(err); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger); }); - it('rejects with PreconditionFailed when no rule matches', done => { + it('rejects with PreconditionFailed when no rule matches', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -863,14 +828,15 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::222:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.PreconditionFailed, true); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.PreconditionFailed, true); + }); }); - it('rejects with BadRole when the bucket config role has more than two ARNs', done => { + it('rejects with BadRole when the bucket config role has more than two ARNs', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -897,14 +863,15 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::222:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.BadRole, true); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.BadRole, true); + }); }); - it('picks the rule matching the backend destination when several share a StorageClass', done => { + it('picks the rule matching the backend destination when several share a StorageClass', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -943,15 +910,14 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::333:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, (err, src, dst) => { - assert.ifError(err); - assert.strictEqual(src, 'arn:aws:iam::111:role/src'); - assert.strictEqual(dst, 'arn:aws:iam::333:role/dst'); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(([src, dst]) => { + assert.strictEqual(src, 'arn:aws:iam::111:role/src'); + assert.strictEqual(dst, 'arn:aws:iam::333:role/dst'); + }); }); - it('rejects when the backend role does not match its rule Account substitution', done => { + it('rejects when the backend role does not match its rule Account substitution', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -990,14 +956,15 @@ describe('ReplicateObject', () => { role: 'arn:aws:iam::999:role/dst', }]); - task._setupRolesOnce(entry, fakeLogger, err => { - assert(err); - assert.strictEqual(err.is.BadRole, true); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(() => assert.fail('expected error')) + .catch(err => { + assert(err); + assert.strictEqual(err.is.BadRole, true); + }); }); - it('falls back to literal compare for legacy configs without Account', done => { + it('falls back to literal compare for legacy configs without Account', () => { task.site = 'site'; sinon.stub(task, '_setupSourceClients').returns(); task.S3source = { @@ -1027,12 +994,11 @@ describe('ReplicateObject', () => { }], }, }).setSite('site'); - task._setupRolesOnce(entry, fakeLogger, (err, src, dst) => { - assert.ifError(err); - assert.strictEqual(src, 'arn:aws:iam::111:role/src'); - assert.strictEqual(dst, 'arn:aws:iam::222:role/legacy'); - done(); - }); + return task._setupRolesOnce(entry, fakeLogger) + .then(([src, dst]) => { + assert.strictEqual(src, 'arn:aws:iam::111:role/src'); + assert.strictEqual(dst, 'arn:aws:iam::222:role/legacy'); + }); }); }); @@ -1143,7 +1109,7 @@ describe('ReplicateObject', () => { }); describe('_putMetadataOnce with conflict', () => { - it('skips the request when conflict revision is equal to source (already at destination)', done => { + it('skips the request when conflict revision is equal to source (already at destination)', () => { sinon.stub(task, '_publishMetadataWriteMetrics').returns(); const { newer, newerEncoded } = makeMicroVersionIds(); const entry = QueueEntry.createFromKafkaEntry(replicationEntry); @@ -1152,14 +1118,11 @@ describe('ReplicateObject', () => { 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(); - }); + return task._putMetadataOnce(entry, false, conflict, fakeLogger) + .then(() => sinon.assert.notCalled(sendStub)); }); - it('proceeds with the request when conflict revision is older than source', done => { + it('proceeds with the request when conflict revision is older than source', () => { sinon.stub(task, '_publishMetadataWriteMetrics').returns(); const { olderEncoded, newer } = makeMicroVersionIds(); const entry = QueueEntry.createFromKafkaEntry(replicationEntry); @@ -1168,11 +1131,8 @@ describe('ReplicateObject', () => { 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(); - }); + return task._putMetadataOnce(entry, false, conflict, fakeLogger) + .then(() => sinon.assert.calledOnce(sendStub)); }); });