Track WAL in MANIFEST: add method to check WAL consistency (#7236)

Summary:
Add a method `CheckWals` in `WalSet` to check the logs on disk. See `CheckWals`'s comments.
This method will be used to check consistency of WALs during DB recovery.

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

Test Plan: a set of tests are added to wal_edit_test.cc.

Reviewed By: riversand963

Differential Revision: D23036505

Pulled By: cheng-chang

fbshipit-source-id: 5b1d6857ac173429b00f950c32c4a5b8d063a732

db/wal_edit.cc

@@ -189,4 +189,46 @@ Status WalSet::DeleteWals(const WalDeletions& wals) {
 
 void WalSet::Reset() { wals_.clear(); }
 
+Status WalSet::CheckWals(
+    Env* env,
+    const std::unordered_map<WalNumber, std::string>& logs_on_disk) const {
+  assert(env != nullptr);
+
+  Status s;
+  for (const auto& wal : wals_) {
+    const uint64_t log_number = wal.first;
+    const WalMetadata& wal_meta = wal.second;
+
+    if (!wal_meta.HasSyncedSize()) {
+      // The WAL and WAL directory is not even synced,
+      // so the WAL's inode may not be persisted,
+      // then the WAL might not show up when listing WAL directory.
+      continue;
+    }
+
+    if (logs_on_disk.find(log_number) == logs_on_disk.end()) {
+      std::stringstream ss;
+      ss << "Missing WAL with log number: " << log_number << ".";
+      s = Status::Corruption(ss.str());
+      break;
+    }
+
+    uint64_t log_file_size = 0;
+    s = env->GetFileSize(logs_on_disk.at(log_number), &log_file_size);
+    if (!s.ok()) {
+      break;
+    }
+    if (log_file_size < wal_meta.GetSyncedSizeInBytes()) {
+      std::stringstream ss;
+      ss << "Size mismatch: WAL (log number: " << log_number
+         << ") in MANIFEST is " << wal_meta.GetSyncedSizeInBytes()
+         << " bytes , but actually is " << log_file_size << " bytes on disk.";
+      s = Status::Corruption(ss.str());
+      break;
+    }
+  }
+
+  return s;
+}
+
 }  // namespace ROCKSDB_NAMESPACE

db/wal_edit.h

@@ -12,6 +12,7 @@
 #include <map>
 #include <ostream>
 #include <string>
+#include <unordered_map>
 #include <vector>
 
 #include "logging/event_logger.h"
@@ -150,6 +151,16 @@ class WalSet {
 
   const std::map<WalNumber, WalMetadata>& GetWals() const { return wals_; }
 
+  // Checks whether there are missing or corrupted WALs.
+  // Returns Status::OK if there is no missing nor corrupted WAL,
+  // otherwise returns Status::Corruption.
+  // logs_on_disk is a map from log number to the log filename.
+  // Note that logs_on_disk may contain logs that is obsolete but
+  // haven't been deleted from disk.
+  Status CheckWals(
+      Env* env,
+      const std::unordered_map<WalNumber, std::string>& logs_on_disk) const;
+
  private:
   std::map<WalNumber, WalMetadata> wals_;
 };

db/wal_edit_test.cc

@@ -5,6 +5,8 @@
 
 #include "db/wal_edit.h"
 
+#include "db/db_test_util.h"
+#include "file/file_util.h"
 #include "port/port.h"
 #include "port/stack_trace.h"
 #include "test_util/testharness.h"
@@ -131,6 +133,111 @@ TEST(WalSet, DeleteNonClosedWal) {
               std::string::npos);
 }
 
+class WalSetTest : public DBTestBase {
+ public:
+  WalSetTest() : DBTestBase("WalSetTest", /* env_do_fsync */ true) {}
+
+  void SetUp() override {
+    test_dir_ = test::PerThreadDBPath("wal_set_test");
+    ASSERT_OK(env_->CreateDir(test_dir_));
+  }
+
+  void TearDown() override {
+    EXPECT_OK(DestroyDir(env_, test_dir_));
+    logs_on_disk_.clear();
+    wals_.Reset();
+  }
+
+  void CreateWalOnDisk(WalNumber number, const std::string& fname,
+                       uint64_t size_bytes) {
+    std::unique_ptr<WritableFile> f;
+    std::string fpath = Path(fname);
+    ASSERT_OK(env_->NewWritableFile(fpath, &f, EnvOptions()));
+    std::string content(size_bytes, '0');
+    ASSERT_OK(f->Append(content));
+    ASSERT_OK(f->Close());
+
+    logs_on_disk_[number] = fpath;
+  }
+
+  void AddWalToWalSet(WalNumber number, uint64_t size_bytes) {
+    // Create WAL.
+    ASSERT_OK(wals_.AddWal(WalAddition(number)));
+    // Close WAL.
+    WalMetadata wal(size_bytes);
+    wal.SetClosed();
+    ASSERT_OK(wals_.AddWal(WalAddition(number, wal)));
+  }
+
+  Status CheckWals() const { return wals_.CheckWals(env_, logs_on_disk_); }
+
+ private:
+  std::string test_dir_;
+  std::unordered_map<WalNumber, std::string> logs_on_disk_;
+  WalSet wals_;
+
+  std::string Path(const std::string& fname) { return test_dir_ + "/" + fname; }
+};
+
+TEST_F(WalSetTest, CheckEmptyWals) { ASSERT_OK(CheckWals()); }
+
+TEST_F(WalSetTest, CheckWals) {
+  for (int number = 1; number < 10; number++) {
+    uint64_t size = rand() % 100;
+    std::stringstream ss;
+    ss << "log" << number;
+    std::string fname = ss.str();
+    CreateWalOnDisk(number, fname, size);
+    // log 0 - 5 are obsolete.
+    if (number > 5) {
+      AddWalToWalSet(number, size);
+    }
+  }
+  ASSERT_OK(CheckWals());
+}
+
+TEST_F(WalSetTest, CheckMissingWals) {
+  for (int number = 1; number < 10; number++) {
+    uint64_t size = rand() % 100;
+    AddWalToWalSet(number, size);
+    // logs with even number are missing from disk.
+    if (number % 2) {
+      std::stringstream ss;
+      ss << "log" << number;
+      std::string fname = ss.str();
+      CreateWalOnDisk(number, fname, size);
+    }
+  }
+
+  Status s = CheckWals();
+  ASSERT_TRUE(s.IsCorruption()) << s.ToString();
+  // The first log with even number is missing.
+  std::stringstream expected_err;
+  expected_err << "Missing WAL with log number: " << 2;
+  ASSERT_TRUE(s.ToString().find(expected_err.str()) != std::string::npos)
+      << s.ToString();
+}
+
+TEST_F(WalSetTest, CheckWalsWithShrinkedSize) {
+  for (int number = 1; number < 10; number++) {
+    uint64_t size = rand() % 100 + 1;
+    AddWalToWalSet(number, size);
+    // logs with even number have shrinked size.
+    std::stringstream ss;
+    ss << "log" << number;
+    std::string fname = ss.str();
+    CreateWalOnDisk(number, fname, (number % 2) ? size : size - 1);
+  }
+
+  Status s = CheckWals();
+  ASSERT_TRUE(s.IsCorruption()) << s.ToString();
+  // The first log with even number has wrong size.
+  std::stringstream expected_err;
+  expected_err << "Size mismatch: WAL (log number: " << 2 << ")";
+  ASSERT_TRUE(s.ToString().find(expected_err.str()) != std::string::npos)
+      << s.ToString();
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {