2 files changed, 173 insertions, 10 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js
index 7968284..a9559c6 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/app.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js
@@ -986,6 +986,24 @@ const CHUNK_SIZE = 1024 * 1024;
 // Seconds of already-watched video kept in the SourceBuffer, and the queue depth
 // past which we start making room before being forced to.
 const BUFFER_BEHIND_S = 60;
+// How far past the playhead we are willing to pull. The browser caps a video
+// SourceBuffer at a few hundred megabytes and refuses the append that goes
+// past, so "as fast as the network allows" is not a strategy for a film: the
+// node remuxes with `-c copy`, so a 500 MB file puts 500 MB on the wire, and a
+// ten-megabit second fills the ceiling in the first minute. Buffering by time
+// rather than by bytes keeps a two-hour film and a two-minute clip alike.
+const BUFFER_AHEAD_S = 90;
+// While we deliberately hold credit back, the node must still hear from us: its
+// own stall timeout is two minutes, and a paused film is not a gone viewer.
+const CREDIT_KEEPALIVE_MS = 20000;
+// Segments allowed in flight while there is room to put them. This is a window,
+// topped up as segments land, and not a debt released in one go: accumulating a
+// credit per append and handing the lot over when the buffer finally had room
+// sent 6 MB in a burst, overshot the target by a minute of film, and then said
+// nothing for the next forty-six seconds. Measured in Chrome against real
+// fragmented MP4. A stream that arrives in gulps has no margin for a network
+// that hesitates, and looks like a hang while it is quiet.
+const STREAM_WINDOW = 8;
 const QUEUE_HIGH_WATER = 12;
 const PIPELINE_WINDOW = 8;
 
@@ -2666,6 +2684,14 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
   const appendingRef = useRef(false);
   const endedRef = useRef(false);
   const durationRef = useRef(0);
+  // Segments the node is allowed to have in flight but has not sent yet, and
+  // when we last said anything to it at all.
+  const outstandingRef = useRef(0);
+  const lastPokeRef = useRef(0);
+  // Diagnostics reported to the node: how many appends the browser refused for
+  // want of room, and whether the element itself says it is starved.
+  const quotaRef = useRef(0);
+  const stalledRef = useRef(false);
 
   /**
    * Drop what has already been watched.
@@ -2689,6 +2715,18 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
     }
   }, []);
 
+  /** Seconds of film held past the playhead. */
+  const bufferedAhead = useCallback(() => {
+    const sb = sbRef.current;
+    const v = videoRef.current;
+    if (!sb || !v || !sb.buffered.length) return 0;
+    try {
+      return sb.buffered.end(sb.buffered.length - 1) - v.currentTime;
+    } catch {
+      return 0;
+    }
+  }, []);
+
   const flushQueue = useCallback(() => {
     const sb = sbRef.current;
     if (!sb || appendingRef.current || sb.updating) return;
@@ -2706,6 +2744,7 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
     } catch (e) {
       appendingRef.current = false;
       if (e.name === 'QuotaExceededError') {
+        quotaRef.current += 1;
         // The segment stays at the head of the queue and is tried again once
         // there is room. Dropping it — which is what this used to do — leaves a
         // hole in the middle of the film and no error anywhere.
@@ -2719,6 +2758,51 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
     }
   }, [evictBehind]);
 
