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

Summary:
Pull Request resolved: https://github.com/facebook/rocksdb/pull/14852

Detection layer of the CPU corruption injector (https://github.com/facebook/rocksdb/pull/14858). With `--verify_cpu_corruption_dir=<dir>`, db_stress reads back the full keyspace after every write/manual flush/manual 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

Bonus: a minor refactoring into the surrounding error handling code in these ops

**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."
```

4.See PR https://github.com/facebook/rocksdb/pull/14866 test plan's spread in the outcome for verification of detection

Reviewed By: pdillinger

Differential Revision: D107999834

fbshipit-source-id: 18decbf51dc56eec62b735136e2dfb4e1175b773

db_stress_tool/db_stress_common.h

@@ -353,6 +353,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,14 @@ 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, "",
+    "When non-empty, activates a slow, meticulous verification mode intended "
+    "only for use with a CPU fault injector; on the first corruption found it "
+    "writes a result file here and fails the run. See "
+    "StressTest::MaybeVerifyCpuCorruption for the full behavior and "
+    "output-file contract. 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

@@ -9,9 +9,11 @@
 //
 
 #include <cstdlib>
+#include <iomanip>
 #include <ios>
 #include <memory>
 #include <mutex>
+#include <optional>
 #include <thread>
 #include <unordered_set>
 
@@ -286,6 +288,95 @@ std::string GetFaultInjectionLogPath(const std::string& db_label) {
          ".bin";
 }
 
+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 << "{\n"
+      << "  \"kind\": \"" << corruption.kind << "\",\n"
+      << "  \"cf\": " << cf << ",\n"
+      << "  \"key\": " << key << ",\n"
+      << "  \"value_from_db\": \""
+      << corruption.value_from_db.ToString(/* hex */ true) << "\",\n"
+      << "  \"value_from_expected\": \""
+      << corruption.value_from_expected.ToString(/* hex */ true) << "\",\n"
+      << "  \"op_status\": " << std::quoted(corruption.op_status) << "\n"
+      << "}\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());
+    port::ImmediateExit(1);
+  }
+  if (fputs(oss.str().c_str(), f) == EOF) {
+    fprintf(stderr, "Failed to write CPU corruption result file: %s\n",
+            path.c_str());
+    port::ImmediateExit(1);
+  }
+  if (fclose(f) != 0) {
+    fprintf(stderr, "Failed to close CPU corruption result file: %s\n",
+            path.c_str());
+    port::ImmediateExit(1);
+  }
+}
+
 }  // namespace
 
 const std::string& StressTest::GetDbLabel() const { return db_label_; }
@@ -927,6 +1018,97 @@ void StressTest::VerificationAbort(SharedState* shared, int cf, int64_t key,
   shared->SetVerificationFailure();
 }
 
+// Under --verify_cpu_corruption_dir, right after each write op
+// (put/delete/deleterange), flush, and compaction (compactrange/compactfiles),
+// verifies that op for corruption and, on the first one found, writes a result
+// file into the 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). Surfacing these with an immediate read right after the op
+// (instead of waiting for a later read or the end-of-run db verification) 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.
+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;
+
+  if (shared->HasVerificationFailedYet()) {
+    return;
+  }
+
+  // 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;
@@ -1582,8 +1764,7 @@ void StressTest::OperateDb(ThreadState* thread) {
           GenerateColumnFamilies(FLAGS_column_families, rand_column_family);
 
       if (thread->rand.OneInOpt(FLAGS_flush_one_in)) {
-        Status status = TestFlush(rand_column_families);
-        ProcessStatus(shared, "Flush", status);
+        TestFlush(thread, rand_column_families);
       }
 
       if (thread->rand.OneInOpt(FLAGS_get_live_files_apis_one_in)) {
@@ -3568,23 +3749,21 @@ void StressTest::TestCompactFiles(ThreadState* thread,
                                  static_cast<int>(output_level));
       if (!s.ok()) {
         thread->stats.AddNumCompactFilesFailed(1);
-        // TOOD (hx235): allow an exact list of tolerable failures under stress
-        // test
-        bool non_ok_status_allowed =
-            s.IsManualCompactionPaused() || s.IsCompactionAborted() ||
-            IsErrorInjectedAndRetryable(s) || s.IsAborted() ||
-            s.IsInvalidArgument() || s.IsNotSupported();
-        if (!non_ok_status_allowed) {
-          fprintf(stderr,
-                  "Unable to perform CompactFiles(): %s under specified "
-                  "CompactionOptions: %s (Empty string or "
-                  "missing field indicates default option or value is used)\n",
-                  s.ToString().c_str(), compact_opt_oss.str().c_str());
-          thread->shared->SafeTerminate();
-        }
       } else {
         thread->stats.AddNumCompactFilesSucceed(1);
       }
+      // Verify before the fail-fast: a corruption that CompactFiles itself
+      // returns must be recorded (CORRUPTION) before we terminate, or it
+      // mis-buckets as CRASH. No-op unless CPU-corruption verification is on.
+      MaybeVerifyCpuCorruption(thread, "compactfiles", s);
+      if (!s.ok() && !IsTolerableCompactionFailure(s)) {
+        fprintf(stderr,
+                "Unable to perform CompactFiles(): %s under specified "
+                "CompactionOptions: %s (Empty string or missing field "
+                "indicates default option or value is used)\n",
+                s.ToString().c_str(), compact_opt_oss.str().c_str());
+        thread->shared->SafeTerminate();
+      }
       break;
     }
   }
@@ -3614,16 +3793,24 @@ void StressTest::TestPromoteL0(ThreadState* thread,
   }
 }
 
-Status StressTest::TestFlush(const std::vector<int>& rand_column_families) {
+void StressTest::TestFlush(ThreadState* thread,
+                           const std::vector<int>& rand_column_families) {
   FlushOptions flush_opts;
   assert(flush_opts.wait);
+  Status status;
   if (FLAGS_atomic_flush) {
-    return db_->Flush(flush_opts, column_families_);
-  }
+    status = db_->Flush(flush_opts, column_families_);
+  } else {
     std::vector<ColumnFamilyHandle*> cfhs;
-  std::for_each(rand_column_families.begin(), rand_column_families.end(),
+    std::for_each(
+        rand_column_families.begin(), rand_column_families.end(),
         [this, &cfhs](int k) { cfhs.push_back(column_families_[k]); });
-  return db_->Flush(flush_opts, cfhs);
+    status = db_->Flush(flush_opts, cfhs);
+  }
+  // Verify before the fail-fast (a no-op unless CPU-corruption verification is
+  // on), then route the status through the standard flush error handling.
+  MaybeVerifyCpuCorruption(thread, "flush", status);
+  ProcessStatus(thread->shared, "Flush", status);
 }
 
 Status StressTest::TestResetStats() { return db_->ResetStats(); }
@@ -3852,22 +4039,19 @@ void StressTest::TestCompactRange(ThreadState* thread, int64_t rand_key,
                         << cro.blob_garbage_collection_age_cutoff;
   Status status = db_->CompactRange(cro, column_family, &start_key, &end_key);
 
-  if (!status.ok()) {
-    // TOOD (hx235): allow an exact list of tolerable failures under stress test
-    bool non_ok_status_allowed =
-        status.IsManualCompactionPaused() || status.IsCompactionAborted() ||
-        IsErrorInjectedAndRetryable(status) || status.IsAborted() ||
-        status.IsInvalidArgument() || status.IsNotSupported();
-    if (!non_ok_status_allowed) {
+  // Verify before the fail-fast: a corruption that CompactRange itself returns
+  // must be recorded (CORRUPTION) before we flag failure, or it mis-buckets.
+  // No-op unless CPU-corruption verification is on.
+  MaybeVerifyCpuCorruption(thread, "compactrange", status);
+  if (!status.ok() && !IsTolerableCompactionFailure(status)) {
     fprintf(stderr,
             "Unable to perform CompactRange(): %s under specified "
-              "CompactRangeOptions: %s (Empty string or "
-              "missing field indicates default option or value is used)\n",
+            "CompactRangeOptions: %s (Empty string or missing field "
+            "indicates default option or value is used)\n",
             status.ToString().c_str(), compact_range_opt_oss.str().c_str());
     // Fail fast to preserve the DB state.
     thread->shared->SetVerificationFailure();
   }
-  }
 
   if (pre_snapshot != nullptr) {
     // Temporarily disable error injection for validation

db_stress_tool/db_stress_test_base.h

@@ -36,6 +36,14 @@ class StressTest {
            !status_to_io_status(Status(error_s)).GetDataLoss();
   }
 
+  static bool IsTolerableCompactionFailure(const Status& s) {
+    // TOOD (hx235): allow an exact list of tolerable failures under stress
+    // test
+    return s.IsManualCompactionPaused() || s.IsCompactionAborted() ||
+           IsErrorInjectedAndRetryable(s) || s.IsAborted() ||
+           s.IsInvalidArgument() || s.IsNotSupported();
+  }
+
   // Returns true if the status is an expected transactional error, including
   // lock conflicts (deadlock or timeout) from MaybeAddKeyToTxnForRYW writing
   // to the same key space without the stress-test-level mutex, and TryAgain
@@ -380,7 +388,8 @@ class StressTest {
 
   void TestCompactFiles(ThreadState* thread, ColumnFamilyHandle* column_family);
 
-  Status TestFlush(const std::vector<int>& rand_column_families);
+  void TestFlush(ThreadState* thread,
+                 const std::vector<int>& rand_column_families);
 
   Status TestResetStats();
 
@@ -433,6 +442,14 @@ class StressTest {
   void VerificationAbort(SharedState* shared, int cf, int64_t key,
                          const Slice& value, const WideColumns& columns) const;
 
+  // Under --verify_cpu_corruption_dir, 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. See
+  // the definition in db_stress_test_base.cc for the full behavior, startup
+  // requirements, performance cost, and result-file contract.
+  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/no_batched_ops_stress.cc

@@ -2021,7 +2021,22 @@ class NonBatchedOpsStressTest : public StressTest {
       if (IsErrorInjectedAndRetryable(s)) {
         assert(!initial_wal_write_may_succeed);
         return s;
-      } else if (FLAGS_inject_error_severity == 2) {
+      }
+    } else {
+      PrintWriteRecoveryWaitTimeIfNeeded(
+          raw_env, initial_write_s, initial_wal_write_may_succeed,
+          wait_for_recover_start_time, "TestPut");
+      pending_expected_value.Commit();
+      thread->stats.AddBytesForWrites(1, sz);
+      PrintKeyValue(rand_column_family, static_cast<uint32_t>(rand_key), value,
+                    sz);
+    }
+    // Single write verification point (no_batched owns it): on success a
+    // read-back after Commit; on failure the op status before the fail-fast.
+    // No-op unless CPU-corruption verification is on.
+    MaybeVerifyCpuCorruption(thread, "put", s);
+    if (!s.ok()) {
+      if (FLAGS_inject_error_severity == 2) {
         if (!is_db_stopped_ && s.severity() >= Status::Severity::kFatalError) {
           is_db_stopped_ = true;
         } else if (!is_db_stopped_ ||
@@ -2033,14 +2048,6 @@ class NonBatchedOpsStressTest : public StressTest {
         fprintf(stderr, "put or merge error: %s\n", s.ToString().c_str());
         thread->shared->SafeTerminate();
       }
-    } else {
-      PrintWriteRecoveryWaitTimeIfNeeded(
-          raw_env, initial_write_s, initial_wal_write_may_succeed,
-          wait_for_recover_start_time, "TestPut");
-      pending_expected_value.Commit();
-      thread->stats.AddBytesForWrites(1, sz);
-      PrintKeyValue(rand_column_family, static_cast<uint32_t>(rand_key), value,
-                    sz);
     }
     return s;
   }
@@ -2126,7 +2133,17 @@ class NonBatchedOpsStressTest : public StressTest {
         if (IsErrorInjectedAndRetryable(s)) {
           assert(!initial_wal_write_may_succeed);
           return s;
-        } else if (FLAGS_inject_error_severity == 2) {
+        }
+      } else {
+        PrintWriteRecoveryWaitTimeIfNeeded(
+            raw_env, initial_write_s, initial_wal_write_may_succeed,
+            wait_for_recover_start_time, "TestDelete");
+        pending_expected_value.Commit();
+        thread->stats.AddDeletes(1);
+      }
+      MaybeVerifyCpuCorruption(thread, "delete", s);
+      if (!s.ok()) {
+        if (FLAGS_inject_error_severity == 2) {
           if (!is_db_stopped_ &&
               s.severity() >= Status::Severity::kFatalError) {
             is_db_stopped_ = true;
@@ -2139,12 +2156,6 @@ class NonBatchedOpsStressTest : public StressTest {
           fprintf(stderr, "delete error: %s\n", s.ToString().c_str());
           thread->shared->SafeTerminate();
         }
-      } else {
-        PrintWriteRecoveryWaitTimeIfNeeded(
-            raw_env, initial_write_s, initial_wal_write_may_succeed,
-            wait_for_recover_start_time, "TestDelete");
-        pending_expected_value.Commit();
-        thread->stats.AddDeletes(1);
       }
     } else {
       PendingExpectedValue pending_expected_value =
@@ -2198,7 +2209,17 @@ class NonBatchedOpsStressTest : public StressTest {
         if (IsErrorInjectedAndRetryable(s)) {
           assert(!initial_wal_write_may_succeed);
           return s;
-        } else if (FLAGS_inject_error_severity == 2) {
+        }
+      } else {
+        PrintWriteRecoveryWaitTimeIfNeeded(
+            raw_env, initial_write_s, initial_wal_write_may_succeed,
+            wait_for_recover_start_time, "TestDelete");
+        pending_expected_value.Commit();
+        thread->stats.AddSingleDeletes(1);
+      }
+      MaybeVerifyCpuCorruption(thread, "singledelete", s);
+      if (!s.ok()) {
+        if (FLAGS_inject_error_severity == 2) {
           if (!is_db_stopped_ &&
               s.severity() >= Status::Severity::kFatalError) {
             is_db_stopped_ = true;
@@ -2211,12 +2232,6 @@ class NonBatchedOpsStressTest : public StressTest {
           fprintf(stderr, "single delete error: %s\n", s.ToString().c_str());
           thread->shared->SafeTerminate();
         }
-      } else {
-        PrintWriteRecoveryWaitTimeIfNeeded(
-            raw_env, initial_write_s, initial_wal_write_may_succeed,
-            wait_for_recover_start_time, "TestDelete");
-        pending_expected_value.Commit();
-        thread->stats.AddSingleDeletes(1);
       }
     }
     return s;
@@ -2292,17 +2307,6 @@ class NonBatchedOpsStressTest : public StressTest {
       if (IsErrorInjectedAndRetryable(s)) {
         assert(!initial_wal_write_may_succeed);
         return s;
-      } else if (FLAGS_inject_error_severity == 2) {
-        if (!is_db_stopped_ && s.severity() >= Status::Severity::kFatalError) {
-          is_db_stopped_ = true;
-        } else if (!is_db_stopped_ ||
-                   s.severity() < Status::Severity::kFatalError) {
-          fprintf(stderr, "delete range error: %s\n", s.ToString().c_str());
-          thread->shared->SafeTerminate();
-        }
-      } else {
-        fprintf(stderr, "delete range error: %s\n", s.ToString().c_str());
-        thread->shared->SafeTerminate();
       }
     } else {
       PrintWriteRecoveryWaitTimeIfNeeded(
@@ -2315,6 +2319,24 @@ class NonBatchedOpsStressTest : public StressTest {
       thread->stats.AddRangeDeletions(1);
       thread->stats.AddCoveredByRangeDeletions(covered);
     }
+    // Single write verification point (no_batched owns it): on success a
+    // read-back after Commit; on failure the op status before the fail-fast.
+    // No-op unless CPU-corruption verification is on.
+    MaybeVerifyCpuCorruption(thread, "deleterange", s);
+    if (!s.ok()) {
+      if (FLAGS_inject_error_severity == 2) {
+        if (!is_db_stopped_ && s.severity() >= Status::Severity::kFatalError) {
+          is_db_stopped_ = true;
+        } else if (!is_db_stopped_ ||
+                   s.severity() < Status::Severity::kFatalError) {
+          fprintf(stderr, "delete range error: %s\n", s.ToString().c_str());
+          thread->shared->SafeTerminate();
+        }
+      } else {
+        fprintf(stderr, "delete range error: %s\n", s.ToString().c_str());
+        thread->shared->SafeTerminate();
+      }
+    }
     return s;
   }