path: root/packages/meshbay-node/tests/test_task_lifetime.py

"""
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 only a handful, so a few 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. It was two slots at the
time, which is how few it took.

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(encoding="utf-8")


@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 "self._tasks.discard(" in spawn and "add_done_callback(" 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_the_slot_count_matches_what_a_slot_now_costs(source):
    """States the number, so a change is deliberate rather than drifted into.

    It was two, and two was right while a stream was a burst: the client took
    segments as fast as it could append them and the slot came back within the
    minute. Bounding the read-ahead to ninety seconds of film (the buffer
    ceiling fix) changed what a slot is — it is now held for as long as someone
    is watching, so the count is a count of simultaneous viewers.
    """
    n = int(re.search(r"MAX_CONCURRENT_TRANSCODES\s*=\s*(\d+)", source).group(1))
    assert n >= 8, (
        f"the cap is {n}; with a slot held for the length of a film, that is "
        f"{n} people watching before the node refuses everyone else")
    assert "for as long as the film runs" in source, (
        "the number moved but the comment explaining what a slot costs did not")


# ── 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.
    """
    start = session.index("async def _do_file_request")
    # Up to whatever the next member is. This used to end at
    # "\n    def _do_stream_segment" — a neighbour removed in MNP 2.0 — and an
    # `index()` on a name that no longer exists fails the test for a reason
    # that has nothing to do with what it is about.
    nxt = re.search(r"\n    (?:@|(?:async )?def )", session[start:])
    fn = session[start:start + nxt.start()] if nxt else session[start:]
    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")