"""Transfer slots and leases: what a member opens before a download or an upload, what the node tells everyone when one frees, and the sweep that reclaims leases a vanished peer left behind.""" import asyncio import logging from typing import TYPE_CHECKING from meshbay_common import MNP_VERSION from meshbay_common.protocol import MNP from meshbay_node import transfers as transfers_mod from meshbay_node.transfers import TransferSlots from meshbay_node.transport.webrtc import sctp_stats from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED log = logging.getLogger("meshbay_node.transport.webrtc_server") if TYPE_CHECKING: # annotations only: the facade imports this module from meshbay_node.transport.webrtc_server import WebRTCPeerSession # 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 MIB = 1024 * 1024 class TransferMixin: 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 _lease_of(self, tr) -> str: """What the `tr` on a request actually is, from the node's own record. `tr` is drawn by the client (§5.5) and arrives on every chunk request and every upload chunk. It was read as a boolean: *present* meant "this is a leased transfer", and nothing asked whether the node had ever granted such a lease — so any non-empty string skipped the leaseless ceiling and every cap the operator set. `touch()` has always answered exactly this question (`False` if it is not granted) and its answer was discarded. Three outcomes, because they deserve different treatment: - **`granted`** — a live lease of *this session*, and the transfer is under the caps it was granted against. The session is checked as well as the id: a lease belongs to a connection, and touching somebody else's would refresh their idle timer. A connection of the same device to the same group counts as this session here (`_shares`), because one connection carries 5 to 6 MB/s whatever the line can do and a client spreads a large download over several. - **`queued`** — the node has this lease and has not granted it. The client is jumping its own queue; refused, and no shipped client does it (the transfer store awaits the grant before it reads a byte). - **`none`** — the node has no such lease. Deliberately *not* a refusal: it is what a reconnect looks like from here, where the session's leases died with the old connection and the client is re-opening them, and it is what a client that never asked looks like. Both are then bounded by the leaseless ceiling instead — which is the residual §5.5 already states: a client that lies gets that bound's worth of files at a time, not the whole library. """ slots = self._ctx.get("_transfer_slots") if slots is None: return LEASE_NONE lease = slots.leases.get(tr) if lease is None or not self._shares(lease): return LEASE_NONE if lease.state != "granted": return LEASE_QUEUED slots.touch(tr) return LEASE_GRANTED def _shares(self, lease) -> bool: """Whether this connection may move bytes under `lease`. Its own lease, or one opened by another connection of the same device, for the same member of the same group. The slot was charged to that member when the lease was granted, so a second connection reading under it costs nobody anything they had not already agreed to. It cannot close the lease and is not told its state: those stay the opener's. """ if lease.session_key == self._registry_key: return True return bool(lease.device and lease.device == self._pinned_pk and lease.member == (self._group_id or "", self._user_id or "")) def _note_unleased(self, tr: str) -> None: """Say once that this connection transferred outside its lease. Once per session, not per chunk: the interesting fact is that it happened, and a per-chunk line would bury it under itself. The bound is the leaseless ceiling either way; this is what makes the residual visible to the operator rather than merely stated in a document. """ if self._unleased_noted: return self._unleased_noted = True log.info("transfer: %s sent chunk requests under an unknown lease %s " "— bounded by the leaseless ceiling", (self._user_id or "?")[:8], str(tr)[:8]) self._audit("transfer_unleased", str(tr)[:16]) def _account(self, tr: str, nbytes: int) -> None: slots = self._ctx.get("_transfer_slots") if slots is not None: slots.account(tr, nbytes, session_key=self._registry_key) def _transfer_end_line(self, lease, reason: str) -> str: """What a transfer moved, how fast, and what its association looked like. The node wrote nothing about throughput, so a slow download left no trace in the journal and took a test bench to explain. The SCTP figures are the association's totals since it opened, not this transfer's share: a connection carries several transfers and they share one window. """ line = (f"transfer: end {lease.kind} {lease.tr[:8]} {reason} " f"{lease.moved / MIB:.1f} MB") if lease.moved and lease.elapsed > 0: line += (f" in {lease.elapsed:.1f} s " f"({lease.moved / MIB / lease.elapsed:.2f} MB/s)") if len(lease.lanes) > 1: line += f" over {len(lease.lanes)} connections" owner = (self if lease.session_key == self._registry_key else next(iter(self._sessions_everywhere(lease.session_key)), None)) return line + sctp_stats.describe(getattr(owner, "_pc", None)) 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), # `member_cap`, not the node-wide default: this group's own limit is # what `_has_room` enforces and what the handshake ack announces, so # reading the default here would have the widget contradicting both # — "1 of 2" in a group where the operator signed 5, or two slots # offered in a group limited to one. "cap": slots.member_cap(lease.kind, lease.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, device=self._pinned_pk) 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)) log.info("%s", self._transfer_end_line(lease, 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) for lease in gone: log.info("%s", self._transfer_end_line( lease, transfers_mod.REASON_SESSION_GONE)) 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) if lease.state != "queued": log.info("%s", self._transfer_end_line(lease, 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)