Summary of where we are:

- Login tests: 50/50 pass
  - Room tests: 119/173 pass (76%), with failures from:
    - 2 tests checking resource_usage (missing from shim)
    - 4 cancellation tests using Twisted Deferred
    - 1 ratelimit test (needs time advancement fix)
    - Several tests with missing await in test files imported from other modules
    - Some tests with actual logic differences (e.g., member list permissions)
    - MSC4293 tests failing (federated test infrastructure)

  The core infrastructure works — IsolatedAsyncioTestCase properly drives async tests, room creation works, DB operations work, cache invalidation works. The remaining failures are mechanical (adding await to more test files)
   or specific test features that need porting.

synapse/http/federation/srv_resolver.py

@@ -189,7 +189,7 @@ class SrvResolver:
 
     async def _lookup_dnspython(self, name: str) -> list[dict]:
         """Use dnspython to resolve SRV records."""
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
 
         def _resolve() -> list[dict]:
             try:

synapse/logging/context.py

@@ -854,7 +854,7 @@ def run_in_background(
         res = f(*args, **kwargs)
     except Exception as e:
         # Return a failed Task so callers can handle it asynchronously
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         future: asyncio.Future[Any] = loop.create_future()
         future.set_exception(e)
         return future
@@ -863,7 +863,7 @@ def run_in_background(
         # Schedule the coroutine as a Task on the event loop.
         # This ensures fire-and-forget callers actually run the coroutine.
         coro = run_coroutine_in_background(res)
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         return loop.create_task(coro)
 
     if isinstance(res, (asyncio.Task, asyncio.Future)):
@@ -875,7 +875,7 @@ def run_in_background(
         return res
 
     # Plain value — return a resolved Future so callers can await it
-    loop = asyncio.get_event_loop()
+    loop = asyncio.get_running_loop()
     future: asyncio.Future[Any] = loop.create_future()
     future.set_result(res)
     return future

synapse/storage/databases/main/events_worker.py

@@ -869,7 +869,7 @@ 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()
+                loop = asyncio.get_running_loop()
                 _fetching_future: asyncio.Future[dict[str, EventCacheEntry]] = loop.create_future()
                 fetching_deferred: ObservableDeferred[dict[str, EventCacheEntry]] = (
                     ObservableDeferred(_fetching_future, consumeErrors=True)
@@ -1167,6 +1167,9 @@ class EventsWorkerStore(SQLBaseStore):
 
     async def _fetch_thread(self) -> None:
         """Services requests for events from `_event_fetch_list`."""
+        # Capture the event loop here (in the async context) so that
+        # _fetch_loop/_fetch_event_list can use it from the DB thread.
+        self._fetch_event_loop = asyncio.get_running_loop()
         exc = None
         try:
             await self.db_pool.runWithConnection(self._fetch_loop)
@@ -1302,7 +1305,7 @@ class EventsWorkerStore(SQLBaseStore):
                 )
 
                 # We only want to resolve futures from the main thread
-                loop = asyncio.get_event_loop()
+                loop = self._fetch_event_loop
 
                 def fire() -> None:
                     for _, d in event_list:
@@ -1314,7 +1317,7 @@ class EventsWorkerStore(SQLBaseStore):
                 logger.exception("do_fetch")
 
                 # We only want to resolve futures from the main thread
-                loop = asyncio.get_event_loop()
+                loop = self._fetch_event_loop
 
                 def fire_errback(exc: Exception) -> None:
                     for _, d in event_list:

synapse/storage/native_database.py

@@ -176,7 +176,7 @@ class NativeConnectionPool:
             return func(conn, *args, **kwargs)
 
         # Run in thread pool via asyncio
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         return await loop.run_in_executor(self._executor, _inner)
 
     async def runInteraction(
@@ -212,7 +212,7 @@ class NativeConnectionPool:
                 raise
 
         # Run in thread pool via asyncio
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         return await loop.run_in_executor(self._executor, _inner)
 
     def close(self) -> None:

synapse/storage/util/partial_state_events_tracker.py

@@ -91,7 +91,7 @@ class PartialStateEventsTracker:
         )
 
         # create an observer for each lazy-joined event
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         observers: dict[str, asyncio.Future[None]] = {
             event_id: loop.create_future() for event_id in partial_state_event_ids
         }
@@ -162,7 +162,7 @@ class PartialCurrentStateTracker:
     async def await_full_state(self, room_id: str) -> None:
         # We add the future immediately so that the DB call to check for
         # partial state doesn't race when we unpartial the room.
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         d: asyncio.Future[None] = loop.create_future()
         self._observers.setdefault(room_id, set()).add(d)
 

synapse/util/caches/descriptors.py

@@ -291,7 +291,7 @@ class DeferredCacheDescriptor(_CacheDescriptorBase):
                     ret = _asyncio.ensure_future(result.__await__())
                 else:
                     # Plain value — wrap in resolved future
-                    loop = _asyncio.get_event_loop()
+                    loop = _asyncio.get_running_loop()
                     f = loop.create_future()
                     f.set_result(result)
                     ret = f

synapse/util/caches/future_cache.py

@@ -312,7 +312,7 @@ class FutureCache(Generic[VT]):
 
         Returns a FutureCacheEntry that can be resolved with the results.
         """
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         future: asyncio.Future[dict[Hashable, VT]] = loop.create_future()
 
         entry = FutureCacheEntry(future)

synapse/util/clock.py

@@ -174,10 +174,7 @@ class NativeClock:
 
     def _get_loop(self) -> asyncio.AbstractEventLoop:
         if self._loop is None:
-            try:
-                self._loop = asyncio.get_running_loop()
-            except RuntimeError:
-                self._loop = asyncio.get_event_loop()
+            self._loop = asyncio.get_running_loop()
         return self._loop
 
     def shutdown(self) -> None:
@@ -281,7 +278,7 @@ class NativeClock:
 
                 await self.sleep(Duration(seconds=interval))
 
-        loop = asyncio.get_event_loop()
+        loop = asyncio.get_running_loop()
         task_obj = loop.create_task(_loop())
         call = NativeLoopingCall(task_obj)
         self._looping_calls.add(call)

tests/storage/test_stream.py

@@ -75,37 +75,37 @@ class PaginationTestCase(HomeserverTestCase):
         config["experimental_features"] = {"msc3874_enabled": True}
         return config
 
-    def prepare(
+    async def prepare(
         self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer
     ) -> None:
-        self.user_id = self.register_user("test", "test")
-        self.tok = self.login("test", "test")
-        self.room_id = self.helper.create_room_as(self.user_id, tok=self.tok)
+        self.user_id = await self.register_user("test", "test")
+        self.tok = await self.login("test", "test")
+        self.room_id = await self.helper.create_room_as(self.user_id, tok=self.tok)
 
-        self.second_user_id = self.register_user("second", "test")
-        self.second_tok = self.login("second", "test")
-        self.helper.join(
+        self.second_user_id = await self.register_user("second", "test")
+        self.second_tok = await self.login("second", "test")
+        await self.helper.join(
             room=self.room_id, user=self.second_user_id, tok=self.second_tok
         )
 
-        self.third_user_id = self.register_user("third", "test")
-        self.third_tok = self.login("third", "test")
-        self.helper.join(room=self.room_id, user=self.third_user_id, tok=self.third_tok)
+        self.third_user_id = await self.register_user("third", "test")
+        self.third_tok = await self.login("third", "test")
+        await self.helper.join(room=self.room_id, user=self.third_user_id, tok=self.third_tok)
 
         # Store a token which is after all the room creation events.
-        self.from_token = self.get_success(
+        self.from_token = await self.get_success(
             self.hs.get_event_sources().get_current_token_for_pagination(self.room_id)
         )
 
         # An initial event with a relation from second user.
-        res = self.helper.send_event(
+        res = await self.helper.send_event(
             room_id=self.room_id,
             type=EventTypes.Message,
             content={"msgtype": "m.text", "body": "Message 1"},
             tok=self.tok,
         )
         self.event_id_1 = res["event_id"]
-        res = self.helper.send_event(
+        res = await self.helper.send_event(
             room_id=self.room_id,
             type="m.reaction",
             content={
@@ -120,14 +120,14 @@ class PaginationTestCase(HomeserverTestCase):
         self.event_id_annotation = res["event_id"]
 
         # Another event with a relation from third user.
-        res = self.helper.send_event(
+        res = await self.helper.send_event(
             room_id=self.room_id,
             type=EventTypes.Message,
             content={"msgtype": "m.text", "body": "Message 2"},
             tok=self.tok,
         )
         self.event_id_2 = res["event_id"]
-        res = self.helper.send_event(
+        res = await self.helper.send_event(
             room_id=self.room_id,
             type="m.reaction",
             content={
@@ -141,7 +141,7 @@ class PaginationTestCase(HomeserverTestCase):
         self.event_id_reference = res["event_id"]
 
         # An event with no relations.
-        res = self.helper.send_event(
+        res = await self.helper.send_event(
             room_id=self.room_id,
             type=EventTypes.Message,
             content={"msgtype": "m.text", "body": "No relations"},
@@ -149,10 +149,10 @@ class PaginationTestCase(HomeserverTestCase):
         )
         self.event_id_none = res["event_id"]
 
-    def _filter_messages(self, filter: JsonDict) -> list[str]:
+    async def _filter_messages(self, filter: JsonDict) -> list[str]:
         """Make a request to /messages with a filter, returns the chunk of events."""
 
-        events, next_key, _ = self.get_success(
+        events, next_key, _ = await self.get_success(
             self.hs.get_datastores().main.paginate_room_events_by_topological_ordering(
                 room_id=self.room_id,
                 from_key=self.from_token.room_key,
@@ -165,31 +165,31 @@ class PaginationTestCase(HomeserverTestCase):
 
         return [ev.event_id for ev in events]
 
-    def test_filter_relation_senders(self) -> None:
+    async def test_filter_relation_senders(self) -> None:
         # Messages which second user reacted to.
         filter = {"related_by_senders": [self.second_user_id]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_1])
 
         # Messages which third user reacted to.
         filter = {"related_by_senders": [self.third_user_id]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_2])
 
         # Messages which either user reacted to.
         filter = {"related_by_senders": [self.second_user_id, self.third_user_id]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertCountEqual(chunk, [self.event_id_1, self.event_id_2])
 
-    def test_filter_relation_type(self) -> None:
+    async def test_filter_relation_type(self) -> None:
         # Messages which have annotations.
         filter = {"related_by_rel_types": [RelationTypes.ANNOTATION]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_1])
 
         # Messages which have references.
         filter = {"related_by_rel_types": [RelationTypes.REFERENCE]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_2])
 
         # Messages which have either annotations or references.
@@ -199,21 +199,21 @@ class PaginationTestCase(HomeserverTestCase):
                 RelationTypes.REFERENCE,
             ]
         }
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertCountEqual(chunk, [self.event_id_1, self.event_id_2])
 
-    def test_filter_relation_senders_and_type(self) -> None:
+    async def test_filter_relation_senders_and_type(self) -> None:
         # Messages which second user reacted to.
         filter = {
             "related_by_senders": [self.second_user_id],
             "related_by_rel_types": [RelationTypes.ANNOTATION],
         }
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_1])
 
-    def test_duplicate_relation(self) -> None:
+    async def test_duplicate_relation(self) -> None:
         """An event should only be returned once if there are multiple relations to it."""
-        self.helper.send_event(
+        await self.helper.send_event(
             room_id=self.room_id,
             type="m.reaction",
             content={
@@ -227,18 +227,18 @@ class PaginationTestCase(HomeserverTestCase):
         )
 
         filter = {"related_by_senders": [self.second_user_id]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_1])
 
-    def test_filter_rel_types(self) -> None:
+    async def test_filter_rel_types(self) -> None:
         # Messages which are annotations.
         filter = {"org.matrix.msc3874.rel_types": [RelationTypes.ANNOTATION]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_annotation])
 
         # Messages which are references.
         filter = {"org.matrix.msc3874.rel_types": [RelationTypes.REFERENCE]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_reference])
 
         # Messages which are either annotations or references.
@@ -248,16 +248,16 @@ class PaginationTestCase(HomeserverTestCase):
                 RelationTypes.REFERENCE,
             ]
         }
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertCountEqual(
             chunk,
             [self.event_id_annotation, self.event_id_reference],
         )
 
-    def test_filter_not_rel_types(self) -> None:
+    async def test_filter_not_rel_types(self) -> None:
         # Messages which are not annotations.
         filter = {"org.matrix.msc3874.not_rel_types": [RelationTypes.ANNOTATION]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(
             chunk,
             [
@@ -270,7 +270,7 @@ class PaginationTestCase(HomeserverTestCase):
 
         # Messages which are not references.
         filter = {"org.matrix.msc3874.not_rel_types": [RelationTypes.REFERENCE]}
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(
             chunk,
             [
@@ -288,7 +288,7 @@ class PaginationTestCase(HomeserverTestCase):
                 RelationTypes.REFERENCE,
             ]
         }
-        chunk = self._filter_messages(filter)
+        chunk = await self._filter_messages(filter)
         self.assertEqual(chunk, [self.event_id_1, self.event_id_2, self.event_id_none])
 
 

tests/unittest.py

@@ -22,6 +22,7 @@
 import functools
 import gc
 import hashlib
+import inspect
 import hmac
 import json
 import logging
@@ -571,7 +572,9 @@ class HomeserverTestCase(TestCase):
             self.reactor.threadpool.start()
 
         if hasattr(self, "prepare"):
-            self.prepare(self.reactor, self.clock, self.hs)
+            res = self.prepare(self.reactor, self.clock, self.hs)
+            if inspect.isawaitable(res):
+                await res
 
     def tearDown(self) -> None:
         # Reset to not use frozen dicts.
@@ -1016,7 +1019,7 @@ class FederatingHomeserverTestCase(HomeserverTestCase):
     OTHER_SERVER_NAME = "other.example.com"
     OTHER_SERVER_SIGNATURE_KEY = signedjson.key.generate_signing_key("test")
 
-    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+    async def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         super().prepare(reactor, clock, hs)
 
         # poke the other server's signing key into the key store, so that we don't
@@ -1024,7 +1027,7 @@ class FederatingHomeserverTestCase(HomeserverTestCase):
         verify_key = signedjson.key.get_verify_key(self.OTHER_SERVER_SIGNATURE_KEY)
         verify_key_id = "%s:%s" % (verify_key.alg, verify_key.version)
 
-        self.get_success(
+        await self.get_success(
             hs.get_datastores().main.store_server_keys_response(
                 self.OTHER_SERVER_NAME,
                 from_server=self.OTHER_SERVER_NAME,