From f057544e313c796c05751eaa1b7370dba8574ca1 Mon Sep 17 00:00:00 2001 From: Wang Qi Date: Wed, 5 Aug 2026 11:52:53 +0800 Subject: [PATCH] Fix: SIGTERM not handle well in ragflow server chat channel (#17828) --- api/channels/bootstrap.py | 13 +++++++++++-- api/ragflow_server.py | 20 ++++++++++++++------ 2 files changed, 25 insertions(+), 8 deletions(-) diff --git a/api/channels/bootstrap.py b/api/channels/bootstrap.py index a607d40383..5139cd11e5 100644 --- a/api/channels/bootstrap.py +++ b/api/channels/bootstrap.py @@ -219,13 +219,17 @@ async def _start_channel(running: dict, account_id: str, channel: str, credentia return True -async def _reconcile(running: dict, failed: dict) -> None: +async def _reconcile(running: dict, failed: dict, stop_event: threading.Event) -> None: """Diff desired (DB) vs running channels and apply start/stop/restart. ``failed`` remembers configs that could not be started so they are not retried (and re-logged) every tick until their credentials change. """ + if stop_event.is_set(): + return desired = await asyncio.to_thread(_desired_channels) + if stop_event.is_set(): + return # Stop channels that were removed or whose credentials/type changed. for account_id in list(running.keys()): @@ -266,7 +270,12 @@ async def run_channels(stop_event: threading.Event) -> None: try: while not stop_event.is_set(): try: - await _reconcile(running, failed) + await _reconcile(running, failed, stop_event) + except RuntimeError as ex: + if stop_event.is_set(): + LOGGER.info("chat channel reconcile stopped") + break + LOGGER.error("chat channel reconcile failed: %s", ex) except Exception as ex: LOGGER.error("chat channel reconcile failed: %s", ex) diff --git a/api/ragflow_server.py b/api/ragflow_server.py index 5f580075d6..f9c9d62c08 100644 --- a/api/ragflow_server.py +++ b/api/ragflow_server.py @@ -48,6 +48,7 @@ from agent.plugin import GlobalPluginManager from rag.utils.redis_conn import RedisDistributedLock stop_event = threading.Event() +chat_channel_thread = None RAGFLOW_DEBUGPY_LISTEN = int(os.environ.get("RAGFLOW_DEBUGPY_LISTEN", "0")) @@ -71,11 +72,16 @@ def update_progress(): stop_event.wait(6) +def stop_background_services(): + stop_event.set() + if chat_channel_thread and chat_channel_thread.is_alive() and chat_channel_thread is not threading.current_thread(): + chat_channel_thread.join(timeout=5) + + def signal_handler(sig, frame): logging.info("Received interrupt signal, shutting down...") shutdown_all_mcp_sessions() - stop_event.set() - stop_event.wait(1) + stop_background_services() sys.exit(0) @@ -137,17 +143,18 @@ if __name__ == "__main__": t.start() def start_chat_channels(): + global chat_channel_thread try: from api.channels.bootstrap import start_channel_server logging.info("Starting chat channel server thread") - t = threading.Thread( + chat_channel_thread = threading.Thread( target=start_channel_server, args=(stop_event,), daemon=True, name="chat-channels", ) - t.start() + chat_channel_thread.start() except Exception: logging.exception("Failed to start chat channel server") @@ -165,6 +172,7 @@ if __name__ == "__main__": app.run(host=settings.HOST_IP, port=settings.HOST_PORT, use_reloader=RuntimeConfig.DEBUG, debug=False) except Exception as e: logging.exception(f"Unhandled exception: {e}") - stop_event.set() - stop_event.wait(1) + stop_background_services() os.kill(os.getpid(), signal.SIGKILL) + finally: + stop_background_services()