Better error handling in BackupEngine

Summary:
Couple of changes here:
* NewBackupEngine() and NewReadOnlyBackupEngine() are now removed. They were deprecated since RocksDB 3.8. Changing these to new functions should be pretty straight-forward. As a followup, I'll fix all fbcode callsights
* Instead of initializing backup engine in the constructor, we initialize it in a separate function now. That way, we can catch all errors and return appropriate status code.
* We catch all errors during initializations and return them to the client properly.
* Added new tests to backupable_db_test, to make sure that we can't open BackupEngine when there are Env errors.
* Transitioned backupable_db_test to use BackupEngine rather than BackupableDB. From the two available APIs, judging by the current use-cases, it looks like BackupEngine API won. It's much more flexible since it doesn't require StackableDB.

Test Plan: Added a new unit test to backupable_db_test

Reviewers: yhchiang, sdong, AaronFeldman

Reviewed By: AaronFeldman

Subscribers: dhruba, leveldb

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

HISTORY.md

@@ -2,9 +2,10 @@
 
 ## Unreleased
 
-### Public API changes
+### Public API Changes
 * Deprecated WriteOptions::timeout_hint_us. We no longer support write timeout. If you really need this option, talk to us and we might consider returning it.
 * Deprecated purge_redundant_kvs_while_flush option.
+* Removed BackupEngine::NewBackupEngine() and NewReadOnlyBackupEngine() that were deprecated in RocksDB 3.8. Please use BackupEngine::Open() instead.
 
 ## 3.12.0 (7/2/2015)
 ### New Features

include/rocksdb/utilities/backupable_db.h

@@ -178,10 +178,6 @@ class BackupEngineReadOnly {
  public:
   virtual ~BackupEngineReadOnly() {}
 
-  static BackupEngineReadOnly* NewReadOnlyBackupEngine(
-      Env* db_env, const BackupableDBOptions& options)
-      __attribute__((deprecated("Please use Open() instead")));
-
   static Status Open(Env* db_env, const BackupableDBOptions& options,
                      BackupEngineReadOnly** backup_engine_ptr);
 
@@ -208,10 +204,6 @@ class BackupEngine {
  public:
   virtual ~BackupEngine() {}
 
-  static BackupEngine* NewBackupEngine(Env* db_env,
-                                       const BackupableDBOptions& options)
-    __attribute__((deprecated("Please use Open() instead")));
-
   static Status Open(Env* db_env,
                      const BackupableDBOptions& options,
                      BackupEngine** backup_engine_ptr);
@@ -272,6 +264,7 @@ class BackupableDB : public StackableDB {
 
  private:
   BackupEngine* backup_engine_;
+  Status status_;
 };
 
 // Use this class to access information about backups and restore from them
@@ -317,6 +310,7 @@ class RestoreBackupableDB {
 
  private:
   BackupEngine* backup_engine_;
+  Status status_;
 };
 
 }  // namespace rocksdb

utilities/backupable/backupable_db.cc

@@ -144,6 +144,8 @@ class BackupEngineImpl : public BackupEngine {
                                restore_options);
   }
 
+  Status Initialize();
+
  private:
   void DeleteChildren(const std::string& dir, uint32_t file_type_filter = 0);
 
