// Copyright 2019 Stellar Development Foundation and contributors. Licensed

// under the Apache License, Version 2.0. See the COPYING file at the root

// of this distribution or at http://www.apache.org/licenses/LICENSE-2.0

#include "herder/TransactionQueue.h"

#include "crypto/Hex.h"

#include "crypto/SecretKey.h"

#include "herder/SurgePricingUtils.h"

#include "herder/TxQueueLimiter.h"

#include "ledger/LedgerHashUtils.h"

#include "ledger/LedgerManager.h"

#include "ledger/LedgerTxn.h"

#include "ledger/LedgerTxnImpl.h"

#include "main/Application.h"

#include "overlay/OverlayManager.h"

#include "transactions/FeeBumpTransactionFrame.h"

#include "transactions/MutableTransactionResult.h"

#include "transactions/OperationFrame.h"

#include "transactions/TransactionBridge.h"

#include "transactions/TransactionUtils.h"

#include "util/BitSet.h"

#include "util/GlobalChecks.h"

#include "util/HashOfHash.h"

#include "util/Math.h"

#include "util/MetricsRegistry.h"

#include "util/ProtocolVersion.h"

#include "util/TarjanSCCCalculator.h"

#include "util/XDROperators.h"

#include "util/numeric128.h"

#include <Tracy.hpp>

#include <algorithm>

#include <fmt/format.h>

#include <functional>

#include <limits>

#include <medida/meter.h>

#include <medida/timer.h>

#include <numeric>

#include <optional>

#include <random>

#ifdef BUILD_TESTS

#include "test/TxTests.h"

#include "transactions/test/TransactionTestFrame.h"

#endif

namespace stellar

