ReplicatedPG::finish_promote: handle results->snaps is empty case

If results->snaps winds up empty after filtering removed snaps,
we need to treat the object as if we had gotten an ENOENT.

PartialFix: #11296
Backport: firefly, hammer
Signed-off-by: Samuel Just <sjust@redhat.com>

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index c8595d8f0d1aac15e7f9a95a12e3bb3124213252..853c81d14abc30615d936f6f09962da4f500809f 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -6512,6 +6512,35 @@ void ReplicatedPG::finish_promote(int r, CopyResults *results,
     return;
   }
 
+  if (r != -ENOENT && soid.is_snap()) {
+    if (results->snaps.empty()) {
+      // we must have read "snap" content from the head object in
+      // the base pool.  use snap_seq to construct what snaps should
+      // be for this clone (what is was before we evicted the clean
+      // clone from this pool, and what it will be when we flush and
+      // the clone eventually happens in the base pool).
+      SnapSet& snapset = obc->ssc->snapset;
+      vector<snapid_t>::iterator p = snapset.snaps.begin();
+      while (p != snapset.snaps.end() && *p > soid.snap)
+       ++p;
+      while (p != snapset.snaps.end() && *p > results->snap_seq) {
+       results->snaps.push_back(*p);
+       ++p;
+      }
+    }
+
+    dout(20) << __func__ << " snaps " << results->snaps << dendl;
+    filter_snapc(results->snaps);
+
+    dout(20) << __func__ << " filtered snaps " << results->snaps << dendl;
+    if (results->snaps.empty()) {
+      dout(20) << __func__
+              << " snaps are empty, clone is invalid,"
+              << " setting r to ENOENT" << dendl;
+      r = -ENOENT;
+    }
+  }
+
   if (r == -ENOENT && results->started_temp_obj) {
     dout(10) << __func__ << " abort; will clean up partial work" << dendl;
     ObjectContextRef tempobc = get_object_context(results->temp_oid, true);
@@ -6616,24 +6645,7 @@ void ReplicatedPG::finish_promote(int r, CopyResults *results,
     tctx->new_obs.oi.user_version = results->user_version;
 
     if (soid.snap != CEPH_NOSNAP) {
-      if (!results->snaps.empty()) {
-       tctx->new_obs.oi.snaps = results->snaps;
-      } else {
-       // we must have read "snap" content from the head object in
-       // the base pool.  use snap_seq to construct what snaps should
-       // be for this clone (what is was before we evicted the clean
-       // clone from this pool, and what it will be when we flush and
-       // the clone eventually happens in the base pool).
-       SnapSet& snapset = obc->ssc->snapset;
-       vector<snapid_t>::iterator p = snapset.snaps.begin();
-       while (p != snapset.snaps.end() && *p > soid.snap)
-         ++p;
-       while (p != snapset.snaps.end() && *p > results->snap_seq) {
-         tctx->new_obs.oi.snaps.push_back(*p);
-         ++p;
-       }
-      }
-      dout(20) << __func__ << " snaps " << tctx->new_obs.oi.snaps << dendl;
+      tctx->new_obs.oi.snaps = results->snaps;
       assert(!tctx->new_obs.oi.snaps.empty());
       assert(obc->ssc->snapset.clone_size.count(soid.snap));
       assert(obc->ssc->snapset.clone_size[soid.snap] ==