// 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 };