fixes

complement.sh: run tests from all test packages
... to save us remembering to add them to the list each time a new one is added.
2024-04-23 12:05:37 +01:00 · 2024-04-23 11:42:20 +01:00 · 2024-04-19 17:10:44 +01:00 · 2024-04-19 15:26:28 +00:00 · 2024-04-19 09:43:25 +01:00 · 2024-04-19 09:43:04 +01:00
74 changed files with 2010 additions and 632 deletions
@@ -30,7 +30,7 @@ jobs:
        run: docker buildx inspect

      - name: Install Cosign
-        uses: sigstore/cosign-installer@v3.4.0
+        uses: sigstore/cosign-installer@v3.5.0

      - name: Checkout repository
        uses: actions/checkout@v4
@@ -19,7 +19,7 @@ jobs:
          fetch-depth: 0

      - name: Setup mdbook
-        uses: peaceiris/actions-mdbook@adeb05db28a0c0004681db83893d56c0388ea9ea # v1.2.0
+        uses: peaceiris/actions-mdbook@ee69d230fe19748b7abf22df32acaa93833fad08 # v2.0.0
        with:
          mdbook-version: '0.4.17'

@@ -53,7 +53,7 @@ jobs:
      - uses: actions/checkout@v4

      - name: Setup mdbook
-        uses: peaceiris/actions-mdbook@adeb05db28a0c0004681db83893d56c0388ea9ea # v1.2.0
+        uses: peaceiris/actions-mdbook@ee69d230fe19748b7abf22df32acaa93833fad08 # v2.0.0
        with:
          mdbook-version: '0.4.17'

@@ -56,7 +56,7 @@ jobs:
          fetch-depth: 0

      - name: Setup mdbook
-        uses: peaceiris/actions-mdbook@adeb05db28a0c0004681db83893d56c0388ea9ea # v1.2.0
+        uses: peaceiris/actions-mdbook@ee69d230fe19748b7abf22df32acaa93833fad08 # v2.0.0
        with:
          mdbook-version: '0.4.17'

@@ -80,7 +80,7 @@ jobs:

      # Deploy to the target directory.
      - name: Deploy to gh pages
-        uses: peaceiris/actions-gh-pages@373f7f263a76c20808c831209c920827a82a2847 # v3.9.3
+        uses: peaceiris/actions-gh-pages@4f9cc6602d3f66b9c108549d475ec49e8ef4d45e # v4.0.0
        with:
          github_token: ${{ secrets.GITHUB_TOKEN }}
          publish_dir: ./book
@@ -110,7 +110,7 @@ jobs:

      # Deploy to the target directory.
      - name: Deploy to gh pages
-        uses: peaceiris/actions-gh-pages@373f7f263a76c20808c831209c920827a82a2847 # v3.9.3
+        uses: peaceiris/actions-gh-pages@4f9cc6602d3f66b9c108549d475ec49e8ef4d45e # v4.0.0
        with:
          github_token: ${{ secrets.GITHUB_TOKEN }}
          publish_dir: ./dev-docs/_build/html
@@ -81,7 +81,7 @@ jobs:
    steps:
      - uses: actions/checkout@v4
      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2
      - uses: matrix-org/setup-python-poetry@v1
        with:
@@ -148,7 +148,7 @@ jobs:
        uses: actions/checkout@v4

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2

      - name: Setup Poetry
@@ -208,7 +208,7 @@ jobs:
        with:
          ref: ${{ github.event.pull_request.head.sha }}
      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2
      - uses: matrix-org/setup-python-poetry@v1
        with:
@@ -225,7 +225,7 @@ jobs:
      - uses: actions/checkout@v4

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
        with:
            components: clippy
      - uses: Swatinem/rust-cache@v2
@@ -344,7 +344,7 @@ jobs:
            postgres:${{ matrix.job.postgres-version }}

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2

      - uses: matrix-org/setup-python-poetry@v1
@@ -386,7 +386,7 @@ jobs:
      - uses: actions/checkout@v4

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2

      # There aren't wheels for some of the older deps, so we need to install
@@ -498,7 +498,7 @@ jobs:
        run: cat sytest-blacklist .ci/worker-blacklist > synapse-blacklist-with-workers

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2

      - name: Run SyTest
@@ -642,7 +642,7 @@ jobs:
          path: synapse

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2

      - name: Prepare Complement's Prerequisites
@@ -674,7 +674,7 @@ jobs:
      - uses: actions/checkout@v4

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.65.0
+        uses: dtolnay/rust-toolchain@1.66.0
      - uses: Swatinem/rust-cache@v2

      - run: cargo test
