feat: Read and write LedgerCache to file (#2761)

Fixes #2413.

Author: kuznetsss

docs/config-description.md

@@ -441,6 +441,22 @@ This document provides a list of all available Clio configuration properties in
 - **Constraints**: The value must be one of the following: `sync`, `async`, `none`.
 - **Description**: The strategy used for Cache loading.
 
+### cache.file.path
+
+- **Required**: False
+- **Type**: string
+- **Default value**: None
+- **Constraints**: None
+- **Description**: The path to a file where cache will be saved to on shutdown and loaded from on startup. If the file couldn't be read Clio will load cache as usual (from DB or from rippled).
+
+### cache.file.max_sequence_age
+
+- **Required**: True
+- **Type**: int
+- **Default value**: `5000`
+- **Constraints**: None
+- **Description**: Max allowed difference between the latest sequence in DB and in cache file. If the cache file is too old (contains too low latest sequence) Clio will reject using it.
+
 ### log.channels.[].channel
 
 - **Required**: False

docs/examples/config/example-config.json

@@ -137,7 +137,11 @@
     // "num_cursors_from_account": 3200, // Read the cursors from the account table until we have enough cursors to partition the ledger to load concurrently.
     "num_markers": 48, // The number of markers is the number of coroutines to load the cache concurrently.
     "page_fetch_size": 512, // The number of rows to load for each page.
-    "load": "async" // "sync" to load cache synchronously  or "async" to load cache asynchronously or "none"/"no" to turn off the cache.
+    "load": "async", // "sync" to load cache synchronously  or "async" to load cache asynchronously or "none"/"no" to turn off the cache.
+    "file": {
+      "path": "./cache.bin",
+      "max_sequence_age": 5000
+    }
   },
   "prometheus": {
     "enabled": true,

src/app/ClioApplication.cpp

@@ -55,6 +55,7 @@
 #include <cstdlib>
 #include <memory>
 #include <optional>
+#include <string>
 #include <thread>
 #include <utility>
 #include <vector>
@@ -110,7 +111,23 @@ ClioApplication::run(bool const useNgWebServer)
     auto const dosguardWeights = web::dosguard::Weights::make(config_);
     auto dosGuard = web::dosguard::DOSGuard{config_, whitelistHandler, dosguardWeights};
     auto sweepHandler = web::dosguard::IntervalSweepHandler{config_, ioc, dosGuard};
+
     auto cache = data::LedgerCache{};
+    appStopper_.setOnStop([&cache, this](auto&&) {
+        // TODO(kuznetsss): move this into Stopper::makeOnStopCallback()
+        auto const cacheFilePath = config_.maybeValue<std::string>("cache.file.path");
+        if (not cacheFilePath.has_value()) {
+            return;
+        }
+
+        LOG(util::LogService::info()) << "Saving ledger cache to " << *cacheFilePath;
+        if (auto const [success, duration_ms] = util::timed([&]() { return cache.saveToFile(*cacheFilePath); });
+            success.has_value()) {
+            LOG(util::LogService::info()) << "Successfully saved ledger cache in " << duration_ms << " ms";
+        } else {
+            LOG(util::LogService::error()) << "Error saving LedgerCache to file";
+        }
+    });
 
     // Interface to the database
     auto backend = data::makeBackend(config_, cache);

src/data/BackendInterface.cpp

@@ -270,7 +270,7 @@ BackendInterface::updateRange(uint32_t newMax)
 {
     std::scoped_lock const lck(rngMtx_);
 
-    if (range_.has_value() && newMax < range_->maxSequence) {
+    if (range_.has_value() and newMax < range_->maxSequence) {
         ASSERT(
             false,
             "Range shouldn't exist yet or newMax should be at least range->maxSequence. newMax = {}, "
@@ -280,11 +280,14 @@ BackendInterface::updateRange(uint32_t newMax)
         );
     }
 
-    if (!range_.has_value()) {
-        range_ = {.minSequence = newMax, .maxSequence = newMax};
-    } else {
-        range_->maxSequence = newMax;
-    }
+    updateRangeImpl(newMax);
+}
+
+void
+BackendInterface::forceUpdateRange(uint32_t newMax)
+{
+    std::scoped_lock const lck(rngMtx_);
+    updateRangeImpl(newMax);
 }
 
 void
@@ -410,4 +413,14 @@ BackendInterface::fetchFees(std::uint32_t const seq, boost::asio::yield_context
     return fees;
 }
 
+void
+BackendInterface::updateRangeImpl(uint32_t newMax)
+{
+    if (!range_.has_value()) {
+        range_ = {.minSequence = newMax, .maxSequence = newMax};
+    } else {
+        range_->maxSequence = newMax;
+    }
+}
+
 }  // namespace data

src/data/BackendInterface.hpp

@@ -249,6 +249,15 @@ class BackendInterface {
     void
     updateRange(uint32_t newMax);
 
+    /**
+     * @brief Updates the range of sequences that are stored in the DB without any checks
+     * @note In the most cases you should use updateRange() instead
+     *
+     * @param newMax The new maximum sequence available
+     */
+    void
+    forceUpdateRange(uint32_t newMax);
+
     /**
      * @brief Sets the range of sequences that are stored in the DB.
      *
@@ -776,6 +785,9 @@ class BackendInterface {
      */
     virtual bool
     doFinishWrites() = 0;
+
+    void
+    updateRangeImpl(uint32_t newMax);
 };
 
 }  // namespace data

src/data/CMakeLists.txt

