From a3fe79ed8c1b39e7f4f256eee4e35edc1ca787e8 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 16 Oct 2026 11:35:02 +0000 Subject: [PATCH] Port the room list sort and ranges to Rust `rust/src/sliding_sync/room_lists/sort.rs` holds `sort_rooms` as a `SortPlan`: the decision logic with the stream change cache shortcut, and the caller fetching what each step asks for. `lists.rs` holds the ranges to `SYNC` operations with the full-range no-sort shortcut, the config combining across lists and room subscriptions, and the decision half of `_filter_relevant_rooms_to_send`, including the reset for a connection with too many rooms to update. The Python ranges loop moves, unchanged, into `_sync_list_ops`, so the test can call it. It keeps a quirk the Rust ports: once a room combines with a config from an earlier list, every later room in the same list takes the combined config too. The relevant room map is an `IndexMap`, which keeps Python's dict order, so `indexmap` becomes a direct dependency. It was already in the tree. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG --- Cargo.lock | 1 + rust/Cargo.toml | 2 + rust/src/sliding_sync/room_lists/lists.rs | 215 ++++++++++++++++ rust/src/sliding_sync/room_lists/mod.rs | 143 ++++++++++- rust/src/sliding_sync/room_lists/sort.rs | 169 +++++++++++++ synapse/handlers/sliding_sync/room_lists.py | 140 +++++------ synapse/synapse_rust/sliding_sync.pyi | 30 +++ tests/handlers/test_sliding_sync_rust_pure.py | 234 +++++++++++++++++- 8 files changed, 856 insertions(+), 78 deletions(-) create mode 100644 rust/src/sliding_sync/room_lists/lists.rs create mode 100644 rust/src/sliding_sync/room_lists/sort.rs diff --git a/Cargo.lock b/Cargo.lock index c238030f4f..c137810acf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2028,6 +2028,7 @@ dependencies = [ "hyper", "hyper-util", "icu_segmenter", + "indexmap", "ipnet", "itertools", "lazy_static", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index 98a4c35f8b..463cac4b20 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -84,6 +84,8 @@ once_cell = "1.18.0" # Fast hashing for the integer-keyed maps in state resolution. rustc-hash = "2.1.1" itertools = "0.15.0" +# Maps that keep insertion order, as the Python dicts they port do. +indexmap = "2.10.0" # Used by the `libpq` module to load the system libpq at runtime, so that # libpq is an optional runtime dependency rather than a build-time one. libloading = "0.8" diff --git a/rust/src/sliding_sync/room_lists/lists.rs b/rust/src/sliding_sync/room_lists/lists.rs new file mode 100644 index 0000000000..61c1c1da7b --- /dev/null +++ b/rust/src/sliding_sync/room_lists/lists.rs @@ -0,0 +1,215 @@ +/* + * 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: + * . + */ + +//! A list's ranges to its `SYNC` operations, the room configs the lists +//! and room subscriptions combine into, and which of those rooms may have +//! updates to send (`_filter_relevant_rooms_to_send`). + +use std::collections::HashSet; +use std::sync::Arc; + +use indexmap::IndexMap; + +use crate::sliding_sync::config::RoomSyncConfig; +use crate::sliding_sync::connection::{HaveSentRoom, PerConnectionState}; + +/// `MINIMUM_NOT_USED_AGE_EXPIRY`, in milliseconds: a connection with too +/// many rooms to update is only reset after this long unused. +pub const MINIMUM_NOT_USED_AGE_EXPIRY_MS: i64 = 60 * 60 * 1000; + +/// `NUM_ROOMS_THRESHOLD`: more rooms with updates than this, on a +/// connection unused for [`MINIMUM_NOT_USED_AGE_EXPIRY_MS`], resets it. +pub const NUM_ROOMS_THRESHOLD: usize = 100; + +/// The rooms a request covers, to their combined configs, in the order +/// the lists and room subscriptions add them. +pub type RelevantRoomMap = IndexMap, RoomSyncConfig>; + +/// `SlidingWindowList.Operation`: always a `SYNC`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SyncOp { + pub range: (u64, u64), + pub room_ids: Vec>, +} + +/// Whether a list with these `ranges` and `count` rooms needs sorting: +/// not when one range covers every room from the start, and not without +/// ranges. +pub fn needs_sort(ranges: &[(u64, u64)], count: usize) -> bool { + match ranges { + [] => false, + [(0, end)] => (*end as u128) < (count as u128).saturating_sub(1), + _ => true, + } +} + +/// The number of top rooms the ranges need sorted: the end of the last +/// one, inclusive. +pub fn sort_limit(ranges: &[(u64, u64)]) -> usize { + ranges + .iter() + .map(|(_, end)| { + usize::try_from(*end) + .unwrap_or(usize::MAX) + .saturating_add(1) + }) + .max() + .unwrap_or(0) +} + +/// `_sync_list_ops`: the `SYNC` operations of a list's `ranges` over its +/// sorted rooms, adding each room in a range to `relevant_room_map` with +/// the list's config combined with the room's existing one. +/// +/// As in Python, once a room combines with an existing config, the rooms +/// after it in the list take the combined config too. +pub fn sync_list_ops( + ranges: &[(u64, u64)], + sorted_room_ids: &[Arc], + room_sync_config: &RoomSyncConfig, + relevant_room_map: &mut RelevantRoomMap, +) -> Vec { + let mut room_sync_config = room_sync_config.clone(); + let mut ops = Vec::with_capacity(ranges.len()); + for &(start, end) in ranges { + // An end before the start takes no rooms, as in Python. + let max_num_rooms = if end < start { + 0 + } else { + (end - start).saturating_add(1) + }; + let skip = usize::try_from(start).unwrap_or(usize::MAX); + let mut room_ids = Vec::new(); + for room_id in sorted_room_ids.iter().skip(skip) { + if room_ids.len() as u64 >= max_num_rooms { + break; + } + if let Some(existing) = relevant_room_map.get(room_id) { + room_sync_config = existing.combine(&room_sync_config); + } + relevant_room_map.insert(Arc::clone(room_id), room_sync_config.clone()); + room_ids.push(Arc::clone(room_id)); + } + ops.push(SyncOp { + range: (start, end), + room_ids, + }); + } + ops +} + +/// Add a room subscription's room to `relevant_room_map`, combined with +/// the room's existing config. +pub fn add_room_subscription( + relevant_room_map: &mut RelevantRoomMap, + room_id: &Arc, + room_sync_config: RoomSyncConfig, +) { + let config = match relevant_room_map.get(room_id) { + Some(existing) => existing.combine(&room_sync_config), + None => room_sync_config, + }; + relevant_room_map.insert(Arc::clone(room_id), config); +} + +/// `SlidingSyncUnknownPosition`: the connection is reset, as it has too +/// many rooms to update. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct UnknownPosition; + +/// The rooms of an incremental sync that may have updates, the decision +/// half of `_filter_relevant_rooms_to_send`: rooms whose config expanded, +/// rooms never sent or with updates left over, and +/// `rooms_that_have_updates`, the rooms the events stream has new events +/// in since the `from_token`. Their order is `relevant_room_map`'s. +pub fn rooms_to_send( + relevant_room_map: &RelevantRoomMap, + previous_state: &PerConnectionState, + rooms_that_have_updates: &HashSet>, + now_ms: i64, +) -> Result>, UnknownPosition> { + let mut rooms_should_send: HashSet<&Arc> = HashSet::new(); + for (room_id, room_config) in relevant_room_map { + if let Some(prev) = previous_state.room_configs.get(&**room_id) { + // Timeline limit increases and required state additions. + if prev.combine(room_config) != *prev { + rooms_should_send.insert(room_id); + continue; + } + } + match previous_state.rooms.have_sent_room(room_id) { + HaveSentRoom::Never | HaveSentRoom::Previously(_) => { + rooms_should_send.insert(room_id); + } + // Everything up to the `from_token` was sent. + HaveSentRoom::Live => {} + } + } + + // A client with lots to catch up on does better starting over with a + // small range, unless it synced recently and would loop. + if rooms_that_have_updates.len() > NUM_ROOMS_THRESHOLD { + if let Some(last_used_ts) = previous_state.last_used_ts { + if now_ms - last_used_ts > MINIMUM_NOT_USED_AGE_EXPIRY_MS { + return Err(UnknownPosition); + } + } + } + + Ok(relevant_room_map + .keys() + .filter(|room_id| { + rooms_should_send.contains(room_id) || rooms_that_have_updates.contains(*room_id) + }) + .cloned() + .collect()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn config(limit: i64) -> RoomSyncConfig { + RoomSyncConfig { + timeline_limit: limit, + ..Default::default() + } + } + + #[test] + fn full_range_needs_no_sort() { + assert!(!needs_sort(&[], 5)); + assert!(!needs_sort(&[(0, 4)], 5)); + assert!(!needs_sort(&[(0, 0)], 0)); + assert!(!needs_sort(&[(0, 10)], 5)); + assert!(needs_sort(&[(0, 3)], 5)); + assert!(needs_sort(&[(1, 10)], 5)); + assert!(needs_sort(&[(0, 10), (0, 10)], 5)); + assert_eq!(sort_limit(&[(0, 3), (5, 9)]), 10); + } + + #[test] + fn combined_config_carries_on_down_the_list() { + let rooms: Vec> = ["!a", "!b", "!c"].into_iter().map(Arc::from).collect(); + let mut relevant = RelevantRoomMap::new(); + relevant.insert(Arc::from("!b"), config(10)); + let ops = sync_list_ops(&[(0, 1), (2, 5)], &rooms, &config(1), &mut relevant); + assert_eq!(ops[0].room_ids, rooms[..2].to_vec()); + assert_eq!(ops[1].room_ids, rooms[2..].to_vec()); + assert_eq!(relevant["!a"].timeline_limit, 1); + assert_eq!(relevant["!b"].timeline_limit, 10); + // Python's leak: `!c` takes `!b`'s combined config. + assert_eq!(relevant["!c"].timeline_limit, 10); + } +} diff --git a/rust/src/sliding_sync/room_lists/mod.rs b/rust/src/sliding_sync/room_lists/mod.rs index 4fc5734f69..13414bf02b 100644 --- a/rust/src/sliding_sync/room_lists/mod.rs +++ b/rust/src/sliding_sync/room_lists/mod.rs @@ -16,13 +16,23 @@ //! sorted and cut to the requested ranges //! (`synapse.handlers.sliding_sync.room_lists`). +use std::collections::{BTreeMap, HashMap, HashSet}; use std::sync::Arc; -use pyo3::{prelude::*, types::PyModule}; +use pyo3::{ + exceptions::{PyKeyError, PyValueError}, + prelude::*, + types::PyModule, +}; +use crate::sliding_sync::config::{config_from_py, to_test_tuple, RoomSyncConfig}; +use crate::sliding_sync::connection::{HaveSentRoom, PerConnectionState, RoomStatusMap}; use crate::storage::sync_store::PersistedEventPosition; +use crate::types::stream_token::RoomStreamToken; pub mod filter; +pub mod lists; +pub mod sort; /// A user's membership in a room at the `to_token`, as the sliding sync /// tables record it: `RoomsForUserSlidingSync`, which also stands for @@ -75,7 +85,138 @@ pub(crate) fn room_from_py(room: PyRoomForUser) -> RoomForUser { } } +type PyConfig = (i64, BTreeMap>); + +/// `sort_rooms` through the Rust implementation, with the stream change +/// cache as `earliest_cache_position` and `cache` (room to its max position +/// of last change), and the database as `recheck` (room to the stream of +/// its last event before the `to_token`) and `joined_positions`. Returns +/// the sorted room IDs. Raises `KeyError` as Python does. +#[pyfunction] +#[pyo3(signature = (rooms, earliest_cache_position, cache, to_stream, limit, recheck, joined_positions))] +fn sort_rooms_for_test( + rooms: Vec, + earliest_cache_position: i64, + cache: HashMap, + to_stream: i64, + limit: Option, + recheck: HashMap>, + joined_positions: HashMap, +) -> PyResult> { + let rooms: Vec = rooms.into_iter().map(room_from_py).collect(); + let rooms: Vec<&RoomForUser> = rooms.iter().collect(); + let mut plan = sort::SortPlan::new(&rooms, earliest_cache_position, |room_id| { + cache.get(room_id).copied() + }); + for room_id in plan.rooms_to_recheck(limit, to_stream) { + let result = recheck + .get(&*room_id) + .ok_or_else(|| PyValueError::new_err(format!("no recheck for {room_id}")))?; + plan.apply_recheck(&room_id, *result); + } + let sorted = match plan.early_result(&rooms, limit) { + Some(sorted) => sorted, + None => { + let joined: HashMap, i64> = sort::SortPlan::joined_rooms(&rooms) + .into_iter() + .filter_map(|room_id| { + joined_positions + .get(&*room_id) + .map(|position| (room_id, *position)) + }) + .collect(); + plan.sort_all(&rooms, &joined) + .map_err(|e| PyKeyError::new_err(e.to_string()))? + } + }; + Ok(sorted.iter().map(|room| room.room_id.to_string()).collect()) +} + +/// `_sync_list_ops` through the Rust implementation, the relevant room map +/// as a list of `(room_id, config)`. Returns `(ops, relevant_room_map)`, +/// each op as `(range, room_ids)`. +#[pyfunction] +#[allow(clippy::type_complexity)] +fn sync_list_ops_for_test( + ranges: Vec<(u64, u64)>, + sorted_room_ids: Vec, + room_sync_config: PyConfig, + relevant_room_map: Vec<(String, PyConfig)>, +) -> (Vec<((u64, u64), Vec)>, Vec<(String, PyConfig)>) { + let mut relevant: lists::RelevantRoomMap = relevant_room_map + .into_iter() + .map(|(room_id, (limit, map))| (Arc::from(room_id), config_from_py(limit, map))) + .collect(); + let sorted: Vec> = sorted_room_ids.into_iter().map(Arc::from).collect(); + let ops = lists::sync_list_ops( + &ranges, + &sorted, + &config_from_py(room_sync_config.0, room_sync_config.1), + &mut relevant, + ); + ( + ops.into_iter() + .map(|op| { + ( + op.range, + op.room_ids.iter().map(|r| r.to_string()).collect(), + ) + }) + .collect(), + relevant + .iter() + .map(|(room_id, config)| (room_id.to_string(), to_test_tuple(config))) + .collect(), + ) +} + +/// The decision of `_filter_relevant_rooms_to_send` through the Rust +/// implementation, each room's status as `(flag, token)` with a token an +/// int. Returns the room IDs to send, or raises `ValueError` for an +/// unknown position. +#[pyfunction] +fn rooms_to_send_for_test( + relevant_room_map: Vec<(String, PyConfig)>, + room_configs: HashMap, + statuses: HashMap)>, + last_used_ts: Option, + rooms_that_have_updates: HashSet, + now_ms: i64, +) -> PyResult> { + let relevant: lists::RelevantRoomMap = relevant_room_map + .into_iter() + .map(|(room_id, (limit, map))| (Arc::from(room_id), config_from_py(limit, map))) + .collect(); + let mut rooms = HashMap::new(); + for (room_id, (flag, token)) in statuses { + let status = match (flag.as_str(), token) { + ("never", _) => HaveSentRoom::Never, + ("live", _) => HaveSentRoom::Live, + (_, Some(token)) => HaveSentRoom::Previously(RoomStreamToken::live(token)), + _ => return Err(PyValueError::new_err("bad status")), + }; + rooms.insert(room_id, status); + } + let room_configs: HashMap = room_configs + .into_iter() + .map(|(room_id, (limit, map))| (room_id, config_from_py(limit, map))) + .collect(); + let state: PerConnectionState = PerConnectionState { + last_used_ts, + rooms: RoomStatusMap::new(rooms), + room_configs: Arc::new(room_configs), + ..Default::default() + }; + let updates: HashSet> = rooms_that_have_updates.into_iter().map(Arc::from).collect(); + lists::rooms_to_send(&relevant, &state, &updates, now_ms) + .map(|rooms| rooms.iter().map(|r| r.to_string()).collect()) + .map_err(|_| PyValueError::new_err("unknown position")) +} + pub fn register_test_functions(m: &Bound<'_, PyModule>) -> PyResult<()> { filter::register_test_functions(m)?; + m.add_function(wrap_pyfunction!(sort_rooms_for_test, m)?)?; + m.add_function(wrap_pyfunction!(sync_list_ops_for_test, m)?)?; + m.add_function(wrap_pyfunction!(rooms_to_send_for_test, m)?)?; Ok(()) } diff --git a/rust/src/sliding_sync/room_lists/sort.rs b/rust/src/sliding_sync/room_lists/sort.rs new file mode 100644 index 0000000000..7af21bcac3 --- /dev/null +++ b/rust/src/sliding_sync/room_lists/sort.rs @@ -0,0 +1,169 @@ +/* + * 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: + * . + */ + +//! `sort_rooms`: a list's rooms by the stream ordering of the last event +//! the user can see in each, latest first. +//! +//! Python's method reads the events stream change cache and the database +//! as it goes. Here [`SortPlan`] is the decision logic, and the caller +//! fetches what it asks for between the steps: +//! +//! 1. [`SortPlan::new`] reads the cached positions. +//! 2. With a `limit`, [`SortPlan::rooms_to_recheck`] are cached positions +//! past the `to_token`, whose last event before it the caller looks up +//! and passes to [`SortPlan::apply_recheck`]. +//! 3. [`SortPlan::early_result`] is the answer if the cache covered the +//! top `limit` rooms. +//! 4. Otherwise the caller fetches the last event positions of +//! [`SortPlan::joined_rooms`] and [`SortPlan::sort_all`] sorts. + +use std::collections::HashMap; +use std::sync::Arc; + +use anyhow::anyhow; + +use crate::sliding_sync::room_lists::RoomForUser; + +/// The positions [`sort_rooms`](self) has decided on so far. +#[derive(Debug, Clone)] +pub struct SortPlan { + /// The earliest position the events stream change cache knows. + earliest_cache_position: i64, + /// The last activity of each room the user is not joined to: their + /// membership event, as they see nothing after it. + last_activity: HashMap, i64>, + /// The positions known to be in the stream change cache's range. + cached_positions: HashMap, i64>, +} + +impl SortPlan { + /// Read the positions the cache gives. `max_pos_of_last_change` is + /// the cache's `get_max_pos_of_last_change`. + pub fn new( + rooms: &[&RoomForUser], + earliest_cache_position: i64, + max_pos_of_last_change: impl Fn(&str) -> Option, + ) -> Self { + let mut last_activity = HashMap::new(); + let mut cached_positions = HashMap::new(); + for room in rooms { + if &*room.membership == "join" { + if let Some(position) = max_pos_of_last_change(&room.room_id) { + cached_positions.insert(Arc::clone(&room.room_id), position); + } + } else { + // FIXME, as in Python: invited or knocking users might see + // further, depending on the history visibility. + let stream = room.event_pos.stream; + last_activity.insert(Arc::clone(&room.room_id), stream); + if stream > earliest_cache_position { + cached_positions.insert(Arc::clone(&room.room_id), stream); + } + } + } + SortPlan { + earliest_cache_position, + last_activity, + cached_positions, + } + } + + /// Whether the cached positions cover the top `limit` rooms. + fn enough_cached(&self, limit: Option) -> bool { + limit.is_some_and(|limit| self.cached_positions.len() >= limit) + } + + /// The rooms whose cached position is past `to_stream`, which need + /// their last event before the `to_token` looked up. Only with a + /// `limit` the cache could answer. + pub fn rooms_to_recheck(&self, limit: Option, to_stream: i64) -> Vec> { + if !self.enough_cached(limit) { + return Vec::new(); + } + self.cached_positions + .iter() + .filter(|(_, position)| **position > to_stream) + .map(|(room_id, _)| Arc::clone(room_id)) + .collect() + } + + /// Record a recheck: `last_event_stream` is the stream ordering of the + /// room's last event before the `to_token`, if any. A position outside + /// the cache's range drops the room from the cached positions. + pub fn apply_recheck(&mut self, room_id: &str, last_event_stream: Option) { + match last_event_stream { + Some(stream) if stream > self.earliest_cache_position => { + self.cached_positions.insert(Arc::from(room_id), stream); + } + _ => { + self.cached_positions.remove(room_id); + } + } + } + + /// The top rooms from the cache alone, if it covers `limit` of them: + /// the rooms with a cached position, latest first, ties in `rooms` + /// order. + pub fn early_result<'a>( + &self, + rooms: &[&'a RoomForUser], + limit: Option, + ) -> Option> { + if !self.enough_cached(limit) { + return None; + } + let mut sorted: Vec<&RoomForUser> = rooms + .iter() + .filter(|room| self.cached_positions.contains_key(&room.room_id)) + .copied() + .collect(); + sorted.sort_by_key(|room| std::cmp::Reverse(self.cached_positions[&room.room_id])); + Some(sorted) + } + + /// The joined rooms whose last event position the full sort needs. + pub fn joined_rooms(rooms: &[&RoomForUser]) -> Vec> { + rooms + .iter() + .filter(|room| &*room.membership == "join") + .map(|room| Arc::clone(&room.room_id)) + .collect() + } + + /// Sort every room, given `joined_positions`, the last event position + /// before the `to_token` of each joined room. Fails for a joined room + /// without one, where Python raises `KeyError`. + pub fn sort_all<'a>( + mut self, + rooms: &[&'a RoomForUser], + joined_positions: &HashMap, i64>, + ) -> anyhow::Result> { + self.last_activity.extend( + joined_positions + .iter() + .map(|(room_id, position)| (Arc::clone(room_id), *position)), + ); + let mut keyed = Vec::with_capacity(rooms.len()); + for room in rooms { + let position = self + .last_activity + .get(&room.room_id) + .copied() + .ok_or_else(|| anyhow!("no last event position for {}", room.room_id))?; + keyed.push((position, *room)); + } + keyed.sort_by_key(|(position, _)| std::cmp::Reverse(*position)); + Ok(keyed.into_iter().map(|(_, room)| room).collect()) + } +} diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index f4d1f484e7..3e8dfe0d36 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -21,6 +21,7 @@ from typing import ( Literal, Mapping, MutableMapping, + Sequence, cast, ) @@ -215,6 +216,57 @@ def filter_membership_for_sync( ) +def _sync_list_ops( + ranges: Sequence[tuple[int, int]], + sorted_room_ids: Sequence[str], + room_sync_config: RoomSyncConfig, + relevant_room_map: dict[str, RoomSyncConfig], +) -> list[SlidingSyncResult.SlidingWindowList.Operation]: + """The `SYNC` operations of a list's `ranges` over its sorted rooms. + Adds each room in a range to `relevant_room_map` with the list's + `room_sync_config`, combined with the room's existing config. + + Note: once a room combines with an existing config, the rooms after it + in the list take the combined config too. + """ + ops: list[SlidingSyncResult.SlidingWindowList.Operation] = [] + for range in ranges: + room_ids_in_list: list[str] = [] + + # We're going to loop through the sorted list of rooms starting + # at the range start index and keep adding rooms until we fill + # up the range or run out of rooms. + # + # Both sides of range are inclusive so we `+ 1` + max_num_rooms = range[1] - range[0] + 1 + for room_id in sorted_room_ids[range[0] :]: + if len(room_ids_in_list) >= max_num_rooms: + break + + # Take the superset of the `RoomSyncConfig` for each room. + # + # Update our `relevant_room_map` with the room we're going + # to display and need to fetch more info about. + existing_room_sync_config = relevant_room_map.get(room_id) + if existing_room_sync_config is not None: + room_sync_config = existing_room_sync_config.combine_room_sync_config( + room_sync_config + ) + + relevant_room_map[room_id] = room_sync_config + + room_ids_in_list.append(room_id) + + ops.append( + SlidingSyncResult.SlidingWindowList.Operation( + op=OperationType.SYNC, + range=range, + room_ids=room_ids_in_list, + ) + ) + return ops + + class SlidingSyncRoomLists: """Handles calculating the room lists from sliding sync requests""" @@ -547,44 +599,12 @@ class SlidingSyncRoomLists: limit=max(range[1] + 1 for range in list_config.ranges), ) - for range in list_config.ranges: - room_ids_in_list: list[str] = [] - - # We're going to loop through the sorted list of rooms starting - # at the range start index and keep adding rooms until we fill - # up the range or run out of rooms. - # - # Both sides of range are inclusive so we `+ 1` - max_num_rooms = range[1] - range[0] + 1 - for room_membership in sorted_room_info[range[0] :]: - room_id = room_membership.room_id - - if len(room_ids_in_list) >= max_num_rooms: - break - - # Take the superset of the `RoomSyncConfig` for each room. - # - # Update our `relevant_room_map` with the room we're going - # to display and need to fetch more info about. - existing_room_sync_config = relevant_room_map.get( - room_id - ) - if existing_room_sync_config is not None: - room_sync_config = existing_room_sync_config.combine_room_sync_config( - room_sync_config - ) - - relevant_room_map[room_id] = room_sync_config - - room_ids_in_list.append(room_id) - - ops.append( - SlidingSyncResult.SlidingWindowList.Operation( - op=OperationType.SYNC, - range=range, - room_ids=room_ids_in_list, - ) - ) + ops = _sync_list_ops( + list_config.ranges, + [room.room_id for room in sorted_room_info], + room_sync_config, + relevant_room_map, + ) lists[list_key] = SlidingSyncResult.SlidingWindowList( count=len(filtered_sync_room_map), @@ -735,44 +755,12 @@ class SlidingSyncRoomLists: ops: list[SlidingSyncResult.SlidingWindowList.Operation] = [] if list_config.ranges: - for range in list_config.ranges: - room_ids_in_list: list[str] = [] - - # We're going to loop through the sorted list of rooms starting - # at the range start index and keep adding rooms until we fill - # up the range or run out of rooms. - # - # Both sides of range are inclusive so we `+ 1` - max_num_rooms = range[1] - range[0] + 1 - for room_membership in sorted_room_info[range[0] :]: - room_id = room_membership.room_id - - if len(room_ids_in_list) >= max_num_rooms: - break - - # Take the superset of the `RoomSyncConfig` for each room. - # - # Update our `relevant_room_map` with the room we're going - # to display and need to fetch more info about. - existing_room_sync_config = relevant_room_map.get( - room_id - ) - if existing_room_sync_config is not None: - room_sync_config = existing_room_sync_config.combine_room_sync_config( - room_sync_config - ) - - relevant_room_map[room_id] = room_sync_config - - room_ids_in_list.append(room_id) - - ops.append( - SlidingSyncResult.SlidingWindowList.Operation( - op=OperationType.SYNC, - range=range, - room_ids=room_ids_in_list, - ) - ) + ops = _sync_list_ops( + list_config.ranges, + [room.room_id for room in sorted_room_info], + room_sync_config, + relevant_room_map, + ) lists[list_key] = SlidingSyncResult.SlidingWindowList( count=len(sorted_room_info), diff --git a/synapse/synapse_rust/sliding_sync.pyi b/synapse/synapse_rust/sliding_sync.pyi index 1dee93c902..6a274f297d 100644 --- a/synapse/synapse_rust/sliding_sync.pyi +++ b/synapse/synapse_rust/sliding_sync.pyi @@ -182,3 +182,33 @@ def filter_rooms_for_test( sent_before: set[str], ) -> list[str]: """`filters` is JSON. Raises `NotImplementedError` as Python does.""" + +def sort_rooms_for_test( + rooms: list[_PyRoom], + earliest_cache_position: int, + cache: dict[str, int], + to_stream: int, + limit: int | None, + recheck: dict[str, int | None], + joined_positions: dict[str, int], +) -> list[str]: + """Raises `KeyError` as Python does.""" + +def sync_list_ops_for_test( + ranges: list[tuple[int, int]], + sorted_room_ids: list[str], + room_sync_config: tuple[int, dict[str, list[str]]], + relevant_room_map: list[tuple[str, tuple[int, dict[str, list[str]]]]], +) -> tuple[ + list[tuple[tuple[int, int], list[str]]], + list[tuple[str, tuple[int, dict[str, list[str]]]]], +]: ... +def rooms_to_send_for_test( + relevant_room_map: list[tuple[str, tuple[int, dict[str, list[str]]]]], + room_configs: dict[str, tuple[int, dict[str, list[str]]]], + statuses: dict[str, tuple[str, int | None]], + last_used_ts: int | None, + rooms_that_have_updates: set[str], + now_ms: int, +) -> list[str]: + """Raises `ValueError` where Python raises `SlidingSyncUnknownPosition`.""" diff --git a/tests/handlers/test_sliding_sync_rust_pure.py b/tests/handlers/test_sliding_sync_rust_pure.py index fa418b1042..6d121cbbd4 100644 --- a/tests/handlers/test_sliding_sync_rust_pure.py +++ b/tests/handlers/test_sliding_sync_rust_pure.py @@ -22,6 +22,7 @@ from types import SimpleNamespace from typing import AbstractSet, Any, Coroutine, Mapping, TypeVar, cast from synapse.api.constants import EventTypes, Membership +from synapse.api.errors import SlidingSyncUnknownPosition from synapse.events import EventBase from synapse.handlers.sliding_sync import ( MAX_NUMBER_PREVIOUS_STATE_KEYS_TO_REMEMBER, @@ -31,11 +32,17 @@ from synapse.handlers.sliding_sync import ( ) from synapse.handlers.sliding_sync.room_lists import ( SlidingSyncRoomLists, + _sync_list_ops, filter_membership_for_sync, ) from synapse.storage.roommember import RoomsForUserSlidingSync from synapse.synapse_rust import sliding_sync as rust -from synapse.types import PersistedEventPosition, RoomStreamToken, StreamToken +from synapse.types import ( + PersistedEventPosition, + RoomStreamToken, + StrCollection, + StreamToken, +) from synapse.types.handlers.sliding_sync import ( HaveSentRoom, HaveSentRoomFlag, @@ -760,3 +767,228 @@ class RoomListFilterTestCase(unittest.TestCase): result = None self.assertEqual(result, python, (rooms, filters)) self.assertGreater(not_implemented, 0) + + +class _FakeStreamCache: + def __init__(self, earliest: int, positions: dict[str, int]) -> None: + self.earliest = earliest + self.positions = positions + + def get_earliest_known_position(self) -> int: + return self.earliest + + def get_max_pos_of_last_change(self, room_id: str) -> int | None: + return self.positions.get(room_id) + + +class _FakeSortStore: + def __init__( + self, + stream_cache: _FakeStreamCache, + recheck: dict[str, int | None], + joined_positions: dict[str, int], + ) -> None: + self._events_stream_cache = stream_cache + self.recheck = recheck + self.joined_positions = joined_positions + + async def get_last_event_pos_in_room_before_stream_ordering( + self, room_id: str, end_token: RoomStreamToken + ) -> tuple[str, PersistedEventPosition] | None: + stream = self.recheck[room_id] + if stream is None: + return None + return ("$event", PersistedEventPosition("master", stream)) + + async def bulk_get_last_event_pos_in_room_before_stream_ordering( + self, room_ids: list[str], end_token: RoomStreamToken + ) -> dict[str, int]: + return { + room_id: self.joined_positions[room_id] + for room_id in room_ids + if room_id in self.joined_positions + } + + +class _FakeUpdatesStore: + def __init__(self, rooms_with_updates: set[str]) -> None: + self.rooms_with_updates = rooms_with_updates + + async def get_rooms_that_have_updates_since_sliding_sync_table( + self, room_ids: StrCollection, from_key: RoomStreamToken + ) -> set[str]: + return {room_id for room_id in room_ids if room_id in self.rooms_with_updates} + + +class RoomListSortTestCase(unittest.TestCase): + """`rust/src/sliding_sync/room_lists/sort.rs` and `lists.rs`""" + + def test_sort_rooms(self) -> None: + rng = random.Random(11) + early = 0 + for _ in range(2000): + rooms = [_random_room(rng, room_id) for room_id in _ROOMS] + # Joined rooms are the common case. + rooms = [ + (r[0], r[1], Membership.JOIN, *r[3:]) if rng.random() < 0.6 else r + for r in rooms + ] + joined = [r[0] for r in rooms if r[2] == Membership.JOIN] + earliest = rng.randint(0, 50) + to_stream = rng.randint(40, 100) + cache = { + room_id: rng.randint(earliest + 1, 120) + for room_id in joined + if rng.random() < 0.7 + } + recheck = { + room_id: rng.choice([None, rng.randint(1, to_stream)]) + for room_id in _ROOMS + } + joined_positions = { + room_id: rng.randint(1, to_stream) + for room_id in joined + if rng.random() < 0.98 + } + limit = rng.choice([None, 1, 2, 3, 5, 8]) + + store = _FakeSortStore( + _FakeStreamCache(earliest, cache), recheck, joined_positions + ) + fake_room_lists = cast(SlidingSyncRoomLists, SimpleNamespace(store=store)) + to_token = cast( + StreamToken, + SimpleNamespace(room_key=RoomStreamToken(stream=to_stream)), + ) + try: + python: list[str] | None = [ + room.room_id + for room in _run( + SlidingSyncRoomLists.sort_rooms( + fake_room_lists, + {room[0]: _python_room(room) for room in rooms}, + to_token, + limit=limit, + ) + ) + ] + except KeyError: + python = None + try: + result: list[str] | None = rust.sort_rooms_for_test( + rooms, earliest, cache, to_stream, limit, recheck, joined_positions + ) + except KeyError: + result = None + self.assertEqual(result, python, (rooms, cache, limit)) + if python is not None and len(python) < len(rooms): + early += 1 + self.assertGreater(early, 0) + + def test_sync_list_ops(self) -> None: + rng = random.Random(12) + for _ in range(2000): + sorted_room_ids = rng.sample(_ROOMS, rng.randint(0, len(_ROOMS))) + ranges = [] + for _ in range(rng.randint(1, 3)): + start = rng.randint(0, 9) + # Pydantic allows an end before the start, but not below 0. + ranges.append((start, max(0, start + rng.randint(-1, 4)))) + config = _python_config(rng.randint(0, 10), _random_required_state(rng)) + relevant = { + room_id: _python_config(rng.randint(0, 10), _random_required_state(rng)) + for room_id in rng.sample(_ROOMS, rng.randint(0, 3)) + } + rust_relevant = [(k, _to_rust(v)) for k, v in relevant.items()] + + ops = _sync_list_ops(ranges, sorted_room_ids, config, relevant) + rust_ops, rust_map = rust.sync_list_ops_for_test( + ranges, sorted_room_ids, _to_rust(config), rust_relevant + ) + context = (ranges, sorted_room_ids, config, rust_relevant) + self.assertEqual( + rust_ops, + [(tuple(op.range), list(op.room_ids)) for op in ops], + context, + ) + self.assertEqual( + [(k, (limit, _as_sets(m))) for k, (limit, m) in rust_map], + [ + (k, (v.timeline_limit, _as_sets(v.required_state_map))) + for k, v in relevant.items() + ], + context, + ) + + def test_rooms_to_send(self) -> None: + rng = random.Random(13) + now_ms = 10 * _HOUR_MS + reset = 0 + for _ in range(1000): + many = rng.random() < 0.2 + room_ids = [f"!many{i}:test" for i in range(130)] if many else list(_ROOMS) + relevant = { + room_id: _python_config(rng.randint(0, 5), _random_required_state(rng)) + for room_id in room_ids + if rng.random() < 0.8 + } + room_configs = { + room_id: _python_config(rng.randint(0, 5), _random_required_state(rng)) + for room_id in room_ids + if rng.random() < 0.5 + } + statuses = _random_statuses(rng) + if many: + statuses = dict.fromkeys(room_ids, ("live", None)) + rooms_with_updates = { + room_id for room_id in room_ids if rng.random() < (0.9 if many else 0.3) + } + last_used_ts = rng.choice([None, now_ms - 2 * _HOUR_MS, now_ms - 60_000]) + + previous_state = PerConnectionState( + last_used_ts=last_used_ts, + rooms=RoomStatusMap( + statuses={ + k: HaveSentRoom( + HaveSentRoomFlag(flag), + RoomStreamToken(stream=token) if token else None, + ) + for k, (flag, token) in statuses.items() + } + ), + room_configs=room_configs, + ) + fake_room_lists = cast( + SlidingSyncRoomLists, + SimpleNamespace( + store=_FakeUpdatesStore(rooms_with_updates), + _clock=_FakeClock(now_ms), + ), + ) + from_token = cast( + StreamToken, SimpleNamespace(room_key=RoomStreamToken(stream=1)) + ) + try: + python: list[str] | None = list( + _run( + SlidingSyncRoomLists._filter_relevant_rooms_to_send( + fake_room_lists, previous_state, from_token, relevant + ) + ) + ) + except SlidingSyncUnknownPosition: + python = None + reset += 1 + try: + result: list[str] | None = rust.rooms_to_send_for_test( + [(k, _to_rust(v)) for k, v in relevant.items()], + {k: _to_rust(v) for k, v in room_configs.items()}, + statuses, + last_used_ts, + {room_id for room_id in relevant if room_id in rooms_with_updates}, + now_ms, + ) + except ValueError: + result = None + self.assertEqual(result, python) + self.assertGreater(reset, 0)