Fix subset checkpoints with incomplete MANIFEST tails (#15123)

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

Subset checkpoint post-processing appends column-family drop records to a copied MANIFEST. Recovery can tolerate an incomplete atomic group at the tail while retaining the reader high-water size, so appending at that size can produce a checkpoint that fails recovery.

Track the valid logical append boundary together with its MANIFEST file number, refresh it after recovery and successful writes, reset it across reactive MANIFEST switches, and trim only subset-checkpoint copies before appending drop records. Whole-DB checkpoints and the source MANIFEST remain unchanged.

Reviewed By: anand1976

Differential Revision: D116449394

fbshipit-source-id: a162dd6a2558f0f9ec40d878ee83956a89df698b

db/db_filesnapshot.cc

@@ -415,7 +415,14 @@ Status DBImpl::GetLiveFilesStorageInfoImpl(
 
   // Capture some final info before releasing mutex
   const uint64_t manifest_number = versions_->manifest_file_number();
-  const uint64_t manifest_size = versions_->manifest_file_size();
+  uint64_t manifest_size = versions_->manifest_file_size();
+  if (cf_subset && !excluded_cf_ids->empty()) {
+    Status manifest_s = versions_->GetManifestAppendBoundary(&manifest_size);
+    if (!manifest_s.ok()) {
+      mutex_.Unlock();
+      return manifest_s;
+    }
+  }
   const uint64_t options_number = versions_->options_file_number();
   const uint64_t options_size = versions_->options_file_size_;
   const uint64_t min_log_num = MinLogNumberToKeep();

db/version_edit_handler.cc

@@ -464,6 +464,9 @@ void VersionEditHandler::CheckIterationResult(const log::Reader& reader,
     assert(version_set_->manifest_file_size_ > 0);
     version_set_->manifest_recovery_last_valid_record_end_ =
         last_valid_record_end_;
+    version_set_->manifest_last_valid_record_end_ = last_valid_record_end_;
+    version_set_->manifest_last_valid_record_end_file_number_ =
+        version_set_->manifest_file_number_;
     version_set_->next_file_number_.store(version_edit_params_.GetNextFile() +
                                           1);
     SequenceNumber last_seq = version_edit_params_.GetLastSequence();

db/version_edit_handler.h

@@ -48,7 +48,10 @@ class VersionEditHandlerBase {
   virtual void CheckIterationResult(const log::Reader& /*reader*/,
                                     Status* /*s*/) {}
 
-  void ClearReadBuffer() { read_buffer_.Clear(); }
+  void ResetReadState() {
+    read_buffer_.Clear();
+    last_valid_record_end_ = 0;
+  }
 
   Status status_;
 
@@ -425,7 +428,7 @@ class ManifestTailer : public VersionEditHandlerPointInTime {
 
   void PrepareToReadNewManifest() {
     initialized_ = false;
-    ClearReadBuffer();
+    ResetReadState();
   }
 
   std::unordered_set<ColumnFamilyData*>& GetUpdatedColumnFamilies() {

db/version_set.cc

@@ -5376,6 +5376,8 @@ VersionSet::VersionSet(
       current_version_number_(0),
       manifest_file_size_(0),
       manifest_recovery_last_valid_record_end_(0),
+      manifest_last_valid_record_end_(0),
+      manifest_last_valid_record_end_file_number_(0),
       force_new_manifest_on_open_(false),
       last_compacted_manifest_file_size_(0),
       file_options_(storage_options),
@@ -5587,6 +5589,8 @@ void VersionSet::Reset() {
   manifest_writers_.clear();
   manifest_file_size_ = 0;
   manifest_recovery_last_valid_record_end_ = 0;
+  manifest_last_valid_record_end_ = 0;
+  manifest_last_valid_record_end_file_number_ = 0;
   force_new_manifest_on_open_ = false;
   last_compacted_manifest_file_size_ = 0;
   TuneMaxManifestFileSize();
@@ -6200,6 +6204,8 @@ Status VersionSet::ProcessManifestWrites(
       descriptor_last_sequence_ = max_last_sequence;
       manifest_file_number_ = pending_manifest_file_number_;
       manifest_file_size_ = new_manifest_file_size;
+      manifest_last_valid_record_end_ = new_manifest_file_size;
+      manifest_last_valid_record_end_file_number_ = manifest_file_number_;
       prev_log_number_ = first_writer.edit_list.front()->GetPrevLogNumber();
     }
   } else {
@@ -6492,9 +6498,24 @@ std::unique_ptr<log::Writer> VersionSet::CreateManifestWriter(
       /*track_and_verify_wals=*/false, block_offset);
 }
 
+Status VersionSet::GetManifestAppendBoundary(uint64_t* manifest_size) const {
+  assert(manifest_size != nullptr);
+  if (manifest_last_valid_record_end_file_number_ != manifest_file_number_) {
+    return Status::TryAgain(
+        "Current MANIFEST changed while determining its append boundary");
+  }
+  if (manifest_last_valid_record_end_ == 0 ||
+      manifest_last_valid_record_end_ > manifest_file_size_) {
+    return Status::Corruption("Invalid MANIFEST append boundary");
+  }
+  *manifest_size = manifest_last_valid_record_end_;
+  return Status::OK();
+}
+
 Status VersionSet::ReopenManifestForAppend(const std::string& manifest_path) {
   assert(db_options_->reuse_manifest_on_open);
   assert(manifest_recovery_last_valid_record_end_ > 0);
+  assert(manifest_last_valid_record_end_file_number_ == manifest_file_number_);
 
   // Disabled under best_efforts_recovery: that mode rebuilds the
   // MANIFEST + CURRENT from scratch via the side-effect of

db/version_set.h

@@ -1619,17 +1619,23 @@ class VersionSet {
   // in the manifest?" (e.g. the DB::OpenAndCompact floor vs. a
   // manifest_file_size captured on the primary). It is maximal, so a
   // fully-recovered prefix never tests short; it excludes garbage, so a corrupt
-  // tail never tests long. It is not a safe append/truncate offset -- use
-  // manifest_recovery_last_valid_record_end_.
+  // tail never tests long. It is not a safe append/truncate offset; use
+  // GetManifestAppendBoundary() for that.
   //
-  // Today this == manifest_recovery_last_valid_record_end_ == a clean
-  // manifest_file_size_; a future padded/footered format may make
-  // last_valid_record_end_ <= manifest_file_size_ <= this (changing only this
-  // accessor's body, not callers).
+  // Immediately after recovery, this ==
+  // manifest_recovery_last_valid_record_end_ == a clean manifest_file_size_;
+  // a future padded/footered format may make manifest_last_valid_record_end_
+  // <= manifest_file_size_ <= this (changing only this accessor's body, not
+  // callers).
   uint64_t manifest_recovery_maximal_valid_size() const {
     return manifest_recovery_last_valid_record_end_;
   }
 
+  // Returns the valid prefix length at which records can be appended to a
+  // copy of the current MANIFEST.
+  // REQUIRES: DB mutex held, or the DB is not yet visible to other threads.
+  Status GetManifestAppendBoundary(uint64_t* manifest_size) const;
+
   Status GetMetadataForFile(uint64_t number, int* filelevel,
                             FileMetaData** metadata, ColumnFamilyData** cfd);
 
@@ -1883,10 +1889,18 @@ class VersionSet {
   // (Close), and backup metadata.
   uint64_t manifest_recovery_last_valid_record_end_;
 
+  // Safe append boundary for a copy of the current MANIFEST. Initialized from
+  // the recovery boundary and advanced after every successful MANIFEST write.
+  // Unlike manifest_file_size_, it excludes any tolerated tail garbage.
+  uint64_t manifest_last_valid_record_end_;
+
+  // MANIFEST file number associated with manifest_last_valid_record_end_.
+  uint64_t manifest_last_valid_record_end_file_number_;
+
   // True when reuse_manifest_on_open was requested but the recovered MANIFEST
   // could not be safely reopened for append. DB open must install a fresh
-  // MANIFEST before returning so other MANIFEST append paths never see the old
-  // tail.
+  // MANIFEST before returning so normal writable operation never appends to
+  // the old tail.
   bool force_new_manifest_on_open_;
 
   // Size of the populated manifest file last time it was re-written from

db/version_set_test.cc

@@ -3004,6 +3004,70 @@ TEST_F(VersionSetAtomicGroupTest,
   EXPECT_EQ(kAtomicGroupSize, num_recovered_edits_);
 }
 
+TEST_F(VersionSetAtomicGroupTest,
+       ReactiveVersionSetResetsManifestAppendBoundaryOnSwitch) {
+  CreateCurrentFile();
+  std::unique_ptr<log::FragmentBufferedReader> manifest_reader;
+  std::unique_ptr<log::Reader::Reporter> manifest_reporter;
+  std::unique_ptr<Status> manifest_reader_status;
+  ASSERT_OK(reactive_versions_->Recover(column_families_, &manifest_reader,
+                                        &manifest_reporter,
+                                        &manifest_reader_status));
+
+  uint64_t old_manifest_end = 0;
+  ASSERT_OK(reactive_versions_->GetManifestAppendBoundary(&old_manifest_end));
+  ASSERT_GT(old_manifest_end, 0);
+  log_writer_.reset();
+
+  constexpr uint64_t kNewManifestFileNumber = 2;
+  const std::string new_manifest_path =
+      DescriptorFileName(dbname_, kNewManifestFileNumber);
+  std::unique_ptr<WritableFileWriter> file_writer;
+  ASSERT_OK(WritableFileWriter::Create(
+      fs_, new_manifest_path, fs_->OptimizeForManifestWrite(env_options_),
+      &file_writer, nullptr));
+  log::Writer new_manifest_writer(std::move(file_writer), /*log_number=*/0,
+                                  /*recycle_log_files=*/false);
+
+  edits_.emplace_back();
+  VersionEdit& incomplete_edit = edits_.back();
+  incomplete_edit.SetLogNumber(0);
+  incomplete_edit.SetNextFile(kNewManifestFileNumber + 1);
+  incomplete_edit.SetLastSequence(last_seqno_++);
+  incomplete_edit.SetDBId(std::string(2 * log::kBlockSize, 'x'));
+  incomplete_edit.MarkAtomicGroup(1);
+  std::string record;
+  ASSERT_TRUE(incomplete_edit.EncodeTo(&record, 0));
+  ASSERT_OK(new_manifest_writer.AddRecord(WriteOptions(), record));
+  ASSERT_OK(new_manifest_writer.Close(WriteOptions()));
+
+  uint64_t new_manifest_size = 0;
+  ASSERT_OK(fs_->GetFileSize(new_manifest_path, IOOptions(), &new_manifest_size,
+                             nullptr));
+  ASSERT_GE(new_manifest_size, old_manifest_end);
+  ASSERT_OK(SetCurrentFile(WriteOptions(), fs_.get(), dbname_,
+                           kNewManifestFileNumber, Temperature::kUnknown,
+                           /*dir_contains_current_file=*/nullptr));
+
+  InstrumentedMutex mu;
+  std::unordered_set<ColumnFamilyData*> cfds_changed;
+  uint64_t new_manifest_end = 0;
+  Status boundary_status;
+  mu.Lock();
+  Status read_status = reactive_versions_->ReadAndApply(
+      &mu, &manifest_reader, manifest_reader_status.get(), &cfds_changed,
+      /*files_to_delete=*/nullptr);
+  if (read_status.ok()) {
+    boundary_status =
+        reactive_versions_->GetManifestAppendBoundary(&new_manifest_end);
+  }
+  mu.Unlock();
+
+  ASSERT_OK(read_status);
+  ASSERT_EQ(1, reactive_versions_->TEST_read_edits_in_atomic_group());
+  ASSERT_TRUE(boundary_status.IsCorruption()) << boundary_status.ToString();
+}
+
 TEST_F(VersionSetAtomicGroupTest,
        HandleIncompleteTrailingAtomicGroupWithVersionSetRecover) {
   const int kAtomicGroupSize = 4;

unreleased_history/bug_fixes/subset_checkpoint_manifest_tail.md

@@ -0,0 +1 @@
+Fixed subset checkpoints to discard tolerated MANIFEST tail corruption before appending column-family drop records.

utilities/checkpoint/checkpoint_test.cc

@@ -16,11 +16,16 @@
 #include <atomic>
 #include <cstdlib>
 #include <iostream>
+#include <limits>
 #include <thread>
 #include <utility>
 
 #include "db/db_impl/db_impl.h"
+#include "db/log_writer.h"
+#include "db/manifest_ops.h"
+#include "db/version_edit.h"
 #include "file/file_util.h"
+#include "file/writable_file_writer.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
 #include "rocksdb/db.h"
@@ -560,6 +565,42 @@ namespace {
 // layout, with handles in that order.
 constexpr size_t kNumSubsetCheckpointCfs = 4;
 
+void AppendIncompleteAtomicGroupToManifest(Env* env,
+                                           const std::string& manifest_path) {
+  uint64_t file_size;
+  ASSERT_OK(env->GetFileSize(manifest_path, &file_size));
+  const std::shared_ptr<FileSystem>& fs = env->GetFileSystem();
+  std::unique_ptr<FSWritableFile> fs_file;
+  ASSERT_OK(fs->ReopenWritableFile(manifest_path, FileOptions(), &fs_file,
+                                   /*dbg=*/nullptr));
+  std::unique_ptr<WritableFileWriter> file_writer(
+      new WritableFileWriter(std::move(fs_file), manifest_path, FileOptions()));
+  log::Writer log_writer(std::move(file_writer), /*log_number=*/0,
+                         /*recycle_log_files=*/false,
+                         /*manual_flush=*/false, kNoCompression,
+                         /*track_and_verify_wals=*/false,
+                         file_size % log::kBlockSize);
+
+  VersionEdit first_edit;
+  first_edit.SetLogNumber(0);
+  first_edit.SetNextFile(100);
+  first_edit.SetLastSequence(100);
+  first_edit.MarkAtomicGroup(2);
+  std::string first_record;
+  ASSERT_TRUE(first_edit.EncodeTo(&first_record, 0));
+  ASSERT_OK(log_writer.AddRecord(WriteOptions(), first_record));
+
+  VersionEdit second_edit;
+  second_edit.SetLogNumber(0);
+  second_edit.SetNextFile(100);
+  second_edit.SetLastSequence(100);
+  second_edit.MarkAtomicGroup(1);
+  std::string second_record;
+  ASSERT_TRUE(second_edit.EncodeTo(&second_record, 0));
+  ASSERT_OK(log_writer.AddRecord(WriteOptions(), second_record));
+  // Omit the final remaining_entries=0 record, leaving the group incomplete.
+}
+
 // Writes one key per column family and flushes, so each family owns exactly one
 // SST file and the families excluded from a subset checkpoint are visible as
 // missing SSTs.
@@ -676,6 +717,78 @@ TEST_F(CheckpointTest, CheckpointSubsetOfCF) {
   VerifyDefaultAndTwoSubsetCheckpoint(env_, snapshot_name_, options);
 }
 
+TEST_F(CheckpointTest,
+       CheckpointSubsetIncompleteAtomicGroupAfterWritableRecovery) {
+  Options options = CurrentOptions();
+  options.optimize_manifest_for_recovery = true;
+  options.reuse_manifest_on_open = false;
+  options.track_and_verify_wals_in_manifest = true;
+  options.avoid_flush_during_recovery = true;
+  options.write_dbid_to_manifest = false;
+  CreateAndReopenWithCF({"one", "two", "three"}, options);
+  PopulateFourCfsForSubsetCheckpoint(db_.get(), handles_);
+  Close();
+
+  std::string manifest_path;
+  uint64_t manifest_file_number = 0;
+  ASSERT_OK(GetCurrentManifestPath(dbname_, env_->GetFileSystem().get(),
+                                   /*is_retry=*/false, &manifest_path,
+                                   &manifest_file_number));
+  AppendIncompleteAtomicGroupToManifest(env_, manifest_path);
+
+  ReopenWithColumnFamilies({kDefaultColumnFamilyName, "one", "two", "three"},
+                           options);
+  std::string manifest_path_after_reopen;
+  ASSERT_OK(GetCurrentManifestPath(
+      dbname_, env_->GetFileSystem().get(), /*is_retry=*/false,
+      &manifest_path_after_reopen, &manifest_file_number));
+  ASSERT_EQ(manifest_path, manifest_path_after_reopen);
+
+  Checkpoint* checkpoint;
+  ASSERT_OK(Checkpoint::Create(db_.get(), &checkpoint));
+  std::unique_ptr<Checkpoint> uchk(checkpoint);
+  ASSERT_OK(uchk->CreateCheckpoint(snapshot_name_, {handles_[0], handles_[2]},
+                                   std::numeric_limits<uint64_t>::max()));
+
+  VerifyDefaultAndTwoSubsetCheckpoint(env_, snapshot_name_, options);
+}
+
+TEST_F(CheckpointTest,
+       CheckpointSubsetIncompleteAtomicGroupAfterReadOnlyRecovery) {
+  Options options = CurrentOptions();
+  options.optimize_manifest_for_recovery = true;
+  options.reuse_manifest_on_open = false;
+  options.track_and_verify_wals_in_manifest = true;
+  options.avoid_flush_during_recovery = true;
+  options.write_dbid_to_manifest = false;
+  CreateAndReopenWithCF({"one", "two", "three"}, options);
+  PopulateFourCfsForSubsetCheckpoint(db_.get(), handles_);
+  Close();
+
+  std::string manifest_path;
+  uint64_t manifest_file_number = 0;
+  ASSERT_OK(GetCurrentManifestPath(dbname_, env_->GetFileSystem().get(),
+                                   /*is_retry=*/false, &manifest_path,
+                                   &manifest_file_number));
+  AppendIncompleteAtomicGroupToManifest(env_, manifest_path);
+
+  ASSERT_OK(ReadOnlyReopenWithColumnFamilies(
+      {kDefaultColumnFamilyName, "one", "two", "three"}, options));
+  std::string manifest_path_after_reopen;
+  ASSERT_OK(GetCurrentManifestPath(
+      dbname_, env_->GetFileSystem().get(), /*is_retry=*/false,
+      &manifest_path_after_reopen, &manifest_file_number));
+  ASSERT_EQ(manifest_path, manifest_path_after_reopen);
+
+  Checkpoint* checkpoint;
+  ASSERT_OK(Checkpoint::Create(db_.get(), &checkpoint));
+  std::unique_ptr<Checkpoint> uchk(checkpoint);
+  ASSERT_OK(uchk->CreateCheckpoint(snapshot_name_, {handles_[0], handles_[2]},
+                                   std::numeric_limits<uint64_t>::max()));
+
+  VerifyDefaultAndTwoSubsetCheckpoint(env_, snapshot_name_, options);
+}
+
 TEST_F(CheckpointTest, CheckpointSubsetEmptyListEqualsAll) {
   Options options = CurrentOptions();
   CreateAndReopenWithCF({"one", "two"}, options);