Merge branch 'develop' of github.com:element-hq/synapse into matrix-org-hotfixes

changelog.d/17557.misc

@@ -0,0 +1 @@
+Add a utility function for generating random event IDs.

changelog.d/17559.doc

@@ -0,0 +1 @@
+Improve docstrings for profile methods.

changelog.d/17564.misc

@@ -0,0 +1 @@
+Speed up responding to media requests.

synapse/handlers/profile.py

@@ -74,6 +74,17 @@ class ProfileHandler:
         self._third_party_rules = hs.get_module_api_callbacks().third_party_event_rules
 
     async def get_profile(self, user_id: str, ignore_backoff: bool = True) -> JsonDict:
+        """
+        Get a user's profile as a JSON dictionary.
+
+        Args:
+            user_id: The user to fetch the profile of.
+            ignore_backoff: True to ignore backoff when fetching over federation.
+
+        Returns:
+            A JSON dictionary. For local queries this will include the displayname and avatar_url
+            fields. For remote queries it may contain arbitrary information.
+        """
         target_user = UserID.from_string(user_id)
 
         if self.hs.is_mine(target_user):
@@ -107,6 +118,15 @@ class ProfileHandler:
                 raise e.to_synapse_error()
 
     async def get_displayname(self, target_user: UserID) -> Optional[str]:
+        """
+        Fetch a user's display name from their profile.
+
+        Args:
+            target_user: The user to fetch the display name of.
+
+        Returns:
+            The user's display name or None if unset.
+        """
         if self.hs.is_mine(target_user):
             try:
                 displayname = await self.store.get_profile_displayname(target_user)
@@ -203,6 +223,15 @@ class ProfileHandler:
             await self._update_join_states(requester, target_user)
 
     async def get_avatar_url(self, target_user: UserID) -> Optional[str]:
+        """
+        Fetch a user's avatar URL from their profile.
+
+        Args:
+            target_user: The user to fetch the avatar URL of.
+
+        Returns:
+            The user's avatar URL or None if unset.
+        """
         if self.hs.is_mine(target_user):
             try:
                 avatar_url = await self.store.get_profile_avatar_url(target_user)
@@ -403,6 +432,12 @@ class ProfileHandler:
     async def _update_join_states(
         self, requester: Requester, target_user: UserID
     ) -> None:
+        """
+        Update the membership events of each room the user is joined to with the
+        new profile information.
+
+        Note that this stomps over any custom display name or avatar URL in member events.
+        """
         if not self.hs.is_mine(target_user):
             return
 

synapse/media/_base.py

@@ -51,15 +51,15 @@ from synapse.api.errors import Codes, cs_error
 from synapse.http.server import finish_request, respond_with_json
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import (
-    defer_to_thread,
+    defer_to_threadpool,
     make_deferred_yieldable,
     run_in_background,
 )
-from synapse.types import ISynapseReactor
 from synapse.util import Clock
 from synapse.util.stringutils import is_ascii
 
 if TYPE_CHECKING:
+    from synapse.server import HomeServer
     from synapse.storage.databases.main.media_repository import LocalMedia
 
 
