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>

1 files changed, 263 insertions, 42 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)