CLDSRV-632 ✨ put metadata edge cases

lib/api/apiUtils/object/createAndStoreObject.js

@@ -218,6 +218,7 @@ function createAndStoreObject(bucketName, bucketMD, objectKey, objMD, authInfo,
                     metadataStoreParams.contentMD5 = constants.emptyFileMd5;
                     return next(null, null, null);
                 }
+
                 // Handle mdOnlyHeader as a metadata only operation. If
                 // the object in question is actually 0 byte or has a body size
                 // then handle normally.
@@ -244,6 +245,7 @@ function createAndStoreObject(bucketName, bucketMD, objectKey, objMD, authInfo,
                     return next(null, dataGetInfo, _md5);
                 }
             }
+
             return dataStore(objectKeyContext, cipherBundle, request, size,
                     streamingV4Params, backendInfo, log, next);
         },
@@ -280,10 +282,12 @@ function createAndStoreObject(bucketName, bucketMD, objectKey, objMD, authInfo,
                 const options = overwritingVersioning(objMD, metadataStoreParams);
                 return process.nextTick(() => next(null, options, infoArr));
             }
+
             if (!bucketMD.isVersioningEnabled() && objMD?.archive?.archiveInfo) {
                 // Ensure we trigger a "delete" event in the oplog for the previously archived object
                 metadataStoreParams.needOplogUpdate = 's3:ReplaceArchivedObject';
             }
+
             return versioningPreprocessing(bucketName, bucketMD,
                 metadataStoreParams.objectKey, objMD, log, (err, options) => {
                     if (err) {
@@ -316,9 +320,11 @@ function createAndStoreObject(bucketName, bucketMD, objectKey, objMD, authInfo,
             metadataStoreParams.versioning = options.versioning;
             metadataStoreParams.isNull = options.isNull;
             metadataStoreParams.deleteNullKey = options.deleteNullKey;
+
             if (options.extraMD) {
                 Object.assign(metadataStoreParams, options.extraMD);
             }
+
             return _storeInMDandDeleteData(bucketName, infoArr,
                 cipherBundle, metadataStoreParams,
                 options.dataToDelete, log, requestMethod, next);

lib/api/apiUtils/object/versioning.js

@@ -103,6 +103,7 @@ function _storeNullVersionMD(bucketName, objKey, nullVersionId, objMD, log, cb)
             log.debug('error from metadata storing null version as new version',
             { error: err });
         }
+
         cb(err);
     });
 }
@@ -252,6 +253,7 @@ function processVersioningState(mst, vstat, nullVersionCompatMode) {
             }
             return { options, nullVersionId };
         }
+
         if (mst.isNull && !mst.isNull2) {
             // if master null version was put with an older
             // Cloudserver (or in compat mode), there is a
@@ -265,6 +267,7 @@ function processVersioningState(mst, vstat, nullVersionCompatMode) {
         }
         return { options, nullVersionId };
     }
