rbd-mirror: wait for in-flight event commit notifications

Fixes: http://tracker.ceph.com/issues/17355
Signed-off-by: Jason Dillaman <dillaman@redhat.com>

src/tools/rbd_mirror/ImageReplayer.cc

@@ -1083,6 +1083,8 @@ void ImageReplayer<I>::process_entry() {
   Context *on_ready = create_context_callback<
     ImageReplayer, &ImageReplayer<I>::handle_process_entry_ready>(this);
   Context *on_commit = new C_ReplayCommitted(this, std::move(m_replay_entry));
+
+  m_event_replay_tracker.start_op();
   m_local_replay->process(m_event_entry, on_ready, on_commit);
 }
 
@@ -1103,14 +1105,12 @@ void ImageReplayer<I>::handle_process_entry_safe(const ReplayEntry& replay_entry
 
   if (r < 0) {
     derr << "failed to commit journal event: " << cpp_strerror(r) << dendl;
-
     handle_replay_complete(r, "failed to commit journal event");
-    return;
-  }
-
-  if (m_remote_journaler) {
+  } else {
+    assert(m_remote_journaler != nullptr);
     m_remote_journaler->committed(replay_entry);
   }
+  m_event_replay_tracker.finish_op();
 }
 
 template <typename I>
@@ -1397,7 +1397,8 @@ void ImageReplayer<I>::shut_down(int r) {
         delete m_event_preprocessor;
         m_event_preprocessor = nullptr;
 
-        ctx->complete(0);
+        // wait for all in-flight replayed events to complete
+        m_event_replay_tracker.wait_for_ops(ctx);
       });
     ctx = new FunctionContext([this, ctx](int r) {
         m_local_journal->remove_listener(

src/tools/rbd_mirror/ImageReplayer.h

@@ -9,6 +9,7 @@
 #include <vector>
 
 #include "include/atomic.h"
+#include "common/AsyncOpTracker.h"
 #include "common/Mutex.h"
 #include "common/WorkQueue.h"
 #include "include/rados/librados.hpp"
@@ -269,6 +270,7 @@ class ImageReplayer {
   cls::journal::Tag m_replay_tag;
   librbd::journal::TagData m_replay_tag_data;
   librbd::journal::EventEntry m_event_entry;
+  AsyncOpTracker m_event_replay_tracker;
 
   struct RemoteJournalerListener : public ::journal::JournalMetadataListener {
     ImageReplayer *replayer;