Cover LoadBalancer with tests (#1394)

Fixes #680. Fixes #1222.

.githooks/check-format

@@ -59,7 +59,9 @@ function grep_code {
     grep -l "${1}" ${sources} -r --include \*.hpp --include \*.cpp
 }
 
-if [[ "$OSTYPE" == "darwin"* ]]; then
+GNU_SED=$(sed --version 2>&1 | grep -q 'GNU' && echo true || echo false)
+
+if [[ "$GNU_SED" == "false" ]]; then # macOS sed
     # make all includes to be <...> style
     grep_code '#include ".*"' | xargs sed -i '' -E 's|#include "(.*)"|#include <\1>|g'
 
@@ -83,9 +85,10 @@ first=$(git diff $sources $cmake_files)
 find $sources -type f \( -name '*.cpp' -o -name '*.hpp' -o -name '*.ipp' \) -print0 | xargs -0 $formatter
 cmake-format -i $cmake_files
 second=$(git diff $sources $cmake_files)
-changes=$(diff <(echo "$first") <(echo "$second") | wc -l | sed -e 's/^[[:space:]]*//')
+changes=$(diff <(echo "$first") <(echo "$second"))
+changes_number=$(echo -n "$changes" | wc -l | sed -e 's/^[[:space:]]*//')
 
-if [ "$changes" != "0" ]; then
+if [ "$changes_number" != "0" ]; then
     cat <<\EOF
 
                                    WARNING
@@ -95,5 +98,8 @@ if [ "$changes" != "0" ]; then
 -----------------------------------------------------------------------------
 
 EOF
+    if [[ "$1" == "--diff" ]]; then
+        echo "$changes"
+    fi
     exit 1
 fi

.github/workflows/build.yml

@@ -21,7 +21,7 @@ jobs:
       - name: Run formatters
         id: run_formatters
         run: |
-          ./.githooks/check-format
+          ./.githooks/check-format --diff
         shell: bash
 
   check_docs:

src/data/BackendInterface.cpp

@@ -235,7 +235,8 @@ BackendInterface::fetchBookOffers(
     LOG(gLog.debug()) << "Fetching " << std::to_string(keys.size()) << " offers took "
                       << std::to_string(getMillis(mid - begin)) << " milliseconds. Fetching next dir took "
                       << std::to_string(succMillis) << " milliseonds. Fetched next dir " << std::to_string(numSucc)
-                      << " times" << " Fetching next page of dir took " << std::to_string(pageMillis) << " milliseconds"
+                      << " times"
+                      << " Fetching next page of dir took " << std::to_string(pageMillis) << " milliseconds"
                       << ". num pages = " << std::to_string(numPages) << ". Fetching all objects took "
                       << std::to_string(getMillis(end - mid))
                       << " milliseconds. total time = " << std::to_string(getMillis(end - begin)) << " milliseconds"

src/data/CassandraBackend.hpp

@@ -792,8 +792,8 @@ class BasicCassandraBackend : public BackendInterface {
     void
     writeSuccessor(std::string&& key, std::uint32_t const seq, std::string&& successor) override
     {
-        LOG(log_.trace()) << "Writing successor. key = " << key.size() << " bytes. " << " seq = " << std::to_string(seq)
-                          << " successor = " << successor.size() << " bytes.";
+        LOG(log_.trace()) << "Writing successor. key = " << key.size() << " bytes. "
+                          << " seq = " << std::to_string(seq) << " successor = " << successor.size() << " bytes.";
         ASSERT(!key.empty(), "Key must not be empty");
         ASSERT(!successor.empty(), "Successor must not be empty");
 

src/data/cassandra/impl/ExecutionStrategy.hpp

@@ -484,7 +484,8 @@ class DefaultExecutionStrategy {
         {
             std::unique_lock<std::mutex> lck(throttleMutex_);
             if (!canAddWriteRequest()) {
-                LOG(log_.trace()) << "Max outstanding requests reached. " << "Waiting for other requests to finish";
+                LOG(log_.trace()) << "Max outstanding requests reached. "
+                                  << "Waiting for other requests to finish";
                 throttleCv_.wait(lck, [this]() { return canAddWriteRequest(); });
             }
         }

src/etl/CMakeLists.txt

@@ -2,11 +2,13 @@ add_library(clio_etl)
 
 target_sources(
   clio_etl
-  PRIVATE NFTHelpers.cpp
+  PRIVATE CacheLoaderSettings.cpp
+          ETLHelpers.cpp
           ETLService.cpp
           ETLState.cpp
           LoadBalancer.cpp
-          CacheLoaderSettings.cpp
+          NetworkValidatedLedgers.cpp
+          NFTHelpers.cpp
           Source.cpp
           impl/ForwardingCache.cpp
           impl/ForwardingSource.cpp

src/etl/ETLHelpers.cpp

@@ -0,0 +1,46 @@
+//------------------------------------------------------------------------------
+/*
+    This file is part of clio: https://github.com/XRPLF/clio
+    Copyright (c) 2024, the clio developers.
+
+    Permission to use, copy, modify, and distribute this software for any
+    purpose with or without fee is hereby granted, provided that the above
+    copyright notice and this permission notice appear in all copies.
+
+    THE  SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+    WITH  REGARD  TO  THIS  SOFTWARE  INCLUDING  ALL  IMPLIED  WARRANTIES  OF
+    MERCHANTABILITY  AND  FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+    ANY  SPECIAL,  DIRECT,  INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+    WHATSOEVER  RESULTING  FROM  LOSS  OF USE, DATA OR PROFITS, WHETHER IN AN
+    ACTION  OF  CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+    OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+*/
+//==============================================================================
+
+#include "etl/ETLHelpers.hpp"
+
+#include "util/Assert.hpp"
+
+#include <ripple/basics/base_uint.h>
+
+#include <cstddef>
+#include <vector>
+
+namespace etl {
+std::vector<ripple::uint256>
+getMarkers(size_t numMarkers)
+{
+    ASSERT(numMarkers <= 256, "Number of markers must be <= 256. Got: {}", numMarkers);
+
+    unsigned char const incr = 256 / numMarkers;
+
+    std::vector<ripple::uint256> markers;
+    markers.reserve(numMarkers);
+    ripple::uint256 base{0};
+    for (size_t i = 0; i < numMarkers; ++i) {
+        markers.push_back(base);
+        base.data()[0] += incr;
+    }
+    return markers;
+}
+}  // namespace etl

src/etl/ETLHelpers.hpp

@@ -20,11 +20,8 @@
 /** @file */
 #pragma once
 
-#include "util/Assert.hpp"
-
 #include <ripple/basics/base_uint.h>
 
-#include <chrono>
 #include <condition_variable>
 #include <cstddef>
 #include <cstdint>
@@ -35,83 +32,6 @@
 #include <vector>
 
 namespace etl {
-/**
- * @brief This datastructure is used to keep track of the sequence of the most recent ledger validated by the network.
- *
- * There are two methods that will wait until certain conditions are met. This datastructure is able to be "stopped".
- * When the datastructure is stopped, any threads currently waiting are unblocked.
- * Any later calls to methods of this datastructure will not wait. Once the datastructure is stopped, the datastructure
- * remains stopped for the rest of its lifetime.
- */
-class NetworkValidatedLedgers {
-    // max sequence validated by network
-    std::optional<uint32_t> max_;
-
-    mutable std::mutex m_;
-    std::condition_variable cv_;
-
-public:
-    /**
-     * @brief A factory function for NetworkValidatedLedgers
-     *
-     * @return A shared pointer to a new instance of NetworkValidatedLedgers
-     */
-    static std::shared_ptr<NetworkValidatedLedgers>
-    make_ValidatedLedgers()
-    {
-        return std::make_shared<NetworkValidatedLedgers>();
-    }
-
-    /**
-     * @brief Notify the datastructure that idx has been validated by the network.
-     *
-     * @param idx Sequence validated by network
-     */
-    void
-    push(uint32_t idx)
-    {
-        std::lock_guard const lck(m_);
-        if (!max_ || idx > *max_)
-            max_ = idx;
-        cv_.notify_all();
-    }
-
-    /**
-     * @brief Get most recently validated sequence.
-     *
-     * If no ledgers are known to have been validated, this function waits until the next ledger is validated
-     *
-     * @return Sequence of most recently validated ledger. empty optional if the datastructure has been stopped
-     */
-    std::optional<uint32_t>
-    getMostRecent()
-    {
-        std::unique_lock lck(m_);
-        cv_.wait(lck, [this]() { return max_; });
-        return max_;
-    }
-
-    /**
-     * @brief Waits for the sequence to be validated by the network.
-     *
-     * @param sequence The sequence to wait for
-     * @param maxWaitMs Maximum time to wait for the sequence to be validated. If empty, wait indefinitely
-     * @return true if sequence was validated, false otherwise a return value of false means the datastructure has been
-     * stopped
-     */
-    bool
-    waitUntilValidatedByNetwork(uint32_t sequence, std::optional<uint32_t> maxWaitMs = {})
-    {
-        std::unique_lock lck(m_);
-        auto pred = [sequence, this]() -> bool { return (max_ && sequence <= *max_); };
-        if (maxWaitMs) {
-            cv_.wait_for(lck, std::chrono::milliseconds(*maxWaitMs));
-        } else {
-            cv_.wait(lck, pred);
-        }
-        return pred();
-    }
-};
 
 // TODO: does the note make sense? lockfree queues provide the same blocking behaviour just without mutex, don't they?
 /**
@@ -228,20 +148,7 @@ class ThreadSafeQueue {
  * @param numMarkers Total markers to partition for
  * @return The markers
  */
-inline std::vector<ripple::uint256>
-getMarkers(size_t numMarkers)
-{
-    ASSERT(numMarkers <= 256, "Number of markers must be <= 256. Got: {}", numMarkers);
-
-    unsigned char const incr = 256 / numMarkers;
-
-    std::vector<ripple::uint256> markers;
-    markers.reserve(numMarkers);
-    ripple::uint256 base{0};
-    for (size_t i = 0; i < numMarkers; ++i) {
-        markers.push_back(base);
-        base.data()[0] += incr;
-    }
-    return markers;
-}
+std::vector<ripple::uint256>
+getMarkers(size_t numMarkers);
+
 }  // namespace etl

src/etl/ETLService.cpp

@@ -22,6 +22,8 @@
 #include "data/BackendInterface.hpp"
 #include "data/LedgerCache.hpp"
 #include "etl/CorruptionDetector.hpp"
+#include "etl/ETLHelpers.hpp"
+#include "feed/SubscriptionManagerInterface.hpp"
 #include "util/Assert.hpp"
 #include "util/Constants.hpp"
 #include "util/config/Config.hpp"
@@ -263,9 +265,9 @@ ETLService::ETLService(
     util::Config const& config,
     boost::asio::io_context& ioc,
     std::shared_ptr<BackendInterface> backend,
-    std::shared_ptr<SubscriptionManagerType> subscriptions,
+    std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
     std::shared_ptr<LoadBalancerType> balancer,
-    std::shared_ptr<NetworkValidatedLedgersType> ledgers
+    std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
 )
     : backend_(backend)
     , loadBalancer_(balancer)

src/etl/ETLService.hpp

@@ -33,7 +33,7 @@
 #include "etl/impl/LedgerLoader.hpp"
 #include "etl/impl/LedgerPublisher.hpp"
 #include "etl/impl/Transformer.hpp"
-#include "feed/SubscriptionManager.hpp"
+#include "feed/SubscriptionManagerInterface.hpp"
 #include "util/log/Logger.hpp"
 
 #include <boost/asio/io_context.hpp>
@@ -77,16 +77,14 @@ namespace etl {
  */
 class ETLService {
     // TODO: make these template parameters in ETLService
-    using SubscriptionManagerType = feed::SubscriptionManager;
     using LoadBalancerType = LoadBalancer;
-    using NetworkValidatedLedgersType = NetworkValidatedLedgers;
     using DataPipeType = etl::impl::ExtractionDataPipe<org::xrpl::rpc::v1::GetLedgerResponse>;
     using CacheType = data::LedgerCache;
     using CacheLoaderType = etl::CacheLoader<CacheType>;
     using LedgerFetcherType = etl::impl::LedgerFetcher<LoadBalancerType>;
-    using ExtractorType = etl::impl::Extractor<DataPipeType, NetworkValidatedLedgersType, LedgerFetcherType>;
+    using ExtractorType = etl::impl::Extractor<DataPipeType, LedgerFetcherType>;
     using LedgerLoaderType = etl::impl::LedgerLoader<LoadBalancerType, LedgerFetcherType>;
-    using LedgerPublisherType = etl::impl::LedgerPublisher<SubscriptionManagerType, CacheType>;
+    using LedgerPublisherType = etl::impl::LedgerPublisher<CacheType>;
     using AmendmentBlockHandlerType = etl::impl::AmendmentBlockHandler<>;
     using TransformerType =
         etl::impl::Transformer<DataPipeType, LedgerLoaderType, LedgerPublisherType, AmendmentBlockHandlerType>;
@@ -95,7 +93,7 @@ class ETLService {
 
     std::shared_ptr<BackendInterface> backend_;
     std::shared_ptr<LoadBalancerType> loadBalancer_;
-    std::shared_ptr<NetworkValidatedLedgersType> networkValidatedLedgers_;
+    std::shared_ptr<NetworkValidatedLedgersInterface> networkValidatedLedgers_;
 
     std::uint32_t extractorThreads_ = 1;
     std::thread worker_;
@@ -128,9 +126,9 @@ class ETLService {
         util::Config const& config,
         boost::asio::io_context& ioc,
         std::shared_ptr<BackendInterface> backend,
-        std::shared_ptr<SubscriptionManagerType> subscriptions,
+        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
         std::shared_ptr<LoadBalancerType> balancer,
-        std::shared_ptr<NetworkValidatedLedgersType> ledgers
+        std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
     );
 
     /**
@@ -151,9 +149,9 @@ class ETLService {
         util::Config const& config,
         boost::asio::io_context& ioc,
         std::shared_ptr<BackendInterface> backend,
-        std::shared_ptr<SubscriptionManagerType> subscriptions,
+        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
         std::shared_ptr<LoadBalancerType> balancer,
-        std::shared_ptr<NetworkValidatedLedgersType> ledgers
+        std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
     )
     {
         auto etl = std::make_shared<ETLService>(config, ioc, backend, subscriptions, balancer, ledgers);

src/etl/ETLState.cpp

@@ -27,21 +27,23 @@
 #include <ripple/protocol/jss.h>
 
 #include <cstdint>
+#include <optional>
 
 namespace etl {
 
-ETLState
-tag_invoke(boost::json::value_to_tag<ETLState>, boost::json::value const& jv)
+std::optional<ETLState>
+tag_invoke(boost::json::value_to_tag<std::optional<ETLState>>, boost::json::value const& jv)
 {
     ETLState state;
     auto const& jsonObject = jv.as_object();
 
-    if (!jsonObject.contains(JS(error))) {
-        if (jsonObject.contains(JS(result)) && jsonObject.at(JS(result)).as_object().contains(JS(info))) {
-            auto const rippledInfo = jsonObject.at(JS(result)).as_object().at(JS(info)).as_object();
-            if (rippledInfo.contains(JS(network_id)))
-                state.networkID.emplace(boost::json::value_to<int64_t>(rippledInfo.at(JS(network_id))));
-        }
+    if (jsonObject.contains(JS(error)))
+        return std::nullopt;
+
+    if (jsonObject.contains(JS(result)) && jsonObject.at(JS(result)).as_object().contains(JS(info))) {
+        auto const rippledInfo = jsonObject.at(JS(result)).as_object().at(JS(info)).as_object();
+        if (rippledInfo.contains(JS(network_id)))
+            state.networkID.emplace(boost::json::value_to<int64_t>(rippledInfo.at(JS(network_id))));
     }
 
     return state;

src/etl/ETLState.hpp

@@ -51,7 +51,7 @@ struct ETLState {
         });
 
         if (serverInfoRippled)
-            return boost::json::value_to<ETLState>(boost::json::value(*serverInfoRippled));
+            return boost::json::value_to<std::optional<ETLState>>(boost::json::value(*serverInfoRippled));
 
         return std::nullopt;
     }
@@ -63,7 +63,7 @@ struct ETLState {
  * @param jv The json value to convert
  * @return The ETLState
  */
-ETLState
-tag_invoke(boost::json::value_to_tag<ETLState>, boost::json::value const& jv);
+std::optional<ETLState>
+tag_invoke(boost::json::value_to_tag<std::optional<ETLState>>, boost::json::value const& jv);
 
 }  // namespace etl