Pairing limits: no shared bucket for portal calls, a cap per side

Every call through the portal arrives off the tunnel, so the per-address limit
skips those (the hub limits them by the real address), and the cap of 50 counts
and evicts portal and local requests apart, so a portal flood never pushes out
a device waiting on this network.

backend/fluksio/api/routes/devices.py

@@ -97,12 +97,16 @@ CODE_LENGTH = 6
 #: The check a device shows beside its code, to compare with the pending list.
 CHECK_LENGTH = 2
 
-#: Never more waiting than this: codes are free to ask for and cost memory.
-#: The oldest makes room for a new one, so a flood cannot lock a device out.
+#: Never more waiting than this from each way in, this network or the portal:
+#: codes are free to ask for and cost memory. The oldest of the same way in
+#: makes room for a new one, so a flood cannot lock a device out, and one
+#: through the portal never pushes out a device waiting here.
 MAX_PENDING = 50
 
-#: Codes one address may ask for per window. A device asks once per code,
-#: every ten minutes, and again after a failure.
+#: Codes one address may ask for per window, here. A device asks once per
+#: code, every ten minutes, and again after a failure. Through the portal the
+#: hub limits each address and the instance by the address it sees, since
+#: here every such call arrives off the tunnel and would share one bucket.
 PAIR_PER_ADDRESS = 5
 PAIR_WINDOW_S = 60.0
 
@@ -197,10 +201,15 @@ class _PendingStore:
         self._redis.zremrangebyscore(_PENDING_INDEX, "-inf", now)
         return list(cast("list[str]", self._redis.zrange(_PENDING_INDEX, 0, -1)))
 
-    def make_room(self, room: int) -> None:
-        """Drop the oldest codes until fewer than ``room`` are left."""
-        codes = self._codes()
-        for code in codes[: max(0, len(codes) - room + 1)]:
+    def make_room(self, room: int, remote: bool) -> None:
+        """Drop the oldest codes asked for the same way, through the portal
+        or not, until fewer than ``room`` of them are left."""
+        same = [
+            code
+            for code in self._codes()
+            if (entry := self.get(code)) is not None and entry.remote == remote
+        ]
+        for code in same[: max(0, len(same) - room + 1)]:
             self.drop(code)
 
     def count(self) -> int:
@@ -486,15 +495,15 @@ def start_pairing(request: Request, body: PairHello | None = None) -> Any:
     the cap and the portal's own limits are between. At the cap the oldest
     request makes room, so a flood delays a device rather than locking it out.
     """
-    if too_many(request, "pair", PAIR_PER_ADDRESS, PAIR_WINDOW_S):
+    remote = cloud_config.via_portal(request.headers)
+    if not remote and too_many(request, "pair", PAIR_PER_ADDRESS, PAIR_WINDOW_S):
         raise HTTPException(
             status_code=429,
             detail="Too many pairing codes asked for from here. Try again shortly.",
             headers={"Retry-After": str(int(PAIR_WINDOW_S))},
         )
-    _pending.make_room(MAX_PENDING)
+    _pending.make_room(MAX_PENDING, remote)
 
-    remote = cloud_config.via_portal(request.headers)
     hello = body or PairHello()
     entry = _Pending(
         secret=secrets.token_urlsafe(16),

backend/tests/api/routes/test_devices.py

@@ -1215,7 +1215,7 @@ def test_a_pending_code_is_found_by_whichever_worker_is_polled() -> None:
     for offset, code in enumerate(("OLD111", "MID222", "NEW333")):
         later = devices_route._Pending(secret="s", expires=time.time() + 60 + offset)
         assert minted.add(code, later)
-    polled.make_room(3)
+    polled.make_room(3, remote=False)
     assert minted.get("OLD111") is None and polled.count() == 2
 
 
@@ -1694,6 +1694,10 @@ def test_one_address_asks_for_a_few_codes_a_minute(client: TestClient) -> None:
     assert refused.headers["retry-after"]
     # Another address is not held to this one's.
     assert client.post(f"{PREFIX}/pair").status_code == 200
+    # Through the portal every call comes off the tunnel, so the hub limits
+    # those by the address it sees, and here they are not one shared bucket.
+    for _ in range(2 * devices_route.PAIR_PER_ADDRESS):
+        assert client.post(f"{PREFIX}/pair", headers=_via_portal()).status_code == 200
 
 
 def test_at_the_cap_the_oldest_request_makes_room(
@@ -1704,24 +1708,27 @@ def test_at_the_cap_the_oldest_request_makes_room(
     monkeypatch.setattr(devices_route, "_pending", store)
     monkeypatch.setattr(devices_route, "PAIR_PER_ADDRESS", 1000)
 
-    started = [
+    def waiting(started: dict) -> bool:
+        polled = client.get(
+            f"{PREFIX}/pair/{started['code']}", params={"secret": started["secret"]}
+        )
+        return polled.status_code == 200
+
+    here = [
         client.post(f"{PREFIX}/pair").json()
         for _ in range(devices_route.MAX_PENDING + 1)
     ]
     assert store.count() == devices_route.MAX_PENDING
-    oldest, newest = started[0], started[-1]
-    assert (
-        client.get(
-            f"{PREFIX}/pair/{oldest['code']}", params={"secret": oldest["secret"]}
-        ).status_code
-        == 404
-    )
-    assert (
-        client.get(
-            f"{PREFIX}/pair/{newest['code']}", params={"secret": newest["secret"]}
-        ).status_code
-        == 200
-    )
+    assert not waiting(here[0]) and waiting(here[1]) and waiting(here[-1])
+
+    # A flood through the portal pushes out its own oldest, never these.
+    portal = [
+        client.post(f"{PREFIX}/pair", headers=_via_portal()).json()
+        for _ in range(devices_route.MAX_PENDING + 1)
+    ]
+    assert store.count() == 2 * devices_route.MAX_PENDING
+    assert not waiting(portal[0]) and waiting(portal[-1])
+    assert all(waiting(started) for started in here[1:])
 
 
 def test_a_report_is_read_port_by_port(