From 188a76f52d2f30609147b1beee7754f2cbd1e778 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 16 Aug 2026 15:28:53 +0200 Subject: fix(node): stop losing transcode slots, and reap ffmpeg without deadlocking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reported from a phone: play a video, close the viewer, open another — the second hangs and the third is refused. Three separate causes, found by instrumenting rather than guessing, after two fixes that addressed real but different bugs. A task nobody holds can be collected mid-flight. asyncio keeps only a weak reference, so `ensure_future` with the result discarded may be garbage-collected while running — "Task was destroyed but it is pending!" — and `_stream_video` never reached the exit of its `async with sem`. `_spawn` holds every background task; all nineteen call sites go through it. Losing the peer must stop its work. The connectionstatechange handler popped the session from a dict and nothing else, so a closed tab went on transcoding for the full 120 s credit timeout. Measured in the log: 91 s of ffmpeg after the connection closed. `shutdown_tasks()` now runs on the way out, and the credit wait checks the channel before sleeping and polls in slices instead of once. And `await proc.wait()` after `kill()` still deadlocks. ffmpeg outruns a credit-paced viewer and fills the stdout pipe; stop reading it and the transport cannot finish closing, SIGKILL or not. Measured against the live node with a 169 MB video, closing the viewer after 20 segments and asking for the next one: 15.1 s then "Server busy" before, 0.1 s / 0.0 s / 0.0 s after. Chunk replies wait for room on the channel. Eight megabyte-sized chunks answered as they arrived queued 8 MB with nothing watching — measured at 7.3 MB of bufferedAmount in milliseconds. Fine on a LAN, minutes of head-of-line delay on a busy link. Upload names accept any script. The rule was ASCII-only, so `été.txt` was refused — and so was `rapport (1).pdf`, which is the form `_free_name` produces itself, meaning the node rejected names it had chosen. Widened to Unicode with the C5a and H2 protections intact, plus a refusal of names that lie about themselves: trailing space or dot, and the right-to-left override. Errors now name the file, so one bad name no longer fails every upload in flight. Co-Authored-By: Claude Opus 5 --- .../tests/test_security_regressions.py | 33 ++- packages/meshbay-node/tests/test_task_lifetime.py | 256 +++++++++++++++++++++ 2 files changed, 288 insertions(+), 1 deletion(-) create mode 100644 packages/meshbay-node/tests/test_task_lifetime.py (limited to 'packages/meshbay-node/tests') diff --git a/packages/meshbay-node/tests/test_security_regressions.py b/packages/meshbay-node/tests/test_security_regressions.py index 77d544b..e13dec0 100644 --- a/packages/meshbay-node/tests/test_security_regressions.py +++ b/packages/meshbay-node/tests/test_security_regressions.py @@ -92,12 +92,43 @@ def test_upload_rejects_unsafe_filenames(name): "My Holiday Video.mkv", "report-2026.pdf", "track_01.flac", + # Reported 2026-08-16: an upload refused as "Invalid filename". The rule was + # ASCII-only, so most of the world could not send a file, and — worse — it + # rejected the "name (1).ext" form that _free_name produces itself, so the + # node refused names it had chosen. + "été.txt", + "naïve café.jpg", + "Ich möchte.pdf", + "日本語.mp4", + "rapport (1).pdf", ]) def test_upload_accepts_ordinary_filenames(name): - """The allowlist must not break normal use.""" + """The allowlist must not break normal use, in any script.""" assert _safe_name_re().match(name), f"should be accepted: {name!r}" +@pytest.mark.parametrize("name", [ + "trailing space ", + "ends.with.dot.", + "..", + "a\u202eexe.txt", # right-to-left override: hides the real extension +]) +def test_upload_rejects_names_that_lie_about_themselves(name): + """Widening to Unicode must not admit names that misrepresent the file.""" + assert not _safe_name_re().match(name), f"should be rejected: {name!r}" + + +def test_the_node_never_generates_a_name_it_would_refuse(tmp_path): + """_free_name resolves a collision by appending " (n)"; that has to be legal.""" + from meshbay_node.transport.webrtc_server import _free_name + (tmp_path / "clip.mp4").touch() + (tmp_path / "clip (1).mp4").touch() + chosen = _free_name(tmp_path, "clip.mp4") + assert chosen not in ("clip.mp4", "clip (1).mp4") + assert _safe_name_re().match(chosen), ( + f"the node picked {chosen!r} and would then reject it on the next upload") + + def _session(tmp_path: Path, user_id: str) -> WebRTCPeerSession: """A peer session wired to a real shared root, with sending stubbed out.""" shared_root = tmp_path / "shared" diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py new file mode 100644 index 0000000..8897e71 --- /dev/null +++ b/packages/meshbay-node/tests/test_task_lifetime.py @@ -0,0 +1,256 @@ +""" +Background tasks the node starts, and the slot one of them owned. + +asyncio keeps only a *weak* reference to a task. A coroutine fired with a bare +`asyncio.ensure_future` and never referenced again can therefore be collected +while it is still running — the loop logs "Task was destroyed but it is +pending!" and nothing else happens. + +For `_stream_video` that was expensive. It holds a transcode slot for its whole +life with `async with sem`, and a destroyed task never reaches `__aexit__`. The +node allows two, so two abandoned streams left it answering "Server busy" to +every request from then on: videos stopped playing entirely, first try included, +until the daemon was restarted. + +Seen in the wild on 2026-08-16 after a viewer switched films mid-stream. +""" + +import re +from pathlib import Path + +import pytest + +SERVER = (Path(__file__).resolve().parents[1] / "src" / "meshbay_node" + / "transport" / "webrtc_server.py") + +pytestmark = pytest.mark.skipif( + not SERVER.exists(), reason="the node sources are not available") + + +@pytest.fixture(scope="module") +def source(): + return SERVER.read_text() + + +@pytest.fixture(scope="module") +def session(source): + """The body of WebRTCPeerSession.""" + i = source.index("class WebRTCPeerSession") + return source[i:source.index("\nclass WebRTCTransport")] + + +def test_the_session_keeps_a_reference_to_what_it_starts(session): + assert "self._tasks: set[asyncio.Task] = set()" in session + spawn = session[session.index("def _spawn("):] + spawn = spawn[:spawn.index("\n def ", 1)] + assert "self._tasks.add(task)" in spawn, "the reference is what keeps it alive" + assert "add_done_callback(self._tasks.discard)" in spawn, ( + "without this the set grows for the life of the session") + + +def test_nothing_in_the_session_is_fired_and_forgotten(session): + """A bare ensure_future here is a task the collector may take.""" + stray = [] + for line in session.splitlines(): + if "asyncio.ensure_future(" in line and "task = asyncio.ensure_future" not in line: + if line.strip().startswith("#") or line.strip().startswith("`"): + continue + stray.append(line.strip()) + assert not stray, ( + "these start a task nobody holds; use self._spawn instead:\n " + + "\n ".join(stray)) + + +def test_the_stream_still_takes_a_slot_for_its_whole_life(session): + """The leak is only interesting because the slot is held this way.""" + body = session[session.index("async def _stream_video(self"):] + body = body[:body.index("\n async def ", 1)] + assert "async with sem:" in body + + +def test_closing_a_session_releases_its_tasks(session): + """A peer that vanishes mid-stream should give the slot back at once. + + The cancelling itself lives in shutdown_tasks, which the state handler also + uses; close() is the variant that additionally shuts the peer connection. + """ + close = session[session.index("async def close(self)"):] + close = close[:close.index("\n\n")] if "\n\n" in close else close + assert "shutdown_tasks()" in close + assert "self._pc.close()" in close + + fn = session[session.index("async def shutdown_tasks(self)"):] + fn = fn[:fn.index("\n async def ", 1)] + assert "_stop_stream()" in fn + assert "task.cancel()" in fn + assert "gather" in fn, "cancelling without awaiting does not run the exits" + + +def test_two_slots_is_the_whole_margin(source): + """States the number the failure hinged on, so a change is deliberate.""" + n = int(re.search(r"MAX_CONCURRENT_TRANSCODES\s*=\s*(\d+)", source).group(1)) + assert n == 2, ( + f"the cap is now {n}; the leak above emptied it in {n} abandoned " + "streams, so if this moves the comments explaining it should too") + + +# ── One viewer, one stream ──────────────────────────────────────────────────── + +def test_a_new_request_retires_the_previous_stream(session): + """Reported: first video fine, second fine, third stuck on "buffering". + + That is the shape of two transcode slots held by two streams that were + never told to stop. `stream_stop` is the only thing that ends one early, + and a browser that is backgrounded, reloaded or simply drops the message + never sends it — leaving STREAM_CREDIT_TIMEOUT, two minutes, as the only + release. Three videos inside two minutes exhausts the node. + + A session can only be watching one film, so the request for the next one is + proof the last is over. It does not depend on any message arriving. + """ + assert "self._spawn(self._replace_stream(msg))" in session, ( + "the stream request must go through the path that retires the old one") + + fn = session[session.index("async def _replace_stream(self"):] + fn = fn[:fn.index("\n async def ", 1)] + assert "self._stop_stream()" in fn + assert "await asyncio.wait_for(asyncio.shield(prev)" in fn, ( + "the slot comes back when the old task exits its `async with sem` — " + "so the new one has to wait for that, not merely ask") + assert "self._stream_task = asyncio.current_task()" in fn + + +def test_the_credit_timeout_is_not_the_only_way_out(source): + """Stated so the two-minute leash is a deliberate backstop, not the plan.""" + timeout = int(re.search(r"STREAM_CREDIT_TIMEOUT\s*=\s*(\d+)", source).group(1)) + slots = int(re.search(r"MAX_CONCURRENT_TRANSCODES\s*=\s*(\d+)", source).group(1)) + assert timeout >= 60, "a short leash would cut off a slow but live viewer" + assert "_replace_stream" in source, ( + f"with {slots} slots and a {timeout}s timeout, {slots} abandoned streams " + "lock the node for that long unless a new request retires the old one") + + +# ── A viewer that simply vanishes ───────────────────────────────────────────── + +def test_losing_the_peer_stops_its_work(source): + """Closing the tab, the browser or the viewer all end here. + + The connection-state handler used to pop the session from the dictionary + and nothing else, which forgets it without stopping it. Measured in the + node's own log on 2026-08-16: + + 11:46:31 WebRTC connection state: closed (peer=45c47e12) + 11:48:02 Stream stalled: no credit from peer=0cc9aaad + 11:48:02 Streamed clip.mp4: 2 segments + + Ninety-one seconds of ffmpeg, and of one of two transcode slots, after the + viewer had gone. Two of those and the next video is refused. + """ + handler = source[source.index('@pc.on("connectionstatechange")'):] + handler = handler[:handler.index("\n offer =")] + assert "shutdown_tasks()" in handler, ( + "a lost peer must have its stream stopped, not merely be forgotten") + assert "pop(peer_id, None)" in handler + + +def test_shutdown_is_separate_from_closing_the_connection(session): + """The state handler runs while aiortc is already tearing pc down.""" + fn = session[session.index("async def shutdown_tasks(self)"):] + fn = fn[:fn.index("\n async def ", 1)] + assert "self._pc.close()" not in fn, ( + "calling pc.close() from the state handler re-enters the teardown") + assert "task.cancel()" in fn and "gather" in fn + + +def test_a_dead_channel_is_noticed_while_waiting_not_after(source): + """A peer that vanishes sends no credit and fires no event. + + Waiting the whole budget on a channel that is already shut is the slot + being held for nothing, which is what the log above shows. + """ + fn = source[source.index("async def _await_stream_credit"):] + fn = fn[:fn.index("\n async def ", 1)] + before = fn[:fn.index("wait_for")] + assert 'readyState != "open"' in before, ( + "the channel must be checked before the wait, not only after it") + assert "STREAM_CREDIT_POLL" in fn, ( + "one long sleep cannot notice a connection dying mid-wait") + + +def test_the_poll_is_much_shorter_than_the_budget(source): + poll = int(re.search(r"STREAM_CREDIT_POLL\s*=\s*(\d+)", source).group(1)) + total = int(re.search(r"STREAM_CREDIT_TIMEOUT\s*=\s*(\d+)", source).group(1)) + assert poll <= 10, f"a {poll}s poll still leaves a slot idle too long" + assert total > poll, "the budget must survive more than one poll" + + +# ── Reaping ffmpeg without deadlocking on its own output ────────────────────── + +def test_the_pipes_are_drained_before_waiting_on_ffmpeg(source): + """The bug behind "close the viewer, next video hangs". + + ffmpeg outruns a credit-paced viewer and fills the stdout pipe. Stop reading + it — which is exactly what closing the player does — and `await proc.wait()` + never returns: asyncio cannot finish closing the transport while that buffer + is full. The task stays alive holding a transcode slot. + + Measured against the live node with a 169 MB video, closing the viewer after + 20 segments and asking for the next one straight away: + + without the drain run 1 served after 15.1s, run 2 refused "Server busy" + with it 0.1s, 0.0s, 0.0s + + Which is the reported failure exactly: one video fine, the next hanging, + the one after refused. + """ + fn = source[source.index("async def _stream_video_inner"):] + fn = fn[:fn.index("\n def _send(")] + finally_block = fn[fn.index("finally:"):] + + assert "proc.kill()" in finally_block + assert "pipe.read()" in finally_block, ( + "the pipe has to be drained or wait() cannot complete") + assert "wait_for(proc.wait()" in finally_block, ( + "an unbounded wait here is a transcode slot held for good") + + +def test_releasing_the_slot_does_not_depend_on_ffmpeg_behaving(source): + """SIGKILL has already been sent; the OS will reap it either way.""" + fn = source[source.index("async def _stream_video_inner"):] + fn = fn[:fn.index("\n def _send(")] + tail = fn[fn.index("wait_for(proc.wait()"):] + assert "except Exception" in tail, ( + "a timeout waiting for ffmpeg must not stop the slot coming back") + + +# ── Downloads while the link is busy ────────────────────────────────────────── + +def test_chunks_wait_for_room_on_the_channel(session): + """A megabyte chunk, eight in flight, and nothing watching the send buffer. + + Measured on the node while two uploads ran: bufferedAmount climbed to 7.3 MB + in a few milliseconds, because every request was answered the instant it + arrived. On a LAN that drains before anyone notices. On a phone that is also + uploading, the reader gets the first chunk and then waits for the link to + work through the rest — which is what "stuck at 1 MB" looks like, one chunk + being exactly one megabyte. + """ + fn = session[session.index("async def _do_file_request"):] + fn = fn[:fn.index("\n def _do_stream_segment")] + assert "DOWNLOAD_BUFFER_HIGH" in fn, "the send buffer has to be watched" + assert "await asyncio.sleep" in fn, "waiting for room is the point" + assert 'readyState != "open"' in fn, ( + "a peer that leaves mid-wait must not be written to") + + +def test_answering_a_chunk_does_not_block_the_message_loop(session): + """It waits for buffer room, and the acks that free it arrive on this loop.""" + assert "self._spawn(self._do_file_request(msg))" in session, ( + "answering inline would stall the uploads whose acks drain the buffer") + + +def test_the_download_ceiling_leaves_room_to_work(source): + high = eval(re.search(r"DOWNLOAD_BUFFER_HIGH\s*=\s*([\d *]+)", source).group(1)) + assert 512 * 1024 <= high <= 8 * 1024 * 1024, ( + f"{high} bytes is either too tight to keep the link busy or too slack " + "to bound the delay") -- cgit v1.2.3