path: root/packages/meshbay-hub/tests/test_video_reconnect.py

"""
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")