Rocksdb Crash Test failed: assertion failed - corruption: Sequence number is being set backwards dur (#14567) (#14567)

Summary:
Handle the case where a transaction commit with `commit_bypass_memtable` succeeds in writing the commit marker to WAL, but then `IngestWBWIAsMemtable` fails (e.g., WAL creation failure during `SwitchMemtable` or invalid column family). Previously this failure path was not distinguished from a direct WBWI ingest, so the DB could continue operating with committed data durable in WAL but not published to memtables.

This resulted in a higher seqno in the WAL that what the DB's latest seqno was.

Now `IngestWBWIAsMemtable` takes an `ingest_wbwi_for_commit` parameter. When true and the ingest fails (without a prior memtable update), the DB sets a fatal background error, stopping all writes until the DB is closed and reopened for WAL recovery. A close and reopen is required because the auto-recovery mechanism works by flushing memtables to disk, but here the committed data was never published to memtables in the first place — it only exists as prepare + commit records in the WAL. Only a full WAL replay during `DB::Open` can reconstruct the committed transaction data and insert it into memtables.

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

Test Plan:
- Added `CommitBypassMemtableTest.SwitchMemtableFailureStopsDBUntilReopen`:
  - Injects an IO error during `SwitchMemtable` WAL creation via `TEST_SYNC_POINT_CALLBACK`
  - Verifies commit returns `Corruption` status
  - Verifies DB enters fatal background error state and rejects further writes
  - Verifies that after close and reopen, committed data is recovered from WAL and the DB resumes normal operation

Reviewed By: xingbowang

Differential Revision: D98638705

Pulled By: joshkang97

fbshipit-source-id: f116b1f257b19984b0413f6aeba0ba4a7c0a5b25

db/db_impl/db_impl.h

@@ -1555,10 +1555,17 @@ class DBImpl : public DB {
   // @param memtable_updated Whether the same write that ingests wbwi has
   // updated memtable. This is useful for determining whether to set bg
   // error when IngestWBWIAsMemtable fails.
+  // @param ingest_wbwi_for_commit Whether wbwi ingestion is publishing the
+  // committed data of a prepared transaction. This means a failure can leave
+  // committed data durable in WAL but not published in memtables.
+  // @param ignore_missing_cf If true, skip column families not found in the DB
+  // instead of returning an error.
   Status IngestWBWIAsMemtable(std::shared_ptr<WriteBatchWithIndex> wbwi,
                               const WBWIMemTable::SeqnoRange& assigned_seqno,
                               uint64_t min_prep_log, SequenceNumber last_seqno,
-                              bool memtable_updated, bool ignore_missing_cf);
+                              bool memtable_updated,
+                              bool ingest_wbwi_for_commit,
+                              bool ignore_missing_cf);
 
   // If disable_memtable is set the application logic must guarantee that the
   // batch will still be skipped from memtable during the recovery. An excption

db/db_impl/db_impl_write.cc

@@ -281,7 +281,7 @@ Status DBImpl::IngestWBWIAsMemtable(
     std::shared_ptr<WriteBatchWithIndex> wbwi,
     const WBWIMemTable::SeqnoRange& assigned_seqno, uint64_t min_prep_log,
     SequenceNumber last_seqno_after_ingest, bool memtable_updated,
-    bool ignore_missing_cf) {
+    bool ingest_wbwi_for_commit, bool ignore_missing_cf) {
   // Keys in new memtable have seqno > last_seqno_after_ingest >= keys in wbwi.
   assert(assigned_seqno.upper_bound <= last_seqno_after_ingest);
   // Keys in the current memtable have seqno <= LastSequence() < keys in wbwi.
@@ -310,8 +310,16 @@ Status DBImpl::IngestWBWIAsMemtable(
           std::to_string(cf_id));
       if (memtable_updated) {
         s = Status::Corruption(
-            "Part of the write batch is applied. Memtable is in a inconsistent "
-            "state. " +
+            "Part of the write batch is applied. Memtable is in an "
+            "inconsistent state due to invalid column family. " +
+            s.ToString());
+        error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable);
+      } else if (ingest_wbwi_for_commit) {
+        s = Status::Corruption(
+            "Commit marker is durable in WAL, but publishing committed WBWI "
+            "data to memtable failed due to invalid column family. This DB "
+            "instance cannot safely resume; close and reopen the DB for WAL "
+            "recovery. " +
             s.ToString());
         error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable);
       }
