LoadBalancer.hpp

#pragma once
 
#include "data/BackendInterface.hpp"
#include "etl/ETLState.hpp"
#include "etl/InitialLoadObserverInterface.hpp"
#include "etl/LoadBalancerInterface.hpp"
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "etl/Source.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "rpc/Errors.hpp"
#include "util/Mutex.hpp"
#include "util/Random.hpp"
#include "util/ResponseExpirationCache.hpp"
#include "util/config/ConfigDefinition.hpp"
#include "util/log/Logger.hpp"
#include "util/prometheus/Counter.hpp"
 
#include <boost/asio.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/json/object.hpp>
#include <boost/json/value.hpp>
#include <grpcpp/grpcpp.h>
#include <org/xrpl/rpc/v1/get_ledger.pb.h>
#include <org/xrpl/rpc/v1/ledger.pb.h>
#include <xrpl/proto/org/xrpl/rpc/v1/xrp_ledger.grpc.pb.h>
 
#include <chrono>
#include <concepts>
#include <cstdint>
#include <expected>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <vector>
 
namespace etl {

struct LoadBalancerTag {
    virtual ~LoadBalancerTag() = default;
};
 
template <typename T>
concept SomeLoadBalancer = std::derived_from<T, LoadBalancerTag>;

class LoadBalancer : public LoadBalancerInterface, LoadBalancerTag {
public:
    using RawLedgerObjectType = org::xrpl::rpc::v1::RawLedgerObject;
    using GetLedgerResponseType = org::xrpl::rpc::v1::GetLedgerResponse;
    using OptionalGetLedgerResponseType = std::optional<GetLedgerResponseType>;
 
private:
    static constexpr std::uint32_t kDEFAULT_DOWNLOAD_RANGES = 16;
 
    util::Logger log_{"ETL"};
    // Forwarding cache must be destroyed after sources because sources have a callback to
    // invalidate cache
    std::optional<util::ResponseExpirationCache> forwardingCache_;
    std::optional<std::string> forwardingXUserValue_;
 
    std::unique_ptr<util::RandomGeneratorInterface> randomGenerator_;
 
    std::vector<SourcePtr> sources_;
    std::optional<ETLState> etlState_;
    std::uint32_t downloadRanges_ = kDEFAULT_DOWNLOAD_RANGES; /*< The number of markers to use when
                                                                 downloading initial ledger */
 
    struct ForwardingCounters {
        std::reference_wrapper<util::prometheus::CounterInt> successDuration;
        std::reference_wrapper<util::prometheus::CounterInt> failDuration;
        std::reference_wrapper<util::prometheus::CounterInt> retries;
        std::reference_wrapper<util::prometheus::CounterInt> cacheHit;
        std::reference_wrapper<util::prometheus::CounterInt> cacheMiss;
    } forwardingCounters_;
 
    // Using mutex instead of atomic_bool because choosing a new source to
    // forward messages should be done with a mutual exclusion otherwise there will be a race
    // condition
    util::Mutex<bool> hasForwardingSource_{false};
 
public:
    static constexpr std::string_view kADMIN_FORWARDING_X_USER_VALUE = "clio_admin";

    static constexpr std::string_view kUSER_FORWARDING_X_USER_VALUE = "clio_user";

    LoadBalancer(
        util::config::ClioConfigDefinition const& config,
        boost::asio::io_context& ioc,
        std::shared_ptr<BackendInterface> backend,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        std::unique_ptr<util::RandomGeneratorInterface> randomGenerator,
        std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
        SourceFactory sourceFactory = makeSource
    );

    static std::shared_ptr<LoadBalancerInterface>
    makeLoadBalancer(
        util::config::ClioConfigDefinition const& config,
        boost::asio::io_context& ioc,
        std::shared_ptr<BackendInterface> backend,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        std::unique_ptr<util::RandomGeneratorInterface> randomGenerator,
        std::shared_ptr<NetworkValidatedLedgersInterface> validatedLedgers,
        SourceFactory sourceFactory = makeSource
    );

    InitialLedgerLoadResult
    loadInitialLedger(
        uint32_t sequence,
        InitialLoadObserverInterface& observer,
        std::chrono::steady_clock::duration retryAfter
    ) override;

    OptionalGetLedgerResponseType
    fetchLedger(
        uint32_t ledgerSequence,
        bool getObjects,
        bool getObjectNeighbors,
        std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2}
    ) override;

    boost::json::value
    toJson() const override;

    std::expected<boost::json::object, rpc::CombinedError>
    forwardToRippled(
        boost::json::object const& request,
        std::optional<std::string> const& clientIp,
        bool isAdmin,
        boost::asio::yield_context yield
    ) override;

    std::optional<ETLState>
    getETLState() noexcept override;

    void
    stop(boost::asio::yield_context yield) override;
 
private:
    template <typename Func>
    void
    execute(
        Func f,
        uint32_t ledgerSequence,
        std::chrono::steady_clock::duration retryAfter = std::chrono::seconds{2}
    );

    void
    chooseForwardingSource();
};
 
}  // namespace etl