@@ -132,6 +132,7 @@ def respond_404(request: SynapseRequest) -> None:
 
 
 async def respond_with_file(
+    hs: "HomeServer",
     request: SynapseRequest,
     media_type: str,
     file_path: str,
@@ -148,7 +149,7 @@ async def respond_with_file(
         add_file_headers(request, media_type, file_size, upload_name)
 
         with open(file_path, "rb") as f:
-            await ThreadedFileSender(request.reactor).beginFileTransfer(f, request)
+            await ThreadedFileSender(hs).beginFileTransfer(f, request)
 
         finish_request(request)
     else:
@@ -632,8 +633,9 @@ class ThreadedFileSender:
     # read.
     TIMEOUT_SECONDS = 90.0
 
-    def __init__(self, reactor: ISynapseReactor) -> None:
-        self.reactor = reactor
+    def __init__(self, hs: "HomeServer") -> None:
+        self.reactor = hs.get_reactor()
+        self.thread_pool = hs.get_media_sender_thread_pool()
 
         self.file: Optional[BinaryIO] = None
         self.deferred: "Deferred[None]" = Deferred()
@@ -661,7 +663,12 @@ class ThreadedFileSender:
 
         # We set the wakeup signal as we should start producing immediately.
         self.wakeup_event.set()
-        run_in_background(defer_to_thread, self.reactor, self._on_thread_read_loop)
+        run_in_background(
+            defer_to_threadpool,
+            self.reactor,
+            self.thread_pool,
+            self._on_thread_read_loop,
+        )
 
         return make_deferred_yieldable(self.deferred)
 

synapse/media/media_storage.py

@@ -57,7 +57,7 @@ from synapse.media._base import ThreadedFileSender
 from synapse.util import Clock
 from synapse.util.file_consumer import BackgroundFileConsumer
 
-from ..types import ISynapseReactor, JsonDict
+from ..types import JsonDict
 from ._base import FileInfo, Responder
 from .filepath import MediaFilePaths
 
@@ -209,7 +209,7 @@ class MediaStorage:
             local_path = os.path.join(self.local_media_directory, path)
             if os.path.exists(local_path):
                 logger.debug("responding with local file %s", local_path)
-                return FileResponder(self.reactor, open(local_path, "rb"))
+                return FileResponder(self.hs, open(local_path, "rb"))
             logger.debug("local file %s did not exist", local_path)
 
         for provider in self.storage_providers:
@@ -332,14 +332,12 @@ class FileResponder(Responder):
             is closed when finished streaming.
     """
 
-    def __init__(self, reactor: ISynapseReactor, open_file: BinaryIO):
-        self.reactor = reactor
+    def __init__(self, hs: "HomeServer", open_file: BinaryIO):
+        self.hs = hs
         self.open_file = open_file
 
     def write_to_consumer(self, consumer: IConsumer) -> Deferred:
-        return ThreadedFileSender(self.reactor).beginFileTransfer(
-            self.open_file, consumer
-        )
+        return ThreadedFileSender(self.hs).beginFileTransfer(self.open_file, consumer)
 
     def __exit__(
         self,

synapse/media/storage_provider.py

@@ -178,7 +178,7 @@ class FileStorageProviderBackend(StorageProvider):
 
         backup_fname = os.path.join(self.base_directory, path)
         if os.path.isfile(backup_fname):
-            return FileResponder(self.reactor, open(backup_fname, "rb"))
+            return FileResponder(self.hs, open(backup_fname, "rb"))
 
         return None
 

synapse/media/thumbnailer.py

@@ -374,11 +374,11 @@ class ThumbnailProvider:
                 await respond_with_multipart_responder(
                     self.hs.get_clock(),
                     request,
-                    FileResponder(self.reactor, open(file_path, "rb")),
+                    FileResponder(self.hs, open(file_path, "rb")),
                     media_info,
                 )
             else:
-                await respond_with_file(request, desired_type, file_path)
+                await respond_with_file(self.hs, request, desired_type, file_path)
         else:
             logger.warning("Failed to generate thumbnail")
             raise SynapseError(400, "Failed to generate thumbnail.")
@@ -456,7 +456,7 @@ class ThumbnailProvider:
         )
 
         if file_path:
-            await respond_with_file(request, desired_type, file_path)
+            await respond_with_file(self.hs, request, desired_type, file_path)
         else:
             logger.warning("Failed to generate thumbnail")
             raise SynapseError(400, "Failed to generate thumbnail.")

synapse/rest/client/room.py

@@ -67,7 +67,8 @@ from synapse.streams.config import PaginationConfig
 from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID, UserID
 from synapse.types.state import StateFilter
 from synapse.util.cancellation import cancellable
-from synapse.util.stringutils import parse_and_validate_server_name, random_string
+from synapse.util.events import generate_fake_event_id
+from synapse.util.stringutils import parse_and_validate_server_name
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -325,7 +326,7 @@ class RoomStateEventRestServlet(RestServlet):
                 )
                 event_id = event.event_id
         except ShadowBanError:
-            event_id = "$" + random_string(43)
+            event_id = generate_fake_event_id()
 
         set_tag("event_id", event_id)
         ret = {"event_id": event_id}
@@ -377,7 +378,7 @@ class RoomSendEventRestServlet(TransactionRestServlet):
             )
             event_id = event.event_id
         except ShadowBanError:
-            event_id = "$" + random_string(43)
+            event_id = generate_fake_event_id()
 
         set_tag("event_id", event_id)
         return 200, {"event_id": event_id}
@@ -1193,7 +1194,7 @@ class RoomRedactEventRestServlet(TransactionRestServlet):
 
             event_id = event.event_id
         except ShadowBanError:
-            event_id = "$" + random_string(43)
+            event_id = generate_fake_event_id()
 
         set_tag("event_id", event_id)
         return 200, {"event_id": event_id}

synapse/server.py

@@ -34,6 +34,7 @@ from typing_extensions import TypeAlias
 
 from twisted.internet.interfaces import IOpenSSLContextFactory
 from twisted.internet.tcp import Port
+from twisted.python.threadpool import ThreadPool
 from twisted.web.iweb import IPolicyForHTTPS
 from twisted.web.resource import Resource
 
@@ -941,3 +942,21 @@ class HomeServer(metaclass=abc.ABCMeta):
     @cache_in_self
     def get_task_scheduler(self) -> TaskScheduler:
         return TaskScheduler(self)
+
+    @cache_in_self
+    def get_media_sender_thread_pool(self) -> ThreadPool:
+        """Fetch the threadpool used to read files when responding to media
+        download requests."""
+
+        # We can choose a large threadpool size as these threads predominately
+        # do IO rather than CPU work.
+        media_threadpool = ThreadPool(
+            name="media_threadpool", minthreads=1, maxthreads=50
+        )
+
+        media_threadpool.start()
+        self.get_reactor().addSystemEventTrigger(
+            "during", "shutdown", media_threadpool.stop
+        )
+
+        return media_threadpool

synapse/storage/databases/main/profile.py

@@ -144,6 +144,16 @@ class ProfileWorkerStore(SQLBaseStore):
         return 50
 
     async def get_profileinfo(self, user_id: UserID) -> ProfileInfo:
+        """
+        Fetch the display name and avatar URL of a user.
+
+        Args:
+            user_id: The user ID to fetch the profile for.
+
+        Returns:
+            The user's display name and avatar URL. Values may be null if unset
+             or if the user doesn't exist.
+        """
         profile = await self.db_pool.simple_select_one(
             table="profiles",
             keyvalues={"full_user_id": user_id.to_string()},
@@ -158,6 +168,15 @@ class ProfileWorkerStore(SQLBaseStore):
         return ProfileInfo(avatar_url=profile[1], display_name=profile[0])
 
     async def get_profile_displayname(self, user_id: UserID) -> Optional[str]:
+        """
+        Fetch the display name of a user.
+
+        Args:
+            user_id: The user to get the display name for.
+
+        Raises:
+            404 if the user does not exist.
+        """
         return await self.db_pool.simple_select_one_onecol(
             table="profiles",
             keyvalues={"full_user_id": user_id.to_string()},
@@ -166,6 +185,15 @@ class ProfileWorkerStore(SQLBaseStore):
         )
 
     async def get_profile_avatar_url(self, user_id: UserID) -> Optional[str]:
+        """
+        Fetch the avatar URL of a user.
+
+        Args:
+            user_id: The user to get the avatar URL for.
+
+        Raises:
+            404 if the user does not exist.
+        """
         return await self.db_pool.simple_select_one_onecol(
             table="profiles",
             keyvalues={"full_user_id": user_id.to_string()},
@@ -174,6 +202,12 @@ class ProfileWorkerStore(SQLBaseStore):
         )
 
     async def create_profile(self, user_id: UserID) -> None:
+        """
+        Create a blank profile for a user.
+
+        Args:
+            user_id: The user to create the profile for.
+        """
         user_localpart = user_id.localpart
         await self.db_pool.simple_insert(
             table="profiles",

synapse/util/events.py

@@ -0,0 +1,29 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2024 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+#
+
+from synapse.util.stringutils import random_string
+
+
+def generate_fake_event_id() -> str:
+    """
+    Generate an event ID from random ASCII characters.
+
+    This is primarily useful for generating fake event IDs in response to
+    requests from shadow-banned users.
+
+    Returns:
+        A string intended to look like an event ID, but with no actual meaning.
+    """
+    return "$" + random_string(43)

tests/handlers/test_federation.py

@@ -44,7 +44,7 @@ from synapse.rest.client import login, room
 from synapse.server import HomeServer
 from synapse.storage.databases.main.events_worker import EventCacheEntry
 from synapse.util import Clock
-from synapse.util.stringutils import random_string
+from synapse.util.events import generate_fake_event_id
 
 from tests import unittest
 from tests.test_utils import event_injection
@@ -52,10 +52,6 @@ from tests.test_utils import event_injection
 logger = logging.getLogger(__name__)
 
 
-def generate_fake_event_id() -> str:
-    return "$fake_" + random_string(43)
-
-
 class FederationTestCase(unittest.FederatingHomeserverTestCase):
     servlets = [
         admin.register_servlets,

tests/server.py

@@ -1166,6 +1166,12 @@ def setup_test_homeserver(
 
     hs.get_auth_handler().validate_hash = validate_hash  # type: ignore[assignment]
 
+    # We need to replace the media threadpool with the fake test threadpool.
+    def thread_pool() -> threadpool.ThreadPool:
+        return reactor.getThreadPool()
+
+    hs.get_media_sender_thread_pool = thread_pool  # type: ignore[method-assign]
+
     # Load any configured modules into the homeserver
     module_api = hs.get_module_api()
     for module, module_config in hs.config.modules.loaded_modules: