From aca58f66df5c21be71a7159620d0b50bb969199f Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Sun, 11 Oct 2026 16:52:48 +0200 Subject: feat: spread a large download over eight connections One WebRTC connection carries 5 to 6 MB/s whatever the line can do. A leased download of 64 MB or more now opens seven more to its node, and a lease serves the other connections of its device in its group. Measured through the whole stack: 5 MB/s before, 24 after. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-node/src/meshbay_node/transfers.py | 18 +++-- .../transport/webrtc/transfer_handlers.py | 28 ++++++-- .../meshbay-node/tests/test_lease_enforcement.py | 82 +++++++++++++++++++++- .../meshbay-node/tests/test_transfer_slots_wire.py | 1 + 4 files changed, 120 insertions(+), 9 deletions(-) (limited to 'packages/meshbay-node') diff --git a/packages/meshbay-node/src/meshbay_node/transfers.py b/packages/meshbay-node/src/meshbay_node/transfers.py index 477ec92..e827362 100644 --- a/packages/meshbay-node/src/meshbay_node/transfers.py +++ b/packages/meshbay-node/src/meshbay_node/transfers.py @@ -14,7 +14,10 @@ properties are load-bearing, and each one is a decision: reconnect with the same `tr` is idempotent, so a reconnect cannot charge a member twice for one transfer. - **A lease is scoped to the connection**, never to the account. It dies with - the session, which is what makes the primary reclaim deterministic. + the session, which is what makes the primary reclaim deterministic. Other + connections of the same device to the same group may read under it (a + download spread over several connections is one transfer and one slot), + and none of them can open, close or outlive it. - **A lease covers a job, not a file.** A directory zip is dozens of files and one lease. - **Nothing is persisted.** A restart drops every session anyway; a lease that @@ -108,6 +111,10 @@ class Lease: moved: int = 0 first_used: float | None = None last_moved: float = 0.0 + # The device key of the connection that opened it, and every connection + # that moved bytes under it (see `TransferMixin._lease_of`). + device: str = "" + lanes: set[str] = field(default_factory=set) @property def member(self) -> tuple[str, str]: @@ -186,6 +193,7 @@ class TransferSlots: def open(self, *, tr: str, kind: str, session_key: str, user_id: str, group_id: str, bytes: int = 0, chunks: int = 0, + device: str = "", now: float | None = None) -> tuple[Lease | None, str]: """Ask for a slot. Returns (lease, error_code); one of them is falsy. @@ -214,7 +222,7 @@ class TransferSlots: lease = Lease(tr=tr, kind=kind, session_key=session_key, user_id=user_id, group_id=group_id, bytes=int(bytes or 0), chunks=int(chunks or 0), - created_at=now, last_seen=now) + device=device, created_at=now, last_seen=now) self.leases[tr] = lease if self._has_room(kind, member): self._grant(lease, now) @@ -240,12 +248,14 @@ class TransferSlots: lease.first_used = lease.last_seen return True - def account(self, tr: str, nbytes: int, now: float | None = None) -> None: - """`nbytes` were served or received under this lease.""" + def account(self, tr: str, nbytes: int, now: float | None = None, + session_key: str = "") -> None: + """`nbytes` were served or received under this lease, on that connection.""" lease = self.leases.get(tr) if lease is None: return lease.moved += nbytes + lease.lanes.add(session_key) lease.last_moved = time.monotonic() if now is None else now def close(self, tr: str, reason: str = REASON_DONE, 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 index 78c74d6..11ad9e6 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/transfer_handlers.py @@ -73,7 +73,10 @@ class TransferMixin: - **`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. + else's would refresh their idle timer. A connection of the same + device to the same group counts as this session here (`_shares`), + because one connection carries 5 to 6 MB/s whatever the line can + do and a client spreads a large download over several. - **`queued`** — the node has this lease and has not granted it. The client is jumping its own queue; refused, and no shipped client does it (the transfer store awaits the grant before it reads a byte). @@ -89,13 +92,28 @@ class TransferMixin: if slots is None: return LEASE_NONE lease = slots.leases.get(tr) - if lease is None or lease.session_key != self._registry_key: + if lease is None or not self._shares(lease): return LEASE_NONE if lease.state != "granted": return LEASE_QUEUED slots.touch(tr) return LEASE_GRANTED + def _shares(self, lease) -> bool: + """Whether this connection may move bytes under `lease`. + + Its own lease, or one opened by another connection of the same device, + for the same member of the same group. The slot was charged to that + member when the lease was granted, so a second connection reading under + it costs nobody anything they had not already agreed to. It cannot + close the lease and is not told its state: those stay the opener's. + """ + if lease.session_key == self._registry_key: + return True + return bool(lease.device + and lease.device == self._pinned_pk + and lease.member == (self._group_id or "", self._user_id or "")) + def _note_unleased(self, tr: str) -> None: """Say once that this connection transferred outside its lease. @@ -115,7 +133,7 @@ class TransferMixin: def _account(self, tr: str, nbytes: int) -> None: slots = self._ctx.get("_transfer_slots") if slots is not None: - slots.account(tr, nbytes) + slots.account(tr, nbytes, session_key=self._registry_key) def _transfer_end_line(self, lease, reason: str) -> str: """What a transfer moved, how fast, and what its association looked like. @@ -131,6 +149,8 @@ class TransferMixin: if lease.moved and lease.elapsed > 0: line += (f" in {lease.elapsed:.1f} s " f"({lease.moved / MIB / lease.elapsed:.2f} MB/s)") + if len(lease.lanes) > 1: + line += f" over {len(lease.lanes)} connections" owner = (self if lease.session_key == self._registry_key else next(iter(self._sessions_everywhere(lease.session_key)), None)) return line + sctp_stats.describe(getattr(owner, "_pc", None)) @@ -212,7 +232,7 @@ class TransferMixin: 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) + bytes=nbytes, chunks=chunks, device=self._pinned_pk) if err: self._send({"type": "error", "detail": err, "code": err, "tr": tr}) return diff --git a/packages/meshbay-node/tests/test_lease_enforcement.py b/packages/meshbay-node/tests/test_lease_enforcement.py index 7b636ac..2104365 100644 --- a/packages/meshbay-node/tests/test_lease_enforcement.py +++ b/packages/meshbay-node/tests/test_lease_enforcement.py @@ -59,9 +59,10 @@ class _Channel: class _Session(WebRTCPeerSession): """A session with the channel replaced by a list, and a real everything else.""" - def __init__(self, ctx, *, key="s1", user="alice"): + def __init__(self, ctx, *, key="s1", user="alice", device=None): self._ctx = ctx self._registry_key = key + self._pinned_pk = device or f"pk-{user}" self._user_id = user self._username = user self._group_id = GROUP @@ -306,3 +307,82 @@ async def test_a_closed_transfer_is_logged_once(group): assert [x.split()[3:5] for x in lines] == [ ["t1", "cancelled"], ["t2", "session_gone"]], lines + + +# ── one download over several connections ─────────────────────────────────── + +async def test_a_second_connection_of_the_device_reads_under_the_lease(group): + """One connection carries 5 to 6 MB/s whatever the line can do, so a client + spreads a large download over several. It is one transfer and one slot.""" + owner = _Session(group, key="s1", user="alice") + lane = _Session(group, key="s2", user="alice") + group["_peers"].update({"s1": owner, "s2": lane}) + owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + + await _request(lane, _ids(group)[0], tr="t1") + assert lane.chunks(), lane.errors() + assert lane._leaseless.in_flight() == 0 + slots = owner._slots() + assert slots.in_use(DOWNLOAD) == 1 + assert slots.leases["t1"].used is True + + await _request(owner, _ids(group)[0], tr="t1") + line = owner._transfer_end_line(slots.leases["t1"], "done") + assert " 2.0 MB" in line and line.endswith("over 2 connections"), line + + +async def test_another_device_of_the_account_does_not(group): + owner = _Session(group, key="s1", user="alice", device="pk-laptop") + other = _Session(group, key="s2", user="alice", device="pk-phone") + group["_peers"].update({"s1": owner, "s2": other}) + owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + + await _request(other, _ids(group)[0], tr="t1") + assert other._leaseless.in_flight() == 1 + + +async def test_the_same_device_key_in_another_group_does_not(group): + owner = _Session(group, key="s1", user="alice") + other = _Session(group, key="s2", user="alice") + other._group_id = "h" * 32 + owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + assert other._lease_of("t1") == "none" + + +async def test_a_lease_without_a_device_is_its_connections_alone(group): + owner = _Session(group, key="s1", user="alice") + lane = _Session(group, key="s2", user="alice") + owner._pinned_pk = lane._pinned_pk = "" + owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + assert lane._lease_of("t1") == "none" + + +async def test_a_queued_lease_is_refused_on_every_connection(group): + owner = _Session(group, key="s1", user="alice") + lane = _Session(group, key="s2", user="alice") + group["_peers"].update({"s1": owner, "s2": lane}) + owner._slots().set_caps(per_member={DOWNLOAD: 1}) + owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + owner._do_transfer_open({"tr": "t2", "kind": DOWNLOAD, "bytes": 2_000_000}) + assert owner.last()["state"] == "queued" + + await _request(lane, _ids(group)[0], tr="t2") + assert not lane.chunks() + assert lane.errors()[-1]["code"] == "lease_not_granted" + + +async def test_only_the_connection_that_opened_a_lease_closes_it(group): + owner = _Session(group, key="s1", user="alice") + lane = _Session(group, key="s2", user="alice") + group["_peers"].update({"s1": owner, "s2": lane}) + owner._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + + lane._do_transfer_close({"tr": "t1"}) + assert lane.errors()[-1]["code"] == "not_your_transfer" + lane._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 2_000_000}) + assert lane.errors()[-1]["code"] == "not_your_transfer" + lane._release_transfers() + assert lane._lease_of("t1") == "granted" + + owner._release_transfers() + assert lane._lease_of("t1") == "none" diff --git a/packages/meshbay-node/tests/test_transfer_slots_wire.py b/packages/meshbay-node/tests/test_transfer_slots_wire.py index 6e9ef94..58e743f 100644 --- a/packages/meshbay-node/tests/test_transfer_slots_wire.py +++ b/packages/meshbay-node/tests/test_transfer_slots_wire.py @@ -36,6 +36,7 @@ class _Session(WebRTCPeerSession): def __init__(self, ctx, *, key, user, group="g1"): self._ctx = ctx self._registry_key = key + self._pinned_pk = f"pk-{user}" self._user_id = user self._group_id = group self.sent: list[dict] = [] -- cgit v1.2.3