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>

3 files changed, 551 insertions, 43 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index a892be2..91d801c 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -109,7 +109,19 @@ JOIN_FAILURE_WINDOW = 600  # seconds
 UPLOAD_DIR_NAME = "uploads"
 # Conservative allowlist: also what keeps markup out of filenames, which the node admin
 # UI used to render unescaped (finding H2).
-SAFE_UPLOAD_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._ -]{0,127}$")
+# An allowlist, still — C5a and H2 depend on it — but one that does not assume
+# the world writes in ASCII. `été.txt` and `rapport (1).pdf` were refused, and
+# the second of those is a name _free_name generates itself, so the node was
+# rejecting files it had named. `\w` is Unicode here, which admits letters and
+# digits of any script while `<`, `>`, `"`, `;`, `/`, `\` and control characters
+# stay out. The first character must be a letter or digit, so ".." and dotfiles
+# cannot start one, and a trailing space or dot is refused because it makes two
+# different files look identical in a list.
+SAFE_UPLOAD_NAME = re.compile(
+    r"^[^\W_]"                            # letter or digit — never '.', '-' or space
+    r"[\w .\-()\[\]'\u2019,&+#@]{0,127}"  # body: word chars plus mild punctuation
+    r"(?<![ .])$",                        # and never ending on a space or a dot
+    re.UNICODE)
 
 
 def _free_name(directory: Path, filename: str) -> str:
@@ -172,10 +184,18 @@ def _extract_dtls_fingerprint(sdp: str) -> bytes:
 
 
 STREAM_SEGMENT_SIZE = 256 * 1024
+# A chunk is a megabyte and the browser keeps eight in flight, so answering them
+# as they arrive queues 8 MB on the channel with nothing watching. On a LAN that
+# drains before anyone notices; on a phone that is also uploading, it is minutes
+# of head-of-line delay for the reader. Above this, wait for room.
+DOWNLOAD_BUFFER_HIGH = 2 * 1024 * 1024
 # What a client may ask for in one go, and how long the node waits for it to ask
 # again before deciding nobody is watching any more.
 STREAM_MAX_CREDIT = 256
 STREAM_CREDIT_TIMEOUT = 120
+# How often that budget is re-examined. A viewer who left stops being
+# charged for a slot within this, rather than within the timeout.
+STREAM_CREDIT_POLL = 3
 
 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"}
 
@@ -283,6 +303,14 @@ class WebRTCPeerSession:
     def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""):
         self._pc = pc
         self._ctx = node_ctx
+        # Every background task this session starts. asyncio keeps only a *weak*
+        # reference to a task, so one that is merely fired and forgotten can be
+        # collected while it is still running — "Task was destroyed but it is
+        # pending!" in the log. For _stream_video that meant its `async with
+        # sem` never reached __aexit__ and the transcode slot was gone for good.
+        # There are two slots: after two abandoned streams the node answered
+        # "Server busy" to everything and no video would start at all.
+        self._tasks: set[asyncio.Task] = set()
         self._channel: RTCDataChannel | None = None
         self._buffer = _DataChannelBuffer(max_message=PRE_HANDSHAKE_MAX_MSG)
         self._pre_proof_fetches = 0
@@ -298,6 +326,13 @@ class WebRTCPeerSession:
         self._stream_credit = 0
         self._stream_credit_evt = asyncio.Event()
         self._stream_stopped = False
+        # The stream this session currently owns. One viewer plays one film at
+        # a time, so a second request means the first is over — see
+        # _replace_stream for why waiting for it to time out is not an option.
+        self._stream_task: asyncio.Task | None = None
+        # Diagnostics only: when the current stream began and how far it got.
+        self._stream_started_at: float = 0.0
+        self._stream_segments: int = 0
         self._gek_challenge: bytes | None = None
         # Same value as the GEK challenge, but kept for the life of the connection:
         # a join_request is signed over it, and it must stay verifiable after the
