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
|
"""
A large download is spread over several connections to its node.
One WebRTC connection carries 5 to 6 MB/s at 22 ms on a line that carries 47:
the browser's UDP socket has a 64 KB kernel buffer, a burst overflows it, and
the node's SCTP sender halves its window at every loss. The ceiling is per
connection, so `MeshBayTransport.lanes` opens more of them and
`pipelinedDownload` sends chunk i over lane i modulo their number.
These run the shipped lanes (transport.js, executed whole) under the shipped
pipeline (lifted out of file-utils.js as text) against a fake clock and a fake
node. See harness/download_lanes_harness.mjs.
"""
import json
import os
import shutil
import subprocess
from pathlib import Path
import pytest
from spa_source import classic_scripts
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
FILE_UTILS = STATIC / "file-utils.js"
HARNESS = Path(__file__).parent / "harness" / "download_lanes_harness.mjs"
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")
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def test_a_large_download_runs_over_eight_connections_in_order():
out = _run(chunks=200)
assert out["outcome"] == "done"
assert out["written"] == 200 and out["inOrder"]
assert out["offers"] == 7, "seven lanes beside the connection already open"
lanes = {k: v for k, v in out["carried"].items() if k != "main"}
assert len(lanes) == 7 and min(lanes.values()) >= 15, out["carried"]
# It starts on the main connection without waiting for the lanes.
assert out["carried"]["main"] > max(lanes.values())
assert out["peakInFlight"] == 16, "two chunks in flight per connection"
def test_a_lane_is_the_main_connection_again_and_never_a_join():
args = _run(chunks=200)["laneArgs"]
assert args == {"groupId": "group", "sameGek": True, "sameIdentity": True,
"joinCode": None, "nodePk": "pk"}
def test_a_small_download_opens_nothing():
out = _run(chunks=10)
assert out["outcome"] == "done" and out["offers"] == 0
assert out["peakInFlight"] == 8
def test_a_read_without_a_lease_opens_nothing():
"""A preview or a track is not a transfer, and the node would count each
lane's read against that lane's own leaseless ceiling."""
out = _run(chunks=200, tr="")
assert out["outcome"] == "done" and out["offers"] == 0
assert out["carried"] == {"main": 200}
def test_a_lane_that_fails_is_dropped_and_its_chunks_asked_again():
out = _run(chunks=200, failLaneAfter=3)
assert out["outcome"] == "done"
assert out["written"] == 200 and out["inOrder"]
assert out["retried"] >= 1
assert out["carried"]["lane1"] <= 3
assert out["offers"] == 8, "the dropped lane is replaced, once"
def test_a_node_that_refuses_lanes_costs_one_round_of_offers():
"""`lanes` is asked at every chunk: without the retry delay a refusing node
would be sent an offer per chunk."""
out = _run(chunks=200, refuseLanes=True)
assert out["outcome"] == "done"
assert out["carried"] == {"main": 200}
assert out["offers"] == 7
def test_idle_lanes_are_closed():
out = _run(chunks=200)
assert out["lanesAfterIdle"] == 0 and out["lanesLeftOpen"] == 0
assert out["timersLeft"] == 0, "the idle timer outlived the lanes"
@pytest.mark.parametrize("close_at", [400, 1500])
def test_closing_the_connection_closes_its_lanes(close_at):
"""At 400 ms the lanes are still negotiating, at 1500 they are carrying."""
out = _run(chunks=2000, closeAt=close_at)
assert out["lanesLeftOpenByClose"] == 0
assert out["offers"] == 7, "a closed connection opened more lanes"
|