{

uint64_t const TransactionQueue::FEE_MULTIPLIER = 10;

std::array<char const*,

static_cast<int>(TransactionQueue::AddResultCode::ADD_STATUS_COUNT)>

TX_STATUS_STRING = std::array{"PENDING", "DUPLICATE", "ERROR",

"TRY_AGAIN_LATER", "FILTERED"};

TransactionQueue::AddResult::AddResult(AddResultCode addCode)

: code(addCode), txResult()

{

}

TransactionQueue::AddResult::AddResult(

AddResultCode addCode, MutableTxResultPtr payload,

xdr::xvector<DiagnosticEvent>&& diagnostics)

: code(addCode)

, txResult(std::move(payload))

, mDiagnosticEvents(std::move(diagnostics))

{

releaseAssert(txResult);

}

TransactionQueue::AddResult::AddResult(AddResultCode addCode,

MutableTxResultPtr payload)

: code(addCode), txResult(std::move(payload))

{

releaseAssert(txResult);

}

TransactionQueue::AddResult::AddResult(AddResultCode addCode,

TransactionFrameBase const& tx,

TransactionResultCode txErrorCode)

: code(addCode), txResult(tx.createTxErrorResult(txErrorCode))

{

}

TransactionQueue::AddResult::AddResult(

AddResultCode addCode, TransactionFrameBase const& tx,

TransactionResultCode txErrorCode,

xdr::xvector<DiagnosticEvent>&& diagnostics)

: code(addCode)

, txResult(tx.createTxErrorResult(txErrorCode))

, mDiagnosticEvents(std::move(diagnostics))

{

}

TransactionQueue::TransactionQueue(Application& app, uint32 pendingDepth,

uint32 banDepth, uint32 poolLedgerMultiplier,

bool isSoroban)

: mApp(app)

, mPendingDepth(pendingDepth)

, mBannedTransactions(banDepth)

, mBroadcastTimer(app)

{

mTxQueueLimiter =

std::make_unique<TxQueueLimiter>(poolLedgerMultiplier, app, isSoroban);

auto const& filteredTypes =

app.getConfig().EXCLUDE_TRANSACTIONS_CONTAINING_OPERATION_TYPE;

mFilteredTypes.insert(filteredTypes.begin(), filteredTypes.end());

mBroadcastSeed =

rand_uniform<uint64>(0, std::numeric_limits<uint64>::max());

}

ClassicTransactionQueue::ClassicTransactionQueue(Application& app,

uint32 pendingDepth,

uint32 banDepth,

uint32 poolLedgerMultiplier)

: TransactionQueue(app, pendingDepth, banDepth, poolLedgerMultiplier, false)

// Arb tx damping is only relevant to classic txs

, mArbTxSeenCounter(

app.getMetrics().NewCounter({"herder", "arb-tx", "seen"}))

, mArbTxDroppedCounter(

app.getMetrics().NewCounter({"herder", "arb-tx", "dropped"}))

{

std::vector<medida::Counter*> sizeByAge;

for (uint32 i = 0; i < mPendingDepth; i++)

{

sizeByAge.emplace_back(&app.getMetrics().NewCounter(

{"herder", "pending-txs", fmt::format(FMT_STRING("age{:d}"), i)}));

}

mQueueMetrics = std::make_unique<QueueMetrics>(

sizeByAge,

app.getMetrics().NewCounter({"herder", "pending-txs", "banned"}),

app.getMetrics().NewSimpleTimer({"herder", "pending-txs"}),

app.getMetrics().NewSimpleTimer({"herder", "pending-txs", "self-"}),

app.getMetrics().NewCounter(

{"herder", "pending-txs", "evicted-due-to-low-fee-count"}),

app.getMetrics().NewCounter(

{"herder", "pending-txs", "evicted-due-to-age-count"}),

app.getMetrics().NewCounter(

{"herder", "pending-txs", "not-included-due-to-low-fee-count"}),

app.getMetrics().NewCounter(

{"herder", "pending-txs", "filtered-due-to-fp-keys"}));

mBroadcastOpCarryover.resize(1,

Resource::makeEmpty(NUM_CLASSIC_TX_RESOURCES));

}

bool

ClassicTransactionQueue::allowTxBroadcast(TransactionFrameBasePtr const& tx)

{

bool allowTx{true};

int32_t const signedAllowance =

mApp.getConfig().FLOOD_ARB_TX_BASE_ALLOWANCE;

if (signedAllowance >= 0)

{

uint32_t const allowance = static_cast<uint32_t>(signedAllowance);

// If arb tx damping is enabled, we only flood the first few arb txs

// touching an asset pair in any given ledger, exponentially

// reducing the odds of further arb ftx broadcast on a

// per-asset-pair basis. This lets _some_ arbitrage occur (and

// retains price-based competition among arbitrageurs earlier in the

// queue) but avoids filling up ledgers with excessive (mostly

// failed) arb attempts.

auto arbPairs = findAllAssetPairsInvolvedInPaymentLoops(tx);

if (!arbPairs.empty())

{

mArbTxSeenCounter.inc();

uint32_t maxBroadcast{0};

std::vector<

UnorderedMap<AssetPair, uint32_t, AssetPairHash>::iterator>

hashMapIters;

// NB: it's essential to reserve() on the hashmap so that we

// can store iterators to positions in it _as we emplace them_

// in the loop that follows, without rehashing. Do not remove.

mArbitrageFloodDamping.reserve(mArbitrageFloodDamping.size() +

arbPairs.size());

for (auto const& key : arbPairs)

{

auto pair = mArbitrageFloodDamping.emplace(key, 0);

hashMapIters.emplace_back(pair.first);

maxBroadcast = std::max(maxBroadcast, pair.first->second);

}

// Admit while no pair on the path has hit the allowance.

allowTx = maxBroadcast < allowance;

// If any pair is over the allowance, dampen transmission

// randomly based on it.

if (!allowTx)

{

std::geometric_distribution<uint32_t> dist(

mApp.getConfig().FLOOD_ARB_TX_DAMPING_FACTOR);

uint32_t k = maxBroadcast - allowance;

allowTx = dist(getGlobalRandomEngine()) >= k;

}

// If we've decided to admit a tx, bump all pairs on the path.

if (allowTx)

{

for (auto i : hashMapIters)

{

i->second++;

}

}

else

{

mArbTxDroppedCounter.inc();

}

}

}

return allowTx;

}

TransactionQueue::~TransactionQueue()

{

// empty destructor needed here due to the dependency on TxQueueLimiter

}

// returns true, if a transaction can be replaced by another

// `minFee` is set when returning false, and is the smallest _full_ fee

// that would allow replace by fee to succeed in this situation

// Note that replace-by-fee logic is done on _inclusion_ fee

static bool

canReplaceByFee(TransactionFrameBasePtr tx, TransactionFrameBasePtr oldTx,

int64_t& minFee)

{

int64_t newFee = tx->getInclusionFee();

uint32_t newNumOps = std::max<uint32_t>(1, tx->getNumOperations());

int64_t oldFee = oldTx->getInclusionFee();

uint32_t oldNumOps = std::max<uint32_t>(1, oldTx->getNumOperations());

// newFee / newNumOps >= FEE_MULTIPLIER * oldFee / oldNumOps

// is equivalent to

// newFee * oldNumOps >= FEE_MULTIPLIER * oldFee * newNumOps

//

// FEE_MULTIPLIER * oldTotalFee does not overflow uint128_t because fees

// are bounded by INT64_MAX, while number of operations and

// FEE_MULTIPLIER are small.

uint128_t oldTotalFee = bigMultiply(oldFee, newNumOps);

uint128_t minFeeN = oldTotalFee * TransactionQueue::FEE_MULTIPLIER;

bool res = newFee >= 0 && bigMultiply(newFee, oldNumOps) >= minFeeN;

if (!res)

{

if (!bigDivide128(minFee, minFeeN, int64_t(oldNumOps),

Rounding::ROUND_UP))

{

minFee = INT64_MAX;

}

else

{

// Add the potential flat component to the resulting min fee.

minFee += tx->getFullFee() - tx->getInclusionFee();

}

}

return res;

}

static bool

isDuplicateTx(TransactionFrameBasePtr oldTx, TransactionFrameBasePtr newTx)

{

auto const& oldEnv = oldTx->getEnvelope();

auto const& newEnv = newTx->getEnvelope();

if (oldEnv.type() == newEnv.type())

{

return oldTx->getFullHash() == newTx->getFullHash();

}

else if (oldEnv.type() == ENVELOPE_TYPE_TX_FEE_BUMP)

{

std::shared_ptr<FeeBumpTransactionFrame const> feeBumpPtr{};

#ifdef BUILD_TESTS

if (oldTx->isTestTx())

{

auto testFrame =

std::static_pointer_cast<TransactionTestFrame const>(oldTx);

feeBumpPtr =

std::static_pointer_cast<FeeBumpTransactionFrame const>(

testFrame->getTxFramePtr());

}

else

#endif

feeBumpPtr =

std::static_pointer_cast<FeeBumpTransactionFrame const>(oldTx);

return feeBumpPtr->getInnerFullHash() == newTx->getFullHash();

}

return false;

}

bool

TransactionQueue::sourceAccountPending(AccountID const& accountID) const

{

return mAccountStates.find(accountID) != mAccountStates.end();

}

TransactionQueue::AddResult

TransactionQueue::canAdd(

TransactionFrameBasePtr tx, AccountStates::iterator& stateIter,

std::vector<std::pair<TransactionFrameBasePtr, bool>>& txsToEvict

#ifdef BUILD_TESTS

,

bool isLoadgenTx

#endif

)

{

ZoneScoped;

if (isBanned(tx->getFullHash()))

{

#ifdef BUILD_TESTS

if (!mApp.getRunInOverlayOnlyMode())

#endif

{

return AddResult(

TransactionQueue::AddResultCode::ADD_STATUS_TRY_AGAIN_LATER);

}

}

if (isFiltered(tx))

{

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_FILTERED);

}

if (!tx->validateSorobanTxForFlooding(mKeysToFilter))

{

mQueueMetrics->mTxsFilteredDueToFootprintKeys.inc();

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_FILTERED);

}