@@ -342,27 +377,35 @@ class WebRTCPeerSession:
                     return
                 self._audit_pre_proof_fetch(mtype)
                 if mtype == MNP.GEK_BUNDLE_FETCH:
-                    asyncio.ensure_future(self._do_gek_bundle_fetch())
+                    self._spawn(self._do_gek_bundle_fetch())
                 else:
-                    asyncio.ensure_future(self._do_keypair_bundle_fetch())
+                    self._spawn(self._do_keypair_bundle_fetch())
             elif mtype == MNP.JOIN_REQUEST and self._nonce_node:
                 # Valid both before the GEK proof (a new member has no GEK to prove
                 # with) and after it (an operator pairing a browser is already
                 # connected). Authority comes from the pairing code and the
                 # signature, never from the session state.
-                asyncio.ensure_future(self._do_join_request(msg))
+                self._spawn(self._do_join_request(msg))
             elif self._user_id is None:
                 self._send({"type": "error", "detail": "Handshake required"})
             elif mtype == MNP.INDEX_SYNC:
                 self._do_index_sync()
             elif mtype == MNP.FILE_REQUEST:
-                self._do_file_request(msg)
+                # Spawned rather than answered inline: the reply waits for room
+                # on the channel, and blocking the message loop for that would
+                # stop everything else this peer is doing — including the
+                # uploads whose acks free the very buffer we are waiting on.
+                # Chunks are matched by file and index on the client, so
+                # answering out of order is safe.
+                self._spawn(self._do_file_request(msg))
             elif mtype == MNP.STREAM_SEGMENT:
                 self._do_stream_segment(msg)
             elif mtype == MNP.CHAT_MESSAGE:
                 self._do_chat_message(msg)
             elif mtype == MNP.CHAT_HISTORY:
                 self._do_chat_history(msg)
+            elif mtype == MNP.PING:
+                self._do_ping(msg)
             elif mtype == MNP.FILE_UPLOAD:
                 self._do_file_upload(msg)
             elif mtype == MNP.DIR_CREATE:
@@ -378,14 +421,24 @@ class WebRTCPeerSession:
             elif mtype == MNP.MEMBER_REVOKE:
                 self._do_member_revoke(msg)
             elif mtype == MNP.KEYPAIR_BUNDLE_STORE:
-                asyncio.ensure_future(self._do_keypair_bundle_store(msg))
+                self._spawn(self._do_keypair_bundle_store(msg))
             elif mtype == MNP.KEYPAIR_BUNDLE_DELETE:
-                asyncio.ensure_future(self._do_keypair_bundle_delete())
+                self._spawn(self._do_keypair_bundle_delete())
             elif mtype == MNP.STREAM_REQUEST:
-                asyncio.ensure_future(self._stream_video(msg))
+                sem = self._ctx.get("_transcode_sem")
+                log.info("stream: req file=%s credits=%s slots_free=%s prev=%s",
+                         str(msg.get("file_id"))[:12], msg.get("credits"),
+                         getattr(sem, "_value", "?"),
+                         "alive" if (self._stream_task and
+                                     not self._stream_task.done()) else "none")
+                self._spawn(self._replace_stream(msg))
             elif mtype == MNP.STREAM_MORE:
                 self._grant_stream_credit(msg)
             elif mtype == MNP.STREAM_STOP:
+                age = (time.monotonic() - self._stream_started_at
+                       if self._stream_started_at else -1)
+                log.info("stream: stop received %.1fs after start, %d segments sent",
+                         age, self._stream_segments)
                 self._stop_stream()
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
@@ -400,7 +453,7 @@ class WebRTCPeerSession:
         if audit and self._user_id:
             if not self._remote_ip:
                 self._remote_ip = _get_remote_ip(self._pc)
