FIXUP

This makes it so we can always `upsert` to avoid duplicates otherwise
I'm not sure of how to not insert duplicates in certain situations
(see FIXME in the diff) which would cause problems down the line
for the unique index being added later.

.github/workflows/docker.yml

@@ -30,7 +30,7 @@ jobs:
         run: docker buildx inspect
 
       - name: Install Cosign
-        uses: sigstore/cosign-installer@v3.5.0
+        uses: sigstore/cosign-installer@v3.6.0
 
       - name: Checkout repository
         uses: actions/checkout@v4

CHANGES.md

@@ -1,3 +1,212 @@
+# Synapse 1.114.0rc1 (2024-08-20)
+
+### Features
+
+- Add a flag to `/versions`, `org.matrix.simplified_msc3575`, to indicate whether experimental sliding sync support has been enabled. ([\#17571](https://github.com/element-hq/synapse/issues/17571))
+- Handle changes in `timeline_limit` in experimental sliding sync. ([\#17579](https://github.com/element-hq/synapse/issues/17579))
+- Correctly track read receipts that should be sent down in experimental sliding sync. ([\#17575](https://github.com/element-hq/synapse/issues/17575), [\#17589](https://github.com/element-hq/synapse/issues/17589), [\#17592](https://github.com/element-hq/synapse/issues/17592))
+
+### Bugfixes
+
+- Start handlers for new media endpoints when media resource configured. ([\#17483](https://github.com/element-hq/synapse/issues/17483))
+- Fix timeline ordering (using `stream_ordering` instead of topological ordering) in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17510](https://github.com/element-hq/synapse/issues/17510))
+- Fix experimental sliding sync implementation to remember any updates in rooms that were not sent down immediately. ([\#17535](https://github.com/element-hq/synapse/issues/17535))
+- Better exclude partially stated rooms if we must await full state in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17538](https://github.com/element-hq/synapse/issues/17538))
+- Handle lower-case http headers in `_Mulitpart_Parser_Protocol`. ([\#17545](https://github.com/element-hq/synapse/issues/17545))
+- Fix fetching federation signing keys from servers that omit `old_verify_keys`. Contributed by @tulir @ Beeper. ([\#17568](https://github.com/element-hq/synapse/issues/17568))
+- Fix bug where we would respond with an error when a remote server asked for media that had a length of 0, using the new multipart federation media endpoint. ([\#17570](https://github.com/element-hq/synapse/issues/17570))
+
+### Improved Documentation
+
+- Clarify default behaviour of the
+  [`auto_accept_invites.worker_to_run_on`](https://element-hq.github.io/synapse/develop/usage/configuration/config_documentation.html#auto-accept-invites)
+  option. ([\#17515](https://github.com/element-hq/synapse/issues/17515))
+- Improve docstrings for profile methods. ([\#17559](https://github.com/element-hq/synapse/issues/17559))
+
+### Internal Changes
+
+- Add more tracing to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17514](https://github.com/element-hq/synapse/issues/17514))
+- Fixup comment in sliding sync implementation. ([\#17531](https://github.com/element-hq/synapse/issues/17531))
+- Replace override of deprecated method `HTTPAdapter.get_connection` with `get_connection_with_tls_context`. ([\#17536](https://github.com/element-hq/synapse/issues/17536))
+- Fix performance of device lists in `/key/changes` and sliding sync. ([\#17537](https://github.com/element-hq/synapse/issues/17537), [\#17548](https://github.com/element-hq/synapse/issues/17548))
+- Bump setuptools from 67.6.0 to 72.1.0. ([\#17542](https://github.com/element-hq/synapse/issues/17542))
+- Add a utility function for generating random event IDs. ([\#17557](https://github.com/element-hq/synapse/issues/17557))
+- Speed up responding to media requests. ([\#17558](https://github.com/element-hq/synapse/issues/17558), [\#17561](https://github.com/element-hq/synapse/issues/17561), [\#17564](https://github.com/element-hq/synapse/issues/17564), [\#17566](https://github.com/element-hq/synapse/issues/17566), [\#17567](https://github.com/element-hq/synapse/issues/17567), [\#17569](https://github.com/element-hq/synapse/issues/17569))
+- Test github token before running release script steps. ([\#17562](https://github.com/element-hq/synapse/issues/17562))
+- Reduce log spam of multipart files. ([\#17563](https://github.com/element-hq/synapse/issues/17563))
+- Refactor per-connection state in experimental sliding sync handler. ([\#17574](https://github.com/element-hq/synapse/issues/17574))
+- Add histogram metrics for sliding sync processing time. ([\#17593](https://github.com/element-hq/synapse/issues/17593))
+
+
+
+### Updates to locked dependencies
+
+* Bump bytes from 1.6.1 to 1.7.1. ([\#17526](https://github.com/element-hq/synapse/issues/17526))
+* Bump lxml from 5.2.2 to 5.3.0. ([\#17550](https://github.com/element-hq/synapse/issues/17550))
+* Bump phonenumbers from 8.13.42 to 8.13.43. ([\#17551](https://github.com/element-hq/synapse/issues/17551))
+* Bump regex from 1.10.5 to 1.10.6. ([\#17527](https://github.com/element-hq/synapse/issues/17527))
+* Bump sentry-sdk from 2.10.0 to 2.12.0. ([\#17553](https://github.com/element-hq/synapse/issues/17553))
+* Bump serde from 1.0.204 to 1.0.206. ([\#17556](https://github.com/element-hq/synapse/issues/17556))
+* Bump serde_json from 1.0.122 to 1.0.124. ([\#17555](https://github.com/element-hq/synapse/issues/17555))
+* Bump sigstore/cosign-installer from 3.5.0 to 3.6.0. ([\#17549](https://github.com/element-hq/synapse/issues/17549))
+* Bump types-pyyaml from 6.0.12.20240311 to 6.0.12.20240808. ([\#17552](https://github.com/element-hq/synapse/issues/17552))
+* Bump types-requests from 2.31.0.20240406 to 2.32.0.20240712. ([\#17524](https://github.com/element-hq/synapse/issues/17524))
+
+# Synapse 1.113.0 (2024-08-13)
+
+No significant changes since 1.113.0rc1.
+
+
+
+
+# Synapse 1.113.0rc1 (2024-08-06)
+
+### Features
+
+- Track which rooms have been sent to clients in the experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17447](https://github.com/element-hq/synapse/issues/17447))
+- Add Account Data extension support to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17477](https://github.com/element-hq/synapse/issues/17477))
+- Add receipts extension support to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17489](https://github.com/element-hq/synapse/issues/17489))
+- Add typing notification extension support to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17505](https://github.com/element-hq/synapse/issues/17505))
+
+### Bugfixes
+
+- Update experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint to handle invite/knock rooms when filtering. ([\#17450](https://github.com/element-hq/synapse/issues/17450))
+- Fix a bug introduced in v1.110.0 which caused `/keys/query` to return incomplete results, leading to high network activity and CPU usage on Matrix clients. ([\#17499](https://github.com/element-hq/synapse/issues/17499))
+
+### Improved Documentation
+
+- Update the [`allowed_local_3pids`](https://element-hq.github.io/synapse/v1.112/usage/configuration/config_documentation.html#allowed_local_3pids) config option's msisdn address to a working example. ([\#17476](https://github.com/element-hq/synapse/issues/17476))
+
+### Internal Changes
+
+- Change sliding sync to use their own token format in preparation for storing per-connection state. ([\#17452](https://github.com/element-hq/synapse/issues/17452))
+- Ensure we don't send down negative `bump_stamp` in experimental sliding sync endpoint. ([\#17478](https://github.com/element-hq/synapse/issues/17478))
+- Do not send down empty room entries down experimental sliding sync endpoint. ([\#17479](https://github.com/element-hq/synapse/issues/17479))
+- Refactor Sliding Sync tests to better utilize the `SlidingSyncBase`. ([\#17481](https://github.com/element-hq/synapse/issues/17481), [\#17482](https://github.com/element-hq/synapse/issues/17482))
+- Add some opentracing tags and logging to the experimental sliding sync implementation. ([\#17501](https://github.com/element-hq/synapse/issues/17501))
+- Split and move Sliding Sync tests so we have some more sane test file sizes. ([\#17504](https://github.com/element-hq/synapse/issues/17504))
+- Update the `limited` field description in the Sliding Sync response to accurately describe what it actually represents. ([\#17507](https://github.com/element-hq/synapse/issues/17507))
+- Easier to understand `timeline` assertions in Sliding Sync tests. ([\#17511](https://github.com/element-hq/synapse/issues/17511))
+- Reset the sliding sync connection if we don't recognize the per-connection state position. ([\#17529](https://github.com/element-hq/synapse/issues/17529))
+
+
+
+### Updates to locked dependencies
+
+* Bump bcrypt from 4.1.3 to 4.2.0. ([\#17495](https://github.com/element-hq/synapse/issues/17495))
+* Bump black from 24.4.2 to 24.8.0. ([\#17522](https://github.com/element-hq/synapse/issues/17522))
+* Bump phonenumbers from 8.13.39 to 8.13.42. ([\#17521](https://github.com/element-hq/synapse/issues/17521))
+* Bump ruff from 0.5.4 to 0.5.5. ([\#17494](https://github.com/element-hq/synapse/issues/17494))
+* Bump serde_json from 1.0.120 to 1.0.121. ([\#17493](https://github.com/element-hq/synapse/issues/17493))
+* Bump serde_json from 1.0.121 to 1.0.122. ([\#17525](https://github.com/element-hq/synapse/issues/17525))
+* Bump towncrier from 23.11.0 to 24.7.1. ([\#17523](https://github.com/element-hq/synapse/issues/17523))
+* Bump types-pyopenssl from 24.1.0.20240425 to 24.1.0.20240722. ([\#17496](https://github.com/element-hq/synapse/issues/17496))
+* Bump types-setuptools from 70.1.0.20240627 to 71.1.0.20240726. ([\#17497](https://github.com/element-hq/synapse/issues/17497))
+
+# Synapse 1.112.0 (2024-07-30)
+
+This security release is to update our locked dependency on Twisted to 24.7.0rc1, which includes a security fix for [CVE-2024-41671 / GHSA-c8m8-j448-xjx7: Disordered HTTP pipeline response in twisted.web, again](https://github.com/twisted/twisted/security/advisories/GHSA-c8m8-j448-xjx7).
+
+Note that this security fix is also available as **Synapse 1.111.1**, which does not include the rest of the changes in Synapse 1.112.0.
+
+This issue means that, if multiple HTTP requests are pipelined in the same TCP connection, Synapse can send responses to the wrong HTTP request.
+If a reverse proxy was configured to use HTTP pipelining, this could result in responses being sent to the wrong user, severely harming confidentiality.
+
+With that said, despite being a high severity issue, **we consider it unlikely that Synapse installations will be affected**.
+The use of HTTP pipelining in this fashion would cause worse performance for clients (request-response latencies would be increased as users' responses would be artificially blocked behind other users' slow requests). Further, Nginx and Haproxy, two common reverse proxies, do not appear to support configuring their upstreams to use HTTP pipelining and thus would not be affected. For both of these reasons, we consider it unlikely that a Synapse deployment would be set up in such a configuration.
+
+Despite that, we cannot rule out that some installations may exist with this unusual setup and so we are releasing this security update today.
+
+**pip users:** Note that by default, upgrading Synapse using pip will not automatically upgrade Twisted. **Please manually install the new version of Twisted** using `pip install Twisted==24.7.0rc1`. Note also that even the `--upgrade-strategy=eager` flag to `pip install -U matrix-synapse` will not upgrade Twisted to a patched version because it is only a release candidate at this time.
+
+### Internal Changes
+
+- Upgrade locked dependency on Twisted to 24.7.0rc1. ([\#17502](https://github.com/element-hq/synapse/issues/17502))
+
+
+# Synapse 1.112.0rc1 (2024-07-23)
+
+Please note that this release candidate does not include the security dependency update
+included in version 1.111.1 as this version was released before 1.111.1.
+The same security fix can be found in the full release of 1.112.0.
+
+### Features
+
+- Add to-device extension support to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17416](https://github.com/element-hq/synapse/issues/17416))
+- Populate `name`/`avatar` fields in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17418](https://github.com/element-hq/synapse/issues/17418))
+- Populate `heroes` and room summary fields (`joined_count`, `invited_count`) in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17419](https://github.com/element-hq/synapse/issues/17419))
+- Populate `is_dm` room field in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17429](https://github.com/element-hq/synapse/issues/17429))
+- Add room subscriptions to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17432](https://github.com/element-hq/synapse/issues/17432))
+- Prepare for authenticated media freeze. ([\#17433](https://github.com/element-hq/synapse/issues/17433))
+- Add E2EE extension support to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint. ([\#17454](https://github.com/element-hq/synapse/issues/17454))
+
+### Bugfixes
+
+- Add configurable option to always include offline users in presence sync results. Contributed by @Michael-Hollister. ([\#17231](https://github.com/element-hq/synapse/issues/17231))
+- Fix bug in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint when using room type filters and the user has one or more remote invites. ([\#17434](https://github.com/element-hq/synapse/issues/17434))
+- Order `heroes` by `stream_ordering` as the Matrix specification states (applies to `/sync`). ([\#17435](https://github.com/element-hq/synapse/issues/17435))
+- Fix rare bug where `/sync` would break for a user when using workers with multiple stream writers. ([\#17438](https://github.com/element-hq/synapse/issues/17438))
+
+### Improved Documentation
+
+- Update the readme image to have a white background, so that it is readable in dark mode. ([\#17387](https://github.com/element-hq/synapse/issues/17387))
+- Add Red Hat Enterprise Linux and Rocky Linux 8 and 9 installation instructions. ([\#17423](https://github.com/element-hq/synapse/issues/17423))
+- Improve documentation for the [`default_power_level_content_override`](https://element-hq.github.io/synapse/latest/usage/configuration/config_documentation.html#default_power_level_content_override) config option. ([\#17451](https://github.com/element-hq/synapse/issues/17451))
+
+### Internal Changes
+
+- Make sure we always use the right logic for enabling the media repo. ([\#17424](https://github.com/element-hq/synapse/issues/17424))
+- Fix argument documentation for method `RateLimiter.record_action`. ([\#17426](https://github.com/element-hq/synapse/issues/17426))
+- Reduce volume of 'Waiting for current token' logs, which were introduced in v1.109.0. ([\#17428](https://github.com/element-hq/synapse/issues/17428))
+- Limit concurrent remote downloads to 6 per IP address, and decrement remote downloads without a content-length from the ratelimiter after the download is complete. ([\#17439](https://github.com/element-hq/synapse/issues/17439))
+- Remove unnecessary call to resume producing in fake channel. ([\#17449](https://github.com/element-hq/synapse/issues/17449))
+- Update experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint to bump room when it is created. ([\#17453](https://github.com/element-hq/synapse/issues/17453))
+- Speed up generating sliding sync responses. ([\#17458](https://github.com/element-hq/synapse/issues/17458))
+- Add cache to `get_rooms_for_local_user_where_membership_is` to speed up sliding sync. ([\#17460](https://github.com/element-hq/synapse/issues/17460))
+- Speed up fetching room keys from backup. ([\#17461](https://github.com/element-hq/synapse/issues/17461))
+- Speed up sorting of the room list in sliding sync. ([\#17468](https://github.com/element-hq/synapse/issues/17468))
+- Implement handling of `$ME` as a state key in sliding sync. ([\#17469](https://github.com/element-hq/synapse/issues/17469))
+
+
+
+### Updates to locked dependencies
+
+* Bump bytes from 1.6.0 to 1.6.1. ([\#17441](https://github.com/element-hq/synapse/issues/17441))
+* Bump hiredis from 2.3.2 to 3.0.0. ([\#17464](https://github.com/element-hq/synapse/issues/17464))
+* Bump jsonschema from 4.22.0 to 4.23.0. ([\#17444](https://github.com/element-hq/synapse/issues/17444))
+* Bump matrix-org/done-action from 2 to 3. ([\#17440](https://github.com/element-hq/synapse/issues/17440))
+* Bump mypy from 1.9.0 to 1.10.1. ([\#17445](https://github.com/element-hq/synapse/issues/17445))
+* Bump pyopenssl from 24.1.0 to 24.2.1. ([\#17465](https://github.com/element-hq/synapse/issues/17465))
+* Bump ruff from 0.5.0 to 0.5.4. ([\#17466](https://github.com/element-hq/synapse/issues/17466))
+* Bump sentry-sdk from 2.6.0 to 2.8.0. ([\#17456](https://github.com/element-hq/synapse/issues/17456))
+* Bump sentry-sdk from 2.8.0 to 2.10.0. ([\#17467](https://github.com/element-hq/synapse/issues/17467))
+* Bump setuptools from 67.6.0 to 70.0.0. ([\#17448](https://github.com/element-hq/synapse/issues/17448))
+* Bump twine from 5.1.0 to 5.1.1. ([\#17443](https://github.com/element-hq/synapse/issues/17443))
+* Bump types-jsonschema from 4.22.0.20240610 to 4.23.0.20240712. ([\#17446](https://github.com/element-hq/synapse/issues/17446))
+* Bump ulid from 1.1.2 to 1.1.3. ([\#17442](https://github.com/element-hq/synapse/issues/17442))
+* Bump zipp from 3.15.0 to 3.19.1. ([\#17427](https://github.com/element-hq/synapse/issues/17427))
+
+
+# Synapse 1.111.1 (2024-07-30)
+
+This security release is to update our locked dependency on Twisted to 24.7.0rc1, which includes a security fix for [CVE-2024-41671 / GHSA-c8m8-j448-xjx7: Disordered HTTP pipeline response in twisted.web, again](https://github.com/twisted/twisted/security/advisories/GHSA-c8m8-j448-xjx7).
+
+This issue means that, if multiple HTTP requests are pipelined in the same TCP connection, Synapse can send responses to the wrong HTTP request.
+If a reverse proxy was configured to use HTTP pipelining, this could result in responses being sent to the wrong user, severely harming confidentiality.
+
+With that said, despite being a high severity issue, **we consider it unlikely that Synapse installations will be affected**.
+The use of HTTP pipelining in this fashion would cause worse performance for clients (request-response latencies would be increased as users' responses would be artificially blocked behind other users' slow requests). Further, Nginx and Haproxy, two common reverse proxies, do not appear to support configuring their upstreams to use HTTP pipelining and thus would not be affected. For both of these reasons, we consider it unlikely that a Synapse deployment would be set up in such a configuration.
+
+Despite that, we cannot rule out that some installations may exist with this unusual setup and so we are releasing this security update today.
+
+**pip users:** Note that by default, upgrading Synapse using pip will not automatically upgrade Twisted. **Please manually install the new version of Twisted** using `pip install Twisted==24.7.0rc1`. Note also that even the `--upgrade-strategy=eager` flag to `pip install -U matrix-synapse` will not upgrade Twisted to a patched version because it is only a release candidate at this time.
+
+
+### Internal Changes
+
+- Upgrade locked dependency on Twisted to 24.7.0rc1. ([\#17502](https://github.com/element-hq/synapse/issues/17502))
+
+
 # Synapse 1.111.0 (2024-07-16)
 
 No significant changes since 1.111.0rc2.

Cargo.lock

@@ -67,9 +67,9 @@ checksum = "79296716171880943b8470b5f8d03aa55eb2e645a4874bdbb28adb49162e012c"
 
 [[package]]
 name = "bytes"
-version = "1.6.1"
+version = "1.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a12916984aab3fa6e39d655a33e09c0071eb36d6ab3aea5c2d78551f1df6d952"
+checksum = "8318a53db07bb3f8dca91a600466bdb3f2eaadeedfdbcf02e1accbad9271ba50"
 
 [[package]]
 name = "cfg-if"
@@ -444,9 +444,9 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.10.5"
+version = "1.10.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b91213439dad192326a0d7c6ee3955910425f441d7038e0d6933b0aec5c4517f"
+checksum = "4219d74c6b67a3654a9fbebc4b419e22126d13d2f3c4a07ee0cb61ff79a79619"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -485,18 +485,18 @@ checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49"
 
 [[package]]
 name = "serde"
-version = "1.0.204"
+version = "1.0.208"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc76f558e0cbb2a839d37354c575f1dc3fdc6546b5be373ba43d95f231bf7c12"
+checksum = "cff085d2cb684faa248efb494c39b68e522822ac0de72ccf08109abde717cfb2"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.204"
+version = "1.0.208"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e0cd7e117be63d3c3678776753929474f3b04a43a080c744d6b0ae2a8c28e222"
+checksum = "24008e81ff7613ed8e5ba0cfaf24e2c2f1e5b8a0495711e44fcd4882fca62bcf"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -505,11 +505,12 @@ dependencies = [
 
 [[package]]
 name = "serde_json"
-version = "1.0.120"
+version = "1.0.125"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4e0d21c9a8cae1235ad58a00c11cb40d4b1e5c784f1ef2c537876ed6ffd8b7c5"
+checksum = "83c8e735a073ccf5be70aa8066aa984eaf2fa000db6c8d0100ae605b366d31ed"
 dependencies = [
  "itoa",
+ "memchr",
  "ryu",
  "serde",
 ]

changelog.d/17387.doc

@@ -1 +0,0 @@
-Update the readme image to have a white background, so that it is readable in dark mode.

changelog.d/17416.feature

@@ -1 +0,0 @@
-Add to-device extension support to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

changelog.d/17418.feature

@@ -1 +0,0 @@
-Populate `name`/`avatar` fields in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

changelog.d/17419.feature

@@ -1 +0,0 @@
-Populate `heroes` and room summary fields (`joined_count`, `invited_count`) in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

changelog.d/17423.doc

@@ -1 +0,0 @@
-Add Red Hat Enterprise Linux and Rocky Linux 8 and 9 installation instructions.

changelog.d/17424.misc

@@ -1 +0,0 @@
-Make sure we always use the right logic for enabling the media repo.

changelog.d/17426.misc

@@ -1 +0,0 @@
-Fix documentation on `RateLimiter#record_action`.

changelog.d/17429.feature

@@ -1 +0,0 @@
-Populate `is_dm` room field in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

changelog.d/17432.feature

@@ -1 +0,0 @@
-Add room subscriptions to experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint.

changelog.d/17433.feature

@@ -1 +0,0 @@
-Prepare for authenticated media freeze.

changelog.d/17434.bugfix

@@ -1 +0,0 @@
-Fix bug in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint when using room type filters and the user has one or more remote invites.

changelog.d/17435.bugfix

@@ -1 +0,0 @@
-Order `heroes` by `stream_ordering` as the Matrix specification states (applies to `/sync`).

changelog.d/17438.bugfix

@@ -1 +0,0 @@
-Fix rare bug where `/sync` would break for a user when using workers with multiple stream writers.

changelog.d/17439.bugfix

@@ -1 +0,0 @@
-Limit concurrent remote downloads to 6 per IP address, and decrement remote downloads without a content-length from the ratelimiter after the download is complete.

changelog.d/17449.bugfix

@@ -1 +0,0 @@
-Remove unnecessary call to resume producing in fake channel.

changelog.d/17451.doc

@@ -1 +0,0 @@
-Improve documentation for the [`default_power_level_content_override`](https://element-hq.github.io/synapse/latest/usage/configuration/config_documentation.html#default_power_level_content_override) config option.

changelog.d/17453.misc

@@ -1 +0,0 @@
-Update experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint to bump room when it is created.

changelog.d/17458.misc

@@ -1 +0,0 @@
-Speed up generating sliding sync responses.

changelog.d/17460.misc

@@ -1 +0,0 @@
-Add cache to `get_rooms_for_local_user_where_membership_is` to speed up sliding sync.

changelog.d/17512.misc

@@ -0,0 +1 @@
+Pre-populate room data used in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint for quick filtering/sorting.

changelog.d/17595.misc

@@ -0,0 +1 @@
+Refactor sliding sync class into multiple files.

changelog.d/17599.misc

@@ -0,0 +1 @@
+Store sliding sync per-connection state in the database.

changelog.d/17600.misc

@@ -0,0 +1 @@
+Make the sliding sync `PerConnectionState` class immutable.

debian/changelog

@@ -1,3 +1,39 @@
+matrix-synapse-py3 (1.114.0~rc1) stable; urgency=medium
+
+  * New synapse release 1.114.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 20 Aug 2024 12:55:28 +0000
+
+matrix-synapse-py3 (1.113.0) stable; urgency=medium
+
+  * New Synapse release 1.113.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 13 Aug 2024 14:36:56 +0100
+
+matrix-synapse-py3 (1.113.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.113.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 06 Aug 2024 12:23:23 +0100
+
+matrix-synapse-py3 (1.112.0) stable; urgency=medium
+
+  * New Synapse release 1.112.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 30 Jul 2024 17:15:48 +0100
+
+matrix-synapse-py3 (1.112.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.112.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 23 Jul 2024 08:58:55 -0600
+
+matrix-synapse-py3 (1.111.1) stable; urgency=medium
+
+  * New Synapse release 1.111.1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 30 Jul 2024 16:13:52 +0100
+
 matrix-synapse-py3 (1.111.0) stable; urgency=medium
 
   * New Synapse release 1.111.0.

debian/templates

@@ -5,7 +5,7 @@ _Description: Name of the server:
  servers via federation. This is normally the public hostname of the
  server running synapse, but can be different if you set up delegation.
  Please refer to the delegation documentation in this case:
- https://github.com/element-hq/synapse/blob/master/docs/delegate.md.
+ https://element-hq.github.io/synapse/latest/delegate.html.
 
 Template: matrix-synapse/report-stats
 Type: boolean

docs/development/room-dag-concepts.md

@@ -21,8 +21,10 @@ incrementing integer, but backfilled events start with `stream_ordering=-1` and
 
 ---
 
- - `/sync` returns things in the order they arrive at the server (`stream_ordering`).
- - `/messages` (and `/backfill` in the federation API) return them in the order determined by the event graph `(topological_ordering, stream_ordering)`.
+ - Incremental `/sync?since=xxx` returns things in the order they arrive at the server
+   (`stream_ordering`).
+ - Initial `/sync`, `/messages` (and `/backfill` in the federation API) return them in
+   the order determined by the event graph `(topological_ordering, stream_ordering)`.
 
 The general idea is that, if you're following a room in real-time (i.e.
 `/sync`), you probably want to see the messages as they arrive at your server,

docs/usage/configuration/config_documentation.md

@@ -246,6 +246,7 @@ Example configuration:
 ```yaml
 presence:
   enabled: false
+  include_offline_users_on_sync: false
 ```
 
 `enabled` can also be set to a special value of "untracked" which ignores updates
@@ -254,6 +255,10 @@ received via clients and federation, while still accepting updates from the
 
 *The "untracked" option was added in Synapse 1.96.0.*
 
+When clients perform an initial or `full_state` sync, presence results for offline users are
+not included by default. Setting `include_offline_users_on_sync` to `true` will always include
+offline users in the results. Defaults to false.
+
 ---
 ### `require_auth_for_profile_requests`
 
@@ -2381,7 +2386,7 @@ enable_registration_without_verification: true
 ---
 ### `registrations_require_3pid`
 
-If this is set, users must provide all of the specified types of 3PID when registering an account.
+If this is set, users must provide all of the specified types of [3PID](https://spec.matrix.org/latest/appendices/#3pid-types) when registering an account.
 
 Note that [`enable_registration`](#enable_registration) must also be set to allow account registration.
 
@@ -2406,6 +2411,9 @@ disable_msisdn_registration: true
 
 Mandate that users are only allowed to associate certain formats of
 3PIDs with accounts on this server, as specified by the `medium` and `pattern` sub-options.
+`pattern` is a [Perl-like regular expression](https://docs.python.org/3/library/re.html#module-re).
+
+More information about 3PIDs, allowed `medium` types and their `address` syntax can be found [in the Matrix spec](https://spec.matrix.org/latest/appendices/#3pid-types).
 
 Example configuration:
 ```yaml
@@ -2415,7 +2423,7 @@ allowed_local_3pids:
   - medium: email
     pattern: '^[^@]+@vector\.im$'
   - medium: msisdn
-    pattern: '\+44'
+    pattern: '^44\d{10}$'
 ```
 ---
 ### `enable_3pid_lookup`
@@ -4677,7 +4685,9 @@ This setting has the following sub-options:
 * `only_for_direct_messages`: Whether invites should be automatically accepted for all room types, or only
    for direct messages. Defaults to false.
 * `only_from_local_users`: Whether to only automatically accept invites from users on this homeserver. Defaults to false.
-* `worker_to_run_on`: Which worker to run this module on. This must match the "worker_name".
+* `worker_to_run_on`: Which worker to run this module on. This must match 
+  the "worker_name". If not set or `null`, invites will be accepted on the
+  main process.
 
 NOTE: Care should be taken not to enable this setting if the `synapse_auto_accept_invite` module is enabled and installed.
 The two modules will compete to perform the same task and may result in undesired behaviour. For example, multiple join

pyproject.toml

@@ -97,7 +97,7 @@ module-name = "synapse.synapse_rust"
 
 [tool.poetry]
 name = "matrix-synapse"
-version = "1.111.0"
+version = "1.114.0rc1"
 description = "Homeserver for the Matrix decentralised comms protocol"
 authors = ["Matrix.org Team and Contributors <packages@matrix.org>"]
 license = "AGPL-3.0-or-later"
@@ -201,8 +201,8 @@ netaddr = ">=0.7.18"
 # add a lower bound to the Jinja2 dependency.
 Jinja2 = ">=3.0"
 bleach = ">=1.4.3"
-# We use `Self`, which were added in `typing-extensions` 4.0.
-typing-extensions = ">=4.0"
+# We use `assert_never`, which were added in `typing-extensions` 4.1.
+typing-extensions = ">=4.1"
 # We enforce that we have a `cryptography` version that bundles an `openssl`
 # with the latest security patches.
 cryptography = ">=3.4.7"
@@ -322,7 +322,7 @@ all = [
 # This helps prevents merge conflicts when running a batch of dependabot updates.
 isort = ">=5.10.1"
 black = ">=22.7.0"
-ruff = "0.5.0"
+ruff = "0.5.5"
 # Type checking only works with the pydantic.v1 compat module from pydantic v2
 pydantic = "^2"
 

scripts-dev/federation_client.py

@@ -43,7 +43,7 @@ import argparse
 import base64
 import json
 import sys
-from typing import Any, Dict, Optional, Tuple
+from typing import Any, Dict, Mapping, Optional, Tuple, Union
 from urllib import parse as urlparse
 
 import requests
@@ -75,7 +75,7 @@ def encode_canonical_json(value: object) -> bytes:
         value,
         # Encode code-points outside of ASCII as UTF-8 rather than \u escapes
         ensure_ascii=False,
-        # Remove unecessary white space.
+        # Remove unnecessary white space.
         separators=(",", ":"),
         # Sort the keys of dictionaries.
         sort_keys=True,
@@ -298,12 +298,23 @@ class MatrixConnectionAdapter(HTTPAdapter):
 
         return super().send(request, *args, **kwargs)
 
-    def get_connection(
-        self, url: str, proxies: Optional[Dict[str, str]] = None
+    def get_connection_with_tls_context(
+        self,
+        request: PreparedRequest,
+        verify: Optional[Union[bool, str]],
+        proxies: Optional[Mapping[str, str]] = None,
+        cert: Optional[Union[Tuple[str, str], str]] = None,
     ) -> HTTPConnectionPool:
-        # overrides the get_connection() method in the base class
-        parsed = urlparse.urlsplit(url)
-        (host, port, ssl_server_name) = self._lookup(parsed.netloc)
+        # overrides the get_connection_with_tls_context() method in the base class
+        parsed = urlparse.urlsplit(request.url)
+
+        # Extract the server name from the request URL, and ensure it's a str.
+        hostname = parsed.netloc
+        if isinstance(hostname, bytes):
+            hostname = hostname.decode("utf-8")
+        assert isinstance(hostname, str)
+
+        (host, port, ssl_server_name) = self._lookup(hostname)
         print(
             f"Connecting to {host}:{port} with SNI {ssl_server_name}", file=sys.stderr
         )

scripts-dev/mypy_synapse_plugin.py

@@ -38,6 +38,7 @@ from mypy.types import (
     NoneType,
     TupleType,
     TypeAliasType,
+    TypeVarType,
     UninhabitedType,
     UnionType,
 )
@@ -233,6 +234,7 @@ IMMUTABLE_CUSTOM_TYPES = {
     "synapse.synapse_rust.push.FilteredPushRules",
     # This is technically not immutable, but close enough.
     "signedjson.types.VerifyKey",
+    "synapse.types.StrCollection",
 }
 
 # Immutable containers only if the values are also immutable.
@@ -298,7 +300,7 @@ def is_cacheable(
 
         elif rt.type.fullname in MUTABLE_CONTAINER_TYPES:
             # Mutable containers are mutable regardless of their underlying type.
-            return False, None
+            return False, f"container {rt.type.fullname} is mutable"
 
         elif "attrs" in rt.type.metadata:
             # attrs classes are only cachable iff it is frozen (immutable itself)
@@ -318,6 +320,9 @@ def is_cacheable(
             else:
                 return False, "non-frozen attrs class"
 
+        elif rt.type.is_enum:
+            # We assume Enum values are immutable
+            return True, None
         else:
             # Ensure we fail for unknown types, these generally means that the
             # above code is not complete.
@@ -326,6 +331,18 @@ def is_cacheable(
                 f"Don't know how to handle {rt.type.fullname} return type instance",
             )
 
+    elif isinstance(rt, TypeVarType):
+        # We consider TypeVars immutable if they are bound to a set of immutable
+        # types.
+        if rt.values:
+            for value in rt.values:
+                ok, note = is_cacheable(value, signature, verbose)
+                if not ok:
+                    return False, f"TypeVar bound not cacheable {value}"
+            return True, None
+
+        return False, "TypeVar is unbound"
+
     elif isinstance(rt, NoneType):
         # None is cachable.
         return True, None

scripts-dev/release.py

@@ -324,6 +324,11 @@ def tag(gh_token: Optional[str]) -> None:
 def _tag(gh_token: Optional[str]) -> None:
     """Tags the release and generates a draft GitHub release"""
 
+    if gh_token:
+        # Test that the GH Token is valid before continuing.
+        gh = Github(gh_token)
+        gh.get_user()
+
     # Make sure we're in a git repo.
     repo = get_repo_and_check_clean_checkout()
 
@@ -418,6 +423,11 @@ def publish(gh_token: str) -> None:
 def _publish(gh_token: str) -> None:
     """Publish release on GitHub."""
 
+    if gh_token:
+        # Test that the GH Token is valid before continuing.
+        gh = Github(gh_token)
+        gh.get_user()
+
     # Make sure we're in a git repo.
     get_repo_and_check_clean_checkout()
 
@@ -460,6 +470,11 @@ def upload(gh_token: Optional[str]) -> None:
 def _upload(gh_token: Optional[str]) -> None:
     """Upload release to pypi."""
 
+    if gh_token:
+        # Test that the GH Token is valid before continuing.
+        gh = Github(gh_token)
+        gh.get_user()
+
     current_version = get_package_version()
     tag_name = f"v{current_version}"
 
@@ -555,6 +570,11 @@ def wait_for_actions(gh_token: Optional[str]) -> None:
 
 
 def _wait_for_actions(gh_token: Optional[str]) -> None:
+    if gh_token:
+        # Test that the GH Token is valid before continuing.
+        gh = Github(gh_token)
+        gh.get_user()
+
     # Find out the version and tag name.
     current_version = get_package_version()
     tag_name = f"v{current_version}"
@@ -711,6 +731,11 @@ Ask the designated people to do the blog and tweets."""
 @cli.command()
 @click.option("--gh-token", envvar=["GH_TOKEN", "GITHUB_TOKEN"], required=True)
 def full(gh_token: str) -> None:
+    if gh_token:
+        # Test that the GH Token is valid before continuing.
+        gh = Github(gh_token)
+        gh.get_user()
+
     click.echo("1. If this is a security release, read the security wiki page.")
     click.echo("2. Check for any release blockers before proceeding.")
     click.echo("    https://github.com/element-hq/synapse/labels/X-Release-Blocker")

synapse/_scripts/synapse_port_db.py

@@ -129,6 +129,11 @@ BOOLEAN_COLUMNS = {
     "remote_media_cache": ["authenticated"],
     "room_stats_state": ["is_federatable"],
     "rooms": ["is_public", "has_auth_chain_index"],
+    "sliding_sync_joined_rooms": ["is_encrypted"],
+    "sliding_sync_membership_snapshots": [
+        "has_known_state",
+        "is_encrypted",
+    ],
     "users": ["shadow_banned", "approved", "locked", "suspended"],
     "un_partial_stated_event_stream": ["rejection_status_changed"],
     "users_who_share_rooms": ["share_private"],

synapse/api/constants.py

@@ -225,6 +225,11 @@ class EventContentFields:
     # This is deprecated in MSC2175.
     ROOM_CREATOR: Final = "creator"
 
+    # The version of the room for `m.room.create` events.
+    ROOM_VERSION: Final = "room_version"
+
+    ROOM_NAME: Final = "name"
+
     # Used in m.room.guest_access events.
     GUEST_ACCESS: Final = "guest_access"
 
@@ -237,6 +242,11 @@ class EventContentFields:
     # an unspecced field added to to-device messages to identify them uniquely-ish
     TO_DEVICE_MSGID: Final = "org.matrix.msgid"
 
+    # `m.room.encryption`` algorithm field
+    ENCRYPTION_ALGORITHM: Final = "algorithm"
+
+    TOMBSTONE_SUCCESSOR_ROOM: Final = "replacement_room"
+
 
 class EventUnsignedContentFields:
     """Fields found inside the 'unsigned' data on events"""

synapse/api/errors.py

@@ -128,6 +128,10 @@ class Codes(str, Enum):
     # MSC2677
     DUPLICATE_ANNOTATION = "M_DUPLICATE_ANNOTATION"
 
+    # MSC3575 we are telling the client they need to expire their sliding sync
+    # connection.
+    UNKNOWN_POS = "M_UNKNOWN_POS"
+
 
 class CodeMessageException(RuntimeError):
     """An exception with integer code, a message string attributes and optional headers.
@@ -847,3 +851,17 @@ class PartialStateConflictError(SynapseError):
             msg=PartialStateConflictError.message(),
             errcode=Codes.UNKNOWN,
         )
+
+
+class SlidingSyncUnknownPosition(SynapseError):
+    """An error that Synapse can return to signal to the client to expire their
+    sliding sync connection (i.e. send a new request without a `?since=`
+    param).
+    """
+
+    def __init__(self) -> None:
+        super().__init__(
+            HTTPStatus.BAD_REQUEST,
+            msg="Unknown position",
+            errcode=Codes.UNKNOWN_POS,
+        )

synapse/app/generic_worker.py

@@ -98,6 +98,7 @@ from synapse.storage.databases.main.roommember import RoomMemberWorkerStore
 from synapse.storage.databases.main.search import SearchStore
 from synapse.storage.databases.main.session import SessionStore
 from synapse.storage.databases.main.signatures import SignatureWorkerStore
+from synapse.storage.databases.main.sliding_sync import SlidingSyncStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.databases.main.stats import StatsStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
@@ -159,6 +160,7 @@ class GenericWorkerStore(
     SessionStore,
     TaskSchedulerWorkerStore,
     ExperimentalFeaturesStore,
+    SlidingSyncStore,
 ):
     # Properties that multiple storage classes define. Tell mypy what the
     # expected type is.
@@ -206,6 +208,21 @@ class GenericWorkerServer(HomeServer):
                                 "/_synapse/admin": admin_resource,
                             }
                         )
+
+                        if "federation" not in res.names:
+                            # Only load the federation media resource separately if federation
+                            # resource is not specified since federation resource includes media
+                            # resource.
+                            resources[FEDERATION_PREFIX] = TransportLayerServer(
+                                self, servlet_groups=["media"]
+                            )
+                        if "client" not in res.names:
+                            # Only load the client media resource separately if client
+                            # resource is not specified since client resource includes media
+                            # resource.
+                            resources[CLIENT_API_PREFIX] = ClientRestResource(
+                                self, servlet_groups=["media"]
+                            )
                     else:
                         logger.warning(
                             "A 'media' listener is configured but the media"

synapse/app/homeserver.py

@@ -101,6 +101,12 @@ class SynapseHomeServer(HomeServer):
                     # Skip loading openid resource if federation is defined
                     # since federation resource will include openid
                     continue
+                if name == "media" and (
+                    "federation" in res.names or "client" in res.names
+                ):
+                    # Skip loading media resource if federation or client are defined
+                    # since federation & client resources will include media
+                    continue
                 if name == "health":
                     # Skip loading, health resource is always included
                     continue
@@ -231,6 +237,14 @@ class SynapseHomeServer(HomeServer):
                     "'media' resource conflicts with enable_media_repo=False"
                 )
 
+        if name == "media":
+            resources[FEDERATION_PREFIX] = TransportLayerServer(
+                self, servlet_groups=["media"]
+            )
+            resources[CLIENT_API_PREFIX] = ClientRestResource(
+                self, servlet_groups=["media"]
+            )
+
         if name in ["keys", "federation"]:
             resources[SERVER_KEY_PREFIX] = KeyResource(self)
 

synapse/config/server.py

@@ -384,6 +384,11 @@ class ServerConfig(Config):
         # Whether to internally track presence, requires that presence is enabled,
         self.track_presence = self.presence_enabled and presence_enabled != "untracked"
 
+        # Determines if presence results for offline users are included on initial/full sync
+        self.presence_include_offline_users_on_sync = presence_config.get(
+            "include_offline_users_on_sync", False
+        )
+
         # Custom presence router module
         # This is the legacy way of configuring it (the config should now be put in the modules section)
         self.presence_router_module_class = None

synapse/crypto/keyring.py

@@ -589,7 +589,7 @@ class BaseV2KeyFetcher(KeyFetcher):
                 % (server_name,)
             )
 
-        for key_id, key_data in response_json["old_verify_keys"].items():
+        for key_id, key_data in response_json.get("old_verify_keys", {}).items():
             if is_signing_algorithm_supported(key_id):
                 key_base64 = key_data["key"]
                 key_bytes = decode_base64(key_base64)

synapse/events/__init__.py

@@ -554,3 +554,22 @@ def relation_from_event(event: EventBase) -> Optional[_EventRelation]:
             aggregation_key = None
 
     return _EventRelation(parent_id, rel_type, aggregation_key)
+
+
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class StrippedStateEvent:
+    """
+    A stripped down state event. Usually used for remote invite/knocks so the user can
+    make an informed decision on whether they want to join.
+
+    Attributes:
+        type: Event `type`
+        state_key: Event `state_key`
+        sender: Event `sender`
+        content: Event `content`
+    """
+
+    type: str
+    state_key: str
+    sender: str
+    content: Dict[str, Any]

synapse/events/utils.py

@@ -49,7 +49,7 @@ from synapse.api.errors import Codes, SynapseError
 from synapse.api.room_versions import RoomVersion
 from synapse.types import JsonDict, Requester
 
-from . import EventBase, make_event_from_dict
+from . import EventBase, StrippedStateEvent, make_event_from_dict
 
 if TYPE_CHECKING:
     from synapse.handlers.relations import BundledAggregations
@@ -854,3 +854,30 @@ def strip_event(event: EventBase) -> JsonDict:
         "content": event.content,
         "sender": event.sender,
     }
+
+
+def parse_stripped_state_event(raw_stripped_event: Any) -> Optional[StrippedStateEvent]:
+    """
+    Given a raw value from an event's `unsigned` field, attempt to parse it into a
+    `StrippedStateEvent`.
+    """
+    if isinstance(raw_stripped_event, dict):
+        # All of these fields are required
+        type = raw_stripped_event.get("type")
+        state_key = raw_stripped_event.get("state_key")
+        sender = raw_stripped_event.get("sender")
+        content = raw_stripped_event.get("content")
+        if (
+            isinstance(type, str)
+            and isinstance(state_key, str)
+            and isinstance(sender, str)
+            and isinstance(content, dict)
+        ):
+            return StrippedStateEvent(
+                type=type,
+                state_key=state_key,
+                sender=sender,
+                content=content,
+            )
+
+    return None

synapse/federation/transport/server/__init__.py

@@ -271,6 +271,10 @@ SERVLET_GROUPS: Dict[str, Iterable[Type[BaseFederationServlet]]] = {
     "federation": FEDERATION_SERVLET_CLASSES,
     "room_list": (PublicRoomList,),
     "openid": (OpenIdUserInfo,),
+    "media": (
+        FederationMediaDownloadServlet,
+        FederationMediaThumbnailServlet,
+    ),
 }
 
 

synapse/federation/transport/server/federation.py

@@ -912,6 +912,4 @@ FEDERATION_SERVLET_CLASSES: Tuple[Type[BaseFederationServlet], ...] = (
     FederationV1SendKnockServlet,
     FederationMakeKnockServlet,
     FederationAccountStatusServlet,
-    FederationMediaDownloadServlet,
-    FederationMediaThumbnailServlet,
 )

synapse/handlers/admin.py

@@ -197,8 +197,14 @@ class AdminHandler:
             # events that we have and then filtering, this isn't the most
             # efficient method perhaps but it does guarantee we get everything.
             while True:
-                events, _ = await self._store.paginate_room_events(
-                    room_id, from_key, to_key, limit=100, direction=Direction.FORWARDS
+                events, _ = (
+                    await self._store.paginate_room_events_by_topological_ordering(
+                        room_id=room_id,
+                        from_key=from_key,
+                        to_key=to_key,
+                        limit=100,
+                        direction=Direction.FORWARDS,
+                    )
                 )
                 if not events:
                     break

synapse/handlers/device.py

@@ -20,10 +20,20 @@
 #
 #
 import logging
-from typing import TYPE_CHECKING, Dict, Iterable, List, Mapping, Optional, Set, Tuple
+from typing import (
+    TYPE_CHECKING,
+    AbstractSet,
+    Dict,
+    Iterable,
+    List,
+    Mapping,
+    Optional,
+    Set,
+    Tuple,
+)
 
 from synapse.api import errors
-from synapse.api.constants import EduTypes, EventTypes
+from synapse.api.constants import EduTypes, EventTypes, Membership
 from synapse.api.errors import (
     Codes,
     FederationDeniedError,
@@ -38,7 +48,9 @@ from synapse.metrics.background_process_metrics import (
     wrap_as_background_process,
 )
 from synapse.storage.databases.main.client_ips import DeviceLastConnectionInfo
+from synapse.storage.databases.main.state_deltas import StateDelta
 from synapse.types import (
+    DeviceListUpdates,
     JsonDict,
     JsonMapping,
     ScheduledTask,
@@ -214,138 +226,210 @@ class DeviceWorkerHandler:
     @cancellable
     async def get_user_ids_changed(
         self, user_id: str, from_token: StreamToken
-    ) -> JsonDict:
+    ) -> DeviceListUpdates:
         """Get list of users that have had the devices updated, or have newly
         joined a room, that `user_id` may be interested in.
         """
 
         set_tag("user_id", user_id)
         set_tag("from_token", str(from_token))
-        now_room_key = self.store.get_room_max_token()
 
-        room_ids = await self.store.get_rooms_for_user(user_id)
+        now_token = self._event_sources.get_current_token()
 
-        changed = await self.get_device_changes_in_shared_rooms(
-            user_id, room_ids, from_token
+        # We need to work out all the different membership changes for the user
+        # and user they share a room with, to pass to
+        # `generate_sync_entry_for_device_list`. See its docstring for details
+        # on the data required.
+
+        joined_room_ids = await self.store.get_rooms_for_user(user_id)
+
+        # Get the set of rooms that the user has joined/left
+        membership_changes = (
+            await self.store.get_current_state_delta_membership_changes_for_user(
+                user_id, from_key=from_token.room_key, to_key=now_token.room_key
+            )
         )
 
-        # Then work out if any users have since joined
-        rooms_changed = self.store.get_rooms_that_changed(room_ids, from_token.room_key)
+        # Check for newly joined or left rooms. We need to make sure that we add
+        # to newly joined in the case membership goes from join -> leave -> join
+        # again.
+        newly_joined_rooms: Set[str] = set()
+        newly_left_rooms: Set[str] = set()
+        for change in membership_changes:
+            # We check for changes in "joinedness", i.e. if the membership has
+            # changed to or from JOIN.
+            if change.membership == Membership.JOIN:
+                if change.prev_membership != Membership.JOIN:
+                    newly_joined_rooms.add(change.room_id)
+                    newly_left_rooms.discard(change.room_id)
+            elif change.prev_membership == Membership.JOIN:
+                newly_joined_rooms.discard(change.room_id)
+                newly_left_rooms.add(change.room_id)
 
-        member_events = await self.store.get_membership_changes_for_user(
-            user_id, from_token.room_key, now_room_key
+        # We now work out if any other users have since joined or left the rooms
+        # the user is currently in.
+
+        # List of membership changes per room
+        room_to_deltas: Dict[str, List[StateDelta]] = {}
+        # The set of event IDs of membership events (so we can fetch their
+        # associated membership).
+        memberships_to_fetch: Set[str] = set()
+
+        # TODO: Only pull out membership events?
+        state_changes = await self.store.get_current_state_deltas_for_rooms(
+            joined_room_ids, from_token=from_token.room_key, to_token=now_token.room_key
         )
-        rooms_changed.update(event.room_id for event in member_events)
-
-        stream_ordering = from_token.room_key.stream
-
-        possibly_changed = set(changed)
-        possibly_left = set()
-        for room_id in rooms_changed:
-            # Check if the forward extremities have changed. If not then we know
-            # the current state won't have changed, and so we can skip this room.
-            try:
-                if not await self.store.have_room_forward_extremities_changed_since(
-                    room_id, stream_ordering
-                ):
-                    continue
-            except errors.StoreError:
-                pass
-
-            current_state_ids = await self._state_storage.get_current_state_ids(
-                room_id, await_full_state=False
-            )
-
-            # The user may have left the room
-            # TODO: Check if they actually did or if we were just invited.
-            if room_id not in room_ids:
-                for etype, state_key in current_state_ids.keys():
-                    if etype != EventTypes.Member:
-                        continue
-                    possibly_left.add(state_key)
+        for delta in state_changes:
+            if delta.event_type != EventTypes.Member:
                 continue
 
-            # Fetch the current state at the time.
-            try:
-                event_ids = await self.store.get_forward_extremities_for_room_at_stream_ordering(
-                    room_id, stream_ordering=stream_ordering
-                )
-            except errors.StoreError:
-                # we have purged the stream_ordering index since the stream
-                # ordering: treat it the same as a new room
-                event_ids = []
+            room_to_deltas.setdefault(delta.room_id, []).append(delta)
+            if delta.event_id:
+                memberships_to_fetch.add(delta.event_id)
+            if delta.prev_event_id:
+                memberships_to_fetch.add(delta.prev_event_id)
 
-            # special-case for an empty prev state: include all members
-            # in the changed list
-            if not event_ids:
-                log_kv(
-                    {"event": "encountered empty previous state", "room_id": room_id}
-                )
-                for etype, state_key in current_state_ids.keys():
-                    if etype != EventTypes.Member:
-                        continue
-                    possibly_changed.add(state_key)
-                continue
+        # Fetch all the memberships for the membership events
+        event_id_to_memberships = await self.store.get_membership_from_event_ids(
+            memberships_to_fetch
+        )
 
-            current_member_id = current_state_ids.get((EventTypes.Member, user_id))
-            if not current_member_id:
-                continue
+        joined_invited_knocked = (
+            Membership.JOIN,
+            Membership.INVITE,
+            Membership.KNOCK,
+        )
 
-            # mapping from event_id -> state_dict
-            prev_state_ids = await self._state_storage.get_state_ids_for_events(
-                event_ids,
-                await_full_state=False,
-            )
+        # We now want to find any user that have newly joined/invited/knocked,
+        # or newly left, similarly to above.
+        newly_joined_or_invited_or_knocked_users: Set[str] = set()
+        newly_left_users: Set[str] = set()
+        for _, deltas in room_to_deltas.items():
+            for delta in deltas:
+                # Get the prev/new memberships for the delta
+                new_membership = None
+                prev_membership = None
+                if delta.event_id:
+                    m = event_id_to_memberships.get(delta.event_id)
+                    if m is not None:
+                        new_membership = m.membership
+                if delta.prev_event_id:
+                    m = event_id_to_memberships.get(delta.prev_event_id)
+                    if m is not None:
+                        prev_membership = m.membership
 
-            # Check if we've joined the room? If so we just blindly add all the users to
-            # the "possibly changed" users.
-            for state_dict in prev_state_ids.values():
-                member_event = state_dict.get((EventTypes.Member, user_id), None)
-                if not member_event or member_event != current_member_id:
-                    for etype, state_key in current_state_ids.keys():
-                        if etype != EventTypes.Member:
-                            continue
-                        possibly_changed.add(state_key)
-                    break
+                # Check if a user has newly joined/invited/knocked, or left.
+                if new_membership in joined_invited_knocked:
+                    if prev_membership not in joined_invited_knocked:
+                        newly_joined_or_invited_or_knocked_users.add(delta.state_key)
+                        newly_left_users.discard(delta.state_key)
+                elif prev_membership in joined_invited_knocked:
+                    newly_joined_or_invited_or_knocked_users.discard(delta.state_key)
+                    newly_left_users.add(delta.state_key)
 
-            # If there has been any change in membership, include them in the
-            # possibly changed list. We'll check if they are joined below,
-            # and we're not toooo worried about spuriously adding users.
-            for key, event_id in current_state_ids.items():
-                etype, state_key = key
-                if etype != EventTypes.Member:
-                    continue
+        # Now we actually calculate the device list entry with the information
+        # calculated above.
+        device_list_updates = await self.generate_sync_entry_for_device_list(
+            user_id=user_id,
+            since_token=from_token,
+            now_token=now_token,
+            joined_room_ids=joined_room_ids,
+            newly_joined_rooms=newly_joined_rooms,
+            newly_joined_or_invited_or_knocked_users=newly_joined_or_invited_or_knocked_users,
+            newly_left_rooms=newly_left_rooms,
+            newly_left_users=newly_left_users,
+        )
 
-                # check if this member has changed since any of the extremities
-                # at the stream_ordering, and add them to the list if so.
-                for state_dict in prev_state_ids.values():
-                    prev_event_id = state_dict.get(key, None)
-                    if not prev_event_id or prev_event_id != event_id:
-                        if state_key != user_id:
-                            possibly_changed.add(state_key)
-                        break
+        log_kv(
+            {
+                "changed": device_list_updates.changed,
+                "left": device_list_updates.left,
+            }
+        )
 
-        if possibly_changed or possibly_left:
-            possibly_joined = possibly_changed
-            possibly_left = possibly_changed | possibly_left
+        return device_list_updates
 
-            # Double check if we still share rooms with the given user.
-            users_rooms = await self.store.get_rooms_for_users(possibly_left)
-            for changed_user_id, entries in users_rooms.items():
-                if any(rid in room_ids for rid in entries):
-                    possibly_left.discard(changed_user_id)
-                else:
-                    possibly_joined.discard(changed_user_id)
+    @measure_func("_generate_sync_entry_for_device_list")
+    async def generate_sync_entry_for_device_list(
+        self,
+        user_id: str,
+        since_token: StreamToken,
+        now_token: StreamToken,
+        joined_room_ids: AbstractSet[str],
+        newly_joined_rooms: AbstractSet[str],
+        newly_joined_or_invited_or_knocked_users: AbstractSet[str],
+        newly_left_rooms: AbstractSet[str],
+        newly_left_users: AbstractSet[str],
+    ) -> DeviceListUpdates:
+        """Generate the DeviceListUpdates section of sync
 
-        else:
-            possibly_joined = set()
-            possibly_left = set()
+        Args:
+            sync_result_builder
+            newly_joined_rooms: Set of rooms user has joined since previous sync
+            newly_joined_or_invited_or_knocked_users: Set of users that have joined,
+                been invited to a room or are knocking on a room since
+                previous sync.
+            newly_left_rooms: Set of rooms user has left since previous sync
+            newly_left_users: Set of users that have left a room we're in since
+                previous sync
+        """
+        # Take a copy since these fields will be mutated later.
+        newly_joined_or_invited_or_knocked_users = set(
+            newly_joined_or_invited_or_knocked_users
+        )
+        newly_left_users = set(newly_left_users)
 
-        result = {"changed": list(possibly_joined), "left": list(possibly_left)}
+        # We want to figure out what user IDs the client should refetch
+        # device keys for, and which users we aren't going to track changes
+        # for anymore.
+        #
+        # For the first step we check:
+        #   a. if any users we share a room with have updated their devices,
+        #      and
+        #   b. we also check if we've joined any new rooms, or if a user has
+        #      joined a room we're in.
+        #
+        # For the second step we just find any users we no longer share a
+        # room with by looking at all users that have left a room plus users
+        # that were in a room we've left.
 
-        log_kv(result)
+        users_that_have_changed = set()
 
-        return result
+        # Step 1a, check for changes in devices of users we share a room
+        # with
+        users_that_have_changed = await self.get_device_changes_in_shared_rooms(
+            user_id,
+            joined_room_ids,
+            from_token=since_token,
+            now_token=now_token,
+        )
+
+        # Step 1b, check for newly joined rooms
+        for room_id in newly_joined_rooms:
+            joined_users = await self.store.get_users_in_room(room_id)
+            newly_joined_or_invited_or_knocked_users.update(joined_users)
+
+        # TODO: Check that these users are actually new, i.e. either they
+        # weren't in the previous sync *or* they left and rejoined.
+        users_that_have_changed.update(newly_joined_or_invited_or_knocked_users)
+
+        user_signatures_changed = await self.store.get_users_whose_signatures_changed(
+            user_id, since_token.device_list_key
+        )
+        users_that_have_changed.update(user_signatures_changed)
+
+        # Now find users that we no longer track
+        for room_id in newly_left_rooms:
+            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.
+        left_users_rooms = await self.store.get_rooms_for_users(newly_left_users)
+        for user_id, entries in left_users_rooms.items():
+            if any(rid in joined_room_ids for rid in entries):
+                newly_left_users.discard(user_id)
+
+        return DeviceListUpdates(changed=users_that_have_changed, left=newly_left_users)
 
     async def on_federation_query_user_devices(self, user_id: str) -> JsonDict:
         if not self.hs.is_mine(UserID.from_string(user_id)):

synapse/handlers/e2e_keys.py

@@ -291,13 +291,20 @@ class E2eKeysHandler:
 
             # Only try and fetch keys for destinations that are not marked as
             # down.
-            filtered_destinations = await filter_destinations_by_retry_limiter(
-                remote_queries_not_in_cache.keys(),
-                self.clock,
-                self.store,
-                # Let's give an arbitrary grace period for those hosts that are
-                # only recently down
-                retry_due_within_ms=60 * 1000,
+            unfiltered_destinations = remote_queries_not_in_cache.keys()
+            filtered_destinations = set(
+                await filter_destinations_by_retry_limiter(
+                    unfiltered_destinations,
+                    self.clock,
+                    self.store,
+                    # Let's give an arbitrary grace period for those hosts that are
+                    # only recently down
+                    retry_due_within_ms=60 * 1000,
+                )
+            )
+            failures.update(
+                (dest, _NOT_READY_FOR_RETRY_FAILURE)
+                for dest in (unfiltered_destinations - filtered_destinations)
             )
 
             await concurrently_execute(
@@ -1641,6 +1648,9 @@ def _check_device_signature(
         raise SynapseError(400, "Invalid signature", Codes.INVALID_SIGNATURE)
 
 
+_NOT_READY_FOR_RETRY_FAILURE = {"status": 503, "message": "Not ready for retry"}
+
+
 def _exception_to_failure(e: Exception) -> JsonDict:
     if isinstance(e, SynapseError):
         return {"status": e.code, "errcode": e.errcode, "message": str(e)}
@@ -1649,7 +1659,7 @@ def _exception_to_failure(e: Exception) -> JsonDict:
         return {"status": e.code, "message": str(e)}
 
     if isinstance(e, NotRetryingDestination):
-        return {"status": 503, "message": "Not ready for retry"}
+        return _NOT_READY_FOR_RETRY_FAILURE
 
     # include ConnectionRefused and other errors
     #

synapse/handlers/e2e_room_keys.py

@@ -34,7 +34,7 @@ from synapse.api.errors import (
 from synapse.logging.opentracing import log_kv, trace
 from synapse.storage.databases.main.e2e_room_keys import RoomKey
 from synapse.types import JsonDict
-from synapse.util.async_helpers import Linearizer
+from synapse.util.async_helpers import ReadWriteLock
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -58,7 +58,7 @@ class E2eRoomKeysHandler:
         # clients belonging to a user will receive and try to upload a new session at
         # roughly the same time.  Also used to lock out uploads when the key is being
         # changed.
-        self._upload_linearizer = Linearizer("upload_room_keys_lock")
+        self._upload_lock = ReadWriteLock()
 
     @trace
     async def get_room_keys(
@@ -89,7 +89,7 @@ class E2eRoomKeysHandler:
 
         # we deliberately take the lock to get keys so that changing the version
         # works atomically
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.read(user_id):
             # make sure the backup version exists
             try:
                 await self.store.get_e2e_room_keys_version_info(user_id, version)
@@ -132,7 +132,7 @@ class E2eRoomKeysHandler:
         """
 
         # lock for consistency with uploading
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.write(user_id):
             # make sure the backup version exists
             try:
                 version_info = await self.store.get_e2e_room_keys_version_info(
@@ -193,7 +193,7 @@ class E2eRoomKeysHandler:
         # TODO: Validate the JSON to make sure it has the right keys.
 
         # XXX: perhaps we should use a finer grained lock here?
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.write(user_id):
             # Check that the version we're trying to upload is the current version
             try:
                 version_info = await self.store.get_e2e_room_keys_version_info(user_id)
@@ -355,7 +355,7 @@ class E2eRoomKeysHandler:
         # TODO: Validate the JSON to make sure it has the right keys.
 
         # lock everyone out until we've switched version
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.write(user_id):
             new_version = await self.store.create_e2e_room_keys_version(
                 user_id, version_info
             )
@@ -382,7 +382,7 @@ class E2eRoomKeysHandler:
         }
         """
 
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.read(user_id):
             try:
                 res = await self.store.get_e2e_room_keys_version_info(user_id, version)
             except StoreError as e:
@@ -407,7 +407,7 @@ class E2eRoomKeysHandler:
             NotFoundError: if this backup version doesn't exist
         """
 
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.write(user_id):
             try:
                 await self.store.delete_e2e_room_keys_version(user_id, version)
             except StoreError as e:
@@ -437,7 +437,7 @@ class E2eRoomKeysHandler:
             raise SynapseError(
                 400, "Version in body does not match", Codes.INVALID_PARAM
             )
-        async with self._upload_linearizer.queue(user_id):
+        async with self._upload_lock.write(user_id):
             try:
                 old_info = await self.store.get_e2e_room_keys_version_info(
                     user_id, version

synapse/handlers/pagination.py

@@ -507,13 +507,15 @@ class PaginationHandler:
 
         # Initially fetch the events from the database. With any luck, we can return
         # these without blocking on backfill (handled below).
-        events, next_key = await self.store.paginate_room_events(
-            room_id=room_id,
-            from_key=from_token.room_key,
-            to_key=to_room_key,
-            direction=pagin_config.direction,
-            limit=pagin_config.limit,
-            event_filter=event_filter,
+        events, next_key = (
+            await self.store.paginate_room_events_by_topological_ordering(
+                room_id=room_id,
+                from_key=from_token.room_key,
+                to_key=to_room_key,
+                direction=pagin_config.direction,
+                limit=pagin_config.limit,
+                event_filter=event_filter,
+            )
         )
 
         if pagin_config.direction == Direction.BACKWARDS:
@@ -582,13 +584,15 @@ class PaginationHandler:
                 # If we did backfill something, refetch the events from the database to
                 # catch anything new that might have been added since we last fetched.
                 if did_backfill:
-                    events, next_key = await self.store.paginate_room_events(
-                        room_id=room_id,
-                        from_key=from_token.room_key,
-                        to_key=to_room_key,
-                        direction=pagin_config.direction,
-                        limit=pagin_config.limit,
-                        event_filter=event_filter,
+                    events, next_key = (
+                        await self.store.paginate_room_events_by_topological_ordering(
+                            room_id=room_id,
+                            from_key=from_token.room_key,
+                            to_key=to_room_key,
+                            direction=pagin_config.direction,
+                            limit=pagin_config.limit,
+                            event_filter=event_filter,
+                        )
                     )
             else:
                 # Otherwise, we can backfill in the background for eventual

synapse/handlers/profile.py

@@ -74,6 +74,17 @@ class ProfileHandler:
         self._third_party_rules = hs.get_module_api_callbacks().third_party_event_rules
 
     async def get_profile(self, user_id: str, ignore_backoff: bool = True) -> JsonDict:
+        """
+        Get a user's profile as a JSON dictionary.
+
+        Args:
+            user_id: The user to fetch the profile of.
+            ignore_backoff: True to ignore backoff when fetching over federation.
+
+        Returns:
+            A JSON dictionary. For local queries this will include the displayname and avatar_url
+            fields. For remote queries it may contain arbitrary information.
+        """
         target_user = UserID.from_string(user_id)
 
         if self.hs.is_mine(target_user):
@@ -107,6 +118,15 @@ class ProfileHandler:
                 raise e.to_synapse_error()
 
     async def get_displayname(self, target_user: UserID) -> Optional[str]:
+        """
+        Fetch a user's display name from their profile.
+
+        Args:
+            target_user: The user to fetch the display name of.
+
+        Returns:
+            The user's display name or None if unset.
+        """
         if self.hs.is_mine(target_user):
             try:
                 displayname = await self.store.get_profile_displayname(target_user)
@@ -203,6 +223,15 @@ class ProfileHandler:
             await self._update_join_states(requester, target_user)
 
     async def get_avatar_url(self, target_user: UserID) -> Optional[str]:
+        """
+        Fetch a user's avatar URL from their profile.
+
+        Args:
+            target_user: The user to fetch the avatar URL of.
+
+        Returns:
+            The user's avatar URL or None if unset.
+        """
         if self.hs.is_mine(target_user):
             try:
                 avatar_url = await self.store.get_profile_avatar_url(target_user)
@@ -403,6 +432,12 @@ class ProfileHandler:
     async def _update_join_states(
         self, requester: Requester, target_user: UserID
     ) -> None:
+        """
+        Update the membership events of each room the user is joined to with the
+        new profile information.
+
+        Note that this stomps over any custom display name or avatar URL in member events.
+        """
         if not self.hs.is_mine(target_user):
             return
 

synapse/handlers/receipts.py

@@ -286,8 +286,14 @@ class ReceiptEventSource(EventSource[MultiWriterStreamToken, JsonMapping]):
         room_ids: Iterable[str],
         is_guest: bool,
         explicit_room_id: Optional[str] = None,
+        to_key: Optional[MultiWriterStreamToken] = None,
     ) -> Tuple[List[JsonMapping], MultiWriterStreamToken]:
-        to_key = self.get_current_key()
+        """
+        Find read receipts for given rooms (> `from_token` and <= `to_token`)
+        """
+
+        if to_key is None:
+            to_key = self.get_current_key()
 
         if from_key == to_key:
             return [], to_key

synapse/handlers/room.py

@@ -1750,7 +1750,7 @@ class RoomEventSource(EventSource[RoomStreamToken, EventBase]):
                 from_key=from_key,
                 to_key=to_key,
                 limit=limit or 10,
-                order="ASC",
+                direction=Direction.FORWARDS,
             )
 
             events = list(room_events)

synapse/handlers/sliding_sync/extensions.py

@@ -0,0 +1,662 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2023 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+import logging
+from typing import TYPE_CHECKING, Dict, List, Mapping, Optional, Sequence, Set
+
+from typing_extensions import assert_never
+
+from synapse.api.constants import AccountDataTypes
+from synapse.handlers.receipts import ReceiptEventSource
+from synapse.logging.opentracing import trace
+from synapse.types import (
+    DeviceListUpdates,
+    JsonMapping,
+    MultiWriterStreamToken,
+    SlidingSyncStreamToken,
+    StreamToken,
+)
+from synapse.types.handlers.sliding_sync import (
+    HaveSentRoomFlag,
+    MutablePerConnectionState,
+    OperationType,
+    PerConnectionState,
+    SlidingSyncConfig,
+    SlidingSyncResult,
+)
+
+if TYPE_CHECKING:
+    from synapse.server import HomeServer
+
+logger = logging.getLogger(__name__)
+
+
+class SlidingSyncExtensionHandler:
+    """Handles the extensions to sliding sync."""
+
+    def __init__(self, hs: "HomeServer"):
+        self.store = hs.get_datastores().main
+        self.event_sources = hs.get_event_sources()
+        self.device_handler = hs.get_device_handler()
+        self.push_rules_handler = hs.get_push_rules_handler()
+
+    @trace
+    async def get_extensions_response(
+        self,
+        sync_config: SlidingSyncConfig,
+        previous_connection_state: "PerConnectionState",
+        new_connection_state: "MutablePerConnectionState",
+        actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
+        actual_room_ids: Set[str],
+        actual_room_response_map: Dict[str, SlidingSyncResult.RoomResult],
+        to_token: StreamToken,
+        from_token: Optional[SlidingSyncStreamToken],
+    ) -> SlidingSyncResult.Extensions:
+        """Handle extension requests.
+
+        Args:
+            sync_config: Sync configuration
+            new_connection_state: Snapshot of the current per-connection state
+            new_per_connection_state: A mutable copy of the per-connection
+                state, used to record updates to the state during this request.
+            actual_lists: Sliding window API. A map of list key to list results in the
+                Sliding Sync response.
+            actual_room_ids: The actual room IDs in the the Sliding Sync response.
+            actual_room_response_map: A map of room ID to room results in the the
+                Sliding Sync response.
+            to_token: The point in the stream to sync up to.
+            from_token: The point in the stream to sync from.
+        """
+
+        if sync_config.extensions is None:
+            return SlidingSyncResult.Extensions()
+
+        to_device_response = None
+        if sync_config.extensions.to_device is not None:
+            to_device_response = await self.get_to_device_extension_response(
+                sync_config=sync_config,
+                to_device_request=sync_config.extensions.to_device,
+                to_token=to_token,
+            )
+
+        e2ee_response = None
+        if sync_config.extensions.e2ee is not None:
+            e2ee_response = await self.get_e2ee_extension_response(
+                sync_config=sync_config,
+                e2ee_request=sync_config.extensions.e2ee,
+                to_token=to_token,
+                from_token=from_token,
+            )
+
+        account_data_response = None
+        if sync_config.extensions.account_data is not None:
+            account_data_response = await self.get_account_data_extension_response(
+                sync_config=sync_config,
+                actual_lists=actual_lists,
+                actual_room_ids=actual_room_ids,
+                account_data_request=sync_config.extensions.account_data,
+                to_token=to_token,
+                from_token=from_token,
+            )
+
+        receipts_response = None
+        if sync_config.extensions.receipts is not None:
+            receipts_response = await self.get_receipts_extension_response(
+                sync_config=sync_config,
+                previous_connection_state=previous_connection_state,
+                new_connection_state=new_connection_state,
+                actual_lists=actual_lists,
+                actual_room_ids=actual_room_ids,
+                actual_room_response_map=actual_room_response_map,
+                receipts_request=sync_config.extensions.receipts,
+                to_token=to_token,
+                from_token=from_token,
+            )
+
+        typing_response = None
+        if sync_config.extensions.typing is not None:
+            typing_response = await self.get_typing_extension_response(
+                sync_config=sync_config,
+                actual_lists=actual_lists,
+                actual_room_ids=actual_room_ids,
+                actual_room_response_map=actual_room_response_map,
+                typing_request=sync_config.extensions.typing,
+                to_token=to_token,
+                from_token=from_token,
+            )
+
+        return SlidingSyncResult.Extensions(
+            to_device=to_device_response,
+            e2ee=e2ee_response,
+            account_data=account_data_response,
+            receipts=receipts_response,
+            typing=typing_response,
+        )
+
+    def find_relevant_room_ids_for_extension(
+        self,
+        requested_lists: Optional[List[str]],
+        requested_room_ids: Optional[List[str]],
+        actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
+        actual_room_ids: Set[str],
+    ) -> Set[str]:
+        """
+        Handle the reserved `lists`/`rooms` keys for extensions. Extensions should only
+        return results for rooms in the Sliding Sync response. This matches up the
+        requested rooms/lists with the actual lists/rooms in the Sliding Sync response.
+
+        {"lists": []}                    // Do not process any lists.
+        {"lists": ["rooms", "dms"]}      // Process only a subset of lists.
+        {"lists": ["*"]}                 // Process all lists defined in the Sliding Window API. (This is the default.)
+
+        {"rooms": []}                    // Do not process any specific rooms.
+        {"rooms": ["!a:b", "!c:d"]}      // Process only a subset of room subscriptions.
+        {"rooms": ["*"]}                 // Process all room subscriptions defined in the Room Subscription API. (This is the default.)
+
+        Args:
+            requested_lists: The `lists` from the extension request.
+            requested_room_ids: The `rooms` from the extension request.
+            actual_lists: The actual lists from the Sliding Sync response.
+            actual_room_ids: The actual room subscriptions from the Sliding Sync request.
+        """
+
+        # We only want to include account data for rooms that are already in the sliding
+        # sync response AND that were requested in the account data request.
+        relevant_room_ids: Set[str] = set()
+
+        # See what rooms from the room subscriptions we should get account data for
+        if requested_room_ids is not None:
+            for room_id in requested_room_ids:
+                # A wildcard means we process all rooms from the room subscriptions
+                if room_id == "*":
+                    relevant_room_ids.update(actual_room_ids)
+                    break
+
+                if room_id in actual_room_ids:
+                    relevant_room_ids.add(room_id)
+
+        # See what rooms from the sliding window lists we should get account data for
+        if requested_lists is not None:
+            for list_key in requested_lists:
+                # Just some typing because we share the variable name in multiple places
+                actual_list: Optional[SlidingSyncResult.SlidingWindowList] = None
+
+                # A wildcard means we process rooms from all lists
+                if list_key == "*":
+                    for actual_list in actual_lists.values():
+                        # We only expect a single SYNC operation for any list
+                        assert len(actual_list.ops) == 1
+                        sync_op = actual_list.ops[0]
+                        assert sync_op.op == OperationType.SYNC
+
+                        relevant_room_ids.update(sync_op.room_ids)
+
+                    break
+
+                actual_list = actual_lists.get(list_key)
+                if actual_list is not None:
+                    # We only expect a single SYNC operation for any list
+                    assert len(actual_list.ops) == 1
+                    sync_op = actual_list.ops[0]
+                    assert sync_op.op == OperationType.SYNC
+
+                    relevant_room_ids.update(sync_op.room_ids)
+
+        return relevant_room_ids
+
+    @trace
+    async def get_to_device_extension_response(
+        self,
+        sync_config: SlidingSyncConfig,
+        to_device_request: SlidingSyncConfig.Extensions.ToDeviceExtension,
+        to_token: StreamToken,
+    ) -> Optional[SlidingSyncResult.Extensions.ToDeviceExtension]:
+        """Handle to-device extension (MSC3885)
+
+        Args:
+            sync_config: Sync configuration
+            to_device_request: The to-device extension from the request
+            to_token: The point in the stream to sync up to.
+        """
+        user_id = sync_config.user.to_string()
+        device_id = sync_config.requester.device_id
+
+        # Skip if the extension is not enabled
+        if not to_device_request.enabled:
+            return None
+
+        # Check that this request has a valid device ID (not all requests have
+        # to belong to a device, and so device_id is None)
+        if device_id is None:
+            return SlidingSyncResult.Extensions.ToDeviceExtension(
+                next_batch=f"{to_token.to_device_key}",
+                events=[],
+            )
+
+        since_stream_id = 0
+        if to_device_request.since is not None:
+            # We've already validated this is an int.
+            since_stream_id = int(to_device_request.since)
+
+            if to_token.to_device_key < since_stream_id:
+                # The since token is ahead of our current token, so we return an
+                # empty response.
+                logger.warning(
+                    "Got to-device.since from the future. since token: %r is ahead of our current to_device stream position: %r",
+                    since_stream_id,
+                    to_token.to_device_key,
+                )
+                return SlidingSyncResult.Extensions.ToDeviceExtension(
+                    next_batch=to_device_request.since,
+                    events=[],
+                )
+
+            # Delete everything before the given since token, as we know the
+            # device must have received them.
+            deleted = await self.store.delete_messages_for_device(
+                user_id=user_id,
+                device_id=device_id,
+                up_to_stream_id=since_stream_id,
+            )
+
+            logger.debug(
+                "Deleted %d to-device messages up to %d for %s",
+                deleted,
+                since_stream_id,
+                user_id,
+            )
+
+        messages, stream_id = await self.store.get_messages_for_device(
+            user_id=user_id,
+            device_id=device_id,
+            from_stream_id=since_stream_id,
+            to_stream_id=to_token.to_device_key,
+            limit=min(to_device_request.limit, 100),  # Limit to at most 100 events
+        )
+
+        return SlidingSyncResult.Extensions.ToDeviceExtension(
+            next_batch=f"{stream_id}",
+            events=messages,
+        )
+
+    @trace
+    async def get_e2ee_extension_response(
+        self,
+        sync_config: SlidingSyncConfig,
+        e2ee_request: SlidingSyncConfig.Extensions.E2eeExtension,
+        to_token: StreamToken,
+        from_token: Optional[SlidingSyncStreamToken],
+    ) -> Optional[SlidingSyncResult.Extensions.E2eeExtension]:
+        """Handle E2EE device extension (MSC3884)
+
+        Args:
+            sync_config: Sync configuration
+            e2ee_request: The e2ee extension from the request
+            to_token: The point in the stream to sync up to.
+            from_token: The point in the stream to sync from.
+        """
+        user_id = sync_config.user.to_string()
+        device_id = sync_config.requester.device_id
+
+        # Skip if the extension is not enabled
+        if not e2ee_request.enabled:
+            return None
+
+        device_list_updates: Optional[DeviceListUpdates] = None
+        if from_token is not None:
+            # TODO: This should take into account the `from_token` and `to_token`
+            device_list_updates = await self.device_handler.get_user_ids_changed(
+                user_id=user_id,
+                from_token=from_token.stream_token,
+            )
+
+        device_one_time_keys_count: Mapping[str, int] = {}
+        device_unused_fallback_key_types: Sequence[str] = []
+        if device_id:
+            # TODO: We should have a way to let clients differentiate between the states of:
+            #   * no change in OTK count since the provided since token
+            #   * the server has zero OTKs left for this device
+            #  Spec issue: https://github.com/matrix-org/matrix-doc/issues/3298
+            device_one_time_keys_count = await self.store.count_e2e_one_time_keys(
+                user_id, device_id
+            )
+            device_unused_fallback_key_types = (
+                await self.store.get_e2e_unused_fallback_key_types(user_id, device_id)
+            )
+
+        return SlidingSyncResult.Extensions.E2eeExtension(
+            device_list_updates=device_list_updates,
+            device_one_time_keys_count=device_one_time_keys_count,
+            device_unused_fallback_key_types=device_unused_fallback_key_types,
+        )
+
+    @trace
+    async def get_account_data_extension_response(
+        self,
+        sync_config: SlidingSyncConfig,
+        actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
+        actual_room_ids: Set[str],
+        account_data_request: SlidingSyncConfig.Extensions.AccountDataExtension,
+        to_token: StreamToken,
+        from_token: Optional[SlidingSyncStreamToken],
+    ) -> Optional[SlidingSyncResult.Extensions.AccountDataExtension]:
+        """Handle Account Data extension (MSC3959)
+
+        Args:
+            sync_config: Sync configuration
+            actual_lists: Sliding window API. A map of list key to list results in the
+                Sliding Sync response.
+            actual_room_ids: The actual room IDs in the the Sliding Sync response.
+            account_data_request: The account_data extension from the request
+            to_token: The point in the stream to sync up to.
+            from_token: The point in the stream to sync from.
+        """
+        user_id = sync_config.user.to_string()
+
+        # Skip if the extension is not enabled
+        if not account_data_request.enabled:
+            return None
+
+        global_account_data_map: Mapping[str, JsonMapping] = {}
+        if from_token is not None:
+            # TODO: This should take into account the `from_token` and `to_token`
+            global_account_data_map = (
+                await self.store.get_updated_global_account_data_for_user(
+                    user_id, from_token.stream_token.account_data_key
+                )
+            )
+
+            have_push_rules_changed = await self.store.have_push_rules_changed_for_user(
+                user_id, from_token.stream_token.push_rules_key
+            )
+            if have_push_rules_changed:
+                global_account_data_map = dict(global_account_data_map)
+                # TODO: This should take into account the `from_token` and `to_token`
+                global_account_data_map[AccountDataTypes.PUSH_RULES] = (
+                    await self.push_rules_handler.push_rules_for_user(sync_config.user)
+                )
+        else:
+            # TODO: This should take into account the `to_token`
+            all_global_account_data = await self.store.get_global_account_data_for_user(
+                user_id
+            )
+
+            global_account_data_map = dict(all_global_account_data)
+            # TODO: This should take into account the  `to_token`
+            global_account_data_map[AccountDataTypes.PUSH_RULES] = (
+                await self.push_rules_handler.push_rules_for_user(sync_config.user)
+            )
+
+        # Fetch room account data
+        account_data_by_room_map: Mapping[str, Mapping[str, JsonMapping]] = {}
+        relevant_room_ids = self.find_relevant_room_ids_for_extension(
+            requested_lists=account_data_request.lists,
+            requested_room_ids=account_data_request.rooms,
+            actual_lists=actual_lists,
+            actual_room_ids=actual_room_ids,
+        )
+        if len(relevant_room_ids) > 0:
+            if from_token is not None:
+                # TODO: This should take into account the `from_token` and `to_token`
+                account_data_by_room_map = (
+                    await self.store.get_updated_room_account_data_for_user(
+                        user_id, from_token.stream_token.account_data_key
+                    )
+                )
+            else:
+                # TODO: This should take into account the `to_token`
+                account_data_by_room_map = (
+                    await self.store.get_room_account_data_for_user(user_id)
+                )
+
+        # Filter down to the relevant rooms
+        account_data_by_room_map = {
+            room_id: account_data_map
+            for room_id, account_data_map in account_data_by_room_map.items()
+            if room_id in relevant_room_ids
+        }
+
+        return SlidingSyncResult.Extensions.AccountDataExtension(
+            global_account_data_map=global_account_data_map,
+            account_data_by_room_map=account_data_by_room_map,
+        )
+
+    @trace
+    async def get_receipts_extension_response(
+        self,
+        sync_config: SlidingSyncConfig,
+        previous_connection_state: "PerConnectionState",
+        new_connection_state: "MutablePerConnectionState",
+        actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
+        actual_room_ids: Set[str],
+        actual_room_response_map: Dict[str, SlidingSyncResult.RoomResult],
+        receipts_request: SlidingSyncConfig.Extensions.ReceiptsExtension,
+        to_token: StreamToken,
+        from_token: Optional[SlidingSyncStreamToken],
+    ) -> Optional[SlidingSyncResult.Extensions.ReceiptsExtension]:
+        """Handle Receipts extension (MSC3960)
+
+        Args:
+            sync_config: Sync configuration
+            previous_connection_state: The current per-connection state
+            new_connection_state: A mutable copy of the per-connection
+                state, used to record updates to the state.
+            actual_lists: Sliding window API. A map of list key to list results in the
+                Sliding Sync response.
+            actual_room_ids: The actual room IDs in the the Sliding Sync response.
+            actual_room_response_map: A map of room ID to room results in the the
+                Sliding Sync response.
+            account_data_request: The account_data extension from the request
+            to_token: The point in the stream to sync up to.
+            from_token: The point in the stream to sync from.
+        """
+        # Skip if the extension is not enabled
+        if not receipts_request.enabled:
+            return None
+
+        relevant_room_ids = self.find_relevant_room_ids_for_extension(
+            requested_lists=receipts_request.lists,
+            requested_room_ids=receipts_request.rooms,
+            actual_lists=actual_lists,
+            actual_room_ids=actual_room_ids,
+        )
+
+        room_id_to_receipt_map: Dict[str, JsonMapping] = {}
+        if len(relevant_room_ids) > 0:
+            # We need to handle the different cases depending on if we have sent
+            # down receipts previously or not, so we split the relevant rooms
+            # up into different collections based on status.
+            live_rooms = set()
+            previously_rooms: Dict[str, MultiWriterStreamToken] = {}
+            initial_rooms = set()
+
+            for room_id in relevant_room_ids:
+                if not from_token:
+                    initial_rooms.add(room_id)
+                    continue
+
+                # If we're sending down the room from scratch again for some reason, we
+                # should always resend the receipts as well (regardless of if
+                # we've sent them down before). This is to mimic the behaviour
+                # of what happens on initial sync, where you get a chunk of
+                # timeline with all of the corresponding receipts for the events in the timeline.
+                room_result = actual_room_response_map.get(room_id)
+                if room_result is not None and room_result.initial:
+                    initial_rooms.add(room_id)
+                    continue
+
+                room_status = previous_connection_state.receipts.have_sent_room(room_id)
+                if room_status.status == HaveSentRoomFlag.LIVE:
+                    live_rooms.add(room_id)
+                elif room_status.status == HaveSentRoomFlag.PREVIOUSLY:
+                    assert room_status.last_token is not None
+                    previously_rooms[room_id] = room_status.last_token
+                elif room_status.status == HaveSentRoomFlag.NEVER:
+                    initial_rooms.add(room_id)
+                else:
+                    assert_never(room_status.status)
+
+            # The set of receipts that we fetched. Private receipts need to be
+            # filtered out before returning.
+            fetched_receipts = []
+
+            # For live rooms we just fetch all receipts in those rooms since the
+            # `since` token.
+            if live_rooms:
+                assert from_token is not None
+                receipts = await self.store.get_linearized_receipts_for_rooms(
+                    room_ids=live_rooms,
+                    from_key=from_token.stream_token.receipt_key,
+                    to_key=to_token.receipt_key,
+                )
+                fetched_receipts.extend(receipts)
+
+            # For rooms we've previously sent down, but aren't up to date, we
+            # need to use the from token from the room status.
+            if previously_rooms:
+                for room_id, receipt_token in previously_rooms.items():
+                    # TODO: Limit the number of receipts we're about to send down
+                    # for the room, if its too many we should TODO
+                    previously_receipts = (
+                        await self.store.get_linearized_receipts_for_room(
+                            room_id=room_id,
+                            from_key=receipt_token,
+                            to_key=to_token.receipt_key,
+                        )
+                    )
+                    fetched_receipts.extend(previously_receipts)
+
+            # For rooms we haven't previously sent down, we could send all receipts
+            # from that room but we only want to include receipts for events
+            # in the timeline to avoid bloating and blowing up the sync response
+            # as the number of users in the room increases. (this behavior is part of the spec)
+            initial_rooms_and_event_ids = [
+                (room_id, event.event_id)
+                for room_id in initial_rooms
+                if room_id in actual_room_response_map
+                for event in actual_room_response_map[room_id].timeline_events
+            ]
+            if initial_rooms_and_event_ids:
+                initial_receipts = await self.store.get_linearized_receipts_for_events(
+                    room_and_event_ids=initial_rooms_and_event_ids,
+                )
+                fetched_receipts.extend(initial_receipts)
+
+            fetched_receipts = ReceiptEventSource.filter_out_private_receipts(
+                fetched_receipts, sync_config.user.to_string()
+            )
+
+            for receipt in fetched_receipts:
+                # These fields should exist for every receipt
+                room_id = receipt["room_id"]
+                type = receipt["type"]
+                content = receipt["content"]
+
+                room_id_to_receipt_map[room_id] = {"type": type, "content": content}
+
+        # Now we update the per-connection state to track which receipts we have
+        # and haven't sent down.
+        new_connection_state.receipts.record_sent_rooms(relevant_room_ids)
+
+        if from_token:
+            # Now find the set of rooms that may have receipts that we're not sending
+            # down. We only need to check rooms that we have previously returned
+            # receipts for (in `previous_connection_state`) because we only care about
+            # updating `LIVE` rooms to `PREVIOUSLY`. The `PREVIOUSLY` rooms will just
+            # stay pointing at their previous position so we don't need to waste time
+            # checking those and since we default to `NEVER`, rooms that were `NEVER`
+            # sent before don't need to be recorded as we'll handle them correctly when
+            # they come into range for the first time.
+            rooms_no_receipts = [
+                room_id
+                for room_id, room_status in previous_connection_state.receipts._statuses.items()
+                if room_status.status == HaveSentRoomFlag.LIVE
+                and room_id not in relevant_room_ids
+            ]
+            changed_rooms = await self.store.get_rooms_with_receipts_between(
+                rooms_no_receipts,
+                from_key=from_token.stream_token.receipt_key,
+                to_key=to_token.receipt_key,
+            )
+            new_connection_state.receipts.record_unsent_rooms(
+                changed_rooms, from_token.stream_token.receipt_key
+            )
+
+        return SlidingSyncResult.Extensions.ReceiptsExtension(
+            room_id_to_receipt_map=room_id_to_receipt_map,
+        )
+
+    async def get_typing_extension_response(
+        self,
+        sync_config: SlidingSyncConfig,
+        actual_lists: Dict[str, SlidingSyncResult.SlidingWindowList],
+        actual_room_ids: Set[str],
+        actual_room_response_map: Dict[str, SlidingSyncResult.RoomResult],
+        typing_request: SlidingSyncConfig.Extensions.TypingExtension,
+        to_token: StreamToken,
+        from_token: Optional[SlidingSyncStreamToken],
+    ) -> Optional[SlidingSyncResult.Extensions.TypingExtension]:
+        """Handle Typing Notification extension (MSC3961)
+
+        Args:
+            sync_config: Sync configuration
+            actual_lists: Sliding window API. A map of list key to list results in the
+                Sliding Sync response.
+            actual_room_ids: The actual room IDs in the the Sliding Sync response.
+            actual_room_response_map: A map of room ID to room results in the the
+                Sliding Sync response.
+            account_data_request: The account_data extension from the request
+            to_token: The point in the stream to sync up to.
+            from_token: The point in the stream to sync from.
+        """
+        # Skip if the extension is not enabled
+        if not typing_request.enabled:
+            return None
+
+        relevant_room_ids = self.find_relevant_room_ids_for_extension(
+            requested_lists=typing_request.lists,
+            requested_room_ids=typing_request.rooms,
+            actual_lists=actual_lists,
+            actual_room_ids=actual_room_ids,
+        )
+
+        room_id_to_typing_map: Dict[str, JsonMapping] = {}
+        if len(relevant_room_ids) > 0:
+            # Note: We don't need to take connection tracking into account for typing
+            # notifications because they'll get anything still relevant and hasn't timed
+            # out when the room comes into range. We consider the gap where the room
+            # fell out of range, as long enough for any typing notifications to have
+            # timed out (it's not worth the 30 seconds of data we may have missed).
+            typing_source = self.event_sources.sources.typing
+            typing_notifications, _ = await typing_source.get_new_events(
+                user=sync_config.user,
+                from_key=(from_token.stream_token.typing_key if from_token else 0),
+                to_key=to_token.typing_key,
+                # This is a dummy value and isn't used in the function
+                limit=0,
+                room_ids=relevant_room_ids,
+                is_guest=False,
+            )
+
+            for typing_notification in typing_notifications:
+                # These fields should exist for every typing notification
+                room_id = typing_notification["room_id"]
+                type = typing_notification["type"]
+                content = typing_notification["content"]
+
+                room_id_to_typing_map[room_id] = {"type": type, "content": content}
+
+        return SlidingSyncResult.Extensions.TypingExtension(
+            room_id_to_typing_map=room_id_to_typing_map,
+        )

synapse/handlers/sliding_sync/store.py

@@ -0,0 +1,123 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2023 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+import logging
+from typing import TYPE_CHECKING, Optional
+
+import attr
+
+from synapse.logging.opentracing import trace
+from synapse.storage.databases.main import DataStore
+from synapse.types import SlidingSyncStreamToken
+from synapse.types.handlers.sliding_sync import (
+    MutablePerConnectionState,
+    PerConnectionState,
+    SlidingSyncConfig,
+)
+
+if TYPE_CHECKING:
+    pass
+
+logger = logging.getLogger(__name__)
+
+
+@attr.s(auto_attribs=True)
+class SlidingSyncConnectionStore:
+    """In-memory store of per-connection state, including what rooms we have
+    previously sent down a sliding sync connection.
+
+    Note: This is NOT safe to run in a worker setup because connection positions will
+    point to different sets of rooms on different workers. e.g. for the same connection,
+    a connection position of 5 might have totally different states on worker A and
+    worker B.
+
+    One complication that we need to deal with here is needing to handle requests being
+    resent, i.e. if we sent down a room in a response that the client received, we must
+    consider the room *not* sent when we get the request again.
+
+    This is handled by using an integer "token", which is returned to the client
+    as part of the sync token. For each connection we store a mapping from
+    tokens to the room states, and create a new entry when we send down new
+    rooms.
+
+    Note that for any given sliding sync connection we will only store a maximum
+    of two different tokens: the previous token from the request and a new token
+    sent in the response. When we receive a request with a given token, we then
+    clear out all other entries with a different token.
+
+    Attributes:
+        _connections: Mapping from `(user_id, conn_id)` to mapping of `token`
+            to mapping of room ID to `HaveSentRoom`.
+    """
+
+    store: "DataStore"
+
+    async def get_per_connection_state(
+        self,
+        sync_config: SlidingSyncConfig,
+        from_token: Optional[SlidingSyncStreamToken],
+    ) -> PerConnectionState:
+        """Fetch the per-connection state for the token.
+
+        Raises:
+            SlidingSyncUnknownPosition if the connection_token is unknown
+        """
+        if from_token is None or from_token.connection_position == 0:
+            return PerConnectionState()
+
+        conn_id = sync_config.conn_id or ""
+
+        device_id = sync_config.requester.device_id
+        assert device_id is not None
+
+        return await self.store.get_per_connection_state(
+            sync_config.user.to_string(),
+            device_id,
+            conn_id,
+            from_token.connection_position,
+        )
+
+    @trace
+    async def record_new_state(
+        self,
+        sync_config: SlidingSyncConfig,
+        from_token: Optional[SlidingSyncStreamToken],
+        new_connection_state: MutablePerConnectionState,
+    ) -> int:
+        """Record updated per-connection state, returning the connection
+        position associated with the new state.
+        If there are no changes to the state this may return the same token as
+        the existing per-connection state.
+        """
+        if not new_connection_state.has_updates():
+            if from_token is not None:
+                return from_token.connection_position
+            else:
+                return 0
+
+        if from_token is not None and from_token.connection_position == 0:
+            from_token = None
+
+        conn_id = sync_config.conn_id or ""
+
+        device_id = sync_config.requester.device_id
+        assert device_id is not None
+
+        return await self.store.persist_per_connection_state(
+            sync_config.user.to_string(),
+            device_id,
+            conn_id,
+            from_token.connection_position if from_token else None,
+            new_connection_state,
+        )

synapse/handlers/stats.py

@@ -293,7 +293,9 @@ class StatsHandler:
                     "history_visibility"
                 )
             elif delta.event_type == EventTypes.RoomEncryption:
-                room_state["encryption"] = event_content.get("algorithm")
+                room_state["encryption"] = event_content.get(
+                    EventContentFields.ENCRYPTION_ALGORITHM
+                )
             elif delta.event_type == EventTypes.Name:
                 room_state["name"] = event_content.get("name")
             elif delta.event_type == EventTypes.Topic:

synapse/handlers/sync.py

@@ -43,6 +43,7 @@ from prometheus_client import Counter
 
 from synapse.api.constants import (
     AccountDataTypes,
+    Direction,
     EventContentFields,
     EventTypes,
     JoinRules,
@@ -64,6 +65,7 @@ from synapse.logging.opentracing import (
 )
 from synapse.storage.databases.main.event_push_actions import RoomNotifCounts
 from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
+from synapse.storage.databases.main.stream import PaginateFunction
 from synapse.storage.roommember import MemberSummary
 from synapse.types import (
     DeviceListUpdates,
@@ -84,7 +86,7 @@ from synapse.util.async_helpers import concurrently_execute
 from synapse.util.caches.expiringcache import ExpiringCache
 from synapse.util.caches.lrucache import LruCache
 from synapse.util.caches.response_cache import ResponseCache, ResponseCacheContext
-from synapse.util.metrics import Measure, measure_func
+from synapse.util.metrics import Measure
 from synapse.visibility import filter_events_for_client
 
 if TYPE_CHECKING:
@@ -879,22 +881,49 @@ class SyncHandler:
                 since_key = since_token.room_key
 
             while limited and len(recents) < timeline_limit and max_repeat:
-                # If we have a since_key then we are trying to get any events
-                # that have happened since `since_key` up to `end_key`, so we
-                # can just use `get_room_events_stream_for_room`.
-                # Otherwise, we want to return the last N events in the room
-                # in topological ordering.
-                if since_key:
-                    events, end_key = await self.store.get_room_events_stream_for_room(
-                        room_id,
-                        limit=load_limit + 1,
-                        from_key=since_key,
-                        to_key=end_key,
-                    )
-                else:
-                    events, end_key = await self.store.get_recent_events_for_room(
-                        room_id, limit=load_limit + 1, end_token=end_key
-                    )
+                # For initial `/sync`, we want to view a historical section of the
+                # timeline; to fetch events by `topological_ordering` (best
+                # representation of the room DAG as others were seeing it at the time).
+                # This also aligns with the order that `/messages` returns events in.
+                #
+                # For incremental `/sync`, we want to get all updates for rooms since
+                # the last `/sync` (regardless if those updates arrived late or happened
+                # a while ago in the past); to fetch events by `stream_ordering` (in the
+                # order they were received by the server).
+                #
+                # Relevant spec issue: https://github.com/matrix-org/matrix-spec/issues/1917
+                #
+                # FIXME: Using workaround for mypy,
+                # https://github.com/python/mypy/issues/10740#issuecomment-1997047277 and
+                # https://github.com/python/mypy/issues/17479
+                paginate_room_events_by_topological_ordering: PaginateFunction = (
+                    self.store.paginate_room_events_by_topological_ordering
+                )
+                paginate_room_events_by_stream_ordering: PaginateFunction = (
+                    self.store.paginate_room_events_by_stream_ordering
+                )
+                pagination_method: PaginateFunction = (
+                    # Use `topographical_ordering` for historical events
+                    paginate_room_events_by_topological_ordering
+                    if since_key is None
+                    # Use `stream_ordering` for updates
+                    else paginate_room_events_by_stream_ordering
+                )
+                events, end_key = await pagination_method(
+                    room_id=room_id,
+                    # The bounds are reversed so we can paginate backwards
+                    # (from newer to older events) starting at to_bound.
+                    # This ensures we fill the `limit` with the newest events first,
+                    from_key=end_key,
+                    to_key=since_key,
+                    direction=Direction.BACKWARDS,
+                    # We add one so we can determine if there are enough events to saturate
+                    # the limit or not (see `limited`)
+                    limit=load_limit + 1,
+                )
+                # We want to return the events in ascending order (the last event is the
+                # most recent).
+                events.reverse()
 
                 log_kv({"loaded_recents": len(events)})
 
@@ -1750,8 +1779,15 @@ class SyncHandler:
                     )
 
                 if include_device_list_updates:
-                    device_lists = await self._generate_sync_entry_for_device_list(
-                        sync_result_builder,
+                    # include_device_list_updates can only be True if we have a
+                    # since token.
+                    assert since_token is not None
+
+                    device_lists = await self._device_handler.generate_sync_entry_for_device_list(
+                        user_id=user_id,
+                        since_token=since_token,
+                        now_token=sync_result_builder.now_token,
+                        joined_room_ids=sync_result_builder.joined_room_ids,
                         newly_joined_rooms=newly_joined_rooms,
                         newly_joined_or_invited_or_knocked_users=newly_joined_or_invited_or_knocked_users,
                         newly_left_rooms=newly_left_rooms,
@@ -1863,8 +1899,14 @@ class SyncHandler:
                 newly_left_users,
             ) = sync_result_builder.calculate_user_changes()
 
-            device_lists = await self._generate_sync_entry_for_device_list(
-                sync_result_builder,
+            # include_device_list_updates can only be True if we have a
+            # since token.
+            assert since_token is not None
+            device_lists = await self._device_handler.generate_sync_entry_for_device_list(
+                user_id=user_id,
+                since_token=since_token,
+                now_token=sync_result_builder.now_token,
+                joined_room_ids=sync_result_builder.joined_room_ids,
                 newly_joined_rooms=newly_joined_rooms,
                 newly_joined_or_invited_or_knocked_users=newly_joined_or_invited_or_knocked_users,
                 newly_left_rooms=newly_left_rooms,
@@ -2041,94 +2083,6 @@ class SyncHandler:
 
         return sync_result_builder
 
-    @measure_func("_generate_sync_entry_for_device_list")
-    async def _generate_sync_entry_for_device_list(
-        self,
-        sync_result_builder: "SyncResultBuilder",
-        newly_joined_rooms: AbstractSet[str],
-        newly_joined_or_invited_or_knocked_users: AbstractSet[str],
-        newly_left_rooms: AbstractSet[str],
-        newly_left_users: AbstractSet[str],
-    ) -> DeviceListUpdates:
-        """Generate the DeviceListUpdates section of sync
-
-        Args:
-            sync_result_builder
-            newly_joined_rooms: Set of rooms user has joined since previous sync
-            newly_joined_or_invited_or_knocked_users: Set of users that have joined,
-                been invited to a room or are knocking on a room since
-                previous sync.
-            newly_left_rooms: Set of rooms user has left since previous sync
-            newly_left_users: Set of users that have left a room we're in since
-                previous sync
-        """
-
-        user_id = sync_result_builder.sync_config.user.to_string()
-        since_token = sync_result_builder.since_token
-        assert since_token is not None
-
-        # Take a copy since these fields will be mutated later.
-        newly_joined_or_invited_or_knocked_users = set(
-            newly_joined_or_invited_or_knocked_users
-        )
-        newly_left_users = set(newly_left_users)
-
-        # We want to figure out what user IDs the client should refetch
-        # device keys for, and which users we aren't going to track changes
-        # for anymore.
-        #
-        # For the first step we check:
-        #   a. if any users we share a room with have updated their devices,
-        #      and
-        #   b. we also check if we've joined any new rooms, or if a user has
-        #      joined a room we're in.
-        #
-        # For the second step we just find any users we no longer share a
-        # room with by looking at all users that have left a room plus users
-        # that were in a room we've left.
-
-        users_that_have_changed = set()
-
-        joined_room_ids = sync_result_builder.joined_room_ids
-
-        # Step 1a, check for changes in devices of users we share a room
-        # with
-        users_that_have_changed = (
-            await self._device_handler.get_device_changes_in_shared_rooms(
-                user_id,
-                joined_room_ids,
-                from_token=since_token,
-                now_token=sync_result_builder.now_token,
-            )
-        )
-
-        # Step 1b, check for newly joined rooms
-        for room_id in newly_joined_rooms:
-            joined_users = await self.store.get_users_in_room(room_id)
-            newly_joined_or_invited_or_knocked_users.update(joined_users)
-
-        # TODO: Check that these users are actually new, i.e. either they
-        # weren't in the previous sync *or* they left and rejoined.
-        users_that_have_changed.update(newly_joined_or_invited_or_knocked_users)
-
-        user_signatures_changed = await self.store.get_users_whose_signatures_changed(
-            user_id, since_token.device_list_key
-        )
-        users_that_have_changed.update(user_signatures_changed)
-
-        # Now find users that we no longer track
-        for room_id in newly_left_rooms:
-            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.
-        left_users_rooms = await self.store.get_rooms_for_users(newly_left_users)
-        for user_id, entries in left_users_rooms.items():
-            if any(rid in joined_room_ids for rid in entries):
-                newly_left_users.discard(user_id)
-
-        return DeviceListUpdates(changed=users_that_have_changed, left=newly_left_users)
-
     @trace
     async def _generate_sync_entry_for_to_device(
         self, sync_result_builder: "SyncResultBuilder"
@@ -2270,7 +2224,11 @@ class SyncHandler:
             user=user,
             from_key=presence_key,
             is_guest=sync_config.is_guest,
-            include_offline=include_offline,
+            include_offline=(
+                True
+                if self.hs_config.server.presence_include_offline_users_on_sync
+                else include_offline
+            ),
         )
         assert presence_key
         sync_result_builder.now_token = now_token.copy_and_replace(
@@ -2637,9 +2595,10 @@ class SyncHandler:
         # a "gap" in the timeline, as described by the spec for /sync.
         room_to_events = await self.store.get_room_events_stream_for_rooms(
             room_ids=sync_result_builder.joined_room_ids,
-            from_key=since_token.room_key,
-            to_key=now_token.room_key,
+            from_key=now_token.room_key,
+            to_key=since_token.room_key,
             limit=timeline_limit + 1,
+            direction=Direction.BACKWARDS,
         )
 
         # We loop through all room ids, even if there are no new events, in case
@@ -2650,6 +2609,9 @@ class SyncHandler:
             newly_joined = room_id in newly_joined_rooms
             if room_entry:
                 events, start_key = room_entry
+                # We want to return the events in ascending order (the last event is the
+                # most recent).
+                events.reverse()
 
                 prev_batch_token = now_token.copy_and_replace(
                     StreamKeyType.ROOM, start_key

synapse/handlers/typing.py

@@ -565,7 +565,12 @@ class TypingNotificationEventSource(EventSource[int, JsonMapping]):
         room_ids: Iterable[str],
         is_guest: bool,
         explicit_room_id: Optional[str] = None,
+        to_key: Optional[int] = None,
     ) -> Tuple[List[JsonMapping], int]:
+        """
+        Find typing notifications for given rooms (> `from_token` and <= `to_token`)
+        """
+
         with Measure(self.clock, "typing.get_new_events"):
             from_key = int(from_key)
             handler = self.get_typing_handler()
@@ -574,7 +579,9 @@ class TypingNotificationEventSource(EventSource[int, JsonMapping]):
             for room_id in room_ids:
                 if room_id not in handler._room_serials:
                     continue
-                if handler._room_serials[room_id] <= from_key:
+                if handler._room_serials[room_id] <= from_key or (
+                    to_key is not None and handler._room_serials[room_id] > to_key
+                ):
                     continue
 
                 events.append(self._make_event_for(room_id))

synapse/http/client.py

@@ -1057,11 +1057,11 @@ class _MultipartParserProtocol(protocol.Protocol):
         if not self.parser:
 
             def on_header_field(data: bytes, start: int, end: int) -> None:
-                if data[start:end] == b"Location":
+                if data[start:end].lower() == b"location":
                     self.has_redirect = True
-                if data[start:end] == b"Content-Disposition":
+                if data[start:end].lower() == b"content-disposition":
                     self.in_disposition = True
-                if data[start:end] == b"Content-Type":
+                if data[start:end].lower() == b"content-type":
                     self.in_content_type = True
 
             def on_header_value(data: bytes, start: int, end: int) -> None:
@@ -1088,7 +1088,6 @@ class _MultipartParserProtocol(protocol.Protocol):
                     return
                 # otherwise we are in the file part
                 else:
-                    logger.info("Writing multipart file data to stream")
                     try:
                         self.stream.write(data[start:end])
                     except Exception as e:

synapse/http/proxy.py

@@ -62,6 +62,15 @@ HOP_BY_HOP_HEADERS = {
     "Upgrade",
 }
 
+if hasattr(Headers, "_canonicalNameCaps"):
+    # Twisted < 24.7.0rc1
+    _canonicalHeaderName = Headers()._canonicalNameCaps  # type: ignore[attr-defined]
+else:
+    # Twisted >= 24.7.0rc1
+    # But note that `_encodeName` still exists on prior versions,
+    # it just encodes differently
+    _canonicalHeaderName = Headers()._encodeName
+
 
 def parse_connection_header_value(
     connection_header_value: Optional[bytes],
@@ -85,11 +94,10 @@ def parse_connection_header_value(
         The set of header names that should not be copied over from the remote response.
         The keys are capitalized in canonical capitalization.
     """
-    headers = Headers()
     extra_headers_to_remove: Set[str] = set()
     if connection_header_value:
         extra_headers_to_remove = {
-            headers._canonicalNameCaps(connection_option.strip()).decode("ascii")
+            _canonicalHeaderName(connection_option.strip()).decode("ascii")
             for connection_option in connection_header_value.split(b",")
         }
 

synapse/http/site.py

@@ -658,7 +658,7 @@ class SynapseSite(ProxySite):
         )
 
         self.site_tag = site_tag
-        self.reactor = reactor
+        self.reactor: ISynapseReactor = reactor
 
         assert config.http_options is not None
         proxied = config.http_options.x_forwarded
@@ -683,7 +683,7 @@ class SynapseSite(ProxySite):
         self.access_logger = logging.getLogger(logger_name)
         self.server_version_string = server_version_string.encode("ascii")
 
-    def log(self, request: SynapseRequest) -> None:
+    def log(self, request: SynapseRequest) -> None:  # type: ignore[override]
         pass
 
 

synapse/media/_base.py

@@ -28,6 +28,7 @@ from types import TracebackType
 from typing import (
     TYPE_CHECKING,
     Awaitable,
+    BinaryIO,
     Dict,
     Generator,
     List,
@@ -37,19 +38,28 @@ from typing import (
 )
 
 import attr
+from zope.interface import implementer
 
+from twisted.internet import interfaces
+from twisted.internet.defer import Deferred
 from twisted.internet.interfaces import IConsumer
-from twisted.protocols.basic import FileSender
+from twisted.python.failure import Failure
 from twisted.web.server import Request
 
 from synapse.api.errors import Codes, cs_error
 from synapse.http.server import finish_request, respond_with_json
 from synapse.http.site import SynapseRequest
-from synapse.logging.context import make_deferred_yieldable
+from synapse.logging.context import (
+    defer_to_threadpool,
+    make_deferred_yieldable,
+    run_in_background,
+)
 from synapse.util import Clock
+from synapse.util.async_helpers import DeferredEvent
 from synapse.util.stringutils import is_ascii
 
 if TYPE_CHECKING:
+    from synapse.server import HomeServer
     from synapse.storage.databases.main.media_repository import LocalMedia
 
 
@@ -122,6 +132,7 @@ def respond_404(request: SynapseRequest) -> None:
 
 
 async def respond_with_file(
+    hs: "HomeServer",
     request: SynapseRequest,
     media_type: str,
     file_path: str,
@@ -138,7 +149,7 @@ async def respond_with_file(
         add_file_headers(request, media_type, file_size, upload_name)
 
         with open(file_path, "rb") as f:
-            await make_deferred_yieldable(FileSender().beginFileTransfer(f, request))
+            await ThreadedFileSender(hs).beginFileTransfer(f, request)
 
         finish_request(request)
     else:
@@ -601,3 +612,151 @@ def _parseparam(s: bytes) -> Generator[bytes, None, None]:
         f = s[:end]
         yield f.strip()
         s = s[end:]
+
+
+@implementer(interfaces.IPushProducer)
+class ThreadedFileSender:
+    """
+    A producer that sends the contents of a file to a consumer, reading from the
+    file on a thread.
+
+    This works by having a loop in a threadpool repeatedly reading from the
+    file, until the consumer pauses the producer. There is then a loop in the
+    main thread that waits until the consumer resumes the producer and then
+    starts reading in the threadpool again.
+
+    This is done to ensure that we're never waiting in the threadpool, as
+    otherwise its easy to starve it of threads.
+    """
+
+    # How much data to read in one go.
+    CHUNK_SIZE = 2**14
+
+    # How long we wait for the consumer to be ready again before aborting the
+    # read.
+    TIMEOUT_SECONDS = 90.0
+
+    def __init__(self, hs: "HomeServer") -> None:
+        self.reactor = hs.get_reactor()
+        self.thread_pool = hs.get_media_sender_thread_pool()
+
+        self.file: Optional[BinaryIO] = None
+        self.deferred: "Deferred[None]" = Deferred()
+        self.consumer: Optional[interfaces.IConsumer] = None
+
+        # Signals if the thread should keep reading/sending data. Set means
+        # continue, clear means pause.
+        self.wakeup_event = DeferredEvent(self.reactor)
+
+        # Signals if the thread should terminate, e.g. because the consumer has
+        # gone away.
+        self.stop_writing = False
+
+    def beginFileTransfer(
+        self, file: BinaryIO, consumer: interfaces.IConsumer
+    ) -> "Deferred[None]":
+        """
+        Begin transferring a file
+        """
+        self.file = file
+        self.consumer = consumer
+
+        self.consumer.registerProducer(self, True)
+
+        # We set the wakeup signal as we should start producing immediately.
+        self.wakeup_event.set()
+        run_in_background(self.start_read_loop)
+
+        return make_deferred_yieldable(self.deferred)
+
+    def resumeProducing(self) -> None:
+        """interfaces.IPushProducer"""
+        self.wakeup_event.set()
+
+    def pauseProducing(self) -> None:
+        """interfaces.IPushProducer"""
+        self.wakeup_event.clear()
+
+    def stopProducing(self) -> None:
+        """interfaces.IPushProducer"""
+
+        # Unregister the consumer so we don't try and interact with it again.
+        if self.consumer:
+            self.consumer.unregisterProducer()
+
+        self.consumer = None
+
+        # Terminate the loop.
+        self.stop_writing = True
+        self.wakeup_event.set()
+
+        if not self.deferred.called:
+            self.deferred.errback(Exception("Consumer asked us to stop producing"))
+
+    async def start_read_loop(self) -> None:
+        """This is the loop that drives reading/writing"""
+        try:
+            while not self.stop_writing:
+                # Start the loop in the threadpool to read data.
+                more_data = await defer_to_threadpool(
+                    self.reactor, self.thread_pool, self._on_thread_read_loop
+                )
+                if not more_data:
+                    # Reached EOF, we can just return.
+                    return
+
+                if not self.wakeup_event.is_set():
+                    ret = await self.wakeup_event.wait(self.TIMEOUT_SECONDS)
+                    if not ret:
+                        raise Exception("Timed out waiting to resume")
+        except Exception:
+            self._error(Failure())
+        finally:
+            self._finish()
+
+    def _on_thread_read_loop(self) -> bool:
+        """This is the loop that happens on a thread.
+
+        Returns:
+            Whether there is more data to send.
+        """
+
+        while not self.stop_writing and self.wakeup_event.is_set():
+            # The file should always have been set before we get here.
+            assert self.file is not None
+
+            chunk = self.file.read(self.CHUNK_SIZE)
+            if not chunk:
+                return False
+
+            self.reactor.callFromThread(self._write, chunk)
+
+        return True
+
+    def _write(self, chunk: bytes) -> None:
+        """Called from the thread to write a chunk of data"""
+        if self.consumer:
+            self.consumer.write(chunk)
+
+    def _error(self, failure: Failure) -> None:
+        """Called when there was a fatal error"""
+        if self.consumer:
+            self.consumer.unregisterProducer()
+            self.consumer = None
+
+        if not self.deferred.called:
+            self.deferred.errback(failure)
+
+    def _finish(self) -> None:
+        """Called when we have finished writing (either on success or
+        failure)."""
+        if self.file:
+            self.file.close()
+            self.file = None
+
+        if self.consumer:
+            self.consumer.unregisterProducer()
+            self.consumer = None
+
+        if not self.deferred.called:
+            self.deferred.callback(None)

synapse/media/media_storage.py

@@ -49,15 +49,11 @@ from zope.interface import implementer
 from twisted.internet import interfaces
 from twisted.internet.defer import Deferred
 from twisted.internet.interfaces import IConsumer
-from twisted.protocols.basic import FileSender
 
 from synapse.api.errors import NotFoundError
-from synapse.logging.context import (
-    defer_to_thread,
-    make_deferred_yieldable,
-    run_in_background,
-)
+from synapse.logging.context import defer_to_thread, run_in_background
 from synapse.logging.opentracing import start_active_span, trace, trace_with_opname
+from synapse.media._base import ThreadedFileSender
 from synapse.util import Clock
 from synapse.util.file_consumer import BackgroundFileConsumer
 
@@ -213,7 +209,7 @@ class MediaStorage:
             local_path = os.path.join(self.local_media_directory, path)
             if os.path.exists(local_path):
                 logger.debug("responding with local file %s", local_path)
-                return FileResponder(open(local_path, "rb"))
+                return FileResponder(self.hs, open(local_path, "rb"))
             logger.debug("local file %s did not exist", local_path)
 
         for provider in self.storage_providers:
@@ -336,13 +332,12 @@ class FileResponder(Responder):
             is closed when finished streaming.
     """
 
-    def __init__(self, open_file: IO):
+    def __init__(self, hs: "HomeServer", open_file: BinaryIO):
+        self.hs = hs
         self.open_file = open_file
 
     def write_to_consumer(self, consumer: IConsumer) -> Deferred:
-        return make_deferred_yieldable(
-            FileSender().beginFileTransfer(self.open_file, consumer)
-        )
+        return ThreadedFileSender(self.hs).beginFileTransfer(self.open_file, consumer)
 
     def __exit__(
         self,
@@ -549,7 +544,7 @@ class MultipartFileConsumer:
         Calculate the content length of the multipart response
         in bytes.
         """
-        if not self.length:
+        if self.length is None:
             return None
         # calculate length of json field and content-type, disposition headers
         json_field = json.dumps(self.json_field)

synapse/media/storage_provider.py

@@ -145,6 +145,7 @@ class FileStorageProviderBackend(StorageProvider):
 
     def __init__(self, hs: "HomeServer", config: str):
         self.hs = hs
+        self.reactor = hs.get_reactor()
         self.cache_directory = hs.config.media.media_store_path
         self.base_directory = config
 
@@ -165,7 +166,7 @@ class FileStorageProviderBackend(StorageProvider):
         shutil_copyfile: Callable[[str, str], str] = shutil.copyfile
         with start_active_span("shutil_copyfile"):
             await defer_to_thread(
-                self.hs.get_reactor(),
+                self.reactor,
                 shutil_copyfile,
                 primary_fname,
                 backup_fname,
@@ -177,7 +178,7 @@ class FileStorageProviderBackend(StorageProvider):
 
         backup_fname = os.path.join(self.base_directory, path)
         if os.path.isfile(backup_fname):
-            return FileResponder(open(backup_fname, "rb"))
+            return FileResponder(self.hs, open(backup_fname, "rb"))
 
         return None
 

synapse/media/thumbnailer.py

@@ -259,6 +259,7 @@ class ThumbnailProvider:
         media_storage: MediaStorage,
     ):
         self.hs = hs
+        self.reactor = hs.get_reactor()
         self.media_repo = media_repo
         self.media_storage = media_storage
         self.store = hs.get_datastores().main
@@ -373,11 +374,11 @@ class ThumbnailProvider:
                 await respond_with_multipart_responder(
                     self.hs.get_clock(),
                     request,
-                    FileResponder(open(file_path, "rb")),
+                    FileResponder(self.hs, open(file_path, "rb")),
                     media_info,
                 )
             else:
-                await respond_with_file(request, desired_type, file_path)
+                await respond_with_file(self.hs, request, desired_type, file_path)
         else:
             logger.warning("Failed to generate thumbnail")
             raise SynapseError(400, "Failed to generate thumbnail.")
@@ -455,7 +456,7 @@ class ThumbnailProvider:
         )
 
         if file_path:
-            await respond_with_file(request, desired_type, file_path)
+            await respond_with_file(self.hs, request, desired_type, file_path)
         else:
             logger.warning("Failed to generate thumbnail")
             raise SynapseError(400, "Failed to generate thumbnail.")

synapse/notifier.py

@@ -773,6 +773,7 @@ class Notifier:
         stream_token = await self.event_sources.bound_future_token(stream_token)
 
         start = self.clock.time_msec()
+        logged = False
         while True:
             current_token = self.event_sources.get_current_token()
             if stream_token.is_before_or_eq(current_token):
@@ -783,11 +784,13 @@ class Notifier:
             if now - start > 10_000:
                 return False
 
-            logger.info(
-                "Waiting for current token to reach %s; currently at %s",
-                stream_token,
-                current_token,
-            )
+            if not logged:
+                logger.info(
+                    "Waiting for current token to reach %s; currently at %s",
+                    stream_token,
+                    current_token,
+                )
+                logged = True
 
             # TODO: be better
             await self.clock.sleep(0.5)

synapse/rest/__init__.py

@@ -18,7 +18,8 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
-from typing import TYPE_CHECKING, Callable
+import logging
+from typing import TYPE_CHECKING, Callable, Dict, Iterable, List, Optional, Tuple
 
 from synapse.http.server import HttpServer, JsonResource
 from synapse.rest import admin
@@ -67,11 +68,64 @@ from synapse.rest.client import (
     voip,
 )
 
+logger = logging.getLogger(__name__)
+
 if TYPE_CHECKING:
     from synapse.server import HomeServer
 
 RegisterServletsFunc = Callable[["HomeServer", HttpServer], None]
 
+CLIENT_SERVLET_FUNCTIONS: Tuple[RegisterServletsFunc, ...] = (
+    versions.register_servlets,
+    initial_sync.register_servlets,
+    room.register_deprecated_servlets,
+    events.register_servlets,
+    room.register_servlets,
+    login.register_servlets,
+    profile.register_servlets,
+    presence.register_servlets,
+    directory.register_servlets,
+    voip.register_servlets,
+    pusher.register_servlets,
+    push_rule.register_servlets,
+    logout.register_servlets,
+    sync.register_servlets,
+    filter.register_servlets,
+    account.register_servlets,
+    register.register_servlets,
+    auth.register_servlets,
+    receipts.register_servlets,
+    read_marker.register_servlets,
+    room_keys.register_servlets,
+    keys.register_servlets,
+    tokenrefresh.register_servlets,
+    tags.register_servlets,
+    account_data.register_servlets,
+    reporting.register_servlets,
+    openid.register_servlets,
+    notifications.register_servlets,
+    devices.register_servlets,
+    thirdparty.register_servlets,
+    sendtodevice.register_servlets,
+    user_directory.register_servlets,
+    room_upgrade_rest_servlet.register_servlets,
+    capabilities.register_servlets,
+    account_validity.register_servlets,
+    relations.register_servlets,
+    password_policy.register_servlets,
+    knock.register_servlets,
+    appservice_ping.register_servlets,
+    admin.register_servlets_for_client_rest_resource,
+    mutual_rooms.register_servlets,
+    login_token_request.register_servlets,
+    rendezvous.register_servlets,
+    auth_issuer.register_servlets,
+)
+
+SERVLET_GROUPS: Dict[str, Iterable[RegisterServletsFunc]] = {
+    "client": CLIENT_SERVLET_FUNCTIONS,
+}
+
 
 class ClientRestResource(JsonResource):
     """Matrix Client API REST resource.
@@ -83,80 +137,56 @@ class ClientRestResource(JsonResource):
        * etc
     """
 
-    def __init__(self, hs: "HomeServer"):
+    def __init__(self, hs: "HomeServer", servlet_groups: Optional[List[str]] = None):
         JsonResource.__init__(self, hs, canonical_json=False)
-        self.register_servlets(self, hs)
+        if hs.config.media.can_load_media_repo:
+            # This import is here to prevent a circular import failure
+            from synapse.rest.client import media
+
+            SERVLET_GROUPS["media"] = (media.register_servlets,)
+        self.register_servlets(self, hs, servlet_groups)
 
     @staticmethod
-    def register_servlets(client_resource: HttpServer, hs: "HomeServer") -> None:
+    def register_servlets(
+        client_resource: HttpServer,
+        hs: "HomeServer",
+        servlet_groups: Optional[Iterable[str]] = None,
+    ) -> None:
         # Some servlets are only registered on the main process (and not worker
         # processes).
         is_main_process = hs.config.worker.worker_app is None
 
-        versions.register_servlets(hs, client_resource)
+        if not servlet_groups:
+            servlet_groups = SERVLET_GROUPS.keys()
 
-        # Deprecated in r0
-        initial_sync.register_servlets(hs, client_resource)
-        room.register_deprecated_servlets(hs, client_resource)
+        for servlet_group in servlet_groups:
+            # Fail on unknown servlet groups.
+            if servlet_group not in SERVLET_GROUPS:
+                if servlet_group == "media":
+                    logger.warn(
+                        "media.can_load_media_repo needs to be configured for the media servlet to be available"
+                    )
+                raise RuntimeError(
+                    f"Attempting to register unknown client servlet: '{servlet_group}'"
+                )
 
-        # Partially deprecated in r0
-        events.register_servlets(hs, client_resource)
+            for servletfunc in SERVLET_GROUPS[servlet_group]:
+                if not is_main_process and servletfunc in [
+                    pusher.register_servlets,
+                    logout.register_servlets,
+                    auth.register_servlets,
+                    tokenrefresh.register_servlets,
+                    reporting.register_servlets,
+                    openid.register_servlets,
+                    thirdparty.register_servlets,
+                    room_upgrade_rest_servlet.register_servlets,
+                    account_validity.register_servlets,
+                    admin.register_servlets_for_client_rest_resource,
+                    mutual_rooms.register_servlets,
+                    login_token_request.register_servlets,
+                    rendezvous.register_servlets,
+                    auth_issuer.register_servlets,
+                ]:
+                    continue
 
-        room.register_servlets(hs, client_resource)
-        login.register_servlets(hs, client_resource)
-        profile.register_servlets(hs, client_resource)
-        presence.register_servlets(hs, client_resource)
-        directory.register_servlets(hs, client_resource)
-        voip.register_servlets(hs, client_resource)
-        if is_main_process:
-            pusher.register_servlets(hs, client_resource)
-        push_rule.register_servlets(hs, client_resource)
-        if is_main_process:
-            logout.register_servlets(hs, client_resource)
-        sync.register_servlets(hs, client_resource)
-        filter.register_servlets(hs, client_resource)
-        account.register_servlets(hs, client_resource)
-        register.register_servlets(hs, client_resource)
-        if is_main_process:
-            auth.register_servlets(hs, client_resource)
-        receipts.register_servlets(hs, client_resource)
-        read_marker.register_servlets(hs, client_resource)
-        room_keys.register_servlets(hs, client_resource)
-        keys.register_servlets(hs, client_resource)
-        if is_main_process:
-            tokenrefresh.register_servlets(hs, client_resource)
-        tags.register_servlets(hs, client_resource)
-        account_data.register_servlets(hs, client_resource)
-        if is_main_process:
-            reporting.register_servlets(hs, client_resource)
-            openid.register_servlets(hs, client_resource)
-        notifications.register_servlets(hs, client_resource)
-        devices.register_servlets(hs, client_resource)
-        if is_main_process:
-            thirdparty.register_servlets(hs, client_resource)
-        sendtodevice.register_servlets(hs, client_resource)
-        user_directory.register_servlets(hs, client_resource)
-        if is_main_process:
-            room_upgrade_rest_servlet.register_servlets(hs, client_resource)
-        capabilities.register_servlets(hs, client_resource)
-        if is_main_process:
-            account_validity.register_servlets(hs, client_resource)
-        relations.register_servlets(hs, client_resource)
-        password_policy.register_servlets(hs, client_resource)
-        knock.register_servlets(hs, client_resource)
-        appservice_ping.register_servlets(hs, client_resource)
-        if hs.config.media.can_load_media_repo:
-            from synapse.rest.client import media
-
-            media.register_servlets(hs, client_resource)
-
-        # moving to /_synapse/admin
-        if is_main_process:
-            admin.register_servlets_for_client_rest_resource(hs, client_resource)
-
-        # unstable
-        if is_main_process:
-            mutual_rooms.register_servlets(hs, client_resource)
-            login_token_request.register_servlets(hs, client_resource)
-            rendezvous.register_servlets(hs, client_resource)
-            auth_issuer.register_servlets(hs, client_resource)
+                servletfunc(hs, client_resource)

synapse/rest/client/keys.py

@@ -256,9 +256,15 @@ class KeyChangesServlet(RestServlet):
 
         user_id = requester.user.to_string()
 
-        results = await self.device_handler.get_user_ids_changed(user_id, from_token)
+        device_list_updates = await self.device_handler.get_user_ids_changed(
+            user_id, from_token
+        )
 
-        return 200, results
+        response: JsonDict = {}
+        response["changed"] = list(device_list_updates.changed)
+        response["left"] = list(device_list_updates.left)
+
+        return 200, response
 
 
 class OneTimeKeyServlet(RestServlet):

synapse/rest/client/room.py

@@ -67,7 +67,8 @@ from synapse.streams.config import PaginationConfig
 from synapse.types import JsonDict, Requester, StreamToken, ThirdPartyInstanceID, UserID
 from synapse.types.state import StateFilter
 from synapse.util.cancellation import cancellable
-from synapse.util.stringutils import parse_and_validate_server_name, random_string
+from synapse.util.events import generate_fake_event_id
+from synapse.util.stringutils import parse_and_validate_server_name
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -325,7 +326,7 @@ class RoomStateEventRestServlet(RestServlet):
                 )
                 event_id = event.event_id
         except ShadowBanError:
-            event_id = "$" + random_string(43)
+            event_id = generate_fake_event_id()
 
         set_tag("event_id", event_id)
         ret = {"event_id": event_id}
@@ -377,7 +378,7 @@ class RoomSendEventRestServlet(TransactionRestServlet):
             )
             event_id = event.event_id
         except ShadowBanError:
-            event_id = "$" + random_string(43)
+            event_id = generate_fake_event_id()
 
         set_tag("event_id", event_id)
         return 200, {"event_id": event_id}
@@ -1193,7 +1194,7 @@ class RoomRedactEventRestServlet(TransactionRestServlet):
 
             event_id = event.event_id
         except ShadowBanError:
-            event_id = "$" + random_string(43)
+            event_id = generate_fake_event_id()
 
         set_tag("event_id", event_id)
         return 200, {"event_id": event_id}

synapse/rest/client/sync.py

@@ -52,9 +52,9 @@ from synapse.http.servlet import (
     parse_string,
 )
 from synapse.http.site import SynapseRequest
-from synapse.logging.opentracing import trace_with_opname
+from synapse.logging.opentracing import log_kv, set_tag, trace_with_opname
 from synapse.rest.admin.experimental_features import ExperimentalFeature
-from synapse.types import JsonDict, Requester, StreamToken
+from synapse.types import JsonDict, Requester, SlidingSyncStreamToken, StreamToken
 from synapse.types.rest.client import SlidingSyncBody
 from synapse.util import json_decoder
 from synapse.util.caches.lrucache import LruCache
@@ -881,7 +881,6 @@ class SlidingSyncRestServlet(RestServlet):
         )
 
         user = requester.user
-        device_id = requester.device_id
 
         timeout = parse_integer(request, "timeout", default=0)
         # Position in the stream
@@ -889,22 +888,50 @@ class SlidingSyncRestServlet(RestServlet):
 
         from_token = None
         if from_token_string is not None:
-            from_token = await StreamToken.from_string(self.store, from_token_string)
+            from_token = await SlidingSyncStreamToken.from_string(
+                self.store, from_token_string
+            )
 
         # TODO: We currently don't know whether we're going to use sticky params or
         # maybe some filters like sync v2  where they are built up once and referenced
         # by filter ID. For now, we will just prototype with always passing everything
         # in.
         body = parse_and_validate_json_object_from_request(request, SlidingSyncBody)
-        logger.info("Sliding sync request: %r", body)
+
+        # Tag and log useful data to differentiate requests.
+        set_tag(
+            "sliding_sync.sync_type", "initial" if from_token is None else "incremental"
+        )
+        set_tag("sliding_sync.conn_id", body.conn_id or "")
+        log_kv(
+            {
+                "sliding_sync.lists": {
+                    list_name: {
+                        "ranges": list_config.ranges,
+                        "timeline_limit": list_config.timeline_limit,
+                    }
+                    for list_name, list_config in (body.lists or {}).items()
+                },
+                "sliding_sync.room_subscriptions": list(
+                    (body.room_subscriptions or {}).keys()
+                ),
+                # We also include the number of room subscriptions because logs are
+                # limited to 1024 characters and the large room ID list above can be cut
+                # off.
+                "sliding_sync.num_room_subscriptions": len(
+                    (body.room_subscriptions or {}).keys()
+                ),
+            }
+        )
 
         sync_config = SlidingSyncConfig(
             user=user,
-            device_id=device_id,
+            requester=requester,
             # FIXME: Currently, we're just manually copying the fields from the
-            # `SlidingSyncBody` into the config. How can we gurantee into the future
+            # `SlidingSyncBody` into the config. How can we guarantee into the future
             # that we don't forget any? I would like something more structured like
             # `copy_attributes(from=body, to=config)`
+            conn_id=body.conn_id,
             lists=body.lists,
             room_subscriptions=body.room_subscriptions,
             extensions=body.extensions,
@@ -927,7 +954,6 @@ class SlidingSyncRestServlet(RestServlet):
 
         return 200, response_content
 
-    # TODO: Is there a better way to encode things?
     async def encode_response(
         self,
         requester: Requester,
@@ -1018,6 +1044,11 @@ class SlidingSyncRestServlet(RestServlet):
             if room_result.initial:
                 serialized_rooms[room_id]["initial"] = room_result.initial
 
+            if room_result.unstable_expanded_timeline:
+                serialized_rooms[room_id][
+                    "unstable_expanded_timeline"
+                ] = room_result.unstable_expanded_timeline
+
             # This will be omitted for invite/knock rooms with `stripped_state`
             if (
                 room_result.required_state is not None
@@ -1081,15 +1112,69 @@ class SlidingSyncRestServlet(RestServlet):
     async def encode_extensions(
         self, requester: Requester, extensions: SlidingSyncResult.Extensions
     ) -> JsonDict:
-        result = {}
+        serialized_extensions: JsonDict = {}
 
         if extensions.to_device is not None:
-            result["to_device"] = {
+            serialized_extensions["to_device"] = {
                 "next_batch": extensions.to_device.next_batch,
                 "events": extensions.to_device.events,
             }
 
-        return result
+        if extensions.e2ee is not None:
+            serialized_extensions["e2ee"] = {
+                # We always include this because
+                # https://github.com/vector-im/element-android/issues/3725. The spec
+                # isn't terribly clear on when this can be omitted and how a client
+                # would tell the difference between "no keys present" and "nothing
+                # changed" in terms of whole field absent / individual key type entry
+                # absent Corresponding synapse issue:
+                # https://github.com/matrix-org/synapse/issues/10456
+                "device_one_time_keys_count": extensions.e2ee.device_one_time_keys_count,
+                # https://github.com/matrix-org/matrix-doc/blob/54255851f642f84a4f1aaf7bc063eebe3d76752b/proposals/2732-olm-fallback-keys.md
+                # states that this field should always be included, as long as the
+                # server supports the feature.
+                "device_unused_fallback_key_types": extensions.e2ee.device_unused_fallback_key_types,
+            }
+
+            if extensions.e2ee.device_list_updates is not None:
+                serialized_extensions["e2ee"]["device_lists"] = {}
+
+                serialized_extensions["e2ee"]["device_lists"]["changed"] = list(
+                    extensions.e2ee.device_list_updates.changed
+                )
+                serialized_extensions["e2ee"]["device_lists"]["left"] = list(
+                    extensions.e2ee.device_list_updates.left
+                )
+
+        if extensions.account_data is not None:
+            serialized_extensions["account_data"] = {
+                # Same as the the top-level `account_data.events` field in Sync v2.
+                "global": [
+                    {"type": account_data_type, "content": content}
+                    for account_data_type, content in extensions.account_data.global_account_data_map.items()
+                ],
+                # Same as the joined room's account_data field in Sync v2, e.g the path
+                # `rooms.join["!foo:bar"].account_data.events`.
+                "rooms": {
+                    room_id: [
+                        {"type": account_data_type, "content": content}
+                        for account_data_type, content in event_map.items()
+                    ]
+                    for room_id, event_map in extensions.account_data.account_data_by_room_map.items()
+                },
+            }
+
+        if extensions.receipts is not None:
+            serialized_extensions["receipts"] = {
+                "rooms": extensions.receipts.room_id_to_receipt_map,
+            }
+
+        if extensions.typing is not None:
+            serialized_extensions["typing"] = {
+                "rooms": extensions.typing.room_id_to_typing_map,
+            }
+
+        return serialized_extensions
 
 
 def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None:

synapse/rest/client/versions.py

@@ -64,6 +64,7 @@ class VersionsRestServlet(RestServlet):
 
     async def on_GET(self, request: SynapseRequest) -> Tuple[int, JsonDict]:
         msc3881_enabled = self.config.experimental.msc3881_enabled
+        msc3575_enabled = self.config.experimental.msc3575_enabled
 
         if self.auth.has_access_token(request):
             requester = await self.auth.get_user_by_req(
@@ -77,6 +78,9 @@ class VersionsRestServlet(RestServlet):
             msc3881_enabled = await self.store.is_feature_enabled(
                 user_id, ExperimentalFeature.MSC3881
             )
+            msc3575_enabled = await self.store.is_feature_enabled(
+                user_id, ExperimentalFeature.MSC3575
+            )
 
         return (
             200,
@@ -169,6 +173,8 @@ class VersionsRestServlet(RestServlet):
                     ),
                     # MSC4151: Report room API (Client-Server API)
                     "org.matrix.msc4151": self.config.experimental.msc4151_enabled,
+                    # Simplified sliding sync
+                    "org.matrix.simplified_msc3575": msc3575_enabled,
                 },
             },
         )

synapse/server.py

@@ -34,6 +34,7 @@ from typing_extensions import TypeAlias
 
 from twisted.internet.interfaces import IOpenSSLContextFactory
 from twisted.internet.tcp import Port
+from twisted.python.threadpool import ThreadPool
 from twisted.web.iweb import IPolicyForHTTPS
 from twisted.web.resource import Resource
 
@@ -123,6 +124,7 @@ from synapse.http.client import (
 )
 from synapse.http.matrixfederationclient import MatrixFederationHttpClient
 from synapse.media.media_repository import MediaRepository
+from synapse.metrics import register_threadpool
 from synapse.metrics.common_usage_metrics import CommonUsageMetricsManager
 from synapse.module_api import ModuleApi
 from synapse.module_api.callbacks import ModuleApiCallbacks
@@ -559,6 +561,7 @@ class HomeServer(metaclass=abc.ABCMeta):
     def get_sync_handler(self) -> SyncHandler:
         return SyncHandler(self)
 
+    @cache_in_self
     def get_sliding_sync_handler(self) -> SlidingSyncHandler:
         return SlidingSyncHandler(self)
 
@@ -940,3 +943,24 @@ class HomeServer(metaclass=abc.ABCMeta):
     @cache_in_self
     def get_task_scheduler(self) -> TaskScheduler:
         return TaskScheduler(self)
+
+    @cache_in_self
+    def get_media_sender_thread_pool(self) -> ThreadPool:
+        """Fetch the threadpool used to read files when responding to media
+        download requests."""
+
+        # We can choose a large threadpool size as these threads predominately
+        # do IO rather than CPU work.
+        media_threadpool = ThreadPool(
+            name="media_threadpool", minthreads=1, maxthreads=50
+        )
+
+        media_threadpool.start()
+        self.get_reactor().addSystemEventTrigger(
+            "during", "shutdown", media_threadpool.stop
+        )
+
+        # Register the threadpool with our metrics.
+        register_threadpool("media", media_threadpool)
+
+        return media_threadpool

synapse/storage/_base.py

@@ -127,6 +127,8 @@ class SQLBaseStore(metaclass=ABCMeta):
         # Purge other caches based on room state.
         self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
         self._attempt_to_invalidate_cache("get_partial_current_state_ids", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
     def _invalidate_state_caches_all(self, room_id: str) -> None:
         """Invalidates caches that are based on the current state, but does
@@ -153,6 +155,8 @@ class SQLBaseStore(metaclass=ABCMeta):
             "_get_rooms_for_local_user_where_membership_is_inner", None
         )
         self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
     def _attempt_to_invalidate_cache(
         self, cache_name: str, key: Optional[Collection[Any]]

synapse/storage/controllers/persist_events.py

@@ -502,8 +502,15 @@ class EventsPersistenceStorageController:
         """
         state = await self._calculate_current_state(room_id)
         delta = await self._calculate_state_delta(room_id, state)
+        sliding_sync_table_changes = (
+            await self.persist_events_store._calculate_sliding_sync_table_changes(
+                room_id, [], delta
+            )
+        )
 
-        await self.persist_events_store.update_current_state(room_id, delta)
+        await self.persist_events_store.update_current_state(
+            room_id, delta, sliding_sync_table_changes
+        )
 
     async def _calculate_current_state(self, room_id: str) -> StateMap[str]:
         """Calculate the current state of a room, based on the forward extremities

synapse/storage/database.py

@@ -35,6 +35,7 @@ from typing import (
     Iterable,
     Iterator,
     List,
+    Mapping,
     Optional,
     Sequence,
     Tuple,
@@ -64,6 +65,7 @@ from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.background_updates import BackgroundUpdater
 from synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine
 from synapse.storage.types import Connection, Cursor, SQLQueryParameters
+from synapse.types import StrCollection
 from synapse.util.async_helpers import delay_cancellation
 from synapse.util.iterutils import batch_iter
 
@@ -1095,6 +1097,50 @@ class DatabasePool:
 
         txn.execute(sql, vals)
 
+    @staticmethod
+    def simple_insert_returning_txn(
+        txn: LoggingTransaction,
+        table: str,
+        values: Dict[str, Any],
+        returning: StrCollection,
+    ) -> Tuple[Any, ...]:
+        """Executes a `INSERT INTO... RETURNING...` statement (or equivalent for
+        SQLite versions that don't support it).
+        """
+
+        if txn.database_engine.supports_returning:
+            keys, vals = zip(*values.items())
+
+            sql = "INSERT INTO %s (%s) VALUES(%s) RETURNING %s" % (
+                table,
+                ", ".join(k for k in keys),
+                ", ".join("?" for _ in keys),
+                ", ".join(k for k in returning),
+            )
+
+            txn.execute(sql, vals)
+            row = txn.fetchone()
+            assert row is not None
+            return row
+        else:
+            # For old versions of SQLite we do a standard insert and then can
+            # use `last_insert_rowid` to get at the row we just inserted
+            DatabasePool.simple_insert_txn(
+                txn,
+                table=table,
+                values=values,
+            )
+            txn.execute("SELECT last_insert_rowid()")
+            row = txn.fetchone()
+            assert row is not None
+            (rowid,) = row
+
+            row = DatabasePool.simple_select_one_txn(
+                txn, table=table, keyvalues={"rowid": rowid}, retcols=returning
+            )
+            assert row is not None
+            return row
+
     async def simple_insert_many(
         self,
         table: str,
@@ -1254,9 +1300,9 @@ class DatabasePool:
         self,
         txn: LoggingTransaction,
         table: str,
-        keyvalues: Dict[str, Any],
-        values: Dict[str, Any],
-        insertion_values: Optional[Dict[str, Any]] = None,
+        keyvalues: Mapping[str, Any],
+        values: Mapping[str, Any],
+        insertion_values: Optional[Mapping[str, Any]] = None,
         where_clause: Optional[str] = None,
     ) -> bool:
         """
@@ -1299,9 +1345,9 @@ class DatabasePool:
         self,
         txn: LoggingTransaction,
         table: str,
-        keyvalues: Dict[str, Any],
-        values: Dict[str, Any],
-        insertion_values: Optional[Dict[str, Any]] = None,
+        keyvalues: Mapping[str, Any],
+        values: Mapping[str, Any],
+        insertion_values: Optional[Mapping[str, Any]] = None,
         where_clause: Optional[str] = None,
         lock: bool = True,
     ) -> bool:
@@ -1322,7 +1368,7 @@ class DatabasePool:
 
         if lock:
             # We need to lock the table :(
-            self.engine.lock_table(txn, table)
+            txn.database_engine.lock_table(txn, table)
 
         def _getwhere(key: str) -> str:
             # If the value we're passing in is None (aka NULL), we need to use
@@ -1376,13 +1422,13 @@ class DatabasePool:
         # successfully inserted
         return True
 
+    @staticmethod
     def simple_upsert_txn_native_upsert(
-        self,
         txn: LoggingTransaction,
         table: str,
-        keyvalues: Dict[str, Any],
-        values: Dict[str, Any],
-        insertion_values: Optional[Dict[str, Any]] = None,
+        keyvalues: Mapping[str, Any],
+        values: Mapping[str, Any],
+        insertion_values: Optional[Mapping[str, Any]] = None,
         where_clause: Optional[str] = None,
     ) -> bool:
         """
@@ -1535,8 +1581,8 @@ class DatabasePool:
 
             self.simple_upsert_txn_emulated(txn, table, _keys, _vals, lock=False)
 
+    @staticmethod
     def simple_upsert_many_txn_native_upsert(
-        self,
         txn: LoggingTransaction,
         table: str,
         key_names: Collection[str],
@@ -1966,8 +2012,8 @@ class DatabasePool:
     def simple_update_txn(
         txn: LoggingTransaction,
         table: str,
-        keyvalues: Dict[str, Any],
-        updatevalues: Dict[str, Any],
+        keyvalues: Mapping[str, Any],
+        updatevalues: Mapping[str, Any],
     ) -> int:
         """
         Update rows in the given database table.

synapse/storage/databases/main/__init__.py

@@ -33,6 +33,7 @@ from synapse.storage.database import (
     LoggingDatabaseConnection,
     LoggingTransaction,
 )
+from synapse.storage.databases.main.sliding_sync import SlidingSyncStore
 from synapse.storage.databases.main.stats import UserSortOrder
 from synapse.storage.engines import BaseDatabaseEngine
 from synapse.storage.types import Cursor
@@ -156,6 +157,7 @@ class DataStore(
     LockStore,
     SessionStore,
     TaskSchedulerWorkerStore,
+    SlidingSyncStore,
 ):
     def __init__(
         self,

synapse/storage/databases/main/cache.py

@@ -268,13 +268,23 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
             self._curr_state_delta_stream_cache.entity_has_changed(data.room_id, token)  # type: ignore[attr-defined]
 
             if data.type == EventTypes.Member:
-                self.get_rooms_for_user.invalidate((data.state_key,))  # type: ignore[attr-defined]
+                self._attempt_to_invalidate_cache(
+                    "get_rooms_for_user", (data.state_key,)
+                )
+            elif data.type == EventTypes.RoomEncryption:
+                self._attempt_to_invalidate_cache(
+                    "get_room_encryption", (data.room_id,)
+                )
+            elif data.type == EventTypes.Create:
+                self._attempt_to_invalidate_cache("get_room_type", (data.room_id,))
         elif row.type == EventsStreamAllStateRow.TypeId:
             assert isinstance(data, EventsStreamAllStateRow)
             # Similar to the above, but the entire caches are invalidated. This is
             # unfortunate for the membership caches, but should recover quickly.
             self._curr_state_delta_stream_cache.entity_has_changed(data.room_id, token)  # type: ignore[attr-defined]
-            self.get_rooms_for_user.invalidate_all()  # type: ignore[attr-defined]
+            self._attempt_to_invalidate_cache("get_rooms_for_user", None)
+            self._attempt_to_invalidate_cache("get_room_type", (data.room_id,))
+            self._attempt_to_invalidate_cache("get_room_encryption", (data.room_id,))
         else:
             raise Exception("Unknown events stream row type %s" % (row.type,))
 
@@ -303,6 +313,8 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
             "get_unread_event_push_actions_by_room_for_user", (room_id,)
         )
 
+        self._attempt_to_invalidate_cache("_get_max_event_pos", (room_id,))
+
         # The `_get_membership_from_event_id` is immutable, except for the
         # case where we look up an event *before* persisting it.
         self._attempt_to_invalidate_cache("_get_membership_from_event_id", (event_id,))
@@ -345,6 +357,10 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
             self._attempt_to_invalidate_cache(
                 "get_forgotten_rooms_for_user", (state_key,)
             )
+        elif etype == EventTypes.Create:
+            self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        elif etype == EventTypes.RoomEncryption:
+            self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
         if relates_to:
             self._attempt_to_invalidate_cache(
@@ -405,6 +421,8 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
         self._attempt_to_invalidate_cache("get_thread_summary", None)
         self._attempt_to_invalidate_cache("get_thread_participated", None)
         self._attempt_to_invalidate_cache("get_threads", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
         self._attempt_to_invalidate_cache("_get_state_group_for_event", None)
 
@@ -457,6 +475,8 @@ class CacheInvalidationWorkerStore(SQLBaseStore):
         self._attempt_to_invalidate_cache("get_forgotten_rooms_for_user", None)
         self._attempt_to_invalidate_cache("_get_membership_from_event_id", None)
         self._attempt_to_invalidate_cache("get_room_version_id", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
         # And delete state caches.
 

synapse/storage/databases/main/event_federation.py

@@ -1313,6 +1313,11 @@ class EventFederationWorkerStore(SignatureWorkerStore, EventsWorkerStore, SQLBas
         # We want to make the cache more effective, so we clamp to the last
         # change before the given ordering.
         last_change = self._events_stream_cache.get_max_pos_of_last_change(room_id)  # type: ignore[attr-defined]
+        if last_change is None:
+            # If the room isn't in the cache we know that the last change was
+            # somewhere before the earliest known position of the cache, so we
+            # can clamp to that.
+            last_change = self._events_stream_cache.get_earliest_known_position()  # type: ignore[attr-defined]
 
         # We don't always have a full stream_to_exterm_id table, e.g. after
         # the upgrade that introduced it, so we make sure we never ask for a

synapse/storage/databases/main/events_worker.py

@@ -457,6 +457,8 @@ class EventsWorkerStore(SQLBaseStore):
     ) -> Optional[EventBase]:
         """Get an event from the database by event_id.
 
+        Events for unknown room versions will also be filtered out.
+
         Args:
             event_id: The event_id of the event to fetch
 
@@ -511,6 +513,10 @@ class EventsWorkerStore(SQLBaseStore):
     ) -> Dict[str, EventBase]:
         """Get events from the database
 
+        Unknown events will be omitted from the response.
+
+        Events for unknown room versions will also be filtered out.
+
         Args:
             event_ids: The event_ids of the events to fetch
 
@@ -553,6 +559,8 @@ class EventsWorkerStore(SQLBaseStore):
 
         Unknown events will be omitted from the response.
 
+        Events for unknown room versions will also be filtered out.
+
         Args:
             event_ids: The event_ids of the events to fetch
 

synapse/storage/databases/main/profile.py

@@ -144,6 +144,16 @@ class ProfileWorkerStore(SQLBaseStore):
         return 50
 
     async def get_profileinfo(self, user_id: UserID) -> ProfileInfo:
+        """
+        Fetch the display name and avatar URL of a user.
+
+        Args:
+            user_id: The user ID to fetch the profile for.
+
+        Returns:
+            The user's display name and avatar URL. Values may be null if unset
+             or if the user doesn't exist.
+        """
         profile = await self.db_pool.simple_select_one(
             table="profiles",
             keyvalues={"full_user_id": user_id.to_string()},
@@ -158,6 +168,15 @@ class ProfileWorkerStore(SQLBaseStore):
         return ProfileInfo(avatar_url=profile[1], display_name=profile[0])
 
     async def get_profile_displayname(self, user_id: UserID) -> Optional[str]:
+        """
+        Fetch the display name of a user.
+
+        Args:
+            user_id: The user to get the display name for.
+
+        Raises:
+            404 if the user does not exist.
+        """
         return await self.db_pool.simple_select_one_onecol(
             table="profiles",
             keyvalues={"full_user_id": user_id.to_string()},
@@ -166,6 +185,15 @@ class ProfileWorkerStore(SQLBaseStore):
         )
 
     async def get_profile_avatar_url(self, user_id: UserID) -> Optional[str]:
+        """
+        Fetch the avatar URL of a user.
+
+        Args:
+            user_id: The user to get the avatar URL for.
+
+        Raises:
+            404 if the user does not exist.
+        """
         return await self.db_pool.simple_select_one_onecol(
             table="profiles",
             keyvalues={"full_user_id": user_id.to_string()},
@@ -174,6 +202,12 @@ class ProfileWorkerStore(SQLBaseStore):
         )
 
     async def create_profile(self, user_id: UserID) -> None:
+        """
+        Create a blank profile for a user.
+
+        Args:
+            user_id: The user to create the profile for.
+        """
         user_localpart = user_id.localpart
         await self.db_pool.simple_insert(
             table="profiles",

synapse/storage/databases/main/purge_events.py

@@ -454,6 +454,10 @@ class PurgeEventsStore(StateGroupWorkerStore, CacheInvalidationWorkerStore):
             # so must be deleted first.
             "local_current_membership",
             "room_memberships",
+            # Note: the sliding_sync_ tables have foreign keys to the `events` table
+            # so must be deleted first.
+            "sliding_sync_joined_rooms",
+            "sliding_sync_membership_snapshots",
             "events",
             "federation_inbound_events_staging",
             "receipts_graph",

synapse/storage/databases/main/receipts.py

@@ -43,6 +43,7 @@ from synapse.storage.database import (
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
+    make_tuple_in_list_sql_clause,
 )
 from synapse.storage.engines._base import IsolationLevel
 from synapse.storage.util.id_generators import MultiWriterIdGenerator
@@ -51,10 +52,12 @@ from synapse.types import (
     JsonMapping,
     MultiWriterStreamToken,
     PersistedPosition,
+    StrCollection,
 )
 from synapse.util import json_encoder
 from synapse.util.caches.descriptors import cached, cachedList
 from synapse.util.caches.stream_change_cache import StreamChangeCache
+from synapse.util.iterutils import batch_iter
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -479,6 +482,83 @@ class ReceiptsWorkerStore(SQLBaseStore):
         }
         return results
 
+    async def get_linearized_receipts_for_events(
+        self,
+        room_and_event_ids: Collection[Tuple[str, str]],
+    ) -> Sequence[JsonMapping]:
+        """Get all receipts for the given set of events.
+
+        Arguments:
+            room_and_event_ids: A collection of 2-tuples of room ID and
+                event IDs to fetch receipts for
+
+        Returns:
+            A list of receipts, one per room.
+        """
+
+        def get_linearized_receipts_for_events_txn(
+            txn: LoggingTransaction,
+            room_id_event_id_tuples: Collection[Tuple[str, str]],
+        ) -> List[Tuple[str, str, str, str, Optional[str], str]]:
+            clause, args = make_tuple_in_list_sql_clause(
+                self.database_engine, ("room_id", "event_id"), room_id_event_id_tuples
+            )
+
+            sql = f"""
+                SELECT room_id, receipt_type, user_id, event_id, thread_id, data
+                FROM receipts_linearized
+                WHERE {clause}
+            """
+
+            txn.execute(sql, args)
+
+            return txn.fetchall()
+
+        # room_id -> event_id -> receipt_type -> user_id -> receipt data
+        room_to_content: Dict[str, Dict[str, Dict[str, Dict[str, JsonMapping]]]] = {}
+        for batch in batch_iter(room_and_event_ids, 1000):
+            batch_results = await self.db_pool.runInteraction(
+                "get_linearized_receipts_for_events",
+                get_linearized_receipts_for_events_txn,
+                batch,
+            )
+
+            for (
+                room_id,
+                receipt_type,
+                user_id,
+                event_id,
+                thread_id,
+                data,
+            ) in batch_results:
+                content = room_to_content.setdefault(room_id, {})
+                user_receipts = content.setdefault(event_id, {}).setdefault(
+                    receipt_type, {}
+                )
+
+                receipt_data = db_to_json(data)
+                if thread_id is not None:
+                    receipt_data["thread_id"] = thread_id
+
+                # MSC4102: always replace threaded receipts with unthreaded ones
+                # if there is a clash. Specifically:
+                # - if there is no existing receipt, great, set the data.
+                # - if there is an existing receipt, is it threaded (thread_id
+                #    present)? YES: replace if this receipt has no thread id.
+                # NO: do not replace. This means we will drop some receipts, but
+                # MSC4102 is designed to drop semantically meaningless receipts,
+                # so this is okay. Previously, we would drop meaningful data!
+                if user_id in user_receipts:
+                    if "thread_id" in user_receipts[user_id] and not thread_id:
+                        user_receipts[user_id] = receipt_data
+                else:
+                    user_receipts[user_id] = receipt_data
+
+        return [
+            {"type": EduTypes.RECEIPT, "room_id": room_id, "content": content}
+            for room_id, content in room_to_content.items()
+        ]
+
     @cached(
         num_args=2,
     )
@@ -550,6 +630,46 @@ class ReceiptsWorkerStore(SQLBaseStore):
 
         return results
 
+    async def get_rooms_with_receipts_between(
+        self,
+        room_ids: StrCollection,
+        from_key: MultiWriterStreamToken,
+        to_key: MultiWriterStreamToken,
+    ) -> StrCollection:
+        """Given a set of room_ids, find out which ones (may) have receipts
+        between the two tokens (> `from_token` and <= `to_token`)."""
+
+        room_ids = self._receipts_stream_cache.get_entities_changed(
+            room_ids, from_key.stream
+        )
+        if not room_ids:
+            return []
+
+        def f(txn: LoggingTransaction, room_ids: StrCollection) -> StrCollection:
+            clause, args = make_in_list_sql_clause(
+                self.database_engine, "room_id", room_ids
+            )
+
+            sql = f"""
+                SELECT DISTINCT room_id FROM receipts_linearized
+                WHERE {clause} AND ? < stream_id AND stream_id <= ?
+            """
+            args.append(from_key.stream)
+            args.append(to_key.get_max_stream_pos())
+
+            txn.execute(sql, args)
+
+            return [room_id for room_id, in txn]
+
+        results: List[str] = []
+        for batch in batch_iter(room_ids, 1000):
+            batch_result = await self.db_pool.runInteraction(
+                "get_rooms_with_receipts_between", f, batch
+            )
+            results.extend(batch_result)
+
+        return results
+
     async def get_users_sent_receipts_between(
         self, last_id: int, current_id: int
     ) -> List[str]:
@@ -954,6 +1074,12 @@ class ReceiptsBackgroundUpdateStore(SQLBaseStore):
             self.RECEIPTS_GRAPH_UNIQUE_INDEX_UPDATE_NAME,
             self._background_receipts_graph_unique_index,
         )
+        self.db_pool.updates.register_background_index_update(
+            update_name="receipts_room_id_event_id_index",
+            index_name="receipts_linearized_event_id",
+            table="receipts_linearized",
+            columns=("room_id", "event_id"),
+        )
 
     async def _populate_receipt_event_stream_ordering(
         self, progress: JsonDict, batch_size: int

synapse/storage/databases/main/roommember.py

@@ -39,6 +39,7 @@ from typing import (
 import attr
 
 from synapse.api.constants import EventTypes, Membership
+from synapse.logging.opentracing import trace
 from synapse.metrics import LaterGauge
 from synapse.metrics.background_process_metrics import wrap_as_background_process
 from synapse.storage._base import SQLBaseStore, db_to_json, make_in_list_sql_clause
@@ -422,6 +423,7 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
                 return invite
         return None
 
+    @trace
     async def get_rooms_for_local_user_where_membership_is(
         self,
         user_id: str,
@@ -1335,6 +1337,12 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
                 keyvalues={"user_id": user_id, "room_id": room_id},
                 updatevalues={"forgotten": 1},
             )
+            self.db_pool.simple_update_txn(
+                txn,
+                table="sliding_sync_membership_snapshots",
+                keyvalues={"user_id": user_id, "room_id": room_id},
+                updatevalues={"forgotten": 1},
+            )
 
             self._invalidate_cache_and_stream(txn, self.did_forget, (user_id, room_id))
             self._invalidate_cache_and_stream(

synapse/storage/databases/main/sliding_sync.py

@@ -0,0 +1,506 @@
+#
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2023 New Vector, Ltd
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# See the GNU Affero General Public License for more details:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+
+from typing import TYPE_CHECKING, AbstractSet, Dict, List, Mapping, Optional, Set, cast
+
+import attr
+
+from synapse.api.errors import SlidingSyncUnknownPosition
+from synapse.logging.opentracing import log_kv
+from synapse.storage._base import SQLBaseStore, db_to_json
+from synapse.storage.database import LoggingTransaction
+from synapse.types import MultiWriterStreamToken, RoomStreamToken
+from synapse.types.handlers.sliding_sync import (
+    HaveSentRoom,
+    HaveSentRoomFlag,
+    MutablePerConnectionState,
+    PerConnectionState,
+    RoomStatusMap,
+    RoomSyncConfig,
+)
+from synapse.util import json_encoder
+from synapse.util.caches.descriptors import cached
+
+if TYPE_CHECKING:
+    from synapse.storage.databases.main import DataStore
+
+
+class SlidingSyncStore(SQLBaseStore):
+    async def persist_per_connection_state(
+        self,
+        user_id: str,
+        device_id: str,
+        conn_id: str,
+        previous_connection_position: Optional[int],
+        per_connection_state: "MutablePerConnectionState",
+    ) -> int:
+        """Persist updates to the per-connection state for a sliding sync
+        connection.
+
+        Returns:
+            The connection position of the newly persisted state.
+        """
+
+        store = cast("DataStore", self)
+        return await self.db_pool.runInteraction(
+            "persist_per_connection_state",
+            self.persist_per_connection_state_txn,
+            user_id=user_id,
+            device_id=device_id,
+            conn_id=conn_id,
+            previous_connection_position=previous_connection_position,
+            per_connection_state=await PerConnectionStateDB.from_state(
+                per_connection_state, store
+            ),
+        )
+
+    def persist_per_connection_state_txn(
+        self,
+        txn: LoggingTransaction,
+        user_id: str,
+        device_id: str,
+        conn_id: str,
+        previous_connection_position: Optional[int],
+        per_connection_state: "PerConnectionStateDB",
+    ) -> int:
+        # First we fetch the (or create) the connection key associated with the
+        # previous connection position.
+        if previous_connection_position is not None:
+            # The `previous_connection_position` is a user-supplied value, so we
+            # need to make sure that the one they supplied is actually theirs.
+            sql = """
+                SELECT connection_key
+                FROM sliding_sync_connection_positions
+                INNER JOIN sliding_sync_connections USING (connection_key)
+                WHERE
+                    connection_position = ?
+                    AND user_id = ? AND device_id = ? AND conn_id = ?
+            """
+            txn.execute(
+                sql, (previous_connection_position, user_id, device_id, conn_id)
+            )
+            row = txn.fetchone()
+            if row is None:
+                raise SlidingSyncUnknownPosition()
+
+            (connection_key,) = row
+        else:
+            # We're restarting the connection, so we clear all existing
+            # connections. We do this here to ensure that if we get lots of
+            # one-shot requests we don't stack up lots of entries.
+            self.db_pool.simple_delete_txn(
+                txn,
+                table="sliding_sync_connections",
+                keyvalues={
+                    "user_id": user_id,
+                    "device_id": device_id,
+                    "conn_id": conn_id,
+                },
+            )
+
+            (connection_key,) = self.db_pool.simple_insert_returning_txn(
+                txn,
+                table="sliding_sync_connections",
+                values={
+                    "user_id": user_id,
+                    "device_id": device_id,
+                    "conn_id": conn_id,
+                    "created_ts": self._clock.time_msec(),
+                },
+                returning=("connection_key",),
+            )
+
+        # Define a new connection position for the updates
+        (connection_position,) = self.db_pool.simple_insert_returning_txn(
+            txn,
+            table="sliding_sync_connection_positions",
+            values={
+                "connection_key": connection_key,
+                "created_ts": self._clock.time_msec(),
+            },
+            returning=("connection_position",),
+        )
+
+        # We need to deduplicate the `required_state` JSON. We do this by
+        # fetching all JSON associated with the connection and comparing that
+        # with the updates to `required_state`
+
+        # Dict from required state json -> required state ID
+        required_state_to_id: Dict[str, int] = {}
+        if previous_connection_position is not None:
+            rows = self.db_pool.simple_select_list_txn(
+                txn,
+                table="sliding_sync_connection_required_state",
+                keyvalues={"connection_key": connection_key},
+                retcols=("required_state_id", "required_state"),
+            )
+            for required_state_id, required_state in rows:
+                required_state_to_id[required_state] = required_state_id
+
+        room_to_state_ids: Dict[str, int] = {}
+        unique_required_state: Dict[str, List[str]] = {}
+        for room_id, room_state in per_connection_state.room_configs.items():
+            serialized_state = json_encoder.encode(
+                # We store the required state as a sorted list of event type /
+                # state key tuples.
+                sorted(
+                    (event_type, state_key)
+                    for event_type, state_keys in room_state.required_state_map.items()
+                    for state_key in state_keys
+                )
+            )
+
+            existing_state_id = required_state_to_id.get(serialized_state)
+            if existing_state_id is not None:
+                room_to_state_ids[room_id] = existing_state_id
+            else:
+                unique_required_state.setdefault(serialized_state, []).append(room_id)
+
+        # Insert any new `required_state` json we haven't previously seen.
+        for serialized_required_state, room_ids in unique_required_state.items():
+            (required_state_id,) = self.db_pool.simple_insert_returning_txn(
+                txn,
+                table="sliding_sync_connection_required_state",
+                values={
+                    "connection_key": connection_key,
+                    "required_state": serialized_required_state,
+                },
+                returning=("required_state_id",),
+            )
+            for room_id in room_ids:
+                room_to_state_ids[room_id] = required_state_id
+
+        # Copy over state from the previous connection position (we'll overwrite
+        # these rows with any changes).
+        if previous_connection_position is not None:
+            sql = """
+                INSERT INTO sliding_sync_connection_streams
+                (connection_position, stream, room_id, room_status, last_position)
+                SELECT ?, stream, room_id, room_status, last_position
+                FROM sliding_sync_connection_streams
+                WHERE connection_position = ?
+            """
+            txn.execute(sql, (connection_position, previous_connection_position))
+
+            sql = """
+                INSERT INTO sliding_sync_connection_room_configs
+                (connection_position, room_id, timeline_limit, required_state_id)
+                SELECT ?, room_id, timeline_limit, required_state_id
+                FROM sliding_sync_connection_room_configs
+                WHERE connection_position = ?
+            """
+            txn.execute(sql, (connection_position, previous_connection_position))
+
+        # We now upsert the changes to the various streams.
+        key_values = []
+        value_values = []
+        for room_id, have_sent_room in per_connection_state.rooms._statuses.items():
+            key_values.append((connection_position, "rooms", room_id))
+            value_values.append(
+                (have_sent_room.status.value, have_sent_room.last_token)
+            )
+
+        for room_id, have_sent_room in per_connection_state.receipts._statuses.items():
+            key_values.append((connection_position, "receipts", room_id))
+            value_values.append(
+                (have_sent_room.status.value, have_sent_room.last_token)
+            )
+
+        self.db_pool.simple_upsert_many_txn(
+            txn,
+            table="sliding_sync_connection_streams",
+            key_names=(
+                "connection_position",
+                "stream",
+                "room_id",
+            ),
+            key_values=key_values,
+            value_names=(
+                "room_status",
+                "last_position",
+            ),
+            value_values=value_values,
+        )
+
+        # ... and upsert changes to the room configs.
+        keys = []
+        values = []
+        for room_id, room_config in per_connection_state.room_configs.items():
+            keys.append((connection_position, room_id))
+            values.append((room_config.timeline_limit, room_to_state_ids[room_id]))
+
+        self.db_pool.simple_upsert_many_txn(
+            txn,
+            table="sliding_sync_connection_room_configs",
+            key_names=(
+                "connection_position",
+                "room_id",
+            ),
+            key_values=keys,
+            value_names=(
+                "timeline_limit",
+                "required_state_id",
+            ),
+            value_values=values,
+        )
+
+        # Persist changes to the room lists
+        for list_name, list_room_ids in per_connection_state.list_to_rooms.items():
+            self.db_pool.simple_delete_txn(
+                txn,
+                table="sliding_sync_connection_room_lists",
+                keyvalues={"connection_key": connection_key, "list_name": list_name},
+            )
+            self.db_pool.simple_insert_many_txn(
+                txn,
+                table="sliding_sync_connection_room_lists",
+                keys=("connection_key", "list_name", "room_id"),
+                values=[
+                    (connection_key, list_name, room_id) for room_id in list_room_ids
+                ],
+            )
+
+        return connection_position
+
+    @cached(iterable=True, max_entries=100000)
+    async def get_per_connection_state(
+        self, user_id: str, device_id: str, conn_id: str, connection_position: int
+    ) -> "PerConnectionState":
+        """Get the per-connection state for the given connection position."""
+
+        per_connection_state_db = await self.db_pool.runInteraction(
+            "get_per_connection_state",
+            self._get_per_connection_state_txn,
+            user_id=user_id,
+            device_id=device_id,
+            conn_id=conn_id,
+            connection_position=connection_position,
+        )
+        store = cast("DataStore", self)
+        return await per_connection_state_db.to_state(store)
+
+    def _get_per_connection_state_txn(
+        self,
+        txn: LoggingTransaction,
+        user_id: str,
+        device_id: str,
+        conn_id: str,
+        connection_position: int,
+    ) -> "PerConnectionStateDB":
+        # The `previous_connection_position` is a user-supplied value, so we
+        # need to make sure that the one they supplied is actually theirs.
+        sql = """
+            SELECT connection_key
+            FROM sliding_sync_connection_positions
+            INNER JOIN sliding_sync_connections USING (connection_key)
+            WHERE
+                connection_position = ?
+                AND user_id = ? AND device_id = ? AND conn_id = ?
+        """
+        txn.execute(sql, (connection_position, user_id, device_id, conn_id))
+        row = txn.fetchone()
+        if row is None:
+            raise SlidingSyncUnknownPosition()
+
+        (connection_key,) = row
+
+        # Now that we have seen the client has received and used the connection
+        # position, we can delete all the other connection positions.
+        sql = """
+            DELETE FROM sliding_sync_connection_positions
+            WHERE connection_key = ? AND connection_position != ?
+        """
+        txn.execute(sql, (connection_key, connection_position))
+
+        # Fetch and create a mapping from required state ID to the actual
+        # required state for the connection.
+        rows = self.db_pool.simple_select_list_txn(
+            txn,
+            table="sliding_sync_connection_required_state",
+            keyvalues={"connection_key": connection_key},
+            retcols=(
+                "required_state_id",
+                "required_state",
+            ),
+        )
+
+        required_state_map: Dict[int, Dict[str, Set[str]]] = {}
+        for row in rows:
+            state = required_state_map[row[0]] = {}
+            for event_type, state_keys in db_to_json(row[1]):
+                state[event_type] = set(state_keys)
+
+        # Get all the room configs, looking up the required state from the map
+        # above.
+        room_config_rows = self.db_pool.simple_select_list_txn(
+            txn,
+            table="sliding_sync_connection_room_configs",
+            keyvalues={"connection_position": connection_position},
+            retcols=(
+                "room_id",
+                "timeline_limit",
+                "required_state_id",
+            ),
+        )
+
+        room_configs: Dict[str, RoomSyncConfig] = {}
+        for (
+            room_id,
+            timeline_limit,
+            required_state_id,
+        ) in room_config_rows:
+            room_configs[room_id] = RoomSyncConfig(
+                timeline_limit=timeline_limit,
+                required_state_map=required_state_map[required_state_id],
+            )
+
+        # Now look up the per-room stream data.
+        rooms: Dict[str, HaveSentRoom[str]] = {}
+        receipts: Dict[str, HaveSentRoom[str]] = {}
+
+        receipt_rows = self.db_pool.simple_select_list_txn(
+            txn,
+            table="sliding_sync_connection_streams",
+            keyvalues={"connection_position": connection_position},
+            retcols=(
+                "stream",
+                "room_id",
+                "room_status",
+                "last_position",
+            ),
+        )
+        for stream, room_id, room_status, last_position in receipt_rows:
+            have_sent_room: HaveSentRoom[str] = HaveSentRoom(
+                status=HaveSentRoomFlag(room_status), last_token=last_position
+            )
+            if stream == "rooms":
+                rooms[room_id] = have_sent_room
+            elif stream == "receipts":
+                receipts[room_id] = have_sent_room
+
+        # Fetch any stored lists for the connection
+        rows = self.db_pool.simple_select_list_txn(
+            txn,
+            table="sliding_sync_connection_room_lists",
+            keyvalues={
+                connection_key: connection_key,
+            },
+            retcols=("list_name", "room_id"),
+        )
+        list_to_rooms: Dict[str, Set[str]] = {}
+        for list_name, room_id in rows:
+            list_to_rooms.setdefault(list_name, set()).add(room_id)
+
+        return PerConnectionStateDB(
+            rooms=RoomStatusMap(rooms),
+            receipts=RoomStatusMap(receipts),
+            room_configs=room_configs,
+            list_to_rooms=list_to_rooms,
+        )
+
+
+@attr.s(auto_attribs=True, frozen=True)
+class PerConnectionStateDB:
+    """An equivalent to `PerConnectionState` that holds data in a format stored
+    in the DB.
+
+    The principle difference is that the tokens for the different streams are
+    serialized to strings.
+
+    When persisting this *only* contains updates to the state.
+    """
+
+    rooms: "RoomStatusMap[str]"
+    receipts: "RoomStatusMap[str]"
+
+    room_configs: Mapping[str, "RoomSyncConfig"]
+    list_to_rooms: Mapping[str, AbstractSet[str]]
+
+    @staticmethod
+    async def from_state(
+        per_connection_state: "MutablePerConnectionState", store: "DataStore"
+    ) -> "PerConnectionStateDB":
+        """Convert from a standard `PerConnectionState`"""
+        rooms = {
+            room_id: HaveSentRoom(
+                status=status.status,
+                last_token=(
+                    await status.last_token.to_string(store)
+                    if status.last_token is not None
+                    else None
+                ),
+            )
+            for room_id, status in per_connection_state.rooms.get_updates().items()
+        }
+
+        receipts = {
+            room_id: HaveSentRoom(
+                status=status.status,
+                last_token=(
+                    await status.last_token.to_string(store)
+                    if status.last_token is not None
+                    else None
+                ),
+            )
+            for room_id, status in per_connection_state.receipts.get_updates().items()
+        }
+
+        log_kv(
+            {
+                "rooms": rooms,
+                "receipts": receipts,
+                "room_configs": per_connection_state.room_configs.maps[0],
+            }
+        )
+
+        return PerConnectionStateDB(
+            rooms=RoomStatusMap(rooms),
+            receipts=RoomStatusMap(receipts),
+            room_configs=per_connection_state.room_configs.maps[0],
+            list_to_rooms=per_connection_state.list_to_rooms.maps[0],
+        )
+
+    async def to_state(self, store: "DataStore") -> "PerConnectionState":
+        """Convert into a standard `PerConnectionState`"""
+        rooms = {
+            room_id: HaveSentRoom(
+                status=status.status,
+                last_token=(
+                    await RoomStreamToken.parse(store, status.last_token)
+                    if status.last_token is not None
+                    else None
+                ),
+            )
+            for room_id, status in self.rooms._statuses.items()
+        }
+
+        receipts = {
+            room_id: HaveSentRoom(
+                status=status.status,
+                last_token=(
+                    await MultiWriterStreamToken.parse(store, status.last_token)
+                    if status.last_token is not None
+                    else None
+                ),
+            )
+            for room_id, status in self.receipts._statuses.items()
+        }
+
+        return PerConnectionState(
+            rooms=RoomStatusMap(rooms),
+            receipts=RoomStatusMap(receipts),
+            room_configs=self.room_configs,
+            list_to_rooms=self.list_to_rooms,
+        )

synapse/storage/databases/main/state.py

@@ -30,6 +30,7 @@ from typing import (
     Iterable,
     List,
     Mapping,
+    MutableMapping,
     Optional,
     Set,
     Tuple,
@@ -72,10 +73,18 @@ logger = logging.getLogger(__name__)
 
 _T = TypeVar("_T")
 
-
 MAX_STATE_DELTA_HOPS = 100
 
 
+# Freeze so it's immutable and we can use it as a cache value
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class Sentinel:
+    pass
+
+
+ROOM_UNKNOWN_SENTINEL = Sentinel()
+
+
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class EventMetadata:
     """Returned by `get_metadata_for_events`"""
@@ -300,51 +309,189 @@ class StateGroupWorkerStore(EventsWorkerStore, SQLBaseStore):
 
     @cached(max_entries=10000)
     async def get_room_type(self, room_id: str) -> Optional[str]:
-        """Get the room type for a given room. The server must be joined to the
-        given room.
-        """
-
-        row = await self.db_pool.simple_select_one(
-            table="room_stats_state",
-            keyvalues={"room_id": room_id},
-            retcols=("room_type",),
-            allow_none=True,
-            desc="get_room_type",
-        )
-
-        if row is not None:
-            return row[0]
-
-        # If we haven't updated `room_stats_state` with the room yet, query the
-        # create event directly.
-        create_event = await self.get_create_event_for_room(room_id)
-        room_type = create_event.content.get(EventContentFields.ROOM_TYPE)
-        return room_type
+        raise NotImplementedError()
 
     @cachedList(cached_method_name="get_room_type", list_name="room_ids")
     async def bulk_get_room_type(
         self, room_ids: Set[str]
-    ) -> Mapping[str, Optional[str]]:
-        """Bulk fetch room types for the given rooms, the server must be in all
-        the rooms given.
+    ) -> Mapping[str, Union[Optional[str], Sentinel]]:
+        """
+        Bulk fetch room types for the given rooms (via current state).
+
+        Since this function is cached, any missing values would be cached as `None`. In
+        order to distinguish between an unencrypted room that has `None` encryption and
+        a room that is unknown to the server where we might want to omit the value
+        (which would make it cached as `None`), instead we use the sentinel value
+        `ROOM_UNKNOWN_SENTINEL`.
+
+        Returns:
+            A mapping from room ID to the room's type (`None` is a valid room type).
+            Rooms unknown to this server will return `ROOM_UNKNOWN_SENTINEL`.
         """
 
-        rows = await self.db_pool.simple_select_many_batch(
-            table="room_stats_state",
-            column="room_id",
-            iterable=room_ids,
-            retcols=("room_id", "room_type"),
-            desc="bulk_get_room_type",
+        def txn(
+            txn: LoggingTransaction,
+        ) -> MutableMapping[str, Union[Optional[str], Sentinel]]:
+            clause, args = make_in_list_sql_clause(
+                txn.database_engine, "room_id", room_ids
+            )
+
+            # We can't rely on `room_stats_state.room_type` if the server has left the
+            # room because the `room_id` will still be in the table but everything will
+            # be set to `None` but `None` is a valid room type value. We join against
+            # the `room_stats_current` table which keeps track of the
+            # `current_state_events` count (and a proxy value `local_users_in_room`
+            # which can used to assume the server is participating in the room and has
+            # current state) to ensure that the data in `room_stats_state` is up-to-date
+            # with the current state.
+            #
+            # FIXME: Use `room_stats_current.current_state_events` instead of
+            # `room_stats_current.local_users_in_room` once
+            # https://github.com/element-hq/synapse/issues/17457 is fixed.
+            sql = f"""
+                SELECT room_id, room_type
+                FROM room_stats_state
+                INNER JOIN room_stats_current USING (room_id)
+                WHERE
+                    {clause}
+                    AND local_users_in_room > 0
+            """
+
+            txn.execute(sql, args)
+
+            room_id_to_type_map = {}
+            for row in txn:
+                room_id_to_type_map[row[0]] = row[1]
+
+            return room_id_to_type_map
+
+        results = await self.db_pool.runInteraction(
+            "bulk_get_room_type",
+            txn,
         )
 
         # If we haven't updated `room_stats_state` with the room yet, query the
         # create events directly. This should happen only rarely so we don't
         # mind if we do this in a loop.
-        results = dict(rows)
         for room_id in room_ids - results.keys():
-            create_event = await self.get_create_event_for_room(room_id)
-            room_type = create_event.content.get(EventContentFields.ROOM_TYPE)
-            results[room_id] = room_type
+            try:
+                create_event = await self.get_create_event_for_room(room_id)
+                room_type = create_event.content.get(EventContentFields.ROOM_TYPE)
+                results[room_id] = room_type
+            except NotFoundError:
+                # We use the sentinel value to distinguish between `None` which is a
+                # valid room type and a room that is unknown to the server so the value
+                # is just unset.
+                results[room_id] = ROOM_UNKNOWN_SENTINEL
+
+        return results
+
+    @cached(max_entries=10000)
+    async def get_room_encryption(self, room_id: str) -> Optional[str]:
+        raise NotImplementedError()
+
+    @cachedList(cached_method_name="get_room_encryption", list_name="room_ids")
+    async def bulk_get_room_encryption(
+        self, room_ids: Set[str]
+    ) -> Mapping[str, Union[Optional[str], Sentinel]]:
+        """
+        Bulk fetch room encryption for the given rooms (via current state).
+
+        Since this function is cached, any missing values would be cached as `None`. In
+        order to distinguish between an unencrypted room that has `None` encryption and
+        a room that is unknown to the server where we might want to omit the value
+        (which would make it cached as `None`), instead we use the sentinel value
+        `ROOM_UNKNOWN_SENTINEL`.
+
+        Returns:
+            A mapping from room ID to the room's encryption algorithm if the room is
+            encrypted, otherwise `None`. Rooms unknown to this server will return
+            `ROOM_UNKNOWN_SENTINEL`.
+        """
+
+        def txn(
+            txn: LoggingTransaction,
+        ) -> MutableMapping[str, Union[Optional[str], Sentinel]]:
+            clause, args = make_in_list_sql_clause(
+                txn.database_engine, "room_id", room_ids
+            )
+
+            # We can't rely on `room_stats_state.encryption` if the server has left the
+            # room because the `room_id` will still be in the table but everything will
+            # be set to `None` but `None` is a valid encryption value. We join against
+            # the `room_stats_current` table which keeps track of the
+            # `current_state_events` count (and a proxy value `local_users_in_room`
+            # which can used to assume the server is participating in the room and has
+            # current state) to ensure that the data in `room_stats_state` is up-to-date
+            # with the current state.
+            #
+            # FIXME: Use `room_stats_current.current_state_events` instead of
+            # `room_stats_current.local_users_in_room` once
+            # https://github.com/element-hq/synapse/issues/17457 is fixed.
+            sql = f"""
+                SELECT room_id, encryption
+                FROM room_stats_state
+                INNER JOIN room_stats_current USING (room_id)
+                WHERE
+                    {clause}
+                    AND local_users_in_room > 0
+            """
+
+            txn.execute(sql, args)
+
+            room_id_to_encryption_map = {}
+            for row in txn:
+                room_id_to_encryption_map[row[0]] = row[1]
+
+            return room_id_to_encryption_map
+
+        results = await self.db_pool.runInteraction(
+            "bulk_get_room_encryption",
+            txn,
+        )
+
+        # If we haven't updated `room_stats_state` with the room yet, query the state
+        # directly. This should happen only rarely so we don't mind if we do this in a
+        # loop.
+        encryption_event_ids: List[str] = []
+        for room_id in room_ids - results.keys():
+            state_map = await self.get_partial_filtered_current_state_ids(
+                room_id,
+                state_filter=StateFilter.from_types(
+                    [
+                        (EventTypes.Create, ""),
+                        (EventTypes.RoomEncryption, ""),
+                    ]
+                ),
+            )
+            # We can use the create event as a canary to tell whether the server has
+            # seen the room before
+            create_event_id = state_map.get((EventTypes.Create, ""))
+            encryption_event_id = state_map.get((EventTypes.RoomEncryption, ""))
+
+            if create_event_id is None:
+                # We use the sentinel value to distinguish between `None` which is a
+                # valid room type and a room that is unknown to the server so the value
+                # is just unset.
+                results[room_id] = ROOM_UNKNOWN_SENTINEL
+                continue
+
+            if encryption_event_id is None:
+                results[room_id] = None
+            else:
+                encryption_event_ids.append(encryption_event_id)
+
+        encryption_event_map = await self.get_events(encryption_event_ids)
+
+        for encryption_event_id in encryption_event_ids:
+            encryption_event = encryption_event_map.get(encryption_event_id)
+            # If the curent state says there is an encryption event, we should have it
+            # in the database.
+            assert encryption_event is not None
+
+            results[encryption_event.room_id] = encryption_event.content.get(
+                EventContentFields.ENCRYPTION_ALGORITHM
+            )
 
         return results
 

synapse/storage/databases/main/state_deltas.py

@@ -24,9 +24,13 @@ from typing import List, Optional, Tuple
 
 import attr
 
+from synapse.logging.opentracing import trace
 from synapse.storage._base import SQLBaseStore
-from synapse.storage.database import LoggingTransaction
+from synapse.storage.database import LoggingTransaction, make_in_list_sql_clause
+from synapse.storage.databases.main.stream import _filter_results_by_stream
+from synapse.types import RoomStreamToken, StrCollection
 from synapse.util.caches.stream_change_cache import StreamChangeCache
+from synapse.util.iterutils import batch_iter
 
 logger = logging.getLogger(__name__)
 
@@ -156,3 +160,138 @@ class StateDeltasStore(SQLBaseStore):
             "get_max_stream_id_in_current_state_deltas",
             self._get_max_stream_id_in_current_state_deltas_txn,
         )
+
+    def get_current_state_deltas_for_room_txn(
+        self,
+        txn: LoggingTransaction,
+        room_id: str,
+        *,
+        from_token: Optional[RoomStreamToken],
+        to_token: Optional[RoomStreamToken],
+    ) -> List[StateDelta]:
+        """
+        Get the state deltas between two tokens.
+
+        (> `from_token` and <= `to_token`)
+        """
+        from_clause = ""
+        from_args = []
+        if from_token is not None:
+            from_clause = "AND ? < stream_id"
+            from_args = [from_token.stream]
+
+        to_clause = ""
+        to_args = []
+        if to_token is not None:
+            to_clause = "AND stream_id <= ?"
+            to_args = [to_token.get_max_stream_pos()]
+
+        sql = f"""
+                SELECT instance_name, stream_id, type, state_key, event_id, prev_event_id
+                FROM current_state_delta_stream
+                WHERE room_id = ? {from_clause} {to_clause}
+                ORDER BY stream_id ASC
+            """
+        txn.execute(sql, [room_id] + from_args + to_args)
+
+        return [
+            StateDelta(
+                stream_id=row[1],
+                room_id=room_id,
+                event_type=row[2],
+                state_key=row[3],
+                event_id=row[4],
+                prev_event_id=row[5],
+            )
+            for row in txn
+            if _filter_results_by_stream(from_token, to_token, row[0], row[1])
+        ]
+
+    @trace
+    async def get_current_state_deltas_for_room(
+        self,
+        room_id: str,
+        *,
+        from_token: Optional[RoomStreamToken],
+        to_token: Optional[RoomStreamToken],
+    ) -> List[StateDelta]:
+        """
+        Get the state deltas between two tokens.
+
+        (> `from_token` and <= `to_token`)
+        """
+
+        if (
+            from_token is not None
+            and not self._curr_state_delta_stream_cache.has_entity_changed(
+                room_id, from_token.stream
+            )
+        ):
+            return []
+
+        return await self.db_pool.runInteraction(
+            "get_current_state_deltas_for_room",
+            self.get_current_state_deltas_for_room_txn,
+            room_id,
+            from_token=from_token,
+            to_token=to_token,
+        )
+
+    @trace
+    async def get_current_state_deltas_for_rooms(
+        self,
+        room_ids: StrCollection,
+        from_token: RoomStreamToken,
+        to_token: RoomStreamToken,
+    ) -> List[StateDelta]:
+        """Get the state deltas between two tokens for the set of rooms."""
+
+        room_ids = self._curr_state_delta_stream_cache.get_entities_changed(
+            room_ids, from_token.stream
+        )
+        if not room_ids:
+            return []
+
+        def get_current_state_deltas_for_rooms_txn(
+            txn: LoggingTransaction,
+            room_ids: StrCollection,
+        ) -> List[StateDelta]:
+            clause, args = make_in_list_sql_clause(
+                self.database_engine, "room_id", room_ids
+            )
+
+            sql = f"""
+                SELECT instance_name, stream_id, room_id, type, state_key, event_id, prev_event_id
+                FROM current_state_delta_stream
+                WHERE {clause} AND ? < stream_id AND stream_id <= ?
+                ORDER BY stream_id ASC
+            """
+            args.append(from_token.stream)
+            args.append(to_token.get_max_stream_pos())
+
+            txn.execute(sql, args)
+
+            return [
+                StateDelta(
+                    stream_id=row[1],
+                    room_id=row[2],
+                    event_type=row[3],
+                    state_key=row[4],
+                    event_id=row[5],
+                    prev_event_id=row[6],
+                )
+                for row in txn
+                if _filter_results_by_stream(from_token, to_token, row[0], row[1])
+            ]
+
+        results = []
+        for batch in batch_iter(room_ids, 1000):
+            deltas = await self.db_pool.runInteraction(
+                "get_current_state_deltas_for_rooms",
+                get_current_state_deltas_for_rooms_txn,
+                batch,
+            )
+
+            results.extend(deltas)
+
+        return results

synapse/storage/databases/main/stream.py

@@ -50,7 +50,9 @@ from typing import (
     Dict,
     Iterable,
     List,
+    Mapping,
     Optional,
+    Protocol,
     Set,
     Tuple,
     cast,
@@ -59,7 +61,7 @@ from typing import (
 
 import attr
 from immutabledict import immutabledict
-from typing_extensions import Literal
+from typing_extensions import Literal, assert_never
 
 from twisted.internet import defer
 
@@ -67,7 +69,7 @@ from synapse.api.constants import Direction, EventTypes, Membership
 from synapse.api.filtering import Filter
 from synapse.events import EventBase
 from synapse.logging.context import make_deferred_yieldable, run_in_background
-from synapse.logging.opentracing import trace
+from synapse.logging.opentracing import tag_args, trace
 from synapse.storage._base import SQLBaseStore
 from synapse.storage.database import (
     DatabasePool,
@@ -78,10 +80,11 @@ from synapse.storage.database import (
 from synapse.storage.databases.main.events_worker import EventsWorkerStore
 from synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine
 from synapse.storage.util.id_generators import MultiWriterIdGenerator
-from synapse.types import PersistedEventPosition, RoomStreamToken
-from synapse.util.caches.descriptors import cached
+from synapse.types import PersistedEventPosition, RoomStreamToken, StrCollection
+from synapse.util.caches.descriptors import cached, cachedList
 from synapse.util.caches.stream_change_cache import StreamChangeCache
 from synapse.util.cancellation import cancellable
+from synapse.util.iterutils import batch_iter
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer
@@ -96,6 +99,18 @@ _STREAM_TOKEN = "stream"
 _TOPOLOGICAL_TOKEN = "topological"
 
 
+class PaginateFunction(Protocol):
+    async def __call__(
+        self,
+        *,
+        room_id: str,
+        from_key: RoomStreamToken,
+        to_key: Optional[RoomStreamToken] = None,
+        direction: Direction = Direction.BACKWARDS,
+        limit: int = 0,
+    ) -> Tuple[List[EventBase], RoomStreamToken]: ...
+
+
 # Used as return values for pagination APIs
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class _EventDictReturn:
@@ -279,7 +294,7 @@ def generate_pagination_bounds(
 
 
 def generate_next_token(
-    direction: Direction, last_topo_ordering: int, last_stream_ordering: int
+    direction: Direction, last_topo_ordering: Optional[int], last_stream_ordering: int
 ) -> RoomStreamToken:
     """
     Generate the next room stream token based on the currently returned data.
@@ -446,7 +461,6 @@ def _filter_results_by_stream(
     The `instance_name` arg is optional to handle historic rows, and is
     interpreted as if it was "master".
     """
-
     if instance_name is None:
         instance_name = "master"
 
@@ -659,33 +673,43 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
 
     async def get_room_events_stream_for_rooms(
         self,
+        *,
         room_ids: Collection[str],
         from_key: RoomStreamToken,
-        to_key: RoomStreamToken,
+        to_key: Optional[RoomStreamToken] = None,
+        direction: Direction = Direction.BACKWARDS,
         limit: int = 0,
-        order: str = "DESC",
     ) -> Dict[str, Tuple[List[EventBase], RoomStreamToken]]:
         """Get new room events in stream ordering since `from_key`.
 
         Args:
             room_ids
-            from_key: Token from which no events are returned before
-            to_key: Token from which no events are returned after. (This
-                is typically the current stream token)
+            from_key: The token to stream from (starting point and heading in the given
+                direction)
+            to_key: The token representing the end stream position (end point)
             limit: Maximum number of events to return
-            order: Either "DESC" or "ASC". Determines which events are
-                returned when the result is limited. If "DESC" then the most
-                recent `limit` events are returned, otherwise returns the
-                oldest `limit` events.
+            direction: Indicates whether we are paginating forwards or backwards
+                from `from_key`.
 
         Returns:
             A map from room id to a tuple containing:
                 - list of recent events in the room
                 - stream ordering key for the start of the chunk of events returned.
+
+            When Direction.FORWARDS: from_key < x <= to_key, (ascending order)
+            When Direction.BACKWARDS: from_key >= x > to_key, (descending order)
         """
-        room_ids = self._events_stream_cache.get_entities_changed(
-            room_ids, from_key.stream
-        )
+        if direction == Direction.FORWARDS:
+            room_ids = self._events_stream_cache.get_entities_changed(
+                room_ids, from_key.stream
+            )
+        elif direction == Direction.BACKWARDS:
+            if to_key is not None:
+                room_ids = self._events_stream_cache.get_entities_changed(
+                    room_ids, to_key.stream
+                )
+        else:
+            assert_never(direction)
 
         if not room_ids:
             return {}
@@ -697,12 +721,12 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
                 defer.gatherResults(
                     [
                         run_in_background(
-                            self.get_room_events_stream_for_room,
-                            room_id,
-                            from_key,
-                            to_key,
-                            limit,
-                            order=order,
+                            self.paginate_room_events_by_stream_ordering,
+                            room_id=room_id,
+                            from_key=from_key,
+                            to_key=to_key,
+                            direction=direction,
+                            limit=limit,
                         )
                         for room_id in rm_ids
                     ],
@@ -726,69 +750,122 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             if self._events_stream_cache.has_entity_changed(room_id, from_id)
         }
 
-    async def get_room_events_stream_for_room(
+    async def paginate_room_events_by_stream_ordering(
         self,
+        *,
         room_id: str,
         from_key: RoomStreamToken,
-        to_key: RoomStreamToken,
+        to_key: Optional[RoomStreamToken] = None,
+        direction: Direction = Direction.BACKWARDS,
         limit: int = 0,
-        order: str = "DESC",
     ) -> Tuple[List[EventBase], RoomStreamToken]:
-        """Get new room events in stream ordering since `from_key`.
+        """
+        Paginate events by `stream_ordering` in the room from the `from_key` in the
+        given `direction` to the `to_key` or `limit`.
 
         Args:
             room_id
-            from_key: Token from which no events are returned before
-            to_key: Token from which no events are returned after. (This
-                is typically the current stream token)
+            from_key: The token to stream from (starting point and heading in the given
+                direction)
+            to_key: The token representing the end stream position (end point)
+            direction: Indicates whether we are paginating forwards or backwards
+                from `from_key`.
             limit: Maximum number of events to return
-            order: Either "DESC" or "ASC". Determines which events are
-                returned when the result is limited. If "DESC" then the most
-                recent `limit` events are returned, otherwise returns the
-                oldest `limit` events.
 
         Returns:
-            The list of events (in ascending stream order) and the token from the start
-            of the chunk of events returned.
-        """
-        if from_key == to_key:
-            return [], from_key
+            The results as a list of events and a token that points to the end
+            of the result set. If no events are returned then the end of the
+            stream has been reached (i.e. there are no events between `from_key`
+            and `to_key`).
 
-        has_changed = self._events_stream_cache.has_entity_changed(
-            room_id, from_key.stream
-        )
+            When Direction.FORWARDS: from_key < x <= to_key, (ascending order)
+            When Direction.BACKWARDS: from_key >= x > to_key, (descending order)
+        """
+
+        # FIXME: When going forwards, we should enforce that the `to_key` is not `None`
+        # because we always need an upper bound when querying the events stream (as
+        # otherwise we'll potentially pick up events that are not fully persisted).
+
+        # We should only be working with `stream_ordering` tokens here
+        assert from_key is None or from_key.topological is None
+        assert to_key is None or to_key.topological is None
+
+        # We can bail early if we're looking forwards, and our `to_key` is already
+        # before our `from_key`.
+        if (
+            direction == Direction.FORWARDS
+            and to_key is not None
+            and to_key.is_before_or_eq(from_key)
+        ):
+            # Token selection matches what we do below if there are no rows
+            return [], to_key if to_key else from_key
+        # Or vice-versa, if we're looking backwards and our `from_key` is already before
+        # our `to_key`.
+        elif (
+            direction == Direction.BACKWARDS
+            and to_key is not None
+            and from_key.is_before_or_eq(to_key)
+        ):
+            # Token selection matches what we do below if there are no rows
+            return [], to_key if to_key else from_key
+
+        # We can do a quick sanity check to see if any events have been sent in the room
+        # since the earlier token.
+        has_changed = True
+        if direction == Direction.FORWARDS:
+            has_changed = self._events_stream_cache.has_entity_changed(
+                room_id, from_key.stream
+            )
+        elif direction == Direction.BACKWARDS:
+            if to_key is not None:
+                has_changed = self._events_stream_cache.has_entity_changed(
+                    room_id, to_key.stream
+                )
+        else:
+            assert_never(direction)
 
         if not has_changed:
-            return [], from_key
+            # Token selection matches what we do below if there are no rows
+            return [], to_key if to_key else from_key
+
+        order, from_bound, to_bound = generate_pagination_bounds(
+            direction, from_key, to_key
+        )
+
+        bounds = generate_pagination_where_clause(
+            direction=direction,
+            # The empty string will shortcut downstream code to only use the
+            # `stream_ordering` column
+            column_names=("", "stream_ordering"),
+            from_token=from_bound,
+            to_token=to_bound,
+            engine=self.database_engine,
+        )
 
         def f(txn: LoggingTransaction) -> List[_EventDictReturn]:
-            # To handle tokens with a non-empty instance_map we fetch more
-            # results than necessary and then filter down
-            min_from_id = from_key.stream
-            max_to_id = to_key.get_max_stream_pos()
-
-            sql = """
-                SELECT event_id, instance_name, topological_ordering, stream_ordering
+            sql = f"""
+                SELECT event_id, instance_name, stream_ordering
                 FROM events
                 WHERE
                     room_id = ?
                     AND not outlier
-                    AND stream_ordering > ? AND stream_ordering <= ?
-                ORDER BY stream_ordering %s LIMIT ?
-            """ % (
-                order,
-            )
-            txn.execute(sql, (room_id, min_from_id, max_to_id, 2 * limit))
+                    AND {bounds}
+                ORDER BY stream_ordering {order} LIMIT ?
+            """
+            txn.execute(sql, (room_id, 2 * limit))
 
             rows = [
                 _EventDictReturn(event_id, None, stream_ordering)
-                for event_id, instance_name, topological_ordering, stream_ordering in txn
-                if _filter_results(
-                    from_key,
-                    to_key,
-                    instance_name,
-                    topological_ordering,
-                    stream_ordering,
+                for event_id, instance_name, stream_ordering in txn
+                if _filter_results_by_stream(
+                    lower_token=(
+                        to_key if direction == Direction.BACKWARDS else from_key
+                    ),
+                    upper_token=(
+                        from_key if direction == Direction.BACKWARDS else to_key
+                    ),
+                    instance_name=instance_name,
+                    stream_ordering=stream_ordering,
                 )
             ][:limit]
             return rows
@@ -799,18 +876,20 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             [r.event_id for r in rows], get_prev_content=True
         )
 
-        if order.lower() == "desc":
-            ret.reverse()
-
         if rows:
-            key = RoomStreamToken(stream=min(r.stream_ordering for r in rows))
+            next_key = generate_next_token(
+                direction=direction,
+                last_topo_ordering=None,
+                last_stream_ordering=rows[-1].stream_ordering,
+            )
         else:
-            # Assume we didn't get anything because there was nothing to
-            # get.
-            key = from_key
+            # TODO (erikj): We should work out what to do here instead. (same as
+            # `_paginate_room_events_by_topological_ordering_txn(...)`)
+            next_key = to_key if to_key else from_key
 
-        return ret, key
+        return ret, next_key
 
+    @trace
     async def get_current_state_delta_membership_changes_for_user(
         self,
         user_id: str,
@@ -1116,7 +1195,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
 
         rows, token = await self.db_pool.runInteraction(
             "get_recent_event_ids_for_room",
-            self._paginate_room_events_txn,
+            self._paginate_room_events_by_topological_ordering_txn,
             room_id,
             from_token=end_token,
             limit=limit,
@@ -1185,11 +1264,76 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
 
         return None
 
+    async def get_last_event_pos_in_room(
+        self,
+        room_id: str,
+        event_types: Optional[StrCollection] = None,
+    ) -> Optional[Tuple[str, PersistedEventPosition]]:
+        """
+        Returns the ID and event position of the last event in a room.
+
+        Based on `get_last_event_pos_in_room_before_stream_ordering(...)`
+
+        Args:
+            room_id
+            event_types: Optional allowlist of event types to filter by
+
+        Returns:
+            The ID of the most recent event and it's position, or None if there are no
+            events in the room that match the given event types.
+        """
+
+        def _get_last_event_pos_in_room_txn(
+            txn: LoggingTransaction,
+        ) -> Optional[Tuple[str, PersistedEventPosition]]:
+            event_type_clause = ""
+            event_type_args: List[str] = []
+            if event_types is not None and len(event_types) > 0:
+                event_type_clause, event_type_args = make_in_list_sql_clause(
+                    txn.database_engine, "type", event_types
+                )
+                event_type_clause = f"AND {event_type_clause}"
+
+            sql = f"""
+            SELECT event_id, stream_ordering, instance_name
+            FROM events
+            LEFT JOIN rejections USING (event_id)
+            WHERE room_id = ?
+                {event_type_clause}
+                AND NOT outlier
+                AND rejections.event_id IS NULL
+            ORDER BY stream_ordering DESC
+            LIMIT 1
+            """
+
+            txn.execute(
+                sql,
+                [room_id] + event_type_args,
+            )
+
+            row = cast(Optional[Tuple[str, int, str]], txn.fetchone())
+            if row is not None:
+                event_id, stream_ordering, instance_name = row
+
+                return event_id, PersistedEventPosition(
+                    # If instance_name is null we default to "master"
+                    instance_name or "master",
+                    stream_ordering,
+                )
+
+            return None
+
+        return await self.db_pool.runInteraction(
+            "get_last_event_pos_in_room",
+            _get_last_event_pos_in_room_txn,
+        )
+
+    @trace
     async def get_last_event_pos_in_room_before_stream_ordering(
         self,
         room_id: str,
         end_token: RoomStreamToken,
-        event_types: Optional[Collection[str]] = None,
+        event_types: Optional[StrCollection] = None,
     ) -> Optional[Tuple[str, PersistedEventPosition]]:
         """
         Returns the ID and event position of the last event in a room at or before a
@@ -1293,6 +1437,129 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             get_last_event_pos_in_room_before_stream_ordering_txn,
         )
 
+    async def bulk_get_last_event_pos_in_room_before_stream_ordering(
+        self,
+        room_ids: StrCollection,
+        end_token: RoomStreamToken,
+    ) -> Dict[str, int]:
+        """Bulk fetch the stream position of the latest events in the given
+        rooms
+        """
+
+        min_token = end_token.stream
+        results: Dict[str, int] = {}
+
+        # First, we check for the rooms in the stream change cache to see if we
+        # can just use the latest position from it.
+        missing_room_ids: Set[str] = set()
+        for room_id in room_ids:
+            stream_pos = self._events_stream_cache.get_max_pos_of_last_change(room_id)
+            if stream_pos and stream_pos <= min_token:
+                results[room_id] = stream_pos
+            else:
+                missing_room_ids.add(room_id)
+
+        if not missing_room_ids:
+            return results
+
+        # Next, we query the stream position from the DB. At first we fetch all
+        # positions less than the *max* stream pos in the token, then filter
+        # them down. We do this as a) this is a cheaper query, and b) the vast
+        # majority of rooms will have a latest token from before the min stream
+        # pos.
+
+        uncapped_results = await self._bulk_get_max_event_pos(missing_room_ids)
+
+        # Check that the stream position for the rooms are from before the
+        # minimum position of the token. If not then we need to fetch more
+        # rows.
+        recheck_rooms: Set[str] = set()
+        for room_id, stream in uncapped_results.items():
+            if stream <= min_token:
+                results[room_id] = stream
+            else:
+                recheck_rooms.add(room_id)
+
+        if not recheck_rooms:
+            return results
+
+        for room_id in recheck_rooms:
+            result = await self.get_last_event_pos_in_room_before_stream_ordering(
+                room_id, end_token
+            )
+            if result is not None:
+                results[room_id] = result[1].stream
+
+        return results
+
+    @cached()
+    async def _get_max_event_pos(self, room_id: str) -> int:
+        raise NotImplementedError()
+
+    @cachedList(cached_method_name="_get_max_event_pos", list_name="room_ids")
+    async def _bulk_get_max_event_pos(
+        self, room_ids: StrCollection
+    ) -> Mapping[str, int]:
+        """Fetch the max position of a persisted event in the room."""
+
+        now_token = self.get_room_max_token()
+        max_pos = now_token.get_max_stream_pos()
+
+        results: Dict[str, int] = {}
+        missing_room_ids: Set[str] = set()
+        for room_id in room_ids:
+            stream_pos = self._events_stream_cache.get_max_pos_of_last_change(room_id)
+            if stream_pos is not None:
+                results[room_id] = stream_pos
+            else:
+                missing_room_ids.add(room_id)
+
+        if not missing_room_ids:
+            return results
+
+        def bulk_get_max_event_pos_txn(
+            txn: LoggingTransaction, batched_room_ids: StrCollection
+        ) -> Dict[str, int]:
+            clause, args = make_in_list_sql_clause(
+                self.database_engine, "room_id", batched_room_ids
+            )
+            sql = f"""
+                SELECT room_id, (
+                    SELECT stream_ordering FROM events AS e
+                    LEFT JOIN rejections USING (event_id)
+                    WHERE e.room_id = r.room_id
+                        AND e.stream_ordering <= ?
+                        AND NOT outlier
+                        AND rejection_reason IS NULL
+                    ORDER BY stream_ordering DESC
+                    LIMIT 1
+                )
+                FROM rooms AS r
+                WHERE {clause}
+            """
+            txn.execute(sql, [max_pos] + args)
+            return {row[0]: row[1] for row in txn}
+
+        recheck_rooms: Set[str] = set()
+        for batched in batch_iter(room_ids, 1000):
+            batch_results = await self.db_pool.runInteraction(
+                "_bulk_get_max_event_pos", bulk_get_max_event_pos_txn, batched
+            )
+            for room_id, stream_ordering in batch_results.items():
+                if stream_ordering <= now_token.stream:
+                    results.update(batch_results)
+                else:
+                    recheck_rooms.add(room_id)
+
+        for room_id in recheck_rooms:
+            result = await self.get_last_event_pos_in_room_before_stream_ordering(
+                room_id, now_token
+            )
+            if result is not None:
+                results[room_id] = result[1].stream
+
+        return results
+
     async def get_current_room_stream_token_for_room_id(
         self, room_id: str
     ) -> RoomStreamToken:
@@ -1501,7 +1768,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             topological=topological_ordering, stream=stream_ordering
         )
 
-        rows, start_token = self._paginate_room_events_txn(
+        rows, start_token = self._paginate_room_events_by_topological_ordering_txn(
             txn,
             room_id,
             before_token,
@@ -1511,7 +1778,7 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         )
         events_before = [r.event_id for r in rows]
 
-        rows, end_token = self._paginate_room_events_txn(
+        rows, end_token = self._paginate_room_events_by_topological_ordering_txn(
             txn,
             room_id,
             after_token,
@@ -1674,14 +1941,14 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
     def has_room_changed_since(self, room_id: str, stream_id: int) -> bool:
         return self._events_stream_cache.has_entity_changed(room_id, stream_id)
 
-    def _paginate_room_events_txn(
+    def _paginate_room_events_by_topological_ordering_txn(
         self,
         txn: LoggingTransaction,
         room_id: str,
         from_token: RoomStreamToken,
         to_token: Optional[RoomStreamToken] = None,
         direction: Direction = Direction.BACKWARDS,
-        limit: int = -1,
+        limit: int = 0,
         event_filter: Optional[Filter] = None,
     ) -> Tuple[List[_EventDictReturn], RoomStreamToken]:
         """Returns list of events before or after a given token.
@@ -1703,6 +1970,24 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             been reached (i.e. there are no events between `from_token` and
             `to_token`), or `limit` is zero.
         """
+        # We can bail early if we're looking forwards, and our `to_key` is already
+        # before our `from_token`.
+        if (
+            direction == Direction.FORWARDS
+            and to_token is not None
+            and to_token.is_before_or_eq(from_token)
+        ):
+            # Token selection matches what we do below if there are no rows
+            return [], to_token if to_token else from_token
+        # Or vice-versa, if we're looking backwards and our `from_token` is already before
+        # our `to_token`.
+        elif (
+            direction == Direction.BACKWARDS
+            and to_token is not None
+            and from_token.is_before_or_eq(to_token)
+        ):
+            # Token selection matches what we do below if there are no rows
+            return [], to_token if to_token else from_token
 
         args: List[Any] = [room_id]
 
@@ -1787,7 +2072,6 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             "bounds": bounds,
             "order": order,
         }
-
         txn.execute(sql, args)
 
         # Filter the result set.
@@ -1819,27 +2103,30 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
         return rows, next_token
 
     @trace
-    async def paginate_room_events(
+    @tag_args
+    async def paginate_room_events_by_topological_ordering(
         self,
+        *,
         room_id: str,
         from_key: RoomStreamToken,
         to_key: Optional[RoomStreamToken] = None,
         direction: Direction = Direction.BACKWARDS,
-        limit: int = -1,
+        limit: int = 0,
         event_filter: Optional[Filter] = None,
     ) -> Tuple[List[EventBase], RoomStreamToken]:
-        """Returns list of events before or after a given token.
-
-        When Direction.FORWARDS: from_key < x <= to_key
-        When Direction.BACKWARDS: from_key >= x > to_key
+        """
+        Paginate events by `topological_ordering` (tie-break with `stream_ordering`) in
+        the room from the `from_key` in the given `direction` to the `to_key` or
+        `limit`.
 
         Args:
             room_id
-            from_key: The token used to stream from
-            to_key: A token which if given limits the results to only those before
+            from_key: The token to stream from (starting point and heading in the given
+                direction)
+            to_key: The token representing the end stream position (end point)
             direction: Indicates whether we are paginating forwards or backwards
                 from `from_key`.
-            limit: The maximum number of events to return.
+            limit: Maximum number of events to return
             event_filter: If provided filters the events to those that match the filter.
 
         Returns:
@@ -1847,8 +2134,18 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             of the result set. If no events are returned then the end of the
             stream has been reached (i.e. there are no events between `from_key`
             and `to_key`).
+
+            When Direction.FORWARDS: from_key < x <= to_key, (ascending order)
+            When Direction.BACKWARDS: from_key >= x > to_key, (descending order)
         """
 
+        # FIXME: When going forwards, we should enforce that the `to_key` is not `None`
+        # because we always need an upper bound when querying the events stream (as
+        # otherwise we'll potentially pick up events that are not fully persisted).
+
+        # We have these checks outside of the transaction function (txn) to save getting
+        # a DB connection and switching threads if we don't need to.
+        #
         # We can bail early if we're looking forwards, and our `to_key` is already
         # before our `from_key`.
         if (
@@ -1871,8 +2168,8 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             return [], to_key if to_key else from_key
 
         rows, token = await self.db_pool.runInteraction(
-            "paginate_room_events",
-            self._paginate_room_events_txn,
+            "paginate_room_events_by_topological_ordering",
+            self._paginate_room_events_by_topological_ordering_txn,
             room_id,
             from_key,
             to_key,
@@ -1983,3 +2280,14 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
             return RoomStreamToken(stream=last_position.stream - 1)
 
         return None
+
+    @trace
+    def get_rooms_that_might_have_updates(
+        self, room_ids: StrCollection, from_token: RoomStreamToken
+    ) -> StrCollection:
+        """Filters given room IDs down to those that might have updates, i.e.
+        removes rooms that definitely do not have updates.
+        """
+        return self._events_stream_cache.get_entities_changed(
+            room_ids, from_token.stream
+        )

synapse/storage/engines/_base.py

@@ -28,6 +28,11 @@ if TYPE_CHECKING:
     from synapse.storage.database import LoggingDatabaseConnection
 
 
+# A string that will be replaced with the appropriate auto increment directive
+# for the database engine, expands to an auto incrementing integer primary key.
+AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER = "$%AUTO_INCREMENT_PRIMARY_KEY%$"
+
+
 class IsolationLevel(IntEnum):
     READ_COMMITTED: int = 1
     REPEATABLE_READ: int = 2

synapse/storage/engines/postgres.py

@@ -25,6 +25,7 @@ from typing import TYPE_CHECKING, Any, Mapping, NoReturn, Optional, Tuple, cast
 import psycopg2.extensions
 
 from synapse.storage.engines._base import (
+    AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER,
     BaseDatabaseEngine,
     IncorrectDatabaseSetup,
     IsolationLevel,
@@ -256,4 +257,10 @@ class PostgresEngine(
         executing the script in its own transaction. The script transaction is
         left open and it is the responsibility of the caller to commit it.
         """
+        # Replace auto increment placeholder with the appropriate directive
+        script = script.replace(
+            AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER,
+            "BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY",
+        )
+
         cursor.execute(f"COMMIT; BEGIN TRANSACTION; {script}")

synapse/storage/engines/sqlite.py

@@ -25,6 +25,7 @@ import threading
 from typing import TYPE_CHECKING, Any, List, Mapping, Optional
 
 from synapse.storage.engines import BaseDatabaseEngine
+from synapse.storage.engines._base import AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER
 from synapse.storage.types import Cursor
 
 if TYPE_CHECKING:
@@ -168,6 +169,11 @@ class Sqlite3Engine(BaseDatabaseEngine[sqlite3.Connection, sqlite3.Cursor]):
         > first. No other implicit transaction control is performed; any transaction
         > control must be added to sql_script.
         """
+        # Replace auto increment placeholder with the appropriate directive
+        script = script.replace(
+            AUTO_INCREMENT_PRIMARY_KEYPLACEHOLDER, "INTEGER PRIMARY KEY AUTOINCREMENT"
+        )
+
         # The implementation of `executescript` can be found at
         # https://github.com/python/cpython/blob/3.11/Modules/_sqlite/cursor.c#L1035.
         cursor.executescript(f"BEGIN TRANSACTION; {script}")

synapse/storage/schema/__init__.py

@@ -19,7 +19,7 @@
 #
 #
 
-SCHEMA_VERSION = 86  # remember to update the list below when updating
+SCHEMA_VERSION = 87  # remember to update the list below when updating
 """Represents the expectations made by the codebase about the database schema
 
 This should be incremented whenever the codebase changes its requirements on the
@@ -142,6 +142,10 @@ Changes in SCHEMA_VERSION = 85
 
 Changes in SCHEMA_VERSION = 86
     - Add a column `authenticated` to the tables `local_media_repository` and `remote_media_cache`
+
+Changes in SCHEMA_VERSION = 87
+    - Add tables to store Sliding Sync data for quick filtering/sorting
+      (`sliding_sync_joined_rooms`, `sliding_sync_membership_snapshots`)
 """
 
 

synapse/storage/schema/main/delta/86/02_receipts_event_id_index.sql

@@ -0,0 +1,15 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2024 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+    (8602, 'receipts_room_id_event_id_index', '{}');

synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql

@@ -0,0 +1,153 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2024 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+-- This table is a list/queue used to keep track of which rooms need to be inserted into
+-- `sliding_sync_joined_rooms`. We do this to avoid reading from `current_state_events`
+-- during the background update to populate `sliding_sync_joined_rooms` which works but
+-- it takes a lot of work for the database to grab `DISTINCT` room_ids given how many
+-- state events there are for each room.
+CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms_to_recalculate(
+    room_id TEXT NOT NULL REFERENCES rooms(room_id),
+    PRIMARY KEY (room_id)
+);
+
+-- A table for storing room meta data (current state relevant to sliding sync) that the
+-- local server is still participating in (someone local is joined to the room).
+--
+-- We store the joined rooms in separate table from `sliding_sync_membership_snapshots`
+-- because we need up-to-date information for joined rooms and it can be shared across
+-- everyone who is joined.
+--
+-- This table is kept in sync with `current_state_events` which means if the server is
+-- no longer participating in a room, the row will be deleted.
+CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms(
+    room_id TEXT NOT NULL REFERENCES rooms(room_id),
+    -- The `stream_ordering` of the most-recent/latest event in the room
+    event_stream_ordering BIGINT NOT NULL REFERENCES events(stream_ordering),
+    -- The `stream_ordering` of the last event according to the `bump_event_types`
+    bump_stamp BIGINT,
+    -- `m.room.create` -> `content.type` (current state)
+    --
+    -- Useful for the `spaces`/`not_spaces` filter in the Sliding Sync API
+    room_type TEXT,
+    -- `m.room.name` -> `content.name` (current state)
+    --
+    -- Useful for the room meta data and `room_name_like` filter in the Sliding Sync API
+    room_name TEXT,
+    -- `m.room.encryption` -> `content.algorithm` (current state)
+    --
+    -- Useful for the `is_encrypted` filter in the Sliding Sync API
+    is_encrypted BOOLEAN DEFAULT FALSE NOT NULL,
+    -- `m.room.tombstone` -> `content.replacement_room` (according to the current state at the
+    -- time of the membership).
+    --
+    -- Useful for the `include_old_rooms` functionality in the Sliding Sync API
+    tombstone_successor_room_id TEXT,
+    PRIMARY KEY (room_id)
+);
+
+-- So we can purge rooms easily.
+--
+-- The primary key is already `room_id`
+
+-- So we can sort by `stream_ordering
+CREATE UNIQUE INDEX IF NOT EXISTS sliding_sync_joined_rooms_event_stream_ordering ON sliding_sync_joined_rooms(event_stream_ordering);
+
+-- A table for storing a snapshot of room meta data (historical current state relevant
+-- for sliding sync) at the time of a local user's membership. Only has rows for the
+-- latest membership event for a given local user in a room which matches
+-- `local_current_membership` .
+--
+-- We store all memberships including joins. This makes it easy to reference this table
+-- to find all membership for a given user and shares the same semantics as
+-- `local_current_membership`. And we get to avoid some table maintenance; if we only
+-- stored non-joins, we would have to delete the row for the user when the user joins
+-- the room. Stripped state doesn't include the `m.room.tombstone` event, so we just
+-- assume that the room doesn't have a tombstone.
+--
+-- For remote invite/knocks where the server is not participating in the room, we will
+-- use stripped state events to populate this table. We assume that if any stripped
+-- state is given, it will include all possible stripped state events types. For
+-- example, if stripped state is given but `m.room.encryption` isn't included, we will
+-- assume that the room is not encrypted.
+--
+-- We don't include `bump_stamp` here because we can just use the `stream_ordering` from
+-- the membership event itself as the `bump_stamp`.
+CREATE TABLE IF NOT EXISTS sliding_sync_membership_snapshots(
+    room_id TEXT NOT NULL REFERENCES rooms(room_id),
+    user_id TEXT NOT NULL,
+    -- Useful to be able to tell leaves from kicks (where the `user_id` is different from the `sender`)
+    sender TEXT NOT NULL,
+    membership_event_id TEXT NOT NULL REFERENCES events(event_id),
+    membership TEXT NOT NULL,
+    -- This is an integer just to match `room_memberships` and also means we don't need
+    -- to do any casting.
+    forgotten INTEGER DEFAULT 0 NOT NULL,
+    -- `stream_ordering` of the `membership_event_id`
+    event_stream_ordering BIGINT NOT NULL REFERENCES events(stream_ordering),
+    -- For remote invites/knocks that don't include any stripped state, we want to be
+    -- able to distinguish between a room with `None` as valid value for some state and
+    -- room where the state is completely unknown. Basically, this should be True unless
+    -- no stripped state was provided for a remote invite/knock (False).
+    has_known_state BOOLEAN DEFAULT FALSE NOT NULL,
+    -- `m.room.create` -> `content.type` (according to the current state at the time of
+    -- the membership).
+    --
+    -- Useful for the `spaces`/`not_spaces` filter in the Sliding Sync API
+    room_type TEXT,
+    -- `m.room.name` -> `content.name` (according to the current state at the time of
+    -- the membership).
+    --
+    -- Useful for the room meta data and `room_name_like` filter in the Sliding Sync API
+    room_name TEXT,
+    -- `m.room.encryption` -> `content.algorithm` (according to the current state at the
+    -- time of the membership).
+    --
+    -- Useful for the `is_encrypted` filter in the Sliding Sync API
+    is_encrypted BOOLEAN DEFAULT FALSE NOT NULL,
+    -- `m.room.tombstone` -> `content.replacement_room` (according to the current state at the
+    -- time of the membership).
+    --
+    -- Useful for the `include_old_rooms` functionality in the Sliding Sync API
+    tombstone_successor_room_id TEXT,
+    PRIMARY KEY (room_id, user_id)
+);
+
+-- So we can purge rooms easily.
+--
+-- Since we're using a multi-column index as the primary key (room_id, user_id), the
+-- first index column (room_id) is always usable for searching so we don't need to
+-- create a separate index for it.
+--
+-- CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_room_id ON sliding_sync_membership_snapshots(room_id);
+
+-- So we can fetch all rooms for a given user
+CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_user_id ON sliding_sync_membership_snapshots(user_id);
+-- So we can sort by `stream_ordering
+CREATE UNIQUE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_event_stream_ordering ON sliding_sync_membership_snapshots(event_stream_ordering);
+
+
+-- Add a series of background updates to populate the new `sliding_sync_joined_rooms` table:
+--
+--   1. Add a background update to prefill `sliding_sync_joined_rooms_to_recalculate`.
+--      We do a one-shot bulk insert from the `rooms` table to prefill.
+--   2. Add a background update to populate the new `sliding_sync_joined_rooms` table
+--
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+  (8701, 'sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update', '{}');
+INSERT INTO background_updates (ordering, update_name, progress_json, depends_on) VALUES
+  (8701, 'sliding_sync_joined_rooms_bg_update', '{}', 'sliding_sync_prefill_joined_rooms_to_recalculate_table_bg_update');
+
+-- Add a background updates to populate the new `sliding_sync_membership_snapshots` table
+INSERT INTO background_updates (ordering, update_name, progress_json) VALUES
+  (8701, 'sliding_sync_membership_snapshots_bg_update', '{}');

synapse/storage/schema/main/delta/87/02_per_connection_state.sql

@@ -0,0 +1,78 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2024 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+
+-- Table to track active sliding sync connections.
+--
+-- A new connection will be created for every sliding sync request without a
+-- `since` token for a given `conn_id` for a device.#
+--
+-- Once a new connection is created and used we delete all other connections for
+-- the `conn_id`.
+CREATE TABLE sliding_sync_connections(
+    connection_key $%AUTO_INCREMENT_PRIMARY_KEY%$,
+    user_id TEXT NOT NULL,
+    device_id TEXT NOT NULL,
+    conn_id TEXT NOT NULL,
+    created_ts BIGINT NOT NULL
+);
+
+CREATE INDEX sliding_sync_connections_idx ON sliding_sync_connections(user_id, device_id, conn_id);
+
+-- We track per-connection state by associating changes to the state with
+-- connection positions. This ensures that we correctly track state even if we
+-- see retries of requests.
+--
+-- If the client starts a "new" connection (by not specifying a since token),
+-- we'll clear out the other connections (to ensure that we don't end up with
+-- lots of connection keys).
+CREATE TABLE sliding_sync_connection_positions(
+    connection_position $%AUTO_INCREMENT_PRIMARY_KEY%$,
+    connection_key BIGINT NOT NULL REFERENCES sliding_sync_connections(connection_key) ON DELETE CASCADE,
+    created_ts BIGINT NOT NULL
+);
+
+CREATE INDEX sliding_sync_connection_positions_key ON sliding_sync_connection_positions(connection_key);
+
+
+-- To save space we deduplicate the `required_state` json by assigning IDs to
+-- different values.
+CREATE TABLE sliding_sync_connection_required_state(
+    required_state_id $%AUTO_INCREMENT_PRIMARY_KEY%$,
+    connection_key BIGINT NOT NULL REFERENCES sliding_sync_connections(connection_key) ON DELETE CASCADE,
+    required_state TEXT NOT NULL  -- We store this as a json list of event type / state key tuples.
+);
+
+CREATE INDEX sliding_sync_connection_required_state_conn_pos ON sliding_sync_connections(connection_key);
+
+
+-- Stores the room configs we have seen for rooms in a connection.
+CREATE TABLE sliding_sync_connection_room_configs(
+    connection_position BIGINT NOT NULL REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE,
+    room_id TEXT NOT NULL,
+    timeline_limit BIGINT NOT NULL,
+    required_state_id BIGINT NOT NULL REFERENCES sliding_sync_connection_required_state(required_state_id)
+);
+
+CREATE UNIQUE INDEX sliding_sync_connection_room_configs_idx ON sliding_sync_connection_room_configs(connection_position, room_id);
+
+-- Stores what data we have sent for given streams down given connections.
+CREATE TABLE sliding_sync_connection_streams(
+    connection_position BIGINT NOT NULL REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE,
+    stream TEXT NOT NULL,  -- e.g. "events" or "receipts"
+    room_id TEXT NOT NULL,
+    room_status TEXT NOT NULL,  -- "live" or "previously", i.e. the `HaveSentRoomFlag` value
+    last_position TEXT  -- For "previously" the token for the stream we have sent up to.
+);
+
+CREATE UNIQUE INDEX sliding_sync_connection_streams_idx ON sliding_sync_connection_streams(connection_position, room_id, stream);

synapse/storage/schema/main/delta/87/03_per_connection_lists.sql

@@ -0,0 +1,22 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2024 New Vector, Ltd
+--
+-- This program is free software: you can redistribute it and/or modify
+-- it under the terms of the GNU Affero General Public License as
+-- published by the Free Software Foundation, either version 3 of the
+-- License, or (at your option) any later version.
+--
+-- See the GNU Affero General Public License for more details:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+
+-- Stores the room lists for a connection
+CREATE TABLE sliding_sync_connection_room_lists(
+    connection_key BIGINT NOT NULL REFERENCES sliding_sync_connections(connection_key) ON DELETE CASCADE,
+    list_name TEXT NOT NULL,
+    room_id TEXT NOT NULL
+);
+
+CREATE INDEX sliding_sync_connection_room_lists_idx ON sliding_sync_connection_room_lists(connection_key);

synapse/types/__init__.py

@@ -777,6 +777,13 @@ class RoomStreamToken(AbstractMultiWriterStreamToken):
 
         return super().bound_stream_token(max_stream)
 
+    def __str__(self) -> str:
+        instances = ", ".join(f"{k}: {v}" for k, v in sorted(self.instance_map.items()))
+        return (
+            f"RoomStreamToken(stream: {self.stream}, topological: {self.topological}, "
+            f"instances: {{{instances}}})"
+        )
+
 
 @attr.s(frozen=True, slots=True, order=False)
 class MultiWriterStreamToken(AbstractMultiWriterStreamToken):
@@ -873,6 +880,13 @@ class MultiWriterStreamToken(AbstractMultiWriterStreamToken):
 
         return True
 
+    def __str__(self) -> str:
+        instances = ", ".join(f"{k}: {v}" for k, v in sorted(self.instance_map.items()))
+        return (
+            f"MultiWriterStreamToken(stream: {self.stream}, "
+            f"instances: {{{instances}}})"
+        )
+
 
 class StreamKeyType(Enum):
     """Known stream types.
@@ -1131,12 +1145,64 @@ class StreamToken:
 
         return True
 
+    def __str__(self) -> str:
+        return (
+            f"StreamToken(room: {self.room_key}, presence: {self.presence_key}, "
+            f"typing: {self.typing_key}, receipt: {self.receipt_key}, "
+            f"account_data: {self.account_data_key}, push_rules: {self.push_rules_key}, "
+            f"to_device: {self.to_device_key}, device_list: {self.device_list_key}, "
+            f"groups: {self.groups_key}, un_partial_stated_rooms: {self.un_partial_stated_rooms_key})"
+        )
+
 
 StreamToken.START = StreamToken(
     RoomStreamToken(stream=0), 0, 0, MultiWriterStreamToken(stream=0), 0, 0, 0, 0, 0, 0
 )
 
 
+@attr.s(slots=True, frozen=True, auto_attribs=True)
+class SlidingSyncStreamToken:
+    """The same as a `StreamToken`, but includes an extra field at the start for
+    the sliding sync connection token (separated by a '/'). This is used to
+    store per-connection state.
+
+    This then looks something like:
+        5/s2633508_17_338_6732159_1082514_541479_274711_265584_1_379
+
+    Attributes:
+        stream_token: Token representing the position of all the standard
+            streams.
+        connection_position: Token used by sliding sync to track updates to any
+            per-connection state stored by Synapse.
+    """
+
+    stream_token: StreamToken
+    connection_position: int
+
+    @staticmethod
+    @cancellable
+    async def from_string(store: "DataStore", string: str) -> "SlidingSyncStreamToken":
+        """Creates a SlidingSyncStreamToken from its textual representation."""
+        try:
+            connection_position_str, stream_token_str = string.split("/", 1)
+            connection_position = int(connection_position_str)
+            stream_token = await StreamToken.from_string(store, stream_token_str)
+
+            return SlidingSyncStreamToken(
+                stream_token=stream_token,
+                connection_position=connection_position,
+            )
+        except CancelledError:
+            raise
+        except Exception:
+            raise SynapseError(400, "Invalid stream token")
+
+    async def to_string(self, store: "DataStore") -> str:
+        """Serializes the token to a string"""
+        stream_token_str = await self.stream_token.to_string(store)
+        return f"{self.connection_position}/{stream_token_str}"
+
+
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class PersistedPosition:
     """Position of a newly persisted row with instance that persisted it."""
@@ -1219,11 +1285,12 @@ class ReadReceipt:
 @attr.s(slots=True, frozen=True, auto_attribs=True)
 class DeviceListUpdates:
     """
-    An object containing a diff of information regarding other users' device lists, intended for
-    a recipient to carry out device list tracking.
+    An object containing a diff of information regarding other users' device lists,
+    intended for a recipient to carry out device list tracking.
 
     Attributes:
-        changed: A set of users whose device lists have changed recently.
+        changed: A set of users who have updated their device identity or
+            cross-signing keys, or who now share an encrypted room with.
         left: A set of users who the recipient no longer needs to track the device lists of.
             Typically when those users no longer share any end-to-end encryption enabled rooms.
     """