Verify values in secondary database against expected state (#13281)

Summary:
TLDR: This PR enables secondary DB verification inside the "simple" crash tests (`NonBatchedOpsStressTest`). Essentially, we want to be able to verify that the secondary is a valid "prefix" of the primary. This PR allows us to do this by piggybacking on the existing verification of the primary through `Get()` requests.

I originally proposed replaying the trace file to recreate the `ExpectedState` as of a specific sequence number. This could be used to run verifications against the secondary database. I did some experimenting in #13266 and got a "mostly working" implementation of this approach. I could sometimes get through entire key space verifications but eventually one of the keys would fail verification. I have not figured out the root cause yet, but I assume that something caused the sequence number to trace record alignment to break.

The approach in this PR is considerably simpler. We can just check that the secondary database's value is in the correct "range," which we already have functionality for checking that. Compared to the approach in #13266, this approach is _much, much simpler_ since we do not have to go through the whole headache of replaying the trace and creating an entire new `ExpectedState`. (Look at #13266 to see how much of a mess that creates.) I think this approach is better than my original approach in almost most aspects: it's faster, uses less space, and has less room for implementation errors.

Other nice aspects of this approach:
1. We don't need to block the primary. (Another approach you could imagine would be to block writes to the primary, have the secondary catch up, do the whole verification, and then re-enable writes to the primary.)
2. We don't need to block the secondary or do any special coordination (locks, sync points, etc). (If we insist on one "golden" expected value to be read from the secondary, then we need to make sure that another thread does not call `TryCatchUpWithPrimary` while we are trying to perform a `Get()`)
3. More "realistic" usage of the secondary. For instance, writes to the primary and secondary would continue on in production while we try to read from the secondary.

The main drawback of course is that we verify against a range of expected values, rather than one particular expected value. However, I think this is acceptable and "good enough" especially with all of other the aforementioned benefits.

Historical context: There is some very old code that attempted to verify secondaries, but is not enabled. This code has not been touched or executed in an extremely long time, and the crash tests started failing when I tried enabling it, most likely because the code is not compatible with certain other crash test options. This code is for the "continuous verification" and involves long iterator scans over the secondary database. Some of the code involved the cross CF consistency test type. I don't think the old checks are what we really want for our purposes of verifying the secondary functionality. Since I don't think we will get much value out of this old "continuous verification" code, I integrated my secondary verification with the "regular" database verification. This also makes the rollout simpler on my end, since I can control whether my secondary verifications are enabled through one `test_secondary` configuration. To make sure the old code does not execute for our recurring crash test runs, I had to enforce that `continuous_verification_interval` is 0 whenever `test_secondary` is set.

Monitoring: I will want to monitor the Sandcastle "simple" runs for failures where `test_secondary` is set. All of my error messages are prefixed with "Secondary" so it should be easy to tell if this PR causes any crash test issues.

Future work:
1. Extend this to followers. I think the same verification method should work, so most of the code from this PR should be reusable
2. Add additional checks to make sure the sequence number of the follower/secondary is actually increasing. For instance, if the primary's sequence number has advanced, and in that period the secondary has not (even after calling `TryCatchUpWithPrimary`), then we know there is a problem
3. Potentially checking things other than `Get()` for the secondary (i.e. iterators). I think the focus here should be testing replication-specific logic, and since we will already have separate unit tests, we do not need to repeat all of tests against both the primary and the secondary.

Pull Request resolved: #13281

Test Plan:
The primary crash test commands I ran were:
```
python3 tools/db_crashtest.py --simple blackbox --test_secondary=1
python3 tools/db_crashtest.py --simple whitebox --test_secondary=1
```

As a sanity check, I added an `assert(false)` right after my secondary verification code to make sure that my code was actually being run.

Reviewed By: anand1976

Differential Revision: D67953821

Pulled By: archang19

fbshipit-source-id: 0bd853580ea53566be41639f5499eb9b5e0e9376

Author: archang19

db_stress_tool/cf_consistency_stress.cc

@@ -1040,17 +1040,17 @@ class CfConsistencyStressTest : public StressTest {
     assert(thread);
     Status status;
 
-    DB* db_ptr = cmp_db_ ? cmp_db_ : db_;
-    const auto& cfhs = cmp_db_ ? cmp_cfhs_ : column_families_;
+    DB* db_ptr = secondary_db_ ? secondary_db_ : db_;
+    const auto& cfhs = secondary_db_ ? secondary_cfhs_ : column_families_;
 
     // Take a snapshot to preserve the state of primary db.
     ManagedSnapshot snapshot_guard(db_);
 
     SharedState* shared = thread->shared;
     assert(shared);
 
-    if (cmp_db_) {
-      status = cmp_db_->TryCatchUpWithPrimary();
+    if (secondary_db_) {
+      status = secondary_db_->TryCatchUpWithPrimary();
       if (!status.ok()) {
         fprintf(stderr, "TryCatchUpWithPrimary: %s\n",
                 status.ToString().c_str());
@@ -1083,7 +1083,7 @@ class CfConsistencyStressTest : public StressTest {
     // `FLAGS_rate_limit_user_ops` to avoid slowing any validation.
     ReadOptions ropts(FLAGS_verify_checksum, true);
     ropts.total_order_seek = true;
-    if (nullptr == cmp_db_) {
+    if (nullptr == secondary_db_) {
       ropts.snapshot = snapshot_guard.snapshot();
     }
     uint32_t crc = 0;

db_stress_tool/db_stress_test_base.cc

@@ -69,7 +69,7 @@ StressTest::StressTest()
       new_column_family_name_(1),
       num_times_reopened_(0),
       db_preload_finished_(false),
-      cmp_db_(nullptr),
+      secondary_db_(nullptr),
       is_db_stopped_(false) {
   if (FLAGS_destroy_db_initially) {
     std::vector<std::string> files;
@@ -114,11 +114,11 @@ void StressTest::CleanUp() {
   delete db_;
   db_ = nullptr;
 
-  for (auto* cf : cmp_cfhs_) {
+  for (auto* cf : secondary_cfhs_) {
     delete cf;
   }
-  cmp_cfhs_.clear();
-  delete cmp_db_;
+  secondary_cfhs_.clear();
+  delete secondary_db_;
 }
 
 std::shared_ptr<Cache> StressTest::NewCache(size_t capacity,
@@ -3696,9 +3696,10 @@ void StressTest::Open(SharedState* shared, bool reopen) {
       tmp_opts.env = db_stress_env;
       const std::string& secondary_path = FLAGS_secondaries_base;
       s = DB::OpenAsSecondary(tmp_opts, FLAGS_db, secondary_path,
-                              cf_descriptors, &cmp_cfhs_, &cmp_db_);
+                              cf_descriptors, &secondary_cfhs_, &secondary_db_);
       assert(s.ok());
-      assert(cmp_cfhs_.size() == static_cast<size_t>(FLAGS_column_families));
+      assert(secondary_cfhs_.size() ==
+             static_cast<size_t>(FLAGS_column_families));
     }
   } else {
     DBWithTTL* db_with_ttl;

db_stress_tool/db_stress_test_base.h

@@ -412,9 +412,8 @@ class StressTest {
   std::atomic<bool> db_preload_finished_;
   std::shared_ptr<SstQueryFilterConfigsManager::Factory> sqfc_factory_;
 
-  // Fields used for continuous verification from another thread
-  DB* cmp_db_;
-  std::vector<ColumnFamilyHandle*> cmp_cfhs_;
+  DB* secondary_db_;
+  std::vector<ColumnFamilyHandle*> secondary_cfhs_;
   bool is_db_stopped_;
 };
 

db_stress_tool/no_batched_ops_stress.cc

@@ -153,6 +153,49 @@ class NonBatchedOpsStressTest : public StressTest {
                           from_db.data(), from_db.size());
           }
         }
+
+        if (secondary_db_) {
+          assert(secondary_cfhs_.size() == column_families_.size());
+          // We are going to read in the expected values before catching the
+          // secondary up to the primary. This sets the lower bound of the
+          // acceptable values that can be returned from the secondary. After
+          // each Get() to the secondary, we are going to read in the expected
+          // value again to determine the upper bound. As long as the returned
+          // value from Get() is within these bounds, we consider that okay. The
+          // lower bound will always be moving forwards anyways as
+          // TryCatchUpWithPrimary() gets called.
+          std::vector<ExpectedValue> pre_read_expected_values;
+          for (int64_t i = start; i < end; ++i) {
+            pre_read_expected_values.push_back(
+                shared->Get(static_cast<int>(cf), i));
+          }
+
+          Status s = secondary_db_->TryCatchUpWithPrimary();
+          if (!s.ok()) {
+            VerificationAbort(shared,
+                              "Secondary failed to catch up to the primary");
+          }
+
+          for (int64_t i = start; i < end; ++i) {
+            if (thread->shared->HasVerificationFailedYet()) {
+              break;
+            }
+
+            const std::string key = Key(i);
+            std::string from_db;
+
+            s = secondary_db_->Get(options, column_families_[cf], key,
+                                   &from_db);
+
+            assert(!pre_read_expected_values.empty() &&
+                   static_cast<size_t>(i - start) <
+                       pre_read_expected_values.size());
+            VerifyValueRange(static_cast<int>(cf), i, options, shared, from_db,
+                             /* msg_prefix */ "Secondary get verification", s,
+                             pre_read_expected_values[i - start]);
+          }
+        }
+
       } else if (method == VerificationMethod::kGetEntity) {
         for (int64_t i = start; i < end; ++i) {
           if (thread->shared->HasVerificationFailedYet()) {
@@ -333,12 +376,19 @@ class NonBatchedOpsStressTest : public StressTest {
   }
 
   void ContinuouslyVerifyDb(ThreadState* thread) const override {
-    if (!cmp_db_) {
+    // For automated crash tests, we only want to run this continous
+    // verification when continuous_verification_interval > 0 and there is
+    // a secondary db. This continous verification currently fails when there is
+    // a secondary db during the iterator scan. The stack trace mentions
+    // BlobReader/BlobSource but it may not necessarily be related to BlobDB.
+    // Regardless, we only want to run this function if we are experimenting and
+    // explicitly setting continuous_verification_interval.
+    if (!secondary_db_ || !FLAGS_continuous_verification_interval) {
       return;
     }
-    assert(cmp_db_);
-    assert(!cmp_cfhs_.empty());
-    Status s = cmp_db_->TryCatchUpWithPrimary();
+    assert(secondary_db_);
+    assert(!secondary_cfhs_.empty());
+    Status s = secondary_db_->TryCatchUpWithPrimary();
     if (!s.ok()) {
       assert(false);
       exit(1);
@@ -372,7 +422,7 @@ class NonBatchedOpsStressTest : public StressTest {
 
     {
       uint32_t crc = 0;
-      std::unique_ptr<Iterator> it(cmp_db_->NewIterator(read_opts));
+      std::unique_ptr<Iterator> it(secondary_db_->NewIterator(read_opts));
       s = checksum_column_family(it.get(), &crc);
       if (!s.ok()) {
         fprintf(stderr, "Computing checksum of default cf: %s\n",
@@ -381,19 +431,21 @@ class NonBatchedOpsStressTest : public StressTest {
       }
     }
 
-    for (auto* handle : cmp_cfhs_) {
+    for (auto* handle : secondary_cfhs_) {
       if (thread->rand.OneInOpt(3)) {
         // Use Get()
         uint64_t key = rand64.Uniform(static_cast<uint64_t>(max_key));
         std::string key_str = Key(key);
         std::string value;
         std::string key_ts;
-        s = cmp_db_->Get(read_opts, handle, key_str, &value,
-                         FLAGS_user_timestamp_size > 0 ? &key_ts : nullptr);
+        s = secondary_db_->Get(
+            read_opts, handle, key_str, &value,
+            FLAGS_user_timestamp_size > 0 ? &key_ts : nullptr);
         s.PermitUncheckedError();
       } else {
         // Use range scan
-        std::unique_ptr<Iterator> iter(cmp_db_->NewIterator(read_opts, handle));
+        std::unique_ptr<Iterator> iter(
+            secondary_db_->NewIterator(read_opts, handle));
         uint32_t rnd = (thread->rand.Next()) % 4;
         if (0 == rnd) {
           // SeekToFirst() + Next()*5
@@ -2810,6 +2862,84 @@ class NonBatchedOpsStressTest : public StressTest {
     return true;
   }
 
+  // Compared to VerifyOrSyncValue, VerifyValueRange takes in a
+  // pre_read_expected_value to determine the lower bound of acceptable values.
+  // Anything from the pre_read_expected_value to the post_read_expected_value
+  // is considered acceptable. VerifyValueRange does not perform the initial
+  // "sync" step and does not compare the exact data/lengths for the values.
+  // This verification is suitable for verifying secondary or follower databases
+  bool VerifyValueRange(int cf, int64_t key, const ReadOptions& opts,
+                        SharedState* shared, const std::string& value_from_db,
+                        const std::string& msg_prefix, const Status& s,
+                        const ExpectedValue& pre_read_expected_value) const {
+    if (shared->HasVerificationFailedYet()) {
+      return false;
+    }
+    const ExpectedValue post_read_expected_value = shared->Get(cf, key);
+    char expected_value_data[kValueMaxLen];
+    size_t expected_value_data_size =
+        GenerateValue(post_read_expected_value.GetValueBase(),
+                      expected_value_data, sizeof(expected_value_data));
+
+    std::ostringstream read_u64ts;
+    if (opts.timestamp) {
+      read_u64ts << " while read with timestamp: ";
+      uint64_t read_ts;
+      if (DecodeU64Ts(*opts.timestamp, &read_ts).ok()) {
+        read_u64ts << std::to_string(read_ts) << ", ";
+      } else {
+        read_u64ts << s.ToString()
+                   << " Encoded read timestamp: " << opts.timestamp->ToString()
+                   << ", ";
+      }
+    }
+
+    // Compare value_from_db with the range of possible values from
+    // pre_read_expected_value to post_read_expected_value
+    if (s.ok()) {
+      const Slice slice(value_from_db);
+      const uint32_t value_base_from_db = GetValueBase(slice);
+      if (ExpectedValueHelper::MustHaveNotExisted(pre_read_expected_value,
+                                                  post_read_expected_value)) {
+        VerificationAbort(shared,
+                          msg_prefix +
+                              ": Unexpected value found that should not exist" +
+                              read_u64ts.str(),
+                          cf, key, value_from_db, "");
+        return false;
+      }
+      if (!ExpectedValueHelper::InExpectedValueBaseRange(
+              value_base_from_db, pre_read_expected_value,
+              post_read_expected_value)) {
+        VerificationAbort(
+            shared,
+            msg_prefix +
+                ": Unexpected value found outside of the value base range" +
+                read_u64ts.str(),
+            cf, key, value_from_db,
+            Slice(expected_value_data, expected_value_data_size));
+        return false;
+      }
+    } else if (s.IsNotFound()) {
+      if (ExpectedValueHelper::MustHaveExisted(pre_read_expected_value,
+                                               post_read_expected_value)) {
+        VerificationAbort(shared,
+                          msg_prefix + ": Value not found which should exist" +
+                              read_u64ts.str() + s.ToString(),
+                          cf, key, "",
+                          Slice(expected_value_data, expected_value_data_size));
+        return false;
+      }
+    } else {
+      VerificationAbort(
+          shared,
+          msg_prefix + "Non-OK status" + read_u64ts.str() + s.ToString(), cf,
+          key, "", Slice(expected_value_data, expected_value_data_size));
+      return false;
+    }
+    return true;
+  }
+
   void PrepareTxnDbOptions(SharedState* shared,
                            TransactionDBOptions& txn_db_opts) override {
     txn_db_opts.rollback_deletion_type_callback =

tools/db_crashtest.py

@@ -485,6 +485,7 @@ def is_direct_io_supported(dbname):
     "write_buffer_size": 32 * 1024 * 1024,
     "level_compaction_dynamic_level_bytes": lambda: random.randint(0, 1),
     "paranoid_file_checks": lambda: random.choice([0, 1, 1, 1]),
+    "test_secondary": lambda: random.choice([0, 1]),
 }
 
 blackbox_simple_default_params = {
@@ -1022,6 +1023,9 @@ def finalize_and_sanitize(src_params):
     if dest_params.get("track_and_verify_wals", 0) == 1:
         dest_params["metadata_write_fault_one_in"] = 0
         dest_params["write_fault_one_in"] = 0
+    # Continuous verification fails with secondaries inside NonBatchedOpsStressTest
+    if dest_params.get("test_secondary") == 1:
+        dest_params["continuous_verification_interval"] = 0
     return dest_params