+
     // backward-compat: keep a reference to the existing null
     // versioned key
     if (mst.nullVersionId) {
@@ -295,6 +298,7 @@ function getMasterState(objMD) {
     if (!objMD) {
         return {};
     }
+
     const mst = {
         exists: true,
         versionId: objMD.versionId,
@@ -304,10 +308,12 @@ function getMasterState(objMD) {
         nullVersionId: objMD.nullVersionId,
         nullUploadId: objMD.nullUploadId,
     };
+
     if (objMD.location) {
         mst.objLocation = Array.isArray(objMD.location) ?
             objMD.location : [objMD.location];
     }
+
     return mst;
 }
 /** versioningPreprocessing - return versioning information for S3 to handle
@@ -329,19 +335,22 @@ function versioningPreprocessing(bucketName, bucketMD, objectKey, objMD,
     log, callback) {
     const mst = getMasterState(objMD);
     const vCfg = bucketMD.getVersioningConfiguration();
-    // bucket is not versioning configured
+
     if (!vCfg) {
         const options = { dataToDelete: mst.objLocation };
         return process.nextTick(callback, null, options);
     }
-    // bucket is versioning configured
+
     const { options, nullVersionId, delOptions } =
           processVersioningState(mst, vCfg.Status, config.nullVersionCompatMode);
+
     return async.series([
         function storeNullVersionMD(next) {
             if (!nullVersionId) {
                 return process.nextTick(next);
             }
+
+            options.nullVersionId = nullVersionId;
             return _storeNullVersionMD(bucketName, objectKey, nullVersionId, objMD, log, next);
         },
         function prepareNullVersionDeletion(next) {

lib/routes/routeBackbeat.js

@@ -40,6 +40,13 @@ const { listLifecycleOrphanDeleteMarkers } = require('../api/backbeat/listLifecy
 const { objectDeleteInternal } = require('../api/objectDelete');
 const quotaUtils = require('../api/apiUtils/quotas/quotaUtils');
 const { handleAuthorizationResults } = require('../api/api');
+const { versioningPreprocessing }
+    = require('../api/apiUtils/object/versioning');
+const {promisify} = require('util');
+
+const versioningPreprocessingPromised = promisify(versioningPreprocessing);
+metadata.getObjectMDPromised = promisify(metadata.getObjectMD);
+metadata.getBucketAndObjectMDPromised = promisify(metadata.getBucketAndObjectMD);
 
 const { CURRENT_TYPE, NON_CURRENT_TYPE, ORPHAN_DM_TYPE } = constants.lifecycleListing;
 
@@ -508,23 +515,22 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
         if (err) {
             return callback(err);
         }
+
         let omVal;
+
         try {
             omVal = JSON.parse(payload);
         } catch {
-            // FIXME: add error type MalformedJSON
             return callback(errors.MalformedPOSTRequest);
         }
+
         const { headers, bucketName, objectKey } = request;
-        // check if it's metadata only operation
+
         if (headers['x-scal-replication-content'] === 'METADATA') {
             if (!objMd) {
-                // if the target does not exist, return an error to
-                // backbeat, who will have to retry the operation as a
-                // complete replication
                 return callback(errors.ObjNotFound);
             }
-            // use original data locations and encryption info
+
             [
                 'location',
                 'x-amz-server-side-encryption',
@@ -611,6 +617,16 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
         // To prevent this, the versionId field is only included in options when it is defined.
         if (versionId !== undefined) {
             options.versionId = versionId;
+            omVal.versionId = versionId;
+
+            if (isNull) {
+                if (!nullVersionCompatMode) {
+                  omVal.isNull2 = true;
+                }
+
+                omVal.isNull = isNull;
+            }
+
             // In the MongoDB metadata backend, setting the versionId option leads to the creation
             // or update of the version object, the master object is only updated if its versionId
             // is the same as the version. This can lead to inconsistencies when replicating objects
@@ -641,6 +657,7 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
                 if (!request.query?.accountId) {
                     return next();
                 }
+
                 return getCanonicalIdsByAccountId(request.query.accountId, log, (err, res) => {
                     if (err) {
                         return next(err);
@@ -650,6 +667,45 @@ function putMetadata(request, response, bucketInfo, objMd, log, callback) {
                     return next();
                 });
             },
+            async () => {
+                // If we create a new version of an object (so objMd is null),
+                // we should make sure that the masterVersion is versionned.
+                // If an object already exists, we just want to update the metadata
+                // of the existing object and not create a new one
+                if (versioning && !objMd) {
+                    let masterMD;
+
+                    try {
+                        masterMD = await metadata.getObjectMDPromised(bucketName, objectKey, {}, log);
+                    } catch (err) {
+                        if (err.is?.NoSuchKey) {
+                            log.debug('no master found for versioned object', {
+                                method: 'putMetadata',
+                                bucketName,
+                                objectKey,
+                            });
+                        } else {
+                            throw err;
+                        }
+                    }
+
+                    if (!masterMD) {
+                        return;
+                    }
+
+                    const versioningPreprocessingResult =
+                        await versioningPreprocessingPromised(bucketName, bucketInfo, objectKey, masterMD, log);
+
+                    if (versioningPreprocessingResult?.nullVersionId) {
+                        omVal.nullVersionId = versioningPreprocessingResult.nullVersionId;
+                        options.deleteNullKey = versioningPreprocessingResult.deleteNullKey;
+
+                        if (versioningPreprocessingResult.extraMD) {
+                            Object.assign(omVal, options.extraMD);
+                        }
+                    }
+                }
+            },
             next => {
                 log.trace('putting object version', {
                     objectKey: request.objectKey, omVal, options });