Bust _membership_stream_cache cache when current state changes
#17732
Conversation
| def all_entities_changed(self, stream_pos: int) -> None: | ||
| """ | ||
| Mark all entities as changed. This is useful when the cache is invalidated and | ||
| there may be some potential change for all of the entities. | ||
| """ |
There was a problem hiding this comment.
Re: all_entities_changed(stream_pos): Does this concept make sense?
I don't think it makes sense to drop all of the keys as we're essentially not sure if something has changed so we have to update them to say "something might have changed but we don't know for sure". I think this is the way and is just "unfortunate for the membership caches"
| @@ -219,6 +219,8 @@ def process_replication_rows( | |||
| room_id = row.keys[0] | |||
| members_changed = set(row.keys[1:]) | |||
| self._invalidate_state_caches(room_id, members_changed) | |||
| for user_id in members_changed: | |||
| self._membership_stream_cache.entity_has_changed(user_id, token) # type: ignore[attr-defined] | |||
There was a problem hiding this comment.
Kinda weird to just stick this here (same with the others in process_replication_rows). Better way to organize this?
| stream_id: This is expected to be the minimum `stream_ordering` for the | ||
| batch of events that we are persisting; which means we do not end up in a | ||
| situation where workers see events before the `current_state_delta` updates. | ||
| FIXME: However, this function also gets called with next upcoming | ||
| `stream_ordering` when we re-sync the state of a partial stated room (see | ||
| `update_current_state(...)`) which may be "correct" but it would be good to | ||
| nail down what exactly is the expected value here. |
There was a problem hiding this comment.
Previous conversation: #17512 (comment)
I decided to define it in some way given we're using it for cache busting below and was curious if it is actually correct. Still not confident whether it's perfect for cache busting but might be good enough.
| for user_id in members_to_cache_bust: | ||
| txn.call_after( | ||
| self.store._membership_stream_cache.entity_has_changed, | ||
| user_id, | ||
| stream_id, | ||
| ) |
There was a problem hiding this comment.
This matches what we do for _curr_state_delta_stream_cache just above this
| for user_id in members_to_cache_bust: | ||
| txn.call_after( | ||
| self.store._membership_stream_cache.entity_has_changed, | ||
| user_id, | ||
| stream_id, | ||
| ) |
There was a problem hiding this comment.
This is the actual call that busts the the membership cache for the tests. I assume that is because this is what busts in monolith mode vs the other calls I've added are more for workers over replication
| self._curr_state_delta_stream_cache.entity_has_changed( # type: ignore[attr-defined] | ||
| room_id, token | ||
| ) | ||
| for user_id in members_changed: | ||
| self._membership_stream_cache.entity_has_changed(user_id, token) # type: ignore[attr-defined] |
There was a problem hiding this comment.
Wherever we are busting _curr_state_delta_stream_cache, we should also be busting _membership_stream_cache (at-least in the general area, expand the hidden diff to find if not visible)
We've forgotten to bust _curr_state_delta_stream_cache in various places which is why it's added and sometimes not.
Bust
_membership_stream_cachecache when current state changes. This is particularly a problem in a state reset scenario where the membership might change without a corresponding event.This is a general Synapse thing so by it's nature it helps out Sliding Sync.
Fix #17368
Match when are busting
_curr_state_delta_stream_cacheDev notes
Pull Request Checklist
EventStoretoEventWorkerStore.".code blocks.(run the linters)