int64_t newFullFee = tx->getFullFee();

if (newFullFee < 0 || tx->getInclusionFee() < 0)

{

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR, *tx,

txMALFORMED);

}

stateIter = mAccountStates.find(tx->getSourceID());

TransactionFrameBasePtr currentTx;

auto ledgerVersion = mApp.getLedgerManager()

.getLastClosedLedgerHeader()

.header.ledgerVersion;

auto diagnosticEvents =

DiagnosticEventManager::createForValidation(mApp.getConfig());

if (stateIter != mAccountStates.end())

{

auto const& transaction = stateIter->second.mTransaction;

if (transaction)

{

currentTx = transaction->mTx;

// Check if the tx is a duplicate

if (isDuplicateTx(currentTx, tx))

{

return AddResult(

TransactionQueue::AddResultCode::ADD_STATUS_DUPLICATE);

}

// Any transaction older than the current one is invalid

if (tx->getSeqNum() < currentTx->getSeqNum())

{

// If the transaction is older than the one in the queue, we

// reject it

return AddResult(

TransactionQueue::AddResultCode::ADD_STATUS_ERROR, *tx,

txBAD_SEQ);

}

// Before rejecting Soroban transactions due to source account

// limit, check validity of its declared resources, and return an

// appropriate error message

if (tx->isSoroban())

{

if (!tx->checkSorobanResources(

mApp.getLedgerManager()

.getLastClosedSorobanNetworkConfig(),

ledgerVersion, diagnosticEvents))

{

return AddResult(AddResultCode::ADD_STATUS_ERROR, *tx,

txSOROBAN_INVALID,

diagnosticEvents.finalize());

}

}

if (tx->getEnvelope().type() != ENVELOPE_TYPE_TX_FEE_BUMP)

{

// If there's already a transaction in the queue, we reject

// any new transaction

return AddResult(TransactionQueue::AddResultCode::

ADD_STATUS_TRY_AGAIN_LATER);

}

else

{

if (tx->getSeqNum() != currentTx->getSeqNum())

{

// New fee-bump transaction is rejected

return AddResult(TransactionQueue::AddResultCode::

ADD_STATUS_TRY_AGAIN_LATER);

}

int64_t minFee;

if (!canReplaceByFee(tx, currentTx, minFee))

{

auto txResult = tx->createTxErrorResult(txINSUFFICIENT_FEE);

txResult->setInsufficientFeeErrorWithFeeCharged(minFee);

return AddResult(

TransactionQueue::AddResultCode::ADD_STATUS_ERROR,

std::move(txResult));

}

if (currentTx->getFeeSourceID() == tx->getFeeSourceID())

{

newFullFee -= currentTx->getFullFee();

}

}

}

}

