diff options
Diffstat (limited to 'packages/meshbay-hub/src/meshbay_hub')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/search-page.js | 226 | ||||
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/transport.js | 87 |
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; |