Blocking I/O in async completion and Starlette middleware can stall concurrent requests
维护者通常 1 天内回复
还没有人认领这个 Issue。
评估
- 难度
- 4/5
- 预计耗时
- 3-5 天
- 新手友好度
- 65/100
- Issue 类型
- 缺陷
- 描述清晰度
- 描述清楚
- 活跃度
- 活跃
- 技术栈
- python
- 领域
- backend, performance
调研方向
Start by tracing the async completion wrappers in src/sap_cloud_sdk/aicore/completion.py and src/sap_cloud_sdk/aicore/__init__.py, then read the Starlette middleware in src/sap_cloud_sdk/core/runtime_context/starlette.py and src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py. The issue includes an example thread offloading approach and regression tests in tests/core/unit/test_async_blocking.py; compare it with the native async option and check existing caching, context propagation, and retry behavior. Done means slow synchronous dependencies no longer stall unrelated async work while the stated context and retry behavior remain correct.
由索引模型根据 Issue 内容生成。
描述
Some async SDK paths invoke synchronous operations that may perform network or filesystem I/O:
- AI Core async completion reloads credentials synchronously after an authentication failure, both through the SDK completion wrapper and the LiteLLM credential-rotation patch. Reloading may read mounted secrets or call Destination Service.
- Starlette runtime context middleware invokes synchronous providers. IAS verification may fetch JWKS signing keys over HTTP.
- Starlette IAS telemetry middleware also invokes synchronous token verification.
While these operations run, they block the event loop, delaying unrelated requests and async tasks. The impact is particularly visible during slow network responses, credential refreshes, or JWKS cache misses.
Possible approaches
- Implement native async I/O. Add async Destination and OAuth requests, credential resolution, and JWKS retrieval. Preserve existing caching, key rotation, validation, and synchronous APIs.
- Offload synchronous operations to worker threads. Use Starlette's run_in_threadpool in its middleware and asyncio.to_thread in SDK async paths. This requires fewer changes, but introduces thread-pool capacity and thread-safety considerations; cancelling an await does not stop the underlying operation.
Either approach should verify that slow dependencies do not stall concurrent async work, request context remains available downstream, and credential reload still retries correctly.
Example patch for approach 2
The following patch illustrates thread offloading as one possible solution. It batches context provider extraction into one worker call, preserving provider order. It also includes regression tests that hold a synchronous dependency blocked while the event loop releases it, and check downstream context and retry behavior.
Local validation: 253 targeted tests passed across async blocking, runtime context, Starlette IAS telemetry, and AI Core unit tests. Lint and formatting checks passed.
Suggested implementation and regression tests
diff --git a/src/sap_cloud_sdk/aicore/__init__.py b/src/sap_cloud_sdk/aicore/__init__.py
index 9fd0f85..940569e 100644
--- a/src/sap_cloud_sdk/aicore/__init__.py
+++ b/src/sap_cloud_sdk/aicore/__init__.py
@@ -4,6 +4,7 @@ This module provides utilities to load AI Core credentials from mounted secrets
variables and configure them for use with LiteLLM.
"""
+import asyncio
import json
import logging
import os
@@ -430,7 +431,8 @@ def patch_litellm_for_credential_rotation() -> None:
try:
return await _orig_acompletion(*args, **kwargs)
except _litellm.AuthenticationError:
- _reload_reactive()
+ # Secret loading and Destination resolution have no async APIs.
+ await asyncio.to_thread(_reload_reactive)
return await _orig_acompletion(*args, **kwargs)
_litellm.completion = _completion
diff --git a/src/sap_cloud_sdk/aicore/completion.py b/src/sap_cloud_sdk/aicore/completion.py
index d97dc63..2e30952 100644
--- a/src/sap_cloud_sdk/aicore/completion.py
+++ b/src/sap_cloud_sdk/aicore/completion.py
@@ -64,6 +64,7 @@ adoption of the SDK.
from __future__ import annotations
+import asyncio
import logging
from collections.abc import AsyncIterator, Iterator
from typing import Any
@@ -198,7 +199,8 @@ async def acompletion(*args: Any, **kwargs: Any) -> Any:
try:
result = await litellm.acompletion(*args, **kwargs)
except litellm.AuthenticationError:
- _reload_reactive()
+ # Secret loading and Destination resolution only expose synchronous APIs.
+ await asyncio.to_thread(_reload_reactive)
result = await litellm.acompletion(*args, **kwargs)
except Exception as exc:
translated = _maybe_translate_filter_error(exc)
diff --git a/src/sap_cloud_sdk/core/runtime_context/starlette.py b/src/sap_cloud_sdk/core/runtime_context/starlette.py
index eb0fcb0..b191b71 100644
--- a/src/sap_cloud_sdk/core/runtime_context/starlette.py
+++ b/src/sap_cloud_sdk/core/runtime_context/starlette.py
@@ -10,6 +10,7 @@ from sap_cloud_sdk.core.runtime_context._envelope import RequestEnvelope
from sap_cloud_sdk.core.runtime_context._protocol import ContextProvider
try:
+ from starlette.concurrency import run_in_threadpool
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
@@ -47,6 +48,12 @@ class StarletteContextMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next: Any) -> Response:
envelope = RequestEnvelope(headers=dict(request.headers))
- ctx = _merge([p.extract(envelope) for p in self._providers])
+ # The provider API is synchronous; IAS verification uses PyJWKClient,
+ # which can fetch JWKS over the network. Keep extraction in one worker
+ # call to preserve provider order without a handoff for each provider.
+ ctx = await run_in_threadpool(self._extract_context, envelope)
async with async_sdk_context(ctx):
return await call_next(request)
+
+ def _extract_context(self, envelope: RequestEnvelope) -> RuntimeContext:
+ return _merge([p.extract(envelope) for p in self._providers])
diff --git a/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py b/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py
index 14162ea..f7bf565 100644
--- a/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py
+++ b/src/sap_cloud_sdk/core/telemetry/middleware/starlette_a2a.py
@@ -19,6 +19,7 @@ from sap_cloud_sdk.ias import set_auth_context
from sap_cloud_sdk.ias._token import IASClaims
try:
+ from starlette.concurrency import run_in_threadpool
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
@@ -43,7 +44,10 @@ class _IASMiddleware(BaseHTTPMiddleware):
self._token_verifier = token_verifier
async def dispatch(self, request: Request, call_next: Any) -> Response:
- verified_claims, attrs = _verify_and_extract(request, self._token_verifier)
+ # PyJWKClient has no async API and may fetch signing keys over the network.
+ verified_claims, attrs = await run_in_threadpool(
+ _verify_and_extract, request, self._token_verifier
+ )
set_auth_context(verified_claims)
token = self._attrs_var.set(attrs)
try:
diff --git a/tests/core/unit/test_async_blocking.py b/tests/core/unit/test_async_blocking.py
new file mode 100644
index 0000000..54eb111
--- /dev/null
+++ b/tests/core/unit/test_async_blocking.py
@@ -0,0 +1,122 @@
+"""Blocking SDK dependencies must not stall unrelated async work."""
+
+import asyncio
+from contextvars import ContextVar
+import threading
+from unittest.mock import AsyncMock, MagicMock
+
+import litellm
+import pytest
+
+from sap_cloud_sdk import aicore
+from sap_cloud_sdk.core.runtime_context import IASContextProvider, get_context
+from sap_cloud_sdk.core.runtime_context.providers._ias import GLOBAL_TENANT_ID
+from sap_cloud_sdk.core.runtime_context.starlette import StarletteContextMiddleware
+from sap_cloud_sdk.core.telemetry.constants import ATTR_SAP_TENANT_ID
+from sap_cloud_sdk.core.telemetry.middleware.starlette_a2a import _IASMiddleware
+from sap_cloud_sdk.ias import IASClaims, VerifiedIASClaims, get_auth_context
+
+
+class BlockingDependency:
+ """Hold worker I/O until the event loop explicitly releases it."""
+
+ def __init__(self):
+ self.loop = asyncio.get_running_loop()
+ self.started = asyncio.Event()
+ self.release = threading.Event()
+
+ def __call__(self):
+ self.loop.call_soon_threadsafe(self.started.set)
+ if not self.release.wait(timeout=5):
+ raise RuntimeError("Event loop could not release blocking I/O")
+
+ async def run(self, operation):
+ task = asyncio.create_task(operation)
+ try:
+ await asyncio.wait_for(self.started.wait(), timeout=2)
+ # This point is reached while the synchronous dependency is blocked.
+ # A direct call on the event loop would prevent reaching it in time.
+ assert not task.done()
+ finally:
+ self.release.set()
+ await task
+ return task.result()
+
+
[email protected]
[email protected]("middleware_kind", ["telemetry", "runtime_context"])
+async def test_slow_verification_yields_and_preserves_downstream_context(
+ middleware_kind,
+):
+ blocker = BlockingDependency()
+ caller_value = ContextVar("caller_value", default=None)
+ caller_value.set("request-local")
+ claims = IASClaims(sap_gtid="tenant-1")
+
+ def verify(authorization):
+ assert authorization == "Bearer token"
+ assert caller_value.get() == "request-local"
+ blocker()
+ return VerifiedIASClaims(claims=claims)
+
+ request = MagicMock()
+ request.headers = {"authorization": "Bearer token"}
+ attrs = ContextVar("test_attrs", default={})
+ if middleware_kind == "telemetry":
+ middleware = _IASMiddleware(MagicMock(), attrs, verify)
+ else:
+ middleware = StarletteContextMiddleware(
+ MagicMock(), providers=[IASContextProvider(token_verifier=verify)]
+ )
+
+ response = object()
+
+ async def call_next(req):
+ assert req is request
+ assert caller_value.get() == "request-local"
+ if middleware_kind == "telemetry":
+ assert get_auth_context() is claims
+ assert attrs.get() == {ATTR_SAP_TENANT_ID: "tenant-1"}
+ else:
+ assert get_context().get(GLOBAL_TENANT_ID) == "tenant-1"
+ return response
+
+ assert await blocker.run(middleware.dispatch(request, call_next)) is response
+ assert attrs.get() == {}
+ assert get_context().get(GLOBAL_TENANT_ID) is None
+
+
[email protected]
[email protected]("patched", [False, True])
[email protected]("retry_fails", [False, True])
+async def test_slow_credential_reload_yields_and_retries_once(
+ monkeypatch, patched, retry_fails
+):
+ blocker = BlockingDependency()
+ error = litellm.AuthenticationError(
+ message="401", llm_provider="sap", model="sap/x"
+ )
+ response = object()
+ completion = AsyncMock(side_effect=[error, error if retry_fails else response])
+ reload_config = MagicMock(side_effect=blocker)
+ monkeypatch.setattr(litellm, "acompletion", completion)
+ monkeypatch.setattr(aicore, "set_aicore_config", reload_config)
+ if patched:
+ monkeypatch.setattr(litellm, "_sap_aicore_patched", False, raising=False)
+ # Restore both functions after the patch installs its wrappers.
+ monkeypatch.setattr(litellm, "completion", litellm.completion)
+ aicore.patch_litellm_for_credential_rotation()
+ call = litellm.acompletion
+ else:
+ call = aicore.acompletion
+
+ if retry_fails:
+ with pytest.raises(litellm.AuthenticationError) as caught:
+ await blocker.run(call(model="sap/x", messages=[]))
+ assert caught.value is error
+ else:
+ assert await blocker.run(call(model="sap/x", messages=[])) is response
+ reload_config.assert_called_once_with()
+ assert completion.await_count == 2
+ for invocation in completion.await_args_list:
+ assert invocation.kwargs == {"model": "sap/x", "messages": []}
- 主要语言
- Python
- 星标
- 29
- 派生
- 45
- 平均合并
- 2 天 4 小时
- 30 天内合并 PR
- 51
环境准备
- 没有 Dockerfile 或 Docker Compose 文件
- 有 Pull Request 模板
- 阅读贡献指南
从这里开始
- 先读完整个 Issue,再读项目的贡献指南。
- 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
- Fork 仓库,在一个分支上完成修改。
- 提交 Pull Request,并在描述里引用这个 Issue 编号。
SAP/cloud-sdk-python 的其他 Issue
-
难度 4/5 3-5 天 新手友好度 15/100
SAP/cloud-sdk-python#474 ·
维护者通常 1 天内回复
-
Relax opentelemetry-api cap (~=1.44.0) to allow 1.45.2 (fixes CVE-2026-114603 / GHSA-xcq7-p3pf-p88h)未关闭
难度 3/5 1-2 天 新手友好度 64/100
SAP/cloud-sdk-python#472 ·
维护者通常 1 天内回复
-
bug
难度 4/5 3-5 天 新手友好度 25/100
SAP/cloud-sdk-python#456 ·
维护者通常 1 天内回复
-
How/Whether langgraph-checkpoint-hana would be supported by sap-cloud-sdk?可能已有人在做 @gpretto 于 3 天前认领。 未关闭question
难度 5/5 一周以上 新手友好度 15/100
SAP/cloud-sdk-python#427 · 2 条评论 · 已指派 1 人 ·
维护者通常 1 天内回复
-
`list_mcp_tools` hangs when MCP endpoint returns plain-text error as `text/event-stream`可能重新可做 @cassiofariasmachado 于 54 天前认领,目前没有进行中的 PR。 未关闭bug
SAP/cloud-sdk-python#348 · 1 条评论 · 已指派 1 人 ·
维护者通常 1 天内回复
查看 SAP/cloud-sdk-python 的全部 Issue
相似的 Issue
-
难度 2/5 1-3 小时 新手友好度 66/100
deepchem/deepchem#5289 · 1 条评论 ·
维护者通常 2 天内回复
-
难度 1/5 1 小时以内 新手友好度 75/100
-
难度 2/5 1-3 小时 新手友好度 60/100
UKGovernmentBEIS/inspect_evals#2892 · 1 条评论 ·
维护者通常 1 天内回复
-
Versioning docs: say that no output from the 0.46 warning check means nothing is affected可能已有人在做 @grelinfo 今天认领。 未关闭documentation priority:low size:S
难度 1/5 1 小时以内 新手友好度 72/100
维护者通常 1 天内回复
-
难度 1/5 1 小时以内 新手友好度 88/100
alphacrack/readme2demo#293 ·