Serialize subset-checkpoint OPTIONS persistence with SetOptions (#15386)

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

Fixes crash-test failure T292562874:
`Checkpoint failed: Corruption: [RocksDBOptionsParser]:failed the verification on BlockBasedTable::: filter_policy.id`

Root cause:
- D123104845 ("Commit authoritative OPTIONS file numbers in MANIFEST", 77d94c989243) made subset checkpoints (`CreateCheckpoint` with `include_cf_ids` that exclude some CFs) write a new OPTIONS file. `DBImpl::GetLiveFilesStorageInfoImpl` captures the snapshot under the DB mutex with `cfd->GetLatestCFOptions()`. That is a shallow copy: `table_factory`, and through it `filter_policy`, are `shared_ptr`s to the live objects. `DBImpl::CreateOptionsFileForSubsetCheckpoint` then calls `PersistRocksDBOptions` without holding `options_mutex_`. `PersistRocksDBOptions` serializes the file, then re-parses it and checks it against the same in-memory options.
- `RibbonFilterPolicy::bloom_before_level_` is a `kMutable` atomic that `SetOptions(table_factory.filter_policy.bloom_before_level=...)` changes in place, and `RibbonFilterPolicy::GetId()` includes it (`ribbonfilter:<bits>:<bloom_before_level>`). A concurrent `SetOptions` between the write and the verify makes the in-memory `filter_policy.id` differ from the file, so verification returns Corruption and the checkpoint fails.
- Every other OPTIONS writer (`WriteOptionsFile`, CF create/drop) holds `options_mutex_`. `SetOptions` holds it for its whole mutate-and-persist sequence. The new subset-checkpoint writer was the only one that skipped it.

Trigger in the crash test: db_stress randomly sets `table_factory.filter_policy.bloom_before_level` through `SetOptions` (when `keepRibbonFilterPolicyOnly`, i.e. `bloom_before_level` != INT_MAX with ribbon filters; see the OPTIONS dump with `ribbonfilter:18:1`, `ribbonfilter:8.9:2147483647`, `ribbonfilter:5.67:-1` across CFs). It also takes subset checkpoints. The two run concurrently on different threads.

Fix: hold `options_mutex_` in `CreateOptionsFileForSubsetCheckpoint` while persisting and verifying, the same as `WriteOptionsFile`. The only caller is `CheckpointImpl::CreateCheckpoint`, which holds no DB locks at that point, so lock ordering (options_mutex_ before mutex_) is kept.

Note: the snapshot's scalar fields are still captured at live-file time. Live pointer sub-options such as the filter policy reflect their state at persist time. This was already true of the shallow snapshot. The change only makes the persisted file self-consistent with what is verified.

Reviewed By: xingbowang

Differential Revision: D124728996

fbshipit-source-id: adb5511c2ddd73ab138fd7fd21109199ff4f5b1e

db/db_filesnapshot.cc

@@ -221,6 +221,12 @@ Status DBImpl::CreateOptionsFileForSubsetCheckpoint(
     return Status::OK();
   }
 
+  // The snapshot's CF options are shallow copies that share live pointer
+  // sub-options (e.g. RibbonFilterPolicy, whose bloom_before_level is mutated
+  // in place by SetOptions). Hold options_mutex_, as WriteOptionsFile does, so
+  // SetOptions cannot change them between writing the file and verifying it.
+  InstrumentedMutexLock ol(&options_mutex_);
+
   const std::string options_path =
       OptionsFileName(checkpoint_dir, options_snapshot.options_file_number);
   return PersistRocksDBOptions(

utilities/checkpoint/checkpoint_test.cc

@@ -14,6 +14,7 @@
 #include <unistd.h>
 #endif
 #include <atomic>
+#include <chrono>
 #include <cstdlib>
 #include <iostream>
 #include <limits>
@@ -31,10 +32,12 @@
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
 #include "rocksdb/file_system.h"
+#include "rocksdb/filter_policy.h"
 #include "rocksdb/metadata.h"
 #include "rocksdb/rocksdb_namespace.h"
 #include "rocksdb/sst_file_manager.h"
 #include "rocksdb/statistics.h"
+#include "rocksdb/table.h"
 #include "rocksdb/utilities/backup_engine.h"
 #include "rocksdb/utilities/options_util.h"
 #include "rocksdb/utilities/transaction_db.h"
@@ -780,6 +783,60 @@ TEST_F(CheckpointTest, SubsetOptionsAndManifestUseSameMembershipSnapshot) {
   VerifyDefaultAndTwoSubsetCheckpoint(env_, snapshot_name_, options);
 }
 
+TEST_F(CheckpointTest, SubsetOptionsFileRacesWithMutableFilterPolicy) {
+  // RibbonFilterPolicy::bloom_before_level is mutated in place by SetOptions,
+  // and the subset checkpoint's options snapshot shares the live policy
+  // object. Persisting that snapshot must not interleave with SetOptions.
+  Options options = CurrentOptions();
+  BlockBasedTableOptions table_options;
+  table_options.filter_policy.reset(
+      NewRibbonFilterPolicy(10, /*bloom_before_level=*/1));
+  options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+  CreateAndReopenWithCF({"one", "two", "three"}, options);
+  PopulateFourCfsForSubsetCheckpoint(db_.get(), handles_);
+
+  const std::thread::id checkpoint_thread_id = std::this_thread::get_id();
+  std::atomic<bool> injected{false};
+  std::atomic<bool> set_options_done{false};
+  Status set_options_status;
+  std::thread set_options_thread;
+  SyncPoint::GetInstance()->SetCallBack(
+      "PersistRocksDBOptions:written", [&](void*) {
+        if (std::this_thread::get_id() != checkpoint_thread_id ||
+            injected.exchange(true)) {
+          return;
+        }
+        set_options_thread = std::thread([&]() {
+          set_options_status = db_->SetOptions(
+              handles_[0],
+              {{"table_factory.filter_policy.bloom_before_level", "2"}});
+          set_options_done.store(true);
+        });
+        // Give an unserialized SetOptions the chance to land between writing
+        // the checkpoint OPTIONS file and verifying it.
+        for (int i = 0; i < 100 && !set_options_done.load(); ++i) {
+          std::this_thread::sleep_for(std::chrono::milliseconds(10));
+        }
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+  Defer sync_point_cleanup([]() {
+    SyncPoint::GetInstance()->DisableProcessing();
+    SyncPoint::GetInstance()->ClearAllCallBacks();
+  });
+
+  Checkpoint* checkpoint;
+  ASSERT_OK(Checkpoint::Create(db_.get(), &checkpoint));
+  std::unique_ptr<Checkpoint> uchk(checkpoint);
+  Status checkpoint_status =
+      uchk->CreateCheckpoint(snapshot_name_, {handles_[0], handles_[2]});
+  if (set_options_thread.joinable()) {
+    set_options_thread.join();
+  }
+  ASSERT_TRUE(injected.load());
+  ASSERT_OK(set_options_status);
+  ASSERT_OK(checkpoint_status);
+}
+
 TEST_F(CheckpointTest,
        CheckpointSubsetIncompleteAtomicGroupAfterWritableRecovery) {
   Options options = CurrentOptions();