prototype status check enforcement (#6798)

Summary: Tried making Status object enforce that it is checked in some way. In cases it is not checked, `PermitUncheckedError()` must be called explicitly. Added a way to run tests (`ASSERT_STATUS_CHECKED=1 make -j48 check`) on a whitelist. The effort appears significant to get each test to pass with this assertion, so I only fixed up enough to get one test (`options_test`) working and added it to the whitelist. Pull Request resolved: https://github.com/facebook/rocksdb/pull/6798 Reviewed By: pdillinger Differential Revision: D21377404 Pulled By: ajkr fbshipit-source-id: 73236f9c8df38f01cf24ecac4a6d1661b72d077e
2020-05-08 12:36:49 -07:00 · 2020-05-08 12:36:49 -07:00 · 1c84660457
parent 12825894a2
commit 1c84660457
11 changed files with 389 additions and 125 deletions
--- a/14
+++ b/14
@ -166,6 +166,12 @@ else
 	CXXFLAGS += -fno-rtti
 endif

+ifdef ASSERT_STATUS_CHECKED
+ifeq ($(filter -DROCKSDB_ASSERT_STATUS_CHECKED,$(OPT)),)
+	OPT += -DROCKSDB_ASSERT_STATUS_CHECKED
+endif
+endif
+
 $(warning Warning: Compiling in debug mode. Don't use the resulting binary in production)
 endif

@ -657,6 +663,14 @@ PARALLEL_TEST = \
 ifdef COMPILE_WITH_UBSAN
        TESTS := $(shell echo $(TESTS) | sed 's/\boptions_settable_test\b//g')
 endif
+ifdef ASSERT_STATUS_CHECKED
+	# This is a new check for which we will add support incrementally. The
+	# whitelist can be removed once support is fully added.
+        TESTS_WHITELIST = \
+		options_test
+        TESTS := $(filter $(TESTS_WHITELIST),$(TESTS))
+        PARALLEL_TEST := $(filter $(TESTS_WHITELIST),$(PARALLEL_TEST))
+endif
 SUBSET := $(TESTS)
 ifdef ROCKSDBTESTS_START
        SUBSET := $(shell echo $(SUBSET) | sed 's/^.*$(ROCKSDBTESTS_START)/$(ROCKSDBTESTS_START)/')
--- a/file/writable_file_writer.h
+++ b/file/writable_file_writer.h
@ -46,6 +46,7 @@ class WritableFileWriter {
    for (auto& listener : listeners_) {
      listener->OnFileWriteFinish(info);
    }
+    info.status.PermitUncheckedError();
  }
 #endif  // ROCKSDB_LITE

@ -126,7 +127,10 @@ class WritableFileWriter {

  WritableFileWriter& operator=(const WritableFileWriter&) = delete;

-  ~WritableFileWriter() { Close(); }
+  ~WritableFileWriter() {
+    auto s = Close();
+    s.PermitUncheckedError();
+  }

  std::string file_name() const { return file_name_; }

--- a/hdfs/env_hdfs.h
+++ b/hdfs/env_hdfs.h
@ -238,8 +238,6 @@ class HdfsEnv : public Env {

 namespace ROCKSDB_NAMESPACE {

-static const Status notsup;
-
 class HdfsEnv : public Env {

 public:
@ -260,79 +258,81 @@ class HdfsEnv : public Env {
      const std::string& /*fname*/,
      std::unique_ptr<RandomAccessFile>* /*result*/,
      const EnvOptions& /*options*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status NewWritableFile(const std::string& /*fname*/,
                                 std::unique_ptr<WritableFile>* /*result*/,
                                 const EnvOptions& /*options*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status NewDirectory(const std::string& /*name*/,
                              std::unique_ptr<Directory>* /*result*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status FileExists(const std::string& /*fname*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status GetChildren(const std::string& /*path*/,
                             std::vector<std::string>* /*result*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status DeleteFile(const std::string& /*fname*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status CreateDir(const std::string& /*name*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status CreateDirIfMissing(const std::string& /*name*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status DeleteDir(const std::string& /*name*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status GetFileSize(const std::string& /*fname*/,
                             uint64_t* /*size*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status GetFileModificationTime(const std::string& /*fname*/,
                                         uint64_t* /*time*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status RenameFile(const std::string& /*src*/,
                            const std::string& /*target*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status LinkFile(const std::string& /*src*/,
                          const std::string& /*target*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status LockFile(const std::string& /*fname*/,
                          FileLock** /*lock*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

-  virtual Status UnlockFile(FileLock* /*lock*/) override { return notsup; }
+  virtual Status UnlockFile(FileLock* /*lock*/) override {
+    return Status::NotSupported();
+  }

  virtual Status NewLogger(const std::string& /*fname*/,
                           std::shared_ptr<Logger>* /*result*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  Status IsDirectory(const std::string& /*path*/, bool* /*is_dir*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual void Schedule(void (* /*function*/)(void* arg), void* /*arg*/,
@ -352,7 +352,7 @@ class HdfsEnv : public Env {
  }

  virtual Status GetTestDirectory(std::string* /*path*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual uint64_t NowMicros() override { return 0; }
@ -360,16 +360,16 @@ class HdfsEnv : public Env {
  virtual void SleepForMicroseconds(int /*micros*/) override {}

  virtual Status GetHostName(char* /*name*/, uint64_t /*len*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status GetCurrentTime(int64_t* /*unix_time*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual Status GetAbsolutePath(const std::string& /*db_path*/,
                                 std::string* /*outputpath*/) override {
-    return notsup;
+    return Status::NotSupported();
  }

  virtual void SetBackgroundThreads(int /*number*/,
--- a/include/rocksdb/io_status.h
+++ b/include/rocksdb/io_status.h
@ -170,6 +170,9 @@ inline IOStatus::IOStatus(Code _code, SubCode _subcode, const Slice& msg,
 }

 inline IOStatus::IOStatus(const IOStatus& s) : Status(s.code_, s.subcode_) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  retryable_ = s.retryable_;
  data_loss_ = s.data_loss_;
  scope_ = s.scope_;
@ -179,6 +182,10 @@ inline IOStatus& IOStatus::operator=(const IOStatus& s) {
  // The following condition catches both aliasing (when this == &s),
  // and the common case where both s and *this are ok.
  if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    code_ = s.code_;
    subcode_ = s.subcode_;
    retryable_ = s.retryable_;
@ -204,6 +211,10 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
 #endif
 {
  if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    code_ = std::move(s.code_);
    s.code_ = kOk;
    subcode_ = std::move(s.subcode_);
@ -211,7 +222,7 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
    retryable_ = s.retryable_;
    data_loss_ = s.data_loss_;
    scope_ = s.scope_;
-    scope_ = kIOErrorScopeFileSystem;
+    s.scope_ = kIOErrorScopeFileSystem;
    delete[] state_;
    state_ = nullptr;
    std::swap(state_, s.state_);
@ -220,10 +231,18 @@ inline IOStatus& IOStatus::operator=(IOStatus&& s)
 }

 inline bool IOStatus::operator==(const IOStatus& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  return (code_ == rhs.code_);
 }

 inline bool IOStatus::operator!=(const IOStatus& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  return !(*this == rhs);
 }

--- a/include/rocksdb/status.h
+++ b/include/rocksdb/status.h
@ -25,7 +25,12 @@ class Status {
 public:
  // Create a success status.
  Status() : code_(kOk), subcode_(kNone), sev_(kNoError), state_(nullptr) {}
-  ~Status() { delete[] state_; }
+  ~Status() {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    assert(checked_);
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    delete[] state_;
+  }

  // Copy the specified status.
  Status(const Status& s);
@ -43,6 +48,15 @@ class Status {
  bool operator==(const Status& rhs) const;
  bool operator!=(const Status& rhs) const;

+  // In case of intentionally swallowing an error, user must explicitly call
+  // this function. That way we are easily able to search the code to find where
+  // error swallowing occurs.
+  void PermitUncheckedError() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+  }
+
  enum Code : unsigned char {
    kOk = 0,
    kNotFound = 1,
@ -63,7 +77,12 @@ class Status {
    kMaxCode
  };

-  Code code() const { return code_; }
+  Code code() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code_;
+  }

  enum SubCode : unsigned char {
    kNone = 0,
@ -83,7 +102,12 @@ class Status {
    kMaxSubCode
  };

-  SubCode subcode() const { return subcode_; }
+  SubCode subcode() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return subcode_;
+  }

  enum Severity : unsigned char {
    kNoError = 0,
@ -95,10 +119,20 @@ class Status {
  };

  Status(const Status& s, Severity sev);
-  Severity severity() const { return sev_; }
+  Severity severity() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return sev_;
+  }

  // Returns a C style string indicating the message of the Status
-  const char* getState() const { return state_; }
+  const char* getState() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return state_;
+  }

  // Return a success status.
  static Status OK() { return Status(); }
@ -233,65 +267,156 @@ class Status {
  }

  // Returns true iff the status indicates success.
-  bool ok() const { return code() == kOk; }
+  bool ok() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kOk;
+  }

  // Returns true iff the status indicates success *with* something
  // overwritten
  bool IsOkOverwritten() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return code() == kOk && subcode() == kOverwritten;
  }

  // Returns true iff the status indicates a NotFound error.
-  bool IsNotFound() const { return code() == kNotFound; }
+  bool IsNotFound() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kNotFound;
+  }

  // Returns true iff the status indicates a Corruption error.
-  bool IsCorruption() const { return code() == kCorruption; }
+  bool IsCorruption() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kCorruption;
+  }

  // Returns true iff the status indicates a NotSupported error.
-  bool IsNotSupported() const { return code() == kNotSupported; }
+  bool IsNotSupported() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kNotSupported;
+  }

  // Returns true iff the status indicates an InvalidArgument error.
-  bool IsInvalidArgument() const { return code() == kInvalidArgument; }
+  bool IsInvalidArgument() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kInvalidArgument;
+  }

  // Returns true iff the status indicates an IOError.
-  bool IsIOError() const { return code() == kIOError; }
+  bool IsIOError() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kIOError;
+  }

  // Returns true iff the status indicates an MergeInProgress.
-  bool IsMergeInProgress() const { return code() == kMergeInProgress; }
+  bool IsMergeInProgress() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kMergeInProgress;
+  }

  // Returns true iff the status indicates Incomplete
