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

"""
A big film stops at around 100 MB.

Reported against 0.5: upload a 500 MB video, play it, and the player loads
roughly 100 MB and then hangs on "buffering" for good.

100 MB is not a number in our code. It is where the browser stops: a video
SourceBuffer is capped at a few hundred megabytes and `appendBuffer` throws
QuotaExceededError past it. The node remuxes with `-c copy`, so the bytes on
the wire are the file's own — a 500 MB film really does try to put 500 MB into
that buffer, and on a fast link it reaches the ceiling in the first minute,
long before anyone has watched enough for eviction to have anything to drop.

Two defects, and the second is the one that makes it permanent.

**Nothing bounded how far ahead we pulled.** Credit was granted once per
append: the node sent exactly as fast as the browser could append, which is as
fast as the network allows, which for a film is very much faster than watching
it. Memory was bounded by the browser's ceiling rather than by anything we
chose. Buffering by *time* past the playhead instead makes a two-hour film cost
the same as a two-minute clip.

**And then a bound in seconds turned out to be the wrong unit.** The browser
limits BYTES, so ninety seconds had to be sized for the highest-bitrate file in
a library and every ordinary file held a fraction of what the same browser
would have taken: measured here against a 100 MB ceiling, a 1.0 Mbit/s film
reached its ninety seconds on 20.7 MB and a 2.3 Mbit/s one on 45.6 MB. On a
mobile network that unused ceiling is the whole of the margin there is for a
dead spot. The read-ahead is now a byte budget converted to seconds at the
file's own bitrate, floored at the fixed bound so nothing pulls less than it
used to, and *walked up* rather than declared — because the real ceiling is
per-engine and per-device, and the alternative to walking up to it is
overshooting it, which costs a refused append every time it is tried.

**The pipeline could not restart itself.** An append refused for quota fires no
`updateend`. `updateend` was where credit was granted, so no credit went out;
the node then sent nothing, so no segment arrived to call `flushQueue` again.
Every wakeup the append path had was downstream of the append that had just
failed. Playback continuing past the segment, which is exactly what frees the
room needed to recover, woke nothing at all. The player deadlocked against
itself.

The first version of these tests modelled the pipeline and passed while the
player still hung, because a model of a fix written by whoever wrote the fix
agrees with it by construction. They now run the shipped `bufferedAhead`,
`evictBehind`, `flushQueue` and `pump`, lifted out of app.js as text, against a
fake SourceBuffer — see harness/mse_harness.mjs. What is modelled is the
browser, not us.
"""

import json
import re
import shutil
import subprocess
from pathlib import Path

import node_tree
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 _const(app: str, name: str) -> float:
    """One of the player's numeric constants, arithmetic and all.

    Read out of the source rather than restated here, so a change to a constant
    is a change to what these tests assert — the same rule the harness follows.
    """
    m = re.search(rf"^const {name} = ([\d\s.*/+-]+);$", app, re.M)
    assert m, f"{name} is not a numeric constant in video-player.js"
    return float(eval(m.group(1), {"__builtins__": {}}, {}))  # noqa: S307


# ── The shipped functions, run against a browser that has a ceiling ───────────

HARNESS = Path(__file__).parent / "harness" / "mse_harness.mjs"


def _harness(**cfg) -> dict:
    proc = subprocess.run(
        ["node", str(HARNESS), str(APP), json.dumps(cfg)],
        capture_output=True, text=True)
    assert proc.returncode == 0, proc.stderr
    return json.loads(proc.stdout)


@pytest.fixture(scope="module")
def idle():
    """Nobody pressed play — autoplay is blocked on a phone more often than not."""
    return _harness(playing=False)


@pytest.fixture(scope="module")
def watched():
    return _harness(playing=True)


def test_the_ceiling_is_never_reached_when_nobody_presses_play(idle):
    """The reported hang, from the side that produces it.

    A film left on the loading screen used to pull until the browser refused an
    append, and that refusal was unrecoverable. Nothing should get near it.
    """
    assert not idle["hitCeiling"], (
        f"filled the buffer to the ceiling ({idle['heldInBufferMB']} MB)")
    assert idle["quotaRefusals"] == 0, (
        f"{idle['quotaRefusals']} appends refused for quota — the state the "
        "player cannot get out of on its own")


def test_the_ceiling_is_never_reached_while_watching(watched):
    assert not watched["hitCeiling"], (
        f"filled the buffer to the ceiling ({watched['heldInBufferMB']} MB)")
    assert watched["quotaRefusals"] == 0


