From de4f798f01e2ad478639b103d293b86079aaf0bc Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 15 Sep 2016 17:34:59 +0100 Subject: [PATCH 01/11] Handling expiring stream extrems correctly. --- synapse/storage/__init__.py | 2 ++ synapse/storage/event_federation.py | 23 ++++++++++++++++++++--- 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 0099a3f5bb..9996f195a0 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -222,6 +222,8 @@ class DataStore(RoomMemberStore, RoomStore, self._find_stream_orderings_for_times, 60 * 60 * 1000 ) + self._stream_order_on_start = self.get_room_max_stream_ordering() + super(DataStore, self).__init__(hs) def take_presence_startup_info(self): diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 59b4cf1e53..765b5a5bcb 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -348,7 +348,14 @@ class EventFederationStore(SQLBaseStore): # We want to make the cache more effective, so we clamp to the last # change before the given ordering. last_change = self._events_stream_cache.get_max_pos_of_last_change(room_id) - stream_ordering = min(last_change, stream_ordering) + + # We don't always have a full stream_to_exterm_id table, e.g. after + # the upgrade that introduced it, so we make sure we never ask for a + # try and pin to a stream_ordering from before a restart + last_change = max(self._stream_order_on_start, last_change) + + if last_change > self.stream_ordering_month_ago: + stream_ordering = min(last_change, stream_ordering) return self._get_forward_extremeties_for_room(room_id, stream_ordering) @@ -386,9 +393,19 @@ class EventFederationStore(SQLBaseStore): def _delete_old_forward_extrem_cache(self): def _delete_old_forward_extrem_cache_txn(txn): + sql = (""" + DELETE FROM stream_ordering_to_exterm + WHERE + ( + SELECT max(stream_ordering) AS stream_ordering + FROM stream_ordering_to_exterm + WHERE room_id = stream_ordering_to_exterm.room_id + ) > ? + AND stream_ordering < ? + """) txn.execute( - "DELETE FROM stream_ordering_to_exterm WHERE stream_ordering < ?", - (self.stream_ordering_month_ago,) + sql, + (self.stream_ordering_month_ago, self.stream_ordering_month_ago,) ) return self.runInteraction( "_delete_old_forward_extrem_cache", From c33b25fd8d7a91652dc7f887178a47343c6c9ed2 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 15 Sep 2016 17:35:20 +0100 Subject: [PATCH 02/11] Change the way we calculate new_limit in /publicRooms and add POST API --- synapse/handlers/room_list.py | 48 +++++++++++++++++++++++----------- synapse/rest/client/v1/room.py | 29 ++++++++++++++++++++ 2 files changed, 62 insertions(+), 15 deletions(-) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 28bc35f8a3..c5ee3b3cd9 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -39,7 +39,8 @@ class RoomListHandler(BaseHandler): super(RoomListHandler, self).__init__(hs) self.response_cache = ResponseCache(hs) - def get_local_public_room_list(self, limit=None, since_token=None): + def get_local_public_room_list(self, limit=None, since_token=None, + search_filter=None): result = self.response_cache.get((limit, since_token)) if not result: result = self.response_cache.set( @@ -49,7 +50,8 @@ class RoomListHandler(BaseHandler): return result @defer.inlineCallbacks - def _get_public_room_list(self, limit=None, since_token=None): + def _get_public_room_list(self, limit=None, since_token=None, + search_filter=None): if since_token and since_token != "END": since_token = RoomListNextBatch.from_token(since_token) else: @@ -115,22 +117,18 @@ class RoomListHandler(BaseHandler): sorted_rooms = sorted_rooms[:since_token.current_limit] sorted_rooms.reverse() - new_limit = None - if limit: - if sorted_rooms[limit:]: - new_limit = limit - if since_token: - if since_token.direction_is_forward: - new_limit += since_token.current_limit - else: - new_limit = since_token.current_limit - new_limit - new_limit = max(0, new_limit) - sorted_rooms = sorted_rooms[:limit] + rooms_to_scan = sorted_rooms + if limit and not search_filter: + rooms_to_scan = sorted_rooms[:limit] chunk = [] @defer.inlineCallbacks def handle_room(room_id): + if limit and len(chunk) > limit: + # We've already got enough, so lets just drop it. + return + num_joined_users = rooms_to_num_joined[room_id] if num_joined_users == 0: return @@ -212,10 +210,29 @@ class RoomListHandler(BaseHandler): chunk.append(result) - yield concurrently_execute(handle_room, sorted_rooms, 10) + yield concurrently_execute(handle_room, rooms_to_scan, 10) chunk.sort(key=lambda e: (-e["num_joined_members"], e["room_id"])) + new_limit = None + if chunk: + addition = 1 + if since_token: + addition += since_token.current_limit + + if not since_token or since_token.direction_is_forward: + last_room_id = chunk[-1]["room_id"] + else: + last_room_id = chunk[0]["room_id"] + addition *= -1 + + try: + new_limit = sorted_rooms.index(last_room_id) + addition + if new_limit >= len(sorted_rooms): + new_limit = None + except ValueError: + pass + results = { "chunk": chunk, } @@ -253,7 +270,8 @@ class RoomListHandler(BaseHandler): defer.returnValue(results) @defer.inlineCallbacks - def get_remote_public_room_list(self, server_name, limit=None, since_token=None): + def get_remote_public_room_list(self, server_name, limit=None, since_token=None, + search_filter=None): res = yield self.hs.get_replication_layer().get_public_rooms( server_name, limit=limit, since_token=since_token, ) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index db0cd4380a..be14ae1061 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -337,6 +337,35 @@ class PublicRoomListRestServlet(ClientV1RestServlet): defer.returnValue((200, data)) + @defer.inlineCallbacks + def on_POST(self, request): + # FIXME + # yield self.auth.get_user_by_req(request) + + server = parse_string(request, "server", default=None) + content = parse_json_object_from_request(request) + + limit = int(content.get("limit", 100)) + since_token = content.get("since", None) + search_filter = content.get("filter", None) + + handler = self.hs.get_room_list_handler() + if server: + data = yield handler.get_remote_public_room_list( + server, + limit=limit, + since_token=since_token, + search_filter=search_filter, + ) + else: + data = yield handler.get_local_public_room_list( + limit=limit, + since_token=since_token, + search_filter=search_filter, + ) + + defer.returnValue((200, data)) + # TODO: Needs unit testing class RoomMemberListRestServlet(ClientV1RestServlet): From 098db4aa521704e8c98c369f0c6d11ca9e02d899 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 15 Sep 2016 17:50:16 +0100 Subject: [PATCH 03/11] Add very basic filter API to /publicRooms --- synapse/handlers/room_list.py | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index c5ee3b3cd9..3a75176ef4 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -45,7 +45,7 @@ class RoomListHandler(BaseHandler): if not result: result = self.response_cache.set( (limit, since_token), - self._get_public_room_list(limit, since_token) + self._get_public_room_list(limit, since_token, search_filter) ) return result @@ -119,13 +119,13 @@ class RoomListHandler(BaseHandler): rooms_to_scan = sorted_rooms if limit and not search_filter: - rooms_to_scan = sorted_rooms[:limit] + rooms_to_scan = sorted_rooms[:limit + 1] chunk = [] @defer.inlineCallbacks def handle_room(room_id): - if limit and len(chunk) > limit: + if limit and len(chunk) > limit + 1: # We've already got enough, so lets just drop it. return @@ -208,14 +208,27 @@ class RoomListHandler(BaseHandler): if avatar_url: result["avatar_url"] = avatar_url - chunk.append(result) + logger.info("search_filter: %r", search_filter) + if search_filter and search_filter.get("generic_search_term", None): + generic_search_term = search_filter["generic_search_term"] + if generic_search_term in result.get("name", ""): + chunk.append(result) + elif generic_search_term in result.get("topic", ""): + chunk.append(result) + elif generic_search_term in result.get("canonical_alias", ""): + chunk.append(result) + else: + chunk.append(result) yield concurrently_execute(handle_room, rooms_to_scan, 10) chunk.sort(key=lambda e: (-e["num_joined_members"], e["room_id"])) new_limit = None - if chunk: + if chunk and (not limit or len(chunk) > limit): + if limit: + chunk = chunk[:limit] + addition = 1 if since_token: addition += since_token.current_limit From 418bcd43093b10fc93487376e23973278850b502 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 08:37:39 +0100 Subject: [PATCH 04/11] Add new storage function to slave store --- synapse/replication/slave/storage/events.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/replication/slave/storage/events.py b/synapse/replication/slave/storage/events.py index cc32c66792..0c26e96e98 100644 --- a/synapse/replication/slave/storage/events.py +++ b/synapse/replication/slave/storage/events.py @@ -62,6 +62,7 @@ class SlavedEventStore(BaseSlavedStore): ) self.stream_ordering_month_ago = 0 + self._stream_order_on_start = self.get_room_max_stream_ordering() # Cached functions can't be accessed through a class instance so we need # to reach inside the __dict__ to extract them. From 74d4cdee250e4134dd14e4edfb8f5833e28c02b0 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 09:05:11 +0100 Subject: [PATCH 05/11] Don't cache searched in /publicRooms --- synapse/handlers/room_list.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 3a75176ef4..37213f4bd8 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -41,11 +41,15 @@ class RoomListHandler(BaseHandler): def get_local_public_room_list(self, limit=None, since_token=None, search_filter=None): + if search_filter: + # We explicitly don't bother caching searches. + return self._get_public_room_list(limit, since_token, search_filter) + result = self.response_cache.get((limit, since_token)) if not result: result = self.response_cache.set( (limit, since_token), - self._get_public_room_list(limit, since_token, search_filter) + self._get_public_room_list(limit, since_token) ) return result From e58a9d781c7808b66f6eda221c9ce91ccd3cd8d6 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 10:19:32 +0100 Subject: [PATCH 06/11] Filter remote rooms lists locally --- synapse/handlers/room_list.py | 34 ++++++++++++++++++++--------- synapse/storage/event_federation.py | 2 +- 2 files changed, 25 insertions(+), 11 deletions(-) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 37213f4bd8..9383f2486c 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -212,16 +212,7 @@ class RoomListHandler(BaseHandler): if avatar_url: result["avatar_url"] = avatar_url - logger.info("search_filter: %r", search_filter) - if search_filter and search_filter.get("generic_search_term", None): - generic_search_term = search_filter["generic_search_term"] - if generic_search_term in result.get("name", ""): - chunk.append(result) - elif generic_search_term in result.get("topic", ""): - chunk.append(result) - elif generic_search_term in result.get("canonical_alias", ""): - chunk.append(result) - else: + if _matches_room_entry(result, search_filter): chunk.append(result) yield concurrently_execute(handle_room, rooms_to_scan, 10) @@ -291,8 +282,16 @@ class RoomListHandler(BaseHandler): search_filter=None): res = yield self.hs.get_replication_layer().get_public_rooms( server_name, limit=limit, since_token=since_token, + search_filter=search_filter, ) + if search_filter: + res["chunk"] = [ + entry + for entry in dict(res.get("chunk", [])) + if _matches_room_entry(entry, search_filter) + ] + defer.returnValue(res) @@ -329,3 +328,18 @@ class RoomListNextBatch(namedtuple("RoomListNextBatch", ( return self._replace( **kwds ) + + +def _matches_room_entry(room_entry, search_filter): + if search_filter and search_filter.get("generic_search_term", None): + generic_search_term = search_filter["generic_search_term"] + if generic_search_term in room_entry.get("name", ""): + return True + elif generic_search_term in room_entry.get("topic", ""): + return True + elif generic_search_term in room_entry.get("canonical_alias", ""): + return True + else: + return True + + return False diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 765b5a5bcb..53289f556b 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -376,7 +376,7 @@ class EventFederationStore(SQLBaseStore): INNER JOIN ( SELECT room_id, MAX(stream_ordering) AS stream_ordering FROM stream_ordering_to_exterm - WHERE stream_ordering < ? GROUP BY room_id + WHERE stream_ordering <= ? GROUP BY room_id ) AS rms USING (room_id, stream_ordering) WHERE room_id = ? """) From 23b6701a2869d50fefbc949fbb449de07636b5b8 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 10:24:15 +0100 Subject: [PATCH 07/11] Support filtering remote room lists --- synapse/federation/federation_client.py | 7 +++++-- synapse/federation/transport/client.py | 5 ++++- synapse/handlers/room_list.py | 12 +++++++++--- 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index f0a684fc13..06d0320b1a 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -718,11 +718,14 @@ class FederationClient(FederationBase): raise RuntimeError("Failed to send to any server.") - def get_public_rooms(self, destination, limit=None, since_token=None): + def get_public_rooms(self, destination, limit=None, since_token=None, + search_filter=None): if destination == self.server_name: return - return self.transport_layer.get_public_rooms(destination, limit, since_token) + return self.transport_layer.get_public_rooms( + destination, limit, since_token, search_filter + ) @defer.inlineCallbacks def query_auth(self, destination, room_id, event_id, local_auth): diff --git a/synapse/federation/transport/client.py b/synapse/federation/transport/client.py index f508b70f11..db45c7826c 100644 --- a/synapse/federation/transport/client.py +++ b/synapse/federation/transport/client.py @@ -248,7 +248,8 @@ class TransportLayerClient(object): @defer.inlineCallbacks @log_function - def get_public_rooms(self, remote_server, limit, since_token): + def get_public_rooms(self, remote_server, limit, since_token, + search_filter=None): path = PREFIX + "/publicRooms" args = {} @@ -257,6 +258,8 @@ class TransportLayerClient(object): if since_token: args["since"] = [since_token] + # TODO(erikj): Actually send the search_filter across federation. + response = yield self.client.get_json( destination=remote_server, path=path, diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 9383f2486c..09189edb65 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -280,17 +280,23 @@ class RoomListHandler(BaseHandler): @defer.inlineCallbacks def get_remote_public_room_list(self, server_name, limit=None, since_token=None, search_filter=None): + if search_filter: + # We currently don't support searching across federation, so we have + # to do it manually without pagination + limit = None + since_token = None + res = yield self.hs.get_replication_layer().get_public_rooms( server_name, limit=limit, since_token=since_token, search_filter=search_filter, ) if search_filter: - res["chunk"] = [ + res = {"chunk": [ entry - for entry in dict(res.get("chunk", [])) + for entry in list(res.get("chunk", [])) if _matches_room_entry(entry, search_filter) - ] + ]} defer.returnValue(res) From d84319ae1048837c766eb2e56f91eb521aeed4a6 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 10:31:59 +0100 Subject: [PATCH 08/11] Add remote reoom cache --- synapse/handlers/room_list.py | 25 +++++++++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index 09189edb65..ec0a293a37 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -38,6 +38,7 @@ class RoomListHandler(BaseHandler): def __init__(self, hs): super(RoomListHandler, self).__init__(hs) self.response_cache = ResponseCache(hs) + self.remote_response_cache = ResponseCache(hs, timeout_ms=30 * 1000) def get_local_public_room_list(self, limit=None, since_token=None, search_filter=None): @@ -286,9 +287,8 @@ class RoomListHandler(BaseHandler): limit = None since_token = None - res = yield self.hs.get_replication_layer().get_public_rooms( + res = yield self._get_remote_list_cached( server_name, limit=limit, since_token=since_token, - search_filter=search_filter, ) if search_filter: @@ -300,6 +300,27 @@ class RoomListHandler(BaseHandler): defer.returnValue(res) + def _get_remote_list_cached(self, server_name, limit=None, since_token=None, + search_filter=None): + repl_layer = self.hs.get_replication_layer() + if search_filter: + # We can't cache when asking for search + return repl_layer.get_public_rooms( + server_name, limit=limit, since_token=since_token, + search_filter=search_filter, + ) + + result = self.remote_response_cache.get((server_name, limit, since_token)) + if not result: + result = self.remote_response_cache.set( + (server_name, limit, since_token), + repl_layer.get_public_rooms( + server_name, limit=limit, since_token=since_token, + search_filter=search_filter, + ) + ) + return result + class RoomListNextBatch(namedtuple("RoomListNextBatch", ( "stream_ordering", # stream_ordering of the first public room list From b7b62bf9eac3f1f92a4913cfdf1f568752dcf8c0 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 11:00:29 +0100 Subject: [PATCH 09/11] Comment --- synapse/handlers/room_list.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/synapse/handlers/room_list.py b/synapse/handlers/room_list.py index ec0a293a37..f15987b265 100644 --- a/synapse/handlers/room_list.py +++ b/synapse/handlers/room_list.py @@ -220,6 +220,8 @@ class RoomListHandler(BaseHandler): chunk.sort(key=lambda e: (-e["num_joined_members"], e["room_id"])) + # Work out the new limit of the batch for pagination, or None if we + # know there are no more results that would be returned. new_limit = None if chunk and (not limit or len(chunk) > limit): if limit: From 2e67cabd7f9b8f01ba776cef5737491bc794527e Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 11:32:51 +0100 Subject: [PATCH 10/11] Make POST /publicRooms require auth --- synapse/rest/client/v1/room.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index be14ae1061..5584bfbfc0 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -339,8 +339,7 @@ class PublicRoomListRestServlet(ClientV1RestServlet): @defer.inlineCallbacks def on_POST(self, request): - # FIXME - # yield self.auth.get_user_by_req(request) + yield self.auth.get_user_by_req(request) server = parse_string(request, "server", default=None) content = parse_json_object_from_request(request) From a68807d4260778cff706fc37ef1badd4fd0d8bf2 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Sep 2016 11:34:01 +0100 Subject: [PATCH 11/11] Comment --- synapse/storage/event_federation.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 53289f556b..3d62451de9 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -393,6 +393,8 @@ class EventFederationStore(SQLBaseStore): def _delete_old_forward_extrem_cache(self): def _delete_old_forward_extrem_cache_txn(txn): + # Delete entries older than a month, while making sure we don't delete + # the only entries for a room. sql = (""" DELETE FROM stream_ordering_to_exterm WHERE