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
|
"""
A reconnect must not fail the download it is reconnecting for.
Reported live on 0.20: a large download from a node under load stopped with
"Handshake required" in the transfers panel. `_sendAndWait` let every message
through while the reconnect's own connect() ran, because the handshake it sends
has to go through; so a chunk whose retry came due in that window went out on a
channel that had opened and not finished its handshake. The node answered
"Handshake required" (webrtc/dispatch.py), which no retry treats as a dropped
connection. Now only the opening messages pass, and anything else waits for the
session or fails as a dropped connection, which the download waits out.
The replaced channel's close event is the second half: it reached the new
connection, marked it closed and failed what it had in flight.
These run the shipped transport and pipeline against a node that refuses what
comes before the handshake: harness/reconnect_race_harness.mjs.
"""
import json
import os
import re
import shutil
import subprocess
from pathlib import Path
import pytest
from spa_source import classic_scripts, transport_source
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
FILE_UTILS = STATIC / "file-utils.js"
HARNESS = Path(__file__).parent / "harness" / "reconnect_race_harness.mjs"
DISPATCH = (Path(__file__).resolve().parents[2] / "meshbay-node" / "src" / "meshbay_node"
/ "transport" / "webrtc" / "dispatch.py")
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not FILE_UTILS.exists(),
reason="node or the SPA sources are not available")
def _run(**cfg) -> dict:
transport = os.pathsep.join(
str(STATIC / name) for name in classic_scripts() if name.startswith("transport"))
proc = subprocess.run(
["node", str(HARNESS), str(FILE_UTILS), transport, json.dumps(cfg)],
capture_output=True, text=True, encoding="utf-8", timeout=120)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def test_retries_that_come_due_during_the_handshake_wait_for_it():
# The chunks in flight time out just before the connection is declared
# failed, and their retries land while the new channel is handshaking.
out = _run()
assert out["reconnected"] == 1 and out["connections"] == 2
assert out["refusedBeforeHandshake"] == 0, "a chunk was asked for before the handshake"
assert out["outcome"] == "done"
assert out["written"] == 400 and out["inOrder"]
@pytest.mark.parametrize("cfg", [
{"failAfter": 20000}, # failure declared before anything times out
{"failAfter": 31000, "rttMs": 800}, # a slow handshake
{"openMs": 2500}, # a slow channel
])
def test_the_download_survives_other_timings(cfg):
out = _run(**cfg)
assert out["refusedBeforeHandshake"] == 0
assert out["outcome"] == "done" and out["written"] == 400 and out["inOrder"]
@pytest.mark.parametrize("late", [800, 1500, 3000])
def test_the_replaced_channels_close_event_is_ignored(late):
out = _run(lateCloseMs=late)
assert out["outcome"] == "done" and out["written"] == 400
def _node_opening_types() -> set[str]:
"""What the node answers before the handshake, read from its dispatch."""
from meshbay_common.protocol import MNP
src = DISPATCH.read_text(encoding="utf-8")
before = src[src.index("def _dispatch_message"):src.index("elif self._user_id is None")]
names = set(re.findall(r"MNP\.([A-Z_]+)", before))
return {getattr(MNP, n) for n in names}
def test_the_opening_messages_are_the_ones_the_node_takes_before_the_handshake():
m = re.search(r"const SESSION_OPENING = new Set\(\[(.*?)\]\);", transport_source(), re.S)
client = set(re.findall(r"'([a-z_]+)'", m.group(1)))
# device_hello is the one connect() sends after the proof and before it
# returns: the node takes it then, and the session is not up until it has.
assert client - _node_opening_types() == {"device_hello"}
assert _node_opening_types() <= client
|