mempoolbridge.cpp raw
1 // Copyright (c) 2026 The Limenka developers
2 // Distributed under the MIT software license, see the accompanying
3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
4
5 #include <node/mempoolbridge.h>
6
7 #include <chainparams.h>
8 #include <hash.h>
9 #include <kernel/cs_main.h>
10 #include <logging.h>
11 #include <net.h>
12 #include <netbase.h>
13 #include <node/context.h>
14 #include <random.h>
15 #include <serialize.h>
16 #include <streams.h>
17 #include <tinyformat.h>
18 #include <txmempool.h>
19 #include <util/time.h>
20 #include <validation.h>
21 #include <validationinterface.h>
22
23 #include <boost/asio.hpp>
24
25 #include <algorithm>
26 #include <array>
27 #include <chrono>
28 #include <cstring>
29 #include <deque>
30 #include <vector>
31
32 namespace node {
33
34 namespace {
35 // ---------------------------------------------------------------------------
36 // Foreign network table: message start + default port per chain.
37 // ---------------------------------------------------------------------------
38 const std::vector<BridgeNetwork> KNOWN_NETWORKS{
39 {"mainnet", {0xf9, 0xbe, 0xb4, 0xd9}, 8333, {}},
40 {"testnet3", {0x07, 0x09, 0x11, 0x0b}, 18333, {}},
41 {"testnet4", {0x1c, 0x16, 0x3f, 0x28}, 48333, {}},
42 {"signet", {0x0a, 0x03, 0xcf, 0x40}, 38333, {}},
43 {"regtest", {0xfa, 0xbf, 0xb5, 0xda}, 18444, {}},
44 // Limenka's own fork network: lets the bridge run against a second
45 // limenka node (functional tests, cross-node mesh).
46 {"fork", {0xfa, 0xbf, 0xd5, 0xc6}, 19444, {}},
47 };
48
49 constexpr std::chrono::seconds RECONNECT_INTERVAL{5};
50 constexpr size_t SEEN_MAX{100000};
51 constexpr size_t TX_CACHE_MAX{10000};
52 constexpr size_t MAX_INV_PER_MESSAGE{50000};
53 constexpr size_t INVENTORY_BROADCAST_MAX_LOCAL{1000};
54
55 // Flood guard per peer: at most this many transactions accepted per window.
56 constexpr size_t MAX_TX_PER_WINDOW{1000};
57 constexpr std::chrono::seconds TX_WINDOW{10};
58
59 } // namespace
60
61 // ---------------------------------------------------------------------------
62 // Peer: one connection to one foreign node. Reads and writes are async on
63 // the bridge io_context; all socket operations are serialized through it.
64 // ---------------------------------------------------------------------------
65 class MempoolBridge::Peer final : public std::enable_shared_from_this<Peer>
66 {
67 public:
68 Peer(MempoolBridge& bridge, const BridgeNetwork& net, const CService& remote)
69 : m_bridge{bridge}, m_net{net}, m_remote{remote},
70 m_socket{*bridge.m_io}
71 {
72 }
73
74 void Start()
75 {
76 auto self{shared_from_this()};
77 boost::system::error_code ec;
78 const auto ep = ResolveEndpoint(ec);
79 if (ec) { m_dead = true; return; }
80 m_socket.async_connect(ep, [self](const boost::system::error_code& err) {
81 if (err) {
82 LogPrintf("mempoolbridge: connect %s failed: %s\n",
83 self->m_remote.ToStringAddrPort(), err.message());
84 self->m_dead = true;
85 return;
86 }
87 LogPrintf("mempoolbridge: tcp connected to %s\n", self->m_remote.ToStringAddrPort());
88 self->SendVersion();
89 self->DoReadHeader();
90 });
91 }
92
93 bool Dead() const { return m_dead; }
94 bool Ready() const { return m_ready; }
95 const CService& Remote() const { return m_remote; }
96 const std::string& NetworkName() const { return m_net.name; }
97
98 void SendInv(const CTransactionRef& tx)
99 {
100 // Announce by wtxid (MSG_WTX, BIP339): peers serve witness
101 // transactions only when requested by wtxid - a txid request gets a
102 // witness-stripped tx, which would fail limenka validation.
103 DataStream ss;
104 ss << CompactSizeWriter(1) << CInv{MSG_WTX, tx->GetWitnessHash()};
105 SendMessage(NetMsgType::INV, ss);
106 }
107
108 void SendTx(const CTransactionRef& tx)
109 {
110 DataStream ss;
111 ss << TX_WITH_WITNESS(*tx);
112 SendMessage(NetMsgType::TX, ss);
113 }
114
115 void Disconnect()
116 {
117 boost::asio::post(*m_bridge.m_io, [self = shared_from_this()]() {
118 boost::system::error_code ec;
119 self->m_socket.close(ec);
120 self->m_dead = true;
121 });
122 }
123
124 private:
125 MempoolBridge& m_bridge;
126 const BridgeNetwork& m_net;
127 CService m_remote;
128 boost::asio::ip::tcp::socket m_socket;
129
130 bool m_dead{false};
131 bool m_ready{false};
132 bool m_version_sent{false};
133
134 std::array<uint8_t, CMessageHeader::HEADER_SIZE> m_hdr{};
135 std::vector<uint8_t> m_payload;
136 std::deque<std::vector<uint8_t>> m_send_q;
137 bool m_writing{false};
138
139 // Flood guard.
140 int64_t m_window_start{0};
141 size_t m_window_tx{0};
142
143 boost::asio::ip::tcp::endpoint ResolveEndpoint(boost::system::error_code& ec)
144 {
145 if (m_remote.IsIPv4()) {
146 struct in_addr a4;
147 m_remote.GetInAddr(&a4);
148 boost::asio::ip::address_v4::bytes_type b{};
149 std::memcpy(b.data(), &a4, 4);
150 return {boost::asio::ip::address_v4{b}, m_remote.GetPort()};
151 }
152 struct in6_addr a6;
153 m_remote.GetIn6Addr(&a6);
154 boost::asio::ip::address_v6::bytes_type b{};
155 std::memcpy(b.data(), &a6, 16);
156 return {boost::asio::ip::address_v6{b}, m_remote.GetPort()};
157 }
158
159 void SendVersion()
160 {
161 if (m_version_sent) return;
162 m_version_sent = true;
163
164 uint64_t nonce;
165 GetRandBytes({reinterpret_cast<unsigned char*>(&nonce), sizeof(nonce)});
166
167 DataStream ss;
168 ss << PROTOCOL_VERSION;
169 ss << uint64_t{NODE_WITNESS};
170 ss << GetTime<std::chrono::seconds>().count();
171 SerAddr(ss, m_remote); // their address
172 SerAddr(ss, m_remote); // our address (same socket target)
173 ss << nonce;
174 ss << std::string{"/limenka-bridge:1.0.0/"};
175 ss << int32_t{0}; // start height: we do not sync blocks
176 ss << uint8_t{1}; // relay: yes, we want their mempool
177 SendMessage(NetMsgType::VERSION, ss);
178 }
179
180 static void SerAddr(DataStream& ss, const CService& svc)
181 {
182 ss << uint64_t{NODE_NONE};
183 std::array<uint8_t, 16> ip{};
184 if (svc.IsIPv4()) {
185 ip[10] = 0xff;
186 ip[11] = 0xff;
187 struct in_addr a4;
188 svc.GetInAddr(&a4);
189 std::memcpy(ip.data() + 12, &a4, 4);
190 } else {
191 struct in6_addr a6;
192 svc.GetIn6Addr(&a6);
193 std::memcpy(ip.data(), &a6, 16);
194 }
195 ss << ip;
196 ss << uint16_t{svc.GetPort()};
197 }
198
199 void SendMessage(const char* type, DataStream& payload)
200 {
201 CMessageHeader hdr{m_net.magic, type, static_cast<uint32_t>(payload.size())};
202 const uint256 checksum = Hash(MakeByteSpan(payload));
203 std::memcpy(hdr.pchChecksum, checksum.begin(), CMessageHeader::CHECKSUM_SIZE);
204
205 std::vector<uint8_t> out;
206 out.reserve(CMessageHeader::HEADER_SIZE + payload.size());
207 VectorWriter vw{out, 0, hdr};
208 vw.write(MakeByteSpan(payload));
209
210 auto self{shared_from_this()};
211 boost::asio::post(*m_bridge.m_io, [self, bytes = std::move(out)]() mutable {
212 if (self->m_dead) return;
213 self->m_send_q.push_back(std::move(bytes));
214 self->MaybeWrite();
215 });
216 }
217
218 void MaybeWrite()
219 {
220 if (m_writing || m_send_q.empty()) return;
221 m_writing = true;
222 auto self{shared_from_this()};
223 const auto& bytes = m_send_q.front();
224 boost::asio::async_write(m_socket, boost::asio::buffer(bytes),
225 [self](const boost::system::error_code& err, size_t) {
226 self->m_writing = false;
227 self->m_send_q.pop_front();
228 if (err) { self->m_dead = true; return; }
229 self->MaybeWrite();
230 });
231 }
232
233 void DoReadHeader()
234 {
235 auto self{shared_from_this()};
236 boost::asio::async_read(m_socket, boost::asio::buffer(m_hdr),
237 [self](const boost::system::error_code& err, size_t n) {
238 if (err) { self->m_dead = true; return; }
239 if (n != CMessageHeader::HEADER_SIZE) { self->m_dead = true; return; }
240 if (std::memcmp(self->m_hdr.data(), self->m_net.magic.data(), 4) != 0) {
241 // Wrong network magic on this connection.
242 self->m_dead = true;
243 return;
244 }
245 uint32_t len;
246 std::memcpy(&len, self->m_hdr.data() + CMessageHeader::MESSAGE_SIZE_OFFSET, 4);
247 if (len > MAX_PROTOCOL_MESSAGE_LENGTH) { self->m_dead = true; return; }
248 self->m_payload.resize(len);
249 if (len == 0) {
250 self->DispatchMessage();
251 self->DoReadHeader();
252 return;
253 }
254 boost::asio::async_read(self->m_socket, boost::asio::buffer(self->m_payload),
255 [self](const boost::system::error_code& err2, size_t n2) {
256 if (err2 || n2 != self->m_payload.size()) { self->m_dead = true; return; }
257 const uint256 checksum = Hash(MakeByteSpan(self->m_payload));
258 if (std::memcmp(self->m_hdr.data() + CMessageHeader::CHECKSUM_OFFSET,
259 checksum.begin(), CMessageHeader::CHECKSUM_SIZE) != 0) {
260 self->m_dead = true;
261 return;
262 }
263 self->DispatchMessage();
264 self->DoReadHeader();
265 });
266 });
267 }
268
269 void DispatchMessage()
270 {
271 std::string type{
272 reinterpret_cast<const char*>(m_hdr.data() + 4),
273 strnlen(reinterpret_cast<const char*>(m_hdr.data() + 4), CMessageHeader::MESSAGE_TYPE_SIZE)};
274 try {
275 HandleMessage(type);
276 } catch (const std::exception& e) {
277 LogDebug(BCLog::NET, "mempoolbridge: %s message from %s failed: %s\n",
278 type, m_remote.ToStringAddrPort(), e.what());
279 }
280 }
281
282 void HandleMessage(const std::string& type)
283 {
284 if (type == NetMsgType::VERSION) {
285 // BIP339: wtxidrelay must be sent between VERSION and VERACK.
286 // We announce and request by wtxid so peers serve us
287 // full-witness transactions.
288 DataStream empty;
289 SendMessage(NetMsgType::WTXIDRELAY, empty);
290 SendMessage(NetMsgType::VERACK, empty);
291 return;
292 }
293 if (type == NetMsgType::VERACK) {
294 m_ready = true;
295 LogPrintf("mempoolbridge: connected %s peer %s\n", m_net.name, m_remote.ToStringAddrPort());
296 return;
297 }
298 if (type == NetMsgType::PING) {
299 DataStream out;
300 if (m_payload.size() >= 8) {
301 DataStream ss{m_payload};
302 uint64_t nonce{0};
303 ss >> nonce;
304 out << nonce;
305 }
306 SendMessage(NetMsgType::PONG, out);
307 return;
308 }
309 if (type == NetMsgType::PONG) return;
310 if (type == NetMsgType::INV) {
311 HandleInv();
312 return;
313 }
314 if (type == NetMsgType::TX) {
315 HandleTx();
316 return;
317 }
318 if (type == NetMsgType::GETDATA) {
319 HandleGetData();
320 return;
321 }
322 if (type == NetMsgType::GETHEADERS || type == NetMsgType::HEADERS) return;
323 // Everything else (addr, getaddr, feefilter, sendheaders, sendcmpct,
324 // wtxidrelay, notfound, filter*, block announcements) is ignored:
325 // the bridge deals in transaction gossip only.
326 }
327
328 void HandleInv()
329 {
330 DataStream ss{m_payload};
331 const uint64_t count = ReadCompactSize(ss);
332 const size_t n = std::min<uint64_t>(count, MAX_INV_PER_MESSAGE);
333 std::vector<CInv> to_fetch;
334 to_fetch.reserve(std::min<size_t>(n, INVENTORY_BROADCAST_MAX_LOCAL));
335 for (size_t i = 0; i < n; ++i) {
336 CInv inv;
337 ss >> inv;
338 if (to_fetch.size() >= INVENTORY_BROADCAST_MAX_LOCAL) continue;
339 if (!(inv.type == MSG_TX || inv.type == MSG_WTX || inv.type == MSG_WITNESS_TX)) continue;
340 if (m_bridge.HasTx(inv.hash)) continue;
341 to_fetch.push_back(inv);
342 }
343 if (to_fetch.empty()) return;
344 DataStream out;
345 out << to_fetch;
346 SendMessage(NetMsgType::GETDATA, out);
347 }
348
349 void HandleTx()
350 {
351 const int64_t now = GetTime<std::chrono::seconds>().count();
352 if (now - m_window_start > TX_WINDOW.count()) {
353 m_window_start = now;
354 m_window_tx = 0;
355 }
356 if (++m_window_tx > MAX_TX_PER_WINDOW) return; // flood guard
357
358 DataStream ss{m_payload};
359 CMutableTransaction mtx;
360 ss >> TX_WITH_WITNESS(mtx);
361 if (ss.size() != 0) return; // trailing bytes - malformed
362 m_bridge.OnForeignTx(m_net.name, MakeTransactionRef(std::move(mtx)));
363 }
364
365 void HandleGetData()
366 {
367 DataStream ss{m_payload};
368 const uint64_t count = ReadCompactSize(ss);
369 const size_t n = std::min<uint64_t>(count, MAX_INV_PER_MESSAGE);
370 for (size_t i = 0; i < n; ++i) {
371 CInv inv;
372 ss >> inv;
373 if (inv.type != MSG_TX && inv.type != MSG_WTX && inv.type != MSG_WITNESS_TX) continue;
374 if (auto tx = m_bridge.GetTx(inv.hash)) SendTx(tx);
375 }
376 }
377 };
378
379 // ---------------------------------------------------------------------------
380 // Bridge
381 // ---------------------------------------------------------------------------
382
383 MempoolBridge::MempoolBridge(node::NodeContext& node) : m_node{node} {}
384
385 MempoolBridge::~MempoolBridge()
386 {
387 Stop();
388 }
389
390 std::optional<std::pair<std::string, CService>> MempoolBridge::ParseBridgePeer(const std::string& spec)
391 {
392 const size_t c1 = spec.find(':');
393 if (c1 == std::string::npos) return std::nullopt;
394 const std::string net_name = spec.substr(0, c1);
395 std::string hostport = spec.substr(c1 + 1);
396 if (hostport.empty()) return std::nullopt;
397
398 const auto it = std::find_if(KNOWN_NETWORKS.begin(), KNOWN_NETWORKS.end(),
399 [&](const BridgeNetwork& n) { return n.name == net_name; });
400 if (it == KNOWN_NETWORKS.end()) return std::nullopt;
401
402 // If no port and no brackets, append the network default.
403 const size_t c2 = hostport.rfind(':');
404 if (c2 == std::string::npos && hostport.find(']') == std::string::npos) {
405 hostport = strprintf("%s:%u", hostport, it->default_port);
406 }
407 const auto svc = Lookup(hostport, it->default_port, /*fAllowLookup=*/true);
408 if (!svc) return std::nullopt;
409 return std::make_pair(net_name, *svc);
410 }
411
412 bool MempoolBridge::Start(const std::vector<std::string>& peer_specs)
413 {
414 if (m_running.exchange(true)) return true;
415
416 {
417 LOCK(m_net_mutex);
418 m_networks = KNOWN_NETWORKS;
419 for (const auto& spec : peer_specs) {
420 const auto parsed = ParseBridgePeer(spec);
421 if (!parsed) {
422 LogPrintf("mempoolbridge: ignoring invalid -bridgepeer=%s\n", spec);
423 continue;
424 }
425 const auto it = std::find_if(m_networks.begin(), m_networks.end(),
426 [&](const BridgeNetwork& n) { return n.name == parsed->first; });
427 if (it != m_networks.end()) it->peers.push_back(parsed->second);
428 }
429 m_networks.erase(std::remove_if(m_networks.begin(), m_networks.end(),
430 [](const BridgeNetwork& n) { return n.peers.empty(); }),
431 m_networks.end());
432 }
433
434 if (m_networks.empty()) {
435 m_running = false;
436 return false;
437 }
438
439 m_io = std::make_unique<boost::asio::io_context>();
440 if (m_node.validation_signals) {
441 m_node.validation_signals->RegisterSharedValidationInterface(shared_from_this());
442 }
443 m_thread = std::thread([this]() {
444 while (m_running) {
445 MaintainConnections();
446 // Park on a timer so a failed connect batch cannot spin the
447 // loop (run() returns only when the timer fires or Stop()).
448 boost::asio::steady_timer timer{*m_io, RECONNECT_INTERVAL};
449 timer.async_wait([](const boost::system::error_code&) {});
450 try {
451 m_io->restart();
452 m_io->run();
453 } catch (const std::exception& e) {
454 LogPrintf("mempoolbridge: io loop error: %s\n", e.what());
455 break;
456 }
457 }
458 });
459
460 LogPrintf("mempoolbridge: started with %d network(s)\n", int(m_networks.size()));
461 return true;
462 }
463
464 void MempoolBridge::Stop()
465 {
466 if (!m_running.exchange(false)) return;
467 if (m_io) m_io->stop();
468 if (m_thread.joinable()) m_thread.join();
469 {
470 LOCK(m_net_mutex);
471 for (auto& peer : m_peers) peer->Disconnect();
472 m_peers.clear();
473 }
474 if (m_node.validation_signals) {
475 m_node.validation_signals->UnregisterSharedValidationInterface(shared_from_this());
476 }
477 LogPrintf("mempoolbridge: stopped\n");
478 }
479
480 void MempoolBridge::TransactionAddedToMempool(const NewMempoolTransactionInfo& info, uint64_t mempool_sequence)
481 {
482 if (!m_io) return;
483 const CTransactionRef tx = info.info.m_tx;
484 const uint256 wtxid = tx->GetWitnessHash();
485 {
486 // Serve this tx to foreign peers by txid or wtxid (BIP339).
487 LOCK(m_cache_mutex);
488 if (m_tx_cache.size() > TX_CACHE_MAX) m_tx_cache.clear();
489 m_tx_cache[tx->GetHash()] = tx;
490 m_tx_cache[wtxid] = tx;
491 }
492
493 std::vector<BridgeNetwork> nets;
494 {
495 LOCK(m_net_mutex);
496 nets = m_networks;
497 }
498 auto self{shared_from_this()};
499 for (const auto& net : nets) {
500 if (!NoteSeen(net.name, wtxid)) continue;
501 boost::asio::post(*m_io, [self, tx, wtxid, net]() { self->AnnounceToNetwork(net, tx, wtxid); });
502 }
503 }
504
505 void MempoolBridge::AnnounceToNetwork(const BridgeNetwork& net, const CTransactionRef& tx, const uint256& wtxid)
506 {
507 std::vector<std::shared_ptr<Peer>> peers;
508 {
509 LOCK(m_net_mutex);
510 for (const auto& peer : m_peers) {
511 if (peer->NetworkName() == net.name && peer->Ready() && !peer->Dead()) {
512 peers.push_back(peer);
513 }
514 }
515 }
516 if (peers.empty()) return;
517 LogDebug(BCLog::MEMPOOL, "mempoolbridge: announcing %s to %s (%d peer(s))\n",
518 tx->GetHash().ToString(), net.name, int(peers.size()));
519 for (const auto& peer : peers) peer->SendInv(tx);
520 }
521
522 bool MempoolBridge::NoteSeen(const std::string& network, const uint256& wtxid)
523 {
524 LOCK(m_seen_mutex);
525 auto& set = m_seen[network];
526 if (set.size() > SEEN_MAX) set.clear();
527 return set.insert(wtxid).second;
528 }
529
530 void MempoolBridge::OnForeignTx(const std::string& from_network, CTransactionRef tx)
531 {
532 const uint256 txid = tx->GetHash();
533 const uint256 wtxid = tx->GetWitnessHash();
534
535 // Never echo a transaction back to the network it arrived from, and
536 // remember it so the mempool announcement does not re-send it there.
537 NoteSeen(from_network, wtxid);
538
539 {
540 LOCK(m_cache_mutex);
541 if (m_tx_cache.size() > TX_CACHE_MAX) m_tx_cache.clear();
542 m_tx_cache[txid] = tx;
543 m_tx_cache[wtxid] = tx;
544 }
545
546 ChainstateManager* chainman = m_node.chainman.get();
547 if (!chainman || !chainman->ActiveChainstate().GetMempool()) return;
548
549 LogDebug(BCLog::MEMPOOL, "mempoolbridge: tx %s from %s, validating\n",
550 txid.ToString(), from_network);
551
552 {
553 LOCK(::cs_main);
554 const MempoolAcceptResult result = AcceptToMemoryPool(chainman->ActiveChainstate(), tx,
555 GetTime<std::chrono::seconds>().count(),
556 /*bypass_limits=*/false,
557 /*test_accept=*/false);
558 if (result.m_result_type == MempoolAcceptResult::ResultType::VALID) {
559 LogDebug(BCLog::MEMPOOL, "mempoolbridge: tx %s from %s accepted into limenka mempool\n",
560 txid.ToString(), from_network);
561 } else {
562 LogDebug(BCLog::MEMPOOL, "mempoolbridge: tx %s from %s rejected: %s\n",
563 txid.ToString(), from_network, result.m_state.ToString());
564 }
565 }
566 }
567
568 bool MempoolBridge::HasTx(const uint256& hash) const
569 {
570 {
571 LOCK(m_cache_mutex);
572 if (m_tx_cache.count(hash)) return true;
573 }
574 ChainstateManager* chainman = m_node.chainman.get();
575 if (!chainman || !chainman->ActiveChainstate().GetMempool()) return false;
576 CTxMemPool& pool = *chainman->ActiveChainstate().GetMempool();
577 LOCK(pool.cs);
578 return pool.GetEntry(Txid::FromUint256(hash)) != nullptr;
579 }
580
581 CTransactionRef MempoolBridge::GetTx(const uint256& hash) const
582 {
583 {
584 LOCK(m_cache_mutex);
585 const auto it = m_tx_cache.find(hash);
586 if (it != m_tx_cache.end()) return it->second;
587 }
588 ChainstateManager* chainman = m_node.chainman.get();
589 if (!chainman || !chainman->ActiveChainstate().GetMempool()) return nullptr;
590 CTxMemPool& pool = *chainman->ActiveChainstate().GetMempool();
591 LOCK(pool.cs);
592 const CTxMemPoolEntry* entry = pool.GetEntry(Txid::FromUint256(hash));
593 if (!entry) return nullptr;
594 return entry->GetSharedTx();
595 }
596
597 void MempoolBridge::MaintainConnections()
598 {
599 if (!m_io) return;
600 LOCK(m_net_mutex);
601 // Drop dead peers.
602 for (auto it = m_peers.begin(); it != m_peers.end();) {
603 if ((*it)->Dead()) {
604 it = m_peers.erase(it);
605 } else {
606 ++it;
607 }
608 }
609 // One live connection per configured peer address.
610 for (const auto& net : m_networks) {
611 for (const auto& addr : net.peers) {
612 const bool exists = std::any_of(m_peers.begin(), m_peers.end(),
613 [&](const std::shared_ptr<Peer>& p) {
614 return p->NetworkName() == net.name && p->Remote() == addr && !p->Dead();
615 });
616 if (exists) continue;
617 LogPrintf("mempoolbridge: connecting to %s %s\n", net.name, addr.ToStringAddrPort());
618 auto peer = std::make_shared<Peer>(*this, net, addr);
619 m_peers.push_back(peer);
620 peer->Start();
621 }
622 }
623 }
624
625 } // namespace node
626