fix(hub): bound the video read-ahead by the playhead, not by the network

A 500 MB film loaded about 100 MB and hung on "buffering" for good. 100 MB is
not a number in our code: it is where the browser stops. ffmpeg remuxes with
`-c copy`, so the bytes on the wire are the file's own, and credit granted per
append meant taking them as fast as the network allowed — which for a film is
very much faster than watching it. The SourceBuffer ceiling arrived in the
first minute.

Past it every append was refused, and the refusal was unrecoverable: a refused
append fires no `updateend`, `updateend` was where credit was granted, so the
node sent nothing and no segment arrived to retry the append. Every wakeup the
pipeline had was downstream of the append that had just failed. Playback
continuing — the one thing that frees room — woke nothing at all.

Credit now follows the buffer instead of the writes. `pump()` is the only
place it is granted, it keeps `STREAM_WINDOW` segments in flight while less
than `BUFFER_AHEAD_S` of film is held past the playhead, and it is driven by a
one-second clock and by playback progress, never by arriving data. Buffering
by time makes a two-hour film cost what a two-minute clip costs.

A window rather than a debt, and this took a second measurement to get right:
accumulating a credit per append and releasing the balance when the buffer
finally drained sent six megabytes in one burst, overshot by a minute of film,
then said nothing for forty-six seconds. Measured in Chrome against real
fragmented MP4.

Two smaller things found on the way. `updateend` fires for `remove()` as well
as `appendBuffer()`, so crediting from it paid the node for the player's own
evictions. And a viewer that is deliberately far enough ahead grants nothing
for minutes, which the node read as a closed tab — it now sends `stream_more`
with n=0, which grants no room but proves someone is there.

The first version of the test modelled the credit loop and passed while the
player still hung: a model written by whoever wrote the fix agrees with it by
construction. `tests/harness/mse_harness.mjs` lifts the real functions out of
app.js as text and runs them against a SourceBuffer that has a ceiling. What is
modelled is the browser.

