From ac8077bcee3328be7c7b2b99ee7249d1bd883a9c Mon Sep 17 00:00:00 2001 From: Matthew Hodgson Date: Sun, 22 Mar 2026 02:20:59 +0000 Subject: [PATCH] =?UTF-8?q?=E2=8F=BA=20The=20remaining=2037=20break=20down?= =?UTF-8?q?=20as:=20=20=20-=2011=20http/client.py=20=E2=80=94=20Twisted=20?= =?UTF-8?q?HTTP=20client=20(needs=20full=20aiohttp=20rewrite)=20=20=20-=20?= =?UTF-8?q?5=20connectproxyclient.py=20=E2=80=94=20Twisted=20proxy=20clien?= =?UTF-8?q?t=20(needs=20full=20rewrite)=20=20=20-=204=20background=5Fproce?= =?UTF-8?q?ss=5Fmetrics.py=20=E2=80=94=20type=20annotations=20(harmless)?= =?UTF-8?q?=20=20=20-=204=20=5Fscripts/synapse=5Fport=5Fdb.py=20=E2=80=94?= =?UTF-8?q?=20CLI=20script=20(low=20priority)=20=20=20-=203=20server.py=20?= =?UTF-8?q?=E2=80=94=20startup/shutdown=20(type=20annotations=20+=20defer.?= =?UTF-8?q?ensureDeferred)=20=20=20-=203=20app/=5Fbase.py=20=E2=80=94=20ap?= =?UTF-8?q?p=20startup=20(defer.ensureDeferred)=20=20=20-=202=20http/serve?= =?UTF-8?q?r.py=20=E2=80=94=20type=20annotation=20+=20isinstance=20check?= =?UTF-8?q?=20=20=20-=201=20manhole.py=20=E2=80=94=20interactive=20manhole?= =?UTF-8?q?=20=20=20-=201=20util/=5F=5Finit=5F=5F.py=20=E2=80=94=20unwrapF?= =?UTF-8?q?irstError=20compat=20=20=20-=201=20crypto/keyring.py=20?= =?UTF-8?q?=E2=80=94=20type=20annotation=20=20=20-=201=20admin=5Fcmd.py=20?= =?UTF-8?q?=E2=80=94=20CLI=20script=20=20=20-=201=20=5Fscripts/update=5Fsy?= =?UTF-8?q?napse=5Fdatabase.py=20=E2=80=94=20CLI=20script?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Here's a summary of what was accomplished in this session: Completed: Replaced all .addErrback/.addCallback/.addBoth calls on production code that was receiving asyncio Tasks instead of Deferreds. Files fixed: 1. cached_call.py — Replaced Deferred/Failure/.addBoth with asyncio.Task/add_done_callback 2. response_cache.py — Replaced .addBoth with add_done_callback, .called with has_called() 3. push_rule.py — Replaced gather_results(...).addErrback(unwrapFirstError) with await gather_results(...) 4. bulk_push_rule_evaluator.py — Same pattern 5. events_worker.py — Replaced defer.Deferred() with asyncio.Future(), defer.gatherResults with asyncio.gather, .errback()/.callback() with .set_exception()/.set_result() 6. partial_state_events_tracker.py — Same pattern, both PartialStateEventsTracker and PartialCurrentStateTracker 7. initial_sync.py — Replaced .addCallback chain with async wrapper, removed gather_results().addErrback() 8. message.py — Replaced .addErrback on Task with try/except wrapper 9. federation_server.py — Removed gather_results().addErrback() 10. transactions.py — Replaced .addErrback with add_done_callback 11. e2e_keys.py, appservice.py, sender/__init__.py, stream.py, persist_events.py, notifier.py, batching_queue.py — All migrated from Twisted to asyncio Also fixed: - NativeLoopingCall missing .running property - Heapq tiebreaker for _pending_sleeps (Future comparison error) - FakeChannel.await_result now advances NativeClock fake time - run_in_background returns resolved Futures for plain values (not unawaited coroutines) Remaining 37 defer.* calls are in HTTP client (needs full rewrite), type annotations, startup code, and CLI scripts — all expected. --- synapse/federation/federation_server.py | 17 ++-- synapse/federation/sender/__init__.py | 20 ++-- synapse/handlers/appservice.py | 98 ++++++++----------- synapse/handlers/e2e_keys.py | 20 ++-- synapse/handlers/initial_sync.py | 66 ++++++------- synapse/handlers/message.py | 24 ++--- synapse/logging/context.py | 18 ++-- synapse/notifier.py | 48 +++++---- synapse/push/bulk_push_rule_evaluator.py | 53 ++++------ synapse/rest/client/transactions.py | 13 ++- synapse/storage/controllers/persist_events.py | 8 +- .../storage/databases/main/events_worker.py | 57 ++++++----- synapse/storage/databases/main/push_rule.py | 52 ++++------ synapse/storage/databases/main/stream.py | 34 +++---- .../util/partial_state_events_tracker.py | 47 ++++----- synapse/util/async_helpers.py | 18 ++++ synapse/util/batching_queue.py | 30 +++--- synapse/util/caches/cached_call.py | 49 +++++----- synapse/util/caches/response_cache.py | 12 +-- synapse/util/clock.py | 13 ++- tests/server.py | 19 +++- tests/unittest.py | 1 + 22 files changed, 341 insertions(+), 376 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 5d25efd44b..21bbc28383 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -69,7 +69,6 @@ from synapse.federation.units import Edu, Transaction, serialize_and_filter_pdus from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME from synapse.http.servlet import assert_params_in_dict from synapse.logging.context import ( - make_deferred_yieldable, nested_logging_context, run_in_background, ) @@ -90,7 +89,6 @@ from synapse.storage.databases.main.lock import Lock from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary from synapse.storage.roommember import MemberSummary from synapse.types import JsonDict, StateMap, UserID, get_domain_from_id -from synapse.util import unwrapFirstError from synapse.util.async_helpers import Linearizer, concurrently_execute, gather_results from synapse.util.caches.response_cache import ResponseCache from synapse.util.duration import Duration @@ -408,16 +406,13 @@ class FederationServer(FederationBase): # We process PDUs and EDUs in parallel. This is important as we don't # want to block things like to device messages from reaching clients # behind the potentially expensive handling of PDUs. - pdu_results, _ = await make_deferred_yieldable( - gather_results( - ( - run_in_background( - self._handle_pdus_in_txn, origin, transaction, request_time - ), - run_in_background(self._handle_edus_in_txn, origin, transaction), + pdu_results, _ = await gather_results( + ( + run_in_background( + self._handle_pdus_in_txn, origin, transaction, request_time ), - consumeErrors=True, - ).addErrback(unwrapFirstError) + run_in_background(self._handle_edus_in_txn, origin, transaction), + ), ) response = {"pdus": pdu_results} diff --git a/synapse/federation/sender/__init__.py b/synapse/federation/sender/__init__.py index 3702992579..14dc680903 100644 --- a/synapse/federation/sender/__init__.py +++ b/synapse/federation/sender/__init__.py @@ -143,10 +143,7 @@ from typing import ( import attr from prometheus_client import Counter -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio import synapse.metrics from synapse.api.constants import EventTypes, Membership @@ -158,7 +155,7 @@ from synapse.federation.sender.per_destination_queue import ( ) from synapse.federation.sender.transaction_manager import TransactionManager from synapse.federation.units import Edu -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.metrics import ( SERVER_NAME_LABEL, LaterGauge, @@ -746,14 +743,11 @@ class FederationSender(AbstractFederationSender): event = event_cache.event events_by_room.setdefault(event.room_id, []).append(event) - await make_deferred_yieldable( - defer.gatherResults( - [ - run_in_background(handle_room_events, evs) - for evs in events_by_room.values() - ], - consumeErrors=True, - ) + await asyncio.gather( + *( + run_in_background(handle_room_events, evs) + for evs in events_by_room.values() + ), ) logger.debug("Successfully handled up to %i", next_token) diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index 8bfedc61bf..ab17d26615 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -28,17 +28,14 @@ from typing import ( from prometheus_client import Counter -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio import synapse from synapse.api.constants import EduTypes, EventTypes from synapse.appservice import ApplicationService from synapse.events import EventBase from synapse.handlers.presence import format_user_presence_state -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.metrics import ( SERVER_NAME_LABEL, event_processing_loop_counter, @@ -191,14 +188,11 @@ class ApplicationServicesHandler: for event in events: await handle_event(event) - await make_deferred_yieldable( - defer.gatherResults( - [ - run_in_background(handle_room_events, evs) - for evs in events_by_room.values() - ], - consumeErrors=True, - ) + await asyncio.gather( + *( + run_in_background(handle_room_events, evs) + for evs in events_by_room.values() + ), ) await self.store.set_appservice_last_pos(upper_bound) @@ -763,21 +757,19 @@ class ApplicationServicesHandler: ) -> list[JsonDict]: services = self._get_services_for_3pn(protocol) - results = await make_deferred_yieldable( - defer.DeferredList( - [ - run_in_background( - self.appservice_api.query_3pe, service, kind, protocol, fields - ) - for service in services - ], - consumeErrors=True, - ) + results = await asyncio.gather( + *( + run_in_background( + self.appservice_api.query_3pe, service, kind, protocol, fields + ) + for service in services + ), + return_exceptions=True, ) ret = [] - for success, result in results: - if success: + for result in results: + if not isinstance(result, BaseException): ret.extend(result) return ret @@ -910,26 +902,24 @@ class ApplicationServicesHandler: continue # Query each service in parallel. - results = await make_deferred_yieldable( - defer.DeferredList( - [ - run_in_background( # type: ignore[call-overload] - self.appservice_api.claim_client_keys, - # We know this must be an app service. - self.store.get_app_service_by_id(service_id), - service_query, - ) - for service_id, service_query in query_by_appservice.items() - ], - consumeErrors=True, - ) + results = await asyncio.gather( + *( + run_in_background( # type: ignore[call-overload] + self.appservice_api.claim_client_keys, + # We know this must be an app service. + self.store.get_app_service_by_id(service_id), + service_query, + ) + for service_id, service_query in query_by_appservice.items() + ), + return_exceptions=True, ) # Patch together the results -- they are all independent (since they # require exclusive control over the users, which is the outermost key). claimed_keys: dict[str, dict[str, dict[str, JsonDict]]] = {} - for success, result in results: - if success: + for result in results: + if not isinstance(result, BaseException): claimed_keys.update(result[0]) missing.extend(result[1]) @@ -966,27 +956,25 @@ class ApplicationServicesHandler: continue # Query each service in parallel. - results = await make_deferred_yieldable( - defer.DeferredList( - [ - run_in_background( # type: ignore[call-overload] - self.appservice_api.query_keys, - # We know this must be an app service. - self.store.get_app_service_by_id(service_id), - service_query, - ) - for service_id, service_query in query_by_appservice.items() - ], - consumeErrors=True, - ) + results = await asyncio.gather( + *( + run_in_background( # type: ignore[call-overload] + self.appservice_api.query_keys, + # We know this must be an app service. + self.store.get_app_service_by_id(service_id), + service_query, + ) + for service_id, service_query in query_by_appservice.items() + ), + return_exceptions=True, ) # Patch together the results -- they are all independent (since they # require exclusive control over the users). They get returned as a single # dictionary. key_queries: dict[str, dict[str, dict[str, JsonDict]]] = {} - for success, result in results: - if success: + for result in results: + if not isinstance(result, BaseException): key_queries.update(result) return key_queries diff --git a/synapse/handlers/e2e_keys.py b/synapse/handlers/e2e_keys.py index 234d0aacd0..e9ecae2341 100644 --- a/synapse/handlers/e2e_keys.py +++ b/synapse/handlers/e2e_keys.py @@ -28,15 +28,12 @@ from signedjson.key import VerifyKey, decode_verify_key_bytes from signedjson.sign import SignatureVerifyException, verify_signed_json from unpaddedbase64 import decode_base64 -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio from synapse.api.constants import EduTypes from synapse.api.errors import CodeMessageException, Codes, NotFoundError, SynapseError from synapse.handlers.device import DeviceWriterHandler -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.logging.opentracing import log_kv, set_tag, tag_args, trace from synapse.types import ( JsonDict, @@ -809,14 +806,11 @@ class E2eKeysHandler: set_tag("error", True) set_tag("reason", str(failure)) - await make_deferred_yieldable( - defer.gatherResults( - [ - run_in_background(claim_client_keys, destination) - for destination in remote_queries - ], - consumeErrors=True, - ) + await asyncio.gather( + *( + run_in_background(claim_client_keys, destination) + for destination in remote_queries + ), ) logger.info( diff --git a/synapse/handlers/initial_sync.py b/synapse/handlers/initial_sync.py index 1e5e98a59b..006e13f6a0 100644 --- a/synapse/handlers/initial_sync.py +++ b/synapse/handlers/initial_sync.py @@ -34,7 +34,7 @@ from synapse.events.utils import SerializeEventConfig from synapse.events.validator import EventValidator from synapse.handlers.presence import format_user_presence_state from synapse.handlers.receipts import ReceiptEventSource -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.storage.roommember import RoomsForUser from synapse.streams.config import PaginationConfig from synapse.types import ( @@ -46,7 +46,6 @@ from synapse.types import ( StreamToken, UserID, ) -from synapse.util import unwrapFirstError from synapse.util.async_helpers import concurrently_execute, gather_results from synapse.util.caches.response_cache import ResponseCache from synapse.visibility import filter_and_transform_events_for_client @@ -199,31 +198,35 @@ class InitialSyncHandler: try: if event.membership == Membership.JOIN: room_end_token = now_token.room_key - deferred_room_state = run_in_background( - self._state_storage_controller.get_current_state, event.room_id - ) + + async def _get_current_state() -> dict: + return await self._state_storage_controller.get_current_state(event.room_id) + + deferred_room_state = run_in_background(_get_current_state) elif event.membership == Membership.LEAVE: room_end_token = RoomStreamToken( stream=event.event_pos.stream, ) - deferred_room_state = run_in_background( - self._state_storage_controller.get_state_for_events, - [event.event_id], - ).addCallback(lambda states: states[event.event_id]) - (messages, token), current_state = await make_deferred_yieldable( - gather_results( - ( - run_in_background( - self.store.get_recent_events_for_room, - event.room_id, - limit=pagin_config.limit, - end_token=room_end_token, - ), - deferred_room_state, + async def _get_state_for_event() -> dict: + states = await self._state_storage_controller.get_state_for_events( + [event.event_id], ) + return states[event.event_id] + + deferred_room_state = run_in_background(_get_state_for_event) + + (messages, token), current_state = await gather_results( + ( + run_in_background( + self.store.get_recent_events_for_room, + event.room_id, + limit=pagin_config.limit, + end_token=room_end_token, + ), + deferred_room_state, ) - ).addErrback(unwrapFirstError) + ) messages = await filter_and_transform_events_for_client( self._storage_controllers, @@ -480,20 +483,17 @@ class InitialSyncHandler: requester.user.to_string(), ) - presence, receipts, (messages, token) = await make_deferred_yieldable( - gather_results( - ( - run_in_background(get_presence), - run_in_background(get_receipts), - run_in_background( - self.store.get_recent_events_for_room, - room_id, - limit=pagin_config.limit, - end_token=now_token.room_key, - ), + presence, receipts, (messages, token) = await gather_results( + ( + run_in_background(get_presence), + run_in_background(get_receipts), + run_in_background( + self.store.get_recent_events_for_room, + room_id, + limit=pagin_config.limit, + end_token=now_token.room_key, ), - consumeErrors=True, - ).addErrback(unwrapFirstError) + ), ) messages = await filter_and_transform_events_for_client( diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 3229ff2bfa..4c8108da09 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -69,7 +69,7 @@ from synapse.events.validator import EventValidator from synapse.handlers.directory import DirectoryHandler from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME from synapse.logging import opentracing -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.replication.http.send_events import ReplicationSendEventsRestServlet from synapse.storage.databases.main.events_worker import EventRedactBehaviour from synapse.types import ( @@ -83,7 +83,7 @@ from synapse.types import ( create_requester, ) from synapse.types.state import StateFilter -from synapse.util import log_failure, unwrapFirstError +from synapse.util import log_failure from synapse.util.async_helpers import Linearizer, gather_results from synapse.util.caches.expiringcache import ExpiringCache from synapse.util.duration import Duration @@ -1579,10 +1579,14 @@ class EventCreationHandler: # We now persist the event (and update the cache in parallel, since we # don't want to block on it). - # - # Note: mypy gets confused if we inline dl and check with twisted#11770. - # Some kind of bug in mypy's deduction? - deferreds = ( + + async def _cache_joined_hosts() -> None: + try: + await self.cache_joined_hosts_for_events(events_and_context) + except Exception: + log_failure(None, "cache_joined_hosts_for_event failed") + + tasks = ( run_in_background( self._persist_events, requester=requester, @@ -1590,13 +1594,9 @@ class EventCreationHandler: ratelimit=ratelimit, extra_users=extra_users, ), - run_in_background( - self.cache_joined_hosts_for_events, events_and_context - ).addErrback(log_failure, "cache_joined_hosts_for_event failed"), + run_in_background(_cache_joined_hosts), ) - result, _ = await make_deferred_yieldable( - gather_results(deferreds, consumeErrors=True) - ).addErrback(unwrapFirstError) + result, _ = await gather_results(tasks) return result diff --git a/synapse/logging/context.py b/synapse/logging/context.py index a10cb04271..43a1aad309 100644 --- a/synapse/logging/context.py +++ b/synapse/logging/context.py @@ -841,10 +841,11 @@ def run_in_background( try: res = f(*args, **kwargs) except Exception as e: - # Return a coroutine that raises the exception - async def _raise() -> Any: - raise e - return _raise() + # Return a failed Task so callers can handle it asynchronously + loop = asyncio.get_event_loop() + future: asyncio.Future[Any] = loop.create_future() + future.set_exception(e) + return future if isinstance(res, typing.Coroutine): # Schedule the coroutine as a Task on the event loop. @@ -861,10 +862,11 @@ def run_in_background( res.add_done_callback(_reset) return res - # Plain value — wrap in a coroutine - async def _return_value() -> Any: - return res - return _return_value() + # Plain value — return a resolved Future so callers can await it + loop = asyncio.get_event_loop() + future: asyncio.Future[Any] = loop.create_future() + future.set_result(res) + return future def run_coroutine_in_background( diff --git a/synapse/notifier.py b/synapse/notifier.py index e8b30a0c89..f7759d5c7f 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -35,11 +35,7 @@ from typing import ( import attr from prometheus_client import Counter -try: - from twisted.internet import defer - from twisted.internet.defer import Deferred -except ImportError: - pass +import asyncio from synapse.api.constants import EduTypes, EventTypes, HistoryVisibility, Membership from synapse.api.errors import AuthError @@ -147,13 +143,13 @@ class _NotifierUserStream: self.last_notified_ms = time_now_ms # Set of listeners that we need to wake up when there has been a change. - self.listeners: set[Deferred[StreamToken]] = set() + self.listeners: set[asyncio.Future[StreamToken]] = set() def update_and_fetch_deferreds( self, current_token: StreamToken, time_now_ms: int, - ) -> Collection["Deferred[StreamToken]"]: + ) -> Collection["asyncio.Future[StreamToken]"]: """Update the stream for this user because of a new event from an event source, and return the set of deferreds to wake up. @@ -189,27 +185,35 @@ class _NotifierUserStream: def count_listeners(self) -> int: return len(self.listeners) - def new_listener(self, token: StreamToken) -> "Deferred[StreamToken]": - """Returns a deferred that is resolved when there is a new token + def new_listener(self, token: StreamToken) -> "asyncio.Future[StreamToken]": + """Returns a future that is resolved when there is a new token greater than the given token. Args: token: The token from which we are streaming from, i.e. we shouldn't notify for things that happened before this. """ + loop = asyncio.get_running_loop() + # Immediately wake up stream if something has already since happened # since their last token. if token != self.current_token: - return defer.succeed(self.current_token) + f: asyncio.Future[StreamToken] = loop.create_future() + f.set_result(self.current_token) + return f - # Create a new deferred and add it to the set of listeners. We add a + # Create a new future and add it to the set of listeners. We add a # cancel handler to remove it from the set again, to handle timeouts. - deferred: "Deferred[StreamToken]" = Deferred( - canceller=lambda d: self.listeners.discard(d) - ) - self.listeners.add(deferred) + future: asyncio.Future[StreamToken] = loop.create_future() + self.listeners.add(future) - return deferred + def _on_cancel(f: asyncio.Future[StreamToken]) -> None: + if f.cancelled(): + self.listeners.discard(f) + + future.add_done_callback(_on_cancel) + + return future @attr.s(slots=True, frozen=True, auto_attribs=True) @@ -370,7 +374,7 @@ class Notifier: time_now_ms = self.clock.time_msec() current_token = self.event_sources.get_current_token() - listeners: list["Deferred[StreamToken]"] = [] + listeners: list["asyncio.Future[StreamToken]"] = [] for user_stream in user_streams: try: listeners.extend( @@ -381,7 +385,8 @@ class Notifier: with PreserveLoggingContext(): for listener in listeners: - listener.callback(current_token) + if not listener.done(): + listener.set_result(current_token) users_woken_by_stream_counter.labels( stream=StreamKeyType.UN_PARTIAL_STATED_ROOMS, @@ -591,7 +596,7 @@ class Notifier: time_now_ms = self.clock.time_msec() current_token = self.event_sources.get_current_token() - listeners: list["Deferred[StreamToken]"] = [] + listeners: list["asyncio.Future[StreamToken]"] = [] for user_stream in user_streams: try: listeners.extend( @@ -602,13 +607,14 @@ class Notifier: except Exception: logger.exception("Failed to notify listener") - # We resolve all these deferreds in one go so that we only need to + # We resolve all these futures in one go so that we only need to # call `PreserveLoggingContext` once, as it has a bunch of overhead # (to calculate performance stats) if listeners: with PreserveLoggingContext(): for listener in listeners: - listener.callback(current_token) + if not listener.done(): + listener.set_result(current_token) if user_streams: users_woken_by_stream_counter.labels( diff --git a/synapse/push/bulk_push_rule_evaluator.py b/synapse/push/bulk_push_rule_evaluator.py index 681822aaa5..6650a912fb 100644 --- a/synapse/push/bulk_push_rule_evaluator.py +++ b/synapse/push/bulk_push_rule_evaluator.py @@ -26,16 +26,10 @@ from typing import ( Collection, Mapping, Sequence, - cast, ) from prometheus_client import Counter -try: - from twisted.internet.defer import Deferred -except ImportError: - pass - from synapse.api.constants import ( MAIN_TIMELINE, EventContentFields, @@ -47,7 +41,7 @@ from synapse.api.room_versions import PushRuleRoomFlag from synapse.event_auth import auth_types_for_event, get_user_power_level from synapse.events import EventBase, relation_from_event from synapse.events.snapshot import EventContext, EventPersistencePair -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.metrics import SERVER_NAME_LABEL from synapse.state import CREATE_KEY, POWER_KEY from synapse.storage.databases.main.roommember import EventIdMembership @@ -56,7 +50,6 @@ from synapse.storage.roommember import ProfileInfo from synapse.synapse_rust.push import FilteredPushRules, PushRuleEvaluator from synapse.types import JsonValue from synapse.types.state import StateFilter -from synapse.util import unwrapFirstError from synapse.util.async_helpers import gather_results from synapse.util.caches import register_cache from synapse.util.metrics import measure_func @@ -404,31 +397,25 @@ class BulkPushRuleEvaluator: (power_levels, sender_power_level), related_events, profiles, - ) = await make_deferred_yieldable( - cast( - "Deferred[tuple[int, tuple[dict, int | None], dict[str, dict[str, JsonValue]], Mapping[str, ProfileInfo]]]", - gather_results( - ( - run_in_background( # type: ignore[call-overload] - self.store.get_number_joined_users_in_room, - event.room_id, # type: ignore[arg-type] - ), - run_in_background( - self._get_power_levels_and_sender_level, - event, - context, - event_id_to_event, - ), - run_in_background(self._related_events, event), - run_in_background( # type: ignore[call-overload] - self.store.get_subset_users_in_room_with_profiles, - event.room_id, - rules_by_user.keys(), - ), - ), - consumeErrors=True, - ).addErrback(unwrapFirstError), - ) + ) = await gather_results( + ( + run_in_background( # type: ignore[call-overload] + self.store.get_number_joined_users_in_room, + event.room_id, # type: ignore[arg-type] + ), + run_in_background( + self._get_power_levels_and_sender_level, + event, + context, + event_id_to_event, + ), + run_in_background(self._related_events, event), + run_in_background( # type: ignore[call-overload] + self.store.get_subset_users_in_room_with_profiles, + event.room_id, + rules_by_user.keys(), + ), + ), ) # Find the event's thread ID. diff --git a/synapse/rest/client/transactions.py b/synapse/rest/client/transactions.py index 32a3d43ee9..34c01950f4 100644 --- a/synapse/rest/client/transactions.py +++ b/synapse/rest/client/transactions.py @@ -27,9 +27,9 @@ from typing import TYPE_CHECKING, Awaitable, Callable, Hashable from typing_extensions import ParamSpec +import asyncio + try: - from twisted.internet.defer import Deferred - from twisted.python.failure import Failure from twisted.web.iweb import IRequest except ImportError: pass @@ -136,12 +136,11 @@ class HttpTransactionCache: # if the request fails with an exception, remove it # from the transaction map. This is done to ensure that we don't # cache transient errors like rate-limiting errors, etc. - def remove_from_map(err: Failure) -> None: - self.transactions.pop(txn_key, None) - # we deliberately do not propagate the error any further, as we - # expect the observers to have reported it. + def remove_from_map(t: asyncio.Task) -> None: + if t.cancelled() or t.exception() is not None: + self.transactions.pop(txn_key, None) - deferred.addErrback(remove_from_map) + deferred.add_done_callback(remove_from_map) return make_deferred_yieldable(observable.observe()) diff --git a/synapse/storage/controllers/persist_events.py b/synapse/storage/controllers/persist_events.py index 5428fcc436..bee8efd4ba 100644 --- a/synapse/storage/controllers/persist_events.py +++ b/synapse/storage/controllers/persist_events.py @@ -40,10 +40,7 @@ from typing import ( import attr from prometheus_client import Counter, Histogram -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio from synapse.api.constants import EventTypes, Membership from synapse.events import EventBase @@ -232,8 +229,9 @@ class _EventPeristenceQueue(Generic[_PersistResult]): # the new task has been merged into the last task in the queue end_item = queue[-1] else: + loop = asyncio.get_running_loop() deferred: ObservableDeferred[_PersistResult] = ObservableDeferred( - defer.Deferred(), consumeErrors=True + loop.create_future(), consumeErrors=True ) end_item = _EventPersistQueueItem( diff --git a/synapse/storage/databases/main/events_worker.py b/synapse/storage/databases/main/events_worker.py index a40a7745b7..002dabf766 100644 --- a/synapse/storage/databases/main/events_worker.py +++ b/synapse/storage/databases/main/events_worker.py @@ -38,10 +38,7 @@ from typing import ( import attr from prometheus_client import Gauge -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio from synapse.api.constants import Direction, EventTypes from synapse.api.errors import NotFoundError, SynapseError @@ -89,7 +86,6 @@ from synapse.storage.util.sequence import build_sequence_generator from synapse.types import JsonDict, get_domain_from_id from synapse.types.state import StateFilter from synapse.types.storage import _BackgroundUpdates -from synapse.util import unwrapFirstError from synapse.util.async_helpers import ObservableDeferred, delay_cancellation from synapse.util.caches.descriptors import cached, cachedList from synapse.util.caches.lrucache import AsyncLruCache @@ -316,7 +312,7 @@ class EventsWorkerStore(SQLBaseStore): self._event_fetch_lock = threading.Condition() self._event_fetch_list: list[ - tuple[Iterable[str], "defer.Deferred[dict[str, _EventRow]]"] + tuple[Iterable[str], "asyncio.Future[dict[str, _EventRow]]"] ] = [] self._event_fetch_ongoing = 0 event_fetch_ongoing_gauge.labels(**{SERVER_NAME_LABEL: self.server_name}).set( @@ -873,8 +869,10 @@ class EventsWorkerStore(SQLBaseStore): # to all the events we pulled from the DB (this will result in this # function returning more events than requested, but that can happen # already due to `_get_events_from_db`). + loop = asyncio.get_event_loop() + _fetching_future: asyncio.Future[dict[str, EventCacheEntry]] = loop.create_future() fetching_deferred: ObservableDeferred[dict[str, EventCacheEntry]] = ( - ObservableDeferred(defer.Deferred(), consumeErrors=True) + ObservableDeferred(_fetching_future, consumeErrors=True) ) for event_id in missing_events_ids: self._current_event_fetches[event_id] = fetching_deferred @@ -897,7 +895,8 @@ class EventsWorkerStore(SQLBaseStore): missing_events.update(db_missing_events) except Exception as e: with PreserveLoggingContext(): - fetching_deferred.errback(e) + if not _fetching_future.done(): + _fetching_future.set_exception(e) raise e finally: # Ensure that we mark these events as no longer being fetched. @@ -905,7 +904,8 @@ class EventsWorkerStore(SQLBaseStore): self._current_event_fetches.pop(event_id, None) with PreserveLoggingContext(): - fetching_deferred.callback(missing_events) + if not _fetching_future.done(): + _fetching_future.set_result(missing_events) return missing_events @@ -920,12 +920,9 @@ class EventsWorkerStore(SQLBaseStore): if already_fetching_deferreds: # Wait for the other event requests to finish and add their results # to ours. - results = await make_deferred_yieldable( - defer.gatherResults( - (d.observe() for d in already_fetching_deferreds), - consumeErrors=True, - ) - ).addErrback(unwrapFirstError) + results = await asyncio.gather( + *(d.observe() for d in already_fetching_deferreds), + ) for result in results: # We filter out events that we haven't asked for as we might get @@ -1234,8 +1231,9 @@ class EventsWorkerStore(SQLBaseStore): # Fail any outstanding fetches since no one else will handle them. assert exc is not None with PreserveLoggingContext(): - for _, deferred in event_fetches_to_fail: - deferred.errback(exc) + for _, future in event_fetches_to_fail: + if not future.done(): + future.set_exception(exc) def _fetch_loop(self, conn: LoggingDatabaseConnection) -> None: """Takes a database connection and waits for requests for events from @@ -1269,7 +1267,7 @@ class EventsWorkerStore(SQLBaseStore): def _fetch_event_list( self, conn: LoggingDatabaseConnection, - event_list: list[tuple[Iterable[str], "defer.Deferred[dict[str, _EventRow]]"]], + event_list: list[tuple[Iterable[str], "asyncio.Future[dict[str, _EventRow]]"]], ) -> None: """Handle a load of requests from the _event_fetch_list queue @@ -1303,23 +1301,27 @@ class EventsWorkerStore(SQLBaseStore): events_to_fetch, ) - # We only want to resolve deferreds from the main thread + # We only want to resolve futures from the main thread + loop = asyncio.get_event_loop() + def fire() -> None: for _, d in event_list: - d.callback(row_dict) + if not d.done(): + d.set_result(row_dict) - with PreserveLoggingContext(): - self.hs.get_reactor().callFromThread(fire) + loop.call_soon_threadsafe(fire) except Exception as e: logger.exception("do_fetch") - # We only want to resolve deferreds from the main thread + # We only want to resolve futures from the main thread + loop = asyncio.get_event_loop() + def fire_errback(exc: Exception) -> None: for _, d in event_list: - d.errback(exc) + if not d.done(): + d.set_exception(exc) - with PreserveLoggingContext(): - self.hs.get_reactor().callFromThread(fire_errback, e) + loop.call_soon_threadsafe(fire_errback, e) @trace async def _get_events_from_db( @@ -1544,7 +1546,8 @@ class EventsWorkerStore(SQLBaseStore): that weren't requested. """ - events_d: "defer.Deferred[dict[str, _EventRow]]" = defer.Deferred() + loop = asyncio.get_running_loop() + events_d: asyncio.Future[dict[str, _EventRow]] = loop.create_future() with self._event_fetch_lock: self._event_fetch_list.append((events, events_d)) self._event_fetch_lock.notify() diff --git a/synapse/storage/databases/main/push_rule.py b/synapse/storage/databases/main/push_rule.py index c0f692dab5..2ebd608f25 100644 --- a/synapse/storage/databases/main/push_rule.py +++ b/synapse/storage/databases/main/push_rule.py @@ -29,14 +29,9 @@ from typing import ( cast, ) -try: - from twisted.internet import defer -except ImportError: - pass - from synapse.api.errors import StoreError from synapse.config.homeserver import ExperimentalConfig -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.replication.tcp.streams import PushRulesStream from synapse.storage._base import SQLBaseStore from synapse.storage.database import ( @@ -54,7 +49,6 @@ from synapse.storage.push_rule import InconsistentRuleException, RuleNotFoundExc from synapse.storage.util.id_generators import IdGenerator, MultiWriterIdGenerator from synapse.synapse_rust.push import FilteredPushRules, PushRule, PushRules from synapse.types import JsonDict -from synapse.util import unwrapFirstError from synapse.util.async_helpers import gather_results from synapse.util.caches.descriptors import cached, cachedList from synapse.util.caches.stream_change_cache import StreamChangeCache @@ -273,33 +267,27 @@ class PushRulesWorkerStore( user_id: [] for user_id in user_ids } - # gatherResults loses all type information. - rows, enabled_map_by_user = await make_deferred_yieldable( - gather_results( - ( - cast( - Any, - run_in_background( - self.db_pool.simple_select_many_batch, - table="push_rules", - column="user_name", - iterable=user_ids, - retcols=( - "user_name", - "rule_id", - "priority_class", - "priority", - "conditions", - "actions", - ), - desc="bulk_get_push_rules", - batch_size=1000, - ), + # gather_results loses all type information. + rows, enabled_map_by_user = await gather_results( + ( + run_in_background( + self.db_pool.simple_select_many_batch, + table="push_rules", + column="user_name", + iterable=user_ids, + retcols=( + "user_name", + "rule_id", + "priority_class", + "priority", + "conditions", + "actions", ), - run_in_background(self.bulk_get_push_rules_enabled, user_ids), + desc="bulk_get_push_rules", + batch_size=1000, ), - consumeErrors=True, - ).addErrback(unwrapFirstError) + run_in_background(self.bulk_get_push_rules_enabled, user_ids), + ), ) # Sort by highest priority_class, then highest priority. diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 822cd01c2f..34a70c8fce 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -58,15 +58,12 @@ from typing import ( import attr from typing_extensions import assert_never -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio from synapse.api.constants import Direction, EventTypes, Membership from synapse.api.filtering import Filter from synapse.events import EventBase -from synapse.logging.context import make_deferred_yieldable, run_in_background +from synapse.logging.context import run_in_background from synapse.logging.opentracing import tag_args, trace from synapse.storage._base import SQLBaseStore from synapse.storage.database import ( @@ -704,21 +701,18 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore): results = {} room_ids = list(room_ids) for rm_ids in (room_ids[i : i + 20] for i in range(0, len(room_ids), 20)): - res = await make_deferred_yieldable( - defer.gatherResults( - [ - run_in_background( - self.paginate_room_events_by_stream_ordering, - room_id=room_id, - from_key=from_key, - to_key=to_key, - direction=direction, - limit=limit, - ) - for room_id in rm_ids - ], - consumeErrors=True, - ) + res = await asyncio.gather( + *( + run_in_background( + self.paginate_room_events_by_stream_ordering, + room_id=room_id, + from_key=from_key, + to_key=to_key, + direction=direction, + limit=limit, + ) + for room_id in rm_ids + ), ) results.update(dict(zip(rm_ids, res))) diff --git a/synapse/storage/util/partial_state_events_tracker.py b/synapse/storage/util/partial_state_events_tracker.py index 3021832ae0..3fa7c36add 100644 --- a/synapse/storage/util/partial_state_events_tracker.py +++ b/synapse/storage/util/partial_state_events_tracker.py @@ -19,21 +19,15 @@ # # +import asyncio import logging from collections import defaultdict from typing import Collection -try: - from twisted.internet import defer - from twisted.internet.defer import Deferred -except ImportError: - pass - -from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable +from synapse.logging.context import PreserveLoggingContext from synapse.logging.opentracing import trace_with_opname from synapse.storage.databases.main.events_worker import EventsWorkerStore from synapse.storage.databases.main.room import RoomWorkerStore -from synapse.util import unwrapFirstError from synapse.util.cancellation import cancellable logger = logging.getLogger(__name__) @@ -44,9 +38,9 @@ class PartialStateEventsTracker: def __init__(self, store: EventsWorkerStore): self._store = store - # a map from event id to a set of Deferreds which are waiting for that event to be + # a map from event id to a set of Futures which are waiting for that event to be # un-partial-stated. - self._observers: dict[str, set[Deferred[None]]] = defaultdict(set) + self._observers: dict[str, set[asyncio.Future[None]]] = defaultdict(set) def notify_un_partial_stated(self, event_id: str) -> None: """Notify that we now have full state for a given event @@ -68,7 +62,8 @@ class PartialStateEventsTracker: ) with PreserveLoggingContext(): for o in observers: - o.callback(None) + if not o.done(): + o.set_result(None) @trace_with_opname("PartialStateEventsTracker.await_full_state") @cancellable @@ -96,8 +91,9 @@ class PartialStateEventsTracker: ) # create an observer for each lazy-joined event - observers: dict[str, Deferred[None]] = { - event_id: Deferred() for event_id in partial_state_event_ids + loop = asyncio.get_event_loop() + observers: dict[str, asyncio.Future[None]] = { + event_id: loop.create_future() for event_id in partial_state_event_ids } for event_id, observer in observers.items(): self._observers[event_id].add(observer) @@ -111,15 +107,10 @@ class PartialStateEventsTracker: ).items(): # there may have been a call to notify_un_partial_stated during the # db query, so the observers may already have been called. - if not partial and not observers[event_id].called: - observers[event_id].callback(None) + if not partial and not observers[event_id].done(): + observers[event_id].set_result(None) - await make_deferred_yieldable( - defer.gatherResults( - observers.values(), - consumeErrors=True, - ) - ).addErrback(unwrapFirstError) + await asyncio.gather(*observers.values()) logger.info("Events %s all un-partial-stated", observers.keys()) finally: # remove any observers we created. This should happen when the notification @@ -141,9 +132,9 @@ class PartialCurrentStateTracker: def __init__(self, store: RoomWorkerStore): self._store = store - # a map from room id to a set of Deferreds which are waiting for that room to be + # a map from room id to a set of Futures which are waiting for that room to be # un-partial-stated. - self._observers: dict[str, set[Deferred[None]]] = defaultdict(set) + self._observers: dict[str, set[asyncio.Future[None]]] = defaultdict(set) def notify_un_partial_stated(self, room_id: str) -> None: """Notify that we now have full current state for a given room @@ -163,14 +154,16 @@ class PartialCurrentStateTracker: ) with PreserveLoggingContext(): for o in observers: - o.callback(None) + if not o.done(): + o.set_result(None) @trace_with_opname("PartialCurrentStateTracker.await_full_state") @cancellable async def await_full_state(self, room_id: str) -> None: - # We add the deferred immediately so that the DB call to check for + # We add the future immediately so that the DB call to check for # partial state doesn't race when we unpartial the room. - d: Deferred[None] = Deferred() + loop = asyncio.get_event_loop() + d: asyncio.Future[None] = loop.create_future() self._observers.setdefault(room_id, set()).add(d) try: @@ -185,7 +178,7 @@ class PartialCurrentStateTracker: stack_info=True, ) - await make_deferred_yieldable(d) + await d logger.info("Room has un-partial-stated") finally: diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py index 8ea0150767..d8f08e7fbd 100644 --- a/synapse/util/async_helpers.py +++ b/synapse/util/async_helpers.py @@ -421,6 +421,24 @@ class ObservableFuture(Generic[_T]): raise ValueError(f"{self!r} has no result yet") return self._result[1] + def callback(self, result: _T) -> None: + """Resolve the underlying future with a result (Deferred compat).""" + self._future.set_result(result) + + def errback(self, exc: BaseException | None = None) -> None: + """Reject the underlying future with an exception (Deferred compat). + + If no exception is given, uses the current exception from sys.exc_info(). + """ + if exc is None: + import sys + + _, exc_val, _ = sys.exc_info() + if exc_val is None: + raise RuntimeError("errback() called with no active exception") + exc = exc_val + self._future.set_exception(exc) + class NativeLinearizer: """asyncio-native equivalent of Linearizer. diff --git a/synapse/util/batching_queue.py b/synapse/util/batching_queue.py index b9e603a7df..28f4bae18c 100644 --- a/synapse/util/batching_queue.py +++ b/synapse/util/batching_queue.py @@ -31,12 +31,9 @@ from typing import ( from prometheus_client import Gauge -try: - from twisted.internet import defer -except ImportError: - pass +import asyncio -from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable +from synapse.logging.context import PreserveLoggingContext from synapse.metrics import SERVER_NAME_LABEL from synapse.util.clock import Clock from synapse.util.duration import Duration @@ -149,15 +146,10 @@ class BatchingQueue(Generic[V, R]): key. """ - # First we create a defer and add it and the value to the list of + # First we create a future and add it and the value to the list of # pending items. - import asyncio as _aq - try: - loop = _aq.get_running_loop() - d: Any = loop.create_future() - except RuntimeError: - # No running loop — use Twisted Deferred as fallback - d = defer.Deferred() + loop = asyncio.get_running_loop() + d: asyncio.Future[V] = loop.create_future() self._next_values.setdefault(key, []).append((value, d)) # If we're not currently processing the key fire off a background @@ -166,7 +158,7 @@ class BatchingQueue(Generic[V, R]): self.hs.run_as_background_process(self._name, self._process_queue, key) with self._number_in_flight_metric.track_inprogress(): - return await make_deferred_yieldable(d) + return await d async def _process_queue(self, key: Hashable) -> None: """A background task to repeatedly pull things off the queue for the @@ -197,16 +189,16 @@ class BatchingQueue(Generic[V, R]): results = await self._process_batch_callback(values) with PreserveLoggingContext(): - for _, deferred in next_values: - deferred.callback(results) + for _, future in next_values: + future.set_result(results) except Exception as e: with PreserveLoggingContext(): - for _, deferred in next_values: - if deferred.called: + for _, future in next_values: + if future.done(): continue - deferred.errback(e) + future.set_exception(e) finally: self._processing_keys.discard(key) diff --git a/synapse/util/caches/cached_call.py b/synapse/util/caches/cached_call.py index 546d6b6dce..04a85128d4 100644 --- a/synapse/util/caches/cached_call.py +++ b/synapse/util/caches/cached_call.py @@ -18,15 +18,10 @@ # [This file includes modifications made by New Vector Limited] # # +import asyncio import enum from typing import Awaitable, Callable, Generic, TypeVar -try: - from twisted.internet.defer import Deferred - from twisted.python.failure import Failure -except ImportError: - pass - from synapse.logging.context import make_deferred_yieldable, run_in_background TV = TypeVar("TV") @@ -69,7 +64,7 @@ class CachedCall(Generic[TV]): CachedCall object. """ - __slots__ = ["_callable", "_deferred", "_result"] + __slots__ = ["_callable", "_task", "_result", "_exception"] def __init__(self, f: Callable[[], Awaitable[TV]]): """ @@ -78,28 +73,33 @@ class CachedCall(Generic[TV]): at once (per instance of CachedCall) """ self._callable: Callable[[], Awaitable[TV]] | None = f - self._deferred: Deferred | None = None - self._result: _Sentinel | TV | Failure = _Sentinel.sentinel + self._task: asyncio.Task | None = None + self._result: _Sentinel | TV = _Sentinel.sentinel + self._exception: BaseException | None = None async def get(self) -> TV: """Kick off the call if necessary, and return the result""" # Fire off the callable now if this is our first time - if not self._deferred: + if not self._task: assert self._callable is not None - self._deferred = run_in_background(self._callable) + self._task = run_in_background(self._callable) # we will never need the callable again, so make sure it can be GCed self._callable = None - # once the deferred completes, store the result. We cannot simply leave the - # result in the deferred, since `awaiting` a deferred destroys its result. - # (Also, if it's a Failure, GCing the deferred would log a critical error - # about unhandled Failures) - def got_result(r: TV | Failure) -> None: - self._result = r + # once the task completes, store the result. + def got_result(t: asyncio.Task) -> None: + if t.cancelled(): + self._exception = asyncio.CancelledError() + return + exc = t.exception() + if exc is not None: + self._exception = exc + else: + self._result = t.result() - self._deferred.addBoth(got_result) + self._task.add_done_callback(got_result) # TODO: consider cancellation semantics. Currently, if the call to get() # is cancelled, the underlying call will continue (and any future calls @@ -107,15 +107,14 @@ class CachedCall(Generic[TV]): # the fact the underlying call may be logged to a cancelled logcontext, # and any eventual exception may not be reported. - # we can now await the deferred, and once it completes, return the result. - if isinstance(self._result, _Sentinel): - await make_deferred_yieldable(self._deferred) - assert not isinstance(self._result, _Sentinel) + # we can now await the task, and once it completes, return the result. + if isinstance(self._result, _Sentinel) and self._exception is None: + await make_deferred_yieldable(self._task) - if isinstance(self._result, Failure): - self._result.raiseException() - raise AssertionError("unexpected return from Failure.raiseException") + if self._exception is not None: + raise self._exception + assert not isinstance(self._result, _Sentinel) return self._result diff --git a/synapse/util/caches/response_cache.py b/synapse/util/caches/response_cache.py index aaa1cd326e..63bedd294e 100644 --- a/synapse/util/caches/response_cache.py +++ b/synapse/util/caches/response_cache.py @@ -31,11 +31,6 @@ from typing import ( import attr -try: - from twisted.internet import defer -except ImportError: - pass - from synapse.logging.context import make_deferred_yieldable, run_in_background from synapse.logging.opentracing import ( active_span, @@ -237,7 +232,7 @@ class ResponseCache(Generic[KV]): ) self._result_cache[key] = entry - def on_complete(r: RV) -> RV: + def on_complete(f: Any) -> None: # if this cache has a non-zero timeout, and the callback has not cleared # the should_cache bit, we leave it in the cache for now and schedule # its removal later. @@ -256,12 +251,11 @@ class ResponseCache(Generic[KV]): else: # otherwise, remove the result immediately. self.unset(key) - return r # make sure we do this *after* adding the entry to result_cache, # in case the result is already complete (in which case flipping the order would # leave us with a stuck entry in the cache). - result.addBoth(on_complete) + deferred.add_done_callback(on_complete) return entry def unset(self, key: KV) -> None: @@ -388,7 +382,7 @@ class ResponseCache(Generic[KV]): result = entry.result.observe() if self._enable_logging: - if result.called: + if entry.result.has_called(): logger.info( "[%s]: using completed cached result for [%s]", self._name, key ) diff --git a/synapse/util/clock.py b/synapse/util/clock.py index dd47d94276..0c931e4693 100644 --- a/synapse/util/clock.py +++ b/synapse/util/clock.py @@ -629,6 +629,11 @@ class NativeLoopingCall: def __init__(self, task: "asyncio.Task[None]") -> None: self._task = task + @property + def running(self) -> bool: + """Whether the looping call is still running.""" + return not self._task.done() + def stop(self) -> None: """Stop the looping call.""" self._task.cancel() @@ -725,7 +730,8 @@ class NativeClock: # Pending sleeps: list of (wake_time, future) import heapq self._fake_time: float = time_mod.time() - self._pending_sleeps: list[tuple[float, asyncio.Future]] = [] + self._pending_sleeps: list[tuple[float, int, asyncio.Future]] = [] + self._sleep_seq = 0 # tiebreaker for heapq when wake_times are equal self._use_fake_time = False # Set to True by tests def _get_loop(self) -> asyncio.AbstractEventLoop: @@ -758,7 +764,8 @@ class NativeClock: loop = self._get_loop() future: asyncio.Future[None] = loop.create_future() wake_time = self._fake_time + duration.as_secs() - heapq.heappush(self._pending_sleeps, (wake_time, future)) + self._sleep_seq += 1 + heapq.heappush(self._pending_sleeps, (wake_time, self._sleep_seq, future)) await future else: await asyncio.sleep(duration.as_secs()) @@ -774,7 +781,7 @@ class NativeClock: # Fire any sleeps that are now due while self._pending_sleeps and self._pending_sleeps[0][0] <= self._fake_time: import heapq - _, future = heapq.heappop(self._pending_sleeps) + _, _, future = heapq.heappop(self._pending_sleeps) if not future.done(): future.set_result(None) diff --git a/tests/server.py b/tests/server.py index ff9559892f..9ff0b16424 100644 --- a/tests/server.py +++ b/tests/server.py @@ -156,6 +156,7 @@ class FakeChannel: _producer: Optional[Union[IPullProducer, IPushProducer]] = None resource_usage: ContextResourceUsage | None = None _request: Request | None = None + _clock: Any = None # NativeClock, for advancing fake time @property def request(self) -> Request: @@ -319,10 +320,21 @@ class FakeChannel: if _time.monotonic() > deadline: raise TimedOutException("Timed out waiting for request to finish.") + # Advance NativeClock fake time (fires pending sleeps) + if self._clock is not None: + self._clock.advance(0.0) + self._reactor.advance(0.1) # Drive asyncio event loop for DB operations, task completions, etc. - if not loop.is_closed() and not loop.is_running(): - loop.run_until_complete(asyncio.sleep(0)) + # Use a small real sleep to allow thread pool callbacks to be delivered. + if not loop.is_closed(): + async def _drain() -> None: + """Run multiple event loop ticks to drain pending work.""" + for _ in range(20): + await asyncio.sleep(0.001) + if self._clock is not None: + self._clock.advance(0.0) + loop.run_until_complete(_drain()) def extract_cookies(self, cookies: MutableMapping[str, str]) -> None: """Process the contents of any Set-Cookie headers in the response @@ -384,6 +396,7 @@ def make_request( await_result: bool = True, custom_headers: Iterable[CustomHeaderType] | None = None, client_ip: str = "127.0.0.1", + clock: Any = None, ) -> FakeChannel: """ Make a web request using the given method, path and content, and render it @@ -440,7 +453,7 @@ def make_request( if isinstance(content, str): content = content.encode("utf8") - channel = FakeChannel(site, reactor, ip=client_ip) + channel = FakeChannel(site, reactor, ip=client_ip, clock=clock) req = request( channel, diff --git a/tests/unittest.py b/tests/unittest.py index 24385ef815..94af5a4ff8 100644 --- a/tests/unittest.py +++ b/tests/unittest.py @@ -803,6 +803,7 @@ class HomeserverTestCase(TestCase): await_result, custom_headers, client_ip, + clock=self.clock, ) def setup_test_homeserver(