Properly purge state groups tables when purging a room

Author: MatMaul

changelog.d/18024.bug

@@ -0,0 +1 @@
+Properly purge state groups tables when purging a room with the admin API.

synapse/storage/controllers/purge_events.py

@@ -42,8 +42,7 @@ async def purge_room(self, room_id: str) -> None:
         """Deletes all record of a room"""
 
         with nested_logging_context(room_id):
-            state_groups_to_delete = await self.stores.main.purge_room(room_id)
-            await self.stores.state.purge_room_state(room_id, state_groups_to_delete)
+            await self.stores.main.purge_room(room_id)
 
     async def purge_history(
         self, room_id: str, token: str, delete_local_events: bool

synapse/storage/databases/main/purge_events.py

@@ -20,7 +20,7 @@
 #
 
 import logging
-from typing import Any, List, Set, Tuple, cast
+from typing import Any, Set, Tuple, cast
 
 from synapse.api.errors import SynapseError
 from synapse.storage.database import LoggingTransaction
@@ -332,7 +332,7 @@ def _purge_history_txn(
 
         return referenced_state_groups
 
-    async def purge_room(self, room_id: str) -> List[int]:
+    async def purge_room(self, room_id: str) -> None:
         """Deletes all record of a room
 
         Args:
@@ -348,26 +348,23 @@ async def purge_room(self, room_id: str) -> List[int]:
         # purge any of those rows which were added during the first.
 
         logger.info("[purge] Starting initial main purge of [1/2]")
-        state_groups_to_delete = await self.db_pool.runInteraction(
+        await self.db_pool.runInteraction(
             "purge_room",
             self._purge_room_txn,
             room_id=room_id,
             isolation_level=IsolationLevel.READ_COMMITTED,
         )
 
         logger.info("[purge] Starting secondary main purge of [2/2]")
-        state_groups_to_delete.extend(
-            await self.db_pool.runInteraction(
-                "purge_room",
-                self._purge_room_txn,
-                room_id=room_id,
-            ),
+        await self.db_pool.runInteraction(
+            "purge_room",
+            self._purge_room_txn,
+            room_id=room_id,
         )
-        logger.info("[purge] Done with main purge")
 
-        return state_groups_to_delete
+        logger.info("[purge] Done with main purge")
 
-    def _purge_room_txn(self, txn: LoggingTransaction, room_id: str) -> List[int]:
+    def _purge_room_txn(self, txn: LoggingTransaction, room_id: str) -> None:
         # This collides with event persistence so we cannot write new events and metadata into
         # a room while deleting it or this transaction will fail.
         if isinstance(self.database_engine, PostgresEngine):
@@ -376,19 +373,6 @@ def _purge_room_txn(self, txn: LoggingTransaction, room_id: str) -> List[int]:
                 (room_id,),
             )
 
-        # First, fetch all the state groups that should be deleted, before
-        # we delete that information.
-        txn.execute(
-            """
-                SELECT DISTINCT state_group FROM events
-                INNER JOIN event_to_state_groups USING(event_id)
-                WHERE events.room_id = ?
-            """,
-            (room_id,),
-        )
-
-        state_groups = [row[0] for row in txn]
-
         # Get all the auth chains that are referenced by events that are to be
         # deleted.
         txn.execute(
@@ -484,6 +468,8 @@ def _purge_room_txn(self, txn: LoggingTransaction, room_id: str) -> List[int]:
             logger.info("[purge] removing from %s", table)
             txn.execute("DELETE FROM %s WHERE room_id=?" % (table,), (room_id,))
 
+        self._purge_room_state_txn(txn, room_id)
+
         # Other tables we do NOT need to clear out:
         #
         #  - blocked_rooms
@@ -509,4 +495,37 @@ def _purge_room_txn(self, txn: LoggingTransaction, room_id: str) -> List[int]:
 
         self._invalidate_caches_for_room_and_stream(txn, room_id)
 
-        return state_groups
+    def _purge_room_state_txn(
+        self,
+        txn: LoggingTransaction,
+        room_id: str,
+    ) -> None:
+        # Delete all edges that reference a state group linked to room_id
+        logger.info("[purge] removing %s from state_group_edges", room_id)
+        txn.execute(
+            """
+            DELETE FROM state_group_edges AS sge WHERE sge.state_group IN (
+                SELECT id FROM state_groups AS sg WHERE sg.room_id = ?
+            )""",
+            (room_id,),
+        )
+
+        # state_groups_state table has a room_id column but no index on it, unlike state_groups,
+        # so we delete them by matching the room_id through the state_groups table.
+        logger.info("[purge] removing %s from state_groups_state", room_id)
+        txn.execute(
+            """
+            DELETE FROM state_groups_state AS sgs WHERE sgs.state_group IN (
+                SELECT id FROM state_groups AS sg WHERE sg.room_id = ?
+            )""",
+            (room_id,),
+        )
+
+        logger.info("[purge] removing %s from state_groups", room_id)
+        self.db_pool.simple_delete_many_txn(
+            txn,
+            table="state_groups",
+            column="room_id",
+            values=[room_id],
+            keyvalues={},
+        )

synapse/storage/databases/state/store.py

@@ -839,61 +839,3 @@ async def get_previous_state_groups(
         )
 
         return dict(rows)
-
-    async def purge_room_state(
-        self, room_id: str, state_groups_to_delete: Collection[int]
-    ) -> None:
-        """Deletes all record of a room from state tables
-
-        Args:
-            room_id:
-            state_groups_to_delete: State groups to delete
-        """
-
-        logger.info("[purge] Starting state purge")
-        await self.db_pool.runInteraction(
-            "purge_room_state",
-            self._purge_room_state_txn,
-            room_id,
-            state_groups_to_delete,
-        )
-        logger.info("[purge] Done with state purge")
-
-    def _purge_room_state_txn(
-        self,
-        txn: LoggingTransaction,
-        room_id: str,
-        state_groups_to_delete: Collection[int],
-    ) -> None:
-        # first we have to delete the state groups states
-        logger.info("[purge] removing %s from state_groups_state", room_id)
-
-        self.db_pool.simple_delete_many_txn(
-            txn,
-            table="state_groups_state",
-            column="state_group",
-            values=state_groups_to_delete,
-            keyvalues={},
-        )
-
-        # ... and the state group edges
-        logger.info("[purge] removing %s from state_group_edges", room_id)
-
-        self.db_pool.simple_delete_many_txn(
-            txn,
-            table="state_group_edges",
-            column="state_group",
-            values=state_groups_to_delete,
-            keyvalues={},
-        )
-
-        # ... and the state groups
-        logger.info("[purge] removing %s from state_groups", room_id)
-
-        self.db_pool.simple_delete_many_txn(
-            txn,
-            table="state_groups",
-            column="id",
-            values=state_groups_to_delete,
-            keyvalues={},
-        )

tests/rest/admin/test_room.py

@@ -3050,7 +3050,7 @@ def _block_room(self, room_id: str) -> None:
     "pusher_throttle",
     "room_account_data",
     "room_tags",
-    # "state_groups",  # Current impl leaves orphaned state groups around.
+    "state_groups",
     "state_groups_state",
     "federation_inbound_events_staging",
 ]