From 5dea4d37d160e5766aac6f1723a8b485c5b6c211 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Fri, 13 Nov 2015 10:31:15 +0000 Subject: Update some comments Add a couple of type annotations, docstrings, and other comments, in the interest of keeping track of what types I have. Merged from pull request #370. --- synapse/rest/client/v2_alpha/sync.py | 56 ++++++++++++++++++++++++++++++++++++ 1 file changed, 56 insertions(+) (limited to 'synapse/rest/client/v2_alpha/sync.py') diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index d24507effa..997a61abbb 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -165,6 +165,20 @@ class SyncRestServlet(RestServlet): return {"events": filter.filter_presence(formatted)} def encode_joined(self, rooms, filter, time_now, token_id): + """ + Encode the joined rooms in a sync result + + :param list[synapse.handlers.sync.JoinedSyncResult] rooms: list of sync + results for rooms this user is joined to + :param FilterCollection filter: filters to apply to the results + :param int time_now: current time - used as a baseline for age + calculations + :param int token_id: ID of the user's auth token - used for namespacing + of transaction IDs + + :return: the joined rooms list, in our response format + :rtype: dict[str, dict[str, object]] + """ joined = {} for room in rooms: joined[room.room_id] = self.encode_room( @@ -174,6 +188,20 @@ class SyncRestServlet(RestServlet): return joined def encode_invited(self, rooms, filter, time_now, token_id): + """ + Encode the invited rooms in a sync result + + :param list[synapse.handlers.sync.InvitedSyncResult] rooms: list of + sync results for rooms this user is joined to + :param FilterCollection filter: filters to apply to the results + :param int time_now: current time - used as a baseline for age + calculations + :param int token_id: ID of the user's auth token - used for namespacing + of transaction IDs + + :return: the invited rooms list, in our response format + :rtype: dict[str, dict[str, object]] + """ invited = {} for room in rooms: invite = serialize_event( @@ -189,6 +217,20 @@ class SyncRestServlet(RestServlet): return invited def encode_archived(self, rooms, filter, time_now, token_id): + """ + Encode the archived rooms in a sync result + + :param list[synapse.handlers.sync.ArchivedSyncResult] rooms: list of + sync results for rooms this user is joined to + :param FilterCollection filter: filters to apply to the results + :param int time_now: current time - used as a baseline for age + calculations + :param int token_id: ID of the user's auth token - used for namespacing + of transaction IDs + + :return: the invited rooms list, in our response format + :rtype: dict[str, dict[str, object]] + """ joined = {} for room in rooms: joined[room.room_id] = self.encode_room( @@ -199,6 +241,20 @@ class SyncRestServlet(RestServlet): @staticmethod def encode_room(room, filter, time_now, token_id, joined=True): + """ + :param JoinedSyncResult|ArchivedSyncResult room: sync result for a + single room + :param FilterCollection filter: filters to apply to the results + :param int time_now: current time - used as a baseline for age + calculations + :param int token_id: ID of the user's auth token - used for namespacing + of transaction IDs + :param joined: True if the user is joined to this room - will mean + we handle ephemeral events + + :return: the room, encoded in our response format + :rtype: dict[str, object] + """ event_map = {} state_events = filter.filter_room_state(room.state) state_event_ids = [] -- cgit 1.4.1 From 5ab4b0afe8b5126213cab2be7c3700eb7dd49789 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Thu, 12 Nov 2015 16:34:42 +0000 Subject: Make handlers.sync return a state dictionary, instead of an event list. Basically this moves the process of flattening the existing dictionary into a list up to rest.client.*, instead of doing it in handlers.sync. This simplifies a bit of the code in handlers.sync, but it is also going to be somewhat beneficial in the next stage of my hacking on SPEC-254. Merged from PR #371 --- synapse/handlers/sync.py | 70 ++++++++++++++++++++---------------- synapse/rest/client/v2_alpha/sync.py | 2 +- 2 files changed, 40 insertions(+), 32 deletions(-) (limited to 'synapse/rest/client/v2_alpha/sync.py') diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index ed93e5a2df..8b154fa7e7 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -49,7 +49,7 @@ class TimelineBatch(collections.namedtuple("TimelineBatch", [ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [ "room_id", # str "timeline", # TimelineBatch - "state", # list[FrozenEvent] + "state", # dict[(str, str), FrozenEvent] "ephemeral", "private_user_data", ])): @@ -70,7 +70,7 @@ class JoinedSyncResult(collections.namedtuple("JoinedSyncResult", [ class ArchivedSyncResult(collections.namedtuple("JoinedSyncResult", [ "room_id", # str "timeline", # TimelineBatch - "state", # list[FrozenEvent] + "state", # dict[(str, str), FrozenEvent] "private_user_data", ])): __slots__ = [] @@ -257,12 +257,11 @@ class SyncHandler(BaseHandler): current_state = yield self.state_handler.get_current_state( room_id ) - current_state_events = current_state.values() defer.returnValue(JoinedSyncResult( room_id=room_id, timeline=batch, - state=current_state_events, + state=current_state, ephemeral=ephemeral_by_room.get(room_id, []), private_user_data=self.private_user_data_for_room( room_id, tags_by_room @@ -361,7 +360,7 @@ class SyncHandler(BaseHandler): defer.returnValue(ArchivedSyncResult( room_id=room_id, timeline=batch, - state=leave_state[leave_event_id].values(), + state=leave_state[leave_event_id], private_user_data=self.private_user_data_for_room( room_id, tags_by_room ), @@ -440,7 +439,10 @@ class SyncHandler(BaseHandler): for room_id in joined_room_ids: recents = events_by_room_id.get(room_id, []) - state = [event for event in recents if event.is_state()] + state = { + (event.type, event.state_key): event + for event in recents if event.is_state()} + if recents: prev_batch = now_token.copy_and_replace( "room_key", recents[0].internal_metadata.before @@ -575,7 +577,6 @@ class SyncHandler(BaseHandler): current_state = yield self.state_handler.get_current_state( room_id ) - current_state_events = current_state.values() state_at_previous_sync = yield self.get_state_at_previous_sync( room_id, since_token=since_token @@ -584,7 +585,7 @@ class SyncHandler(BaseHandler): state_events_delta = yield self.compute_state_delta( since_token=since_token, previous_state=state_at_previous_sync, - current_state=current_state_events, + current_state=current_state, ) state_events_delta, _ = yield self.check_joined_room( @@ -632,7 +633,7 @@ class SyncHandler(BaseHandler): [leave_event.event_id], None ) - state_events_at_leave = leave_state[leave_event.event_id].values() + state_events_at_leave = leave_state[leave_event.event_id] state_at_previous_sync = yield self.get_state_at_previous_sync( leave_event.room_id, since_token=since_token @@ -661,7 +662,7 @@ class SyncHandler(BaseHandler): def get_state_at_previous_sync(self, room_id, since_token): """ Get the room state at the previous sync the client made. Returns: - A Deferred list of Events. + A Deferred map from ((type, state_key)->Event) """ last_events, token = yield self.store.get_recent_events_for_room( room_id, end_token=since_token.room_key, limit=1, @@ -673,11 +674,12 @@ class SyncHandler(BaseHandler): last_event ) if last_event.is_state(): - state = [last_event] + last_context.current_state.values() + state = last_context.current_state.copy() + state[(last_event.type, last_event.state_key)] = last_event else: - state = last_context.current_state.values() + state = last_context.current_state else: - state = () + state = {} defer.returnValue(state) def compute_state_delta(self, since_token, previous_state, current_state): @@ -685,21 +687,23 @@ class SyncHandler(BaseHandler): state the client got when it last performed a sync. :param str since_token: the point we are comparing against - :param list[synapse.events.FrozenEvent] previous_state: the state to - compare to - :param list[synapse.events.FrozenEvent] current_state: the new state + :param dict[(str,str), synapse.events.FrozenEvent] previous_state: the + state to compare to + :param dict[(str,str), synapse.events.FrozenEvent] current_state: the + new state - :returns: A list of events. + :returns A new event dictionary """ # TODO(mjark) Check if the state events were received by the server # after the previous sync, since we need to include those state # updates even if they occured logically before the previous event. # TODO(mjark) Check for new redactions in the state events. - previous_dict = {event.event_id: event for event in previous_state} - state_delta = [] - for event in current_state: - if event.event_id not in previous_dict: - state_delta.append(event) + + state_delta = {} + for key, event in current_state.iteritems(): + if (key not in previous_state or + previous_state[key].event_id != event.event_id): + state_delta[key] = event return state_delta @defer.inlineCallbacks @@ -708,21 +712,25 @@ class SyncHandler(BaseHandler): Check if the user has just joined the given room. If so, return the full state for the room, instead of the delta since the last sync. + :param sync_config: + :param room_id: + :param dict[(str,str), synapse.events.FrozenEvent] state_delta: the + difference in state since the last sync + :returns A deferred Tuple (state_delta, limited) """ joined = False limited = False - for event in state_delta: - if ( - event.type == EventTypes.Member - and event.state_key == sync_config.user.to_string() - ): - if event.content["membership"] == Membership.JOIN: - joined = True + + join_event = state_delta.get(( + EventTypes.Member, sync_config.user.to_string()), None) + if join_event is not None: + if join_event.content["membership"] == Membership.JOIN: + joined = True if joined: - res = yield self.state_handler.get_current_state(room_id) - state_delta = res.values() + state_delta = yield self.state_handler.get_current_state(room_id) + # the timeline is inherently limited if we've just joined limited = True defer.returnValue((state_delta, limited)) diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index 997a61abbb..272a00bc85 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -256,7 +256,7 @@ class SyncRestServlet(RestServlet): :rtype: dict[str, object] """ event_map = {} - state_events = filter.filter_room_state(room.state) + state_events = filter.filter_room_state(room.state.values()) state_event_ids = [] for event in state_events: # TODO(mjark): Respect formatting requirements in the filter. -- cgit 1.4.1 From e4d622aaaf0df503f942d016a5bf798dd52899d1 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Tue, 10 Nov 2015 18:29:25 +0000 Subject: Implementation of state rollback in /sync Implementation of SPEC-254: roll back the state dictionary to how it looked at the start of the timeline. Merged PR https://github.com/matrix-org/synapse/pull/373 --- synapse/rest/client/v2_alpha/sync.py | 67 ++++++++++++++++++++++++++++++++++-- synapse/storage/events.py | 6 ++-- 2 files changed, 69 insertions(+), 4 deletions(-) (limited to 'synapse/rest/client/v2_alpha/sync.py') diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index 272a00bc85..efd8281558 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -20,6 +20,7 @@ from synapse.http.servlet import ( ) from synapse.handlers.sync import SyncConfig from synapse.types import StreamToken +from synapse.events import FrozenEvent from synapse.events.utils import ( serialize_event, format_event_for_client_v2_without_event_id, ) @@ -256,7 +257,13 @@ class SyncRestServlet(RestServlet): :rtype: dict[str, object] """ event_map = {} - state_events = filter.filter_room_state(room.state.values()) + state_dict = room.state + timeline_events = filter.filter_room_timeline(room.timeline.events) + + state_dict = SyncRestServlet._rollback_state_for_timeline( + state_dict, timeline_events) + + state_events = filter.filter_room_state(state_dict.values()) state_event_ids = [] for event in state_events: # TODO(mjark): Respect formatting requirements in the filter. @@ -266,7 +273,6 @@ class SyncRestServlet(RestServlet): ) state_event_ids.append(event.event_id) - timeline_events = filter.filter_room_timeline(room.timeline.events) timeline_event_ids = [] for event in timeline_events: # TODO(mjark): Respect formatting requirements in the filter. @@ -297,6 +303,63 @@ class SyncRestServlet(RestServlet): return result + @staticmethod + def _rollback_state_for_timeline(state, timeline): + """ + Wind the state dictionary backwards, so that it represents the + state at the start of the timeline, rather than at the end. + + :param dict[(str, str), synapse.events.EventBase] state: the + state dictionary. Will be updated to the state before the timeline. + :param list[synapse.events.EventBase] timeline: the event timeline + :return: updated state dictionary + """ + logger.debug("Processing state dict %r; timeline %r", state, + [e.get_dict() for e in timeline]) + + result = state.copy() + + for timeline_event in reversed(timeline): + if not timeline_event.is_state(): + continue + + event_key = (timeline_event.type, timeline_event.state_key) + + logger.debug("Considering %s for removal", event_key) + + state_event = result.get(event_key) + if (state_event is None or + state_event.event_id != timeline_event.event_id): + # the event in the timeline isn't present in the state + # dictionary. + # + # the most likely cause for this is that there was a fork in + # the event graph, and the state is no longer valid. Really, + # the event shouldn't be in the timeline. We're going to ignore + # it for now, however. + logger.warn("Found state event %r in timeline which doesn't " + "match state dictionary", timeline_event) + continue + + prev_event_id = timeline_event.unsigned.get("replaces_state", None) + logger.debug("Replacing %s with %s in state dict", + timeline_event.event_id, prev_event_id) + + if prev_event_id is None: + del result[event_key] + else: + result[event_key] = FrozenEvent({ + "type": timeline_event.type, + "state_key": timeline_event.state_key, + "content": timeline_event.unsigned['prev_content'], + "sender": timeline_event.unsigned['prev_sender'], + "event_id": prev_event_id, + "room_id": timeline_event.room_id, + }) + logger.debug("New value: %r", result.get(event_key)) + + return result + def register_servlets(hs, http_server): SyncRestServlet(hs).register(http_server) diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 4a365ff639..5d35ca90b9 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -831,7 +831,8 @@ class EventsStore(SQLBaseStore): allow_none=True, ) if prev: - ev.unsigned["prev_content"] = prev.get_dict()["content"] + ev.unsigned["prev_content"] = prev.content + ev.unsigned["prev_sender"] = prev.sender self._get_event_cache.prefill( (ev.event_id, check_redacted, get_prev_content), ev @@ -888,7 +889,8 @@ class EventsStore(SQLBaseStore): get_prev_content=False, ) if prev: - ev.unsigned["prev_content"] = prev.get_dict()["content"] + ev.unsigned["prev_content"] = prev.content + ev.unsigned["prev_sender"] = prev.sender self._get_event_cache.prefill( (ev.event_id, check_redacted, get_prev_content), ev -- cgit 1.4.1