From 83f9a6fd2102f62f720fb1fcb2d1b777bdb6a37f Mon Sep 17 00:00:00 2001
From: Andrew Kryczka <andrewkr@fb.com>
Date: Wed, 14 Dec 2016 16:29:23 -0800
Subject: [PATCH] Fail BackupEngine::Open upon meta-file read error

Summary:
We used to treat any failure to read a backup's meta-file as if the backup were corrupted; however, we should distinguish corruption errors from errors in the backup Env. This fixes an issue where callers would get inconsistent results from GetBackupInfo() if they called it on an engine that encountered Env error during initialization. Now we fail Initialize() in this case so callers cannot invoke GetBackupInfo() on such engines.
Closes https://github.com/facebook/rocksdb/pull/1654

Differential Revision: D4318573

Pulled By: ajkr

fbshipit-source-id: f7a7c54
---
 utilities/backupable/backupable_db.cc      |  9 +++++--
 utilities/backupable/backupable_db_test.cc | 31 ++++++++++++++++++++--
 2 files changed, 36 insertions(+), 4 deletions(-)

diff --git a/utilities/backupable/backupable_db.cc b/utilities/backupable/backupable_db.cc
index aaddfd7866..d6d9feded7 100644
--- a/utilities/backupable/backupable_db.cc
+++ b/utilities/backupable/backupable_db.cc
@@ -613,11 +613,16 @@ Status BackupEngineImpl::Initialize() {
           &abs_path_to_size);
       Status s =
           backup.second->LoadFromFile(options_.backup_dir, abs_path_to_size);
-      if (!s.ok()) {
+      if (s.IsCorruption()) {
         Log(options_.info_log, "Backup %u corrupted -- %s", backup.first,
             s.ToString().c_str());
         corrupt_backups_.insert(std::make_pair(
               backup.first, std::make_pair(s, std::move(backup.second))));
+      } else if (!s.ok()) {
+        // Distinguish corruption errors from errors in the backup Env.
+        // Errors in the backup Env (i.e., this code path) will cause Open() to
+        // fail, whereas corruption errors would not cause Open() failures.
+        return s;
       } else {
         Log(options_.info_log, "Loading backup %" PRIu32 " OK:\n%s",
             backup.first, backup.second->GetInfoString().c_str());
@@ -1615,7 +1620,7 @@ Status BackupEngineImpl::BackupMeta::LoadFromFile(
       try {
         size = abs_path_to_size.at(abs_path);
       } catch (std::out_of_range&) {
-        return Status::NotFound("Size missing for pathname: " + abs_path);
+        return Status::Corruption("Size missing for pathname: " + abs_path);
       }
     }
 
diff --git a/utilities/backupable/backupable_db_test.cc b/utilities/backupable/backupable_db_test.cc
index 3f4115c565..8d4468f76b 100644
--- a/utilities/backupable/backupable_db_test.cc
+++ b/utilities/backupable/backupable_db_test.cc
@@ -148,8 +148,12 @@ class TestEnv : public EnvWrapper {
 
   class DummySequentialFile : public SequentialFile {
    public:
-    DummySequentialFile() : SequentialFile(), rnd_(5) {}
+    explicit DummySequentialFile(bool fail_reads)
+        : SequentialFile(), rnd_(5), fail_reads_(fail_reads) {}
     virtual Status Read(size_t n, Slice* result, char* scratch) override {
+      if (fail_reads_) {
+        return Status::IOError();
+      }
       size_t read_size = (n > size_left) ? size_left : n;
       for (size_t i = 0; i < read_size; ++i) {
         scratch[i] = rnd_.Next() & 255;
@@ -166,13 +170,15 @@ class TestEnv : public EnvWrapper {
    private:
     size_t size_left = 200;
     Random rnd_;
+    bool fail_reads_;
   };
 
   Status NewSequentialFile(const std::string& f, unique_ptr<SequentialFile>* r,
                            const EnvOptions& options) override {
     MutexLock l(&mutex_);
     if (dummy_sequential_file_) {
-      r->reset(new TestEnv::DummySequentialFile());
+      r->reset(
+          new TestEnv::DummySequentialFile(dummy_sequential_file_fail_reads_));
       return Status::OK();
     } else {
       return EnvWrapper::NewSequentialFile(f, r, options);
@@ -224,6 +230,10 @@ class TestEnv : public EnvWrapper {
     MutexLock l(&mutex_);
     dummy_sequential_file_ = dummy_sequential_file;
   }
+  void SetDummySequentialFileFailReads(bool dummy_sequential_file_fail_reads) {
+    MutexLock l(&mutex_);
+    dummy_sequential_file_fail_reads_ = dummy_sequential_file_fail_reads;
+  }
 
   void SetGetChildrenFailure(bool fail) { get_children_failure_ = fail; }
   Status GetChildren(const std::string& dir,
@@ -273,6 +283,7 @@ class TestEnv : public EnvWrapper {
  private:
   port::Mutex mutex_;
   bool dummy_sequential_file_ = false;
+  bool dummy_sequential_file_fail_reads_ = false;
   std::vector<std::string> written_files_;
   std::vector<std::string> filenames_for_mocked_attrs_;
   uint64_t limit_written_files_ = 1000000;
@@ -1277,6 +1288,22 @@ TEST_F(BackupableDBTest, EnvFailures) {
     test_backup_env_->SetNewDirectoryFailure(false);
   }
 
+  // Read from meta-file failure
+  {
+    DestroyDB(dbname_, options_);
+    OpenDBAndBackupEngine(true);
+    FillDB(db_.get(), 0, 100);
+    ASSERT_TRUE(backup_engine_->CreateNewBackup(db_.get(), true).ok());
+    CloseDBAndBackupEngine();
+    test_backup_env_->SetDummySequentialFile(true);
+    test_backup_env_->SetDummySequentialFileFailReads(true);
+    backupable_options_->destroy_old_data = false;
+    ASSERT_NOK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                  &backup_engine));
+    test_backup_env_->SetDummySequentialFile(false);
+    test_backup_env_->SetDummySequentialFileFailReads(false);
+  }
+
   // no failure
   {
     ASSERT_OK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,