Merge branch 'main' of meshbay.org:meshbay

1 files changed, 93 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_reconnect_race.py b/packages/meshbay-hub/tests/test_reconnect_race.py
new file mode 100644
index 0000000..01c5182
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_reconnect_race.py
@@ -0,0 +1,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