-  bool IsIncomplete() const { return code() == kIncomplete; }
+  bool IsIncomplete() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kIncomplete;
+  }

  // Returns true iff the status indicates Shutdown In progress
-  bool IsShutdownInProgress() const { return code() == kShutdownInProgress; }
+  bool IsShutdownInProgress() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kShutdownInProgress;
+  }

-  bool IsTimedOut() const { return code() == kTimedOut; }
+  bool IsTimedOut() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kTimedOut;
+  }

-  bool IsAborted() const { return code() == kAborted; }
+  bool IsAborted() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kAborted;
+  }

  bool IsLockLimit() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return code() == kAborted && subcode() == kLockLimit;
  }

  // Returns true iff the status indicates that a resource is Busy and
  // temporarily could not be acquired.
-  bool IsBusy() const { return code() == kBusy; }
+  bool IsBusy() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kBusy;
+  }

-  bool IsDeadlock() const { return code() == kBusy && subcode() == kDeadlock; }
+  bool IsDeadlock() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kBusy && subcode() == kDeadlock;
+  }

  // Returns true iff the status indicated that the operation has Expired.
-  bool IsExpired() const { return code() == kExpired; }
+  bool IsExpired() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kExpired;
+  }

  // Returns true iff the status indicates a TryAgain error.
  // This usually means that the operation failed, but may succeed if
  // re-attempted.