@@ -192,7 +194,7 @@ class BackupEngineImpl : public BackupEngine {
 
     Status AddFile(std::shared_ptr<FileInfo> file_info);
 
-    void Delete(bool delete_meta = true);
+    Status Delete(bool delete_meta = true);
 
     bool Empty() {
       return files_.empty();
@@ -419,6 +421,7 @@ class BackupEngineImpl : public BackupEngine {
     }
   };
 
+  bool initialized_;
   channel<CopyWorkItem> files_to_copy_;
   std::vector<std::thread> threads_;
 
@@ -459,58 +462,83 @@ class BackupEngineImpl : public BackupEngine {
   BackupStatistics backup_statistics_;
 };
 
-BackupEngine* BackupEngine::NewBackupEngine(
-    Env* db_env, const BackupableDBOptions& options) {
-  return new BackupEngineImpl(db_env, options);
-}
-
-Status BackupEngine::Open(Env* env,
-                          const BackupableDBOptions& options,
+Status BackupEngine::Open(Env* env, const BackupableDBOptions& options,
                           BackupEngine** backup_engine_ptr) {
-  *backup_engine_ptr = new BackupEngineImpl(env, options);
+  std::unique_ptr<BackupEngineImpl> backup_engine(
+      new BackupEngineImpl(env, options));
+  auto s = backup_engine->Initialize();
+  if (!s.ok()) {
+    *backup_engine_ptr = nullptr;
+    return s;
+  }
+  *backup_engine_ptr = backup_engine.release();
   return Status::OK();
 }
 
 BackupEngineImpl::BackupEngineImpl(Env* db_env,
                                    const BackupableDBOptions& options,
                                    bool read_only)
-    : stop_backup_(false),
+    : initialized_(false),
+      stop_backup_(false),
       options_(options),
       db_env_(db_env),
       backup_env_(options.backup_env != nullptr ? options.backup_env : db_env_),
       copy_file_buffer_size_(kDefaultCopyFileBufferSize),
-      read_only_(read_only) {
+      read_only_(read_only) {}
+
+BackupEngineImpl::~BackupEngineImpl() {
+  files_to_copy_.sendEof();
+  for (auto& t : threads_) {
+    t.join();
+  }
+  LogFlush(options_.info_log);
+}
 
+Status BackupEngineImpl::Initialize() {
+  assert(!initialized_);
+  initialized_ = true;
   if (read_only_) {
     Log(options_.info_log, "Starting read_only backup engine");
   }
   options_.Dump(options_.info_log);
 
   if (!read_only_) {
-    // create all the dirs we need
-    backup_env_->CreateDirIfMissing(GetAbsolutePath());
-    backup_env_->NewDirectory(GetAbsolutePath(), &backup_directory_);
+    // gather the list of directories that we need to create
+    std::vector<std::pair<std::string, std::unique_ptr<Directory>*>>
+        directories;
+    directories.emplace_back(GetAbsolutePath(), &backup_directory_);
     if (options_.share_table_files) {
       if (options_.share_files_with_checksum) {
-        backup_env_->CreateDirIfMissing(GetAbsolutePath(
-            GetSharedFileWithChecksumRel()));
-        backup_env_->NewDirectory(GetAbsolutePath(
-            GetSharedFileWithChecksumRel()), &shared_directory_);
+        directories.emplace_back(
+            GetAbsolutePath(GetSharedFileWithChecksumRel()),
+            &shared_directory_);
       } else {
-        backup_env_->CreateDirIfMissing(GetAbsolutePath(GetSharedFileRel()));
-        backup_env_->NewDirectory(GetAbsolutePath(GetSharedFileRel()),
-                                  &shared_directory_);
+        directories.emplace_back(GetAbsolutePath(GetSharedFileRel()),
+                                 &shared_directory_);
+      }
+    }
+    directories.emplace_back(GetAbsolutePath(GetPrivateDirRel()),
+                             &private_directory_);
+    directories.emplace_back(GetBackupMetaDir(), &meta_directory_);
+    // create all the dirs we need
+    for (const auto& d : directories) {
+      auto s = backup_env_->CreateDirIfMissing(d.first);
+      if (s.ok()) {
+        s = backup_env_->NewDirectory(d.first, d.second);
+      }
+      if (!s.ok()) {
+        return s;
       }
     }
-    backup_env_->CreateDirIfMissing(GetAbsolutePath(GetPrivateDirRel()));
-    backup_env_->NewDirectory(GetAbsolutePath(GetPrivateDirRel()),
-                              &private_directory_);
-    backup_env_->CreateDirIfMissing(GetBackupMetaDir());
-    backup_env_->NewDirectory(GetBackupMetaDir(), &meta_directory_);
   }
 
   std::vector<std::string> backup_meta_files;
-  backup_env_->GetChildren(GetBackupMetaDir(), &backup_meta_files);
+  {
+    auto s = backup_env_->GetChildren(GetBackupMetaDir(), &backup_meta_files);
+    if (!s.ok()) {
+      return s;
+    }
+  }
   // create backups_ structure
   for (auto& file : backup_meta_files) {
     if (file == "." || file == "..") {
@@ -521,10 +549,10 @@ BackupEngineImpl::BackupEngineImpl(Env* db_env,
     sscanf(file.c_str(), "%u", &backup_id);
     if (backup_id == 0 || file != rocksdb::ToString(backup_id)) {
       if (!read_only_) {
-        Log(options_.info_log, "Unrecognized meta file %s, deleting",
-            file.c_str());
         // invalid file name, delete that
-        backup_env_->DeleteFile(GetBackupMetaDir() + "/" + file);
+        auto s = backup_env_->DeleteFile(GetBackupMetaDir() + "/" + file);
+        Log(options_.info_log, "Unrecognized meta file %s, deleting -- %s",
+            file.c_str(), s.ToString().c_str());
       }
       continue;
     }
@@ -540,8 +568,13 @@ BackupEngineImpl::BackupEngineImpl(Env* db_env,
     Log(options_.info_log,
         "Backup Engine started with destroy_old_data == true, deleting all "
         "backups");
-    PurgeOldBackups(0);
-    (void) GarbageCollect();
+    auto s = PurgeOldBackups(0);
+    if (s.ok()) {
+      s = GarbageCollect();
+    }
+    if (!s.ok()) {
+      return s;
+    }
     // start from beginning
     latest_backup_id_ = 0;
   } else {  // Load data from storage
@@ -586,19 +619,27 @@ BackupEngineImpl::BackupEngineImpl(Env* db_env,
     later_ids.push_back(itr->first);
   }
   for (auto id : later_ids) {
+    Status s;
     if (!read_only_) {
-      DeleteBackup(id);
+      s = DeleteBackup(id);
     } else {
       auto backup = backups_.find(id);
       // We just found it couple of lines earlier!
       assert(backup != backups_.end());
-      backup->second->Delete(false);
+      s = backup->second->Delete(false);
       backups_.erase(backup);
     }
+    if (!s.ok()) {
+      Log(options_.info_log, "Failed deleting backup %" PRIu32 " -- %s", id,
+          s.ToString().c_str());
+    }
   }
 
   if (!read_only_) {
-    PutLatestBackupFileContents(latest_backup_id_);  // Ignore errors
+    auto s = PutLatestBackupFileContents(latest_backup_id_);
+    if (!s.ok()) {
+      return s;
+    }
   }
 
   // set up threads perform copies from files_to_copy_ in the background
@@ -622,19 +663,14 @@ BackupEngineImpl::BackupEngineImpl(Env* db_env,
   }
 
   Log(options_.info_log, "Initialized BackupEngine");
-}
 
-BackupEngineImpl::~BackupEngineImpl() {
-  files_to_copy_.sendEof();
-  for (int i = 0; i < options_.max_background_operations; i++) {
-    threads_[i].join();
-  }
-  LogFlush(options_.info_log);
+  return Status::OK();
 }
 
 Status BackupEngineImpl::CreateNewBackup(DB* db, bool flush_before_backup) {
+  assert(initialized_);
   if (options_.max_background_operations > 1 &&
-          options_.backup_rate_limit != 0) {
+      options_.backup_rate_limit != 0) {
     return Status::InvalidArgument(
         "Multi-threaded backups cannot use a backup_rate_limit");
   }
@@ -838,6 +874,7 @@ Status BackupEngineImpl::CreateNewBackup(DB* db, bool flush_before_backup) {
 }
 
 Status BackupEngineImpl::PurgeOldBackups(uint32_t num_backups_to_keep) {
+  assert(initialized_);
   assert(!read_only_);
   Log(options_.info_log, "Purging old backups, keeping %u",
       num_backups_to_keep);
@@ -848,24 +885,34 @@ Status BackupEngineImpl::PurgeOldBackups(uint32_t num_backups_to_keep) {
     itr++;
   }
   for (auto backup_id : to_delete) {
-    DeleteBackup(backup_id);
+    auto s = DeleteBackup(backup_id);
+    if (!s.ok()) {
+      return s;
+    }
   }
   return Status::OK();
 }
 
 Status BackupEngineImpl::DeleteBackup(BackupID backup_id) {
+  assert(initialized_);
   assert(!read_only_);
   Log(options_.info_log, "Deleting backup %u", backup_id);
   auto backup = backups_.find(backup_id);
   if (backup != backups_.end()) {
-    backup->second->Delete();
+    auto s = backup->second->Delete();
+    if (!s.ok()) {
+      return s;
+    }
     backups_.erase(backup);
   } else {
     auto corrupt = corrupt_backups_.find(backup_id);
     if (corrupt == corrupt_backups_.end()) {
       return Status::NotFound("Backup not found");
     }
-    corrupt->second.second->Delete();
+    auto s = corrupt->second.second->Delete();
+    if (!s.ok()) {
+      return s;
+    }
     corrupt_backups_.erase(corrupt);
   }
 
@@ -892,6 +939,7 @@ Status BackupEngineImpl::DeleteBackup(BackupID backup_id) {
 }
 
 void BackupEngineImpl::GetBackupInfo(std::vector<BackupInfo>* backup_info) {
+  assert(initialized_);
   backup_info->reserve(backups_.size());
   for (auto& backup : backups_) {
     if (!backup.second->Empty()) {
@@ -906,6 +954,7 @@ void BackupEngineImpl::GetBackupInfo(std::vector<BackupInfo>* backup_info) {
 void
 BackupEngineImpl::GetCorruptedBackups(
     std::vector<BackupID>* corrupt_backup_ids) {
+  assert(initialized_);
   corrupt_backup_ids->reserve(corrupt_backups_.size());
   for (auto& backup : corrupt_backups_) {
     corrupt_backup_ids->push_back(backup.first);
@@ -915,8 +964,9 @@ BackupEngineImpl::GetCorruptedBackups(
 Status BackupEngineImpl::RestoreDBFromBackup(
     BackupID backup_id, const std::string& db_dir, const std::string& wal_dir,
     const RestoreOptions& restore_options) {
+  assert(initialized_);
   if (options_.max_background_operations > 1 &&
-          options_.restore_rate_limit != 0) {
+      options_.restore_rate_limit != 0) {
     return Status::InvalidArgument(
         "Multi-threaded restores cannot use a restore_rate_limit");
   }
@@ -1351,8 +1401,13 @@ Status BackupEngineImpl::GarbageCollect() {
 
   // delete obsolete shared files
   std::vector<std::string> shared_children;
-  backup_env_->GetChildren(GetAbsolutePath(GetSharedFileRel()),
-                           &shared_children);
+  {
+    auto s = backup_env_->GetChildren(GetAbsolutePath(GetSharedFileRel()),
+                                      &shared_children);
+    if (!s.ok()) {
+      return s;
+    }
+  }
   for (auto& child : shared_children) {
     std::string rel_fname = GetSharedFileRel(child);
     auto child_itr = backuped_file_infos_.find(rel_fname);
@@ -1362,17 +1417,21 @@ Status BackupEngineImpl::GarbageCollect() {
       // this might be a directory, but DeleteFile will just fail in that
       // case, so we're good
       Status s = backup_env_->DeleteFile(GetAbsolutePath(rel_fname));
-      if (s.ok()) {
-        Log(options_.info_log, "Deleted %s", rel_fname.c_str());
-      }
+      Log(options_.info_log, "Deleting %s -- %s", rel_fname.c_str(),
+          s.ToString().c_str());
       backuped_file_infos_.erase(rel_fname);
     }
   }
 
   // delete obsolete private files
   std::vector<std::string> private_children;
-  backup_env_->GetChildren(GetAbsolutePath(GetPrivateDirRel()),
-                           &private_children);
+  {
+    auto s = backup_env_->GetChildren(GetAbsolutePath(GetPrivateDirRel()),
+                                      &private_children);
+    if (!s.ok()) {
+      return s;
+    }
+  }
   for (auto& child : private_children) {
     BackupID backup_id = 0;
     bool tmp_dir = child.find(".tmp") != std::string::npos;
@@ -1389,14 +1448,12 @@ Status BackupEngineImpl::GarbageCollect() {
     backup_env_->GetChildren(full_private_path, &subchildren);
     for (auto& subchild : subchildren) {
       Status s = backup_env_->DeleteFile(full_private_path + subchild);
-      if (s.ok()) {
-        Log(options_.info_log, "Deleted %s",
-            (full_private_path + subchild).c_str());
-      }
+      Log(options_.info_log, "Deleting %s -- %s",
+          (full_private_path + subchild).c_str(), s.ToString().c_str());
     }
     // finally delete the private dir
     Status s = backup_env_->DeleteDir(full_private_path);
-    Log(options_.info_log, "Deleted dir %s -- %s", full_private_path.c_str(),
+    Log(options_.info_log, "Deleting dir %s -- %s", full_private_path.c_str(),
         s.ToString().c_str());
   }
 
@@ -1432,16 +1489,18 @@ Status BackupEngineImpl::BackupMeta::AddFile(
   return Status::OK();
 }
 
-void BackupEngineImpl::BackupMeta::Delete(bool delete_meta) {
+Status BackupEngineImpl::BackupMeta::Delete(bool delete_meta) {
+  Status s;
   for (const auto& file : files_) {
     --file->refs;  // decrease refcount
   }
   files_.clear();
   // delete meta file
-  if (delete_meta) {
-    env_->DeleteFile(meta_filename_);
+  if (delete_meta && env_->FileExists(meta_filename_)) {
+    s = env_->DeleteFile(meta_filename_);
   }
   timestamp_ = 0;
+  return s;
 }
 
 // each backup meta file is of the format:
@@ -1607,58 +1666,75 @@ class BackupEngineReadOnlyImpl : public BackupEngineReadOnly {
                                                      restore_options);
   }
 
+  Status Initialize() { return backup_engine_->Initialize(); }
+
  private:
   std::unique_ptr<BackupEngineImpl> backup_engine_;
 };
 
-BackupEngineReadOnly* BackupEngineReadOnly::NewReadOnlyBackupEngine(
-    Env* db_env, const BackupableDBOptions& options) {
-  if (options.destroy_old_data) {
-    assert(false);
-    return nullptr;
-  }
-  return new BackupEngineReadOnlyImpl(db_env, options);
-}
-
 Status BackupEngineReadOnly::Open(Env* env, const BackupableDBOptions& options,
                                   BackupEngineReadOnly** backup_engine_ptr) {
   if (options.destroy_old_data) {
-    assert(false);
     return Status::InvalidArgument(
         "Can't destroy old data with ReadOnly BackupEngine");
   }
-  *backup_engine_ptr = new BackupEngineReadOnlyImpl(env, options);
+  std::unique_ptr<BackupEngineReadOnlyImpl> backup_engine(
+      new BackupEngineReadOnlyImpl(env, options));
+  auto s = backup_engine->Initialize();
+  if (!s.ok()) {
+    *backup_engine_ptr = nullptr;
+    return s;
+  }
+  *backup_engine_ptr = backup_engine.release();
   return Status::OK();
 }
 
 // --- BackupableDB methods --------
 
 BackupableDB::BackupableDB(DB* db, const BackupableDBOptions& options)
-    : StackableDB(db),
-      backup_engine_(new BackupEngineImpl(db->GetEnv(), options)) {}
+    : StackableDB(db) {
+  auto backup_engine_impl = new BackupEngineImpl(db->GetEnv(), options);
+  status_ = backup_engine_impl->Initialize();
+  backup_engine_ = backup_engine_impl;
+}
 
 BackupableDB::~BackupableDB() {
   delete backup_engine_;
 }
 
 Status BackupableDB::CreateNewBackup(bool flush_before_backup) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->CreateNewBackup(this, flush_before_backup);
 }
 
 void BackupableDB::GetBackupInfo(std::vector<BackupInfo>* backup_info) {
+  if (!status_.ok()) {
+    return;
+  }
   backup_engine_->GetBackupInfo(backup_info);
 }
 
 void
 BackupableDB::GetCorruptedBackups(std::vector<BackupID>* corrupt_backup_ids) {
+  if (!status_.ok()) {
+    return;
+  }
   backup_engine_->GetCorruptedBackups(corrupt_backup_ids);
 }
 
 Status BackupableDB::PurgeOldBackups(uint32_t num_backups_to_keep) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->PurgeOldBackups(num_backups_to_keep);
 }
 
 Status BackupableDB::DeleteBackup(BackupID backup_id) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->DeleteBackup(backup_id);
 }
 
@@ -1667,14 +1743,20 @@ void BackupableDB::StopBackup() {
 }
 
 Status BackupableDB::GarbageCollect() {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->GarbageCollect();
 }
 
 // --- RestoreBackupableDB methods ------
 
 RestoreBackupableDB::RestoreBackupableDB(Env* db_env,
-                                         const BackupableDBOptions& options)
-    : backup_engine_(new BackupEngineImpl(db_env, options)) {}
+                                         const BackupableDBOptions& options) {
+  auto backup_engine_impl = new BackupEngineImpl(db_env, options);
+  status_ = backup_engine_impl->Initialize();
+  backup_engine_ = backup_engine_impl;
+}
 
 RestoreBackupableDB::~RestoreBackupableDB() {
   delete backup_engine_;
@@ -1682,17 +1764,26 @@ RestoreBackupableDB::~RestoreBackupableDB() {
 
 void
 RestoreBackupableDB::GetBackupInfo(std::vector<BackupInfo>* backup_info) {
+  if (!status_.ok()) {
+    return;
+  }
   backup_engine_->GetBackupInfo(backup_info);
 }
 
 void RestoreBackupableDB::GetCorruptedBackups(
     std::vector<BackupID>* corrupt_backup_ids) {
+  if (!status_.ok()) {
+    return;
+  }
   backup_engine_->GetCorruptedBackups(corrupt_backup_ids);
 }
 
 Status RestoreBackupableDB::RestoreDBFromBackup(
     BackupID backup_id, const std::string& db_dir, const std::string& wal_dir,
     const RestoreOptions& restore_options) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->RestoreDBFromBackup(backup_id, db_dir, wal_dir,
                                              restore_options);
 }
@@ -1700,19 +1791,31 @@ Status RestoreBackupableDB::RestoreDBFromBackup(
 Status RestoreBackupableDB::RestoreDBFromLatestBackup(
     const std::string& db_dir, const std::string& wal_dir,
     const RestoreOptions& restore_options) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->RestoreDBFromLatestBackup(db_dir, wal_dir,
                                                    restore_options);
 }
 
 Status RestoreBackupableDB::PurgeOldBackups(uint32_t num_backups_to_keep) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->PurgeOldBackups(num_backups_to_keep);
 }
 
 Status RestoreBackupableDB::DeleteBackup(BackupID backup_id) {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->DeleteBackup(backup_id);
 }
 
 Status RestoreBackupableDB::GarbageCollect() {
+  if (!status_.ok()) {
+    return status_;
+  }
   return backup_engine_->GarbageCollect();
 }
 

utilities/backupable/backupable_db_test.cc

@@ -12,6 +12,7 @@
 #include <iostream>
 
 #include "port/port.h"
+#include "port/stack_trace.h"
 #include "rocksdb/types.h"
 #include "rocksdb/transaction_log.h"
 #include "rocksdb/utilities/backupable_db.h"
@@ -217,12 +218,44 @@ class TestEnv : public EnvWrapper {
     dummy_sequential_file_ = dummy_sequential_file;
   }
 
+  void SetGetChildrenFailure(bool fail) { get_children_failure_ = fail; }
+  Status GetChildren(const std::string& dir,
+                     std::vector<std::string>* r) override {
+    if (get_children_failure_) {
+      return Status::IOError("SimulatedFailure");
+    }
+    return EnvWrapper::GetChildren(dir, r);
+  }
+
+  void SetCreateDirIfMissingFailure(bool fail) {
+    create_dir_if_missing_failure_ = fail;
+  }
+  Status CreateDirIfMissing(const std::string& d) override {
+    if (create_dir_if_missing_failure_) {
+      return Status::IOError("SimulatedFailure");
+    }
+    return EnvWrapper::CreateDirIfMissing(d);
+  }
+
+  void SetNewDirectoryFailure(bool fail) { new_directory_failure_ = fail; }
+  virtual Status NewDirectory(const std::string& name,
+                              unique_ptr<Directory>* result) override {
+    if (new_directory_failure_) {
+      return Status::IOError("SimulatedFailure");
+    }
+    return EnvWrapper::NewDirectory(name, result);
+  }
+
  private:
   port::Mutex mutex_;
   bool dummy_sequential_file_ = false;
   std::vector<std::string> written_files_;
   uint64_t limit_written_files_ = 1000000;
   uint64_t limit_delete_files_ = 1000000;
+
+  bool get_children_failure_ = false;
+  bool create_dir_if_missing_failure_ = false;
+  bool new_directory_failure_ = false;
 };  // TestEnv
 
 class FileManager : public EnvWrapper {
@@ -392,9 +425,9 @@ class BackupableDBTest : public testing::Test {
     return db;
   }
 
-  void OpenBackupableDB(bool destroy_old_data = false, bool dummy = false,
-                        bool share_table_files = true,
-                        bool share_with_checksums = false) {
+  void OpenDBAndBackupEngine(bool destroy_old_data = false, bool dummy = false,
+                             bool share_table_files = true,
+                             bool share_with_checksums = false) {
     // reset all the defaults
     test_backup_env_->SetLimitWrittenFiles(1000000);
     test_db_env_->SetLimitWrittenFiles(1000000);
@@ -407,25 +440,30 @@ class BackupableDBTest : public testing::Test {
     } else {
       ASSERT_OK(DB::Open(options_, dbname_, &db));
     }
+    db_.reset(db);
     backupable_options_->destroy_old_data = destroy_old_data;
     backupable_options_->share_table_files = share_table_files;
     backupable_options_->share_files_with_checksum = share_with_checksums;
-    db_.reset(new BackupableDB(db, *backupable_options_));
+    BackupEngine* backup_engine;
+    ASSERT_OK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                 &backup_engine));
+    backup_engine_.reset(backup_engine);
   }
 
-  void CloseBackupableDB() {
-    db_.reset(nullptr);
+  void CloseDBAndBackupEngine() {
+    db_.reset();
+    backup_engine_.reset();
   }
 
-  void OpenRestoreDB() {
+  void OpenBackupEngine() {
     backupable_options_->destroy_old_data = false;
-    restore_db_.reset(
-        new RestoreBackupableDB(test_db_env_.get(), *backupable_options_));
+    BackupEngine* backup_engine;
+    ASSERT_OK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                 &backup_engine));
+    backup_engine_.reset(backup_engine);
   }
 
-  void CloseRestoreDB() {
-    restore_db_.reset(nullptr);
-  }
+  void CloseBackupEngine() { backup_engine_.reset(nullptr); }
 
   // restores backup backup_id and asserts the existence of
   // [start_exist, end_exist> and not-existence of
@@ -437,17 +475,17 @@ class BackupableDBTest : public testing::Test {
                                uint32_t end_exist, uint32_t end = 0,
                                bool keep_log_files = false) {
     RestoreOptions restore_options(keep_log_files);
-    bool opened_restore = false;
-    if (restore_db_.get() == nullptr) {
-      opened_restore = true;
-      OpenRestoreDB();
+    bool opened_backup_engine = false;
+    if (backup_engine_.get() == nullptr) {
+      opened_backup_engine = true;
+      OpenBackupEngine();
     }
     if (backup_id > 0) {
-      ASSERT_OK(restore_db_->RestoreDBFromBackup(backup_id, dbname_, dbname_,
-                                                 restore_options));
+      ASSERT_OK(backup_engine_->RestoreDBFromBackup(backup_id, dbname_, dbname_,
+                                                    restore_options));
     } else {
-      ASSERT_OK(restore_db_->RestoreDBFromLatestBackup(dbname_, dbname_,
-                                                       restore_options));
+      ASSERT_OK(backup_engine_->RestoreDBFromLatestBackup(dbname_, dbname_,
+                                                          restore_options));
     }
     DB* db = OpenDB();
     AssertExists(db, start_exist, end_exist);
@@ -455,8 +493,8 @@ class BackupableDBTest : public testing::Test {
       AssertEmpty(db, end_exist, end);
     }
     delete db;
-    if (opened_restore) {
-      CloseRestoreDB();
+    if (opened_backup_engine) {
+      CloseBackupEngine();
     }
   }
 
@@ -486,8 +524,8 @@ class BackupableDBTest : public testing::Test {
 
   // all the dbs!
   DummyDB* dummy_db_; // BackupableDB owns dummy_db_
-  unique_ptr<BackupableDB> db_;
-  unique_ptr<RestoreBackupableDB> restore_db_;
+  unique_ptr<DB> db_;
+  unique_ptr<BackupEngine> backup_engine_;
 
   // options
   Options options_;
@@ -503,7 +541,7 @@ void AppendPath(const std::string& path, std::vector<std::string>& v) {
 
 // this will make sure that backup does not copy the same file twice
 TEST_F(BackupableDBTest, NoDoubleCopy) {
-  OpenBackupableDB(true, true);
+  OpenDBAndBackupEngine(true, true);
 
   // should write 5 DB files + LATEST_BACKUP + one meta file
   test_backup_env_->SetLimitWrittenFiles(7);
@@ -512,16 +550,12 @@ TEST_F(BackupableDBTest, NoDoubleCopy) {
   dummy_db_->live_files_ = { "/00010.sst", "/00011.sst",
                              "/CURRENT",   "/MANIFEST-01" };
   dummy_db_->wal_files_ = {{"/00011.log", true}, {"/00012.log", false}};
-  ASSERT_OK(db_->CreateNewBackup(false));
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
   std::vector<std::string> should_have_written = {
-    "/shared/00010.sst.tmp",
-    "/shared/00011.sst.tmp",
-    "/private/1.tmp/CURRENT",
-    "/private/1.tmp/MANIFEST-01",
-    "/private/1.tmp/00011.log",
-    "/meta/1.tmp",
-    "/LATEST_BACKUP.tmp"
-  };
+      "/shared/00010.sst.tmp",    "/shared/00011.sst.tmp",
+      "/private/1.tmp/CURRENT",   "/private/1.tmp/MANIFEST-01",
+      "/private/1.tmp/00011.log", "/meta/1.tmp",
+      "/LATEST_BACKUP.tmp"};
   AppendPath(dbname_ + "_backup", should_have_written);
   test_backup_env_->AssertWrittenFiles(should_have_written);
 
@@ -532,7 +566,7 @@ TEST_F(BackupableDBTest, NoDoubleCopy) {
   dummy_db_->live_files_ = { "/00010.sst", "/00015.sst",
                              "/CURRENT",   "/MANIFEST-01" };
   dummy_db_->wal_files_ = {{"/00011.log", true}, {"/00012.log", false}};
-  ASSERT_OK(db_->CreateNewBackup(false));
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
   // should not open 00010.sst - it's already there
   should_have_written = {
     "/shared/00015.sst.tmp",
@@ -545,7 +579,7 @@ TEST_F(BackupableDBTest, NoDoubleCopy) {
   AppendPath(dbname_ + "_backup", should_have_written);
   test_backup_env_->AssertWrittenFiles(should_have_written);
 
-  ASSERT_OK(db_->DeleteBackup(1));
+  ASSERT_OK(backup_engine_->DeleteBackup(1));
   ASSERT_TRUE(test_backup_env_->FileExists(backupdir_ + "/shared/00010.sst"));
   // 00011.sst was only in backup 1, should be deleted
   ASSERT_FALSE(test_backup_env_->FileExists(backupdir_ + "/shared/00011.sst"));
@@ -558,7 +592,7 @@ TEST_F(BackupableDBTest, NoDoubleCopy) {
   test_backup_env_->GetFileSize(backupdir_ + "/shared/00015.sst", &size);
   ASSERT_EQ(200UL, size);
 
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 }
 
 // Verify that backup works when the database environment is not the same as
@@ -571,7 +605,7 @@ TEST_F(BackupableDBTest, DifferentEnvs) {
   test_db_env_.reset(new TestEnv(mock_env_.get()));
   options_.env = test_db_env_.get();
 
-  OpenBackupableDB(true, true);
+  OpenDBAndBackupEngine(true, true);
 
   // should write 5 DB files + LATEST_BACKUP + one meta file
   test_backup_env_->SetLimitWrittenFiles(7);
@@ -580,9 +614,18 @@ TEST_F(BackupableDBTest, DifferentEnvs) {
   dummy_db_->live_files_ = { "/00010.sst", "/00011.sst",
                              "/CURRENT",   "/MANIFEST-01" };
   dummy_db_->wal_files_ = {{"/00011.log", true}, {"/00012.log", false}};
-  ASSERT_OK(db_->CreateNewBackup(false));
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
 
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
+
+  // try simple backup and verify correctness
+  OpenDBAndBackupEngine(true);
+  FillDB(db_.get(), 0, 100);
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
+  CloseDBAndBackupEngine();
+  DestroyDB(dbname_, Options());
+
+  AssertBackupConsistency(0, 0, 100, 500);
 }
 
 // test various kind of corruptions that may happen:
@@ -599,11 +642,11 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   Random rnd(6);
   Status s;
 
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
   // create five backups
   for (int i = 0; i < 5; ++i) {
     FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(!!(rnd.Next() % 2)));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), !!(rnd.Next() % 2)));
   }
 
   // ---------- case 1. - fail a write -----------
@@ -611,11 +654,11 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   FillDB(db_.get(), keys_iteration * 5, keys_iteration * 6);
   test_backup_env_->SetLimitWrittenFiles(2);
   // should fail
-  s = db_->CreateNewBackup(!!(rnd.Next() % 2));
+  s = backup_engine_->CreateNewBackup(db_.get(), !!(rnd.Next() % 2));
   ASSERT_TRUE(!s.ok());
   test_backup_env_->SetLimitWrittenFiles(1000000);
   // latest backup should have all the keys
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
   AssertBackupConsistency(0, 0, keys_iteration * 5, keys_iteration * 6);
 
   // ---------- case 2. - corrupt/delete latest backup -----------
@@ -624,10 +667,10 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   ASSERT_OK(file_manager_->DeleteFile(backupdir_ + "/LATEST_BACKUP"));
   AssertBackupConsistency(0, 0, keys_iteration * 5);
   // create backup 6, point LATEST_BACKUP to 5
-  OpenBackupableDB();
+  OpenDBAndBackupEngine();
   FillDB(db_.get(), keys_iteration * 5, keys_iteration * 6);
-  ASSERT_OK(db_->CreateNewBackup(false));
-  CloseBackupableDB();
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
+  CloseDBAndBackupEngine();
   ASSERT_OK(file_manager_->WriteToFile(backupdir_ + "/LATEST_BACKUP", "5"));
   AssertBackupConsistency(0, 0, keys_iteration * 5, keys_iteration * 6);
   // assert that all 6 data is gone!
@@ -638,16 +681,16 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   ASSERT_OK(file_manager_->CorruptFile(backupdir_ + "/meta/5", 3));
   // since 5 meta is now corrupted, latest backup should be 4
   AssertBackupConsistency(0, 0, keys_iteration * 4, keys_iteration * 5);
-  OpenRestoreDB();
-  s = restore_db_->RestoreDBFromBackup(5, dbname_, dbname_);
+  OpenBackupEngine();
+  s = backup_engine_->RestoreDBFromBackup(5, dbname_, dbname_);
   ASSERT_TRUE(!s.ok());
-  CloseRestoreDB();
+  CloseBackupEngine();
   ASSERT_OK(file_manager_->DeleteRandomFileInDir(backupdir_ + "/private/4"));
   // 4 is corrupted, 3 is the latest backup now
   AssertBackupConsistency(0, 0, keys_iteration * 3, keys_iteration * 5);
-  OpenRestoreDB();
-  s = restore_db_->RestoreDBFromBackup(4, dbname_, dbname_);
-  CloseRestoreDB();
+  OpenBackupEngine();
+  s = backup_engine_->RestoreDBFromBackup(4, dbname_, dbname_);
+  CloseBackupEngine();
   ASSERT_TRUE(!s.ok());
 
   // --------- case 4. corrupted checksum value ----
@@ -659,9 +702,9 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   // mismatch and abort restore process
   ASSERT_OK(file_manager_->CorruptChecksum(backupdir_ + "/meta/2", true));
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/meta/2"));
-  OpenRestoreDB();
+  OpenBackupEngine();
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/meta/2"));
-  s = restore_db_->RestoreDBFromBackup(2, dbname_, dbname_);
+  s = backup_engine_->RestoreDBFromBackup(2, dbname_, dbname_);
   ASSERT_TRUE(!s.ok());
 
   // make sure that no corrupt backups have actually been deleted!
@@ -677,11 +720,11 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/private/5"));
 
   // delete the corrupt backups and then make sure they're actually deleted
-  ASSERT_OK(restore_db_->DeleteBackup(5));
-  ASSERT_OK(restore_db_->DeleteBackup(4));
-  ASSERT_OK(restore_db_->DeleteBackup(3));
-  ASSERT_OK(restore_db_->DeleteBackup(2));
-  (void) restore_db_->GarbageCollect();
+  ASSERT_OK(backup_engine_->DeleteBackup(5));
+  ASSERT_OK(backup_engine_->DeleteBackup(4));
+  ASSERT_OK(backup_engine_->DeleteBackup(3));
+  ASSERT_OK(backup_engine_->DeleteBackup(2));
+  (void)backup_engine_->GarbageCollect();
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/meta/5") == false);
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/private/5") == false);
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/meta/4") == false);
@@ -691,14 +734,14 @@ TEST_F(BackupableDBTest, CorruptionsTest) {
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/meta/2") == false);
   ASSERT_TRUE(file_manager_->FileExists(backupdir_ + "/private/2") == false);
 
-  CloseRestoreDB();
+  CloseBackupEngine();
   AssertBackupConsistency(0, 0, keys_iteration * 1, keys_iteration * 5);
 
   // new backup should be 2!
-  OpenBackupableDB();
+  OpenDBAndBackupEngine();
   FillDB(db_.get(), keys_iteration * 1, keys_iteration * 2);
-  ASSERT_OK(db_->CreateNewBackup(!!(rnd.Next() % 2)));
-  CloseBackupableDB();
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), !!(rnd.Next() % 2)));
+  CloseDBAndBackupEngine();
   AssertBackupConsistency(2, 0, keys_iteration * 2, keys_iteration * 5);
 }
 
@@ -709,13 +752,13 @@ TEST_F(BackupableDBTest, NoDeleteWithReadOnly) {
   Random rnd(6);
   Status s;
 
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
   // create five backups
   for (int i = 0; i < 5; ++i) {
     FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(!!(rnd.Next() % 2)));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), !!(rnd.Next() % 2)));
   }
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
   ASSERT_OK(file_manager_->WriteToFile(backupdir_ + "/LATEST_BACKUP", "4"));
 
   backupable_options_->destroy_old_data = false;
@@ -756,11 +799,11 @@ TEST_F(BackupableDBTest, OfflineIntegrationTest) {
       // in last iteration, put smaller amount of data,
       int fill_up_to = std::min(keys_iteration * (i + 1), max_key);
       // ---- insert new data and back up ----
-      OpenBackupableDB(destroy_data);
+      OpenDBAndBackupEngine(destroy_data);
       destroy_data = false;
       FillDB(db_.get(), keys_iteration * i, fill_up_to);
-      ASSERT_OK(db_->CreateNewBackup(iter == 0));
-      CloseBackupableDB();
+      ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), iter == 0));
+      CloseDBAndBackupEngine();
       DestroyDB(dbname_, Options());
 
       // ---- make sure it's empty ----
@@ -769,15 +812,15 @@ TEST_F(BackupableDBTest, OfflineIntegrationTest) {
       delete db;
 
       // ---- restore the DB ----
-      OpenRestoreDB();
-      if (i >= 3) { // test purge old backups
+      OpenBackupEngine();
+      if (i >= 3) {  // test purge old backups
         // when i == 4, purge to only 1 backup
         // when i == 3, purge to 2 backups
-        ASSERT_OK(restore_db_->PurgeOldBackups(5 - i));
+        ASSERT_OK(backup_engine_->PurgeOldBackups(5 - i));
       }
       // ---- make sure the data is there ---
       AssertBackupConsistency(0, 0, fill_up_to, max_key);
-      CloseRestoreDB();
+      CloseBackupEngine();
     }
   }
 }
@@ -791,14 +834,12 @@ TEST_F(BackupableDBTest, OnlineIntegrationTest) {
   // delete old data
   DestroyDB(dbname_, Options());
 
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
   // write some data, backup, repeat
   for (int i = 0; i < 5; ++i) {
     if (i == 4) {
       // delete backup number 2, online delete!
-      OpenRestoreDB();
-      ASSERT_OK(restore_db_->DeleteBackup(2));
-      CloseRestoreDB();
+      ASSERT_OK(backup_engine_->DeleteBackup(2));
     }
     // in last iteration, put smaller amount of data,
     // so that backups can share sst files
@@ -806,10 +847,10 @@ TEST_F(BackupableDBTest, OnlineIntegrationTest) {
     FillDB(db_.get(), keys_iteration * i, fill_up_to);
     // we should get consistent results with flush_before_backup
     // set to both true and false
-    ASSERT_OK(db_->CreateNewBackup(!!(rnd.Next() % 2)));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), !!(rnd.Next() % 2)));
   }
   // close and destroy
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
   DestroyDB(dbname_, Options());
 
   // ---- make sure it's empty ----
@@ -818,11 +859,11 @@ TEST_F(BackupableDBTest, OnlineIntegrationTest) {
   delete db;
 
   // ---- restore every backup and verify all the data is there ----
-  OpenRestoreDB();
+  OpenBackupEngine();
   for (int i = 1; i <= 5; ++i) {
     if (i == 2) {
       // we deleted backup 2
-      Status s = restore_db_->RestoreDBFromBackup(2, dbname_, dbname_);
+      Status s = backup_engine_->RestoreDBFromBackup(2, dbname_, dbname_);
       ASSERT_TRUE(!s.ok());
     } else {
       int fill_up_to = std::min(keys_iteration * i, max_key);
@@ -831,11 +872,11 @@ TEST_F(BackupableDBTest, OnlineIntegrationTest) {
   }
 
   // delete some backups -- this should leave only backups 3 and 5 alive
-  ASSERT_OK(restore_db_->DeleteBackup(4));
-  ASSERT_OK(restore_db_->PurgeOldBackups(2));
+  ASSERT_OK(backup_engine_->DeleteBackup(4));
+  ASSERT_OK(backup_engine_->PurgeOldBackups(2));
 
   std::vector<BackupInfo> backup_info;
-  restore_db_->GetBackupInfo(&backup_info);
+  backup_engine_->GetBackupInfo(&backup_info);
   ASSERT_EQ(2UL, backup_info.size());
 
   // check backup 3
@@ -843,30 +884,30 @@ TEST_F(BackupableDBTest, OnlineIntegrationTest) {
   // check backup 5
   AssertBackupConsistency(5, 0, max_key);
 
-  CloseRestoreDB();
+  CloseBackupEngine();
 }
 
 TEST_F(BackupableDBTest, FailOverwritingBackups) {
   options_.write_buffer_size = 1024 * 1024 * 1024;  // 1GB
   // create backups 1, 2, 3, 4, 5
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
   for (int i = 0; i < 5; ++i) {
-    CloseBackupableDB();
+    CloseDBAndBackupEngine();
     DeleteLogFiles();
-    OpenBackupableDB(false);
+    OpenDBAndBackupEngine(false);
     FillDB(db_.get(), 100 * i, 100 * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(true));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
   }
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   // restore 3
-  OpenRestoreDB();
-  ASSERT_OK(restore_db_->RestoreDBFromBackup(3, dbname_, dbname_));
-  CloseRestoreDB();
+  OpenBackupEngine();
+  ASSERT_OK(backup_engine_->RestoreDBFromBackup(3, dbname_, dbname_));
+  CloseBackupEngine();
 
-  OpenBackupableDB(false);
+  OpenDBAndBackupEngine(false);
   FillDB(db_.get(), 0, 300);
-  Status s = db_->CreateNewBackup(true);
+  Status s = backup_engine_->CreateNewBackup(db_.get(), true);
   // the new backup fails because new table files
   // clash with old table files from backups 4 and 5
   // (since write_buffer_size is huge, we can be sure that
@@ -874,21 +915,21 @@ TEST_F(BackupableDBTest, FailOverwritingBackups) {
   // a file generated by a new backup is the same as
   // sst file generated by backup 4)
   ASSERT_TRUE(s.IsCorruption());
-  ASSERT_OK(db_->DeleteBackup(4));
-  ASSERT_OK(db_->DeleteBackup(5));
+  ASSERT_OK(backup_engine_->DeleteBackup(4));
+  ASSERT_OK(backup_engine_->DeleteBackup(5));
   // now, the backup can succeed
-  ASSERT_OK(db_->CreateNewBackup(true));
-  CloseBackupableDB();
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
+  CloseDBAndBackupEngine();
 }
 
 TEST_F(BackupableDBTest, NoShareTableFiles) {
   const int keys_iteration = 5000;
-  OpenBackupableDB(true, false, false);
+  OpenDBAndBackupEngine(true, false, false);
   for (int i = 0; i < 5; ++i) {
     FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(!!(i % 2)));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), !!(i % 2)));
   }
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   for (int i = 0; i < 5; ++i) {
     AssertBackupConsistency(i + 1, 0, keys_iteration * (i + 1),
@@ -899,12 +940,12 @@ TEST_F(BackupableDBTest, NoShareTableFiles) {
 // Verify that you can backup and restore with share_files_with_checksum on
 TEST_F(BackupableDBTest, ShareTableFilesWithChecksums) {
   const int keys_iteration = 5000;
-  OpenBackupableDB(true, false, true, true);
+  OpenDBAndBackupEngine(true, false, true, true);
   for (int i = 0; i < 5; ++i) {
     FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(!!(i % 2)));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), !!(i % 2)));
   }
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   for (int i = 0; i < 5; ++i) {
     AssertBackupConsistency(i + 1, 0, keys_iteration * (i + 1),
@@ -917,12 +958,12 @@ TEST_F(BackupableDBTest, ShareTableFilesWithChecksums) {
 TEST_F(BackupableDBTest, ShareTableFilesWithChecksumsTransition) {
   const int keys_iteration = 5000;
   // set share_files_with_checksum to false
-  OpenBackupableDB(true, false, true, false);
+  OpenDBAndBackupEngine(true, false, true, false);
   for (int i = 0; i < 5; ++i) {
     FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(true));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
   }
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   for (int i = 0; i < 5; ++i) {
     AssertBackupConsistency(i + 1, 0, keys_iteration * (i + 1),
@@ -930,12 +971,12 @@ TEST_F(BackupableDBTest, ShareTableFilesWithChecksumsTransition) {
   }
 
   // set share_files_with_checksum to true and do some more backups
-  OpenBackupableDB(true, false, true, true);
+  OpenDBAndBackupEngine(true, false, true, true);
   for (int i = 5; i < 10; ++i) {
     FillDB(db_.get(), keys_iteration * i, keys_iteration * (i + 1));
-    ASSERT_OK(db_->CreateNewBackup(true));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
   }
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   for (int i = 0; i < 5; ++i) {
     AssertBackupConsistency(i + 1, 0, keys_iteration * (i + 5 + 1),
@@ -944,8 +985,8 @@ TEST_F(BackupableDBTest, ShareTableFilesWithChecksumsTransition) {
 }
 
 TEST_F(BackupableDBTest, DeleteTmpFiles) {
-  OpenBackupableDB();
-  CloseBackupableDB();
+  OpenDBAndBackupEngine();
+  CloseDBAndBackupEngine();
   std::string shared_tmp = backupdir_ + "/shared/00006.sst.tmp";
   std::string private_tmp_dir = backupdir_ + "/private/10.tmp";
   std::string private_tmp_file = private_tmp_dir + "/00003.sst";
@@ -953,10 +994,10 @@ TEST_F(BackupableDBTest, DeleteTmpFiles) {
   file_manager_->CreateDir(private_tmp_dir);
   file_manager_->WriteToFile(private_tmp_file, "tmp");
   ASSERT_TRUE(file_manager_->FileExists(private_tmp_dir));
-  OpenBackupableDB();
+  OpenDBAndBackupEngine();
   // Need to call this explicitly to delete tmp files
-  (void) db_->GarbageCollect();
-  CloseBackupableDB();
+  (void)backup_engine_->GarbageCollect();
+  CloseDBAndBackupEngine();
   ASSERT_FALSE(file_manager_->FileExists(shared_tmp));
   ASSERT_FALSE(file_manager_->FileExists(private_tmp_file));
   ASSERT_FALSE(file_manager_->FileExists(private_tmp_dir));
@@ -966,18 +1007,18 @@ TEST_F(BackupableDBTest, KeepLogFiles) {
   backupable_options_->backup_log_files = false;
   // basically infinite
   options_.WAL_ttl_seconds = 24 * 60 * 60;
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
   FillDB(db_.get(), 0, 100);
   ASSERT_OK(db_->Flush(FlushOptions()));
   FillDB(db_.get(), 100, 200);
-  ASSERT_OK(db_->CreateNewBackup(false));
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
   FillDB(db_.get(), 200, 300);
   ASSERT_OK(db_->Flush(FlushOptions()));
   FillDB(db_.get(), 300, 400);
   ASSERT_OK(db_->Flush(FlushOptions()));
   FillDB(db_.get(), 400, 500);
   ASSERT_OK(db_->Flush(FlushOptions()));
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   // all data should be there if we call with keep_log_files = true
   AssertBackupConsistency(0, 0, 500, 600, true);
@@ -999,23 +1040,23 @@ TEST_F(BackupableDBTest, RateLimiting) {
     // rate-limiting backups must be single-threaded
     backupable_options_->max_background_operations = 1;
     options_.compression = kNoCompression;
-    OpenBackupableDB(true);
+    OpenDBAndBackupEngine(true);
     size_t bytes_written = FillDB(db_.get(), 0, 100000);
 
     auto start_backup = env_->NowMicros();
-    ASSERT_OK(db_->CreateNewBackup(false));
+    ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), false));
     auto backup_time = env_->NowMicros() - start_backup;
     auto rate_limited_backup_time = (bytes_written * kMicrosPerSec) /
                                     backupable_options_->backup_rate_limit;
     ASSERT_GT(backup_time, 0.8 * rate_limited_backup_time);
 
-    CloseBackupableDB();
+    CloseDBAndBackupEngine();
 
-    OpenRestoreDB();
+    OpenBackupEngine();
     auto start_restore = env_->NowMicros();
-    ASSERT_OK(restore_db_->RestoreDBFromLatestBackup(dbname_, dbname_));
+    ASSERT_OK(backup_engine_->RestoreDBFromLatestBackup(dbname_, dbname_));
     auto restore_time = env_->NowMicros() - start_restore;
-    CloseRestoreDB();
+    CloseBackupEngine();
     auto rate_limited_restore_time = (bytes_written * kMicrosPerSec) /
                                      backupable_options_->restore_rate_limit;
     ASSERT_GT(restore_time, 0.8 * rate_limited_restore_time);
@@ -1026,12 +1067,12 @@ TEST_F(BackupableDBTest, RateLimiting) {
 
 TEST_F(BackupableDBTest, ReadOnlyBackupEngine) {
   DestroyDB(dbname_, Options());
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
   FillDB(db_.get(), 0, 100);
-  ASSERT_OK(db_->CreateNewBackup(true));
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
   FillDB(db_.get(), 100, 200);
-  ASSERT_OK(db_->CreateNewBackup(true));
-  CloseBackupableDB();
+  ASSERT_OK(backup_engine_->CreateNewBackup(db_.get(), true));
+  CloseDBAndBackupEngine();
   DestroyDB(dbname_, Options());
 
   backupable_options_->destroy_old_data = false;
@@ -1058,7 +1099,7 @@ TEST_F(BackupableDBTest, ReadOnlyBackupEngine) {
 
 TEST_F(BackupableDBTest, GarbageCollectionBeforeBackup) {
   DestroyDB(dbname_, Options());
-  OpenBackupableDB(true);
+  OpenDBAndBackupEngine(true);
 
   env_->CreateDirIfMissing(backupdir_ + "/shared");
   std::string file_five = backupdir_ + "/shared/000005.sst";
@@ -1068,23 +1109,60 @@ TEST_F(BackupableDBTest, GarbageCollectionBeforeBackup) {
 
   FillDB(db_.get(), 0, 100);
   // backup overwrites file 000005.sst
-  ASSERT_TRUE(db_->CreateNewBackup(true).ok());
+  ASSERT_TRUE(backup_engine_->CreateNewBackup(db_.get(), true).ok());
 
   std::string new_file_five_contents;
   ASSERT_OK(ReadFileToString(env_, file_five, &new_file_five_contents));
   // file 000005.sst was overwritten
   ASSERT_TRUE(new_file_five_contents != file_five_contents);
 
-  CloseBackupableDB();
+  CloseDBAndBackupEngine();
 
   AssertBackupConsistency(0, 0, 100);
 }
 
+// Test that we properly propagate Env failures
+TEST_F(BackupableDBTest, EnvFailures) {
+  BackupEngine* backup_engine;
+
+  // get children failure
+  {
+    test_backup_env_->SetGetChildrenFailure(true);
+    ASSERT_NOK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                  &backup_engine));
+    test_backup_env_->SetGetChildrenFailure(false);
+  }
+
+  // created dir failure
+  {
+    test_backup_env_->SetCreateDirIfMissingFailure(true);
+    ASSERT_NOK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                  &backup_engine));
+    test_backup_env_->SetCreateDirIfMissingFailure(false);
+  }
+
+  // new directory failure
+  {
+    test_backup_env_->SetNewDirectoryFailure(true);
+    ASSERT_NOK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                  &backup_engine));
+    test_backup_env_->SetNewDirectoryFailure(false);
+  }
+
+  // no failure
+  {
+    ASSERT_OK(BackupEngine::Open(test_db_env_.get(), *backupable_options_,
+                                 &backup_engine));
+    delete backup_engine;
+  }
+}
+
 }  // anon namespace
 
 } //  namespace rocksdb
 
 int main(int argc, char** argv) {
+  rocksdb::port::InstallStackTraceHandler();
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();
 }