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>

7 files changed, 530 insertions, 33 deletions

diff --git a/packages/meshbay-hub/tests/harness/offer_retry_harness.mjs b/packages/meshbay-hub/tests/harness/offer_retry_harness.mjs
new file mode 100644
index 0000000..3a4b568
--- /dev/null
+++ b/packages/meshbay-hub/tests/harness/offer_retry_harness.mjs
@@ -0,0 +1,84 @@
+// Run the SHIPPED `postOffer` against a fake hub and a fake clock.
+//
+// `postOffer` and the constants that govern it are lifted out of transport.js
+// as text. What is modelled is the hub — a list of the statuses it answers in
+// turn, each with an optional Retry-After — and the clock.
+//
+// Usage: node offer_retry_harness.mjs <path to transport.js> <json config>
+import { readFileSync } from 'fs';
+
+const src = readFileSync(process.argv[2], 'utf8');
+const cfg = JSON.parse(process.argv[3] || '{}');
+
+const {
+  // What the hub answers, one entry per POST: a status, or [status, retryAfter].
+  answers = [200],
+  // After this many ms the caller closes its transport. null: never.
+  closeAt = null,
+  runUntil = 600000,
+} = cfg;
+
+let now = 0;
+let nextId = 1;
+let timers = [];
+globalThis.setTimeout = (fn, ms) => {
+  const t = { at: now + (ms || 0), fn, id: nextId++ };
+  timers.push(t);
+  return t.id;
+};
+globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); };
+const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); };
+async function run(until) {
+  await flush();
+  while (timers.length) {
+    const due = timers.reduce((a, b) => (b.at < a.at ? b : a));
+    if (due.at > until) break;
+    timers = timers.filter((t) => t !== due);
+    now = due.at;
+    due.fn();
+    await flush();
+  }
+  now = until;
+}
+// Jitter at its midpoint, so the delays asserted are the nominal ones.
+Math.random = () => 0.5;
+
+const posts = [];
+const call = async () => {
+  const a = answers[Math.min(posts.length, answers.length - 1)];
+  const [status, retryAfter] = Array.isArray(a) ? a : [a, null];
+  posts.push(now);
+  return {
+    ok: status >= 200 && status < 300,
+    status,
+    headers: new Headers(retryAfter === null ? {} : { 'Retry-After': String(retryAfter) }),
+    json: async () => ({ detail: `status ${status}` }),
+  };
+};
+
+const lift = (signature, end) => {
+  const start = src.indexOf(signature);
+  if (start < 0) throw new Error(`${signature} is gone from transport.js`);
+  return src.slice(start, src.indexOf(end, start) + end.length);
+};
+const make = new Function(
+  `${lift('const OFFER_RETRY_STATUSES', ';\n')}
+   ${lift('const OFFER_RETRY_DELAYS_MS', ';\n')}
+   ${lift('const OFFER_RETRY_AFTER_MAX_MS', ';\n')}
+   ${lift('async function postOffer(', '\n}\n')}
+   return postOffer;`,
+);
+const postOffer = make();
+
+let closed = false;
+if (closeAt !== null) setTimeout(() => { closed = true; }, closeAt);
+const retries = [];
+let outcome = null;
+postOffer(call, 'https://hub.example/v1/nodes/n/webrtc/offer', {}, {
+  isClosed: () => closed,
+  onRetry: (status, delay) => retries.push({ status, delay }),
+}).then(() => { outcome = { result: 'answered', at: now }; })
+  .catch((e) => { outcome = { result: 'failed', at: now, status: e.status ?? null }; });
+
+await run(runUntil);
+console.log(JSON.stringify({ ...(outcome || { result: 'pending' }), posts, retries }));
diff --git a/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs b/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs
index c05e23b..5f95ede 100644
--- a/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs
+++ b/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs
@@ -80,7 +80,7 @@ const session = { bundleKey: 'k' };
 const _loadBundleKey = async () => 'k';
 
 // One group's index, answered on the clock rather than over a network.
