DB::GetSortedWalFiles() to ensure file deletion is disabled (#8591)

Summary:
If DB::GetSortedWalFiles() runs without file deletion disbled, file might get deleted in the middle and error is returned to users. It makes the function hard to use. Fix it by disabling file deletion if it is not done.

Fix another minor issue of logging within DB mutex, which should not be done unless a major failure happens.

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

Test Plan: Run all existing tests

Reviewed By: pdillinger

Differential Revision: D29969412

fbshipit-source-id: d5f42b5271608a35b9b07687ce18157d7447b0de

db/db_filesnapshot.cc

@@ -127,7 +127,22 @@ Status DBImpl::GetSortedWalFiles(VectorLogPtr& files) {
       bg_cv_.Wait();
     }
   }
-  return wal_manager_.GetSortedWalFiles(files);
+
+  // Disable deletion in order to avoid the case where a file is deleted in
+  // the middle of the process so IO error is returned.
+  Status s = DisableFileDeletions();
+  bool file_deletion_supported = !s.IsNotSupported();
+  if (s.ok() || !file_deletion_supported) {
+    s = wal_manager_.GetSortedWalFiles(files);
+    if (file_deletion_supported) {
+      Status s2 = EnableFileDeletions(false);
+      if (!s2.ok() && s.ok()) {
+        s = s2;
+      }
+    }
+  }
+
+  return s;
 }
 
 Status DBImpl::GetCurrentWalFile(std::unique_ptr<LogFile>* current_log_file) {

db/db_impl/db_impl_files.cc

@@ -38,20 +38,26 @@ uint64_t DBImpl::MinObsoleteSstNumberToKeep() {
 }
 
 Status DBImpl::DisableFileDeletions() {
+  Status s;
+  int my_disable_delete_obsolete_files;
+  {
     InstrumentedMutexLock l(&mutex_);
-  return DisableFileDeletionsWithLock();
-}
-
-Status DBImpl::DisableFileDeletionsWithLock() {
-  mutex_.AssertHeld();
-  ++disable_delete_obsolete_files_;
-  if (disable_delete_obsolete_files_ == 1) {
+    s = DisableFileDeletionsWithLock();
+    my_disable_delete_obsolete_files = disable_delete_obsolete_files_;
+  }
+  if (my_disable_delete_obsolete_files == 1) {
     ROCKS_LOG_INFO(immutable_db_options_.info_log, "File Deletions Disabled");
   } else {
     ROCKS_LOG_WARN(immutable_db_options_.info_log,
                    "File Deletions Disabled, but already disabled. Counter: %d",
-                   disable_delete_obsolete_files_);
+                   my_disable_delete_obsolete_files);
   }
+  return s;
+}
+
+Status DBImpl::DisableFileDeletionsWithLock() {
+  mutex_.AssertHeld();
+  ++disable_delete_obsolete_files_;
   return Status::OK();
 }
 

db/error_handler.cc

@@ -394,6 +394,7 @@ const Status& ErrorHandler::SetBGError(const IOStatus& bg_io_err,
   if (BackgroundErrorReason::kManifestWrite == reason ||
       BackgroundErrorReason::kManifestWriteNoWAL == reason) {
     // Always returns ok
+    ROCKS_LOG_INFO(db_options_.info_log, "Disabling File Deletions");
     db_->DisableFileDeletionsWithLock().PermitUncheckedError();
   }