BackendInterface.hpp

#pragma once
 
#include "data/DBHelpers.hpp"
#include "data/LedgerCacheInterface.hpp"
#include "data/Types.hpp"
#include "etl/CorruptionDetector.hpp"
#include "util/Retry.hpp"
#include "util/Spawn.hpp"
#include "util/log/Logger.hpp"
 
#include <boost/asio/executor_work_guard.hpp>
#include <boost/asio/io_context.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/json.hpp>
#include <boost/json/object.hpp>
#include <boost/utility/result_of.hpp>
#include <boost/uuid/uuid.hpp>
#include <xrpl/basics/base_uint.h>
#include <xrpl/protocol/AccountID.h>
#include <xrpl/protocol/Fees.h>
#include <xrpl/protocol/LedgerHeader.h>
 
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <optional>
#include <shared_mutex>
#include <string>
#include <thread>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace data {

class DatabaseError : public std::exception {
    std::string message_{"Transient database error. Please retry the request"};
 
public:
    DatabaseError() = default;

    explicit DatabaseError(std::string message) : message_{std::move(message)}
    {
    }

    [[nodiscard]] char const*
    what() const noexcept override
    {
        return message_.c_str();
    }
};

static constexpr std::chrono::milliseconds kDefaultWaitBetweenRetry{500};

