From bb17e190684a8a3b0888feffb6f2cd442bd84c6d Mon Sep 17 00:00:00 2001 From: krisztian-gajdar Date: Wed, 7 Oct 2026 15:52:12 +0200 Subject: [PATCH 1/2] fix(server): give upstream identity reads their own budget Identity metadata reads for hybrid SIE admission went through the same per-upstream limiter as inference calls. A process with many hybrid models on one upstream refreshes each identity about every 20 seconds, so the refreshes alone could spend the upstream's whole rate cap. Identity reads now go through a second limiter per upstream: a tenth of rate_cap.requests_per_minute, at least one read a minute, one read in flight, and a circuit breaker of its own. Inference calls never draw on it. A read it refuses is never sent and backs off like a failed read. The identity limiter reports nothing to the worker telemetry, whose upstream instruments keep describing inference calls. --- packages/sie_server/REMOTE_BACKENDS.md | 17 +++- .../src/sie_server/adapters/remote/_limits.py | 61 ++++++++++-- .../src/sie_server/config/sie_identity.py | 9 +- .../adapters/test_remote_upstream_limits.py | 85 ++++++++++++++++- .../tests/config/test_sie_identity.py | 94 ++++++++++++++++++- 5 files changed, 246 insertions(+), 20 deletions(-) diff --git a/packages/sie_server/REMOTE_BACKENDS.md b/packages/sie_server/REMOTE_BACKENDS.md index f9d01f56f..597fda6e3 100644 --- a/packages/sie_server/REMOTE_BACKENDS.md +++ b/packages/sie_server/REMOTE_BACKENDS.md @@ -200,7 +200,9 @@ 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. Identity metadata +reads for [SIE identity fallback](#admitting-sie-identity-fallback) do not +count against these limits; they have a smaller budget of their own. 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,11 +381,18 @@ 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. +Metadata reads have limits of their own for each upstream in each worker +process, apart from its rate cap: 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 read +these limits refuse is not sent and counts as a failed read. + 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 -different weights revision or identity, the server refuses the model and does -not start. A hot reload of that model runs the same comparison, reusing a +on the upstream at startup. When the upstream cannot be reached, the metadata +limits refuse the read, or the upstream reports a different weights revision or +identity, the server refuses the model and does not start. A hot reload of that +model runs the same comparison, reusing a matching observation up to 30 seconds old, and a refused reload is logged. A refused reload of a loaded model leaves it unloaded. To start while the upstream is down, remove the model's `routing` block, then add it back by hot reload once the upstream answers. 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..940473823 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,14 @@ 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. + +Reads of an SIE upstream's identity metadata for hybrid admission go through a +second limiter per upstream name, never through the first. 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. """ from __future__ import annotations @@ -42,7 +50,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 +83,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 +108,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 +253,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 +279,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 identity metadata reads 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..5c75daf46 100644 --- a/packages/sie_server/src/sie_server/config/sie_identity.py +++ b/packages/sie_server/src/sie_server/config/sie_identity.py @@ -13,7 +13,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 identity_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 @@ -82,7 +82,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(), + identity_limiter(upstream_name).call(), ): try: metadata: Any = client.get_model(model) @@ -135,8 +135,9 @@ def current(self, now: float) -> tuple[tuple[str, str] | None, float]: def _refresh(observation: _Observation, upstream_name: str, upstream: Upstream, remote_model: str) -> None: """Replace the observation with a completed read. - 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 upstream's identity 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: 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..e4c57bb68 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,95 @@ def test_a_completed_refusal_replaces_the_identity_at_once(remote, monkeypatch) assert len(requests) == 2 +def _counted_reads(monkeypatch) -> list[tuple]: + reads = [] + read = sie_identity._read_identity + + def counted(*args): + reads.append(args) + return read(*args) + + monkeypatch.setattr(sie_identity, "_read_identity", counted) + return reads + + +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_identity_refreshes_leave_the_inference_budget_untouched(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(50): + assert sie_identity.sie_identity_refusal(model(), device="cpu") is None + clock[0] += sie_identity._IDENTITY_AGE_S + + assert len(requests) == 50 + with inference.batch(upstream.rate_cap.requests_per_minute): + pass + + +def test_a_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]) + assert sie_identity.sie_upstream_identity(model()) == (REVISION, IDENTITY, 30.0) + refreshed = _tracked_refresh(monkeypatch) + reads = _counted_reads(monkeypatch) + clock[0] += 25.0 + + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 5.0) + assert refreshed.wait(5) + clock[0] += sie_identity._REFUSAL_AGE_S - 0.5 + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 3.5) + assert len(reads) == 1 + + refreshed.clear() + clock[0] += 0.5 + assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 3.0) + assert refreshed.wait(5) + assert len(reads) == 2 + assert len(requests) == 1 + + +def test_a_second_identity_read_while_one_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]) + + with identity_limiter("team").call(): + assert sie_identity.sie_upstream_identity(model()) == ( + "hybrid upstream identity is unavailable or outside its age" + ) + clock[0] += sie_identity._REFUSAL_AGE_S - 0.5 + assert sie_identity.sie_upstream_identity(model()) == "hybrid upstream identity is unavailable or outside its age" + assert requests == [] + + clock[0] += 0.5 + assert sie_identity.sie_upstream_identity(model()) == (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 + 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 + + assert sie_identity.sie_identity_refusal(model(), device="cpu") is None + 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) From 9c96f6909c7ce3d535e1f8c4c96e21f1cb254bbd Mon Sep 17 00:00:00 2001 From: krisztian-gajdar Date: Wed, 7 Oct 2026 16:18:49 +0200 Subject: [PATCH 2/2] fix(server): keep waited identity reads on the inference limiter Background identity refreshes never consume the inference rate cap: only they draw on the per-upstream identity budget. An identity read that its caller waits for, at configuration load, at hot reload and before a single-node bridge, goes through the upstream's inference limiter again, so a server starts with more hybrid models on one upstream than the identity budget allows reads a minute. _identity_observation passes the limiter for its path into _refresh and _read_identity. --- packages/sie_server/REMOTE_BACKENDS.md | 34 +++-- .../src/sie_server/adapters/remote/_limits.py | 19 +-- .../src/sie_server/config/sie_identity.py | 39 ++++-- .../tests/config/test_sie_identity.py | 120 ++++++++++++++---- 4 files changed, 150 insertions(+), 62 deletions(-) diff --git a/packages/sie_server/REMOTE_BACKENDS.md b/packages/sie_server/REMOTE_BACKENDS.md index 597fda6e3..d9a5dc36b 100644 --- a/packages/sie_server/REMOTE_BACKENDS.md +++ b/packages/sie_server/REMOTE_BACKENDS.md @@ -200,9 +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. Identity metadata -reads for [SIE identity fallback](#admitting-sie-identity-fallback) do not -count against these limits; they have a smaller budget of their own. 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 @@ -381,18 +382,27 @@ 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. -Metadata reads have limits of their own for each upstream in each worker -process, apart from its rate cap: 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 read -these limits refuse is not sent and counts as a failed read. +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, the metadata -limits refuse the read, or the upstream reports a different weights revision or -identity, the server refuses the model and does not start. A hot reload of that -model runs the same comparison, reusing a +on the upstream at startup. When the upstream cannot be reached, or reports a +different weights revision or identity, the server refuses the model and does +not start. A hot reload of that model runs the same comparison, reusing a matching observation up to 30 seconds old, and a refused reload is logged. A refused reload of a loaded model leaves it unloaded. To start while the upstream is down, remove the model's `routing` block, then add it back by hot reload once the upstream answers. 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 940473823..f60acf5da 100644 --- a/packages/sie_server/src/sie_server/adapters/remote/_limits.py +++ b/packages/sie_server/src/sie_server/adapters/remote/_limits.py @@ -24,13 +24,16 @@ through when the budget is full, and the budget then refills before anything else is sent. -Reads of an SIE upstream's identity metadata for hybrid admission go through a -second limiter per upstream name, never through the first. 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. +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 @@ -283,7 +286,7 @@ def upstream_limiter(name: str) -> UpstreamLimiter: def identity_limiter(name: str) -> UpstreamLimiter: - """The process-wide limiter for identity metadata reads from upstream ``name``. + """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. 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 5c75daf46..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 identity_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, - identity_limiter(upstream_name).call(), + limiter(upstream_name).call(), ): try: metadata: Any = client.get_model(model) @@ -132,16 +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, or that the upstream's identity limiter refuses before - sending it, 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 @@ -160,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: @@ -186,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/config/test_sie_identity.py b/packages/sie_server/tests/config/test_sie_identity.py index e4c57bb68..907c01b85 100644 --- a/packages/sie_server/tests/config/test_sie_identity.py +++ b/packages/sie_server/tests/config/test_sie_identity.py @@ -521,16 +521,22 @@ def test_a_completed_refusal_replaces_the_identity_at_once(remote, monkeypatch) assert len(requests) == 2 -def _counted_reads(monkeypatch) -> list[tuple]: - reads = [] - read = sie_identity._read_identity +def _background_refreshes(monkeypatch) -> list[threading.Thread]: + started: list[threading.Thread] = [] - def counted(*args): - reads.append(args) - return read(*args) + class Recorded(threading.Thread): + def start(self) -> None: + started.append(self) + super().start() + + monkeypatch.setattr(sie_identity, "threading", SimpleNamespace(Thread=Recorded)) + return started - monkeypatch.setattr(sie_identity, "_read_identity", counted) - return reads + +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: @@ -539,74 +545,132 @@ def _inference_limiter_without_refill(monkeypatch, upstream: Upstream) -> Upstre return inference -def test_identity_refreshes_leave_the_inference_budget_untouched(remote, monkeypatch) -> None: +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): - assert sie_identity.sie_identity_refusal(model(), device="cpu") is None - clock[0] += sie_identity._IDENTITY_AGE_S + sie_identity.sie_upstream_identity(model(), wait=False) + _settle(refreshes) + clock[0] += 25.0 - assert len(requests) == 50 + assert len(refreshes) == len(requests) == 50 with inference.batch(upstream.rate_cap.requests_per_minute): pass -def test_a_refresh_over_the_identity_budget_is_never_sent_and_backs_off_like_a_failed_read(remote, monkeypatch) -> None: +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]) - assert sie_identity.sie_upstream_identity(model()) == (REVISION, IDENTITY, 30.0) - refreshed = _tracked_refresh(monkeypatch) - reads = _counted_reads(monkeypatch) + 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) - assert refreshed.wait(5) + _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(reads) == 1 + assert len(refreshes) == 2 - refreshed.clear() clock[0] += 0.5 assert sie_identity.sie_upstream_identity(model(), wait=False) == (REVISION, IDENTITY, 3.0) - assert refreshed.wait(5) - assert len(reads) == 2 + _settle(refreshes) + assert len(refreshes) == 3 assert len(requests) == 1 -def test_a_second_identity_read_while_one_is_in_flight_is_never_sent_and_backs_off(remote, monkeypatch) -> None: +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(): - assert sie_identity.sie_upstream_identity(model()) == ( - "hybrid upstream identity is unavailable or outside its age" - ) + 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()) == "hybrid upstream identity is unavailable or outside its age" + 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 - assert sie_identity.sie_upstream_identity(model()) == (REVISION, IDENTITY, 30.0) + 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) - assert sie_identity.sie_identity_refusal(model(), device="cpu") is None + 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