ceph/src/test/rbd_mirror/image_replayer/journal/test_mock_Replayer.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 #include "test/rbd_mirror/test_mock_fixture.h"
   5 #include "librbd/journal/Types.h"
   6 #include "librbd/journal/TypeTraits.h"
   7 #include "tools/rbd_mirror/Threads.h"
   8 #include "tools/rbd_mirror/image_replayer/CloseImageRequest.h"
   9 #include "tools/rbd_mirror/image_replayer/ReplayerListener.h"
  10 #include "tools/rbd_mirror/image_replayer/Utils.h"
  11 #include "tools/rbd_mirror/image_replayer/journal/Replayer.h"
  12 #include "tools/rbd_mirror/image_replayer/journal/EventPreprocessor.h"
  13 #include "tools/rbd_mirror/image_replayer/journal/ReplayStatusFormatter.h"
  14 #include "tools/rbd_mirror/image_replayer/journal/StateBuilder.h"
  15 #include "test/journal/mock/MockJournaler.h"
  16 #include "test/librbd/mock/MockImageCtx.h"
  17 #include "test/rbd_mirror/mock/MockContextWQ.h"
  18 #include "test/rbd_mirror/mock/MockSafeTimer.h"
  19 #include <boost/intrusive_ptr.hpp>
  20
  21 namespace librbd {
  22
  23 namespace {
  24
  25 struct MockTestJournal;
  26
  27 struct MockTestImageCtx : public librbd::MockImageCtx {
  28   explicit MockTestImageCtx(librbd::ImageCtx &image_ctx,
  29                             MockTestJournal& mock_test_journal)
  30     : librbd::MockImageCtx(image_ctx), journal(&mock_test_journal) {
  31   }
  32
  33   MockTestJournal* journal = nullptr;
  34 };
  35
  36 struct MockTestJournal : public MockJournal {
  37   MOCK_METHOD2(start_external_replay, void(journal::Replay<MockTestImageCtx> **,
  38                                            Context *on_start));
  39   MOCK_METHOD0(stop_external_replay, void());
  40 };
  41
  42 } // anonymous namespace
  43
  44 namespace journal {
  45
  46 template <>
  47 struct TypeTraits<librbd::MockTestImageCtx> {
  48   typedef ::journal::MockJournaler Journaler;
  49   typedef ::journal::MockReplayEntryProxy ReplayEntry;
  50 };
  51
  52 template<>
  53 struct Replay<MockTestImageCtx> {
  54   MOCK_METHOD2(decode, int(bufferlist::const_iterator *, EventEntry *));
  55   MOCK_METHOD3(process, void(const EventEntry &, Context *, Context *));
  56   MOCK_METHOD1(flush, void(Context*));
  57   MOCK_METHOD2(shut_down, void(bool, Context*));
  58 };
  59
  60 } // namespace journal
  61 } // namespace librbd
  62
  63 namespace boost {
  64
  65 template<>
  66 struct intrusive_ptr<librbd::MockTestJournal> {
  67   intrusive_ptr() {
  68   }
  69   intrusive_ptr(librbd::MockTestJournal* mock_test_journal)
  70     : mock_test_journal(mock_test_journal) {
  71   }
  72
  73   librbd::MockTestJournal* operator->() {
  74     return mock_test_journal;
  75   }
  76
  77   void reset() {
  78     mock_test_journal = nullptr;
  79   }
  80
  81   const librbd::MockTestJournal* get() const {
  82     return mock_test_journal;
  83   }
  84
  85   template<typename T>
  86   bool operator==(T* t) const {
  87     return (mock_test_journal == t);
  88   }
  89
  90   librbd::MockTestJournal* mock_test_journal = nullptr;
  91 };
  92
  93 } // namespace boost
  94
  95 namespace rbd {
  96 namespace mirror {
  97
  98 template <>
  99 struct Threads<librbd::MockTestImageCtx> {
 100   MockSafeTimer *timer;
 101   ceph::mutex &timer_lock;
 102
 103   MockContextWQ *work_queue;
 104
 105   Threads(Threads<librbd::ImageCtx>* threads)
 106     : timer(new MockSafeTimer()),
 107       timer_lock(threads->timer_lock),
 108       work_queue(new MockContextWQ()) {
 109   }
 110   ~Threads() {
 111     delete timer;
 112     delete work_queue;
 113   }
 114 };
 115
 116 namespace {
 117
 118 struct MockReplayerListener : public image_replayer::ReplayerListener {
 119   MOCK_METHOD0(handle_notification, void());
 120 };
 121
 122 } // anonymous namespace
 123
 124 namespace image_replayer {
 125
 126 template<>
 127 struct CloseImageRequest<librbd::MockTestImageCtx> {
 128   static CloseImageRequest* s_instance;
 129   librbd::MockTestImageCtx **image_ctx = nullptr;
 130   Context *on_finish = nullptr;
 131
 132   static CloseImageRequest* create(librbd::MockTestImageCtx **image_ctx,
 133                                    Context *on_finish) {
 134     ceph_assert(s_instance != nullptr);
 135     s_instance->image_ctx = image_ctx;
 136     s_instance->on_finish = on_finish;
 137     return s_instance;
 138   }
 139
 140   CloseImageRequest() {
 141     ceph_assert(s_instance == nullptr);
 142     s_instance = this;
 143   }
 144
 145   ~CloseImageRequest() {
 146     ceph_assert(s_instance == this);
 147     s_instance = nullptr;
 148   }
 149
 150   MOCK_METHOD0(send, void());
 151 };
 152
 153 CloseImageRequest<librbd::MockTestImageCtx>* CloseImageRequest<librbd::MockTestImageCtx>::s_instance = nullptr;
 154
 155 namespace journal {
 156
 157 template <>
 158 struct EventPreprocessor<librbd::MockTestImageCtx> {
 159   static EventPreprocessor *s_instance;
 160
 161   static EventPreprocessor *create(librbd::MockTestImageCtx &local_image_ctx,
 162                                    ::journal::MockJournaler &remote_journaler,
 163                                    const std::string &local_mirror_uuid,
 164                                    librbd::journal::MirrorPeerClientMeta *client_meta,
 165                                    MockContextWQ *work_queue) {
 166     ceph_assert(s_instance != nullptr);
 167     return s_instance;
 168   }
 169
 170   static void destroy(EventPreprocessor* processor) {
 171   }
 172
 173   EventPreprocessor() {
 174     ceph_assert(s_instance == nullptr);
 175     s_instance = this;
 176   }
 177
 178   ~EventPreprocessor() {
 179     ceph_assert(s_instance == this);
 180     s_instance = nullptr;
 181   }
 182
 183   MOCK_METHOD1(is_required, bool(const librbd::journal::EventEntry &));
 184   MOCK_METHOD2(preprocess, void(librbd::journal::EventEntry *, Context *));
 185 };
 186
 187 template<>
 188 struct ReplayStatusFormatter<librbd::MockTestImageCtx> {
 189   static ReplayStatusFormatter* s_instance;
 190
 191   static ReplayStatusFormatter* create(::journal::MockJournaler *journaler,
 192                                        const std::string &mirror_uuid) {
 193     ceph_assert(s_instance != nullptr);
 194     return s_instance;
 195   }
 196
 197   static void destroy(ReplayStatusFormatter* formatter) {
 198   }
 199
 200   ReplayStatusFormatter() {
 201     ceph_assert(s_instance == nullptr);
 202     s_instance = this;
 203   }
 204
 205   ~ReplayStatusFormatter() {
 206     ceph_assert(s_instance == this);
 207     s_instance = nullptr;
 208   }
 209
 210   MOCK_METHOD2(get_or_send_update, bool(std::string *description, Context *on_finish));
 211 };
 212
 213 template<>
 214 struct StateBuilder<librbd::MockTestImageCtx> {
 215   StateBuilder(librbd::MockTestImageCtx& local_image_ctx,
 216                ::journal::MockJournaler& remote_journaler,
 217                const librbd::journal::MirrorPeerClientMeta& remote_client_meta)
 218     : local_image_ctx(&local_image_ctx),
 219       remote_journaler(&remote_journaler),
 220       remote_client_meta(remote_client_meta) {
 221   }
 222
 223   librbd::MockTestImageCtx* local_image_ctx;
 224   std::string remote_mirror_uuid = "remote mirror uuid";
 225   ::journal::MockJournaler* remote_journaler = nullptr;
 226   librbd::journal::MirrorPeerClientMeta remote_client_meta;
 227 };
 228
 229 EventPreprocessor<librbd::MockTestImageCtx>* EventPreprocessor<librbd::MockTestImageCtx>::s_instance = nullptr;
 230 ReplayStatusFormatter<librbd::MockTestImageCtx>* ReplayStatusFormatter<librbd::MockTestImageCtx>::s_instance = nullptr;
 231
 232 } // namespace journal
 233 } // namespace image_replayer
 234 } // namespace mirror
 235 } // namespace rbd
 236
 237 #include "tools/rbd_mirror/image_replayer/journal/Replayer.cc"
 238
 239 namespace rbd {
 240 namespace mirror {
 241 namespace image_replayer {
 242 namespace journal {
 243
 244 using ::testing::_;
 245 using ::testing::AtLeast;
 246 using ::testing::DoAll;
 247 using ::testing::InSequence;
 248 using ::testing::Invoke;
 249 using ::testing::MatcherCast;
 250 using ::testing::Return;
 251 using ::testing::ReturnArg;
 252 using ::testing::SaveArg;
 253 using ::testing::SetArgPointee;
 254 using ::testing::WithArg;
 255
 256 class TestMockImageReplayerJournalReplayer : public TestMockFixture {
 257 public:
 258   typedef Replayer<librbd::MockTestImageCtx> MockReplayer;
 259   typedef EventPreprocessor<librbd::MockTestImageCtx> MockEventPreprocessor;
 260   typedef ReplayStatusFormatter<librbd::MockTestImageCtx> MockReplayStatusFormatter;
 261   typedef StateBuilder<librbd::MockTestImageCtx> MockStateBuilder;
 262   typedef Threads<librbd::MockTestImageCtx> MockThreads;
 263   typedef CloseImageRequest<librbd::MockTestImageCtx> MockCloseImageRequest;
 264   typedef librbd::journal::Replay<librbd::MockTestImageCtx> MockReplay;
 265
 266   void SetUp() override {
 267     TestMockFixture::SetUp();
 268
 269     librbd::RBD rbd;
 270     ASSERT_EQ(0, create_image(rbd, m_local_io_ctx, m_image_name, m_image_size));
 271     ASSERT_EQ(0, open_image(m_local_io_ctx, m_image_name, &m_local_image_ctx));
 272   }
 273
 274   bufferlist encode_tag_data(const librbd::journal::TagData &tag_data) {
 275     bufferlist bl;
 276     encode(tag_data, bl);
 277     return bl;
 278   }
 279
 280   void expect_work_queue_repeatedly(MockThreads &mock_threads) {
 281     EXPECT_CALL(*mock_threads.work_queue, queue(_, _))
 282       .WillRepeatedly(Invoke([this](Context *ctx, int r) {
 283           m_threads->work_queue->queue(ctx, r);
 284         }));
 285   }
 286
 287   void expect_add_event_after_repeatedly(MockThreads &mock_threads) {
 288     EXPECT_CALL(*mock_threads.timer, add_event_after(_, _))
 289       .WillRepeatedly(
 290         DoAll(Invoke([this](double seconds, Context *ctx) {
 291                        m_threads->timer->add_event_after(seconds, ctx);
 292                      }),
 293           ReturnArg<1>()));
 294     EXPECT_CALL(*mock_threads.timer, cancel_event(_))
 295       .WillRepeatedly(
 296         Invoke([this](Context *ctx) {
 297           return m_threads->timer->cancel_event(ctx);
 298         }));
 299   }
 300
 301   void expect_init(::journal::MockJournaler &mock_journaler, int r) {
 302     EXPECT_CALL(mock_journaler, init(_))
 303       .WillOnce(CompleteContext(m_threads->work_queue, r));
 304   }
 305
 306   void expect_stop_replay(::journal::MockJournaler &mock_journaler, int r) {
 307     EXPECT_CALL(mock_journaler, stop_replay(_))
 308       .WillOnce(CompleteContext(r));
 309   }
 310
 311   void expect_shut_down(MockReplay &mock_replay, bool cancel_ops, int r) {
 312     EXPECT_CALL(mock_replay, shut_down(cancel_ops, _))
 313       .WillOnce(WithArg<1>(CompleteContext(m_threads->work_queue, r)));
 314   }
 315
 316   void expect_get_cached_client(::journal::MockJournaler &mock_journaler,
 317                                 const std::string& client_id,
 318                                 const cls::journal::Client& client,
 319                                 const librbd::journal::ClientMeta& client_meta,
 320                                 int r) {
 321     librbd::journal::ClientData client_data;
 322     client_data.client_meta = client_meta;
 323
 324     cls::journal::Client client_copy{client};
 325     encode(client_data, client_copy.data);
 326
 327     EXPECT_CALL(mock_journaler, get_cached_client(client_id, _))
 328       .WillOnce(DoAll(SetArgPointee<1>(client_copy),
 329                       Return(r)));
 330   }
 331
 332   void expect_start_external_replay(librbd::MockTestJournal &mock_journal,
 333                                     MockReplay *mock_replay, int r) {
 334     EXPECT_CALL(mock_journal, start_external_replay(_, _))
 335       .WillOnce(DoAll(SetArgPointee<0>(mock_replay),
 336                       WithArg<1>(CompleteContext(m_threads->work_queue, r))));
 337   }
 338
 339   void expect_is_tag_owner(librbd::MockTestJournal &mock_journal,
 340                            bool is_owner) {
 341     EXPECT_CALL(mock_journal, is_tag_owner()).WillOnce(Return(is_owner));
 342   }
 343
 344   void expect_is_resync_requested(librbd::MockTestJournal &mock_journal,
 345                                   int r, bool resync_requested) {
 346     EXPECT_CALL(mock_journal, is_resync_requested(_)).WillOnce(
 347       DoAll(SetArgPointee<0>(resync_requested),
 348             Return(r)));
 349   }
 350
 351   void expect_get_commit_tid_in_debug(
 352     ::journal::MockReplayEntry &mock_replay_entry) {
 353     // It is used in debug messages and depends on debug level
 354     EXPECT_CALL(mock_replay_entry, get_commit_tid())
 355     .Times(AtLeast(0))
 356       .WillRepeatedly(Return(0));
 357   }
 358
 359   void expect_get_tag_tid_in_debug(librbd::MockTestJournal &mock_journal) {
 360     // It is used in debug messages and depends on debug level
 361     EXPECT_CALL(mock_journal, get_tag_tid()).Times(AtLeast(0))
 362       .WillRepeatedly(Return(0));
 363   }
 364
 365   void expect_committed(::journal::MockReplayEntry &mock_replay_entry,
 366                         ::journal::MockJournaler &mock_journaler, int times) {
 367     EXPECT_CALL(mock_replay_entry, get_data()).Times(times);
 368     EXPECT_CALL(mock_journaler, committed(
 369                   MatcherCast<const ::journal::MockReplayEntryProxy&>(_)))
 370       .Times(times);
 371   }
 372
 373   void expect_try_pop_front(::journal::MockJournaler &mock_journaler,
 374                             uint64_t replay_tag_tid, bool entries_available) {
 375     EXPECT_CALL(mock_journaler, try_pop_front(_, _))
 376       .WillOnce(DoAll(SetArgPointee<0>(::journal::MockReplayEntryProxy()),
 377                       SetArgPointee<1>(replay_tag_tid),
 378                       Return(entries_available)));
 379   }
 380
 381   void expect_try_pop_front_return_no_entries(
 382     ::journal::MockJournaler &mock_journaler, Context *on_finish) {
 383     EXPECT_CALL(mock_journaler, try_pop_front(_, _))
 384       .WillOnce(DoAll(Invoke([on_finish](::journal::MockReplayEntryProxy *e,
 385                                          uint64_t *t) {
 386                                on_finish->complete(0);
 387                              }),
 388                       Return(false)));
 389   }
 390
 391   void expect_get_tag(::journal::MockJournaler &mock_journaler,
 392                       const cls::journal::Tag &tag, int r) {
 393     EXPECT_CALL(mock_journaler, get_tag(_, _, _))
 394       .WillOnce(DoAll(SetArgPointee<1>(tag),
 395                       WithArg<2>(CompleteContext(r))));
 396   }
 397
 398   void expect_allocate_tag(librbd::MockTestJournal &mock_journal, int r) {
 399     EXPECT_CALL(mock_journal, allocate_tag(_, _, _))
 400       .WillOnce(WithArg<2>(CompleteContext(r)));
 401   }
 402
 403   void expect_preprocess(MockEventPreprocessor &mock_event_preprocessor,
 404                          bool required, int r) {
 405     EXPECT_CALL(mock_event_preprocessor, is_required(_))
 406       .WillOnce(Return(required));
 407     if (required) {
 408       EXPECT_CALL(mock_event_preprocessor, preprocess(_, _))
 409         .WillOnce(WithArg<1>(CompleteContext(r)));
 410     }
 411   }
 412
 413   void expect_process(MockReplay &mock_replay,
 414                       int on_ready_r, int on_commit_r) {
 415     EXPECT_CALL(mock_replay, process(_, _, _))
 416       .WillOnce(DoAll(WithArg<1>(CompleteContext(on_ready_r)),
 417                       WithArg<2>(CompleteContext(on_commit_r))));
 418   }
 419
 420   void expect_flush(MockReplay& mock_replay, int r) {
 421     EXPECT_CALL(mock_replay, flush(_))
 422       .WillOnce(CompleteContext(m_threads->work_queue, r));
 423   }
 424
 425   void expect_flush_commit_position(::journal::MockJournaler& mock_journal,
 426                                     int r) {
 427     EXPECT_CALL(mock_journal, flush_commit_position(_))
 428       .WillOnce(CompleteContext(m_threads->work_queue, r));
 429   }
 430
 431   void expect_get_tag_data(librbd::MockTestJournal& mock_local_journal,
 432                            const librbd::journal::TagData& tag_data) {
 433     EXPECT_CALL(mock_local_journal, get_tag_data())
 434       .WillOnce(Return(tag_data));
 435   }
 436
 437   void expect_send(MockCloseImageRequest &mock_close_image_request, int r) {
 438     EXPECT_CALL(mock_close_image_request, send())
 439       .WillOnce(Invoke([this, &mock_close_image_request, r]() {
 440             *mock_close_image_request.image_ctx = nullptr;
 441             m_threads->work_queue->queue(mock_close_image_request.on_finish, r);
 442           }));
 443   }
 444
 445   void expect_notification(MockThreads& mock_threads,
 446                            MockReplayerListener& mock_replayer_listener) {
 447     EXPECT_CALL(mock_replayer_listener, handle_notification())
 448       .WillOnce(Invoke([this]() {
 449           std::unique_lock locker{m_lock};
 450           m_notified = true;
 451           m_cond.notify_all();
 452         }));
 453   }
 454
 455   int wait_for_notification() {
 456     std::unique_lock locker{m_lock};
 457     while (!m_notified) {
 458       if (m_cond.wait_for(locker, 10s) == std::cv_status::timeout) {
 459         return -ETIMEDOUT;
 460       }
 461     }
 462     m_notified = false;
 463     return 0;
 464   }
 465
 466   int init_entry_replayer(MockReplayer& mock_replayer,
 467                           MockThreads& mock_threads,
 468                           MockReplayerListener& mock_replayer_listener,
 469                           librbd::MockTestJournal& mock_local_journal,
 470                           ::journal::MockJournaler& mock_remote_journaler,
 471                           MockReplay& mock_local_journal_replay,
 472                           librbd::journal::Listener** local_journal_listener,
 473                           ::journal::ReplayHandler** remote_replay_handler,
 474                           ::journal::JournalMetadataListener** remote_journal_listener) {
 475     expect_init(mock_remote_journaler, 0);
 476     EXPECT_CALL(mock_remote_journaler, add_listener(_))
 477       .WillOnce(SaveArg<0>(remote_journal_listener));
 478     expect_get_cached_client(mock_remote_journaler, "local mirror uuid", {},
 479                              {librbd::journal::MirrorPeerClientMeta{}}, 0);
 480     expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
 481                                  0);
 482     EXPECT_CALL(mock_local_journal, add_listener(_))
 483       .WillOnce(SaveArg<0>(local_journal_listener));
 484     expect_is_tag_owner(mock_local_journal, false);
 485     expect_is_resync_requested(mock_local_journal, 0, false);
 486     EXPECT_CALL(mock_remote_journaler, start_live_replay(_, _))
 487       .WillOnce(SaveArg<0>(remote_replay_handler));
 488     expect_notification(mock_threads, mock_replayer_listener);
 489
 490     C_SaferCond init_ctx;
 491     mock_replayer.init(&init_ctx);
 492     int r = init_ctx.wait();
 493     if (r < 0) {
 494       return r;
 495     }
 496
 497     return wait_for_notification();
 498   }
 499
 500   int shut_down_entry_replayer(MockReplayer& mock_replayer,
 501                                MockThreads& mock_threads,
 502                                librbd::MockTestJournal& mock_local_journal,
 503                                ::journal::MockJournaler& mock_remote_journaler,
 504                                MockReplay& mock_local_journal_replay) {
 505     expect_shut_down(mock_local_journal_replay, true, 0);
 506     EXPECT_CALL(mock_local_journal, remove_listener(_));
 507     EXPECT_CALL(mock_local_journal, stop_external_replay());
 508     MockCloseImageRequest mock_close_image_request;
 509     expect_send(mock_close_image_request, 0);
 510     expect_stop_replay(mock_remote_journaler, 0);
 511     EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 512
 513     C_SaferCond shutdown_ctx;
 514     mock_replayer.shut_down(&shutdown_ctx);
 515     return shutdown_ctx.wait();
 516   }
 517
 518   librbd::ImageCtx* m_local_image_ctx = nullptr;
 519
 520   ceph::mutex m_lock = ceph::make_mutex(
 521     "TestMockImageReplayerJournalReplayer");
 522   ceph::condition_variable m_cond;
 523   bool m_notified = false;
 524 };
 525
 526 TEST_F(TestMockImageReplayerJournalReplayer, InitShutDown) {
 527   librbd::MockTestJournal mock_local_journal;
 528   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 529                                                 mock_local_journal};
 530   ::journal::MockJournaler mock_remote_journaler;
 531   MockReplayerListener mock_replayer_listener;
 532   MockThreads mock_threads{m_threads};
 533   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 534                                       mock_remote_journaler,
 535                                       {});
 536   MockReplayer mock_replayer{
 537     &mock_threads, "local mirror uuid", &mock_state_builder,
 538     &mock_replayer_listener};
 539
 540   expect_work_queue_repeatedly(mock_threads);
 541
 542   InSequence seq;
 543
 544   MockReplay mock_local_journal_replay;
 545   MockEventPreprocessor mock_event_preprocessor;
 546   MockReplayStatusFormatter mock_replay_status_formatter;
 547   librbd::journal::Listener* local_journal_listener = nullptr;
 548   ::journal::ReplayHandler* remote_replay_handler = nullptr;
 549   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
 550   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
 551                                    mock_replayer_listener, mock_local_journal,
 552                                    mock_remote_journaler,
 553                                    mock_local_journal_replay,
 554                                    &local_journal_listener,
 555                                    &remote_replay_handler,
 556                                    &remote_journaler_listener));
 557
 558   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
 559                                         mock_local_journal,
 560                                         mock_remote_journaler,
 561                                         mock_local_journal_replay));
 562 }
 563
 564 TEST_F(TestMockImageReplayerJournalReplayer, InitNoLocalJournal) {
 565   librbd::MockTestJournal mock_local_journal;
 566   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 567                                                 mock_local_journal};
 568   ::journal::MockJournaler mock_remote_journaler;
 569   MockReplayerListener mock_replayer_listener;
 570   MockThreads mock_threads{m_threads};
 571
 572   mock_local_image_ctx.journal = nullptr;
 573   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 574                                       mock_remote_journaler,
 575                                       {});
 576   MockReplayer mock_replayer{
 577     &mock_threads, "local mirror uuid", &mock_state_builder,
 578     &mock_replayer_listener};
 579
 580   expect_work_queue_repeatedly(mock_threads);
 581
 582   InSequence seq;
 583
 584   MockCloseImageRequest mock_close_image_request;
 585   expect_send(mock_close_image_request, 0);
 586
 587   C_SaferCond init_ctx;
 588   mock_replayer.init(&init_ctx);
 589   ASSERT_EQ(-EINVAL, init_ctx.wait());
 590 }
 591
 592 TEST_F(TestMockImageReplayerJournalReplayer, InitRemoteJournalerError) {
 593   librbd::MockTestJournal mock_local_journal;
 594   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 595                                                 mock_local_journal};
 596   ::journal::MockJournaler mock_remote_journaler;
 597   MockReplayerListener mock_replayer_listener;
 598   MockThreads mock_threads{m_threads};
 599   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 600                                       mock_remote_journaler,
 601                                       {});
 602   MockReplayer mock_replayer{
 603     &mock_threads, "local mirror uuid", &mock_state_builder,
 604     &mock_replayer_listener};
 605
 606   expect_work_queue_repeatedly(mock_threads);
 607
 608   InSequence seq;
 609
 610   expect_init(mock_remote_journaler, -EINVAL);
 611   MockCloseImageRequest mock_close_image_request;
 612   expect_send(mock_close_image_request, 0);
 613
 614   C_SaferCond init_ctx;
 615   mock_replayer.init(&init_ctx);
 616   ASSERT_EQ(-EINVAL, init_ctx.wait());
 617 }
 618
 619 TEST_F(TestMockImageReplayerJournalReplayer, InitRemoteJournalerGetClientError) {
 620   librbd::MockTestJournal mock_local_journal;
 621   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 622                                                 mock_local_journal};
 623   ::journal::MockJournaler mock_remote_journaler;
 624   MockReplayerListener mock_replayer_listener;
 625   MockThreads mock_threads{m_threads};
 626   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 627                                       mock_remote_journaler,
 628                                       {});
 629   MockReplayer mock_replayer{
 630     &mock_threads, "local mirror uuid", &mock_state_builder,
 631     &mock_replayer_listener};
 632
 633   expect_work_queue_repeatedly(mock_threads);
 634
 635   InSequence seq;
 636
 637   expect_init(mock_remote_journaler, 0);
 638   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 639   expect_get_cached_client(mock_remote_journaler, "local mirror uuid", {},
 640                            {librbd::journal::MirrorPeerClientMeta{}}, -EINVAL);
 641   MockCloseImageRequest mock_close_image_request;
 642   expect_send(mock_close_image_request, 0);
 643   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 644
 645   C_SaferCond init_ctx;
 646   mock_replayer.init(&init_ctx);
 647   ASSERT_EQ(-EINVAL, init_ctx.wait());
 648 }
 649
 650 TEST_F(TestMockImageReplayerJournalReplayer, InitLocalJournalStartExternalReplayError) {
 651   librbd::MockTestJournal mock_local_journal;
 652   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 653                                                 mock_local_journal};
 654   ::journal::MockJournaler mock_remote_journaler;
 655   MockReplayerListener mock_replayer_listener;
 656   MockThreads mock_threads{m_threads};
 657   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 658                                       mock_remote_journaler,
 659                                       {});
 660   MockReplayer mock_replayer{
 661     &mock_threads, "local mirror uuid", &mock_state_builder,
 662     &mock_replayer_listener};
 663
 664   expect_work_queue_repeatedly(mock_threads);
 665
 666   InSequence seq;
 667
 668   expect_init(mock_remote_journaler, 0);
 669   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 670   expect_get_cached_client(mock_remote_journaler, "local mirror uuid", {},
 671                            {librbd::journal::MirrorPeerClientMeta{}}, 0);
 672   expect_start_external_replay(mock_local_journal, nullptr, -EINVAL);
 673   MockCloseImageRequest mock_close_image_request;
 674   expect_send(mock_close_image_request, 0);
 675   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 676
 677   C_SaferCond init_ctx;
 678   mock_replayer.init(&init_ctx);
 679   ASSERT_EQ(-EINVAL, init_ctx.wait());
 680 }
 681
 682 TEST_F(TestMockImageReplayerJournalReplayer, InitIsPromoted) {
 683   librbd::MockTestJournal mock_local_journal;
 684   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 685                                                 mock_local_journal};
 686   ::journal::MockJournaler mock_remote_journaler;
 687   MockReplayerListener mock_replayer_listener;
 688   MockThreads mock_threads{m_threads};
 689   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 690                                       mock_remote_journaler,
 691                                       {});
 692   MockReplayer mock_replayer{
 693     &mock_threads, "local mirror uuid", &mock_state_builder,
 694     &mock_replayer_listener};
 695
 696   expect_work_queue_repeatedly(mock_threads);
 697
 698   InSequence seq;
 699
 700   expect_init(mock_remote_journaler, 0);
 701   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 702   expect_get_cached_client(mock_remote_journaler, "local mirror uuid", {},
 703                            {librbd::journal::MirrorPeerClientMeta{}}, 0);
 704   MockReplay mock_local_journal_replay;
 705   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
 706                                0);
 707   EXPECT_CALL(mock_local_journal, add_listener(_));
 708   expect_is_tag_owner(mock_local_journal, true);
 709   expect_notification(mock_threads, mock_replayer_listener);
 710
 711   C_SaferCond init_ctx;
 712   mock_replayer.init(&init_ctx);
 713   ASSERT_EQ(0, init_ctx.wait());
 714   ASSERT_EQ(0, wait_for_notification());
 715
 716   expect_shut_down(mock_local_journal_replay, true, 0);
 717   EXPECT_CALL(mock_local_journal, remove_listener(_));
 718   EXPECT_CALL(mock_local_journal, stop_external_replay());
 719   MockCloseImageRequest mock_close_image_request;
 720   expect_send(mock_close_image_request, 0);
 721   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 722
 723   C_SaferCond shutdown_ctx;
 724   mock_replayer.shut_down(&shutdown_ctx);
 725   ASSERT_EQ(0, shutdown_ctx.wait());
 726 }
 727
 728 TEST_F(TestMockImageReplayerJournalReplayer, InitDisconnected) {
 729   librbd::MockTestJournal mock_local_journal;
 730   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 731                                                 mock_local_journal};
 732   mock_local_image_ctx.config.set_val("rbd_mirroring_resync_after_disconnect",
 733                                       "false");
 734
 735   ::journal::MockJournaler mock_remote_journaler;
 736   MockReplayerListener mock_replayer_listener;
 737   MockThreads mock_threads{m_threads};
 738   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 739                                       mock_remote_journaler,
 740                                       {});
 741   MockReplayer mock_replayer{
 742     &mock_threads, "local mirror uuid", &mock_state_builder,
 743     &mock_replayer_listener};
 744
 745   expect_work_queue_repeatedly(mock_threads);
 746
 747   InSequence seq;
 748
 749   expect_init(mock_remote_journaler, 0);
 750   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 751   expect_get_cached_client(mock_remote_journaler, "local mirror uuid",
 752                            {{}, {}, {},
 753                             cls::journal::CLIENT_STATE_DISCONNECTED},
 754                            {librbd::journal::MirrorPeerClientMeta{
 755                               mock_local_image_ctx.id}}, 0);
 756   MockCloseImageRequest mock_close_image_request;
 757   expect_send(mock_close_image_request, 0);
 758   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 759
 760   C_SaferCond init_ctx;
 761   mock_replayer.init(&init_ctx);
 762   ASSERT_EQ(-ENOTCONN, init_ctx.wait());
 763   ASSERT_FALSE(mock_replayer.is_resync_requested());
 764 }
 765
 766 TEST_F(TestMockImageReplayerJournalReplayer, InitDisconnectedResync) {
 767   librbd::MockTestJournal mock_local_journal;
 768   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 769                                                 mock_local_journal};
 770   mock_local_image_ctx.config.set_val("rbd_mirroring_resync_after_disconnect",
 771                                       "true");
 772
 773   ::journal::MockJournaler mock_remote_journaler;
 774   MockReplayerListener mock_replayer_listener;
 775   MockThreads mock_threads{m_threads};
 776   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 777                                       mock_remote_journaler,
 778                                       {});
 779   MockReplayer mock_replayer{
 780     &mock_threads, "local mirror uuid", &mock_state_builder,
 781     &mock_replayer_listener};
 782
 783   expect_work_queue_repeatedly(mock_threads);
 784
 785   InSequence seq;
 786
 787   expect_init(mock_remote_journaler, 0);
 788   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 789   expect_get_cached_client(mock_remote_journaler, "local mirror uuid",
 790                            {{}, {}, {},
 791                             cls::journal::CLIENT_STATE_DISCONNECTED},
 792                            {librbd::journal::MirrorPeerClientMeta{
 793                               mock_local_image_ctx.id}}, 0);
 794   MockCloseImageRequest mock_close_image_request;
 795   expect_send(mock_close_image_request, 0);
 796   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 797
 798   C_SaferCond init_ctx;
 799   mock_replayer.init(&init_ctx);
 800   ASSERT_EQ(-ENOTCONN, init_ctx.wait());
 801   ASSERT_TRUE(mock_replayer.is_resync_requested());
 802 }
 803
 804 TEST_F(TestMockImageReplayerJournalReplayer, InitResyncRequested) {
 805   librbd::MockTestJournal mock_local_journal;
 806   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 807                                                 mock_local_journal};
 808   ::journal::MockJournaler mock_remote_journaler;
 809   MockReplayerListener mock_replayer_listener;
 810   MockThreads mock_threads{m_threads};
 811   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 812                                       mock_remote_journaler,
 813                                       {});
 814   MockReplayer mock_replayer{
 815     &mock_threads, "local mirror uuid", &mock_state_builder,
 816     &mock_replayer_listener};
 817
 818   expect_work_queue_repeatedly(mock_threads);
 819
 820   InSequence seq;
 821
 822   expect_init(mock_remote_journaler, 0);
 823   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 824   expect_get_cached_client(mock_remote_journaler, "local mirror uuid", {},
 825                            {librbd::journal::MirrorPeerClientMeta{}}, 0);
 826   MockReplay mock_local_journal_replay;
 827   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
 828                                0);
 829   EXPECT_CALL(mock_local_journal, add_listener(_));
 830   expect_is_tag_owner(mock_local_journal, false);
 831   expect_is_resync_requested(mock_local_journal, 0, true);
 832   expect_notification(mock_threads, mock_replayer_listener);
 833
 834   C_SaferCond init_ctx;
 835   mock_replayer.init(&init_ctx);
 836   ASSERT_EQ(0, init_ctx.wait());
 837   ASSERT_EQ(0, wait_for_notification());
 838
 839   expect_shut_down(mock_local_journal_replay, true, 0);
 840   EXPECT_CALL(mock_local_journal, remove_listener(_));
 841   EXPECT_CALL(mock_local_journal, stop_external_replay());
 842   MockCloseImageRequest mock_close_image_request;
 843   expect_send(mock_close_image_request, 0);
 844   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 845
 846   C_SaferCond shutdown_ctx;
 847   mock_replayer.shut_down(&shutdown_ctx);
 848   ASSERT_EQ(0, shutdown_ctx.wait());
 849 }
 850
 851 TEST_F(TestMockImageReplayerJournalReplayer, InitResyncRequestedError) {
 852   librbd::MockTestJournal mock_local_journal;
 853   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 854                                                 mock_local_journal};
 855   ::journal::MockJournaler mock_remote_journaler;
 856   MockReplayerListener mock_replayer_listener;
 857   MockThreads mock_threads{m_threads};
 858   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 859                                       mock_remote_journaler,
 860                                       {});
 861   MockReplayer mock_replayer{
 862     &mock_threads, "local mirror uuid", &mock_state_builder,
 863     &mock_replayer_listener};
 864
 865   expect_work_queue_repeatedly(mock_threads);
 866
 867   InSequence seq;
 868
 869   expect_init(mock_remote_journaler, 0);
 870   EXPECT_CALL(mock_remote_journaler, add_listener(_));
 871   expect_get_cached_client(mock_remote_journaler, "local mirror uuid", {},
 872                            {librbd::journal::MirrorPeerClientMeta{}}, 0);
 873   MockReplay mock_local_journal_replay;
 874   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
 875                                0);
 876   EXPECT_CALL(mock_local_journal, add_listener(_));
 877   expect_is_tag_owner(mock_local_journal, false);
 878   expect_is_resync_requested(mock_local_journal, -EINVAL, false);
 879   expect_notification(mock_threads, mock_replayer_listener);
 880
 881   C_SaferCond init_ctx;
 882   mock_replayer.init(&init_ctx);
 883   ASSERT_EQ(0, init_ctx.wait());
 884   ASSERT_EQ(0, wait_for_notification());
 885   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
 886
 887   expect_shut_down(mock_local_journal_replay, true, 0);
 888   EXPECT_CALL(mock_local_journal, remove_listener(_));
 889   EXPECT_CALL(mock_local_journal, stop_external_replay());
 890   MockCloseImageRequest mock_close_image_request;
 891   expect_send(mock_close_image_request, 0);
 892   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 893
 894   C_SaferCond shutdown_ctx;
 895   mock_replayer.shut_down(&shutdown_ctx);
 896   ASSERT_EQ(0, shutdown_ctx.wait());
 897 }
 898
 899 TEST_F(TestMockImageReplayerJournalReplayer, ShutDownLocalJournalReplayError) {
 900   librbd::MockTestJournal mock_local_journal;
 901   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 902                                                 mock_local_journal};
 903   ::journal::MockJournaler mock_remote_journaler;
 904   MockReplayerListener mock_replayer_listener;
 905   MockThreads mock_threads{m_threads};
 906   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 907                                       mock_remote_journaler,
 908                                       {});
 909   MockReplayer mock_replayer{
 910     &mock_threads, "local mirror uuid", &mock_state_builder,
 911     &mock_replayer_listener};
 912
 913   expect_work_queue_repeatedly(mock_threads);
 914
 915   InSequence seq;
 916
 917   MockReplay mock_local_journal_replay;
 918   MockEventPreprocessor mock_event_preprocessor;
 919   MockReplayStatusFormatter mock_replay_status_formatter;
 920   librbd::journal::Listener* local_journal_listener = nullptr;
 921   ::journal::ReplayHandler* remote_replay_handler = nullptr;
 922   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
 923   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
 924                                    mock_replayer_listener, mock_local_journal,
 925                                    mock_remote_journaler,
 926                                    mock_local_journal_replay,
 927                                    &local_journal_listener,
 928                                    &remote_replay_handler,
 929                                    &remote_journaler_listener));
 930
 931   expect_shut_down(mock_local_journal_replay, true, -EINVAL);
 932   EXPECT_CALL(mock_local_journal, remove_listener(_));
 933   EXPECT_CALL(mock_local_journal, stop_external_replay());
 934   MockCloseImageRequest mock_close_image_request;
 935   expect_send(mock_close_image_request, 0);
 936   expect_stop_replay(mock_remote_journaler, 0);
 937   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 938
 939   C_SaferCond shutdown_ctx;
 940   mock_replayer.shut_down(&shutdown_ctx);
 941   ASSERT_EQ(-EINVAL, shutdown_ctx.wait());
 942 }
 943
 944 TEST_F(TestMockImageReplayerJournalReplayer, CloseLocalImageError) {
 945   librbd::MockTestJournal mock_local_journal;
 946   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 947                                                 mock_local_journal};
 948   ::journal::MockJournaler mock_remote_journaler;
 949   MockReplayerListener mock_replayer_listener;
 950   MockThreads mock_threads{m_threads};
 951   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 952                                       mock_remote_journaler,
 953                                       {});
 954   MockReplayer mock_replayer{
 955     &mock_threads, "local mirror uuid", &mock_state_builder,
 956     &mock_replayer_listener};
 957
 958   expect_work_queue_repeatedly(mock_threads);
 959
 960   InSequence seq;
 961
 962   MockReplay mock_local_journal_replay;
 963   MockEventPreprocessor mock_event_preprocessor;
 964   MockReplayStatusFormatter mock_replay_status_formatter;
 965   librbd::journal::Listener* local_journal_listener = nullptr;
 966   ::journal::ReplayHandler* remote_replay_handler = nullptr;
 967   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
 968   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
 969                                    mock_replayer_listener, mock_local_journal,
 970                                    mock_remote_journaler,
 971                                    mock_local_journal_replay,
 972                                    &local_journal_listener,
 973                                    &remote_replay_handler,
 974                                    &remote_journaler_listener));
 975
 976   expect_shut_down(mock_local_journal_replay, true, 0);
 977   EXPECT_CALL(mock_local_journal, remove_listener(_));
 978   EXPECT_CALL(mock_local_journal, stop_external_replay());
 979   MockCloseImageRequest mock_close_image_request;
 980   expect_send(mock_close_image_request, -EINVAL);
 981   expect_stop_replay(mock_remote_journaler, 0);
 982   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
 983
 984   C_SaferCond shutdown_ctx;
 985   mock_replayer.shut_down(&shutdown_ctx);
 986   ASSERT_EQ(-EINVAL, shutdown_ctx.wait());
 987 }
 988
 989 TEST_F(TestMockImageReplayerJournalReplayer, StopRemoteJournalerError) {
 990   librbd::MockTestJournal mock_local_journal;
 991   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
 992                                                 mock_local_journal};
 993   ::journal::MockJournaler mock_remote_journaler;
 994   MockReplayerListener mock_replayer_listener;
 995   MockThreads mock_threads{m_threads};
 996   MockStateBuilder mock_state_builder(mock_local_image_ctx,
 997                                       mock_remote_journaler,
 998                                       {});
 999   MockReplayer mock_replayer{
1000     &mock_threads, "local mirror uuid", &mock_state_builder,
1001     &mock_replayer_listener};
1002
1003   expect_work_queue_repeatedly(mock_threads);
1004
1005   InSequence seq;
1006
1007   MockReplay mock_local_journal_replay;
1008   MockEventPreprocessor mock_event_preprocessor;
1009   MockReplayStatusFormatter mock_replay_status_formatter;
1010   librbd::journal::Listener* local_journal_listener = nullptr;
1011   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1012   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1013   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1014                                    mock_replayer_listener, mock_local_journal,
1015                                    mock_remote_journaler,
1016                                    mock_local_journal_replay,
1017                                    &local_journal_listener,
1018                                    &remote_replay_handler,
1019                                    &remote_journaler_listener));
1020
1021   expect_shut_down(mock_local_journal_replay, true, 0);
1022   EXPECT_CALL(mock_local_journal, remove_listener(_));
1023   EXPECT_CALL(mock_local_journal, stop_external_replay());
1024   MockCloseImageRequest mock_close_image_request;
1025   expect_send(mock_close_image_request, 0);
1026   expect_stop_replay(mock_remote_journaler, -EPERM);
1027   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
1028
1029   C_SaferCond shutdown_ctx;
1030   mock_replayer.shut_down(&shutdown_ctx);
1031   ASSERT_EQ(-EPERM, shutdown_ctx.wait());
1032 }
1033
1034 TEST_F(TestMockImageReplayerJournalReplayer, Replay) {
1035   librbd::MockTestJournal mock_local_journal;
1036   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1037                                                 mock_local_journal};
1038   ::journal::MockJournaler mock_remote_journaler;
1039   MockReplayerListener mock_replayer_listener;
1040   MockThreads mock_threads{m_threads};
1041   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1042                                       mock_remote_journaler,
1043                                       {});
1044   MockReplayer mock_replayer{
1045     &mock_threads, "local mirror uuid", &mock_state_builder,
1046     &mock_replayer_listener};
1047
1048   ::journal::MockReplayEntry mock_replay_entry;
1049   expect_work_queue_repeatedly(mock_threads);
1050   expect_add_event_after_repeatedly(mock_threads);
1051   expect_get_commit_tid_in_debug(mock_replay_entry);
1052   expect_get_tag_tid_in_debug(mock_local_journal);
1053   expect_committed(mock_replay_entry, mock_remote_journaler, 2);
1054
1055   InSequence seq;
1056
1057   MockReplay mock_local_journal_replay;
1058   MockEventPreprocessor mock_event_preprocessor;
1059   MockReplayStatusFormatter mock_replay_status_formatter;
1060   librbd::journal::Listener* local_journal_listener = nullptr;
1061   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1062   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1063   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1064                                    mock_replayer_listener, mock_local_journal,
1065                                    mock_remote_journaler,
1066                                    mock_local_journal_replay,
1067                                    &local_journal_listener,
1068                                    &remote_replay_handler,
1069                                    &remote_journaler_listener));
1070
1071   cls::journal::Tag tag =
1072     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
1073                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1074                             true, 0, 0})};
1075
1076   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1077
1078   // replay_flush
1079   expect_shut_down(mock_local_journal_replay, false, 0);
1080   EXPECT_CALL(mock_local_journal, stop_external_replay());
1081   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1082                                0);
1083   expect_get_tag(mock_remote_journaler, tag, 0);
1084   expect_allocate_tag(mock_local_journal, 0);
1085
1086   // process
1087   EXPECT_CALL(mock_local_journal_replay, decode(_, _)).WillOnce(Return(0));
1088   expect_preprocess(mock_event_preprocessor, false, 0);
1089   expect_process(mock_local_journal_replay, 0, 0);
1090
1091   // the next event with preprocess
1092   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1093   EXPECT_CALL(mock_local_journal_replay, decode(_, _)).WillOnce(Return(0));
1094   expect_preprocess(mock_event_preprocessor, true, 0);
1095   expect_process(mock_local_journal_replay, 0, 0);
1096
1097   // attempt to process the next event
1098   C_SaferCond replay_ctx;
1099   expect_try_pop_front_return_no_entries(mock_remote_journaler, &replay_ctx);
1100
1101   // fire
1102   remote_replay_handler->handle_entries_available();
1103   ASSERT_EQ(0, replay_ctx.wait());
1104
1105   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1106                                         mock_local_journal,
1107                                         mock_remote_journaler,
1108                                         mock_local_journal_replay));
1109 }
1110
1111 TEST_F(TestMockImageReplayerJournalReplayer, DecodeError) {
1112   librbd::MockTestJournal mock_local_journal;
1113   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1114                                                 mock_local_journal};
1115   ::journal::MockJournaler mock_remote_journaler;
1116   MockReplayerListener mock_replayer_listener;
1117   MockThreads mock_threads{m_threads};
1118   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1119                                       mock_remote_journaler,
1120                                       {});
1121   MockReplayer mock_replayer{
1122     &mock_threads, "local mirror uuid", &mock_state_builder,
1123     &mock_replayer_listener};
1124
1125   ::journal::MockReplayEntry mock_replay_entry;
1126   expect_work_queue_repeatedly(mock_threads);
1127   expect_add_event_after_repeatedly(mock_threads);
1128   expect_get_commit_tid_in_debug(mock_replay_entry);
1129   expect_get_tag_tid_in_debug(mock_local_journal);
1130
1131   InSequence seq;
1132
1133   MockReplay mock_local_journal_replay;
1134   MockEventPreprocessor mock_event_preprocessor;
1135   MockReplayStatusFormatter mock_replay_status_formatter;
1136   librbd::journal::Listener* local_journal_listener = nullptr;
1137   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1138   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1139   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1140                                    mock_replayer_listener, mock_local_journal,
1141                                    mock_remote_journaler,
1142                                    mock_local_journal_replay,
1143                                    &local_journal_listener,
1144                                    &remote_replay_handler,
1145                                    &remote_journaler_listener));
1146
1147   cls::journal::Tag tag =
1148     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
1149                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1150                             true, 0, 0})};
1151
1152   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1153
1154   // replay_flush
1155   expect_shut_down(mock_local_journal_replay, false, 0);
1156   EXPECT_CALL(mock_local_journal, stop_external_replay());
1157   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1158                                0);
1159   expect_get_tag(mock_remote_journaler, tag, 0);
1160   expect_allocate_tag(mock_local_journal, 0);
1161
1162   // process
1163   EXPECT_CALL(mock_replay_entry, get_data());
1164   EXPECT_CALL(mock_local_journal_replay, decode(_, _))
1165     .WillOnce(Return(-EINVAL));
1166   expect_notification(mock_threads, mock_replayer_listener);
1167
1168   // fire
1169   remote_replay_handler->handle_entries_available();
1170   wait_for_notification();
1171
1172   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
1173
1174   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1175                                         mock_local_journal,
1176                                         mock_remote_journaler,
1177                                         mock_local_journal_replay));
1178 }
1179
1180 TEST_F(TestMockImageReplayerJournalReplayer, DelayedReplay) {
1181   librbd::MockTestJournal mock_local_journal;
1182   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1183                                                 mock_local_journal};
1184   ::journal::MockJournaler mock_remote_journaler;
1185   MockReplayerListener mock_replayer_listener;
1186   MockThreads mock_threads{m_threads};
1187   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1188                                       mock_remote_journaler,
1189                                       {});
1190   MockReplayer mock_replayer{
1191     &mock_threads, "local mirror uuid", &mock_state_builder,
1192     &mock_replayer_listener};
1193
1194   ::journal::MockReplayEntry mock_replay_entry;
1195   expect_work_queue_repeatedly(mock_threads);
1196   expect_add_event_after_repeatedly(mock_threads);
1197   expect_get_commit_tid_in_debug(mock_replay_entry);
1198   expect_get_tag_tid_in_debug(mock_local_journal);
1199   expect_committed(mock_replay_entry, mock_remote_journaler, 1);
1200
1201   InSequence seq;
1202
1203   MockReplay mock_local_journal_replay;
1204   MockEventPreprocessor mock_event_preprocessor;
1205   MockReplayStatusFormatter mock_replay_status_formatter;
1206   librbd::journal::Listener* local_journal_listener = nullptr;
1207   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1208   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1209   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1210                                    mock_replayer_listener, mock_local_journal,
1211                                    mock_remote_journaler,
1212                                    mock_local_journal_replay,
1213                                    &local_journal_listener,
1214                                    &remote_replay_handler,
1215                                    &remote_journaler_listener));
1216
1217   cls::journal::Tag tag =
1218     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
1219                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1220                             true, 0, 0})};
1221
1222   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1223
1224   // replay_flush
1225   expect_shut_down(mock_local_journal_replay, false, 0);
1226   EXPECT_CALL(mock_local_journal, stop_external_replay());
1227   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1228                                0);
1229   expect_get_tag(mock_remote_journaler, tag, 0);
1230   expect_allocate_tag(mock_local_journal, 0);
1231
1232   // process with delay
1233   EXPECT_CALL(mock_replay_entry, get_data());
1234   librbd::journal::EventEntry event_entry(
1235     librbd::journal::AioDiscardEvent(123, 345, 0), ceph_clock_now());
1236   EXPECT_CALL(mock_local_journal_replay, decode(_, _))
1237     .WillOnce(DoAll(SetArgPointee<1>(event_entry),
1238                     Return(0)));
1239
1240   Context* delayed_task_ctx = nullptr;
1241   EXPECT_CALL(*mock_threads.timer, add_event_after(_, _))
1242     .WillOnce(
1243       DoAll(Invoke([this, &delayed_task_ctx](double seconds, Context *ctx) {
1244                 std::unique_lock locker{m_lock};
1245                 delayed_task_ctx = ctx;
1246                 m_cond.notify_all();
1247             }),
1248             ReturnArg<1>()));
1249   expect_preprocess(mock_event_preprocessor, false, 0);
1250   expect_process(mock_local_journal_replay, 0, 0);
1251
1252   // attempt to process the next event
1253   C_SaferCond replay_ctx;
1254   expect_try_pop_front_return_no_entries(mock_remote_journaler, &replay_ctx);
1255
1256   // fire
1257   mock_local_image_ctx.mirroring_replay_delay = 600;
1258   remote_replay_handler->handle_entries_available();
1259   {
1260     std::unique_lock locker{m_lock};
1261     while (delayed_task_ctx == nullptr) {
1262       if (m_cond.wait_for(locker, 10s) == std::cv_status::timeout) {
1263         FAIL() << "timed out waiting for task";
1264         break;
1265       }
1266     }
1267   }
1268   {
1269     std::unique_lock timer_locker{mock_threads.timer_lock};
1270     delayed_task_ctx->complete(0);
1271   }
1272   ASSERT_EQ(0, replay_ctx.wait());
1273
1274   // add a pending (delayed) entry before stop
1275   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1276   C_SaferCond decode_ctx;
1277   EXPECT_CALL(mock_local_journal_replay, decode(_, _))
1278     .WillOnce(DoAll(Invoke([&decode_ctx](bufferlist::const_iterator* it,
1279                                          librbd::journal::EventEntry *e) {
1280                              decode_ctx.complete(0);
1281                            }),
1282                     Return(0)));
1283
1284   remote_replay_handler->handle_entries_available();
1285   ASSERT_EQ(0, decode_ctx.wait());
1286
1287   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1288                                         mock_local_journal,
1289                                         mock_remote_journaler,
1290                                         mock_local_journal_replay));
1291 }
1292
1293 TEST_F(TestMockImageReplayerJournalReplayer, ReplayNoMemoryError) {
1294   librbd::MockTestJournal mock_local_journal;
1295   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1296                                                 mock_local_journal};
1297   ::journal::MockJournaler mock_remote_journaler;
1298   MockReplayerListener mock_replayer_listener;
1299   MockThreads mock_threads{m_threads};
1300   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1301                                       mock_remote_journaler,
1302                                       {});
1303   MockReplayer mock_replayer{
1304     &mock_threads, "local mirror uuid", &mock_state_builder,
1305     &mock_replayer_listener};
1306
1307   expect_work_queue_repeatedly(mock_threads);
1308
1309   InSequence seq;
1310
1311   MockReplay mock_local_journal_replay;
1312   MockEventPreprocessor mock_event_preprocessor;
1313   MockReplayStatusFormatter mock_replay_status_formatter;
1314   librbd::journal::Listener* local_journal_listener = nullptr;
1315   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1316   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1317   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1318                                    mock_replayer_listener, mock_local_journal,
1319                                    mock_remote_journaler,
1320                                    mock_local_journal_replay,
1321                                    &local_journal_listener,
1322                                    &remote_replay_handler,
1323                                    &remote_journaler_listener));
1324
1325   expect_notification(mock_threads, mock_replayer_listener);
1326   remote_replay_handler->handle_complete(-ENOMEM);
1327
1328   wait_for_notification();
1329   ASSERT_EQ(false, mock_replayer.is_replaying());
1330   ASSERT_EQ(-ENOMEM, mock_replayer.get_error_code());
1331
1332   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1333                                         mock_local_journal,
1334                                         mock_remote_journaler,
1335                                         mock_local_journal_replay));
1336 }
1337
1338 TEST_F(TestMockImageReplayerJournalReplayer, LocalJournalForcePromoted) {
1339   librbd::MockTestJournal mock_local_journal;
1340   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1341                                                 mock_local_journal};
1342   ::journal::MockJournaler mock_remote_journaler;
1343   MockReplayerListener mock_replayer_listener;
1344   MockThreads mock_threads{m_threads};
1345   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1346                                       mock_remote_journaler,
1347                                       {});
1348   MockReplayer mock_replayer{
1349     &mock_threads, "local mirror uuid", &mock_state_builder,
1350     &mock_replayer_listener};
1351
1352   expect_work_queue_repeatedly(mock_threads);
1353
1354   InSequence seq;
1355
1356   MockReplay mock_local_journal_replay;
1357   MockEventPreprocessor mock_event_preprocessor;
1358   MockReplayStatusFormatter mock_replay_status_formatter;
1359   librbd::journal::Listener* local_journal_listener = nullptr;
1360   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1361   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1362   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1363                                    mock_replayer_listener, mock_local_journal,
1364                                    mock_remote_journaler,
1365                                    mock_local_journal_replay,
1366                                    &local_journal_listener,
1367                                    &remote_replay_handler,
1368                                    &remote_journaler_listener));
1369
1370   expect_notification(mock_threads, mock_replayer_listener);
1371   local_journal_listener->handle_promoted();
1372   wait_for_notification();
1373
1374   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1375                                         mock_local_journal,
1376                                         mock_remote_journaler,
1377                                         mock_local_journal_replay));
1378 }
1379
1380 TEST_F(TestMockImageReplayerJournalReplayer, LocalJournalResyncRequested) {
1381   librbd::MockTestJournal mock_local_journal;
1382   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1383                                                 mock_local_journal};
1384   ::journal::MockJournaler mock_remote_journaler;
1385   MockReplayerListener mock_replayer_listener;
1386   MockThreads mock_threads{m_threads};
1387   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1388                                       mock_remote_journaler,
1389                                       {});
1390   MockReplayer mock_replayer{
1391     &mock_threads, "local mirror uuid", &mock_state_builder,
1392     &mock_replayer_listener};
1393
1394   expect_work_queue_repeatedly(mock_threads);
1395
1396   InSequence seq;
1397
1398   MockReplay mock_local_journal_replay;
1399   MockEventPreprocessor mock_event_preprocessor;
1400   MockReplayStatusFormatter mock_replay_status_formatter;
1401   librbd::journal::Listener* local_journal_listener = nullptr;
1402   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1403   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1404   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1405                                    mock_replayer_listener, mock_local_journal,
1406                                    mock_remote_journaler,
1407                                    mock_local_journal_replay,
1408                                    &local_journal_listener,
1409                                    &remote_replay_handler,
1410                                    &remote_journaler_listener));
1411
1412   expect_notification(mock_threads, mock_replayer_listener);
1413   local_journal_listener->handle_resync();
1414   wait_for_notification();
1415
1416   ASSERT_TRUE(mock_replayer.is_resync_requested());
1417
1418   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1419                                         mock_local_journal,
1420                                         mock_remote_journaler,
1421                                         mock_local_journal_replay));
1422 }
1423
1424 TEST_F(TestMockImageReplayerJournalReplayer, RemoteJournalDisconnected) {
1425   librbd::MockTestJournal mock_local_journal;
1426   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1427                                                 mock_local_journal};
1428   mock_local_image_ctx.config.set_val("rbd_mirroring_resync_after_disconnect",
1429                                       "true");
1430
1431   ::journal::MockJournaler mock_remote_journaler;
1432   MockReplayerListener mock_replayer_listener;
1433   MockThreads mock_threads{m_threads};
1434   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1435                                       mock_remote_journaler,
1436                                       {});
1437   MockReplayer mock_replayer{
1438     &mock_threads, "local mirror uuid", &mock_state_builder,
1439     &mock_replayer_listener};
1440
1441   expect_work_queue_repeatedly(mock_threads);
1442
1443   InSequence seq;
1444
1445   MockReplay mock_local_journal_replay;
1446   MockEventPreprocessor mock_event_preprocessor;
1447   MockReplayStatusFormatter mock_replay_status_formatter;
1448   librbd::journal::Listener* local_journal_listener = nullptr;
1449   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1450   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1451   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1452                                    mock_replayer_listener, mock_local_journal,
1453                                    mock_remote_journaler,
1454                                    mock_local_journal_replay,
1455                                    &local_journal_listener,
1456                                    &remote_replay_handler,
1457                                    &remote_journaler_listener));
1458
1459   expect_get_cached_client(mock_remote_journaler, "local mirror uuid",
1460                            {{}, {}, {},
1461                             cls::journal::CLIENT_STATE_DISCONNECTED},
1462                            {librbd::journal::MirrorPeerClientMeta{
1463                               mock_local_image_ctx.id}}, 0);
1464   expect_notification(mock_threads, mock_replayer_listener);
1465
1466   remote_journaler_listener->handle_update(nullptr);
1467   wait_for_notification();
1468
1469   ASSERT_EQ(-ENOTCONN, mock_replayer.get_error_code());
1470   ASSERT_FALSE(mock_replayer.is_replaying());
1471   ASSERT_TRUE(mock_replayer.is_resync_requested());
1472
1473   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1474                                         mock_local_journal,
1475                                         mock_remote_journaler,
1476                                         mock_local_journal_replay));
1477 }
1478
1479 TEST_F(TestMockImageReplayerJournalReplayer, Flush) {
1480   librbd::MockTestJournal mock_local_journal;
1481   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1482                                                 mock_local_journal};
1483   ::journal::MockJournaler mock_remote_journaler;
1484   MockReplayerListener mock_replayer_listener;
1485   MockThreads mock_threads{m_threads};
1486   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1487                                       mock_remote_journaler,
1488                                       {});
1489   MockReplayer mock_replayer{
1490     &mock_threads, "local mirror uuid", &mock_state_builder,
1491     &mock_replayer_listener};
1492
1493   expect_work_queue_repeatedly(mock_threads);
1494
1495   InSequence seq;
1496
1497   MockReplay mock_local_journal_replay;
1498   MockEventPreprocessor mock_event_preprocessor;
1499   MockReplayStatusFormatter mock_replay_status_formatter;
1500   librbd::journal::Listener* local_journal_listener = nullptr;
1501   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1502   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1503   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1504                                    mock_replayer_listener, mock_local_journal,
1505                                    mock_remote_journaler,
1506                                    mock_local_journal_replay,
1507                                    &local_journal_listener,
1508                                    &remote_replay_handler,
1509                                    &remote_journaler_listener));
1510
1511   expect_flush(mock_local_journal_replay, 0);
1512   expect_flush_commit_position(mock_remote_journaler, 0);
1513
1514   C_SaferCond ctx;
1515   mock_replayer.flush(&ctx);
1516   ASSERT_EQ(0, ctx.wait());
1517
1518   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1519                                         mock_local_journal,
1520                                         mock_remote_journaler,
1521                                         mock_local_journal_replay));
1522 }
1523
1524 TEST_F(TestMockImageReplayerJournalReplayer, FlushError) {
1525   librbd::MockTestJournal mock_local_journal;
1526   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1527                                                 mock_local_journal};
1528   ::journal::MockJournaler mock_remote_journaler;
1529   MockReplayerListener mock_replayer_listener;
1530   MockThreads mock_threads{m_threads};
1531   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1532                                       mock_remote_journaler,
1533                                       {});
1534   MockReplayer mock_replayer{
1535     &mock_threads, "local mirror uuid", &mock_state_builder,
1536     &mock_replayer_listener};
1537
1538   expect_work_queue_repeatedly(mock_threads);
1539
1540   InSequence seq;
1541
1542   MockReplay mock_local_journal_replay;
1543   MockEventPreprocessor mock_event_preprocessor;
1544   MockReplayStatusFormatter mock_replay_status_formatter;
1545   librbd::journal::Listener* local_journal_listener = nullptr;
1546   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1547   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1548   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1549                                    mock_replayer_listener, mock_local_journal,
1550                                    mock_remote_journaler,
1551                                    mock_local_journal_replay,
1552                                    &local_journal_listener,
1553                                    &remote_replay_handler,
1554                                    &remote_journaler_listener));
1555
1556   expect_flush(mock_local_journal_replay, -EINVAL);
1557
1558   C_SaferCond ctx;
1559   mock_replayer.flush(&ctx);
1560   ASSERT_EQ(-EINVAL, ctx.wait());
1561
1562   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1563                                         mock_local_journal,
1564                                         mock_remote_journaler,
1565                                         mock_local_journal_replay));
1566 }
1567
1568 TEST_F(TestMockImageReplayerJournalReplayer, FlushCommitPositionError) {
1569   librbd::MockTestJournal mock_local_journal;
1570   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1571                                                 mock_local_journal};
1572   ::journal::MockJournaler mock_remote_journaler;
1573   MockReplayerListener mock_replayer_listener;
1574   MockThreads mock_threads{m_threads};
1575   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1576                                       mock_remote_journaler,
1577                                       {});
1578   MockReplayer mock_replayer{
1579     &mock_threads, "local mirror uuid", &mock_state_builder,
1580     &mock_replayer_listener};
1581
1582   expect_work_queue_repeatedly(mock_threads);
1583
1584   InSequence seq;
1585
1586   MockReplay mock_local_journal_replay;
1587   MockEventPreprocessor mock_event_preprocessor;
1588   MockReplayStatusFormatter mock_replay_status_formatter;
1589   librbd::journal::Listener* local_journal_listener = nullptr;
1590   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1591   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1592   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1593                                    mock_replayer_listener, mock_local_journal,
1594                                    mock_remote_journaler,
1595                                    mock_local_journal_replay,
1596                                    &local_journal_listener,
1597                                    &remote_replay_handler,
1598                                    &remote_journaler_listener));
1599
1600   expect_flush(mock_local_journal_replay, 0);
1601   expect_flush_commit_position(mock_remote_journaler, -EINVAL);
1602
1603   C_SaferCond ctx;
1604   mock_replayer.flush(&ctx);
1605   ASSERT_EQ(-EINVAL, ctx.wait());
1606
1607   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1608                                         mock_local_journal,
1609                                         mock_remote_journaler,
1610                                         mock_local_journal_replay));
1611 }
1612
1613
1614 TEST_F(TestMockImageReplayerJournalReplayer, ReplayFlushShutDownError) {
1615   librbd::MockTestJournal mock_local_journal;
1616   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1617                                                 mock_local_journal};
1618   ::journal::MockJournaler mock_remote_journaler;
1619   MockReplayerListener mock_replayer_listener;
1620   MockThreads mock_threads{m_threads};
1621   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1622                                       mock_remote_journaler,
1623                                       {});
1624   MockReplayer mock_replayer{
1625     &mock_threads, "local mirror uuid", &mock_state_builder,
1626     &mock_replayer_listener};
1627
1628   ::journal::MockReplayEntry mock_replay_entry;
1629   expect_get_commit_tid_in_debug(mock_replay_entry);
1630   expect_work_queue_repeatedly(mock_threads);
1631
1632   InSequence seq;
1633
1634   MockReplay mock_local_journal_replay;
1635   MockEventPreprocessor mock_event_preprocessor;
1636   MockReplayStatusFormatter mock_replay_status_formatter;
1637   librbd::journal::Listener* local_journal_listener = nullptr;
1638   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1639   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1640   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1641                                    mock_replayer_listener, mock_local_journal,
1642                                    mock_remote_journaler,
1643                                    mock_local_journal_replay,
1644                                    &local_journal_listener,
1645                                    &remote_replay_handler,
1646                                    &remote_journaler_listener));
1647
1648   expect_try_pop_front(mock_remote_journaler, 1, true);
1649   expect_shut_down(mock_local_journal_replay, false, -EINVAL);
1650   EXPECT_CALL(mock_local_journal, stop_external_replay());
1651   expect_notification(mock_threads, mock_replayer_listener);
1652   remote_replay_handler->handle_entries_available();
1653
1654   wait_for_notification();
1655   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
1656
1657   EXPECT_CALL(mock_local_journal, remove_listener(_));
1658   MockCloseImageRequest mock_close_image_request;
1659   expect_send(mock_close_image_request, 0);
1660   expect_stop_replay(mock_remote_journaler, 0);
1661   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
1662
1663   C_SaferCond shutdown_ctx;
1664   mock_replayer.shut_down(&shutdown_ctx);
1665   ASSERT_EQ(0, shutdown_ctx.wait());
1666 }
1667
1668 TEST_F(TestMockImageReplayerJournalReplayer, ReplayFlushStartError) {
1669   librbd::MockTestJournal mock_local_journal;
1670   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1671                                                 mock_local_journal};
1672   ::journal::MockJournaler mock_remote_journaler;
1673   MockReplayerListener mock_replayer_listener;
1674   MockThreads mock_threads{m_threads};
1675   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1676                                       mock_remote_journaler,
1677                                       {});
1678   MockReplayer mock_replayer{
1679     &mock_threads, "local mirror uuid", &mock_state_builder,
1680     &mock_replayer_listener};
1681
1682   ::journal::MockReplayEntry mock_replay_entry;
1683   expect_get_commit_tid_in_debug(mock_replay_entry);
1684   expect_work_queue_repeatedly(mock_threads);
1685
1686   InSequence seq;
1687
1688   MockReplay mock_local_journal_replay;
1689   MockEventPreprocessor mock_event_preprocessor;
1690   MockReplayStatusFormatter mock_replay_status_formatter;
1691   librbd::journal::Listener* local_journal_listener = nullptr;
1692   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1693   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1694   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1695                                    mock_replayer_listener, mock_local_journal,
1696                                    mock_remote_journaler,
1697                                    mock_local_journal_replay,
1698                                    &local_journal_listener,
1699                                    &remote_replay_handler,
1700                                    &remote_journaler_listener));
1701
1702   expect_try_pop_front(mock_remote_journaler, 1, true);
1703   expect_shut_down(mock_local_journal_replay, false, 0);
1704   EXPECT_CALL(mock_local_journal, stop_external_replay());
1705   expect_start_external_replay(mock_local_journal, nullptr, -EINVAL);
1706   expect_notification(mock_threads, mock_replayer_listener);
1707   remote_replay_handler->handle_entries_available();
1708
1709   wait_for_notification();
1710   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
1711
1712   EXPECT_CALL(mock_local_journal, remove_listener(_));
1713   MockCloseImageRequest mock_close_image_request;
1714   expect_send(mock_close_image_request, 0);
1715   expect_stop_replay(mock_remote_journaler, 0);
1716   EXPECT_CALL(mock_remote_journaler, remove_listener(_));
1717
1718   C_SaferCond shutdown_ctx;
1719   mock_replayer.shut_down(&shutdown_ctx);
1720   ASSERT_EQ(0, shutdown_ctx.wait());
1721 }
1722
1723 TEST_F(TestMockImageReplayerJournalReplayer, GetTagError) {
1724   librbd::MockTestJournal mock_local_journal;
1725   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1726                                                 mock_local_journal};
1727   ::journal::MockJournaler mock_remote_journaler;
1728   MockReplayerListener mock_replayer_listener;
1729   MockThreads mock_threads{m_threads};
1730   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1731                                       mock_remote_journaler,
1732                                       {});
1733   MockReplayer mock_replayer{
1734     &mock_threads, "local mirror uuid", &mock_state_builder,
1735     &mock_replayer_listener};
1736
1737   ::journal::MockReplayEntry mock_replay_entry;
1738   expect_get_commit_tid_in_debug(mock_replay_entry);
1739   expect_work_queue_repeatedly(mock_threads);
1740
1741   InSequence seq;
1742
1743   MockReplay mock_local_journal_replay;
1744   MockEventPreprocessor mock_event_preprocessor;
1745   MockReplayStatusFormatter mock_replay_status_formatter;
1746   librbd::journal::Listener* local_journal_listener = nullptr;
1747   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1748   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1749   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1750                                    mock_replayer_listener, mock_local_journal,
1751                                    mock_remote_journaler,
1752                                    mock_local_journal_replay,
1753                                    &local_journal_listener,
1754                                    &remote_replay_handler,
1755                                    &remote_journaler_listener));
1756
1757   cls::journal::Tag tag =
1758     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
1759                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1760                             true, 0, 0})};
1761   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1762   expect_shut_down(mock_local_journal_replay, false, 0);
1763   EXPECT_CALL(mock_local_journal, stop_external_replay());
1764   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1765                                0);
1766   expect_get_tag(mock_remote_journaler, tag, -EINVAL);
1767   expect_notification(mock_threads, mock_replayer_listener);
1768   remote_replay_handler->handle_entries_available();
1769
1770   wait_for_notification();
1771   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
1772
1773   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1774                                         mock_local_journal,
1775                                         mock_remote_journaler,
1776                                         mock_local_journal_replay));
1777 }
1778
1779 TEST_F(TestMockImageReplayerJournalReplayer, AllocateTagDemotion) {
1780   librbd::MockTestJournal mock_local_journal;
1781   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1782                                                 mock_local_journal};
1783   ::journal::MockJournaler mock_remote_journaler;
1784   MockReplayerListener mock_replayer_listener;
1785   MockThreads mock_threads{m_threads};
1786   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1787                                       mock_remote_journaler,
1788                                       {});
1789   MockReplayer mock_replayer{
1790     &mock_threads, "local mirror uuid", &mock_state_builder,
1791     &mock_replayer_listener};
1792
1793   ::journal::MockReplayEntry mock_replay_entry;
1794   expect_work_queue_repeatedly(mock_threads);
1795   expect_notification(mock_threads, mock_replayer_listener);
1796   expect_get_commit_tid_in_debug(mock_replay_entry);
1797   expect_get_tag_tid_in_debug(mock_local_journal);
1798   expect_committed(mock_replay_entry, mock_remote_journaler, 1);
1799
1800   InSequence seq;
1801
1802   MockReplay mock_local_journal_replay;
1803   MockEventPreprocessor mock_event_preprocessor;
1804   MockReplayStatusFormatter mock_replay_status_formatter;
1805   librbd::journal::Listener* local_journal_listener = nullptr;
1806   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1807   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1808   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1809                                    mock_replayer_listener, mock_local_journal,
1810                                    mock_remote_journaler,
1811                                    mock_local_journal_replay,
1812                                    &local_journal_listener,
1813                                    &remote_replay_handler,
1814                                    &remote_journaler_listener));
1815
1816   cls::journal::Tag tag =
1817     {1, 0, encode_tag_data({librbd::Journal<>::ORPHAN_MIRROR_UUID,
1818                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1819                             true, 0, 0})};
1820
1821   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1822   expect_shut_down(mock_local_journal_replay, false, 0);
1823   EXPECT_CALL(mock_local_journal, stop_external_replay());
1824   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1825                                0);
1826   expect_get_tag(mock_remote_journaler, tag, 0);
1827   expect_get_tag_data(mock_local_journal, {});
1828   expect_allocate_tag(mock_local_journal, 0);
1829   EXPECT_CALL(mock_local_journal_replay, decode(_, _)).WillOnce(Return(0));
1830   expect_preprocess(mock_event_preprocessor, false, 0);
1831   expect_process(mock_local_journal_replay, 0, 0);
1832
1833   remote_replay_handler->handle_entries_available();
1834   wait_for_notification();
1835   ASSERT_FALSE(mock_replayer.is_replaying());
1836
1837   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1838                                         mock_local_journal,
1839                                         mock_remote_journaler,
1840                                         mock_local_journal_replay));
1841 }
1842
1843 TEST_F(TestMockImageReplayerJournalReplayer, AllocateTagError) {
1844   librbd::MockTestJournal mock_local_journal;
1845   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1846                                                 mock_local_journal};
1847   ::journal::MockJournaler mock_remote_journaler;
1848   MockReplayerListener mock_replayer_listener;
1849   MockThreads mock_threads{m_threads};
1850   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1851                                       mock_remote_journaler,
1852                                       {});
1853   MockReplayer mock_replayer{
1854     &mock_threads, "local mirror uuid", &mock_state_builder,
1855     &mock_replayer_listener};
1856
1857   ::journal::MockReplayEntry mock_replay_entry;
1858   expect_work_queue_repeatedly(mock_threads);
1859   expect_get_commit_tid_in_debug(mock_replay_entry);
1860   expect_get_tag_tid_in_debug(mock_local_journal);
1861
1862   InSequence seq;
1863
1864   MockReplay mock_local_journal_replay;
1865   MockEventPreprocessor mock_event_preprocessor;
1866   MockReplayStatusFormatter mock_replay_status_formatter;
1867   librbd::journal::Listener* local_journal_listener = nullptr;
1868   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1869   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1870   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1871                                    mock_replayer_listener, mock_local_journal,
1872                                    mock_remote_journaler,
1873                                    mock_local_journal_replay,
1874                                    &local_journal_listener,
1875                                    &remote_replay_handler,
1876                                    &remote_journaler_listener));
1877
1878   cls::journal::Tag tag =
1879     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
1880                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1881                             true, 0, 0})};
1882
1883   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1884   expect_shut_down(mock_local_journal_replay, false, 0);
1885   EXPECT_CALL(mock_local_journal, stop_external_replay());
1886   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1887                                0);
1888   expect_get_tag(mock_remote_journaler, tag, 0);
1889   expect_allocate_tag(mock_local_journal, -EINVAL);
1890   expect_notification(mock_threads, mock_replayer_listener);
1891   remote_replay_handler->handle_entries_available();
1892
1893   wait_for_notification();
1894   ASSERT_FALSE(mock_replayer.is_replaying());
1895   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
1896
1897   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1898                                         mock_local_journal,
1899                                         mock_remote_journaler,
1900                                         mock_local_journal_replay));
1901 }
1902
1903 TEST_F(TestMockImageReplayerJournalReplayer, PreprocessError) {
1904   librbd::MockTestJournal mock_local_journal;
1905   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1906                                                 mock_local_journal};
1907   ::journal::MockJournaler mock_remote_journaler;
1908   MockReplayerListener mock_replayer_listener;
1909   MockThreads mock_threads{m_threads};
1910   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1911                                       mock_remote_journaler,
1912                                       {});
1913   MockReplayer mock_replayer{
1914     &mock_threads, "local mirror uuid", &mock_state_builder,
1915     &mock_replayer_listener};
1916
1917   ::journal::MockReplayEntry mock_replay_entry;
1918   expect_work_queue_repeatedly(mock_threads);
1919   expect_get_commit_tid_in_debug(mock_replay_entry);
1920   expect_get_tag_tid_in_debug(mock_local_journal);
1921
1922   InSequence seq;
1923
1924   MockReplay mock_local_journal_replay;
1925   MockEventPreprocessor mock_event_preprocessor;
1926   MockReplayStatusFormatter mock_replay_status_formatter;
1927   librbd::journal::Listener* local_journal_listener = nullptr;
1928   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1929   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1930   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1931                                    mock_replayer_listener, mock_local_journal,
1932                                    mock_remote_journaler,
1933                                    mock_local_journal_replay,
1934                                    &local_journal_listener,
1935                                    &remote_replay_handler,
1936                                    &remote_journaler_listener));
1937
1938   cls::journal::Tag tag =
1939     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
1940                             librbd::Journal<>::LOCAL_MIRROR_UUID,
1941                             true, 0, 0})};
1942
1943   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
1944   expect_shut_down(mock_local_journal_replay, false, 0);
1945   EXPECT_CALL(mock_local_journal, stop_external_replay());
1946   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
1947                                0);
1948   expect_get_tag(mock_remote_journaler, tag, 0);
1949   expect_allocate_tag(mock_local_journal, 0);
1950   EXPECT_CALL(mock_replay_entry, get_data());
1951   EXPECT_CALL(mock_local_journal_replay, decode(_, _)).WillOnce(Return(0));
1952   expect_preprocess(mock_event_preprocessor, true, -EINVAL);
1953
1954   expect_notification(mock_threads, mock_replayer_listener);
1955   remote_replay_handler->handle_entries_available();
1956
1957   wait_for_notification();
1958   ASSERT_FALSE(mock_replayer.is_replaying());
1959   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
1960
1961   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
1962                                         mock_local_journal,
1963                                         mock_remote_journaler,
1964                                         mock_local_journal_replay));
1965 }
1966
1967 TEST_F(TestMockImageReplayerJournalReplayer, ProcessError) {
1968   librbd::MockTestJournal mock_local_journal;
1969   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
1970                                                 mock_local_journal};
1971   ::journal::MockJournaler mock_remote_journaler;
1972   MockReplayerListener mock_replayer_listener;
1973   MockThreads mock_threads{m_threads};
1974   MockStateBuilder mock_state_builder(mock_local_image_ctx,
1975                                       mock_remote_journaler,
1976                                       {});
1977   MockReplayer mock_replayer{
1978     &mock_threads, "local mirror uuid", &mock_state_builder,
1979     &mock_replayer_listener};
1980
1981   ::journal::MockReplayEntry mock_replay_entry;
1982   expect_work_queue_repeatedly(mock_threads);
1983   expect_get_commit_tid_in_debug(mock_replay_entry);
1984   expect_get_tag_tid_in_debug(mock_local_journal);
1985   expect_notification(mock_threads, mock_replayer_listener);
1986
1987   InSequence seq;
1988
1989   MockReplay mock_local_journal_replay;
1990   MockEventPreprocessor mock_event_preprocessor;
1991   MockReplayStatusFormatter mock_replay_status_formatter;
1992   librbd::journal::Listener* local_journal_listener = nullptr;
1993   ::journal::ReplayHandler* remote_replay_handler = nullptr;
1994   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
1995   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
1996                                    mock_replayer_listener, mock_local_journal,
1997                                    mock_remote_journaler,
1998                                    mock_local_journal_replay,
1999                                    &local_journal_listener,
2000                                    &remote_replay_handler,
2001                                    &remote_journaler_listener));
2002
2003   cls::journal::Tag tag =
2004     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
2005                             librbd::Journal<>::LOCAL_MIRROR_UUID,
2006                             true, 0, 0})};
2007
2008   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
2009   expect_shut_down(mock_local_journal_replay, false, 0);
2010   EXPECT_CALL(mock_local_journal, stop_external_replay());
2011   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
2012                                0);
2013   expect_get_tag(mock_remote_journaler, tag, 0);
2014   expect_allocate_tag(mock_local_journal, 0);
2015   EXPECT_CALL(mock_replay_entry, get_data());
2016   EXPECT_CALL(mock_local_journal_replay, decode(_, _)).WillOnce(Return(0));
2017   expect_preprocess(mock_event_preprocessor, false, 0);
2018   expect_process(mock_local_journal_replay, 0, -EINVAL);
2019
2020   // attempt to process the next event
2021   C_SaferCond replay_ctx;
2022   expect_try_pop_front_return_no_entries(mock_remote_journaler, &replay_ctx);
2023   remote_replay_handler->handle_entries_available();
2024
2025   wait_for_notification();
2026   ASSERT_FALSE(mock_replayer.is_replaying());
2027   ASSERT_EQ(-EINVAL, mock_replayer.get_error_code());
2028
2029   ASSERT_EQ(0, replay_ctx.wait());
2030
2031   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
2032                                         mock_local_journal,
2033                                         mock_remote_journaler,
2034                                         mock_local_journal_replay));
2035 }
2036
2037 TEST_F(TestMockImageReplayerJournalReplayer, ImageNameUpdated) {
2038   librbd::MockTestJournal mock_local_journal;
2039   librbd::MockTestImageCtx mock_local_image_ctx{*m_local_image_ctx,
2040                                                 mock_local_journal};
2041   ::journal::MockJournaler mock_remote_journaler;
2042   MockReplayerListener mock_replayer_listener;
2043   MockThreads mock_threads{m_threads};
2044   MockStateBuilder mock_state_builder(mock_local_image_ctx,
2045                                       mock_remote_journaler,
2046                                       {});
2047   MockReplayer mock_replayer{
2048     &mock_threads, "local mirror uuid", &mock_state_builder,
2049     &mock_replayer_listener};
2050
2051   ::journal::MockReplayEntry mock_replay_entry;
2052   expect_work_queue_repeatedly(mock_threads);
2053   expect_add_event_after_repeatedly(mock_threads);
2054   expect_get_commit_tid_in_debug(mock_replay_entry);
2055   expect_get_tag_tid_in_debug(mock_local_journal);
2056   expect_committed(mock_replay_entry, mock_remote_journaler, 1);
2057   expect_notification(mock_threads, mock_replayer_listener);
2058
2059   InSequence seq;
2060
2061   MockReplay mock_local_journal_replay;
2062   MockEventPreprocessor mock_event_preprocessor;
2063   MockReplayStatusFormatter mock_replay_status_formatter;
2064   librbd::journal::Listener* local_journal_listener = nullptr;
2065   ::journal::ReplayHandler* remote_replay_handler = nullptr;
2066   ::journal::JournalMetadataListener* remote_journaler_listener = nullptr;
2067   ASSERT_EQ(0, init_entry_replayer(mock_replayer, mock_threads,
2068                                    mock_replayer_listener, mock_local_journal,
2069                                    mock_remote_journaler,
2070                                    mock_local_journal_replay,
2071                                    &local_journal_listener,
2072                                    &remote_replay_handler,
2073                                    &remote_journaler_listener));
2074
2075   mock_local_image_ctx.name = "NEW NAME";
2076   cls::journal::Tag tag =
2077     {1, 0, encode_tag_data({librbd::Journal<>::LOCAL_MIRROR_UUID,
2078                             librbd::Journal<>::LOCAL_MIRROR_UUID,
2079                             true, 0, 0})};
2080
2081   expect_try_pop_front(mock_remote_journaler, tag.tid, true);
2082   expect_shut_down(mock_local_journal_replay, false, 0);
2083   EXPECT_CALL(mock_local_journal, stop_external_replay());
2084   expect_start_external_replay(mock_local_journal, &mock_local_journal_replay,
2085                                0);
2086   expect_get_tag(mock_remote_journaler, tag, 0);
2087   expect_allocate_tag(mock_local_journal, 0);
2088   EXPECT_CALL(mock_local_journal_replay, decode(_, _)).WillOnce(Return(0));
2089   expect_preprocess(mock_event_preprocessor, false, 0);
2090   expect_process(mock_local_journal_replay, 0, 0);
2091
2092   // attempt to process the next event
2093   C_SaferCond replay_ctx;
2094   expect_try_pop_front_return_no_entries(mock_remote_journaler, &replay_ctx);
2095
2096   remote_replay_handler->handle_entries_available();
2097   wait_for_notification();
2098
2099   auto image_spec = util::compute_image_spec(m_local_io_ctx, "NEW NAME");
2100   ASSERT_EQ(image_spec, mock_replayer.get_image_spec());
2101
2102   ASSERT_EQ(0, replay_ctx.wait());
2103   ASSERT_TRUE(mock_replayer.is_replaying());
2104
2105   ASSERT_EQ(0, shut_down_entry_replayer(mock_replayer, mock_threads,
2106                                         mock_local_journal,
2107                                         mock_remote_journaler,
2108                                         mock_local_journal_replay));
2109 }
2110
2111 } // namespace journal
2112 } // namespace image_replayer
2113 } // namespace mirror
2114 } // namespace rbd