Add stream change cache

4 files changed, 34 insertions, 1 deletions

diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 11655bf60f..3c88ba9860 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -223,6 +223,18 @@ class DataStore(RoomMemberStore, RoomStore,
             "DeviceListFederationStreamChangeCache", device_list_max,
         )
 
+        curr_state_delta_prefill, min_curr_state_delta_id = self._get_cache_dict(
+            db_conn, "current_state_delta_stream",
+            entity_column="room_id",
+            stream_column="stream_id",
+            max_value=events_max,  # As we share the stream id with events token
+            limit=1000,
+        )
+        self._curr_state_delta_stream_cache = StreamChangeCache(
+            "_curr_state_delta_stream_cache", min_curr_state_delta_id,
+            prefilled_cache=curr_state_delta_prefill,
+        )
+
         cur = LoggingTransaction(
             db_conn.cursor(),
             name="_find_stream_orderings_for_times_txn",
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index dfb57f9d12..77861488d2 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -755,6 +755,10 @@ class EventsStore(SQLBaseStore):
                     ]
                 )
 
+                self._curr_state_delta_stream_cache.enttity_has_changed(
+                    room_id, max_stream_order,
+                )
+
                 # Invalidate the various caches
 
                 # Figure out the changes of membership to invalidate the
diff --git a/synapse/storage/user_directory.py b/synapse/storage/user_directory.py
index b1957cb873..15b8ea0460 100644
--- a/synapse/storage/user_directory.py
+++ b/synapse/storage/user_directory.py
@@ -204,7 +204,9 @@ class UserDirectoryStore(SQLBaseStore):
         )
 
     def get_current_state_deltas(self, prev_stream_id):
-        # TODO: Add stream change cache
+        if not self._curr_state_delta_stream_cache.has_any_entity_changed(prev_stream_id):
+            return []
+
         # TODO: Add limit
         sql = """
             SELECT stream_id, room_id, type, state_key, event_id, prev_event_id
diff --git a/synapse/util/caches/stream_change_cache.py b/synapse/util/caches/stream_change_cache.py
index 70fe00ce0b..c498aee46c 100644
--- a/synapse/util/caches/stream_change_cache.py
+++ b/synapse/util/caches/stream_change_cache.py
@@ -89,6 +89,21 @@ class StreamChangeCache(object):
 
         return result
 
+    def has_any_entity_changed(self, stream_pos):
+        """Returns if any entity has changed
+        """
+        assert type(stream_pos) is int
+
+        if stream_pos >= self._earliest_known_stream_pos:
+            self.metrics.inc_hits()
+            if stream_pos >= max(self._cache):
+                return False
+            else:
+                return True
+        else:
+            self.metrics.inc_misses()
+            return True
+
     def get_all_entities_changed(self, stream_pos):
         """Returns all entites that have had new things since the given
         position. If the position is too old it will return None.