Add background job to clear unreferenced state groups

changelog.d/18154.feature

@@ -0,0 +1 @@
+Add background job to clear unreferenced state groups.

docs/development/database_schema.md

@@ -162,7 +162,7 @@ by a unique name, the current status (stored in JSON), and some dependency infor
 * Whether the update requires a previous update to be complete.
 * A rough ordering for which to complete updates.
 
-A new background updates needs to be added to the `background_updates` table:
+A new background update needs to be added to the `background_updates` table:
 
 ```sql
 INSERT INTO background_updates (ordering, update_name, depends_on, progress_json) VALUES

synapse/_scripts/synapse_port_db.py

@@ -191,6 +191,11 @@
 
 
 IGNORED_TABLES = {
+    # Porting the auto generated sequence in this table is non-trivial.
+    # And anything not ported, will get automatically added back by the
+    # `delete_unreferenced_state_groups_bg_update` background task.
+    # This makes it safe to ignore porting this table.
+    "state_groups_pending_deletion",
     # We don't port these tables, as they're a faff and we can regenerate
     # them anyway.
     "user_directory",

synapse/storage/controllers/purge_events.py

@@ -19,13 +19,13 @@
 #
 #
 
-import itertools
 import logging
-from typing import TYPE_CHECKING, Collection, Mapping, Set
+from typing import TYPE_CHECKING, Mapping
 
 from synapse.logging.context import nested_logging_context
 from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.storage.databases import Databases
+from synapse.storage.databases.state.bg_updates import find_unreferenced_groups
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -72,76 +72,19 @@ async def purge_history(
             )
 
             logger.info("[purge] finding state groups that can be deleted")
-            sg_to_delete = await self._find_unreferenced_groups(state_groups)
+            # sg_to_delete = await self._find_unreferenced_groups(state_groups)
+            sg_to_delete = await find_unreferenced_groups(
+                self.stores.main.db_pool,
+                self.stores.state.db_pool,
+                state_groups,
+            )
 
             # Mark these state groups as pending deletion, they will actually
             # get deleted automatically later.
             await self.stores.state_deletion.mark_state_groups_as_pending_deletion(
                 sg_to_delete
             )
 
-    async def _find_unreferenced_groups(
-        self, state_groups: Collection[int]
-    ) -> Set[int]:
-        """Used when purging history to figure out which state groups can be
-        deleted.
-
-        Args:
-            state_groups: Set of state groups referenced by events
-                that are going to be deleted.
-
-        Returns:
-            The set of state groups that can be deleted.
-        """
-        # Set of events that we have found to be referenced by events
-        referenced_groups = set()
-
-        # Set of state groups we've already seen
-        state_groups_seen = set(state_groups)
-
-        # Set of state groups to handle next.
-        next_to_search = set(state_groups)
-        while next_to_search:
-            # We bound size of groups we're looking up at once, to stop the
-            # SQL query getting too big
-            if len(next_to_search) < 100:
-                current_search = next_to_search
-                next_to_search = set()
-            else:
-                current_search = set(itertools.islice(next_to_search, 100))
-                next_to_search -= current_search
-
-            referenced = await self.stores.main.get_referenced_state_groups(
-                current_search
-            )
-            referenced_groups |= referenced
-
-            # We don't continue iterating up the state group graphs for state
-            # groups that are referenced.
-            current_search -= referenced
-
-            edges = await self.stores.state.get_previous_state_groups(current_search)
-
-            prevs = set(edges.values())
-            # We don't bother re-handling groups we've already seen
-            prevs -= state_groups_seen
-            next_to_search |= prevs
-            state_groups_seen |= prevs
-
-            # We also check to see if anything referencing the state groups are
-            # also unreferenced. This helps ensure that we delete unreferenced
-            # state groups, if we don't then we will de-delta them when we
-            # delete the other state groups leading to increased DB usage.
-            next_edges = await self.stores.state.get_next_state_groups(current_search)
-            nexts = set(next_edges.keys())
-            nexts -= state_groups_seen
-            next_to_search |= nexts
-            state_groups_seen |= nexts
-
-        to_delete = state_groups_seen - referenced_groups
-
-        return to_delete
-
     @wrap_as_background_process("_delete_state_groups_loop")
     async def _delete_state_groups_loop(self) -> None:
         """Background task that deletes any state groups that may be pending