feat(dup): verify private log validity before starting to duplicate

src/dist/replication/lib/duplication/replica_duplicator.cpp

@@ -141,5 +141,24 @@ error_s replica_duplicator::update_progress(const duplication_progress &p)
     return error_s::ok();
 }
 
+void replica_duplicator::verify_start_decree(decree start_decree)
+{
+    decree confirmed_decree = progress().confirmed_decree;
+    decree last_decree = progress().last_decree;
+    decree max_gced_decree = get_max_gced_decree();
+    dassert(max_gced_decree < start_decree,
+            "the logs haven't yet duplicated were accidentally truncated "
+            "[max_gced_decree: {}, start_decree: {}, confirmed_decree: {}, last_decree: {}]",
+            max_gced_decree,
+            start_decree,
+            confirmed_decree,
+            last_decree);
+}
+
+decree replica_duplicator::get_max_gced_decree() const
+{
+    return _replica->private_log()->max_gced_decree(_replica->get_gpid());
+}
+
 } // namespace replication
 } // namespace dsn

src/dist/replication/lib/duplication/replica_duplicator.h

@@ -91,6 +91,14 @@ class replica_duplicator : public replica_base, public pipeline::base
 
     std::string to_string() const;
 
+    // To ensure mutation logs after start_decree is available
+    // for duplication. If not, it means the eventual consistency
+    // of duplication is no longer guaranteed due to the missing logs.
+    // For current implementation the system will fail fast.
+    void verify_start_decree(decree start_decree);
+
+    decree get_max_gced_decree() const;
+
 private:
     friend class replica_duplicator_test;
     friend class duplication_sync_timer_test;

src/dist/replication/lib/mutation_log.cpp

@@ -1001,6 +1001,30 @@ decree mutation_log::max_commit_on_disk() const
     return _private_max_commit_on_disk;
 }
 
+decree mutation_log::max_gced_decree(gpid gpid) const
+{
+    zauto_lock l(_lock);
+    return max_gced_decree_no_lock(gpid);
+}
+
+decree mutation_log::max_gced_decree_no_lock(gpid gpid) const
+{
+    dassert(_is_private, "");
+
+    decree result = invalid_decree;
+    for (auto &log : _log_files) {
+        auto it = log.second->previous_log_max_decrees().find(gpid);
+        if (it != log.second->previous_log_max_decrees().end()) {
+            if (result == invalid_decree) {
+                result = it->second.max_decree;
+            } else {
+                result = std::min(result, it->second.max_decree);
+            }
+        }
+    }
+    return result;
+}
+
 void mutation_log::check_valid_start_offset(gpid gpid, int64_t valid_start_offset) const
 {
     zauto_lock l(_lock);

src/dist/replication/lib/mutation_log.h

@@ -324,6 +324,12 @@ class mutation_log : public ref_counter
     // thread safe
     decree max_commit_on_disk() const;
 
+    // Returns `invalid_decree` when plog directory is empty, maybe data corruption occurred
+    // and the entire directory was tagged ".err".
+    // thread-safe & private log only
+    decree max_gced_decree(gpid gpid) const;
+    decree max_gced_decree_no_lock(gpid gpid) const;
+
     // thread-safe
     std::map<int, log_file_ptr> get_log_file_map() const;