Presence logic for handling newly joined rooms is very inefficient and does duplicate work · Issue #8956 · matrix-org/synapse

When a user joins a room Synapse needs to decide what presence updates to send out. For local users joining a room we need to send a presence update to all users in the room, for remote users joining we need to send out all local users presence to the remote user.

This is implemented by streaming the current state delta stream, however when the server joins a federated room for the first time the state delta will include join events for all members. This leads to a situation where we correctly process the fact that a local user has joined a room and send out presence updates, and then incorrectly process each remote join event as if they're a new user join, leading to a lot of unnecessary work and superfluous presence events being generated and sent over federation.

synapse/synapse/handlers/presence.py

Lines 860 to 953 in 6d02eb2

    
               async def _handle_state_delta(self, deltas): 
        
                   """Process current state deltas to find new joins that need to be 
        
                   handled. 
        
                   """ 
        
                   for delta in deltas: 
        
                       typ = delta["type"] 
        
                       state_key = delta["state_key"] 
        
                       room_id = delta["room_id"] 
        
                       event_id = delta["event_id"] 
        
                       prev_event_id = delta["prev_event_id"] 
        
                       logger.debug("Handling: %r %r, %s", typ, state_key, event_id) 
        
                       if typ != EventTypes.Member: 
        
                           continue 
        
                       if event_id is None: 
        
                           # state has been deleted, so this is not a join. We only care about 
        
                           # joins. 
        
                           continue 
        
                       event = await self.store.get_event(event_id, allow_none=True) 
        
                       if not event or event.content.get("membership") != Membership.JOIN: 
        
                           # We only care about joins 
        
                           continue 
        
                       if prev_event_id: 
        
                           prev_event = await self.store.get_event(prev_event_id, allow_none=True) 
        
                           if ( 
        
                               prev_event 
        
                               and prev_event.content.get("membership") == Membership.JOIN 
        
                           ): 
        
                               # Ignore changes to join events. 
        
                               continue 
        
                       await self._on_user_joined_room(room_id, state_key) 
        
               async def _on_user_joined_room(self, room_id: str, user_id: str) -> None: 
        
                   """Called when we detect a user joining the room via the current state 
        
                   delta stream. 
        
                   """ 
        
                   if self.is_mine_id(user_id): 
        
                       # If this is a local user then we need to send their presence 
        
                       # out to hosts in the room (who don't already have it) 
        
                       # TODO: We should be able to filter the hosts down to those that 
        
                       # haven't previously seen the user 
        
                       state = await self.current_state_for_user(user_id) 
        
                       hosts = await self.state.get_current_hosts_in_room(room_id) 
        
                       # Filter out ourselves. 
        
                       hosts = {host for host in hosts if host != self.server_name} 
        
                       self.federation.send_presence_to_destinations( 
        
                           states=[state], destinations=hosts 
        
                       ) 
        
                   else: 
        
                       # A remote user has joined the room, so we need to: 
        
                       #   1. Check if this is a new server in the room 
        
                       #   2. If so send any presence they don't already have for 
        
                       #      local users in the room. 
        
                       # TODO: We should be able to filter the users down to those that 
        
                       # the server hasn't previously seen 
        
                       # TODO: Check that this is actually a new server joining the 
        
                       # room. 
        
                       users = await self.state.get_current_users_in_room(room_id) 
        
                       user_ids = list(filter(self.is_mine_id, users)) 
        
                       states_d = await self.current_state_for_users(user_ids) 
        
                       # Filter out old presence, i.e. offline presence states where 
        
                       # the user hasn't been active for a week. We can change this 
        
                       # depending on what we want the UX to be, but at the least we 
        
                       # should filter out offline presence where the state is just the 
        
                       # default state. 
        
                       now = self.clock.time_msec() 
        
                       states = [ 
        
                           state 
        
                           for state in states_d.values() 
        
                           if state.state != PresenceState.OFFLINE 
        
                           or now - state.last_active_ts < 7 * 24 * 60 * 60 * 1000 
        
                           or state.status_msg is not None 
        
                       ] 
        
                       if states: 
        
                           self.federation.send_presence_to_destinations( 
        
                               states=states, destinations=[get_domain_from_id(user_id)] 
        
                           )

On top of that, we should also only send presence updates when this is the first time a remote server shares a room with the local user.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

Presence logic for handling newly joined rooms is very inefficient and does duplicate work #8956

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	async def _handle_state_delta(self, deltas):
	"""Process current state deltas to find new joins that need to be
	handled.
	"""
	for delta in deltas:
	typ = delta["type"]
	state_key = delta["state_key"]
	room_id = delta["room_id"]
	event_id = delta["event_id"]
	prev_event_id = delta["prev_event_id"]

	logger.debug("Handling: %r %r, %s", typ, state_key, event_id)

	if typ != EventTypes.Member:
	continue

	if event_id is None:
	# state has been deleted, so this is not a join. We only care about
	# joins.
	continue

	event = await self.store.get_event(event_id, allow_none=True)
	if not event or event.content.get("membership") != Membership.JOIN:
	# We only care about joins
	continue

	if prev_event_id:
	prev_event = await self.store.get_event(prev_event_id, allow_none=True)
	if (
	prev_event
	and prev_event.content.get("membership") == Membership.JOIN
	):
	# Ignore changes to join events.
	continue

	await self._on_user_joined_room(room_id, state_key)

	async def _on_user_joined_room(self, room_id: str, user_id: str) -> None:
	"""Called when we detect a user joining the room via the current state
	delta stream.
	"""

	if self.is_mine_id(user_id):
	# If this is a local user then we need to send their presence
	# out to hosts in the room (who don't already have it)

	# TODO: We should be able to filter the hosts down to those that
	# haven't previously seen the user

	state = await self.current_state_for_user(user_id)
	hosts = await self.state.get_current_hosts_in_room(room_id)

	# Filter out ourselves.
	hosts = {host for host in hosts if host != self.server_name}

	self.federation.send_presence_to_destinations(
	states=[state], destinations=hosts
	)
	else:
	# A remote user has joined the room, so we need to:
	# 1. Check if this is a new server in the room
	# 2. If so send any presence they don't already have for
	# local users in the room.

	# TODO: We should be able to filter the users down to those that
	# the server hasn't previously seen

	# TODO: Check that this is actually a new server joining the
	# room.

	users = await self.state.get_current_users_in_room(room_id)
	user_ids = list(filter(self.is_mine_id, users))

	states_d = await self.current_state_for_users(user_ids)

	# Filter out old presence, i.e. offline presence states where
	# the user hasn't been active for a week. We can change this
	# depending on what we want the UX to be, but at the least we
	# should filter out offline presence where the state is just the
	# default state.
	now = self.clock.time_msec()
	states = [
	state
	for state in states_d.values()
	if state.state != PresenceState.OFFLINE
	or now - state.last_active_ts < 7 * 24 * 60 * 60 * 1000
	or state.status_msg is not None
	]

	if states:
	self.federation.send_presence_to_destinations(
	states=states, destinations=[get_domain_from_id(user_id)]
	)

Uh oh!

Presence logic for handling newly joined rooms is very inefficient and does duplicate work #8956

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions