1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
"""A transient hub state on node sign-in must not crash the daemon.
`_login_with_retry` retries a 401 (the node key is not linked yet — a human has
to link it, and the daemon must stay alive so its key can be read). It used to
`raise` on every other status, so a **429** (the daemon's own retries hitting
the sign-in rate limit) or a **502/503** (the hub restarting during a deploy)
killed the process — systemd then crash-looped it, which is what "impossible de
démarrer le node" looked like after a reset left the node with a fresh, unlinked
key. Those transient statuses are now retried with a back-off that respects
`Retry-After`.
"""
import asyncio
import httpx
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node.config import Config, GroupConfig, HubConfig, KeystoreConfig, NodeConfig
from meshbay_node.daemon import NodeDaemon
def _daemon(tmp_path):
cfg = Config(
hub=HubConfig(url="http://localhost:9999", username="testuser"),
node=NodeConfig(quic_port=29011, ui_port=29012),
groups=[],
keystore=KeystoreConfig(path=tmp_path / "keystore.enc"),
data_dir=tmp_path / "data",
)
return NodeDaemon(cfg)
def _http_error(status: int, headers: dict | None = None) -> httpx.HTTPStatusError:
req = httpx.Request("POST", "http://localhost:9999/v1/nodes/auth")
resp = httpx.Response(status, headers=headers or {}, request=req)
return httpx.HTTPStatusError(f"{status}", request=req, response=resp)
class _Hub:
"""A hub whose `startup` raises the given sequence, then returns a session."""
def __init__(self, seq):
self._seq = list(seq)
self.calls = 0
async def startup(self, endpoint_hint=None):
self.calls += 1
item = self._seq.pop(0)
if isinstance(item, Exception):
raise item
return item
@pytest.mark.asyncio
@pytest.mark.parametrize("status", [429, 500, 502, 503, 504])
async def test_a_transient_status_is_retried_not_fatal(tmp_path, status, monkeypatch):
slept = []
async def _sleep(d):
slept.append(d)
monkeypatch.setattr(asyncio, "sleep", _sleep)
daemon = _daemon(tmp_path)
session = object()
hub = _Hub([_http_error(status), session]) # transient, then success
got = await daemon._login_with_retry(hub)
assert got is session # it recovered instead of crashing
assert hub.calls == 2 # retried once
assert slept # it backed off
@pytest.mark.asyncio
async def test_retry_after_is_respected(tmp_path, monkeypatch):
slept = []
async def _sleep(d):
slept.append(d)
monkeypatch.setattr(asyncio, "sleep", _sleep)
daemon = _daemon(tmp_path)
hub = _Hub([_http_error(429, {"Retry-After": "42"}), object()])
await daemon._login_with_retry(hub)
assert 42 in slept
@pytest.mark.asyncio
async def test_a_401_still_retries_and_stays_alive(tmp_path, monkeypatch):
async def _sleep(d):
pass
monkeypatch.setattr(asyncio, "sleep", _sleep)
daemon = _daemon(tmp_path)
session = object()
hub = _Hub([_http_error(401), session])
got = await daemon._login_with_retry(hub)
assert got is session
assert daemon._state.get("status") in ("waiting_for_node_key", "waiting_for_account")
@pytest.mark.asyncio
async def test_a_genuine_client_error_still_raises(tmp_path, monkeypatch):
"""A 400/422 is a bug, not a transient state — it must not be swallowed."""
async def _sleep(d):
pass
monkeypatch.setattr(asyncio, "sleep", _sleep)
daemon = _daemon(tmp_path)
hub = _Hub([_http_error(400), object()])
with pytest.raises(httpx.HTTPStatusError):
await daemon._login_with_retry(hub)
|