Merge pull request ceph#64272 from linuxbox2/wip-matt-70853

rgwlc: fix removal of delete markers (SAL)

Author: mattbenjamin

src/rgw/driver/rados/rgw_rados.cc

@@ -6921,12 +6921,12 @@ int RGWRados::get_obj_state(const DoutPrefixProvider *dpp, RGWObjectCtx *rctx,
   RGWObjStateManifest* sm = nullptr;
   int r = get_obj_state(dpp, rctx, bucket_info, obj, &sm,
                         follow_olh, y, assume_noent);
-  if (r < 0) {
-    return r;
-  }
   if (pstate) {
     *pstate = &sm->state;
   }
+  if (r < 0) {
+    return r;
+  }
   if (pmanifest) {
     if (sm->manifest) {
       *pmanifest = &(*sm->manifest);
@@ -9532,6 +9532,8 @@ int RGWRados::follow_olh(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_in
   }
 
   if (olh.removed) {
+    /* the object is a delete marker */
+    state->is_dm = true;
     return -ENOENT;
   }
 

src/rgw/driver/rados/rgw_sal_rados.cc

@@ -2782,6 +2782,9 @@ int RadosObject::load_obj_state(const DoutPrefixProvider* dpp, optional_yield y,
 
   int ret = store->getRados()->get_obj_state(dpp, rados_ctx, bucket->get_info(), get_obj(), &pstate, &manifest, follow_olh, y);
   if (ret < 0) {
+    if (ret == -ENOENT) {
+      state.is_dm = pstate->is_dm;
+    }
     return ret;
   }
 

src/rgw/rgw_lc.cc

@@ -623,10 +623,15 @@ static int remove_expired_obj(const DoutPrefixProvider* dpp,
   auto obj = oc.bucket->get_object(obj_key);
   ret = obj->load_obj_state(dpp, null_yield, true);
   if (ret < 0) {
-    ldpp_dout(oc.dpp, 0) <<
-      fmt::format("ERROR: get_obj_state() failed in {} for object k={} error r={}",
-		  __func__, oc.o.key.to_string(), ret) << dendl;
-    return ret;
+    /* for delete markers, we expect load_obj_state() to "fail"
+     * with -ENOENT */
+    if (! (o.is_delete_marker() &&
+	   (ret == -ENOENT))) {
+      ldpp_dout(oc.dpp, 0) <<
+	fmt::format("ERROR: get_obj_state() failed in {} for object k={} error r={}",
+		    __func__, oc.o.key.to_string(), ret) << dendl;
+      return ret;
+    }
   }
 
   auto have_notify = !event_types.empty();
@@ -1280,13 +1285,14 @@ class LCOpAction_DMExpiration : public LCOpAction {
 			<< oc.wq->thr_name() << dendl;
       return false;
     }
+    /* don't remove the delete marker if that would expose a non-current
+     * version as current */
     if (oc.next_has_same_name(o.key.name)) {
       ldpp_dout(dpp, 20) << __func__ << "(): key=" << o.key
-			<< ": next is same object, skipping "
+			<< ": dm expiration would expose a non-current version, skipping "
 			<< oc.wq->thr_name() << dendl;
       return false;
     }
-
     *exp_time = real_clock::now();
 
     return true;

src/rgw/rgw_rest.cc

@@ -1931,7 +1931,20 @@ int RGWHandler_REST::read_permissions(RGWOp* op_obj, optional_yield y)
     return -EINVAL;
   }
 
-  return do_read_permissions(op_obj, only_bucket, y);
+  auto ret = do_read_permissions(op_obj, only_bucket, y);
+  switch (s->op) {
+  case OP_HEAD:
+  case OP_GET:
+    if (ret == -ENOENT /* note, access already accounted for */) [[unlikely]] {
+      (void) s->object->load_obj_state(s, s->yield, true /* follow_olh */);
+      auto tf = s->object->is_delete_marker() ? "true" : "false";
+      dump_header(s, "x-amz-delete-marker", tf);
+    }
+  default:
+    break;
+  }
+
+  return ret;
 }
 
 void RGWRESTMgr::register_resource(string resource, RGWRESTMgr *mgr)

src/rgw/rgw_sal.h

@@ -1223,6 +1223,8 @@ class Object {
     virtual void set_compressed() = 0;
     /** Check if this object is compressed */
     virtual bool is_compressed() = 0;
+    /** True if this object is a delete marker (newest version is deleted) */
+    virtual bool is_delete_marker() = 0;
     /** Check if object is synced */
     virtual bool is_sync_completed(const DoutPrefixProvider* dpp,
                                    optional_yield y,
@@ -1344,8 +1346,6 @@ class Object {
     virtual void set_hash_source(std::string s) = 0;
     /** Build an Object Identifier string for this object */
     virtual std::string get_oid(void) const = 0;
-    /** True if this object is a delete marker (newest version is deleted) */
-    virtual bool get_delete_marker(void) = 0;
     /** True if this object is stored in the extra data pool */
     virtual bool get_in_extra_data(void) = 0;
     /** True if this object exists in the store */

src/rgw/rgw_sal_filter.h

@@ -789,6 +789,7 @@ class FilterObject : public Object {
   virtual bool is_prefetch_data() override { return next->is_prefetch_data(); }
   virtual void set_compressed() override { return next->set_compressed(); }
   virtual bool is_compressed() override { return next->is_compressed(); }
+  virtual bool is_delete_marker() override { return next->is_delete_marker(); }
   bool is_sync_completed(const DoutPrefixProvider* dpp, optional_yield y,
                          const ceph::real_time& obj_mtime) override {
     return next->is_sync_completed(dpp, y, obj_mtime);
@@ -863,7 +864,6 @@ class FilterObject : public Object {
   virtual std::string get_hash_source(void) override { return next->get_hash_source(); };
   virtual void set_hash_source(std::string s) override { return next->set_hash_source(s); };
   virtual std::string get_oid(void) const override { return next->get_oid(); };
-  virtual bool get_delete_marker(void) override { return next->get_delete_marker(); };
   virtual bool get_in_extra_data(void) override { return next->get_in_extra_data(); };
   virtual bool exists(void) override { return next->exists(); };
   virtual void set_in_extra_data(bool i) override { return next->set_in_extra_data(i); };

src/rgw/rgw_sal_store.h

@@ -25,6 +25,7 @@ struct RGWObjState {
   bool is_atomic{false};
   bool has_attrs{false};
   bool exists{false};
+  bool is_dm{false};
   uint64_t size{0}; //< size of raw object
   uint64_t accounted_size{0}; //< size before compression, encryption
   ceph::real_time mtime;
@@ -279,7 +280,6 @@ class StoreObject : public Object {
   protected:
     RGWObjState state;
     Bucket* bucket = nullptr;
-    bool delete_marker{false};
     jspan_context trace_ctx{false, false};
 
   public:
@@ -299,6 +299,7 @@ class StoreObject : public Object {
     virtual bool is_prefetch_data() override { return state.prefetch_data; }
     virtual void set_compressed() override { state.compressed = true; }
     virtual bool is_compressed() override { return state.compressed; }
+    virtual bool is_delete_marker() override { return state.is_dm; }
     virtual void invalidate() override {
       rgw_obj obj = state.obj;
       bool is_atomic = state.is_atomic;
@@ -342,7 +343,6 @@ class StoreObject : public Object {
     virtual std::string get_hash_source(void) override { return state.obj.index_hash_source; }
     virtual void set_hash_source(std::string s) override { state.obj.index_hash_source = s; }
     virtual std::string get_oid(void) const override { return state.obj.key.get_oid(); }
-    virtual bool get_delete_marker(void) override { return delete_marker; }
     virtual bool get_in_extra_data(void) override { return state.obj.is_in_extra_data(); }
     virtual bool exists(void) override { return state.exists; }
     virtual void set_in_extra_data(bool i) override { state.obj.set_in_extra_data(i); }