Port the sync state delta arithmetic to Rust

`_calculate_state`, the `use_state_after` roll-back over state deltas of
the full-state path, and the delta loop of the incremental path. The
database reads that produce their inputs stay with the store.

The two Python loops were inline in the handler; they are now functions,
`_roll_back_state_over_deltas` and `_apply_current_state_deltas`, so that
`tests/sync/test_rust_state_delta.py` can compare them with the Rust on
random state maps and deltas. Behaviour is unchanged.

One deliberate difference: when the start and end of the timeline
disagree on a key that is in neither the timeline nor the previous state,
`_calculate_state` sends whichever event its set iteration visits last.
The Rust sends the one from the end of the timeline, the state the client
should end up with. The test allows the Python either.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0194eNQP5weBtVMg8NKtc6VU

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::<SyncHandler>()?;
     child_module.add_class::<filter::FilterCollection>()?;
+    state_delta::register_test_functions(py, &child_module)?;
     child_module.add("RustSyncFallback", py.get_type::<RustSyncFallback>())?;
 
     m.add_submodule(&child_module)?;

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ */
+
+//! 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<str>, &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<str>> = timeline_end.values().collect();
+    let timeline_start_ids: HashSet<&Arc<str>> = timeline_start.values().collect();
+    let mut previous_timeline_end_ids: HashSet<&Arc<str>> =
+        previous_timeline_end.values().collect();
+    let timeline_contains_ids: HashSet<&Arc<str>> = 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<str>,
+    pub state_key: Arc<str>,
+    /// `None` when the key was removed from the current state.
+    pub event_id: Option<Arc<str>>,
+    pub prev_event_id: Option<Arc<str>>,
+}
+
+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<StateKey>,
+    /// 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<StateKey>,
+}
+
+/// 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<StateMap> {
+    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<Bound<'py, PyDict>> {
+    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<StateKey>) -> HashSet<(String, String)> {
+    keys.iter()
+        .map(|key| (key.0.to_string(), key.1.to_string()))
+        .collect()
+}
+
+type PyDelta = (String, String, Option<String>, Option<String>);
+
+fn deltas_from_py(deltas: Vec<PyDelta>) -> Vec<StateDelta> {
+    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<Bound<'py, PyDict>> {
+    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<PyDelta>,
+) -> PyResult<Bound<'py, PyDict>> {
+    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<PyDelta>,
+    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")]));
+    }
+}

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],

synapse/synapse_rust/sync.pyi

@@ -10,6 +10,8 @@
 # See the GNU Affero General Public License for more details:
 # <https://www.gnu.org/licenses/agpl-3.0.html>.
 
+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`."""

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+
+"""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,
+            )