Ensure Close() before LinkFile() for WALs in Checkpoint (#12734)

Summary:
POSIX semantics for LinkFile (hard links) allow linking a file
that is still being written two, with both the source and destination
showing any subsequent writes to the source. This may not be practical
semantics for some FileSystem implementations such as remote storage.
They might only link the flushed or sync-ed file contents at time of
LinkFile, or might even have undefined behavior if LinkFile is called on
a file still open for write (not yet "sealed"). This change builds on https://github.com/facebook/rocksdb/issues/12731
to bring more hygiene to our handling of WAL files in Checkpoint.

Specifically, we now Close WAL files as soon as they are either
(a) inactive and fully synced, or (b) inactive and obsolete (so maybe
never fully synced), rather than letting Close() happen in handling
obsolete files (maybe a background thread). This should not be a
performance issue as Close() should be trivial cost relative to other
IO ops, but just in case:
* We don't Close() while holding a mutex, to avoid blocking, and
* The old behavior is available with a new kill switch option
  `background_close_inactive_wals`.

Stacked on https://github.com/facebook/rocksdb/issues/12731

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

Test Plan:
Extended existing unit test, especially adding a hygiene
check to FaultInjectionTestFS to detect LinkFile() on a file still open
for writes. FaultInjectionTestFS already has relevant tracking data, and
tests can opt out of the new check, as in a smoke test I have left for
the old, deprecated functionality `background_close_inactive_wals=true`.

Also ran lengthy blackbox_crash_test to ensure the hygiene check is OK
with the crash test. (The only place I can find we use LinkFile in
production is Checkpoint.)

Reviewed By: cbi42

Differential Revision: D58295284

Pulled By: pdillinger

fbshipit-source-id: 64d90ed8477e2366c19eaf9c4c5ad60b82cac5c6

3 files changed, 33 insertions, 5 deletions

diff --git a/utilities/checkpoint/checkpoint_test.cc b/utilities/checkpoint/checkpoint_test.cc
index 111e81f32..3b867bfa6 100644
--- a/utilities/checkpoint/checkpoint_test.cc
+++ b/utilities/checkpoint/checkpoint_test.cc
@@ -762,27 +762,38 @@ TEST_F(CheckpointTest, CheckpointWithParallelWrites) {
 
 class CheckpointTestWithWalParams
     : public CheckpointTest,
-      public testing::WithParamInterface<std::tuple<uint64_t, bool, bool>> {
+      public testing::WithParamInterface<
+          std::tuple<uint64_t, bool, bool, bool>> {
  public:
   uint64_t GetLogSizeForFlush() { return std::get<0>(GetParam()); }
   bool GetWalsInManifest() { return std::get<1>(GetParam()); }
   bool GetManualWalFlush() { return std::get<2>(GetParam()); }
+  bool GetBackgroundCloseInactiveWals() { return std::get<3>(GetParam()); }
 };
 
-INSTANTIATE_TEST_CASE_P(CheckpointTestWithWalParams,
-                        CheckpointTestWithWalParams,
+INSTANTIATE_TEST_CASE_P(NormalWalParams, CheckpointTestWithWalParams,
                         ::testing::Combine(::testing::Values(0U, 100000000U),
-                                           ::testing::Bool(),
-                                           ::testing::Bool()));
+                                           ::testing::Bool(), ::testing::Bool(),
+                                           ::testing::Values(false)));
+
+INSTANTIATE_TEST_CASE_P(DeprecatedWalParams, CheckpointTestWithWalParams,
+                        ::testing::Values(std::make_tuple(100000000U, true,
+                                                          false, true)));
 
 TEST_P(CheckpointTestWithWalParams, CheckpointWithUnsyncedDataDropped) {
   Options options = CurrentOptions();
   options.max_write_buffer_number = 4;
   options.track_and_verify_wals_in_manifest = GetWalsInManifest();
   options.manual_wal_flush = GetManualWalFlush();
+  options.background_close_inactive_wals = GetBackgroundCloseInactiveWals();
   auto fault_fs = std::make_shared<FaultInjectionTestFS>(FileSystem::Default());
   std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
 
+  if (options.background_close_inactive_wals) {
+    // Disable this hygiene check when the fix is disabled
+    fault_fs->SetAllowLinkOpenFile();
+  }
+
   options.env = fault_fs_env.get();
   Reopen(options);
   ASSERT_OK(Put("key1", "val1"));
diff --git a/utilities/fault_injection_fs.cc b/utilities/fault_injection_fs.cc
index 4244b0a4e..1cdc1256f 100644
--- a/utilities/fault_injection_fs.cc
+++ b/utilities/fault_injection_fs.cc
@@ -17,6 +17,7 @@
 #include "utilities/fault_injection_fs.h"
 
 #include <algorithm>
+#include <cstdio>
 #include <functional>
 #include <utility>
 
@@ -886,6 +887,12 @@ IOStatus FaultInjectionTestFS::LinkFile(const std::string& s,
   if (io_s.ok()) {
     {
       MutexLock l(&mutex_);
+      if (!allow_link_open_file_ &&
+          open_managed_files_.find(s) != open_managed_files_.end()) {
+        fprintf(stderr, "Attempt to LinkFile while open for write: %s\n",
+                s.c_str());
+        abort();
+      }
       if (db_file_state_.find(s) != db_file_state_.end()) {
         db_file_state_[t] = db_file_state_[s];
       }
diff --git a/utilities/fault_injection_fs.h b/utilities/fault_injection_fs.h
index b49a69d2c..ea76bf538 100644
--- a/utilities/fault_injection_fs.h
+++ b/utilities/fault_injection_fs.h
@@ -203,6 +203,7 @@ class FaultInjectionTestFS : public FileSystemWrapper {
         filesystem_active_(true),
         filesystem_writable_(false),
         read_unsynced_data_(true),
+        allow_link_open_file_(false),
         thread_local_error_(new ThreadLocalPtr(DeleteThreadLocalErrorContext)),
         enable_write_error_injection_(false),
         enable_metadata_write_error_injection_(false),
@@ -365,6 +366,14 @@ class FaultInjectionTestFS : public FileSystemWrapper {
     read_unsynced_data_ = read_unsynced_data;
   }
   bool ReadUnsyncedData() const { return read_unsynced_data_; }
+
+  // FaultInjectionTestFS normally includes a hygiene check for FileSystem
+  // implementations that only support LinkFile() on closed files (not open
+  // for write). Setting this to true bypasses the check.
+  void SetAllowLinkOpenFile(bool allow_link_open_file = true) {
+    allow_link_open_file_ = allow_link_open_file;
+  }
+
   void AssertNoOpenFile() { assert(open_managed_files_.empty()); }
 
   IOStatus GetError() { return error_; }
@@ -565,6 +574,7 @@ class FaultInjectionTestFS : public FileSystemWrapper {
   bool filesystem_writable_;  // Bypass FaultInjectionTestFS and go directly
                               // to underlying FS for writable files
   bool read_unsynced_data_;   // See SetReadUnsyncedData()
+  bool allow_link_open_file_;  // See SetAllowLinkOpenFile()
   IOStatus error_;
 
   enum ErrorType : int {