@@ -376,8 +384,16 @@ Status DBImpl::IngestWBWIAsMemtable(
       if (i != 0 || memtable_updated) {
         // escalate error to non-recoverable
         s = Status::Corruption(
-            "Part of the write batch is applied. Memtable is in a inconsistent "
-            "state. " +
+            "Part of the write batch is applied. Memtable is in an "
+            "inconsistent state due to SwitchMemtable failure. " +
+            s.ToString());
+        error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable);
+      } else if (ingest_wbwi_for_commit) {
+        s = Status::Corruption(
+            "Commit marker is durable in WAL, but publishing committed WBWI "
+            "data to memtable failed due to SwitchMemtable failure. This DB "
+            "instance cannot safely resume; close and reopen the DB for WAL "
+            "recovery. " +
             s.ToString());
         error_handler_.SetBGError(s, BackgroundErrorReason::kMemTable);
       } else {
@@ -1260,11 +1276,11 @@ Status DBImpl::WriteImpl(const WriteOptions& write_options,
       if (two_write_queues_) {
         assert(ub <= versions_->LastAllocatedSequence());
       }
-      status =
-          IngestWBWIAsMemtable(wbwi, {/*lower_bound=*/lb, /*upper_bound=*/ub},
-                               /*min_prep_log=*/log_ref, last_sequence,
-                               /*memtable_updated=*/memtable_update_count > 0,
-                               write_options.ignore_missing_column_families);
+      status = IngestWBWIAsMemtable(
+          wbwi, {/*lower_bound=*/lb, /*upper_bound=*/ub},
+          /*min_prep_log=*/log_ref, last_sequence,
+          /*memtable_updated=*/memtable_update_count > 0,
+          ingest_wbwi_for_commit, write_options.ignore_missing_column_families);
       RecordTick(stats_, NUMBER_WBWI_INGEST);
     }
   }
@@ -2884,6 +2900,7 @@ Status DBImpl::SwitchMemtable(ColumnFamilyData* cfd, WriteContext* context,
     // of mutable_cf_options.write_buffer_size.
     io_s = CreateWAL(write_options, new_log_number, recycle_log_number,
                      preallocate_block_size, info, &new_log);
+    TEST_SYNC_POINT_CALLBACK("DBImpl::SwitchMemtable:AfterCreateWAL", &io_s);
     if (s.ok()) {
       s = io_s;
     }

db/event_helpers.cc

@@ -259,6 +259,7 @@ void EventHelpers::NotifyOnErrorRecoveryEnd(
     db_mutex->Lock();
   } else {
     old_bg_error.PermitUncheckedError();
+    new_bg_error.PermitUncheckedError();
   }
 }
 

utilities/transactions/transaction_test.cc

@@ -9636,6 +9636,63 @@ TEST_P(CommitBypassMemtableTest, AtomicFlushTest) {
   }
 }
 
+TEST_P(CommitBypassMemtableTest, SwitchMemtableFailureStopsDBUntilReopen) {
+  SetUpTransactionDB(/*atomic_flush=*/false);
+
+  auto* db_impl = static_cast_with_check<DBImpl>(txn_db->GetRootDB());
+
+  WriteOptions wopts;
+  wopts.sync = true;
+  TransactionOptions txn_opts;
+  txn_opts.commit_bypass_memtable = true;
+  std::unique_ptr<Transaction> txn{
+      txn_db->BeginTransaction(wopts, txn_opts, nullptr)};
+  ASSERT_NE(txn, nullptr);
+  ASSERT_OK(txn->SetName("xid_switch_memtable_failure"));
+  ASSERT_OK(txn->Put("k1", "v1"));
+  ASSERT_OK(txn->Put("k2", "v2"));
+  ASSERT_OK(txn->Prepare());
+
+  IOStatus injected_error =
+      IOStatus::IOError("Injected WAL creation failure in SwitchMemtable");
+  injected_error.SetRetryable(true);
+  bool failure_injected = false;
+  SyncPoint::GetInstance()->SetCallBack(
+      "DBImpl::SwitchMemtable:AfterCreateWAL", [&](void* arg) {
+        if (failure_injected) {
+          return;
+        }
+        *static_cast<IOStatus*>(arg) = injected_error;
+        failure_injected = true;
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  Status commit_status = txn->Commit();
+  ASSERT_TRUE(commit_status.IsCorruption()) << commit_status.ToString();
+  ASSERT_TRUE(failure_injected);
+
+  txn.reset();
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  Status bg_error = db_impl->TEST_GetBGError();
+  ASSERT_TRUE(bg_error.IsCorruption()) << bg_error.ToString();
+  ASSERT_EQ(bg_error.severity(), Status::Severity::kFatalError);
+
+  Status put_status = txn_db->Put(wopts, "k3", "v3");
+  ASSERT_TRUE(put_status.IsCorruption()) << put_status.ToString();
+  ASSERT_EQ(put_status.severity(), Status::Severity::kFatalError);
+
+  ASSERT_OK(txn_db->Close());
+  db_.reset();  // destroys txn_db (owned by db_)
+  ASSERT_OK(TransactionDB::Open(options, txn_db_opts, dbname_, &txn_db));
+  db_.reset(txn_db);
+
+  VerifyDBFromMap({{"k1", "v1"}, {"k2", "v2"}});
+  ASSERT_OK(txn_db->Put(wopts, "k3", "v3"));
+}
+
 TEST_P(CommitBypassMemtableTest, MergeAndMultiCF) {
   // disable_flush allows testing Get path with memtables.
   for (bool disable_flush : {false, true}) {