Merge branch 'develop' into madlittlemods/rust-db-access-using-python-db-pool

Conflicts: rust/src/lib.rs synapse/storage/util/id_generators.py
2026-08-28 09:24:41 +00:00 · 2026-06-09 17:52:16 -05:00
parent 287c0656d6 f53f104449
commit 4e555ac022
53 changed files with 1051 additions and 196 deletions
@@ -1,3 +1,29 @@
+# Synapse 1.155.0rc1 (2026-06-09)
+
+## Bugfixes
+
+- Limit the to-device EDU size to a reasonable value to mitigate long queues of to-device messages preventing outgoing federation because of the size of the transaction. ([\#19617](https://github.com/element-hq/synapse/issues/19617))
+- Work around bug that sometimes breaks joining restricted rooms that require a remote join. Contributed by @tulir @ Beeper. ([\#19730](https://github.com/element-hq/synapse/issues/19730))
+- Update Sliding Sync to return a new response immediately if a room subscription has changed and produced a new response. ([\#19734](https://github.com/element-hq/synapse/issues/19734), [\#19792](https://github.com/element-hq/synapse/issues/19792))
+- Fix the `/capabilities` endpoint returning a 500 error on non-media workers when [MSC4452: Preview URL capabilities API](https://github.com/matrix-org/matrix-spec-proposals/pull/4452) is enabled. ([\#19839](https://github.com/element-hq/synapse/issues/19839))
+
+## Improved Documentation
+
+- Document how to see Rust build failure output when using `poetry install`. ([\#19818](https://github.com/element-hq/synapse/issues/19818))
+- Document that the SQLite version included in Ubuntu LTS, aside from ESM-only versions, is included in our support policy. ([\#19823](https://github.com/element-hq/synapse/issues/19823))
+
+## Internal Changes
+
+- Port the Python Event classes to Rust. ([\#19701](https://github.com/element-hq/synapse/issues/19701), [\#19816](https://github.com/element-hq/synapse/issues/19816), [\#19817](https://github.com/element-hq/synapse/issues/19817), [\#19819](https://github.com/element-hq/synapse/issues/19819))
+- Added tests to ensure that email notification links are sanitized. Contributed by Noah Markert. ([\#19741](https://github.com/element-hq/synapse/issues/19741))
+- Add `GcpJsonFormatter` logging formatter for use with Google Cloud Logging and GKE deployments. ([\#19775](https://github.com/element-hq/synapse/issues/19775))
+- Add more logging to the to-device message replication stream. ([\#19801](https://github.com/element-hq/synapse/issues/19801), [\#19821](https://github.com/element-hq/synapse/issues/19821))
+- Port `Requester` class to Rust. ([\#19828](https://github.com/element-hq/synapse/issues/19828))
+
+# Synapse 1.154.0 (2026-06-04)
+
+No significant changes since 1.154.0rc1.
+
 # Synapse 1.154.0rc1 (2026-05-27)

 ## Features
@@ -1 +0,0 @@
-A long queue of to-device messages could prevent outgoing federation because of the size of the transaction, let's limit the to-device EDU size to a reasonable value.
@@ -1 +0,0 @@
-Port the python Event classes to Rust.
@@ -1 +0,0 @@
-Work around bug that sometimes breaks joining restricted rooms that require a remote join. Contributed by @tulir @ Beeper.
@@ -1 +0,0 @@
-Added tests to ensure that email notification links are sanitized. Contributed by Noah Markert.
@@ -1 +0,0 @@
-Add `GcpJsonFormatter` logging formatter for use with Google Cloud Logging and GKE deployments.
@@ -1 +0,0 @@
-Add more logging to the to-device message replication stream.
@@ -1 +0,0 @@
-Port the python Event classes to Rust.
@@ -1 +0,0 @@
-Port the python Event classes to Rust.
@@ -1 +0,0 @@
-Document how to see Rust build failure output when using `poetry install`.
@@ -1 +0,0 @@
-Port the python Event classes to Rust.
@@ -1,3 +1,15 @@
+matrix-synapse-py3 (1.155.0~rc1) stable; urgency=medium
+
+  * New Synapse release 1.155.0rc1.
+
+ -- Synapse Packaging team <packages@matrix.org>  Tue, 09 Jun 2026 14:58:03 +0100
+
+matrix-synapse-py3 (1.154.0) stable; urgency=medium
+
+  * New Synapse release 1.154.0.
+
+ -- Synapse Packaging team <packages@matrix.org>  Thu, 04 Jun 2026 14:16:23 +0100
+
 matrix-synapse-py3 (1.154.0~rc1) stable; urgency=medium

  * New Synapse release 1.154.0rc1.
@@ -22,7 +22,7 @@ people building from source should ensure they can fetch recent versions of Rust

 The oldest supported version of SQLite is the version
 [provided](https://packages.debian.org/oldstable/libsqlite3-0) by
-[Debian oldstable](https://wiki.debian.org/DebianOldStable).
+[Debian oldstable](https://wiki.debian.org/DebianOldStable) or the oldest maintenance/security-supported [Ubuntu LTS](https://endoflife.date/ubuntu) (Ubuntu versions with only Expanded Security Maintenance are not included).


 ### Context
@@ -1,6 +1,6 @@
 [project]
 name = "matrix-synapse"
-version = "1.154.0rc1"
+version = "1.155.0rc1"
 description = "Homeserver for the Matrix decentralised comms protocol"
 readme = "README.rst"
 authors = [
@@ -22,6 +22,7 @@ pub mod rendezvous;
 pub mod room_versions;
 pub mod segmenter;
 pub mod storage;
+pub mod types;

 lazy_static! {
    static ref LOGGING_HANDLE: ResetHandle = pyo3_log::init();
@@ -75,6 +76,7 @@ fn synapse_rust(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
    msc4388_rendezvous::register_module(py, m)?;
    segmenter::register_module(py, m)?;
    room_versions::register_module(py, m)?;
+    types::register_module(py, m)?;

    Ok(())
 }
@@ -0,0 +1,340 @@
+/*
+ * This file is licensed under the Affero General Public License (AGPL) version 3.
+ *
+ * Copyright (C) 2026 Element Creations 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>.
+ *
+ */
+
+//! Rust implementations of types from `synapse.types`.
+
+use std::collections::HashSet;
+
+use once_cell::sync::OnceCell;
+use pyo3::prelude::*;
+use pyo3::{
+    exceptions::PyKeyError,
+    types::{PyDict, PyDictMethods, PyList},
+};
+
+/// A reference to the `synapse.types.UserID` class.
+static USER_ID_CLASS: OnceCell<Py<PyAny>> = OnceCell::new();
+
+/// Access to the `synapse.types.UserID` class.
+fn user_id_class(py: Python<'_>) -> PyResult<&Bound<'_, PyAny>> {
+    Ok(USER_ID_CLASS
+        .get_or_try_init(|| -> PyResult<_> {
+            Ok(py.import("synapse.types")?.getattr("UserID")?.unbind())
+        })?
+        .bind(py))
+}
+
+/// Represents the user making a request.
+#[pyclass(frozen, skip_from_py_object, get_all, eq)]
+#[derive(Debug, PartialEq, Eq)]
+pub struct Requester {
+    /// The ID of the user making the request, in string form (see
+    /// [`Self::user`] for accessing the parsed `UserID`).
+    user_id: String,
+    /// The ID of the access token used for this request, or None for
+    /// appservices, guests, and tokens generated by the admin API
+    access_token_id: Option<i64>,
+    /// True if the user making this request is a guest
+    is_guest: bool,
+    /// Any scopes associated with the access token used for this request, or an
+    /// empty set if no token or a non-oauth token was used
+    scope: HashSet<String>,
+    /// True if the user making this request is shadow banned
+    shadow_banned: bool,
+    /// The device_id which was set at authentication time, or None for
+    /// appservices, guests, and tokens generated by the admin API
+    device_id: Option<String>,
+    /// The ID of the AS requesting on behalf of the user, or None.
+    app_service_id: Option<String>,
+    /// The entity that authenticated when making the request.
+    ///
+    /// This is different to the `user_id` when an admin user or the server is
+    /// "puppeting" the user.
+    authenticated_entity: String,
+}
+
+#[pymethods]
+impl Requester {
+    #[new]
+    #[pyo3(signature = (
+        user,
+        access_token_id,
+        is_guest,
+        scope,
+        shadow_banned,
+        device_id,
+        app_service_id,
+        authenticated_entity,
+    ))]
+    #[allow(clippy::too_many_arguments)]
+    fn new(
+        user: &Bound<'_, PyAny>,
+        access_token_id: Option<i64>,
+        is_guest: &Bound<'_, PyAny>,
+        scope: HashSet<String>,
+        shadow_banned: &Bound<'_, PyAny>,
+        device_id: Option<String>,
+        app_service_id: Option<String>,
+        authenticated_entity: String,
+    ) -> PyResult<Self> {
+        // The `user` argument should be a `UserID`, which has a `to_string` for
+        // getting the string form.
+        let user_id = user.call_method0("to_string")?.extract::<String>()?;
+
+        // The `is_guest` and `shadow_banned` arguments are expected to be
+        // Python bools, but unfortunately Synapse often passes them as truthy
+        // values (mainly due to reading from SQLite, which returns 0/1 for
+        // bools).
+        let is_guest = is_guest.is_truthy()?;
+        let shadow_banned = shadow_banned.is_truthy()?;
+
+        Ok(Requester {
+            user_id,
+            access_token_id,
+            is_guest,
+            scope,
+            shadow_banned,
+            device_id,
+            app_service_id,
+            authenticated_entity,
+        })
+    }
+
+    /// The user making the request, as a Python `UserID`.
+    #[getter]
+    fn user<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
+        user_id_class(py)?.call_method1("from_string", (&self.user_id,))
+    }
+
+    /// Converts self to a type that can be serialized as JSON, and then
+    /// deserialized by [`Self::deserialize`]
+    fn serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyDict>> {
+        let dict = PyDict::new(py);
+        dict.set_item("user_id", &self.user_id)?;
+        dict.set_item("access_token_id", self.access_token_id)?;
+        dict.set_item("is_guest", self.is_guest)?;
+        dict.set_item("scope", PyList::new(py, &self.scope)?)?;
+        dict.set_item("shadow_banned", self.shadow_banned)?;
+        dict.set_item("device_id", self.device_id.as_deref())?;
+        // NB: the wire key is "app_server_id" (server, not service). Changing
+        // this is non-trivial as it would break replication during a rolling
+        // upgrade.
+        dict.set_item("app_server_id", self.app_service_id.as_deref())?;
+        dict.set_item("authenticated_entity", &self.authenticated_entity)?;
+        Ok(dict)
+    }
+
+    /// Converts a dict that was produced by [`Self::serialize`] back into a
+    /// [`Requester`].
+    #[staticmethod]
+    fn deserialize(py: Python<'_>, input: &Bound<'_, PyAny>) -> PyResult<Self> {
+        let user_id = input.get_item("user_id")?.extract::<String>()?;
+        let access_token_id = input
+            .get_item("access_token_id")?
+            .extract::<Option<i64>>()?;
+        let is_guest = input.get_item("is_guest")?.is_truthy()?;
+
+        // `serialize` stores the scope as a list, so extract it as a `Vec`
+        // (which accepts any sequence) and collect into a set. For backwards
+        // compatibility, "scope" is optional and defaults to an empty set if
+        // not present.
+        let scope = match input.get_item("scope") {
+            Ok(scope) => scope.extract::<Vec<String>>()?.into_iter().collect(),
+            Err(err) if err.is_instance_of::<PyKeyError>(py) => HashSet::new(),
+            Err(err) => return Err(err),
+        };
+
+        let shadow_banned = input.get_item("shadow_banned")?.is_truthy()?;
+        let device_id = input.get_item("device_id")?.extract::<Option<String>>()?;
+
+        // The wire key is "app_server_id", not "app_service_id" — see `serialize`.
+        let app_service_id = input
+            .get_item("app_server_id")?
+            .extract::<Option<String>>()?;
+
+        let authenticated_entity = input
+            .get_item("authenticated_entity")?
+            .extract::<String>()?;
+
+        Ok(Requester {
+            user_id,
+            access_token_id,
+            is_guest,
+            scope,
+            shadow_banned,
+            device_id,
+            app_service_id,
+            authenticated_entity,
+        })
+    }
+
+    fn __repr__(&self) -> String {
+        format!(
+            "Requester(user_id={}, access_token_id={:?}, is_guest={}, scope={:?}, \
+             shadow_banned={}, device_id={:?}, app_service_id={:?}, authenticated_entity={})",
+            self.user_id,
+            self.access_token_id,
+            self.is_guest,
+            self.scope,
+            self.shadow_banned,
+            self.device_id,
+            self.app_service_id,
+            self.authenticated_entity,
+        )
+    }
+}
+
+/// Called when registering modules with python.
+pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
+    let child_module = PyModule::new(py, "types")?;
+    child_module.add_class::<Requester>()?;
+
+    m.add_submodule(&child_module)?;
+
+    // We need to manually add the module to sys.modules to make `from
+    // synapse.synapse_rust.types import Requester` work.
+    py.import("sys")?
+        .getattr("modules")?
+        .set_item("synapse.synapse_rust.types", child_module)?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A `Requester` with every field populated, for use as a test fixture.
+    fn sample_requester() -> Requester {
+        Requester {
+            user_id: "@alice:example.com".to_string(),
+            access_token_id: Some(42),
+            is_guest: false,
+            scope: HashSet::from(["urn:matrix:client:api:*".to_string()]),
+            shadow_banned: false,
+            device_id: Some("ABCDEFG".to_string()),
+            app_service_id: None,
+            authenticated_entity: "@alice:example.com".to_string(),
+        }
+    }
+
+    #[test]
+    fn test_serialize() {
+        Python::initialize();
+        Python::attach(|py| -> Result<(), PyErr> {
+            let requester = sample_requester();
+            let dict = requester.serialize(py)?;
+            let dict = dict.as_any();
+
+            assert_eq!(
+                dict.get_item("user_id")?.extract::<String>()?,
+                "@alice:example.com"
+            );
+            assert_eq!(dict.get_item("access_token_id")?.extract::<i64>()?, 42);
+            assert!(!dict.get_item("is_guest")?.extract::<bool>()?);
+            assert_eq!(
+                dict.get_item("scope")?.extract::<Vec<String>>()?,
+                vec!["urn:matrix:client:api:*".to_string()]
+            );
+            assert!(!dict.get_item("shadow_banned")?.extract::<bool>()?);
+            assert_eq!(
+                dict.get_item("device_id")?.extract::<Option<String>>()?,
+                Some("ABCDEFG".to_string())
+            );
+            assert_eq!(
+                dict.get_item("authenticated_entity")?.extract::<String>()?,
+                "@alice:example.com"
+            );
+
+            // The `app_service_id` field is serialized under the wire key
+            // "app_server_id" (server, not service), and there must be no
+            // "app_service_id" key.
+            assert!(dict.get_item("app_server_id")?.is_none());
+            assert!(!dict.contains("app_service_id")?);
+
+            Ok(())
+        })
+        .unwrap();
+    }
+
+    #[test]
+    fn test_serialize_deserialize_round_trip() {
+        Python::initialize();
+        Python::attach(|py| {
+            // Use a requester that exercises the optional fields and the
+            // `app_service_id` -> "app_server_id" wire-key mapping.
+            let requester = Requester {
+                user_id: "@bob:example.com".to_string(),
+                access_token_id: None,
+                is_guest: true,
+                scope: HashSet::from(["a".to_string(), "b".to_string()]),
+                shadow_banned: true,
+                device_id: None,
+                app_service_id: Some("my_appservice".to_string()),
+                authenticated_entity: "@admin:example.com".to_string(),
+            };
+
+            let dict = requester.serialize(py).unwrap();
+            let deserialized = Requester::deserialize(py, dict.as_any()).unwrap();
+            let deserialized = Bound::new(py, deserialized).unwrap();
+
+            assert_eq!(&requester, deserialized.get());
+        });
+    }
+
+    #[test]
+    fn test_deserialize_defaults_scope_when_missing() {
+        Python::initialize();
+        Python::attach(|py| {
+            // An older serialized form may omit "scope"; it should default to
+            // an empty set rather than erroring.
+            let dict = PyDict::new(py);
+            dict.set_item("user_id", "@alice:example.com").unwrap();
+            dict.set_item("access_token_id", py.None()).unwrap();
+            dict.set_item("is_guest", false).unwrap();
+            dict.set_item("shadow_banned", false).unwrap();
+            dict.set_item("device_id", py.None()).unwrap();
+            dict.set_item("app_server_id", py.None()).unwrap();
+            dict.set_item("authenticated_entity", "@alice:example.com")
+                .unwrap();
+
+            let requester = Requester::deserialize(py, dict.as_any()).unwrap();
+            assert!(requester.scope.is_empty());
+        });
+    }
+
+    #[test]
+    fn test_deserialize_coerces_truthy_bools() {
+        Python::initialize();
+        Python::attach(|py| {
+            // SQLite returns 0/1 for booleans, so non-bool truthy values must
+            // be coerced for `is_guest` and `shadow_banned`.
+            let dict = PyDict::new(py);
+            dict.set_item("user_id", "@alice:example.com").unwrap();
+            dict.set_item("access_token_id", py.None()).unwrap();
+            dict.set_item("is_guest", 1).unwrap();
+            dict.set_item("scope", PyList::empty(py)).unwrap();
+            dict.set_item("shadow_banned", 0).unwrap();
+            dict.set_item("device_id", py.None()).unwrap();
+            dict.set_item("app_server_id", py.None()).unwrap();
+            dict.set_item("authenticated_entity", "@alice:example.com")
+                .unwrap();
+
+            let requester = Requester::deserialize(py, dict.as_any()).unwrap();
+            assert!(requester.is_guest);
+            assert!(!requester.shadow_banned);
+        });
+    }
+}
@@ -1,5 +1,5 @@
 $schema: https://element-hq.github.io/synapse/latest/schema/v1/meta.schema.json
-$id: https://element-hq.github.io/synapse/schema/synapse/v1.154/synapse-config.schema.json
+$id: https://element-hq.github.io/synapse/schema/synapse/v1.155/synapse-config.schema.json
 type: object
 properties:
  modules:
@@ -371,7 +371,9 @@ class BaseAuth:
        """
        ip_addr = request.get_client_ip_if_available()

-        if ip_addr and (not requester.app_service or self._track_appservice_user_ips):
+        if ip_addr and (
+            not requester.app_service_id or self._track_appservice_user_ips
+        ):
            user_agent = get_request_user_agent(request)
            access_token = self.get_access_token_from_request(request)

@@ -381,7 +383,7 @@ class BaseAuth:
            # table during the transition
            recorded_device_id = (
                "dummy-device"
-                if requester.device_id is None and requester.app_service is not None
+                if requester.device_id is None and requester.app_service_id is not None
                else requester.device_id
            )
            await self.store.insert_client_ip(
@@ -106,8 +106,8 @@ class InternalAuth(BaseAuth):
                parent_span.set_tag("user_id", requester.user.to_string())
                if requester.device_id is not None:
                    parent_span.set_tag("device_id", requester.device_id)
-                if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                if requester.app_service_id is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service_id)
            return requester

    async def get_user_by_req_experimental_feature(
@@ -311,8 +311,8 @@ class MasDelegatedAuth(BaseAuth):
                parent_span.set_tag("user_id", requester.user.to_string())
                if requester.device_id is not None:
                    parent_span.set_tag("device_id", requester.device_id)
-                if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                if requester.app_service_id is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service_id)
            return requester

    async def get_user_by_access_token(
@@ -420,8 +420,8 @@ class MSC3861DelegatedAuth(BaseAuth):
                parent_span.set_tag("user_id", requester.user.to_string())
                if requester.device_id is not None:
                    parent_span.set_tag("device_id", requester.device_id)
-                if requester.app_service is not None:
-                    parent_span.set_tag("appservice_id", requester.app_service.id)
+                if requester.app_service_id is not None:
+                    parent_span.set_tag("appservice_id", requester.app_service_id)
            return requester

    async def _wrapped_get_user_by_req(
@@ -88,7 +88,7 @@ class AuthBlocking:
                # We never block the server from doing actions on behalf of
                # users.
                return
-            if requester.app_service and not self._track_appservice_user_ips:
+            if requester.app_service_id and not self._track_appservice_user_ips:
                # If we're authenticated as an appservice then we only block
                # auth if `track_appservice_user_ips` is set, as that option
                # implicitly means that application services are part of MAU
@@ -163,7 +163,12 @@ class Ratelimiter:
        if requester:
            # Disable rate limiting of users belonging to any AS that is configured
            # not to be rate limited in its registration file (rate_limited: true|false).
-            if requester.app_service and not requester.app_service.is_rate_limited():
+            app_service = (
+                self.store.get_app_service_by_id(requester.app_service_id)
+                if requester.app_service_id
+                else None
+            )
+            if app_service and not app_service.is_rate_limited():
                return True, -1.0

            # Check if ratelimiting has been disabled for the user.
@@ -139,6 +139,10 @@ class ContentRepositoryConfig(Config):
    section = "media"

    def read_config(self, config: JsonDict, **kwargs: Any) -> None:
+        # We need to set this configuration flag even if this worker
+        # is not a media repo worker, as it's exposed in `/capabilities`
+        self.url_preview_enabled = bool(config.get("url_preview_enabled", False))
+
        # Only enable the media repo if either the media repo is enabled or the
        # current worker app is the media repo.
        if (
@@ -242,7 +246,6 @@ class ContentRepositoryConfig(Config):
        self.thumbnail_requirements = parse_thumbnail_requirements(
            config.get("thumbnail_sizes", DEFAULT_THUMBNAIL_SIZES)
        )
-        self.url_preview_enabled = bool(config.get("url_preview_enabled", False))

        if self.url_preview_enabled:
            check_requirements("url-preview")
@@ -419,7 +419,7 @@ def _serialize_event(
                        and event_token_id == config.requester.access_token_id
                    )
                    or config.requester.is_guest
-                    or config.requester.app_service
+                    or config.requester.app_service_id
                ):
                    d["unsigned"]["transaction_id"] = txn_id

@@ -427,7 +427,7 @@ class AdminHandler:

        r = task.params.get("requester")
        assert r is not None
-        admin = Requester.deserialize(self._store, r)
+        admin = Requester.deserialize(r)

        user_id = task.params.get("user_id")
        assert user_id is not None
@@ -132,7 +132,11 @@ class DirectoryHandler:
                Codes.INVALID_PARAM,
            )

-        service = requester.app_service
+        service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
        if service:
            if not service.is_room_alias_in_namespace(room_alias_str):
                raise SynapseError(
@@ -343,7 +343,7 @@ class MessageHandler:
        Returns:
            A dict of user_id to profile info
        """
-        if not requester.app_service:
+        if not requester.app_service_id:
            # We check AS auth after fetching the room membership, as it
            # requires us to pull out all joined members anyway.
            membership, _ = await self.auth.check_user_in_room_or_world_readable(
@@ -365,12 +365,14 @@ class MessageHandler:
        # If this is an AS, double check that they are allowed to see the members.
        # This can either be because the AS user is in the room or because there
        # is a user in the room that the AS is "interested in"
-        if (
-            requester.app_service
-            and requester.user.to_string() not in users_with_profile
-        ):
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        if app_service and requester.user.to_string() not in users_with_profile:
            for uid in users_with_profile:
-                if requester.app_service.is_interested_in_user(uid):
+                if app_service.is_interested_in_user(uid):
                    break
            else:
                # Loop fell through, AS has no interested users in room
@@ -846,7 +848,7 @@ class EventCreationHandler:
            return

        # exempt AS users from needing consent
-        if requester.app_service is not None:
+        if requester.app_service_id is not None:
            return

        user_id = requester.authenticated_entity
@@ -1425,8 +1427,10 @@ class EventCreationHandler:
            else:
                context = await self.state.calculate_context_info(event)

-        if requester:
-            context.app_service = requester.app_service
+        if requester and requester.app_service_id:
+            context.app_service = self.store.get_app_service_by_id(
+                requester.app_service_id
+            )

        res, new_content = await self._third_party_event_rules.check_event_allowed(
            event, context
@@ -661,8 +661,8 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
        key = (room_id,)

        as_id = object()
-        if requester.app_service:
-            as_id = requester.app_service.id
+        if requester.app_service_id:
+            as_id = requester.app_service_id

        # We first linearise by the application service (to try to limit concurrent joins
        # by application services), and then by room ID.
@@ -184,34 +184,45 @@ class SlidingSyncHandler:
                timeout_ms -= after_wait_ts - before_wait_ts
                timeout_ms = max(timeout_ms, 0)

-        # We're going to respond immediately if the timeout is 0 or if this is an
-        # initial sync (without a `from_token`) so we can avoid calling
-        # `notifier.wait_for_events()`.
-        if timeout_ms == 0 or from_token is None:
-            now_token = self.event_sources.get_current_token()
-            result = await self.current_sync_for_user(
+        # Compute a response immediately. We always need to do this before
+        # waiting for new data (unlike in /v3/sync), as the request config might
+        # have changed (e.g. new room subscriptions, etc).
+        now_token = self.event_sources.get_current_token()
+        result = await self.current_sync_for_user(
+            sync_config,
+            from_token=from_token,
+            to_token=now_token,
+        )
+
+        # Return immediately if we have a result, the timeout is 0, or this is
+        # an initial sync.
+        if result or timeout_ms == 0 or from_token is None:
+            return result, did_wait
+
+        # Otherwise, we wait for something to happen and report it to the user.
+        async def current_sync_callback(
+            before_token: StreamToken, after_token: StreamToken
+        ) -> SlidingSyncResult:
+            return await self.current_sync_for_user(
                sync_config,
                from_token=from_token,
-                to_token=now_token,
+                to_token=after_token,
            )
-        else:
-            # Otherwise, we wait for something to happen and report it to the user.
-            async def current_sync_callback(
-                before_token: StreamToken, after_token: StreamToken
-            ) -> SlidingSyncResult:
-                return await self.current_sync_for_user(
-                    sync_config,
-                    from_token=from_token,
-                    to_token=after_token,
-                )

-            result = await self.notifier.wait_for_events(
-                sync_config.user.to_string(),
-                timeout_ms,
-                current_sync_callback,
-                from_token=from_token.stream_token,
-            )
-            did_wait = True
+        result = await self.notifier.wait_for_events(
+            sync_config.user.to_string(),
+            timeout_ms,
+            current_sync_callback,
+            # We *wait* from `now_token` as we have already computed the sync
+            # response up to `now_token` above, so as a minor optimization, we
+            # can wait for something new to arrive after `now_token`.
+            #
+            # We still generate the sync response using `from_token` in the
+            # callback above though, as to generate the correct response it
+            # needs to know the "real" `from_token`.
+            from_token=now_token,
+        )
+        did_wait = True

        return result, did_wait

@@ -852,11 +852,15 @@ class SlidingSyncRoomLists:
                            previous_connection_state.room_configs.get(room_id)
                        )
                        if prev_room_sync_config is not None:
-                            # Always include rooms whose timeline limit has increased.
-                            # (see the "XXX: Odd behavior" described below)
+                            # Always include rooms whose effective config has
+                            # expanded. This covers timeline-limit increases and
+                            # required-state additions introduced by room
+                            # subscriptions overriding list-derived params.
                            if (
-                                prev_room_sync_config.timeline_limit
-                                < room_config.timeline_limit
+                                prev_room_sync_config.combine_room_sync_config(
+                                    room_config
+                                )
+                                != prev_room_sync_config
                            ):
                                rooms_should_send.add(room_id)
                                continue
@@ -89,7 +89,7 @@ class ReplicationRemoteJoinRestServlet(ReplicationEndpoint):
        remote_room_hosts = content["remote_room_hosts"]
        event_content = content["content"]

-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
        request.requester = requester

        logger.info("remote_join: %s into room: %s", user_id, room_id)
@@ -153,7 +153,7 @@ class ReplicationRemoteKnockRestServlet(ReplicationEndpoint):
        remote_room_hosts = content["remote_room_hosts"]
        event_content = content["content"]

-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
        request.requester = requester

        logger.debug("remote_knock: %s on room: %s", user_id, room_id)
@@ -219,7 +219,7 @@ class ReplicationRemoteRejectInviteRestServlet(ReplicationEndpoint):
        txn_id = content["txn_id"]
        event_content = content["content"]

-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
        request.requester = requester

        # hopefully we're now on the master, so this won't recurse!
@@ -283,7 +283,7 @@ class ReplicationRemoteRescindKnockRestServlet(ReplicationEndpoint):
        txn_id = content["txn_id"]
        event_content = content["content"]

-        requester = Requester.deserialize(self.store, content["requester"])
+        requester = Requester.deserialize(content["requester"])
        request.requester = requester

        # hopefully we're now on the master, so this won't recurse!
@@ -141,9 +141,7 @@ class ReplicationSendEventsRestServlet(ReplicationEndpoint):
                )
                event.internal_metadata.outlier = event_payload["outlier"]

-                requester = Requester.deserialize(
-                    self.store, event_payload["requester"]
-                )
+                requester = Requester.deserialize(event_payload["requester"])
                context = EventContext.deserialize(
                    self._storage_controllers, event_payload["context"]
                )
@@ -305,7 +305,7 @@ class DeactivateAccountRestServlet(RestServlet):

        # allow ASes to deactivate their own users:
        # ASes don't need user-interactive auth
-        if not requester.app_service:
+        if not requester.app_service_id:
            await self.auth_handler.validate_user_via_ui_auth(
                requester,
                request,
@@ -55,25 +55,32 @@ class AppservicePingRestServlet(RestServlet):
        self.as_api = hs.get_application_service_api()
        self.scheduler = hs.get_application_service_scheduler()
        self.auth = hs.get_auth()
+        self.store = hs.get_datastores().main

    async def on_POST(
        self, request: SynapseRequest, appservice_id: str
    ) -> tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)

-        if not requester.app_service:
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+
+        if not app_service:
            raise SynapseError(
                HTTPStatus.FORBIDDEN,
                "Only application services can use the /appservice/ping endpoint",
                Codes.FORBIDDEN,
            )
-        elif requester.app_service.id != appservice_id:
+        elif app_service.id != appservice_id:
            raise SynapseError(
                HTTPStatus.FORBIDDEN,
                "Mismatching application service ID in path",
                Codes.FORBIDDEN,
            )
-        elif not requester.app_service.url:
+        elif not app_service.url:
            raise SynapseError(
                HTTPStatus.BAD_REQUEST,
                "The application service does not have a URL set",
@@ -85,11 +92,11 @@ class AppservicePingRestServlet(RestServlet):

        start = time.monotonic()
        try:
-            await self.as_api.ping(requester.app_service, txn_id)
+            await self.as_api.ping(app_service, txn_id)

            # We got a OK response, so if the AS needs to be recovered then lets recover it now.
            # This sets off a task in the background and so is safe to execute and forget.
-            self.scheduler.txn_ctrl.force_retry(requester.app_service)
+            self.scheduler.txn_ctrl.force_retry(app_service)
        except RequestTimedOutError as e:
            raise SynapseError(
                HTTPStatus.GATEWAY_TIMEOUT,
@@ -105,7 +105,7 @@ class DeleteDevicesRestServlet(RestServlet):
            else:
                raise e

-        if requester.app_service:
+        if requester.app_service_id:
            # MSC4190 can skip UIA for this endpoint
            pass
        else:
@@ -177,7 +177,7 @@ class DeviceRestServlet(RestServlet):
            else:
                raise

-        if requester.app_service:
+        if requester.app_service_id:
            # MSC4190 allows appservices to delete devices through this endpoint without UIA
            # It's also allowed with MSC3861 enabled
            pass
@@ -212,7 +212,7 @@ class DeviceRestServlet(RestServlet):
        body = parse_and_validate_json_object_from_request(request, self.PutBody)

        # MSC4190 allows appservices to create devices through this endpoint
-        if requester.app_service:
+        if requester.app_service_id:
            created = await self.device_handler.upsert_device(
                user_id=requester.user.to_string(),
                device_id=device_id,
@@ -110,14 +110,19 @@ class ClientDirectoryServer(RestServlet):
        room_alias_obj = RoomAlias.from_string(room_alias)
        requester = await self.auth.get_user_by_req(request)

-        if requester.app_service:
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        if app_service:
            await self.directory_handler.delete_appservice_association(
-                requester.app_service, room_alias_obj
+                app_service, room_alias_obj
            )

            logger.info(
                "Application service at %s deleted alias %s",
-                requester.app_service.url,
+                app_service.url,
                room_alias_obj.to_string(),
            )

@@ -199,13 +204,13 @@ class ClientAppserviceDirectoryListServer(RestServlet):
        visibility: Literal["public", "private"],
    ) -> tuple[int, JsonDict]:
        requester = await self.auth.get_user_by_req(request)
-        if not requester.app_service:
+        if not requester.app_service_id:
            raise AuthError(
                403, "Only appservices can edit the appservice published room list"
            )

        await self.directory_handler.edit_published_appservice_room_list(
-            requester.app_service.id, network_id, room_id, visibility
+            requester.app_service_id, network_id, room_id, visibility
        )

        return 200, {}
@@ -535,7 +535,7 @@ class SigningKeyUploadServlet(RestServlet):
        # setup, and that is allowed without UIA, per MSC3967.
        # If yes, then we need to authenticate the change.
        # MSC4190 can skip UIA for replacing cross-signing keys as well.
-        if is_cross_signing_setup and not requester.app_service:
+        if is_cross_signing_setup and not requester.app_service_id:
            # With MSC3861, UIA is not possible. Instead, the auth service has to
            # explicitly mark the master key as replaceable.
            if self.hs.config.mas.enabled:
@@ -203,7 +203,11 @@ class LoginRestServlet(RestServlet):
        try:
            if login_submission["type"] == LoginRestServlet.APPSERVICE_TYPE:
                requester = await self.auth.get_user_by_req(request)
-                appservice = requester.app_service
+                appservice = (
+                    self._main_store.get_app_service_by_id(requester.app_service_id)
+                    if requester.app_service_id
+                    else None
+                )

                if appservice is None:
                    raise InvalidClientTokenError(
@@ -336,7 +336,7 @@ class RoomStateEventRestServlet(RestServlet):
                )

        origin_server_ts = None
-        if requester.app_service:
+        if requester.app_service_id:
            origin_server_ts = parse_integer(request, "ts")

        sticky_duration_ms: int | None = None
@@ -435,7 +435,7 @@ class RoomSendEventRestServlet(TransactionRestServlet):
        content = parse_json_object_from_request(request)

        origin_server_ts = None
-        if requester.app_service:
+        if requester.app_service_id:
            origin_server_ts = parse_integer(request, "ts")

        sticky_duration_ms: int | None = None
@@ -82,8 +82,8 @@ class HttpTransactionCache:
            assert requester.user is not None, "Guest requester must have a user ID set"
            return (path, "guest", requester.user)

-        elif requester.app_service is not None:
-            return (path, "appservice", requester.app_service.id)
+        elif requester.app_service_id is not None:
+            return (path, "appservice", requester.app_service_id)

        # Use the user ID and device ID as the transaction key.
        elif requester.device_id:
@@ -43,6 +43,7 @@ class CreateResource(RestServlet):
        super().__init__()

        self.media_repo = media_repo
+        self.store = hs.get_datastores().main
        self.clock = hs.get_clock()
        self.auth = hs.get_auth()
        self.max_pending_media_uploads = hs.config.media.max_pending_media_uploads
@@ -60,7 +61,12 @@ class CreateResource(RestServlet):
        # If the create media requests for the user are over the limit, drop them.
        await self._create_media_rate_limiter.ratelimit(requester)

-        if not requester.app_service or requester.app_service.is_rate_limited():
+        app_service = (
+            self.store.get_app_service_by_id(requester.app_service_id)
+            if requester.app_service_id
+            else None
+        )
+        if not app_service or app_service.is_rate_limited():
            (
                reached_pending_limit,
                first_expiration_ts,
@@ -828,7 +828,8 @@ class MultiWriterIdGenerator(AbstractStreamIdGenerator):
                # do.
                break

-        # Hacky debug logging to attempt to trace https://github.com/element-hq/synapse/issues/19795
+        # Hacky debug logging to attempt to trace https://github.com/element-hq/synapse/issues/19795.
+        # If this is the to-device stream, and we are a writer for that stream, log some stats
        if (
            issue9533_logger.isEnabledFor(logging.DEBUG)
            # Only log if we are the instance that is doing the persisting
@@ -0,0 +1,68 @@
+# This file is licensed under the Affero General Public License (AGPL) version 3.
+#
+# Copyright (C) 2026 Element Creations 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 synapse.types import JsonDict, UserID
+
+class Requester:
+    def __init__(
+        self,
+        user: UserID,
+        access_token_id: int | None,
+        is_guest: bool,
+        scope: set[str],
+        shadow_banned: bool,
+        device_id: str | None,
+        app_service_id: str | None,
+        authenticated_entity: str,
+    ) -> None: ...
+    @property
+    def user_id(self) -> str:
+        """The ID of the user making the request, in string form (see `user`
+        for the parsed UserID)"""
+
+    @property
+    def user(self) -> UserID:
+        """The user making the request"""
+    @property
+    def access_token_id(self) -> int | None:
+        """The ID of the access token used for this request, or
+        None for appservices, guests, and tokens generated by the admin API"""
+    @property
+    def is_guest(self) -> bool:
+        """True if the user making this request is a guest user"""
+    @property
+    def scope(self) -> set[str]:
+        """Any scopes associated with the access token used for this request, or
+        an empty set if no token or a non-oauth token was used"""
+    @property
+    def shadow_banned(self) -> bool:
+        """True if the user making this request has been shadow-banned."""
+    @property
+    def device_id(self) -> str | None:
+        """The device_id which was set at authentication time, or
+        None for appservices, guests, and tokens generated by the admin API"""
+    @property
+    def app_service_id(self) -> str | None:
+        """The ID of the AS requesting on behalf of the user, or None."""
+    @property
+    def authenticated_entity(self) -> str:
+        """The entity that authenticated when making the request.
+
+        This is different to the user_id when an admin user or the server is
+        "puppeting" the user."""
+    def serialize(self) -> JsonDict:
+        """Converts self to a type that can be serialized as JSON, and then
+        deserialized by `deserialize`"""
+    @staticmethod
+    def deserialize(input: JsonDict) -> Requester:
+        """Converts a dict that was produced by `serialize` back into a
+        Requester."""
@@ -61,6 +61,7 @@ from twisted.internet.interfaces import (
 )

 from synapse.api.errors import Codes, SynapseError
+from synapse.synapse_rust.types import Requester
 from synapse.util.cancellation import cancellable
 from synapse.util.stringutils import parse_and_validate_server_name

@@ -70,7 +71,6 @@ if TYPE_CHECKING:
    from synapse.appservice.api import ApplicationService
    from synapse.events import EventBase
    from synapse.storage.databases.main import DataStore, PurgeEventsStore
-    from synapse.storage.databases.main.appservice import ApplicationServiceWorkerStore
    from synapse.storage.util.id_generators import MultiWriterIdGenerator


@@ -138,82 +138,6 @@ class ISynapseReactor(
    """The interfaces necessary for Synapse to function."""


-@attr.s(frozen=True, slots=True, auto_attribs=True)
-class Requester:
-    """
-    Represents the user making a request
-
-    Attributes:
-        user:  id of the user making the request
-        access_token_id:  *ID* of the access token used for this request, or
-            None for appservices, guests, and tokens generated by the admin API
-        is_guest:  True if the user making this request is a guest user
-        shadow_banned:  True if the user making this request has been shadow-banned.
-        device_id:  device_id which was set at authentication time, or
-            None for appservices, guests, and tokens generated by the admin API
-        app_service:  the AS requesting on behalf of the user
-        authenticated_entity: The entity that authenticated when making the request.
-            This is different to the user_id when an admin user or the server is
-            "puppeting" the user.
-    """
-
-    user: "UserID"
-    access_token_id: int | None
-    is_guest: bool
-    scope: set[str]
-    shadow_banned: bool
-    device_id: str | None
-    app_service: Optional["ApplicationService"]
-    authenticated_entity: str
-
-    def serialize(self) -> dict[str, Any]:
-        """Converts self to a type that can be serialized as JSON, and then
-        deserialized by `deserialize`
-
-        Returns:
-            dict
-        """
-        return {
-            "user_id": self.user.to_string(),
-            "access_token_id": self.access_token_id,
-            "is_guest": self.is_guest,
-            "scope": list(self.scope),
-            "shadow_banned": self.shadow_banned,
-            "device_id": self.device_id,
-            "app_server_id": self.app_service.id if self.app_service else None,
-            "authenticated_entity": self.authenticated_entity,
-        }
-
-    @staticmethod
-    def deserialize(
-        store: "ApplicationServiceWorkerStore", input: dict[str, Any]
-    ) -> "Requester":
-        """Converts a dict that was produced by `serialize` back into a
-        Requester.
-
-        Args:
-            store: Used to convert AS ID to AS object
-            input: A dict produced by `serialize`
-
-        Returns:
-            Requester
-        """
-        appservice = None
-        if input["app_server_id"]:
-            appservice = store.get_app_service_by_id(input["app_server_id"])
-
-        return Requester(
-            user=UserID.from_string(input["user_id"]),
-            access_token_id=input["access_token_id"],
-            is_guest=input["is_guest"],
-            scope=set(input.get("scope", [])),
-            shadow_banned=input["shadow_banned"],
-            device_id=input["device_id"],
-            app_service=appservice,
-            authenticated_entity=input["authenticated_entity"],
-        )
-
-
 def create_requester(
    user_id: Union[str, "UserID"],
    access_token_id: int | None = None,
@@ -258,7 +182,7 @@ def create_requester(
        scope,
        shadow_banned,
        device_id,
-        app_service,
+        app_service.id if app_service else None,
        authenticated_entity,
    )

@@ -203,6 +203,9 @@ class SlidingSyncResult:
        highlight_count: int

        def __bool__(self) -> bool:
+            """Are there any updates that should be returned immediately to
+            the client?
+            """
            return (
                # If this is the first time the client is seeing the room, we should not filter it out
                # under any circumstance.
@@ -270,6 +273,8 @@ class SlidingSyncResult:
            events: Sequence[JsonMapping]

            def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                return bool(self.events)

        @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -294,23 +299,37 @@ class SlidingSyncResult:
            device_unused_fallback_key_types: Sequence[str]

            def __bool__(self) -> bool:
-                # Note that "signed_curve25519" is always returned in key count responses
-                # regardless of whether we uploaded any keys for it. This is necessary until
+                """Are there any updates that should be returned immediately to
+                the client?"""
+                # Note that "signed_curve25519" is always returned in key count
+                # responses regardless of whether we uploaded any keys for it.
+                # This is necessary until
                # https://github.com/matrix-org/matrix-doc/issues/3298 is fixed.
                #
                # Also related:
                # https://github.com/element-hq/element-android/issues/3725 and
                # https://github.com/matrix-org/synapse/issues/10456
-                default_otk = self.device_one_time_keys_count.get("signed_curve25519")
-                more_than_default_otk = len(self.device_one_time_keys_count) > 1 or (
-                    default_otk is not None and default_otk > 0
-                )
+                #
+                # This is why we don't incorporate `device_one_time_keys_count`
+                # (or `device_unused_fallback_key_types`) into the `__bool__`
+                # check.
+                #
+                # FIXME: Ideally we'd detect if either of those fields have
+                # changed since the last sync, but we do not currently track
+                # such state.
+                #
+                # Note that the client will receive these fields eventually when
+                # we respond to the sync request (usually sync timeouts are set
+                # to ~30s), we just won't immediately respond (even if there are
+                # changes). This delay is acceptable for clients, as a) these
+                # fields do not trigger UI (and so don't affect user perceivable
+                # latency) and b) are handled in the background by the clients
+                # anyway. The only risk being that one-time keys could be exhausted
+                # before the client knows about adding some more. But for example,
+                # if the client is syncing with a timeout of 30s, the window of
+                # staleness is so small for this not to matter.

-                return bool(
-                    more_than_default_otk
-                    or self.device_list_updates
-                    or self.device_unused_fallback_key_types
-                )
+                return bool(self.device_list_updates)

        @attr.s(slots=True, frozen=True, auto_attribs=True)
        class AccountDataExtension:
@@ -327,6 +346,8 @@ class SlidingSyncResult:
            account_data_by_room_map: Mapping[str, Mapping[str, JsonMapping]]

            def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                return bool(
                    self.global_account_data_map or self.account_data_by_room_map
                )
@@ -343,6 +364,8 @@ class SlidingSyncResult:
            room_id_to_receipt_map: Mapping[str, JsonMapping]

            def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                return bool(self.room_id_to_receipt_map)

        @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -357,6 +380,8 @@ class SlidingSyncResult:
            room_id_to_typing_map: Mapping[str, JsonMapping]

            def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                return bool(self.room_id_to_typing_map)

        @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -391,6 +416,8 @@ class SlidingSyncResult:
            prev_batch: ThreadSubscriptionsToken | None

            def __bool__(self) -> bool:
+                """Are there any updates that should be returned immediately to
+                the client?"""
                return (
                    bool(self.subscribed)
                    or bool(self.unsubscribed)
@@ -405,6 +432,8 @@ class SlidingSyncResult:
        thread_subscriptions: ThreadSubscriptionsExtension | None = None

        def __bool__(self) -> bool:
+            """Are there any updates that should be returned immediately to
+            the client?"""
            return bool(
                self.to_device
                or self.e2ee
@@ -420,9 +449,14 @@ class SlidingSyncResult:
    extensions: Extensions

    def __bool__(self) -> bool:
-        """Make the result appear empty if there are no updates. This is used
-        to tell if the notifier needs to wait for more events when polling for
-        events.
+        """Are there any updates that should be returned immediately to
+        the client?
+
+        This is used to determine if a sliding sync response should be returned
+        immediately or if the notifier needs to wait for further updates, and
+        thus MUST return false if there is no new data since the last sync. This
+        is subtly different than just checking if any of the fields are set,
+        since some fields are always included (like `bump_stamp`).
        """
        # We don't include `self.lists` here, as a) `lists` is always non-empty even if
        # there are no changes, and b) since we're sorting rooms by `stream_ordering` of
@@ -108,6 +108,7 @@ class AuthTestCase(unittest.HomeserverTestCase):

    def test_get_user_by_req_appservice_valid_token(self) -> None:
        app_service = Mock(
+            id="as_id",
            token="foobar",
            url="a_url",
            sender=self.test_user_id,
@@ -132,6 +133,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            sender=self.test_user_id.to_string(),
            ip_range_whitelist=IPSet(["192.168.0.0/16"]),
        )
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)
        self.store.get_user_by_access_token = AsyncMock(return_value=None)

@@ -151,6 +153,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            sender=self.test_user_id,
            ip_range_whitelist=IPSet(["192.168.0.0/16"]),
        )
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)
        self.store.get_user_by_access_token = AsyncMock(return_value=None)

@@ -179,6 +182,7 @@ class AuthTestCase(unittest.HomeserverTestCase):

    def test_get_user_by_req_appservice_missing_token(self) -> None:
        app_service = Mock(token="foobar", url="a_url", sender=self.test_user_id)
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)
        self.store.get_user_by_access_token = AsyncMock(return_value=None)

@@ -199,6 +203,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            ip_range_whitelist=None,
        )
        app_service.is_interested_in_user = Mock(return_value=True)
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)

        class FakeUserInfo:
@@ -226,6 +231,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            ip_range_whitelist=None,
        )
        app_service.is_interested_in_user = Mock(return_value=False)
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)
        self.store.get_user_by_access_token = AsyncMock(return_value=None)

@@ -251,6 +257,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            ip_range_whitelist=None,
        )
        app_service.is_interested_in_user = Mock(return_value=True)
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)
        # This just needs to return a truth-y value.
        self.store.get_user_by_id = AsyncMock(return_value={"is_guest": False})
@@ -285,6 +292,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            ip_range_whitelist=None,
        )
        app_service.is_interested_in_user = Mock(return_value=True)
+        app_service.id = "as_id"
        self.store.get_app_service_by_token = Mock(return_value=app_service)
        # This just needs to return a truth-y value.
        self.store.get_user_by_id = AsyncMock(return_value={"is_guest": False})
@@ -457,7 +465,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            is_guest=False,
            scope=set(),
            shadow_banned=False,
-            app_service=appservice,
+            app_service_id=appservice.id,
            authenticated_entity="@appservice:server",
        )
        self.get_success(self.auth_blocking.check_auth_blocking(requester=requester))
@@ -488,7 +496,7 @@ class AuthTestCase(unittest.HomeserverTestCase):
            is_guest=False,
            scope=set(),
            shadow_banned=False,
-            app_service=appservice,
+            app_service_id=appservice.id,
            authenticated_entity="@appservice:server",
        )
        self.get_failure(
@@ -40,6 +40,9 @@ class TestRatelimiter(unittest.HomeserverTestCase):
            rate_limited=True,
            sender=UserID.from_string("@as:example.com"),
        )
+        # The ratelimiter now resolves the AS via get_app_service_by_id, so the
+        # appservice must be in the store's cache for the lookup to hit.
+        self.hs.get_datastores().main.services_cache.append(appservice)
        as_requester = create_requester("@user:example.com", app_service=appservice)

        limiter = Ratelimiter(
@@ -76,6 +79,9 @@ class TestRatelimiter(unittest.HomeserverTestCase):
            rate_limited=False,
            sender=UserID.from_string("@as:example.com"),
        )
+        # The ratelimiter now resolves the AS via get_app_service_by_id, so the
+        # appservice must be in the store's cache for the lookup to hit.
+        self.hs.get_datastores().main.services_cache.append(appservice)
        as_requester = create_requester("@user:example.com", app_service=appservice)

        limiter = Ratelimiter(
@@ -290,6 +290,173 @@ class SlidingSyncE2eeExtensionTestCase(SlidingSyncBase):
            [],
        )

+    def test_wait_for_new_data_timeout_with_otks(self) -> None:
+        """
+        Test that an incremental Sliding Sync with the e2ee extension enabled
+        does not return immediately when the user has uploaded one-time keys
+        (i.e. `device_one_time_keys_count` contains entries beyond the default
+        `signed_curve25519: 0`).
+        """
+        test_device_id = "TESTDEVICE"
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass", device_id=test_device_id)
+
+        # Upload one-time keys for the user/device so that
+        # `device_one_time_keys_count` is non-default in the response.
+        self.get_success(
+            self.e2e_keys_handler.upload_keys_for_user(
+                user1_id,
+                test_device_id,
+                {
+                    "one_time_keys": {
+                        "alg1:k1": "key1",
+                        "alg2:k2": {"key": "key2", "signatures": {"k1": "sig1"}},
+                    }
+                },
+            )
+        )
+
+        sync_body = {
+            "lists": {},
+            "extensions": {
+                "e2ee": {
+                    "enabled": True,
+                }
+            },
+        }
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # Make an incremental Sliding Sync request with a timeout
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?timeout=10000&pos={from_token}",
+            content=sync_body,
+            access_token=user1_tok,
+            await_result=False,
+        )
+        # Block for 5 seconds to make sure we are `notifier.wait_for_events(...)`
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=5000)
+
+        # Wake-up `notifier.wait_for_events(...)` that will cause us to test
+        # `SlidingSyncResult.__bool__` for new results. The non-default
+        # `device_one_time_keys_count` must not be considered new data.
+        self._bump_notifier_wait_for_events(
+            user1_id, wake_stream_key=StreamKeyType.ACCOUNT_DATA
+        )
+
+        # Block for a little bit more to ensure we don't see any new results.
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=4000)
+        # Wait for the sync to complete (wait for the rest of the 10 second timeout,
+        # 5000 + 4000 + 1200 > 10000)
+        channel.await_result(timeout_ms=1200)
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Device lists are present for incremental syncs but empty because no device changes
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]
+            .get("device_lists", {})
+            .get("changed"),
+            [],
+        )
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"].get("device_lists", {}).get("left"),
+            [],
+        )
+
+        # The one-time key counts are present, but they should not have caused
+        # the sync to return early.
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]["device_one_time_keys_count"],
+            {
+                "alg1": 1,
+                "alg2": 1,
+                "signed_curve25519": 0,
+            },
+        )
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]["device_unused_fallback_key_types"],
+            [],
+        )
+
+    def test_wait_for_new_data_timeout_with_fallback_keys(self) -> None:
+        """
+        Test that an incremental Sliding Sync with the e2ee extension enabled
+        does not return immediately when the user has uploaded fallback keys
+        (i.e. `device_unused_fallback_key_types` is non-empty).
+        """
+        test_device_id = "TESTDEVICE"
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass", device_id=test_device_id)
+
+        # Upload a fallback key for the user/device so that
+        # `device_unused_fallback_key_types` is non-empty in the response.
+        self.get_success(
+            self.e2e_keys_handler.upload_keys_for_user(
+                user1_id,
+                test_device_id,
+                {"fallback_keys": {"alg1:k1": "fallback_key1"}},
+            )
+        )
+
+        sync_body = {
+            "lists": {},
+            "extensions": {
+                "e2ee": {
+                    "enabled": True,
+                }
+            },
+        }
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # Make an incremental Sliding Sync request with a timeout
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?timeout=10000&pos={from_token}",
+            content=sync_body,
+            access_token=user1_tok,
+            await_result=False,
+        )
+        # Block for 5 seconds to make sure we are `notifier.wait_for_events(...)`
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=5000)
+
+        # Wake-up `notifier.wait_for_events(...)` that will cause us to test
+        # `SlidingSyncResult.__bool__` for new results. The non-empty
+        # `device_unused_fallback_key_types` must not be considered new data.
+        self._bump_notifier_wait_for_events(
+            user1_id, wake_stream_key=StreamKeyType.ACCOUNT_DATA
+        )
+
+        # Block for a little bit more to ensure we don't see any new results.
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=4000)
+
+        # Wait for the sync to complete (wait for the rest of the 10 second timeout,
+        # 5000 + 4000 + 1200 > 10000)
+        channel.await_result(timeout_ms=1200)
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        # Device lists are present for incremental syncs but empty because no device changes
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]
+            .get("device_lists", {})
+            .get("changed"),
+            [],
+        )
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"].get("device_lists", {}).get("left"),
+            [],
+        )
+
+        # The unused fallback key types are present, but they should not have
+        # caused the sync to return early.
+        self.assertEqual(
+            channel.json_body["extensions"]["e2ee"]["device_unused_fallback_key_types"],
+            ["alg1"],
+        )
+
    def test_device_lists(self) -> None:
        """
        Test that device list updates are included in the response
@@ -22,6 +22,7 @@ import synapse.rest.admin
 from synapse.api.constants import EventTypes, HistoryVisibility
 from synapse.rest.client import login, room, sync
 from synapse.server import HomeServer
+from synapse.types import JsonDict
 from synapse.util.clock import Clock

 from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
@@ -126,6 +127,124 @@ class SlidingSyncRoomSubscriptionsTestCase(SlidingSyncBase):
            response_body["rooms"][room_id1],
        )

+    def test_room_subscription_required_state_expansion_returns_immediately(
+        self,
+    ) -> None:
+        """
+        Test that adding a room subscription with stronger params than the list causes an
+        incremental long-poll to return immediately, even without new stream activity.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        sync_body: JsonDict = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 0]],
+                    "required_state": [],
+                    "timeline_limit": 0,
+                }
+            },
+            "conn_id": "conn_id",
+        }
+        _, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        sync_body["room_subscriptions"] = {
+            room_id1: {
+                "required_state": [
+                    [EventTypes.Create, ""],
+                ],
+                "timeline_limit": 0,
+            }
+        }
+
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?timeout=10000&pos={from_token}",
+            content=sync_body,
+            access_token=user1_tok,
+            await_result=False,
+        )
+        channel.await_result(timeout_ms=3000)
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+
+        room_response = channel.json_body["rooms"][room_id1]
+        self.assertNotIn("initial", room_response)
+        self._assertRequiredStateIncludes(
+            room_response["required_state"],
+            {
+                state_map[(EventTypes.Create, "")],
+            },
+            exact=True,
+        )
+
+    def test_room_subscription_required_state_change_returns_immediately(self) -> None:
+        """
+        Test that expanding an existing room subscription's required state causes an
+        incremental long-poll to return immediately, even without new stream activity.
+        """
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        room_id1 = self.helper.create_room_as(
+            user1_id, tok=user1_tok, extra_content={"name": "Foo"}
+        )
+
+        sync_body: JsonDict = {
+            "room_subscriptions": {
+                room_id1: {
+                    "required_state": [
+                        [EventTypes.Create, ""],
+                    ],
+                    "timeline_limit": 0,
+                }
+            },
+            "conn_id": "conn_id",
+        }
+        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+        self._assertRequiredStateIncludes(
+            response_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Create, "")],
+            },
+            exact=True,
+        )
+
+        sync_body["room_subscriptions"][room_id1]["required_state"] = [
+            [EventTypes.Create, ""],
+            [EventTypes.Name, ""],
+        ]
+
+        channel = self.make_request(
+            "POST",
+            self.sync_endpoint + f"?timeout=10000&pos={from_token}",
+            content=sync_body,
+            access_token=user1_tok,
+            await_result=False,
+        )
+        channel.await_result(timeout_ms=3000)
+        self.assertEqual(channel.code, 200, channel.json_body)
+
+        room_response = channel.json_body["rooms"][room_id1]
+        self.assertNotIn("initial", room_response)
+        self._assertRequiredStateIncludes(
+            room_response["required_state"],
+            {
+                state_map[(EventTypes.Name, "")],
+            },
+            exact=True,
+        )
+
    def test_room_subscriptions_with_leave_membership(self) -> None:
        """
        Test `room_subscriptions` with a leave room should give us timeline and state
@@ -25,8 +25,10 @@ from synapse.rest.client import knock, login, room, sync
 from synapse.server import HomeServer
 from synapse.storage.databases.main.events import DeltaState, SlidingSyncTableChanges
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration

 from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
+from tests.server import TimedOutException
 from tests.test_utils.event_injection import mark_event_as_partial_state

 logger = logging.getLogger(__name__)
@@ -1924,7 +1926,12 @@ class SlidingSyncRoomsRequiredStateTestCase(SlidingSyncBase):

    def test_rooms_required_state_expand_retract_expand(self) -> None:
        """Test that when expanding, retracting and then expanding the required
-        state, we get the changes that happened."""
+        state, we get the changes that happened.
+
+        Also see `test_changing_required_state_returns_immediately`, which tests
+        that the sync stream is woken up immediately when changing the required
+        state, and not just on the next change to the room.
+        """

        user1_id = self.register_user("user1", "pass")
        user1_tok = self.login(user1_id, "pass")
@@ -2245,3 +2252,75 @@ class SlidingSyncRoomsRequiredStateTestCase(SlidingSyncBase):
            response_body["rooms"][room_id]["required_state"][0]["event_id"],
            first_event_id,
        )
+
+    def test_changing_required_state_returns_immediately(self) -> None:
+        """Test that if we change the `required_state`, then we return immediately
+        with the new `required_state`."""
+
+        user1_id = self.register_user("user1", "pass")
+        user1_tok = self.login(user1_id, "pass")
+
+        room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok)
+
+        # Make an initial sync request with no required state
+        sync_body = {
+            "lists": {
+                "foo-list": {
+                    "ranges": [[0, 1]],
+                    "required_state": [],
+                    "timeline_limit": 0,
+                }
+            }
+        }
+        response_body, from_token = self.do_sync(sync_body, tok=user1_tok)
+
+        # We should see no required state
+        self.assertIsNone(response_body["rooms"][room_id1].get("required_state"))
+
+        # Get the state_map before we change the state as this is the final state we
+        # expect to see when we update the required state.
+        state_map = self.get_success(
+            self.storage_controllers.state.get_current_state(room_id1)
+        )
+
+        # There is no new data, and so making another sync request will block.
+        channel = self.make_sync_request(
+            sync_body,
+            since=from_token,
+            tok=user1_tok,
+            timeout=Duration(seconds=10),
+            await_result=False,
+        )
+
+        # Request will block for 10 seconds as there no updates.
+        with self.assertRaises(TimedOutException):
+            channel.await_result(timeout_ms=9500)
+
+        # Wait for the request to actually finish. (We do this to ensure log
+        # contexts don't leak between tests).
+        channel.await_result(timeout_ms=1000)
+
+        # Now update the Sliding Sync requests to include a `required_state`
+        # event, and make another sync request.
+        sync_body["lists"]["foo-list"]["required_state"] = [
+            [EventTypes.Create, ""],
+        ]
+
+        channel = self.make_sync_request(
+            sync_body,
+            since=from_token,
+            tok=user1_tok,
+            timeout=Duration(seconds=10),
+            await_result=False,
+        )
+
+        # We should see the new `required_state` immediately without waiting
+        channel.await_result(timeout_ms=0)
+        response_body = channel.json_body
+        self._assertRequiredStateIncludes(
+            response_body["rooms"][room_id1]["required_state"],
+            {
+                state_map[(EventTypes.Create, "")],
+            },
+            exact=True,
+        )
@@ -12,6 +12,7 @@
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 #
 import logging
+import urllib.parse
 from typing import Any, Iterable, Literal
 from unittest.mock import AsyncMock

@@ -43,6 +44,7 @@ from synapse.types import (
    StreamToken,
 )
 from synapse.util.clock import Clock
+from synapse.util.duration import Duration
 from synapse.util.stringutils import random_string

 from tests import unittest
@@ -82,7 +84,13 @@ class SlidingSyncBase(unittest.HomeserverTestCase):
        return config

    def make_sync_request(
-        self, sync_body: JsonDict, *, since: str | None = None, tok: str
+        self,
+        sync_body: JsonDict,
+        *,
+        since: str | None = None,
+        tok: str,
+        timeout: Duration | None = None,
+        await_result: bool = True,
    ) -> FakeChannel:
        """Make a sliding sync request with given body.

@@ -90,25 +98,40 @@ class SlidingSyncBase(unittest.HomeserverTestCase):
            sync_body: The full request body to use
            since: Optional since token
            tok: Access token to use
-
+            timeout_ms: Optional timeout in milliseconds to use for the request.
+            await_result: Whether to block and wait for the result before returning.
        Returns:
            A tuple of the response body and the `pos` field.
        """

        sync_path = self.sync_endpoint
+
+        query_params: dict[str, Any] = {}
        if since:
-            sync_path += f"?pos={since}"
+            query_params["pos"] = since
+        if timeout is not None:
+            query_params["timeout"] = timeout.as_millis()
+
+        if query_params:
+            query_str = urllib.parse.urlencode(query_params)
+            sync_path += f"?{query_str}"

        channel = self.make_request(
            method="POST",
            path=sync_path,
            content=sync_body,
            access_token=tok,
+            await_result=await_result,
        )
        return channel

    def do_sync(
-        self, sync_body: JsonDict, *, since: str | None = None, tok: str
+        self,
+        sync_body: JsonDict,
+        *,
+        since: str | None = None,
+        tok: str,
+        timeout: Duration | None = None,
    ) -> tuple[JsonDict, str]:
        """Do a sliding sync request with given body.

@@ -118,11 +141,14 @@ class SlidingSyncBase(unittest.HomeserverTestCase):
            sync_body: The full request body to use
            since: Optional since token
            tok: Access token to use
+            timeout: Optional timeout to use for the request.

        Returns:
            A tuple of the response body and the `pos` field.
        """
-        channel = self.make_sync_request(sync_body, since=since, tok=tok)
+        channel = self.make_sync_request(
+            sync_body, since=since, tok=tok, timeout=timeout
+        )
        self.assertEqual(channel.code, 200, channel.json_body)

        return channel.json_body, channel.json_body["pos"]
@@ -52,7 +52,7 @@ class HttpTransactionCacheTestCase(unittest.TestCase):
        self.mock_request = Mock()
        self.mock_request.path = b"/foo/bar"
        self.mock_requester = Mock()
-        self.mock_requester.app_service = None
+        self.mock_requester.app_service_id = None
        self.mock_requester.is_guest = False
        self.mock_requester.access_token_id = 1234
				`@@ -1 +0,0 @@`
				`A long queue of to-device messages could prevent outgoing federation because of the size of the transaction, let's limit the to-device EDU size to a reasonable value.`
				`@@ -1 +0,0 @@`
				`Work around bug that sometimes breaks joining restricted rooms that require a remote join. Contributed by @tulir @ Beeper.`
				`@@ -1 +0,0 @@`
				`Added tests to ensure that email notification links are sanitized. Contributed by Noah Markert.`
				`@@ -1 +0,0 @@`
				Add `GcpJsonFormatter` logging formatter for use with Google Cloud Logging and GKE deployments.
				`@@ -1 +0,0 @@`
				`Add more logging to the to-device message replication stream.`
				`@@ -1 +0,0 @@`
				Document how to see Rust build failure output when using `poetry install`.