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 +++++++++++++++++++++- 1 file changed, 61 insertions(+), 3 deletions(-) (limited to 'packages/meshbay-hub/src') 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() -- cgit v1.2.3