diff --git a/synapse/storage/databases/main/events_bg_updates.py b/synapse/storage/databases/main/events_bg_updates.py index 46966f5e99..33294ed0da 100644 --- a/synapse/storage/databases/main/events_bg_updates.py +++ b/synapse/storage/databases/main/events_bg_updates.py @@ -290,11 +290,6 @@ class EventsBackgroundUpdatesStore( replaces_index="ev_edges_id", ) - self.db_pool.updates.register_background_update_handler( - _BackgroundUpdates.EVENTS_POPULATE_STATE_KEY_REJECTIONS, - self._background_events_populate_state_key_rejections, - ) - # Add an index that would be useful for jumping to date using # get_event_id_for_timestamp. self.db_pool.updates.register_background_index_update( @@ -1471,86 +1466,6 @@ class EventsBackgroundUpdatesStore( return batch_size - async def _background_events_populate_state_key_rejections( - self, progress: JsonDict, batch_size: int - ) -> int: - """Back-populate `events.state_key` and `events.rejection_reason""" - - min_stream_ordering_exclusive = progress["min_stream_ordering_exclusive"] - max_stream_ordering_inclusive = progress["max_stream_ordering_inclusive"] - - def _populate_txn(txn: LoggingTransaction) -> bool: - """Returns True if we're done.""" - - # first we need to find an endpoint. - # we need to find the final row in the batch of batch_size, which means - # we need to skip over (batch_size-1) rows and get the next row. - txn.execute( - """ - SELECT stream_ordering FROM events - WHERE stream_ordering > ? AND stream_ordering <= ? - ORDER BY stream_ordering - LIMIT 1 OFFSET ? - """, - ( - min_stream_ordering_exclusive, - max_stream_ordering_inclusive, - batch_size - 1, - ), - ) - - row = txn.fetchone() - if row: - endpoint = row[0] - else: - # if the query didn't return a row, we must be almost done. We just - # need to go up to the recorded max_stream_ordering. - endpoint = max_stream_ordering_inclusive - - where_clause = "stream_ordering > ? AND stream_ordering <= ?" - args = [min_stream_ordering_exclusive, endpoint] - - # now do the updates. - txn.execute( - f""" - UPDATE events - SET state_key = (SELECT state_key FROM state_events se WHERE se.event_id = events.event_id), - rejection_reason = (SELECT reason FROM rejections rej WHERE rej.event_id = events.event_id) - WHERE ({where_clause}) - """, - args, - ) - - logger.info( - "populated new `events` columns up to %i/%i: updated %i rows", - endpoint, - max_stream_ordering_inclusive, - txn.rowcount, - ) - - if endpoint >= max_stream_ordering_inclusive: - # we're done - return True - - progress["min_stream_ordering_exclusive"] = endpoint - self.db_pool.updates._background_update_progress_txn( - txn, - _BackgroundUpdates.EVENTS_POPULATE_STATE_KEY_REJECTIONS, - progress, - ) - return False - - done = await self.db_pool.runInteraction( - desc="events_populate_state_key_rejections", func=_populate_txn - ) - - if done: - await self.db_pool.updates._end_background_update( - _BackgroundUpdates.EVENTS_POPULATE_STATE_KEY_REJECTIONS - ) - - return batch_size - async def _sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update( self, progress: JsonDict, _batch_size: int ) -> int: diff --git a/synapse/storage/schema/main/delta/95/03_foreground_events_populate_state_key_rejections.py b/synapse/storage/schema/main/delta/95/03_foreground_events_populate_state_key_rejections.py new file mode 100644 index 0000000000..6e71db0d01 --- /dev/null +++ b/synapse/storage/schema/main/delta/95/03_foreground_events_populate_state_key_rejections.py @@ -0,0 +1,64 @@ +# +# 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 json +import logging + +from synapse.storage.database import LoggingTransaction +from synapse.storage.engines import BaseDatabaseEngine + +logger = logging.getLogger(__name__) + + +def run_create(txn: LoggingTransaction, database_engine: BaseDatabaseEngine) -> None: + """ + This migration forces the `events_populate_state_key_rejections` background + update scheduled by `72/03bg_populate_events_columns.py` to be completed + in the foreground if it is still outstanding. + """ + + # Clear the background update whilst also checking it + txn.execute( + """ + DELETE FROM background_updates + WHERE update_name = 'events_populate_state_key_rejections' + RETURNING progress_json + """ + ) + row = txn.fetchone() + if row is None: + # The background update has completed, so there is nothing to do. + return + + progress = json.loads(row[0]) + min_stream_ordering_exclusive = progress["min_stream_ordering_exclusive"] + max_stream_ordering_inclusive = progress["max_stream_ordering_inclusive"] + + logger.warning( + "`events_populate_state_key_rejections` has not completed in background; running in foreground!" + ) + + # Backpopulate the `state_key` and `rejection_reason` columns according to the unpopulated range + # specified in the background update progress dict. + # For simplicity, do this in one query. + txn.execute( + """ + UPDATE events + SET state_key = (SELECT state_key FROM state_events se WHERE se.event_id = events.event_id), + rejection_reason = (SELECT reason FROM rejections rej WHERE rej.event_id = events.event_id) + WHERE ? < stream_ordering AND stream_ordering <= ? + """, + (min_stream_ordering_exclusive, max_stream_ordering_inclusive), + ) + logger.info("Populated new `events` columns for %i rows", txn.rowcount) diff --git a/synapse/types/storage/__init__.py b/synapse/types/storage/__init__.py index 6b857aeb0e..34ec1eed9c 100644 --- a/synapse/types/storage/__init__.py +++ b/synapse/types/storage/__init__.py @@ -34,8 +34,6 @@ class _BackgroundUpdates: 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" - EVENTS_JUMP_TO_DATE_INDEX = "events_jump_to_date_index" CURRENT_STATE_EVENTS_STREAM_ORDERING_INDEX_UPDATE_NAME = (