Merge branch 'fix/large-download-paths'

Concurrent-transfer limits, with the queue, the pause and the flag day.

A node now caps how many transfers it runs at once (8 downloads, 8 uploads,
node-wide) and how many one member may run in one group (2 by default,
operator-signed). Beyond that the node answers "queued" and the client waits its
turn, visibly, in the transfers panel — and a slot that frees starts whatever is
next, skipping past a member who is at their own cap rather than letting them
stall everyone behind them.

Browsing is never subject to a slot: not the poster grid, not the covers, not
opening a photo to look at it. That is structural — a transfer is what the
transfers widget shows — and the exemption is bounded rather than open, at two
files in flight per session, because an exemption with no bound is a leaseless
branch under another name.

Transfers can be cancelled, and now paused and resumed. A paused one holds
nothing: its slot goes back at once and resuming rejoins the queue at the tail.
Uploads survive the connection that started them and resume where the node
stopped, asked for inside the seal rather than on a clear message. What they
leave behind when they are abandoned is reaped, which closes a disk leak that
predates this work.

MNP 3.0 makes the lease compulsory and refuses 2.x at the handshake, with the
desktop client checking `client.minimum` before connecting so an un-updated one
says "update" instead of failing every connection in a protocol vocabulary.

Fourteen defects were found on the way, eight of them by a person clicking
Download and pasting a console — none of which 2075 tests could reach. Section
12 of ~/next/improve-downloads.md is that report, including the three this work
introduced itself and the one that turned out to be caused by an instruction to
hard-reload after each deployment.

Node suite 1209 passed, hub suite 866 passed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HCGdheDLxGReuKHga3BtST

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]]: