diff options
Diffstat (limited to 'src/net_processing.cpp')
| -rw-r--r-- | src/net_processing.cpp | 269 |
1 files changed, 169 insertions, 100 deletions
diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 71ebd72b8..4b43b2cdf 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -25,9 +25,7 @@ #include <scheduler.h> #include <tinyformat.h> #include <txmempool.h> -#include <ui_interface.h> #include <util/system.h> -#include <util/moneystr.h> #include <util/strencodings.h> #include <util/validation.h> @@ -70,11 +68,13 @@ static constexpr int32_t MAX_PEER_TX_IN_FLIGHT = 100; /** Maximum number of announced transactions from a peer */ static constexpr int32_t MAX_PEER_TX_ANNOUNCEMENTS = 2 * MAX_INV_SZ; /** How many microseconds to delay requesting transactions from inbound peers */ -static constexpr int64_t INBOUND_PEER_TX_DELAY = 2 * 1000000; +static constexpr int64_t INBOUND_PEER_TX_DELAY = 2 * 1000000; // 2 seconds /** How long to wait (in microseconds) before downloading a transaction from an additional peer */ -static constexpr int64_t GETDATA_TX_INTERVAL = 60 * 1000000; +static constexpr int64_t GETDATA_TX_INTERVAL = 60 * 1000000; // 1 minute /** Maximum delay (in microseconds) for transaction requests to avoid biasing some peers over others. */ -static constexpr int64_t MAX_GETDATA_RANDOM_DELAY = 2 * 1000000; +static constexpr int64_t MAX_GETDATA_RANDOM_DELAY = 2 * 1000000; // 2 seconds +/** How long to wait (in microseconds) before expiring an in-flight getdata request to a peer */ +static constexpr int64_t TX_EXPIRY_INTERVAL = 10 * GETDATA_TX_INTERVAL; static_assert(INBOUND_PEER_TX_DELAY >= MAX_GETDATA_RANDOM_DELAY, "To preserve security, MAX_GETDATA_RANDOM_DELAY should not exceed INBOUND_PEER_DELAY"); /** Limit to avoid sending big packets. Not used in processing incoming GETDATA for compatibility */ @@ -345,8 +345,11 @@ struct CNodeState { //! Store all the transactions a peer has recently announced std::set<uint256> m_tx_announced; - //! Store transactions which were requested by us - std::set<uint256> m_tx_in_flight; + //! Store transactions which were requested by us, with timestamp + std::map<uint256, int64_t> m_tx_in_flight; + + //! Periodically check for stuck getdata requests + int64_t m_check_expiry_timer{0}; }; TxDownloadState m_tx_download; @@ -422,7 +425,7 @@ static void PushNodeVersion(CNode *pnode, CConnman* connman, int64_t nTime) CAddress addrMe = CAddress(CService(), nLocalNodeServices); connman->PushMessage(pnode, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERSION, PROTOCOL_VERSION, (uint64_t)nLocalNodeServices, nTime, addrYou, addrMe, - nonce, strSubVersion, nNodeStartingHeight, ::fRelayTxes)); + nonce, strSubVersion, nNodeStartingHeight, ::g_relay_txes)); if (fLogIPs) { LogPrint(BCLog::NET, "send version message: version %d, blocks=%d, us=%s, them=%s, peer=%d\n", PROTOCOL_VERSION, nNodeStartingHeight, addrMe.ToString(), addrYou.ToString(), nodeid); @@ -579,7 +582,7 @@ static bool TipMayBeStale(const Consensus::Params &consensusParams) EXCLUSIVE_LO static bool CanDirectFetch(const Consensus::Params &consensusParams) EXCLUSIVE_LOCKS_REQUIRED(cs_main) { - return chainActive.Tip()->GetBlockTime() > GetAdjustedTime() - consensusParams.nPowTargetSpacing * 20; + return ::ChainActive().Tip()->GetBlockTime() > GetAdjustedTime() - consensusParams.nPowTargetSpacing * 20; } static bool PeerHasHeader(CNodeState *state, const CBlockIndex *pindex) EXCLUSIVE_LOCKS_REQUIRED(cs_main) @@ -605,7 +608,7 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec // Make sure pindexBestKnownBlock is up to date, we'll need it. ProcessBlockAvailability(nodeid); - if (state->pindexBestKnownBlock == nullptr || state->pindexBestKnownBlock->nChainWork < chainActive.Tip()->nChainWork || state->pindexBestKnownBlock->nChainWork < nMinimumChainWork) { + if (state->pindexBestKnownBlock == nullptr || state->pindexBestKnownBlock->nChainWork < ::ChainActive().Tip()->nChainWork || state->pindexBestKnownBlock->nChainWork < nMinimumChainWork) { // This peer has nothing interesting. return; } @@ -613,7 +616,7 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec if (state->pindexLastCommonBlock == nullptr) { // Bootstrap quickly by guessing a parent of our best tip is the forking point. // Guessing wrong in either direction is not a problem. - state->pindexLastCommonBlock = chainActive[std::min(state->pindexBestKnownBlock->nHeight, chainActive.Height())]; + state->pindexLastCommonBlock = ::ChainActive()[std::min(state->pindexBestKnownBlock->nHeight, ::ChainActive().Height())]; } // If the peer reorganized, our previous pindexLastCommonBlock may not be an ancestor @@ -655,7 +658,7 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec // We wouldn't download this block or its descendants from this peer. return; } - if (pindex->nStatus & BLOCK_HAVE_DATA || chainActive.Contains(pindex)) { + if (pindex->nStatus & BLOCK_HAVE_DATA || ::ChainActive().Contains(pindex)) { if (pindex->HaveTxsDownloaded()) state->pindexLastCommonBlock = pindex; } else if (mapBlocksInFlight.count(pindex->GetBlockHash()) == 0) { @@ -704,30 +707,40 @@ void UpdateTxRequestTime(const uint256& txid, int64_t request_time) EXCLUSIVE_LO } } - -void RequestTx(CNodeState* state, const uint256& txid, int64_t nNow) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +int64_t CalculateTxGetDataTime(const uint256& txid, int64_t current_time, bool use_inbound_delay) EXCLUSIVE_LOCKS_REQUIRED(cs_main) { - CNodeState::TxDownloadState& peer_download_state = state->m_tx_download; - if (peer_download_state.m_tx_announced.size() >= MAX_PEER_TX_ANNOUNCEMENTS || peer_download_state.m_tx_announced.count(txid)) { - // Too many queued announcements from this peer, or we already have - // this announcement - return; - } - peer_download_state.m_tx_announced.insert(txid); - int64_t process_time; int64_t last_request_time = GetTxRequestTime(txid); // First time requesting this tx if (last_request_time == 0) { - process_time = nNow; + process_time = current_time; } else { // Randomize the delay to avoid biasing some peers over others (such as due to // fixed ordering of peer processing in ThreadMessageHandler) process_time = last_request_time + GETDATA_TX_INTERVAL + GetRand(MAX_GETDATA_RANDOM_DELAY); } - // We delay processing announcements from non-preferred (eg inbound) peers - if (!state->fPreferredDownload) process_time += INBOUND_PEER_TX_DELAY; + // We delay processing announcements from inbound peers + if (use_inbound_delay) process_time += INBOUND_PEER_TX_DELAY; + + return process_time; +} + +void RequestTx(CNodeState* state, const uint256& txid, int64_t nNow) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +{ + CNodeState::TxDownloadState& peer_download_state = state->m_tx_download; + if (peer_download_state.m_tx_announced.size() >= MAX_PEER_TX_ANNOUNCEMENTS || + peer_download_state.m_tx_process_time.size() >= MAX_PEER_TX_ANNOUNCEMENTS || + peer_download_state.m_tx_announced.count(txid)) { + // Too many queued announcements from this peer, or we already have + // this announcement + return; + } + peer_download_state.m_tx_announced.insert(txid); + + // Calculate the time to try requesting this transaction. Use + // fPreferredDownload as a proxy for outbound peers. + int64_t process_time = CalculateTxGetDataTime(txid, nNow, !state->fPreferredDownload); peer_download_state.m_tx_process_time.emplace(process_time, txid); } @@ -1071,7 +1084,7 @@ static bool MaybePunishNode(NodeId nodeid, const CValidationState& state, bool v static bool BlockRequestAllowed(const CBlockIndex* pindex, const Consensus::Params& consensusParams) EXCLUSIVE_LOCKS_REQUIRED(cs_main) { AssertLockHeld(cs_main); - if (chainActive.Contains(pindex)) return true; + if (::ChainActive().Contains(pindex)) return true; return pindex->IsValid(BLOCK_VALID_SCRIPTS) && (pindexBestHeader != nullptr) && (pindexBestHeader->GetBlockTime() - pindex->GetBlockTime() < STALE_RELAY_AGE_LIMIT) && (GetBlockProofEquivalentTime(*pindexBestHeader, *pindex, *pindexBestHeader, consensusParams) < STALE_RELAY_AGE_LIMIT); @@ -1183,7 +1196,7 @@ void PeerLogicValidation::NewPoWValidBlock(const CBlockIndex *pindex, const std: /** * Update our best height and announce any block hashes which weren't previously - * in chainActive to our peers. + * in ::ChainActive() to our peers. */ void PeerLogicValidation::UpdatedBlockTip(const CBlockIndex *pindexNew, const CBlockIndex *pindexFork, bool fInitialDownload) { const int nNewHeight = pindexNew->nHeight; @@ -1240,7 +1253,7 @@ void PeerLogicValidation::BlockChecked(const CBlock& block, const CValidationSta // the tip yet so we have no way to check this directly here. Instead we // just check that there are currently no other blocks in flight. else if (state.IsValid() && - !IsInitialBlockDownload() && + !::ChainstateActive().IsInitialBlockDownload() && mapBlocksInFlight.count(hash) == mapBlocksInFlight.size()) { if (it != mapBlockSource.end()) { MaybeSetPeerAsAnnouncingHeaderAndIDs(it->second.first, connman); @@ -1264,13 +1277,13 @@ bool static AlreadyHave(const CInv& inv) EXCLUSIVE_LOCKS_REQUIRED(cs_main) case MSG_WITNESS_TX: { assert(recentRejects); - if (chainActive.Tip()->GetBlockHash() != hashRecentRejectsChainTip) + if (::ChainActive().Tip()->GetBlockHash() != hashRecentRejectsChainTip) { // If the chain tip has changed previously rejected transactions // might be now valid, e.g. due to a nLockTime'd tx becoming valid, // or a double-spend. Reset the rejects filter and give those // txs a second chance. - hashRecentRejectsChainTip = chainActive.Tip()->GetBlockHash(); + hashRecentRejectsChainTip = ::ChainActive().Tip()->GetBlockHash(); recentRejects->reset(); } @@ -1395,7 +1408,7 @@ void static ProcessGetBlockData(CNode* pfrom, const CChainParams& chainparams, c } // Avoid leaking prune-height by never sending blocks below the NODE_NETWORK_LIMITED threshold if (send && !pfrom->fWhitelisted && ( - (((pfrom->GetLocalServices() & NODE_NETWORK_LIMITED) == NODE_NETWORK_LIMITED) && ((pfrom->GetLocalServices() & NODE_NETWORK) != NODE_NETWORK) && (chainActive.Tip()->nHeight - pindex->nHeight > (int)NODE_NETWORK_LIMITED_MIN_BLOCKS + 2 /* add two blocks buffer extension for possible races */) ) + (((pfrom->GetLocalServices() & NODE_NETWORK_LIMITED) == NODE_NETWORK_LIMITED) && ((pfrom->GetLocalServices() & NODE_NETWORK) != NODE_NETWORK) && (::ChainActive().Tip()->nHeight - pindex->nHeight > (int)NODE_NETWORK_LIMITED_MIN_BLOCKS + 2 /* add two blocks buffer extension for possible races */) ) )) { LogPrint(BCLog::NET, "Ignore block request below NODE_NETWORK_LIMITED threshold from peer=%d\n", pfrom->GetId()); @@ -1465,7 +1478,7 @@ void static ProcessGetBlockData(CNode* pfrom, const CChainParams& chainparams, c // instead we respond with the full, non-compact block. bool fPeerWantsWitness = State(pfrom->GetId())->fWantsCmpctWitness; int nSendFlags = fPeerWantsWitness ? 0 : SERIALIZE_TRANSACTION_NO_WITNESS; - if (CanDirectFetch(consensusParams) && pindex->nHeight >= chainActive.Height() - MAX_CMPCTBLOCK_DEPTH) { + if (CanDirectFetch(consensusParams) && pindex->nHeight >= ::ChainActive().Height() - MAX_CMPCTBLOCK_DEPTH) { if ((fPeerWantsWitness || !fWitnessesPresentInARecentCompactBlock) && a_recent_compact_block && a_recent_compact_block->header.GetHash() == pindex->GetBlockHash()) { connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::CMPCTBLOCK, *a_recent_compact_block)); } else { @@ -1485,7 +1498,7 @@ void static ProcessGetBlockData(CNode* pfrom, const CChainParams& chainparams, c // and we want it right after the last block so they don't // wait for other stuff first. std::vector<CInv> vInv; - vInv.push_back(CInv(MSG_BLOCK, chainActive.Tip()->GetBlockHash())); + vInv.push_back(CInv(MSG_BLOCK, ::ChainActive().Tip()->GetBlockHash())); connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::INV, vInv)); pfrom->hashContinue.SetNull(); } @@ -1542,16 +1555,28 @@ void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnm } } + // Unknown types in the GetData stay in vRecvGetData and block any future + // message from this peer, see vRecvGetData check in ProcessMessages(). + // Depending on future p2p changes, we might either drop unknown getdata on + // the floor or disconnect the peer. + pfrom->vRecvGetData.erase(pfrom->vRecvGetData.begin(), it); if (!vNotFound.empty()) { // Let the peer know that we didn't find what it asked for, so it doesn't - // have to wait around forever. Currently only SPV clients actually care - // about this message: it's needed when they are recursively walking the - // dependencies of relevant unconfirmed transactions. SPV clients want to - // do that because they want to know about (and store and rebroadcast and - // risk analyze) the dependencies of transactions relevant to them, without - // having to download the entire memory pool. + // have to wait around forever. + // SPV clients care about this message: it's needed when they are + // recursively walking the dependencies of relevant unconfirmed + // transactions. SPV clients want to do that because they want to know + // about (and store and rebroadcast and risk analyze) the dependencies + // of transactions relevant to them, without having to download the + // entire memory pool. + // Also, other nodes can use these messages to automatically request a + // transaction from some other peer that annnounced it, and stop + // waiting for us to respond. + // In normal operation, we often send NOTFOUND messages for parents of + // transactions that we relay; if a peer is missing a parent, they may + // assume we have them and request the parents from us. connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::NOTFOUND, vNotFound)); } } @@ -1606,7 +1631,7 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve // nUnconnectingHeaders gets reset back to 0. if (!LookupBlockIndex(headers[0].hashPrevBlock) && nCount < MAX_BLOCKS_TO_ANNOUNCE) { nodestate->nUnconnectingHeaders++; - connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexBestHeader), uint256())); + connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), uint256())); LogPrint(BCLog::NET, "received header %s: missing prev block %s, sending getheaders (%d) to end (peer=%d, nUnconnectingHeaders=%d)\n", headers[0].GetHash().ToString(), headers[0].hashPrevBlock.ToString(), @@ -1663,26 +1688,26 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve // because it is set in UpdateBlockAvailability. Some nullptr checks // are still present, however, as belt-and-suspenders. - if (received_new_header && pindexLast->nChainWork > chainActive.Tip()->nChainWork) { + if (received_new_header && pindexLast->nChainWork > ::ChainActive().Tip()->nChainWork) { nodestate->m_last_block_announcement = GetTime(); } if (nCount == MAX_HEADERS_RESULTS) { // Headers message had its maximum size; the peer may have more headers. - // TODO: optimize: if pindexLast is an ancestor of chainActive.Tip or pindexBestHeader, continue + // TODO: optimize: if pindexLast is an ancestor of ::ChainActive().Tip or pindexBestHeader, continue // from there instead. LogPrint(BCLog::NET, "more getheaders (%d) to end to peer=%d (startheight:%d)\n", pindexLast->nHeight, pfrom->GetId(), pfrom->nStartingHeight); - connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexLast), uint256())); + connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexLast), uint256())); } bool fCanDirectFetch = CanDirectFetch(chainparams.GetConsensus()); // If this set of headers is valid and ends in a block with at least as // much work as our tip, download as much as possible. - if (fCanDirectFetch && pindexLast->IsValid(BLOCK_VALID_TREE) && chainActive.Tip()->nChainWork <= pindexLast->nChainWork) { + if (fCanDirectFetch && pindexLast->IsValid(BLOCK_VALID_TREE) && ::ChainActive().Tip()->nChainWork <= pindexLast->nChainWork) { std::vector<const CBlockIndex*> vToFetch; const CBlockIndex *pindexWalk = pindexLast; // Calculate all the blocks we'd need to switch to pindexLast, up to a limit. - while (pindexWalk && !chainActive.Contains(pindexWalk) && vToFetch.size() <= MAX_BLOCKS_IN_TRANSIT_PER_PEER) { + while (pindexWalk && !::ChainActive().Contains(pindexWalk) && vToFetch.size() <= MAX_BLOCKS_IN_TRANSIT_PER_PEER) { if (!(pindexWalk->nStatus & BLOCK_HAVE_DATA) && !mapBlocksInFlight.count(pindexWalk->GetBlockHash()) && (!IsWitnessEnabled(pindexWalk->pprev, chainparams.GetConsensus()) || State(pfrom->GetId())->fHaveWitness)) { @@ -1695,7 +1720,7 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve // very large reorg at a time we think we're close to caught up to // the main chain -- this shouldn't really happen. Bail out on the // direct fetch and rely on parallel download instead. - if (!chainActive.Contains(pindexWalk)) { + if (!::ChainActive().Contains(pindexWalk)) { LogPrint(BCLog::NET, "Large reorg, won't direct fetch to %s (%d)\n", pindexLast->GetBlockHash().ToString(), pindexLast->nHeight); @@ -1728,7 +1753,7 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve } // If we're in IBD, we want outbound peers that will serve us a useful // chain. Disconnect peers that are on chains with insufficient work. - if (IsInitialBlockDownload() && nCount != MAX_HEADERS_RESULTS) { + if (::ChainstateActive().IsInitialBlockDownload() && nCount != MAX_HEADERS_RESULTS) { // When nCount < MAX_HEADERS_RESULTS, we know we have no more // headers to fetch from this peer. if (nodestate->pindexBestKnownBlock && nodestate->pindexBestKnownBlock->nChainWork < nMinimumChainWork) { @@ -1736,7 +1761,7 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve // us sync -- disconnect if using an outbound slot (unless // whitelisted or addnode). // Note: We compare their tip to nMinimumChainWork (rather than - // chainActive.Tip()) because we won't start block download + // ::ChainActive().Tip()) because we won't start block download // until we have a headers chain that has at least // nMinimumChainWork, even if a peer has a chain past our tip, // as an anti-DoS measure. @@ -1750,7 +1775,7 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve if (!pfrom->fDisconnect && IsOutboundDisconnectionCandidate(pfrom) && nodestate->pindexBestKnownBlock != nullptr) { // If this is an outbound peer, check to see if we should protect // it from the bad/lagging chain logic. - if (g_outbound_peers_with_protect_from_disconnect < MAX_OUTBOUND_PEERS_TO_PROTECT_FROM_DISCONNECT && nodestate->pindexBestKnownBlock->nChainWork >= chainActive.Tip()->nChainWork && !nodestate->m_chain_sync.m_protect) { + if (g_outbound_peers_with_protect_from_disconnect < MAX_OUTBOUND_PEERS_TO_PROTECT_FROM_DISCONNECT && nodestate->pindexBestKnownBlock->nChainWork >= ::ChainActive().Tip()->nChainWork && !nodestate->m_chain_sync.m_protect) { LogPrint(BCLog::NET, "Protecting outbound peer=%d from eviction\n", pfrom->GetId()); nodestate->m_chain_sync.m_protect = true; ++g_outbound_peers_with_protect_from_disconnect; @@ -1994,7 +2019,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr if (!pfrom->fInbound) { // Advertise our address - if (fListen && !IsInitialBlockDownload()) + if (fListen && !::ChainstateActive().IsInitialBlockDownload()) { CAddress addr = GetLocalAddress(&pfrom->addr, pfrom->GetLocalServices()); FastRandomContext insecure_rand; @@ -2189,7 +2214,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr return false; } - bool fBlocksOnly = !fRelayTxes; + bool fBlocksOnly = !g_relay_txes; // Allow whitelisted peers to send data other than blocks in blocks only mode if whitelistrelay is true if (pfrom->fWhitelisted && gArgs.GetBoolArg("-whitelistrelay", DEFAULT_WHITELISTRELAY)) @@ -2220,7 +2245,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr // fell back to inv we probably have a reorg which we should get the headers for first, // we now only provide a getheaders response here. When we receive the headers, we will // then ask for the blocks we need. - connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexBestHeader), inv.hash)); + connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), inv.hash)); LogPrint(BCLog::NET, "getheaders (%d) %s to peer=%d\n", pindexBestHeader->nHeight, inv.hash.ToString(), pfrom->GetId()); } } @@ -2229,7 +2254,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr pfrom->AddInventoryKnown(inv); if (fBlocksOnly) { LogPrint(BCLog::NET, "transaction (%s) inv sent in violation of protocol peer=%d\n", inv.hash.ToString(), pfrom->GetId()); - } else if (!fAlreadyHave && !fImporting && !fReindex && !IsInitialBlockDownload()) { + } else if (!fAlreadyHave && !fImporting && !fReindex && !::ChainstateActive().IsInitialBlockDownload()) { RequestTx(State(pfrom->GetId()), inv.hash, nNow); } } @@ -2291,14 +2316,14 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr LOCK(cs_main); // Find the last block the caller has in the main chain - const CBlockIndex* pindex = FindForkInGlobalIndex(chainActive, locator); + const CBlockIndex* pindex = FindForkInGlobalIndex(::ChainActive(), locator); // Send the rest of the chain if (pindex) - pindex = chainActive.Next(pindex); + pindex = ::ChainActive().Next(pindex); int nLimit = 500; LogPrint(BCLog::NET, "getblocks %d to %s limit %d from peer=%d\n", (pindex ? pindex->nHeight : -1), hashStop.IsNull() ? "end" : hashStop.ToString(), nLimit, pfrom->GetId()); - for (; pindex; pindex = chainActive.Next(pindex)) + for (; pindex; pindex = ::ChainActive().Next(pindex)) { if (pindex->GetBlockHash() == hashStop) { @@ -2308,7 +2333,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr // If pruning, don't inv blocks unless we have on disk and are likely to still have // for some reasonable time window (1 hour) that block relay might require. const int nPrunedBlocksLikelyToHave = MIN_BLOCKS_TO_KEEP - 3600 / chainparams.GetConsensus().nPowTargetSpacing; - if (fPruneMode && (!(pindex->nStatus & BLOCK_HAVE_DATA) || pindex->nHeight <= chainActive.Tip()->nHeight - nPrunedBlocksLikelyToHave)) + if (fPruneMode && (!(pindex->nStatus & BLOCK_HAVE_DATA) || pindex->nHeight <= ::ChainActive().Tip()->nHeight - nPrunedBlocksLikelyToHave)) { LogPrint(BCLog::NET, " getblocks stopping, pruned or too old block at %d %s\n", pindex->nHeight, pindex->GetBlockHash().ToString()); break; @@ -2350,7 +2375,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr return true; } - if (pindex->nHeight < chainActive.Height() - MAX_BLOCKTXN_DEPTH) { + if (pindex->nHeight < ::ChainActive().Height() - MAX_BLOCKTXN_DEPTH) { // If an older block is requested (should never happen in practice, // but can happen in tests) send a block response instead of a // blocktxn response. Sending a full block response instead of a @@ -2387,7 +2412,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr } LOCK(cs_main); - if (IsInitialBlockDownload() && !pfrom->fWhitelisted) { + if (::ChainstateActive().IsInitialBlockDownload() && !pfrom->fWhitelisted) { LogPrint(BCLog::NET, "Ignoring getheaders from peer=%d because node is in initial block download\n", pfrom->GetId()); return true; } @@ -2410,23 +2435,23 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr else { // Find the last block the caller has in the main chain - pindex = FindForkInGlobalIndex(chainActive, locator); + pindex = FindForkInGlobalIndex(::ChainActive(), locator); if (pindex) - pindex = chainActive.Next(pindex); + pindex = ::ChainActive().Next(pindex); } // we must use CBlocks, as CBlockHeaders won't include the 0x00 nTx count at the end std::vector<CBlock> vHeaders; int nLimit = MAX_HEADERS_RESULTS; LogPrint(BCLog::NET, "getheaders %d to %s from peer=%d\n", (pindex ? pindex->nHeight : -1), hashStop.IsNull() ? "end" : hashStop.ToString(), pfrom->GetId()); - for (; pindex; pindex = chainActive.Next(pindex)) + for (; pindex; pindex = ::ChainActive().Next(pindex)) { vHeaders.push_back(pindex->GetBlockHeader()); if (--nLimit <= 0 || pindex->GetBlockHash() == hashStop) break; } - // pindex can be nullptr either if we sent chainActive.Tip() OR - // if our peer has chainActive.Tip() (and thus we are sending an empty + // pindex can be nullptr either if we sent ::ChainActive().Tip() OR + // if our peer has ::ChainActive().Tip() (and thus we are sending an empty // headers message). In both cases it's safe to update // pindexBestHeaderSent to be our tip. // @@ -2437,7 +2462,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr // without the new block. By resetting the BestHeaderSent, we ensure we // will re-announce the new block via headers (or compact blocks again) // in the SendMessages logic. - nodestate->pindexBestHeaderSent = pindex ? pindex : chainActive.Tip(); + nodestate->pindexBestHeaderSent = pindex ? pindex : ::ChainActive().Tip(); connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::HEADERS, vHeaders)); return true; } @@ -2445,7 +2470,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr if (strCommand == NetMsgType::TX) { // Stop processing the transaction early if // We are in blocks only mode and peer is either not whitelisted or whitelistrelay is off - if (!fRelayTxes && (!pfrom->fWhitelisted || !gArgs.GetBoolArg("-whitelistrelay", DEFAULT_WHITELISTRELAY))) + if (!g_relay_txes && (!pfrom->fWhitelisted || !gArgs.GetBoolArg("-whitelistrelay", DEFAULT_WHITELISTRELAY))) { LogPrint(BCLog::NET, "transaction sent in violation of protocol peer=%d\n", pfrom->GetId()); return true; @@ -2609,8 +2634,8 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr if (!LookupBlockIndex(cmpctblock.header.hashPrevBlock)) { // Doesn't connect (or is genesis), instead of DoSing in AcceptBlockHeader, request deeper headers - if (!IsInitialBlockDownload()) - connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexBestHeader), uint256())); + if (!::ChainstateActive().IsInitialBlockDownload()) + connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexBestHeader), uint256())); return true; } @@ -2654,7 +2679,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr // If this was a new header with more work than our tip, update the // peer's last block announcement time - if (received_new_header && pindex->nChainWork > chainActive.Tip()->nChainWork) { + if (received_new_header && pindex->nChainWork > ::ChainActive().Tip()->nChainWork) { nodestate->m_last_block_announcement = GetTime(); } @@ -2664,7 +2689,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr if (pindex->nStatus & BLOCK_HAVE_DATA) // Nothing to do here return true; - if (pindex->nChainWork <= chainActive.Tip()->nChainWork || // We know something better + if (pindex->nChainWork <= ::ChainActive().Tip()->nChainWork || // We know something better pindex->nTx != 0) { // We had this block at some point, but pruned it if (fAlreadyInFlight) { // We requested this block for some reason, but our mempool will probably be useless @@ -2688,7 +2713,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr // We want to be a bit conservative just to be extra careful about DoS // possibilities in compact block processing... - if (pindex->nHeight <= chainActive.Height() + 2) { + if (pindex->nHeight <= ::ChainActive().Height() + 2) { if ((!fAlreadyInFlight && nodestate->nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) || (fAlreadyInFlight && blockInFlightIt->second.first == pfrom->GetId())) { std::list<QueuedBlock>::iterator* queuedBlockIt = nullptr; @@ -3148,8 +3173,27 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr } if (strCommand == NetMsgType::NOTFOUND) { - // We do not care about the NOTFOUND message, but logging an Unknown Command - // message would be undesirable as we transmit it ourselves. + // Remove the NOTFOUND transactions from the peer + LOCK(cs_main); + CNodeState *state = State(pfrom->GetId()); + std::vector<CInv> vInv; + vRecv >> vInv; + if (vInv.size() <= MAX_PEER_TX_IN_FLIGHT + MAX_BLOCKS_IN_TRANSIT_PER_PEER) { + for (CInv &inv : vInv) { + if (inv.type == MSG_TX || inv.type == MSG_WITNESS_TX) { + // If we receive a NOTFOUND message for a txid we requested, erase + // it from our data structures for this peer. + auto in_flight_it = state->m_tx_download.m_tx_in_flight.find(inv.hash); + if (in_flight_it == state->m_tx_download.m_tx_in_flight.end()) { + // Skip any further work if this is a spurious NOTFOUND + // message. + continue; + } + state->m_tx_download.m_tx_in_flight.erase(in_flight_it); + state->m_tx_download.m_tx_announced.erase(inv.hash); + } + } + } return true; } @@ -3221,6 +3265,7 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter return false; // this maintains the order of responses + // and prevents vRecvGetData to grow unbounded if (!pfrom->vRecvGetData.empty()) return true; if (!pfrom->orphan_work_set.empty()) return true; @@ -3288,23 +3333,22 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter if (m_enable_bip61) { connman->PushMessage(pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::REJECT, strCommand, REJECT_MALFORMED, std::string("error parsing message"))); } - if (strstr(e.what(), "end of data")) - { + if (strstr(e.what(), "end of data")) { // Allow exceptions from under-length message on vRecv LogPrint(BCLog::NET, "%s(%s, %u bytes): Exception '%s' caught, normally caused by a message being shorter than its stated length\n", __func__, SanitizeString(strCommand), nMessageSize, e.what()); - } - else if (strstr(e.what(), "size too large")) - { + } else if (strstr(e.what(), "size too large")) { // Allow exceptions from over-long size LogPrint(BCLog::NET, "%s(%s, %u bytes): Exception '%s' caught\n", __func__, SanitizeString(strCommand), nMessageSize, e.what()); - } - else if (strstr(e.what(), "non-canonical ReadCompactSize()")) - { + } else if (strstr(e.what(), "non-canonical ReadCompactSize()")) { // Allow exceptions from non-canonical encoding LogPrint(BCLog::NET, "%s(%s, %u bytes): Exception '%s' caught\n", __func__, SanitizeString(strCommand), nMessageSize, e.what()); - } - else - { + } else if (strstr(e.what(), "Superfluous witness record")) { + // Allow exceptions from illegal witness encoding + LogPrint(BCLog::NET, "%s(%s, %u bytes): Exception '%s' caught\n", __func__, SanitizeString(strCommand), nMessageSize, e.what()); + } else if (strstr(e.what(), "Unknown transaction optional data")) { + // Allow exceptions from unknown witness encoding + LogPrint(BCLog::NET, "%s(%s, %u bytes): Exception '%s' caught\n", __func__, SanitizeString(strCommand), nMessageSize, e.what()); + } else { PrintExceptionContinue(&e, "ProcessMessages()"); } } @@ -3338,7 +3382,7 @@ void PeerLogicValidation::ConsiderEviction(CNode *pto, int64_t time_in_seconds) // their chain has more work than ours, we should sync to it, // unless it's invalid, in which case we should find that out and // disconnect from them elsewhere). - if (state.pindexBestKnownBlock != nullptr && state.pindexBestKnownBlock->nChainWork >= chainActive.Tip()->nChainWork) { + if (state.pindexBestKnownBlock != nullptr && state.pindexBestKnownBlock->nChainWork >= ::ChainActive().Tip()->nChainWork) { if (state.m_chain_sync.m_timeout != 0) { state.m_chain_sync.m_timeout = 0; state.m_chain_sync.m_work_header = nullptr; @@ -3350,7 +3394,7 @@ void PeerLogicValidation::ConsiderEviction(CNode *pto, int64_t time_in_seconds) // where we checked against our tip. // Either way, set a new timeout based on current tip. state.m_chain_sync.m_timeout = time_in_seconds + CHAIN_SYNC_TIMEOUT; - state.m_chain_sync.m_work_header = chainActive.Tip(); + state.m_chain_sync.m_work_header = ::ChainActive().Tip(); state.m_chain_sync.m_sent_getheaders = false; } else if (state.m_chain_sync.m_timeout > 0 && time_in_seconds > state.m_chain_sync.m_timeout) { // No evidence yet that our peer has synced to a chain with work equal to that @@ -3363,7 +3407,7 @@ void PeerLogicValidation::ConsiderEviction(CNode *pto, int64_t time_in_seconds) } else { assert(state.m_chain_sync.m_work_header); LogPrint(BCLog::NET, "sending getheaders to outbound peer=%d to verify chain work (current best known block:%s, benchmark blockhash: %s)\n", pto->GetId(), state.pindexBestKnownBlock != nullptr ? state.pindexBestKnownBlock->GetBlockHash().ToString() : "<none>", state.m_chain_sync.m_work_header->GetBlockHash().ToString()); - connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(state.m_chain_sync.m_work_header->pprev), uint256())); + connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(state.m_chain_sync.m_work_header->pprev), uint256())); state.m_chain_sync.m_sent_getheaders = true; constexpr int64_t HEADERS_RESPONSE_TIME = 120; // 2 minutes // Bump the timeout to allow a response, which could clear the timeout @@ -3525,7 +3569,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto) // Address refresh broadcast int64_t nNow = GetTimeMicros(); - if (!IsInitialBlockDownload() && pto->nNextLocalAddrSend < nNow) { + if (!::ChainstateActive().IsInitialBlockDownload() && pto->nNextLocalAddrSend < nNow) { AdvertiseLocal(pto); pto->nNextLocalAddrSend = PoissonNextSend(nNow, AVG_LOCAL_ADDRESS_BROADCAST_INTERVAL); } @@ -3561,7 +3605,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto) // Start block sync if (pindexBestHeader == nullptr) - pindexBestHeader = chainActive.Tip(); + pindexBestHeader = ::ChainActive().Tip(); bool fFetch = state.fPreferredDownload || (nPreferredDownload == 0 && !pto->fClient && !pto->fOneShot); // Download if this is a nice peer, or we have no nice peers and this one might do. if (!state.fSyncStarted && !pto->fClient && !fImporting && !fReindex) { // Only actively request headers from a single peer, unless we're close to today. @@ -3580,7 +3624,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto) if (pindexStart->pprev) pindexStart = pindexStart->pprev; LogPrint(BCLog::NET, "initial getheaders (%d) to peer=%d (startheight:%d)\n", pindexStart->nHeight, pto->GetId(), pto->nStartingHeight); - connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexStart), uint256())); + connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETHEADERS, ::ChainActive().GetLocator(pindexStart), uint256())); } } @@ -3607,11 +3651,11 @@ bool PeerLogicValidation::SendMessages(CNode* pto) bool fFoundStartingHeader = false; // Try to find first header that our peer doesn't have, and // then send all headers past that one. If we come across any - // headers that aren't on chainActive, give up. + // headers that aren't on ::ChainActive(), give up. for (const uint256 &hash : pto->vBlockHashesToAnnounce) { const CBlockIndex* pindex = LookupBlockIndex(hash); assert(pindex); - if (chainActive[pindex->nHeight] != pindex) { + if (::ChainActive()[pindex->nHeight] != pindex) { // Bail out if we reorged away from this block fRevertToInv = true; break; @@ -3707,9 +3751,9 @@ bool PeerLogicValidation::SendMessages(CNode* pto) // Warn if we're announcing a block that is not on the main chain. // This should be very rare and could be optimized out. // Just log for now. - if (chainActive[pindex->nHeight] != pindex) { + if (::ChainActive()[pindex->nHeight] != pindex) { LogPrint(BCLog::NET, "Announcing block %s not on main chain (tip=%s)\n", - hashToAnnounce.ToString(), chainActive.Tip()->GetBlockHash().ToString()); + hashToAnnounce.ToString(), ::ChainActive().Tip()->GetBlockHash().ToString()); } // If the peer's chain has this block, don't inv it back. @@ -3926,7 +3970,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto) // Message: getdata (blocks) // std::vector<CInv> vGetData; - if (!pto->fClient && ((fFetch && !pto->m_limited_node) || !IsInitialBlockDownload()) && state.nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) { + if (!pto->fClient && ((fFetch && !pto->m_limited_node) || !::ChainstateActive().IsInitialBlockDownload()) && state.nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) { std::vector<const CBlockIndex*> vToDownload; NodeId staller = -1; FindNextBlocksToDownload(pto->GetId(), MAX_BLOCKS_IN_TRANSIT_PER_PEER - state.nBlocksInFlight, vToDownload, staller, consensusParams); @@ -3948,9 +3992,33 @@ bool PeerLogicValidation::SendMessages(CNode* pto) // // Message: getdata (non-blocks) // + + // For robustness, expire old requests after a long timeout, so that + // we can resume downloading transactions from a peer even if they + // were unresponsive in the past. + // Eventually we should consider disconnecting peers, but this is + // conservative. + if (state.m_tx_download.m_check_expiry_timer <= nNow) { + for (auto it=state.m_tx_download.m_tx_in_flight.begin(); it != state.m_tx_download.m_tx_in_flight.end();) { + if (it->second <= nNow - TX_EXPIRY_INTERVAL) { + LogPrint(BCLog::NET, "timeout of inflight tx %s from peer=%d\n", it->first.ToString(), pto->GetId()); + state.m_tx_download.m_tx_announced.erase(it->first); + state.m_tx_download.m_tx_in_flight.erase(it++); + } else { + ++it; + } + } + // On average, we do this check every TX_EXPIRY_INTERVAL. Randomize + // so that we're not doing this for all peers at the same time. + state.m_tx_download.m_check_expiry_timer = nNow + TX_EXPIRY_INTERVAL/2 + GetRand(TX_EXPIRY_INTERVAL); + } + auto& tx_process_time = state.m_tx_download.m_tx_process_time; while (!tx_process_time.empty() && tx_process_time.begin()->first <= nNow && state.m_tx_download.m_tx_in_flight.size() < MAX_PEER_TX_IN_FLIGHT) { - const uint256& txid = tx_process_time.begin()->second; + const uint256 txid = tx_process_time.begin()->second; + // Erase this entry from tx_process_time (it may be added back for + // processing at a later time, see below) + tx_process_time.erase(tx_process_time.begin()); CInv inv(MSG_TX | GetFetchFlags(pto), txid); if (!AlreadyHave(inv)) { // If this transaction was last requested more than 1 minute ago, @@ -3964,20 +4032,20 @@ bool PeerLogicValidation::SendMessages(CNode* pto) vGetData.clear(); } UpdateTxRequestTime(inv.hash, nNow); - state.m_tx_download.m_tx_in_flight.insert(inv.hash); + state.m_tx_download.m_tx_in_flight.emplace(inv.hash, nNow); } else { // This transaction is in flight from someone else; queue // up processing to happen after the download times out // (with a slight delay for inbound peers, to prefer // requests to outbound peers). - RequestTx(&state, txid, nNow); + int64_t next_process_time = CalculateTxGetDataTime(txid, nNow, !state.fPreferredDownload); + tx_process_time.emplace(next_process_time, txid); } } else { // We have already seen this transaction, no need to download. state.m_tx_download.m_tx_announced.erase(inv.hash); state.m_tx_download.m_tx_in_flight.erase(inv.hash); } - tx_process_time.erase(tx_process_time.begin()); } @@ -4024,4 +4092,5 @@ public: mapOrphanTransactions.clear(); mapOrphanTransactionsByPrev.clear(); } -} instance_of_cnetprocessingcleanup; +}; +static CNetProcessingCleanup instance_of_cnetprocessingcleanup; |