From b0666bed81ccb0f886fd648ceb6df7f180adb8d0 Mon Sep 17 00:00:00 2001 From: Jason Robinson Date: Tue, 21 Jul 2026 15:28:09 +0300 Subject: [PATCH] Fix `lazy_loaded_members_cache` Do a `sha256(other user + field name) -> sha256(value)` to ensure we miss the cache on values changing back/forth. --- synapse/handlers/sync.py | 50 +++++++------ tests/handlers/test_sync.py | 143 ++++++++++++++++++++++++++++++++++++ 2 files changed, 169 insertions(+), 24 deletions(-) diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 9dc6167b27..f0ae4efc3c 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -20,6 +20,7 @@ # import hashlib import itertools +import json import logging from typing import ( TYPE_CHECKING, @@ -348,10 +349,10 @@ class SyncHandler: ) # ExpiringCache((User, Device)) # -> LruCache( - # sha256(Other User ID + Field Name + Field value) -> bool + # sha256(Other User ID + Field Name) -> sha256(Field value) # ) self.lazy_loaded_profile_fields_cache: ExpiringCache[ - tuple[str, str | None], LruCache[str, bool] + tuple[str, str | None], LruCache[str, str] ] = ExpiringCache( cache_name="lazy_loaded_profile_fields_cache", server_name=self.server_name, @@ -361,11 +362,10 @@ class SyncHandler: expiry_ms=LAZY_LOADED_PROFILE_FIELDS_CACHE_MAX_AGE, ) """This cache contains fields and values we have sent to clients as profile - updates, for a particular user + device combo. The cache entry is a combination - of the user + field name + value, all hashed into sha256, an existing value - indicating the field has recently been sent. The boolean value does not hold - other significance. A missing cache entry means "we have not sent this user + - field name + value combo to the syncing user". + updates, for a particular user + device combo. The cache entry is a sha256 + of the user + field name, with the value being a sha256 of the field value. + If the field value changes for a particular user, the hash will change + and the cache will be missed. We don't manually remove entries from this cache, though it may be ignored in cases where the sync must send the field down to the client. @@ -1079,18 +1079,17 @@ class SyncHandler: def get_lazy_loaded_profile_fields_cache( self, cache_key: tuple[str, str | None] - ) -> LruCache[str, bool]: + ) -> LruCache[str, str]: """This cache contains fields and values we have sent to clients as profile - updates, for a particular user + device combo. The cache entry is a combination - of the user + field name + value, all hashed into sha256, an existing value - indicating the field has recently been sent. The boolean value does not hold - other significance. A missing cache entry means "we have not sent this user + - field name + value combo to the syncing user". + updates, for a particular user + device combo. The cache entry is a sha256 + of the user + field name, with the value being a sha256 of the field value. + If the field value changes for a particular user, the hash will change + and the cache will be missed. We don't manually remove entries from this cache, though it may be ignored in cases where the sync must send the field down to the client. """ - cache: LruCache[str, bool] | None = self.lazy_loaded_profile_fields_cache.get( + cache: LruCache[str, str] | None = self.lazy_loaded_profile_fields_cache.get( cache_key ) if cache is None: @@ -2390,6 +2389,10 @@ class SyncHandler: # Hopefully clients can just filter these out. profile_data_by_user = await self.store.get_profile_data_for_users(users) + # Note, we've already collected field updates above via `updates`, + # outside of events in the timeline when lazy loading. When lazy loading, + # we're already always sending the fields that have changed, regardless + # of the lazy loading cache. for other_user_id in users: profile_data = profile_data_by_user.get(other_user_id) if profile_data is None: @@ -2412,17 +2415,16 @@ class SyncHandler: ) cache = self.get_lazy_loaded_profile_fields_cache(cache_key) # Only send this users field if we haven't recently sent it. - # Our cache value to check against is a sha256 of a string of - # user ID + field name + value, which ensures if the value - # changes, we'll miss the cache, thus sending the field update - # to the syncing user. - import json - + # Our cache contains previously set values as pairs of + # sha256(other_used_id + field_name) -> sha256(value), + # which ensures if the value changes, we'll miss the cache, + # thus sending the field update to the syncing user. cache_value = hashlib.sha256( + f"{other_user_id}-{field_name}".encode("utf8"), + ).hexdigest() + value_hash = hashlib.sha256( json.dumps( [ - other_user_id, - field_name, profile_data.get(field_name), ], sort_keys=True, @@ -2430,13 +2432,13 @@ class SyncHandler: ensure_ascii=False, ).encode("utf8") ).hexdigest() - if cache.get(cache_value) is None: + if cache.get(cache_value) != value_hash: per_user_updates[field_name] = profile_data.get(field_name) # Update our cache to indicate this user/field combo # has been recently sent. cache.set( cache_value, - True, + value_hash, ) else: # Include only the diff, unless the user recently joined, diff --git a/tests/handlers/test_sync.py b/tests/handlers/test_sync.py index ef26d1a515..8e09c6cdbf 100644 --- a/tests/handlers/test_sync.py +++ b/tests/handlers/test_sync.py @@ -2664,6 +2664,9 @@ class SyncProfileUpdatesTestCase(tests.unittest.HomeserverTestCase): new_value=cast(JsonValue | dict[str, JsonValue], value), ) ) + # Also send an event + self.helper.send(self.joined_room, "Foo", tok=self.other_tok) + # Sync incremental_result = self.get_success( self.sync_handler.wait_for_sync_for_user( requester, @@ -2703,6 +2706,9 @@ class SyncProfileUpdatesTestCase(tests.unittest.HomeserverTestCase): new_value=cast(JsonValue | dict[str, JsonValue], new_value), ) ) + # Also send an event + self.helper.send(self.joined_room, "Foo", tok=self.other_tok) + # Sync incremental_result = self.get_success( self.sync_handler.wait_for_sync_for_user( requester, @@ -2732,6 +2738,143 @@ class SyncProfileUpdatesTestCase(tests.unittest.HomeserverTestCase): new_value, ) + @override_config({"include_profile_updates_in_sync": True}) + def test_lazy_loading_cache_and_multiple_updates_to_the_same_field( + self, + ) -> None: + """Test that with MSC4429 enabled the incremental lazy sync response + includes an update to a field, even when the value changes back to a + value set and cached previously. + """ + requester = create_requester(self.user) + filter_json = { + "org.matrix.msc4429.profile_fields": {"ids": ["field"]}, + "room": { + "state": { + "lazy_load_members": True, + }, + }, + } + initial_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json=filter_json, + ), + ), + request_key=generate_request_key(), + ) + ) + self.assertFalse( + "@other_user:test" in initial_result.profile_updates, + ) + + # Update the field + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="value", + ) + ) + # Also send an event + self.helper.send(self.joined_room, "Foo", tok=self.other_tok) + # Sync + incremental_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + since_token=initial_result.next_batch, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json=filter_json, + ), + ), + request_key=generate_request_key(), + ) + ) + # We should have the field change in our sync response. + # It will also be added to the lazy loading cache, so the same field value + # isn't sent again immediately. + assert incremental_result.profile_updates["@other_user:test"] is not None + self.assertEqual( + incremental_result.profile_updates["@other_user:test"]["field"], + "value", + ) + + # Update the field again, busting our cache + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="new value", + ) + ) + # Also send an event + self.helper.send(self.joined_room, "Foo", tok=self.other_tok) + # Sync + incremental_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + since_token=incremental_result.next_batch, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json=filter_json, + ), + ), + request_key=generate_request_key(), + ) + ) + # Even though the field was added to the lazy loading members cache, + # it should come through as an update, as the field value changed. + assert incremental_result.profile_updates["@other_user:test"] is not None + self.assertEqual( + incremental_result.profile_updates["@other_user:test"]["field"], + "new value", + ) + + # Update the field again, but to the previous value + self.get_success( + self.profile_handler.set_field( + target_user=UserID.from_string(self.other_user), + requester=create_requester(self.other_user), + field_name="field", + new_value="value", + ) + ) + # Also send an event + self.helper.send(self.joined_room, "Foo", tok=self.other_tok) + # Sync + incremental_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + since_token=incremental_result.next_batch, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json=filter_json, + ), + ), + request_key=generate_request_key(), + ) + ) + # Even though we've quite recently sent down this value, we should still + # see it again as it is a change + assert incremental_result.profile_updates["@other_user:test"] is not None + self.assertEqual( + incremental_result.profile_updates["@other_user:test"]["field"], + "value", + ) + class SyncStateAfterTestCase(tests.unittest.HomeserverTestCase): """Tests Sync Handler state behavior when using `use_state_after."""