fix(hub): a group whose node is down is no longer something the reader waits for

A node is a machine in somebody's house, so with a handful of groups one is
always off. Search treated that as the exception and charged the reader for it.

Measured on twelve groups against a virtual clock, a live group answering in
200 ms and a dead one taking the full deadline: one node down in the first batch
put a blank page and a progress bar in front of the reader for **ten seconds**,
while two groups of that same batch had answered in two hundred milliseconds and
nine others had not been dialled at all. Four down, spread out — the shape a real
reader has — was ten seconds to the first result and **forty** to the last.

Two causes, neither of them the connection deadline. Results were drawn once a
batch was complete rather than as each arrived, so an index already in hand waited
on a node that was not answering. And the batches were sequential, so a dead group
did not merely cost its own deadline, it postponed every group behind it.

So the three is a ceiling on concurrency and never a batch — `inFlight` starts the
next group the moment one ends — and each index is drawn when it lands.

That alone is not enough, and the tests say where it stops: a ceiling still lets
silent nodes hold every place at once, and with four of twelve down the last three
live groups still waited out a deadline. So the browser remembers which groups
were silent and dials them last, which puts all eight on screen in 600 ms. The
list is advisory and rewritten from what each sweep saw: a private window, storage
that refuses, a first visit or a node that has come back all fall through to the
hub's own order, cost one sweep, and correct themselves.

The "n groups unreachable" line also waited for the sweep to finish, which is the
one moment it is no longer needed. It now appears as they are found.

First result, before and after, twelve groups: one down 10 s → 200 ms; four down
10 s → 200 ms with every reachable group on screen by 600 ms; three down and
listed first 10 s → 200 ms from the second visit on.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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