<commit_before>const async = require('async');
const AWS = require('aws-sdk');
const util = require('util');

const errors = require('arsenal').errors;
const jsutil = require('arsenal').jsutil;
const ObjectMDLocation = require('arsenal').models.ObjectMDLocation;

const BackbeatClient = require('../../../lib/clients/BackbeatClient');
const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy');

const mapLimitWaitPendingIfError = require('../../../lib/util/mapLimitWaitPendingIfError');
const { attachReqUids } = require('../../../lib/clients/utils');
const getExtMetrics = require('../utils/getExtMetrics');
const BackbeatTask = require('../../../lib/tasks/BackbeatTask');
const { getAccountCredentials } = require('../../../lib/credentials/AccountCredentials');
const RoleCredentials = require('../../../lib/credentials/RoleCredentials');
const { metricsExtension, metricsTypeQueued, metricsTypeCompleted, replicationStages } = require('../constants');

const notifConstants = require('../../notification/constants');
const messageUtil = require('../../notification/utils/message');
const configUtil = require('../../notification/utils/config');

function _extractAccountIdFromRole(role) {
    return role.split(':')[4];
}

class ReplicateObject extends BackbeatTask {
    /**
     * Process a single replication entry
     *
     * @constructor
     * @param {QueueProcessor} qp - queue processor instance
     */
    constructor(qp) {
        const qpState = qp.getStateVars();
        super();
        Object.assign(this, qpState);
        this.retryParams = this.repConfig.queueProcessor.retry.scality;
        if (this.destConfig && this.destConfig.bootstrapList) {
            const destination = this.destConfig.bootstrapList
                .find(endpoint => endpoint.site === this.site) || {};
            this.destType = destination.type;
            const retryParams =
                this.repConfig.queueProcessor.retry[destination.type];
            if (retryParams) {
                this.retryParams = retryParams;
            }
        }
        this.sourceRole = null;
        this.targetRole = null;
        this.destBackbeatHost = null;
        this.s3sourceCredentials = null;
        this.s3destCredentials = null;
        this.S3source = null;
        this.backbeatSource = null;
        this.backbeatSourceProxy = null;
        this.backbeatDest = null;

        if (this.notificationConfigManager) {
            // callback version of notifcationConfigManager's getConfig function
            this.getNotificationConfig = util.callbackify(this.notificationConfigManager.getConfig
                .bind(this.notificationConfigManager));
        }
    }

    _createCredentials(where, authConfig, roleArn, log) {
        const accountCredentials = getAccountCredentials(authConfig, log);
        if (accountCredentials) {
            return accountCredentials;
        }
        let vaultclient;
        if (where === 'source') {
            vaultclient = this.vaultclientCache.getClient('source:s3');
        } else { // target
            const { host, port } = this.destHosts.pickHost();
            vaultclient = this.vaultclientCache.getClient('dest:s3',
                host, port);
        }
        return new RoleCredentials(vaultclient, 'replication', roleArn, log);
    }

