From 18a45503011a572325e09b56d5ab799a15ee83d4 Mon Sep 17 00:00:00 2001
From: Jason Dillaman <dillaman@redhat.com>
Date: Wed, 9 Dec 2020 22:30:17 -0500
Subject: [PATCH] librbd/mirror: unlink peer might recursively loop

If the mirror peer set is (incorrectly) empty, it's not currently
possible for the unlink peer state machine to properly delete the
snapshot. This can result in a recursive loop between the create
primary snapshot state machine and the unlink peer state machine
until the stack depth grows too large.

Fixes: https://tracker.ceph.com/issues/48525
Signed-off-by: Jason Dillaman <dillaman@redhat.com>
---
 .../mirror/snapshot/UnlinkPeerRequest.cc      | 15 +++++-----
 .../mirror/snapshot/UnlinkPeerRequest.h       |  2 ++
 .../snapshot/test_mock_UnlinkPeerRequest.cc   | 29 +++++++++++++++++++
 3 files changed, 38 insertions(+), 8 deletions(-)

diff --git a/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc b/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
index cb059c760c1..d82d42d5bad 100644
--- a/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
+++ b/src/librbd/mirror/snapshot/UnlinkPeerRequest.cc
@@ -63,7 +63,7 @@ void UnlinkPeerRequest<I>::unlink_peer() {
   m_image_ctx->image_lock.lock_shared();
   int r = -ENOENT;
   cls::rbd::MirrorSnapshotNamespace* mirror_ns = nullptr;
-  bool newer_mirror_snapshots = false;
+  m_newer_mirror_snapshots = false;
   for (auto snap_it = m_image_ctx->snap_info.find(m_snap_id);
        snap_it != m_image_ctx->snap_info.end(); ++snap_it) {
     if (snap_it->first == m_snap_id) {
@@ -73,7 +73,7 @@ void UnlinkPeerRequest<I>::unlink_peer() {
     } else if (boost::get<cls::rbd::MirrorSnapshotNamespace>(
                  &snap_it->second.snap_namespace) != nullptr) {
       ldout(cct, 20) << "located newer mirror snapshot" << dendl;
-      newer_mirror_snapshots = true;
+      m_newer_mirror_snapshots = true;
       break;
     }
   }
@@ -95,7 +95,7 @@ void UnlinkPeerRequest<I>::unlink_peer() {
   // if there is or will be no more peers in the mirror snapshot and we have
   // a more recent mirror snapshot, remove the older one
   if ((mirror_ns->mirror_peer_uuids.count(m_mirror_peer_uuid) == 0) ||
-      (mirror_ns->mirror_peer_uuids.size() == 1U && newer_mirror_snapshots)) {
+      (mirror_ns->mirror_peer_uuids.size() <= 1U && m_newer_mirror_snapshots)) {
     m_image_ctx->image_lock.unlock_shared();
     remove_snapshot();
     return;
@@ -184,15 +184,14 @@ void UnlinkPeerRequest<I>::remove_snapshot() {
   }
 
   auto info = boost::get<cls::rbd::MirrorSnapshotNamespace>(
-    &snap_namespace);
-  ceph_assert(info);
+    snap_namespace);
 
-  if (info->mirror_peer_uuids.size() > 1 ||
-      info->mirror_peer_uuids.count(m_mirror_peer_uuid) == 0) {
+  info.mirror_peer_uuids.erase(m_mirror_peer_uuid);
+  if (!info.mirror_peer_uuids.empty() || !m_newer_mirror_snapshots) {
     ldout(cct, 20) << "skipping removal of snapshot: "
                    << "snap_id=" << m_snap_id << ": "
                    << "mirror_peer_uuid=" << m_mirror_peer_uuid << ", "
-                   << "mirror_peer_uuids=" << info->mirror_peer_uuids << dendl;
+                   << "mirror_peer_uuids=" << info.mirror_peer_uuids << dendl;
     finish(0);
     return;
   }
diff --git a/src/librbd/mirror/snapshot/UnlinkPeerRequest.h b/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
index 184f7ccd956..9ef47269d87 100644
--- a/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
+++ b/src/librbd/mirror/snapshot/UnlinkPeerRequest.h
@@ -69,6 +69,8 @@ private:
   std::string m_mirror_peer_uuid;
   Context *m_on_finish;
 
+  bool m_newer_mirror_snapshots = false;
+
   void refresh_image();
   void handle_refresh_image(int r);
 
diff --git a/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc b/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc
index 2492e06f9bd..456b6ccdca1 100644
--- a/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc
+++ b/src/test/librbd/mirror/snapshot/test_mock_UnlinkPeerRequest.cc
@@ -182,6 +182,35 @@ TEST_F(TestMockMirrorSnapshotUnlinkPeerRequest, RemoveSnapshot) {
   ASSERT_EQ(0, ctx.wait());
 }
 
+TEST_F(TestMockMirrorSnapshotUnlinkPeerRequest, SnapshotRemoveEmptyPeers) {
+  REQUIRE_FORMAT_V2();
+
+  librbd::ImageCtx *ictx;
+  ASSERT_EQ(0, open_image(m_image_name, &ictx));
+
+  MockTestImageCtx mock_image_ctx(*ictx);
+  cls::rbd::MirrorSnapshotNamespace ns{
+    cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {},
+    "", CEPH_NOSNAP};
+  auto snap_id = snap_create(mock_image_ctx, ns, "mirror_snap");
+  ns.mirror_peer_uuids = {"peer_uuid"};
+  snap_create(mock_image_ctx, ns, "mirror_snap2");
+
+  expect_get_snap_info(mock_image_ctx, snap_id);
+
+  InSequence seq;
+
+  expect_is_refresh_required(mock_image_ctx, true);
+  expect_refresh_image(mock_image_ctx, 0);
+  expect_remove_snapshot(mock_image_ctx, snap_id, 0);
+
+  C_SaferCond ctx;
+  auto req = new MockUnlinkPeerRequest(&mock_image_ctx, snap_id, "peer_uuid",
+                                       &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
 TEST_F(TestMockMirrorSnapshotUnlinkPeerRequest, SnapshotDNE) {
   REQUIRE_FORMAT_V2();
 
-- 
2.39.5