Repository: matrix-org/synapse

Issue:
CI run against latest deps is failing
See https://github.com/matrix-org/synapse/actions/runs/2407332978


Discussion:

Patch:
diff --git a/changelog.d/12832.bugfix b/changelog.d/12832.bugfix
new file mode 100644
index 000000000..497d5184e
--- /dev/null
+++ b/changelog.d/12832.bugfix
@@ -0,0 +1,1 @@
+Fixed a bug which allowed multiple async operations to access database locks concurrently. Contributed by @sumnerevans @ Beeper.
diff --git a/changelog.d/12836.misc b/changelog.d/12836.misc
new file mode 100644
index 000000000..85909c6a2
--- /dev/null
+++ b/changelog.d/12836.misc
@@ -0,0 +1,1 @@
+Remove Mutual Rooms ([MSC2666](https://github.com/matrix-org/matrix-spec-proposals/pull/2666)) endpoint dependency on the User Directory.
\ No newline at end of file
diff --git a/changelog.d/12912.misc b/changelog.d/12912.misc
new file mode 100644
index 000000000..6396fd9d3
--- /dev/null
+++ b/changelog.d/12912.misc
@@ -0,0 +1,1 @@
+Bump types-jsonschema from 4.4.1 to 4.4.6.
diff --git a/poetry.lock b/poetry.lock
index f64d70941..6b4686545 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -1355,7 +1355,7 @@ python-versions = "*"
 
 [[package]]
 name = "types-jsonschema"
-version = "4.4.1"
+version = "4.4.6"
 description = "Typing stubs for jsonschema"
 category = "dev"
 optional = false
@@ -2618,8 +2618,8 @@ types-ipaddress = [
     {file = "types_ipaddress-1.0.8-py3-none-any.whl", hash = "sha256:4933b74da157ba877b1a705d64f6fa7742745e9ffd65e51011f370c11ebedb55"},
 ]
 types-jsonschema = [
-    {file = "types-jsonschema-4.4.1.tar.gz", hash = "sha256:bd68b75217ebbb33b0242db10047581dad3b061a963a46ee80d4a9044080663e"},
-    {file = "types_jsonschema-4.4.1-py3-none-any.whl", hash = "sha256:ab3ecfdc912d6091cc82f4b7556cfbf1a7cbabc26da0ceaa1cbbc232d1d09971"},
+    {file = "types-jsonschema-4.4.6.tar.gz", hash = "sha256:7f2a804618756768c7c0616f8c794b61fcfe3077c7ee1ad47dcf01c5e5f692bb"},
+    {file = "types_jsonschema-4.4.6-py3-none-any.whl", hash = "sha256:1db9031ca49a8444d01bd2ce8cf2f89318382b04610953b108321e6f8fb03390"},
 ]
 types-opentracing = [
     {file = "types-opentracing-2.4.7.tar.gz", hash = "sha256:be60e9618355aa892571ace002e6b353702538b1c0dc4fbc1c921219d6658830"},
diff --git a/synapse/events/validator.py b/synapse/events/validator.py
index 360d24274..29fa9b388 100644
--- a/synapse/events/validator.py
+++ b/synapse/events/validator.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import collections.abc
-from typing import Iterable, Type, Union
+from typing import Iterable, Type, Union, cast
 
 import jsonschema
 
@@ -103,7 +103,12 @@ class EventValidator:
             except jsonschema.ValidationError as e:
                 if e.path:
                     # example: "users_default": '0' is not of type 'integer'
-                    message = '"' + e.path[-1] + '": ' + e.message  # noqa: B306
+                    # cast safety: path entries can be integers, if we fail to validate
+                    # items in an array. However the POWER_LEVELS_SCHEMA doesn't expect
+                    # to see any arrays.
+                    message = (
+                        '"' + cast(str, e.path[-1]) + '": ' + e.message  # noqa: B306
+                    )
                     # jsonschema.ValidationError.message is a valid attribute
                 else:
                     # example: '0' is not of type 'integer'
diff --git a/synapse/rest/client/mutual_rooms.py b/synapse/rest/client/mutual_rooms.py
index 27bfaf0b2..38ef4e459 100644
--- a/synapse/rest/client/mutual_rooms.py
+++ b/synapse/rest/client/mutual_rooms.py
@@ -42,21 +42,10 @@ class UserMutualRoomsServlet(RestServlet):
         super().__init__()
         self.auth = hs.get_auth()
         self.store = hs.get_datastores().main
-        self.user_directory_search_enabled = (
-            hs.config.userdirectory.user_directory_search_enabled
-        )
 
     async def on_GET(
         self, request: SynapseRequest, user_id: str
     ) -> Tuple[int, JsonDict]:
-
-        if not self.user_directory_search_enabled:
-            raise SynapseError(
-                code=400,
-                msg="User directory searching is disabled. Cannot determine shared rooms.",
-                errcode=Codes.UNKNOWN,
-            )
-
         UserID.from_string(user_id)
 
         requester = await self.auth.get_user_by_req(request)
@@ -67,8 +56,8 @@ class UserMutualRoomsServlet(RestServlet):
                 errcode=Codes.FORBIDDEN,
             )
 
-        rooms = await self.store.get_mutual_rooms_for_users(
-            requester.user.to_string(), user_id
+        rooms = await self.store.get_mutual_rooms_between_users(
+            frozenset((requester.user.to_string(), user_id))
         )
 
         return 200, {"joined": list(rooms)}
diff --git a/synapse/storage/databases/main/lock.py b/synapse/storage/databases/main/lock.py
index bedacaf0d..2d7633fbd 100644
--- a/synapse/storage/databases/main/lock.py
+++ b/synapse/storage/databases/main/lock.py
@@ -13,7 +13,7 @@
 # limitations under the License.
 import logging
 from types import TracebackType
-from typing import TYPE_CHECKING, Optional, Tuple, Type
+from typing import TYPE_CHECKING, Optional, Set, Tuple, Type
 from weakref import WeakValueDictionary
 
 from twisted.internet.interfaces import IReactorCore
@@ -84,6 +84,8 @@ class LockStore(SQLBaseStore):
             self._on_shutdown,
         )
 
+        self._acquiring_locks: Set[Tuple[str, str]] = set()
+
     @wrap_as_background_process("LockStore._on_shutdown")
     async def _on_shutdown(self) -> None:
         """Called when the server is shutting down"""
@@ -103,6 +105,21 @@ class LockStore(SQLBaseStore):
         context manager if the lock is successfully acquired, which *must* be
         used (otherwise the lock will leak).
         """
+        if (lock_name, lock_key) in self._acquiring_locks:
+            return None
+        try:
+            self._acquiring_locks.add((lock_name, lock_key))
+            return await self._try_acquire_lock(lock_name, lock_key)
+        finally:
+            self._acquiring_locks.discard((lock_name, lock_key))
+
+    async def _try_acquire_lock(
+        self, lock_name: str, lock_key: str
+    ) -> Optional["Lock"]:
+        """Try to acquire a lock for the given name/key. Will return an async
+        context manager if the lock is successfully acquired, which *must* be
+        used (otherwise the lock will leak).
+        """
 
         # Check if this process has taken out a lock and if it's still valid.
         lock = self._live_tokens.get((lock_name, lock_key))
diff --git a/synapse/storage/databases/main/roommember.py b/synapse/storage/databases/main/roommember.py
index cc528fcf2..e222b7bd1 100644
--- a/synapse/storage/databases/main/roommember.py
+++ b/synapse/storage/databases/main/roommember.py
@@ -670,6 +670,30 @@ class RoomMemberWorkerStore(EventsWorkerStore):
 
         return user_who_share_room
 
+    @cached(cache_context=True, iterable=True)
+    async def get_mutual_rooms_between_users(
+        self, user_ids: FrozenSet[str], cache_context: _CacheContext
+    ) -> FrozenSet[str]:
+        """
+        Returns the set of rooms that all users in `user_ids` share.
+
+        Args:
+            user_ids: A frozen set of all users to investigate and return
+              overlapping joined rooms for.
+            cache_context
+        """
+        shared_room_ids: Optional[FrozenSet[str]] = None
+        for user_id in user_ids:
+            room_ids = await self.get_rooms_for_user(
+                user_id, on_invalidate=cache_context.invalidate
+            )
+            if shared_room_ids is not None:
+                shared_room_ids &= room_ids
+            else:
+                shared_room_ids = room_ids
+
+        return shared_room_ids or frozenset()
+
     async def get_joined_users_from_context(
         self, event: EventBase, context: EventContext
     ) -> Dict[str, ProfileInfo]:
diff --git a/synapse/storage/databases/main/user_directory.py b/synapse/storage/databases/main/user_directory.py
index 028db69af..2282242e9 100644
--- a/synapse/storage/databases/main/user_directory.py
+++ b/synapse/storage/databases/main/user_directory.py
@@ -729,49 +729,6 @@ class UserDirectoryStore(UserDirectoryBackgroundUpdateStore):
         users.update(rows)
         return list(users)
 
-    async def get_mutual_rooms_for_users(
-        self, user_id: str, other_user_id: str
-    ) -> Set[str]:
-        """
-        Returns the rooms that a local user shares with another local or remote user.
-
-        Args:
-            user_id: The MXID of a local user
-            other_user_id: The MXID of the other user
-
-        Returns:
-            A set of room ID's that the users share.
-        """
-
-        def _get_mutual_rooms_for_users_txn(
-            txn: LoggingTransaction,
-        ) -> List[Dict[str, str]]:
-            txn.execute(
-                """
-                SELECT p1.room_id
-                FROM users_in_public_rooms as p1
-                INNER JOIN users_in_public_rooms as p2
-                    ON p1.room_id = p2.room_id
-                    AND p1.user_id = ?
-                    AND p2.user_id = ?
-                UNION
-                SELECT room_id
-                FROM users_who_share_private_rooms
-                WHERE
-                    user_id = ?
-                    AND other_user_id = ?
-                """,
-                (user_id, other_user_id, user_id, other_user_id),
-            )
-            rows = self.db_pool.cursor_to_dict(txn)
-            return rows
-
-        rows = await self.db_pool.runInteraction(
-            "get_mutual_rooms_for_users", _get_mutual_rooms_for_users_txn
-        )
-
-        return {row["room_id"] for row in rows}
-
     async def get_user_directory_stream_pos(self) -> Optional[int]:
         """
         Get the stream ID of the user directory stream.


Test patch:
diff --git a/tests/rest/client/test_mutual_rooms.py b/tests/rest/client/test_mutual_rooms.py
index 7b7d283bb..a4327f7ac 100644
--- a/tests/rest/client/test_mutual_rooms.py
+++ b/tests/rest/client/test_mutual_rooms.py
@@ -36,12 +36,10 @@ class UserMutualRoomsTest(unittest.HomeserverTestCase):
 
     def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
         config = self.default_config()
-        config["update_user_directory"] = True
         return self.setup_test_homeserver(config=config)
 
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.store = hs.get_datastores().main
-        self.handler = hs.get_user_directory_handler()
 
     def _get_mutual_rooms(self, token: str, other_user: str) -> FakeChannel:
         return self.make_request(
diff --git a/tests/storage/databases/main/test_lock.py b/tests/storage/databases/main/test_lock.py
index 74c6224eb..3cc2a58d8 100644
--- a/tests/storage/databases/main/test_lock.py
+++ b/tests/storage/databases/main/test_lock.py
@@ -12,6 +12,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from twisted.internet import defer, reactor
+from twisted.internet.base import ReactorBase
+from twisted.internet.defer import Deferred
+
 from synapse.server import HomeServer
 from synapse.storage.databases.main.lock import _LOCK_TIMEOUT_MS
 
@@ -22,6 +26,56 @@ class LockTestCase(unittest.HomeserverTestCase):
     def prepare(self, reactor, clock, hs: HomeServer):
         self.store = hs.get_datastores().main
 
+    def test_acquire_contention(self):
+        # Track the number of tasks holding the lock.
+        # Should be at most 1.
+        in_lock = 0
+        max_in_lock = 0
+
+        release_lock: "Deferred[None]" = Deferred()
+
+        async def task():
+            nonlocal in_lock
+            nonlocal max_in_lock
+
+            lock = await self.store.try_acquire_lock("name", "key")
+            if not lock:
+                return
+
+            async with lock:
+                in_lock += 1
+                max_in_lock = max(max_in_lock, in_lock)
+
+                # Block to allow other tasks to attempt to take the lock.
+                await release_lock
+
+                in_lock -= 1
+
+        # Start 3 tasks.
+        task1 = defer.ensureDeferred(task())
+        task2 = defer.ensureDeferred(task())
+        task3 = defer.ensureDeferred(task())
+
+        # Give the reactor a kick so that the database transaction returns.
+        self.pump()
+
+        release_lock.callback(None)
+
+        # Run the tasks to completion.
+        # To work around `Linearizer`s using a different reactor to sleep when
+        # contended (#12841), we call `runUntilCurrent` on
+        # `twisted.internet.reactor`, which is a different reactor to that used
+        # by the homeserver.
+        assert isinstance(reactor, ReactorBase)
+        self.get_success(task1)
+        reactor.runUntilCurrent()
+        self.get_success(task2)
+        reactor.runUntilCurrent()
+        self.get_success(task3)
+
+        # At most one task should have held the lock at a time.
+        self.assertEqual(max_in_lock, 1)
+
     def test_simple_lock(self):
         """Test that we can take out a lock and that while we hold it nobody
         else can take it out.