Move uploadId listing and adapt/add tests

Issue: CLDSRV-669

Author: williamlardier

lib/api/apiUtils/object/abortMultipartUpload.js

@@ -3,12 +3,10 @@ const async = require('async');
 const constants = require('../../../../constants');
 const { data } = require('../../../data/wrapper');
 const locationConstraintCheck = require('../object/locationConstraintCheck');
-const { standardMetadataValidateBucketAndObj } =
-    require('../../../metadata/metadataUtils');
+const metadataUtils = require('../../../metadata/metadataUtils');
 const { validateQuotas } = require('../quotas/quotaUtils');
 const services = require('../../../services');
 const metadata = require('../../../metadata/wrapper');
-const { versioning } = require('arsenal');
 
 function abortMultipartUpload(authInfo, bucketName, objectKey, uploadId, log,
     callback, request) {
@@ -31,7 +29,8 @@ function abortMultipartUpload(authInfo, bucketName, objectKey, uploadId, log,
 
     async.waterfall([
         function checkDestBucketVal(next) {
-            standardMetadataValidateBucketAndObj(metadataValParams, authzIdentityResult, log,
+            // FIX: Call the function from the imported module object.
+            metadataUtils.standardMetadataValidateBucketAndObj(metadataValParams, authzIdentityResult, log,
                 (err, destinationBucket, objectMD) => {
                     if (err) {
                         log.error('error validating request', { error: err });
@@ -105,9 +104,13 @@ function abortMultipartUpload(authInfo, bucketName, objectKey, uploadId, log,
         // supported by Metadata.
         function ensureCleanupIsRequired(mpuBucket, storedParts, destBucket,
             objectMD, skipDataDelete, next) {
+            if (!objectMD) {
+                return next(null, mpuBucket, storedParts, destBucket, null, skipDataDelete);
+            }
+
             // If objectMD exists and has matching uploadId, use it directly
             // This handles all non-versioned cases, and some versioned cases.
-            if (objectMD && objectMD.uploadId === uploadId) {
+            if (objectMD.uploadId === uploadId) {
                 return next(null, mpuBucket, storedParts, destBucket, objectMD, skipDataDelete);
             }
 
@@ -117,64 +120,21 @@ function abortMultipartUpload(authInfo, bucketName, objectKey, uploadId, log,
                 return next(null, mpuBucket, storedParts, destBucket, null, skipDataDelete);
             }
 
-            // Otherwise, we list all versions of the object. We try to stop as early
-            // as possible, by using pagination.
-            let keyMarker = null;
-            let versionIdMarker = null;
-            let foundVersion = null;
-            let shouldContinue = true;
-
-            return async.whilst(
-                () => shouldContinue && !foundVersion,
-                callback => {
-                    const listParams = {
-                        listingType: 'DelimiterVersions',
-                        // To only list the specific key, we need to add the versionId separator
-                        prefix: `${objectKey}${versioning.VersioningConstants.VersionId.Separator}`,
-                        maxKeys: 1000,
-                    };
-
-                    if (keyMarker) {
-                        listParams.keyMarker = keyMarker;
-                    }
-                    if (versionIdMarker) {
-                        listParams.versionIdMarker = versionIdMarker;
-                    }
-
-                    return services.getObjectListing(bucketName, listParams, log, (err, listResponse) => {
-                        if (err) {
-                            log.error('error listing object versions', { error: err });
-                            return callback(err);
-                        }
-
-                        // Check each version in current batch for matching uploadId
-                        const matchedVersion = (listResponse.Versions || []).find(version =>
-                            version.key === objectKey &&
-                            version.value &&
-                            version.value.uploadId === uploadId
-                        );
-
-                        if (matchedVersion) {
-                            foundVersion = matchedVersion.value;
-                        }
-
-                        // Set up for next iteration if needed
-                        if (listResponse.IsTruncated && !foundVersion) {
-                            keyMarker = listResponse.NextKeyMarker;
-                            versionIdMarker = listResponse.NextVersionIdMarker;
-                        } else {
-                            shouldContinue = false;
-                        }
-
-                        return callback();
+            // Otherwise, list all versions to find one with a matching uploadId.
+            return services.findObjectVersionByUploadId(bucketName, objectKey, uploadId, log, (err, foundVersion) => {
+                if (err) {
+                    log.error('error finding object version by uploadId, proceeding without cleanup', {
+                        error: err,
+                        method: 'abortMultipartUpload.ensureCleanupIsRequired',
                     });
-                },
-                err => next(null, mpuBucket, storedParts, destBucket, err ? null : foundVersion, skipDataDelete)
-            );
+                    // On error, continue the abort without an objectMD to clean up.
+                    return next(null, mpuBucket, storedParts, destBucket, null, skipDataDelete);
+                }
+                return next(null, mpuBucket, storedParts, destBucket, foundVersion, skipDataDelete);
+            });
         },
         function deleteObjectMetadata(mpuBucket, storedParts, destBucket, objMDWithMatchingUploadID,
             skipDataDelete, next) {
-            // If no objMDWithMatchingUploadID, nothing to delete
             if (!objMDWithMatchingUploadID) {
                 return next(null, mpuBucket, storedParts, destBucket, objMDWithMatchingUploadID, skipDataDelete);
             }

lib/services.js

@@ -6,6 +6,7 @@ const { errors, s3middleware } = require('arsenal');
 const ObjectMD = require('arsenal').models.ObjectMD;
 const BucketInfo = require('arsenal').models.BucketInfo;
 const ObjectMDArchive = require('arsenal').models.ObjectMDArchive;
+const { versioning } = require('arsenal');
 const acl = require('./metadata/acl');
 const constants = require('../constants');
 const { config } = require('./Config');
@@ -418,6 +419,71 @@ const services = {
             });
     },
 
+    /**
+     * Finds a specific object version by its uploadId by listing and filtering all versions.
+     * This is used during MPU abort to clean up potentially orphaned object metadata.
+     * @param {string} bucketName - The name of the bucket.
+     * @param {string} objectKey - The key of the object.
+     * @param {string} uploadId - The uploadId to search for.
+     * @param {object} log - The logger instance.
+     * @param {function} cb - The callback, called with (err, foundVersion).
+     * @returns {undefined}
+     */
+    findObjectVersionByUploadId(bucketName, objectKey, uploadId, log, cb) {
+        let keyMarker = null;
+        let versionIdMarker = null;
+        let foundVersion = null;
+        let shouldContinue = true;
+
+        return async.whilst(
+            () => shouldContinue && !foundVersion,
+            callback => {
+                const listParams = {
+                    listingType: 'DelimiterVersions',
+                    // To only list the specific key, we need to add the versionId separator
+                    prefix: `${objectKey}${versioning.VersioningConstants.VersionId.Separator}`,
+                    maxKeys: 1000,
+                };
+
+                if (keyMarker) {
+                    listParams.keyMarker = keyMarker;
+                }
+                if (versionIdMarker) {
+                    listParams.versionIdMarker = versionIdMarker;
+                }
+
+                return this.getObjectListing(bucketName, listParams, log, (err, listResponse) => {
+                    if (err) {
+                        log.error('error listing object versions', { error: err });
+                        return callback(err);
+                    }
+
+                    // Check each version in current batch for matching uploadId
+                    const matchedVersion = (listResponse.Versions || []).find(version =>
+                        version.key === objectKey &&
+                        version.value &&
+                        version.value.uploadId === uploadId
+                    );
+
+                    if (matchedVersion) {
+                        foundVersion = matchedVersion.value;
+                    }
+
+                    // Set up for next iteration if needed
+                    if (listResponse.IsTruncated && !foundVersion) {
+                        keyMarker = listResponse.NextKeyMarker;
+                        versionIdMarker = listResponse.NextVersionIdMarker;
+                    } else {
+                        shouldContinue = false;
+                    }
+
+                    return callback();
+                });
+            },
+            err => cb(err, err ? null : foundVersion)
+        );
+    },
+
     /**
      * Gets list of objects ready to be lifecycled
      * @param {object} bucketName - bucket in which objectMetadata is stored