See #28765
First, this PR enables keeping track of per-peer reconciliation sets, containing those transactions which we intend to exchange efficiently. The remaining transactions are announced via flooding, as usual.
Second, this PR enables periodically initiating a reconciliation round via a new p2p message.
Erlay Project Tracking: #28646
~Marking this draft until~
~Another task is to sync parent PR with this version. I have a local branch that compiles, but tests of the full Erlay there needs a little care. If you want to see how this PR works in the broader context, the parent should be good enough to get it. Otherwise, I intend to update it soon.~
The following sections might be updated with supplementary metadata relevant to reviewers and maintainers.
For detailed information about the code coverage, see the test coverage report.
See the guideline for information on the review process. A summary of reviews will appear here.
Reviewers, this pull request conflicts with the following ones:
If you consider this pull request important, please also help to review the conflicting pull requests. Ideally, start with the one that should be merged first.
5741+ // 1. Flood parent+child.
5742+ // 2. Reconcile parent+child.
5743+ // 3. Flood parent, reconcile child.
5744+ // We choose (2) because it has the easiest implementation.
5745+ // The latency impact is not that bad:
5746+ // 1. If the parent is in the reocnciliation set, the two
5778+ // We don't care about a laggy peer (1) because we probably can't help them even if we flood transactions.
5779+ // However, exploiting (2) should not prevent us from relaying certain transactions.
5780+ //
5781+ // Transactions which don't make it to the set due to the limit are announced via fan-out.
5782+ const size_t recon_set_size = m_txreconciliation->GetPeerSetSize(pto->GetId());
5783+ if (!flood_target && txs_to_reconcile.size() + recon_set_size < MAX_PEER_TX_ANNOUNCEMENTS) {
<= as MAX_PEER_TX_ANNOUNCEMENTS should be the maximum number of elements/transactions we announce to our peers. Note, while only strictly inferior transaction requests are considered L1409 in net_processing.cpp, this is different for the processing of NOTFOUND, L4793, still in net_processing.cpp.
<=, but I don’t get the second sentence :(
120+ bool ShouldFloodTo(uint256 wtxid, NodeId peer_id) const;
121+
122+ /**
123+ * Check whether a particular transaction is being currently reconciled with a given peer.
124+ */
125+ bool CurrentlyReconcilingTx(NodeId peer_id, const uint256 wtxid) const;
IsAlreadyInPeerSet or something more speaking than transaction is already in peer local set but not announced yet.
5834
5835+ //
5836+ // Message: reconciliation request
5837+ //
5838+ {
5839+ if (!m_chainman.ActiveChainstate().IsInitialBlockDownload()) {
86+ * the peer just announced the transaction to us.
87+ */
88+ void TryRemovingFromReconSet(NodeId peer_id, const uint256 wtxid_to_remove);
89+
90+ /**
91+ * Step 2. If a it's time to request a reconciliation from the peer, this function will return
5836+ // Message: reconciliation request
5837+ //
5838+ {
5839+ if (!m_chainman.ActiveChainstate().IsInitialBlockDownload()) {
5840+ if (m_txreconciliation && m_txreconciliation->IsPeerRegistered(pto->GetId())) {
5841+ auto reconciliation_request_data = m_txreconciliation->MaybeRequestReconciliation(pto->GetId());
reconciliation_request_data could be called reconciliation_request_parameters to dissociate clearly we’re sending reconciliation config info and not the set itself.
162 * Limits the impact of low-fee transaction floods. */
163 static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND = 7;
164 /** Maximum number of inventory items to send per transmission. */
165-static constexpr unsigned int INVENTORY_BROADCAST_MAX = INVENTORY_BROADCAST_PER_SECOND * count_seconds(INBOUND_INVENTORY_BROADCAST_INTERVAL);
166+static constexpr unsigned int INVENTORY_BROADCAST_MAX = INVENTORY_BROADCAST_PER_SECOND *
167+ count_seconds(std::max(INBOUND_INVENTORY_BROADCAST_INTERVAL, INBOUND_INVENTORY_BROADCAST_INTERVAL_RECON));
std::max between INBOUND_INVENTORY_BROADCAST_INTERVAL and INBOUND_INVENTORY_BROADCAST_INTERVAL_RECON as the former is always superior to the latter. Maybe could be rather a static_assert to enforce the order at compilation time in case of future changes.
5595 fSendTrickle = true;
5596 if (pto->IsInboundConn()) {
5597- tx_relay->m_next_inv_send_time = NextInvToInbounds(current_time, INBOUND_INVENTORY_BROADCAST_INTERVAL);
5598+ if (supports_recon) {
5599+ // Use shorter intervals for reconciliation peers because we use
5600+ // low-fanout, and 1) we need to make faster; 2) we won't get much
20+
21+/** Announce transactions via full wtxid to a limited number of inbound and outbound peers. */
22+constexpr double INBOUND_FANOUT_DESTINATIONS_FRACTION = 0.1;
23+constexpr double OUTBOUND_FANOUT_DESTINATIONS_FRACTION = 0.1;
24+/** Coefficient used to estimate reconciliation set differences. */
25+constexpr double RECON_Q = 0.25;
31+/**
32+ * Interval between initiating reconciliations with peers.
33+ * This value allows to reconcile ~(7 tx/s * 8s) transactions during normal operation.
34+ * More frequent reconciliations would cause significant constant bandwidth overhead
35+ * due to reconciliation metadata (sketch sizes etc.), which would nullify the efficiency.
36+ * Less frequent reconciliations would introduce high transaction relay latency.
131+ */
132+ std::chrono::microseconds m_next_recon_request GUARDED_BY(m_txreconciliation_mutex){0};
133+ void UpdateNextReconRequest(std::chrono::microseconds now) EXCLUSIVE_LOCKS_REQUIRED(m_txreconciliation_mutex)
134+ {
135+ // We have one timer for the entire queue. This is safe because we initiate reconciliations
136+ // with outbound connections, which are unlikely to game this timer in a serious way.
Phase::INIT_REQUESTED and which would prevent MaybeRequestReconciliation() to move forward due to the check L258 ?
Phase::INIT_REQUESTED is somehow related to being stuck… if one peer stays in that state, it doesn’t prevent us from requesting recon from other peers. Only will delay by m_next_recon_request once?
5821@@ -5611,11 +5822,31 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
5822 tx_relay->m_tx_inventory_known_filter.insert(txid);
5823 }
5824 }
5825+
5826+ // Populating local reconciliation set.
5827+ if (txs_to_reconcile.size() != 0) {
1d9b1f7d4bbf5ca5a8f43180b994b7b2fc03447f
0 if (!txs_to_reconcile.empty()) {
38+constexpr std::chrono::microseconds RECON_REQUEST_INTERVAL{8s};
39+
40+/**
41+ * Represents phase of the current reconciliation round with a peer.
42+ */
43+enum Phase {
5070c97459282346cdcff7af08914c702462fe0d
0enum class Phase {
134+ {
135+ // We have one timer for the entire queue. This is safe because we initiate reconciliations
136+ // with outbound connections, which are unlikely to game this timer in a serious way.
137+ size_t we_initiate_to_count = std::count_if(m_states.begin(), m_states.end(),
138+ [](std::pair<NodeId, std::variant<uint64_t, TxReconciliationState>> indexed_state) {
139+ auto* cur_state = std::get_if<TxReconciliationState>(&indexed_state.second);
7cffb4cc0211b0494a7563d9a80a67633fa61255
0 const auto* cur_state = std::get_if<TxReconciliationState>(&indexed_state.second);
244+ AssertLockNotHeld(m_txreconciliation_mutex);
245+ if (!IsPeerRegistered(peer_id)) return std::nullopt;
246+ LOCK(m_txreconciliation_mutex);
247+ auto& recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
248+
249+ if (m_queue.size() > 0) {
0 if (!m_queue.empty()) {
300+ auto recon_state = m_states.find(peer_id);
301+ return (recon_state != m_states.end() &&
302+ std::holds_alternative<TxReconciliationState>(recon_state->second));
303+ }
304+
305+ bool ShouldFloodTo(uint256 wtxid, NodeId peer_id) const EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
4cb08ebcd6190535bc2300e35557dd8b730b8951
0 bool ShouldFloodTo(const uint256& wtxid, NodeId peer_id) const EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
315+
316+ const bool we_initiate = recon_state.m_we_initiate;
317+ // Find all peers of the similar reconciliation direction.
318+ std::for_each(m_states.begin(), m_states.end(),
319+ [&eligible_peers, we_initiate](std::pair<NodeId, std::variant<uint64_t, TxReconciliationState>> indexed_state) {
320+ const auto cur_state = std::get<TxReconciliationState>(indexed_state.second);
4cb08ebcd6190535bc2300e35557dd8b730b8951
0 const auto& cur_state = std::get<TxReconciliationState>(indexed_state.second);
314+ std::vector<NodeId> eligible_peers;
315+
316+ const bool we_initiate = recon_state.m_we_initiate;
317+ // Find all peers of the similar reconciliation direction.
318+ std::for_each(m_states.begin(), m_states.end(),
319+ [&eligible_peers, we_initiate](std::pair<NodeId, std::variant<uint64_t, TxReconciliationState>> indexed_state) {
4cb08ebcd6190535bc2300e35557dd8b730b8951
0 [&eligible_peers, we_initiate](auto indexed_state) {
335+
336+ const size_t peer_index = it - eligible_peers.begin();
337+ return txidHasher(wtxid) % flood_index_modulo == peer_index % flood_index_modulo;
338+ }
339+
340+ bool CurrentlyReconcilingTx(NodeId peer_id, const uint256 wtxid) const EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
1d9b1f7d4bbf5ca5a8f43180b994b7b2fc03447f: there are other instances where wtxid is not passed by ref
0 bool CurrentlyReconcilingTx(NodeId peer_id, const uint256& wtxid) const EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
121+ const bool we_initiate = !is_peer_inbound && is_peer_recon_responder;
122+
123+ // If we ever announce support for both requesting and responding, this will need
124+ // tie-breaking. For now, this is mutually exclusive because both are based on the
125+ // inbound flag.
126+ assert(!(they_initiate && we_initiate));
c60235bacceecbe7f83bb70f761a0adb3e5bf65b nit but I think it’s easier to read
0 assert(!they_initiate || !we_initiate));
124+ // tie-breaking. For now, this is mutually exclusive because both are based on the
125+ // inbound flag.
126+ assert(!(they_initiate && we_initiate));
127+
128+ // The peer set both flags to false, we treat it as a protocol violation.
129+ if (!(they_initiate || we_initiate)) return PROTOCOL_VIOLATION;
c60235bacceecbe7f83bb70f761a0adb3e5bf65b nit but I think it’s easier to read
0 if (!they_initiate && !we_initiate) return PROTOCOL_VIOLATION;
353+ // 3. Otherwise, return false.
354+ const size_t starting_point = txidHasher(wtxid) % eligible_peers.size();
355+ if (starting_point <= peer_position && peer_position < starting_point + round_down_flood_targets) {
356+ return true;
357+ } else if (peer_position == starting_point + round_down_flood_targets) {
358+ return rand() < (flood_targets - round_down_flood_targets) * RAND_MAX;
FastRandomContext), not the standard library built-in one which has no guarantees about its quality or performance. If you really can’t use FastRandomContext for some reason, use C++ RNG functions, not C ones (rand() is not thread-safe, in addition to possibly being poor quality).
131@@ -124,6 +132,35 @@ class TxReconciliationTracker::Impl
132 return ReconciliationRegisterResult::SUCCESS;
133 }
134
135+ void AddToSet(NodeId peer_id, const std::vector<uint256>& txs_to_reconcile) EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
136+ {
137+ AssertLockNotHeld(m_txreconciliation_mutex);
138+ Assume(txs_to_reconcile.size() > 0);
139+ assert(IsPeerRegistered(peer_id));
This will lock m_txreconciliation_mutex inside IsPeerRegistered(), check the state, and unlock again. On the next line we will then lock it again, and get a reference to the actual state. I don’t think that’s what we want:
IsPeerRegistered and getting the state two lines below, so the assertion doesn’t actually prevent race conditions.You should check the TxReconciliationState within the same critical section as using it (perhaps by making IsPeerRegistered require m_txreconciliation_mutex being held?).
152+ }
153+
154+ void TryRemovingFromSet(NodeId peer_id, const uint256& wtxid_to_remove) EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
155+ {
156+ AssertLockNotHeld(m_txreconciliation_mutex);
157+ assert(IsPeerRegistered(peer_id));
140+ LOCK(m_txreconciliation_mutex);
141+ auto& recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
142+
143+ size_t added = 0;
144+ for (auto& wtxid: txs_to_reconcile) {
145+ if (recon_state.m_local_set.insert(wtxid).second) {
added += recon_state.m_local_set.insert(wtxid).second;.
139+ assert(IsPeerRegistered(peer_id));
140+ LOCK(m_txreconciliation_mutex);
141+ auto& recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
142+
143+ size_t added = 0;
144+ for (auto& wtxid: txs_to_reconcile) {
:.
202+ bool ShouldFloodTo(NodeId peer_id, const uint256& wtxid) const EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
203+ {
204+ AssertLockNotHeld(m_txreconciliation_mutex);
205+ if (!IsPeerRegistered(peer_id)) return true;
206+ LOCK(m_txreconciliation_mutex);
207+ const auto recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
TxReconciliationState. Sure you don’t want to get a reference here?
85@@ -72,6 +86,12 @@ class TxReconciliationTracker::Impl
86 private:
87 mutable Mutex m_txreconciliation_mutex;
88
89+ /**
90+ * We need a ReconciliationTracker-wide randomness to decide to which peers we should flood a
91+ * given transaction based on a (w)txid.
92+ */
93+ const SaltedTxidHasher txidHasher;
m_txid_hasher.
230+ } else {
231+ flood_targets = eligible_peers.size() * INBOUND_FANOUT_DESTINATIONS_FRACTION;
232+ if (flood_targets == 0) return false;
233+ }
234+
235+ const size_t round_down_flood_targets = floor(flood_targets);
floor() here. Casting a floating-point type to integer type will automatically round down.
263@@ -264,6 +264,11 @@ extern const char* WTXIDRELAY;
264 * txreconciliation, as described by BIP 330.
265 */
266 extern const char* SENDTXRCNCL;
267+/**
268+ * Contains a 4-byte local reconciliation set size and 4-byte q-coefficient.
312+ AssertLockNotHeld(m_txreconciliation_mutex);
313+ LOCK(m_txreconciliation_mutex);
314+ if (!IsPeerRegistered(peer_id)) return true;
315+ const auto& recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
316+
317+ // We assume that reconciliation is always initiated from inbound to outbound to avoid
89@@ -53,6 +90,17 @@ class TxReconciliationState
90 */
91 uint64_t m_k0, m_k1;
92
93+ /**
94+ * Store all wtxids which we would announce to the peer (policy checks passed, etc.)
Have you considered to enforce policy checks like feefilter at REQTXRCNCL sending instead of at reconciliation set fulfilling in INV sending ?
Due to RECON_REQUEST_INTERVAL, I think some transactions could become stalled, wasting bandwidth in case of fast-paced mempool congestion change (not to exclude when we see few transactions issuers on the network provoking spikes by themselves). Though if correct, more likely a marginal performance improvement better left for follow-up.
net_processing.cpp.
Yeah, I see your concern (although I tested Erlay in mainnet and it was ok), let’s see if other reviewers think it’s needed now.
txs_to_reconcile and sending the inv after reconciliation? Given that RECON_REQUEST_INTERVAL is 8sec and AVG_FEEFILTER_BROADCAST_INTERVAL is 10min, I don’t think this is too much of a concern (at least for this PR)?
216+ size_t added = 0;
217+ for (auto& wtxid: txs_to_reconcile) {
218+ added += recon_state.m_local_set.insert(wtxid).second;
219+ }
220+
221+ LogPrint(BCLog::NET, "Added %i new transactions to the reconciliation set for peer=%d. " /* Continued */
BCLog::TXRECONCILIATION
154+ AssertLockNotHeld(m_txreconciliation_mutex);
155+ LOCK(m_txreconciliation_mutex);
156+ assert(IsPeerRegistered(peer_id));
157+ auto& recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
158+
159+ recon_state.m_local_set.erase(wtxid_to_remove);
231+ double flood_targets;
232+ if (we_initiate) {
233+ flood_targets = OUTBOUND_FANOUT_DESTINATIONS;
234+ } else {
235+ flood_targets = eligible_peers.size() * INBOUND_FANOUT_DESTINATIONS_FRACTION;
236+ if (flood_targets == 0) return false;
== to compare floating-point numbers seems not ideal due to precision issues, but it seems like we can never hit the return here anyway since we assert that eligible_peers.size() > 0 above, so maybe the entire line could be deleted or be an assert?
241+ const auto it = std::find(eligible_peers.begin(), eligible_peers.end(), peer_id);
242+ Assume(it != eligible_peers.end());
243+ const size_t peer_position = it - eligible_peers.begin();
244+ // The requirements to this algorithm is the following:
245+ // 1. Every transaction should be assigned to *some* peer, at least assuming a static list
246+ // of peers. For this function that means no randomness.
insecure_rand.randrange, “no randomness” seems a bit misleading.
243+ const size_t peer_position = it - eligible_peers.begin();
244+ // The requirements to this algorithm is the following:
245+ // 1. Every transaction should be assigned to *some* peer, at least assuming a static list
246+ // of peers. For this function that means no randomness.
247+ // 2. The choice doesn't leak the internal order of peers (m_states) to the external
248+ // observer. This is achieved by hashing the txid.
239+ const size_t round_down_flood_targets = flood_targets;
240+
241+ const auto it = std::find(eligible_peers.begin(), eligible_peers.end(), peer_id);
242+ Assume(it != eligible_peers.end());
243+ const size_t peer_position = it - eligible_peers.begin();
244+ // The requirements to this algorithm is the following:
225+
226+ void TryRemovingFromSet(NodeId peer_id, const uint256& wtxid_to_remove) EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
227+ {
228+ AssertLockNotHeld(m_txreconciliation_mutex);
229+ LOCK(m_txreconciliation_mutex);
230+ assert(IsPeerRegistered(peer_id));
I would suggest not to have an assert here. Generally (imo) we should only add asserts when crashing the node is better than continuing. In this case simply adding a if(!IsPeerRegistered()) return; is better than crashing if the user of the API makes a mistake. You would be treating non-existence of the peer the same as the peer having an empty set (i.e. nothing is removed as it didn’t exist in the first place).
Additionally, the way you are using this API right now is not generally safe.
0if (m_txreconciliation && m_txreconciliation->IsPeerRegistered(pfrom.GetId())) {
1 m_txreconciliation->TryRemovingFromSet(pfrom.GetId(), wtxid);
2}
The peer could be de-registered from the tracker in between the IsPeerRegistered and TryRemovingFromSet calls causing the node to crash for no good reason. This is only safe right now because cs_main is held here as well as when a peer is forgotten from the tracker. Since the tracker has its own mutex it should not rely on another mutex such as cs_main.
5722+ State(pto->GetId())->m_recently_announced_invs.insert(txid);
5723+ State(pto->GetId())->m_recently_announced_invs.insert(wtxid);
5724+
5725+ bool adding_to_recon_set = false;
5726+ // Check if peer supports reconciliations.
5727+ if (supports_recon) {
I think it would make sense to move this block to its own function. This entire block for the tx announcement logic in SendMessages is already quite big.
Maybe this can even move to the tracker?
Adding to this:
Afaict it’s impossible for our fuzzers to exercise this code at the moment. Encapsulating this logic within the TxReconciliationTracker would make that possible.
208+ void AddToSet(NodeId peer_id, const std::vector<uint256>& txs_to_reconcile) EXCLUSIVE_LOCKS_REQUIRED(!m_txreconciliation_mutex)
209+ {
210+ AssertLockNotHeld(m_txreconciliation_mutex);
211+ Assume(txs_to_reconcile.size() > 0);
212+ LOCK(m_txreconciliation_mutex);
213+ assert(IsPeerRegistered(peer_id));
238+ const auto it = std::find(eligible_peers.begin(), eligible_peers.end(), peer_id);
239+ Assume(it != eligible_peers.end());
240+ const size_t peer_position = it - eligible_peers.begin();
241+ // The requirements to this algorithm is the following:
242+ // 1. Every transaction should be assigned to *some* peer.
243+ // 2. The choice doesn't leak the internal order of peers (m_states) to the external
One more conceptual question for the flooding algorithm:
Doesn’t this still leak internal order by flooding preferentially to adjacent nodes in eligible_peers if flood_targets > 1?
Would it be a simpler alternative to have a deterministic randomizer that takes the wtxid, and then pick flood_targets random peers (and pick a last one only with an appropiate probability if flood_targets is not an integer), in the same way that way PeerManagerImpl::RelayAddress() picks 2 peers for addr relay? (but without the 24h reset happening there)
Doesn’t this still leak internal order by flooding preferentially to adjacent nodes in eligible_peers if flood_targets > 1?
Good point. Yes. I should think about the fix.
Would it be a simpler alternative to have a deterministic randomizer that takes the wtxid, and then pick flood_targets random peers (and pick a last one only with an appropiate probability if flood_targets is not an integer), in the same way that way PeerManagerImpl::RelayAddress() picks 2 peers for addr relay? (but without the 24h reset happening there)
I’ll try.
5677@@ -5678,7 +5678,11 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
5678 }
5679 if (tx_relay->m_bloom_filter && !tx_relay->m_bloom_filter->IsRelevantAndUpdate(*txinfo.tx)) continue;
5680 // Send
5681- State(pto->GetId())->m_recently_announced_invs.insert(hash);
5682+
5683+ // Make a transaction requestable by both txid and wtxid, to avoid making
5684+ // an assumption that a child arrives after the parent.
5685+ State(pto->GetId())->m_recently_announced_invs.insert(txid);
Trying to understand commit 869c6824235ff2af09d24b13cfab9a6cabd52b00: “It’s possible that a parent will be scheduled to relay same time as child, but a child arrives earlier. " Is this only possible with reconciliation? The previous commit tries to make sure that the child is also scheduled for recon, if the parent was - so can it only happen in the case where the recon set is at max capacity?
Also, since now twice as many entries will be saved in m_recently_announced_invs, does it need a larger capacity?
Is this only possible with reconciliation?
Yes I think so
so can it only happen in the case where the recon set is at max capacity?
Not only that. The order is lost when set difference is decoded. The decoding party only knows wtxids and will request them in random order, via a regular INV. The ordering could be handled on the responder side, but I’m afraid it would be hard and messy. I can try though.
The logic in the last commit is not super-reliable either, so I think it’s better to have belt-and-suspenders like this.
Unless they break something else of course.
Also, since now twice as many entries will be saved in m_recently_announced_invs, does it need a larger capacity?
good point.
161+ * 2a) announcing both ways simultaneously (inefficiency);
162+ * 2b) inference based on the announced transactions (privacy leak).
163+ * That's why it's ok to make this delay low as well, and lower delay is generally good to
164+ * facilitate good transaction relay speed when slow reconciliations prevail. */
165 static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL{5s};
166+static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL_RECON{2s};
245+ if (!m_queue.empty()) {
246+ // Request transaction reconciliation periodically to efficiently exchange transactions.
247+ // To make reconciliation predictable and efficient, we reconcile with peers in order
248+ // based on the queue, taking a delay between requests.
249+ auto current_time = GetTime<std::chrono::seconds>();
250+ if (m_next_recon_request <= current_time && m_queue.front() == peer_id) {
SENDTXRCNCL but stalls the VERACK, they would be registered and added to m_queue - but SendMessages() will never get to the part where REQTXRCNCL is sent for this peer because we abort early here before hanshake completion. Therefore, once that peer is in the front of the queue, it won’t get cleared and prevents us from reconciling with other peers while connected (up to DEFAULT_PEER_CONNECT_TIMEOUT = 60 seconds).
I don’t see that yet. No matter what happens to this peer, it will be moved to the end of the queue, and the next peer will be requested in couple seconds.
I was thinking of the following:
1.) Peer X gets connected, sends SENDTXRCNCL, RegisterPeer puts it into the back of the queue.
2.) X doesn’t send VERACK
3.) We do regular reconciliation requests with all other peers of the queue, so now X is in the front of it
4.) Since X is not succesfully connected, SendMessages aborts early and we don’t reach the code where REQTXRCNCL is processed
5.) X won’t be removed from the front of the queue, and we don’t request reconciliations from any other peer because of that.
I made this code extra safe:
SendMessages;Also reminding that this only applies to outbound.
Having separate queues for every peer would nullify all our previous tuning and performance benchmarks (both mine and from others), that’s why I’d rather stick to the current approach.
27+constexpr double OUTBOUND_FANOUT_DESTINATIONS = 1;
28+/**
29+ * If there's a chance a transaction is not streamlined along the first couple hops, it would take
30+ * very long to relay.
31+ */
32+static_assert(OUTBOUND_FANOUT_DESTINATIONS >= 1);
in ffbb2b814e82284c7c77c5a5b23058cbc7c0149c:
Based on this, is the “The requirements to this algorithm is the following: 1. Every transaction should be assigned to some peer.” a hard requirement? If so, it seems the logic in f0685a1781 net_processing if recon parent -> recon child seems to break this, since you might have chosen just 1 peer to flood the child to, but then switch to recon because you’re reconing the parent with that peer. It doesn’t seem like ShouldFloodTo uses any state about whether we’ve chosen other peers for flooding, so we might end up not flooding the tx to anybody? Am I missing something / is this okay?
Every transaction should be assigned to some peer." a hard requirement?
I meant to relay in some way (either flooded or reconciled), so that the transaction doesn’t get stuck. That’s it. That resolves your question right? Fixing code comment for now.
5846
5847+ //
5848+ // Message: reconciliation request
5849+ //
5850+ {
5851+ if (!m_chainman.ActiveChainstate().IsInitialBlockDownload() && m_txreconciliation) {
NetMsgType::REQTXRCNCL handling is added, we ignore the request. That’s fine, but should we also have the logic for falling back to flooding the wtxids in the reconciliation set when it gets full / after a timeout?
-txreconciliation=1 CLI flag… Why do you think this would be useful for such tester?
75@@ -76,6 +76,35 @@ class TxReconciliationTracker
76 ReconciliationRegisterResult RegisterPeer(NodeId peer_id, bool is_peer_inbound,
77 uint32_t peer_recon_version, uint64_t remote_salt);
78
79+ /**
80+ * Step 1. Add new transactions we want to announce to the peer to the local reconciliation set
81+ * of the peer, so that those transactions will be reconciled later.
82+ * The caller *must* check that the peer is registered for reconciliations.
83+ */
84+ void AddToSet(NodeId peer_id, const std::vector<uint256>& txs_to_reconcile);
86+ /**
87+ * Before Step 2, we might want to remove a wtxid from the reconciliation set, for example if
88+ * the peer just announced the transaction to us.
89+ * The caller *must* check that the peer is registered for reconciliations.
90+ */
91+ void TryRemovingFromSet(NodeId peer_id, const uint256& wtxid_to_remove);
96+ * know what we need:
97+ * - size of our reconciliation set for the peer
98+ * - our q-coefficient with the peer, formatted to be transmitted as integer value
99+ * If the peer was not previously registered for reconciliations, returns nullopt.
100+ */
101+ std::optional<std::pair<uint16_t, uint16_t>> MaybeRequestReconciliation(NodeId peer_id);
0 std::optional<std::pair<uint16_t, uint16_t>> MaybeRequestReconciliation(NodeId peer_id, std::chrono::microsecond now);
Passing in the time here would be nice for testing.
I’ve drafted up a fuzz target for the TxReconciliationTracker here: https://github.com/dergoegge/bitcoin/commits/2022-10-erlay2
I think having a fuzz target is worth while, now that there is quite a bit more going on besides registering/forgetting peers. I am happy to PR that separately after this is merged.
Some suggestions inline after working on that target:
318+ std::vector<NodeId> eligible_peers;
319+ const bool we_initiate = recon_state.m_we_initiate;
320+ // Find all peers of the same reconciliation direction.
321+ std::for_each(m_states.begin(), m_states.end(),
322+ [&eligible_peers, we_initiate](auto indexed_state) {
323+ const auto& cur_state = std::get<TxReconciliationState>(indexed_state.second);
This throws a std::bad_variant_access for peers that aren’t fully registered yet (found by the fuzzer).
Maybe create an internal helper method that only iterates through all fully registered peers/states to avoid this entirely?
5370+ // We must look into the reconciliation queue first. Since the queue applies to all peers,
5371+ // this peer might block other reconciliation if we don't make this call regularly and
5372+ // unconditionally.
5373+ bool reconcile = false;
5374+ if (m_txreconciliation) {
5375+ // reconcile = m_txreconciliation->IsPeerNextToReconcileWith(pto->GetId(), current_time);
5803+ {
5804+ if (!m_chainman.ActiveChainstate().IsInitialBlockDownload()) {
5805+ if (reconcile) {
5806+ const auto reconciliation_request_params = m_txreconciliation->InitiateReconciliationRequest(pto->GetId());
5807+ if (reconciliation_request_params) {
5808+ const auto [local_set_size, local_q_formatted] = (*reconciliation_request_params);
373+ const double inbound_destinations = all_inbound_peers * INBOUND_FANOUT_DESTINATIONS_FRACTION;
374+ const size_t outbound_destinations = std::min<int>(0, OUTBOUND_FANOUT_DESTINATIONS - already_flooded_to_outbound);
375+
376+ std::vector<std::pair<uint64_t, NodeId>> best_inbound_peers(size_t(inbound_destinations) + 1, std::make_pair(0, 0));
377+ std::vector<std::pair<uint64_t, NodeId>> best_outbound_peers(outbound_destinations, std::make_pair(0, 0));
378+ ;
02e480d174e4edb1ffedb643fc454ad9b77e9001
75@@ -76,6 +76,44 @@ class TxReconciliationTracker
76 ReconciliationRegisterResult RegisterPeer(NodeId peer_id, bool is_peer_inbound,
77 uint32_t peer_recon_version, uint64_t remote_salt);
78
79+ /** TODO */
403+ return result;
404+ }
405+
406+ // For a given peer, see whether the transaction should be flooded
407+ bool ShouldFloodTo(const uint256& wtxid, const std::vector<uint256> parents_wtxids,
408+ const std::vector<uint256> wtxids_to_reconcile, CSipHasher deterministic_randomizer, NodeId peer_id,
parents_wtxids and wtxids_to_reconcile, same for TxReconciliationTracker::ShouldFloodTo.
ShouldFloodTo function?
Curious, what we care about a low transaction relay latency ? Like I would say lower latency > makes it a) harder to observe original transaction broadcast by deanonymization attacker b) disincentive transaction issuers front-running the standard tx-relay rules to place their transactions first in the mempools in case of congestion and c) for instant/0confs flows improve UX, do we have more properties ?
Yeah low-transaction relay latency is still to reconsider in light of all the timevalue DoS that you might have as a Lightning node running on top of Bitcoin Core. Implications with things like #21224.
3719@@ -3699,6 +3720,9 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
3720 if (!fAlreadyHave && !m_chainman.ActiveChainstate().IsInitialBlockDownload()) {
3721 AddTxAnnouncement(pfrom, gtxid, current_time);
3722 }
3723+ if (m_txreconciliation && m_txreconciliation->IsPeerRegistered(pfrom.GetId()) && gtxid.IsWtxid()) {
3724+ m_txreconciliation->TryRemovingFromSet(pfrom.GetId(), gtxid.GetHash());
BLOCKTXN should be TryRemovingFromSet().
152+ * For reconciliation peers the delay is chosen according the following
153+ * considerations:
154+ * 1. Reconciliation. When the transaction is reconciled, this delay is applied to adding to
155+ * reconciliation sets, not actual reconciliation (less frequent). That should happen rather
156+ * fast, so that sets are in sync and reconciliation is efficient. At the same time, not too
157+ * fast to avoid privacy leaks (e.g., infer connections via set probing).
92+ *
93+ * The caller *must* check that the peer is registered for reconciliations.
94+ */
95+ bool TryRemovingFromSet(NodeId peer_id, const uint256& wtxid_to_remove);
96+
97+ bool IsPeerNextToReconcileWith(NodeId peer_id, std::chrono::microseconds now);
TxReconciliationTracker::IsPeerNextToReconcileWith() does, like assumptions on peers queue and phase we’re in.
98@@ -99,7 +99,9 @@ static constexpr int32_t MAX_PEER_TX_REQUEST_IN_FLIGHT = 100;
99 /** Maximum number of transactions to consider for requesting, per peer. It provides a reasonable DoS limit to
100 * per-peer memory usage spent on announcements, while covering peers continuously sending INVs at the maximum
101 * rate (by our own policy, see INVENTORY_BROADCAST_PER_SECOND) for several minutes, while not receiving
102- * the actual transaction (from any peer) in response to requests for them. */
103+ * the actual transaction (from any peer) in response to requests for them.
104+ * Also limits a maximum number of elements to store in the reconciliation set.
MAX_SET_SIZE=3000.
206+ if (!IsPeerRegistered(peer_id)) return 0;
207+ auto& recon_state = std::get<TxReconciliationState>(m_states.find(peer_id)->second);
208+
209+ size_t added = 0;
210+ for (auto& wtxid : txs_to_reconcile) {
211+ added += recon_state.m_local_set.insert(wtxid).second;
MAX_SET_SIZE) is enforced by ShouldFanoutTo() which happens before the call to AddToSet(), don’t know if it could be more conservative with another check here.
340+ }
341+ }
342+ };
343+
344+ for (auto indexed_state : m_states) {
345+ const auto& cur_state = std::get<TxReconciliationState>(indexed_state.second);
std::visit.
5683+
5684+ bool fanout = true;
5685+ if (reconciles_txs) {
5686+ LOCK(m_mempool.cs);
5687+ auto txiter = m_mempool.GetIter(txinfo.tx->GetHash());
5688+ assert(txiter);
txinfo is queried, and the time we call m_mempool.GetIter() some other thread could remove the tx from the mempool (because no lock is held), so that this assert would be hit.
210+ bool we_initiate, float limit) const EXCLUSIVE_LOCKS_REQUIRED(m_txreconciliation_mutex)
211+ {
212+ // To handle fractional values, we add one peer optimistically and then probabilistically
213+ // drop it later.
214+ // Initiate the randomness here so that it's not influenced by the following code.
215+ double fractional_peer;
limit) and double (fractional_peer) is probably not ideal.
204@@ -192,6 +205,87 @@ class TxReconciliationTracker::Impl
205 LOCK(m_txreconciliation_mutex);
206 return IsPeerRegistered(peer_id);
207 }
208+
209+ std::vector<NodeId> GetFanoutTargets(const uint256& wtxid, CSipHasher& deterministic_randomizer,
wtxid is unused
ShouldFanoutTo() and GetFanoutTargets()), especially since this is not covered in the BIP.
4042@@ -4021,6 +4043,10 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
4043 return;
4044 }
4045
4046+ if (m_txreconciliation && m_txreconciliation->IsPeerRegistered(pfrom.GetId())) {
4047+ m_txreconciliation->TryRemovingFromSet(pfrom.GetId(), wtxid);
4048+ }
0 if (m_txreconciliation) m_txreconciliation->TryRemovingFromSet(pfrom.GetId(), wtxid);
You are already checking internally that the peer is registered.
5743+ } else {
5744+ size_t inbounds_fanouted = 0, outbounds_fanouted = 0;
5745+ m_connman.ForEachNode([&inbounds_fanouted, &outbounds_fanouted, this](CNode* pnode) {
5746+ inbounds_fanouted += pnode->IsInboundConn() && pnode->m_relays_txs;
5747+ outbounds_fanouted += pnode->IsFullOutboundConn() && pnode->m_relays_txs && !m_txreconciliation->IsPeerRegistered(pnode->GetId());
5748+ });
0 {
1 LOCK(m_peer_mutex);
2 for (auto [id, peer] : m_peer_map) {
3 const auto state{State(id)};
4 if (!state) continue;
5
6 if (auto tx_relay = peer->GetTxRelay()) {
7 LOCK(tx_relay->m_bloom_filter_mutex);
8 inbounds_fanouted += state->m_is_inbound && tx_relay->m_relay_txs;
9 outbounds_fanouted += !state->m_is_inbound && tx_relay->m_relay_txs && !m_txreconciliation->IsPeerRegistered(id);
10 }
11 }
12 }
5741+ // sets for this one child is not good either.
5742+ fanout = true;
5743+ } else {
5744+ size_t inbounds_fanouted = 0, outbounds_fanouted = 0;
5745+ m_connman.ForEachNode([&inbounds_fanouted, &outbounds_fanouted, this](CNode* pnode) {
5746+ inbounds_fanouted += pnode->IsInboundConn() && pnode->m_relays_txs;
0 inbounds_fanouted += pnode->IsInboundConn() && pnode->m_relays_txs && !m_txreconciliation->IsPeerRegistered(pnode->GetId());
Right? because otherwise this includes erlay peers which we are not flooding to by default.
5747+ outbounds_fanouted += pnode->IsFullOutboundConn() && pnode->m_relays_txs && !m_txreconciliation->IsPeerRegistered(pnode->GetId());
5748+ });
5749+
5750+ auto fanout_randomizer = m_connman.GetDeterministicRandomizer(RANDOMIZER_ID_FANOUTTARGET);
5751+ fanout = m_txreconciliation->ShouldFanoutTo(wtxid, fanout_randomizer, pto->GetId(),
5752+ std::make_pair(m_connman.GetNodeCount(ConnectionDirection::In), inbounds_fanouted),
m_connman.GetNodeCount(ConnectionDirection::In) includes all inbound connections, even ones that do want transactions relayed to them. IIUC, you want this to be “number of all inbound peers that sent fRelay=true” and the second member of the pair should be “number of all non-erlay inbound peers that sent fRelay=true”
31+ self.extra_args = [['-txreconciliation']]
32+
33+ def run_test(self):
34+ t = int(time.time())
35+ self.nodes[0].setmocktime(t)
36+ self.generate(self.nodes[0], 200) # mature coinbase UTXO used later
In 00df0aefff938e849bdab5278d6658d5c9d5d064, If the purpose is maturing them, I think we could do:
0 self.generate(self.nodes[0], COINBASE_MATURITY) # mature coinbase UTXO used later
5794@@ -5787,6 +5795,21 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
5795 if (!vInv.empty())
5796 m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
5797
5798+ //
5799+ // Message: reconciliation request
5800+ //
5801+ {
5802+ if (!m_chainman.ActiveChainstate().IsInitialBlockDownload()) {
200@@ -184,6 +201,8 @@ static constexpr double MAX_ADDR_RATE_PER_SECOND{0.1};
201 static constexpr size_t MAX_ADDR_PROCESSING_TOKEN_BUCKET{MAX_ADDR_TO_SEND};
202 /** The compactblocks version we support. See BIP 152. */
203 static constexpr uint64_t CMPCTBLOCKS_VERSION{2};
204+/** Used to determine whether to fanout or to reconcile a transaction with a given peer */
171@@ -166,5 +172,5 @@ void TxReconciliationTracker::ForgetPeer(NodeId peer_id)
172
173 bool TxReconciliationTracker::IsPeerRegistered(NodeId peer_id) const
174 {
175- return m_impl->IsPeerRegistered(peer_id);
176+ return m_impl->IsPeerRegisteredExternal(peer_id);
Note to reviewers, this is used in AddToSet(), which is itself called in the transaction to relay selection at message NetMsgType::INV generation in SendMessages().
This allows one m_txreconciliation_mutex lock taking to cover the insert in the m_states entry.
16@@ -17,7 +17,7 @@ namespace {
17 /** Static salt component used to compute short txids for sketch construction, see BIP-330. */
18 const std::string RECON_STATIC_SALT = "Tx Relay Salting";
19 const HashWriter RECON_SALT_HASHER = TaggedHash(RECON_STATIC_SALT);
20-
21+constexpr size_t MAX_SET_SIZE = 3000;
AddToSet).
152+ //
153+ // Transactions which don't make it to the set due to the limit are announced via fan-out.
154+ if (recon_state.m_local_set.size() >= MAX_SET_SIZE) return false;
155+
156+ Assume(recon_state.m_local_set.insert(wtxid).second);
157+ LogPrintLevel(BCLog::TXRECONCILIATION, BCLog::Level::Debug, "Added a transaction to the reconciliation set for peer=%d. " /* Continued */
wtxid could be added to the LogPrintLevel to ease debug.
5656@@ -5650,9 +5657,13 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
5657 }
5658
5659 if (auto tx_relay = peer->GetTxRelay(); tx_relay != nullptr) {
5660+ // Lock way before it's used to maintain lock ordering.
5661+ LOCK(m_mempool.cs);
5662+ LOCK(m_peer_mutex);
LOCK2 could be used.
5757+ auto txiter = m_mempool.GetIter(txinfo.tx->GetHash());
5758+ if (txiter) {
5759+ if ((*txiter)->GetCountWithDescendants() > 1) {
5760+ // If a transaction has in-mempool children, always fanout it.
5761+ // Until package relay is implemented, this is needed to avoid
5762+ // breaking parent+child relay expectations in some cases.
5764+ // Potentially reconciling parent+child would mean that for every
5765+ // child we need to to check if any of the parents is currently
5766+ // reconciled so that the child isn't fanouted ahead. But then
5767+ // it gets tricky when reconciliation sets are full: a) the child
5768+ // can't just be added; b) removing parents from reconciliation
5769+ // sets for this one child is not good either.
I think the issue can be even worst if you’re considering the receiver-initiated approach, where the receiver might have already the child from another peer and the BIP331’s getskgtxns is in-flight. Both the parent and child wtxid reconciliation set entries are going to be wasted, I think.
I believe the best approach to combine it with package relay support for now is just to look if it doesn’t raise privacy or CPU robustness approach for now (at least until one is finalized and deployed), and swallow the bullet in term of bandwidth savings performance.
Reviewed up to dfde439 - Commit “p2p: Add transactions to reconciliation sets”
About Erlay:
p2p_reqtxrcncl.py is the source of failing checks for all the unsuccessful CI tasks.
295+ // number of inbound targets.
296+ const double inbound_targets = (inbounds_nonrcncl_tx_relay + inbound_rcncl_peers) * INBOUND_FANOUT_DESTINATIONS_FRACTION;
297+ destinations = inbound_targets - inbounds_nonrcncl_tx_relay;
298+ }
299+
300+ if (destinations < 0.01) {
vInv announcement. A node operator could have a low maxconnections.
152+ * 2a) announcing both ways simultaneously (inefficiency);
153+ * 2b) inference based on the announced transactions (privacy leak).
154+ * That's why it's ok to make this delay low as well, and lower delay is generally good to
155+ * facilitate good transaction relay speed when slow reconciliations prevail. */
156 static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL{5s};
157+static constexpr auto INBOUND_INVENTORY_BROADCAST_INTERVAL_RECON{2s};