2 files changed, 233 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()
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)