dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-29 07:48:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix lazy_loaded_members_cache

Browse Source 
Do a `sha256(other user + field name) -> sha256(value)` to ensure we miss the cache on values changing back/forth.
This commit is contained in:
Jason Robinson
2026-07-21 15:28:09 +03:00
parent 12d4f2d089
commit b0666bed81
2 changed files with 169 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/sync.py
+26 -24
View File
@@ -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,
tests/handlers/test_sync.py
+143
View File
@@ -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."""