def test_the_read_ahead_is_bounded_by_the_playhead(app, idle, watched):
    """What replaced "as fast as the network allows".

    The bound is the one the player computed for this film, not the constant:
    the constant is only its floor. Asserting against the constant would say
    nothing about a film whose bitrate raises the bound above it, which is now
    most of them.
    """
    for name, run in (("idle", idle), ("watching", watched)):
        limit = run["aheadLimitS"]
        assert limit >= _const(app, "BUFFER_AHEAD_S"), (
            f"{name}: the bound came out at {limit}s, below the fixed floor — "
            "some film now buffers less than it did before the budget existed")
        # One window of segments may land past the gate before it shuts, which
        # is the flow control working, not the gate leaking.
        assert run["bufferedAheadS"] < limit * 1.5, (
            f"{name}: {run['bufferedAheadS']}s buffered against a {limit}s "
            "bound — the gate is not holding")


def test_a_watched_film_keeps_being_fed(watched):
    """The gate must throttle the stream, not stop it.

    Holding credit for good would be just as broken as never holding it, and
    would look the same from the sofa.
    """
    assert watched["watchedS"] > 500, (
        "playback did not advance, so this run says nothing about throttling")
    assert watched["sentMB"] > 40, (
        f"only {watched['sentMB']} MB reached the player in ten minutes of "
        "playback — the gate is holding credit it should have released")
    assert watched["removes"] > 0, "nothing was ever evicted behind the playhead"


def test_memory_stays_bounded_over_a_long_watch(app, watched):
    """The budget is a bound on memory, so it is the one to assert against.

    What the player may hold is the read-ahead budget plus the minute it keeps
    behind the playhead for a small seek back. Anything past that is eviction
    failing to keep up, which is the fault this guards.
    """
    mb = 1024 * 1024
    per_s = watched["bitrateMbits"] * 1e6 / 8 / mb
    allowed = _const(app, "BUFFER_AHEAD_MAX_BYTES") / mb \
        + _const(app, "BUFFER_BEHIND_S") * per_s
    # One window of segments may be in flight past the budget when the run ends.
    allowed += 8 * 256 * 1024 / mb
    assert watched["heldInBufferMB"] < allowed, (
        f"holding {watched['heldInBufferMB']} MB against {allowed:.1f} MB of "
        "budget and eviction window — eviction is not keeping up")


# ── The byte budget, and what it must not cost ────────────────────────────────

# A library's worth of bitrates against one ceiling, and the same low-bitrate
# film against a ceiling small enough that the budget cannot have it. Sizes are
# a two-hour film at each rate; nothing here names a real title.
_BITRATE_SWEEP = [
    ("0.6 Mbit/s", {"fileMB": 493.5, "durationS": 7200, "capMB": 100}),
    ("2.3 Mbit/s", {"fileMB": 2000, "durationS": 7200, "capMB": 100}),
    ("4.7 Mbit/s", {"fileMB": 4000, "durationS": 7200, "capMB": 100}),
    ("9.3 Mbit/s", {"fileMB": 8000, "durationS": 7200, "capMB": 100}),
    ("0.6 Mbit/s, small ceiling", {"fileMB": 493.5, "durationS": 7200, "capMB": 40}),
    ("2.3 Mbit/s, small ceiling", {"fileMB": 2000, "durationS": 7200, "capMB": 40}),
]


@pytest.fixture(scope="module")
def sweep():
    return {name: _harness(playing=True, netMBs=35, wallS=900, **cfg)
            for name, cfg in _BITRATE_SWEEP}


def _fits_under_the_ceiling(app: str, name: str, run: dict) -> bool:
    """Whether even the *floor* bound fits in this browser's buffer.

    Where it does not, the film was already unservable before the budget
    existed — the fixed ninety seconds plus the minute kept behind is more than
    the ceiling, so there is nowhere to put it. See MESHBAY_DESIGN.md §15.3.
    """
    cap_mb = dict(_BITRATE_SWEEP)[name]["capMB"]
    per_s = run["bitrateMbits"] * 1e6 / 8 / (1024 * 1024)
    need = (_const(app, "BUFFER_AHEAD_S") + _const(app, "BUFFER_BEHIND_S")) * per_s
    return need <= cap_mb


def test_no_film_buffers_less_than_the_fixed_bound_used_to(app, sweep):
    """The floor, which is the whole no-regression claim.

    The budget may only ever raise the read-ahead. A film whose bitrate spends
    it in under the fixed bound keeps that bound exactly, and nothing that
    played before pulls less than it did.
    """
    floor = _const(app, "BUFFER_AHEAD_S")
    for name, run in sweep.items():
        assert run["aheadLimitS"] >= floor, (
            f"{name}: bound came out at {run['aheadLimitS']}s, under the {floor}s "
            "floor — this film now buffers less than it did before")


