fix(hub): Search reaches each group with one offer, and a busy hub is not a dead node0.15

The other half of the 4G failure. The page negotiated every group twice: the
sweep opened a connection, read the index and closed it, then the warm-up
opened the same group again. The sweep, the warm-up and the tiles each had a
concurrency ceiling of their own, and together they went past what the hub
admits per account. Whatever the hub refused was then reported as "node
unreachable" and remembered as down, which put that group last next time.

- Every connection goes through ConnectionPool, which holds the page's one
  ceiling (six at once, sized for twenty groups on a phone) and keeps what
  the sweep opened for the tiles. A visit costs one offer per group. A
  refresh costs none for a connection that answers a four-second ping, and
  a connection that died while the phone slept is replaced, not waited on.
- A connection whose index is being read is held against eviction. With
  more groups than the pool keeps, it was otherwise the least recently used
  one.
- Negotiations still under way when the page closes close what they get,
  and a sweep cut short that way remembers nobody as down.
- transport.js sends an offer again on 429, 502 or 503, honouring
  Retry-After, with jittered waits of about twenty seconds at worst. Search
  counts each retry as progress. A 404, 403 or 504 still fails at once, so
  a dead node costs no time.

The fan-out tests assumed a ceiling of three and were re-measured: four dead
groups of twelve now hold nothing back, even on a first visit. The pool and
the retry run as shipped code, lifted as text, against a fake clock. Each
guard was checked by removing it and seeing its test fail.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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")