diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index afb93e42b4..b5e561142f 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -61,7 +61,7 @@ from synapse.types import ( ) from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES from synapse.types.handlers.sliding_sync import ( - HaveSentRoomFlag, + HaveSentFlag, MutablePerConnectionState, PerConnectionState, RoomLazyMembershipChanges, @@ -669,14 +669,14 @@ class SlidingSyncHandler: ignore_timeline_bound = False if from_token and not newly_joined and not state_reset_out_of_room: room_status = previous_connection_state.rooms.have_sent_room(room_id) - if room_status.status == HaveSentRoomFlag.LIVE: + if room_status.status == HaveSentFlag.LIVE: from_bound = from_token.stream_token.room_key initial = False - elif room_status.status == HaveSentRoomFlag.PREVIOUSLY: + elif room_status.status == HaveSentFlag.PREVIOUSLY: assert room_status.last_token is not None from_bound = room_status.last_token initial = False - elif room_status.status == HaveSentRoomFlag.NEVER: + elif room_status.status == HaveSentFlag.NEVER: from_bound = None initial = True else: diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 0e787f36c5..ada65d1c49 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -49,7 +49,7 @@ from synapse.types import ( UserID, ) from synapse.types.handlers.sliding_sync import ( - HaveSentRoomFlag, + HaveSentFlag, MutablePerConnectionState, OperationType, PerConnectionState, @@ -537,12 +537,12 @@ class SlidingSyncExtensionHandler: room_status = previous_connection_state.account_data.have_sent_room( room_id ) - if room_status.status == HaveSentRoomFlag.LIVE: + if room_status.status == HaveSentFlag.LIVE: live_rooms.add(room_id) - elif room_status.status == HaveSentRoomFlag.PREVIOUSLY: + elif room_status.status == HaveSentFlag.PREVIOUSLY: assert room_status.last_token is not None previously_rooms[room_id] = room_status.last_token - elif room_status.status == HaveSentRoomFlag.NEVER: + elif room_status.status == HaveSentFlag.NEVER: initial_rooms.add(room_id) else: assert_never(room_status.status) @@ -747,12 +747,12 @@ class SlidingSyncExtensionHandler: continue room_status = previous_connection_state.receipts.have_sent_room(room_id) - if room_status.status == HaveSentRoomFlag.LIVE: + if room_status.status == HaveSentFlag.LIVE: live_rooms.add(room_id) - elif room_status.status == HaveSentRoomFlag.PREVIOUSLY: + elif room_status.status == HaveSentFlag.PREVIOUSLY: assert room_status.last_token is not None previously_rooms[room_id] = room_status.last_token - elif room_status.status == HaveSentRoomFlag.NEVER: + elif room_status.status == HaveSentFlag.NEVER: initial_rooms.add(room_id) else: assert_never(room_status.status) @@ -867,7 +867,7 @@ class SlidingSyncExtensionHandler: rooms_no_receipts = [ room_id for room_id, room_status in previous_connection_state.receipts._statuses.items() - if room_status.status == HaveSentRoomFlag.LIVE + if room_status.status == HaveSentFlag.LIVE and room_id not in relevant_room_ids ] changed_rooms = await self.store.get_rooms_with_receipts_between( diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 836cee6c20..45120a2968 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -62,7 +62,7 @@ from synapse.types import ( UserID, ) from synapse.types.handlers.sliding_sync import ( - HaveSentRoomFlag, + HaveSentFlag, OperationType, PerConnectionState, RoomSyncConfig, @@ -893,14 +893,14 @@ class SlidingSyncRoomLists: if ( # The room was never sent down before so the client needs to know # about it regardless of any updates. - status.status == HaveSentRoomFlag.NEVER + status.status == HaveSentFlag.NEVER # `PREVIOUSLY` literally means the "room was sent down before *AND* # there are updates we haven't sent down" so we already know this # room has updates. - or status.status == HaveSentRoomFlag.PREVIOUSLY + or status.status == HaveSentFlag.PREVIOUSLY ): rooms_should_send.add(room_id) - elif status.status == HaveSentRoomFlag.LIVE: + elif status.status == HaveSentFlag.LIVE: # We know that we've sent all updates up until `from_token`, # so we just need to check if there have been updates since # then. @@ -2002,7 +2002,7 @@ class SlidingSyncRoomLists: for room_id in sync_room_map.keys() if sync_room_map[room_id].event_id is None and previous_connection_state.rooms.have_sent_room(room_id).status - != HaveSentRoomFlag.NEVER + != HaveSentFlag.NEVER } # Assemble a new sync room map but only with the `filtered_room_id_set` @@ -2170,7 +2170,7 @@ class SlidingSyncRoomLists: for room_id in sync_room_map.keys() if sync_room_map[room_id].event_id is None and previous_connection_state.rooms.have_sent_room(room_id).status - != HaveSentRoomFlag.NEVER + != HaveSentFlag.NEVER } # Assemble a new sync room map but only with the `filtered_room_id_set` diff --git a/synapse/handlers/sliding_sync/store.py b/synapse/handlers/sliding_sync/store.py index 65febe58aa..2439c0123b 100644 --- a/synapse/handlers/sliding_sync/store.py +++ b/synapse/handlers/sliding_sync/store.py @@ -55,7 +55,7 @@ class SlidingSyncConnectionStore: Attributes: _connections: Mapping from `(user_id, conn_id)` to mapping of `token` - to mapping of room ID to `HaveSentRoom`. + to mapping of room ID to `HaveSent`. """ clock: Clock diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py index 0fc3f7a436..ece89e1836 100644 --- a/synapse/storage/databases/main/sliding_sync.py +++ b/synapse/storage/databases/main/sliding_sync.py @@ -31,8 +31,8 @@ from synapse.storage.database import ( from synapse.storage.engines import PostgresEngine from synapse.types import MultiWriterStreamToken, RoomStreamToken from synapse.types.handlers.sliding_sync import ( - HaveSentRoom, - HaveSentRoomFlag, + HaveSent, + HaveSentFlag, MutablePerConnectionState, PerConnectionState, ProfileFieldStatusMap, @@ -635,9 +635,9 @@ class SlidingSyncStore(SQLBaseStore): ) # Now look up the per-room stream data. - rooms: dict[str, HaveSentRoom[str]] = {} - receipts: dict[str, HaveSentRoom[str]] = {} - account_data: dict[str, HaveSentRoom[str]] = {} + rooms: dict[str, HaveSent[str]] = {} + receipts: dict[str, HaveSent[str]] = {} + account_data: dict[str, HaveSent[str]] = {} receipt_rows = self.db_pool.simple_select_list_txn( txn, @@ -651,8 +651,8 @@ class SlidingSyncStore(SQLBaseStore): ), ) for stream, room_id, room_status, last_token in receipt_rows: - have_sent_room: HaveSentRoom[str] = HaveSentRoom( - status=HaveSentRoomFlag(room_status), last_token=last_token + have_sent_room: HaveSent[str] = HaveSent( + status=HaveSentFlag(room_status), last_token=last_token ) if stream == "rooms": rooms[room_id] = have_sent_room @@ -666,7 +666,7 @@ class SlidingSyncStore(SQLBaseStore): logger.warning("Unrecognized sliding sync stream in DB %r", stream) # Now look up the per-profile field stream data. - profile_updates: dict[str, dict[str, HaveSentRoom[str]]] = {} + profile_updates: dict[str, dict[str, HaveSent[str]]] = {} profile_update_rows = self.db_pool.simple_select_list_txn( txn, @@ -680,8 +680,8 @@ class SlidingSyncStore(SQLBaseStore): ), ) for user_id, field_name, field_status, last_token in profile_update_rows: - have_sent_field: HaveSentRoom[str] = HaveSentRoom( - status=HaveSentRoomFlag(field_status), last_token=last_token + have_sent_field: HaveSent[str] = HaveSent( + status=HaveSentFlag(field_status), last_token=last_token ) if user_id not in profile_updates: profile_updates[user_id] = {} @@ -889,7 +889,7 @@ class PerConnectionStateDB: ) -> "PerConnectionStateDB": """Convert from a standard `PerConnectionState`""" rooms = { - room_id: HaveSentRoom( + room_id: HaveSent( status=status.status, last_token=( await status.last_token.to_string(store) @@ -901,7 +901,7 @@ class PerConnectionStateDB: } receipts = { - room_id: HaveSentRoom( + room_id: HaveSent( status=status.status, last_token=( await status.last_token.to_string(store) @@ -913,7 +913,7 @@ class PerConnectionStateDB: } account_data = { - room_id: HaveSentRoom( + room_id: HaveSent( status=status.status, last_token=( str(status.last_token) if status.last_token is not None else None @@ -922,13 +922,13 @@ class PerConnectionStateDB: for room_id, status in per_connection_state.account_data.get_updates().items() } - profile_updates: dict[str, dict[str, HaveSentRoom[str]]] = {} + profile_updates: dict[str, dict[str, HaveSent[str]]] = {} for ( user_id, field_statuses, ) in per_connection_state.profile_updates.get_updates().items(): profile_updates[user_id] = { - field_name: HaveSentRoom( + field_name: HaveSent( status=status.status, last_token=( await status.last_token.to_string(store) @@ -962,7 +962,7 @@ class PerConnectionStateDB: async def to_state(self, store: "DataStore") -> "PerConnectionState": """Convert into a standard `PerConnectionState`""" rooms = { - room_id: HaveSentRoom( + room_id: HaveSent( status=status.status, last_token=( await RoomStreamToken.parse(store, status.last_token) @@ -974,7 +974,7 @@ class PerConnectionStateDB: } receipts = { - room_id: HaveSentRoom( + room_id: HaveSent( status=status.status, last_token=( await MultiWriterStreamToken.parse(store, status.last_token) @@ -986,7 +986,7 @@ class PerConnectionStateDB: } account_data = { - room_id: HaveSentRoom( + room_id: HaveSent( status=status.status, last_token=( int(status.last_token) if status.last_token is not None else None @@ -995,10 +995,10 @@ class PerConnectionStateDB: for room_id, status in self.account_data._statuses.items() } - profile_updates: dict[str, dict[str, HaveSentRoom[MultiWriterStreamToken]]] = {} + profile_updates: dict[str, dict[str, HaveSent[MultiWriterStreamToken]]] = {} for user_id, field_statuses in self.profile_updates._statuses.items(): profile_updates[user_id] = { - field_name: HaveSentRoom( + field_name: HaveSent( status=status.status, last_token=( await MultiWriterStreamToken.parse(store, status.last_token) diff --git a/synapse/storage/schema/main/delta/87/02_per_connection_state.sql b/synapse/storage/schema/main/delta/87/02_per_connection_state.sql index 59bc14a2c9..7a424c184e 100644 --- a/synapse/storage/schema/main/delta/87/02_per_connection_state.sql +++ b/synapse/storage/schema/main/delta/87/02_per_connection_state.sql @@ -74,7 +74,7 @@ CREATE TABLE sliding_sync_connection_streams( connection_position BIGINT NOT NULL REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE, stream TEXT NOT NULL, -- e.g. "events" or "receipts" room_id TEXT NOT NULL, - room_status TEXT NOT NULL, -- "live" or "previously", i.e. the `HaveSentRoomFlag` value + room_status TEXT NOT NULL, -- "live" or "previously", i.e. the `HaveSentFlag` value last_token TEXT -- For "previously" the token for the stream we have sent up to. ); diff --git a/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql b/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql index 124ff9e463..f599ae7cef 100644 --- a/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql +++ b/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql @@ -19,7 +19,7 @@ CREATE TABLE sliding_sync_connection_profile_updates( connection_position BIGINT NOT NULL REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE, user_id TEXT NOT NULL, field_name TEXT NOT NULL, - field_status TEXT NOT NULL, -- "live" or "previously", i.e. the `HaveSentRoomFlag` value + field_status TEXT NOT NULL, -- "live" or "previously", i.e. the `HaveSentFlag` value last_token TEXT -- For "previously" the token for the stream we have sent up to. ); diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index 0c62f9a4c7..dcb46758be 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -767,8 +767,8 @@ class RoomSyncConfig: return False -class HaveSentRoomFlag(Enum): - """Flag for whether we have sent the room down a sliding sync connection. +class HaveSentFlag(Enum): + """Flag for whether we have sent some data down a sliding sync connection. The valid state changes here are: NEVER -> LIVE @@ -776,15 +776,15 @@ class HaveSentRoomFlag(Enum): PREVIOUSLY -> LIVE """ - # The room has never been sent down (or we have forgotten we have sent it + # The data has never been sent down (or we have forgotten we have sent it # down). NEVER = "never" - # We have previously sent the room down, but there are updates that we + # We have previously sent the data down, but there are updates that we # haven't sent down. PREVIOUSLY = "previously" - # We have sent the room down and the client has received all updates. + # We have sent the data down and the client has received all updates. LIVE = "live" @@ -792,40 +792,40 @@ T = TypeVar("T", str, RoomStreamToken, MultiWriterStreamToken, int) @attr.s(auto_attribs=True, slots=True, frozen=True) -class HaveSentRoom(Generic[T]): - """Whether we have sent the room data down a sliding sync connection. +class HaveSent(Generic[T]): + """Whether we have sent some data down a sliding sync connection. We are generic over the type of token used, e.g. `RoomStreamToken` or `MultiWriterStreamToken`. Attributes: - status: Flag of if we have or haven't sent down the room + status: Flag of if we have or haven't sent down the data last_token: If the flag is `PREVIOUSLY` then this is non-null and contains the last stream token of the last updates we sent down - the room, i.e. we still need to send everything since then to the + the data, i.e. we still need to send everything since then to the client. """ - status: HaveSentRoomFlag + status: HaveSentFlag last_token: T | None @staticmethod - def live() -> "HaveSentRoom[T]": - return HaveSentRoom(HaveSentRoomFlag.LIVE, None) + def live() -> "HaveSent[T]": + return HaveSent(HaveSentFlag.LIVE, None) @staticmethod - def previously(last_token: T) -> "HaveSentRoom[T]": + def previously(last_token: T) -> "HaveSent[T]": """Constructor for `PREVIOUSLY` flag.""" - return HaveSentRoom(HaveSentRoomFlag.PREVIOUSLY, last_token) + return HaveSent(HaveSentFlag.PREVIOUSLY, last_token) @staticmethod - def never() -> "HaveSentRoom[T]": + def never() -> "HaveSent[T]": # We use a singleton to avoid repeatedly instantiating new `never` # values. - return _HAVE_SENT_ROOM_NEVER + return _HAVE_SENT_NEVER -_HAVE_SENT_ROOM_NEVER: HaveSentRoom[Any] = HaveSentRoom(HaveSentRoomFlag.NEVER, None) +_HAVE_SENT_NEVER: HaveSent[Any] = HaveSent(HaveSentFlag.NEVER, None) @attr.s(auto_attribs=True, slots=True, frozen=True) @@ -833,12 +833,12 @@ class RoomStatusMap(Generic[T]): """For a given stream, e.g. events, records what we have or have not sent down for that stream in a given room.""" - # `room_id` -> `HaveSentRoom` - _statuses: Mapping[str, HaveSentRoom[T]] = attr.Factory(dict) + # `room_id` -> `HaveSent` + _statuses: Mapping[str, HaveSent[T]] = attr.Factory(dict) - def have_sent_room(self, room_id: str) -> HaveSentRoom[T]: + def have_sent_room(self, room_id: str) -> HaveSent[T]: """Return whether we have previously sent the room down""" - return self._statuses.get(room_id, HaveSentRoom.never()) + return self._statuses.get(room_id, HaveSent.never()) def get_mutable(self) -> "MutableRoomStatusMap[T]": """Get a mutable copy of this state.""" @@ -862,11 +862,11 @@ class MutableRoomStatusMap(RoomStatusMap[T]): # We use a ChainMap here so that we can easily track what has been updated # and what hasn't. Note that when we persist the per connection state this # will get flattened to a normal dict (via calling `.copy()`) - _statuses: ChainMap[str, HaveSentRoom[T]] + _statuses: ChainMap[str, HaveSent[T]] def __init__( self, - statuses: Mapping[str, HaveSentRoom[T]], + statuses: Mapping[str, HaveSent[T]], ) -> None: # ChainMap requires a mutable mapping, but we're not actually going to # mutate it. @@ -876,18 +876,18 @@ class MutableRoomStatusMap(RoomStatusMap[T]): statuses=ChainMap({}, statuses), ) - def get_updates(self) -> Mapping[str, HaveSentRoom[T]]: + def get_updates(self) -> Mapping[str, HaveSent[T]]: """Return only the changes that were made""" return self._statuses.maps[0] def record_sent_rooms(self, room_ids: StrCollection) -> None: """Record that we have sent these rooms in the response""" for room_id in room_ids: - current_status = self._statuses.get(room_id, HaveSentRoom.never()) - if current_status.status == HaveSentRoomFlag.LIVE: + current_status = self._statuses.get(room_id, HaveSent.never()) + if current_status.status == HaveSentFlag.LIVE: continue - self._statuses[room_id] = HaveSentRoom.live() + self._statuses[room_id] = HaveSent.live() def record_unsent_rooms(self, room_ids: StrCollection, from_token: T) -> None: """Record that we have not sent these rooms in the response, but there @@ -904,11 +904,11 @@ class MutableRoomStatusMap(RoomStatusMap[T]): # sent anything down this time either so we leave it as NEVER. for room_id in room_ids: - current_status = self._statuses.get(room_id, HaveSentRoom.never()) - if current_status.status != HaveSentRoomFlag.LIVE: + current_status = self._statuses.get(room_id, HaveSent.never()) + if current_status.status != HaveSentFlag.LIVE: continue - self._statuses[room_id] = HaveSentRoom.previously(from_token) + self._statuses[room_id] = HaveSent.previously(from_token) @attr.s(auto_attribs=True, slots=True, frozen=True) @@ -916,12 +916,12 @@ class ProfileFieldStatusMap(Generic[T]): """For a given profile field, records what we have or have not sent down for that field in a given user profile.""" - # `user_id` -> `field_name` -> `HaveSentRoom` - _statuses: Mapping[str, Mapping[str, HaveSentRoom[T]]] = attr.Factory(dict) + # `user_id` -> `field_name` -> `HaveSent` + _statuses: Mapping[str, Mapping[str, HaveSent[T]]] = attr.Factory(dict) - def have_sent_field(self, user_id: str, field_name: str) -> HaveSentRoom[T]: + def have_sent_field(self, user_id: str, field_name: str) -> HaveSent[T]: """Return whether we have previously sent the field for this user""" - return self._statuses.get(user_id, {}).get(field_name, HaveSentRoom.never()) + return self._statuses.get(user_id, {}).get(field_name, HaveSent.never()) def get_mutable(self) -> "MutableProfileFieldStatusMap[T]": """Get a mutable copy of this state.""" @@ -944,11 +944,11 @@ class MutableProfileFieldStatusMap(ProfileFieldStatusMap[T]): # We use a ChainMap here so that we can easily track what has been updated # and what hasn't. Note that when we persist the per connection state this # will get flattened to a normal dict (via calling `.copy()`) - _statuses: ChainMap[str, Mapping[str, HaveSentRoom[T]]] + _statuses: ChainMap[str, Mapping[str, HaveSent[T]]] def __init__( self, - statuses: Mapping[str, Mapping[str, HaveSentRoom[T]]], + statuses: Mapping[str, Mapping[str, HaveSent[T]]], ) -> None: # ChainMap requires a mutable mapping, but we're not actually going to # mutate it. @@ -958,7 +958,7 @@ class MutableProfileFieldStatusMap(ProfileFieldStatusMap[T]): statuses=ChainMap({}, statuses_mutable), ) - def get_updates(self) -> Mapping[str, Mapping[str, HaveSentRoom[T]]]: + def get_updates(self) -> Mapping[str, Mapping[str, HaveSent[T]]]: """Return only the changes that were made""" return self._statuses.maps[0] @@ -967,16 +967,14 @@ class MutableProfileFieldStatusMap(ProfileFieldStatusMap[T]): if user_id not in self._statuses: self._statuses[user_id] = {} - user_fields = cast( - MutableMapping[str, HaveSentRoom[T]], self._statuses[user_id] - ) + user_fields = cast(MutableMapping[str, HaveSent[T]], self._statuses[user_id]) for field_name in field_names: - current_status = user_fields.get(field_name, HaveSentRoom.never()) - if current_status.status == HaveSentRoomFlag.LIVE: + current_status = user_fields.get(field_name, HaveSent.never()) + if current_status.status == HaveSentFlag.LIVE: continue - user_fields[field_name] = HaveSentRoom.live() + user_fields[field_name] = HaveSent.live() def record_unsent_fields( self, user_id: str, field_names: list[str], from_token: T @@ -987,16 +985,14 @@ class MutableProfileFieldStatusMap(ProfileFieldStatusMap[T]): if user_id not in self._statuses: return - user_fields = cast( - MutableMapping[str, HaveSentRoom[T]], self._statuses[user_id] - ) + user_fields = cast(MutableMapping[str, HaveSent[T]], self._statuses[user_id]) for field_name in field_names: - current_status = user_fields.get(field_name, HaveSentRoom.never()) - if current_status.status != HaveSentRoomFlag.LIVE: + current_status = user_fields.get(field_name, HaveSent.never()) + if current_status.status != HaveSentFlag.LIVE: continue - user_fields[field_name] = HaveSentRoom.previously(from_token) + user_fields[field_name] = HaveSent.previously(from_token) @attr.s(auto_attribs=True, frozen=True) diff --git a/tests/rest/client/sliding_sync/test_extension_account_data.py b/tests/rest/client/sliding_sync/test_extension_account_data.py index 8e08b4a2ac..8c677efa69 100644 --- a/tests/rest/client/sliding_sync/test_extension_account_data.py +++ b/tests/rest/client/sliding_sync/test_extension_account_data.py @@ -370,7 +370,7 @@ class SlidingSyncAccountDataExtensionTestCase(SlidingSyncBase): On incremental sync, we return all account data for a given room but only for rooms that we request and are being returned in the Sliding Sync response. - (HaveSentRoomFlag.LIVE) + (HaveSentFlag.LIVE) """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") @@ -538,7 +538,7 @@ class SlidingSyncAccountDataExtensionTestCase(SlidingSyncBase): """Tests that we don't return account data for rooms that are out of range, but then do send all account data once they're in range. - (initial/HaveSentRoomFlag.NEVER) + (initial/HaveSentFlag.NEVER) """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") @@ -728,7 +728,7 @@ class SlidingSyncAccountDataExtensionTestCase(SlidingSyncBase): """Tests that we don't return account data for rooms that fall out of range, but then do send all account data that has changed they're back in range. - (HaveSentRoomFlag.PREVIOUSLY) + (HaveSentFlag.PREVIOUSLY) """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass")