changelog

As 3.13 has now been released, recent 3.12 releases should be stable enough.

CHANGES.md

@@ -1,3 +1,63 @@
+# Synapse 1.118.0 (2024-10-29)
+
+No significant changes since 1.118.0rc1.
+
+### Python 3.8 support will be dropped in the next release
+
+Python 3.8 is now [end-of-life](https://devguide.python.org/versions/). As per our [Deprecation Policy for Platform Dependencies](https://element-hq.github.io/synapse/latest/deprecation_policy.html#policy), Synapse will be dropping support for Python 3.8 in the next release; Synapse 1.119.0.
+
+Synapse 1.118.x will be the final release to support Python 3.8. If you are running Synapse with Python 3.8, please upgrade before the 1.119.0 release, due in less than one month.
+
+### Python 3.13 and PostgreSQL 17 support
+
+On the other end of the spectrum, Synapse 1.118.0 is the first release to support [Python 3.13](https://www.python.org/downloads/release/python-3130/)! [PostgreSQL 17](https://www.postgresql.org/about/news/postgresql-17-released-2936/) is also supported as of this release.
+
+
+# Synapse 1.118.0rc1 (2024-10-22)
+
+### Features
+
+- Added the `display_name_claim` option to the JWT configuration. This option allows specifying the claim key that contains the user's display name in the JWT payload. ([\#17708](https://github.com/element-hq/synapse/issues/17708))
+- Implement [MSC4210](https://github.com/matrix-org/matrix-spec-proposals/pull/4210): Remove legacy mentions. Contributed by @tulir @ Beeper. ([\#17783](https://github.com/element-hq/synapse/issues/17783))
+
+### Bugfixes
+
+- Fix saving of PNG thumbnails, when the original image is in the CMYK color space. ([\#17736](https://github.com/element-hq/synapse/issues/17736))
+- Fix bug with sliding sync where the server would not return state that was added to the `required_state` config. ([\#17785](https://github.com/element-hq/synapse/issues/17785), [\#17805](https://github.com/element-hq/synapse/issues/17805))
+- Fix a bug in [MSC4186](https://github.com/matrix-org/matrix-spec-proposals/pull/4186) Sliding Sync that would cause rooms to stay forgotten and hidden even after rejoining. ([\#17835](https://github.com/element-hq/synapse/issues/17835))
+
+### Improved Documentation
+
+- Clarify when the `user_may_invite` and `user_may_send_3pid_invite` module callbacks are called. ([\#17627](https://github.com/element-hq/synapse/issues/17627))
+- Correct documentation to refer to the `--config-path` argument instead of `--config-file`. ([\#17802](https://github.com/element-hq/synapse/issues/17802))
+- Fix typo in `target_cache_memory_usage` docs. ([\#17825](https://github.com/element-hq/synapse/issues/17825))
+
+### Internal Changes
+
+- Slight optimization when fetching state/events for Sliding Sync. ([\#17718](https://github.com/element-hq/synapse/issues/17718))
+- Add Python 3.13 and Postgres 17 to the test matrix. ([\#17752](https://github.com/element-hq/synapse/issues/17752))
+- Test github token before running release script steps. ([\#17803](https://github.com/element-hq/synapse/issues/17803))
+- Build debian packages for new Ubuntu versions, and stop building for no longer supported versions. ([\#17824](https://github.com/element-hq/synapse/issues/17824))
+- Enable the `.org.matrix.msc4028.encrypted_event` push rule by default in accordance with [MSC4028](https://github.com/matrix-org/matrix-spec-proposals/pull/4028). Note that the corresponding experimental feature must still be switched on for this push rule to have any effect. ([\#17826](https://github.com/element-hq/synapse/issues/17826))
+- Fix some typing issues uncovered by upgrading mypy to 1.11.x. ([\#17842](https://github.com/element-hq/synapse/issues/17842))
+
+
+
+### Updates to locked dependencies
+
+* Bump mypy from 1.10.1 to 1.11.2. ([\#17842](https://github.com/element-hq/synapse/issues/17842))
+* Bump mypy-zope from 1.0.5 to 1.0.7. ([\#17827](https://github.com/element-hq/synapse/issues/17827))
+* Bump phonenumbers from 8.13.46 to 8.13.47. ([\#17797](https://github.com/element-hq/synapse/issues/17797))
+* Bump psycopg2 from 2.9.9 to 2.9.10. ([\#17843](https://github.com/element-hq/synapse/issues/17843))
+* Bump ruff from 0.6.8 to 0.6.9. ([\#17794](https://github.com/element-hq/synapse/issues/17794))
+* Bump sentry-sdk from 2.14.0 to 2.15.0. ([\#17795](https://github.com/element-hq/synapse/issues/17795))
+* Bump sentry-sdk from 2.15.0 to 2.16.0. ([\#17829](https://github.com/element-hq/synapse/issues/17829))
+* Bump sentry-sdk from 2.16.0 to 2.17.0. ([\#17844](https://github.com/element-hq/synapse/issues/17844))
+* Bump sigstore/cosign-installer from 3.6.0 to 3.7.0. ([\#17798](https://github.com/element-hq/synapse/issues/17798))
+* Bump tomli from 2.0.1 to 2.0.2. ([\#17796](https://github.com/element-hq/synapse/issues/17796))
+* Bump types-requests from 2.32.0.20240914 to 2.32.0.20241016. ([\#17841](https://github.com/element-hq/synapse/issues/17841))
+* Bump types-setuptools from 75.1.0.20240917 to 75.1.0.20241014. ([\#17828](https://github.com/element-hq/synapse/issues/17828))
+
 # Synapse 1.117.0 (2024-10-15)
 
 No significant changes since 1.117.0rc1.

changelog.d/17627.doc

@@ -1 +0,0 @@
-Clarify when the `user_may_invite` and `user_may_send_3pid_invite` module callbacks are called.

changelog.d/17708.feature

@@ -1 +0,0 @@
-Added the `display_name_claim` option to the JWT configuration. This option allows specifying the claim key that contains the user's display name in the JWT payload.

changelog.d/17718.misc

@@ -1 +0,0 @@
-Slight optimization when fetching state/events for Sliding Sync.

changelog.d/17736.bugfix

@@ -1 +0,0 @@
-Fix saving of PNG thumbnails, when the original image is in the CMYK color space.

changelog.d/17752.misc

@@ -1 +0,0 @@
-Add Python 3.13 and Postgres 17 to the test matrix.

changelog.d/17783.feature

@@ -1 +0,0 @@
-Implement [MSC4210](https://github.com/matrix-org/matrix-spec-proposals/pull/4210): Remove legacy mentions. Contributed by @tulir @ Beeper.

changelog.d/17785.bugfix

@@ -1 +0,0 @@
-Fix bug with sliding sync where the server would not return state that was added to the `required_state` config.

changelog.d/17802.doc

@@ -1 +0,0 @@
-Correct documentation to refer to the `--config-path` argument instead of `--config-file`.

changelog.d/17803.misc

@@ -1 +0,0 @@
-Test github token before running release script steps.

changelog.d/17805.bugfix

@@ -1 +0,0 @@
-Fix bug with sliding sync where the server would not return state that was added to the `required_state` config.

changelog.d/17824.misc

@@ -1 +0,0 @@
-Build debian packages for new Ubuntu versions, and stop building for no longer supported versions.

changelog.d/17825.doc

@@ -1 +0,0 @@
-Fix typo in `target_cache_memory_usage` docs.

changelog.d/17826.misc

@@ -1 +0,0 @@
-Enable the `.org.matrix.msc4028.encrypted_event` push rule by default in accordance with [MSC4028](https://github.com/matrix-org/matrix-spec-proposals/pull/4028). Note that the corresponding experimental feature must still be switched on for this push rule to have any effect.

changelog.d/17830.misc

@@ -0,0 +1 @@
+Include the destination in the error of 'Destination mismatch' on federation requests.

changelog.d/17835.bugfix

@@ -1 +0,0 @@
-Fix a bug in [MSC4186](https://github.com/matrix-org/matrix-spec-proposals/pull/4186) Sliding Sync that would cause rooms to stay forgotten and hidden even after rejoining.

changelog.d/17842.misc

@@ -1 +0,0 @@
-Fix some typing issues uncovered by upgrading mypy to 1.11.x.

changelog.d/17847.bugfix

@@ -0,0 +1,2 @@
+Fix a bug in the admin redact endpoint where the background task would not run if a worker was specified in
+the config option `run_background_tasks_on`.

changelog.d/17861.bugfix

@@ -0,0 +1 @@
+Fix detection when the built Rust library was outdated when using source installations.

changelog.d/17887.misc

@@ -0,0 +1 @@
+Bump the default Python version in the Synapse Dockerfile from 3.11 -> 3.12.

debian/changelog

@@ -1,3 +1,15 @@
+matrix-synapse-py3 (1.118.0) stable; urgency=medium
+
+  * New Synapse release 1.118.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 29 Oct 2024 15:29:53 +0100
+
+matrix-synapse-py3 (1.118.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.118.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 22 Oct 2024 11:48:14 +0100
+
 matrix-synapse-py3 (1.117.0) stable; urgency=medium
 
   * New Synapse release 1.117.0.

docker/Dockerfile

@@ -20,7 +20,7 @@
 # `poetry export | pip install -r /dev/stdin`, but beware: we have experienced bugs in
 # in `poetry export` in the past.
 
-ARG PYTHON_VERSION=3.11
+ARG PYTHON_VERSION=3.12
 
 ###
 ### Stage 0: generate requirements.txt

docs/admin_api/user_admin_api.md

@@ -1365,6 +1365,9 @@ _Added in Synapse 1.72.0._
 
 ## Redact all the events of a user
 
+This endpoint allows an admin to redact the events of a given user. There are no restrictions on redactions for a 
+local user. By default, we puppet the user who sent the message to redact it themselves. Redactions for non-local users are issued using the admin user, and will fail in rooms where the admin user is not admin/does not have the specified power level to issue redactions. 
+
 The API is 
 ```
 POST /_synapse/admin/v1/user/$user_id/redact

pyproject.toml

@@ -97,7 +97,7 @@ module-name = "synapse.synapse_rust"
 
 [tool.poetry]
 name = "matrix-synapse"
-version = "1.117.0"
+version = "1.118.0"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "AGPL-3.0-or-later"

synapse/federation/transport/server/_base.py

@@ -113,7 +113,7 @@ class Authenticator:
                 ):
                     raise AuthenticationError(
                         HTTPStatus.UNAUTHORIZED,
-                        "Destination mismatch in auth header",
+                        f"Destination mismatch in auth header, received: {destination!r}",
                         Codes.UNAUTHORIZED,
                     )
         if (

synapse/handlers/admin.py

@@ -73,6 +73,8 @@ class AdminHandler:
             self._redact_all_events, REDACT_ALL_EVENTS_ACTION_NAME
         )
 
+        self.hs = hs
+
     async def get_redact_task(self, redact_id: str) -> Optional[ScheduledTask]:
         """Get the current status of an active redaction process
 
@@ -423,8 +425,10 @@ class AdminHandler:
         user_id = task.params.get("user_id")
         assert user_id is not None
 
+        # puppet the user if they're ours, otherwise use admin to redact
         requester = create_requester(
-            user_id, authenticated_entity=admin.user.to_string()
+            user_id if self.hs.is_mine_id(user_id) else admin.user.to_string(),
+            authenticated_entity=admin.user.to_string(),
         )
 
         reason = task.params.get("reason")

synapse/handlers/sync.py

@@ -1397,7 +1397,6 @@ class SyncHandler:
             timeline_contains=timeline_state,
             timeline_start=state_at_timeline_start,
             timeline_end=state_at_timeline_end,
-            previous_timeline_start={},
             previous_timeline_end={},
             lazy_load_members=lazy_load_members,
         )
@@ -1536,17 +1535,6 @@ class SyncHandler:
             await_full_state=await_full_state,
         )
 
-        state_at_previous_sync_start = (
-            {}
-            if since_token.prev_batch is None
-            else await self._state_storage_controller.get_state_ids_at(
-                room_id,
-                stream_position=since_token.prev_batch,
-                state_filter=state_filter,
-                await_full_state=await_full_state,
-            )
-        )
-
         state_at_timeline_end = await self._state_storage_controller.get_state_ids_at(
             room_id,
             stream_position=end_token,
@@ -1558,7 +1546,6 @@ class SyncHandler:
             timeline_contains=timeline_state,
             timeline_start=state_at_timeline_start,
             timeline_end=state_at_timeline_end,
-            previous_timeline_start=state_at_previous_sync_start,
             previous_timeline_end=state_at_previous_sync,
             lazy_load_members=lazy_load_members,
         )
@@ -1978,7 +1965,7 @@ class SyncHandler:
         # this is due to some of the underlying streams not supporting the ability
         # to query up to a given point.
         # Always use the `now_token` in `SyncResultBuilder`
-        now_token = self.event_sources.get_current_token(prev_batch=since_token)
+        now_token = self.event_sources.get_current_token()
         log_kv({"now_token": now_token})
 
         # Since we fetched the users room list before calculating the `now_token` (see
@@ -2993,7 +2980,6 @@ def _calculate_state(
     timeline_contains: StateMap[str],
     timeline_start: StateMap[str],
     timeline_end: StateMap[str],
-    previous_timeline_start: StateMap[str],
     previous_timeline_end: StateMap[str],
     lazy_load_members: bool,
 ) -> StateMap[str]:
@@ -3021,7 +3007,6 @@ def _calculate_state(
 
     timeline_end_ids = set(timeline_end.values())
     timeline_start_ids = set(timeline_start.values())
-    previous_timeline_start_ids = set(previous_timeline_start.values())
     previous_timeline_end_ids = set(previous_timeline_end.values())
     timeline_contains_ids = set(timeline_contains.values())
 
@@ -3097,7 +3082,7 @@ def _calculate_state(
 
     state_ids = (
         (timeline_end_ids | timeline_start_ids)
-        - (previous_timeline_end_ids | previous_timeline_start_ids)
+        - previous_timeline_end_ids
         - timeline_contains_ids
     )
 

synapse/server.py

@@ -249,6 +249,7 @@ class HomeServer(metaclass=abc.ABCMeta):
     """
 
     REQUIRED_ON_BACKGROUND_TASK_STARTUP = [
+        "admin",
         "account_validity",
         "auth",
         "deactivate_account",

synapse/streams/events.py

@@ -77,7 +77,7 @@ class EventSources:
         self.store = hs.get_datastores().main
         self._instance_name = hs.get_instance_name()
 
-    def get_current_token(self, prev_batch: StreamToken = None) -> StreamToken:
+    def get_current_token(self) -> StreamToken:
         push_rules_key = self.store.get_max_push_rules_stream_id()
         to_device_key = self.store.get_to_device_stream_token()
         device_list_key = self.store.get_device_stream_token()
@@ -97,7 +97,6 @@ class EventSources:
             # Groups key is unused.
             groups_key=0,
             un_partial_stated_rooms_key=un_partial_stated_rooms_key,
-            prev_batch=prev_batch,
         )
         return token
 

synapse/types/__init__.py

@@ -980,30 +980,16 @@ class StreamToken:
     groups_key: int
     un_partial_stated_rooms_key: int
 
-    prev_batch: Optional["StreamToken"] = None
-    _BATCH_SEPARATOR = "~"
-
     _SEPARATOR = "_"
     START: ClassVar["StreamToken"]
 
     @classmethod
     @cancellable
-    async def from_string(
-        cls, store: "DataStore", string: str, prev_batch: Optional["StreamToken"] = None
-    ) -> "StreamToken":
+    async def from_string(cls, store: "DataStore", string: str) -> "StreamToken":
         """
         Creates a RoomStreamToken from its textual representation.
         """
         try:
-            if string.count(cls._BATCH_SEPARATOR) == 1:
-                # We have a prev_token
-                batches = string.split(cls._BATCH_SEPARATOR)
-                prev_batch = await StreamToken.from_string(store, batches[1])
-                batch = await StreamToken.from_string(
-                    store, batches[0], prev_batch=prev_batch
-                )
-                return batch
-
             keys = string.split(cls._SEPARATOR)
             while len(keys) < len(attr.fields(cls)):
                 # i.e. old token from before receipt_key
@@ -1020,7 +1006,6 @@ class StreamToken:
                 device_list_key,
                 groups_key,
                 un_partial_stated_rooms_key,
-                prev_batch,
             ) = keys
 
             return cls(
@@ -1040,34 +1025,24 @@ class StreamToken:
         except Exception:
             raise SynapseError(400, "Invalid stream token")
 
-    async def to_string(
-        self, store: "DataStore", include_prev_batch: bool = True
-    ) -> str:
-        if include_prev_batch and self.prev_batch:
-            return self._BATCH_SEPARATOR.join(
-                [
-                    await self.to_string(store, include_prev_batch=False),
-                    await self.prev_batch.to_string(store, include_prev_batch=False),
-                ]
-            )
-        else:
-            return self._SEPARATOR.join(
-                [
-                    await self.room_key.to_string(store),
-                    str(self.presence_key),
-                    str(self.typing_key),
-                    await self.receipt_key.to_string(store),
-                    str(self.account_data_key),
-                    str(self.push_rules_key),
-                    str(self.to_device_key),
-                    str(self.device_list_key),
-                    # Note that the groups key is no longer used, but it is still
-                    # serialized so that there will not be confusion in the future
-                    # if additional tokens are added.
-                    str(self.groups_key),
-                    str(self.un_partial_stated_rooms_key),
-                ]
-            )
+    async def to_string(self, store: "DataStore") -> str:
+        return self._SEPARATOR.join(
+            [
+                await self.room_key.to_string(store),
+                str(self.presence_key),
+                str(self.typing_key),
+                await self.receipt_key.to_string(store),
+                str(self.account_data_key),
+                str(self.push_rules_key),
+                str(self.to_device_key),
+                str(self.device_list_key),
+                # Note that the groups key is no longer used, but it is still
+                # serialized so that there will not be confusion in the future
+                # if additional tokens are added.
+                str(self.groups_key),
+                str(self.un_partial_stated_rooms_key),
+            ]
+        )
 
     @property
     def room_stream_id(self) -> int:

synapse/util/rust.py

@@ -19,9 +19,12 @@
 #
 #
 
+import json
 import os
-import sys
+import urllib.parse
 from hashlib import blake2b
+from importlib.metadata import Distribution, PackageNotFoundError
+from typing import Optional
 
 import synapse
 from synapse.synapse_rust import get_rust_file_digest
@@ -32,22 +35,17 @@ def check_rust_lib_up_to_date() -> None:
     be rebuilt.
     """
 
-    if not _dist_is_editable():
-        return
-
-    synapse_dir = os.path.dirname(synapse.__file__)
-    synapse_root = os.path.abspath(os.path.join(synapse_dir, ".."))
-
-    # Double check we've not gone into site-packages...
-    if os.path.basename(synapse_root) == "site-packages":
-        return
-
-    # ... and it looks like the root of a python project.
-    if not os.path.exists("pyproject.toml"):
-        return
+    # Get the location of the editable install.
+    synapse_root = get_synapse_source_directory()
+    if synapse_root is None:
+        return None
 
     # Get the hash of all Rust source files
-    hash = _hash_rust_files_in_directory(os.path.join(synapse_root, "rust", "src"))
+    rust_path = os.path.join(synapse_root, "rust", "src")
+    if not os.path.exists(rust_path):
+        return None
+
+    hash = _hash_rust_files_in_directory(rust_path)
 
     if hash != get_rust_file_digest():
         raise Exception("Rust module outdated. Please rebuild using `poetry install`")
@@ -82,10 +80,55 @@ def _hash_rust_files_in_directory(directory: str) -> str:
     return hasher.hexdigest()
 
 
-def _dist_is_editable() -> bool:
-    """Is distribution an editable install?"""
-    for path_item in sys.path:
-        egg_link = os.path.join(path_item, "matrix-synapse.egg-link")
-        if os.path.isfile(egg_link):
-            return True
-    return False
+def get_synapse_source_directory() -> Optional[str]:
+    """Try and find the source directory of synapse for editable installs (like
+    those used in development).
+
+    Returns None if not an editable install (or otherwise can't find the source
+    directory).
+    """
+
+    # Try and find the installed matrix-synapse package.
+    try:
+        package = Distribution.from_name("matrix-synapse")
+    except PackageNotFoundError:
+        # The package is not found, so it's not installed and so must be being
+        # pulled out from a local directory (usually the current one).
+        synapse_dir = os.path.dirname(synapse.__file__)
+        synapse_root = os.path.abspath(os.path.join(synapse_dir, ".."))
+
+        # Double check we've not gone into site-packages...
+        if os.path.basename(synapse_root) == "site-packages":
+            return None
+
+        # ... and it looks like the root of a python project.
+        if not os.path.exists("pyproject.toml"):
+            return None
+
+        return synapse_root
+
+    # Read the `direct_url.json` metadata for the package. This won't exist for
+    # packages installed via a repository/etc.
+    # c.f. https://packaging.python.org/en/latest/specifications/direct-url/
+    direct_url_json = package.read_text("direct_url.json")
+    if direct_url_json is None:
+        return None
+
+    # c.f. https://packaging.python.org/en/latest/specifications/direct-url/ for
+    # the format
+    direct_url_dict: dict = json.loads(direct_url_json)
+
+    # `url` must exist as a key, and point to where we fetched the repo from.
+    project_url = urllib.parse.urlparse(direct_url_dict["url"])
+
+    # If its not a local file then we must have built the rust libs either a)
+    # after we downloaded the package, or b) we built the download wheel.
+    if project_url.scheme != "file":
+        return None
+
+    # And finally if its not an editable install then the files can't have
+    # changed since we installed the package.
+    if not direct_url_dict.get("dir_info", {}).get("editable", False):
+        return None
+
+    return project_url.path

tests/handlers/test_sync.py

@@ -710,293 +710,9 @@ class SyncTestCase(tests.unittest.HomeserverTestCase):
             [e.event_id for e in room_sync.timeline.events],
             [e4_event, e5_event],
         )
-
-    def test_state_after_on_branches_winner_at_end_of_timeline(self) -> None:
-        r"""Test `state` and `state_after` where not all information is in `state` + `timeline`.
-
-            -----|---------- initial sync
-                 |
-        unrelated state event
-                 |
-                 S1
-            -----|---------- incremental sync 1
-               ↗    ↖
-              |      S2
-            --|------|------ incremental sync 2
-              E3     E4
-            --|------|------ incremental sync 3
-              |      |
-               \    ↗        S2 wins
-                 E5
-            -----|---------- incremental sync 4
-
-        The "interesting" sync is sync 3. At the end of sync 3 the server doesn't know which branch will win.
-
-        """
-        alice = self.register_user("alice", "password")
-        alice_tok = self.login(alice, "password")
-        alice_requester = create_requester(alice)
-        room_id = self.helper.create_room_as(alice, is_public=True, tok=alice_tok)
-
-        # Do an initial sync to get a known starting point.
-        initial_sync_result = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-            )
-        )
-
-        # Send an unrelated state event which doesn't change across the branches
-        unrelated_state_event = self.helper.send_state(
-            room_id, "m.something.else", {"node": "S1"}, tok=alice_tok
-        )["event_id"]
-
-        # Send S1
-        s1_event = self.helper.send_state(
-            room_id, "m.call.member", {"node": "S1"}, tok=alice_tok
-        )["event_id"]
-
-        # Incremental sync 1
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=initial_sync_result.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [unrelated_state_event, s1_event],
-        )
-
-        # Send S2 -> S1
-        s2_event = self.helper.send_state(
-            room_id, "m.call.member", {"node": "S2"}, tok=alice_tok
-        )["event_id"]
-
-        # Incremental sync 2
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=incremental_sync.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [s2_event],
-        )
-
-        # Send two regular events on different branches:
-        # E3 -> S1
-        # E4 -> S2
-        with self._patch_get_latest_events([s1_event]):
-            e3_event = self.helper.send(room_id, "E3", tok=alice_tok)["event_id"]
-        with self._patch_get_latest_events([s2_event]):
-            e4_event = self.helper.send(room_id, "E4", tok=alice_tok)["event_id"]
-
-        # Incremental sync 3
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=incremental_sync.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [
-                e3_event,
-                e4_event,
-            ],  # We have two events from different timelines neither of which are state events
-        )
-
-        # Send E5 which resolves the branches
-        e5_event = self.helper.send(room_id, "E5", tok=alice_tok)["event_id"]
-
-        # Incremental sync 4
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=incremental_sync.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [e5_event],
-        )
-        # FIXED: S2 is the winning state event but and the last that the client saw!
-
-    def test_state_after_on_branches_winner_at_start_of_timeline(self) -> None:
-        r"""Test `state` and `state_after` where not all information is in `state` + `timeline`.
-
-        -----|---------- initial sync
-             |
-             S1
-        -----|---------- incremental sync 1
-           ↗    ↖
-          |      S2
-        --|------|------ incremental sync 2
-          S3     E4
-        --|------|------ incremental sync 3
-          |      |
-           ↖    /        S3 wins
-             E5
-        -----|---------- incremental sync 4
-
-        The "interesting" sync is sync 3. At the end of sync 3 the server doesn't know which branch will win.
-
-        """
-        alice = self.register_user("alice", "password")
-        alice_tok = self.login(alice, "password")
-        alice_requester = create_requester(alice)
-        room_id = self.helper.create_room_as(alice, is_public=True, tok=alice_tok)
-
-        # Do an initial sync to get a known starting point.
-        initial_sync_result = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-            )
-        )
-
-        # Send an unrelated state event which doesn't change across the branches
-        unrelated_state_event = self.helper.send_state(
-            room_id, "m.something.else", {"node": "S1"}, tok=alice_tok
-        )["event_id"]
-
-        # Send S1
-        s1_event = self.helper.send_state(
-            room_id, "m.call.member", {"node": "S1"}, tok=alice_tok
-        )["event_id"]
-
-        # Incremental sync 1
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=initial_sync_result.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [unrelated_state_event, s1_event],
-        )
-
-        # Send S2 -> S1
-        s2_event = self.helper.send_state(
-            room_id, "m.call.member", {"node": "S2"}, tok=alice_tok
-        )["event_id"]
-
-        # Incremental sync 2
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=incremental_sync.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [s2_event],
-        )
-
-        # Send two events on different branches:
-        # S3 -> S1
-        # E4 -> S2
-        with self._patch_get_latest_events([s1_event]):
-            s3_event = self.helper.send_state(
-                room_id, "m.call.member", {"node": "S3"}, tok=alice_tok
-            )["event_id"]
-        with self._patch_get_latest_events([s2_event]):
-            e4_event = self.helper.send(room_id, "E4", tok=alice_tok)["event_id"]
-
-        # Incremental sync 3
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=incremental_sync.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
-        self.assertEqual(room_sync.state, {})
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [
-                s3_event,
-                e4_event,
-            ],  # We have two events from different timelines
-        )
-
-        # Send E5 which resolves the branches with S3 winning
-        e5_event = self.helper.send(room_id, "E5", tok=alice_tok)["event_id"]
-
-        # Incremental sync 4
-        incremental_sync = self.get_success(
-            self.sync_handler.wait_for_sync_for_user(
-                alice_requester,
-                generate_sync_config(alice),
-                sync_version=SyncVersion.SYNC_V2,
-                request_key=generate_request_key(),
-                since_token=incremental_sync.next_batch,
-            )
-        )
-        room_sync = incremental_sync.joined[0]
-
-        self.assertEqual(room_sync.room_id, room_id)
         self.assertEqual(
             [e.event_id for e in room_sync.state.values()],
-            [s3_event],  # S3 is the winning state event
-        )
-        self.assertEqual(
-            [e.event_id for e in room_sync.timeline.events],
-            [e5_event],
+            [s2_event],
         )
 
     @parameterized.expand(

tests/rest/admin/test_user.py

@@ -23,6 +23,7 @@ import hashlib
 import hmac
 import json
 import os
+import time
 import urllib.parse
 from binascii import unhexlify
 from http import HTTPStatus
@@ -56,6 +57,7 @@ from synapse.types import JsonDict, UserID, create_requester
 from synapse.util import Clock
 
 from tests import unittest
+from tests.replication._base import BaseMultiWorkerStreamTestCase
 from tests.test_utils import SMALL_PNG
 from tests.unittest import override_config
 
@@ -5127,7 +5129,6 @@ class UserRedactionTestCase(unittest.HomeserverTestCase):
         """
         Test that request to redact events in all rooms user is member of is successful
         """
-
         # join rooms, send some messages
         originals = []
         for rm in [self.rm1, self.rm2, self.rm3]:
@@ -5404,3 +5405,98 @@ class UserRedactionTestCase(unittest.HomeserverTestCase):
                     matches.append((event_id, event))
         # we redacted 6 messages
         self.assertEqual(len(matches), 6)
+
+
+class UserRedactionBackgroundTaskTestCase(BaseMultiWorkerStreamTestCase):
+    servlets = [
+        synapse.rest.admin.register_servlets,
+        login.register_servlets,
+        admin.register_servlets,
+        room.register_servlets,
+        sync.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.admin = self.register_user("thomas", "pass", True)
+        self.admin_tok = self.login("thomas", "pass")
+
+        self.bad_user = self.register_user("teresa", "pass")
+        self.bad_user_tok = self.login("teresa", "pass")
+
+        # create rooms - room versions 11+ store the `redacts` key in content while
+        # earlier ones don't so we use a mix of room versions
+        self.rm1 = self.helper.create_room_as(
+            self.admin, tok=self.admin_tok, room_version="7"
+        )
+        self.rm2 = self.helper.create_room_as(self.admin, tok=self.admin_tok)
+        self.rm3 = self.helper.create_room_as(
+            self.admin, tok=self.admin_tok, room_version="11"
+        )
+
+    @override_config({"run_background_tasks_on": "worker1"})
+    def test_redact_messages_all_rooms(self) -> None:
+        """
+        Test that redact task successfully runs when `run_background_tasks_on` is specified
+        """
+        self.make_worker_hs(
+            "synapse.app.generic_worker",
+            extra_config={
+                "worker_name": "worker1",
+                "run_background_tasks_on": "worker1",
+                "redis": {"enabled": True},
+            },
+        )
+
+        # join rooms, send some messages
+        original_event_ids = set()
+        for rm in [self.rm1, self.rm2, self.rm3]:
+            join = self.helper.join(rm, self.bad_user, tok=self.bad_user_tok)
+            original_event_ids.add(join["event_id"])
+            for i in range(15):
+                event = {"body": f"hello{i}", "msgtype": "m.text"}
+                res = self.helper.send_event(
+                    rm, "m.room.message", event, tok=self.bad_user_tok, expect_code=200
+                )
+                original_event_ids.add(res["event_id"])
+
+        # redact all events in all rooms
+        channel = self.make_request(
+            "POST",
+            f"/_synapse/admin/v1/user/{self.bad_user}/redact",
+            content={"rooms": []},
+            access_token=self.admin_tok,
+        )
+        self.assertEqual(channel.code, 200)
+        id = channel.json_body.get("redact_id")
+
+        timeout_s = 10
+        start_time = time.time()
+        redact_result = ""
+        while redact_result != "complete":
+            if start_time + timeout_s < time.time():
+                self.fail("Timed out waiting for redactions.")
+
+            channel2 = self.make_request(
+                "GET",
+                f"/_synapse/admin/v1/user/redact_status/{id}",
+                access_token=self.admin_tok,
+            )
+            redact_result = channel2.json_body["status"]
+            if redact_result == "failed":
+                self.fail("Redaction task failed.")
+
+        redaction_ids = set()
+        for rm in [self.rm1, self.rm2, self.rm3]:
+            filter = json.dumps({"types": [EventTypes.Redaction]})
+            channel = self.make_request(
+                "GET",
+                f"rooms/{rm}/messages?filter={filter}&limit=50",
+                access_token=self.admin_tok,
+            )
+            self.assertEqual(channel.code, 200)
+
+            for event in channel.json_body["chunk"]:
+                if event["type"] == "m.room.redaction":
+                    redaction_ids.add(event["redacts"])
+
+        self.assertIncludes(redaction_ids, original_event_ids, exact=True)