Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

3 files changed, 4317 insertions, 0 deletions

diff --git a/src/test/rbd_mirror/image_replayer/snapshot/test_mock_ApplyImageStateRequest.cc b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_ApplyImageStateRequest.cc
new file mode 100644
index 000000000..904d36854
--- /dev/null
+++ b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_ApplyImageStateRequest.cc
@@ -0,0 +1,641 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/rbd_mirror/test_mock_fixture.h"
+#include "librbd/internal.h"
+#include "librbd/Operations.h"
+#include "librbd/image/GetMetadataRequest.h"
+#include "tools/rbd_mirror/Threads.h"
+#include "tools/rbd_mirror/image_replayer/snapshot/ApplyImageStateRequest.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librbd/mock/MockImageCtx.h"
+
+namespace librbd {
+
+namespace {
+
+struct MockTestImageCtx : public librbd::MockImageCtx {
+  explicit MockTestImageCtx(librbd::ImageCtx &image_ctx)
+    : librbd::MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace image {
+
+template <>
+struct GetMetadataRequest<MockTestImageCtx> {
+  std::map<std::string, bufferlist>* pairs = nullptr;
+  Context* on_finish = nullptr;
+
+  static GetMetadataRequest* s_instance;
+  static GetMetadataRequest* create(librados::IoCtx& io_ctx,
+                                    const std::string& oid,
+                                    bool filter_internal,
+                                    const std::string& filter_key_prefix,
+                                    const std::string& last_key,
+                                    size_t max_results,
+                                    std::map<std::string, bufferlist>* pairs,
+                                    Context* on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->pairs = pairs;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  GetMetadataRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+GetMetadataRequest<MockTestImageCtx>* GetMetadataRequest<MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace image
+} // namespace librbd
+
+#include "tools/rbd_mirror/image_replayer/snapshot/ApplyImageStateRequest.cc"
+
+using ::testing::_;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::StrEq;
+using ::testing::WithArg;
+
+namespace rbd {
+namespace mirror {
+namespace image_replayer {
+namespace snapshot {
+
+class TestMockImageReplayerSnapshotApplyImageStateRequest : public TestMockFixture {
+public:
+  typedef ApplyImageStateRequest<librbd::MockTestImageCtx> MockApplyImageStateRequest;
+  typedef librbd::image::GetMetadataRequest<librbd::MockTestImageCtx> MockGetMetadataRequest;
+
+  void SetUp() override {
+    TestMockFixture::SetUp();
+
+    librbd::RBD rbd;
+    ASSERT_EQ(0, create_image(rbd, m_local_io_ctx, m_image_name, m_image_size));
+    ASSERT_EQ(0, open_image(m_local_io_ctx, m_image_name, &m_local_image_ctx));
+
+    ASSERT_EQ(0, create_image(rbd, m_remote_io_ctx, m_image_name, m_image_size));
+    ASSERT_EQ(0, open_image(m_remote_io_ctx, m_image_name, &m_remote_image_ctx));
+
+    m_mock_local_image_ctx = new librbd::MockTestImageCtx(*m_local_image_ctx);
+    m_mock_remote_image_ctx = new librbd::MockTestImageCtx(*m_remote_image_ctx);
+  }
+
+  void TearDown() override {
+    delete m_mock_remote_image_ctx;
+    delete m_mock_local_image_ctx;
+    TestMockFixture::TearDown();
+  }
+
+  void expect_rename_image(const std::string& name, int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations, execute_rename(name, _))
+      .WillOnce(WithArg<1>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  void expect_update_features(uint64_t features, bool enable, int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations,
+                execute_update_features(features, enable, _, 0U))
+      .WillOnce(WithArg<2>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  void expect_get_metadata(MockGetMetadataRequest& mock_get_metadata_request,
+                           const std::map<std::string, bufferlist>& pairs,
+                           int r) {
+    EXPECT_CALL(mock_get_metadata_request, send())
+      .WillOnce(Invoke([this, &mock_get_metadata_request, pairs, r]() {
+          *mock_get_metadata_request.pairs = pairs;
+          m_threads->work_queue->queue(mock_get_metadata_request.on_finish, r);
+        }));
+  }
+
+  void expect_update_metadata(const std::vector<std::string>& remove,
+                              const std::map<std::string, bufferlist>& pairs,
+                              int r) {
+    for (auto& key : remove) {
+      bufferlist bl;
+      ceph::encode(key, bl);
+      EXPECT_CALL(get_mock_io_ctx(m_mock_local_image_ctx->md_ctx),
+                  exec(m_mock_local_image_ctx->header_oid, _, StrEq("rbd"),
+                  StrEq("metadata_remove"), ContentsEqual(bl), _, _, _))
+        .WillOnce(Return(r));
+      if (r < 0) {
+        return;
+      }
+    }
+
+    if (!pairs.empty()) {
+      bufferlist bl;
+      ceph::encode(pairs, bl);
+      EXPECT_CALL(get_mock_io_ctx(m_mock_local_image_ctx->md_ctx),
+                  exec(m_mock_local_image_ctx->header_oid, _, StrEq("rbd"),
+                  StrEq("metadata_set"), ContentsEqual(bl), _, _, _))
+        .WillOnce(Return(r));
+    }
+  }
+
+  void expect_unprotect_snapshot(const std::string& name, int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations,
+                execute_snap_unprotect({cls::rbd::UserSnapshotNamespace{}},
+                                       name, _))
+      .WillOnce(WithArg<2>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  void expect_remove_snapshot(const std::string& name, int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations,
+                execute_snap_remove({cls::rbd::UserSnapshotNamespace{}},
+                                     name, _))
+      .WillOnce(WithArg<2>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  void expect_protect_snapshot(const std::string& name, int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations,
+                execute_snap_protect({cls::rbd::UserSnapshotNamespace{}},
+                                     name, _))
+      .WillOnce(WithArg<2>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  void expect_rename_snapshot(uint64_t snap_id, const std::string& name,
+                              int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations,
+                execute_snap_rename(snap_id, name, _))
+      .WillOnce(WithArg<2>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+
+  void expect_set_snap_limit(uint64_t limit, int r) {
+    EXPECT_CALL(*m_mock_local_image_ctx->operations,
+                execute_snap_set_limit(limit, _))
+      .WillOnce(WithArg<1>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  librbd::ImageCtx *m_local_image_ctx;
+  librbd::ImageCtx *m_remote_image_ctx;
+
+  librbd::MockTestImageCtx *m_mock_local_image_ctx = nullptr;
+  librbd::MockTestImageCtx *m_mock_remote_image_ctx = nullptr;
+
+};
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, NoChanges) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, RenameImage) {
+  InSequence seq;
+
+  expect_rename_image("new name", 0);
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = "new name";
+  image_state.features = m_remote_image_ctx->features;
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, RenameImageError) {
+  InSequence seq;
+
+  expect_rename_image("new name", -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = "new name";
+  image_state.features = m_remote_image_ctx->features;
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, UpdateFeatures) {
+  InSequence seq;
+
+  expect_update_features(RBD_FEATURE_DEEP_FLATTEN, false, 0);
+  expect_update_features(RBD_FEATURE_OBJECT_MAP, true, 0);
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = RBD_FEATURE_EXCLUSIVE_LOCK |
+                         RBD_FEATURE_OBJECT_MAP;
+  m_mock_local_image_ctx->features = RBD_FEATURE_EXCLUSIVE_LOCK |
+                                     RBD_FEATURE_DEEP_FLATTEN;
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, UpdateFeaturesError) {
+  InSequence seq;
+
+  expect_update_features(RBD_FEATURE_DEEP_FLATTEN, false, -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = RBD_FEATURE_EXCLUSIVE_LOCK |
+                         RBD_FEATURE_OBJECT_MAP;
+  m_mock_local_image_ctx->features = RBD_FEATURE_EXCLUSIVE_LOCK |
+                                     RBD_FEATURE_DEEP_FLATTEN;
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, UpdateImageMeta) {
+  InSequence seq;
+
+  bufferlist data_bl;
+  ceph::encode("data", data_bl);
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request,
+                      {{"key1", {}}, {"key2", {}}}, 0);
+  expect_update_metadata({"key2"}, {{"key1", data_bl}}, 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.metadata = {{"key1", data_bl}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, GetImageMetaError) {
+  InSequence seq;
+
+  bufferlist data_bl;
+  ceph::encode("data", data_bl);
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request,
+                      {{"key1", {}}, {"key2", {}}}, -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.metadata = {{"key1", data_bl}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, UpdateImageMetaError) {
+  InSequence seq;
+
+  bufferlist data_bl;
+  ceph::encode("data", data_bl);
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request,
+                      {{"key1", {}}, {"key2", {}}}, 0);
+  expect_update_metadata({"key2"}, {{"key1", data_bl}}, -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.metadata = {{"key1", data_bl}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, UnprotectSnapshot) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_unprotect_snapshot("snap1", 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.snapshots = {
+    {1U, {cls::rbd::UserSnapshotNamespace{}, "snap1",
+          RBD_PROTECTION_STATUS_UNPROTECTED}}};
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_PROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, UnprotectSnapshotError) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_unprotect_snapshot("snap1", -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.snapshots = {
+    {1U, {cls::rbd::UserSnapshotNamespace{}, "snap1",
+          RBD_PROTECTION_STATUS_UNPROTECTED}}};
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_PROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, RemoveSnapshot) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_remove_snapshot("snap1", 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_UNPROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, RemoveSnapshotError) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_remove_snapshot("snap1", -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_UNPROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, ProtectSnapshot) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_protect_snapshot("snap1", 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.snapshots = {
+    {1U, {cls::rbd::UserSnapshotNamespace{}, "snap1",
+          RBD_PROTECTION_STATUS_PROTECTED}}};
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_UNPROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, ProtectSnapshotError) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_protect_snapshot("snap1", -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.snapshots = {
+    {1U, {cls::rbd::UserSnapshotNamespace{}, "snap1",
+          RBD_PROTECTION_STATUS_PROTECTED}}};
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_UNPROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, RenameSnapshot) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_rename_snapshot(11, "snap1-renamed", 0);
+
+  expect_set_snap_limit(0, 0);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.snapshots = {
+    {1U, {cls::rbd::UserSnapshotNamespace{}, "snap1-renamed",
+          RBD_PROTECTION_STATUS_PROTECTED}}};
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_PROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(0, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, RenameSnapshotError) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_rename_snapshot(11, "snap1-renamed", -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+  image_state.snapshots = {
+    {1U, {cls::rbd::UserSnapshotNamespace{}, "snap1-renamed",
+          RBD_PROTECTION_STATUS_PROTECTED}}};
+  m_mock_local_image_ctx->snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+                           0U, {}, RBD_PROTECTION_STATUS_PROTECTED, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, 11}, {2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotApplyImageStateRequest, SetSnapshotLimitError) {
+  InSequence seq;
+
+  MockGetMetadataRequest mock_get_metadata_request;
+  expect_get_metadata(mock_get_metadata_request, {}, 0);
+
+  expect_set_snap_limit(0, -EINVAL);
+
+  librbd::mirror::snapshot::ImageState image_state;
+  image_state.name = m_image_name;
+  image_state.features = m_remote_image_ctx->features;
+
+  C_SaferCond ctx;
+  auto req = MockApplyImageStateRequest::create(
+    "local mirror uuid", "remote mirror uuid", m_mock_local_image_ctx,
+    m_mock_remote_image_ctx, image_state, &ctx);
+  req->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+} // namespace journal
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd
diff --git a/src/test/rbd_mirror/image_replayer/snapshot/test_mock_CreateLocalImageRequest.cc b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_CreateLocalImageRequest.cc
new file mode 100644
index 000000000..58214f3e8
--- /dev/null
+++ b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_CreateLocalImageRequest.cc
@@ -0,0 +1,356 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/rbd_mirror/test_mock_fixture.h"
+#include "librbd/internal.h"
+#include "librbd/ImageState.h"
+#include "librbd/Operations.h"
+#include "tools/rbd_mirror/PoolMetaCache.h"
+#include "tools/rbd_mirror/Threads.h"
+#include "tools/rbd_mirror/image_replayer/CreateImageRequest.h"
+#include "tools/rbd_mirror/image_replayer/snapshot/CreateLocalImageRequest.h"
+#include "tools/rbd_mirror/image_replayer/snapshot/StateBuilder.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/rbd_mirror/mock/MockContextWQ.h"
+#include "test/rbd_mirror/mock/MockSafeTimer.h"
+#include <boost/intrusive_ptr.hpp>
+
+namespace librbd {
+
+namespace {
+
+struct MockTestImageCtx : public librbd::MockImageCtx {
+  explicit MockTestImageCtx(librbd::ImageCtx &image_ctx)
+    : librbd::MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace util {
+
+static std::string s_image_id;
+
+template <>
+std::string generate_image_id<MockTestImageCtx>(librados::IoCtx&) {
+  ceph_assert(!s_image_id.empty());
+  return s_image_id;
+}
+
+} // namespace util
+} // namespace librbd
+
+namespace rbd {
+namespace mirror {
+
+template <>
+struct Threads<librbd::MockTestImageCtx> {
+  ceph::mutex &timer_lock;
+  SafeTimer *timer;
+  librbd::asio::ContextWQ *work_queue;
+
+  Threads(Threads<librbd::ImageCtx> *threads)
+    : timer_lock(threads->timer_lock), timer(threads->timer),
+      work_queue(threads->work_queue) {
+  }
+};
+
+namespace image_replayer {
+
+template<>
+struct CreateImageRequest<librbd::MockTestImageCtx> {
+  static CreateImageRequest* s_instance;
+  Context *on_finish = nullptr;
+
+  static CreateImageRequest* create(Threads<librbd::MockTestImageCtx>* threads,
+                                    librados::IoCtx &local_io_ctx,
+                                    const std::string &global_image_id,
+                                    const std::string &remote_mirror_uuid,
+                                    const std::string &local_image_name,
+				    const std::string &local_image_id,
+                                    librbd::MockTestImageCtx *remote_image_ctx,
+                                    PoolMetaCache* pool_meta_cache,
+                                    cls::rbd::MirrorImageMode mirror_image_mode,
+                                    Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->on_finish = on_finish;
+    s_instance->construct(local_image_id);
+    return s_instance;
+  }
+
+  CreateImageRequest() {
+    ceph_assert(s_instance == nullptr);
+    s_instance = this;
+  }
+  ~CreateImageRequest() {
+    s_instance = nullptr;
+  }
+
+  MOCK_METHOD1(construct, void(const std::string&));
+  MOCK_METHOD0(send, void());
+};
+
+CreateImageRequest<librbd::MockTestImageCtx>*
+  CreateImageRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
+
+namespace snapshot {
+
+template<>
+struct StateBuilder<librbd::MockTestImageCtx> {
+  std::string local_image_id;
+  std::string remote_mirror_uuid;
+};
+
+} // namespace snapshot
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd
+
+#include "tools/rbd_mirror/image_replayer/snapshot/CreateLocalImageRequest.cc"
+
+using ::testing::_;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::StrEq;
+
+namespace rbd {
+namespace mirror {
+namespace image_replayer {
+namespace snapshot {
+
+class TestMockImageReplayerSnapshotCreateLocalImageRequest : public TestMockFixture {
+public:
+  typedef CreateLocalImageRequest<librbd::MockTestImageCtx> MockCreateLocalImageRequest;
+  typedef Threads<librbd::MockTestImageCtx> MockThreads;
+  typedef CreateImageRequest<librbd::MockTestImageCtx> MockCreateImageRequest;
+  typedef StateBuilder<librbd::MockTestImageCtx> MockStateBuilder;
+
+  void SetUp() override {
+    TestMockFixture::SetUp();
+
+    librbd::RBD rbd;
+    ASSERT_EQ(0, create_image(rbd, m_remote_io_ctx, m_image_name, m_image_size));
+    ASSERT_EQ(0, open_image(m_remote_io_ctx, m_image_name, &m_remote_image_ctx));
+    m_mock_remote_image_ctx = new librbd::MockTestImageCtx(*m_remote_image_ctx);
+  }
+
+  void TearDown() override {
+    delete m_mock_remote_image_ctx;
+    TestMockFixture::TearDown();
+  }
+
+  void snap_create(librbd::ImageCtx *image_ctx, const std::string &snap_name) {
+    librbd::NoOpProgressContext prog_ctx;
+    ASSERT_EQ(0, image_ctx->operations->snap_create(cls::rbd::UserSnapshotNamespace(),
+					            snap_name, 0, prog_ctx));
+    ASSERT_EQ(0, image_ctx->operations->snap_protect(cls::rbd::UserSnapshotNamespace(),
+						     snap_name));
+    ASSERT_EQ(0, image_ctx->state->refresh());
+  }
+
+  int clone_image(librbd::ImageCtx *parent_image_ctx,
+                  const std::string &snap_name, const std::string &clone_name) {
+    snap_create(parent_image_ctx, snap_name);
+
+    int order = 0;
+    return librbd::clone(m_remote_io_ctx, parent_image_ctx->name.c_str(),
+                         snap_name.c_str(), m_remote_io_ctx,
+                         clone_name.c_str(), parent_image_ctx->features,
+                         &order, 0, 0);
+  }
+
+  void expect_mirror_image_set(const std::string& image_id,
+                               const cls::rbd::MirrorImage& mirror_image,
+                               int r) {
+    bufferlist bl;
+    encode(image_id, bl);
+    encode(mirror_image, bl);
+
+    EXPECT_CALL(get_mock_io_ctx(m_local_io_ctx),
+                exec(RBD_MIRRORING, _, StrEq("rbd"),
+                     StrEq("mirror_image_set"), ContentsEqual(bl), _, _, _))
+      .WillOnce(Return(r));
+  }
+
+  void expect_mirror_image_remove(const std::string& image_id, int r) {
+    bufferlist bl;
+    encode(image_id, bl);
+
+    EXPECT_CALL(get_mock_io_ctx(m_local_io_ctx),
+                exec(StrEq("rbd_mirroring"), _, StrEq("rbd"),
+                     StrEq("mirror_image_remove"),
+                     ContentsEqual(bl), _, _, _))
+      .WillOnce(Return(r));
+  }
+
+  void expect_create_image(MockCreateImageRequest& mock_create_image_request,
+                           const std::string& image_id, int r) {
+    EXPECT_CALL(mock_create_image_request, construct(image_id));
+    EXPECT_CALL(mock_create_image_request, send())
+      .WillOnce(Invoke([this, &mock_create_image_request, r]() {
+          m_threads->work_queue->queue(mock_create_image_request.on_finish, r);
+        }));
+  }
+
+  MockCreateLocalImageRequest* create_request(
+      MockThreads& mock_threads,
+      MockStateBuilder& mock_state_builder,
+      const std::string& global_image_id,
+      Context* on_finish) {
+    return new MockCreateLocalImageRequest(
+      &mock_threads, m_local_io_ctx, m_mock_remote_image_ctx,
+      global_image_id, &m_pool_meta_cache, nullptr, &mock_state_builder,
+      on_finish);
+  }
+
+  PoolMetaCache m_pool_meta_cache{g_ceph_context};
+
+  librbd::ImageCtx *m_remote_image_ctx;
+  librbd::MockTestImageCtx *m_mock_remote_image_ctx = nullptr;
+};
+
+TEST_F(TestMockImageReplayerSnapshotCreateLocalImageRequest, Success) {
+  InSequence seq;
+
+  librbd::util::s_image_id = "local image id";
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_CREATING}, 0);
+
+  MockCreateImageRequest mock_create_image_request;
+  expect_create_image(mock_create_image_request, "local image id", 0);
+
+  C_SaferCond ctx;
+  MockThreads mock_threads(m_threads);
+  MockStateBuilder mock_state_builder;
+  auto request = create_request(
+    mock_threads, mock_state_builder, "global image id", &ctx);
+  request->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  ASSERT_EQ("local image id", mock_state_builder.local_image_id);
+}
+
+TEST_F(TestMockImageReplayerSnapshotCreateLocalImageRequest, AddMirrorImageError) {
+  InSequence seq;
+
+  librbd::util::s_image_id = "local image id";
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_CREATING}, -EINVAL);
+
+  C_SaferCond ctx;
+  MockThreads mock_threads(m_threads);
+  MockStateBuilder mock_state_builder;
+  auto request = create_request(
+    mock_threads, mock_state_builder, "global image id", &ctx);
+  request->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotCreateLocalImageRequest, CreateImageError) {
+  InSequence seq;
+
+  librbd::util::s_image_id = "local image id";
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_CREATING}, 0);
+
+  MockCreateImageRequest mock_create_image_request;
+  expect_create_image(mock_create_image_request, "local image id", -EINVAL);
+
+  C_SaferCond ctx;
+  MockThreads mock_threads(m_threads);
+  MockStateBuilder mock_state_builder;
+  auto request = create_request(
+    mock_threads, mock_state_builder, "global image id", &ctx);
+  request->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotCreateLocalImageRequest, CreateImageDuplicate) {
+  InSequence seq;
+
+  librbd::util::s_image_id = "local image id";
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_CREATING}, 0);
+
+  MockCreateImageRequest mock_create_image_request;
+  expect_create_image(mock_create_image_request, "local image id", -EBADF);
+
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_DISABLING}, 0);
+
+  expect_mirror_image_remove("local image id", 0);
+
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_CREATING}, 0);
+
+  expect_create_image(mock_create_image_request, "local image id", 0);
+
+  C_SaferCond ctx;
+  MockThreads mock_threads(m_threads);
+  MockStateBuilder mock_state_builder;
+  auto request = create_request(
+    mock_threads, mock_state_builder, "global image id", &ctx);
+  request->send();
+  ASSERT_EQ(0, ctx.wait());
+
+  ASSERT_EQ("local image id", mock_state_builder.local_image_id);
+}
+
+TEST_F(TestMockImageReplayerSnapshotCreateLocalImageRequest, DisableMirrorImageError) {
+  InSequence seq;
+
+  librbd::util::s_image_id = "local image id";
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_DISABLING}, -EINVAL);
+
+  C_SaferCond ctx;
+  MockThreads mock_threads(m_threads);
+  MockStateBuilder mock_state_builder;
+  mock_state_builder.local_image_id = "local image id";
+  auto request = create_request(
+    mock_threads, mock_state_builder, "global image id", &ctx);
+  request->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotCreateLocalImageRequest, RemoveMirrorImageError) {
+  InSequence seq;
+
+  librbd::util::s_image_id = "local image id";
+  expect_mirror_image_set("local image id",
+                          {cls::rbd::MIRROR_IMAGE_MODE_SNAPSHOT,
+                           "global image id",
+                           cls::rbd::MIRROR_IMAGE_STATE_DISABLING}, 0);
+
+  expect_mirror_image_remove("local image id", -EINVAL);
+
+  C_SaferCond ctx;
+  MockThreads mock_threads(m_threads);
+  MockStateBuilder mock_state_builder;
+  mock_state_builder.local_image_id = "local image id";
+  auto request = create_request(
+    mock_threads, mock_state_builder, "global image id", &ctx);
+  request->send();
+  ASSERT_EQ(-EINVAL, ctx.wait());
+}
+
+} // namespace journal
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd
diff --git a/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc
new file mode 100644
index 000000000..e82381b09
--- /dev/null
+++ b/src/test/rbd_mirror/image_replayer/snapshot/test_mock_Replayer.cc
@@ -0,0 +1,3320 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "test/rbd_mirror/test_mock_fixture.h"
+#include "librbd/deep_copy/ImageCopyRequest.h"
+#include "librbd/deep_copy/SnapshotCopyRequest.h"
+#include "librbd/mirror/ImageStateUpdateRequest.h"
+#include "librbd/mirror/snapshot/CreateNonPrimaryRequest.h"
+#include "librbd/mirror/snapshot/GetImageStateRequest.h"
+#include "librbd/mirror/snapshot/ImageMeta.h"
+#include "librbd/mirror/snapshot/UnlinkPeerRequest.h"
+#include "tools/rbd_mirror/InstanceWatcher.h"
+#include "tools/rbd_mirror/Threads.h"
+#include "tools/rbd_mirror/image_replayer/CloseImageRequest.h"
+#include "tools/rbd_mirror/image_replayer/ReplayerListener.h"
+#include "tools/rbd_mirror/image_replayer/Utils.h"
+#include "tools/rbd_mirror/image_replayer/snapshot/ApplyImageStateRequest.h"
+#include "tools/rbd_mirror/image_replayer/snapshot/Replayer.h"
+#include "tools/rbd_mirror/image_replayer/snapshot/StateBuilder.h"
+#include "test/librados_test_stub/MockTestMemIoCtxImpl.h"
+#include "test/librbd/mock/MockImageCtx.h"
+#include "test/librbd/mock/MockOperations.h"
+#include "test/rbd_mirror/mock/MockContextWQ.h"
+#include "test/rbd_mirror/mock/MockSafeTimer.h"
+
+namespace librbd {
+namespace {
+
+struct MockTestImageCtx : public librbd::MockImageCtx {
+  explicit MockTestImageCtx(librbd::ImageCtx &image_ctx)
+    : librbd::MockImageCtx(image_ctx) {
+  }
+};
+
+} // anonymous namespace
+
+namespace deep_copy {
+
+template <>
+struct ImageCopyRequest<MockTestImageCtx> {
+  uint64_t src_snap_id_start;
+  uint64_t src_snap_id_end;
+  uint64_t dst_snap_id_start;
+  librbd::deep_copy::ObjectNumber object_number;
+  librbd::SnapSeqs snap_seqs;
+
+  static ImageCopyRequest* s_instance;
+  static ImageCopyRequest* create(MockTestImageCtx *src_image_ctx,
+                                  MockTestImageCtx *dst_image_ctx,
+                                  librados::snap_t src_snap_id_start,
+                                  librados::snap_t src_snap_id_end,
+                                  librados::snap_t dst_snap_id_start,
+                                  bool flatten,
+                                  const ObjectNumber &object_number,
+                                  const SnapSeqs &snap_seqs,
+                                  Handler *handler,
+                                  Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->src_snap_id_start = src_snap_id_start;
+    s_instance->src_snap_id_end = src_snap_id_end;
+    s_instance->dst_snap_id_start = dst_snap_id_start;
+    s_instance->object_number = object_number;
+    s_instance->snap_seqs = snap_seqs;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  Context* on_finish = nullptr;
+
+  ImageCopyRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+template <>
+struct SnapshotCopyRequest<MockTestImageCtx> {
+  librados::snap_t src_snap_id_start;
+  librados::snap_t src_snap_id_end;
+  librados::snap_t dst_snap_id_start;
+  SnapSeqs* snap_seqs = nullptr;
+
+  static SnapshotCopyRequest* s_instance;
+  static SnapshotCopyRequest* create(MockTestImageCtx *src_image_ctx,
+                                     MockTestImageCtx *dst_image_ctx,
+                                     librados::snap_t src_snap_id_start,
+                                     librados::snap_t src_snap_id_end,
+                                     librados::snap_t dst_snap_id_start,
+                                     bool flatten,
+                                     ::MockContextWQ *work_queue,
+                                     SnapSeqs *snap_seqs,
+                                     Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->src_snap_id_start = src_snap_id_start;
+    s_instance->src_snap_id_end = src_snap_id_end;
+    s_instance->dst_snap_id_start = dst_snap_id_start;
+    s_instance->snap_seqs = snap_seqs;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  Context* on_finish = nullptr;
+
+  SnapshotCopyRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+ImageCopyRequest<MockTestImageCtx>* ImageCopyRequest<MockTestImageCtx>::s_instance = nullptr;
+SnapshotCopyRequest<MockTestImageCtx>* SnapshotCopyRequest<MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace deep_copy
+
+namespace mirror {
+
+template <>
+struct ImageStateUpdateRequest<MockTestImageCtx> {
+  static ImageStateUpdateRequest* s_instance;
+  static ImageStateUpdateRequest* create(
+      librados::IoCtx& io_ctx,
+      const std::string& image_id,
+      cls::rbd::MirrorImageState mirror_image_state,
+      const cls::rbd::MirrorImage& mirror_image,
+      Context* on_finish) {
+    ceph_assert(s_instance != nullptr);
+    EXPECT_EQ(cls::rbd::MIRROR_IMAGE_STATE_ENABLED,
+              mirror_image_state);
+    EXPECT_EQ(cls::rbd::MirrorImage{}, mirror_image);
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  Context* on_finish = nullptr;
+  ImageStateUpdateRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+ImageStateUpdateRequest<MockTestImageCtx>* ImageStateUpdateRequest<MockTestImageCtx>::s_instance = nullptr;
+
+namespace snapshot {
+
+template <>
+struct CreateNonPrimaryRequest<MockTestImageCtx> {
+  bool demoted = false;
+  std::string primary_mirror_uuid;
+  uint64_t primary_snap_id;
+  SnapSeqs snap_seqs;
+  uint64_t* snap_id = nullptr;
+
+  static CreateNonPrimaryRequest* s_instance;
+  static CreateNonPrimaryRequest* create(MockTestImageCtx *image_ctx,
+                                         bool demoted,
+                                         const std::string &primary_mirror_uuid,
+                                         uint64_t primary_snap_id,
+                                         const SnapSeqs& snap_seqs,
+                                         const ImageState &image_state,
+                                         uint64_t *snap_id,
+                                         Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->demoted = demoted;
+    s_instance->primary_mirror_uuid = primary_mirror_uuid;
+    s_instance->primary_snap_id = primary_snap_id;
+    s_instance->snap_seqs = snap_seqs;
+    s_instance->snap_id = snap_id;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  Context* on_finish = nullptr;
+
+  CreateNonPrimaryRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+template <>
+struct GetImageStateRequest<MockTestImageCtx> {
+  uint64_t snap_id = CEPH_NOSNAP;
+
+  static GetImageStateRequest* s_instance;
+  static GetImageStateRequest* create(MockTestImageCtx *image_ctx,
+                                      uint64_t snap_id,
+                                      ImageState *image_state,
+                                      Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->snap_id = snap_id;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  Context* on_finish = nullptr;
+
+  GetImageStateRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+template <>
+struct ImageMeta<MockTestImageCtx> {
+  MOCK_METHOD1(load, void(Context*));
+
+  bool resync_requested = false;
+};
+
+template <>
+struct UnlinkPeerRequest<MockTestImageCtx> {
+  uint64_t snap_id;
+  std::string mirror_peer_uuid;
+
+  static UnlinkPeerRequest* s_instance;
+  static UnlinkPeerRequest*create (MockTestImageCtx *image_ctx,
+                                   uint64_t snap_id,
+                                   const std::string &mirror_peer_uuid,
+                                   Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->snap_id = snap_id;
+    s_instance->mirror_peer_uuid = mirror_peer_uuid;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  Context* on_finish = nullptr;
+
+  UnlinkPeerRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+CreateNonPrimaryRequest<MockTestImageCtx>* CreateNonPrimaryRequest<MockTestImageCtx>::s_instance = nullptr;
+GetImageStateRequest<MockTestImageCtx>* GetImageStateRequest<MockTestImageCtx>::s_instance = nullptr;
+UnlinkPeerRequest<MockTestImageCtx>* UnlinkPeerRequest<MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace snapshot
+} // namespace mirror
+} // namespace librbd
+
+namespace rbd {
+namespace mirror {
+
+template <>
+struct InstanceWatcher<librbd::MockTestImageCtx> {
+  MOCK_METHOD1(cancel_sync_request, void(const std::string&));
+  MOCK_METHOD2(notify_sync_request, void(const std::string&,
+                                         Context*));
+  MOCK_METHOD1(notify_sync_complete, void(const std::string&));
+};
+
+template <>
+struct Threads<librbd::MockTestImageCtx> {
+  MockSafeTimer *timer;
+  ceph::mutex &timer_lock;
+
+  MockContextWQ *work_queue;
+
+  Threads(Threads<librbd::ImageCtx>* threads)
+    : timer(new MockSafeTimer()),
+      timer_lock(threads->timer_lock),
+      work_queue(new MockContextWQ()) {
+  }
+  ~Threads() {
+    delete timer;
+    delete work_queue;
+  }
+};
+
+namespace {
+
+struct MockReplayerListener : public image_replayer::ReplayerListener {
+  MOCK_METHOD0(handle_notification, void());
+};
+
+} // anonymous namespace
+
+namespace image_replayer {
+
+template<>
+struct CloseImageRequest<librbd::MockTestImageCtx> {
+  static CloseImageRequest* s_instance;
+  librbd::MockTestImageCtx **image_ctx = nullptr;
+  Context *on_finish = nullptr;
+
+  static CloseImageRequest* create(librbd::MockTestImageCtx **image_ctx,
+                                   Context *on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->image_ctx = image_ctx;
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  CloseImageRequest() {
+    ceph_assert(s_instance == nullptr);
+    s_instance = this;
+  }
+
+  ~CloseImageRequest() {
+    ceph_assert(s_instance == this);
+    s_instance = nullptr;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+CloseImageRequest<librbd::MockTestImageCtx>* CloseImageRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
+
+namespace snapshot {
+
+template <>
+struct ApplyImageStateRequest<librbd::MockTestImageCtx> {
+  Context* on_finish = nullptr;
+
+  static ApplyImageStateRequest* s_instance;
+  static ApplyImageStateRequest* create(
+      const std::string& local_mirror_uuid,
+      const std::string& remote_mirror_uuid,
+      librbd::MockTestImageCtx* local_image_ctx,
+      librbd::MockTestImageCtx* remote_image_ctx,
+      const librbd::mirror::snapshot::ImageState& image_state,
+      Context* on_finish) {
+    ceph_assert(s_instance != nullptr);
+    s_instance->on_finish = on_finish;
+    return s_instance;
+  }
+
+  ApplyImageStateRequest() {
+    s_instance = this;
+  }
+
+  MOCK_METHOD0(send, void());
+};
+
+template<>
+struct StateBuilder<librbd::MockTestImageCtx> {
+  StateBuilder(librbd::MockTestImageCtx& local_image_ctx,
+               librbd::MockTestImageCtx& remote_image_ctx,
+               librbd::mirror::snapshot::ImageMeta<librbd::MockTestImageCtx>&
+                 local_image_meta)
+    : local_image_ctx(&local_image_ctx),
+      remote_image_ctx(&remote_image_ctx),
+      local_image_meta(&local_image_meta) {
+  }
+
+  librbd::MockTestImageCtx* local_image_ctx;
+  librbd::MockTestImageCtx* remote_image_ctx;
+
+  std::string remote_mirror_uuid = "remote mirror uuid";
+
+  librbd::mirror::snapshot::ImageMeta<librbd::MockTestImageCtx>*
+    local_image_meta = nullptr;
+};
+
+ApplyImageStateRequest<librbd::MockTestImageCtx>* ApplyImageStateRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
+
+} // namespace snapshot
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd
+
+#include "tools/rbd_mirror/image_replayer/snapshot/Replayer.cc"
+
+namespace rbd {
+namespace mirror {
+namespace image_replayer {
+namespace snapshot {
+
+using ::testing::_;
+using ::testing::DoAll;
+using ::testing::InSequence;
+using ::testing::Invoke;
+using ::testing::Return;
+using ::testing::ReturnArg;
+using ::testing::StrEq;
+using ::testing::WithArg;
+
+class TestMockImageReplayerSnapshotReplayer : public TestMockFixture {
+public:
+  typedef Replayer<librbd::MockTestImageCtx> MockReplayer;
+  typedef ApplyImageStateRequest<librbd::MockTestImageCtx> MockApplyImageStateRequest;
+  typedef StateBuilder<librbd::MockTestImageCtx> MockStateBuilder;
+  typedef InstanceWatcher<librbd::MockTestImageCtx> MockInstanceWatcher;
+  typedef Threads<librbd::MockTestImageCtx> MockThreads;
+  typedef CloseImageRequest<librbd::MockTestImageCtx> MockCloseImageRequest;
+  typedef librbd::deep_copy::ImageCopyRequest<librbd::MockTestImageCtx> MockImageCopyRequest;
+  typedef librbd::deep_copy::SnapshotCopyRequest<librbd::MockTestImageCtx> MockSnapshotCopyRequest;
+  typedef librbd::mirror::ImageStateUpdateRequest<librbd::MockTestImageCtx> MockImageStateUpdateRequest;
+  typedef librbd::mirror::snapshot::CreateNonPrimaryRequest<librbd::MockTestImageCtx> MockCreateNonPrimaryRequest;
+  typedef librbd::mirror::snapshot::GetImageStateRequest<librbd::MockTestImageCtx> MockGetImageStateRequest;
+  typedef librbd::mirror::snapshot::ImageMeta<librbd::MockTestImageCtx> MockImageMeta;
+  typedef librbd::mirror::snapshot::UnlinkPeerRequest<librbd::MockTestImageCtx> MockUnlinkPeerRequest;
+
+  void SetUp() override {
+    TestMockFixture::SetUp();
+
+    librbd::RBD rbd;
+    ASSERT_EQ(0, create_image(rbd, m_local_io_ctx, m_image_name, m_image_size));
+    ASSERT_EQ(0, open_image(m_local_io_ctx, m_image_name, &m_local_image_ctx));
+
+    ASSERT_EQ(0, create_image(rbd, m_remote_io_ctx, m_image_name,
+                              m_image_size));
+    ASSERT_EQ(0, open_image(m_remote_io_ctx, m_image_name,
+                            &m_remote_image_ctx));
+  }
+
+  void expect_work_queue_repeatedly(MockThreads &mock_threads) {
+    EXPECT_CALL(*mock_threads.work_queue, queue(_, _))
+      .WillRepeatedly(Invoke([this](Context *ctx, int r) {
+          m_threads->work_queue->queue(ctx, r);
+        }));
+  }
+
+  void expect_add_event_after_repeatedly(MockThreads &mock_threads) {
+    EXPECT_CALL(*mock_threads.timer, add_event_after(_, _))
+      .WillRepeatedly(
+        DoAll(Invoke([this](double seconds, Context *ctx) {
+                       m_threads->timer->add_event_after(seconds, ctx);
+                     }),
+          ReturnArg<1>()));
+    EXPECT_CALL(*mock_threads.timer, cancel_event(_))
+      .WillRepeatedly(
+        Invoke([this](Context *ctx) {
+          return m_threads->timer->cancel_event(ctx);
+        }));
+  }
+
+  void expect_register_update_watcher(librbd::MockTestImageCtx& mock_image_ctx,
+                                      librbd::UpdateWatchCtx** update_watch_ctx,
+                                      uint64_t watch_handle, int r) {
+    EXPECT_CALL(*mock_image_ctx.state, register_update_watcher(_, _))
+      .WillOnce(Invoke([update_watch_ctx, watch_handle, r]
+                       (librbd::UpdateWatchCtx* ctx, uint64_t* handle) {
+          if (r >= 0) {
+            *update_watch_ctx = ctx;
+            *handle = watch_handle;
+          }
+          return r;
+        }));
+  }
+
+  void expect_unregister_update_watcher(librbd::MockTestImageCtx& mock_image_ctx,
+                                        uint64_t watch_handle, int r) {
+    EXPECT_CALL(*mock_image_ctx.state, unregister_update_watcher(watch_handle, _))
+      .WillOnce(WithArg<1>(Invoke([this, r](Context* ctx) {
+          m_threads->work_queue->queue(ctx, r);
+        })));
+  }
+
+  void expect_load_image_meta(MockImageMeta& mock_image_meta,
+                              bool resync_requested, int r) {
+    EXPECT_CALL(mock_image_meta, load(_))
+      .WillOnce(Invoke([this, &mock_image_meta, resync_requested, r](Context* ctx) {
+          mock_image_meta.resync_requested = resync_requested;
+          m_threads->work_queue->queue(ctx, r);
+        }));
+  }
+
+  void expect_is_refresh_required(librbd::MockTestImageCtx& mock_image_ctx,
+                                  bool is_required) {
+    EXPECT_CALL(*mock_image_ctx.state, is_refresh_required())
+      .WillOnce(Return(is_required));
+  }
+
+  void expect_refresh(librbd::MockTestImageCtx& mock_image_ctx,
+                      const std::map<uint64_t, librbd::SnapInfo>& snaps,
+                      int r) {
+    EXPECT_CALL(*mock_image_ctx.state, refresh(_))
+      .WillOnce(Invoke([this, &mock_image_ctx, snaps, r](Context* ctx) {
+        mock_image_ctx.snap_info = snaps;
+        m_threads->work_queue->queue(ctx, r);
+      }));
+  }
+
+  void expect_notify_update(librbd::MockTestImageCtx& mock_image_ctx) {
+    EXPECT_CALL(mock_image_ctx, notify_update(_))
+      .WillOnce(Invoke([this](Context* ctx) {
+        m_threads->work_queue->queue(ctx, 0);
+      }));
+  }
+
+  void expect_prune_non_primary_snapshot(librbd::MockTestImageCtx& mock_image_ctx,
+                                         uint64_t snap_id, int r) {
+    EXPECT_CALL(mock_image_ctx, get_snap_info(snap_id))
+      .WillOnce(Invoke([&mock_image_ctx](uint64_t snap_id) -> librbd::SnapInfo* {
+        auto it = mock_image_ctx.snap_info.find(snap_id);
+        if (it == mock_image_ctx.snap_info.end()) {
+          return nullptr;
+        }
+        return &it->second;
+      }));
+    EXPECT_CALL(*mock_image_ctx.operations, snap_remove(_, _, _))
+      .WillOnce(WithArg<2>(Invoke([this, r](Context* ctx) {
+        m_threads->work_queue->queue(ctx, r);
+      })));
+  }
+
+  void expect_snapshot_copy(MockSnapshotCopyRequest& mock_snapshot_copy_request,
+                            uint64_t src_snap_id_start,
+                            uint64_t src_snap_id_end,
+                            uint64_t dst_snap_id_start,
+                            const librbd::SnapSeqs& snap_seqs, int r) {
+    EXPECT_CALL(mock_snapshot_copy_request, send())
+      .WillOnce(Invoke([this, &req=mock_snapshot_copy_request,
+                        src_snap_id_start, src_snap_id_end, dst_snap_id_start,
+                        snap_seqs, r]() {
+        ASSERT_EQ(src_snap_id_start, req.src_snap_id_start);
+        ASSERT_EQ(src_snap_id_end, req.src_snap_id_end);
+        ASSERT_EQ(dst_snap_id_start, req.dst_snap_id_start);
+        *req.snap_seqs = snap_seqs;
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_get_image_state(MockGetImageStateRequest& mock_get_image_state_request,
+                              uint64_t snap_id, int r) {
+    EXPECT_CALL(mock_get_image_state_request, send())
+      .WillOnce(Invoke([this, &req=mock_get_image_state_request, snap_id, r]() {
+        ASSERT_EQ(snap_id, req.snap_id);
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_create_non_primary_request(MockCreateNonPrimaryRequest& mock_create_non_primary_request,
+                                         bool demoted,
+                                         const std::string& primary_mirror_uuid,
+                                         uint64_t primary_snap_id,
+                                         const librbd::SnapSeqs& snap_seqs,
+                                         uint64_t snap_id, int r) {
+    EXPECT_CALL(mock_create_non_primary_request, send())
+      .WillOnce(Invoke([this, &req=mock_create_non_primary_request, demoted,
+                        primary_mirror_uuid, primary_snap_id, snap_seqs,
+                        snap_id, r]() {
+        ASSERT_EQ(demoted, req.demoted);
+        ASSERT_EQ(primary_mirror_uuid, req.primary_mirror_uuid);
+        ASSERT_EQ(primary_snap_id, req.primary_snap_id);
+        ASSERT_EQ(snap_seqs, req.snap_seqs);
+        *req.snap_id = snap_id;
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_update_mirror_image_state(MockImageStateUpdateRequest& mock_image_state_update_request,
+                                        int r) {
+    EXPECT_CALL(mock_image_state_update_request, send())
+      .WillOnce(Invoke([this, &req=mock_image_state_update_request, r]() {
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_notify_sync_request(MockInstanceWatcher& mock_instance_watcher,
+                                  const std::string& image_id, int r) {
+    EXPECT_CALL(mock_instance_watcher, notify_sync_request(image_id, _))
+      .WillOnce(WithArg<1>(Invoke([this, r](Context* ctx) {
+        m_threads->work_queue->queue(ctx, r);
+      })));
+  }
+
+  void expect_notify_sync_complete(MockInstanceWatcher& mock_instance_watcher,
+                                   const std::string& image_id) {
+    EXPECT_CALL(mock_instance_watcher, notify_sync_complete(image_id));
+  }
+
+  void expect_cancel_sync_request(MockInstanceWatcher& mock_instance_watcher,
+                                  const std::string& image_id) {
+    EXPECT_CALL(mock_instance_watcher, cancel_sync_request(image_id));
+  }
+
+  void expect_image_copy(MockImageCopyRequest& mock_image_copy_request,
+                         uint64_t src_snap_id_start, uint64_t src_snap_id_end,
+                         uint64_t dst_snap_id_start,
+                         const librbd::deep_copy::ObjectNumber& object_number,
+                         const librbd::SnapSeqs& snap_seqs, int r) {
+    EXPECT_CALL(mock_image_copy_request, send())
+      .WillOnce(Invoke([this, &req=mock_image_copy_request, src_snap_id_start,
+                        src_snap_id_end, dst_snap_id_start, object_number,
+                        snap_seqs, r]() {
+        ASSERT_EQ(src_snap_id_start, req.src_snap_id_start);
+        ASSERT_EQ(src_snap_id_end, req.src_snap_id_end);
+        ASSERT_EQ(dst_snap_id_start, req.dst_snap_id_start);
+        ASSERT_EQ(object_number, req.object_number);
+        ASSERT_EQ(snap_seqs, req.snap_seqs);
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_unlink_peer(MockUnlinkPeerRequest& mock_unlink_peer_request,
+                          uint64_t snap_id, const std::string& mirror_peer_uuid,
+                          int r) {
+    EXPECT_CALL(mock_unlink_peer_request, send())
+      .WillOnce(Invoke([this, &req=mock_unlink_peer_request, snap_id,
+                        mirror_peer_uuid, r]() {
+        ASSERT_EQ(snap_id, req.snap_id);
+        ASSERT_EQ(mirror_peer_uuid, req.mirror_peer_uuid);
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_apply_image_state(
+      MockApplyImageStateRequest& mock_request, int r) {
+    EXPECT_CALL(mock_request, send())
+      .WillOnce(Invoke([this, &req=mock_request, r]() {
+        m_threads->work_queue->queue(req.on_finish, r);
+      }));
+  }
+
+  void expect_mirror_image_snapshot_set_copy_progress(
+      librbd::MockTestImageCtx& mock_test_image_ctx, uint64_t snap_id,
+      bool completed, uint64_t last_copied_object, int r) {
+    bufferlist bl;
+    encode(snap_id, bl);
+    encode(completed, bl);
+    encode(last_copied_object, bl);
+
+    EXPECT_CALL(get_mock_io_ctx(mock_test_image_ctx.md_ctx),
+                exec(mock_test_image_ctx.header_oid, _, StrEq("rbd"),
+                     StrEq("mirror_image_snapshot_set_copy_progress"),
+                     ContentsEqual(bl), _, _, _))
+      .WillOnce(Return(r));
+  }
+
+  void expect_send(MockCloseImageRequest &mock_close_image_request, int r) {
+    EXPECT_CALL(mock_close_image_request, send())
+      .WillOnce(Invoke([this, &mock_close_image_request, r]() {
+            *mock_close_image_request.image_ctx = nullptr;
+            m_threads->work_queue->queue(mock_close_image_request.on_finish, r);
+          }));
+  }
+
+  void expect_notification(MockThreads& mock_threads,
+                           MockReplayerListener& mock_replayer_listener) {
+    EXPECT_CALL(mock_replayer_listener, handle_notification())
+      .WillRepeatedly(Invoke([this]() {
+          std::unique_lock locker{m_lock};
+          ++m_notifications;
+          m_cond.notify_all();
+        }));
+  }
+
+  int wait_for_notification(uint32_t count) {
+    std::unique_lock locker{m_lock};
+    for (uint32_t idx = 0; idx < count; ++idx) {
+      while (m_notifications == 0) {
+        if (m_cond.wait_for(locker, 10s) == std::cv_status::timeout) {
+          return -ETIMEDOUT;
+        }
+      }
+      --m_notifications;
+    }
+    return 0;
+  }
+
+  int init_entry_replayer(MockReplayer& mock_replayer,
+                          MockThreads& mock_threads,
+                          librbd::MockTestImageCtx& mock_local_image_ctx,
+                          librbd::MockTestImageCtx& mock_remote_image_ctx,
+                          MockReplayerListener& mock_replayer_listener,
+                          MockImageMeta& mock_image_meta,
+                          librbd::UpdateWatchCtx** update_watch_ctx) {
+    expect_register_update_watcher(mock_local_image_ctx, update_watch_ctx, 123,
+                                   0);
+    expect_register_update_watcher(mock_remote_image_ctx, update_watch_ctx, 234,
+                                   0);
+    expect_load_image_meta(mock_image_meta, false, 0);
+    expect_is_refresh_required(mock_local_image_ctx, false);
+    expect_is_refresh_required(mock_remote_image_ctx, false);
+
+    C_SaferCond init_ctx;
+    mock_replayer.init(&init_ctx);
+    int r = init_ctx.wait();
+    if (r < 0) {
+      return r;
+    }
+
+    return wait_for_notification(2);
+  }
+
+  int shut_down_entry_replayer(MockReplayer& mock_replayer,
+                               MockThreads& mock_threads,
+                               librbd::MockTestImageCtx& mock_local_image_ctx,
+                               librbd::MockTestImageCtx& mock_remote_image_ctx) {
+    expect_unregister_update_watcher(mock_remote_image_ctx, 234, 0);
+    expect_unregister_update_watcher(mock_local_image_ctx, 123, 0);
+
+    C_SaferCond shutdown_ctx;
+    mock_replayer.shut_down(&shutdown_ctx);
+    return shutdown_ctx.wait();
+  }
+
+  librbd::ImageCtx* m_local_image_ctx = nullptr;
+  librbd::ImageCtx* m_remote_image_ctx = nullptr;
+
+  PoolMetaCache m_pool_meta_cache{g_ceph_context};
+
+  ceph::mutex m_lock = ceph::make_mutex(
+    "TestMockImageReplayerSnapshotReplayer");
+  ceph::condition_variable m_cond;
+  uint32_t m_notifications = 0;
+};
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InitShutDown) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, SyncSnapshot) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  // it should sync two snapshots and skip two (user and mirror w/o matching
+  // peer uuid)
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::UserSnapshotNamespace{},
+     0, {}, 0, 0, {}}},
+    {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {""},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {4U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+
+  // init
+  expect_register_update_watcher(mock_local_image_ctx, &update_watch_ctx, 123,
+                                 0);
+  expect_register_update_watcher(mock_remote_image_ctx, &update_watch_ctx, 234,
+                                 0);
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 11, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // sync snap4
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {{1, CEPH_NOSNAP}}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::UserSnapshotNamespace{},
+       0, {}, 0, 0, {}}},
+      {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {""},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {4U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {5U, librbd::SnapInfo{"snap5", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}}
+    }, 0);
+  expect_snapshot_copy(mock_snapshot_copy_request, 1, 4, 11,
+                       {{1, 11}, {2, 12}, {4, CEPH_NOSNAP}}, 0);
+  expect_get_image_state(mock_get_image_state_request, 4, 0);
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 4,
+                                    {{1, 11}, {2, 12}, {4, CEPH_NOSNAP}}, 14,
+                                    0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  expect_image_copy(mock_image_copy_request, 1, 4, 11, {},
+                    {{1, 11}, {2, 12}, {4, CEPH_NOSNAP}}, 0);
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 14, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // prune non-primary snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {12U, librbd::SnapInfo{"snap2", cls::rbd::UserSnapshotNamespace{},
+       0, {}, 0, 0, {}}},
+      {14U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         4, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::UserSnapshotNamespace{},
+       0, {}, 0, 0, {}}},
+      {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {""},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {4U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}}
+    }, 0);
+  expect_prune_non_primary_snapshot(mock_local_image_ctx, 11, 0);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {14U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         4, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {4U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}}
+    }, 0);
+
+  // fire init
+  C_SaferCond init_ctx;
+  mock_replayer.init(&init_ctx);
+  ASSERT_EQ(0, init_ctx.wait());
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(4));
+
+  // shut down
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedSyncInitial) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject an incomplete sync snapshot with last_copied_object_number > 0
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, false, 123, {{1, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  // re-sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0,
+                    librbd::deep_copy::ObjectNumber{123U},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 11, true, 123, 0);
+  expect_notify_update(mock_local_image_ctx);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(2));
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedSyncDelta) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject an incomplete sync snapshot with last_copied_object_number > 0
+  // after a complete snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       2, false, 123, {{2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  // re-sync snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 12, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 1, 2, 11,
+                    librbd::deep_copy::ObjectNumber{123U},
+                    {{2, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 12, true, 123, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // prune non-primary snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         2, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_prune_non_primary_snapshot(mock_local_image_ctx, 11, 0);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         2, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(2));
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedSyncDeltaDemote) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject an incomplete sync snapshot with last_copied_object_number > 0
+  // after a primary demotion snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY_DEMOTED,
+       {"remote mirror peer uuid"}, "local mirror uuid", 11, true, 0,
+       {{11, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED,
+       {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       2, false, 123, {{2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  // re-sync snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 12, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 1, 2, 11,
+                    librbd::deep_copy::ObjectNumber{123U},
+                    {{2, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 12, true, 123, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED,
+         {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         2, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(2));
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedPendingSyncInitial) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject an incomplete sync snapshot with last_copied_object_number == 0
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, false, 0, {{1, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  // re-sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 11, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 11, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(2));
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedPendingSyncDelta) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject an incomplete sync snapshot with last_copied_object_number == 0
+  // after a complete snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {{1, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       2, false, 0, {{2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  // prune non-primary snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  expect_prune_non_primary_snapshot(mock_local_image_ctx, 12, 0);
+
+  // sync snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {{1, CEPH_NOSNAP}}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 1, 2, 11,
+                       {{2, CEPH_NOSNAP}}, 0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 2, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 2,
+                                    {{2, CEPH_NOSNAP}}, 13, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 1, 2, 11, {},
+                    {{2, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 13, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // prune non-primary snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {13U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         2, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_prune_non_primary_snapshot(mock_local_image_ctx, 11, 0);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {13U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         2, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(2));
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, InterruptedPendingSyncDeltaDemote) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject an incomplete sync snapshot with last_copied_object_number == 0
+  // after a primary demotion snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY_DEMOTED,
+       {"remote mirror peer uuid"}, "local mirror uuid", 11, true, 0,
+       {{11, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED,
+       {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       2, false, 0, {{2, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+
+  // prune non-primary snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  expect_prune_non_primary_snapshot(mock_local_image_ctx, 12, 0);
+
+  // sync snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED,
+         {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 1, 2, 11,
+                       {{2, CEPH_NOSNAP}}, 0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 2, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 2,
+                                    {{2, CEPH_NOSNAP}}, 13, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 1, 2, 11, {},
+                    {{2, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 13, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED,
+         {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {13U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         2, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(2));
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RemoteImageDemoted) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject a demotion snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED,
+       {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    true, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 11, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(2));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, LocalImagePromoted) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject a promotion snapshot
+  mock_local_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY,
+       {"remote mirror peer uuid"}, "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, ResyncRequested) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // idle
+  expect_load_image_meta(mock_image_meta, true, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RegisterLocalUpdateWatcherError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayerListener mock_replayer_listener;
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  // init
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  expect_register_update_watcher(mock_local_image_ctx, &update_watch_ctx, 123,
+                                 -EINVAL);
+
+  // fire init
+  C_SaferCond init_ctx;
+  mock_replayer.init(&init_ctx);
+  ASSERT_EQ(-EINVAL, init_ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RegisterRemoteUpdateWatcherError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayerListener mock_replayer_listener;
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  // init
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  expect_register_update_watcher(mock_local_image_ctx, &update_watch_ctx, 123,
+                                 0);
+  expect_register_update_watcher(mock_remote_image_ctx, &update_watch_ctx, 234,
+                                 -EINVAL);
+
+  expect_unregister_update_watcher(mock_local_image_ctx, 123, 0);
+
+  // fire init
+  C_SaferCond init_ctx;
+  mock_replayer.init(&init_ctx);
+  ASSERT_EQ(-EINVAL, init_ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, UnregisterRemoteUpdateWatcherError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+
+  // shut down
+  expect_unregister_update_watcher(mock_remote_image_ctx, 234, -EINVAL);
+  expect_unregister_update_watcher(mock_local_image_ctx, 123, 0);
+
+  C_SaferCond shutdown_ctx;
+  mock_replayer.shut_down(&shutdown_ctx);
+  ASSERT_EQ(0, shutdown_ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, UnregisterLocalUpdateWatcherError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+
+  // shut down
+  expect_unregister_update_watcher(mock_remote_image_ctx, 234, 0);
+  expect_unregister_update_watcher(mock_local_image_ctx, 123, -EINVAL);
+
+  C_SaferCond shutdown_ctx;
+  mock_replayer.shut_down(&shutdown_ctx);
+  ASSERT_EQ(0, shutdown_ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, LoadImageMetaError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // sync
+  expect_load_image_meta(mock_image_meta, false, -EINVAL);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RefreshLocalImageError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // sync
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(mock_local_image_ctx, {}, -EINVAL);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RefreshRemoteImageError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // sync
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(mock_remote_image_ctx, {}, -EINVAL);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, CopySnapshotsError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       -EINVAL);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, GetImageStateError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, -EINVAL);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, CreateNonPrimarySnapshotError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, -EINVAL);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, UpdateMirrorImageStateError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, -EIO);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EIO, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RequestSyncError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id,
+                             -ECANCELED);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-ECANCELED, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, CopyImageError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP,true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, -EINVAL);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, UpdateNonPrimarySnapshotError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 11, true, 0, -EINVAL);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, UnlinkPeerError) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       1, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap2
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 1, 2, 11, {{2, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 2, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 2,
+                                    {{2, CEPH_NOSNAP}}, 12, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 1, 2, 11, {},
+                    {{2, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 12, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     -EINVAL);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, SplitBrain) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject a primary demote to local image
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED, {}, "", CEPH_NOSNAP,
+       true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // detect split-brain
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EEXIST, mock_replayer.get_error_code());
+  ASSERT_EQ(std::string{"split-brain"}, mock_replayer.get_error_description());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RemoteSnapshotMissingSplitBrain) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject a missing remote start snap (deleted)
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {},
+       "remote mirror uuid", 1, true, 0,
+       {{1, CEPH_NOSNAP}}},
+     0, {}, 0, 0, {}}}};
+  mock_remote_image_ctx.snap_info = {
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // split-brain due to missing snapshot 1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(-EEXIST, mock_replayer.get_error_code());
+  ASSERT_EQ(std::string{"split-brain"}, mock_replayer.get_error_description());
+
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, RemoteFailover) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject a primary demote to local image
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+     0, {}, 0, 0, {}}},
+    {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY_DEMOTED,
+       {"remote mirror peer uuid"}, "local mirror uuid", 12U, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_local_image_ctx.snap_ids = {
+    {{cls::rbd::UserSnapshotNamespace{}, "snap1"}, 11},
+    {{cls::rbd::MirrorSnapshotNamespace{}, "snap2"}, 12}};
+  mock_local_image_ctx.snap_info = {
+    {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+     0, {}, 0, 0, {}}},
+    {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED, {}, "", CEPH_NOSNAP,
+       true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // attach to promoted remote image
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 2, 3, 12,
+                       {{2, 12}, {3, CEPH_NOSNAP}}, 0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 3, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 3,
+                                    {{1, 11}, {2, 12}, {3, CEPH_NOSNAP}}, 13,
+                                    0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 2, 3, 12, {},
+                    {{1, 11}, {2, 12}, {3, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 13, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 2, "remote mirror peer uuid", 0);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+         0, {}, 0, 0, {}}},
+      {12U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY_DEMOTED, {}, "", CEPH_NOSNAP,
+         true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {13U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {},
+         "remote mirror uuid", 3, true, 0,
+         {{1, 11}, {2, 12}, {3, CEPH_NOSNAP}}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {1U, librbd::SnapInfo{"snap1", cls::rbd::UserSnapshotNamespace{},
+         0, {}, 0, 0, {}}},
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY_DEMOTED,
+         {"remote mirror peer uuid"}, "local mirror uuid", 12U, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {}, "", CEPH_NOSNAP, true, 0,
+         {}},
+       0, {}, 0, 0, {}}}
+    }, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(2));
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, UnlinkRemoteSnapshot) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  // it should attempt to unlink from remote snap1 since we don't need it
+  // anymore
+  mock_local_image_ctx.snap_info = {
+    {14U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+       4, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}},
+    {4U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+
+  // init
+  expect_register_update_watcher(mock_local_image_ctx, &update_watch_ctx, 123,
+                                 0);
+  expect_register_update_watcher(mock_remote_image_ctx, &update_watch_ctx, 234,
+                                 0);
+
+  // unlink snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockUnlinkPeerRequest mock_unlink_peer_request;
+  expect_unlink_peer(mock_unlink_peer_request, 1, "remote mirror peer uuid",
+                     0);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, true);
+  expect_refresh(
+    mock_remote_image_ctx, {
+      {2U, librbd::SnapInfo{"snap2", cls::rbd::UserSnapshotNamespace{},
+       0, {}, 0, 0, {}}},
+      {3U, librbd::SnapInfo{"snap3", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {""},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}},
+      {4U, librbd::SnapInfo{"snap4", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+         "", CEPH_NOSNAP, true, 0, {}},
+       0, {}, 0, 0, {}}}
+    }, 0);
+
+  // fire init
+  C_SaferCond init_ctx;
+  mock_replayer.init(&init_ctx);
+  ASSERT_EQ(0, init_ctx.wait());
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(3));
+
+  // shut down
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, SkipImageSync) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"},
+       "", 0U, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+
+  // init
+  expect_register_update_watcher(mock_local_image_ctx, &update_watch_ctx, 123,
+                                 0);
+  expect_register_update_watcher(mock_remote_image_ctx, &update_watch_ctx, 234,
+                                 0);
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  expect_apply_image_state(mock_apply_state_request, 0);
+  expect_mirror_image_snapshot_set_copy_progress(
+    mock_local_image_ctx, 11, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+
+  // idle
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, true);
+  expect_refresh(
+    mock_local_image_ctx, {
+      {11U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+         cls::rbd::MIRROR_SNAPSHOT_STATE_NON_PRIMARY, {}, "remote mirror uuid",
+         1, true, 0, {{1, CEPH_NOSNAP}}},
+       0, {}, 0, 0, {}}},
+    }, 0);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+
+  // fire init
+  C_SaferCond init_ctx;
+  mock_replayer.init(&init_ctx);
+  ASSERT_EQ(0, init_ctx.wait());
+
+  // wait for sync to complete
+  ASSERT_EQ(0, wait_for_notification(3));
+
+  // shut down
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, ImageNameUpdated) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // change the name of the image
+  mock_local_image_ctx.name = "NEW NAME";
+
+  // idle
+  expect_load_image_meta(mock_image_meta, true, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  // wait for sync to complete and expect replay complete
+  ASSERT_EQ(0, wait_for_notification(2));
+  auto image_spec = image_replayer::util::compute_image_spec(m_local_io_ctx,
+                                                             "NEW NAME");
+  ASSERT_EQ(image_spec, mock_replayer.get_image_spec());
+  ASSERT_FALSE(mock_replayer.is_replaying());
+
+  // shut down
+  ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
+                                        mock_local_image_ctx,
+                                        mock_remote_image_ctx));
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, ApplyImageStatePendingShutdown) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  C_SaferCond shutdown_ctx;
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  EXPECT_CALL(mock_apply_state_request, send())
+    .WillOnce(Invoke([this, &req=mock_apply_state_request,
+                      &replayer=mock_replayer, &ctx=shutdown_ctx]() {
+      // inject a shutdown, to be pended due to STATE_REPLAYING
+      replayer.shut_down(&ctx);
+      m_threads->work_queue->queue(req.on_finish, 0);
+    }));
+  expect_cancel_sync_request(mock_instance_watcher, mock_local_image_ctx.id);
+  expect_mirror_image_snapshot_set_copy_progress(
+      mock_local_image_ctx, 11, true, 0, 0);
+  expect_notify_update(mock_local_image_ctx);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // shutdown should be resumed
+  expect_unregister_update_watcher(mock_remote_image_ctx, 234, 0);
+  expect_unregister_update_watcher(mock_local_image_ctx, 123, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  ASSERT_EQ(0, wait_for_notification(1));
+  ASSERT_FALSE(mock_replayer.is_replaying());
+  ASSERT_EQ(0, mock_replayer.get_error_code());
+
+  ASSERT_EQ(0, shutdown_ctx.wait());
+}
+
+TEST_F(TestMockImageReplayerSnapshotReplayer, ApplyImageStateErrorPendingShutdown) {
+  librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx};
+  librbd::MockTestImageCtx mock_remote_image_ctx{*m_remote_image_ctx};
+
+  MockThreads mock_threads(m_threads);
+  expect_work_queue_repeatedly(mock_threads);
+
+  MockReplayerListener mock_replayer_listener;
+  expect_notification(mock_threads, mock_replayer_listener);
+
+  InSequence seq;
+
+  MockInstanceWatcher mock_instance_watcher;
+  MockImageMeta mock_image_meta;
+  MockStateBuilder mock_state_builder(mock_local_image_ctx,
+                                      mock_remote_image_ctx,
+                                      mock_image_meta);
+  MockReplayer mock_replayer{&mock_threads, &mock_instance_watcher,
+                             "local mirror uuid", &m_pool_meta_cache,
+                             &mock_state_builder, &mock_replayer_listener};
+  C_SaferCond shutdown_ctx;
+  m_pool_meta_cache.set_remote_pool_meta(
+    m_remote_io_ctx.get_id(),
+    {"remote mirror uuid", "remote mirror peer uuid"});
+
+  librbd::UpdateWatchCtx* update_watch_ctx = nullptr;
+  ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
+                                   mock_local_image_ctx,
+                                   mock_remote_image_ctx,
+                                   mock_replayer_listener,
+                                   mock_image_meta,
+                                   &update_watch_ctx));
+
+  // inject snapshot
+  mock_remote_image_ctx.snap_info = {
+    {1U, librbd::SnapInfo{"snap1", cls::rbd::MirrorSnapshotNamespace{
+       cls::rbd::MIRROR_SNAPSHOT_STATE_PRIMARY, {"remote mirror peer uuid"}, "",
+       CEPH_NOSNAP, true, 0, {}},
+     0, {}, 0, 0, {}}}};
+
+  // sync snap1
+  expect_load_image_meta(mock_image_meta, false, 0);
+  expect_is_refresh_required(mock_local_image_ctx, false);
+  expect_is_refresh_required(mock_remote_image_ctx, false);
+  MockSnapshotCopyRequest mock_snapshot_copy_request;
+  expect_snapshot_copy(mock_snapshot_copy_request, 0, 1, 0, {{1, CEPH_NOSNAP}},
+                       0);
+  MockGetImageStateRequest mock_get_image_state_request;
+  expect_get_image_state(mock_get_image_state_request, 1, 0);
+  MockCreateNonPrimaryRequest mock_create_non_primary_request;
+  expect_create_non_primary_request(mock_create_non_primary_request,
+                                    false, "remote mirror uuid", 1,
+                                    {{1, CEPH_NOSNAP}}, 11, 0);
+  MockImageStateUpdateRequest mock_image_state_update_request;
+  expect_update_mirror_image_state(mock_image_state_update_request, 0);
+  expect_notify_sync_request(mock_instance_watcher, mock_local_image_ctx.id, 0);
+  MockImageCopyRequest mock_image_copy_request;
+  expect_image_copy(mock_image_copy_request, 0, 1, 0, {},
+                    {{1, CEPH_NOSNAP}}, 0);
+  MockApplyImageStateRequest mock_apply_state_request;
+  EXPECT_CALL(mock_apply_state_request, send())
+    .WillOnce(Invoke([this, &req=mock_apply_state_request,
+                      &replayer=mock_replayer, &ctx=shutdown_ctx]() {
+      // inject a shutdown, to be pended due to STATE_REPLAYING
+      replayer.shut_down(&ctx);
+      m_threads->work_queue->queue(req.on_finish, -EINVAL);
+    }));
+  expect_cancel_sync_request(mock_instance_watcher, mock_local_image_ctx.id);
+  expect_notify_sync_complete(mock_instance_watcher, mock_local_image_ctx.id);
+
+  // shutdown should be resumed
+  expect_unregister_update_watcher(mock_remote_image_ctx, 234, 0);
+  expect_unregister_update_watcher(mock_local_image_ctx, 123, 0);
+
+  // wake-up replayer
+  update_watch_ctx->handle_notify();
+
+  ASSERT_EQ(0, shutdown_ctx.wait());
+}
+
+} // namespace snapshot
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd