diff options
Diffstat (limited to 'packages')
3 files changed, 413 insertions, 33 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 679b3f9..26d4e4c 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/search-page.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/search-page.js @@ -16,7 +16,10 @@ import { transfers } from './transfers.js'; import { mergeUnitEntries } from './source-merge.js'; import { useStickyBand } from './sticky.js'; -const BATCH_SIZE = 3; +// 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; // 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 @@ -214,40 +217,102 @@ async function fetchGroupIndex(groupId, token, bundleKey, username, userId) { } } -async function fetchAllIndexes(groups, token, username, userId, onProgress, onBatch) { +/** + * Run `fn` over `items`, `n` at a time, starting the next the moment one ends. + * + * Deliberately not batches. A batch waits for its slowest member before the + * next one starts, and a group whose node is down is always the slowest member + * there is: it costs the full connection deadline. Measured on twelve groups + * with four of them down — the shape a reader really has — batches of three + * showed the first result after ten seconds and finished after forty, because + * every dead group stalled its own batch and postponed all the ones behind it. + * A dead group here occupies one of `n` places and holds up nothing else. + */ +async function inFlight(items, n, fn) { + let next = 0; + const worker = async () => { + while (next < items.length) { + const i = next; + next += 1; + await fn(items[i], i); + } + }; + await Promise.all( + Array.from({ length: Math.min(n, items.length) }, worker)); +} + +// Which groups did not answer last time, so they can be dialled last. +// +// A ceiling of three still leaves one bad case: if the first three groups the +// hub lists are all down, all three places are held for the full deadline and +// nothing else starts. A node that is off tends to stay off, so the browser +// remembers and puts the known-silent ones at the back — where their deadline +// runs beside results that are already on screen instead of in front of them. +// +// Per browser and advisory: unreadable storage, a private window, a first visit +// or a stale list all degrade to the hub's own order, which is the behaviour +// without this. It is rewritten after every sweep, so a node coming back costs +// one sweep of being last and then returns to its place. +const DOWN_KEY = 'meshbay.search.down'; + +function lastKnownDown() { + try { + return new Set(JSON.parse(localStorage.getItem(DOWN_KEY) || '[]')); + } catch { + return new Set(); + } +} + +function rememberDown(ids) { + try { + localStorage.setItem(DOWN_KEY, JSON.stringify([...ids])); + } catch { /* private window, or storage refused: the order is advisory */ } +} + +async function fetchAllIndexes(groups, token, username, userId, onProgress, onResult) { const bundleKey = session.bundleKey || await _loadBundleKey(); if (bundleKey) session.bundleKey = bundleKey; const total = groups.length; let done = 0; const unreachable = []; + const downNow = new Set(); const results = new Map(); - for (let i = 0; i < groups.length; i += BATCH_SIZE) { - const batch = groups.slice(i, i + BATCH_SIZE); - await Promise.all(batch.map(async (g) => { - try { - const result = await fetchGroupIndex(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) { - results.set(g.id, { - ...result, - groupName: g.name, - groupOwner: g.owner_username, - }); - cacheGroupIndex(g.id, g.name, g.owner_username, result.entries, result.roots); - } else { - unreachable.push(g.name || g.id); - } - } catch { + // Stable, so groups that answered keep the hub's order among themselves. + const wasDown = lastKnownDown(); + const queue = [...groups].sort( + (a, b) => (wasDown.has(a.id) ? 1 : 0) - (wasDown.has(b.id) ? 1 : 0)); + + await inFlight(queue, MAX_IN_FLIGHT, async (g) => { + try { + const result = await fetchGroupIndex(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) { + results.set(g.id, { + ...result, + groupName: g.name, + groupOwner: g.owner_username, + }); + cacheGroupIndex(g.id, g.name, g.owner_username, result.entries, result.roots); + // Drawn now, not when this group's neighbours are done. Its index is + // already in hand; holding it back until a group that is not answering + // has finished not answering is ten seconds of blank page for work that + // completed in two hundred milliseconds. + onResult(results); + } else { unreachable.push(g.name || g.id); + downNow.add(g.id); } - done++; - onProgress({ done, total, unreachable: [...unreachable] }); - })); - onBatch(new Map(results)); - } + } catch { + unreachable.push(g.name || g.id); + downNow.add(g.id); + } + done++; + onProgress({ done, total, unreachable: [...unreachable] }); + }); + rememberDown(downNow); return { results, unreachable }; } @@ -478,13 +543,10 @@ function SearchPage({ token, username, userId, groups, onPlayQueue, userPrefs }) (async () => { const groupIds = [...indexedGroups.keys()].slice(0, MAX_POOL_SIZE); - for (let i = 0; i < groupIds.length; i += BATCH_SIZE) { - if (cancelled) break; - const batch = groupIds.slice(i, i + BATCH_SIZE); - await Promise.all(batch.map(async (gid) => { - try { await connectGroup(gid); } catch { /* skip */ } - })); - } + await inFlight(groupIds, MAX_IN_FLIGHT, async (gid) => { + if (cancelled) return; + try { await connectGroup(gid); } catch { /* skip */ } + }); })(); return () => { cancelled = true; }; @@ -762,7 +824,7 @@ function SearchPage({ token, username, userId, groups, onPlayQueue, userPrefs }) </div> `} - ${!fetching && progress.unreachable.length > 0 && html` + ${progress.unreachable.length > 0 && html` <p class="search-unreachable"> ${t('search.unreachable', { n: progress.unreachable.length })} </p> diff --git a/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs b/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs new file mode 100644 index 0000000..74b4bb3 --- /dev/null +++ b/packages/meshbay-hub/tests/harness/search_fanout_harness.mjs @@ -0,0 +1,144 @@ +// Run the SHIPPED `fetchAllIndexes` against a fake clock and report *when the +// reader sees something*. +// +// The question this answers is not how long the whole sweep takes. It is how +// long a group whose node is down makes everybody else wait — which is a +// property of the fan-out, not of any one connection's deadline. +// +// `fetchAllIndexes`, `inFlight` and the concurrency ceiling are lifted out of +// search-page.js as text +// and executed; what is modelled is the clock and one group's reachability. +// +// Usage: node search_fanout_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 { + groups = 12, + // Indexes (0-based, in the hub's listing order) of the groups whose node is + // down. They take `deadMs`; every other group answers in `liveMs`. + dead = [], + liveMs = 200, + deadMs = 10000, + // Group indexes the browser remembers as silent from a previous sweep. + knownDown = null, + runUntil = 600000, +} = cfg; + +// The browser's own store, modelled: a first visit has nothing, and a private +// window throws on both halves. +const refuses = knownDown === 'refuses'; +const stored = (knownDown === null || refuses) + ? null + : JSON.stringify(knownDown.map((n) => `g-${n}`)); +let written = null; +globalThis.localStorage = { + getItem: () => { + if (refuses) throw new Error('storage disabled'); + return stored; + }, + setItem: (_k, v) => { + if (refuses) throw new Error('storage disabled'); + written = v; + }, +}; + +// ── 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); }; + +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; +} + +// ── The environment ────────────────────────────────────────────────────────── + +const shown = []; // [{ at, groups }] — one entry per render the reader gets + +const session = { bundleKey: 'k' }; +const _loadBundleKey = async () => 'k'; +const cacheGroupIndex = () => {}; + +// One group's index, answered on the clock rather than over a network. +const fetchGroupIndex = (groupId) => new Promise((resolve, reject) => { + const n = Number(String(groupId).split('-')[1]); + const isDead = dead.includes(n); + setTimeout( + () => (isDead ? reject(new Error('Connection timeout')) + : resolve({ entries: [{ id: `e${n}` }], roots: {} })), + isDead ? deadMs : liveMs, + ); +}); + +// ── The code under test, as text ───────────────────────────────────────────── + +const ceiling = /^const MAX_IN_FLIGHT = (\d+);/m.exec(src); +if (!ceiling) throw new Error('MAX_IN_FLIGHT is gone from search-page.js'); + +// Both functions, because the sweep is the two of them: `inFlight` decides what +// runs when, and lifting only its caller would leave the harness measuring a +// concurrency it had supplied itself. +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( + 'session', '_loadBundleKey', 'cacheGroupIndex', 'fetchGroupIndex', 'localStorage', + `const MAX_IN_FLIGHT = ${ceiling[1]}; + const DOWN_KEY = 'harness'; + ${lift('function lastKnownDown(')} + ${lift('function rememberDown(')} + ${lift('async function inFlight(')} + ${lift('async function fetchAllIndexes(')} + return fetchAllIndexes;`, +); +const fetchAllIndexes = make( + session, _loadBundleKey, cacheGroupIndex, fetchGroupIndex, globalThis.localStorage); + +// ── The scenario ───────────────────────────────────────────────────────────── + +const list = Array.from({ length: groups }, (_, i) => ({ id: `g-${i}`, name: `G${i}` })); + +let finishedAt = null; +fetchAllIndexes( + list, 'tok', 'alice', 'u1', + () => {}, // progress counters only + (results) => { shown.push({ at: now, groups: results.size }); }, +).then(() => { finishedAt = now; }); + +await run(runUntil); + +console.log(JSON.stringify({ + // When the reader first sees any result at all, and how many. + firstPaintAt: shown.length ? shown[0].at : null, + firstPaintGroups: shown.length ? shown[0].groups : 0, + // Every render, so a test can see whether live groups waited on dead ones. + renders: shown, + finishedAt, + inFlight: Number(ceiling[1]), + // What the browser will remember for next time. + remembered: written === null ? null : JSON.parse(written), +})); diff --git a/packages/meshbay-hub/tests/test_search_fanout.py b/packages/meshbay-hub/tests/test_search_fanout.py new file mode 100644 index 0000000..eda9465 --- /dev/null +++ b/packages/meshbay-hub/tests/test_search_fanout.py @@ -0,0 +1,174 @@ +""" +A group whose node is down must not be something the reader waits for. + +Some group is always down — a node is a machine in somebody's house. Search +dials every group the reader belongs to, so "one of them is off tonight" is the +normal case and not the exception, and what it costs is the whole of how the +page feels. + +It used to cost everything. `fetchAllIndexes` went in batches of three and +waited for the slowest of each before starting the next, and it drew results +only once a batch was complete. Measured on twelve groups on a virtual clock, +with a live group answering in 200 ms and a dead one taking the full 10 s +deadline: + + all reachable first result 200 ms done 800 ms + one down, first batch first result 10 s done 10.6 s + four down, spread first result 10 s done 40 s + three down, together first result 10 s with nothing to show + +So a single node being off put a blank page and a progress bar in front of the +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 +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. + +These run the shipped `fetchAllIndexes`, `inFlight`, `lastKnownDown` and +`rememberDown`, lifted out of search-page.js as text, against a fake clock and a +fake browser store — see harness/search_fanout_harness.mjs. The numbers below are +the ones in the table, as assertions. +""" + +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_fanout_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") + +LIVE_MS = 200 # a node that answers +DEAD_MS = 10000 # a node that does not, to the connection deadline + + +def _sweep(**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_a_reachable_group_is_drawn_as_soon_as_it_answers(): + """Not when the two groups it happens to share a place in the queue with do.""" + out = _sweep(groups=12, dead=[]) + assert out["firstPaintAt"] == LIVE_MS + assert out["firstPaintGroups"] == 1, ( + "a result waited for others before being drawn") + + +def test_one_node_being_down_does_not_delay_the_first_result(): + """The reported case: a node down in what used to be the first batch.""" + out = _sweep(groups=12, dead=[1]) + assert out["firstPaintAt"] == LIVE_MS + + +def test_several_nodes_down_do_not_delay_the_first_result(): + """ + Four of twelve, spread out — the shape a reader with a few groups really + has. Batches made this ten seconds to the first result and forty to the end, + because each dead group stalled its own batch and postponed the rest. + """ + out = _sweep(groups=12, dead=[1, 4, 7, 10]) + assert out["firstPaintAt"] == LIVE_MS + assert out["finishedAt"] < 4 * DEAD_MS, ( + "dead groups are being waited for one after another") + + +def test_on_a_first_visit_dead_groups_can_still_hold_the_last_few_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. + """ + 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, ( + "the shape of the first visit has changed — re-measure it rather than " + "adjusting this number") + + +def test_once_the_silent_groups_are_known_no_live_one_waits_for_them(): + """ + The property that matters, in the state a reader is normally in: every group + that answers is on screen before any deadline has expired. + """ + out = _sweep(groups=12, dead=[1, 4, 7, 10], knownDown=[1, 4, 7, 10]) + live = 12 - 4 + early = [r for r in out["renders"] if r["at"] < DEAD_MS] + assert early and early[-1]["groups"] == live, ( + "a reachable group was not on screen until a deadline expired") + assert early[-1]["at"] <= 1000, ( + f"the live groups took {early[-1]['at']} ms to all appear") + + +def test_a_group_that_did_not_answer_is_dialled_last_next_time(): + """ + The last bad case, 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. + """ + first_visit = _sweep(groups=12, dead=[0, 1, 2]) + assert first_visit["firstPaintAt"] == DEAD_MS + LIVE_MS + assert first_visit["remembered"] == ["g-0", "g-1", "g-2"], ( + "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]) + assert again["firstPaintAt"] == LIVE_MS + + +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") + assert out["firstPaintAt"] == DEAD_MS + LIVE_MS + assert out["remembered"] is None + assert out["finishedAt"] is not None, "the sweep did not finish" + + +def test_a_node_that_comes_back_is_not_punished_for_ever(): + """ + The remembered list is rewritten from what this sweep saw, so a group that + answers again is no longer at the back on the visit after it. + """ + out = _sweep(groups=12, dead=[], knownDown=[1, 4]) + assert out["remembered"] == [], "a recovered node stayed on the silent list" + + +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. + """ + out = _sweep(groups=12, dead=list(range(12))) + assert out["inFlight"] == 3 + assert out["finishedAt"] == 4 * DEAD_MS |