-  bool IsTryAgain() const { return code() == kTryAgain; }
+  bool IsTryAgain() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kTryAgain;
+  }

  // Returns true iff the status indicates the proposed compaction is too large
-  bool IsCompactionTooLarge() const { return code() == kCompactionTooLarge; }
+  bool IsCompactionTooLarge() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kCompactionTooLarge;
+  }

  // Returns true iff the status indicates Column Family Dropped
-  bool IsColumnFamilyDropped() const { return code() == kColumnFamilyDropped; }
+  bool IsColumnFamilyDropped() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
+    return code() == kColumnFamilyDropped;
+  }

  // Returns true iff the status indicates a NoSpace error
  // This is caused by an I/O error returning the specific "out of space"
@ -299,6 +424,9 @@ class Status {
  // with a specific subcode, enabling users to take the appropriate action
  // if needed
  bool IsNoSpace() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return (code() == kIOError) && (subcode() == kNoSpace);
  }

@ -306,6 +434,9 @@ class Status {
  // cases where we limit the memory used in certain operations (eg. the size
  // of a write batch) in order to avoid out of memory exceptions.
  bool IsMemoryLimit() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return (code() == kAborted) && (subcode() == kMemoryLimit);
  }

@ -314,17 +445,26 @@ class Status {
  // directory" error condition. A PathNotFound error is an I/O error with
  // a specific subcode, enabling users to take appropriate action if necessary
  bool IsPathNotFound() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return (code() == kIOError) && (subcode() == kPathNotFound);
  }

  // Returns true iff the status indicates manual compaction paused. This
  // is caused by a call to PauseManualCompaction
  bool IsManualCompactionPaused() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return (code() == kIncomplete) && (subcode() == kManualCompactionPaused);
  }

  // Returns true iff the status indicates a TxnNotPrepared error.
  bool IsTxnNotPrepared() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    return (code() == kInvalidArgument) && (subcode() == kTxnNotPrepared);
  }

