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

Make profile_updates table support several affected fields in one update

Browse Source
This commit is contained in:
Olivier 'reivilibre
2026-07-31 14:34:46 +01:00
parent f9a5712e46
commit cdda205c39
6 changed files with 164 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/sync.py
+9 -7
View File
@@ -2378,17 +2378,19 @@ class SyncHandler:
updated_user_fields: dict[str, set[str]] = {}
# Set fields from updates
for update in updates:
# Skip the update if there is no field update (a joined or left room
# action), the client didn't ask for this field, or we're not
# interested in this user.
if (
not update.field_name
or update.field_name not in profile_fields
# Skip the update if there is no field update (a joined or left room action),
update.action != ProfileUpdateAction.UPDATE
or update.affected_fields is None
# or if the client isn't interested in any of the fields
or update.affected_fields.isdisjoint(profile_fields)
# or we're not interested in this user.
or update.user_id not in users
):
continue
updated_user_fields.setdefault(update.user_id, set()).add(
update.field_name
updated_user_fields.setdefault(update.user_id, set()).update(
# Add any fields that were affected and that we're interested in
update.affected_fields & profile_fields
)
# Note: there's a small race condition here where a profile update may
synapse/replication/tcp/streams/_base.py
+4 -4
View File
@@ -774,11 +774,11 @@ class ProfileUpdatesStreamRow:
"""The full user ID with the profile update."""
action: ProfileUpdateAction
"""The action, either 'update' for a field update, 'left_room' if the user left
a room or `joined_room` if the user joined a room, see ProfileUpdateAction constant.
a room or `joined_room` if the user joined a room, see ProfileUpdateAction enum.
"""
field_name: str | None
"""The profile field that was updated, see https://spec.matrix.org/unstable/client-server-api/#profiles.
This is None if `action` is not 'update'.
affected_fields: frozenset[str] | None
"""Names of the profile fields that were added, updated or removed, see https://spec.matrix.org/unstable/client-server-api/#profiles.
This is None if `action` is not `update`.
"""
synapse/storage/databases/main/profile.py
+110 -70
View File
@@ -19,7 +19,7 @@
#
#
import json
from typing import TYPE_CHECKING, Collection, cast
from typing import TYPE_CHECKING, AbstractSet, Collection, cast
import attr
from canonicaljson import encode_canonical_json
@@ -32,7 +32,7 @@ from synapse.api.constants import (
)
from synapse.api.errors import Codes, StoreError
from synapse.replication.tcp.streams._base import ProfileUpdatesStream
from synapse.storage._base import SQLBaseStore, make_in_list_sql_clause
from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
from synapse.storage.database import (
DatabasePool,
LoggingDatabaseConnection,
@@ -42,6 +42,7 @@ from synapse.storage.databases.main.roommember import ProfileInfo
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
from synapse.storage.util.id_generators import MultiWriterIdGenerator
from synapse.types import JsonDict, JsonValue, UserID
from synapse.util.json import json_encoder
if TYPE_CHECKING:
from synapse.server import HomeServer
@@ -58,7 +59,7 @@ class ProfileUpdate:
stream_id: int
user_id: str
action: str
field_name: str | None
affected_fields: frozenset[str] | None
class ProfileWorkerStore(SQLBaseStore):
@@ -351,7 +352,7 @@ class ProfileWorkerStore(SQLBaseStore):
async def get_updated_profile_updates(
self, *, from_id: int, to_id: int, limit: int
) -> list[tuple[int, str, str, str | None]]:
) -> list[tuple[int, str, str, frozenset[str] | None]]:
"""Get updates to profile updates between two stream IDs.
Bounds: from_id < ... <= to_id
@@ -369,11 +370,11 @@ class ProfileWorkerStore(SQLBaseStore):
def _get_updated_profile_updates_txn(
txn: LoggingTransaction,
) -> list[tuple[int, str, str, str | None]]:
) -> list[tuple[int, str, str, frozenset[str] | None]]:
txn.execute(
"""
SELECT
stream_id, user_id, action, field_name
stream_id, user_id, action, affected_fields
FROM profile_updates
WHERE
? < stream_id AND stream_id <= ?
@@ -381,7 +382,21 @@ class ProfileWorkerStore(SQLBaseStore):
""",
(from_id, to_id, limit),
)
return cast(list[tuple[int, str, str, str | None]], txn.fetchall())
return [
(
stream_id,
user_id,
action,
(
# affected_fields is a JSON array, turn it to a frozenset[str]
frozenset(db_to_json(affected_fields))
if affected_fields is not None
else None
),
)
for stream_id, user_id, action, affected_fields in txn
]
return await self.db_pool.runInteraction(
"get_updated_profile_updates", _get_updated_profile_updates_txn
@@ -392,7 +407,7 @@ class ProfileWorkerStore(SQLBaseStore):
*,
from_id: int,
to_id: int,
field_names: Collection[str],
field_names: AbstractSet[str],
) -> list[ProfileUpdate]:
"""Get profile update markers for the given fields in a stream range.
@@ -405,6 +420,7 @@ class ProfileWorkerStore(SQLBaseStore):
Returns:
list of ProfileUpdates update rows
The `affected_fields` entry in the ProfileUpdates will be filtered.
"""
if from_id >= to_id:
return []
@@ -415,29 +431,59 @@ class ProfileWorkerStore(SQLBaseStore):
def _get_profile_updates_for_fields_txn(
txn: LoggingTransaction,
) -> list[ProfileUpdate]:
clause, args = make_in_list_sql_clause(
txn.database_engine, "field_name", field_names
wanted_field_in_elems_clause, wanted_field_in_elems_args = (
make_in_list_sql_clause(
txn.database_engine, "field_names.value", field_names
)
)
if isinstance(txn.database_engine, PostgresEngine):
# Note that if we had a GIN index on `affected_fields`, this would defeat it.
# If we decide we want one, we should consider using the `?|` operator or its
# clearer-named `jsonb_exists_any` equivalent.
all_field_names_table_expression = (
"jsonb_array_elements_text(affected_fields) AS field_names(value)"
)
else:
# json_each is a table-valued function that gives `value` as one of its column names
all_field_names_table_expression = (
"json_each(affected_fields) AS field_names"
)
txn.execute(
f"""
SELECT stream_id, user_id, action, field_name
FROM profile_updates
SELECT stream_id, user_id, action, affected_fields
FROM profile_updates
WHERE ? < stream_id AND stream_id <= ?
AND ({clause} OR action != ?)
AND (
(EXISTS (SELECT 1 FROM {all_field_names_table_expression} WHERE {wanted_field_in_elems_clause}))
OR action != ?
)
ORDER BY stream_id ASC
""",
(from_id, to_id, *args, ProfileUpdateAction.UPDATE.value),
(
from_id,
to_id,
*wanted_field_in_elems_args,
ProfileUpdateAction.UPDATE.value,
),
)
rows = cast(list[tuple[int, str, str, str | None]], txn.fetchall())
updates: list[ProfileUpdate] = []
for stream_id, user_id, action, field_name in rows:
for stream_id, user_id, action, affected_fields_dbjson in rows:
updates.append(
ProfileUpdate(
stream_id=stream_id,
user_id=user_id,
action=action,
field_name=field_name,
affected_fields=(
# Get the field names that were affected by this update
# and intersect with the field names we care about
frozenset(db_to_json(affected_fields_dbjson)) & field_names
)
if affected_fields_dbjson is not None
else None,
)
)
@@ -453,7 +499,7 @@ class ProfileWorkerStore(SQLBaseStore):
from_id: int,
to_id: int,
user_id: str,
field_names: set[str],
field_names: AbstractSet[str],
include_users: set[str] | None = None,
) -> list[ProfileUpdate]:
"""Get profile update markers for a user in a stream range.
@@ -486,9 +532,23 @@ class ProfileWorkerStore(SQLBaseStore):
def _get_profile_updates_for_user_and_fields_txn(
txn: LoggingTransaction,
) -> list[ProfileUpdate]:
field_clause, field_args = make_in_list_sql_clause(
txn.database_engine, "pu.field_name", field_names
wanted_field_in_elems_clause, wanted_field_in_elems_args = (
make_in_list_sql_clause(
txn.database_engine, "field_names.value", field_names
)
)
if isinstance(txn.database_engine, PostgresEngine):
# Note that if we had a GIN index on `affected_fields`, this would defeat it.
# If we decide we want one, we should consider using the `?|` operator or its
# clearer-named `jsonb_exists_any` equivalent.
all_field_names_table_expression = "jsonb_array_elements_text(pu.affected_fields) AS field_names(value)"
else:
# json_each is a table-valued function that gives `value` as one of its column names
all_field_names_table_expression = (
"json_each(pu.affected_fields) AS field_names"
)
user_clause = ""
user_args: list[str] = []
if include_users is not None:
@@ -504,14 +564,17 @@ class ProfileWorkerStore(SQLBaseStore):
# and the `user_id` and `field_names` match.
txn.execute(
f"""
SELECT pu.stream_id, pu.user_id, pu.action, pu.field_name
SELECT pu.stream_id, pu.user_id, pu.action, pu.affected_fields
FROM profile_updates AS pu
INNER JOIN profile_updates_per_user AS puf
INNER JOIN profile_updates_per_user AS puf
ON pu.stream_id = puf.stream_id
WHERE ? < pu.stream_id AND pu.stream_id <= ?
AND puf.user_id = ?
{user_clause}
AND ({field_clause} OR pu.action != ?)
AND (
(EXISTS (SELECT 1 FROM {all_field_names_table_expression} WHERE {wanted_field_in_elems_clause}))
OR pu.action != ?
)
ORDER BY pu.stream_id ASC
""",
(
@@ -519,20 +582,26 @@ class ProfileWorkerStore(SQLBaseStore):
to_id,
user_id,
*user_args,
*field_args,
*wanted_field_in_elems_args,
ProfileUpdateAction.UPDATE.value,
),
)
rows = cast(list[tuple[int, str, str, str | None]], txn.fetchall())
updates: list[ProfileUpdate] = []
for stream_id, updated_user_id, action, field_name in rows:
for stream_id, updated_user_id, action, affected_fields_dbjson in rows:
updates.append(
ProfileUpdate(
stream_id=stream_id,
user_id=updated_user_id,
action=action,
field_name=field_name,
affected_fields=(
# Get the field names that were affected by this update
# and intersect with the field names we care about
frozenset(db_to_json(affected_fields_dbjson)) & field_names
)
if affected_fields_dbjson is not None
else None,
)
)
@@ -801,7 +870,7 @@ class ProfileWorkerStore(SQLBaseStore):
txn: LoggingTransaction,
user_id: UserID,
action: ProfileUpdateAction,
field_names: list[str] | None,
field_names: Collection[str] | None,
user_rooms: set[str] | None = None,
target_users: set[str] | None = None,
) -> int | None:
@@ -874,54 +943,25 @@ class ProfileWorkerStore(SQLBaseStore):
# Record the profile update
inserted_ts = self.clock.time_msec()
if field_names:
stream_ids = self._profile_updates_id_gen.get_next_mult_txn(
txn, len(field_names)
)
values: list[tuple[int, str, str, str, str | None, int]] = [
(
stream_id,
self._instance_name,
user_id.to_string(),
action.value,
field_name,
inserted_ts,
)
for stream_id, field_name in zip(stream_ids, field_names)
]
else:
stream_ids = [self._profile_updates_id_gen.get_next_txn(txn)]
values = [
(
stream_ids[0],
self._instance_name,
user_id.to_string(),
action.value,
None,
inserted_ts,
)
]
self.db_pool.simple_insert_many_txn(
stream_id = self._profile_updates_id_gen.get_next_txn(txn)
self.db_pool.simple_insert_txn(
txn,
table="profile_updates",
keys=[
"stream_id",
"instance_name",
"user_id",
"action",
"field_name",
"inserted_ts",
],
values=values,
values={
"stream_id": stream_id,
"instance_name": self._instance_name,
"user_id": user_id.to_string(),
"action": action.value,
"affected_fields": json_encoder.encode(sorted(field_names))
if field_names
else None,
"inserted_ts": inserted_ts,
},
)
# Add per user tracking rows for each generated stream ID
inserted_ts = self.clock.time_msec()
per_user_values = [
(stream_id, user_id, inserted_ts)
for user_id in users
for stream_id in stream_ids
]
per_user_values = [(stream_id, user_id, inserted_ts) for user_id in users]
self.db_pool.simple_insert_many_txn(
txn,
table="profile_updates_per_user",
@@ -932,7 +972,7 @@ class ProfileWorkerStore(SQLBaseStore):
],
values=per_user_values,
)
return stream_ids[-1]
return stream_id
async def set_profile_field(
self,
synapse/storage/databases/main/roommember.py
+1 -1
View File
@@ -902,7 +902,7 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
(
user_id,
*exclude_room_id_args,
state_key_args,
*state_key_args,
*exclude_room_id_args,
),
)
synapse/storage/schema/main/delta/94/07_profile_updates.sql
+16 -6
View File
@@ -11,7 +11,9 @@
-- See the GNU Affero General Public License for more details:
-- <https://www.gnu.org/licenses/agpl-3.0.html>.
-- Track updates to profile fields for MSC4429 legacy /sync.
-- Track updates to profile fields.
-- For MSC4429 legacy /sync and others.
-- See https://github.com/element-hq/synapse/issues/19981 for potential future directions of this table.
CREATE TABLE IF NOT EXISTS profile_updates (
stream_id BIGINT NOT NULL PRIMARY KEY,
instance_name TEXT NOT NULL,
@@ -22,17 +24,25 @@ CREATE TABLE IF NOT EXISTS profile_updates (
-- Profile action that has happened, see ProfileUpdateAction enum.
action TEXT NOT NULL,
-- Profile field name that has been updated,
-- see https://spec.matrix.org/unstable/client-server-api/#profiles
-- This is only required if "action" is "update"
field_name TEXT NULL,
-- JSON array of the profile field names that have been
-- added, updated or removed in this update.
-- See https://spec.matrix.org/unstable/client-server-api/#profiles
-- This is only present if `action` is `update`.
--
-- We support multiple field updates at once because it is easy to foresee features
-- involving multiple fields (where getting the illusion of a torn write might be harmful),
-- as well as synchronisation over federation being likely to lead to multiple field changes
-- at once.
affected_fields JSONB NULL,
-- Unix timestamp for debugging purposes
inserted_ts BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS profile_updates_by_user ON profile_updates (user_id, stream_id);
CREATE INDEX IF NOT EXISTS profile_updates_by_field ON profile_updates (field_name, stream_id);
-- We aren't creating a GIN index on `affected_fields` at this time because we don't expect
-- field names to be very selective and therefore an index might not be that useful.
-- Track which local users should receive each profile update.
CREATE TABLE IF NOT EXISTS profile_updates_per_user (
tests/handlers/test_profile.py
+24 -19
View File
@@ -313,14 +313,19 @@ class ProfileTestCase(unittest.HomeserverTestCase):
)
self.assertEqual(
updates[0],
(2, "@1234abcd:test", ProfileUpdateAction.UPDATE.value, field_name),
(
2,
"@1234abcd:test",
ProfileUpdateAction.UPDATE.value,
{field_name},
),
)
fields_updates = self.get_success(
self.store.get_profile_updates_for_fields(
from_id=1,
to_id=2,
field_names=[field_name],
field_names={field_name},
)
)
self.assertEqual(
@@ -329,7 +334,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=2,
user_id="@1234abcd:test",
action=ProfileUpdateAction.UPDATE.value,
field_name=field_name,
affected_fields=frozenset({field_name}),
),
)
@@ -350,7 +355,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
)
self.assertEqual(
delete_updates[0],
(3, "@1234abcd:test", ProfileUpdateAction.UPDATE.value, field_name),
(3, "@1234abcd:test", ProfileUpdateAction.UPDATE.value, {field_name}),
)
@override_config({"include_profile_updates_in_sync": True})
@@ -392,13 +397,13 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=3,
user_id="@millie:test",
action="joined_room",
field_name=None,
affected_fields=None,
),
ProfileUpdate(
stream_id=4,
user_id=self.frank.to_string(),
action="update",
field_name="m.status",
affected_fields=frozenset({"m.status"}),
),
],
)
@@ -417,7 +422,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=4,
user_id=self.frank.to_string(),
action="update",
field_name="m.status",
affected_fields=frozenset({"m.status"}),
),
],
)
@@ -436,19 +441,19 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=2,
user_id="@roger:test",
action="joined_room",
field_name=None,
affected_fields=None,
),
ProfileUpdate(
stream_id=3,
user_id="@millie:test",
action="joined_room",
field_name=None,
affected_fields=None,
),
ProfileUpdate(
stream_id=4,
user_id=self.frank.to_string(),
action="update",
field_name="m.status",
affected_fields=frozenset({"m.status"}),
),
],
)
@@ -507,7 +512,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=2,
user_id="@roger:test",
action="joined_room",
field_name=None,
affected_fields=None,
),
],
)
@@ -563,13 +568,13 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=4,
user_id="@gracie:test",
action="joined_room",
field_name=None,
affected_fields=None,
),
ProfileUpdate(
stream_id=5,
user_id=self.frank.to_string(),
action="update",
field_name="m.status",
affected_fields=frozenset({"m.status"}),
),
],
)
@@ -588,7 +593,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=5,
user_id=self.frank.to_string(),
action="update",
field_name="m.status",
affected_fields=frozenset({"m.status"}),
),
],
)
@@ -611,13 +616,13 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=4,
user_id="@gracie:test",
action="joined_room",
field_name=None,
affected_fields=None,
),
ProfileUpdate(
stream_id=6,
user_id=self.frank.to_string(),
action="left_room",
field_name=None,
affected_fields=None,
),
],
)
@@ -638,13 +643,13 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=6,
user_id=self.frank.to_string(),
action="left_room",
field_name=None,
affected_fields=None,
),
ProfileUpdate(
stream_id=7,
user_id="@gracie:test",
action="left_room",
field_name=None,
affected_fields=None,
),
],
)
@@ -665,7 +670,7 @@ class ProfileTestCase(unittest.HomeserverTestCase):
stream_id=5,
user_id=self.frank.to_string(),
action="update",
field_name="m.status",
affected_fields=frozenset({"m.status"}),
),
],
)