cleanup 3

Author: SylvainSenechal

bin/ensureServiceUser

@@ -12,7 +12,8 @@ const version = require('../package.json').version;
 const async = require('async');
 const { IAMClient, GetUserCommand, CreateUserCommand, ListPoliciesCommand, CreatePolicyCommand,
         ListAttachedUserPoliciesCommand, AttachUserPolicyCommand, GetRoleCommand, CreateRoleCommand,
-        ListAttachedRolePoliciesCommand, AttachRolePolicyCommand, ListAccessKeysCommand, CreateAccessKeyCommand } = require('@aws-sdk/client-iam');
+        ListAttachedRolePoliciesCommand, AttachRolePolicyCommand, ListAccessKeysCommand, CreateAccessKeyCommand,
+        NoSuchEntityException } = require('@aws-sdk/client-iam');
 const { STSClient, GetCallerIdentityCommand } = require('@aws-sdk/client-sts');
 
 const systemPrefix = '/scality-internal/';
@@ -302,7 +303,7 @@ function collectResource(v, done) {
     v.collect()
         .then(res => done(null, res))
         .catch(err => {
-            if (err.code === 'NoSuchEntity') {
+            if (err instanceof NoSuchEntityException) {
                 return done(null, null);
             }
             done(err);

extensions/replication/tasks/CopyLocationTask.js

@@ -13,7 +13,8 @@ const {
     MultipleBackendInitiateMPUCommand,
     MultipleBackendPutMPUPartCommand,
     MultipleBackendCompleteMPUCommand,
-    MultipleBackendAbortMPUCommand
+    MultipleBackendAbortMPUCommand,
+    addContentLengthMiddleware,
 } = require('@scality/cloudserverclient');
 const { LifecycleMetrics } = require('../../lifecycle/LifecycleMetrics');
 const ReplicationMetric = require('../ReplicationMetric');
@@ -223,6 +224,54 @@ class CopyLocationTask extends BackbeatTask {
         log.debug('getting object data', actionEntry.getLogInfo());
         const doneOnce = jsutil.once(done);
         const size = objMD.getContentLength();
+        
+        // Use abort controller to cancel requests on error
+        const abortController = new AbortController();
+        let sourceStreamAborted = false;
+        
+        const performPutObject = incomingMsg => {
+            log.debug('putting data', actionEntry.getLogInfo());
+            
+            // Set up stream error handler if we have a stream
+            if (incomingMsg) {
+                incomingMsg.on('error', err => {
+                    if (!sourceStreamAborted) {
+                        if (err.$metadata?.httpStatusCode === 404) {
+                            log.error('the source object was not found', Object.assign({
+                                method: 'CopyLocationTask._getAndPutObjectOnce',
+                                peer: this.sourceConfig.s3,
+                                error: err.message,
+                                httpStatus: err.$metadata?.httpStatusCode,
+                            }, actionEntry.getLogInfo()));
+                            return doneOnce(errors.ObjNotFound);
+                        }
+                        log.error('an error occurred when streaming data from S3',
+                                  Object.assign({
+                                      method: 'CopyLocationTask._getAndPutObjectOnce',
+                                      peer: this.sourceConfig.s3,
+                                      error: err.message,
+                                  }, actionEntry.getLogInfo()));
+                        return doneOnce(err);
+                    }
+                    return undefined;
+                });
+            }
+            
+            const putDone = err => {
+                if (err && incomingMsg && !sourceStreamAborted) {
+                    // Abort the source stream on PUT error
+                    sourceStreamAborted = true;
+                    if (incomingMsg.destroy) {
+                        incomingMsg.destroy();
+                    }
+                }
+                return doneOnce(err);
+            };
+            
+            return this._sendMultipleBackendPutObject(
+                actionEntry, objMD, size, incomingMsg, log, putDone);
+        };
+        
         if (size !== 0) {
             const { bucket, key, version } = actionEntry.getAttribute('target');
             const getObjectCommand = new GetObjectCommand({
@@ -233,13 +282,12 @@ class CopyLocationTask extends BackbeatTask {
                 RequestUids: log.getSerializedUids(),
             });
 
-            return this.backbeatClient.send(getObjectCommand)
-                .then(response => {
-                    log.debug('putting data', actionEntry.getLogInfo());
-                    return this._sendMultipleBackendPutObject(
-                        actionEntry, objMD, size, response.Body, log, doneOnce);
-                })
+            return this.backbeatClient.send(getObjectCommand, { abortSignal: abortController.signal })
+                .then(response => performPutObject(response.Body))
                 .catch(err => {
+                    if (!sourceStreamAborted) {
+                        sourceStreamAborted = true;
+                    }
                     if (err.$metadata?.httpStatusCode === 404) {
                         log.error('the source object was not found', Object.assign({
                             method: 'CopyLocationTask._getAndPutObjectOnce',
@@ -259,9 +307,7 @@ class CopyLocationTask extends BackbeatTask {
                     return doneOnce(err);
                 });
         }
-        log.debug('putting data', actionEntry.getLogInfo());
-        return this._sendMultipleBackendPutObject(
-            actionEntry, objMD, size, null, log, doneOnce);
+        return performPutObject(undefined);
     }
 
     /**
@@ -295,6 +341,7 @@ class CopyLocationTask extends BackbeatTask {
             Body: incomingMsg,
             RequestUids: log.getSerializedUids(),
         });
+        addContentLengthMiddleware(command, size);
 
         return this.backbeatClient.send(command)
             .then(data => {
@@ -362,6 +409,9 @@ class CopyLocationTask extends BackbeatTask {
         }, actionEntry.getLogInfo()));
         // A 0-byte object has no range, otherwise range is inclusive.
         const size = range ? range.end - range.start + 1 : 0;
+        // Create AbortController even for 0-byte parts for consistency
+        const abortController = new AbortController();
+        
         const { bucket, key, version } = actionEntry.getAttribute('target');
         if (size !== 0) {
             const getObjectCommand = new GetObjectCommand({
@@ -373,9 +423,9 @@ class CopyLocationTask extends BackbeatTask {
                 RequestUids: log.getSerializedUids(),
             });
 
-            return this.backbeatClient.send(getObjectCommand)
+            return this.backbeatClient.send(getObjectCommand, { abortSignal: abortController.signal })
                 .then(response => this._putMPUPart(actionEntry, objMD, response.Body, size,
-                        uploadId, partNumber, log, done))
+                        uploadId, partNumber, log, abortController, done))
                 .catch(err => {
                     if (err.$metadata?.httpStatusCode === 404) {
                         return done(err);
@@ -389,8 +439,9 @@ class CopyLocationTask extends BackbeatTask {
                     return done(err);
                 });
         }
-        return this._putMPUPart(actionEntry, objMD, null, size,
-            uploadId, partNumber, log, done);
+        // For 0-byte parts, pass undefined body but still provide abortController
+        return this._putMPUPart(actionEntry, objMD, undefined, size,
+            uploadId, partNumber, log, abortController, done);
     }
 
     /**
@@ -513,45 +564,94 @@ class CopyLocationTask extends BackbeatTask {
      * @param {String} uploadId - The upload ID of the initiated MPU
      * @param {Number} partNumber - The part number of the part
      * @param {Werelogs} log - The logger instance
+     * @param {AbortController} abortController - The abort controller for the source GET request
      * @param {Function} cb - The callback to call
      * @return {undefined}
      */
     _putMPUPart(actionEntry, objMD, incomingMsg, size, uploadId, partNumber,
-                log, cb) {
+                log, abortController, cb) {
         const doneOnce = jsutil.once(cb);
-        log.debug('putting data', actionEntry.getLogInfo());
+        let sourceStreamAborted = false;
+        let destRequestAborted = false;
 
-        const { bucket, key } = actionEntry.getAttribute('target');
-        const command = new MultipleBackendPutMPUPartCommand({
-            Bucket: bucket,
-            Key: key,
-            ContentLength: size,
-            StorageType: this.destType,
-            StorageClass: this.site,
-            PartNumber: partNumber,
-            UploadId: uploadId,
-            Body: incomingMsg,
-            RequestUids: log.getSerializedUids(),
-        });
-        
-        return this.backbeatClient.send(command)
-            .then(data => {
-                this._replicationMetric
-                    .withEntry(actionEntry)
-                    .withMetricType(metricsTypeCompleted)
-                    .withObjectSize(size)
-                    .publish();
-                return doneOnce(null, data);
-            })
-            .catch(err => {
-                log.error('an error occurred on putting MPU part to S3',
-                Object.assign({
-                    method: 'CopyLocationTask._putMPUPart',
-                    error: err.message,
-                    httpStatus: err.$metadata?.httpStatusCode,
-                }, actionEntry.getLogInfo()));
-                return doneOnce(err);
+        const performPutMPUPart = () => {
+            if (incomingMsg) {
+                log.debug('putting data', actionEntry.getLogInfo());
+            }
+
+            const { bucket, key } = actionEntry.getAttribute('target');
+            const command = new MultipleBackendPutMPUPartCommand({
+                Bucket: bucket,
+                Key: key,
+                ContentLength: size,
+                StorageType: this.destType,
+                StorageClass: this.site,
+                PartNumber: partNumber,
+                UploadId: uploadId,
+                Body: incomingMsg,
+                RequestUids: log.getSerializedUids(),
             });
+            addContentLengthMiddleware(command, size);
+
+            return this.backbeatClient.send(command)
+                .then(data => {
+                    this._replicationMetric
+                        .withEntry(actionEntry)
+                        .withMetricType(metricsTypeCompleted)
+                        .withObjectSize(size)
+                        .publish();
+                    return doneOnce(null, data);
+                })
+                .catch(err => {
+                    if (!destRequestAborted) {
+                        destRequestAborted = true;
+                        if (abortController) {
+                            abortController.abort();
+                            sourceStreamAborted = true;
+                        }
+                        if (incomingMsg && incomingMsg.destroy) {
+                            incomingMsg.destroy();
+                        }
+                        log.error('an error occurred on putting MPU part to S3',
+                        Object.assign({
+                            method: 'CopyLocationTask._putMPUPart',
+                            error: err.message,
+                            httpStatus: err.$metadata?.httpStatusCode,
+                        }, actionEntry.getLogInfo()));
+                    }
+                    return doneOnce(err);
+                });
+        };
+
+        if (incomingMsg) {
+            incomingMsg.on('error', err => {
+                if (!sourceStreamAborted) {
+                    sourceStreamAborted = true;
+                    if (abortController) {
+                        abortController.abort();
+                    }
+                    destRequestAborted = true;
+                    if (err.$metadata?.httpStatusCode === 404) {
+                        log.error('source object not found when streaming MPU part',
+                        Object.assign({
+                            method: 'CopyLocationTask._putMPUPart',
+                            error: err.message,
+                            httpStatus: err.$metadata?.httpStatusCode,
+                        }, actionEntry.getLogInfo()));
+                        return doneOnce(errors.ObjNotFound);
+                    }
+                    log.error('an error occurred when streaming MPU part data from S3',
+                    Object.assign({
+                        method: 'CopyLocationTask._putMPUPart',
+                        error: err.message,
+                    }, actionEntry.getLogInfo()));
+                    return doneOnce(err);
+                }
+                return undefined;
+            });
+        }
+
+        return performPutMPUPart();
     }
 
     _getAndPutMultipartUpload(actionEntry, objMD, log, cb) {