diff --git a/packages/sie_server/REMOTE_BACKENDS.md b/packages/sie_server/REMOTE_BACKENDS.md index f9d01f56f..d9a5dc36b 100644 --- a/packages/sie_server/REMOTE_BACKENDS.md +++ b/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 diff --git a/packages/sie_server/src/sie_server/adapters/remote/_limits.py b/packages/sie_server/src/sie_server/adapters/remote/_limits.py index bb49ac682..f60acf5da 100644 --- a/packages/sie_server/src/sie_server/adapters/remote/_limits.py +++ b/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] diff --git a/packages/sie_server/src/sie_server/config/sie_identity.py b/packages/sie_server/src/sie_server/config/sie_identity.py index 45eeaa843..abda3b19e 100644 --- a/packages/sie_server/src/sie_server/config/sie_identity.py +++ b/packages/sie_server/src/sie_server/config/sie_identity.py @@ -6,6 +6,7 @@ import threading import time from collections import OrderedDict +from collections.abc import Callable from dataclasses import dataclass from typing import Any @@ -13,7 +14,7 @@ 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()) diff --git a/packages/sie_server/tests/adapters/test_remote_upstream_limits.py b/packages/sie_server/tests/adapters/test_remote_upstream_limits.py index 554718c77..e27916e05 100644 --- a/packages/sie_server/tests/adapters/test_remote_upstream_limits.py +++ b/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( diff --git a/packages/sie_server/tests/config/test_sie_identity.py b/packages/sie_server/tests/config/test_sie_identity.py index 3ff40e6c4..907c01b85 100644 --- a/packages/sie_server/tests/config/test_sie_identity.py +++ b/packages/sie_server/tests/config/test_sie_identity.py @@ -14,12 +14,15 @@ import yaml from fastapi import HTTPException, Request from opentelemetry import trace +from sie_server.adapters.errors import UpstreamRefusedError +from sie_server.adapters.remote import _limits +from sie_server.adapters.remote._limits import UpstreamLimiter, identity_limiter from sie_server.api.routing import route_request from sie_server.config import hybrid_admission, sie_identity from sie_server.config.equivalence import model_contract_digest, remote_profile_contract_digest from sie_server.config.model import ModelConfig from sie_server.config.routing import validate_model_routing -from sie_server.config.upstreams import Upstream, install_upstreams +from sie_server.config.upstreams import RateCap, Upstream, install_upstreams from sie_server.core.loader import expand_profile_variants from sie_server.core.registry import ModelRegistry from sie_server.ipc_server import IpcServer @@ -518,6 +521,159 @@ def test_a_completed_refusal_replaces_the_identity_at_once(remote, monkeypatch) assert len(requests) == 2 +def _background_refreshes(monkeypatch) -> list[threading.Thread]: + started: list[threading.Thread] = [] + + class Recorded(threading.Thread): + def start(self) -> None: + started.append(self) + super().start() + + monkeypatch.setattr(sie_identity, "threading", SimpleNamespace(Thread=Recorded)) + return started + + +def _settle(refreshes: list[threading.Thread]) -> None: + for refresh in refreshes: + refresh.join(5) + assert not refresh.is_alive() + + +def _inference_limiter_without_refill(monkeypatch, upstream: Upstream) -> UpstreamLimiter: + inference = UpstreamLimiter("team", upstream, clock=lambda: 0.0) + monkeypatch.setitem(_limits._LIMITERS, "team", (upstream, inference)) + return inference + + +def test_background_identity_refreshes_never_consume_the_inference_rate_cap(remote, monkeypatch) -> None: + upstream, requests, _constructions, _payload = remote + clock = [1000.0] + monkeypatch.setattr(sie_identity.time, "monotonic", lambda: clock[0]) + inference = _inference_limiter_without_refill(monkeypatch, upstream) + refreshes = _background_refreshes(monkeypatch) + + for _ in range(50): + sie_identity.sie_upstream_identity(model(), wait=False) + _settle(refreshes) + clock[0] += 25.0 + + assert len(refreshes) == len(requests) == 50 + with inference.batch(upstream.rate_cap.requests_per_minute): + pass + + +def test_identity_reads_a_check_waits_for_count_against_the_inference_rate_cap(remote, monkeypatch) -> None: + upstream, requests, _constructions, _payload = remote + clock = [1000.0] + monkeypatch.setattr(sie_identity.time, "monotonic", lambda: clock[0]) + inference = _inference_limiter_without_refill(monkeypatch, upstream) + budget = upstream.rate_cap.requests_per_minute + + with identity_limiter("team").call(): + for _ in range(3): + assert sie_identity.sie_identity_refusal(model(), device="cpu") is None + clock[0] += sie_identity._IDENTITY_AGE_S + + assert len(requests) == 3 + with pytest.raises(UpstreamRefusedError, match="rate cap"), inference.batch(budget - 2): + pass + with inference.batch(budget - 3): + pass + + +def test_a_server_with_more_hybrid_models_on_one_upstream_than_identity_reads_a_minute_starts(remote, tmp_path) -> None: + upstream, requests, _constructions, payload = remote + one_read_a_minute = RateCap(requests_per_minute=10, max_concurrency=4) + install_upstreams({"team": upstream.model_copy(update={"rate_cap": one_read_a_minute})}) + profiles = [f"stable-{index}" for index in range(3)] + payload[0] = httpx.Response( + 200, json=metadata(profiles={name: {"identity": IDENTITY, "remote_contract_sha256": None} for name in profiles}) + ) + models = tmp_path / "models" + models.mkdir() + for index, profile in enumerate(profiles): + data = model(f"operator/model:{profile}").model_dump(mode="json") + data["sie_id"] = f"local/model-{index}" + (models / f"model-{index}.yaml").write_text(yaml.safe_dump(data)) + + registry = ModelRegistry(models_dir=models, device="cpu", enable_hot_reload=False) + + assert all(registry.has_model(f"local/model-{index}") for index in range(len(profiles))) + assert len(requests) == len(profiles) + + +def test_a_background_refresh_over_the_identity_budget_is_never_sent_and_backs_off_like_a_failed_read( + remote, monkeypatch +) -> None: + upstream, requests, _constructions, _payload = remote + one_read_a_minute = RateCap(requests_per_minute=10, max_concurrency=4) + install_upstreams({"team": upstream.model_copy(update={"rate_cap": one_read_a_minute})}) + clock = [1000.0] + monkeypatch.setattr(sie_identity.time, "monotonic", lambda: clock[0]) + refreshes = _background_refreshes(monkeypatch) + sie_identity.sie_upstream_identity(model(), wait=False) + _settle(refreshes) + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 30.0) + clock[0] += 25.0 + + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 5.0) + _settle(refreshes) + clock[0] += sie_identity._REFUSAL_AGE_S - 0.5 + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 3.5) + assert len(refreshes) == 2 + + clock[0] += 0.5 + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 3.0) + _settle(refreshes) + assert len(refreshes) == 3 + assert len(requests) == 1 + + +def test_a_background_refresh_while_another_identity_read_is_in_flight_is_never_sent_and_backs_off( + remote, monkeypatch +) -> None: + _upstream, requests, _constructions, _payload = remote + clock = [1000.0] + monkeypatch.setattr(sie_identity.time, "monotonic", lambda: clock[0]) + refreshes = _background_refreshes(monkeypatch) + + with identity_limiter("team").call(): + sie_identity.sie_upstream_identity(model(), wait=False) + _settle(refreshes) + clock[0] += sie_identity._REFUSAL_AGE_S - 0.5 + assert ( + sie_identity.sie_upstream_identity(model(), wait=False) + == "hybrid upstream identity is unavailable or outside its age" + ) + assert len(refreshes) == 1 + assert requests == [] + + clock[0] += 0.5 + sie_identity.sie_upstream_identity(model(), wait=False) + _settle(refreshes) + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 30.0) + assert len(requests) == 1 + + +def test_inference_calls_never_spend_the_identity_budget(remote, monkeypatch) -> None: + upstream, requests, _constructions, _payload = remote + clock = [1000.0] + monkeypatch.setattr(sie_identity.time, "monotonic", lambda: clock[0]) + inference = _inference_limiter_without_refill(monkeypatch, upstream) + for _ in range(upstream.rate_cap.requests_per_minute): + with inference.call(): + pass + with pytest.raises(UpstreamRefusedError, match="rate cap"), inference.call(): + pass + refreshes = _background_refreshes(monkeypatch) + + sie_identity.sie_upstream_identity(model(), wait=False) + _settle(refreshes) + + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 30.0) + assert len(requests) == 1 + + async def test_remote_lane_snapshot_reports_the_sie_admission(remote, tmp_path, monkeypatch) -> None: monkeypatch.setenv("SIE_IPC_SOCKET_PATH", str(tmp_path / "ipc.sock")) registry = ModelRegistry(device="cpu", enable_hot_reload=False)