Introduce support for MSC4429: Profile Updates for Legacy Sync

changelog.d/19556.feature

@@ -0,0 +1,2 @@
+Add optional support for [MSC4429: Profile Updates for Legacy Sync](https://github.com/matrix-org/matrix-spec-proposals/pull/4429).
+Currently defaults to not enabled, and is limited to local users only for the sync results.

docker/complement/conf/start_for_complement.sh

@@ -60,12 +60,13 @@ if [[ -n "$SYNAPSE_COMPLEMENT_USE_WORKERS" ]]; then
       federation_inbound, \
       federation_reader, \
       federation_sender, \
+      profile_updates, \
       synchrotron, \
       client_reader, \
       appservice, \
       pusher, \
       device_lists:2, \
-      stream_writers=account_data+presence+receipts+to_device+typing"
+      stream_writers=account_data+presence+profile_updates+receipts+to_device+typing"
 
   fi
   log "Workers requested: $SYNAPSE_WORKER_TYPES"

docker/complement/conf/workers-shared-extra.yaml.j2

@@ -15,6 +15,8 @@ enable_registration_without_verification: true
 bcrypt_rounds: 4
 url_preview_enabled: true
 url_preview_ip_range_blacklist: []
+# MSC4429 Profile updates down legacy /sync
+include_profile_updates_in_sync: true
 
 ## Registration ##
 

docker/configure_workers_and_start.py

@@ -269,7 +269,10 @@
             "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/join/",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/knock/",
-            "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/",
+            # The [^/] differentiates this endpoint from
+            # `ProfileRestFieldsServlet`, which we want to instead go to the
+            # `profile_updates` worker below.
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/[^/]+",
         ],
         "shared_extra_conf": {},
         "worker_extra_conf": "",
@@ -308,6 +311,15 @@
         "shared_extra_conf": {},
         "worker_extra_conf": "",
     },