-            asyncio.ensure_future(audit.log_event(
+            self._spawn(audit.log_event(
                 user_id=self._user_id,
                 event=event,
                 ip=self._remote_ip,
@@ -521,7 +574,7 @@ class WebRTCPeerSession:
         self._user_id = self._pending_sub
         self._group_id = self._pending_group
         self._username = self._pending_username
-        asyncio.ensure_future(self._load_pinned_pk())
+        self._spawn(self._load_pinned_pk())
 
         self._peer_registry()[self._user_id] = self
 
@@ -687,7 +740,7 @@ class WebRTCPeerSession:
         if not audit:
             return
         self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
-        asyncio.ensure_future(audit.log_event(
+        self._spawn(audit.log_event(
             user_id=self._user_id or getattr(self, "_pending_sub", "unknown"),
             event=event,
             ip=self._remote_ip,
@@ -920,7 +973,8 @@ class WebRTCPeerSession:
         ctx = self._group_ctx()
         shared_root = ctx.get("shared_root")
         if not shared_root:
-            self._send({"type": "error", "detail": "No shared directory"})
+            self._send({"type": "error", "detail": "No shared directory",
+                        "filename": filename})
             return
 
         name = str(msg.get("name", "")).strip()
@@ -964,7 +1018,8 @@ class WebRTCPeerSession:
         ctx = self._group_ctx()
         shared_root = ctx.get("shared_root")
         if not shared_root:
-            self._send({"type": "error", "detail": "No shared directory"})
+            self._send({"type": "error", "detail": "No shared directory",
+                        "filename": filename})
             return
 
         target = safe_subdir(shared_root, msg.get("dir") or "")
@@ -1105,7 +1160,7 @@ class WebRTCPeerSession:
         if not audit:
             return
         self._remote_ip = self._remote_ip or _get_remote_ip(self._pc)
-        asyncio.ensure_future(audit.log_event(
+        self._spawn(audit.log_event(
             user_id=getattr(self, "_pending_sub", "unknown"),
             event="pre_proof_fetch",
             ip=self._remote_ip,
@@ -1118,7 +1173,7 @@ class WebRTCPeerSession:
         audit = self._ctx.get("audit_store")
         if audit:
             self._remote_ip = _get_remote_ip(self._pc)
-            asyncio.ensure_future(audit.log_event(
+            self._spawn(audit.log_event(
                 user_id="unknown",
                 event="auth_failed",
                 ip=self._remote_ip,
@@ -1126,6 +1181,19 @@ class WebRTCPeerSession:
                 detail=reason,
             ))
 
+    def _spawn(self, coro) -> asyncio.Task:
+        """Run a coroutine in the background and hold on to it.
+
+        The reference is what keeps the task alive; the done callback is what
+        stops the set growing. Anything that owns a resource for its lifetime —
+        a transcode slot, an ffmpeg process — must go through here rather than
+        `asyncio.ensure_future`.
+        """
+        task = asyncio.ensure_future(coro)
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+        return task
+
     def _group_ctx(self) -> dict:
         if "groups" in self._ctx and self._group_id:
             return self._ctx["groups"][self._group_id]
@@ -1184,7 +1252,7 @@ class WebRTCPeerSession:
             return []
         return out[:2000]
 
-    def _do_file_request(self, msg: dict) -> None:
+    async def _do_file_request(self, msg: dict) -> None:
         ctx = self._group_ctx()
         file_id = msg["file_id"]
         chunk_index = msg["chunk_index"]
@@ -1199,6 +1267,9 @@ class WebRTCPeerSession:
             self._send({"type": "error", "detail": "File not on disk"})
             return
 
+        log.debug("dl: req file=%s chunk=%s buffered=%s",
+                 file_id[:12], chunk_index,
+                 getattr(self._channel, "bufferedAmount", "?"))
         file_hash = bytes.fromhex(entry.id)
         chunk_data = _read_and_encrypt(
             self._ctx["sk_node"],
@@ -1208,12 +1279,27 @@ class WebRTCPeerSession:
             file_hash,
             entry.id,
         )
+        # Backpressure. Without it the node hands the whole window to the
+        # channel at once and the reader sees the first chunk, then nothing for
+        # as long as the link takes to drain the rest.
+        waited = 0.0
+        while (self._channel is not None
+               and getattr(self._channel, "bufferedAmount", 0) > DOWNLOAD_BUFFER_HIGH
+               and self._channel.readyState == "open"
+               and waited < 60):
+            await asyncio.sleep(0.05)
+            waited += 0.05
+        if self._channel is None or self._channel.readyState != "open":
+            return
         self._send(chunk_data)
+        log.debug("dl: sent file=%s chunk=%s bytes=%s buffered=%s",
+                 file_id[:12], chunk_index, len(chunk_data.get("ct") or b""),
+                 getattr(self._channel, "bufferedAmount", "?"))
         if chunk_index == 0:
             self._audit("file_download", entry.name)
 
     def _do_stream_segment(self, msg: dict) -> None:
-        asyncio.ensure_future(self._do_stream_segment_async(msg))
+        self._spawn(self._do_stream_segment_async(msg))
 
     async def _do_stream_segment_async(self, msg: dict) -> None:
         """
@@ -1291,7 +1377,7 @@ class WebRTCPeerSession:
             self._user_names()[self._user_id] = sender_name
         if chat_store:
             raw = payload.encode() if isinstance(payload, str) else payload
-            asyncio.ensure_future(chat_store.save_message(
+            self._spawn(chat_store.save_message(
                 sender_id=self._user_id,
                 iteration=msg.get("iteration", 0),
                 payload=raw,
@@ -1320,7 +1406,7 @@ class WebRTCPeerSession:
         if hub_ws and self._group_id:
             try:
                 import json as _json
-                asyncio.ensure_future(hub_ws.send(_json.dumps({
+                self._spawn(hub_ws.send(_json.dumps({
                     "type": "chat_notify",
                     "group_id": self._group_id,
                     "sender_name": sender_name,
@@ -1336,6 +1422,15 @@ class WebRTCPeerSession:
         self._send({"type": "ack", "v": MNP_VERSION})
         self._audit("chat_message")
 
+    def _do_ping(self, msg: dict) -> None:
+        """Answer a liveness probe on an open channel, echoing the caller's token.
+
+        Echoed rather than bare so a client can match the answer to the probe it
+        sent and measure a round trip, instead of being reassured by a reply to
+        some earlier one.
+        """
+        self._send({"type": MNP.PONG, "v": MNP_VERSION, "token": msg.get("token")})
+
     def _do_chat_history(self, msg: dict) -> None:
         chat_store = self._group_ctx().get("chat_store")
         if not chat_store:
@@ -1343,19 +1438,30 @@ class WebRTCPeerSession:
                 "type": MNP.CHAT_HISTORY_RESPONSE,
                 "v": MNP_VERSION,
                 "messages": [],
+                "has_more": False,
             })
             return
 
-        since = msg.get("since", 0)
-        limit = msg.get("limit", 100)
-        asyncio.ensure_future(self._send_chat_history(chat_store, since, limit))
+        # `before` pages backwards from the newest, which is the direction a chat
+        # is actually read. `since` remains for callers that want everything
+        # after a point in time; the browser no longer uses it.
+        before = msg.get("before")
+        limit = max(1, min(int(msg.get("limit", 100)), 200))
+        self._spawn(self._send_chat_history(chat_store, before, limit))
 
-    async def _send_chat_history(self, chat_store, since: float, limit: int) -> None:
-        msgs = await chat_store.get_messages(since=since, limit=limit)
+    async def _send_chat_history(self, chat_store, before, limit: int) -> None:
+        if before:
+            msgs = await chat_store.get_before(int(before), limit=limit)
+        else:
+            msgs = await chat_store.get_recent(limit=limit)
+        # Whether the "load older" control has anything left to fetch. Asked
+        # about the oldest row returned, so an empty page correctly says no.
+        has_more = await chat_store.has_before(msgs[0].id) if msgs else False
         names = self._user_names()
         self._send({
             "type": MNP.CHAT_HISTORY_RESPONSE,
             "v": MNP_VERSION,
+            "has_more": has_more,
             "messages": [
                 {
                     "id": m.id,
@@ -1378,16 +1484,19 @@ class WebRTCPeerSession:
         data = msg.get("data")
 
         if not filename or data is None:
-            self._send({"type": "error", "detail": "Missing filename or data"})
+            self._send({"type": "error", "detail": "Missing filename or data",
+                        "filename": filename})
             return
 
         if not SAFE_UPLOAD_NAME.match(filename):
-            self._send({"type": "error", "detail": "Invalid filename"})
+            self._send({"type": "error", "detail": "Invalid filename",
+                        "filename": filename})
             return
 
         shared_root = ctx.get("shared_root")
         if not shared_root:
-            self._send({"type": "error", "detail": "No shared directory"})
+            self._send({"type": "error", "detail": "No shared directory",
+                        "filename": filename})
             return
 
         # One destination, chosen here and not by the client: uploads/ at the root
@@ -1412,18 +1521,21 @@ class WebRTCPeerSession:
             # Backstop: _free_name already guarantees this, and it stays because
             # it asserts the invariant where the write happens.
             if final_path.exists():
-                self._send({"type": "error", "detail": "File already exists"})
+                self._send({"type": "error", "detail": "File already exists",
+                        "filename": filename})
                 return
             state = {"next_index": 0, "bytes": 0, "stored_name": stored_name}
             self._uploads[upload_key] = state
         elif state is None:
-            self._send({"type": "error", "detail": "Upload not started"})
+            self._send({"type": "error", "detail": "Upload not started",
+                        "filename": filename})
             return
 
         # Reject out-of-order or replayed chunks — otherwise chunk_index>0 appends
         # blindly to whatever .part file is already on disk.
         if chunk_index != state["next_index"]:
-            self._send({"type": "error", "detail": "Unexpected chunk index"})
+            self._send({"type": "error", "detail": "Unexpected chunk index",
+                        "filename": filename})
             return
 
         if isinstance(data, str):
@@ -1434,7 +1546,8 @@ class WebRTCPeerSession:
         if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES:
             self._uploads.pop(upload_key, None)
             tmp_path.unlink(missing_ok=True)
-            self._send({"type": "error", "detail": "Upload exceeds size limit"})
+            self._send({"type": "error", "detail": "Upload exceeds size limit",
+                        "filename": filename})
             return
 
         with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f:
@@ -1616,16 +1729,16 @@ class WebRTCPeerSession:
         )
 
         if pending["op"] == OP_FILE_DELETE:
-            asyncio.ensure_future(
+            self._spawn(
                 self._admin_exec_file_delete(pending, transcript, sig_bytes))
         elif pending["op"] == OP_DIR_DELETE:
-            asyncio.ensure_future(
+            self._spawn(
                 self._admin_exec_dir_delete(pending, transcript, sig_bytes))
         elif pending["op"] == OP_MEMBER_REVOKE:
-            asyncio.ensure_future(
+            self._spawn(
                 self._admin_exec_member_revoke(pending, transcript, sig_bytes))
         elif pending["op"] == OP_INVITE_CREATE:
-            asyncio.ensure_future(
+            self._spawn(
                 self._admin_exec_invite_create(pending, transcript, sig_bytes))
         else:
             self._send({"type": "error", "detail": "Unknown admin operation"})
@@ -1717,6 +1830,8 @@ class WebRTCPeerSession:
         })
 
     def _grant_stream_credit(self, msg: dict) -> None:
+        log.debug("stream credit +%s (had %d, sent %d)",
+                  msg.get("n"), self._stream_credit, self._stream_segments)
         """The client has room for more segments."""
         try:
             n = int(msg.get("n", 1))
@@ -1745,17 +1860,29 @@ class WebRTCPeerSession:
         fast as it is produced, and the browser holds a four gigabyte film in a
         JavaScript array while MediaSource consumes it a segment at a time.
         """
+        waited = 0.0
         while self._stream_credit <= 0:
             if self._stream_stopped:
                 return False
+            # Checked before the wait as well as after it: a peer that vanishes
+            # sends no credit and fires no event, so waiting the full timeout
+            # on a channel that is already shut is pure dead time on a slot.
+            if self._channel is None or self._channel.readyState != "open":
+                return False
             self._stream_credit_evt.clear()
             try:
+                # In slices rather than one long sleep, so a connection that
+                # dies mid-wait is noticed in seconds instead of minutes. The
+                # total budget is unchanged.
                 await asyncio.wait_for(self._stream_credit_evt.wait(),
-                                       timeout=STREAM_CREDIT_TIMEOUT)
+                                       timeout=STREAM_CREDIT_POLL)
             except asyncio.TimeoutError:
-                log.info("Stream stalled: no credit from peer=%s",
-                         (self._user_id or "?")[:8])
-                return False
+                waited += STREAM_CREDIT_POLL
+                if waited >= STREAM_CREDIT_TIMEOUT:
+                    log.info("Stream stalled: no credit from peer=%s",
+                             (self._user_id or "?")[:8])
+                    return False
+                continue
             if self._stream_stopped:
                 return False
             if self._channel is None or self._channel.readyState != "open":
@@ -1763,6 +1890,40 @@ class WebRTCPeerSession:
         self._stream_credit -= 1
         return True
 
+    async def _replace_stream(self, msg: dict) -> None:
+        """Retire this session's previous stream before starting another.
+
+        A viewer plays one film at a time, so a second request means the first
+        one is finished whatever the client managed to tell us. Relying on
+        `stream_stop` alone was not enough: a browser that is backgrounded,
+        reloaded or simply loses the message never sends it, and the only other
+        thing that ends a stream is STREAM_CREDIT_TIMEOUT — two minutes during
+        which ffmpeg keeps running and holds one of the node's two transcode
+        slots.
+
+        That is the reported failure exactly: first video fine, second fine,
+        third answered "Server busy" because the first two were still holding
+        both slots. The client shows that as "buffering" forever.
+
+        Waiting for the old task is what makes the slot available: it is the
+        exit of its `async with sem` that releases it.
+        """
+        prev = self._stream_task
+        if prev is not None and not prev.done():
+            t0 = time.monotonic()
+            log.info("stream: retiring previous stream")
+            self._stop_stream()
+            try:
+                await asyncio.wait_for(asyncio.shield(prev), timeout=15)
+                log.info("stream: previous stream ended in %.1fs",
+                         time.monotonic() - t0)
+            except asyncio.TimeoutError:
+                log.warning("stream: previous stream STILL RUNNING after 15s")
+            except Exception:
+                pass   # it failed on its own; the slot is free either way
+        self._stream_task = asyncio.current_task()
+        await self._stream_video(msg)
+
     async def _stream_video(self, msg: dict) -> None:
         """Stream a video file as fMP4 segments via MSE-compatible output."""
         # One ffmpeg per request with no cap lets any member exhaust the node's
@@ -1775,8 +1936,13 @@ class WebRTCPeerSession:
         if sem.locked() and sem._value <= 0:
             self._send({"type": "error", "detail": "Server busy, retry shortly"})
             return
+        log.info("stream: waiting for a slot (free=%s)", sem._value)
         async with sem:
-            await self._stream_video_inner(msg)
+            log.info("stream: slot acquired (free=%s)", sem._value)
+            try:
+                await self._stream_video_inner(msg)
+            finally:
+                log.info("stream: slot released (free=%s)", sem._value + 1)
 
     async def _stream_video_inner(self, msg: dict) -> None:
         ctx = self._group_ctx()
@@ -1832,11 +1998,18 @@ class WebRTCPeerSession:
         self._stream_stopped = False
 
         index = 0
+        self._stream_started_at = time.monotonic()
+        self._stream_segments = 0
+        reason = "eof"
+        log.info("stream: stream_init sent file=%s paced=%s credits=%d",
+                 file_id[:12], paced, self._stream_credit)
         try:
             while True:
                 if paced and not await self._await_stream_credit():
+                    reason = "no-credit-or-gone"
                     break
                 if self._stream_stopped:
+                    reason = "stopped-by-peer"
                     log.info("Stream stopped by peer=%s after %d segments",
                              (self._user_id or "?")[:8], index)
                     break
@@ -1855,6 +2028,7 @@ class WebRTCPeerSession:
                     "plaintext_size": len(data),
                 })
                 index += 1
+                self._stream_segments = index
                 await asyncio.sleep(0)
         except Exception as e:
             log.error("Stream error: %s", e)
@@ -1863,7 +2037,29 @@ class WebRTCPeerSession:
                 proc.kill()
             except ProcessLookupError:
                 pass
-            await proc.wait()
+            # `await proc.wait()` on its own is the deadlock the asyncio docs
+            # warn about: ffmpeg fills the stdout pipe we have stopped reading,
+            # and the transport cannot finish closing until that buffer is
+            # drained. Measured on 2026-08-16 with stream: — a viewer closed
+            # the player after 99 segments (25 MB) and the task sat here past
+            # the 15 s handover timeout, holding a transcode slot. The node has
+            # two, so the next video waited and the one after was refused.
+            #
+            # Drain first, then wait with a bound. The slot must come back even
+            # if the process is being stubborn: it has already had SIGKILL, and
+            # the OS will reap it whether or not we are still watching.
+            for pipe in (proc.stdout, proc.stderr):
+                if pipe is None:
+                    continue
+                try:
+                    await asyncio.wait_for(pipe.read(), timeout=2)
+                except Exception:
+                    pass
+            try:
+                await asyncio.wait_for(proc.wait(), timeout=5)
+            except Exception:
+                log.warning("stream: ffmpeg did not reap in 5s — "
+                            "releasing the slot regardless")
 
         if not self._stream_stopped:
             self._send({
@@ -1871,6 +2067,8 @@ class WebRTCPeerSession:
                 "v": MNP_VERSION,
                 "file_id": file_id,
             })
+        log.info("stream: stream ended reason=%s segments=%d after %.1fs",
+                 reason, index, time.monotonic() - self._stream_started_at)
         log.info("Streamed %s: %d segments", entry.name, index)
         self._audit("stream_video", entry.name)
 
@@ -1881,10 +2079,25 @@ class WebRTCPeerSession:
             log.warning("WebRTC send skipped: channel=%s",
                         self._channel.readyState if self._channel else "none")
 
+    async def shutdown_tasks(self) -> None:
+        """Stop everything this session is doing and give back what it holds.
+
+        Separate from close() because the connection-state handler runs while
+        aiortc is already tearing the peer connection down — calling pc.close()
+        from in there would re-enter it. What matters for the transcode slot is
+        here: cancelling the task runs the exit of its `async with sem`.
+        """
+        self._stop_stream()
+        for task in list(self._tasks):
+            task.cancel()
+        if self._tasks:
+            await asyncio.gather(*self._tasks, return_exceptions=True)
+
     async def close(self) -> None:
         self._audit("disconnect")
         if self._user_id:
             self._peer_registry().pop(self._user_id, None)
+        await self.shutdown_tasks()
         await self._pc.close()
 
 
@@ -1981,7 +2194,15 @@ class WebRTCTransport:
             state = pc.connectionState
             log.info("WebRTC connection state: %s (peer=%s)", state, peer_id)
             if state in ("failed", "closed"):
-                self._sessions.pop(peer_id, None)
+                gone = self._sessions.pop(peer_id, None)
+                if gone is not None:
+                    # Popping only forgets the session. Its stream went on
+                    # transcoding until the credit timeout — measured at 91s
+                    # after the connection closed — holding one of the node's
+                    # two slots the whole time. Closing the viewer, the tab or
+                    # the browser all arrive here, so this is the one place
+                    # that covers every way of walking away.
+                    await gone.shutdown_tasks()
 
         offer = RTCSessionDescription(sdp=offer_sdp, type="offer")
         await pc.setRemoteDescription(offer)
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")