LedgerSnapshot ls(mApp);

// Subtle: transactions are rejected based on the source account limit

// prior to this point. This is safe because we can't evict transactions

// from the same source account, so a newer transaction won't replace an

// old one.

auto canAddRes = mTxQueueLimiter->canAddTx(tx, currentTx, txsToEvict,

ledgerVersion, mBroadcastSeed);

if (!canAddRes.first)

{

ban({tx});

mQueueMetrics->mTxsNotAcceptedDueToLowFeeCounter.inc();

if (canAddRes.second != 0)

{

auto txResult = tx->createValidationSuccessResult();

txResult->setInsufficientFeeErrorWithFeeCharged(canAddRes.second);

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR,

std::move(txResult));

}

return AddResult(

TransactionQueue::AddResultCode::ADD_STATUS_TRY_AGAIN_LATER);

}

auto closeTime = mApp.getLedgerManager()

.getLastClosedLedgerHeader()

.header.scpValue.closeTime;

if (protocolVersionStartsFrom(ledgerVersion, ProtocolVersion::V_19))

{

// This is done so minSeqLedgerGap is validated against the next

// ledgerSeq, which is what will be used at apply time

ls.getLedgerHeader().currentToModify().ledgerSeq =

mApp.getLedgerManager().getLastClosedLedgerNum() + 1;

}

// Loadgen txs were generated by this local node, and therefore can skip

// validation, and be added directly to the queue.

#ifdef BUILD_TESTS

if (!isLoadgenTx)

#endif

{

auto validationResult = tx->checkValid(

mApp.getAppConnector(), ls, 0, 0,

getUpperBoundCloseTimeOffset(mApp, closeTime), diagnosticEvents);

if (!validationResult->isSuccess())

{

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR,

std::move(validationResult),

diagnosticEvents.finalize());

}

}

