Erlay: bandwidth-efficient transaction relay protocol #18261

3437+
3438+        uint64_t local_salt = connman->m_local_recon_salt;
3439+
3440+        std::string salt1, salt2;
3441+        if (remote_salt < local_salt) {
3442+            salt1 = std::to_string(remote_salt);

749+    for (size_t i = 0; i < remote_missing_short_ids.size(); ++i) {
750+        auto local_tx = m_recon_state->local_short_id_mapping.find(remote_missing_short_ids[i]);
751+        if (local_tx == m_recon_state->local_short_id_mapping.end()) {
752+            LogPrint(BCLog::NET, "Unknown transaction requested by short id=%d by peer=%d\n", remote_missing_short_ids[i], GetId());
753+        }
754+        remote_missing.push_back(local_tx->second);

3446@@ -3417,6 +3447,274 @@ bool ProcessMessage(CNode* pfrom, const std::string& msg_type, CDataStream& vRec
3447         return true;
3448     }
3449 
3450+    std::chrono::microseconds current_time = GetTime<std::chrono::microseconds>();
3451+
3452+    // Record an (expected) reconciliation request with parameters to respond when time comes.
3453+    // All initial reconciliation responses will be done at the same time to prevent tx-related privacy leaks.
3454+    if (strCommand == NetMsgType::REQRECON) {
3455+        if (!pfrom->m_recon_state->sender) return true;

407 {
408     UNDEFINED = 0,
409     MSG_TX = 1,
410     MSG_BLOCK = 2,
411-    // The following can only occur in getdata. Invs always use TX or BLOCK.
412+    MSG_WTX = 5, // Defined in BIP XXX

233@@ -234,6 +234,46 @@ extern const char *GETBLOCKTXN;
234  * @since protocol version 70014 as described by BIP 152
235  */
236 extern const char *BLOCKTXN;
237+/**
238+ * Indicates that a node prefers to relay transactions via wtxid, rather than
239+ * txid.
240+ * @since protocol version 70016 as described by BIP XXX.

125+        int num_roots = minisketch_decode(state, count + 1, roots.data());
126+        CHECK(overfill || num_roots >= 0);
127+        CHECK(num_roots < 1 || minisketch_decode(state, num_roots - 1, roots.data()) == -1); // Decoding with a too-low maximum should fail.
128+        if (!overfill) {
129+            std::sort(roots.begin(), roots.begin() + num_roots);
130+//            fprintf(stderr, "Solut: ");

2374+            bool be_recon_requestor, be_recon_responder;
2375+            // Currently reconciliation requests flow only in one direction inbound->outbound.
2376+            if (pfrom.IsFullOutboundConn()) {
2377+                be_recon_requestor = true;
2378+                be_recon_responder = false;
2379+            } else {

259@@ -260,6 +260,14 @@ extern const char* CFCHECKPT;
260  * @since protocol version 70016 as described by BIP 339.
261  */
262 extern const char* WTXIDRELAY;
263+/**
264+ * Contains 2 1-byte bools, a 4-byte version number and an 8-byte salt.
265+ * Indicates that a node is willing to participate in transaction reconciliation,
266+ * either as a sender or a receiver.
267+ * The salt is used to compute short txids needed for efficient reconciliation.
268+ * @since protocol version 80001 as described by BIP 330

145@@ -146,6 +146,10 @@ static constexpr uint32_t MAX_GETCFILTERS_SIZE = 1000;
146 static constexpr uint32_t MAX_GETCFHEADERS_SIZE = 2000;
147 /** the maximum percentage of addresses from our addrman to return in response to a getaddr message. */
148 static constexpr size_t MAX_PCT_ADDR_TO_SEND = 23;
149+/** Static component of the salt used to compute short txids for transaction reconciliation. */
150+static const std::string RECON_STATIC_SALT = "Tx Relay Salting";
151+/** Default coefficient used to estimate set difference for tx reconciliation. */
152+static constexpr double DEFAULT_RECON_Q = 0.02;

2616@@ -2559,6 +2617,62 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
2617         return;
2618     }
2619 
2620+    // Received from an inbound peer planning to reconcilie transactions with us, or

2622+    // If received from outgoing, adds the peer to the reconciliation queue.
2623+    // Feature negotiation of tx reconciliation should happen between VERSION and
2624+    // VERACK, to avoid relay problems from switching after a connection is up.
2625+    if (msg_type == NetMsgType::SENDRECON) {
2626+        if (!pfrom.m_tx_relay) return;
2627+        if (peer->m_recon_state != nullptr) return; // Do not support reconciliation salt/version updates.

2633+        if (recon_version != 1) return;
2634+
2635+         // Do not flood through inbound connections which support reconciliation to save bandwidth.
2636+         bool flood_to = false;
2637+         if (pfrom.IsInboundConn()) {
2638+             if (!recon_sender) return;

2666+        peer->m_recon_state->m_salt = full_salt;
2667+
2668+        // Reconcile with all outbound peers supporting reconciliation (even if we flood to them),
2669+        // to not miss transactions they have for us but won't flood.
2670+        if (peer->m_recon_state->m_responder) {
2671+            m_recon_queue.push_back(&pfrom);

832@@ -827,6 +833,30 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec
833 
834 } // namespace
835 
836+size_t PeerManager::GetTxRelayOutboundCountByRelayType(bool flooding) const

832@@ -827,6 +833,30 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec
833 
834 } // namespace
835 
836+size_t PeerManager::GetTxRelayOutboundCountByRelayType(bool flooding) const
837+{
838+    size_t result = 0;
839+    m_connman.ForEachNode([flooding, &result](CNode* pnode) {
840+        if (!pnode->m_tx_relay)

838+    size_t result = 0;
839+    m_connman.ForEachNode([flooding, &result](CNode* pnode) {
840+        if (!pnode->m_tx_relay)
841+            return;
842+        if (!pnode->IsFullOutboundConn() && !pnode->IsManualConn())
843+            return;

2369@@ -2370,6 +2370,28 @@ static void ProcessGetCFCheckPt(CNode& peer, CDataStream& vRecv, const CChainPar
2370     connman.PushMessage(&peer, std::move(msg));
2371 }
2372 
2373+/**
2374+ * Announce transactions a peer is missing after reconciliation is done.
2375+ * No need to add transactions to peer's filter or do checks
2376+ * because it was already done when adding to the reconciliation set.
2377+ */
2378+void static AnnounceTxs(std::vector<uint256> remote_missing_wtxids, CNode* pto, CNetMsgMaker msgMaker, CConnman* connman)

2377+ */
2378+void static AnnounceTxs(std::vector<uint256> remote_missing_wtxids, CNode* pto, CNetMsgMaker msgMaker, CConnman* connman)
2379+{
2380+    if (remote_missing_wtxids.size() != 0) {
2381+        std::vector<CInv> remote_missing_invs;
2382+        for (uint256 wtxid: remote_missing_wtxids) {

934@@ -917,7 +935,7 @@ void PeerManager::ReattemptInitialBroadcast(CScheduler& scheduler) const
935 
936         if (tx != nullptr) {
937             LOCK(cs_main);
938-            RelayTransaction(txid, tx->GetWitnessHash(), m_connman);
939+            RelayTransaction(txid, tx->GetWitnessHash(), m_connman, false);

1555         if (state->m_wtxid_relay) {
1556-            pnode->PushTxInventory(wtxid);
1557+            pnode->PushTxInventory(wtxid, flood);
1558         } else {
1559-            pnode->PushTxInventory(txid);
1560+            pnode->PushTxInventory(txid, flood);

2104@@ -2087,7 +2105,7 @@ void PeerManager::ProcessOrphanTx(std::set<uint256>& orphan_work_set)
2105 
2106         if (AcceptToMemoryPool(m_mempool, state, porphanTx, &removed_txn, false /* bypass_limits */)) {
2107             LogPrint(BCLog::MEMPOOL, "   accepted orphan tx %s\n", orphanHash.ToString());
2108-            RelayTransaction(orphanHash, porphanTx->GetWitnessHash(), m_connman);
2109+            RelayTransaction(orphanHash, porphanTx->GetWitnessHash(), m_connman, true);

458+     * Salt is generated randomly per-connection to prevent linking of
459+     * connections belonging to the same physical node.
460+     * Also, salts should be different per-connection to prevent halting
461+     * of relay of particular transactions due to collisions in short IDs.
462+     */
463+    uint64_t m_local_recon_salt;

516+         */
517+        double m_local_q;
518+    };
519+
520+    /// nullptr if we're not reconciling (neither passively nor actively) with this peer.
521+    std::unique_ptr<ReconState> m_recon_state;

474+     * asymmetrical, there is always a requestor and a responder. At the end of the
475+     * sequence, nodes are supposed to exchange transactions, so that both of them
476+     * have all relevant transactions. For more protocol details, refer to BIP-0330.
477+     */
478+    struct ReconState {
479+        ReconState(){};

148+     * Transaction reconciliation should happen with peers in the same order,
149+     * because the efficiency gain is the highest when reconciliation set difference
150+     * is predictable. This queue is used to maintain the order of
151+     * peers chosen for reconciliation.
152+     */
153+    std::deque<CNode*> m_recon_queue;

4527@@ -4504,11 +4528,11 @@ bool PeerManager::SendMessages(CNode* pto)
4528                 bool fSendTrickle = pto->HasPermission(PF_NOBAN);
4529                 if (pto->m_tx_relay->nNextInvSend < current_time) {
4530                     fSendTrickle = true;
4531-                    if (pto->IsInboundConn()) {
4532-                        pto->m_tx_relay->nNextInvSend = std::chrono::microseconds{m_connman.PoissonNextSendInbound(count_microseconds(current_time), INVENTORY_BROADCAST_INTERVAL)};
4533-                    } else {
4534-                        // Use half the delay for outbound peers, as there is less privacy concern for them.
4535+                    if (peer->m_recon_state || !pto->IsInboundConn()) {

4630+                            // Always flood to non-reconciliation nodes supporting tx relay.
4631+                            // For reconciliation nodes, flood if flood_to is set up and transaction is meant for flooding.
4632+                            vInv.push_back(inv);
4633+                        else {
4634+                            // Storing to populate the reconciliation set.
4635+                            txs_to_reconcile.push_back(txinfo.tx->GetWitnessHash());

4680+                            // We don't care about a laggy peer (1) because we probably can't help them even if we flood transactions.
4681+                            // However, exploiting (2) should not prevent us from relaying certain transactions.
4682+                            // Since computing a sketch over too many elements is too expensive, we will just flood some transactions here.
4683+                            std::vector<uint256> txs_to_flood = std::vector<uint256>(txs_to_reconcile.end() - recon_set_overflow, txs_to_reconcile.end());
4684+                            txs_to_reconcile.resize(txs_to_reconcile.size() - recon_set_overflow);
4685+                            AnnounceTxs(txs_to_flood, *pto, msgMaker, m_connman);

122@@ -123,12 +123,12 @@ static constexpr std::chrono::hours AVG_LOCAL_ADDRESS_BROADCAST_INTERVAL{24};
123 static constexpr std::chrono::seconds AVG_ADDRESS_BROADCAST_INTERVAL{30};
124 /** Average delay between trickled inventory transmissions in seconds.
125  *  Blocks and peers with noban permission bypass this, outbound peers get half this delay. */
126-static const unsigned int INVENTORY_BROADCAST_INTERVAL = 5;
127+static const unsigned int INVENTORY_BROADCAST_INTERVAL = 2;

128 /** Maximum rate of inventory items to send per second.
129  *  Limits the impact of low-fee transaction floods. */
130 static constexpr unsigned int INVENTORY_BROADCAST_PER_SECOND = 7;
131 /** Maximum number of inventory items to send per transmission. */
132-static constexpr unsigned int INVENTORY_BROADCAST_MAX = INVENTORY_BROADCAST_PER_SECOND * INVENTORY_BROADCAST_INTERVAL;
133+static constexpr unsigned int INVENTORY_BROADCAST_MAX = INVENTORY_BROADCAST_PER_SECOND * INVENTORY_BROADCAST_INTERVAL * 4;

175@@ -176,6 +176,12 @@ static const unsigned int MAX_RECON_SET = 10000;
176  * Less frequent reconciliations would introduce high transaction relay latency.
177  */
178 static constexpr std::chrono::microseconds RECON_REQUEST_INTERVAL{std::chrono::seconds{16}};
179+/**
180+ * Interval between responding to peers' reconciliation requests.
181+ * We don't respond to reconciliation requests right away because that would enable monitoring
182+ * when we receive transactions (privacy leak).

4039+        uint16_t peer_recon_set_size, peer_q;
4040+        vRecv >> peer_recon_set_size >> peer_q;
4041+        peer->m_recon_state->m_incoming_recon = Peer::ReconState::ReconPhase::INIT_REQUESTED;
4042+        peer->m_recon_state->m_remote_set_size = peer_recon_set_size;
4043+        peer->m_recon_state->m_remote_q = double(peer_q / Q_PRECISION);
4044+        peer->m_recon_state->m_next_recon_respond = NextReconRespond(current_time);

194+* but the result will be nonsense (false-positive decoding).
195+* Given this coef, a false positive probability will be of 1 in 2**coef.
196+*/
197+static constexpr unsigned int RECON_FALSE_POSITIVE_COEF = 16;
198+static_assert(RECON_FALSE_POSITIVE_COEF <= 256,
199+    "Reducing reconciliation false positives beyond 1 in 2**256 is not supported");

4873+        // Message: reconciliation response
4874+        //
4875+        if (peer->m_recon_state) {
4876+            // Respond to a requested reconciliation to enable efficient transaction exchange.
4877+            // Respond only periodically to a) limit CPU usage for sketch computation,
4878+            // and, b) limit transaction posesssion privacy leak.

4883+                if (response_sketch != nullptr) {
4884+                    // It is possible to create a sketch if we had at least one transaction to send to the peer.
4885+                    size_t ser_size = minisketch_serialized_size(response_sketch);
4886+                    uint8_t skdata[MAX_SKETCH_CAPACITY * BYTES_PER_SKETCH_CAPACITY];
4887+                    minisketch_serialize(response_sketch, skdata);
4888+                    response_skdata.resize(ser_size);

4889+                    std::copy(skdata, skdata + ser_size, response_skdata.begin());
4890+                }
4891+                m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::SKETCH, response_skdata));
4892+
4893+                peer->m_recon_state->m_incoming_recon = Peer::ReconState::ReconPhase::INIT_RESPONDED;
4894+                peer->m_recon_state->m_local_set.clear();

689+         */
690+        void FinalizeReconciliation(bool clear_local_set, LocalQAction action, size_t actual_local_missing, size_t actual_remote_missing)
691+        {
692+            // According to the erlay spec, reconciliation is initiated by inbound peers.
693+            if (m_sender) {
694+                assert(m_incoming_recon != ReconPhase::NONE);

2650+            // We currently don't support reconciliations with outbound peers which
2651+            // don't want to be reconciliation responders (send us their sketches),
2652+            // or want to be reconciliation senders (request our sketches).
2653+            // Just ignore SENDRECON and use normal flooding for transaction relay with them.
2654+            if (recon_sender) return;
2655+            if (!recon_responder) return;

925@@ -899,6 +926,14 @@ void PeerManager::UpdateNextReconRequest(std::chrono::microseconds now)
926     m_next_recon_request = now + RECON_REQUEST_INTERVAL / m_recon_queue.size();
927 }
928 
929+std::chrono::microseconds PeerManager::NextReconRespond(std::chrono::microseconds now)
930+{
931+    if (m_next_recon_respond < now) {
932+        m_next_recon_respond = now + RECON_RESPONSE_INTERVAL;

647+        {
648+            std::vector<uint32_t> short_ids;
649+            for (uint256 wtxid: local_set) {
650+                uint32_t short_txid = ComputeShortID(wtxid);
651+                short_ids.push_back(short_txid);
652+                m_local_short_id_mapping.insert(std::pair<uint32_t, uint256>(short_txid, wtxid));

4219+
4220+        std::vector<uint8_t> skdata;
4221+        vRecv >> skdata;
4222+
4223+        // Attempt to decode the received sketch with a local sketch.
4224+        if (skdata.size() / BYTES_PER_SKETCH_CAPACITY > MAX_SKETCH_CAPACITY) return;

4227+        uint8_t remote_sketch_serialized[MAX_SKETCH_CAPACITY * BYTES_PER_SKETCH_CAPACITY];
4228+        std::copy(skdata.begin(), skdata.end(), remote_sketch_serialized);
4229+        minisketch_deserialize(remote_sketch, remote_sketch_serialized);
4230+
4231+        minisketch* working_sketch; // Contains sketch of the set difference
4232+        minisketch* local_sketch = peer->m_recon_state->ComputeSketch(peer->m_recon_state->m_local_set, remote_sketch_capacity);

4261@@ -4252,6 +4262,16 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
4262             AnnounceTxs(remote_missing, pfrom, msgMaker, m_connman);
4263             m_connman.PushMessage(&pfrom, msgMaker.Make(NetMsgType::RECONCILDIFF, uint8_t(RECON_SUCCESS), local_missing));
4264             peer->m_recon_state->FinalizeReconciliation(true, Peer::ReconState::LocalQAction::Q_RECOMPUTE, local_missing.size(), remote_missing.size());
4265+        } else {
4266+            // Initial reconciliation failed.
4267+            // Store the received sketch and the local sketch, request extension.
4268+
4269+            assert(remote_sketch != nullptr);

2667+        } else {
2668+            salt1 = ToString(local_salt);
2669+            salt2 = ToString(remote_salt);
2670+        }
2671+        uint256 full_salt;
2672+        CSHA256().Write((unsigned char*)RECON_STATIC_SALT.data(), RECON_STATIC_SALT.size()).

485+
486+        /// Whether this peer will respond to reconciliation requests.
487+        bool m_responder;
488+
489+        /**
490+         * Whether we should flood transactions to the peer.

503+         * These short IDs will be salted so that they are not the same
504+         * across all pairs of peers, because otherwise it would enable network-wide
505+         * collisions which may (intentionally or not) halt relay of certain transactions.
506+         * Both of the peers contribute to the salt.
507+         */
508+        uint256 m_salt;

517+         * to better predict future set size differences.
518+         */
519+        double m_local_q;
520+    };
521+
522+    RecursiveMutex cs_recon_state;

2658+            // TODO: Flood only through a limited number of outbound connections.
2659+            flood_to = true;
2660+        }
2661+
2662+        uint64_t local_salt = peer->m_local_recon_salt;
2663+        std::string salt1, salt2;

1196     {
1197         if (m_tx_relay == nullptr) return;
1198         LOCK(m_tx_relay->cs_tx_inventory);
1199         if (!m_tx_relay->filterInventoryKnown.contains(hash)) {
1200-            m_tx_relay->setInventoryTxToSend.insert(hash);
1201+            m_tx_relay->setInventoryTxToSend.insert(std::pair<uint256, bool>(hash, flood));

163+ * Sets the bound on the following objects:
164+ * - reconciliation set
165+ * - reconciliation set snapshot
166+ * - reconciliation short-full id mapping
167+ */
168+static const unsigned int MAX_RECON_SET = 10000;

2625+    // from an outgoing peer demonstrating readiness to do reconciliations.
2626+    // If received from outgoing, adds the peer to the reconciliation queue.
2627+    // Feature negotiation of tx reconciliation should happen between VERSION and
2628+    // VERACK, to avoid relay problems from switching after a connection is up.
2629+    if (msg_type == NetMsgType::SENDRECON) {
2630+        if (!pfrom.m_tx_relay) return;

4644@@ -4608,9 +4645,17 @@ bool PeerManager::SendMessages(CNode* pto)
4645                             continue;
4646                         }
4647                         if (pto->m_tx_relay->pfilter && !pto->m_tx_relay->pfilter->IsRelevantAndUpdate(*txinfo.tx)) continue;
4648-                        // Send
4649+
4650+                        bool flood_tx = it->second;
4651+                        if (!peer->m_recon_state || (peer->m_recon_state->m_flood_to && flood_tx))

1008@@ -1009,9 +1009,10 @@ class CNode
1009 
1010         mutable RecursiveMutex cs_tx_inventory;
1011         CRollingBloomFilter filterInventoryKnown GUARDED_BY(cs_tx_inventory){50000, 0.000001};
1012-        // Set of transaction ids we still have to announce.
1013-        // They are sorted by the mempool before relay, so the order is not important.
1014-        std::set<uint256> setInventoryTxToSend;
1015+        // Set of transaction ids we still have to announce, and whether we may flood them
1016+        // in case if peer is meant to receive flooding, as opposed to reconcile.
1017+        // Transactions are sorted by the mempool before relay, so the order is not important.

146@@ -146,6 +147,65 @@ static constexpr uint32_t MAX_GETCFILTERS_SIZE = 1000;
147 static constexpr uint32_t MAX_GETCFHEADERS_SIZE = 2000;
148 /** the maximum percentage of addresses from our addrman to return in response to a getaddr message. */
149 static constexpr size_t MAX_PCT_ADDR_TO_SEND = 23;
150+/** Static component of the salt used to compute short txids for transaction reconciliation. */
151+static const std::string RECON_STATIC_SALT = "Tx Relay Salting";
152+/** Used to convert a floating point reconciliation coefficient q to an int for transmission. */
153+static constexpr double Q_PRECISION{2 << 12};

183+ * when we receive transactions (privacy leak).
184+ */
185+static constexpr std::chrono::microseconds RECON_RESPONSE_INTERVAL{std::chrono::seconds{2}};
186+/** The size of the field, used to compute sketches to reconcile transactions (see BIP-330). */
187+static constexpr unsigned int RECON_FIELD_SIZE = 32;
188+static_assert(RECON_FIELD_SIZE % 8 == 0, "Field size should be divisible by 8");

699+                minisketch_add_uint64(sketch, short_id);
700+            }
701+            return sketch;
702+        }
703+
704+        minisketch* ComputeExtendedSketch() {

41+
42+def mul2(x):
43+    """Compute 2*x in GF(2^FIELD_BITS)"""
44+    return (x << 1) ^ (FIELD_MODULUS if x.bit_length() >= FIELD_BITS else 0)
45+
46+def mul(x, y):

555@@ -556,9 +556,11 @@ class CNode
556 
557         mutable RecursiveMutex cs_tx_inventory;
558         CRollingBloomFilter filterInventoryKnown GUARDED_BY(cs_tx_inventory){50000, 0.000001};
559-        // Set of transaction ids we still have to announce.
560-        // They are sorted by the mempool before relay, so the order is not important.
561-        std::set<uint256> setInventoryTxToSend;
562+        // Set of transaction ids we still have to announce, and whether we may flood them:

168+ * This value allows to reconcile ~100 transactions (7 tx/s * 16s) during normal system operation at capacity.
169+ * More frequent reconciliations would cause significant constant bandwidth overhead due to
170+ * reconciliation metadata (sketch sizes etc.), which would nullify the efficiency.
171+ * Less frequent reconciliations would introduce high transaction relay latency.
172+ */
173+static constexpr std::chrono::microseconds RECON_REQUEST_INTERVAL{std::chrono::seconds{16}};

177+ * when we receive transactions (privacy leak).
178+ */
179+static constexpr std::chrono::microseconds RECON_RESPONSE_INTERVAL{std::chrono::seconds{2}};
180+
181+/** Represents a reconciliation result, used to decide what to do when the reconciliation is over. */
182+enum ReconResult {

2354@@ -2267,6 +2355,29 @@ static void ProcessGetCFCheckPt(CNode& peer, CDataStream& vRecv, const CChainPar
2355     connman.PushMessage(&peer, std::move(msg));
2356 }
2357 
2358+/**
2359+ * Announce transactions a peer is missing after reconciliation is done.
2360+ * No need to add transactions to peer's filter or do checks
2361+ * because it was already done when adding to the reconciliation set.
2362+ */
2363+void static AnnounceTxs(const std::vector<uint256>& remote_missing_wtxids, CNode& pto, const CNetMsgMaker& msgMaker, CConnman& connman)

2360+ * No need to add transactions to peer's filter or do checks
2361+ * because it was already done when adding to the reconciliation set.
2362+ */
2363+void static AnnounceTxs(const std::vector<uint256>& remote_missing_wtxids, CNode& pto, const CNetMsgMaker& msgMaker, CConnman& connman)
2364+{
2365+    if (remote_missing_wtxids.size() != 0) {

2361+ * because it was already done when adding to the reconciliation set.
2362+ */
2363+void static AnnounceTxs(const std::vector<uint256>& remote_missing_wtxids, CNode& pto, const CNetMsgMaker& msgMaker, CConnman& connman)
2364+{
2365+    if (remote_missing_wtxids.size() != 0) {
2366+        std::vector<CInv> remote_missing_invs;

3942@@ -3751,6 +3943,165 @@ void PeerManager::ProcessMessage(CNode& pfrom, const std::string& msg_type, CDat
3943         return;
3944     }
3945 
3946+    std::chrono::microseconds current_time = GetTime<std::chrono::microseconds>();
3947+
3948+    // Record an (expected) reconciliation request with parameters to respond when time comes.
3949+    // All initial reconciliation responses will be done at the same time to prevent tx-related privacy leaks.
3950+    if (msg_type == NetMsgType::REQRECON) {

4080+        return;
4081+    }
4082+
4083+    // Among transactions requested by short ID here, we should send only those transactions
4084+    // sketched (stored in local set snapshot), because otherwise we would leak privacy (mempool content).
4085+    if (msg_type == NetMsgType::RECONCILDIFF) {

10@@ -11,6 +11,7 @@
11 #include <sync.h>
12 #include <txrequest.h>
13 #include <validationinterface.h>
14+#include <minisketch/include/minisketch.h>

31@@ -31,6 +32,19 @@ static const bool DEFAULT_PEERBLOOMFILTERS = false;
32 static const bool DEFAULT_PEERBLOCKFILTERS = false;
33 /** Threshold for marking a node to be discouraged, e.g. disconnected and added to the discouragement filter. */
34 static const int DISCOURAGEMENT_THRESHOLD{100};
35+/** The size of the field, used to compute sketches to reconcile transactions (see BIP-330). */
36+static constexpr unsigned int RECON_FIELD_SIZE = 32;
37+static constexpr unsigned int BYTES_PER_SKETCH_CAPACITY = RECON_FIELD_SIZE / 8;

277+        /**
278+         * Reconciliation involves computing a space-efficient representation of transaction identifiers (a sketch).
279+         * A sketch has a capacity meaning it allows reconciling at most a certain number of elements. (see BIP-330).
280+         * Considering whether we are going to send a sketch to a peer or use locally, we estimate the set difference.
281+         */
282+        Minisketch ComputeSketch(const std::set<uint256> local_set, uint16_t& capacity)

407+            }
408+            return remote_missing;
409+        }
410+    };
411+
412+    RecursiveMutex cs_recon_state;

566+     * Transaction reconciliation should happen with peers in the same order,
567+     * because the efficiency gain is the highest when reconciliation set difference
568+     * is predictable. This queue is used to maintain the order of
569+     * peers chosen for reconciliation.
570+     */
571+    RecursiveMutex cs_recon_queue;

0@@ -0,0 +1,595 @@
1+#!/usr/bin/env python3
2+# Copyhigh (c) 2016-2017 The Bitcoin Core developers
3+# Distributed under the MIT software license, see the accompanying
4+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
5+"""Test reconciliation-based transaction relay protocol.
6+

20+from enum import IntEnum
21+from io import BytesIO
22+import random
23+import hashlib
24+import time
25+import struct

154+        msg = msg_getdata(inv=[CInv(MSG_WTX, h=wtxid) for wtxid in ask_wtxids])
155+        self.send_message(msg)
156+
157+class ReconResult(IntEnum):
158+    RECON_FAILED = 0
159+    RECON_SUCCESS = 1

132+def ser_uint128(u):
133+    rs = b""
134+    for i in range(4):
135+        rs += struct.pack("<I", u & 0xFFFFFFFF)
136+        u >>= 32
137+    return rs

235+    for i in range(nit):
236+        t = struct.unpack("<B", f.read(1))[0]
237+        r.append(t)
238+    return r
239+
240+

26@@ -27,6 +27,7 @@
27 import struct
28 import sys
29 import threading
30+import socket

59@@ -59,6 +60,11 @@
60     msg_pong,
61     msg_sendaddrv2,
62     msg_sendcmpct,
63+    msg_sendrecon,

112@@ -107,6 +113,11 @@
113     b"verack": msg_verack,
114     b"version": msg_version,
115     b"wtxidrelay": msg_wtxidrelay,
116+    b"sendrecon": msg_sendrecon,

407@@ -379,6 +408,7 @@ def on_sendaddrv2(self, message): pass
408     def on_sendcmpct(self, message): pass
409     def on_sendheaders(self, message): pass
410     def on_tx(self, message): pass
411+    def on_sendrecon(self, message): pass

241+     * of relay of particular transactions due to collisions in short IDs.
242+     */
243+    const uint64_t m_local_recon_salt;
244+
245+    Mutex m_recon_state_mutex;
246+    /// nullptr if we're not reconciling (neither passively nor actively) with this peer.

 6@@ -7,6 +7,7 @@
 7 #define BITCOIN_NET_PROCESSING_H
 8 
 9 #include <net.h>
10+#include <reconciliation.h>

3406+
3407+            // Flood those transactions which were received either via flooding, or inbound reconciliation,
3408+            // but NOT via outbound reconciliation. Flooding then is mainly used for initial propagation
3409+            // of new transactions across a network of reachable nodes quickly.
3410+            LOCK(peer->m_recon_state_mutex);
3411+            bool flood = !(peer->m_recon_state && peer->m_recon_state->m_responder);

4159@@ -3944,6 +4160,163 @@ void PeerManagerImpl::ProcessMessage(CNode& pfrom, const std::string& msg_type,
4160         return;
4161     }
4162 
4163+    std::chrono::microseconds current_time = GetTime<std::chrono::microseconds>();

4915@@ -4543,6 +4916,7 @@ bool PeerManagerImpl::SendMessages(CNode* pto)
4916         // Message: inventory
4917         //
4918         std::vector<CInv> vInv;
4919+        PeerRef peer = GetPeerRef(pto->GetId());

689@@ -688,12 +690,12 @@ class CNode
690         }
691     }
692 
693-    void PushTxInventory(const uint256& hash)
694+    void PushTxInventory(const uint256& hash, const bool flood)

69@@ -69,6 +70,6 @@ class PeerManager : public CValidationInterface, public NetEventsInterface
70 };
71 
72 /** Relay transaction to every node */
73-void RelayTransaction(const uint256& txid, const uint256& wtxid, const CConnman& connman) EXCLUSIVE_LOCKS_REQUIRED(cs_main);
74+void RelayTransaction(const uint256& txid, const uint256& wtxid, const CConnman& connman, bool flood) EXCLUSIVE_LOCKS_REQUIRED(cs_main);

0@@ -0,0 +1,335 @@
1+#include <minisketch/include/minisketch.h>