Sanity check on Open

Summary:
Everytime a client opens a DB, we do a sanity check that:
* checks the existance of all the necessary files
* verifies that file sizes are correct

Some of the code was stolen from https://reviews.facebook.net/D16935

Test Plan: added a unit test

Reviewers: dhruba, haobo, sdong

Reviewed By: dhruba

CC: leveldb

Differential Revision: https://reviews.facebook.net/D17097

HISTORY.md

@@ -13,8 +13,8 @@
 * Chagned Options.prefix_extractor from raw pointer to shared_ptr (take ownership)
   Changed HashSkipListRepFactory and HashLinkListRepFactory constructor to not take SliceTransform object (use Options.prefix_extractor implicitly)
 * Added Env::GetThreadPoolQueueLen(), which returns the waiting queue length of thread pools
-* Added DB::CheckConsistency(), which checks the consistency of live files' metadata
+* Added a command "checkconsistency" in ldb tool, which checks
-  Added a corresponding command "checkconsistency" in ldb tool
+  if file system state matches DB state (file existence and file sizes)
 
 ### New Features
 * If we find one truncated record at the end of the MANIFEST or WAL files,

db/corruption_test.cc

@@ -376,6 +376,38 @@ TEST(CorruptionTest, UnrelatedKeys) {
   ASSERT_EQ(Value(1000, &tmp2).ToString(), v);
 }
 
+TEST(CorruptionTest, FileSystemStateCorrupted) {
+  for (int iter = 0; iter < 2; ++iter) {
+    Options options;
+    options.paranoid_checks = true;
+    options.create_if_missing = true;
+    Reopen(&options);
+    Build(10);
+    ASSERT_OK(db_->Flush(FlushOptions()));
+    DBImpl* dbi = reinterpret_cast<DBImpl*>(db_);
+    std::vector<LiveFileMetaData> metadata;
+    dbi->GetLiveFilesMetaData(&metadata);
+    ASSERT_GT(metadata.size(), 0);
+    std::string filename = dbname_ + metadata[0].name;
+
+    delete db_;
+
+    if (iter == 0) {  // corrupt file size
+      unique_ptr<WritableFile> file;
+      env_.NewWritableFile(filename, &file, EnvOptions());
+      file->Append(Slice("corrupted sst"));
+      file.reset();
+    } else {  // delete the file
+      env_.DeleteFile(filename);
+    }
+
+    Status x = TryReopen(&options);
+    ASSERT_TRUE(x.IsCorruption());
+    DestroyDB(dbname_, options_);
+    Reopen(&options);
+  }
+}
+
 }  // namespace rocksdb
 
 int main(int argc, char** argv) {

db/db_impl.cc

@@ -969,6 +969,9 @@ Status DBImpl::Recover(bool read_only, bool error_if_log_file_exist) {
   }
 
   Status s = versions_->Recover();
+  if (options_.paranoid_checks && s.ok()) {
+    s = CheckConsistency();
+  }
   if (s.ok()) {
     SequenceNumber max_sequence(0);
 
@@ -3828,11 +3831,38 @@ Status DBImpl::DeleteFile(std::string name) {
   return status;
 }
 
-void DBImpl::GetLiveFilesMetaData(std::vector<LiveFileMetaData> *metadata) {
+void DBImpl::GetLiveFilesMetaData(std::vector<LiveFileMetaData>* metadata) {
   MutexLock l(&mutex_);
   return versions_->GetLiveFilesMetaData(metadata);
 }
 
+Status DBImpl::CheckConsistency() {
+  mutex_.AssertHeld();
+  std::vector<LiveFileMetaData> metadata;
+  versions_->GetLiveFilesMetaData(&metadata);
+
+  std::string corruption_messages;
+  for (const auto& md : metadata) {
+    std::string file_path = dbname_ + md.name;
+    uint64_t fsize = 0;
+    Status s = env_->GetFileSize(file_path, &fsize);
+    if (!s.ok()) {
+      corruption_messages +=
+          "Can't access " + md.name + ": " + s.ToString() + "\n";
+    } else if (fsize != md.size) {
+      corruption_messages += "Sst file size mismatch: " + md.name +
+                             ". Size recorded in manifest " +
+                             std::to_string(md.size) + ", actual size " +
+                             std::to_string(fsize) + "\n";
+    }
+  }
+  if (corruption_messages.size() == 0) {
+    return Status::OK();
+  } else {
+    return Status::Corruption(corruption_messages);
+  }
+}
+
 void DBImpl::TEST_GetFilesMetaData(
     std::vector<std::vector<FileMetaData>>* metadata) {
   MutexLock l(&mutex_);

db/db_impl.h

@@ -94,6 +94,10 @@ class DBImpl : public DB {
   virtual void GetLiveFilesMetaData(
     std::vector<LiveFileMetaData> *metadata);
 
+  // checks if all live files exist on file system and that their file sizes
+  // match to our in-memory records
+  virtual Status CheckConsistency();
+
   virtual Status GetDbIdentity(std::string& identity);
 
   Status RunManualCompaction(int input_level,

db/db_impl_readonly.cc

@@ -99,46 +99,4 @@ Status DB::OpenForReadOnly(const Options& options, const std::string& dbname,
   return s;
 }
 
-Status DB::CheckConsistency(const Options& options,
-    const std::string& name) {
-  DB *db = nullptr;
-  Status st;
-
-  st = DB::OpenForReadOnly(options, name, &db);
-  if (!st.ok()) {
-    return st;
-  }
-
-  std::vector<LiveFileMetaData> metadata;
-  db->GetLiveFilesMetaData(&metadata);
-
-  for (const auto& md : metadata) {
-    std::string file_path = name + md.name;
-
-    if (!db->GetEnv()->FileExists(file_path)) {
-      st = Status::Corruption("sst file " + md.name + " doesn't exist");
-      break;
-    }
-
-    uint64_t fsize = 0;
-    st = db->GetEnv()->GetFileSize(file_path, &fsize);
-    if (!st.ok()) {
-      st = Status::Corruption(
-          "Failed to determine the actual size of file " + md.name +
-          ": " + st.ToString());
-      break;
-    }
-    if (fsize != md.size) {
-      st = Status::Corruption(
-          "sst file size mismatch: " + md.name +
-          ". Size recorded in manifest " + std::to_string(md.size) +
-          ", actual size " + std::to_string(fsize));
-      break;
-    }
-  }
-
-  delete db;
-  return st;
-}
-
 }   // namespace rocksdb

include/rocksdb/db.h

@@ -91,14 +91,6 @@ class DB {
       const std::string& name, DB** dbptr,
       bool error_if_log_file_exist = false);
 
-  // Check the consistency of live files' metadata.
-  // It will return Corruption Status when a file in manifest
-  // doesn't actually exist or doesn't match the actual file size.
-  // Note: This call should be invoked only when the database is
-  // not already open and serving data.
-  static Status CheckConsistency(const Options& options,
-      const std::string& name);
-
   DB() { }
   virtual ~DB();
 

util/ldb_cmd.cc

@@ -1765,10 +1765,13 @@ void CheckConsistencyCommand::Help(string& ret) {
 
 void CheckConsistencyCommand::DoCommand() {
   Options opt = PrepareOptionsForOpenDB();
+  opt.paranoid_checks = true;
   if (!exec_state_.IsNotStarted()) {
     return;
   }
-  Status st = DB::CheckConsistency(opt, db_path_);
+  DB* db;
+  Status st = DB::OpenForReadOnly(opt, db_path_, &db, false);
+  delete db;
   if (st.ok()) {
     fprintf(stdout, "OK\n");
   } else {