From dcb7c85dbf03b727a462a6956bf4863e0796d3d6 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Oct 2026 10:28:39 +0000 Subject: [PATCH] Parse the sliding sync request body in Rust Python validates the body with pydantic as before. Rust parses the same bytes with serde into `SlidingSyncBody`, which follows the pydantic model field for field, including where it tells an explicit `null` from a missing field. Lists and room subscriptions keep document order, and a repeated key keeps its first position and last value, as the dict `json.loads` builds. `SlidingSyncRequest` is the body plus what Python adds: the user and device, the room key and connection position of `pos`, the room key and string suffix of the token the response runs up to, the admin's soft-failed policy, whether the background updates have finished, the excluded rooms and the extension flags. `SlidingSyncHandler._build_rust_request` builds it. The servlet keeps the body's bytes in `SlidingSyncConfig.body_bytes` when the Rust handler is in use. A config built without a request is re-encoded. `tests/rest/client/sliding_sync/test_rust_request.py` parses 400 generated bodies (unknown fields, nulls, repeated keys, escapes) both ways and compares, and checks the re-encoding parses the same. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb --- rust/src/sliding_sync/mod.rs | 3 + rust/src/sliding_sync/request.rs | 548 +++++++++++++++++ synapse/handlers/sliding_sync/__init__.py | 54 ++ synapse/rest/client/sync.py | 6 + synapse/synapse_rust/sliding_sync.pyi | 48 ++ synapse/types/handlers/sliding_sync.py | 3 + .../client/sliding_sync/test_rust_request.py | 563 ++++++++++++++++++ 7 files changed, 1225 insertions(+) create mode 100644 rust/src/sliding_sync/request.rs create mode 100644 tests/rest/client/sliding_sync/test_rust_request.py diff --git a/rust/src/sliding_sync/mod.rs b/rust/src/sliding_sync/mod.rs index 024890bf1e..4f46167572 100644 --- a/rust/src/sliding_sync/mod.rs +++ b/rust/src/sliding_sync/mod.rs @@ -36,6 +36,8 @@ use crate::deferred::{create_deferred_with, OnCancel}; use crate::runtime::RustRuntime; use crate::storage::sync_store::{SyncDbStats, SyncStore}; +pub mod request; + /// Computes sliding sync responses with the Rust implementation, reading /// the database through the native Postgres driver. #[pyclass(frozen)] @@ -158,6 +160,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> let child_module = PyModule::new(py, "sliding_sync")?; child_module.add_class::()?; child_module.add_class::()?; + child_module.add_class::()?; m.add_submodule(&child_module)?; diff --git a/rust/src/sliding_sync/request.rs b/rust/src/sliding_sync/request.rs new file mode 100644 index 0000000000..c5afe58e89 --- /dev/null +++ b/rust/src/sliding_sync/request.rs @@ -0,0 +1,548 @@ +/* + * 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: + * . + */ + +//! One sliding sync request as Rust sees it: the request body, plus what +//! Python adds to it. +//! +//! Python validates the body with pydantic (`SlidingSyncBody`), which also +//! produces the error messages. Rust then parses the same bytes with serde +//! into [`SlidingSyncBody`]. The two parses are held equal by +//! `tests/rest/client/sliding_sync/test_rust_request.py`. The types follow +//! the pydantic model field for field, including where it tells an explicit +//! `null` from a missing field (`slow_get_all_rooms`, and an extension's +//! `enabled`, `lists` and `rooms`). +//! +//! Rust only parses bodies pydantic has accepted, so it does not repeat the +//! validation: the `timeline_limit` maximum, the range bounds, the list key +//! length, the number of lists and the format of the `since` tokens. One +//! accepted body does not parse here: an integer outside `i64` (`u64` for a +//! range), which Python takes as an arbitrary-precision integer. Python then +//! computes the request itself. + +use std::collections::{HashMap, HashSet}; +use std::fmt; +use std::marker::PhantomData; +use std::sync::Arc; + +use pyo3::{exceptions::PyValueError, prelude::*}; +use pythonize::pythonize; +use serde::{ + de::{MapAccess, Visitor}, + ser::SerializeSeq, + Deserialize, Deserializer, Serialize, Serializer, +}; + +use crate::types::stream_token::RoomStreamToken; +use crate::visibility::SoftFailedPolicy; + +/// A JSON object read into its entries in document order. A repeated key +/// keeps its first position and its last value, as the Python dict that +/// `json.loads` builds does. +/// +/// Serialises as a list of `[key, value]` pairs, so the order survives the +/// trip to Python in tests. +#[derive(Debug, Clone, PartialEq)] +pub struct OrderedMap(pub Vec<(String, V)>); + +impl Default for OrderedMap { + fn default() -> Self { + OrderedMap(Vec::new()) + } +} + +impl OrderedMap { + pub fn iter(&self) -> impl Iterator { + self.0.iter().map(|(k, v)| (k.as_str(), v)) + } + + pub fn len(&self) -> usize { + self.0.len() + } + + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } + + pub fn get(&self, key: &str) -> Option<&V> { + self.0.iter().find(|(k, _)| k == key).map(|(_, v)| v) + } +} + +impl<'de, V: Deserialize<'de>> Deserialize<'de> for OrderedMap { + fn deserialize>(deserializer: D) -> Result { + struct OrderedMapVisitor(PhantomData); + + impl<'de, V: Deserialize<'de>> Visitor<'de> for OrderedMapVisitor { + type Value = OrderedMap; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + formatter.write_str("a JSON object") + } + + fn visit_map>(self, mut access: A) -> Result { + let mut entries: Vec<(String, V)> = Vec::new(); + let mut positions: HashMap = HashMap::new(); + while let Some((key, value)) = access.next_entry::()? { + match positions.get(&key) { + Some(&position) => entries[position].1 = value, + None => { + positions.insert(key.clone(), entries.len()); + entries.push((key, value)); + } + } + } + Ok(OrderedMap(entries)) + } + } + + deserializer.deserialize_map(OrderedMapVisitor(PhantomData)) + } +} + +impl Serialize for OrderedMap { + fn serialize(&self, serializer: S) -> Result { + let mut seq = serializer.serialize_seq(Some(self.0.len()))?; + for entry in &self.0 { + seq.serialize_element(entry)?; + } + seq.end() + } +} + +fn some_false() -> Option { + Some(false) +} + +fn all_lists_or_rooms() -> Option> { + Some(vec!["*".to_owned()]) +} + +fn hundred() -> i64 { + 100 +} + +/// `SlidingSyncBody`: the request body. +#[derive(Debug, Clone, Default, PartialEq, Deserialize, Serialize)] +pub struct SlidingSyncBody { + #[serde(default)] + pub conn_id: Option, + /// The sliding window lists, by the client's name for them. + #[serde(default)] + pub lists: Option>, + /// The room subscriptions, by room ID. + #[serde(default)] + pub room_subscriptions: Option>, + #[serde(default)] + pub extensions: Option, +} + +/// `SlidingSyncBody.SlidingSyncList`. +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct SlidingSyncList { + /// `(event type, state key)` pairs, with the `*`, `$LAZY` and `$ME` + /// wildcards still in them. + pub required_state: Vec<(String, String)>, + /// At most 1000. Pydantic sets no minimum. + pub timeline_limit: i64, + /// Inclusive `[start, end]` ranges. `None` means the whole list. + #[serde(default)] + pub ranges: Option>, + /// `false` when missing, `None` for an explicit `null`. + #[serde(default = "some_false")] + pub slow_get_all_rooms: Option, + #[serde(default)] + pub filters: Option, +} + +/// `SlidingSyncBody.RoomSubscription`. +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct RoomSubscription { + pub required_state: Vec<(String, String)>, + pub timeline_limit: i64, +} + +/// `SlidingSyncList.Filters`. Every field is optional, and a missing field +/// means no filter on it. +#[derive(Debug, Clone, Default, PartialEq, Deserialize, Serialize)] +pub struct Filters { + #[serde(default)] + pub is_dm: Option, + #[serde(default)] + pub spaces: Option>, + #[serde(default)] + pub is_encrypted: Option, + #[serde(default)] + pub is_invite: Option, + /// `None` inside the list is a room without a type. + #[serde(default)] + pub room_types: Option>>, + #[serde(default)] + pub not_room_types: Option>>, + #[serde(default)] + pub room_name_like: Option, + #[serde(default)] + pub tags: Option>, + #[serde(default)] + pub not_tags: Option>, +} + +/// `SlidingSyncBody.Extensions`. An unknown extension is ignored. +#[derive(Debug, Clone, Default, PartialEq, Deserialize, Serialize)] +pub struct Extensions { + #[serde(default)] + pub to_device: Option, + #[serde(default)] + pub e2ee: Option, + #[serde(default)] + pub account_data: Option, + #[serde(default)] + pub receipts: Option, + #[serde(default)] + pub typing: Option, + /// MSC4308. Only the unstable name is accepted. + #[serde(default, rename = "io.element.msc4308.thread_subscriptions")] + pub thread_subscriptions: Option, + /// MSC4354. `None` when absent: pydantic rejects an explicit `null`. + #[serde(default, rename = "org.matrix.msc4354.sticky_events")] + pub sticky_events: Option, + /// MSC4262. As `sticky_events`. + #[serde(default, rename = "org.matrix.msc4262.profiles")] + pub profiles: Option, +} + +/// `Extensions.ToDeviceExtension` (MSC3885). +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct ToDeviceExtension { + #[serde(default = "some_false")] + pub enabled: Option, + #[serde(default = "hundred")] + pub limit: i64, + /// The device inbox stream position, as a string of digits. + #[serde(default)] + pub since: Option, +} + +/// `Extensions.E2eeExtension` (MSC3884). +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct E2eeExtension { + #[serde(default = "some_false")] + pub enabled: Option, +} + +/// The account data (MSC3959), receipts (MSC3960) and typing (MSC3961) +/// extensions, which share their fields. +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct ListsAndRoomsExtension { + #[serde(default = "some_false")] + pub enabled: Option, + /// The list names to cover. `["*"]`, every list, when missing. + #[serde(default = "all_lists_or_rooms")] + pub lists: Option>, + /// The room subscriptions to cover. `["*"]`, every one, when missing. + #[serde(default = "all_lists_or_rooms")] + pub rooms: Option>, +} + +/// `Extensions.ThreadSubscriptionsExtension` (MSC4308). +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct ThreadSubscriptionsExtension { + #[serde(default = "some_false")] + pub enabled: Option, + #[serde(default = "hundred")] + pub limit: i64, +} + +/// `Extensions.StickyEventsExtension` (MSC4354). +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct StickyEventsExtension { + #[serde(default)] + pub enabled: bool, + #[serde(default = "hundred")] + pub limit: i64, + /// A `sticky_` token, unparsed. + #[serde(default)] + pub since: Option, +} + +/// `Extensions.ProfilesExtension` (MSC4262). +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)] +pub struct ProfilesExtension { + #[serde(default)] + pub enabled: bool, + #[serde(default)] + pub fields: Option>, +} + +impl SlidingSyncBody { + /// Parse a body pydantic has accepted. + pub fn parse(body: &[u8]) -> Result { + serde_json::from_slice(body) + } +} + +/// Where an incremental request starts: the `pos` the client sent, less +/// the stream keys Rust does not read yet. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FromToken { + pub room_key: RoomStreamToken, + pub connection_position: i64, +} + +/// What every part of a request's computation reads, fixed for the whole +/// request. +#[derive(Debug)] +pub struct SlidingSyncRequestData { + pub body: SlidingSyncBody, + pub user_id: Arc, + pub device_id: Option>, + /// `None` for an initial request. + pub from: Option, + /// The room key of the token the response is computed up to. + pub to_room_key: RoomStreamToken, + /// The rest of that token's string form, after the room key and its + /// `_`. A room's `prev_batch` is a new room key followed by it. + pub to_suffix: Arc, + /// `Some` for a server admin: what they asked to see of soft-failed + /// events. + pub admin_policy: Option, + /// `have_finished_sliding_sync_background_jobs`. Rust covers only the + /// path that reads the sliding sync tables, so a request made before + /// they are filled is Python's. + pub have_finished_bg_jobs: bool, + /// `rooms_to_exclude_from_sync`. + pub rooms_to_exclude: Arc>>, + pub msc4306_enabled: bool, + pub msc4354_enabled: bool, + /// `include_profile_updates_in_sync`, which gates MSC4262. + pub profiles_enabled: bool, +} + +/// One sliding sync request. Built by Python in +/// `SlidingSyncHandler._build_rust_request` for each computation of the +/// response, as the token it runs up to changes between the first try and +/// the one after waiting. +#[pyclass(frozen, skip_from_py_object)] +pub struct SlidingSyncRequest { + pub(crate) data: Arc, +} + +#[pymethods] +impl SlidingSyncRequest { + /// Raises `ValueError` if the body does not parse. + #[new] + #[pyo3(signature = (body, *, user_id, device_id, from_room_key, connection_position, to_room_key, to_suffix, is_server_admin, return_soft_failed_events, return_policy_server_spammy_events, have_finished_bg_jobs, rooms_to_exclude, msc4306_enabled, msc4354_enabled, profiles_enabled))] + #[allow(clippy::too_many_arguments)] + fn py_new( + body: &[u8], + user_id: &str, + device_id: Option<&str>, + from_room_key: Option, + connection_position: Option, + to_room_key: RoomStreamToken, + to_suffix: &str, + is_server_admin: bool, + return_soft_failed_events: bool, + return_policy_server_spammy_events: bool, + have_finished_bg_jobs: bool, + rooms_to_exclude: HashSet, + msc4306_enabled: bool, + msc4354_enabled: bool, + profiles_enabled: bool, + ) -> PyResult { + let body = SlidingSyncBody::parse(body) + .map_err(|e| PyValueError::new_err(format!("parsing the sliding sync body: {e}")))?; + let from = match (from_room_key, connection_position) { + (Some(room_key), Some(connection_position)) => Some(FromToken { + room_key, + connection_position, + }), + (None, None) => None, + _ => { + return Err(PyValueError::new_err( + "from_room_key and connection_position go together", + )) + } + }; + let admin_policy = is_server_admin.then_some(SoftFailedPolicy { + return_soft_failed_events, + return_policy_server_spammy_events, + }); + Ok(SlidingSyncRequest { + data: Arc::new(SlidingSyncRequestData { + body, + user_id: Arc::from(user_id), + device_id: device_id.map(Arc::from), + from, + to_room_key, + to_suffix: Arc::from(to_suffix), + admin_policy, + have_finished_bg_jobs, + rooms_to_exclude: Arc::new(rooms_to_exclude.into_iter().map(Arc::from).collect()), + msc4306_enabled, + msc4354_enabled, + profiles_enabled, + }), + }) + } + + /// The parsed body, with every default filled in and each JSON object + /// keyed by name (lists, room subscriptions) as a list of `[name, + /// value]` pairs. For the differential test. + fn body_for_test<'py>(&self, py: Python<'py>) -> PyResult> { + Ok(pythonize(py, &self.data.body)?) + } + + #[getter] + fn user_id(&self) -> &str { + &self.data.user_id + } + + #[getter] + fn device_id(&self) -> Option<&str> { + self.data.device_id.as_deref() + } + + #[getter] + #[allow(clippy::wrong_self_convention)] + fn from_room_key(&self) -> Option { + self.data.from.as_ref().map(|from| from.room_key.clone()) + } + + #[getter] + fn connection_position(&self) -> Option { + self.data.from.as_ref().map(|from| from.connection_position) + } + + #[getter] + fn to_room_key(&self) -> RoomStreamToken { + self.data.to_room_key.clone() + } + + #[getter] + fn to_suffix(&self) -> &str { + &self.data.to_suffix + } + + #[getter] + fn is_server_admin(&self) -> bool { + self.data.admin_policy.is_some() + } + + #[getter] + fn have_finished_bg_jobs(&self) -> bool { + self.data.have_finished_bg_jobs + } + + #[getter] + fn rooms_to_exclude(&self) -> HashSet { + self.data + .rooms_to_exclude + .iter() + .map(|room_id| room_id.to_string()) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_body() { + let body = SlidingSyncBody::parse(b"{}").unwrap(); + assert_eq!(body, SlidingSyncBody::default()); + } + + #[test] + fn list_defaults_and_nulls() { + let body = SlidingSyncBody::parse( + br#"{ + "lists": { + "a": {"required_state": [["m.room.name", ""]], "timeline_limit": 5}, + "b": {"required_state": [], "timeline_limit": 0, "slow_get_all_rooms": null, + "ranges": [[0, 9], [20, 29]], "filters": {"room_types": [null, "m.space"]}} + } + }"#, + ) + .unwrap(); + let lists = body.lists.unwrap(); + let a = lists.get("a").unwrap(); + assert_eq!(a.slow_get_all_rooms, Some(false)); + assert_eq!(a.ranges, None); + assert_eq!( + a.required_state, + vec![("m.room.name".to_owned(), String::new())] + ); + let b = lists.get("b").unwrap(); + assert_eq!(b.slow_get_all_rooms, None); + assert_eq!(b.ranges, Some(vec![(0, 9), (20, 29)])); + assert_eq!( + b.filters.as_ref().unwrap().room_types, + Some(vec![None, Some("m.space".to_owned())]) + ); + } + + #[test] + fn ordered_map_keeps_document_order_and_the_last_value() { + let map: OrderedMap = serde_json::from_str(r#"{"z": 1, "a": 2, "z": 3}"#).unwrap(); + assert_eq!(map.0, vec![("z".to_owned(), 3), ("a".to_owned(), 2)]); + } + + #[test] + fn extension_defaults() { + let body = SlidingSyncBody::parse( + br#"{"extensions": { + "to_device": {}, + "account_data": {"enabled": true, "lists": null}, + "io.element.msc4308.thread_subscriptions": {"enabled": true}, + "thread_subscriptions": {"enabled": false}, + "org.matrix.msc4354.sticky_events": {"since": "sticky_4"}, + "unknown": {"enabled": true} + }}"#, + ) + .unwrap(); + let extensions = body.extensions.unwrap(); + let to_device = extensions.to_device.unwrap(); + assert_eq!(to_device.enabled, Some(false)); + assert_eq!(to_device.limit, 100); + let account_data = extensions.account_data.unwrap(); + assert_eq!(account_data.lists, None); + assert_eq!(account_data.rooms, Some(vec!["*".to_owned()])); + assert_eq!(extensions.thread_subscriptions.unwrap().enabled, Some(true)); + let sticky = extensions.sticky_events.unwrap(); + assert!(!sticky.enabled); + assert_eq!(sticky.since.as_deref(), Some("sticky_4")); + assert_eq!(extensions.profiles, None); + assert_eq!(extensions.e2ee, None); + } + + #[test] + fn unknown_fields_are_ignored() { + let body = SlidingSyncBody::parse( + br#"{"conn_id": "c", "foo": [1, 2], + "room_subscriptions": {"!r:x": {"required_state": [], "timeline_limit": 1, "bar": null}}}"#, + ) + .unwrap(); + assert_eq!(body.conn_id.as_deref(), Some("c")); + assert_eq!(body.room_subscriptions.unwrap().len(), 1); + } + + #[test] + fn out_of_range_integers_do_not_parse() { + assert!(SlidingSyncBody::parse( + br#"{"lists": {"a": {"required_state": [], "timeline_limit": 1, "ranges": [[0, 18446744073709551616]]}}}"# + ) + .is_err()); + } +} diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 59b88d90de..fd665f7e0e 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -48,6 +48,7 @@ from synapse.storage.roommember import ( ) from synapse.synapse_rust.sliding_sync import ( SlidingSyncHandler as RustSlidingSyncHandler, + SlidingSyncRequest as RustSlidingSyncRequest, SlidingSyncStats as RustSlidingSyncStats, ) from synapse.types import ( @@ -188,6 +189,7 @@ class SlidingSyncHandler: self.relations_handler = hs.get_relations_handler() self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync self.is_mine_id = hs.is_mine_id + self._hs_config = hs.config self.connection_store = SlidingSyncConnectionStore(self.clock, self.store) self.extensions = SlidingSyncExtensionHandler(hs) @@ -210,6 +212,58 @@ class SlidingSyncHandler: # set it. self._rust_sliding_sync_strict = False + @property + def parses_request_body(self) -> bool: + """Whether the servlet should keep the request body's bytes in the + `SlidingSyncConfig`, for the Rust implementation to parse.""" + return self._rust_sliding_sync is not None + + async def _build_rust_request( + self, + sync_config: SlidingSyncConfig, + from_token: SlidingSyncStreamToken | None, + to_token: StreamToken, + ) -> RustSlidingSyncRequest: + """What the Rust implementation reads about the request: the body, and + what Python knows about the user, the tokens and the server. + + Raises: + ValueError if Rust cannot parse the body. Python validated it, so + this is a body Rust does not cover (see `rust/src/sliding_sync/ + request.rs`), and Python computes the request. + """ + user_id = sync_config.user.to_string() + body = sync_config.body_bytes + if body is None: + # Not from a request. Only unset fields can hold `Absent`. + body = sync_config.model_dump_json( + by_alias=True, + exclude_unset=True, + include={"conn_id", "lists", "room_subscriptions", "extensions"}, + ).encode() + is_server_admin = await self.store.is_server_admin(user_id) + client_config = await self.store.get_admin_client_config_for_user(user_id) + # A room key never holds the separator. + to_suffix = (await to_token.to_string(self.store)).split("_", 1)[1] + config = self._hs_config + return RustSlidingSyncRequest( + body, + user_id=user_id, + device_id=sync_config.requester.device_id, + from_room_key=from_token.stream_token.room_key if from_token else None, + connection_position=from_token.connection_position if from_token else None, + to_room_key=to_token.room_key, + to_suffix=to_suffix, + is_server_admin=is_server_admin, + return_soft_failed_events=client_config.return_soft_failed_events, + return_policy_server_spammy_events=client_config.return_policy_server_spammy_events, + have_finished_bg_jobs=await self.store.have_finished_sliding_sync_background_jobs(), + rooms_to_exclude=set(self.rooms_to_exclude_globally), + msc4306_enabled=config.experimental.msc4306_enabled, + msc4354_enabled=config.experimental.msc4354_enabled, + profiles_enabled=config.server.include_profile_updates_in_sync, + ) + async def wait_for_sync_for_user( self, requester: Requester, diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index 6915b51521..b970d742fc 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -990,6 +990,11 @@ class SlidingSyncRestServlet(RestServlet): # by filter ID. For now, we will just prototype with always passing everything # in. body = parse_and_validate_json_object_from_request(request, SlidingSyncBody) + # The Rust implementation parses the body itself, from the same bytes. + body_bytes = None + if self.sliding_sync_handler.parses_request_body: + request.content.seek(0) # type: ignore[union-attr] + body_bytes = request.content.read() # type: ignore[union-attr] # Tag and log useful data to differentiate requests. set_tag( @@ -1028,6 +1033,7 @@ class SlidingSyncRestServlet(RestServlet): lists=body.lists, room_subscriptions=body.room_subscriptions, extensions=body.extensions, + body_bytes=body_bytes, ) ( diff --git a/synapse/synapse_rust/sliding_sync.pyi b/synapse/synapse_rust/sliding_sync.pyi index 138c72e6e9..8f6c670513 100644 --- a/synapse/synapse_rust/sliding_sync.pyi +++ b/synapse/synapse_rust/sliding_sync.pyi @@ -12,6 +12,8 @@ """Simplified sliding sync (MSC4186) in Rust. See `rust/src/sliding_sync/`.""" +from synapse.types import JsonDict, RoomStreamToken + class SlidingSyncHandler: """Computes sliding sync responses in Rust, reading the database through the native Postgres driver. Built by `RustHandlers`.""" @@ -43,3 +45,49 @@ class SlidingSyncStats: def rooms_computed(self) -> int: ... @property def lazy_member_lookups(self) -> int: ... + +class SlidingSyncRequest: + """One sliding sync request as the Rust implementation sees it: the body, + parsed from the bytes pydantic validated, plus what Python adds.""" + + def __init__( + self, + body: bytes, + *, + user_id: str, + device_id: str | None, + from_room_key: RoomStreamToken | None, + connection_position: int | None, + to_room_key: RoomStreamToken, + to_suffix: str, + is_server_admin: bool, + return_soft_failed_events: bool, + return_policy_server_spammy_events: bool, + have_finished_bg_jobs: bool, + rooms_to_exclude: set[str], + msc4306_enabled: bool, + msc4354_enabled: bool, + profiles_enabled: bool, + ) -> None: + """Raises `ValueError` if the body does not parse.""" + def body_for_test(self) -> JsonDict: + """The parsed body with every default filled in, and the lists and + room subscriptions as lists of `[name, value]` pairs.""" + @property + def user_id(self) -> str: ... + @property + def device_id(self) -> str | None: ... + @property + def from_room_key(self) -> RoomStreamToken | None: ... + @property + def connection_position(self) -> int | None: ... + @property + def to_room_key(self) -> RoomStreamToken: ... + @property + def to_suffix(self) -> str: ... + @property + def is_server_admin(self) -> bool: ... + @property + def have_finished_bg_jobs(self) -> bool: ... + @property + def rooms_to_exclude(self) -> set[str]: ... diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index f6d30460b5..fadeb24222 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -73,6 +73,9 @@ class SlidingSyncConfig(SlidingSyncBody): user: UserID requester: Requester + # The request body as the client sent it, which the Rust implementation + # parses itself. `None` when the config was not built from a request. + body_bytes: bytes | None = None model_config = ConfigDict( extra="ignore", frozen=True, diff --git a/tests/rest/client/sliding_sync/test_rust_request.py b/tests/rest/client/sliding_sync/test_rust_request.py new file mode 100644 index 0000000000..73af5a7f38 --- /dev/null +++ b/tests/rest/client/sliding_sync/test_rust_request.py @@ -0,0 +1,563 @@ +# +# 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: +# . +# + +"""The Rust parse of the sliding sync request body +(`rust/src/sliding_sync/request.rs`) against pydantic's `SlidingSyncBody`.""" + +import json +import random +from typing import Any +from unittest import SkipTest +from unittest.mock import patch + +from pydantic import ValidationError + +from twisted.internet.testing import MemoryReactor + +from synapse.rest import admin +from synapse.rest.client import login, room, sync +from synapse.server import HomeServer +from synapse.synapse_rust.sliding_sync import SlidingSyncRequest +from synapse.types import ( + Absent, + JsonDict, + RoomStreamToken, + SlidingSyncStreamToken, + StreamToken, +) +from synapse.types.handlers.sliding_sync import SlidingSyncConfig +from synapse.types.rest.client import SlidingSyncBody +from synapse.util.clock import Clock + +import tests.utils +from tests import unittest +from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase + +# A JSON object as the generator builds it: its entries in order, so that a +# key can repeat. +Entries = list[tuple[str, Any]] + + +class _Object: + def __init__(self, entries: Entries) -> None: + self.entries = entries + + +def _encode(value: Any, rng: random.Random) -> str: + """Encode `value` as JSON text, keeping the repeated keys of an + `_Object` and varying the whitespace and string escapes.""" + space = rng.choice(["", " ", "\n "]) + if isinstance(value, _Object): + items = [ + f"{_encode(key, rng)}:{space}{_encode(item, rng)}" + for key, item in value.entries + ] + return "{" + f",{space}".join(items) + "}" + if isinstance(value, list): + return "[" + f",{space}".join(_encode(item, rng) for item in value) + "]" + return json.dumps(value, ensure_ascii=rng.random() < 0.5) + + +_STATE_PARTS = [ + "*", + "$LAZY", + "$ME", + "", + "m.room.member", + "m.room.name", + "m.space.child", + "@alice:test", + "org.example.é", + "☃", +] +_NAMES = ["a", "b", "list", "", "é", "x" * 64, "☃", "a b"] +_ROOM_IDS = ["!a:test", "!b:test", "!c:test", "!é:test"] + + +class _BodyGenerator: + """Random request bodies, most of which pydantic accepts.""" + + def __init__(self, rng: random.Random) -> None: + self.rng = rng + + def maybe(self, value: Any, p: float = 0.5) -> list[Any]: + """`[value]` with probability `p`, else `[]`: an optional field.""" + return [value] if self.rng.random() < p else [] + + def nullable(self, value: Any) -> Any: + return None if self.rng.random() < 0.15 else value + + def unknown_fields(self) -> Entries: + return self.maybe( + ("org.example.unknown", self.rng.choice([1, "x", None, []])), 0.1 + ) + + def coin(self) -> bool: + return self.rng.random() < 0.5 + + def strings(self, pool: list[str], allow_none: bool = False) -> list[Any]: + values: list[Any] = list(pool) + ([None] if allow_none else []) + return self.rng.sample(values, self.rng.randint(0, min(4, len(values)))) + + def required_state(self) -> list[list[str]]: + return [ + [self.rng.choice(_STATE_PARTS), self.rng.choice(_STATE_PARTS)] + for _ in range(self.rng.randint(0, 5)) + ] + + def room_params(self) -> Entries: + return [ + ("required_state", self.required_state()), + ("timeline_limit", self.rng.choice([0, 1, 10, 1000, -3])), + *self.unknown_fields(), + ] + + def filters(self) -> _Object: + rng = self.rng + fields: Entries = [ + ("is_dm", self.nullable(self.coin())), + ("spaces", self.nullable(self.strings(_ROOM_IDS))), + ("is_encrypted", self.nullable(self.coin())), + ("is_invite", self.nullable(self.coin())), + ( + "room_types", + self.nullable(self.strings(["m.space", "org.example"], True)), + ), + ( + "not_room_types", + self.nullable(self.strings(["m.space", "org.example"], True)), + ), + ("room_name_like", self.nullable(rng.choice(["foo", "", "É"]))), + ("tags", self.nullable(self.strings(["m.favourite", "u.work"]))), + ("not_tags", self.nullable(self.strings(["m.lowpriority", "u.x"]))), + ] + entries = [field for field in fields if rng.random() < 0.4] + return _Object(entries + self.unknown_fields()) + + def sliding_list(self) -> _Object: + rng = self.rng + entries = self.room_params() + if rng.random() < 0.7: + ranges = [ + sorted([rng.randint(0, 200), rng.randint(0, 200)]) + for _ in range(rng.randint(0, 3)) + ] + entries.append(("ranges", self.nullable(ranges))) + entries += self.maybe(("slow_get_all_rooms", self.nullable(self.coin())), 0.2) + entries += self.maybe(("filters", self.nullable(self.filters())), 0.4) + rng.shuffle(entries) + return _Object(entries) + + def keyed(self, keys: list[str], make: Any) -> _Object: + """An object of up to four entries, sometimes with a key repeated.""" + entries = [ + (self.rng.choice(keys), make()) for _ in range(self.rng.randint(0, 4)) + ] + return _Object(entries) + + def lists_and_rooms_extension(self) -> _Object: + entries: Entries = [ + *self.maybe(("enabled", self.nullable(self.coin()))), + *self.maybe(("lists", self.nullable(self.strings(["*", *_NAMES])))), + *self.maybe(("rooms", self.nullable(self.strings(["*", *_ROOM_IDS])))), + ] + return _Object(entries + self.unknown_fields()) + + def extensions(self) -> _Object: + rng = self.rng + candidates: Entries = [ + ( + "to_device", + _Object( + [ + *self.maybe(("enabled", self.nullable(self.coin()))), + *self.maybe(("limit", rng.choice([0, 7, 100, -1]))), + *self.maybe(("since", self.nullable(str(rng.randint(0, 99))))), + ] + ), + ), + ("e2ee", _Object(self.maybe(("enabled", self.nullable(self.coin()))))), + ("account_data", self.lists_and_rooms_extension()), + ("receipts", self.lists_and_rooms_extension()), + ("typing", self.lists_and_rooms_extension()), + ( + "io.element.msc4308.thread_subscriptions", + _Object( + [ + *self.maybe(("enabled", self.nullable(self.coin()))), + *self.maybe(("limit", rng.choice([1, 100]))), + ] + ), + ), + # Only the unstable name counts: this one is ignored. + ("thread_subscriptions", _Object([("enabled", True)])), + ( + "org.matrix.msc4354.sticky_events", + _Object( + [ + *self.maybe(("enabled", self.coin())), + *self.maybe(("limit", rng.choice([0, 5, 100]))), + *self.maybe( + ("since", rng.choice(["sticky_0", "sticky_m4~1.6"])) + ), + ] + ), + ), + ( + "org.matrix.msc4262.profiles", + _Object( + [ + *self.maybe(("enabled", self.coin())), + *self.maybe(("fields", self.strings(["displayname", "m.tz"]))), + ] + ), + ), + ("org.example.unknown_extension", _Object([("enabled", True)])), + ] + entries = [ + (name, self.nullable(value)) + if name not in _NOT_NULLABLE_EXTENSIONS + else (name, value) + for name, value in candidates + if rng.random() < 0.4 + ] + return _Object(entries) + + def body(self) -> _Object: + entries: Entries = [ + *self.maybe(("conn_id", self.nullable(self.rng.choice(["", "room-list"])))), + *self.maybe( + ("lists", self.nullable(self.keyed(_NAMES, self.sliding_list))), 0.8 + ), + *self.maybe( + ( + "room_subscriptions", + self.nullable( + self.keyed(_ROOM_IDS, lambda: _Object(self.room_params())) + ), + ), + ), + *self.maybe(("extensions", self.nullable(self.extensions()))), + *self.unknown_fields(), + ] + self.rng.shuffle(entries) + return _Object(entries) + + +# Pydantic rejects an explicit `null` for these: they are `Absent` or a value. +_NOT_NULLABLE_EXTENSIONS = { + "org.matrix.msc4354.sticky_events", + "org.matrix.msc4262.profiles", +} + + +def _expected(body: SlidingSyncBody) -> JsonDict: + """What `SlidingSyncRequest.body_for_test` should return for `body`.""" + + def room_params(params: SlidingSyncBody.CommonRoomParameters) -> JsonDict: + return { + "required_state": [list(pair) for pair in params.required_state], + "timeline_limit": params.timeline_limit, + } + + def sliding_list(sliding_list: SlidingSyncBody.SlidingSyncList) -> JsonDict: + return { + **room_params(sliding_list), + "ranges": ( + None + if sliding_list.ranges is None + else [list(r) for r in sliding_list.ranges] + ), + "slow_get_all_rooms": sliding_list.slow_get_all_rooms, + "filters": ( + None + if sliding_list.filters is None + else sliding_list.filters.model_dump() + ), + } + + def dump(model: Any) -> Any: + return None if model is None else model.model_dump() + + extensions = None + if body.extensions is not None: + ext = body.extensions + sticky = None + if ext.sticky_events is not Absent: + sticky = { + "enabled": ext.sticky_events.enabled, + "limit": ext.sticky_events.limit, + "since": ( + ext.sticky_events.since.serialise() + if ext.sticky_events.since is not Absent + else None + ), + } + profiles = None + if ext.profiles is not Absent: + profiles = { + "enabled": ext.profiles.enabled, + "fields": ( + list(ext.profiles.fields) + if ext.profiles.fields is not Absent + else None + ), + } + extensions = { + "to_device": dump(ext.to_device), + "e2ee": dump(ext.e2ee), + "account_data": dump(ext.account_data), + "receipts": dump(ext.receipts), + "typing": dump(ext.typing), + "io.element.msc4308.thread_subscriptions": dump(ext.thread_subscriptions), + "org.matrix.msc4354.sticky_events": sticky, + "org.matrix.msc4262.profiles": profiles, + } + + return { + "conn_id": body.conn_id, + "lists": ( + None + if body.lists is None + else [[name, sliding_list(value)] for name, value in body.lists.items()] + ), + "room_subscriptions": ( + None + if body.room_subscriptions is None + else [ + [room_id, room_params(value)] + for room_id, value in body.room_subscriptions.items() + ] + ), + "extensions": extensions, + } + + +def _as_json(value: Any) -> Any: + """`value` with its tuples as lists, for comparing.""" + return json.loads(json.dumps(value)) + + +def _rust_request(body: bytes) -> SlidingSyncRequest: + return SlidingSyncRequest( + body, + user_id="@alice:test", + device_id="DEVICE", + from_room_key=None, + connection_position=None, + to_room_key=RoomStreamToken(stream=1), + to_suffix="0_0", + is_server_admin=False, + return_soft_failed_events=False, + return_policy_server_spammy_events=False, + have_finished_bg_jobs=True, + rooms_to_exclude=set(), + msc4306_enabled=False, + msc4354_enabled=False, + profiles_enabled=False, + ) + + +class RustRequestParseTestCase(unittest.TestCase): + """Rust parses every body pydantic accepts the same way.""" + + def test_generated_bodies(self) -> None: + accepted = 0 + for seed in range(400): + rng = random.Random(seed) + text = _encode(_BodyGenerator(rng).body(), rng) + try: + body = SlidingSyncBody.model_validate(json.loads(text)) + except ValidationError: + continue + accepted += 1 + expected = _as_json(_expected(body)) + self.assertEqual( + _as_json(_rust_request(text.encode()).body_for_test()), + expected, + f"seed {seed}: {text}", + ) + # What `_build_rust_request` sends for a config that was not built + # from a request. + reencoded = body.model_dump_json(by_alias=True, exclude_unset=True) + self.assertEqual( + _as_json(_rust_request(reencoded.encode()).body_for_test()), + expected, + f"seed {seed}, re-encoded: {reencoded}", + ) + # The generator is no use if pydantic rejects what it makes. + self.assertGreater(accepted, 300) + + def test_repeated_list_name(self) -> None: + """A repeated key keeps its first position and its last value, as in + the dict `json.loads` builds.""" + text = ( + '{"lists": {' + '"b": {"required_state": [], "timeline_limit": 1},' + '"a": {"required_state": [], "timeline_limit": 2},' + '"b": {"required_state": [], "timeline_limit": 3}}}' + ) + body = SlidingSyncBody.model_validate(json.loads(text)) + rust = _rust_request(text.encode()).body_for_test() + self.assertEqual(_as_json(rust), _as_json(_expected(body))) + self.assertEqual([name for name, _ in rust["lists"]], ["b", "a"]) + self.assertEqual(rust["lists"][0][1]["timeline_limit"], 3) + + def test_unparseable_body(self) -> None: + """A body Rust does not cover raises `ValueError`, and Python computes + the request.""" + text = ( + '{"lists": {"a": {"required_state": [], "timeline_limit": 1,' + ' "ranges": [[0, 100000000000000000000000]]}}}' + ) + SlidingSyncBody.model_validate(json.loads(text)) + with self.assertRaises(ValueError): + _rust_request(text.encode()) + + def test_tokens_go_together(self) -> None: + with self.assertRaises(ValueError): + SlidingSyncRequest( + b"{}", + user_id="@alice:test", + device_id=None, + from_room_key=RoomStreamToken(stream=1), + connection_position=None, + to_room_key=RoomStreamToken(stream=1), + to_suffix="0", + is_server_admin=False, + return_soft_failed_events=False, + return_policy_server_spammy_events=False, + have_finished_bg_jobs=True, + rooms_to_exclude=set(), + msc4306_enabled=False, + msc4354_enabled=False, + profiles_enabled=False, + ) + + +class RustRequestBuildTestCase(SlidingSyncBase): + """The servlet keeps the body's bytes, and the handler builds the Rust + request from them.""" + + servlets = [ + admin.register_servlets, + login.register_servlets, + room.register_servlets, + sync.register_servlets, + ] + + def default_config(self) -> JsonDict: + config = super().default_config() + config["experimental_features"]["rust_sliding_sync"] = "on" + config["exclude_rooms_from_sync"] = ["!excluded:test"] + return config + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + if tests.utils.RUST_SLIDING_SYNC_FOR_TESTS not in (None, "on"): + raise SkipTest("SYNAPSE_TEST_RUST_SLIDING_SYNC overrides the config") + hs = self.setup_test_homeserver() + if hs.get_sliding_sync_handler()._rust_sliding_sync is None: + raise SkipTest("needs the native Rust Postgres driver") + return hs + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + super().prepare(reactor, clock, hs) + self.handler = hs.get_sliding_sync_handler() + self.user = self.register_user("alice", "pass", admin=True) + self.tok = self.login("alice", "pass") + + def _capture_configs(self, text: str, since: str | None = None) -> list[Any]: + """Send `text` as a sliding sync body, and return the arguments of each + `current_sync_for_user` call it led to.""" + calls: list[Any] = [] + original = self.handler.current_sync_for_user + + async def capture( + sync_config: SlidingSyncConfig, + to_token: StreamToken, + from_token: SlidingSyncStreamToken | None = None, + ) -> Any: + calls.append((sync_config, to_token, from_token)) + return await original(sync_config, to_token=to_token, from_token=from_token) + + path = self.sync_endpoint + (f"?pos={since}" if since else "") + with patch.object(self.handler, "current_sync_for_user", side_effect=capture): + channel = self.make_request( + "POST", path, content=text.encode(), access_token=self.tok + ) + self.assertEqual(channel.code, 200, channel.json_body) + return calls + + def test_body_bytes_and_request(self) -> None: + text = ( + '{"lists": {"a": {"ranges": [[0, 9]],' + ' "required_state": [["m.room.name", ""]], "timeline_limit": 3}},' + ' "conn_id": "c"}' + ) + [(sync_config, to_token, from_token)] = self._capture_configs(text) + self.assertEqual(sync_config.body_bytes, text.encode()) + self.assertIsNone(from_token) + + request = self.get_success( + self.handler._build_rust_request(sync_config, from_token, to_token) + ) + self.assertEqual(request.user_id, self.user) + self.assertEqual(request.device_id, sync_config.requester.device_id) + self.assertIsNone(request.from_room_key) + self.assertIsNone(request.connection_position) + self.assertEqual(request.to_room_key, to_token.room_key) + self.assertEqual( + f"{self.get_success(to_token.room_key.to_string(self.store))}_" + f"{request.to_suffix}", + self.get_success(to_token.to_string(self.store)), + ) + self.assertTrue(request.is_server_admin) + self.assertTrue(request.have_finished_bg_jobs) + self.assertEqual(request.rooms_to_exclude, {"!excluded:test"}) + self.assertEqual( + _as_json(request.body_for_test()), + _as_json(_expected(SlidingSyncBody.model_validate(json.loads(text)))), + ) + + def test_incremental_request(self) -> None: + text = '{"lists": {"a": {"required_state": [], "timeline_limit": 1}}}' + [(sync_config, to_token, _)] = self._capture_configs(text) + channel = self.make_request( + "POST", self.sync_endpoint, content=text.encode(), access_token=self.tok + ) + pos = channel.json_body["pos"] + [(sync_config, to_token, from_token)] = self._capture_configs(text, pos) + assert from_token is not None + request = self.get_success( + self.handler._build_rust_request(sync_config, from_token, to_token) + ) + self.assertEqual(request.from_room_key, from_token.stream_token.room_key) + self.assertEqual(request.connection_position, from_token.connection_position) + + def test_config_without_body_bytes(self) -> None: + """A config not built from a request is re-encoded for Rust.""" + text = '{"lists": {"a": {"required_state": [["*", "*"]], "timeline_limit": 2}}}' + [(sync_config, to_token, _)] = self._capture_configs(text) + config = sync_config.model_copy(update={"body_bytes": None}) + request = self.get_success( + self.handler._build_rust_request(config, None, to_token) + ) + self.assertEqual( + _as_json(request.body_for_test()), + _as_json(_expected(SlidingSyncBody.model_validate(json.loads(text)))), + ) + + @property + def store(self) -> Any: + return self.hs.get_datastores().main