From c43d8981191ea993d78dea8be93fe513ad107c3e Mon Sep 17 00:00:00 2001 From: Kegan Dougal Date: Mon, 12 Jan 2015 17:38:30 +0000 Subject: SYN-178: Fix off by one. --- synapse/storage/stream.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index bedc3c6c52..744c821dfe 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -284,8 +284,12 @@ class StreamStore(SQLBaseStore): rows.reverse() # As we selected with reverse ordering if rows: - topo = rows[0]["topological_ordering"] - toke = rows[0]["stream_ordering"] + # XXX: Always subtract 1 since the start token always goes + # backwards (parity with paginate_room_events). It isn't + # obvious that this is correct; we should clarify the algorithm + # used here. + topo = rows[0]["topological_ordering"] - 1 + toke = rows[0]["stream_ordering"] - 1 start_token = "t%s-%s" % (topo, toke) token = (start_token, end_token) -- cgit 1.4.1 From 895fcb377e3ebc43d67df1ac66413d92aacffca1 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 13 Jan 2015 14:14:21 +0000 Subject: Fix stream token ordering --- synapse/storage/stream.py | 173 +++++++++++++++++++++++++++------------------- 1 file changed, 101 insertions(+), 72 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 744c821dfe..563c8e3bbb 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -39,6 +39,8 @@ from ._base import SQLBaseStore from synapse.api.errors import SynapseError from synapse.util.logutils import log_function +from collections import namedtuple + import logging @@ -52,58 +54,76 @@ _STREAM_TOKEN = "stream" _TOPOLOGICAL_TOKEN = "topological" -def _parse_stream_token(string): - try: - if string[0] != 's': - raise - return int(string[1:]) - except: - raise SynapseError(400, "Invalid token") - - -def _parse_topological_token(string): - try: - if string[0] != 't': - raise - parts = string[1:].split('-', 1) - return (int(parts[0]), int(parts[1])) - except: - raise SynapseError(400, "Invalid token") - - -def is_stream_token(string): - try: - _parse_stream_token(string) - return True - except: - return False - - -def is_topological_token(string): - try: - _parse_topological_token(string) - return True - except: - return False - - -def _get_token_bound(token, comparison): - try: - s = _parse_stream_token(token) - return "%s %s %d" % ("stream_ordering", comparison, s) - except: - pass - - try: - top, stream = _parse_topological_token(token) - return "%s %s %d AND %s %s %d" % ( - "topological_ordering", comparison, top, - "stream_ordering", comparison, stream, - ) - except: - pass +class _StreamToken(namedtuple("_StreamToken", "topological stream")): + """Tokens are positions between events. The token "s1" comes after event 1. + + s0 s1 + | | + [0] V [1] V [2] + + Tokens can either be a point in the live event stream or a cursor going + through historic events. + + When traversing the live event stream events are ordered by when they + arrived at the homeserver. + + When traversing historic events the events are ordered by their depth in + the event graph "topological_ordering" and then by when they arrived at the + homeserver "stream_ordering". + + Live tokens start with an "s" followed by the "stream_ordering" id of the + event it comes after. Historic tokens start with a "t" followed by the + "topological_ordering" id of the event it comes after, follewed by "-", + followed by the "stream_ordering" id of the event it comes after. + """ + __slots__ = [] + + @classmethod + def parse(cls, string): + try: + if string[0] == 's': + return cls(None, int(string[1:])) + if string[0] == 't': + parts = string[1:].split('-', 1) + return cls(int(parts[1]), int(parts[0])) + except: + pass + raise SynapseError(400, "Invalid token %r" % (string,)) + + @classmethod + def parse_stream_token(cls, string): + try: + if string[0] == 's': + return cls(None, int(string[1:])) + except: + pass + raise SynapseError(400, "Invalid token %r" % (string,)) + + def __str__(self): + if self.topological is not None: + return "t%d-%d" % (self.topological, self.stream) + else: + return "s%d" % (self.stream,) + + def lower_bound(self): + if self.topological is None: + return "(%d < %s)" % (self.stream, "stream_ordering") + else: + return "(%d < %s OR (%d == %s AND %d < %s))" % ( + self.topological, "topological_ordering", + self.topological, "topological_ordering", + self.stream, "stream_ordering", + ) - raise SynapseError(400, "Invalid token") + def upper_bound(self): + if self.topological is None: + return "(%d >= %s)" % (self.stream, "stream_ordering") + else: + return "(%d > %s OR (%d == %s AND %d >= %s))" % ( + self.topological, "topological_ordering", + self.topological, "topological_ordering", + self.stream, "stream_ordering", + ) class StreamStore(SQLBaseStore): @@ -162,8 +182,8 @@ class StreamStore(SQLBaseStore): limit = MAX_STREAM_SIZE # From and to keys should be integers from ordering. - from_id = _parse_stream_token(from_key) - to_id = _parse_stream_token(to_key) + from_id = _StreamToken.parse_stream_token(from_key) + to_id = _StreamToken.parse_stream_token(to_key) if from_key == to_key: return defer.succeed(([], to_key)) @@ -181,7 +201,7 @@ class StreamStore(SQLBaseStore): } def f(txn): - txn.execute(sql, (user_id, user_id, from_id, to_id,)) + txn.execute(sql, (user_id, user_id, from_id.stream, to_id.stream,)) rows = self.cursor_to_dict(txn) @@ -211,17 +231,21 @@ class StreamStore(SQLBaseStore): # Tokens really represent positions between elements, but we use # the convention of pointing to the event before the gap. Hence # we have a bit of asymmetry when it comes to equalities. - from_comp = '<=' if direction == 'b' else '>' - to_comp = '>' if direction == 'b' else '<=' - order = "DESC" if direction == 'b' else "ASC" - args = [room_id] - - bounds = _get_token_bound(from_key, from_comp) - if to_key: - bounds = "%s AND %s" % ( - bounds, _get_token_bound(to_key, to_comp) - ) + if direction == 'b': + order = "DESC" + bounds = _StreamToken.parse(from_key).upper_bound() + if to_key: + bounds = "%s AND %s" % ( + bounds, _StreamToken.parse(to_key).lower_bound() + ) + else: + order = "ASC" + bounds = _StreamToken.parse(from_key).lower_bound() + if to_key: + bounds = "%s AND %s" % ( + bounds, _StreamToken.parse(to_key).upper_bound() + ) if int(limit) > 0: args.append(int(limit)) @@ -249,9 +273,13 @@ class StreamStore(SQLBaseStore): topo = rows[-1]["topological_ordering"] toke = rows[-1]["stream_ordering"] if direction == 'b': - topo -= 1 + # Tokens are positions between events. + # This token points *after* the last event in the chunk. + # We need it to point to the event before it in the chunk + # when we are going backwards so we subtract one from the + # stream part. toke -= 1 - next_token = "t%s-%s" % (topo, toke) + next_token = str(_StreamToken(topo, toke)) else: # TODO (erikj): We should work out what to do here instead. next_token = to_key if to_key else from_key @@ -284,13 +312,14 @@ class StreamStore(SQLBaseStore): rows.reverse() # As we selected with reverse ordering if rows: - # XXX: Always subtract 1 since the start token always goes - # backwards (parity with paginate_room_events). It isn't - # obvious that this is correct; we should clarify the algorithm - # used here. - topo = rows[0]["topological_ordering"] - 1 + # Tokens are positions between events. + # This token points *after* the last event in the chunk. + # We need it to point to the event before it in the chunk + # since we are going backwards so we subtract one from the + # stream part. + topo = rows[0]["topological_ordering"] toke = rows[0]["stream_ordering"] - 1 - start_token = "t%s-%s" % (topo, toke) + start_token = str(_StreamToken(topo, toke)) token = (start_token, end_token) else: -- cgit 1.4.1 From fda63064fc21881c8aabbfef36916b3b00af5299 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 13 Jan 2015 14:43:26 +0000 Subject: get_room_events isn't called anywhere --- synapse/storage/stream.py | 30 ------------------------------ 1 file changed, 30 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 563c8e3bbb..8ac2adab05 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -127,36 +127,6 @@ class _StreamToken(namedtuple("_StreamToken", "topological stream")): class StreamStore(SQLBaseStore): - @log_function - def get_room_events(self, user_id, from_key, to_key, room_id, limit=0, - direction='f', with_feedback=False): - # We deal with events request in two different ways depending on if - # this looks like an /events request or a pagination request. - is_events = ( - direction == 'f' - and user_id - and is_stream_token(from_key) - and to_key and is_stream_token(to_key) - ) - - if is_events: - return self.get_room_events_stream( - user_id=user_id, - from_key=from_key, - to_key=to_key, - room_id=room_id, - limit=limit, - with_feedback=with_feedback, - ) - else: - return self.paginate_room_events( - from_key=from_key, - to_key=to_key, - room_id=room_id, - limit=limit, - with_feedback=with_feedback, - ) - @log_function def get_room_events_stream(self, user_id, from_key, to_key, room_id, limit=0, with_feedback=False): -- cgit 1.4.1 From a56008842b43089433768f569f35b2d14523ac39 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 27 Jan 2015 16:24:22 +0000 Subject: Start implementing incremental initial sync --- synapse/events/utils.py | 1 + synapse/handlers/sync.py | 233 +++++++++++++++++++++++++++++++++++++++++----- synapse/storage/stream.py | 41 ++++++-- 3 files changed, 241 insertions(+), 34 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/events/utils.py b/synapse/events/utils.py index b7f1ad4b40..42fb0371e5 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -137,6 +137,7 @@ def serialize_event(e, time_now_ms, client_event=True, strip_ids=False): d.pop("depth", None) d.pop("unsigned", None) d.pop("origin", None) + d.pop("prev_state", None) if strip_ids: d.pop("room_id", None) diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index bbabaf3df1..f8629a588f 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -37,14 +37,18 @@ SyncConfig = collections.namedtuple("SyncConfig", [ ]) -RoomSyncResult = collections.namedtuple("RoomSyncResult", [ +class RoomSyncResult(collections.namedtuple("RoomSyncResult", [ "room_id", "limited", "published", - "events", # dict of event + "events", "state", "prev_batch", -]) +])): + __slots__ = [] + + def __nonzero__(self): + return bool(self.events or self.state) class SyncResult(collections.namedtuple("SyncResult", [ @@ -56,7 +60,9 @@ class SyncResult(collections.namedtuple("SyncResult", [ __slots__ = [] def __nonzero__(self): - return self.private_user_data or self.public_user_data or self.rooms + return bool( + self.private_user_data or self.public_user_data or self.rooms + ) class SyncHandler(BaseHandler): @@ -67,7 +73,13 @@ class SyncHandler(BaseHandler): self.clock = hs.get_clock() def wait_for_sync_for_user(self, sync_config, since_token=None, timeout=0): - if timeout == 0: + """Get the sync for a client if we have new data for it now. Otherwise + wait for new data to arrive on the server. If the timeout expires, then + return an empty sync result. + Returns: + A Deferred SyncResult. + """ + if timeout == 0 or since_token is None: return self.current_sync_for_user(sync_config, since_token) else: def current_sync_callback(since_token): @@ -79,13 +91,25 @@ class SyncHandler(BaseHandler): ) def current_sync_for_user(self, sync_config, since_token=None): + """Get the sync for client needed to match what the server has now. + Returns: + A Deferred SyncResult. + """ if since_token is None: return self.initial_sync(sync_config) else: - return self.incremental_sync(sync_config) + if sync_config.gap: + return self.incremental_sync_with_gap(sync_config, since_token) + else: + #TODO(mjark): Handle gapless sync + pass @defer.inlineCallbacks def initial_sync(self, sync_config): + """Get a sync for a client which is starting without any state + Returns: + A Deferred SyncResult. + """ if sync_config.sort == "timeline,desc": # TODO(mjark): Handle going through events in reverse order?. # What does "most recent events" mean when applying the limits mean @@ -114,25 +138,86 @@ class SyncHandler(BaseHandler): rooms = [] for event in room_list: - #TODO (mjark): Apply the event filter in sync_config. - recent_events, token = yield self.store.get_recent_events_for_room( - event.room_id, - limit=sync_config.limit, - end_token=now_token.room_key, - ) - prev_batch_token = now_token.copy_and_replace("room_key", token[0]) - current_state_events = yield self.state_handler.get_current_state( - event.room_id + room_sync = yield self.initial_sync_for_room( + event.room_id, sync_config, now_token, published_room_ids ) + rooms.append(room_sync) + + defer.returnValue(SyncResult( + public_user_data=presence, + private_user_data=[], + rooms=rooms, + next_batch=now_token, + )) + + @defer.inlineCallbacks + def intial_sync_for_room(self, room_id, sync_config, now_token, + published_room_ids): + """Sync a room for a client which is starting without any state + Returns: + A Deferred RoomSyncResult. + """ + recent_events, token = yield self.store.get_recent_events_for_room( + room_id, + limit=sync_config.limit, + end_token=now_token.room_key, + ) + prev_batch_token = now_token.copy_and_replace("room_key", token[0]) + current_state_events = yield self.state_handler.get_current_state( + room_id + ) + + defer.returnValue(RoomSyncResult( + room_id=room_id, + published=room_id in published_room_ids, + events=recent_events, + prev_batch=prev_batch_token, + state=current_state_events, + limited=True, + )) + + + @defer.inlineCallbacks + def incremental_sync_with_gap(self, sync_config, since_token): + """ Get the incremental delta needed to bring the client up to + date with the server. + Returns: + A Deferred SyncResult. + """ + if sync_config.sort == "timeline,desc": + # TODO(mjark): Handle going through events in reverse order?. + # What does "most recent events" mean when applying the limits mean + # in this case? + raise NotImplementedError() + + now_token = yield self.event_sources.get_current_token() + + presence_stream = self.event_sources.sources["presence"] + pagination_config = PaginationConfig( + from_token=since_token, to_token=now_token + ) + presence, _ = yield presence_stream.get_pagination_rows( + user=sync_config.user, + pagination_config=pagination_config.get_source_config("presence"), + key=None + ) + room_list = yield self.store.get_rooms_for_user_where_membership_is( + user_id=sync_config.user.to_string(), + membership_list=[Membership.INVITE, Membership.JOIN] + ) + + # TODO (mjark): Does public mean "published"? + published_rooms = yield self.store.get_rooms(is_public=True) + published_room_ids = set(r["room_id"] for r in published_rooms) - rooms.append(RoomSyncResult( - room_id=event.room_id, - published=event.room_id in published_room_ids, - events=recent_events, - prev_batch=prev_batch_token, - state=current_state_events, - limited=True, - )) + rooms = [] + for event in room_list: + room_sync = yield self.incremental_sync_with_gap_for_room( + event.room_id, sync_config, since_token, now_token, + published_room_ids + ) + if room_sync: + rooms.append(room_sync) defer.returnValue(SyncResult( public_user_data=presence, @@ -143,5 +228,103 @@ class SyncHandler(BaseHandler): @defer.inlineCallbacks - def incremental_sync(self, sync_config): - pass + def incremental_sync_with_gap_for_room(self, room_id, sync_config, + since_token, now_token, + published_room_ids): + """ Get the incremental delta needed to bring the client up to date for + the room. Gives the client the most recent events and the changes to + state. + Returns: + A Deferred RoomSyncResult + """ + # TODO(mjark): Check if they have joined the room between + # the previous sync and this one. + # TODO(mjark): Apply the event filter in sync_config + # TODO(mjark): Check for redactions we might have missed. + # TODO(mjark): Typing notifications. + recents, token = yield self.store.get_recent_events_for_room( + room_id, + limit=sync_config.limit + 1, + from_token=since_token.room_key, + end_token=now_token.room_key, + ) + + logging.debug("Recents %r", recents) + + if len(recents) > sync_config.limit: + limited = True + recents = recents[1:] + else: + limited = False + + prev_batch_token = now_token.copy_and_replace("room_key", token[0]) + + # TODO(mjark): This seems racy since this isn't being passed a + # token to indicate what point in the stream this is + current_state_events = yield self.state_handler.get_current_state( + room_id + ) + + state_at_previous_sync = yield self.get_state_at_previous_sync( + room_id, since_token=since_token + ) + + state_events_delta = yield self.compute_state_delta( + since_token=since_token, + previous_state=state_at_previous_sync, + current_state=current_state_events, + ) + + room_sync = RoomSyncResult( + room_id=room_id, + published=room_id in published_room_ids, + events=recents, + prev_batch=prev_batch_token, + state=state_events_delta, + limited=limited, + ) + + logging.debug("Room sync: %r", room_sync) + + defer.returnValue(room_sync) + + @defer.inlineCallbacks + 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. + """ + last_events, token = yield self.store.get_recent_events_for_room( + room_id, end_token=since_token.room_key, limit=1, + ) + + if last_events: + last_event = last_events[0] + last_context = yield self.state_handler.compute_event_context( + last_event + ) + if last_event.is_state(): + state = [last_event] + last_context.current_state.values() + else: + state = last_context.current_state.values() + else: + state = () + defer.returnValue(state) + + + def compute_state_delta(self, since_token, previous_state, current_state): + """ Works out the differnce in state between the current state and the + state the client got when it last performed a sync. + Returns: + A list of events. + """ + # 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) + return state_delta diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 8ac2adab05..06aca1a4e5 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -265,17 +265,38 @@ class StreamStore(SQLBaseStore): return self.runInteraction("paginate_room_events", f) def get_recent_events_for_room(self, room_id, limit, end_token, - with_feedback=False): + with_feedback=False, from_token=None): # TODO (erikj): Handle compressed feedback - sql = ( - "SELECT stream_ordering, topological_ordering, event_id FROM events " - "WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0 " - "ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? " - ) + end_token = _StreamToken.parse_stream_token(end_token) - def f(txn): - txn.execute(sql, (room_id, end_token, limit,)) + if from_token is None: + sql = ( + "SELECT stream_ordering, topological_ordering, event_id" + " FROM events" + " WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0" + " ORDER BY topological_ordering DESC, stream_ordering DESC" + " LIMIT ?" + ) + else: + from_token = _StreamToken.parse_stream_token(from_token) + sql = ( + "SELECT stream_ordering, topological_ordering, event_id" + " FROM events" + " WHERE room_id = ? AND stream_ordering > ?" + " AND stream_ordering <= ? AND outlier = 0" + " ORDER BY topological_ordering DESC, stream_ordering DESC" + " LIMIT ?" + ) + + + def get_recent_events_for_room_txn(txn): + if from_token is None: + txn.execute(sql, (room_id, end_token.stream, limit,)) + else: + txn.execute(sql, ( + room_id, from_token.stream, end_token.stream, limit + )) rows = self.cursor_to_dict(txn) @@ -303,7 +324,9 @@ class StreamStore(SQLBaseStore): return events, token - return self.runInteraction("get_recent_events_for_room", f) + return self.runInteraction( + "get_recent_events_for_room", get_recent_events_for_room_txn + ) def get_room_events_max_id(self): return self.runInteraction( -- cgit 1.4.1 From e020574d65a994858ac53c45070ae5016090d2f3 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 27 Jan 2015 20:19:36 +0000 Subject: Fix Formatting --- synapse/handlers/sync.py | 13 +++++-------- synapse/notifier.py | 4 ++-- synapse/rest/client/v2_alpha/sync.py | 5 ++--- synapse/storage/stream.py | 1 - 4 files changed, 9 insertions(+), 14 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 9f5f73eab6..82a2c6986a 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -52,10 +52,10 @@ class RoomSyncResult(collections.namedtuple("RoomSyncResult", [ class SyncResult(collections.namedtuple("SyncResult", [ - "next_batch", # Token for the next sync - "private_user_data", # List of private events for the user. - "public_user_data", # List of public events for all users. - "rooms", # RoomSyncResult for each room. + "next_batch", # Token for the next sync + "private_user_data", # List of private events for the user. + "public_user_data", # List of public events for all users. + "rooms", # RoomSyncResult for each room. ])): __slots__ = [] @@ -181,7 +181,6 @@ class SyncHandler(BaseHandler): limited=True, )) - @defer.inlineCallbacks def incremental_sync_with_gap(self, sync_config, since_token): """ Get the incremental delta needed to bring the client up to @@ -231,7 +230,6 @@ class SyncHandler(BaseHandler): next_batch=now_token, )) - @defer.inlineCallbacks def incremental_sync_with_gap_for_room(self, room_id, sync_config, since_token, now_token, @@ -316,7 +314,6 @@ class SyncHandler(BaseHandler): state = () defer.returnValue(state) - def compute_state_delta(self, since_token, previous_state, current_state): """ Works out the differnce in state between the current state and the state the client got when it last performed a sync. @@ -327,7 +324,7 @@ class SyncHandler(BaseHandler): # 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} + 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: diff --git a/synapse/notifier.py b/synapse/notifier.py index 922bf064d0..e3b6ead620 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -214,7 +214,7 @@ class Notifier(object): deferred = defer.Deferred() - from_token=StreamToken("s0","0","0") + from_token = StreamToken("s0", "0", "0") listener = [_NotificationListener( user=user, @@ -231,6 +231,7 @@ class Notifier(object): result = yield callback() if timeout: timed_out = [False] + def _timeout_listener(): timed_out[0] = True listener[0].notify(self, [], from_token, from_token) @@ -252,7 +253,6 @@ class Notifier(object): defer.returnValue(result) - def get_events_for(self, user, rooms, pagination_config, timeout): """ For the given user and rooms, return any new events for them. If there are no new events wait for up to `timeout` milliseconds for any diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index cc667ebafc..0c17208cd3 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -68,7 +68,6 @@ class SyncRestServlet(RestServlet): } """ - PATTERN = client_v2_pattern("/sync$") ALLOWED_SORT = set(["timeline,asc", "timeline,desc"]) ALLOWED_PRESENCE = set(["online", "offline", "idle"]) @@ -114,12 +113,12 @@ class SyncRestServlet(RestServlet): sync_config = SyncConfig( user=user, - device="TODO", # TODO(mjark) Get the device_id from access_token + device="TODO", # TODO(mjark) Get the device_id from access_token gap=gap, limit=limit, sort=sort, backfill=backfill, - filter="TODO", # TODO(mjark) Add the filter to the config. + filter="TODO", # TODO(mjark) Add the filter to the config. ) if since is not None: diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 06aca1a4e5..db1816ea84 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -289,7 +289,6 @@ class StreamStore(SQLBaseStore): " LIMIT ?" ) - def get_recent_events_for_room_txn(txn): if from_token is None: txn.execute(sql, (room_id, end_token.stream, limit,)) -- cgit 1.4.1 From b0b80074e04dcb2b70dff56b0368060a19c065d3 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 29 Jan 2015 01:48:36 +0000 Subject: SYN-252: Supply the stream and topological parts in the correct order to the constructor --- synapse/storage/stream.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 8ac2adab05..062ca06fb3 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -82,10 +82,10 @@ class _StreamToken(namedtuple("_StreamToken", "topological stream")): def parse(cls, string): try: if string[0] == 's': - return cls(None, int(string[1:])) + return cls(topological=None, stream=int(string[1:])) if string[0] == 't': parts = string[1:].split('-', 1) - return cls(int(parts[1]), int(parts[0])) + return cls(topological=int(parts[0]), stream=int(parts[1])) except: pass raise SynapseError(400, "Invalid token %r" % (string,)) @@ -94,7 +94,7 @@ class _StreamToken(namedtuple("_StreamToken", "topological stream")): def parse_stream_token(cls, string): try: if string[0] == 's': - return cls(None, int(string[1:])) + return cls(topological=None, stream=int(string[1:])) except: pass raise SynapseError(400, "Invalid token %r" % (string,)) -- cgit 1.4.1 From e016f4043b81ffdedf71c4459772f66757386e44 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Thu, 29 Jan 2015 14:40:28 +0000 Subject: Use get_room_events_stream to get changes to the rooms if the number of changes is small --- synapse/handlers/sync.py | 56 +++++++++++++++++++++++++++++++++++++---------- synapse/storage/stream.py | 7 ++++++ 2 files changed, 52 insertions(+), 11 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 9e1188da56..e93dfe005d 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -216,23 +216,57 @@ class SyncHandler(BaseHandler): typing_by_room = {event["room_id"]: event for event in typing} logger.debug("Typing %r", typing_by_room) - room_list = yield self.store.get_rooms_for_user_where_membership_is( - user_id=sync_config.user.to_string(), - membership_list=[Membership.INVITE, Membership.JOIN] - ) + rm_handler = self.hs.get_handlers().room_member_handler + room_ids = yield rm_handler.get_rooms_for_user(sync_config.user) # TODO (mjark): Does public mean "published"? published_rooms = yield self.store.get_rooms(is_public=True) published_room_ids = set(r["room_id"] for r in published_rooms) + room_events, _ = yield self.store.get_room_events_stream( + sync_config.user.to_string(), + from_key=since_token.room_key, + to_key=now_token.room_key, + room_id=None, + limit=sync_config.limit + 1, + ) + rooms = [] - for event in room_list: - room_sync = yield self.incremental_sync_with_gap_for_room( - event.room_id, sync_config, since_token, now_token, - published_room_ids, typing_by_room - ) - if room_sync: - rooms.append(room_sync) + if len(room_events) <= sync_config.limit: + # There is no gap in any of the rooms. Therefore we can just + # partition the new events by room and return them. + events_by_room_id = {} + for event in room_events: + events_by_room_id.setdefault(event.room_id, []).append(event) + + for room_id in room_ids: + recents = events_by_room_id.get(room_id, []) + state = [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 + ) + else: + prev_batch = now_token + room_sync = RoomSyncResult( + room_id=room_id, + published=room_id in published_room_ids, + events=recents, + prev_batch=prev_batch, + state=state, + limited=False, + typing=typing_by_room.get(room_id, None) + ) + if room_sync is not None: + rooms.append(room_sync) + else: + for room_id in room_ids: + room_sync = yield self.incremental_sync_with_gap_for_room( + room_id, sync_config, since_token, now_token, + published_room_ids, typing_by_room + ) + if room_sync: + rooms.append(room_sync) defer.returnValue(SyncResult( public_user_data=presence, diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index db1816ea84..93ccfd8c10 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -181,6 +181,13 @@ class StreamStore(SQLBaseStore): get_prev_content=True ) + for event, row in zip(ret, rows): + stream = row["stream_ordering"] + topo = event.depth + internal = event.internal_metadata + internal.before = str(_StreamToken(topo, stream - 1)) + internal.after = str(_StreamToken(topo, stream)) + if rows: key = "s%d" % max([r["stream_ordering"] for r in rows]) else: -- cgit 1.4.1 From 22dd1cde2d83a2448074816108b85d1957315236 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 30 Jan 2015 11:32:35 +0000 Subject: Filter the recent events before applying the limit when doing an incremental sync with a gap --- synapse/api/filtering.py | 2 -- synapse/handlers/sync.py | 53 ++++++++++++++++++++++++++---------- synapse/rest/client/v2_alpha/sync.py | 2 +- synapse/storage/stream.py | 21 ++++++++++---- 4 files changed, 54 insertions(+), 24 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/api/filtering.py b/synapse/api/filtering.py index b7e5d3222f..fa4de2614d 100644 --- a/synapse/api/filtering.py +++ b/synapse/api/filtering.py @@ -12,8 +12,6 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -from twisted.internet import defer - from synapse.api.errors import SynapseError from synapse.types import UserID, RoomID diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 5768702192..0df1851b0e 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -278,6 +278,40 @@ class SyncHandler(BaseHandler): next_batch=now_token, )) + @defer.inlineCallbacks + def load_filtered_recents(self, room_id, sync_config, since_token, + now_token): + limited = True + recents = [] + filtering_factor = 2 + load_limit = max(sync_config.limit * filtering_factor, 100) + max_repeat = 3 # Only try a few times per room, otherwise + room_key = now_token.room_key + + while limited and len(recents) < sync_config.limit and max_repeat: + events, room_key = yield self.store.get_recent_events_for_room( + room_id, + limit=load_limit + 1, + from_token=since_token.room_key, + end_token=room_key, + ) + loaded_recents = sync_config.filter.filter_room_events(events) + loaded_recents.extend(recents) + recents = loaded_recents + if len(events) <= load_limit: + limited = False + max_repeat -= 1 + + if len(recents) > sync_config.limit: + recents = recents[-sync_config.limit:] + room_key = recents[0].internal_metadata.before + + prev_batch_token = now_token.copy_and_replace( + "room_key", room_key + ) + + defer.returnValue((recents, prev_batch_token, limited)) + @defer.inlineCallbacks def incremental_sync_with_gap_for_room(self, room_id, sync_config, since_token, now_token, @@ -288,28 +322,17 @@ class SyncHandler(BaseHandler): Returns: A Deferred RoomSyncResult """ + # TODO(mjark): Check if they have joined the room between # the previous sync and this one. - # TODO(mjark): Apply the event filter in sync_config taking care to get - # enough events to reach the limit # TODO(mjark): Check for redactions we might have missed. - recents, token = yield self.store.get_recent_events_for_room( - room_id, - limit=sync_config.limit + 1, - from_token=since_token.room_key, - end_token=now_token.room_key, + + recents, prev_batch_token, limited = self.load_filtered_recents( + room_id, sync_config, since_token, ) logging.debug("Recents %r", recents) - if len(recents) > sync_config.limit: - limited = True - recents = recents[1:] - else: - limited = False - - prev_batch_token = now_token.copy_and_replace("room_key", token[0]) - # TODO(mjark): This seems racy since this isn't being passed a # token to indicate what point in the stream this is current_state_events = yield self.state_handler.get_current_state( diff --git a/synapse/rest/client/v2_alpha/sync.py b/synapse/rest/client/v2_alpha/sync.py index c1277d2675..46ea50d118 100644 --- a/synapse/rest/client/v2_alpha/sync.py +++ b/synapse/rest/client/v2_alpha/sync.py @@ -116,7 +116,7 @@ class SyncRestServlet(RestServlet): user.localpart, filter_id ) except: - filter = Filter({}) + filter = Filter({}) # filter = filter.apply_overrides(http_request) #if filter.matches(event): # # stuff diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 2ea5e1a021..73504c8b52 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -181,15 +181,11 @@ class StreamStore(SQLBaseStore): get_prev_content=True ) - for event, row in zip(ret, rows): - stream = row["stream_ordering"] - topo = event.depth - internal = event.internal_metadata - internal.before = str(_StreamToken(topo, stream - 1)) - internal.after = str(_StreamToken(topo, stream)) + self._set_before_and_after(ret, rows) if rows: key = "s%d" % max([r["stream_ordering"] for r in rows]) + else: # Assume we didn't get anything because there was nothing to # get. @@ -267,6 +263,8 @@ class StreamStore(SQLBaseStore): get_prev_content=True ) + self._set_before_and_after(events, rows) + return events, next_token, return self.runInteraction("paginate_room_events", f) @@ -328,6 +326,8 @@ class StreamStore(SQLBaseStore): get_prev_content=True ) + self._set_before_and_after(events, rows) + return events, token return self.runInteraction( @@ -354,3 +354,12 @@ class StreamStore(SQLBaseStore): key = res[0]["m"] return "s%d" % (key,) + + @staticmethod + def _set_before_and_after(events, rows): + for event, row in zip(events, rows): + stream = row["stream_ordering"] + topo = event.depth + internal = event.internal_metadata + internal.before = str(_StreamToken(topo, stream - 1)) + internal.after = str(_StreamToken(topo, stream)) -- cgit 1.4.1 From 8498d348d818aa2d2cb9bb9bb2775103840f355d Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 30 Jan 2015 11:42:09 +0000 Subject: Fix token formatting --- synapse/handlers/sync.py | 6 +++--- synapse/storage/stream.py | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) (limited to 'synapse/storage/stream.py') diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index b83fcad655..3c68e2a9ec 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -164,7 +164,7 @@ class SyncHandler(BaseHandler): A Deferred RoomSyncResult. """ - recents, prev_batch_token, limited = self.load_filtered_recents( + recents, prev_batch_token, limited = yield self.load_filtered_recents( room_id, sync_config, now_token, ) @@ -288,7 +288,7 @@ class SyncHandler(BaseHandler): room_key = now_token.room_key while limited and len(recents) < sync_config.limit and max_repeat: - events, room_key = yield self.store.get_recent_events_for_room( + events, (room_key,_) = yield self.store.get_recent_events_for_room( room_id, limit=load_limit + 1, from_token=since_token.room_key if since_token else None, @@ -326,7 +326,7 @@ class SyncHandler(BaseHandler): # the previous sync and this one. # TODO(mjark): Check for redactions we might have missed. - recents, prev_batch_token, limited = self.load_filtered_recents( + recents, prev_batch_token, limited = yield self.load_filtered_recents( room_id, sync_config, now_token, since_token, ) diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 73504c8b52..3ccb6f8a61 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -316,9 +316,9 @@ class StreamStore(SQLBaseStore): toke = rows[0]["stream_ordering"] - 1 start_token = str(_StreamToken(topo, toke)) - token = (start_token, end_token) + token = (start_token, str(end_token)) else: - token = (end_token, end_token) + token = (str(end_token), str(end_token)) events = self._get_events_txn( txn, -- cgit 1.4.1