@@ -14,6 +14,9 @@ target_sources(
           cassandra/impl/SslContext.cpp
           cassandra/Handle.cpp
           cassandra/SettingsProvider.cpp
+          impl/InputFile.cpp
+          impl/LedgerCacheFile.cpp
+          impl/OutputFile.cpp
 )
 
 target_link_libraries(clio_data PUBLIC cassandra-cpp-driver::cassandra-cpp-driver clio_util)

src/data/LedgerCache.cpp

@@ -20,16 +20,22 @@
 #include "data/LedgerCache.hpp"
 
 #include "data/Types.hpp"
+#include "data/impl/LedgerCacheFile.hpp"
 #include "etl/Models.hpp"
 #include "util/Assert.hpp"
 
 #include <xrpl/basics/base_uint.h>
 
 #include <cstddef>
 #include <cstdint>
+#include <cstdlib>
+#include <cstring>
+#include <map>
 #include <mutex>
 #include <optional>
 #include <shared_mutex>
+#include <string>
+#include <utility>
 #include <vector>
 
 namespace data {
@@ -251,4 +257,34 @@ LedgerCache::getSuccessorHitRate() const
     return static_cast<float>(successorHitCounter_.get().value()) / successorReqCounter_.get().value();
 }
 
+std::expected<void, std::string>
+LedgerCache::saveToFile(std::string const& path) const
+{
+    if (not isFull()) {
+        return std::unexpected{"Ledger cache is not full"};
+    }
+
+    impl::LedgerCacheFile file{path};
+    std::unique_lock lock{mtx_};
+    impl::LedgerCacheFile::DataView data{.latestSeq = latestSeq_, .map = map_, .deleted = deleted_};
+    return file.write(data);
+}
+
+std::expected<void, std::string>
+LedgerCache::loadFromFile(std::string const& path, uint32_t minLatestSequence)
+{
+    impl::LedgerCacheFile file{path};
+    auto data = file.read(minLatestSequence);
+    if (not data.has_value()) {
+        return std::unexpected(std::move(data).error());
+    }
+    auto [latestSeq, map, deleted] = std::move(data).value();
+    std::unique_lock lock{mtx_};
+    latestSeq_ = latestSeq;
+    map_ = std::move(map);
+    deleted_ = std::move(deleted);
+    full_ = true;
+    return {};
+}
+
 }  // namespace data

src/data/LedgerCache.hpp

@@ -37,6 +37,7 @@
 #include <map>
 #include <optional>
 #include <shared_mutex>
+#include <string>
 #include <unordered_set>
 #include <vector>
 
@@ -46,11 +47,16 @@ namespace data {
  * @brief Cache for an entire ledger.
  */
 class LedgerCache : public LedgerCacheInterface {
+public:
+    /** @brief An entry of the cache */
     struct CacheEntry {
         uint32_t seq = 0;
         Blob blob;
     };
 
+    using CacheMap = std::map<ripple::uint256, CacheEntry>;
+
+private:
     // counters for fetchLedgerObject(s) hit rate
     std::reference_wrapper<util::prometheus::CounterInt> objectReqCounter_{PrometheusService::counterInt(
         "ledger_cache_counter_total_number",
@@ -73,8 +79,8 @@ class LedgerCache : public LedgerCacheInterface {
         util::prometheus::Labels({{"type", "cache_hit"}, {"fetch", "successor_key"}})
     )};
 
-    std::map<ripple::uint256, CacheEntry> map_;
-    std::map<ripple::uint256, CacheEntry> deleted_;
+    CacheMap map_;
+    CacheMap deleted_;
 
     mutable std::shared_mutex mtx_;
     std::condition_variable_any cv_;
@@ -138,6 +144,19 @@ class LedgerCache : public LedgerCacheInterface {
 
     void
     waitUntilCacheContainsSeq(uint32_t seq) override;
+
+    /**
+     * @brief Save the cache to file
+     * @note This operation takes about 7 seconds and it keeps mtx_ exclusively locked
+     *
+     * @param path The file path to save the cache to
+     * @return An error as a string if any
+     */
+    std::expected<void, std::string>
+    saveToFile(std::string const& path) const;
+
+    std::expected<void, std::string>
+    loadFromFile(std::string const& path, uint32_t minLatestSequence) override;
 };
 
 }  // namespace data

src/data/LedgerCacheInterface.hpp

@@ -27,7 +27,9 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <expected>
 #include <optional>
+#include <string>
 #include <vector>
 
 namespace data {
@@ -168,6 +170,17 @@ class LedgerCacheInterface {
      */
     virtual void
     waitUntilCacheContainsSeq(uint32_t seq) = 0;
+
+    /**
+     * @brief Load the cache from file
+     * @note This operation takes about 7 seconds and it keeps mtx_ exclusively locked
+     *
+     * @param path The file path to load data from
+     * @param minLatestSequence The minimum allowed value of the latestLedgerSequence in cache file
+     * @return An error as a string if any
+     */
+    [[nodiscard]] virtual std::expected<void, std::string>
+    loadFromFile(std::string const& path, uint32_t minLatestSequence) = 0;
 };
 
 }  // namespace data

src/data/Types.hpp

@@ -247,6 +247,9 @@ struct MPTHoldersAndCursor {
 struct LedgerRange {
     std::uint32_t minSequence = 0;
     std::uint32_t maxSequence = 0;
+
+    bool
+    operator==(LedgerRange const&) const = default;
 };
 
 /**