diff options
Diffstat (limited to 'packages')
| -rw-r--r-- | packages/meshbay-hub/src/meshbay_hub/static/video-player.js | 173 | ||||
| -rw-r--r-- | packages/meshbay-hub/tests/test_video_reconnect.py | 252 |
2 files changed, 416 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; diff --git a/packages/meshbay-hub/tests/test_video_reconnect.py b/packages/meshbay-hub/tests/test_video_reconnect.py new file mode 100644 index 0000000..beeeace --- /dev/null +++ b/packages/meshbay-hub/tests/test_video_reconnect.py @@ -0,0 +1,252 @@ +""" +A reconnection used to throw the read-ahead away. + +The transport drops — a tunnel, a dead spot, a screen lock — and comes back. +The old stream died with it, so something has to be asked for again, and what +the player asked for was always the same thing: the stream, restarted at the +playhead. That goes through `reinitAt`, whose whole job is to empty the +SourceBuffer, because a seek has to. + +But a reconnection is not a seek. The viewer did not navigate anywhere; the +film has been playing out of the buffer the entire time the link was gone, and +the buffer is the only reason it was still playing. Emptying it at the moment +the link returns is what finally stops the picture — and it is the *one* moment +the buffer is worth most. + +This was invisible while the read-ahead was a fixed ninety seconds: a dead spot +long enough to notice had already drained the buffer, so there was nothing left +to throw away. Once the read-ahead became a byte budget worth minutes +(`test_video_buffer_ceiling.py`), it became the thing that stops the budget +paying for anything at all — several minutes of lead, bought and then discarded +by the reconnection it was bought for. + +So `reconnectPlan` decides, and there are two modes that are not +interchangeable: + +**resume** keeps the buffer and carries the stream on from the *end* of it. The +playhead must not move — it is already in the right place and still running. + +**seek** discards the buffer and restarts at the playhead. The playhead must +move. This is the path that has always run, and it is still what happens when +there is no read-ahead worth keeping, or when the stream already ended. + +`reconnectPlan` is pure and at module scope so that this file can *run* it +rather than read it. The rest is read, which is weak evidence and the only +evidence available for code that lives inside a component's effect. +""" + +import json +import re +import shutil +import subprocess +from pathlib import Path + +import pytest + +STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" +APP = STATIC / "video-player.js" + +pytestmark = pytest.mark.skipif( + shutil.which("node") is None or not APP.exists(), + reason="node or the SPA sources are not available") + + +@pytest.fixture(scope="module") +def app(): + return APP.read_text() + + +def _player(app: str) -> str: + i = app.index("function VideoPlayer(") + nxt = app.find("\nfunction ", i + 1) + return app[i:nxt if nxt > 0 else len(app)] + + +def _fn(player: str, name: str) -> str: + """One `const <name> = ...` defined inside the effect, as text.""" + i = player.index(f"const {name} = ") + return player[i:player.index("\n };", i)] + + +# ── The decision, executed ──────────────────────────────────────────────────── + +def _plan(app: str, cases: list[dict]) -> list[dict]: + """Run the SHIPPED `reconnectPlan` over a table of cases. + + Lifted out as text and executed, in the manner of harness/mse_harness.mjs: + a second implementation written here would agree with the first by + construction and prove nothing. + """ + i = app.index("function reconnectPlan(") + body = app[i:app.index("\n}\n", i) + 3] + const = re.search(r"^const RECONNECT_KEEP_MIN_S = .+;$", app, re.M) + assert const, "RECONNECT_KEEP_MIN_S is not a constant in video-player.js" + script = ( + f"{const.group(0)}\n{body}\n" + f"console.log(JSON.stringify({json.dumps(cases)}.map(" + "(c) => reconnectPlan(c.playhead, c.range, c.ended, c.cast))));" + ) + proc = subprocess.run(["node", "--input-type=module", "-e", script], + capture_output=True, text=True) + assert proc.returncode == 0, proc.stderr + return json.loads(proc.stdout) + + +def test_a_buffer_worth_keeping_is_carried_on_from_its_end(app): + """The whole point: the minutes already paid for survive the reconnection.""" + plan, = _plan(app, [{"playhead": 100, "range": [40, 400], "ended": False}]) + assert plan["mode"] == "resume", ( + "five minutes of read-ahead in hand and the reconnection still throws " + "it away") + assert plan["at"] == 400, ( + f"resuming at {plan['at']} rather than the end of the buffer — " + "anything earlier re-fetches film already held") + + +def test_an_empty_buffer_still_restarts_at_the_playhead(app): + """The old path, unchanged, for the case it was written for.""" + plan, = _plan(app, [{"playhead": 100, "range": None, "ended": False}]) + assert plan == {"mode": "seek", "at": 100} + + +def test_a_buffer_too_short_to_matter_takes_the_proven_path(app): + """Below the threshold there is nothing to save and no reason to be clever. + + Resuming buys a couple of seconds and costs a second code path through a + SourceBuffer that has just been reconnected; seeking is proven. + """ + keep = float(re.search(r"const RECONNECT_KEEP_MIN_S = (\d+)", + app).group(1)) + below, at_threshold = _plan(app, [ + {"playhead": 100, "range": [40, 100 + keep - 1], "ended": False}, + {"playhead": 100, "range": [40, 100 + keep], "ended": False}, + ]) + assert below["mode"] == "seek", ( + f"{keep - 1}s of read-ahead took the resume path") + assert at_threshold["mode"] == "resume", ( + f"exactly {keep}s of read-ahead did not, so the threshold excludes " + "its own boundary") + + +def test_a_cast_restarts_at_the_playhead_because_it_never_had_the_buffer(app): + """The receiver is fed from the wire, not from this SourceBuffer. + + `platform.cast.push` hands the relay each segment as it arrives, so + everything held in the SourceBuffer is material the receiver never saw. + Carrying on from the end of it would restart the relay there and skip the + television forward by the entire read-ahead — with a budget of minutes, + minutes of film silently missed. + """ + plan, = _plan(app, [{"playhead": 100, "range": [40, 400], + "ended": False, "cast": True}]) + assert plan == {"mode": "seek", "at": 100} + + +def test_a_finished_stream_is_not_restarted_to_serve_nothing(app): + """`endedRef` means the node already sent everything. + + Asking to carry on from the end of the buffer would spawn an ffmpeg to + serve a position at or past the end of the film. Seeking is what this did + before the plan existed and is deliberately left alone. + """ + plan, = _plan(app, [{"playhead": 100, "range": [40, 400], "ended": True}]) + assert plan["mode"] == "seek" + + +# ── Resuming must not do a seek's work ──────────────────────────────────────── + +def test_resuming_does_not_empty_the_buffer(app): + """The one thing this whole change exists to stop.""" + resume = _fn(_player(app), "resumeAt") + assert "sb.remove(0, Infinity)" not in resume, ( + "resumeAt empties the SourceBuffer, which is exactly what it was " + "written to avoid") + assert "sb.abort()" in resume, ( + "ffmpeg was killed mid-fragment, so the parser holds half of one and " + "the next stream's header lands on top of it") + assert "sb.timestampOffset = start" in resume, ( + "the new fragments are not placed on the film's timeline, so they " + "land at zero instead of beside what is already buffered") + + +def test_resuming_does_not_move_the_playhead(app): + """`landPlayhead` jumps the film to `seekTarget` once it is buffered. + + A resume that set one would jump forward to the end of the buffer — over + the very minutes it just went to the trouble of keeping. + """ + resume = _fn(_player(app), "resumeAt") + assert "seekTargetRef.current = null" in resume, ( + "resumeAt sets a seek target, so the film jumps to the end of the " + "buffer instead of playing through it") + assert "seekTargetRef.current = start" not in resume + + +def test_a_resume_that_would_leave_a_gap_falls_back(app): + """A gap in the middle of a film is a permanent silent stall. + + The node lands on a keyframe at or before what was asked for, and clamping + only ever moves it earlier, so this should not happen. "Should not happen" + is not the same as "cannot", and the proven path is one line away. + """ + resume = _fn(_player(app), "resumeAt") + assert "reinitAt(start)" in resume, ( + "nothing catches a stream that starts past the end of the buffer, so " + "the film stalls on the gap for the rest of its run") + + +def test_the_reconnection_asks_the_plan_rather_than_assuming(app): + player = _player(app) + i = player.index("addReconnectListener(") + listener = player[i:player.index("\n });", i)] + assert "reconnectPlan(" in listener, ( + "the reconnection still restarts at the playhead unconditionally") + assert "requestResume(" in listener, "nothing ever takes the resume path" + assert "seek(plan.at)" in listener, "nothing takes the seek path any more" + + +def test_resuming_does_not_show_the_loading_screen(app): + """The film never stopped, so there is nothing to tell the viewer to await.""" + resume_req = _fn(_player(app), "requestResume") + assert "setPhase('loading')" not in resume_req, ( + "a reconnection that the buffer covered still blanks the picture for " + "a loading screen") + + +def test_the_resume_is_not_debounced(app): + """The 350 ms exists for a finger on a scrubber; a reconnection happens once.""" + resume_req = _fn(_player(app), "requestResume") + assert "SEEK_DEBOUNCE_MS" not in resume_req, ( + "the resume waits out the scrubber debounce before asking, so the " + "stream restarts a third of a second later than it could") + + +# ── The flag that tells the two landings apart ──────────────────────────────── + +def test_the_two_landings_are_told_apart_and_the_flag_is_cleared(app): + player = _player(app) + init = player[player.index("transport.onStreamInit = "):] + init = init[:init.index("\n };")] + assert "const resuming = resumingRef.current;" in init + assert "resumingRef.current = false;" in init, ( + "the flag outlives the stream_init it was raised for, so the next one " + "keeps a buffer nobody asked it to keep") + assert "resuming ? resumeAt : reinitAt" in init, ( + "both landings still go through the same path") + + +def test_a_new_stream_and_a_seek_both_clear_the_resume_flag(app): + """The `awaitingInitRef` trap, one ref along — see MESHBAY_DESIGN.md §8.5. + + A resume in flight when the film changes, or when the viewer seeks, would + have the next `stream_init` keep a buffer belonging to something else. + """ + player = _player(app) + effect = player[player.index("useEffect(() => {\n let cancelled = false;"):] + reset = effect[:effect.index("const requestSeek = ")] + assert "resumingRef.current = false" in reset, ( + "a resume in flight when the film changes is inherited by the new one") + seek = _fn(player, "requestSeek") + assert "resumingRef.current = false" in seek, ( + "a seek after an unlanded resume takes the resume branch and keeps the " + "buffer it was supposed to discard") |