def test_a_film_nobody_started_does_not_spend_the_budget(app):
    """A film on the loading screen cannot be interrupted, so it needs no lead.

    Autoplay is blocked on a phone more often than not, so this is the ordinary
    first state of every playback and not a corner. Left to grow there, the
    budget filled a small browser's entire buffer — 40 MB against the 8 MB the
    fixed bound took — with film on a mobile connection that is paying for it,
    and reached the ceiling that the fixed bound never came near.
    """
    idle_small = _harness(playing=False, fileMB=493.5, durationS=7200,
                          capMB=40, netMBs=35, wallS=900)
    assert idle_small["aheadLimitS"] == _const(app, "BUFFER_AHEAD_S"), (
        f"an unstarted film is allowed {idle_small['aheadLimitS']}s of "
        "read-ahead — the budget grows before anyone has pressed play")
    assert not idle_small["hitCeiling"], (
        f"a film nobody started filled the buffer to the ceiling "
        f"({idle_small['heldInBufferMB']} MB)")
    assert idle_small["quotaRefusals"] == 0


def test_an_ordinary_film_buffers_far_past_the_fixed_bound(app, sweep):
    """And the point of the exercise.

    Most of a library is well under five megabits, and that is where the
    unused ceiling was. A bound that never actually rises is a budget that
    bought nothing.
    """
    floor = _const(app, "BUFFER_AHEAD_S")
    low = sweep["0.6 Mbit/s"]
    assert low["aheadLimitS"] > floor * 3, (
        f"a 0.6 Mbit/s film is allowed {low['aheadLimitS']}s of read-ahead "
        f"against a {floor}s floor — the byte budget is not being spent")
    assert low["bufferedAheadS"] > floor * 3, (
        f"allowed {low['aheadLimitS']}s but only reached "
        f"{low['bufferedAheadS']}s — something other than the bound is holding")


def test_finding_the_ceiling_costs_a_handful_of_refusals_not_thousands(sweep):
    """The budget is walked up to, never overshot wholesale.

    A flat budget discovers a browser's ceiling by being refused at it, and
    that refusal repeats: a 48 MB budget against a 40 MB ceiling was measured
    at 1386 refused appends on a film that had none before it. Each one is a
    thrown exception, a warning on a phone's console and an append's work
    thrown away, so the count is the thing to bound.
    """
    for name, run in sweep.items():
        assert run["quotaRefusals"] < 50, (
            f"{name}: {run['quotaRefusals']} appends refused for quota — the "
            "budget is overshooting the ceiling rather than walking up to it")


def test_the_walk_settles_and_playback_survives_it(app, sweep):
    """Discovering the ceiling must not cost the film.

    Every rate that was servable at all has to still be watchable while the
    budget is being found, and the run has to end below the ceiling rather than
    pinned against it. A rate whose *floor* bound does not fit the ceiling is
    excluded, not because the budget is allowed to break it, but because it was
    already broken: the 9.3 Mbit/s run plays the same 100.8 s with the budget
    as it did without one, to the tenth of a second. That wedge is §15.3, not
    this commit.
    """
    for name, run in sweep.items():
        assert not run["hitCeiling"], f"{name}: finished pinned at the ceiling"
        if not _fits_under_the_ceiling(app, name, run):
            continue
        assert run["watchedS"] > 600, (
            f"{name}: only {run['watchedS']}s played in 900s of wall clock — "
            "playback did not survive the walk")


def test_a_refused_append_is_not_retried_until_something_frees_room(app):
    """Why the walk is cheap.

    An append refused for quota with nothing to evict will be refused again
    until room appears, and pump() calls flushQueue on a clock — so the retry
    alone cost a refusal several times a second for as long as it took the film
    to play far enough for eviction to have anything to drop. `evictBehind` is
    the only thing that gives a SourceBuffer room back, so it is the only thing
    that may lift the hold; a timer or the playhead would move long before the
    answer changes.
    """
    player = _player(app)
    flush = player[player.index("const flushQueue = useCallback("):]
    flush = flush[:flush.index("\n  }, [")]
    assert "if (quotaHoldRef.current) return;" in flush, (
        "flushQueue retries a refused append on every pump tick")
    evict = player[player.index("const evictBehind = useCallback("):]
    evict = evict[:evict.index("\n  }, [")]
    assert "quotaHoldRef.current = false" in evict, (
        "nothing lifts the hold when room is freed, so the first refusal stops "
        "every append for the rest of the film")
    # A seek backwards leaves the playhead behind everything buffered, and
    # reinitAt empties the buffer outright — so the hold must not survive it.
    reinit = player[player.index("const reinitAt = async ("):]
    reinit = reinit[:reinit.index("\n    };")]
    assert "quotaHoldRef.current = false" in reinit, (
        "a hold set before a seek survives the buffer it was held against")


