From 950f4aa6d107e127fb6dc0579beedbff0e6f1af5 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 25 Sep 2026 10:25:44 +0200 Subject: refactor(client): move the connection pool and the media tiles to modules ConnectionPool (with connectToGroup and its limits) leaves search-page.js for connection-pool.js, and LazyTile/MediaThumb leave video-app.js for media-tiles.js, cut as text. The shell and the Music and Photos apps now reach them without importing the search page or the Videos app. Co-Authored-By: Claude Opus 5.5 --- .../src/meshbay_hub/static/connection-pool.js | 260 +++++++++++++++++++++ 1 file changed, 260 insertions(+) create mode 100644 packages/meshbay-hub/src/meshbay_hub/static/connection-pool.js (limited to 'packages/meshbay-hub/src/meshbay_hub/static/connection-pool.js') 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 }; -- cgit v1.2.3