rollout epoch based divergence

seckcoder · seckcoder · commit a91cd03cf228 · 2024-05-21T21:43:12.000Z
diff --git a/cloud/replication_test.cc b/cloud/replication_test.cc
@@ -217,8 +217,7 @@ class ReplicationTest : public testing::Test {
   // it will catch up until end of log
   //
   // Returns the number of log records applied
-  size_t catchUpFollower(std::optional<size_t> num_records = std::nullopt,
-                         bool allow_new_manifest_writes = true);
+  size_t catchUpFollower(std::optional<size_t> num_records = std::nullopt);
 
   WriteOptions wo() const {
     WriteOptions w;
@@ -429,11 +428,7 @@ DB* ReplicationTest::openLeader(Options options, uint64_t snapshot_replication_e
       s = db->ApplyReplicationLogRecord(
           log_records_[leaderSeq].first, log_records_[leaderSeq].second,
           [this](Slice) { return ColumnFamilyOptions(leaderOptions()); },
-          true /* allow_new_manifest_writes */,
-          snapshot_replication_epoch_,
-          &info,
-          DB::AR_EVICT_OBSOLETE_FILES |
-              DB::AR_EPOCH_BASED_DIVERGENCE_DETECTION);
+          snapshot_replication_epoch_, &info, DB::AR_EVICT_OBSOLETE_FILES);
       assert(s.ok());
       assert(!info.diverged_manifest_writes);
     }
@@ -480,8 +475,7 @@ DB* ReplicationTest::openFollower(Options options) {
   return db;
 }
 
-size_t ReplicationTest::catchUpFollower(std::optional<size_t> num_records,
-                                        bool allow_new_manifest_writes) {
+size_t ReplicationTest::catchUpFollower(std::optional<size_t> num_records) {
   MutexLock lock(&log_records_mutex_);
   DB::ApplyReplicationLogRecordInfo info;
   size_t ret = 0;
@@ -496,15 +490,11 @@ size_t ReplicationTest::catchUpFollower(std::optional<size_t> num_records,
         [this](Slice) {
           return ColumnFamilyOptions(follower_db_->GetOptions());
         },
-        allow_new_manifest_writes,
         snapshot_replication_epoch_,
         &info, flags);
     assert(s.ok());
     ++ret;
   }
-  if (info.has_new_manifest_writes) {
-    assert(info.has_manifest_writes);
-  }
   for (auto& cf : info.added_column_families) {
     auto inserted =
         follower_cfs_.try_emplace(cf->GetName(), std::move(cf)).second;
@@ -1112,20 +1102,6 @@ TEST_F(ReplicationTest, LogNumberDontGoBackwards) {
   EXPECT_GE(logNum, minLogNumberToKeep);
 }
 
-TEST_F(ReplicationTest, AllowNewManifestWrite) {
-  auto leader = openLeader(), follower = openFollower();
-
-  ASSERT_OK(leader->Put(wo(), "k1", "v1"));
-  ASSERT_OK(leader->Flush({}));
-  catchUpFollower(2);
-  // The new manifest write won't be applied
-  catchUpFollower(1, false);
-  uint64_t followerMUS, leaderMUS;
-  ASSERT_OK(follower->GetManifestUpdateSequence(&followerMUS));
-  ASSERT_OK(leader->GetManifestUpdateSequence(&leaderMUS));
-  EXPECT_LT(followerMUS, leaderMUS);
-}
-
 // Memtable switch record won't be generated if all memtables are empty
 TEST_F(ReplicationTest, NoMemSwitchRecordIfEmpty) {
   auto leader = openLeader();
diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc
@@ -1321,15 +1321,12 @@ std::string DescribeVersionEdit(const VersionEdit& e, ColumnFamilyData* cfd) {
 Status DBImpl::ApplyReplicationLogRecord(ReplicationLogRecord record,
                                          std::string replication_sequence,
                                          CFOptionsFactory cf_options_factory,
-                                         bool allow_new_manifest_writes,
                                          uint64_t snapshot_replication_epoch,
                                          ApplyReplicationLogRecordInfo* info,
                                          unsigned flags) {
   JobContext job_context(0, false);
   Status s;
   bool evictObsoleteFiles = flags & AR_EVICT_OBSOLETE_FILES;
-  bool enableEpochBasedDivergenceDetection =
-      flags & AR_EPOCH_BASED_DIVERGENCE_DETECTION;
 
   {
     WriteThread::Writer w;
@@ -1469,8 +1466,7 @@ Status DBImpl::ApplyReplicationLogRecord(ReplicationLogRecord record,
           // at epoch `e` is generated while there is no manifest writes for the
           // latest epoch yet.
           if (latest_applied_update_sequence <= current_update_sequence) {
-            if (enableEpochBasedDivergenceDetection &&
-                !versions_->IsReplicationEpochsEmpty()) {
+            if (!versions_->IsReplicationEpochsEmpty()) {
               auto inferred_epoch_of_mus = versions_->GetReplicationEpochForMUS(
                   latest_applied_update_sequence);
               // If mus is smaller than mus in the epoch set, the replication
@@ -1506,9 +1502,9 @@ Status DBImpl::ApplyReplicationLogRecord(ReplicationLogRecord record,
             // don't apply the manifest write if it's already applied
             continue;
           } else {
-            if (enableEpochBasedDivergenceDetection &&
-                !versions_->IsReplicationEpochsEmpty() &&
-                versions_->replication_epochs_.GetLargestEpoch() < snapshot_replication_epoch) {
+            if (!versions_->IsReplicationEpochsEmpty() &&
+                versions_->replication_epochs_.GetLargestEpoch() <
+                    snapshot_replication_epoch) {
               // This should be the first mus of `snapshotEpoch`
               if (replication_epoch != snapshot_replication_epoch) {
                 info->diverged_manifest_writes = true;
@@ -1523,11 +1519,6 @@ Status DBImpl::ApplyReplicationLogRecord(ReplicationLogRecord record,
             }
           }
 
-          info->has_new_manifest_writes = true;
-          if (!allow_new_manifest_writes) {
-            // We don't expect new manifest writes, break early
-            break;
-          }
           ++current_update_sequence;
           if (e.GetManifestUpdateSequence() != current_update_sequence) {
             std::ostringstream oss;
diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h
@@ -427,7 +427,6 @@ class DBImpl : public DB {
   Status ApplyReplicationLogRecord(ReplicationLogRecord record,
                                    std::string replication_sequence,
                                    CFOptionsFactory cf_options_factory,
-                                   bool allow_new_manifest_writes,
                                    uint64_t snapshot_replication_epoch,
                                    ApplyReplicationLogRecordInfo* info,
                                    unsigned flags) override;
diff --git a/db/db_test.cc b/db/db_test.cc
@@ -3444,7 +3444,6 @@ class ModelDB : public DB {
   Status ApplyReplicationLogRecord(ReplicationLogRecord /*record*/,
                                    std::string /*replication_sequence*/,
                                    CFOptionsFactory /* cf_options_factory */,
-                                   bool /* allow_new_manifest_writes */,
                                    uint64_t /* snapshot_replication_epoch */,
                                    ApplyReplicationLogRecordInfo* /*info*/,
                                    unsigned /*flags*/) override {
diff --git a/include/rocksdb/db.h b/include/rocksdb/db.h
@@ -1481,13 +1481,6 @@ class DB {
     // if its manifest update sequence number is lower and equal than the DB's
     // manifest update sequence number.
     uint64_t latest_applied_manifest_update_seq{0};
-    // If true, the replication event contains manifest writes with manifest
-    // update sequence number greater than the DB's manifest update sequence
-    // number. The new manifest writes may or may not be applied depending on
-    // value of `allow_new_manifest_writes`
-    // TODO(wei): remove once epoch based diverged manifest write detection is
-    // rolled out
-    bool has_new_manifest_writes{false};
     bool diverged_manifest_writes{false};
 
     // added_column_families contains column family handles for all column
@@ -1508,21 +1501,14 @@ class DB {
   // options need to be returned. The function is invoked done outside of the DB
   // mutex.
   //
-  // If `allow_new_manifest_writes` is false, and there are new
-  // manifest writes after DB's manfiest update sequence, this function will set
-  // `has_new_manifest_writes_` to true and return early without applying
-  // the new manifest writes.
-  //
   // REQUIRES: info needs to be provided, can't be nullptr.
   enum ApplyReplicationLogRecordFlags : unsigned {
     AR_EVICT_OBSOLETE_FILES = 1U << 0,
-    AR_EPOCH_BASED_DIVERGENCE_DETECTION = 1U << 1,
   };
   using CFOptionsFactory = std::function<ColumnFamilyOptions(Slice)>;
   virtual Status ApplyReplicationLogRecord(ReplicationLogRecord record,
                                            std::string replication_sequence,
                                            CFOptionsFactory cf_options_factory,
-                                           bool allow_new_manifest_writes,
                                            uint64_t snapshot_replication_epoch,
                                            ApplyReplicationLogRecordInfo* info,
                                            unsigned flags = 0) = 0;
diff --git a/include/rocksdb/utilities/stackable_db.h b/include/rocksdb/utilities/stackable_db.h
@@ -529,13 +529,12 @@ class StackableDB : public DB {
   Status ApplyReplicationLogRecord(ReplicationLogRecord record,
                                    std::string replication_sequence,
                                    CFOptionsFactory cf_options_factory,
-                                   bool allow_new_manifest_writes,
                                    uint64_t snapshot_replication_epoch,
                                    ApplyReplicationLogRecordInfo* info,
                                    unsigned flags) override {
     return db_->ApplyReplicationLogRecord(
         record, replication_sequence, std::move(cf_options_factory),
-        allow_new_manifest_writes, snapshot_replication_epoch, info, flags);
+        snapshot_replication_epoch, info, flags);
   }
   Status GetReplicationRecordDebugString(const ReplicationLogRecord& record,
                                          std::string* out) const override {
