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

@@ -4486,6 +4486,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

@@ -585,6 +585,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

schema/synapse-config.schema.yaml

@@ -5541,6 +5541,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/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
         )