1 files changed, 61 insertions, 3 deletions
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()