From 992d14cb862617bf56bbfbfd0fe6cb4ac478b50e Mon Sep 17 00:00:00 2001 From: Goffi Date: Tue, 21 Apr 2026 12:42:52 +0200 Subject: [PATCH] core (xmpp): Message processing queue improvements: Messages are now queued per conversation (i.e. one-to-one chats, or MUCs). This allows messages to be treated in parallel instead of waiting sequentially for each one to be processed. --- libervia/backend/core/xmpp.py | 204 +++++++++++++++++++++++++++++----- 1 file changed, 176 insertions(+), 28 deletions(-) diff --git a/libervia/backend/core/xmpp.py b/libervia/backend/core/xmpp.py index 3745bfc2..421d9c47 100644 --- a/libervia/backend/core/xmpp.py +++ b/libervia/backend/core/xmpp.py @@ -17,14 +17,14 @@ # along with this program. If not, see . import calendar -from collections.abc import Iterable +from collections.abc import Awaitable, Callable, Iterable import copy from functools import partial import mimetypes from pathlib import Path import sys import time -from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, cast +from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, cast from urllib.parse import unquote, urlparse import uuid @@ -33,6 +33,7 @@ from treq.api import default_pool from treq.client import HTTPClient from twisted.internet import defer, error as internet_error, reactor from twisted.internet import ssl +from twisted.internet.base import DelayedCall from twisted.internet.endpoints import TCP4ClientEndpoint, UNIXClientEndpoint from twisted.python import failure from twisted.web.client import Agent @@ -56,7 +57,7 @@ from libervia.backend.memory import cache from libervia.backend.memory import encryption from libervia.backend.memory import persistent from libervia.backend.models.core import MessageData -from libervia.backend.models.tor import TorData, TorMode +from libervia.backend.models.tor import TorMode from libervia.backend.tools import xml_tools from libervia.backend.tools import utils from libervia.backend.tools.common import data_format @@ -83,6 +84,9 @@ NS_ROSTER_VER = "urn:xmpp:features:rosterver" # with roster jids ROSTER_VER_KEY = "@version@" +# Cleaning timeout (in seconds) +MESSAGE_QUEUE_TIMEOUT = 300 + class ClientPluginWrapper: """Use a plugin with default value if plugin is missing""" @@ -1397,18 +1401,175 @@ class SatXMPPComponent(SatXMPPEntity, core_types.SatXMPPComponent, component.Com return client +class TimestampedElement(domish.Element): + """domish.Element with a timestamp""" + + def __init__(self, element: domish.Element) -> None: + super().__init__((element.uri, element.name)) + self.defaultUri = element.defaultUri + self.attributes = element.attributes + self.children = element.children + self.localPrefixes = element.localPrefixes + self.parent = element.parent + + self._received_timestamp: float = time.time() + + @property + def received_timestamp(self) -> float: + """Element creation timestamp. + + It is used when message is received in processing workflow, hence its + name. + """ + return self._received_timestamp + + def __repr__(self) -> str: + return ( + f"<{self.__class__.__name__} {self.toXml()} " + f"timestamp={self._received_timestamp}>" + ) + + +class MessagesQueue: + + def __init__( + self, + client: SatXMPPEntity, + key: str, + manager: "MessagesQueueManager", + ) -> None: + self.client: SatXMPPEntity = client + self.key: str = key + self.queue: defer.DeferredQueue[TimestampedElement] = defer.DeferredQueue() + self._manager: MessagesQueueManager = manager + self._timer: DelayedCall | None = None + defer.ensureDeferred(self._process_loop()) + + @property + def size(self) -> int: + """Number of messages waiting in the queue""" + return len(self.queue.pending) + + def put(self, message_elt: TimestampedElement) -> None: + """Add a message to the queue.""" + self.queue.put(message_elt) + self._reset_timer() + + def _reset_timer(self) -> None: + if self._timer is not None and self._timer.active(): + self._timer.cancel() + self._timer = reactor.callLater(MESSAGE_QUEUE_TIMEOUT, self._cleanup) + + def _cleanup(self) -> None: + """Called when timeout expires""" + if not self.queue.pending: + # The queue is empty, we can destroy it. + self._manager.remove_queue(self.key) + else: + # There are still message pending, we wait again. + self._reset_timer() + + async def _process_loop(self) -> None: + """Process messages - runs until queue is destroyed""" + while True: + message = await self.queue.get() + try: + await self._manager._process_callback(self.client, message) + except Exception: + log.exception(f"Error processing message in {self.key}") + + +class MessagesQueueManager: + """Manages queues of messages.""" + + def __init__( + self, + process_callback: Callable[[SatXMPPEntity, domish.Element], Awaitable[None]], + ) -> None: + self.queues: dict[str, MessagesQueue] = {} + self._process_callback = process_callback + + @staticmethod + def get_queue_key(client: SatXMPPEntity, message_elt: domish.Element) -> str: + """Get the key associate with the MessagesQueue. + + Queue are grouped per one2one conversation and per MUC. The key is generated from + "from" and "to" attribute to have message routed with the right queue. + """ + try: + from_jid = jid.JID(message_elt["from"]) + except KeyError: + from_jid = client.server_jid + + try: + to_jid = jid.JID(message_elt["to"]) + except KeyError: + to_jid = client.jid + + msg_type = message_elt.getAttribute("type", "normal") + + if msg_type == "groupchat": + room_jid_s = from_jid.userhost() + return f"muc:{room_jid_s}" + else: + # We sort the pair to have the same key for both directions. + key = tuple(sorted([from_jid.userhost(), to_jid.userhost()])) + return f"one2one:{key[0]}:{key[1]}" + + def get_or_create_queue( + self, client: SatXMPPEntity, message_elt: TimestampedElement + ) -> MessagesQueue: + """Get existing queue or create new one""" + key = self.get_queue_key(client, message_elt) + if key not in self.queues: + log.debug(f"Creating queue for {key}") + self.queues[key] = MessagesQueue( + client=client, + key=key, + manager=self, + ) + return self.queues[key] + + def put(self, client: SatXMPPEntity, message_elt: TimestampedElement) -> None: + """Put message in queue. + + The queue will be created if needed. + @param client: XMPP client + @param message_elt: Message to put in queue. + """ + queue = self.get_or_create_queue(client, message_elt) + queue.put(message_elt) + + def remove_queue(self, key: str) -> None: + """Remove an unused queue.""" + try: + queue = self.queues.pop(key) + except KeyError: + pass + else: + if queue._timer is not None and queue._timer.active(): + queue._timer.cancel() + + def get_stats(self) -> dict[str, Any]: + """Debug/monitoring stats""" + return { + "active_queues": len(self.queues), + "queue_sizes": {key: queue.size for key, queue in self.queues.items()}, + } + + class SatMessageProtocol(xmppim.MessageProtocol): def __init__(self): xmppim.MessageProtocol.__init__(self) - self.messages_queue = defer.DeferredQueue() + self.queue_manager = MessagesQueueManager(self.process_message) def setHandlerParent(self, parent): super().setHandlerParent(parent) - defer.ensureDeferred(self.process_messages()) @property - def client(self): + def client(self) -> SatXMPPEntity: + assert self.parent is not None return self.parent def normalize_ns(self, elt: domish.Element, namespace: Optional[str]) -> None: @@ -1518,10 +1679,10 @@ class SatMessageProtocol(xmppim.MessageProtocol): # delay and timestamp try: - received_timestamp = message_elt._received_timestamp + received_timestamp = message_elt.received_timestamp except AttributeError: - # message_elt._received_timestamp should have been set in onMessage - # but if parse_message is called directly, it can be missing + # message_elt.received_timestamp is avaiable with TimestampedElement, but if + # parse_message is called directly, it can be missing log.debug( "missing received timestamp for {message_elt}".format( message_elt=message_elt @@ -1543,32 +1704,19 @@ class SatMessageProtocol(xmppim.MessageProtocol): G.host.trigger.point("message_parse", client, message_elt, data) return data - def onMessage(self, message_elt: domish.Element) -> None: - message_elt._received_timestamp = time.time() - self.messages_queue.put(message_elt) + def onMessage(self, message: domish.Element) -> None: + """Called when a is received. - async def process_messages(self) -> None: - """Process message in order - - Messages are processed in a queue to avoid race conditions and ensure orderly - processing. + This put the message in queue, so they are processed in order. """ - client = self.parent - if client is None: - log.error("client should not be None!") - raise exceptions.InternalError() - while True: - message_elt = await self.messages_queue.get() - try: - await self.process_message(client, message_elt) - except Exception: - log.exception(f"Can't process message {message_elt.toXml()}") + self.queue_manager.put(self.client, TimestampedElement(message)) def _on_processing_timeout( self, message_elt: domish.Element, async_point_d: defer.Deferred ) -> None: + key = MessagesQueueManager.get_queue_key(self.client, message_elt) log.error( - "Processing of following message took too long, cancelling:" + f"Processing of following message took too long in {key}, cancelling:" f"{message_elt.toXml()}" ) async_point_d.cancel()