diff --git a/synapse/storage/databases/main/events_bg_updates.py b/synapse/storage/databases/main/events_bg_updates.py index 9a11f9b9bb..f2b6b466c8 100644 --- a/synapse/storage/databases/main/events_bg_updates.py +++ b/synapse/storage/databases/main/events_bg_updates.py @@ -280,11 +280,6 @@ class EventsBackgroundUpdatesStore( ################################################################################ - self.db_pool.updates.register_background_update_handler( - _BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS, - self._background_drop_invalid_event_edges_rows, - ) - self.db_pool.updates.register_background_index_update( _BackgroundUpdates.EVENT_EDGES_REPLACE_INDEX, index_name="event_edges_event_id_prev_event_id_idx", @@ -1497,102 +1492,6 @@ class EventsBackgroundUpdatesStore( return 0 - async def _background_drop_invalid_event_edges_rows( - self, progress: JsonDict, batch_size: int - ) -> int: - """Drop invalid rows from event_edges - - This only runs for postgres. For SQLite, it all happens synchronously. - - Firstly, drop any rows with is_state=True. These may have been added a long time - ago, but they are no longer used. - - We also drop rows that do not correspond to entries in `events`, and add a - foreign key. - """ - - last_event_id = progress.get("last_event_id", "") - - def drop_invalid_event_edges_txn(txn: LoggingTransaction) -> bool: - """Returns True if we're done.""" - - # first we need to find an endpoint. - txn.execute( - """ - SELECT event_id FROM event_edges - WHERE event_id > ? - ORDER BY event_id - LIMIT 1 OFFSET ? - """, - (last_event_id, batch_size), - ) - - endpoint = None - row = txn.fetchone() - - if row: - endpoint = row[0] - - where_clause = "ee.event_id > ?" - args = [last_event_id] - if endpoint: - where_clause += " AND ee.event_id <= ?" - args.append(endpoint) - - # now delete any that: - # - have is_state=TRUE, or - # - do not correspond to a row in `events` - txn.execute( - f""" - DELETE FROM event_edges - WHERE event_id IN ( - SELECT ee.event_id - FROM event_edges ee - LEFT JOIN events ev USING (event_id) - WHERE ({where_clause}) AND - (is_state OR ev.event_id IS NULL) - )""", - args, - ) - - logger.info( - "cleaned up event_edges up to %s: removed %i/%i rows", - endpoint, - txn.rowcount, - batch_size, - ) - - if endpoint is not None: - self.db_pool.updates._background_update_progress_txn( - txn, - _BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS, - {"last_event_id": endpoint}, - ) - return False - - # if that was the final batch, we validate the foreign key. - # - # The constraint should have been in place and enforced for new rows since - # before we started deleting invalid rows, so there's no chance for any - # invalid rows to have snuck in the meantime. In other words, this really - # ought to succeed. - logger.info("cleaned up event_edges; enabling foreign key") - txn.execute( - "ALTER TABLE event_edges VALIDATE CONSTRAINT event_edges_event_id_fkey" - ) - return True - - done = await self.db_pool.runInteraction( - desc="drop_invalid_event_edges", func=drop_invalid_event_edges_txn - ) - - if done: - await self.db_pool.updates._end_background_update( - _BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS - ) - - return batch_size - async def _background_events_populate_state_key_rejections( self, progress: JsonDict, batch_size: int ) -> int: diff --git a/synapse/storage/schema/main/delta/95/01_foreground_event_edges_state_cleanup.py b/synapse/storage/schema/main/delta/95/01_foreground_event_edges_state_cleanup.py new file mode 100644 index 0000000000..a914b20f07 --- /dev/null +++ b/synapse/storage/schema/main/delta/95/01_foreground_event_edges_state_cleanup.py @@ -0,0 +1,74 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# + +import logging + +from synapse.storage.database import LoggingTransaction +from synapse.storage.engines import BaseDatabaseEngine, Sqlite3Engine + +logger = logging.getLogger(__name__) + + +def run_create(txn: LoggingTransaction, database_engine: BaseDatabaseEngine) -> None: + """ + This migration forces the `event_edges_drop_invalid_rows` background update + scheduled by `71/01rebuild_event_edges.sql.postgres` to be completed + in the foreground if it is still outstanding. + """ + + if isinstance(database_engine, Sqlite3Engine): + # SQLite already did everything synchronously in + # `71/01rebuild_event_edges.sql.sqlite` + # So there's nothing to do + return + + # Clear the background update whilst also checking it + txn.execute( + """ + DELETE FROM background_updates + WHERE update_name = 'event_edges_drop_invalid_rows' + RETURNING 1 + """ + ) + if txn.fetchone() is None: + # The background update has completed, so there is nothing to do. + return + + logger.warning( + "`event_edges_drop_invalid_rows` has not completed in background; running in foreground!" + ) + + # now delete any that: + # - have is_state=TRUE, or + # - do not correspond to a row in `events` + txn.execute( + """ + DELETE FROM event_edges + WHERE event_id IN ( + SELECT ee.event_id + FROM event_edges ee + LEFT JOIN events ev USING (event_id) + WHERE (is_state OR ev.event_id IS NULL) + ) + """, + ) + logger.info("Deleted %i legacy state edges from `event_edges` table", txn.rowcount) + + logger.info("Enabling foreign key") + txn.execute( + """ + ALTER TABLE event_edges + VALIDATE CONSTRAINT event_edges_event_id_fkey + """ + ) diff --git a/synapse/types/storage/__init__.py b/synapse/types/storage/__init__.py index 6b857aeb0e..2d2b31cccc 100644 --- a/synapse/types/storage/__init__.py +++ b/synapse/types/storage/__init__.py @@ -31,7 +31,6 @@ class _BackgroundUpdates: INDEX_STREAM_ORDERING2_TS = "index_stream_ordering2_ts" REPLACE_STREAM_ORDERING_COLUMN = "replace_stream_ordering_column" - EVENT_EDGES_DROP_INVALID_ROWS = "event_edges_drop_invalid_rows" EVENT_EDGES_REPLACE_INDEX = "event_edges_replace_index" EVENTS_POPULATE_STATE_KEY_REJECTIONS = "events_populate_state_key_rejections"