What peer messages mean

2251    {
2252        auto most_recent_block_txs = std::make_unique<std::map<GenTxid, CTransactionRef>>();
2253        for (const auto& tx : pblock->vtx) {
2254            most_recent_block_txs->emplace(tx->GetHash(), tx);
2255            most_recent_block_txs->emplace(tx->GetWitnessHash(), tx);
2256        }
2257
2258        LOCK(m_most_recent_block_mutex);
2259        m_most_recent_block_hash = hashBlock;
2260        m_most_recent_block = pblock;
2261        m_most_recent_compact_block = pcmpctblock;
2262        m_most_recent_block_txs = std::move(most_recent_block_txs);
2263    }
2264
2265    m_connman.ForEachNode([this, pindex, &lazy_ser, &hashBlock](CNode* pnode) EXCLUSIVE_LOCKS_REQUIRED(::cs_main) {
2266        AssertLockHeld(::cs_main);
2267
2268        if (pnode->GetCommonVersion() < INVALID_CB_NO_BAN_VERSION || pnode->fDisconnect)
2269            return;
2270        ProcessBlockAvailability(pnode->GetId());
2271        CNodeState &state = *State(pnode->GetId());
2272        // If the peer has, or we announced to them the previous block already,
2273        // but we don't think they have this one, go ahead and announce it
2274        if (state.m_requested_hb_cmpctblocks && !PeerHasHeader(&state, pindex) && PeerHasHeader(&state, pindex->pprev)) {
2275
2276            LogDebug(BCLog::NET, "%s sending header-and-ids %s to peer=%d\n", "PeerManager::NewPoWValidBlock",
2277                    hashBlock.ToString(), pnode->GetId());
2278
2279            const CSerializedNetMsg& ser_cmpctblock{lazy_ser.get()};
2280            PushMessage(*pnode, ser_cmpctblock.Copy());
2281            state.pindexBestHeaderSent = pindex;
2282        }
2283    });
2284}
2285
2286/**
2287 * Update our best height and announce any block hashes which weren't previously
2288 * in m_chainman.ActiveChain() to our peers.
2289 */
2290void PeerManagerImpl::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload)
2291{
2292    SetBestBlock(pindexNew->nHeight, std::chrono::seconds{pindexNew->GetBlockTime()});
2293
2294    // Don't relay inventory during initial block download.
2295    if (fInitialDownload) return;
2296
2297    // Find the hashes of all blocks that weren't previously in the best chain.
2298    std::vector<uint256> vHashes;
2299    const CBlockIndex *pindexToAnnounce = pindexNew;
2300    while (pindexToAnnounce != pindexFork) {
2301        vHashes.push_back(pindexToAnnounce->GetBlockHash());
2302        pindexToAnnounce = pindexToAnnounce->pprev;
2303        if (vHashes.size() == MAX_BLOCKS_TO_ANNOUNCE) {
2304            // Limit announcements in case of a huge reorganization.
2305            // Rely on the peer's synchronization mechanism in that case.
2306            break;
2307        }
2308    }
2309
2310    {
2311        LOCK(m_peer_mutex);
2312        for (auto& it : m_peer_map) {
2313            Peer& peer = *it.second;
2314            LOCK(peer.m_block_inv_mutex);
2315            for (const uint256& hash : vHashes | std::views::reverse) {
2316                peer.m_blocks_for_headers_relay.push_back(hash);
2317            }
2318        }
2319    }
2320
2321    m_connman.WakeMessageHandler();
2322}
2323
2324/**
2325 * Handle invalid block rejection and consequent peer discouragement, maintain which
2326 * peers announce compact blocks.
2327 */
2328void PeerManagerImpl::BlockChecked(const std::shared_ptr<const CBlock>& block, const BlockValidationState& state)
2329{
2330    LOCK(cs_main);
2331
2332    const uint256 hash(block->GetHash());
2333    std::map<uint256, std::pair<NodeId, bool>>::iterator it = mapBlockSource.find(hash);
2334
2335    // If the block failed validation, we know where it came from and we're still connected
2336    // to that peer, maybe punish.
2337    if (state.IsInvalid() &&
2338        it != mapBlockSource.end() &&
2339        State(it->second.first)) {
2340            MaybePunishNodeForBlock(/*nodeid=*/ it->second.first, state, /*via_compact_block=*/ !it->second.second);
2341    }
2342    // Check that:
2343    // 1. The block is valid
2344    // 2. We're not in initial block download
2345    // 3. This is currently the best block we're aware of. We haven't updated
2346    //    the tip yet so we have no way to check this directly here. Instead we
2347    //    just check that there are currently no other blocks in flight.
2348    else if (state.IsValid() &&
2349             !m_chainman.IsInitialBlockDownload() &&
2350             mapBlocksInFlight.count(hash) == mapBlocksInFlight.size()) {
2351        if (it != mapBlockSource.end()) {
2352            MaybeSetPeerAsAnnouncingHeaderAndIDs(it->second.first);
2353        }
2354    }
2355    if (it != mapBlockSource.end())
2356        mapBlockSource.erase(it);
2357}
2358
2359//////////////////////////////////////////////////////////////////////////////
2360//
2361// Messages
2362//
2363
2364bool PeerManagerImpl::AlreadyHaveBlock(const uint256& block_hash)
2365{
2366    return m_chainman.m_blockman.LookupBlockIndex(block_hash) != nullptr;
2367}
2368
2369void PeerManagerImpl::SendPings()
2370{
2371    LOCK(m_peer_mutex);
2372    for(auto& it : m_peer_map) it.second->m_ping_queued = true;
2373}
2374
2375std::vector<Wtxid> InvToSendBucket::TakeForProcessing(CTxMemPool& mempool)
2376{
2377    AssertLockHeld(mempool.cs);
2378
2379    size_t n_to_take = static_cast<size_t>(std::max<double>(count_bucket.value() - count_floor, 0));
2380
2381    std::vector<Wtxid> best;
2382
2383    auto itervec = mempool.ExtractBestByMiningScoreWithTopology(backlog, n_to_take);
2384    bool tokens_left = true;
2385    for (auto txiter : itervec) {
2386        auto& wtxid = txiter->GetTx().GetWitnessHash();
2387        if (tokens_left) {
2388            best.push_back(wtxid);
2389            if (!decrement(txiter->GetTx().ComputeTotalSize())) {
2390                tokens_left = false;
2391            }
2392        } else {
2393            backlog.push_back(wtxid);
2394        }
2395    }
2396
2397    // if the backlog is now empty, consider shrinking it if it's oversized
2398    if (backlog.empty() && backlog.capacity() > INVENTORY_BUCKET_BACKLOG_CAPACITY) {
2399        std::vector<Wtxid> dummy;
2400        dummy.reserve(INVENTORY_BUCKET_BACKLOG_CAPACITY);
2401        dummy.swap(backlog);
2402    }
2403
2404    return best;
2405}
2406
2407void PeerManagerImpl::ProcessInvBacklog(NodeClock::time_point now, bool backlog_bumped)
2408{
2409    // Don't run the body of this function unless it's been a little
2410    // while since the last run, or we just added a new tx to the backlog.
2411    if (!backlog_bumped && now <= m_next_inv_bucket_check.load()) return;
2412    m_next_inv_bucket_check = now + INVENTORY_BUCKET_CHECK_DELAY;
2413
2414    LOCK(m_inv_to_send_mutex);
2415    m_inbound_inv_bucket.increment(now);
2416    m_outbound_inv_bucket.increment(now);
2417
2418    // Regular heartbeat logging when there's a backlog
2419    if (!m_next_inv_bucket_heartbeat.has_value()) {
2420        if (m_inbound_inv_bucket.backlog.size() >= INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN || m_outbound_inv_bucket.backlog.size() >= INVENTORY_BUCKET_BACKLOG_HEARTBEAT_MIN) {
2421            m_next_inv_bucket_heartbeat = now;
2422        }
2423    }
2424    if (m_next_inv_bucket_heartbeat.has_value() && now >= *m_next_inv_bucket_heartbeat) {
2425        LogDebug(BCLog::NET, "Transaction rate-limiting backlog inbound=%d itok=%.1f isz=%.1f outbound=%d otok=%.1f osz=%.1f",
2426                 m_inbound_inv_bucket.backlog.size(),
2427                 m_inbound_inv_bucket.count_bucket.value(),
2428                 m_inbound_inv_bucket.size_bucket.value(),
2429                 m_outbound_inv_bucket.backlog.size(),
2430                 m_outbound_inv_bucket.count_bucket.value(),
2431                 m_outbound_inv_bucket.size_bucket.value());
2432        if (m_inbound_inv_bucket.backlog.empty() && m_outbound_inv_bucket.backlog.empty()) {
2433            m_next_inv_bucket_heartbeat = std::nullopt;
2434        } else {
2435            m_next_inv_bucket_heartbeat = now + INVENTORY_BUCKET_BACKLOG_HEARTBEAT;
2436        }
2437    }
2438
2439    // Early exit to skip pointlessly touching mempool lock
2440    bool in_avail = m_inbound_inv_bucket.avail();
2441    bool out_avail = m_outbound_inv_bucket.avail();
2442    if (!in_avail && !out_avail) return;
2443
2444    std::vector<Wtxid> for_inbound;
2445    std::vector<Wtxid> for_outbound;
2446
2447    {
2448        LOCK(m_mempool.cs);
2449        if (in_avail) for_inbound = m_inbound_inv_bucket.TakeForProcessing(m_mempool);
2450        if (out_avail) for_outbound = m_outbound_inv_bucket.TakeForProcessing(m_mempool);
2451    }
2452
2453    if (!for_inbound.empty() || !for_outbound.empty()) {
2454        bool any_inbound_connected = false;
2455        bool any_outbound_connected = false;
2456        for (const PeerRef& peer_ref : GetAllPeers()) {
2457            if (!peer_ref) continue;
2458            Peer& peer{*peer_ref};
2459            auto tx_relay = peer.GetTxRelay();
2460            if (!tx_relay) continue;
2461
2462            LOCK(tx_relay->m_tx_inventory_mutex);
2463            // Only queue transactions for announcement once the version handshake
2464            // is completed. The time of arrival for these transactions is
2465            // otherwise at risk of leaking to a spy, if the spy is able to
2466            // distinguish transactions received during the handshake from the rest
2467            // in the announcement.
2468            if (tx_relay->m_next_inv_send_time == 0s) continue;
2469            if (peer.m_is_inbound) {
2470                any_inbound_connected = true;
2471            } else {
2472                any_outbound_connected = true;
2473            }
2474            for (auto& i : (peer.m_is_inbound ? for_inbound : for_outbound)) {
2475                tx_relay->m_tx_inventory_to_send.push_back(i);
2476            }
2477        }
2478
2479        // if the node has no in/outbound connections, clear the corresponding backlog entirely
2480        // this reduces wasted memory, and avoids having the bucket artificially empty for when
2481        // future peers do connect.
2482        if (!any_inbound_connected) m_inbound_inv_bucket.backlog.clear();
2483        if (!any_outbound_connected) m_outbound_inv_bucket.backlog.clear();
2484    }
2485}
2486
2487void PeerManagerImpl::InitiateTxBroadcastToAll(const Wtxid& wtxid)
2488{
2489    {
2490        LOCK(m_inv_to_send_mutex);
2491        m_inbound_inv_bucket.backlog.push_back(wtxid);
2492        m_outbound_inv_bucket.backlog.push_back(wtxid);
2493    }
2494    ProcessInvBacklog(NodeClock::now(), /*backlog_bumped=*/true);
2495}
2496
2497node::TransactionError PeerManagerImpl::InitiateTxBroadcastPrivate(const CTransactionRef& tx)
2498{
2499    const auto txstr{strprintf("txid=%s, wtxid=%s", tx->GetHash().ToString(), tx->GetWitnessHash().ToString())};
2500    switch (m_tx_for_private_broadcast.Add(tx)) {