Fix: SIGTERM not handle well in ragflow server chat channel (#17828)

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)
 

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()