diff options
Diffstat (limited to 'packages/meshbay-node/src')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 305 |
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) |