From fbfa7bcfd11bbb56297be3633fd16cd378dae8ad Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Wed, 23 Sep 2026 18:08:43 +0200 Subject: [PATCH 1/6] Fix push badge count ignoring a room's notifications when another room's summary is up to date (#20237) The count of unread notifications is computed in two phases: 1. From the summaries (`event_push_summary`) when those are up to date with the user's last read receipt 2. Otherwise from counting the push actions (`event_push_actions`) after that receipt The problem in this process is that the threads whose summary is up to date are identified with only their `thread_id`. But all main timelines share the same `"main"` thread_id, so as soon as one room has an up-to-date summary, phase 2 skips the main timeline of every room. The fix identifies the up-to-date summaries with their `(room_id, thread_id)` pair. Authored by @sandhose. Found while investigating the `TestThreadedReceipts` Complement flake (#15517, #18537), but it is a bug on its own. ### Pull Request Checklist * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [x] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --------- Co-authored-by: Quentin Gliech Co-authored-by: Devon Hudson --- changelog.d/20237.bugfix | 1 + .../databases/main/event_push_actions.py | 29 ++++--- tests/storage/test_event_push_actions.py | 76 +++++++++++++++++++ 3 files changed, 90 insertions(+), 16 deletions(-) create mode 100644 changelog.d/20237.bugfix diff --git a/changelog.d/20237.bugfix b/changelog.d/20237.bugfix new file mode 100644 index 0000000000..3d6c6bf027 --- /dev/null +++ b/changelog.d/20237.bugfix @@ -0,0 +1 @@ +Fix a bug where push badge counts could ignore notifications in one room based on another room's summarised counts. diff --git a/synapse/storage/databases/main/event_push_actions.py b/synapse/storage/databases/main/event_push_actions.py index 9c5fd35906..afbbe6b61a 100644 --- a/synapse/storage/databases/main/event_push_actions.py +++ b/synapse/storage/databases/main/event_push_actions.py @@ -435,12 +435,13 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas """ txn.execute(sql, args) - seen_thread_ids = set() + # The (room ID, thread ID) pairs we found an up-to-date summary for. + seen_room_thread_ids = set() room_to_count: dict[str, int] = defaultdict(int) for room_id, thread_id, notif_count in txn: room_to_count[room_id] += notif_count - seen_thread_ids.add(thread_id) + seen_room_thread_ids.add((room_id, thread_id)) # Now get any event push actions that haven't been rotated using the same OR # join and filter by receipt and event push summary rotated up to stream ordering. @@ -460,35 +461,31 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas for room_id, thread_id, notif_count in txn: # Note: only count push actions we have valid summaries for with up to date receipt. - if thread_id not in seen_thread_ids: + if (room_id, thread_id) not in seen_room_thread_ids: continue room_to_count[room_id] += notif_count - thread_id_clause, thread_ids_args = make_in_list_sql_clause( - self.database_engine, "epa.thread_id", seen_thread_ids - ) - - # Finally re-check event_push_actions for any rooms not in the summary, ignoring - # the rotated up-to position. This handles the case where a read receipt has arrived - # but not been rotated meaning the summary table is out of date, so we go back to - # the push actions table. + # Finally re-check event_push_actions for any room/threads not in the summary, + # ignoring the rotated up-to position. This handles the case where a read receipt + # has arrived but not been rotated meaning the summary table is out of date, so we + # go back to the push actions table. sql = f""" {receipts_cte} - SELECT epa.room_id, COUNT(CASE WHEN epa.notif = 1 THEN 1 END) AS notif_count + SELECT epa.room_id, epa.thread_id, COUNT(CASE WHEN epa.notif = 1 THEN 1 END) AS notif_count FROM event_push_actions AS epa {receipts_joins} WHERE user_id = ? - AND NOT {thread_id_clause} AND epa.notif = 1 AND (threaded_receipt_stream_ordering IS NULL OR stream_ordering > threaded_receipt_stream_ordering) AND (unthreaded_receipt_stream_ordering IS NULL OR stream_ordering > unthreaded_receipt_stream_ordering) - GROUP BY epa.room_id + GROUP BY epa.room_id, epa.thread_id """ - args.extend(thread_ids_args) txn.execute(sql, args) - for room_id, notif_count in txn: + for room_id, thread_id, notif_count in txn: + if (room_id, thread_id) in seen_room_thread_ids: + continue room_to_count[room_id] += notif_count return room_to_count diff --git a/tests/storage/test_event_push_actions.py b/tests/storage/test_event_push_actions.py index bfe7049fb2..158426f83a 100644 --- a/tests/storage/test_event_push_actions.py +++ b/tests/storage/test_event_push_actions.py @@ -472,6 +472,82 @@ class EventPushActionsStoreTestCase(HomeserverTestCase): self.get_success(self.store._rotate_notifs()) _assert_badge(1) + def test_count_aggregation_badge_recount_is_scoped_per_room(self) -> None: + """ + Regression test: a room whose summary row is out of date must be recounted + from `event_push_actions`, even when another room has an up-to-date summary + for the same thread ID. + + The set of threads a valid summary was found for used to be keyed on the + thread ID alone, so a single room with an up-to-date `main` summary excluded + `main` from the recount in *every* room, dropping those rooms' counts. + """ + user_id, token, other_id, other_token, room_id = self._create_users_and_room() + + stale_room_id = self.helper.create_room_as(user_id, tok=token) + self.helper.join(stale_room_id, other_id, tok=other_token) + + def _send(room: str) -> str: + return self.helper.send_event( + room, + type="m.room.message", + content={"msgtype": "m.text", "body": "msg"}, + tok=other_token, + )["event_id"] + + def _read(room: str, event_id: str) -> None: + self.get_success( + self.store.insert_receipt( + room, + "m.read", + user_id=user_id, + event_ids=[event_id], + thread_id=None, + data={}, + ) + ) + + def _badge(room: str) -> int: + counts = self.get_success( + self.store.db_pool.runInteraction( + "get-aggregate-unread-counts", + self.store._get_unread_counts_by_room_for_user_txn, + user_id, + ) + ) + return counts.get(room, 0) + + # `room_id` keeps an up-to-date summary throughout, so its `main` thread is + # always one we found a valid summary for. + first = _send(room_id) + _send(room_id) + + stale_first = _send(stale_room_id) + stale_second = _send(stale_room_id) + _send(stale_room_id) + + # Read one event in each room and rotate, so that both summary rows record + # the receipt they were calculated against. + _read(room_id, first) + _read(stale_room_id, stale_first) + self.get_success(self.store._rotate_notifs()) + + self.assertEqual(_badge(room_id), 1) + self.assertEqual(_badge(stale_room_id), 2) + + # A second receipt, which rotation has not processed yet: `stale_room_id`'s + # summary row no longer matches it, so its count has to be recovered from + # `event_push_actions`. + _read(stale_room_id, stale_second) + + self.assertEqual(_badge(room_id), 1) + self.assertEqual(_badge(stale_room_id), 1) + # A new event, not yet rotated, while the summary row is still stale. + _send(stale_room_id) + + self.assertEqual(_badge(room_id), 1) + self.assertEqual(_badge(stale_room_id), 2) + def test_count_aggregation_threads(self) -> None: """ This is essentially the same test as test_count_aggregation, but adds From e8987e801ecdbdc8ccfa6819db40b2baf25265f8 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 23 Sep 2026 17:11:43 +0100 Subject: [PATCH 2/6] Remove the ability to send state events that are MSC4354 sticky (#20256) Part of: MSC4354 whose Experimental Feature tracking issue is https://github.com/element-hq/synapse/issues/19409 Remove the ability to create sticky (MSC4354) state events as a client This was removed from the MSC in https://github.com/matrix-org/matrix-spec-proposals/commit/d7d1546d3aa2323ec5d97a50c9f61ab92cd9e14b --------- Signed-off-by: Olivier 'reivilibre --- changelog.d/20256.removal | 1 + synapse/rest/client/room.py | 11 +---------- 2 files changed, 2 insertions(+), 10 deletions(-) create mode 100644 changelog.d/20256.removal diff --git a/changelog.d/20256.removal b/changelog.d/20256.removal new file mode 100644 index 0000000000..e258976f5f --- /dev/null +++ b/changelog.d/20256.removal @@ -0,0 +1 @@ +Remove the ability to send state events that are sticky, as this feature was removed from [MSC4354: Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354). \ No newline at end of file diff --git a/synapse/rest/client/room.py b/synapse/rest/client/room.py index 1b8774fde3..4ffb6a89b3 100644 --- a/synapse/rest/client/room.py +++ b/synapse/rest/client/room.py @@ -217,7 +217,6 @@ class RoomStateEventRestServlet(RestServlet): self.clock = hs.get_clock() self._event_serializer = hs.get_event_client_serializer() self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker - self._msc4354_enabled = hs.config.experimental.msc4354_enabled def register(self, http_server: HttpServer) -> None: # /rooms/$roomid/state/$eventtype @@ -339,10 +338,6 @@ class RoomStateEventRestServlet(RestServlet): if requester.app_service_id: origin_server_ts = parse_integer(request, "ts") - sticky_duration_ms: int | None = None - if self._msc4354_enabled: - sticky_duration_ms = parse_integer(request, StickyEvent.QUERY_PARAM_NAME) - delay = _parse_request_for_delayed_event_delay(request) if delay is not None: delay_id = await self.delayed_events_handler.add( @@ -353,7 +348,7 @@ class RoomStateEventRestServlet(RestServlet): origin_server_ts=origin_server_ts, content=content, delay=delay, - sticky_duration_ms=sticky_duration_ms, + sticky_duration_ms=None, ) set_tag("delay_id", delay_id) @@ -381,10 +376,6 @@ class RoomStateEventRestServlet(RestServlet): "room_id": room_id, "sender": requester.user.to_string(), } - if sticky_duration_ms is not None: - event_dict[StickyEvent.EVENT_FIELD_NAME] = StickyEventField( - duration_ms=sticky_duration_ms - ) if state_key is not None: event_dict["state_key"] = state_key From c0b7224e850506c94b9f1594f14585b1547eb363 Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Wed, 23 Sep 2026 18:33:17 +0200 Subject: [PATCH 3/6] Reject `limit_profile_requests_to_users_who_share_rooms` without `require_auth_for_profile_requests` (#20231) As I was working on https://github.com/element-hq/synapse/pull/20218, I noticed what seemed to be an illegal configuration of synapse. - `require_auth_for_profile_requests`: blocks profile requests unless authenticated - `limit_profile_requests_to_users_who_share_rooms`: blocks profile requests unless authenticated user share a room with requested user This, I think, should be an illegal config: ``` require_auth_for_profile_requests = false limit_profile_requests_to_users_who_share_rooms = true ``` As of now, with such a config the shared-room check is never applied: a profile can be requested anonymously, and also by an authenticated user who doesn't share a room. ### Pull Request Checklist * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [x] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --------- Co-authored-by: Olivier 'reivilibre' --- changelog.d/20231.bugfix | 1 + docs/upgrade.md | 17 +++++++++++++ .../configuration/config_documentation.md | 6 +++-- schema/synapse-config.schema.yaml | 7 ++++++ synapse/config/server.py | 14 +++++++++++ synapse/handlers/profile.py | 6 ++--- tests/config/test_server.py | 24 +++++++++++++++++++ 7 files changed, 70 insertions(+), 5 deletions(-) create mode 100644 changelog.d/20231.bugfix diff --git a/changelog.d/20231.bugfix b/changelog.d/20231.bugfix new file mode 100644 index 0000000000..cc9b403e3e --- /dev/null +++ b/changelog.d/20231.bugfix @@ -0,0 +1 @@ +Requires `require_auth_for_profile_requests` to be enabled to enable `limit_profile_requests_to_users_who_share_rooms`. diff --git a/docs/upgrade.md b/docs/upgrade.md index 9eac9c5ebc..1febde5555 100644 --- a/docs/upgrade.md +++ b/docs/upgrade.md @@ -118,6 +118,23 @@ stacking them up. You can monitor the currently running background updates with [the Admin API](usage/administration/admin_api/background_updates.html#status). +# Upgrading to v1.163.0 + +## `limit_profile_requests_to_users_who_share_rooms` now requires `require_auth_for_profile_requests` + +The shared-room check enabled by +[`limit_profile_requests_to_users_who_share_rooms`](usage/configuration/config_documentation.md#limit_profile_requests_to_users_who_share_rooms) +is only applied to authenticated requests, and profile requests are only +authenticated when +[`require_auth_for_profile_requests`](usage/configuration/config_documentation.md#require_auth_for_profile_requests) +is enabled. Enabling the former without the latter therefore never restricted +any profile lookup. + +Synapse will now refuse to start with that combination. If you have +`limit_profile_requests_to_users_who_share_rooms: true` in your configuration, +either also set `require_auth_for_profile_requests: true`, or remove the option to +keep the behaviour you have today. + # Upgrading to v1.161.0 ## Deprecation of `matrix_rtc.livekit_service_url` diff --git a/docs/usage/configuration/config_documentation.md b/docs/usage/configuration/config_documentation.md index f9b76f4f21..d875337c0e 100644 --- a/docs/usage/configuration/config_documentation.md +++ b/docs/usage/configuration/config_documentation.md @@ -311,7 +311,8 @@ presence: --- ### `require_auth_for_profile_requests` -*(boolean)* Whether to require authentication to retrieve profile data (avatars, display names) of other users through the client API. Note that profile data is also available via the federation API, unless `allow_profile_lookup_over_federation` is set to false. Defaults to `false`. +*(boolean)* Whether to require authentication to retrieve profile data (avatars, display names) of other users through the client API. Note that profile data is also available via the federation API, unless `allow_profile_lookup_over_federation` is set to false. +This option must be enabled in order to use `limit_profile_requests_to_users_who_share_rooms`. Defaults to `false`. Example configuration: ```yaml @@ -320,7 +321,8 @@ require_auth_for_profile_requests: true --- ### `limit_profile_requests_to_users_who_share_rooms` -*(boolean)* Use this option to require a user to share a room with another user in order to retrieve their profile information. Only checked on Client-Server requests. Profile requests from other servers should be checked by the requesting server. Defaults to `false`. +*(boolean)* Use this option to require a user to share a room with another user in order to retrieve their profile information. Only checked on Client-Server requests. Profile requests from other servers should be checked by the requesting server. +The check can only be applied to authenticated requests, so this option requires `require_auth_for_profile_requests` to be enabled as well. Synapse will refuse to start if this option is enabled without it. Defaults to `false`. Example configuration: ```yaml diff --git a/schema/synapse-config.schema.yaml b/schema/synapse-config.schema.yaml index d78ca3610b..1a929912a3 100644 --- a/schema/synapse-config.schema.yaml +++ b/schema/synapse-config.schema.yaml @@ -252,6 +252,9 @@ properties: display names) of other users through the client API. Note that profile data is also available via the federation API, unless `allow_profile_lookup_over_federation` is set to false. + + This option must be enabled in order to use + `limit_profile_requests_to_users_who_share_rooms`. default: false examples: - true @@ -262,6 +265,10 @@ properties: order to retrieve their profile information. Only checked on Client-Server requests. Profile requests from other servers should be checked by the requesting server. + + The check can only be applied to authenticated requests, so this option + requires `require_auth_for_profile_requests` to be enabled as well. + Synapse will refuse to start if this option is enabled without it. default: false examples: - true diff --git a/synapse/config/server.py b/synapse/config/server.py index 6c9345e5c2..56ac229c00 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -569,6 +569,20 @@ class ServerConfig(Config): False, ) + # The shared-room check needs to know who is asking, and profile requests + # are only authenticated when `require_auth_for_profile_requests` is set. + # Without it the limit would silently never apply, so refuse to start. + if ( + self.limit_profile_requests_to_users_who_share_rooms + and not self.require_auth_for_profile_requests + ): + raise ConfigError( + "'limit_profile_requests_to_users_who_share_rooms' can only be" + " enforced on authenticated requests, so" + " 'require_auth_for_profile_requests' must also be enabled.", + ("limit_profile_requests_to_users_who_share_rooms",), + ) + # Whether to retrieve and display profile data for a user when they # are invited to a room self.include_profile_data_on_invite = config.get( diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 112ac4c8c5..7521975cc5 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -985,9 +985,9 @@ class ProfileHandler: self, target_user: UserID, requester: UserID | None = None ) -> None: """Checks whether a profile query is allowed. If the - 'require_auth_for_profile_requests' config flag is set to True and a - 'requester' is provided, the query is only allowed if the two users - share a room. + 'limit_profile_requests_to_users_who_share_rooms' config flag is set to + True and a 'requester' is provided, the query is only allowed if the two + users share a room. Args: target_user: The owner of the queried profile. diff --git a/tests/config/test_server.py b/tests/config/test_server.py index f718e20a2a..12c57d33da 100644 --- a/tests/config/test_server.py +++ b/tests/config/test_server.py @@ -229,6 +229,30 @@ class ServerConfigTestCase(unittest.TestCase): with self.assertRaises(ConfigError): _read_config(generate_config(disallowed_value)) + def test_limit_profile_requests_requires_auth(self) -> None: + """ + Test that `limit_profile_requests_to_users_who_share_rooms` can only be + enabled together with `require_auth_for_profile_requests`, as the shared-room + check is only applied to authenticated requests + """ + + def generate_config(limit: bool, require_auth: bool) -> JsonDict: + return { + "limit_profile_requests_to_users_who_share_rooms": limit, + "require_auth_for_profile_requests": require_auth, + } + + _read_config(generate_config(limit=False, require_auth=False)) + _read_config(generate_config(limit=False, require_auth=True)) + _read_config(generate_config(limit=True, require_auth=True)) + + with self.assertRaises(ConfigError): + _read_config(generate_config(limit=True, require_auth=False)) + + # `require_auth_for_profile_requests` defaults to false + with self.assertRaises(ConfigError): + _read_config({"limit_profile_requests_to_users_who_share_rooms": True}) + @parameterized.expand( [ [ From 64f0590705c5669cbed7b6c24fae7f4c4231b7f8 Mon Sep 17 00:00:00 2001 From: Paul Chobert Date: Thu, 24 Sep 2026 10:05:33 +0200 Subject: [PATCH 4/6] MSC4140: cancel a user's delayed events when their account is deactivated (#20247) [MSC4140](https://github.com/matrix-org/matrix-spec-proposals/blob/main/proposals/4140-delayed-events-futures.md), "Account deactivation": > when an account is deactivated, the homeserver MUST cancel that account's delayed events which have not yet been added to a room's event DAG. These cancelled records MAY be removed immediately, including their stored event content, as an exception to the usual finalised-record retention policy. Deactivation currently leaves the user's scheduled delayed events, including their content, in place, and they are still attempted at their scheduled time: | Delayed event scheduled by the user | What happens after deactivation today | | --- | --- | | Message or state event, fires after deactivation has made the user leave the room | The send fails with a 403 (user not in room) and the record is dropped | | Message or state event, fires before deactivation has made the user leave the room (leaving happens room by room in the background) | The event is sent | | `m.room.member` join for themselves in a public room | The deactivated user re-joins the room | ### What changes - Deactivating an account (client or admin API) removes the user's unsent delayed events as its first step, and re-arms the send timer for whatever is scheduled next. - Deactivation can run on a worker, while the send timer lives on the main process, so the cancellation goes through a new replication request. - A delayed event whose send has already started is left alone, as with a normal cancel: it is already on its way into the DAG, and the send path removes its record itself. Suspension and locking are unchanged. ### After #19038 Today a cancelled delayed event is simply deleted, so this PR deletes the user's records too. #19038 changes cancellation to keep the record and mark it as cancelled ("finalised"), so that clients can look up what happened to a delayed event. Once it has landed, deactivation should probably finalise the user's records as cancelled the same way, rather than delete them. The MSC allows either: it permits removing the records immediately, content included, as an exception to the usual retention of finalised records, which is worth doing at least when the user asks to be erased. ### Pull Request Checklist * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [x] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --------- Co-authored-by: Andrew Ferrazzutti Co-authored-by: Andrew Ferrazzutti --- changelog.d/20247.feature | 1 + synapse/handlers/deactivate_account.py | 15 +- synapse/handlers/delayed_events.py | 28 +++ synapse/replication/http/delayed_events.py | 36 ++++ .../storage/databases/main/delayed_events.py | 29 +++ tests/rest/client/test_delayed_events.py | 189 +++++++++++++++++- 6 files changed, 294 insertions(+), 4 deletions(-) create mode 100644 changelog.d/20247.feature diff --git a/changelog.d/20247.feature b/changelog.d/20247.feature new file mode 100644 index 0000000000..4dbffdc07e --- /dev/null +++ b/changelog.d/20247.feature @@ -0,0 +1 @@ +Cancel a user's scheduled [MSC4140](https://github.com/matrix-org/matrix-spec-proposals/pull/4140) delayed events when their account is deactivated. diff --git a/synapse/handlers/deactivate_account.py b/synapse/handlers/deactivate_account.py index 34596ade16..a1aa2a51dc 100644 --- a/synapse/handlers/deactivate_account.py +++ b/synapse/handlers/deactivate_account.py @@ -27,7 +27,13 @@ from synapse.api.errors import SynapseError from synapse.replication.http.deactivate_account import ( ReplicationNotifyAccountDeactivatedServlet, ) -from synapse.types import Codes, Requester, UserID, create_requester +from synapse.types import ( + Codes, + Requester, + UserID, + create_requester, + get_localpart_from_id, +) if TYPE_CHECKING: from synapse.server import HomeServer @@ -47,6 +53,7 @@ class DeactivateAccountHandler: self._room_member_handler = hs.get_room_member_handler() self._identity_handler = hs.get_identity_handler() self._profile_handler = hs.get_profile_handler() + self._delayed_events_handler = hs.get_delayed_events_handler() self._pusher_pool = hs.get_pusherpool() self.user_directory_handler = hs.get_user_directory_handler() self._server_name = hs.hostname @@ -109,6 +116,12 @@ class DeactivateAccountHandler: id_server, ) + # Cancel the user's delayed events (MSC4140). Do this first, so that none + # of them get sent while the rest of the deactivation is in progress. + await self._delayed_events_handler.cancel_all_for_user( + get_localpart_from_id(user_id) + ) + # FIXME: Theoretically there is a race here wherein user resets # password using threepid. diff --git a/synapse/handlers/delayed_events.py b/synapse/handlers/delayed_events.py index ac2da15112..c86194f77c 100644 --- a/synapse/handlers/delayed_events.py +++ b/synapse/handlers/delayed_events.py @@ -28,6 +28,7 @@ from synapse.logging.opentracing import set_tag from synapse.metrics import SERVER_NAME_LABEL, event_processing_positions from synapse.replication.http.delayed_events import ( ReplicationAddedDelayedEventRestServlet, + ReplicationCancelDelayedEventsForUserRestServlet, ) from synapse.storage.databases.main.delayed_events import ( DelayedEventDetails, @@ -129,6 +130,9 @@ class DelayedEventsHandler: ) else: self._repl_client = ReplicationAddedDelayedEventRestServlet.make_client(hs) + self._cancel_all_for_user_client = ( + ReplicationCancelDelayedEventsForUserRestServlet.make_client(hs) + ) @property def _is_master(self) -> bool: @@ -440,6 +444,30 @@ class DelayedEventsHandler: if self._next_send_ts_changed(next_send_ts): self._schedule_next_at_or_none(next_send_ts) + async def cancel_all_for_user(self, user_localpart: str) -> None: + """ + Cancels the scheduled delivery of all delayed events owned by the local user + with the given localpart, e.g. because their account is being deactivated. + + Delayed events that are already being sent are left alone. + + Goes through replication if this is not the main process, as only the + main process handles sending delayed events. + """ + if not self._is_master: + await self._cancel_all_for_user_client( + instance_name=MAIN_PROCESS_INSTANCE_NAME, + user_localpart=user_localpart, + ) + return + + await make_deferred_yieldable(self._initialized_from_db) + + next_send_ts = await self._store.cancel_delayed_events_for_user(user_localpart) + + if self._next_send_ts_changed(next_send_ts): + self._schedule_next_at_or_none(next_send_ts) + async def restart(self, request: SynapseRequest, delay_id: str) -> None: """ Restarts the scheduled delivery of the matching delayed event. diff --git a/synapse/replication/http/delayed_events.py b/synapse/replication/http/delayed_events.py index 26eaf68dae..461c3e044c 100644 --- a/synapse/replication/http/delayed_events.py +++ b/synapse/replication/http/delayed_events.py @@ -58,5 +58,41 @@ class ReplicationAddedDelayedEventRestServlet(ReplicationEndpoint): return 200, {} +class ReplicationCancelDelayedEventsForUserRestServlet(ReplicationEndpoint): + """Cancel all of a local user's delayed events, on behalf of another worker. + + Request format: + + POST /_synapse/replication/cancel_delayed_events_for_user/:user_localpart + + {} + """ + + NAME = "cancel_delayed_events_for_user" + PATH_ARGS = ("user_localpart",) + CACHE = False + + def __init__(self, hs: "HomeServer"): + super().__init__(hs) + + self.handler = hs.get_delayed_events_handler() + + @staticmethod + async def _serialize_payload(user_localpart: str) -> JsonDict: # type: ignore[override] + """ + Args: + user_localpart: The localpart of the local user whose delayed events to cancel. + """ + return {} + + async def _handle_request( # type: ignore[override] + self, request: Request, content: JsonDict, user_localpart: str + ) -> tuple[int, JsonDict]: + await self.handler.cancel_all_for_user(user_localpart) + + return 200, {} + + def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: ReplicationAddedDelayedEventRestServlet(hs).register(http_server) + ReplicationCancelDelayedEventsForUserRestServlet(hs).register(http_server) diff --git a/synapse/storage/databases/main/delayed_events.py b/synapse/storage/databases/main/delayed_events.py index 35f78e3f97..1cd06dc14e 100644 --- a/synapse/storage/databases/main/delayed_events.py +++ b/synapse/storage/databases/main/delayed_events.py @@ -621,6 +621,35 @@ class DelayedEventsStore(SQLBaseStore): "cancel_delayed_state_events", cancel_delayed_state_events_txn ) + async def cancel_delayed_events_for_user( + self, user_localpart: str + ) -> Timestamp | None: + """ + Cancels all delayed events owned by the given local user, i.e. remove them as long as they haven't been processed. + + Args: + user_localpart: The localpart of the user whose delayed events to cancel. + + Returns: The send time of the next delayed event to be sent, if any. + """ + + def cancel_delayed_events_for_user_txn( + txn: LoggingTransaction, + ) -> Timestamp | None: + self.db_pool.simple_delete_txn( + txn, + table="delayed_events", + keyvalues={ + "user_localpart": user_localpart, + "is_processed": False, + }, + ) + return self._get_next_delayed_event_send_ts_txn(txn) + + return await self.db_pool.runInteraction( + "cancel_delayed_events_for_user", cancel_delayed_events_for_user_txn + ) + async def delete_processed_delayed_event(self, delay_id: DelayID) -> None: """ Delete the matching delayed event, as long as it has been marked as processed. diff --git a/tests/rest/client/test_delayed_events.py b/tests/rest/client/test_delayed_events.py index 644b373f36..93e933d991 100644 --- a/tests/rest/client/test_delayed_events.py +++ b/tests/rest/client/test_delayed_events.py @@ -20,15 +20,23 @@ from parameterized import parameterized from twisted.internet.testing import MemoryReactor -from synapse.api.errors import Codes +from synapse.api.errors import Codes, NotFoundError from synapse.rest import admin -from synapse.rest.client import delayed_events, login, room, sync, versions +from synapse.rest.client import ( + account, + delayed_events, + login, + room, + sync, + versions, +) from synapse.server import HomeServer -from synapse.types import JsonDict +from synapse.types import JsonDict, create_requester from synapse.util.clock import Clock from synapse.util.duration import Duration from tests import unittest +from tests.replication._base import BaseMultiWorkerStreamTestCase from tests.server import FakeChannel from tests.unittest import HomeserverTestCase @@ -67,6 +75,7 @@ class DelayedEventsTestCase(HomeserverTestCase): servlets = [ admin.register_servlets, + account.register_servlets, delayed_events.register_servlets, login.register_servlets, room.register_servlets, @@ -765,6 +774,130 @@ class DelayedEventsTestCase(HomeserverTestCase): self._find_sent_delayed_event(self.user1_access_token, delay_id, False) self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + @parameterized.expand((("client_api", False), ("admin_api", True))) + def test_delayed_events_are_cancelled_on_deactivation( + self, _name: str, by_admin: bool + ) -> None: + store = self.hs.get_datastores().main + handler = self.hs.get_delayed_events_handler() + + # user1 schedules a delayed message event and a delayed state event. + state_key = "to_cancel_on_deactivation" + user1_delay_ids = [] + # TODO: Schedule these through the dedicated delayed events endpoint + # instead of the delay query parameter on `/send` and `/state`, once + # https://github.com/element-hq/synapse/pull/19354 has landed. + for path in ( + _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 900), + _get_path_for_delayed_state(self.room_id, _EVENT_TYPE, state_key, 900), + ): + channel = self.make_request( + "POST" if "/send/" in path else "PUT", + path, + {"key": "value"}, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + user1_delay_ids.append(channel.json_body["delay_id"]) + + # user2 schedules a delayed event to be sent later than user1's. + channel = self.make_request( + "POST", + _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 2000), + {"key": "value"}, + self.user2_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + user2_delay_id = channel.json_body["delay_id"] + + self.assertEqual(2, len(self._get_delayed_events())) + + # The next send is scheduled for user1's delayed events. + assert handler._next_delayed_event_call is not None + self.assertLess( + handler._next_delayed_event_call.getTime() - self.reactor.seconds(), 1 + ) + + if by_admin: + self.register_user("admin", "pass", admin=True) + channel = self.make_request( + "POST", + f"/_synapse/admin/v1/deactivate/{self.user1_user_id}", + {}, + self.login("admin", "pass"), + ) + else: + channel = self.make_request( + "POST", + "account/deactivate", + { + "auth": { + "type": "m.login.password", + "user": self.user1_user_id, + "password": "pass", + }, + }, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + + # user1's delayed events are gone... + self.assertListEqual( + [], self.get_success(store.get_all_delayed_events_for_user("user1")) + ) + for delay_id in user1_delay_ids: + self.get_failure( + store.get_delayed_event_for_user(delay_id, "user1"), NotFoundError + ) + + # ...and the next send has been rescheduled for user2's delayed event. + assert handler._next_delayed_event_call is not None + self.assertGreater( + handler._next_delayed_event_call.getTime() - self.reactor.seconds(), 1 + ) + + # Nothing gets sent when user1's delayed events would have timed out. + self.reactor.advance(1) + for delay_id in user1_delay_ids: + self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self.helper.get_state( + self.room_id, + _EVENT_TYPE, + self.user2_access_token, + state_key=state_key, + expect_code=HTTPStatus.NOT_FOUND, + ) + + # user2's delayed event is unaffected, and still gets sent. + self._find_sent_delayed_event(self.user2_access_token, user2_delay_id, False) + self.reactor.advance(1) + for delay_id in user1_delay_ids: + self._find_sent_delayed_event(self.user2_access_token, delay_id, False) + self._find_sent_delayed_event(self.user2_access_token, user2_delay_id, True) + + def test_deactivation_cancels_the_timer_when_no_delayed_events_remain( + self, + ) -> None: + handler = self.hs.get_delayed_events_handler() + + channel = self.make_request( + "POST", + _get_path_for_delayed_send(self.room_id, _EVENT_TYPE, 900), + {}, + self.user1_access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + self.assertIsNotNone(handler._next_delayed_event_call) + + self.get_success( + self.hs.get_deactivate_account_handler().deactivate_account( + self.user1_user_id, + erase_data=False, + requester=create_requester(self.user1_user_id), + ) + ) + self.assertIsNone(handler._next_delayed_event_call) + def _get_delayed_events(self) -> list[JsonDict]: channel = self.make_request( "GET", @@ -839,6 +972,56 @@ class DelayedEventsTestCase(HomeserverTestCase): self.fail("Did not find event with matching delay_id") +class DelayedEventsWorkerTestCase(BaseMultiWorkerStreamTestCase): + """Tests delayed events when some of the work happens off the main process.""" + + servlets = [ + admin.register_servlets, + delayed_events.register_servlets, + login.register_servlets, + room.register_servlets, + sync.register_servlets, + ] + + def default_config(self) -> JsonDict: + config = super().default_config() + config["max_event_delay_duration"] = "24h" + return config + + def test_delayed_events_are_cancelled_on_deactivation_from_worker(self) -> None: + worker_hs = self.make_worker_hs("synapse.app.generic_worker") + store = self.hs.get_datastores().main + handler = self.hs.get_delayed_events_handler() + + user_id = self.register_user("user", "pass") + access_token = self.login("user", "pass") + room_id = self.helper.create_room_as(user_id, tok=access_token) + + channel = self.make_request( + "POST", + _get_path_for_delayed_send(room_id, _EVENT_TYPE, 900), + {}, + access_token, + ) + self.assertEqual(HTTPStatus.OK, channel.code, channel.result) + self.assertIsNotNone(handler._next_delayed_event_call) + + # Deactivate the account on the worker. The main process, which is the + # one to send delayed events, must get to know about it. + self.get_success( + worker_hs.get_deactivate_account_handler().deactivate_account( + user_id, + erase_data=False, + requester=create_requester(user_id), + ) + ) + + self.assertListEqual( + [], self.get_success(store.get_all_delayed_events_for_user("user")) + ) + self.assertIsNone(handler._next_delayed_event_call) + + def _get_path_for_delayed_state( room_id: str, event_type: str, state_key: str, delay_ms: int ) -> str: From 8f6f4e03c18248a9f659464a7b09640bc4a8f906 Mon Sep 17 00:00:00 2001 From: Jason Robinson Date: Thu, 24 Sep 2026 12:58:57 +0300 Subject: [PATCH 5/6] Fix a few readme links regarding ESS (#19070) Didn't add a changelog, is it needed for this? ### Pull Request Checklist * [x] Pull request is based on the develop branch * [x] Pull request includes a [changelog file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog). The entry should: - Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from `EventStore` to `EventWorkerStore`.". - Use markdown where necessary, mostly for `code blocks`. - End with either a period (.) or an exclamation mark (!). - Start with a capital letter. - Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry. * [ ] [Code style](https://element-hq.github.io/synapse/latest/code_style.html) is correct (run the [linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters)) --- README.rst | 2 +- changelog.d/19070.doc | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) create mode 100644 changelog.d/19070.doc diff --git a/README.rst b/README.rst index 4db5d461a8..599dafefb7 100644 --- a/README.rst +++ b/README.rst @@ -59,7 +59,7 @@ below for more useful documentation links. Enterprise quality support for Synapse including SLAs is available as part of an `Element Server Suite (ESS) `_ subscription. -If you are an existing ESS subscriber then you can raise a `support request `_ +If you are an existing ESS subscriber then you can raise a `support request `_ and access the `Element product documentation `_. 🤝 Community support diff --git a/changelog.d/19070.doc b/changelog.d/19070.doc new file mode 100644 index 0000000000..0be0703a60 --- /dev/null +++ b/changelog.d/19070.doc @@ -0,0 +1 @@ +Update ESS support link to point to the new customer portal domain. \ No newline at end of file From 62cb420a0e3aea6ef25274021d91167508d671ef Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Thu, 24 Sep 2026 15:28:06 +0100 Subject: [PATCH 6/6] Send all of a room's MSC4354 Sticky Events down oldschool `/sync` when a user joins it. (#20233) Part of: MSC4354 whose experimental feature tracking issue is https://github.com/element-hq/synapse/issues/19409 Part of: #19662 --------- Signed-off-by: Olivier 'reivilibre --- changelog.d/20233.bugfix | 1 + synapse/api/constants.py | 2 + synapse/handlers/sync.py | 42 +++++++ tests/rest/client/test_sync_sticky_events.py | 120 +++++++++++++++++++ 4 files changed, 165 insertions(+) create mode 100644 changelog.d/20233.bugfix diff --git a/changelog.d/20233.bugfix b/changelog.d/20233.bugfix new file mode 100644 index 0000000000..c9d5a6ee8e --- /dev/null +++ b/changelog.d/20233.bugfix @@ -0,0 +1 @@ +Send all of a room's [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) down legacy `/sync` when a user joins it. diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 17e6a4cfdb..bfdb3ef926 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -502,6 +502,8 @@ class StickyEvent: Maximum number of sticky events to include in /sync. This is the default specified in the MSC. Chosen arbitrarily. + + Does not apply when joining a new room, see https://github.com/matrix-org/matrix-spec-proposals/pull/4354#discussion_r3021907998 """ diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index c66034cd0e..a3ebf8bb1d 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -27,6 +27,7 @@ from typing import ( TYPE_CHECKING, AbstractSet, Any, + Collection, Mapping, Sequence, ) @@ -701,6 +702,37 @@ class SyncHandler: return now_token, sticky_by_room + async def sticky_events_for_newly_joined_rooms( + self, + now_token: StreamToken, + newly_joined_rooms: Collection[str], + ) -> dict[str, list[str]]: + """Get all the sticky events for each newly-joined room the user is in + Args: + now_token: Where the server is currently up to. + newly_joined_rooms: Room IDs of rooms that are newly-joined + Returns: + Dict from room ID to list of sticky event IDs + """ + now = self.clock.time_msec() + with Measure( + self.clock, + name="sticky_events_for_newly_joined_rooms", + server_name=self.server_name, + ): + _, sticky_by_room = await self.store.get_sticky_events_in_rooms( + newly_joined_rooms, + # Since the start of time + from_id=0, + to_id=now_token.sticky_events_key, + now=now, + # Unfortunately, we're meant to return all sticky events in one go + # (we don't have a good alternative). + # See: https://github.com/matrix-org/matrix-spec-proposals/pull/4354#discussion_r3021907998 + limit=None, + ) + return sticky_by_room + async def _load_filtered_recents( self, room_id: str, @@ -2729,6 +2761,16 @@ class SyncHandler: newly_joined_rooms = room_changes.newly_joined_rooms newly_left_rooms = room_changes.newly_left_rooms + if self.hs_config.experimental.msc4354_enabled and newly_joined_rooms: + # If we have any newly-joined rooms, load all sticky events from them. + # We then send down all the sticky events, including historical ones, + # rather than just the ones that are new since the `since` token. + sticky_by_room.update( + await self.sticky_events_for_newly_joined_rooms( + sync_result_builder.now_token, newly_joined_rooms + ) + ) + # 4. We need to apply further processing to `room_entries` (rooms considered # joined or archived). async def handle_room_entries(room_entry: "RoomSyncResultBuilder") -> None: diff --git a/tests/rest/client/test_sync_sticky_events.py b/tests/rest/client/test_sync_sticky_events.py index 2e52a0ad16..095d5c1022 100644 --- a/tests/rest/client/test_sync_sticky_events.py +++ b/tests/rest/client/test_sync_sticky_events.py @@ -13,6 +13,7 @@ import json import sqlite3 from dataclasses import dataclass +from http import HTTPStatus from unittest.mock import patch from urllib.parse import quote @@ -351,6 +352,125 @@ class SyncStickyEventsTestCase(unittest.HomeserverTestCase): f"Sticky event from ignored user {sticky_event_id} should not be in sticky section", ) + def test_newly_joined_room_gets_all_sticky_events(self) -> None: + """ + Test that joining a room delivers that room's whole sticky event backlog, + even though those sticky events are behind the client's sync position. + + > When the user joins a room, the server MUST include all unexpired sticky + > events for that room in their subsequent sync response. + > — https://github.com/matrix-org/matrix-spec-proposals/blob/d42301b17859d046953563dc7f046edd51789a12/proposals/4354-sticky-events.md#sync-api-changes + """ + # A filter that excludes message events, so that the sticky events come down + # the sticky section rather than the timeline section. + filter_json = json.dumps( + {"room": {"timeline": {"types": ["io.element.example"]}}} + ) + + sticky_event_id = self.helper.send_sticky_event( + self.room_id, + EventTypes.Message, + duration=Duration(minutes=1), + content={"body": "sticky message", "msgtype": "m.text"}, + tok=self.token, + )["event_id"] + + # user2 syncs before joining the room, which moves their sticky events stream + # position past the sticky event above. + user2_id = self.register_user("user2", "pass") + user2_token = self.login(user2_id, "pass") + channel = self.make_request( + "GET", + f"/sync?filter={quote(filter_json)}", + access_token=user2_token, + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + since = channel.json_body["next_batch"] + + # user2 joins the room + self.helper.join(self.room_id, user2_id, tok=user2_token) + + # The next (incremental) sync gives user2 the historical sticky event. + channel = self.make_request( + "GET", + f"/sync?filter={quote(filter_json)}&since={since}", + access_token=user2_token, + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + sticky_events = channel.json_body["rooms"]["join"][self.room_id][ + "msc4354_sticky" + ]["events"] + self.assertEqual( + {sticky_event["event_id"] for sticky_event in sticky_events}, + {sticky_event_id}, + ) + + def test_displayname_change_is_not_newly_joined_room(self) -> None: + """ + Test that a displayname change (`m.room.member` from `join` to `join`) + does not cause the room to be treated as newly joined. + That is to say, we do NOT send down all sticky events again in this case. + """ + # A filter that excludes message events, so that the sticky events come down + # the sticky section rather than the timeline section. + # We include `m.room.member` so that we witness the membership change + # and receive a sync entry for the room. + filter_json = json.dumps({"room": {"timeline": {"types": [EventTypes.Member]}}}) + + sticky_event_id = self.helper.send_sticky_event( + self.room_id, + EventTypes.Message, + duration=Duration(minutes=1), + content={"body": "sticky message", "msgtype": "m.text"}, + tok=self.token, + )["event_id"] + + # user2 joins the room and syncs, getting the sticky event and a `since` + # token past it. + user2_id = self.register_user("user2", "pass") + user2_token = self.login(user2_id, "pass") + self.helper.join(self.room_id, user2_id, tok=user2_token) + channel = self.make_request( + "GET", + f"/sync?filter={quote(filter_json)}", + access_token=user2_token, + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + sticky_events = channel.json_body["rooms"]["join"][self.room_id][ + "msc4354_sticky" + ]["events"] + self.assertEqual( + [sticky_event["event_id"] for sticky_event in sticky_events], + [sticky_event_id], + ) + since = channel.json_body["next_batch"] + + # user2 changes their display name, sending `m.room.member` + self.helper.send_state( + self.room_id, + EventTypes.Member, + { + "membership": "join", + "displayname": "some_per_room_name", + }, + tok=user2_token, + state_key=user2_id, + ) + + # The next (incremental) sync must not transmit the sticky event again. + channel = self.make_request( + "GET", + f"/sync?filter={quote(filter_json)}&since={since}", + access_token=user2_token, + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + room_entry = channel.json_body["rooms"]["join"][self.room_id] + self.assertNotIn( + "msc4354_sticky", + room_entry, + f"Display name change should not resend the room's sticky events: {room_entry}", + ) + def test_history_visibility_bypass_for_sticky_events(self) -> None: """ Test that joined users can see sticky events even when history visibility