def test_the_budget_starts_from_nothing_on_every_new_stream(app):
    """A new stream starts from a known state, and this is two more refs.

    What one browser refused for a nine-megabit film says nothing about the
    next one, and a budget left at the floor by the film before would silently
    cap every film after it for the life of the page.
    """
    player = _player(app)
    i = player.index("appendingRef.current = false;\n    endedRef.current = false;")
    reset = player[i:i + 2000]
    for ref, value in (("bitrateRef", "0"),
                       ("aheadBytesRef", "0"),
                       ("aheadCapRef", "BUFFER_AHEAD_MAX_BYTES"),
                       ("quotaHoldRef", "false")):
        assert f"{ref}.current = {value};" in reset, (
            f"{ref} is not reset at the start of a stream, so the next film "
            "inherits what this one learnt")


# ── The shape the fix depends on ──────────────────────────────────────────────

def test_credit_is_granted_in_exactly_one_place(app):
    """Granting from `updateend` is the deadlock. It must not come back.

    A second grant site is how this regresses: it would work, until the append
    it hangs off is the one the ceiling refuses.
    """
    player = _player(app)
    sites = player.count("grantStreamCredit(")
    assert sites == 2, (
        f"{sites} calls to grantStreamCredit — expected exactly two, both "
        "inside pump(): the keepalive and the release")
    pump = player[player.index("const pump = useCallback("):]
    pump = pump[:pump.index("\n  }, [")]
    assert pump.count("grantStreamCredit(") == 2, (
        "credit is granted outside pump(), so an append that is refused for "
        "quota can still be the only thing that would have woken the pipeline")


def test_something_other_than_data_drives_the_pipeline(app):
    """The recovery path cannot depend on a segment arriving."""
    player = _player(app)
    assert "setInterval(pump" in player, (
        "no clock drives pump(): once the ceiling refuses an append, nothing "
        "arrives and nothing retries")
    assert "clearInterval(pumpTimer)" in player, "the pump timer outlives the player"
    assert "addEventListener('timeupdate', pump)" in player, (
        "playback progress is what frees room to evict, and it wakes nothing")


def test_a_buffered_viewer_still_tells_the_node_it_is_there(app):
    """Holding credit back must not read as a closed tab."""
    player = _player(app)
    assert "grantStreamCredit(0)" in player, (
        "a viewer that is far enough ahead grants nothing and says nothing, so "
        "the node's stall timeout ends a film that is merely paused")


@pytest.mark.skipif(not node_tree.available(), reason="node sources unavailable")
def test_the_node_ends_a_stream_on_silence_not_on_stinginess():
    """The other half of the keepalive: the node has to honour it."""
    body = node_tree.method("_await_stream_credit")
    assert "self._stream_heard_at" in body, (
        "the stall budget still accumulates over the whole wait, so a keepalive "
        "that grants no credit cannot keep a paused film alive")
    assert "waited += STREAM_CREDIT_POLL" not in body, (
        "the budget still accumulates over the whole wait rather than being "
        "measured from the last thing the peer said")
    grant = node_tree.method("_grant_stream_credit")
    assert "self._stream_heard_at = time.monotonic()" in grant, (
        "n=0 does not refresh the timeout, so the keepalive is a no-op")


def test_appending_does_not_earn_credit(app):
    """What may be in flight is a question about the buffer, not about appends.

    Tying the two was the original design and it was wrong twice over.
    `updateend` fires for `remove()` as well, so the player paid the node for
    its own evictions; and crediting per append meant taking segments as fast
    as they could be written, which is as fast as the network allows.
    """
    player = _player(app)
    # The real handler, not the one-liner in settled()
    marker = "sb.addEventListener('updateend', () =>"
    handler = player[player.index(marker):]
    handler = handler[:handler.index("\n          });")]
    assert "grantStreamCredit" not in handler, (
        "credit is granted from updateend, which fires for remove() too")
    assert "outstandingRef" not in handler, (
        "the in-flight window is adjusted from updateend rather than from the "
        "buffer, so an eviction still counts as room for another segment")


def test_credit_is_a_window_and_not_a_debt(app):
    """It must be topped up, not paid off.

    Accumulating a credit per append and handing over the whole balance when
    the buffer finally had room sent six megabytes in one burst, overshot the
    target by a minute of film, and then said nothing for forty-six seconds.
    Measured in Chrome against real fragmented MP4.
    """
    player = _player(app)
    pump = player[player.index("const pump = useCallback("):]
    pump = pump[:pump.index("\n  }, [")]
    assert "STREAM_WINDOW - outstandingRef.current" in pump, (
        "pump() no longer tops a window up to what is allowed in flight")
    src = APP.read_text()
    window = int(re.search(r"const STREAM_WINDOW = (\d+)", src).group(1))
    assert 2 <= window <= 16, (
        f"a window of {window} segments is either too small to keep the pipe "
        "busy or big enough to be a burst again")
    assert "outstandingRef.current = Math.max(0, outstandingRef.current - 1)" in player, (
        "nothing decrements the window when a segment lands, so it fills once "
        "and never reopens")