fix(modules): serialize configuration reload lifecycle (#6122)

app/core/module.py

@@ -58,7 +58,6 @@ class ModuleManager(metaclass=Singleton):
         """
         logger.info("正在停止所有模块...")
         for module_id, module in self._running_modules.items():
-            if hasattr(module, "stop"):
             try:
                 module.stop()
                 logger.debug(f"Moudle Stoped：{module_id}")

app/modules/__init__.py

@@ -1,8 +1,10 @@
+import threading
 from abc import abstractmethod, ABCMeta
 from typing import Generic, Tuple, Union, TypeVar, Type, Dict, Optional, Callable
 from pathlib import Path
 
 from app.helper.service import ServiceConfigHelper
+from app.log import logger
 from app.schemas import Notification, NotificationConf, MediaServerConf, DownloaderConf
 from app.schemas.types import ModuleType, DownloaderType, MediaServerType, MessageChannel, StorageSchema, \
     OtherModulesType, SystemConfigKey
@@ -15,7 +17,20 @@ class _ModuleBase(ConfigReloadMixin, metaclass=ABCMeta):
     输入参数与输出参数一致的，或没有输出的，可以被多个模块重复实现
     """
 
-    def on_config_changed(self):
+    def __init__(self) -> None:
+        """初始化模块生命周期锁"""
+        super().__init__()
+        self._reload_lock = threading.RLock()
+
+    def on_config_changed(self) -> None:
+        """串行停止旧资源并按最新配置重新初始化模块"""
+        with self._reload_lock:
+            try:
+                self.stop()
+            except Exception as err:
+                logger.error(
+                    f"停止 {self.get_reload_name()} 旧资源失败，继续按最新配置初始化：{err}"
+                )
             self.init_module()
 
     def get_reload_name(self):

app/modules/discord/__init__.py

@@ -58,7 +58,6 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]):
         if not Discord:
             logger.error("Discord 依赖未就绪（需要安装 discord.py==2.6.4），模块未启动")
             return
-        self.stop()
         super().init_service(
             service_name=Discord.__name__.lower(), service_type=Discord
         )
@@ -89,12 +88,13 @@ class DiscordModule(_ModuleBase, _MessageBase[Discord]):
         """
         return 4
 
-    def stop(self):
-        """
-        停止模块
-        """
+    def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
+            try:
                 client.stop()
+            except Exception as err:
+                logger.error(f"停止Discord模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         """

app/modules/feishu/__init__.py

@@ -10,7 +10,6 @@ from app.schemas.types import ModuleType
 
 class FeishuModule(_ModuleBase, _MessageBase[Feishu]):
     def init_module(self) -> None:
-        self.stop()
         super().init_service(service_name=Feishu.__name__.lower(), service_type=Feishu)
         self._channel = MessageChannel.Feishu
 
@@ -30,9 +29,9 @@ class FeishuModule(_ModuleBase, _MessageBase[Feishu]):
     def get_priority() -> int:
         return 2
 
-    def stop(self):
+    def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
-            if hasattr(client, "stop"):
             try:
                 client.stop()
             except Exception as err:

app/modules/filter/__init__.py

@@ -92,13 +92,6 @@ class FilterModule(_ModuleBase):
         self.rule_set = deepcopy(self.builtin_rule_set)
         self.__init_custom_rules()
 
-    def on_config_changed(self) -> None:
-        """
-        自定义过滤或 Meta 识别配置变更后重建规则集并刷新 Rust Meta 配置缓存。
-        """
-        clear_rust_parse_options_cache()
-        self.init_module()
-
     def __init_custom_rules(self):
         """
         加载用户自定义规则，如跟内置规则冲突，以用户自定义规则为准
@@ -137,10 +130,8 @@ class FilterModule(_ModuleBase):
         return 4
 
     def stop(self) -> None:
-        """
-        停止过滤器模块。
-        """
-        pass
+        """停止模块"""
+        clear_rust_parse_options_cache()
 
     def test(self) -> None:
         """

app/modules/plex/__init__.py

