From eb3c84cf45e67c25116de55deeb9a78275dd8e6c Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Mon, 26 Aug 2024 17:31:26 -0500 Subject: [PATCH] Kick-off background update for out-of-date snapshots --- synapse/storage/database.py | 4 +- .../databases/main/events_bg_updates.py | 5 ++ synapse/storage/prepare_database.py | 47 ++++++++++++++----- 3 files changed, 42 insertions(+), 14 deletions(-) diff --git a/synapse/storage/database.py b/synapse/storage/database.py index ba2616b479..da50fd7f83 100644 --- a/synapse/storage/database.py +++ b/synapse/storage/database.py @@ -1323,7 +1323,7 @@ class DatabasePool: if lock: # We need to lock the table :( - self.engine.lock_table(txn, table) + txn.database_engine.lock_table(txn, table) def _getwhere(key: str) -> str: # If the value we're passing in is None (aka NULL), we need to use @@ -1377,8 +1377,8 @@ class DatabasePool: # successfully inserted return True + @staticmethod def simple_upsert_txn_native_upsert( - self, txn: LoggingTransaction, table: str, keyvalues: Mapping[str, Any], diff --git a/synapse/storage/databases/main/events_bg_updates.py b/synapse/storage/databases/main/events_bg_updates.py index af0a1ba880..1287b995fb 100644 --- a/synapse/storage/databases/main/events_bg_updates.py +++ b/synapse/storage/databases/main/events_bg_updates.py @@ -1552,6 +1552,11 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS def _get_rooms_to_update_txn(txn: LoggingTransaction) -> List[str]: # Fetch the set of room IDs that we want to update + # + # We use `current_state_events` table as the barometer for whether the + # server is still participating in the room because if we're + # `no_longer_in_room`, this table would be cleared out for the given + # `room_id`. txn.execute( """ SELECT DISTINCT room_id FROM current_state_events diff --git a/synapse/storage/prepare_database.py b/synapse/storage/prepare_database.py index 01559da03f..b127e83e61 100644 --- a/synapse/storage/prepare_database.py +++ b/synapse/storage/prepare_database.py @@ -42,6 +42,7 @@ from synapse.storage.database import ( LoggingDatabaseConnection, LoggingTransaction, ) +from synapse.storage.databases.main.events_bg_updates import _BackgroundUpdates from synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine from synapse.storage.schema import SCHEMA_COMPAT_VERSION, SCHEMA_VERSION from synapse.storage.types import Cursor @@ -576,12 +577,12 @@ def _upgrade_existing_database( # foreground update for # `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by # https://github.com/element-hq/synapse/issues/TODO) - _clear_stale_data_in_sliding_sync_tables( + _resolve_stale_data_in_sliding_sync_tables( txn=cur, ) -def _clear_stale_data_in_sliding_sync_tables( +def _resolve_stale_data_in_sliding_sync_tables( txn: LoggingTransaction, ) -> None: """ @@ -598,7 +599,7 @@ def _clear_stale_data_in_sliding_sync_tables( `sliding_sync_membership_snapshots`. This way, if a row exists in the sliding sync tables, we are able to rely on it - (accurate data). So if a row doesn't exist, we use a fallback to get the same info + (accurate data). And if a row doesn't exist, we use a fallback to get the same info until the background updates fill in the rows or a new event comes in triggering it to be fully inserted. @@ -608,17 +609,20 @@ def _clear_stale_data_in_sliding_sync_tables( https://github.com/element-hq/synapse/issues/TODO) """ - _clear_stale_data_in_sliding_sync_joined_rooms_table(txn) - _clear_stale_data_in_sliding_sync_membership_snapshots_table(txn) + _resolve_stale_data_in_sliding_sync_joined_rooms_table(txn) + _resolve_stale_data_in_sliding_sync_membership_snapshots_table(txn) -def _clear_stale_data_in_sliding_sync_joined_rooms_table( +def _resolve_stale_data_in_sliding_sync_joined_rooms_table( txn: LoggingTransaction, ) -> None: """ - Clears stale/out-of-date entries from the `sliding_sync_joined_rooms` table. + Clears stale/out-of-date entries from the `sliding_sync_joined_rooms` table and + kicks-off the background update to catch-up with what we missed while Synapse was + downgraded. - See `_clear_out_of_date_sliding_sync_tables()` description above for more context. + See `_resolve_stale_data_in_sliding_sync_tables()` description above for more + context. """ # Find the point when we stopped writing to the `sliding_sync_joined_rooms` table @@ -668,13 +672,16 @@ def _clear_stale_data_in_sliding_sync_joined_rooms_table( ) -def _clear_stale_data_in_sliding_sync_membership_snapshots_table( +def _resolve_stale_data_in_sliding_sync_membership_snapshots_table( txn: LoggingTransaction, ) -> None: """ - Clears stale/out-of-date entries from the `sliding_sync_membership_snapshots` table. + Clears stale/out-of-date entries from the `sliding_sync_membership_snapshots` table + and kicks-off the background update to catch-up with what we missed while Synapse + was downgraded. - See `_clear_out_of_date_sliding_sync_tables()` description above for more context. + See `_resolve_stale_data_in_sliding_sync_tables()` description above for more + context. """ # Find the point when we stopped writing to the `sliding_sync_membership_snapshots` table @@ -698,7 +705,7 @@ def _clear_stale_data_in_sliding_sync_membership_snapshots_table( txn.execute( """ SELECT DISTINCT(user_id, room_id) - FROM room_memberships + FROM local_current_membership WHERE event_stream_ordering > ? ORDER BY event_stream_ordering DESC """, @@ -723,6 +730,22 @@ def _clear_stale_data_in_sliding_sync_membership_snapshots_table( values=chunk, ) + # Now kick-off the background update to catch-up with what we missed while Synapse + # was downgraded. + DatabasePool.simple_upsert_txn_native_upsert( + txn, + table="background_updates", + keyvalues={ + "update_name": _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE + }, + values={}, + # Only insert the row if it doesn't already exist. If it already exists, we will + # eventually fill in the rows we're trying to populate. + insertion_values={ + "progress_json": f'{ "last_event_stream_ordering": {str(max_stream_ordering_sliding_sync_membership_snapshots_table)} }', + }, + ) + def _apply_module_schemas( txn: Cursor, database_engine: BaseDatabaseEngine, config: HomeServerConfig