Sync recovery SST directory before reused MANIFEST append (#14780)

Summary:
- When `reuse_manifest_on_open` reuses the current MANIFEST, `DB::Open` recovery can flush WAL data into a new L0 SST and append the corresponding `VersionEdit` to that already-current MANIFEST.
- If open later fails and the process crashes, the MANIFEST edit can be durable while the recovered SST directory entry is not, leaving the DB pointing at a missing SST.
- Fsync the recovered SST's data directory before adding the file to the recovery edit when appending to a reused MANIFEST.
- Add a regression test that injects failure after MANIFEST sync, simulates crash cleanup of files created after the last directory sync, and verifies the recovered key remains readable.

## Task
- T272584339

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

Test Plan: CI

Reviewed By: anand1976

Differential Revision: D106201774

Pulled By: xingbowang

fbshipit-source-id: a44a7d1263d5bc1d82b995c90eef1a825eab4182

db/db_basic_test.cc

@@ -761,6 +761,111 @@ TEST_F(DBBasicTest, ReuseManifestOnOpenFullStackComposition) {
   EXPECT_EQ("v", Get("k"));
 }
 
+// Regression test for WAL recovery while publishing a fresh MANIFEST. The test
+// stores SSTs in a separate DB path, injects failure after CURRENT points at
+// the new MANIFEST, and simulates crash cleanup; the recovered SST must survive
+// because the synced MANIFEST references it.
+TEST_F(DBBasicTest, RecoverySstDirSyncedBeforeFreshManifestPublish) {
+  auto fault_fs = std::make_shared<FaultInjectionTestFS>(env_->GetFileSystem());
+  std::unique_ptr<Env> fault_env(NewCompositeEnv(fault_fs));
+
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.disable_auto_compactions = true;
+  options.env = fault_env.get();
+  options.reuse_manifest_on_open = false;
+  options.db_paths.emplace_back(dbname_ + "_2", 1ULL << 30);
+
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("base", "value"));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(Put("recovered", "value"));
+  ASSERT_OK(db_->FlushWAL(true));
+  Close();
+
+  fault_fs->ResetState();
+
+  std::atomic<bool> fail_after_current_publish{true};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ProcessManifestWrites:AfterSetCurrentFile", [&](void* arg) {
+        if (fail_after_current_publish.exchange(false)) {
+          ASSERT_NE(nullptr, arg);
+          IOStatus* io_s = static_cast<IOStatus*>(arg);
+          ASSERT_OK(*io_s);
+          *io_s = IOStatus::IOError("injected current publish aftermath");
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Status s = TryReopen(options);
+  ASSERT_TRUE(s.IsIOError()) << s.ToString();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  ASSERT_OK(fault_fs->DeleteFilesCreatedAfterLastDirSync(IOOptions(), nullptr));
+
+  s = TryReopen(options);
+  ASSERT_OK(s);
+  ASSERT_EQ("value", Get("base"));
+  ASSERT_EQ("value", Get("recovered"));
+  Close();
+}
+
+// Regression test for WAL recovery while appending to a reused MANIFEST. The
+// first reopen forces recovery to create an SST and then fail after MANIFEST
+// sync. The simulated crash cleanup deletes files without a prior directory
+// sync; the recovered SST must survive because the synced MANIFEST references
+// it.
+TEST_F(DBBasicTest,
+       ReuseManifestOnOpenSyncsRecoverySstDirBeforeManifestAppend) {
+  auto fault_fs = std::make_shared<FaultInjectionTestFS>(env_->GetFileSystem());
+  std::unique_ptr<Env> fault_env(NewCompositeEnv(fault_fs));
+
+  Options options = CurrentOptions();
+  options.create_if_missing = true;
+  options.disable_auto_compactions = true;
+  options.env = fault_env.get();
+  options.reuse_manifest_on_open = true;
+
+  DestroyAndReopen(options);
+  ASSERT_OK(Put("base", "value"));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(Put("recovered", "value"));
+  ASSERT_OK(db_->FlushWAL(true));
+  Close();
+
+  fault_fs->ResetState();
+
+  std::atomic<bool> fail_after_manifest_sync{true};
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->SetCallBack(
+      "VersionSet::ProcessManifestWrites:AfterSyncManifest", [&](void* arg) {
+        if (fail_after_manifest_sync.exchange(false)) {
+          ASSERT_NE(nullptr, arg);
+          IOStatus* io_s = static_cast<IOStatus*>(arg);
+          ASSERT_OK(*io_s);
+          *io_s = IOStatus::IOError("injected manifest sync aftermath");
+        }
+      });
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->EnableProcessing();
+
+  Status s = TryReopen(options);
+  ASSERT_TRUE(s.IsIOError()) << s.ToString();
+
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
+  ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
+
+  ASSERT_OK(fault_fs->DeleteFilesCreatedAfterLastDirSync(IOOptions(), nullptr));
+
+  s = TryReopen(options);
+  ASSERT_OK(s);
+  ASSERT_EQ("value", Get("base"));
+  ASSERT_EQ("value", Get("recovered"));
+  Close();
+}
+
 // Direct unit test for WritableFileWriter's initial_file_size parameter:
 // verifies the visible size accessors report the existing on-disk size
 // immediately, rather than the constructor's zero default.

db/db_impl/db_impl_open.cc

@@ -2234,6 +2234,16 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
 
   constexpr int level = 0;
 
+  if (s.ok() && has_output) {
+    // Before publishing any recovery edit, make the new SST's directory entry
+    // durable. Reused MANIFESTs have no CURRENT update, and fresh MANIFEST
+    // publication should not be relied on to order the recovered SST entry.
+    s = GetDataDir(cfd, meta.fd.GetPathId())
+            ->FsyncWithDirOptions(
+                IOOptions(), nullptr,
+                DirFsyncOptions(DirFsyncOptions::FsyncReason::kNewFileSynced));
+  }
+
   if (s.ok() && has_output) {
     edit->AddFile(level, meta.fd.GetNumber(), meta.fd.GetPathId(),
                   meta.fd.GetFileSize(), meta.smallest, meta.largest,

db/version_set.cc

@@ -6284,6 +6284,8 @@ Status VersionSet::ProcessManifestWrites(
       io_s = SetCurrentFile(
           write_options, fs_.get(), dbname_, pending_manifest_file_number_,
           file_options_.temperature, dir_contains_current_file);
+      TEST_SYNC_POINT_CALLBACK(
+          "VersionSet::ProcessManifestWrites:AfterSetCurrentFile", &io_s);
       if (!io_s.ok()) {
         s = io_s;
         // Quarantine old manifest file in case new manifest file's CURRENT