fix(server): give background identity refreshes their own budget

packages/sie_server/REMOTE_BACKENDS.md

@@ -200,7 +200,10 @@ received no answer, so it is `QUEUE_FULL`. Any other failure is
 
 Each upstream has a required rate cap and a circuit breaker. The limits are
 shared by that worker process's adapters, not across replicas: adding remote
-worker replicas increases the aggregate permitted traffic. Under
+worker replicas increases the aggregate permitted traffic. Background identity
+refreshes never consume the inference rate cap; they have a smaller budget of
+their own, described under
+[Admitting SIE identity fallback](#admitting-sie-identity-fallback). Under
 `remote_only`, unavailable upstreams, open breakers and reached caps return
 retryable `503` responses with `Retry-After`. In a cluster the remote worker
 answers such a request at once instead of redelivering it, and the gateway
@@ -379,6 +382,22 @@ refreshes metadata. A concurrent refresh
 refuses another bridge instead of waiting or starting a second metadata request.
 Changes to the installed upstream discard the previous observation.
 
+Background identity refreshes never consume the inference rate cap. A check
+that does not wait, such as the admission a cluster's remote lane reports and
+checks before an admitted remote attempt, refreshes the observation in the
+background before it expires. These refreshes draw on limits of their own for
+each upstream in each worker process: a tenth of
+`rate_cap.requests_per_minute`, at least one read a minute, one read at a
+time, and a circuit breaker with the upstream's `breaker` settings. Inference
+calls never use this budget. A refresh these limits refuse is not sent and
+counts as a failed read. While such checks continue, each model's observation
+is read about every 20 seconds, so the limits keep about a third as many
+models admitted as they allow reads a minute: two per process for an upstream
+capped at 60 requests a minute. Past that, admissions lapse and requests stay
+local. Reads that a check waits for, at configuration load, at hot reload and
+before a single-node bridge, count against the upstream's rate cap and breaker
+like inference calls.
+
 Because configuration load runs this comparison, a single-node server whose
 models directory holds a hybrid SIE-identity `encode` or `score` model depends
 on the upstream at startup. When the upstream cannot be reached, or reports a

packages/sie_server/src/sie_server/adapters/remote/_limits.py

@@ -23,6 +23,17 @@
 refused before anything is sent. A batch larger than the whole budget is let
 through when the budget is full, and the budget then refills before anything
 else is sent.
+
+Background refreshes of an SIE upstream's identity metadata for hybrid
+admission go through a second limiter per upstream name, never through the
+first, so they never consume the inference rate cap. It has a budget of its
+own: a tenth of ``rate_cap.requests_per_minute``, at least one read a minute,
+with one read in flight. Its circuit breaker, set by the same ``breaker``
+settings, is its own too. Inference calls never draw on it, and its refusals
+and breaker state are not reported to the worker telemetry, whose upstream
+instruments describe inference calls. An identity read that its caller waits
+for, such as the check at configuration load, goes through the first limiter
+like an inference call.
 """
 
 from __future__ import annotations
@@ -42,7 +53,7 @@
     UpstreamRefusedError,
     UpstreamUnavailableError,
 )
-from sie_server.config.upstreams import Upstream, installed_upstreams
+from sie_server.config.upstreams import RateCap, Upstream, installed_upstreams
 from sie_server.observability.worker_telemetry import worker_telemetry
 
 _SLOT_RETRY_AFTER_S = 1
@@ -75,10 +86,17 @@ def unused(self) -> int:
 
 
 class UpstreamLimiter:
-    """The budget, the concurrency cap and the circuit breaker for one upstream."""
+    """The budget, the concurrency cap and the circuit breaker for one upstream.
+
+    Its refusals and breaker changes are reported to the worker telemetry
+    unless ``telemetry`` is false.
+    """
 
-    def __init__(self, name: str, upstream: Upstream, *, clock: Callable[[], float] = time.monotonic) -> None:
+    def __init__(
+        self, name: str, upstream: Upstream, *, clock: Callable[[], float] = time.monotonic, telemetry: bool = True
+    ) -> None:
         self.name = name
+        self._telemetry = telemetry
         self._clock = clock
         self._lock = threading.Lock()
         self._capacity = float(upstream.rate_cap.requests_per_minute)
@@ -93,7 +111,8 @@ def __init__(self, name: str, upstream: Upstream, *, clock: Callable[[], float]
         self._failures: deque[float] = deque()
         self._open_until: float | None = None
         self._probing = False
-        worker_telemetry().upstream_breaker_changed(upstream=name, open=False)
+        if telemetry:
+            worker_telemetry().upstream_breaker_changed(upstream=name, open=False)
 
     @contextmanager
     def batch(self, requests: int, *, concurrency: int | None = None) -> Iterator[None]:
@@ -237,21 +256,24 @@ def _open(self, now: float) -> None:
         was_closed = self._open_until is None
         self._open_until = now + self._cooldown_s
         self._failures.clear()
-        if was_closed:
+        if was_closed and self._telemetry:
             worker_telemetry().upstream_breaker_changed(upstream=self.name, open=True)
 
     def _close(self) -> None:
         self._open_until = None
-        worker_telemetry().upstream_breaker_changed(upstream=self.name, open=False)
+        if self._telemetry:
+            worker_telemetry().upstream_breaker_changed(upstream=self.name, open=False)
 
     def _refusal(self, refusal: UpstreamRefusal, wait_s: float, *, requests: int = 1) -> UpstreamRefusedError:
-        worker_telemetry().upstream_refused(upstream=self.name, refusal=refusal, requests=requests)
+        if self._telemetry:
+            worker_telemetry().upstream_refused(upstream=self.name, refusal=refusal, requests=requests)
         retry_after_s = min(RETRY_AFTER_MAX_S, max(RETRY_AFTER_MIN_S, math.ceil(wait_s)))
         return UpstreamRefusedError(self.name, refusal, retry_after_s=retry_after_s)
 
 
 _REGISTRY_LOCK = threading.Lock()
 _LIMITERS: dict[str, tuple[Upstream, UpstreamLimiter]] = {}
+_IDENTITY_LIMITERS: dict[str, tuple[Upstream, UpstreamLimiter]] = {}
 
 
 def upstream_limiter(name: str) -> UpstreamLimiter:
@@ -260,12 +282,34 @@ def upstream_limiter(name: str) -> UpstreamLimiter:
     Installing the upstreams again, as a restart of the configuration does,
     starts every limiter afresh.
     """
+    return _installed_limiter(_LIMITERS, name, lambda upstream: UpstreamLimiter(name, upstream))
+
+
+def identity_limiter(name: str) -> UpstreamLimiter:
+    """The process-wide limiter for background identity metadata refreshes from upstream ``name``.
+
+    Its configuration is the upstream's, with ``rate_cap`` set to a tenth of
+    ``requests_per_minute``, at least one, and a ``max_concurrency`` of one.
+    Installing the upstreams again starts it afresh, as it does
+    :func:`upstream_limiter`.
+    """
+
+    def build(upstream: Upstream) -> UpstreamLimiter:
+        rate_cap = RateCap(requests_per_minute=max(1, upstream.rate_cap.requests_per_minute // 10), max_concurrency=1)
+        return UpstreamLimiter(name, upstream.model_copy(update={"rate_cap": rate_cap}), telemetry=False)
+
+    return _installed_limiter(_IDENTITY_LIMITERS, name, build)
+
+
+def _installed_limiter(
+    limiters: dict[str, tuple[Upstream, UpstreamLimiter]], name: str, build: Callable[[Upstream], UpstreamLimiter]
+) -> UpstreamLimiter:
     upstream = installed_upstreams().get(name)
     if upstream is None:
         raise RuntimeError(f"upstream {name!r} is not defined in the startup configuration")
     with _REGISTRY_LOCK:
-        entry = _LIMITERS.get(name)
+        entry = limiters.get(name)
         if entry is None or entry[0] is not upstream:
-            entry = (upstream, UpstreamLimiter(name, upstream))
-            _LIMITERS[name] = entry
+            entry = (upstream, build(upstream))
+            limiters[name] = entry
         return entry[1]

packages/sie_server/src/sie_server/config/sie_identity.py

@@ -6,14 +6,15 @@
 import threading
 import time
 from collections import OrderedDict
+from collections.abc import Callable
 from dataclasses import dataclass
 from typing import Any
 
 import httpx
 from sie_sdk import RequestError, ServerError, SIEClient, SIEConnectionError
 
 from sie_server.adapters.errors import UpstreamUnavailableError
-from sie_server.adapters.remote._limits import upstream_limiter
+from sie_server.adapters.remote._limits import UpstreamLimiter, identity_limiter, upstream_limiter
 from sie_server.config.engine import EngineConfig
 from sie_server.config.equivalence import canonical_digest
 from sie_server.config.model import UPSTREAM_MODEL_PATTERN, ModelConfig, is_immutable_revision
@@ -72,7 +73,9 @@ def close(self) -> None:
         self._transport.close()
 
 
-def _read_identity(upstream_name: str, upstream: Upstream, remote_model: str) -> tuple[str, str] | None:
+def _read_identity(
+    upstream_name: str, upstream: Upstream, remote_model: str, limiter: Callable[[str], UpstreamLimiter]
+) -> tuple[str, str] | None:
     if not remote_serving_enabled() or installed_upstreams().get(upstream_name) is not upstream:
         return None
     model, separator, profile = remote_model.partition(":")
@@ -82,7 +85,7 @@ def _read_identity(upstream_name: str, upstream: Upstream, remote_model: str) ->
         http_client.headers["Accept-Encoding"] = "identity"
         with (
             SIEClient(upstream.base_url, api_key="", http_client=http_client) as client,
-            upstream_limiter(upstream_name).call(),
+            limiter(upstream_name).call(),
         ):
             try:
                 metadata: Any = client.get_model(model)
@@ -132,15 +135,22 @@ def current(self, now: float) -> tuple[tuple[str, str] | None, float]:
 _OBSERVATIONS: OrderedDict[tuple[int, str, str], _Observation] = OrderedDict()
 
 
-def _refresh(observation: _Observation, upstream_name: str, upstream: Upstream, remote_model: str) -> None:
-    """Replace the observation with a completed read.
+def _refresh(
+    observation: _Observation,
+    upstream_name: str,
+    upstream: Upstream,
+    remote_model: str,
+    limiter: Callable[[str], UpstreamLimiter],
+) -> None:
+    """Replace the observation with a completed read through ``limiter``.
 
-    A read that fails keeps the previous observation until its own expiry and
-    holds off the next read for the refusal age.
+    A read that fails, or that the limiter refuses before sending it, keeps the
+    previous observation until its own expiry and holds off the next read for
+    the refusal age.
     """
     completed, value = False, None
     try:
-        value = _read_identity(upstream_name, upstream, remote_model)
+        value = _read_identity(upstream_name, upstream, remote_model, limiter)
         completed = True
     except (UpstreamUnavailableError, UpstreamCredentialError, httpx.HTTPError, OSError, ValueError, RecursionError):
         pass
@@ -159,10 +169,12 @@ def _identity_observation(
 ) -> tuple[tuple[str, str] | None, float]:
     """Return the current observation and its age in seconds.
 
-    With ``wait`` an expired observation is refreshed before returning. Without
+    With ``wait`` an expired observation is refreshed before returning, and the
+    read counts against the upstream's rate cap like an inference call. Without
     it the refresh runs in the background, starting before expiry, and the
-    caller never waits on the upstream. A refresh already in flight is never
-    started twice.
+    caller never waits on the upstream. A background refresh draws only on the
+    upstream's identity limiter, so it never consumes the inference rate cap. A
+    refresh already in flight is never started twice.
     """
     key = id(upstream), upstream_name, remote_model
     with _LOCK:
@@ -185,10 +197,10 @@ def _identity_observation(
         observation.loading = True
     if not wait:
         threading.Thread(
-            target=_refresh, args=(observation, upstream_name, upstream, remote_model), daemon=True
+            target=_refresh, args=(observation, upstream_name, upstream, remote_model, identity_limiter), daemon=True
         ).start()
         return current
-    _refresh(observation, upstream_name, upstream, remote_model)
+    _refresh(observation, upstream_name, upstream, remote_model, upstream_limiter)
     with _LOCK:
         return observation.current(time.monotonic())
 

packages/sie_server/tests/adapters/test_remote_upstream_limits.py

@@ -21,7 +21,7 @@
 from sie_server.adapters.remote import sie as remote_sie
 from sie_server.adapters.remote._batching import call_each
 from sie_server.adapters.remote._http import RemoteUpstreamError
-from sie_server.adapters.remote._limits import UpstreamLimiter, upstream_limiter
+from sie_server.adapters.remote._limits import UpstreamLimiter, identity_limiter, upstream_limiter
 from sie_server.config.upstreams import Upstream, UpstreamConfigError, install_upstreams, load_upstreams
 from sie_server.core.upstream_client import upstream_sync_client
 from sie_server.ipc_types import EncodeBatchItem
@@ -332,6 +332,89 @@ def test_every_adapter_shares_one_limiter_per_upstream_until_the_upstreams_are_i
     assert fresh is not first
 
 
+def test_identity_reads_have_one_limiter_per_upstream_apart_from_the_inference_limiter() -> None:
+    try:
+        install_upstreams({UPSTREAM: upstream_config()})
+        first = identity_limiter(UPSTREAM)
+        same = identity_limiter(UPSTREAM)
+        inference = upstream_limiter(UPSTREAM)
+        install_upstreams({UPSTREAM: upstream_config()})
+        fresh = identity_limiter(UPSTREAM)
+
+        with pytest.raises(RuntimeError, match="not defined"):
+            identity_limiter("nobody")
+    finally:
+        install_upstreams({})
+
+    assert first is same
+    assert first is not inference
+    assert fresh is not first
+
+
+@pytest.mark.parametrize(("rpm", "reads"), [(600, 60), (60, 6), (19, 1), (5, 1)])
+def test_identity_reads_have_a_tenth_of_the_rate_cap_and_at_least_one_a_minute_of_their_own(
+    events: Events, rpm: int, reads: int
+) -> None:
+    try:
+        install_upstreams({UPSTREAM: upstream_config(rpm)})
+        identity, inference = identity_limiter(UPSTREAM), upstream_limiter(UPSTREAM)
+        for _ in range(reads):
+            call(identity)
+        refusal = refused(identity, "rate_cap")
+        with inference.batch(rpm):
+            pass
+    finally:
+        install_upstreams({})
+
+    assert refusal.retry_after_s == 60 // reads
+    assert (events.refused, events.breaker) == ([], [(UPSTREAM, False)])
+
+
+def test_inference_calls_never_take_the_identity_budget_or_its_one_slot(events: Events) -> None:
+    try:
+        install_upstreams({UPSTREAM: upstream_config(rpm=20, concurrency=4)})
+        identity, inference = identity_limiter(UPSTREAM), upstream_limiter(UPSTREAM)
+        for _ in range(16):
+            call(inference)
+        with ExitStack() as held:
+            for _ in range(4):
+                held.enter_context(inference.call())
+            refused(inference, "concurrency_cap")
+            with identity.call():
+                refused(identity, "concurrency_cap")
+            call(identity)
+        refused(inference, "rate_cap")
+    finally:
+        install_upstreams({})
+
+    assert events.refused == [(UPSTREAM, "concurrency_cap"), (UPSTREAM, "rate_cap")]
+
+
+@pytest.mark.parametrize("failing", ["identity", "inference"])
+def test_identity_reads_and_inference_calls_have_separate_breakers(failing: str) -> None:
+    try:
+        install_upstreams({UPSTREAM: upstream_config(failures=1)})
+        identity, inference = identity_limiter(UPSTREAM), upstream_limiter(UPSTREAM)
+        broken, working = (identity, inference) if failing == "identity" else (inference, identity)
+        fail(broken)
+        refused(broken, "breaker_open")
+        call(working)
+    finally:
+        install_upstreams({})
+
+
+def test_a_limiter_without_telemetry_reports_no_refusal_or_breaker_change(events: Events, clock: Clock) -> None:
+    quiet = UpstreamLimiter(UPSTREAM, upstream_config(rpm=2, failures=1, cooldown_s=60), clock=clock, telemetry=False)
+    fail(quiet)
+    refused(quiet, "breaker_open")
+    clock.advance(60)
+    call(quiet)
+    call(quiet)
+    refused(quiet, "rate_cap")
+
+    assert (events.refused, events.breaker) == ([], [])
+
+
 def test_a_breaker_defaults_to_five_failures_in_thirty_seconds_and_a_minute_of_cooldown(tmp_path: Path) -> None:
     path = tmp_path / "upstreams.yaml"
     path.write_text(