Mark inoperable cache as corrupted (#1203)

Cache could become inoperable due to several reasons:
- interrupted compaction leads to files on level-0, and this leads to
excessive memory usage during read operations;
- different IO errors for protected cache couldn't be solved with
RepairDb due to read-only nature of this cache (for mutable cache there
will be an attempt to repair it);

Do not remove corrupted cache and just notify a user about it.

Relates-To: OLPSUP-14088
Signed-off-by: Andrey Kashcheev <[email protected]>

Author: andrey-kashcheev

olp-cpp-sdk-core/include/olp/core/cache/DefaultCache.h

@@ -60,9 +60,10 @@ class CORE_API DefaultCache : public KeyValueCache {
    * @brief The storage open result type.
    */
   enum StorageOpenResult {
-    Success,             /*!< The operation succeeded. */
-    OpenDiskPathFailure, /*!< The disk cache failure. */
-    NotReady             /*!< The DefaultCache is closed. */
+    Success,                 /*!< The operation succeeded. */
+    OpenDiskPathFailure,     /*!< The disk cache failure. */
+    ProtectedCacheCorrupted, /*!< The protected disk cache is corrupted. */
+    NotReady                 /*!< The DefaultCache is closed. */
   };
 
   /**
@@ -140,6 +141,10 @@ class CORE_API DefaultCache : public KeyValueCache {
    * operation in parallel for the time of the compacting operation. Be aware
    * that automatic asynchronous compacting operation is triggered internally
    * once the database size exceeds the CacheSettings::max_disk_storage size.
+   *
+   * @note After the compaction is finished the cache is checked on level-0 file
+   * presence. If there are still some files present another round of compaction
+   * is performed.
    */
   void Compact();
 

olp-cpp-sdk-core/src/cache/DefaultCacheImpl.cpp

@@ -31,6 +31,7 @@
 
 namespace {
 using CacheType = olp::cache::DefaultCache::CacheType;
+using StorageOpenResult = olp::cache::DefaultCache::StorageOpenResult;
 
 constexpr auto kLogTag = "DefaultCache";
 constexpr auto kExpirySuffix = "::expiry";
@@ -119,6 +120,21 @@ int64_t GetElapsedTime(std::chrono::steady_clock::time_point start) {
 bool IsInternalKey(const std::string& key) {
   return key.find(kInternalKeysPrefix) == 0u;
 }
+
+olp::cache::DefaultCache::StorageOpenResult ToStorageOpenResult(
+    olp::cache::OpenResult input) {
+  switch (input) {
+    case olp::cache::OpenResult::Fail:
+      return StorageOpenResult::OpenDiskPathFailure;
+    case olp::cache::OpenResult::Corrupted:
+      return StorageOpenResult::ProtectedCacheCorrupted;
+    case olp::cache::OpenResult::Repaired:
+    case olp::cache::OpenResult::Success:
+      return StorageOpenResult::Success;
+  }
+
+  return {};
+}
 }  // namespace
 
 namespace olp {
@@ -817,13 +833,13 @@ DefaultCache::StorageOpenResult DefaultCacheImpl::SetupProtectedCache() {
   auto status = protected_cache_->Open(
       settings_.disk_path_protected.get(), settings_.disk_path_protected.get(),
       protected_storage_settings, OpenOptions::ReadOnly);
-  if (status == OpenResult::Fail) {
-    OLP_SDK_LOG_ERROR_F(kLogTag, "Failed to reopen protected cache %s",
+  if (status != OpenResult::Success) {
+    OLP_SDK_LOG_ERROR_F(kLogTag, "Failed to open protected cache %s",
                         settings_.disk_path_protected.get().c_str());
 
     protected_cache_.reset();
     settings_.disk_path_protected = boost::none;
-    return DefaultCache::OpenDiskPathFailure;
+    return ToStorageOpenResult(status);
   }
 
   return DefaultCache::Success;
@@ -836,13 +852,15 @@ DefaultCache::StorageOpenResult DefaultCacheImpl::SetupMutableCache() {
   auto status = mutable_cache_->Open(settings_.disk_path_mutable.get(),
                                      settings_.disk_path_mutable.get(),
                                      storage_settings, OpenOptions::Default);
-  if (status == OpenResult::Fail) {
+  if (status == OpenResult::Repaired) {
+    OLP_SDK_LOG_INFO(kLogTag, "Mutable cache was repaired");
+  } else if (status == OpenResult::Fail) {
     OLP_SDK_LOG_ERROR_F(kLogTag, "Failed to open the mutable cache %s",
                         settings_.disk_path_mutable.get().c_str());
 
     mutable_cache_.reset();
     settings_.disk_path_mutable = boost::none;
-    return DefaultCache::OpenDiskPathFailure;
+    return StorageOpenResult::OpenDiskPathFailure;
   }
 
   // read protected keys

olp-cpp-sdk-core/src/cache/DiskCache.cpp

@@ -42,6 +42,7 @@ namespace cache {
 
 namespace {
 constexpr auto kLogTag = "DiskCache";
+constexpr auto kLevelDbLostFolder = "/lost";
 
 leveldb::Slice ToLeveldbSlice(const std::string& slice) {
   return leveldb::Slice(slice);
@@ -52,7 +53,13 @@ static bool RepairCache(const std::string& data_path) {
   auto status = leveldb::RepairDB(data_path, leveldb::Options());
   if (status.ok()) {
     OLP_SDK_LOG_INFO(kLogTag, "RepairCache: repaired - " << data_path);
-    DiskCacheEnv::Env()->DeleteDir(data_path + "/lost");
+    const auto lost_folder_path = data_path + kLevelDbLostFolder;
+    if (utils::Dir::Exists(lost_folder_path)) {
+      OLP_SDK_LOG_INFO_F(
+          kLogTag, "RepairCache: some data may have been lost - deleting '%s'",
+          kLevelDbLostFolder);
+      utils::Dir::Remove(lost_folder_path);
+    }
     return true;
   }
   OLP_SDK_LOG_ERROR(kLogTag,
@@ -120,6 +127,22 @@ client::ApiError GetApiError(const leveldb::Status& status) {
   return client::ApiError(code, status.ToString());
 }
 
+int CheckCompactionFinished(leveldb::DB& db) {
+  std::string property_result;
+  db.GetProperty("leveldb.num-files-at-level0", &property_result);
+
+  const auto files_at_level0 = std::stoi(property_result);
+  if (files_at_level0 == 0) {
+    return true;
+  }
+
+  OLP_SDK_LOG_INFO_F(
+      kLogTag, "CheckCompactionFinished: L0 files present, files_at_level0=%d",
+      files_at_level0);
+
+  return false;
+}
+
 }  // anonymous namespace
 
 DiskCache::DiskCache() = default;
@@ -153,8 +176,13 @@ void DiskCache::Compact() {
   // doing it already. We don't need two at the same time.
   if (database_ && !compacting_.exchange(true)) {
     OLP_SDK_LOG_INFO(kLogTag, "Compact: Compacting database started");
-    database_->CompactRange(nullptr, nullptr);
+
+    do {
+      database_->CompactRange(nullptr, nullptr);
+    } while (!CheckCompactionFinished(*database_));
+
     compacting_ = false;
+
     OLP_SDK_LOG_INFO(kLogTag, "Compact: Compacting database finished");
   }
 }
@@ -216,13 +244,18 @@ OpenResult DiskCache::Open(const std::string& data_path,
     status = leveldb::DB::Open(open_options, versioned_data_path, &db);
   }
 
-  // If the database is r/w and corrupted, attempt to repair & reopen
-  if ((status.IsCorruption() || status.IsIOError()) &&
-      RepairCache(versioned_data_path) == true) {
-    status = leveldb::DB::Open(open_options, versioned_data_path, &db);
-    if (status.ok()) {
-      database_.reset(db);
-      return OpenResult::Repaired;
+  if (status.IsCorruption() || status.IsIOError()) {
+    if (is_read_only) {
+      OLP_SDK_LOG_ERROR_F(
+          kLogTag, "Open: cache corrupted, cache_path='%s', error='%s'",
+          versioned_data_path.c_str(), status.ToString().c_str());
+      return OpenResult::Corrupted;
+    } else if (RepairCache(versioned_data_path)) {
+      status = leveldb::DB::Open(open_options, versioned_data_path, &db);
+      if (status.ok()) {
+        database_.reset(db);
+        return OpenResult::Repaired;
+      }
     }
   }
 
@@ -235,7 +268,13 @@ OpenResult DiskCache::Open(const std::string& data_path,
     error_ = NoError{};
   }
 
+  if (is_read_only && !CheckCompactionFinished(*db)) {
+    OLP_SDK_LOG_ERROR(kLogTag, "Open: interrupted compaction detected");
+    return OpenResult::Corrupted;
+  }
+
   database_.reset(db);
+
   return OpenResult::Success;
 }
 

olp-cpp-sdk-core/src/cache/DiskCache.h

@@ -19,8 +19,8 @@
 
 #pragma once
 
-#include <time.h>
 #include <atomic>
+#include <ctime>
 #include <functional>
 #include <limits>
 #include <map>
@@ -55,6 +55,8 @@ class SizeCountingEnv;
 enum class OpenResult {
   /// Opening the store failed. Use openError() for details.
   Fail,
+  /// The store was corrupted or store compaction was interrupted.
+  Corrupted,
   /// The store was corrupted and has been repaired. Internal integrity might be
   /// broken.
   Repaired,
@@ -138,7 +140,7 @@ class DiskCache {
   /// scans.
   std::unique_ptr<leveldb::Iterator> NewIterator(leveldb::ReadOptions options);
 
-  /// Allow batch writting so that we can delete and write multiple values at
+  /// Allow batch writing so that we can delete and write multiple values at
   /// the same time.
   OperationOutcome ApplyBatch(std::unique_ptr<leveldb::WriteBatch> batch);