Merge remote-tracking branch 'origin/develop' into clokep/rel-workers

2022-10-04 10:27:46 -04:00
parent 837909d17d d8663f5e63
commit 2a7b3e9b2d
65 changed files with 986 additions and 257 deletions
--- a/.github/workflows/dependabot_changelog.yml
+++ b/.github/workflows/dependabot_changelog.yml
@@ -3,16 +3,13 @@ on:
  pull_request:
    types:
      - opened
-      - reopened
+      - reopened  # For debugging!

 permissions:
  # Needed to be able to push the commit. See 
  #     https://docs.github.com/en/code-security/dependabot/working-with-dependabot/automating-dependabot-with-github-actions#enable-auto-merge-on-a-pull-request
  # for a similar example
  contents: write
-  # The pull_requests "synchronize" event doesn't seem to fire with just `contents: write`, so
-  # CI doesn't run with the new changelog. Maybe `pull_requests: write` will fix this?
-  pull-requests: write

 jobs:
  add-changelog:
@@ -31,5 +28,19 @@ jobs:
          git commit -m "Changelog"
          git push
        shell: bash
-  # THIS WORKFLOW HAS VARIOUS WRITE PERMISSIONS---do not add other jobs here unless they
+      # The `git push` above does not trigger CI on the dependabot PR.
+      #
+      # By default, workflows can't trigger other workflows when they're just using the
+      # default `GITHUB_TOKEN` access token. (This is intended to stop you from writing
+      # recursive workflow loops by accident, because that'll get very expensive very
+      # quickly.) Instead, you have to manually call out to another workflow, or else
+      # make your changes (i.e. the `git push` above) using a personal access token.
+      # See
+      # https://docs.github.com/en/actions/using-workflows/triggering-a-workflow#triggering-a-workflow-from-a-workflow
+      #
+      # I have tried and failed to find a way to trigger CI on the "merge ref" of the PR.
+      # See git commit history for previous attempts. If anyone desperately wants to try
+      # again in the future, make a matrix-bot account and use its access token to git push.
+
+  # THIS WORKFLOW HAS WRITE PERMISSIONS---do not add other jobs here unless they
  # are sufficiently locked down to dependabot only as above.
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -29,7 +29,7 @@ jobs:
        run: docker buildx inspect
          
      - name: Log in to DockerHub
-        uses: docker/login-action@v1
+        uses: docker/login-action@v2
        with:
          username: ${{ secrets.DOCKERHUB_USERNAME }}
          password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -48,7 +48,7 @@ jobs:
            type=pep440,pattern={{raw}}

      - name: Build and push all platforms
-        uses: docker/build-push-action@v2
+        uses: docker/build-push-action@v3
        with:
          push: true
          labels: "gitsha1=${{ github.sha }}"
--- a/.github/workflows/latest_deps.yml
+++ b/.github/workflows/latest_deps.yml
@@ -76,7 +76,7 @@ jobs:
            -e POSTGRES_PASSWORD=postgres \
            -e POSTGRES_INITDB_ARGS="--lc-collate C --lc-ctype C --encoding UTF8" \
            postgres:${{ matrix.postgres-version }}
-      - uses: actions/setup-python@v2
+      - uses: actions/setup-python@v4
        with:
          python-version: "3.x"
      - run: pip install .[all,test]
@@ -155,7 +155,7 @@ jobs:
        if: ${{ always() }}
        run: /sytest/scripts/tap_to_gha.pl /logs/results.tap
      - name: Upload SyTest logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v3
        if: ${{ always() }}
        with:
          name: Sytest Logs - ${{ job.status }} - (${{ join(matrix.*, ', ') }})
--- a/.github/workflows/release-artifacts.yml
+++ b/.github/workflows/release-artifacts.yml
@@ -11,6 +11,7 @@ on:

    # we do the full build on tags.
    tags: ["v*"]
+  workflow_dispatch:

 concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
@@ -25,7 +26,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
-      - uses: actions/setup-python@v2
+      - uses: actions/setup-python@v4
      - id: set-distros
        run: |
          # if we're running from a tag, get the full list of distros; otherwise just use debian:sid
@@ -60,7 +61,7 @@ jobs:
          install: true

      - name: Set up docker layer caching
-        uses: actions/cache@v2
+        uses: actions/cache@v3
        with:
          path: /tmp/.buildx-cache
          key: ${{ runner.os }}-buildx-${{ github.sha }}
@@ -68,7 +69,7 @@ jobs:
            ${{ runner.os }}-buildx-

      - name: Set up python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v4

      - name: Build the packages
        # see https://github.com/docker/build-push-action/issues/252
