diff --git a/changelog.d/17557.misc b/changelog.d/17557.misc new file mode 100644 index 0000000000..535f4b6e5f --- /dev/null +++ b/changelog.d/17557.misc @@ -0,0 +1 @@ +Add a utility function for generating random event IDs. \ No newline at end of file diff --git a/changelog.d/17559.doc b/changelog.d/17559.doc new file mode 100644 index 0000000000..e54a122b74 --- /dev/null +++ b/changelog.d/17559.doc @@ -0,0 +1 @@ +Improve docstrings for profile methods. diff --git a/changelog.d/17564.misc b/changelog.d/17564.misc new file mode 100644 index 0000000000..cfa8089a81 --- /dev/null +++ b/changelog.d/17564.misc @@ -0,0 +1 @@ +Speed up responding to media requests. diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 6663d4b271..af8cd838ee 100644 --- a/synapse/handlers/profile.py +++ b/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 diff --git a/synapse/media/_base.py b/synapse/media/_base.py index ad80098e9f..89dea39163 100644 --- a/synapse/media/_base.py +++ b/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) diff --git a/synapse/media/media_storage.py b/synapse/media/media_storage.py index e06273c92f..cf4208eb71 100644 --- a/synapse/media/media_storage.py +++ b/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, diff --git a/synapse/media/storage_provider.py b/synapse/media/storage_provider.py index 355df999d2..300952025a 100644 --- a/synapse/media/storage_provider.py +++ b/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 diff --git a/synapse/media/thumbnailer.py b/synapse/media/thumbnailer.py index 3380315b27..042851021c 100644 --- a/synapse/media/thumbnailer.py +++ b/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.") diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py index 903c74f6d8..7d57904d69 100644 --- a/synapse/rest/client/room.py +++ b/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} diff --git a/synapse/server.py b/synapse/server.py index 46b9d83a04..8b07bb39a0 100644 --- a/synapse/server.py +++ b/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 diff --git a/synapse/storage/databases/main/profile.py b/synapse/storage/databases/main/profile.py index 996aea808d..41cf08211f 100644 --- a/synapse/storage/databases/main/profile.py +++ b/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", diff --git a/synapse/util/events.py b/synapse/util/events.py new file mode 100644 index 0000000000..ad9b946578 --- /dev/null +++ b/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: +# . +# +# + +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) diff --git a/tests/handlers/test_federation.py b/tests/handlers/test_federation.py index 3fe5b0a1b4..9847893fce 100644 --- a/tests/handlers/test_federation.py +++ b/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, diff --git a/tests/server.py b/tests/server.py index 3e377585ce..95aff6f66c 100644 --- a/tests/server.py +++ b/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: