diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index b87fa31532..481aabe7cd 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -1549,20 +1549,24 @@ class PersistEventsStore: txn, {m for m in members_to_cache_bust if not self.hs.is_mine_id(m)} ) - # TODO: We can probably remove this function in favor of other stuff. - # TODO: This doesn't take into account redactions @classmethod def _get_relevant_sliding_sync_current_state_event_ids_txn( cls, txn: LoggingTransaction, room_id: str - ) -> MutableStateMap[str]: + ) -> Tuple[MutableStateMap[str], int]: """ Fetch the current state event IDs for the relevant (to the `sliding_sync_joined_rooms` table) state types for the given room. Returns: - StateMap of event IDs necessary to to fetch the relevant state values needed - to insert into the - `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots`. + A tuple of: + 1. StateMap of event IDs necessary to to fetch the relevant state values + needed to insert into the + `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots`. + 2. The corresponding latest `stream_id` in the + `current_state_delta_stream` table. This is useful to compare against + the `current_state_delta_stream` table later so you can check whether + the current state has changed since you last fetched the current + state. """ # Fetch the current state event IDs from the database ( @@ -1587,75 +1591,25 @@ class PersistEventsStore: (event_type, state_key): event_id for event_id, event_type, state_key in txn } - return current_state_map - - # TODO: We can probably remove this function in favor of other stuff. - # TODO: Should we put this next to the other `_get_sliding_sync_*` function? - @classmethod - def _get_sliding_sync_insert_values_from_state_ids_map_txn( - cls, txn: LoggingTransaction, state_map: StateMap[str] - ) -> SlidingSyncStateInsertValues: - """ - Fetch events in the `state_map` and extract the relevant state values needed to - insert into the `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` - tables. - - Returns: - Map from column names (`room_type`, `is_encrypted`, `room_name`) to relevant - state values needed to insert into - the `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables. - """ - # Map of values to insert/update in the `sliding_sync_membership_snapshots` table - sliding_sync_insert_map: SlidingSyncStateInsertValues = {} - # Fetch the raw event JSON from the database - ( - event_id_in_list_clause, - event_id_args, - ) = make_in_list_sql_clause( - txn.database_engine, - "event_id", - state_map.values(), - ) txn.execute( - f""" - SELECT type, state_key, json FROM event_json - INNER JOIN events USING (event_id) - WHERE {event_id_in_list_clause} + """ + SELECT stream_id + FROM current_state_delta_stream + WHERE + room_id = ? + ORDER BY stream_id DESC + LIMIT 1 """, - event_id_args, + (room_id,), ) + row = txn.fetchone() + # If we're able to fetch the `current_state_events` above, we should have rows + # in `current_state_delta_stream` as well. + assert row, "Failed to fetch the `last_current_state_delta_stream_id`" + last_current_state_delta_stream_id = row[0] - # Parse the raw event JSON - for row in txn: - event_type, state_key, json = row - event_json = db_to_json(json) + return current_state_map, last_current_state_delta_stream_id - if event_type == EventTypes.Create: - room_type = event_json.get("content", {}).get( - EventContentFields.ROOM_TYPE - ) - sliding_sync_insert_map["room_type"] = room_type - elif event_type == EventTypes.RoomEncryption: - encryption_algorithm = event_json.get("content", {}).get( - EventContentFields.ENCRYPTION_ALGORITHM - ) - is_encrypted = encryption_algorithm is not None - sliding_sync_insert_map["is_encrypted"] = is_encrypted - elif event_type == EventTypes.Name: - room_name = event_json.get("content", {}).get( - EventContentFields.ROOM_NAME - ) - sliding_sync_insert_map["room_name"] = room_name - else: - # We only expect to see events according to the - # `SLIDING_SYNC_RELEVANT_STATE_SET`. - raise AssertionError( - f"Unexpected event (we should not be fetching extra events): ({event_type}, {state_key})" - ) - - return sliding_sync_insert_map - - # TODO: Should we put this next to the other `_get_sliding_sync_*` functions? @classmethod def _get_sliding_sync_insert_values_from_state_map( cls, state_map: StateMap[EventBase] @@ -1699,7 +1653,6 @@ class PersistEventsStore: return sliding_sync_insert_map - # TODO: Should we put this next to the other `_get_sliding_sync_*` function? @classmethod def _get_sliding_sync_insert_values_from_stripped_state_txn( cls, txn: LoggingTransaction, unsigned_stripped_state_events: Any diff --git a/synapse/storage/databases/main/events_bg_updates.py b/synapse/storage/databases/main/events_bg_updates.py index 8e12645d6b..b56325a4a2 100644 --- a/synapse/storage/databases/main/events_bg_updates.py +++ b/synapse/storage/databases/main/events_bg_updates.py @@ -39,11 +39,12 @@ from synapse.storage.databases.main.events import ( PersistEventsStore, SlidingSyncMembershipInfo, SlidingSyncMembershipSnapshotSharedInsertValues, + SlidingSyncStateInsertValues, ) -from synapse.storage.databases.main.events_worker import EventsWorkerStore -from synapse.storage.engines import BaseDatabaseEngine +from synapse.storage.databases.main.state_deltas import StateDeltasStore +from synapse.storage.databases.main.stream import StreamWorkerStore from synapse.storage.types import Cursor -from synapse.types import JsonDict, StateMap, StrCollection +from synapse.types import JsonDict, RoomStreamToken, StateMap, StrCollection from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES from synapse.types.state import StateFilter @@ -111,7 +112,7 @@ class _CalculateChainCover: finished_room_map: Dict[str, Tuple[int, int]] -class EventsBackgroundUpdatesStore(EventsWorkerStore, SQLBaseStore): +class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseStore): def __init__( self, database: DatabasePool, @@ -1549,46 +1550,7 @@ class EventsBackgroundUpdatesStore(EventsWorkerStore, SQLBaseStore): """ last_room_id = progress.get("last_room_id", "") - def make_sql_clause_for_get_last_event_pos_in_room( - database_engine: BaseDatabaseEngine, - event_types: Optional[StrCollection] = None, - ) -> Tuple[str, list]: - """ - Returns the ID and event position of the last event in a room at or before a - stream ordering. - - Based on `get_last_event_pos_in_room_before_stream_ordering(...)` - - Args: - database_engine - event_types: Optional allowlist of event types to filter by - - Returns: - A tuple of SQL query and the args - """ - event_type_clause = "" - event_type_args: List[str] = [] - if event_types is not None and len(event_types) > 0: - event_type_clause, event_type_args = make_in_list_sql_clause( - database_engine, "type", event_types - ) - event_type_clause = f"AND {event_type_clause}" - - sql = f""" - SELECT stream_ordering - FROM events - LEFT JOIN rejections USING (event_id) - WHERE room_id = ? - {event_type_clause} - AND NOT outlier - AND rejections.event_id IS NULL - ORDER BY stream_ordering DESC - LIMIT 1 - """ - - return sql, event_type_args - - def _txn(txn: LoggingTransaction) -> int: + def _get_rooms_to_update_txn(txn: LoggingTransaction) -> List[str]: # Fetch the set of room IDs that we want to update txn.execute( """ @@ -1600,85 +1562,162 @@ class EventsBackgroundUpdatesStore(EventsWorkerStore, SQLBaseStore): (last_room_id, batch_size), ) - rooms_to_update_rows = txn.fetchall() - if not rooms_to_update_rows: - return 0 + rooms_to_update_rows = cast(List[Tuple[str]], txn.fetchall()) - for (room_id,) in rooms_to_update_rows: - # TODO: Handle redactions - current_state_map = PersistEventsStore._get_relevant_sliding_sync_current_state_event_ids_txn( - txn, room_id + return [row[0] for row in rooms_to_update_rows] + + rooms_to_update = await self.db_pool.runInteraction( + "_sliding_sync_joined_rooms_backfill._get_rooms_to_update_txn", + _get_rooms_to_update_txn, + ) + + if not rooms_to_update: + await self.db_pool.updates._end_background_update( + _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL + ) + return 0 + + # Map from room_id to insert/update state values in the `sliding_sync_joined_rooms` table + joined_room_updates: Dict[str, SlidingSyncStateInsertValues] = {} + # Map from room_id to stream_ordering/bump_stamp/last_current_state_delta_stream_id values + joined_room_stream_ordering_updates: Dict[str, Tuple[int, int, int]] = {} + for room_id in rooms_to_update: + current_state_ids_map, last_current_state_delta_stream_id = ( + await self.db_pool.runInteraction( + "_sliding_sync_joined_rooms_backfill._get_relevant_sliding_sync_current_state_event_ids_txn", + PersistEventsStore._get_relevant_sliding_sync_current_state_event_ids_txn, + room_id, ) - # We're iterating over rooms pulled from the current_state_events table - # so we should have some current state for each room - assert current_state_map + ) + # We're iterating over rooms pulled from the current_state_events table + # so we should have some current state for each room + assert current_state_ids_map - sliding_sync_joined_rooms_insert_map = PersistEventsStore._get_sliding_sync_insert_values_from_state_ids_map_txn( - txn, current_state_map + fetched_events = await self.get_events(current_state_ids_map.values()) + + current_state_map: StateMap[EventBase] = { + state_key: fetched_events[event_id] + for state_key, event_id in current_state_ids_map.items() + } + + state_insert_values = ( + PersistEventsStore._get_sliding_sync_insert_values_from_state_map( + current_state_map ) - # We should have some insert values for each room, even if they are `None` - assert sliding_sync_joined_rooms_insert_map + ) + # We should have some insert values for each room, even if they are `None` + assert state_insert_values + joined_room_updates[room_id] = state_insert_values + # Figure out the stream_ordering of the latest event in the room + most_recent_event_pos_results = await self.get_last_event_pos_in_room( + room_id, event_types=None + ) + assert ( + most_recent_event_pos_results + ), "We should not be seeing `None` here because the room should at-least have a create event" + # Figure out the latest bump_stamp in the room + bump_stamp_event_pos_results = await self.get_last_event_pos_in_room( + room_id, event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES + ) + assert bump_stamp_event_pos_results, ( + "We should not be seeing `None` here because the room should at-least have a create event " + + "(unless `SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES` no longer includes the room create event)" + ) + joined_room_stream_ordering_updates[room_id] = ( + most_recent_event_pos_results[1].stream, + bump_stamp_event_pos_results[1].stream, + last_current_state_delta_stream_id, + ) + + def _backfill_table_txn(txn: LoggingTransaction) -> None: + last_successful_room_id: Optional[str] = None + for room_id, insert_map in joined_room_updates.items(): ( - most_recent_event_stream_ordering_clause, - most_recent_event_stream_ordering_args, - ) = make_sql_clause_for_get_last_event_pos_in_room( - txn.database_engine, event_types=None - ) - bump_stamp_clause, bump_stamp_args = ( - make_sql_clause_for_get_last_event_pos_in_room( - txn.database_engine, - event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES, + event_stream_ordering, + bump_stamp, + last_current_state_delta_stream_id, + ) = joined_room_stream_ordering_updates[room_id] + + # Check if the current state has been updated since we gathered it + state_deltas_since_we_gathered_current_state = ( + self.get_current_state_deltas_for_room_txn( + txn, + room_id, + from_token=RoomStreamToken( + stream=last_current_state_delta_stream_id + ), + to_token=None, ) ) + for state_delta in state_deltas_since_we_gathered_current_state: + # We only need to check if the state is relevant to the + # `sliding_sync_joined_rooms` table. + if ( + state_delta.event_type, + state_delta.state_key, + ) in SLIDING_SYNC_RELEVANT_STATE_SET: + # Save our progress before we exit early + if last_successful_room_id is not None: + self.db_pool.updates._background_update_progress_txn( + txn, + _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL, + {"last_room_id": room_id}, + ) + # Raising exception so we can just exit and try again + raise Exception( + "Current state was updated after we gathered it to update " + + "`sliding_sync_joined_rooms` in the background update. " + + "Raising exception so we can just try again." + ) # Pulling keys/values separately is safe and will produce congruent # lists - insert_keys = sliding_sync_joined_rooms_insert_map.keys() - insert_values = sliding_sync_joined_rooms_insert_map.values() - + insert_keys = insert_map.keys() + insert_values = insert_map.values() + # Since we partially update the `sliding_sync_joined_rooms` as new state + # is sent, we need to update the fields `ON CONFLICT`. We just have to be careful + # we're not overwriting it with stale data. sql = f""" INSERT INTO sliding_sync_joined_rooms (room_id, event_stream_ordering, bump_stamp, {", ".join(insert_keys)}) VALUES ( - ?, - ({most_recent_event_stream_ordering_clause}), - ({bump_stamp_clause}), + ?, ?, ?, {", ".join("?" for _ in insert_values)} ) ON CONFLICT (room_id) DO UPDATE SET - event_stream_ordering = EXCLUDED.event_stream_ordering, - bump_stamp = EXCLUDED.bump_stamp, + event_stream_ordering = CASE + WHEN event_stream_ordering IS NULL OR event_stream_ordering < EXCLUDED.event_stream_ordering + THEN EXCLUDED.event_stream_ordering + ELSE event_stream_ordering + END, + bump_stamp = CASE + WHEN bump_stamp IS NULL OR bump_stamp < EXCLUDED.bump_stamp + THEN EXCLUDED.bump_stamp + ELSE bump_stamp + END, {", ".join(f"{key} = EXCLUDED.{key}" for key in insert_keys)} """ - args = ( - [room_id, room_id] - + most_recent_event_stream_ordering_args - + [room_id] - + bump_stamp_args - + list(insert_values) + args = [room_id, event_stream_ordering, bump_stamp] + list( + insert_values ) txn.execute(sql, args) - self.db_pool.updates._background_update_progress_txn( - txn, - _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL, - {"last_room_id": rooms_to_update_rows[-1][0]}, - ) + # Keep track of the last successful room_id + last_successful_room_id = room_id - return len(rooms_to_update_rows) - - count = await self.db_pool.runInteraction( - "sliding_sync_joined_rooms_backfill", _txn + await self.db_pool.runInteraction( + "sliding_sync_joined_rooms_backfill", _backfill_table_txn ) - if not count: - await self.db_pool.updates._end_background_update( - _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL - ) + # Update the progress + await self.db_pool.updates._background_update_progress( + _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL, + {"last_room_id": rooms_to_update[-1]}, + ) - return count + return len(rooms_to_update) async def _sliding_sync_membership_snapshots_backfill( self, progress: JsonDict, batch_size: int diff --git a/synapse/storage/databases/main/state_deltas.py b/synapse/storage/databases/main/state_deltas.py index eaa13da368..ba52fff652 100644 --- a/synapse/storage/databases/main/state_deltas.py +++ b/synapse/storage/databases/main/state_deltas.py @@ -161,45 +161,80 @@ class StateDeltasStore(SQLBaseStore): self._get_max_stream_id_in_current_state_deltas_txn, ) + def get_current_state_deltas_for_room_txn( + self, + txn: LoggingTransaction, + room_id: str, + *, + from_token: Optional[RoomStreamToken], + to_token: Optional[RoomStreamToken], + ) -> List[StateDelta]: + """ + Get the state deltas between two tokens. + + (> `from_token` and <= `to_token`) + """ + from_clause = "" + from_args = [] + if from_token is not None: + from_clause = "AND ? < stream_id" + from_args = [from_token.stream] + + to_clause = "" + to_args = [] + if to_token is not None: + to_clause = "AND stream_id <= ?" + to_args = [to_token.get_max_stream_pos()] + + sql = f""" + SELECT instance_name, stream_id, type, state_key, event_id, prev_event_id + FROM current_state_delta_stream + WHERE room_id = ? {from_clause} {to_clause} + ORDER BY stream_id ASC + """ + txn.execute(sql, [room_id] + from_args + to_args) + + return [ + StateDelta( + stream_id=row[1], + room_id=room_id, + event_type=row[2], + state_key=row[3], + event_id=row[4], + prev_event_id=row[5], + ) + for row in txn + if _filter_results_by_stream(from_token, to_token, row[0], row[1]) + ] + @trace async def get_current_state_deltas_for_room( - self, room_id: str, from_token: RoomStreamToken, to_token: RoomStreamToken + self, + room_id: str, + *, + from_token: Optional[RoomStreamToken], + to_token: Optional[RoomStreamToken], ) -> List[StateDelta]: - """Get the state deltas between two tokens.""" + """ + Get the state deltas between two tokens. - if not self._curr_state_delta_stream_cache.has_entity_changed( - room_id, from_token.stream + (> `from_token` and <= `to_token`) + """ + + if ( + from_token is not None + and not self._curr_state_delta_stream_cache.has_entity_changed( + room_id, from_token.stream + ) ): return [] - def get_current_state_deltas_for_room_txn( - txn: LoggingTransaction, - ) -> List[StateDelta]: - sql = """ - SELECT instance_name, stream_id, type, state_key, event_id, prev_event_id - FROM current_state_delta_stream - WHERE room_id = ? AND ? < stream_id AND stream_id <= ? - ORDER BY stream_id ASC - """ - txn.execute( - sql, (room_id, from_token.stream, to_token.get_max_stream_pos()) - ) - - return [ - StateDelta( - stream_id=row[1], - room_id=room_id, - event_type=row[2], - state_key=row[3], - event_id=row[4], - prev_event_id=row[5], - ) - for row in txn - if _filter_results_by_stream(from_token, to_token, row[0], row[1]) - ] - return await self.db_pool.runInteraction( - "get_current_state_deltas_for_room", get_current_state_deltas_for_room_txn + "get_current_state_deltas_for_room", + self.get_current_state_deltas_for_room_txn, + room_id, + from_token=from_token, + to_token=to_token, ) @trace diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 3054174717..395735cd43 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1263,6 +1263,67 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore): return None + async def get_last_event_pos_in_room( + self, + room_id: str, + event_types: Optional[StrCollection] = None, + ) -> Optional[Tuple[str, PersistedEventPosition]]: + """ + Returns the ID and event position of the last event in a room. + + Based on `get_last_event_pos_in_room_before_stream_ordering(...)` + + Args: + room_id + event_types: Optional allowlist of event types to filter by + + Returns: + The ID of the most recent event and it's position, or None if there are no + events in the room that match the given event types. + """ + + def _get_last_event_pos_in_room_txn( + txn: LoggingTransaction, + ) -> Optional[Tuple[str, PersistedEventPosition]]: + event_type_clause = "" + event_type_args: List[str] = [] + if event_types is not None and len(event_types) > 0: + event_type_clause, event_type_args = make_in_list_sql_clause( + txn.database_engine, "type", event_types + ) + event_type_clause = f"AND {event_type_clause}" + + sql = f""" + SELECT event_id, stream_ordering, instance_name + FROM events + LEFT JOIN rejections USING (event_id) + WHERE room_id = ? + {event_type_clause} + AND NOT outlier + AND rejections.event_id IS NULL + ORDER BY stream_ordering DESC + LIMIT 1 + """ + + txn.execute( + sql, + [room_id] + event_type_args, + ) + + row = cast(Tuple[str, int, str], txn.fetchone()) + event_id, stream_ordering, instance_name = row + + return event_id, PersistedEventPosition( + # If instance_name is null we default to "master" + instance_name or "master", + stream_ordering, + ) + + return await self.db_pool.runInteraction( + "get_last_event_pos_in_room", + _get_last_event_pos_in_room_txn, + ) + @trace async def get_last_event_pos_in_room_before_stream_ordering( self,