// Note: stateIter corresponds to getSourceID() which is not necessarily

// the same as getFeeSourceID()

// Loadgen transactions are given unlimited funds, and therefore do no need

// to be checked for fees

#ifdef BUILD_TESTS

if (!isLoadgenTx && !mApp.getRunInOverlayOnlyMode())

#endif

{

auto const feeSource = ls.getAccount(tx->getFeeSourceID());

auto feeStateIter = mAccountStates.find(tx->getFeeSourceID());

int64_t totalFees = feeStateIter == mAccountStates.end()

? 0

: feeStateIter->second.mTotalFees;

if (getAvailableBalance(ls.getLedgerHeader().current(),

feeSource.current()) -

newFullFee <

totalFees)

{

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR,

*tx, txINSUFFICIENT_BALANCE);

}

}

if (protocolVersionIsBefore(ledgerVersion, ProtocolVersion::V_25) &&

!tx->validateSorobanMemo())

{

diagnosticEvents.pushError(SCE_VALUE, SCEC_INVALID_INPUT,

"Soroban transactions are not allowed to "

"use memo or muxed source account");

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR, *tx,

txSOROBAN_INVALID, diagnosticEvents.finalize());

}

if (!tx->validateHostFn())

{

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR, *tx,

txSOROBAN_INVALID, diagnosticEvents.finalize());

}

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_PENDING,

tx->createValidationSuccessResult());

}

void

TransactionQueue::releaseFeeMaybeEraseAccountState(TransactionFrameBasePtr tx)

{

auto iter = mAccountStates.find(tx->getFeeSourceID());

releaseAssert(iter != mAccountStates.end() &&

iter->second.mTotalFees >= tx->getFullFee());

iter->second.mTotalFees -= tx->getFullFee();

if (!iter->second.mTransaction && iter->second.mTotalFees == 0)

{

mAccountStates.erase(iter);

}

}

void

TransactionQueue::prepareDropTransaction(AccountState& as)

{

releaseAssert(as.mTransaction);

mTxQueueLimiter->removeTransaction(as.mTransaction->mTx);

mKnownTxHashes.erase(as.mTransaction->mTx->getFullHash());

CLOG_DEBUG(Tx, "Dropping {} transaction",

hexAbbrev(as.mTransaction->mTx->getFullHash()));

releaseFeeMaybeEraseAccountState(as.mTransaction->mTx);

}

// Heuristic: an "arbitrage transaction" as identified by this function as

// any tx that has 1 or more path payments in it that collectively form a

// payment _loop_. That is: a tx that performs a sequence of order-book

// conversions of at least some quantity of some asset _back_ to itself via

// some number of intermediate steps. Typically these are only a single

// path-payment op, but for thoroughness sake we're also going to cover

// cases where there's any atomic _sequence_ of path payment ops that cause

// a conversion-loop.

//

// Such transactions are not going to be outright banned, note: just damped

// so that they do not overload the network. Currently people are submitting

// thousands of such txs per second in an attempt to win races for

// arbitrage, and we just want to make those races a behave more like

// bidding wars than pure resource-wasting races.

//

// This function doesn't catch all forms of arbitrage -- there are an

// unlimited number of types, many of which involve holding assets,

// interacting with real-world actors, etc. and are indistinguishable from

// "real" traffic -- but it does cover the case of zero-risk (fee-only)

// instantaneous-arbitrage attempts, which users are (at the time of

// writing) flooding the network with.

std::vector<AssetPair>

TransactionQueue::findAllAssetPairsInvolvedInPaymentLoops(

TransactionFrameBasePtr tx)