@@ -84,7 +85,7 @@ jobs:
          mv /tmp/.buildx-cache-new /tmp/.buildx-cache

      - name: Upload debs as artifacts
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v3
        with:
          name: debs
          path: debs/*
@@ -106,7 +107,11 @@ jobs:
    steps:
      - uses: actions/checkout@v3

-      - uses: actions/setup-python@v3
+      - uses: actions/setup-python@v4
+        with:
+          # setup-python@v4 doesn't impose a default python version. Need to use 3.x
+          # here, because `python` on osx points to Python 2.7.
+          python-version: "3.x"

      - name: Install cibuildwheel
        run: python -m pip install cibuildwheel==2.9.0 poetry==1.2.0
@@ -145,7 +150,7 @@ jobs:
      - name: Build sdist
        run: python -m build --sdist

-      - uses: actions/upload-artifact@v2
+      - uses: actions/upload-artifact@v3
        with:
          name: Sdist
          path: dist/*.tar.gz
@@ -162,7 +167,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - name: Download all workflow run artifacts
-        uses: actions/download-artifact@v2
+        uses: actions/download-artifact@v3
      - name: Build a tarball for the debs
        run: tar -cvJf debs.tar.xz debs
      - name: Attach to release
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -4,6 +4,7 @@ on:
  push:
    branches: ["develop", "release-*"]
  pull_request:
+  workflow_dispatch:

 concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
@@ -31,7 +32,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
-      - uses: actions/setup-python@v2
+      - uses: actions/setup-python@v4
      - uses: matrix-org/setup-python-poetry@v1
        with:
          extras: "all"
@@ -42,7 +43,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
-      - uses: actions/setup-python@v2
+      - uses: actions/setup-python@v4
      - run: "pip install 'click==8.1.1' 'GitPython>=3.1.20'"
      - run: scripts-dev/check_schema_delta.py --force-colors

@@ -59,14 +60,14 @@ jobs:
        run: scripts-dev/check_line_terminators.sh

  lint-newsfile:
-    if: ${{ github.base_ref == 'develop'  || contains(github.base_ref, 'release-') }}
+    if: ${{ (github.base_ref == 'develop'  || contains(github.base_ref, 'release-')) && github.actor != 'dependabot[bot]' }}
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
        with:
          ref: ${{ github.event.pull_request.head.sha }}
          fetch-depth: 0
-      - uses: actions/setup-python@v2
+      - uses: actions/setup-python@v4
      - run: "pip install 'towncrier>=18.6.0rc1'"
      - run: scripts-dev/check-newsfragment.sh
        env:
@@ -141,7 +142,7 @@ jobs:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
-      - uses: actions/setup-python@v2
+      - uses: actions/setup-python@v4
      - id: get-matrix
        run: .ci/scripts/calculate_jobs.py
    outputs:
@@ -331,7 +332,7 @@ jobs:
        if: ${{ always() }}
        run: /sytest/scripts/tap_to_gha.pl /logs/results.tap
      - name: Upload SyTest logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v3
        if: ${{ always() }}
        with:
          name: Sytest Logs - ${{ job.status }} - (${{ join(matrix.job.*, ', ') }})
--- a/.github/workflows/twisted_trunk.yml
+++ b/.github/workflows/twisted_trunk.yml
@@ -112,7 +112,7 @@ jobs:
        if: ${{ always() }}
        run: /sytest/scripts/tap_to_gha.pl /logs/results.tap
      - name: Upload SyTest logs
-        uses: actions/upload-artifact@v2
+        uses: actions/upload-artifact@v3
        if: ${{ always() }}
        with:
          name: Sytest Logs - ${{ job.status }} - (${{ join(matrix.*, ', ') }})
--- a/changelog.d/13776.feature
+++ b/changelog.d/13776.feature
@@ -0,0 +1 @@
+Experimental support for thread-specific notifications ([MSC3773](https://github.com/matrix-org/matrix-spec-proposals/pull/3773)).
--- a/changelog.d/13815.feature
+++ b/changelog.d/13815.feature
@@ -0,0 +1 @@
+Keep track when an event pulled over federation fails its signature check so we can intelligently back-off in the future.
--- a/changelog.d/13978.misc
+++ b/changelog.d/13978.misc
@@ -0,0 +1 @@
+Bump docker/login-action from 1 to 2.
--- a/changelog.d/13979.misc
+++ b/changelog.d/13979.misc
@@ -0,0 +1 @@
+Bump actions/download-artifact from 2 to 3.
--- a/changelog.d/13980.misc
+++ b/changelog.d/13980.misc
@@ -0,0 +1 @@
+Bump actions/cache from 2 to 3.
--- a/changelog.d/13983.misc
+++ b/changelog.d/13983.misc
@@ -0,0 +1 @@
+Bump actions/setup-python from 2 to 4.
--- a/changelog.d/13984.misc
+++ b/changelog.d/13984.misc
@@ -0,0 +1 @@
+Bump types-pyopenssl from 22.0.0 to 22.0.10.
--- a/changelog.d/13985.misc
+++ b/changelog.d/13985.misc
@@ -0,0 +1 @@
+Bump jsonschema from 4.4.0 to 4.16.0.
--- a/changelog.d/13986.misc
+++ b/changelog.d/13986.misc
@@ -0,0 +1 @@
+Bump types-setuptools from 57.4.9 to 65.4.0.0.
--- a/changelog.d/13987.misc
+++ b/changelog.d/13987.misc
@@ -0,0 +1 @@
+Bump types-requests from 2.27.11 to 2.28.11.
--- a/changelog.d/13988.misc
+++ b/changelog.d/13988.misc
@@ -0,0 +1 @@
+Bump isort from 5.7.0 to 5.10.1.
--- a/changelog.d/13996.feature
+++ b/changelog.d/13996.feature
@@ -0,0 +1 @@
+Send application service access tokens as a header (and query parameter). Implement [MSC2832](https://github.com/matrix-org/matrix-spec-proposals/pull/2832).
--- a/changelog.d/13997.feature
+++ b/changelog.d/13997.feature
@@ -0,0 +1 @@
+Ignore server ACL changes when generating pushes. Implement [MSC3786](https://github.com/matrix-org/matrix-spec-proposals/pull/3786).
--- a/changelog.d/14003.doc
+++ b/changelog.d/14003.doc
@@ -0,0 +1 @@
+Linkify urls in config documentation.
--- a/changelog.d/14020.misc
+++ b/changelog.d/14020.misc
@@ -0,0 +1 @@
+Clear out stale entries in `event_push_actions_staging` table.
--- a/changelog.d/14022.misc
+++ b/changelog.d/14022.misc
@@ -0,0 +1 @@
+Bump docker/build-push-action from 2 to 3.
--- a/changelog.d/14023.misc
+++ b/changelog.d/14023.misc
@@ -0,0 +1 @@
+Bump actions/upload-artifact from 2 to 3.
--- a/changelog.d/14024.removal
+++ b/changelog.d/14024.removal
@@ -0,0 +1 @@
+Announce that legacy metric names are deprecated, will be turned off by default in Synapse v1.71.0 and removed altogether in Synapse v1.73.0. See the upgrade notes for more information.
--- a/changelog.d/14025.bugfix
+++ b/changelog.d/14025.bugfix
@@ -0,0 +1 @@
+Do not return an unspecified `original_event` field when using the stable `/relations` endpoint. Introduced in Synapse v1.57.0.
--- a/changelog.d/14027.misc
+++ b/changelog.d/14027.misc
@@ -0,0 +1 @@
+Prototype a workflow to automatically add changelogs to dependabot PRs.
--- a/changelog.d/14032.feature
+++ b/changelog.d/14032.feature
@@ -0,0 +1 @@
+Advertise Matrix 1.3 support on `/_matrix/client/versions`.
--- a/changelog.d/14046.misc
+++ b/changelog.d/14046.misc
@@ -0,0 +1 @@
+Bump actions/setup-python from 2 to 4.
--- a/docs/metrics-howto.md
+++ b/docs/metrics-howto.md
@@ -135,6 +135,8 @@ Synapse 1.2 updates the Prometheus metrics to match the naming
 convention of the upstream `prometheus_client`. The old names are
 considered deprecated and will be removed in a future version of
 Synapse.
+**The old names will be disabled by default in Synapse v1.71.0 and removed
+altogether in Synapse v1.73.0.**

 | New Name                                                                     | Old Name                                                               |
 | ---------------------------------------------------------------------------- | ---------------------------------------------------------------------- |
@@ -146,6 +148,13 @@ Synapse.
 | synapse_federation_client_events_processed_total                             | synapse_federation_client_events_processed                             |
 | synapse_event_processing_loop_count_total                                    | synapse_event_processing_loop_count                                    |
 | synapse_event_processing_loop_room_count_total                               | synapse_event_processing_loop_room_count                               |
+| synapse_util_caches_cache_hits                                               | synapse_util_caches_cache:hits                                         |
+| synapse_util_caches_cache_size                                               | synapse_util_caches_cache:size                                         |
+| synapse_util_caches_cache_evicted_size                                       | synapse_util_caches_cache:evicted_size                                 |
+| synapse_util_caches_cache                                                    | synapse_util_caches_cache:total                                        |
+| synapse_util_caches_response_cache_size                                      | synapse_util_caches_response_cache:size                                |
+| synapse_util_caches_response_cache_hits                                      | synapse_util_caches_response_cache:hits                                |
+| synapse_util_caches_response_cache_evicted_size                              | synapse_util_caches_response_cache:evicted_size                        |
 | synapse_util_metrics_block_count_total                                       | synapse_util_metrics_block_count                                       |
 | synapse_util_metrics_block_time_seconds_total                                | synapse_util_metrics_block_time_seconds                                |
 | synapse_util_metrics_block_ru_utime_seconds_total                            | synapse_util_metrics_block_ru_utime_seconds                            |
@@ -261,7 +270,7 @@ Standard Metric Names

 As of synapse version 0.18.2, the format of the process-wide metrics has
 been changed to fit prometheus standard naming conventions. Additionally
-the units have been changed to seconds, from miliseconds.
+the units have been changed to seconds, from milliseconds.

 | New name                                 | Old name                          |
 | ---------------------------------------- | --------------------------------- |
--- a/docs/upgrade.md
+++ b/docs/upgrade.md
@@ -100,6 +100,34 @@ vice versa.
 Once all workers are upgraded to v1.69 (or downgraded to v1.68), receipts
 replication will resume as normal.

+
+## Deprecation of legacy Prometheus metric names
+
+In current versions of Synapse, some Prometheus metrics are emitted under two different names,
+with one of the names being older but non-compliant with OpenMetrics and Prometheus conventions
+and one of the names being newer but compliant.
+
+Synapse v1.71.0 will turn the old metric names off *by default*.
+For administrators that still rely on them and have not had chance to update their
+uses of the metrics, it's possible to specify `enable_legacy_metrics: true` in
+the configuration to re-enable them temporarily.
+
+Synapse v1.73.0 will **remove legacy metric names altogether** and it will no longer
+be possible to re-enable them.
+
+The Grafana dashboard, Prometheus recording rules and Prometheus Consoles included
+in the `contrib` directory in the Synapse repository have been updated to no longer
+rely on the legacy names. These can be used on a current version of Synapse
+because current versions of Synapse emit both old and new names.
+
+You may need to update your alerting rules or any other rules that depend on
+the names of Prometheus metrics.
+If you want to test your changes before legacy names are disabled by default,
+you may specify `enable_legacy_metrics: false` in your homeserver configuration.
+
+A list of affected metrics is available on the [Metrics How-to page](https://matrix-org.github.io/synapse/v1.69/metrics-howto.html?highlight=metrics%20deprecated#renaming-of-metrics--deprecation-of-old-names-in-12).
+
+
 # Upgrading to v1.68.0

 Two changes announced in the upgrade notes for v1.67.0 have now landed in v1.68.0.
--- a/docs/usage/configuration/config_documentation.md
+++ b/docs/usage/configuration/config_documentation.md
@@ -179,7 +179,7 @@ This will tell other servers to send traffic to port 443 instead.

 This option currently defaults to false.

-See https://matrix-org.github.io/synapse/latest/delegate.html for more
+See [Delegation of incoming federation traffic](../../delegate.md) for more
 information.

 Example configuration:
@@ -2436,6 +2436,31 @@ Example configuration:
 enable_metrics: true
 ```
 ---
+### `enable_legacy_metrics`
+
+Set to `true` to publish both legacy and non-legacy Prometheus metric names,
+or to `false` to only publish non-legacy Prometheus metric names.
+Defaults to `true`. Has no effect if `enable_metrics` is `false`.
+**In Synapse v1.71.0, this will default to `false` before being removed in Synapse v1.73.0.**
+
+Legacy metric names include:
+- metrics containing colons in the name, such as `synapse_util_caches_response_cache:hits`, because colons are supposed to be reserved for user-defined recording rules;
+- counters that don't end with the `_total` suffix, such as `synapse_federation_client_sent_edus`, therefore not adhering to the OpenMetrics standard.
+
+These legacy metric names are unconventional and not compliant with OpenMetrics standards.
+They are included for backwards compatibility.
+
+Example configuration:
+```yaml
+enable_legacy_metrics: false
+```
+
+See https://github.com/matrix-org/synapse/issues/11106 for context.
+
+*Since v1.67.0.*
+
+**Will be removed in v1.73.0.**
+---
 ### `sentry`

 Use this option to enable sentry integration. Provide the DSN assigned to you by sentry
@@ -2952,7 +2977,7 @@ Options for each entry include:

     * `module`: The class name of a custom mapping module. Default is
       `synapse.handlers.oidc.JinjaOidcMappingProvider`.
-        See https://matrix-org.github.io/synapse/latest/sso_mapping_providers.html#openid-mapping-providers
+        See [OpenID Mapping Providers](../../sso_mapping_providers.md#openid-mapping-providers)
        for information on implementing a custom mapping provider.

     * `config`: Configuration for the mapping provider module. This section will
@@ -3393,13 +3418,15 @@ This option has the following sub-options:
   the user directory. If false, search results will only contain users
    visible in public rooms and users sharing a room with the requester.
    Defaults to false.
+
    NB. If you set this to true, and the last time the user_directory search
    indexes were (re)built was before Synapse 1.44, you'll have to
    rebuild the indexes in order to search through all known users.
+    
    These indexes are built the first time Synapse starts; admins can
-    manually trigger a rebuild via API following the instructions at
-         https://matrix-org.github.io/synapse/latest/usage/administration/admin_api/background_updates.html#run
-    Set to true to return search results containing all known users, even if that
+    manually trigger a rebuild via the API following the instructions
+    [for running background updates](../administration/admin_api/background_updates.md#run),
+    set to true to return search results containing all known users, even if that
    user does not share a room with the requester.
 * `prefer_local_users`: Defines whether to prefer local users in search query results.
   If set to true, local users are more likely to appear above remote users when searching the
--- a/poetry.lock
+++ b/poetry.lock
@@ -399,15 +399,16 @@ scripts = ["click (>=6.0)", "twisted (>=16.4.0)"]

 [[package]]
 name = "isort"
-version = "5.7.0"
+version = "5.10.1"
 description = "A Python utility / library to sort Python imports."
 category = "dev"
 optional = false
-python-versions = ">=3.6,<4.0"
+python-versions = ">=3.6.1,<4.0"

 [package.extras]
 colors = ["colorama (>=0.4.3,<0.5.0)"]
 pipfile_deprecated_finder = ["pipreqs", "requirementslib"]
+plugins = ["setuptools"]
 requirements_deprecated_finder = ["pip-api", "pipreqs"]

 [[package]]
@@ -455,7 +456,7 @@ i18n = ["Babel (>=2.7)"]

 [[package]]
 name = "jsonschema"
-version = "4.4.0"
+version = "4.16.0"
 description = "An implementation of JSON Schema validation for Python"
 category = "main"
 optional = false
@@ -465,12 +466,13 @@ python-versions = ">=3.7"
 attrs = ">=17.4.0"
 importlib-metadata = {version = "*", markers = "python_version < \"3.8\""}
 importlib-resources = {version = ">=1.4.0", markers = "python_version < \"3.9\""}
+pkgutil-resolve-name = {version = ">=1.3.10", markers = "python_version < \"3.9\""}
 pyrsistent = ">=0.14.0,<0.17.0 || >0.17.0,<0.17.1 || >0.17.1,<0.17.2 || >0.17.2"
 typing-extensions = {version = "*", markers = "python_version < \"3.8\""}

 [package.extras]
 format = ["fqdn", "idna", "isoduration", "jsonpointer (>1.13)", "rfc3339-validator", "rfc3987", "uri-template", "webcolors (>=1.11)"]
-format_nongpl = ["fqdn", "idna", "isoduration", "jsonpointer (>1.13)", "rfc3339-validator", "rfc3986-validator (>0.1.0)", "uri-template", "webcolors (>=1.11)"]
+format-nongpl = ["fqdn", "idna", "isoduration", "jsonpointer (>1.13)", "rfc3339-validator", "rfc3986-validator (>0.1.0)", "uri-template", "webcolors (>=1.11)"]

 [[package]]
 name = "keyring"
@@ -690,6 +692,14 @@ python-versions = "*"
 [package.extras]
 testing = ["coverage", "nose"]

+[[package]]
+name = "pkgutil_resolve_name"
+version = "1.3.10"
+description = "Resolve a name to an object."
+category = "main"
+optional = false
+python-versions = ">=3.6"
+
 [[package]]
 name = "platformdirs"
 version = "2.5.1"
@@ -1442,8 +1452,8 @@ optional = false
 python-versions = "*"

 [[package]]
-name = "types-pyopenssl"
-version = "22.0.0"
+name = "types-pyOpenSSL"
+version = "22.0.10"
 description = "Typing stubs for pyOpenSSL"
 category = "dev"
 optional = false
@@ -1462,7 +1472,7 @@ python-versions = "*"

 [[package]]
 name = "types-requests"
-version = "2.27.11"
+version = "2.28.11"
 description = "Typing stubs for requests"
 category = "dev"
 optional = false
@@ -1473,7 +1483,7 @@ types-urllib3 = "<1.27"

 [[package]]
 name = "types-setuptools"
-version = "57.4.9"
+version = "65.4.0.0"
 description = "Typing stubs for setuptools"
 category = "dev"
 optional = false
@@ -1626,7 +1636,7 @@ url_preview = ["lxml"]
 [metadata]
 lock-version = "1.1"
 python-versions = "^3.7.1"
-content-hash = "1b14fc274d9e2a495a7f864150f3ffcf4d9f585e09a67e53301ae4ef3c2f3e48"
+content-hash = "9d74da808739e4c3d15a2d3473f01ad419f62aec8bf28613b03bd69136c4745b"

 [metadata.files]
 attrs = [
@@ -1970,8 +1980,8 @@ incremental = [
    {file = "incremental-21.3.0.tar.gz", hash = "sha256:02f5de5aff48f6b9f665d99d48bfc7ec03b6e3943210de7cfc88856d755d6f57"},
 ]
 isort = [
-    {file = "isort-5.7.0-py3-none-any.whl", hash = "sha256:fff4f0c04e1825522ce6949973e83110a6e907750cd92d128b0d14aaaadbffdc"},
-    {file = "isort-5.7.0.tar.gz", hash = "sha256:c729845434366216d320e936b8ad6f9d681aab72dc7cbc2d51bedc3582f3ad1e"},
+    {file = "isort-5.10.1-py3-none-any.whl", hash = "sha256:6f62d78e2f89b4500b080fe3a81690850cd254227f27f75c3a0c491a1f351ba7"},
+    {file = "isort-5.10.1.tar.gz", hash = "sha256:e8443a5e7a020e9d7f97f1d7d9cd17c88bcb3bc7e218bf9cf5095fe550be2951"},
 ]
 jaeger-client = [
    {file = "jaeger-client-4.8.0.tar.gz", hash = "sha256:3157836edab8e2c209bd2d6ae61113db36f7ee399e66b1dcbb715d87ab49bfe0"},
@@ -1985,18 +1995,15 @@ jinja2 = [
    {file = "Jinja2-3.0.3.tar.gz", hash = "sha256:611bb273cd68f3b993fabdc4064fc858c5b47a973cb5aa7999ec1ba405c87cd7"},
 ]
 jsonschema = [
-    {file = "jsonschema-4.4.0-py3-none-any.whl", hash = "sha256:77281a1f71684953ee8b3d488371b162419767973789272434bbc3f29d9c8823"},
-    {file = "jsonschema-4.4.0.tar.gz", hash = "sha256:636694eb41b3535ed608fe04129f26542b59ed99808b4f688aa32dcf55317a83"},
+    {file = "jsonschema-4.16.0-py3-none-any.whl", hash = "sha256:9e74b8f9738d6a946d70705dc692b74b5429cd0960d58e79ffecfc43b2221eb9"},
+    {file = "jsonschema-4.16.0.tar.gz", hash = "sha256:165059f076eff6971bae5b742fc029a7b4ef3f9bcf04c14e4776a7605de14b23"},
 ]
 keyring = [
    {file = "keyring-23.5.0-py3-none-any.whl", hash = "sha256:b0d28928ac3ec8e42ef4cc227822647a19f1d544f21f96457965dc01cf555261"},
    {file = "keyring-23.5.0.tar.gz", hash = "sha256:9012508e141a80bd1c0b6778d5c610dd9f8c464d75ac6774248500503f972fb9"},
 ]
 ldap3 = [
-    {file = "ldap3-2.9.1-py2.6.egg", hash = "sha256:5ab7febc00689181375de40c396dcad4f2659cd260fc5e94c508b6d77c17e9d5"},
-    {file = "ldap3-2.9.1-py2.7.egg", hash = "sha256:2bc966556fc4d4fa9f445a1c31dc484ee81d44a51ab0e2d0fd05b62cac75daa6"},
    {file = "ldap3-2.9.1-py2.py3-none-any.whl", hash = "sha256:5869596fc4948797020d3f03b7939da938778a0f9e2009f7a072ccf92b8e8d70"},
-    {file = "ldap3-2.9.1-py3.9.egg", hash = "sha256:5630d1383e09ba94839e253e013f1aa1a2cf7a547628ba1265cb7b9a844b5687"},
    {file = "ldap3-2.9.1.tar.gz", hash = "sha256:f3e7fc4718e3f09dda568b57100095e0ce58633bcabbed8667ce3f8fbaa4229f"},
 ]
 lxml = [
@@ -2259,6 +2266,10 @@ pkginfo = [
    {file = "pkginfo-1.8.2-py2.py3-none-any.whl", hash = "sha256:c24c487c6a7f72c66e816ab1796b96ac6c3d14d49338293d2141664330b55ffc"},
    {file = "pkginfo-1.8.2.tar.gz", hash = "sha256:542e0d0b6750e2e21c20179803e40ab50598d8066d51097a0e382cba9eb02bff"},
 ]
+pkgutil_resolve_name = [
+    {file = "pkgutil_resolve_name-1.3.10-py3-none-any.whl", hash = "sha256:ca27cc078d25c5ad71a9de0a7a330146c4e014c2462d9af19c6b828280649c5e"},
+    {file = "pkgutil_resolve_name-1.3.10.tar.gz", hash = "sha256:357d6c9e6a755653cfd78893817c0853af365dd51ec97f3d358a819373bbd174"},
+]
 platformdirs = [
    {file = "platformdirs-2.5.1-py3-none-any.whl", hash = "sha256:bcae7cab893c2d310a711b70b24efb93334febe65f8de776ee320b517471e227"},
    {file = "platformdirs-2.5.1.tar.gz", hash = "sha256:7535e70dfa32e84d4b34996ea99c5e432fa29a708d0f4e394bbcb2a8faa4f16d"},
@@ -2287,34 +2298,12 @@ psycopg2cffi-compat = [
    {file = "psycopg2cffi-compat-1.1.tar.gz", hash = "sha256:d25e921748475522b33d13420aad5c2831c743227dc1f1f2585e0fdb5c914e05"},
 ]
 pyasn1 = [
-    {file = "pyasn1-0.4.8-py2.4.egg", hash = "sha256:fec3e9d8e36808a28efb59b489e4528c10ad0f480e57dcc32b4de5c9d8c9fdf3"},
-    {file = "pyasn1-0.4.8-py2.5.egg", hash = "sha256:0458773cfe65b153891ac249bcf1b5f8f320b7c2ce462151f8fa74de8934becf"},
-    {file = "pyasn1-0.4.8-py2.6.egg", hash = "sha256:5c9414dcfede6e441f7e8f81b43b34e834731003427e5b09e4e00e3172a10f00"},
-    {file = "pyasn1-0.4.8-py2.7.egg", hash = "sha256:6e7545f1a61025a4e58bb336952c5061697da694db1cae97b116e9c46abcf7c8"},
    {file = "pyasn1-0.4.8-py2.py3-none-any.whl", hash = "sha256:39c7e2ec30515947ff4e87fb6f456dfc6e84857d34be479c9d4a4ba4bf46aa5d"},
-    {file = "pyasn1-0.4.8-py3.1.egg", hash = "sha256:78fa6da68ed2727915c4767bb386ab32cdba863caa7dbe473eaae45f9959da86"},
-    {file = "pyasn1-0.4.8-py3.2.egg", hash = "sha256:08c3c53b75eaa48d71cf8c710312316392ed40899cb34710d092e96745a358b7"},
-    {file = "pyasn1-0.4.8-py3.3.egg", hash = "sha256:03840c999ba71680a131cfaee6fab142e1ed9bbd9c693e285cc6aca0d555e576"},
-    {file = "pyasn1-0.4.8-py3.4.egg", hash = "sha256:7ab8a544af125fb704feadb008c99a88805126fb525280b2270bb25cc1d78a12"},
-    {file = "pyasn1-0.4.8-py3.5.egg", hash = "sha256:e89bf84b5437b532b0803ba5c9a5e054d21fec423a89952a74f87fa2c9b7bce2"},
-    {file = "pyasn1-0.4.8-py3.6.egg", hash = "sha256:014c0e9976956a08139dc0712ae195324a75e142284d5f87f1a87ee1b068a359"},
-    {file = "pyasn1-0.4.8-py3.7.egg", hash = "sha256:99fcc3c8d804d1bc6d9a099921e39d827026409a58f2a720dcdb89374ea0c776"},
    {file = "pyasn1-0.4.8.tar.gz", hash = "sha256:aef77c9fb94a3ac588e87841208bdec464471d9871bd5050a287cc9a475cd0ba"},
 ]
 pyasn1-modules = [
    {file = "pyasn1-modules-0.2.8.tar.gz", hash = "sha256:905f84c712230b2c592c19470d3ca8d552de726050d1d1716282a1f6146be65e"},
-    {file = "pyasn1_modules-0.2.8-py2.4.egg", hash = "sha256:0fe1b68d1e486a1ed5473f1302bd991c1611d319bba158e98b106ff86e1d7199"},
-    {file = "pyasn1_modules-0.2.8-py2.5.egg", hash = "sha256:fe0644d9ab041506b62782e92b06b8c68cca799e1a9636ec398675459e031405"},
-    {file = "pyasn1_modules-0.2.8-py2.6.egg", hash = "sha256:a99324196732f53093a84c4369c996713eb8c89d360a496b599fb1a9c47fc3eb"},
-    {file = "pyasn1_modules-0.2.8-py2.7.egg", hash = "sha256:0845a5582f6a02bb3e1bde9ecfc4bfcae6ec3210dd270522fee602365430c3f8"},
    {file = "pyasn1_modules-0.2.8-py2.py3-none-any.whl", hash = "sha256:a50b808ffeb97cb3601dd25981f6b016cbb3d31fbf57a8b8a87428e6158d0c74"},
-    {file = "pyasn1_modules-0.2.8-py3.1.egg", hash = "sha256:f39edd8c4ecaa4556e989147ebf219227e2cd2e8a43c7e7fcb1f1c18c5fd6a3d"},
-    {file = "pyasn1_modules-0.2.8-py3.2.egg", hash = "sha256:b80486a6c77252ea3a3e9b1e360bc9cf28eaac41263d173c032581ad2f20fe45"},
-    {file = "pyasn1_modules-0.2.8-py3.3.egg", hash = "sha256:65cebbaffc913f4fe9e4808735c95ea22d7a7775646ab690518c056784bc21b4"},
-    {file = "pyasn1_modules-0.2.8-py3.4.egg", hash = "sha256:15b7c67fabc7fc240d87fb9aabf999cf82311a6d6fb2c70d00d3d0604878c811"},
-    {file = "pyasn1_modules-0.2.8-py3.5.egg", hash = "sha256:426edb7a5e8879f1ec54a1864f16b882c2837bfd06eee62f2c982315ee2473ed"},
-    {file = "pyasn1_modules-0.2.8-py3.6.egg", hash = "sha256:cbac4bc38d117f2a49aeedec4407d23e8866ea4ac27ff2cf7fb3e5b570df19e0"},
-    {file = "pyasn1_modules-0.2.8-py3.7.egg", hash = "sha256:c29a5e5cc7a3f05926aff34e097e84f8589cd790ce0ed41b67aed6857b26aafd"},
 ]
 pycodestyle = [
    {file = "pycodestyle-2.8.0-py2.py3-none-any.whl", hash = "sha256:720f8b39dde8b293825e7ff02c475f3077124006db4f440dcbc9a20b76548a20"},
@@ -2452,6 +2441,13 @@ pyyaml = [
    {file = "PyYAML-6.0-cp310-cp310-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:f84fbc98b019fef2ee9a1cb3ce93e3187a6df0b2538a651bfb890254ba9f90b5"},
    {file = "PyYAML-6.0-cp310-cp310-win32.whl", hash = "sha256:2cd5df3de48857ed0544b34e2d40e9fac445930039f3cfe4bcc592a1f836d513"},
    {file = "PyYAML-6.0-cp310-cp310-win_amd64.whl", hash = "sha256:daf496c58a8c52083df09b80c860005194014c3698698d1a57cbcfa182142a3a"},
+    {file = "PyYAML-6.0-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:d4b0ba9512519522b118090257be113b9468d804b19d63c71dbcf4a48fa32358"},
+    {file = "PyYAML-6.0-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:81957921f441d50af23654aa6c5e5eaf9b06aba7f0a19c18a538dc7ef291c5a1"},
+    {file = "PyYAML-6.0-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:afa17f5bc4d1b10afd4466fd3a44dc0e245382deca5b3c353d8b757f9e3ecb8d"},
+    {file = "PyYAML-6.0-cp311-cp311-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:dbad0e9d368bb989f4515da330b88a057617d16b6a8245084f1b05400f24609f"},
+    {file = "PyYAML-6.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:432557aa2c09802be39460360ddffd48156e30721f5e8d917f01d31694216782"},
+    {file = "PyYAML-6.0-cp311-cp311-win32.whl", hash = "sha256:bfaef573a63ba8923503d27530362590ff4f576c626d86a9fed95822a8255fd7"},
+    {file = "PyYAML-6.0-cp311-cp311-win_amd64.whl", hash = "sha256:01b45c0191e6d66c470b6cf1b9531a771a83c1c4208272ead47a3ae4f2f603bf"},
    {file = "PyYAML-6.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:897b80890765f037df3403d22bab41627ca8811ae55e9a722fd0392850ec4d86"},
    {file = "PyYAML-6.0-cp36-cp36m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:50602afada6d6cbfad699b0c7bb50d5ccffa7e46a3d738092afddc1f9758427f"},
    {file = "PyYAML-6.0-cp36-cp36m-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:48c346915c114f5fdb3ead70312bd042a953a8ce5c7106d5bfb1a5254e47da92"},
@@ -2755,21 +2751,21 @@ types-psycopg2 = [
    {file = "types-psycopg2-2.9.9.tar.gz", hash = "sha256:4f9d4d52eeb343dc00fd5ed4f1513a8a5c18efba0a072eb82706d15cf4f20a2e"},
    {file = "types_psycopg2-2.9.9-py3-none-any.whl", hash = "sha256:cec9291d4318ad70b407310f8304b3d40f6d0358f09870448f7a65e3027c80af"},
 ]
-types-pyopenssl = [
-    {file = "types-pyOpenSSL-22.0.0.tar.gz", hash = "sha256:d86dde7f6fe2f1ac9fe0b6282e489f649f480364bdaa9d6a4696d52505f4477e"},
-    {file = "types_pyOpenSSL-22.0.0-py3-none-any.whl", hash = "sha256:da685f57b864979f36df0157895139c8244ad4aad19b551f1678206fbad0108a"},
+types-pyOpenSSL = [
+    {file = "types-pyOpenSSL-22.0.10.tar.gz", hash = "sha256:f943b834f5b97e5e808764c2f6e37be1a2e226c46792296f61558196acfcc3a1"},
+    {file = "types_pyOpenSSL-22.0.10-py3-none-any.whl", hash = "sha256:63baea211768bea580a769ac5c0d637ae8cd3150314aadc5726ca22e4c4f241a"},
 ]
 types-pyyaml = [
    {file = "types-PyYAML-6.0.4.tar.gz", hash = "sha256:6252f62d785e730e454dfa0c9f0fb99d8dae254c5c3c686903cf878ea27c04b7"},
    {file = "types_PyYAML-6.0.4-py3-none-any.whl", hash = "sha256:693b01c713464a6851f36ff41077f8adbc6e355eda929addfb4a97208aea9b4b"},
 ]
 types-requests = [
-    {file = "types-requests-2.27.11.tar.gz", hash = "sha256:6a7ed24b21780af4a5b5e24c310b2cd885fb612df5fd95584d03d87e5f2a195a"},
-    {file = "types_requests-2.27.11-py3-none-any.whl", hash = "sha256:506279bad570c7b4b19ac1f22e50146538befbe0c133b2cea66a9b04a533a859"},
+    {file = "types-requests-2.28.11.tar.gz", hash = "sha256:7ee827eb8ce611b02b5117cfec5da6455365b6a575f5e3ff19f655ba603e6b4e"},
+    {file = "types_requests-2.28.11-py3-none-any.whl", hash = "sha256:af5f55e803cabcfb836dad752bd6d8a0fc8ef1cd84243061c0e27dee04ccf4fd"},
 ]
 types-setuptools = [
-    {file = "types-setuptools-57.4.9.tar.gz", hash = "sha256:536ef74744f8e1e4be4fc719887f886e74e4cf3c792b4a06984320be4df450b5"},
-    {file = "types_setuptools-57.4.9-py3-none-any.whl", hash = "sha256:948dc6863373750e2cd0b223a84f1fb608414cde5e55cf38ea657b93aeb411d2"},
+    {file = "types-setuptools-65.4.0.0.tar.gz", hash = "sha256:d9021d6a70690b34e7bd2947e7ab10167c646fbf062508cb56581be2e2a1615e"},
+    {file = "types_setuptools-65.4.0.0-py3-none-any.whl", hash = "sha256:ce178b3f7dbd6c0e67f8eee7ae29c1be280ade7e5188bdd9e620843de4060d85"},
 ]
 types-urllib3 = [
    {file = "types-urllib3-1.26.10.tar.gz", hash = "sha256:a26898f530e6c3f43f25b907f2b884486868ffd56a9faa94cbf9b3eb6e165d6a"},
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -267,7 +267,7 @@ all = [

 [tool.poetry.dev-dependencies]
 ## We pin black so that our tests don't start failing on new releases.
-isort = "==5.7.0"
+isort = "==5.10.1"
 black = "==22.3.0"
 flake8-comprehensions = "*"
 flake8-bugbear = "==21.3.2"
--- a/rust/src/push/base_rules.rs
+++ b/rust/src/push/base_rules.rs
@@ -173,7 +173,7 @@ pub const BASE_APPEND_OVERRIDE_RULES: &[PushRule] = &[
        default_enabled: true,
    },
    PushRule {
-        rule_id: Cow::Borrowed("global/override/.org.matrix.msc3786.rule.room.server_acl"),
+        rule_id: Cow::Borrowed("global/override/.m.rule.room.server_acl"),
        priority_class: 5,
        conditions: Cow::Borrowed(&[
            Condition::Known(KnownCondition::EventMatch(EventMatchCondition {
--- a/rust/src/push/mod.rs
+++ b/rust/src/push/mod.rs
@@ -401,7 +401,6 @@ impl PushRules {
 pub struct FilteredPushRules {
    push_rules: PushRules,
    enabled_map: BTreeMap<String, bool>,
-    msc3786_enabled: bool,
    msc3772_enabled: bool,
 }

@@ -411,13 +410,11 @@ impl FilteredPushRules {
    pub fn py_new(
        push_rules: PushRules,
        enabled_map: BTreeMap<String, bool>,
-        msc3786_enabled: bool,
        msc3772_enabled: bool,
    ) -> Self {
        Self {
            push_rules,
            enabled_map,
-            msc3786_enabled,
            msc3772_enabled,
        }
    }
@@ -437,12 +434,6 @@ impl FilteredPushRules {
            .iter()
            .filter(|rule| {
                // Ignore disabled experimental push rules
-                if !self.msc3786_enabled
-                    && rule.rule_id == "global/override/.org.matrix.msc3786.rule.room.server_acl"
-                {
-                    return false;
-                }
-
                if !self.msc3772_enabled
                    && rule.rule_id == "global/underride/.org.matrix.msc3772.thread_reply"
                {
--- a/stubs/synapse/synapse_rust/push.pyi
+++ b/stubs/synapse/synapse_rust/push.pyi
@@ -26,11 +26,7 @@ class PushRules:

 class FilteredPushRules:
    def __init__(
-        self,
-        push_rules: PushRules,
-        enabled_map: Dict[str, bool],
-        msc3786_enabled: bool,
-        msc3772_enabled: bool,
+        self, push_rules: PushRules, enabled_map: Dict[str, bool], msc3772_enabled: bool
    ): ...
    def rules(self) -> Collection[Tuple[PushRule, bool]]: ...

--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -31,6 +31,9 @@ MAX_ALIAS_LENGTH = 255
 # the maximum length for a user id is 255 characters
 MAX_USERID_LENGTH = 255

+# Constant value used for the pseudo-thread which is the main timeline.
+MAIN_TIMELINE: Final = "main"
+

 class Membership:

--- a/synapse/api/filtering.py
+++ b/synapse/api/filtering.py
@@ -84,6 +84,7 @@ ROOM_EVENT_FILTER_SCHEMA = {
        "contains_url": {"type": "boolean"},
        "lazy_load_members": {"type": "boolean"},
        "include_redundant_members": {"type": "boolean"},
+        "org.matrix.msc3773.unread_thread_notifications": {"type": "boolean"},
        # Include or exclude events with the provided labels.
        # cf https://github.com/matrix-org/matrix-doc/pull/2326
        "org.matrix.labels": {"type": "array", "items": {"type": "string"}},
@@ -240,6 +241,9 @@ class FilterCollection:
    def include_redundant_members(self) -> bool:
        return self._room_state_filter.include_redundant_members

+    def unread_thread_notifications(self) -> bool:
+        return self._room_timeline_filter.unread_thread_notifications
+
    async def filter_presence(
        self, events: Iterable[UserPresenceState]
    ) -> List[UserPresenceState]:
@@ -304,6 +308,12 @@ class Filter:
        self.include_redundant_members = filter_json.get(
            "include_redundant_members", False
        )
+        if hs.config.experimental.msc3773_enabled:
+            self.unread_thread_notifications: bool = filter_json.get(
+                "org.matrix.msc3773.unread_thread_notifications", False
+            )
+        else:
+            self.unread_thread_notifications = False

        self.types = filter_json.get("types", None)
        self.not_types = filter_json.get("not_types", [])
--- a/synapse/appservice/api.py
+++ b/synapse/appservice/api.py
@@ -120,7 +120,11 @@ class ApplicationServiceApi(SimpleHttpClient):

        uri = service.url + ("/users/%s" % urllib.parse.quote(user_id))
        try:
-            response = await self.get_json(uri, {"access_token": service.hs_token})
+            response = await self.get_json(
+                uri,
+                {"access_token": service.hs_token},
+                headers={"Authorization": f"Bearer {service.hs_token}"},
+            )
            if response is not None:  # just an empty json object
                return True
        except CodeMessageException as e:
@@ -140,7 +144,11 @@ class ApplicationServiceApi(SimpleHttpClient):

        uri = service.url + ("/rooms/%s" % urllib.parse.quote(alias))
        try:
-            response = await self.get_json(uri, {"access_token": service.hs_token})
+            response = await self.get_json(
+                uri,
+                {"access_token": service.hs_token},
+                headers={"Authorization": f"Bearer {service.hs_token}"},
+            )
            if response is not None:  # just an empty json object
                return True
        except CodeMessageException as e:
@@ -181,7 +189,9 @@ class ApplicationServiceApi(SimpleHttpClient):
                **fields,
                b"access_token": service.hs_token,
            }
-            response = await self.get_json(uri, args=args)
+            response = await self.get_json(
+                uri, args=args, headers={"Authorization": f"Bearer {service.hs_token}"}
+            )
            if not isinstance(response, list):
                logger.warning(
                    "query_3pe to %s returned an invalid response %r", uri, response
@@ -217,7 +227,11 @@ class ApplicationServiceApi(SimpleHttpClient):
                urllib.parse.quote(protocol),
            )
            try:
-                info = await self.get_json(uri, {"access_token": service.hs_token})
+                info = await self.get_json(
+                    uri,
+                    {"access_token": service.hs_token},
+                    headers={"Authorization": f"Bearer {service.hs_token}"},
+                )

                if not _is_valid_3pe_metadata(info):
                    logger.warning(
@@ -313,6 +327,7 @@ class ApplicationServiceApi(SimpleHttpClient):
                uri=uri,
                json_body=body,
                args={"access_token": service.hs_token},
+                headers={"Authorization": f"Bearer {service.hs_token}"},
            )
            if logger.isEnabledFor(logging.DEBUG):
                logger.debug(
--- a/synapse/config/experimental.py
+++ b/synapse/config/experimental.py
@@ -95,13 +95,12 @@ class ExperimentalConfig(Config):
        # MSC2815 (allow room moderators to view redacted event content)
        self.msc2815_enabled: bool = experimental.get("msc2815_enabled", False)

-        # MSC3786 (Add a default push rule to ignore m.room.server_acl events)
-        self.msc3786_enabled: bool = experimental.get("msc3786_enabled", False)
-
        # MSC3771: Thread read receipts
        self.msc3771_enabled: bool = experimental.get("msc3771_enabled", False)
        # MSC3772: A push rule for mutual relations.
        self.msc3772_enabled: bool = experimental.get("msc3772_enabled", False)
+        # MSC3773: Thread notifications
+        self.msc3773_enabled: bool = experimental.get("msc3773_enabled", False)

        # MSC3715: dir param on /relations.
        self.msc3715_enabled: bool = experimental.get("msc3715_enabled", False)
--- a/synapse/config/metrics.py
+++ b/synapse/config/metrics.py
@@ -43,32 +43,6 @@ class MetricsConfig(Config):
    def read_config(self, config: JsonDict, **kwargs: Any) -> None:
        self.enable_metrics = config.get("enable_metrics", False)

-        """
-        ### `enable_legacy_metrics` (experimental)
-
-        **Experimental: this option may be removed or have its behaviour
-        changed at any time, with no notice.**
-
-        Set to `true` to publish both legacy and non-legacy Prometheus metric names,
-        or to `false` to only publish non-legacy Prometheus metric names.
-        Defaults to `true`. Has no effect if `enable_metrics` is `false`.
-
-        Legacy metric names include:
-        - metrics containing colons in the name, such as `synapse_util_caches_response_cache:hits`, because colons are supposed to be reserved for user-defined recording rules;
-        - counters that don't end with the `_total` suffix, such as `synapse_federation_client_sent_edus`, therefore not adhering to the OpenMetrics standard.
-
-        These legacy metric names are unconventional and not compliant with OpenMetrics standards.
-        They are included for backwards compatibility.
-
-        Example configuration:
-        ```yaml
-        enable_legacy_metrics: false
-        ```
-
-        See https://github.com/matrix-org/synapse/issues/11106 for context.
-
-        *Since v1.67.0.*
-        """
        self.enable_legacy_metrics = config.get("enable_legacy_metrics", True)

        self.report_stats = config.get("report_stats", None)
--- a/synapse/federation/federation_base.py
+++ b/synapse/federation/federation_base.py
@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Awaitable, Callable, Optional

 from synapse.api.constants import MAX_DEPTH, EventContentFields, EventTypes, Membership
 from synapse.api.errors import Codes, SynapseError
@@ -58,7 +58,12 @@ class FederationBase:

    @trace
    async def _check_sigs_and_hash(
-        self, room_version: RoomVersion, pdu: EventBase
+        self,
+        room_version: RoomVersion,
+        pdu: EventBase,
+        record_failure_callback: Optional[
+            Callable[[EventBase, str], Awaitable[None]]
+        ] = None,
    ) -> EventBase:
        """Checks that event is correctly signed by the sending server.

@@ -70,6 +75,11 @@ class FederationBase:
        Args:
            room_version: The room version of the PDU
            pdu: the event to be checked
+            record_failure_callback: A callback to run whenever the given event
+                fails signature or hash checks. This includes exceptions
+                that would be normally be thrown/raised but also things like
+                checking for event tampering where we just return the redacted
+                event.

        Returns:
              * the original event if the checks pass
@@ -80,7 +90,12 @@ class FederationBase:
          InvalidEventSignatureError if the signature check failed. Nothing
             will be logged in this case.
        """
-        await _check_sigs_on_pdu(self.keyring, room_version, pdu)
+        try:
+            await _check_sigs_on_pdu(self.keyring, room_version, pdu)
+        except InvalidEventSignatureError as exc:
+            if record_failure_callback:
+                await record_failure_callback(pdu, str(exc))
+            raise exc

        if not check_event_content_hash(pdu):
            # let's try to distinguish between failures because the event was
@@ -116,6 +131,10 @@ class FederationBase:
                        "event_id": pdu.event_id,
                    }
                )
+                if record_failure_callback:
+                    await record_failure_callback(
+                        pdu, "Event content has been tampered with"
+                    )
            return redacted_event

        spam_check = await self.spam_checker.check_event_for_spam(pdu)
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -278,7 +278,7 @@ class FederationClient(FederationBase):
        pdus = [event_from_pdu_json(p, room_version) for p in transaction_data_pdus]

        # Check signatures and hash of pdus, removing any from the list that fail checks
-        pdus[:] = await self._check_sigs_and_hash_and_fetch(
+        pdus[:] = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
            dest, pdus, room_version=room_version
        )

@@ -328,7 +328,17 @@ class FederationClient(FederationBase):

            # Check signatures are correct.
            try:
-                signed_pdu = await self._check_sigs_and_hash(room_version, pdu)
+
+                async def _record_failure_callback(
+                    event: EventBase, cause: str
+                ) -> None:
+                    await self.store.record_event_failed_pull_attempt(
+                        event.room_id, event.event_id, cause
+                    )
+
+                signed_pdu = await self._check_sigs_and_hash(
+                    room_version, pdu, _record_failure_callback
+                )
            except InvalidEventSignatureError as e:
                errmsg = f"event id {pdu.event_id}: {e}"
                logger.warning("%s", errmsg)
@@ -547,24 +557,28 @@ class FederationClient(FederationBase):
            len(auth_event_map),
        )

-        valid_auth_events = await self._check_sigs_and_hash_and_fetch(
+        valid_auth_events = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
            destination, auth_event_map.values(), room_version
        )

-        valid_state_events = await self._check_sigs_and_hash_and_fetch(
-            destination, state_event_map.values(), room_version
+        valid_state_events = (
+            await self._check_sigs_and_hash_for_pulled_events_and_fetch(
+                destination, state_event_map.values(), room_version
+            )
        )

        return valid_state_events, valid_auth_events

    @trace
-    async def _check_sigs_and_hash_and_fetch(
+    async def _check_sigs_and_hash_for_pulled_events_and_fetch(
        self,
        origin: str,
        pdus: Collection[EventBase],
        room_version: RoomVersion,
    ) -> List[EventBase]:
-        """Checks the signatures and hashes of a list of events.
+        """
+        Checks the signatures and hashes of a list of pulled events we got from
+        federation and records any signature failures as failed pull attempts.

        If a PDU fails its signature check then we check if we have it in
        the database, and if not then request it from the sender's server (if that
@@ -597,11 +611,17 @@ class FederationClient(FederationBase):

        valid_pdus: List[EventBase] = []

+        async def _record_failure_callback(event: EventBase, cause: str) -> None:
+            await self.store.record_event_failed_pull_attempt(
+                event.room_id, event.event_id, cause
+            )
+
        async def _execute(pdu: EventBase) -> None:
            valid_pdu = await self._check_sigs_and_hash_and_fetch_one(
                pdu=pdu,
                origin=origin,
                room_version=room_version,
+                record_failure_callback=_record_failure_callback,
            )

            if valid_pdu:
@@ -618,6 +638,9 @@ class FederationClient(FederationBase):
        pdu: EventBase,
        origin: str,
        room_version: RoomVersion,
+        record_failure_callback: Optional[
+            Callable[[EventBase, str], Awaitable[None]]
+        ] = None,
    ) -> Optional[EventBase]:
        """Takes a PDU and checks its signatures and hashes.

@@ -634,6 +657,11 @@ class FederationClient(FederationBase):
            origin
            pdu
            room_version
+            record_failure_callback: A callback to run whenever the given event
+                fails signature or hash checks. This includes exceptions
+                that would be normally be thrown/raised but also things like
+                checking for event tampering where we just return the redacted
+                event.

        Returns:
            The PDU (possibly redacted) if it has valid signatures and hashes.
@@ -641,7 +669,9 @@ class FederationClient(FederationBase):
        """

        try:
-            return await self._check_sigs_and_hash(room_version, pdu)
+            return await self._check_sigs_and_hash(
+                room_version, pdu, record_failure_callback
+            )
        except InvalidEventSignatureError as e:
            logger.warning(
                "Signature on retrieved event %s was invalid (%s). "
@@ -694,7 +724,7 @@ class FederationClient(FederationBase):

        auth_chain = [event_from_pdu_json(p, room_version) for p in res["auth_chain"]]

-        signed_auth = await self._check_sigs_and_hash_and_fetch(
+        signed_auth = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
            destination, auth_chain, room_version=room_version
        )

@@ -1401,7 +1431,7 @@ class FederationClient(FederationBase):
                event_from_pdu_json(e, room_version) for e in content.get("events", [])
            ]

-            signed_events = await self._check_sigs_and_hash_and_fetch(
+            signed_events = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
                destination, events, room_version=room_version
            )
        except HttpResponseException as e:
--- a/synapse/handlers/relations.py
+++ b/synapse/handlers/relations.py
@@ -78,6 +78,7 @@ class RelationsHandler:
        direction: str = "b",
        from_token: Optional[StreamToken] = None,
        to_token: Optional[StreamToken] = None,
+        include_original_event: bool = False,
    ) -> JsonDict:
        """Get related events of a event, ordered by topological ordering.

@@ -94,6 +95,7 @@ class RelationsHandler:
                oldest first (`"f"`).
            from_token: Fetch rows from the given token, or from the start if None.
            to_token: Fetch rows up to the given token, or up to the end if None.
+            include_original_event: Whether to include the parent event.

        Returns:
            The pagination chunk.
@@ -138,25 +140,24 @@ class RelationsHandler:
            is_peeking=(member_event_id is None),
        )

-        now = self._clock.time_msec()
-        # Do not bundle aggregations when retrieving the original event because
-        # we want the content before relations are applied to it.
-        original_event = self._event_serializer.serialize_event(
-            event, now, bundle_aggregations=None
-        )
        # The relations returned for the requested event do include their
        # bundled aggregations.
        aggregations = await self.get_bundled_aggregations(
            events, requester.user.to_string()
        )
-        serialized_events = self._event_serializer.serialize_events(
-            events, now, bundle_aggregations=aggregations
-        )

-        return_value = {
-            "chunk": serialized_events,
-            "original_event": original_event,
+        now = self._clock.time_msec()
+        return_value: JsonDict = {
+            "chunk": self._event_serializer.serialize_events(
+                events, now, bundle_aggregations=aggregations
+            ),
        }
+        if include_original_event:
+            # Do not bundle aggregations when retrieving the original event because
+            # we want the content before relations are applied to it.
+            return_value["original_event"] = self._event_serializer.serialize_event(
+                event, now, bundle_aggregations=None
+            )

        if next_token:
            return_value["next_batch"] = await next_token.to_string(self._main_store)
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -40,7 +40,7 @@ from synapse.handlers.relations import BundledAggregations
 from synapse.logging.context import current_context
 from synapse.logging.opentracing import SynapseTags, log_kv, set_tag, start_active_span
 from synapse.push.clientformat import format_push_rules_for_user
-from synapse.storage.databases.main.event_push_actions import NotifCounts
+from synapse.storage.databases.main.event_push_actions import RoomNotifCounts
 from synapse.storage.roommember import MemberSummary
 from synapse.storage.state import StateFilter
 from synapse.types import (
@@ -128,6 +128,7 @@ class JoinedSyncResult:
    ephemeral: List[JsonDict]
    account_data: List[JsonDict]
    unread_notifications: JsonDict
+    unread_thread_notifications: JsonDict
    summary: Optional[JsonDict]
    unread_count: int

@@ -278,6 +279,8 @@ class SyncHandler:

        self.rooms_to_exclude = hs.config.server.rooms_to_exclude_from_sync

+        self._msc3773_enabled = hs.config.experimental.msc3773_enabled
+
    async def wait_for_sync_for_user(
        self,
        requester: Requester,
@@ -1288,7 +1291,7 @@ class SyncHandler:

    async def unread_notifs_for_room_id(
        self, room_id: str, sync_config: SyncConfig
-    ) -> NotifCounts:
+    ) -> RoomNotifCounts:
        with Measure(self.clock, "unread_notifs_for_room_id"):

            return await self.store.get_unread_event_push_actions_by_room_for_user(
@@ -2353,6 +2356,7 @@ class SyncHandler:
                    ephemeral=ephemeral,
                    account_data=account_data_events,
                    unread_notifications=unread_notifications,
+                    unread_thread_notifications={},
                    summary=summary,
                    unread_count=0,
                )
@@ -2360,10 +2364,36 @@ class SyncHandler:
                if room_sync or always_include:
                    notifs = await self.unread_notifs_for_room_id(room_id, sync_config)

-                    unread_notifications["notification_count"] = notifs.notify_count
-                    unread_notifications["highlight_count"] = notifs.highlight_count
+                    # Notifications for the main timeline.
+                    notify_count = notifs.main_timeline.notify_count
+                    highlight_count = notifs.main_timeline.highlight_count
+                    unread_count = notifs.main_timeline.unread_count

-                    room_sync.unread_count = notifs.unread_count
+                    # Check the sync configuration.
+                    if (
+                        self._msc3773_enabled
+                        and sync_config.filter_collection.unread_thread_notifications()
+                    ):
+                        # And add info for each thread.
+                        room_sync.unread_thread_notifications = {
+                            thread_id: {
+                                "notification_count": thread_notifs.notify_count,
+                                "highlight_count": thread_notifs.highlight_count,
+                            }
+                            for thread_id, thread_notifs in notifs.threads.items()
+                            if thread_id is not None
+                        }
+
+                    else:
+                        # Combine the unread counts for all threads and main timeline.
+                        for thread_notifs in notifs.threads.values():
+                            notify_count += thread_notifs.notify_count
+                            highlight_count += thread_notifs.highlight_count
+                            unread_count += thread_notifs.unread_count
+
+                    unread_notifications["notification_count"] = notify_count
+                    unread_notifications["highlight_count"] = highlight_count
+                    room_sync.unread_count = unread_count

                    sync_result_builder.joined.append(room_sync)

--- a/synapse/push/bulk_push_rule_evaluator.py
+++ b/synapse/push/bulk_push_rule_evaluator.py
@@ -31,7 +31,7 @@ from typing import (

 from prometheus_client import Counter

-from synapse.api.constants import EventTypes, Membership, RelationTypes
+from synapse.api.constants import MAIN_TIMELINE, EventTypes, Membership, RelationTypes
 from synapse.event_auth import auth_types_for_event, get_user_power_level
 from synapse.events import EventBase, relation_from_event
 from synapse.events.snapshot import EventContext
@@ -280,7 +280,7 @@ class BulkPushRuleEvaluator:
        # If the event does not have a relation, then cannot have any mutual
        # relations or thread ID.
        relations = {}
-        thread_id = "main"
+        thread_id = MAIN_TIMELINE
        if relation:
            relations = await self._get_mutual_relations(
                relation.parent_id,
--- a/synapse/push/push_tools.py
+++ b/synapse/push/push_tools.py
@@ -39,7 +39,12 @@ async def get_badge_count(store: DataStore, user_id: str, group_by_room: bool) -
    await concurrently_execute(get_room_unread_count, joins, 10)

    for notifs in room_notifs:
-        if notifs.notify_count == 0:
+        # Combine the counts from all the threads.
+        notify_count = notifs.main_timeline.notify_count + sum(
+            n.notify_count for n in notifs.threads.values()
+        )
+
+        if notify_count == 0:
            continue

        if group_by_room:
@@ -47,7 +52,7 @@ async def get_badge_count(store: DataStore, user_id: str, group_by_room: bool) -
            badge += 1
        else:
            # increment the badge count by the number of unread messages in the room
-            badge += notifs.notify_count
+            badge += notify_count
    return badge


--- a/synapse/rest/client/relations.py
+++ b/synapse/rest/client/relations.py
@@ -82,6 +82,11 @@ class RelationPaginationServlet(RestServlet):
        if to_token_str:
            to_token = await StreamToken.from_string(self.store, to_token_str)

+        # The unstable version of this API returns an extra field for client
+        # compatibility, see https://github.com/matrix-org/synapse/issues/12930.
+        assert request.path is not None
+        include_original_event = request.path.startswith(b"/_matrix/client/unstable/")
+
        result = await self._relations_handler.get_relations(
            requester=requester,
            event_id=parent_id,
@@ -92,6 +97,7 @@ class RelationPaginationServlet(RestServlet):
            direction=direction,
            from_token=from_token,
            to_token=to_token,
+            include_original_event=include_original_event,
        )

        return 200, result
--- a/synapse/rest/client/sync.py
+++ b/synapse/rest/client/sync.py
@@ -509,6 +509,10 @@ class SyncRestServlet(RestServlet):
            ephemeral_events = room.ephemeral
            result["ephemeral"] = {"events": ephemeral_events}
            result["unread_notifications"] = room.unread_notifications
+            if room.unread_thread_notifications:
+                result[
+                    "org.matrix.msc3773.unread_thread_notifications"
+                ] = room.unread_thread_notifications
            result["summary"] = room.summary
            if self._msc2654_enabled:
                result["org.matrix.msc2654.unread_count"] = room.unread_count
--- a/synapse/rest/client/versions.py
+++ b/synapse/rest/client/versions.py
@@ -75,6 +75,7 @@ class VersionsRestServlet(RestServlet):
                    "r0.6.1",
                    "v1.1",
                    "v1.2",
+                    "v1.3",
                ],
                # as per MSC1497:
                "unstable_features": {
@@ -103,8 +104,9 @@ class VersionsRestServlet(RestServlet):
                    "org.matrix.msc3030": self.config.experimental.msc3030_enabled,
                    # Adds support for thread relations, per MSC3440.
                    "org.matrix.msc3440.stable": True,  # TODO: remove when "v1.3" is added above
-                    # Support for thread read receipts.
+                    # Support for thread read receipts & notification counts.
                    "org.matrix.msc3771": self.config.experimental.msc3771_enabled,
+                    "org.matrix.msc3773": self.config.experimental.msc3773_enabled,
                    # Allows moderators to fetch redacted event content as described in MSC2815
                    "fi.mau.msc2815": self.config.experimental.msc2815_enabled,
                    # Adds support for login token requests as per MSC3882
--- a/synapse/storage/database.py
+++ b/synapse/storage/database.py
@@ -94,7 +94,7 @@ UNIQUE_INDEX_BACKGROUND_UPDATES = {
    "event_search": "event_search_event_id_idx",
    "local_media_repository_thumbnails": "local_media_repository_thumbnails_method_idx",
    "remote_media_cache_thumbnails": "remote_media_repository_thumbnails_method_idx",
-    "event_push_summary": "event_push_summary_unique_index",
+    "event_push_summary": "event_push_summary_unique_index2",
    "receipts_linearized": "receipts_linearized_unique_index",
    "receipts_graph": "receipts_graph_unique_index",
 }
--- a/synapse/storage/databases/main/event_push_actions.py
+++ b/synapse/storage/databases/main/event_push_actions.py
@@ -88,7 +88,7 @@ from typing import (

 import attr

-from synapse.api.constants import ReceiptTypes
+from synapse.api.constants import MAIN_TIMELINE, ReceiptTypes
 from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
 from synapse.storage.database import (
@@ -157,7 +157,7 @@ class UserPushAction(EmailPushAction):
@attr.s(slots=True, auto_attribs=True)
 class NotifCounts:
    """
-    The per-user, per-room count of notifications. Used by sync and push.
+    The per-user, per-room, per-thread count of notifications. Used by sync and push.
    """

    notify_count: int = 0
@@ -165,6 +165,21 @@ class NotifCounts:
    highlight_count: int = 0


+@attr.s(slots=True, auto_attribs=True)
+class RoomNotifCounts:
+    """
+    The per-user, per-room count of notifications. Used by sync and push.
+    """
+
+    main_timeline: NotifCounts
+    # Map of thread ID to the notification counts.
+    threads: Dict[str, NotifCounts]
+
+    def __len__(self) -> int:
+        # To properly account for the amount of space in any caches.
+        return len(self.threads) + 1
+
+
 def _serialize_action(
    actions: Collection[Union[Mapping, str]], is_highlight: bool
 ) -> str:
@@ -205,6 +220,9 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
    ):
        super().__init__(database, db_conn, hs)

+        # Track when the process started.
+        self._started_ts = self._clock.time_msec()
+
        # These get correctly set by _find_stream_orderings_for_times_txn
        self.stream_ordering_month_ago: Optional[int] = None
        self.stream_ordering_day_ago: Optional[int] = None
@@ -224,6 +242,10 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                self._rotate_notifs, 30 * 1000
            )

+            self._clear_old_staging_loop = self._clock.looping_call(
+                self._clear_old_push_actions_staging, 30 * 60 * 1000
+            )
+
        self.db_pool.updates.register_background_index_update(
            "event_push_summary_unique_index",
            index_name="event_push_summary_unique_index",
@@ -331,12 +353,12 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas

        return result

-    @cached(tree=True, max_entries=5000)
+    @cached(tree=True, max_entries=5000, iterable=True)
    async def get_unread_event_push_actions_by_room_for_user(
        self,
        room_id: str,
        user_id: str,
-    ) -> NotifCounts:
+    ) -> RoomNotifCounts:
        """Get the notification count, the highlight count and the unread message count
        for a given user in a given room after their latest read receipt.

@@ -349,8 +371,9 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            user_id: The user to retrieve the counts for.

        Returns
-            A NotifCounts object containing the notification count, the highlight count
-            and the unread message count.
+            A RoomNotifCounts object containing the notification count, the
+            highlight count and the unread message count for both the main timeline
+            and threads.
        """
        return await self.db_pool.runInteraction(
            "get_unread_event_push_actions_by_room",
@@ -364,7 +387,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        txn: LoggingTransaction,
        room_id: str,
        user_id: str,
-    ) -> NotifCounts:
+    ) -> RoomNotifCounts:
        # Get the stream ordering of the user's latest receipt in the room.
        result = self.get_last_unthreaded_receipt_for_user_txn(
            txn,
@@ -399,7 +422,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        room_id: str,
        user_id: str,
        receipt_stream_ordering: int,
-    ) -> NotifCounts:
+    ) -> RoomNotifCounts:
        """Get the number of unread messages for a user/room that have happened
        since the given stream ordering.

@@ -411,12 +434,19 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                receipt in the room. If there are no receipts, the stream ordering
                of the user's join event.

-        Returns
-            A NotifCounts object containing the notification count, the highlight count
-            and the unread message count.
+        Returns:
+            A RoomNotifCounts object containing the notification count, the
+            highlight count and the unread message count for both the main timeline
+            and threads.
        """

-        counts = NotifCounts()
+        main_counts = NotifCounts()
+        thread_counts: Dict[str, NotifCounts] = {}
+
+        def _get_thread(thread_id: str) -> NotifCounts:
+            if thread_id == MAIN_TIMELINE:
+                return main_counts
+            return thread_counts.setdefault(thread_id, NotifCounts())

        # First we pull the counts from the summary table.
        #
@@ -433,52 +463,61 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        # receipt).
        txn.execute(
            """
-                SELECT stream_ordering, notif_count, COALESCE(unread_count, 0)
+                SELECT stream_ordering, notif_count, COALESCE(unread_count, 0), thread_id
                FROM event_push_summary
                WHERE room_id = ? AND user_id = ?
                AND (
                    (last_receipt_stream_ordering IS NULL AND stream_ordering > ?)
                    OR last_receipt_stream_ordering = ?
-                )
+                ) AND (notif_count != 0 OR COALESCE(unread_count, 0) != 0)
            """,
            (room_id, user_id, receipt_stream_ordering, receipt_stream_ordering),
        )
-        row = txn.fetchone()
+        max_summary_stream_ordering = 0
+        for summary_stream_ordering, notif_count, unread_count, thread_id in txn:
+            counts = _get_thread(thread_id)
+            counts.notify_count += notif_count
+            counts.unread_count += unread_count

-        summary_stream_ordering = 0
-        if row:
-            summary_stream_ordering = row[0]
-            counts.notify_count += row[1]
-            counts.unread_count += row[2]
+            # Summaries will only be used if they have not been invalidated by
+            # a recent receipt; track the latest stream ordering or a valid summary.
+            #
+            # Note that since there's only one read receipt in the room per user,
+            # valid summaries are contiguous.
+            max_summary_stream_ordering = max(
+                summary_stream_ordering, max_summary_stream_ordering
+            )

        # Next we need to count highlights, which aren't summarised
        sql = """
-            SELECT COUNT(*) FROM event_push_actions
+            SELECT COUNT(*), thread_id FROM event_push_actions
            WHERE user_id = ?
                AND room_id = ?
                AND stream_ordering > ?
                AND highlight = 1
+            GROUP BY thread_id
        """
        txn.execute(sql, (user_id, room_id, receipt_stream_ordering))
-        row = txn.fetchone()
-        if row:
-            counts.highlight_count += row[0]
+        for highlight_count, thread_id in txn:
+            _get_thread(thread_id).highlight_count += highlight_count

        # Finally we need to count push actions that aren't included in the
        # summary returned above. This might be due to recent events that haven't
        # been summarised yet or the summary is out of date due to a recent read
        # receipt.
        start_unread_stream_ordering = max(
-            receipt_stream_ordering, summary_stream_ordering
+            receipt_stream_ordering, max_summary_stream_ordering
        )
-        notify_count, unread_count = self._get_notif_unread_count_for_user_room(
+        unread_counts = self._get_notif_unread_count_for_user_room(
            txn, room_id, user_id, start_unread_stream_ordering
        )

-        counts.notify_count += notify_count
-        counts.unread_count += unread_count
+        for notif_count, unread_count, thread_id in unread_counts:
+            counts = _get_thread(thread_id)
+            counts.notify_count += notif_count
+            counts.unread_count += unread_count

-        return counts
+        return RoomNotifCounts(main_counts, thread_counts)

    def _get_notif_unread_count_for_user_room(
        self,
@@ -487,7 +526,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        user_id: str,
        stream_ordering: int,
        max_stream_ordering: Optional[int] = None,
-    ) -> Tuple[int, int]:
+    ) -> List[Tuple[int, int, str]]:
        """Returns the notify and unread counts from `event_push_actions` for
        the given user/room in the given range.

@@ -503,13 +542,14 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                If this is not given, then no maximum is applied.

        Return:
-            A tuple of the notif count and unread count in the given range.
+            A tuple of the notif count and unread count in the given range for
+            each thread.
        """

        # If there have been no events in the room since the stream ordering,
        # there can't be any push actions either.
        if not self._events_stream_cache.has_entity_changed(room_id, stream_ordering):
-            return 0, 0
+            return []

        clause = ""
        args = [user_id, room_id, stream_ordering]
@@ -520,26 +560,23 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            # If the max stream ordering is less than the min stream ordering,
            # then obviously there are zero push actions in that range.
            if max_stream_ordering <= stream_ordering:
-                return 0, 0
+                return []

        sql = f"""
            SELECT
               COUNT(CASE WHEN notif = 1 THEN 1 END),
-               COUNT(CASE WHEN unread = 1 THEN 1 END)
-             FROM event_push_actions ea
-             WHERE user_id = ?
+               COUNT(CASE WHEN unread = 1 THEN 1 END),
+               thread_id
+            FROM event_push_actions ea
+            WHERE user_id = ?
               AND room_id = ?
               AND ea.stream_ordering > ?
               {clause}
+            GROUP BY thread_id
        """

        txn.execute(sql, args)
-        row = txn.fetchone()
-
-        if row:
-            return cast(Tuple[int, int], row)
-
-        return 0, 0
+        return cast(List[Tuple[int, int, str]], txn.fetchall())

    async def get_push_action_users_in_range(
        self, min_stream_ordering: int, max_stream_ordering: int
@@ -791,7 +828,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        # can be used to insert into the `event_push_actions_staging` table.
        def _gen_entry(
            user_id: str, actions: Collection[Union[Mapping, str]]
-        ) -> Tuple[str, str, str, int, int, int, str]:
+        ) -> Tuple[str, str, str, int, int, int, str, int]:
            is_highlight = 1 if _action_has_highlight(actions) else 0
            notif = 1 if "notify" in actions else 0
            return (
@@ -802,6 +839,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                is_highlight,  # highlight column
                int(count_as_unread),  # unread column
                thread_id,  # thread_id column
+                self._clock.time_msec(),  # inserted_ts column
            )

        await self.db_pool.simple_insert_many(
@@ -814,6 +852,7 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
                "highlight",
                "unread",
                "thread_id",
+                "inserted_ts",
            ),
            values=[
                _gen_entry(user_id, actions)
@@ -1090,26 +1129,34 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas

            # Fetch the notification counts between the stream ordering of the
            # latest receipt and what was previously summarised.
-            notif_count, unread_count = self._get_notif_unread_count_for_user_room(
+            unread_counts = self._get_notif_unread_count_for_user_room(
                txn, room_id, user_id, stream_ordering, old_rotate_stream_ordering
            )

-            # Replace the previous summary with the new counts.
-            #
-            # TODO(threads): Upsert per-thread instead of setting them all to main.
-            self.db_pool.simple_upsert_txn(
+            # First mark the summary for all threads in the room as cleared.
+            self.db_pool.simple_update_txn(
                txn,
                table="event_push_summary",
-                keyvalues={"room_id": room_id, "user_id": user_id},
-                values={
-                    "notif_count": notif_count,
-                    "unread_count": unread_count,
+                keyvalues={"user_id": user_id, "room_id": room_id},
+                updatevalues={
+                    "notif_count": 0,
+                    "unread_count": 0,
                    "stream_ordering": old_rotate_stream_ordering,
                    "last_receipt_stream_ordering": stream_ordering,
-                    "thread_id": "main",
                },
            )

+            # Then any updated threads get their notification count and unread
+            # count updated.
+            self.db_pool.simple_update_many_txn(
+                txn,
+                table="event_push_summary",
+                key_names=("room_id", "user_id", "thread_id"),
+                key_values=[(room_id, user_id, row[2]) for row in unread_counts],
+                value_names=("notif_count", "unread_count"),
+                value_values=[(row[0], row[1]) for row in unread_counts],
+            )
+
        # We always update `event_push_summary_last_receipt_stream_id` to
        # ensure that we don't rescan the same receipts for remote users.

@@ -1195,23 +1242,23 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas

        # Calculate the new counts that should be upserted into event_push_summary
        sql = """
-            SELECT user_id, room_id,
+            SELECT user_id, room_id, thread_id,
                coalesce(old.%s, 0) + upd.cnt,
                upd.stream_ordering
            FROM (
-                SELECT user_id, room_id, count(*) as cnt,
+                SELECT user_id, room_id, thread_id, count(*) as cnt,
                    max(ea.stream_ordering) as stream_ordering
                FROM event_push_actions AS ea
-                LEFT JOIN event_push_summary AS old USING (user_id, room_id)
+                LEFT JOIN event_push_summary AS old USING (user_id, room_id, thread_id)
                WHERE ? < ea.stream_ordering AND ea.stream_ordering <= ?
                    AND (
                        old.last_receipt_stream_ordering IS NULL
                        OR old.last_receipt_stream_ordering < ea.stream_ordering
                    )
                    AND %s = 1
-                GROUP BY user_id, room_id
+                GROUP BY user_id, room_id, thread_id
            ) AS upd
-            LEFT JOIN event_push_summary AS old USING (user_id, room_id)
+            LEFT JOIN event_push_summary AS old USING (user_id, room_id, thread_id)
        """

        # First get the count of unread messages.
@@ -1225,11 +1272,11 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        # object because we might not have the same amount of rows in each of them. To do
        # this, we use a dict indexed on the user ID and room ID to make it easier to
        # populate.
-        summaries: Dict[Tuple[str, str], _EventPushSummary] = {}
+        summaries: Dict[Tuple[str, str, str], _EventPushSummary] = {}
        for row in txn:
-            summaries[(row[0], row[1])] = _EventPushSummary(
-                unread_count=row[2],
-                stream_ordering=row[3],
+            summaries[(row[0], row[1], row[2])] = _EventPushSummary(
+                unread_count=row[3],
+                stream_ordering=row[4],
                notif_count=0,
            )

@@ -1240,34 +1287,35 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        )

        for row in txn:
-            if (row[0], row[1]) in summaries:
-                summaries[(row[0], row[1])].notif_count = row[2]
+            if (row[0], row[1], row[2]) in summaries:
+                summaries[(row[0], row[1], row[2])].notif_count = row[3]
            else:
                # Because the rules on notifying are different than the rules on marking
                # a message unread, we might end up with messages that notify but aren't
                # marked unread, so we might not have a summary for this (user, room)
                # tuple to complete.
-                summaries[(row[0], row[1])] = _EventPushSummary(
+                summaries[(row[0], row[1], row[2])] = _EventPushSummary(
                    unread_count=0,
-                    stream_ordering=row[3],
-                    notif_count=row[2],
+                    stream_ordering=row[4],
+                    notif_count=row[3],
                )

        logger.info("Rotating notifications, handling %d rows", len(summaries))

-        # TODO(threads): Update on a per-thread basis.
        self.db_pool.simple_upsert_many_txn(
            txn,
            table="event_push_summary",
-            key_names=("user_id", "room_id"),
-            key_values=[(user_id, room_id) for user_id, room_id in summaries],
-            value_names=("notif_count", "unread_count", "stream_ordering", "thread_id"),
+            key_names=("user_id", "room_id", "thread_id"),
+            key_values=[
+                (user_id, room_id, thread_id)
+                for user_id, room_id, thread_id in summaries
+            ],
+            value_names=("notif_count", "unread_count", "stream_ordering"),
            value_values=[
                (
                    summary.notif_count,
                    summary.unread_count,
                    summary.stream_ordering,
-                    "main",
                )
                for summary in summaries.values()
            ],
@@ -1279,7 +1327,10 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
        )

    async def _remove_old_push_actions_that_have_rotated(self) -> None:
-        """Clear out old push actions that have been summarised."""
+        """
+        Clear out old push actions that have been summarised (and are older than
+        1 day ago).
+        """

        # We want to clear out anything that is older than a day that *has* already
        # been rotated.
@@ -1340,6 +1391,53 @@ class EventPushActionsWorkerStore(ReceiptsWorkerStore, StreamWorkerStore, SQLBas
            if done:
                break

+    @wrap_as_background_process("_clear_old_push_actions_staging")
+    async def _clear_old_push_actions_staging(self) -> None:
+        """Clear out any old event push actions from the staging table for
+        events that we failed to persist.
+        """
+
+        # We delete anything more than an hour old, on the assumption that we'll
+        # never take more than an hour to persist an event.
+        delete_before_ts = self._clock.time_msec() - 60 * 60 * 1000
+
+        if self._started_ts > delete_before_ts:
+            # We need to wait for at least an hour before we started deleting,
+            # so that we know it's safe to delete rows with NULL `inserted_ts`.
+            return
+
+        # We don't have an index on `inserted_ts`, instead we assume that the
+        # number of "live" rows in `event_push_actions_staging` is small enough
+        # that an infrequent periodic scan won't cause a problem.
+        #
+        # Note: we also delete any columns with NULL `inserted_ts`, this is safe
+        # as we added a default value to new rows and so they must be at least
+        # an hour old.
+        limit = 1000
+        sql = """
+            DELETE FROM event_push_actions_staging WHERE event_id IN (
+                SELECT event_id FROM event_push_actions_staging WHERE
+                inserted_ts < ? OR inserted_ts IS NULL
+                LIMIT ?
+            )
+        """
+
+        def _clear_old_push_actions_staging_txn(txn: LoggingTransaction) -> bool:
+            txn.execute(sql, (delete_before_ts, limit))
+            return txn.rowcount >= limit
+
+        while True:
+            # Returns true if we have more stuff to delete from the table.
+            deleted = await self.db_pool.runInteraction(
+                "_clear_old_push_actions_staging", _clear_old_push_actions_staging_txn
+            )
+
+            if not deleted:
+                return
+
+            # We sleep to ensure that we don't overwhelm the DB.
+            await self._clock.sleep(1.0)
+

 class EventPushActionsStore(EventPushActionsWorkerStore):
    EPA_HIGHLIGHT_INDEX = "epa_highlight_index"
--- a/synapse/storage/databases/main/push_rule.py
+++ b/synapse/storage/databases/main/push_rule.py
@@ -81,15 +81,10 @@ def _load_rules(
        for rawrule in rawrules
    ]

-    push_rules = PushRules(
-        ruleslist,
-    )
+    push_rules = PushRules(ruleslist)

    filtered_rules = FilteredPushRules(
-        push_rules,
-        enabled_map,
-        msc3786_enabled=experimental_config.msc3786_enabled,
-        msc3772_enabled=experimental_config.msc3772_enabled,
+        push_rules, enabled_map, msc3772_enabled=experimental_config.msc3772_enabled
    )

    return filtered_rules
--- a/synapse/storage/schema/init.py
+++ b/synapse/storage/schema/init.py
@@ -85,13 +85,14 @@ Changes in SCHEMA_VERSION = 73;
      events over federation.
    - Add indexes to various tables (`event_failed_pull_attempts`, `insertion_events`,
      `batch_events`) to make it easy to delete all associated rows when purging a room.
+    - `inserted_ts` column is added to `event_push_actions_staging` table.
 """


 SCHEMA_COMPAT_VERSION = (
-    # The groups tables are no longer accessible, so synapses with SCHEMA_VERSION < 72
-    # could break.
-    72
+    # The threads_id column must exist for event_push_actions, event_push_summary,
+    # receipts_linearized, and receipts_graph.
+    73
 )
 """Limit on how far the synapse codebase can be rolled back without breaking db compat

--- a/synapse/storage/schema/main/delta/73/05old_push_actions.sql.postgres
+++ b/synapse/storage/schema/main/delta/73/05old_push_actions.sql.postgres
@@ -0,0 +1,22 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- Add a column so that we know when a push action was inserted, to make it
+-- easier to clear out old ones.
+ALTER TABLE event_push_actions_staging ADD COLUMN inserted_ts BIGINT;
+
+-- We now add a default for *new* rows. We don't do this above as we don't want
+-- to have to update every remove with the new default.
+ALTER TABLE event_push_actions_staging ALTER COLUMN inserted_ts SET DEFAULT extract(epoch from now()) * 1000;
--- a/synapse/storage/schema/main/delta/73/05old_push_actions.sql.sqlite
+++ b/synapse/storage/schema/main/delta/73/05old_push_actions.sql.sqlite
@@ -0,0 +1,24 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- On SQLite we must be in monolith mode and updating the database from Synapse,
+-- so its safe to assume that `event_push_actions_staging` should be empty (as
+-- over restart an event must either have been fully persisted or we'll
+-- recalculate the push actions)
+DELETE FROM event_push_actions_staging;
+
+-- Add a column so that we know when a push action was inserted, to make it
+-- easier to clear out old ones.
+ALTER TABLE event_push_actions_staging ADD COLUMN inserted_ts BIGINT;
--- a/synapse/storage/schema/main/delta/73/06thread_notifications_backfill.sql
+++ b/synapse/storage/schema/main/delta/73/06thread_notifications_backfill.sql
@@ -0,0 +1,29 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- Forces the background updates from 06thread_notifications.sql to run in the
+-- foreground as code will now require those to be "done".
+
+DELETE FROM background_updates WHERE update_name = 'event_push_backfill_thread_id';
+
+-- Overwrite any null thread_id columns.
+UPDATE event_push_actions_staging SET thread_id = 'main' WHERE thread_id IS NULL;
+UPDATE event_push_actions SET thread_id = 'main' WHERE thread_id IS NULL;
+UPDATE event_push_summary SET thread_id = 'main' WHERE thread_id IS NULL;
+
+-- Do not run the event_push_summary_unique_index job if it is pending; the
+-- thread_id field will be made required.
+DELETE FROM background_updates WHERE update_name = 'event_push_summary_unique_index';
+DROP INDEX IF EXISTS event_push_summary_unique_index;
--- a/synapse/storage/schema/main/delta/73/07thread_notifications_not_null.sql.postgres
+++ b/synapse/storage/schema/main/delta/73/07thread_notifications_not_null.sql.postgres
@@ -0,0 +1,19 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- The columns can now be made non-nullable.
+ALTER TABLE event_push_actions_staging ALTER COLUMN thread_id SET NOT NULL;
+ALTER TABLE event_push_actions ALTER COLUMN thread_id SET NOT NULL;
+ALTER TABLE event_push_summary ALTER COLUMN thread_id SET NOT NULL;
--- a/synapse/storage/schema/main/delta/73/07thread_notifications_not_null.sql.sqlite
+++ b/synapse/storage/schema/main/delta/73/07thread_notifications_not_null.sql.sqlite
@@ -0,0 +1,101 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+-- SQLite doesn't support modifying columns to an existing table, so it must
+-- be recreated.
+
+-- Create the new tables.
+CREATE TABLE event_push_actions_staging_new (
+    event_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    actions TEXT NOT NULL,
+    notif SMALLINT NOT NULL,
+    highlight SMALLINT NOT NULL,
+    unread SMALLINT,
+    thread_id TEXT NOT NULL,
+    inserted_ts BIGINT
+);
+
+CREATE TABLE event_push_actions_new (
+    room_id TEXT NOT NULL,
+    event_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    profile_tag VARCHAR(32),
+    actions TEXT NOT NULL,
+    topological_ordering BIGINT,
+    stream_ordering BIGINT,
+    notif SMALLINT,
+    highlight SMALLINT,
+    unread SMALLINT,
+    thread_id TEXT NOT NULL,
+    CONSTRAINT event_id_user_id_profile_tag_uniqueness UNIQUE (room_id, event_id, user_id, profile_tag)
+);
+
+CREATE TABLE event_push_summary_new (
+    user_id TEXT NOT NULL,
+    room_id TEXT NOT NULL,
+    notif_count BIGINT NOT NULL,
+    stream_ordering BIGINT NOT NULL,
+    unread_count BIGINT,
+    last_receipt_stream_ordering BIGINT,
+    thread_id TEXT NOT NULL
+);
+
+-- Swap the indexes.
+DROP INDEX IF EXISTS event_push_actions_staging_id;
+CREATE INDEX event_push_actions_staging_id ON event_push_actions_staging_new(event_id);
+
+DROP INDEX IF EXISTS event_push_actions_room_id_user_id;
+DROP INDEX IF EXISTS event_push_actions_rm_tokens;
+DROP INDEX IF EXISTS event_push_actions_stream_ordering;
+DROP INDEX IF EXISTS event_push_actions_u_highlight;
+DROP INDEX IF EXISTS event_push_actions_highlights_index;
+CREATE INDEX event_push_actions_room_id_user_id on event_push_actions_new(room_id, user_id);
+CREATE INDEX event_push_actions_rm_tokens on event_push_actions_new( user_id, room_id, topological_ordering, stream_ordering );
+CREATE INDEX event_push_actions_stream_ordering on event_push_actions_new( stream_ordering, user_id );
+CREATE INDEX event_push_actions_u_highlight ON event_push_actions_new (user_id, stream_ordering);
+CREATE INDEX event_push_actions_highlights_index ON event_push_actions_new (user_id, room_id, topological_ordering, stream_ordering);
+
+-- Copy the data.
+INSERT INTO event_push_actions_staging_new (event_id, user_id, actions, notif, highlight, unread, thread_id, inserted_ts)
+    SELECT event_id, user_id, actions, notif, highlight, unread, thread_id, inserted_ts
+    FROM event_push_actions_staging;
+
+INSERT INTO event_push_actions_new (room_id, event_id, user_id, profile_tag, actions, topological_ordering, stream_ordering, notif, highlight, unread, thread_id)
+    SELECT room_id, event_id, user_id, profile_tag, actions, topological_ordering, stream_ordering, notif, highlight, unread, thread_id
+    FROM event_push_actions;
+
+INSERT INTO event_push_summary_new (user_id, room_id, notif_count, stream_ordering, unread_count, last_receipt_stream_ordering, thread_id)
+    SELECT user_id, room_id, notif_count, stream_ordering, unread_count, last_receipt_stream_ordering, thread_id
+    FROM event_push_summary;
+
+-- Drop the old tables.
+DROP TABLE event_push_actions_staging;
+DROP TABLE event_push_actions;
+DROP TABLE event_push_summary;
+
+-- Rename the tables.
+ALTER TABLE event_push_actions_staging_new RENAME TO event_push_actions_staging;
+ALTER TABLE event_push_actions_new RENAME TO event_push_actions;
+ALTER TABLE event_push_summary_new RENAME TO event_push_summary;
+
+-- Re-run background updates from 72/02event_push_actions_index.sql and
+-- 72/06thread_notifications.sql.
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+  (7307, 'event_push_summary_unique_index2', '{}')
+  ON CONFLICT (update_name) DO NOTHING;
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+  (7307, 'event_push_actions_stream_highlight_index', '{}')
+  ON CONFLICT (update_name) DO NOTHING;
--- a/tests/appservice/test_api.py
+++ b/tests/appservice/test_api.py
@@ -69,10 +69,14 @@ class ApplicationServiceApiTestCase(unittest.HomeserverTestCase):

        self.request_url = None

-        async def get_json(url: str, args: Mapping[Any, Any]) -> List[JsonDict]:
-            if not args.get(b"access_token"):
+        async def get_json(
+            url: str, args: Mapping[Any, Any], headers: Mapping[Any, Any]
+        ) -> List[JsonDict]:
+            # Ensure the access token is passed as both a header and query arg.
+            if not headers.get("Authorization") or not args.get(b"access_token"):
                raise RuntimeError("Access token not provided")

+            self.assertEqual(headers.get("Authorization"), f"Bearer {TOKEN}")
            self.assertEqual(args.get(b"access_token"), TOKEN)
            self.request_url = url
            if url == URL_USER:
--- a/tests/federation/test_federation_client.py
+++ b/tests/federation/test_federation_client.py
@@ -23,14 +23,23 @@ from twisted.test.proto_helpers import MemoryReactor

 from synapse.api.room_versions import RoomVersions
 from synapse.events import EventBase
+from synapse.rest import admin
+from synapse.rest.client import login, room
 from synapse.server import HomeServer
 from synapse.types import JsonDict
 from synapse.util import Clock

+from tests.test_utils import event_injection
 from tests.unittest import FederatingHomeserverTestCase


 class FederationClientTest(FederatingHomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        room.register_servlets,
+        login.register_servlets,
+    ]
+
    def prepare(self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer):
        super().prepare(reactor, clock, homeserver)

@@ -231,6 +240,72 @@ class FederationClientTest(FederatingHomeserverTestCase):

        return remote_pdu

+    def test_backfill_invalid_signature_records_failed_pull_attempts(
+        self,
+    ) -> None:
+        """
+        Test to make sure that events from /backfill with invalid signatures get
+        recorded as failed pull attempts.
+        """
+        OTHER_USER = f"@user:{self.OTHER_SERVER_NAME}"
+        main_store = self.hs.get_datastores().main
+
+        # Create the room
+        user_id = self.register_user("kermit", "test")
+        tok = self.login("kermit", "test")
+        room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
+
+        # We purposely don't run `add_hashes_and_signatures_from_other_server`
+        # over this because we want the signature check to fail.
+        pulled_event, _ = self.get_success(
+            event_injection.create_event(
+                self.hs,
+                room_id=room_id,
+                sender=OTHER_USER,
+                type="test_event_type",
+                content={"body": "garply"},
+            )
+        )
+
+        # We expect an outbound request to /backfill, so stub that out
+        self._mock_agent.request.side_effect = lambda *args, **kwargs: defer.succeed(
+            _mock_response(
+                {
+                    "origin": "yet.another.server",
+                    "origin_server_ts": 900,
+                    # Mimic the other server returning our new `pulled_event`
+                    "pdus": [pulled_event.get_pdu_json()],
+                }
+            )
+        )
+
+        self.get_success(
+            self.hs.get_federation_client().backfill(
+                # We use "yet.another.server" instead of
+                # `self.OTHER_SERVER_NAME` because we want to see the behavior
+                # from `_check_sigs_and_hash_and_fetch_one` where it tries to
+                # fetch the PDU again from the origin server if the signature
+                # fails. Just want to make sure that the failure is counted from
+                # both code paths.
+                dest="yet.another.server",
+                room_id=room_id,
+                limit=1,
+                extremities=[pulled_event.event_id],
+            ),
+        )
+
+        # Make sure our failed pull attempt was recorded
+        backfill_num_attempts = self.get_success(
+            main_store.db_pool.simple_select_one_onecol(
+                table="event_failed_pull_attempts",
+                keyvalues={"event_id": pulled_event.event_id},
+                retcol="num_attempts",
+            )
+        )
+        # This is 2 because it failed once from `self.OTHER_SERVER_NAME` and the
+        # other from "yet.another.server"
+        self.assertEqual(backfill_num_attempts, 2)
+

 def _mock_response(resp: JsonDict):
    body = json.dumps(resp).encode("utf-8")
--- a/tests/replication/slave/storage/test_events.py
+++ b/tests/replication/slave/storage/test_events.py
@@ -22,7 +22,10 @@ from synapse.api.room_versions import RoomVersions
 from synapse.events import FrozenEvent, _EventInternalMetadata, make_event_from_dict
 from synapse.handlers.room import RoomEventSource
 from synapse.replication.slave.storage.events import SlavedEventStore
-from synapse.storage.databases.main.event_push_actions import NotifCounts
+from synapse.storage.databases.main.event_push_actions import (
+    NotifCounts,
+    RoomNotifCounts,
+)
 from synapse.storage.roommember import GetRoomsForUserWithStreamOrdering, RoomsForUser
 from synapse.types import PersistedEventPosition

@@ -178,7 +181,9 @@ class SlavedEventStoreTestCase(BaseSlavedStoreTestCase):
        self.check(
            "get_unread_event_push_actions_by_room_for_user",
            [ROOM_ID, USER_ID_2],
-            NotifCounts(highlight_count=0, unread_count=0, notify_count=0),
+            RoomNotifCounts(
+                NotifCounts(highlight_count=0, unread_count=0, notify_count=0), {}
+            ),
        )

        self.persist(
@@ -191,7 +196,9 @@ class SlavedEventStoreTestCase(BaseSlavedStoreTestCase):
        self.check(
            "get_unread_event_push_actions_by_room_for_user",
            [ROOM_ID, USER_ID_2],
-            NotifCounts(highlight_count=0, unread_count=0, notify_count=1),
+            RoomNotifCounts(
+                NotifCounts(highlight_count=0, unread_count=0, notify_count=1), {}
+            ),
        )

        self.persist(
@@ -206,7 +213,9 @@ class SlavedEventStoreTestCase(BaseSlavedStoreTestCase):
        self.check(
            "get_unread_event_push_actions_by_room_for_user",
            [ROOM_ID, USER_ID_2],
-            NotifCounts(highlight_count=1, unread_count=0, notify_count=2),
+            RoomNotifCounts(
+                NotifCounts(highlight_count=1, unread_count=0, notify_count=2), {}
+            ),
        )

    def test_get_rooms_for_user_with_stream_ordering(self):
--- a/tests/rest/client/test_relations.py
+++ b/tests/rest/client/test_relations.py
@@ -654,6 +654,14 @@ class RelationsTestCase(BaseRelationsTestCase):
        )

        # We also expect to get the original event (the id of which is self.parent_id)
+        # when requesting the unstable endpoint.
+        self.assertNotIn("original_event", channel.json_body)
+        channel = self.make_request(
+            "GET",
+            f"/_matrix/client/unstable/rooms/{self.room}/relations/{self.parent_id}?limit=1",
+            access_token=self.user_token,
+        )
+        self.assertEqual(200, channel.code, channel.json_body)
        self.assertEqual(
            channel.json_body["original_event"]["event_id"], self.parent_id
        )
@@ -755,11 +763,6 @@ class RelationPaginationTestCase(BaseRelationsTestCase):
            channel.json_body["chunk"][0],
        )

-        # We also expect to get the original event (the id of which is self.parent_id)
-        self.assertEqual(
-            channel.json_body["original_event"]["event_id"], self.parent_id
-        )
-
        # Make sure next_batch has something in it that looks like it could be a
        # valid token.
        self.assertIsInstance(
--- a/tests/storage/test_event_push_actions.py
+++ b/tests/storage/test_event_push_actions.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-from typing import Tuple
+from typing import Optional, Tuple

 from twisted.test.proto_helpers import MemoryReactor

@@ -20,6 +20,7 @@ from synapse.rest import admin
 from synapse.rest.client import login, room
 from synapse.server import HomeServer
 from synapse.storage.databases.main.event_push_actions import NotifCounts
+from synapse.types import JsonDict
 from synapse.util import Clock

 from tests.unittest import HomeserverTestCase
@@ -133,13 +134,14 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
                )
            )
            self.assertEqual(
-                counts,
+                counts.main_timeline,
                NotifCounts(
                    notify_count=noitf_count,
                    unread_count=0,
                    highlight_count=highlight_count,
                ),
            )
+            self.assertEqual(counts.threads, {})

        def _create_event(highlight: bool = False) -> str:
            result = self.helper.send_event(
@@ -186,6 +188,7 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
        _assert_counts(0, 0)

        _create_event()
+        _assert_counts(1, 0)
        _rotate()
        _assert_counts(1, 0)

@@ -236,6 +239,168 @@ class EventPushActionsStoreTestCase(HomeserverTestCase):
        _rotate()
        _assert_counts(0, 0)

+    def test_count_aggregation_threads(self) -> None:
+        """
+        This is essentially the same test as test_count_aggregation, but adds
+        events to the main timeline and to a thread.
+        """
+
+        user_id, token, _, other_token, room_id = self._create_users_and_room()
+        thread_id: str
+
+        last_event_id: str
+
+        def _assert_counts(
+            noitf_count: int,
+            highlight_count: int,
+            thread_notif_count: int,
+            thread_highlight_count: int,
+        ) -> None:
+            counts = self.get_success(
+                self.store.db_pool.runInteraction(
+                    "get-unread-counts",
+                    self.store._get_unread_counts_by_receipt_txn,
+                    room_id,
+                    user_id,
+                )
+            )
+            self.assertEqual(
+                counts.main_timeline,
+                NotifCounts(
+                    notify_count=noitf_count,
+                    unread_count=0,
+                    highlight_count=highlight_count,
+                ),
+            )
+            if thread_notif_count or thread_highlight_count:
+                self.assertEqual(
+                    counts.threads,
+                    {
+                        thread_id: NotifCounts(
+                            notify_count=thread_notif_count,
+                            unread_count=0,
+                            highlight_count=thread_highlight_count,
+                        ),
+                    },
+                )
+            else:
+                self.assertEqual(counts.threads, {})
+
+        def _create_event(
+            highlight: bool = False, thread_id: Optional[str] = None
+        ) -> str:
+            content: JsonDict = {
+                "msgtype": "m.text",
+                "body": user_id if highlight else "msg",
+            }
+            if thread_id:
+                content["m.relates_to"] = {
+                    "rel_type": "m.thread",
+                    "event_id": thread_id,
+                }
+
+            result = self.helper.send_event(
+                room_id,
+                type="m.room.message",
+                content=content,
+                tok=other_token,
+            )
+            nonlocal last_event_id
+            last_event_id = result["event_id"]
+            return last_event_id
+
+        def _rotate() -> None:
+            self.get_success(self.store._rotate_notifs())
+
+        def _mark_read(event_id: str, thread_id: Optional[str] = None) -> None:
+            self.get_success(
+                self.store.insert_receipt(
+                    room_id,
+                    "m.read",
+                    user_id=user_id,
+                    event_ids=[event_id],
+                    thread_id=thread_id,
+                    data={},
+                )
+            )
+
+        _assert_counts(0, 0, 0, 0)
+        thread_id = _create_event()
+        _assert_counts(1, 0, 0, 0)
+        _rotate()
+        _assert_counts(1, 0, 0, 0)
+
+        _create_event(thread_id=thread_id)
+        _assert_counts(1, 0, 1, 0)
+        _rotate()
+        _assert_counts(1, 0, 1, 0)
+
+        _create_event()
+        _assert_counts(2, 0, 1, 0)
+        _rotate()
+        _assert_counts(2, 0, 1, 0)
+
+        event_id = _create_event(thread_id=thread_id)
+        _assert_counts(2, 0, 2, 0)
+        _rotate()
+        _assert_counts(2, 0, 2, 0)
+
+        _create_event()
+        _create_event(thread_id=thread_id)
+        _mark_read(event_id)
+        _assert_counts(1, 0, 1, 0)
+
+        _mark_read(last_event_id)
+        _assert_counts(0, 0, 0, 0)
+
+        _create_event()
+        _create_event(thread_id=thread_id)
+        _assert_counts(1, 0, 1, 0)
+        _rotate()
+        _assert_counts(1, 0, 1, 0)
+
+        # Delete old event push actions, this should not affect the (summarised) count.
+        self.get_success(self.store._remove_old_push_actions_that_have_rotated())
+        _assert_counts(1, 0, 1, 0)
+
+        _mark_read(last_event_id)
+        _assert_counts(0, 0, 0, 0)
+
+        _create_event(True)
+        _assert_counts(1, 1, 0, 0)
+        _rotate()
+        _assert_counts(1, 1, 0, 0)
+
+        event_id = _create_event(True, thread_id)
+        _assert_counts(1, 1, 1, 1)
+        _rotate()
+        _assert_counts(1, 1, 1, 1)
+
+        # Check that adding another notification and rotating after highlight
+        # works.
+        _create_event()
+        _rotate()
+        _assert_counts(2, 1, 1, 1)
+
+        _create_event(thread_id=thread_id)
+        _rotate()
+        _assert_counts(2, 1, 2, 1)
+
+        # Check that sending read receipts at different points results in the
+        # right counts.
+        _mark_read(event_id)
+        _assert_counts(1, 0, 1, 0)
+        _mark_read(last_event_id)
+        _assert_counts(0, 0, 0, 0)
+
+        _create_event(True)
+        _create_event(True, thread_id)
+        _assert_counts(1, 1, 1, 1)
+        _mark_read(last_event_id)
+        _assert_counts(0, 0, 0, 0)
+        _rotate()
+        _assert_counts(0, 0, 0, 0)
+
    def test_find_first_stream_ordering_after_ts(self) -> None:
        def add_event(so: int, ts: int) -> None:
            self.get_success(
--- a/tests/test_federation.py
+++ b/tests/test_federation.py
@@ -86,8 +86,8 @@ class MessageAcceptTests(unittest.HomeserverTestCase):

        federation_event_handler._check_event_auth = _check_event_auth
        self.client = self.homeserver.get_federation_client()
-        self.client._check_sigs_and_hash_and_fetch = lambda dest, pdus, **k: succeed(
-            pdus
+        self.client._check_sigs_and_hash_for_pulled_events_and_fetch = (
+            lambda dest, pdus, **k: succeed(pdus)
        )

        # Send the join, it should return None (which is not an error)
				`@@ -0,0 +1 @@`
				`Experimental support for thread-specific notifications ([MSC3773](https://github.com/matrix-org/matrix-spec-proposals/pull/3773)).`
				`@@ -0,0 +1 @@`
				`Keep track when an event pulled over federation fails its signature check so we can intelligently back-off in the future.`
				`@@ -0,0 +1 @@`
				`Bump actions/download-artifact from 2 to 3.`
				`@@ -0,0 +1 @@`
				`Bump types-pyopenssl from 22.0.0 to 22.0.10.`
				`@@ -0,0 +1 @@`
				`Bump types-setuptools from 57.4.9 to 65.4.0.0.`
				`@@ -0,0 +1 @@`
				`Bump types-requests from 2.27.11 to 2.28.11.`
				`@@ -0,0 +1 @@`
				`Send application service access tokens as a header (and query parameter). Implement [MSC2832](https://github.com/matrix-org/matrix-spec-proposals/pull/2832).`
				`@@ -0,0 +1 @@`
				`Ignore server ACL changes when generating pushes. Implement [MSC3786](https://github.com/matrix-org/matrix-spec-proposals/pull/3786).`
				`@@ -0,0 +1 @@`
				Clear out stale entries in `event_push_actions_staging` table.
				`@@ -0,0 +1 @@`
				`Announce that legacy metric names are deprecated, will be turned off by default in Synapse v1.71.0 and removed altogether in Synapse v1.73.0. See the upgrade notes for more information.`