-const fetchGroupIndex = (groupId) => new Promise((resolve, reject) => {
+const fetchGroupIndex = (_pool, groupId) => new Promise((resolve, reject) => {
   const n = Number(String(groupId).split('-')[1]);
   const isDead = dead.includes(n);
   setTimeout(
@@ -123,7 +123,7 @@ const list = Array.from({ length: groups }, (_, i) => ({ id: `g-${i}`, name: `G$
 
 let finishedAt = null;
 fetchAllIndexes(
-  list, 'tok', 'alice', 'u1',
+  null, list, 'tok', 'alice', 'u1',
   () => {},                                            // progress counters only
   (results) => { shown.push({ at: now, groups: results.size }); },
 ).then(() => { finishedAt = now; });
diff --git a/packages/meshbay-hub/tests/harness/search_pool_harness.mjs b/packages/meshbay-hub/tests/harness/search_pool_harness.mjs
new file mode 100644
index 0000000..521a97e
--- /dev/null
+++ b/packages/meshbay-hub/tests/harness/search_pool_harness.mjs
@@ -0,0 +1,221 @@
+// Run the SHIPPED `ConnectionPool`, `fetchGroupIndex` and `fetchAllIndexes`
+// against a fake clock and fake nodes, and count offers.
+//
+// The question this answers is how many connections Search negotiates, how many
+// at once, and whether one it is reading from can be closed under it — the
+// things that decided whether a phone ran into the hub's per-account ceilings.
+// The pool, the index fetch and the sweep are lifted out of search-page.js as
+// text; what is modelled is the clock and the nodes (`connectToGroup`, which
+// is where an offer is made).
+//
+// Usage: node search_pool_harness.mjs <path to search-page.js> <json config>
+import { readFileSync } from 'fs';
+
+const src = readFileSync(process.argv[2], 'utf8');
+const cfg = JSON.parse(process.argv[3] || '{}');
+
+const {
+  scenario = 'sweep',
+  groups = 5,
+  // Group indexes whose node never answers; `connectToGroup` fails after deadMs.
+  dead = [],
+  connectMs = 2000,       // a phone on 4G
+  deadMs = 10000,
+  indexMs = 300,
+  runUntil = 600000,
+} = cfg;
+
+// ── The clock ────────────────────────────────────────────────────────────────
+
+let now = 0;
+let nextId = 1;
+let timers = [];
+globalThis.setTimeout = (fn, ms) => {
+  const t = { at: now + (ms || 0), fn, id: nextId++ };
+  timers.push(t);
+  return t.id;
+};
+globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); };
+// The pool orders its connections by `Date.now()`, so that is the fake clock too.
+Date.now = () => now;
+const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); };
+async function run(until) {
+  await flush();
+  while (timers.length) {
+    const due = timers.reduce((a, b) => (b.at < a.at ? b : a));
+    if (due.at > until) break;
+    timers = timers.filter((t) => t !== due);
+    now = due.at;
+    due.fn();
+    await flush();
+  }
+  now = until;
+}
+const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
+
+// ── The nodes ────────────────────────────────────────────────────────────────
+
+let offers = 0;
+let negotiating = 0;
+let peakNegotiating = 0;
+let readsOnClosed = 0;
+let pings = 0;
+const transports = [];
+// Groups whose held connection has died quietly, as a phone's does in its sleep.
+const silentlyDead = new Set();
+
+class FakeTransport {
+  constructor(groupId) {
+    this.groupId = groupId;
+    this.closedFlag = false;
+    transports.push(this);
+  }
+  get connected() { return !this.closedFlag; }
+  close() { this.closedFlag = true; }
+  async ping(timeoutMs) {
+    pings += 1;
+    if (silentlyDead.has(this.groupId)) {
+      await sleep(timeoutMs);
+      throw new Error('Response timeout');
+    }
+    await sleep(100);
+  }
+  async fetchIndex() {
+    // Indexes differ in size, so reads end at different times and overlap
+    // other groups' negotiations — which uniform timings never do.
+    const n = Number(String(this.groupId).split('-')[1]);
+    await sleep(indexMs * (1 + (n % 4)));
+    if (this.closedFlag) { readsOnClosed += 1; throw new Error('Transport closed'); }
+    return { entries: [{ id: `e-${this.groupId}` }] };
+  }
+}
+
+async function connectToGroup(_hub, groupId) {
+  offers += 1;
+  negotiating += 1;
+  peakNegotiating = Math.max(peakNegotiating, negotiating);
+  try {
+    const n = Number(String(groupId).split('-')[1]);
+    if (dead.includes(n)) {
+      await sleep(deadMs);
+      throw new Error('Connection timeout');
+    }
+    await sleep(connectMs);
+    return { transport: new FakeTransport(groupId), ack: {} };
+  } finally {
+    negotiating -= 1;
+  }
+}
+
+globalThis.window = {};
+let remembered = null;
+globalThis.localStorage = { getItem: () => null, setItem: (_k, v) => { remembered = v; } };
+const session = { bundleKey: 'k' };
+const _loadBundleKey = async () => 'k';
+const cacheGroupIndex = () => {};
+
+// ── The code under test, as text ─────────────────────────────────────────────
+
+const constant = (name) => {
+  const m = new RegExp(`^const ${name} = (\\d+);`, 'm').exec(src);
+  if (!m) throw new Error(`${name} is gone from search-page.js`);
+  return `const ${name} = ${m[1]};`;
+};
+const lift = (signature) => {
+  const start = src.indexOf(signature);
+  if (start < 0) throw new Error(`${signature} is gone from search-page.js`);
+  return src.slice(start, src.indexOf('\n}\n', start) + 3);
+};
+
+const make = new Function(
+  'connectToGroup', 'window', 'session', '_loadBundleKey', 'cacheGroupIndex',
+  'localStorage',
+  `${constant('MAX_IN_FLIGHT')}
+   ${constant('MAX_POOL_SIZE')}
+   ${constant('REUSE_PING_MS')}
+   const DOWN_KEY = 'harness';
+   ${lift('class ConnectionPool {')}
+   ${lift('async function fetchGroupIndex(')}
+   ${lift('function lastKnownDown(')}
+   ${lift('function rememberDown(')}
+   ${lift('async function inFlight(')}
+   ${lift('async function fetchAllIndexes(')}
+   return { ConnectionPool, fetchAllIndexes, MAX_IN_FLIGHT, MAX_POOL_SIZE };`,
+);
+const code = make(connectToGroup, globalThis.window, session, _loadBundleKey,
+                  cacheGroupIndex, globalThis.localStorage);
+
+// ── The scenarios ────────────────────────────────────────────────────────────
+
+const list = Array.from({ length: groups }, (_, i) => ({ id: `g-${i}`, name: `G${i}` }));
+let evicted = 0;
+const pool = new code.ConnectionPool('https://hub.example', () => { evicted += 1; });
+const out = { maxInFlight: code.MAX_IN_FLIGHT, maxPool: code.MAX_POOL_SIZE };
+
+// Groups whose index has landed — the only ones with tiles on screen.
+const shown = new Set();
+const sweep = async () => {
+  const r = await code.fetchAllIndexes(pool, list, 'tok', 'alice', 'u1', () => {},
+    (results) => { for (const id of results.keys()) shown.add(id); });
+  return { found: r.results.size, unreachable: r.unreachable.length, at: now };
+};
+
+if (scenario === 'sweep') {
+  sweep().then((r) => { Object.assign(out, r); });
+  await run(runUntil);
+} else if (scenario === 'refresh') {
+  // A sweep, then the Search page's own refresh button: the second must reuse.
+  sweep().then(async () => {
+    out.offersFirst = offers;
+    const again = await sweep();
+    out.offersSecond = offers - out.offersFirst;
+    Object.assign(out, again);
+  });
+  await run(runUntil);
+} else if (scenario === 'slept') {
+  // A sweep, then the phone sleeps and two connections die without saying so.
+  sweep().then(async () => {
+    out.offersFirst = offers;
+    silentlyDead.add('g-0');
+    silentlyDead.add('g-1');
+    const again = await sweep();
+    out.offersSecond = offers - out.offersFirst;
+    Object.assign(out, again);
+  });
+  await run(runUntil);
+} else if (scenario === 'tiles') {
+  // A tile of every group asks for its connection while the sweep is on.
+  sweep().then((r) => { Object.assign(out, r); });
+  for (const g of list) pool.connect(g.id, 'tok', 'k', 'alice', 'u1').catch(() => {});
+  await run(runUntil);
+} else if (scenario === 'scroll') {
+  // Tiles of groups already on screen keep using their connections while the
+  // sweep is still reading other groups' indexes — which makes a connection an
+  // index is being read from the least recently used one in the pool.
+  let sweeping = true;
+  sweep().then((r) => { sweeping = false; Object.assign(out, r); });
+  const touch = async () => {
+    while (sweeping) {
+      await sleep(250);
+      for (const id of shown) {
+        if (pool.has(id)) pool.connect(id, 'tok', 'k', 'alice', 'u1').catch(() => {});
+      }
+    }
+  };
+  touch();
+  await run(runUntil);
+} else if (scenario === 'unmount') {
+  // The page goes away while connections are still being negotiated.
+  sweep().catch(() => {});
+  await run(connectMs / 2);
+  pool.closeAll();
+  await run(runUntil);
+  out.openAfter = transports.filter((t) => !t.closedFlag).length;
+}
+
+Object.assign(out, {
+  offers, peakNegotiating, readsOnClosed, pings, evicted, poolSize: pool.size,
+  remembered: remembered === null ? null : JSON.parse(remembered),
+  openTransports: transports.filter((t) => !t.closedFlag).length,
+});
+console.log(JSON.stringify(out));
diff --git a/packages/meshbay-hub/tests/test_offer_retry.py b/packages/meshbay-hub/tests/test_offer_retry.py
new file mode 100644
index 0000000..3c968db
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_offer_retry.py
@@ -0,0 +1,78 @@
+"""
+An offer the hub refuses for load is sent again; one refused for cause is not.
+
+A phone's Search reported groups as unreachable whose node was answering every
+offer that reached it: the hub had refused those offers with 429, one of its
+per-account ceilings, and the browser took that for the node. A 429 — or a 502
+or 503 while the hub restarts behind its proxy — says the hub is busy, so
+`postOffer` waits (the hub's `Retry-After` when it gives one) and sends the same
+offer again. A 404 for a node that is not connected, or a 504 for one that did
+not answer, fails at once: retrying those would make a dead node cost time.
+
+These run the shipped `postOffer`, lifted out of transport.js as text, against
+a fake hub and a fake clock — see harness/offer_retry_harness.mjs.
+"""
+
+import json
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+
+STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
+TRANSPORT = STATIC / "transport.js"
+HARNESS = Path(__file__).parent / "harness" / "offer_retry_harness.mjs"
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("node") is None or not TRANSPORT.exists(),
+    reason="node or the SPA sources are not available")
+
+
+def _post(**cfg) -> dict:
+    proc = subprocess.run(
+        ["node", str(HARNESS), str(TRANSPORT), json.dumps(cfg)],
+        capture_output=True, text=True)
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+def test_a_busy_hub_is_asked_again_when_it_says():
+    out = _post(answers=[[429, 1], 200])
+    assert out["result"] == "answered"
+    assert out["posts"] == [0, 1000], "Retry-After was not honoured"
+
+
+def test_without_retry_after_the_waits_grow():
+    out = _post(answers=[429, 429, 503, 200])
+    assert out["result"] == "answered"
+    assert out["posts"] == [0, 500, 1500, 3500]
+
+
+def test_a_node_that_is_not_there_fails_at_once():
+    for status in (404, 403, 504):
+        out = _post(answers=[status])
+        assert out["result"] == "failed"
+        assert out["status"] == status
+        assert out["posts"] == [0], f"{status} was retried"
+
+
+def test_a_hub_that_stays_busy_is_given_up_on():
+    out = _post(answers=[429])
+    assert out["result"] == "failed"
+    assert out["status"] == 429
+    assert len(out["posts"]) == 6
+    assert out["at"] <= 20000, "a busy hub cost more than the worst case stated"
+
+
+def test_a_retry_after_that_asks_too_much_is_capped():
+    out = _post(answers=[[429, 3600], 200])
+    assert out["posts"] == [0, 10000]
+
+
+def test_a_caller_that_gave_up_sends_nothing_more():
+    """Search's deadline, or a page that went away, closes the transport while
+    it waits; the offer must not go out on its behalf afterwards."""
+    out = _post(answers=[429], closeAt=100)
+    assert out["result"] == "failed"
+    assert out["posts"] == [0]
diff --git a/packages/meshbay-hub/tests/test_search_fanout.py b/packages/meshbay-hub/tests/test_search_fanout.py
index eda9465..196aa5a 100644
--- a/packages/meshbay-hub/tests/test_search_fanout.py
+++ b/packages/meshbay-hub/tests/test_search_fanout.py
@@ -22,17 +22,21 @@ reader for ten seconds, while two groups of the same batch had answered in two
 hundred milliseconds and nine others had not been dialled at all.
 
 Three things fix it, and each is asserted below: a result is drawn when it
-arrives rather than when its neighbours finish; the ceiling of three is a
+arrives rather than when its neighbours finish; the ceiling is a
 ceiling on *concurrency* rather than a batch, so a dead group holds one place
 instead of stalling a batch and everything queued behind it; and a group that
 did not answer last time is dialled last.
 
 The third is not a refinement of the second, it is what makes the steady state
-correct. A ceiling alone still lets dead groups occupy every place at once —
-with four of twelve down, five live groups appear straight away and the last
-three wait out a deadline. Ordering by what was silent last time puts all eight
-on screen in 600 ms. The cost is that a browser seeing these groups for the
-first time has nothing to order by and pays the first sweep, once.
+correct. A ceiling alone still lets dead groups occupy every place at once — as
+many dead groups as there are places, listed first, and nothing is drawn until
+a deadline expires. Ordering by what was silent last time puts them behind
+every live group. The cost is that a browser seeing these groups for the first
+time has nothing to order by and pays the first sweep, once.
+
+The ceiling is six (search-page.js, `MAX_IN_FLIGHT`). It was three, which was
+also exactly what the hub admitted in flight per account, so the page ran into
+the hub's refusals as soon as anything else connected beside the sweep.
 
 These run the shipped `fetchAllIndexes`, `inFlight`, `lastKnownDown` and
 `rememberDown`, lifted out of search-page.js as text, against a fake clock and a
@@ -93,21 +97,17 @@ def test_several_nodes_down_do_not_delay_the_first_result():
         "dead groups are being waited for one after another")
 
 
-def test_on_a_first_visit_dead_groups_can_still_hold_the_last_few_back():
+def test_on_a_first_visit_four_dead_groups_hold_nothing_back():
     """
-    The limit of a ceiling, stated rather than glossed over.
-
-    Four dead groups and three places: once all three are held by nodes that are
-    not answering, nothing else is dialled until one of them gives up. The first
-    results are immediate and most arrive at once, but the last few do wait — on
-    a browser that has never run this sweep before and so has nothing to order
-    by. The test after this one is the steady state, which is what a reader
-    actually lives in.
+    Four dead groups of twelve on a browser that has never swept before, so
+    nothing orders them. With three places they held every place at once and
+    the last three live groups waited out a deadline; with six, two places stay
+    free and every live group is on screen before any deadline expires.
     """
     out = _sweep(groups=12, dead=[1, 4, 7, 10])
     assert out["firstPaintAt"] == LIVE_MS
     early = [r for r in out["renders"] if r["at"] < DEAD_MS]
-    assert early[-1]["groups"] == 5, (
+    assert early[-1]["groups"] == 8, (
         "the shape of the first visit has changed — re-measure it rather than "
         "adjusting this number")
 
@@ -128,19 +128,20 @@ def test_once_the_silent_groups_are_known_no_live_one_waits_for_them():
 
 def test_a_group_that_did_not_answer_is_dialled_last_next_time():
     """
-    The last bad case, and the reason the browser remembers.
+    The limit of a ceiling, and the reason the browser remembers.
 
-    The three groups the hub lists first are all down, so on a first visit all
-    three places are held at once and there is nothing to draw until one frees
-    up. Told which groups were silent, the sweep puts them behind everything
-    else and the first result arrives on time.
+    The six groups the hub lists first are all down, so on a first visit every
+    place is held at once and there is nothing to draw until one frees up. Told
+    which groups were silent, the sweep puts them behind everything else and
+    the first result arrives on time.
     """
-    first_visit = _sweep(groups=12, dead=[0, 1, 2])
+    down = list(range(6))
+    first_visit = _sweep(groups=12, dead=down)
     assert first_visit["firstPaintAt"] == DEAD_MS + LIVE_MS
-    assert first_visit["remembered"] == ["g-0", "g-1", "g-2"], (
+    assert first_visit["remembered"] == [f"g-{n}" for n in down], (
         "the sweep must record who was silent, or the next visit repeats this")
 
-    again = _sweep(groups=12, dead=[0, 1, 2], knownDown=[0, 1, 2])
+    again = _sweep(groups=12, dead=down, knownDown=down)
     assert again["firstPaintAt"] == LIVE_MS
 
 
@@ -149,7 +150,7 @@ def test_the_order_is_advisory_and_survives_a_browser_that_refuses_storage():
     A private window throws on both halves of `localStorage`. The sweep must
     then behave exactly as a first visit does, not fail.
     """
-    out = _sweep(groups=12, dead=[0, 1, 2], knownDown="refuses")
+    out = _sweep(groups=12, dead=list(range(6)), knownDown="refuses")
     assert out["firstPaintAt"] == DEAD_MS + LIVE_MS
     assert out["remembered"] is None
     assert out["finishedAt"] is not None, "the sweep did not finish"
@@ -166,9 +167,9 @@ def test_a_node_that_comes_back_is_not_punished_for_ever():
 
 def test_no_more_groups_are_dialled_at_once_than_the_ceiling():
     """
-    The ceiling is still a ceiling: twelve groups with three places and every
-    node down cost four deadlines, not one and not twelve.
+    The ceiling is still a ceiling: twelve groups with six places and every
+    node down cost two deadlines, not one and not twelve.
     """
     out = _sweep(groups=12, dead=list(range(12)))
-    assert out["inFlight"] == 3
-    assert out["finishedAt"] == 4 * DEAD_MS
+    assert out["inFlight"] == 6
+    assert out["finishedAt"] == 2 * DEAD_MS
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")
diff --git a/packages/meshbay-hub/tests/test_transport_contracts.py b/packages/meshbay-hub/tests/test_transport_contracts.py
index 3c2b0b6..28d6a10 100644
--- a/packages/meshbay-hub/tests/test_transport_contracts.py
+++ b/packages/meshbay-hub/tests/test_transport_contracts.py
@@ -531,13 +531,18 @@ def test_search_tries_every_node_the_hub_offers():
 
 
 def test_search_connects_in_one_place():
-    """Two call sites with their own connect is how one of them kept `nodes[0]`."""
+    """Two call sites with their own connect is how one of them kept `nodes[0]`,
+    and how the sweep and the warm-up each negotiated the same group: every
+    connection goes through the pool, and only the pool calls `connectToGroup`."""
     code = _code_only(SEARCH_PAGE.read_text(encoding="utf-8"))
     assert code.count("transport.connect(") == 1
+    # The definition, and the pool's one call.
+    assert code.count("connectToGroup(") == 2
     pool = code[code.index("async _doConnect("):code.index("_evict() {")]
     index = code[code.index("async function fetchGroupIndex("):
                  code.index("async function fetchAllIndexes(")]
-    assert "connectToGroup(" in pool and "connectToGroup(" in index
+    assert "connectToGroup(" in pool
+    assert "pool.connect(" in index and "connectToGroup(" not in index
 
 
 # node:start only links an unlinked node key to the hub account when it is