diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 113d711d4c..c802db9792 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -20,10 +20,14 @@ import logging
from six import iteritems, itervalues
+from prometheus_client import Counter
+
from twisted.internet import defer
from synapse.api.constants import EventTypes, Membership
from synapse.push.clientformat import format_push_rules_for_user
+from synapse.storage.roommember import MemberSummary
+from synapse.storage.state import StateFilter
from synapse.types import RoomStreamToken
from synapse.util.async_helpers import concurrently_execute
from synapse.util.caches.expiringcache import ExpiringCache
@@ -35,6 +39,19 @@ from synapse.visibility import filter_events_for_client
logger = logging.getLogger(__name__)
+
+# Counts the number of times we returned a non-empty sync. `type` is one of
+# "initial_sync", "full_state_sync" or "incremental_sync", `lazy_loaded` is
+# "true" or "false" depending on if the request asked for lazy loaded members or
+# not.
+non_empty_sync_counter = Counter(
+ "synapse_handlers_sync_nonempty_total",
+ "Count of non empty sync responses. type is initial_sync/full_state_sync"
+ "/incremental_sync. lazy_loaded indicates if lazy loaded members were "
+ "enabled for that request.",
+ ["type", "lazy_loaded"],
+)
+
# Store the cache that tracks which lazy-loaded members have been sent to a given
# client for no more than 30 minutes.
LAZY_LOADED_MEMBERS_CACHE_MAX_AGE = 30 * 60 * 1000
@@ -226,14 +243,16 @@ class SyncHandler(object):
@defer.inlineCallbacks
def _wait_for_sync_for_user(self, sync_config, since_token, timeout,
full_state):
+ if since_token is None:
+ sync_type = "initial_sync"
+ elif full_state:
+ sync_type = "full_state_sync"
+ else:
+ sync_type = "incremental_sync"
+
context = LoggingContext.current_context()
if context:
- if since_token is None:
- context.tag = "initial_sync"
- elif full_state:
- context.tag = "full_state_sync"
- else:
- context.tag = "incremental_sync"
+ context.tag = sync_type
if timeout == 0 or since_token is None or full_state:
# we are going to return immediately, so don't bother calling
@@ -241,7 +260,6 @@ class SyncHandler(object):
result = yield self.current_sync_for_user(
sync_config, since_token, full_state=full_state,
)
- defer.returnValue(result)
else:
def current_sync_callback(before_token, after_token):
return self.current_sync_for_user(sync_config, since_token)
@@ -250,7 +268,15 @@ class SyncHandler(object):
sync_config.user.to_string(), timeout, current_sync_callback,
from_token=since_token,
)
- defer.returnValue(result)
+
+ if result:
+ if sync_config.filter_collection.lazy_load_members():
+ lazy_loaded = "true"
+ else:
+ lazy_loaded = "false"
+ non_empty_sync_counter.labels(sync_type, lazy_loaded).inc()
+
+ defer.returnValue(result)
def current_sync_for_user(self, sync_config, since_token=None,
full_state=False):
@@ -444,25 +470,20 @@ class SyncHandler(object):
))
@defer.inlineCallbacks
- def get_state_after_event(self, event, types=None, filtered_types=None):
+ def get_state_after_event(self, event, state_filter=StateFilter.all()):
"""
Get the room state after the given event
Args:
event(synapse.events.EventBase): event of interest
- types(list[(str, str|None)]|None): List of (type, state_key) tuples
- which are used to filter the state fetched. If `state_key` is None,
- all events are returned of the given type.
- May be None, which matches any key.
- filtered_types(list[str]|None): Only apply filtering via `types` to this
- list of event types. Other types of events are returned unfiltered.
- If None, `types` filtering is applied to all events.
+ state_filter (StateFilter): The state filter used to fetch state
+ from the database.
Returns:
A Deferred map from ((type, state_key)->Event)
"""
state_ids = yield self.store.get_state_ids_for_event(
- event.event_id, types, filtered_types=filtered_types,
+ event.event_id, state_filter=state_filter,
)
if event.is_state():
state_ids = state_ids.copy()
@@ -470,18 +491,14 @@ class SyncHandler(object):
defer.returnValue(state_ids)
@defer.inlineCallbacks
- def get_state_at(self, room_id, stream_position, types=None, filtered_types=None):
+ def get_state_at(self, room_id, stream_position, state_filter=StateFilter.all()):
""" Get the room state at a particular stream position
Args:
room_id(str): room for which to get state
stream_position(StreamToken): point at which to get state
- types(list[(str, str|None)]|None): List of (type, state_key) tuples
- which are used to filter the state fetched. If `state_key` is None,
- all events are returned of the given type.
- filtered_types(list[str]|None): Only apply filtering via `types` to this
- list of event types. Other types of events are returned unfiltered.
- If None, `types` filtering is applied to all events.
+ state_filter (StateFilter): The state filter used to fetch state
+ from the database.
Returns:
A Deferred map from ((type, state_key)->Event)
@@ -497,7 +514,7 @@ class SyncHandler(object):
if last_events:
last_event = last_events[-1]
state = yield self.get_state_after_event(
- last_event, types, filtered_types=filtered_types,
+ last_event, state_filter=state_filter,
)
else:
@@ -525,6 +542,8 @@ class SyncHandler(object):
A deferred dict describing the room summary
"""
+ # FIXME: we could/should get this from room_stats when matthew/stats lands
+
# FIXME: this promulgates https://github.com/matrix-org/synapse/issues/3305
last_events, _ = yield self.store.get_recent_event_ids_for_room(
room_id, end_token=now_token.room_key, limit=1,
@@ -536,45 +555,69 @@ class SyncHandler(object):
last_event = last_events[-1]
state_ids = yield self.store.get_state_ids_for_event(
- last_event.event_id, [
- (EventTypes.Member, None),
+ last_event.event_id,
+ state_filter=StateFilter.from_types([
(EventTypes.Name, ''),
(EventTypes.CanonicalAlias, ''),
- ]
+ ]),
)
- member_ids = {
- state_key: event_id
- for (t, state_key), event_id in state_ids.iteritems()
- if t == EventTypes.Member
- }
+ # this is heavily cached, thus: fast.
+ details = yield self.store.get_room_summary(room_id)
+
name_id = state_ids.get((EventTypes.Name, ''))
canonical_alias_id = state_ids.get((EventTypes.CanonicalAlias, ''))
summary = {}
-
- # FIXME: it feels very heavy to load up every single membership event
- # just to calculate the counts.
- member_events = yield self.store.get_events(member_ids.values())
-
- joined_user_ids = []
- invited_user_ids = []
-
- for ev in member_events.values():
- if ev.content.get("membership") == Membership.JOIN:
- joined_user_ids.append(ev.state_key)
- elif ev.content.get("membership") == Membership.INVITE:
- invited_user_ids.append(ev.state_key)
+ empty_ms = MemberSummary([], 0)
# TODO: only send these when they change.
- summary["m.joined_member_count"] = len(joined_user_ids)
- summary["m.invited_member_count"] = len(invited_user_ids)
+ summary["m.joined_member_count"] = (
+ details.get(Membership.JOIN, empty_ms).count
+ )
+ summary["m.invited_member_count"] = (
+ details.get(Membership.INVITE, empty_ms).count
+ )
- if name_id or canonical_alias_id:
- defer.returnValue(summary)
+ # if the room has a name or canonical_alias set, we can skip
+ # calculating heroes. we assume that if the event has contents, it'll
+ # be a valid name or canonical_alias - i.e. we're checking that they
+ # haven't been "deleted" by blatting {} over the top.
+ if name_id:
+ name = yield self.store.get_event(name_id, allow_none=True)
+ if name and name.content:
+ defer.returnValue(summary)
+
+ if canonical_alias_id:
+ canonical_alias = yield self.store.get_event(
+ canonical_alias_id, allow_none=True,
+ )
+ if canonical_alias and canonical_alias.content:
+ defer.returnValue(summary)
- # FIXME: order by stream ordering, not alphabetic
+ joined_user_ids = [
+ r[0] for r in details.get(Membership.JOIN, empty_ms).members
+ ]
+ invited_user_ids = [
+ r[0] for r in details.get(Membership.INVITE, empty_ms).members
+ ]
+ gone_user_ids = (
+ [r[0] for r in details.get(Membership.LEAVE, empty_ms).members] +
+ [r[0] for r in details.get(Membership.BAN, empty_ms).members]
+ )
+ # FIXME: only build up a member_ids list for our heroes
+ member_ids = {}
+ for membership in (
+ Membership.JOIN,
+ Membership.INVITE,
+ Membership.LEAVE,
+ Membership.BAN
+ ):
+ for user_id, event_id in details.get(membership, empty_ms).members:
+ member_ids[user_id] = event_id
+
+ # FIXME: order by stream ordering rather than as returned by SQL
me = sync_config.user.to_string()
if (joined_user_ids or invited_user_ids):
summary['m.heroes'] = sorted(
@@ -586,7 +629,11 @@ class SyncHandler(object):
)[0:5]
else:
summary['m.heroes'] = sorted(
- [user_id for user_id in member_ids.keys() if user_id != me]
+ [
+ user_id
+ for user_id in gone_user_ids
+ if user_id != me
+ ]
)[0:5]
if not sync_config.filter_collection.lazy_load_members():
@@ -663,8 +710,7 @@ class SyncHandler(object):
with Measure(self.clock, "compute_state_delta"):
- types = None
- filtered_types = None
+ members_to_fetch = None
lazy_load_members = sync_config.filter_collection.lazy_load_members()
include_redundant_members = (
@@ -675,16 +721,21 @@ class SyncHandler(object):
# We only request state for the members needed to display the
# timeline:
- types = [
- (EventTypes.Member, state_key)
- for state_key in set(
- event.sender # FIXME: we also care about invite targets etc.
- for event in batch.events
- )
- ]
+ members_to_fetch = set(
+ event.sender # FIXME: we also care about invite targets etc.
+ for event in batch.events
+ )
- # only apply the filtering to room members
- filtered_types = [EventTypes.Member]
+ if full_state:
+ # always make sure we LL ourselves so we know we're in the room
+ # (if we are) to fix https://github.com/vector-im/riot-web/issues/7209
+ # We only need apply this on full state syncs given we disabled
+ # LL for incr syncs in #3840.
+ members_to_fetch.add(sync_config.user.to_string())
+
+ state_filter = StateFilter.from_lazy_load_member_list(members_to_fetch)
+ else:
+ state_filter = StateFilter.all()
timeline_state = {
(event.type, event.state_key): event.event_id
@@ -694,19 +745,17 @@ class SyncHandler(object):
if full_state:
if batch:
current_state_ids = yield self.store.get_state_ids_for_event(
- batch.events[-1].event_id, types=types,
- filtered_types=filtered_types,
+ batch.events[-1].event_id, state_filter=state_filter,
)
state_ids = yield self.store.get_state_ids_for_event(
- batch.events[0].event_id, types=types,
- filtered_types=filtered_types,
+ batch.events[0].event_id, state_filter=state_filter,
)
else:
current_state_ids = yield self.get_state_at(
- room_id, stream_position=now_token, types=types,
- filtered_types=filtered_types,
+ room_id, stream_position=now_token,
+ state_filter=state_filter,
)
state_ids = current_state_ids
@@ -719,19 +768,31 @@ class SyncHandler(object):
lazy_load_members=lazy_load_members,
)
elif batch.limited:
- state_at_previous_sync = yield self.get_state_at(
- room_id, stream_position=since_token, types=types,
- filtered_types=filtered_types,
+ state_at_timeline_start = yield self.store.get_state_ids_for_event(
+ batch.events[0].event_id, state_filter=state_filter,
)
- current_state_ids = yield self.store.get_state_ids_for_event(
- batch.events[-1].event_id, types=types,
- filtered_types=filtered_types,
+ # for now, we disable LL for gappy syncs - see
+ # https://github.com/vector-im/riot-web/issues/7211#issuecomment-419976346
+ # N.B. this slows down incr syncs as we are now processing way
+ # more state in the server than if we were LLing.
+ #
+ # We still have to filter timeline_start to LL entries (above) in order
+ # for _calculate_state's LL logic to work, as we have to include LL
+ # members for timeline senders in case they weren't loaded in the initial
+ # sync. We do this by (counterintuitively) by filtering timeline_start
+ # members to just be ones which were timeline senders, which then ensures
+ # all of the rest get included in the state block (if we need to know
+ # about them).
+ state_filter = StateFilter.all()
+
+ state_at_previous_sync = yield self.get_state_at(
+ room_id, stream_position=since_token,
+ state_filter=state_filter,
)
- state_at_timeline_start = yield self.store.get_state_ids_for_event(
- batch.events[0].event_id, types=types,
- filtered_types=filtered_types,
+ current_state_ids = yield self.store.get_state_ids_for_event(
+ batch.events[-1].event_id, state_filter=state_filter,
)
state_ids = _calculate_state(
@@ -739,15 +800,28 @@ class SyncHandler(object):
timeline_start=state_at_timeline_start,
previous=state_at_previous_sync,
current=current_state_ids,
+ # we have to include LL members in case LL initial sync missed them
lazy_load_members=lazy_load_members,
)
else:
state_ids = {}
if lazy_load_members:
- if types:
+ if members_to_fetch and batch.events:
+ # We're returning an incremental sync, with no
+ # "gap" since the previous sync, so normally there would be
+ # no state to return.
+ # But we're lazy-loading, so the client might need some more
+ # member events to understand the events in this timeline.
+ # So we fish out all the member events corresponding to the
+ # timeline here, and then dedupe any redundant ones below.
+
state_ids = yield self.store.get_state_ids_for_event(
- batch.events[0].event_id, types=types,
- filtered_types=filtered_types,
+ batch.events[0].event_id,
+ # we only want members!
+ state_filter=StateFilter.from_types(
+ (EventTypes.Member, member)
+ for member in members_to_fetch
+ ),
)
if lazy_load_members and not include_redundant_members:
@@ -767,7 +841,7 @@ class SyncHandler(object):
logger.debug("filtering state from %r...", state_ids)
state_ids = {
t: event_id
- for t, event_id in state_ids.iteritems()
+ for t, event_id in iteritems(state_ids)
if cache.get(t[1]) != event_id
}
logger.debug("...to %r", state_ids)
@@ -1574,6 +1648,19 @@ class SyncHandler(object):
newly_joined_room=newly_joined,
)
+ # When we join the room (or the client requests full_state), we should
+ # send down any existing tags. Usually the user won't have tags in a
+ # newly joined room, unless either a) they've joined before or b) the
+ # tag was added by synapse e.g. for server notice rooms.
+ if full_state:
+ user_id = sync_result_builder.sync_config.user.to_string()
+ tags = yield self.store.get_tags_for_room(user_id, room_id)
+
+ # If there aren't any tags, don't send the empty tags list down
+ # sync
+ if not tags:
+ tags = None
+
account_data_events = []
if tags is not None:
account_data_events.append({
@@ -1602,10 +1689,24 @@ class SyncHandler(object):
)
summary = {}
+
+ # we include a summary in room responses when we're lazy loading
+ # members (as the client otherwise doesn't have enough info to form
+ # the name itself).
if (
sync_config.filter_collection.lazy_load_members() and
(
+ # we recalulate the summary:
+ # if there are membership changes in the timeline, or
+ # if membership has changed during a gappy sync, or
+ # if this is an initial sync.
any(ev.type == EventTypes.Member for ev in batch.events) or
+ (
+ # XXX: this may include false positives in the form of LL
+ # members which have snuck into state
+ batch.limited and
+ any(t == EventTypes.Member for (t, k) in state)
+ ) or
since_token is None
)
):
@@ -1635,6 +1736,16 @@ class SyncHandler(object):
unread_notifications["highlight_count"] = notifs["highlight_count"]
sync_result_builder.joined.append(room_sync)
+
+ if batch.limited and since_token:
+ user_id = sync_result_builder.sync_config.user.to_string()
+ logger.info(
+ "Incremental gappy sync of %s for user %s with %d state events" % (
+ room_id,
+ user_id,
+ len(state),
+ )
+ )
elif room_builder.rtype == "archived":
room_sync = ArchivedSyncResult(
room_id=room_id,
@@ -1728,17 +1839,17 @@ def _calculate_state(
event_id_to_key = {
e: key
for key, e in itertools.chain(
- timeline_contains.items(),
- previous.items(),
- timeline_start.items(),
- current.items(),
+ iteritems(timeline_contains),
+ iteritems(previous),
+ iteritems(timeline_start),
+ iteritems(current),
)
}
- c_ids = set(e for e in current.values())
- ts_ids = set(e for e in timeline_start.values())
- p_ids = set(e for e in previous.values())
- tc_ids = set(e for e in timeline_contains.values())
+ c_ids = set(e for e in itervalues(current))
+ ts_ids = set(e for e in itervalues(timeline_start))
+ p_ids = set(e for e in itervalues(previous))
+ tc_ids = set(e for e in itervalues(timeline_contains))
# If we are lazyloading room members, we explicitly add the membership events
# for the senders in the timeline into the state block returned by /sync,
@@ -1752,7 +1863,7 @@ def _calculate_state(
if lazy_load_members:
p_ids.difference_update(
- e for t, e in timeline_start.iteritems()
+ e for t, e in iteritems(timeline_start)
if t[0] == EventTypes.Member
)
|
