When server leaves room check for stale device lists. (#6801)

When a server leaves a room it may stop sharing a room with remote
users, and thus not get any updates to their device lists. So we need to
check for this case and delete those device lists from the cache.

We don't need to do this if we stop sharing a room because the remote
user leaves the room, because we track that case via looking at
membership changes.
This commit is contained in:
Erik Johnston 2020-01-30 16:10:30 +00:00 committed by GitHub
parent c80a9fe13d
commit a5bab2d058
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 83 additions and 6 deletions

1
changelog.d/6801.bugfix Normal file
View File

@ -0,0 +1 @@
Fix bug where Synapse didn't invalidate cache of remote users' devices when Synapse left a room.

View File

@ -15,7 +15,7 @@
# limitations under the License. # limitations under the License.
import logging import logging
from typing import Iterable, List from typing import Iterable, List, Set
from six import iteritems, itervalues from six import iteritems, itervalues
@ -40,7 +40,7 @@ from synapse.storage.roommember import (
ProfileInfo, ProfileInfo,
RoomsForUser, RoomsForUser,
) )
from synapse.types import get_domain_from_id from synapse.types import Collection, get_domain_from_id
from synapse.util.async_helpers import Linearizer from synapse.util.async_helpers import Linearizer
from synapse.util.caches import intern_string from synapse.util.caches import intern_string
from synapse.util.caches.descriptors import cached, cachedInlineCallbacks, cachedList from synapse.util.caches.descriptors import cached, cachedInlineCallbacks, cachedList
@ -439,6 +439,39 @@ class RoomMemberWorkerStore(EventsWorkerStore):
return results return results
async def get_users_server_still_shares_room_with(
self, user_ids: Collection[str]
) -> Set[str]:
"""Given a list of users return the set that the server still share a
room with.
"""
if not user_ids:
return set()
def _get_users_server_still_shares_room_with_txn(txn):
sql = """
SELECT state_key FROM current_state_events
WHERE
type = 'm.room.member'
AND membership = 'join'
AND %s
GROUP BY state_key
"""
clause, args = make_in_list_sql_clause(
self.database_engine, "state_key", user_ids
)
txn.execute(sql % (clause,), args)
return set(row[0] for row in txn)
return await self.db.runInteraction(
"get_users_server_still_shares_room_with",
_get_users_server_still_shares_room_with_txn,
)
@defer.inlineCallbacks @defer.inlineCallbacks
def get_rooms_for_user(self, user_id, on_invalidate=None): def get_rooms_for_user(self, user_id, on_invalidate=None):
"""Returns a set of room_ids the user is currently joined to. """Returns a set of room_ids the user is currently joined to.

View File

@ -18,7 +18,7 @@
import itertools import itertools
import logging import logging
from collections import deque, namedtuple from collections import deque, namedtuple
from typing import Iterable, List, Optional, Tuple from typing import Iterable, List, Optional, Set, Tuple
from six import iteritems from six import iteritems
from six.moves import range from six.moves import range
@ -318,6 +318,11 @@ class EventsPersistenceStorage(object):
# room # room
state_delta_for_room = {} state_delta_for_room = {}
# Set of remote users which were in rooms the server has left. We
# should check if we still share any rooms and if not we mark their
# device lists as stale.
potentially_left_users = set() # type: Set[str]
if not backfilled: if not backfilled:
with Measure(self._clock, "_calculate_state_and_extrem"): with Measure(self._clock, "_calculate_state_and_extrem"):
# Work out the new "current state" for each room. # Work out the new "current state" for each room.
@ -421,7 +426,11 @@ class EventsPersistenceStorage(object):
# the room then we delete the current state and # the room then we delete the current state and
# extremities. # extremities.
is_still_joined = await self._is_server_still_joined( is_still_joined = await self._is_server_still_joined(
room_id, ev_ctx_rm, delta, current_state room_id,
ev_ctx_rm,
delta,
current_state,
potentially_left_users,
) )
if not is_still_joined: if not is_still_joined:
logger.info("Server no longer in room %s", room_id) logger.info("Server no longer in room %s", room_id)
@ -444,6 +453,8 @@ class EventsPersistenceStorage(object):
backfilled=backfilled, backfilled=backfilled,
) )
await self._handle_potentially_left_users(potentially_left_users)
async def _calculate_new_extremities( async def _calculate_new_extremities(
self, self,
room_id: str, room_id: str,
@ -688,6 +699,7 @@ class EventsPersistenceStorage(object):
ev_ctx_rm: List[Tuple[FrozenEvent, EventContext]], ev_ctx_rm: List[Tuple[FrozenEvent, EventContext]],
delta: DeltaState, delta: DeltaState,
current_state: Optional[StateMap[str]], current_state: Optional[StateMap[str]],
potentially_left_users: Set[str],
) -> bool: ) -> bool:
"""Check if the server will still be joined after the given events have """Check if the server will still be joined after the given events have
been persised. been persised.
@ -699,6 +711,9 @@ class EventsPersistenceStorage(object):
and what the new current state will be. and what the new current state will be.
current_state: The new current state if it already been calculated, current_state: The new current state if it already been calculated,
otherwise None. otherwise None.
potentially_left_users: If the server has left the room, then joined
remote users will be added to this set to indicate that the
server may no longer be sharing a room with them.
""" """
if not any( if not any(
@ -741,5 +756,33 @@ class EventsPersistenceStorage(object):
is_still_joined = any(row["membership"] == Membership.JOIN for row in rows) is_still_joined = any(row["membership"] == Membership.JOIN for row in rows)
if is_still_joined: if is_still_joined:
return True return True
else:
# The server will leave the room, so we go and find out which remote
# users will still be joined when we leave.
remote_event_ids = [
event_id
for (typ, state_key,), event_id in current_state.items()
if typ == EventTypes.Member and not self.is_mine_id(state_key)
]
rows = await self.main_store.get_membership_from_event_ids(remote_event_ids)
potentially_left_users.update(
row["user_id"] for row in rows if row["membership"] == Membership.JOIN
)
return False return False
async def _handle_potentially_left_users(self, user_ids: Set[str]):
"""Given a set of remote users check if the server still shares a room with
them. If not then mark those users' device cache as stale.
"""
if not user_ids:
return
joined_users = await self.main_store.get_users_server_still_shares_room_with(
user_ids
)
left_users = user_ids - joined_users
for user_id in left_users:
await self.main_store.mark_remote_user_device_list_as_unsubscribed(user_id)