@@ -1,3 +1,49 @@
+# Synapse 1.105.0 (2024-04-16)
+
+No significant changes since 1.105.0rc1.
+
+
+
+
+# Synapse 1.105.0rc1 (2024-04-11)
+
+### Features
+
+- Stabilize support for [MSC4010](https://github.com/matrix-org/matrix-spec-proposals/pull/4010) which clarifies the interaction of push rules and account data. Contributed by @clokep. ([\#17022](https://github.com/element-hq/synapse/issues/17022))
+- Stabilize support for [MSC3981](https://github.com/matrix-org/matrix-spec-proposals/pull/3981): `/relations` recursion. Contributed by @clokep. ([\#17023](https://github.com/element-hq/synapse/issues/17023))
+- Add support for moving `/pushrules` off of main process. ([\#17037](https://github.com/element-hq/synapse/issues/17037), [\#17038](https://github.com/element-hq/synapse/issues/17038))
+
+### Bugfixes
+
+- Fix various long-standing bugs which could cause incorrect state to be returned from `/sync` in certain situations. ([\#16930](https://github.com/element-hq/synapse/issues/16930), [\#16932](https://github.com/element-hq/synapse/issues/16932), [\#16942](https://github.com/element-hq/synapse/issues/16942), [\#17064](https://github.com/element-hq/synapse/issues/17064), [\#17065](https://github.com/element-hq/synapse/issues/17065), [\#17066](https://github.com/element-hq/synapse/issues/17066))
+- Fix server notice rooms not always being created as unencrypted rooms, even when `encryption_enabled_by_default_for_room_type` is in use (server notices are always unencrypted). ([\#17033](https://github.com/element-hq/synapse/issues/17033))
+- Fix the `.m.rule.encrypted_room_one_to_one` and `.m.rule.room_one_to_one` default underride push rules being in the wrong order. Contributed by @Sumpy1. ([\#17043](https://github.com/element-hq/synapse/issues/17043))
+
+### Internal Changes
+
+- Refactor auth chain fetching to reduce duplication. ([\#17044](https://github.com/element-hq/synapse/issues/17044))
+- Improve database performance by adding a missing index to `access_tokens.refresh_token_id`. ([\#17045](https://github.com/element-hq/synapse/issues/17045), [\#17054](https://github.com/element-hq/synapse/issues/17054))
+- Improve database performance by reducing number of receipts fetched when sending push notifications. ([\#17049](https://github.com/element-hq/synapse/issues/17049))
+
+
+
+### Updates to locked dependencies
+
+* Bump packaging from 23.2 to 24.0. ([\#17027](https://github.com/element-hq/synapse/issues/17027))
+* Bump regex from 1.10.3 to 1.10.4. ([\#17028](https://github.com/element-hq/synapse/issues/17028))
+* Bump ruff from 0.3.2 to 0.3.5. ([\#17060](https://github.com/element-hq/synapse/issues/17060))
+* Bump serde_json from 1.0.114 to 1.0.115. ([\#17041](https://github.com/element-hq/synapse/issues/17041))
+* Bump types-pillow from 10.2.0.20240125 to 10.2.0.20240406. ([\#17061](https://github.com/element-hq/synapse/issues/17061))
+* Bump types-requests from 2.31.0.20240125 to 2.31.0.20240406. ([\#17063](https://github.com/element-hq/synapse/issues/17063))
+* Bump typing-extensions from 4.9.0 to 4.11.0. ([\#17062](https://github.com/element-hq/synapse/issues/17062))
+
+# Synapse 1.104.0 (2024-04-02)
+
+### Bugfixes
+
+- Fix regression when using OIDC provider. Introduced in v1.104.0rc1. ([\#17031](https://github.com/element-hq/synapse/issues/17031))
+
+
 # Synapse 1.104.0rc1 (2024-03-26)

 ### Features
@@ -13,9 +13,9 @@ dependencies = [

 [[package]]
 name = "anyhow"
-version = "1.0.81"
+version = "1.0.82"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0952808a6c2afd1aa8947271f3a60f1a6763c7b912d210184c5149b5cf147247"
+checksum = "f538837af36e6f6a9be0faa67f9a314f8119e4e4b5867c6ab40ed60360142519"

 [[package]]
 name = "arc-swap"
@@ -29,6 +29,12 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa"

+[[package]]
+name = "base64"
+version = "0.21.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567"
+
 [[package]]
 name = "bitflags"
 version = "1.3.2"
@@ -53,12 +59,27 @@ dependencies = [
 "generic-array",
 ]

+[[package]]
+name = "bytes"
+version = "1.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "514de17de45fdb8dc022b1a7975556c53c86f9f0aa5f534b98977b171857c2c9"
+
 [[package]]
 name = "cfg-if"
 version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd"

+[[package]]
+name = "cpufeatures"
+version = "0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "53fe5e26ff1b7aef8bca9c6080520cfb8d9333c7568e1829cef191a9723e5504"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "crypto-common"
 version = "0.1.6"
@@ -80,6 +101,12 @@ dependencies = [
 "subtle",
 ]

+[[package]]
+name = "fnv"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
+
 [[package]]
 name = "generic-array"
 version = "0.14.6"
@@ -90,6 +117,30 @@ dependencies = [
 "version_check",
 ]

+[[package]]
+name = "headers"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "322106e6bd0cba2d5ead589ddb8150a13d7c4217cf80d7c4f682ca994ccc6aa9"
+dependencies = [
+ "base64",
+ "bytes",
+ "headers-core",
+ "http",
+ "httpdate",
+ "mime",
+ "sha1",
+]
+
+[[package]]
+name = "headers-core"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4"
+dependencies = [
+ "http",
+]
+
 [[package]]
 name = "heck"
 version = "0.4.1"
@@ -102,6 +153,23 @@ version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70"

+[[package]]
+name = "http"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "21b9ddb458710bc376481b842f5da65cdf31522de232c1ca8146abce2a358258"
+dependencies = [
+ "bytes",
+ "fnv",
+ "itoa",
+]
+
+[[package]]
+name = "httpdate"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
+
 [[package]]
 name = "indoc"
 version = "2.0.4"
@@ -122,9 +190,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"

 [[package]]
 name = "libc"
-version = "0.2.135"
+version = "0.2.153"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "68783febc7782c6c5cb401fbda4de5a9898be1762314da0bb2c10ced61f18b0c"
+checksum = "9c198f91728a82281a64e1f4f9eeb25d82cb32a5de251c6bd1b5154d63a8e7bd"

 [[package]]
 name = "lock_api"
@@ -157,6 +225,12 @@ dependencies = [
 "autocfg",
 ]

+[[package]]
+name = "mime"
+version = "0.3.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
+
 [[package]]
 name = "once_cell"
 version = "1.15.0"
@@ -306,9 +380,9 @@ dependencies = [

 [[package]]
 name = "regex"
-version = "1.10.3"
+version = "1.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b62dbe01f0b06f9d8dc7d49e05a0785f153b00b2c227856282f671e0318c9b15"
+checksum = "c117dbdfde9c8308975b6a18d71f3f385c89461f7b3fb054288ecf2a2058ba4c"
 dependencies = [
 "aho-corasick",
 "memchr",
@@ -367,15 +441,26 @@ dependencies = [

 [[package]]
 name = "serde_json"
-version = "1.0.114"
+version = "1.0.115"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c5f09b1bd632ef549eaa9f60a1f8de742bdbc698e6cee2095fc84dde5f549ae0"
+checksum = "12dc5c46daa8e9fdf4f5e71b6cf9a53f2487da0e86e55808e2d35539666497dd"
 dependencies = [
 "itoa",
 "ryu",
 "serde",
 ]

+[[package]]
+name = "sha1"
+version = "0.10.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f04293dc80c3993519f2d7f6f511707ee7094fe0c6d3406feb330cdb3540eba3"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
+]
+
 [[package]]
 name = "smallvec"
 version = "1.10.0"
@@ -405,7 +490,10 @@ version = "0.1.0"
 dependencies = [
 "anyhow",
 "blake2",
+ "bytes",
+ "headers",
 "hex",
+ "http",
 "lazy_static",
 "log",
 "pyo3",
@@ -0,0 +1 @@
+Adds validation to ensure that the `limit` parameter on `/publicRooms` is non-negative.
@@ -0,0 +1 @@
+Return `400 M_NOT_JSON` upon receiving invalid JSON in query parameters across various client and admin endpoints, rather than an internal server error.
@@ -0,0 +1 @@
+Make the CSAPI endpoint `/keys/device_signing/upload` idempotent.
@@ -1 +0,0 @@
-OIDC: try to JWT decode userinfo response if JSON parsing failed.
@@ -0,0 +1 @@
+Use new receipts column to optimise receipt and push action SQL queries. Contributed by Nick @ Beeper (@fizzadar).
@@ -0,0 +1 @@
+Fix mypy with latest Twisted release.
@@ -0,0 +1 @@
+Add a prompt in the contributing guide to manually configure icu4c.
@@ -0,0 +1 @@
+Bump minimum supported Rust version to 1.66.0.
@@ -0,0 +1 @@
+Add helpers to transform Twisted requests to Rust http Requests/Responses.
@@ -0,0 +1 @@
+Support delegating the rendezvous mechanism described MSC4108 to an external implementation.
@@ -0,0 +1 @@
+Use new receipts column to optimise receipt and push action SQL queries. Contributed by Nick @ Beeper (@fizzadar).
@@ -0,0 +1 @@
+Clarify what part of message retention is still experimental.
@@ -0,0 +1 @@
+`complement.sh`: run tests from all test packages.
@@ -1,3 +1,21 @@
+matrix-synapse-py3 (1.105.0) stable; urgency=medium
+
+  * New Synapse release 1.105.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 16 Apr 2024 15:53:23 +0100
+
+matrix-synapse-py3 (1.105.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.105.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 11 Apr 2024 12:15:49 +0100
+
+matrix-synapse-py3 (1.104.0) stable; urgency=medium
+
+  * New Synapse release 1.104.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 02 Apr 2024 17:15:45 +0100
+
 matrix-synapse-py3 (1.104.0~rc1) stable; urgency=medium

  * New Synapse release 1.104.0rc1.
@@ -102,6 +102,8 @@ experimental_features:
  msc3391_enabled: true
  # Filtering /messages by relation type.
  msc3874_enabled: true
+  # no UIA for x-signing upload for the first time
+  msc3967_enabled: true

 server_notices:
  system_mxid_localpart: _server
@@ -310,6 +310,13 @@ WORKERS_CONFIG: Dict[str, Dict[str, Any]] = {
        "shared_extra_conf": {},
        "worker_extra_conf": "",
    },
+    "push_rules": {
+        "app": "synapse.app.generic_worker",
+        "listener_resources": ["client", "replication"],
+        "endpoint_patterns": ["^/_matrix/client/(api/v1|r0|v3|unstable)/pushrules/"],
+        "shared_extra_conf": {},
+        "worker_extra_conf": "",
+    },
 }

 # Templates for sections that may be inserted multiple times in config files
@@ -401,6 +408,7 @@ def add_worker_roles_to_shared_config(
        "receipts",
        "to_device",
        "typing",
+        "push_rules",
    ]

    # Worker-type specific sharding config. Now a single worker can fulfill multiple
@@ -86,6 +86,8 @@ poetry install --extras all
 This will install the runtime and developer dependencies for the project.  Be sure to check
 that the `poetry install` step completed cleanly.

+For OSX users, be sure to set `PKG_CONFIG_PATH` to support `icu4c`. Run `brew info icu4c` for more details.
+
 ## Running Synapse via poetry

 To start a local instance of Synapse in the locked poetry environment, create a config file:
@@ -7,8 +7,10 @@ follow the semantics described in
 and allow server and room admins to configure how long messages should
 be kept in a homeserver's database before being purged from it.
 **Please note that, as this feature isn't part of the Matrix
-specification yet, this implementation is to be considered as
-experimental.**
+specification yet, the use of `m.room.retention` events for per-room
+retention policies is to be considered as experimental. However, the use
+of a default message retention policy is considered a stable feature
+in Synapse.**

 A message retention policy is mainly defined by its `max_lifetime`
 parameter, which defines how long a message can be kept around after
@@ -532,6 +532,13 @@ the stream writer for the `presence` stream:

    ^/_matrix/client/(api/v1|r0|v3|unstable)/presence/

+##### The `push_rules` stream
+
+The following endpoints should be routed directly to the worker configured as
+the stream writer for the `push` stream:
+
+    ^/_matrix/client/(api/v1|r0|v3|unstable)/pushrules/
+
 #### Restrict outbound federation traffic to a specific set of workers

 The
@@ -1,4 +1,4 @@
-# This file is automatically @generated by Poetry 1.7.1 and should not be changed by hand.
+# This file is automatically @generated by Poetry 1.8.2 and should not be changed by hand.

 [[package]]
 name = "alabaster"
@@ -1602,13 +1602,13 @@ tests = ["Sphinx", "doubles", "flake8", "flake8-quotes", "gevent", "mock", "pyte

 [[package]]
 name = "packaging"
-version = "23.2"
+version = "24.0"
 description = "Core utilities for Python packages"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "packaging-23.2-py3-none-any.whl", hash = "sha256:8c491190033a9af7e1d931d0b5dacc2ef47509b34dd0de67ed209b5203fc88c7"},
-    {file = "packaging-23.2.tar.gz", hash = "sha256:048fb0e9405036518eaaf48a55953c750c11e1a1b68e0dd1a9d62ed0c092cfc5"},
+    {file = "packaging-24.0-py3-none-any.whl", hash = "sha256:2ddfb553fdf02fb784c234c7ba6ccc288296ceabec964ad2eae3777778130bc5"},
+    {file = "packaging-24.0.tar.gz", hash = "sha256:eb82c5e3e56209074766e6885bb04b8c38a0c015d0a30036ebe7ece34c9989e9"},
 ]

 [[package]]
@@ -1848,17 +1848,17 @@ files = [

 [[package]]
 name = "pyasn1-modules"
-version = "0.3.0"
+version = "0.4.0"
 description = "A collection of ASN.1-based protocols modules"
 optional = false
-python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,!=3.5.*,>=2.7"
+python-versions = ">=3.8"
 files = [
-    {file = "pyasn1_modules-0.3.0-py2.py3-none-any.whl", hash = "sha256:d3ccd6ed470d9ffbc716be08bd90efbd44d0734bc9303818f7336070984a162d"},
-    {file = "pyasn1_modules-0.3.0.tar.gz", hash = "sha256:5bd01446b736eb9d31512a30d46c1ac3395d676c6f3cafa4c03eb54b9925631c"},
+    {file = "pyasn1_modules-0.4.0-py3-none-any.whl", hash = "sha256:be04f15b66c206eed667e0bb5ab27e2b1855ea54a842e5037738099e8ca4ae0b"},
+    {file = "pyasn1_modules-0.4.0.tar.gz", hash = "sha256:831dbcea1b177b28c9baddf4c6d1013c24c3accd14a1873fffaa6a2e905f17b6"},
 ]

 [package.dependencies]
-pyasn1 = ">=0.4.6,<0.6.0"
+pyasn1 = ">=0.4.6,<0.7.0"

 [[package]]
 name = "pycparser"
@@ -1983,13 +1983,13 @@ typing-extensions = ">=4.6.0,<4.7.0 || >4.7.0"

 [[package]]
 name = "pygithub"
-version = "2.2.0"
+version = "2.3.0"
 description = "Use the full Github API v3"
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "PyGithub-2.2.0-py3-none-any.whl", hash = "sha256:41042ea53e4c372219db708c38d2ca1fd4fadab75475bac27d89d339596cfad1"},
-    {file = "PyGithub-2.2.0.tar.gz", hash = "sha256:e39be7c4dc39418bdd6e3ecab5931c636170b8b21b4d26f9ecf7e6102a3b51c3"},
+    {file = "PyGithub-2.3.0-py3-none-any.whl", hash = "sha256:65b499728be3ce7b0cd2cd760da3b32f0f4d7bc55e5e0677617f90f6564e793e"},
+    {file = "PyGithub-2.3.0.tar.gz", hash = "sha256:0148d7347a1cdeed99af905077010aef81a4dad988b0ba51d4108bf66b443f7e"},
 ]

 [package.dependencies]
@@ -2444,28 +2444,28 @@ files = [

 [[package]]
 name = "ruff"
-version = "0.3.2"
+version = "0.3.7"
 description = "An extremely fast Python linter and code formatter, written in Rust."
 optional = false
 python-versions = ">=3.7"
 files = [
-    {file = "ruff-0.3.2-py3-none-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl", hash = "sha256:77f2612752e25f730da7421ca5e3147b213dca4f9a0f7e0b534e9562c5441f01"},
-    {file = "ruff-0.3.2-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:9966b964b2dd1107797be9ca7195002b874424d1d5472097701ae8f43eadef5d"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b83d17ff166aa0659d1e1deaf9f2f14cbe387293a906de09bc4860717eb2e2da"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:bb875c6cc87b3703aeda85f01c9aebdce3d217aeaca3c2e52e38077383f7268a"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:be75e468a6a86426430373d81c041b7605137a28f7014a72d2fc749e47f572aa"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:967978ac2d4506255e2f52afe70dda023fc602b283e97685c8447d036863a302"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:1231eacd4510f73222940727ac927bc5d07667a86b0cbe822024dd00343e77e9"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:2c6d613b19e9a8021be2ee1d0e27710208d1603b56f47203d0abbde906929a9b"},
-    {file = "ruff-0.3.2-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:c8439338a6303585d27b66b4626cbde89bb3e50fa3cae86ce52c1db7449330a7"},
-    {file = "ruff-0.3.2-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:de8b480d8379620cbb5ea466a9e53bb467d2fb07c7eca54a4aa8576483c35d36"},
-    {file = "ruff-0.3.2-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:b74c3de9103bd35df2bb05d8b2899bf2dbe4efda6474ea9681280648ec4d237d"},
-    {file = "ruff-0.3.2-py3-none-musllinux_1_2_i686.whl", hash = "sha256:f380be9fc15a99765c9cf316b40b9da1f6ad2ab9639e551703e581a5e6da6745"},
-    {file = "ruff-0.3.2-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:0ac06a3759c3ab9ef86bbeca665d31ad3aa9a4b1c17684aadb7e61c10baa0df4"},
-    {file = "ruff-0.3.2-py3-none-win32.whl", hash = "sha256:9bd640a8f7dd07a0b6901fcebccedadeb1a705a50350fb86b4003b805c81385a"},
-    {file = "ruff-0.3.2-py3-none-win_amd64.whl", hash = "sha256:0c1bdd9920cab5707c26c8b3bf33a064a4ca7842d91a99ec0634fec68f9f4037"},
-    {file = "ruff-0.3.2-py3-none-win_arm64.whl", hash = "sha256:5f65103b1d76e0d600cabd577b04179ff592064eaa451a70a81085930e907d0b"},
-    {file = "ruff-0.3.2.tar.gz", hash = "sha256:fa78ec9418eb1ca3db392811df3376b46471ae93792a81af2d1cbb0e5dcb5142"},
+    {file = "ruff-0.3.7-py3-none-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl", hash = "sha256:0e8377cccb2f07abd25e84fc5b2cbe48eeb0fea9f1719cad7caedb061d70e5ce"},
+    {file = "ruff-0.3.7-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:15a4d1cc1e64e556fa0d67bfd388fed416b7f3b26d5d1c3e7d192c897e39ba4b"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:d28bdf3d7dc71dd46929fafeec98ba89b7c3550c3f0978e36389b5631b793663"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:379b67d4f49774ba679593b232dcd90d9e10f04d96e3c8ce4a28037ae473f7bb"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:c060aea8ad5ef21cdfbbe05475ab5104ce7827b639a78dd55383a6e9895b7c51"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_ppc64.manylinux2014_ppc64.whl", hash = "sha256:ebf8f615dde968272d70502c083ebf963b6781aacd3079081e03b32adfe4d58a"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d48098bd8f5c38897b03604f5428901b65e3c97d40b3952e38637b5404b739a2"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:da8a4fda219bf9024692b1bc68c9cff4b80507879ada8769dc7e985755d662ea"},
+    {file = "ruff-0.3.7-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c44e0149f1d8b48c4d5c33d88c677a4aa22fd09b1683d6a7ff55b816b5d074f"},
+    {file = "ruff-0.3.7-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:3050ec0af72b709a62ecc2aca941b9cd479a7bf2b36cc4562f0033d688e44fa1"},
+    {file = "ruff-0.3.7-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:a29cc38e4c1ab00da18a3f6777f8b50099d73326981bb7d182e54a9a21bb4ff7"},
+    {file = "ruff-0.3.7-py3-none-musllinux_1_2_i686.whl", hash = "sha256:5b15cc59c19edca917f51b1956637db47e200b0fc5e6e1878233d3a938384b0b"},
+    {file = "ruff-0.3.7-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:e491045781b1e38b72c91247cf4634f040f8d0cb3e6d3d64d38dcf43616650b4"},
+    {file = "ruff-0.3.7-py3-none-win32.whl", hash = "sha256:bc931de87593d64fad3a22e201e55ad76271f1d5bfc44e1a1887edd0903c7d9f"},
+    {file = "ruff-0.3.7-py3-none-win_amd64.whl", hash = "sha256:5ef0e501e1e39f35e03c2acb1d1238c595b8bb36cf7a170e7c1df1b73da00e74"},
+    {file = "ruff-0.3.7-py3-none-win_arm64.whl", hash = "sha256:789e144f6dc7019d1f92a812891c645274ed08af6037d11fc65fcbc183b7d59f"},
+    {file = "ruff-0.3.7.tar.gz", hash = "sha256:d5c1aebee5162c2226784800ae031f660c350e7a3402c4d1f8ea4e97e232e3ba"},
 ]

 [[package]]
@@ -2954,13 +2954,13 @@ docs = ["sphinx (<7.0.0)"]

 [[package]]
 name = "twine"
-version = "4.0.2"
+version = "5.0.0"
 description = "Collection of utilities for publishing packages on PyPI"
 optional = false
-python-versions = ">=3.7"
+python-versions = ">=3.8"
 files = [
-    {file = "twine-4.0.2-py3-none-any.whl", hash = "sha256:929bc3c280033347a00f847236564d1c52a3e61b1ac2516c97c48f3ceab756d8"},
-    {file = "twine-4.0.2.tar.gz", hash = "sha256:9e102ef5fdd5a20661eb88fad46338806c3bd32cf1db729603fe3697b1bc83c8"},
+    {file = "twine-5.0.0-py3-none-any.whl", hash = "sha256:a262933de0b484c53408f9edae2e7821c1c45a3314ff2df9bdd343aa7ab8edc0"},
+    {file = "twine-5.0.0.tar.gz", hash = "sha256:89b0cc7d370a4b66421cc6102f269aa910fe0f1861c124f573cf2ddedbc10cf4"},
 ]

 [package.dependencies]
@@ -3109,13 +3109,13 @@ files = [

 [[package]]
 name = "types-pillow"
-version = "10.2.0.20240125"
+version = "10.2.0.20240415"
 description = "Typing stubs for Pillow"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "types-Pillow-10.2.0.20240125.tar.gz", hash = "sha256:c449b2c43b9fdbe0494a7b950e6b39a4e50516091213fec24ef3f33c1d017717"},
-    {file = "types_Pillow-10.2.0.20240125-py3-none-any.whl", hash = "sha256:322dbae32b4b7918da5e8a47c50ac0f24b0aa72a804a23857620f2722b03c858"},
+    {file = "types-Pillow-10.2.0.20240415.tar.gz", hash = "sha256:dd6058027639bcdc66ba78b228cc25fdae42524c2150c78c804da427e7e76e70"},
+    {file = "types_Pillow-10.2.0.20240415-py3-none-any.whl", hash = "sha256:f933332b7e96010bae9b9cf82a4c9979ff0c270d63f5c5bbffb2d789b85cd00b"},
 ]

 [[package]]
@@ -3156,13 +3156,13 @@ files = [

 [[package]]
 name = "types-requests"
-version = "2.31.0.20240125"
+version = "2.31.0.20240406"
 description = "Typing stubs for requests"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "types-requests-2.31.0.20240125.tar.gz", hash = "sha256:03a28ce1d7cd54199148e043b2079cdded22d6795d19a2c2a6791a4b2b5e2eb5"},
-    {file = "types_requests-2.31.0.20240125-py3-none-any.whl", hash = "sha256:9592a9a4cb92d6d75d9b491a41477272b710e021011a2a3061157e2fb1f1a5d1"},
+    {file = "types-requests-2.31.0.20240406.tar.gz", hash = "sha256:4428df33c5503945c74b3f42e82b181e86ec7b724620419a2966e2de604ce1a1"},
+    {file = "types_requests-2.31.0.20240406-py3-none-any.whl", hash = "sha256:6216cdac377c6b9a040ac1c0404f7284bd13199c0e1bb235f4324627e8898cf5"},
 ]

 [package.dependencies]
@@ -3181,13 +3181,13 @@ files = [

 [[package]]
 name = "typing-extensions"
-version = "4.9.0"
+version = "4.11.0"
 description = "Backported and Experimental Type Hints for Python 3.8+"
 optional = false
 python-versions = ">=3.8"
 files = [
-    {file = "typing_extensions-4.9.0-py3-none-any.whl", hash = "sha256:af72aea155e91adfc61c3ae9e0e342dbc0cba726d6cba4b6c72c1f34e47291cd"},
-    {file = "typing_extensions-4.9.0.tar.gz", hash = "sha256:23478f88c37f27d76ac8aee6c905017a143b0b1b886c3c9f66bc2fd94f9f5783"},
+    {file = "typing_extensions-4.11.0-py3-none-any.whl", hash = "sha256:c1f94d72897edaf4ce775bb7558d5b79d8126906a14ea5ed1635921406c0387a"},
+    {file = "typing_extensions-4.11.0.tar.gz", hash = "sha256:83f085bd5ca59c80295fc2a82ab5dac679cbe02b9f33f7d83af68e241bea51b0"},
 ]

 [[package]]
@@ -3451,4 +3451,4 @@ user-search = ["pyicu"]
 [metadata]
 lock-version = "2.0"
 python-versions = "^3.8.0"
-content-hash = "b510fa05f4ea33194bec079f5d04ebb3f9ffbb5c1ea96a0341d57ba770ef81e6"
+content-hash = "1951f2b4623138d47db08a405edd970e67599d05804bb459af21a085e1665f69"
@@ -96,7 +96,7 @@ module-name = "synapse.synapse_rust"

 [tool.poetry]
 name = "matrix-synapse"
-version = "1.104.0rc1"
+version = "1.105.0"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "AGPL-3.0-or-later"
@@ -321,7 +321,7 @@ all = [
 # This helps prevents merge conflicts when running a batch of dependabot updates.
 isort = ">=5.10.1"
 black = ">=22.7.0"
-ruff = "0.3.2"
+ruff = "0.3.7"
 # Type checking only works with the pydantic.v1 compat module from pydantic v2
 pydantic = "^2"

@@ -7,7 +7,7 @@ name = "synapse"
 version = "0.1.0"

 edition = "2021"
-rust-version = "1.65.0"
+rust-version = "1.66.0"

 [lib]
 name = "synapse"
@@ -23,6 +23,9 @@ name = "synapse.synapse_rust"

 [dependencies]
 anyhow = "1.0.63"
+bytes = "1.6.0"
+headers = "0.4.0"
+http = "1.1.0"
 lazy_static = "1.4.0"
 log = "0.4.17"
 pyo3 = { version = "0.20.0", features = [
@@ -0,0 +1,60 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2024 New Vector, Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * See the GNU Affero General Public License for more details:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ */
+
+#![allow(clippy::new_ret_no_self)]
+
+use std::collections::HashMap;
+
+use http::{HeaderMap, StatusCode};
+use pyo3::{exceptions::PyValueError, import_exception};
+
+import_exception!(synapse.api.errors, SynapseError);
+
+impl SynapseError {
+    pub fn new(
+        code: StatusCode,
+        message: String,
+        errcode: &'static str,
+        additional_fields: Option<HashMap<String, String>>,
+        headers: Option<HeaderMap>,
+    ) -> pyo3::PyErr {
+        // Transform the HeaderMap into a HashMap<String, String>
+        let headers = if let Some(headers) = headers {
+            let mut map = HashMap::with_capacity(headers.len());
+            for (key, value) in headers.iter() {
+                let Ok(value) = value.to_str() else {
+                    // This should never happen, but we don't want to panic in case it does
+                    return PyValueError::new_err(
+                        "Could not construct SynapseError: header value is not valid ASCII",
+                    );
+                };
+
+                map.insert(key.as_str().to_owned(), value.to_owned());
+            }
+            Some(map)
+        } else {
+            None
+        };
+
+        SynapseError::new_err((code.as_u16(), message, errcode, additional_fields, headers))
+    }
+}
+
+import_exception!(synapse.api.errors, NotFoundError);
+
+impl NotFoundError {
+    pub fn new() -> pyo3::PyErr {
+        NotFoundError::new_err(())
+    }
+}
@@ -0,0 +1,165 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2024 New Vector, Ltd
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * See the GNU Affero General Public License for more details:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ */
+
+use bytes::{Buf, BufMut, Bytes, BytesMut};
+use headers::{Header, HeaderMapExt};
+use http::{HeaderName, HeaderValue, Method, Request, Response, StatusCode, Uri};
+use pyo3::{
+    exceptions::PyValueError,
+    types::{PyBytes, PySequence, PyTuple},
+    PyAny, PyResult,
+};
+
+use crate::errors::SynapseError;
+
+/// Read a file-like Python object by chunks
+///
+/// # Errors
+///
+/// Returns an error if calling the `read` on the Python object failed
+fn read_io_body(body: &PyAny, chunk_size: usize) -> PyResult<Bytes> {
+    let mut buf = BytesMut::new();
+    loop {
+        let bytes: &PyBytes = body.call_method1("read", (chunk_size,))?.downcast()?;
+        if bytes.as_bytes().is_empty() {
+            return Ok(buf.into());
+        }
+        buf.put(bytes.as_bytes());
+    }
+}
+
+/// Transform a Twisted `IRequest` to an [`http::Request`]
+///
+/// It uses the following members of `IRequest`:
+///   - `content`, which is expected to be a file-like object with a `read` method
+///   - `uri`, which is expected to be a valid URI as `bytes`
+///   - `method`, which is expected to be a valid HTTP method as `bytes`
+///   - `requestHeaders`, which is expected to have a `getAllRawHeaders` method
+///
+/// # Errors
+///
+/// Returns an error if the Python object doesn't properly implement `IRequest`
+pub fn http_request_from_twisted(request: &PyAny) -> PyResult<Request<Bytes>> {
+    let content = request.getattr("content")?;
+    let body = read_io_body(content, 4096)?;
+
+    let mut req = Request::new(body);
+
+    let uri: &PyBytes = request.getattr("uri")?.downcast()?;
+    *req.uri_mut() =
+        Uri::try_from(uri.as_bytes()).map_err(|_| PyValueError::new_err("invalid uri"))?;
+
+    let method: &PyBytes = request.getattr("method")?.downcast()?;
+    *req.method_mut() = Method::from_bytes(method.as_bytes())
+        .map_err(|_| PyValueError::new_err("invalid method"))?;
+
+    let headers_iter = request
+        .getattr("requestHeaders")?
+        .call_method0("getAllRawHeaders")?
+        .iter()?;
+
+    for header in headers_iter {
+        let header = header?;
+        let header: &PyTuple = header.downcast()?;
+        let name: &PyBytes = header.get_item(0)?.downcast()?;
+        let name = HeaderName::from_bytes(name.as_bytes())
+            .map_err(|_| PyValueError::new_err("invalid header name"))?;
+
+        let values: &PySequence = header.get_item(1)?.downcast()?;
+        for index in 0..values.len()? {
+            let value: &PyBytes = values.get_item(index)?.downcast()?;
+            let value = HeaderValue::from_bytes(value.as_bytes())
+                .map_err(|_| PyValueError::new_err("invalid header value"))?;
+            req.headers_mut().append(name.clone(), value);
+        }
+    }
+
+    Ok(req)
+}
+
+/// Send an [`http::Response`] through a Twisted `IRequest`
+///
+/// It uses the following members of `IRequest`:
+///
+///  - `responseHeaders`, which is expected to have a `addRawHeader(bytes, bytes)` method
+///  - `setResponseCode(int)` method
+///  - `write(bytes)` method
+///  - `finish()` method
+///
+///  # Errors
+///
+/// Returns an error if the Python object doesn't properly implement `IRequest`
+pub fn http_response_to_twisted<B>(request: &PyAny, response: Response<B>) -> PyResult<()>
+where
+    B: Buf,
+{
+    let (parts, mut body) = response.into_parts();
+
+    request.call_method1("setResponseCode", (parts.status.as_u16(),))?;
+
+    let response_headers = request.getattr("responseHeaders")?;
+    for (name, value) in parts.headers.iter() {
+        response_headers.call_method1("addRawHeader", (name.as_str(), value.as_bytes()))?;
+    }
+
+    while body.remaining() != 0 {
+        let chunk = body.chunk();
+        request.call_method1("write", (chunk,))?;
+        body.advance(chunk.len());
+    }
+
+    request.call_method0("finish")?;
+
+    Ok(())
+}
+
+/// An extension trait for [`HeaderMap`] that provides typed access to headers, and throws the
+/// right python exceptions when the header is missing or fails to parse.
+///
+/// [`HeaderMap`]: headers::HeaderMap
+pub trait HeaderMapPyExt: HeaderMapExt {
+    /// Get a header from the map, returning an error if it is missing or invalid.
+    fn typed_get_required<H>(&self) -> PyResult<H>
+    where
+        H: Header,
+    {
+        self.typed_get_optional::<H>()?.ok_or_else(|| {
+            SynapseError::new(
+                StatusCode::BAD_REQUEST,
+                format!("Missing required header: {}", H::name()),
+                "M_MISSING_PARAM",
+                None,
+                None,
+            )
+        })
+    }
+
+    /// Get a header from the map, returning `None` if it is missing and an error if it is invalid.
+    fn typed_get_optional<H>(&self) -> PyResult<Option<H>>
+    where
+        H: Header,
+    {
+        self.typed_try_get::<H>().map_err(|_| {
+            SynapseError::new(
+                StatusCode::BAD_REQUEST,
+                format!("Invalid header: {}", H::name()),
+                "M_INVALID_PARAM",
+                None,
+                None,
+            )
+        })
+    }
+}
+
+impl<T: HeaderMapExt> HeaderMapPyExt for T {}
@@ -3,7 +3,9 @@ use pyo3::prelude::*;
 use pyo3_log::ResetHandle;

 pub mod acl;
+pub mod errors;
 pub mod events;
+pub mod http;
 pub mod push;

 lazy_static! {
@@ -304,12 +304,12 @@ pub const BASE_APPEND_UNDERRIDE_RULES: &[PushRule] = &[
        default_enabled: true,
    },
    PushRule {
-        rule_id: Cow::Borrowed("global/underride/.m.rule.room_one_to_one"),
+        rule_id: Cow::Borrowed("global/underride/.m.rule.encrypted_room_one_to_one"),
        priority_class: 1,
        conditions: Cow::Borrowed(&[
            Condition::Known(KnownCondition::EventMatch(EventMatchCondition {
                key: Cow::Borrowed("type"),
-                pattern: Cow::Borrowed("m.room.message"),
+                pattern: Cow::Borrowed("m.room.encrypted"),
            })),
            Condition::Known(KnownCondition::RoomMemberCount {
                is: Some(Cow::Borrowed("2")),
@@ -320,12 +320,12 @@ pub const BASE_APPEND_UNDERRIDE_RULES: &[PushRule] = &[
        default_enabled: true,
    },
    PushRule {
-        rule_id: Cow::Borrowed("global/underride/.m.rule.encrypted_room_one_to_one"),
+        rule_id: Cow::Borrowed("global/underride/.m.rule.room_one_to_one"),
        priority_class: 1,
        conditions: Cow::Borrowed(&[
            Condition::Known(KnownCondition::EventMatch(EventMatchCondition {
                key: Cow::Borrowed("type"),
-                pattern: Cow::Borrowed("m.room.encrypted"),
+                pattern: Cow::Borrowed("m.room.message"),
            })),
            Condition::Known(KnownCondition::RoomMemberCount {
                is: Some(Cow::Borrowed("2")),
@@ -214,8 +214,6 @@ fi

 extra_test_args=()

-test_packages="./tests/csapi ./tests ./tests/msc3874 ./tests/msc3890 ./tests/msc3391 ./tests/msc3930 ./tests/msc3902"
-
 # Enable dirty runs, so tests will reuse the same container where possible.
 # This significantly speeds up tests, but increases the possibility of test pollution.
 export COMPLEMENT_ENABLE_DIRTY_RUNS=1
@@ -278,7 +276,12 @@ fi
 export PASS_SYNAPSE_LOG_TESTING=1

 # Run the tests!
-echo "Images built; running complement with ${extra_test_args[@]} $@ $test_packages"
 cd "$COMPLEMENT_DIR"

-go test -v -tags "synapse_blacklist" -count=1 "${extra_test_args[@]}" "$@" $test_packages
+# This isn't whitespace-safe but *does* work on the prehistoric version of bash
+# on OSX.
+test_packages=( $(find ./tests -type d) )
+
+echo "Images built; running complement with ${extra_test_args[@]} $@ ${test_packages[@]}"
+
+go test -v -tags "synapse_blacklist" -count=1 "${extra_test_args[@]}" "$@" "${test_packages[@]}"
@@ -60,7 +60,7 @@ from synapse.logging.context import (
 )
 from synapse.notifier import ReplicationNotifier
 from synapse.storage.database import DatabasePool, LoggingTransaction, make_conn
-from synapse.storage.databases.main import FilteringWorkerStore, PushRuleStore
+from synapse.storage.databases.main import FilteringWorkerStore
 from synapse.storage.databases.main.account_data import AccountDataWorkerStore
 from synapse.storage.databases.main.client_ips import ClientIpBackgroundUpdateStore
 from synapse.storage.databases.main.deviceinbox import DeviceInboxBackgroundUpdateStore
@@ -77,10 +77,8 @@ from synapse.storage.databases.main.media_repository import (
 )
 from synapse.storage.databases.main.presence import PresenceBackgroundUpdateStore
 from synapse.storage.databases.main.profile import ProfileWorkerStore
-from synapse.storage.databases.main.pusher import (
-    PusherBackgroundUpdatesStore,
-    PusherWorkerStore,
-)
+from synapse.storage.databases.main.push_rule import PusherWorkerStore
+from synapse.storage.databases.main.pusher import PusherBackgroundUpdatesStore
 from synapse.storage.databases.main.receipts import ReceiptsBackgroundUpdateStore
 from synapse.storage.databases.main.registration import (
    RegistrationBackgroundUpdateStore,
@@ -245,7 +243,6 @@ class Store(
    AccountDataWorkerStore,
    FilteringWorkerStore,
    ProfileWorkerStore,
-    PushRuleStore,
    PusherWorkerStore,
    PusherBackgroundUpdatesStore,
    PresenceBackgroundUpdateStore,
@@ -393,11 +393,6 @@ class ExperimentalConfig(Config):
        # MSC3967: Do not require UIA when first uploading cross signing keys
        self.msc3967_enabled = experimental.get("msc3967_enabled", False)

-        # MSC3981: Recurse relations
-        self.msc3981_recurse_relations = experimental.get(
-            "msc3981_recurse_relations", False
-        )
-
        # MSC3861: Matrix architecture change to delegate authentication via OIDC
        try:
            self.msc3861 = MSC3861(**experimental.get("msc3861", {}))
@@ -409,11 +404,6 @@ class ExperimentalConfig(Config):
        # Check that none of the other config options conflict with MSC3861 when enabled
        self.msc3861.check_config_conflicts(self.root)

-        # MSC4010: Do not allow setting m.push_rules account data.
-        self.msc4010_push_rules_account_data = experimental.get(
-            "msc4010_push_rules_account_data", False
-        )
-
        self.msc4028_push_encrypted_events = experimental.get(
            "msc4028_push_encrypted_events", False
        )
@@ -421,3 +411,14 @@ class ExperimentalConfig(Config):
        self.msc4069_profile_inhibit_propagation = experimental.get(
            "msc4069_profile_inhibit_propagation", False
        )
+
+        # MSC4108: Mechanism to allow OIDC sign in and E2EE set up via QR code
+        self.msc4108_delegation_endpoint: Optional[str] = experimental.get(
+            "msc4108_delegation_endpoint", None
+        )
+
+        if self.msc4108_delegation_endpoint is not None and not self.msc3861.enabled:
+            raise ConfigError(
+                "MSC4108 requires MSC3861 to be enabled",
+                ("experimental", "msc4108_delegation_endpoint"),
+            )
@@ -156,6 +156,8 @@ class WriterLocations:
            can only be a single instance.
        presence: The instances that write to the presence stream. Currently
            can only be a single instance.
+        push_rules: The instances that write to the push stream. Currently
+            can only be a single instance.
    """

    events: List[str] = attr.ib(
@@ -182,6 +184,10 @@ class WriterLocations:
        default=["master"],
        converter=_instance_to_list_converter,
    )
+    push_rules: List[str] = attr.ib(
+        default=["master"],
+        converter=_instance_to_list_converter,
+    )


@attr.s(auto_attribs=True)
@@ -341,6 +347,7 @@ class WorkerConfig(Config):
            "account_data",
            "receipts",
            "presence",
+            "push_rules",
        ):
            instances = _instance_to_list_converter(getattr(self.writers, stream))
            for instance in instances:
@@ -378,6 +385,11 @@ class WorkerConfig(Config):
                "Must only specify one instance to handle `presence` messages."
            )

+        if len(self.writers.push_rules) != 1:
+            raise ConfigError(
+                "Must only specify one instance to handle `push` messages."
+            )
+
        self.events_shard_config = RoutableShardedWorkerHandlingConfig(
            self.writers.events
        )
@@ -1476,6 +1476,42 @@ class E2eKeysHandler:
        else:
            return exists, self.clock.time_msec() < ts_replacable_without_uia_before

+    async def has_different_keys(self, user_id: str, body: JsonDict) -> bool:
+        """
+        Check if a key provided in `body` differs from the same key stored in the DB. Returns
+        true on the first difference. If a key exists in `body` but does not exist in the DB,
+        returns True. If `body` has no keys, this always returns False.
+        Note by 'key' we mean Matrix key rather than JSON key.
+
+        The purpose of this function is to detect whether or not we need to apply UIA checks.
+        We must apply UIA checks if any key in the database is being overwritten. If a key is
+        being inserted for the first time, or if the key exactly matches what is in the database,
+        then no UIA check needs to be performed.
+
+        Args:
+            user_id: The user who sent the `body`.
+            body: The JSON request body from POST /keys/device_signing/upload
+        Returns:
+            True if any key in `body` has a different value in the database.
+        """
+        # Ensure that each key provided in the request body exactly matches the one we have stored.
+        # The first time we see the DB having a different key to the matching request key, bail.
+        # Note: we do not care if the DB has a key which the request does not specify, as we only
+        # care about *replacements* or *insertions* (i.e UPSERT)
+        req_body_key_to_db_key = {
+            "master_key": "master",
+            "self_signing_key": "self_signing",
+            "user_signing_key": "user_signing",
+        }
+        for req_body_key, db_key in req_body_key_to_db_key.items():
+            if req_body_key in body:
+                existing_key = await self.store.get_e2e_cross_signing_key(
+                    user_id, db_key
+                )
+                if existing_key != body[req_body_key]:
+                    return True
+        return False
+

 def _check_cross_signing_key(
    key: JsonDict, user_id: str, key_type: str, signing_key: Optional[VerifyKey] = None
@@ -956,6 +956,7 @@ class RoomCreationHandler:
            room_alias=room_alias,
            power_level_content_override=power_level_content_override,
            creator_join_profile=creator_join_profile,
+            ignore_forced_encryption=ignore_forced_encryption,
        )

        # we avoid dropping the lock between invites, as otherwise joins can
@@ -51,6 +51,7 @@ from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
 from synapse.logging import opentracing
 from synapse.metrics import event_processing_positions
 from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.replication.http.push import ReplicationCopyPusherRestServlet
 from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.types import (
    JsonDict,
@@ -181,6 +182,12 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
            hs.config.server.forgotten_room_retention_period
        )

+        self._is_push_writer = (
+            hs.get_instance_name() in hs.config.worker.writers.push_rules
+        )
+        self._push_writer = hs.config.worker.writers.push_rules[0]
+        self._copy_push_client = ReplicationCopyPusherRestServlet.make_client(hs)
+
    def _on_user_joined_room(self, event_id: str, room_id: str) -> None:
        """Notify the rate limiter that a room join has occurred.

@@ -1301,9 +1308,17 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                    old_room_id, new_room_id, user_id
                )
                # Copy over push rules
-                await self.store.copy_push_rules_from_room_to_room_for_user(
-                    old_room_id, new_room_id, user_id
-                )
+                if self._is_push_writer:
+                    await self.store.copy_push_rules_from_room_to_room_for_user(
+                        old_room_id, new_room_id, user_id
+                    )
+                else:
+                    await self._copy_push_client(
+                        instance_name=self._push_writer,
+                        user_id=user_id,
+                        old_room_id=old_room_id,
+                        new_room_id=new_room_id,
+                    )
            except Exception:
                logger.exception(
                    "Error copying tags and/or push rules from rooms %s to %s for user %s. "
@@ -953,7 +953,7 @@ class SyncHandler:
        batch: TimelineBatch,
        sync_config: SyncConfig,
        since_token: Optional[StreamToken],
-        now_token: StreamToken,
+        end_token: StreamToken,
        full_state: bool,
    ) -> MutableStateMap[EventBase]:
        """Works out the difference in state between the end of the previous sync and
@@ -964,7 +964,9 @@ class SyncHandler:
            batch: The timeline batch for the room that will be sent to the user.
            sync_config:
            since_token: Token of the end of the previous batch. May be `None`.
-            now_token: Token of the end of the current batch.
+            end_token: Token of the end of the current batch. Normally this will be
+                the same as the global "now_token", but if the user has left the room,
+                the point just after their leave event.
            full_state: Whether to force returning the full state.
                `lazy_load_members` still applies when `full_state` is `True`.

@@ -1044,7 +1046,7 @@ class SyncHandler:
                    room_id,
                    sync_config.user,
                    batch,
-                    now_token,
+                    end_token,
                    members_to_fetch,
                    timeline_state,
                )
@@ -1058,7 +1060,7 @@ class SyncHandler:
                    room_id,
                    batch,
                    since_token,
-                    now_token,
+                    end_token,
                    members_to_fetch,
                    timeline_state,
                )
@@ -1130,7 +1132,7 @@ class SyncHandler:
        room_id: str,
        syncing_user: UserID,
        batch: TimelineBatch,
-        now_token: StreamToken,
+        end_token: StreamToken,
        members_to_fetch: Optional[Set[str]],
        timeline_state: StateMap[str],
    ) -> StateMap[str]:
@@ -1143,7 +1145,9 @@ class SyncHandler:
            room_id: The room we are calculating for.
            syncing_user: The user that is calling `/sync`.
            batch: The timeline batch for the room that will be sent to the user.
-            now_token: Token of the end of the current batch.
+            end_token: Token of the end of the current batch. Normally this will be
+                the same as the global "now_token", but if the user has left the room,
+                the point just after their leave event.
            members_to_fetch: If lazy-loading is enabled, the memberships needed for
                events in the timeline.
            timeline_state: The contribution to the room state from state events in
@@ -1183,15 +1187,16 @@ class SyncHandler:
            await_full_state = True
            lazy_load_members = False

-        if batch:
-            state_at_timeline_end = (
-                await self._state_storage_controller.get_state_ids_for_event(
-                    batch.events[-1].event_id,
-                    state_filter=state_filter,
-                    await_full_state=await_full_state,
-                )
-            )
+        state_at_timeline_end = await self.get_state_at(
+            room_id,
+            stream_position=end_token,
+            state_filter=state_filter,
+            await_full_state=await_full_state,
+        )

+        if batch:
+            # Strictly speaking, this returns the state *after* the first event in the
+            # timeline, but that is good enough here.
            state_at_timeline_start = (
                await self._state_storage_controller.get_state_ids_for_event(
                    batch.events[0].event_id,
@@ -1200,13 +1205,6 @@ class SyncHandler:
                )
            )
        else:
-            state_at_timeline_end = await self.get_state_at(
-                room_id,
-                stream_position=now_token,
-                state_filter=state_filter,
-                await_full_state=await_full_state,
-            )
-
            state_at_timeline_start = state_at_timeline_end

        state_ids = _calculate_state(
@@ -1223,7 +1221,7 @@ class SyncHandler:
        room_id: str,
        batch: TimelineBatch,
        since_token: StreamToken,
-        now_token: StreamToken,
+        end_token: StreamToken,
        members_to_fetch: Optional[Set[str]],
        timeline_state: StateMap[str],
    ) -> StateMap[str]:
@@ -1239,7 +1237,9 @@ class SyncHandler:
            room_id: The room we are calculating for.
            batch: The timeline batch for the room that will be sent to the user.
            since_token: Token of the end of the previous batch.
-            now_token: Token of the end of the current batch.
+            end_token: Token of the end of the current batch. Normally this will be
+                the same as the global "now_token", but if the user has left the room,
+                the point just after their leave event.
            members_to_fetch: If lazy-loading is enabled, the memberships needed for
                events in the timeline. Otherwise, `None`.
            timeline_state: The contribution to the room state from state events in
@@ -1259,25 +1259,70 @@ class SyncHandler:
            await_full_state = True
            lazy_load_members = False

-        if batch.limited:
-            if batch:
-                state_at_timeline_start = (
-                    await self._state_storage_controller.get_state_ids_for_event(
-                        batch.events[0].event_id,
-                        state_filter=state_filter,
-                        await_full_state=await_full_state,
-                    )
-                )
-            else:
-                # We can get here if the user has ignored the senders of all
-                # the recent events.
-                state_at_timeline_start = await self.get_state_at(
+        # For a non-gappy sync if the events in the timeline are simply a linear
+        # chain (i.e. no merging/branching of the graph), then we know the state
+        # delta between the end of the previous sync and start of the new one is
+        # empty.
+        #
+        # c.f. #16941 for an example of why we can't do this for all non-gappy
+        # syncs.
+        is_linear_timeline = True
+        if batch.events:
+            # We need to make sure the first event in our batch points to the
+            # last event in the previous batch.
+            last_event_id_prev_batch = (
+                await self.store.get_last_event_in_room_before_stream_ordering(
                    room_id,
-                    stream_position=now_token,
+                    end_token=since_token.room_key,
+                )
+            )
+
+            prev_event_id = last_event_id_prev_batch
+            for e in batch.events:
+                if e.prev_event_ids() != [prev_event_id]:
+                    is_linear_timeline = False
+                    break
+                prev_event_id = e.event_id
+
+        if is_linear_timeline and not batch.limited:
+            state_ids: StateMap[str] = {}
+            if lazy_load_members:
+                if members_to_fetch and batch.events:
+                    # We're lazy-loading, so the client might need some more
+                    # member events to understand the events in this timeline.
+                    # So we fish out all the member events corresponding to the
+                    # timeline here. The caller will then dedupe any redundant
+                    # ones.
+
+                    state_ids = await self._state_storage_controller.get_state_ids_for_event(
+                        batch.events[0].event_id,
+                        # we only want members!
+                        state_filter=StateFilter.from_types(
+                            (EventTypes.Member, member) for member in members_to_fetch
+                        ),
+                        await_full_state=False,
+                    )
+            return state_ids
+
+        if batch:
+            state_at_timeline_start = (
+                await self._state_storage_controller.get_state_ids_for_event(
+                    batch.events[0].event_id,
                    state_filter=state_filter,
                    await_full_state=await_full_state,
                )
+            )
+        else:
+            # We can get here if the user has ignored the senders of all
+            # the recent events.
+            state_at_timeline_start = await self.get_state_at(
+                room_id,
+                stream_position=end_token,
+                state_filter=state_filter,
+                await_full_state=await_full_state,
+            )

+        if batch.limited:
            # for now, we disable LL for gappy syncs - see
            # https://github.com/vector-im/riot-web/issues/7211#issuecomment-419976346
            # N.B. this slows down incr syncs as we are now processing way
@@ -1292,58 +1337,28 @@ class SyncHandler:
            # about them).
            state_filter = StateFilter.all()

-            state_at_previous_sync = await self.get_state_at(
-                room_id,
-                stream_position=since_token,
-                state_filter=state_filter,
-                await_full_state=await_full_state,
-            )
+        state_at_previous_sync = await self.get_state_at(
+            room_id,
+            stream_position=since_token,
+            state_filter=state_filter,
+            await_full_state=await_full_state,
+        )

-            if batch:
-                state_at_timeline_end = (
-                    await self._state_storage_controller.get_state_ids_for_event(
-                        batch.events[-1].event_id,
-                        state_filter=state_filter,
-                        await_full_state=await_full_state,
-                    )
-                )
-            else:
-                # We can get here if the user has ignored the senders of all
-                # the recent events.
-                state_at_timeline_end = await self.get_state_at(
-                    room_id,
-                    stream_position=now_token,
-                    state_filter=state_filter,
-                    await_full_state=await_full_state,
-                )
+        state_at_timeline_end = await self.get_state_at(
+            room_id,
+            stream_position=end_token,
+            state_filter=state_filter,
+            await_full_state=await_full_state,
+        )

-            state_ids = _calculate_state(
-                timeline_contains=timeline_state,
-                timeline_start=state_at_timeline_start,
-                timeline_end=state_at_timeline_end,
-                previous_timeline_end=state_at_previous_sync,
-                lazy_load_members=lazy_load_members,
-            )
-        else:
-            state_ids = {}
-            if lazy_load_members:
-                if members_to_fetch and batch.events:
-                    # We're returning an incremental sync, with no
-                    # "gap" since the previous sync, so normally there would be
-                    # no state to return.
-                    # But we're lazy-loading, so the client might need some more
-                    # member events to understand the events in this timeline.
-                    # So we fish out all the member events corresponding to the
-                    # timeline here. The caller will then dedupe any redundant ones.
+        state_ids = _calculate_state(
+            timeline_contains=timeline_state,
+            timeline_start=state_at_timeline_start,
+            timeline_end=state_at_timeline_end,
+            previous_timeline_end=state_at_previous_sync,
+            lazy_load_members=lazy_load_members,
+        )

-                    state_ids = await self._state_storage_controller.get_state_ids_for_event(
-                        batch.events[0].event_id,
-                        # we only want members!
-                        state_filter=StateFilter.from_types(
-                            (EventTypes.Member, member) for member in members_to_fetch
-                        ),
-                        await_full_state=False,
-                    )
        return state_ids

    async def _find_missing_partial_state_memberships(
@@ -2344,6 +2359,7 @@ class SyncHandler:
                        full_state=False,
                        since_token=since_token,
                        upto_token=leave_token,
+                        end_token=leave_token,
                        out_of_band=leave_event.internal_metadata.is_out_of_band_membership(),
                    )
                )
@@ -2381,6 +2397,7 @@ class SyncHandler:
                    full_state=False,
                    since_token=None if newly_joined else since_token,
                    upto_token=prev_batch_token,
+                    end_token=now_token,
                )
            else:
                entry = RoomSyncResultBuilder(
@@ -2391,6 +2408,7 @@ class SyncHandler:
                    full_state=False,
                    since_token=since_token,
                    upto_token=since_token,
+                    end_token=now_token,
                )

            room_entries.append(entry)
@@ -2449,6 +2467,7 @@ class SyncHandler:
                        full_state=True,
                        since_token=since_token,
                        upto_token=now_token,
+                        end_token=now_token,
                    )
                )
            elif event.membership == Membership.INVITE:
@@ -2478,6 +2497,7 @@ class SyncHandler:
                        full_state=True,
                        since_token=since_token,
                        upto_token=leave_token,
+                        end_token=leave_token,
                    )
                )

@@ -2548,6 +2568,7 @@ class SyncHandler:
                {
                    "since_token": since_token,
                    "upto_token": upto_token,
+                    "end_token": room_builder.end_token,
                }
            )

@@ -2621,7 +2642,7 @@ class SyncHandler:
                    batch,
                    sync_config,
                    since_token,
-                    now_token,
+                    room_builder.end_token,
                    full_state=full_state,
                )
            else:
@@ -2781,6 +2802,61 @@ def _calculate_state(
            e for t, e in timeline_start.items() if t[0] == EventTypes.Member
        )

+    # Naively, we would just return the difference between the state at the start
+    # of the timeline (`timeline_start_ids`) and that at the end of the previous sync
+    # (`previous_timeline_end_ids`). However, that fails in the presence of forks in
+    # the DAG.
+    #
+    # For example, consider a DAG such as the following:
+    #
+    #       E1
+    #     ↗    ↖
+    #    |      S2
+    #    |      ↑
+    #  --|------|----
+    #    |      |
+    #    E3     |
+    #     ↖    /
+    #       E4
+    #
+    # ... and a filter that means we only return 2 events, represented by the dashed
+    # horizontal line. Assuming S2 was *not* included in the previous sync, we need to
+    # include it in the `state` section.
+    #
+    # Note that the state at the start of the timeline (E3) does not include S2. So,
+    # to make sure it gets included in the calculation here, we actually look at
+    # the state at the *end* of the timeline, and subtract any events that are present
+    # in the timeline.
+    #
+    # ----------
+    #
+    # Aside 1: You may then wonder if we need to include `timeline_start` in the
+    # calculation. Consider a linear DAG:
+    #
+    #      E1
+    #      ↑
+    #      S2
+    #      ↑
+    #  ----|------
+    #      |
+    #      E3
+    #      ↑
+    #      S4
+    #      ↑
+    #      E5
+    #
+    # ... where S2 and S4 change the same piece of state; and where we have a filter
+    # that returns 3 events (E3, S4, E5). We still need to tell the client about S2,
+    # because it might affect the display of E3. However, the state at the end of the
+    # timeline only tells us about S4; if we don't inspect `timeline_start` we won't
+    # find out about S2.
+    #
+    # (There are yet more complicated cases in which a state event is excluded from the
+    # timeline, but whose effect actually lands in the DAG in the *middle* of the
+    # timeline. We have no way to represent that in the /sync response, and we don't
+    # even try; it is ether omitted or plonked into `state` as if it were at the start
+    # of the timeline, depending on what else is in the timeline.)
+
    state_ids = (
        (timeline_end_ids | timeline_start_ids)
        - previous_timeline_end_ids
@@ -2883,13 +2959,30 @@ class RoomSyncResultBuilder:

    Attributes:
        room_id
+
        rtype: One of `"joined"` or `"archived"`
+
        events: List of events to include in the room (more events may be added
            when generating result).
+
        newly_joined: If the user has newly joined the room
+
        full_state: Whether the full state should be sent in result
+
        since_token: Earliest point to return events from, or None
-        upto_token: Latest point to return events from.
+
+        upto_token: Latest point to return events from. If `events` is populated,
+           this is set to the token at the start of `events`
+
+        end_token: The last point in the timeline that the client should see events
+           from. Normally this will be the same as the global `now_token`, but in
+           the case of rooms where the user has left the room, this will be the point
+           just after their leave event.
+
+           This is used in the calculation of the state which is returned in `state`:
+           any state changes *up to* `end_token` (and not beyond!) which are not
+           reflected in the timeline need to be returned in `state`.
+
        out_of_band: whether the events in the room are "out of band" events
            and the server isn't in the room.
    """
@@ -2901,5 +2994,5 @@ class RoomSyncResultBuilder:
    full_state: bool
    since_token: Optional[StreamToken]
    upto_token: StreamToken
-
+    end_token: StreamToken
    out_of_band: bool = False
@@ -262,7 +262,8 @@ class _ProxyResponseBody(protocol.Protocol):
            self._request.finish()
        else:
            # Abort the underlying request since our remote request also failed.
-            self._request.transport.abortConnection()
+            if self._request.channel:
+                self._request.channel.forceAbortClient()


 class ProxySite(Site):
@@ -153,9 +153,9 @@ def return_json_error(
    # Only respond with an error response if we haven't already started writing,
    # otherwise lets just kill the connection
    if request.startedWriting:
-        if request.transport:
+        if request.channel:
            try:
-                request.transport.abortConnection()
+                request.channel.forceAbortClient()
            except Exception:
                # abortConnection throws if the connection is already closed
                pass
@@ -909,7 +909,18 @@ def set_cors_headers(request: "SynapseRequest") -> None:
    request.setHeader(
        b"Access-Control-Allow-Methods", b"GET, HEAD, POST, PUT, DELETE, OPTIONS"
    )
-    if request.experimental_cors_msc3886:
+    if request.path is not None and request.path.startswith(
+        b"/_matrix/client/unstable/org.matrix.msc4108/rendezvous"
+    ):
+        request.setHeader(
+            b"Access-Control-Allow-Headers",
+            b"Content-Type, If-Match, If-None-Match",
+        )
+        request.setHeader(
+            b"Access-Control-Expose-Headers",
+            b"Synapse-Trace-Id, Server, ETag",
+        )
+    elif request.experimental_cors_msc3886:
        request.setHeader(
            b"Access-Control-Allow-Headers",
            b"X-Requested-With, Content-Type, Authorization, Date, If-Match, If-None-Match",
@@ -19,9 +19,11 @@
 #
 #

-""" This module contains base REST classes for constructing REST servlets. """
+"""This module contains base REST classes for constructing REST servlets."""
+
 import enum
 import logging
+import urllib.parse as urlparse
 from http import HTTPStatus
 from typing import (
    TYPE_CHECKING,
@@ -65,17 +67,49 @@ def parse_integer(request: Request, name: str, default: int) -> int: ...


@overload
-def parse_integer(request: Request, name: str, *, required: Literal[True]) -> int: ...
+def parse_integer(
+    request: Request, name: str, *, default: int, negative: bool
+) -> int: ...


@overload
 def parse_integer(
-    request: Request, name: str, default: Optional[int] = None, required: bool = False
+    request: Request, name: str, *, default: int, negative: bool = False
+) -> int: ...
+
+
+@overload
+def parse_integer(
+    request: Request, name: str, *, required: Literal[True], negative: bool = False
+) -> int: ...
+
+
+@overload
+def parse_integer(
+    request: Request, name: str, *, default: Literal[None], negative: bool = False
+) -> None: ...
+
+
+@overload
+def parse_integer(request: Request, name: str, *, negative: bool) -> Optional[int]: ...
+
+
+@overload
+def parse_integer(
+    request: Request,
+    name: str,
+    default: Optional[int] = None,
+    required: bool = False,
+    negative: bool = False,
 ) -> Optional[int]: ...


 def parse_integer(
-    request: Request, name: str, default: Optional[int] = None, required: bool = False
+    request: Request,
+    name: str,
+    default: Optional[int] = None,
+    required: bool = False,
+    negative: bool = False,
 ) -> Optional[int]:
    """Parse an integer parameter from the request string

@@ -85,16 +119,17 @@ def parse_integer(
        default: value to use if the parameter is absent, defaults to None.
        required: whether to raise a 400 SynapseError if the parameter is absent,
            defaults to False.
-
+        negative: whether to allow negative integers, defaults to True.
    Returns:
        An int value or the default.

    Raises:
-        SynapseError: if the parameter is absent and required, or if the
-            parameter is present and not an integer.
+        SynapseError: if the parameter is absent and required, if the
+            parameter is present and not an integer, or if the
+            parameter is illegitimate negative.
    """
    args: Mapping[bytes, Sequence[bytes]] = request.args  # type: ignore
-    return parse_integer_from_args(args, name, default, required)
+    return parse_integer_from_args(args, name, default, required, negative)


@overload
@@ -120,6 +155,7 @@ def parse_integer_from_args(
    name: str,
    default: Optional[int] = None,
    required: bool = False,
+    negative: bool = False,
 ) -> Optional[int]: ...


@@ -128,6 +164,7 @@ def parse_integer_from_args(
    name: str,
    default: Optional[int] = None,
    required: bool = False,
+    negative: bool = True,
 ) -> Optional[int]:
    """Parse an integer parameter from the request string

@@ -137,33 +174,37 @@ def parse_integer_from_args(
        default: value to use if the parameter is absent, defaults to None.
        required: whether to raise a 400 SynapseError if the parameter is absent,
            defaults to False.
+        negative: whether to allow negative integers, defaults to True.

    Returns:
        An int value or the default.

    Raises:
-        SynapseError: if the parameter is absent and required, or if the
-            parameter is present and not an integer.
+        SynapseError: if the parameter is absent and required, if the
+            parameter is present and not an integer, or if the
+            parameter is illegitimate negative.
    """
    name_bytes = name.encode("ascii")

-    if name_bytes in args:
-        try:
-            return int(args[name_bytes][0])
-        except Exception:
-            message = "Query parameter %r must be an integer" % (name,)
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST, message, errcode=Codes.INVALID_PARAM
-            )
-    else:
-        if required:
-            message = "Missing integer query parameter %r" % (name,)
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST, message, errcode=Codes.MISSING_PARAM
-            )
-        else:
+    if name_bytes not in args:
+        if not required:
            return default

+        message = f"Missing required integer query parameter {name}"
+        raise SynapseError(HTTPStatus.BAD_REQUEST, message, errcode=Codes.MISSING_PARAM)
+
+    try:
+        integer = int(args[name_bytes][0])
+    except Exception:
+        message = f"Query parameter {name} must be an integer"
+        raise SynapseError(HTTPStatus.BAD_REQUEST, message, errcode=Codes.INVALID_PARAM)
+
+    if not negative and integer < 0:
+        message = f"Query parameter {name} must be a positive integer."
+        raise SynapseError(HTTPStatus.BAD_REQUEST, message, errcode=Codes.INVALID_PARAM)
+
+    return integer
+

@overload
 def parse_boolean(request: Request, name: str, default: bool) -> bool: ...
@@ -410,6 +451,87 @@ def parse_string(
    )


+def parse_json(
+    request: Request,
+    name: str,
+    default: Optional[dict] = None,
+    required: bool = False,
+    encoding: str = "ascii",
+) -> Optional[JsonDict]:
+    """
+    Parse a JSON parameter from the request query string.
+
+    Args:
+        request: the twisted HTTP request.
+        name: the name of the query parameter.
+        default: value to use if the parameter is absent,
+            defaults to None.
+        required: whether to raise a 400 SynapseError if the
+           parameter is absent, defaults to False.
+        encoding: The encoding to decode the string content with.
+
+    Returns:
+        A JSON value, or `default` if the named query parameter was not found
+        and `required` was False.
+
+    Raises:
+        SynapseError if the parameter is absent and required, or if the
+            parameter is present and not a JSON object.
+    """
+    args: Mapping[bytes, Sequence[bytes]] = request.args  # type: ignore
+    return parse_json_from_args(
+        args,
+        name,
+        default,
+        required=required,
+        encoding=encoding,
+    )
+
+
+def parse_json_from_args(
+    args: Mapping[bytes, Sequence[bytes]],
+    name: str,
+    default: Optional[dict] = None,
+    required: bool = False,
+    encoding: str = "ascii",
+) -> Optional[JsonDict]:
+    """
+    Parse a JSON parameter from the request query string.
+
+    Args:
+        args: a mapping of request args as bytes to a list of bytes (e.g. request.args).
+        name: the name of the query parameter.
+        default: value to use if the parameter is absent,
+            defaults to None.
+        required: whether to raise a 400 SynapseError if the
+            parameter is absent, defaults to False.
+        encoding: the encoding to decode the string content with.
+
+        A JSON value, or `default` if the named query parameter was not found
+        and `required` was False.
+
+    Raises:
+        SynapseError if the parameter is absent and required, or if the
+            parameter is present and not a JSON object.
+    """
+    name_bytes = name.encode("ascii")
+
+    if name_bytes not in args:
+        if not required:
+            return default
+
+        message = f"Missing required integer query parameter {name}"
+        raise SynapseError(HTTPStatus.BAD_REQUEST, message, errcode=Codes.MISSING_PARAM)
+
+    json_str = parse_string_from_args(args, name, required=True, encoding=encoding)
+
+    try:
+        return json_decoder.decode(urlparse.unquote(json_str))
+    except Exception:
+        message = f"Query parameter {name} must be a valid JSON object"
+        raise SynapseError(HTTPStatus.BAD_REQUEST, message, errcode=Codes.NOT_JSON)
+
+
 EnumT = TypeVar("EnumT", bound=enum.Enum)


@@ -150,7 +150,8 @@ class SynapseRequest(Request):
                self.get_method(),
                self.get_redacted_uri(),
            )
-            self.transport.abortConnection()
+            if self.channel:
+                self.channel.forceAbortClient()
            return
        super().handleContentChunk(data)

@@ -77,5 +77,46 @@ class ReplicationRemovePusherRestServlet(ReplicationEndpoint):
        return 200, {}


+class ReplicationCopyPusherRestServlet(ReplicationEndpoint):
+    """Copies push rules from an old room to new room.
+
+    Request format:
+
+        POST /_synapse/replication/copy_push_rules/:user_id/:old_room_id/:new_room_id
+
+        {}
+
+    """
+
+    NAME = "copy_push_rules"
+    PATH_ARGS = ("user_id", "old_room_id", "new_room_id")
+    CACHE = False
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__(hs)
+
+        self._store = hs.get_datastores().main
+
+    @staticmethod
+    async def _serialize_payload(user_id: str, old_room_id: str, new_room_id: str) -> JsonDict:  # type: ignore[override]
+        return {}
+
+    async def _handle_request(  # type: ignore[override]
+        self,
+        request: Request,
+        content: JsonDict,
+        user_id: str,
+        old_room_id: str,
+        new_room_id: str,
+    ) -> Tuple[int, JsonDict]:
+
+        await self._store.copy_push_rules_from_room_to_room_for_user(
+            old_room_id, new_room_id, user_id
+        )
+
+        return 200, {}
+
+
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
    ReplicationRemovePusherRestServlet(hs).register(http_server)
+    ReplicationCopyPusherRestServlet(hs).register(http_server)
@@ -66,6 +66,7 @@ from synapse.replication.tcp.streams import (
    FederationStream,
    PresenceFederationStream,
    PresenceStream,
+    PushRulesStream,
    ReceiptsStream,
    Stream,
    ToDeviceStream,
@@ -178,6 +179,12 @@ class ReplicationCommandHandler:

                continue

+            if isinstance(stream, PushRulesStream):
+                if hs.get_instance_name() in hs.config.worker.writers.push_rules:
+                    self._streams_to_replicate.append(stream)
+
+                continue
+
            # Only add any other streams if we're on master.
            if hs.config.worker.worker_app is not None:
                continue
@@ -23,7 +23,7 @@ from http import HTTPStatus
 from typing import TYPE_CHECKING, Tuple

 from synapse.api.constants import Direction
-from synapse.api.errors import Codes, NotFoundError, SynapseError
+from synapse.api.errors import NotFoundError, SynapseError
 from synapse.federation.transport.server import Authenticator
 from synapse.http.servlet import RestServlet, parse_enum, parse_integer, parse_string
 from synapse.http.site import SynapseRequest
@@ -61,22 +61,8 @@ class ListDestinationsRestServlet(RestServlet):
    async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
        await assert_requester_is_admin(self._auth, request)

-        start = parse_integer(request, "from", default=0)
-        limit = parse_integer(request, "limit", default=100)
-
-        if start < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        if limit < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter limit must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
+        start = parse_integer(request, "from", default=0, negative=False)
+        limit = parse_integer(request, "limit", default=100, negative=False)

        destination = parse_string(request, "destination")

@@ -195,22 +181,8 @@ class DestinationMembershipRestServlet(RestServlet):
        if not await self._store.is_destination_known(destination):
            raise NotFoundError("Unknown destination")

-        start = parse_integer(request, "from", default=0)
-        limit = parse_integer(request, "limit", default=100)
-
-        if start < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        if limit < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter limit must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
+        start = parse_integer(request, "from", default=0, negative=False)
+        limit = parse_integer(request, "limit", default=100, negative=False)

        direction = parse_enum(request, "dir", Direction, default=Direction.FORWARDS)

@@ -311,29 +311,17 @@ class DeleteMediaByDateSize(RestServlet):
    ) -> Tuple[int, JsonDict]:
        await assert_requester_is_admin(self.auth, request)

-        before_ts = parse_integer(request, "before_ts", required=True)
-        size_gt = parse_integer(request, "size_gt", default=0)
+        before_ts = parse_integer(request, "before_ts", required=True, negative=False)
+        size_gt = parse_integer(request, "size_gt", default=0, negative=False)
        keep_profiles = parse_boolean(request, "keep_profiles", default=True)

-        if before_ts < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter before_ts must be a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-        elif before_ts < 30000000000:  # Dec 1970 in milliseconds, Aug 2920 in seconds
+        if before_ts < 30000000000:  # Dec 1970 in milliseconds, Aug 2920 in seconds
            raise SynapseError(
                HTTPStatus.BAD_REQUEST,
                "Query parameter before_ts you provided is from the year 1970. "
                + "Double check that you are providing a timestamp in milliseconds.",
                errcode=Codes.INVALID_PARAM,
            )
-        if size_gt < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter size_gt must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )

        # This check is useless, we keep it for the legacy endpoint only.
        if server_name is not None and self.server_name != server_name:
@@ -389,22 +377,8 @@ class UserMediaRestServlet(RestServlet):
        if user is None:
            raise NotFoundError("Unknown user")

-        start = parse_integer(request, "from", default=0)
-        limit = parse_integer(request, "limit", default=100)
-
-        if start < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        if limit < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter limit must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
+        start = parse_integer(request, "from", default=0, negative=False)
+        limit = parse_integer(request, "limit", default=100, negative=False)

        # If neither `order_by` nor `dir` is set, set the default order
        # to newest media is on top for backward compatibility.
@@ -447,22 +421,8 @@ class UserMediaRestServlet(RestServlet):
        if user is None:
            raise NotFoundError("Unknown user")

-        start = parse_integer(request, "from", default=0)
-        limit = parse_integer(request, "limit", default=100)
-
-        if start < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        if limit < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter limit must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
+        start = parse_integer(request, "from", default=0, negative=False)
+        limit = parse_integer(request, "limit", default=100, negative=False)

        # If neither `order_by` nor `dir` is set, set the default order
        # to newest media is on top for backward compatibility.
@@ -21,7 +21,6 @@
 import logging
 from http import HTTPStatus
 from typing import TYPE_CHECKING, List, Optional, Tuple, cast
-from urllib import parse as urlparse

 import attr

@@ -38,6 +37,7 @@ from synapse.http.servlet import (
    assert_params_in_dict,
    parse_enum,
    parse_integer,
+    parse_json,
    parse_json_object_from_request,
    parse_string,
 )
@@ -51,7 +51,6 @@ from synapse.storage.databases.main.room import RoomSortOrder
 from synapse.streams.config import PaginationConfig
 from synapse.types import JsonDict, RoomID, ScheduledTask, UserID, create_requester
 from synapse.types.state import StateFilter
-from synapse.util import json_decoder

 if TYPE_CHECKING:
    from synapse.api.auth import Auth
@@ -776,14 +775,8 @@ class RoomEventContextServlet(RestServlet):
        limit = parse_integer(request, "limit", default=10)

        # picking the API shape for symmetry with /messages
-        filter_str = parse_string(request, "filter", encoding="utf-8")
-        if filter_str:
-            filter_json = urlparse.unquote(filter_str)
-            event_filter: Optional[Filter] = Filter(
-                self._hs, json_decoder.decode(filter_json)
-            )
-        else:
-            event_filter = None
+        filter_json = parse_json(request, "filter", encoding="utf-8")
+        event_filter = Filter(self._hs, filter_json) if filter_json else None

        event_context = await self.room_context_handler.get_event_context(
            requester,
@@ -914,21 +907,16 @@ class RoomMessagesRestServlet(RestServlet):
        )
        # Twisted will have processed the args by now.
        assert request.args is not None
+
+        filter_json = parse_json(request, "filter", encoding="utf-8")
+        event_filter = Filter(self._hs, filter_json) if filter_json else None
+
        as_client_event = b"raw" not in request.args
-        filter_str = parse_string(request, "filter", encoding="utf-8")
-        if filter_str:
-            filter_json = urlparse.unquote(filter_str)
-            event_filter: Optional[Filter] = Filter(
-                self._hs, json_decoder.decode(filter_json)
-            )
-            if (
-                event_filter
-                and event_filter.filter_json.get("event_format", "client")
-                == "federation"
-            ):
-                as_client_event = False
-        else:
-            event_filter = None
+        if (
+            event_filter
+            and event_filter.filter_json.get("event_format", "client") == "federation"
+        ):
+            as_client_event = False

        msgs = await self._pagination_handler.get_messages(
            room_id=room_id,
@@ -63,38 +63,12 @@ class UserMediaStatisticsRestServlet(RestServlet):
            ),
        )

-        start = parse_integer(request, "from", default=0)
-        if start < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
+        start = parse_integer(request, "from", default=0, negative=False)
+        limit = parse_integer(request, "limit", default=100, negative=False)
+        from_ts = parse_integer(request, "from_ts", default=0, negative=False)
+        until_ts = parse_integer(request, "until_ts", negative=False)

-        limit = parse_integer(request, "limit", default=100)
-        if limit < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter limit must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        from_ts = parse_integer(request, "from_ts", default=0)
-        if from_ts < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from_ts must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        until_ts = parse_integer(request, "until_ts")
        if until_ts is not None:
-            if until_ts < 0:
-                raise SynapseError(
-                    HTTPStatus.BAD_REQUEST,
-                    "Query parameter until_ts must be a string representing a positive integer.",
-                    errcode=Codes.INVALID_PARAM,
-                )
            if until_ts <= from_ts:
                raise SynapseError(
                    HTTPStatus.BAD_REQUEST,
@@ -90,22 +90,8 @@ class UsersRestServletV2(RestServlet):
    async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
        await assert_requester_is_admin(self.auth, request)

-        start = parse_integer(request, "from", default=0)
-        limit = parse_integer(request, "limit", default=100)
-
-        if start < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter from must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
-
-        if limit < 0:
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Query parameter limit must be a string representing a positive integer.",
-                errcode=Codes.INVALID_PARAM,
-            )
+        start = parse_integer(request, "from", default=0, negative=False)
+        limit = parse_integer(request, "limit", default=100, negative=False)

        user_id = parse_string(request, "user_id")
        name = parse_string(request, "name", encoding="utf-8")
@@ -81,8 +81,7 @@ class AccountDataServlet(RestServlet):
            raise AuthError(403, "Cannot add account data for other users.")

        # Raise an error if the account data type cannot be set directly.
-        if self._hs.config.experimental.msc4010_push_rules_account_data:
-            _check_can_set_account_data_type(account_data_type)
+        _check_can_set_account_data_type(account_data_type)

        body = parse_json_object_from_request(request)

@@ -108,10 +107,7 @@ class AccountDataServlet(RestServlet):
            raise AuthError(403, "Cannot get account data for other users.")

        # Push rules are stored in a separate table and must be queried separately.
-        if (
-            self._hs.config.experimental.msc4010_push_rules_account_data
-            and account_data_type == AccountDataTypes.PUSH_RULES
-        ):
+        if account_data_type == AccountDataTypes.PUSH_RULES:
            account_data: Optional[JsonMapping] = (
                await self._push_rules_handler.push_rules_for_user(requester.user)
            )
@@ -162,8 +158,7 @@ class UnstableAccountDataServlet(RestServlet):
            raise AuthError(403, "Cannot delete account data for other users.")

        # Raise an error if the account data type cannot be set directly.
-        if self._hs.config.experimental.msc4010_push_rules_account_data:
-            _check_can_set_account_data_type(account_data_type)
+        _check_can_set_account_data_type(account_data_type)

        await self.handler.remove_account_data_for_user(user_id, account_data_type)

@@ -209,15 +204,7 @@ class RoomAccountDataServlet(RestServlet):
            )

        # Raise an error if the account data type cannot be set directly.
-        if self._hs.config.experimental.msc4010_push_rules_account_data:
-            _check_can_set_account_data_type(account_data_type)
-        elif account_data_type == ReceiptTypes.FULLY_READ:
-            raise SynapseError(
-                405,
-                "Cannot set m.fully_read through this API."
-                " Use /rooms/!roomId:server.name/read_markers",
-                Codes.BAD_JSON,
-            )
+        _check_can_set_account_data_type(account_data_type)

        body = parse_json_object_from_request(request)

@@ -256,10 +243,7 @@ class RoomAccountDataServlet(RestServlet):
            )

        # Room-specific push rules are not currently supported.
-        if (
-            self._hs.config.experimental.msc4010_push_rules_account_data
-            and account_data_type == AccountDataTypes.PUSH_RULES
-        ):
+        if account_data_type == AccountDataTypes.PUSH_RULES:
            account_data: Optional[JsonMapping] = {}
        else:
            account_data = await self.store.get_account_data_for_room_and_type(
@@ -317,8 +301,7 @@ class UnstableRoomAccountDataServlet(RestServlet):
            )

        # Raise an error if the account data type cannot be set directly.
-        if self._hs.config.experimental.msc4010_push_rules_account_data:
-            _check_can_set_account_data_type(account_data_type)
+        _check_can_set_account_data_type(account_data_type)

        await self.handler.remove_account_data_for_room(
            user_id, room_id, account_data_type
@@ -409,7 +409,18 @@ class SigningKeyUploadServlet(RestServlet):
            # But first-time setup is fine

        elif self.hs.config.experimental.msc3967_enabled:
-            # If we already have a master key then cross signing is set up and we require UIA to reset
+            # MSC3967 allows this endpoint to 200 OK for idempotency. Resending exactly the same
+            # keys should just 200 OK without doing a UIA prompt.
+            keys_are_different = await self.e2e_keys_handler.has_different_keys(
+                user_id, body
+            )
+            if not keys_are_different:
+                # FIXME: we do not fallthrough to upload_signing_keys_for_user because confusingly
+                # if we do, we 500 as it looks like it tries to INSERT the same key twice, causing a
+                # unique key constraint violation. This sounds like a bug?
+                return 200, {}
+            # the keys are different, is x-signing set up? If no, then the keys don't exist which is
+            # why they are different. If yes, then we need to UIA to change them.
            if is_cross_signing_setup:
                await self.auth_handler.validate_user_via_ui_auth(
                    requester,
@@ -420,7 +431,6 @@ class SigningKeyUploadServlet(RestServlet):
                    can_skip_ui_auth=False,
                )
            # Otherwise we don't require UIA since we are setting up cross signing for first time
-
        else:
            # Previous behaviour is to always require UIA but allow it to be skipped
            await self.auth_handler.validate_user_via_ui_auth(
@@ -59,12 +59,14 @@ class PushRuleRestServlet(RestServlet):
        self.auth = hs.get_auth()
        self.store = hs.get_datastores().main
        self.notifier = hs.get_notifier()
-        self._is_worker = hs.config.worker.worker_app is not None
+        self._is_push_worker = (
+            hs.get_instance_name() in hs.config.worker.writers.push_rules
+        )
        self._push_rules_handler = hs.get_push_rules_handler()
        self._push_rule_linearizer = Linearizer(name="push_rules")

    async def on_PUT(self, request: SynapseRequest, path: str) -> Tuple[int, JsonDict]:
-        if self._is_worker:
+        if not self._is_push_worker:
            raise Exception("Cannot handle PUT /push_rules on worker")

        requester = await self.auth.get_user_by_req(request)
@@ -137,7 +139,7 @@ class PushRuleRestServlet(RestServlet):
    async def on_DELETE(
        self, request: SynapseRequest, path: str
    ) -> Tuple[int, JsonDict]:
-        if self._is_worker:
+        if not self._is_push_worker:
            raise Exception("Cannot handle DELETE /push_rules on worker")

        requester = await self.auth.get_user_by_req(request)
@@ -55,7 +55,6 @@ class RelationPaginationServlet(RestServlet):
        self.auth = hs.get_auth()
        self._store = hs.get_datastores().main
        self._relations_handler = hs.get_relations_handler()
-        self._support_recurse = hs.config.experimental.msc3981_recurse_relations

    async def on_GET(
        self,
@@ -70,12 +69,9 @@ class RelationPaginationServlet(RestServlet):
        pagination_config = await PaginationConfig.from_request(
            self._store, request, default_limit=5, default_dir=Direction.BACKWARDS
        )
-        if self._support_recurse:
-            recurse = parse_boolean(request, "recurse", default=False) or parse_boolean(
-                request, "org.matrix.msc3981.recurse", default=False
-            )
-        else:
-            recurse = False
+        recurse = parse_boolean(request, "recurse", default=False) or parse_boolean(
+            request, "org.matrix.msc3981.recurse", default=False
+        )

        # The unstable version of this API returns an extra field for client
        # compatibility, see https://github.com/matrix-org/synapse/issues/12930.
@@ -2,7 +2,7 @@
 # This file is licensed under the Affero General Public License (AGPL) version 3.
 #
 # Copyright 2022 The Matrix.org Foundation C.I.C.
-# Copyright (C) 2023 New Vector, Ltd
+# Copyright (C) 2023-2024 New Vector, Ltd
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -34,7 +34,7 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)


-class RendezvousServlet(RestServlet):
+class MSC3886RendezvousServlet(RestServlet):
    """
    This is a placeholder implementation of [MSC3886](https://github.com/matrix-org/matrix-spec-proposals/pull/3886)
    simple client rendezvous capability that is used by the "Sign in with QR" functionality.
@@ -76,6 +76,30 @@ class RendezvousServlet(RestServlet):
    # PUT, GET and DELETE are not implemented as they should be fulfilled by the redirect target.


+class MSC4108DelegationRendezvousServlet(RestServlet):
+    PATTERNS = client_patterns(
+        "/org.matrix.msc4108/rendezvous$", releases=[], v1=False, unstable=True
+    )
+
+    def __init__(self, hs: "HomeServer"):
+        super().__init__()
+        redirection_target: Optional[str] = (
+            hs.config.experimental.msc4108_delegation_endpoint
+        )
+        assert (
+            redirection_target is not None
+        ), "Servlet is only registered if there is a delegation target"
+        self.endpoint = redirection_target.encode("utf-8")
+
+    async def on_POST(self, request: SynapseRequest) -> None:
+        respond_with_redirect(
+            request, self.endpoint, statusCode=TEMPORARY_REDIRECT, cors=True
+        )
+
+
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:
    if hs.config.experimental.msc3886_endpoint is not None:
-        RendezvousServlet(hs).register(http_server)
+        MSC3886RendezvousServlet(hs).register(http_server)
+
+    if hs.config.experimental.msc4108_delegation_endpoint is not None:
+        MSC4108DelegationRendezvousServlet(hs).register(http_server)
@@ -52,6 +52,7 @@ from synapse.http.servlet import (
    parse_boolean,
    parse_enum,
    parse_integer,
+    parse_json,
    parse_json_object_from_request,
    parse_string,
    parse_strings_from_args,
@@ -65,7 +66,6 @@ from synapse.rest.client.transactions import HttpTransactionCache
 from synapse.streams.config import PaginationConfig
 from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID, UserID
 from synapse.types.state import StateFilter
-from synapse.util import json_decoder
 from synapse.util.cancellation import cancellable
 from synapse.util.stringutils import parse_and_validate_server_name, random_string

@@ -499,7 +499,7 @@ class PublicRoomListRestServlet(RestServlet):
            if server:
                raise e

-        limit: Optional[int] = parse_integer(request, "limit", 0)
+        limit: Optional[int] = parse_integer(request, "limit", 0, negative=False)
        since_token = parse_string(request, "since")

        if limit == 0:
@@ -703,21 +703,16 @@ class RoomMessageListRestServlet(RestServlet):
        )
        # Twisted will have processed the args by now.
        assert request.args is not None
+
+        filter_json = parse_json(request, "filter", encoding="utf-8")
+        event_filter = Filter(self._hs, filter_json) if filter_json else None
+
        as_client_event = b"raw" not in request.args
-        filter_str = parse_string(request, "filter", encoding="utf-8")
-        if filter_str:
-            filter_json = urlparse.unquote(filter_str)
-            event_filter: Optional[Filter] = Filter(
-                self._hs, json_decoder.decode(filter_json)
-            )
-            if (
-                event_filter
-                and event_filter.filter_json.get("event_format", "client")
-                == "federation"
-            ):
-                as_client_event = False
-        else:
-            event_filter = None
+        if (
+            event_filter
+            and event_filter.filter_json.get("event_format", "client") == "federation"
+        ):
+            as_client_event = False

        msgs = await self.pagination_handler.get_messages(
            room_id=room_id,
@@ -898,14 +893,8 @@ class RoomEventContextServlet(RestServlet):
        limit = parse_integer(request, "limit", default=10)

        # picking the API shape for symmetry with /messages
-        filter_str = parse_string(request, "filter", encoding="utf-8")
-        if filter_str:
-            filter_json = urlparse.unquote(filter_str)
-            event_filter: Optional[Filter] = Filter(
-                self._hs, json_decoder.decode(filter_json)
-            )
-        else:
-            event_filter = None
+        filter_json = parse_json(request, "filter", encoding="utf-8")
+        event_filter = Filter(self._hs, filter_json) if filter_json else None

        event_context = await self.room_context_handler.get_event_context(
            requester, room_id, event_id, limit, event_filter
@@ -132,13 +132,17 @@ class VersionsRestServlet(RestServlet):
                    # Adds support for relation-based redactions as per MSC3912.
                    "org.matrix.msc3912": self.config.experimental.msc3912_enabled,
                    # Whether recursively provide relations is supported.
-                    "org.matrix.msc3981": self.config.experimental.msc3981_recurse_relations,
+                    # TODO This is no longer needed once unstable MSC3981 does not need to be supported.
+                    "org.matrix.msc3981": True,
                    # Adds support for deleting account data.
                    "org.matrix.msc3391": self.config.experimental.msc3391_enabled,
                    # Allows clients to inhibit profile update propagation.
                    "org.matrix.msc4069": self.config.experimental.msc4069_profile_inhibit_propagation,
                    # Allows clients to handle push for encrypted events.
                    "org.matrix.msc4028": self.config.experimental.msc4028_push_encrypted_events,
+                    # MSC4108: Mechanism to allow OIDC sign in and E2EE set up via QR code
+                    "org.matrix.msc4108": self.config.experimental.msc4108_delegation_endpoint
+                    is not None,
                },
            },
        )
@@ -72,9 +72,6 @@ class PreviewUrlResource(RestServlet):
        # XXX: if get_user_by_req fails, what should we do in an async render?
        requester = await self.auth.get_user_by_req(request)
        url = parse_string(request, "url", required=True)
-        ts = parse_integer(request, "ts")
-        if ts is None:
-            ts = self.clock.time_msec()
-
+        ts = parse_integer(request, "ts", default=self.clock.time_msec())
        og = await self.url_previewer.preview(url, requester.user, ts)
        respond_with_json_bytes(request, 200, og, send_cors=True)
@@ -63,7 +63,7 @@ from .openid import OpenIdStore
 from .presence import PresenceStore
 from .profile import ProfileStore
 from .purge_events import PurgeEventsStore
-from .push_rule import PushRuleStore
+from .push_rule import PushRulesWorkerStore
 from .pusher import PusherStore
 from .receipts import ReceiptsStore
 from .registration import RegistrationStore
@@ -130,7 +130,6 @@ class DataStore(
    RejectionsStore,
    FilteringWorkerStore,
    PusherStore,
-    PushRuleStore,
    ApplicationServiceTransactionStore,
    EventPushActionsStore,
    ServerMetricsStore,
@@ -140,6 +139,7 @@ class DataStore(
    SearchStore,
    TagsStore,
    AccountDataStore,
+    PushRulesWorkerStore,
    StreamWorkerStore,
    OpenIdStore,
    ClientIpWorkerStore,
@@ -27,6 +27,7 @@ from typing import (
    Collection,
    Dict,
    FrozenSet,
+    Generator,
    Iterable,
    List,
    Optional,
@@ -279,64 +280,16 @@ class EventFederationWorkerStore(SignatureWorkerStore, EventsWorkerStore, SQLBas

        # Now we look up all links for the chains we have, adding chains that
        # are reachable from any event.
-        #
-        # This query is structured to first get all chain IDs reachable, and
-        # then pull out all links from those chains. This does pull out more
-        # rows than is strictly necessary, however there isn't a way of
-        # structuring the recursive part of query to pull out the links without
-        # also returning large quantities of redundant data (which can make it a
-        # lot slower).
-        sql = """
-            WITH RECURSIVE links(chain_id) AS (
-                SELECT
-                    DISTINCT origin_chain_id
-                FROM event_auth_chain_links WHERE %s
-                UNION
-                SELECT
-                    target_chain_id
-                FROM event_auth_chain_links
-                INNER JOIN links ON (chain_id = origin_chain_id)
-            )
-            SELECT
-                origin_chain_id, origin_sequence_number,
-                target_chain_id, target_sequence_number
-            FROM links
-            INNER JOIN event_auth_chain_links ON (chain_id = origin_chain_id)
-        """

        # A map from chain ID to max sequence number *reachable* from any event ID.
        chains: Dict[int, int] = {}
-
-        # Add all linked chains reachable from initial set of chains.
-        chains_to_fetch = set(event_chains.keys())
-        while chains_to_fetch:
-            batch2 = tuple(itertools.islice(chains_to_fetch, 1000))
-            chains_to_fetch.difference_update(batch2)
-            clause, args = make_in_list_sql_clause(
-                txn.database_engine, "origin_chain_id", batch2
-            )
-            txn.execute(sql % (clause,), args)
-
-            links: Dict[int, List[Tuple[int, int, int]]] = {}
-
-            for (
-                origin_chain_id,
-                origin_sequence_number,
-                target_chain_id,
-                target_sequence_number,
-            ) in txn:
-                links.setdefault(origin_chain_id, []).append(
-                    (origin_sequence_number, target_chain_id, target_sequence_number)
-                )
-
+        for links in self._get_chain_links(txn, set(event_chains.keys())):
            for chain_id in links:
                if chain_id not in event_chains:
                    continue

                _materialize(chain_id, event_chains[chain_id], links, chains)

-            chains_to_fetch.difference_update(chains)
-
        # Add the initial set of chains, excluding the sequence corresponding to
        # initial event.
        for chain_id, seq_no in event_chains.items():
@@ -380,6 +333,68 @@ class EventFederationWorkerStore(SignatureWorkerStore, EventsWorkerStore, SQLBas

        return results

+    @classmethod
+    def _get_chain_links(
+        cls, txn: LoggingTransaction, chains_to_fetch: Set[int]
+    ) -> Generator[Dict[int, List[Tuple[int, int, int]]], None, None]:
+        """Fetch all auth chain links from the given set of chains, and all
+        links from those chains, recursively.
+
+        Note: This may return links that are not reachable from the given
+        chains.
+
+        Returns a generator that produces dicts from origin chain ID to 3-tuple
+        of origin sequence number, target chain ID and target sequence number.
+        """
+
+        # This query is structured to first get all chain IDs reachable, and
+        # then pull out all links from those chains. This does pull out more
+        # rows than is strictly necessary, however there isn't a way of
+        # structuring the recursive part of query to pull out the links without
+        # also returning large quantities of redundant data (which can make it a
+        # lot slower).
+        sql = """
+            WITH RECURSIVE links(chain_id) AS (
+                SELECT
+                    DISTINCT origin_chain_id
+                FROM event_auth_chain_links WHERE %s
+                UNION
+                SELECT
+                    target_chain_id
+                FROM event_auth_chain_links
+                INNER JOIN links ON (chain_id = origin_chain_id)
+            )
+            SELECT
+                origin_chain_id, origin_sequence_number,
+                target_chain_id, target_sequence_number
+            FROM links
+            INNER JOIN event_auth_chain_links ON (chain_id = origin_chain_id)
+        """
+
+        while chains_to_fetch:
+            batch2 = tuple(itertools.islice(chains_to_fetch, 1000))
+            chains_to_fetch.difference_update(batch2)
+            clause, args = make_in_list_sql_clause(
+                txn.database_engine, "origin_chain_id", batch2
+            )
+            txn.execute(sql % (clause,), args)
+
+            links: Dict[int, List[Tuple[int, int, int]]] = {}
+
+            for (
+                origin_chain_id,
+                origin_sequence_number,
+                target_chain_id,
+                target_sequence_number,
+            ) in txn:
+                links.setdefault(origin_chain_id, []).append(
+                    (origin_sequence_number, target_chain_id, target_sequence_number)
+                )
+
+            chains_to_fetch.difference_update(links)
+
+            yield links
+
    def _get_auth_chain_ids_txn(
        self, txn: LoggingTransaction, event_ids: Collection[str], include_given: bool
    ) -> Set[str]:
@@ -564,53 +579,9 @@ class EventFederationWorkerStore(SignatureWorkerStore, EventsWorkerStore, SQLBas

        # Now we look up all links for the chains we have, adding chains that
        # are reachable from any event.
-        #
-        # This query is structured to first get all chain IDs reachable, and
-        # then pull out all links from those chains. This does pull out more
-        # rows than is strictly necessary, however there isn't a way of
-        # structuring the recursive part of query to pull out the links without
-        # also returning large quantities of redundant data (which can make it a
-        # lot slower).
-        sql = """
-            WITH RECURSIVE links(chain_id) AS (
-                SELECT
-                    DISTINCT origin_chain_id
-                FROM event_auth_chain_links WHERE %s
-                UNION
-                SELECT
-                    target_chain_id
-                FROM event_auth_chain_links
-                INNER JOIN links ON (chain_id = origin_chain_id)
-            )
-            SELECT
-                origin_chain_id, origin_sequence_number,
-                target_chain_id, target_sequence_number
-            FROM links
-            INNER JOIN event_auth_chain_links ON (chain_id = origin_chain_id)
-        """
-
-        # (We need to take a copy of `seen_chains` as we want to mutate it in
-        # the loop)
-        chains_to_fetch = set(seen_chains)
-        while chains_to_fetch:
-            batch2 = tuple(itertools.islice(chains_to_fetch, 1000))
-            clause, args = make_in_list_sql_clause(
-                txn.database_engine, "origin_chain_id", batch2
-            )
-            txn.execute(sql % (clause,), args)
-
-            links: Dict[int, List[Tuple[int, int, int]]] = {}
-
-            for (
-                origin_chain_id,
-                origin_sequence_number,
-                target_chain_id,
-                target_sequence_number,
-            ) in txn:
-                links.setdefault(origin_chain_id, []).append(
-                    (origin_sequence_number, target_chain_id, target_sequence_number)
-                )

+        # (We need to take a copy of `seen_chains` as the function mutates it)
+        for links in self._get_chain_links(txn, set(seen_chains)):
            for chains in set_to_chain:
                for chain_id in links:
                    if chain_id not in chains:
@@ -618,7 +589,6 @@ class EventFederationWorkerStore(SignatureWorkerStore, EventsWorkerStore, SQLBas

                    _materialize(chain_id, chains[chain_id], links, chains)

-                chains_to_fetch.difference_update(chains)
                seen_chains.update(chains)

        # Now for each chain we figure out the maximum sequence number reachable
@@ -106,7 +106,7 @@ from synapse.storage.database import (
 )
 from synapse.storage.databases.main.receipts import ReceiptsWorkerStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
-from synapse.types import JsonDict
+from synapse.types import JsonDict, StrCollection
 from synapse.util import json_encoder
 from synapse.util.caches.descriptors import cached

@@ -385,7 +385,6 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            WITH all_receipts AS (
                SELECT room_id, thread_id, MAX(event_stream_ordering) AS max_receipt_stream_ordering
                FROM receipts_linearized
-                LEFT JOIN events USING (room_id, event_id)
                WHERE
                    {receipt_types_clause}
                    AND user_id = ?
@@ -621,13 +620,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                SELECT notif_count, COALESCE(unread_count, 0), thread_id
                FROM event_push_summary
                LEFT JOIN (
-                    SELECT thread_id, MAX(stream_ordering) AS threaded_receipt_stream_ordering
+                    SELECT thread_id, MAX(event_stream_ordering) AS threaded_receipt_stream_ordering
                    FROM receipts_linearized
-                    LEFT JOIN events USING (room_id, event_id)
                    WHERE
                        user_id = ?
                        AND room_id = ?
-                        AND stream_ordering > ?
+                        AND event_stream_ordering > ?
                        AND {receipt_types_clause}
                    GROUP BY thread_id
                ) AS receipts USING (thread_id)
@@ -659,13 +657,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        sql = f"""
            SELECT COUNT(*), thread_id FROM event_push_actions
            LEFT JOIN (
-                SELECT thread_id, MAX(stream_ordering) AS threaded_receipt_stream_ordering
+                SELECT thread_id, MAX(event_stream_ordering) AS threaded_receipt_stream_ordering
                FROM receipts_linearized
-                LEFT JOIN events USING (room_id, event_id)
                WHERE
                    user_id = ?
                    AND room_id = ?
-                    AND stream_ordering > ?
+                    AND event_stream_ordering > ?
                    AND {receipt_types_clause}
                GROUP BY thread_id
            ) AS receipts USING (thread_id)
@@ -738,13 +735,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                thread_id
            FROM event_push_actions
            LEFT JOIN (
-                SELECT thread_id, MAX(stream_ordering) AS threaded_receipt_stream_ordering
+                SELECT thread_id, MAX(event_stream_ordering) AS threaded_receipt_stream_ordering
                FROM receipts_linearized
-                LEFT JOIN events USING (room_id, event_id)
                WHERE
                    user_id = ?
                    AND room_id = ?
-                    AND stream_ordering > ?
+                    AND event_stream_ordering > ?
                    AND {receipt_types_clause}
                GROUP BY thread_id
            ) AS receipts USING (thread_id)
@@ -859,37 +855,85 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas

        return await self.db_pool.runInteraction("get_push_action_users_in_range", f)

-    def _get_receipts_by_room_txn(
-        self, txn: LoggingTransaction, user_id: str
+    def _get_receipts_for_room_and_threads_txn(
+        self,
+        txn: LoggingTransaction,
+        user_id: str,
+        room_ids: StrCollection,
+        thread_ids: StrCollection,
    ) -> Dict[str, _RoomReceipt]:
        """
-        Generate a map of room ID to the latest stream ordering that has been
-        read by the given user.
+        Get (private) read receipts for a user in each of the given room IDs
+        and thread IDs.

-        Args:
-            txn:
-            user_id: The user to fetch receipts for.
+        Note: The corresponding room ID for each thread must appear in
+        `room_ids` arg.

        Returns:
            A map including all rooms the user is in with a receipt. It maps
            room IDs to _RoomReceipt instances
        """
-        receipt_types_clause, args = make_in_list_sql_clause(
+
+        receipt_types_clause, receipts_args = make_in_list_sql_clause(
            self.database_engine,
            "receipt_type",
            (ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE),
        )

+        thread_ids_clause, thread_ids_args = make_in_list_sql_clause(
+            self.database_engine,
+            "thread_id",
+            thread_ids,
+        )
+
+        room_ids_clause, room_ids_args = make_in_list_sql_clause(
+            self.database_engine,
+            "room_id",
+            room_ids,
+        )
+
+        # We use the union of two (almost identical) queries here, the first to
+        # fetch the specific thread receipts and the second to fetch the
+        # unthreaded receipts.
+        #
+        # This SQL is optimized to use the indices we have on
+        # `receipts_linearized`.
+        #
+        # We compare room ID and thread IDs independently due to the above,
+        # which means that this query might return more rows than we need if the
+        # same thread ID appears across different rooms (e.g. 'main' thread ID).
+        # This doesn't cause any logic issues, and isn't a performance concern
+        # given this function generally gets called with only one room and
+        # thread ID.
        sql = f"""
+            SELECT room_id, thread_id, MAX(event_stream_ordering)
+            FROM receipts_linearized
+            WHERE {receipt_types_clause}
+                AND {thread_ids_clause}
+                AND {room_ids_clause}
+            AND user_id = ?
+            GROUP BY room_id, thread_id
+
+        UNION ALL
+
            SELECT room_id, thread_id, MAX(stream_ordering)
            FROM receipts_linearized
            INNER JOIN events USING (room_id, event_id)
            WHERE {receipt_types_clause}
+                AND {room_ids_clause}
+                AND thread_id IS NULL
            AND user_id = ?
            GROUP BY room_id, thread_id
        """

-        args.extend((user_id,))
+        args = list(receipts_args)
+        args.extend(thread_ids_args)
+        args.extend(room_ids_args)
+        args.append(user_id)
+        args.extend(receipts_args)
+        args.extend(room_ids_args)
+        args.append(user_id)
+
        txn.execute(sql, args)

        result: Dict[str, _RoomReceipt] = {}
@@ -925,12 +969,6 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            The list will have between 0~limit entries.
        """

-        receipts_by_room = await self.db_pool.runInteraction(
-            "get_unread_push_actions_for_user_in_range_http_receipts",
-            self._get_receipts_by_room_txn,
-            user_id=user_id,
-        )
-
        def get_push_actions_txn(
            txn: LoggingTransaction,
        ) -> List[Tuple[str, str, str, int, str, bool]]:
@@ -952,6 +990,27 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            "get_unread_push_actions_for_user_in_range_http", get_push_actions_txn
        )

+        room_ids = set()
+        thread_ids = []
+        for (
+            _,
+            room_id,
+            thread_id,
+            _,
+            _,
+            _,
+        ) in push_actions:
+            room_ids.add(room_id)
+            thread_ids.append(thread_id)
+
+        receipts_by_room = await self.db_pool.runInteraction(
+            "get_unread_push_actions_for_user_in_range_http_receipts",
+            self._get_receipts_for_room_and_threads_txn,
+            user_id=user_id,
+            room_ids=room_ids,
+            thread_ids=thread_ids,
+        )
+
        notifs = [
            HttpPushAction(
                event_id=event_id,
@@ -998,12 +1057,6 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            The list will have between 0~limit entries.
        """

-        receipts_by_room = await self.db_pool.runInteraction(
-            "get_unread_push_actions_for_user_in_range_email_receipts",
-            self._get_receipts_by_room_txn,
-            user_id=user_id,
-        )
-
        def get_push_actions_txn(
            txn: LoggingTransaction,
        ) -> List[Tuple[str, str, str, int, str, bool, int]]:
@@ -1026,6 +1079,28 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            "get_unread_push_actions_for_user_in_range_email", get_push_actions_txn
        )

+        room_ids = set()
+        thread_ids = []
+        for (
+            _,
+            room_id,
+            thread_id,
+            _,
+            _,
+            _,
+            _,
+        ) in push_actions:
+            room_ids.add(room_id)
+            thread_ids.append(thread_id)
+
+        receipts_by_room = await self.db_pool.runInteraction(
+            "get_unread_push_actions_for_user_in_range_email_receipts",
+            self._get_receipts_for_room_and_threads_txn,
+            user_id=user_id,
+            room_ids=room_ids,
+            thread_ids=thread_ids,
+        )
+
        # Make a list of dicts from the two sets of results.
        notifs = [
            EmailPushAction(
@@ -1362,9 +1437,8 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        )

        sql = """
-            SELECT r.stream_id, r.room_id, r.user_id, r.thread_id, e.stream_ordering
+            SELECT r.stream_id, r.room_id, r.user_id, r.thread_id, r.event_stream_ordering
            FROM receipts_linearized AS r
-            INNER JOIN events AS e USING (event_id)
            WHERE ? < r.stream_id AND r.stream_id <= ? AND user_id LIKE ?
            ORDER BY r.stream_id ASC
            LIMIT ?
@@ -53,11 +53,7 @@ from synapse.storage.databases.main.receipts import ReceiptsWorkerStore
 from synapse.storage.databases.main.roommember import RoomMemberWorkerStore
 from synapse.storage.engines import PostgresEngine, Sqlite3Engine
 from synapse.storage.push_rule import InconsistentRuleException, RuleNotFoundException
-from synapse.storage.util.id_generators import (
-    AbstractStreamIdGenerator,
-    IdGenerator,
-    StreamIdGenerator,
-)
+from synapse.storage.util.id_generators import IdGenerator, StreamIdGenerator
 from synapse.synapse_rust.push import FilteredPushRules, PushRule, PushRules
 from synapse.types import JsonDict
 from synapse.util import json_encoder, unwrapFirstError
@@ -130,6 +126,8 @@ class PushRulesWorkerStore(
    `get_max_push_rules_stream_id` which can be called in the initializer.
    """

+    _push_rules_stream_id_gen: StreamIdGenerator
+
    def __init__(
        self,
        database: DatabasePool,
@@ -138,6 +136,10 @@ class PushRulesWorkerStore(
    ):
        super().__init__(database, db_conn, hs)

+        self._is_push_writer = (
+            hs.get_instance_name() in hs.config.worker.writers.push_rules
+        )
+
        # In the worker store this is an ID tracker which we overwrite in the non-worker
        # class below that is used on the main process.
        self._push_rules_stream_id_gen = StreamIdGenerator(
@@ -145,7 +147,7 @@ class PushRulesWorkerStore(
            hs.get_replication_notifier(),
            "push_rules_stream",
            "stream_id",
-            is_writer=hs.config.worker.worker_app is None,
+            is_writer=self._is_push_writer,
        )

        push_rules_prefill, push_rules_id = self.db_pool.get_cache_dict(
@@ -162,6 +164,9 @@ class PushRulesWorkerStore(
            prefilled_cache=push_rules_prefill,
        )

+        self._push_rule_id_gen = IdGenerator(db_conn, "push_rules", "id")
+        self._push_rules_enable_id_gen = IdGenerator(db_conn, "push_rules_enable", "id")
+
    def get_max_push_rules_stream_id(self) -> int:
        """Get the position of the push rules stream.

@@ -383,23 +388,6 @@ class PushRulesWorkerStore(
            "get_all_push_rule_updates", get_all_push_rule_updates_txn
        )

-
-class PushRuleStore(PushRulesWorkerStore):
-    # Because we have write access, this will be a StreamIdGenerator
-    # (see PushRulesWorkerStore.__init__)
-    _push_rules_stream_id_gen: AbstractStreamIdGenerator
-
-    def __init__(
-        self,
-        database: DatabasePool,
-        db_conn: LoggingDatabaseConnection,
-        hs: "HomeServer",
-    ):
-        super().__init__(database, db_conn, hs)
-
-        self._push_rule_id_gen = IdGenerator(db_conn, "push_rules", "id")
-        self._push_rules_enable_id_gen = IdGenerator(db_conn, "push_rules_enable", "id")
-
    async def add_push_rule(
        self,
        user_id: str,
@@ -410,6 +398,9 @@ class PushRuleStore(PushRulesWorkerStore):
        before: Optional[str] = None,
        after: Optional[str] = None,
    ) -> None:
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        conditions_json = json_encoder.encode(conditions)
        actions_json = json_encoder.encode(actions)
        async with self._push_rules_stream_id_gen.get_next() as stream_id:
@@ -455,6 +446,9 @@ class PushRuleStore(PushRulesWorkerStore):
        before: str,
        after: str,
    ) -> None:
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        relative_to_rule = before or after

        sql = """
@@ -524,6 +518,9 @@ class PushRuleStore(PushRulesWorkerStore):
        conditions_json: str,
        actions_json: str,
    ) -> None:
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        if isinstance(self.database_engine, PostgresEngine):
            # Postgres doesn't do FOR UPDATE on aggregate functions, so select the rows first
            # then re-select the count/max below.
@@ -575,6 +572,9 @@ class PushRuleStore(PushRulesWorkerStore):
        actions_json: str,
        update_stream: bool = True,
    ) -> None:
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        """Specialised version of simple_upsert_txn that picks a push_rule_id
        using the _push_rule_id_gen if it needs to insert the rule. It assumes
        that the "push_rules" table is locked"""
@@ -653,6 +653,8 @@ class PushRuleStore(PushRulesWorkerStore):
            user_id: The matrix ID of the push rule owner
            rule_id: The rule_id of the rule to be deleted
        """
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")

        def delete_push_rule_txn(
            txn: LoggingTransaction,
@@ -704,6 +706,9 @@ class PushRuleStore(PushRulesWorkerStore):
        Raises:
            RuleNotFoundException if the rule does not exist.
        """
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        async with self._push_rules_stream_id_gen.get_next() as stream_id:
            event_stream_ordering = self._stream_id_gen.get_current_token()
            await self.db_pool.runInteraction(
@@ -727,6 +732,9 @@ class PushRuleStore(PushRulesWorkerStore):
        enabled: bool,
        is_default_rule: bool,
    ) -> None:
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        new_id = self._push_rules_enable_id_gen.get_next()

        if not is_default_rule:
@@ -796,6 +804,9 @@ class PushRuleStore(PushRulesWorkerStore):
        Raises:
            RuleNotFoundException if the rule does not exist.
        """
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        actions_json = json_encoder.encode(actions)

        def set_push_rule_actions_txn(
@@ -865,6 +876,9 @@ class PushRuleStore(PushRulesWorkerStore):
        op: str,
        data: Optional[JsonDict] = None,
    ) -> None:
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        values = {
            "stream_id": stream_id,
            "event_stream_ordering": event_stream_ordering,
@@ -882,9 +896,6 @@ class PushRuleStore(PushRulesWorkerStore):
            self.push_rules_stream_cache.entity_has_changed, user_id, stream_id
        )

-    def get_max_push_rules_stream_id(self) -> int:
-        return self._push_rules_stream_id_gen.get_current_token()
-
    async def copy_push_rule_from_room_to_room(
        self, new_room_id: str, user_id: str, rule: PushRule
    ) -> None:
@@ -895,6 +906,9 @@ class PushRuleStore(PushRulesWorkerStore):
            user_id : ID of user the push rule belongs to.
            rule: A push rule.
        """
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        # Create new rule id
        rule_id_scope = "/".join(rule.rule_id.split("/")[:-1])
        new_rule_id = rule_id_scope + "/" + new_room_id
@@ -930,6 +944,9 @@ class PushRuleStore(PushRulesWorkerStore):
            new_room_id: ID of the new room.
            user_id: ID of user to copy push rules for.
        """
+        if not self._is_push_writer:
+            raise Exception("Not a push writer")
+
        # Retrieve push rules for this user
        user_push_rules = await self.get_push_rules_for_user(user_id)

@@ -178,14 +178,13 @@ class ReceiptsWorkerStore(SQLBaseStore):
        )

        sql = f"""
-            SELECT event_id, stream_ordering
+            SELECT event_id, event_stream_ordering
            FROM receipts_linearized
-            INNER JOIN events USING (room_id, event_id)
            WHERE {clause}
            AND user_id = ?
            AND room_id = ?
            AND thread_id IS NULL
-            ORDER BY stream_ordering DESC
+            ORDER BY event_stream_ordering DESC
            LIMIT 1
        """

@@ -735,10 +734,13 @@ class ReceiptsWorkerStore(SQLBaseStore):
                thread_clause = "r.thread_id = ?"
                thread_args = (thread_id,)

+            # If the receipt doesn't have a stream ordering it is because we
+            # don't have the associated event, and so must be a remote receipt.
+            # Hence it's safe to just allow new receipts to clobber it.
            sql = f"""
-            SELECT stream_ordering, event_id FROM events
-            INNER JOIN receipts_linearized AS r USING (event_id, room_id)
-            WHERE r.room_id = ? AND r.receipt_type = ? AND r.user_id = ? AND {thread_clause}
+            SELECT r.event_stream_ordering, r.event_id FROM receipts_linearized AS r
+            WHERE r.room_id = ? AND r.receipt_type = ? AND r.user_id = ?
+            AND r.event_stream_ordering IS NOT NULL AND {thread_clause}
            """
            txn.execute(
                sql,
@@ -2108,6 +2108,13 @@ class RegistrationBackgroundUpdateStore(RegistrationWorkerStore):
            unique=False,
        )

+        self.db_pool.updates.register_background_index_update(
+            update_name="access_tokens_refresh_token_id_idx",
+            index_name="access_tokens_refresh_token_id_idx",
+            table="access_tokens",
+            columns=("refresh_token_id",),
+        )
+
    async def _background_update_set_deactivated_flag(
        self, progress: JsonDict, batch_size: int
    ) -> int:
@@ -0,0 +1,15 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2023 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+    (8404, 'access_tokens_refresh_token_id_idx', '{}');
@@ -1101,6 +1101,56 @@ class E2eKeysHandlerTestCase(unittest.HomeserverTestCase):
            },
        )

+    def test_has_different_keys(self) -> None:
+        """check that has_different_keys returns True when the keys provided are different to what
+        is in the database."""
+        local_user = "@boris:" + self.hs.hostname
+        keys1 = {
+            "master_key": {
+                # private key: 2lonYOM6xYKdEsO+6KrC766xBcHnYnim1x/4LFGF8B0
+                "user_id": local_user,
+                "usage": ["master"],
+                "keys": {
+                    "ed25519:nqOvzeuGWT/sRx3h7+MHoInYj3Uk2LD/unI9kDYcHwk": "nqOvzeuGWT/sRx3h7+MHoInYj3Uk2LD/unI9kDYcHwk"
+                },
+            }
+        }
+        self.get_success(self.handler.upload_signing_keys_for_user(local_user, keys1))
+        is_different = self.get_success(
+            self.handler.has_different_keys(
+                local_user,
+                {
+                    "master_key": keys1["master_key"],
+                },
+            )
+        )
+        self.assertEqual(is_different, False)
+        # change the usage => different keys
+        keys1["master_key"]["usage"] = ["develop"]
+        is_different = self.get_success(
+            self.handler.has_different_keys(
+                local_user,
+                {
+                    "master_key": keys1["master_key"],
+                },
+            )
+        )
+        self.assertEqual(is_different, True)
+        keys1["master_key"]["usage"] = ["master"]  # reset
+        # change the key => different keys
+        keys1["master_key"]["keys"] = {
+            "ed25519:nqOvzeuGWT/sRx3h7+MHoInYj3Uk2LD/unIc0rncs": "nqOvzeuGWT/sRx3h7+MHoInYj3Uk2LD/unIc0rncs"
+        }
+        is_different = self.get_success(
+            self.handler.has_different_keys(
+                local_user,
+                {
+                    "master_key": keys1["master_key"],
+                },
+            )
+        )
+        self.assertEqual(is_different, True)
+
    def test_query_devices_remote_sync(self) -> None:
        """Tests that querying keys for a remote user that we share a room with,
        but haven't yet fetched the keys for, returns the cross signing keys
@@ -17,14 +17,16 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-from typing import Collection, List, Optional
+from typing import Collection, ContextManager, List, Optional
 from unittest.mock import AsyncMock, Mock, patch

+from parameterized import parameterized
+
 from twisted.test.proto_helpers import MemoryReactor

 from synapse.api.constants import EventTypes, JoinRules
 from synapse.api.errors import Codes, ResourceLimitError
-from synapse.api.filtering import Filtering
+from synapse.api.filtering import FilterCollection, Filtering
 from synapse.api.room_versions import RoomVersion, RoomVersions
 from synapse.events import EventBase
 from synapse.events.snapshot import EventContext
@@ -33,7 +35,7 @@ from synapse.handlers.sync import SyncConfig, SyncResult
 from synapse.rest import admin
 from synapse.rest.client import knock, login, room
 from synapse.server import HomeServer
-from synapse.types import UserID, create_requester
+from synapse.types import JsonDict, UserID, create_requester
 from synapse.util import Clock

 import tests.unittest
@@ -258,13 +260,7 @@ class SyncTestCase(tests.unittest.HomeserverTestCase):
        # Eve tries to join the room. We monkey patch the internal logic which selects
        # the prev_events used when creating the join event, such that the ban does not
        # precede the join.
-        mocked_get_prev_events = patch.object(
-            self.hs.get_datastores().main,
-            "get_prev_events_for_room",
-            new_callable=AsyncMock,
-            return_value=[last_room_creation_event_id],
-        )
-        with mocked_get_prev_events:
+        with self._patch_get_latest_events([last_room_creation_event_id]):
            self.helper.join(room_id, eve, tok=eve_token)

        # Eve makes a second, incremental sync.
@@ -288,6 +284,460 @@ class SyncTestCase(tests.unittest.HomeserverTestCase):
        )
        self.assertEqual(eve_initial_sync_after_join.joined, [])

+    def test_state_includes_changes_on_forks(self) -> None:
+        """State changes that happen on a fork of the DAG must be included in `state`
+
+        Given the following DAG:
+
+             E1
+           ↗    ↖
+          |      S2
+          |      ↑
+        --|------|----
+          |      |
+          E3     |
+           ↖    /
+             E4
+
+        ... and a filter that means we only return 2 events, represented by the dashed
+        horizontal line: `S2` must be included in the `state` section.
+        """
+        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 as Alice 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)
+            )
+        )
+        last_room_creation_event_id = (
+            initial_sync_result.joined[0].timeline.events[-1].event_id
+        )
+
+        # Send a state event, and a regular event, both using the same prev ID
+        with self._patch_get_latest_events([last_room_creation_event_id]):
+            s2_event = self.helper.send_state(room_id, "s2", {}, tok=alice_tok)[
+                "event_id"
+            ]
+            e3_event = self.helper.send(room_id, "e3", tok=alice_tok)["event_id"]
+
+        # Send a final event, joining the two branches of the dag
+        e4_event = self.helper.send(room_id, "e4", tok=alice_tok)["event_id"]
+
+        # do an incremental sync, with a filter that will ensure we only get two of
+        # the three new events.
+        incremental_sync = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                alice_requester,
+                generate_sync_config(
+                    alice,
+                    filter_collection=FilterCollection(
+                        self.hs, {"room": {"timeline": {"limit": 2}}}
+                    ),
+                ),
+                since_token=initial_sync_result.next_batch,
+            )
+        )
+
+        # The state event should appear in the 'state' section of the response.
+        room_sync = incremental_sync.joined[0]
+        self.assertEqual(room_sync.room_id, room_id)
+        self.assertTrue(room_sync.timeline.limited)
+        self.assertEqual(
+            [e.event_id for e in room_sync.timeline.events],
+            [e3_event, e4_event],
+        )
+        self.assertEqual(
+            [e.event_id for e in room_sync.state.values()],
+            [s2_event],
+        )
+
+    def test_state_includes_changes_on_forks_when_events_excluded(self) -> None:
+        """A variation on the previous test, but where one event is filtered
+
+        The DAG is the same as the previous test, but E4 is excluded by the filter.
+
+             E1
+           ↗    ↖
+          |      S2
+          |      ↑
+        --|------|----
+          |      |
+          E3     |
+           ↖    /
+            (E4)
+
+        """
+
+        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 as Alice 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)
+            )
+        )
+        last_room_creation_event_id = (
+            initial_sync_result.joined[0].timeline.events[-1].event_id
+        )
+
+        # Send a state event, and a regular event, both using the same prev ID
+        with self._patch_get_latest_events([last_room_creation_event_id]):
+            s2_event = self.helper.send_state(room_id, "s2", {}, tok=alice_tok)[
+                "event_id"
+            ]
+            e3_event = self.helper.send(room_id, "e3", tok=alice_tok)["event_id"]
+
+        # Send a final event, joining the two branches of the dag
+        self.helper.send(room_id, "e4", type="not_a_normal_message", tok=alice_tok)[
+            "event_id"
+        ]
+
+        # do an incremental sync, with a filter that will only return E3, excluding S2
+        # and E4.
+        incremental_sync = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                alice_requester,
+                generate_sync_config(
+                    alice,
+                    filter_collection=FilterCollection(
+                        self.hs,
+                        {
+                            "room": {
+                                "timeline": {
+                                    "limit": 1,
+                                    "not_types": ["not_a_normal_message"],
+                                }
+                            }
+                        },
+                    ),
+                ),
+                since_token=initial_sync_result.next_batch,
+            )
+        )
+
+        # The state event should appear in the 'state' section of the response.
+        room_sync = incremental_sync.joined[0]
+        self.assertEqual(room_sync.room_id, room_id)
+        self.assertTrue(room_sync.timeline.limited)
+        self.assertEqual(
+            [e.event_id for e in room_sync.timeline.events],
+            [e3_event],
+        )
+        self.assertEqual(
+            [e.event_id for e in room_sync.state.values()],
+            [s2_event],
+        )
+
+    def test_state_includes_changes_on_long_lived_forks(self) -> None:
+        """State changes that happen on a fork of the DAG must be included in `state`
+
+        Given the following DAG:
+
+             E1
+           ↗    ↖
+          |      S2
+          |      ↑
+        --|------|----
+          E3     |
+        --|------|----
+          |      E4
+          |      |
+
+        ... and a filter that means we only return 1 event, represented by the dashed
+        horizontal lines: `S2` must be included in the `state` section on the second sync.
+        """
+        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 as Alice 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)
+            )
+        )
+        last_room_creation_event_id = (
+            initial_sync_result.joined[0].timeline.events[-1].event_id
+        )
+
+        # Send a state event, and a regular event, both using the same prev ID
+        with self._patch_get_latest_events([last_room_creation_event_id]):
+            s2_event = self.helper.send_state(room_id, "s2", {}, tok=alice_tok)[
+                "event_id"
+            ]
+            e3_event = self.helper.send(room_id, "e3", tok=alice_tok)["event_id"]
+
+        # Do an incremental sync, this will return E3 but *not* S2 at this
+        # point.
+        incremental_sync = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                alice_requester,
+                generate_sync_config(
+                    alice,
+                    filter_collection=FilterCollection(
+                        self.hs, {"room": {"timeline": {"limit": 1}}}
+                    ),
+                ),
+                since_token=initial_sync_result.next_batch,
+            )
+        )
+        room_sync = incremental_sync.joined[0]
+        self.assertEqual(room_sync.room_id, room_id)
+        self.assertTrue(room_sync.timeline.limited)
+        self.assertEqual(
+            [e.event_id for e in room_sync.timeline.events],
+            [e3_event],
+        )
+        self.assertEqual(
+            [e.event_id for e in room_sync.state.values()],
+            [],
+        )
+
+        # Now send another event that points to S2, but not E3.
+        with self._patch_get_latest_events([s2_event]):
+            e4_event = self.helper.send(room_id, "e4", tok=alice_tok)["event_id"]
+
+        # Doing an incremental sync should return S2 in state.
+        incremental_sync = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                alice_requester,
+                generate_sync_config(
+                    alice,
+                    filter_collection=FilterCollection(
+                        self.hs, {"room": {"timeline": {"limit": 1}}}
+                    ),
+                ),
+                since_token=incremental_sync.next_batch,
+            )
+        )
+        room_sync = incremental_sync.joined[0]
+        self.assertEqual(room_sync.room_id, room_id)
+        self.assertFalse(room_sync.timeline.limited)
+        self.assertEqual(
+            [e.event_id for e in room_sync.timeline.events],
+            [e4_event],
+        )
+        self.assertEqual(
+            [e.event_id for e in room_sync.state.values()],
+            [s2_event],
+        )
+
+    def test_state_includes_changes_on_ungappy_syncs(self) -> None:
+        """Test `state` where the sync is not gappy.
+
+        We start with a DAG like this:
+
+             E1
+           ↗    ↖
+          |      S2
+          |
+        --|---
+          |
+          E3
+
+        ... and initialsync with `limit=1`, represented by the horizontal dashed line.
+        At this point, we do not expect S2 to appear in the response at all (since
+        it is excluded from the timeline by the `limit`, and the state is based on the
+        state after the most recent event before the sync token (E3), which doesn't
+        include S2.
+
+        Now more events arrive, and we do an incremental sync:
+
+             E1
+           ↗    ↖
+          |      S2
+          |      ↑
+          E3     |
+          ↑      |
+        --|------|----
+          |      |
+          E4     |
+           ↖    /
+             E5
+
+        This is the last chance for us to tell the client about S2, so it *must* be
+        included in the response.
+        """
+        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)
+            )
+        )
+        last_room_creation_event_id = (
+            initial_sync_result.joined[0].timeline.events[-1].event_id
+        )
+
+        # Send a state event, and a regular event, both using the same prev ID
+        with self._patch_get_latest_events([last_room_creation_event_id]):
+            s2_event = self.helper.send_state(room_id, "s2", {}, tok=alice_tok)[
+                "event_id"
+            ]
+            e3_event = self.helper.send(room_id, "e3", tok=alice_tok)["event_id"]
+
+        # Another initial sync, with limit=1
+        initial_sync_result = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                alice_requester,
+                generate_sync_config(
+                    alice,
+                    filter_collection=FilterCollection(
+                        self.hs, {"room": {"timeline": {"limit": 1}}}
+                    ),
+                ),
+            )
+        )
+        room_sync = initial_sync_result.joined[0]
+        self.assertEqual(room_sync.room_id, room_id)
+        self.assertEqual(
+            [e.event_id for e in room_sync.timeline.events],
+            [e3_event],
+        )
+        self.assertNotIn(s2_event, [e.event_id for e in room_sync.state.values()])
+
+        # More events, E4 and E5
+        with self._patch_get_latest_events([e3_event]):
+            e4_event = self.helper.send(room_id, "e4", tok=alice_tok)["event_id"]
+        e5_event = self.helper.send(room_id, "e5", tok=alice_tok)["event_id"]
+
+        # Now incremental sync
+        incremental_sync = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                alice_requester,
+                generate_sync_config(alice),
+                since_token=initial_sync_result.next_batch,
+            )
+        )
+
+        # The state event should appear in the 'state' section of the response.
+        room_sync = incremental_sync.joined[0]
+        self.assertEqual(room_sync.room_id, room_id)
+        self.assertFalse(room_sync.timeline.limited)
+        self.assertEqual(
+            [e.event_id for e in room_sync.timeline.events],
+            [e4_event, e5_event],
+        )
+        self.assertEqual(
+            [e.event_id for e in room_sync.state.values()],
+            [s2_event],
+        )
+
+    @parameterized.expand(
+        [
+            (False, False),
+            (True, False),
+            (False, True),
+            (True, True),
+        ]
+    )
+    def test_archived_rooms_do_not_include_state_after_leave(
+        self, initial_sync: bool, empty_timeline: bool
+    ) -> None:
+        """If the user leaves the room, state changes that happen after they leave are not returned.
+
+        We try with both a zero and a normal timeline limit,
+        and we try both an initial sync and an incremental sync for both.
+        """
+        if empty_timeline and not initial_sync:
+            # FIXME synapse doesn't return the room at all in this situation!
+            self.skipTest("Synapse does not correctly handle this case")
+
+        # Alice creates the room, and bob joins.
+        alice = self.register_user("alice", "password")
+        alice_tok = self.login(alice, "password")
+
+        bob = self.register_user("bob", "password")
+        bob_tok = self.login(bob, "password")
+        bob_requester = create_requester(bob)
+
+        room_id = self.helper.create_room_as(alice, is_public=True, tok=alice_tok)
+        self.helper.join(room_id, bob, tok=bob_tok)
+
+        initial_sync_result = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                bob_requester, generate_sync_config(bob)
+            )
+        )
+
+        # Alice sends a message and a state
+        before_message_event = self.helper.send(room_id, "before", tok=alice_tok)[
+            "event_id"
+        ]
+        before_state_event = self.helper.send_state(
+            room_id, "test_state", {"body": "before"}, tok=alice_tok
+        )["event_id"]
+
+        # Bob leaves
+        leave_event = self.helper.leave(room_id, bob, tok=bob_tok)["event_id"]
+
+        # Alice sends some more stuff
+        self.helper.send(room_id, "after", tok=alice_tok)["event_id"]
+        self.helper.send_state(room_id, "test_state", {"body": "after"}, tok=alice_tok)[
+            "event_id"
+        ]
+
+        # And now, Bob resyncs.
+        filter_dict: JsonDict = {"room": {"include_leave": True}}
+        if empty_timeline:
+            filter_dict["room"]["timeline"] = {"limit": 0}
+        sync_room_result = self.get_success(
+            self.sync_handler.wait_for_sync_for_user(
+                bob_requester,
+                generate_sync_config(
+                    bob, filter_collection=FilterCollection(self.hs, filter_dict)
+                ),
+                since_token=None if initial_sync else initial_sync_result.next_batch,
+            )
+        ).archived[0]
+
+        if empty_timeline:
+            # The timeline should be empty
+            self.assertEqual(sync_room_result.timeline.events, [])
+
+            # And the state should include the leave event...
+            self.assertEqual(
+                sync_room_result.state[("m.room.member", bob)].event_id, leave_event
+            )
+            # ... and the state change before he left.
+            self.assertEqual(
+                sync_room_result.state[("test_state", "")].event_id, before_state_event
+            )
+        else:
+            # The last three events in the timeline should be those leading up to the
+            # leave
+            self.assertEqual(
+                [e.event_id for e in sync_room_result.timeline.events[-3:]],
+                [before_message_event, before_state_event, leave_event],
+            )
+            # ... And the state should be empty
+            self.assertEqual(sync_room_result.state, {})
+
+    def _patch_get_latest_events(self, latest_events: List[str]) -> ContextManager:
+        """Monkey-patch `get_prev_events_for_room`
+
+        Returns a context manager which will replace the implementation of
+        `get_prev_events_for_room` with one which returns `latest_events`.
+        """
+        return patch.object(
+            self.hs.get_datastores().main,
+            "get_prev_events_for_room",
+            new_callable=AsyncMock,
+            return_value=latest_events,
+        )
+
    def test_call_invite_in_public_room_not_returned(self) -> None:
        user = self.register_user("alice", "password")
        tok = self.login(user, "password")
@@ -401,14 +851,26 @@ _request_key = 0


 def generate_sync_config(
-    user_id: str, device_id: Optional[str] = "device_id"
+    user_id: str,
+    device_id: Optional[str] = "device_id",
+    filter_collection: Optional[FilterCollection] = None,
 ) -> SyncConfig:
-    """Generate a sync config (with a unique request key)."""
+    """Generate a sync config (with a unique request key).
+
+    Args:
+        user_id: user who is syncing.
+        device_id: device that is syncing. Defaults to "device_id".
+        filter_collection: filter to apply. Defaults to the default filter (ie,
+           return everything, with a default limit)
+    """
+    if filter_collection is None:
+        filter_collection = Filtering(Mock()).DEFAULT_FILTER_COLLECTION
+
    global _request_key
    _request_key += 1
    return SyncConfig(
        user=UserID.from_string(user_id),
-        filter_collection=Filtering(Mock()).DEFAULT_FILTER_COLLECTION,
+        filter_collection=filter_collection,
        is_guest=False,
        request_key=("request_key", _request_key),
        device_id=device_id,
@@ -277,7 +277,8 @@ class DeleteMediaByDateSizeTestCase(_AdminMediaTests):
        self.assertEqual(400, channel.code, msg=channel.json_body)
        self.assertEqual(Codes.MISSING_PARAM, channel.json_body["errcode"])
        self.assertEqual(
-            "Missing integer query parameter 'before_ts'", channel.json_body["error"]
+            "Missing required integer query parameter before_ts",
+            channel.json_body["error"],
        )

    def test_invalid_parameter(self) -> None:
@@ -320,7 +321,7 @@ class DeleteMediaByDateSizeTestCase(_AdminMediaTests):
        self.assertEqual(400, channel.code, msg=channel.json_body)
        self.assertEqual(Codes.INVALID_PARAM, channel.json_body["errcode"])
        self.assertEqual(
-            "Query parameter size_gt must be a string representing a positive integer.",
+            "Query parameter size_gt must be a positive integer.",
            channel.json_body["error"],
        )

@@ -21,6 +21,7 @@
 import json
 import time
 import urllib.parse
+from http import HTTPStatus
 from typing import List, Optional
 from unittest.mock import AsyncMock, Mock

@@ -2190,6 +2191,33 @@ class RoomMessagesTestCase(unittest.HomeserverTestCase):
        chunk = channel.json_body["chunk"]
        self.assertEqual(len(chunk), 0, [event["content"] for event in chunk])

+    def test_room_message_filter_query_validation(self) -> None:
+        # Test json validation in (filter) query parameter.
+        # Does not test the validity of the filter, only the json validation.
+
+        # Check Get with valid json filter parameter, expect 200.
+        valid_filter_str = '{"types": ["m.room.message"]}'
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/admin/v1/rooms/{self.room_id}/messages?dir=b&filter={valid_filter_str}",
+            access_token=self.admin_user_tok,
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+        # Check Get with invalid json filter parameter, expect 400 NOT_JSON.
+        invalid_filter_str = "}}}{}"
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/admin/v1/rooms/{self.room_id}/messages?dir=b&filter={invalid_filter_str}",
+            access_token=self.admin_user_tok,
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.BAD_REQUEST, channel.json_body)
+        self.assertEqual(
+            channel.json_body["errcode"], Codes.NOT_JSON, channel.json_body
+        )
+

 class JoinAliasRoomTestCase(unittest.HomeserverTestCase):
    servlets = [
@@ -2522,6 +2550,39 @@ class JoinAliasRoomTestCase(unittest.HomeserverTestCase):
            else:
                self.fail("Event %s from events_after not found" % j)

+    def test_room_event_context_filter_query_validation(self) -> None:
+        # Test json validation in (filter) query parameter.
+        # Does not test the validity of the filter, only the json validation.
+
+        # Create a user with room and event_id.
+        user_id = self.register_user("test", "test")
+        user_tok = self.login("test", "test")
+        room_id = self.helper.create_room_as(user_id, tok=user_tok)
+        event_id = self.helper.send(room_id, "message 1", tok=user_tok)["event_id"]
+
+        # Check Get with valid json filter parameter, expect 200.
+        valid_filter_str = '{"types": ["m.room.message"]}'
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/admin/v1/rooms/{room_id}/context/{event_id}?filter={valid_filter_str}",
+            access_token=self.admin_user_tok,
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+        # Check Get with invalid json filter parameter, expect 400 NOT_JSON.
+        invalid_filter_str = "}}}{}"
+        channel = self.make_request(
+            "GET",
+            f"/_synapse/admin/v1/rooms/{room_id}/context/{event_id}?filter={invalid_filter_str}",
+            access_token=self.admin_user_tok,
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.BAD_REQUEST, channel.json_body)
+        self.assertEqual(
+            channel.json_body["errcode"], Codes.NOT_JSON, channel.json_body
+        )
+

 class MakeRoomAdminTestCase(unittest.HomeserverTestCase):
    servlets = [
@@ -35,7 +35,6 @@ from synapse.util import Clock
 from tests import unittest
 from tests.server import FakeChannel
 from tests.test_utils.event_injection import inject_event
-from tests.unittest import override_config


 class BaseRelationsTestCase(unittest.HomeserverTestCase):
@@ -957,7 +956,6 @@ class RelationPaginationTestCase(BaseRelationsTestCase):


 class RecursiveRelationTestCase(BaseRelationsTestCase):
-    @override_config({"experimental_features": {"msc3981_recurse_relations": True}})
    def test_recursive_relations(self) -> None:
        """Generate a complex, multi-level relationship tree and query it."""
        # Create a thread with a few messages in it.
@@ -1003,7 +1001,7 @@ class RecursiveRelationTestCase(BaseRelationsTestCase):
        channel = self.make_request(
            "GET",
            f"/_matrix/client/v1/rooms/{self.room}/relations/{self.parent_id}"
-            "?dir=f&limit=20&org.matrix.msc3981.recurse=true",
+            "?dir=f&limit=20&recurse=true",
            access_token=self.user_token,
        )
        self.assertEqual(200, channel.code, channel.json_body)
@@ -1024,7 +1022,6 @@ class RecursiveRelationTestCase(BaseRelationsTestCase):
            ],
        )

-    @override_config({"experimental_features": {"msc3981_recurse_relations": True}})
    def test_recursive_relations_with_filter(self) -> None:
        """The event_type and rel_type still apply."""
        # Create a thread with a few messages in it.
@@ -1052,7 +1049,7 @@ class RecursiveRelationTestCase(BaseRelationsTestCase):
        channel = self.make_request(
            "GET",
            f"/_matrix/client/v1/rooms/{self.room}/relations/{self.parent_id}/{RelationTypes.ANNOTATION}"
-            "?dir=f&limit=20&org.matrix.msc3981.recurse=true",
+            "?dir=f&limit=20&recurse=true",
            access_token=self.user_token,
        )
        self.assertEqual(200, channel.code, channel.json_body)
@@ -1065,7 +1062,7 @@ class RecursiveRelationTestCase(BaseRelationsTestCase):
        channel = self.make_request(
            "GET",
            f"/_matrix/client/v1/rooms/{self.room}/relations/{self.parent_id}/{RelationTypes.ANNOTATION}/m.reaction"
-            "?dir=f&limit=20&org.matrix.msc3981.recurse=true",
+            "?dir=f&limit=20&recurse=true",
            access_token=self.user_token,
        )
        self.assertEqual(200, channel.code, channel.json_body)
@@ -27,8 +27,10 @@ from synapse.util import Clock

 from tests import unittest
 from tests.unittest import override_config
+from tests.utils import HAS_AUTHLIB

-endpoint = "/_matrix/client/unstable/org.matrix.msc3886/rendezvous"
+msc3886_endpoint = "/_matrix/client/unstable/org.matrix.msc3886/rendezvous"
+msc4108_endpoint = "/_matrix/client/unstable/org.matrix.msc4108/rendezvous"


 class RendezvousServletTestCase(unittest.HomeserverTestCase):
@@ -41,11 +43,35 @@ class RendezvousServletTestCase(unittest.HomeserverTestCase):
        return self.hs

    def test_disabled(self) -> None:
-        channel = self.make_request("POST", endpoint, {}, access_token=None)
+        channel = self.make_request("POST", msc3886_endpoint, {}, access_token=None)
+        self.assertEqual(channel.code, 404)
+        channel = self.make_request("POST", msc4108_endpoint, {}, access_token=None)
        self.assertEqual(channel.code, 404)

    @override_config({"experimental_features": {"msc3886_endpoint": "/asd"}})
-    def test_redirect(self) -> None:
-        channel = self.make_request("POST", endpoint, {}, access_token=None)
+    def test_msc3886_redirect(self) -> None:
+        channel = self.make_request("POST", msc3886_endpoint, {}, access_token=None)
        self.assertEqual(channel.code, 307)
        self.assertEqual(channel.headers.getRawHeaders("Location"), ["/asd"])
+
+    @unittest.skip_unless(HAS_AUTHLIB, "requires authlib")
+    @override_config(
+        {
+            "disable_registration": True,
+            "experimental_features": {
+                "msc4108_delegation_endpoint": "https://asd",
+                "msc3861": {
+                    "enabled": True,
+                    "issuer": "https://issuer",
+                    "client_id": "client_id",
+                    "client_auth_method": "client_secret_post",
+                    "client_secret": "client_secret",
+                    "admin_token": "admin_token_value",
+                },
+            },
+        }
+    )
+    def test_msc4108_delegation(self) -> None:
+        channel = self.make_request("POST", msc4108_endpoint, {}, access_token=None)
+        self.assertEqual(channel.code, 307)
+        self.assertEqual(channel.headers.getRawHeaders("Location"), ["https://asd"])
@@ -2175,6 +2175,31 @@ class RoomMessageListTestCase(RoomBase):
        chunk = channel.json_body["chunk"]
        self.assertEqual(len(chunk), 0, [event["content"] for event in chunk])

+    def test_room_message_filter_query_validation(self) -> None:
+        # Test json validation in (filter) query parameter.
+        # Does not test the validity of the filter, only the json validation.
+
+        # Check Get with valid json filter parameter, expect 200.
+        valid_filter_str = '{"types": ["m.room.message"]}'
+        channel = self.make_request(
+            "GET",
+            f"/rooms/{self.room_id}/messages?access_token=x&dir=b&filter={valid_filter_str}",
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+        # Check Get with invalid json filter parameter, expect 400 NOT_JSON.
+        invalid_filter_str = "}}}{}"
+        channel = self.make_request(
+            "GET",
+            f"/rooms/{self.room_id}/messages?access_token=x&dir=b&filter={invalid_filter_str}",
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.BAD_REQUEST, channel.json_body)
+        self.assertEqual(
+            channel.json_body["errcode"], Codes.NOT_JSON, channel.json_body
+        )
+

 class RoomMessageFilterTestCase(RoomBase):
    """Tests /rooms/$room_id/messages REST events."""
@@ -3213,6 +3238,33 @@ class ContextTestCase(unittest.HomeserverTestCase):
        self.assertDictEqual(events_after[0].get("content"), {}, events_after[0])
        self.assertEqual(events_after[1].get("content"), {}, events_after[1])

+    def test_room_event_context_filter_query_validation(self) -> None:
+        # Test json validation in (filter) query parameter.
+        # Does not test the validity of the filter, only the json validation.
+        event_id = self.helper.send(self.room_id, "message 7", tok=self.tok)["event_id"]
+
+        # Check Get with valid json filter parameter, expect 200.
+        valid_filter_str = '{"types": ["m.room.message"]}'
+        channel = self.make_request(
+            "GET",
+            f"/rooms/{self.room_id}/context/{event_id}?filter={valid_filter_str}",
+            access_token=self.tok,
+        )
+        self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body)
+
+        # Check Get with invalid json filter parameter, expect 400 NOT_JSON.
+        invalid_filter_str = "}}}{}"
+        channel = self.make_request(
+            "GET",
+            f"/rooms/{self.room_id}/context/{event_id}?filter={invalid_filter_str}",
+            access_token=self.tok,
+        )
+
+        self.assertEqual(channel.code, HTTPStatus.BAD_REQUEST, channel.json_body)
+        self.assertEqual(
+            channel.json_body["errcode"], Codes.NOT_JSON, channel.json_body
+        )
+

 class RoomAliasListTestCase(unittest.HomeserverTestCase):
    servlets = [
@@ -170,8 +170,8 @@ class RestHelper:
        targ: Optional[str] = None,
        expect_code: int = HTTPStatus.OK,
        tok: Optional[str] = None,
-    ) -> None:
-        self.change_membership(
+    ) -> JsonDict:
+        return self.change_membership(
            room=room,
            src=src,
            targ=targ,
@@ -189,8 +189,8 @@ class RestHelper:
        appservice_user_id: Optional[str] = None,
        expect_errcode: Optional[Codes] = None,
        expect_additional_fields: Optional[dict] = None,
-    ) -> None:
-        self.change_membership(
+    ) -> JsonDict:
+        return self.change_membership(
            room=room,
            src=user,
            targ=user,
@@ -242,8 +242,8 @@ class RestHelper:
        user: Optional[str] = None,
        expect_code: int = HTTPStatus.OK,
        tok: Optional[str] = None,
-    ) -> None:
-        self.change_membership(
+    ) -> JsonDict:
+        return self.change_membership(
            room=room,
            src=user,
            targ=user,
@@ -282,7 +282,7 @@ class RestHelper:
        expect_code: int = HTTPStatus.OK,
        expect_errcode: Optional[str] = None,
        expect_additional_fields: Optional[dict] = None,
-    ) -> None:
+    ) -> JsonDict:
        """
        Send a membership state event into a room.

@@ -298,6 +298,9 @@ class RestHelper:
                using an application service access token in `tok`.
            expect_code: The expected HTTP response code
            expect_errcode: The expected Matrix error code
+
+        Returns:
+            The JSON response
        """
        temp_id = self.auth_user_id
        self.auth_user_id = src
@@ -356,6 +359,7 @@ class RestHelper:
                )

        self.auth_user_id = temp_id
+        return channel.json_body

    def send(
        self,
				`@@ -0,0 +1 @@`
				Adds validation to ensure that the `limit` parameter on `/publicRooms` is non-negative.
				`@@ -0,0 +1 @@`
				Return `400 M_NOT_JSON` upon receiving invalid JSON in query parameters across various client and admin endpoints, rather than an internal server error.
				`@@ -0,0 +1 @@`
				Make the CSAPI endpoint `/keys/device_signing/upload` idempotent.
				`@@ -1 +0,0 @@`
				`OIDC: try to JWT decode userinfo response if JSON parsing failed.`
				`@@ -0,0 +1 @@`
				`Use new receipts column to optimise receipt and push action SQL queries. Contributed by Nick @ Beeper (@fizzadar).`
				`@@ -0,0 +1 @@`
				`Add a prompt in the contributing guide to manually configure icu4c.`
				`@@ -0,0 +1 @@`
				`Bump minimum supported Rust version to 1.66.0.`
				`@@ -0,0 +1 @@`
				`Add helpers to transform Twisted requests to Rust http Requests/Responses.`
				`@@ -0,0 +1 @@`
				`Support delegating the rendezvous mechanism described MSC4108 to an external implementation.`
				`@@ -0,0 +1 @@`
				`Clarify what part of message retention is still experimental.`