From 2d657f40ebd697e4332c95d7a57bbb292ff46012 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Thu, 24 Sep 2026 00:10:41 +0200 Subject: fix(hub): offer ceilings that bound a member without failing an ordinary one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A phone on 4G found one of five groups missing from Search on half its visits, and all but one on a fifth of them. Every offer that reached the node was answered in about a second; the hub refused the others with 429 — 93 of them in half an hour on meshbay.org, all from that phone, all to nodes that were up. Two ceilings did it. Three pending offers per account was exactly what one Search page dialled at once, so the first tile or reconnection beside the sweep was refused. Thirty offers a minute per address and per node was spent by four or five reloads of a page whose groups share one node. Sized now against an account with twenty groups on three devices: - 32 offers pending per account, across devices and nodes. - A budget per account and per node, a burst of 120 refilled at two a second. This is what bounds a member's cost to one machine (H6), and it leaves their other nodes alone. Counted by account, because a mobile carrier puts many subscribers behind one IPv4 address. - 600 a minute per address and per node, as a coarse guard in front of authentication only. Both refusals carry Retry-After, which the browser now honours. The node's own ceiling on peer sessions is unchanged; its comment no longer quotes the old per-account number. Co-Authored-By: Claude Opus 5.5 --- .../meshbay-hub/src/meshbay_hub/api/signaling.py | 64 +++++++- .../meshbay-hub/tests/test_signaling_limits.py | 172 +++++++++++++++++++++ 2 files changed, 233 insertions(+), 3 deletions(-) create mode 100644 packages/meshbay-hub/tests/test_signaling_limits.py (limited to 'packages/meshbay-hub') diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py index 8a10822..60d5e20 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py @@ -15,6 +15,8 @@ Flow: import asyncio import json import logging +import math +import time import uuid from fastapi import APIRouter, Depends, HTTPException, Request @@ -51,13 +53,60 @@ class WebRTCOfferResponse(BaseModel): MAX_SDP_BYTES = 16 * 1024 # an SDP offer is ~2 KB -MAX_PENDING_PER_USER = 3 # concurrent in-flight offers per account + +# Offers of one account waiting for their node's answer, across all its devices +# and all nodes. An offer is pending for the node's round trip — well under a +# second — or up to the 15 s answer timeout when the node is connected and +# silent. Sized for an account with twenty groups on three devices: Search +# negotiates six at once per page (search-page.js), so three pages are eighteen, +# plus a phone's pooled connections all reconnecting as it wakes. It was 3, which +# is exactly what one Search page dialled at once, so the first tile or the first +# reconnection beside a sweep was refused and its group reported unreachable. +MAX_PENDING_PER_USER = 32 + +# Offers one account may send one node: a burst of OFFER_BURST, refilled at +# OFFER_REFILL_PER_S. This is what bounds a member's cost to a node (H6) — each +# offer makes it allocate a peer connection — and it is counted per account +# rather than per address, because a 4G carrier puts many subscribers behind one +# IPv4 address and hands each a whole IPv6 /64. The burst covers twenty groups +# hosted on one node, reloaded several times over on more than one device; the +# refill is two a second, sustained. The node bounds its own total separately +# (MAX_PEER_SESSIONS in webrtc_server.py). +OFFER_BURST = 120 +OFFER_REFILL_PER_S = 2.0 _pending_per_user: dict[str, int] = {} +# (user id, node id) -> (tokens left, when they were counted) +_offer_buckets: dict[tuple[str, str], tuple[float, float]] = {} +_OFFER_BUCKETS_PRUNE_AT = 4096 + + +def _take_offer(user_id: str, node_id: str, now: float) -> float | None: + """Spend one of this account's offers to this node. + + None when it may go ahead, otherwise how many seconds until it may. + """ + if len(_offer_buckets) > _OFFER_BUCKETS_PRUNE_AT: + # A bucket that would have refilled completely carries no information. + full_after = OFFER_BURST / OFFER_REFILL_PER_S + for key, (_, at) in list(_offer_buckets.items()): + if now - at >= full_after: + del _offer_buckets[key] + key = (user_id, node_id) + tokens, at = _offer_buckets.get(key, (float(OFFER_BURST), now)) + tokens = min(float(OFFER_BURST), tokens + (now - at) * OFFER_REFILL_PER_S) + if tokens < 1.0: + _offer_buckets[key] = (tokens, now) + return (1.0 - tokens) / OFFER_REFILL_PER_S + _offer_buckets[key] = (tokens - 1.0, now) + return None @router.post("/{node_id}/webrtc/offer", response_model=WebRTCOfferResponse) -@limiter.limit("30/minute") +# Per address and per node, and only a coarse guard in front of authentication: +# the account's budget above is the limit that means something. 600 because an +# IPv4 address on a mobile network is shared by many subscribers. +@limiter.limit("600/minute") async def webrtc_offer( node_id: str, body: WebRTCOfferRequest, @@ -147,8 +196,17 @@ async def webrtc_offer( state = "revoked" if "revoked" in statuses else next(iter(statuses), "suspended") raise HTTPException(status_code=403, detail=f"Group is {state}") + # Both refusals say when to come back, and transport.js does: a 429 here is + # the hub being busy, never the node being down. + # The pending check comes first so that an offer refused by it does not + # spend the node's budget: retries of a busy moment would otherwise drain it. if _pending_per_user.get(current_user.id, 0) >= MAX_PENDING_PER_USER: - raise HTTPException(status_code=429, detail="Too many pending connections") + raise HTTPException(status_code=429, detail="Too many pending connections", + headers={"Retry-After": "1"}) + wait = _take_offer(current_user.id, node_id, time.monotonic()) + if wait is not None: + raise HTTPException(status_code=429, detail="Too many connections to this node", + headers={"Retry-After": str(max(1, math.ceil(wait)))}) peer_id = str(uuid.uuid4()) answer_future: asyncio.Future = asyncio.get_event_loop().create_future() diff --git a/packages/meshbay-hub/tests/test_signaling_limits.py b/packages/meshbay-hub/tests/test_signaling_limits.py new file mode 100644 index 0000000..2326011 --- /dev/null +++ b/packages/meshbay-hub/tests/test_signaling_limits.py @@ -0,0 +1,172 @@ +""" +The hub's ceilings on offers must bound a member without failing an ordinary one. + +They did fail one. `MAX_PENDING_PER_USER` was 3 and Search dialled three groups +at once, so a single tile or reconnection beside the sweep was refused with 429, +and the page reported the group's node as unreachable while that node was +answering every other offer in under a second. The per-address limit, 30 a +minute per node, was spent by four or five reloads of a page on a phone whose +groups share one node. Measured on meshbay.org on 2026-09-23: 93 offers refused +in half an hour, all from one phone, all to nodes that were up. + +What these tests hold is the shape the limits must admit — twenty groups on one +node, dialled together — and that each refusal says when to come back, because +the browser now does (transport.js, `postOffer`). +""" + +import asyncio +import json +import time + +import pytest +from test_availability_between_members import ( + _add_member, + _announce_node, + _make_group, + _make_user, +) + + +class _SlowNode: + """A node that answers each offer once `release` is set.""" + + def __init__(self, node_id: str): + self.node_id = node_id + self.release = asyncio.Event() + self.offers = 0 + + async def send_text(self, text): + from meshbay_hub.api.signaling import handle_webrtc_answer + + msg = json.loads(text) + self.offers += 1 + + async def answer(): + await self.release.wait() + handle_webrtc_answer({"peer_id": msg["peer_id"], "sdp": "v=0\r\nanswer", + "ice_candidates": []}, self.node_id) + + asyncio.get_running_loop().create_task(answer()) + + +async def _member_with_node(client, prefix: str): + owner = await _make_user(client, f"{prefix}_owner") + member = await _make_user(client, f"{prefix}_member") + group_id = await _make_group(client, owner, f"{prefix}-group") + await _add_member(client, owner, group_id, member) + node_id = await _announce_node(client, owner) + return member, group_id, node_id + + +def _offer(client, node_id, user): + return client.post(f"/v1/nodes/{node_id}/webrtc/offer", + json={"sdp": "v=0\r\noffer", "ice_candidates": []}, + headers={"Authorization": f"Bearer {user['token']}"}) + + +@pytest.mark.asyncio +async def test_twenty_groups_dialled_together_are_all_brokered(client): + """An account with twenty groups on one node, all dialled at once — more + than one Search page ever does — is refused nothing.""" + from meshbay_hub.api import revocation as rev + + member, group_id, node_id = await _member_with_node(client, "sig_twenty") + node = _SlowNode(node_id) + rev._connected_nodes[node_id] = node + rev._node_groups[node_id] = [group_id] + try: + pending = [asyncio.ensure_future(_offer(client, node_id, member)) for _ in range(20)] + while node.offers < 20: + await asyncio.sleep(0.01) + node.release.set() + statuses = [r.status_code for r in await asyncio.gather(*pending)] + assert statuses == [200] * 20, statuses + finally: + rev._connected_nodes.pop(node_id, None) + rev._node_groups.pop(node_id, None) + + +@pytest.mark.asyncio +async def test_the_pending_ceiling_refuses_with_a_time_to_come_back(client): + from meshbay_hub.api import revocation as rev + from meshbay_hub.api.signaling import MAX_PENDING_PER_USER + + assert MAX_PENDING_PER_USER >= 32, "sized for three devices of a twenty-group account" + member, group_id, node_id = await _member_with_node(client, "sig_pending") + node = _SlowNode(node_id) + rev._connected_nodes[node_id] = node + rev._node_groups[node_id] = [group_id] + try: + held = [asyncio.ensure_future(_offer(client, node_id, member)) + for _ in range(MAX_PENDING_PER_USER)] + while node.offers < MAX_PENDING_PER_USER: + await asyncio.sleep(0.01) + refused = await _offer(client, node_id, member) + assert refused.status_code == 429, refused.text + assert refused.headers.get("Retry-After") == "1" + node.release.set() + assert all(r.status_code == 200 for r in await asyncio.gather(*held)) + # Once they are answered the account has its places back. + node.release = asyncio.Event() + node.release.set() + again = await _offer(client, node_id, member) + assert again.status_code == 200, again.text + finally: + rev._connected_nodes.pop(node_id, None) + rev._node_groups.pop(node_id, None) + + +@pytest.mark.asyncio +async def test_an_exhausted_node_budget_says_when_and_spares_other_nodes(client): + """The per-node budget bounds what one member costs one node, and only that + node: an account that has spent it on one machine still reaches the others.""" + from meshbay_hub.api import revocation as rev + from meshbay_hub.api import signaling + + member, group_id, node_id = await _member_with_node(client, "sig_budget") + # A second machine hosting the same group, as a group with two nodes has. + _, _, other_node = await _member_with_node(client, "sig_budget2") + node, other = _SlowNode(node_id), _SlowNode(other_node) + node.release.set() + other.release.set() + rev._connected_nodes[node_id] = node + rev._node_groups[node_id] = [group_id] + rev._connected_nodes[other_node] = other + rev._node_groups[other_node] = [group_id] + try: + signaling._offer_buckets[(member["user_id"], node_id)] = (0.0, time.monotonic()) + refused = await _offer(client, node_id, member) + assert refused.status_code == 429, refused.text + assert int(refused.headers["Retry-After"]) >= 1 + assert node.offers == 0, "a refused offer still reached the node" + + ok = await _offer(client, other_node, member) + assert ok.status_code == 200, ok.text + finally: + signaling._offer_buckets.pop((member["user_id"], node_id), None) + for n in (node_id, other_node): + rev._connected_nodes.pop(n, None) + rev._node_groups.pop(n, None) + + +def test_the_node_budget_admits_a_burst_and_refills(): + """The arithmetic, without a clock: a twenty-group page reloaded six times + in a row fits, the offer after the burst waits half a second, and a minute + of quiet gives the whole burst back.""" + from meshbay_hub.api import signaling + + key = ("budget-user", "budget-node") + signaling._offer_buckets.pop(key, None) + try: + t0 = 1000.0 + assert signaling.OFFER_BURST >= 120 + for _ in range(signaling.OFFER_BURST): + assert signaling._take_offer(*key, t0) is None + wait = signaling._take_offer(*key, t0) + assert wait == pytest.approx(1 / signaling.OFFER_REFILL_PER_S) + assert signaling._take_offer(*key, t0 + wait) is None + later = t0 + signaling.OFFER_BURST / signaling.OFFER_REFILL_PER_S + wait + for _ in range(signaling.OFFER_BURST): + assert signaling._take_offer(*key, later) is None + finally: + signaling._offer_buckets.pop(key, None) -- cgit v1.2.3