fix: make every client close what it built and nothing else
A client used to close whatever transport, recorder or cache it happened to hold, injected or not, so the first client to shut down killed the backend its siblings were still using. That is why the explicit-sharing path the architecture prescribes was unusable in practice and downstream projects fell back to one private instance per client. The mirror image of the same gap: the redis clients the factories build for the limiter and the breaker were never closed at all, because nobody kept a reference to them once they were handed to the retry middleware. Ownership is now stated once, the way RedisLimiter already stated it: whoever builds a resource closes it, injected ones are left alone. The constructor is the full-injection path, so it owns nothing by default and only the factories mark what they built. RedisCache gains the same rule for its own client, and the three copies of the "probe for aclose, fall back to close" dance collapse into a single helper so the next correction cannot land in only one of them.
This commit is contained in:
+294
-14
@@ -45,6 +45,20 @@ _ENV = {
|
||||
"PGW_TELEMETRY_BACKEND": "none",
|
||||
}
|
||||
|
||||
_OCR_ENV = {
|
||||
"OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866",
|
||||
"OCR__MONKEY__1__API_KEY": "none",
|
||||
"OCR__MONKEY__1__MODEL": "monkey-ocr",
|
||||
"OCR__MONKEY__1__TIMEOUT_S": "120",
|
||||
"LLM_MAX_RETRIES": "3",
|
||||
"LLM_RETRY_BASE_DELAY": "2.0",
|
||||
"LLM_RETRY_MAX_DELAY": "30.0",
|
||||
"LLM_CIRCUIT_BREAKER_THRESHOLD": "5",
|
||||
"LLM_CIRCUIT_BREAKER_COOLDOWN": "60",
|
||||
"PGW_CACHE_BACKEND": "none",
|
||||
"PGW_TELEMETRY_BACKEND": "none",
|
||||
}
|
||||
|
||||
|
||||
def _sse(content='{"answer": 1}'):
|
||||
chunk = json.dumps({"choices": [{"delta": {"content": content}}]})
|
||||
@@ -359,20 +373,7 @@ class TestTelemetryTextCapWiring:
|
||||
"""
|
||||
|
||||
_CAP_ENV = dict(_ENV, PGW_TELEMETRY_TEXT_CAP="8")
|
||||
_OCR_CAP_ENV = {
|
||||
"OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866",
|
||||
"OCR__MONKEY__1__API_KEY": "none",
|
||||
"OCR__MONKEY__1__MODEL": "monkey-ocr",
|
||||
"OCR__MONKEY__1__TIMEOUT_S": "120",
|
||||
"LLM_MAX_RETRIES": "3",
|
||||
"LLM_RETRY_BASE_DELAY": "2.0",
|
||||
"LLM_RETRY_MAX_DELAY": "30.0",
|
||||
"LLM_CIRCUIT_BREAKER_THRESHOLD": "5",
|
||||
"LLM_CIRCUIT_BREAKER_COOLDOWN": "60",
|
||||
"PGW_CACHE_BACKEND": "none",
|
||||
"PGW_TELEMETRY_BACKEND": "none",
|
||||
"PGW_TELEMETRY_TEXT_CAP": "8",
|
||||
}
|
||||
_OCR_CAP_ENV = dict(_OCR_ENV, PGW_TELEMETRY_TEXT_CAP="8")
|
||||
|
||||
def test_gateway_from_settings_wires_the_cap(self):
|
||||
settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV)
|
||||
@@ -555,3 +556,282 @@ class TestReferenceProtocolCompat:
|
||||
): ...
|
||||
|
||||
assert isinstance(_client(), proto)
|
||||
|
||||
|
||||
# —— 资源所有权纪律(issue #15 D 组): 谁建的谁关,注入的一律不碰 ——
|
||||
|
||||
_CACHE_ENV = dict(
|
||||
_ENV, PGW_CACHE_BACKEND="memory", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="3600"
|
||||
)
|
||||
|
||||
|
||||
class _Closable:
|
||||
"""记 close 次数的假组件;所有权纪律的唯一观测点。"""
|
||||
|
||||
def __init__(self):
|
||||
self.closed = 0
|
||||
|
||||
async def aclose(self):
|
||||
self.closed += 1
|
||||
|
||||
|
||||
class _SyncClosable:
|
||||
"""只有同步 close 的假 recorder(SQLiteRecorder 形态,收敛后的 helper 须探测到)。"""
|
||||
|
||||
def __init__(self):
|
||||
self.closed = 0
|
||||
|
||||
def close(self):
|
||||
self.closed += 1
|
||||
|
||||
|
||||
def _parts(*names):
|
||||
return {name: _Closable() for name in names}
|
||||
|
||||
|
||||
def _patch_builders(monkeypatch, built, *, transport_path):
|
||||
"""把工厂的自建点换成可计数假件;transport 无注入入口,故恒自建。"""
|
||||
monkeypatch.setattr(transport_path, lambda **kwargs: built["transport"])
|
||||
monkeypatch.setattr("polygateway.client._build_limiter", lambda s, src: built["limiter"])
|
||||
monkeypatch.setattr("polygateway.client._build_breaker", lambda s: built["breaker"])
|
||||
monkeypatch.setattr("polygateway.client._build_telemetry", lambda s: built["telemetry"])
|
||||
if "cache" in built:
|
||||
monkeypatch.setattr("polygateway.client._build_cache", lambda s: built["cache"])
|
||||
|
||||
|
||||
class TestGatewayClientOwnership:
|
||||
"""`__init__` 是全量注入路径,经它传入的一切都归调用方(设计 §3.4)。"""
|
||||
|
||||
_GATEWAY_TRANSPORT = "polygateway.client.OpenAICompatTransport"
|
||||
|
||||
async def test_injected_components_are_never_closed(self):
|
||||
"""共享 recorder/transport 被第一个关闭的 client 弄死,正是 R5 显式共享走不通的原因。"""
|
||||
injected = _parts(*("transport", "telemetry", "cache", "limiter", "breaker"))
|
||||
client = _client(
|
||||
transport=injected["transport"],
|
||||
telemetry=injected["telemetry"],
|
||||
cache=injected["cache"],
|
||||
cache_namespace="proj",
|
||||
cache_ttl_s=3600,
|
||||
limiter=injected["limiter"],
|
||||
breaker=injected["breaker"],
|
||||
)
|
||||
await client.aclose()
|
||||
assert {name: part.closed for name, part in injected.items()} == {
|
||||
"transport": 0,
|
||||
"telemetry": 0,
|
||||
"cache": 0,
|
||||
"limiter": 0,
|
||||
"breaker": 0,
|
||||
}
|
||||
|
||||
async def test_factory_closes_every_component_it_built(self, monkeypatch):
|
||||
"""泄漏钉子: 自建的 redis limiter/breaker 今天没人关,连引用都没留。"""
|
||||
built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
|
||||
_patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
|
||||
client = GatewayClient.from_settings(GatewaySettings.from_env("LLM", env=_CACHE_ENV))
|
||||
await client.aclose()
|
||||
assert {name: part.closed for name, part in built.items()} == {
|
||||
"transport": 1,
|
||||
"telemetry": 1,
|
||||
"cache": 1,
|
||||
"limiter": 1,
|
||||
"breaker": 1,
|
||||
}
|
||||
|
||||
async def test_factory_keeps_hands_off_injected_components(self, monkeypatch):
|
||||
built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
|
||||
_patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
|
||||
injected = _parts("telemetry", "cache", "limiter", "breaker")
|
||||
client = GatewayClient.from_settings(
|
||||
GatewaySettings.from_env("LLM", env=_CACHE_ENV),
|
||||
limiter=injected["limiter"],
|
||||
breaker=injected["breaker"],
|
||||
cache=injected["cache"],
|
||||
telemetry=injected["telemetry"],
|
||||
)
|
||||
await client.aclose()
|
||||
assert all(part.closed == 0 for part in injected.values())
|
||||
assert built["transport"].closed == 1 # 工厂恒自建 transport,归 client
|
||||
|
||||
async def test_aclose_is_idempotent(self, monkeypatch):
|
||||
built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
|
||||
_patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
|
||||
client = GatewayClient.from_settings(GatewaySettings.from_env("LLM", env=_CACHE_ENV))
|
||||
await client.aclose()
|
||||
await client.aclose()
|
||||
assert all(part.closed == 1 for part in built.values())
|
||||
|
||||
async def test_sync_only_recorder_is_closed(self, monkeypatch):
|
||||
"""SQLiteRecorder 只有同步 `close()`;收敛成 helper 之后这条分支不得丢。"""
|
||||
built = _parts("transport", "cache", "limiter", "breaker")
|
||||
recorder = _SyncClosable()
|
||||
built["telemetry"] = recorder
|
||||
_patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
|
||||
client = GatewayClient.from_settings(GatewaySettings.from_env("LLM", env=_CACHE_ENV))
|
||||
await client.aclose()
|
||||
assert recorder.closed == 1
|
||||
|
||||
|
||||
def _embedding_client(**overrides):
|
||||
from polygateway.embedding import EmbeddingClient
|
||||
|
||||
defaults = {
|
||||
"scope": "embed",
|
||||
"sources": [_source()],
|
||||
"selector": RoundRobinSelector(),
|
||||
"limiter": _Closable(),
|
||||
"breaker": _Closable(),
|
||||
"transport": _Closable(),
|
||||
"retry": RetryPolicy(3, 2.0, 30.0),
|
||||
"backpressure": BackpressurePolicy(300.0, 0.01),
|
||||
"batch_size": 2,
|
||||
}
|
||||
defaults.update(overrides)
|
||||
return EmbeddingClient(**defaults)
|
||||
|
||||
|
||||
class TestEmbeddingClientOwnership:
|
||||
"""三处必须各钉一次: 收敛成 helper 之后,有人把逻辑复制回去也得当场被发现。"""
|
||||
|
||||
async def test_injected_components_are_never_closed(self):
|
||||
injected = _parts("transport", "telemetry", "limiter", "breaker")
|
||||
client = _embedding_client(
|
||||
transport=injected["transport"],
|
||||
telemetry=injected["telemetry"],
|
||||
limiter=injected["limiter"],
|
||||
breaker=injected["breaker"],
|
||||
)
|
||||
await client.aclose()
|
||||
assert all(part.closed == 0 for part in injected.values())
|
||||
|
||||
async def test_factory_closes_every_component_it_built(self, monkeypatch):
|
||||
from polygateway.config import EmbeddingSettings
|
||||
from polygateway.embedding import EmbeddingClient
|
||||
|
||||
built = _parts("transport", "telemetry", "limiter", "breaker")
|
||||
_patch_builders(
|
||||
monkeypatch,
|
||||
built,
|
||||
transport_path="polygateway.transports.openai_compat.OpenAICompatTransport",
|
||||
)
|
||||
settings = EmbeddingSettings(
|
||||
gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2
|
||||
)
|
||||
client = EmbeddingClient.from_settings(settings)
|
||||
await client.aclose()
|
||||
assert all(part.closed == 1 for part in built.values())
|
||||
|
||||
async def test_factory_keeps_hands_off_injected_components(self, monkeypatch):
|
||||
from polygateway.config import EmbeddingSettings
|
||||
from polygateway.embedding import EmbeddingClient
|
||||
|
||||
built = _parts("transport", "telemetry", "limiter", "breaker")
|
||||
_patch_builders(
|
||||
monkeypatch,
|
||||
built,
|
||||
transport_path="polygateway.transports.openai_compat.OpenAICompatTransport",
|
||||
)
|
||||
injected = _parts("telemetry", "limiter", "breaker")
|
||||
settings = EmbeddingSettings(
|
||||
gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2
|
||||
)
|
||||
client = EmbeddingClient.from_settings(
|
||||
settings,
|
||||
limiter=injected["limiter"],
|
||||
breaker=injected["breaker"],
|
||||
telemetry=injected["telemetry"],
|
||||
)
|
||||
await client.aclose()
|
||||
assert all(part.closed == 0 for part in injected.values())
|
||||
assert built["transport"].closed == 1
|
||||
|
||||
|
||||
def _ocr_client(**overrides):
|
||||
from polygateway.ocr import OcrClient
|
||||
|
||||
defaults = {
|
||||
"scope": "ocr",
|
||||
"sources": [_source(name="m1", provider="monkey", model="monkey-ocr")],
|
||||
"selector": RoundRobinSelector(),
|
||||
"limiter": _Closable(),
|
||||
"breaker": _Closable(),
|
||||
"transport": _Closable(),
|
||||
"retry": RetryPolicy(3, 2.0, 30.0),
|
||||
"backpressure": BackpressurePolicy(300.0, 0.01),
|
||||
}
|
||||
defaults.update(overrides)
|
||||
return OcrClient(**defaults)
|
||||
|
||||
|
||||
class TestOcrClientOwnership:
|
||||
async def test_injected_components_are_never_closed(self):
|
||||
injected = _parts("transport", "telemetry", "limiter", "breaker")
|
||||
client = _ocr_client(
|
||||
transport=injected["transport"],
|
||||
telemetry=injected["telemetry"],
|
||||
limiter=injected["limiter"],
|
||||
breaker=injected["breaker"],
|
||||
)
|
||||
await client.aclose()
|
||||
assert all(part.closed == 0 for part in injected.values())
|
||||
|
||||
async def test_factory_closes_every_component_it_built(self, monkeypatch):
|
||||
from polygateway.config import OcrSettings
|
||||
from polygateway.ocr import OcrClient
|
||||
|
||||
built = _parts("transport", "telemetry", "limiter", "breaker")
|
||||
_patch_builders(
|
||||
monkeypatch,
|
||||
built,
|
||||
transport_path="polygateway.transports.monkey_ocr.MonkeyOcrTransport",
|
||||
)
|
||||
client = OcrClient.from_settings(OcrSettings.from_env("OCR", env=dict(_OCR_ENV)))
|
||||
await client.aclose()
|
||||
assert all(part.closed == 1 for part in built.values())
|
||||
|
||||
async def test_factory_keeps_hands_off_injected_components(self, monkeypatch):
|
||||
from polygateway.config import OcrSettings
|
||||
from polygateway.ocr import OcrClient
|
||||
|
||||
built = _parts("transport", "telemetry", "limiter", "breaker")
|
||||
_patch_builders(
|
||||
monkeypatch,
|
||||
built,
|
||||
transport_path="polygateway.transports.monkey_ocr.MonkeyOcrTransport",
|
||||
)
|
||||
injected = _parts("telemetry", "limiter", "breaker")
|
||||
client = OcrClient.from_settings(
|
||||
OcrSettings.from_env("OCR", env=dict(_OCR_ENV)),
|
||||
limiter=injected["limiter"],
|
||||
breaker=injected["breaker"],
|
||||
telemetry=injected["telemetry"],
|
||||
)
|
||||
await client.aclose()
|
||||
assert all(part.closed == 0 for part in injected.values())
|
||||
assert built["transport"].closed == 1
|
||||
|
||||
|
||||
class TestRedisCacheOwnership:
|
||||
"""组件内部自建的连接归组件自己;照抄 RedisLimiter._owns_client 的正确先例。"""
|
||||
|
||||
async def test_injected_client_is_not_closed(self):
|
||||
from polygateway.backends.redis_cache import RedisCache
|
||||
|
||||
client = _Closable()
|
||||
await RedisCache(client).aclose()
|
||||
assert client.closed == 0
|
||||
|
||||
async def test_self_built_client_is_closed_once(self, monkeypatch):
|
||||
from types import SimpleNamespace
|
||||
|
||||
from polygateway.backends import redis_cache
|
||||
|
||||
built = _Closable()
|
||||
monkeypatch.setattr(
|
||||
redis_cache, "aioredis", SimpleNamespace(from_url=lambda url, **kwargs: built)
|
||||
)
|
||||
cache = redis_cache.RedisCache.from_url("redis://localhost:6379/0")
|
||||
await cache.aclose()
|
||||
await cache.aclose() # 幂等: 不重复关
|
||||
assert built.closed == 1
|
||||
|
||||
Reference in New Issue
Block a user