@ -342,6 +482,9 @@ class Status {
  SubCode subcode_;
  Severity sev_;
  const char* state_;
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  mutable bool checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED

  explicit Status(Code _code, SubCode _subcode = kNone)
      : code_(_code), subcode_(_subcode), sev_(kNoError), state_(nullptr) {}
@ -355,16 +498,26 @@ class Status {

 inline Status::Status(const Status& s)
    : code_(s.code_), subcode_(s.subcode_), sev_(s.sev_) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  state_ = (s.state_ == nullptr) ? nullptr : CopyState(s.state_);
 }
 inline Status::Status(const Status& s, Severity sev)
    : code_(s.code_), subcode_(s.subcode_), sev_(sev) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  state_ = (s.state_ == nullptr) ? nullptr : CopyState(s.state_);
 }
 inline Status& Status::operator=(const Status& s) {
  // The following condition catches both aliasing (when this == &s),
  // and the common case where both s and *this are ok.
  if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    code_ = s.code_;
    subcode_ = s.subcode_;
    sev_ = s.sev_;
@ -379,6 +532,9 @@ inline Status::Status(Status&& s)
    noexcept
 #endif
    : Status() {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  s.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  *this = std::move(s);
 }

@ -388,6 +544,10 @@ inline Status& Status::operator=(Status&& s)
 #endif
 {
  if (this != &s) {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+    s.checked_ = true;
+    checked_ = false;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
    code_ = std::move(s.code_);
    s.code_ = kOk;
    subcode_ = std::move(s.subcode_);
@ -402,10 +562,18 @@ inline Status& Status::operator=(Status&& s)
 }

 inline bool Status::operator==(const Status& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  return (code_ == rhs.code_);
 }

 inline bool Status::operator!=(const Status& rhs) const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+  rhs.checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  return !(*this == rhs);
 }

--- a/options/cf_options.cc
+++ b/options/cf_options.cc
@ -459,8 +459,9 @@ std::unordered_map<std::string, OptionTypeInfo>
          [](const ConfigOptions& /*opts*/, const std::string& /*name*/,
             const std::string& value, char* addr) {
            auto mop = reinterpret_cast<std::shared_ptr<MergeOperator>*>(addr);
-            ObjectRegistry::NewInstance()->NewSharedObject<MergeOperator>(value,
-                                                                          mop);
+            ObjectRegistry::NewInstance()
+                ->NewSharedObject<MergeOperator>(value, mop)
+                .PermitUncheckedError();
            return Status::OK();
          }}},
        {"compaction_style",
--- a/options/options_helper.cc
+++ b/options/options_helper.cc
@ -1135,7 +1135,7 @@ Status GetTableFactoryFromMap(
      return s;
    }
    table_factory->reset(new BlockBasedTableFactory(bbt_opt));
-    return Status::OK();
+    return s;
  } else if (factory_name == PlainTableFactory().Name()) {
    PlainTableOptions pt_opt;
    s = GetPlainTableOptionsFromMap(config_options, PlainTableOptions(),
@ -1144,12 +1144,12 @@ Status GetTableFactoryFromMap(
      return s;
    }
    table_factory->reset(new PlainTableFactory(pt_opt));
-    return Status::OK();
+    return s;
  }
  // Return OK for not supported table factories as TableFactory
  // Deserialization is optional.
  table_factory->reset();
-  return Status::OK();
+  return s;
 }

 std::unordered_map<std::string, EncodingType>
