diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 548 |
1 files changed, 531 insertions, 17 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 dfabe9b..507650a 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -70,6 +70,7 @@ from meshbay_common.adminop import ( OP_MEMBER_UPLOAD, OP_APPS_ENABLED, OP_SET_SCAN_SETTINGS, + OP_TRANSFER_LIMITS, OP_TMDB_CONFIG, OP_TMDB_ENABLED, OP_VIDEO_ROOT, @@ -119,6 +120,7 @@ from meshbay_common.protocol import ( MNP, chunk_ciphertext, file_chunk_wire, + UPLOAD_PROBE_INDEX, file_upload_ack_wire, file_upload_payload, ) @@ -127,6 +129,9 @@ from meshbay_node.transport.wire import index_sync_message from meshbay_node.indexer import GroupIndex from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node import linkpreview, ops, platform +from meshbay_node import transfers as transfers_mod +from meshbay_node import uploads as uploads_mod +from meshbay_node.transfers import TransferSlots # Re-imported under its original name: every call site and existing test in # this module still refers to it as `_probe_video`. The implementation lives # in media_probe.py so the indexer package (imported just above) can call it @@ -258,6 +263,11 @@ 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 +# How often transfer leases are swept. Nothing depends on it being +# prompt -- the session teardown is the reclaim that matters and is +# immediate; this catches peers that vanished without the connection +# noticing, so it trades latency for a timer that hardly ever runs. +TRANSFER_SWEEP_SECS = 15 def _pack(obj: dict) -> bytes: @@ -418,7 +428,13 @@ class WebRTCPeerSession: self._join_attempts = 0 self._nonce_client: bytes = b"" self._admin_ops: dict[str, dict] = {} # op_id → pending admin operation - self._uploads: dict[str, dict] = {} # filename → {next_index, bytes} + # Uploads in progress live in the group context, not here: see + # `_partial_uploads` and `uploads.py`. + # + # Leaseless reads, though, *are* this connection's: the bound is on what + # one session may do while claiming to be browsing, not a pool shared + # between them. Three tabs open is browsing in three tabs. + self._leaseless = transfers_mod.LeaselessReads() # Diagnostics only (_WEBRTC_TRACE): when the last DataChannel message # arrived, so the heartbeat can report silence duration. self._last_msg_at: float = 0.0 @@ -524,6 +540,10 @@ class WebRTCPeerSession: self._spawn(self._do_link_preview_request(msg)) elif mtype == MNP.PING: self._do_ping(msg) + elif mtype == MNP.TRANSFER_OPEN: + self._do_transfer_open(msg) + elif mtype == MNP.TRANSFER_CLOSE: + self._do_transfer_close(msg) elif mtype == MNP.FILE_UPLOAD: self._do_file_upload(msg) elif mtype == MNP.DIR_CREATE: @@ -554,6 +574,8 @@ class WebRTCPeerSession: self._do_member_upload(msg) elif mtype == MNP.APPS_ENABLED: self._do_apps_enabled(msg) + elif mtype == MNP.TRANSFER_LIMITS: + self._do_transfer_limits(msg) elif mtype == MNP.SET_SCAN_SETTINGS: self._do_set_scan_settings(msg) elif mtype == MNP.TMDB_CONFIG: @@ -922,6 +944,19 @@ class WebRTCPeerSession: # No `chat_encrypted` beside it: there is no switch. A peer that # reached this point speaks MNP 2.0, and 2.0 has no plaintext chat. "chat_epoch": int(self._group_ctx().get("chat_epoch", 0) or 0), + # This member's own transfer caps in this group, so the interface + # can say "2 of 2 of your slots are busy" rather than draw a bare + # spinner. Absent reads as "no limit known" and the hint is simply + # not drawn — never as "unlimited", which would have the interface + # contradicting the node. + "transfer_limits": { + "download": self._slots().member_cap( + transfers_mod.DOWNLOAD, + (self._group_id or "", self._user_id or "")), + "upload": self._slots().member_cap( + transfers_mod.UPLOAD, + (self._group_id or "", self._user_id or "")), + }, # So a client that connects mid-scan shows the indexing state # immediately, instead of waiting for the next periodic # INDEX_PROGRESS push. Never a path or filename — see @@ -2704,6 +2739,64 @@ class WebRTCPeerSession: self._issue_admin_challenge( OP_SET_SCAN_SETTINGS, f"{reconcile:g},{debounce:g}") + MIN_TRANSFER_LIMIT = 1 + MAX_TRANSFER_LIMIT = 32 + + def _do_transfer_limits(self, msg: dict) -> None: + """How many transfers one member may run at once in this group. + + Zero is not "unlimited" and is refused: a member who may not transfer at + all is a member the operator revokes, and reading 0 as no-limit would + make the most dangerous value the easiest to type by accident. + """ + try: + downloads = int(msg.get("downloads")) + uploads = int(msg.get("uploads")) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid transfer limits"}) + return + for value in (downloads, uploads): + if not (self.MIN_TRANSFER_LIMIT <= value <= self.MAX_TRANSFER_LIMIT): + self._send({"type": "error", + "detail": f"transfer limits must be between " + f"{self.MIN_TRANSFER_LIMIT} and " + f"{self.MAX_TRANSFER_LIMIT}"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge(OP_TRANSFER_LIMITS, + f"d={downloads},u={uploads}") + + async def _admin_exec_transfer_limits( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + try: + parts = dict(p.split("=") for p in pending["subject"].split(",")) + downloads, uploads = int(parts["d"]), int(parts["u"]) + except (ValueError, KeyError): + self._send({"type": "error", "detail": "Invalid transfer limits"}) + return + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"transfer_limits:{pending['subject']}") + return + try: + result = await self._run_op( + ops.set_transfer_limits, self._group_id or "", downloads, uploads) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("transfer_limits", pending["subject"]) + + notice = {"type": MNP.TRANSFER_LIMITS_ACK, "v": MNP_VERSION, + "limits": result["limits"]} + for session in list(self._peer_registry().values()): + try: + session._send(notice) + except Exception: + pass + async def _admin_exec_set_scan_settings( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: @@ -3309,6 +3402,199 @@ class WebRTCPeerSession: "total_bytes": progress.total_bytes, } + # ── Transfer slots ─────────────────────────────────────────────────────── + + def _slots(self) -> "TransferSlots": + """The node's transfer pools, shared across every peer and every group. + + On the transport context, not the session: it counts the node's + transfers, not one browser's. Built once, for the same reason the + transcode semaphore is — rebuilding it per call would hand every caller + its own budget and cap nothing at all. + """ + slots = self._ctx.get("_transfer_slots") + if slots is None: + slots = TransferSlots() + n = self._ctx.get("max_concurrent_downloads") + u = self._ctx.get("max_concurrent_uploads") + if n: + slots.caps[transfers_mod.DOWNLOAD] = int(n) + if u: + slots.caps[transfers_mod.UPLOAD] = int(u) + self._ctx["_transfer_slots"] = slots + log.info("transfer: %s", slots.summary()) + # Refreshed from the group context rather than only at construction: a + # node serves several groups, each with its own signed cap, and the + # pools are built by whichever group happens to transfer first. + limits = self._group_ctx().get("transfer_limits") + if limits and self._group_id: + slots.group_limits[self._group_id] = dict(limits) + return slots + + def _transfer_state_msg(self, lease, state: str, reason: str = "") -> dict: + slots = self._slots() + out = { + "type": MNP.TRANSFER_STATE, + "v": MNP_VERSION, + "tr": lease.tr, + "state": state, + "kind": lease.kind, + "used": slots.member_in_use(lease.kind, lease.member), + "cap": slots.per_member.get(lease.kind, + transfers_mod.DEFAULT_MAX_PER_MEMBER), + "node_used": slots.in_use(lease.kind), + "node_cap": slots.caps.get(lease.kind, + transfers_mod.DEFAULT_MAX_CONCURRENT), + } + if state == "queued": + out["ahead"] = slots.ahead_of(lease) + if reason: + out["reason"] = reason + return out + + def _notify_transfer(self, lease, state: str, reason: str = "") -> None: + """Push a lease's state to the connection that owns it. + + By session key, never by account: a lease belongs to one connection, and + telling a member's other device that *its* transfer was granted is how a + queue starts lying. + """ + session = self._peer_registry().get(lease.session_key) + for candidate in ([session] if session else + self._sessions_everywhere(lease.session_key)): + try: + candidate._send(self._transfer_state_msg(lease, state, reason)) + except Exception: + pass + + def _sessions_everywhere(self, session_key: str) -> list["WebRTCPeerSession"]: + """The session with this key, whichever group it is in. + + `_peer_registry` is per group (finding H1) and the pools are node-wide, + so a slot freed in one group can grant one in another: the peer to tell + is not necessarily in this session's own registry. + """ + groups = self._ctx.get("groups") + registries = ([g.get("_peers", {}) for g in groups.values()] + if groups else [self._ctx.get("_peers", {})]) + return [reg[session_key] for reg in registries if session_key in reg] + + def _announce(self, granted: list, ended: list | None = None) -> None: + for lease, reason in (ended or []): + self._notify_transfer( + lease, "queued" if lease.state == "queued" else "closed", reason) + for lease in granted: + self._notify_transfer(lease, "granted") + + def _do_transfer_open(self, msg: dict) -> None: + tr = str(msg.get("tr") or "")[:64] + kind = str(msg.get("kind") or transfers_mod.DOWNLOAD) + if not tr: + self._send({"type": "error", "detail": "Missing transfer id", + "code": "bad_transfer_id"}) + return + slots = self._slots() + try: + nbytes = int(msg.get("bytes") or 0) + chunks = int(msg.get("chunks") or 0) + except (TypeError, ValueError): + self._send({"type": "error", "detail": "Invalid transfer size", + "code": "bad_transfer_size", "tr": tr}) + return + lease, err = slots.open( + tr=tr, kind=kind, session_key=self._registry_key, + user_id=self._user_id or "", group_id=self._group_id or "", + bytes=nbytes, chunks=chunks) + if err: + self._send({"type": "error", "detail": err, "code": err, "tr": tr}) + return + self._send(self._transfer_state_msg(lease, lease.state)) + # INFO, not DEBUG. This is the line that answers "did the client ever + # ask for a slot, and what was it told" when somebody reports a stuck + # transfer — and a whole afternoon was spent concluding "the node saw + # nothing" from a journal that could not have shown it. One line per + # transfer is not a volume problem; turning the root logger up to DEBUG + # to see it is, because aiortc logs every SCTP chunk. + log.info("transfer: open %s %s -> %s (%s)", + kind, tr[:8], lease.state, slots.summary()) + self._ensure_transfer_sweeper() + + def _do_transfer_close(self, msg: dict) -> None: + tr = str(msg.get("tr") or "")[:64] + reason = str(msg.get("reason") or transfers_mod.REASON_DONE)[:32] + slots = self._slots() + held = slots.leases.get(tr) + if held is not None and held.session_key != self._registry_key: + # Closing somebody else's transfer would be a denial of service one + # random id away. + self._send({"type": "error", "detail": "not_your_transfer", + "code": "not_your_transfer", "tr": tr}) + return + lease, granted = slots.close(tr, reason) + if lease is not None: + self._send(self._transfer_state_msg(lease, "closed", reason)) + self._announce(granted) + + def _release_transfers(self) -> None: + """Give back everything this connection held. Called from teardown.""" + slots = self._ctx.get("_transfer_slots") + if slots is None: + return + gone, granted = slots.release_session(self._registry_key) + if gone: + log.info("transfer: session gone, released %d (%s)", + len(gone), slots.summary()) + self._announce(granted) + + def _ensure_transfer_sweeper(self) -> None: + """Start the maintenance task, once, and only while it has work. + + It reclaims what a session teardown cannot see — a grant nobody took up, + a transfer that went quiet — and logs the one line that answers "was + this peer ever in a queue" when somebody reports a stuck transfer. It + stops when the last lease goes, so an idle node runs no timer. + """ + running = self._ctx.get("_transfer_sweeper") + if running is not None and not running.done(): + return + + ctx = self._ctx + + async def _sweep_loop() -> None: + while True: + await asyncio.sleep(TRANSFER_SWEEP_SECS) + slots = ctx.get("_transfer_slots") + if slots is None or not slots.leases: + return + ended, granted = slots.sweep() + for lease, reason in ended: + log.info("transfer: reclaimed %s (%s)", lease.tr[:8], reason) + self._announce(granted, ended) + log.debug("transfer: %s", slots.summary()) + + # Deliberately NOT `self._spawn`, which is otherwise the only way to + # start a task here. `_spawn` ties a task to *this session's* set, and + # `shutdown_tasks` cancels those when the peer leaves — so the sweeper + # would die with whichever connection happened to open the first + # transfer, and every other peer's abandoned lease would then never be + # reclaimed. It belongs to the node, so the strong reference that keeps + # it off the garbage collector lives on the transport context; the rule + # `_spawn` exists for (asyncio holds only a weak reference) is satisfied + # by that reference, not by which set it is in. + task = asyncio.ensure_future(_sweep_loop()) + ctx["_transfer_sweeper"] = task + + def _finished(done: asyncio.Task) -> None: + if ctx.get("_transfer_sweeper") is done: + ctx["_transfer_sweeper"] = None + if not done.cancelled() and done.exception() is not None: + # Nothing awaits this task, so an exception here would otherwise + # be swallowed and idle leases would silently stop being + # reclaimed — the failure mode is a node that fills up over days. + log.error("transfer: sweeper died: %r", done.exception()) + + task.add_done_callback(_finished) + def _register_peer(self) -> None: """Add this connection to its group's peer set. @@ -3379,6 +3665,17 @@ class WebRTCPeerSession: async def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() + # A chunk request is what "this transfer is alive" looks like. Nothing + # marked a lease used, so `used` stayed False for the whole download and + # the sweeper revoked the grant every 30 s as never-taken-up — while the + # file was transferring at 20 MB/s. Found in the node's own log, which + # repeated the same two reclaims every 30 s for as long as the daemon + # ran. + tr = msg.get("tr") + if tr: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.touch(str(tr)[:64]) file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) @@ -3397,6 +3694,25 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not on disk"}) return + # A real index entry, asked for without a lease: browsing, or a client + # helping itself to the whole library outside every cap. + # + # Both look identical here — which is why the bound is a small count of + # files rather than a judgement about what the read is for. Thumbnails, + # posters and cover art never reach this line: they resolve through + # `_try_serve_thumbnail` above, out of a cache the node built itself, + # and are never leased, never counted, never queued. + if not tr: + if not self._leaseless.admit(str(file_id)): + self._send({ + "type": "error", + "detail": "Too many files open at once without a transfer. " + "Download this one instead of previewing it.", + "code": "transfer_required", + "file_id": file_id, + }) + return + log.debug("dl: req file=%s chunk=%s buffered=%s", file_id[:12], chunk_index, getattr(self._channel, "bufferedAmount", "?")) @@ -3421,6 +3737,11 @@ class WebRTCPeerSession: getattr(self._channel, "bufferedAmount", "?")) if chunk_index == 0: self._audit("file_download", entry.name) + # The last chunk is the only "close" a leaseless read has. Without this + # the session carries the entry until it goes idle, and the person who + # just looked at two photos cannot look at a third for a minute. + if not tr and (chunk_index + 1) * CHUNK_SIZE >= entry.size: + self._leaseless.finish(str(file_id)) @staticmethod async def _fetch_and_cache_poster(media_cache, tmdb_client, poster_path: str | None) -> str | None: @@ -4468,6 +4789,19 @@ class WebRTCPeerSession: if k not in ("type", "v")}) self._send(resp) + def _partial_uploads(self, ctx: dict) -> uploads_mod.PartialUploads: + """This group's uploads in progress, created on first use. + + In the group context rather than on the session, so a client that + reconnects finds its own upload where it left it — and so the reaper has + something to ask "is anyone still writing this?". + """ + store = ctx.get("partial_uploads") + if store is None: + store = uploads_mod.PartialUploads() + ctx["partial_uploads"] = store + return store + def _do_file_upload(self, msg: dict) -> None: """ One chunk of an upload, sealed under the group key (MNP 2.0). @@ -4488,6 +4822,30 @@ class WebRTCPeerSession: ctx = self._group_ctx() upload_id = str(msg.get("upload_id") or "")[:64] + # Say the slot is being used, chunk by chunk, exactly as `_do_file_req` + # does for a download. + # + # A grant nobody takes up is reclaimed after GRANT_DEADLINE_SECS and, on + # the third miss, abandoned. Uploads were not gated by the lease, so the + # file still arrived — but the widget follows the lease, so a 3.5 GB + # upload showed "waiting, 0 ahead" for a minute and a half while it was + # in fact transferring, and the node logged three reclaims against a + # transfer that never stopped. Measured, from the journal: + # + # 11:52:49 open upload 919ebf54 -> granted + # 11:53:19 reclaimed 919ebf54 (not_taken_up) + # 11:54:19 reclaimed 919ebf54 (abandoned) + # 11:55:48 Upload complete: ... (3 522 297 517 bytes) + # + # The download twin of this was fixed on 2026-09-08 (§12.1 of + # ~/next/improve-downloads.md); the same omission was still here, + # invisible until uploads started taking a real lease. + tr = msg.get("tr") + if tr: + slots = self._ctx.get("_transfer_slots") + if slots is not None: + slots.touch(str(tr)[:64]) + gek = ctx.get("gek") if not gek: self._send({"type": "error", "detail": "Group encryption not initialized", @@ -4623,43 +4981,78 @@ class WebRTCPeerSession: "root_unavailable") return - upload_key = f"{rel_dir}/{filename}" - state = self._uploads.get(upload_key) + # Held by the group, not by this connection. + # + # This used to be `self._uploads`, on the session. A dropped link threw + # the position away and the next chunk was refused with `not_started`: + # an upload interrupted at 99% could only be started again from zero, on + # a connection flaky enough to have interrupted it once. And the state + # it lost was the only thing that knew about the `.part` file left + # behind — see `uploads.orphaned_parts`, which is the other half of this. + # + # Keyed by member as well as by name, because a shared directory means + # two people can be sending IMG_1234.jpg at the same moment and neither + # may inherit the other's position. + uploads = self._partial_uploads(ctx) + user_id = self._user_id or "" + state = uploads.get(user_id, rel_dir, filename) # A shared directory means two people can send the same name. Refusing the # second is safe but silly — everyone's camera produces IMG_1234.jpg — so # a free name is found instead. Never a replacement. - stored_name = state["stored_name"] if state else _free_name(target_dir, filename) - tmp_path = target_dir / f"{stored_name}.part" + stored_name = state.stored_name if state else _free_name(target_dir, filename) + tmp_path = target_dir / f"{stored_name}{uploads_mod.PART_SUFFIX}" final_path = target_dir / stored_name + if chunk_index == UPLOAD_PROBE_INDEX: + # "Where am I?", asked inside the seal rather than on a clear + # message, because the answer is about a file whose name is exactly + # what sealing this path was for. + # + # It writes nothing, creates no state and reserves no name: a client + # that asks and then goes away has cost this node one reply. Every + # check above has already run, so it cannot be used to ask questions + # about a directory the caller may not write to. + self._send(file_upload_ack_wire( + gek, self._group_id or "", + upload_id=upload_id, + chunk_index=UPLOAD_PROBE_INDEX, + filename=filename, + # Only what is really on disk. Without state, `_free_name` above + # picked a name nothing has claimed yet, and reporting it would + # promise a destination the real chunk 0 may not choose. + stored_as=state.stored_name if state else "", + dir=rel_dir, + resume_from=state.next_index if state else 0, + )) + return + if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. if final_path.exists(): _refuse("File already exists", "already_exists") return - state = {"next_index": 0, "bytes": 0, "stored_name": stored_name} - self._uploads[upload_key] = state + state = uploads.start(user_id, rel_dir, filename, stored_name, + part_path=tmp_path) elif state is None: _refuse("Upload not started", "not_started") 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"]: + if chunk_index != state.next_index: _refuse("Unexpected chunk index", "bad_chunk_index") return - if state["bytes"] + len(chunk_bytes) > MAX_UPLOAD_BYTES: - self._uploads.pop(upload_key, None) + if state.bytes + len(chunk_bytes) > MAX_UPLOAD_BYTES: + uploads.drop(user_id, rel_dir, filename) tmp_path.unlink(missing_ok=True) _refuse("Upload exceeds size limit", "too_large") return with open(tmp_path, "wb" if chunk_index == 0 else "ab") as f: f.write(chunk_bytes) - state["next_index"] = chunk_index + 1 - state["bytes"] += len(chunk_bytes) + uploads.advance(user_id, rel_dir, filename, chunk_index, len(chunk_bytes)) self._send(file_upload_ack_wire( gek, self._group_id or "", @@ -4673,10 +5066,10 @@ class WebRTCPeerSession: )) if chunk_index + 1 >= total_chunks: - self._uploads.pop(upload_key, None) + uploads.drop(user_id, rel_dir, filename) tmp_path.rename(final_path) log.info("Upload complete: %s (%d chunks, %d bytes)", - stored_name, total_chunks, state["bytes"]) + stored_name, total_chunks, state.bytes) self._audit("file_upload", f"{rel_dir}/{stored_name}") self._register_uploader(ctx, rel_dir, stored_name) @@ -4932,6 +5325,9 @@ class WebRTCPeerSession: elif pending["op"] == OP_APPS_ENABLED: self._spawn( self._admin_exec_apps_enabled(pending, transcript, sig_bytes)) + elif pending["op"] == OP_TRANSFER_LIMITS: + self._spawn( + self._admin_exec_transfer_limits(pending, transcript, sig_bytes)) elif pending["op"] == OP_SET_SCAN_SETTINGS: self._spawn( self._admin_exec_set_scan_settings(pending, transcript, sig_bytes)) @@ -5231,13 +5627,28 @@ 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) + ctx = self._ctx + log.info("stream: waiting for a slot (%d of %d in use)", + ctx.get("_streams_in_flight", 0), self._stream_capacity()) async with sem: - log.info("stream: slot acquired (free=%s)", sem._value) + # Counted here rather than read back out of the semaphore's private + # `_value`: `set_capacity` needs to know how many slots are held in + # order to resize without letting the pool overshoot, and a number + # this code maintains itself is one that survives the semaphore + # object being replaced underneath it. + ctx["_streams_in_flight"] = ctx.get("_streams_in_flight", 0) + 1 + log.info("stream: slot acquired (%d of %d in use)", + ctx["_streams_in_flight"], self._stream_capacity()) try: await self._stream_video_inner(msg) finally: - log.info("stream: slot released (free=%s)", sem._value + 1) + ctx["_streams_in_flight"] = max( + 0, ctx.get("_streams_in_flight", 1) - 1) + log.info("stream: slot released (%d of %d in use)", + ctx["_streams_in_flight"], self._stream_capacity()) + + def _stream_capacity(self) -> int: + return self._ctx.get("max_concurrent_streams") or MAX_CONCURRENT_TRANSCODES async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() @@ -5492,6 +5903,12 @@ class WebRTCPeerSession: here: cancelling the task runs the exit of its `async with sem`. """ self._stop_stream() + # Before the tasks are cancelled: a lease is not held by a task, so + # nothing else would give it back, and this hook is the one place every + # way of walking away arrives at (see the connectionstatechange handler, + # which calls it for a closed tab, a quit browser and a dead network + # alike). + self._release_transfers() for task in list(self._tasks): task.cancel() if self._tasks: @@ -5499,6 +5916,7 @@ class WebRTCPeerSession: async def close(self) -> None: self._audit("disconnect") + self._release_transfers() if self._user_id: self._unregister_peer() await self.shutdown_tasks() @@ -5581,6 +5999,8 @@ class WebRTCTransport: denylist: Any | None = None, stun_servers: list[str] | None = None, max_concurrent_streams: int | None = None, + max_concurrent_downloads: int | None = None, + max_concurrent_uploads: int | None = None, transcode_incompatible_video: bool = True, ): self._ctx: dict[str, Any] = { @@ -5593,6 +6013,10 @@ class WebRTCTransport: # None means "the operator said nothing" — the default applies. It # is read once, when the first stream builds the semaphore. "max_concurrent_streams": max_concurrent_streams, + # Read once, when the first transfer builds the pools. None means + # the operator said nothing and transfers.py's defaults apply. + "max_concurrent_downloads": max_concurrent_downloads, + "max_concurrent_uploads": max_concurrent_uploads, # Operator opt-out (node.toml) for the HEVC-etc. transcode # fallback in _stream_video_inner — real CPU cost, unlike copy. "transcode_incompatible_video": transcode_incompatible_video, @@ -5605,6 +6029,96 @@ class WebRTCTransport: self._stun = stun_servers or list(DEFAULT_STUN_SERVERS) self._sessions: dict[str, WebRTCPeerSession] = {} + def set_capacity(self, *, max_concurrent_streams: int | None = None, + max_concurrent_downloads: int | None = None, + max_concurrent_uploads: int | None = None) -> dict: + """Resize a live pool without restarting the daemon. + + `ops.set_node_settings` used to do this by assigning + `webrtc._stream_sem`, an attribute that has never existed — the pool is + `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always + False. So the hot-swap was a no-op and **`max_concurrent_streams` has + never taken effect from the Node page without a restart**, contrary to + draft-v6 §2.11. This is the one implementation, on the object that owns + the state, so the next two caps do not each grow their own copy of the + mistake. + + What resizing means, stated because it is a decision and not a + detail: **the new cap governs new streams; the ones already running are + never interrupted.** A slot is held for the length of a film, so + lowering the cap below what is in flight cannot take a viewer's film + away — it stops the next one starting. The replacement pool is therefore + created with the permits that remain (`new - in_flight`, floored at + zero), not with a full set, or lowering the cap would briefly allow more + viewers than either the old value or the new one. + """ + changed: dict = {} + if max_concurrent_streams is not None: + n = int(max_concurrent_streams) + if n < 1: + raise ValueError("max_concurrent_streams must be positive") + before = self._ctx.get("max_concurrent_streams") + self._ctx["max_concurrent_streams"] = n + if self._ctx.get("_transcode_sem") is not None: + in_flight = self._ctx.get("_streams_in_flight", 0) + self._ctx["_transcode_sem"] = asyncio.Semaphore( + max(0, n - in_flight)) + log.info("stream: capacity %s -> %d (%d in flight, %d free now)", + before, n, in_flight, max(0, n - in_flight)) + else: + # Nothing has streamed yet; the pool is built from this value on + # first use, so there is nothing to resize. + log.info("stream: capacity %s -> %d (no pool built yet)", + before, n) + changed["max_concurrent_streams"] = n + + pools = {} + if max_concurrent_downloads is not None: + pools[transfers_mod.DOWNLOAD] = int(max_concurrent_downloads) + if max_concurrent_uploads is not None: + pools[transfers_mod.UPLOAD] = int(max_concurrent_uploads) + for key, value in pools.items(): + if value < 1: + raise ValueError(f"max_concurrent_{key}s must be positive") + if pools: + # Kept on the context whether or not a pool exists yet: the pools + # are built on the first transfer, and would otherwise come up with + # the defaults after an operator had already changed them. + for key, value in pools.items(): + self._ctx[f"max_concurrent_{key}s"] = value + changed[f"max_concurrent_{key}s"] = value + slots = self._ctx.get("_transfer_slots") + if slots is not None: + granted = slots.set_caps(node=pools) + log.info("transfer: capacity now %s (%d started at once)", + slots.summary(), len(granted)) + # Raising a cap can start queued transfers immediately, and the + # peers waiting on them have to be told: a grant nobody hears + # about is the "stuck at waiting" report this design exists to + # prevent. + for lease in granted: + self._notify_granted(lease) + return changed + + def _notify_granted(self, lease) -> None: + """Tell the connection that owns `lease` it may start. + + On the transport rather than the session because a cap change has no + session behind it — it arrives from the loopback API. + """ + groups = self._ctx.get("groups") + registries = ([g.get("_peers", {}) for g in groups.values()] + if groups else [self._ctx.get("_peers", {})]) + for reg in registries: + session = reg.get(lease.session_key) + if session is not None: + try: + session._send( + session._transfer_state_msg(lease, "granted")) + except Exception: + pass + return + async def handle_offer( self, offer_sdp: str, peer_id: str, ) -> tuple[str, list[dict]]: |