zmqpublishnotifier.cpp raw
1 // Copyright (c) 2015-2022 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 <zmq/zmqpublishnotifier.h>
6
7 #include <chain.h>
8 #include <chainparams.h>
9 #include <crypto/common.h>
10 #include <kernel/cs_main.h>
11 #include <logging.h>
12 #include <netaddress.h>
13 #include <netbase.h>
14 #include <node/blockstorage.h>
15 #include <primitives/block.h>
16 #include <primitives/transaction.h>
17 #include <rpc/server.h>
18 #include <serialize.h>
19 #include <streams.h>
20 #include <sync.h>
21 #include <uint256.h>
22 #include <zmq/zmqutil.h>
23
24 #include <zmq.h>
25
26 #include <cassert>
27 #include <cstdarg>
28 #include <cstddef>
29 #include <cstdint>
30 #include <cstring>
31 #include <map>
32 #include <optional>
33 #include <string>
34 #include <utility>
35 #include <vector>
36
37 namespace Consensus {
38 struct Params;
39 }
40
41 static std::multimap<std::string, CZMQAbstractPublishNotifier*> mapPublishNotifiers;
42
43 static const char *MSG_HASHBLOCK = "hashblock";
44 static const char *MSG_HASHTX = "hashtx";
45 static const char *MSG_HASHWALLETTXMEMPOOL = "hashwallettx-mempool";
46 static const char *MSG_HASHWALLETTXBLOCK = "hashwallettx-block";
47 static const char *MSG_RAWBLOCK = "rawblock";
48 static const char *MSG_RAWTX = "rawtx";
49 static const char *MSG_RAWWALLETTXMEMPOOL = "rawwallettx-mempool";
50 static const char *MSG_RAWWALLETTXBLOCK = "rawwallettx-block";
51 static const char *MSG_SEQUENCE = "sequence";
52
53 // Internal function to send multipart message
54 static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
55 {
56 va_list args;
57 va_start(args, size);
58
59 while (1)
60 {
61 zmq_msg_t msg;
62
63 int rc = zmq_msg_init_size(&msg, size);
64 if (rc != 0)
65 {
66 zmqError("Unable to initialize ZMQ msg");
67 va_end(args);
68 return -1;
69 }
70
71 void *buf = zmq_msg_data(&msg);
72 memcpy(buf, data, size);
73
74 data = va_arg(args, const void*);
75
76 rc = zmq_msg_send(&msg, sock, data ? ZMQ_SNDMORE : 0);
77 if (rc == -1)
78 {
79 zmqError("Unable to send ZMQ msg");
80 zmq_msg_close(&msg);
81 va_end(args);
82 return -1;
83 }
84
85 zmq_msg_close(&msg);
86
87 if (!data)
88 break;
89
90 size = va_arg(args, size_t);
91 }
92 va_end(args);
93 return 0;
94 }
95
96 static bool IsZMQAddressIPV6(const std::string &zmq_address)
97 {
98 const std::string tcp_prefix = "tcp://";
99 const size_t tcp_index = zmq_address.rfind(tcp_prefix);
100 const size_t colon_index = zmq_address.rfind(':');
101 if (tcp_index == 0 && colon_index != std::string::npos) {
102 const std::string ip = zmq_address.substr(tcp_prefix.length(), colon_index - tcp_prefix.length());
103 const std::optional<CNetAddr> addr{LookupHost(ip, false)};
104 if (addr.has_value() && addr.value().IsIPv6()) return true;
105 }
106 return false;
107 }
108
109 bool CZMQAbstractPublishNotifier::Initialize(void *pcontext)
110 {
111 assert(!psocket);
112
113 // check if address is being used by other publish notifier
114 std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator i = mapPublishNotifiers.find(address);
115
116 if (i==mapPublishNotifiers.end())
117 {
118 psocket = zmq_socket(pcontext, ZMQ_PUB);
119 if (!psocket)
120 {
121 zmqError("Failed to create socket");
122 return false;
123 }
124
125 LogDebug(BCLog::ZMQ, "Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark);
126
127 int rc = zmq_setsockopt(psocket, ZMQ_SNDHWM, &outbound_message_high_water_mark, sizeof(outbound_message_high_water_mark));
128 if (rc != 0)
129 {
130 zmqError("Failed to set outbound message high water mark");
131 zmq_close(psocket);
132 return false;
133 }
134
135 const int so_keepalive_option {1};
136 rc = zmq_setsockopt(psocket, ZMQ_TCP_KEEPALIVE, &so_keepalive_option, sizeof(so_keepalive_option));
137 if (rc != 0) {
138 zmqError("Failed to set SO_KEEPALIVE");
139 zmq_close(psocket);
140 return false;
141 }
142
143 // On some systems (e.g. OpenBSD) the ZMQ_IPV6 must not be enabled, if the address to bind isn't IPv6
144 const int enable_ipv6 { IsZMQAddressIPV6(address) ? 1 : 0};
145 rc = zmq_setsockopt(psocket, ZMQ_IPV6, &enable_ipv6, sizeof(enable_ipv6));
146 if (rc != 0) {
147 zmqError("Failed to set ZMQ_IPV6");
148 zmq_close(psocket);
149 return false;
150 }
151
152 rc = zmq_bind(psocket, address.c_str());
153 if (rc != 0)
154 {
155 zmqError("Failed to bind address");
156 zmq_close(psocket);
157 return false;
158 }
159
160 // register this notifier for the address, so it can be reused for other publish notifier
161 mapPublishNotifiers.insert(std::make_pair(address, this));
162 return true;
163 }
164 else
165 {
166 LogDebug(BCLog::ZMQ, "Reusing socket for address %s\n", address);
167 LogDebug(BCLog::ZMQ, "Outbound message high water mark for %s at %s is %d\n", type, address, outbound_message_high_water_mark);
168
169 psocket = i->second->psocket;
170 mapPublishNotifiers.insert(std::make_pair(address, this));
171
172 return true;
173 }
174 }
175
176 void CZMQAbstractPublishNotifier::Shutdown()
177 {
178 // Early return if Initialize was not called
179 if (!psocket) return;
180
181 int count = mapPublishNotifiers.count(address);
182
183 // remove this notifier from the list of publishers using this address
184 typedef std::multimap<std::string, CZMQAbstractPublishNotifier*>::iterator iterator;
185 std::pair<iterator, iterator> iterpair = mapPublishNotifiers.equal_range(address);
186
187 for (iterator it = iterpair.first; it != iterpair.second; ++it)
188 {
189 if (it->second==this)
190 {
191 mapPublishNotifiers.erase(it);
192 break;
193 }
194 }
195
196 if (count == 1)
197 {
198 LogDebug(BCLog::ZMQ, "Close socket at address %s\n", address);
199 int linger = 0;
200 zmq_setsockopt(psocket, ZMQ_LINGER, &linger, sizeof(linger));
201 zmq_close(psocket);
202 }
203
204 psocket = nullptr;
205 }
206
207 bool CZMQAbstractPublishNotifier::SendZmqMessage(const char *command, const void* data, size_t size)
208 {
209 assert(psocket);
210
211 /* send three parts, command & data & a LE 4byte sequence number */
212 unsigned char msgseq[sizeof(uint32_t)];
213 WriteLE32(msgseq, nSequence);
214 int rc = zmq_send_multipart(psocket, command, strlen(command), data, size, msgseq, (size_t)sizeof(uint32_t), nullptr);
215 if (rc == -1)
216 return false;
217
218 /* increment memory only sequence number after sending */
219 nSequence++;
220
221 return true;
222 }
223
224 bool CZMQPublishHashBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
225 {
226 uint256 hash = pindex->GetBlockHash();
227 LogDebug(BCLog::ZMQ, "Publish hashblock %s to %s\n", hash.GetHex(), this->address);
228 uint8_t data[32];
229 for (unsigned int i = 0; i < 32; i++) {
230 data[31 - i] = hash.begin()[i];
231 }
232 return SendZmqMessage(MSG_HASHBLOCK, data, 32);
233 }
234
235 bool CZMQPublishHashTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
236 {
237 uint256 hash = transaction.GetHash();
238 LogDebug(BCLog::ZMQ, "Publish hashtx %s to %s\n", hash.GetHex(), this->address);
239 uint8_t data[32];
240 for (unsigned int i = 0; i < 32; i++) {
241 data[31 - i] = hash.begin()[i];
242 }
243 return SendZmqMessage(MSG_HASHTX, data, 32);
244 }
245
246 bool CZMQPublishHashWalletTransactionNotifier::NotifyWalletTransaction(const CTransaction &transaction, const uint256 &hashBlock){
247 uint256 hash = transaction.GetHash();
248 LogDebug(BCLog::ZMQ, "Publish hashwallettx %s to %s\n", hash.GetHex(), this->address);
249 uint8_t data[32];
250 for (unsigned int i = 0; i < 32; i++)
251 data[31 - i] = hash.begin()[i];
252
253 const char *command;
254
255 if (!hashBlock.IsNull())
256 command = MSG_HASHWALLETTXBLOCK;
257 else
258 command = MSG_HASHWALLETTXMEMPOOL;
259
260 return SendZmqMessage(command, data, 32);
261 }
262
263 bool CZMQPublishRawBlockNotifier::NotifyBlock(const CBlockIndex *pindex)
264 {
265 LogDebug(BCLog::ZMQ, "Publish rawblock %s to %s\n", pindex->GetBlockHash().GetHex(), this->address);
266
267 std::vector<uint8_t> block{};
268 if (!m_get_block_by_index(block, *pindex)) {
269 LogDebug(BCLog::ZMQ, "Error: Can't read block %s from disk\n", pindex->GetBlockHash().ToString());
270 return false;
271 }
272
273 return SendZmqMessage(MSG_RAWBLOCK, block.data(), block.size());
274 }
275
276 bool CZMQPublishRawTransactionNotifier::NotifyTransaction(const CTransaction &transaction)
277 {
278 uint256 hash = transaction.GetHash();
279 LogDebug(BCLog::ZMQ, "Publish rawtx %s to %s\n", hash.GetHex(), this->address);
280 DataStream ss;
281 ss << TX_WITH_WITNESS(transaction);
282 return SendZmqMessage(MSG_RAWTX, &(*ss.begin()), ss.size());
283 }
284
285 // Helper function to send a 'sequence' topic message with the following structure:
286 // <32-byte hash> | <1-byte label> | <8-byte LE sequence> (optional)
287 static bool SendSequenceMsg(CZMQAbstractPublishNotifier& notifier, uint256 hash, char label, std::optional<uint64_t> sequence = {})
288 {
289 unsigned char data[sizeof(hash) + sizeof(label) + sizeof(uint64_t)];
290 for (unsigned int i = 0; i < sizeof(hash); ++i) {
291 data[sizeof(hash) - 1 - i] = hash.begin()[i];
292 }
293 data[sizeof(hash)] = label;
294 if (sequence) WriteLE64(data + sizeof(hash) + sizeof(label), *sequence);
295 return notifier.SendZmqMessage(MSG_SEQUENCE, data, sequence ? sizeof(data) : sizeof(hash) + sizeof(label));
296 }
297
298 bool CZMQPublishSequenceNotifier::NotifyBlockConnect(const CBlockIndex *pindex)
299 {
300 uint256 hash = pindex->GetBlockHash();
301 LogDebug(BCLog::ZMQ, "Publish sequence block connect %s to %s\n", hash.GetHex(), this->address);
302 return SendSequenceMsg(*this, hash, /* Block (C)onnect */ 'C');
303 }
304
305 bool CZMQPublishSequenceNotifier::NotifyBlockDisconnect(const CBlockIndex *pindex)
306 {
307 uint256 hash = pindex->GetBlockHash();
308 LogDebug(BCLog::ZMQ, "Publish sequence block disconnect %s to %s\n", hash.GetHex(), this->address);
309 return SendSequenceMsg(*this, hash, /* Block (D)isconnect */ 'D');
310 }
311
312 bool CZMQPublishSequenceNotifier::NotifyTransactionAcceptance(const CTransaction &transaction, uint64_t mempool_sequence)
313 {
314 uint256 hash = transaction.GetHash();
315 LogDebug(BCLog::ZMQ, "Publish hashtx mempool acceptance %s to %s\n", hash.GetHex(), this->address);
316 return SendSequenceMsg(*this, hash, /* Mempool (A)cceptance */ 'A', mempool_sequence);
317 }
318
319 bool CZMQPublishSequenceNotifier::NotifyTransactionRemoval(const CTransaction &transaction, uint64_t mempool_sequence)
320 {
321 uint256 hash = transaction.GetHash();
322 LogDebug(BCLog::ZMQ, "Publish hashtx mempool removal %s to %s\n", hash.GetHex(), this->address);
323 return SendSequenceMsg(*this, hash, /* Mempool (R)emoval */ 'R', mempool_sequence);
324 }
325
326 bool CZMQPublishRawWalletTransactionNotifier::NotifyWalletTransaction(const CTransaction &transaction, const uint256 &hashBlock){
327 uint256 hash = transaction.GetHash();
328 LogDebug(BCLog::ZMQ, "Publish rawwallettx %s to %s\n", hash.GetHex(), this->address);
329 DataStream ss;
330 ss << TX_WITH_WITNESS(transaction);
331
332 const char *command;
333
334 if (!hashBlock.IsNull())
335 command = MSG_RAWWALLETTXBLOCK;
336 else
337 command = MSG_RAWWALLETTXMEMPOOL;
338
339 return SendZmqMessage(command, &(*ss.begin()), ss.size());
340 }
341