diff --git a/changelog.d/17658.misc b/changelog.d/17658.misc
new file mode 100644
index 0000000000..0bdbc1140d
--- /dev/null
+++ b/changelog.d/17658.misc
@@ -0,0 +1 @@
+Get `bump_stamp` from [new sliding sync tables](https://github.com/element-hq/synapse/pull/17512) which should be faster.
diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py
index 444cc32f36..7340c6ec05 100644
--- a/synapse/handlers/sliding_sync/__init__.py
+++ b/synapse/handlers/sliding_sync/__init__.py
@@ -1040,29 +1040,67 @@ class SlidingSyncHandler:
)
)
- # By default, just choose the membership event position
+ # Figure out the last bump event in the room
+ #
+ # By default, just choose the membership event position for any non-join membership
bump_stamp = room_membership_for_user_at_to_token.event_pos.stream
-
- # Figure out the last bump event in the room if we're in the room.
+ # If we're joined to the room, we need to find the last bump event before the
+ # `to_token`
if room_membership_for_user_at_to_token.membership == Membership.JOIN:
- last_bump_event_result = (
- await self.store.get_last_event_pos_in_room_before_stream_ordering(
- room_id,
- to_token.room_key,
- event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES,
- )
+ # We can quickly query for the latest bump event in the room using the
+ # sliding sync tables.
+ latest_room_bump_stamp = await self.store.get_latest_bump_stamp_for_room(
+ room_id
)
- # But if we found a bump event, use that instead
- if last_bump_event_result is not None:
- _, new_bump_event_pos = last_bump_event_result
+ min_to_token_position = to_token.room_key.stream
- # If we've just joined a remote room, then the last bump event may
- # have been backfilled (and so have a negative stream ordering).
- # These negative stream orderings can't sensibly be compared, so
- # instead we use the membership event position.
- if new_bump_event_pos.stream > 0:
- bump_stamp = new_bump_event_pos.stream
+ # If we can rely on the new sliding sync tables and the `bump_stamp` is
+ # `None`, just fallback to the membership event position. This can happen
+ # when we've just joined a remote room and all the events are backfilled.
+ if (
+ # FIXME: The background job check can be removed once we bump
+ # `SCHEMA_COMPAT_VERSION` and run the foreground update for
+ # `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots`
+ # (tracked by https://github.com/element-hq/synapse/issues/17623)
+ await self.store.have_finished_sliding_sync_background_jobs()
+ and latest_room_bump_stamp is None
+ ):
+ pass
+
+ # The `bump_stamp` stored in the database might be ahead of our token. Since
+ # `bump_stamp` is only a `stream_ordering` position, we can't be 100% sure
+ # that's before the `to_token` in all scenarios. The only scenario we can be
+ # sure of is if the `bump_stamp` is totally before the minimum position from
+ # the token.
+ #
+ # We don't need to check if the background update has finished, as if the
+ # returned bump stamp is not None then it must be up to date.
+ elif (
+ latest_room_bump_stamp is not None
+ and latest_room_bump_stamp < min_to_token_position
+ ):
+ bump_stamp = latest_room_bump_stamp
+
+ # Otherwise, if it's within or after the `to_token`, we need to find the
+ # last bump event before the `to_token`.
+ else:
+ last_bump_event_result = (
+ await self.store.get_last_event_pos_in_room_before_stream_ordering(
+ room_id,
+ to_token.room_key,
+ event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES,
+ )
+ )
+ if last_bump_event_result is not None:
+ _, new_bump_event_pos = last_bump_event_result
+
+ # If we've just joined a remote room, then the last bump event may
+ # have been backfilled (and so have a negative stream ordering).
+ # These negative stream orderings can't sensibly be compared, so
+ # instead we use the membership event position.
+ if new_bump_event_pos.stream > 0:
+ bump_stamp = new_bump_event_pos.stream
unstable_expanded_timeline = False
prev_room_sync_config = previous_connection_state.room_configs.get(room_id)
diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py
index d423d80efa..e5f63019fd 100644
--- a/synapse/storage/databases/main/events.py
+++ b/synapse/storage/databases/main/events.py
@@ -327,6 +327,13 @@ class PersistEventsStore:
async with stream_ordering_manager as stream_orderings:
for (event, _), stream in zip(events_and_contexts, stream_orderings):
+ # XXX: We can't rely on `stream_ordering`/`instance_name` being correct
+ # at this point. We could be working with events that were previously
+ # persisted as an `outlier` with one `stream_ordering` but are now being
+ # persisted again and de-outliered and are being assigned a different
+ # `stream_ordering` here that won't end up being used.
+ # `_update_outliers_txn()` will fix this discrepancy (always use the
+ # `stream_ordering` from the first time it was persisted).
event.internal_metadata.stream_ordering = stream
event.internal_metadata.instance_name = self._instance_name
@@ -470,11 +477,11 @@ class PersistEventsStore:
membership_infos_to_insert_membership_snapshots.append(
# XXX: We don't use `SlidingSyncMembershipInfoWithEventPos` here
# because we're sourcing the event from `events_and_contexts`, we
- # can't rely on `stream_ordering`/`instance_name` being correct. We
- # could be working with events that were previously persisted as an
- # `outlier` with one `stream_ordering` but are now being persisted
- # again and de-outliered and assigned a different `stream_ordering`
- # that won't end up being used. Since we call
+ # can't rely on `stream_ordering`/`instance_name` being correct at
+ # this point. We could be working with events that were previously
+ # persisted as an `outlier` with one `stream_ordering` but are now
+ # being persisted again and de-outliered and assigned a different
+ # `stream_ordering` that won't end up being used. Since we call
# `_calculate_sliding_sync_table_changes()` before
# `_update_outliers_txn()` which fixes this discrepancy (always use
# the `stream_ordering` from the first time it was persisted), we're
@@ -591,11 +598,17 @@ class PersistEventsStore:
event_types=SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES,
)
)
- bump_stamp_to_fully_insert = (
- most_recent_bump_event_pos_results[1].stream
- if most_recent_bump_event_pos_results is not None
- else None
- )
+ if most_recent_bump_event_pos_results is not None:
+ _, new_bump_event_pos = most_recent_bump_event_pos_results
+
+ # If we've just joined a remote room, then the last bump event may
+ # have been backfilled (and so have a negative stream ordering).
+ # These negative stream orderings can't sensibly be compared, so
+ # instead just leave it as `None` in the table and we will use their
+ # membership event position as the bump event position in the
+ # Sliding Sync API.
+ if new_bump_event_pos.stream > 0:
+ bump_stamp_to_fully_insert = new_bump_event_pos.stream
current_state_ids_map = dict(
await self.store.get_partial_filtered_current_state_ids(
@@ -2123,31 +2136,26 @@ class PersistEventsStore:
if len(events_and_contexts) == 0:
return
- # We only update the sliding sync tables for non-backfilled events.
- #
- # Check if the first event is a backfilled event (with a negative
- # `stream_ordering`). If one event is backfilled, we assume this whole batch was
- # backfilled.
- first_event_stream_ordering = events_and_contexts[0][
- 0
- ].internal_metadata.stream_ordering
- # This should exist for persisted events
- assert first_event_stream_ordering is not None
- if first_event_stream_ordering < 0:
- return
-
# Since the list is sorted ascending by `stream_ordering`, the last event should
# have the highest `stream_ordering`.
max_stream_ordering = events_and_contexts[-1][
0
].internal_metadata.stream_ordering
+ # `stream_ordering` should be assigned for persisted events
+ assert max_stream_ordering is not None
+ # Check if the event is a backfilled event (with a negative `stream_ordering`).
+ # If one event is backfilled, we assume this whole batch was backfilled.
+ if max_stream_ordering < 0:
+ # We only update the sliding sync tables for non-backfilled events.
+ return
+
max_bump_stamp = None
for event, _ in reversed(events_and_contexts):
# Sanity check that all events belong to the same room
assert event.room_id == room_id
if event.type in SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES:
- # This should exist for persisted events
+ # `stream_ordering` should be assigned for persisted events
assert event.internal_metadata.stream_ordering is not None
max_bump_stamp = event.internal_metadata.stream_ordering
@@ -2156,11 +2164,6 @@ class PersistEventsStore:
# matching bump event which should have the highest `stream_ordering`.
break
- # We should have exited earlier if there were no events
- assert (
- max_stream_ordering is not None
- ), "Expected to have a stream_ordering if we have events"
-
# Handle updating the `sliding_sync_joined_rooms` table.
#
txn.execute(
diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py
index dc747d7ac0..83939d10b0 100644
--- a/synapse/storage/databases/main/sliding_sync.py
+++ b/synapse/storage/databases/main/sliding_sync.py
@@ -41,6 +41,46 @@ logger = logging.getLogger(__name__)
class SlidingSyncStore(SQLBaseStore):
+ async def get_latest_bump_stamp_for_room(
+ self,
+ room_id: str,
+ ) -> Optional[int]:
+ """
+ Get the `bump_stamp` for the room.
+
+ The `bump_stamp` is the `stream_ordering` of the last event according to the
+ `bump_event_types`. This helps clients sort more readily without them needing to
+ pull in a bunch of the timeline to determine the last activity.
+ `bump_event_types` is a thing because for example, we don't want display name
+ changes to mark the room as unread and bump it to the top. For encrypted rooms,
+ we just have to consider any activity as a bump because we can't see the content
+ and the client has to figure it out for themselves.
+
+ This should only be called where the server is participating
+ in the room (someone local is joined).
+
+ Returns:
+ The `bump_stamp` for the room (which can be `None`).
+ """
+
+ return cast(
+ Optional[int],
+ await self.db_pool.simple_select_one_onecol(
+ table="sliding_sync_joined_rooms",
+ keyvalues={"room_id": room_id},
+ retcol="bump_stamp",
+ # FIXME: This should be `False` once we bump `SCHEMA_COMPAT_VERSION` and run the
+ # foreground update for
+ # `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked
+ # by https://github.com/element-hq/synapse/issues/17623)
+ #
+ # The should be `allow_none=False` in the future because event though
+ # `bump_stamp` itself can be `None`, we should have a row in the
+ # `sliding_sync_joined_rooms` table for any joined room.
+ allow_none=True,
+ ),
+ )
+
async def persist_per_connection_state(
self,
user_id: str,
diff --git a/tests/storage/test_sliding_sync_tables.py b/tests/storage/test_sliding_sync_tables.py
index 621f46fff8..de80ad53cd 100644
--- a/tests/storage/test_sliding_sync_tables.py
+++ b/tests/storage/test_sliding_sync_tables.py
@@ -106,6 +106,12 @@ class SlidingSyncTablesTestCaseBase(HomeserverTestCase):
assert persist_events_store is not None
self.persist_events_store = persist_events_store
+ persist_controller = self.hs.get_storage_controllers().persistence
+ assert persist_controller is not None
+ self.persist_controller = persist_controller
+
+ self.state_handler = self.hs.get_state_handler()
+
def _get_sliding_sync_joined_rooms(self) -> Dict[str, _SlidingSyncJoinedRoomResult]:
"""
Return the rows from the `sliding_sync_joined_rooms` table.
@@ -260,10 +266,8 @@ class SlidingSyncTablesTestCaseBase(HomeserverTestCase):
)
)
context = EventContext.for_outlier(self.hs.get_storage_controllers())
- persist_controller = self.hs.get_storage_controllers().persistence
- assert persist_controller is not None
persisted_event, _, _ = self.get_success(
- persist_controller.persist_event(invite_event, context)
+ self.persist_controller.persist_event(invite_event, context)
)
self._remote_invite_count += 1
@@ -316,10 +320,8 @@ class SlidingSyncTablesTestCaseBase(HomeserverTestCase):
)
)
context = EventContext.for_outlier(self.hs.get_storage_controllers())
- persist_controller = self.hs.get_storage_controllers().persistence
- assert persist_controller is not None
persisted_event, _, _ = self.get_success(
- persist_controller.persist_event(kick_event, context)
+ self.persist_controller.persist_event(kick_event, context)
)
return persisted_event
@@ -926,6 +928,201 @@ class SlidingSyncTablesTestCase(SlidingSyncTablesTestCaseBase):
user2_snapshot,
)
+ def test_joined_room_bump_stamp_backfill(self) -> None:
+ """
+ Test that `bump_stamp` ignores backfilled events, i.e. events with a
+ negative stream ordering.
+ """
+ user1_id = self.register_user("user1", "pass")
+ _user1_tok = self.login(user1_id, "pass")
+
+ # Create a remote room
+ creator = "@user:other"
+ room_id = "!foo:other"
+ room_version = RoomVersions.V10
+ shared_kwargs = {
+ "room_id": room_id,
+ "room_version": room_version.identifier,
+ }
+
+ create_tuple = self.get_success(
+ create_event(
+ self.hs,
+ prev_event_ids=[],
+ type=EventTypes.Create,
+ state_key="",
+ content={
+ # The `ROOM_CREATOR` field could be removed if we used a room
+ # version > 10 (in favor of relying on `sender`)
+ EventContentFields.ROOM_CREATOR: creator,
+ EventContentFields.ROOM_VERSION: room_version.identifier,
+ },
+ sender=creator,
+ **shared_kwargs,
+ )
+ )
+ creator_tuple = self.get_success(
+ create_event(
+ self.hs,
+ prev_event_ids=[create_tuple[0].event_id],
+ auth_event_ids=[create_tuple[0].event_id],
+ type=EventTypes.Member,
+ state_key=creator,
+ content={"membership": Membership.JOIN},
+ sender=creator,
+ **shared_kwargs,
+ )
+ )
+ room_name_tuple = self.get_success(
+ create_event(
+ self.hs,
+ prev_event_ids=[creator_tuple[0].event_id],
+ auth_event_ids=[create_tuple[0].event_id, creator_tuple[0].event_id],
+ type=EventTypes.Name,
+ state_key="",
+ content={
+ EventContentFields.ROOM_NAME: "my super duper room",
+ },
+ sender=creator,
+ **shared_kwargs,
+ )
+ )
+ # We add a message event as a valid "bump type"
+ msg_tuple = self.get_success(
+ create_event(
+ self.hs,
+ prev_event_ids=[room_name_tuple[0].event_id],
+ auth_event_ids=[create_tuple[0].event_id, creator_tuple[0].event_id],
+ type=EventTypes.Message,
+ content={"body": "foo", "msgtype": "m.text"},
+ sender=creator,
+ **shared_kwargs,
+ )
+ )
+ invite_tuple = self.get_success(
+ create_event(
+ self.hs,
+ prev_event_ids=[msg_tuple[0].event_id],
+ auth_event_ids=[create_tuple[0].event_id, creator_tuple[0].event_id],
+ type=EventTypes.Member,
+ state_key=user1_id,
+ content={"membership": Membership.INVITE},
+ sender=creator,
+ **shared_kwargs,
+ )
+ )
+
+ remote_events_and_contexts = [
+ create_tuple,
+ creator_tuple,
+ room_name_tuple,
+ msg_tuple,
+ invite_tuple,
+ ]
+
+ # Ensure the local HS knows the room version
+ self.get_success(self.store.store_room(room_id, creator, False, room_version))
+
+ # Persist these events as backfilled events.
+ for event, context in remote_events_and_contexts:
+ self.get_success(
+ self.persist_controller.persist_event(event, context, backfilled=True)
+ )
+
+ # Now we join the local user to the room. We want to make this feel as close to
+ # the real `process_remote_join()` as possible but we'd like to avoid some of
+ # the auth checks that would be done in the real code.
+ #
+ # FIXME: The test was originally written using this less-real
+ # `persist_event(...)` shortcut but it would be nice to use the real remote join
+ # process in a `FederatingHomeserverTestCase`.
+ flawed_join_tuple = self.get_success(
+ create_event(
+ self.hs,
+ prev_event_ids=[invite_tuple[0].event_id],
+ # This doesn't work correctly to create an `EventContext` that includes
+ # both of these state events. I assume it's because we're working on our
+ # local homeserver which has the remote state set as `outlier`. We have
+ # to create our own EventContext below to get this right.
+ auth_event_ids=[create_tuple[0].event_id, invite_tuple[0].event_id],
+ type=EventTypes.Member,
+ state_key=user1_id,
+ content={"membership": Membership.JOIN},
+ sender=user1_id,
+ **shared_kwargs,
+ )
+ )
+ # We have to create our own context to get the state set correctly. If we use
+ # the `EventContext` from the `flawed_join_tuple`, the `current_state_events`
+ # table will only have the join event in it which should never happen in our
+ # real server.
+ join_event = flawed_join_tuple[0]
+ join_context = self.get_success(
+ self.state_handler.compute_event_context(
+ join_event,
+ state_ids_before_event={
+ (e.type, e.state_key): e.event_id
+ for e in [create_tuple[0], invite_tuple[0], room_name_tuple[0]]
+ },
+ partial_state=False,
+ )
+ )
+ join_event, _join_event_pos, _room_token = self.get_success(
+ self.persist_controller.persist_event(join_event, join_context)
+ )
+
+ # Make sure the tables are populated correctly
+ sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms()
+ self.assertIncludes(
+ set(sliding_sync_joined_rooms_results.keys()),
+ {room_id},
+ exact=True,
+ )
+ self.assertEqual(
+ sliding_sync_joined_rooms_results[room_id],
+ _SlidingSyncJoinedRoomResult(
+ room_id=room_id,
+ # This should be the last event in the room (the join membership)
+ event_stream_ordering=join_event.internal_metadata.stream_ordering,
+ # Since all of the bump events are backfilled, the `bump_stamp` should
+ # still be `None`. (and we will fallback to the users membership event
+ # position in the Sliding Sync API)
+ bump_stamp=None,
+ room_type=None,
+ # We still pick up state of the room even if it's backfilled
+ room_name="my super duper room",
+ is_encrypted=False,
+ tombstone_successor_room_id=None,
+ ),
+ )
+
+ sliding_sync_membership_snapshots_results = (
+ self._get_sliding_sync_membership_snapshots()
+ )
+ self.assertIncludes(
+ set(sliding_sync_membership_snapshots_results.keys()),
+ {
+ (room_id, user1_id),
+ },
+ exact=True,
+ )
+ self.assertEqual(
+ sliding_sync_membership_snapshots_results.get((room_id, user1_id)),
+ _SlidingSyncMembershipSnapshotResult(
+ room_id=room_id,
+ user_id=user1_id,
+ sender=user1_id,
+ membership_event_id=join_event.event_id,
+ membership=Membership.JOIN,
+ event_stream_ordering=join_event.internal_metadata.stream_ordering,
+ has_known_state=True,
+ room_type=None,
+ room_name="my super duper room",
+ is_encrypted=False,
+ tombstone_successor_room_id=None,
+ ),
+ )
+
@parameterized.expand(
# Test both an insert an upsert into the
# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` to exercise
@@ -1036,11 +1233,9 @@ class SlidingSyncTablesTestCase(SlidingSyncTablesTestCaseBase):
context = self.get_success(unpersisted_context.persist(event))
events_to_persist.append((event, context))
- persist_controller = self.hs.get_storage_controllers().persistence
- assert persist_controller is not None
for event, context in events_to_persist:
self.get_success(
- persist_controller.persist_event(
+ self.persist_controller.persist_event(
event,
context,
)
|
