TxQ.h

#pragma once
 
#include <xrpl/beast/utility/Journal.h>
#include <xrpl/beast/utility/Zero.h>
#include <xrpl/json/json_value.h>
#include <xrpl/ledger/ApplyView.h>
#include <xrpl/ledger/OpenView.h>
#include <xrpl/ledger/ReadView.h>
#include <xrpl/protocol/AccountID.h>
#include <xrpl/protocol/Protocol.h>
#include <xrpl/protocol/RippleLedgerHash.h>
#include <xrpl/protocol/STAmount.h>
#include <xrpl/protocol/STLedgerEntry.h>
#include <xrpl/protocol/STTx.h>
#include <xrpl/protocol/SeqProxy.h>
#include <xrpl/protocol/TER.h>
#include <xrpl/protocol/Units.h>
#include <xrpl/protocol/XRPAmount.h>
#include <xrpl/tx/applySteps.h>
 
#include <boost/circular_buffer.hpp>
#include <boost/intrusive/set.hpp>
 
#include <cstddef>
#include <cstdint>
#include <expected>
#include <limits>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <utility>
#include <vector>
 
namespace xrpl {
 
class Application;
class Config;

class TxQ
{
public:
    static constexpr FeeLevel64 kBaseLevel{256};

    struct Setup
    {
        explicit Setup() = default;

        std::size_t ledgersInQueue = 20;
        std::size_t queueSizeMin = 2000;
        std::uint32_t retrySequencePercent = 25;
        FeeLevel64 minimumEscalationMultiplier = kBaseLevel * 500;
        std::uint32_t minimumTxnInLedger = 32;
        std::uint32_t minimumTxnInLedgerSA = 1000;
        std::uint32_t targetTxnInLedger = 256;
        std::optional<std::uint32_t> maximumTxnInLedger;
        std::uint32_t normalConsensusIncreasePercent = 20;
        std::uint32_t slowConsensusDecreasePercent = 50;
        std::uint32_t maximumTxnPerAccount = 10;
        std::uint32_t minimumLastLedgerBuffer = 2;
        bool standAlone = false;
    };

    struct Metrics
    {
        explicit Metrics() = default;

        std::size_t txCount{};
        std::optional<std::size_t> txQMaxSize;
        std::size_t txInLedger{};
        std::size_t txPerLedger{};
        FeeLevel64 referenceFeeLevel{};
        FeeLevel64 minProcessingFeeLevel{};
        FeeLevel64 medFeeLevel{};
        FeeLevel64 openLedgerFeeLevel{};
    };

    struct TxDetails
    {
        TxDetails(
            FeeLevel64 feeLevel,
            std::optional<LedgerIndex> const& lastValid,
            TxConsequences const& consequences,
            AccountID const& account,
            SeqProxy seqProxy,
            std::shared_ptr<STTx const> const& txn,
            int retriesRemaining,
            TER preflightResult,
            std::optional<TER> lastResult)
            : feeLevel(feeLevel)
            , lastValid(lastValid)
            , consequences(consequences)
            , account(account)
            , seqProxy(seqProxy)
            , txn(txn)
            , retriesRemaining(retriesRemaining)
            , preflightResult(preflightResult)
            , lastResult(lastResult)
        {
        }

        FeeLevel64 feeLevel;
        std::optional<LedgerIndex> lastValid;
        TxConsequences consequences;
        AccountID account;
        SeqProxy seqProxy;
        std::shared_ptr<STTx const> txn;
        int retriesRemaining;
        TER preflightResult;
        std::optional<TER> lastResult;
    };

    TxQ(Setup const& setup, beast::Journal j);

    virtual ~TxQ();

    ApplyResult
    apply(
        Application& app,
        OpenView& view,
        std::shared_ptr<STTx const> const& tx,
        ApplyFlags flags,
        beast::Journal j);

    bool
    accept(Application& app, OpenView& view);

    void
    processClosedLedger(Application& app, ReadView const& view, bool timeLeap);

    SeqProxy
    nextQueuableSeq(SLE::ConstRef sleAccount) const;

    Metrics
    getMetrics(OpenView const& view) const;
 
    struct FeeAndSeq
    {
        XRPAmount fee;
        std::uint32_t accountSeq;
        std::uint32_t availableSeq;
    };

    std::expected<FeeAndSeq, TER>
    getTxRequiredFeeAndSeq(OpenView const& view, std::shared_ptr<STTx const> const& tx) const;

    std::vector<TxDetails>
    getAccountTxs(AccountID const& account) const;

    std::vector<TxDetails>
    getTxs() const;

