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 ·