+  /**
+   * Decide whether the node may send more, and keep the pipeline moving.
+   *
+   * This is the only place credit is granted, and the only thing that can
+   * restart a pipeline the buffer ceiling has stopped. That second job is why
+   * it exists: an append refused for quota fires no `updateend`, so it grants
+   * no credit, so the node sends nothing, so no segment arrives to call
+   * `flushQueue` again. Every wakeup the append path had was downstream of the
+   * append that just failed — the player deadlocked against itself and sat on
+   * "buffering" for good, which is what a 500 MB film did at around 100 MB.
+   *
+   * So the clock drives this, not the data.
+   */
+  const pump = useCallback(() => {
+    const transport = transportRef.current;
+    evictBehind();
+    flushQueue();
+    if (endedRef.current && queueRef.current.length === 0) return;
+    if (!transport || !transport.connected) return;
+
+    if (bufferedAhead() > BUFFER_AHEAD_S
+        || queueRef.current.length > QUEUE_HIGH_WATER) {
+      // Far enough ahead. Grant nothing, but do not go silent: two minutes of
+      // silence is how the node decides nobody is watching, and pausing a film
+      // for two minutes is an ordinary thing to do.
+      const now = Date.now();
+      if (now - lastPokeRef.current > CREDIT_KEEPALIVE_MS) {
+        lastPokeRef.current = now;
+        transport.grantStreamCredit(0);
+      }
+      return;
+    }
+
+    // Top the window back up to what is allowed in flight, rather than paying
+    // off everything owed at once. Called on every arriving segment as well as
+    // on the clock, so credit trickles out as room appears instead of being
+    // released in one gulp when the buffer finally drains.
+    const room = STREAM_WINDOW - outstandingRef.current;
+    if (room > 0) {
+      outstandingRef.current += room;
+      lastPokeRef.current = Date.now();
+      transport.grantStreamCredit(room);
+    }
+  }, [evictBehind, flushQueue, bufferedAhead]);
+
   useEffect(() => {
     let cancelled = false;
     // Reset here, not in the teardown of the run before: switching video while
@@ -2731,6 +2815,10 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
     appendingRef.current = false;
     endedRef.current = false;
     queueRef.current = [];
+    outstandingRef.current = 0;
+    lastPokeRef.current = Date.now();
+    quotaRef.current = 0;
+    stalledRef.current = false;
     const transport = transportRef.current;
     if (!transport || !transport.connected) {
       setError(t('video.err_transport'));
@@ -2738,6 +2826,9 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
       return;
     }
 
+    const onStarved = () => { stalledRef.current = true; pump(); };
+    const onFed = () => { stalledRef.current = false; };
+
     const onSeeking = () => {
       const v = videoRef.current;
       if (!v || !v.buffered.length) return;
@@ -2784,14 +2875,13 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
           sbRef.current = sb;
           sb.mode = 'sequence';
           sb.addEventListener('updateend', () => {
+            // No credit is granted here, deliberately. Appending is not the
+            // same question as having room, and tying the two meant `remove()`
+            // — which fires this event too — paid the node for the player's own
+            // evictions. What may be in flight is decided from the buffer, in
+            // pump(), and nowhere else.
             appendingRef.current = false;
-            // One segment consumed, so the node may send one more. The credit
-            // is granted here, after the append, because this is the point at
-            // which the memory is genuinely free again.
-            const transport = transportRef.current;
-            if (transport && transport.connected) transport.grantStreamCredit(1);
-            if (queueRef.current.length > QUEUE_HIGH_WATER) evictBehind();
-            flushQueue();
+            pump();
           });
           setPhase('streaming');
           flushQueue();
@@ -2800,6 +2890,13 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
         if (videoRef.current) {
           videoRef.current.src = url;
           videoRef.current.addEventListener('seeking', onSeeking);
+          videoRef.current.addEventListener('timeupdate', pump);
+          // The element's own verdict. "buffering" on screen is this, and it
+          // is the one thing the node cannot infer from a stream it is feeding.
+          videoRef.current.addEventListener('waiting', onStarved);
+          videoRef.current.addEventListener('stalled', onStarved);
+          videoRef.current.addEventListener('playing', onFed);
+          videoRef.current.addEventListener('canplay', onFed);
         }
       };
 
@@ -2810,11 +2907,18 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
         // would otherwise be decrypted against the wrong file — which fails,
         // loudly, in the console, for something that is simply not ours.
         if (msg.file_id && msg.file_id !== entry.id) return;
+        // One of the in-flight segments landed, whatever becomes of it: the
+        // window has room again. Counted before the decrypt so that a segment
+        // we fail to read still frees its slot rather than shrinking the
+        // window by one for the rest of the film.
+        outstandingRef.current = Math.max(0, outstandingRef.current - 1);
         try {
           const plaintext = await window.MeshBayCrypto.decryptChunkBin(
             gekRef.current, entry.id, msg.segment_index, msg.nonce, msg.ct);
           queueRef.current.push(plaintext);
-          flushQueue();
+          // pump(), not flushQueue(): arriving data is the moment to top the
+          // window back up, and that is what keeps the stream continuous.
+          pump();
         } catch (e) {
           console.error('[MSE] decrypt error:', e);
         }
@@ -2828,7 +2932,12 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
         flushQueue();
       };
 
-      transport.requestStream(entry.id);
+      // The opening window, and the count that tracks it. Asking for more here
+      // than pump() maintains would leave the node holding credit this side
+      // does not know about, which is the whole window's worth of overshoot on
+      // the very first breath of the stream.
+      outstandingRef.current = STREAM_WINDOW;
+      transport.requestStream(entry.id, STREAM_WINDOW);
     };
 
     // Closing the tab, or backgrounding it on a phone, never runs a React
