fix(node): stop losing transcode slots, and reap ffmpeg without deadlocking

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 <noreply@anthropic.com>

2 files changed, 288 insertions, 1 deletions

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