{

std::map<Asset, size_t> assetToNum;

std::vector<Asset> numToAsset;

std::vector<BitSet> graph;

auto internAsset = [&](Asset const& a) -> size_t {

size_t n = numToAsset.size();

auto pair = assetToNum.emplace(a, n);

if (pair.second)

{

numToAsset.emplace_back(a);

graph.emplace_back(BitSet());

}

return pair.first->second;

};

auto internEdge = [&](Asset const& src, Asset const& dst) {

auto si = internAsset(src);

auto di = internAsset(dst);

graph.at(si).set(di);

};

auto internSegment = [&](Asset const& src, Asset const& dst,

std::vector<Asset> const& path) {

Asset const* prev = &src;

for (auto const& a : path)

{

internEdge(*prev, a);

prev = &a;

}

internEdge(*prev, dst);

};

for (auto const& op : tx->getRawOperations())

{

switch (op.body.type())

{

case PATH_PAYMENT_STRICT_RECEIVE:

{

auto const& pop = op.body.pathPaymentStrictReceiveOp();

internSegment(pop.sendAsset, pop.destAsset, pop.path);

}

break;

case PATH_PAYMENT_STRICT_SEND:

{

auto const& pop = op.body.pathPaymentStrictSendOp();

internSegment(pop.sendAsset, pop.destAsset, pop.path);

}

break;

default:

continue;

}

}

// We build a TarjanSCCCalculator for the graph of all the edges we've

// seen, and return the set of edges that participate in nontrivial SCCs

// (which are loops). This is O(|v| + |e|) and just operations on a

// vector of pairs of integers.

TarjanSCCCalculator tsc;

tsc.calculateSCCs(graph.size(), [&graph](size_t i) -> BitSet const& {

// NB: this closure must be written with the explicit const&

// returning type signature, otherwise it infers wrong and

// winds up returning a dangling reference at its site of use.

return graph.at(i);

});

std::vector<AssetPair> ret;

for (BitSet const& scc : tsc.mSCCs)

{

if (scc.count() > 1)

{

for (size_t src = 0; scc.nextSet(src); ++src)

{

BitSet edgesFromSrcInSCC = graph.at(src);

edgesFromSrcInSCC.inplaceIntersection(scc);

for (size_t dst = 0; edgesFromSrcInSCC.nextSet(dst); ++dst)

{

ret.emplace_back(

AssetPair{numToAsset.at(src), numToAsset.at(dst)});

}

}

}

}

return ret;

}

TransactionQueue::AddResult

TransactionQueue::tryAdd(TransactionFrameBasePtr tx, bool submittedFromSelf

#ifdef BUILD_TESTS

,

bool isLoadgenTx

#endif

)

{

ZoneScoped;

if (!tx->XDRProvidesValidFee())

{

return AddResult(TransactionQueue::AddResultCode::ADD_STATUS_ERROR, *tx,

txMALFORMED);

}

AccountStates::iterator stateIter;

std::vector<std::pair<TransactionFrameBasePtr, bool>> txsToEvict;

auto res = canAdd(tx, stateIter, txsToEvict

#ifdef BUILD_TESTS

,

isLoadgenTx

#endif

);

if (res.code != TransactionQueue::AddResultCode::ADD_STATUS_PENDING)

{

return res;

}

// only evict if successful

if (stateIter == mAccountStates.end())

{

stateIter =

mAccountStates.emplace(tx->getSourceID(), AccountState{}).first;

}

auto& oldTx = stateIter->second.mTransaction;

if (oldTx)

{

// Drop current transaction associated with this account, replace

// with `tx`

prepareDropTransaction(stateIter->second);

*oldTx = {tx, mApp.getClock().now(), submittedFromSelf};

}

else

{

// New transaction for this account, insert it and update age

stateIter->second.mTransaction = {tx, mApp.getClock().now(),

submittedFromSelf};

mQueueMetrics->mSizeByAge[stateIter->second.mAge]->inc();

}

// Update fee accounting

auto& thisAccountState = mAccountStates[tx->getFeeSourceID()];

thisAccountState.mTotalFees += tx->getFullFee();

// make space so that we can add this transaction

// this will succeed as `canAdd` ensures that this is the case

int evictedCount = 0;

mTxQueueLimiter->evictTransactions(

txsToEvict, *tx,

[this, &evictedCount](TransactionFrameBasePtr const& txToEvict) {

++evictedCount;

ban({txToEvict});

});

mQueueMetrics->mTxsEvictedByHigherFeeTxCounter.inc(evictedCount);

mTxQueueLimiter->addTransaction(tx);

mKnownTxHashes[tx->getFullHash()] = tx;

broadcast(false);

return res;

}

