fix(cache): isolate async Redis clients per event loop

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:

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'] = {}

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 连接池配置变化应触发客户端重建。

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"},
+    }