@@ -44,13 +44,14 @@ class PlexModule(_ModuleBase, _MediaServerBase[Plex]):
         """
         return 3
 
-    def stop(self):
-        """
-        停止模块服务
-        """
+    def stop(self) -> None:
+        """停止模块"""
         for server in self.get_instances().values():
+            try:
                 if server:
                     server.close()
+            except Exception as err:
+                logger.error(f"停止Plex模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         """

app/modules/qqbot/__init__.py

@@ -46,7 +46,6 @@ class QQBotModule(_ModuleBase, _MessageBase[QQBot]):
     )
 
     def init_module(self) -> None:
-        self.stop()
         super().init_service(service_name=QQBot.__name__.lower(), service_type=QQBot)
         self._channel = MessageChannel.QQ
 
@@ -67,9 +66,12 @@ class QQBotModule(_ModuleBase, _MessageBase[QQBot]):
         return 10
 
     def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
-            if hasattr(client, "stop"):
+            try:
                 client.stop()
+            except Exception as err:
+                logger.error(f"停止QQ Bot模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         if not self.get_instances():

app/modules/slack/__init__.py

@@ -69,12 +69,13 @@ class SlackModule(_ModuleBase, _MessageBase[Slack]):
         """
         return 3
 
-    def stop(self):
-        """
-        停止模块
-        """
+    def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
+            try:
                 client.stop()
+            except Exception as err:
+                logger.error(f"停止Slack模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         """

app/modules/telegram/__init__.py

@@ -62,12 +62,13 @@ class TelegramModule(_ModuleBase, _MessageBase[Telegram]):
         """
         return 0
 
-    def stop(self):
-        """
-        停止模块
-        """
+    def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
+            try:
                 client.stop()
+            except Exception as err:
+                logger.error(f"停止Telegram模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         """

app/modules/telegram/telegram.py

@@ -1536,14 +1536,19 @@ class Telegram:
         # 清理菜单命令
         self._bot.delete_my_commands()
 
-    def stop(self):
+    def stop(self) -> None:
         """
         停止Telegram消息接收服务
         """
         # 停止所有typing任务
         for chat_id in list(self._typing_tasks.keys()):
             self._stop_typing_task(chat_id)
-        if self._bot:
-            self._bot.stop_polling()
+        if not self._bot:
+            return
+
+        self._bot.stop_bot()
+        if self._polling_thread:
             self._polling_thread.join()
+        self._polling_thread = None
+        self._bot = None
         logger.info("Telegram消息接收服务已停止")

app/modules/themoviedb/__init__.py

@@ -43,12 +43,6 @@ class TheMovieDbModule(_ModuleBase):
         self.category = CategoryHelper()
         self.scraper = TmdbScraper()
 
-    def on_config_changed(self):
-        # 停止模块
-        self.stop()
-        # 初始化模块
-        self.init_module()
-
     @staticmethod
     def get_name() -> str:
         return "TheMovieDb"
@@ -74,8 +68,12 @@ class TheMovieDbModule(_ModuleBase):
         """
         return 1
 
-    def stop(self):
+    def stop(self) -> None:
+        """停止模块"""
+        # 缓存持久化失败不能阻断 HTTP 客户端关闭
+        try:
             self.cache.save()
+        finally:
             self.tmdb.close()
 
     def test(self) -> Tuple[bool, str]:

app/modules/trimemedia/__init__.py

@@ -61,10 +61,14 @@ class TrimeMediaModule(_ModuleBase, _MediaServerBase[TrimeMedia]):
                 logger.info(f"飞牛影视 {name} 连接断开，尝试重连 ...")
                 server.reconnect()
 
-    def stop(self):
+    def stop(self) -> None:
+        """停止模块"""
         for server in self.get_instances().values():
+            try:
                 if server.is_authenticated():
                     server.disconnect()
+            except Exception as err:
+                logger.error(f"停止飞牛影视模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         """

app/modules/ugreen/__init__.py

@@ -60,10 +60,14 @@ class UgreenModule(_ModuleBase, _MediaServerBase[Ugreen]):
                 logger.info(f"绿联影视 {name} 连接断开，尝试重连 ...")
                 server.reconnect()
 
-    def stop(self):
+    def stop(self) -> None:
+        """停止模块"""
         for server in self.get_instances().values():
+            try:
                 if server.is_authenticated():
                     server.disconnect()
+            except Exception as err:
+                logger.error(f"停止绿联影视模块实例失败：{err}")
 
     def test(self) -> Optional[Tuple[bool, str]]:
         """

app/modules/wechat/__init__.py

@@ -24,7 +24,6 @@ class WechatModule(_ModuleBase, _MessageBase[WeChat]):
         """
         初始化模块
         """
-        self.stop()
         super().init_service(service_name=WeChat.__name__.lower(),
                              service_type=self._create_client)
         self._channel = MessageChannel.Wechat
@@ -54,10 +53,11 @@ class WechatModule(_ModuleBase, _MessageBase[WeChat]):
         """
         return 1
 
-    def stop(self):
+    def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
-            if hasattr(client, "stop"):
             try:
+                if hasattr(client, "stop"):
                     client.stop()
             except Exception as err:
                 logger.error(f"停止微信模块实例失败：{err}")

app/modules/wechatclawbot/__init__.py

@@ -23,7 +23,6 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]):
 
     def init_module(self) -> None:
         """初始化模块。"""
