Fail with UpdateConflict on concurrent metadata update

Issue: ARSN-8449

lib/errors/arsenalErrors.ts

@@ -1024,6 +1024,13 @@ export const TooManyRequests: ErrorFormat = {
     code: 429,
 };
 
+export const UpdateConflict: ErrorFormat = {
+    code: 500,
+
+    description:
+        'The request was rejected because there was a conflict attempting to update a resource.'
+};
+
 // --------------------- cdmiclient ---------------------
 
 export const ReadOnly: ErrorFormat = {

lib/models/ObjectMD.ts

@@ -89,6 +89,10 @@ export type ObjectMDData = {
     // the master is set as a placeholder and gets updated with the new latest
     // version data after a certain amount of time.
     isPHD: boolean;
+    // Count the number of 'updates' to the object, incremented every time the
+    // metadata "document" is updated. This is used by backend to detect
+    // conflicting updates to the same object.
+    revisionCount?: number
 };
 
 /**

lib/storage/metadata/mongoclient/MongoClientInterface.js

@@ -51,6 +51,7 @@ const SOCKET_TIMEOUT_MS = 360000;
 const CONCURRENT_CURSORS = 10;
 
 const initialInstanceID = process.env.INITIAL_INSTANCE_ID;
+const METADATA_CONFLICT_DETECTION = process.env.METADATA_CONFLICT_DETECTION;
 
 let uidCounter = 0;
 
@@ -808,6 +809,30 @@ class MongoClientInterface {
         });
     }
 
+    _updateCondition(key, isVersion, objVal) {
+        const cond = { _id: key };
+        if (!METADATA_CONFLICT_DETECTION) {
+            return cond;
+        }
+
+        if (objVal.revisionCount) {
+            cond['value.revisionCount'] = objVal.revisionCount;
+        } else if (isVersion) {
+            // This is the case where the document was created with an older version of Arsenal,
+            // and which did not have the revisionCount protection
+
+            // We don't want to do this for master object, which get overwritten in non-versioned
+            // buckets without following the read object MD / update / write object MD pattern
+            cond['value.revisionCount'] = { $exists: false };
+        }
+
+        /* eslint-disable no-param-reassign */
+        objVal.revisionCount = (objVal.revisionCount || 0) + 1;
+        /* eslint-enable no-param-reassign */
+
+        return cond;
+    }
+
     /**
      * In this case the caller provides a versionId. We assume that
      * objVal already contains the destination versionId. We first
@@ -831,9 +856,7 @@ class MongoClientInterface {
     putObjectVerCase4(c, bucketName, objName, objVal, params, log, cb) {
         const versionKey = formatVersionKey(objName, params.versionId, params.vFormat);
         const masterKey = formatMasterKey(objName, params.vFormat);
-        c.updateOne({
+        c.updateOne(this._updateCondition(versionKey, true, objVal), {
-            _id: versionKey,
-        }, {
             $set: {
                 _id: versionKey,
                 value: objVal,
@@ -893,6 +916,12 @@ class MongoClientInterface {
                 return cb(errors.InternalError);
             });
         })).catch(err => {
+            if (err.code === 11000) { // E11000 duplicate key error
+                log.error('putObjectVerCase4: conflict upserting object version',
+                    { error: err.message });
+                return cb(errors.UpdateConflict);
+            }
+
             log.error(
                 'putObjectVerCase4: error upserting object version',
                 { error: err.message });
@@ -913,16 +942,20 @@ class MongoClientInterface {
      */
     putObjectNoVer(c, bucketName, objName, objVal, params, log, cb) {
         const masterKey = formatMasterKey(objName, params.vFormat);
-        c.updateOne({
+        c.updateOne(this._updateCondition(masterKey, false, objVal), {
-            _id: masterKey,
-        }, {
             $set: {
                 _id: masterKey,
                 value: objVal,
             },
         }, {
             upsert: true,
-        }).then(() => cb()).catch((err) => {
+        }).then(cb).catch((err) => {
+            if (err.code === 11000) { // E11000 duplicate key error
+                log.error('putObjectVerCase4: conflict upserting object version',
+                    { error: err.message });
+                return cb(errors.UpdateConflict);
+            }
+
             log.error('putObjectNoVer: error putting obect with no versioning', { error: err.message });
             return cb(errors.InternalError);
         });