Ensure local invited & knocking users leave before purge. (#16559)

This is mostly useful for federated rooms where some users
would get stuck in the invite or knock state when the room
was purged from their homeserver.

changelog.d/16559.bugfix

@@ -0,0 +1 @@
+Fix a long-standing bug where invited/knocking users would not leave during a room purge.

synapse/handlers/room.py

@@ -1939,9 +1939,10 @@ class RoomShutdownHandler:
         else:
             logger.info("Shutting down room %r", room_id)
 
-        users = await self.store.get_users_in_room(room_id)
-        for user_id in users:
-            if not self.hs.is_mine_id(user_id):
+        users = await self.store.get_local_users_related_to_room(room_id)
+        for user_id, membership in users:
+            # If the user is not in the room (or is banned), nothing to do.
+            if membership not in (Membership.JOIN, Membership.INVITE, Membership.KNOCK):
                 continue
 
             logger.info("Kicking %r from %r...", user_id, room_id)

synapse/storage/databases/main/roommember.py

@@ -482,6 +482,22 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
             desc="get_local_users_in_room",
         )
 
+    async def get_local_users_related_to_room(
+        self, room_id: str
+    ) -> List[Tuple[str, str]]:
+        """
+        Retrieves a list of the current roommembers who are local to the server and their membership status.
+        """
+        return cast(
+            List[Tuple[str, str]],
+            await self.db_pool.simple_select_list(
+                table="local_current_membership",
+                keyvalues={"room_id": room_id},
+                retcols=("user_id", "membership"),
+                desc="get_local_users_in_room",
+            ),
+        )
+
     async def check_local_user_in_room(self, user_id: str, room_id: str) -> bool:
         """
         Check whether a given local user is currently joined to the given room.

tests/rest/admin/test_room.py

@@ -29,7 +29,7 @@ from synapse.handlers.pagination import (
     PURGE_ROOM_ACTION_NAME,
     SHUTDOWN_AND_PURGE_ROOM_ACTION_NAME,
 )
-from synapse.rest.client import directory, events, login, room
+from synapse.rest.client import directory, events, knock, login, room, sync
 from synapse.server import HomeServer
 from synapse.types import UserID
 from synapse.util import Clock
@@ -49,6 +49,8 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         login.register_servlets,
         events.register_servlets,
         room.register_servlets,
+        knock.register_servlets,
+        sync.register_servlets,
         room.register_deprecated_servlets,
     ]
 
@@ -254,6 +256,55 @@ class DeleteRoomTestCase(unittest.HomeserverTestCase):
         self._is_blocked(self.room_id, expect=False)
         self._has_no_members(self.room_id)
 
+    def test_purge_room_unjoined(self) -> None:
+        """Test to purge a room when there are invited or knocked users."""
+        # Test that room is not purged
+        with self.assertRaises(AssertionError):
+            self._is_purged(self.room_id)
+
+        # Test that room is not blocked
+        self._is_blocked(self.room_id, expect=False)
+
+        # Assert one user in room
+        self._is_member(room_id=self.room_id, user_id=self.other_user)
+        self.helper.send_state(
+            self.room_id,
+            EventTypes.JoinRules,
+            {"join_rule": "knock"},
+            tok=self.other_user_tok,
+        )
+
+        # Invite a user.
+        invited_user = self.register_user("invited", "pass")
+        self.helper.invite(
+            self.room_id, self.other_user, invited_user, tok=self.other_user_tok
+        )
+
+        # Have a user knock.
+        knocked_user = self.register_user("knocked", "pass")
+        knocked_user_tok = self.login("knocked", "pass")
+        self.helper.knock(self.room_id, knocked_user, tok=knocked_user_tok)
+
+        channel = self.make_request(
+            "DELETE",
+            self.url.encode("ascii"),
+            content={"block": False, "purge": True},
+            access_token=self.admin_user_tok,
+        )
+
+        self.assertEqual(200, channel.code, msg=channel.json_body)
+        self.assertEqual(None, channel.json_body["new_room_id"])
+        self.assertCountEqual(
+            [self.other_user, invited_user, knocked_user],
+            channel.json_body["kicked_users"],
+        )
+        self.assertIn("failed_to_kick_users", channel.json_body)
+        self.assertIn("local_aliases", channel.json_body)
+
+        self._is_purged(self.room_id)
+        self._is_blocked(self.room_id, expect=False)
+        self._has_no_members(self.room_id)
+
     def test_block_room_and_not_purge(self) -> None:
         """Test to block a room without purging it.
         Members will not be moved to a new room and will not receive a message.