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>

2 files changed, 246 insertions, 67 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/search-page.js b/packages/meshbay-hub/src/meshbay_hub/static/search-page.js
index 0a96237..61565e8 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/search-page.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/search-page.js
@@ -16,10 +16,18 @@ import { transfers } from './transfers.js';
 import { mergeUnitEntries } from './source-merge.js';
 import { useStickyBand } from './sticky.js';
 
-// How many groups Search dials at once — a ceiling on concurrency, never a
-// batch. See `inFlight` below for why the difference is the whole of what a
-// reader waits for when a node is down.
-const MAX_IN_FLIGHT = 3;
+// How many connections this page negotiates at once — a ceiling on concurrency,
+// never a batch (see `inFlight` below), and one ceiling for everything here: the
+// sweep, the warm-up and a tile asking for its group all go through
+// `ConnectionPool.connect`, which holds it. Three separate ceilings used to add
+// up behind each other's backs, past what the hub admits per account.
+//
+// Six, sized for twenty groups on a phone: a connection there takes about two
+// seconds, so three at a time is fourteen seconds to reach them all and six is
+// seven, and a node that is down holds one place in six rather than one in
+// three. The hub admits 32 pending offers per account (signaling.py), which
+// leaves room for this page on three devices at once plus their reconnections.
+const MAX_IN_FLIGHT = 6;
 // One WebRTC peer connection per group the search view touches. The cap bounds
 // how many a busy account (many groups on the hub) keeps open at once; groups
 // past it connect lazily when a tile of theirs scrolls into view (video-app.js's
@@ -39,9 +47,10 @@ const DEBOUNCE_MS = 200;
 // worked from there, for the same work against the same node.
 //
 // A node that is not answering produces no progress event and still fails in
-// `SEARCH_STALL_MS`, which is what keeps the fan-out bounded: `fetchAllIndexes`
-// goes in batches of three and waits for the slowest of each, so the number that
-// matters for a page full of unreachable groups is this one, unchanged.
+// `SEARCH_STALL_MS`, which is what keeps the fan-out bounded: a dead group holds
+// one of the `MAX_IN_FLIGHT` places for this long, so the number that matters
+// for a page full of unreachable groups is this one. A hub that refuses an offer
+// for load *is* answering, and transport.js reports each retry as progress.
 // `SEARCH_MAX_MS` bounds the other case — a node that answers ICE and then stops
 // — because a deadline that only ever resets has none.
 const SEARCH_STALL_MS = 10000;
