diff --git a/synapse/config/workers.py b/synapse/config/workers.py index 3bfe300d49..171109da6b 100644 --- a/synapse/config/workers.py +++ b/synapse/config/workers.py @@ -145,7 +145,6 @@ class WriterLocations: device_lists: The instances that write to the device list stream. thread_subscriptions: The instances that write to the thread subscriptions stream. - profile_updates: The instances that write to the profile updates stream. quarantined_media_changes: The instances that write to the quarantined media changes stream. """ diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index 1cbe7fe1a1..1721c03ffd 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -2416,10 +2416,15 @@ class SyncHandler: # 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 + cache_value = hashlib.sha256( - f"{other_user_id}-{field_name}-{profile_data.get(field_name)}".encode( - "utf8" - ) + json.dumps( + [other_user_id, field_name, profile_data.get(field_name)], + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ).encode("utf8") ).hexdigest() if cache.get(cache_value) is None: per_user_updates[field_name] = profile_data.get(field_name) diff --git a/synapse/storage/databases/main/profile.py b/synapse/storage/databases/main/profile.py index 111d6ecc36..0aae4a7783 100644 --- a/synapse/storage/databases/main/profile.py +++ b/synapse/storage/databases/main/profile.py @@ -884,10 +884,12 @@ class ProfileWorkerStore(SQLBaseStore): values=values, ) - # Add per user tracking rows pointing to the latest stream ID + # Add per user tracking rows for each generated stream ID inserted_ts = self.clock.time_msec() per_user_values = [ - (stream_ids[-1], user_id, inserted_ts) for user_id in target_users + (stream_id, user_id, inserted_ts) + for user_id in target_users + for stream_id in stream_ids ] self.db_pool.simple_insert_many_txn( txn, @@ -1049,25 +1051,20 @@ class ProfileWorkerStore(SQLBaseStore): ) res = txn.fetchall() if res: - stream_ids = [row[0] for row in res] - user_clause, user_args = make_in_list_sql_clause( txn.database_engine, "user_id", users_to_update, ) - stream_id_clause, stream_id_args = make_in_list_sql_clause( - txn.database_engine, - "stream_id", - stream_ids, - ) txn.execute( f""" DELETE FROM profile_updates_per_user WHERE {user_clause} - AND {stream_id_clause} + AND stream_id IN ( + SELECT stream_id FROM profile_updates WHERE user_id = ? + ) """, - (*user_args, *stream_id_args), + (*user_args, user_id.to_string()), ) # Now record the "left room" action in the stream