1 files changed, 260 insertions, 0 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/connection-pool.js b/packages/meshbay-hub/src/meshbay_hub/static/connection-pool.js
new file mode 100644
index 0000000..aac8225
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/static/connection-pool.js
@@ -0,0 +1,260 @@
+// The connections to groups' nodes that search opens — and the music queue,
+// which plays across groups — one per group, with one ceiling on how many are
+// negotiated at once. A module of its own so the shell can hold a pool without
+// loading the search page.
+
+import { hubFetch, ensureFreshToken } from './hub-client.js';
+
+// How many connections search negotiates at once — a ceiling on concurrency,
+// never a batch (see `inFlight` in search-page.js), and one ceiling for everything: 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 the search 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 (media-tiles.js's
+// LazyTile → onNeedConn). Was 3, which meant any account with more than three
+// groups thrashed the pool: an evicted transport left a stale `_tRef` on every
+// tile of that group, and MediaThumb / useMediaMeta gave up on it with no
+// retry — so posters rendered for a moment, then fell back to a spinner for
+// good. grenet (one shared group) never hit it; cbesson (many) always did.
+const MAX_POOL_SIZE = 12;
+
+// How long a connection attempt may make **no progress**, not how long it may
+// take. Search used a flat 10 s for the whole of `connect()`, which is the hub
+// round trip, ICE gathering (capped at 4 s in transport.js), DTLS, the
+// DataChannel opening and the handshake's own round trips. Opening the same
+// group from the sidebar has no such deadline and gets the transport's own 30 s,
+// so a link slow enough to need twelve seconds — a phone on 4G — failed here and
+// 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: 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;
+const SEARCH_MAX_MS = 30000;
+
+// -- Connecting to a group ----------------------------------------------------
+
+/**
+ * A live connection to one of the nodes serving `groupId`, and its ack — or
+ * null when the hub lists no node at all.
+ *
+ * Every node the hub lists, in turn, exactly as group-page.js does since
+ * 2026-09-11: the list is in hub registration order, and its head is not
+ * necessarily a node that answers. Search took `nodes[0]` and stopped, so a
+ * group with a second, working node counted as unreachable here while it
+ * opened fine from the sidebar. A refusal naming a state of this browser (a
+ * code, a passphrase, a device) is the same from every node and stops the
+ * walk; `not_hosted`, a timeout or a failed connection moves on.
+ */
+async function connectToGroup(hubBase, groupId, token, bundleKey, username, userId) {
+  const nodesData = await hubFetch(`/v1/groups/${groupId}/nodes`, { token });
+  if (!nodesData.nodes || !nodesData.nodes.length) return null;
+
+  const live = (await ensureFreshToken()) || token;
+  let lastErr = null;
+  for (const n of nodesData.nodes) {
+    const transport = new window.MeshBayTransport(hubBase, live);
+    transport.onNeedToken = async () => (await ensureFreshToken()) || token;
+    let stallTimer, capTimer;
+    const stopTimers = () => {
+      clearTimeout(stallTimer);
+      clearTimeout(capTimer);
+      transport.onConnectProgress = null;
+    };
+    try {
+      const ack = await Promise.race([
+        transport.connect(
+          n.node_id, live, groupId, null, null, bundleKey,
+          username, userId, null),
+        new Promise((_, reject) => {
+          const giveUp = () => reject(new Error('Connection timeout'));
+          stallTimer = setTimeout(giveUp, SEARCH_STALL_MS);
+          capTimer = setTimeout(giveUp, SEARCH_MAX_MS);
+          // Each step the transport reports — the peer answering ICE, the
+          // channel opening — buys another window, never more than the cap.
+          transport.onConnectProgress = () => {
+            clearTimeout(stallTimer);
+            stallTimer = setTimeout(giveUp, SEARCH_STALL_MS);
+          };
+        }),
+      ]);
+      stopTimers();
+      return { transport, ack };
+    } catch (e) {
+      stopTimers();
+      lastErr = e;
+      try { transport.close(); } catch {}
+      if (e.reason && e.reason !== 'not_hosted') throw e;
+    }
+  }
+  throw (lastErr || new Error('no node served this group'));
+}
+
+// -- Connection pool ----------------------------------------------------------
+
+/**
+ * Every connection the search 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, 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;
+  }
+
+  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;
+    }
+    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();
+  }
+
+  /** 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 {
+      // 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._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, ack } = found;
+
+    let gek = null;
+    if (transport.gekRaw && window.MeshBayCrypto) {
+      gek = await window.MeshBayCrypto.importGEK(
+        window.MeshBayCrypto.b64encode(transport.gekRaw));
+    }
+    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;
+      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();
+  }
+}
+
+export { ConnectionPool, MAX_IN_FLIGHT, MAX_POOL_SIZE };