Merge remote-tracking branch 'origin/develop' into matrix-org-hotfixes

2023-03-29 13:10:57 +01:00
parent 53352a2b80 78cdb72cd6
commit 34a2dff973
150 changed files with 2973 additions and 1075 deletions
@@ -35,9 +35,9 @@ sed -i \
 # compatible (as far the package metadata declares, anyway); pip's package resolver
 # is more lax.
 #
-# Rather than `poetry install --no-dev`, we drop all dev dependencies from the
-# toml file. This means we don't have to ensure compatibility between old deps and
-# dev tools.
+# Rather than `poetry install --no-dev`, we drop all dev dependencies and the dev-docs
+# group from the toml file. This means we don't have to ensure compatibility between
+# old deps and dev tools.

 pip install toml wheel

@@ -47,6 +47,7 @@ with open('pyproject.toml', 'r') as f:
    data = toml.loads(f.read())

 del data['tool']['poetry']['dev-dependencies']
+del data['tool']['poetry']['group']['dev-docs']

 with open('pyproject.toml', 'w') as f:
    toml.dump(data, f)
@@ -13,25 +13,10 @@ on:
  workflow_dispatch:

 jobs:
-  pages:
-    name: GitHub Pages
+  pre:
+    name: Calculate variables for GitHub Pages deployment
    runs-on: ubuntu-latest
    steps:
-      - uses: actions/checkout@v3
-
-      - name: Setup mdbook
-        uses: peaceiris/actions-mdbook@adeb05db28a0c0004681db83893d56c0388ea9ea # v1.2.0
-        with:
-          mdbook-version: '0.4.17'
-
-      - name: Build the documentation
-        # mdbook will only create an index.html if we're including docs/README.md in SUMMARY.md.
-        # However, we're using docs/README.md for other purposes and need to pick a new page
-        # as the default. Let's opt for the welcome page instead.
-        run: |
-          mdbook build
-          cp book/welcome_and_overview.html book/index.html
-
      # Figure out the target directory.
      #
      # The target directory depends on the name of the branch
@@ -55,11 +40,65 @@ jobs:

          # finally, set the 'branch-version' var.
          echo "branch-version=$branch" >> "$GITHUB_OUTPUT"
-          
+    outputs:
+      branch-version: ${{ steps.vars.outputs.branch-version }}
+
+################################################################################
+  pages-docs:
+    name: GitHub Pages
+    runs-on: ubuntu-latest
+    needs:
+      - pre
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Setup mdbook
+        uses: peaceiris/actions-mdbook@adeb05db28a0c0004681db83893d56c0388ea9ea # v1.2.0
+        with:
+          mdbook-version: '0.4.17'
+
+      - name: Build the documentation
+        # mdbook will only create an index.html if we're including docs/README.md in SUMMARY.md.
+        # However, we're using docs/README.md for other purposes and need to pick a new page
+        # as the default. Let's opt for the welcome page instead.
+        run: |
+          mdbook build
+          cp book/welcome_and_overview.html book/index.html
+
      # Deploy to the target directory.
      - name: Deploy to gh pages
        uses: peaceiris/actions-gh-pages@bd8c6b06eba6b3d25d72b7a1767993c0aeee42e7 # v3.9.2
        with:
          github_token: ${{ secrets.GITHUB_TOKEN }}
          publish_dir: ./book
-          destination_dir: ./${{ steps.vars.outputs.branch-version }}
+          destination_dir: ./${{ needs.pre.outputs.branch-version }}
+
+################################################################################
+  pages-devdocs:
+    name: GitHub Pages (developer docs)
+    runs-on: ubuntu-latest
+    needs:
+      - pre
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: "Set up Sphinx"
+        uses: matrix-org/setup-python-poetry@v1
+        with:
+          python-version: "3.x"
+          poetry-version: "1.3.2"
+          groups: "dev-docs"
+          extras: ""
+
+      - name: Build the documentation
+        run: |
+          cd dev-docs
+          poetry run make html
+
+      # Deploy to the target directory.
+      - name: Deploy to gh pages
+        uses: peaceiris/actions-gh-pages@bd8c6b06eba6b3d25d72b7a1767993c0aeee42e7 # v3.9.2
+        with:
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          publish_dir: ./dev-docs/_build/html
+          destination_dir: ./dev-docs/${{ needs.pre.outputs.branch-version }}
@@ -27,7 +27,7 @@ jobs:
    steps:
      - uses: actions/checkout@v3
      - name: Install Rust
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          toolchain: stable
      - uses: Swatinem/rust-cache@v2
@@ -61,7 +61,7 @@ jobs:
      - uses: actions/checkout@v3

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          toolchain: stable
      - uses: Swatinem/rust-cache@v2
@@ -134,7 +134,7 @@ jobs:
      - uses: actions/checkout@v3

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          toolchain: stable
      - uses: Swatinem/rust-cache@v2
@@ -34,6 +34,14 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
+      - name: Install Rust
+        # There don't seem to be versioned releases of this action per se: for each rust
+        # version there is a branch which gets constantly rebased on top of master.
+        # We pin to a specific commit for paranoia's sake.
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
+        with:
+            toolchain: 1.58.1
+      - uses: Swatinem/rust-cache@v2
      - uses: matrix-org/setup-python-poetry@v1
        with:
          python-version: "3.x"
@@ -95,6 +103,14 @@ jobs:
      - uses: actions/checkout@v3
        with:
          ref: ${{ github.event.pull_request.head.sha }}
+      - name: Install Rust
+        # There don't seem to be versioned releases of this action per se: for each rust
+        # version there is a branch which gets constantly rebased on top of master.
+        # We pin to a specific commit for paranoia's sake.
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
+        with:
+            toolchain: 1.58.1
+      - uses: Swatinem/rust-cache@v2
      - uses: matrix-org/setup-python-poetry@v1
        with:
          poetry-version: "1.3.2"
@@ -113,7 +129,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: 1.58.1
            components: clippy
@@ -135,7 +151,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: nightly-2022-12-01
            components: clippy
@@ -155,7 +171,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          # We use nightly so that it correctly groups together imports
          toolchain: nightly-2022-12-01
@@ -223,7 +239,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: 1.58.1
      - uses: Swatinem/rust-cache@v2
@@ -268,7 +284,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: 1.58.1
      - uses: Swatinem/rust-cache@v2
@@ -389,7 +405,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: 1.58.1
      - uses: Swatinem/rust-cache@v2
@@ -534,7 +550,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: 1.58.1
      - uses: Swatinem/rust-cache@v2
@@ -565,7 +581,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: 1.58.1
      - uses: Swatinem/rust-cache@v2
@@ -588,7 +604,7 @@ jobs:
        # There don't seem to be versioned releases of this action per se: for each rust
        # version there is a branch which gets constantly rebased on top of master.
        # We pin to a specific commit for paranoia's sake.
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
            toolchain: nightly-2022-12-01
      - uses: Swatinem/rust-cache@v2
@@ -5,6 +5,13 @@ on:
    - cron: 0 8 * * *

  workflow_dispatch:
+    inputs:
+      twisted_ref:
+        description: Commit, branch or tag to checkout from upstream Twisted.
+        required: false
+        default: 'trunk'
+        type: string
+

 concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
@@ -18,7 +25,7 @@ jobs:
      - uses: actions/checkout@v3

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          toolchain: stable
      - uses: Swatinem/rust-cache@v2
@@ -29,7 +36,7 @@ jobs:
          extras: "all"
      - run: |
          poetry remove twisted
-          poetry add --extras tls git+https://github.com/twisted/twisted.git#trunk
+          poetry add --extras tls git+https://github.com/twisted/twisted.git#${{ inputs.twisted_ref }}
          poetry install --no-interaction --extras "all test"
      - name: Remove warn_unused_ignores from mypy config
        run: sed '/warn_unused_ignores = True/d' -i mypy.ini
@@ -43,7 +50,7 @@ jobs:
      - run: sudo apt-get -qq install xmlsec1

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          toolchain: stable
      - uses: Swatinem/rust-cache@v2
@@ -82,7 +89,7 @@ jobs:
      - uses: actions/checkout@v3

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@e12eda571dc9a5ee5d58eecf4738ec291c66f295
+        uses: dtolnay/rust-toolchain@fc3253060d0c959bea12a59f10f8391454a0b02d
        with:
          toolchain: stable
      - uses: Swatinem/rust-cache@v2
@@ -53,6 +53,7 @@ __pycache__/
 /coverage.*
 /dist/
 /docs/build/
+/dev-docs/_build/
 /htmlcov
 /pip-wheel-metadata/

@@ -61,7 +62,7 @@ book/

 # complement
 /complement-*
-/master.tar.gz
+/main.tar.gz

 # rust
 /target/
