diff --git a/rust/src/sync/mod.rs b/rust/src/sync/mod.rs index 55449d0728..6512099809 100644 --- a/rust/src/sync/mod.rs +++ b/rust/src/sync/mod.rs @@ -22,7 +22,8 @@ //! //! The pure parts of the per-room loop are each a port of the Python they //! are named after, checked against it by the tests in `tests/sync/`: -//! [`filter`] ports `synapse.api.filtering`. +//! [`filter`] ports `synapse.api.filtering`, and [`state_delta`] the state +//! delta arithmetic of the sync handler. //! [`crate::visibility`] and [`crate::types::stream_token`] are shared with //! later ports. @@ -34,6 +35,7 @@ use crate::runtime::RustRuntime; use crate::storage::store::Store; pub mod filter; +pub mod state_delta; create_exception!( synapse.synapse_rust.sync, @@ -74,6 +76,7 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> let child_module = PyModule::new(py, "sync")?; child_module.add_class::()?; child_module.add_class::()?; + state_delta::register_test_functions(py, &child_module)?; child_module.add("RustSyncFallback", py.get_type::())?; m.add_submodule(&child_module)?; diff --git a/rust/src/sync/state_delta.rs b/rust/src/sync/state_delta.rs new file mode 100644 index 0000000000..a4c49b05e5 --- /dev/null +++ b/rust/src/sync/state_delta.rs @@ -0,0 +1,429 @@ +/* + * 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 state delta arithmetic of `synapse.handlers.sync`: what state to +//! send for a room, given the state at a few points and what the timeline +//! holds. The database reads that produce those inputs are the store's. + +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; + +use pyo3::{ + prelude::*, + types::{PyDict, PyModule, PyModuleMethods}, +}; + +use crate::events::constants::event_type::M_ROOM_MEMBER; +use crate::state::{StateKey, StateMap}; + +/// `_calculate_state`: the state to include in a sync response. +/// +/// `timeline_contains` is the state in the timeline, `timeline_start` and +/// `timeline_end` the state at its ends, and `previous_timeline_end` the +/// state at the end of the previous sync (empty for an initial sync). With +/// `lazy_load_members`, `timeline_start` has been filtered to the members +/// the client needs, and those are sent even if the previous sync had them. +/// +/// The Python docstring explains why the state at the end of the timeline +/// takes part: a fork in the DAG can put a state event the client needs +/// outside the state at the timeline's start. +/// +/// One difference from the Python: when the start and end of the timeline +/// disagree on a key and neither event is in the timeline or the previous +/// state, the Python sends whichever its set iteration happens to visit +/// last. This sends the one from the end of the timeline, which is the +/// state the client should end up with. +pub fn calculate_state( + timeline_contains: &StateMap, + timeline_start: &StateMap, + timeline_end: &StateMap, + previous_timeline_end: &StateMap, + lazy_load_members: bool, +) -> StateMap { + let mut event_id_to_state_key: HashMap<&Arc, &StateKey> = HashMap::new(); + for map in [ + timeline_contains, + timeline_start, + timeline_end, + previous_timeline_end, + ] { + for (key, event_id) in map { + event_id_to_state_key.insert(event_id, key); + } + } + + let timeline_end_ids: HashSet<&Arc> = timeline_end.values().collect(); + let timeline_start_ids: HashSet<&Arc> = timeline_start.values().collect(); + let mut previous_timeline_end_ids: HashSet<&Arc> = + previous_timeline_end.values().collect(); + let timeline_contains_ids: HashSet<&Arc> = timeline_contains.values().collect(); + + // With lazy loading, the memberships in `timeline_start` are the ones + // the timeline's senders need, and they may never have been sent: take + // them out of what the previous sync is assumed to have covered. + if lazy_load_members { + for (key, event_id) in timeline_start { + if &*key.0 == M_ROOM_MEMBER { + previous_timeline_end_ids.remove(event_id); + } + } + } + + // The start's entries first, so that the end's overwrite them. + timeline_start_ids + .iter() + .chain(&timeline_end_ids) + .filter(|id| !previous_timeline_end_ids.contains(*id)) + .filter(|id| !timeline_contains_ids.contains(*id)) + .map(|id| ((*event_id_to_state_key[*id]).clone(), Arc::clone(id))) + .collect() +} + +/// A row of `current_state_delta_stream`, in stream order. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StateDelta { + pub event_type: Arc, + pub state_key: Arc, + /// `None` when the key was removed from the current state. + pub event_id: Option>, + pub prev_event_id: Option>, +} + +impl StateDelta { + fn key(&self) -> StateKey { + StateKey(Arc::clone(&self.event_type), Arc::clone(&self.state_key)) + } +} + +/// Roll `state` back over `deltas`, which run forwards from the point +/// `state` is wanted at to the point it was read at. The deltas are walked +/// backwards so that the earliest change to a key is what sticks. +/// +/// This is the `use_state_after` loop in `_compute_state_delta_for_full_ +/// sync`, which reads the current state and rolls it back to the end of the +/// timeline. +pub fn roll_back_state(state: &mut StateMap, deltas: &[StateDelta]) { + for delta in deltas.iter().rev() { + match &delta.prev_event_id { + Some(prev_event_id) => { + state.insert(delta.key(), Arc::clone(prev_event_id)); + } + None => { + state.remove(&delta.key()); + } + } + } +} + +/// What applying the deltas since the last sync gives, for the +/// `use_state_after` incremental path. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct AppliedDeltas { + /// The state to send, on top of `initial`. + pub state: StateMap, + /// The keys in `state` that the deltas changed. + pub changed: HashSet, + /// Keys removed from the current state that the caller has to look up + /// in state groups at the end of the timeline: the keys the timeline + /// changed, and the user's own membership. See the Python comment on + /// why a server leaving a room produces these. + pub cleared: HashSet, +} + +/// The delta loop of `_compute_state_delta_for_incremental_sync` with +/// `use_state_after`: apply `deltas` (in stream order, so the latest wins) +/// to `initial`, the lazy-loaded memberships fetched first. +pub fn apply_state_deltas( + initial: StateMap, + deltas: &[StateDelta], + timeline_state: &StateMap, + user_id: &str, +) -> AppliedDeltas { + let mut applied = AppliedDeltas { + state: initial, + ..AppliedDeltas::default() + }; + let own_membership = StateKey::new(M_ROOM_MEMBER, user_id); + for delta in deltas { + let key = delta.key(); + match &delta.event_id { + None => { + if timeline_state.contains_key(&key) || key == own_membership { + applied.cleared.insert(key); + } + } + Some(event_id) => { + applied.state.insert(key.clone(), Arc::clone(event_id)); + applied.changed.insert(key.clone()); + applied.cleared.remove(&key); + } + } + } + applied +} + +// Entry points for the differential tests. + +fn state_map_from_py(dict: &Bound<'_, PyDict>) -> PyResult { + let mut map = StateMap::new(); + for (key, event_id) in dict.iter() { + let (event_type, state_key): (String, String) = key.extract()?; + let event_id: String = event_id.extract()?; + map.insert(StateKey::new(&event_type, &state_key), Arc::from(event_id)); + } + Ok(map) +} + +fn state_map_to_py<'py>(py: Python<'py>, map: &StateMap) -> PyResult> { + let dict = PyDict::new(py); + for (key, event_id) in map { + dict.set_item((&*key.0, &*key.1), &**event_id)?; + } + Ok(dict) +} + +fn keys_to_py(keys: &HashSet) -> HashSet<(String, String)> { + keys.iter() + .map(|key| (key.0.to_string(), key.1.to_string())) + .collect() +} + +type PyDelta = (String, String, Option, Option); + +fn deltas_from_py(deltas: Vec) -> Vec { + deltas + .into_iter() + .map( + |(event_type, state_key, event_id, prev_event_id)| StateDelta { + event_type: Arc::from(event_type), + state_key: Arc::from(state_key), + event_id: event_id.map(Arc::from), + prev_event_id: prev_event_id.map(Arc::from), + }, + ) + .collect() +} + +/// `_calculate_state` through the Rust implementation. +#[pyfunction] +fn calculate_state_for_test<'py>( + py: Python<'py>, + timeline_contains: &Bound<'py, PyDict>, + timeline_start: &Bound<'py, PyDict>, + timeline_end: &Bound<'py, PyDict>, + previous_timeline_end: &Bound<'py, PyDict>, + lazy_load_members: bool, +) -> PyResult> { + let state = calculate_state( + &state_map_from_py(timeline_contains)?, + &state_map_from_py(timeline_start)?, + &state_map_from_py(timeline_end)?, + &state_map_from_py(previous_timeline_end)?, + lazy_load_members, + ); + state_map_to_py(py, &state) +} + +/// The full-sync `use_state_after` roll-back through the Rust +/// implementation. Deltas are `(type, state_key, event_id, prev_event_id)` +/// in stream order. +#[pyfunction] +fn roll_back_state_for_test<'py>( + py: Python<'py>, + state: &Bound<'py, PyDict>, + deltas: Vec, +) -> PyResult> { + let mut state = state_map_from_py(state)?; + roll_back_state(&mut state, &deltas_from_py(deltas)); + state_map_to_py(py, &state) +} + +/// The incremental `use_state_after` delta loop through the Rust +/// implementation: `(state, changed_keys, cleared_keys)`. +#[pyfunction] +#[allow(clippy::type_complexity)] +fn apply_state_deltas_for_test<'py>( + py: Python<'py>, + initial: &Bound<'py, PyDict>, + deltas: Vec, + timeline_state: &Bound<'py, PyDict>, + user_id: &str, +) -> PyResult<( + Bound<'py, PyDict>, + HashSet<(String, String)>, + HashSet<(String, String)>, +)> { + let applied = apply_state_deltas( + state_map_from_py(initial)?, + &deltas_from_py(deltas), + &state_map_from_py(timeline_state)?, + user_id, + ); + Ok(( + state_map_to_py(py, &applied.state)?, + keys_to_py(&applied.changed), + keys_to_py(&applied.cleared), + )) +} + +pub fn register_test_functions(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_function(wrap_pyfunction!(calculate_state_for_test, m)?)?; + m.add_function(wrap_pyfunction!(roll_back_state_for_test, m)?)?; + m.add_function(wrap_pyfunction!(apply_state_deltas_for_test, m)?)?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn state(entries: &[(&str, &str, &str)]) -> StateMap { + entries + .iter() + .map(|(t, k, e)| (StateKey::new(t, k), Arc::from(*e))) + .collect() + } + + fn delta(t: &str, k: &str, event_id: Option<&str>, prev: Option<&str>) -> StateDelta { + StateDelta { + event_type: Arc::from(t), + state_key: Arc::from(k), + event_id: event_id.map(Arc::from), + prev_event_id: prev.map(Arc::from), + } + } + + #[test] + fn state_from_a_fork_comes_from_the_timeline_end() { + // The docstring's example: S2 is in the state at the end of the + // timeline but not at its start, and not in the timeline. + let result = calculate_state( + &state(&[]), + &state(&[("m.room.create", "", "$e1")]), + &state(&[("m.room.create", "", "$e1"), ("m.room.topic", "", "$s2")]), + &state(&[("m.room.create", "", "$e1")]), + false, + ); + assert_eq!(result, state(&[("m.room.topic", "", "$s2")])); + } + + #[test] + fn lazy_loaded_members_are_sent_again() { + let member = ("m.room.member", "@a:x", "$m"); + let without = calculate_state( + &state(&[]), + &state(&[member]), + &state(&[member]), + &state(&[member]), + false, + ); + assert!(without.is_empty()); + let with = calculate_state( + &state(&[]), + &state(&[member]), + &state(&[member]), + &state(&[member]), + true, + ); + assert_eq!(with, state(&[member])); + // A non-member key the previous sync had is still left out. + let topic = ("m.room.topic", "", "$t"); + let with = calculate_state( + &state(&[]), + &state(&[topic]), + &state(&[topic]), + &state(&[topic]), + true, + ); + assert!(with.is_empty()); + } + + #[test] + fn timeline_state_is_not_repeated() { + let topic = ("m.room.topic", "", "$t"); + let result = calculate_state( + &state(&[topic]), + &state(&[]), + &state(&[topic]), + &state(&[]), + false, + ); + assert!(result.is_empty()); + } + + #[test] + fn rolling_back_keeps_the_earliest_change() { + let mut current = state(&[("m.room.topic", "", "$t3"), ("m.room.name", "", "$n")]); + roll_back_state( + &mut current, + &[ + delta("m.room.topic", "", Some("$t2"), Some("$t1")), + delta("m.room.name", "", Some("$n"), None), + delta("m.room.topic", "", Some("$t3"), Some("$t2")), + ], + ); + assert_eq!(current, state(&[("m.room.topic", "", "$t1")])); + } + + #[test] + fn applying_deltas_tracks_changed_and_cleared_keys() { + let applied = apply_state_deltas( + state(&[("m.room.member", "@b:x", "$mb")]), + &[ + delta("m.room.topic", "", Some("$t1"), None), + delta("m.room.topic", "", Some("$t2"), Some("$t1")), + // The server left: the timeline's keys and the user's + // membership are cleared, other keys are dropped. + delta("m.room.name", "", None, Some("$n")), + delta("m.room.member", "@me:x", None, Some("$me")), + delta("m.room.member", "@c:x", None, Some("$mc")), + ], + &state(&[("m.room.name", "", "$n2")]), + "@me:x", + ); + assert_eq!( + applied.state, + state(&[ + ("m.room.member", "@b:x", "$mb"), + ("m.room.topic", "", "$t2") + ]) + ); + assert_eq!( + applied.changed, + HashSet::from([StateKey::new("m.room.topic", "")]) + ); + assert_eq!( + applied.cleared, + HashSet::from([ + StateKey::new("m.room.name", ""), + StateKey::new("m.room.member", "@me:x"), + ]) + ); + + // A later delta with an event un-clears the key. + let applied = apply_state_deltas( + StateMap::new(), + &[ + delta("m.room.name", "", None, Some("$n")), + delta("m.room.name", "", Some("$n2"), None), + ], + &state(&[("m.room.name", "", "$n2")]), + "@me:x", + ); + assert!(applied.cleared.is_empty()); + assert_eq!(applied.state, state(&[("m.room.name", "", "$n2")])); + } +} diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index eca20b89bd..30358544ad 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -63,6 +63,7 @@ from synapse.logging.opentracing import ( from synapse.metrics import SERVER_NAME_LABEL from synapse.storage.databases.main.event_push_actions import RoomNotifCounts from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary +from synapse.storage.databases.main.state_deltas import StateDelta from synapse.storage.databases.main.stream import PaginateFunction from synapse.storage.invite_rule import InviteRule from synapse.storage.roommember import MemberSummary @@ -1558,20 +1559,7 @@ class SyncHandler: to_token=self.store.get_room_max_token(), ) if deltas: - mutable_state_ids = dict(state_ids) - - # We iterate over the deltas backwards so that if there are - # multiple changes of the same type/state_key we'll - # correctly pick the earliest delta. - for delta in reversed(deltas): - if delta.prev_event_id: - mutable_state_ids[(delta.event_type, delta.state_key)] = ( - delta.prev_event_id - ) - elif (delta.event_type, delta.state_key) in mutable_state_ids: - mutable_state_ids.pop((delta.event_type, delta.state_key)) - - state_ids = mutable_state_ids + state_ids = _roll_back_state_over_deltas(state_ids, deltas) return state_ids @@ -1721,46 +1709,13 @@ class SyncHandler: to_token=end_token.room_key, ) ) - # A delta with `event_id=None` means the key was removed from the - # current state. Two things cause this: - # - A state reset removed the key. With `/sync`, there is no way to tell the client a - # key was removed, so there is nothing to send for it. - # - The server left the room (its last local user left). It then - # copies the room state from **before the batch** that contains the - # leave into `current_state_delta_stream`, as deltas with - # `event_id=None`. No delta refers to the events in that batch: the - # leaving user's membership only gets an `event_id=None` delta, with - # their membership before the leave as `prev_event_id`. So a key - # that batch changed: - # - only has the `event_id=None` delta if it already existed, and - # any earlier delta for it is stale. That batch ends the - # timeline, so we look up the keys in the timeline in state - # groups, which still have the state at `end_token`. - # - has no delta at all if it is new. - # TODO: Recover it. - # Keys that only changed earlier keep their earlier delta, and - # other keys need no update. - cleared_state_keys: set[tuple[str, str]] = set() - for delta in deltas: - key = (delta.event_type, delta.state_key) - if delta.event_id is None: - # Look up the key if: - # - it is in the timeline (see above). - # - OR it is the user's own membership, in case a filter - # removed their leave from the timeline. - if key in timeline_state or key == ( - EventTypes.Member, - sync_config.user.to_string(), - ): - cleared_state_keys.add(key) - continue - - # Note that deltas are in stream ordering, so if there are - # multiple deltas for a given type/state_key we'll always pick - # the latest one. - delta_state_ids[key] = delta.event_id - changed_keys.add(key) - cleared_state_keys.discard(key) + cleared_state_keys = _apply_current_state_deltas( + delta_state_ids, + changed_keys, + deltas, + timeline_state, + sync_config.user.to_string(), + ) if cleared_state_keys: state_at_end = await self._state_storage_controller.get_state_ids_at( @@ -3646,6 +3601,76 @@ def _action_has_highlight(actions: list[JsonDict]) -> bool: return False +def _roll_back_state_over_deltas( + state_ids: StateMap[str], deltas: Sequence[StateDelta] +) -> StateMap[str]: + """Roll `state_ids` back over `deltas`, which run forwards from the point + the state is wanted at to the point it was read at. + + The deltas are walked backwards so that if there are multiple changes of + the same type/state_key we correctly pick the earliest delta. + """ + mutable_state_ids = dict(state_ids) + for delta in reversed(deltas): + if delta.prev_event_id: + mutable_state_ids[(delta.event_type, delta.state_key)] = delta.prev_event_id + elif (delta.event_type, delta.state_key) in mutable_state_ids: + mutable_state_ids.pop((delta.event_type, delta.state_key)) + return mutable_state_ids + + +def _apply_current_state_deltas( + delta_state_ids: MutableStateMap[str], + changed_keys: set[StateKey], + deltas: Sequence[StateDelta], + timeline_state: StateMap[str], + user_id: str, +) -> set[StateKey]: + """Apply `deltas`, in stream order, to `delta_state_ids`, recording the + keys they change in `changed_keys`. Returns the keys removed from the + current state that the caller has to look up at the end of the timeline. + + A delta with `event_id=None` means the key was removed from the + current state. Two things cause this: + - A state reset removed the key. With `/sync`, there is no way to tell the client a + key was removed, so there is nothing to send for it. + - The server left the room (its last local user left). It then + copies the room state from **before the batch** that contains the + leave into `current_state_delta_stream`, as deltas with + `event_id=None`. No delta refers to the events in that batch: the + leaving user's membership only gets an `event_id=None` delta, with + their membership before the leave as `prev_event_id`. So a key + that batch changed: + - only has the `event_id=None` delta if it already existed, and + any earlier delta for it is stale. That batch ends the + timeline, so we look up the keys in the timeline in state + groups, which still have the state at `end_token`. + - has no delta at all if it is new. + TODO: Recover it. + Keys that only changed earlier keep their earlier delta, and + other keys need no update. + """ + cleared_state_keys: set[StateKey] = set() + for delta in deltas: + key = (delta.event_type, delta.state_key) + if delta.event_id is None: + # Look up the key if: + # - it is in the timeline (see above). + # - OR it is the user's own membership, in case a filter + # removed their leave from the timeline. + if key in timeline_state or key == (EventTypes.Member, user_id): + cleared_state_keys.add(key) + continue + + # Note that deltas are in stream ordering, so if there are + # multiple deltas for a given type/state_key we'll always pick + # the latest one. + delta_state_ids[key] = delta.event_id + changed_keys.add(key) + cleared_state_keys.discard(key) + return cleared_state_keys + + def _calculate_state( timeline_contains: StateMap[str], timeline_start: StateMap[str], diff --git a/synapse/synapse_rust/sync.pyi b/synapse/synapse_rust/sync.pyi index 7e2b2a07ac..f1479f981b 100644 --- a/synapse/synapse_rust/sync.pyi +++ b/synapse/synapse_rust/sync.pyi @@ -10,6 +10,8 @@ # See the GNU Affero General Public License for more details: # . +from collections.abc import Mapping + from synapse.synapse_rust.events import Event from synapse.types import JsonDict, JsonMapping @@ -71,3 +73,30 @@ class FilterCollection: def check_room_account_data(self, event: JsonDict) -> bool: ... def check_presence(self, user_id: str) -> bool: ... def check_global_account_data(self, event: JsonDict) -> bool: ... + +# The Rust ports of the state delta arithmetic in `synapse.handlers.sync`. +# Deltas are `(event_type, state_key, event_id, prev_event_id)` in stream +# order. + +def calculate_state_for_test( + timeline_contains: Mapping[tuple[str, str], str], + timeline_start: Mapping[tuple[str, str], str], + timeline_end: Mapping[tuple[str, str], str], + previous_timeline_end: Mapping[tuple[str, str], str], + lazy_load_members: bool, +) -> dict[tuple[str, str], str]: + """`_calculate_state`.""" + +def roll_back_state_for_test( + state: Mapping[tuple[str, str], str], + deltas: list[tuple[str, str, str | None, str | None]], +) -> dict[tuple[str, str], str]: + """`_roll_back_state_over_deltas`.""" + +def apply_state_deltas_for_test( + initial: Mapping[tuple[str, str], str], + deltas: list[tuple[str, str, str | None, str | None]], + timeline_state: Mapping[tuple[str, str], str], + user_id: str, +) -> tuple[dict[tuple[str, str], str], set[tuple[str, str]], set[tuple[str, str]]]: + """`_apply_current_state_deltas`.""" diff --git a/tests/sync/test_rust_state_delta.py b/tests/sync/test_rust_state_delta.py new file mode 100644 index 0000000000..5533291e62 --- /dev/null +++ b/tests/sync/test_rust_state_delta.py @@ -0,0 +1,136 @@ +# +# 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 state delta arithmetic of `synapse.handlers.sync` against +`rust/src/sync/state_delta.rs`, on random state maps and deltas.""" + +import random + +from parameterized import parameterized + +from synapse.api.constants import EventTypes +from synapse.handlers.sync import ( + _apply_current_state_deltas, + _calculate_state, + _roll_back_state_over_deltas, +) +from synapse.storage.databases.main.state_deltas import StateDelta +from synapse.synapse_rust.sync import ( + apply_state_deltas_for_test, + calculate_state_for_test, + roll_back_state_for_test, +) +from synapse.types import ( + StateKey, + StateMap, +) + +from tests import unittest +from tests.sync.fixtures import ( + ME, + SEEDS, +) + +KEYS: list[StateKey] = [ + (EventTypes.Create, ""), + (EventTypes.Topic, ""), + (EventTypes.Name, ""), + (EventTypes.Member, ME), + (EventTypes.Member, "@alice:test"), + (EventTypes.Member, "@bob:test"), +] + + +def _random_state_map(rng: random.Random, event_ids: list[str]) -> StateMap[str]: + """A random state map over `KEYS`. Event IDs are shared across maps, as + they are between the state at different points of one room, so the + same event ID always carries the same key.""" + return { + key: event_ids[KEYS.index(key) * 3 + rng.randint(0, 2)] + for key in rng.sample(KEYS, rng.randint(0, len(KEYS))) + } + + +class StateDeltaTestCase(unittest.TestCase): + """The state delta arithmetic of `synapse.handlers.sync` against + `rust/src/sync/state_delta.rs`.""" + + EVENT_IDS = [f"${i}" for i in range(len(KEYS) * 3)] + + @parameterized.expand(SEEDS) + def test_random_calculate_state_agrees(self, seed: int) -> None: + rng = random.Random(seed) + for _ in range(100): + contains, start, end, previous = ( + _random_state_map(rng, self.EVENT_IDS) for _ in range(4) + ) + lazy = rng.choice([True, False]) + context = f"maps={(contains, start, end, previous)} lazy={lazy}" + expected = _calculate_state(contains, start, end, previous, lazy) + actual = calculate_state_for_test(contains, start, end, previous, lazy) + self.assertEqual(set(actual), set(expected), context) + for key, event_id in actual.items(): + if event_id == expected[key]: + continue + # The start and end of the timeline disagree on this key. + # Python sends whichever its set iteration visits last; the + # Rust sends the end's. + self.assertEqual(event_id, end[key], context) + self.assertEqual(expected[key], start[key], context) + + @staticmethod + def _random_deltas(rng: random.Random, event_ids: list[str]) -> list[StateDelta]: + deltas = [] + for i in range(rng.randint(0, 8)): + key = rng.choice(KEYS) + deltas.append( + StateDelta( + stream_id=i, + room_id="!a:test", + event_type=key[0], + state_key=key[1], + event_id=rng.choice([None, rng.choice(event_ids)]), + prev_event_id=rng.choice([None, rng.choice(event_ids)]), + ) + ) + return deltas + + @parameterized.expand(SEEDS) + def test_random_delta_loops_agree(self, seed: int) -> None: + rng = random.Random(seed) + for _ in range(100): + state = _random_state_map(rng, self.EVENT_IDS) + deltas = self._random_deltas(rng, self.EVENT_IDS) + rust_deltas = [ + (d.event_type, d.state_key, d.event_id, d.prev_event_id) for d in deltas + ] + context = f"state={state} deltas={rust_deltas}" + self.assertEqual( + roll_back_state_for_test(state, rust_deltas), + _roll_back_state_over_deltas(state, deltas), + context, + ) + + timeline_state = _random_state_map(rng, self.EVENT_IDS) + initial = _random_state_map(rng, self.EVENT_IDS) + expected_state = dict(initial) + expected_changed: set[StateKey] = set() + expected_cleared = _apply_current_state_deltas( + expected_state, expected_changed, deltas, timeline_state, ME + ) + self.assertEqual( + apply_state_deltas_for_test(initial, rust_deltas, timeline_state, ME), + (expected_state, expected_changed, expected_cleared), + context, + )