fix(gateway): fold route + authenticated principal into idempotency key
Review on this PR (tracked issue #84256) found the profile-only fix left two more dimensions of the shared OpenAI-compat Idempotency-Key cache unscoped: 1. No endpoint/route discriminator. _run_idempotent() is shared by both POST /v1/chat/completions and POST /v1/responses, but _IdempotencyCache ._store keeps the fingerprint only as a value inside the key's slot — not part of the key itself. A client reusing one Idempotency-Key across both routes would have the second route's settled response silently overwrite the first route's storage slot, so a legitimate retry of the first request would recompute instead of replaying its own cached result. Each call site now passes a stable logical route discriminator ("chat_completions" / "responses") folded directly into the cache key, rather than relying on request.path (which would wrongly split /v1/... and its /p/<profile>/v1/... alias into different namespaces). 2. No authenticated principal. The sibling durable /v1/runs admission API's _run_idempotency_scope() already composes sha256(profile \0 expected-api-key-or-sentinel) for its non-room case. _run_idempotent() now delegates directly to self._run_idempotency_scope(request) for its principal/profile scope instead of re-deriving the profile alone, so both endpoints share one definition of "who is this request for" and a caller whose API_SERVER_KEY changes mid-cache-lifetime can no longer replay a previous principal's cached response. Adds regression tests proving: route isolation (including the exact chat -> responses -> chat retry scenario from the review), concurrent cross-profile inflight isolation, principal/key-rotation isolation, and that same-route/same-profile/same-principal dedupe is unchanged. Mutation-verified with three independent negative controls: reverting the whole diff (TypeError on the now-required route kwarg), keeping route as a parameter but dropping it from the key (only the two route-isolation tests fail), and keeping the shared-scope call but reverting it to a bare profile lookup (only the two principal-isolation tests fail).
This commit is contained in:
nftpoetrist
@@ -555,6 +555,7 @@ class OpenAICompatRoutesMixin:
|
||||
outcome, err = await self._run_idempotent(
|
||||
request, body, _compute_completion, log_label="chat completions",
|
||||
fingerprint_keys=["model", "provider", "model_options", "messages", "tools", "tool_choice", "stream"],
|
||||
route="chat_completions",
|
||||
)
|
||||
if err is not None:
|
||||
return err
|
||||
@@ -599,22 +600,38 @@ class OpenAICompatRoutesMixin:
|
||||
|
||||
async def _run_idempotent(
|
||||
self, request: "web.Request", body: Dict[str, Any], compute, *,
|
||||
log_label: str, fingerprint_keys: List[str]) -> tuple:
|
||||
"""Run ``compute()`` once per profile + Idempotency-Key + body fingerprint ->
|
||||
``((result, usage), None)`` or ``(None, 500 response)``.
|
||||
log_label: str, fingerprint_keys: List[str], route: str) -> tuple:
|
||||
"""Run ``compute()`` once per (profile + authenticated principal) + logical route +
|
||||
Idempotency-Key + body fingerprint -> ``((result, usage), None)`` or
|
||||
``(None, 500 response)``.
|
||||
|
||||
The profile is folded into the cache key (mirroring ``_run_idempotency_scope``'s
|
||||
``_api_request_profile.get() or "default"``) so two ``/p/<profile>/...`` mirrors
|
||||
under ``gateway.multiplex_profiles`` never share a cached response for a client-
|
||||
supplied Idempotency-Key that happens to collide across profiles.
|
||||
The cache key's authority/namespace portion is delegated to
|
||||
``self._run_idempotency_scope(request)`` — the exact opaque
|
||||
``sha256(profile \\0 expected-api-key-or-sentinel)`` scope the sibling durable ``/v1/runs``
|
||||
admission API already computes via ``_run_idempotency_scope()`` — so this cache shares one
|
||||
definition of "who is this request for" with that endpoint rather than growing a second,
|
||||
subtly different one. That keeps two ``/p/<profile>/...`` mirrors under
|
||||
``gateway.multiplex_profiles`` from sharing a cached response for a client-supplied
|
||||
Idempotency-Key that happens to collide across profiles, and keeps a caller whose
|
||||
API_SERVER_KEY changes mid-cache-lifetime from replaying a previous principal's response.
|
||||
|
||||
``route`` is a stable, call-site-supplied logical endpoint discriminator (e.g.
|
||||
``"chat_completions"`` / ``"responses"``) — not ``request.path`` — because ``/v1/...`` and
|
||||
its ``/p/<profile>/v1/...`` alias must resolve to the same logical route after profile
|
||||
resolution. It is folded into the cache key itself (not only relied on to differ via
|
||||
``fingerprint_keys``): ``_IdempotencyCache._store`` keeps the fingerprint only as a value
|
||||
inside the key's slot, so without a route-distinct key, a client that (accidentally or by
|
||||
automation) reuses one Idempotency-Key across both ``/v1/chat/completions`` and
|
||||
``/v1/responses`` would have the second route's settled response silently overwrite the
|
||||
first route's slot, and a legitimate retry of the first request would recompute instead of
|
||||
replaying its own cached result.
|
||||
"""
|
||||
from gateway.platforms.api_server import (
|
||||
_api_request_profile, _error_response, _idem_cache, _make_request_fingerprint)
|
||||
from gateway.platforms.api_server import _error_response, _idem_cache, _make_request_fingerprint
|
||||
idempotency_key = request.headers.get("Idempotency-Key")
|
||||
try:
|
||||
if idempotency_key:
|
||||
profile = _api_request_profile.get() or "default"
|
||||
scoped_key = f"{profile}\0{idempotency_key}"
|
||||
principal_scope = self._run_idempotency_scope(request)
|
||||
scoped_key = f"{principal_scope}\0{route}\0{idempotency_key}"
|
||||
fp = _make_request_fingerprint(body, keys=fingerprint_keys)
|
||||
result, usage = await _idem_cache.get_or_set(scoped_key, fp, compute)
|
||||
else:
|
||||
@@ -892,6 +909,7 @@ class OpenAICompatRoutesMixin:
|
||||
outcome, err = await self._run_idempotent(
|
||||
request, body, _compute_response, log_label="responses",
|
||||
fingerprint_keys=["input", "instructions", "previous_response_id", "conversation", "model", "provider", "model_options", "tools"],
|
||||
route="responses",
|
||||
)
|
||||
if err is not None:
|
||||
return err
|
||||
|
||||
@@ -151,10 +151,12 @@ class TestIdempotencyCache:
|
||||
|
||||
class TestRunIdempotentProfileScope:
|
||||
"""``_run_idempotent`` (the chat completions / responses idempotency wrapper) must scope
|
||||
its cache key by the request's ``/p/<profile>/`` identity, mirroring
|
||||
``_run_idempotency_scope``'s ``_api_request_profile.get() or "default"`` — otherwise two
|
||||
multiplexed profiles sharing a client-supplied Idempotency-Key silently share a cached
|
||||
response instead of each running its own turn."""
|
||||
its cache key by the request's composite authority namespace — profile, authenticated
|
||||
principal, and logical route — mirroring the sibling durable ``/v1/runs`` admission API's
|
||||
``_run_idempotency_scope()`` (profile + expected API key) plus an explicit per-call-site
|
||||
route discriminator. See tracked issue #84256 for the full composite-namespace contract
|
||||
this class exercises: profile isolation, principal isolation, route isolation, and
|
||||
same-namespace dedup/replay must all hold simultaneously."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_same_key_different_profiles_do_not_share_a_cached_response(self, adapter, monkeypatch):
|
||||
@@ -171,14 +173,16 @@ class TestRunIdempotentProfileScope:
|
||||
token_a = _api_request_profile.set("profile-a")
|
||||
try:
|
||||
outcome_a, err_a = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"])
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
finally:
|
||||
_api_request_profile.reset(token_a)
|
||||
|
||||
token_b = _api_request_profile.set("profile-b")
|
||||
try:
|
||||
outcome_b, err_b = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"])
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
finally:
|
||||
_api_request_profile.reset(token_b)
|
||||
|
||||
@@ -186,6 +190,53 @@ class TestRunIdempotentProfileScope:
|
||||
assert len(calls) == 2, "each profile must run its own turn, not reuse the other's cached response"
|
||||
assert outcome_a != outcome_b
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_different_profiles_have_isolated_inflight_tasks(self, adapter, monkeypatch):
|
||||
"""#84256 acceptance contract: two profiles racing the *same* client Idempotency-Key
|
||||
must each get their own in-flight compute task, not one profile blocking on (or
|
||||
replaying) the other's still-running turn."""
|
||||
monkeypatch.setattr("gateway.platforms.api_server._idem_cache", _IdempotencyCache())
|
||||
request = MagicMock()
|
||||
request.headers = {"Idempotency-Key": "client-supplied-key"}
|
||||
body = {"model": "gpt-5.5", "messages": [{"role": "user", "content": "hi"}]}
|
||||
started = {"profile-a": asyncio.Event(), "profile-b": asyncio.Event()}
|
||||
gate = {"profile-a": asyncio.Event(), "profile-b": asyncio.Event()}
|
||||
calls = {"profile-a": 0, "profile-b": 0}
|
||||
|
||||
def _make_compute(profile: str):
|
||||
async def compute():
|
||||
calls[profile] += 1
|
||||
started[profile].set()
|
||||
await gate[profile].wait()
|
||||
return (f"response-{profile}", {"total_tokens": calls[profile]})
|
||||
return compute
|
||||
|
||||
async def _run(profile: str):
|
||||
token = _api_request_profile.set(profile)
|
||||
try:
|
||||
return await adapter._run_idempotent(
|
||||
request, body, _make_compute(profile), log_label="test",
|
||||
fingerprint_keys=["model", "messages"], route="chat_completions")
|
||||
finally:
|
||||
_api_request_profile.reset(token)
|
||||
|
||||
task_a = asyncio.create_task(_run("profile-a"))
|
||||
task_b = asyncio.create_task(_run("profile-b"))
|
||||
|
||||
# Both computes must be running concurrently (neither waiting on the other's slot)
|
||||
# before either is allowed to finish.
|
||||
await asyncio.wait_for(started["profile-a"].wait(), timeout=2)
|
||||
await asyncio.wait_for(started["profile-b"].wait(), timeout=2)
|
||||
assert calls == {"profile-a": 1, "profile-b": 1}
|
||||
|
||||
gate["profile-a"].set()
|
||||
gate["profile-b"].set()
|
||||
(outcome_a, err_a), (outcome_b, err_b) = await asyncio.gather(task_a, task_b)
|
||||
|
||||
assert err_a is None and err_b is None
|
||||
assert outcome_a[0] == "response-profile-a"
|
||||
assert outcome_b[0] == "response-profile-b"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_same_key_same_profile_still_dedupes(self, adapter, monkeypatch):
|
||||
"""Regression guard: profile-scoping the cache key must not break same-profile dedup,
|
||||
@@ -203,9 +254,11 @@ class TestRunIdempotentProfileScope:
|
||||
token = _api_request_profile.set("profile-a")
|
||||
try:
|
||||
outcome_1, err_1 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"])
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
outcome_2, err_2 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"])
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
finally:
|
||||
_api_request_profile.reset(token)
|
||||
|
||||
@@ -228,14 +281,154 @@ class TestRunIdempotentProfileScope:
|
||||
return (f"response-{len(calls)}", {"total_tokens": len(calls)})
|
||||
|
||||
outcome_1, err_1 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"])
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
outcome_2, err_2 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"])
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
|
||||
assert err_1 is None and err_2 is None
|
||||
assert len(calls) == 1
|
||||
assert outcome_1 == outcome_2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_same_key_different_routes_do_not_share_a_cached_response(self, adapter, monkeypatch):
|
||||
"""Endpoint/route isolation (#84256, review point 2): a client reusing one
|
||||
Idempotency-Key across /v1/chat/completions and /v1/responses must not have either
|
||||
route's cached response leak into the other, and — critically — the second route's
|
||||
settled write must not silently overwrite the first route's storage slot."""
|
||||
monkeypatch.setattr("gateway.platforms.api_server._idem_cache", _IdempotencyCache())
|
||||
request = MagicMock()
|
||||
request.headers = {"Idempotency-Key": "client-supplied-key"}
|
||||
body = {"model": "gpt-5.5", "messages": [{"role": "user", "content": "hi"}]}
|
||||
calls = []
|
||||
|
||||
async def compute():
|
||||
calls.append(1)
|
||||
return (f"response-{len(calls)}", {"total_tokens": len(calls)})
|
||||
|
||||
token = _api_request_profile.set("profile-a")
|
||||
try:
|
||||
outcome_chat, err_chat = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
outcome_resp, err_resp = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="responses")
|
||||
finally:
|
||||
_api_request_profile.reset(token)
|
||||
|
||||
assert err_chat is None and err_resp is None
|
||||
assert len(calls) == 2, "each route must run its own turn, not reuse the other route's cached response"
|
||||
assert outcome_chat != outcome_resp
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_chat_then_responses_then_chat_retry_replays_first_chat_result(self, adapter, monkeypatch):
|
||||
"""Same defect as above, phrased as the reviewer's exact retry scenario: chat (A) ->
|
||||
responses (B) -> a legitimate retry of A must replay A's own cached result rather than
|
||||
recomputing it (proving B's settled write didn't clobber A's storage slot)."""
|
||||
monkeypatch.setattr("gateway.platforms.api_server._idem_cache", _IdempotencyCache())
|
||||
request = MagicMock()
|
||||
request.headers = {"Idempotency-Key": "client-supplied-key"}
|
||||
body = {"model": "gpt-5.5", "messages": [{"role": "user", "content": "hi"}]}
|
||||
chat_calls = []
|
||||
responses_calls = []
|
||||
|
||||
async def compute_chat():
|
||||
chat_calls.append(1)
|
||||
return (f"chat-response-{len(chat_calls)}", {"total_tokens": len(chat_calls)})
|
||||
|
||||
async def compute_responses():
|
||||
responses_calls.append(1)
|
||||
return (f"responses-response-{len(responses_calls)}", {"total_tokens": len(responses_calls)})
|
||||
|
||||
token = _api_request_profile.set("profile-a")
|
||||
try:
|
||||
outcome_chat_1, err_1 = await adapter._run_idempotent(
|
||||
request, body, compute_chat, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
outcome_resp, err_2 = await adapter._run_idempotent(
|
||||
request, body, compute_responses, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="responses")
|
||||
outcome_chat_retry, err_3 = await adapter._run_idempotent(
|
||||
request, body, compute_chat, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
finally:
|
||||
_api_request_profile.reset(token)
|
||||
|
||||
assert err_1 is None and err_2 is None and err_3 is None
|
||||
assert len(chat_calls) == 1, "the chat retry must replay the first chat call's cached result"
|
||||
assert len(responses_calls) == 1
|
||||
assert outcome_chat_1 == outcome_chat_retry
|
||||
assert outcome_chat_1 != outcome_resp
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_principal_rotation_does_not_replay_old_principals_response(self, adapter, monkeypatch):
|
||||
"""Authenticated-principal isolation (#84256, review point 3): if the profile's
|
||||
expected API key changes while the process-global cache is still warm, a newly
|
||||
authorized caller reusing the same Idempotency-Key/body must not receive the previous
|
||||
principal's cached response."""
|
||||
monkeypatch.setattr("gateway.platforms.api_server._idem_cache", _IdempotencyCache())
|
||||
request = MagicMock()
|
||||
request.headers = {"Idempotency-Key": "client-supplied-key"}
|
||||
body = {"model": "gpt-5.5", "messages": [{"role": "user", "content": "hi"}]}
|
||||
calls = []
|
||||
|
||||
async def compute():
|
||||
calls.append(1)
|
||||
return (f"response-{len(calls)}", {"total_tokens": len(calls)})
|
||||
|
||||
monkeypatch.setattr(adapter, "_expected_api_key", lambda: "key-for-principal-one")
|
||||
outcome_1, err_1 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
|
||||
monkeypatch.setattr(adapter, "_expected_api_key", lambda: "key-for-principal-two")
|
||||
outcome_2, err_2 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
|
||||
assert err_1 is None and err_2 is None
|
||||
assert len(calls) == 2, "a rotated principal must run its own turn, not reuse the old principal's response"
|
||||
assert outcome_1 != outcome_2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_same_key_same_profile_same_principal_still_dedupes_via_run_idempotency_scope(
|
||||
self, adapter, monkeypatch):
|
||||
"""Cross-check that ``_run_idempotent`` really delegates its principal/profile scope to
|
||||
``self._run_idempotency_scope()`` (the same helper the sibling ``/v1/runs`` admission API
|
||||
uses), rather than reimplementing an independent definition."""
|
||||
monkeypatch.setattr("gateway.platforms.api_server._idem_cache", _IdempotencyCache())
|
||||
request = MagicMock()
|
||||
request.headers = {"Idempotency-Key": "client-supplied-key"}
|
||||
body = {"model": "gpt-5.5", "messages": [{"role": "user", "content": "hi"}]}
|
||||
scope_calls = []
|
||||
real_scope = adapter._run_idempotency_scope
|
||||
|
||||
def _spy_scope(req):
|
||||
scope_calls.append(req)
|
||||
return real_scope(req)
|
||||
|
||||
monkeypatch.setattr(adapter, "_run_idempotency_scope", _spy_scope)
|
||||
calls = []
|
||||
|
||||
async def compute():
|
||||
calls.append(1)
|
||||
return (f"response-{len(calls)}", {"total_tokens": len(calls)})
|
||||
|
||||
outcome_1, err_1 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
outcome_2, err_2 = await adapter._run_idempotent(
|
||||
request, body, compute, log_label="test", fingerprint_keys=["model", "messages"],
|
||||
route="chat_completions")
|
||||
|
||||
assert err_1 is None and err_2 is None
|
||||
assert len(scope_calls) == 2, "_run_idempotent must consult _run_idempotency_scope() on every call"
|
||||
assert len(calls) == 1
|
||||
assert outcome_1 == outcome_2
|
||||
assert outcome_1 == outcome_2
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Adapter initialization
|
||||
|
||||
Reference in New Issue
Block a user