Repository: bitcoin/bitcoin
File: src/private_broadcast.cpp (C++)

Code before:

#include <private_broadcast.h>
#include <util/check.h>

#include <algorithm>
#include <ranges>

/// If a transaction is not received back from the network for this duration
/// after it is broadcast, then we consider it stale / for rebroadcasting.
static constexpr auto STALE_DURATION{1min};
static constexpr size_t MAX_STORED_TRANSACTIONS{1'000};

bool PrivateBroadcast::Add(const CTransactionRef& tx)
    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
    LOCK(m_mutex);
    auto [it, inserted]{m_transactions.try_emplace(tx)};
    if (!inserted && it->second.final_state.has_value()) {
        // Re-schedule a previously finished transaction for private broadcast again.
        it->second.final_state.reset();
        it->second.sent_to.clear();
        inserted = true;
    }

    // Bound memory usage by pruning oldest finished transactions. If we still exceed the cap
    // but all transactions are unfinished, that's ok.
    if (m_transactions.size() > MAX_STORED_TRANSACTIONS) {
        struct FinishedEntry {
            NodeClock::time_point time;
            CTransactionRef tx;
        };
        std::vector<FinishedEntry> finished;
        finished.reserve(m_transactions.size());
        for (const auto& [txref, state] : m_transactions) {
            if (const auto final_state{state.final_state}) {
                finished.push_back(FinishedEntry{.time = state.final_state->time, .tx = txref});
            }
        }
        std::ranges::sort(finished, [](const FinishedEntry& a, const FinishedEntry& b) {
            return a.time < b.time;
        });

        for (const auto& entry : finished) {
            m_transactions.erase(entry.tx);
            if (m_transactions.size() <= MAX_STORED_TRANSACTIONS) break;
        }
    }
    return inserted;
}

std::optional<size_t> PrivateBroadcast::StopBroadcasting(const CTransactionRef& tx, util::Expected<CService, std::string> final_state)

Review comment (suggestion):
I think the current code is fine, but just mentioning: trimming only during "add" could leave us with more than 1000 finalized transactions. For example, if 1100 are added, all of them pending, then no new ones are added (no calls to `Add()`, so the trimming code is never executed anymore) and one by one those 1100 are finalized. So, in theory, to enforce this, the trimming code should be run also when a state changes from non-finalized to finalized. Maybe an overkill, feel free to ignore.

Code after:

#include <private_broadcast.h>
#include <util/check.h>

#include <algorithm>

/// If a transaction is not received back from the network for this duration
/// after it is broadcast, then we consider it stale / for rebroadcasting.
static constexpr auto STALE_DURATION{1min};

bool PrivateBroadcast::Add(const CTransactionRef& tx)
    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
    LOCK(m_mutex);
    const bool inserted{m_transactions.try_emplace(tx).second};
    return inserted;
}

std::optional<size_t> PrivateBroadcast::Remove(const CTransactionRef& tx)
    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
    LOCK(m_mutex);
    const auto handle{m_transactions.extract(tx)};
    if (handle) {
        const auto p{DerivePriority(handle.mapped())};
        return p.num_confirmed;
    }
    return std::nullopt;
}

std::optional<CTransactionRef> PrivateBroadcast::PickTxForSend(const NodeId& will_send_to_nodeid, const CService& will_send_to_address)
    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)
{
    LOCK(m_mutex);

    const auto it{std::ranges::max_element(
            m_transactions,
            [](const auto& a, const auto& b) { return a < b; },
            [](const auto& el) { return DerivePriority(el.second); })};

    if (it != m_transactions.end()) {
        auto& [tx, sent_to]{*it};
        sent_to.emplace_back(will_send_to_nodeid, will_send_to_address, NodeClock::now());
        return tx;
    }

    return std::nullopt;
}

std::optional<CTransactionRef> PrivateBroadcast::GetTxForNode(const NodeId& nodeid)
    EXCLUSIVE_LOCKS_REQUIRED(!m_mutex)