Fix redactions

Handle deduplicating multiple fetch event requests
Newsfile
2021-05-10 15:48:27 +01:00 · 2021-05-10 14:49:39 +01:00 · 2021-05-06 14:45:20 +01:00 · 2021-05-06 14:42:42 +01:00 · 2021-05-06 14:36:59 +01:00 · 2021-05-06 14:06:38 +01:00
113 changed files with 1140 additions and 712 deletions
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -1,3 +1,83 @@
+Synapse 1.33.1 (2021-05-06)
+===========================
+
+Bugfixes
+--------
+
+- Fix bug where `/sync` would break if using the latest version of `attrs` dependency, by pinning to a previous version. ([\#9937](https://github.com/matrix-org/synapse/issues/9937))
+
+
+Synapse 1.33.0 (2021-05-05)
+===========================
+
+Features
+--------
+
+- Build Debian packages for Ubuntu 21.04 (Hirsute Hippo). ([\#9909](https://github.com/matrix-org/synapse/issues/9909))
+
+
+Synapse 1.33.0rc2 (2021-04-29)
+==============================
+
+Bugfixes
+--------
+
+- Fix tight loop when handling presence replication when using workers. Introduced in v1.33.0rc1. ([\#9900](https://github.com/matrix-org/synapse/issues/9900))
+
+
+Synapse 1.33.0rc1 (2021-04-28)
+==============================
+
+Features
+--------
+
+- Update experimental support for [MSC3083](https://github.com/matrix-org/matrix-doc/pull/3083): restricting room access via group membership. ([\#9800](https://github.com/matrix-org/synapse/issues/9800), [\#9814](https://github.com/matrix-org/synapse/issues/9814))
+- Add experimental support for handling presence on a worker. ([\#9819](https://github.com/matrix-org/synapse/issues/9819), [\#9820](https://github.com/matrix-org/synapse/issues/9820), [\#9828](https://github.com/matrix-org/synapse/issues/9828), [\#9850](https://github.com/matrix-org/synapse/issues/9850))
+- Return a new template when an user attempts to renew their account multiple times with the same token, stating that their account is set to expire. This replaces the invalid token template that would previously be shown in this case. This change concerns the optional account validity feature. ([\#9832](https://github.com/matrix-org/synapse/issues/9832))
+
+
+Bugfixes
+--------
+
+- Fixes the OIDC SSO flow when using a `public_baseurl` value including a non-root URL path. ([\#9726](https://github.com/matrix-org/synapse/issues/9726))
+- Fix thumbnail generation for some sites with non-standard content types. Contributed by @rkfg. ([\#9788](https://github.com/matrix-org/synapse/issues/9788))
+- Add some sanity checks to identity server passed to 3PID bind/unbind endpoints. ([\#9802](https://github.com/matrix-org/synapse/issues/9802))
+- Limit the size of HTTP responses read over federation. ([\#9833](https://github.com/matrix-org/synapse/issues/9833))
+- Fix a bug which could cause Synapse to get stuck in a loop of resyncing device lists. ([\#9867](https://github.com/matrix-org/synapse/issues/9867))
+- Fix a long-standing bug where errors from federation did not propagate to the client. ([\#9868](https://github.com/matrix-org/synapse/issues/9868))
+
+
+Improved Documentation
+----------------------
+
+- Add a note to the docker docs mentioning that we mirror upstream's supported Docker platforms. ([\#9801](https://github.com/matrix-org/synapse/issues/9801))
+
+
+Internal Changes
+----------------
+
+- Add a dockerfile for running Synapse in worker-mode under Complement. ([\#9162](https://github.com/matrix-org/synapse/issues/9162))
+- Apply `pyupgrade` across the codebase. ([\#9786](https://github.com/matrix-org/synapse/issues/9786))
+- Move some replication processing out of `generic_worker`. ([\#9796](https://github.com/matrix-org/synapse/issues/9796))
+- Replace `HomeServer.get_config()` with inline references. ([\#9815](https://github.com/matrix-org/synapse/issues/9815))
+- Rename some handlers and config modules to not duplicate the top-level module. ([\#9816](https://github.com/matrix-org/synapse/issues/9816))
+- Fix a long-standing bug which caused `max_upload_size` to not be correctly enforced. ([\#9817](https://github.com/matrix-org/synapse/issues/9817))
+- Reduce CPU usage of the user directory by reusing existing calculated room membership. ([\#9821](https://github.com/matrix-org/synapse/issues/9821))
+- Small speed up for joining large remote rooms. ([\#9825](https://github.com/matrix-org/synapse/issues/9825))
+- Introduce flake8-bugbear to the test suite and fix some of its lint violations. ([\#9838](https://github.com/matrix-org/synapse/issues/9838))
+- Only store the raw data in the in-memory caches, rather than objects that include references to e.g. the data stores. ([\#9845](https://github.com/matrix-org/synapse/issues/9845))
+- Limit length of accepted email addresses. ([\#9855](https://github.com/matrix-org/synapse/issues/9855))
+- Remove redundant `synapse.types.Collection` type definition. ([\#9856](https://github.com/matrix-org/synapse/issues/9856))
+- Handle recently added rate limits correctly when using `--no-rate-limit` with the demo scripts. ([\#9858](https://github.com/matrix-org/synapse/issues/9858))
+- Disable invite rate-limiting by default when running the unit tests. ([\#9871](https://github.com/matrix-org/synapse/issues/9871))
+- Pass a reactor into `SynapseSite` to make testing easier. ([\#9874](https://github.com/matrix-org/synapse/issues/9874))
+- Make `DomainSpecificString` an `attrs` class. ([\#9875](https://github.com/matrix-org/synapse/issues/9875))
+- Add type hints to `synapse.api.auth` and `synapse.api.auth_blocking` modules. ([\#9876](https://github.com/matrix-org/synapse/issues/9876))
+- Remove redundant `_PushHTTPChannel` test class. ([\#9878](https://github.com/matrix-org/synapse/issues/9878))
+- Remove backwards-compatibility code for Python versions < 3.6. ([\#9879](https://github.com/matrix-org/synapse/issues/9879))
+- Small performance improvement around handling new local presence updates. ([\#9887](https://github.com/matrix-org/synapse/issues/9887))
+
+
 Synapse 1.32.2 (2021-04-22)
 ===========================

--- a/changelog.d/9162.misc
+++ b/changelog.d/9162.misc
@@ -1 +0,0 @@
-Add a dockerfile for running Synapse in worker-mode under Complement.
--- a/changelog.d/9702.misc
+++ b/changelog.d/9702.misc
@@ -1 +0,0 @@
-Speed up federation transmission by using fewer database calls. Contributed by @ShadowJonathan.
--- a/changelog.d/9726.bugfix
+++ b/changelog.d/9726.bugfix
@@ -1 +0,0 @@
-Fixes the OIDC SSO flow when using a `public_baseurl` value including a non-root URL path.
--- a/changelog.d/9786.misc
+++ b/changelog.d/9786.misc
@@ -1 +0,0 @@
-Apply `pyupgrade` across the codebase.
--- a/changelog.d/9788.bugfix
+++ b/changelog.d/9788.bugfix
@@ -1 +0,0 @@
-Fix thumbnail generation for some sites with non-standard content types. Contributed by @rkfg.
--- a/changelog.d/9796.misc
+++ b/changelog.d/9796.misc
@@ -1 +0,0 @@
-Move some replication processing out of `generic_worker`.
--- a/changelog.d/9800.feature
+++ b/changelog.d/9800.feature
@@ -1 +0,0 @@
-Update experimental support for [MSC3083](https://github.com/matrix-org/matrix-doc/pull/3083): restricting room access via group membership.
--- a/changelog.d/9801.doc
+++ b/changelog.d/9801.doc
@@ -1 +0,0 @@
-Add a note to the docker docs mentioning that we mirror upstream's supported Docker platforms.
--- a/changelog.d/9802.bugfix
+++ b/changelog.d/9802.bugfix
@@ -1 +0,0 @@
-Add some sanity checks to identity server passed to 3PID bind/unbind endpoints.
--- a/changelog.d/9814.feature
+++ b/changelog.d/9814.feature
@@ -1 +0,0 @@
-Update experimental support for [MSC3083](https://github.com/matrix-org/matrix-doc/pull/3083): restricting room access via group membership.
--- a/changelog.d/9815.misc
+++ b/changelog.d/9815.misc
@@ -1 +0,0 @@
-Replace `HomeServer.get_config()` with inline references.
--- a/changelog.d/9816.misc
+++ b/changelog.d/9816.misc
@@ -1 +0,0 @@
-Rename some handlers and config modules to not duplicate the top-level module.
--- a/changelog.d/9817.misc
+++ b/changelog.d/9817.misc
@@ -1 +0,0 @@
-Fix a long-standing bug which caused `max_upload_size` to not be correctly enforced.
--- a/changelog.d/9819.feature
+++ b/changelog.d/9819.feature
@@ -1 +0,0 @@
-Add experimental support for handling presence on a worker.
--- a/changelog.d/9820.feature
+++ b/changelog.d/9820.feature
@@ -1 +0,0 @@
-Add experimental support for handling presence on a worker.
--- a/changelog.d/9821.misc
+++ b/changelog.d/9821.misc
@@ -1 +0,0 @@
-Reduce CPU usage of the user directory by reusing existing calculated room membership.
--- a/changelog.d/9825.misc
+++ b/changelog.d/9825.misc
@@ -1 +0,0 @@
-Small speed up for joining large remote rooms.
--- a/changelog.d/9828.feature
+++ b/changelog.d/9828.feature
@@ -1 +0,0 @@
-Add experimental support for handling presence on a worker.
--- a/changelog.d/9832.feature
+++ b/changelog.d/9832.feature
@@ -1 +0,0 @@
-Don't return an error when a user attempts to renew their account multiple times with the same token. Instead, state when their account is set to expire. This change concerns the optional account validity feature.
--- a/changelog.d/9833.bugfix
+++ b/changelog.d/9833.bugfix
@@ -1 +0,0 @@
-Limit the size of HTTP responses read over federation.
--- a/changelog.d/9838.misc
+++ b/changelog.d/9838.misc
@@ -1 +0,0 @@
-Introduce flake8-bugbear to the test suite and fix some of its lint violations.
--- a/changelog.d/9845.misc
+++ b/changelog.d/9845.misc
@@ -1 +0,0 @@
-Only store the raw data in the in-memory caches, rather than objects that include references to e.g. the data stores.
--- a/changelog.d/9850.feature
+++ b/changelog.d/9850.feature
@@ -1 +0,0 @@
-Add experimental support for handling presence on a worker.
--- a/changelog.d/9855.misc
+++ b/changelog.d/9855.misc
@@ -1 +0,0 @@
-Limit length of accepted email addresses.
--- a/changelog.d/9856.misc
+++ b/changelog.d/9856.misc
@@ -1 +0,0 @@
-Remove redundant `synapse.types.Collection` type definition.
--- a/changelog.d/9858.misc
+++ b/changelog.d/9858.misc
@@ -1 +0,0 @@
-Handle recently added rate limits correctly when using `--no-rate-limit` with the demo scripts.
--- a/changelog.d/9867.bugfix
+++ b/changelog.d/9867.bugfix
@@ -1 +0,0 @@
-Fix a bug which could cause Synapse to get stuck in a loop of resyncing device lists.
--- a/changelog.d/9871.misc
+++ b/changelog.d/9871.misc
@@ -1 +0,0 @@
-Disable invite rate-limiting by default when running the unit tests.
--- a/changelog.d/9874.misc
+++ b/changelog.d/9874.misc
@@ -1 +0,0 @@
-Pass a reactor into `SynapseSite` to make testing easier.
--- a/changelog.d/9875.misc
+++ b/changelog.d/9875.misc
@@ -1 +0,0 @@
-Make `DomainSpecificString` an `attrs` class.
--- a/changelog.d/9876.misc
+++ b/changelog.d/9876.misc
@@ -1 +0,0 @@
-Add type hints to `synapse.api.auth` and `synapse.api.auth_blocking` modules.
--- a/changelog.d/9878.misc
+++ b/changelog.d/9878.misc
@@ -1 +0,0 @@
-Remove redundant `_PushHTTPChannel` test class.
--- a/changelog.d/9881.feature
+++ b/changelog.d/9881.feature
@@ -0,0 +1 @@
+Add experimental option to track memory usage of the caches.
--- a/changelog.d/9885.misc
+++ b/changelog.d/9885.misc
@@ -0,0 +1 @@
+Add type hints to presence handler.
--- a/changelog.d/9886.misc
+++ b/changelog.d/9886.misc
@@ -0,0 +1 @@
+Reduce memory usage of the LRU caches.
--- a/changelog.d/9887.misc
+++ b/changelog.d/9887.misc
@@ -1 +0,0 @@
-Small performance improvement around handling new local presence updates.
--- a/changelog.d/9889.feature
+++ b/changelog.d/9889.feature
@@ -0,0 +1 @@
+Add support for `DELETE /_synapse/admin/v1/rooms/<room_id>`.
--- a/changelog.d/9889.removal
+++ b/changelog.d/9889.removal
@@ -0,0 +1 @@
+Mark as deprecated `POST /_synapse/admin/v1/rooms/<room_id>/delete`.
--- a/changelog.d/9895.bugfix
+++ b/changelog.d/9895.bugfix
@@ -0,0 +1 @@
+Fix a bug introduced in v1.32.0 where the associated connection was improperly logged for SQL logging statements.
--- a/changelog.d/9896.bugfix
+++ b/changelog.d/9896.bugfix
@@ -0,0 +1 @@
+Correct the type hint for the `user_may_create_room_alias` method of spam checkers. It is provided a `RoomAlias`, not a `str`.
--- a/changelog.d/9896.misc
+++ b/changelog.d/9896.misc
@@ -0,0 +1 @@
+Add type hints to the `synapse.handlers` module.
--- a/changelog.d/9902.feature
+++ b/changelog.d/9902.feature
@@ -0,0 +1 @@
+Add limits to how often Synapse will GC, ensuring that large servers do not end up GC thrashing if `gc_thresholds` has not been correctly set.
--- a/changelog.d/9904.misc
+++ b/changelog.d/9904.misc
@@ -0,0 +1 @@
+Time response time for external cache requests.
--- a/changelog.d/9905.feature
+++ b/changelog.d/9905.feature
@@ -0,0 +1 @@
+Improve performance of sending events for worker-based deployments using Redis.
--- a/changelog.d/9910.bugfix
+++ b/changelog.d/9910.bugfix
@@ -0,0 +1 @@
+Fix bug where user directory could get out of sync if room visibility and membership changed in quick succession.
--- a/changelog.d/9910.feature
+++ b/changelog.d/9910.feature
@@ -0,0 +1 @@
+Improve performance after joining a large room when presence is enabled.
--- a/changelog.d/9911.doc
+++ b/changelog.d/9911.doc
@@ -0,0 +1 @@
+Add `port` argument to the Postgres database sample config section.
--- a/changelog.d/9913.docker
+++ b/changelog.d/9913.docker
@@ -0,0 +1 @@
+Added startup_delay to docker healthcheck to reduce waiting time for coming online, updated readme for extra options, contributed by @Maquis196.
--- a/changelog.d/9915.feature
+++ b/changelog.d/9915.feature
@@ -0,0 +1 @@
+Support stable identifiers from [MSC1772](https://github.com/matrix-org/matrix-doc/pull/1772).
--- a/changelog.d/9916.feature
+++ b/changelog.d/9916.feature
@@ -0,0 +1 @@
+Improve performance after joining a large room when presence is enabled.
--- a/changelog.d/9919.feature
+++ b/changelog.d/9919.feature
@@ -0,0 +1 @@
+Omit empty fields from the `/sync` response.  Contributed by @deepbluev7.
--- a/changelog.d/9928.bugfix
+++ b/changelog.d/9928.bugfix
@@ -0,0 +1 @@
+Include the `origin_server_ts` property in the experimental [MSC2946](https://github.com/matrix-org/matrix-doc/pull/2946) support to allow clients to properly sort rooms.
--- a/changelog.d/9939.misc
+++ b/changelog.d/9939.misc
@@ -0,0 +1 @@
+Ensure we only have one copy of an event in memory at a time.
--- a/contrib/experiments/test_messaging.py
+++ b/contrib/experiments/test_messaging.py
@@ -224,16 +224,14 @@ class HomeServer(ReplicationHandler):
        destinations = yield self.get_servers_for_context(room_name)

        try:
-            yield self.replication_layer.send_pdus(
-                [
-                    Pdu.create_new(
-                        context=room_name,
-                        pdu_type="sy.room.message",
-                        content={"sender": sender, "body": body},
-                        origin=self.server_name,
-                        destinations=destinations,
-                    )
-                ]
+            yield self.replication_layer.send_pdu(
+                Pdu.create_new(
+                    context=room_name,
+                    pdu_type="sy.room.message",
+                    content={"sender": sender, "body": body},
+                    origin=self.server_name,
+                    destinations=destinations,
+                )
            )
        except Exception as e:
            logger.exception(e)
@@ -255,7 +253,7 @@ class HomeServer(ReplicationHandler):
                origin=self.server_name,
                destinations=destinations,
            )
-            yield self.replication_layer.send_pdus([pdu])
+            yield self.replication_layer.send_pdu(pdu)
        except Exception as e:
            logger.exception(e)

@@ -267,18 +265,16 @@ class HomeServer(ReplicationHandler):
        destinations = yield self.get_servers_for_context(room_name)

        try:
-            yield self.replication_layer.send_pdus(
-                [
-                    Pdu.create_new(
-                        context=room_name,
-                        is_state=True,
-                        pdu_type="sy.room.member",
-                        state_key=invitee,
-                        content={"membership": "invite"},
-                        origin=self.server_name,
-                        destinations=destinations,
-                    )
-                ]
+            yield self.replication_layer.send_pdu(
+                Pdu.create_new(
+                    context=room_name,
+                    is_state=True,
+                    pdu_type="sy.room.member",
+                    state_key=invitee,
+                    content={"membership": "invite"},
+                    origin=self.server_name,
+                    destinations=destinations,
+                )
            )
        except Exception as e:
            logger.exception(e)
--- a/debian/changelog
+++ b/debian/changelog
@@ -1,3 +1,15 @@
+matrix-synapse-py3 (1.33.1) stable; urgency=medium
+
+  * New synapse release 1.33.1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 06 May 2021 14:06:33 +0100
+
+matrix-synapse-py3 (1.33.0) stable; urgency=medium
+
+  * New synapse release 1.33.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Wed, 05 May 2021 14:15:27 +0100
+
 matrix-synapse-py3 (1.32.2) stable; urgency=medium

  * New synapse release 1.32.2.
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -88,5 +88,5 @@ EXPOSE 8008/tcp 8009/tcp 8448/tcp

 ENTRYPOINT ["/start.py"]

-HEALTHCHECK --interval=1m --timeout=5s \
+HEALTHCHECK --start-period=5s --interval=15s --timeout=5s \
    CMD curl -fSs http://localhost:8008/health || exit 1
--- a/docker/README.md
+++ b/docker/README.md
@@ -191,6 +191,16 @@ whilst running the above `docker run` commands.
 ```
   --no-healthcheck
 ```
+
+## Disabling the healthcheck in docker-compose file
+
+If you wish to disable the healthcheck via docker-compose, append the following to your service configuration.
+
+```
+  healthcheck:
+    disable: true
+```
+
 ## Setting custom healthcheck on docker run

 If you wish to point the healthcheck at a different port with docker command, add the following
@@ -202,14 +212,15 @@ If you wish to point the healthcheck at a different port with docker command, ad
 ## Setting the healthcheck in docker-compose file

 You can add the following to set a custom healthcheck in a docker compose file.
-You will need version >2.1 for this to work. 
+You will need docker-compose version >2.1 for this to work. 

 ```
 healthcheck:
  test: ["CMD", "curl", "-fSs", "http://localhost:8008/health"]
-  interval: 1m
-  timeout: 10s
+  interval: 15s
+  timeout: 5s
  retries: 3
+  start_period: 5s
 ```

 ## Using jemalloc
--- a/docs/admin_api/rooms.md
+++ b/docs/admin_api/rooms.md
@@ -427,7 +427,7 @@ the new room. Users on other servers will be unaffected.
 The API is:

 ```
-POST /_synapse/admin/v1/rooms/<room_id>/delete
+DELETE /_synapse/admin/v1/rooms/<room_id>
 ```

 with a body of:
@@ -528,6 +528,15 @@ You will have to manually handle, if you so choose, the following:
 * Users that would have been booted from the room (and will have been force-joined to the Content Violation room).
 * Removal of the Content Violation room if desired.

+## Deprecated endpoint
+
+The previous deprecated API will be removed in a future release, it was:
+
+```
+POST /_synapse/admin/v1/rooms/<room_id>/delete
+```
+
+It behaves the same way than the current endpoint except the path and the method.

 # Make Room Admin API

--- a/docs/sample_config.yaml
+++ b/docs/sample_config.yaml
@@ -152,6 +152,16 @@ presence:
 #
 #gc_thresholds: [700, 10, 10]

+# The minimum time in seconds between each GC for a generation, regardless of
+# the GC thresholds. This ensures that we don't do GC too frequently.
+#
+# A value of `[1s, 10s, 30s]` indicates that a second must pass between consecutive
+# generation 0 GCs, etc.
+#
+# Defaults to `[1s, 10s, 30s]`.
+#
+#gc_min_interval: [0.5s, 30s, 1m]
+
 # Set the limit on the returned events in the timeline in the get
 # and sync operations. The default value is 100. -1 means no upper limit.
 #
@@ -810,6 +820,7 @@ caches:
 #    password: secretpassword
 #    database: synapse
 #    host: localhost
+#    port: 5432
 #    cp_min: 5
 #    cp_max: 10
 #
--- a/mypy.ini
+++ b/mypy.ini
@@ -41,7 +41,6 @@ files =
  synapse/push,
  synapse/replication,
  synapse/rest,
-  synapse/secrets.py,
  synapse/server.py,
  synapse/server_notices,
  synapse/spam_checker_api,
@@ -172,3 +171,6 @@ ignore_missing_imports = True

 [mypy-txacme.*]
 ignore_missing_imports = True
+
+[mypy-pympler.*]
+ignore_missing_imports = True
--- a/scripts-dev/build_debian_packages
+++ b/scripts-dev/build_debian_packages
@@ -21,9 +21,10 @@ DISTS = (
    "debian:buster",
    "debian:bullseye",
    "debian:sid",
-    "ubuntu:bionic",
-    "ubuntu:focal",
-    "ubuntu:groovy",
+    "ubuntu:bionic",   # 18.04 LTS (our EOL forced by Py36 on 2021-12-23)
+    "ubuntu:focal",    # 20.04 LTS (our EOL forced by Py38 on 2024-10-14)
+    "ubuntu:groovy",   # 20.10 (EOL 2021-07-07)
+    "ubuntu:hirsute",  # 21.04 (EOL 2022-01-05)
 )

 DESC = '''\
--- a/synapse/init.py
+++ b/synapse/init.py
@@ -21,8 +21,8 @@ import os
 import sys

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

 # Twisted and canonicaljson will fail to import when this file is executed to
@@ -47,7 +47,7 @@ try:
 except ImportError:
    pass

-__version__ = "1.32.2"
+__version__ = "1.33.1"

 if bool(os.environ.get("SYNAPSE_TEST_PATCH_LOG_CONTEXTS", False)):
    # We import here so that we don't have to install a bunch of deps when
--- a/synapse/api/constants.py
+++ b/synapse/api/constants.py
@@ -110,6 +110,8 @@ class EventTypes:

    Dummy = "org.matrix.dummy_event"

+    SpaceChild = "m.space.child"
+    SpaceParent = "m.space.parent"
    MSC1772_SPACE_CHILD = "org.matrix.msc1772.space.child"
    MSC1772_SPACE_PARENT = "org.matrix.msc1772.space.parent"

@@ -174,6 +176,7 @@ class EventContentFields:
    SELF_DESTRUCT_AFTER = "org.matrix.self_destruct_after"

    # cf https://github.com/matrix-org/matrix-doc/pull/1772
+    ROOM_TYPE = "type"
    MSC1772_ROOM_TYPE = "org.matrix.msc1772.type"


--- a/synapse/app/generic_worker.py
+++ b/synapse/app/generic_worker.py
@@ -454,6 +454,10 @@ def start(config_options):
        config.server.update_user_directory = False

    synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts
+    synapse.util.caches.TRACK_MEMORY_USAGE = config.caches.track_memory_usage
+
+    if config.server.gc_seconds:
+        synapse.metrics.MIN_TIME_BETWEEN_GCS = config.server.gc_seconds

    hs = GenericWorkerServer(
        config.server_name,
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -341,6 +341,10 @@ def setup(config_options):
        sys.exit(0)

    events.USE_FROZEN_DICTS = config.use_frozen_dicts
+    synapse.util.caches.TRACK_MEMORY_USAGE = config.caches.track_memory_usage
+
+    if config.server.gc_seconds:
+        synapse.metrics.MIN_TIME_BETWEEN_GCS = config.server.gc_seconds

    hs = SynapseHomeServer(
        config.server_name,
--- a/synapse/config/cache.py
+++ b/synapse/config/cache.py
@@ -17,6 +17,8 @@ import re
 import threading
 from typing import Callable, Dict

+from synapse.python_dependencies import DependencyException, check_requirements
+
 from ._base import Config, ConfigError

 # The prefix for all cache factor-related environment variables
@@ -189,6 +191,15 @@ class CacheConfig(Config):
                )
            self.cache_factors[cache] = factor

+        self.track_memory_usage = cache_config.get("track_memory_usage", False)
+        if self.track_memory_usage:
+            try:
+                check_requirements("cache_memory")
+            except DependencyException as e:
+                raise ConfigError(
+                    e.message  # noqa: B306, DependencyException.message is a property
+                )
+
        # Resize all caches (if necessary) with the new factors we've loaded
        self.resize_all_caches()

--- a/synapse/config/database.py
+++ b/synapse/config/database.py
@@ -58,6 +58,7 @@ DEFAULT_CONFIG = """\
 #    password: secretpassword
 #    database: synapse
 #    host: localhost
+#    port: 5432
 #    cp_min: 5
 #    cp_max: 10
 #
--- a/synapse/config/server.py
+++ b/synapse/config/server.py
@@ -19,7 +19,7 @@ import logging
 import os.path
 import re
 from textwrap import indent
-from typing import Any, Dict, Iterable, List, Optional, Set
+from typing import Any, Dict, Iterable, List, Optional, Set, Tuple

 import attr
 import yaml
@@ -572,6 +572,7 @@ class ServerConfig(Config):
            _warn_if_webclient_configured(self.listeners)

        self.gc_thresholds = read_gc_thresholds(config.get("gc_thresholds", None))
+        self.gc_seconds = self.read_gc_intervals(config.get("gc_min_interval", None))

        @attr.s
        class LimitRemoteRoomsConfig:
@@ -917,6 +918,16 @@ class ServerConfig(Config):
        #
        #gc_thresholds: [700, 10, 10]

+        # The minimum time in seconds between each GC for a generation, regardless of
+        # the GC thresholds. This ensures that we don't do GC too frequently.
+        #
+        # A value of `[1s, 10s, 30s]` indicates that a second must pass between consecutive
+        # generation 0 GCs, etc.
+        #
+        # Defaults to `[1s, 10s, 30s]`.
+        #
+        #gc_min_interval: [0.5s, 30s, 1m]
+
        # Set the limit on the returned events in the timeline in the get
        # and sync operations. The default value is 100. -1 means no upper limit.
        #
@@ -1305,6 +1316,24 @@ class ServerConfig(Config):
            help="Turn on the twisted telnet manhole service on the given port.",
        )

+    def read_gc_intervals(self, durations) -> Optional[Tuple[float, float, float]]:
+        """Reads the three durations for the GC min interval option, returning seconds."""
+        if durations is None:
+            return None
+
+        try:
+            if len(durations) != 3:
+                raise ValueError()
+            return (
+                self.parse_duration(durations[0]) / 1000,
+                self.parse_duration(durations[1]) / 1000,
+                self.parse_duration(durations[2]) / 1000,
+            )
+        except Exception:
+            raise ConfigError(
+                "Value of `gc_min_interval` must be a list of three durations if set"
+            )
+

 def is_threepid_reserved(reserved_threepids, threepid):
    """Check the threepid against the reserved threepid config
--- a/synapse/events/spamcheck.py
+++ b/synapse/events/spamcheck.py
@@ -20,6 +20,7 @@ from typing import TYPE_CHECKING, Any, Collection, Dict, List, Optional, Tuple,
 from synapse.rest.media.v1._base import FileInfo
 from synapse.rest.media.v1.media_storage import ReadableFileWrapper
 from synapse.spam_checker_api import RegistrationBehaviour
+from synapse.types import RoomAlias
 from synapse.util.async_helpers import maybe_awaitable

 if TYPE_CHECKING:
@@ -113,7 +114,9 @@ class SpamChecker:

        return True

-    async def user_may_create_room_alias(self, userid: str, room_alias: str) -> bool:
+    async def user_may_create_room_alias(
+        self, userid: str, room_alias: RoomAlias
+    ) -> bool:
        """Checks if a given user may create a room alias

        If this method returns false, the association request will be rejected.
--- a/synapse/federation/federation_client.py
+++ b/synapse/federation/federation_client.py
@@ -451,6 +451,28 @@ class FederationClient(FederationBase):

        return signed_auth

+    def _is_unknown_endpoint(
+        self, e: HttpResponseException, synapse_error: Optional[SynapseError] = None
+    ) -> bool:
+        """
+        Returns true if the response was due to an endpoint being unimplemented.
+
+        Args:
+            e: The error response received from the remote server.
+            synapse_error: The above error converted to a SynapseError. This is
+                automatically generated if not provided.
+
+        """
+        if synapse_error is None:
+            synapse_error = e.to_synapse_error()
+        # There is no good way to detect an "unknown" endpoint.
+        #
+        # Dendrite returns a 404 (with no body); synapse returns a 400
+        # with M_UNRECOGNISED.
+        return e.code == 404 or (
+            e.code == 400 and synapse_error.errcode == Codes.UNRECOGNIZED
+        )
+
    async def _try_destination_list(
        self,
        description: str,
@@ -468,9 +490,9 @@ class FederationClient(FederationBase):
            callback:  Function to run for each server. Passed a single
                argument: the server_name to try.

-                If the callback raises a CodeMessageException with a 300/400 code,
-                attempts to perform the operation stop immediately and the exception is
-                reraised.
+                If the callback raises a CodeMessageException with a 300/400 code or
+                an UnsupportedRoomVersionError, attempts to perform the operation
+                stop immediately and the exception is reraised.

                Otherwise, if the callback raises an Exception the error is logged and the
                next server tried. Normally the stacktrace is logged but this is
@@ -492,8 +514,7 @@ class FederationClient(FederationBase):
                continue

            try:
-                res = await callback(destination)
-                return res
+                return await callback(destination)
            except InvalidResponseError as e:
                logger.warning("Failed to %s via %s: %s", description, destination, e)
            except UnsupportedRoomVersionError:
@@ -502,17 +523,15 @@ class FederationClient(FederationBase):
                synapse_error = e.to_synapse_error()
                failover = False

+                # Failover on an internal server error, or if the destination
+                # doesn't implemented the endpoint for some reason.
                if 500 <= e.code < 600:
                    failover = True

-                elif failover_on_unknown_endpoint:
-                    # there is no good way to detect an "unknown" endpoint. Dendrite
-                    # returns a 404 (with no body); synapse returns a 400
-                    # with M_UNRECOGNISED.
-                    if e.code == 404 or (
-                        e.code == 400 and synapse_error.errcode == Codes.UNRECOGNIZED
-                    ):
-                        failover = True
+                elif failover_on_unknown_endpoint and self._is_unknown_endpoint(
+                    e, synapse_error
+                ):
+                    failover = True

                if not failover:
                    raise synapse_error from e
@@ -570,9 +589,8 @@ class FederationClient(FederationBase):
            UnsupportedRoomVersionError: if remote responds with
                a room version we don't understand.

-            SynapseError: if the chosen remote server returns a 300/400 code.
-
-            RuntimeError: if no servers were reachable.
+            SynapseError: if the chosen remote server returns a 300/400 code, or
+                no servers successfully handle the request.
        """
        valid_memberships = {Membership.JOIN, Membership.LEAVE}
        if membership not in valid_memberships:
@@ -642,9 +660,8 @@ class FederationClient(FederationBase):
            ``auth_chain``.

        Raises:
-            SynapseError: if the chosen remote server returns a 300/400 code.
-
-            RuntimeError: if no servers were reachable.
+            SynapseError: if the chosen remote server returns a 300/400 code, or
+                no servers successfully handle the request.
        """

        async def send_request(destination) -> Dict[str, Any]:
@@ -673,7 +690,7 @@ class FederationClient(FederationBase):
            if create_event is None:
                # If the state doesn't have a create event then the room is
                # invalid, and it would fail auth checks anyway.
-                raise SynapseError(400, "No create event in state")
+                raise InvalidResponseError("No create event in state")

            # the room version should be sane.
            create_room_version = create_event.content.get(
@@ -746,16 +763,11 @@ class FederationClient(FederationBase):
                content=pdu.get_pdu_json(time_now),
            )
        except HttpResponseException as e:
-            if e.code in [400, 404]:
-                err = e.to_synapse_error()
-
-                # If we receive an error response that isn't a generic error, or an
-                # unrecognised endpoint error, we  assume that the remote understands
-                # the v2 invite API and this is a legitimate error.
-                if err.errcode not in [Codes.UNKNOWN, Codes.UNRECOGNIZED]:
-                    raise err
-            else:
-                raise e.to_synapse_error()
+            # If an error is received that is due to an unrecognised endpoint,
+            # fallback to the v1 endpoint. Otherwise consider it a legitmate error
+            # and raise.
+            if not self._is_unknown_endpoint(e):
+                raise

        logger.debug("Couldn't send_join with the v2 API, falling back to the v1 API")

@@ -802,6 +814,11 @@ class FederationClient(FederationBase):

        Returns:
            The event as a dict as returned by the remote server
+
+        Raises:
+            SynapseError: if the remote server returns an error or if the server
+                only supports the v1 endpoint and a room version other than "1"
+                or "2" is requested.
        """
        time_now = self._clock.time_msec()

@@ -817,28 +834,19 @@ class FederationClient(FederationBase):
                },
            )
        except HttpResponseException as e:
-            if e.code in [400, 404]:
-                err = e.to_synapse_error()
-
-                # If we receive an error response that isn't a generic error, we
-                # assume that the remote understands the v2 invite API and this
-                # is a legitimate error.
-                if err.errcode != Codes.UNKNOWN:
-                    raise err
-
-                # Otherwise, we assume that the remote server doesn't understand
-                # the v2 invite API. That's ok provided the room uses old-style event
-                # IDs.
+            # If an error is received that is due to an unrecognised endpoint,
+            # fallback to the v1 endpoint if the room uses old-style event IDs.
+            # Otherwise consider it a legitmate error and raise.
+            err = e.to_synapse_error()
+            if self._is_unknown_endpoint(e, err):
                if room_version.event_format != EventFormatVersions.V1:
                    raise SynapseError(
                        400,
                        "User's homeserver does not support this room version",
                        Codes.UNSUPPORTED_ROOM_VERSION,
                    )
-            elif e.code in (403, 429):
-                raise e.to_synapse_error()
            else:
-                raise
+                raise err

        # Didn't work, try v1 API.
        # Note the v1 API returns a tuple of `(200, content)`
@@ -865,9 +873,8 @@ class FederationClient(FederationBase):
            pdu: event to be sent

        Raises:
-            SynapseError if the chosen remote server returns a 300/400 code.
-
-            RuntimeError if no servers were reachable.
+            SynapseError: if the chosen remote server returns a 300/400 code, or
+                no servers successfully handle the request.
        """

        async def send_request(destination: str) -> None:
@@ -889,16 +896,11 @@ class FederationClient(FederationBase):
                content=pdu.get_pdu_json(time_now),
            )
        except HttpResponseException as e:
-            if e.code in [400, 404]:
-                err = e.to_synapse_error()
-
-                # If we receive an error response that isn't a generic error, or an
-                # unrecognised endpoint error, we  assume that the remote understands
-                # the v2 invite API and this is a legitimate error.
-                if err.errcode not in [Codes.UNKNOWN, Codes.UNRECOGNIZED]:
-                    raise err
-            else:
-                raise e.to_synapse_error()
+            # If an error is received that is due to an unrecognised endpoint,
+            # fallback to the v1 endpoint. Otherwise consider it a legitmate error
+            # and raise.
+            if not self._is_unknown_endpoint(e):
+                raise

        logger.debug("Couldn't send_leave with the v2 API, falling back to the v1 API")

--- a/synapse/federation/sender/init.py
+++ b/synapse/federation/sender/init.py
@@ -14,26 +14,19 @@

 import abc
 import logging
-from typing import (
-    TYPE_CHECKING,
-    Collection,
-    Dict,
-    Hashable,
-    Iterable,
-    List,
-    Optional,
-    Set,
-    Tuple,
-)
+from typing import TYPE_CHECKING, Dict, Hashable, Iterable, List, Optional, Set, Tuple

 from prometheus_client import Counter

+from twisted.internet import defer
+
 import synapse.metrics
 from synapse.api.presence import UserPresenceState
 from synapse.events import EventBase
 from synapse.federation.sender.per_destination_queue import PerDestinationQueue
 from synapse.federation.sender.transaction_manager import TransactionManager
 from synapse.federation.units import Edu
+from synapse.logging.context import make_deferred_yieldable, run_in_background
 from synapse.metrics import (
    LaterGauge,
    event_processing_loop_counter,
@@ -262,27 +255,15 @@ class FederationSender(AbstractFederationSender):
                if not events and next_token >= self._last_poked_id:
                    break

-                async def get_destinations_for_event(
-                    event: EventBase,
-                ) -> Collection[str]:
-                    """Computes the destinations to which this event must be sent.
-
-                    This returns an empty tuple when there are no destinations to send to,
-                    or if this event is not from this homeserver and it is not sending
-                    it on behalf of another server.
-
-                    Will also filter out destinations which this sender is not responsible for,
-                    if multiple federation senders exist.
-                    """
-
+                async def handle_event(event: EventBase) -> None:
                    # Only send events for this server.
                    send_on_behalf_of = event.internal_metadata.get_send_on_behalf_of()
                    is_mine = self.is_mine_id(event.sender)
                    if not is_mine and send_on_behalf_of is None:
-                        return ()
+                        return

                    if not event.internal_metadata.should_proactively_send():
-                        return ()
+                        return

                    destinations = None  # type: Optional[Set[str]]
                    if not event.prev_event_ids():
@@ -317,7 +298,7 @@ class FederationSender(AbstractFederationSender):
                                "Failed to calculate hosts in room for event: %s",
                                event.event_id,
                            )
-                            return ()
+                            return

                    destinations = {
                        d
@@ -327,15 +308,17 @@ class FederationSender(AbstractFederationSender):
                        )
                    }

-                    destinations.discard(self.server_name)
-
                    if send_on_behalf_of is not None:
                        # If we are sending the event on behalf of another server
                        # then it already has the event and there is no reason to
                        # send the event to it.
                        destinations.discard(send_on_behalf_of)

+                    logger.debug("Sending %s to %r", event, destinations)
+
                    if destinations:
+                        await self._send_pdu(event, destinations)
+
                        now = self.clock.time_msec()
                        ts = await self.store.get_received_ts(event.event_id)

@@ -343,29 +326,24 @@ class FederationSender(AbstractFederationSender):
                            "federation_sender"
                        ).observe((now - ts) / 1000)

-                        return destinations
-                    return ()
+                async def handle_room_events(events: Iterable[EventBase]) -> None:
+                    with Measure(self.clock, "handle_room_events"):
+                        for event in events:
+                            await handle_event(event)

-                async def get_federatable_events_and_destinations(
-                    events: Iterable[EventBase],
-                ) -> List[Tuple[EventBase, Collection[str]]]:
-                    with Measure(self.clock, "get_destinations_for_events"):
-                        # Fetch federation destinations per event,
-                        # skip if get_destinations_for_event returns an empty collection,
-                        # return list of event->destinations pairs.
-                        return [
-                            (event, dests)
-                            for (event, dests) in [
-                                (event, await get_destinations_for_event(event))
-                                for event in events
-                            ]
-                            if dests
-                        ]
+                events_by_room = {}  # type: Dict[str, List[EventBase]]
+                for event in events:
+                    events_by_room.setdefault(event.room_id, []).append(event)

-                events_and_dests = await get_federatable_events_and_destinations(events)
-
-                # Send corresponding events to each destination queue
-                await self._distribute_events(events_and_dests)
+                await make_deferred_yieldable(
+                    defer.gatherResults(
+                        [
+                            run_in_background(handle_room_events, evs)
+                            for evs in events_by_room.values()
+                        ],
+                        consumeErrors=True,
+                    )
+                )

                await self.store.update_federation_out_pos("events", next_token)

@@ -383,7 +361,7 @@ class FederationSender(AbstractFederationSender):
                    events_processed_counter.inc(len(events))

                    event_processing_loop_room_count.labels("federation_sender").inc(
-                        len({event.room_id for event in events})
+                        len(events_by_room)
                    )

                event_processing_loop_counter.labels("federation_sender").inc()
@@ -395,53 +373,34 @@ class FederationSender(AbstractFederationSender):
        finally:
            self._is_processing = False

-    async def _distribute_events(
-        self,
-        events_and_dests: Iterable[Tuple[EventBase, Collection[str]]],
-    ) -> None:
-        """Distribute events to the respective per_destination queues.
+    async def _send_pdu(self, pdu: EventBase, destinations: Iterable[str]) -> None:
+        # We loop through all destinations to see whether we already have
+        # a transaction in progress. If we do, stick it in the pending_pdus
+        # table and we'll get back to it later.

-        Also persists last-seen per-room stream_ordering to 'destination_rooms'.
+        destinations = set(destinations)
+        destinations.discard(self.server_name)
+        logger.debug("Sending to: %s", str(destinations))

-        Args:
-            events_and_dests: A list of tuples, which are (event: EventBase, destinations: Collection[str]).
-                              Every event is paired with its intended destinations (in federation).
-        """
-        # Tuples of room_id + destination to their max-seen stream_ordering
-        room_with_dest_stream_ordering = {}  # type: Dict[Tuple[str, str], int]
+        if not destinations:
+            return

-        # List of events to send to each destination
-        events_by_dest = {}  # type: Dict[str, List[EventBase]]
+        sent_pdus_destination_dist_total.inc(len(destinations))
+        sent_pdus_destination_dist_count.inc()

-        # For each event-destinations pair...
-        for event, destinations in events_and_dests:
+        assert pdu.internal_metadata.stream_ordering

-            # (we got this from the database, it's filled)
-            assert event.internal_metadata.stream_ordering
-
-            sent_pdus_destination_dist_total.inc(len(destinations))
-            sent_pdus_destination_dist_count.inc()
-
-            # ...iterate over those destinations..
-            for destination in destinations:
-                # ...update their stream-ordering...
-                room_with_dest_stream_ordering[(event.room_id, destination)] = max(
-                    event.internal_metadata.stream_ordering,
-                    room_with_dest_stream_ordering.get((event.room_id, destination), 0),
-                )
-
-                # ...and add the event to each destination queue.
-                events_by_dest.setdefault(destination, []).append(event)
-
-        # Bulk-store destination_rooms stream_ids
-        await self.store.bulk_store_destination_rooms_entries(
-            room_with_dest_stream_ordering
+        # track the fact that we have a PDU for these destinations,
+        # to allow us to perform catch-up later on if the remote is unreachable
+        # for a while.
+        await self.store.store_destination_rooms_entries(
+            destinations,
+            pdu.room_id,
+            pdu.internal_metadata.stream_ordering,
        )

-        for destination, pdus in events_by_dest.items():
-            logger.debug("Sending %d pdus to %s", len(pdus), destination)
-
-            self._get_per_destination_queue(destination).send_pdus(pdus)
+        for destination in destinations:
+            self._get_per_destination_queue(destination).send_pdu(pdu)

    async def send_read_receipt(self, receipt: ReadReceipt) -> None:
        """Send a RR to any other servers in the room
--- a/synapse/federation/sender/per_destination_queue.py
+++ b/synapse/federation/sender/per_destination_queue.py
@@ -154,22 +154,19 @@ class PerDestinationQueue:
            + len(self._pending_edus_keyed)
        )

-    def send_pdus(self, pdus: Iterable[EventBase]) -> None:
-        """Add PDUs to the queue, and start the transmission loop if necessary
+    def send_pdu(self, pdu: EventBase) -> None:
+        """Add a PDU to the queue, and start the transmission loop if necessary

        Args:
-            pdus: pdus to send
+            pdu: pdu to send
        """
        if not self._catching_up or self._last_successful_stream_ordering is None:
            # only enqueue the PDU if we are not catching up (False) or do not
            # yet know if we have anything to catch up (None)
-            self._pending_pdus.extend(pdus)
+            self._pending_pdus.append(pdu)
        else:
-            self._catchup_last_skipped = max(
-                pdu.internal_metadata.stream_ordering
-                for pdu in pdus
-                if pdu.internal_metadata.stream_ordering is not None
-            )
+            assert pdu.internal_metadata.stream_ordering
+            self._catchup_last_skipped = pdu.internal_metadata.stream_ordering

        self.attempt_new_transaction()

--- a/synapse/handlers/directory.py
+++ b/synapse/handlers/directory.py
@@ -14,7 +14,7 @@

 import logging
 import string
-from typing import Iterable, List, Optional
+from typing import TYPE_CHECKING, Iterable, List, Optional

 from synapse.api.constants import MAX_ALIAS_LENGTH, EventTypes
 from synapse.api.errors import (
@@ -27,15 +27,19 @@ from synapse.api.errors import (
    SynapseError,
 )
 from synapse.appservice import ApplicationService
-from synapse.types import Requester, RoomAlias, UserID, get_domain_from_id
+from synapse.storage.databases.main.directory import RoomAliasMapping
+from synapse.types import JsonDict, Requester, RoomAlias, UserID, get_domain_from_id

 from ._base import BaseHandler

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
 logger = logging.getLogger(__name__)


 class DirectoryHandler(BaseHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super().__init__(hs)

        self.state = hs.get_state_handler()
@@ -60,7 +64,7 @@ class DirectoryHandler(BaseHandler):
        room_id: str,
        servers: Optional[Iterable[str]] = None,
        creator: Optional[str] = None,
-    ):
+    ) -> None:
        # general association creation for both human users and app services

        for wchar in string.whitespace:
@@ -74,7 +78,7 @@ class DirectoryHandler(BaseHandler):
        # TODO(erikj): Add transactions.
        # TODO(erikj): Check if there is a current association.
        if not servers:
-            users = await self.state.get_current_users_in_room(room_id)
+            users = await self.store.get_users_in_room(room_id)
            servers = {get_domain_from_id(u) for u in users}

        if not servers:
@@ -104,8 +108,9 @@ class DirectoryHandler(BaseHandler):
        """

        user_id = requester.user.to_string()
+        room_alias_str = room_alias.to_string()

-        if len(room_alias.to_string()) > MAX_ALIAS_LENGTH:
+        if len(room_alias_str) > MAX_ALIAS_LENGTH:
            raise SynapseError(
                400,
                "Can't create aliases longer than %s characters" % MAX_ALIAS_LENGTH,
@@ -114,7 +119,7 @@ class DirectoryHandler(BaseHandler):

        service = requester.app_service
        if service:
-            if not service.is_interested_in_alias(room_alias.to_string()):
+            if not service.is_interested_in_alias(room_alias_str):
                raise SynapseError(
                    400,
                    "This application service has not reserved this kind of alias.",
@@ -138,7 +143,7 @@ class DirectoryHandler(BaseHandler):
                raise AuthError(403, "This user is not permitted to create this alias")

            if not self.config.is_alias_creation_allowed(
-                user_id, room_id, room_alias.to_string()
+                user_id, room_id, room_alias_str
            ):
                # Lets just return a generic message, as there may be all sorts of
                # reasons why we said no. TODO: Allow configurable error messages
@@ -211,7 +216,7 @@ class DirectoryHandler(BaseHandler):

    async def delete_appservice_association(
        self, service: ApplicationService, room_alias: RoomAlias
-    ):
+    ) -> None:
        if not service.is_interested_in_alias(room_alias.to_string()):
            raise SynapseError(
                400,
@@ -220,7 +225,7 @@ class DirectoryHandler(BaseHandler):
            )
        await self._delete_association(room_alias)

-    async def _delete_association(self, room_alias: RoomAlias):
+    async def _delete_association(self, room_alias: RoomAlias) -> str:
        if not self.hs.is_mine(room_alias):
            raise SynapseError(400, "Room alias must be local")

@@ -228,17 +233,19 @@ class DirectoryHandler(BaseHandler):

        return room_id

-    async def get_association(self, room_alias: RoomAlias):
+    async def get_association(self, room_alias: RoomAlias) -> JsonDict:
        room_id = None
        if self.hs.is_mine(room_alias):
-            result = await self.get_association_from_room_alias(room_alias)
+            result = await self.get_association_from_room_alias(
+                room_alias
+            )  # type: Optional[RoomAliasMapping]

            if result:
                room_id = result.room_id
                servers = result.servers
        else:
            try:
-                result = await self.federation.make_query(
+                fed_result = await self.federation.make_query(
                    destination=room_alias.domain,
                    query_type="directory",
                    args={"room_alias": room_alias.to_string()},
@@ -248,13 +255,13 @@ class DirectoryHandler(BaseHandler):
            except CodeMessageException as e:
                logging.warning("Error retrieving alias")
                if e.code == 404:
-                    result = None
+                    fed_result = None
                else:
                    raise

-            if result and "room_id" in result and "servers" in result:
-                room_id = result["room_id"]
-                servers = result["servers"]
+            if fed_result and "room_id" in fed_result and "servers" in fed_result:
+                room_id = fed_result["room_id"]
+                servers = fed_result["servers"]

        if not room_id:
            raise SynapseError(
@@ -263,7 +270,7 @@ class DirectoryHandler(BaseHandler):
                Codes.NOT_FOUND,
            )

-        users = await self.state.get_current_users_in_room(room_id)
+        users = await self.store.get_users_in_room(room_id)
        extra_servers = {get_domain_from_id(u) for u in users}
        servers = set(extra_servers) | set(servers)

@@ -275,7 +282,7 @@ class DirectoryHandler(BaseHandler):

        return {"room_id": room_id, "servers": servers}

-    async def on_directory_query(self, args):
+    async def on_directory_query(self, args: JsonDict) -> JsonDict:
        room_alias = RoomAlias.from_string(args["room_alias"])
        if not self.hs.is_mine(room_alias):
            raise SynapseError(400, "Room Alias is not hosted on this homeserver")
@@ -293,7 +300,7 @@ class DirectoryHandler(BaseHandler):

    async def _update_canonical_alias(
        self, requester: Requester, user_id: str, room_id: str, room_alias: RoomAlias
-    ):
+    ) -> None:
        """
        Send an updated canonical alias event if the removed alias was set as
        the canonical alias or listed in the alt_aliases field.
@@ -344,7 +351,9 @@ class DirectoryHandler(BaseHandler):
                ratelimit=False,
            )

-    async def get_association_from_room_alias(self, room_alias: RoomAlias):
+    async def get_association_from_room_alias(
+        self, room_alias: RoomAlias
+    ) -> Optional[RoomAliasMapping]:
        result = await self.store.get_association_from_room_alias(room_alias)
        if not result:
            # Query AS to see if it exists
@@ -372,7 +381,7 @@ class DirectoryHandler(BaseHandler):
        # either no interested services, or no service with an exclusive lock
        return True

-    async def _user_can_delete_alias(self, alias: RoomAlias, user_id: str):
+    async def _user_can_delete_alias(self, alias: RoomAlias, user_id: str) -> bool:
        """Determine whether a user can delete an alias.

        One of the following must be true:
@@ -394,14 +403,13 @@ class DirectoryHandler(BaseHandler):
        if not room_id:
            return False

-        res = await self.auth.check_can_change_room_list(
+        return await self.auth.check_can_change_room_list(
            room_id, UserID.from_string(user_id)
        )
-        return res

    async def edit_published_room_list(
        self, requester: Requester, room_id: str, visibility: str
-    ):
+    ) -> None:
        """Edit the entry of the room in the published room list.

        requester
@@ -469,7 +477,7 @@ class DirectoryHandler(BaseHandler):

    async def edit_published_appservice_room_list(
        self, appservice_id: str, network_id: str, room_id: str, visibility: str
-    ):
+    ) -> None:
        """Add or remove a room from the appservice/network specific public
        room list.

@@ -499,5 +507,4 @@ class DirectoryHandler(BaseHandler):
                room_id, requester.user.to_string()
            )

-        aliases = await self.store.get_aliases_for_room(room_id)
-        return aliases
+        return await self.store.get_aliases_for_room(room_id)
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -103,7 +103,7 @@ class EventStreamHandler(BaseHandler):
                    # Send down presence.
                    if event.state_key == auth_user_id:
                        # Send down presence for everyone in the room.
-                        users = await self.state.get_current_users_in_room(
+                        users = await self.store.get_users_in_room(
                            event.room_id
                        )  # type: Iterable[str]
                    else:
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -2446,7 +2446,9 @@ class FederationHandler(BaseHandler):
        # If we are going to send this event over federation we precaclculate
        # the joined hosts.
        if event.internal_metadata.get_send_on_behalf_of():
-            await self.event_creation_handler.cache_joined_hosts_for_event(event)
+            await self.event_creation_handler.cache_joined_hosts_for_event(
+                event, context
+            )

        return context

--- a/synapse/handlers/identity.py
+++ b/synapse/handlers/identity.py
@@ -17,7 +17,7 @@
 """Utilities for interacting with Identity Servers"""
 import logging
 import urllib.parse
-from typing import Awaitable, Callable, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Awaitable, Callable, Dict, List, Optional, Tuple

 from synapse.api.errors import (
    CodeMessageException,
@@ -41,13 +41,16 @@ from synapse.util.stringutils import (

 from ._base import BaseHandler

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
 logger = logging.getLogger(__name__)

 id_server_scheme = "https://"


 class IdentityHandler(BaseHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super().__init__(hs)

        # An HTTP client for contacting trusted URLs.
@@ -80,7 +83,7 @@ class IdentityHandler(BaseHandler):
        request: SynapseRequest,
        medium: str,
        address: str,
-    ):
+    ) -> None:
        """Used to ratelimit requests to `/requestToken` by IP and address.

        Args:
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -15,7 +15,7 @@
 # limitations under the License.
 import logging
 import random
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Tuple

 from canonicaljson import encode_canonical_json

@@ -51,6 +51,7 @@ from synapse.storage.state import StateFilter
 from synapse.types import Requester, RoomAlias, StreamToken, UserID, create_requester
 from synapse.util import json_decoder, json_encoder
 from synapse.util.async_helpers import Linearizer
+from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.metrics import measure_func
 from synapse.visibility import filter_events_for_client

@@ -66,7 +67,7 @@ logger = logging.getLogger(__name__)
 class MessageHandler:
    """Contains some read only APIs to get state about a room"""

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        self.auth = hs.get_auth()
        self.clock = hs.get_clock()
        self.state = hs.get_state_handler()
@@ -91,7 +92,7 @@ class MessageHandler:
        room_id: str,
        event_type: str,
        state_key: str,
-    ) -> dict:
+    ) -> Optional[EventBase]:
        """Get data from a room.

        Args:
@@ -115,6 +116,10 @@ class MessageHandler:
            data = await self.state.get_current_state(room_id, event_type, state_key)
        elif membership == Membership.LEAVE:
            key = (event_type, state_key)
+            # If the membership is not JOIN, then the event ID should exist.
+            assert (
+                membership_event_id is not None
+            ), "check_user_in_room_or_world_readable returned invalid data"
            room_state = await self.state_store.get_state_for_events(
                [membership_event_id], StateFilter.from_types([key])
            )
@@ -186,10 +191,12 @@ class MessageHandler:

            event = last_events[0]
            if visible_events:
-                room_state = await self.state_store.get_state_for_events(
+                room_state_events = await self.state_store.get_state_for_events(
                    [event.event_id], state_filter=state_filter
                )
-                room_state = room_state[event.event_id]
+                room_state = room_state_events[
+                    event.event_id
+                ]  # type: Mapping[Any, EventBase]
            else:
                raise AuthError(
                    403,
@@ -210,10 +217,14 @@ class MessageHandler:
                )
                room_state = await self.store.get_events(state_ids.values())
            elif membership == Membership.LEAVE:
-                room_state = await self.state_store.get_state_for_events(
+                # If the membership is not JOIN, then the event ID should exist.
+                assert (
+                    membership_event_id is not None
+                ), "check_user_in_room_or_world_readable returned invalid data"
+                room_state_events = await self.state_store.get_state_for_events(
                    [membership_event_id], state_filter=state_filter
                )
-                room_state = room_state[membership_event_id]
+                room_state = room_state_events[membership_event_id]

        now = self.clock.time_msec()
        events = await self._event_serializer.serialize_events(
@@ -248,7 +259,7 @@ class MessageHandler:
                    "Getting joined members after leaving is not implemented"
                )

-        users_with_profile = await self.state.get_current_users_in_room(room_id)
+        users_with_profile = await self.store.get_users_in_room_with_profiles(room_id)

        # If this is an AS, double check that they are allowed to see the members.
        # This can either be because the AS user is in the room or because there
@@ -447,6 +458,19 @@ class EventCreationHandler:

        self._external_cache = hs.get_external_cache()

+        # Stores the state groups we've recently added to the joined hosts
+        # external cache. Note that the timeout must be significantly less than
+        # the TTL on the external cache.
+        self._external_cache_joined_hosts_updates = (
+            None
+        )  # type: Optional[ExpiringCache]
+        if self._external_cache.is_enabled():
+            self._external_cache_joined_hosts_updates = ExpiringCache(
+                "_external_cache_joined_hosts_updates",
+                self.clock,
+                expiry_ms=30 * 60 * 1000,
+            )
+
    async def create_event(
        self,
        requester: Requester,
@@ -957,7 +981,7 @@ class EventCreationHandler:

        await self.action_generator.handle_push_actions_for_event(event, context)

-        await self.cache_joined_hosts_for_event(event)
+        await self.cache_joined_hosts_for_event(event, context)

        try:
            # If we're a worker we need to hit out to the master.
@@ -998,7 +1022,9 @@ class EventCreationHandler:
            await self.store.remove_push_actions_from_staging(event.event_id)
            raise

-    async def cache_joined_hosts_for_event(self, event: EventBase) -> None:
+    async def cache_joined_hosts_for_event(
+        self, event: EventBase, context: EventContext
+    ) -> None:
        """Precalculate the joined hosts at the event, when using Redis, so that
        external federation senders don't have to recalculate it themselves.
        """
@@ -1006,6 +1032,9 @@ class EventCreationHandler:
        if not self._external_cache.is_enabled():
            return

+        # If external cache is enabled we should always have this.
+        assert self._external_cache_joined_hosts_updates is not None
+
        # We actually store two mappings, event ID -> prev state group,
        # state group -> joined hosts, which is much more space efficient
        # than event ID -> joined hosts.
@@ -1013,16 +1042,21 @@ class EventCreationHandler:
        # Note: We have to cache event ID -> prev state group, as we don't
        # store that in the DB.
        #
-        # Note: We always set the state group -> joined hosts cache, even if
-        # we already set it, so that the expiry time is reset.
+        # Note: We set the state group -> joined hosts cache if it hasn't been
+        # set for a while, so that the expiry time is reset.

        state_entry = await self.state.resolve_state_groups_for_events(
            event.room_id, event_ids=event.prev_event_ids()
        )

        if state_entry.state_group:
+            if state_entry.state_group in self._external_cache_joined_hosts_updates:
+                return
+
            joined_hosts = await self.store.get_joined_hosts(event.room_id, state_entry)

+            # Note that the expiry times must be larger than the expiry time in
+            # _external_cache_joined_hosts_updates.
            await self._external_cache.set(
                "event_to_prev_state_group",
                event.event_id,
@@ -1036,6 +1070,8 @@ class EventCreationHandler:
                expiry_ms=60 * 60 * 1000,
            )

+            self._external_cache_joined_hosts_updates[state_entry.state_group] = None
+
    async def _validate_canonical_alias(
        self, directory_handler, room_alias_str: str, expected_room_id: str
    ) -> None:
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -28,6 +28,7 @@ from bisect import bisect
 from contextlib import contextmanager
 from typing import (
    TYPE_CHECKING,
+    Callable,
    Collection,
    Dict,
    FrozenSet,
@@ -232,23 +233,23 @@ class BasePresenceHandler(abc.ABC):
        """

    async def update_external_syncs_row(
-        self, process_id, user_id, is_syncing, sync_time_msec
-    ):
+        self, process_id: str, user_id: str, is_syncing: bool, sync_time_msec: int
+    ) -> None:
        """Update the syncing users for an external process as a delta.

        This is a no-op when presence is handled by a different worker.

        Args:
-            process_id (str): An identifier for the process the users are
+            process_id: An identifier for the process the users are
                syncing against. This allows synapse to process updates
                as user start and stop syncing against a given process.
-            user_id (str): The user who has started or stopped syncing
-            is_syncing (bool): Whether or not the user is now syncing
-            sync_time_msec(int): Time in ms when the user was last syncing
+            user_id: The user who has started or stopped syncing
+            is_syncing: Whether or not the user is now syncing
+            sync_time_msec: Time in ms when the user was last syncing
        """
        pass

-    async def update_external_syncs_clear(self, process_id):
+    async def update_external_syncs_clear(self, process_id: str) -> None:
        """Marks all users that had been marked as syncing by a given process
        as offline.

@@ -304,7 +305,7 @@ class _NullContextManager(ContextManager[None]):


 class WorkerPresenceHandler(BasePresenceHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super().__init__(hs)
        self.hs = hs

@@ -327,7 +328,7 @@ class WorkerPresenceHandler(BasePresenceHandler):

        # user_id -> last_sync_ms. Lists the users that have stopped syncing but
        # we haven't notified the presence writer of that yet
-        self.users_going_offline = {}
+        self.users_going_offline = {}  # type: Dict[str, int]

        self._bump_active_client = ReplicationBumpPresenceActiveTime.make_client(hs)
        self._set_state_client = ReplicationPresenceSetState.make_client(hs)
@@ -346,24 +347,21 @@ class WorkerPresenceHandler(BasePresenceHandler):
            self._on_shutdown,
        )

-    def _on_shutdown(self):
+    def _on_shutdown(self) -> None:
        if self._presence_enabled:
            self.hs.get_tcp_replication().send_command(
                ClearUserSyncsCommand(self.instance_id)
            )

-    def send_user_sync(self, user_id, is_syncing, last_sync_ms):
+    def send_user_sync(self, user_id: str, is_syncing: bool, last_sync_ms: int) -> None:
        if self._presence_enabled:
            self.hs.get_tcp_replication().send_user_sync(
                self.instance_id, user_id, is_syncing, last_sync_ms
            )

-    def mark_as_coming_online(self, user_id):
+    def mark_as_coming_online(self, user_id: str) -> None:
        """A user has started syncing. Send a UserSync to the presence writer,
        unless they had recently stopped syncing.
-
-        Args:
-            user_id (str)
        """
        going_offline = self.users_going_offline.pop(user_id, None)
        if not going_offline:
@@ -371,18 +369,15 @@ class WorkerPresenceHandler(BasePresenceHandler):
            # were offline
            self.send_user_sync(user_id, True, self.clock.time_msec())

-    def mark_as_going_offline(self, user_id):
+    def mark_as_going_offline(self, user_id: str) -> None:
        """A user has stopped syncing. We wait before notifying the presence
        writer as its likely they'll come back soon. This allows us to avoid
        sending a stopped syncing immediately followed by a started syncing
        notification to the presence writer
-
-        Args:
-            user_id (str)
        """
        self.users_going_offline[user_id] = self.clock.time_msec()

-    def send_stop_syncing(self):
+    def send_stop_syncing(self) -> None:
        """Check if there are any users who have stopped syncing a while ago and
        haven't come back yet. If there are poke the presence writer about them.
        """
@@ -430,7 +425,9 @@ class WorkerPresenceHandler(BasePresenceHandler):

        return _user_syncing()

-    async def notify_from_replication(self, states, stream_id):
+    async def notify_from_replication(
+        self, states: List[UserPresenceState], stream_id: int
+    ) -> None:
        parties = await get_interested_parties(self.store, self.presence_router, states)
        room_ids_to_states, users_to_states = parties

@@ -478,7 +475,12 @@ class WorkerPresenceHandler(BasePresenceHandler):
            if count > 0
        ]

-    async def set_state(self, target_user, state, ignore_status_msg=False):
+    async def set_state(
+        self,
+        target_user: UserID,
+        state: JsonDict,
+        ignore_status_msg: bool = False,
+    ) -> None:
        """Set the presence state of the user."""
        presence = state["presence"]

@@ -508,7 +510,7 @@ class WorkerPresenceHandler(BasePresenceHandler):
            ignore_status_msg=ignore_status_msg,
        )

-    async def bump_presence_active_time(self, user):
+    async def bump_presence_active_time(self, user: UserID) -> None:
        """We've seen the user do something that indicates they're interacting
        with the app.
        """
@@ -592,8 +594,8 @@ class PresenceHandler(BasePresenceHandler):
        # we assume that all the sync requests on that process have stopped.
        # Stored as a dict from process_id to set of user_id, and a dict of
        # process_id to millisecond timestamp last updated.
-        self.external_process_to_current_syncs = {}  # type: Dict[int, Set[str]]
-        self.external_process_last_updated_ms = {}  # type: Dict[int, int]
+        self.external_process_to_current_syncs = {}  # type: Dict[str, Set[str]]
+        self.external_process_last_updated_ms = {}  # type: Dict[str, int]

        self.external_sync_linearizer = Linearizer(name="external_sync_linearizer")

@@ -633,7 +635,7 @@ class PresenceHandler(BasePresenceHandler):
        self._event_pos = self.store.get_current_events_token()
        self._event_processing = False

-    async def _on_shutdown(self):
+    async def _on_shutdown(self) -> None:
        """Gets called when shutting down. This lets us persist any updates that
        we haven't yet persisted, e.g. updates that only changes some internal
        timers. This allows changes to persist across startup without having to
@@ -662,7 +664,7 @@ class PresenceHandler(BasePresenceHandler):
            )
        logger.info("Finished _on_shutdown")

-    async def _persist_unpersisted_changes(self):
+    async def _persist_unpersisted_changes(self) -> None:
        """We periodically persist the unpersisted changes, as otherwise they
        may stack up and slow down shutdown times.
        """
@@ -762,7 +764,7 @@ class PresenceHandler(BasePresenceHandler):
                        states, destinations
                    )

-    async def _handle_timeouts(self):
+    async def _handle_timeouts(self) -> None:
        """Checks the presence of users that have timed out and updates as
        appropriate.
        """
@@ -814,7 +816,7 @@ class PresenceHandler(BasePresenceHandler):

        return await self._update_states(changes)

-    async def bump_presence_active_time(self, user):
+    async def bump_presence_active_time(self, user: UserID) -> None:
        """We've seen the user do something that indicates they're interacting
        with the app.
        """
@@ -911,17 +913,17 @@ class PresenceHandler(BasePresenceHandler):
        return []

    async def update_external_syncs_row(
-        self, process_id, user_id, is_syncing, sync_time_msec
-    ):
+        self, process_id: str, user_id: str, is_syncing: bool, sync_time_msec: int
+    ) -> None:
        """Update the syncing users for an external process as a delta.

        Args:
-            process_id (str): An identifier for the process the users are
+            process_id: An identifier for the process the users are
                syncing against. This allows synapse to process updates
                as user start and stop syncing against a given process.
-            user_id (str): The user who has started or stopped syncing
-            is_syncing (bool): Whether or not the user is now syncing
-            sync_time_msec(int): Time in ms when the user was last syncing
+            user_id: The user who has started or stopped syncing
+            is_syncing: Whether or not the user is now syncing
+            sync_time_msec: Time in ms when the user was last syncing
        """
        with (await self.external_sync_linearizer.queue(process_id)):
            prev_state = await self.current_state_for_user(user_id)
@@ -958,7 +960,7 @@ class PresenceHandler(BasePresenceHandler):

            self.external_process_last_updated_ms[process_id] = self.clock.time_msec()

-    async def update_external_syncs_clear(self, process_id):
+    async def update_external_syncs_clear(self, process_id: str) -> None:
        """Marks all users that had been marked as syncing by a given process
        as offline.

@@ -979,12 +981,12 @@ class PresenceHandler(BasePresenceHandler):
            )
            self.external_process_last_updated_ms.pop(process_id, None)

-    async def current_state_for_user(self, user_id):
+    async def current_state_for_user(self, user_id: str) -> UserPresenceState:
        """Get the current presence state for a user."""
        res = await self.current_state_for_users([user_id])
        return res[user_id]

-    async def _persist_and_notify(self, states):
+    async def _persist_and_notify(self, states: List[UserPresenceState]) -> None:
        """Persist states in the database, poke the notifier and send to
        interested remote servers
        """
@@ -1005,7 +1007,7 @@ class PresenceHandler(BasePresenceHandler):
        # stream (which is updated by `store.update_presence`).
        await self.maybe_send_presence_to_interested_destinations(states)

-    async def incoming_presence(self, origin, content):
+    async def incoming_presence(self, origin: str, content: JsonDict) -> None:
        """Called when we receive a `m.presence` EDU from a remote server."""
        if not self._presence_enabled:
            return
@@ -1055,7 +1057,9 @@ class PresenceHandler(BasePresenceHandler):
            federation_presence_counter.inc(len(updates))
            await self._update_states(updates)

-    async def set_state(self, target_user, state, ignore_status_msg=False):
+    async def set_state(
+        self, target_user: UserID, state: JsonDict, ignore_status_msg: bool = False
+    ) -> None:
        """Set the presence state of the user."""
        status_msg = state.get("status_msg", None)
        presence = state["presence"]
@@ -1089,7 +1093,7 @@ class PresenceHandler(BasePresenceHandler):

        await self._update_states([prev_state.copy_and_replace(**new_fields)])

-    async def is_visible(self, observed_user, observer_user):
+    async def is_visible(self, observed_user: UserID, observer_user: UserID) -> bool:
        """Returns whether a user can see another user's presence."""
        observer_room_ids = await self.store.get_rooms_for_user(
            observer_user.to_string()
@@ -1144,7 +1148,7 @@ class PresenceHandler(BasePresenceHandler):
        )
        return rows

-    def notify_new_event(self):
+    def notify_new_event(self) -> None:
        """Called when new events have happened. Handles users and servers
        joining rooms and require being sent presence.
        """
@@ -1163,7 +1167,7 @@ class PresenceHandler(BasePresenceHandler):

        run_as_background_process("presence.notify_new_event", _process_presence)

-    async def _unsafe_process(self):
+    async def _unsafe_process(self) -> None:
        # Loop round handling deltas until we're up to date
        while True:
            with Measure(self.clock, "presence_delta"):
@@ -1179,7 +1183,16 @@ class PresenceHandler(BasePresenceHandler):
                max_pos, deltas = await self.store.get_current_state_deltas(
                    self._event_pos, room_max_stream_ordering
                )
-                await self._handle_state_delta(deltas)
+
+                # We may get multiple deltas for different rooms, but we want to
+                # handle them on a room by room basis, so we batch them up by
+                # room.
+                deltas_by_room: Dict[str, List[JsonDict]] = {}
+                for delta in deltas:
+                    deltas_by_room.setdefault(delta["room_id"], []).append(delta)
+
+                for room_id, deltas_for_room in deltas_by_room.items():
+                    await self._handle_state_delta(room_id, deltas_for_room)

                self._event_pos = max_pos

@@ -1188,17 +1201,21 @@ class PresenceHandler(BasePresenceHandler):
                    max_pos
                )

-    async def _handle_state_delta(self, deltas):
-        """Process current state deltas to find new joins that need to be
-        handled.
+    async def _handle_state_delta(self, room_id: str, deltas: List[JsonDict]) -> None:
+        """Process current state deltas for the room to find new joins that need
+        to be handled.
        """
-        # A map of destination to a set of user state that they should receive
-        presence_destinations = {}  # type: Dict[str, Set[UserPresenceState]]
+
+        # Sets of newly joined users. Note that if the local server is
+        # joining a remote room for the first time we'll see both the joining
+        # user and all remote users as newly joined.
+        newly_joined_users = set()

        for delta in deltas:
+            assert room_id == delta["room_id"]
+
            typ = delta["type"]
            state_key = delta["state_key"]
-            room_id = delta["room_id"]
            event_id = delta["event_id"]
            prev_event_id = delta["prev_event_id"]

@@ -1227,72 +1244,55 @@ class PresenceHandler(BasePresenceHandler):
                    # Ignore changes to join events.
                    continue

-            # Retrieve any user presence state updates that need to be sent as a result,
-            # and the destinations that need to receive it
-            destinations, user_presence_states = await self._on_user_joined_room(
-                room_id, state_key
-            )
+            newly_joined_users.add(state_key)

-            # Insert the destinations and respective updates into our destinations dict
-            for destination in destinations:
-                presence_destinations.setdefault(destination, set()).update(
-                    user_presence_states
-                )
+        if not newly_joined_users:
+            # If nobody has joined then there's nothing to do.
+            return

-        # Send out user presence updates for each destination
-        for destination, user_state_set in presence_destinations.items():
-            self._federation_queue.send_presence_to_destinations(
-                destinations=[destination], states=user_state_set
-            )
+        # We want to send:
+        #   1. presence states of all local users in the room to newly joined
+        #      remote servers
+        #   2. presence states of newly joined users to all remote servers in
+        #      the room.
+        #
+        # TODO: Only send presence states to remote hosts that don't already
+        # have them (because they already share rooms).

-    async def _on_user_joined_room(
-        self, room_id: str, user_id: str
-    ) -> Tuple[List[str], List[UserPresenceState]]:
-        """Called when we detect a user joining the room via the current state
-        delta stream. Returns the destinations that need to be updated and the
-        presence updates to send to them.
+        # Get all the users who were already in the room, by fetching the
+        # current users in the room and removing the newly joined users.
+        users = await self.store.get_users_in_room(room_id)
+        prev_users = set(users) - newly_joined_users

-        Args:
-            room_id: The ID of the room that the user has joined.
-            user_id: The ID of the user that has joined the room.
+        # Construct sets for all the local users and remote hosts that were
+        # already in the room
+        prev_local_users = []
+        prev_remote_hosts = set()
+        for user_id in prev_users:
+            if self.is_mine_id(user_id):
+                prev_local_users.append(user_id)
+            else:
+                prev_remote_hosts.add(get_domain_from_id(user_id))

-        Returns:
-            A tuple of destinations and presence updates to send to them.
-        """
-        if self.is_mine_id(user_id):
-            # If this is a local user then we need to send their presence
-            # out to hosts in the room (who don't already have it)
+        # Similarly, construct sets for all the local users and remote hosts
+        # that were *not* already in the room. Care needs to be taken with the
+        # calculating the remote hosts, as a host may have already been in the
+        # room even if there is a newly joined user from that host.
+        newly_joined_local_users = []
+        newly_joined_remote_hosts = set()
+        for user_id in newly_joined_users:
+            if self.is_mine_id(user_id):
+                newly_joined_local_users.append(user_id)
+            else:
+                host = get_domain_from_id(user_id)
+                if host not in prev_remote_hosts:
+                    newly_joined_remote_hosts.add(host)

-            # TODO: We should be able to filter the hosts down to those that
-            # haven't previously seen the user
-
-            remote_hosts = await self.state.get_current_hosts_in_room(room_id)
-
-            # Filter out ourselves.
-            filtered_remote_hosts = [
-                host for host in remote_hosts if host != self.server_name
-            ]
-
-            state = await self.current_state_for_user(user_id)
-            return filtered_remote_hosts, [state]
-        else:
-            # A remote user has joined the room, so we need to:
-            #   1. Check if this is a new server in the room
-            #   2. If so send any presence they don't already have for
-            #      local users in the room.
-
-            # TODO: We should be able to filter the users down to those that
-            # the server hasn't previously seen
-
-            # TODO: Check that this is actually a new server joining the
-            # room.
-
-            remote_host = get_domain_from_id(user_id)
-
-            users = await self.state.get_current_users_in_room(room_id)
-            user_ids = list(filter(self.is_mine_id, users))
-
-            states_d = await self.current_state_for_users(user_ids)
+        # Send presence states of all local users in the room to newly joined
+        # remote servers. (We actually only send states for local users already
+        # in the room, as we'll send states for newly joined local users below.)
+        if prev_local_users and newly_joined_remote_hosts:
+            local_states = await self.current_state_for_users(prev_local_users)

            # Filter out old presence, i.e. offline presence states where
            # the user hasn't been active for a week. We can change this
@@ -1302,16 +1302,30 @@ class PresenceHandler(BasePresenceHandler):
            now = self.clock.time_msec()
            states = [
                state
-                for state in states_d.values()
+                for state in local_states.values()
                if state.state != PresenceState.OFFLINE
                or now - state.last_active_ts < 7 * 24 * 60 * 60 * 1000
                or state.status_msg is not None
            ]

-            return [remote_host], states
+            self._federation_queue.send_presence_to_destinations(
+                destinations=newly_joined_remote_hosts,
+                states=states,
+            )
+
+        # Send presence states of newly joined users to all remote servers in
+        # the room
+        if newly_joined_local_users and (
+            prev_remote_hosts or newly_joined_remote_hosts
+        ):
+            local_states = await self.current_state_for_users(newly_joined_local_users)
+            self._federation_queue.send_presence_to_destinations(
+                destinations=prev_remote_hosts | newly_joined_remote_hosts,
+                states=list(local_states.values()),
+            )


-def should_notify(old_state, new_state):
+def should_notify(old_state: UserPresenceState, new_state: UserPresenceState) -> bool:
    """Decides if a presence state change should be sent to interested parties."""
    if old_state == new_state:
        return False
@@ -1347,7 +1361,9 @@ def should_notify(old_state, new_state):
    return False


-def format_user_presence_state(state, now, include_user_id=True):
+def format_user_presence_state(
+    state: UserPresenceState, now: int, include_user_id: bool = True
+) -> JsonDict:
    """Convert UserPresenceState to a format that can be sent down to clients
    and to other servers.

@@ -1385,11 +1401,11 @@ class PresenceEventSource:
    @log_function
    async def get_new_events(
        self,
-        user,
-        from_key,
-        room_ids=None,
-        include_offline=True,
-        explicit_room_id=None,
+        user: UserID,
+        from_key: Optional[int],
+        room_ids: Optional[List[str]] = None,
+        include_offline: bool = True,
+        explicit_room_id: Optional[str] = None,
        **kwargs,
    ) -> Tuple[List[UserPresenceState], int]:
        # The process for getting presence events are:
@@ -1594,7 +1610,7 @@ class PresenceEventSource:
            if update.state != PresenceState.OFFLINE
        ]

-    def get_current_key(self):
+    def get_current_key(self) -> int:
        return self.store.get_current_presence_token()

    @cached(num_args=2, cache_context=True)
@@ -1654,15 +1670,20 @@ class PresenceEventSource:
        return users_interested_in


-def handle_timeouts(user_states, is_mine_fn, syncing_user_ids, now):
+def handle_timeouts(
+    user_states: List[UserPresenceState],
+    is_mine_fn: Callable[[str], bool],
+    syncing_user_ids: Set[str],
+    now: int,
+) -> List[UserPresenceState]:
    """Checks the presence of users that have timed out and updates as
    appropriate.

    Args:
-        user_states(list): List of UserPresenceState's to check.
-        is_mine_fn (fn): Function that returns if a user_id is ours
-        syncing_user_ids (set): Set of user_ids with active syncs.
-        now (int): Current time in ms.
+        user_states: List of UserPresenceState's to check.
+        is_mine_fn: Function that returns if a user_id is ours
+        syncing_user_ids: Set of user_ids with active syncs.
+        now: Current time in ms.

    Returns:
        List of UserPresenceState updates
@@ -1679,14 +1700,16 @@ def handle_timeouts(user_states, is_mine_fn, syncing_user_ids, now):
    return list(changes.values())


-def handle_timeout(state, is_mine, syncing_user_ids, now):
+def handle_timeout(
+    state: UserPresenceState, is_mine: bool, syncing_user_ids: Set[str], now: int
+) -> Optional[UserPresenceState]:
    """Checks the presence of the user to see if any of the timers have elapsed

    Args:
-        state (UserPresenceState)
-        is_mine (bool): Whether the user is ours
-        syncing_user_ids (set): Set of user_ids with active syncs.
-        now (int): Current time in ms.
+        state
+        is_mine: Whether the user is ours
+        syncing_user_ids: Set of user_ids with active syncs.
+        now: Current time in ms.

    Returns:
        A UserPresenceState update or None if no update.
@@ -1738,23 +1761,29 @@ def handle_timeout(state, is_mine, syncing_user_ids, now):
    return state if changed else None


-def handle_update(prev_state, new_state, is_mine, wheel_timer, now):
+def handle_update(
+    prev_state: UserPresenceState,
+    new_state: UserPresenceState,
+    is_mine: bool,
+    wheel_timer: WheelTimer,
+    now: int,
+) -> Tuple[UserPresenceState, bool, bool]:
    """Given a presence update:
        1. Add any appropriate timers.
        2. Check if we should notify anyone.

    Args:
-        prev_state (UserPresenceState)
-        new_state (UserPresenceState)
-        is_mine (bool): Whether the user is ours
-        wheel_timer (WheelTimer)
-        now (int): Time now in ms
+        prev_state
+        new_state
+        is_mine: Whether the user is ours
+        wheel_timer
+        now: Time now in ms

    Returns:
        3-tuple: `(new_state, persist_and_notify, federation_ping)` where:
            - new_state: is the state to actually persist
-            - persist_and_notify (bool): whether to persist and notify people
-            - federation_ping (bool): whether we should send a ping over federation
+            - persist_and_notify: whether to persist and notify people
+            - federation_ping: whether we should send a ping over federation
    """
    user_id = new_state.user_id

@@ -2026,18 +2055,40 @@ class PresenceFederationQueue:
            )
            return result["updates"], result["upto_token"], result["limited"]

+        # If the from_token is the current token then there's nothing to return
+        # and we can trivially no-op.
+        if from_token == self._next_id - 1:
+            return [], upto_token, False
+
        # We can find the correct position in the queue by noting that there is
        # exactly one entry per stream ID, and that the last entry has an ID of
        # `self._next_id - 1`, so we can count backwards from the end.
        #
+        # Since we are returning all states in the range `from_token < stream_id
+        # <= upto_token` we look for the index with a `stream_id` of `from_token
+        # + 1`.
+        #
        # Since the start of the queue is periodically truncated we need to
        # handle the case where `from_token` stream ID has already been dropped.
-        start_idx = max(from_token - self._next_id, -len(self._queue))
+        start_idx = max(from_token + 1 - self._next_id, -len(self._queue))

        to_send = []  # type: List[Tuple[int, Tuple[str, str]]]
        limited = False
        new_id = upto_token
        for _, stream_id, destinations, user_ids in self._queue[start_idx:]:
+            if stream_id <= from_token:
+                # Paranoia check that we are actually only sending states that
+                # are have stream_id strictly greater than from_token. We should
+                # never hit this.
+                logger.warning(
+                    "Tried returning presence federation stream ID: %d less than from_token: %d (next_id: %d, len: %d)",
+                    stream_id,
+                    from_token,
+                    self._next_id,
+                    len(self._queue),
+                )
+                continue
+
            if stream_id > upto_token:
                break

--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -1327,7 +1327,7 @@ class RoomShutdownHandler:
            new_room_id = None
            logger.info("Shutting down room %r", room_id)

-        users = await self.state.get_current_users_in_room(room_id)
+        users = await self.store.get_users_in_room(room_id)
        kicked_users = []
        failed_to_kick_users = []
        for user_id in users:
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -1044,7 +1044,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):


 class RoomMemberMasterHandler(RoomMemberHandler):
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super().__init__(hs)

        self.distributor = hs.get_distributor()
--- a/synapse/handlers/space_summary.py
+++ b/synapse/handlers/space_summary.py
@@ -288,6 +288,7 @@ class SpaceSummaryHandler:
            ev.data
            for ev in res.events
            if ev.event_type == EventTypes.MSC1772_SPACE_CHILD
+            or ev.event_type == EventTypes.SpaceChild
        )

    async def _is_room_accessible(self, room_id: str, requester: Optional[str]) -> bool:
@@ -331,7 +332,9 @@ class SpaceSummaryHandler:
        )

        # TODO: update once MSC1772 lands
-        room_type = create_event.content.get(EventContentFields.MSC1772_ROOM_TYPE)
+        room_type = create_event.content.get(EventContentFields.ROOM_TYPE)
+        if not room_type:
+            room_type = create_event.content.get(EventContentFields.MSC1772_ROOM_TYPE)

        entry = {
            "room_id": stats["room_id"],
@@ -344,6 +347,7 @@ class SpaceSummaryHandler:
                stats["history_visibility"] == HistoryVisibility.WORLD_READABLE
            ),
            "guest_can_join": stats["guest_access"] == "can_join",
+            "creation_ts": create_event.origin_server_ts,
            "room_type": room_type,
        }

@@ -360,8 +364,9 @@ class SpaceSummaryHandler:
            [
                event_id
                for key, event_id in current_state_ids.items()
-                # TODO: update once MSC1772 lands
+                # TODO: update once MSC1772 has been FCP for a period of time.
                if key[0] == EventTypes.MSC1772_SPACE_CHILD
+                or key[0] == EventTypes.SpaceChild
            ]
        )

--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -1190,7 +1190,7 @@ class SyncHandler:

            # Step 1b, check for newly joined rooms
            for room_id in newly_joined_rooms:
-                joined_users = await self.state.get_current_users_in_room(room_id)
+                joined_users = await self.store.get_users_in_room(room_id)
                newly_joined_or_invited_users.update(joined_users)

            # TODO: Check that these users are actually new, i.e. either they
@@ -1206,7 +1206,7 @@ class SyncHandler:

            # Now find users that we no longer track
            for room_id in newly_left_rooms:
-                left_users = await self.state.get_current_users_in_room(room_id)
+                left_users = await self.store.get_users_in_room(room_id)
                newly_left_users.update(left_users)

            # Remove any users that we still share a room with.
@@ -1361,7 +1361,7 @@ class SyncHandler:

        extra_users_ids = set(newly_joined_or_invited_users)
        for room_id in newly_joined_rooms:
-            users = await self.state.get_current_users_in_room(room_id)
+            users = await self.store.get_users_in_room(room_id)
            extra_users_ids.update(users)
        extra_users_ids.discard(user.to_string())

--- a/synapse/handlers/ui_auth/checkers.py
+++ b/synapse/handlers/ui_auth/checkers.py
@@ -13,7 +13,7 @@
 # limitations under the License.

 import logging
-from typing import Any
+from typing import TYPE_CHECKING, Any

 from twisted.web.client import PartialDownloadError

@@ -22,13 +22,16 @@ from synapse.api.errors import Codes, LoginError, SynapseError
 from synapse.config.emailconfig import ThreepidBehaviour
 from synapse.util import json_decoder

+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
 logger = logging.getLogger(__name__)


 class UserInteractiveAuthChecker:
    """Abstract base class for an interactive auth checker"""

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        pass

    def is_enabled(self) -> bool:
@@ -57,10 +60,10 @@ class UserInteractiveAuthChecker:
 class DummyAuthChecker(UserInteractiveAuthChecker):
    AUTH_TYPE = LoginType.DUMMY

-    def is_enabled(self):
+    def is_enabled(self) -> bool:
        return True

-    async def check_auth(self, authdict, clientip):
+    async def check_auth(self, authdict: dict, clientip: str) -> Any:
        return True


@@ -70,24 +73,24 @@ class TermsAuthChecker(UserInteractiveAuthChecker):
    def is_enabled(self):
        return True

-    async def check_auth(self, authdict, clientip):
+    async def check_auth(self, authdict: dict, clientip: str) -> Any:
        return True


 class RecaptchaAuthChecker(UserInteractiveAuthChecker):
    AUTH_TYPE = LoginType.RECAPTCHA

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        super().__init__(hs)
        self._enabled = bool(hs.config.recaptcha_private_key)
        self._http_client = hs.get_proxied_http_client()
        self._url = hs.config.recaptcha_siteverify_api
        self._secret = hs.config.recaptcha_private_key

-    def is_enabled(self):
+    def is_enabled(self) -> bool:
        return self._enabled

-    async def check_auth(self, authdict, clientip):
+    async def check_auth(self, authdict: dict, clientip: str) -> Any:
        try:
            user_response = authdict["response"]
        except KeyError:
@@ -132,11 +135,11 @@ class RecaptchaAuthChecker(UserInteractiveAuthChecker):


 class _BaseThreepidAuthChecker:
-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        self.hs = hs
        self.store = hs.get_datastore()

-    async def _check_threepid(self, medium, authdict):
+    async def _check_threepid(self, medium: str, authdict: dict) -> dict:
        if "threepid_creds" not in authdict:
            raise LoginError(400, "Missing threepid_creds", Codes.MISSING_PARAM)

@@ -206,31 +209,31 @@ class _BaseThreepidAuthChecker:
 class EmailIdentityAuthChecker(UserInteractiveAuthChecker, _BaseThreepidAuthChecker):
    AUTH_TYPE = LoginType.EMAIL_IDENTITY

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        UserInteractiveAuthChecker.__init__(self, hs)
        _BaseThreepidAuthChecker.__init__(self, hs)

-    def is_enabled(self):
+    def is_enabled(self) -> bool:
        return self.hs.config.threepid_behaviour_email in (
            ThreepidBehaviour.REMOTE,
            ThreepidBehaviour.LOCAL,
        )

-    async def check_auth(self, authdict, clientip):
+    async def check_auth(self, authdict: dict, clientip: str) -> Any:
        return await self._check_threepid("email", authdict)


 class MsisdnAuthChecker(UserInteractiveAuthChecker, _BaseThreepidAuthChecker):
    AUTH_TYPE = LoginType.MSISDN

-    def __init__(self, hs):
+    def __init__(self, hs: "HomeServer"):
        UserInteractiveAuthChecker.__init__(self, hs)
        _BaseThreepidAuthChecker.__init__(self, hs)

-    def is_enabled(self):
+    def is_enabled(self) -> bool:
        return bool(self.hs.config.account_threepid_delegate_msisdn)

-    async def check_auth(self, authdict, clientip):
+    async def check_auth(self, authdict: dict, clientip: str) -> Any:
        return await self._check_threepid("msisdn", authdict)


--- a/synapse/metrics/init.py
+++ b/synapse/metrics/init.py
@@ -535,6 +535,13 @@ class ReactorLastSeenMetric:

 REGISTRY.register(ReactorLastSeenMetric())

+# The minimum time in seconds between GCs for each generation, regardless of the current GC
+# thresholds and counts.
+MIN_TIME_BETWEEN_GCS = (1.0, 10.0, 30.0)
+
+# The time (in seconds since the epoch) of the last time we did a GC for each generation.
+_last_gc = [0.0, 0.0, 0.0]
+

 def runUntilCurrentTimer(reactor, func):
    @functools.wraps(func)
@@ -575,11 +582,16 @@ def runUntilCurrentTimer(reactor, func):
            return ret

        # Check if we need to do a manual GC (since its been disabled), and do
-        # one if necessary.
+        # one if necessary. Note we go in reverse order as e.g. a gen 1 GC may
+        # promote an object into gen 2, and we don't want to handle the same
+        # object multiple times.
        threshold = gc.get_threshold()
        counts = gc.get_count()
        for i in (2, 1, 0):
-            if threshold[i] < counts[i]:
+            # We check if we need to do one based on a straightforward
+            # comparison between the threshold and count. We also do an extra
+            # check to make sure that we don't a GC too often.
+            if threshold[i] < counts[i] and MIN_TIME_BETWEEN_GCS[i] < end - _last_gc[i]:
                if i == 0:
                    logger.debug("Collecting gc %d", i)
                else:
@@ -589,6 +601,8 @@ def runUntilCurrentTimer(reactor, func):
                unreachable = gc.collect(i)
                end = time.time()

+                _last_gc[i] = end
+
                gc_time.labels(i).observe(end - start)
                gc_unreachable.labels(i).set(unreachable)

--- a/synapse/python_dependencies.py
+++ b/synapse/python_dependencies.py
@@ -78,14 +78,15 @@ REQUIREMENTS = [
    # we use attr.validators.deep_iterable, which arrived in 19.1.0 (Note:
    # Fedora 31 only has 19.1, so if we want to upgrade we should wait until 33
    # is out in November.)
-    "attrs>=19.1.0",
+    # Note: 21.1.0 broke `/sync`, see #9936
+    "attrs>=19.1.0,<21.1.0",
    "netaddr>=0.7.18",
    "Jinja2>=2.9",
    "bleach>=1.4.3",
    "typing-extensions>=3.7.4",
    # We enforce that we have a `cryptography` version that bundles an `openssl`
    # with the latest security patches.
-    "cryptography>=3.4.7;python_version>='3.6'",
+    "cryptography>=3.4.7",
 ]

 CONDITIONAL_REQUIREMENTS = {
@@ -100,14 +101,9 @@ CONDITIONAL_REQUIREMENTS = {
    # that use the protocol, such as Let's Encrypt.
    "acme": [
        "txacme>=0.9.2",
-        # txacme depends on eliot. Eliot 1.8.0 is incompatible with
-        # python 3.5.2, as per https://github.com/itamarst/eliot/issues/418
-        "eliot<1.8.0;python_version<'3.5.3'",
    ],
    "saml2": [
-        # pysaml2 6.4.0 is incompatible with Python 3.5 (see https://github.com/IdentityPython/pysaml2/issues/749)
-        "pysaml2>=4.5.0,<6.4.0;python_version<'3.6'",
-        "pysaml2>=4.5.0;python_version>='3.6'",
+        "pysaml2>=4.5.0",
    ],
    "oidc": ["authlib>=0.14.0"],
    # systemd-python is necessary for logging to the systemd journal via
@@ -121,6 +117,8 @@ CONDITIONAL_REQUIREMENTS = {
    # hiredis is not a *strict* dependency, but it makes things much faster.
    # (if it is not installed, we fall back to slow code.)
    "redis": ["txredisapi>=1.4.7", "hiredis"],
+    # Required to use experimental `caches.track_memory_usage` config option.
+    "cache_memory": ["pympler"],
 }

 ALL_OPTIONAL_REQUIREMENTS = set()  # type: Set[str]
--- a/synapse/replication/tcp/external_cache.py
+++ b/synapse/replication/tcp/external_cache.py
@@ -15,7 +15,7 @@
 import logging
 from typing import TYPE_CHECKING, Any, Optional

-from prometheus_client import Counter
+from prometheus_client import Counter, Histogram

 from synapse.logging.context import make_deferred_yieldable
 from synapse.util import json_decoder, json_encoder
@@ -35,6 +35,20 @@ get_counter = Counter(
    labelnames=["cache_name", "hit"],
 )

+response_timer = Histogram(
+    "synapse_external_cache_response_time_seconds",
+    "Time taken to get a response from Redis for a cache get/set request",
+    labelnames=["method"],
+    buckets=(
+        0.001,
+        0.002,
+        0.005,
+        0.01,
+        0.02,
+        0.05,
+    ),
+)
+

 logger = logging.getLogger(__name__)

@@ -72,13 +86,14 @@ class ExternalCache:

        logger.debug("Caching %s %s: %r", cache_name, key, encoded_value)

-        return await make_deferred_yieldable(
-            self._redis_connection.set(
-                self._get_redis_key(cache_name, key),
-                encoded_value,
-                pexpire=expiry_ms,
+        with response_timer.labels("set").time():
+            return await make_deferred_yieldable(
+                self._redis_connection.set(
+                    self._get_redis_key(cache_name, key),
+                    encoded_value,
+                    pexpire=expiry_ms,
+                )
            )
-        )

    async def get(self, cache_name: str, key: str) -> Optional[Any]:
        """Look up a key/value in the named cache."""
@@ -86,9 +101,10 @@ class ExternalCache:
        if self._redis_connection is None:
            return None

-        result = await make_deferred_yieldable(
-            self._redis_connection.get(self._get_redis_key(cache_name, key))
-        )
+        with response_timer.labels("get").time():
+            result = await make_deferred_yieldable(
+                self._redis_connection.get(self._get_redis_key(cache_name, key))
+            )

        logger.debug("Got cache result %s %s: %r", cache_name, key, result)

--- a/synapse/rest/admin/rooms.py
+++ b/synapse/rest/admin/rooms.py
@@ -37,9 +37,11 @@ from synapse.types import JsonDict, RoomAlias, RoomID, UserID, create_requester
 from synapse.util import json_decoder

 if TYPE_CHECKING:
+    from synapse.api.auth import Auth
+    from synapse.handlers.pagination import PaginationHandler
+    from synapse.handlers.room import RoomShutdownHandler
    from synapse.server import HomeServer

-
 logger = logging.getLogger(__name__)


@@ -146,50 +148,14 @@ class DeleteRoomRestServlet(RestServlet):
    async def on_POST(
        self, request: SynapseRequest, room_id: str
    ) -> Tuple[int, JsonDict]:
-        requester = await self.auth.get_user_by_req(request)
-        await assert_user_is_admin(self.auth, requester.user)
-
-        content = parse_json_object_from_request(request)
-
-        block = content.get("block", False)
-        if not isinstance(block, bool):
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Param 'block' must be a boolean, if given",
-                Codes.BAD_JSON,
-            )
-
-        purge = content.get("purge", True)
-        if not isinstance(purge, bool):
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Param 'purge' must be a boolean, if given",
-                Codes.BAD_JSON,
-            )
-
-        force_purge = content.get("force_purge", False)
-        if not isinstance(force_purge, bool):
-            raise SynapseError(
-                HTTPStatus.BAD_REQUEST,
-                "Param 'force_purge' must be a boolean, if given",
-                Codes.BAD_JSON,
-            )
-
-        ret = await self.room_shutdown_handler.shutdown_room(
-            room_id=room_id,
-            new_room_user_id=content.get("new_room_user_id"),
-            new_room_name=content.get("room_name"),
-            message=content.get("message"),
-            requester_user_id=requester.user.to_string(),
-            block=block,
+        return await _delete_room(
+            request,
+            room_id,
+            self.auth,
+            self.room_shutdown_handler,
+            self.pagination_handler,
        )

-        # Purge room
-        if purge:
-            await self.pagination_handler.purge_room(room_id, force=force_purge)
-
-        return (200, ret)
-

 class ListRoomRestServlet(RestServlet):
    """
@@ -282,7 +248,22 @@ class ListRoomRestServlet(RestServlet):


 class RoomRestServlet(RestServlet):
-    """Get room details.
+    """Manage a room.
+
+    On GET : Get details of a room.
+
+    On DELETE : Delete a room from server.
+
+    It is a combination and improvement of shutdown and purge room.
+
+    Shuts down a room by removing all local users from the room.
+    Blocking all future invites and joins to the room is optional.
+
+    If desired any local aliases will be repointed to a new room
+    created by `new_room_user_id` and kicked users will be auto-
+    joined to the new room.
+
+    If 'purge' is true, it will remove all traces of a room from the database.

    TODO: Add on_POST to allow room creation without joining the room
    """
@@ -293,6 +274,8 @@ class RoomRestServlet(RestServlet):
        self.hs = hs
        self.auth = hs.get_auth()
        self.store = hs.get_datastore()
+        self.room_shutdown_handler = hs.get_room_shutdown_handler()
+        self.pagination_handler = hs.get_pagination_handler()

    async def on_GET(
        self, request: SynapseRequest, room_id: str
@@ -308,6 +291,17 @@ class RoomRestServlet(RestServlet):

        return (200, ret)

+    async def on_DELETE(
+        self, request: SynapseRequest, room_id: str
+    ) -> Tuple[int, JsonDict]:
+        return await _delete_room(
+            request,
+            room_id,
+            self.auth,
+            self.room_shutdown_handler,
+            self.pagination_handler,
+        )
+

 class RoomMembersRestServlet(RestServlet):
    """
@@ -694,3 +688,55 @@ class RoomEventContextServlet(RestServlet):
        )

        return 200, results
+
+
+async def _delete_room(
+    request: SynapseRequest,
+    room_id: str,
+    auth: "Auth",
+    room_shutdown_handler: "RoomShutdownHandler",
+    pagination_handler: "PaginationHandler",
+) -> Tuple[int, JsonDict]:
+    requester = await auth.get_user_by_req(request)
+    await assert_user_is_admin(auth, requester.user)
+
+    content = parse_json_object_from_request(request)
+
+    block = content.get("block", False)
+    if not isinstance(block, bool):
+        raise SynapseError(
+            HTTPStatus.BAD_REQUEST,
+            "Param 'block' must be a boolean, if given",
+            Codes.BAD_JSON,
+        )
+
+    purge = content.get("purge", True)
+    if not isinstance(purge, bool):
+        raise SynapseError(
+            HTTPStatus.BAD_REQUEST,
+            "Param 'purge' must be a boolean, if given",
+            Codes.BAD_JSON,
+        )
+
+    force_purge = content.get("force_purge", False)
+    if not isinstance(force_purge, bool):
+        raise SynapseError(
+            HTTPStatus.BAD_REQUEST,
+            "Param 'force_purge' must be a boolean, if given",
+            Codes.BAD_JSON,
+        )
+
+    ret = await room_shutdown_handler.shutdown_room(
+        room_id=room_id,
+        new_room_user_id=content.get("new_room_user_id"),
+        new_room_name=content.get("room_name"),
+        message=content.get("message"),
+        requester_user_id=requester.user.to_string(),
+        block=block,
+    )
+
+    # Purge room
+    if purge:
+        await pagination_handler.purge_room(room_id, force=force_purge)
+
+    return (200, ret)
--- a/synapse/rest/admin/users.py
+++ b/synapse/rest/admin/users.py
@@ -14,6 +14,7 @@
 import hashlib
 import hmac
 import logging
+import secrets
 from http import HTTPStatus
 from typing import TYPE_CHECKING, Dict, List, Optional, Tuple

@@ -375,7 +376,7 @@ class UserRegisterServlet(RestServlet):
        """
        self._clear_old_nonces()

-        nonce = self.hs.get_secrets().token_hex(64)
+        nonce = secrets.token_hex(64)
        self.nonces[nonce] = int(self.reactor.seconds())
        return 200, {"nonce": nonce}

--- a/synapse/rest/client/v2_alpha/sync.py
+++ b/synapse/rest/client/v2_alpha/sync.py
@@ -14,6 +14,7 @@

 import itertools
 import logging
+from collections import defaultdict
 from typing import TYPE_CHECKING, Tuple

 from synapse.api.constants import PresenceState
@@ -229,24 +230,49 @@ class SyncRestServlet(RestServlet):
        )

        logger.debug("building sync response dict")
-        return {
-            "account_data": {"events": sync_result.account_data},
-            "to_device": {"events": sync_result.to_device},
-            "device_lists": {
-                "changed": list(sync_result.device_lists.changed),
-                "left": list(sync_result.device_lists.left),
-            },
-            "presence": SyncRestServlet.encode_presence(sync_result.presence, time_now),
-            "rooms": {"join": joined, "invite": invited, "leave": archived},
-            "groups": {
-                "join": sync_result.groups.join,
-                "invite": sync_result.groups.invite,
-                "leave": sync_result.groups.leave,
-            },
-            "device_one_time_keys_count": sync_result.device_one_time_keys_count,
-            "org.matrix.msc2732.device_unused_fallback_key_types": sync_result.device_unused_fallback_key_types,
-            "next_batch": await sync_result.next_batch.to_string(self.store),
-        }
+
+        response: dict = defaultdict(dict)
+        response["next_batch"] = await sync_result.next_batch.to_string(self.store)
+
+        if sync_result.account_data:
+            response["account_data"] = {"events": sync_result.account_data}
+        if sync_result.presence:
+            response["presence"] = SyncRestServlet.encode_presence(
+                sync_result.presence, time_now
+            )
+
+        if sync_result.to_device:
+            response["to_device"] = {"events": sync_result.to_device}
+
+        if sync_result.device_lists.changed:
+            response["device_lists"]["changed"] = list(sync_result.device_lists.changed)
+        if sync_result.device_lists.left:
+            response["device_lists"]["left"] = list(sync_result.device_lists.left)
+
+        if sync_result.device_one_time_keys_count:
+            response[
+                "device_one_time_keys_count"
+            ] = sync_result.device_one_time_keys_count
+        if sync_result.device_unused_fallback_key_types:
+            response[
+                "org.matrix.msc2732.device_unused_fallback_key_types"
+            ] = sync_result.device_unused_fallback_key_types
+
+        if joined:
+            response["rooms"]["join"] = joined
+        if invited:
+            response["rooms"]["invite"] = invited
+        if archived:
+            response["rooms"]["leave"] = archived
+
+        if sync_result.groups.join:
+            response["groups"]["join"] = sync_result.groups.join
+        if sync_result.groups.invite:
+            response["groups"]["invite"] = sync_result.groups.invite
+        if sync_result.groups.leave:
+            response["groups"]["leave"] = sync_result.groups.leave
+
+        return response

    @staticmethod
    def encode_presence(events, time_now):
--- a/synapse/rest/consent/consent_resource.py
+++ b/synapse/rest/consent/consent_resource.py
@@ -32,14 +32,6 @@ TEMPLATE_LANGUAGE = "en"

 logger = logging.getLogger(__name__)

-# use hmac.compare_digest if we have it (python 2.7.7), else just use equality
-if hasattr(hmac, "compare_digest"):
-    compare_digest = hmac.compare_digest
-else:
-
-    def compare_digest(a, b):
-        return a == b
-

 class ConsentResource(DirectServeHtmlResource):
    """A twisted Resource to display a privacy policy and gather consent to it
@@ -209,5 +201,5 @@ class ConsentResource(DirectServeHtmlResource):
            .encode("ascii")
        )

-        if not compare_digest(want_mac, userhmac):
+        if not hmac.compare_digest(want_mac, userhmac):
            raise SynapseError(HTTPStatus.FORBIDDEN, "HMAC incorrect")
--- a/synapse/rest/media/v1/filepath.py
+++ b/synapse/rest/media/v1/filepath.py
@@ -21,7 +21,7 @@ from typing import Callable, List
 NEW_FORMAT_ID_RE = re.compile(r"^\d\d\d\d-\d\d-\d\d")


-def _wrap_in_base_path(func: "Callable[..., str]") -> "Callable[..., str]":
+def _wrap_in_base_path(func: Callable[..., str]) -> Callable[..., str]:
    """Takes a function that returns a relative path and turns it into an
    absolute path based on the location of the primary media store
    """
--- a/synapse/secrets.py
+++ b/synapse/secrets.py
@@ -1,44 +0,0 @@
-# Copyright 2018 New Vector Ltd
-#
-# 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.
-
-"""
-Injectable secrets module for Synapse.
-
-See https://docs.python.org/3/library/secrets.html#module-secrets for the API
-used in Python 3.6, and the API emulated in Python 2.7.
-"""
-import sys
-
-# secrets is available since python 3.6
-if sys.version_info[0:2] >= (3, 6):
-    import secrets
-
-    class Secrets:
-        def token_bytes(self, nbytes: int = 32) -> bytes:
-            return secrets.token_bytes(nbytes)
-
-        def token_hex(self, nbytes: int = 32) -> str:
-            return secrets.token_hex(nbytes)
-
-
-else:
-    import binascii
-    import os
-
-    class Secrets:
-        def token_bytes(self, nbytes: int = 32) -> bytes:
-            return os.urandom(nbytes)
-
-        def token_hex(self, nbytes: int = 32) -> str:
-            return binascii.hexlify(self.token_bytes(nbytes)).decode("ascii")
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -126,7 +126,6 @@ from synapse.rest.media.v1.media_repository import (
    MediaRepository,
    MediaRepositoryResource,
 )
-from synapse.secrets import Secrets
 from synapse.server_notices.server_notices_manager import ServerNoticesManager
 from synapse.server_notices.server_notices_sender import ServerNoticesSender
 from synapse.server_notices.worker_server_notices_sender import (
@@ -641,10 +640,6 @@ class HomeServer(metaclass=abc.ABCMeta):
    def get_groups_attestation_renewer(self) -> GroupAttestionRenewer:
        return GroupAttestionRenewer(self)

-    @cache_in_self
-    def get_secrets(self) -> Secrets:
-        return Secrets()
-
    @cache_in_self
    def get_stats_handler(self) -> StatsHandler:
        return StatsHandler(self)
--- a/synapse/state/init.py
+++ b/synapse/state/init.py
@@ -213,19 +213,23 @@ class StateHandler:
        return ret.state

    async def get_current_users_in_room(
-        self, room_id: str, latest_event_ids: Optional[List[str]] = None
+        self, room_id: str, latest_event_ids: List[str]
    ) -> Dict[str, ProfileInfo]:
        """
        Get the users who are currently in a room.

+        Note: This is much slower than using the equivalent method
+        `DataStore.get_users_in_room` or `DataStore.get_users_in_room_with_profiles`,
+        so this should only be used when wanting the users at a particular point
+        in the room.
+
        Args:
            room_id: The ID of the room.
            latest_event_ids: Precomputed list of latest event IDs. Will be computed if None.
        Returns:
            Dictionary of user IDs to their profileinfo.
        """
-        if not latest_event_ids:
-            latest_event_ids = await self.store.get_latest_event_ids_in_room(room_id)
+
        assert latest_event_ids is not None

        logger.debug("calling resolve_state_groups from get_current_users_in_room")
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -69,6 +69,7 @@ class SQLBaseStore(metaclass=ABCMeta):
            self._attempt_to_invalidate_cache("is_host_joined", (room_id, host))

        self._attempt_to_invalidate_cache("get_users_in_room", (room_id,))
+        self._attempt_to_invalidate_cache("get_users_in_room_with_profiles", (room_id,))
        self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
        self._attempt_to_invalidate_cache("get_current_state_ids", (room_id,))

@@ -114,7 +115,7 @@ def db_to_json(db_content: Union[memoryview, bytes, bytearray, str]) -> Any:
        db_content = db_content.tobytes()

    # Decode it to a Unicode string before feeding it to the JSON decoder, since
-    # Python 3.5 does not support deserializing bytes.
+    # it only supports handling strings
    if isinstance(db_content, (bytes, bytearray)):
        db_content = db_content.decode("utf8")

--- a/synapse/storage/database.py
+++ b/synapse/storage/database.py
@@ -171,10 +171,7 @@ class LoggingDatabaseConnection:


 # The type of entry which goes on our after_callbacks and exception_callbacks lists.
-#
-# Python 3.5.2 doesn't support Callable with an ellipsis, so we wrap it in quotes so
-# that mypy sees the type but the runtime python doesn't.
-_CallbackListEntry = Tuple["Callable[..., None]", Iterable[Any], Dict[str, Any]]
+_CallbackListEntry = Tuple[Callable[..., None], Iterable[Any], Dict[str, Any]]


 R = TypeVar("R")
@@ -221,7 +218,7 @@ class LoggingTransaction:
        self.after_callbacks = after_callbacks
        self.exception_callbacks = exception_callbacks

-    def call_after(self, callback: "Callable[..., None]", *args: Any, **kwargs: Any):
+    def call_after(self, callback: Callable[..., None], *args: Any, **kwargs: Any):
        """Call the given callback on the main twisted thread after the
        transaction has finished. Used to invalidate the caches on the
        correct thread.
@@ -233,7 +230,7 @@ class LoggingTransaction:
        self.after_callbacks.append((callback, args, kwargs))

    def call_on_exception(
-        self, callback: "Callable[..., None]", *args: Any, **kwargs: Any
+        self, callback: Callable[..., None], *args: Any, **kwargs: Any
    ):
        # if self.exception_callbacks is None, that means that whatever constructed the
        # LoggingTransaction isn't expecting there to be any callbacks; assert that
@@ -485,7 +482,7 @@ class DatabasePool:
        desc: str,
        after_callbacks: List[_CallbackListEntry],
        exception_callbacks: List[_CallbackListEntry],
-        func: "Callable[..., R]",
+        func: Callable[..., R],
        *args: Any,
        **kwargs: Any,
    ) -> R:
@@ -618,7 +615,7 @@ class DatabasePool:
    async def runInteraction(
        self,
        desc: str,
-        func: "Callable[..., R]",
+        func: Callable[..., R],
        *args: Any,
        db_autocommit: bool = False,
        **kwargs: Any,
@@ -678,7 +675,7 @@ class DatabasePool:

    async def runWithConnection(
        self,
-        func: "Callable[..., R]",
+        func: Callable[..., R],
        *args: Any,
        db_autocommit: bool = False,
        **kwargs: Any,
@@ -718,7 +715,9 @@ class DatabasePool:
            # pool).
            assert not self.engine.in_transaction(conn)

-            with LoggingContext("runWithConnection", parent_context) as context:
+            with LoggingContext(
+                str(curr_context), parent_context=parent_context
+            ) as context:
                sched_duration_sec = monotonic_time() - start_time
                sql_scheduling_timer.observe(sched_duration_sec)
                context.add_database_scheduled(sched_duration_sec)
--- a/synapse/storage/databases/main/censor_events.py
+++ b/synapse/storage/databases/main/censor_events.py
@@ -181,7 +181,7 @@ class CensorEventsStore(EventsWorkerStore, CacheInvalidationWorkerStore, SQLBase
            # changed its content in the database. We can't call
            # self._invalidate_cache_and_stream because self.get_event_cache isn't of the
            # right type.
-            txn.call_after(self._get_event_cache.invalidate, (event.event_id,))
+            txn.call_after(self._invalidate_get_event_cache, event.event_id)
            # Send that invalidation to replication so that other workers also invalidate
            # the event cache.
            self._send_invalidation_to_replication(
--- a/synapse/storage/databases/main/events_worker.py
+++ b/synapse/storage/databases/main/events_worker.py
@@ -14,7 +14,6 @@

 import logging
 import threading
-from collections import namedtuple
 from typing import (
    Collection,
    Container,
@@ -25,7 +24,9 @@ from typing import (
    Tuple,
    overload,
 )
+from weakref import WeakValueDictionary

+import attr
 from constantly import NamedConstant, Names
 from typing_extensions import Literal

@@ -73,7 +74,10 @@ EVENT_QUEUE_ITERATIONS = 3  # No. times we block waiting for requests for events
 EVENT_QUEUE_TIMEOUT_S = 0.1  # Timeout when waiting for requests for events


-_EventCacheEntry = namedtuple("_EventCacheEntry", ("event", "redacted_event"))
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class _EventCacheEntry:
+    event: EventBase
+    redacted_event: Optional[EventBase]


 class EventRedactBehaviour(Names):
@@ -157,9 +161,14 @@ class EventsWorkerStore(SQLBaseStore):

        self._get_event_cache = LruCache(
            cache_name="*getEvent*",
-            keylen=3,
            max_size=hs.config.caches.event_cache_size,
        )
+        # We seperately track which events we have in memory. This is mainly to
+        # guard against loading the same event into memory multiple times when
+        # `_get_event_cache` overflows.
+        self._in_memory_events = (
+            WeakValueDictionary()
+        )  # type: WeakValueDictionary[str, _EventCacheEntry]

        self._event_fetch_lock = threading.Condition()
        self._event_fetch_list = []
@@ -519,6 +528,7 @@ class EventsWorkerStore(SQLBaseStore):

    def _invalidate_get_event_cache(self, event_id):
        self._get_event_cache.invalidate((event_id,))
+        self._in_memory_events.pop(event_id, None)

    def _get_events_from_cache(self, events, allow_rejected, update_metrics=True):
        """Fetch events from the caches
@@ -539,6 +549,9 @@ class EventsWorkerStore(SQLBaseStore):
            ret = self._get_event_cache.get(
                (event_id,), None, update_metrics=update_metrics
            )
+            if not ret:
+                ret = self._in_memory_events.get(event_id)
+
            if not ret:
                continue

@@ -708,6 +721,9 @@ class EventsWorkerStore(SQLBaseStore):
            if events_to_fetch:
                logger.debug("Also fetching redaction events %s", events_to_fetch)

+        # The events to return
+        result_map = {}  # type: Dict[str, _EventCacheEntry]
+
        # build a map from event_id to EventBase
        event_map = {}
        for event_id, row in fetched_events.items():
@@ -720,6 +736,18 @@ class EventsWorkerStore(SQLBaseStore):
            if not allow_rejected and rejected_reason:
                continue

+            # Check whether we already have this event in memory. This can
+            # happen multiple requests for the same event happen at the same
+            # time. (Ideally we'd have make it so that this doesn't happen, but
+            # that would require a larger refactor).
+            cached_entry = self._in_memory_events.get(event_id)
+            if cached_entry is not None:
+                # We need to add to the event_map as we read from it to fetch redactions.
+                event_map[event_id] = cached_entry.event
+                result_map[event_id] = cached_entry
+                self._get_event_cache.set((event_id,), cached_entry)
+                continue
+
            # If the event or metadata cannot be parsed, log the error and act
            # as if the event is unknown.
            try:
@@ -813,8 +841,10 @@ class EventsWorkerStore(SQLBaseStore):

        # finally, we can decide whether each one needs redacting, and build
        # the cache entries.
-        result_map = {}
        for event_id, original_ev in event_map.items():
+            if event_id in result_map:
+                continue
+
            redactions = fetched_events[event_id]["redactions"]
            redacted_event = self._maybe_redact_event_row(
                original_ev, redactions, event_map
@@ -825,6 +855,7 @@ class EventsWorkerStore(SQLBaseStore):
            )

            self._get_event_cache.set((event_id,), cache_entry)
+            self._in_memory_events[event_id] = cache_entry
            result_map[event_id] = cache_entry

        return result_map
@@ -1056,7 +1087,11 @@ class EventsWorkerStore(SQLBaseStore):
            set[str]: The events we have already seen.
        """
        # if the event cache contains the event, obviously we've seen it.
-        results = {x for x in event_ids if self._get_event_cache.contains(x)}
+        results = {
+            x
+            for x in event_ids
+            if self._get_event_cache.contains((x,)) or x in self._in_memory_events
+        }

        def have_seen_events_txn(txn, chunk):
            sql = "SELECT event_id FROM events as e WHERE "
--- a/synapse/storage/databases/main/roommember.py
+++ b/synapse/storage/databases/main/roommember.py
@@ -205,8 +205,12 @@ class RoomMemberWorkerStore(EventsWorkerStore):

        def _get_users_in_room_with_profiles(txn) -> Dict[str, ProfileInfo]:
            sql = """
-                SELECT user_id, display_name, avatar_url FROM room_memberships
-                WHERE room_id = ? AND membership = ?
+                SELECT state_key, display_name, avatar_url FROM room_memberships as m
+                INNER JOIN current_state_events as c
+                ON m.event_id = c.event_id
+                AND m.room_id = c.room_id
+                AND m.user_id = c.state_key
+                WHERE c.type = 'm.room.member' AND c.room_id = ? AND m.membership = ?
            """
            txn.execute(sql, (room_id, Membership.JOIN))

--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
Erik Johnston	e5c6aafe61	Fix redactions	2021-05-10 15:48:27 +01:00
Erik Johnston	c0a7348580	Handle deduplicating multiple fetch event requests	2021-05-10 14:49:39 +01:00
Erik Johnston	875a8fec34	Newsfile	2021-05-06 14:45:20 +01:00
Erik Johnston	9d1118dde8	Ensure we only have one copy of an event in memory at a time This ensures that if the get event cache overflows we don't end up with multiple copies of the event in RAM at the same time (which could lead to memory bloat)	2021-05-06 14:42:42 +01:00
Erik Johnston	a8803e2b6e	Merge remote-tracking branch 'origin/master' into develop	2021-05-06 14:36:59 +01:00
Erik Johnston	ac88aca7f7	1.33.1	2021-05-06 14:06:38 +01:00
Erik Johnston	24f07a83e6	Pin attrs to <21.1.0 (#9937 ) Fixes #9936	2021-05-06 14:06:06 +01:00
Patrick Cloke	70f0ffd2fc	Follow-up to #9915 to correct the identifier for room types.	2021-05-05 16:31:16 -04:00
Patrick Cloke	d783880083	Include the time of the create event in Spaces Summary. (#9928 ) This is an update based on changes to MSC2946. The origin_server_ts of the m.room.create event is copied into the creation_ts field for each room returned from the spaces summary.	2021-05-05 13:33:05 -04:00
Erik Johnston	37623e3382	Increase perf of handling presence when joining large rooms. (#9916 )	2021-05-05 17:27:05 +01:00
Patrick Cloke	e2a443550e	Support stable MSC1772 spaces identifiers. (#9915 ) Support both the unstable and stable identifiers. A future release will disable the unstable identifiers.	2021-05-05 11:56:51 -04:00
Erik Johnston	ef889c98a6	Optionally track memory usage of each LruCache (#9881 ) This will double count slightly in the presence of interned strings. It's off by default as it can consume a lot of resources.	2021-05-05 16:54:36 +01:00
Erik Johnston	1fb9a2d0bf	Limit how often GC happens by time. (#9902 ) Synapse can be quite memory intensive, and unless care is taken to tune the GC thresholds it can end up thrashing, causing noticable performance problems for large servers. We fix this by limiting how often we GC a given generation, regardless of current counts/thresholds. This does not help with the reverse problem where the thresholds are set too high, but that should only happen in situations where they've been manually configured. Adds a `gc_min_seconds_between` config option to override the defaults. Fixes #9890.	2021-05-05 16:53:45 +01:00
Erik Johnston	de8f0a03a3	Don't set the external cache if its been done recently (#9905 )	2021-05-05 16:53:22 +01:00
Erik Johnston	d0aee697ac	Use get_current_users_in_room from store and not StateHandler (#9910 )	2021-05-05 16:49:34 +01:00
Christopher May-Townsend	d5305000f1	Docker healthcheck timings - add startup delay and changed interval (#9913 ) * Add healthcheck startup delay by 5secs and reduced interval check to 15s to reduce waiting time for docker aware edge routers bringing an instance online	2021-05-05 17:33:04 +02:00
DeepBlueV7.X	e9eb3549d3	Leave out optional keys from /sync (#9919 ) This leaves out all optional keys from /sync. This should be fine for all clients tested against conduit already, but it may break some clients, as such we should check, that at least most of them don't break horribly and maybe back out some of the individual changes. (We can probably always leave out groups for example, while the others may cause more issues.) Signed-off-by: Nicolas Werner <nicolas.werner@hotmail.de>	2021-05-05 14:37:56 +01:00
Brendan Abolivier	a61b13c0a1	Merge branch 'master' into develop	2021-05-05 14:25:49 +01:00
Brendan Abolivier	0644ac0989	1.33.0	2021-05-05 14:15:54 +01:00
Erik Johnston	e3bc4617fc	Time external cache response time (#9904 )	2021-05-04 15:14:22 +01:00
Andrew Morgan	b85821aca2	Add port parameter to the sample config for psycopg2 args (#9911 ) Adds the `port` option with the default value to the sample config file.	2021-05-04 13:28:59 +01:00
Dan Callahan	56c4b47df3	Build Debian packages for Ubuntu 21.04 Hirsute (#9909 ) Signed-off-by: Dan Callahan <danc@element.io>	2021-04-30 15:36:05 +01:00
Andrew Morgan	4d624f467a	Merge tag 'v1.33.0rc2' into develop Synapse 1.33.0rc2 (2021-04-29) ============================== Bugfixes -------- - Fix tight loop when handling presence replication when using workers. Introduced in v1.33.0rc1. ([\#9900](https://github.com/matrix-org/synapse/issues/9900))	2021-04-29 14:35:14 +01:00
Andrew Morgan	d11f2dfee5	typo in changelog	2021-04-29 14:31:14 +01:00
Patrick Cloke	bb4b11846f	Add missing type hints to handlers and fix a Spam Checker type hint. (#9896 ) The user_may_create_room_alias method on spam checkers declared the room_alias parameter as a str when in reality it is passed a RoomAlias object.	2021-04-29 07:17:28 -04:00
Andrew Morgan	e9444cc74d	1.33.0rc2	2021-04-29 11:45:37 +01:00
ThibF	0085dc5abc	Delete room endpoint (#9889 ) Support the delete of a room through DELETE request and mark previous request as deprecated through documentation. Signed-off-by: Thibault Ferrante <thibault.ferrante@pm.me>	2021-04-29 10:31:45 +01:00
Erik Johnston	802560211a	Merge remote-tracking branch 'origin/release-v1.33.0' into develop	2021-04-28 14:43:10 +01:00
Erik Johnston	e4ab8676b4	Fix tight loop handling presence replication. (#9900 ) Only affects workers. Introduced in #9819. Fixes #9899.	2021-04-28 14:42:50 +01:00
Patrick Cloke	10a08ab88a	Use the parent's logging context name for runWithConnection. (#9895 ) This fixes a regression where the logging context for runWithConnection was reported as runWithConnection instead of the connection name, e.g. "POST-XYZ".	2021-04-28 07:44:52 -04:00
Andrew Morgan	fa6679e794	Merge tag 'v1.33.0rc1' into develop Synapse 1.33.0rc1 (2021-04-28) ============================== Features -------- - Update experimental support for [MSC3083](https://github.com/matrix-org/matrix-doc/pull/3083): restricting room access via group membership. ([\#9800](https://github.com/matrix-org/synapse/issues/9800), [\#9814](https://github.com/matrix-org/synapse/issues/9814)) - Add experimental support for handling presence on a worker. ([\#9819](https://github.com/matrix-org/synapse/issues/9819), [\#9820](https://github.com/matrix-org/synapse/issues/9820), [\#9828](https://github.com/matrix-org/synapse/issues/9828), [\#9850](https://github.com/matrix-org/synapse/issues/9850)) - Return a new template when an user attempts to renew their account multiple times with the same token, stating that their account is set to expire. This replaces the invalid token template that would previously be shown in this case. This change concerns the optional account validity feature. ([\#9832](https://github.com/matrix-org/synapse/issues/9832)) Bugfixes -------- - Fixes the OIDC SSO flow when using a `public_baseurl` value including a non-root URL path. ([\#9726](https://github.com/matrix-org/synapse/issues/9726)) - Fix thumbnail generation for some sites with non-standard content types. Contributed by @rkfg. ([\#9788](https://github.com/matrix-org/synapse/issues/9788)) - Add some sanity checks to identity server passed to 3PID bind/unbind endpoints. ([\#9802](https://github.com/matrix-org/synapse/issues/9802)) - Limit the size of HTTP responses read over federation. ([\#9833](https://github.com/matrix-org/synapse/issues/9833)) - Fix a bug which could cause Synapse to get stuck in a loop of resyncing device lists. ([\#9867](https://github.com/matrix-org/synapse/issues/9867)) - Fix a long-standing bug where errors from federation did not propagate to the client. ([\#9868](https://github.com/matrix-org/synapse/issues/9868)) Improved Documentation ---------------------- - Add a note to the docker docs mentioning that we mirror upstream's supported Docker platforms. ([\#9801](https://github.com/matrix-org/synapse/issues/9801)) Internal Changes ---------------- - Add a dockerfile for running Synapse in worker-mode under Complement. ([\#9162](https://github.com/matrix-org/synapse/issues/9162)) - Apply `pyupgrade` across the codebase. ([\#9786](https://github.com/matrix-org/synapse/issues/9786)) - Move some replication processing out of `generic_worker`. ([\#9796](https://github.com/matrix-org/synapse/issues/9796)) - Replace `HomeServer.get_config()` with inline references. ([\#9815](https://github.com/matrix-org/synapse/issues/9815)) - Rename some handlers and config modules to not duplicate the top-level module. ([\#9816](https://github.com/matrix-org/synapse/issues/9816)) - Fix a long-standing bug which caused `max_upload_size` to not be correctly enforced. ([\#9817](https://github.com/matrix-org/synapse/issues/9817)) - Reduce CPU usage of the user directory by reusing existing calculated room membership. ([\#9821](https://github.com/matrix-org/synapse/issues/9821)) - Small speed up for joining large remote rooms. ([\#9825](https://github.com/matrix-org/synapse/issues/9825)) - Introduce flake8-bugbear to the test suite and fix some of its lint violations. ([\#9838](https://github.com/matrix-org/synapse/issues/9838)) - Only store the raw data in the in-memory caches, rather than objects that include references to e.g. the data stores. ([\#9845](https://github.com/matrix-org/synapse/issues/9845)) - Limit length of accepted email addresses. ([\#9855](https://github.com/matrix-org/synapse/issues/9855)) - Remove redundant `synapse.types.Collection` type definition. ([\#9856](https://github.com/matrix-org/synapse/issues/9856)) - Handle recently added rate limits correctly when using `--no-rate-limit` with the demo scripts. ([\#9858](https://github.com/matrix-org/synapse/issues/9858)) - Disable invite rate-limiting by default when running the unit tests. ([\#9871](https://github.com/matrix-org/synapse/issues/9871)) - Pass a reactor into `SynapseSite` to make testing easier. ([\#9874](https://github.com/matrix-org/synapse/issues/9874)) - Make `DomainSpecificString` an `attrs` class. ([\#9875](https://github.com/matrix-org/synapse/issues/9875)) - Add type hints to `synapse.api.auth` and `synapse.api.auth_blocking` modules. ([\#9876](https://github.com/matrix-org/synapse/issues/9876)) - Remove redundant `_PushHTTPChannel` test class. ([\#9878](https://github.com/matrix-org/synapse/issues/9878)) - Remove backwards-compatibility code for Python versions < 3.6. ([\#9879](https://github.com/matrix-org/synapse/issues/9879)) - Small performance improvement around handling new local presence updates. ([\#9887](https://github.com/matrix-org/synapse/issues/9887))	2021-04-28 12:12:29 +01:00
Andrew Morgan	8ba086980d	Reword account validity template change to sound less like a bugfix	2021-04-28 12:07:49 +01:00
Erik Johnston	391bfe9a7b	Reduce memory footprint of caches (#9886 )	2021-04-28 11:59:28 +01:00
Andrew Morgan	787de3190f	1.33.0rc1	2021-04-28 11:43:33 +01:00
Andrew Morgan	4e0fd35bc9	Revert "Experimental Federation Speedup (#9702 )" This reverts commit `05e8c70c05`.	2021-04-28 11:38:33 +01:00
Erik Johnston	dd2d32dcdb	Add type hints to presence handler (#9885 )	2021-04-28 11:07:47 +01:00
Andrew Morgan	fe604a022a	Remove various bits of compatibility code for Python <3.6 (#9879 ) I went through and removed a bunch of cruft that was lying around for compatibility with old Python versions. This PR also will now prevent Synapse from starting unless you're running Python 3.6+.	2021-04-27 13:13:07 +01:00
Patrick Cloke	1350b053da	Pass errors back to the client when trying multiple federation destinations. (#9868 ) This ensures that something like an auth error (403) will be returned to the requester instead of attempting to try more servers, which will likely result in the same error, and then passing back a generic 400 error.	2021-04-27 07:30:34 -04:00
				`@@ -1 +0,0 @@`
				`Add a dockerfile for running Synapse in worker-mode under Complement.`
				`@@ -1 +0,0 @@`
				`Speed up federation transmission by using fewer database calls. Contributed by @ShadowJonathan.`
				`@@ -1 +0,0 @@`
				Fixes the OIDC SSO flow when using a `public_baseurl` value including a non-root URL path.
				`@@ -1 +0,0 @@`
				`Fix thumbnail generation for some sites with non-standard content types. Contributed by @rkfg.`
				`@@ -1 +0,0 @@`
				Move some replication processing out of `generic_worker`.
				`@@ -1 +0,0 @@`
				`Update experimental support for [MSC3083](https://github.com/matrix-org/matrix-doc/pull/3083): restricting room access via group membership.`
				`@@ -1 +0,0 @@`
				`Add a note to the docker docs mentioning that we mirror upstream's supported Docker platforms.`
				`@@ -1 +0,0 @@`
				`Add some sanity checks to identity server passed to 3PID bind/unbind endpoints.`
				`@@ -1 +0,0 @@`
				Replace `HomeServer.get_config()` with inline references.