diff options
Diffstat (limited to 'packages')
9 files changed, 776 insertions, 100 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; diff --git a/packages/meshbay-hub/tests/harness/offer_retry_harness.mjs b/packages/meshbay-hub/tests/harness/offer_retry_harness.mjs new file mode 100644 index 0000000..3a4b568 --- /dev/null +++ b/packages/meshbay-hub/tests/harness/offer_retry_harness.mjs @@ -0,0 +1,84 @@ +// Run the SHIPPED `postOffer` against a fake hub and a fake clock. +// +// `postOffer` and the constants that govern it are lifted out of transport.js +// as text. What is modelled is the hub — a list of the statuses it answers in +// turn, each with an optional Retry-After — and the clock. +// +// Usage: node offer_retry_harness.mjs <path to transport.js> <json config> +import { readFileSync } from 'fs'; + +const src = readFileSync(process.argv[2], 'utf8'); +const cfg = JSON.parse(process.argv[3] || '{}'); + +const { + // What the hub answers, one entry per POST: a status, or [status, retryAfter]. + answers = [200], + // After this many ms the caller closes its transport. null: never. + closeAt = null, + runUntil = 600000, +} = cfg; + +let now = 0; +let nextId = 1; +let timers = []; +globalThis.setTimeout = (fn, ms) => { + const t = { at: now + (ms || 0), fn, id: nextId++ }; + timers.push(t); + return t.id; +}; +globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); }; +const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); }; +async function run(until) { + await flush(); + while (timers.length) { + const due = timers.reduce((a, b) => (b.at < a.at ? b : a)); + if (due.at > until) break; + timers = timers.filter((t) => t !== due); + now = due.at; + due.fn(); + await flush(); + } + now = until; +} +// Jitter at its midpoint, so the delays asserted are the nominal ones. +Math.random = () => 0.5; + +const posts = []; +const call = async () => { + const a = answers[Math.min(posts.length, answers.length - 1)]; + const [status, retryAfter] = Array.isArray(a) ? a : [a, null]; + posts.push(now); + return { + ok: status >= 200 && status < 300, + status, + headers: new Headers(retryAfter === null ? {} : { 'Retry-After': String(retryAfter) }), + json: async () => ({ detail: `status ${status}` }), + }; +}; + +const lift = (signature, end) => { + const start = src.indexOf(signature); + if (start < 0) throw new Error(`${signature} is gone from transport.js`); + return src.slice(start, src.indexOf(end, start) + end.length); +}; +const make = new Function( + `${lift('const OFFER_RETRY_STATUSES', ';\n')} + ${lift('const OFFER_RETRY_DELAYS_MS', ';\n')} + ${lift('const OFFER_RETRY_AFTER_MAX_MS', ';\n')} + ${lift('async function postOffer(', '\n}\n')} + return postOffer;`, +); +const postOffer = make(); + +let closed = false; +if (closeAt !== null) setTimeout(() => { closed = true; }, closeAt); +const retries = []; +let outcome = null; +postOffer(call, 'https://hub.example/v1/nodes/n/webrtc/offer', {}, { + isClosed: () => closed, + onRetry: (status, delay) => retries.push({ status, delay }), +}).then(() => { outcome = { result: 'answered', at: now }; }) + .catch((e) => { outcome = { result: 'failed', at: now, status: e.status ?? null }; }); + +await run(runUntil); +console.log(JSON.stringify({ ...(outcome || { result: 'pending' }), posts, retries })); diff --git a/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs b/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs index c05e23b..5f95ede 100644 --- a/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs +++ b/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs @@ -80,7 +80,7 @@ const session = { bundleKey: 'k' }; const _loadBundleKey = async () => 'k'; // One group's index, answered on the clock rather than over a network. -const fetchGroupIndex = (groupId) => new Promise((resolve, reject) => { +const fetchGroupIndex = (_pool, groupId) => new Promise((resolve, reject) => { const n = Number(String(groupId).split('-')[1]); const isDead = dead.includes(n); setTimeout( @@ -123,7 +123,7 @@ const list = Array.from({ length: groups }, (_, i) => ({ id: `g-${i}`, name: `G$ let finishedAt = null; fetchAllIndexes( - list, 'tok', 'alice', 'u1', + null, list, 'tok', 'alice', 'u1', () => {}, // progress counters only (results) => { shown.push({ at: now, groups: results.size }); }, ).then(() => { finishedAt = now; }); diff --git a/packages/meshbay-hub/tests/harness/search_pool_harness.mjs b/packages/meshbay-hub/tests/harness/search_pool_harness.mjs new file mode 100644 index 0000000..521a97e --- /dev/null +++ b/packages/meshbay-hub/tests/harness/search_pool_harness.mjs @@ -0,0 +1,221 @@ +// Run the SHIPPED `ConnectionPool`, `fetchGroupIndex` and `fetchAllIndexes` +// against a fake clock and fake nodes, and count offers. +// +// The question this answers is how many connections Search negotiates, how many +// at once, and whether one it is reading from can be closed under it — the +// things that decided whether a phone ran into the hub's per-account ceilings. +// The pool, the index fetch and the sweep are lifted out of search-page.js as +// text; what is modelled is the clock and the nodes (`connectToGroup`, which +// is where an offer is made). +// +// Usage: node search_pool_harness.mjs <path to search-page.js> <json config> +import { readFileSync } from 'fs'; + +const src = readFileSync(process.argv[2], 'utf8'); +const cfg = JSON.parse(process.argv[3] || '{}'); + +const { + scenario = 'sweep', + groups = 5, + // Group indexes whose node never answers; `connectToGroup` fails after deadMs. + dead = [], + connectMs = 2000, // a phone on 4G + deadMs = 10000, + indexMs = 300, + runUntil = 600000, +} = cfg; + +// ── The clock ──────────────────────────────────────────────────────────────── + +let now = 0; +let nextId = 1; +let timers = []; +globalThis.setTimeout = (fn, ms) => { + const t = { at: now + (ms || 0), fn, id: nextId++ }; + timers.push(t); + return t.id; +}; +globalThis.clearTimeout = (id) => { timers = timers.filter((t) => t.id !== id); }; +// The pool orders its connections by `Date.now()`, so that is the fake clock too. +Date.now = () => now; +const flush = async () => { for (let i = 0; i < 50; i++) await Promise.resolve(); }; +async function run(until) { + await flush(); + while (timers.length) { + const due = timers.reduce((a, b) => (b.at < a.at ? b : a)); + if (due.at > until) break; + timers = timers.filter((t) => t !== due); + now = due.at; + due.fn(); + await flush(); + } + now = until; +} +const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + +// ── The nodes ──────────────────────────────────────────────────────────────── + +let offers = 0; +let negotiating = 0; +let peakNegotiating = 0; +let readsOnClosed = 0; +let pings = 0; +const transports = []; +// Groups whose held connection has died quietly, as a phone's does in its sleep. +const silentlyDead = new Set(); + +class FakeTransport { + constructor(groupId) { + this.groupId = groupId; + this.closedFlag = false; + transports.push(this); + } + get connected() { return !this.closedFlag; } + close() { this.closedFlag = true; } + async ping(timeoutMs) { + pings += 1; + if (silentlyDead.has(this.groupId)) { + await sleep(timeoutMs); + throw new Error('Response timeout'); + } + await sleep(100); + } + async fetchIndex() { + // Indexes differ in size, so reads end at different times and overlap + // other groups' negotiations — which uniform timings never do. + const n = Number(String(this.groupId).split('-')[1]); + await sleep(indexMs * (1 + (n % 4))); + if (this.closedFlag) { readsOnClosed += 1; throw new Error('Transport closed'); } + return { entries: [{ id: `e-${this.groupId}` }] }; + } +} + +async function connectToGroup(_hub, groupId) { + offers += 1; + negotiating += 1; + peakNegotiating = Math.max(peakNegotiating, negotiating); + try { + const n = Number(String(groupId).split('-')[1]); + if (dead.includes(n)) { + await sleep(deadMs); + throw new Error('Connection timeout'); + } + await sleep(connectMs); + return { transport: new FakeTransport(groupId), ack: {} }; + } finally { + negotiating -= 1; + } +} + +globalThis.window = {}; +let remembered = null; +globalThis.localStorage = { getItem: () => null, setItem: (_k, v) => { remembered = v; } }; +const session = { bundleKey: 'k' }; +const _loadBundleKey = async () => 'k'; +const cacheGroupIndex = () => {}; + +// ── The code under test, as text ───────────────────────────────────────────── + +const constant = (name) => { + const m = new RegExp(`^const ${name} = (\\d+);`, 'm').exec(src); + if (!m) throw new Error(`${name} is gone from search-page.js`); + return `const ${name} = ${m[1]};`; +}; +const lift = (signature) => { + const start = src.indexOf(signature); + if (start < 0) throw new Error(`${signature} is gone from search-page.js`); + return src.slice(start, src.indexOf('\n}\n', start) + 3); +}; + +const make = new Function( + 'connectToGroup', 'window', 'session', '_loadBundleKey', 'cacheGroupIndex', + 'localStorage', + `${constant('MAX_IN_FLIGHT')} + ${constant('MAX_POOL_SIZE')} + ${constant('REUSE_PING_MS')} + const DOWN_KEY = 'harness'; + ${lift('class ConnectionPool {')} + ${lift('async function fetchGroupIndex(')} + ${lift('function lastKnownDown(')} + ${lift('function rememberDown(')} + ${lift('async function inFlight(')} + ${lift('async function fetchAllIndexes(')} + return { ConnectionPool, fetchAllIndexes, MAX_IN_FLIGHT, MAX_POOL_SIZE };`, +); +const code = make(connectToGroup, globalThis.window, session, _loadBundleKey, + cacheGroupIndex, globalThis.localStorage); + +// ── The scenarios ──────────────────────────────────────────────────────────── + +const list = Array.from({ length: groups }, (_, i) => ({ id: `g-${i}`, name: `G${i}` })); +let evicted = 0; +const pool = new code.ConnectionPool('https://hub.example', () => { evicted += 1; }); +const out = { maxInFlight: code.MAX_IN_FLIGHT, maxPool: code.MAX_POOL_SIZE }; + +// Groups whose index has landed — the only ones with tiles on screen. +const shown = new Set(); +const sweep = async () => { + const r = await code.fetchAllIndexes(pool, list, 'tok', 'alice', 'u1', () => {}, + (results) => { for (const id of results.keys()) shown.add(id); }); + return { found: r.results.size, unreachable: r.unreachable.length, at: now }; +}; + +if (scenario === 'sweep') { + sweep().then((r) => { Object.assign(out, r); }); + await run(runUntil); +} else if (scenario === 'refresh') { + // A sweep, then the Search page's own refresh button: the second must reuse. + sweep().then(async () => { + out.offersFirst = offers; + const again = await sweep(); + out.offersSecond = offers - out.offersFirst; + Object.assign(out, again); + }); + await run(runUntil); +} else if (scenario === 'slept') { + // A sweep, then the phone sleeps and two connections die without saying so. + sweep().then(async () => { + out.offersFirst = offers; + silentlyDead.add('g-0'); + silentlyDead.add('g-1'); + const again = await sweep(); + out.offersSecond = offers - out.offersFirst; + Object.assign(out, again); + }); + await run(runUntil); +} else if (scenario === 'tiles') { + // A tile of every group asks for its connection while the sweep is on. + sweep().then((r) => { Object.assign(out, r); }); + for (const g of list) pool.connect(g.id, 'tok', 'k', 'alice', 'u1').catch(() => {}); + await run(runUntil); +} else if (scenario === 'scroll') { + // Tiles of groups already on screen keep using their connections while the + // sweep is still reading other groups' indexes — which makes a connection an + // index is being read from the least recently used one in the pool. + let sweeping = true; + sweep().then((r) => { sweeping = false; Object.assign(out, r); }); + const touch = async () => { + while (sweeping) { + await sleep(250); + for (const id of shown) { + if (pool.has(id)) pool.connect(id, 'tok', 'k', 'alice', 'u1').catch(() => {}); + } + } + }; + touch(); + await run(runUntil); +} else if (scenario === 'unmount') { + // The page goes away while connections are still being negotiated. + sweep().catch(() => {}); + await run(connectMs / 2); + pool.closeAll(); + await run(runUntil); + out.openAfter = transports.filter((t) => !t.closedFlag).length; +} + +Object.assign(out, { + offers, peakNegotiating, readsOnClosed, pings, evicted, poolSize: pool.size, + remembered: remembered === null ? null : JSON.parse(remembered), + openTransports: transports.filter((t) => !t.closedFlag).length, +}); +console.log(JSON.stringify(out)); diff --git a/packages/meshbay-hub/tests/test_offer_retry.py b/packages/meshbay-hub/tests/test_offer_retry.py new file mode 100644 index 0000000..3c968db --- /dev/null +++ b/packages/meshbay-hub/tests/test_offer_retry.py @@ -0,0 +1,78 @@ +""" +An offer the hub refuses for load is sent again; one refused for cause is not. + +A phone's Search reported groups as unreachable whose node was answering every +offer that reached it: the hub had refused those offers with 429, one of its +per-account ceilings, and the browser took that for the node. A 429 — or a 502 +or 503 while the hub restarts behind its proxy — says the hub is busy, so +`postOffer` waits (the hub's `Retry-After` when it gives one) and sends the same +offer again. A 404 for a node that is not connected, or a 504 for one that did +not answer, fails at once: retrying those would make a dead node cost time. + +These run the shipped `postOffer`, lifted out of transport.js as text, against +a fake hub and a fake clock — see harness/offer_retry_harness.mjs. +""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" +TRANSPORT = STATIC / "transport.js" +HARNESS = Path(__file__).parent / "harness" / "offer_retry_harness.mjs" + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not TRANSPORT.exists(), + reason="node or the SPA sources are not available") + + +def _post(**cfg) -> dict: + proc = subprocess.run( + ["node", str(HARNESS), str(TRANSPORT), json.dumps(cfg)], + capture_output=True, text=True) + assert proc.returncode == 0, proc.stderr + return json.loads(proc.stdout) + + +def test_a_busy_hub_is_asked_again_when_it_says(): + out = _post(answers=[[429, 1], 200]) + assert out["result"] == "answered" + assert out["posts"] == [0, 1000], "Retry-After was not honoured" + + +def test_without_retry_after_the_waits_grow(): + out = _post(answers=[429, 429, 503, 200]) + assert out["result"] == "answered" + assert out["posts"] == [0, 500, 1500, 3500] + + +def test_a_node_that_is_not_there_fails_at_once(): + for status in (404, 403, 504): + out = _post(answers=[status]) + assert out["result"] == "failed" + assert out["status"] == status + assert out["posts"] == [0], f"{status} was retried" + + +def test_a_hub_that_stays_busy_is_given_up_on(): + out = _post(answers=[429]) + assert out["result"] == "failed" + assert out["status"] == 429 + assert len(out["posts"]) == 6 + assert out["at"] <= 20000, "a busy hub cost more than the worst case stated" + + +def test_a_retry_after_that_asks_too_much_is_capped(): + out = _post(answers=[[429, 3600], 200]) + assert out["posts"] == [0, 10000] + + +def test_a_caller_that_gave_up_sends_nothing_more(): + """Search's deadline, or a page that went away, closes the transport while + it waits; the offer must not go out on its behalf afterwards.""" + out = _post(answers=[429], closeAt=100) + assert out["result"] == "failed" + assert out["posts"] == [0] diff --git a/packages/meshbay-hub/tests/test_search_fanout.py b/packages/meshbay-hub/tests/test_search_fanout.py index eda9465..196aa5a 100644 --- a/packages/meshbay-hub/tests/test_search_fanout.py +++ b/packages/meshbay-hub/tests/test_search_fanout.py @@ -22,17 +22,21 @@ reader for ten seconds, while two groups of the same batch had answered in two hundred milliseconds and nine others had not been dialled at all. Three things fix it, and each is asserted below: a result is drawn when it -arrives rather than when its neighbours finish; the ceiling of three is a +arrives rather than when its neighbours finish; the ceiling is a ceiling on *concurrency* rather than a batch, so a dead group holds one place instead of stalling a batch and everything queued behind it; and a group that did not answer last time is dialled last. The third is not a refinement of the second, it is what makes the steady state -correct. A ceiling alone still lets dead groups occupy every place at once — -with four of twelve down, five live groups appear straight away and the last -three wait out a deadline. Ordering by what was silent last time puts all eight -on screen in 600 ms. The cost is that a browser seeing these groups for the -first time has nothing to order by and pays the first sweep, once. +correct. A ceiling alone still lets dead groups occupy every place at once — as +many dead groups as there are places, listed first, and nothing is drawn until +a deadline expires. Ordering by what was silent last time puts them behind +every live group. The cost is that a browser seeing these groups for the first +time has nothing to order by and pays the first sweep, once. + +The ceiling is six (search-page.js, `MAX_IN_FLIGHT`). It was three, which was +also exactly what the hub admitted in flight per account, so the page ran into +the hub's refusals as soon as anything else connected beside the sweep. These run the shipped `fetchAllIndexes`, `inFlight`, `lastKnownDown` and `rememberDown`, lifted out of search-page.js as text, against a fake clock and a @@ -93,21 +97,17 @@ def test_several_nodes_down_do_not_delay_the_first_result(): "dead groups are being waited for one after another") -def test_on_a_first_visit_dead_groups_can_still_hold_the_last_few_back(): +def test_on_a_first_visit_four_dead_groups_hold_nothing_back(): """ - The limit of a ceiling, stated rather than glossed over. - - Four dead groups and three places: once all three are held by nodes that are - not answering, nothing else is dialled until one of them gives up. The first - results are immediate and most arrive at once, but the last few do wait — on - a browser that has never run this sweep before and so has nothing to order - by. The test after this one is the steady state, which is what a reader - actually lives in. + Four dead groups of twelve on a browser that has never swept before, so + nothing orders them. With three places they held every place at once and + the last three live groups waited out a deadline; with six, two places stay + free and every live group is on screen before any deadline expires. """ out = _sweep(groups=12, dead=[1, 4, 7, 10]) assert out["firstPaintAt"] == LIVE_MS early = [r for r in out["renders"] if r["at"] < DEAD_MS] - assert early[-1]["groups"] == 5, ( + assert early[-1]["groups"] == 8, ( "the shape of the first visit has changed — re-measure it rather than " "adjusting this number") @@ -128,19 +128,20 @@ def test_once_the_silent_groups_are_known_no_live_one_waits_for_them(): def test_a_group_that_did_not_answer_is_dialled_last_next_time(): """ - The last bad case, and the reason the browser remembers. + The limit of a ceiling, and the reason the browser remembers. - The three groups the hub lists first are all down, so on a first visit all - three places are held at once and there is nothing to draw until one frees - up. Told which groups were silent, the sweep puts them behind everything - else and the first result arrives on time. + The six groups the hub lists first are all down, so on a first visit every + place is held at once and there is nothing to draw until one frees up. Told + which groups were silent, the sweep puts them behind everything else and + the first result arrives on time. """ - first_visit = _sweep(groups=12, dead=[0, 1, 2]) + down = list(range(6)) + first_visit = _sweep(groups=12, dead=down) assert first_visit["firstPaintAt"] == DEAD_MS + LIVE_MS - assert first_visit["remembered"] == ["g-0", "g-1", "g-2"], ( + assert first_visit["remembered"] == [f"g-{n}" for n in down], ( "the sweep must record who was silent, or the next visit repeats this") - again = _sweep(groups=12, dead=[0, 1, 2], knownDown=[0, 1, 2]) + again = _sweep(groups=12, dead=down, knownDown=down) assert again["firstPaintAt"] == LIVE_MS @@ -149,7 +150,7 @@ def test_the_order_is_advisory_and_survives_a_browser_that_refuses_storage(): A private window throws on both halves of `localStorage`. The sweep must then behave exactly as a first visit does, not fail. """ - out = _sweep(groups=12, dead=[0, 1, 2], knownDown="refuses") + out = _sweep(groups=12, dead=list(range(6)), knownDown="refuses") assert out["firstPaintAt"] == DEAD_MS + LIVE_MS assert out["remembered"] is None assert out["finishedAt"] is not None, "the sweep did not finish" @@ -166,9 +167,9 @@ def test_a_node_that_comes_back_is_not_punished_for_ever(): def test_no_more_groups_are_dialled_at_once_than_the_ceiling(): """ - The ceiling is still a ceiling: twelve groups with three places and every - node down cost four deadlines, not one and not twelve. + The ceiling is still a ceiling: twelve groups with six places and every + node down cost two deadlines, not one and not twelve. """ out = _sweep(groups=12, dead=list(range(12))) - assert out["inFlight"] == 3 - assert out["finishedAt"] == 4 * DEAD_MS + assert out["inFlight"] == 6 + assert out["finishedAt"] == 2 * DEAD_MS diff --git a/packages/meshbay-hub/tests/test_search_pool.py b/packages/meshbay-hub/tests/test_search_pool.py new file mode 100644 index 0000000..67e66fa --- /dev/null +++ b/packages/meshbay-hub/tests/test_search_pool.py @@ -0,0 +1,108 @@ +""" +Search reaches each group with one offer, and never more than a few at a time. + +A phone on 4G reported one group of five missing from Search on half its +visits, and all but one on a fifth of them. The nodes were answering every +offer that reached them in about a second; the hub was refusing the others. The +page negotiated each group twice — the sweep opened a connection, read the +index, closed it, and the warm-up opened the same group again — and the sweep, +the warm-up and the tiles each kept a concurrency ceiling of their own, which +added up past what the hub admitted per account. Every refusal read as "node +unreachable". + +Now every connection goes through one `ConnectionPool`, which holds the only +ceiling and keeps what the sweep opened. These run the shipped pool, index +fetch and sweep, lifted out of search-page.js as text, against a fake clock and +fake nodes that take two seconds to answer — see harness/search_pool_harness.mjs. +""" + +import json +import shutil +import subprocess +from pathlib import Path + +import pytest + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" +SEARCH_PAGE = STATIC / "search-page.js" +HARNESS = Path(__file__).parent / "harness" / "search_pool_harness.mjs" + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not SEARCH_PAGE.exists(), + reason="node or the SPA sources are not available") + + +def _run(**cfg) -> dict: + proc = subprocess.run( + ["node", str(HARNESS), str(SEARCH_PAGE), json.dumps(cfg)], + capture_output=True, text=True) + assert proc.returncode == 0, proc.stderr + return json.loads(proc.stdout) + + +def test_each_group_costs_one_offer_and_stays_open(): + out = _run(scenario="sweep", groups=5) + assert out["found"] == 5 + assert out["offers"] == 5, "a group was negotiated more than once" + assert out["openTransports"] == 5, "the sweep closed what the tiles will need" + + +def test_twenty_groups_stay_within_the_ceiling_and_the_pool(): + out = _run(scenario="sweep", groups=20) + assert out["found"] == 20 + assert out["offers"] == 20 + assert out["peakNegotiating"] == out["maxInFlight"] + assert out["openTransports"] == out["maxPool"], "the pool kept more than its size" + + +def test_an_index_is_never_read_from_a_connection_evicted_under_it(): + """With more groups than the pool keeps, and tiles on screen using the + connections of groups already found, the connection a slow index is being + read from is the least recently used one there — the first to be evicted, + which would fail that group for a reason of this page's own making.""" + out = _run(scenario="scroll", groups=20, indexMs=5000) + assert out["readsOnClosed"] == 0 + assert out["found"] == 20 + + +def test_dead_nodes_cost_their_own_places_and_nothing_else(): + out = _run(scenario="sweep", groups=20, dead=[0, 3, 7]) + assert out["found"] == 17 + assert out["unreachable"] == 3 + # Three dead groups hold three of six places for one deadline, side by side. + assert out["at"] < 10000 + 20 * 2000 / 6 + 1000 + + +def test_tiles_asking_during_the_sweep_share_its_connections(): + """Tiles mount as soon as a group's index lands, while the sweep is still + dialling the others. They must neither negotiate a group a second time nor + add places of their own beside the sweep's.""" + out = _run(scenario="tiles", groups=20) + assert out["offers"] == 20 + assert out["peakNegotiating"] <= out["maxInFlight"] + + +def test_refreshing_reuses_every_connection_that_is_alive(): + out = _run(scenario="refresh", groups=5) + assert out["offersSecond"] == 0 + assert out["pings"] == 5, "a reused connection must prove it is alive first" + assert out["found"] == 5 + + +def test_a_connection_that_died_in_its_sleep_is_replaced_not_waited_on(): + """A phone that slept keeps reporting `connected` on channels that are gone. + Those are found by a short ping and renegotiated; the rest are reused.""" + out = _run(scenario="slept", groups=5) + assert out["offersSecond"] == 2 + assert out["found"] == 5 + assert out["unreachable"] == 0 + + +def test_leaving_the_page_leaves_no_connection_behind(): + """Negotiations still under way when the page closes its pool close what + they obtain, rather than adding it to a pool nobody will close again.""" + out = _run(scenario="unmount", groups=8) + assert out["openAfter"] == 0 + assert out["offers"] == out["maxInFlight"], "groups were dialled after the page closed" + assert out["remembered"] is None, ( + "groups the closed page never reached were remembered as down") diff --git a/packages/meshbay-hub/tests/test_transport_contracts.py b/packages/meshbay-hub/tests/test_transport_contracts.py index 3c2b0b6..28d6a10 100644 --- a/packages/meshbay-hub/tests/test_transport_contracts.py +++ b/packages/meshbay-hub/tests/test_transport_contracts.py @@ -531,13 +531,18 @@ def test_search_tries_every_node_the_hub_offers(): def test_search_connects_in_one_place(): - """Two call sites with their own connect is how one of them kept `nodes[0]`.""" + """Two call sites with their own connect is how one of them kept `nodes[0]`, + and how the sweep and the warm-up each negotiated the same group: every + connection goes through the pool, and only the pool calls `connectToGroup`.""" code = _code_only(SEARCH_PAGE.read_text(encoding="utf-8")) assert code.count("transport.connect(") == 1 + # The definition, and the pool's one call. + assert code.count("connectToGroup(") == 2 pool = code[code.index("async _doConnect("):code.index("_evict() {")] index = code[code.index("async function fetchGroupIndex("): code.index("async function fetchAllIndexes(")] - assert "connectToGroup(" in pool and "connectToGroup(" in index + assert "connectToGroup(" in pool + assert "pool.connect(" in index and "connectToGroup(" not in index # node:start only links an unlinked node key to the hub account when it is |