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
|
"""
Search reaches each group with one offer, and never more than a few at a time.
A phone on 4G reported one group of five missing from Search on half its
visits, and all but one on a fifth of them. The nodes were answering every
offer that reached them in about a second; the hub was refusing the others. The
page negotiated each group twice — the sweep opened a connection, read the
index, closed it, and the warm-up opened the same group again — and the sweep,
the warm-up and the tiles each kept a concurrency ceiling of their own, which
added up past what the hub admitted per account. Every refusal read as "node
unreachable".
Now every connection goes through one `ConnectionPool`, which holds the only
ceiling and keeps what the sweep opened. These run the shipped pool, index
fetch and sweep, lifted out of search-page.js as text, against a fake clock and
fake nodes that take two seconds to answer — see harness/search_pool_harness.mjs.
"""
import json
import shutil
import subprocess
from pathlib import Path
import pytest
STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
SEARCH_PAGE = STATIC / "search-page.js"
HARNESS = Path(__file__).parent / "harness" / "search_pool_harness.mjs"
pytestmark = pytest.mark.skipif(
shutil.which("node") is None or not SEARCH_PAGE.exists(),
reason="node or the SPA sources are not available")
def _run(**cfg) -> dict:
proc = subprocess.run(
["node", str(HARNESS), str(SEARCH_PAGE), json.dumps(cfg)],
capture_output=True, text=True)
assert proc.returncode == 0, proc.stderr
return json.loads(proc.stdout)
def test_each_group_costs_one_offer_and_stays_open():
out = _run(scenario="sweep", groups=5)
assert out["found"] == 5
assert out["offers"] == 5, "a group was negotiated more than once"
assert out["openTransports"] == 5, "the sweep closed what the tiles will need"
def test_twenty_groups_stay_within_the_ceiling_and_the_pool():
out = _run(scenario="sweep", groups=20)
assert out["found"] == 20
assert out["offers"] == 20
assert out["peakNegotiating"] == out["maxInFlight"]
assert out["openTransports"] == out["maxPool"], "the pool kept more than its size"
def test_an_index_is_never_read_from_a_connection_evicted_under_it():
"""With more groups than the pool keeps, and tiles on screen using the
connections of groups already found, the connection a slow index is being
read from is the least recently used one there — the first to be evicted,
which would fail that group for a reason of this page's own making."""
out = _run(scenario="scroll", groups=20, indexMs=5000)
assert out["readsOnClosed"] == 0
assert out["found"] == 20
def test_dead_nodes_cost_their_own_places_and_nothing_else():
out = _run(scenario="sweep", groups=20, dead=[0, 3, 7])
assert out["found"] == 17
assert out["unreachable"] == 3
# Three dead groups hold three of six places for one deadline, side by side.
assert out["at"] < 10000 + 20 * 2000 / 6 + 1000
def test_tiles_asking_during_the_sweep_share_its_connections():
"""Tiles mount as soon as a group's index lands, while the sweep is still
dialling the others. They must neither negotiate a group a second time nor
add places of their own beside the sweep's."""
out = _run(scenario="tiles", groups=20)
assert out["offers"] == 20
assert out["peakNegotiating"] <= out["maxInFlight"]
def test_refreshing_reuses_every_connection_that_is_alive():
out = _run(scenario="refresh", groups=5)
assert out["offersSecond"] == 0
assert out["pings"] == 5, "a reused connection must prove it is alive first"
assert out["found"] == 5
def test_a_connection_that_died_in_its_sleep_is_replaced_not_waited_on():
"""A phone that slept keeps reporting `connected` on channels that are gone.
Those are found by a short ping and renegotiated; the rest are reused."""
out = _run(scenario="slept", groups=5)
assert out["offersSecond"] == 2
assert out["found"] == 5
assert out["unreachable"] == 0
def test_leaving_the_page_leaves_no_connection_behind():
"""Negotiations still under way when the page closes its pool close what
they obtain, rather than adding it to a pool nobody will close again."""
out = _run(scenario="unmount", groups=8)
assert out["openAfter"] == 0
assert out["offers"] == out["maxInFlight"], "groups were dialled after the page closed"
assert out["remembered"] is None, (
"groups the closed page never reached were remembered as down")
|