@@ -1,3 +1,20 @@
+Synapse 1.80.0 (2023-03-28)
+===========================
+
+No significant changes since 1.80.0rc2.
+
+
+Synapse 1.80.0rc2 (2023-03-22)
+==============================
+
+Bugfixes
+--------
+
+- Fix a bug in which the [`POST /_matrix/client/v3/rooms/{roomId}/report/{eventId}`](https://spec.matrix.org/v1.6/client-server-api/#post_matrixclientv3roomsroomidreporteventid) endpoint would return the wrong error if the user did not have permission to view the event. This aligns Synapse's implementation with [MSC2249](https://github.com/matrix-org/matrix-spec-proposals/pull/2249). ([\#15298](https://github.com/matrix-org/synapse/issues/15298), [\#15300](https://github.com/matrix-org/synapse/issues/15300))
+- Fix a bug introduced in Synapse 1.75.0rc1 where the [SQLite port_db script](https://matrix-org.github.io/synapse/latest/postgres.html#porting-from-sqlite)
+  would fail to open the SQLite database. ([\#15301](https://github.com/matrix-org/synapse/issues/15301))
+
+
 Synapse 1.80.0rc1 (2023-03-21)
 ==============================

@@ -294,9 +294,9 @@ dependencies = [

 [[package]]
 name = "regex"
-version = "1.7.1"
+version = "1.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "48aaa5748ba571fb95cd2c85c09f629215d3a6ece942baa100950af03a34f733"
+checksum = "8b1f693b24f6ac912f4893ef08244d70b6067480d2f1a46e950c9691e6749d1d"
 dependencies = [
 "aho-corasick",
 "memchr",
@@ -305,9 +305,9 @@ dependencies = [

 [[package]]
 name = "regex-syntax"
-version = "0.6.27"
+version = "0.6.29"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3f87b73ce11b1619a3c6332f45341e0047173771e8b8b73f87bfeefb7b56244"
+checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1"

 [[package]]
 name = "ryu"
@@ -323,22 +323,22 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"

 [[package]]
 name = "serde"
-version = "1.0.157"
+version = "1.0.158"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "707de5fcf5df2b5788fca98dd7eab490bc2fd9b7ef1404defc462833b83f25ca"
+checksum = "771d4d9c4163ee138805e12c710dd365e4f44be8be0503cb1bb9eb989425d9c9"
 dependencies = [
 "serde_derive",
 ]

 [[package]]
 name = "serde_derive"
-version = "1.0.157"
+version = "1.0.158"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78997f4555c22a7971214540c4a661291970619afd56de19f77e0de86296e1e5"
+checksum = "e801c1712f48475582b7696ac71e0ca34ebb30e09338425384269d9717c62cad"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 2.0.2",
+ "syn 2.0.10",
 ]

 [[package]]
@@ -377,9 +377,9 @@ dependencies = [

 [[package]]
 name = "syn"
-version = "2.0.2"
+version = "2.0.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "59d3276aee1fa0c33612917969b5172b5be2db051232a6e4826f1a1a9191b045"
+checksum = "5aad1363ed6d37b84299588d62d3a7d95b5a5c2d9aad5c85609fda12afaa1f40"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -0,0 +1 @@
+Use `immutabledict` instead of `frozendict`.
@@ -0,0 +1 @@
+Add denormalised event stream ordering column to membership state tables for future use. Contributed by Nick @ Beeper (@fizzadar).
@@ -0,0 +1 @@
+Prune user's old devices on login if they have too many.
@@ -0,0 +1 @@
+Add a primitive helper script for listing worker endpoints.
@@ -0,0 +1 @@
+Add developer documentation for the Federation Sender and add a documentation mechanism using Sphinx.
@@ -0,0 +1 @@
+Make the pushers rely on the `device_id` instead of the `access_token_id` for various operations.
@@ -0,0 +1 @@
+Bump sentry-sdk from 1.15.0 to 1.17.0.
@@ -0,0 +1 @@
+Fix a long-standing bug where edits of non-`m.room.message` events would not be correctly bundled.
@@ -0,0 +1 @@
+Fix a bug introduced in Synapse v1.55.0 which could delay remote homeservers being able to decrypt encrypted messages sent by local users.
@@ -1 +0,0 @@
-Fix a bug in which the [`POST /_matrix/client/v3/rooms/{roomId}/report/{eventId}`](https://spec.matrix.org/v1.6/client-server-api/#post_matrixclientv3roomsroomidreporteventid) endpoint would return the wrong error if the user did not have permission to view the event. This aligns Synapse's implementation with [MSC2249](https://github.com/matrix-org/matrix-spec-proposals/pull/2249).
@@ -1 +0,0 @@
-Fix a bug in which the [`POST /_matrix/client/v3/rooms/{roomId}/report/{eventId}`](https://spec.matrix.org/v1.6/client-server-api/#post_matrixclientv3roomsroomidreporteventid) endpoint would return the wrong error if the user did not have permission to view the event. This aligns Synapse's implementation with [MSC2249](https://github.com/matrix-org/matrix-spec-proposals/pull/2249).
@@ -1,3 +0,0 @@
-Fix a bug introduced in Synapse 1.75.0rc1 where the [SQLite port_db script](https://matrix-org.github.io/synapse/latest/postgres.html#porting-from-sqlite)
-would fail to open the SQLite database.
-
@@ -0,0 +1 @@
+Allow running the Twisted trunk job against other branches.
@@ -0,0 +1 @@
+Remind the releaser to ask for changelog feedback in [#synapse-dev](https://matrix.to/#/#synapse-dev:matrix.org).
@@ -0,0 +1 @@
+Bump dtolnay/rust-toolchain from e12eda571dc9a5ee5d58eecf4738ec291c66f295 to fc3253060d0c959bea12a59f10f8391454a0b02d.
@@ -0,0 +1,2 @@
+Add a check to [SQLite port_db script](https://matrix-org.github.io/synapse/latest/postgres.html#porting-from-sqlite)
+to ensure that the sqlite database passed to the script exists before trying to port from it.
@@ -0,0 +1 @@
+Fix a bug introduced in Synapse 1.76.0 where responses from worker deployments could include an internal `_INT_STREAM_POS` key.
@@ -0,0 +1 @@
+Reject events with an invalid "mentions" property pert [MSC3952](https://github.com/matrix-org/matrix-spec-proposals/pull/3952).
@@ -0,0 +1 @@
+Experimental support for passing One Time Key requests to application services ([MSC3983](https://github.com/matrix-org/matrix-spec-proposals/pull/3983)).
@@ -0,0 +1 @@
+As an optimisation, use `TRUNCATE` on Postgres when clearing the user directory tables.
@@ -0,0 +1 @@
+Fix `.gitignore` rule for the Complement source tarball downloaded automatically by `complement.sh`.
@@ -0,0 +1 @@
+Fix a long-standing bug preventing users from joining rooms, that they had been unbanned from, over federation. Contributed by Nico.
@@ -0,0 +1 @@
+Bump serde from 1.0.157 to 1.0.158.
@@ -0,0 +1 @@
+Bump regex from 1.7.1 to 1.7.3.
@@ -0,0 +1 @@
+Bump types-pyopenssl from 23.0.0.4 to 23.1.0.0.
@@ -0,0 +1 @@
+Bump furo from 2022.12.7 to 2023.3.23.
@@ -0,0 +1 @@
+Bump ruff from 0.0.252 to 0.0.259.
@@ -0,0 +1 @@
+Bump cryptography from 40.0.0 to 40.0.1.
@@ -0,0 +1 @@
+Bump mypy-zope from 0.9.0 to 0.9.1.
@@ -0,0 +1 @@
+Allow loading `/password_policy` endpoint on workers.
@@ -0,0 +1 @@
+Fix bug in worker mode where on a rolling restart of workers the "typing" worker would consume 100% CPU until it got restarted.
@@ -0,0 +1 @@
+Add developer documentation for the Federation Sender and add a documentation mechanism using Sphinx.
@@ -0,0 +1 @@
+Speed up pydantic CI job.
@@ -0,0 +1 @@
+Speed up sample config CI job.
@@ -0,0 +1 @@
+Fix a typo in login requests ratelimit defaults.
@@ -1,3 +1,15 @@
+matrix-synapse-py3 (1.80.0) stable; urgency=medium
+
+  * New Synapse release 1.80.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 28 Mar 2023 11:10:33 +0100
+
+matrix-synapse-py3 (1.80.0~rc2) stable; urgency=medium
+
+  * New Synapse release 1.80.0rc2.
+
+ -- Synapse Packaging team <packages@matrix.org>  Wed, 22 Mar 2023 08:30:16 -0700
+
 matrix-synapse-py3 (1.80.0~rc1) stable; urgency=medium

  * New Synapse release 1.80.0rc1.
@@ -0,0 +1,20 @@
+# Minimal makefile for Sphinx documentation
+#
+
+# You can set these variables from the command line, and also
+# from the environment for the first two.
+SPHINXOPTS    ?=
+SPHINXBUILD   ?= sphinx-build
+SOURCEDIR     = .
+BUILDDIR      = _build
+
+# Put it first so that "make" without argument is like "make help".
+help:
+	@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
+
+.PHONY: help Makefile
+
+# Catch-all target: route all unknown targets to Sphinx using the new
+# "make mode" option.  $(O) is meant as a shortcut for $(SPHINXOPTS).
+%: Makefile
+	@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
@@ -0,0 +1,50 @@
+# Configuration file for the Sphinx documentation builder.
+#
+# For the full list of built-in configuration values, see the documentation:
+# https://www.sphinx-doc.org/en/master/usage/configuration.html
+
+# -- Project information -----------------------------------------------------
+# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
+
+project = "Synapse development"
+copyright = "2023, The Matrix.org Foundation C.I.C."
+author = "The Synapse Maintainers and Community"
+
+# -- General configuration ---------------------------------------------------
+# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
+
+extensions = [
+    "autodoc2",
+    "myst_parser",
+]
+
+templates_path = ["_templates"]
+exclude_patterns = ["_build", "Thumbs.db", ".DS_Store"]
+
+
+# -- Options for Autodoc2 ----------------------------------------------------
+
+autodoc2_docstring_parser_regexes = [
+    # this will render all docstrings as 'MyST' Markdown
+    (r".*", "myst"),
+]
+
+autodoc2_packages = [
+    {
+        "path": "../synapse",
+        # Don't render documentation for everything as a matter of course
+        "auto_mode": False,
+    },
+]
+
+
+# -- Options for MyST (Markdown) ---------------------------------------------
+
+# myst_heading_anchors = 2
+
+
+# -- Options for HTML output -------------------------------------------------
+# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
+
+html_theme = "furo"
+html_static_path = ["_static"]
@@ -0,0 +1,22 @@
+.. Synapse Developer Documentation documentation master file, created by
+   sphinx-quickstart on Mon Mar 13 08:59:51 2023.
+   You can adapt this file completely to your liking, but it should at least
+   contain the root `toctree` directive.
+
+Welcome to the Synapse Developer Documentation!
+===========================================================
+
+.. toctree::
+   :maxdepth: 2
+   :caption: Contents:
+
+   modules/federation_sender
+
+
+
+Indices and tables
+==================
+
+* :ref:`genindex`
+* :ref:`modindex`
+* :ref:`search`
@@ -0,0 +1,5 @@
+Federation Sender
+=================
+
+```{autodoc2-docstring} synapse.federation.sender
+```
@@ -172,6 +172,7 @@ WORKERS_CONFIG: Dict[str, Dict[str, Any]] = {
            "^/_matrix/client/v1/rooms/.*/timestamp_to_event$",
            "^/_matrix/client/(api/v1|r0|v3|unstable)/search",
            "^/_matrix/client/(r0|v3|unstable)/user/.*/filter(/|$)",
+            "^/_matrix/client/(r0|v3|unstable)/password_policy$",
        ],
        "shared_extra_conf": {},
        "worker_extra_conf": "",
@@ -1521,7 +1521,7 @@ This option specifies several limits for login:
      address. Defaults to `per_second: 0.003`, `burst_count: 5`.

 * `account` ratelimits login requests based on the account the
-  client is attempting to log into. Defaults to `per_second: 0.03`,
+  client is attempting to log into. Defaults to `per_second: 0.003`,
  `burst_count: 5`.

 * `failed_attempts` ratelimits login requests based on the account the
@@ -247,6 +247,7 @@ information.
    ^/_matrix/client/(r0|v3|unstable)/register$
    ^/_matrix/client/(r0|v3|unstable)/register/available$
    ^/_matrix/client/v1/register/m.login.registration_token/validity$
+    ^/_matrix/client/(r0|v3|unstable)/password_policy$

    # Event sending requests
    ^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/redact
@@ -89,7 +89,7 @@ manifest-path = "rust/Cargo.toml"

 [tool.poetry]
 name = "matrix-synapse"
-version = "1.80.0rc1"
+version = "1.80.0"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "Apache-2.0"
@@ -153,15 +153,13 @@ python = "^3.7.1"
 # ----------------------
 # we use the TYPE_CHECKER.redefine method added in jsonschema 3.0.0
 jsonschema = ">=3.0.0"
-# frozendict 2.1.2 is broken on Debian 10: https://github.com/Marco-Sulla/python-frozendict/issues/41
-# We cannot test our wheels against the 2.3.5 release in CI. Putting in an upper bound for this
-# because frozendict has been more trouble than it's worth; we would like to move to immutabledict.
-frozendict = ">=1,!=2.1.2,<2.3.5"
+# We choose 2.0 as a lower bound: the most recent backwards incompatible release.
+# It seems generally available, judging by https://pkgs.org/search/?q=immutabledict
+immutabledict = ">=2.0"
 # We require 2.1.0 or higher for type hints. Previous guard was >= 1.1.0
 unpaddedbase64 = ">=2.1.0"
-# We require 1.5.0 to work around an issue when running against the C implementation of
-# frozendict: https://github.com/matrix-org/python-canonicaljson/issues/36
-canonicaljson = "^1.5.0"
+# We require 2.0.0 for immutabledict support.
+canonicaljson = "^2.0.0"
 # we use the type definitions added in signedjson 1.1.
 signedjson = "^1.1.0"
 # validating SSL certs for IP addresses requires service_identity 18.1.
@@ -313,7 +311,7 @@ all = [
 # We pin black so that our tests don't start failing on new releases.
 isort = ">=5.10.1"
 black = ">=22.3.0"
-ruff = "0.0.252"
+ruff = "0.0.259"

 # Typechecking
 mypy = "*"
@@ -352,6 +350,18 @@ towncrier = ">=18.6.0rc1"
 # Used for checking the Poetry lockfile
 tomli = ">=1.2.3"

+
+# Dependencies for building the development documentation
+[tool.poetry.group.dev-docs]
+optional = true
+
+[tool.poetry.group.dev-docs.dependencies]
+sphinx = {version = "^6.1", python = "^3.8"}
+sphinx-autodoc2 = {version = "^0.4.2", python = "^3.8"}
+myst-parser = {version = "^1.0.0", python = "^3.8"}
+furo = ">=2022.12.7,<2024.0.0"
+
+
 [build-system]
 # The upper bounds here are defensive, intended to prevent situations like
 # #13849 and #14079 where we see buildtime or runtime errors caused by build
@@ -91,6 +91,7 @@ else
          "synapse" "docker" "tests"
          "scripts-dev"
          "contrib" "synmark" "stubs" ".ci"
+          "dev-docs"
      )
  fi
 fi
@@ -280,7 +280,7 @@ def _prepare() -> None:
    )

    print("Opening the changelog in your browser...")
-    print("Please ask others to give it a check.")
+    print("Please ask #synapse-dev to give it a check.")
    click.launch(
        f"https://github.com/matrix-org/synapse/blob/{synapse_repo.active_branch.name}/CHANGES.md"
    )
@@ -1,39 +0,0 @@
-# Copyright 2020 The Matrix.org Foundation C.I.C.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-# Stub for frozendict.
-
-from __future__ import annotations
-
-from typing import Any, Hashable, Iterable, Iterator, Mapping, Tuple, TypeVar, overload
-
-_KT = TypeVar("_KT", bound=Hashable)  # Key type.
-_VT = TypeVar("_VT")  # Value type.
-
-class frozendict(Mapping[_KT, _VT]):
-    @overload
-    def __init__(self, **kwargs: _VT) -> None: ...
-    @overload
-    def __init__(self, __map: Mapping[_KT, _VT], **kwargs: _VT) -> None: ...
-    @overload
-    def __init__(
-        self, __iterable: Iterable[Tuple[_KT, _VT]], **kwargs: _VT
-    ) -> None: ...
-    def __getitem__(self, key: _KT) -> _VT: ...
-    def __contains__(self, key: Any) -> bool: ...
-    def copy(self, **add_or_replace: Any) -> frozendict: ...
-    def __iter__(self) -> Iterator[_KT]: ...
-    def __len__(self) -> int: ...
-    def __repr__(self) -> str: ...
-    def __hash__(self) -> int: ...
@@ -17,9 +17,9 @@
 """ This is an implementation of a Matrix homeserver.
 """

-import json
 import os
 import sys
+from typing import Any, Dict

 from synapse.util.rust import check_rust_lib_up_to_date
 from synapse.util.stringutils import strtobool
@@ -61,11 +61,20 @@ try:
 except ImportError:
    pass

-# Use the standard library json implementation instead of simplejson.
+# Teach canonicaljson how to serialise immutabledicts.
 try:
-    from canonicaljson import set_json_library
+    from canonicaljson import register_preserialisation_callback
+    from immutabledict import immutabledict

-    set_json_library(json)
+    def _immutabledict_cb(d: immutabledict) -> Dict[str, Any]:
+        try:
+            return d._dict
+        except Exception:
+            # Paranoia: fall back to a `dict()` call, in case a future version of
+            # immutabledict removes `_dict` from the implementation.
+            return dict(d)
+
+    register_preserialisation_callback(immutabledict, _immutabledict_cb)
 except ImportError:
    pass

@@ -0,0 +1,302 @@
+#!/usr/bin/env python
+# Copyright 2022-2023 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+import logging
+import re
+from collections import defaultdict
+from dataclasses import dataclass
+from typing import Dict, Iterable, Optional, Pattern, Set, Tuple
+
+import yaml
+
+from synapse.config.homeserver import HomeServerConfig
+from synapse.federation.transport.server import (
+    TransportLayerServer,
+    register_servlets as register_federation_servlets,
+)
+from synapse.http.server import HttpServer, ServletCallback
+from synapse.rest import ClientRestResource
+from synapse.rest.key.v2 import RemoteKey
+from synapse.server import HomeServer
+from synapse.storage import DataStore
+
+logger = logging.getLogger("generate_workers_map")
+
+
+class MockHomeserver(HomeServer):
+    DATASTORE_CLASS = DataStore  # type: ignore
+
+    def __init__(self, config: HomeServerConfig, worker_app: Optional[str]) -> None:
+        super().__init__(config.server.server_name, config=config)
+        self.config.worker.worker_app = worker_app
+
+
+GROUP_PATTERN = re.compile(r"\(\?P<[^>]+?>(.+?)\)")
+
+
+@dataclass
+class EndpointDescription:
+    """
+    Describes an endpoint and how it should be routed.
+    """
+
+    # The servlet class that handles this endpoint
+    servlet_class: object
+
+    # The category of this endpoint. Is read from the `CATEGORY` constant in the servlet
+    # class.
+    category: Optional[str]
+
+    # TODO:
+    #  - does it need to be routed based on a stream writer config?
+    #  - does it benefit from any optimised, but optional, routing?
+    #  - what 'opinionated synapse worker class' (event_creator, synchrotron, etc) does
+    #    it go in?
+
+
+class EnumerationResource(HttpServer):
+    """
+    Accepts servlet registrations for the purposes of building up a description of
+    all endpoints.
+    """
+
+    def __init__(self, is_worker: bool) -> None:
+        self.registrations: Dict[Tuple[str, str], EndpointDescription] = {}
+        self._is_worker = is_worker
+
+    def register_paths(
+        self,
+        method: str,
+        path_patterns: Iterable[Pattern],
+        callback: ServletCallback,
+        servlet_classname: str,
+    ) -> None:
+        # federation servlet callbacks are wrapped, so unwrap them.
+        callback = getattr(callback, "__wrapped__", callback)
+
+        # fish out the servlet class
+        servlet_class = callback.__self__.__class__  # type: ignore
+
+        if self._is_worker and method in getattr(
+            servlet_class, "WORKERS_DENIED_METHODS", ()
+        ):
+            # This endpoint would cause an error if called on a worker, so pretend it
+            # was never registered!
+            return
+
+        sd = EndpointDescription(
+            servlet_class=servlet_class,
+            category=getattr(servlet_class, "CATEGORY", None),
+        )
+
+        for pat in path_patterns:
+            self.registrations[(method, pat.pattern)] = sd
+
+
+def get_registered_paths_for_hs(
+    hs: HomeServer,
+) -> Dict[Tuple[str, str], EndpointDescription]:
+    """
+    Given a homeserver, get all registered endpoints and their descriptions.
+    """
+
+    enumerator = EnumerationResource(is_worker=hs.config.worker.worker_app is not None)
+    ClientRestResource.register_servlets(enumerator, hs)
+    federation_server = TransportLayerServer(hs)
+
+    # we can't use `federation_server.register_servlets` but this line does the
+    # same thing, only it uses this enumerator
+    register_federation_servlets(
+        federation_server.hs,
+        resource=enumerator,
+        ratelimiter=federation_server.ratelimiter,
+        authenticator=federation_server.authenticator,
+        servlet_groups=federation_server.servlet_groups,
+    )
+
+    # the key server endpoints are separate again
+    RemoteKey(hs).register(enumerator)
+
+    return enumerator.registrations
+
+
+def get_registered_paths_for_default(
+    worker_app: Optional[str], base_config: HomeServerConfig
+) -> Dict[Tuple[str, str], EndpointDescription]:
+    """
+    Given the name of a worker application and a base homeserver configuration,
+    returns:
+
+        Dict from (method, path) to EndpointDescription
+
+    TODO Don't require passing in a config
+    """
+
+    hs = MockHomeserver(base_config, worker_app)
+    # TODO We only do this to avoid an error, but don't need the database etc
+    hs.setup()
+    return get_registered_paths_for_hs(hs)
+
+
+def elide_http_methods_if_unconflicting(
+    registrations: Dict[Tuple[str, str], EndpointDescription],
+    all_possible_registrations: Dict[Tuple[str, str], EndpointDescription],
+) -> Dict[Tuple[str, str], EndpointDescription]:
+    """
+    Elides HTTP methods (by replacing them with `*`) if all possible registered methods
+    can be handled by the worker whose registration map is `registrations`.
+
+    i.e. the only endpoints left with methods (other than `*`) should be the ones where
+    the worker can't handle all possible methods for that path.
+    """
+
+    def paths_to_methods_dict(
+        methods_and_paths: Iterable[Tuple[str, str]]
+    ) -> Dict[str, Set[str]]:
+        """
+        Given (method, path) pairs, produces a dict from path to set of methods
+        available at that path.
+        """
+        result: Dict[str, Set[str]] = {}
+        for method, path in methods_and_paths:
+            result.setdefault(path, set()).add(method)
+        return result
+
+    all_possible_reg_methods = paths_to_methods_dict(all_possible_registrations)
+    reg_methods = paths_to_methods_dict(registrations)
+
+    output = {}
+
+    for path, handleable_methods in reg_methods.items():
+        if handleable_methods == all_possible_reg_methods[path]:
+            any_method = next(iter(handleable_methods))
+            # TODO This assumes that all methods have the same servlet.
+            #      I suppose that's possibly dubious?
+            output[("*", path)] = registrations[(any_method, path)]
+        else:
+            for method in handleable_methods:
+                output[(method, path)] = registrations[(method, path)]
+
+    return output
+
+
+def simplify_path_regexes(
+    registrations: Dict[Tuple[str, str], EndpointDescription]
+) -> Dict[Tuple[str, str], EndpointDescription]:
+    """
+    Simplify all the path regexes for the dict of endpoint descriptions,
+    so that we don't use the Python-specific regex extensions
+    (and also to remove needlessly specific detail).
+    """
+
+    def simplify_path_regex(path: str) -> str:
+        """
+        Given a regex pattern, replaces all named capturing groups (e.g. `(?P<blah>xyz)`)
+        with a simpler version available in more common regex dialects (e.g. `.*`).
+        """
+
+        # TODO it's hard to choose between these two;
+        #      `.*` is a vague simplification
+        # return GROUP_PATTERN.sub(r"\1", path)
+        return GROUP_PATTERN.sub(r".*", path)
+
+    return {(m, simplify_path_regex(p)): v for (m, p), v in registrations.items()}
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Updates a synapse database to the latest schema and optionally runs background updates"
+            " on it."
+        )
+    )
+    parser.add_argument("-v", action="store_true")
+    parser.add_argument(
+        "--config-path",
+        type=argparse.FileType("r"),
+        required=True,
+        help="Synapse configuration file",
+    )
+
+    args = parser.parse_args()
+
+    # TODO
+    # logging.basicConfig(**logging_config)
+
+    # Load, process and sanity-check the config.
+    hs_config = yaml.safe_load(args.config_path)
+
+    config = HomeServerConfig()
+    config.parse_config_dict(hs_config, "", "")
+
+    master_paths = get_registered_paths_for_default(None, config)
+    worker_paths = get_registered_paths_for_default(
+        "synapse.app.generic_worker", config
+    )
+
+    all_paths = {**master_paths, **worker_paths}
+
+    elided_worker_paths = elide_http_methods_if_unconflicting(worker_paths, all_paths)
+    elide_http_methods_if_unconflicting(master_paths, all_paths)
+
+    # TODO SSO endpoints (pick_idp etc) NOT REGISTERED BY THIS SCRIPT
+
+    categories_to_methods_and_paths: Dict[
+        Optional[str], Dict[Tuple[str, str], EndpointDescription]
+    ] = defaultdict(dict)
+
+    for (method, path), desc in elided_worker_paths.items():
+        categories_to_methods_and_paths[desc.category][method, path] = desc
+
+    for category, contents in categories_to_methods_and_paths.items():
+        print_category(category, contents)
+
+
+def print_category(
+    category_name: Optional[str],
+    elided_worker_paths: Dict[Tuple[str, str], EndpointDescription],
+) -> None:
+    """
+    Prints out a category, in documentation page style.
+
+    Example:
+    ```
+    # Category name
+    /path/xyz
+
+    GET /path/abc
+    ```
+    """
+
+    if category_name:
+        print(f"# {category_name}")
+    else:
+        print("# (Uncategorised requests)")
+
+    for ln in sorted(
+        p for m, p in simplify_path_regexes(elided_worker_paths) if m == "*"
+    ):
+        print(ln)
+    print()
+    for ln in sorted(
+        f"{m:6} {p}" for m, p in simplify_path_regexes(elided_worker_paths) if m != "*"
+    ):
+        print(ln)
+    print()
+
+
+if __name__ == "__main__":
+    main()
@@ -18,6 +18,7 @@
 import argparse
 import curses
 import logging
+import os
 import sys
 import time
 import traceback
@@ -67,7 +68,10 @@ from synapse.storage.databases.main.media_repository import (
    MediaRepositoryBackgroundUpdateStore,
 )
 from synapse.storage.databases.main.presence import PresenceBackgroundUpdateStore
-from synapse.storage.databases.main.pusher import PusherWorkerStore
+from synapse.storage.databases.main.pusher import (
+    PusherBackgroundUpdatesStore,
+    PusherWorkerStore,
+)
 from synapse.storage.databases.main.receipts import ReceiptsBackgroundUpdateStore
 from synapse.storage.databases.main.registration import (
    RegistrationBackgroundUpdateStore,
@@ -225,6 +229,7 @@ class Store(
    AccountDataWorkerStore,
    PushRuleStore,
    PusherWorkerStore,
+    PusherBackgroundUpdatesStore,
    PresenceBackgroundUpdateStore,
    ReceiptsBackgroundUpdateStore,
    RelationsWorkerStore,
@@ -1326,6 +1331,13 @@ def main() -> None:
        filename="port-synapse.log" if args.curses else None,
    )

+    if not os.path.isfile(args.sqlite_database):
+        sys.stderr.write(
+            "The sqlite database you specified does not exist, please check that you have the"
+            "correct path."
+        )
+        sys.exit(1)
+
    sqlite_config = {
        "name": "sqlite3",
        "args": {
@@ -388,6 +388,62 @@ class ApplicationServiceApi(SimpleHttpClient):
        failed_transactions_counter.labels(service.id).inc()
        return False

+    async def claim_client_keys(
+        self, service: "ApplicationService", query: List[Tuple[str, str, str]]
+    ) -> Tuple[Dict[str, Dict[str, Dict[str, JsonDict]]], List[Tuple[str, str, str]]]:
+        """Claim one time keys from an application service.
+
+        Args:
+            query: An iterable of tuples of (user ID, device ID, algorithm).
+
+        Returns:
+            A tuple of:
+                A map of user ID -> a map device ID -> a map of key ID -> JSON dict.
+
+                A copy of the input which has not been fulfilled because the
+                appservice doesn't support this endpoint or has not returned
+                data for that tuple.
+        """
+        if service.url is None:
+            return {}, query
+
+        # This is required by the configuration.
+        assert service.hs_token is not None
+
+        # Create the expected payload shape.
+        body: Dict[str, Dict[str, List[str]]] = {}
+        for user_id, device, algorithm in query:
+            body.setdefault(user_id, {}).setdefault(device, []).append(algorithm)
+
+        uri = f"{service.url}/_matrix/app/unstable/org.matrix.msc3983/keys/claim"
+        try:
+            response = await self.post_json_get_json(
+                uri,
+                body,
+                headers={"Authorization": [f"Bearer {service.hs_token}"]},
+            )
+        except CodeMessageException as e:
+            # The appservice doesn't support this endpoint.
+            if e.code == 404 or e.code == 405:
+                return {}, query
+            logger.warning("claim_keys to %s received %s", uri, e.code)
+            return {}, query
+        except Exception as ex:
+            logger.warning("claim_keys to %s threw exception %s", uri, ex)
+            return {}, query
+
+        # Check if the appservice fulfilled all of the queried user/device/algorithms
+        # or if some are still missing.
+        #
+        # TODO This places a lot of faith in the response shape being correct.
+        missing = [
+            (user_id, device, algorithm)
+            for user_id, device, algorithm in query
+            if algorithm not in response.get(user_id, {}).get(device, [])
+        ]
+
+        return response, missing
+
    def _serialize(
        self, service: "ApplicationService", events: Iterable[EventBase]
    ) -> List[JsonDict]:
@@ -74,6 +74,11 @@ class ExperimentalConfig(Config):
            "msc3202_transaction_extensions", False
        )

+        # MSC3983: Proxying OTK claim requests to exclusive ASes.
+        self.msc3983_appservice_otk_claims: bool = experimental.get(
+            "msc3983_appservice_otk_claims", False
+        )
+
        # MSC3706 (server-side support for partial state in /send_join responses)
        # Synapse will always serve partial state responses to requests using the stable
        # query parameter `omit_members`. If this flag is set, Synapse will also serve
@@ -51,7 +51,7 @@ def check_event_content_hash(
    # some malformed events lack a 'hashes'. Protect against it being missing
    # or a weird type by basically treating it the same as an unhashed event.
    hashes = event.get("hashes")
-    # nb it might be a frozendict or a dict
+    # nb it might be a immutabledict or a dict
    if not isinstance(hashes, collections.abc.Mapping):
        raise SynapseError(
            400, "Malformed 'hashes': %s" % (type(hashes),), Codes.UNAUTHORIZED
@@ -462,7 +462,7 @@ class FrozenEvent(EventBase):
        # Signatures is a dict of dicts, and this is faster than doing a
        # copy.deepcopy
        signatures = {
-            name: {sig_id: sig for sig_id, sig in sigs.items()}
+            name: dict(sigs.items())
            for name, sigs in event_dict.pop("signatures", {}).items()
        }

@@ -510,7 +510,7 @@ class FrozenEventV2(EventBase):
        # Signatures is a dict of dicts, and this is faster than doing a
        # copy.deepcopy
        signatures = {
-            name: {sig_id: sig for sig_id, sig in sigs.items()}
+            name: dict(sigs.items())
            for name, sigs in event_dict.pop("signatures", {}).items()
        }

@@ -15,7 +15,7 @@ from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, List, Optional, Tuple

 import attr
-from frozendict import frozendict
+from immutabledict import immutabledict

 from synapse.appservice import ApplicationService
 from synapse.events import EventBase
@@ -489,4 +489,4 @@ def _decode_state_dict(
    if input is None:
        return None

-    return frozendict({(etype, state_key): v for etype, state_key, v in input})
+    return immutabledict({(etype, state_key): v for etype, state_key, v in input})
@@ -355,7 +355,7 @@ def serialize_event(
    time_now_ms = int(time_now_ms)

    # Should this strip out None's?
-    d = {k: v for k, v in e.get_dict().items()}
+    d = dict(e.get_dict().items())

    d["event_id"] = e.event_id

@@ -567,7 +567,7 @@ PowerLevelsContent = Mapping[str, Union[_PowerLevel, Mapping[str, _PowerLevel]]]
 def copy_and_fixup_power_levels_contents(
    old_power_levels: PowerLevelsContent,
 ) -> Dict[str, Union[int, Dict[str, int]]]:
-    """Copy the content of a power_levels event, unfreezing frozendicts along the way.
+    """Copy the content of a power_levels event, unfreezing immutabledicts along the way.

    We accept as input power level values which are strings, provided they represent an
    integer, e.g. `"`100"` instead of 100. Such strings are converted to integers
@@ -12,11 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import collections.abc
-from typing import Iterable, Type, Union, cast
+from typing import Iterable, List, Type, Union, cast

 import jsonschema
+from pydantic import Field, StrictBool, StrictStr

-from synapse.api.constants import MAX_ALIAS_LENGTH, EventTypes, Membership
+from synapse.api.constants import (
+    MAX_ALIAS_LENGTH,
+    EventContentFields,
+    EventTypes,
+    Membership,
+)
 from synapse.api.errors import Codes, SynapseError
 from synapse.api.room_versions import EventFormatVersions
 from synapse.config.homeserver import HomeServerConfig
@@ -28,6 +34,8 @@ from synapse.events.utils import (
    validate_canonicaljson,
 )
 from synapse.federation.federation_server import server_matches_acl_event
+from synapse.http.servlet import validate_json_object
+from synapse.rest.models import RequestBodyModel
 from synapse.types import EventID, JsonDict, RoomID, UserID


@@ -88,27 +96,27 @@ class EventValidator:
                            Codes.INVALID_PARAM,
                        )

-        if event.type == EventTypes.Retention:
+        elif event.type == EventTypes.Retention:
            self._validate_retention(event)

-        if event.type == EventTypes.ServerACL:
+        elif event.type == EventTypes.ServerACL:
            if not server_matches_acl_event(config.server.server_name, event):
                raise SynapseError(
                    400, "Can't create an ACL event that denies the local server"
                )

-        if event.type == EventTypes.PowerLevels:
+        elif event.type == EventTypes.PowerLevels:
            try:
                jsonschema.validate(
                    instance=event.content,
                    schema=POWER_LEVELS_SCHEMA,
-                    cls=plValidator,
+                    cls=POWER_LEVELS_VALIDATOR,
                )
            except jsonschema.ValidationError as e:
                if e.path:
                    # example: "users_default": '0' is not of type 'integer'
                    # cast safety: path entries can be integers, if we fail to validate
-                    # items in an array. However the POWER_LEVELS_SCHEMA doesn't expect
+                    # items in an array. However, the POWER_LEVELS_SCHEMA doesn't expect
                    # to see any arrays.
                    message = (
                        '"' + cast(str, e.path[-1]) + '": ' + e.message  # noqa: B306
@@ -125,6 +133,15 @@ class EventValidator:
                    errcode=Codes.BAD_JSON,
                )

+        # If the event contains a mentions key, validate it.
+        if (
+            EventContentFields.MSC3952_MENTIONS in event.content
+            and config.experimental.msc3952_intentional_mentions
+        ):
+            validate_json_object(
+                event.content[EventContentFields.MSC3952_MENTIONS], Mentions
+            )
+
    def _validate_retention(self, event: EventBase) -> None:
        """Checks that an event that defines the retention policy for a room respects the
        format enforced by the spec.
@@ -253,12 +270,17 @@ POWER_LEVELS_SCHEMA = {
 }


+class Mentions(RequestBodyModel):
+    user_ids: List[StrictStr] = Field(default_factory=list)
+    room: StrictBool = False
+
+
 # This could return something newer than Draft 7, but that's the current "latest"
 # validator.
-def _create_power_level_validator() -> Type[jsonschema.Draft7Validator]:
-    validator = jsonschema.validators.validator_for(POWER_LEVELS_SCHEMA)
+def _create_validator(schema: JsonDict) -> Type[jsonschema.Draft7Validator]:
+    validator = jsonschema.validators.validator_for(schema)

-    # by default jsonschema does not consider a frozendict to be an object so
+    # by default jsonschema does not consider a immutabledict to be an object so
    # we need to use a custom type checker
    # https://python-jsonschema.readthedocs.io/en/stable/validate/?highlight=object#validating-with-additional-types
    type_checker = validator.TYPE_CHECKER.redefine(
@@ -268,4 +290,4 @@ def _create_power_level_validator() -> Type[jsonschema.Draft7Validator]:
    return jsonschema.validators.extend(validator, type_checker=type_checker)


-plValidator = _create_power_level_validator()
+POWER_LEVELS_VALIDATOR = _create_validator(POWER_LEVELS_SCHEMA)
@@ -86,7 +86,7 @@ from synapse.storage.databases.main.lock import Lock
 from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
 from synapse.storage.roommember import MemberSummary
 from synapse.types import JsonDict, StateMap, get_domain_from_id
-from synapse.util import json_decoder, unwrapFirstError
+from synapse.util import unwrapFirstError
 from synapse.util.async_helpers import Linearizer, concurrently_execute, gather_results
 from synapse.util.caches.response_cache import ResponseCache
 from synapse.util.stringutils import parse_server_name
@@ -135,6 +135,7 @@ class FederationServer(FederationBase):
        self.state = hs.get_state_handler()
        self._event_auth_handler = hs.get_event_auth_handler()
        self._room_member_handler = hs.get_room_member_handler()
+        self._e2e_keys_handler = hs.get_e2e_keys_handler()

        self._state_storage_controller = hs.get_storage_controllers().state

@@ -1012,15 +1013,14 @@ class FederationServer(FederationBase):
                query.append((user_id, device_id, algorithm))

        log_kv({"message": "Claiming one time keys.", "user, device pairs": query})
-        results = await self.store.claim_e2e_one_time_keys(query)
+        results = await self._e2e_keys_handler.claim_local_one_time_keys(query)

-        json_result: Dict[str, Dict[str, dict]] = {}
-        for user_id, device_keys in results.items():
-            for device_id, keys in device_keys.items():
-                for key_id, json_str in keys.items():
-                    json_result.setdefault(user_id, {})[device_id] = {
-                        key_id: json_decoder.decode(json_str)
-                    }
+        json_result: Dict[str, Dict[str, Dict[str, JsonDict]]] = {}
+        for result in results:
+            for user_id, device_keys in result.items():
+                for device_id, keys in device_keys.items():
+                    for key_id, key in keys.items():
+                        json_result.setdefault(user_id, {})[device_id] = {key_id: key}

        logger.info(
            "Claimed one-time-keys: %s",
@@ -244,7 +244,7 @@ class FederationRemoteSendQueue(AbstractFederationSender):

        self.notifier.on_new_replication_data()

-    def send_device_messages(self, destination: str, immediate: bool = False) -> None:
+    def send_device_messages(self, destination: str, immediate: bool = True) -> None:
        """As per FederationSender"""
        # We don't need to replicate this as it gets sent down a different
        # stream.
@@ -11,6 +11,119 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+"""
+The Federation Sender is responsible for sending Persistent Data Units (PDUs)
+and Ephemeral Data Units (EDUs) to other homeservers using
+the `/send` Federation API.
+
+
+## How do PDUs get sent?
+
+The Federation Sender is made aware of new PDUs due to `FederationSender.notify_new_events`.
+When the sender is notified about a newly-persisted PDU that originates from this homeserver
+and is not an out-of-band event, we pass the PDU to the `_PerDestinationQueue` for each
+remote homeserver that is in the room at that point in the DAG.
+
+
+### Per-Destination Queues
+
+There is one `PerDestinationQueue` per 'destination' homeserver.
+The `PerDestinationQueue` maintains the following information about the destination:
+
+- whether the destination is currently in [catch-up mode (see below)](#catch-up-mode);
+- a queue of PDUs to be sent to the destination; and
+- a queue of EDUs to be sent to the destination (not considered in this section).
+
+Upon a new PDU being enqueued, `attempt_new_transaction` is called to start a new
+transaction if there is not already one in progress.
+
+
+### Transactions and the Transaction Transmission Loop
+
+Each federation HTTP request to the `/send` endpoint is referred to as a 'transaction'.
+The body of the HTTP request contains a list of PDUs and EDUs to send to the destination.
+
+The *Transaction Transmission Loop* (`_transaction_transmission_loop`) is responsible
+for emptying the queued PDUs (and EDUs) from a `PerDestinationQueue` by sending
+them to the destination.
+
+There can only be one transaction in flight for a given destination at any time.
+(Other than preventing us from overloading the destination, this also makes it easier to
+reason about because we process events sequentially for each destination.
+This is useful for *Catch-Up Mode*, described later.)
+
+The loop continues so long as there is anything to send. At each iteration of the loop, we:
+
+- dequeue up to 50 PDUs (and up to 100 EDUs).
+- make the `/send` request to the destination homeserver with the dequeued PDUs and EDUs.
+- if successful, make note of the fact that we succeeded in transmitting PDUs up to
+  the given `stream_ordering` of the latest PDU by
+- if unsuccessful, back off from the remote homeserver for some time.
+  If we have been unsuccessful for too long (when the backoff interval grows to exceed 1 hour),
+  the in-memory queues are emptied and we enter [*Catch-Up Mode*, described below](#catch-up-mode).
+
+
+### Catch-Up Mode
+
+When the `PerDestinationQueue` has the catch-up flag set, the *Catch-Up Transmission Loop*
+(`_catch_up_transmission_loop`) is used in lieu of the regular `_transaction_transmission_loop`.
+(Only once the catch-up mode has been exited can the regular tranaction transmission behaviour
+be resumed.)
+
+*Catch-Up Mode*, entered upon Synapse startup or once a homeserver has fallen behind due to
+connection problems, is responsible for sending PDUs that have been missed by the destination
+homeserver. (PDUs can be missed because the `PerDestinationQueue` is volatile — i.e. resets
+on startup — and it does not hold PDUs forever if `/send` requests to the destination fail.)
+
+The catch-up mechanism makes use of the `last_successful_stream_ordering` column in the
+`destinations` table (which gives the `stream_ordering` of the most recent successfully
+sent PDU) and the `stream_ordering` column in the `destination_rooms` table (which gives,
+for each room, the `stream_ordering` of the most recent PDU that needs to be sent to this
+destination).
+
+Each iteration of the loop pulls out 50 `destination_rooms` entries with the oldest
+`stream_ordering`s that are greater than the `last_successful_stream_ordering`.
+In other words, from the set of latest PDUs in each room to be sent to the destination,
+the 50 oldest such PDUs are pulled out.
+
+These PDUs could, in principle, now be directly sent to the destination. However, as an
+optimisation intended to prevent overloading destination homeservers, we instead attempt
+to send the latest forward extremities so long as the destination homeserver is still
+eligible to receive those.
+This reduces load on the destination **in aggregate** because all Synapse homeservers
+will behave according to this principle and therefore avoid sending lots of different PDUs
+at different points in the DAG to a recovering homeserver.
+*This optimisation is not currently valid in rooms which are partial-state on this homeserver,
+since we are unable to determine whether the destination homeserver is eligible to receive
+the latest forward extremities unless this homeserver sent those PDUs — in this case, we
+just send the latest PDUs originating from this server and skip this optimisation.*
+
+Whilst PDUs are sent through this mechanism, the position of `last_successful_stream_ordering`
+is advanced as normal.
+Once there are no longer any rooms containing outstanding PDUs to be sent to the destination
+*that are not already in the `PerDestinationQueue` because they arrived since Catch-Up Mode
+was enabled*, Catch-Up Mode is exited and we return to `_transaction_transmission_loop`.
+
+
+#### A note on failures and back-offs
+
+If a remote server is unreachable over federation, we back off from that server,
+with an exponentially-increasing retry interval.
+Whilst we don't automatically retry after the interval, we prevent making new attempts
+until such time as the back-off has cleared.
+Once the back-off is cleared and a new PDU or EDU arrives for transmission, the transmission
+loop resumes and empties the queue by making federation requests.
+
+If the backoff grows too large (> 1 hour), the in-memory queue is emptied (to prevent
+unbounded growth) and Catch-Up Mode is entered.
+
+It is worth noting that the back-off for a remote server is cleared once an inbound
+request from that remote server is received (see `notify_remote_server_up`).
+At this point, the transaction transmission loop is also started up, to proactively
+send missed PDUs and EDUs to the destination (i.e. you don't need to wait for a new PDU
+or EDU, destined for that destination, to be created in order to send out missed PDUs and
+EDUs).
+"""

 import abc
 import logging
@@ -783,7 +896,7 @@ class FederationSender(AbstractFederationSender):
        else:
            queue.send_edu(edu)

-    def send_device_messages(self, destination: str, immediate: bool = False) -> None:
+    def send_device_messages(self, destination: str, immediate: bool = True) -> None:
        if destination == self.server_name:
            logger.warning("Not sending device update to ourselves")
            return
@@ -108,6 +108,7 @@ class PublicRoomList(BaseFederationServlet):
    """

    PATH = "/publicRooms"
+    CATEGORY = "Federation requests"

    def __init__(
        self,
@@ -212,6 +213,7 @@ class OpenIdUserInfo(BaseFederationServlet):
    """

    PATH = "/openid/userinfo"
+    CATEGORY = "Federation requests"

    REQUIRE_AUTH = False

@@ -70,6 +70,7 @@ class BaseFederationServerServlet(BaseFederationServlet):

 class FederationSendServlet(BaseFederationServerServlet):
    PATH = "/send/(?P<transaction_id>[^/]*)/?"
+    CATEGORY = "Inbound federation transaction request"

    # We ratelimit manually in the handler as we queue up the requests and we
    # don't want to fill up the ratelimiter with blocked requests.
@@ -138,6 +139,7 @@ class FederationSendServlet(BaseFederationServerServlet):

 class FederationEventServlet(BaseFederationServerServlet):
    PATH = "/event/(?P<event_id>[^/]*)/?"
+    CATEGORY = "Federation requests"

    # This is when someone asks for a data item for a given server data_id pair.
    async def on_GET(
@@ -152,6 +154,7 @@ class FederationEventServlet(BaseFederationServerServlet):

 class FederationStateV1Servlet(BaseFederationServerServlet):
    PATH = "/state/(?P<room_id>[^/]*)/?"
+    CATEGORY = "Federation requests"

    # This is when someone asks for all data for a given room.
    async def on_GET(
@@ -170,6 +173,7 @@ class FederationStateV1Servlet(BaseFederationServerServlet):

 class FederationStateIdsServlet(BaseFederationServerServlet):
    PATH = "/state_ids/(?P<room_id>[^/]*)/?"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -187,6 +191,7 @@ class FederationStateIdsServlet(BaseFederationServerServlet):

 class FederationBackfillServlet(BaseFederationServerServlet):
    PATH = "/backfill/(?P<room_id>[^/]*)/?"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -225,6 +230,7 @@ class FederationTimestampLookupServlet(BaseFederationServerServlet):
    """

    PATH = "/timestamp_to_event/(?P<room_id>[^/]*)/?"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -246,6 +252,7 @@ class FederationTimestampLookupServlet(BaseFederationServerServlet):

 class FederationQueryServlet(BaseFederationServerServlet):
    PATH = "/query/(?P<query_type>[^/]*)"
+    CATEGORY = "Federation requests"

    # This is when we receive a server-server Query
    async def on_GET(
@@ -262,6 +269,7 @@ class FederationQueryServlet(BaseFederationServerServlet):

 class FederationMakeJoinServlet(BaseFederationServerServlet):
    PATH = "/make_join/(?P<room_id>[^/]*)/(?P<user_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -297,6 +305,7 @@ class FederationMakeJoinServlet(BaseFederationServerServlet):

 class FederationMakeLeaveServlet(BaseFederationServerServlet):
    PATH = "/make_leave/(?P<room_id>[^/]*)/(?P<user_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -312,6 +321,7 @@ class FederationMakeLeaveServlet(BaseFederationServerServlet):

 class FederationV1SendLeaveServlet(BaseFederationServerServlet):
    PATH = "/send_leave/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_PUT(
        self,
@@ -327,6 +337,7 @@ class FederationV1SendLeaveServlet(BaseFederationServerServlet):

 class FederationV2SendLeaveServlet(BaseFederationServerServlet):
    PATH = "/send_leave/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    PREFIX = FEDERATION_V2_PREFIX

@@ -344,6 +355,7 @@ class FederationV2SendLeaveServlet(BaseFederationServerServlet):

 class FederationMakeKnockServlet(BaseFederationServerServlet):
    PATH = "/make_knock/(?P<room_id>[^/]*)/(?P<user_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -366,6 +378,7 @@ class FederationMakeKnockServlet(BaseFederationServerServlet):

 class FederationV1SendKnockServlet(BaseFederationServerServlet):
    PATH = "/send_knock/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_PUT(
        self,
@@ -381,6 +394,7 @@ class FederationV1SendKnockServlet(BaseFederationServerServlet):

 class FederationEventAuthServlet(BaseFederationServerServlet):
    PATH = "/event_auth/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -395,6 +409,7 @@ class FederationEventAuthServlet(BaseFederationServerServlet):

 class FederationV1SendJoinServlet(BaseFederationServerServlet):
    PATH = "/send_join/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_PUT(
        self,
@@ -412,6 +427,7 @@ class FederationV1SendJoinServlet(BaseFederationServerServlet):

 class FederationV2SendJoinServlet(BaseFederationServerServlet):
    PATH = "/send_join/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    PREFIX = FEDERATION_V2_PREFIX

@@ -455,6 +471,7 @@ class FederationV2SendJoinServlet(BaseFederationServerServlet):

 class FederationV1InviteServlet(BaseFederationServerServlet):
    PATH = "/invite/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_PUT(
        self,
@@ -479,6 +496,7 @@ class FederationV1InviteServlet(BaseFederationServerServlet):

 class FederationV2InviteServlet(BaseFederationServerServlet):
    PATH = "/invite/(?P<room_id>[^/]*)/(?P<event_id>[^/]*)"
+    CATEGORY = "Federation requests"

    PREFIX = FEDERATION_V2_PREFIX

@@ -515,6 +533,7 @@ class FederationV2InviteServlet(BaseFederationServerServlet):

 class FederationThirdPartyInviteExchangeServlet(BaseFederationServerServlet):
    PATH = "/exchange_third_party_invite/(?P<room_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_PUT(
        self,
@@ -529,6 +548,7 @@ class FederationThirdPartyInviteExchangeServlet(BaseFederationServerServlet):

 class FederationClientKeysQueryServlet(BaseFederationServerServlet):
    PATH = "/user/keys/query"
+    CATEGORY = "Federation requests"

    async def on_POST(
        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
@@ -538,6 +558,7 @@ class FederationClientKeysQueryServlet(BaseFederationServerServlet):

 class FederationUserDevicesQueryServlet(BaseFederationServerServlet):
    PATH = "/user/devices/(?P<user_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_GET(
        self,
@@ -551,6 +572,7 @@ class FederationUserDevicesQueryServlet(BaseFederationServerServlet):

 class FederationClientKeysClaimServlet(BaseFederationServerServlet):
    PATH = "/user/keys/claim"
+    CATEGORY = "Federation requests"

    async def on_POST(
        self, origin: str, content: JsonDict, query: Dict[bytes, List[bytes]]
@@ -561,6 +583,7 @@ class FederationClientKeysClaimServlet(BaseFederationServerServlet):

 class FederationGetMissingEventsServlet(BaseFederationServerServlet):
    PATH = "/get_missing_events/(?P<room_id>[^/]*)"
+    CATEGORY = "Federation requests"

    async def on_POST(
        self,
@@ -586,6 +609,7 @@ class FederationGetMissingEventsServlet(BaseFederationServerServlet):

 class On3pidBindServlet(BaseFederationServerServlet):
    PATH = "/3pid/onbind"
+    CATEGORY = "Federation requests"

    REQUIRE_AUTH = False

@@ -618,6 +642,7 @@ class On3pidBindServlet(BaseFederationServerServlet):

 class FederationVersionServlet(BaseFederationServlet):
    PATH = "/version"
+    CATEGORY = "Federation requests"

    REQUIRE_AUTH = False

@@ -640,6 +665,7 @@ class FederationVersionServlet(BaseFederationServlet):

 class FederationRoomHierarchyServlet(BaseFederationServlet):
    PATH = "/hierarchy/(?P<room_id>[^/]*)"
+    CATEGORY = "Federation requests"

    def __init__(
        self,
@@ -672,6 +698,7 @@ class RoomComplexityServlet(BaseFederationServlet):

    PATH = "/rooms/(?P<room_id>[^/]*)/complexity"
    PREFIX = FEDERATION_UNSTABLE_PREFIX
+    CATEGORY = "Federation requests (unstable)"

    def __init__(
        self,
@@ -12,7 +12,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import TYPE_CHECKING, Collection, Dict, Iterable, List, Optional, Union
+from typing import (
+    TYPE_CHECKING,
+    Collection,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Tuple,
+    Union,
+)

 from prometheus_client import Counter

@@ -829,3 +838,66 @@ class ApplicationServicesHandler:
        if unknown_user:
            return await self.query_user_exists(user_id)
        return True
+
+    async def claim_e2e_one_time_keys(
+        self, query: Iterable[Tuple[str, str, str]]
+    ) -> Tuple[
+        Iterable[Dict[str, Dict[str, Dict[str, JsonDict]]]], List[Tuple[str, str, str]]
+    ]:
+        """Claim one time keys from application services.
+
+        Args:
+            query: An iterable of tuples of (user ID, device ID, algorithm).
+
+        Returns:
+            A tuple of:
+                An iterable of maps of user ID -> a map device ID -> a map of key ID -> JSON bytes.
+
+                A copy of the input which has not been fulfilled (either because
+                they are not appservice users or the appservice does not support
+                providing OTKs).
+        """
+        services = self.store.get_app_services()
+
+        # Partition the users by appservice.
+        query_by_appservice: Dict[str, List[Tuple[str, str, str]]] = {}
+        missing = []
+        for user_id, device, algorithm in query:
+            if not self.store.get_if_app_services_interested_in_user(user_id):
+                missing.append((user_id, device, algorithm))
+                continue
+
+            # Find the associated appservice.
+            for service in services:
+                if service.is_exclusive_user(user_id):
+                    query_by_appservice.setdefault(service.id, []).append(
+                        (user_id, device, algorithm)
+                    )
+                    continue
+
+        # Query each service in parallel.
+        results = await make_deferred_yieldable(
+            defer.DeferredList(
+                [
+                    run_in_background(
+                        self.appservice_api.claim_client_keys,
+                        # We know this must be an app service.
+                        self.store.get_app_service_by_id(service_id),  # type: ignore[arg-type]
+                        service_query,
+                    )
+                    for service_id, service_query in query_by_appservice.items()
+                ],
+                consumeErrors=True,
+            )
+        )
+
+        # Patch together the results -- they are all independent (since they
+        # require exclusive control over the users). They get returned as a list
+        # and the caller combines them.
+        claimed_keys: List[Dict[str, Dict[str, Dict[str, JsonDict]]]] = []
+        for success, result in results:
+            if success:
+                claimed_keys.append(result[0])
+                missing.extend(result[1])
+
+        return claimed_keys, missing
@@ -1504,8 +1504,10 @@ class AuthHandler:
        )

        # delete pushers associated with this access token
+        # XXX(quenting): This is only needed until the 'set_device_id_for_pushers'
+        # background update completes.
        if token.token_id is not None:
-            await self.hs.get_pusherpool().remove_pushers_by_access_token(
+            await self.hs.get_pusherpool().remove_pushers_by_access_tokens(
                token.user_id, (token.token_id,)
            )

@@ -1535,7 +1537,9 @@ class AuthHandler:
            )

        # delete pushers associated with the access tokens
-        await self.hs.get_pusherpool().remove_pushers_by_access_token(
+        # XXX(quenting): This is only needed until the 'set_device_id_for_pushers'
+        # background update completes.
+        await self.hs.get_pusherpool().remove_pushers_by_access_tokens(
            user_id, (token_id for _, token_id, _ in tokens_and_devices)
        )

@@ -485,7 +485,7 @@ class DeviceHandler(DeviceWorkerHandler):
            device_ids = [d for d in device_ids if d != except_device_id]
        await self.delete_devices(user_id, device_ids)

-    async def delete_devices(self, user_id: str, device_ids: List[str]) -> None:
+    async def delete_devices(self, user_id: str, device_ids: StrCollection) -> None:
        """Delete several devices

        Args:
@@ -503,6 +503,8 @@ class DeviceHandler(DeviceWorkerHandler):
            else:
                raise

+        await self.hs.get_pusherpool().remove_pushers_by_devices(user_id, device_ids)
+
        # Delete data specific to each device. Not optimised as it is not
        # considered as part of a critical path.
        for device_id in device_ids:
@@ -13,7 +13,6 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 import logging
 from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Mapping, Optional, Tuple

@@ -53,6 +52,7 @@ class E2eKeysHandler:
        self.store = hs.get_datastores().main
        self.federation = hs.get_federation_client()
        self.device_handler = hs.get_device_handler()
+        self._appservice_handler = hs.get_application_service_handler()
        self.is_mine = hs.is_mine
        self.clock = hs.get_clock()

@@ -88,6 +88,10 @@ class E2eKeysHandler:
            max_count=10,
        )

+        self._query_appservices_for_otks = (
+            hs.config.experimental.msc3983_appservice_otk_claims
+        )
+
    @trace
    @cancellable
    async def query_devices(
@@ -542,6 +546,42 @@ class E2eKeysHandler:

        return ret

+    async def claim_local_one_time_keys(
+        self, local_query: List[Tuple[str, str, str]]
+    ) -> Iterable[Dict[str, Dict[str, Dict[str, JsonDict]]]]:
+        """Claim one time keys for local users.
+
+        1. Attempt to claim OTKs from the database.
+        2. Ask application services if they provide OTKs.
+        3. Attempt to fetch fallback keys from the database.
+
+        Args:
+            local_query: An iterable of tuples of (user ID, device ID, algorithm).
+
+        Returns:
+            An iterable of maps of user ID -> a map device ID -> a map of key ID -> JSON bytes.
+        """
+
+        otk_results, not_found = await self.store.claim_e2e_one_time_keys(local_query)
+
+        # If the application services have not provided any keys via the C-S
+        # API, query it directly for one-time keys.
+        if self._query_appservices_for_otks:
+            (
+                appservice_results,
+                not_found,
+            ) = await self._appservice_handler.claim_e2e_one_time_keys(not_found)
+        else:
+            appservice_results = []
+
+        # For each user that does not have a one-time keys available, see if
+        # there is a fallback key.
+        fallback_results = await self.store.claim_e2e_fallback_keys(not_found)
+
+        # Return the results in order, each item from the input query should
+        # only appear once in the combined list.
+        return (otk_results, *appservice_results, fallback_results)
+
    @trace
    async def claim_one_time_keys(
        self, query: Dict[str, Dict[str, Dict[str, str]]], timeout: Optional[int]
@@ -561,17 +601,18 @@ class E2eKeysHandler:
        set_tag("local_key_query", str(local_query))
        set_tag("remote_key_query", str(remote_queries))

-        results = await self.store.claim_e2e_one_time_keys(local_query)
+        results = await self.claim_local_one_time_keys(local_query)

        # A map of user ID -> device ID -> key ID -> key.
        json_result: Dict[str, Dict[str, Dict[str, JsonDict]]] = {}
+        for result in results:
+            for user_id, device_keys in result.items():
+                for device_id, keys in device_keys.items():
+                    for key_id, key in keys.items():
+                        json_result.setdefault(user_id, {})[device_id] = {key_id: key}
+
+        # Remote failures.
        failures: Dict[str, JsonDict] = {}
-        for user_id, device_keys in results.items():
-            for device_id, keys in device_keys.items():
-                for key_id, json_str in keys.items():
-                    json_result.setdefault(user_id, {})[device_id] = {
-                        key_id: json_decoder.decode(json_str)
-                    }

        @trace
        async def claim_client_keys(destination: str) -> None:
@@ -583,7 +583,7 @@ class FederationEventHandler:

            await self._check_event_auth(origin, event, context)
            if context.rejected:
-                raise SynapseError(400, "Join event was rejected")
+                raise SynapseError(403, "Join event was rejected")

            # the remote server is responsible for sending our join event to the rest
            # of the federation. Indeed, attempting to do so will result in problems
@@ -16,7 +16,7 @@
 """Contains functions for registering clients."""

 import logging
-from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple
+from typing import TYPE_CHECKING, Iterable, List, Optional, Set, Tuple

 from prometheus_client import Counter
 from typing_extensions import TypedDict
@@ -40,6 +40,7 @@ from synapse.appservice import ApplicationService
 from synapse.config.server import is_threepid_reserved
 from synapse.handlers.device import DeviceHandler
 from synapse.http.servlet import assert_params_in_dict
+from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.replication.http.login import RegisterDeviceReplicationServlet
 from synapse.replication.http.register import (
    ReplicationPostRegisterActionsServlet,
@@ -48,6 +49,7 @@ from synapse.replication.http.register import (
 from synapse.spam_checker_api import RegistrationBehaviour
 from synapse.types import RoomAlias, UserID, create_requester
 from synapse.types.state import StateFilter
+from synapse.util.iterutils import batch_iter

 if TYPE_CHECKING:
    from synapse.server import HomeServer
@@ -110,6 +112,10 @@ class RegistrationHandler:
        self._server_notices_mxid = hs.config.servernotices.server_notices_mxid
        self._server_name = hs.hostname

+        # The set of users that we're currently pruning devices for. Ensures
+        # that we don't have two such jobs for the same user running at once.
+        self._currently_pruning_devices_for_users: Set[str] = set()
+
        self.spam_checker = hs.get_spam_checker()

        if hs.config.worker.worker_app:
@@ -121,7 +127,10 @@ class RegistrationHandler:
                ReplicationPostRegisterActionsServlet.make_client(hs)
            )
        else:
-            self.device_handler = hs.get_device_handler()
+            device_handler = hs.get_device_handler()
+            assert isinstance(device_handler, DeviceHandler)
+            self.device_handler = device_handler
+
            self._register_device_client = self.register_device_inner
            self.pusher_pool = hs.get_pusherpool()

@@ -851,6 +860,9 @@ class RegistrationHandler:
        # This can only run on the main process.
        assert isinstance(self.device_handler, DeviceHandler)

+        # Prune the user's device list if they already have a lot of devices.
+        await self._maybe_prune_too_many_devices(user_id)
+
        registered_device_id = await self.device_handler.check_device_registered(
            user_id,
            device_id,
@@ -919,6 +931,40 @@ class RegistrationHandler:
            "refresh_token": refresh_token,
        }

+    async def _maybe_prune_too_many_devices(self, user_id: str) -> None:
+        """Delete any excess old devices this user may have."""
+
+        if user_id in self._currently_pruning_devices_for_users:
+            return
+
+        # We also cap the number of users whose devices we prune at the same
+        # time, to avoid performance problems.
+        if len(self._currently_pruning_devices_for_users) > 5:
+            return
+
+        device_ids = await self.store.check_too_many_devices_for_user(user_id)
+        if not device_ids:
+            return
+
+        # Now spawn a background loop that deletes said devices.
+        async def _prune_too_many_devices_loop() -> None:
+            if user_id in self._currently_pruning_devices_for_users:
+                return
+
+            self._currently_pruning_devices_for_users.add(user_id)
+
+            try:
+                for batch in batch_iter(device_ids, 10):
+                    await self.device_handler.delete_devices(user_id, batch)
+
+                    await self.clock.sleep(60)
+            finally:
+                self._currently_pruning_devices_for_users.discard(user_id)
+
+        run_as_background_process(
+            "_prune_too_many_devices_loop", _prune_too_many_devices_loop
+        )
+
    async def post_registration_actions(
        self, user_id: str, auth_result: dict, access_token: Optional[str]
    ) -> None:
@@ -1013,11 +1059,11 @@ class RegistrationHandler:
            user_tuple = await self.store.get_user_by_access_token(token)
            # The token better still exist.
            assert user_tuple
-            token_id = user_tuple.token_id
+            device_id = user_tuple.device_id

            await self.pusher_pool.add_or_update_pusher(
                user_id=user_id,
-                access_token=token_id,
+                device_id=device_id,
                kind="email",
                app_id="m.email",
                app_display_name="Email Notifications",
@@ -864,64 +864,69 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        # `is_partial_state_room` also indicates whether `partial_state_before_join` is
        # partial.

-        # TODO: Refactor into dictionary of explicitly allowed transitions
-        # between old and new state, with specific error messages for some
-        # transitions and generic otherwise
-        old_state_id = partial_state_before_join.get(
-            (EventTypes.Member, target.to_string())
-        )
-        if old_state_id:
-            old_state = await self.store.get_event(old_state_id, allow_none=True)
-            old_membership = old_state.content.get("membership") if old_state else None
-            if action == "unban" and old_membership != "ban":
-                raise SynapseError(
-                    403,
-                    "Cannot unban user who was not banned"
-                    " (membership=%s)" % old_membership,
-                    errcode=Codes.BAD_STATE,
-                )
-            if old_membership == "ban" and action not in ["ban", "unban", "leave"]:
-                raise SynapseError(
-                    403,
-                    "Cannot %s user who was banned" % (action,),
-                    errcode=Codes.BAD_STATE,
-                )
-
-            if old_state:
-                same_content = content == old_state.content
-                same_membership = old_membership == effective_membership_state
-                same_sender = requester.user.to_string() == old_state.sender
-                if same_sender and same_membership and same_content:
-                    # duplicate event.
-                    # we know it was persisted, so must have a stream ordering.
-                    assert old_state.internal_metadata.stream_ordering
-                    return (
-                        old_state.event_id,
-                        old_state.internal_metadata.stream_ordering,
-                    )
-
-            if old_membership in ["ban", "leave"] and action == "kick":
-                raise AuthError(403, "The target user is not in the room")
-
-            # we don't allow people to reject invites to the server notice
-            # room, but they can leave it once they are joined.
-            if (
-                old_membership == Membership.INVITE
-                and effective_membership_state == Membership.LEAVE
-            ):
-                is_blocked = await self.store.is_server_notice_room(room_id)
-                if is_blocked:
-                    raise SynapseError(
-                        HTTPStatus.FORBIDDEN,
-                        "You cannot reject this invite",
-                        errcode=Codes.CANNOT_LEAVE_SERVER_NOTICE_ROOM,
-                    )
-        else:
-            if action == "kick":
-                raise AuthError(403, "The target user is not in the room")
-
        is_host_in_room = await self._is_host_in_room(partial_state_before_join)

+        # if we are not in the room, we won't have the current state
+        if is_host_in_room:
+            # TODO: Refactor into dictionary of explicitly allowed transitions
+            # between old and new state, with specific error messages for some
+            # transitions and generic otherwise
+            old_state_id = partial_state_before_join.get(
+                (EventTypes.Member, target.to_string())
+            )
+
+            if old_state_id:
+                old_state = await self.store.get_event(old_state_id, allow_none=True)
+                old_membership = (
+                    old_state.content.get("membership") if old_state else None
+                )
+                if action == "unban" and old_membership != "ban":
+                    raise SynapseError(
+                        403,
+                        "Cannot unban user who was not banned"
+                        " (membership=%s)" % old_membership,
+                        errcode=Codes.BAD_STATE,
+                    )
+                if old_membership == "ban" and action not in ["ban", "unban", "leave"]:
+                    raise SynapseError(
+                        403,
+                        "Cannot %s user who was banned" % (action,),
+                        errcode=Codes.BAD_STATE,
+                    )
+
+                if old_state:
+                    same_content = content == old_state.content
+                    same_membership = old_membership == effective_membership_state
+                    same_sender = requester.user.to_string() == old_state.sender
+                    if same_sender and same_membership and same_content:
+                        # duplicate event.
+                        # we know it was persisted, so must have a stream ordering.
+                        assert old_state.internal_metadata.stream_ordering
+                        return (
+                            old_state.event_id,
+                            old_state.internal_metadata.stream_ordering,
+                        )
+
+                if old_membership in ["ban", "leave"] and action == "kick":
+                    raise AuthError(403, "The target user is not in the room")
+
+                # we don't allow people to reject invites to the server notice
+                # room, but they can leave it once they are joined.
+                if (
+                    old_membership == Membership.INVITE
+                    and effective_membership_state == Membership.LEAVE
+                ):
+                    is_blocked = await self.store.is_server_notice_room(room_id)
+                    if is_blocked:
+                        raise SynapseError(
+                            HTTPStatus.FORBIDDEN,
+                            "You cannot reject this invite",
+                            errcode=Codes.CANNOT_LEAVE_SERVER_NOTICE_ROOM,
+                        )
+            else:
+                if action == "kick":
+                    raise AuthError(403, "The target user is not in the room")
+
        if effective_membership_state == Membership.JOIN:
            if requester.is_guest:
                guest_can_join = await self._can_guest_join(partial_state_before_join)
@@ -52,6 +52,11 @@ FEDERATION_TIMEOUT = 60 * 1000
 FEDERATION_PING_INTERVAL = 40 * 1000


+# How long to remember a typing notification happened in a room before
+# forgetting about it.
+FORGET_TIMEOUT = 10 * 60 * 1000
+
+
 class FollowerTypingHandler:
    """A typing handler on a different process than the writer that is updated
    via replication.
@@ -83,7 +88,10 @@ class FollowerTypingHandler:
        self.wheel_timer: WheelTimer[RoomMember] = WheelTimer(bucket_size=5000)
        self._latest_room_serial = 0

+        self._rooms_updated: Set[str] = set()
+
        self.clock.looping_call(self._handle_timeouts, 5000)
+        self.clock.looping_call(self._prune_old_typing, FORGET_TIMEOUT)

    def _reset(self) -> None:
        """Reset the typing handler's data caches."""
@@ -92,6 +100,8 @@ class FollowerTypingHandler:
        # map room IDs to sets of users currently typing
        self._room_typing = {}

+        self._rooms_updated = set()
+
        self._member_last_federation_poke = {}
        self.wheel_timer = WheelTimer(bucket_size=5000)

@@ -178,6 +188,7 @@ class FollowerTypingHandler:
            prev_typing = self._room_typing.get(row.room_id, set())
            now_typing = set(row.user_ids)
            self._room_typing[row.room_id] = now_typing
+            self._rooms_updated.add(row.room_id)

            if self.federation:
                run_as_background_process(
@@ -209,6 +220,19 @@ class FollowerTypingHandler:
    def get_current_token(self) -> int:
        return self._latest_room_serial

+    def _prune_old_typing(self) -> None:
+        """Prune rooms that haven't seen typing updates since last time.
+
+        This is safe to do as clients should time out old typing notifications.
+        """
+        stale_rooms = self._room_serials.keys() - self._rooms_updated
+
+        for room_id in stale_rooms:
+            self._room_serials.pop(room_id, None)
+            self._room_typing.pop(room_id, None)
+
+        self._rooms_updated = set()
+

 class TypingWriterHandler(FollowerTypingHandler):
    def __init__(self, hs: "HomeServer"):
@@ -388,6 +412,7 @@ class TypingWriterHandler(FollowerTypingHandler):
        self._typing_stream_change_cache.entity_has_changed(
            member.room_id, self._latest_room_serial
        )
+        self._rooms_updated.add(member.room_id)

        self.notifier.on_new_event(
            StreamKeyType.TYPING, self._latest_room_serial, rooms=[member.room_id]
@@ -778,17 +778,13 @@ def parse_json_object_from_request(
 Model = TypeVar("Model", bound=BaseModel)


-def parse_and_validate_json_object_from_request(
-    request: Request, model_type: Type[Model]
-) -> Model:
-    """Parse a JSON object from the body of a twisted HTTP request, then deserialise and
-    validate using the given pydantic model.
+def validate_json_object(content: JsonDict, model_type: Type[Model]) -> Model:
+    """Validate a deserialized JSON object using the given pydantic model.

    Raises:
        SynapseError if the request body couldn't be decoded as JSON or
            if it wasn't a JSON object.
    """
-    content = parse_json_object_from_request(request, allow_empty_body=False)
    try:
        instance = model_type.parse_obj(content)
    except ValidationError as e:
@@ -811,6 +807,20 @@ def parse_and_validate_json_object_from_request(
    return instance


+def parse_and_validate_json_object_from_request(
+    request: Request, model_type: Type[Model]
+) -> Model:
+    """Parse a JSON object from the body of a twisted HTTP request, then deserialise and
+    validate using the given pydantic model.
+
+    Raises:
+        SynapseError if the request body couldn't be decoded as JSON or
+            if it wasn't a JSON object.
+    """
+    content = parse_json_object_from_request(request, allow_empty_body=False)
+    return validate_json_object(content, model_type)
+
+
 def assert_params_in_dict(body: JsonDict, required: Iterable[str]) -> None:
    absent = []
    for k in required:
@@ -103,7 +103,7 @@ class PusherConfig:

    id: Optional[str]
    user_name: str
-    access_token: Optional[int]
+
    profile_tag: str
    kind: str
    app_id: str
@@ -119,6 +119,11 @@ class PusherConfig:
    enabled: bool
    device_id: Optional[str]

+    # XXX(quenting): The access_token is not persisted anymore for new pushers, but we
+    # keep it when reading from the database, so that we don't get stale pushers
+    # while the "set_device_id_for_pushers" background update is running.
+    access_token: Optional[int]
+
    def as_dict(self) -> Dict[str, Any]:
        """Information that can be retrieved about a pusher after creation."""
        return {
@@ -25,7 +25,7 @@ from synapse.metrics.background_process_metrics import (
 from synapse.push import Pusher, PusherConfig, PusherConfigException
 from synapse.push.pusher import PusherFactory
 from synapse.replication.http.push import ReplicationRemovePusherRestServlet
-from synapse.types import JsonDict, RoomStreamToken
+from synapse.types import JsonDict, RoomStreamToken, StrCollection
 from synapse.util.async_helpers import concurrently_execute
 from synapse.util.threepids import canonicalise_email

@@ -97,7 +97,6 @@ class PusherPool:
    async def add_or_update_pusher(
        self,
        user_id: str,
-        access_token: Optional[int],
        kind: str,
        app_id: str,
        app_display_name: str,
@@ -128,6 +127,22 @@ class PusherPool:
        # stream ordering, so it will process pushes from this point onwards.
        last_stream_ordering = self.store.get_room_max_stream_ordering()

+        # Before we actually persist the pusher, we check if the user already has one
+        # for this app ID and pushkey. If so, we want to keep the access token and
+        # device ID in place, since this could be one device modifying
+        # (e.g. enabling/disabling) another device's pusher.
+        # XXX(quenting): Even though we're not persisting the access_token_id for new
+        # pushers anymore, we still need to copy existing access_token_ids over when
+        # updating a pusher, in case the "set_device_id_for_pushers" background update
+        # hasn't run yet.
+        access_token_id = None
+        existing_config = await self._get_pusher_config_for_user_by_app_id_and_pushkey(
+            user_id, app_id, pushkey
+        )
+        if existing_config:
+            device_id = existing_config.device_id
+            access_token_id = existing_config.access_token
+
        # we try to create the pusher just to validate the config: it
        # will then get pulled out of the database,
        # recreated, added and started: this means we have only one
@@ -136,7 +151,6 @@ class PusherPool:
            PusherConfig(
                id=None,
                user_name=user_id,
-                access_token=access_token,
                profile_tag=profile_tag,
                kind=kind,
                app_id=app_id,
@@ -151,23 +165,12 @@ class PusherPool:
                failing_since=None,
                enabled=enabled,
                device_id=device_id,
+                access_token=access_token_id,
            )
        )

-        # Before we actually persist the pusher, we check if the user already has one
-        # this app ID and pushkey. If so, we want to keep the access token and device ID
-        # in place, since this could be one device modifying (e.g. enabling/disabling)
-        # another device's pusher.
-        existing_config = await self._get_pusher_config_for_user_by_app_id_and_pushkey(
-            user_id, app_id, pushkey
-        )
-        if existing_config:
-            access_token = existing_config.access_token
-            device_id = existing_config.device_id
-
        await self.store.add_pusher(
            user_id=user_id,
-            access_token=access_token,
            kind=kind,
            app_id=app_id,
            app_display_name=app_display_name,
@@ -180,6 +183,7 @@ class PusherPool:
            profile_tag=profile_tag,
            enabled=enabled,
            device_id=device_id,
+            access_token_id=access_token_id,
        )
        pusher = await self.process_pusher_change_by_id(app_id, pushkey, user_id)

@@ -199,7 +203,7 @@ class PusherPool:
                )
                await self.remove_pusher(p.app_id, p.pushkey, p.user_name)

-    async def remove_pushers_by_access_token(
+    async def remove_pushers_by_access_tokens(
        self, user_id: str, access_tokens: Iterable[int]
    ) -> None:
        """Remove the pushers for a given user corresponding to a set of
@@ -209,6 +213,8 @@ class PusherPool:
            user_id: user to remove pushers for
            access_tokens: access token *ids* to remove pushers for
        """
+        # XXX(quenting): This is only needed until the "set_device_id_for_pushers"
+        # background update finishes
        tokens = set(access_tokens)
        for p in await self.store.get_pushers_by_user_id(user_id):
            if p.access_token in tokens:
@@ -220,6 +226,26 @@ class PusherPool:
                )
                await self.remove_pusher(p.app_id, p.pushkey, p.user_name)

+    async def remove_pushers_by_devices(
+        self, user_id: str, devices: StrCollection
+    ) -> None:
+        """Remove the pushers for a given user corresponding to a set of devices
+
+        Args:
+            user_id: user to remove pushers for
+            devices: device IDs to remove pushers for
+        """
+        device_ids = set(devices)
+        for p in await self.store.get_pushers_by_user_id(user_id):
+            if p.device_id in device_ids:
+                logger.info(
+                    "Removing pusher for app id %s, pushkey %s, user %s",
+                    p.app_id,
+                    p.pushkey,
+                    p.user_name,
+                )
+                await self.remove_pusher(p.app_id, p.pushkey, p.user_name)
+
    def on_new_notifications(self, max_token: RoomStreamToken) -> None:
        if not self.pushers:
            # nothing to do here.
@@ -345,7 +345,7 @@ class ReplicationEndpoint(metaclass=abc.ABCMeta):
                _outgoing_request_counter.labels(cls.NAME, 200).inc()

                # Wait on any streams that the remote may have written to.
-                for stream_name, position in result.get(
+                for stream_name, position in result.pop(
                    _STREAM_POSITION_KEY, {}
                ).items():
                    await replication.wait_for_stream_position(
@@ -138,8 +138,7 @@ class ClientRestResource(JsonResource):
            capabilities.register_servlets(hs, client_resource)
            account_validity.register_servlets(hs, client_resource)
        relations.register_servlets(hs, client_resource)
-        if is_main_process:
-            password_policy.register_servlets(hs, client_resource)
+        password_policy.register_servlets(hs, client_resource)
        knock.register_servlets(hs, client_resource)
        appservice_ping.register_servlets(hs, client_resource)

@@ -425,7 +425,6 @@ class UserRestServletV2(RestServlet):
                    ):
                        await self.pusher_pool.add_or_update_pusher(
                            user_id=user_id,
-                            access_token=None,
                            kind="email",
                            app_id="m.email",
                            app_display_name="Email Notifications",
@@ -43,19 +43,22 @@ def client_patterns(
    Returns:
        An iterable of patterns.
    """
-    patterns = []
+    versions = []

-    if unstable:
-        unstable_prefix = CLIENT_API_PREFIX + "/unstable"
-        patterns.append(re.compile("^" + unstable_prefix + path_regex))
    if v1:
-        v1_prefix = CLIENT_API_PREFIX + "/api/v1"
-        patterns.append(re.compile("^" + v1_prefix + path_regex))
-    for release in releases:
-        new_prefix = CLIENT_API_PREFIX + f"/{release}"
-        patterns.append(re.compile("^" + new_prefix + path_regex))
+        versions.append("api/v1")
+    versions.extend(releases)
+    if unstable:
+        versions.append("unstable")

-    return patterns
+    if len(versions) == 1:
+        versions_str = versions[0]
+    elif len(versions) > 1:
+        versions_str = "(" + "|".join(versions) + ")"
+    else:
+        raise RuntimeError("Must have at least one version for a URL")
+
+    return [re.compile("^" + CLIENT_API_PREFIX + "/" + versions_str + path_regex)]


 def set_timeline_upper_limit(filter_json: JsonDict, filter_timeline_limit: int) -> None:
@@ -576,6 +576,9 @@ class AddThreepidMsisdnSubmitTokenServlet(RestServlet):

 class ThreepidRestServlet(RestServlet):
    PATTERNS = client_patterns("/account/3pid$")
+    # This is used as a proxy for all the 3pid endpoints.
+
+    CATEGORY = "Client API requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -834,6 +837,7 @@ def assert_valid_next_link(hs: "HomeServer", next_link: str) -> None:

 class WhoamiRestServlet(RestServlet):
    PATTERNS = client_patterns("/account/whoami$")
+    CATEGORY = "Client API requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -38,6 +38,7 @@ class AccountDataServlet(RestServlet):
    PATTERNS = client_patterns(
        "/user/(?P<user_id>[^/]*)/account_data/(?P<account_data_type>[^/]*)"
    )
+    CATEGORY = "Account data requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -136,6 +137,7 @@ class RoomAccountDataServlet(RestServlet):
        "/rooms/(?P<room_id>[^/]*)"
        "/account_data/(?P<account_data_type>[^/]*)"
    )
+    CATEGORY = "Account data requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -40,6 +40,7 @@ logger = logging.getLogger(__name__)

 class DevicesRestServlet(RestServlet):
    PATTERNS = client_patterns("/devices$")
+    CATEGORY = "Client API requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -123,6 +124,7 @@ class DeleteDevicesRestServlet(RestServlet):

 class DeviceRestServlet(RestServlet):
    PATTERNS = client_patterns("/devices/(?P<device_id>[^/]*)$")
+    CATEGORY = "Client API requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -33,6 +33,7 @@ logger = logging.getLogger(__name__)

 class EventStreamRestServlet(RestServlet):
    PATTERNS = client_patterns("/events$", v1=True)
+    CATEGORY = "Sync requests"

    DEFAULT_LONGPOLL_TIME_MS = 30000

@@ -76,6 +77,7 @@ class EventStreamRestServlet(RestServlet):

 class EventRestServlet(RestServlet):
    PATTERNS = client_patterns("/events/(?P<event_id>[^/]*)$", v1=True)
+    CATEGORY = "Client API requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -31,6 +31,7 @@ logger = logging.getLogger(__name__)

 class GetFilterRestServlet(RestServlet):
    PATTERNS = client_patterns("/user/(?P<user_id>[^/]*)/filter/(?P<filter_id>[^/]*)")
+    CATEGORY = "Encryption requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -69,6 +70,7 @@ class GetFilterRestServlet(RestServlet):

 class CreateFilterRestServlet(RestServlet):
    PATTERNS = client_patterns("/user/(?P<user_id>[^/]*)/filter")
+    CATEGORY = "Encryption requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -28,6 +28,7 @@ if TYPE_CHECKING:
 # TODO: Needs unit testing
 class InitialSyncRestServlet(RestServlet):
    PATTERNS = client_patterns("/initialSync$", v1=True)
+    CATEGORY = "Sync requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -89,6 +89,7 @@ class KeyUploadServlet(RestServlet):
    """

    PATTERNS = client_patterns("/keys/upload(/(?P<device_id>[^/]+))?$")
+    CATEGORY = "Encryption requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -182,6 +183,7 @@ class KeyQueryServlet(RestServlet):
    """

    PATTERNS = client_patterns("/keys/query$")
+    CATEGORY = "Encryption requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -225,6 +227,7 @@ class KeyChangesServlet(RestServlet):
    """

    PATTERNS = client_patterns("/keys/changes$")
+    CATEGORY = "Encryption requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -274,6 +277,7 @@ class OneTimeKeyServlet(RestServlet):
    """

    PATTERNS = client_patterns("/keys/claim$")
+    CATEGORY = "Encryption requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -40,6 +40,7 @@ class KnockRoomAliasServlet(RestServlet):
    """

    PATTERNS = client_patterns("/knock/(?P<room_identifier>[^/]*)")
+    CATEGORY = "Event sending requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -72,6 +72,8 @@ class LoginResponse(TypedDict, total=False):

 class LoginRestServlet(RestServlet):
    PATTERNS = client_patterns("/login$", v1=True)
+    CATEGORY = "Registration/login requests"
+
    CAS_TYPE = "m.login.cas"
    SSO_TYPE = "m.login.sso"
    TOKEN_TYPE = "m.login.token"
@@ -537,6 +539,7 @@ def _get_auth_flow_dict_for_idp(idp: SsoIdentityProvider) -> JsonDict:

 class RefreshTokenServlet(RestServlet):
    PATTERNS = client_patterns("/refresh$")
+    CATEGORY = "Registration/login requests"

    def __init__(self, hs: "HomeServer"):
        self._auth_handler = hs.get_auth_handler()
@@ -590,6 +593,7 @@ class SsoRedirectServlet(RestServlet):
            + "/(r0|v3)/login/sso/redirect/(?P<idp_id>[A-Za-z0-9_.~-]+)$"
        )
    ]
+    CATEGORY = "SSO requests needed for all SSO providers"

    def __init__(self, hs: "HomeServer"):
        # make sure that the relevant handlers are instantiated, so that they
@@ -31,6 +31,7 @@ logger = logging.getLogger(__name__)

 class PasswordPolicyServlet(RestServlet):
    PATTERNS = client_patterns("/password_policy$")
+    CATEGORY = "Registration/login requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -33,6 +33,7 @@ logger = logging.getLogger(__name__)

 class PresenceStatusRestServlet(RestServlet):
    PATTERNS = client_patterns("/presence/(?P<user_id>[^/]*)/status", v1=True)
+    CATEGORY = "Presence requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -29,6 +29,7 @@ if TYPE_CHECKING:

 class ProfileDisplaynameRestServlet(RestServlet):
    PATTERNS = client_patterns("/profile/(?P<user_id>[^/]*)/displayname", v1=True)
+    CATEGORY = "Event sending requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -86,6 +87,7 @@ class ProfileDisplaynameRestServlet(RestServlet):

 class ProfileAvatarURLRestServlet(RestServlet):
    PATTERNS = client_patterns("/profile/(?P<user_id>[^/]*)/avatar_url", v1=True)
+    CATEGORY = "Event sending requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -142,6 +144,7 @@ class ProfileAvatarURLRestServlet(RestServlet):

 class ProfileRestServlet(RestServlet):
    PATTERNS = client_patterns("/profile/(?P<user_id>[^/]*)", v1=True)
+    CATEGORY = "Event sending requests"

    def __init__(self, hs: "HomeServer"):
        super().__init__()
@@ -44,6 +44,9 @@ class PushRuleRestServlet(RestServlet):
        "Unrecognised request: You probably wanted a trailing slash"
    )

+    WORKERS_DENIED_METHODS = ["PUT", "DELETE"]
+    CATEGORY = "Push rule requests"
+
    def __init__(self, hs: "HomeServer"):
        super().__init__()
        self.auth = hs.get_auth()
@@ -126,7 +126,6 @@ class PushersSetRestServlet(RestServlet):
        try:
            await self.pusher_pool.add_or_update_pusher(
                user_id=user.to_string(),
-                access_token=requester.access_token_id,
                kind=content["kind"],
                app_id=content["app_id"],
                app_display_name=content["app_display_name"],
--- a/Show More
+++ b/Show More
				`@@ -0,0 +1 @@`
				`Add denormalised event stream ordering column to membership state tables for future use. Contributed by Nick @ Beeper (@fizzadar).`
				`@@ -0,0 +1 @@`
				`Prune user's old devices on login if they have too many.`
				`@@ -0,0 +1 @@`
				`Add a primitive helper script for listing worker endpoints.`
				`@@ -0,0 +1 @@`
				`Add developer documentation for the Federation Sender and add a documentation mechanism using Sphinx.`
				`@@ -0,0 +1 @@`
				Make the pushers rely on the `device_id` instead of the `access_token_id` for various operations.
				`@@ -0,0 +1 @@`
				Fix a long-standing bug where edits of non-`m.room.message` events would not be correctly bundled.
				`@@ -0,0 +1 @@`
				`Fix a bug introduced in Synapse v1.55.0 which could delay remote homeservers being able to decrypt encrypted messages sent by local users.`
				`@@ -1 +0,0 @@`
				Fix a bug in which the [`POST /_matrix/client/v3/rooms/{roomId}/report/{eventId}`](https://spec.matrix.org/v1.6/client-server-api/#post_matrixclientv3roomsroomidreporteventid) endpoint would return the wrong error if the user did not have permission to view the event. This aligns Synapse's implementation with [MSC2249](https://github.com/matrix-org/matrix-spec-proposals/pull/2249).
				`@@ -0,0 +1 @@`
				`Allow running the Twisted trunk job against other branches.`
				`@@ -0,0 +1 @@`
				`Remind the releaser to ask for changelog feedback in [#synapse-dev](https://matrix.to/#/#synapse-dev:matrix.org).`