18 files changed, 558 insertions, 290 deletions

diff --git a/src/tools/rbd_mirror/CMakeLists.txt b/src/tools/rbd_mirror/CMakeLists.txt
index 3fb0536c762..1a9d0fb1aab 100644
--- a/src/tools/rbd_mirror/CMakeLists.txt
+++ b/src/tools/rbd_mirror/CMakeLists.txt
@@ -25,6 +25,7 @@ set(rbd_mirror_internal
   image_replayer/IsPrimaryRequest.cc
   image_replayer/OpenImageRequest.cc
   image_replayer/OpenLocalImageRequest.cc
+  image_replayer/PrepareLocalImageRequest.cc
   image_replayer/ReplayStatusFormatter.cc
   image_sync/ImageCopyRequest.cc
   image_sync/ObjectCopyRequest.cc
diff --git a/src/tools/rbd_mirror/ImageDeleter.cc b/src/tools/rbd_mirror/ImageDeleter.cc
index 00c4c3952f2..b2eb615a7e4 100644
--- a/src/tools/rbd_mirror/ImageDeleter.cc
+++ b/src/tools/rbd_mirror/ImageDeleter.cc
@@ -134,7 +134,7 @@ private:
 
 ImageDeleter::ImageDeleter(ContextWQ *work_queue, SafeTimer *timer,
                            Mutex *timer_lock)
-  : m_running(1),
+  : m_running(true),
     m_work_queue(work_queue),
     m_delete_lock("rbd::mirror::ImageDeleter::Delete"),
     m_image_deleter_thread(this),
@@ -149,7 +149,7 @@ ImageDeleter::ImageDeleter(ContextWQ *work_queue, SafeTimer *timer,
 ImageDeleter::~ImageDeleter() {
   dout(20) << "enter" << dendl;
 
-  m_running.set(0);
+  m_running = false;
   {
     Mutex::Locker l (m_delete_lock);
     m_delete_queue_cond.Signal();
@@ -164,13 +164,13 @@ ImageDeleter::~ImageDeleter() {
 
 void ImageDeleter::run() {
   dout(20) << "enter" << dendl;
-  while(m_running.read()) {
+  while(m_running) {
     m_delete_lock.Lock();
     while (m_delete_queue.empty()) {
       dout(20) << "waiting for delete requests" << dendl;
       m_delete_queue_cond.Wait(m_delete_lock);
 
-      if (!m_running.read()) {
+      if (!m_running) {
         m_delete_lock.Unlock();
         dout(20) << "return" << dendl;
         return;
@@ -183,7 +183,7 @@ void ImageDeleter::run() {
 
     bool move_to_next = process_image_delete();
     if (!move_to_next) {
-      if (!m_running.read()) {
+      if (!m_running) {
        dout(20) << "return" << dendl;
        return;
       }
@@ -198,7 +198,6 @@ void ImageDeleter::run() {
 
 void ImageDeleter::schedule_image_delete(RadosRef local_rados,
                                          int64_t local_pool_id,
-                                         const std::string& local_image_id,
                                          const std::string& global_image_id) {
   dout(20) << "enter" << dendl;
 
@@ -206,14 +205,14 @@ void ImageDeleter::schedule_image_delete(RadosRef local_rados,
 
   auto del_info = find_delete_info(local_pool_id, global_image_id);
   if (del_info != nullptr) {
-    dout(20) << "image " << local_image_id << " (" << global_image_id << ") "
+    dout(20) << "image " << global_image_id << " "
              << "was already scheduled for deletion" << dendl;
     return;
   }
 
-  m_delete_queue.push_front(unique_ptr<DeleteInfo>(
-        new DeleteInfo(local_rados, local_pool_id, local_image_id,
-                       global_image_id)));
+  m_delete_queue.push_front(
+    unique_ptr<DeleteInfo>(new DeleteInfo(local_rados, local_pool_id,
+                                          global_image_id)));
   m_delete_queue_cond.Signal();
 }
 
@@ -272,25 +271,42 @@ bool ImageDeleter::process_image_delete() {
   r = m_active_delete->local_rados->ioctx_create2(
     m_active_delete->local_pool_id, ioctx);
   if (r < 0) {
-    derr << "error accessing local pool: " << cpp_strerror(r) << dendl;
+    derr << "error accessing local pool " << m_active_delete->local_pool_id
+         << ": " << cpp_strerror(r) << dendl;
     enqueue_failed_delete(r);
     return true;
   }
 
   dout(20) << "connected to local pool: " << ioctx.get_pool_name() << dendl;
 
+  auto &global_image_id = m_active_delete->global_image_id;
+  std::string local_image_id;
+  r = librbd::cls_client::mirror_image_get_image_id(
+    &ioctx, global_image_id, &local_image_id);
+  if (r == -ENOENT) {
+    dout(10) << "image " << global_image_id << " is not mirrored" << dendl;
+    complete_active_delete(r);
+    return true;
+  } else if (r < 0) {
+    derr << "error retrieving local id for image " << global_image_id
+         << ": " << cpp_strerror(r) << dendl;
+    enqueue_failed_delete(r);
+    return true;
+  }
+
   bool is_primary = false;
-  r = Journal<>::is_tag_owner(ioctx, m_active_delete->local_image_id,
-                              &is_primary, m_work_queue);
+  C_SaferCond tag_owner_ctx;
+  Journal<>::is_tag_owner(ioctx, local_image_id, &is_primary,
+                          m_work_queue, &tag_owner_ctx);
+  r = tag_owner_ctx.wait();
   if (r < 0 && r != -ENOENT) {
-    derr << "error retrieving image primary info: " << cpp_strerror(r)
-         << dendl;
+    derr << "error retrieving image primary info for image " << global_image_id
+         << ": " << cpp_strerror(r) << dendl;
     enqueue_failed_delete(r);
     return true;
   }
   if (is_primary) {
-    dout(10) << "local image is the primary image, aborting deletion..."
-             << dendl;
+    dout(10) << "image " << global_image_id << " is local primary" << dendl;
     complete_active_delete(-EISPRM);
     return true;
   }
@@ -298,31 +314,28 @@ bool ImageDeleter::process_image_delete() {
   dout(20) << "local image is not the primary" << dendl;
 
   bool has_snapshots;
-  r = image_has_snapshots_and_children(&ioctx, m_active_delete->local_image_id,
-                                       &has_snapshots);
+  r = image_has_snapshots_and_children(&ioctx, local_image_id, &has_snapshots);
   if (r < 0) {
     enqueue_failed_delete(r);
     return true;
   }
 
-  mirror_image.global_image_id = m_active_delete->global_image_id;
+  mirror_image.global_image_id = global_image_id;
   mirror_image.state = cls::rbd::MIRROR_IMAGE_STATE_DISABLING;
-  r = cls_client::mirror_image_set(&ioctx, m_active_delete->local_image_id,
-                                   mirror_image);
+  r = cls_client::mirror_image_set(&ioctx, local_image_id, mirror_image);
   if (r == -ENOENT) {
     dout(10) << "local image is not mirrored, aborting deletion..." << dendl;
     complete_active_delete(r);
     return true;
   } else if (r == -EEXIST || r == -EINVAL) {
-    derr << "cannot disable mirroring for image id "
-         << m_active_delete->local_image_id
-         << ": global_image_id has changed/reused, aborting deletion: "
+    derr << "cannot disable mirroring for image " << global_image_id
+         << ": global_image_id has changed/reused: "
          << cpp_strerror(r) << dendl;
     complete_active_delete(r);
     return true;
   } else if (r < 0) {
-    derr << "cannot disable mirroring for image id "
-         << m_active_delete->local_image_id << ": " << cpp_strerror(r) << dendl;
+    derr << "cannot disable mirroring for image " << global_image_id
+         << ": " << cpp_strerror(r) << dendl;
     enqueue_failed_delete(r);
     return true;
   }
@@ -332,12 +345,11 @@ bool ImageDeleter::process_image_delete() {
   if (has_snapshots) {
     dout(20) << "local image has snapshots" << dendl;
 
-    ImageCtx *imgctx = new ImageCtx("", m_active_delete->local_image_id,
-                                    nullptr, ioctx, false);
+    ImageCtx *imgctx = new ImageCtx("", local_image_id, nullptr, ioctx, false);
     r = imgctx->state->open(false);
     if (r < 0) {
-      derr << "error opening image id " << m_active_delete->local_image_id
-           << ": " << cpp_strerror(r) << dendl;
+      derr << "error opening image " << global_image_id << " ("
+           << local_image_id << "): " << cpp_strerror(r) << dendl;
       enqueue_failed_delete(r);
       return true;
     }
@@ -375,12 +387,12 @@ bool ImageDeleter::process_image_delete() {
         dout(20) << "snapshot " << imgctx->name << "@" << snap.name
                  << " is protected, issuing unprotect command" << dendl;
 
-        r = imgctx->operations->snap_unprotect(cls::rbd::UserSnapshotNamespace(),
-					       snap.name.c_str());
+        r = imgctx->operations->snap_unprotect(
+          cls::rbd::UserSnapshotNamespace(), snap.name.c_str());
         if (r == -EBUSY) {
           // there are still clones of snapshots of this image, therefore send
           // the delete request to the end of the queue
-          dout(10) << "local image id " << m_active_delete->local_image_id << " has "
+          dout(10) << "local image id " << local_image_id << " has "
                    << "snapshots with cloned children, postponing deletion..."
                    << dendl;
           imgctx->state->close();
@@ -415,10 +427,10 @@ bool ImageDeleter::process_image_delete() {
   }
 
   librbd::NoOpProgressContext ctx;
-  r = librbd::remove(ioctx, "", m_active_delete->local_image_id, ctx, true);
+  r = librbd::remove(ioctx, "", local_image_id, ctx, true);
   if (r < 0 && r != -ENOENT) {
-    derr << "error removing image " << m_active_delete->local_image_id << " "
-         << "(" << m_active_delete->global_image_id << ") from local pool: "
+    derr << "error removing image " << global_image_id << " "
+         << "(" << local_image_id << ") from local pool: "
          << cpp_strerror(r) << dendl;
     enqueue_failed_delete(r);
     return true;
@@ -431,7 +443,7 @@ bool ImageDeleter::process_image_delete() {
              << dendl;
   }
 
-  r = cls_client::mirror_image_remove(&ioctx, m_active_delete->local_image_id);
+  r = cls_client::mirror_image_remove(&ioctx, local_image_id);
   if (r < 0 && r != -ENOENT) {
     derr << "error removing image from mirroring directory: "
          << cpp_strerror(r) << dendl;
@@ -440,8 +452,7 @@ bool ImageDeleter::process_image_delete() {
   }
 
   dout(10) << "Successfully deleted image "
-           << m_active_delete->local_image_id << " "
-           << "(" << m_active_delete->global_image_id << ")" << dendl;
+           << global_image_id << " " << "(" << local_image_id << ")" << dendl;
 
   complete_active_delete(0);
   return true;
@@ -581,7 +592,6 @@ void ImageDeleter::DeleteInfo::notify(int r) {
 
 void ImageDeleter::DeleteInfo::to_string(stringstream& ss) {
   ss << "[" << "local_pool_id=" << local_pool_id << ", ";
-  ss << "local_image_id=" << local_image_id << ", ";
   ss << "global_image_id=" << global_image_id << "]";
 }
 
@@ -590,7 +600,6 @@ void ImageDeleter::DeleteInfo::print_status(Formatter *f, stringstream *ss,
   if (f) {
     f->open_object_section("delete_info");
     f->dump_int("local_pool_id", local_pool_id);
-    f->dump_string("local_image_id", local_image_id);
     f->dump_string("global_image_id", global_image_id);
     if (print_failure_info) {
       f->dump_string("error_code", cpp_strerror(error_code));
@@ -608,7 +617,7 @@ vector<string> ImageDeleter::get_delete_queue_items() {
 
   Mutex::Locker l(m_delete_lock);
   for (const auto& del_info : m_delete_queue) {
-    items.push_back(del_info->local_image_id);
+    items.push_back(del_info->global_image_id);
   }
 
   return items;
@@ -619,7 +628,7 @@ vector<pair<string, int> > ImageDeleter::get_failed_queue_items() {
 
   Mutex::Locker l(m_delete_lock);
   for (const auto& del_info : m_failed_queue) {
-    items.push_back(make_pair(del_info->local_image_id,
+    items.push_back(make_pair(del_info->global_image_id,
                               del_info->error_code));
   }
 
diff --git a/src/tools/rbd_mirror/ImageDeleter.h b/src/tools/rbd_mirror/ImageDeleter.h
index ff5f4e535f4..c91f2fbc6ff 100644
--- a/src/tools/rbd_mirror/ImageDeleter.h
+++ b/src/tools/rbd_mirror/ImageDeleter.h
@@ -15,15 +15,16 @@
 #ifndef CEPH_RBD_MIRROR_IMAGEDELETER_H
 #define CEPH_RBD_MIRROR_IMAGEDELETER_H
 
-#include <deque>
-#include <vector>
-#include "include/atomic.h"
 #include "common/Mutex.h"
 #include "common/Cond.h"
 #include "common/Thread.h"
 #include "common/Timer.h"
 #include "types.h"
 
+#include <deque>
+#include <vector>
+#include <atomic>
+
 class ContextWQ;
 
 namespace rbd {
@@ -45,7 +46,6 @@ public:
 
   void schedule_image_delete(RadosRef local_rados,
                              int64_t local_pool_id,
-                             const std::string& local_image_id,
                              const std::string& global_image_id);
   void wait_for_scheduled_deletion(int64_t local_pool_id,
                                    const std::string &global_image_id,
@@ -77,20 +77,16 @@ private:
   struct DeleteInfo {
     RadosRef local_rados;
     int64_t local_pool_id;
-    std::string local_image_id;
     std::string global_image_id;
-    int error_code;
-    int retries;
-    bool notify_on_failed_retry;
-    Context *on_delete;
+    int error_code = 0;
+    int retries = 0;
+    bool notify_on_failed_retry = true;
+    Context *on_delete = nullptr;
 
     DeleteInfo(RadosRef local_rados, int64_t local_pool_id,
-               const std::string& local_image_id,
                const std::string& global_image_id) :
       local_rados(local_rados), local_pool_id(local_pool_id),
-      local_image_id(local_image_id), global_image_id(global_image_id),
-      error_code(0), retries(0), notify_on_failed_retry(true),
-      on_delete(nullptr) {
+      global_image_id(global_image_id) {
     }
 
     bool match(int64_t local_pool_id, const std::string &global_image_id) {
@@ -103,7 +99,7 @@ private:
                       bool print_failure_info=false);
   };
 
-  atomic_t m_running;
+  std::atomic<unsigned> m_running { 0 };
 
   ContextWQ *m_work_queue;
 
diff --git a/src/tools/rbd_mirror/ImageReplayer.cc b/src/tools/rbd_mirror/ImageReplayer.cc
index fa4cb470b1f..3276dba3b8f 100644
--- a/src/tools/rbd_mirror/ImageReplayer.cc
+++ b/src/tools/rbd_mirror/ImageReplayer.cc
@@ -26,6 +26,7 @@
 #include "tools/rbd_mirror/image_replayer/BootstrapRequest.h"
 #include "tools/rbd_mirror/image_replayer/CloseImageRequest.h"
 #include "tools/rbd_mirror/image_replayer/EventPreprocessor.h"
+#include "tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.h"
 #include "tools/rbd_mirror/image_replayer/ReplayStatusFormatter.h"
 
 #define dout_context g_ceph_context
@@ -335,7 +336,8 @@ void ImageReplayer<I>::add_remote_image(const std::string &mirror_uuid,
 
 template <typename I>
 void ImageReplayer<I>::remove_remote_image(const std::string &mirror_uuid,
-                                           const std::string &image_id) {
+                                           const std::string &image_id,
+					   bool schedule_delete) {
   Mutex::Locker locker(m_lock);
   m_remote_images.erase({mirror_uuid, image_id});
 }
@@ -370,18 +372,12 @@ void ImageReplayer<I>::start(Context *on_finish, bool manual)
       dout(5) << "stopped manually, ignoring start without manual flag"
 	      << dendl;
       r = -EPERM;
-    } else if (m_remote_images.empty()) {
-      derr << "no remote images associated with replayer" << dendl;
-      r = -EINVAL;
     } else {
       m_state = STATE_STARTING;
       m_last_r = 0;
       m_state_desc.clear();
       m_manual_stop = false;
 
-      // TODO bootstrap will need to support multiple remote images
-      m_remote_image = *m_remote_images.begin();
-
       if (on_finish != nullptr) {
         assert(m_on_start_finish == nullptr);
         m_on_start_finish = on_finish;
@@ -405,6 +401,52 @@ void ImageReplayer<I>::start(Context *on_finish, bool manual)
     return;
   }
 
+  prepare_local_image();
+}
+
+template <typename I>
+void ImageReplayer<I>::prepare_local_image() {
+  dout(20) << dendl;
+
+  Context *ctx = create_context_callback<
+    ImageReplayer, &ImageReplayer<I>::handle_prepare_local_image>(this);
+  auto req = PrepareLocalImageRequest<I>::create(
+    m_local_ioctx, m_global_image_id, &m_local_image_id,
+    &m_local_image_tag_owner, m_threads->work_queue, ctx);
+  req->send();
+}
+
+template <typename I>
+void ImageReplayer<I>::handle_prepare_local_image(int r) {
+  dout(20) << "r=" << r << dendl;
+
+  if (r == -ENOENT) {
+    dout(20) << "local image does not exist" << dendl;
+  } else if (r < 0) {
+    on_start_fail(r, "error preparing local image for replay");
+    return;
+  } else if (m_local_image_tag_owner == librbd::Journal<>::LOCAL_MIRROR_UUID) {
+    dout(5) << "local image is primary" << dendl;
+    on_start_fail(0, "local image is primary");
+    return;
+  }
+
+  // local image doesn't exist or is non-primary
+  bootstrap();
+}
+
+template <typename I>
+void ImageReplayer<I>::bootstrap() {
+  dout(20) << dendl;
+
+  if (m_remote_images.empty()) {
+    on_start_fail(0, "waiting for primary remote image");
+    return;
+  }
+
+  // TODO bootstrap will need to support multiple remote images
+  m_remote_image = *m_remote_images.begin();
+
   CephContext *cct = static_cast<CephContext *>(m_local->cct());
   journal::Settings settings;
   settings.commit_interval = cct->_conf->rbd_mirror_journal_commit_age;
@@ -412,16 +454,10 @@ void ImageReplayer<I>::start(Context *on_finish, bool manual)
 
   m_remote_journaler = new Journaler(m_threads->work_queue,
                                      m_threads->timer,
-				     &m_threads->timer_lock,
+                                     &m_threads->timer_lock,
                                      m_remote_image.io_ctx,
                                      m_remote_image.image_id,
                                      m_local_mirror_uuid, settings);
-  bootstrap();
-}
-
-template <typename I>
-void ImageReplayer<I>::bootstrap() {
-  dout(20) << dendl;
 
   Context *ctx = create_context_callback<
     ImageReplayer, &ImageReplayer<I>::handle_bootstrap>(this);
@@ -458,10 +494,12 @@ void ImageReplayer<I>::handle_bootstrap(int r) {
   }
 
   if (r == -EREMOTEIO) {
+    m_local_image_tag_owner = "";
     dout(5) << "remote image is non-primary or local image is primary" << dendl;
     on_start_fail(0, "remote image is non-primary or local image is primary");
     return;
   } else if (r == -EEXIST) {
+    m_local_image_tag_owner = "";
     on_start_fail(r, "split-brain detected");
     return;
   } else if (r < 0) {
@@ -1511,7 +1549,6 @@ void ImageReplayer<I>::handle_shut_down(int r) {
     if (m_stopping_for_resync) {
       m_image_deleter->schedule_image_delete(m_local,
                                              m_local_pool_id,
-                                             m_local_image_id,
                                              m_global_image_id);
       m_stopping_for_resync = false;
     }
diff --git a/src/tools/rbd_mirror/ImageReplayer.h b/src/tools/rbd_mirror/ImageReplayer.h
index 79a744d9842..72f03f77b6c 100644
--- a/src/tools/rbd_mirror/ImageReplayer.h
+++ b/src/tools/rbd_mirror/ImageReplayer.h
@@ -4,11 +4,6 @@
 #ifndef CEPH_RBD_MIRROR_IMAGE_REPLAYER_H
 #define CEPH_RBD_MIRROR_IMAGE_REPLAYER_H
 
-#include <map>
-#include <string>
-#include <vector>
-
-#include "include/atomic.h"
 #include "common/AsyncOpTracker.h"
 #include "common/Mutex.h"
 #include "common/WorkQueue.h"
@@ -23,10 +18,16 @@
 #include "ImageDeleter.h"
 #include "ProgressContext.h"
 #include "types.h"
-#include <set>
+
 #include <boost/noncopyable.hpp>
 #include <boost/optional.hpp>
 
+#include <set>
+#include <map>
+#include <atomic>
+#include <string>
+#include <vector>
+
 class AdminSocketHook;
 
 namespace journal {
@@ -109,7 +110,8 @@ public:
                         const std::string &remote_image_id,
                         librados::IoCtx &remote_io_ctx);
   void remove_remote_image(const std::string &remote_mirror_uuid,
-                           const std::string &remote_image_id);
+                           const std::string &remote_image_id,
+			   bool schedule_delete);
   bool remote_images_empty() const;
 
   inline int64_t get_local_pool_id() const {
@@ -118,10 +120,6 @@ public:
   inline const std::string& get_global_image_id() const {
     return m_global_image_id;
   }
-  inline std::string get_local_image_id() {
-    Mutex::Locker locker(m_lock);
-    return m_local_image_id;
-  }
 
   void start(Context *on_finish = nullptr, bool manual = false);
   void stop(Context *on_finish = nullptr, bool manual = false,
@@ -146,6 +144,9 @@ protected:
    * <starting>                                             *
    *    |                                                   *
    *    v                                           (error) *
+   * PREPARE_LOCAL_IMAGE  * * * * * * * * * * * * * * * * * *
+   *    |                                                   *
+   *    v                                           (error) *
    * BOOTSTRAP_IMAGE  * * * * * * * * * * * * * * * * * * * *
    *    |                                                   *
    *    v                                           (error) *
@@ -304,6 +305,7 @@ private:
     nullptr;
   librados::IoCtx m_local_ioctx;
   ImageCtxT *m_local_image_ctx = nullptr;
+  std::string m_local_image_tag_owner;
 
   decltype(ImageCtxT::journal) m_local_journal = nullptr;
   librbd::journal::Replay<ImageCtxT> *m_local_replay = nullptr;
@@ -388,6 +390,9 @@ private:
   void handle_shut_down(int r);
   void handle_remote_journal_metadata_updated();
 
+  void prepare_local_image();
+  void handle_prepare_local_image(int r);
+
   void bootstrap();
   void handle_bootstrap(int r);
 
diff --git a/src/tools/rbd_mirror/InstanceReplayer.cc b/src/tools/rbd_mirror/InstanceReplayer.cc
index 359e2c66bbb..d2426d0d0e4 100644
--- a/src/tools/rbd_mirror/InstanceReplayer.cc
+++ b/src/tools/rbd_mirror/InstanceReplayer.cc
@@ -156,13 +156,13 @@ void InstanceReplayer<I>::acquire_image(const std::string &global_image_id,
   }
 
   auto image_replayer = it->second;
+  if (!peer_mirror_uuid.empty()) {
+    auto iter = m_peers.find(Peer(peer_mirror_uuid));
+    assert(iter != m_peers.end());
+    auto io_ctx = iter->io_ctx;
 
-  auto iter = m_peers.find(Peer(peer_mirror_uuid));
-  assert(iter != m_peers.end());
-  auto io_ctx = iter->io_ctx;
-
-  image_replayer->add_remote_image(peer_mirror_uuid, peer_image_id, io_ctx);
-
+    image_replayer->add_remote_image(peer_mirror_uuid, peer_image_id, io_ctx);
+  }
   start_image_replayer(image_replayer);
 
   m_threads->work_queue->queue(on_finish, 0);
@@ -190,11 +190,13 @@ void InstanceReplayer<I>::release_image(const std::string &global_image_id,
   }
 
   auto image_replayer = it->second;
-
-  image_replayer->remove_remote_image(peer_mirror_uuid, peer_image_id);
+  if (!peer_mirror_uuid.empty()) {
+    image_replayer->remove_remote_image(peer_mirror_uuid, peer_image_id,
+					schedule_delete);
+  }
 
   if (!image_replayer->remote_images_empty()) {
-    dout(20) << global_image_id << ": still has remote images" << dendl;
+    dout(20) << global_image_id << ": still has peer images" << dendl;
     m_threads->work_queue->queue(on_finish, 0);
     return;
   }
@@ -211,15 +213,8 @@ void InstanceReplayer<I>::release_image(const std::string &global_image_id,
     on_finish = new FunctionContext(
       [this, image_replayer, on_finish] (int r) {
         auto global_image_id = image_replayer->get_global_image_id();
-        auto local_image_id = image_replayer->get_local_image_id();
-        if (local_image_id.empty()) {
-          dout(20) << global_image_id << ": unknown local_image_id"
-                   << " (image does not exist or primary), skipping delete"
-                   << dendl;
-        } else {
-          m_image_deleter->schedule_image_delete(
-            m_local_rados, m_local_pool_id, local_image_id, global_image_id);
-        }
+        m_image_deleter->schedule_image_delete(
+          m_local_rados, m_local_pool_id, global_image_id);
         on_finish->complete(0);
       });
   }
diff --git a/src/tools/rbd_mirror/Mirror.cc b/src/tools/rbd_mirror/Mirror.cc
index 86c6939182f..f37d2559922 100644
--- a/src/tools/rbd_mirror/Mirror.cc
+++ b/src/tools/rbd_mirror/Mirror.cc
@@ -215,7 +215,7 @@ Mirror::~Mirror()
 
 void Mirror::handle_signal(int signum)
 {
-  m_stopping.set(1);
+  m_stopping = true;
   {
     Mutex::Locker l(m_lock);
     m_cond.Signal();
@@ -250,7 +250,7 @@ int Mirror::init()
 void Mirror::run()
 {
   dout(20) << "enter" << dendl;
-  while (!m_stopping.read()) {
+  while (!m_stopping) {
     m_local_cluster_watcher->refresh_pools();
     Mutex::Locker l(m_lock);
     if (!m_manual_stop) {
@@ -275,7 +275,7 @@ void Mirror::print_status(Formatter *f, stringstream *ss)
 
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
@@ -314,7 +314,7 @@ void Mirror::start()
   dout(20) << "enter" << dendl;
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
@@ -330,7 +330,7 @@ void Mirror::stop()
   dout(20) << "enter" << dendl;
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
@@ -346,7 +346,7 @@ void Mirror::restart()
   dout(20) << "enter" << dendl;
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
@@ -362,7 +362,7 @@ void Mirror::flush()
   dout(20) << "enter" << dendl;
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read() || m_manual_stop) {
+  if (m_stopping || m_manual_stop) {
     return;
   }
 
@@ -376,7 +376,7 @@ void Mirror::release_leader()
   dout(20) << "enter" << dendl;
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
diff --git a/src/tools/rbd_mirror/Mirror.h b/src/tools/rbd_mirror/Mirror.h
index 4ff9d512399..2253156d660 100644
--- a/src/tools/rbd_mirror/Mirror.h
+++ b/src/tools/rbd_mirror/Mirror.h
@@ -4,19 +4,19 @@
 #ifndef CEPH_RBD_MIRROR_H
 #define CEPH_RBD_MIRROR_H
 
-#include <map>
-#include <memory>
-#include <set>
-
 #include "common/ceph_context.h"
 #include "common/Mutex.h"
-#include "include/atomic.h"
 #include "include/rados/librados.hpp"
 #include "ClusterWatcher.h"
 #include "PoolReplayer.h"
 #include "ImageDeleter.h"
 #include "types.h"
 
+#include <set>
+#include <map>
+#include <memory>
+#include <atomic>
+
 namespace librbd { struct ImageCtx; }
 
 namespace rbd {
@@ -67,7 +67,7 @@ private:
   std::shared_ptr<ImageDeleter> m_image_deleter;
   ImageSyncThrottlerRef<> m_image_sync_throttler;
   std::map<PoolPeer, std::unique_ptr<PoolReplayer> > m_pool_replayers;
-  atomic_t m_stopping;
+  std::atomic<bool> m_stopping = { false };
   bool m_manual_stop = false;
   MirrorAdminSocketHook *m_asok_hook;
 };
diff --git a/src/tools/rbd_mirror/PoolReplayer.cc b/src/tools/rbd_mirror/PoolReplayer.cc
index 6747ddc15aa..0bd06f624ae 100644
--- a/src/tools/rbd_mirror/PoolReplayer.cc
+++ b/src/tools/rbd_mirror/PoolReplayer.cc
@@ -21,7 +21,6 @@
 #include "InstanceWatcher.h"
 #include "LeaderWatcher.h"
 #include "Threads.h"
-#include "pool_watcher/RefreshImagesRequest.h"
 
 #define dout_context g_ceph_context
 #define dout_subsys ceph_subsys_rbd_mirror
@@ -206,20 +205,6 @@ private:
 
 } // anonymous namespace
 
-struct PoolReplayer::C_RefreshLocalImages : public Context {
-  PoolReplayer *pool_replayer;
-  Context *on_finish;
-  ImageIds image_ids;
-
-  C_RefreshLocalImages(PoolReplayer *pool_replayer, Context *on_finish)
-    : pool_replayer(pool_replayer), on_finish(on_finish) {
-  }
-
-  void finish(int r) override {
-    pool_replayer->handle_refresh_local_images(r, std::move(image_ids), on_finish);
-  }
-};
-
 PoolReplayer::PoolReplayer(Threads<librbd::ImageCtx> *threads,
 			   std::shared_ptr<ImageDeleter> image_deleter,
 			   ImageSyncThrottlerRef<> image_sync_throttler,
@@ -232,7 +217,8 @@ PoolReplayer::PoolReplayer(Threads<librbd::ImageCtx> *threads,
   m_peer(peer),
   m_args(args),
   m_local_pool_id(local_pool_id),
-  m_pool_watcher_listener(this),
+  m_local_pool_watcher_listener(this, true),
+  m_remote_pool_watcher_listener(this, false),
   m_asok_hook(nullptr),
   m_pool_replayer_thread(this),
   m_leader_listener(this)
@@ -243,7 +229,7 @@ PoolReplayer::~PoolReplayer()
 {
   delete m_asok_hook;
 
-  m_stopping.set(1);
+  m_stopping = true;
   {
     Mutex::Locker l(m_lock);
     m_cond.Signal();
@@ -261,7 +247,8 @@ PoolReplayer::~PoolReplayer()
     m_instance_replayer->shut_down();
   }
 
-  assert(!m_pool_watcher);
+  assert(!m_local_pool_watcher);
+  assert(!m_remote_pool_watcher);
 }
 
 bool PoolReplayer::is_blacklisted() const {
@@ -400,6 +387,11 @@ int PoolReplayer::init_rados(const std::string &cluster_name,
     }
   }
 
+  if (!g_ceph_context->_conf->admin_socket.empty()) {
+    cct->_conf->set_val_or_die("admin_socket",
+                               "$run_dir/$name.$pid.$cluster.$cctid.asok");
+  }
+
   // disable unnecessary librbd cache
   cct->_conf->set_val_or_die("rbd_cache", "false");
   cct->_conf->apply_changes(nullptr);
@@ -423,7 +415,7 @@ void PoolReplayer::run()
 {
   dout(20) << "enter" << dendl;
 
-  while (!m_stopping.read()) {
+  while (!m_stopping) {
     std::string asok_hook_name = m_local_io_ctx.get_pool_name() + " " +
                                  m_peer.cluster_name;
     if (m_asok_hook_name != asok_hook_name || m_asok_hook == nullptr) {
@@ -435,9 +427,10 @@ void PoolReplayer::run()
     }
 
     Mutex::Locker locker(m_lock);
-    if (m_pool_watcher && m_pool_watcher->is_blacklisted()) {
+    if ((m_local_pool_watcher && m_local_pool_watcher->is_blacklisted()) ||
+	(m_remote_pool_watcher && m_remote_pool_watcher->is_blacklisted())) {
       m_blacklisted = true;
-      m_stopping.set(1);
+      m_stopping = true;
       break;
     }
 
@@ -476,6 +469,13 @@ void PoolReplayer::print_status(Formatter *f, stringstream *ss)
     f->close_section();
   }
 
+  f->dump_string("local_cluster_admin_socket",
+                 reinterpret_cast<CephContext *>(m_local_io_ctx.cct())->_conf->
+                     admin_socket);
+  f->dump_string("remote_cluster_admin_socket",
+                 reinterpret_cast<CephContext *>(m_remote_io_ctx.cct())->_conf->
+                     admin_socket);
+
   m_instance_replayer->print_status(f, ss);
 
   f->close_section();
@@ -488,7 +488,7 @@ void PoolReplayer::start()
 
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
@@ -501,10 +501,10 @@ void PoolReplayer::stop(bool manual)
 
   Mutex::Locker l(m_lock);
   if (!manual) {
-    m_stopping.set(1);
+    m_stopping = true;
     m_cond.Signal();
     return;
-  } else if (m_stopping.read()) {
+  } else if (m_stopping) {
     return;
   }
 
@@ -517,7 +517,7 @@ void PoolReplayer::restart()
 
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read()) {
+  if (m_stopping) {
     return;
   }
 
@@ -530,7 +530,7 @@ void PoolReplayer::flush()
 
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read() || m_manual_stop) {
+  if (m_stopping || m_manual_stop) {
     return;
   }
 
@@ -543,7 +543,7 @@ void PoolReplayer::release_leader()
 
   Mutex::Locker l(m_lock);
 
-  if (m_stopping.read() || !m_leader_watcher) {
+  if (m_stopping || !m_leader_watcher) {
     return;
   }
 
@@ -551,44 +551,46 @@ void PoolReplayer::release_leader()
 }
 
 void PoolReplayer::handle_update(const std::string &mirror_uuid,
-				 const ImageIds &added_image_ids,
-				 const ImageIds &removed_image_ids) {
-  assert(!mirror_uuid.empty());
-  if (m_stopping.read()) {
+				 ImageIds &&added_image_ids,
+				 ImageIds &&removed_image_ids) {
+  if (m_stopping) {
     return;
   }
 
-  dout(10) << dendl;
+  dout(10) << "mirror_uuid=" << mirror_uuid << ", "
+           << "added_count=" << added_image_ids.size() << ", "
+           << "removed_count=" << removed_image_ids.size() << dendl;
   Mutex::Locker locker(m_lock);
   if (!m_leader_watcher->is_leader()) {
     return;
   }
 
-  if (m_peer.uuid != mirror_uuid) {
-    m_instance_replayer->remove_peer(m_peer.uuid);
-    m_instance_replayer->add_peer(mirror_uuid, m_remote_io_ctx);
-    m_peer.uuid = mirror_uuid;
-  }
+  if (m_initial_mirror_image_ids.find(mirror_uuid) ==
+        m_initial_mirror_image_ids.end() &&
+      m_initial_mirror_image_ids.size() < 2) {
+    m_initial_mirror_image_ids[mirror_uuid] = added_image_ids;
 
-  // first callback will be a full directory -- so see if we need to remove
-  // any local images that no longer exist on the remote side
-  if (!m_init_image_ids.empty()) {
-    dout(20) << "scanning initial local image set" << dendl;
-    for (auto &image_id : added_image_ids) {
-      auto it = m_init_image_ids.find(image_id);
-      if (it != m_init_image_ids.end()) {
-        m_init_image_ids.erase(it);
+    if (m_initial_mirror_image_ids.size() == 2) {
+      dout(10) << "local and remote pools refreshed" << dendl;
+
+      // both local and remote initial pool listing received. derive
+      // removal notifications for the remote pool
+      auto &local_image_ids = m_initial_mirror_image_ids.begin()->second;
+      auto &remote_image_ids = m_initial_mirror_image_ids.rbegin()->second;
+      for (auto &local_image_id : local_image_ids) {
+        if (remote_image_ids.find(local_image_id) == remote_image_ids.end()) {
+          removed_image_ids.emplace(local_image_id.global_id, "");
+        }
       }
+      local_image_ids.clear();
+      remote_image_ids.clear();
     }
+  }
 
-    // the remaining images in m_init_image_ids must be deleted
-    for (auto &image_id : m_init_image_ids) {
-      dout(20) << "scheduling the deletion of init image: "
-               << image_id.global_id << " (" << image_id.id << ")" << dendl;
-      m_image_deleter->schedule_image_delete(m_local_rados, m_local_pool_id,
-                                             image_id.id, image_id.global_id);
-    }
-    m_init_image_ids.clear();
+  if (!mirror_uuid.empty() && m_peer.uuid != mirror_uuid) {
+    m_instance_replayer->remove_peer(m_peer.uuid);
+    m_instance_replayer->add_peer(mirror_uuid, m_remote_io_ctx);
+    m_peer.uuid = mirror_uuid;
   }
 
   m_update_op_tracker.start_op();
@@ -599,19 +601,19 @@ void PoolReplayer::handle_update(const std::string &mirror_uuid,
 
   C_Gather *gather_ctx = new C_Gather(g_ceph_context, ctx);
 
-  for (auto &image_id : removed_image_ids) {
+  for (auto &image_id : added_image_ids) {
     // for now always send to myself (the leader)
     std::string &instance_id = m_instance_watcher->get_instance_id();
-    m_instance_watcher->notify_image_release(instance_id, image_id.global_id,
-                                             mirror_uuid, image_id.id, true,
+    m_instance_watcher->notify_image_acquire(instance_id, image_id.global_id,
+                                             mirror_uuid, image_id.id,
                                              gather_ctx->new_sub());
   }
 
-  for (auto &image_id : added_image_ids) {
+  for (auto &image_id : removed_image_ids) {
     // for now always send to myself (the leader)
     std::string &instance_id = m_instance_watcher->get_instance_id();
-    m_instance_watcher->notify_image_acquire(instance_id, image_id.global_id,
-                                             mirror_uuid, image_id.id,
+    m_instance_watcher->notify_image_release(instance_id, image_id.global_id,
+                                             mirror_uuid, image_id.id, true,
                                              gather_ctx->new_sub());
   }
 
@@ -620,68 +622,73 @@ void PoolReplayer::handle_update(const std::string &mirror_uuid,
 
 void PoolReplayer::handle_post_acquire_leader(Context *on_finish) {
   dout(20) << dendl;
-  refresh_local_images(on_finish);
+  init_local_pool_watcher(on_finish);
 }
 
 void PoolReplayer::handle_pre_release_leader(Context *on_finish) {
   dout(20) << dendl;
-  shut_down_pool_watcher(on_finish);
+  shut_down_pool_watchers(on_finish);
 }
 
-void PoolReplayer::refresh_local_images(Context *on_finish) {
+void PoolReplayer::init_local_pool_watcher(Context *on_finish) {
   dout(20) << dendl;
 
+  Mutex::Locker locker(m_lock);
+  assert(!m_local_pool_watcher);
+  m_local_pool_watcher.reset(new PoolWatcher<>(
+    m_threads, m_local_io_ctx, m_local_pool_watcher_listener));
+  m_initial_mirror_image_ids.clear();
+
   // ensure the initial set of local images is up-to-date
   // after acquiring the leader role
-  auto ctx = new C_RefreshLocalImages(this, on_finish);
-  auto req = pool_watcher::RefreshImagesRequest<>::create(
-    m_local_io_ctx, &ctx->image_ids, ctx);
-  req->send();
+  auto ctx = new FunctionContext([this, on_finish](int r) {
+      handle_init_local_pool_watcher(r, on_finish);
+    });
+  m_local_pool_watcher->init(create_async_context_callback(
+    m_threads->work_queue, ctx));
 }
 
-void PoolReplayer::handle_refresh_local_images(int r, ImageIds &&image_ids,
-					       Context *on_finish) {
+void PoolReplayer::handle_init_local_pool_watcher(int r, Context *on_finish) {
   dout(20) << "r=" << r << dendl;
-
-  {
-    Mutex::Locker locker(m_lock);
-    m_init_image_ids = std::move(image_ids);
-  }
-
   if (r < 0) {
     derr << "failed to retrieve local images: " << cpp_strerror(r) << dendl;
     on_finish->complete(r);
     return;
   }
 
-  init_pool_watcher(on_finish);
+  init_remote_pool_watcher(on_finish);
 }
 
-void PoolReplayer::init_pool_watcher(Context *on_finish) {
+void PoolReplayer::init_remote_pool_watcher(Context *on_finish) {
   dout(20) << dendl;
 
   Mutex::Locker locker(m_lock);
-  assert(!m_pool_watcher);
-  m_pool_watcher.reset(new PoolWatcher<>(
-    m_threads, m_remote_io_ctx, m_pool_watcher_listener));
-  m_pool_watcher->init(create_async_context_callback(
+  assert(!m_remote_pool_watcher);
+  m_remote_pool_watcher.reset(new PoolWatcher<>(
+    m_threads, m_remote_io_ctx, m_remote_pool_watcher_listener));
+  m_remote_pool_watcher->init(create_async_context_callback(
     m_threads->work_queue, on_finish));
 
   m_cond.Signal();
 }
 
-void PoolReplayer::shut_down_pool_watcher(Context *on_finish) {
+void PoolReplayer::shut_down_pool_watchers(Context *on_finish) {
   dout(20) << dendl;
 
   {
     Mutex::Locker locker(m_lock);
-    if (m_pool_watcher) {
+    if (m_local_pool_watcher) { 
       Context *ctx = new FunctionContext([this, on_finish](int r) {
-          handle_shut_down_pool_watcher(r, on_finish);
-      });
+          handle_shut_down_pool_watchers(r, on_finish);
+	});
       ctx = create_async_context_callback(m_threads->work_queue, ctx);
 
-      m_pool_watcher->shut_down(ctx);
+      auto gather_ctx = new C_Gather(g_ceph_context, ctx);
+      m_local_pool_watcher->shut_down(gather_ctx->new_sub());
+      if (m_remote_pool_watcher) {
+	m_remote_pool_watcher->shut_down(gather_ctx->new_sub());
+      }
+      gather_ctx->activate();
       return;
     }
   }
@@ -689,13 +696,17 @@ void PoolReplayer::shut_down_pool_watcher(Context *on_finish) {
   on_finish->complete(0);
 }
 
-void PoolReplayer::handle_shut_down_pool_watcher(int r, Context *on_finish) {
+void PoolReplayer::handle_shut_down_pool_watchers(int r, Context *on_finish) {
   dout(20) << "r=" << r << dendl;
 
   {
     Mutex::Locker locker(m_lock);
-    assert(m_pool_watcher);
-    m_pool_watcher.reset();
+    assert(m_local_pool_watcher);
+    m_local_pool_watcher.reset();
+
+    if (m_remote_pool_watcher) {
+      m_remote_pool_watcher.reset();
+    }
   }
   wait_for_update_ops(on_finish);
 }
diff --git a/src/tools/rbd_mirror/PoolReplayer.h b/src/tools/rbd_mirror/PoolReplayer.h
index 7de10a5c6a5..87a6589356e 100644
--- a/src/tools/rbd_mirror/PoolReplayer.h
+++ b/src/tools/rbd_mirror/PoolReplayer.h
@@ -4,16 +4,10 @@
 #ifndef CEPH_RBD_MIRROR_POOL_REPLAYER_H
 #define CEPH_RBD_MIRROR_POOL_REPLAYER_H
 
-#include <map>
-#include <memory>
-#include <set>
-#include <string>
-
 #include "common/AsyncOpTracker.h"
 #include "common/Cond.h"
 #include "common/Mutex.h"
 #include "common/WorkQueue.h"
-#include "include/atomic.h"
 #include "include/rados/librados.hpp"
 
 #include "ClusterWatcher.h"
@@ -22,6 +16,12 @@
 #include "ImageDeleter.h"
 #include "types.h"
 
+#include <set>
+#include <map>
+#include <memory>
+#include <atomic>
+#include <string>
+
 class AdminSocketHook;
 
 namespace librbd { class ImageCtx; }
@@ -63,24 +63,24 @@ public:
 private:
   struct PoolWatcherListener : public PoolWatcher<>::Listener {
     PoolReplayer *pool_replayer;
+    bool local;
 
-    PoolWatcherListener(PoolReplayer *pool_replayer)
-      : pool_replayer(pool_replayer) {
+    PoolWatcherListener(PoolReplayer *pool_replayer, bool local)
+      : pool_replayer(pool_replayer), local(local) {
     }
 
     void handle_update(const std::string &mirror_uuid,
-                       const ImageIds &added_image_ids,
-                       const ImageIds &removed_image_ids) override {
-      pool_replayer->handle_update(mirror_uuid, added_image_ids,
-				   removed_image_ids);
+                       ImageIds &&added_image_ids,
+                       ImageIds &&removed_image_ids) override {
+      pool_replayer->handle_update((local ? "" : mirror_uuid),
+				   std::move(added_image_ids),
+                                   std::move(removed_image_ids));
     }
   };
 
-  struct C_RefreshLocalImages;
-
   void handle_update(const std::string &mirror_uuid,
-                     const ImageIds &added_image_ids,
-                     const ImageIds &removed_image_ids);
+                     ImageIds &&added_image_ids,
+                     ImageIds &&removed_image_ids);
 
   int init_rados(const std::string &cluster_name,
                  const std::string &client_name,
@@ -89,13 +89,13 @@ private:
   void handle_post_acquire_leader(Context *on_finish);
   void handle_pre_release_leader(Context *on_finish);
 
-  void refresh_local_images(Context *on_finish);
-  void handle_refresh_local_images(int r, ImageIds &&image_ids,
-                                   Context *on_finish);
+  void init_local_pool_watcher(Context *on_finish);
+  void handle_init_local_pool_watcher(int r, Context *on_finish);
+
+  void init_remote_pool_watcher(Context *on_finish);
 
-  void init_pool_watcher(Context *on_finish);
-  void shut_down_pool_watcher(Context *on_finish);
-  void handle_shut_down_pool_watcher(int r, Context *on_finish);
+  void shut_down_pool_watchers(Context *on_finish);
+  void handle_shut_down_pool_watchers(int r, Context *on_finish);
 
   void wait_for_update_ops(Context *on_finish);
   void handle_wait_for_update_ops(int r, Context *on_finish);
@@ -105,7 +105,7 @@ private:
   ImageSyncThrottlerRef<> m_image_sync_throttler;
   mutable Mutex m_lock;
   Cond m_cond;
-  atomic_t m_stopping;
+  std::atomic<bool> m_stopping = { false };
   bool m_manual_stop = false;
   bool m_blacklisted = false;
 
@@ -119,15 +119,18 @@ private:
 
   int64_t m_local_pool_id = -1;
 
-  PoolWatcherListener m_pool_watcher_listener;
-  std::unique_ptr<PoolWatcher<> > m_pool_watcher;
+  PoolWatcherListener m_local_pool_watcher_listener;
+  std::unique_ptr<PoolWatcher<> > m_local_pool_watcher;
+
+  PoolWatcherListener m_remote_pool_watcher_listener;
+  std::unique_ptr<PoolWatcher<> > m_remote_pool_watcher;
 
   std::unique_ptr<InstanceReplayer<librbd::ImageCtx>> m_instance_replayer;
 
   std::string m_asok_hook_name;
   AdminSocketHook *m_asok_hook;
 
-  std::set<ImageId> m_init_image_ids;
+  std::map<std::string, ImageIds> m_initial_mirror_image_ids;
 
   class PoolReplayerThread : public Thread {
     PoolReplayer *m_pool_replayer;
diff --git a/src/tools/rbd_mirror/PoolWatcher.cc b/src/tools/rbd_mirror/PoolWatcher.cc
index 6a855ff2185..18c6df3840f 100644
--- a/src/tools/rbd_mirror/PoolWatcher.cc
+++ b/src/tools/rbd_mirror/PoolWatcher.cc
@@ -479,7 +479,7 @@ void PoolWatcher<I>::notify_listener() {
   }
 
   if (!removed_image_ids.empty()) {
-    m_listener.handle_update(mirror_uuid, {}, removed_image_ids);
+    m_listener.handle_update(mirror_uuid, {}, std::move(removed_image_ids));
     removed_image_ids.clear();
   }
 
@@ -529,7 +529,8 @@ void PoolWatcher<I>::notify_listener() {
     mirror_uuid = m_mirror_uuid;
   }
 
-  m_listener.handle_update(mirror_uuid, added_image_ids, removed_image_ids);
+  m_listener.handle_update(mirror_uuid, std::move(added_image_ids),
+                           std::move(removed_image_ids));
 
   {
     Mutex::Locker locker(m_lock);
diff --git a/src/tools/rbd_mirror/PoolWatcher.h b/src/tools/rbd_mirror/PoolWatcher.h
index aebd981b864..aec063b3e7c 100644
--- a/src/tools/rbd_mirror/PoolWatcher.h
+++ b/src/tools/rbd_mirror/PoolWatcher.h
@@ -37,8 +37,8 @@ public:
     }
 
     virtual void handle_update(const std::string &mirror_uuid,
-                               const ImageIds &added_image_ids,
-                               const ImageIds &removed_image_ids) = 0;
+                               ImageIds &&added_image_ids,
+                               ImageIds &&removed_image_ids) = 0;
   };
 
   PoolWatcher(Threads<ImageCtxT> *threads, librados::IoCtx &remote_io_ctx,
diff --git a/src/tools/rbd_mirror/image_replayer/BootstrapRequest.cc b/src/tools/rbd_mirror/image_replayer/BootstrapRequest.cc
index 7d7d5ed1a6c..8818945b10f 100644
--- a/src/tools/rbd_mirror/image_replayer/BootstrapRequest.cc
+++ b/src/tools/rbd_mirror/image_replayer/BootstrapRequest.cc
@@ -20,7 +20,6 @@
 #include "librbd/Journal.h"
 #include "librbd/Utils.h"
 #include "librbd/journal/Types.h"
-#include "tools/rbd_mirror/ImageSync.h"
 #include "tools/rbd_mirror/ProgressContext.h"
 #include "tools/rbd_mirror/ImageSyncThrottler.h"
 
@@ -79,7 +78,7 @@ template <typename I>
 void BootstrapRequest<I>::send() {
   *m_do_resync = false;
 
-  get_local_image_id();
+  get_remote_tag_class();
 }
 
 template <typename I>
@@ -93,45 +92,6 @@ void BootstrapRequest<I>::cancel() {
 }
 
 template <typename I>
-void BootstrapRequest<I>::get_local_image_id() {
-  dout(20) << dendl;
-
-  update_progress("GET_LOCAL_IMAGE_ID");
-
-  // attempt to cross-reference a local image by the global image id
-  librados::ObjectReadOperation op;
-  librbd::cls_client::mirror_image_get_image_id_start(&op, m_global_image_id);
-
-  librados::AioCompletion *aio_comp = create_rados_callback<
-    BootstrapRequest<I>, &BootstrapRequest<I>::handle_get_local_image_id>(
-      this);
-  int r = m_local_io_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op, &m_out_bl);
-  assert(r == 0);
-  aio_comp->release();
-}
-
-template <typename I>
-void BootstrapRequest<I>::handle_get_local_image_id(int r) {
-  dout(20) << ": r=" << r << dendl;
-
-  if (r == 0) {
-    bufferlist::iterator iter = m_out_bl.begin();
-    r = librbd::cls_client::mirror_image_get_image_id_finish(
-      &iter, &m_local_image_id);
-  }
-
-  if (r == -ENOENT) {
-    dout(10) << ": image not registered locally" << dendl;
-  } else if (r < 0) {
-    derr << ": failed to retrieve local image id: " << cpp_strerror(r) << dendl;
-    finish(r);
-    return;
-  }
-
-  get_remote_tag_class();
-}
-
-template <typename I>
 void BootstrapRequest<I>::get_remote_tag_class() {
   dout(20) << dendl;
 
@@ -453,7 +413,6 @@ void BootstrapRequest<I>::handle_create_local_image(int r) {
     return;
   }
 
-  m_created_local_image = true;
   open_local_image();
 }
 
@@ -472,8 +431,8 @@ void BootstrapRequest<I>::update_client_image() {
 
   dout(20) << dendl;
 
-  librbd::journal::MirrorPeerClientMeta client_meta;
-  client_meta.image_id = m_local_image_id;
+  librbd::journal::MirrorPeerClientMeta client_meta{m_local_image_id};
+  client_meta.state = librbd::journal::MIRROR_PEER_STATE_SYNCING;
 
   librbd::journal::ClientData client_data(client_meta);
   bufferlist data_bl;
@@ -503,7 +462,8 @@ void BootstrapRequest<I>::handle_update_client_image(int r) {
     return;
   }
 
-  m_client_meta->image_id = m_local_image_id;
+  *m_client_meta = {m_local_image_id};
+  m_client_meta->state = librbd::journal::MIRROR_PEER_STATE_SYNCING;
   get_remote_tags();
 }
 
@@ -513,8 +473,7 @@ void BootstrapRequest<I>::get_remote_tags() {
 
   update_progress("GET_REMOTE_TAGS");
 
-  if (m_created_local_image ||
-      m_client_meta->state == librbd::journal::MIRROR_PEER_STATE_SYNCING) {
+  if (m_client_meta->state == librbd::journal::MIRROR_PEER_STATE_SYNCING) {
     // optimization -- no need to compare remote tags if we just created
     // the image locally or sync was interrupted
     image_sync();
diff --git a/src/tools/rbd_mirror/image_replayer/BootstrapRequest.h b/src/tools/rbd_mirror/image_replayer/BootstrapRequest.h
index e367b2ae482..6e755689488 100644
--- a/src/tools/rbd_mirror/image_replayer/BootstrapRequest.h
+++ b/src/tools/rbd_mirror/image_replayer/BootstrapRequest.h
@@ -87,9 +87,6 @@ private:
    * <start>
    *    |
    *    v
-   * GET_LOCAL_IMAGE_ID * * * * * * * * * * * * * * * * *
-   *    |                                               *
-   *    v                                               *
    * GET_REMOTE_TAG_CLASS * * * * * * * * * * * * * * * *
    *    |                                               *
    *    v                                               *
@@ -170,14 +167,10 @@ private:
   uint64_t m_remote_tag_class = 0;
   ImageCtxT *m_remote_image_ctx = nullptr;
   bool m_primary = false;
-  bool m_created_local_image = false;
   int m_ret_val = 0;
 
   bufferlist m_out_bl;
 
-  void get_local_image_id();
-  void handle_get_local_image_id(int r);
-
   void get_remote_tag_class();
   void handle_get_remote_tag_class(int r);
 
diff --git a/src/tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.cc b/src/tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.cc
new file mode 100644
index 00000000000..b26ac05e942
--- /dev/null
+++ b/src/tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.cc
@@ -0,0 +1,160 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#include "tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.h"
+#include "include/rados/librados.hpp"
+#include "cls/rbd/cls_rbd_client.h"
+#include "common/errno.h"
+#include "librbd/ImageCtx.h"
+#include "librbd/Journal.h"
+#include "librbd/Utils.h"
+#include "tools/rbd_mirror/Threads.h"
+#include <type_traits>
+
+#define dout_context g_ceph_context
+#define dout_subsys ceph_subsys_rbd_mirror
+#undef dout_prefix
+#define dout_prefix *_dout << "rbd::mirror::image_replayer::" \
+                           << "PrepareLocalImageRequest: " << this << " " \
+                           << __func__ << ": "
+
+namespace rbd {
+namespace mirror {
+namespace image_replayer {
+
+using librbd::util::create_context_callback;
+using librbd::util::create_rados_callback;
+
+template <typename I>
+void PrepareLocalImageRequest<I>::send() {
+  dout(20) << dendl;
+  get_local_image_id();
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::get_local_image_id() {
+  dout(20) << dendl;
+
+  // attempt to cross-reference a local image by the global image id
+  librados::ObjectReadOperation op;
+  librbd::cls_client::mirror_image_get_image_id_start(&op, m_global_image_id);
+
+  m_out_bl.clear();
+  librados::AioCompletion *aio_comp = create_rados_callback<
+    PrepareLocalImageRequest<I>,
+    &PrepareLocalImageRequest<I>::handle_get_local_image_id>(
+      this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op, &m_out_bl);
+  assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::handle_get_local_image_id(int r) {
+  if (r == 0) {
+    bufferlist::iterator iter = m_out_bl.begin();
+    r = librbd::cls_client::mirror_image_get_image_id_finish(
+      &iter, m_local_image_id);
+  }
+
+  dout(20) << "r=" << r << ", "
+           << "local_image_id=" << *m_local_image_id << dendl;
+
+  if (r < 0) {
+    if (r == -ENOENT) {
+      dout(10) << "image not registered locally" << dendl;
+    } else {
+      derr << "failed to retrieve local image id: " << cpp_strerror(r)
+           << dendl;
+    }
+    finish(r);
+    return;
+  }
+
+  get_mirror_state();
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::get_mirror_state() {
+  dout(20) << dendl;
+
+  librados::ObjectReadOperation op;
+  librbd::cls_client::mirror_image_get_start(&op, *m_local_image_id);
+
+  m_out_bl.clear();
+  librados::AioCompletion *aio_comp = create_rados_callback<
+    PrepareLocalImageRequest<I>,
+    &PrepareLocalImageRequest<I>::handle_get_mirror_state>(this);
+  int r = m_io_ctx.aio_operate(RBD_MIRRORING, aio_comp, &op, &m_out_bl);
+  assert(r == 0);
+  aio_comp->release();
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::handle_get_mirror_state(int r) {
+  dout(20) << ": r=" << r << dendl;
+
+  cls::rbd::MirrorImage mirror_image;
+  if (r == 0) {
+    bufferlist::iterator iter = m_out_bl.begin();
+    r = librbd::cls_client::mirror_image_get_finish(&iter, &mirror_image);
+  }
+
+  if (r < 0) {
+    derr << "failed to retrieve image mirror state: " << cpp_strerror(r)
+         << dendl;
+    finish(r);
+    return;
+  }
+
+  // TODO save current mirror state to determine if we should
+  // delete a partially formed image
+  // (e.g. MIRROR_IMAGE_STATE_CREATING/DELETING)
+
+  get_tag_owner();
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::get_tag_owner() {
+  // deduce the class type for the journal to support unit tests
+  using Journal = typename std::decay<
+    typename std::remove_pointer<decltype(std::declval<I>().journal)>
+    ::type>::type;
+
+  dout(20) << dendl;
+
+  Context *ctx = create_context_callback<
+    PrepareLocalImageRequest<I>,
+    &PrepareLocalImageRequest<I>::handle_get_tag_owner>(this);
+  Journal::get_tag_owner(m_io_ctx, *m_local_image_id, m_tag_owner,
+                         m_work_queue, ctx);
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::handle_get_tag_owner(int r) {
+  dout(20) << "r=" << r << ", "
+           << "tag_owner=" << *m_tag_owner << dendl;
+
+  if (r < 0) {
+    derr << "failed to retrieve journal tag owner: " << cpp_strerror(r)
+         << dendl;
+    finish(r);
+    return;
+  }
+
+  finish(0);
+}
+
+template <typename I>
+void PrepareLocalImageRequest<I>::finish(int r) {
+  dout(20) << "r=" << r << dendl;
+
+  m_on_finish->complete(r);
+  delete this;
+}
+
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd
+
+template class rbd::mirror::image_replayer::PrepareLocalImageRequest<librbd::ImageCtx>;
diff --git a/src/tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.h b/src/tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.h
new file mode 100644
index 00000000000..913bfd1c242
--- /dev/null
+++ b/src/tools/rbd_mirror/image_replayer/PrepareLocalImageRequest.h
@@ -0,0 +1,92 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+#ifndef RBD_MIRROR_IMAGE_REPLAYER_PREPARE_LOCAL_IMAGE_REQUEST_H
+#define RBD_MIRROR_IMAGE_REPLAYER_PREPARE_LOCAL_IMAGE_REQUEST_H
+
+#include "include/buffer.h"
+#include <string>
+
+namespace librados { struct IoCtx; }
+namespace librbd { struct ImageCtx; }
+
+struct Context;
+struct ContextWQ;
+
+namespace rbd {
+namespace mirror {
+namespace image_replayer {
+
+template <typename ImageCtxT = librbd::ImageCtx>
+class PrepareLocalImageRequest {
+public:
+  static PrepareLocalImageRequest *create(librados::IoCtx &io_ctx,
+                                          const std::string &global_image_id,
+                                          std::string *local_image_id,
+                                          std::string *tag_owner,
+                                          ContextWQ *work_queue,
+                                          Context *on_finish) {
+    return new PrepareLocalImageRequest(io_ctx, global_image_id, local_image_id,
+                                        tag_owner, work_queue, on_finish);
+  }
+
+  PrepareLocalImageRequest(librados::IoCtx &io_ctx,
+                           const std::string &global_image_id,
+                           std::string *local_image_id,
+                           std::string *tag_owner,
+                           ContextWQ *work_queue,
+                           Context *on_finish)
+    : m_io_ctx(io_ctx), m_global_image_id(global_image_id),
+      m_local_image_id(local_image_id), m_tag_owner(tag_owner),
+      m_work_queue(work_queue), m_on_finish(on_finish) {
+  }
+
+  void send();
+
+private:
+  /**
+   * @verbatim
+   *
+   * <start>
+   *    |
+   *    v
+   * GET_LOCAL_IMAGE_ID
+   *    |
+   *    v
+   * GET_MIRROR_STATE
+   *    |
+   *    v
+   * <finish>
+
+   * @endverbatim
+   */
+
+  librados::IoCtx &m_io_ctx;
+  std::string m_global_image_id;
+  std::string *m_local_image_id;
+  std::string *m_tag_owner;
+  ContextWQ *m_work_queue;
+  Context *m_on_finish;
+
+  bufferlist m_out_bl;
+
+  void get_local_image_id();
+  void handle_get_local_image_id(int r);
+
+  void get_mirror_state();
+  void handle_get_mirror_state(int r);
+
+  void get_tag_owner();
+  void handle_get_tag_owner(int r);
+
+  void finish(int r);
+
+};
+
+} // namespace image_replayer
+} // namespace mirror
+} // namespace rbd
+
+extern template class rbd::mirror::image_replayer::PrepareLocalImageRequest<librbd::ImageCtx>;
+
+#endif // RBD_MIRROR_IMAGE_REPLAYER_PREPARE_LOCAL_IMAGE_REQUEST_H
diff --git a/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc b/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc
index faca72b47d4..e86765c2d4f 100644
--- a/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc
+++ b/src/tools/rbd_mirror/image_replayer/ReplayStatusFormatter.cc
@@ -196,7 +196,13 @@ void ReplayStatusFormatter<I>::handle_update_tag_cache(uint64_t master_tag_tid,
     }
   }
 
-  if (tag_data.predecessor.tag_tid == 0) {
+  if (tag_data.predecessor.mirror_uuid !=
+        librbd::Journal<>::LOCAL_MIRROR_UUID &&
+      tag_data.predecessor.mirror_uuid !=
+        librbd::Journal<>::ORPHAN_MIRROR_UUID) {
+    dout(20) << "hit remote image non-primary epoch" << dendl;
+    tag_data.predecessor.tag_tid = mirror_tag_tid;
+  } else if (tag_data.predecessor.tag_tid == 0) {
     // We failed. Don't consider this fatal, just terminate retrieving.
     dout(20) << "making fake tag" << dendl;
     tag_data.predecessor.tag_tid = mirror_tag_tid;
diff --git a/src/tools/rbd_mirror/image_sync/ObjectCopyRequest.cc b/src/tools/rbd_mirror/image_sync/ObjectCopyRequest.cc
index 2d5f1470e8a..92817b7a788 100644
--- a/src/tools/rbd_mirror/image_sync/ObjectCopyRequest.cc
+++ b/src/tools/rbd_mirror/image_sync/ObjectCopyRequest.cc
@@ -334,7 +334,7 @@ void ObjectCopyRequest<I>::send_update_object_map() {
   bool sent = m_local_image_ctx->object_map->template aio_update<
     ObjectCopyRequest<I>, &ObjectCopyRequest<I>::handle_update_object_map>(
       snap_object_state.first, m_object_number, snap_object_state.second, {},
-      this);
+      {}, this);
   assert(sent);
   m_local_image_ctx->snap_lock.put_read();
 }