2 files changed, 318 insertions, 0 deletions
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