    json::Value
    doRPC(Application& app) const;
 
private:
    // Implementation for nextQueuableSeq().  The passed lock must be held.
    SeqProxy
    nextQueuableSeqImpl(SLE::ConstRef sleAccount, std::scoped_lock<std::mutex> const&) const;

    class FeeMetrics
    {
    private:
        std::size_t const minimumTxnCount_;
        std::size_t const targetTxnCount_;
        std::optional<std::size_t> const maximumTxnCount_;
        std::size_t txnsExpected_;
        boost::circular_buffer<std::size_t> recentTxnCounts_;
        FeeLevel64 escalationMultiplier_;
        beast::Journal const j_;
 
    public:
        FeeMetrics(Setup const& setup, beast::Journal j)
            : minimumTxnCount_(
                  setup.standAlone ? setup.minimumTxnInLedgerSA : setup.minimumTxnInLedger)
            , targetTxnCount_(
                  setup.targetTxnInLedger < minimumTxnCount_ ? minimumTxnCount_
                                                             : setup.targetTxnInLedger)
            , maximumTxnCount_([&]() -> std::optional<std::size_t> {
                if (!setup.maximumTxnInLedger)
                    return std::nullopt;
                return *setup.maximumTxnInLedger < targetTxnCount_ ? targetTxnCount_
                                                                   : *setup.maximumTxnInLedger;
            }())
            , txnsExpected_(minimumTxnCount_)
            , recentTxnCounts_(setup.ledgersInQueue)
            , escalationMultiplier_(setup.minimumEscalationMultiplier)
            , j_(j)
        {
        }

        std::size_t
        update(Application& app, ReadView const& view, bool timeLeap, TxQ::Setup const& setup);

        struct Snapshot
        {
            // Number of transactions expected per ledger.
            // One more than this value will be accepted
            // before escalation kicks in.
            std::size_t const txnsExpected{};
            // Based on the median fee of the LCL. Used
            // when fee escalation kicks in.
            FeeLevel64 const escalationMultiplier;
        };

        [[nodiscard]] Snapshot
        getSnapshot() const
        {
            return {.txnsExpected = txnsExpected_, .escalationMultiplier = escalationMultiplier_};
        }

        static FeeLevel64
        scaleFeeLevel(Snapshot const& snapshot, OpenView const& view);

        static std::pair<bool, FeeLevel64>
        escalatedSeriesFeeLevel(
            Snapshot const& snapshot,
            OpenView const& view,
            std::size_t extraCount,
            std::size_t seriesSize);
    };

    class MaybeTx
    {
    public:
        boost::intrusive::set_member_hook<> byFeeListHook;

        std::shared_ptr<STTx const> txn;

        FeeLevel64 const feeLevel;
        TxID const txID;
        AccountID const account;
        std::optional<LedgerIndex> const lastValid;
        SeqProxy const seqProxy;
        int retriesRemaining{kRetriesAllowed};
        ApplyFlags const flags;
        std::optional<TER> lastResult;
        std::optional<PreflightResult const> pfResult;

        static constexpr int kRetriesAllowed = 10;

        static LedgerHash parentHashComp;
 
    public:
        MaybeTx(
            std::shared_ptr<STTx const> const&,
            TxID const& txID,
            FeeLevel64 feeLevel,
            ApplyFlags const flags,
            PreflightResult const& pfResult);

        ApplyResult
        apply(Application& app, OpenView& view, beast::Journal j);

        [[nodiscard]] TxConsequences const&
        consequences() const
        {
            return pfResult->consequences;  // NOLINT(bugprone-unchecked-optional-access) invariant:
                                            // pfResult is never empty
        }

        [[nodiscard]] TxDetails
        getTxDetails() const
        {
            return {
                feeLevel,
                lastValid,
                consequences(),
                account,
                seqProxy,
                txn,
                retriesRemaining,
                pfResult->ter,  // NOLINT(bugprone-unchecked-optional-access) invariant: pfResult is
                                // never empty
                lastResult};
        }
    };

    class OrderCandidates
    {
    public:
        explicit OrderCandidates() = default;

        bool
        operator()(MaybeTx const& lhs, MaybeTx const& rhs) const
        {
            if (lhs.feeLevel == rhs.feeLevel)
                return (lhs.txID ^ MaybeTx::parentHashComp) < (rhs.txID ^ MaybeTx::parentHashComp);
            return lhs.feeLevel > rhs.feeLevel;
        }
    };

    class TxQAccount
    {
    public:
        using TxMap = std::map<SeqProxy, MaybeTx>;

