fix(hub): keep the video read-ahead across a reconnection

A reconnection restarted the stream at the playhead, and that empties the
source buffer — spending the whole read-ahead at the one moment it was the
thing carrying the film. It carries on from the end of the buffer instead,
except where the buffer is short, the stream already ended, or a cast is
active: the receiver is fed from the wire and never had that buffer, so
resuming there would skip it forward by the lot.

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

1 files changed, 164 insertions, 9 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/static/video-player.js b/packages/meshbay-hub/src/meshbay_hub/static/video-player.js
index 0fed88c..6eaacc5 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/video-player.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/video-player.js
@@ -63,6 +63,10 @@ const STREAM_WINDOW = 8;
 // kills an ffmpeg and spawns another. Only where the finger stops is worth a
 // restart.
 const SEEK_DEBOUNCE_MS = 350;
+// How much read-ahead is worth carrying across a reconnection. Under this
+// there is nothing much to lose, and restarting at the playhead — the path
+// that has always run — is simpler and no worse.
+const RECONNECT_KEEP_MIN_S = 10;
 // A position is remembered per file, in this browser. Below the first threshold
 // there is nothing to resume; above the second the film is finished and
 // offering to resume thirty seconds before the credits is a nuisance.
@@ -224,6 +228,48 @@ function castSubtitleFor(sub, start) {
   };
 }
 
+/**
+ * What a reconnection should ask the node for, and on which terms.
+ *
+ * The old stream died with the connection, so something has to be asked for
+ * again; the question is *where*, and the answer used to be "the playhead",
+ * unconditionally. That goes through `reinitAt`, which empties the
+ * SourceBuffer — so a reconnection spent the entire read-ahead at the one
+ * moment it was worth most: the link is back, the film never stopped because
+ * the buffer was carrying it, and emptying the buffer is what finally stops
+ * it. Harmless while the read-ahead was ninety seconds and a dead spot had
+ * already drained it; with a budget of minutes it is the thing that stops the
+ * budget paying for anything.
+ *
+ * So where there is read-ahead worth keeping, carry on from the *end* of it
+ * and leave what is there alone. The two modes are not interchangeable:
+ * resuming keeps the buffer and must not move the playhead, seeking discards
+ * the buffer and must move it.
+ *
+ * Pure, and at module scope, so it can be tested by being run rather than by
+ * being read.
+ */
+function reconnectPlan(playhead, range, ended, castActive) {
+  // A stream that already ended has nothing left to fetch, and asking for a
+  // position at the end of the film would start an ffmpeg to serve nothing.
+  // Seeking is what this did before and is left exactly as it was.
+  //
+  // A cast is the same answer for a quite different reason: **the receiver
+  // does not share this buffer.** The relay is fed from segments as they
+  // arrive off the wire, so everything sitting in the SourceBuffer is material
+  // it never saw — carrying on from the end of it would restart the relay
+  // there and jump the receiver forward by the whole read-ahead, which with a
+  // budget of minutes is minutes of film silently skipped on somebody's
+  // television. What this protects is the *local* buffer; where playback is
+  // not local, restarting at the playhead is the correct answer and not merely
+  // the cautious one.
+  if (!ended && !castActive
+      && range && range[1] - playhead >= RECONNECT_KEEP_MIN_S) {
+    return { mode: 'resume', at: range[1] };
+  }
+  return { mode: 'seek', at: playhead };
+}
+
 function _mseSupported(codec) {
   if (!window.MediaSource) return false;
   const mime = `video/mp4; codecs="${codec}"`;
@@ -343,6 +389,10 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
   const awaitingInitRef = useRef(false);
   const seekTargetRef = useRef(null);
   const seekTimerRef = useRef(null);
+  // True between asking the node to carry the stream on from the end of the
+  // buffer and that request's `stream_init` arriving. It is what tells the two
+  // landings apart: one keeps the buffer, the other empties it.
+  const resumingRef = useRef(false);
   // The seek is built inside the effect, where the transport and `cancelled`
   // live; the render needs to reach it for "start from the beginning".
   const requestSeekRef = useRef(null);
@@ -663,6 +713,10 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
     // the old one the moment that much is buffered.
     awaitingInitRef.current = false;
     seekTargetRef.current = null;
+    // And the same trap once more: a resume asked for on the film being left
+    // would have the new film's first stream_init keep a buffer belonging to
+    // the old one, on a SourceBuffer built for neither.
+    resumingRef.current = false;
     clearTimeout(seekTimerRef.current);
     const transport = transportRef.current;
     if (!transport || !transport.connected) {
@@ -716,6 +770,10 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
           ranges: describeRanges(),
         });
         awaitingInitRef.current = true;
