diff options
Diffstat (limited to 'synapse')
| -rw-r--r-- | synapse/__init__.py | 2 | ||||
| -rw-r--r-- | synapse/federation/send_queue.py | 14 | ||||
| -rw-r--r-- | synapse/federation/transaction_queue.py | 73 | ||||
| -rw-r--r-- | synapse/federation/transport/client.py | 128 | ||||
| -rw-r--r-- | synapse/handlers/appservice.py | 37 | ||||
| -rw-r--r-- | synapse/handlers/message.py | 104 | ||||
| -rw-r--r-- | synapse/handlers/register.py | 36 | ||||
| -rw-r--r-- | synapse/metrics/__init__.py | 43 | ||||
| -rw-r--r-- | synapse/metrics/metric.py | 32 | ||||
| -rw-r--r-- | synapse/python_dependencies.py | 2 | ||||
| -rw-r--r-- | synapse/rest/client/v1/room.py | 7 | ||||
| -rw-r--r-- | synapse/rest/client/v2_alpha/register.py | 32 | ||||
| -rw-r--r-- | synapse/storage/__init__.py | 16 | ||||
| -rw-r--r-- | synapse/storage/_base.py | 2 | ||||
| -rw-r--r-- | synapse/storage/engines/__init__.py | 5 | ||||
| -rw-r--r-- | synapse/storage/events.py | 3 | ||||
| -rw-r--r-- | synapse/storage/events_worker.py | 20 | ||||
| -rw-r--r-- | synapse/types.py | 2 | ||||
| -rw-r--r-- | synapse/util/caches/response_cache.py | 32 | ||||
| -rw-r--r-- | synapse/util/caches/stream_change_cache.py | 4 |
20 files changed, 414 insertions, 180 deletions
diff --git a/synapse/__init__.py b/synapse/__init__.py index a9d5198aba..7f6090baf8 100644 --- a/synapse/__init__.py +++ b/synapse/__init__.py @@ -16,4 +16,4 @@ """ This is a reference implementation of a Matrix home server. """ -__version__ = "0.27.2" +__version__ = "0.27.3-rc2" diff --git a/synapse/federation/send_queue.py b/synapse/federation/send_queue.py index 93e5acebc1..945832283f 100644 --- a/synapse/federation/send_queue.py +++ b/synapse/federation/send_queue.py @@ -35,7 +35,7 @@ from synapse.storage.presence import UserPresenceState from synapse.util.metrics import Measure import synapse.metrics -from blist import sorteddict +from sortedcontainers import SortedDict from collections import namedtuple import logging @@ -56,19 +56,19 @@ class FederationRemoteSendQueue(object): self.is_mine_id = hs.is_mine_id self.presence_map = {} # Pending presence map user_id -> UserPresenceState - self.presence_changed = sorteddict() # Stream position -> user_id + self.presence_changed = SortedDict() # Stream position -> user_id self.keyed_edu = {} # (destination, key) -> EDU - self.keyed_edu_changed = sorteddict() # stream position -> (destination, key) + self.keyed_edu_changed = SortedDict() # stream position -> (destination, key) - self.edus = sorteddict() # stream position -> Edu + self.edus = SortedDict() # stream position -> Edu - self.failures = sorteddict() # stream position -> (destination, Failure) + self.failures = SortedDict() # stream position -> (destination, Failure) - self.device_messages = sorteddict() # stream position -> destination + self.device_messages = SortedDict() # stream position -> destination self.pos = 1 - self.pos_time = sorteddict() + self.pos_time = SortedDict() # EVERYTHING IS SAD. In particular, python only makes new scopes when # we make a new function, so we need to make a new function so the inner diff --git a/synapse/federation/transaction_queue.py b/synapse/federation/transaction_queue.py index a141ec9953..963d938edd 100644 --- a/synapse/federation/transaction_queue.py +++ b/synapse/federation/transaction_queue.py @@ -169,7 +169,7 @@ class TransactionQueue(object): while True: last_token = yield self.store.get_federation_out_pos("events") next_token, events = yield self.store.get_all_new_events_stream( - last_token, self._last_poked_id, limit=20, + last_token, self._last_poked_id, limit=100, ) logger.debug("Handling %s -> %s", last_token, next_token) @@ -177,24 +177,33 @@ class TransactionQueue(object): if not events and next_token >= self._last_poked_id: break - for event in events: + @defer.inlineCallbacks + def handle_event(event): # Only send events for this server. send_on_behalf_of = event.internal_metadata.get_send_on_behalf_of() is_mine = self.is_mine_id(event.event_id) if not is_mine and send_on_behalf_of is None: - continue - - # Get the state from before the event. - # We need to make sure that this is the state from before - # the event and not from after it. - # Otherwise if the last member on a server in a room is - # banned then it won't receive the event because it won't - # be in the room after the ban. - destinations = yield self.state.get_current_hosts_in_room( - event.room_id, latest_event_ids=[ - prev_id for prev_id, _ in event.prev_events - ], - ) + return + + try: + # Get the state from before the event. + # We need to make sure that this is the state from before + # the event and not from after it. + # Otherwise if the last member on a server in a room is + # banned then it won't receive the event because it won't + # be in the room after the ban. + destinations = yield self.state.get_current_hosts_in_room( + event.room_id, latest_event_ids=[ + prev_id for prev_id, _ in event.prev_events + ], + ) + except Exception: + logger.exception( + "Failed to calculate hosts in room for event: %s", + event.event_id, + ) + return + destinations = set(destinations) if send_on_behalf_of is not None: @@ -207,12 +216,44 @@ class TransactionQueue(object): self._send_pdu(event, destinations) - events_processed_counter.inc_by(len(events)) + @defer.inlineCallbacks + def handle_room_events(events): + for event in events: + yield handle_event(event) + + events_by_room = {} + for event in events: + events_by_room.setdefault(event.room_id, []).append(event) + + yield logcontext.make_deferred_yieldable(defer.gatherResults( + [ + logcontext.run_in_background(handle_room_events, evs) + for evs in events_by_room.itervalues() + ], + consumeErrors=True + )) yield self.store.update_federation_out_pos( "events", next_token ) + if events: + now = self.clock.time_msec() + ts = yield self.store.get_received_ts(events[-1].event_id) + + synapse.metrics.event_processing_lag.set( + now - ts, "federation_sender", + ) + synapse.metrics.event_processing_last_ts.set( + ts, "federation_sender", + ) + + events_processed_counter.inc_by(len(events)) + + synapse.metrics.event_processing_positions.set( + next_token, "federation_sender", + ) + finally: self._is_processing = False diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index 50a967a7ec..6db8efa6dd 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -21,6 +21,7 @@ from synapse.api.urls import FEDERATION_PREFIX as PREFIX from synapse.util.logutils import log_function import logging +import urllib logger = logging.getLogger(__name__) @@ -50,7 +51,7 @@ class TransportLayerClient(object): logger.debug("get_room_state dest=%s, room=%s", destination, room_id) - path = PREFIX + "/state/%s/" % room_id + path = _create_path(PREFIX, "/state/%s/", room_id) return self.client.get_json( destination, path=path, args={"event_id": event_id}, ) @@ -72,7 +73,7 @@ class TransportLayerClient(object): logger.debug("get_room_state_ids dest=%s, room=%s", destination, room_id) - path = PREFIX + "/state_ids/%s/" % room_id + path = _create_path(PREFIX, "/state_ids/%s/", room_id) return self.client.get_json( destination, path=path, args={"event_id": event_id}, ) @@ -94,7 +95,7 @@ class TransportLayerClient(object): logger.debug("get_pdu dest=%s, event_id=%s", destination, event_id) - path = PREFIX + "/event/%s/" % (event_id, ) + path = _create_path(PREFIX, "/event/%s/", event_id) return self.client.get_json(destination, path=path, timeout=timeout) @log_function @@ -120,7 +121,7 @@ class TransportLayerClient(object): # TODO: raise? return - path = PREFIX + "/backfill/%s/" % (room_id,) + path = _create_path(PREFIX, "/backfill/%s/", room_id) args = { "v": event_tuples, @@ -158,9 +159,11 @@ class TransportLayerClient(object): # generated by the json_data_callback. json_data = transaction.get_dict() + path = _create_path(PREFIX, "/send/%s/", transaction.transaction_id) + response = yield self.client.put_json( transaction.destination, - path=PREFIX + "/send/%s/" % transaction.transaction_id, + path=path, data=json_data, json_data_callback=json_data_callback, long_retries=True, @@ -178,7 +181,7 @@ class TransportLayerClient(object): @log_function def make_query(self, destination, query_type, args, retry_on_dns_fail, ignore_backoff=False): - path = PREFIX + "/query/%s" % query_type + path = _create_path(PREFIX, "/query/%s", query_type) content = yield self.client.get_json( destination=destination, @@ -223,7 +226,7 @@ class TransportLayerClient(object): "make_membership_event called with membership='%s', must be one of %s" % (membership, ",".join(valid_memberships)) ) - path = PREFIX + "/make_%s/%s/%s" % (membership, room_id, user_id) + path = _create_path(PREFIX, "/make_%s/%s/%s", membership, room_id, user_id) ignore_backoff = False retry_on_dns_fail = False @@ -249,7 +252,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function def send_join(self, destination, room_id, event_id, content): - path = PREFIX + "/send_join/%s/%s" % (room_id, event_id) + path = _create_path(PREFIX, "/send_join/%s/%s", room_id, event_id) response = yield self.client.put_json( destination=destination, @@ -262,7 +265,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function def send_leave(self, destination, room_id, event_id, content): - path = PREFIX + "/send_leave/%s/%s" % (room_id, event_id) + path = _create_path(PREFIX, "/send_leave/%s/%s", room_id, event_id) response = yield self.client.put_json( destination=destination, @@ -281,7 +284,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function def send_invite(self, destination, room_id, event_id, content): - path = PREFIX + "/invite/%s/%s" % (room_id, event_id) + path = _create_path(PREFIX, "/invite/%s/%s", room_id, event_id) response = yield self.client.put_json( destination=destination, @@ -323,7 +326,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function def exchange_third_party_invite(self, destination, room_id, event_dict): - path = PREFIX + "/exchange_third_party_invite/%s" % (room_id,) + path = _create_path(PREFIX, "/exchange_third_party_invite/%s", room_id,) response = yield self.client.put_json( destination=destination, @@ -336,7 +339,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function def get_event_auth(self, destination, room_id, event_id): - path = PREFIX + "/event_auth/%s/%s" % (room_id, event_id) + path = _create_path(PREFIX, "/event_auth/%s/%s", room_id, event_id) content = yield self.client.get_json( destination=destination, @@ -348,7 +351,7 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function def send_query_auth(self, destination, room_id, event_id, content): - path = PREFIX + "/query_auth/%s/%s" % (room_id, event_id) + path = _create_path(PREFIX, "/query_auth/%s/%s", room_id, event_id) content = yield self.client.post_json( destination=destination, @@ -410,7 +413,7 @@ class TransportLayerClient(object): Returns: A dict containg the device keys. """ - path = PREFIX + "/user/devices/" + user_id + path = _create_path(PREFIX, "/user/devices/%s", user_id) content = yield self.client.get_json( destination=destination, @@ -460,7 +463,7 @@ class TransportLayerClient(object): @log_function def get_missing_events(self, destination, room_id, earliest_events, latest_events, limit, min_depth, timeout): - path = PREFIX + "/get_missing_events/%s" % (room_id,) + path = _create_path(PREFIX, "/get_missing_events/%s", room_id,) content = yield self.client.post_json( destination=destination, @@ -480,7 +483,7 @@ class TransportLayerClient(object): def get_group_profile(self, destination, group_id, requester_user_id): """Get a group profile """ - path = PREFIX + "/groups/%s/profile" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/profile", group_id,) return self.client.get_json( destination=destination, @@ -499,7 +502,7 @@ class TransportLayerClient(object): requester_user_id (str) content (dict): The new profile of the group """ - path = PREFIX + "/groups/%s/profile" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/profile", group_id,) return self.client.post_json( destination=destination, @@ -513,7 +516,7 @@ class TransportLayerClient(object): def get_group_summary(self, destination, group_id, requester_user_id): """Get a group summary """ - path = PREFIX + "/groups/%s/summary" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/summary", group_id,) return self.client.get_json( destination=destination, @@ -526,7 +529,7 @@ class TransportLayerClient(object): def get_rooms_in_group(self, destination, group_id, requester_user_id): """Get all rooms in a group """ - path = PREFIX + "/groups/%s/rooms" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/rooms", group_id,) return self.client.get_json( destination=destination, @@ -539,7 +542,7 @@ class TransportLayerClient(object): content): """Add a room to a group """ - path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,) + path = _create_path(PREFIX, "/groups/%s/room/%s", group_id, room_id,) return self.client.post_json( destination=destination, @@ -553,7 +556,10 @@ class TransportLayerClient(object): config_key, content): """Update room in group """ - path = PREFIX + "/groups/%s/room/%s/config/%s" % (group_id, room_id, config_key,) + path = _create_path( + PREFIX, "/groups/%s/room/%s/config/%s", + group_id, room_id, config_key, + ) return self.client.post_json( destination=destination, @@ -566,7 +572,7 @@ class TransportLayerClient(object): def remove_room_from_group(self, destination, group_id, requester_user_id, room_id): """Remove a room from a group """ - path = PREFIX + "/groups/%s/room/%s" % (group_id, room_id,) + path = _create_path(PREFIX, "/groups/%s/room/%s", group_id, room_id,) return self.client.delete_json( destination=destination, @@ -579,7 +585,7 @@ class TransportLayerClient(object): def get_users_in_group(self, destination, group_id, requester_user_id): """Get users in a group """ - path = PREFIX + "/groups/%s/users" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/users", group_id,) return self.client.get_json( destination=destination, @@ -592,7 +598,7 @@ class TransportLayerClient(object): def get_invited_users_in_group(self, destination, group_id, requester_user_id): """Get users that have been invited to a group """ - path = PREFIX + "/groups/%s/invited_users" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/invited_users", group_id,) return self.client.get_json( destination=destination, @@ -605,7 +611,10 @@ class TransportLayerClient(object): def accept_group_invite(self, destination, group_id, user_id, content): """Accept a group invite """ - path = PREFIX + "/groups/%s/users/%s/accept_invite" % (group_id, user_id) + path = _create_path( + PREFIX, "/groups/%s/users/%s/accept_invite", + group_id, user_id, + ) return self.client.post_json( destination=destination, @@ -618,7 +627,7 @@ class TransportLayerClient(object): def join_group(self, destination, group_id, user_id, content): """Attempts to join a group """ - path = PREFIX + "/groups/%s/users/%s/join" % (group_id, user_id) + path = _create_path(PREFIX, "/groups/%s/users/%s/join", group_id, user_id) return self.client.post_json( destination=destination, @@ -631,7 +640,7 @@ class TransportLayerClient(object): def invite_to_group(self, destination, group_id, user_id, requester_user_id, content): """Invite a user to a group """ - path = PREFIX + "/groups/%s/users/%s/invite" % (group_id, user_id) + path = _create_path(PREFIX, "/groups/%s/users/%s/invite", group_id, user_id) return self.client.post_json( destination=destination, @@ -647,7 +656,7 @@ class TransportLayerClient(object): invited. """ - path = PREFIX + "/groups/local/%s/users/%s/invite" % (group_id, user_id) + path = _create_path(PREFIX, "/groups/local/%s/users/%s/invite", group_id, user_id) return self.client.post_json( destination=destination, @@ -661,7 +670,7 @@ class TransportLayerClient(object): user_id, content): """Remove a user fron a group """ - path = PREFIX + "/groups/%s/users/%s/remove" % (group_id, user_id) + path = _create_path(PREFIX, "/groups/%s/users/%s/remove", group_id, user_id) return self.client.post_json( destination=destination, @@ -678,7 +687,7 @@ class TransportLayerClient(object): kicked from the group. """ - path = PREFIX + "/groups/local/%s/users/%s/remove" % (group_id, user_id) + path = _create_path(PREFIX, "/groups/local/%s/users/%s/remove", group_id, user_id) return self.client.post_json( destination=destination, @@ -693,7 +702,7 @@ class TransportLayerClient(object): the attestations """ - path = PREFIX + "/groups/%s/renew_attestation/%s" % (group_id, user_id) + path = _create_path(PREFIX, "/groups/%s/renew_attestation/%s", group_id, user_id) return self.client.post_json( destination=destination, @@ -708,11 +717,12 @@ class TransportLayerClient(object): """Update a room entry in a group summary """ if category_id: - path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % ( + path = _create_path( + PREFIX, "/groups/%s/summary/categories/%s/rooms/%s", group_id, category_id, room_id, ) else: - path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,) + path = _create_path(PREFIX, "/groups/%s/summary/rooms/%s", group_id, room_id,) return self.client.post_json( destination=destination, @@ -728,11 +738,12 @@ class TransportLayerClient(object): """Delete a room entry in a group summary """ if category_id: - path = PREFIX + "/groups/%s/summary/categories/%s/rooms/%s" % ( + path = _create_path( + PREFIX + "/groups/%s/summary/categories/%s/rooms/%s", group_id, category_id, room_id, ) else: - path = PREFIX + "/groups/%s/summary/rooms/%s" % (group_id, room_id,) + path = _create_path(PREFIX, "/groups/%s/summary/rooms/%s", group_id, room_id,) return self.client.delete_json( destination=destination, @@ -745,7 +756,7 @@ class TransportLayerClient(object): def get_group_categories(self, destination, group_id, requester_user_id): """Get all categories in a group """ - path = PREFIX + "/groups/%s/categories" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/categories", group_id,) return self.client.get_json( destination=destination, @@ -758,7 +769,7 @@ class TransportLayerClient(object): def get_group_category(self, destination, group_id, requester_user_id, category_id): """Get category info in a group """ - path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,) + path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,) return self.client.get_json( destination=destination, @@ -772,7 +783,7 @@ class TransportLayerClient(object): content): """Update a category in a group """ - path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,) + path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,) return self.client.post_json( destination=destination, @@ -787,7 +798,7 @@ class TransportLayerClient(object): category_id): """Delete a category in a group """ - path = PREFIX + "/groups/%s/categories/%s" % (group_id, category_id,) + path = _create_path(PREFIX, "/groups/%s/categories/%s", group_id, category_id,) return self.client.delete_json( destination=destination, @@ -800,7 +811,7 @@ class TransportLayerClient(object): def get_group_roles(self, destination, group_id, requester_user_id): """Get all roles in a group """ - path = PREFIX + "/groups/%s/roles" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/roles", group_id,) return self.client.get_json( destination=destination, @@ -813,7 +824,7 @@ class TransportLayerClient(object): def get_group_role(self, destination, group_id, requester_user_id, role_id): """Get a roles info """ - path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,) + path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,) return self.client.get_json( destination=destination, @@ -827,7 +838,7 @@ class TransportLayerClient(object): content): """Update a role in a group """ - path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,) + path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,) return self.client.post_json( destination=destination, @@ -841,7 +852,7 @@ class TransportLayerClient(object): def delete_group_role(self, destination, group_id, requester_user_id, role_id): """Delete a role in a group """ - path = PREFIX + "/groups/%s/roles/%s" % (group_id, role_id,) + path = _create_path(PREFIX, "/groups/%s/roles/%s", group_id, role_id,) return self.client.delete_json( destination=destination, @@ -856,11 +867,12 @@ class TransportLayerClient(object): """Update a users entry in a group """ if role_id: - path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % ( + path = _create_path( + PREFIX, "/groups/%s/summary/roles/%s/users/%s", group_id, role_id, user_id, ) else: - path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,) + path = _create_path(PREFIX, "/groups/%s/summary/users/%s", group_id, user_id,) return self.client.post_json( destination=destination, @@ -875,7 +887,7 @@ class TransportLayerClient(object): content): """Sets the join policy for a group """ - path = PREFIX + "/groups/%s/settings/m.join_policy" % (group_id,) + path = _create_path(PREFIX, "/groups/%s/settings/m.join_policy", group_id,) return self.client.put_json( destination=destination, @@ -891,11 +903,12 @@ class TransportLayerClient(object): """Delete a users entry in a group """ if role_id: - path = PREFIX + "/groups/%s/summary/roles/%s/users/%s" % ( + path = _create_path( + PREFIX, "/groups/%s/summary/roles/%s/users/%s", group_id, role_id, user_id, ) else: - path = PREFIX + "/groups/%s/summary/users/%s" % (group_id, user_id,) + path = _create_path(PREFIX, "/groups/%s/summary/users/%s", group_id, user_id,) return self.client.delete_json( destination=destination, @@ -918,3 +931,22 @@ class TransportLayerClient(object): data=content, ignore_backoff=True, ) + + +def _create_path(prefix, path, *args): + """Creates a path from the prefix, path template and args. Ensures that + all args are url encoded. + + Example: + + _create_path(PREFIX, "/event/%s/", event_id) + + Args: + prefix (str) + path (str): String template for the path + args: ([str]): Args to insert into path. Each arg will be url encoded + + Returns: + str + """ + return prefix + path % tuple(urllib.quote(arg, "") for arg in args) diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py index 3dd3fa2a27..0245197c02 100644 --- a/synapse/handlers/appservice.py +++ b/synapse/handlers/appservice.py @@ -18,7 +18,9 @@ from twisted.internet import defer import synapse from synapse.api.constants import EventTypes from synapse.util.metrics import Measure -from synapse.util.logcontext import make_deferred_yieldable, preserve_fn +from synapse.util.logcontext import ( + make_deferred_yieldable, preserve_fn, run_in_background, +) import logging @@ -84,11 +86,16 @@ class ApplicationServicesHandler(object): if not events: break + events_by_room = {} for event in events: + events_by_room.setdefault(event.room_id, []).append(event) + + @defer.inlineCallbacks + def handle_event(event): # Gather interested services services = yield self._get_services_for_event(event) if len(services) == 0: - continue # no services need notifying + return # no services need notifying # Do we know this user exists? If not, poke the user # query API for all services which match that user regex. @@ -108,9 +115,33 @@ class ApplicationServicesHandler(object): service, event ) - events_processed_counter.inc_by(len(events)) + @defer.inlineCallbacks + def handle_room_events(events): + for event in events: + yield handle_event(event) + + yield make_deferred_yieldable(defer.gatherResults([ + run_in_background(handle_room_events, evs) + for evs in events_by_room.itervalues() + ], consumeErrors=True)) yield self.store.set_appservice_last_pos(upper_bound) + + now = self.clock.time_msec() + ts = yield self.store.get_received_ts(events[-1].event_id) + + synapse.metrics.event_processing_positions.set( + upper_bound, "appservice_sender", + ) + + events_processed_counter.inc_by(len(events)) + + synapse.metrics.event_processing_lag.set( + now - ts, "appservice_sender", + ) + synapse.metrics.event_processing_last_ts.set( + ts, "appservice_sender", + ) finally: self.is_processing = False diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 6de6e13b7b..54cd691f91 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -454,40 +454,39 @@ class EventCreationHandler(object): """ builder = self.event_builder_factory.new(event_dict) - with (yield self.limiter.queue(builder.room_id)): - self.validator.validate_new(builder) - - if builder.type == EventTypes.Member: - membership = builder.content.get("membership", None) - target = UserID.from_string(builder.state_key) - - if membership in {Membership.JOIN, Membership.INVITE}: - # If event doesn't include a display name, add one. - profile = self.profile_handler - content = builder.content - - try: - if "displayname" not in content: - content["displayname"] = yield profile.get_displayname(target) - if "avatar_url" not in content: - content["avatar_url"] = yield profile.get_avatar_url(target) - except Exception as e: - logger.info( - "Failed to get profile information for %r: %s", - target, e - ) + self.validator.validate_new(builder) + + if builder.type == EventTypes.Member: + membership = builder.content.get("membership", None) + target = UserID.from_string(builder.state_key) + + if membership in {Membership.JOIN, Membership.INVITE}: + # If event doesn't include a display name, add one. + profile = self.profile_handler + content = builder.content + + try: + if "displayname" not in content: + content["displayname"] = yield profile.get_displayname(target) + if "avatar_url" not in content: + content["avatar_url"] = yield profile.get_avatar_url(target) + except Exception as e: + logger.info( + "Failed to get profile information for %r: %s", + target, e + ) - if token_id is not None: - builder.internal_metadata.token_id = token_id + if token_id is not None: + builder.internal_metadata.token_id = token_id - if txn_id is not None: - builder.internal_metadata.txn_id = txn_id + if txn_id is not None: + builder.internal_metadata.txn_id = txn_id - event, context = yield self.create_new_client_event( - builder=builder, - requester=requester, - prev_event_ids=prev_event_ids, - ) + event, context = yield self.create_new_client_event( + builder=builder, + requester=requester, + prev_event_ids=prev_event_ids, + ) defer.returnValue((event, context)) @@ -557,27 +556,34 @@ class EventCreationHandler(object): See self.create_event and self.send_nonmember_event. """ - event, context = yield self.create_event( - requester, - event_dict, - token_id=requester.access_token_id, - txn_id=txn_id - ) - spam_error = self.spam_checker.check_event_for_spam(event) - if spam_error: - if not isinstance(spam_error, basestring): - spam_error = "Spam is not permitted here" - raise SynapseError( - 403, spam_error, Codes.FORBIDDEN + # We limit the number of concurrent event sends in a room so that we + # don't fork the DAG too much. If we don't limit then we can end up in + # a situation where event persistence can't keep up, causing + # extremities to pile up, which in turn leads to state resolution + # taking longer. + with (yield self.limiter.queue(event_dict["room_id"])): + event, context = yield self.create_event( + requester, + event_dict, + token_id=requester.access_token_id, + txn_id=txn_id ) - yield self.send_nonmember_event( - requester, - event, - context, - ratelimit=ratelimit, - ) + spam_error = self.spam_checker.check_event_for_spam(event) + if spam_error: + if not isinstance(spam_error, basestring): + spam_error = "Spam is not permitted here" + raise SynapseError( + 403, spam_error, Codes.FORBIDDEN + ) + + yield self.send_nonmember_event( + requester, + event, + context, + ratelimit=ratelimit, + ) defer.returnValue(event) @measure_func("create_new_client_event") diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py index dd03705279..f83c6b3cf8 100644 --- a/synapse/handlers/register.py +++ b/synapse/handlers/register.py @@ -23,7 +23,7 @@ from synapse.api.errors import ( ) from synapse.http.client import CaptchaServerHttpClient from synapse import types -from synapse.types import UserID +from synapse.types import UserID, create_requester, RoomID, RoomAlias from synapse.util.async import run_on_reactor, Linearizer from synapse.util.threepids import check_3pid_allowed from ._base import BaseHandler @@ -205,10 +205,17 @@ class RegistrationHandler(BaseHandler): token = None attempts += 1 + # auto-join the user to any rooms we're supposed to dump them into + fake_requester = create_requester(user_id) + for r in self.hs.config.auto_join_rooms: + try: + yield self._join_user_to_room(fake_requester, r) + except Exception as e: + logger.error("Failed to join new user to %r: %r", r, e) + # We used to generate default identicons here, but nowadays # we want clients to generate their own as part of their branding # rather than there being consistent matrix-wide ones, so we don't. - defer.returnValue((user_id, token)) @defer.inlineCallbacks @@ -483,3 +490,28 @@ class RegistrationHandler(BaseHandler): ) defer.returnValue((user_id, access_token)) + + @defer.inlineCallbacks + def _join_user_to_room(self, requester, room_identifier): + room_id = None + room_member_handler = self.hs.get_room_member_handler() + if RoomID.is_valid(room_identifier): + room_id = room_identifier + elif RoomAlias.is_valid(room_identifier): + room_alias = RoomAlias.from_string(room_identifier) + room_id, remote_room_hosts = ( + yield room_member_handler.lookup_room_alias(room_alias) + ) + room_id = room_id.to_string() + else: + raise SynapseError(400, "%s was not legal room ID or room alias" % ( + room_identifier, + )) + + yield room_member_handler.update_membership( + requester=requester, + target=requester.user, + room_id=room_id, + remote_room_hosts=remote_room_hosts, + action="join", + ) diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py index 50d99d7a5c..e3b831db67 100644 --- a/synapse/metrics/__init__.py +++ b/synapse/metrics/__init__.py @@ -17,12 +17,13 @@ import logging import functools import time import gc +import platform from twisted.internet import reactor from .metric import ( CounterMetric, CallbackMetric, DistributionMetric, CacheMetric, - MemoryUsageMetric, + MemoryUsageMetric, GaugeMetric, ) from .process_collector import register_process_collector @@ -30,6 +31,7 @@ from .process_collector import register_process_collector logger = logging.getLogger(__name__) +running_on_pypy = platform.python_implementation() == 'PyPy' all_metrics = [] all_collectors = [] @@ -63,6 +65,13 @@ class Metrics(object): """ return self._register(CounterMetric, *args, **kwargs) + def register_gauge(self, *args, **kwargs): + """ + Returns: + GaugeMetric + """ + return self._register(GaugeMetric, *args, **kwargs) + def register_callback(self, *args, **kwargs): """ Returns: @@ -142,6 +151,32 @@ reactor_metrics = get_metrics_for("python.twisted.reactor") tick_time = reactor_metrics.register_distribution("tick_time") pending_calls_metric = reactor_metrics.register_distribution("pending_calls") +synapse_metrics = get_metrics_for("synapse") + +# Used to track where various components have processed in the event stream, +# e.g. federation sending, appservice sending, etc. +event_processing_positions = synapse_metrics.register_gauge( + "event_processing_positions", labels=["name"], +) + +# Used to track the current max events stream position +event_persisted_position = synapse_metrics.register_gauge( + "event_persisted_position", +) + +# Used to track the received_ts of the last event processed by various +# components +event_processing_last_ts = synapse_metrics.register_gauge( + "event_processing_last_ts", labels=["name"], +) + +# Used to track the lag processing events. This is the time difference +# between the last processed event's received_ts and the time it was +# finished being processed. +event_processing_lag = synapse_metrics.register_gauge( + "event_processing_lag", labels=["name"], +) + def runUntilCurrentTimer(func): @@ -174,6 +209,9 @@ def runUntilCurrentTimer(func): tick_time.inc_by(end - start) pending_calls_metric.inc_by(num_pending) + if running_on_pypy: + return ret + # Check if we need to do a manual GC (since its been disabled), and do # one if necessary. threshold = gc.get_threshold() @@ -206,6 +244,7 @@ try: # We manually run the GC each reactor tick so that we can get some metrics # about time spent doing GC, - gc.disable() + if not running_on_pypy: + gc.disable() except AttributeError: pass diff --git a/synapse/metrics/metric.py b/synapse/metrics/metric.py index ff5aa8c0e1..89bd47c3f7 100644 --- a/synapse/metrics/metric.py +++ b/synapse/metrics/metric.py @@ -115,7 +115,7 @@ class CounterMetric(BaseMetric): # dict[list[str]]: value for each set of label values. the keys are the # label values, in the same order as the labels in self.labels. # - # (if the metric is a scalar, the (single) key is the empty list). + # (if the metric is a scalar, the (single) key is the empty tuple). self.counts = {} # Scalar metrics are never empty @@ -145,6 +145,36 @@ class CounterMetric(BaseMetric): ) +class GaugeMetric(BaseMetric): + """A metric that can go up or down + """ + + def __init__(self, *args, **kwargs): + super(GaugeMetric, self).__init__(*args, **kwargs) + + # dict[list[str]]: value for each set of label values. the keys are the + # label values, in the same order as the labels in self.labels. + # + # (if the metric is a scalar, the (single) key is the empty tuple). + self.guages = {} + + def set(self, v, *values): + if len(values) != self.dimension(): + raise ValueError( + "Expected as many values to inc() as labels (%d)" % (self.dimension()) + ) + + # TODO: should assert that the tag values are all strings + + self.guages[values] = v + + def render(self): + return flatten( + self._render_for_labels(k, self.guages[k]) + for k in sorted(self.guages.keys()) + ) + + class CallbackMetric(BaseMetric): """A metric that returns the numeric value returned by a callback whenever it is rendered. Typically this is used to implement gauges that yield the diff --git a/synapse/python_dependencies.py b/synapse/python_dependencies.py index 40eedb63cb..f9596bddaf 100644 --- a/synapse/python_dependencies.py +++ b/synapse/python_dependencies.py @@ -34,8 +34,8 @@ REQUIREMENTS = { "bcrypt": ["bcrypt>=3.1.0"], "pillow": ["PIL"], "pydenticon": ["pydenticon"], - "blist": ["blist"], "pysaml2>=3.0.0": ["saml2>=3.0.0"], + "sortedcontainers": ["sortedcontainers"], "pymacaroons-pynacl": ["pymacaroons"], "msgpack-python>=0.3.0": ["msgpack"], "phonenumbers>=8.2.0": ["phonenumbers"], diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index d06cbdc35e..2ad0e5943b 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -165,17 +165,12 @@ class RoomStateEventRestServlet(ClientV1RestServlet): content=content, ) else: - event, context = yield self.event_creation_hander.create_event( + event = yield self.event_creation_hander.create_and_send_nonmember_event( requester, event_dict, - token_id=requester.access_token_id, txn_id=txn_id, ) - yield self.event_creation_hander.send_nonmember_event( - requester, event, context, - ) - ret = {} if event: ret = {"event_id": event.event_id} diff --git a/synapse/rest/client/v2_alpha/register.py b/synapse/rest/client/v2_alpha/register.py index 0ba62bddc1..f317c919dc 100644 --- a/synapse/rest/client/v2_alpha/register.py +++ b/synapse/rest/client/v2_alpha/register.py @@ -20,7 +20,6 @@ import synapse import synapse.types from synapse.api.auth import get_access_token_from_request, has_access_token from synapse.api.constants import LoginType -from synapse.types import RoomID, RoomAlias from synapse.api.errors import SynapseError, Codes, UnrecognizedRequestError from synapse.http.servlet import ( RestServlet, parse_json_object_from_request, assert_params_in_request, parse_string @@ -405,14 +404,6 @@ class RegisterRestServlet(RestServlet): generate_token=False, ) - # auto-join the user to any rooms we're supposed to dump them into - fake_requester = synapse.types.create_requester(registered_user_id) - for r in self.hs.config.auto_join_rooms: - try: - yield self._join_user_to_room(fake_requester, r) - except Exception as e: - logger.error("Failed to join new user to %r: %r", r, e) - # remember that we've now registered that user account, and with # what user ID (since the user may not have specified) self.auth_handler.set_session_data( @@ -446,29 +437,6 @@ class RegisterRestServlet(RestServlet): return 200, {} @defer.inlineCallbacks - def _join_user_to_room(self, requester, room_identifier): - room_id = None - if RoomID.is_valid(room_identifier): - room_id = room_identifier - elif RoomAlias.is_valid(room_identifier): - room_alias = RoomAlias.from_string(room_identifier) - room_id, remote_room_hosts = ( - yield self.room_member_handler.lookup_room_alias(room_alias) - ) - room_id = room_id.to_string() - else: - raise SynapseError(400, "%s was not legal room ID or room alias" % ( - room_identifier, - )) - - yield self.room_member_handler.update_membership( - requester=requester, - target=requester.user, - room_id=room_id, - action="join", - ) - - @defer.inlineCallbacks def _do_appservice_registration(self, username, as_token, body): user_id = yield self.registration_handler.appservice_register( username, as_token diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 4800584b59..8cdfd50f90 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -266,16 +266,16 @@ class DataStore(RoomMemberStore, RoomStore, def count_r30_users(self): """ Counts the number of 30 day retained users, defined as:- - * Users who have created their accounts more than 30 days + * Users who have created their accounts more than 30 days ago * Where last seen at most 30 days ago - * Where account creation and last_seen are > 30 days + * Where account creation and last_seen are > 30 days apart Returns counts globaly for a given user as well as breaking by platform """ def _count_r30_users(txn): thirty_days_in_secs = 86400 * 30 - now = int(self._clock.time_msec()) + now = int(self._clock.time()) thirty_days_ago_in_secs = now - thirty_days_in_secs sql = """ @@ -289,11 +289,11 @@ class DataStore(RoomMemberStore, RoomStore, user_id, last_seen, CASE - WHEN user_agent LIKE '%Android%' THEN 'android' - WHEN user_agent LIKE '%iOS%' THEN 'ios' - WHEN user_agent LIKE '%Electron%' THEN 'electron' - WHEN user_agent LIKE '%Mozilla%' THEN 'web' - WHEN user_agent LIKE '%Gecko%' THEN 'web' + WHEN user_agent LIKE '%%Android%%' THEN 'android' + WHEN user_agent LIKE '%%iOS%%' THEN 'ios' + WHEN user_agent LIKE '%%Electron%%' THEN 'electron' + WHEN user_agent LIKE '%%Mozilla%%' THEN 'web' + WHEN user_agent LIKE '%%Gecko%%' THEN 'web' ELSE 'unknown' END AS platform diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 2fbebd4907..2262776ab2 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -376,7 +376,7 @@ class SQLBaseStore(object): Returns: A list of dicts where the key is the column header. """ - col_headers = list(intern(column[0]) for column in cursor.description) + col_headers = list(intern(str(column[0])) for column in cursor.description) results = list( dict(zip(col_headers, row)) for row in cursor ) diff --git a/synapse/storage/engines/__init__.py b/synapse/storage/engines/__init__.py index 338b495611..8c868ece75 100644 --- a/synapse/storage/engines/__init__.py +++ b/synapse/storage/engines/__init__.py @@ -18,6 +18,7 @@ from .postgres import PostgresEngine from .sqlite3 import Sqlite3Engine import importlib +import platform SUPPORTED_MODULE = { @@ -31,6 +32,10 @@ def create_engine(database_config): engine_class = SUPPORTED_MODULE.get(name, None) if engine_class: + # pypy requires psycopg2cffi rather than psycopg2 + if (name == "psycopg2" and + platform.python_implementation() == "PyPy"): + name = "psycopg2cffi" module = importlib.import_module(name) return engine_class(module, database_config) diff --git a/synapse/storage/events.py b/synapse/storage/events.py index ece5e6c41f..da44b52fd6 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -444,6 +444,9 @@ class EventsStore(EventsWorkerStore): new_forward_extremeties=new_forward_extremeties, ) persist_event_counter.inc_by(len(chunk)) + synapse.metrics.event_persisted_position.set( + chunk[-1][0].internal_metadata.stream_ordering, + ) for event, context in chunk: if context.app_service: origin_type = "local" diff --git a/synapse/storage/events_worker.py b/synapse/storage/events_worker.py index 2e23dd78ba..a937b9bceb 100644 --- a/synapse/storage/events_worker.py +++ b/synapse/storage/events_worker.py @@ -51,6 +51,26 @@ _EventCacheEntry = namedtuple("_EventCacheEntry", ("event", "redacted_event")) class EventsWorkerStore(SQLBaseStore): + def get_received_ts(self, event_id): + """Get received_ts (when it was persisted) for the event. + + Raises an exception for unknown events. + + Args: + event_id (str) + + Returns: + Deferred[int|None]: Timestamp in milliseconds, or None for events + that were persisted before received_ts was implemented. + """ + return self._simple_select_one_onecol( + table="events", + keyvalues={ + "event_id": event_id, + }, + retcol="received_ts", + desc="get_received_ts", + ) @defer.inlineCallbacks def get_event(self, event_id, check_redacted=True, diff --git a/synapse/types.py b/synapse/types.py index 7cb24cecb2..cc7c182a78 100644 --- a/synapse/types.py +++ b/synapse/types.py @@ -169,7 +169,7 @@ class DomainSpecificString( except Exception: return False - __str__ = to_string + __repr__ = to_string class UserID(DomainSpecificString): diff --git a/synapse/util/caches/response_cache.py b/synapse/util/caches/response_cache.py index 7f68289723..066fa423fd 100644 --- a/synapse/util/caches/response_cache.py +++ b/synapse/util/caches/response_cache.py @@ -41,6 +41,18 @@ class ResponseCache(object): return len(self.pending_result_cache) def get(self, key): + """Look up the given key. + + Returns a deferred which doesn't follow the synapse logcontext rules, + so you'll probably want to make_deferred_yieldable it. + + Args: + key (str): + + Returns: + twisted.internet.defer.Deferred|None: None if there is no entry + for this key; otherwise a deferred result. + """ result = self.pending_result_cache.get(key) if result is not None: self._metrics.inc_hits() @@ -50,6 +62,26 @@ class ResponseCache(object): return None def set(self, key, deferred): + """Set the entry for the given key to the given deferred. + + *deferred* should run its callbacks in the sentinel logcontext (ie, + you should wrap normal synapse deferreds with + logcontext.run_in_background). + + Returns a new Deferred which also doesn't follow the synapse logcontext + rules, so you will want to make_deferred_yieldable it + + (TODO: before using this more widely, it might make sense to refactor + it and get() so that they do the necessary wrapping rather than having + to do it everywhere ResponseCache is used.) + + Args: + key (str): + deferred (twisted.internet.defer.Deferred): + + Returns: + twisted.internet.defer.Deferred + """ result = ObservableDeferred(deferred, consumeErrors=True) self.pending_result_cache[key] = result diff --git a/synapse/util/caches/stream_change_cache.py b/synapse/util/caches/stream_change_cache.py index 941d873ab8..2ff46090a6 100644 --- a/synapse/util/caches/stream_change_cache.py +++ b/synapse/util/caches/stream_change_cache.py @@ -16,7 +16,7 @@ from synapse.util.caches import register_cache, CACHE_SIZE_FACTOR -from blist import sorteddict +from sortedcontainers import SortedDict import logging @@ -35,7 +35,7 @@ class StreamChangeCache(object): def __init__(self, name, current_stream_pos, max_size=10000, prefilled_cache={}): self._max_size = int(max_size * CACHE_SIZE_FACTOR) self._entity_to_key = {} - self._cache = sorteddict() + self._cache = SortedDict() self._earliest_known_stream_pos = current_stream_pos self.name = name self.metrics = register_cache(self.name, self._cache) |