mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-05 14:48:40 +08:00
fix(cache): close shared redis pool once and keep redis caches across restarts
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
committed by
InfinityPacer
parent
0fae514cb1
commit
6fc2c535f7
10
app/adapters/cache/redis.py
vendored
10
app/adapters/cache/redis.py
vendored
@@ -781,8 +781,11 @@ class RedisBackend(AtomicCacheBackend):
|
||||
return self.redis_helper.items(region=region)
|
||||
|
||||
def close(self) -> None:
|
||||
"""关闭同步 Redis 连接池。"""
|
||||
self.redis_helper.close()
|
||||
"""释放当前缓存实例;共享的 Redis 连接池不在此关闭。
|
||||
|
||||
所有 Redis 缓存共用同一个 ``RedisHelper`` 连接池,由关闭流程的「Redis缓存连接」步骤统一关闭。
|
||||
单个缓存在这里关池会断开其他仍在使用的缓存,随后又被自动重连。
|
||||
"""
|
||||
|
||||
@staticmethod
|
||||
def is_redis() -> bool:
|
||||
@@ -850,8 +853,7 @@ class AsyncRedisBackend(AsyncCacheBackend):
|
||||
yield item
|
||||
|
||||
async def close(self) -> None:
|
||||
"""关闭异步 Redis 连接池。"""
|
||||
await self.redis_helper.close()
|
||||
"""释放当前异步缓存实例;共享的异步 Redis 连接池由关闭流程的「异步Redis缓存连接」步骤统一关闭。"""
|
||||
|
||||
@staticmethod
|
||||
def is_redis() -> bool:
|
||||
|
||||
@@ -185,6 +185,9 @@ def reset_wallpaper_providers() -> None:
|
||||
class WallpaperHelper(metaclass=Singleton):
|
||||
"""
|
||||
壁纸帮助类
|
||||
|
||||
壁纸缓存键不区分来源,启动装配与生命周期撤销时需要清空;每个方法只缓存一条地址,
|
||||
因此固定使用进程内缓存(local_only),清空不触及 Redis,也不跨进程沿用旧来源的结果。
|
||||
"""
|
||||
|
||||
def clear_cache(self) -> None:
|
||||
@@ -234,21 +237,21 @@ class WallpaperHelper(metaclass=Singleton):
|
||||
wallpaper = self.get_static_wallpaper()
|
||||
return [wallpaper] if wallpaper else []
|
||||
|
||||
@cached(maxsize=1, ttl=3600)
|
||||
@cached(maxsize=1, ttl=3600, local_only=True)
|
||||
def get_tmdb_wallpaper(self) -> Optional[str]:
|
||||
"""
|
||||
获取TMDB每日壁纸
|
||||
"""
|
||||
return _tmdb_wallpaper_provider()
|
||||
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True)
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True, local_only=True)
|
||||
def get_tmdb_wallpapers(self, num: int = 10) -> List[str]:
|
||||
"""
|
||||
获取7天的TMDB每日壁纸
|
||||
"""
|
||||
return _tmdb_wallpaper_list_provider(num)
|
||||
|
||||
@cached(maxsize=1, ttl=3600)
|
||||
@cached(maxsize=1, ttl=3600, local_only=True)
|
||||
def get_bing_wallpaper(self) -> Optional[str]:
|
||||
"""
|
||||
获取Bing每日壁纸
|
||||
@@ -266,7 +269,7 @@ class WallpaperHelper(metaclass=Singleton):
|
||||
print(str(err))
|
||||
return None
|
||||
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True)
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True, local_only=True)
|
||||
def get_bing_wallpapers(self, num: int = 7) -> List[str]:
|
||||
"""
|
||||
获取7天的Bing每日壁纸
|
||||
@@ -283,21 +286,21 @@ class WallpaperHelper(metaclass=Singleton):
|
||||
print(str(err))
|
||||
return []
|
||||
|
||||
@cached(maxsize=1, ttl=3600)
|
||||
@cached(maxsize=1, ttl=3600, local_only=True)
|
||||
def get_mediaserver_wallpaper(self) -> Optional[str]:
|
||||
"""
|
||||
获取媒体服务器壁纸
|
||||
"""
|
||||
return _mediaserver_wallpaper_provider()
|
||||
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True)
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True, local_only=True)
|
||||
def get_mediaserver_wallpapers(self, num: int = 10) -> List[str]:
|
||||
"""
|
||||
获取媒体服务器壁纸列表
|
||||
"""
|
||||
return _mediaserver_wallpaper_list_provider(num)
|
||||
|
||||
@cached(maxsize=1, ttl=3600)
|
||||
@cached(maxsize=1, ttl=3600, local_only=True)
|
||||
def get_customize_wallpaper(self) -> Optional[str]:
|
||||
"""
|
||||
获取自定义壁纸api壁纸
|
||||
@@ -307,7 +310,7 @@ class WallpaperHelper(metaclass=Singleton):
|
||||
return wallpaper_list[0]
|
||||
return None
|
||||
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True)
|
||||
@cached(maxsize=1, ttl=3600, skip_empty=True, local_only=True)
|
||||
def get_customize_wallpapers(self) -> List[str]:
|
||||
"""
|
||||
获取自定义壁纸api壁纸
|
||||
|
||||
@@ -91,11 +91,10 @@ class BangumiModule(MediaAuxiliaryProviderMixin, _ModuleBase):
|
||||
"""
|
||||
关闭Bangumi客户端
|
||||
"""
|
||||
# 请求缓存键已包含 API 地址,切换代理或域名后自然使用新键;这里不清缓存,
|
||||
# 以免 Redis 缓存在每次重载和关闭时都被清空。
|
||||
if self.bangumiapi:
|
||||
try:
|
||||
self.bangumiapi.clear_cache()
|
||||
finally:
|
||||
self.bangumiapi.close()
|
||||
self.bangumiapi.close()
|
||||
|
||||
def test(self) -> Tuple[bool, str]:
|
||||
"""
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import asyncio
|
||||
import hashlib
|
||||
from pathlib import Path
|
||||
from typing import Optional, Tuple, Union
|
||||
|
||||
@@ -307,8 +308,10 @@ class FanartModule(_ModuleBase):
|
||||
"""
|
||||
|
||||
def init_module(self) -> None:
|
||||
"""初始化 Fanart 模块并清理旧配置 generation 的请求缓存。"""
|
||||
self.clear_cache()
|
||||
"""初始化 Fanart 模块。
|
||||
|
||||
请求缓存键包含 API Key 指纹,更换 Key 后自然使用新键,无需在每次启动或重载时清空缓存。
|
||||
"""
|
||||
|
||||
def stop(self):
|
||||
"""停止 Fanart 模块;请求资源由每次调用按当前配置创建。"""
|
||||
@@ -416,7 +419,7 @@ class FanartModule(_ModuleBase):
|
||||
query = self.__fanart_query(mediainfo=mediainfo)
|
||||
if not query:
|
||||
return None
|
||||
result = self.__request_fanart(*query)
|
||||
result = self.__request_fanart(*query, _key_token=self.__key_token())
|
||||
return self.__extract_images(mediainfo=mediainfo, result=result, season=season)
|
||||
|
||||
async def __async_obtain_fanart_images(
|
||||
@@ -430,7 +433,7 @@ class FanartModule(_ModuleBase):
|
||||
query = self.__fanart_query(mediainfo=mediainfo)
|
||||
if not query:
|
||||
return None
|
||||
result = await self.__async_request_fanart(*query)
|
||||
result = await self.__async_request_fanart(*query, _key_token=self.__key_token())
|
||||
return self.__extract_images(mediainfo=mediainfo, result=result, season=season)
|
||||
|
||||
@staticmethod
|
||||
@@ -573,7 +576,7 @@ class FanartModule(_ModuleBase):
|
||||
@classmethod
|
||||
@cached(maxsize=get_runtime_setting('CONF').fanart, ttl=get_runtime_setting('CONF').meta, shared_key="get")
|
||||
def __request_fanart(
|
||||
cls, media_type: MediaType, queryid: Union[str, int]
|
||||
cls, media_type: MediaType, queryid: Union[str, int], _key_token: str = ""
|
||||
) -> Optional[dict]:
|
||||
image_url = cls.__fanart_url(media_type=media_type, queryid=queryid)
|
||||
try:
|
||||
@@ -592,7 +595,7 @@ class FanartModule(_ModuleBase):
|
||||
@classmethod
|
||||
@cached(maxsize=get_runtime_setting('CONF').fanart, ttl=get_runtime_setting('CONF').meta, shared_key="get")
|
||||
async def __async_request_fanart(
|
||||
cls, media_type: MediaType, queryid: Union[str, int]
|
||||
cls, media_type: MediaType, queryid: Union[str, int], _key_token: str = ""
|
||||
) -> Optional[dict]:
|
||||
image_url = cls.__fanart_url(media_type=media_type, queryid=queryid)
|
||||
try:
|
||||
@@ -609,6 +612,15 @@ class FanartModule(_ModuleBase):
|
||||
logger.error(f"获取{queryid}的Fanart图片失败:{str(err)}")
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def __key_token() -> str:
|
||||
"""返回当前 API Key 的指纹,只用于区分缓存键,不把 Key 原文写进缓存。
|
||||
|
||||
Key 无效时 Fanart 返回空图集并被长期缓存;换 Key 后指纹变化,旧结果不再命中。
|
||||
"""
|
||||
api_key = str(get_runtime_setting("FANART_API_KEY") or "")
|
||||
return hashlib.sha256(api_key.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
@classmethod
|
||||
def __fanart_url(cls, media_type: MediaType, queryid: Union[str, int]) -> str:
|
||||
"""
|
||||
|
||||
@@ -585,15 +585,6 @@ async def stop_modules() -> bool:
|
||||
lambda: _call_existing_singleton(ThreadHelper, "shutdown"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"Redis缓存连接",
|
||||
lambda: _call_existing_singleton(get_cache_composition().sync_redis_owner, "close"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"异步Redis缓存连接",
|
||||
lambda: _call_existing_singleton(get_cache_composition().async_redis_owner, "close"),
|
||||
)
|
||||
# Web Agent 的取消 finally 可能还要写入最终展示快照,必须先完成任务收尾,再关闭写入准入。
|
||||
web_agent_drained = await run_step(
|
||||
"Web Agent后台任务",
|
||||
@@ -635,6 +626,17 @@ async def stop_modules() -> bool:
|
||||
logger.error(
|
||||
"数据库任务未收敛,保留全部 Provider 和数据库连接以供诊断与重试"
|
||||
)
|
||||
# Redis 连接池由全部缓存共享,前面的 Agent 收尾、数据库任务和 Provider 撤销都可能读写缓存;
|
||||
# 放在最后关闭,避免关闭后又被自动重连且再无步骤收口。
|
||||
await run_step(
|
||||
"Redis缓存连接",
|
||||
lambda: _call_existing_singleton(get_cache_composition().sync_redis_owner, "close"),
|
||||
offload=True,
|
||||
)
|
||||
await run_step(
|
||||
"异步Redis缓存连接",
|
||||
lambda: _call_existing_singleton(get_cache_composition().async_redis_owner, "close"),
|
||||
)
|
||||
return all_converged
|
||||
|
||||
|
||||
|
||||
@@ -99,3 +99,14 @@ def test_bangumi_test_uses_generation_snapshot_until_reload(monkeypatch):
|
||||
assert request_utils.return_value.get_res.call_args.args[0] == (
|
||||
"https://new-bangumi.example/api/"
|
||||
)
|
||||
|
||||
|
||||
def test_bangumi_stop_closes_client_without_clearing_cache() -> None:
|
||||
"""模块停止只关闭客户端;缓存键已含 API 地址,不在重载或关闭时清空缓存。"""
|
||||
module = BangumiModule.__new__(BangumiModule)
|
||||
module.bangumiapi = MagicMock()
|
||||
|
||||
module.stop()
|
||||
|
||||
module.bangumiapi.close.assert_called_once_with()
|
||||
module.bangumiapi.clear_cache.assert_not_called()
|
||||
|
||||
@@ -3,7 +3,7 @@ import os
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from unittest.mock import AsyncMock
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -581,6 +581,20 @@ def test_memory_backend_uses_zero_default_ttl():
|
||||
assert cache.get("key", region="zero_default_ttl") is None
|
||||
|
||||
|
||||
def test_redis_backend_close_keeps_shared_pool(monkeypatch):
|
||||
"""关闭单个 Redis 缓存不得关闭全部缓存共享的连接池,连接池由关闭流程统一收口。"""
|
||||
helper = MagicMock()
|
||||
async_helper = MagicMock()
|
||||
async_helper.close = AsyncMock()
|
||||
monkeypatch.setattr("app.adapters.cache.redis.RedisHelper", lambda: helper)
|
||||
monkeypatch.setattr("app.adapters.cache.redis.AsyncRedisHelper", lambda: async_helper)
|
||||
|
||||
RedisBackend().close()
|
||||
asyncio.run(AsyncRedisBackend().close())
|
||||
|
||||
helper.close.assert_not_called()
|
||||
async_helper.close.assert_not_awaited()
|
||||
|
||||
def test_redis_backend_treats_zero_ttl_as_expired():
|
||||
"""
|
||||
Redis backend 应删除 ttl=0 的同名 key,避免向 Redis 发送无效 EX 0。
|
||||
|
||||
@@ -52,3 +52,55 @@ def test_fanart_clear_cache_registers_async_cleanup_in_running_loop() -> None:
|
||||
assert registry.records == ()
|
||||
|
||||
asyncio.run(scenario())
|
||||
|
||||
|
||||
class _FakeFanartResponse:
|
||||
"""模拟成功的 Fanart HTTP 响应。"""
|
||||
|
||||
def json(self) -> dict:
|
||||
"""返回最小图集载荷。"""
|
||||
return {"name": "fanart"}
|
||||
|
||||
|
||||
def test_fanart_cache_key_follows_api_key(monkeypatch) -> None:
|
||||
"""同一媒体在 API Key 不变时命中缓存,换 Key 后重新请求而不依赖清空缓存。"""
|
||||
from app.domain.context import MediaInfo
|
||||
from app.runtime.config import settings
|
||||
from app.schemas.types import MediaType
|
||||
|
||||
requested_urls: list[str] = []
|
||||
|
||||
class _FakeRequestUtils:
|
||||
"""记录请求地址的同步 HTTP 替身。"""
|
||||
|
||||
def __init__(self, **_kwargs) -> None:
|
||||
pass
|
||||
|
||||
def get_res(self, url: str, **_kwargs) -> _FakeFanartResponse:
|
||||
requested_urls.append(url)
|
||||
return _FakeFanartResponse()
|
||||
|
||||
monkeypatch.setattr("app.modules.fanart.RequestUtils", _FakeRequestUtils)
|
||||
monkeypatch.setattr(settings, "FANART_ENABLE", True)
|
||||
monkeypatch.setattr(settings, "FANART_API_KEY", "key-a")
|
||||
module = FanartModule()
|
||||
# 使用测试专属 ID,避免与其他用例共享进程内缓存条目。
|
||||
mediainfo = MediaInfo(type=MediaType.MOVIE, tmdb_id=987654321, title="Fanart Cache Key")
|
||||
|
||||
module._FanartModule__obtain_fanart_images(mediainfo=mediainfo)
|
||||
module._FanartModule__obtain_fanart_images(mediainfo=mediainfo)
|
||||
monkeypatch.setattr(settings, "FANART_API_KEY", "key-b")
|
||||
module._FanartModule__obtain_fanart_images(mediainfo=mediainfo)
|
||||
|
||||
assert len(requested_urls) == 2
|
||||
assert requested_urls[0].endswith("api_key=key-a")
|
||||
assert requested_urls[1].endswith("api_key=key-b")
|
||||
|
||||
|
||||
def test_fanart_init_module_keeps_request_cache() -> None:
|
||||
"""启动和重载不再清空请求缓存,Redis 缓存得以跨重启保留。"""
|
||||
module = FanartModule()
|
||||
with patch.object(FanartModule, "clear_cache") as clear_cache:
|
||||
module.init_module()
|
||||
|
||||
clear_cache.assert_not_called()
|
||||
|
||||
@@ -1318,6 +1318,27 @@ def test_stop_modules_drains_web_agent_tasks_before_persistence(monkeypatch):
|
||||
]
|
||||
|
||||
|
||||
def test_stop_modules_closes_shared_redis_pool_last(monkeypatch):
|
||||
"""Redis 连接池必须在 Provider 撤销等仍会读写缓存的步骤之后关闭,避免关闭后被重连。"""
|
||||
order = []
|
||||
dependencies = _patch_module_shutdown_dependencies(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
modules_initializer,
|
||||
"get_configured_agent_chat_persistence",
|
||||
MagicMock(return_value=None),
|
||||
)
|
||||
monkeypatch.setattr(modules_initializer, "database_runtime_active", lambda: False)
|
||||
dependencies["reset_module_providers"].side_effect = lambda: order.append(
|
||||
"providers"
|
||||
) or True
|
||||
dependencies["close_database"].side_effect = lambda: order.append("connection")
|
||||
dependencies["redis"].side_effect = lambda: order.append("redis")
|
||||
dependencies["async_redis"].side_effect = lambda: order.append("async-redis")
|
||||
|
||||
asyncio.run(modules_initializer.stop_modules())
|
||||
|
||||
assert order == ["providers", "connection", "redis", "async-redis"]
|
||||
|
||||
def test_stop_modules_retains_providers_when_database_worker_remains_active(
|
||||
monkeypatch,
|
||||
) -> None:
|
||||
|
||||
@@ -197,3 +197,44 @@ def test_image_legacy_helpers_keep_canonical_class_identity():
|
||||
|
||||
assert legacy_image.ImageHelper is ImageHelper
|
||||
assert legacy_wallpaper.WallpaperHelper is WallpaperHelper
|
||||
|
||||
|
||||
def test_wallpaper_caches_stay_in_process_under_redis_backend() -> None:
|
||||
"""Redis 缓存模式下壁纸缓存仍留在进程内,生命周期清空不触及 Redis。"""
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
script = """
|
||||
import json
|
||||
from app.testing.bootstrap import ensure_sites_stub, isolate_config_dir
|
||||
isolate_config_dir()
|
||||
ensure_sites_stub()
|
||||
from app.startup.initializers.cache import configure_cache_dependencies
|
||||
configure_cache_dependencies()
|
||||
from app.application.image import WallpaperHelper
|
||||
|
||||
backends = {}
|
||||
for name in ("get_tmdb_wallpaper", "get_bing_wallpapers", "get_customize_wallpapers"):
|
||||
wrapper = getattr(WallpaperHelper, name)
|
||||
cells = [cell.cell_contents for cell in wrapper.__closure__ or ()]
|
||||
backend = next(value for value in cells if hasattr(value, "is_redis"))
|
||||
backends[name] = backend.is_redis()
|
||||
print(json.dumps(backends))
|
||||
"""
|
||||
environment = {**os.environ, "CACHE_BACKEND_TYPE": "redis"}
|
||||
completed = subprocess.run(
|
||||
[sys.executable, "-c", script],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=environment,
|
||||
)
|
||||
|
||||
result = json.loads(completed.stdout.strip().splitlines()[-1])
|
||||
assert result == {
|
||||
"get_tmdb_wallpaper": False,
|
||||
"get_bing_wallpapers": False,
|
||||
"get_customize_wallpapers": False,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user