@@ -2852,16 +2961,58 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
     window.addEventListener('pagehide', onPageHide);
     document.addEventListener('visibilitychange', onVisibility);
 
+    // `timeupdate` is silent while the film is paused, and the append path
+    // cannot wake itself once the ceiling has refused a segment. This is the
+    // clock that guarantees something is still driving the pipeline.
+    const pumpTimer = setInterval(pump, 1000);
+
+    // What the player sees, into the node's log. A hang on a phone shows the
+    // node feeding a stream quite happily; the half that says otherwise is in
+    // here, and there is no console to read it from.
+    const diagTimer = setInterval(() => {
+      const v = videoRef.current, sb = sbRef.current;
+      const t = transportRef.current;
+      if (!t || !v) return;
+      let ranges = '';
+      try {
+        for (let i = 0; sb && i < sb.buffered.length; i++) {
+          ranges += `${sb.buffered.start(i).toFixed(0)}-${sb.buffered.end(i).toFixed(0)} `;
+        }
+      } catch { ranges = '?'; }
+      t.sendStreamDiag({
+        t: +v.currentTime.toFixed(1),
+        ahead: +bufferedAhead().toFixed(1),
+        ranges: ranges.trim(),
+        ready: v.readyState,       // 0 = nothing, 4 = enough to play through
+        paused: v.paused,
+        stalled: stalledRef.current,
+        q: queueRef.current.length,
+        inflight: outstandingRef.current,
+        appending: appendingRef.current,
+        updating: sb ? sb.updating : null,
+        quota: quotaRef.current,
+        ms: msRef.current ? msRef.current.readyState : null,
+        err: v.error ? `${v.error.code}:${v.error.message}` : null,
+      });
+    }, 5000);
+
     startStream().catch(err => {
       if (!cancelled) { setError(err.message); setPhase('error'); }
     });
 
     return () => {
       cancelled = true;
+      clearInterval(pumpTimer);
+      clearInterval(diagTimer);
       window.removeEventListener('pagehide', onPageHide);
       document.removeEventListener('visibilitychange', onVisibility);
       if (videoRef.current) {
         videoRef.current.removeEventListener('seeking', onSeeking);
+        videoRef.current.removeEventListener('timeupdate', pump);
+        videoRef.current.removeEventListener('waiting', onStarved);
+        videoRef.current.removeEventListener('stalled', onStarved);
+        videoRef.current.removeEventListener('playing', onFed);
+        videoRef.current.removeEventListener('canplay', onFed);
       }
       if (transport) {
         // Tell the node first: dropping the handlers only makes us deaf, and a
@@ -2885,7 +3036,7 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
       sbRef.current = null;
       msRef.current = null;
     };
-  }, [entry, flushQueue]);
+  }, [entry, flushQueue, pump]);
 
   useEffect(() => {
     if (phase === 'streaming' && videoRef.current) {
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index 7a3943e..1b9abb3 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -590,6 +590,18 @@ class MeshBayTransport {
   }
 
   /**
+   * Tell the node what the player sees.
+   *
+   * A hang on a phone is unreadable from here: there is no console to open and
+   * the node's own log shows a stream it is feeding perfectly well. This puts
+   * the two halves in one file. The node only logs it.
+   */
+  sendStreamDiag(diag) {
+    if (!this._connected) return;
+    try { this._send({ type: 'client_diag', v: '0.1', ...diag }); } catch { /* gone */ }
+  }
+
+  /**
    * Nobody is watching any more.
    *
    * Closing the viewer used to say nothing to the node, which went on