@ -1306,9 +1306,9 @@ Status OptionTypeInfo::SerializeOption(const ConfigOptions& config_options,
  // we skip it in the serialization.
  Status s;
  if (opt_addr == nullptr || IsDeprecated()) {
-    return Status::OK();
+    s = Status::OK();
  } else if (string_func != nullptr) {
-    return string_func(config_options, opt_name, opt_addr, opt_value);
+    s = string_func(config_options, opt_name, opt_addr, opt_value);
  } else if (SerializeSingleOptionHelper(opt_addr, type, opt_value)) {
    s = Status::OK();
  } else {
--- a/options/options_parser.cc
+++ b/options/options_parser.cc
@ -75,55 +75,68 @@ Status PersistRocksDBOptions(const ConfigOptions& config_options_in,

  std::string options_file_content;

-  writable->Append(option_file_header + "[" +
-                   opt_section_titles[kOptionSectionVersion] +
-                   "]\n"
-                   "  rocksdb_version=" +
-                   ToString(ROCKSDB_MAJOR) + "." + ToString(ROCKSDB_MINOR) +
-                   "." + ToString(ROCKSDB_PATCH) + "\n");
-  writable->Append("  options_file_version=" +
-                   ToString(ROCKSDB_OPTION_FILE_MAJOR) + "." +
-                   ToString(ROCKSDB_OPTION_FILE_MINOR) + "\n");
-  writable->Append("\n[" + opt_section_titles[kOptionSectionDBOptions] +
-                   "]\n  ");
-
-  s = GetStringFromDBOptions(config_options, db_opt, &options_file_content);
-  if (!s.ok()) {
-    writable->Close();
-    return s;
+  s = writable->Append(option_file_header + "[" +
+                       opt_section_titles[kOptionSectionVersion] +
+                       "]\n"
+                       "  rocksdb_version=" +
+                       ToString(ROCKSDB_MAJOR) + "." + ToString(ROCKSDB_MINOR) +
+                       "." + ToString(ROCKSDB_PATCH) + "\n");
+  if (s.ok()) {
+    s = writable->Append(
+        "  options_file_version=" + ToString(ROCKSDB_OPTION_FILE_MAJOR) + "." +
+        ToString(ROCKSDB_OPTION_FILE_MINOR) + "\n");
+  }
+  if (s.ok()) {
+    s = writable->Append("\n[" + opt_section_titles[kOptionSectionDBOptions] +
+                         "]\n  ");
  }
-  writable->Append(options_file_content + "\n");

-  for (size_t i = 0; i < cf_opts.size(); ++i) {
+  if (s.ok()) {
+    s = GetStringFromDBOptions(config_options, db_opt, &options_file_content);
+  }
+  if (s.ok()) {
+    s = writable->Append(options_file_content + "\n");
+  }
+
+  for (size_t i = 0; s.ok() && i < cf_opts.size(); ++i) {
    // CFOptions section
-    writable->Append("\n[" + opt_section_titles[kOptionSectionCFOptions] +
-                     " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
-    s = GetStringFromColumnFamilyOptions(config_options, cf_opts[i],
-                                         &options_file_content);
-    if (!s.ok()) {
-      writable->Close();
-      return s;
+    s = writable->Append("\n[" + opt_section_titles[kOptionSectionCFOptions] +
+                         " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
+    if (s.ok()) {
+      s = GetStringFromColumnFamilyOptions(config_options, cf_opts[i],
+                                           &options_file_content);
+    }
+    if (s.ok()) {
+      s = writable->Append(options_file_content + "\n");
    }
-    writable->Append(options_file_content + "\n");
    // TableOptions section
    auto* tf = cf_opts[i].table_factory.get();
    if (tf != nullptr) {
-      writable->Append("[" + opt_section_titles[kOptionSectionTableOptions] +
-                       tf->Name() + " \"" + EscapeOptionString(cf_names[i]) +
-                       "\"]\n  ");
-      options_file_content.clear();
-      s = tf->GetOptionString(config_options, &options_file_content);
-      if (!s.ok()) {
-        return s;
+      if (s.ok()) {
+        s = writable->Append(
+            "[" + opt_section_titles[kOptionSectionTableOptions] + tf->Name() +
+            " \"" + EscapeOptionString(cf_names[i]) + "\"]\n  ");
+      }
+      if (s.ok()) {
+        options_file_content.clear();
+        s = tf->GetOptionString(config_options, &options_file_content);
+      }
+      if (s.ok()) {
+        s = writable->Append(options_file_content + "\n");
      }
-      writable->Append(options_file_content + "\n");
    }
  }
-  writable->Sync(true /* use_fsync */);
-  writable->Close();
-
-  return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
-      config_options, db_opt, cf_names, cf_opts, file_name, fs);
+  if (s.ok()) {
+    s = writable->Sync(true /* use_fsync */);
+  }
+  if (s.ok()) {
+    s = writable->Close();
+  }
+  if (s.ok()) {
+    return RocksDBOptionsParser::VerifyRocksDBOptionsFromFile(
+        config_options, db_opt, cf_names, cf_opts, file_name, fs);
+  }
+  return s;
 }

 RocksDBOptionsParser::RocksDBOptionsParser() { Reset(); }
@ -464,7 +477,7 @@ Status RocksDBOptionsParser::EndSection(
      }
    }
  }
-  return Status::OK();
+  return s;
 }

 Status RocksDBOptionsParser::ValidityCheck() {
@ -609,15 +622,35 @@ Status RocksDBOptionsParser::VerifyDBOptions(
        char buffer[kBufferSize];
        std::string base_value;
        std::string file_value;
-        opt_info.SerializeOption(config_options, pair.first, base_addr,
-                                 &base_value);
-        opt_info.SerializeOption(config_options, pair.first, file_addr,
-                                 &file_value);
+        int offset =
+            snprintf(buffer, sizeof(buffer),
+                     "[RocksDBOptionsParser]: "
+                     "failed the verification on ColumnFamilyOptions::%s",
+                     pair.first.c_str());
+        Status s = opt_info.SerializeOption(config_options, pair.first,
+                                            base_addr, &base_value);
+        if (s.ok()) {
+          s = opt_info.SerializeOption(config_options, pair.first, file_addr,
+                                       &file_value);
+        }
        snprintf(buffer, sizeof(buffer),
                 "[RocksDBOptionsParser]: "
                 "failed the verification on DBOptions::%s --- "
                 "The specified one is %s while the persisted one is %s.\n",
                 pair.first.c_str(), base_value.c_str(), file_value.c_str());
+        assert(offset >= 0);
+        assert(static_cast<size_t>(offset) < sizeof(buffer));
+        if (s.ok()) {
+          snprintf(
+              buffer + offset, sizeof(buffer) - static_cast<size_t>(offset),
+              "--- The specified one is %s while the persisted one is %s.\n",
+              base_value.c_str(), file_value.c_str());
+        } else {
+          snprintf(buffer + offset,
+                   sizeof(buffer) - static_cast<size_t>(offset),
+                   "--- Unable to re-serialize an option: %s.\n",
+                   s.ToString().c_str());
+        }
        return Status::InvalidArgument(Slice(buffer, strlen(buffer)));
      }
    }
@ -659,15 +692,30 @@ Status RocksDBOptionsParser::VerifyCFOptions(
        char buffer[kBufferSize];
        std::string base_value;
        std::string file_value;
-        opt_info.SerializeOption(config_options, pair.first, base_addr,
-                                 &base_value);
-        opt_info.SerializeOption(config_options, pair.first, file_addr,
-                                 &file_value);
-        snprintf(buffer, sizeof(buffer),
-                 "[RocksDBOptionsParser]: "
-                 "failed the verification on ColumnFamilyOptions::%s --- "
-                 "The specified one is %s while the persisted one is %s.\n",
-                 pair.first.c_str(), base_value.c_str(), file_value.c_str());
+        Status s = opt_info.SerializeOption(config_options, pair.first,
+                                            base_addr, &base_value);
+        if (s.ok()) {
+          s = opt_info.SerializeOption(config_options, pair.first, file_addr,
+                                       &file_value);
+        }
+        int offset =
+            snprintf(buffer, sizeof(buffer),
+                     "[RocksDBOptionsParser]: "
+                     "failed the verification on ColumnFamilyOptions::%s",
+                     pair.first.c_str());
+        assert(offset >= 0);
+        assert(static_cast<size_t>(offset) < sizeof(buffer));
+        if (s.ok()) {
+          snprintf(
+              buffer + offset, sizeof(buffer) - static_cast<size_t>(offset),
+              "--- The specified one is %s while the persisted one is %s.\n",
+              base_value.c_str(), file_value.c_str());
+        } else {
+          snprintf(buffer + offset,
+                   sizeof(buffer) - static_cast<size_t>(offset),
+                   "--- Unable to re-serialize an option: %s.\n",
+                   s.ToString().c_str());
+        }
        return Status::InvalidArgument(Slice(buffer, sizeof(buffer)));
      }  // if (! matches)
    }    // CheckSanityLevel
--- a/options/options_test.cc
+++ b/options/options_test.cc
@ -2101,7 +2101,7 @@ TEST_F(OptionsParserTest, Comment) {
      "  # if a section is blank, we will use the default\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_OK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2132,7 +2132,7 @@ TEST_F(OptionsParserTest, ExtraSpace) {
      "  # if a section is blank, we will use the default\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_OK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2150,7 +2150,7 @@ TEST_F(OptionsParserTest, MissingDBOptions) {
      "  # if a section is blank, we will use the default\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_NOK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2180,7 +2180,7 @@ TEST_F(OptionsParserTest, DoubleDBOptions) {
      "  # if a section is blank, we will use the default\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_NOK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2208,7 +2208,7 @@ TEST_F(OptionsParserTest, NoDefaultCFOptions) {
      "  # if a section is blank, we will use the default\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_NOK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2238,7 +2238,7 @@ TEST_F(OptionsParserTest, DefaultCFOptionsMustBeTheFirst) {
      "  # if a section is blank, we will use the default\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_NOK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2267,7 +2267,7 @@ TEST_F(OptionsParserTest, DuplicateCFOptions) {
      "[CFOptions \"something_else\"]\n";

  const std::string kTestFileName = "test-rocksdb-options.ini";
-  env_->WriteToNewFile(kTestFileName, options_file_content);
+  ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
  RocksDBOptionsParser parser;
  ASSERT_NOK(
      parser.Parse(kTestFileName, fs_.get(), false, 4096 /* readahead_size */));
@ -2335,8 +2335,12 @@ TEST_F(OptionsParserTest, IgnoreUnknownOptions) {
        "  # if a section is blank, we will use the default\n";

    const std::string kTestFileName = "test-rocksdb-options.ini";
-    env_->DeleteFile(kTestFileName);
-    env_->WriteToNewFile(kTestFileName, options_file_content);
+    auto s = env_->FileExists(kTestFileName);
+    ASSERT_TRUE(s.ok() || s.IsNotFound());
+    if (s.ok()) {
+      ASSERT_OK(env_->DeleteFile(kTestFileName));
+    }
+    ASSERT_OK(env_->WriteToNewFile(kTestFileName, options_file_content));
    RocksDBOptionsParser parser;
    ASSERT_NOK(parser.Parse(kTestFileName, fs_.get(), false,
                            4096 /* readahead_size */));
@ -2384,7 +2388,7 @@ TEST_F(OptionsParserTest, ParseVersion) {
    snprintf(buffer, kLength - 1, file_template.c_str(), iv.c_str());

    parser.Reset();
-    env_->WriteToNewFile(iv, buffer);
+    ASSERT_OK(env_->WriteToNewFile(iv, buffer));
    ASSERT_NOK(parser.Parse(iv, fs_.get(), false, 0 /* readahead_size */));
  }

@ -2393,7 +2397,7 @@ TEST_F(OptionsParserTest, ParseVersion) {
  for (auto vv : valid_versions) {
    snprintf(buffer, kLength - 1, file_template.c_str(), vv.c_str());
    parser.Reset();
-    env_->WriteToNewFile(vv, buffer);
+    ASSERT_OK(env_->WriteToNewFile(vv, buffer));
    ASSERT_OK(parser.Parse(vv, fs_.get(), false, 0 /* readahead_size */));
  }
 }
--- a/test_util/testutil.h
+++ b/test_util/testutil.h
@ -594,14 +594,17 @@ inline std::string EncodeInt(uint64_t x) {
                                const std::string& content) {
      std::unique_ptr<WritableFile> r;
      auto s = NewWritableFile(file_name, &r, EnvOptions());
-      if (!s.ok()) {
-        return s;
+      if (s.ok()) {
+        s = r->Append(content);
      }
-      r->Append(content);
-      r->Flush();
-      r->Close();
-      assert(files_[file_name] == content);
-      return Status::OK();
+      if (s.ok()) {
+        s = r->Flush();
+      }
+      if (s.ok()) {
+        s = r->Close();
+      }
+      assert(!s.ok() || files_[file_name] == content);
+      return s;
    }

    // The following text is boilerplate that forwards all methods to target()
--- a/util/status.cc
+++ b/util/status.cc
@ -75,6 +75,9 @@ Status::Status(Code _code, SubCode _subcode, const Slice& msg,
 }

 std::string Status::ToString() const {
+#ifdef ROCKSDB_ASSERT_STATUS_CHECKED
+  checked_ = true;
+#endif  // ROCKSDB_ASSERT_STATUS_CHECKED
  char tmp[30];
  const char* type;
  switch (code_) {