From ecbdcd572606644e65ca1112203de8850987b1d1 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 19 Sep 2026 18:25:43 +0800 Subject: [PATCH] fix(cache): isolate async Redis clients per event loop --- app/adapters/cache/redis.py | 149 +++++++++++++++++++++---------- app/runtime/progress.py | 8 +- tests/test_cache_system.py | 67 ++++++++++++++ tests/test_scheduler_progress.py | 40 +++++++++ 4 files changed, 217 insertions(+), 47 deletions(-) diff --git a/app/adapters/cache/redis.py b/app/adapters/cache/redis.py index 2166fd130..006bcfa2e 100644 --- a/app/adapters/cache/redis.py +++ b/app/adapters/cache/redis.py @@ -2,6 +2,7 @@ import asyncio import json import pickle import threading +import weakref from typing import Any, AsyncGenerator, Generator, Optional, Tuple, Union from urllib.parse import quote, unquote @@ -381,19 +382,41 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): self.redis_url = get_runtime_setting('CACHE_BACKEND_URL') self.client: Optional[Redis] = None self._loop: Optional[asyncio.AbstractEventLoop] = None - self._connect_lock: Optional[asyncio.Lock] = None - self._connect_lock_loop: Optional[asyncio.AbstractEventLoop] = None + self._clients: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary() + self._connect_locks: weakref.WeakKeyDictionary = weakref.WeakKeyDictionary() + self._state_lock = threading.RLock() def _get_connect_lock(self, current_loop: asyncio.AbstractEventLoop) -> asyncio.Lock: """ 获取当前事件循环对应的异步连接锁 """ - if self._connect_lock is None or self._connect_lock_loop is not current_loop: - self._connect_lock = asyncio.Lock() - self._connect_lock_loop = current_loop - return self._connect_lock + with self._state_lock: + connect_lock = self._connect_locks.get(current_loop) + if connect_lock is None: + connect_lock = asyncio.Lock() + self._connect_locks[current_loop] = connect_lock + return connect_lock - async def _connect(self): + async def _close_client_instance(self, client: Redis) -> None: + """在客户端所属事件循环中关闭连接池。""" + try: + close = getattr(client, "aclose", None) or getattr(client, "close") + await close() + except Exception: + pass + await self._close_client_pool(client) + + async def _close_client_pool(self, client: Redis) -> None: + """关闭异步 Redis 客户端显式持有的连接池。""" + pool = getattr(client, "connection_pool", None) + close_pool = getattr(pool, "aclose", None) + if close_pool is not None: + try: + await close_pool() + except Exception: + pass + + async def _connect(self) -> Redis: """ 建立异步Redis连接 """ @@ -402,12 +425,12 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): client = None try: async with connect_lock: - # 检测事件循环是否发生变化，如果变化则重新连接 - if self.client is not None and self._loop is not current_loop: - logger.debug("Event loop changed, reconnecting Redis (async)") - await self._close_client() - if self.client is not None: - return + with self._state_lock: + client = self._clients.get(current_loop) + if client is not None: + self.client = client + self._loop = current_loop + return client self.redis_url = get_runtime_setting('CACHE_BACKEND_URL') connection_pool = AsyncBlockingConnectionPool.from_url( self.redis_url, @@ -419,31 +442,51 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): timeout=get_runtime_setting('CACHE_REDIS_POOL_TIMEOUT'), ) client = Redis(connection_pool=connection_pool) - self._loop = current_loop # 测试连接，确保Redis可用 await client.ping() + with self._state_lock: + self._clients[current_loop] = client self.client = client + self._loop = current_loop logger.info(f"Successfully connected to Redis (async)：{self.redis_url}") - await self.set_memory_limit() + await self.set_memory_limit(client=client) + return client except Exception as e: if client: - await client.close() + await self._close_client_instance(client) + with self._state_lock: + if self._clients.get(current_loop) is client: + self._clients.pop(current_loop, None) + if self.client is client: + self.client = None + self._loop = None logger.error(f"Failed to connect to Redis (async): {e}") - self.client = None - self._loop = None raise RuntimeError("Redis async connection failed") from e async def _close_client(self): """ - 关闭当前Redis客户端连接 + 关闭当前事件循环拥有的Redis客户端连接 """ - if self.client: - try: - await self.client.close() - except Exception: - pass - self.client = None - self._loop = None + current_loop = asyncio.get_running_loop() + connect_lock = self._get_connect_lock(current_loop) + async with connect_lock: + with self._state_lock: + client = self._clients.pop(current_loop, None) + client_is_current = self.client is client + if client is not None: + try: + if client_is_current and self.client is client: + await self.client.close() + else: + close = getattr(client, "aclose", None) or getattr(client, "close") + await close() + except Exception: + pass + await self._close_client_pool(client) + with self._state_lock: + if self.client is client: + self.client = None + self._loop = None async def on_config_changed(self): """缓存配置变化后异步重建 Redis 连接。""" @@ -455,19 +498,27 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): """返回异步 Redis 配置重载名称。""" return "Redis (async)" - async def set_memory_limit(self, policy: Optional[str] = "allkeys-lru"): + async def set_memory_limit( + self, + policy: Optional[str] = "allkeys-lru", + client: Optional[Redis] = None, + ): """ 动态设置Redis最大内存和内存淘汰策略 :param policy: 淘汰策略（如'allkeys-lru'） + :param client: 已建立的异步 Redis 客户端 """ + if client is None: + client = await self._connect() try: # 如果有显式值，则直接使用，为0时说明不限制，如果未配置，开启BIG_MEMORY_MODE时为"1024mb"，未开启时为"256mb" maxmemory = get_runtime_setting('CACHE_REDIS_MAXMEMORY') or ( "1024mb" if get_runtime_setting('BIG_MEMORY_MODE') else "256mb" ) - await self.client.config_set("maxmemory", maxmemory) - await self.client.config_set("maxmemory-policy", policy) + target_client = self.client if self.client is client else client + await target_client.config_set("maxmemory", maxmemory) + await target_client.config_set("maxmemory-policy", policy) logger.debug(f"Redis maxmemory set to {maxmemory}, policy: {policy} (async)") except Exception as e: logger.error(f"Failed to set Redis maxmemory or policy (async): {e}") @@ -513,12 +564,13 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): :param kwargs: 其他参数 """ try: - await self._connect() + current_client = await self._connect() + client = self.client if self.client is current_client else current_client redis_key = self.__make_redis_key(region, key) # 对值进行序列化 serialized_value = serialize(value) kwargs.pop("maxsize", None) - await self.client.set(redis_key, serialized_value, ex=ttl, **kwargs) + await client.set(redis_key, serialized_value, ex=ttl, **kwargs) except Exception as e: logger.error(f"Failed to set key (async): {key} in region: {region}, error: {e}") @@ -531,9 +583,10 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): :return: 存在返回True，否则返回False """ try: - await self._connect() + current_client = await self._connect() + client = self.client if self.client is current_client else current_client redis_key = self.__make_redis_key(region, key) - result = await self.client.exists(redis_key) + result = await client.exists(redis_key) return result == 1 except Exception as e: logger.error(f"Failed to exists key (async): {key} region: {region}, error: {e}") @@ -548,9 +601,10 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): :return: 返回缓存的值，如果缓存不存在返回None """ try: - await self._connect() + current_client = await self._connect() + client = self.client if self.client is current_client else current_client redis_key = self.__make_redis_key(region, key) - value = await self.client.get(redis_key) + value = await client.get(redis_key) if value is not None: return deserialize(value) return None @@ -566,9 +620,10 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): :param region: 缓存的区 """ try: - await self._connect() + current_client = await self._connect() + client = self.client if self.client is current_client else current_client redis_key = self.__make_redis_key(region, key) - await self.client.delete(redis_key) + await client.delete(redis_key) except Exception as e: logger.error(f"Failed to delete key (async): {key} in region: {region}, error: {e}") @@ -579,17 +634,18 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): :param region: 缓存的区 """ try: - await self._connect() + current_client = await self._connect() + client = self.client if self.client is current_client else current_client if region: cache_region = self.__get_region(region) redis_key = f"{cache_region}:key:*" - async with self.client.pipeline() as pipe: - async for key in self.client.scan_iter(redis_key): + async with client.pipeline() as pipe: + async for key in client.scan_iter(redis_key): await pipe.delete(key) await pipe.execute() logger.debug(f"Cleared Redis cache for region (async): {region}") else: - await self.client.flushdb() + await client.flushdb() logger.info("Cleared all Redis cache (async)") except Exception as e: logger.error(f"Failed to clear cache (async), region: {region}, error: {e}") @@ -602,17 +658,18 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): :return: 返回键值对生成器 """ try: - await self._connect() + current_client = await self._connect() + client = self.client if self.client is current_client else current_client if region: cache_region = self.__get_region(region) redis_key = f"{cache_region}:key:*" - async for key in self.client.scan_iter(redis_key): - value = await self.client.get(key) + async for key in client.scan_iter(redis_key): + value = await client.get(key) if value is not None: yield self.__get_original_key(key), deserialize(value) else: - async for key in self.client.scan_iter("*"): - value = await self.client.get(key) + async for key in client.scan_iter("*"): + value = await client.get(key) if value is not None: yield self.__get_original_key(key), deserialize(value) except Exception as e: diff --git a/app/runtime/progress.py b/app/runtime/progress.py index b256c4a32..1950865d1 100644 --- a/app/runtime/progress.py +++ b/app/runtime/progress.py @@ -166,7 +166,13 @@ class AsyncProgressHelper: """ current = await self._progress.get(self._key) if not current: - return + # 读取快照失败时仍需落地终态，避免 Redis 短暂故障让任务永久显示运行中。 + current = { + "enable": False, + "value": 0, + "text": "", + "data": {}, + } if data is not None: if not current.get('data'): current['data'] = {} diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 34e5e6761..e1cb36ca4 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -14,6 +14,7 @@ from app.adapters.cache.backends import ( RedisBackend, ) from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper, serialize +from app.foundation.singleton import Singleton from app.runtime.cache import ( AsyncFileCache, AsyncMemoryBackend, @@ -900,6 +901,72 @@ def test_async_redis_helper_uses_blocking_pool_settings(monkeypatch): assert ("maxmemory-policy", "allkeys-lru") in config_calls +def test_async_redis_helper_does_not_close_client_from_foreign_loop(monkeypatch): + """ + 异步 Redis 客户端按事件循环隔离，切换循环时不得跨循环关闭旧连接池。 + """ + singleton_key = (AsyncRedisHelper, (), frozenset()) + monkeypatch.delitem(Singleton._instances, singleton_key, raising=False) + clients = [] + + class FakePool: + """记录连接池所属事件循环。""" + + def __init__(self): + self.loop = asyncio.get_running_loop() + self.wrong_loop_close = False + + async def aclose(self): + """模拟连接池关闭并校验事件循环归属。""" + if asyncio.get_running_loop() is not self.loop: + self.wrong_loop_close = True + raise RuntimeError("pool belongs to a different event loop") + + class FakeClient: + """模拟带事件循环归属的异步 Redis 客户端。""" + + def __init__(self, connection_pool): + self.connection_pool = connection_pool + self.loop = asyncio.get_running_loop() + self.wrong_loop_close = False + clients.append(self) + + async def ping(self): + """模拟 Redis ping。""" + + async def config_set(self, _key, _value): + """模拟 Redis 配置写入。""" + + async def close(self): + """模拟客户端关闭并校验事件循环归属。""" + if asyncio.get_running_loop() is not self.loop: + self.wrong_loop_close = True + raise RuntimeError("client belongs to a different event loop") + + def fake_from_url(_url, **_kwargs): + """为当前事件循环创建独立连接池。""" + return FakePool() + + monkeypatch.setattr( + "app.adapters.cache.redis.AsyncBlockingConnectionPool.from_url", + fake_from_url, + ) + monkeypatch.setattr("app.adapters.cache.redis.Redis", FakeClient) + + async def connect(): + """在当前事件循环建立异步 Redis 客户端。""" + return await AsyncRedisHelper()._connect() + + first = asyncio.run(connect()) + second = asyncio.run(connect()) + + assert first is not second + assert len(clients) == 2 + assert not first.wrong_loop_close + assert not first.connection_pool.wrong_loop_close + assert second.loop is not first.loop + + def test_redis_helpers_watch_pool_settings(): """ Redis 连接池配置变化应触发客户端重建。 diff --git a/tests/test_scheduler_progress.py b/tests/test_scheduler_progress.py index a9d015f76..6c74911e4 100644 --- a/tests/test_scheduler_progress.py +++ b/tests/test_scheduler_progress.py @@ -6,6 +6,7 @@ from uuid import uuid4 import pytest from app.runtime.loop import main_loop_registry +from app.runtime.progress import AsyncProgressHelper from app.scheduler.facade import Scheduler from app.scheduler.registry import ExecutionRegistry @@ -235,3 +236,42 @@ def test_scheduler_returns_none_for_unknown_job(): scheduler._registry = ExecutionRegistry(scheduler._lock) assert scheduler.get_progress(job_id) is None + + +def test_async_progress_end_rebuilds_terminal_snapshot_after_read_failure(): + """进度快照读取失败时，终态写入仍应生成可展示的完成状态。""" + + class BrokenProgress: + """模拟读取失败但仍可写入的进度后端。""" + + def __init__(self): + self.value = None + + async def get(self, _key): + """模拟 Redis 读取失败后的空结果。""" + return None + + async def set(self, _key, value): + """记录进度终态写入。""" + self.value = value + + async def run_end(): + """执行一次终态写入。""" + progress = AsyncProgressHelper("scheduler:read-failure") + backend = BrokenProgress() + progress._progress = backend + await progress.end( + text="任务执行完成", + data={"status": "success"}, + value=100, + ) + return backend.value + + detail = asyncio.run(run_end()) + + assert detail == { + "enable": False, + "value": 100.0, + "text": "任务执行完成", + "data": {"status": "success"}, + }