rgw/multipart: use cls_version to avoid racing between part upload and multipart complete

Signed-off-by: Jane Zhu <jzhu116@bloomberg.net>

13 files changed, 282 insertions, 83 deletions

diff --git a/src/rgw/driver/daos/rgw_sal_daos.cc b/src/rgw/driver/daos/rgw_sal_daos.cc
index 0558dbf461e..cf6820a9111 100644
--- a/src/rgw/driver/daos/rgw_sal_daos.cc
+++ b/src/rgw/driver/daos/rgw_sal_daos.cc
@@ -1196,7 +1196,8 @@ int DaosObject::DaosDeleteOp::delete_obj(const DoutPrefixProvider* dpp,
 }
 
 int DaosObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y,
-                              uint32_t flags) {
+                              uint32_t flags, std::list<rgw_obj_index_key>* remove_objs,
+                              RGWObjVersionTracker* objv) {
   ldpp_dout(dpp, 20) << "DEBUG: delete_object" << dendl;
   DaosObject::DaosDeleteOp del_op(this);
   del_op.params.bucket_owner = bucket->get_info().owner;
@@ -1677,7 +1678,8 @@ int DaosMultipartUpload::complete(
     map<int, string>& part_etags, list<rgw_obj_index_key>& remove_objs,
     uint64_t& accounted_size, bool& compressed, RGWCompressionInfo& cs_info,
     off_t& off, std::string& tag, ACLOwner& owner, uint64_t olh_epoch,
-    rgw::sal::Object* target_obj) {
+    rgw::sal::Object* target_obj,
+    prefix_map_t& processed_prefixes) {
   ldpp_dout(dpp, 20) << "DEBUG: complete" << dendl;
   char final_etag[CEPH_CRYPTO_MD5_DIGESTSIZE];
   char final_etag_str[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 16];
@@ -1922,6 +1924,15 @@ int DaosMultipartUpload::complete(
   return ret;
 }
 
+int DaosMultipartUpload::cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+    CephContext *cct, optional_yield y,
+    const rgw_obj& obj,
+    std::list<rgw_obj_index_key>& remove_objs,
+    prefix_map_t& processed_prefixes)
+{
+  return -ENOTSUP;
+}
+
 int DaosMultipartUpload::get_info(const DoutPrefixProvider* dpp,
                                   optional_yield y, rgw_placement_rule** rule,
                                   rgw::sal::Attrs* attrs) {
diff --git a/src/rgw/driver/daos/rgw_sal_daos.h b/src/rgw/driver/daos/rgw_sal_daos.h
index 8cff96088ee..7cc20260227 100644
--- a/src/rgw/driver/daos/rgw_sal_daos.h
+++ b/src/rgw/driver/daos/rgw_sal_daos.h
@@ -596,7 +596,8 @@ class DaosObject : public StoreObject {
   virtual ~DaosObject();
 
   virtual int delete_object(const DoutPrefixProvider* dpp, optional_yield y,
-                            uint32_t flags) override;
+                            uint32_t flags, std::list<rgw_obj_index_key>* remove_objs,
+                            RGWObjVersionTracker* objv) override;
   virtual int copy_object(
       const ACLOwner& owner, const rgw_user& remote_user,
       req_info* info, const rgw_zone_id& source_zone,
@@ -861,7 +862,13 @@ class DaosMultipartUpload : public StoreMultipartUpload {
                        uint64_t& accounted_size, bool& compressed,
                        RGWCompressionInfo& cs_info, off_t& off,
                        std::string& tag, ACLOwner& owner, uint64_t olh_epoch,
-                       rgw::sal::Object* target_obj) override;
+                       rgw::sal::Object* target_obj,
+                       prefix_map_t& processed_prefixes) override;
+  virtual int cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+      CephContext *cct, optional_yield y,
+      const rgw_obj& obj,
+      std::list<rgw_obj_index_key>& remove_objs,
+      prefix_map_t& processed_prefixes) override;
   virtual int get_info(const DoutPrefixProvider* dpp, optional_yield y,
                        rgw_placement_rule** rule,
                        rgw::sal::Attrs* attrs = nullptr) override;
diff --git a/src/rgw/driver/motr/rgw_sal_motr.cc b/src/rgw/driver/motr/rgw_sal_motr.cc
index bb3af3a63a1..b999673ac18 100644
--- a/src/rgw/driver/motr/rgw_sal_motr.cc
+++ b/src/rgw/driver/motr/rgw_sal_motr.cc
@@ -585,7 +585,7 @@ int MotrBucket::remove(const DoutPrefixProvider *dpp, bool delete_children, opti
 
       std::unique_ptr<rgw::sal::Object> object = get_object(key);
 
-      ret = object->delete_object(dpp, null_yield, rgw::sal::FLAG_LOG_OP);
+      ret = object->delete_object(dpp, null_yield, rgw::sal::FLAG_LOG_OP, nullptr, nullptr);
       if (ret < 0 && ret != -ENOENT) {
         ldpp_dout(dpp, 0) << "ERROR: remove_bucket rgw_remove_object failed rc=" << ret << dendl;
 	      return ret;
@@ -1502,7 +1502,11 @@ int MotrObject::MotrDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional
   return 0;
 }
 
-int MotrObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags)
+int MotrObject::delete_object(const DoutPrefixProvider* dpp,
+    optional_yield y,
+    uint32_t flags,
+    std::list<rgw_obj_index_key>* remove_objs,
+    RGWObjVersionTracker* objv)
 {
   MotrObject::MotrDeleteOp del_op(this);
   del_op.params.bucket_owner = bucket->get_info().owner;
@@ -2668,7 +2672,8 @@ int MotrMultipartUpload::complete(const DoutPrefixProvider *dpp,
 				   RGWCompressionInfo& cs_info, off_t& off,
 				   std::string& tag, ACLOwner& owner,
 				   uint64_t olh_epoch,
-				   rgw::sal::Object* target_obj)
+				   rgw::sal::Object* target_obj,
+				   prefix_map_t& processed_prefixes)
 {
   char final_etag[CEPH_CRYPTO_MD5_DIGESTSIZE];
   char final_etag_str[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 16];
@@ -2878,6 +2883,15 @@ int MotrMultipartUpload::complete(const DoutPrefixProvider *dpp,
                                   M0_IC_DEL, meta_obj->get_key().get_oid(), bl);
 }
 
+int MotrMultipartUpload::cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+    CephContext *cct, optional_yield y,
+    const rgw_obj& obj,
+    std::list<rgw_obj_index_key>& remove_objs,
+    prefix_map_t& processed_prefixes)
+{
+  return -ENOTSUP;
+}
+
 int MotrMultipartUpload::get_info(const DoutPrefixProvider *dpp, optional_yield y, rgw_placement_rule** rule, rgw::sal::Attrs* attrs)
 {
   if (!rule && !attrs) {
diff --git a/src/rgw/driver/motr/rgw_sal_motr.h b/src/rgw/driver/motr/rgw_sal_motr.h
index 4d9c189993b..f92074b9d94 100644
--- a/src/rgw/driver/motr/rgw_sal_motr.h
+++ b/src/rgw/driver/motr/rgw_sal_motr.h
@@ -657,7 +657,9 @@ class MotrObject : public StoreObject {
 
     virtual int delete_object(const DoutPrefixProvider* dpp,
         optional_yield y,
-        uint32_t flags) override;
+        uint32_t flags,
+        td::list<rgw_obj_index_key>* remove_objs,
+        GWObjVersionTracker* objv) override;
     virtual int copy_object(const ACLOwner& owner,
         const rgw_user& remote_user,
         req_info* info, const rgw_zone_id& source_zone,
@@ -933,7 +935,13 @@ public:
 		       RGWCompressionInfo& cs_info, off_t& off,
 		       std::string& tag, ACLOwner& owner,
 		       uint64_t olh_epoch,
-		       rgw::sal::Object* target_obj) override;
+		       rgw::sal::Object* target_obj,
+		       prefix_map_t& processed_prefixes) override;
+  virtual int cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+           CephContext *cct, optional_yield y,
+           const rgw_obj& obj,
+           std::list<rgw_obj_index_key>& remove_objs,
+           prefix_map_t& processed_prefixes) override;
   virtual int get_info(const DoutPrefixProvider *dpp, optional_yield y, rgw_placement_rule** rule, rgw::sal::Attrs* attrs = nullptr) override;
   virtual std::unique_ptr<Writer> get_writer(const DoutPrefixProvider *dpp,
 			  optional_yield y,
diff --git a/src/rgw/driver/posix/rgw_sal_posix.cc b/src/rgw/driver/posix/rgw_sal_posix.cc
index 8ffe4a0d0ca..d676213727e 100644
--- a/src/rgw/driver/posix/rgw_sal_posix.cc
+++ b/src/rgw/driver/posix/rgw_sal_posix.cc
@@ -2730,7 +2730,9 @@ int POSIXBucket::rename(const DoutPrefixProvider* dpp, optional_yield y, Object*
 
 int POSIXObject::delete_object(const DoutPrefixProvider* dpp,
 				optional_yield y,
-				uint32_t flags)
+				uint32_t flags,
+                                std::list<rgw_obj_index_key>* remove_objs,
+				RGWObjVersionTracker* objv)
 {
   POSIXBucket *b = static_cast<POSIXBucket*>(get_bucket());
   if (!b) {
@@ -3505,7 +3507,7 @@ int POSIXObject::POSIXReadOp::get_attr(const DoutPrefixProvider* dpp, const char
 int POSIXObject::POSIXDeleteOp::delete_obj(const DoutPrefixProvider* dpp,
 					   optional_yield y, uint32_t flags)
 {
-  return source->delete_object(dpp, y, flags);
+  return source->delete_object(dpp, y, flags, nullptr, nullptr);
 }
 
 int POSIXObject::copy(const DoutPrefixProvider *dpp, optional_yield y,
@@ -3703,7 +3705,8 @@ int POSIXMultipartUpload::complete(const DoutPrefixProvider *dpp,
 				    RGWCompressionInfo& cs_info, off_t& ofs,
 				    std::string& tag, ACLOwner& owner,
 				    uint64_t olh_epoch,
-				    rgw::sal::Object* target_obj)
+				    rgw::sal::Object* target_obj,
+				    prefix_map_t& processed_prefixes)
 {
   char final_etag[CEPH_CRYPTO_MD5_DIGESTSIZE];
   char final_etag_str[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 16];
@@ -3852,6 +3855,15 @@ int POSIXMultipartUpload::complete(const DoutPrefixProvider *dpp,
   return 0;
 }
 
+int POSIXMultipartUpload::cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+    CephContext *cct, optional_yield y,
+    const rgw_obj& obj,
+    std::list<rgw_obj_index_key>& remove_objs,
+    prefix_map_t& processed_prefixes)
+{
+  return -ENOTSUP;
+}
+
 int POSIXMultipartUpload::get_info(const DoutPrefixProvider *dpp, optional_yield y,
 				   rgw_placement_rule** rule, rgw::sal::Attrs* attrs)
 {
diff --git a/src/rgw/driver/posix/rgw_sal_posix.h b/src/rgw/driver/posix/rgw_sal_posix.h
index 5291ba500f9..efe3bfd7a50 100644
--- a/src/rgw/driver/posix/rgw_sal_posix.h
+++ b/src/rgw/driver/posix/rgw_sal_posix.h
@@ -632,7 +632,9 @@ public:
 
   virtual int delete_object(const DoutPrefixProvider* dpp,
 			    optional_yield y,
-			    uint32_t flags) override;
+			    uint32_t flags,
+			    std::list<rgw_obj_index_key>* remove_objs,
+			    RGWObjVersionTracker* objv) override;
   virtual int copy_object(const ACLOwner& owner,
                const rgw_user& remote_user,
                req_info* info, const rgw_zone_id& source_zone,
@@ -879,7 +881,13 @@ public:
 		       RGWCompressionInfo& cs_info, off_t& ofs,
 		       std::string& tag, ACLOwner& owner,
 		       uint64_t olh_epoch,
-		       rgw::sal::Object* target_obj) override;
+		       rgw::sal::Object* target_obj,
+		       prefix_map_t& processed_prefixes) override;
+  virtual int cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+                                     CephContext *cct, optional_yield y,
+                                     const rgw_obj& obj,
+                                     std::list<rgw_obj_index_key>& remove_objs,
+                                     prefix_map_t& processed_prefixes) override;
   virtual int get_info(const DoutPrefixProvider *dpp, optional_yield y,
 		       rgw_placement_rule** rule, rgw::sal::Attrs* attrs) override;
 
diff --git a/src/rgw/driver/rados/rgw_bucket.cc b/src/rgw/driver/rados/rgw_bucket.cc
index 94d3066e5b5..b130617e1c3 100644
--- a/src/rgw/driver/rados/rgw_bucket.cc
+++ b/src/rgw/driver/rados/rgw_bucket.cc
@@ -149,7 +149,7 @@ int rgw_remove_object(const DoutPrefixProvider *dpp, rgw::sal::Driver* driver, r
 
   std::unique_ptr<rgw::sal::Object> object = bucket->get_object(key);
 
-  return object->delete_object(dpp, y, rgw::sal::FLAG_LOG_OP);
+  return object->delete_object(dpp, y, rgw::sal::FLAG_LOG_OP, nullptr, nullptr);
 }
 
 static void set_err_msg(std::string *sink, std::string msg)
diff --git a/src/rgw/driver/rados/rgw_object_expirer_core.cc b/src/rgw/driver/rados/rgw_object_expirer_core.cc
index c285443d0b0..a5d788ea469 100644
--- a/src/rgw/driver/rados/rgw_object_expirer_core.cc
+++ b/src/rgw/driver/rados/rgw_object_expirer_core.cc
@@ -225,7 +225,7 @@ int RGWObjectExpirer::garbage_single_object(const DoutPrefixProvider *dpp, objex
 
   std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(key);
   obj->set_atomic();
-  ret = obj->delete_object(dpp, null_yield, rgw::sal::FLAG_LOG_OP);
+  ret = obj->delete_object(dpp, null_yield, rgw::sal::FLAG_LOG_OP, nullptr, nullptr);
 
   return ret;
 }
diff --git a/src/rgw/driver/rados/rgw_putobj_processor.cc b/src/rgw/driver/rados/rgw_putobj_processor.cc
index 79be7b3209b..f04ed1db8d4 100644
--- a/src/rgw/driver/rados/rgw_putobj_processor.cc
+++ b/src/rgw/driver/rados/rgw_putobj_processor.cc
@@ -22,6 +22,8 @@
 #include "services/svc_zone.h"
 #include "rgw_sal_rados.h"
 
+#include "cls/version/cls_version_client.h"
+
 #define dout_subsys ceph_subsys_rgw
 
 using namespace std;
@@ -573,7 +575,9 @@ int MultipartObjectProcessor::complete(
   }
 
   librados::ObjectWriteOperation op;
+  op.assert_exists();
   cls_rgw_mp_upload_part_info_update(op, p, info);
+  cls_version_inc(op);
   r = rgw_rados_operate(rctx.dpp, meta_obj_ref.ioctx, meta_obj_ref.obj.oid, &op, rctx.y);
   ldpp_dout(rctx.dpp, 20) << "Update meta: " << meta_obj_ref.obj.oid << " part " << p << " prefix " << info.manifest.get_prefix() << " return " << r << dendl;
 
@@ -588,8 +592,10 @@ int MultipartObjectProcessor::complete(
     op = librados::ObjectWriteOperation{};
     op.assert_exists(); // detect races with abort
     op.omap_set(m);
+    cls_version_inc(op);
     r = rgw_rados_operate(rctx.dpp, meta_obj_ref.ioctx, meta_obj_ref.obj.oid, &op, rctx.y);
   }
+
   if (r < 0) {
     return r == -ENOENT ? -ERR_NO_SUCH_UPLOAD : r;
   }
diff --git a/src/rgw/driver/rados/rgw_rados.cc b/src/rgw/driver/rados/rgw_rados.cc
index 23c149574ae..f9c95553820 100644
--- a/src/rgw/driver/rados/rgw_rados.cc
+++ b/src/rgw/driver/rados/rgw_rados.cc
@@ -5856,6 +5856,10 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
 
   store->remove_rgw_head_obj(op);
 
+  if (params.check_objv != nullptr) {
+    cls_version_check(op, *params.check_objv, VER_COND_EQ);
+  }
+
   auto& ioctx = ref.ioctx;
   r = rgw_rados_operate(dpp, ioctx, ref.obj.oid, &op, y);
 
@@ -6050,7 +6054,7 @@ int RGWRados::get_obj_state_impl(const DoutPrefixProvider *dpp, RGWObjectCtx *oc
   int r = -ENOENT;
 
   if (!assume_noent) {
-    r = RGWRados::raw_obj_stat(dpp, raw_obj, &s->size, &s->mtime, &s->epoch, &s->attrset, (s->prefetch_data ? &s->data : NULL), NULL, y);
+    r = RGWRados::raw_obj_stat(dpp, raw_obj, &s->size, &s->mtime, &s->epoch, &s->attrset, (s->prefetch_data ? &s->data : NULL), &s->objv_tracker, y);
   }
 
   if (r == -ENOENT) {
@@ -6783,6 +6787,10 @@ int RGWRados::Object::Read::prepare(optional_yield y, const DoutPrefixProvider *
     return -ENOENT;
   }
 
+  if (params.objv_tracker) {
+    *params.objv_tracker = astate->objv_tracker;
+  }
+
   RGWBucketInfo& bucket_info = source->get_bucket_info();
 
   if (params.part_num) {
@@ -8709,6 +8717,7 @@ int RGWRados::raw_obj_stat(const DoutPrefixProvider *dpp,
   if (first_chunk) {
     op.read(0, cct->_conf->rgw_max_chunk_size, first_chunk, NULL);
   }
+
   bufferlist outbl;
   r = rgw_rados_operate(dpp, ref.ioctx, ref.obj.oid, &op, &outbl, y);
 
diff --git a/src/rgw/driver/rados/rgw_rados.h b/src/rgw/driver/rados/rgw_rados.h
index bfae32888f6..4943cd9d002 100644
--- a/src/rgw/driver/rados/rgw_rados.h
+++ b/src/rgw/driver/rados/rgw_rados.h
@@ -771,6 +771,7 @@ public:
 	uint64_t *epoch;
         int* part_num = nullptr;
         std::optional<int> parts_count;
+        RGWObjVersionTracker *objv_tracker = nullptr;
 
         Params() : lastmod(nullptr), obj_size(nullptr), attrs(nullptr),
 		   target_obj(nullptr), epoch(nullptr)
@@ -854,8 +855,9 @@ public:
         rgw_zone_set *zones_trace;
 	bool abortmp;
 	uint64_t parts_accounted_size;
+	obj_version *check_objv;
 
-        DeleteParams() : versioning_status(0), null_verid(false), olh_epoch(0), bilog_flags(0), remove_objs(NULL), high_precision_time(false), zones_trace(nullptr), abortmp(false), parts_accounted_size(0) {}
+        DeleteParams() : versioning_status(0), null_verid(false), olh_epoch(0), bilog_flags(0), remove_objs(NULL), high_precision_time(false), zones_trace(nullptr), abortmp(false), parts_accounted_size(0), check_objv(nullptr) {}
       } params;
 
       struct DeleteResult {
diff --git a/src/rgw/driver/rados/rgw_sal_rados.cc b/src/rgw/driver/rados/rgw_sal_rados.cc
index 50c738d5435..5cfc0d87ddf 100644
--- a/src/rgw/driver/rados/rgw_sal_rados.cc
+++ b/src/rgw/driver/rados/rgw_sal_rados.cc
@@ -2318,6 +2318,7 @@ int RadosObject::read_attrs(const DoutPrefixProvider* dpp, RGWRados::Object::Rea
   read_op.params.target_obj = target_obj;
   read_op.params.obj_size = &state.size;
   read_op.params.lastmod = &state.mtime;
+  read_op.params.objv_tracker = &state.objv_tracker;
 
   return read_op.prepare(y, dpp);
 }
@@ -2848,6 +2849,9 @@ int RadosObject::RadosDeleteOp::delete_obj(const DoutPrefixProvider* dpp, option
   parent_op.params.abortmp = params.abortmp;
   parent_op.params.parts_accounted_size = params.parts_accounted_size;
   parent_op.params.null_verid = params.null_verid;
+  if (params.objv_tracker) {
+      parent_op.params.check_objv = params.objv_tracker->version_for_check();
+  }
 
   int ret = parent_op.delete_obj(y, dpp, flags & FLAG_LOG_OP);
   if (ret < 0)
@@ -2861,7 +2865,9 @@ int RadosObject::RadosDeleteOp::delete_obj(const DoutPrefixProvider* dpp, option
 
 int RadosObject::delete_object(const DoutPrefixProvider* dpp,
 			       optional_yield y,
-			       uint32_t flags)
+			       uint32_t flags,
+			       std::list<rgw_obj_index_key>* remove_objs,
+			       RGWObjVersionTracker* objv)
 {
   RGWRados::Object del_target(store->getRados(), bucket->get_info(), *rados_ctx, get_obj());
   RGWRados::Object::Delete del_op(&del_target);
@@ -2869,6 +2875,10 @@ int RadosObject::delete_object(const DoutPrefixProvider* dpp,
   del_op.params.bucket_owner = bucket->get_info().owner;
   del_op.params.versioning_status = (flags & FLAG_PREVENT_VERSIONING)
                                     ? 0 : bucket->get_info().versioning_status();
+  del_op.params.remove_objs = remove_objs;
+  if (objv) {
+      del_op.params.check_objv = objv->version_for_check();
+  }
 
   return del_op.delete_obj(y, dpp, flags & FLAG_LOG_OP);
 }
@@ -2964,13 +2974,84 @@ int RadosObject::swift_versioning_copy(const ACLOwner& owner, const rgw_user& re
                                         y);
 }
 
+int RadosMultipartUpload::cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+                                                 CephContext *cct, optional_yield y,
+                                                 const rgw_obj& obj,
+                                                 list<rgw_obj_index_key>& remove_objs,
+                                                 prefix_map_t& processed_prefixes)
+{
+  bool truncated;
+  int ret;
+  int max_parts = 1000;
+  int marker = 0;
+  cls_rgw_obj_chain chain;
+
+  do {
+    ret = list_parts(dpp, cct, max_parts, marker, &marker, &truncated, y);
+
+    if (ret < 0) {
+      ldpp_dout(dpp, 20) << __func__ << ": RadosMultipartUpload::list_parts returned " << ret << dendl;
+      return (ret == -ENOENT) ? -ERR_NO_SUCH_UPLOAD : ret;
+    }
+
+    for (auto part_it = parts.begin(); part_it != parts.end(); ++part_it) {
+      RadosMultipartPart* part = dynamic_cast<RadosMultipartPart*>(part_it->second.get());
+
+      auto& part_prefixes = processed_prefixes[part->info.num];
+
+      if (!part->info.manifest.empty()) {
+        auto manifest_prefix = part->info.manifest.get_prefix();
+        if (not manifest_prefix.empty() && part_prefixes.find(manifest_prefix) == part_prefixes.end()) {
+          store->getRados()->update_gc_chain(dpp, obj, part->info.manifest, &chain);
+
+          RGWObjManifest::obj_iterator oiter = part->info.manifest.obj_begin(dpp);
+          if (oiter != part->info.manifest.obj_end(dpp)) {
+            rgw_raw_obj raw_head = oiter.get_location().get_raw_obj(store->getRados());
+
+            rgw_obj head_obj;
+            RGWSI_Tier_RADOS::raw_obj_to_obj(bucket->get_key(), raw_head, &head_obj);
+
+            rgw_obj_index_key remove_key;
+            head_obj.key.get_index_key(&remove_key);
+            remove_objs.push_back(remove_key);
+          }
+        }
+      }
+      cleanup_part_history(dpp, y, part, remove_objs, part_prefixes);
+    }
+  } while (truncated);
+
+  if (store->getRados()->get_gc() == nullptr) {
+    //Delete objects inline if gc hasn't been initialised (in case when bypass gc is specified)
+    store->getRados()->delete_objs_inline(dpp, chain, mp_obj.get_upload_id(), y);
+  } else {
+    /* use upload id as tag and do it synchronously */
+    auto [ret, leftover_chain] = store->getRados()->send_chain_to_gc(chain, mp_obj.get_upload_id(), y);
+    if (ret < 0 && leftover_chain) {
+      ldpp_dout(dpp, 5) << __func__ << ": gc->send_chain() returned " << ret << dendl;
+      if (ret == -ENOENT) {
+        return -ERR_NO_SUCH_UPLOAD;
+      }
+      //Delete objects inline if send chain to gc fails
+      store->getRados()->delete_objs_inline(dpp, *leftover_chain, mp_obj.get_upload_id(), y);
+    }
+  }
+  return 0;
+}
+
 int RadosMultipartUpload::cleanup_part_history(const DoutPrefixProvider* dpp,
                                                optional_yield y,
                                                RadosMultipartPart *part,
-                                               list<rgw_obj_index_key>& remove_objs)
+                                               list<rgw_obj_index_key>& remove_objs,
+                                               boost::container::flat_set<std::string>& processed_prefixes)
 {
   cls_rgw_obj_chain chain;
   for (auto& ppfx : part->get_past_prefixes()) {
+    auto [it, inserted] = processed_prefixes.emplace(ppfx);
+    if (!inserted) {
+      continue; // duplicate
+    }
+
     rgw_obj past_obj;
     past_obj.init_ns(bucket->get_key(), ppfx + "." + std::to_string(part->info.num), mp_ns);
     rgw_obj_index_key past_key;
@@ -3018,77 +3099,105 @@ int RadosMultipartUpload::abort(const DoutPrefixProvider *dpp, CephContext *cct,
   int ret;
   uint64_t parts_accounted_size = 0;
 
-  do {
-    ret = list_parts(dpp, cct, 1000, marker, &marker, &truncated, y);
+  prefix_map_t processed_prefixes;
+
+  static constexpr auto MAX_DELETE_RETRIES = 15u;
+  for (auto i = 0u; i < MAX_DELETE_RETRIES; i++) {
+    ret = meta_obj->get_obj_attrs(y, dpp);
     if (ret < 0) {
-      ldpp_dout(dpp, 20) << __func__ << ": RadosMultipartUpload::list_parts returned " <<
-	ret << dendl;
+      ldpp_dout(dpp, 0) << __func__ << ": ERROR: failed to get obj attrs, obj=" << meta_obj
+                        << " ret=" << ret << dendl;
       return (ret == -ENOENT) ? -ERR_NO_SUCH_UPLOAD : ret;
     }
 
-    for (auto part_it = parts.begin();
-	 part_it != parts.end();
-	 ++part_it) {
-      RadosMultipartPart* obj_part = dynamic_cast<RadosMultipartPart*>(part_it->second.get());
-      if (obj_part->info.manifest.empty()) {
-	std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(
-				    rgw_obj_key(obj_part->oid, std::string(), RGW_OBJ_NS_MULTIPART));
-	obj->set_hash_source(mp_obj.get_key());
-	ret = obj->delete_object(dpp, y, 0);
-        if (ret < 0 && ret != -ENOENT)
-          return ret;
-      } else {
-	auto target = meta_obj->get_obj();
-	store->getRados()->update_gc_chain(dpp, target, obj_part->info.manifest, &chain);
-        RGWObjManifest::obj_iterator oiter = obj_part->info.manifest.obj_begin(dpp);
-        if (oiter != obj_part->info.manifest.obj_end(dpp)) {
-	  std::unique_ptr<rgw::sal::Object> head = bucket->get_object(rgw_obj_key());
-          rgw_raw_obj raw_head = oiter.get_location().get_raw_obj(store->getRados());
-	  dynamic_cast<rgw::sal::RadosObject*>(head.get())->raw_obj_to_obj(raw_head);
-
-          rgw_obj_index_key key;
-          head->get_key().get_index_key(&key);
-          remove_objs.push_back(key);
-
-          cleanup_part_history(dpp, null_yield, obj_part, remove_objs);
+    RGWObjVersionTracker objv_tracker = meta_obj->get_version_tracker();
+
+    do {
+      ret = list_parts(dpp, cct, 1000, marker, &marker, &truncated, y);
+      if (ret < 0) {
+        ldpp_dout(dpp, 20) << __func__ << ": RadosMultipartUpload::list_parts returned " << ret << dendl;
+        return (ret == -ENOENT) ? -ERR_NO_SUCH_UPLOAD : ret;
+      }
+
+      for (auto part_it = parts.begin(); part_it != parts.end(); ++part_it) {
+        RadosMultipartPart* obj_part = dynamic_cast<RadosMultipartPart*>(part_it->second.get());
+
+        if (obj_part->info.manifest.empty()) {
+          std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(
+            rgw_obj_key(obj_part->oid, std::string(), RGW_OBJ_NS_MULTIPART));
+          obj->set_hash_source(mp_obj.get_key());
+          ret = obj->delete_object(dpp, y, 0, nullptr, nullptr);
+          if (ret < 0 && ret != -ENOENT)
+            return ret;
+        } else {
+          auto manifest_prefix = obj_part->info.manifest.get_prefix();
+          auto [it, inserted] = processed_prefixes.emplace(obj_part->info.num, boost::container::flat_set<std::string>{});
+          if (not manifest_prefix.empty()) {
+            if (it->second.find(manifest_prefix) != it->second.end()) {
+              continue;
+            }
+            it->second.emplace(manifest_prefix);
+          }
+
+          auto target = meta_obj->get_obj();
+          store->getRados()->update_gc_chain(dpp, target, obj_part->info.manifest, &chain);
+          RGWObjManifest::obj_iterator oiter = obj_part->info.manifest.obj_begin(dpp);
+          if (oiter != obj_part->info.manifest.obj_end(dpp)) {
+            std::unique_ptr<rgw::sal::Object> head = bucket->get_object(rgw_obj_key());
+            rgw_raw_obj raw_head = oiter.get_location().get_raw_obj(store->getRados());
+            dynamic_cast<rgw::sal::RadosObject*>(head.get())->raw_obj_to_obj(raw_head);
+
+            rgw_obj_index_key key;
+            head->get_key().get_index_key(&key);
+            remove_objs.push_back(key);
+
+            cleanup_part_history(dpp, null_yield, obj_part, remove_objs, it->second);
+          }
         }
+        parts_accounted_size += obj_part->info.accounted_size;
       }
-      parts_accounted_size += obj_part->info.accounted_size;
-    }
-  } while (truncated);
+    } while (truncated);
 
-  if (store->getRados()->get_gc() == nullptr) {
-    //Delete objects inline if gc hasn't been initialised (in case when bypass gc is specified)
-    store->getRados()->delete_objs_inline(dpp, chain, mp_obj.get_upload_id(), y);
-  } else {
-    /* use upload id as tag and do it synchronously */
-    auto [ret, leftover_chain] = store->getRados()->send_chain_to_gc(chain, mp_obj.get_upload_id(), y);
-    if (ret < 0 && leftover_chain) {
-      ldpp_dout(dpp, 5) << __func__ << ": gc->send_chain() returned " << ret << dendl;
-      if (ret == -ENOENT) {
-        return -ERR_NO_SUCH_UPLOAD;
+    if (store->getRados()->get_gc() == nullptr) {
+      //Delete objects inline if gc hasn't been initialised (in case when bypass gc is specified)
+      store->getRados()->delete_objs_inline(dpp, chain, mp_obj.get_upload_id(), y);
+    } else {
+      /* use upload id as tag and do it synchronously */
+      auto [ret, leftover_chain] = store->getRados()->send_chain_to_gc(chain, mp_obj.get_upload_id(), y);
+      if (ret < 0 && leftover_chain) {
+        ldpp_dout(dpp, 5) << __func__ << ": gc->send_chain() returned " << ret << dendl;
+        if (ret == -ENOENT) {
+          return -ERR_NO_SUCH_UPLOAD;
+        }
+        //Delete objects inline if send chain to gc fails
+        store->getRados()->delete_objs_inline(dpp, *leftover_chain, mp_obj.get_upload_id(), y);
       }
-      //Delete objects inline if send chain to gc fails
-      store->getRados()->delete_objs_inline(dpp, *leftover_chain, mp_obj.get_upload_id(), y);
     }
-  }
 
-  std::unique_ptr<rgw::sal::Object::DeleteOp> del_op = meta_obj->get_delete_op();
-  del_op->params.bucket_owner = bucket->get_info().owner;
-  del_op->params.versioning_status = 0;
-  if (!remove_objs.empty()) {
-    del_op->params.remove_objs = &remove_objs;
-  }
+    std::unique_ptr<rgw::sal::Object::DeleteOp> del_op = meta_obj->get_delete_op();
+    del_op->params.bucket_owner = bucket->get_info().owner;
+    del_op->params.versioning_status = 0;
+    if (!remove_objs.empty()) {
+      del_op->params.remove_objs = &remove_objs;
+    }
 
-  del_op->params.abortmp = true;
-  del_op->params.parts_accounted_size = parts_accounted_size;
+    del_op->params.abortmp = true;
+    del_op->params.parts_accounted_size = parts_accounted_size;
+    del_op->params.objv_tracker = &objv_tracker;
 
-  // and also remove the metadata obj
-  ret = del_op->delete_obj(dpp, y, 0);
-  if (ret < 0) {
-    ldpp_dout(dpp, 20) << __func__ << ": del_op.delete_obj returned " <<
-      ret << dendl;
+    // and also remove the metadata obj
+    ret = del_op->delete_obj(dpp, y, 0);
+    if (ret != -ECANCELED) {
+      if (ret < 0) {
+        ldpp_dout(dpp, 20) << __func__ << ": del_op.delete_obj returned " << ret << dendl;
+      }
+      break;
+    }
+    ldpp_dout(dpp, 20) << "deleting meta_obj is cancelled due to mismatch cls_version: " << objv_tracker << dendl;
+    chain.objs.clear();
+    marker = 0;
   }
+
   return (ret == -ENOENT) ? -ERR_NO_SUCH_UPLOAD : ret;
 }
 
@@ -3269,7 +3378,8 @@ int RadosMultipartUpload::complete(const DoutPrefixProvider *dpp,
 				   RGWCompressionInfo& cs_info, off_t& ofs,
 				   std::string& tag, ACLOwner& owner,
 				   uint64_t olh_epoch,
-				   rgw::sal::Object* target_obj)
+				   rgw::sal::Object* target_obj,
+				   prefix_map_t& processed_prefixes)
 {
   char final_etag[CEPH_CRYPTO_MD5_DIGESTSIZE];
   char final_etag_str[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 16];
@@ -3341,6 +3451,8 @@ int RadosMultipartUpload::complete(const DoutPrefixProvider *dpp,
       rgw_obj src_obj;
       src_obj.init_ns(bucket->get_key(), oid, mp_ns);
 
+      auto [it, inserted] = processed_prefixes.emplace(part->info.num, boost::container::flat_set<std::string>{});
+
       if (obj_part.manifest.empty()) {
         ldpp_dout(dpp, 0) << "ERROR: empty manifest for object part: obj="
 			 << src_obj << dendl;
@@ -3352,6 +3464,7 @@ int RadosMultipartUpload::complete(const DoutPrefixProvider *dpp,
         if (not manifest_prefix.empty()) {
           // It has an explicit prefix. Override the default one.
           src_obj.init_ns(bucket->get_key(), manifest_prefix + "." + std::to_string(part->info.num), mp_ns);
+	  it->second.emplace(manifest_prefix);
         }
       }
 
@@ -3397,7 +3510,7 @@ int RadosMultipartUpload::complete(const DoutPrefixProvider *dpp,
 
       remove_objs.push_back(remove_key);
 
-      cleanup_part_history(dpp, y, part, remove_objs);
+      cleanup_part_history(dpp, y, part, remove_objs, it->second);
 
       ofs += obj_part.size;
       accounted_size += obj_part.accounted_size;
diff --git a/src/rgw/driver/rados/rgw_sal_rados.h b/src/rgw/driver/rados/rgw_sal_rados.h
index 98d0bc9d005..705b665d46d 100644
--- a/src/rgw/driver/rados/rgw_sal_rados.h
+++ b/src/rgw/driver/rados/rgw_sal_rados.h
@@ -559,7 +559,9 @@ class RadosObject : public StoreObject {
       rados_ctx->invalidate(get_obj());
     }
     virtual int delete_object(const DoutPrefixProvider* dpp,
-			      optional_yield y, uint32_t flags) override;
+			      optional_yield y, uint32_t flags,
+			      std::list<rgw_obj_index_key>* remove_objs,
+			      RGWObjVersionTracker* objv) override;
     virtual int copy_object(const ACLOwner& owner,
                const rgw_user& remote_user,
                req_info* info, const rgw_zone_id& source_zone,
@@ -825,7 +827,13 @@ public:
 		       RGWCompressionInfo& cs_info, off_t& ofs,
 		       std::string& tag, ACLOwner& owner,
 		       uint64_t olh_epoch,
-		       rgw::sal::Object* target_obj) override;
+		       rgw::sal::Object* target_obj,
+		       prefix_map_t& processed_prefixes) override;
+  virtual int cleanup_orphaned_parts(const DoutPrefixProvider *dpp,
+                                     CephContext *cct, optional_yield y,
+                                     const rgw_obj& obj,
+                                     std::list<rgw_obj_index_key>& remove_objs,
+                                     prefix_map_t& processed_prefixes) override;
   virtual int get_info(const DoutPrefixProvider *dpp, optional_yield y, rgw_placement_rule** rule, rgw::sal::Attrs* attrs = nullptr) override;
   virtual std::unique_ptr<Writer> get_writer(const DoutPrefixProvider *dpp,
 			  optional_yield y,
@@ -838,7 +846,8 @@ protected:
   int cleanup_part_history(const DoutPrefixProvider* dpp,
                            optional_yield y,
                            RadosMultipartPart* part,
-                           std::list<rgw_obj_index_key>& remove_objs);
+                           std::list<rgw_obj_index_key>& remove_objs,
+                           boost::container::flat_set<std::string>& processed_prefixes);
 };
 
 class MPRadosSerializer : public StoreMPSerializer {