Same changelog as #15773

Fix TLS when using the ProxyAgent
Same fix as https://github.com/matrix-org/synapse/pull/15746 Thanks to @realtyem for pointing it out! `creatorForNetloc(...)` doesn't come with typing and expects `host` to be `bytes` instead of a `str`. `ProxyAgent` was introduced with the federation outbound proxy: https://github.com/matrix-org/synapse/pull/15773
2023-07-06 09:55:28 -05:00 · 2023-07-06 09:51:06 -05:00 · 2023-07-05 18:53:55 -05:00 · 2023-07-05 18:45:42 -05:00 · 2023-07-05 17:25:00 +01:00 · 2023-07-05 10:22:21 -05:00
225 changed files with 9320 additions and 6349 deletions
@@ -29,11 +29,12 @@ IS_PR = os.environ["GITHUB_REF"].startswith("refs/pull/")

 # First calculate the various trial jobs.
 #
-# For each type of test we only run on Py3.7 on PRs
+# For PRs, we only run each type of test with the oldest Python version supported (which
+# is Python 3.8 right now)

 trial_sqlite_tests = [
    {
-        "python-version": "3.7",
+        "python-version": "3.8",
        "database": "sqlite",
        "extras": "all",
    }
@@ -46,13 +47,13 @@ if not IS_PR:
            "database": "sqlite",
            "extras": "all",
        }