1 files changed, 159 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/harness/mse_harness.mjs b/packages/meshbay-hub/tests/harness/mse_harness.mjs
new file mode 100644
index 0000000..0daacfb
--- /dev/null
+++ b/packages/meshbay-hub/tests/harness/mse_harness.mjs
@@ -0,0 +1,159 @@
+// Run the SHIPPED player functions against a fake SourceBuffer.
+//
+// The point is that nothing here is a paraphrase of app.js: `bufferedAhead`,
+// `evictBehind`, `flushQueue` and `pump` are lifted out of the file as text and
+// executed. A model of a fix, written by whoever wrote the fix, agrees with it
+// by construction — which is how a passing test sat next to a player that still
+// hung. What is modelled here is the *browser*: a SourceBuffer with a ceiling,
+// and `updateend` firing for removals as well as appends.
+//
+// Usage: node mse_harness.mjs <path to app.js> <json config>
+import { readFileSync } from 'fs';
+
+const app = readFileSync(process.argv[2], 'utf8');
+const cfg = JSON.parse(process.argv[3] || '{}');
+
+const {
+  playing = false,          // does the viewer actually press play
+  capMB = 100,              // where the browser refuses the append
+  fileMB = 493.5,           // the film, from a real upload
+  durationS = 3936,
+  netMBs = 35,              // measured node throughput
+  wallS = 600,
+} = cfg;
+
+const grab = (name) => {
+  const start = app.indexOf(`const ${name} = useCallback(`);
+  if (start < 0) throw new Error(`${name} not found in app.js`);
+  const deps = app.indexOf('\n  }, [', start);
+  const end = app.indexOf(');', deps) + 2;
+  return app.slice(start, end);
+};
+
+const useCallback = (fn) => fn;
+const src = ['bufferedAhead', 'evictBehind', 'flushQueue', 'pump']
+  .map(grab).join('\n');
+
+const SEG = 256 * 1024;
+const CAP = capMB * 1048576;
+const BITRATE = fileMB * 1048576 / durationS;
+
+// Read from app.js too, so a change to the constants is a change to the test.
+const constOf = (name) => {
+  const m = app.match(new RegExp(`const ${name} = (\\d+)`));
+  if (!m) throw new Error(`${name} not found`);
+  return Number(m[1]);
+};
+const BUFFER_BEHIND_S = constOf('BUFFER_BEHIND_S');
+const STREAM_WINDOW = constOf('STREAM_WINDOW');
+const BUFFER_AHEAD_S = constOf('BUFFER_AHEAD_S');
+const QUEUE_HIGH_WATER = constOf('QUEUE_HIGH_WATER');
+const CREDIT_KEEPALIVE_MS = constOf('CREDIT_KEEPALIVE_MS');
+
+let bytes = 0, ranges = [], appended = 0, removes = 0;
+let granted = 0, keepalives = 0, sent = 0, credit = 0, quotaRefusals = 0;
+
+const sb = {
+  updating: false,
+  get buffered() {
+    return {
+      get length() { return ranges.length; },
+      start: (i) => ranges[i][0],
+      end: (i) => ranges[i][1],
+    };
+  },
+  appendBuffer(chunk) {
+    if (bytes + chunk.byteLength > CAP) {
+      quotaRefusals++;
+      const e = new Error('quota'); e.name = 'QuotaExceededError'; throw e;
+    }
+    const at = ranges.length ? ranges[ranges.length - 1][1] : 0;
+    ranges.push([at, at + chunk.byteLength / BITRATE]);
+    bytes += chunk.byteLength;
+    appended++;
+  },
+  remove(a, b) {
+    removes++;
+    let dropped = 0;
+    ranges = ranges.filter(([s, e]) => {
+      if (e <= b && s >= a) { dropped += (e - s) * BITRATE; return false; }
+      return true;
+    });
+    bytes -= dropped;
+    // A real remove() is asynchronous and fires updateend when it lands. That
+    // event is indistinguishable from an append's unless the player kept track.
+    pendingRemoveEvents++;
+  },
+};
+let pendingRemoveEvents = 0;
+
+const video = { currentTime: 0 };
+const sbRef = { current: sb }, videoRef = { current: video };
+const msRef = { current: { readyState: 'open', endOfStream() {} } };
+const queueRef = { current: [] };
+const appendingRef = { current: false }, endedRef = { current: false };
+const outstandingRef = { current: 0 }, lastPokeRef = { current: 0 };
+const quotaRef = { current: 0 }, stalledRef = { current: false };
+const transportRef = {
+  current: {
+    connected: true,
+    grantStreamCredit(n) {
+      if (n === 0) { keepalives++; return; }
+      granted += n; credit += n;
+    },
+  },
+};
+
+const fns = new Function(
+  'sbRef,videoRef,msRef,queueRef,appendingRef,endedRef,outstandingRef,' +
+  'lastPokeRef,transportRef,BUFFER_BEHIND_S,BUFFER_AHEAD_S,QUEUE_HIGH_WATER,' +
+  'CREDIT_KEEPALIVE_MS,STREAM_WINDOW,quotaRef,stalledRef,console,useCallback',
+  src + '\n return {bufferedAhead, evictBehind, flushQueue, pump};'
+)(sbRef, videoRef, msRef, queueRef, appendingRef, endedRef, outstandingRef,
+  lastPokeRef, transportRef, BUFFER_BEHIND_S, BUFFER_AHEAD_S, QUEUE_HIGH_WATER,
+  CREDIT_KEEPALIVE_MS, STREAM_WINDOW, quotaRef, stalledRef, console, useCallback);
+
+// The player's own `updateend` listener, transcribed — the one part of the
+// component that is a listener rather than a callback, and the place the
+// remove/append distinction lives.
+const updateend = () => {
+  appendingRef.current = false;
+  fns.pump();
+};
+
+credit = STREAM_WINDOW;
+outstandingRef.current = STREAM_WINDOW;
+let wall = 0;
+const TICK = 0.05;
+while (wall < wallS) {
+  wall += TICK;
+  if (playing) {
+    const end = ranges.length ? ranges[ranges.length - 1][1] : 0;
+    video.currentTime = Math.min(video.currentTime + TICK, end);
+  }
+  fns.pump();                                    // the 1 s timer and timeupdate
+  while (pendingRemoveEvents > 0) { pendingRemoveEvents--; updateend(); }
+
+  let budget = netMBs * 1048576 * TICK;
+  while (credit > 0 && budget >= SEG) {
+    credit--; budget -= SEG; sent++;
+    outstandingRef.current = Math.max(0, outstandingRef.current - 1);
+    queueRef.current.push({ byteLength: SEG });
+    fns.flushQueue();
+    if (appendingRef.current) updateend();
+    while (pendingRemoveEvents > 0) { pendingRemoveEvents--; updateend(); }
+  }
+}
+
+console.log(JSON.stringify({
+  sentMB: +(sent * SEG / 1048576).toFixed(1),
+  heldInBufferMB: +(bytes / 1048576).toFixed(1),
+  queueDepth: queueRef.current.length,
+  bufferedAheadS: +fns.bufferedAhead().toFixed(1),
+  watchedS: +video.currentTime.toFixed(1),
+  grants: granted,
+  keepalives,
+  removes,
+  quotaRefusals: quotaRef.current,
+  hitCeiling: bytes >= CAP * 0.99,
+}));