        AccountID const account;
        TxMap transactions;
        /* If this account has had any transaction retry more than
            `retriesAllowed` times so that it was dropped from the
            queue, then all other transactions for this account will
            be given at most 2 attempts before being removed. Helps
            prevent wasting resources on retries that are more likely
            to fail.
        */
        bool retryPenalty = false;
        /* If this account has had any transaction fail or expire,
            then when the queue is nearly full, transactions from
            this account will be discarded. Helps prevent the queue
            from getting filled and wedged.
        */
        bool dropPenalty = false;
 
    public:
        explicit TxQAccount(std::shared_ptr<STTx const> const& txn);
        explicit TxQAccount(AccountID const& account);

        [[nodiscard]] std::size_t
        getTxnCount() const
        {
            return transactions.size();
        }

        [[nodiscard]] bool
        empty() const
        {
            return getTxnCount() == 0u;
        }

        [[nodiscard]] TxMap::const_iterator
        getPrevTx(SeqProxy seqProx) const;

        MaybeTx&
        add(MaybeTx&&);

        bool
        remove(SeqProxy seqProx);
    };
 
    // Helper function returns requiredFeeLevel.
    static FeeLevel64
    getRequiredFeeLevel(
        OpenView& view,
        ApplyFlags flags,
        FeeMetrics::Snapshot const& metricsSnapshot,
        std::scoped_lock<std::mutex> const& lock);
 
    // Helper function for TxQ::apply.  If a transaction's fee is high enough,
    // attempt to directly apply that transaction to the ledger.
    std::optional<ApplyResult>
    tryDirectApply(
        Application& app,
        OpenView& view,
        std::shared_ptr<STTx const> const& tx,
        ApplyFlags flags,
        beast::Journal j);
 
    // Helper function that removes a replaced entry in _byFee.
    std::optional<TxQAccount::TxMap::iterator>
    removeFromByFee(
        std::optional<TxQAccount::TxMap::iterator> const& replacedTxIter,
        std::shared_ptr<STTx const> const& tx);
 
    using FeeHook = boost::intrusive::
        member_hook<MaybeTx, boost::intrusive::set_member_hook<>, &MaybeTx::byFeeListHook>;
 
    using FeeMultiSet =
        boost::intrusive::multiset<MaybeTx, FeeHook, boost::intrusive::compare<OrderCandidates>>;
 
    using AccountMap = std::map<AccountID, TxQAccount>;

    Setup const setup_;
    beast::Journal const j_;

    FeeMetrics feeMetrics_;
    FeeMultiSet byFee_;
    AccountMap byAccount_;
    std::optional<size_t> maxSize_;

    LedgerHash parentHash_{beast::kZero};

    std::mutex mutable mutex_;
 
private:
    template <size_t FillPercentage = 100>
    bool
    isFull() const;

    TER
    canBeHeld(
        STTx const&,
        ApplyFlags const,
        OpenView const&,
        SLE::ConstRef sleAccount,
        AccountMap::iterator const&,
        std::optional<TxQAccount::TxMap::iterator> const&,
        std::scoped_lock<std::mutex> const& lock);

    FeeMultiSet::iterator_type erase(FeeMultiSet::const_iterator_type);
    FeeMultiSet::iterator_type eraseAndAdvance(FeeMultiSet::const_iterator_type);
    TxQAccount::TxMap::iterator
    erase(
        TxQAccount& txQAccount,
        TxQAccount::TxMap::const_iterator begin,
        TxQAccount::TxMap::const_iterator end);

    ApplyResult
    tryClearAccountQueueUpThruTx(
        Application& app,
        OpenView& view,
        STTx const& tx,
        AccountMap::iterator const& accountIter,
        TxQAccount::TxMap::iterator,
        FeeLevel64 feeLevelPaid,
        PreflightResult const& pfResult,
        std::size_t const txExtraCount,
        ApplyFlags flags,
        FeeMetrics::Snapshot const& metricsSnapshot,
        beast::Journal j);
};

TxQ::Setup
setupTxQ(Config const&);
 
template <class T>
XRPAmount
toDrops(FeeLevel<T> const& level, XRPAmount baseFee)
{
    return mulDiv(level, baseFee, TxQ::kBaseLevel).value_or(XRPAmount(STAmount::kMaxNativeN));
}
 
inline FeeLevel64
toFeeLevel(XRPAmount const& drops, XRPAmount const& baseFee)
{
    return mulDiv(drops, TxQ::kBaseLevel, baseFee)
        .value_or(FeeLevel64(std::numeric_limits<std::uint64_t>::max()));
}
 
}  // namespace xrpl