TxQ.h

//------------------------------------------------------------------------------
/*
    This file is part of rippled: https://github.com/ripple/rippled
    Copyright (c) 2012-19 Ripple Labs Inc.
 
    Permission to use, copy, modify, and/or 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.
*/
//==============================================================================
 
#ifndef RIPPLE_TXQ_H_INCLUDED
#define RIPPLE_TXQ_H_INCLUDED
 
#include <ripple/app/tx/applySteps.h>
#include <ripple/ledger/ApplyView.h>
#include <ripple/ledger/OpenView.h>
#include <ripple/protocol/RippleLedgerHash.h>
#include <ripple/protocol/STTx.h>
#include <ripple/protocol/SeqProxy.h>
#include <ripple/protocol/TER.h>
#include <boost/circular_buffer.hpp>
#include <boost/intrusive/set.hpp>
#include <optional>
 
namespace ripple {
 
class Application;
class Config;
 
class TxQ
{
public:
    static constexpr FeeLevel64 baseLevel{256};
 
    struct Setup
    {
        explicit Setup() = default;
 
        std::size_t ledgersInQueue = 20;
        std::size_t queueSizeMin = 2000;
        std::uint32_t retrySequencePercent = 25;
        FeeLevel64 minimumEscalationMultiplier = baseLevel * 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();
 
    std::pair<TER, bool>
    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(std::shared_ptr<SLE const> const& sleAccount) const;
 
    Metrics
    getMetrics(OpenView const& view) const;
 
    struct FeeAndSeq
    {
        XRPAmount fee;
        std::uint32_t accountSeq;
        std::uint32_t availableSeq;
    };
 
    FeeAndSeq
    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(
        std::shared_ptr<SLE const> const& sleAccount,
        std::lock_guard<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_(
                  setup.maximumTxnInLedger
                      ? *setup.maximumTxnInLedger < targetTxnCount_
                          ? targetTxnCount_
                          : *setup.maximumTxnInLedger
                      : std::optional<std::size_t>(std::nullopt))
            , 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;
        };
 
        Snapshot
        getSnapshot() const
        {
            return {txnsExpected_, 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;
        ApplyFlags const flags;
        std::optional<TER> lastResult;
        std::optional<PreflightResult const> pfresult;
 
        static constexpr int retriesAllowed = 10;
 
        static LedgerHash parentHashComp;
 
    public:
        MaybeTx(
            std::shared_ptr<STTx const> const&,
            TxID const& txID,
            FeeLevel64 feeLevel,
            ApplyFlags const flags,
            PreflightResult const& pfresult);
 
        std::pair<TER, bool>
        apply(Application& app, OpenView& view, beast::Journal j);
 
        TxConsequences const&
        consequences() const
        {
            return pfresult->consequences;
        }
 
        TxDetails
        getTxDetails() const
        {
            return {
                feeLevel,
                lastValid,
                consequences(),
                account,
                seqProxy,
                txn,
                retriesRemaining,
                pfresult->ter,
                lastResult};
        }
    };
 
    class OrderCandidates
    {
    public:
        explicit OrderCandidates() = default;
 
        bool
        operator()(const MaybeTx& lhs, const MaybeTx& 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(const AccountID& account);
 
        std::size_t
        getTxnCount() const
        {
            return transactions.size();
        }
 
        bool
        empty() const
        {
            return !getTxnCount();
        }
 
        TxMap::const_iterator
        getPrevTx(SeqProxy seqProx) const;
 
        MaybeTx&
        add(MaybeTx&&);
 
        bool
        remove(SeqProxy seqProx);
    };
 
    // Helper function returns requiredFeeLevel.
    FeeLevel64
    getRequiredFeeLevel(
        OpenView& view,
        ApplyFlags flags,
        FeeMetrics::Snapshot const& metricsSnapshot,
        std::lock_guard<std::mutex> const& lock) const;
 
    // Helper function for TxQ::apply.  If a transaction's fee is high enough,
    // attempt to directly apply that transaction to the ledger.
    std::optional<std::pair<TER, bool>>
    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_;
 
#if !NDEBUG
 
    LedgerHash parentHash_{beast::zero};
#endif
 
    std::mutex mutable mutex_;
 
private:
    template <size_t fillPercentage = 100>
    bool
    isFull() const;
 
    TER
    canBeHeld(
        STTx const&,
        ApplyFlags const,
        OpenView const&,
        std::shared_ptr<SLE const> const& sleAccount,
        AccountMap::iterator const&,
        std::optional<TxQAccount::TxMap::iterator> const&,
        std::lock_guard<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);
 
    std::pair<TER, bool>
    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
setup_TxQ(Config const&);
 
template <class T>
XRPAmount
toDrops(FeeLevel<T> const& level, XRPAmount baseFee)
{
    if (auto const drops = mulDiv(level, baseFee, TxQ::baseLevel); drops.first)
        return drops.second;
 
    return XRPAmount(STAmount::cMaxNativeN);
}
 
inline FeeLevel64
toFeeLevel(XRPAmount const& drops, XRPAmount const& baseFee)
{
    if (auto const feeLevel = mulDiv(drops, TxQ::baseLevel, baseFee);
        feeLevel.first)
        return feeLevel.second;
 
    return FeeLevel64(std::numeric_limits<std::uint64_t>::max());
}
 
}  // namespace ripple
 
#endif