+    "profile_updates": {
+        "app": "synapse.app.generic_worker",
+        "listener_resources": ["client", "replication"],
+        "endpoint_patterns": [
+            "^/_matrix/client/(unstable/uk.tcpip.msc4133|api/v1|r0|v3|unstable)/profile/.+/"
+        ],
+        "shared_extra_conf": {},
+        "worker_extra_conf": "",
+    },
     "device_lists": {
         "app": "synapse.app.generic_worker",
         "listener_resources": ["client", "replication"],
@@ -517,6 +529,7 @@ def add_worker_roles_to_shared_config(
         "typing",
         "push_rules",
         "thread_subscriptions",
+        "profile_updates",
     }
 
     # Worker-type specific sharding config. Now a single worker can fulfill multiple

docs/usage/configuration/config_documentation.md

@@ -318,6 +318,16 @@ Example configuration:
 include_profile_data_on_invite: false
 ```
 ---
+### `include_profile_updates_in_sync`
+
+*(boolean)* Use this option to include updates of other users profiles in sync responses, for users who share rooms.
+Note, requires an MSC4429 compatible client, and is currently limited to legacy sync and local users only. Defaults to `false`.
+
+Example configuration:
+```yaml
+include_profile_updates_in_sync: true
+```
+---
 ### `allow_public_rooms_without_auth`
 
 *(boolean)* If set to true, removes the need for authentication to access the server's public rooms directory through the client API, meaning that anyone can query the room directory. Defaults to `false`.
@@ -4486,6 +4496,8 @@ This setting has the following sub-options:
 
 * `device_lists` (string): Name of a worker assigned to the `device_lists` stream.
 
+* `profile_updates` (string): Name of a worker assigned to the `profile_updates` stream.
+
 Example configuration:
 ```yaml
 stream_writers:

docs/workers.md

@@ -577,6 +577,7 @@ configured as stream writer for the `quarantined_media_changes` stream:
 
     ^/_synapse/admin/v1/quarantine_media/.*$
 
+
 #### Restrict outbound federation traffic to a specific set of workers
 
 The

rust/src/config/mod.rs

@@ -48,6 +48,7 @@ pub struct AuthConfig {
 #[derive(FromPyObject, Clone)]
 pub struct ServerConfig {
     pub max_event_delay_ms: Option<u64>,
+    pub include_profile_updates_in_sync: bool,
 }
 
 #[derive(FromPyObject, Clone)]

rust/src/handlers/versions.rs

@@ -257,6 +257,9 @@ pub struct UnstableFeatureMap {
     /// MSC4380: Invite blocking
     #[serde(rename = "org.matrix.msc4380.stable")]
     msc4380: bool,
+    /// MSC4429: Profile updates for legacy /sync.
+    #[serde(rename = "org.matrix.msc4429")]
+    msc4429: bool,
     /// MSC4445: Sync timeline order
     #[serde(rename = "org.matrix.msc4445.initial_sync_timeline_topological_ordering")]
     msc4445_initial_sync_timeline_topological_ordering: bool,
@@ -316,6 +319,7 @@ pub fn synapse_config_to_global_unstable_feature_map(
         msc4169: config.experimental.msc4169_enabled,
         msc4354: config.experimental.msc4354_enabled,
         msc4380: true,
+        msc4429: config.server.include_profile_updates_in_sync,
         msc4445_initial_sync_timeline_topological_ordering: true,
         msc4491_enabled: config.experimental.msc4491_enabled,
         msc4143_enabled: config.experimental.msc4143_enabled,

schema/synapse-config.schema.yaml

@@ -235,6 +235,17 @@ properties:
     default: true
     examples:
       - false
+  include_profile_updates_in_sync:
+    type: boolean
+    description: >-
+      Use this option to include updates of other users profiles in sync responses,
+      for users who share rooms.
+
+      Note, requires an MSC4429 compatible client, and is currently limited to
+      legacy sync and local users only.
+    default: false
+    examples:
+      - true
   allow_public_rooms_without_auth:
     type: boolean
     description:
@@ -5541,6 +5552,9 @@ properties:
       device_lists:
         type: string
         description: Name of a worker assigned to the `device_lists` stream.
+      profile_updates:
+        type: string
+        description: Name of a worker assigned to the `profile_updates` stream.
     default: {}
     examples:
       - events: worker1

scripts-dev/check_schema_delta.py

@@ -14,6 +14,11 @@
 
 SCHEMA_FILE_REGEX = re.compile(r"^synapse/storage/schema/(.*)/delta/(.*)/(.*)$")
 
+# Keep this in sync with synapse.storage.engines._base. The CI job for this
+# script deliberately installs only its lightweight parsing dependencies, so we
+# avoid importing Synapse here.
+AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER = "$%AUTO_INCREMENT_PRIMARY_KEY%$"
+
 # The base branch we want to check against. We use the main development branch
 # on the assumption that is what we are developing against.
 DEVELOP_BRANCH = "develop"
@@ -81,7 +86,7 @@ def main(force_colors: bool) -> None:
     bad_delta_files = []
     changed_delta_files = []
     for diff in diffs:
-        if diff.b_path is None:
+        if diff.deleted_file or diff.b_path is None:
             # We don't lint deleted files.
             continue
 
@@ -196,6 +201,10 @@ def check_schema_delta(delta_files: list[str], force_colors: bool) -> bool:
             )
             return True
 
+        delta_contents = _replace_auto_increment_primary_key_placeholder(
+            delta_contents, sql_lang
+        )
+
         statements = sqlglot.parse(delta_contents, read=sql_lang)
 
         for statement in statements:
@@ -244,5 +253,18 @@ def check_schema_delta(delta_files: list[str], force_colors: bool) -> bool:
     return success
 
 
+def _replace_auto_increment_primary_key_placeholder(
+    delta_contents: str, sql_lang: str
+) -> str:
+    """Replace Synapse's auto-increment PK placeholder with parseable SQL."""
+
+    if sql_lang == "sqlite":
+        replacement = "INTEGER PRIMARY KEY AUTOINCREMENT"
+    else:
+        replacement = "BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY"
+
+    return delta_contents.replace(AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER, replacement)
+
+
 if __name__ == "__main__":
     main()

scripts-dev/complement.sh

@@ -286,6 +286,7 @@ main() {
     ./tests/msc4155
     ./tests/msc4306
     ./tests/msc4222
+    ./tests/msc4429
   )
 
   # Export the list of test packages as a space-separated environment variable, so other

synapse/_scripts/synapse_port_db.py

@@ -917,6 +917,10 @@ def alter_table(txn: LoggingTransaction) -> None:
                 "quarantined_media_id_seq",
                 [("quarantined_media_changes", "stream_id")],
             )
+            await self._setup_sequence(
+                "profile_updates_sequence",
+                [("profile_updates", "stream_id")],
+            )
 
             # Step 3. Get tables.
             self.progress.set_state("Fetching tables")

synapse/api/constants.py

@@ -410,6 +410,56 @@ class ProfileFields:
     AVATAR_URL: Final = "avatar_url"
 
 
+class ProfileUpdateAction(str, enum.Enum):
+    """
+    Enum representing the action of a row in the profile updates stream tables.
+    These actions are used to determine how profiles, and what data is included in the
+    sync responses, depending on field updates and room membership changes.
+    """
+
+    JOINED_ROOM = "joined_room"
+    """
+    This profile update row action represents a user joining a room.
+
+    When gathering an incremental sync non-lazy response for profile updates,
+    we always include the full profile of users who have joined a room the syncing
+    user is a member of, where full profile means all the current profile values the
+    client asked for, regardless of whether they have changed recently. This ensures
+    that clients have profile re-populated for any users who have recently left
+    shared rooms.
+
+    A scenario example would be as follows:
+
+        * Alice leaves a room with Bob
+        * Bob's client clears all profile fields from Alice
+        * Alice joins a room with Bob
+        * Bob's client does an incremental non-lazy sync
+
+    At the end of the flow Bob should receive all the profile fields the client
+    is interested in, not just the potential diff, which non-lazy incremental sync
+    normally includes. This update action currently has no meaning for sync responses
+    that are not incremental and non-lazy.
+    """
+    LEFT_ROOM = "left_room"
+    """
+    This profile update row action represents a user leaving a room.
+
+    Clients will want to know when they no longer share rooms with a user. This
+    profile action row allows the sync code to deliver a `null` response for those
+    profiles, so clients can clear their cache containing the users profile data
+    they are no longer interested in.
+    """
+    UPDATE = "update"
+    """
+    This profile update row action represents a user updating a profile field.
+
+    Depending on the type of sync (initial/incremental, lazy/non-lazy), either the
+    diff of profile field updates or all the current profile fields are included
+    in the sync response. In the latter case the profile update action row signifies
+    a change, but the client may still get fields that have not changed.
+    """
+
+
 class StickyEventField(TypedDict):
     """
     Dict content of the `sticky` part of an event.

synapse/api/filtering.py

@@ -123,13 +123,23 @@
         "filter": FILTER_SCHEMA,
         "room_filter": ROOM_FILTER_SCHEMA,
         "room_event_filter": ROOM_EVENT_FILTER_SCHEMA,
+        "profile_fields_filter": {
+            "type": "object",
+            "properties": {
+                "ids": {"type": "array", "items": {"type": "string"}},
+            },
+            "additionalProperties": True,
+        },
     },
     "properties": {
         "presence": {"$ref": "#/definitions/filter"},
         "account_data": {"$ref": "#/definitions/filter"},
         "room": {"$ref": "#/definitions/room_filter"},
         "event_format": {"type": "string", "enum": ["client", "federation"]},
         "event_fields": {"type": "array", "items": {"type": "string"}},
+        "org.matrix.msc4429.profile_fields": {
+            "$ref": "#/definitions/profile_fields_filter"
+        },
     },
     "additionalProperties": True,  # Allow new fields for forward compatibility
 }
@@ -217,6 +227,13 @@ def __init__(self, hs: "HomeServer", filter_json: JsonMapping):
         self.event_fields = filter_json.get("event_fields", [])
         self.event_format = filter_json.get("event_format", "client")
 
+        self.profile_fields: set[str] = set()
+        if hs.config.server.include_profile_updates_in_sync:
+            profile_fields_filter = filter_json.get("org.matrix.msc4429.profile_fields")
+
+            if isinstance(profile_fields_filter, Mapping):
+                self.profile_fields = set(profile_fields_filter.get("ids", []))
+
     def __repr__(self) -> str:
         return "<FilterCollection %s>" % (json.dumps(self._filter_json),)
 

synapse/config/server.py

@@ -545,6 +545,12 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
                 " 'allow_public_rooms_over_federation' is set."
             )
 
+        # Whether to support MSC4299 profile updates down legacy /sync
+        self.include_profile_updates_in_sync = config.get(
+            "include_profile_updates_in_sync",
+            False,
+        )
+
         # Check if the legacy "restrict_public_rooms_to_local_users" flag is set. This
         # flag is now obsolete but we need to check it for backward-compatibility.
         if config.get("restrict_public_rooms_to_local_users", False):

synapse/config/workers.py

@@ -142,6 +142,9 @@ class WriterLocations:
         push_rules: The instances that write to the push stream. Currently
             can only be a single instance.
         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.
     """
@@ -179,7 +182,11 @@ class WriterLocations:
         converter=_instance_to_list_converter,
     )
     thread_subscriptions: list[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
+        converter=_instance_to_list_converter,
+    )
+    profile_updates: list[str] = attr.ib(
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     quarantined_media_changes: list[str] = attr.ib(
@@ -361,8 +368,7 @@ def read_config(
         writers = config.get("stream_writers") or {}
         self.writers = WriterLocations(**writers)
 
-        # Check that the configured writers for events and typing also appears in
-        # `instance_map`.
+        # Check that the configured writers also appear in `instance_map`.
         for stream in (
             "events",
             "typing",
@@ -371,6 +377,9 @@ def read_config(
             "receipts",
             "presence",
             "push_rules",
+            "device_lists",
+            "thread_subscriptions",
+            "profile_updates",
         ):
             instances = _instance_to_list_converter(getattr(self.writers, stream))
             for instance in instances:
@@ -421,6 +430,16 @@ def read_config(
                 "Must specify at least one instance to handle `device_lists` messages."
             )
 
+        if len(self.writers.thread_subscriptions) == 0:
+            raise ConfigError(
+                "Must specify at least one instance to handle `thread_subscriptions` messages."
+            )
+
+        if len(self.writers.profile_updates) == 0:
+            raise ConfigError(
+                "Must specify at least one instance to handle `profile_updates` messages."
+            )
+
         self.events_shard_config = RoutableShardedWorkerHandlingConfig(
             self.writers.events
         )

synapse/handlers/deactivate_account.py

@@ -172,7 +172,7 @@ async def deactivate_account(
             # Note that displayname and avatar URL may persist as historical state events
             # in rooms, but these cases behave like message history, following
             # https://spec.matrix.org/v1.17/client-server-api/#post_matrixclientv3accountdeactivate
-            await self._profile_handler.delete_profile_upon_deactivation(
+            await self._profile_handler.dispatch_delete_profile_upon_deactivation(
                 user, requester, by_admin
             )