-        for version in ("3.8", "3.9", "3.10", "3.11")
+        for version in ("3.9", "3.10", "3.11")
    )


 trial_postgres_tests = [
    {
-        "python-version": "3.7",
+        "python-version": "3.8",
        "database": "postgres",
        "postgres-version": "11",
        "extras": "all",
@@ -71,7 +72,7 @@ if not IS_PR:

 trial_no_extra_tests = [
    {
-        "python-version": "3.7",
+        "python-version": "3.8",
        "database": "sqlite",
        "extras": "",
    }
@@ -22,7 +22,21 @@ concurrency:
  cancel-in-progress: true

 jobs:
+  check_repo:
+    # Prevent this workflow from running on any fork of Synapse other than matrix-org/synapse, as it is
+    # only useful to the Synapse core team.
+    # All other workflow steps depend on this one, thus if 'should_run_workflow' is not 'true', the rest
+    # of the workflow will be skipped as well.
+    runs-on: ubuntu-latest
+    outputs:
+      should_run_workflow: ${{ steps.check_condition.outputs.should_run_workflow }}
+    steps:
+      - id: check_condition
+        run: echo "should_run_workflow=${{ github.repository == 'matrix-org/synapse' }}" >> "$GITHUB_OUTPUT"
+
  mypy:
+    needs: check_repo
+    if: needs.check_repo.outputs.should_run_workflow == 'true'
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
@@ -47,6 +61,8 @@ jobs:
        run: sed '/warn_unused_ignores = True/d' -i mypy.ini
      - run: poetry run mypy
  trial:
+    needs: check_repo
+    if: needs.check_repo.outputs.should_run_workflow == 'true'
    runs-on: ubuntu-latest
    strategy:
      matrix:
@@ -105,6 +121,8 @@ jobs:


  sytest:
+    needs: check_repo
+    if: needs.check_repo.outputs.should_run_workflow == 'true'
    runs-on: ubuntu-latest
    container:
      image: matrixdotorg/sytest-synapse:testing
@@ -156,7 +174,8 @@ jobs:


  complement:
-    if: "${{ !failure() && !cancelled() }}"
+    needs: check_repo
+    if: "!failure() && !cancelled() && needs.check_repo.outputs.should_run_workflow == 'true'"
    runs-on: ubuntu-latest

    strategy:
@@ -192,7 +211,7 @@ jobs:
  # Open an issue if the build fails, so we know about it.
  # Only do this if we're not experimenting with this action in a PR.
  open-issue:
-    if: "failure() && github.event_name != 'push' && github.event_name != 'pull_request'"
+    if: "failure() && github.event_name != 'push' && github.event_name != 'pull_request' && needs.check_repo.outputs.should_run_workflow == 'true'"
    needs:
      # TODO: should mypy be included here? It feels more brittle than the others.
      - mypy
@@ -34,6 +34,7 @@ jobs:
      - id: set-distros
        run: |
          # if we're running from a tag, get the full list of distros; otherwise just use debian:sid
+          # NOTE: inside the actual Dockerfile-dhvirtualenv, the image name is expanded into its full image path
          dists='["debian:sid"]'
          if [[ $GITHUB_REF == refs/tags/* ]]; then
              dists=$(scripts-dev/build_debian_packages.py --show-dists-json)
@@ -143,7 +144,7 @@ jobs:

      - name: Only build a single wheel on PR
        if: startsWith(github.ref, 'refs/pull/')
-        run: echo "CIBW_BUILD="cp37-manylinux_${{ matrix.arch }}"" >> $GITHUB_ENV
+        run: echo "CIBW_BUILD="cp38-manylinux_${{ matrix.arch }}"" >> $GITHUB_ENV

      - name: Build wheels
        run: python -m cibuildwheel --output-dir wheelhouse
@@ -35,7 +35,7 @@ jobs:
    steps:
      - uses: actions/checkout@v3
      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2
      - uses: matrix-org/setup-python-poetry@v1
        with:
@@ -92,6 +92,10 @@ jobs:
      - name: Checkout repository
        uses: actions/checkout@v3

+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@1.60.0
+      - uses: Swatinem/rust-cache@v2
+
      - name: Setup Poetry
        uses: matrix-org/setup-python-poetry@v1
        with:
@@ -103,10 +107,6 @@ jobs:
          # To make CI green, err towards caution and install the project.
          install-project: "true"

-      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
-      - uses: Swatinem/rust-cache@v2
-
      # Cribbed from
      # https://github.com/AustinScola/mypy-cache-github-action/blob/85ea4f2972abed39b33bd02c36e341b28ca59213/src/restore.ts#L10-L17
      - name: Restore/persist mypy's cache
@@ -150,7 +150,7 @@ jobs:
        with:
          ref: ${{ github.event.pull_request.head.sha }}
      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2
      - uses: matrix-org/setup-python-poetry@v1
        with:
@@ -167,7 +167,7 @@ jobs:
      - uses: actions/checkout@v3

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

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2

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

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2

      # There aren't wheels for some of the older deps, so we need to install
@@ -320,7 +320,7 @@ jobs:

      - uses: actions/setup-python@v4
        with:
-          python-version: '3.7'
+          python-version: '3.8'

      - name: Prepare old deps
        if: steps.cache-poetry-old-deps.outputs.cache-hit != 'true'
@@ -362,7 +362,7 @@ jobs:
    runs-on: ubuntu-latest
    strategy:
      matrix:
-        python-version: ["pypy-3.7"]
+        python-version: ["pypy-3.8"]
        extras: ["all"]

    steps:
@@ -399,8 +399,8 @@ jobs:
      env:
        SYTEST_BRANCH: ${{ github.head_ref }}
        POSTGRES: ${{ matrix.job.postgres && 1}}
-        MULTI_POSTGRES: ${{ (matrix.job.postgres == 'multi-postgres') && 1}}
-        ASYNCIO_REACTOR: ${{ (matrix.job.reactor == 'asyncio') && 1 }}
+        MULTI_POSTGRES: ${{ (matrix.job.postgres == 'multi-postgres') || '' }}
+        ASYNCIO_REACTOR: ${{ (matrix.job.reactor == 'asyncio') || '' }}
        WORKERS: ${{ matrix.job.workers && 1 }}
        BLACKLIST: ${{ matrix.job.workers && 'synapse-blacklist-with-workers' }}
        TOP: ${{ github.workspace }}
@@ -416,7 +416,7 @@ jobs:
        run: cat sytest-blacklist .ci/worker-blacklist > synapse-blacklist-with-workers

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2

      - name: Run SyTest
@@ -477,7 +477,7 @@ jobs:
    strategy:
      matrix:
        include:
-          - python-version: "3.7"
+          - python-version: "3.8"
            postgres-version: "11"

          - python-version: "3.11"
@@ -556,7 +556,7 @@ jobs:
          path: synapse

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2

      - uses: actions/setup-go@v4
@@ -584,7 +584,7 @@ jobs:
      - uses: actions/checkout@v3

      - name: Install Rust
-        uses: dtolnay/rust-toolchain@1.58.1
+        uses: dtolnay/rust-toolchain@1.60.0
      - uses: Swatinem/rust-cache@v2

      - run: cargo test
@@ -18,7 +18,22 @@ concurrency:
  cancel-in-progress: true

 jobs:
+  check_repo:
+    # Prevent this workflow from running on any fork of Synapse other than matrix-org/synapse, as it is
+    # only useful to the Synapse core team.
+    # All other workflow steps depend on this one, thus if 'should_run_workflow' is not 'true', the rest
+    # of the workflow will be skipped as well.
+    if: github.repository == 'matrix-org/synapse'
+    runs-on: ubuntu-latest
+    outputs:
+      should_run_workflow: ${{ steps.check_condition.outputs.should_run_workflow }}
+    steps:
+      - id: check_condition
+        run: echo "should_run_workflow=${{ github.repository == 'matrix-org/synapse' }}" >> "$GITHUB_OUTPUT"
+
  mypy:
+    needs: check_repo
+    if: needs.check_repo.outputs.should_run_workflow == 'true'
    runs-on: ubuntu-latest

    steps:
@@ -41,6 +56,8 @@ jobs:
      - run: poetry run mypy

  trial:
+    needs: check_repo
+    if: needs.check_repo.outputs.should_run_workflow == 'true'
    runs-on: ubuntu-latest

    steps:
@@ -75,6 +92,8 @@ jobs:
          || true

  sytest:
+    needs: check_repo
+    if: needs.check_repo.outputs.should_run_workflow == 'true'
    runs-on: ubuntu-latest
    container:
      image: matrixdotorg/sytest-synapse:buster
@@ -119,7 +138,8 @@ jobs:
            /logs/**/*.log*

  complement:
-    if: "${{ !failure() && !cancelled() }}"
+    needs: check_repo
+    if: "!failure() && !cancelled() && needs.check_repo.outputs.should_run_workflow == 'true'"
    runs-on: ubuntu-latest

    strategy:
@@ -166,7 +186,7 @@ jobs:

  # open an issue if the build fails, so we know about it.
  open-issue:
-    if: failure()
+    if: failure() && needs.check_repo.outputs.should_run_workflow == 'true'
    needs:
      - mypy
      - trial
@@ -34,6 +34,7 @@ __pycache__/
 /logs
 /media_store/
 /uploads
+/homeserver-config-overrides.d

 # For direnv users
 /.envrc
@@ -4,9 +4,9 @@ version = 3

 [[package]]
 name = "aho-corasick"
-version = "0.7.19"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4f55bd91a0978cbfd91c457a164bab8b4001c833b7f323132c0a4e1922dd44e"
+checksum = "43f6cb1bf222025340178f382c426f13757b2960e89779dfcb319c32542a5a41"
 dependencies = [
 "memchr",
 ]
@@ -132,9 +132,9 @@ dependencies = [

 [[package]]
 name = "log"
-version = "0.4.18"
+version = "0.4.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "518ef76f2f87365916b142844c16d8fefd85039bc5699050210a7778ee1cd1de"
+checksum = "b06a4cde4c0f271a446782e3eff8de789548ce57dbc8eca9292c27f4a42004b4"

 [[package]]
 name = "memchr"
@@ -229,9 +229,9 @@ dependencies = [

 [[package]]
 name = "pyo3-log"
-version = "0.8.1"
+version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9c8b57fe71fb5dcf38970ebedc2b1531cf1c14b1b9b4c560a182a57e115575c"
+checksum = "c94ff6535a6bae58d7d0b85e60d4c53f7f84d0d0aa35d6a28c3f3e70bfe51444"
 dependencies = [
 "arc-swap",
 "log",
@@ -291,9 +291,9 @@ dependencies = [

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

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

 [[package]]
 name = "ryu"
@@ -320,18 +320,18 @@ checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"

 [[package]]
 name = "serde"
-version = "1.0.163"
+version = "1.0.164"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2113ab51b87a539ae008b5c6c02dc020ffa39afd2d83cffcb3f4eb2722cebec2"
+checksum = "9e8c8cf938e98f769bc164923b06dce91cea1751522f46f8466461af04c9027d"
 dependencies = [
 "serde_derive",
 ]

 [[package]]
 name = "serde_derive"
-version = "1.0.163"
+version = "1.0.164"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8c805777e3930c8883389c602315a24224bcc738b63905ef87cd1420353ea93e"
+checksum = "d9735b638ccc51c28bf6914d90a2e9725b377144fc612c49a611fddd1b631d68"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -340,9 +340,9 @@ dependencies = [

 [[package]]
 name = "serde_json"
-version = "1.0.96"
+version = "1.0.99"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "057d394a50403bcac12672b2b18fb387ab6d289d957dab67dd201875391e52f1"
+checksum = "46266871c240a00b8f503b877622fe33430b3c7d963bdc0f2adc511e54a1eae3"
 dependencies = [
 "itoa",
 "ryu",
@@ -1 +0,0 @@
-Experimental [MSC3861](https://github.com/matrix-org/matrix-spec-proposals/pull/3861) support: delegate auth to an OIDC provider.
@@ -1 +0,0 @@
-Remove some unused code.
@@ -0,0 +1 @@
+Add foreign key constraint to `event_forward_extremities`.
@@ -0,0 +1 @@
+Allow configuring the set of workers to proxy outbound federation traffic through via `outbound_federation_restricted_to`.
@@ -0,0 +1 @@
+Add read/write style cross-worker locks.
@@ -0,0 +1 @@
+Use lower isolation level when cleaning old presence stream data to avoid serialization errors.
@@ -0,0 +1 @@
+Add `not_user_type` param to the list accounts admin API.
@@ -0,0 +1 @@
+Add tracing to media `/upload` code paths.
@@ -0,0 +1 @@
+Remove support for Python 3.7.
@@ -0,0 +1 @@
+Fixed header levels on the Admin API "Users" documentation page. Contributed by @sumnerevans at @beeper.
@@ -0,0 +1 @@
+Add a timeout that aborts any Postgres statement taking more than 1 hour.
@@ -0,0 +1 @@
+Fix the `devenv up` configuration which was ignoring the config overrides.
@@ -0,0 +1 @@
+Optimised cleanup of old entries in device_lists_stream.
@@ -0,0 +1,3 @@
+Pin `pydantic` to ^=1.7.4 to avoid backwards-incompatible API changes from the 2.0.0 release.
+Resolves https://github.com/matrix-org/synapse/issues/15858.
+Contributed by @PaarthShah.
@@ -0,0 +1 @@
+Correctly resize thumbnails with pillow version >=10.
@@ -0,0 +1 @@
+Allow configuring the set of workers to proxy outbound federation traffic through via `outbound_federation_restricted_to`.
@@ -29,7 +29,7 @@
        "level": "error"
      },
      {
-        "line": "my-matrix-server-federation-sender-1 | 2023-01-25 20:56:20,995 - synapse.http.matrixfederationclient - 709 - WARNING - federation_transaction_transmission_loop-3 - {PUT-O-3} [example.com] Request failed: PUT matrix://example.com/_matrix/federation/v1/send/1674680155797: HttpResponseException('403: Forbidden')",
+        "line": "my-matrix-server-federation-sender-1 | 2023-01-25 20:56:20,995 - synapse.http.matrixfederationclient - 709 - WARNING - federation_transaction_transmission_loop-3 - {PUT-O-3} [example.com] Request failed: PUT matrix-federation://example.com/_matrix/federation/v1/send/1674680155797: HttpResponseException('403: Forbidden')",
        "level": "warning"
      },
      {
@@ -1,3 +1,57 @@
+matrix-synapse-py3 (1.87.0) stable; urgency=medium
+
+  * New Synapse release 1.87.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 04 Jul 2023 16:24:00 +0100
+
+matrix-synapse-py3 (1.87.0~rc1) stable; urgency=medium
+
+  * New synapse release 1.87.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 27 Jun 2023 15:27:04 +0000
+
+matrix-synapse-py3 (1.86.0) stable; urgency=medium
+
+  * New Synapse release 1.86.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 20 Jun 2023 17:22:46 +0200
+
+matrix-synapse-py3 (1.86.0~rc2) stable; urgency=medium
+
+  * New Synapse release 1.86.0rc2.
+
+ -- Synapse Packaging team <packages@matrix.org>  Wed, 14 Jun 2023 12:16:27 +0200
+
+matrix-synapse-py3 (1.86.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.86.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 13 Jun 2023 14:30:45 +0200
+
+matrix-synapse-py3 (1.85.2) stable; urgency=medium
+
+  * New Synapse release 1.85.2.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 08 Jun 2023 13:04:18 +0100
+
+matrix-synapse-py3 (1.85.1) stable; urgency=medium
+
+  * New Synapse release 1.85.1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Wed, 07 Jun 2023 10:51:12 +0100
+
+matrix-synapse-py3 (1.85.0) stable; urgency=medium
+
+  * New Synapse release 1.85.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 06 Jun 2023 09:39:29 +0100
+
+matrix-synapse-py3 (1.85.0~rc2) stable; urgency=medium
+
+  * New Synapse release 1.85.0rc2.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 01 Jun 2023 09:16:18 -0700
+
 matrix-synapse-py3 (1.85.0~rc1) stable; urgency=medium

  * New Synapse release 1.85.0rc1.
@@ -27,7 +27,7 @@ ARG PYTHON_VERSION=3.11
 ###
 # We hardcode the use of Debian bullseye here because this could change upstream
 # and other Dockerfiles used for testing are expecting bullseye.
-FROM docker.io/python:${PYTHON_VERSION}-slim-bullseye as requirements
+FROM docker.io/library/python:${PYTHON_VERSION}-slim-bullseye as requirements

 # RUN --mount is specific to buildkit and is documented at
 # https://github.com/moby/buildkit/blob/master/frontend/dockerfile/docs/syntax.md#build-mounts-run---mount.
@@ -87,7 +87,7 @@ RUN if [ -z "$TEST_ONLY_IGNORE_POETRY_LOCKFILE" ]; then \
 ###
 ### Stage 1: builder
 ###
-FROM docker.io/python:${PYTHON_VERSION}-slim-bullseye as builder
+FROM docker.io/library/python:${PYTHON_VERSION}-slim-bullseye as builder

 # install the OS build deps
 RUN \
@@ -158,7 +158,7 @@ RUN --mount=type=cache,target=/synapse/target,sharing=locked \
 ### Stage 2: runtime
 ###

-FROM docker.io/python:${PYTHON_VERSION}-slim-bullseye
+FROM docker.io/library/python:${PYTHON_VERSION}-slim-bullseye

 LABEL org.opencontainers.image.url='https://matrix.org/docs/projects/server/synapse'
 LABEL org.opencontainers.image.documentation='https://github.com/matrix-org/synapse/blob/master/docker/README.md'
@@ -24,16 +24,16 @@ ARG distro=""
 # https://launchpad.net/~jyrki-pulliainen/+archive/ubuntu/dh-virtualenv, but
 # it's not obviously easier to use that than to build our own.)

-FROM ${distro} as builder
+FROM docker.io/library/${distro} as builder

 RUN apt-get update -qq -o Acquire::Languages=none
 RUN env DEBIAN_FRONTEND=noninteractive apt-get install \
-        -yqq --no-install-recommends \
-        build-essential \
-        ca-certificates \
-        devscripts \
-        equivs \
-        wget
+    -yqq --no-install-recommends \
+    build-essential \
+    ca-certificates \
+    devscripts \
+    equivs \
+    wget

 # fetch and unpack the package
 # We are temporarily using a fork of dh-virtualenv due to an incompatibility with Python 3.11, which ships with
@@ -55,40 +55,36 @@ RUN cd /dh-virtualenv && DEB_BUILD_OPTIONS=nodoc dpkg-buildpackage -us -uc -b
 ###
 ### Stage 1
 ###
-FROM ${distro}
+FROM docker.io/library/${distro}

 # Get the distro we want to pull from as a dynamic build variable
 # (We need to define it in each build stage)
 ARG distro=""
 ENV distro ${distro}

-# Python < 3.7 assumes LANG="C" means ASCII-only and throws on printing unicode
-# http://bugs.python.org/issue19846
-ENV LANG C.UTF-8
-
 # Install the build dependencies
 #
 # NB: keep this list in sync with the list of build-deps in debian/control
 # TODO: it would be nice to do that automatically.
 RUN apt-get update -qq -o Acquire::Languages=none \
    && env DEBIAN_FRONTEND=noninteractive apt-get install \
-        -yqq --no-install-recommends -o Dpkg::Options::=--force-unsafe-io \
-        build-essential \
-        curl \
-        debhelper \
-        devscripts \
-        libsystemd-dev \
-        lsb-release \
-        pkg-config \
-        python3-dev \
-        python3-pip \
-        python3-setuptools \
-        python3-venv \
-        sqlite3 \
-        libpq-dev \
-        libicu-dev \
-        pkg-config \
-        xmlsec1
+    -yqq --no-install-recommends -o Dpkg::Options::=--force-unsafe-io \
+    build-essential \
+    curl \
+    debhelper \
+    devscripts \
+    libsystemd-dev \
+    lsb-release \
+    pkg-config \
+    python3-dev \
+    python3-pip \
+    python3-setuptools \
+    python3-venv \
+    sqlite3 \
+    libpq-dev \
+    libicu-dev \
+    pkg-config \
+    xmlsec1

 # Install rust and ensure it's in the PATH
 ENV RUSTUP_HOME=/rust
@@ -7,7 +7,7 @@ ARG FROM=matrixdotorg/synapse:$SYNAPSE_VERSION
 # target image. For repeated rebuilds, this is much faster than apt installing
 # each time.

-FROM debian:bullseye-slim AS deps_base
+FROM docker.io/library/debian:bullseye-slim AS deps_base
    RUN \
       --mount=type=cache,target=/var/cache/apt,sharing=locked \
       --mount=type=cache,target=/var/lib/apt,sharing=locked \
@@ -21,7 +21,7 @@ FROM debian:bullseye-slim AS deps_base
 # which makes it much easier to copy (but we need to make sure we use an image
 # based on the same debian version as the synapse image, to make sure we get
 # the expected version of libc.
-FROM redis:6-bullseye AS redis_base
+FROM docker.io/library/redis:7-bullseye AS redis_base

 # now build the final image, based on the the regular Synapse docker image
 FROM $FROM
@@ -73,7 +73,8 @@ The following environment variables are supported in `generate` mode:
  will log sensitive information such as access tokens.
  This should not be needed unless you are a developer attempting to debug something
  particularly tricky.
-
+* `SYNAPSE_LOG_TESTING`: if set, Synapse will log additional information useful
+  for testing.

 ## Postgres

@@ -7,6 +7,7 @@
 # https://github.com/matrix-org/synapse/blob/develop/docker/README-testing.md#testing-with-postgresql-and-single-or-multi-process-synapse

 ARG SYNAPSE_VERSION=latest
+# This is an intermediate image, to be built locally (not pulled from a registry).
 ARG FROM=matrixdotorg/synapse-workers:$SYNAPSE_VERSION

 FROM $FROM
@@ -19,8 +20,8 @@ FROM $FROM
    # the same debian version as Synapse's docker image (so the versions of the
    # shared libraries match).
    RUN adduser --system --uid 999 postgres --home /var/lib/postgresql
-    COPY --from=postgres:13-bullseye /usr/lib/postgresql /usr/lib/postgresql
-    COPY --from=postgres:13-bullseye /usr/share/postgresql /usr/share/postgresql
+    COPY --from=docker.io/library/postgres:13-bullseye /usr/lib/postgresql /usr/lib/postgresql
+    COPY --from=docker.io/library/postgres:13-bullseye /usr/share/postgresql /usr/share/postgresql
    RUN mkdir /var/run/postgresql && chown postgres /var/run/postgresql
    ENV PATH="${PATH}:/usr/lib/postgresql/13/bin"
    ENV PGDATA=/var/lib/postgresql/data
@@ -92,8 +92,6 @@ allow_device_name_lookup_over_federation: true
 ## Experimental Features ##

 experimental_features:
-  # Enable history backfilling support
-  msc2716_enabled: true
  # client-side support for partial state in /send_join responses
  faster_joins: true
  # Enable support for polls
@@ -49,17 +49,35 @@ handlers:
    class: logging.StreamHandler
    formatter: precise

-{% if not SYNAPSE_LOG_SENSITIVE %}
-{#
-  If SYNAPSE_LOG_SENSITIVE is unset, then override synapse.storage.SQL to INFO
-  so that DEBUG entries (containing sensitive information) are not emitted.
-#}
 loggers:
+    # This is just here so we can leave `loggers` in the config regardless of whether
+    # we configure other loggers below (avoid empty yaml dict error).
+    _placeholder:
+        level: "INFO"
+
+    {% if not SYNAPSE_LOG_SENSITIVE %}
+    {#
+      If SYNAPSE_LOG_SENSITIVE is unset, then override synapse.storage.SQL to INFO
+      so that DEBUG entries (containing sensitive information) are not emitted.
+    #}
    synapse.storage.SQL:
        # beware: increasing this to DEBUG will make synapse log sensitive
        # information such as access tokens.
        level: INFO
-{% endif %}
+    {% endif %}
+
+    {% if SYNAPSE_LOG_TESTING %}
+    {#
+      If Synapse is under test, log a few more useful things for a developer
+      attempting to debug something particularly tricky.
+
+      With `synapse.visibility.filtered_event_debug`, it logs when events are (maybe
+      unexpectedly) filtered out of responses in tests. It's just nice to be able to
+      look at the CI log and figure out why an event isn't being returned.
+    #}
+    synapse.visibility.filtered_event_debug:
+        level: DEBUG
+    {% endif %}

 root:
    level: {{ SYNAPSE_LOG_LEVEL or "INFO" }}
@@ -40,6 +40,8 @@
 #         log level. INFO is the default.
 #   * SYNAPSE_LOG_SENSITIVE: If unset, SQL and SQL values won't be logged,
 #         regardless of the SYNAPSE_LOG_LEVEL setting.
+#   * SYNAPSE_LOG_TESTING: if set, Synapse will log additional information useful
+#     for testing.
 #
 # NOTE: According to Complement's ENTRYPOINT expectations for a homeserver image (as defined
 # in the project's README), this script may be run multiple times, and functionality should
@@ -242,7 +244,6 @@ WORKERS_CONFIG: Dict[str, Dict[str, Any]] = {
            "^/_matrix/client/(api/v1|r0|v3|unstable)/join/",
            "^/_matrix/client/(api/v1|r0|v3|unstable)/knock/",
            "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/",
-            "^/_matrix/client/(v1|unstable/org.matrix.msc2716)/rooms/.*/batch_send",
        ],
        "shared_extra_conf": {},
        "worker_extra_conf": "",
@@ -947,6 +948,7 @@ def generate_worker_log_config(
    extra_log_template_args["SYNAPSE_LOG_SENSITIVE"] = environ.get(
        "SYNAPSE_LOG_SENSITIVE"
    )
+    extra_log_template_args["SYNAPSE_LOG_TESTING"] = environ.get("SYNAPSE_LOG_TESTING")

    # Render and write the file
    log_config_filepath = f"/conf/workers/{worker_name}.log.config"
@@ -10,7 +10,7 @@ ARG PYTHON_VERSION=3.9
 ###
 # We hardcode the use of Debian bullseye here because this could change upstream
 # and other Dockerfiles used for testing are expecting bullseye.
-FROM docker.io/python:${PYTHON_VERSION}-slim-bullseye
+FROM docker.io/library/python:${PYTHON_VERSION}-slim-bullseye

 # Install Rust and other dependencies (stolen from normal Dockerfile)
 # install the OS build deps
@@ -419,7 +419,7 @@ The following query parameters are available:

 * `from` (required) - The token to start returning events from. This token can be obtained from a prev_batch
  or next_batch token returned by the /sync endpoint, or from an end token returned by a previous request to this endpoint.
-* `to` - The token to spot returning events at.
+* `to` - The token to stop returning events at.
 * `limit` - The maximum number of events to return. Defaults to `10`.
 * `filter` - A JSON RoomEventFilter to filter returned events with.
 * `dir` - The direction to return events from. Either `f` for forwards or `b` for backwards. Setting
@@ -242,6 +242,9 @@ The following parameters should be set in the URL:

 - `dir` - Direction of media order. Either `f` for forwards or `b` for backwards.
  Setting this value to `b` will reverse the above sort order. Defaults to `f`.
+- `not_user_type` - Exclude certain user types, such as bot users, from the request.
+   Can be provided multiple times. Possible values are `bot`, `support` or "empty string".
+   "empty string" here means to exclude users without a type.

 Caution. The database only has indexes on the columns `name` and `creation_ts`.
 This means that if a different sort order is used (`is_guest`, `admin`,
@@ -1180,7 +1183,7 @@ The following parameters should be set in the URL:
 - `user_id` - The fully qualified MXID: for example, `@user:server.com`. The user must
  be local.

-### Check username availability
+## Check username availability

 Checks to see if a username is available, and valid, for the server. See [the client-server 
 API](https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-register-available)
@@ -1198,7 +1201,7 @@ GET /_synapse/admin/v1/username_available?username=$localpart
 The request and response format is the same as the
 [/_matrix/client/r0/register/available](https://matrix.org/docs/spec/client_server/r0.6.0#get-matrix-client-r0-register-available) API.

-### Find a user based on their ID in an auth provider
+## Find a user based on their ID in an auth provider

 The API is:

@@ -1237,7 +1240,7 @@ Returns a `404` HTTP status code if no user was found, with a response body like
 _Added in Synapse 1.68.0._


-### Find a user based on their Third Party ID (ThreePID or 3PID)
+## Find a user based on their Third Party ID (ThreePID or 3PID)

 The API is:

@@ -6,7 +6,7 @@ This is a work-in-progress set of notes with two goals:

 See also [MSC3902](https://github.com/matrix-org/matrix-spec-proposals/pull/3902).

-The key idea is described by [MSC706](https://github.com/matrix-org/matrix-spec-proposals/pull/3902). This allows servers to
+The key idea is described by [MSC3706](https://github.com/matrix-org/matrix-spec-proposals/pull/3706). This allows servers to
 request a lightweight response to the federation `/send_join` endpoint.
 This is called a **faster join**, also known as a **partial join**. In these
 notes we'll usually use the word "partial" as it matches the database schema.
@@ -348,6 +348,42 @@ callback returns `False`, Synapse falls through to the next one. The value of th
 callback that does not return `False` will be used. If this happens, Synapse will not call
 any of the subsequent implementations of this callback.

+
+### `check_login_for_spam`
+
+_First introduced in Synapse v1.87.0_
+
+```python
+async def check_login_for_spam(
+    user_id: str,
+    device_id: Optional[str],
+    initial_display_name: Optional[str],
+    request_info: Collection[Tuple[Optional[str], str]],
+    auth_provider_id: Optional[str] = None,
+) -> Union["synapse.module_api.NOT_SPAM", "synapse.module_api.errors.Codes"]
+```
+
+Called when a user logs in.
+
+The arguments passed to this callback are:
+
+* `user_id`: The user ID the user is logging in with
+* `device_id`: The device ID the user is re-logging into.
+* `initial_display_name`: The device display name, if any.
+* `request_info`: A collection of tuples, which first item is a user agent, and which
+  second item is an IP address. These user agents and IP addresses are the ones that were
+  used during the login process.
+* `auth_provider_id`: The identifier of the SSO authentication provider, if any.
+
+If multiple modules implement this callback, they will be considered in order. If a
+callback returns `synapse.module_api.NOT_SPAM`, Synapse falls through to the next one.
+The value of the first callback that does not return `synapse.module_api.NOT_SPAM` will
+be used. If this happens, Synapse will not call any of the subsequent implementations of
+this callback.
+
+*Note:* This will not be called when a user registers.
+
+
 ## Example

 The example below is a module that implements the spam checker callback
@@ -200,7 +200,7 @@ When following this route please make sure that the [Platform-specific prerequis
 System requirements:

 - POSIX-compliant system (tested on Linux & OS X)
- Python 3.7 or later, up to Python 3.11.
+- Python 3.8 or later, up to Python 3.11.
 - At least 1GB of free RAM if you want to join large public rooms like #matrix:matrix.org

 If building on an uncommon architecture for which pre-built wheels are
@@ -88,6 +88,37 @@ process, for example:
    dpkg -i matrix-synapse-py3_1.3.0+stretch1_amd64.deb
    ```

+# Upgrading to v1.88.0
+
+## Minimum supported Python version
+
+The minimum supported Python version has been increased from v3.7 to v3.8.
+You will need Python 3.8 to run Synapse v1.88.0 (due out July 18th, 2023).
+
+If you use current versions of the Matrix.org-distributed Debian
+packages or Docker images, no action is required.
+
+
+# Upgrading to v1.86.0
+
+## Minimum supported Rust version
+
+The minimum supported Rust version has been increased from v1.58.1 to v1.60.0.
+Users building from source will need to ensure their `rustc` version is up to
+date.
+
+
+# Upgrading to v1.85.0
+
+## Application service registration with "user" property deprecation
+
+Application services should ensure they call the `/register` endpoint with a
+`username` property. The legacy `user` property is considered deprecated and
+should no longer be included.
+
+A future version of Synapse (v1.88.0 or later) will remove support for legacy
+application service login.
+
 # Upgrading to v1.84.0

 ## Deprecation of `worker_replication_*` configuration settings
@@ -27,9 +27,8 @@ What servers are currently participating in this room?
 Run this sql query on your db:
 ```sql
 SELECT DISTINCT split_part(state_key, ':', 2)
-    FROM current_state_events AS c
-    INNER JOIN room_memberships AS m USING (room_id, event_id)
-    WHERE room_id = '!cURbafjkfsMDVwdRDQ:matrix.org' AND membership = 'join';
+FROM current_state_events
+WHERE room_id = '!cURbafjkfsMDVwdRDQ:matrix.org' AND membership = 'join';
 ```

 What users are registered on my server?
@@ -1196,6 +1196,32 @@ Example configuration:
 allow_device_name_lookup_over_federation: true
 ```
 ---
+### `federation`
+
+The federation section defines some sub-options related to federation.
+
+The following options are related to configuring timeout and retry logic for one request,
+independently of the others.
+Short retry algorithm is used when something or someone will wait for the request to have an
+answer, while long retry is used for requests that happen in the background,
+like sending a federation transaction.
+
+* `client_timeout`: timeout for the federation requests. Default to 60s.
+* `max_short_retry_delay`: maximum delay to be used for the short retry algo. Default to 2s.
+* `max_long_retry_delay`: maximum delay to be used for the short retry algo. Default to 60s.
+* `max_short_retries`: maximum number of retries for the short retry algo. Default to 3 attempts.
+* `max_long_retries`: maximum number of retries for the long retry algo. Default to 10 attempts.
+
+Example configuration:
+```yaml
+federation:
+  client_timeout: 180s
+  max_short_retry_delay: 7s
+  max_long_retry_delay: 100s
+  max_short_retries: 5
+  max_long_retries: 20
+```
+---
 ## Caching

 Options related to caching.
@@ -2570,7 +2596,50 @@ Example configuration:
 ```yaml
 nonrefreshable_access_token_lifetime: 24h
 ```
+---
+### `ui_auth`

+The amount of time to allow a user-interactive authentication session to be active.
+
+This defaults to 0, meaning the user is queried for their credentials
+before every action, but this can be overridden to allow a single
+validation to be re-used.  This weakens the protections afforded by
+the user-interactive authentication process, by allowing for multiple
+(and potentially different) operations to use the same validation session.
+
+This is ignored for potentially "dangerous" operations (including
+deactivating an account, modifying an account password, adding a 3PID,
+and minting additional login tokens).
+
+Use the `session_timeout` sub-option here to change the time allowed for credential validation.
+
+Example configuration:
+```yaml
+ui_auth:
+    session_timeout: "15s"
+```
+---
+### `login_via_existing_session`
+
+Matrix supports the ability of an existing session to mint a login token for
+another client.
+
+Synapse disables this by default as it has security ramifications -- a malicious
+client could use the mechanism to spawn more than one session.
+
+The duration of time the generated token is valid for can be configured with the
+`token_timeout` sub-option.
+
+User-interactive authentication is required when this is enabled unless the
+`require_ui_auth` sub-option is set to `False`.
+
+Example configuration:
+```yaml
+login_via_existing_session:
+    enabled: true
+    require_ui_auth: false
+    token_timeout: "5m"
+```
 ---
 ## Metrics
 Config options related to metrics.
@@ -3415,28 +3484,6 @@ password_config:
      require_uppercase: true
 ```
 ---
-### `ui_auth`
-
-The amount of time to allow a user-interactive authentication session to be active.
-
-This defaults to 0, meaning the user is queried for their credentials
-before every action, but this can be overridden to allow a single
-validation to be re-used.  This weakens the protections afforded by
-the user-interactive authentication process, by allowing for multiple
-(and potentially different) operations to use the same validation session.
-
-This is ignored for potentially "dangerous" operations (including
-deactivating an account, modifying an account password, and
-adding a 3PID).
-
-Use the `session_timeout` sub-option here to change the time allowed for credential validation.
-
-Example configuration:
-```yaml
-ui_auth:
-    session_timeout: "15s"
-```
---
 ## Push
 Configuration settings related to push notifications

@@ -3883,13 +3930,14 @@ federation_sender_instances:
 ---
 ### `instance_map`

-When using workers this should be a map from [`worker_name`](#worker_name) to the
-HTTP replication listener of the worker, if configured, and to the main process.
-Each worker declared under [`stream_writers`](../../workers.md#stream-writers) needs
-a HTTP replication listener, and that listener should be included in the `instance_map`.
-The main process also needs an entry on the `instance_map`, and it should be listed under
-`main` **if even one other worker exists**. Ensure the port matches with what is declared 
-inside the `listener` block for a `replication` listener.
+When using workers this should be a map from [`worker_name`](#worker_name) to the HTTP
+replication listener of the worker, if configured, and to the main process. Each worker
+declared under [`stream_writers`](../../workers.md#stream-writers) and
+[`outbound_federation_restricted_to`](#outbound_federation_restricted_to) needs a HTTP replication listener, and that
+listener should be included in the `instance_map`. The main process also needs an entry
+on the `instance_map`, and it should be listed under `main` **if even one other worker
+exists**. Ensure the port matches with what is declared inside the `listener` block for
+a `replication` listener.


 Example configuration:
@@ -3919,6 +3967,22 @@ stream_writers:
  typing: worker1
 ```
 ---
+### `outbound_federation_restricted_to`
+
+When using workers, you can restrict outbound federation traffic to only go through a
+specific subset of workers. Any worker specified here must also be in the
+[`instance_map`](#instance_map).
+
+```yaml
+outbound_federation_restricted_to:
+  - federation_sender1
+  - federation_sender2
+```
+
+Also see the [worker
+documentation](../../workers.md#restrict-outbound-federation-traffic-to-a-specific-set-of-workers)
+for more info.
+---
 ### `run_background_tasks_on`

 The [worker](../../workers.md#background-tasks) that is used to run
@@ -232,7 +232,6 @@ information.
    ^/_matrix/client/v1/rooms/.*/hierarchy$
    ^/_matrix/client/(v1|unstable)/rooms/.*/relations/
    ^/_matrix/client/v1/rooms/.*/threads$
-    ^/_matrix/client/unstable/org.matrix.msc2716/rooms/.*/batch_send$
    ^/_matrix/client/unstable/im.nheko.summary/rooms/.*/summary$
    ^/_matrix/client/(r0|v3|unstable)/account/3pid$
    ^/_matrix/client/(r0|v3|unstable)/account/whoami$
@@ -532,6 +531,26 @@ the stream writer for the `presence` stream:

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

+#### Restrict outbound federation traffic to a specific set of workers
+
+The `outbound_federation_restricted_to` configuration is useful to make sure outbound
+federation traffic only goes through a specified subset of workers. This allows you to
+set more strict access controls (like a firewall) for all workers and only allow the
+`federation_sender`'s to contact the outside world.
+
+```yaml
+instance_map:
+    main:
+        host: localhost
+        port: 8030
+    federation_sender1:
+        host: localhost
+        port: 8034
+
+outbound_federation_restricted_to:
+  - federation_sender1
+```
+
 #### Background tasks

 There is also support for moving background tasks to a separate
@@ -178,7 +178,7 @@
                  EOF
                '';
                # Start synapse when `devenv up` is run.
-                processes.synapse.exec = "poetry run python -m synapse.app.homeserver -c homeserver.yaml --config-directory homeserver-config-overrides.d";
+                processes.synapse.exec = "poetry run python -m synapse.app.homeserver -c homeserver.yaml -c homeserver-config-overrides.d";

                # Define the perl modules we require to run SyTest.
                #
@@ -2,17 +2,29 @@
 namespace_packages = True
 plugins = pydantic.mypy, mypy_zope:plugin, scripts-dev/mypy_synapse_plugin.py
 follow_imports = normal
-check_untyped_defs = True
 show_error_codes = True
 show_traceback = True
 mypy_path = stubs
 warn_unreachable = True
-warn_unused_ignores = True
 local_partial_types = True
 no_implicit_optional = True
+
+# Strict checks, see mypy --help
+warn_unused_configs = True
+# disallow_any_generics = True
+disallow_subclassing_any = True
+# disallow_untyped_calls = True
 disallow_untyped_defs = True
-strict_equality = True
+disallow_incomplete_defs = True
+# check_untyped_defs = True
+# disallow_untyped_decorators = True
 warn_redundant_casts = True
+warn_unused_ignores = True
+# warn_return_any = True
+# no_implicit_reexport = True
+strict_equality = True
+strict_concatenate = True
+
 # Run mypy type checking with the minimum supported Python version to catch new usage
 # that isn't backwards-compatible (types, overloads, etc).
 python_version = 3.8
@@ -31,6 +43,7 @@ warn_unused_ignores = False

 [mypy-synapse.util.caches.treecache]
 disallow_untyped_defs = False
+disallow_incomplete_defs = False

 ;; Dependencies without annotations
 ;; Before ignoring a module, check to see if type stubs are available.
@@ -40,18 +53,18 @@ disallow_untyped_defs = False
 ;; which we can pull in as a dev dependency by adding to `pyproject.toml`'s
 ;; `[tool.poetry.dev-dependencies]` list.

+# https://github.com/lepture/authlib/issues/460
 [mypy-authlib.*]
 ignore_missing_imports = True

 [mypy-ijson.*]
 ignore_missing_imports = True

-[mypy-lxml]
-ignore_missing_imports = True
-
+# https://github.com/msgpack/msgpack-python/issues/448
 [mypy-msgpack]
 ignore_missing_imports = True

+# https://github.com/wolever/parameterized/issues/143
 [mypy-parameterized.*]
 ignore_missing_imports = True

@@ -73,6 +86,7 @@ ignore_missing_imports = True
 [mypy-srvlookup.*]
 ignore_missing_imports = True

+# https://github.com/twisted/treq/pull/366
 [mypy-treq.*]
 ignore_missing_imports = True

@@ -89,7 +89,7 @@ manifest-path = "rust/Cargo.toml"

 [tool.poetry]
 name = "matrix-synapse"
-version = "1.85.0rc1"
+version = "1.87.0"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "Apache-2.0"
@@ -147,7 +147,7 @@ synapse_review_recent_signups = "synapse._scripts.review_recent_signups:main"
 update_synapse_database = "synapse._scripts.update_synapse_database:main"

 [tool.poetry.dependencies]
-python = "^3.7.1"
+python = "^3.8.0"

 # Mandatory Dependencies
 # ----------------------
@@ -203,11 +203,9 @@ ijson = ">=3.1.4"
 matrix-common = "^1.3.0"
 # We need packaging.requirements.Requirement, added in 16.1.
 packaging = ">=16.1"
-# At the time of writing, we only use functions from the version `importlib.metadata`
-# which shipped in Python 3.8. This corresponds to version 1.4 of the backport.
-importlib_metadata = { version = ">=1.4", python = "<3.8" }
 # This is the most recent version of Pydantic with available on common distros.
-pydantic = ">=1.7.4"
+# We are currently incompatible with >=2.0.0: (https://github.com/matrix-org/synapse/issues/15858)
+pydantic = "^1.7.4"

 # This is for building the rust components during "poetry install", which
 # currently ignores the `build-system.requires` directive (c.f.
@@ -311,9 +309,10 @@ 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.265"
+ruff = "0.0.275"

 # Typechecking
+lxml-stubs = ">=0.4.0"
 mypy = "*"
 mypy-zope = "*"
 types-bleach = ">=4.1.0"
@@ -7,7 +7,7 @@ name = "synapse"
 version = "0.1.0"

 edition = "2021"
-rust-version = "1.58.1"
+rust-version = "1.60.0"

 [lib]
 name = "synapse"
@@ -13,8 +13,6 @@
 // limitations under the License.

 #![feature(test)]
-use std::collections::BTreeSet;
-
 use synapse::push::{
    evaluator::PushRuleEvaluator, Condition, EventMatchCondition, FilteredPushRules, JsonValue,
    PushRules, SimpleJsonValue,
@@ -197,7 +195,6 @@ fn bench_eval_message(b: &mut Bencher) {
        false,
        false,
        false,
-        false,
    );

    b.iter(|| eval.run(&rules, Some("bob"), Some("person")));
@@ -142,11 +142,11 @@ pub const BASE_APPEND_OVERRIDE_RULES: &[PushRule] = &[
        default_enabled: true,
    },
    PushRule {
-        rule_id: Cow::Borrowed(".org.matrix.msc3952.is_user_mention"),
+        rule_id: Cow::Borrowed("global/override/.m.rule.is_user_mention"),
        priority_class: 5,
        conditions: Cow::Borrowed(&[Condition::Known(
            KnownCondition::ExactEventPropertyContainsType(EventPropertyIsTypeCondition {
-                key: Cow::Borrowed("content.org\\.matrix\\.msc3952\\.mentions.user_ids"),
+                key: Cow::Borrowed("content.m\\.mentions.user_ids"),
                value_type: Cow::Borrowed(&EventMatchPatternType::UserId),
            }),
        )]),
@@ -163,11 +163,11 @@ pub const BASE_APPEND_OVERRIDE_RULES: &[PushRule] = &[
        default_enabled: true,
    },
    PushRule {
-        rule_id: Cow::Borrowed(".org.matrix.msc3952.is_room_mention"),
+        rule_id: Cow::Borrowed("global/override/.m.rule.is_room_mention"),
        priority_class: 5,
        conditions: Cow::Borrowed(&[
            Condition::Known(KnownCondition::EventPropertyIs(EventPropertyIsCondition {
-                key: Cow::Borrowed("content.org\\.matrix\\.msc3952\\.mentions.room"),
+                key: Cow::Borrowed("content.m\\.mentions.room"),
                value: Cow::Borrowed(&SimpleJsonValue::Bool(true)),
            })),
            Condition::Known(KnownCondition::SenderNotificationPermission {
@@ -70,7 +70,9 @@ pub struct PushRuleEvaluator {
    /// The "content.body", if any.
    body: String,

-    /// True if the event has a mentions property and MSC3952 support is enabled.
+    /// True if the event has a m.mentions property. (Note that this is a separate
+    /// flag instead of checking flattened_keys since the m.mentions property
+    /// might be an empty map and not appear in flattened_keys.
    has_mentions: bool,

    /// The number of users in the room.
@@ -155,9 +157,7 @@ impl PushRuleEvaluator {
            let rule_id = &push_rule.rule_id().to_string();

            // For backwards-compatibility the legacy mention rules are disabled
-            // if the event contains the 'm.mentions' property (and if the
-            // experimental feature is enabled, both of these are represented
-            // by the has_mentions flag).
+            // if the event contains the 'm.mentions' property.
            if self.has_mentions
                && (rule_id == "global/override/.m.rule.contains_display_name"
                    || rule_id == "global/content/.m.rule.contains_user_name"
@@ -562,7 +562,7 @@ fn test_requires_room_version_supports_condition() {
    };
    let rules = PushRules::new(vec![custom_rule]);
    result = evaluator.run(
-        &FilteredPushRules::py_new(rules, BTreeMap::new(), true, false, true, false, false),
+        &FilteredPushRules::py_new(rules, BTreeMap::new(), true, false, true, false),
        None,
        None,
    );
@@ -527,7 +527,6 @@ pub struct FilteredPushRules {
    msc1767_enabled: bool,
    msc3381_polls_enabled: bool,
    msc3664_enabled: bool,
-    msc3952_intentional_mentions: bool,
    msc3958_suppress_edits_enabled: bool,
 }

@@ -540,7 +539,6 @@ impl FilteredPushRules {
        msc1767_enabled: bool,
        msc3381_polls_enabled: bool,
        msc3664_enabled: bool,
-        msc3952_intentional_mentions: bool,
        msc3958_suppress_edits_enabled: bool,
    ) -> Self {
        Self {
@@ -549,7 +547,6 @@ impl FilteredPushRules {
            msc1767_enabled,
            msc3381_polls_enabled,
            msc3664_enabled,
-            msc3952_intentional_mentions,
            msc3958_suppress_edits_enabled,
        }
    }
@@ -587,10 +584,6 @@ impl FilteredPushRules {
                    return false;
                }

-                if !self.msc3952_intentional_mentions && rule.rule_id.contains("org.matrix.msc3952")
-                {
-                    return false;
-                }
                if !self.msc3958_suppress_edits_enabled
                    && rule.rule_id == "global/override/.com.beeper.suppress_edits"
                {
@@ -20,6 +20,8 @@ from concurrent.futures import ThreadPoolExecutor
 from types import FrameType
 from typing import Collection, Optional, Sequence, Set

+# These are expanded inside the dockerfile to be a fully qualified image name.
+# e.g. docker.io/library/debian:bullseye
 DISTS = (
    "debian:buster",  # oldstable: EOL 2022-08
    "debian:bullseye",
@@ -246,10 +246,6 @@ else
  else
    export PASS_SYNAPSE_COMPLEMENT_DATABASE=sqlite
  fi
-
-  # The tests for importing historical messages (MSC2716)
-  # only pass with monoliths, currently.
-  test_tags="$test_tags,msc2716"
 fi

 if [[ -n "$ASYNCIO_REACTOR" ]]; then
@@ -269,6 +265,10 @@ if [[ -n "$SYNAPSE_TEST_LOG_LEVEL" ]]; then
  export PASS_SYNAPSE_LOG_SENSITIVE=1
 fi

+# Log a few more useful things for a developer attempting to debug something
+# particularly tricky.
+export PASS_SYNAPSE_LOG_TESTING=1
+
 # Run the tests!
 echo "Images built; running complement"
 cd "$COMPLEMENT_DIR"
@@ -136,11 +136,11 @@ def request(
        authorization_headers.append(header)
        print("Authorization: %s" % header, file=sys.stderr)

-    dest = "matrix://%s%s" % (destination, path)
+    dest = "matrix-federation://%s%s" % (destination, path)
    print("Requesting %s" % dest, file=sys.stderr)

    s = requests.Session()
-    s.mount("matrix://", MatrixConnectionAdapter())
+    s.mount("matrix-federation://", MatrixConnectionAdapter())

    headers: Dict[str, str] = {
        "Authorization": authorization_headers[0],
@@ -46,7 +46,6 @@ class FilteredPushRules:
        msc1767_enabled: bool,
        msc3381_polls_enabled: bool,
        msc3664_enabled: bool,
-        msc3952_intentional_mentions: bool,
        msc3958_suppress_edits_enabled: bool,
    ): ...
    def rules(self) -> Collection[Tuple[PushRule, bool]]: ...
@@ -25,8 +25,8 @@ from synapse.util.rust import check_rust_lib_up_to_date
 from synapse.util.stringutils import strtobool

 # Check that we're not running on an unsupported Python version.
-if sys.version_info < (3, 7):
-    print("Synapse requires Python 3.7 or above.")
+if sys.version_info < (3, 8):
+    print("Synapse requires Python 3.8 or above.")
    sys.exit(1)

 # Allow using the asyncio reactor via env var.
@@ -61,6 +61,7 @@ from synapse.storage.databases.main.deviceinbox import DeviceInboxBackgroundUpda
 from synapse.storage.databases.main.devices import DeviceBackgroundUpdateStore
 from synapse.storage.databases.main.e2e_room_keys import EndToEndRoomKeyBackgroundStore
 from synapse.storage.databases.main.end_to_end_keys import EndToEndKeyBackgroundStore
+from synapse.storage.databases.main.event_federation import EventFederationWorkerStore
 from synapse.storage.databases.main.event_push_actions import EventPushActionsStore
 from synapse.storage.databases.main.events_bg_updates import (
    EventsBackgroundUpdatesStore,
@@ -196,6 +197,11 @@ IGNORED_TABLES = {
    "ui_auth_sessions",
    "ui_auth_sessions_credentials",
    "ui_auth_sessions_ips",
+    # Ignore the worker locks table, as a) there shouldn't be any acquired locks
+    # after porting, and b) the circular foreign key constraints make it hard to
+    # port.
+    "worker_read_write_locks_mode",
+    "worker_read_write_locks",
 }


@@ -239,6 +245,7 @@ class Store(
    PresenceBackgroundUpdateStore,
    ReceiptsBackgroundUpdateStore,
    RelationsWorkerStore,
+    EventFederationWorkerStore,
 ):
    def execute(self, f: Callable[..., R], *args: Any, **kwargs: Any) -> Awaitable[R]:
        return self.db_pool.runInteraction(f.__name__, f, *args, **kwargs)
@@ -803,7 +810,9 @@ class Porter:
            )
            # Map from table name to args passed to `handle_table`, i.e. a tuple
            # of: `postgres_size`, `table_size`, `forward_chunk`, `backward_chunk`.
-            tables_to_port_info_map = {r[0]: r[1:] for r in setup_res}
+            tables_to_port_info_map = {
+                r[0]: r[1:] for r in setup_res if r[0] not in IGNORED_TABLES
+            }

            # Step 5. Do the copying.
            #
@@ -1369,6 +1378,9 @@ def main() -> None:
        sys.stderr.write("Database must use the 'psycopg2' connector.\n")
        sys.exit(3)

+    # Don't run the background tasks that get started by the data stores.
+    hs_config["run_background_tasks_on"] = "some_other_process"
+
    config = HomeServerConfig()
    config.parse_config_dict(hs_config, "", "")

@@ -59,7 +59,7 @@ def scope_to_list(scope: str) -> List[str]:
    return scope.strip().split(" ")


-class PrivateKeyJWTWithKid(PrivateKeyJWT):
+class PrivateKeyJWTWithKid(PrivateKeyJWT):  # type: ignore[misc]
    """An implementation of the private_key_jwt client auth method that includes a kid header.

    This is needed because some providers (Keycloak) require the kid header to figure
@@ -123,10 +123,6 @@ class EventTypes:
    SpaceChild: Final = "m.space.child"
    SpaceParent: Final = "m.space.parent"

-    MSC2716_INSERTION: Final = "org.matrix.msc2716.insertion"
-    MSC2716_BATCH: Final = "org.matrix.msc2716.batch"
-    MSC2716_MARKER: Final = "org.matrix.msc2716.marker"
-
    Reaction: Final = "m.reaction"


@@ -222,21 +218,11 @@ class EventContentFields:
    # Used in m.room.guest_access events.
    GUEST_ACCESS: Final = "guest_access"

-    # Used on normal messages to indicate they were historically imported after the fact
-    MSC2716_HISTORICAL: Final = "org.matrix.msc2716.historical"
-    # For "insertion" events to indicate what the next batch ID should be in
-    # order to connect to it
-    MSC2716_NEXT_BATCH_ID: Final = "next_batch_id"
-    # Used on "batch" events to indicate which insertion event it connects to
-    MSC2716_BATCH_ID: Final = "batch_id"
-    # For "marker" events
-    MSC2716_INSERTION_EVENT_REFERENCE: Final = "insertion_event_reference"
-
    # The authorising user for joining a restricted room.
    AUTHORISING_USER: Final = "join_authorised_via_users_server"

    # Use for mentioning users.
-    MSC3952_MENTIONS: Final = "org.matrix.msc3952.mentions"
+    MENTIONS: Final = "m.mentions"

    # an unspecced field added to to-device messages to identify them uniquely-ish
    TO_DEVICE_MSGID: Final = "org.matrix.msgid"
@@ -152,9 +152,9 @@ class Filtering:
        self.DEFAULT_FILTER_COLLECTION = FilterCollection(hs, {})

    async def get_user_filter(
-        self, user_localpart: str, filter_id: Union[int, str]
+        self, user_id: UserID, filter_id: Union[int, str]
    ) -> "FilterCollection":
-        result = await self.store.get_user_filter(user_localpart, filter_id)
+        result = await self.store.get_user_filter(user_id, filter_id)
        return FilterCollection(self._hs, result)

    def add_user_filter(self, user_id: UserID, user_filter: JsonDict) -> Awaitable[int]:
@@ -91,11 +91,6 @@ class RoomVersion:
    # MSC2403: Allows join_rules to be set to 'knock', changes auth rules to allow sending
    # m.room.membership event with membership 'knock'.
    msc2403_knocking: bool
-    # MSC2716: Adds m.room.power_levels -> content.historical field to control
-    # whether "insertion", "chunk", "marker" events can be sent
-    msc2716_historical: bool
-    # MSC2716: Adds support for redacting "insertion", "chunk", and "marker" events
-    msc2716_redactions: bool
    # MSC3389: Protect relation information from redaction.
    msc3389_relation_redactions: bool
    # MSC3787: Adds support for a `knock_restricted` join rule, mixing concepts of
@@ -130,8 +125,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -153,8 +146,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -176,8 +167,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -199,8 +188,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -222,8 +209,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -245,8 +230,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -268,8 +251,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=False,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -291,8 +272,6 @@ class RoomVersions:
        msc3083_join_rules=False,
        msc3375_redaction_rules=False,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -314,8 +293,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=False,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -337,8 +314,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -360,8 +335,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=True,
        msc3667_int_only_power_levels=False,
@@ -383,8 +356,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=False,
        msc3667_int_only_power_levels=False,
@@ -406,8 +377,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=True,
        msc3667_int_only_power_levels=True,
@@ -415,29 +384,6 @@ class RoomVersions:
        msc3931_push_features=(),
        msc3989_redaction_rules=False,
    )
-    MSC2716v4 = RoomVersion(
-        "org.matrix.msc2716v4",
-        RoomDisposition.UNSTABLE,
-        EventFormatVersions.ROOM_V4_PLUS,
-        StateResolutionVersions.V2,
-        enforce_key_validity=True,
-        special_case_aliases_auth=False,
-        strict_canonicaljson=True,
-        limit_notifications_power_levels=True,
-        msc2175_implicit_room_creator=False,
-        msc2176_redaction_rules=False,
-        msc3083_join_rules=False,
-        msc3375_redaction_rules=False,
-        msc2403_knocking=True,
-        msc2716_historical=True,
-        msc2716_redactions=True,
-        msc3389_relation_redactions=False,
-        msc3787_knock_restricted_join_rule=False,
-        msc3667_int_only_power_levels=False,
-        msc3821_redaction_rules=False,
-        msc3931_push_features=(),
-        msc3989_redaction_rules=False,
-    )
    MSC1767v10 = RoomVersion(
        # MSC1767 (Extensible Events) based on room version "10"
        "org.matrix.msc1767.10",
@@ -453,8 +399,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=True,
        msc3667_int_only_power_levels=True,
@@ -476,8 +420,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=True,
        msc3667_int_only_power_levels=True,
@@ -500,8 +442,6 @@ class RoomVersions:
        msc3083_join_rules=True,
        msc3375_redaction_rules=True,
        msc2403_knocking=True,
-        msc2716_historical=False,
-        msc2716_redactions=False,
        msc3389_relation_redactions=False,
        msc3787_knock_restricted_join_rule=True,
        msc3667_int_only_power_levels=True,
@@ -526,7 +466,6 @@ KNOWN_ROOM_VERSIONS: Dict[str, RoomVersion] = {
        RoomVersions.V9,
        RoomVersions.MSC3787,
        RoomVersions.V10,
-        RoomVersions.MSC2716v4,
        RoomVersions.MSC3989,
        RoomVersions.MSC3820opt2,
    )
@@ -386,6 +386,7 @@ def listen_unix(


 def listen_http(
+    hs: "HomeServer",
    listener_config: ListenerConfig,
    root_resource: Resource,
    version_string: str,
@@ -406,6 +407,7 @@ def listen_http(
        version_string,
        max_request_body_size=max_request_body_size,
        reactor=reactor,
+        federation_agent=hs.get_federation_http_client().agent,
    )

    if isinstance(listener_config, TCPListenerConfig):
@@ -83,7 +83,6 @@ from synapse.storage.databases.main.receipts import ReceiptsWorkerStore
 from synapse.storage.databases.main.registration import RegistrationWorkerStore
 from synapse.storage.databases.main.relations import RelationsWorkerStore
 from synapse.storage.databases.main.room import RoomWorkerStore
-from synapse.storage.databases.main.room_batch import RoomBatchStore
 from synapse.storage.databases.main.roommember import RoomMemberWorkerStore
 from synapse.storage.databases.main.search import SearchStore
 from synapse.storage.databases.main.session import SessionStore
@@ -120,7 +119,6 @@ class GenericWorkerStore(
    # the races it creates aren't too bad.
    KeyStore,
    RoomWorkerStore,
-    RoomBatchStore,
    DirectoryWorkerStore,
    PushRulesWorkerStore,
    ApplicationServiceTransactionWorkerStore,
@@ -223,6 +221,7 @@ class GenericWorkerServer(HomeServer):
        root_resource = create_resource_tree(resources, OptionsResource())

        _base.listen_http(
+            self,
            listener_config,
            root_resource,
            self.version_string,
@@ -139,6 +139,7 @@ class SynapseHomeServer(HomeServer):
            root_resource = OptionsResource()

        ports = listen_http(
+            self,
            listener_config,
            create_resource_tree(resources, root_resource),
            self.version_string,
@@ -60,3 +60,13 @@ class AuthConfig(Config):
        self.ui_auth_session_timeout = self.parse_duration(
            ui_auth.get("session_timeout", 0)
        )
+
+        # Logging in with an existing session.
+        login_via_existing = config.get("login_via_existing_session", {})
+        self.login_via_existing_enabled = login_via_existing.get("enabled", False)
+        self.login_via_existing_require_ui_auth = login_via_existing.get(
+            "require_ui_auth", True
+        )
+        self.login_via_existing_token_timeout = self.parse_duration(
+            login_via_existing.get("token_timeout", "5m")
+        )
@@ -192,10 +192,10 @@ class MSC3861:
                ("captcha", "enable_registration_captcha"),
            )

-        if root.experimental.msc3882_enabled:
+        if root.auth.login_via_existing_enabled:
            raise ConfigError(
-                "MSC3882 cannot be enabled when OAuth delegation is enabled",
-                ("experimental_features", "msc3882_enabled"),
+                "Login via existing session cannot be enabled when OAuth delegation is enabled",
+                ("login_via_existing_session", "enabled"),
            )

        if root.registration.refresh_token_lifetime:
@@ -247,9 +247,6 @@ class ExperimentalConfig(Config):
        # MSC3026 (busy presence state)
        self.msc3026_enabled: bool = experimental.get("msc3026_enabled", False)

-        # MSC2716 (importing historical messages)
-        self.msc2716_enabled: bool = experimental.get("msc2716_enabled", False)
-
        # MSC3244 (room version capabilities)
        self.msc3244_enabled: bool = experimental.get("msc3244_enabled", True)

@@ -319,13 +316,6 @@ class ExperimentalConfig(Config):
        # MSC3881: Remotely toggle push notifications for another client
        self.msc3881_enabled: bool = experimental.get("msc3881_enabled", False)

-        # MSC3882: Allow an existing session to sign in a new session
-        self.msc3882_enabled: bool = experimental.get("msc3882_enabled", False)
-        self.msc3882_ui_auth: bool = experimental.get("msc3882_ui_auth", True)
-        self.msc3882_token_timeout = self.parse_duration(
-            experimental.get("msc3882_token_timeout", "5m")
-        )
-
        # MSC3874: Filtering /messages with rel_types / not_rel_types.
        self.msc3874_enabled: bool = experimental.get("msc3874_enabled", False)

@@ -365,11 +355,6 @@ class ExperimentalConfig(Config):
        # MSC3391: Removing account data.
        self.msc3391_enabled = experimental.get("msc3391_enabled", False)

-        # MSC3952: Intentional mentions, this depends on MSC3966.
-        self.msc3952_intentional_mentions = experimental.get(
-            "msc3952_intentional_mentions", False
-        )
-
        # MSC3959: Do not generate notifications for edits.
        self.msc3958_supress_edit_notifs = experimental.get(
            "msc3958_supress_edit_notifs", False
@@ -22,6 +22,8 @@ class FederationConfig(Config):
    section = "federation"

    def read_config(self, config: JsonDict, **kwargs: Any) -> None:
+        federation_config = config.setdefault("federation", {})
+
        # FIXME: federation_domain_whitelist needs sytests
        self.federation_domain_whitelist: Optional[dict] = None
        federation_domain_whitelist = config.get("federation_domain_whitelist", None)
@@ -49,5 +51,19 @@ class FederationConfig(Config):
            "allow_device_name_lookup_over_federation", False
        )

+        # Allow for the configuration of timeout, max request retries
+        # and min/max retry delays in the matrix federation client.
+        self.client_timeout_ms = Config.parse_duration(
+            federation_config.get("client_timeout", "60s")
+        )
+        self.max_long_retry_delay_ms = Config.parse_duration(
+            federation_config.get("max_long_retry_delay", "60s")
+        )
+        self.max_short_retry_delay_ms = Config.parse_duration(
+            federation_config.get("max_short_retry_delay", "2s")
+        )
+        self.max_long_retries = federation_config.get("max_long_retries", 10)
+        self.max_short_retries = federation_config.get("max_short_retries", 3)
+

 _METRICS_FOR_DOMAINS_SCHEMA = {"type": "array", "items": {"type": "string"}}
@@ -15,7 +15,7 @@

 import argparse
 import logging
-from typing import Any, Dict, List, Union
+from typing import Any, Dict, List, Optional, Union

 import attr
 from pydantic import BaseModel, Extra, StrictBool, StrictInt, StrictStr
@@ -148,6 +148,27 @@ class WriterLocations:
    )


+@attr.s(auto_attribs=True)
+class OutboundFederationRestrictedTo:
+    """Whether we limit outbound federation to a certain set of instances.
+
+    Attributes:
+        instances: optional list of instances that can make outbound federation
+            requests. If None then all instances can make federation requests.
+        locations: list of instance locations to connect to proxy via.
+    """
+
+    instances: Optional[List[str]]
+    locations: List[InstanceLocationConfig] = attr.Factory(list)
+
+    def __contains__(self, instance: str) -> bool:
+        # It feels a bit dirty to return `True` if `instances` is `None`, but it makes
+        # sense in downstream usage in the sense that if
+        # `outbound_federation_restricted_to` is not configured, then any instance can
+        # talk to federation (no restrictions so always return `True`).
+        return self.instances is None or instance in self.instances
+
+
 class WorkerConfig(Config):
    """The workers are processes run separately to the main synapse process.
    They have their own pid_file and listener configuration. They use the
@@ -357,6 +378,23 @@ class WorkerConfig(Config):
            new_option_name="update_user_directory_from_worker",
        )

+        outbound_federation_restricted_to = config.get(
+            "outbound_federation_restricted_to", None
+        )
+        self.outbound_federation_restricted_to = OutboundFederationRestrictedTo(
+            outbound_federation_restricted_to
+        )
+        if outbound_federation_restricted_to:
+            for instance in outbound_federation_restricted_to:
+                if instance not in self.instance_map:
+                    raise ConfigError(
+                        "Instance %r is configured in 'outbound_federation_restricted_to' but does not appear in `instance_map` config."
+                        % (instance,)
+                    )
+                self.outbound_federation_restricted_to.locations.append(
+                    self.instance_map[instance]
+                )
+
    def _should_this_worker_perform_duty(
        self,
        config: Dict[str, Any],
@@ -339,13 +339,6 @@ def check_state_dependent_auth_rules(
    if event.type == EventTypes.Redaction:
        check_redaction(event.room_version, event, auth_dict)

-    if (
-        event.type == EventTypes.MSC2716_INSERTION
-        or event.type == EventTypes.MSC2716_BATCH
-        or event.type == EventTypes.MSC2716_MARKER
-    ):
-        check_historical(event.room_version, event, auth_dict)
-
    logger.debug("Allowing! %s", event)


@@ -365,7 +358,6 @@ LENIENT_EVENT_BYTE_LIMITS_ROOM_VERSIONS = {
    RoomVersions.V9,
    RoomVersions.MSC3787,
    RoomVersions.V10,
-    RoomVersions.MSC2716v4,
    RoomVersions.MSC1767v10,
 }

@@ -823,38 +815,6 @@ def check_redaction(
    raise AuthError(403, "You don't have permission to redact events")


-def check_historical(
-    room_version_obj: RoomVersion,
-    event: "EventBase",
-    auth_events: StateMap["EventBase"],
-) -> None:
-    """Check whether the event sender is allowed to send historical related
-    events like "insertion", "batch", and "marker".
-
-    Returns:
-        None
-
-    Raises:
-        AuthError if the event sender is not allowed to send historical related events
-        ("insertion", "batch", and "marker").
-    """
-    # Ignore the auth checks in room versions that do not support historical
-    # events
-    if not room_version_obj.msc2716_historical:
-        return
-
-    user_level = get_user_power_level(event.user_id, auth_events)
-
-    historical_level = get_named_level(auth_events, "historical", 100)
-
-    if user_level < historical_level:
-        raise UnstableSpecAuthError(
-            403,
-            'You don\'t have permission to send send historical related events ("insertion", "batch", and "marker")',
-            errcode=Codes.INSUFFICIENT_POWER,
-        )
-
-
 def _check_power_levels(
    room_version_obj: RoomVersion,
    event: "EventBase",
@@ -198,7 +198,6 @@ class _EventInternalMetadata:
    soft_failed: DictProperty[bool] = DictProperty("soft_failed")
    proactively_send: DictProperty[bool] = DictProperty("proactively_send")
    redacted: DictProperty[bool] = DictProperty("redacted")
-    historical: DictProperty[bool] = DictProperty("historical")

    txn_id: DictProperty[str] = DictProperty("txn_id")
    """The transaction ID, if it was set when the event was created."""
@@ -288,14 +287,6 @@ class _EventInternalMetadata:
        """
        return self._dict.get("redacted", False)

-    def is_historical(self) -> bool:
-        """Whether this is a historical message.
-        This is used by the batchsend historical message endpoint and
-        is needed to and mark the event as backfilled and skip some checks
-        like push notifications.
-        """
-        return self._dict.get("historical", False)
-
    def is_notifiable(self) -> bool:
        """Whether this event can trigger a push notification"""
        return not self.is_outlier() or self.is_out_of_band_membership()
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, List, Optional, Tuple
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple

 import attr
 from immutabledict import immutabledict
@@ -107,33 +107,32 @@ class EventContext(UnpersistedEventContextBase):
        state_delta_due_to_event: If `state_group` and `state_group_before_event` are not None
            then this is the delta of the state between the two groups.

-        prev_group: If it is known, ``state_group``'s prev_group. Note that this being
-            None does not necessarily mean that ``state_group`` does not have
-            a prev_group!
+        state_group_deltas: If not empty, this is a dict collecting a mapping of the state
+            difference between state groups.

-            If the event is a state event, this is normally the same as
-            ``state_group_before_event``.
+            The keys are a tuple of two integers: the initial group and final state group.
+            The corresponding value is a state map representing the state delta between
+            these state groups.

-            If ``state_group`` is None (ie, the event is an outlier), ``prev_group``
-            will always also be ``None``.
+            The dictionary is expected to have at most two entries with state groups of:

-            Note that this *not* (necessarily) the state group associated with
-            ``_prev_state_ids``.
+            1. The state group before the event and after the event.
+            2. The state group preceding the state group before the event and the
+               state group before the event.

-        delta_ids: If ``prev_group`` is not None, the state delta between ``prev_group``
-            and ``state_group``.
+            This information is collected and stored as part of an optimization for persisting
+            events.

        partial_state: if True, we may be storing this event with a temporary,
            incomplete state.
    """

    _storage: "StorageControllers"
+    state_group_deltas: Dict[Tuple[int, int], StateMap[str]]
    rejected: Optional[str] = None
    _state_group: Optional[int] = None
    state_group_before_event: Optional[int] = None
    _state_delta_due_to_event: Optional[StateMap[str]] = None
-    prev_group: Optional[int] = None
-    delta_ids: Optional[StateMap[str]] = None
    app_service: Optional[ApplicationService] = None

    partial_state: bool = False
@@ -145,16 +144,14 @@ class EventContext(UnpersistedEventContextBase):
        state_group_before_event: Optional[int],
        state_delta_due_to_event: Optional[StateMap[str]],
        partial_state: bool,
-        prev_group: Optional[int] = None,
-        delta_ids: Optional[StateMap[str]] = None,
+        state_group_deltas: Dict[Tuple[int, int], StateMap[str]],
    ) -> "EventContext":
        return EventContext(
            storage=storage,
            state_group=state_group,
            state_group_before_event=state_group_before_event,
            state_delta_due_to_event=state_delta_due_to_event,
-            prev_group=prev_group,
-            delta_ids=delta_ids,
+            state_group_deltas=state_group_deltas,
            partial_state=partial_state,
        )

@@ -163,7 +160,7 @@ class EventContext(UnpersistedEventContextBase):
        storage: "StorageControllers",
    ) -> "EventContext":
        """Return an EventContext instance suitable for persisting an outlier event"""
-        return EventContext(storage=storage)
+        return EventContext(storage=storage, state_group_deltas={})

    async def persist(self, event: EventBase) -> "EventContext":
        return self
@@ -183,13 +180,15 @@ class EventContext(UnpersistedEventContextBase):
            "state_group": self._state_group,
            "state_group_before_event": self.state_group_before_event,
            "rejected": self.rejected,
-            "prev_group": self.prev_group,
+            "state_group_deltas": _encode_state_group_delta(self.state_group_deltas),
            "state_delta_due_to_event": _encode_state_dict(
                self._state_delta_due_to_event
            ),
-            "delta_ids": _encode_state_dict(self.delta_ids),
            "app_service_id": self.app_service.id if self.app_service else None,
            "partial_state": self.partial_state,
+            # add dummy delta_ids and prev_group for backwards compatibility
+            "delta_ids": None,
+            "prev_group": None,
        }

    @staticmethod
@@ -204,17 +203,24 @@ class EventContext(UnpersistedEventContextBase):
        Returns:
            The event context.
        """
+        # workaround for backwards/forwards compatibility: if the input doesn't have a value
+        # for "state_group_deltas" just assign an empty dict
+        state_group_deltas = input.get("state_group_deltas", None)
+        if state_group_deltas:
+            state_group_deltas = _decode_state_group_delta(state_group_deltas)
+        else:
+            state_group_deltas = {}
+
        context = EventContext(
            # We use the state_group and prev_state_id stuff to pull the
            # current_state_ids out of the DB and construct prev_state_ids.
            storage=storage,
            state_group=input["state_group"],
            state_group_before_event=input["state_group_before_event"],
-            prev_group=input["prev_group"],
+            state_group_deltas=state_group_deltas,
            state_delta_due_to_event=_decode_state_dict(
                input["state_delta_due_to_event"]
            ),
-            delta_ids=_decode_state_dict(input["delta_ids"]),
            rejected=input["rejected"],
            partial_state=input.get("partial_state", False),
        )
@@ -349,7 +355,7 @@ class UnpersistedEventContext(UnpersistedEventContextBase):
    _storage: "StorageControllers"
    state_group_before_event: Optional[int]
    state_group_after_event: Optional[int]
-    state_delta_due_to_event: Optional[dict]
+    state_delta_due_to_event: Optional[StateMap[str]]
    prev_group_for_state_group_before_event: Optional[int]
    delta_ids_to_state_group_before_event: Optional[StateMap[str]]
    partial_state: bool
@@ -380,26 +386,16 @@ class UnpersistedEventContext(UnpersistedEventContextBase):

        events_and_persisted_context = []
        for event, unpersisted_context in amended_events_and_context:
-            if event.is_state():
-                context = EventContext(
-                    storage=unpersisted_context._storage,
-                    state_group=unpersisted_context.state_group_after_event,
-                    state_group_before_event=unpersisted_context.state_group_before_event,
-                    state_delta_due_to_event=unpersisted_context.state_delta_due_to_event,
-                    partial_state=unpersisted_context.partial_state,
-                    prev_group=unpersisted_context.state_group_before_event,
-                    delta_ids=unpersisted_context.state_delta_due_to_event,
-                )
-            else:
-                context = EventContext(
-                    storage=unpersisted_context._storage,
-                    state_group=unpersisted_context.state_group_after_event,
-                    state_group_before_event=unpersisted_context.state_group_before_event,
-                    state_delta_due_to_event=unpersisted_context.state_delta_due_to_event,
-                    partial_state=unpersisted_context.partial_state,
-                    prev_group=unpersisted_context.prev_group_for_state_group_before_event,
-                    delta_ids=unpersisted_context.delta_ids_to_state_group_before_event,
-                )
+            state_group_deltas = unpersisted_context._build_state_group_deltas()
+
+            context = EventContext(
+                storage=unpersisted_context._storage,
+                state_group=unpersisted_context.state_group_after_event,
+                state_group_before_event=unpersisted_context.state_group_before_event,
+                state_delta_due_to_event=unpersisted_context.state_delta_due_to_event,
+                partial_state=unpersisted_context.partial_state,
+                state_group_deltas=state_group_deltas,
+            )
            events_and_persisted_context.append((event, context))
        return events_and_persisted_context

@@ -452,11 +448,11 @@ class UnpersistedEventContext(UnpersistedEventContextBase):

        # if the event isn't a state event the state group doesn't change
        if not self.state_delta_due_to_event:
-            state_group_after_event = self.state_group_before_event
+            self.state_group_after_event = self.state_group_before_event

        # otherwise if it is a state event we need to get a state group for it
        else:
-            state_group_after_event = await self._storage.state.store_state_group(
+            self.state_group_after_event = await self._storage.state.store_state_group(
                event.event_id,
                event.room_id,
                prev_group=self.state_group_before_event,
@@ -464,16 +460,81 @@ class UnpersistedEventContext(UnpersistedEventContextBase):
                current_state_ids=None,
            )

+        state_group_deltas = self._build_state_group_deltas()
+
        return EventContext.with_state(
            storage=self._storage,
-            state_group=state_group_after_event,
+            state_group=self.state_group_after_event,
            state_group_before_event=self.state_group_before_event,
            state_delta_due_to_event=self.state_delta_due_to_event,
+            state_group_deltas=state_group_deltas,
            partial_state=self.partial_state,
-            prev_group=self.state_group_before_event,
-            delta_ids=self.state_delta_due_to_event,
        )

+    def _build_state_group_deltas(self) -> Dict[Tuple[int, int], StateMap]:
+        """
+        Collect deltas between the state groups associated with this context
+        """
+        state_group_deltas = {}
+
+        # if we know the state group before the event and after the event, add them and the
+        # state delta between them to state_group_deltas
+        if self.state_group_before_event and self.state_group_after_event:
+            # if we have the state groups we should have the delta
+            assert self.state_delta_due_to_event is not None
+            state_group_deltas[
+                (
+                    self.state_group_before_event,
+                    self.state_group_after_event,
+                )
+            ] = self.state_delta_due_to_event
+
+        # the state group before the event may also have a state group which precedes it, if
+        # we have that and the state group before the event, add them and the state
+        # delta between them to state_group_deltas
+        if (
+            self.prev_group_for_state_group_before_event
+            and self.state_group_before_event
+        ):
+            # if we have both state groups we should have the delta between them
+            assert self.delta_ids_to_state_group_before_event is not None
+            state_group_deltas[
+                (
+                    self.prev_group_for_state_group_before_event,
+                    self.state_group_before_event,
+                )
+            ] = self.delta_ids_to_state_group_before_event
+
+        return state_group_deltas
+
+
+def _encode_state_group_delta(
+    state_group_delta: Dict[Tuple[int, int], StateMap[str]]
+) -> List[Tuple[int, int, Optional[List[Tuple[str, str, str]]]]]:
+    if not state_group_delta:
+        return []
+
+    state_group_delta_encoded = []
+    for key, value in state_group_delta.items():
+        state_group_delta_encoded.append((key[0], key[1], _encode_state_dict(value)))
+
+    return state_group_delta_encoded
+
+
+def _decode_state_group_delta(
+    input: List[Tuple[int, int, List[Tuple[str, str, str]]]]
+) -> Dict[Tuple[int, int], StateMap[str]]:
+    if not input:
+        return {}
+
+    state_group_deltas = {}
+    for state_group_1, state_group_2, state_dict in input:
+        state_map = _decode_state_dict(state_dict)
+        assert state_map is not None
+        state_group_deltas[(state_group_1, state_group_2)] = state_map
+
+    return state_group_deltas
+

 def _encode_state_dict(
    state_dict: Optional[StateMap[str]],
@@ -164,21 +164,12 @@ def prune_event_dict(room_version: RoomVersion, event_dict: JsonDict) -> JsonDic
        if room_version.msc2176_redaction_rules:
            add_fields("invite")

-        if room_version.msc2716_historical:
-            add_fields("historical")
-
    elif event_type == EventTypes.Aliases and room_version.special_case_aliases_auth:
        add_fields("aliases")
    elif event_type == EventTypes.RoomHistoryVisibility:
        add_fields("history_visibility")
    elif event_type == EventTypes.Redaction and room_version.msc2176_redaction_rules:
        add_fields("redacts")
-    elif room_version.msc2716_redactions and event_type == EventTypes.MSC2716_INSERTION:
-        add_fields(EventContentFields.MSC2716_NEXT_BATCH_ID)
-    elif room_version.msc2716_redactions and event_type == EventTypes.MSC2716_BATCH:
-        add_fields(EventContentFields.MSC2716_BATCH_ID)
-    elif room_version.msc2716_redactions and event_type == EventTypes.MSC2716_MARKER:
-        add_fields(EventContentFields.MSC2716_INSERTION_EVENT_REFERENCE)

    # Protect the rel_type and event_id fields under the m.relates_to field.
    if room_version.msc3389_relation_redactions:
@@ -134,13 +134,8 @@ class EventValidator:
                )

        # 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
-            )
+        if EventContentFields.MENTIONS in event.content:
+            validate_json_object(event.content[EventContentFields.MENTIONS], Mentions)

    def _validate_retention(self, event: EventBase) -> None:
        """Checks that an event that defines the retention policy for a room respects the
@@ -260,7 +260,9 @@ class FederationClient(FederationBase):
        use_unstable = False
        for user_id, one_time_keys in query.items():
            for device_id, algorithms in one_time_keys.items():
-                if any(count > 1 for count in algorithms.values()):
+                # If more than one algorithm is requested, attempt to use the unstable
+                # endpoint.
+                if sum(algorithms.values()) > 1:
                    use_unstable = True
                if algorithms:
                    # For the stable query, choose only the first algorithm.
@@ -296,6 +298,7 @@ class FederationClient(FederationBase):
        else:
            logger.debug("Skipping unstable claim client keys API")

+        # TODO Potentially attempt multiple queries and combine the results?
        return await self.transport_layer.claim_client_keys(
            user, destination, content, timeout
        )
@@ -515,7 +515,7 @@ class FederationServer(FederationBase):
                    logger.error(
                        "Failed to handle PDU %s",
                        event_id,
-                        exc_info=(f.type, f.value, f.getTracebackObject()),  # type: ignore
+                        exc_info=(f.type, f.value, f.getTracebackObject()),
                    )
                    return {"error": str(e)}

@@ -944,7 +944,7 @@ class FederationServer(FederationBase):
            if not self._is_mine_server_name(authorising_server):
                raise SynapseError(
                    400,
-                    f"Cannot authorise request from resident server: {authorising_server}",
+                    f"Cannot authorise membership event for {authorising_server}. We can only authorise requests from our own homeserver",
                )

            event.signatures.update(
@@ -1016,7 +1016,9 @@ class FederationServer(FederationBase):
            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}
+                        json_result.setdefault(user_id, {}).setdefault(device_id, {})[
+                            key_id
+                        ] = key

        logger.info(
            "Claimed one-time-keys: %s",
@@ -1247,7 +1249,7 @@ class FederationServer(FederationBase):
                    logger.error(
                        "Failed to handle PDU %s",
                        event.event_id,
-                        exc_info=(f.type, f.value, f.getTracebackObject()),  # type: ignore
+                        exc_info=(f.type, f.value, f.getTracebackObject()),
                    )

                received_ts = await self.store.remove_received_event_from_staging(
@@ -109,10 +109,8 @@ was enabled*, Catch-Up Mode is exited and we return to `_transaction_transmissio

 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.
+We automatically retry after the retry interval expires (roughly, the logic to do so
+being triggered every minute).

 If the backoff grows too large (> 1 hour), the in-memory queue is emptied (to prevent
 unbounded growth) and Catch-Up Mode is entered.
@@ -145,7 +143,6 @@ from prometheus_client import Counter
 from typing_extensions import Literal

 from twisted.internet import defer
-from twisted.internet.interfaces import IDelayedCall

 import synapse.metrics
 from synapse.api.presence import UserPresenceState
@@ -184,14 +181,18 @@ sent_pdus_destination_dist_total = Counter(
    "Total number of PDUs queued for sending across all destinations",
 )

-# Time (in s) after Synapse's startup that we will begin to wake up destinations
-# that have catch-up outstanding.
-CATCH_UP_STARTUP_DELAY_SEC = 15
+# Time (in s) to wait before trying to wake up destinations that have
+# catch-up outstanding. This will also be the delay applied at startup
+# before trying the same.
+# Please note that rate limiting still applies, so while the loop is
+# executed every X seconds the destinations may not be wake up because
+# they are being rate limited following previous attempt failures.
+WAKEUP_RETRY_PERIOD_SEC = 60

 # Time (in s) to wait in between waking up each destination, i.e. one destination
-# will be woken up every <x> seconds after Synapse's startup until we have woken
-# every destination has outstanding catch-up.
-CATCH_UP_STARTUP_INTERVAL_SEC = 5
+# will be woken up every <x> seconds until we have woken every destination
+# has outstanding catch-up.
+WAKEUP_INTERVAL_BETWEEN_DESTINATIONS_SEC = 5


 class AbstractFederationSender(metaclass=abc.ABCMeta):
@@ -415,12 +416,10 @@ class FederationSender(AbstractFederationSender):
            / hs.config.ratelimiting.federation_rr_transactions_per_room_per_second
        )

-        # wake up destinations that have outstanding PDUs to be caught up
-        self._catchup_after_startup_timer: Optional[
-            IDelayedCall
-        ] = self.clock.call_later(
-            CATCH_UP_STARTUP_DELAY_SEC,
+        # Regularly wake up destinations that have outstanding PDUs to be caught up
+        self.clock.looping_call(
            run_as_background_process,
+            WAKEUP_RETRY_PERIOD_SEC * 1000.0,
            "wake_destinations_needing_catchup",
            self._wake_destinations_needing_catchup,
        )
@@ -966,7 +965,6 @@ class FederationSender(AbstractFederationSender):

            if not destinations_to_wake:
                # finished waking all destinations!
-                self._catchup_after_startup_timer = None
                break

            last_processed = destinations_to_wake[-1]
@@ -983,4 +981,4 @@ class FederationSender(AbstractFederationSender):
                    last_processed,
                )
                self.wake_destination(destination)
-                await self.clock.sleep(CATCH_UP_STARTUP_INTERVAL_SEC)
+                await self.clock.sleep(WAKEUP_INTERVAL_BETWEEN_DESTINATIONS_SEC)
@@ -164,7 +164,7 @@ class AccountValidityHandler:

        try:
            user_display_name = await self.store.get_profile_displayname(
-                UserID.from_string(user_id).localpart
+                UserID.from_string(user_id)
            )
            if user_display_name is None:
                user_display_name = user_id
@@ -89,7 +89,7 @@ class AdminHandler:
        }

        # Add additional user metadata
-        profile = await self._store.get_profileinfo(user.localpart)
+        profile = await self._store.get_profileinfo(user)
        threepids = await self._store.user_get_threepids(user.to_string())
        external_ids = [
            ({"auth_provider": auth_provider, "external_id": external_id})
@@ -1759,7 +1759,7 @@ class AuthHandler:
            return

        user_profile_data = await self.store.get_profileinfo(
-            UserID.from_string(registered_user_id).localpart
+            UserID.from_string(registered_user_id)
        )

        # Store any extra attributes which will be passed in the login response.
@@ -297,5 +297,5 @@ class DeactivateAccountHandler:
        # Add the user to the directory, if necessary. Note that
        # this must be done after the user is re-activated, because
        # deactivated users are excluded from the user directory.
-        profile = await self.store.get_profileinfo(user.localpart)
+        profile = await self.store.get_profileinfo(user)
        await self.user_directory_handler.handle_local_profile_change(user_id, profile)
@@ -105,14 +105,12 @@ backfill_processing_before_timer = Histogram(
 )


+# TODO: We can refactor this away now that there is only one backfill point again
 class _BackfillPointType(Enum):
    # a regular backwards extremity (ie, an event which we don't yet have, but which
    # is referred to by other events in the DAG)
    BACKWARDS_EXTREMITY = enum.auto()

-    # an MSC2716 "insertion event"
-    INSERTION_PONT = enum.auto()
-

@attr.s(slots=True, auto_attribs=True, frozen=True)
 class _BackfillPoint:
@@ -200,6 +198,7 @@ class FederationHandler:
            )

    @trace
+    @tag_args
    async def maybe_backfill(
        self, room_id: str, current_depth: int, limit: int
    ) -> bool:
@@ -214,6 +213,9 @@ class FederationHandler:
            limit: The number of events that the pagination request will
                return. This is used as part of the heuristic to decide if we
                should back paginate.
+
+        Returns:
+            True if we actually tried to backfill something, otherwise False.
        """
        # Starting the processing time here so we can include the room backfill
        # linearizer lock queue in the timing
@@ -227,6 +229,8 @@ class FederationHandler:
                processing_start_time=processing_start_time,
            )

+    @trace
+    @tag_args
    async def _maybe_backfill_inner(
        self,
        room_id: str,
@@ -247,6 +251,9 @@ class FederationHandler:
            limit: The max number of events to request from the remote federated server.
            processing_start_time: The time when `maybe_backfill` started processing.
                Only used for timing. If `None`, no timing observation will be made.
+
+        Returns:
+            True if we actually tried to backfill something, otherwise False.
        """
        backwards_extremities = [
            _BackfillPoint(event_id, depth, _BackfillPointType.BACKWARDS_EXTREMITY)
@@ -264,32 +271,10 @@ class FederationHandler:
            )
        ]

-        insertion_events_to_be_backfilled: List[_BackfillPoint] = []
-        if self.hs.config.experimental.msc2716_enabled:
-            insertion_events_to_be_backfilled = [
-                _BackfillPoint(event_id, depth, _BackfillPointType.INSERTION_PONT)
-                for event_id, depth in await self.store.get_insertion_event_backward_extremities_in_room(
-                    room_id=room_id,
-                    current_depth=current_depth,
-                    # We only need to end up with 5 extremities combined with
-                    # the backfill points to make the `/backfill` request ...
-                    # (see the other comment above for more context).
-                    limit=50,
-                )
-            ]
-        logger.debug(
-            "_maybe_backfill_inner: backwards_extremities=%s insertion_events_to_be_backfilled=%s",
-            backwards_extremities,
-            insertion_events_to_be_backfilled,
-        )
-
        # we now have a list of potential places to backpaginate from. We prefer to
        # start with the most recent (ie, max depth), so let's sort the list.
        sorted_backfill_points: List[_BackfillPoint] = sorted(
-            itertools.chain(
-                backwards_extremities,
-                insertion_events_to_be_backfilled,
-            ),
+            backwards_extremities,
            key=lambda e: -int(e.depth),
        )

@@ -302,15 +287,30 @@ class FederationHandler:
            len(sorted_backfill_points),
            sorted_backfill_points,
        )
+        set_tag(
+            SynapseTags.RESULT_PREFIX + "sorted_backfill_points",
+            str(sorted_backfill_points),
+        )
+        set_tag(
+            SynapseTags.RESULT_PREFIX + "sorted_backfill_points.length",
+            str(len(sorted_backfill_points)),
+        )

-        # If we have no backfill points lower than the `current_depth` then
-        # either we can a) bail or b) still attempt to backfill. We opt to try
-        # backfilling anyway just in case we do get relevant events.
+        # If we have no backfill points lower than the `current_depth` then either we
+        # can a) bail or b) still attempt to backfill. We opt to try backfilling anyway
+        # just in case we do get relevant events. This is good for eventual consistency
+        # sake but we don't need to block the client for something that is just as
+        # likely not to return anything relevant so we backfill in the background. The
+        # only way, this could return something relevant is if we discover a new branch
+        # of history that extends all the way back to where we are currently paginating
+        # and it's within the 100 events that are returned from `/backfill`.
        if not sorted_backfill_points and current_depth != MAX_DEPTH:
            logger.debug(
                "_maybe_backfill_inner: all backfill points are *after* current depth. Trying again with later backfill points."
            )
-            return await self._maybe_backfill_inner(
+            run_as_background_process(
+                "_maybe_backfill_inner_anyway_with_max_depth",
+                self._maybe_backfill_inner,
                room_id=room_id,
                # We use `MAX_DEPTH` so that we find all backfill points next
                # time (all events are below the `MAX_DEPTH`)
@@ -321,6 +321,9 @@ class FederationHandler:
                # overall otherwise the smaller one will throw off the results.
                processing_start_time=None,
            )
+            # We return `False` because we're backfilling in the background and there is
+            # no new events immediately for the caller to know about yet.
+            return False

        # Even after recursing with `MAX_DEPTH`, we didn't find any
        # backward extremities to backfill from.
@@ -384,10 +387,7 @@ class FederationHandler:
            #   event but not anything before it. This would require looking at the
            #   state *before* the event, ignoring the special casing certain event
            #   types have.
-            if bp.type == _BackfillPointType.INSERTION_PONT:
-                event_ids_to_check = [bp.event_id]
-            else:
-                event_ids_to_check = await self.store.get_successor_events(bp.event_id)
+            event_ids_to_check = await self.store.get_successor_events(bp.event_id)

            events_to_check = await self.store.get_events_as_list(
                event_ids_to_check,
@@ -601,18 +601,6 @@ class FederationEventHandler:
                room_id, [(event, context)]
            )

-            # If we're joining the room again, check if there is new marker
-            # state indicating that there is new history imported somewhere in
-            # the DAG. Multiple markers can exist in the current state with
-            # unique state_keys.
-            #
-            # Do this after the state from the remote join was persisted (via
-            # `persist_events_and_notify`). Otherwise we can run into a
-            # situation where the create event doesn't exist yet in the
-            # `current_state_events`
-            for e in state:
-                await self._handle_marker_event(origin, e)
-
            return stream_id_after_persist

    async def update_state_for_partial_state_event(
@@ -915,13 +903,6 @@ class FederationEventHandler:
            )
        )

-        # We construct the event lists in source order from `/backfill` response because
-        # it's a) easiest, but also b) the order in which we process things matters for
-        # MSC2716 historical batches because many historical events are all at the same
-        # `depth` and we rely on the tenuous sort that the other server gave us and hope
-        # they're doing their best. The brittle nature of this ordering for historical
-        # messages over federation is one of the reasons why we don't want to continue
-        # on MSC2716 until we have online topological ordering.
        events_with_failed_pull_attempts, fresh_events = partition(
            new_events, lambda e: e.event_id in event_ids_with_failed_pull_attempts
        )
@@ -1460,8 +1441,6 @@ class FederationEventHandler:

        await self._run_push_actions_and_persist_event(event, context, backfilled)

-        await self._handle_marker_event(origin, event)
-
        if backfilled or context.rejected:
            return

@@ -1559,94 +1538,6 @@ class FederationEventHandler:
        except Exception:
            logger.exception("Failed to resync device for %s", sender)

-    @trace
-    async def _handle_marker_event(self, origin: str, marker_event: EventBase) -> None:
-        """Handles backfilling the insertion event when we receive a marker
-        event that points to one.
-
-        Args:
-            origin: Origin of the event. Will be called to get the insertion event
-            marker_event: The event to process
-        """
-
-        if marker_event.type != EventTypes.MSC2716_MARKER:
-            # Not a marker event
-            return
-
-        if marker_event.rejected_reason is not None:
-            # Rejected event
-            return
-
-        # Skip processing a marker event if the room version doesn't
-        # support it or the event is not from the room creator.
-        room_version = await self._store.get_room_version(marker_event.room_id)
-        create_event = await self._store.get_create_event_for_room(marker_event.room_id)
-        if not room_version.msc2175_implicit_room_creator:
-            room_creator = create_event.content.get(EventContentFields.ROOM_CREATOR)
-        else:
-            room_creator = create_event.sender
-        if not room_version.msc2716_historical and (
-            not self._config.experimental.msc2716_enabled
-            or marker_event.sender != room_creator
-        ):
-            return
-
-        logger.debug("_handle_marker_event: received %s", marker_event)
-
-        insertion_event_id = marker_event.content.get(
-            EventContentFields.MSC2716_INSERTION_EVENT_REFERENCE
-        )
-
-        if insertion_event_id is None:
-            # Nothing to retrieve then (invalid marker)
-            return
-
-        already_seen_insertion_event = await self._store.have_seen_event(
-            marker_event.room_id, insertion_event_id
-        )
-        if already_seen_insertion_event:
-            # No need to process a marker again if we have already seen the
-            # insertion event that it was pointing to
-            return
-
-        logger.debug(
-            "_handle_marker_event: backfilling insertion event %s", insertion_event_id
-        )
-
-        await self._get_events_and_persist(
-            origin,
-            marker_event.room_id,
-            [insertion_event_id],
-        )
-
-        insertion_event = await self._store.get_event(
-            insertion_event_id, allow_none=True
-        )
-        if insertion_event is None:
-            logger.warning(
-                "_handle_marker_event: server %s didn't return insertion event %s for marker %s",
-                origin,
-                insertion_event_id,
-                marker_event.event_id,
-            )
-            return
-
-        logger.debug(
-            "_handle_marker_event: succesfully backfilled insertion event %s from marker event %s",
-            insertion_event,
-            marker_event,
-        )
-
-        await self._store.insert_insertion_extremity(
-            insertion_event_id, marker_event.room_id
-        )
-
-        logger.debug(
-            "_handle_marker_event: insertion extremity added for %s from marker event %s",
-            insertion_event,
-            marker_event,
-        )
-
    async def backfill_event_id(
        self, destinations: List[str], room_id: str, event_id: str
    ) -> PulledPduInfo:
@@ -60,7 +60,6 @@ from synapse.replication.http.send_event import ReplicationSendEventRestServlet
 from synapse.replication.http.send_events import ReplicationSendEventsRestServlet
 from synapse.storage.databases.main.events_worker import EventRedactBehaviour
 from synapse.types import (
-    MutableStateMap,
    PersistedEventPosition,
    Requester,
    RoomAlias,
@@ -573,7 +572,6 @@ class EventCreationHandler:
        state_event_ids: Optional[List[str]] = None,
        require_consent: bool = True,
        outlier: bool = False,
-        historical: bool = False,
        depth: Optional[int] = None,
        state_map: Optional[StateMap[str]] = None,
        for_batch: bool = False,
@@ -599,7 +597,7 @@ class EventCreationHandler:
            allow_no_prev_events: Whether to allow this event to be created an empty
                list of prev_events. Normally this is prohibited just because most
                events should have a prev_event and we should only use this in special
-                cases like MSC2716.
+                cases (previously useful for MSC2716).
            prev_event_ids:
                the forward extremities to use as the prev_events for the
                new event.
@@ -614,13 +612,10 @@ class EventCreationHandler:
                If non-None, prev_event_ids must also be provided.

            state_event_ids:
-                The full state at a given event. This is used particularly by the MSC2716
-                /batch_send endpoint. One use case is with insertion events which float at
-                the beginning of a historical batch and don't have any `prev_events` to
-                derive from; we add all of these state events as the explicit state so the
-                rest of the historical batch can inherit the same state and state_group.
-                This should normally be left as None, which will cause the auth_event_ids
-                to be calculated based on the room state at the prev_events.
+                The full state at a given event. This was previously used particularly
+                by the MSC2716 /batch_send endpoint. This should normally be left as
+                None, which will cause the auth_event_ids to be calculated based on the
+                room state at the prev_events.

            require_consent: Whether to check if the requester has
                consented to the privacy policy.
@@ -629,10 +624,6 @@ class EventCreationHandler:
                it's from an arbitrary point and floating in the DAG as
                opposed to being inline with the current DAG.

-            historical: Indicates whether the message is being inserted
-                back in time around some existing events. This is used to skip
-                a few checks and mark the event as backfilled.
-
            depth: Override the depth used to order the event in the DAG.
                Should normally be set to None, which will cause the depth to be calculated
                based on the prev_events.
@@ -717,8 +708,6 @@ class EventCreationHandler:

        builder.internal_metadata.outlier = outlier

-        builder.internal_metadata.historical = historical
-
        event, unpersisted_context = await self.create_new_client_event(
            builder=builder,
            requester=requester,
@@ -947,7 +936,6 @@ class EventCreationHandler:
        txn_id: Optional[str] = None,
        ignore_shadow_ban: bool = False,
        outlier: bool = False,
-        historical: bool = False,
        depth: Optional[int] = None,
    ) -> Tuple[EventBase, int]:
        """
@@ -961,19 +949,16 @@ class EventCreationHandler:
            allow_no_prev_events: Whether to allow this event to be created an empty
                list of prev_events. Normally this is prohibited just because most
                events should have a prev_event and we should only use this in special
-                cases like MSC2716.
+                cases (previously useful for MSC2716).
            prev_event_ids:
                The event IDs to use as the prev events.
                Should normally be left as None to automatically request them
                from the database.
            state_event_ids:
-                The full state at a given event. This is used particularly by the MSC2716
-                /batch_send endpoint. One use case is with insertion events which float at
-                the beginning of a historical batch and don't have any `prev_events` to
-                derive from; we add all of these state events as the explicit state so the
-                rest of the historical batch can inherit the same state and state_group.
-                This should normally be left as None, which will cause the auth_event_ids
-                to be calculated based on the room state at the prev_events.
+                The full state at a given event. This was previously used particularly
+                by the MSC2716 /batch_send endpoint. This should normally be left as
+                None, which will cause the auth_event_ids to be calculated based on the
+                room state at the prev_events.
            ratelimit: Whether to rate limit this send.
            txn_id: The transaction ID.
            ignore_shadow_ban: True if shadow-banned users should be allowed to
@@ -981,9 +966,6 @@ class EventCreationHandler:
            outlier: Indicates whether the event is an `outlier`, i.e. if
                it's from an arbitrary point and floating in the DAG as
                opposed to being inline with the current DAG.
-            historical: Indicates whether the message is being inserted
-                back in time around some existing events. This is used to skip
-                a few checks and mark the event as backfilled.
            depth: Override the depth used to order the event in the DAG.
                Should normally be set to None, which will cause the depth to be calculated
                based on the prev_events.
@@ -1053,7 +1035,6 @@ class EventCreationHandler:
                    prev_event_ids=prev_event_ids,
                    state_event_ids=state_event_ids,
                    outlier=outlier,
-                    historical=historical,
                    depth=depth,
                )
                context = await unpersisted_context.persist(event)
@@ -1145,7 +1126,7 @@ class EventCreationHandler:
            allow_no_prev_events: Whether to allow this event to be created an empty
                list of prev_events. Normally this is prohibited just because most
                events should have a prev_event and we should only use this in special
-                cases like MSC2716.
+                cases (previously useful for MSC2716).
            prev_event_ids:
                the forward extremities to use as the prev_events for the
                new event.
@@ -1158,13 +1139,10 @@ class EventCreationHandler:
                based on the room state at the prev_events.

            state_event_ids:
-                The full state at a given event. This is used particularly by the MSC2716
-                /batch_send endpoint. One use case is with insertion events which float at
-                the beginning of a historical batch and don't have any `prev_events` to
-                derive from; we add all of these state events as the explicit state so the
-                rest of the historical batch can inherit the same state and state_group.
-                This should normally be left as None, which will cause the auth_event_ids
-                to be calculated based on the room state at the prev_events.
+                The full state at a given event. This was previously used particularly
+                by the MSC2716 /batch_send endpoint. This should normally be left as
+                None, which will cause the auth_event_ids to be calculated based on the
+                room state at the prev_events.

            depth: Override the depth used to order the event in the DAG.
                Should normally be set to None, which will cause the depth to be calculated
@@ -1261,52 +1239,6 @@ class EventCreationHandler:
            if builder.internal_metadata.outlier:
                event.internal_metadata.outlier = True
                context = EventContext.for_outlier(self._storage_controllers)
-            elif (
-                event.type == EventTypes.MSC2716_INSERTION
-                and state_event_ids
-                and builder.internal_metadata.is_historical()
-            ):
-                # Add explicit state to the insertion event so it has state to derive
-                # from even though it's floating with no `prev_events`. The rest of
-                # the batch can derive from this state and state_group.
-                #
-                # TODO(faster_joins): figure out how this works, and make sure that the
-                #   old state is complete.
-                #   https://github.com/matrix-org/synapse/issues/13003
-                metadata = await self.store.get_metadata_for_events(state_event_ids)
-
-                state_map_for_event: MutableStateMap[str] = {}
-                for state_id in state_event_ids:
-                    data = metadata.get(state_id)
-                    if data is None:
-                        # We're trying to persist a new historical batch of events
-                        # with the given state, e.g. via
-                        # `RoomBatchSendEventRestServlet`. The state can be inferred
-                        # by Synapse or set directly by the client.
-                        #
-                        # Either way, we should have persisted all the state before
-                        # getting here.
-                        raise Exception(
-                            f"State event {state_id} not found in DB,"
-                            " Synapse should have persisted it before using it."
-                        )
-
-                    if data.state_key is None:
-                        raise Exception(
-                            f"Trying to set non-state event {state_id} as state"
-                        )
-
-                    state_map_for_event[(data.event_type, data.state_key)] = state_id
-
-                # TODO(faster_joins): check how MSC2716 works and whether we can have
-                #   partial state here
-                #   https://github.com/matrix-org/synapse/issues/13003
-                context = await self.state.calculate_context_info(
-                    event,
-                    state_ids_before_event=state_map_for_event,
-                    partial_state=False,
-                )
-
            else:
                context = await self.state.calculate_context_info(event)

@@ -1876,28 +1808,6 @@ class EventCreationHandler:
                            403, "Redacting server ACL events is not permitted"
                        )

-                    # Add a little safety stop-gap to prevent people from trying to
-                    # redact MSC2716 related events when they're in a room version
-                    # which does not support it yet. We allow people to use MSC2716
-                    # events in existing room versions but only from the room
-                    # creator since it does not require any changes to the auth
-                    # rules and in effect, the redaction algorithm . In the
-                    # supported room version, we add the `historical` power level to
-                    # auth the MSC2716 related events and adjust the redaction
-                    # algorthim to keep the `historical` field around (redacting an
-                    # event should only strip fields which don't affect the
-                    # structural protocol level).
-                    is_msc2716_event = (
-                        original_event.type == EventTypes.MSC2716_INSERTION
-                        or original_event.type == EventTypes.MSC2716_BATCH
-                        or original_event.type == EventTypes.MSC2716_MARKER
-                    )
-                    if not room_version_obj.msc2716_historical and is_msc2716_event:
-                        raise AuthError(
-                            403,
-                            "Redacting MSC2716 events is not supported in this room version",
-                        )
-
                event_types = event_auth.auth_types_for_event(event.room_version, event)
                prev_state_ids = await context.get_prev_state_ids(
                    StateFilter.from_types(event_types)
@@ -1935,58 +1845,12 @@ class EventCreationHandler:
                if prev_state_ids:
                    raise AuthError(403, "Changing the room create event is forbidden")

-            if event.type == EventTypes.MSC2716_INSERTION:
-                room_version = await self.store.get_room_version_id(event.room_id)
-                room_version_obj = KNOWN_ROOM_VERSIONS[room_version]
-
-                create_event = await self.store.get_create_event_for_room(event.room_id)
-                if not room_version_obj.msc2175_implicit_room_creator:
-                    room_creator = create_event.content.get(
-                        EventContentFields.ROOM_CREATOR
-                    )
-                else:
-                    room_creator = create_event.sender
-
-                # Only check an insertion event if the room version
-                # supports it or the event is from the room creator.
-                if room_version_obj.msc2716_historical or (
-                    self.config.experimental.msc2716_enabled
-                    and event.sender == room_creator
-                ):
-                    next_batch_id = event.content.get(
-                        EventContentFields.MSC2716_NEXT_BATCH_ID
-                    )
-                    conflicting_insertion_event_id = None
-                    if next_batch_id:
-                        conflicting_insertion_event_id = (
-                            await self.store.get_insertion_event_id_by_batch_id(
-                                event.room_id, next_batch_id
-                            )
-                        )
-                    if conflicting_insertion_event_id is not None:
-                        # The current insertion event that we're processing is invalid
-                        # because an insertion event already exists in the room with the
-                        # same next_batch_id. We can't allow multiple because the batch
-                        # pointing will get weird, e.g. we can't determine which insertion
-                        # event the batch event is pointing to.
-                        raise SynapseError(
-                            HTTPStatus.BAD_REQUEST,
-                            "Another insertion event already exists with the same next_batch_id",
-                            errcode=Codes.INVALID_PARAM,
-                        )
-
-            # Mark any `m.historical` messages as backfilled so they don't appear
-            # in `/sync` and have the proper decrementing `stream_ordering` as we import
-            backfilled = False
-            if event.internal_metadata.is_historical():
-                backfilled = True
-
        assert self._storage_controllers.persistence is not None
        (
            persisted_events,
            max_stream_token,
        ) = await self._storage_controllers.persistence.persist_events(
-            events_and_context, backfilled=backfilled
+            events_and_context,
        )

        events_and_pos = []
@@ -1354,7 +1354,7 @@ class OidcProvider:
        finish_request(request)


-class LogoutToken(JWTClaims):
+class LogoutToken(JWTClaims):  # type: ignore[misc]
    """
    Holds and verify claims of a logout token, as per
    https://openid.net/specs/openid-connect-backchannel-1_0.html#LogoutToken
@@ -40,6 +40,11 @@ if TYPE_CHECKING:

 logger = logging.getLogger(__name__)

+# How many single event gaps we tolerate returning in a `/messages` response before we
+# backfill and try to fill in the history. This is an arbitrarily picked number so feel
+# free to tune it in the future.
+BACKFILL_BECAUSE_TOO_MANY_GAPS_THRESHOLD = 3
+

@attr.s(slots=True, auto_attribs=True)
 class PurgeStatus:
@@ -360,7 +365,7 @@ class PaginationHandler:
        except Exception:
            f = Failure()
            logger.error(
-                "[purge] failed", exc_info=(f.type, f.value, f.getTracebackObject())  # type: ignore
+                "[purge] failed", exc_info=(f.type, f.value, f.getTracebackObject())
            )
            self._purges_by_id[purge_id].status = PurgeStatus.STATUS_FAILED
            self._purges_by_id[purge_id].error = f.getErrorMessage()
@@ -486,35 +491,35 @@ class PaginationHandler:
                        room_id, room_token.stream
                    )

-                if not use_admin_priviledge and membership == Membership.LEAVE:
-                    # If they have left the room then clamp the token to be before
-                    # they left the room, to save the effort of loading from the
-                    # database.
+            # If they have left the room then clamp the token to be before
+            # they left the room, to save the effort of loading from the
+            # database.
+            if (
+                pagin_config.direction == Direction.BACKWARDS
+                and not use_admin_priviledge
+                and membership == Membership.LEAVE
+            ):
+                # This is only None if the room is world_readable, in which case
+                # "Membership.JOIN" would have been returned and we should never hit
+                # this branch.
+                assert member_event_id

-                    # This is only None if the room is world_readable, in which
-                    # case "JOIN" would have been returned.
-                    assert member_event_id
-
-                    leave_token = await self.store.get_topological_token_for_event(
-                        member_event_id
-                    )
-                    assert leave_token.topological is not None
-
-                    if leave_token.topological < curr_topo:
-                        from_token = from_token.copy_and_replace(
-                            StreamKeyType.ROOM, leave_token
-                        )
-
-                await self.hs.get_federation_handler().maybe_backfill(
-                    room_id,
-                    curr_topo,
-                    limit=pagin_config.limit,
+                leave_token = await self.store.get_topological_token_for_event(
+                    member_event_id
                )
+                assert leave_token.topological is not None
+
+                if leave_token.topological < curr_topo:
+                    from_token = from_token.copy_and_replace(
+                        StreamKeyType.ROOM, leave_token
+                    )

            to_room_key = None
            if pagin_config.to_token:
                to_room_key = pagin_config.to_token.room_key

+            # Initially fetch the events from the database. With any luck, we can return
+            # these without blocking on backfill (handled below).
            events, next_key = await self.store.paginate_room_events(
                room_id=room_id,
                from_key=from_token.room_key,
@@ -524,6 +529,94 @@ class PaginationHandler:
                event_filter=event_filter,
            )

+            if pagin_config.direction == Direction.BACKWARDS:
+                # We use a `Set` because there can be multiple events at a given depth
+                # and we only care about looking at the unique continum of depths to
+                # find gaps.
+                event_depths: Set[int] = {event.depth for event in events}
+                sorted_event_depths = sorted(event_depths)
+
+                # Inspect the depths of the returned events to see if there are any gaps
+                found_big_gap = False
+                number_of_gaps = 0
+                previous_event_depth = (
+                    sorted_event_depths[0] if len(sorted_event_depths) > 0 else 0
+                )
+                for event_depth in sorted_event_depths:
+                    # We don't expect a negative depth but we'll just deal with it in
+                    # any case by taking the absolute value to get the true gap between
+                    # any two integers.
+                    depth_gap = abs(event_depth - previous_event_depth)
+                    # A `depth_gap` of 1 is a normal continuous chain to the next event
+                    # (1 <-- 2 <-- 3) so anything larger indicates a missing event (it's
+                    # also possible there is no event at a given depth but we can't ever
+                    # know that for sure)
+                    if depth_gap > 1:
+                        number_of_gaps += 1
+
+                    # We only tolerate a small number single-event long gaps in the
+                    # returned events because those are most likely just events we've
+                    # failed to pull in the past. Anything longer than that is probably
+                    # a sign that we're missing a decent chunk of history and we should
+                    # try to backfill it.
+                    #
+                    # XXX: It's possible we could tolerate longer gaps if we checked
+                    # that a given events `prev_events` is one that has failed pull
+                    # attempts and we could just treat it like a dead branch of history
+                    # for now or at least something that we don't need the block the
+                    # client on to try pulling.
+                    #
+                    # XXX: If we had something like MSC3871 to indicate gaps in the
+                    # timeline to the client, we could also get away with any sized gap
+                    # and just have the client refetch the holes as they see fit.
+                    if depth_gap > 2:
+                        found_big_gap = True
+                        break
+                    previous_event_depth = event_depth
+
+                # Backfill in the foreground if we found a big gap, have too many holes,
+                # or we don't have enough events to fill the limit that the client asked
+                # for.
+                missing_too_many_events = (
+                    number_of_gaps > BACKFILL_BECAUSE_TOO_MANY_GAPS_THRESHOLD
+                )
+                not_enough_events_to_fill_response = len(events) < pagin_config.limit
+                if (
+                    found_big_gap
+                    or missing_too_many_events
+                    or not_enough_events_to_fill_response
+                ):
+                    did_backfill = (
+                        await self.hs.get_federation_handler().maybe_backfill(
+                            room_id,
+                            curr_topo,
+                            limit=pagin_config.limit,
+                        )
+                    )
+
+                    # If we did backfill something, refetch the events from the database to
+                    # catch anything new that might have been added since we last fetched.
+                    if did_backfill:
+                        events, next_key = await self.store.paginate_room_events(
+                            room_id=room_id,
+                            from_key=from_token.room_key,
+                            to_key=to_room_key,
+                            direction=pagin_config.direction,
+                            limit=pagin_config.limit,
+                            event_filter=event_filter,
+                        )
+                else:
+                    # Otherwise, we can backfill in the background for eventual
+                    # consistency's sake but we don't need to block the client waiting
+                    # for a costly federation call and processing.
+                    run_as_background_process(
+                        "maybe_backfill_in_the_background",
+                        self.hs.get_federation_handler().maybe_backfill,
+                        room_id,
+                        curr_topo,
+                        limit=pagin_config.limit,
+                    )
+
            next_token = from_token.copy_and_replace(StreamKeyType.ROOM, next_key)

        # if no events are returned from pagination, that implies
@@ -689,7 +782,7 @@ class PaginationHandler:
            f = Failure()
            logger.error(
                "failed",
-                exc_info=(f.type, f.value, f.getTracebackObject()),  # type: ignore
+                exc_info=(f.type, f.value, f.getTracebackObject()),
            )
            self._delete_by_id[delete_id].status = DeleteStatus.STATUS_FAILED
            self._delete_by_id[delete_id].error = f.getErrorMessage()
@@ -648,7 +648,6 @@ class PresenceHandler(BasePresenceHandler):
    def __init__(self, hs: "HomeServer"):
        super().__init__(hs)
        self.hs = hs
-        self.server_name = hs.hostname
        self.wheel_timer: WheelTimer[str] = WheelTimer()
        self.notifier = hs.get_notifier()
        self._presence_enabled = hs.config.server.use_presence
@@ -67,7 +67,7 @@ class ProfileHandler:
        target_user = UserID.from_string(user_id)

        if self.hs.is_mine(target_user):
-            profileinfo = await self.store.get_profileinfo(target_user.localpart)
+            profileinfo = await self.store.get_profileinfo(target_user)
            if profileinfo.display_name is None:
                raise SynapseError(404, "Profile was not found", Codes.NOT_FOUND)

@@ -99,9 +99,7 @@ class ProfileHandler:
    async def get_displayname(self, target_user: UserID) -> Optional[str]:
        if self.hs.is_mine(target_user):
            try:
-                displayname = await self.store.get_profile_displayname(
-                    target_user.localpart
-                )
+                displayname = await self.store.get_profile_displayname(target_user)
            except StoreError as e:
                if e.code == 404:
                    raise SynapseError(404, "Profile was not found", Codes.NOT_FOUND)
@@ -147,7 +145,7 @@ class ProfileHandler:
            raise AuthError(400, "Cannot set another user's displayname")

        if not by_admin and not self.hs.config.registration.enable_set_displayname:
-            profile = await self.store.get_profileinfo(target_user.localpart)
+            profile = await self.store.get_profileinfo(target_user)
            if profile.display_name:
                raise SynapseError(
                    400,
@@ -180,7 +178,7 @@ class ProfileHandler:

        await self.store.set_profile_displayname(target_user, displayname_to_set)

-        profile = await self.store.get_profileinfo(target_user.localpart)
+        profile = await self.store.get_profileinfo(target_user)
        await self.user_directory_handler.handle_local_profile_change(
            target_user.to_string(), profile
        )
@@ -194,9 +192,7 @@ class ProfileHandler:
    async def get_avatar_url(self, target_user: UserID) -> Optional[str]:
        if self.hs.is_mine(target_user):
            try:
-                avatar_url = await self.store.get_profile_avatar_url(
-                    target_user.localpart
-                )
+                avatar_url = await self.store.get_profile_avatar_url(target_user)
            except StoreError as e:
                if e.code == 404:
                    raise SynapseError(404, "Profile was not found", Codes.NOT_FOUND)
@@ -241,7 +237,7 @@ class ProfileHandler:
            raise AuthError(400, "Cannot set another user's avatar_url")

        if not by_admin and not self.hs.config.registration.enable_set_avatar_url:
-            profile = await self.store.get_profileinfo(target_user.localpart)
+            profile = await self.store.get_profileinfo(target_user)
            if profile.avatar_url:
                raise SynapseError(
                    400, "Changing avatar is disabled on this server", Codes.FORBIDDEN
@@ -272,7 +268,7 @@ class ProfileHandler:

        await self.store.set_profile_avatar_url(target_user, avatar_url_to_set)

-        profile = await self.store.get_profileinfo(target_user.localpart)
+        profile = await self.store.get_profileinfo(target_user)
        await self.user_directory_handler.handle_local_profile_change(
            target_user.to_string(), profile
        )
@@ -369,14 +365,10 @@ class ProfileHandler:
        response = {}
        try:
            if just_field is None or just_field == "displayname":
-                response["displayname"] = await self.store.get_profile_displayname(
-                    user.localpart
-                )
+                response["displayname"] = await self.store.get_profile_displayname(user)

            if just_field is None or just_field == "avatar_url":
-                response["avatar_url"] = await self.store.get_profile_avatar_url(
-                    user.localpart
-                )
+                response["avatar_url"] = await self.store.get_profile_avatar_url(user)
        except StoreError as e:
            if e.code == 404:
                raise SynapseError(404, "Profile was not found", Codes.NOT_FOUND)
@@ -27,7 +27,6 @@ logger = logging.getLogger(__name__)

 class ReadMarkerHandler:
    def __init__(self, hs: "HomeServer"):
-        self.server_name = hs.config.server.server_name
        self.store = hs.get_datastores().main
        self.account_data_handler = hs.get_account_data_handler()
        self.read_marker_linearizer = Linearizer(name="read_marker")
@@ -315,7 +315,7 @@ class RegistrationHandler:
                approved=approved,
            )

-            profile = await self.store.get_profileinfo(localpart)
+            profile = await self.store.get_profileinfo(user)
            await self.user_directory_handler.handle_local_profile_change(
                user_id, profile
            )
@@ -205,16 +205,22 @@ class RelationsHandler:
            event_id: The event IDs to look and redact relations of.
            initial_redaction_event: The redaction for the event referred to by
                event_id.
-            relation_types: The types of relations to look for.
+            relation_types: The types of relations to look for. If "*" is in the list,
+                all related events will be redacted regardless of the type.

        Raises:
            ShadowBanError if the requester is shadow-banned
        """
-        related_event_ids = (
-            await self._main_store.get_all_relations_for_event_with_types(
-                event_id, relation_types
+        if "*" in relation_types:
+            related_event_ids = await self._main_store.get_all_relations_for_event(
+                event_id
+            )
+        else:
+            related_event_ids = (
+                await self._main_store.get_all_relations_for_event_with_types(
+                    event_id, relation_types
+                )
            )
-        )

        for related_event_id in related_event_ids:
            try:
@@ -872,6 +872,8 @@ class RoomCreationHandler:
        visibility = config.get("visibility", "private")
        is_public = visibility == "public"

+        self._validate_room_config(config, visibility)
+
        room_id = await self._generate_and_create_room_id(
            creator_id=user_id,
            is_public=is_public,
@@ -1111,20 +1113,7 @@ class RoomCreationHandler:

            return new_event, new_unpersisted_context

-        visibility = room_config.get("visibility", "private")
-        preset_config = room_config.get(
-            "preset",
-            RoomCreationPreset.PRIVATE_CHAT
-            if visibility == "private"
-            else RoomCreationPreset.PUBLIC_CHAT,
-        )
-
-        try:
-            config = self._presets_dict[preset_config]
-        except KeyError:
-            raise SynapseError(
-                400, f"'{preset_config}' is not a valid preset", errcode=Codes.BAD_JSON
-            )
+        preset_config, config = self._room_preset_config(room_config)

        # MSC2175 removes the creator field from the create event.
        if not room_version.msc2175_implicit_room_creator:
@@ -1306,6 +1295,65 @@ class RoomCreationHandler:
        assert last_event.internal_metadata.stream_ordering is not None
        return last_event.internal_metadata.stream_ordering, last_event.event_id, depth

+    def _validate_room_config(
+        self,
+        config: JsonDict,
+        visibility: str,
+    ) -> None:
+        """Checks configuration parameters for a /createRoom request.
+
+        If validation detects invalid parameters an exception may be raised to
+        cause room creation to be aborted and an error response to be returned
+        to the client.
+
+        Args:
+            config: A dict of configuration options. Originally from the body of
+                the /createRoom request
+            visibility: One of "public" or "private"
+        """
+
+        # Validate the requested preset, raise a 400 error if not valid
+        preset_name, preset_config = self._room_preset_config(config)
+
+        # If the user is trying to create an encrypted room and this is forbidden
+        # by the configured default_power_level_content_override, then reject the
+        # request before the room is created.
+        raw_initial_state = config.get("initial_state", [])
+        room_encryption_event = any(
+            s.get("type", "") == EventTypes.RoomEncryption for s in raw_initial_state
+        )
+
+        if preset_config["encrypted"] or room_encryption_event:
+            if self._default_power_level_content_override:
+                override = self._default_power_level_content_override.get(preset_name)
+                if override is not None:
+                    event_levels = override.get("events", {})
+                    room_admin_level = event_levels.get(EventTypes.PowerLevels, 100)
+                    encryption_level = event_levels.get(EventTypes.RoomEncryption, 100)
+                    if encryption_level > room_admin_level:
+                        raise SynapseError(
+                            403,
+                            f"You cannot create an encrypted room. user_level ({room_admin_level}) < send_level ({encryption_level})",
+                        )
+
+    def _room_preset_config(self, room_config: JsonDict) -> Tuple[str, dict]:
+        # The spec says rooms should default to private visibility if
+        # `visibility` is not specified.
+        visibility = room_config.get("visibility", "private")
+        preset_name = room_config.get(
+            "preset",
+            RoomCreationPreset.PRIVATE_CHAT
+            if visibility == "private"
+            else RoomCreationPreset.PUBLIC_CHAT,
+        )
+        try:
+            preset_config = self._presets_dict[preset_name]
+        except KeyError:
+            raise SynapseError(
+                400, f"'{preset_name}' is not a valid preset", errcode=Codes.BAD_JSON
+            )
+        return preset_name, preset_config
+
    def _generate_room_id(self) -> str:
        """Generates a random room ID.

@@ -1490,7 +1538,6 @@ class RoomContextHandler:

 class TimestampLookupHandler:
    def __init__(self, hs: "HomeServer"):
-        self.server_name = hs.hostname
        self.store = hs.get_datastores().main
        self.state_handler = hs.get_state_handler()
        self.federation_client = hs.get_federation_client()
@@ -1,466 +0,0 @@
-import logging
-from typing import TYPE_CHECKING, List, Tuple
-
-from synapse.api.constants import EventContentFields, EventTypes
-from synapse.appservice import ApplicationService
-from synapse.http.servlet import assert_params_in_dict
-from synapse.types import JsonDict, Requester, UserID, create_requester
-from synapse.util.stringutils import random_string
-
-if TYPE_CHECKING:
-    from synapse.server import HomeServer
-
-logger = logging.getLogger(__name__)
-
-
-class RoomBatchHandler:
-    def __init__(self, hs: "HomeServer"):
-        self.hs = hs
-        self.store = hs.get_datastores().main
-        self._state_storage_controller = hs.get_storage_controllers().state
-        self.event_creation_handler = hs.get_event_creation_handler()
-        self.room_member_handler = hs.get_room_member_handler()
-        self.auth = hs.get_auth()
-
-    async def inherit_depth_from_prev_ids(self, prev_event_ids: List[str]) -> int:
-        """Finds the depth which would sort it after the most-recent
-        prev_event_id but before the successors of those events. If no
-        successors are found, we assume it's an historical extremity part of the
-        current batch and use the same depth of the prev_event_ids.
-
-        Args:
-            prev_event_ids: List of prev event IDs
-
-        Returns:
-            Inherited depth
-        """
-        (
-            most_recent_prev_event_id,
-            most_recent_prev_event_depth,
-        ) = await self.store.get_max_depth_of(prev_event_ids)
-
-        # We want to insert the historical event after the `prev_event` but before the successor event
-        #
-        # We inherit depth from the successor event instead of the `prev_event`
-        # because events returned from `/messages` are first sorted by `topological_ordering`
-        # which is just the `depth` and then tie-break with `stream_ordering`.
-        #
-        # We mark these inserted historical events as "backfilled" which gives them a
-        # negative `stream_ordering`. If we use the same depth as the `prev_event`,
-        # then our historical event will tie-break and be sorted before the `prev_event`
-        # when it should come after.
-        #
-        # We want to use the successor event depth so they appear after `prev_event` because
-        # it has a larger `depth` but before the successor event because the `stream_ordering`
-        # is negative before the successor event.
-        assert most_recent_prev_event_id is not None
-        successor_event_ids = await self.store.get_successor_events(
-            most_recent_prev_event_id
-        )
-
-        # If we can't find any successor events, then it's a forward extremity of
-        # historical messages and we can just inherit from the previous historical
-        # event which we can already assume has the correct depth where we want
-        # to insert into.
-        if not successor_event_ids:
-            depth = most_recent_prev_event_depth
-        else:
-            (
-                _,
-                oldest_successor_depth,
-            ) = await self.store.get_min_depth_of(successor_event_ids)
-
-            depth = oldest_successor_depth
-
-        return depth
-
-    def create_insertion_event_dict(
-        self, sender: str, room_id: str, origin_server_ts: int
-    ) -> JsonDict:
-        """Creates an event dict for an "insertion" event with the proper fields
-        and a random batch ID.
-
-        Args:
-            sender: The event author MXID
-            room_id: The room ID that the event belongs to
-            origin_server_ts: Timestamp when the event was sent
-
-        Returns:
-            The new event dictionary to insert.
-        """
-
-        next_batch_id = random_string(8)
-        insertion_event = {
-            "type": EventTypes.MSC2716_INSERTION,
-            "sender": sender,
-            "room_id": room_id,
-            "content": {
-                EventContentFields.MSC2716_NEXT_BATCH_ID: next_batch_id,
-                EventContentFields.MSC2716_HISTORICAL: True,
-            },
-            "origin_server_ts": origin_server_ts,
-        }
-
-        return insertion_event
-
-    async def create_requester_for_user_id_from_app_service(
-        self, user_id: str, app_service: ApplicationService
-    ) -> Requester:
-        """Creates a new requester for the given user_id
-        and validates that the app service is allowed to control
-        the given user.
-
-        Args:
-            user_id: The author MXID that the app service is controlling
-            app_service: The app service that controls the user
-
-        Returns:
-            Requester object
-        """
-
-        await self.auth.validate_appservice_can_control_user_id(app_service, user_id)
-
-        return create_requester(user_id, app_service=app_service)
-
-    async def get_most_recent_full_state_ids_from_event_id_list(
-        self, event_ids: List[str]
-    ) -> List[str]:
-        """Find the most recent event_id and grab the full state at that event.
-        We will use this as a base to auth our historical messages against.
-
-        Args:
-            event_ids: List of event ID's to look at
-
-        Returns:
-            List of event ID's
-        """
-
-        (
-            most_recent_event_id,
-            _,
-        ) = await self.store.get_max_depth_of(event_ids)
-        # mapping from (type, state_key) -> state_event_id
-        assert most_recent_event_id is not None
-        prev_state_map = await self._state_storage_controller.get_state_ids_for_event(
-            most_recent_event_id
-        )
-        # List of state event ID's
-        full_state_ids = list(prev_state_map.values())
-
-        return full_state_ids
-
-    async def persist_state_events_at_start(
-        self,
-        state_events_at_start: List[JsonDict],
-        room_id: str,
-        initial_state_event_ids: List[str],
-        app_service_requester: Requester,
-    ) -> List[str]:
-        """Takes all `state_events_at_start` event dictionaries and creates/persists
-        them in a floating state event chain which don't resolve into the current room
-        state. They are floating because they reference no prev_events which disconnects
-        them from the normal DAG.
-
-        Args:
-            state_events_at_start:
-            room_id: Room where you want the events persisted in.
-            initial_state_event_ids:
-                The base set of state for the historical batch which the floating
-                state chain will derive from. This should probably be the state
-                from the `prev_event` defined by `/batch_send?prev_event_id=$abc`.
-            app_service_requester: The requester of an application service.
-
-        Returns:
-            List of state event ID's we just persisted
-        """
-        assert app_service_requester.app_service
-
-        state_event_ids_at_start = []
-        state_event_ids = initial_state_event_ids.copy()
-
-        # Make the state events float off on their own by specifying no
-        # prev_events for the first one in the chain so we don't have a bunch of
-        # `@mxid joined the room` noise between each batch.
-        prev_event_ids_for_state_chain: List[str] = []
-
-        for index, state_event in enumerate(state_events_at_start):
-            assert_params_in_dict(
-                state_event, ["type", "origin_server_ts", "content", "sender"]
-            )
-
-            logger.debug(
-                "RoomBatchSendEventRestServlet inserting state_event=%s", state_event
-            )
-
-            event_dict = {
-                "type": state_event["type"],
-                "origin_server_ts": state_event["origin_server_ts"],
-                "content": state_event["content"],
-                "room_id": room_id,
-                "sender": state_event["sender"],
-                "state_key": state_event["state_key"],
-            }
-
-            # Mark all events as historical
-            event_dict["content"][EventContentFields.MSC2716_HISTORICAL] = True
-
-            # TODO: This is pretty much the same as some other code to handle inserting state in this file
-            if event_dict["type"] == EventTypes.Member:
-                membership = event_dict["content"].get("membership", None)
-                event_id, _ = await self.room_member_handler.update_membership(
-                    await self.create_requester_for_user_id_from_app_service(
-                        state_event["sender"], app_service_requester.app_service
-                    ),
-                    target=UserID.from_string(event_dict["state_key"]),
-                    room_id=room_id,
-                    action=membership,
-                    content=event_dict["content"],
-                    historical=True,
-                    # Only the first event in the state chain should be floating.
-                    # The rest should hang off each other in a chain.
-                    allow_no_prev_events=index == 0,
-                    prev_event_ids=prev_event_ids_for_state_chain,
-                    # The first event in the state chain is floating with no
-                    # `prev_events` which means it can't derive state from
-                    # anywhere automatically. So we need to set some state
-                    # explicitly.
-                    #
-                    # Make sure to use a copy of this list because we modify it
-                    # later in the loop here. Otherwise it will be the same
-                    # reference and also update in the event when we append
-                    # later.
-                    state_event_ids=state_event_ids.copy(),
-                )
-            else:
-                (
-                    event,
-                    _,
-                ) = await self.event_creation_handler.create_and_send_nonmember_event(
-                    await self.create_requester_for_user_id_from_app_service(
-                        state_event["sender"], app_service_requester.app_service
-                    ),
-                    event_dict,
-                    historical=True,
-                    # Only the first event in the state chain should be floating.
-                    # The rest should hang off each other in a chain.
-                    allow_no_prev_events=index == 0,
-                    prev_event_ids=prev_event_ids_for_state_chain,
-                    # The first event in the state chain is floating with no
-                    # `prev_events` which means it can't derive state from
-                    # anywhere automatically. So we need to set some state
-                    # explicitly.
-                    #
-                    # Make sure to use a copy of this list because we modify it
-                    # later in the loop here. Otherwise it will be the same
-                    # reference and also update in the event when we append later.
-                    state_event_ids=state_event_ids.copy(),
-                )
-                event_id = event.event_id
-
-            state_event_ids_at_start.append(event_id)
-            state_event_ids.append(event_id)
-            # Connect all the state in a floating chain
-            prev_event_ids_for_state_chain = [event_id]
-
-        return state_event_ids_at_start
-
-    async def persist_historical_events(
-        self,
-        events_to_create: List[JsonDict],
-        room_id: str,
-        inherited_depth: int,
-        initial_state_event_ids: List[str],
-        app_service_requester: Requester,
-    ) -> List[str]:
-        """Create and persists all events provided sequentially. Handles the
-        complexity of creating events in chronological order so they can
-        reference each other by prev_event but still persists in
-        reverse-chronoloical order so they have the correct
-        (topological_ordering, stream_ordering) and sort correctly from
-        /messages.
-
-        Args:
-            events_to_create: List of historical events to create in JSON
-                dictionary format.
-            room_id: Room where you want the events persisted in.
-            inherited_depth: The depth to create the events at (you will
-                probably by calling inherit_depth_from_prev_ids(...)).
-            initial_state_event_ids:
-                This is used to set explicit state for the insertion event at
-                the start of the historical batch since it's floating with no
-                prev_events to derive state from automatically.
-            app_service_requester: The requester of an application service.
-
-        Returns:
-            List of persisted event IDs
-        """
-        assert app_service_requester.app_service
-
-        # We expect the first event in a historical batch to be an insertion event
-        assert events_to_create[0]["type"] == EventTypes.MSC2716_INSERTION
-        # We expect the last event in a historical batch to be an batch event
-        assert events_to_create[-1]["type"] == EventTypes.MSC2716_BATCH
-
-        # Make the historical event chain float off on its own by specifying no
-        # prev_events for the first event in the chain which causes the HS to
-        # ask for the state at the start of the batch later.
-        prev_event_ids: List[str] = []
-
-        event_ids = []
-        events_to_persist = []
-        for index, ev in enumerate(events_to_create):
-            assert_params_in_dict(ev, ["type", "origin_server_ts", "content", "sender"])
-
-            assert self.hs.is_mine_id(ev["sender"]), "User must be our own: %s" % (
-                ev["sender"],
-            )
-
-            event_dict = {
-                "type": ev["type"],
-                "origin_server_ts": ev["origin_server_ts"],
-                "content": ev["content"],
-                "room_id": room_id,
-                "sender": ev["sender"],  # requester.user.to_string(),
-                "prev_events": prev_event_ids.copy(),
-            }
-
-            # Mark all events as historical
-            event_dict["content"][EventContentFields.MSC2716_HISTORICAL] = True
-
-            event, unpersisted_context = await self.event_creation_handler.create_event(
-                await self.create_requester_for_user_id_from_app_service(
-                    ev["sender"], app_service_requester.app_service
-                ),
-                event_dict,
-                # Only the first event (which is the insertion event) in the
-                # chain should be floating. The rest should hang off each other
-                # in a chain.
-                allow_no_prev_events=index == 0,
-                prev_event_ids=event_dict.get("prev_events"),
-                # Since the first event (which is the insertion event) in the
-                # chain is floating with no `prev_events`, it can't derive state
-                # from anywhere automatically. So we need to set some state
-                # explicitly.
-                state_event_ids=initial_state_event_ids if index == 0 else None,
-                historical=True,
-                depth=inherited_depth,
-            )
-            context = await unpersisted_context.persist(event)
-            assert context._state_group
-
-            # Normally this is done when persisting the event but we have to
-            # pre-emptively do it here because we create all the events first,
-            # then persist them in another pass below. And we want to share
-            # state_groups across the whole batch so this lookup needs to work
-            # for the next event in the batch in this loop.
-            await self.store.store_state_group_id_for_event_id(
-                event_id=event.event_id,
-                state_group_id=context._state_group,
-            )
-
-            logger.debug(
-                "RoomBatchSendEventRestServlet inserting event=%s, prev_event_ids=%s",
-                event,
-                prev_event_ids,
-            )
-
-            events_to_persist.append((event, context))
-            event_id = event.event_id
-
-            event_ids.append(event_id)
-            prev_event_ids = [event_id]
-
-        # Persist events in reverse-chronological order so they have the
-        # correct stream_ordering as they are backfilled (which decrements).
-        # Events are sorted by (topological_ordering, stream_ordering)
-        # where topological_ordering is just depth.
-        for event, context in reversed(events_to_persist):
-            # This call can't raise `PartialStateConflictError` since we forbid
-            # use of the historical batch API during partial state
-            await self.event_creation_handler.handle_new_client_event(
-                await self.create_requester_for_user_id_from_app_service(
-                    event.sender, app_service_requester.app_service
-                ),
-                events_and_context=[(event, context)],
-            )
-
-        return event_ids
-
-    async def handle_batch_of_events(
-        self,
-        events_to_create: List[JsonDict],
-        room_id: str,
-        batch_id_to_connect_to: str,
-        inherited_depth: int,
-        initial_state_event_ids: List[str],
-        app_service_requester: Requester,
-    ) -> Tuple[List[str], str]:
-        """
-        Handles creating and persisting all of the historical events as well as
-        insertion and batch meta events to make the batch navigable in the DAG.
-
-        Args:
-            events_to_create: List of historical events to create in JSON
-                dictionary format.
-            room_id: Room where you want the events created in.
-            batch_id_to_connect_to: The batch_id from the insertion event you
-                want this batch to connect to.
-            inherited_depth: The depth to create the events at (you will
-                probably by calling inherit_depth_from_prev_ids(...)).
-            initial_state_event_ids:
-                This is used to set explicit state for the insertion event at
-                the start of the historical batch since it's floating with no
-                prev_events to derive state from automatically. This should
-                probably be the state from the `prev_event` defined by
-                `/batch_send?prev_event_id=$abc` plus the outcome of
-                `persist_state_events_at_start`
-            app_service_requester: The requester of an application service.
-
-        Returns:
-            Tuple containing a list of created events and the next_batch_id
-        """
-
-        # Connect this current batch to the insertion event from the previous batch
-        last_event_in_batch = events_to_create[-1]
-        batch_event = {
-            "type": EventTypes.MSC2716_BATCH,
-            "sender": app_service_requester.user.to_string(),
-            "room_id": room_id,
-            "content": {
-                EventContentFields.MSC2716_BATCH_ID: batch_id_to_connect_to,
-                EventContentFields.MSC2716_HISTORICAL: True,
-            },
-            # Since the batch event is put at the end of the batch,
-            # where the newest-in-time event is, copy the origin_server_ts from
-            # the last event we're inserting
-            "origin_server_ts": last_event_in_batch["origin_server_ts"],
-        }
-        # Add the batch event to the end of the batch (newest-in-time)
-        events_to_create.append(batch_event)
-
-        # Add an "insertion" event to the start of each batch (next to the oldest-in-time
-        # event in the batch) so the next batch can be connected to this one.
-        insertion_event = self.create_insertion_event_dict(
-            sender=app_service_requester.user.to_string(),
-            room_id=room_id,
-            # Since the insertion event is put at the start of the batch,
-            # where the oldest-in-time event is, copy the origin_server_ts from
-            # the first event we're inserting
-            origin_server_ts=events_to_create[0]["origin_server_ts"],
-        )
-        next_batch_id = insertion_event["content"][
-            EventContentFields.MSC2716_NEXT_BATCH_ID
-        ]
-        # Prepend the insertion event to the start of the batch (oldest-in-time)
-        events_to_create = [insertion_event] + events_to_create
-
-        # Create and persist all of the historical events
-        event_ids = await self.persist_historical_events(
-            events_to_create=events_to_create,
-            room_id=room_id,
-            inherited_depth=inherited_depth,
-            initial_state_event_ids=initial_state_event_ids,
-            app_service_requester=app_service_requester,
-        )
-
-        return event_ids, next_batch_id
@@ -362,7 +362,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        content: Optional[dict] = None,
        require_consent: bool = True,
        outlier: bool = False,
-        historical: bool = False,
        origin_server_ts: Optional[int] = None,
    ) -> Tuple[str, int]:
        """
@@ -378,16 +377,13 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
            allow_no_prev_events: Whether to allow this event to be created an empty
                list of prev_events. Normally this is prohibited just because most
                events should have a prev_event and we should only use this in special
-                cases like MSC2716.
+                cases (previously useful for MSC2716).
            prev_event_ids: The event IDs to use as the prev events
            state_event_ids:
-                The full state at a given event. This is used particularly by the MSC2716
-                /batch_send endpoint. One use case is the historical `state_events_at_start`;
-                since each is marked as an `outlier`, the `EventContext.for_outlier()` won't
-                have any `state_ids` set and therefore can't derive any state even though the
-                prev_events are set so we need to set them ourself via this argument.
-                This should normally be left as None, which will cause the auth_event_ids
-                to be calculated based on the room state at the prev_events.
+                The full state at a given event. This was previously used particularly
+                by the MSC2716 /batch_send endpoint. This should normally be left as
+                None, which will cause the auth_event_ids to be calculated based on the
+                room state at the prev_events.
            depth: Override the depth used to order the event in the DAG.
                Should normally be set to None, which will cause the depth to be calculated
                based on the prev_events.
@@ -400,9 +396,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
            outlier: Indicates whether the event is an `outlier`, i.e. if
                it's from an arbitrary point and floating in the DAG as
                opposed to being inline with the current DAG.
-            historical: Indicates whether the message is being inserted
-                back in time around some existing events. This is used to skip
-                a few checks and mark the event as backfilled.
            origin_server_ts: The origin_server_ts to use if a new event is created. Uses
                the current timestamp if set to None.

@@ -477,7 +470,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                    depth=depth,
                    require_consent=require_consent,
                    outlier=outlier,
-                    historical=historical,
                )
                context = await unpersisted_context.persist(event)
                prev_state_ids = await context.get_prev_state_ids(
@@ -585,7 +577,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        new_room: bool = False,
        require_consent: bool = True,
        outlier: bool = False,
-        historical: bool = False,
        allow_no_prev_events: bool = False,
        prev_event_ids: Optional[List[str]] = None,
        state_event_ids: Optional[List[str]] = None,
@@ -610,22 +601,16 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
            outlier: Indicates whether the event is an `outlier`, i.e. if
                it's from an arbitrary point and floating in the DAG as
                opposed to being inline with the current DAG.
-            historical: Indicates whether the message is being inserted
-                back in time around some existing events. This is used to skip
-                a few checks and mark the event as backfilled.
            allow_no_prev_events: Whether to allow this event to be created an empty
                list of prev_events. Normally this is prohibited just because most
                events should have a prev_event and we should only use this in special
-                cases like MSC2716.
+                cases (previously useful for MSC2716).
            prev_event_ids: The event IDs to use as the prev events
            state_event_ids:
-                The full state at a given event. This is used particularly by the MSC2716
-                /batch_send endpoint. One use case is the historical `state_events_at_start`;
-                since each is marked as an `outlier`, the `EventContext.for_outlier()` won't
-                have any `state_ids` set and therefore can't derive any state even though the
-                prev_events are set so we need to set them ourself via this argument.
-                This should normally be left as None, which will cause the auth_event_ids
-                to be calculated based on the room state at the prev_events.
+                The full state at a given event. This was previously used particularly
+                by the MSC2716 /batch_send endpoint. This should normally be left as
+                None, which will cause the auth_event_ids to be calculated based on the
+                room state at the prev_events.
            depth: Override the depth used to order the event in the DAG.
                Should normally be set to None, which will cause the depth to be calculated
                based on the prev_events.
@@ -667,7 +652,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                        new_room=new_room,
                        require_consent=require_consent,
                        outlier=outlier,
-                        historical=historical,
                        allow_no_prev_events=allow_no_prev_events,
                        prev_event_ids=prev_event_ids,
                        state_event_ids=state_event_ids,
@@ -691,7 +675,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        new_room: bool = False,
        require_consent: bool = True,
        outlier: bool = False,
-        historical: bool = False,
        allow_no_prev_events: bool = False,
        prev_event_ids: Optional[List[str]] = None,
        state_event_ids: Optional[List[str]] = None,
@@ -718,22 +701,16 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
            outlier: Indicates whether the event is an `outlier`, i.e. if
                it's from an arbitrary point and floating in the DAG as
                opposed to being inline with the current DAG.
-            historical: Indicates whether the message is being inserted
-                back in time around some existing events. This is used to skip
-                a few checks and mark the event as backfilled.
            allow_no_prev_events: Whether to allow this event to be created an empty
                list of prev_events. Normally this is prohibited just because most
                events should have a prev_event and we should only use this in special
-                cases like MSC2716.
+                cases (previously useful for MSC2716).
            prev_event_ids: The event IDs to use as the prev events
            state_event_ids:
-                The full state at a given event. This is used particularly by the MSC2716
-                /batch_send endpoint. One use case is the historical `state_events_at_start`;
-                since each is marked as an `outlier`, the `EventContext.for_outlier()` won't
-                have any `state_ids` set and therefore can't derive any state even though the
-                prev_events are set so we need to set them ourself via this argument.
-                This should normally be left as None, which will cause the auth_event_ids
-                to be calculated based on the room state at the prev_events.
+                The full state at a given event. This was previously used particularly
+                by the MSC2716 /batch_send endpoint. This should normally be left as
+                None, which will cause the auth_event_ids to be calculated based on the
+                room state at the prev_events.
            depth: Override the depth used to order the event in the DAG.
                Should normally be set to None, which will cause the depth to be calculated
                based on the prev_events.
@@ -877,7 +854,6 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                content=content,
                require_consent=require_consent,
                outlier=outlier,
-                historical=historical,
                origin_server_ts=origin_server_ts,
            )

@@ -1498,7 +1474,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        # put the server which owns the alias at the front of the server list.
        if room_alias.domain in servers:
            servers.remove(room_alias.domain)
-        servers.insert(0, room_alias.domain)
+            servers.insert(0, room_alias.domain)

        return RoomID.from_string(room_id), servers

@@ -42,7 +42,6 @@ class StatsHandler:
        self.store = hs.get_datastores().main
        self._storage_controllers = hs.get_storage_controllers()
        self.state = hs.get_state_handler()
-        self.server_name = hs.hostname
        self.clock = hs.get_clock()
        self.notifier = hs.get_notifier()
        self.is_mine_id = hs.is_mine_id
@@ -1037,7 +1037,12 @@ class _ReadBodyWithMaxSizeProtocol(protocol.Protocol):
        if reason.check(ResponseDone):
            self.deferred.callback(self.length)
        elif reason.check(PotentialDataLoss):
-            # stolen from https://github.com/twisted/treq/pull/49/files
+            # This applies to requests which don't set `Content-Length` or a
+            # `Transfer-Encoding` in the response because in this case the end of the
+            # response is indicated by the connection being closed, an event which may
+            # also be due to a transient network problem or other error. But since this
+            # behavior is expected of some servers (like YouTube), let's ignore it.
+            # Stolen from https://github.com/twisted/treq/pull/49/files
            # http://twistedmatrix.com/trac/ticket/4840
            self.deferred.callback(self.length)
        else:
@@ -51,8 +51,10 @@ logger = logging.getLogger(__name__)
@implementer(IAgent)
 class MatrixFederationAgent:
    """An Agent-like thing which provides a `request` method which correctly
-    handles resolving matrix server names when using matrix://. Handles standard
-    https URIs as normal.
+    handles resolving matrix server names when using `matrix-federation://`. Handles
+    standard https URIs as normal. The `matrix-federation://` scheme is internal to
+    Synapse and we purposely want to avoid colliding with the `matrix://` URL scheme
+    which is now specced.

    Doesn't implement any retries. (Those are done in MatrixFederationHttpClient.)

@@ -167,14 +169,14 @@ class MatrixFederationAgent:
        # There must be a valid hostname.
        assert parsed_uri.hostname

-        # If this is a matrix:// URI check if the server has delegated matrix
+        # If this is a matrix-federation:// URI check if the server has delegated matrix
        # traffic using well-known delegation.
        #
        # We have to do this here and not in the endpoint as we need to rewrite
        # the host header with the delegated server name.
        delegated_server = None
        if (
-            parsed_uri.scheme == b"matrix"
+            parsed_uri.scheme == b"matrix-federation"
            and not _is_ip_literal(parsed_uri.hostname)
            and not parsed_uri.port
        ):
@@ -250,7 +252,7 @@ class MatrixHostnameEndpointFactory:

@implementer(IStreamClientEndpoint)
 class MatrixHostnameEndpoint:
-    """An endpoint that resolves matrix:// URLs using Matrix server name
+    """An endpoint that resolves matrix-federation:// URLs using Matrix server name
    resolution (i.e. via SRV). Does not check for well-known delegation.

    Args:
@@ -379,7 +381,7 @@ class MatrixHostnameEndpoint:
        connect to.
        """

-        if self._parsed_uri.scheme != b"matrix":
+        if self._parsed_uri.scheme != b"matrix-federation":
            return [Server(host=self._parsed_uri.host, port=self._parsed_uri.port)]

        # Note: We don't do well-known lookup as that needs to have happened
--- a/Show More
+++ b/Show More
				`@@ -1 +0,0 @@`
				`Experimental [MSC3861](https://github.com/matrix-org/matrix-spec-proposals/pull/3861) support: delegate auth to an OIDC provider.`
				`@@ -0,0 +1 @@`
				Add foreign key constraint to `event_forward_extremities`.
				`@@ -0,0 +1 @@`
				Allow configuring the set of workers to proxy outbound federation traffic through via `outbound_federation_restricted_to`.
				`@@ -0,0 +1 @@`
				`Use lower isolation level when cleaning old presence stream data to avoid serialization errors.`
				`@@ -0,0 +1 @@`
				Add `not_user_type` param to the list accounts admin API.
				`@@ -0,0 +1 @@`
				`Fixed header levels on the Admin API "Users" documentation page. Contributed by @sumnerevans at @beeper.`
				`@@ -0,0 +1 @@`
				`Add a timeout that aborts any Postgres statement taking more than 1 hour.`
				`@@ -0,0 +1 @@`
				Fix the `devenv up` configuration which was ignoring the config overrides.
				`@@ -0,0 +1 @@`
				`Optimised cleanup of old entries in device_lists_stream.`
				`@@ -0,0 +1 @@`
				`Correctly resize thumbnails with pillow version >=10.`