Skip to content

Commit

Permalink
Add support for tx-relay via wtxid
Browse files Browse the repository at this point in the history 
This adds a field to CNodeState that tracks whether to relay transactions with
that peer via wtxid, instead of txid. As of this commit the field will always
be false, but in a later commit we will add a way to negotiate turning this on
via p2p messages exchanged with the peer.
  • Loading branch information
@sdaftuar
sdaftuar committed Jul 19, 2020
1 parent 8e68fc2 commit ac88e2e
Show file tree
Hide file tree
Showing 6 changed files with 106 additions and 44 deletions.
138 changes: 98 additions & 40 deletions src/net_processing.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -410,6 +410,9 @@ struct CNodeState {
//! A rolling bloom filter of all announced tx CInvs to this peer.
CRollingBloomFilter m_recently_announced_invs = CRollingBloomFilter{INVENTORY_MAX_RECENT_RELAY, 0.000001};

//! Whether this peer relays txs via wtxid
bool m_wtxid_relay{false};

CNodeState(CAddress addrIn, std::string addrNameIn, bool is_inbound, bool is_manual) :
address(addrIn), name(std::move(addrNameIn)), m_is_inbound(is_inbound),
m_is_manual_connection (is_manual)
Expand Down Expand Up @@ -836,7 +839,8 @@ void PeerLogicValidation::ReattemptInitialBroadcast(CScheduler& scheduler) const
for (const auto& elem : unbroadcast_txids) {
// Sanity check: all unbroadcast txns should exist in the mempool
if (m_mempool.exists(elem.first)) {
RelayTransaction(elem.first, *connman);
LOCK(cs_main);
RelayTransaction(elem.first, elem.second, *connman);
} else {
m_mempool.RemoveUnbroadcastTx(elem.first, true);
}
Expand Down Expand Up @@ -1405,6 +1409,7 @@ bool static AlreadyHave(const CInv& inv, const CTxMemPool& mempool) EXCLUSIVE_LO
{
case MSG_TX:
case MSG_WITNESS_TX:
case MSG_WTX:
{
assert(recentRejects);
if (::ChainActive().Tip()->GetBlockHash() != hashRecentRejectsChainTip)
Expand All @@ -1419,16 +1424,20 @@ bool static AlreadyHave(const CInv& inv, const CTxMemPool& mempool) EXCLUSIVE_LO

{
LOCK(g_cs_orphans);
if (mapOrphanTransactions.count(inv.hash)) return true;
if (inv.type != MSG_WTX && mapOrphanTransactions.count(inv.hash)) {
return true;
} else if (inv.type == MSG_WTX && g_orphans_by_wtxid.count(inv.hash)) {
return true;
}
}

{
LOCK(g_cs_recent_confirmed_transactions);
if (g_recent_confirmed_transactions->contains(inv.hash)) return true;
}

return recentRejects->contains(inv.hash) ||
mempool.exists(inv.hash);
const bool by_wtxid = (inv.type == MSG_WTX);
return recentRejects->contains(inv.hash) || mempool.exists(inv.hash, by_wtxid);
}
case MSG_BLOCK:
case MSG_WITNESS_BLOCK:
Expand All @@ -1438,11 +1447,17 @@ bool static AlreadyHave(const CInv& inv, const CTxMemPool& mempool) EXCLUSIVE_LO
return true;
}

void RelayTransaction(const uint256& txid, const CConnman& connman)
void RelayTransaction(const uint256& txid, const uint256& wtxid, const CConnman& connman)
{
connman.ForEachNode([&txid](CNode* pnode)
connman.ForEachNode([&txid, &wtxid](CNode* pnode)
{
pnode->PushTxInventory(txid);
AssertLockHeld(cs_main);
CNodeState &state = *State(pnode->GetId());
if (state.m_wtxid_relay) {
pnode->PushTxInventory(wtxid);
} else {
pnode->PushTxInventory(txid);
}
});
}

Expand Down Expand Up @@ -1640,9 +1655,9 @@ void static ProcessGetBlockData(CNode& pfrom, const CChainParams& chainparams, c
}

//! Determine whether or not a peer can request a transaction, and return it (or nullptr if not found or not allowed).
CTransactionRef static FindTxForGetData(const CNode& peer, const uint256& txid, const std::chrono::seconds mempool_req, const std::chrono::seconds now) LOCKS_EXCLUDED(cs_main)
CTransactionRef static FindTxForGetData(const CNode& peer, const uint256& txid_or_wtxid, bool use_wtxid, const std::chrono::seconds mempool_req, const std::chrono::seconds now) LOCKS_EXCLUDED(cs_main)
{
auto txinfo = mempool.info(txid);
auto txinfo = mempool.info(txid_or_wtxid, use_wtxid);
if (txinfo.tx) {
// If a TX could have been INVed in reply to a MEMPOOL request,
// or is older than UNCONDITIONAL_RELAY_DELAY, permit the request
Expand All @@ -1654,13 +1669,12 @@ CTransactionRef static FindTxForGetData(const CNode& peer, const uint256& txid,

{
LOCK(cs_main);

// Otherwise, the transaction must have been announced recently.
if (State(peer.GetId())->m_recently_announced_invs.contains(txid)) {
if (State(peer.GetId())->m_recently_announced_invs.contains(txid_or_wtxid)) {
// If it was, it can be relayed from either the mempool...
if (txinfo.tx) return std::move(txinfo.tx);
// ... or the relay pool.
auto mi = mapRelay.find(txid);
auto mi = mapRelay.find(txid_or_wtxid);
if (mi != mapRelay.end()) return mi->second;
}
}
Expand All @@ -1684,7 +1698,7 @@ void static ProcessGetData(CNode& pfrom, const CChainParams& chainparams, CConnm
// Process as many TX items from the front of the getdata queue as
// possible, since they're common and it's efficient to batch process
// them.
while (it != pfrom.vRecvGetData.end() && (it->type == MSG_TX || it->type == MSG_WITNESS_TX)) {
while (it != pfrom.vRecvGetData.end() && (it->type == MSG_TX || it->type == MSG_WITNESS_TX || it->type == MSG_WTX)) {
if (interruptMsgProc) return;
// The send buffer provides backpressure. If there's no space in
// the buffer, pause processing until the next call.
Expand All @@ -1697,11 +1711,12 @@ void static ProcessGetData(CNode& pfrom, const CChainParams& chainparams, CConnm
continue;
}

CTransactionRef tx = FindTxForGetData(pfrom, inv.hash, mempool_req, now);
CTransactionRef tx = FindTxForGetData(pfrom, inv.hash, inv.type == MSG_WTX, mempool_req, now);
if (tx) {
// WTX and WITNESS_TX imply we serialize with witness
int nSendFlags = (inv.type == MSG_TX ? SERIALIZE_TRANSACTION_NO_WITNESS : 0);
connman.PushMessage(&pfrom, msgMaker.Make(nSendFlags, NetMsgType::TX, *tx));
mempool.RemoveUnbroadcastTx(inv.hash);
mempool.RemoveUnbroadcastTx(tx->GetHash());
// As we're going to send tx, make sure its unconfirmed parents are made requestable.
for (const auto& txin : tx->vin) {
auto txinfo = mempool.info(txin.prevout.hash);
Expand Down Expand Up @@ -1980,7 +1995,7 @@ void static ProcessOrphanTx(CConnman& connman, CTxMemPool& mempool, std::set<uin
if (setMisbehaving.count(fromPeer)) continue;
if (AcceptToMemoryPool(mempool, orphan_state, porphanTx, &removed_txn, false /* bypass_limits */, 0 /* nAbsurdFee */)) {
LogPrint(BCLog::MEMPOOL, " accepted orphan tx %s\n", orphanHash.ToString());
RelayTransaction(orphanHash, connman);
RelayTransaction(orphanHash, porphanTx->GetWitnessHash(), connman);
for (unsigned int i = 0; i < orphanTx.vout.size(); i++) {
auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(orphanHash, i));
if (it_by_prev != mapOrphanTransactionsByPrev.end()) {
Expand Down Expand Up @@ -2841,23 +2856,47 @@ void ProcessMessage(
const CTransaction& tx = *ptx;

const uint256& txid = ptx->GetHash();
pfrom.AddInventoryKnown(txid);
const uint256& wtxid = ptx->GetWitnessHash();

LOCK2(cs_main, g_cs_orphans);

CNodeState* nodestate = State(pfrom.GetId());

const uint256& hash = nodestate->m_wtxid_relay ? wtxid : txid;
pfrom.AddInventoryKnown(hash);
if (nodestate->m_wtxid_relay && txid != wtxid) {
// Insert txid into filterInventoryKnown, even for
// wtxidrelay peers. This prevents re-adding of
// unconfirmed parents to the recently_announced
// filter, when a child tx is requested. See
// ProcessGetData().
pfrom.AddInventoryKnown(txid);
}

TxValidationState state;

CNodeState* nodestate = State(pfrom.GetId());
nodestate->m_tx_download.m_tx_announced.erase(txid);
nodestate->m_tx_download.m_tx_in_flight.erase(txid);
EraseTxRequest(txid);
nodestate->m_tx_download.m_tx_announced.erase(hash);
nodestate->m_tx_download.m_tx_in_flight.erase(hash);
EraseTxRequest(hash);

std::list<CTransactionRef> lRemovedTxn;

if (!AlreadyHave(CInv(MSG_TX, txid), mempool) &&
// We do the AlreadyHave() check using wtxid, rather than txid - in the
// absence of witness malleation, this is strictly better, because the
// recent rejects filter may contain the wtxid but will never contain
// the txid of a segwit transaction that has been rejected.
// In the presence of witness malleation, it's possible that by only
// doing the check with wtxid, we could overlook a transaction which
// was confirmed with a different witness, or exists in our mempool
// with a different witness, but this has limited downside:
// mempool validation does its own lookup of whether we have the txid
// already; and an adversary can already relay us old transactions
// (older than our recency filter) if trying to DoS us, without any need
// for witness malleation.
if (!AlreadyHave(CInv(MSG_WTX, wtxid), mempool) &&
AcceptToMemoryPool(mempool, state, ptx, &lRemovedTxn, false /* bypass_limits */, 0 /* nAbsurdFee */)) {
mempool.check(&::ChainstateActive().CoinsTip());
RelayTransaction(tx.GetHash(), connman);
RelayTransaction(tx.GetHash(), tx.GetWitnessHash(), connman);
for (unsigned int i = 0; i < tx.vout.size(); i++) {
auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(txid, i));
if (it_by_prev != mapOrphanTransactionsByPrev.end()) {
Expand Down Expand Up @@ -2890,10 +2929,17 @@ void ProcessMessage(
uint32_t nFetchFlags = GetFetchFlags(pfrom);
const auto current_time = GetTime<std::chrono::microseconds>();

for (const CTxIn& txin : tx.vin) {
CInv _inv(MSG_TX | nFetchFlags, txin.prevout.hash);
pfrom.AddInventoryKnown(txin.prevout.hash);
if (!AlreadyHave(_inv, mempool)) RequestTx(State(pfrom.GetId()), _inv.hash, current_time);
if (!State(pfrom.GetId())->m_wtxid_relay) {
for (const CTxIn& txin : tx.vin) {
// Here, we only have the txid (and not wtxid) of the
// inputs, so we only request parents from
// non-wtxid-relay peers.
// Eventually we should replace this with an improved
// protocol for getting all unconfirmed parents.
CInv _inv(MSG_TX | nFetchFlags, txin.prevout.hash);
pfrom.AddInventoryKnown(txin.prevout.hash);
if (!AlreadyHave(_inv, mempool)) RequestTx(State(pfrom.GetId()), _inv.hash, current_time);
}
}
AddOrphanTx(ptx, pfrom.GetId());

Expand Down Expand Up @@ -2933,7 +2979,7 @@ void ProcessMessage(
LogPrintf("Not relaying non-mempool transaction %s from forcerelay peer=%d\n", tx.GetHash().ToString(), pfrom.GetId());
} else {
LogPrintf("Force relaying tx %s from peer=%d\n", tx.GetHash().ToString(), pfrom.GetId());
RelayTransaction(tx.GetHash(), connman);
RelayTransaction(tx.GetHash(), tx.GetWitnessHash(), connman);
}
}
}
Expand Down Expand Up @@ -3573,7 +3619,7 @@ void ProcessMessage(
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 (inv.type == MSG_TX || inv.type == MSG_WITNESS_TX || inv.type == MSG_WTX) {
// 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);
Expand Down Expand Up @@ -3858,17 +3904,19 @@ namespace {
class CompareInvMempoolOrder
{
CTxMemPool *mp;
bool m_wtxid_relay;
public:
explicit CompareInvMempoolOrder(CTxMemPool *_mempool)
explicit CompareInvMempoolOrder(CTxMemPool *_mempool, bool use_wtxid)
{
mp = _mempool;
m_wtxid_relay = use_wtxid;
}

bool operator()(std::set<uint256>::iterator a, std::set<uint256>::iterator b)
{
/* As std::make_heap produces a max-heap, we want the entries with the
* fewest ancestors/highest fee to sort later. */
return mp->CompareDepthAndScore(*b, *a);
return mp->CompareDepthAndScore(*b, *a, m_wtxid_relay);
}
};
}
Expand Down Expand Up @@ -4175,8 +4223,8 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
LOCK(pto->m_tx_relay->cs_filter);

for (const auto& txinfo : vtxinfo) {
const uint256& hash = txinfo.tx->GetHash();
CInv inv(MSG_TX, hash);
const uint256& hash = state.m_wtxid_relay ? txinfo.tx->GetWitnessHash() : txinfo.tx->GetHash();
CInv inv(state.m_wtxid_relay ? MSG_WTX : MSG_TX, hash);
pto->m_tx_relay->setInventoryTxToSend.erase(hash);
// Don't send transactions that peers will not put into their mempool
if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) {
Expand Down Expand Up @@ -4211,7 +4259,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
}
// Topologically and fee-rate sort the inventory we send for privacy and priority reasons.
// A heap is used so that not all items need sorting if only a few are being sent.
CompareInvMempoolOrder compareInvMempoolOrder(&m_mempool);
CompareInvMempoolOrder compareInvMempoolOrder(&m_mempool, state.m_wtxid_relay);
std::make_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder);
// No reason to drain out at many times the network's capacity,
// especially since we have many peers and some will draw much shorter delays.
Expand All @@ -4230,18 +4278,20 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
continue;
}
// Not in the mempool anymore? don't bother sending it.
auto txinfo = m_mempool.info(hash);
auto txinfo = m_mempool.info(hash, state.m_wtxid_relay);
if (!txinfo.tx) {
continue;
}
auto txid = txinfo.tx->GetHash();
auto wtxid = txinfo.tx->GetWitnessHash();
// Peer told you to not send transactions at that feerate? Don't bother sending it.
if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) {
continue;
}
if (pto->m_tx_relay->pfilter && !pto->m_tx_relay->pfilter->IsRelevantAndUpdate(*txinfo.tx)) continue;
// Send
State(pto->GetId())->m_recently_announced_invs.insert(hash);
vInv.push_back(CInv(MSG_TX, hash));
vInv.push_back(CInv(state.m_wtxid_relay ? MSG_WTX : MSG_TX, hash));
nRelayedTransactions++;
{
// Expire old relay messages
Expand All @@ -4251,12 +4301,12 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
vRelayExpiration.pop_front();
}

auto ret = mapRelay.insert(std::make_pair(hash, std::move(txinfo.tx)));
auto ret = mapRelay.emplace(txid, std::move(txinfo.tx));
if (ret.second) {
vRelayExpiration.push_back(std::make_pair(nNow + std::chrono::microseconds{RELAY_TX_CACHE_TIME}.count(), ret.first));
vRelayExpiration.emplace_back(nNow + std::chrono::microseconds{RELAY_TX_CACHE_TIME}.count(), ret.first);
}
// Add wtxid-based lookup into mapRelay as well, so that peers can request by wtxid
auto ret2 = mapRelay.emplace(ret.first->second->GetWitnessHash(), ret.first->second);
auto ret2 = mapRelay.emplace(wtxid, ret.first->second);
if (ret2.second) {
vRelayExpiration.emplace_back(nNow + std::chrono::microseconds{RELAY_TX_CACHE_TIME}.count(), ret2.first);
}
Expand All @@ -4266,6 +4316,14 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
vInv.clear();
}
pto->m_tx_relay->filterInventoryKnown.insert(hash);
if (hash != txid) {
// Insert txid into filterInventoryKnown, even for
// wtxidrelay peers. This prevents re-adding of
// unconfirmed parents to the recently_announced
// filter, when a child tx is requested. See
// ProcessGetData().
pto->m_tx_relay->filterInventoryKnown.insert(txid);
}
}
}
}
Expand Down Expand Up @@ -4390,7 +4448,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto)
// 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);
CInv inv(state.m_wtxid_relay ? MSG_WTX : (MSG_TX | GetFetchFlags(*pto)), txid);
if (!AlreadyHave(inv, m_mempool)) {
// If this transaction was last requested more than 1 minute ago,
// then request.
Expand Down
2 changes: 1 addition & 1 deletion src/net_processing.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,6 @@ struct CNodeStateStats {
bool GetNodeStateStats(NodeId nodeid, CNodeStateStats &stats);

/** Relay transaction to every node */
void RelayTransaction(const uint256&, const CConnman& connman);
void RelayTransaction(const uint256& txid, const uint256& wtxid, const CConnman& connman) EXCLUSIVE_LOCKS_REQUIRED(cs_main);

#endif // BITCOIN_NET_PROCESSING_H
3 changes: 2 additions & 1 deletion src/node/transaction.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,8 @@ TransactionError BroadcastTransaction(NodeContext& node, const CTransactionRef t
// best-effort of initial broadcast
node.mempool->AddUnbroadcastTx(hashTx, tx->GetWitnessHash());

RelayTransaction(hashTx, *node.connman);
LOCK(cs_main);
RelayTransaction(hashTx, tx->GetWitnessHash(), *node.connman);
}

return TransactionError::OK;
Expand Down
2 changes: 2 additions & 0 deletions src/protocol.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,8 @@ std::string CInv::GetCommand() const
switch (masked)
{
case MSG_TX: return cmd.append(NetMsgType::TX);
// WTX is not a message type, just an inv type
case MSG_WTX: return cmd.append("wtx");
case MSG_BLOCK: return cmd.append(NetMsgType::BLOCK);
case MSG_FILTERED_BLOCK: return cmd.append(NetMsgType::MERKLEBLOCK);
case MSG_CMPCT_BLOCK: return cmd.append(NetMsgType::CMPCTBLOCK);
Expand Down
3 changes: 2 additions & 1 deletion src/protocol.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -397,10 +397,11 @@ const uint32_t MSG_TYPE_MASK = 0xffffffff >> 2;
* These numbers are defined by the protocol. When adding a new value, be sure
* to mention it in the respective BIP.
*/
enum GetDataMsg {
enum GetDataMsg : uint32_t {
UNDEFINED = 0,
MSG_TX = 1,
MSG_BLOCK = 2,
MSG_WTX = 5, //!< Defined in BIP 339
// The following can only occur in getdata. Invs always use TX or BLOCK.
MSG_FILTERED_BLOCK = 3, //!< Defined in BIP37
MSG_CMPCT_BLOCK = 4, //!< Defined in BIP152
Expand Down
Loading

0 comments on commit ac88e2e

Please sign in to comment.