@@ -110,92 +119,189 @@ async function connectToGroup(hubBase, groupId, token, bundleKey, username, user
 
 // -- Connection pool ----------------------------------------------------------
 
+/**
+ * Every connection this page makes, and the only way it makes one.
+ *
+ * The sweep, the warm-up and a tile asking for its group all come here, so the
+ * page has one ceiling on how many it negotiates at once (`MAX_IN_FLIGHT`) and
+ * one connection per group. The sweep used to open its own, fetch the index,
+ * close it, and leave the warm-up to open the same group again straight after:
+ * two offers per group per visit, and on 4G the two overlapped, which is how a
+ * phone ran into the hub's per-account ceilings with five groups.
+ */
 class ConnectionPool {
   constructor(hubBase, onEvict) {
     this._hubBase = hubBase;
     this._connections = new Map();
     this._connecting = new Map();
     // Called with a groupId whenever this pool closes that group's connection
-    // (eviction or closeAll). SearchPage uses it to drop its own record so it
-    // never hands a tile a `_tRef` pointing at a transport just closed here.
+    // (eviction, close or closeAll). SearchPage uses it to drop its own record
+    // so it never hands a tile a `_tRef` pointing at a transport closed here.
     this._onEvict = onEvict || (() => {});
+    // Negotiations under way, and those waiting for one of the places.
+    this._active = 0;
+    this._waiting = [];
+    this._closed = false;
   }
 
-  async connect(groupId, token, bundleKey, username, userId) {
-    const existing = this._connections.get(groupId);
-    if (existing && existing.transport.connected) {
-      existing.lastUsed = Date.now();
-      return existing;
+  get size() { return this._connections.size; }
+
+  get closed() { return this._closed; }
+
+  /** Whether `groupId` has a connection here that using costs no offer. */
+  has(groupId) {
+    const conn = this._connections.get(groupId);
+    return !!(conn && conn.transport.connected);
+  }
+
+  /**
+   * The group's connection, negotiating one if there is none.
+   *
+   * `hold` keeps it from being evicted until `release` — the sweep holds each
+   * group while its index is on the way, because with more groups than the pool
+   * keeps, the connection it is reading from would otherwise be the oldest one
+   * and closed under it.
+   */
+  async connect(groupId, token, bundleKey, username, userId, { hold = false } = {}) {
+    if (this._closed) throw new Error('Search was closed');
+    let conn = this._connections.get(groupId);
+    if (!(conn && conn.transport.connected)) {
+      let p = this._connecting.get(groupId);
+      if (!p) {
+        p = this._negotiate(groupId, token, bundleKey, username, userId);
+        this._connecting.set(groupId, p);
+        p.finally(() => this._connecting.delete(groupId)).catch(() => {});
+      }
+      conn = await p;
     }
-    if (this._connecting.has(groupId)) return this._connecting.get(groupId);
+    conn.lastUsed = Date.now();
+    if (hold) conn.holds += 1;
+    this._evict();
+    return conn;
+  }
+
+  release(conn) {
+    conn.holds = Math.max(0, conn.holds - 1);
+    this._evict();
+  }
 
-    const p = this._doConnect(groupId, token, bundleKey, username, userId);
-    this._connecting.set(groupId, p);
+  /** Drop one group's connection, e.g. one that stopped answering. */
+  close(groupId) {
+    const conn = this._connections.get(groupId);
+    if (!conn) return;
+    try { conn.transport.close(); } catch {}
+    this._connections.delete(groupId);
+    this._onEvict(groupId);
+  }
+
+  async _negotiate(groupId, token, bundleKey, username, userId) {
+    await this._takePlace();
+    let conn;
     try {
-      const conn = await p;
-      this._connections.set(groupId, conn);
-      this._evict();
-      return conn;
+      // Waiting for a place is not part of any deadline: `connectToGroup`'s
+      // timers start inside `_doConnect`, once this negotiation is really on.
+      if (this._closed) throw new Error('Search was closed');
+      conn = await this._doConnect(groupId, token, bundleKey, username, userId);
     } finally {
-      this._connecting.delete(groupId);
+      this._givePlace();
+    }
+    if (this._closed) {
+      try { conn.transport.close(); } catch {}
+      throw new Error('Search was closed');
+    }
+    // A connection that dropped is replaced, and closed so its own reconnect
+    // loop does not keep negotiating a second one for the same group.
+    const stale = this._connections.get(groupId);
+    if (stale && stale.transport !== conn.transport) {
+      try { stale.transport.close(); } catch {}
+    }
+    this._connections.set(groupId, conn);
+    return conn;
+  }
+
+  _takePlace() {
+    if (this._active < MAX_IN_FLIGHT) {
+      this._active += 1;
+      return Promise.resolve();
     }
+    return new Promise((resolve) => this._waiting.push(resolve));
+  }
+
+  _givePlace() {
+    const next = this._waiting.shift();
+    if (next) next();          // handed straight over: `_active` is unchanged
+    else this._active -= 1;
   }
 
   async _doConnect(groupId, token, bundleKey, username, userId) {
     const found = await connectToGroup(
       this._hubBase, groupId, token, bundleKey, username, userId);
     if (!found) throw new Error('offline');
-    const { transport } = found;
+    const { transport, ack } = found;
 
     let gek = null;
     if (transport.gekRaw && window.MeshBayCrypto) {
       gek = await window.MeshBayCrypto.importGEK(
         window.MeshBayCrypto.b64encode(transport.gekRaw));
     }
-    return { transport, gek, lastUsed: Date.now() };
+    return { transport, gek, ack, lastUsed: Date.now(), holds: 0 };
   }
 
   _evict() {
     while (this._connections.size > MAX_POOL_SIZE) {
       let oldestId = null, oldestTime = Infinity;
       for (const [id, conn] of this._connections) {
+        if (conn.holds > 0) continue;
         if (conn.lastUsed < oldestTime) { oldestTime = conn.lastUsed; oldestId = id; }
       }
+      // Everything over the size is held: it goes when it is released.
       if (!oldestId) break;
-      const conn = this._connections.get(oldestId);
-      try { conn.transport.close(); } catch {}
-      this._connections.delete(oldestId);
-      this._onEvict(oldestId);
+      this.close(oldestId);
     }
   }
 
   closeAll() {
+    // In-flight negotiations see this when they finish and close what they got.
+    this._closed = true;
     for (const [id, conn] of this._connections) {
       try { conn.transport.close(); } catch {}
       this._onEvict(id);
     }
     this._connections.clear();
-    for (const [, p] of this._connecting) {
-      p.then(c => { try { c.transport.close(); } catch {} }).catch(() => {});
-    }
-    this._connecting.clear();
   }
 }
 
 // -- Index fetching -----------------------------------------------------------
 
-async function fetchGroupIndex(groupId, token, bundleKey, username, userId) {
-  const found = await connectToGroup(HUB, groupId, token, bundleKey, username, userId);
-  if (!found) return null;
-  const { transport, ack } = found;
+// How long a connection the pool already holds gets to prove it is alive before
+// its index is asked for. A phone that slept keeps reporting `connected` on a
+// channel that is gone, and an index request on it would wait thirty seconds.
+const REUSE_PING_MS = 4000;
 
+/**
+ * One group's index, over the pool's connection to it — which the page then
+ * keeps for its tiles, so reaching a group costs one offer and not two.
+ */
+async function fetchGroupIndex(pool, groupId, token, bundleKey, username, userId) {
+  if (pool.has(groupId)) {
+    const held = await pool.connect(groupId, token, bundleKey, username, userId);
+    try { await held.transport.ping(REUSE_PING_MS); } catch { pool.close(groupId); }
+  }
+  const conn = await pool.connect(
+    groupId, token, bundleKey, username, userId, { hold: true });
+  const { transport, ack } = conn;
+
+  let unlisted = false;
   try {
     // The operator asked for this group to stay out of the global listing.
     // Decided here, before the index is asked for, so nothing of it is held,
     // cached or merged by this page. A listing preference and not a boundary:
     // the node cannot tell this request from the group page's, and opening the
     // group lists everything.
-    if (ack.search_listed === false) return { unlisted: true };
+    if (ack.search_listed === false) {
+      unlisted = true;
+      return { unlisted: true };
+    }
 
     const indexMsg = await transport.fetchIndex();
     // Plural, with the old scalars as the fallback for a node still speaking
@@ -213,7 +319,9 @@ async function fetchGroupIndex(groupId, token, bundleKey, username, userId) {
     // describes this connection, not the group's content.
     return { entries: indexMsg.entries || [], roots, isNodeAdmin: !!ack.is_node_admin };
   } finally {
-    try { transport.close(); } catch {}
+    pool.release(conn);
+    // Nothing of an unlisted group is shown, so nothing will use its connection.
+    if (unlisted) pool.close(groupId);
   }
 }
 
@@ -269,7 +377,7 @@ function rememberDown(ids) {
   } catch { /* private window, or storage refused: the order is advisory */ }
 }
 
-async function fetchAllIndexes(groups, token, username, userId, onProgress, onResult) {
+async function fetchAllIndexes(pool, groups, token, username, userId, onProgress, onResult) {
   const bundleKey = session.bundleKey || await _loadBundleKey();
   if (bundleKey) session.bundleKey = bundleKey;
 
@@ -286,7 +394,7 @@ async function fetchAllIndexes(groups, token, username, userId, onProgress, onRe
 
   await inFlight(queue, MAX_IN_FLIGHT, async (g) => {
     try {
-      const result = await fetchGroupIndex(g.id, token, bundleKey, username, userId);
+      const result = await fetchGroupIndex(pool, g.id, token, bundleKey, username, userId);
       if (result && result.unlisted) {
         // Left out of Search by its operator: not shown, and not down either.
       } else if (result) {
@@ -304,14 +412,18 @@ async function fetchAllIndexes(groups, token, username, userId, onProgress, onRe
         unreachable.push(g.name || g.id);
         downNow.add(g.id);
       }
-    } catch {
+    } catch (e) {
       unreachable.push(g.name || g.id);
-      downNow.add(g.id);
+      // Still refused after transport.js's retries: the hub was busy, which
+      // says nothing about this group's node, so it keeps its place next time.
+      if (!(e && e.status === 429)) downNow.add(g.id);
     }
     done++;
     onProgress({ done, total, unreachable: [...unreachable] });
   });
-  rememberDown(downNow);
+  // A sweep cut short by the page closing saw nothing about any node: every
+  // group left was refused by this page, not by its node.
+  if (!(pool && pool.closed)) rememberDown(downNow);
   return { results, unreachable };
 }
 
@@ -448,8 +560,9 @@ function SearchPage({ token, username, userId, groups, onPlayQueue, userPrefs })
       setFetching(true);
       setProgress({ done: 0, total: groups.length, unreachable: [] });
 
+      if (!poolRef.current) return;
       await fetchAllIndexes(
-        groups, token, username, userId,
+        poolRef.current, groups, token, username, userId,
         (p) => { if (!cancelled) setProgress(p); },
         (results) => { if (!cancelled) setIndexedGroups(new Map(results)); },
       );
@@ -530,19 +643,30 @@ function SearchPage({ token, username, userId, groups, onPlayQueue, userPrefs })
     return entry;
   }, [token, username, userId, markGroupDown]);
 
-  // Warm up connections as soon as indexing finishes so thumbnails start
-  // loading before the user switches views — but only up to the pool's
-  // capacity. Warming every group would just evict the earlier ones before
-  // the user ever gets there; groups past the cap connect lazily when a tile
-  // of theirs scrolls into view (onNeedConn, below). _thumbBlobCache keeps
+  // Hand the tiles their connections as soon as indexing finishes, so
+  // thumbnails start loading before the user switches views. The sweep left
+  // its connections in the pool, so for those this costs no offer at all; a
+  // group is dialled here only when its connection has gone since and the pool
+  // has room — never to evict one that is already open, which would be an offer
+  // spent to lose another. Groups past the pool's size connect lazily when a
+  // tile of theirs scrolls into view (onNeedConn, below). _thumbBlobCache keeps
   // fetched thumbnails across evictions.
   useEffect(() => {
     if (fetching || indexedGroups.size === 0) return;
     let cancelled = false;
 
     (async () => {
-      const groupIds = [...indexedGroups.keys()].slice(0, MAX_POOL_SIZE);
-      await inFlight(groupIds, MAX_IN_FLIGHT, async (gid) => {
+      const pool = poolRef.current;
+      if (!pool) return;
+      const ids = [...indexedGroups.keys()];
+      const open = ids.filter((gid) => pool.has(gid));
+      for (const gid of open) {
+        if (cancelled) return;
+        try { await connectGroup(gid); } catch { /* skip */ }
+      }
+      const room = Math.max(0, MAX_POOL_SIZE - pool.size);
+      const closed = ids.filter((gid) => !pool.has(gid)).slice(0, room);
+      await inFlight(closed, MAX_IN_FLIGHT, async (gid) => {
         if (cancelled) return;
         try { await connectGroup(gid); } catch { /* skip */ }
       });
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index f6adaac..79cd8e1 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -59,6 +59,57 @@ const UPLOAD_BUFFER_HIGH = 1024 * 1024;
 // that never answers costs a pause rather than the whole attempt.
 const ICE_GATHER_TIMEOUT_MS = 4000;
 
+// Answers to an offer that are about the hub's load, never about the node: 429
+// is one of the hub's per-account ceilings (docs/MESHBAY_DESIGN.md §7.2), and
+// 502/503 is the hub restarting behind its proxy. An offer refused for either is
+// sent again, so it is not reported as a node that cannot be reached — which is
+// what a single 429 used to become in Search, for a group whose node was
+// answering every other offer in under a second. Anything else — 404 for a node
+// that is not connected, 403, 504 for a node that did not answer — fails at once:
+// retrying it would make a dead node cost time instead of costing nothing.
+const OFFER_RETRY_STATUSES = new Set([429, 502, 503]);
+// Worst case about twenty seconds of waiting, all of it on a hub that is
+// answering. Search's own deadline treats each retry as progress and is still
+// bounded by its ceiling.
+const OFFER_RETRY_DELAYS_MS = [500, 1000, 2000, 4000, 8000];
+const OFFER_RETRY_AFTER_MAX_MS = 10000;
+
+/**
+ * POST an offer, sending it again while the hub refuses it for load.
+ *
+ * `Retry-After` is honoured when the hub gives one, and every delay is jittered:
+ * the refusals this exists for come in bursts — several groups dialled at once,
+ * or every connection of a phone reconnecting as it wakes — and retries that
+ * all land on the same millisecond would be refused together again.
+ *
+ * `isClosed` is asked after each wait, so a caller that gave up (Search's
+ * deadline, a page that unmounted) sends no further offer on its behalf.
+ */
+async function postOffer(call, url, init, { isClosed, onRetry } = {}) {
+  for (let attempt = 0; ; attempt += 1) {
+    const resp = await call(url, init);
+    if (resp.ok) return resp;
+    if (!OFFER_RETRY_STATUSES.has(resp.status) || attempt >= OFFER_RETRY_DELAYS_MS.length) {
+      const detail = await resp.json().catch(() => ({}));
+      const err = new Error(`Signaling failed: ${resp.status} ${detail.detail || ''}`);
+      err.status = resp.status;
+      throw err;
+    }
+    const after = Number(resp.headers && resp.headers.get && resp.headers.get('Retry-After'));
+    const base = after > 0
+      ? Math.min(after * 1000, OFFER_RETRY_AFTER_MAX_MS)
+      : OFFER_RETRY_DELAYS_MS[attempt];
+    const delay = Math.round(base * (0.75 + Math.random() * 0.5));
+    if (onRetry) onRetry(resp.status, delay, attempt + 1);
+    await new Promise((resolve) => setTimeout(resolve, delay));
+    if (isClosed && isClosed()) {
+      const err = new Error('Transport closed while the hub was busy');
+      err.status = resp.status;
+      throw err;
+    }
+  }
+}
+
 const STREAM_CREDITS = 24;
 
 function _aborted() {
@@ -823,23 +874,27 @@ class MeshBayTransport {
     // app:// origin is refused by CORS.
     const call = (window.MeshBayPlatform && window.MeshBayPlatform.apiFetch)
       || fetch;
-    const resp = await call(
+    const resp = await postOffer(call,
       `${this._hubUrl}/v1/nodes/${nodeId}/webrtc/offer`, {
-      method: 'POST',
-      headers: {
-        'Content-Type': 'application/json',
-        'Authorization': `Bearer ${this._accessToken}`,
-      },
-      body: JSON.stringify({
-        sdp: this._pc.localDescription.sdp,
-        ice_candidates: [],
-      }),
-    });
-
-    if (!resp.ok) {
-      const detail = await resp.json().catch(() => ({}));
-      throw new Error(`Signaling failed: ${resp.status} ${detail.detail || ''}`);
-    }
+        method: 'POST',
+        headers: {
+          'Content-Type': 'application/json',
+          'Authorization': `Bearer ${this._accessToken}`,
+        },
+        body: JSON.stringify({
+          sdp: this._pc.localDescription.sdp,
+          ice_candidates: [],
+        }),
+      }, {
+        isClosed: () => this._closed,
+        onRetry: (status, delayMs, attempt) => {
+          console.warn('[MeshBay] Hub refused the offer with', status,
+                       '— sending it again in', delayMs, 'ms (attempt', attempt, ')');
+          trace('offer_retry', { status, delay_ms: delayMs, attempt });
+          // The hub answered: a busy hub, not a silent node.
+          this._noteConnectProgress('offer_retry');
+        },
+      });
 
     const answer = await resp.json();
     this._rawAnswerSdp = answer.sdp;