path: root/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py

"""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)