static constexpr std::chrono::milliseconds kMaxWaitBetweenRetry{5'000};

static constexpr util::Retry::Delays kDefaultRetryDelays{
    .initial = kDefaultWaitBetweenRetry,
    .max = kDefaultWaitBetweenRetry
};

template <typename FnType>
auto
retryOnTimeout(
    FnType func,
    boost::asio::yield_context yield,
    util::Retry::Delays delays = kDefaultRetryDelays
)
{
    static util::Logger const log{"Backend"};  // NOLINT(readability-identifier-naming)
 
    auto retry = util::makeRetryExponentialBackoff(delays, yield.get_executor());
 
    while (true) {
        // Suspending a coroutine from inside a catch block is not safe so we copy out the failure.
        std::string failure;
 
        try {
            return func();
        } catch (DatabaseError const& e) {
            failure = e.what();
        }
 
        auto const delayMs =
            std::chrono::duration_cast<std::chrono::milliseconds>(retry.delayValue()).count();
        LOG(log.error()) << failure << " (attempt " << retry.attemptNumber() + 1
                         << "). Retrying in " << delayMs << "ms ...";
 
        retry.wait(yield);
    }
}

template <typename FnType>
auto
retryOnTimeout(FnType func, util::Retry::Delays delays = kDefaultRetryDelays)
{
    static util::Logger const log{"Backend"};  // NOLINT(readability-identifier-naming)
 
    util::ExponentialBackoffStrategy backoff{delays};
    std::size_t attempt = 1;
 
    while (true) {
        try {
            return func();
        } catch (DatabaseError const& e) {
            auto const delay = backoff.getDelay();
            LOG(log.error()) << e.what() << " (attempt " << attempt << "). Retrying in "
                             << std::chrono::duration_cast<std::chrono::milliseconds>(delay).count()
                             << "ms ...";
 
            ++attempt;
 
            std::this_thread::sleep_for(delay);
            backoff.increaseDelay();
        }
    }
}

template <typename FnType>
auto
synchronous(FnType&& func)
{
    boost::asio::io_context ctx;
 
    using R = boost::result_of<FnType(boost::asio::yield_context)>::type;
    if constexpr (!std::is_same_v<R, void>) {
        R res;
        util::spawn(ctx, [_ = boost::asio::make_work_guard(ctx), &func, &res](auto yield) {
            res = func(yield);
        });
 
        ctx.run();
        return res;
    } else {
        util::spawn(ctx, [_ = boost::asio::make_work_guard(ctx), &func](auto yield) {
            func(yield);
        });
        ctx.run();
    }
}

template <typename FnType>
auto
synchronousAndRetryOnTimeout(FnType&& func, util::Retry::Delays delays = kDefaultRetryDelays)
{
    return retryOnTimeout([&]() { return synchronous(func); }, delays);
}

class BackendInterface {
protected:
    util::Logger log_{"Backend"};
    mutable std::shared_mutex rngMtx_;
    std::optional<LedgerRange> range_;
    std::reference_wrapper<LedgerCacheInterface> cache_;
    std::optional<etl::CorruptionDetector> corruptionDetector_;
 
public:
    BackendInterface(LedgerCacheInterface& cache) : cache_{cache}
    {
    }
 
    virtual ~BackendInterface() = default;

    [[nodiscard]] virtual std::chrono::milliseconds
    initialRetryDelay() const
    {
        return kDefaultWaitBetweenRetry;
    }

    [[nodiscard]] virtual std::chrono::milliseconds
    maxRetryDelay() const
    {
        return kMaxWaitBetweenRetry;
    }
 
    // TODO https://github.com/XRPLF/clio/issues/1956: Remove this hack once old ETL is removed.
    // Cache should not be exposed thru BackendInterface

    LedgerCacheInterface const&
    cache() const
    {
        return cache_;
    }

    LedgerCacheInterface&
    cache()
    {
        return cache_;
    }

    void
    setCorruptionDetector(etl::CorruptionDetector detector)
    {
        corruptionDetector_ = std::move(detector);
    }

    virtual std::optional<xrpl::LedgerHeader>
    fetchLedgerBySequence(std::uint32_t sequence, boost::asio::yield_context yield) const = 0;

    virtual std::optional<xrpl::LedgerHeader>
    fetchLedgerByHash(xrpl::uint256 const& hash, boost::asio::yield_context yield) const = 0;

    virtual std::optional<std::uint32_t>
    fetchLatestLedgerSequence(boost::asio::yield_context yield) const = 0;

    std::optional<LedgerRange>
    fetchLedgerRange() const;

    virtual std::vector<xrpl::uint256>
    fetchAccountRoots(
        std::uint32_t number,
        std::uint32_t pageSize,
        std::uint32_t seq,
        boost::asio::yield_context yield
    ) const = 0;

    void
    updateRange(uint32_t newMax);

    void
    forceUpdateRange(uint32_t newMax);

    void
    setRange(uint32_t min, uint32_t max, bool force = false);

    std::optional<xrpl::Fees>
    fetchFees(std::uint32_t seq, boost::asio::yield_context yield) const;

    virtual std::optional<TransactionAndMetadata>
    fetchTransaction(xrpl::uint256 const& hash, boost::asio::yield_context yield) const = 0;

    virtual std::vector<TransactionAndMetadata>
    fetchTransactions(
        std::vector<xrpl::uint256> const& hashes,
        boost::asio::yield_context yield
    ) const = 0;

    virtual TransactionsAndCursor
    fetchAccountTransactions(
        xrpl::AccountID const& account,
        std::uint32_t limit,
        bool forward,
        std::optional<TransactionsCursor> const& txnCursor,
        boost::asio::yield_context yield
    ) const = 0;

    virtual std::vector<TransactionAndMetadata>
    fetchAllTransactionsInLedger(
        std::uint32_t ledgerSequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual std::vector<xrpl::uint256>
    fetchAllTransactionHashesInLedger(
        std::uint32_t ledgerSequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual std::optional<NFT>
    fetchNFT(
        xrpl::uint256 const& tokenID,
        std::uint32_t ledgerSequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual TransactionsAndCursor
    fetchNFTTransactions(
        xrpl::uint256 const& tokenID,
        std::uint32_t limit,
        bool forward,
        std::optional<TransactionsCursor> const& cursorIn,
        boost::asio::yield_context yield
    ) const = 0;

    virtual NFTsAndCursor
    fetchNFTsByIssuer(
        xrpl::AccountID const& issuer,
        std::optional<std::uint32_t> const& taxon,
        std::uint32_t ledgerSequence,
        std::uint32_t limit,
        std::optional<xrpl::uint256> const& cursorIn,
        boost::asio::yield_context yield
    ) const = 0;

    virtual MPTHoldersAndCursor
    fetchMPTHolders(
        xrpl::uint192 const& mptID,
        std::uint32_t const limit,
        std::optional<xrpl::AccountID> const& cursorIn,
        std::uint32_t const ledgerSequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual TransactionsAndCursor
    fetchMPTokenIssuanceTransactions(
        xrpl::uint192 const& mptIssuanceID,
        std::uint32_t limit,
        bool forward,
        std::optional<TransactionsCursor> const& cursorIn,
        boost::asio::yield_context yield
    ) const = 0;

    virtual TransactionsAndCursor
    fetchAccountMPTokenIssuanceTransactions(
        xrpl::uint192 const& mptIssuanceID,
        xrpl::AccountID const& account,
        std::uint32_t limit,
        bool forward,
        std::optional<TransactionsCursor> const& cursorIn,
        boost::asio::yield_context yield
    ) const = 0;

    std::optional<Blob>
    fetchLedgerObject(
        xrpl::uint256 const& key,
        std::uint32_t sequence,
        boost::asio::yield_context yield
    ) const;

    std::optional<std::uint32_t>
    fetchLedgerObjectSeq(
        xrpl::uint256 const& key,
        std::uint32_t sequence,
        boost::asio::yield_context yield
    ) const;

    std::vector<Blob>
    fetchLedgerObjects(
        std::vector<xrpl::uint256> const& keys,
        std::uint32_t sequence,
        boost::asio::yield_context yield
    ) const;

    virtual std::optional<Blob>
    doFetchLedgerObject(
        xrpl::uint256 const& key,
        std::uint32_t sequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual std::optional<std::uint32_t>
    doFetchLedgerObjectSeq(
        xrpl::uint256 const& key,
        std::uint32_t sequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual std::vector<Blob>
    doFetchLedgerObjects(
        std::vector<xrpl::uint256> const& keys,
        std::uint32_t sequence,
        boost::asio::yield_context yield
    ) const = 0;

    virtual std::vector<LedgerObject>
    fetchLedgerDiff(std::uint32_t ledgerSequence, boost::asio::yield_context yield) const = 0;

    LedgerPage
    fetchLedgerPage(
        std::optional<xrpl::uint256> const& cursor,
        std::uint32_t ledgerSequence,
        std::uint32_t limit,
        bool outOfOrder,
        boost::asio::yield_context yield
    );

    std::optional<LedgerObject>
    fetchSuccessorObject(
        xrpl::uint256 key,
        std::uint32_t ledgerSequence,
        boost::asio::yield_context yield
    ) const;

    std::optional<xrpl::uint256>
    fetchSuccessorKey(
        xrpl::uint256 key,
        std::uint32_t ledgerSequence,
        boost::asio::yield_context yield
    ) const;

    virtual std::optional<xrpl::uint256>
    doFetchSuccessorKey(
        xrpl::uint256 key,
        std::uint32_t ledgerSequence,
        boost::asio::yield_context yield
    ) const = 0;

    BookOffersPage
    fetchBookOffers(
        xrpl::uint256 const& book,
        std::uint32_t ledgerSequence,
        std::uint32_t limit,
        boost::asio::yield_context yield
    ) const;

    virtual std::optional<std::string>
    fetchMigratorStatus(
        std::string const& migratorName,
        boost::asio::yield_context yield
    ) const = 0;

    using ClioNodesDataFetchResult =
        std::expected<std::vector<std::pair<boost::uuids::uuid, std::string>>, std::string>;

    [[nodiscard]] virtual ClioNodesDataFetchResult
    fetchClioNodesData(boost::asio::yield_context yield) const = 0;

    std::optional<LedgerRange>
    hardFetchLedgerRange() const;

    virtual std::optional<LedgerRange>
    hardFetchLedgerRange(boost::asio::yield_context yield) const = 0;

    std::optional<LedgerRange>
    hardFetchLedgerRangeNoThrow() const;

    virtual void
    writeLedger(xrpl::LedgerHeader const& ledgerHeader, std::string&& blob) = 0;

    virtual void
    writeLedgerObject(std::string&& key, std::uint32_t seq, std::string&& blob);

    virtual void
    writeTransaction(
        std::string&& hash,
        std::uint32_t seq,
        std::uint32_t date,
        std::string&& transaction,
        std::string&& metadata
    ) = 0;

    virtual void
    writeNFTs(std::vector<NFTsData> const& data) = 0;

    virtual void
    writeAccountTransactions(std::vector<AccountTransactionsData> data) = 0;

    virtual void
    writeAccountTransaction(AccountTransactionsData record) = 0;

    virtual void
    writeNFTTransactions(std::vector<NFTTransactionsData> const& data) = 0;

    virtual void
    writeMPTokenIssuanceTransactions(std::vector<MPTokenIssuanceTransactionsData> const& data) = 0;

    virtual void
    writeAccountMPTokenIssuanceTransactions(
        std::vector<MPTokenIssuanceTransactionsData> const& data
    ) = 0;

    virtual void
    writeMPTHolders(std::vector<MPTHolderData> const& data) = 0;

    virtual void
    writeSuccessor(std::string&& key, std::uint32_t seq, std::string&& successor) = 0;

    virtual void
    writeNodeMessage(boost::uuids::uuid const& uuid, std::string message) = 0;

    virtual void
    startWrites() const = 0;

    bool
    finishWrites(std::uint32_t ledgerSequence);

    virtual void
    waitForWritesToFinish() = 0;

    virtual void
    writeMigratorStatus(std::string const& migratorName, std::string const& status) = 0;

    virtual bool
    isTooBusy() const = 0;

    virtual boost::json::object
    stats() const = 0;
 
private:
    virtual void
    doWriteLedgerObject(std::string&& key, std::uint32_t seq, std::string&& blob) = 0;

    virtual bool
    doFinishWrites() = 0;
 
    void
    updateRangeImpl(uint32_t newMax);
};
 
}  // namespace data
using BackendInterface = data::BackendInterface;