""" Seeking in a stream, and picking a film up where it was left. Until now the scrubber was a lie: `ms.duration` was set to the whole film, so the bar was drawn full length, and `onSeeking` quietly clamped any target back into what happened to be buffered. The stream itself only ever ran forwards from byte zero. Seeking is a stream restarted somewhere else. The node already knew how — the legacy HLS path passes `-ss` — so `stream_req` gained a `start`, ffmpeg is spawned with `-ss` **before** `-i` (seeking by the container index, milliseconds on a 500 MB film rather than tens of seconds of decoding), and the client puts the fragments back on the film's timeline with `SourceBuffer.timestampOffset`. Three things about that are easy to get wrong and are what these tests hold: **ffmpeg restarts its timestamps at zero** however far in it seeks — measured, `-copyts` does not change it for this input. So the offset has to come from the client, and the node has to say which position it actually used. **The channel is ordered, and a seek does not change the file.** Everything between asking for a seek and its `stream_init` belongs to the stream being abandoned, and `file_id` cannot tell them apart. Appending it would put the old material on top of the new. **A seek makes the buffer discontinuous.** Every piece of code that reads `buffered` then has to mean a particular range: "the last one" stops being "the one being watched", and measuring the read-ahead across a gap reports a full buffer while the player starves. """ import re import shutil from pathlib import Path import node_tree import pytest from spa_source import transport_source STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static" APP = STATIC / "video-player.js" TRANSPORT = STATIC / "transport.js" pytestmark = pytest.mark.skipif(not APP.exists(), reason="SPA sources unavailable") @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)] # ── The node ────────────────────────────────────────────────────────────────── @pytest.fixture(scope="module") def stream_fn(): if not node_tree.available(): pytest.skip("node sources unavailable") return node_tree.method("_stream_video_inner") def test_the_seek_is_an_index_lookup_not_a_decode(stream_fn): """`-ss` before `-i`, which is the difference between instant and unusable. After `-i` it means "decode and discard until you get there" — tens of seconds on a long film, per seek. Before it, ffmpeg uses the container's index and starts almost at once. """ args = stream_fn[stream_fn.index("create_subprocess_exec"):] args = args[:args.index("stdout=")] assert "seek_args" in args, "the stream is still spawned without a seek" assert args.index("*seek_args") < args.index('"-i"'), ( "-ss lands after -i, which decodes the whole film up to the seek point") def test_the_node_says_where_it_actually_started(stream_fn): """The client cannot infer it. ffmpeg restarts its output timestamps at zero however far in it seeks, so the offset that puts the fragments back on the timeline has to be told. """ init = stream_fn[stream_fn.index("MNP.STREAM_INIT"):] init = init[:init.index("})")] assert re.search(r'"start":\s*start', init), ( "stream_init carries no start, so the client has nothing to offset by") def test_seeking_past_the_end_is_pulled_back(stream_fn): """Otherwise ffmpeg produces nothing and the player waits for ever.""" assert "duration - 1" in stream_fn and "duration - 5" in stream_fn, ( "a seek to or past the end is not clamped — the stream would be empty " "and the player would sit on 'buffering' with nothing coming") assert "start = max(0.0, start)" in stream_fn, "a negative start is not refused" def test_the_request_carries_it(): text = transport_source() fn = text[text.index("requestStream(fileId"):] fn = fn[:fn.index("\n }")] assert "start" in fn, "requestStream cannot express a seek" # ── Old material must not land on the new stream ────────────────────────────── def test_everything_before_the_new_init_is_dropped(app): """A seek does not change the file, so `file_id` cannot separate them. Ordering can: the node retires the previous stream before sending the new `stream_init`, so anything arriving in between is the film we left. """ player = _player(app) for handler in ("onStreamData", "onStreamEnd"): body = player[player.index(f"transport.{handler} = "):] body = body[:body.index("\n };")] assert "awaitingInitRef.current" in body, ( f"{handler} accepts data from the stream being abandoned; on " "onStreamData that appends the old film over the new one, on " "onStreamEnd it truncates the film at the seek point") def test_a_discarded_segment_still_frees_its_place_in_the_window(app): """The leak that made the third seek hang. `reinitAt` waits for two `updateend` events, and the seek's first segments arrive during that gap and are discarded by the flag above. If the window is only decremented after those early returns, each discarded segment takes a slot with it. Lose the whole window and the player believes eight segments are in flight, grants nothing ever again, and the node waits for credit that cannot come — while its log shows a stream it fed perfectly well. A race, which is why it worked twice and hung on the third try. """ player = _player(app) body = player[player.index("transport.onStreamData = "):] body = body[:body.index("\n };")] decrement = body.index("outstandingRef.current = Math.max(0") for guard in ("awaitingInitRef.current", "msg.file_id !== entry.id"): assert body.index(guard) > decrement, ( f"the window is decremented after the `{guard}` check, so every " "segment discarded there is a slot lost from the window for good") def test_the_leak_deadlocks_the_window_and_the_fix_clears_it(): """The same defect, run rather than read. It was a race — whether the player survived a seek depended on how many segments arrived before `reinitAt` finished — so a run that works proves little on its own. This forces the worst case: the whole window arrives while the flag is up. Then the only question left is arithmetic. """ import json import subprocess harness = Path(__file__).parent / "harness" / "window_leak.mjs" if shutil.which("node") is None: pytest.skip("node is not available") def run(decrement_first: bool) -> dict: proc = subprocess.run( ["node", str(harness), str(APP), json.dumps({"decrementFirst": decrement_first})], capture_output=True, text=True) assert proc.returncode == 0, proc.stderr return json.loads(proc.stdout) was = run(False) assert was["deadlocked"], ( "putting the decrement back after the guards no longer deadlocks the " "window, so this test no longer describes the defect it guards") now = run(True) assert not now["deadlocked"], ( f"the window is still {now['windowAfterDiscards']} after discarding a " "full window's worth of segments — the player would grant no more " "credit and the node would wait for ever") def test_the_window_is_topped_up_by_something_other_than_arrivals(app): """Because the arrival that would have done it is the one being discarded.""" player = _player(app) assert "setInterval(pump" in player, ( "only an accepted segment tops the window up, so a window emptied by " "discards has nothing to refill it") def test_a_new_stream_clears_the_seek_state(app): """The same shape as `appendingRef` before it, and worse. Switching film while a seek is in flight leaves `awaitingInit` true, and only `reinitAt` lowers it — which the next film never reaches, because it builds a new SourceBuffer and takes the first-init path. Every segment of the new film is then discarded as though it belonged to the old one. """ player = _player(app) effect = player[player.index("useEffect(() => {\n let cancelled = false;"):] effect = effect[:effect.index("transport.requestStream")] assert "awaitingInitRef.current = false" in effect, ( "a seek in flight when the film changes silences the next one entirely") assert "seekTargetRef.current = null" in effect, ( "the next film jumps to a position from the previous one") def test_the_flag_is_raised_when_the_seek_is_requested_and_cleared_on_init(app): player = _player(app) seek = player[player.index("const requestSeek = "):] seek = seek[:seek.index("\n };")] assert "awaitingInitRef.current = true" in seek, ( "nothing marks the gap between asking and being answered") reinit = player[player.index("const reinitAt = "):] reinit = reinit[:reinit.index("\n };")] assert "awaitingInitRef.current = false" in reinit, ( "the flag is never lowered, so the new stream is dropped too") # ── The buffer after a seek ─────────────────────────────────────────────────── def test_the_parser_is_reset_before_the_next_stream_is_appended(app): """ffmpeg was killed mid-fragment, so the parser holds half of one.""" player = _player(app) reinit = player[player.index("const reinitAt = "):] reinit = reinit[:reinit.index("\n };")] assert "sb.abort()" in reinit, ( "the SourceBuffer keeps the half fragment it was parsing, and the next " "stream's header lands on top of it") assert "sb.remove(0, Infinity)" in reinit, ( "the old material is kept, so the buffer is discontinuous for the rest " "of the film") assert "sb.timestampOffset = start" in reinit, ( "the new fragments are not placed on the film's timeline") def test_the_read_ahead_is_measured_on_the_range_being_watched(app): """"The last range" stops meaning "the one playing" once there is a gap.""" player = _player(app) for fn in ("bufferedAhead", "evictBehind"): body = player[player.index(f"const {fn} = useCallback("):] body = body[:body.index("\n }, [")] assert "currentRange()" in body, ( f"{fn} still assumes one contiguous range: after a seek it reads " "a range on the far side of a gap") def test_the_mode_places_fragments_rather_than_stacking_them(app): """'sequence' concatenates; a stream that starts at 40 minutes must not.""" player = _player(app) assert "sb.mode = 'segments'" in player, ( "in 'sequence' mode the fragments are laid end to end, so a stream " "starting mid-film is buffered at zero and the scrubber lies") def test_the_playhead_waits_for_the_data(app): """Seeking into an unbuffered region leaves the element with nothing.""" player = _player(app) assert "const landPlayhead" in player, ( "currentTime is set without checking the data for it has arrived") land = player[player.index("const landPlayhead = "):] land = land[:land.index("\n };")] assert "sb.buffered" in land and "seekTargetRef.current = null" in land, ( "the target is not checked against what is buffered, or never cleared") # ── Scrubbing must not be a storm of ffmpeg ─────────────────────────────────── def test_seeks_are_debounced(app): """Dragging fires `seeking` continuously; each one we act on costs a spawn.""" player = _player(app) assert "SEEK_DEBOUNCE_MS" in player, "every intermediate drag position seeks" ms = int(re.search(r"const SEEK_DEBOUNCE_MS = (\d+)", APP.read_text()).group(1)) assert 150 <= ms <= 1000, ( f"{ms} ms is either short enough to still storm the node or long " "enough to feel broken") def test_a_seek_inside_the_buffer_does_not_reach_the_node(app): """The browser already has it; restarting ffmpeg for it would be absurd.""" player = _player(app) body = player[player.index("const onSeeking = "):] body = body[:body.index("\n };")] assert "sb.buffered" in body and "return" in body, ( "onSeeking asks the node even for a position already buffered") # ── Resume ──────────────────────────────────────────────────────────────────── def test_the_position_is_kept_in_this_browser(app): """localStorage: no protocol, no storage for anyone else to keep, and nothing new learns what you watch.""" src = APP.read_text() assert "mb:pos:" in src, "no position is stored" read = src[src.index("function readResumePosition"):] read = read[:read.index("\n}")] assert "catch" in read, ( "localStorage throws in private browsing and with storage disabled, " "and a player that cannot start there is worse than one that forgets") def test_a_finished_film_does_not_offer_to_resume(app): src = APP.read_text() write = src[src.index("function writeResumePosition"):] write = write[:write.index("\n}")] assert "RESUME_MAX_FRACTION" in write and "removeItem" in write, ( "a position at the credits is kept, so reopening the film resumes " "thirty seconds before the end for ever") assert "RESUME_MIN_S" in write, "the first seconds are remembered as a position" def test_the_viewer_can_refuse_the_resume(app): player = _player(app) assert "video.from_start" in player, ( "resuming is imposed with no way back to the beginning") @pytest.mark.parametrize("locale", ["en", "fr", "es", "pt-BR", "zh-CN", "ja", "de", "it", "nl", "pl"]) def test_the_resume_strings_exist_everywhere(locale): text = (STATIC / "locales" / f"{locale}.js").read_text() for key in ("video.resumed_at", "video.from_start"): assert key in text, f"{locale} is missing {key}"