-        self.stop()
         super().init_service(
             service_name=WechatClawBot.__name__.lower(), service_type=WechatClawBot
         )
@@ -49,10 +48,9 @@ class WechatClawBotModule(_ModuleBase, _MessageBase[WechatClawBot]):
         """获取模块优先级。"""
         return 2
 
-    def stop(self):
-        """停止模块。"""
+    def stop(self) -> None:
+        """停止模块"""
         for client in self.get_instances().values():
-            if hasattr(client, "stop"):
             try:
                 client.stop()
             except Exception as err:

tests/test_module_lifecycle.py

@@ -0,0 +1,187 @@
+import threading
+from unittest.mock import Mock, patch
+
+import pytest
+
+from app.modules import _MessageBase
+from app.modules.discord import DiscordModule
+from app.modules.feishu import FeishuModule
+from app.modules.filter import FilterModule
+from app.modules.plex import PlexModule
+from app.modules.qqbot import QQBotModule
+from app.modules.slack import SlackModule
+from app.modules.telegram import TelegramModule
+from app.modules.telegram.telegram import Telegram
+from app.modules.themoviedb import TheMovieDbModule
+from app.modules.trimemedia import TrimeMediaModule
+from app.modules.ugreen import UgreenModule
+from app.modules.wechat import WechatModule
+from app.modules.wechatclawbot import WechatClawBotModule
+
+
+def test_config_reload_stops_before_initializing_latest_generation():
+    """同一模块的重载必须串行，并依次停止和初始化 generation。"""
+    module = TelegramModule()
+    call_order = []
+    reload_started = threading.Event()
+    reload_finished = threading.Event()
+
+    def reload_module():
+        reload_started.set()
+        module.on_config_changed()
+        reload_finished.set()
+
+    with patch.object(
+        module, "stop", side_effect=lambda: call_order.append("stop")
+    ), patch.object(
+        _MessageBase,
+        "init_service",
+        side_effect=lambda **_kwargs: call_order.append("init"),
+    ):
+        module._reload_lock.acquire()
+        try:
+            reload_thread = threading.Thread(target=reload_module)
+            reload_thread.start()
+            assert reload_started.wait(1)
+            assert not reload_finished.wait(0.1)
+        finally:
+            module._reload_lock.release()
+
+        assert reload_finished.wait(1)
+        reload_thread.join()
+
+    assert call_order == ["stop", "init"]
+
+
+def test_initialization_does_not_stop_a_fresh_module_generation():
+    """首次初始化只创建资源，停止旧 generation 由重载入口负责。"""
+    module = TelegramModule()
+
+    with patch.object(module, "stop") as stop, patch.object(
+        _MessageBase, "init_service"
+    ) as init_service:
+        module.init_module()
+
+    stop.assert_not_called()
+    init_service.assert_called_once()
+
+
+def test_config_reload_initializes_latest_generation_after_stop_failure():
+    """旧资源停止异常只记录错误，不阻止最新配置完成初始化。"""
+    module = TelegramModule()
+    call_order = []
+
+    def stop_with_failure():
+        call_order.append("stop")
+        raise RuntimeError("stop failed")
+
+    with patch.object(
+        module,
+        "stop",
+        side_effect=stop_with_failure,
+    ), patch.object(
+        _MessageBase,
+        "init_service",
+        side_effect=lambda **_kwargs: call_order.append("init"),
+    ):
+        module.on_config_changed()
+
+    assert call_order == ["stop", "init"]
+
+
+def test_tmdb_reload_closes_old_client_when_cache_save_fails():
+    """TMDB 缓存保存失败时仍须关闭旧客户端并初始化最新配置。"""
+    module = TheMovieDbModule()
+    module.cache = Mock()
+    module.cache.save.side_effect = OSError("cache write failed")
+    module.tmdb = Mock()
+
+    with patch.object(module, "init_module") as init_module:
+        module.on_config_changed()
+
+    module.tmdb.close.assert_called_once_with()
+    init_module.assert_called_once_with()
+
+
+def test_filter_reload_uses_shared_module_lifecycle_lock():
+    """过滤规则重载必须经过模块基类的串行 stop 和 init。"""
+    module = FilterModule()
+    reload_started = threading.Event()
+    reload_finished = threading.Event()
+    call_order = []
+
+    def reload_module():
+        reload_started.set()
+        module.on_config_changed()
+        reload_finished.set()
+
+    with patch(
+        "app.modules.filter.clear_rust_parse_options_cache",
+        side_effect=lambda: call_order.append("stop"),
+    ), patch.object(
+        module, "init_module", side_effect=lambda: call_order.append("init")
+    ):
+        module._reload_lock.acquire()
+        try:
+            reload_thread = threading.Thread(target=reload_module)
+            reload_thread.start()
+            assert reload_started.wait(1)
+            assert not reload_finished.wait(0.1)
+        finally:
+            module._reload_lock.release()
+
+        assert reload_finished.wait(1)
+        reload_thread.join()
+
+    assert call_order == ["stop", "init"]
+
+
+@pytest.mark.parametrize(
+    ("module_type", "stop_method", "requires_authentication"),
+    [
+        (DiscordModule, "stop", False),
+        (FeishuModule, "stop", False),
+        (QQBotModule, "stop", False),
+        (SlackModule, "stop", False),
+        (TelegramModule, "stop", False),
+        (WechatModule, "stop", False),
+        (WechatClawBotModule, "stop", False),
+        (PlexModule, "close", False),
+        (TrimeMediaModule, "disconnect", True),
+        (UgreenModule, "disconnect", True),
+    ],
+)
+def test_module_stop_isolates_each_service_instance(
+    module_type, stop_method, requires_authentication
+):
+    """单个服务停止失败时必须继续关闭同模块的其余实例。"""
+    module = module_type()
+    failed_client = Mock()
+    healthy_client = Mock()
+    getattr(failed_client, stop_method).side_effect = RuntimeError("stop failed")
+    if requires_authentication:
+        failed_client.is_authenticated.return_value = True
+        healthy_client.is_authenticated.return_value = True
+    module._instances = {"failed": failed_client, "healthy": healthy_client}
+
+    module.stop()
+
+    getattr(failed_client, stop_method).assert_called_once_with()
+    getattr(healthy_client, stop_method).assert_called_once_with()
+
+
+def test_telegram_stop_closes_sdk_and_waits_for_polling_thread():
+    """客户端停止完成后不得保留 SDK worker 或 polling 线程句柄。"""
+    client = Telegram.__new__(Telegram)
+    bot = Mock()
+    client._bot = bot
+    polling_thread = Mock()
+    client._polling_thread = polling_thread
+
+    client.stop()
+    client.stop()
+
+    bot.stop_bot.assert_called_once_with()
+    polling_thread.join.assert_called_once_with()
+    assert client._bot is None
+    assert client._polling_thread is None