void

TransactionQueue::dropTransaction(AccountStates::iterator stateIter)

{

ZoneScoped;

// Remove fees and update queue size for each transaction to be dropped.

// Note prepareDropTransaction may erase other iterators from

// mAccountStates, but it will not erase stateIter because it has at

// least one transaction (otherwise we couldn't reach that line).

releaseAssert(stateIter->second.mTransaction);

prepareDropTransaction(stateIter->second);

// Actually erase the transaction to be dropped.

stateIter->second.mTransaction.reset();

// If the queue for stateIter is now empty, then (1) erase it if it is

// not the fee-source for some other transaction or (2) reset the age

// otherwise.

if (stateIter->second.mTotalFees == 0)

{

mAccountStates.erase(stateIter);

}

else

{

stateIter->second.mAge = 0;

}

}

void

TransactionQueue::removeApplied(Transactions const& appliedTxs)

{

ZoneScoped;

auto now = mApp.getClock().now();

for (auto const& appliedTx : appliedTxs)

{

// If the source account is not in mAccountStates, then it has no

// transactions in the queue so there is nothing to do

auto stateIter = mAccountStates.find(appliedTx->getSourceID());

if (stateIter != mAccountStates.end())

{

// If there are no transactions in the queue for this source

// account, then there is nothing to do

auto const& transaction = stateIter->second.mTransaction;

if (transaction)

{

// We care about matching the sequence number rather than

// the hash, because any transaction with a sequence number

// less-than-or-equal to the highest applied sequence number

// for this source account has either (1) been applied, or

// (2) become invalid.

if (transaction->mTx->getSeqNum() <= appliedTx->getSeqNum())

{

auto& age = stateIter->second.mAge;

mQueueMetrics->mSizeByAge[age]->dec();

age = 0;

// update the metric for the time spent for applied

// transactions using exact match

if (transaction->mTx->getFullHash() ==

appliedTx->getFullHash())

{

auto elapsed = now - transaction->mInsertionTime;

mQueueMetrics->mTransactionsDelay.Update(elapsed);

if (transaction->mSubmittedFromSelf)

{

mQueueMetrics->mTransactionsSelfDelay.Update(

elapsed);

}

}

// WARNING: stateIter and everything that references it

// may be invalid from this point onward and should not

// be used.

dropTransaction(stateIter);

}

}

}

// Ban applied tx

auto& bannedFront = mBannedTransactions.front();

bannedFront.emplace(appliedTx->getFullHash());

CLOG_DEBUG(Tx, "Ban applied transaction {}",

hexAbbrev(appliedTx->getFullHash()));

// do not mark metric for banning as this is the result of normal

// flow of operations

}

}

void

TransactionQueue::ban(Transactions const& banTxs)

{

ZoneScoped;

auto& bannedFront = mBannedTransactions.front();

// Group the transactions by source account and ban all the transactions

// that are explicitly listed

std::map<AccountID, TransactionFrameBasePtr> transactionsByAccount;

for (auto const& tx : banTxs)

{

// Must be a new transaction for this account

releaseAssert(

transactionsByAccount.emplace(tx->getSourceID(), tx).second);

CLOG_DEBUG(Tx, "Ban transaction {}", hexAbbrev(tx->getFullHash()));

if (bannedFront.emplace(tx->getFullHash()).second)

{

mQueueMetrics->mBannedTransactionsCounter.inc();

}

}

for (auto const& kv : transactionsByAccount)

{

// If the source account is not in mAccountStates, then it has no

// transactions in the queue so there is nothing to do

auto stateIter = mAccountStates.find(kv.first);

if (stateIter != mAccountStates.end())

{

auto const& transaction = stateIter->second.mTransaction;

// Only ban transactions that are actually present in the queue.

// Transactions with higher sequence numbers than banned

// transactions remain in the queue.

if (transaction &&

transaction->mTx->getFullHash() == kv.second->getFullHash())

{

mQueueMetrics->mSizeByAge[stateIter->second.mAge]->dec();

// WARNING: stateIter and everything that references it may

// be invalid from this point onward and should not be used.

dropTransaction(stateIter);

}

}

}

}

