ETLService.hpp

#pragma once
 
#include "data/BackendInterface.hpp"
#include "data/LedgerCacheLoadingState.hpp"
#include "data/Types.hpp"
#include "etl/CacheLoaderInterface.hpp"
#include "etl/CacheUpdaterInterface.hpp"
#include "etl/ETLServiceInterface.hpp"
#include "etl/ETLState.hpp"
#include "etl/ExtractorInterface.hpp"
#include "etl/InitialLoadObserverInterface.hpp"
#include "etl/LedgerPublisherInterface.hpp"
#include "etl/LoadBalancerInterface.hpp"
#include "etl/LoaderInterface.hpp"
#include "etl/MonitorInterface.hpp"
#include "etl/MonitorProviderInterface.hpp"
#include "etl/NetworkValidatedLedgersInterface.hpp"
#include "etl/SystemState.hpp"
#include "etl/TaskManagerInterface.hpp"
#include "etl/TaskManagerProviderInterface.hpp"
#include "etl/impl/AmendmentBlockHandler.hpp"
#include "etl/impl/CacheUpdater.hpp"
#include "etl/impl/Extraction.hpp"
#include "etl/impl/LedgerFetcher.hpp"
#include "etl/impl/LedgerPublisher.hpp"
#include "etl/impl/Loading.hpp"
#include "etl/impl/Registry.hpp"
#include "etl/impl/Scheduling.hpp"
#include "etl/impl/TaskManager.hpp"
#include "etl/impl/ext/Cache.hpp"
#include "etl/impl/ext/Core.hpp"
#include "etl/impl/ext/NFT.hpp"
#include "etl/impl/ext/Successor.hpp"
#include "feed/SubscriptionManagerInterface.hpp"
#include "util/async/AnyExecutionContext.hpp"
#include "util/async/AnyOperation.hpp"
#include "util/async/AnyStrand.hpp"
#include "util/config/ConfigDefinition.hpp"
#include "util/log/Logger.hpp"
 
#include <boost/asio/io_context.hpp>
#include <boost/json/object.hpp>
#include <boost/signals2/connection.hpp>
#include <fmt/format.h>
#include <xrpl/basics/Blob.h>
#include <xrpl/basics/base_uint.h>
#include <xrpl/basics/strHex.h>
#include <xrpl/proto/org/xrpl/rpc/v1/get_ledger.pb.h>
#include <xrpl/proto/org/xrpl/rpc/v1/ledger.pb.h>
#include <xrpl/protocol/LedgerHeader.h>
#include <xrpl/protocol/STTx.h>
#include <xrpl/protocol/TxFormats.h>
#include <xrpl/protocol/TxMeta.h>
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <memory>
#include <optional>
 
namespace etl {

class ETLService : public ETLServiceInterface {
    util::Logger log_{"ETL"};
 
    util::async::AnyExecutionContext ctx_;
    std::reference_wrapper<util::config::ClioConfigDefinition const> config_;
    std::shared_ptr<BackendInterface> backend_;
    std::shared_ptr<LoadBalancerInterface> balancer_;
    std::shared_ptr<NetworkValidatedLedgersInterface> ledgers_;
    std::shared_ptr<LedgerPublisherInterface> publisher_;
    std::shared_ptr<CacheLoaderInterface> cacheLoader_;
    std::shared_ptr<CacheUpdaterInterface> cacheUpdater_;
    std::shared_ptr<ExtractorInterface> extractor_;
    std::shared_ptr<LoaderInterface> loader_;
    std::shared_ptr<InitialLoadObserverInterface> initialLoadObserver_;
    std::shared_ptr<TaskManagerProviderInterface> taskManagerProvider_;
    std::shared_ptr<MonitorProviderInterface> monitorProvider_;
    std::shared_ptr<SystemState> state_;
 
    std::optional<uint32_t> startSequence_;
    std::optional<uint32_t> finishSequence_;
 
    std::unique_ptr<MonitorInterface> monitor_;
    std::unique_ptr<TaskManagerInterface> taskMan_;
 
    boost::signals2::scoped_connection monitorNewSeqSubscription_;
    boost::signals2::scoped_connection monitorDbStalledSubscription_;
    boost::signals2::scoped_connection systemStateWriteCommandSubscription_;
    util::async::AnyStrand writeCommandStrand_;
    std::atomic<size_t> runningWriteCommandHandlers_{0};
 
    std::optional<util::async::AnyOperation<void>> mainLoop_;
 
public:
    static std::shared_ptr<ETLServiceInterface>
    makeETLService(
        util::config::ClioConfigDefinition const& config,
        std::shared_ptr<SystemState> state,
        std::unique_ptr<data::LedgerCacheLoadingStateInterface const> cacheLoadingState,
        util::async::AnyExecutionContext ctx,
        std::shared_ptr<BackendInterface> backend,
        std::shared_ptr<feed::SubscriptionManagerInterface> subscriptions,
        std::shared_ptr<LoadBalancerInterface> balancer,
        std::shared_ptr<NetworkValidatedLedgersInterface> ledgers
    );

    ETLService(
        util::async::AnyExecutionContext ctx,
        std::reference_wrapper<util::config::ClioConfigDefinition const> config,
        std::shared_ptr<data::BackendInterface> backend,
        std::shared_ptr<LoadBalancerInterface> balancer,
        std::shared_ptr<NetworkValidatedLedgersInterface> ledgers,
        std::shared_ptr<LedgerPublisherInterface> publisher,
        std::shared_ptr<CacheLoaderInterface> cacheLoader,
        std::shared_ptr<CacheUpdaterInterface> cacheUpdater,
        std::shared_ptr<ExtractorInterface> extractor,
        std::shared_ptr<LoaderInterface> loader,
        std::shared_ptr<InitialLoadObserverInterface> initialLoadObserver,
        std::shared_ptr<TaskManagerProviderInterface> taskManagerProvider,
        std::shared_ptr<MonitorProviderInterface> monitorProvider,
        std::shared_ptr<SystemState> state
    );
 
    ~ETLService() override;
 
    void
    run() override;
 
    void
    stop() override;
 
    boost::json::object
    getInfo() const override;
 
    bool
    isAmendmentBlocked() const override;
 
    bool
    isCorruptionDetected() const override;
 
    std::optional<ETLState>
    getETLState() const override;
 
    std::uint32_t
    lastCloseAgeSeconds() const override;
 
private:
    std::optional<data::LedgerRange>
    loadInitialLedgerIfNeeded();
 
    [[nodiscard]] uint32_t
    syncCacheWithDb();
 
    void
    updateCache(uint32_t seq);
 
    void
    startMonitor(uint32_t seq);
 
    void
    startLoading(uint32_t seq);
 
    void
    attemptTakeoverWriter();
 
    void
    giveUpWriter();
};
 
}  // namespace etl