diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py | 278 | ||||
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 262 |
2 files changed, 281 insertions, 259 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py new file mode 100644 index 0000000..729c0a0 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py @@ -0,0 +1,278 @@ +"""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.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 + + +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. + - **`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 lease.session_key != self._registry_key: + return LEASE_NONE + if lease.state != "granted": + return LEASE_QUEUED + slots.touch(tr) + return LEASE_GRANTED + + 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 _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) + 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) 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 cb95822..a617d29 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -126,7 +126,6 @@ from meshbay_node.roots import ( off_disk, ) from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK -from meshbay_node.transfers import TransferSlots from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin @@ -141,7 +140,8 @@ from meshbay_node.transport.webrtc.channel import ( ) from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.files import FilesMixin -from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED, MAX_MSG +from meshbay_node.transport.webrtc.limits import LEASE_NONE, LEASE_QUEUED, MAX_MSG +from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin log = logging.getLogger(__name__) @@ -193,12 +193,6 @@ JOIN_FAILURE_WINDOW = 600 # seconds # nobody could read, or files scattered wherever someone happened to be looking. -# 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 - # Opt-in, off by default: a per-session heartbeat log (message count, time # since the last message, ICE state) and ICE-state-change logging, on top of @@ -215,7 +209,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( - BlobsMixin, ChatMixin, FilesMixin, + BlobsMixin, ChatMixin, FilesMixin, TransferMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -2691,256 +2685,6 @@ class WebRTCPeerSession( return MAX_UPLOAD_BYTES return max(1, int(float(gb) * GB_BYTES)) - 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. - - **`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 lease.session_key != self._registry_key: - return LEASE_NONE - if lease.state != "granted": - return LEASE_QUEUED - slots.touch(tr) - return LEASE_GRANTED - - 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 _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) - 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. |