refactor(node): move transfer slots and leases out of webrtc_server

TransferMixin in transport/webrtc/transfer_handlers.py, with the sweep
interval.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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.