    _setupRoles(entry, log, cb) {
        this.retry({
            actionDesc: 'get bucket replication configuration',
            logFields: { entry: entry.getLogInfo() },
            actionFunc: done => this._setupRolesOnce(entry, log, 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) {
        if (!this.destHosts) {
            log.warn('cannot process entry: no target site configured',
                { entry: destEntry.getLogInfo() });
            return cb(errors.InternalError);
        }
        this._setupDestClients(this.targetRole, log);

        return this.retry({
            actionDesc: 'lookup target account attributes',
            logFields: { entry: destEntry.getLogInfo() },
            actionFunc: done => this._setTargetAccountMdOnce(
                destEntry, targetRole, log, done),
            // this call uses our own Vault client which does not set
            // the 'retryable' field
            shouldRetryFunc: err =>
            (err.InternalError || err.code === 'InternalError' ||
                err.ServiceUnavailable || err.code === 'ServiceUnavailable'),
            onRetryFunc: () => {
                this.destHosts.pickNextHost();
                this._setupDestClients(this.targetRole, log);
            },
            log,
        }, cb);
    }

    _getAndPutPart(sourceEntry, destEntry, part, log, cb) {
        const partLogger = this.logger.newRequestLogger(log.getUids());
        this.retry({
            actionDesc: 'stream part data',
            logFields: { entry: sourceEntry.getLogInfo(), part },
            actionFunc: done => this._getAndPutPartOnce(
                sourceEntry, destEntry, part, partLogger, done),
            shouldRetryFunc: err => err.retryable,
            onRetryFunc: err => {
                if (err.origin === 'target') {
                    this.destHosts.pickNextHost();
                    this._setupDestClients(this.targetRole, partLogger);
                }
            },
            log: partLogger,
        }, cb);
    }

    _putMetadata(entry, mdOnly, log, cb) {
        this.retry({
            actionDesc: 'update metadata on target',
            logFields: { entry: entry.getLogInfo() },
            actionFunc: done => this._putMetadataOnce(entry, mdOnly,
                log, done),
            shouldRetryFunc: err => err.retryable,
            onRetryFunc: err => {
                if (err.origin === 'target') {
                    this.destHosts.pickNextHost();
                    this._setupDestClients(this.targetRole, log);
                }
            },
            log,
        }, cb);
    }

    _getUpdatedSourceEntry(params) {
        const { sourceEntry, replicationStatus } = params;
        const entry = replicationStatus === 'COMPLETED' ?
              sourceEntry.toCompletedEntry(this.site) :
              sourceEntry.toFailedEntry(this.site);
        const versionId =
              sourceEntry.getReplicationSiteDataStoreVersionId(this.site);
        return entry.setReplicationSiteDataStoreVersionId(this.site,
            versionId);
    }

    /**
     * Publishes the failed notification status
     * into the notification topic
     * @param {Object} sourceEntry the object entry
     * @param {Logger} log the logger instance
     * @param {Function} done callback
     * @return {undefined}
     */
    _publishFailedReplicationStatusNotification(sourceEntry, log, done) {
        const bucket = sourceEntry.getBucket();
        const key = sourceEntry.getObjectKey();
        const value = sourceEntry.getValue();
        const versionId = sourceEntry.getVersionId();
        this.getNotificationConfig(bucket, (err, config) => {
            // we skip if key is a master or if no config is available for
            // the bucket
            if (versionId && config && Object.keys(config).length > 0) {
                const eventType = notifConstants.replicationFailedEvent;
                const ent = {
                    bucket,
                    key,
                    versionId,
                    eventType,
                };
                log.debug('validating entry', {
                    method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                    bucket,
                    key,
                    versionId,
                    eventType,
                });
                // validate and push kafka message to each destination internal topic
                return async.each(this.notificationConfig.destinations,
                    (destination, cb) => {
                        // get destination specific notification config
                        const destBnConf = config.queueConfig.find(
                            c => c.queueArn.split(':').pop()
                            === destination.resource);
                        if (!destBnConf) {
                            // skip, if there is no config for the current
                            // destination resource
                            return cb();
                        }
                        // pass only destination resource specific config to
                        // validate entry
                        const bnConfig = {
                            queueConfig: [destBnConf],
                        };
                        // skip if entry doesn't match config
                        if (!configUtil.validateEntry(bnConfig, ent)) {
                            return cb();
                        }
                        const message
                            = messageUtil.addLogAttributes(value, ent);
                        log.info('publishing replication failed notification', {
                            method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                            bucket,
                            key: message.key,
                            eventType,
                            eventTime: message.dateTime,
                        });
                        const entry = {
                            key: encodeURIComponent(bucket),
                            message: JSON.stringify(message)
                        };
                        return this.notificationProducers[destination.resource].send([entry], err => {
                            if (err) {
                                log.error('error in entry delivery to notification topic', {
                                    method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                                    destination: destination.resource,
                                    entry: sourceEntry.getLogInfo(),
                                    error: err,
                                });
                            }
                            return cb();
                        });
                }, err => {
                    if (err) {
                        log.error('error while pushing replication failed notification', {
                            method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                            entry: sourceEntry.getLogInfo(),
                            error: err,
                        });
                    }
                    return done(null, { committable: false });
                });
            }
            return done(null, { committable: false });
        });
    }

    _publishReplicationStatus(sourceEntry, replicationStatus, params) {
        const { log, reason, kafkaEntry } = params;
        const entryParams = { sourceEntry, replicationStatus };
        const updatedSourceEntry = this._getUpdatedSourceEntry(entryParams);
        const updateData = sourceEntry.getReplicationContent().includes('DATA');
        const kafkaEntries = [updatedSourceEntry.toKafkaEntry(this.site)];
        this.replicationStatusProducer.send(kafkaEntries, err => {
            if (err) {
                log.error('error in entry delivery to replication status topic', {
                    method: 'ReplicateObject._publishReplicationStatus',
                    topic: this.repConfig.replicationStatusTopic,
                    entry: updatedSourceEntry.getLogInfo(),
                    replicationStatus,
                    error: err,
                });
            } else {
                log.info('replication status published', {
                    topic: this.repConfig.replicationStatusTopic,
                    entry: updatedSourceEntry.getLogInfo(),
                    replicationStatus,
                    reason,
                });
                this.metricsHandler.metadataReplicationStatus({ replicationStatus });
                if (updateData) {
                    this.metricsHandler.dataReplicationStatus({ replicationStatus });
                }
            }
            // Commit whether there was an error or not to allow
            // progress of the consumer, as best effort measure when
            // there are errors. We can count on the sweeper to retry
            // entries that failed to be published to kafka (because
            // they will keep their PENDING status).
            if (this.consumer) {
                this.consumer.onEntryCommittable(kafkaEntry);
            }
        });
    }

    _setupRolesOnce(entry, log, cb) {
        log.debug('getting bucket replication',
            { entry: entry.getLogInfo() });
        const entryRolesString = entry.getReplicationRoles();
        let entryRoles;
        if (entryRolesString !== undefined) {
            entryRoles = entryRolesString.split(',');
        }
        if (entryRoles === undefined || entryRoles.length !== 2) {
            log.error('expecting two roles separated by a ' +
                'comma in entry replication configuration',
                {
                    method: 'ReplicateObject._setupRolesOnce',
                    entry: entry.getLogInfo(),
                    roles: entryRolesString,
                });
            return cb(errors.BadRole);
        }
        this.sourceRole = entryRoles[0];
        this.targetRole = entryRoles[1];

        this._setupSourceClients(this.sourceRole, log);

        const req = this.S3source.getBucketReplication(
            { Bucket: entry.getBucket() });
        attachReqUids(req, log);
        return req.send((err, data) => {
            if (err) {
                // eslint-disable-next-line no-param-reassign
                err.origin = 'source';
                log.error('error getting replication ' +
                    'configuration from S3',
                    {
                        method: 'ReplicateObject._setupRolesOnce',
                        entry: entry.getLogInfo(),
                        origin: 'source',
                        peer: this.sourceConfig.s3,
                        error: err.message,
                        httpStatus: err.statusCode,
                    });
                return cb(err);
            }
            const replicationEnabled = (
                data.ReplicationConfiguration.Rules.some(
                    rule => entry.getObjectKey().startsWith(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 two roles separated by a ' +
                    'comma 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);
            }
            if (roles[1] !== 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: roles[1],
                    });
                return cb(errors.BadRole);
            }
            return cb(null, roles[0], roles[1]);
        });
    }

    _setTargetAccountMdOnce(destEntry, targetRole, log, cb) {
        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 };
                            }
                        }
                    }
                    log.error('an error occurred when looking up target ' +
                        'account attributes',
                        {
                            method: 'ReplicateObject._setTargetAccountMdOnce',
                            entry: destEntry.getLogInfo(),
                            origin: 'target',
                            peer,
                            error: err.message,
                        });
                    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();
            });
    }

    _getAndPutData(sourceEntry, destEntry, log, cb) {
        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';
            log.error(errMessage, {
                method: 'ReplicateObject._getAndPutData',
                entry: sourceEntry.getLogInfo(),
            });
            return cb(errors.InternalError.customizeDescription(errMessage));
        }
        // For Replication Replay "manual testing" only, uncomment the following lines.
        // if (Math.random() < 0.5) {
        // return process.nextTick(() => cb(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, destLocations) => {
            if (err) {
                return this._deleteOrphans(destEntry, destLocations, log, () => cb(err));
            }
            return cb(null, destLocations);
        });
    }

    _getAndPutPartOnce(sourceEntry, destEntry, part, log, done) {
        const serviceName = this.serviceName;
        const doneOnce = jsutil.once(done);
        const partObj = new ObjectMDLocation(part);
        const partNumber = partObj.getPartNumber();
        const partSize = partObj.getPartSize();
        let destReq = null;<commit_msg>Allow injecting replication errors<commit_after>const async = require('async');
const AWS = require('aws-sdk');
const util = require('util');

const errors = require('arsenal').errors;
const jsutil = require('arsenal').jsutil;
const ObjectMDLocation = require('arsenal').models.ObjectMDLocation;

const BackbeatClient = require('../../../lib/clients/BackbeatClient');
const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy');

const mapLimitWaitPendingIfError = require('../../../lib/util/mapLimitWaitPendingIfError');
const { attachReqUids } = require('../../../lib/clients/utils');
const getExtMetrics = require('../utils/getExtMetrics');
const BackbeatTask = require('../../../lib/tasks/BackbeatTask');
const { getAccountCredentials } = require('../../../lib/credentials/AccountCredentials');
const RoleCredentials = require('../../../lib/credentials/RoleCredentials');
const { metricsExtension, metricsTypeQueued, metricsTypeCompleted, replicationStages } = require('../constants');

const notifConstants = require('../../notification/constants');
const messageUtil = require('../../notification/utils/message');
const configUtil = require('../../notification/utils/config');

function _extractAccountIdFromRole(role) {
    return role.split(':')[4];
}

// BACKBEAT_INJECT_REPLICATION_ERROR_RATE variable can be set to randomly introduce errors.
// When set, the value is the target percentage of errors.
const BACKBEAT_INJECT_REPLICATION_ERROR_RATE =
    process.env.BACKBEAT_INJECT_REPLICATION_ERROR_RATE / 100;

class ReplicateObject extends BackbeatTask {
    /**
     * Process a single replication entry
     *
     * @constructor
     * @param {QueueProcessor} qp - queue processor instance
     */
    constructor(qp) {
        const qpState = qp.getStateVars();
        super();
        Object.assign(this, qpState);
        this.retryParams = this.repConfig.queueProcessor.retry.scality;
        if (this.destConfig && this.destConfig.bootstrapList) {
            const destination = this.destConfig.bootstrapList
                .find(endpoint => endpoint.site === this.site) || {};
            this.destType = destination.type;
            const retryParams =
                this.repConfig.queueProcessor.retry[destination.type];
            if (retryParams) {
                this.retryParams = retryParams;
            }
        }
        this.sourceRole = null;
        this.targetRole = null;
        this.destBackbeatHost = null;
        this.s3sourceCredentials = null;
        this.s3destCredentials = null;
        this.S3source = null;
        this.backbeatSource = null;
        this.backbeatSourceProxy = null;
        this.backbeatDest = null;

        if (this.notificationConfigManager) {
            // callback version of notifcationConfigManager's getConfig function
            this.getNotificationConfig = util.callbackify(this.notificationConfigManager.getConfig
                .bind(this.notificationConfigManager));
        }
    }

    _createCredentials(where, authConfig, roleArn, log) {
        const accountCredentials = getAccountCredentials(authConfig, log);
        if (accountCredentials) {
            return accountCredentials;
        }
        let vaultclient;
        if (where === 'source') {
            vaultclient = this.vaultclientCache.getClient('source:s3');
        } else { // target
            const { host, port } = this.destHosts.pickHost();
            vaultclient = this.vaultclientCache.getClient('dest:s3',
                host, port);
        }
        return new RoleCredentials(vaultclient, 'replication', roleArn, log);
    }

    _setupRoles(entry, log, cb) {
        this.retry({
            actionDesc: 'get bucket replication configuration',
            logFields: { entry: entry.getLogInfo() },
            actionFunc: done => this._setupRolesOnce(entry, log, 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) {
        if (!this.destHosts) {
            log.warn('cannot process entry: no target site configured',
                { entry: destEntry.getLogInfo() });
            return cb(errors.InternalError);
        }
        this._setupDestClients(this.targetRole, log);

        return this.retry({
            actionDesc: 'lookup target account attributes',
            logFields: { entry: destEntry.getLogInfo() },
            actionFunc: done => this._setTargetAccountMdOnce(
                destEntry, targetRole, log, done),
            // this call uses our own Vault client which does not set
            // the 'retryable' field
            shouldRetryFunc: err =>
            (err.InternalError || err.code === 'InternalError' ||
                err.ServiceUnavailable || err.code === 'ServiceUnavailable'),
            onRetryFunc: () => {
                this.destHosts.pickNextHost();
                this._setupDestClients(this.targetRole, log);
            },
            log,
        }, cb);
    }

    _getAndPutPart(sourceEntry, destEntry, part, log, cb) {
        const partLogger = this.logger.newRequestLogger(log.getUids());
        this.retry({
            actionDesc: 'stream part data',
            logFields: { entry: sourceEntry.getLogInfo(), part },
            actionFunc: done => this._getAndPutPartOnce(
                sourceEntry, destEntry, part, partLogger, done),
            shouldRetryFunc: err => err.retryable,
            onRetryFunc: err => {
                if (err.origin === 'target') {
                    this.destHosts.pickNextHost();
                    this._setupDestClients(this.targetRole, partLogger);
                }
            },
            log: partLogger,
        }, cb);
    }

    _putMetadata(entry, mdOnly, log, cb) {
        this.retry({
            actionDesc: 'update metadata on target',
            logFields: { entry: entry.getLogInfo() },
            actionFunc: done => this._putMetadataOnce(entry, mdOnly,
                log, done),
            shouldRetryFunc: err => err.retryable,
            onRetryFunc: err => {
                if (err.origin === 'target') {
                    this.destHosts.pickNextHost();
                    this._setupDestClients(this.targetRole, log);
                }
            },
            log,
        }, cb);
    }

    _getUpdatedSourceEntry(params) {
        const { sourceEntry, replicationStatus } = params;
        const entry = replicationStatus === 'COMPLETED' ?
              sourceEntry.toCompletedEntry(this.site) :
              sourceEntry.toFailedEntry(this.site);
        const versionId =
              sourceEntry.getReplicationSiteDataStoreVersionId(this.site);
        return entry.setReplicationSiteDataStoreVersionId(this.site,
            versionId);
    }

    /**
     * Publishes the failed notification status
     * into the notification topic
     * @param {Object} sourceEntry the object entry
     * @param {Logger} log the logger instance
     * @param {Function} done callback
     * @return {undefined}
     */
    _publishFailedReplicationStatusNotification(sourceEntry, log, done) {
        const bucket = sourceEntry.getBucket();
        const key = sourceEntry.getObjectKey();
        const value = sourceEntry.getValue();
        const versionId = sourceEntry.getVersionId();
        this.getNotificationConfig(bucket, (err, config) => {
            // we skip if key is a master or if no config is available for
            // the bucket
            if (versionId && config && Object.keys(config).length > 0) {
                const eventType = notifConstants.replicationFailedEvent;
                const ent = {
                    bucket,
                    key,
                    versionId,
                    eventType,
                };
                log.debug('validating entry', {
                    method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                    bucket,
                    key,
                    versionId,
                    eventType,
                });
                // validate and push kafka message to each destination internal topic
                return async.each(this.notificationConfig.destinations,
                    (destination, cb) => {
                        // get destination specific notification config
                        const destBnConf = config.queueConfig.find(
                            c => c.queueArn.split(':').pop()
                            === destination.resource);
                        if (!destBnConf) {
                            // skip, if there is no config for the current
                            // destination resource
                            return cb();
                        }
                        // pass only destination resource specific config to
                        // validate entry
                        const bnConfig = {
                            queueConfig: [destBnConf],
                        };
                        // skip if entry doesn't match config
                        if (!configUtil.validateEntry(bnConfig, ent)) {
                            return cb();
                        }
                        const message
                            = messageUtil.addLogAttributes(value, ent);
                        log.info('publishing replication failed notification', {
                            method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                            bucket,
                            key: message.key,
                            eventType,
                            eventTime: message.dateTime,
                        });
                        const entry = {
                            key: encodeURIComponent(bucket),
                            message: JSON.stringify(message)
                        };
                        return this.notificationProducers[destination.resource].send([entry], err => {
                            if (err) {
                                log.error('error in entry delivery to notification topic', {
                                    method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                                    destination: destination.resource,
                                    entry: sourceEntry.getLogInfo(),
                                    error: err,
                                });
                            }
                            return cb();
                        });
                }, err => {
                    if (err) {
                        log.error('error while pushing replication failed notification', {
                            method: 'ReplicateObject._publishFailedReplicationStatusNotification',
                            entry: sourceEntry.getLogInfo(),
                            error: err,
                        });
                    }
                    return done(null, { committable: false });
                });
            }
            return done(null, { committable: false });
        });
    }

    _publishReplicationStatus(sourceEntry, replicationStatus, params) {
        const { log, reason, kafkaEntry } = params;
        const entryParams = { sourceEntry, replicationStatus };
        const updatedSourceEntry = this._getUpdatedSourceEntry(entryParams);
        const updateData = sourceEntry.getReplicationContent().includes('DATA');
        const kafkaEntries = [updatedSourceEntry.toKafkaEntry(this.site)];
        this.replicationStatusProducer.send(kafkaEntries, err => {
            if (err) {
                log.error('error in entry delivery to replication status topic', {
                    method: 'ReplicateObject._publishReplicationStatus',
                    topic: this.repConfig.replicationStatusTopic,
                    entry: updatedSourceEntry.getLogInfo(),
                    replicationStatus,
                    error: err,
                });
            } else {
                log.info('replication status published', {
                    topic: this.repConfig.replicationStatusTopic,
                    entry: updatedSourceEntry.getLogInfo(),
                    replicationStatus,
                    reason,
                });
                this.metricsHandler.metadataReplicationStatus({ replicationStatus });
                if (updateData) {
                    this.metricsHandler.dataReplicationStatus({ replicationStatus });
                }
            }
            // Commit whether there was an error or not to allow
            // progress of the consumer, as best effort measure when
            // there are errors. We can count on the sweeper to retry
            // entries that failed to be published to kafka (because
            // they will keep their PENDING status).
            if (this.consumer) {
                this.consumer.onEntryCommittable(kafkaEntry);
            }
        });
    }

    _setupRolesOnce(entry, log, cb) {
        log.debug('getting bucket replication',
            { entry: entry.getLogInfo() });
        const entryRolesString = entry.getReplicationRoles();
        let entryRoles;
        if (entryRolesString !== undefined) {
            entryRoles = entryRolesString.split(',');
        }
        if (entryRoles === undefined || entryRoles.length !== 2) {
            log.error('expecting two roles separated by a ' +
                'comma in entry replication configuration',
                {
                    method: 'ReplicateObject._setupRolesOnce',
                    entry: entry.getLogInfo(),
                    roles: entryRolesString,
                });
            return cb(errors.BadRole);
        }
        this.sourceRole = entryRoles[0];
        this.targetRole = entryRoles[1];

        this._setupSourceClients(this.sourceRole, log);

        const req = this.S3source.getBucketReplication(
            { Bucket: entry.getBucket() });
        attachReqUids(req, log);
        return req.send((err, data) => {
            if (err) {
                // eslint-disable-next-line no-param-reassign
                err.origin = 'source';
                log.error('error getting replication ' +
                    'configuration from S3',
                    {
                        method: 'ReplicateObject._setupRolesOnce',
                        entry: entry.getLogInfo(),
                        origin: 'source',
                        peer: this.sourceConfig.s3,
                        error: err.message,
                        httpStatus: err.statusCode,
                    });
                return cb(err);
            }
            const replicationEnabled = (
                data.ReplicationConfiguration.Rules.some(
                    rule => entry.getObjectKey().startsWith(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 two roles separated by a ' +
                    'comma 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);
            }
            if (roles[1] !== 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: roles[1],
                    });
                return cb(errors.BadRole);
            }
            return cb(null, roles[0], roles[1]);
        });
    }

    _setTargetAccountMdOnce(destEntry, targetRole, log, cb) {
        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 };
                            }
                        }
                    }
                    log.error('an error occurred when looking up target ' +
                        'account attributes',
                        {
                            method: 'ReplicateObject._setTargetAccountMdOnce',
                            entry: destEntry.getLogInfo(),
                            origin: 'target',
                            peer,
                            error: err.message,
                        });
                    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();
            });
    }

    _getAndPutData(sourceEntry, destEntry, log, cb) {
        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';
            log.error(errMessage, {
                method: 'ReplicateObject._getAndPutData',
                entry: sourceEntry.getLogInfo(),
            });
            return cb(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')));
            }
        }
        const locations = sourceEntry.getReducedLocations();
        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));
            }
            return cb(null, destLocations);
        });
    }

    _getAndPutPartOnce(sourceEntry, destEntry, part, log, done) {
        const serviceName = this.serviceName;
        const doneOnce = jsutil.once(done);
        const partObj = new ObjectMDLocation(part);
        const partNumber = partObj.getPartNumber();
        const partSize = partObj.getPartSize();
        let destReq = null;
{'fraction_non_alphanumeric': 0.09685920839478711, 'fraction_numerical': 0.0013298193863487996, 'mean_word_length': 1.4993956973652405, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1208004', 'n_tokens_mistral': 9872, 'n_tokens_neox': 9494, 'n_words': 2147}