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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG
This commit is contained in:
Erik JohnstonandClaude Opus 5.5 committed 2026-10-16 11:35:02 +00:00
1 parent c2117d3b16
commit a3fe79ed8c
8 files changed
+856 -78

No files matched your search

Generated
+1
View File
@@ -2028,6 +2028,7 @@ dependencies = [
"hyper",
"hyper-util",
"icu_segmenter",
"indexmap",
"ipnet",
"itertools",
"lazy_static",
+2
View File
@@ -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"
+215
View File
@@ -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:
* <https://www.gnu.org/licenses/agpl-3.0.html>.
*/
//! 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<Arc<str>, RoomSyncConfig>;
/// `SlidingWindowList.Operation`: always a `SYNC`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SyncOp {
pub range: (u64, u64),
pub room_ids: Vec<Arc<str>>,
}
/// 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<str>],
room_sync_config: &RoomSyncConfig,
relevant_room_map: &mut RelevantRoomMap,
) -> Vec<SyncOp> {
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<str>,
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<R: Clone>(
relevant_room_map: &RelevantRoomMap,
previous_state: &PerConnectionState<R>,
rooms_that_have_updates: &HashSet<Arc<str>>,
now_ms: i64,
) -> Result<Vec<Arc<str>>, UnknownPosition> {
let mut rooms_should_send: HashSet<&Arc<str>> = 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<Arc<str>> = ["!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);
}
}
+142 -1
View File
@@ -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<String, Vec<String>>);
/// `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<PyRoomForUser>,
earliest_cache_position: i64,
cache: HashMap<String, i64>,
to_stream: i64,
limit: Option<usize>,
recheck: HashMap<String, Option<i64>>,
joined_positions: HashMap<String, i64>,
) -> PyResult<Vec<String>> {
let rooms: Vec<RoomForUser> = 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<Arc<str>, 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<String>,
room_sync_config: PyConfig,
relevant_room_map: Vec<(String, PyConfig)>,
) -> (Vec<((u64, u64), Vec<String>)>, 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<Arc<str>> = 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<String, PyConfig>,
statuses: HashMap<String, (String, Option<i64>)>,
last_used_ts: Option<i64>,
rooms_that_have_updates: HashSet<String>,
now_ms: i64,
) -> PyResult<Vec<String>> {
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<String, RoomSyncConfig> = room_configs
.into_iter()
.map(|(room_id, (limit, map))| (room_id, config_from_py(limit, map)))
.collect();
let state: PerConnectionState<i64> = PerConnectionState {
last_used_ts,
rooms: RoomStatusMap::new(rooms),
room_configs: Arc::new(room_configs),
..Default::default()
};
let updates: HashSet<Arc<str>> = 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(())
}
+169
View File
@@ -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:
* <https://www.gnu.org/licenses/agpl-3.0.html>.
*/
//! `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<Arc<str>, i64>,
/// The positions known to be in the stream change cache's range.
cached_positions: HashMap<Arc<str>, 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<i64>,
) -> 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<usize>) -> 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<usize>, to_stream: i64) -> Vec<Arc<str>> {
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<i64>) {
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<usize>,
) -> Option<Vec<&'a RoomForUser>> {
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<Arc<str>> {
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<Arc<str>, i64>,
) -> anyhow::Result<Vec<&'a RoomForUser>> {
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())
}
}
+64 -76
View File
@@ -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),
+30
View File
@@ -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`."""
+233 -1
View File
@@ -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)