Skip to content

Commit a057c55

Browse files
Add Electrum protocol version negotiation (#62)
* Add Electrum protocol version negotiation Moves server.version handling out of _ping_loop into a dedicated `handshaking` state in the monitor loop state machine. The handshake runs synchronously after recv/write/notify threads are started but before the ping thread is launched, ensuring the protocol ordering constraint (server.version before server.ping, required since protocol 1.2) is enforced by the architecture, not just by call order within a thread. Changes: - New `handshaking` state between `creating_threads` and `execute_recreation_callback`; ping thread only starts after a successful handshake - Protocol range raised from ["1.2", "1.4"] to ["1.3", "1.4"]; blockchain.block.header, used throughout spectrum.py, was added in 1.3 - Negotiated server software version and protocol version stored as `server_software_version` and `negotiated_protocol` on the socket instance and logged at INFO level - All failure paths (RPCError, timeout, unexpected response format, sub-minimum negotiated version) trigger reconnect via `broken_killing_threads` - Fix logger.exception() missing argument in _create_threads Tested against libbitcoin, ElectrumX, Fulcrum and electrs-esplora. * test: update Electrum socket state expectations --------- Co-authored-by: al-munazzim <al-munazzim@users.noreply.github.com>
1 parent e75ed88 commit a057c55

2 files changed

Lines changed: 84 additions & 10 deletions

File tree

‎src/cryptoadvance/spectrum/elsock.py‎

Lines changed: 61 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,14 @@
1717
logger = logging.getLogger(__name__)
1818

1919

20+
def _ver(version_str: str) -> tuple:
21+
"""Parse a "X.Y" version string into a comparable tuple."""
22+
try:
23+
return tuple(int(x) for x in str(version_str).split("."))
24+
except ValueError:
25+
return (0,)
26+
27+
2028
class ElSockTimeoutException(Exception):
2129
"""Called in different contexts where a timeout is relevant"""
2230

@@ -54,6 +62,10 @@ class ElectrumSocket:
5462
sleep_recv_loop = 0.01 # seconds , the shorter the better performance but 0.001 might be much worse
5563
sleep_write_loop = 0.01 # seconds , the shorter the better performance but 0.001 might be much worse
5664
socket_timeout = 10 # seconds for self._socket.recv(2048) (won't show up in the logs)
65+
66+
CLIENT_NAME = "Spectrum/1.0"
67+
PROTOCOL_MIN = "1.3" # blockchain.block.header was added in 1.3
68+
PROTOCOL_MAX = "1.4"
5769
# fmt: on
5870

5971
def __init__(
@@ -108,6 +120,8 @@ def __init__(
108120
self._requests = []
109121
self._notifications = []
110122
self._wanted_status = "ok" # "ok" or "down"
123+
self.server_software_version = None # set during version negotiation, e.g. "/libbitcoin:4.0.0/"
124+
self.negotiated_protocol = None # set during version negotiation, e.g. "1.4"
111125
# The monitor-thread will create the other threads
112126
self._monitor_thread = create_and_start_bg_thread(self._monitor_loop)
113127
while not (self.status == "ok" or self.status.startswith("broken_")):
@@ -231,11 +245,10 @@ def _create_threads(self) -> bool:
231245
try:
232246
self._recv_thread = create_and_start_bg_thread(self.recv_loop)
233247
self._write_thread = create_and_start_bg_thread(self._write_loop)
234-
self._ping_thread = create_and_start_bg_thread(self._ping_loop)
235248
self._notify_thread = create_and_start_bg_thread(self._notify_loop)
236249
return True
237250
except Exception as e:
238-
logger.exception()
251+
logger.exception(e)
239252
return False
240253

241254
def is_socket_closed(self) -> bool:
@@ -257,14 +270,15 @@ def _monitor_loop(self):
257270
If the ping thread is not alive, the socket connection and threads will be recreated via walking through
258271
this state-machine:
259272
260-
[![](https://mermaid.ink/img/pako:eNqdkj9vAyEMxb_KyWOVWzre0KkZM3VrqU4OOAmCwxFn-kdRvnu5I0mVKmEok-H9eDbiHUCzIehgFBR6triNOLQfjyo0eb09vDdt-9ToSCg2bPuRtSMpIrtZW0d2FHpnvZ8I2WXWjAW5rd23rPCGP0OBpuq-xZ_Da_BquFvkaYDacP9oUH13haUv0kmoj1Rkzt3R-zVqV25VgNmAHSxgoDigNfmPD9MtBbKjgRR0uTS0weRFgQrHjKa9ySlYGiscodugH2kBmIRfvoOGTmKiM3SKyoXaY3hl_t3TbLIq4ZozdvwB373ZuA?type=png)](https://mermaid-js.github.io/mermaid-live-editor/edit#pako:eNqdkj9vAyEMxb_KyWOVWzre0KkZM3VrqU4OOAmCwxFn-kdRvnu5I0mVKmEok-H9eDbiHUCzIehgFBR6triNOLQfjyo0eb09vDdt-9ToSCg2bPuRtSMpIrtZW0d2FHpnvZ8I2WXWjAW5rd23rPCGP0OBpuq-xZ_Da_BquFvkaYDacP9oUH13haUv0kmoj1Rkzt3R-zVqV25VgNmAHSxgoDigNfmPD9MtBbKjgRR0uTS0weRFgQrHjKa9ySlYGiscodugH2kBmIRfvoOGTmKiM3SKyoXaY3hl_t3TbLIq4ZozdvwB373ZuA)
273+
[![](https://mermaid.ink/img/pako:eNqdkk0OwiAQha9CWBq7cenClbdQ04wwFsJfQqmaNL27FFxYQzGRFXnvm5lHmJEyx5HuCe0DBDxK6DyY5r47WxLPaXMhTXMgzCMEabu2d0xhyKZTybt6p9C2Smo9E0FElvcZKXvrLSs8dw-bofm23uJLXIKLcCXyHaAW7o8BP99dJAVY3gtQUc_Uh5AAfCIbArYec72L4UDrKzBVLqh9VaVZqnWKbgk16A1IHhdmpEGgSavD8QaDDnSaXhlMzyQ?type=png)](https://mermaid-js.github.io/mermaid-live-editor/edit#pako:eNqdkk0OwiAQha9CWBq7cenClbdQ04wwFsJfQqmaNL27FFxYQzGRFXnvm5lHmJEyx5HuCe0DBDxK6DyY5r47WxLPaXMhTXMgzCMEabu2d0xhyKZTybt6p9C2Smo9E0FElvcZKXvrLSs8dw-bofm23uJLXIKLcCXyHaAW7o8BP99dJAVY3gtQUc_Uh5AAfCIbArYec72L4UDrKzBVLqh9VaVZqnWKbgk16A1IHhdmpEGgSavD8QaDDnSaXhlMzyQ)
261274
262275
The states are stored in the `ElectrumSocket.state` property. The Constructor of the `ElectrumSocket` is hardly doing more than just setting up the `_monitor_thread` which is an endless loop going through these states:
263276
* `creating_sockets` will create the sockets and pass to `creating_threads` or to `broken_creating_sockets` if that fails
264277
* `broken_creating_sockets` will try to create the socket and sleep for some time if that fails (and endlessly try to do that)
265-
* `creating_threads` will create the write/recv/ping/notify threads and start them
278+
* `creating_threads` will create the write/recv/notify threads (not ping yet) and start them
279+
* `handshaking` calls server.version synchronously. On success it stores the negotiated protocol version and starts the ping thread. On failure it transitions to `broken_killing_threads`.
266280
* `execute_recreation_callback` will call that callback after setting the status to `ok`
267-
* the `ok` state will now simply check the other thready and if one of them is no longer alive (probably the ping-thread as he will exit if ping fails for 4 times) it will transition to `broken_killing_threads`
281+
* the `ok` state will now simply check the other threads and if one of them is no longer alive (probably the ping-thread as he will exit if ping fails for 4 times) it will transition to `broken_killing_threads`
268282
* `broken_killing_threads` will set `self.running` to false and wait for the threads to terminate. Especially the `recv` thread might not terminate until he get internet connection (again). This might take forever. If all threads are terminated, it will transition to `creating_socket`
269283
270284
"""
@@ -296,7 +310,47 @@ def _monitor_loop(self):
296310
if not self._create_threads():
297311
time.sleep(10)
298312
continue
299-
self.status = "execute_recreation_callback"
313+
self.status = "handshaking"
314+
315+
if self.status == "handshaking":
316+
self.server_software_version = None
317+
self.negotiated_protocol = None
318+
_handshake_ok = False
319+
try:
320+
result = self.call(
321+
"server.version",
322+
[self.CLIENT_NAME, [self.PROTOCOL_MIN, self.PROTOCOL_MAX]],
323+
)
324+
if isinstance(result, list) and len(result) == 2:
325+
self.server_software_version, self.negotiated_protocol = result
326+
logger.info(
327+
f"Electrum handshake complete: server={self.server_software_version!r} "
328+
f"protocol={self.negotiated_protocol!r}"
329+
)
330+
if _ver(self.negotiated_protocol) < _ver(self.PROTOCOL_MIN):
331+
logger.error(
332+
f"Negotiated protocol {self.negotiated_protocol!r} is below "
333+
f"required minimum {self.PROTOCOL_MIN!r}. Reconnecting."
334+
)
335+
else:
336+
_handshake_ok = True
337+
else:
338+
logger.error(f"Unexpected server.version response: {result!r}. Reconnecting.")
339+
except RPCError as e:
340+
logger.error(
341+
f"server.version rejected (code={e.code}): {e.message!r}. "
342+
f"Server may not support protocol range "
343+
f"[{self.PROTOCOL_MIN}, {self.PROTOCOL_MAX}]. Reconnecting."
344+
)
345+
except ElSockTimeoutException:
346+
logger.error("server.version timed out. Reconnecting.")
347+
except Exception as e:
348+
logger.error(f"server.version failed: {e}. Reconnecting.")
349+
if not _handshake_ok:
350+
self.status = "broken_killing_threads"
351+
else:
352+
self._ping_thread = create_and_start_bg_thread(self._ping_loop)
353+
self.status = "execute_recreation_callback"
300354

301355
if self.status == "execute_recreation_callback":
302356
# set the new status here before we call the callback
@@ -502,7 +556,7 @@ def _ping_loop(self):
502556
try:
503557
self.ping()
504558
tries = 0
505-
except ElSockTimeoutException as e:
559+
except ElSockTimeoutException:
506560
tries = tries + 1
507561
logger.error(
508562
f"Timeout in ping-loop ({tries}th time, next try in {self.sleep_ping_loop} seconds if threshold not met"

‎tests/integration/elsock_test.py‎

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,13 @@ def callback(something):
5454
)
5555
assert (
5656
caplog.text.count(
57-
"ElectrumSocket Status changed from creating_threads to execute_recreation_callback"
57+
"ElectrumSocket Status changed from creating_threads to handshaking"
58+
)
59+
== 1
60+
)
61+
assert (
62+
caplog.text.count(
63+
"ElectrumSocket Status changed from handshaking to execute_recreation_callback"
5864
)
5965
== 1
6066
)
@@ -74,14 +80,22 @@ def callback(something):
7480
f"...................................... timer: {i} seconds passed (elsock.is_socket_closed() returns {elsock.is_socket_closed()})"
7581
)
7682
time.sleep(1)
83+
for i in range(0, 20):
84+
if elsock.status == "ok":
85+
break
86+
logger.info(
87+
f"...................................... waiting for socket recovery: {i} seconds passed (status={elsock.status})"
88+
)
89+
time.sleep(1)
90+
assert elsock.status == "ok"
7791
logger.info(
7892
f"{datetime.now()}========================The socket connection should now work properly again================================"
7993
)
8094
logger.info(elsock._socket)
8195
ts = elsock.ping()
8296
logger.info(f"second working ping in {ts} ms")
8397
assert ts < 10
84-
assert caplog.text.count("ElectrumSocket Status changed") == 9
98+
assert caplog.text.count("ElectrumSocket Status changed") == 11
8599

86100
assert (
87101
caplog.text.count(
@@ -103,7 +117,13 @@ def callback(something):
103117
)
104118
assert (
105119
caplog.text.count(
106-
"ElectrumSocket Status changed from creating_threads to execute_recreation_callback"
120+
"ElectrumSocket Status changed from creating_threads to handshaking"
121+
)
122+
== 2
123+
)
124+
assert (
125+
caplog.text.count(
126+
"ElectrumSocket Status changed from handshaking to execute_recreation_callback"
107127
)
108128
== 2
109129
)

0 commit comments

Comments
 (0)