From 1c931cb3e708091b198540b34643142d15d14564 Mon Sep 17 00:00:00 2001 From: Eric Eastwood Date: Wed, 14 Aug 2024 19:19:15 -0500 Subject: [PATCH] Add background update for `sliding_sync_joined_rooms` --- synapse/storage/databases/main/events.py | 210 +++++++++------- .../databases/main/events_bg_updates.py | 230 ++++++++++++++++++ synapse/storage/databases/main/stream.py | 2 +- .../delta/87/01_sliding_sync_memberships.sql | 19 +- tests/storage/test_events.py | 104 ++++++++ 5 files changed, 479 insertions(+), 86 deletions(-) diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 3f9ca26321..62a203e252 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -93,6 +93,17 @@ event_counter = Counter( ["type", "origin_type", "origin_entity"], ) +# State event type/key pairs that we need to gather to fill in the +# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` tables. +SLIDING_SYNC_RELEVANT_STATE_SET = { + # So we can fill in the `room_type` column in the `sliding_sync_joined_rooms` table + (EventTypes.Create, ""), + # So we can fill in the `is_encrypted` column in the `sliding_sync_joined_rooms` table + (EventTypes.RoomEncryption, ""), + # So we can fill in the `room_name` column in the `sliding_sync_joined_rooms` table + (EventTypes.Name, ""), +} + @attr.s(slots=True, auto_attribs=True) class DeltaState: @@ -1211,35 +1222,11 @@ class PersistEventsStore: membership_event_id_to_user_id_map[event_id] = state_key[1] if len(membership_event_id_to_user_id_map) > 0: - relevant_state_set = { - (EventTypes.Create, ""), - (EventTypes.RoomEncryption, ""), - (EventTypes.Name, ""), - } - - # Fetch the current state event IDs from the database - ( - event_type_and_state_key_in_list_clause, - event_type_and_state_key_args, - ) = make_tuple_in_list_sql_clause( - self.database_engine, - ("type", "state_key"), - relevant_state_set, + current_state_map = ( + self._get_relevant_sliding_sync_current_state_event_ids_txn( + txn, room_id + ) ) - txn.execute( - f""" - SELECT c.event_id, c.type, c.state_key - FROM current_state_events AS c - WHERE - c.room_id = ? - AND {event_type_and_state_key_in_list_clause} - """, - [room_id] + event_type_and_state_key_args, - ) - current_state_map: MutableStateMap[str] = { - (event_type, state_key): event_id - for event_id, event_type, state_key in txn - } # Since we fetched the current state before we took `to_insert`/`to_delete` # into account, we need to do a couple fixups. # @@ -1248,7 +1235,7 @@ class PersistEventsStore: current_state_map.pop(state_key, None) # Update the current_state_map with what we have `to_insert` for state_key, event_id in to_insert.items(): - if state_key in relevant_state_set: + if state_key in SLIDING_SYNC_RELEVANT_STATE_SET: current_state_map[state_key] = event_id # Map of values to insert/update in the `sliding_sync_membership_snapshots` table @@ -1256,60 +1243,13 @@ class PersistEventsStore: str, Optional[Union[str, bool]] ] = {} if current_state_map: + sliding_sync_membership_snapshots_insert_map = self._get_sliding_sync_insert_values_according_to_current_state_map_txn( + txn, current_state_map + ) # We have current state to work from sliding_sync_membership_snapshots_insert_map["has_known_state"] = ( True ) - - # Fetch the raw event JSON from the database - ( - event_id_in_list_clause, - event_id_args, - ) = make_in_list_sql_clause( - self.database_engine, - "event_id", - current_state_map.values(), - ) - txn.execute( - f""" - SELECT event_id, type, state_key, json FROM event_json - INNER JOIN events USING (event_id) - WHERE {event_id_in_list_clause} - """, - event_id_args, - ) - - # Parse the raw event JSON - for row in txn: - event_id, event_type, state_key, json = row - event_json = db_to_json(json) - - if event_type == EventTypes.Create: - room_type = event_json.get("content", {}).get( - EventContentFields.ROOM_TYPE - ) - sliding_sync_membership_snapshots_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_membership_snapshots_insert_map[ - "is_encrypted" - ] = is_encrypted - elif event_type == EventTypes.Name: - room_name = event_json.get("content", {}).get( - EventContentFields.ROOM_NAME - ) - sliding_sync_membership_snapshots_insert_map[ - "room_name" - ] = room_name - else: - raise AssertionError( - f"Unexpected event (we should not be fetching extra events): ({event_type}, {state_key})" - ) else: # We don't have any `current_state_events` anymore (previously # cleared out because of `no_longer_in_room`). This can happen if @@ -1468,7 +1408,12 @@ class PersistEventsStore: ], ) - # Handle updating the `sliding_sync_joined_rooms` table + # Handle updating the `sliding_sync_joined_rooms` table. We only deal with + # updating the state related columns. The + # `event_stream_ordering`/`bump_stamp` are updated elsewhere in the event + # persisting stack (see + # `_update_sliding_sync_tables_with_new_persisted_events_txn()`) + # event_ids_to_fetch: List[str] = [] create_event_id = None room_encryption_event_id = None @@ -1574,8 +1519,10 @@ class PersistEventsStore: args.extend(iter(insert_values)) - # We don't update `event_stream_ordering` `ON CONFLICT` because it's simpler - # we can just + # We don't update `event_stream_ordering` `ON CONFLICT` because it's + # simpler and we can just rely on + # `_update_sliding_sync_tables_with_new_persisted_events_txn()` to do + # the right thing. # # We don't update `bump_stamp` `ON CONFLICT` because we're dealing with # state here and the only state event that is also a bump event type is @@ -1653,6 +1600,105 @@ class PersistEventsStore: txn, {m for m in members_to_cache_bust if not self.hs.is_mine_id(m)} ) + @classmethod + def _get_relevant_sliding_sync_current_state_event_ids_txn( + cls, txn: LoggingTransaction, room_id: str + ) -> MutableStateMap[str]: + """ + Fetch the current state event IDs for the relevant (to the + `sliding_sync_joined_rooms` table) state types for the given room. + + TODO + """ + # Fetch the current state event IDs from the database + ( + event_type_and_state_key_in_list_clause, + event_type_and_state_key_args, + ) = make_tuple_in_list_sql_clause( + txn.database_engine, + ("type", "state_key"), + SLIDING_SYNC_RELEVANT_STATE_SET, + ) + txn.execute( + f""" + SELECT c.event_id, c.type, c.state_key + FROM current_state_events AS c + WHERE + c.room_id = ? + AND {event_type_and_state_key_in_list_clause} + """, + [room_id] + event_type_and_state_key_args, + ) + current_state_map: MutableStateMap[str] = { + (event_type, state_key): event_id for event_id, event_type, state_key in txn + } + + return current_state_map + + @classmethod + def _get_sliding_sync_insert_values_according_to_current_state_map_txn( + cls, txn: LoggingTransaction, current_state_map: StateMap[str] + ) -> Dict[str, Optional[Union[str, bool]]]: + """ + TODO + + 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: Dict[str, Optional[Union[str, bool]]] = {} + # 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", + current_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} + """, + event_id_args, + ) + + # Parse the raw event JSON + for row in txn: + event_type, state_key, json = row + event_json = db_to_json(json) + + 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` which is what will + # `_get_relevant_sliding_sync_current_state_event_ids_txn()` will + # return. + raise AssertionError( + f"Unexpected event (we should not be fetching extra events): ({event_type}, {state_key})" + ) + + return sliding_sync_insert_map + def _update_sliding_sync_tables_with_new_persisted_events_txn( self, txn: LoggingTransaction, diff --git a/synapse/storage/databases/main/events_bg_updates.py b/synapse/storage/databases/main/events_bg_updates.py index 64d303e330..b59bc2a561 100644 --- a/synapse/storage/databases/main/events_bg_updates.py +++ b/synapse/storage/databases/main/events_bg_updates.py @@ -35,8 +35,10 @@ from synapse.storage.database import ( make_tuple_comparison_clause, ) from synapse.storage.databases.main.events import PersistEventsStore +from synapse.storage.engines import BaseDatabaseEngine from synapse.storage.types import Cursor from synapse.types import JsonDict, StrCollection +from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES if TYPE_CHECKING: from synapse.server import HomeServer @@ -78,6 +80,11 @@ class _BackgroundUpdates: EVENTS_JUMP_TO_DATE_INDEX = "events_jump_to_date_index" + SLIDING_SYNC_JOINED_ROOMS_BACKFILL = "sliding_sync_joined_rooms_backfill" + SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BACKFILL = ( + "sliding_sync_membership_snapshots_backfill" + ) + @attr.s(slots=True, frozen=True, auto_attribs=True) class _CalculateChainCover: @@ -279,6 +286,16 @@ class EventsBackgroundUpdatesStore(SQLBaseStore): where_clause="NOT outlier", ) + # Backfill the sliding sync tables + self.db_pool.updates.register_background_update_handler( + _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL, + self._sliding_sync_joined_rooms_backfill, + ) + self.db_pool.updates.register_background_update_handler( + _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BACKFILL, + self._sliding_sync_membership_snapshots_backfill, + ) + async def _background_reindex_fields_sender( self, progress: JsonDict, batch_size: int ) -> int: @@ -1516,3 +1533,216 @@ class EventsBackgroundUpdatesStore(SQLBaseStore): ) return batch_size + + async def _sliding_sync_joined_rooms_backfill( + self, progress: JsonDict, batch_size: int + ) -> int: + """ + Handles backfilling the `sliding_sync_joined_rooms` table. + """ + 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: + # Fetch the set of room IDs that we want to update + txn.execute( + """ + SELECT DISTINCT room_id FROM current_state_events + WHERE room_id > ? + ORDER BY room_id ASC + LIMIT ? + """, + (last_room_id, batch_size), + ) + + rooms_to_update_rows = txn.fetchall() + if not rooms_to_update_rows: + return 0 + + for (room_id,) in rooms_to_update_rows: + logger.info("asdf Working on room %s", room_id) + current_state_map = PersistEventsStore._get_relevant_sliding_sync_current_state_event_ids_txn( + 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 + + sliding_sync_joined_rooms_insert_map = PersistEventsStore._get_sliding_sync_insert_values_according_to_current_state_map_txn( + txn, current_state_map + ) + # We should have some insert values for each room, even if they are `None` + assert sliding_sync_joined_rooms_insert_map + + ( + 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, + ) + ) + + # 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() + + 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, + {", ".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) + ) + 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]}, + ) + + return len(rooms_to_update_rows) + + count = await self.db_pool.runInteraction( + "sliding_sync_joined_rooms_backfill", _txn + ) + + if not count: + await self.db_pool.updates._end_background_update( + _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL + ) + + return count + + async def _sliding_sync_membership_snapshots_backfill( + self, progress: JsonDict, batch_size: int + ) -> int: + """ + Handles backfilling the `sliding_sync_membership_snapshots` table. + """ + # last_event_stream_ordering = progress.get("last_event_stream_ordering", "") + + def _txn(txn: LoggingTransaction) -> int: + # # Fetch the set of event IDs that we want to update + # txn.execute( + # """ + # SELECT room_id, user_id, event_id FROM local_current_membership + # WHERE event_stream_ordering > ? + # ORDER BY event_stream_ordering ASC + # LIMIT ? + # """, + # (last_event_stream_ordering, batch_size), + # ) + + # rows = txn.fetchall() + # if not rows: + # return 0 + + # # Update the redactions with the received_ts. + # # + # # Note: Not all events have an associated received_ts, so we + # # fallback to using origin_server_ts. If we for some reason don't + # # have an origin_server_ts, lets just use the current timestamp. + # # + # # We don't want to leave it null, as then we'll never try and + # # censor those redactions. + # txn.execute_batch( + # f""" + # INSERT INTO sliding_sync_membership_snapshots + # (room_id, user_id, membership_event_id, membership, event_stream_ordering + # {"," + (", ".join(insert_keys)) if insert_keys else ""}) + # VALUES ( + # ?, ?, ?, + # (SELECT membership FROM room_memberships WHERE event_id = ?), + # (SELECT stream_ordering FROM events WHERE event_id = ?) + # {"," + (", ".join("?" for _ in insert_values)) if insert_values else ""} + # ) + # ON CONFLICT (room_id, user_id) + # DO UPDATE SET + # membership_event_id = EXCLUDED.membership_event_id, + # membership = EXCLUDED.membership, + # event_stream_ordering = EXCLUDED.event_stream_ordering + # {"," + (", ".join(f"{key} = EXCLUDED.{key}" for key in insert_keys)) if insert_keys else ""} + # """, + # (TODO,), + # ) + + # self.db_pool.updates._background_update_progress_txn( + # txn, "redactions_received_ts", {"last_event_id": upper_event_id} + # ) + + # return len(rows) + + # TODO + # return len(rows) + return 0 + + count = await self.db_pool.runInteraction( + "sliding_sync_membership_snapshots_backfill", _txn + ) + + if not count: + await self.db_pool.updates._end_background_update( + _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BACKFILL + ) + + return count diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 4989c960a6..3054174717 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1268,7 +1268,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore): self, room_id: str, end_token: RoomStreamToken, - event_types: Optional[Collection[str]] = None, + event_types: Optional[StrCollection] = None, ) -> Optional[Tuple[str, PersistedEventPosition]]: """ Returns the ID and event position of the last event in a room at or before a diff --git a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql index c8c671cf6c..16b3f84c3d 100644 --- a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql +++ b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql @@ -22,7 +22,7 @@ -- no longer participating in a room, the row will be deleted. CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms( room_id TEXT NOT NULL REFERENCES rooms(room_id), - -- The `stream_ordering` of the latest event in the room + -- The `stream_ordering` of the most-recent/latest event in the room event_stream_ordering BIGINT NOT NULL REFERENCES events(stream_ordering), -- The `stream_ordering` of the last event according to the `bump_event_types` bump_stamp BIGINT, @@ -83,9 +83,22 @@ CREATE TABLE IF NOT EXISTS sliding_sync_membership_snapshots( PRIMARY KEY (room_id, user_id) ); --- So we can purge rooms easily -CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_room_id ON sliding_sync_membership_snapshots(room_id); +-- So we can purge rooms easily. +-- +-- Since we're using a multi-column index as the primary key (room_id, user_id), the +-- first index column (room_id) is always usable for searching so we don't need to +-- create a separate index for it. +-- +-- CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_room_id ON sliding_sync_membership_snapshots(room_id); + -- So we can fetch all rooms for a given user CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_user_id ON sliding_sync_membership_snapshots(user_id); -- So we can sort by `stream_ordering CREATE UNIQUE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_event_stream_ordering ON sliding_sync_membership_snapshots(event_stream_ordering); + + +-- Add some background updates to populate the new tables +INSERT INTO background_updates (ordering, update_name, progress_json) VALUES + (8701, 'sliding_sync_joined_rooms_backfill', '{}'); +INSERT INTO background_updates (ordering, update_name, progress_json) VALUES + (8701, 'sliding_sync_membership_snapshots_backfill', '{}'); diff --git a/tests/storage/test_events.py b/tests/storage/test_events.py index 99b3a5676e..70ba415f7f 100644 --- a/tests/storage/test_events.py +++ b/tests/storage/test_events.py @@ -32,6 +32,7 @@ from synapse.api.room_versions import RoomVersions from synapse.events import EventBase, StrippedStateEvent, make_event_from_dict from synapse.events.snapshot import EventContext from synapse.federation.federation_base import event_from_pdu_json +from synapse.storage.databases.main.events_bg_updates import _BackgroundUpdates from synapse.rest import admin from synapse.rest.client import login, room from synapse.server import HomeServer @@ -2378,3 +2379,106 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): ) # TODO: test_non_join_state_reset + + def test_joined_background_update_missing(self) -> None: + """ + Test that the background update for `sliding_sync_joined_rooms` backfills missing rows + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + + # Create rooms with various levels of state that should appear in the table + # + room_id_no_info = self.helper.create_room_as(user1_id, tok=user1_tok) + + room_id_with_info = self.helper.create_room_as(user1_id, tok=user1_tok) + # Add a room name + self.helper.send_state( + room_id_with_info, + EventTypes.Name, + {"name": "my super duper room"}, + tok=user1_tok, + ) + # Encrypt the room + self.helper.send_state( + room_id_with_info, + EventTypes.RoomEncryption, + {EventContentFields.ENCRYPTION_ALGORITHM: "m.megolm.v1.aes-sha2"}, + tok=user1_tok, + ) + + space_room_id = self.helper.create_room_as( + user1_id, + tok=user1_tok, + extra_content={ + "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE} + }, + ) + # Add a room name + self.helper.send_state( + space_room_id, + EventTypes.Name, + {"name": "my super duper space"}, + tok=user1_tok, + ) + + # Clean-up the `sliding_sync_joined_rooms` table as if the inserts did not + # happen during event creation. + self.get_success( + self.store.db_pool.simple_delete_many( + table="sliding_sync_joined_rooms", + column="room_id", + iterable=(room_id_no_info, room_id_with_info, space_room_id), + keyvalues={}, + desc="RelationsTestCase.test_background_update", + ) + ) + + # We shouldn't find anything in the table because we just deleted them in + # preparation for the test. + sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() + self.assertIncludes( + set(sliding_sync_joined_rooms_results.keys()), + set(), + exact=True, + ) + + # Insert and run the background update. + self.get_success( + self.store.db_pool.simple_insert( + "background_updates", + { + "update_name": _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BACKFILL, + "progress_json": "{}", + }, + ) + ) + # Ugh, have to reset this flag + self.store.db_pool.updates._all_done = False + self.wait_for_background_updates() + + sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() + self.assertIncludes( + set(sliding_sync_joined_rooms_results.keys()), + {room_id_no_info, room_id_with_info, space_room_id}, + exact=True, + ) + # self.assertEqual( + # sliding_sync_joined_rooms_results[room_id1], + # _SlidingSyncJoinedRoomResult( + # room_id=room_id1, + # # Latest event in the room + # event_stream_ordering=room_name_update_event_pos.stream, + # bump_stamp=state_map[ + # (EventTypes.Create, "") + # ].internal_metadata.stream_ordering, + # room_type=None, + # room_name="my super duper room", + # is_encrypted=False, + # ), + # ) + + def test_joined_background_update_partial(self) -> None: + """ + Test that the background update for `sliding_sync_joined_rooms` backfills partially updated rows + """