Cache get_room_changes_for_user

synapse/storage/__init__.py

@@ -131,6 +131,10 @@ def __init__(self, db_conn, hs):
             prefilled_cache=event_cache_prefill,
         )
 
+        self._membership_stream_cache = StreamChangeCache(
+            "MembershipStreamChangeCache", events_max,
+        )
+
         account_max = self._account_data_id_gen.get_max_token(None)
         self._account_data_stream_cache = StreamChangeCache(
             "AccountDataAndTagsChangeCache", account_max,

synapse/storage/roommember.py

@@ -58,6 +58,10 @@ def _store_room_members_txn(self, txn, events):
             txn.call_after(self.get_rooms_for_user.invalidate, (event.state_key,))
             txn.call_after(self.get_joined_hosts_for_room.invalidate, (event.room_id,))
             txn.call_after(self.get_users_in_room.invalidate, (event.room_id,))
+            txn.call_after(
+                self._membership_stream_cache.entity_has_changed,
+                event.state_key, event.internal_metadata.stream_ordering
+            )
 
     def get_room_member(self, user_id, room_id):
         """Retrieve the current state of a room member.

synapse/storage/stream.py

@@ -252,6 +252,13 @@ def get_room_changes_for_user(self, user_id, from_key, to_key):
         if from_key == to_key:
             return defer.succeed([])
 
+        if from_id:
+            has_changed = self._membership_stream_cache.has_entity_changed(
+                user_id, int(from_id)
+            )
+            if not has_changed:
+                return defer.succeed([])
+
         def f(txn):
             if from_id is not None:
                 sql = (