Verify CPU corruption after each op via --verify_cpu_corruption_dir (#14852)

Summary:

Detection layer of the CPU corruption injector (coming up). With `--verify_cpu_corruption_dir=<dir>`, db_stress reads back the full keyspace after every write/flush/compaction op and compares it to the expected-values model, classifying any mismatch by `kind`: `lost` / `resurrected` / `wrong-value` (silent data corruption) or `detected-corruption` (a status/checksum-caught error). Each finding is written to `<dir>/data_corruption.<tid>.json` ({kind, cf, key, value_from_db, value_from_expected, op_status}) and routed through db_stress's standard `VerificationAbort` for a clean exit-1. A startup guard requires `--threads=1` and all fault injection off so the read-back is single-writer and the only corruption present is the injected one

**Test plan:**
1.Startup guard rejects misconfiguration:
```
--threads=2           -> exit 1: "--verify_cpu_corruption_dir requires --threads=1"
--read_fault_one_in=5 -> exit 1: "requires all fault injection off"
```
2.No false positive (clean CORE preset run, no injection):
```
$ db_stress --verify_cpu_corruption_dir=<dir> --threads=1 (full protections, all *_fault_one_in=0) ...
exit 0; no data_corruption.<tid>.json produced; "Verification successful"
```
3.Write-path cpu corruption injection (coming up, e.g, gdb flips a register inside MemTable::Add), then the immediate post-op read-back catches it. Real `<dir>/data_corruption.<tid>.json`:

silent data corruption -- write returned OK but the key is gone on read-back:
```
{"kind":"lost","cf":0,"key":9814,"value_from_db":"","value_from_expected":"010000000504070609080B0A0D0C0F0E","op_status":"Get: NotFound"}
```
detected corruption -- read-back Get returns Corruption via the memtable per-key checksum:
```
{"kind":"detected-corruption","cf":0,"key":139,"value_from_db":"","value_from_expected":"","op_status":"Get: Corruption: Corrupted memtable entry, per key-value checksum verification failed."
```

Differential Revision: D107999834

db_stress_tool/db_stress_common.h

@@ -352,6 +352,7 @@ DECLARE_int32(prepopulate_blob_cache);
 DECLARE_int32(approximate_size_one_in);
 DECLARE_bool(best_efforts_recovery);
 DECLARE_bool(skip_verifydb);
+DECLARE_string(verify_cpu_corruption_dir);
 DECLARE_bool(paranoid_file_checks);
 DECLARE_uint64(batch_protection_bytes_per_key);
 DECLARE_uint32(memtable_protection_bytes_per_key);

db_stress_tool/db_stress_gflags.cc

@@ -130,6 +130,34 @@ DEFINE_bool(enable_pipelined_write, false, "Pipeline WAL/memtable writes");
 
 DEFINE_bool(verify_before_write, false, "Verify before write");
 
+DEFINE_string(
+    verify_cpu_corruption_dir, "",
+    "If non-empty, right after each write op (put/delete/deleterange), "
+    "flush, and compaction (compactrange/compactfiles), db_stress verifies "
+    "that op for corruption and, on the first hit, writes a result file "
+    "into this directory and marks the run as a verification failure. It "
+    "catches: (a) a corruption returned by the op itself or by the "
+    "read-back (Status::Corruption); or (b) a silent data corruption (SDC) "
+    "-- a read-back that SUCCEEDS but returns the wrong result vs the "
+    "committed expected state: a lost key (NotFound), a resurrected key "
+    "(deleted but present), or a wrong value (bytes differ) -- surfaced by "
+    "an immediate read right after the op, instead of waiting for a later "
+    "read operation or the end-of-run db verification. Running this check "
+    "right after the op gives the most immediate and accurate verification "
+    "when an external CPU-fault injector (e.g. gdb) flips a register inside "
+    "that op. REQUIRES (enforced at startup) --threads=1 and all fault "
+    "injection off (every *_fault_one_in = 0 and sync_fault_injection = "
+    "false): the full-keyspace read-back is only well-defined with a single "
+    "writer, and injected I/O faults would otherwise taint it. OUTPUT "
+    "CONTRACT (one file per worker thread): "
+    "<dir>/data_corruption.<thread_id>.json -- a JSON object with fields: "
+    "kind (one of lost|resurrected|wrong-value|detected-corruption), cf "
+    "(int), key (int), value_from_db (hex), value_from_expected (hex), "
+    "op_status (string). PERFORMANCE: the read-back scans the entire keyspace "
+    "after every op (O(max_key) Get calls per op), so this is meant for "
+    "single-op CPU-fault-injection debugging, not general stress runs. Empty "
+    "(default) = off.");
+
 DEFINE_bool(histogram, false, "Print histogram of operation timings");
 
 DEFINE_bool(destroy_db_initially, true,

db_stress_tool/db_stress_test_base.cc

@@ -8,9 +8,11 @@
 // found in the LICENSE file. See the AUTHORS file for names of contributors.
 //
 
+#include <iomanip>
 #include <ios>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <thread>
 #include <unordered_set>
 
@@ -915,6 +917,148 @@ void StressTest::VerificationAbort(SharedState* shared, int cf, int64_t key,
   shared->SetVerificationFailure();
 }
 
+namespace {
+
+Slice ExpectedValueSlice(const ExpectedValue& expected_value, char* scratch) {
+  const size_t size =
+      GenerateValue(expected_value.GetValueBase(), scratch, kValueMaxLen);
+  return Slice(scratch, size);
+}
+
+struct DataCorruption {
+  // "lost" | "resurrected" | "wrong-value" | "detected-corruption"
+  const char* kind;
+  Slice value_from_db;
+  Slice value_from_expected;
+  std::string op_status;
+};
+
+std::optional<DataCorruption> ClassifyReadBack(
+    const Status& read_status, const std::string& db_value,
+    const ExpectedValue& expected_value, char* scratch) {
+  if (read_status.IsCorruption()) {
+    return DataCorruption{"detected-corruption", db_value, Slice(),
+                          "Get: " + read_status.ToString()};
+  }
+  // Fault injection is off (enforced at startup), so the read-back can only be
+  // OK or NotFound here. Anything else is outside this tool's contract -- stop
+  // now rather than guess a classification. ImmediateExit (not exit/abort): the
+  // DB is open with live background threads, so a normal exit would race their
+  // static teardown (see port::ImmediateExit).
+  if (!read_status.ok() && !read_status.IsNotFound()) {
+    fprintf(stderr, "verify_cpu_corruption: unexpected read-back status: %s\n",
+            read_status.ToString().c_str());
+    port::ImmediateExit(1);
+  }
+  const bool present_in_db = read_status.ok();
+  // Exists() asserts the key is not mid-write -- which holds under --threads=1:
+  // the just-run op committed before this read-back.
+  const bool expected_to_exist = expected_value.Exists();
+
+  if (!expected_to_exist && !present_in_db) {
+    return std::nullopt;  // Agree: key absent.
+  }
+  if (expected_to_exist && !present_in_db) {
+    return DataCorruption{"lost", Slice(),
+                          ExpectedValueSlice(expected_value, scratch),
+                          "Get: NotFound"};
+  }
+  if (!expected_to_exist && present_in_db) {
+    return DataCorruption{"resurrected", db_value, Slice(), "Get: OK"};
+  }
+  // Present in both: compare the values.
+  const Slice expected = ExpectedValueSlice(expected_value, scratch);
+  if (Slice(db_value) == expected) {
+    return std::nullopt;  // Agree: same value.
+  }
+  return DataCorruption{"wrong-value", db_value, expected, "Get: OK"};
+}
+
+void WriteDataCorruption(uint32_t thread_id, int cf, int64_t key,
+                         const DataCorruption& corruption) {
+  std::ostringstream oss;
+  oss << "{\"kind\":\"" << corruption.kind << "\",\"cf\":" << cf
+      << ",\"key\":" << key << ",\"value_from_db\":\""
+      << corruption.value_from_db.ToString(/* hex */ true)
+      << "\",\"value_from_expected\":\""
+      << corruption.value_from_expected.ToString(/* hex */ true)
+      << "\",\"op_status\":" << std::quoted(corruption.op_status) << "}\n";
+  const std::string path = FLAGS_verify_cpu_corruption_dir +
+                           "/data_corruption." + std::to_string(thread_id) +
+                           ".json";
+  FILE* f = fopen(path.c_str(), "w");
+  if (f == nullptr) {
+    fprintf(stderr, "Failed to open CPU corruption result file: %s\n",
+            path.c_str());
+    return;
+  }
+  fputs(oss.str().c_str(), f);
+  fclose(f);
+}
+
+}  // namespace
+
+void StressTest::MaybeVerifyCpuCorruption(ThreadState* thread,
+                                          const char* op_label,
+                                          const Status& op_status) {
+  if (FLAGS_verify_cpu_corruption_dir.empty()) {
+    return;
+  }
+  auto shared = thread->shared;
+
+  // The op itself returned Corruption: an integrity check caught the
+  // corruption.
+  if (op_status.IsCorruption()) {
+    WriteDataCorruption(thread->tid, /* cf */ -1, /* key */ -1,
+                        {"detected-corruption", Slice(), Slice(),
+                         std::string(op_label) + ": " + op_status.ToString()});
+    VerificationAbort(shared,
+                      std::string(op_label) +
+                          " detected data corruption: " + op_status.ToString());
+    return;
+  }
+
+  // Single writer (the run is pinned to --threads=1): read every key back and
+  // compare it against the committed expected state, ending the run on the
+  // first data corruption found.
+  // TODO: this full-keyspace read-back duplicates the canonical verifier
+  // VerifyDb (no_batched_ops_stress.cc). Refactor that key/value compare into a
+  // shared per-key helper both can call, so this stays in sync with the model.
+  ReadOptions read_opts;
+  read_opts.verify_checksums = true;
+  // A user-timestamp DB needs a read timestamp; supply one like the canonical
+  // verifier (no_batched_ops_stress.cc).
+  std::string read_ts_str;
+  Slice read_ts;
+  if (FLAGS_user_timestamp_size > 0) {
+    read_ts_str = GetNowNanos();
+    read_ts = read_ts_str;
+    read_opts.timestamp = &read_ts;
+  }
+  const int64_t max_key = shared->GetMaxKey();
+  std::string db_value;
+  char expected_scratch[kValueMaxLen];
+  for (int cf = 0; cf < static_cast<int>(column_families_.size()); ++cf) {
+    for (int64_t key = 0; key < max_key; ++key) {
+      const ExpectedValue expected = shared->Get(cf, key);
+      db_value.clear();
+      const Status s =
+          db_->Get(read_opts, column_families_[cf], Key(key), &db_value);
+      const std::optional<DataCorruption> corruption =
+          ClassifyReadBack(s, db_value, expected, expected_scratch);
+      if (corruption.has_value()) {
+        WriteDataCorruption(thread->tid, cf, key, *corruption);
+        VerificationAbort(
+            shared,
+            std::string(corruption->kind) + " (" + corruption->op_status + ")",
+            cf, key, corruption->value_from_db,
+            corruption->value_from_expected);
+        return;
+      }
+    }
+  }
+}
+
 std::string StressTest::DebugString(const Slice& value,
                                     const WideColumns& columns) {
   std::ostringstream oss;
@@ -1544,7 +1688,8 @@ void StressTest::OperateDb(ThreadState* thread) {
       ColumnFamilyHandle* column_family = column_families_[rand_column_family];
 
       if (thread->rand.OneInOpt(FLAGS_compact_files_one_in)) {
-        TestCompactFiles(thread, column_family);
+        Status s = TestCompactFiles(thread, column_family);
+        MaybeVerifyCpuCorruption(thread, "compactfiles", s);
       }
 
       int64_t rand_key = GenerateOneKey(thread, i);
@@ -1552,7 +1697,8 @@ void StressTest::OperateDb(ThreadState* thread) {
       Slice key = keystr;
 
       if (thread->rand.OneInOpt(FLAGS_compact_range_one_in)) {
-        TestCompactRange(thread, rand_key, key, column_family);
+        Status s = TestCompactRange(thread, rand_key, key, column_family);
+        MaybeVerifyCpuCorruption(thread, "compactrange", s);
         if (thread->shared->HasVerificationFailedYet()) {
           break;
         }
@@ -1567,6 +1713,7 @@ void StressTest::OperateDb(ThreadState* thread) {
 
       if (thread->rand.OneInOpt(FLAGS_flush_one_in)) {
         Status status = TestFlush(rand_column_families);
+        MaybeVerifyCpuCorruption(thread, "flush", status);
         ProcessStatus(shared, "Flush", status);
       }
 
@@ -1792,8 +1939,11 @@ void StressTest::OperateDb(ThreadState* thread) {
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->DisableAllThreadLocalErrorInjection();
         }
-        TestPut(thread, write_opts, read_opts, rand_column_families, rand_keys,
-                value);
+        Status write_status = TestPut(thread, write_opts, read_opts,
+                                      rand_column_families, rand_keys, value);
+        // Verify before re-enabling injection: the read-back must run in the
+        // no-injection window so injected I/O errors cannot taint it.
+        MaybeVerifyCpuCorruption(thread, "put", write_status);
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->EnableAllThreadLocalErrorInjection();
         }
@@ -1803,7 +1953,10 @@ void StressTest::OperateDb(ThreadState* thread) {
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->DisableAllThreadLocalErrorInjection();
         }
-        TestDelete(thread, write_opts, rand_column_families, rand_keys);
+        Status del_status =
+            TestDelete(thread, write_opts, rand_column_families, rand_keys);
+        // Verify before re-enabling injection (see the put case).
+        MaybeVerifyCpuCorruption(thread, "delete", del_status);
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->EnableAllThreadLocalErrorInjection();
         }
@@ -1813,7 +1966,10 @@ void StressTest::OperateDb(ThreadState* thread) {
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->DisableAllThreadLocalErrorInjection();
         }
-        TestDeleteRange(thread, write_opts, rand_column_families, rand_keys);
+        Status delrange_status = TestDeleteRange(
+            thread, write_opts, rand_column_families, rand_keys);
+        // Verify before re-enabling injection (see the put case).
+        MaybeVerifyCpuCorruption(thread, "deleterange", delrange_status);
         if (disable_fault_injection_during_user_write) {
           db_fault_injection_fs_->EnableAllThreadLocalErrorInjection();
         }
@@ -3504,13 +3660,13 @@ Status StressTest::TestGetPropertiesOfAllTables() const {
   return db_->GetPropertiesOfAllTables(&props);
 }
 
-void StressTest::TestCompactFiles(ThreadState* thread,
-                                  ColumnFamilyHandle* column_family) {
+Status StressTest::TestCompactFiles(ThreadState* thread,
+                                    ColumnFamilyHandle* column_family) {
   ROCKSDB_NAMESPACE::ColumnFamilyMetaData cf_meta_data;
   db_->GetColumnFamilyMetaData(column_family, &cf_meta_data);
 
   if (cf_meta_data.levels.empty()) {
-    return;
+    return Status::OK();
   }
 
   // Randomly compact up to three consecutive files from a level
@@ -3569,9 +3725,10 @@ void StressTest::TestCompactFiles(ThreadState* thread,
       } else {
         thread->stats.AddNumCompactFilesSucceed(1);
       }
-      break;
+      return s;
     }
   }
+  return Status::OK();
 }
 
 void StressTest::TestPromoteL0(ThreadState* thread,
@@ -3761,9 +3918,9 @@ Status StressTest::MaybeReleaseSnapshots(ThreadState* thread, uint64_t i) {
   return Status::OK();
 }
 
-void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
-                                  const Slice& start_key,
-                                  ColumnFamilyHandle* column_family) {
+Status StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
+                                    const Slice& start_key,
+                                    ColumnFamilyHandle* column_family) {
   int64_t end_key_num;
   if (std::numeric_limits<int64_t>::max() - rand_key <
       FLAGS_compact_range_width) {
@@ -3878,6 +4035,7 @@ void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
       db_fault_injection_fs_->EnableAllThreadLocalErrorInjection();
     }
   }
+  return status;
 }
 
 uint32_t StressTest::GetRangeHash(ThreadState* thread, const Snapshot* snapshot,

db_stress_tool/db_stress_test_base.h

@@ -252,10 +252,10 @@ class StressTest {
       const std::vector<int64_t>& rand_keys) = 0;
 
   // Issue compact range, starting with start_key, whose integer value
-  // is rand_key.
-  virtual void TestCompactRange(ThreadState* thread, int64_t rand_key,
-                                const Slice& start_key,
-                                ColumnFamilyHandle* column_family);
+  // is rand_key. Returns the CompactRange() status.
+  virtual Status TestCompactRange(ThreadState* thread, int64_t rand_key,
+                                  const Slice& start_key,
+                                  ColumnFamilyHandle* column_family);
 
   virtual void TestPromoteL0(ThreadState* thread,
                              ColumnFamilyHandle* column_family);
@@ -378,7 +378,9 @@ class StressTest {
                                 const std::vector<int>& rand_column_families,
                                 const std::vector<int64_t>& rand_keys);
 
-  void TestCompactFiles(ThreadState* thread, ColumnFamilyHandle* column_family);
+  // Returns the CompactFiles() status (OK if no compaction was performed).
+  Status TestCompactFiles(ThreadState* thread,
+                          ColumnFamilyHandle* column_family);
 
   Status TestFlush(const std::vector<int>& rand_column_families);
 
@@ -433,6 +435,13 @@ class StressTest {
   void VerificationAbort(SharedState* shared, int cf, int64_t key,
                          const Slice& value, const WideColumns& columns) const;
 
+  // Under --verify_cpu_corruption_dir (see that flag's comment for behavior and
+  // the result-file contract), verifies the just-run op (named by `op_label`,
+  // e.g. "put"/"flush"/"compactrange") for a returned/read-back corruption or a
+  // silent data corruption. A no-op when the flag is empty.
+  void MaybeVerifyCpuCorruption(ThreadState* thread, const char* op_label,
+                                const Status& op_status);
+
   static std::string DebugString(const Slice& value,
                                  const WideColumns& columns);
 

db_stress_tool/db_stress_tool.cc

@@ -176,6 +176,45 @@ int db_stress_tool(int argc, char** argv) {
     }
   }
 
+  if (!FLAGS_verify_cpu_corruption_dir.empty()) {
+    // The full-keyspace read-back is only well-defined with a single writer,
+    // and injected I/O faults would taint it -- so require --threads=1 and all
+    // fault injection off (see the flag's comment).
+    if (FLAGS_threads != 1) {
+      fprintf(stderr,
+              "Error: --verify_cpu_corruption_dir requires --threads=1\n");
+      exit(1);  // NOLINT(concurrency-mt-unsafe)
+    }
+    if (FLAGS_read_fault_one_in != 0 || FLAGS_write_fault_one_in != 0 ||
+        FLAGS_metadata_read_fault_one_in != 0 ||
+        FLAGS_metadata_write_fault_one_in != 0 ||
+        FLAGS_open_read_fault_one_in != 0 ||
+        FLAGS_open_write_fault_one_in != 0 ||
+        FLAGS_open_metadata_read_fault_one_in != 0 ||
+        FLAGS_open_metadata_write_fault_one_in != 0 ||
+        FLAGS_secondary_cache_fault_one_in != 0 || FLAGS_sync_fault_injection) {
+      fprintf(stderr,
+              "Error: --verify_cpu_corruption_dir requires all fault injection "
+              "off (every *_fault_one_in = 0 and sync_fault_injection = "
+              "false)\n");
+      exit(1);  // NOLINT(concurrency-mt-unsafe)
+    }
+    // The read-back compares against the expected-state model, which only the
+    // default (state-tracked) stress test maintains. The
+    // batched/cf-consistency/ multi-ops-txns variants set
+    // IsStateTracked()=false, so every key would look absent -- reject them
+    // rather than report false losses.
+    if (FLAGS_test_batches_snapshots || FLAGS_test_cf_consistency ||
+        FLAGS_test_multi_ops_txns) {
+      fprintf(stderr,
+              "Error: --verify_cpu_corruption_dir requires the default "
+              "state-tracked stress test; it is incompatible with "
+              "--test_batches_snapshots, --test_cf_consistency, and "
+              "--test_multi_ops_txns\n");
+      exit(1);  // NOLINT(concurrency-mt-unsafe)
+    }
+  }
+
   FLAGS_rep_factory = StringToRepFactory(FLAGS_memtablerep.c_str());
 
   // The number of background threads should be at least as much the

db_stress_tool/multi_ops_txns_stress.cc

@@ -481,7 +481,7 @@ void MultiOpsTxnsStressTest::TestIngestExternalFile(
   (void)rand_column_families;
 }
 
-void MultiOpsTxnsStressTest::TestCompactRange(
+Status MultiOpsTxnsStressTest::TestCompactRange(
     ThreadState* thread, int64_t /*rand_key*/, const Slice& /*start_key*/,
     ColumnFamilyHandle* column_family) {
   // TODO (yanqin).
@@ -489,6 +489,7 @@ void MultiOpsTxnsStressTest::TestCompactRange(
   // completes.
   (void)thread;
   (void)column_family;
+  return Status::OK();
 }
 
 Status MultiOpsTxnsStressTest::TestBackupRestore(

db_stress_tool/multi_ops_txns_stress.h

@@ -253,9 +253,9 @@ class MultiOpsTxnsStressTest : public StressTest {
                               const std::vector<int>& rand_column_families,
                               const std::vector<int64_t>& rand_keys) override;
 
-  void TestCompactRange(ThreadState* thread, int64_t rand_key,
-                        const Slice& start_key,
-                        ColumnFamilyHandle* column_family) override;
+  Status TestCompactRange(ThreadState* thread, int64_t rand_key,
+                          const Slice& start_key,
+                          ColumnFamilyHandle* column_family) override;
 
   Status TestBackupRestore(ThreadState* thread,
                            const std::vector<int>& rand_column_families,