mirror of
https://github.com/element-hq/synapse.git
synced 2026-11-04 13:28:04 +00:00
Compute and encode sliding sync room entries in Rust
SlidingSyncHandler.generate_room_entries runs get_room_sync_data for a batch of rooms, twenty at a time, and returns each room's entry with the two changes it makes to the connection state. encode_rooms is the servlet's encode_rooms for those entries. A room's state loads reuse the events its timeline loaded, as Python's held-event map does, so state events carry the same prev_content. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01L51DoxB8VnH8UjXJFDgpNG
This commit is contained in:
1 parent
4b3bbecddd
commit
2a0d452c10
4 files changed
+1555
-1
No files matched your search
@@ -0,0 +1,187 @@
|
||||
/*
|
||||
* 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 servlet's `encode_rooms` for the rooms Rust computed, so that their
|
||||
//! events never become Python objects.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use pyo3::prelude::*;
|
||||
use pythonize::pythonize;
|
||||
use serde_json::{json, Map, Value};
|
||||
|
||||
use crate::events::json_object::JsonObject;
|
||||
use crate::events::serialize::{
|
||||
serialize_event, BundleRef, SerializeEventConfig, SerializeLookups,
|
||||
};
|
||||
use crate::events::Event;
|
||||
use crate::sliding_sync::result::RoomResult;
|
||||
use crate::sliding_sync::room_entry::RoomEntry;
|
||||
use crate::sync::relations::Bundle;
|
||||
|
||||
/// The lookups of one room: the bundles and redactions Rust loaded for it,
|
||||
/// and the module callbacks' unsigned additions.
|
||||
struct EntryLookups<'a> {
|
||||
entry: &'a RoomEntry,
|
||||
unsigned_additions: &'a HashMap<String, JsonObject>,
|
||||
}
|
||||
|
||||
impl SerializeLookups for EntryLookups<'_> {
|
||||
fn redaction(&self, event_id: &str) -> Option<&Event> {
|
||||
self.entry.redactions.get(event_id)
|
||||
}
|
||||
|
||||
fn bundle(&self, event_id: &str) -> Option<BundleRef<'_>> {
|
||||
self.entry
|
||||
.bundles
|
||||
.as_ref()?
|
||||
.get(event_id)
|
||||
.map(Bundle::as_ref)
|
||||
}
|
||||
|
||||
fn unsigned_additions(&self, event_id: &str) -> Option<&JsonObject> {
|
||||
self.unsigned_additions.get(event_id)
|
||||
}
|
||||
}
|
||||
|
||||
/// `encode_rooms` for one room.
|
||||
pub fn encode_room(
|
||||
entry: &RoomEntry,
|
||||
time_now_ms: i64,
|
||||
config: &SerializeEventConfig,
|
||||
unsigned_additions: &HashMap<String, JsonObject>,
|
||||
) -> PyResult<Value> {
|
||||
let lookups = EntryLookups {
|
||||
entry,
|
||||
unsigned_additions,
|
||||
};
|
||||
|
||||
// Notifications are only done properly by clients, so these are
|
||||
// placeholders.
|
||||
let mut room = Map::new();
|
||||
room.insert("notification_count".to_owned(), json!(0));
|
||||
room.insert("highlight_count".to_owned(), json!(0));
|
||||
if let Some(bump_stamp) = entry.bump_stamp {
|
||||
room.insert("bump_stamp".to_owned(), json!(bump_stamp));
|
||||
}
|
||||
if let Some(joined_count) = entry.joined_count {
|
||||
room.insert("joined_count".to_owned(), json!(joined_count));
|
||||
}
|
||||
if let Some(invited_count) = entry.invited_count {
|
||||
room.insert("invited_count".to_owned(), json!(invited_count));
|
||||
}
|
||||
if let Some(name) = entry.name.as_deref().filter(|name| !name.is_empty()) {
|
||||
room.insert("name".to_owned(), json!(name));
|
||||
}
|
||||
if let Some(avatar) = entry.avatar.as_deref().filter(|avatar| !avatar.is_empty()) {
|
||||
room.insert("avatar".to_owned(), json!(avatar));
|
||||
}
|
||||
if !entry.heroes.is_empty() {
|
||||
let heroes: Vec<Value> = entry
|
||||
.heroes
|
||||
.iter()
|
||||
.map(|hero| {
|
||||
let mut serialized = Map::new();
|
||||
serialized.insert("user_id".to_owned(), json!(hero.user_id));
|
||||
// The spec spells it "displayname".
|
||||
if let Some(display_name) = &hero.display_name {
|
||||
serialized.insert("displayname".to_owned(), json!(display_name));
|
||||
}
|
||||
if let Some(avatar_url) = &hero.avatar_url {
|
||||
serialized.insert("avatar_url".to_owned(), json!(avatar_url));
|
||||
}
|
||||
Value::Object(serialized)
|
||||
})
|
||||
.collect();
|
||||
room.insert("heroes".to_owned(), Value::Array(heroes));
|
||||
}
|
||||
// Only sent when true, to save bandwidth.
|
||||
if entry.initial {
|
||||
room.insert("initial".to_owned(), json!(true));
|
||||
}
|
||||
if entry.unstable_expanded_timeline {
|
||||
room.insert("unstable_expanded_timeline".to_owned(), json!(true));
|
||||
}
|
||||
if !entry.required_state.is_empty() {
|
||||
let state = entry
|
||||
.required_state
|
||||
.iter()
|
||||
.map(|event| {
|
||||
serialize_event(event, time_now_ms, config, None, false, &lookups)
|
||||
.map(Value::Object)
|
||||
})
|
||||
.collect::<PyResult<Vec<_>>>()?;
|
||||
room.insert("required_state".to_owned(), Value::Array(state));
|
||||
}
|
||||
if !entry.timeline.is_empty() {
|
||||
let with_bundles = entry.bundles.is_some();
|
||||
let timeline = entry
|
||||
.timeline
|
||||
.iter()
|
||||
.map(|filtered| {
|
||||
serialize_event(
|
||||
&filtered.event,
|
||||
time_now_ms,
|
||||
config,
|
||||
Some(&filtered.membership),
|
||||
with_bundles,
|
||||
&lookups,
|
||||
)
|
||||
.map(Value::Object)
|
||||
})
|
||||
.collect::<PyResult<Vec<_>>>()?;
|
||||
room.insert("timeline".to_owned(), Value::Array(timeline));
|
||||
}
|
||||
if let Some(limited) = entry.limited {
|
||||
room.insert("limited".to_owned(), json!(limited));
|
||||
}
|
||||
if let Some(prev_batch) = &entry.prev_batch {
|
||||
room.insert("prev_batch".to_owned(), json!(prev_batch));
|
||||
}
|
||||
if let Some(num_live) = entry.num_live {
|
||||
room.insert("num_live".to_owned(), json!(num_live));
|
||||
}
|
||||
// Absent on rooms that are not DMs.
|
||||
if entry.is_dm {
|
||||
room.insert("is_dm".to_owned(), json!(true));
|
||||
}
|
||||
if !entry.stripped_state.is_empty() {
|
||||
room.insert(
|
||||
"invite_state".to_owned(),
|
||||
Value::Array(entry.stripped_state.clone()),
|
||||
);
|
||||
}
|
||||
Ok(Value::Object(room))
|
||||
}
|
||||
|
||||
/// `encode_rooms` for `rooms`, in order. `unsigned_additions` are the
|
||||
/// module callbacks' additions to the rooms' events, by event ID.
|
||||
#[pyfunction]
|
||||
#[pyo3(signature = (rooms, time_now_ms, config, *, unsigned_additions = None))]
|
||||
pub fn encode_rooms<'py>(
|
||||
py: Python<'py>,
|
||||
rooms: Vec<Bound<'py, RoomResult>>,
|
||||
time_now_ms: i64,
|
||||
config: &SerializeEventConfig,
|
||||
unsigned_additions: Option<HashMap<String, JsonObject>>,
|
||||
) -> PyResult<Vec<Bound<'py, PyAny>>> {
|
||||
let unsigned_additions = unsigned_additions.unwrap_or_default();
|
||||
rooms
|
||||
.iter()
|
||||
.map(|room| {
|
||||
let encoded = encode_room(&room.get().entry, time_now_ms, config, &unsigned_additions)?;
|
||||
Ok(pythonize(py, &encoded)?)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
@@ -26,22 +26,28 @@
|
||||
//! ([`SlidingSyncHandler::defer`]), and a `_blocking` twin runs to completion
|
||||
//! for tests ([`SlidingSyncHandler::block_on`]).
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
use futures::future::BoxFuture;
|
||||
use pyo3::{exceptions::PyRuntimeError, intern, prelude::*, types::PyModule};
|
||||
use request::SlidingSyncRequest;
|
||||
use result::{RoomInput, RoomsResult};
|
||||
|
||||
use crate::database::postgres::pool::{ConnectionPool, PyConnectionPool};
|
||||
use crate::deferred::{create_deferred_with, OnCancel};
|
||||
use crate::runtime::RustRuntime;
|
||||
use crate::storage::sync_store::{SyncDbStats, SyncStore};
|
||||
use crate::storage::sync_store::{anyhow_to_py, SyncDbStats, SyncStore};
|
||||
|
||||
pub mod config;
|
||||
pub mod connection;
|
||||
pub mod encode;
|
||||
pub mod extensions;
|
||||
pub mod request;
|
||||
pub mod required_state;
|
||||
pub mod result;
|
||||
pub mod room;
|
||||
pub mod room_entry;
|
||||
pub mod room_lists;
|
||||
|
||||
/// Computes sliding sync responses with the Rust implementation, reading
|
||||
@@ -102,6 +108,62 @@ impl SlidingSyncHandler {
|
||||
let handle = self.runtime.tokio_handle()?;
|
||||
py.detach(|| handle.block_on(future))
|
||||
}
|
||||
|
||||
/// Start computing the entries of `rooms`: read them from Python, and
|
||||
/// return the future that computes them off the GIL.
|
||||
fn start_room_entries(
|
||||
&self,
|
||||
py: Python<'_>,
|
||||
request: &Bound<'_, SlidingSyncRequest>,
|
||||
rooms: Vec<Bound<'_, RoomInput>>,
|
||||
instance_ids: HashMap<String, i64>,
|
||||
) -> PyResult<BoxFuture<'static, PyResult<RoomsResult>>> {
|
||||
let pool = self.pool(py)?;
|
||||
let request = Arc::clone(&request.get().data);
|
||||
let rooms: Vec<_> = rooms.iter().map(|room| room.get().data.clone()).collect();
|
||||
let store = Arc::clone(&self.store);
|
||||
let now_ms = self.runtime.clock().now_millis() as i64;
|
||||
Ok(Box::pin(async move {
|
||||
let reader = store.sliding_sync_reader(pool);
|
||||
let outcomes =
|
||||
room_entry::generate_room_entries(&reader, &request, &instance_ids, now_ms, &rooms)
|
||||
.await
|
||||
.map_err(anyhow_to_py)?;
|
||||
let stats = SlidingSyncStats::from_db(&reader.stats());
|
||||
Ok(RoomsResult::new(outcomes, request.from.is_some(), stats))
|
||||
}))
|
||||
}
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl SlidingSyncHandler {
|
||||
/// `get_room_sync_data` for each of `rooms`, twenty at a time: a
|
||||
/// deferred resolving to a `RoomsResult`. `instance_ids` are
|
||||
/// `get_id_for_instance` for the writers the rooms' tokens may name.
|
||||
#[pyo3(signature = (request, rooms, *, instance_ids))]
|
||||
fn generate_room_entries<'py>(
|
||||
&self,
|
||||
py: Python<'py>,
|
||||
request: &Bound<'py, SlidingSyncRequest>,
|
||||
rooms: Vec<Bound<'py, RoomInput>>,
|
||||
instance_ids: HashMap<String, i64>,
|
||||
) -> PyResult<Bound<'py, PyAny>> {
|
||||
let future = self.start_room_entries(py, request, rooms, instance_ids)?;
|
||||
self.defer(py, future)
|
||||
}
|
||||
|
||||
/// As [`Self::generate_room_entries`], but blocks.
|
||||
#[pyo3(signature = (request, rooms, *, instance_ids))]
|
||||
fn generate_room_entries_blocking(
|
||||
&self,
|
||||
py: Python<'_>,
|
||||
request: &Bound<'_, SlidingSyncRequest>,
|
||||
rooms: Vec<Bound<'_, RoomInput>>,
|
||||
instance_ids: HashMap<String, i64>,
|
||||
) -> PyResult<RoomsResult> {
|
||||
let future = self.start_room_entries(py, request, rooms, instance_ids)?;
|
||||
self.block_on(py, future)
|
||||
}
|
||||
}
|
||||
|
||||
/// What the Rust implementation did for one request, for the
|
||||
@@ -167,6 +229,10 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
|
||||
child_module.add_class::<SlidingSyncHandler>()?;
|
||||
child_module.add_class::<SlidingSyncStats>()?;
|
||||
child_module.add_class::<request::SlidingSyncRequest>()?;
|
||||
child_module.add_class::<RoomInput>()?;
|
||||
child_module.add_class::<result::RoomResult>()?;
|
||||
child_module.add_class::<RoomsResult>()?;
|
||||
child_module.add_function(wrap_pyfunction!(encode::encode_rooms, &child_module)?)?;
|
||||
child_module.add_class::<crate::storage::sliding_sync_store::PySlidingSyncStoreForTest>()?;
|
||||
config::register_test_functions(&child_module)?;
|
||||
connection::register_test_functions(&child_module)?;
|
||||
|
||||
@@ -0,0 +1,376 @@
|
||||
/*
|
||||
* 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>.
|
||||
*/
|
||||
|
||||
//! What crosses to Python for the rooms of a request: [`RoomInput`] in,
|
||||
//! [`RoomsResult`] out, and a [`RoomResult`] per room Rust computed.
|
||||
//!
|
||||
//! A `RoomResult`'s events never become Python objects, except for module
|
||||
//! callbacks: Python reads only what `current_sync_for_user` and the
|
||||
//! extensions need, and the servlet encodes the room in Rust.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use pyo3::{
|
||||
exceptions::PyValueError,
|
||||
intern,
|
||||
prelude::*,
|
||||
types::{PyList, PyTuple},
|
||||
};
|
||||
|
||||
use crate::caches::lock_ignoring_poison;
|
||||
use crate::events::Event;
|
||||
use crate::sliding_sync::config::{RequiredStateMap, RoomSyncConfig, MEMBER};
|
||||
use crate::sliding_sync::connection::HaveSentRoom;
|
||||
use crate::sliding_sync::room_entry::{RoomEntry, RoomInputData, RoomOutcome};
|
||||
use crate::sliding_sync::room_lists::RoomForUser;
|
||||
use crate::sliding_sync::SlidingSyncStats;
|
||||
use crate::storage::sync_store::PersistedEventPosition;
|
||||
use crate::sync::relations::Bundle;
|
||||
use crate::types::stream_token::RoomStreamToken;
|
||||
|
||||
/// A `required_state_map` from Python: a mapping of types to sets of
|
||||
/// state keys.
|
||||
pub fn required_state_map_from_py(map: &Bound<'_, PyAny>) -> PyResult<RequiredStateMap> {
|
||||
let py = map.py();
|
||||
let mut result = RequiredStateMap::new();
|
||||
for item in map.call_method0(intern!(py, "items"))?.try_iter()? {
|
||||
let (state_type, state_keys): (String, Bound<'_, PyAny>) = item?.extract()?;
|
||||
let mut keys = BTreeSet::new();
|
||||
for key in state_keys.try_iter()? {
|
||||
keys.insert(key?.extract::<String>()?);
|
||||
}
|
||||
result.insert(state_type, keys);
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// The input of one room's `get_room_sync_data`. See [`RoomInputData`].
|
||||
#[pyclass(frozen, skip_from_py_object)]
|
||||
pub struct RoomInput {
|
||||
pub(crate) data: RoomInputData,
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl RoomInput {
|
||||
/// `room_status` is a `HaveSentRoomFlag` value, with its token for
|
||||
/// "previously".
|
||||
#[new]
|
||||
#[pyo3(signature = (*, room_id, sender, membership, event_id, instance_name, stream, room_version_id, timeline_limit, required_state_map, prev_room_sync_config, room_status, room_status_token, newly_joined, newly_left, is_dm))]
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn py_new(
|
||||
room_id: &str,
|
||||
sender: Option<&str>,
|
||||
membership: &str,
|
||||
event_id: Option<&str>,
|
||||
instance_name: &str,
|
||||
stream: i64,
|
||||
room_version_id: Option<&str>,
|
||||
timeline_limit: i64,
|
||||
required_state_map: &Bound<'_, PyAny>,
|
||||
prev_room_sync_config: Option<(i64, Bound<'_, PyAny>)>,
|
||||
room_status: &str,
|
||||
room_status_token: Option<RoomStreamToken>,
|
||||
newly_joined: bool,
|
||||
newly_left: bool,
|
||||
is_dm: bool,
|
||||
) -> PyResult<Self> {
|
||||
let room_status = match (room_status, room_status_token) {
|
||||
("never", _) => HaveSentRoom::Never,
|
||||
("live", _) => HaveSentRoom::Live,
|
||||
("previously", Some(token)) => HaveSentRoom::Previously(token),
|
||||
(status, _) => {
|
||||
return Err(PyValueError::new_err(format!(
|
||||
"bad room status {status:?} or missing token"
|
||||
)))
|
||||
}
|
||||
};
|
||||
let prev_room_sync_config = prev_room_sync_config
|
||||
.map(|(timeline_limit, map)| {
|
||||
Ok::<_, PyErr>(RoomSyncConfig {
|
||||
timeline_limit,
|
||||
required_state_map: required_state_map_from_py(&map)?,
|
||||
})
|
||||
})
|
||||
.transpose()?;
|
||||
Ok(RoomInput {
|
||||
data: RoomInputData {
|
||||
membership: RoomForUser {
|
||||
room_id: Arc::from(room_id),
|
||||
sender: sender.map(Arc::from),
|
||||
membership: Arc::from(membership),
|
||||
event_id: event_id.map(Arc::from),
|
||||
event_pos: PersistedEventPosition {
|
||||
instance_name: Arc::from(instance_name),
|
||||
stream,
|
||||
},
|
||||
room_version_id: room_version_id.map(Arc::from),
|
||||
// Not read by the room's computation.
|
||||
has_known_state: true,
|
||||
room_type: None,
|
||||
is_encrypted: false,
|
||||
},
|
||||
room_sync_config: RoomSyncConfig {
|
||||
timeline_limit,
|
||||
required_state_map: required_state_map_from_py(required_state_map)?,
|
||||
},
|
||||
prev_room_sync_config,
|
||||
room_status,
|
||||
newly_joined,
|
||||
newly_left,
|
||||
is_dm,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn room_id(&self) -> &str {
|
||||
&self.data.membership.room_id
|
||||
}
|
||||
}
|
||||
|
||||
/// One room's entry, computed in Rust: `SlidingSyncResult.RoomResult` as
|
||||
/// far as `current_sync_for_user` and the extensions read it.
|
||||
#[pyclass(frozen, skip_from_py_object)]
|
||||
pub struct RoomResult {
|
||||
pub(crate) entry: Arc<RoomEntry>,
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl RoomResult {
|
||||
fn __bool__(&self) -> bool {
|
||||
self.entry.has_updates()
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn room_id(&self) -> &str {
|
||||
&self.entry.room_id
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn initial(&self) -> bool {
|
||||
self.entry.initial
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn unstable_expanded_timeline(&self) -> bool {
|
||||
self.entry.unstable_expanded_timeline
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn timeline_event_ids(&self) -> Vec<&str> {
|
||||
self.entry
|
||||
.timeline
|
||||
.iter()
|
||||
.map(|filtered| filtered.event.event_id())
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn timeline_senders(&self) -> Vec<&str> {
|
||||
self.entry
|
||||
.timeline
|
||||
.iter()
|
||||
.map(|filtered| filtered.event.sender())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The state keys of the membership events in `required_state`.
|
||||
#[getter]
|
||||
fn required_state_member_keys(&self) -> Vec<&str> {
|
||||
self.entry
|
||||
.required_state
|
||||
.iter()
|
||||
.filter(|event| event.type_() == MEMBER)
|
||||
.filter_map(Event::state_key)
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn hero_user_ids(&self) -> Vec<&str> {
|
||||
self.entry
|
||||
.heroes
|
||||
.iter()
|
||||
.map(|hero| hero.user_id.as_str())
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Every event the room's encoding serializes, for the module
|
||||
/// callbacks that add to their unsigned data.
|
||||
fn events_to_serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
|
||||
let bundled = self
|
||||
.entry
|
||||
.bundles
|
||||
.iter()
|
||||
.flat_map(|bundles| bundles.values().flat_map(Bundle::events));
|
||||
let events = self
|
||||
.entry
|
||||
.timeline
|
||||
.iter()
|
||||
.map(|filtered| &filtered.event)
|
||||
.chain(&self.entry.required_state)
|
||||
.chain(bundled)
|
||||
.map(|event| Bound::new(py, event.shallow_clone()))
|
||||
.collect::<PyResult<Vec<_>>>()?;
|
||||
PyList::new(py, events)
|
||||
}
|
||||
}
|
||||
|
||||
/// A [`RequiredStateMap`] for Python: types to sets of state keys.
|
||||
type PyRequiredStateMap = BTreeMap<String, BTreeSet<String>>;
|
||||
|
||||
/// A room's `RoomLazyMembershipChanges`: `(room_id,
|
||||
/// returned_user_id_to_last_seen_ts_map, invalidated_user_ids)`.
|
||||
type LazyMembership = (Arc<str>, HashMap<String, Option<i64>>, HashSet<String>);
|
||||
|
||||
/// What [`crate::sliding_sync::room_entry::generate_room_entries`] made of
|
||||
/// the rooms.
|
||||
#[pyclass(frozen, skip_from_py_object)]
|
||||
pub struct RoomsResult {
|
||||
/// The entries to send. Taken by [`RoomsResult::take_entries`].
|
||||
entries: Mutex<Vec<Arc<RoomEntry>>>,
|
||||
/// The rooms computed with nothing to send.
|
||||
empty: Vec<Arc<str>>,
|
||||
/// `room_configs[room_id]` for the computed rooms whose config changes.
|
||||
room_config_updates: Vec<(Arc<str>, RoomSyncConfig)>,
|
||||
/// `room_lazy_membership[room_id]` for every computed room.
|
||||
lazy_memberships: Vec<LazyMembership>,
|
||||
fallbacks: Vec<(Arc<str>, String)>,
|
||||
errors: Vec<(Arc<str>, String)>,
|
||||
stats: SlidingSyncStats,
|
||||
}
|
||||
|
||||
impl RoomsResult {
|
||||
/// The result of `outcomes`. An entry with nothing to send is dropped
|
||||
/// from an incremental response.
|
||||
pub fn new(
|
||||
outcomes: Vec<(Arc<str>, RoomOutcome)>,
|
||||
incremental: bool,
|
||||
mut stats: SlidingSyncStats,
|
||||
) -> Self {
|
||||
let mut entries = Vec::new();
|
||||
let mut empty = Vec::new();
|
||||
let mut room_config_updates = Vec::new();
|
||||
let mut lazy_memberships = Vec::new();
|
||||
let mut fallbacks = Vec::new();
|
||||
let mut errors = Vec::new();
|
||||
for (room_id, outcome) in outcomes {
|
||||
match outcome {
|
||||
RoomOutcome::Computed(computed) => {
|
||||
stats.rooms_computed += 1;
|
||||
if computed.lazy_member_lookup {
|
||||
stats.lazy_member_lookups += 1;
|
||||
}
|
||||
let mutations = computed.mutations;
|
||||
if let Some(config) = mutations.room_config {
|
||||
room_config_updates.push((Arc::clone(&room_id), config));
|
||||
}
|
||||
let lazy = mutations.lazy_membership;
|
||||
lazy_memberships.push((
|
||||
Arc::clone(&room_id),
|
||||
lazy.returned_user_id_to_last_seen_ts_map,
|
||||
lazy.invalidated_user_ids,
|
||||
));
|
||||
if computed.entry.has_updates() || !incremental {
|
||||
entries.push(Arc::new(computed.entry));
|
||||
} else {
|
||||
empty.push(room_id);
|
||||
}
|
||||
}
|
||||
RoomOutcome::Fallback(reason) => fallbacks.push((room_id, reason)),
|
||||
RoomOutcome::Error(error) => errors.push((room_id, error)),
|
||||
}
|
||||
}
|
||||
RoomsResult {
|
||||
entries: Mutex::new(entries),
|
||||
empty,
|
||||
room_config_updates,
|
||||
lazy_memberships,
|
||||
fallbacks,
|
||||
errors,
|
||||
stats,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[pymethods]
|
||||
impl RoomsResult {
|
||||
/// The entries to send as `RoomResult`s. A second call returns an
|
||||
/// empty list.
|
||||
fn take_entries<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
|
||||
let entries = std::mem::take(&mut *lock_ignoring_poison(&self.entries));
|
||||
let items = entries
|
||||
.into_iter()
|
||||
.map(|entry| Bound::new(py, RoomResult { entry }))
|
||||
.collect::<PyResult<Vec<_>>>()?;
|
||||
PyList::new(py, items)
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn empty(&self) -> Vec<&str> {
|
||||
self.empty.iter().map(|room_id| &**room_id).collect()
|
||||
}
|
||||
|
||||
/// `(room_id, timeline_limit, required_state_map)` for each room whose
|
||||
/// config the connection records.
|
||||
#[getter]
|
||||
fn room_config_updates<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
|
||||
let items = self
|
||||
.room_config_updates
|
||||
.iter()
|
||||
.map(|(room_id, config)| {
|
||||
let map: PyRequiredStateMap = config.required_state_map.clone();
|
||||
(&**room_id, config.timeline_limit, map).into_pyobject(py)
|
||||
})
|
||||
.collect::<PyResult<Vec<Bound<'py, PyTuple>>>>()?;
|
||||
PyList::new(py, items)
|
||||
}
|
||||
|
||||
/// `(room_id, returned_user_id_to_last_seen_ts_map,
|
||||
/// invalidated_user_ids)` for each room computed.
|
||||
#[getter]
|
||||
fn lazy_memberships<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
|
||||
let items = self
|
||||
.lazy_memberships
|
||||
.iter()
|
||||
.map(|(room_id, returned, invalidated)| {
|
||||
(&**room_id, returned, invalidated).into_pyobject(py)
|
||||
})
|
||||
.collect::<PyResult<Vec<Bound<'py, PyTuple>>>>()?;
|
||||
PyList::new(py, items)
|
||||
}
|
||||
|
||||
/// `(room_id, reason)` for each room Python must compute.
|
||||
#[getter]
|
||||
fn fallbacks(&self) -> Vec<(&str, &str)> {
|
||||
self.fallbacks
|
||||
.iter()
|
||||
.map(|(room_id, reason)| (&**room_id, reason.as_str()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// `(room_id, error)` for each room Rust failed on.
|
||||
#[getter]
|
||||
fn errors(&self) -> Vec<(&str, &str)> {
|
||||
self.errors
|
||||
.iter()
|
||||
.map(|(room_id, error)| (&**room_id, error.as_str()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[getter]
|
||||
fn stats(&self) -> SlidingSyncStats {
|
||||
self.stats
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,925 @@
|
||||
/*
|
||||
* 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>.
|
||||
*/
|
||||
|
||||
//! `get_room_sync_data`: one room's entry in a sliding sync response, in
|
||||
//! the order of the Python, and the batch of rooms Python hands over in
|
||||
//! one call.
|
||||
//!
|
||||
//! The pure decisions are in [`crate::sliding_sync::room`] and
|
||||
//! [`crate::sliding_sync::required_state`]. This module does the reads
|
||||
//! between them.
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::panic::AssertUnwindSafe;
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{bail, Context};
|
||||
use futures::{stream, FutureExt, StreamExt};
|
||||
use indexmap::IndexMap;
|
||||
use serde_json::{Map, Value};
|
||||
|
||||
use crate::events::Event;
|
||||
use crate::sliding_sync::config::{RoomSyncConfig, MEMBER};
|
||||
use crate::sliding_sync::connection::{HaveSentRoom, RoomLazyMembershipChanges};
|
||||
use crate::sliding_sync::request::SlidingSyncRequestData;
|
||||
use crate::sliding_sync::required_state::{
|
||||
required_state_changes, required_state_request, state_filter_with_meta, TimelineMember,
|
||||
};
|
||||
use crate::sliding_sync::room::{
|
||||
always_return_bump_stamp, bump_stamp_from_last_event, final_bump_stamp, needs_heroes, num_live,
|
||||
room_config_to_persist, store_bump_stamp, strip_event, stripped_state, timeline_bounds,
|
||||
timeline_bump_stamp, StoreBumpStamp, StrippedHero, TimelineBounds, BUMP_EVENT_TYPES,
|
||||
};
|
||||
use crate::sliding_sync::room_lists::RoomForUser;
|
||||
use crate::state::{lookup, StateKey, StateMap};
|
||||
use crate::storage::events::RedactBehaviour;
|
||||
use crate::storage::sliding_sync_store::SlidingSyncReader;
|
||||
use crate::storage::sync_store::{PaginationOrder, RoomSummary};
|
||||
use crate::sync::client_events::{filter_events_for_client, ClientFilterContext};
|
||||
use crate::sync::relations::{bundled_aggregations, Bundle};
|
||||
use crate::sync::room::Fallback;
|
||||
use crate::sync::rooms::panic_message;
|
||||
use crate::sync::summary::extract_heroes_from_room_summary;
|
||||
use crate::types::state_filter::StateFilter;
|
||||
use crate::types::stream_token::{Direction, RoomStreamToken};
|
||||
use crate::visibility::FilteredEvent;
|
||||
|
||||
const M_ROOM_NAME: &str = "m.room.name";
|
||||
const M_ROOM_AVATAR: &str = "m.room.avatar";
|
||||
|
||||
/// How many rooms are computed at once, as `concurrently_execute` does.
|
||||
const ROOM_ENTRY_CONCURRENCY: usize = 20;
|
||||
|
||||
/// What `current_sync_for_user` passes `get_room_sync_data` for one room,
|
||||
/// besides the request.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct RoomInputData {
|
||||
/// The user's membership in the room at the `to_token`.
|
||||
pub membership: RoomForUser,
|
||||
pub room_sync_config: RoomSyncConfig,
|
||||
/// What the connection last recorded for the room.
|
||||
pub prev_room_sync_config: Option<RoomSyncConfig>,
|
||||
/// Whether the connection sent the room before.
|
||||
pub room_status: HaveSentRoom<RoomStreamToken>,
|
||||
pub newly_joined: bool,
|
||||
pub newly_left: bool,
|
||||
pub is_dm: bool,
|
||||
}
|
||||
|
||||
/// `SlidingSyncResult.RoomResult`, holding the events Rust loaded.
|
||||
pub struct RoomEntry {
|
||||
pub room_id: Arc<str>,
|
||||
/// May be empty, which counts as an update but is not sent.
|
||||
pub name: Option<String>,
|
||||
pub avatar: Option<String>,
|
||||
pub heroes: Vec<StrippedHero>,
|
||||
pub is_dm: bool,
|
||||
pub initial: bool,
|
||||
pub unstable_expanded_timeline: bool,
|
||||
pub required_state: Vec<Event>,
|
||||
pub timeline: Vec<FilteredEvent>,
|
||||
/// `None` unless the timeline is limited.
|
||||
pub bundles: Option<HashMap<Arc<str>, Bundle>>,
|
||||
/// The redactions of the redacted events among the room's, which the
|
||||
/// serializer adds to their unsigned data.
|
||||
pub redactions: HashMap<Arc<str>, Event>,
|
||||
pub stripped_state: Vec<Value>,
|
||||
/// The `prev_batch` token as a string.
|
||||
pub prev_batch: Option<String>,
|
||||
pub limited: Option<bool>,
|
||||
pub num_live: Option<i64>,
|
||||
pub bump_stamp: Option<i64>,
|
||||
pub joined_count: Option<i64>,
|
||||
pub invited_count: Option<i64>,
|
||||
}
|
||||
|
||||
impl RoomEntry {
|
||||
/// `RoomResult.__bool__`: whether the entry holds an update to return
|
||||
/// at once.
|
||||
pub fn has_updates(&self) -> bool {
|
||||
self.initial
|
||||
|| self.name.is_some()
|
||||
|| self.avatar.is_some()
|
||||
|| !self.heroes.is_empty()
|
||||
|| self.joined_count.is_some()
|
||||
|| self.invited_count.is_some()
|
||||
|| !self.required_state.is_empty()
|
||||
|| !self.timeline.is_empty()
|
||||
|| !self.stripped_state.is_empty()
|
||||
}
|
||||
}
|
||||
|
||||
/// What `get_room_sync_data` writes to the new connection state for the
|
||||
/// room.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct RoomMutations {
|
||||
/// `room_configs[room_id]`, if it changes.
|
||||
pub room_config: Option<RoomSyncConfig>,
|
||||
/// `room_lazy_membership[room_id]`, always set.
|
||||
pub lazy_membership: RoomLazyMembershipChanges,
|
||||
}
|
||||
|
||||
/// One room computed.
|
||||
pub struct RoomComputed {
|
||||
pub entry: RoomEntry,
|
||||
pub mutations: RoomMutations,
|
||||
/// Whether the room's lazy-loaded members were looked up in the
|
||||
/// connection state.
|
||||
pub lazy_member_lookup: bool,
|
||||
}
|
||||
|
||||
/// How one room came out.
|
||||
pub enum RoomOutcome {
|
||||
Computed(Box<RoomComputed>),
|
||||
/// Python must compute the room: [`Fallback`].
|
||||
Fallback(String),
|
||||
/// Computing the room failed.
|
||||
Error(String),
|
||||
}
|
||||
|
||||
/// What computing one room reads, shared by the rooms of a request.
|
||||
pub struct RoomContext<'r, 'a> {
|
||||
pub reader: &'r SlidingSyncReader<'a>,
|
||||
pub request: &'r SlidingSyncRequestData,
|
||||
pub client_filter: ClientFilterContext,
|
||||
/// `get_id_for_instance` for the writers the rooms' tokens may name.
|
||||
pub instance_ids: &'r HashMap<String, i64>,
|
||||
/// The time the visibility checks compare with.
|
||||
pub now_ms: i64,
|
||||
}
|
||||
|
||||
/// One room's view of the request: the room, and where the user's view of
|
||||
/// it ends.
|
||||
struct Room<'r> {
|
||||
input: &'r RoomInputData,
|
||||
room_id: &'r str,
|
||||
membership: &'r str,
|
||||
/// Where a leave or ban stops what the user sees: the membership's
|
||||
/// position, as `event_pos.to_room_stream_token()`.
|
||||
membership_token: RoomStreamToken,
|
||||
}
|
||||
|
||||
impl Room<'_> {
|
||||
fn has_left(&self) -> bool {
|
||||
matches!(self.membership, "leave" | "ban")
|
||||
}
|
||||
}
|
||||
|
||||
/// The events of `event_ids`, by state key. Events that are not found or
|
||||
/// rejected are left out.
|
||||
///
|
||||
/// An event of the room's timeline is that event, with the previous content
|
||||
/// the timeline's load gave it. Python's loads find the same objects through
|
||||
/// its map of the events still held, whatever the event cache holds.
|
||||
async fn state_events(
|
||||
reader: &SlidingSyncReader<'_>,
|
||||
timeline: &HashMap<&str, &Arc<Event>>,
|
||||
event_ids: &[Arc<str>],
|
||||
) -> anyhow::Result<HashMap<StateKey, Event>> {
|
||||
let (held, to_load): (Vec<&Arc<str>>, Vec<&Arc<str>>) = event_ids
|
||||
.iter()
|
||||
.partition(|event_id| timeline.contains_key(&***event_id));
|
||||
let to_load: Vec<Arc<str>> = to_load.into_iter().cloned().collect();
|
||||
let loaded = reader
|
||||
.events(&to_load, RedactBehaviour::Redact, false, false)
|
||||
.await?;
|
||||
let held = held
|
||||
.into_iter()
|
||||
.map(|event_id| timeline[&**event_id].shallow_clone());
|
||||
Ok(loaded
|
||||
.into_iter()
|
||||
.chain(held)
|
||||
.filter_map(|event| {
|
||||
let key = StateKey::new(event.type_(), event.state_key()?);
|
||||
Some((key, event))
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// `get_event`: the event, which must exist.
|
||||
async fn one_event(reader: &SlidingSyncReader<'_>, event_id: &Arc<str>) -> anyhow::Result<Event> {
|
||||
reader
|
||||
.events(
|
||||
std::slice::from_ref(event_id),
|
||||
RedactBehaviour::Redact,
|
||||
false,
|
||||
false,
|
||||
)
|
||||
.await?
|
||||
.into_iter()
|
||||
.next()
|
||||
.with_context(|| format!("Could not find event {event_id}"))
|
||||
}
|
||||
|
||||
/// `get_current_state_ids_at`: the room's state as the user sees it, at
|
||||
/// their leave or ban, or the current state.
|
||||
async fn state_ids_at(
|
||||
context: &RoomContext<'_, '_>,
|
||||
room: &Room<'_>,
|
||||
state_filter: &StateFilter,
|
||||
) -> anyhow::Result<StateMap> {
|
||||
if room.has_left() {
|
||||
context
|
||||
.reader
|
||||
.state_ids_at(room.room_id, &room.membership_token, state_filter)
|
||||
.await
|
||||
} else {
|
||||
context
|
||||
.reader
|
||||
.current_state_ids(room.room_id, state_filter)
|
||||
.await
|
||||
}
|
||||
}
|
||||
|
||||
/// `get_current_state_at`: as [`state_ids_at`], with the events.
|
||||
async fn state_at(
|
||||
context: &RoomContext<'_, '_>,
|
||||
room: &Room<'_>,
|
||||
timeline: &HashMap<&str, &Arc<Event>>,
|
||||
state_filter: &StateFilter,
|
||||
) -> anyhow::Result<HashMap<StateKey, Event>> {
|
||||
let state_ids = state_ids_at(context, room, state_filter).await?;
|
||||
let event_ids: Vec<Arc<str>> = state_ids.into_values().collect();
|
||||
state_events(context.reader, timeline, &event_ids).await
|
||||
}
|
||||
|
||||
/// `_stripped_state` of an invite or knock event.
|
||||
fn stripped_state_of(event: &Event) -> anyhow::Result<Vec<Value>> {
|
||||
let mut unsigned = Map::new();
|
||||
let unsigned_ref = event.unsigned_ref();
|
||||
if let Some(state) = unsigned_ref.invite_room_state()? {
|
||||
unsigned.insert("invite_room_state".to_owned(), Value::Array(state));
|
||||
}
|
||||
if let Some(state) = unsigned_ref.knock_room_state()? {
|
||||
unsigned.insert("knock_room_state".to_owned(), Value::Array(state));
|
||||
}
|
||||
let content = Value::Object(
|
||||
event
|
||||
.content()
|
||||
.as_map()
|
||||
.iter()
|
||||
.map(|(key, value)| (key.to_string(), value.clone()))
|
||||
.collect(),
|
||||
);
|
||||
let membership = event
|
||||
.content()
|
||||
.get_field("membership")
|
||||
.and_then(Value::as_str);
|
||||
let stripped = strip_event(
|
||||
event.type_(),
|
||||
event.state_key().unwrap_or_default(),
|
||||
&content,
|
||||
event.sender(),
|
||||
);
|
||||
Ok(stripped_state(membership, &unsigned, stripped))
|
||||
}
|
||||
|
||||
/// A string field of an event's content, as the Python's `isinstance(value,
|
||||
/// str)` checks read it.
|
||||
fn content_str(event: &Event, key: &str) -> Option<String> {
|
||||
event
|
||||
.content()
|
||||
.get_field(key)
|
||||
.and_then(Value::as_str)
|
||||
.map(str::to_owned)
|
||||
}
|
||||
|
||||
/// `_get_bump_stamp`.
|
||||
async fn bump_stamp(
|
||||
context: &RoomContext<'_, '_>,
|
||||
room_id: &str,
|
||||
timeline: &[FilteredEvent],
|
||||
check_outside_timeline: bool,
|
||||
) -> anyhow::Result<Option<i64>> {
|
||||
let in_timeline = timeline_bump_stamp(timeline.iter().filter_map(|filtered| {
|
||||
let stream = filtered.event.internal_metadata().stream_ordering()?;
|
||||
Some((filtered.event.type_(), stream.get()))
|
||||
}));
|
||||
if in_timeline.is_some() {
|
||||
return Ok(in_timeline);
|
||||
}
|
||||
if !check_outside_timeline {
|
||||
return Ok(None);
|
||||
}
|
||||
let request = context.request;
|
||||
let latest = context.reader.latest_bump_stamp(room_id).await?;
|
||||
match store_bump_stamp(
|
||||
latest,
|
||||
request.have_finished_bg_jobs,
|
||||
request.to_room_key.stream,
|
||||
) {
|
||||
StoreBumpStamp::Done(stamp) => Ok(stamp),
|
||||
StoreBumpStamp::LookupLastEvent => {
|
||||
let types: Vec<String> = BUMP_EVENT_TYPES.iter().map(|t| t.to_string()).collect();
|
||||
let last = context
|
||||
.reader
|
||||
.last_event_of_types_before(room_id, &request.to_room_key, &types)
|
||||
.await?;
|
||||
Ok(bump_stamp_from_last_event(last.map(|(_, pos)| pos.stream)))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The redactions of the redacted events of `entry`.
|
||||
async fn load_redactions(
|
||||
reader: &SlidingSyncReader<'_>,
|
||||
entry: &RoomEntry,
|
||||
) -> anyhow::Result<HashMap<Arc<str>, Event>> {
|
||||
let mut redaction_ids: Vec<Arc<str>> = Vec::new();
|
||||
let bundled = entry
|
||||
.bundles
|
||||
.iter()
|
||||
.flat_map(|bundles| bundles.values().flat_map(Bundle::events));
|
||||
let events = entry
|
||||
.timeline
|
||||
.iter()
|
||||
.map(|filtered| &filtered.event)
|
||||
.chain(&entry.required_state)
|
||||
.chain(bundled);
|
||||
for event in events {
|
||||
if let Some(redacted_by) = event.internal_metadata().redacted_by()? {
|
||||
let redacted_by: Arc<str> = Arc::from(redacted_by);
|
||||
if !redaction_ids.contains(&redacted_by) {
|
||||
redaction_ids.push(redacted_by);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(reader
|
||||
.events(&redaction_ids, RedactBehaviour::Redact, false, false)
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|event| (Arc::from(event.event_id()), event))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// `get_room_sync_data`.
|
||||
pub async fn room_sync_data(
|
||||
context: &RoomContext<'_, '_>,
|
||||
input: &RoomInputData,
|
||||
) -> anyhow::Result<RoomComputed> {
|
||||
let reader = context.reader;
|
||||
let request = context.request;
|
||||
let user_id = &*request.user_id;
|
||||
let member = &input.membership;
|
||||
let room = Room {
|
||||
input,
|
||||
room_id: &member.room_id,
|
||||
membership: &member.membership,
|
||||
membership_token: RoomStreamToken::live(member.event_pos.stream),
|
||||
};
|
||||
let room_id = room.room_id;
|
||||
let membership = room.membership;
|
||||
let config = &input.room_sync_config;
|
||||
let prev_config = input.prev_room_sync_config.as_ref();
|
||||
let from = request.from.as_ref();
|
||||
|
||||
// Only a state reset out of the room leaves a membership with no event.
|
||||
let state_reset_out_of_room = member.event_id.is_none();
|
||||
if state_reset_out_of_room && membership != "leave" {
|
||||
bail!("{room_id}: membership {membership} with no event");
|
||||
}
|
||||
|
||||
let TimelineBounds {
|
||||
from_bound,
|
||||
initial,
|
||||
ignore_timeline_bound,
|
||||
} = timeline_bounds(
|
||||
&room.input.room_status,
|
||||
from.map(|from| &from.room_key),
|
||||
input.newly_joined || state_reset_out_of_room,
|
||||
prev_config,
|
||||
config,
|
||||
);
|
||||
|
||||
// The timeline. Invites and knocks only have stripped state.
|
||||
let mut timeline: Vec<FilteredEvent> = Vec::new();
|
||||
// The timeline's events as loaded, before visibility.
|
||||
let mut loaded_timeline: Vec<Arc<Event>> = Vec::new();
|
||||
let mut bundles = None;
|
||||
let mut limited = None;
|
||||
let mut prev_batch = None;
|
||||
let mut live = None;
|
||||
if config.timeline_limit > 0 && !matches!(membership, "invite" | "knock") {
|
||||
let to_bound = if room.has_left() {
|
||||
&room.membership_token
|
||||
} else {
|
||||
&request.to_room_key
|
||||
};
|
||||
let timeline_from_bound = if ignore_timeline_bound {
|
||||
None
|
||||
} else {
|
||||
from_bound.as_ref()
|
||||
};
|
||||
// A historical section of the timeline in topological order, as
|
||||
// `/messages` returns it, or the updates since the bound in the
|
||||
// order the server received them.
|
||||
let order = if timeline_from_bound.is_none() {
|
||||
PaginationOrder::Topological
|
||||
} else {
|
||||
PaginationOrder::Stream
|
||||
};
|
||||
let (mut events, new_room_key, page_limited) = reader
|
||||
.paginate_room_events(
|
||||
room_id,
|
||||
to_bound,
|
||||
timeline_from_bound,
|
||||
Direction::Backwards,
|
||||
config.timeline_limit,
|
||||
order,
|
||||
)
|
||||
.await?;
|
||||
events.reverse();
|
||||
loaded_timeline = events.into_iter().map(Arc::new).collect();
|
||||
timeline = filter_events_for_client(
|
||||
reader,
|
||||
&context.client_filter,
|
||||
&loaded_timeline,
|
||||
&HashSet::new(),
|
||||
membership != "join",
|
||||
context.now_ms,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let mut positions = Vec::new();
|
||||
if from.is_some() {
|
||||
for filtered in &timeline {
|
||||
let metadata = filtered.event.internal_metadata();
|
||||
let stream = metadata.stream_ordering().with_context(|| {
|
||||
format!("event {} has no stream ordering", filtered.event.event_id())
|
||||
})?;
|
||||
let instance_name = metadata.instance_name().with_context(|| {
|
||||
format!("event {} has no instance name", filtered.event.event_id())
|
||||
})?;
|
||||
positions.push((instance_name, stream.get()));
|
||||
}
|
||||
}
|
||||
live = Some(num_live(
|
||||
positions
|
||||
.iter()
|
||||
.map(|(instance_name, stream)| (instance_name.as_str(), *stream)),
|
||||
from.map(|from| &from.room_key),
|
||||
));
|
||||
|
||||
// With a limited timeline, the client lacks the events it would
|
||||
// need to aggregate relations itself.
|
||||
if page_limited {
|
||||
bundles = Some(
|
||||
bundled_aggregations(
|
||||
reader,
|
||||
&timeline,
|
||||
user_id,
|
||||
&context.client_filter.ignored_users,
|
||||
)
|
||||
.await?,
|
||||
);
|
||||
}
|
||||
limited = Some(page_limited);
|
||||
|
||||
let room_key = new_room_key
|
||||
.to_string_with_instances(|name| context.instance_ids.get(name).copied())
|
||||
.map_err(|e| anyhow::anyhow!("{room_id}: prev_batch: {e}"))?;
|
||||
prev_batch = Some(format!("{room_key}_{}", request.to_suffix));
|
||||
}
|
||||
|
||||
// Python holds no reference to a redacted event, which is loaded again.
|
||||
let held: HashMap<&str, &Arc<Event>> = loaded_timeline
|
||||
.iter()
|
||||
.filter(|event| matches!(event.internal_metadata().redacted_by(), Ok(None)))
|
||||
.map(|event| (event.event_id(), event))
|
||||
.collect();
|
||||
|
||||
// Invites and knocks carry the stripped state the potential joiner
|
||||
// identifies the room by.
|
||||
let mut stripped = Vec::new();
|
||||
if matches!(membership, "invite" | "knock") {
|
||||
let event_id = member
|
||||
.event_id
|
||||
.as_ref()
|
||||
.with_context(|| format!("{room_id}: {membership} with no event"))?;
|
||||
stripped = stripped_state_of(&one_event(reader, event_id).await?)?;
|
||||
}
|
||||
|
||||
// The changes to the current state in the token range, in the order of
|
||||
// Python's dict.
|
||||
let mut deltas: IndexMap<StateKey, Arc<str>> = IndexMap::new();
|
||||
let mut membership_changed = false;
|
||||
let mut name_changed = false;
|
||||
let mut avatar_changed = false;
|
||||
if !initial {
|
||||
let from_bound = from_bound
|
||||
.as_ref()
|
||||
.with_context(|| format!("{room_id}: incremental with no from bound"))?;
|
||||
|
||||
// An out-of-band leave, such as a rejected remote invite, is not in
|
||||
// the state deltas: it is passed in as newly left.
|
||||
if input.newly_left {
|
||||
if let Some(event_id) = &member.event_id {
|
||||
membership_changed = true;
|
||||
let leave = one_event(reader, event_id).await?;
|
||||
if let Some(state_key) = leave.state_key() {
|
||||
deltas.insert(
|
||||
StateKey::new(leave.type_(), state_key),
|
||||
Arc::clone(event_id),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let to_bound = match membership {
|
||||
"leave" | "ban" => Some(&room.membership_token),
|
||||
"join" => Some(&request.to_room_key),
|
||||
"invite" | "knock" => None,
|
||||
other => bail!("Unexpected membership {other} that we don't know how to handle yet"),
|
||||
};
|
||||
let state_deltas = match to_bound {
|
||||
Some(to_bound) => {
|
||||
reader
|
||||
.state_deltas_by_stream_id(room_id, Some(from_bound), Some(to_bound))
|
||||
.await?
|
||||
}
|
||||
None => Vec::new(),
|
||||
};
|
||||
for delta in state_deltas {
|
||||
// A state reset leaves a delta with no event, which is not
|
||||
// handled yet.
|
||||
if let Some(event_id) = &delta.event_id {
|
||||
deltas.insert(
|
||||
StateKey(Arc::clone(&delta.event_type), Arc::clone(&delta.state_key)),
|
||||
Arc::clone(event_id),
|
||||
);
|
||||
}
|
||||
if &*delta.event_type == MEMBER {
|
||||
membership_changed = true;
|
||||
} else if &*delta.event_type == M_ROOM_NAME && delta.state_key.is_empty() {
|
||||
name_changed = true;
|
||||
} else if &*delta.event_type == M_ROOM_AVATAR && delta.state_key.is_empty() {
|
||||
avatar_changed = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let mut room_name = None;
|
||||
if initial || name_changed {
|
||||
let name_state = state_at(
|
||||
context,
|
||||
&room,
|
||||
&held,
|
||||
&StateFilter::from_types([(M_ROOM_NAME, Some(""))]),
|
||||
)
|
||||
.await?;
|
||||
room_name =
|
||||
lookup(&name_state, (M_ROOM_NAME, "")).and_then(|event| content_str(event, "name"));
|
||||
}
|
||||
|
||||
// Heroes, when the room has no name, on the first send or when the
|
||||
// membership changed. The summary also gives the member counts.
|
||||
let mut summary: Option<Arc<RoomSummary>> = None;
|
||||
let mut hero_user_ids: Vec<Arc<str>> = Vec::new();
|
||||
if needs_heroes(room_name.as_deref(), initial, membership_changed) {
|
||||
// The summary of a room left, invited to or knocked on is not known.
|
||||
let details = if membership == "join" {
|
||||
reader.room_summary(room_id).await?
|
||||
} else {
|
||||
Arc::new(RoomSummary::new())
|
||||
};
|
||||
hero_user_ids = extract_heroes_from_room_summary(&details, user_id);
|
||||
summary = Some(details);
|
||||
}
|
||||
|
||||
let mut joined_count = None;
|
||||
let mut invited_count = None;
|
||||
if (initial || membership_changed) && membership == "join" {
|
||||
let counts = match &summary {
|
||||
Some(details) => {
|
||||
let count = |m: &str| details.get(m).map_or(0, |summary| summary.count);
|
||||
(count("join"), count("invite"))
|
||||
}
|
||||
None => {
|
||||
let counts = reader.cached_member_counts(room_id).await?;
|
||||
let count = |m: &str| counts.get(m).copied().unwrap_or(0);
|
||||
(count("join"), count("invite"))
|
||||
}
|
||||
};
|
||||
joined_count = Some(counts.0);
|
||||
invited_count = Some(counts.1);
|
||||
}
|
||||
|
||||
// The state the request asks for, from the timeline and the deltas
|
||||
// before any are added below.
|
||||
let delta_keys: Vec<(String, String)> = deltas
|
||||
.keys()
|
||||
.map(|key| (key.0.to_string(), key.1.to_string()))
|
||||
.collect();
|
||||
let timeline_members: Vec<TimelineMember<'_>> = timeline
|
||||
.iter()
|
||||
.map(|filtered| TimelineMember {
|
||||
sender: filtered.event.sender(),
|
||||
event_type: filtered.event.type_(),
|
||||
state_key: filtered.event.state_key(),
|
||||
})
|
||||
.collect();
|
||||
let state_request = required_state_request(
|
||||
user_id,
|
||||
membership,
|
||||
&config.required_state_map,
|
||||
&timeline_members,
|
||||
limited,
|
||||
initial,
|
||||
delta_keys.iter().map(|(t, k)| (t.as_str(), k.as_str())),
|
||||
);
|
||||
drop(timeline_members);
|
||||
let required_state_filter = &state_request.required_state_filter;
|
||||
let lazy_load_user_ids = &state_request.lazy_load_user_ids;
|
||||
|
||||
let state_filter = state_filter_with_meta(
|
||||
required_state_filter,
|
||||
&hero_user_ids,
|
||||
initial || avatar_changed,
|
||||
);
|
||||
|
||||
// Every lazy-loaded member counts as sent first in this request, until
|
||||
// the connection says otherwise.
|
||||
let mut returned: HashMap<String, Option<i64>> = if state_request.lazy_load_room_members {
|
||||
lazy_load_user_ids
|
||||
.iter()
|
||||
.map(|user_id| (user_id.clone(), None))
|
||||
.collect()
|
||||
} else {
|
||||
HashMap::new()
|
||||
};
|
||||
let mut invalidated = HashSet::new();
|
||||
let mut changed_required_state_map = None;
|
||||
let mut lazy_member_lookup = false;
|
||||
|
||||
let room_state: HashMap<StateKey, Event>;
|
||||
// The heroes' memberships, when not in `room_state`.
|
||||
let mut hero_state: Option<HashMap<StateKey, Event>> = None;
|
||||
if initial {
|
||||
room_state = state_at(context, &room, &held, &state_filter).await?;
|
||||
} else {
|
||||
let from = from.with_context(|| format!("{room_id}: incremental with no from token"))?;
|
||||
if let Some(prev_config) = prev_config {
|
||||
// The members whose state may be sent: explicit, lazy-loaded,
|
||||
// and those whose membership changed.
|
||||
let mut all_required_user_state: HashSet<&str> = state_request
|
||||
.explicit_user_state
|
||||
.iter()
|
||||
.chain(lazy_load_user_ids)
|
||||
.map(String::as_str)
|
||||
.collect();
|
||||
for (state_type, state_key) in &delta_keys {
|
||||
if state_type == MEMBER {
|
||||
all_required_user_state.insert(state_key);
|
||||
}
|
||||
}
|
||||
|
||||
let mut previously_returned: HashSet<String> = HashSet::new();
|
||||
if !all_required_user_state.is_empty() {
|
||||
lazy_member_lookup = true;
|
||||
let user_ids: Vec<Arc<str>> = all_required_user_state
|
||||
.iter()
|
||||
.map(|u| Arc::from(*u))
|
||||
.collect();
|
||||
let last_seen = reader
|
||||
.connection_lazy_members(from.connection_position, room_id, &user_ids)
|
||||
.await?;
|
||||
for (user_id, last_seen_ts) in last_seen {
|
||||
if lazy_load_user_ids.contains(&*user_id) {
|
||||
returned.insert(user_id.to_string(), Some(last_seen_ts));
|
||||
}
|
||||
previously_returned.insert(user_id.to_string());
|
||||
}
|
||||
}
|
||||
|
||||
let changes = required_state_changes(
|
||||
user_id,
|
||||
&prev_config.required_state_map,
|
||||
&config.required_state_map,
|
||||
&previously_returned,
|
||||
lazy_load_user_ids,
|
||||
&delta_keys,
|
||||
);
|
||||
changed_required_state_map = changes.changed_required_state_map;
|
||||
invalidated = changes.lazy_members_invalidated;
|
||||
// Explicit members converted to lazy ones: when they were last
|
||||
// sent is not known, so a timestamp already there is kept.
|
||||
for user_id in changes.extra_users_to_add_to_lazy_cache {
|
||||
returned.entry(user_id).or_insert(None);
|
||||
}
|
||||
if changes.added_state_filter != StateFilter::none() {
|
||||
// State the request newly asks for, on top of the deltas.
|
||||
let added = state_ids_at(context, &room, &changes.added_state_filter).await?;
|
||||
deltas.extend(added);
|
||||
}
|
||||
}
|
||||
|
||||
let delta_state: StateMap = deltas
|
||||
.iter()
|
||||
.map(|(key, event_id)| (key.clone(), Arc::clone(event_id)))
|
||||
.collect();
|
||||
let event_ids: Vec<Arc<str>> = state_filter
|
||||
.filter_state(&delta_state)
|
||||
.into_values()
|
||||
.collect();
|
||||
room_state = state_events(reader, &held, &event_ids).await?;
|
||||
|
||||
if !hero_user_ids.is_empty() {
|
||||
let hero_filter =
|
||||
StateFilter::from_types(hero_user_ids.iter().map(|u| (MEMBER, Some(&**u))));
|
||||
hero_state = Some(state_at(context, &room, &held, &hero_filter).await?);
|
||||
}
|
||||
}
|
||||
|
||||
let room_avatar =
|
||||
lookup(&room_state, (M_ROOM_AVATAR, "")).and_then(|event| content_str(event, "url"));
|
||||
|
||||
let hero_membership_state = hero_state.as_ref().unwrap_or(&room_state);
|
||||
let heroes: Vec<StrippedHero> = hero_user_ids
|
||||
.iter()
|
||||
.filter_map(|hero| {
|
||||
let event = lookup(hero_membership_state, (MEMBER, hero))?;
|
||||
Some(StrippedHero {
|
||||
user_id: hero.to_string(),
|
||||
display_name: content_str(event, "displayname"),
|
||||
avatar_url: content_str(event, "avatar_url"),
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
|
||||
let mut required_state: Vec<Event> = Vec::new();
|
||||
if *required_state_filter != StateFilter::none() {
|
||||
let mut keys: Vec<&StateKey> = room_state
|
||||
.keys()
|
||||
.filter(|key| required_state_filter.matches(&key.0, &key.1))
|
||||
.collect();
|
||||
keys.sort();
|
||||
required_state = keys
|
||||
.into_iter()
|
||||
.map(|key| room_state[key].shallow_clone())
|
||||
.collect();
|
||||
}
|
||||
|
||||
// The bump stamp: the last bump event for a joined room, else the
|
||||
// membership's position if the client needs one.
|
||||
let always_return = always_return_bump_stamp(membership, limited, initial);
|
||||
let new_bump_stamp = if membership == "join" {
|
||||
bump_stamp(context, room_id, &timeline, always_return).await?
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let bump_stamp = final_bump_stamp(new_bump_stamp, always_return, member.event_pos.stream);
|
||||
|
||||
let room_config = room_config_to_persist(
|
||||
config,
|
||||
prev_config,
|
||||
changed_required_state_map.as_ref(),
|
||||
ignore_timeline_bound,
|
||||
limited,
|
||||
);
|
||||
|
||||
let mut entry = RoomEntry {
|
||||
room_id: Arc::clone(&member.room_id),
|
||||
name: room_name,
|
||||
avatar: room_avatar,
|
||||
heroes,
|
||||
is_dm: input.is_dm,
|
||||
initial,
|
||||
unstable_expanded_timeline: ignore_timeline_bound,
|
||||
required_state,
|
||||
timeline,
|
||||
bundles,
|
||||
redactions: HashMap::new(),
|
||||
stripped_state: stripped,
|
||||
prev_batch,
|
||||
limited,
|
||||
num_live: live,
|
||||
bump_stamp,
|
||||
joined_count,
|
||||
invited_count,
|
||||
};
|
||||
entry.redactions = load_redactions(reader, &entry).await?;
|
||||
|
||||
Ok(RoomComputed {
|
||||
entry,
|
||||
mutations: RoomMutations {
|
||||
room_config,
|
||||
lazy_membership: RoomLazyMembershipChanges {
|
||||
returned_user_id_to_last_seen_ts_map: returned,
|
||||
invalidated_user_ids: invalidated,
|
||||
},
|
||||
},
|
||||
lazy_member_lookup,
|
||||
})
|
||||
}
|
||||
|
||||
/// One room's outcome. A panic is this room's error, not the request's.
|
||||
async fn room_outcome(
|
||||
context: &RoomContext<'_, '_>,
|
||||
input: &RoomInputData,
|
||||
) -> (Arc<str>, RoomOutcome) {
|
||||
let outcome = match AssertUnwindSafe(room_sync_data(context, input))
|
||||
.catch_unwind()
|
||||
.await
|
||||
{
|
||||
Ok(Ok(computed)) => RoomOutcome::Computed(Box::new(computed)),
|
||||
Ok(Err(err)) => match err.downcast::<Fallback>() {
|
||||
Ok(fallback) => RoomOutcome::Fallback(fallback.0),
|
||||
Err(err) => RoomOutcome::Error(format!("{err:#}")),
|
||||
},
|
||||
Err(panic) => RoomOutcome::Error(format!("panicked: {}", panic_message(&*panic))),
|
||||
};
|
||||
(Arc::clone(&input.membership.room_id), outcome)
|
||||
}
|
||||
|
||||
/// The entries of `rooms`, twenty at a time, in the order they finish.
|
||||
pub async fn generate_room_entries(
|
||||
reader: &SlidingSyncReader<'_>,
|
||||
request: &SlidingSyncRequestData,
|
||||
instance_ids: &HashMap<String, i64>,
|
||||
now_ms: i64,
|
||||
rooms: &[RoomInputData],
|
||||
) -> anyhow::Result<Vec<(Arc<str>, RoomOutcome)>> {
|
||||
let ignored_users = reader.ignored_users(&request.user_id).await?;
|
||||
let context = RoomContext {
|
||||
reader,
|
||||
request,
|
||||
client_filter: ClientFilterContext {
|
||||
user_id: Arc::clone(&request.user_id),
|
||||
ignored_users: Arc::new(ignored_users),
|
||||
admin_policy: request.admin_policy,
|
||||
msc4354_enabled: request.msc4354_enabled,
|
||||
},
|
||||
instance_ids,
|
||||
now_ms,
|
||||
};
|
||||
let outcomes: Vec<_> = rooms
|
||||
.iter()
|
||||
.map(|room| room_outcome(&context, room))
|
||||
.collect();
|
||||
Ok(stream::iter(outcomes)
|
||||
.buffer_unordered(ROOM_ENTRY_CONCURRENCY)
|
||||
.collect()
|
||||
.await)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn empty_entry() -> RoomEntry {
|
||||
RoomEntry {
|
||||
room_id: Arc::from("!r:test"),
|
||||
name: None,
|
||||
avatar: None,
|
||||
heroes: Vec::new(),
|
||||
is_dm: true,
|
||||
initial: false,
|
||||
unstable_expanded_timeline: true,
|
||||
required_state: Vec::new(),
|
||||
timeline: Vec::new(),
|
||||
bundles: None,
|
||||
redactions: HashMap::new(),
|
||||
stripped_state: Vec::new(),
|
||||
prev_batch: Some("s1_0".to_owned()),
|
||||
limited: Some(true),
|
||||
num_live: Some(0),
|
||||
bump_stamp: Some(5),
|
||||
joined_count: None,
|
||||
invited_count: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn what_counts_as_an_update() {
|
||||
// As in Python: a DM flag, an expanded timeline, a token or a bump
|
||||
// stamp alone are not updates.
|
||||
assert!(!empty_entry().has_updates());
|
||||
// An empty name is, though it is not sent.
|
||||
let entry = RoomEntry {
|
||||
name: Some(String::new()),
|
||||
..empty_entry()
|
||||
};
|
||||
assert!(entry.has_updates());
|
||||
let entry = RoomEntry {
|
||||
joined_count: Some(0),
|
||||
..empty_entry()
|
||||
};
|
||||
assert!(entry.has_updates());
|
||||
let entry = RoomEntry {
|
||||
initial: true,
|
||||
..empty_entry()
|
||||
};
|
||||
assert!(entry.has_updates());
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user