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.
|