+        // A seek supersedes a resume that had been asked for and not landed:
+        // this one empties the buffer, and taking the resume branch on its
+        // `stream_init` would leave the film we navigated away from in place.
+        resumingRef.current = false;
         seekTargetRef.current = target;
         outstandingRef.current = STREAM_WINDOW;
         setPhase('loading');
@@ -726,6 +784,33 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
     requestSeekRef.current = requestSeek;
 
     /**
+     * Ask the node to carry the stream on from `target`, keeping the buffer.
+     *
+     * Not debounced, unlike `requestSeek`: the 350 ms there is for a finger on
+     * the scrubber, and a reconnection happens once. Nor does it show the
+     * loading screen — the film is still playing out of the buffer, and there
+     * is nothing for the viewer to wait for.
+     */
+    const requestResume = (target) => {
+      clearTimeout(seekTimerRef.current);
+      const t = transportRef.current;
+      if (cancelled || !t || !t.connected) return;
+      t.sendStreamDiag({
+        event: 'resume-request', target: +target.toFixed(1),
+        t: videoRef.current ? +videoRef.current.currentTime.toFixed(1) : null,
+        ready: videoRef.current ? videoRef.current.readyState : null,
+        offset: sbRef.current ? sbRef.current.timestampOffset : null,
+        ranges: describeRanges(),
+      });
+      resumingRef.current = true;
+      awaitingInitRef.current = true;
+      seekTargetRef.current = null;
+      outstandingRef.current = STREAM_WINDOW;
+      console.log('[resume] request', +target.toFixed(1));
+      t.requestStream(entry.id, STREAM_WINDOW, target, audioTrackRef.current);
+    };
+
+    /**
      * Move the playhead onto a seek once the data for it has arrived.
      *
      * Setting `currentTime` into a region that is not buffered yet leaves the
@@ -812,6 +897,65 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
       pump();
     };
 
+    /**
+     * Carry the stream on at `start`, keeping everything already buffered.
+     *
+     * `reinitAt`'s counterpart, for the one case where the buffer is the thing
+     * worth saving rather than the thing in the way — see `reconnectPlan`. The
+     * node lands on a keyframe at or before what was asked for, so the new
+     * material overlaps the tail of what is there and the coded frame
+     * processing replaces it; the range stays continuous and the playhead
+     * never notices.
+     */
+    const resumeAt = async (start) => {
+      const sb = sbRef.current;
+      if (!sb) return;
+      // If the node could not start at or before where the buffer ends, the
+      // new material would not join what is there and the film would stall on
+      // the gap for good. It should not happen — a keyframe at or before the
+      // request, and clamping only ever moves it earlier — but a permanent
+      // silent stall is not a risk worth taking on reasoning alone, and the
+      // old path is right there and proven.
+      const range = currentRange();
+      if (!range || start > range[1] + 0.5) {
+        console.warn('[resume] node started at', start, 'past the buffer end',
+                     range ? range[1] : null, '— falling back to reinit');
+        return reinitAt(start);
+      }
+      // Same reason as reinitAt: ffmpeg was killed mid-fragment, so the parser
+      // holds half of one and the next stream's header on top of that is a
+      // decode error. `abort()` resets the parser and leaves what is buffered
+      // alone, which is exactly what this path needs.
+      try { sb.abort(); } catch { /* not in a state that needs it */ }
+      await settled(sb);
+      // ffmpeg restarts its timestamps at zero however far in it was asked to
+      // seek, so this is what puts the new fragments back on the film's
+      // timeline — beside the ones already there rather than on top of them.
+      try { sb.timestampOffset = start; } catch { /* older browsers */ }
+      const tr = transportRef.current;
+      if (tr) {
+        tr.sendStreamDiag({
+          event: 'resume', target: start, offset: sb.timestampOffset,
+          t: videoRef.current ? +videoRef.current.currentTime.toFixed(1) : null,
+          ranges: describeRanges(),
+        });
+      }
+      queueRef.current = [];
+      appendingRef.current = false;
+      endedRef.current = false;
+      awaitingInitRef.current = false;
+      // No seek target. The playhead is already where it belongs and has not
+      // stopped; setting one would have `landPlayhead` jump the film forward
+      // to the end of the buffer — over the very minutes this path exists to
+      // keep. `quotaHold` is left alone for the mirror-image reason: nothing
+      // was emptied, so a buffer that had no room still has none.
+      seekTargetRef.current = null;
+      console.log('[resume] resumeAt done, start:', start,
+                  'kept:', describeRanges());
+      setPhase('streaming');
+      pump();
+    };
+
     const onSeeking = () => {
       if (landingPlayheadRef.current) {
         landingPlayheadRef.current = false;
@@ -850,19 +994,19 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
       // The old stream died with the connection (the node retires it the
       // moment its session goes away — see webrtc_server.py's
       // on_state_change), so there is nothing to resume on the wire, only a
-      // reason to ask again. requestSeek already knows how to land a new
-      // stream_init on the live SourceBuffer without resetting playback —
-      // exactly what dragging the scrubber does — so reusing it here means a
-      // screen-lock reconnect looks like a seek to where the film already
-      // was, not a reload.
+      // reason to ask again. Where to ask from is `reconnectPlan`: the end of
+      // the buffer when there is one worth keeping, the playhead otherwise.
       offReconnect = transport.addReconnectListener(() => {
         if (cancelled) return;
         const v = videoRef.current;
         const seek = requestSeekRef.current;
         if (!v || !seek) return;
-        console.log('[MeshBay] transport reconnected — resuming stream at',
-                    v.currentTime.toFixed(1));
-        seek(v.currentTime);
+        const plan = reconnectPlan(v.currentTime, currentRange(),
+                                   endedRef.current, castActiveRef.current);
+        console.log('[MeshBay] transport reconnected —', plan.mode, 'at',
+                    plan.at.toFixed(1), 'from', v.currentTime.toFixed(1));
+        if (plan.mode === 'resume') requestResume(plan.at);
+        else seek(plan.at);
       });
 
       transport.onStreamInit = (msg) => {
@@ -915,12 +1059,18 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
         if (sbRef.current && msRef.current
             && msRef.current.readyState === 'open') {
           console.log('[seek] stream_init landed, start:', msg.start, 'awaitingInit:', awaitingInitRef.current);
+          // Read and cleared together: whichever landing runs, the next
+          // `stream_init` is a fresh question and must not inherit this one's
+          // answer.
+          const resuming = resumingRef.current;
+          resumingRef.current = false;
           initSegmentRef.current = null;
           if (castActiveRef.current && platform.cast.available) {
             platform.cast.stop().catch(() => {});
             castRestartPendingRef.current = true;
           }
-          reinitAt(msg.start || 0).catch(() => {
+          const land = resuming ? resumeAt : reinitAt;
+          land(msg.start || 0).catch(() => {
             setError(t('video.err_transport'));
             setPhase('error');
           });
@@ -934,6 +1084,11 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose, onDownload }) {
         awaitingInitRef.current = false;
         endedRef.current = false;
         appendingRef.current = false;
+        // This branch builds a new SourceBuffer, so there is no buffer left to
+        // carry anything on from — whatever asked for a resume has had its
+        // answer, and a flag left standing here would send the *next*
+        // stream_init down a path whose precondition is gone.
+        resumingRef.current = false;
         queueRef.current = [];
         sbRef.current = null;
         initSegmentRef.current = null;