diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 921215469f..4c5b935012 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -258,19 +258,39 @@ class SyncHandler(BaseHandler): user_id = sync_config.user.to_string() + room_to_last_ts = {} + + @defer.inlineCallbacks + def _get_last_ts(event): + room_id = event.room_id + if event.membership == Membership.JOIN: + ts = yield self.store.get_last_ts_for_room( + room_id, now_token.room_key + ) + room_to_last_ts[room_id] = ts if ts else 0 + + logger.info("room_to_last_ts: %r", room_to_last_ts) + yield concurrently_execute(_get_last_ts, room_list, 10) + + joined_rooms_list = frozenset([ + room_id for room_id, _f in + sorted(room_to_last_ts.items(), key=lambda item: -item[1]) + ][:20]) + @defer.inlineCallbacks def _generate_room_entry(event): if event.membership == Membership.JOIN: - room_result = yield self.full_state_sync_for_joined_room( - room_id=event.room_id, - sync_config=sync_config, - now_token=now_token, - timeline_since_token=timeline_since_token, - ephemeral_by_room=ephemeral_by_room, - tags_by_room=tags_by_room, - account_data_by_room=account_data_by_room, - ) - joined.append(room_result) + if event.room_id in joined_rooms_list: + room_result = yield self.full_state_sync_for_joined_room( + room_id=event.room_id, + sync_config=sync_config, + now_token=now_token, + timeline_since_token=timeline_since_token, + ephemeral_by_room=ephemeral_by_room, + tags_by_room=tags_by_room, + account_data_by_room=account_data_by_room, + ) + joined.append(room_result) elif event.membership == Membership.INVITE: if event.sender in ignored_users: return diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 95b12559a6..60a8384fae 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -529,6 +529,26 @@ class StreamStore(SQLBaseStore): int(stream), ) + def get_last_ts_for_room(self, room_id, token): + stream_ordering = RoomStreamToken.parse_stream_token(token).stream + + sql = ( + "SELECT origin_server_ts FROM events" + " WHERE room_id = ? AND stream_ordering <= ?" + " ORDER BY topological_ordering DESC, stream_ordering DESC" + " LIMIT 1" + ) + + def f(txn): + txn.execute(sql, (room_id, stream_ordering)) + rows = txn.fetchall() + if rows: + return rows[0][0] + else: + return None + + return self.runInteraction("get_last_ts_for_room", f) + @defer.inlineCallbacks def get_events_around(self, room_id, event_id, before_limit, after_limit): """Retrieve events and pagination tokens around a given event in a