Merge branch 'develop' into rei/se_sharded_tok

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) <https://element.io/pricing>`_ subscription.
 
-If you are an existing ESS subscriber then you can raise a `support request <https://ems.element.io/support>`_
+If you are an existing ESS subscriber then you can raise a `support request <https://customer.element.io/support>`_
 and access the `Element product documentation <https://docs.element.io>`_.
 
 🤝 Community support

changelog.d/19070.doc

@@ -0,0 +1 @@
+Update ESS support link to point to the new customer portal domain.

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`.

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.

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.

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.

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).

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`

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

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

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
     """
 
 

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(

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.
 

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.

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.

synapse/handlers/sync.py

@@ -27,6 +27,7 @@ from typing import (
     TYPE_CHECKING,
     AbstractSet,
     Any,
+    Collection,
     Mapping,
     Sequence,
 )
@@ -707,6 +708,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_token=MultiWriterStreamToken(stream=0),
+                to_token=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,
@@ -2735,6 +2767,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:

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)

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

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.

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

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(
         [
             [

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:

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

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