fix(duplication): prevent plog files from being removed by GC while they are being checked by duplication

src/replica/duplication/load_from_private_log.cpp

@@ -122,6 +122,8 @@ void load_from_private_log::run()
 
 void load_from_private_log::find_log_file_to_start()
 {
+    _duplicator->set_duplication_plog_checking(true);
+
     // `file_map` has already excluded the useless log files during replica init.
     auto file_map = _private_log->get_log_file_map();
 
@@ -166,6 +168,8 @@ void load_from_private_log::find_log_file_to_start(std::map<int, log_file_ptr> l
         }
     }
     start_from_log_file(_current);
+
+    _duplicator->set_duplication_plog_checking(false);
 }
 
 void load_from_private_log::replay_log_block()

src/replica/duplication/replica_duplicator.cpp

@@ -257,5 +257,11 @@ uint64_t replica_duplicator::get_pending_mutations_count() const
     return cnt > 0 ? static_cast<uint64_t>(cnt) : 0;
 }
 
+void replica_duplicator::set_duplication_plog_checking(bool checking)
+{
+    _replica->set_duplication_plog_checking(checking);
+}
+
+
 } // namespace replication
 } // namespace dsn

src/replica/duplication/replica_duplicator.h

@@ -138,7 +138,9 @@ class replica_duplicator : public replica_base, public pipeline::base
     // For metric "dup.pending_mutations_count"
     uint64_t get_pending_mutations_count() const;
 
-    duplication_status::type status() const { return _status; };
+    duplication_status::type status() const { return _status; }
+
+    void set_duplication_plog_checking(bool checking);
 
 private:
     friend class duplication_test_base;

src/replica/replica.h

@@ -254,6 +254,12 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     replica_duplicator_manager *get_duplication_manager() const { return _duplication_mgr.get(); }
     bool is_duplication_master() const { return _is_duplication_master; }
     bool is_duplication_follower() const { return _is_duplication_follower; }
+    bool is_duplication_plog_checking() const { return _is_duplication_plog_checking.load(); }
+    void set_duplication_plog_checking(bool checking)
+    {
+        _is_duplication_plog_checking.store(checking);
+    }
+
 
     //
     // Backup
@@ -633,6 +639,7 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     bool _is_manual_emergency_checkpointing{false};
     bool _is_duplication_master{false};
     bool _is_duplication_follower{false};
+    std::atomic<bool> _is_duplication_plog_checking{false};
 
     // backup
     std::unique_ptr<replica_backup_manager> _backup_mgr;