1 files changed, 108 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_search_pool.py b/packages/meshbay-hub/tests/test_search_pool.py
new file mode 100644
index 0000000..67e66fa
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_search_pool.py
@@ -0,0 +1,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")