Currently, ZMQ listeners cannot detect if they have missed a message. This PR adds a sequence number to each message (each message type has its own counter).
The sequence number is the last part of the multipart zmq message to not break the API, though, we could consider breaking the API in favor of moving the sequence number to the very beginning.
Todo:
I thought a subscriber can’t miss a message.
Edit:
But it can: http://stackoverflow.com/a/15821036. So what can a subscriber do if it detects a missing message?
I thought a subscriber can’t miss a message.
The way we are using ZMQ, the sender will never block. This is sensible as we don’t want to hang the application due to a slow client. But this means at some point one of the send queues will fill up and new messages will be discarded.
But it can: http://stackoverflow.com/a/15821036. So what can a subscriber do if it detects a missing message?
That’s up to the application. Either stop with a fatal error, or re-start/re-sync. An event application usually does some synchronization at start-up, then starts processing incremental changes. After getting out of sync it can repeat the process. The important thing is being able to detect it and thus act on it.
doc/release-notes.md as well as doc/zmq.md (there, don’t forget to specify the size and endianness of the sequence number), I think it is ready for merge otherwise.
47@@ -47,11 +48,18 @@ def run_test(self):
48 print "listen..."
49 msg = self.zmqSubSocket.recv_multipart()
50 topic = str(msg[0])
51+ assert_equal(topic, "hashtx")
52 body = msg[1]
53+ nseq = msg[2]
54+ msgSequence = struct.unpack('<I', str(msg[-1]))[-1]
str() here?
47@@ -47,11 +48,17 @@ def run_test(self):
48 print "listen..."
49 msg = self.zmqSubSocket.recv_multipart()
50 topic = msg[0]
51+ assert_equal(topic, "hashtx")
Nit: Add the b prefix, which is a noop in py2?
0b"hashtx"
19@@ -19,18 +20,21 @@
20 msg = zmqSubSocket.recv_multipart()
21 topic = str(msg[0])
22 body = msg[1]
23-
24+ sequence = "Unknown";
25+ if len(msg[-1]) == 4:
26+ msgSequence = struct.unpack('<I', str(msg[-1]))[-1]
str()?
14+static const char *MSG_RAWBLOCK = "rawblock";
15+static const char *MSG_RAWTX = "rawtx";
16+
17 // Internal function to send multipart message
18-static int zmq_send_multipart(void *sock, const void* data, size_t size, ...)
19+static int zmq_send_multipart_keepalive(void *sock, const void* data, size_t size, ...)
_keepalive?
_keepalive because it does not “close” the multipart message. All messaged are sent with ZMQ_SNDMORE and I wanted to increase the awareness of that by renaming the function.
The way we are using ZMQ, the sender will never block. This is sensible as we don’t want to hang the application due to a slow client.
Could we make that parameterizable? In certain cases I’d rather have message reliability than having the node to be guaranteed to be running in real time.
In certain cases I’d rather have message reliability than having the node to be guaranteed to be running in real time.
Well notifications can be lost, through zmq or otherwise, for example if bitcoind needs to be restarted, or a myriad of other circumstances not under your control. At least you can detect it now:
Your application needs an (application dependent) way to resync anyhow. This is better than pretending to guarantee something.