From 4b36801b53ee5a38c4bf9e54e82cacd8c2462fdf Mon Sep 17 00:00:00 2001 From: wdeveloper16 Date: Mon, 25 May 2026 16:45:40 +0200 Subject: [PATCH] fix: resolve asyncio correctness issues (fire-and-forget tasks, event loop nesting) (#14761) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Fixes the confirmed asyncio anti-patterns from #14755. Only the three verified bugs are addressed; patterns already correctly using `asyncio.new_event_loop()` in a fresh thread are left untouched. ### Changes **`api/apps/restful_apis/tenant_api.py` — fire-and-forget `send_invite_email`** `asyncio.create_task()` was called without storing the `Task` reference. CPython's GC can collect an unfinished task, silently cancelling it and swallowing exceptions. Fixed by storing the task in a module-level `_background_tasks: set[Task]` with a `done_callback` to discard it on completion — the standard Python idiom for safe background tasks. **`api/apps/restful_apis/agent_api.py` — fire-and-forget `background_run`** Same root cause in the webhook "Immediately" execution path. Same fix applied. **`rag/llm/chat_model.py` (`LocalLLM._stream_response`) — `asyncio.get_event_loop()` on running loop** `asyncio.get_event_loop()` returns Quart's running event loop when called from an async context. Calling `loop.run_until_complete()` on it raises `RuntimeError`. Replaced with `asyncio.new_event_loop()` so the generator uses a dedicated fresh loop, closed in a `finally` block. ## What was NOT changed - `llm_service._sync_from_async_stream` and `evaluation_service._sync_from_async_gen`: both already correctly use `asyncio.new_event_loop()` inside a fresh thread. - `llm_service._run_coroutine_sync`: only caller is `rag/app/resume.py` (sync context), so `thread.join()` is correct there. - `requests` in agent tools: sync methods dispatched through thread pools; httpx migration is a separate, larger refactor. ## Test plan - [ ] Invite a team member and confirm the email is sent with no task warnings in logs. - [ ] Trigger a webhook agent in "Immediately" mode; confirm canvas state is persisted after background run. - [ ] Verify `LocalLLM` (Jina backend) chat and streaming work end-to-end. Closes #14755 --------- Co-authored-by: Zhichang Yu --- api/apps/restful_apis/agent_api.py | 9 ++++++++- api/apps/restful_apis/tenant_api.py | 18 +++++++++++++++++- rag/llm/chat_model.py | 4 +++- 3 files changed, 28 insertions(+), 3 deletions(-) diff --git a/api/apps/restful_apis/agent_api.py b/api/apps/restful_apis/agent_api.py index d7641d43d..0803a2b11 100644 --- a/api/apps/restful_apis/agent_api.py +++ b/api/apps/restful_apis/agent_api.py @@ -25,6 +25,7 @@ import json import logging import time from functools import partial, wraps +from typing import Set from api.utils.web_utils import CONTENT_TYPE_MAP, apply_safe_file_response_headers import jwt @@ -63,6 +64,9 @@ from common.constants import RetCode from common.misc_utils import get_uuid, thread_pool_exec from peewee import MySQLDatabase, PostgresqlDatabase +# Keeps strong references to fire-and-forget tasks so they are not GC'd before completion. +_background_tasks: Set[asyncio.Task] = set() + def _require_canvas_access_sync(func): @wraps(func) @@ -2079,7 +2083,10 @@ async def webhook(agent_id: str): except Exception: logging.exception("Failed to append webhook trace") - asyncio.create_task(background_run()) + task = asyncio.create_task(background_run()) + if isinstance(task, asyncio.Task): + _background_tasks.add(task) + task.add_done_callback(_background_tasks.discard) return resp else: async def sse(): diff --git a/api/apps/restful_apis/tenant_api.py b/api/apps/restful_apis/tenant_api.py index 4d45337cb..56f40444a 100644 --- a/api/apps/restful_apis/tenant_api.py +++ b/api/apps/restful_apis/tenant_api.py @@ -15,6 +15,7 @@ # import asyncio import logging +from typing import Set from api.apps import current_user, login_required from api.db import UserTenantRole @@ -33,6 +34,9 @@ from common.constants import RetCode, StatusEnum from common.misc_utils import get_uuid from common.time_utils import delta_seconds +# Keeps strong references to fire-and-forget tasks so they are not GC'd before completion. +_background_tasks: Set[asyncio.Task] = set() + @manager.route("/tenants//users", methods=["GET"]) # noqa: F821 @login_required @@ -97,7 +101,16 @@ async def create(tenant_id): if user: user_name = user.nickname - asyncio.create_task( + def _on_invite_email_done(done_task: asyncio.Task) -> None: + _background_tasks.discard(done_task) + try: + done_task.result() + except asyncio.CancelledError: + logging.warning("Invite email task cancelled: tenant_id=%s to=%s", tenant_id, invite_user_email) + except Exception: + logging.exception("Invite email task failed: tenant_id=%s to=%s", tenant_id, invite_user_email) + + task = asyncio.create_task( send_invite_email( to_email=invite_user_email, invite_url=settings.MAIL_FRONTEND_URL, @@ -105,6 +118,9 @@ async def create(tenant_id): inviter=user_name or current_user.email, ) ) + if isinstance(task, asyncio.Task): + _background_tasks.add(task) + task.add_done_callback(_on_invite_email_done) except Exception as exc: logging.exception(f"Failed to send invite email to {invite_user_email}: {exc}") return get_json_result( diff --git a/rag/llm/chat_model.py b/rag/llm/chat_model.py index 7c1652ddb..82a9e48f9 100644 --- a/rag/llm/chat_model.py +++ b/rag/llm/chat_model.py @@ -766,9 +766,9 @@ class LocalLLM(Base): from rag.svr.jina_server import Generation answer = "" + loop = asyncio.new_event_loop() try: res = self.client.stream_doc(on=endpoint, inputs=prompt, return_type=Generation) - loop = asyncio.get_event_loop() try: while True: answer = loop.run_until_complete(res.__anext__()).text @@ -777,6 +777,8 @@ class LocalLLM(Base): pass except Exception as e: yield answer + "\n**ERROR**: " + str(e) + finally: + loop.close() yield num_tokens_from_string(answer) def chat(self, system, history, gen_conf=None, **kwargs):