#ifdef BUILD_TESTS

TransactionQueue::AccountState

TransactionQueue::getAccountTransactionQueueInfo(

AccountID const& accountID) const

{

auto i = mAccountStates.find(accountID);

if (i == std::end(mAccountStates))

{

return AccountState{};

}

return i->second;

}

size_t

TransactionQueue::countBanned(int index) const

{

return mBannedTransactions[index].size();

}

#endif

void

TransactionQueue::shift()

{

ZoneScoped;

mBannedTransactions.pop_back();

mBannedTransactions.emplace_front();

mArbitrageFloodDamping.clear();

auto sizes = std::vector<int64_t>{};

sizes.resize(mPendingDepth);

auto& bannedFront = mBannedTransactions.front();

auto end = std::end(mAccountStates);

auto it = std::begin(mAccountStates);

int evictedDueToAge = 0;

while (it != end)

{

// If mTransactions is empty then mAge is always 0. This can occur

// if an account is the fee-source for at least one transaction but

// not the sequence-number-source for any transaction in the

// TransactionQueue.

if (it->second.mTransaction)

{

++it->second.mAge;

}

if (mPendingDepth == it->second.mAge)

{

if (it->second.mTransaction)

{

// This never invalidates it because

// it->second.mTransaction

// otherwise we couldn't have reached this line.

prepareDropTransaction(it->second);

CLOG_DEBUG(

Tx, "Ban transaction {}",

hexAbbrev(it->second.mTransaction->mTx->getFullHash()));

bannedFront.insert(it->second.mTransaction->mTx->getFullHash());

mQueueMetrics->mBannedTransactionsCounter.inc();

it->second.mTransaction.reset();

++evictedDueToAge;

}

if (it->second.mTotalFees == 0)

{

it = mAccountStates.erase(it);

}

else

{

it->second.mAge = 0;

}

}

else

{

sizes[it->second.mAge] +=

static_cast<int>(it->second.mTransaction.has_value());

++it;

}

}

mQueueMetrics->mTxsEvictedDueToAgeCounter.inc(evictedDueToAge);

for (size_t i = 0; i < sizes.size(); i++)

{

mQueueMetrics->mSizeByAge[i]->set_count(sizes[i]);

}

mTxQueueLimiter->resetEvictionState();

// pick a new randomizing seed for tie breaking

mBroadcastSeed =

rand_uniform<uint64>(0, std::numeric_limits<uint64>::max());

// Reset flood queue with the new seed (this will drop all existing

// non-broadcasted transactions, which will be re-added in `rebroadcast`)

mTxQueueLimiter->resetBestFeeTxs(mApp.getLedgerManager()

.getLastClosedLedgerHeader()

.header.ledgerVersion,

mBroadcastSeed);

}

bool

TransactionQueue::isBanned(Hash const& hash) const

{

return std::any_of(

std::begin(mBannedTransactions), std::end(mBannedTransactions),

[&](UnorderedSet<Hash> const& transactions) {

return transactions.find(hash) != std::end(transactions);

});

}

TxFrameList

TransactionQueue::getTransactions(LedgerHeader const& lcl) const

{

ZoneScoped;

TxFrameList txs;

uint32_t const nextLedgerSeq = lcl.ledgerSeq + 1;

int64_t const startingSeq = getStartingSequenceNumber(nextLedgerSeq);

for (auto const& m : mAccountStates)

{

if (m.second.mTransaction &&

m.second.mTransaction->mTx->getSeqNum() != startingSeq)

{

txs.emplace_back(m.second.mTransaction->mTx);

}

}

return txs;

}

TransactionFrameBaseConstPtr

TransactionQueue::getTx(Hash const& hash) const

{

ZoneScoped;

auto it = mKnownTxHashes.find(hash);

if (it != mKnownTxHashes.end())

{

return it->second;

}

else

{

return nullptr;

}

}