From c436e0366c18b26d18d1686bd155b60f75f96116 Mon Sep 17 00:00:00 2001 From: Classic298 <27028174+Classic298@users.noreply.github.com> Date: Sun, 1 Mar 2026 19:43:03 +0100 Subject: [PATCH] perf: async DB calls, skip intermediate status writes, elif chain in event emitter (#22107) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three improvements to the socket event emitter hot path (when realtime chat save is enabled): 1. Wrap all synchronous Chats.* DB calls in asyncio.to_thread() to avoid blocking the event loop during streaming. With N concurrent users, sync DB calls serialize all writes and block socket event delivery. 2. Only persist final (done=True) status events to DB. Intermediate statuses (tool calling progress, web search progress, etc.) are ephemeral UI-only data already delivered via socket — writing every one to DB is unnecessary I/O. 3. Convert if/if/if chain to if/elif since event types are mutually exclusive, avoiding unnecessary string comparisons after a match. --- backend/open_webui/socket/main.py | 58 ++++++++++++++++++++----------- 1 file changed, 37 insertions(+), 21 deletions(-) diff --git a/backend/open_webui/socket/main.py b/backend/open_webui/socket/main.py index c1d7e5c29e..57660bd177 100644 --- a/backend/open_webui/socket/main.py +++ b/backend/open_webui/socket/main.py @@ -796,15 +796,23 @@ def get_event_emitter(request_info, update_db=True): and not request_info.get("chat_id", "").startswith("local:") ): - if "type" in event_data and event_data["type"] == "status": - Chats.add_message_status_to_chat_by_id_and_message_id( - request_info["chat_id"], - request_info["message_id"], - event_data.get("data", {}), - ) + event_type = event_data.get("type") - if "type" in event_data and event_data["type"] == "message": - message = Chats.get_message_by_id_and_message_id( + if event_type == "status": + # Only persist final (done) status to DB; intermediate + # statuses are ephemeral UI-only data sent via socket. + status_data = event_data.get("data", {}) + if status_data.get("done", False): + await asyncio.to_thread( + Chats.add_message_status_to_chat_by_id_and_message_id, + request_info["chat_id"], + request_info["message_id"], + status_data, + ) + + elif event_type == "message": + message = await asyncio.to_thread( + Chats.get_message_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], ) @@ -813,7 +821,8 @@ def get_event_emitter(request_info, update_db=True): content = message.get("content", "") content += event_data.get("data", {}).get("content", "") - Chats.upsert_message_to_chat_by_id_and_message_id( + await asyncio.to_thread( + Chats.upsert_message_to_chat_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], { @@ -821,10 +830,11 @@ def get_event_emitter(request_info, update_db=True): }, ) - if "type" in event_data and event_data["type"] == "replace": + elif event_type == "replace": content = event_data.get("data", {}).get("content", "") - Chats.upsert_message_to_chat_by_id_and_message_id( + await asyncio.to_thread( + Chats.upsert_message_to_chat_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], { @@ -832,8 +842,9 @@ def get_event_emitter(request_info, update_db=True): }, ) - if "type" in event_data and event_data["type"] == "embeds": - message = Chats.get_message_by_id_and_message_id( + elif event_type == "embeds": + message = await asyncio.to_thread( + Chats.get_message_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], ) @@ -841,7 +852,8 @@ def get_event_emitter(request_info, update_db=True): embeds = event_data.get("data", {}).get("embeds", []) embeds.extend(message.get("embeds", [])) - Chats.upsert_message_to_chat_by_id_and_message_id( + await asyncio.to_thread( + Chats.upsert_message_to_chat_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], { @@ -849,8 +861,9 @@ def get_event_emitter(request_info, update_db=True): }, ) - if "type" in event_data and event_data["type"] == "files": - message = Chats.get_message_by_id_and_message_id( + elif event_type == "files": + message = await asyncio.to_thread( + Chats.get_message_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], ) @@ -858,7 +871,8 @@ def get_event_emitter(request_info, update_db=True): files = event_data.get("data", {}).get("files", []) files.extend(message.get("files", [])) - Chats.upsert_message_to_chat_by_id_and_message_id( + await asyncio.to_thread( + Chats.upsert_message_to_chat_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], { @@ -866,10 +880,11 @@ def get_event_emitter(request_info, update_db=True): }, ) - if event_data.get("type") in ["source", "citation"]: + elif event_type in ("source", "citation"): data = event_data.get("data", {}) - if data.get("type") == None: - message = Chats.get_message_by_id_and_message_id( + if data.get("type") is None: + message = await asyncio.to_thread( + Chats.get_message_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], ) @@ -877,7 +892,8 @@ def get_event_emitter(request_info, update_db=True): sources = message.get("sources", []) sources.append(data) - Chats.upsert_message_to_chat_by_id_and_message_id( + await asyncio.to_thread( + Chats.upsert_message_to_chat_by_id_and_message_id, request_info["chat_id"], request_info["message_id"], {