diff --git a/rust/src/events/internal_metadata.rs b/rust/src/events/internal_metadata.rs index 9c5cbd49b4..a86df7eba7 100644 --- a/rust/src/events/internal_metadata.rs +++ b/rust/src/events/internal_metadata.rs @@ -653,7 +653,7 @@ impl EventInternalMetadata { Ok(self.read_inner()?.is_outlier()) } - fn is_out_of_band_membership(&self) -> PyResult { + pub fn is_out_of_band_membership(&self) -> PyResult { Ok(self.read_inner()?.is_out_of_band_membership()) } diff --git a/rust/src/storage/sync_store.rs b/rust/src/storage/sync_store.rs index 59c41d7e5e..e3b3ffb8e4 100644 --- a/rust/src/storage/sync_store.rs +++ b/rust/src/storage/sync_store.rs @@ -987,6 +987,30 @@ impl<'a> SyncReader<'a> { .await } + /// `get_position_for_event`: where the event was persisted, or `None` + /// if it is not in the database. + pub async fn event_position( + &self, + event_id: &str, + ) -> anyhow::Result> { + let rows = self + .query_one_shot( + "SELECT instance_name, stream_ordering FROM events WHERE event_id = $1", + &[&event_id], + ) + .await?; + let Some(row) = rows.first() else { + return Ok(None); + }; + let instance_name: Option = row.try_get(0)?; + let stream: i64 = row.try_get(1)?; + Ok(Some(PersistedEventPosition { + // A null instance name is a row from before sharded persisters. + instance_name: Arc::from(instance_name.as_deref().unwrap_or("master")), + stream, + })) + } + /// `get_last_event_pos_in_room_before_stream_ordering`: the last event /// in the room at or before `end_token`, and where it was persisted. pub async fn last_event_before( diff --git a/rust/src/sync/mod.rs b/rust/src/sync/mod.rs index e35ba756e1..46d963f846 100644 --- a/rust/src/sync/mod.rs +++ b/rust/src/sync/mod.rs @@ -16,9 +16,9 @@ //! //! [`SyncHandler`] is the Python entry point, used by //! `synapse.handlers.sync.SyncHandler` when `rust_sync` is on. It computes -//! the per-room part of a sync response, one room at a time, and raises -//! [`RustSyncFallback`] for a room it cannot handle, which Python then -//! computes itself. +//! the `rooms` section of a sync response ([`rooms`]): which rooms to +//! report, then each joined or archived room's entry ([`room`]). It leaves +//! a room it cannot handle to Python, which computes it itself. //! //! 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/`: @@ -28,22 +28,14 @@ //! [`crate::visibility`] and [`crate::types::stream_token`] are shared with //! later ports. -use std::future::Future; -use std::pin::Pin; use std::sync::{Arc, Mutex}; -use pyo3::{ - create_exception, - exceptions::{PyException, PyRuntimeError}, - intern, - prelude::*, - types::PyModule, -}; +use futures::future::BoxFuture; +use pyo3::{exceptions::PyRuntimeError, intern, prelude::*, types::PyModule}; use crate::caches::lock_ignoring_poison; -use crate::clock::Clock; use crate::config::SynapseHomeServerConfig; -use crate::database::postgres::pool::PyConnectionPool; +use crate::database::postgres::pool::{ConnectionPool, PyConnectionPool}; use crate::deferred::{create_deferred_with, OnCancel}; use crate::runtime::RustRuntime; use crate::storage::store::Store; @@ -56,22 +48,14 @@ pub mod ll_members; pub mod request; pub mod room; pub mod room_state; +pub mod rooms; pub mod state_delta; pub mod summary; pub mod timeline; -use client_events::ClientFilterContext; use ll_members::LazyLoadedMembersCaches; -use request::{RoomSyncInput, RoomSyncResult, SyncRequest}; -use room::{Fallback, RoomContext}; - -create_exception!( - synapse.synapse_rust.sync, - RustSyncFallback, - PyException, - "The Rust sync handler cannot compute this room's entry. Compute it with \ - the Python implementation." -); +use request::{RoomSyncResult, SyncRequest}; +use rooms::{RoomChanges, RoomsInput, RoomsResult}; /// Computes sync responses with the Rust implementation, reading the /// database through the native Postgres driver. @@ -89,9 +73,6 @@ pub struct SyncHandler { ll_members: Arc>, } -/// One room's entry, started. See [`SyncHandler::start_room`]. -type RoomFuture = Pin>> + Send>>; - impl SyncHandler { pub fn new( db_pool: Py, @@ -115,60 +96,97 @@ impl SyncHandler { &self.store } - /// Start computing `room`'s entry: read what it needs from Python, and - /// return the future that computes it off the GIL. - fn start_room( - &self, - py: Python<'_>, - request: &Bound<'_, SyncRequest>, - room: &Bound<'_, RoomSyncInput>, - ) -> PyResult { + /// The connection pool to read through, or an error if it is closed. + fn pool(&self, py: Python<'_>) -> PyResult { let pool = self.db_pool.bind(py).getattr(intern!(py, "_pool"))?; if pool.is_none() { return Err(PyRuntimeError::new_err("the database pool is closed")); } - let pool = pool.cast::()?.get().pool.clone(); + Ok(pool.cast::()?.get().pool.clone()) + } + /// Start working out the request's room changes: read what it needs + /// from Python, and return the future that does it off the GIL. + fn start_room_changes( + &self, + py: Python<'_>, + request: &Bound<'_, SyncRequest>, + input: &Bound<'_, RoomsInput>, + ) -> PyResult>> { + let pool = self.pool(py)?; let request = Arc::clone(&request.get().data); - let room = Arc::clone(&room.get().data); + let input = input.get().take()?; let store = Arc::clone(&self.store); - let ll_members = Arc::clone(&self.ll_members); - Ok(Box::pin(async move { let reader = store.reader(pool); - let context = RoomContext { - reader: &reader, - request: &request, - client_filter: ClientFilterContext { - user_id: Arc::clone(&request.user_id), - ignored_users: Arc::clone(&request.ignored_users), - admin_policy: request.admin_policy, - msc4354_enabled: request.msc4354_enabled, - }, - ll_members: &ll_members, - ll_pending: Mutex::default(), - clock: Clock::new(), - }; - match room::generate_room_entry(&context, &room).await { - Ok(result) => { - context.commit_ll_members(); - Ok(result.map(|data| RoomSyncResult { data })) - } - Err(err) => Err(room_error_to_py(err, &room.room_id)), - } + let data = rooms::room_changes(&reader, &request, input) + .await + .map_err(anyhow_to_py)?; + Ok(RoomChanges { + data: Arc::new(data), + }) })) } + + /// Start computing the entries of `changes`' rooms: read each room's + /// extras from Python, and return the future that computes them off the + /// GIL. + #[allow(clippy::too_many_arguments)] + fn start_room_entries( + &self, + py: Python<'_>, + request: &Bound<'_, SyncRequest>, + changes: &Bound<'_, RoomChanges>, + ephemeral_by_room: &Bound<'_, PyAny>, + tags_by_room: &Bound<'_, PyAny>, + account_data_by_room: &Bound<'_, PyAny>, + sticky_by_room: &Bound<'_, PyAny>, + ) -> PyResult>> { + let pool = self.pool(py)?; + let request = Arc::clone(&request.get().data); + let changes = Arc::clone(&changes.get().data); + let rooms = changes + .entries + .iter() + .map(|plan| { + let extras = rooms::RoomExtras::from_py( + &plan.room_id, + ephemeral_by_room, + tags_by_room, + account_data_by_room, + sticky_by_room, + )?; + Ok(rooms::room_input(&changes, plan, extras)) + }) + .collect::>>()?; + let store = Arc::clone(&self.store); + let ll_members = Arc::clone(&self.ll_members); + Ok(Box::pin(async move { + let reader = store.reader(pool); + let outcomes = + rooms::generate_room_entries(&reader, &request, &ll_members, rooms).await; + Ok(RoomsResult::new(&changes, outcomes)) + })) + } + + /// Run `future` to completion on the runtime, releasing the GIL. For the + /// `_blocking` entry points. + fn block_on( + &self, + py: Python<'_>, + future: BoxFuture<'static, PyResult>, + ) -> PyResult { + let handle = self.runtime.tokio_handle()?; + py.detach(|| handle.block_on(future)) + } } -/// A failed room as Python sees it: [`RustSyncFallback`] for a room it -/// should compute itself, an error otherwise. -fn room_error_to_py(err: anyhow::Error, room_id: &str) -> PyErr { - match err.downcast::() { - Ok(fallback) => RustSyncFallback::new_err(fallback.0), - Err(err) => match err.downcast::() { - Ok(err) => err, - Err(err) => PyRuntimeError::new_err(format!("syncing room {room_id}: {err:#}")), - }, +/// An error as Python sees it: the Python exception it wraps, or a +/// `RuntimeError`. +fn anyhow_to_py(err: anyhow::Error) -> PyErr { + match err.downcast::() { + Ok(err) => err, + Err(err) => PyRuntimeError::new_err(format!("{err:#}")), } } @@ -225,34 +243,83 @@ impl SyncHandler { .mark_sent(member, event_id, &next_batch); } - /// `_generate_room_entry` for one joined or archived room. - /// - /// Returns a deferred resolving to a `RoomSyncResult`, or to `None` - /// when the room has nothing to send. Fails with `RustSyncFallback` for - /// a room the Python implementation must compute. Cancelling the - /// deferred stops the computation. - fn generate_room_entry<'py>( + /// `_get_room_changes_for_incremental_sync` or + /// `_get_room_changes_for_initial_sync`: a deferred resolving to a + /// `RoomChanges`. Consumes `input`. Cancelling the deferred stops the + /// computation. + fn room_changes<'py>( &self, py: Python<'py>, request: &Bound<'py, SyncRequest>, - room: &Bound<'py, RoomSyncInput>, + input: &Bound<'py, RoomsInput>, ) -> PyResult> { - let future = self.start_room(py, request, room)?; + let future = self.start_room_changes(py, request, input)?; create_deferred_with(py, &self.runtime, future, OnCancel::Abort) } - /// As [`Self::generate_room_entry`], but blocks (releasing the GIL) and - /// returns the result directly. For tests: see + /// As [`Self::room_changes`], but blocks. For tests: see /// `_BlockingRustSyncHandler`. - fn generate_room_entry_blocking( + fn room_changes_blocking( &self, py: Python<'_>, request: &Bound<'_, SyncRequest>, - room: &Bound<'_, RoomSyncInput>, - ) -> PyResult> { - let future = self.start_room(py, request, room)?; - let handle = self.runtime.tokio_handle()?; - py.detach(|| handle.block_on(future)) + input: &Bound<'_, RoomsInput>, + ) -> PyResult { + let future = self.start_room_changes(py, request, input)?; + self.block_on(py, future) + } + + /// The entries of the joined and archived rooms of `changes`, ten at a + /// time: a deferred resolving to a `RoomsResult`. The `*_by_room` + /// mappings are what `handle_room_entries` reads for each room. + /// Cancelling the deferred stops the computation. + #[pyo3(signature = (request, changes, *, ephemeral_by_room, tags_by_room, account_data_by_room, sticky_by_room))] + #[allow(clippy::too_many_arguments)] + fn generate_room_entries<'py>( + &self, + py: Python<'py>, + request: &Bound<'py, SyncRequest>, + changes: &Bound<'py, RoomChanges>, + ephemeral_by_room: &Bound<'py, PyAny>, + tags_by_room: &Bound<'py, PyAny>, + account_data_by_room: &Bound<'py, PyAny>, + sticky_by_room: &Bound<'py, PyAny>, + ) -> PyResult> { + let future = self.start_room_entries( + py, + request, + changes, + ephemeral_by_room, + tags_by_room, + account_data_by_room, + sticky_by_room, + )?; + create_deferred_with(py, &self.runtime, future, OnCancel::Abort) + } + + /// As [`Self::generate_room_entries`], but blocks. + #[pyo3(signature = (request, changes, *, ephemeral_by_room, tags_by_room, account_data_by_room, sticky_by_room))] + #[allow(clippy::too_many_arguments)] + fn generate_room_entries_blocking( + &self, + py: Python<'_>, + request: &Bound<'_, SyncRequest>, + changes: &Bound<'_, RoomChanges>, + ephemeral_by_room: &Bound<'_, PyAny>, + tags_by_room: &Bound<'_, PyAny>, + account_data_by_room: &Bound<'_, PyAny>, + sticky_by_room: &Bound<'_, PyAny>, + ) -> PyResult { + let future = self.start_room_entries( + py, + request, + changes, + ephemeral_by_room, + tags_by_room, + account_data_by_room, + sticky_by_room, + )?; + self.block_on(py, future) } } @@ -263,12 +330,13 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> child_module.add_class::()?; child_module.add_class::()?; child_module.add_class::()?; - child_module.add_class::()?; child_module.add_class::()?; + child_module.add_class::()?; + child_module.add_class::()?; + child_module.add_class::()?; state_delta::register_test_functions(py, &child_module)?; summary::register_test_functions(py, &child_module)?; crate::storage::sync_store::register_test_classes(&child_module)?; - child_module.add("RustSyncFallback", py.get_type::())?; m.add_submodule(&child_module)?; diff --git a/rust/src/sync/request.rs b/rust/src/sync/request.rs index 6b35833cc7..63bb97cdbb 100644 --- a/rust/src/sync/request.rs +++ b/rust/src/sync/request.rs @@ -12,10 +12,10 @@ * . */ -//! The types that cross the Python boundary for one room's entry: the -//! request-wide [`SyncRequest`], the per-room [`RoomSyncInput`] and the -//! [`RoomSyncResult`] Python turns into a `JoinedSyncResult` or -//! `ArchivedSyncResult`. +//! The types that cross the Python boundary for a room's entry: the +//! request-wide [`SyncRequest`], and the [`RoomSyncResult`] Python turns +//! into a `JoinedSyncResult` or `ArchivedSyncResult`. A room's input, +//! [`RoomSyncInputData`], is built in Rust by [`crate::sync::rooms`]. use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -24,7 +24,7 @@ use pyo3::{ prelude::*, types::{PyDict, PyList, PyTuple}, }; -use pythonize::{depythonize, pythonize}; +use pythonize::pythonize; use serde_json::Value; use crate::events::Event; @@ -103,6 +103,14 @@ pub enum RoomType { Archived, } +/// The `rtype` of a `RoomSyncResultBuilder`. +pub fn rtype_str(rtype: RoomType) -> &'static str { + match rtype { + RoomType::Joined => "joined", + RoomType::Archived => "archived", + } +} + /// One room's `RoomSyncResultBuilder`, plus what `handle_room_entries` /// passes alongside it. pub struct RoomSyncInputData { @@ -124,84 +132,6 @@ pub struct RoomSyncInputData { pub account_data: Vec<(String, Value)>, pub sticky_event_ids: Vec>, pub always_include: bool, - /// `get_room_max_token()` when Python dispatched the room. - pub room_max_token: RoomStreamToken, -} - -/// See [`RoomSyncInputData`]. -#[pyclass(frozen, skip_from_py_object)] -pub struct RoomSyncInput { - pub(crate) data: Arc, -} - -#[pymethods] -impl RoomSyncInput { - #[new] - #[pyo3(signature = (*, room_id, rtype, events, newly_joined, full_state, since, upto, end, out_of_band, ephemeral, tags, account_data, sticky_event_ids, always_include, room_max_token))] - #[allow(clippy::too_many_arguments)] - fn py_new( - room_id: &str, - rtype: &str, - events: Option>>, - newly_joined: bool, - full_state: bool, - since: Option, - upto: RoomStreamToken, - end: RoomStreamToken, - out_of_band: bool, - ephemeral: Vec>, - tags: Option>, - account_data: &Bound<'_, PyAny>, - sticky_event_ids: Vec, - always_include: bool, - room_max_token: RoomStreamToken, - ) -> PyResult { - let rtype = match rtype { - "joined" => RoomType::Joined, - "archived" => RoomType::Archived, - other => { - return Err(pyo3::exceptions::PyValueError::new_err(format!( - "Unrecognized rtype: {other:?}" - ))) - } - }; - let py = account_data.py(); - let mut account_data_items = Vec::new(); - for item in account_data - .call_method0(pyo3::intern!(py, "items"))? - .try_iter()? - { - let (account_data_type, content): (String, Bound<'_, PyAny>) = item?.extract()?; - account_data_items.push((account_data_type, depythonize(&content)?)); - } - Ok(RoomSyncInput { - data: Arc::new(RoomSyncInputData { - room_id: Arc::from(room_id), - rtype, - events: events.map(|events| { - events - .iter() - .map(|event| Arc::new(event.get().shallow_clone())) - .collect() - }), - newly_joined, - full_state, - since, - upto, - end, - out_of_band, - ephemeral: ephemeral - .iter() - .map(|event| depythonize(event)) - .collect::>()?, - tags: tags.map(|tags| depythonize(&tags)).transpose()?, - account_data: account_data_items, - sticky_event_ids: sticky_event_ids.into_iter().map(Arc::from).collect(), - always_include, - room_max_token, - }), - }) - } } /// Where the timeline's `prev_batch` token comes from: the room's upto @@ -268,10 +198,7 @@ impl RoomSyncResult { #[getter] fn rtype(&self) -> &'static str { - match self.data.rtype { - RoomType::Joined => "joined", - RoomType::Archived => "archived", - } + rtype_str(self.data.rtype) } /// The timeline as `(event, membership)` pairs, oldest first. diff --git a/rust/src/sync/room_state.rs b/rust/src/sync/room_state.rs index 44252a4dc1..6da326acb5 100644 --- a/rust/src/sync/room_state.rs +++ b/rust/src/sync/room_state.rs @@ -215,10 +215,12 @@ async fn state_for_full_sync( if request.use_state_after { if joined { // The current state, rolled back to the end of the timeline. + // The deltas are read with no upper bound, after the current + // state, so that they cover every change the current state may + // hold. Rolling back a change made after it was read is still + // right: it is after the end of the timeline too. let mut state_ids = reader.current_state_ids(room_id, &state_filter).await?; - let deltas = reader - .state_deltas(room_id, Some(&room.end), Some(&room.room_max_token)) - .await?; + let deltas = reader.state_deltas(room_id, Some(&room.end), None).await?; if !deltas.is_empty() { roll_back_state(&mut state_ids, &to_deltas(deltas)); } diff --git a/rust/src/sync/rooms.rs b/rust/src/sync/rooms.rs new file mode 100644 index 0000000000..d1d491e2d5 --- /dev/null +++ b/rust/src/sync/rooms.rs @@ -0,0 +1,927 @@ +/* + * 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 `rooms` section of `_generate_sync_entry_for_rooms`, from the room +//! changes on. +//! +//! [`room_changes`] ports `_get_room_changes_for_incremental_sync` and +//! `_get_room_changes_for_initial_sync`: which rooms the response reports, +//! and how. Python then fetches the sticky events of the newly joined +//! rooms, and [`generate_room_entries`] computes the entry of every joined +//! or archived room, ten at a time, as `concurrently_execute` does. +//! +//! What Python computes before the call stays in Python: the joined rooms +//! and membership changes of `get_sync_result_builder`, the ephemeral +//! events, account data, tags and sticky events of each room, and the +//! no-op check, which reads the in-memory stream change cache. Python also +//! drops the invites of ignored users and those the user's invite rules +//! ignore. + +use std::collections::{HashMap, HashSet}; +use std::panic::AssertUnwindSafe; +use std::sync::{Arc, Mutex}; + +use futures::{stream, FutureExt, StreamExt, TryStreamExt}; +use pyo3::{ + exceptions::{PyKeyError, PyRuntimeError}, + intern, + prelude::*, + types::{PyBool, PyList, PyTuple}, +}; +use pythonize::depythonize; + +use crate::caches::lock_ignoring_poison; +use crate::clock::Clock; +use crate::events::constants::event_type::M_ROOM_MEMBER; +use crate::events::Event; +use crate::state::StateKey; +use crate::storage::events::RedactBehaviour; +use crate::storage::sync_store::{PaginationOrder, SyncReader}; +use crate::sync::client_events::ClientFilterContext; +use crate::sync::ll_members::LazyLoadedMembersCaches; +use crate::sync::request::{ + rtype_str, RoomSyncInputData, RoomSyncResult, RoomSyncResultData, RoomType, SyncRequestData, +}; +use crate::sync::room::{generate_room_entry, Fallback, RoomContext}; +use crate::types::state_filter::StateFilter; +use crate::types::stream_token::{persisted_after, Direction, RoomStreamToken}; +use crate::visibility::membership_of; + +/// How many rooms' timelines are fetched at once, as in +/// `get_room_events_stream_for_rooms`. +const TIMELINE_FETCH_CONCURRENCY: usize = 20; + +/// How many rooms' entries are computed at once, as in +/// `_generate_sync_entry_for_rooms`. +const ROOM_ENTRY_CONCURRENCY: usize = 10; + +/// Which of the request's tokens a room's token is built on. A room's +/// tokens differ from the request's only in their room key. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TokenBase { + Since, + Now, +} + +/// A room's `RoomSyncResultBuilder` token: the request token it is built +/// on, with its room key replaced. +#[derive(Debug, Clone)] +pub struct RoomToken { + pub base: TokenBase, + pub room_key: RoomStreamToken, +} + +/// A joined or archived room to compute the entry of: a +/// `RoomSyncResultBuilder`. +pub struct RoomEntryPlan { + pub room_id: Arc, + pub rtype: RoomType, + /// The events between the since and upto tokens, if fetched. + pub events: Option>>, + pub newly_joined: bool, + pub full_state: bool, + /// Whether the room's `since_token` is the request's, rather than None. + pub has_since: bool, + pub upto: RoomToken, + pub end: RoomToken, + pub out_of_band: bool, +} + +/// One of the user's rooms, as `get_rooms_for_local_user_where_membership_is` +/// returns it: a `RoomsForUser`. +pub struct RoomForUser { + pub room_id: Arc, + pub sender: Arc, + pub membership: Arc, + pub event_id: Arc, + pub stream: i64, +} + +/// What [`room_changes`] reads besides the request, from Python's +/// `SyncResultBuilder`. +pub struct RoomsInputData { + /// The room key of the request's since token, `None` for an initial + /// sync. + pub since: Option, + /// The request's `full_state`. + pub full_state: bool, + pub joined_room_ids: Vec>, + /// The joined rooms that may have events after `since`, by the stream + /// change cache. Incremental syncs only. + pub changed_room_ids: HashSet>, + pub forced_newly_joined_room_ids: Vec>, + /// The user's membership events after `since`, oldest first. + pub membership_change_events: Vec>, + /// The user's rooms, in a known room version and not excluded. Initial + /// syncs only. + pub initial_rooms: Vec, +} + +/// A `_RoomChanges`, less the invites Python drops. +pub struct RoomChangesData { + pub since: Option, + pub full_state: bool, + pub entries: Vec, + /// The index in `entries` of each room. + pub entry_index: HashMap, usize>, + pub invited: Vec<(Arc, Arc)>, + pub knocked: Vec<(Arc, Arc)>, + pub newly_joined_rooms: Vec>, + pub newly_left_rooms: Vec>, +} + +fn entry_index(entries: &[RoomEntryPlan]) -> HashMap, usize> { + entries + .iter() + .enumerate() + .map(|(index, plan)| (Arc::clone(&plan.room_id), index)) + .collect() +} + +/// `_get_room_changes_for_incremental_sync` or +/// `_get_room_changes_for_initial_sync`. +pub async fn room_changes( + reader: &SyncReader<'_>, + request: &SyncRequestData, + input: RoomsInputData, +) -> anyhow::Result { + match input.since.clone() { + Some(since) => incremental_room_changes(reader, request, input, &since).await, + None => initial_room_changes(reader, request, input).await, + } +} + +/// The user's membership in the room at `since`: the event, if any. +async fn membership_at( + reader: &SyncReader<'_>, + room_id: &str, + user_id: &str, + since: &RoomStreamToken, +) -> anyhow::Result> { + let state_filter = StateFilter::from_types([(M_ROOM_MEMBER, Some(user_id))]); + let state = reader.state_ids_at(room_id, since, &state_filter).await?; + let Some(event_id) = state.get(&StateKey::new(M_ROOM_MEMBER, user_id)) else { + return Ok(None); + }; + // `get_event(allow_none=True)`. + let mut events = reader + .events( + std::slice::from_ref(event_id), + RedactBehaviour::Redact, + false, + false, + ) + .await?; + Ok(events.pop()) +} + +fn is_join(event: Option<&Event>) -> bool { + event.and_then(membership_of) == Some("join") +} + +async fn incremental_room_changes( + reader: &SyncReader<'_>, + request: &SyncRequestData, + input: RoomsInputData, + since: &RoomStreamToken, +) -> anyhow::Result { + let user_id = &*request.user_id; + let now = &request.now_room_key; + let joined: HashSet<&str> = input.joined_room_ids.iter().map(|r| &**r).collect(); + + // The membership events by room, rooms in the order of their first. + let mut rooms_in_order: Vec> = Vec::new(); + let mut by_room: HashMap, Vec<&Arc>> = HashMap::new(); + for event in &input.membership_change_events { + let room_id: Arc = Arc::from(event.room_id()); + by_room + .entry(Arc::clone(&room_id)) + .or_insert_with(|| { + rooms_in_order.push(Arc::clone(&room_id)); + Vec::new() + }) + .push(event); + } + + let mut newly_joined_rooms: Vec> = input.forced_newly_joined_room_ids.clone(); + // The same rooms, to look them up. + let mut newly_joined: HashSet> = newly_joined_rooms.iter().cloned().collect(); + let mut newly_left_rooms: Vec> = Vec::new(); + let mut entries: Vec = Vec::new(); + let mut invited = Vec::new(); + let mut knocked = Vec::new(); + + for room_id in rooms_in_order { + let events = &by_room[&room_id]; + let non_joins: Vec<&Arc> = events + .iter() + .copied() + .filter(|event| membership_of(event) != Some("join")) + .collect(); + let has_join = non_joins.len() != events.len(); + let in_room = joined.contains(&*room_id); + + // Joined again after leaving: always included, so that device + // list changes are calculated correctly. + if in_room && !non_joins.is_empty() { + newly_joined.insert(Arc::clone(&room_id)); + newly_joined_rooms.push(room_id); + continue; + } + + if in_room || has_join { + let old_membership = membership_at(reader, &room_id, user_id, since).await?; + if !is_join(old_membership.as_ref()) { + newly_joined.insert(Arc::clone(&room_id)); + newly_joined_rooms.push(Arc::clone(&room_id)); + } + } + + if in_room { + continue; + } + let Some(&last_non_join) = non_joins.last() else { + continue; + }; + + // Left the room: either joined before, or joined since and left. + let last = events.last().expect("a room has at least one event"); + if membership_of(last) != Some("join") { + if has_join { + newly_left_rooms.push(Arc::clone(&room_id)); + } else { + let old_membership = membership_at(reader, &room_id, user_id, since).await?; + if is_join(old_membership.as_ref()) { + newly_left_rooms.push(Arc::clone(&room_id)); + } + } + } + + match membership_of(last_non_join) { + Some("invite") => invited.push((Arc::clone(&room_id), Arc::clone(last_non_join))), + Some("knock") => knocked.push((Arc::clone(&room_id), Arc::clone(last_non_join))), + _ => {} + } + + // The last leave or ban. + let Some(leave_event) = non_joins + .iter() + .rev() + .find(|event| matches!(membership_of(event), Some("leave" | "ban"))) + else { + continue; + }; + let position = reader + .event_position(leave_event.event_id()) + .await? + .ok_or_else(|| anyhow::anyhow!("no position for {}", leave_event.event_id()))?; + if !persisted_after(&position.instance_name, position.stream, since) { + continue; + } + let leave_key = RoomStreamToken::live(position.stream); + // An out of band leave, like a remote invite rejection, is the + // timeline. Otherwise the timeline is loaded as usual. + let out_of_band = leave_event + .internal_metadata() + .is_out_of_band_membership() + .map_err(anyhow::Error::from)?; + entries.push(RoomEntryPlan { + newly_joined: newly_joined.contains(&room_id), + room_id, + rtype: RoomType::Archived, + events: out_of_band.then(|| vec![Arc::clone(leave_event)]), + full_state: false, + has_since: true, + upto: RoomToken { + base: TokenBase::Since, + room_key: leave_key.clone(), + }, + end: RoomToken { + base: TokenBase::Since, + room_key: leave_key, + }, + out_of_band, + }); + } + + // The events since `since` in the joined rooms that may have any, the + // most recent if there are too many: a gap. + let limit = request.filter.timeline_limit() + 1; + let to_fetch: Vec<&Arc> = input + .joined_room_ids + .iter() + .filter(|room_id| input.changed_room_ids.contains(*room_id)) + .collect(); + let fetches: Vec<_> = to_fetch + .into_iter() + .map(|room_id| recent_events(reader, room_id, now, since, limit)) + .collect(); + let fetched: HashMap, (Vec, RoomStreamToken)> = stream::iter(fetches) + .buffer_unordered(TIMELINE_FETCH_CONCURRENCY) + .try_collect() + .await?; + + // Every joined room, even without new events, for its other updates. + let mut fetched = fetched; + for room_id in &input.joined_room_ids { + let is_newly_joined = newly_joined.contains(room_id); + let entry = match fetched.remove(room_id) { + Some((mut events, start_key)) => { + // Oldest first. + events.reverse(); + RoomEntryPlan { + room_id: Arc::clone(room_id), + rtype: RoomType::Joined, + events: Some(events.into_iter().map(Arc::new).collect()), + newly_joined: is_newly_joined, + full_state: false, + has_since: !is_newly_joined, + upto: RoomToken { + base: TokenBase::Now, + room_key: start_key, + }, + end: RoomToken { + base: TokenBase::Now, + room_key: now.clone(), + }, + out_of_band: false, + } + } + None => RoomEntryPlan { + room_id: Arc::clone(room_id), + rtype: RoomType::Joined, + events: Some(Vec::new()), + newly_joined: is_newly_joined, + full_state: false, + has_since: true, + upto: RoomToken { + base: TokenBase::Since, + room_key: since.clone(), + }, + end: RoomToken { + base: TokenBase::Now, + room_key: now.clone(), + }, + out_of_band: false, + }, + }; + entries.push(entry); + } + + Ok(RoomChangesData { + since: input.since, + full_state: input.full_state, + entry_index: entry_index(&entries), + entries, + invited, + knocked, + newly_joined_rooms, + newly_left_rooms, + }) +} + +/// `paginate_room_events_by_stream_ordering` backwards from `now` to +/// `since`, for `get_room_events_stream_for_rooms`: the events, newest +/// first, and the token of the start of the page. +async fn recent_events( + reader: &SyncReader<'_>, + room_id: &Arc, + now: &RoomStreamToken, + since: &RoomStreamToken, + limit: i64, +) -> anyhow::Result<(Arc, (Vec, RoomStreamToken))> { + let (events, start_key, _) = reader + .paginate_room_events( + room_id, + now, + Some(since), + Direction::Backwards, + limit, + PaginationOrder::Stream, + ) + .await?; + Ok((Arc::clone(room_id), (events, start_key))) +} + +async fn initial_room_changes( + reader: &SyncReader<'_>, + request: &SyncRequestData, + input: RoomsInputData, +) -> anyhow::Result { + let now = &request.now_room_key; + let mut entries = Vec::new(); + // Room, event ID, and whether it is a knock rather than an invite. + let mut to_load: Vec<(Arc, Arc, bool)> = Vec::new(); + + for room in &input.initial_rooms { + match &*room.membership { + "join" => entries.push(RoomEntryPlan { + room_id: Arc::clone(&room.room_id), + rtype: RoomType::Joined, + events: None, + newly_joined: false, + full_state: true, + has_since: false, + upto: RoomToken { + base: TokenBase::Now, + room_key: now.clone(), + }, + end: RoomToken { + base: TokenBase::Now, + room_key: now.clone(), + }, + out_of_band: false, + }), + // Python drops these too, but they need not be loaded. + "invite" if request.ignored_users.contains(&room.sender) => {} + "invite" => { + to_load.push((Arc::clone(&room.room_id), Arc::clone(&room.event_id), false)) + } + "knock" => to_load.push((Arc::clone(&room.room_id), Arc::clone(&room.event_id), true)), + "leave" | "ban" => { + // Rooms the user was banned or kicked from are always sent. + if !request.filter.include_leave + && &*room.membership == "leave" + && room.sender == request.user_id + { + continue; + } + let leave_key = RoomStreamToken::live(room.stream); + entries.push(RoomEntryPlan { + room_id: Arc::clone(&room.room_id), + rtype: RoomType::Archived, + events: None, + newly_joined: false, + full_state: true, + has_since: false, + upto: RoomToken { + base: TokenBase::Now, + room_key: leave_key.clone(), + }, + end: RoomToken { + base: TokenBase::Now, + room_key: leave_key, + }, + out_of_band: false, + }); + } + _ => {} + } + } + + // `get_event`, for each invite and knock. + let ids: Vec> = to_load.iter().map(|(_, id, _)| Arc::clone(id)).collect(); + let loaded: HashMap, Arc> = reader + .events(&ids, RedactBehaviour::Redact, false, false) + .await? + .into_iter() + .map(|event| (Arc::from(event.event_id()), Arc::new(event))) + .collect(); + let mut invited = Vec::new(); + let mut knocked = Vec::new(); + for (room_id, event_id, is_knock) in to_load { + let event = loaded + .get(&event_id) + .ok_or_else(|| anyhow::anyhow!("could not find event {event_id}"))?; + let list = if is_knock { &mut knocked } else { &mut invited }; + list.push((room_id, Arc::clone(event))); + } + + Ok(RoomChangesData { + since: input.since, + full_state: input.full_state, + entry_index: entry_index(&entries), + entries, + invited, + knocked, + newly_joined_rooms: Vec::new(), + newly_left_rooms: Vec::new(), + }) +} + +/// What `handle_room_entries` passes `_generate_room_entry` besides the +/// room: the room's ephemeral events, tags, account data and sticky events. +pub struct RoomExtras { + pub ephemeral: Vec, + pub tags: Option, + pub account_data: Vec<(String, serde_json::Value)>, + pub sticky_event_ids: Vec>, +} + +impl RoomExtras { + /// `room_id`'s extras, read from the mappings Python passes, as + /// `handle_room_entries` reads them. + pub fn from_py( + room_id: &str, + ephemeral_by_room: &Bound<'_, PyAny>, + tags_by_room: &Bound<'_, PyAny>, + account_data_by_room: &Bound<'_, PyAny>, + sticky_by_room: &Bound<'_, PyAny>, + ) -> PyResult { + let py = ephemeral_by_room.py(); + fn get<'py>( + mapping: &Bound<'py, PyAny>, + room_id: &str, + ) -> PyResult>> { + let value = mapping.call_method1(intern!(mapping.py(), "get"), (room_id,))?; + Ok((!value.is_none()).then_some(value)) + } + + let mut ephemeral = Vec::new(); + if let Some(events) = get(ephemeral_by_room, room_id)? { + for event in events.try_iter()? { + ephemeral.push(depythonize(&event?)?); + } + } + let tags = get(tags_by_room, room_id)? + .map(|tags| depythonize(&tags)) + .transpose()?; + let mut account_data = Vec::new(); + if let Some(items) = get(account_data_by_room, room_id)? { + for item in items.call_method0(intern!(py, "items"))?.try_iter()? { + let (account_data_type, content): (String, Bound<'_, PyAny>) = item?.extract()?; + account_data.push((account_data_type, depythonize(&content)?)); + } + } + let sticky_event_ids = match get(sticky_by_room, room_id)? { + Some(ids) => ids + .extract::>()? + .into_iter() + .map(Arc::from) + .collect(), + None => Vec::new(), + }; + Ok(RoomExtras { + ephemeral, + tags, + account_data, + sticky_event_ids, + }) + } +} + +/// How one room's entry came out. +pub enum RoomOutcome { + /// The entry, or `None` when the room has nothing to send. + Entry(Option>), + /// Python must compute the room: [`Fallback`]. + Fallback(String), + /// Computing the room failed. + Error(String), +} + +/// The input of the room's `_generate_room_entry`. +pub fn room_input( + changes: &RoomChangesData, + plan: &RoomEntryPlan, + extras: RoomExtras, +) -> RoomSyncInputData { + RoomSyncInputData { + room_id: Arc::clone(&plan.room_id), + rtype: plan.rtype, + events: plan.events.clone(), + newly_joined: plan.newly_joined, + full_state: plan.full_state || plan.newly_joined || changes.full_state, + since: if plan.has_since { + changes.since.clone() + } else { + None + }, + upto: plan.upto.room_key.clone(), + end: plan.end.room_key.clone(), + out_of_band: plan.out_of_band, + ephemeral: extras.ephemeral, + tags: extras.tags, + account_data: extras.account_data, + sticky_event_ids: extras.sticky_event_ids, + always_include: changes.full_state, + } +} + +/// One room's entry. Its lazy-loaded members are recorded once it is +/// computed. +async fn room_entry( + reader: &SyncReader<'_>, + request: &SyncRequestData, + ll_members: &Mutex, + room: RoomSyncInputData, +) -> (Arc, RoomOutcome) { + let context = RoomContext { + reader, + request, + client_filter: ClientFilterContext { + user_id: Arc::clone(&request.user_id), + ignored_users: Arc::clone(&request.ignored_users), + admin_policy: request.admin_policy, + msc4354_enabled: request.msc4354_enabled, + }, + ll_members, + ll_pending: Mutex::default(), + clock: Clock::new(), + }; + // A panic is this room's error. It would otherwise fail every room of + // the request, including those that have recorded the members they + // sent, which Python then could not compute. + let outcome = match AssertUnwindSafe(generate_room_entry(&context, &room)) + .catch_unwind() + .await + { + Ok(Ok(result)) => { + context.commit_ll_members(); + RoomOutcome::Entry(result.map(Box::new)) + } + Ok(Err(err)) => match err.downcast::() { + Ok(fallback) => RoomOutcome::Fallback(fallback.0), + Err(err) => RoomOutcome::Error(format!("{err:#}")), + }, + Err(panic) => RoomOutcome::Error(format!("panicked: {}", panic_message(&*panic))), + }; + (Arc::clone(&room.room_id), outcome) +} + +/// The message of a panic's payload. +fn panic_message(panic: &(dyn std::any::Any + Send)) -> &str { + if let Some(message) = panic.downcast_ref::<&str>() { + message + } else if let Some(message) = panic.downcast_ref::() { + message + } else { + "unknown error" + } +} + +/// The entries of `rooms`, ten at a time, in the order they finish. A +/// room's lazy-loaded members are recorded once its entry is computed. +pub async fn generate_room_entries( + reader: &SyncReader<'_>, + request: &SyncRequestData, + ll_members: &Mutex, + rooms: Vec, +) -> Vec<(Arc, RoomOutcome)> { + let rooms: Vec<_> = rooms + .into_iter() + .map(|room| room_entry(reader, request, ll_members, room)) + .collect(); + stream::iter(rooms) + .buffer_unordered(ROOM_ENTRY_CONCURRENCY) + .collect() + .await +} + +/// What [`room_changes`] reads besides the request. See +/// [`RoomsInputData`]. +#[pyclass(frozen, skip_from_py_object)] +pub struct RoomsInput { + pub(crate) data: Mutex>, +} + +#[pymethods] +impl RoomsInput { + #[new] + #[pyo3(signature = (*, since, full_state, joined_room_ids, changed_room_ids, forced_newly_joined_room_ids, membership_change_events, initial_rooms))] + #[allow(clippy::too_many_arguments)] + fn py_new( + since: Option, + full_state: bool, + joined_room_ids: Vec, + changed_room_ids: HashSet, + forced_newly_joined_room_ids: Vec, + membership_change_events: Vec>, + initial_rooms: Vec<(String, String, String, String, i64)>, + ) -> Self { + RoomsInput { + data: Mutex::new(Some(RoomsInputData { + since, + full_state, + joined_room_ids: joined_room_ids.into_iter().map(Arc::from).collect(), + changed_room_ids: changed_room_ids.into_iter().map(Arc::from).collect(), + forced_newly_joined_room_ids: forced_newly_joined_room_ids + .into_iter() + .map(Arc::from) + .collect(), + membership_change_events: membership_change_events + .iter() + .map(|event| Arc::new(event.get().shallow_clone())) + .collect(), + initial_rooms: initial_rooms + .into_iter() + .map( + |(room_id, sender, membership, event_id, stream)| RoomForUser { + room_id: Arc::from(room_id), + sender: Arc::from(sender), + membership: Arc::from(membership), + event_id: Arc::from(event_id), + stream, + }, + ) + .collect(), + })), + } + } +} + +impl RoomsInput { + /// The input, which a [`room_changes`] call consumes. + pub(crate) fn take(&self) -> PyResult { + lock_ignoring_poison(&self.data) + .take() + .ok_or_else(|| PyRuntimeError::new_err("RoomsInput already used")) + } +} + +fn events_to_py<'py>( + py: Python<'py>, + events: &[(Arc, Arc)], +) -> PyResult> { + let items = events + .iter() + .map(|(room_id, event)| { + let event = Bound::new(py, event.shallow_clone())?; + PyTuple::new( + py, + [room_id.into_pyobject(py)?.into_any(), event.into_any()], + ) + }) + .collect::>>()?; + PyList::new(py, items) +} + +fn room_ids_to_py<'py>(py: Python<'py>, room_ids: &[Arc]) -> PyResult> { + PyList::new(py, room_ids.iter().map(|room_id| &**room_id)) +} + +/// A `_RoomChanges`. See [`RoomChangesData`]. +#[pyclass(frozen, skip_from_py_object)] +pub struct RoomChanges { + pub(crate) data: Arc, +} + +#[pymethods] +impl RoomChanges { + #[getter] + fn newly_joined_rooms<'py>(&self, py: Python<'py>) -> PyResult> { + room_ids_to_py(py, &self.data.newly_joined_rooms) + } + + #[getter] + fn newly_left_rooms<'py>(&self, py: Python<'py>) -> PyResult> { + room_ids_to_py(py, &self.data.newly_left_rooms) + } + + /// The invites as `(room_id, event)`, including those Python drops. + #[getter] + fn invited<'py>(&self, py: Python<'py>) -> PyResult> { + events_to_py(py, &self.data.invited) + } + + /// The knocks as `(room_id, event)`. + #[getter] + fn knocked<'py>(&self, py: Python<'py>) -> PyResult> { + events_to_py(py, &self.data.knocked) + } + + /// The number of joined and archived rooms to compute. + fn __len__(&self) -> usize { + self.data.entries.len() + } + + /// The `RoomSyncResultBuilder` fields of a room, for Python to compute + /// it: `(rtype, events, newly_joined, full_state, has_since, + /// upto_from_now, upto_room_key, end_from_now, end_room_key, + /// out_of_band)`. + fn room_entry<'py>(&self, py: Python<'py>, room_id: &str) -> PyResult> { + let plan = self + .data + .entry_index + .get(room_id) + .map(|&index| &self.data.entries[index]) + .ok_or_else(|| PyKeyError::new_err(room_id.to_owned()))?; + let events = match &plan.events { + Some(events) => PyList::new( + py, + events + .iter() + .map(|event| Bound::new(py, event.shallow_clone())) + .collect::>>()?, + )? + .into_any(), + None => py.None().into_bound(py), + }; + PyTuple::new( + py, + [ + rtype_str(plan.rtype).into_pyobject(py)?.into_any(), + events, + PyBool::new(py, plan.newly_joined).to_owned().into_any(), + PyBool::new(py, plan.full_state).to_owned().into_any(), + PyBool::new(py, plan.has_since).to_owned().into_any(), + PyBool::new(py, plan.upto.base == TokenBase::Now) + .to_owned() + .into_any(), + (&plan.upto.room_key).into_pyobject(py)?.into_any(), + PyBool::new(py, plan.end.base == TokenBase::Now) + .to_owned() + .into_any(), + (&plan.end.room_key).into_pyobject(py)?.into_any(), + PyBool::new(py, plan.out_of_band).to_owned().into_any(), + ], + ) + } +} + +/// What [`generate_room_entries`] made of the rooms. +#[pyclass(frozen, skip_from_py_object)] +pub struct RoomsResult { + /// `(entry, upto_from_now, upto_room_key)` for each room with an entry: + /// the room's upto token is the request's now or since token with that + /// room key. Taken by [`RoomsResult::take_entries`]. + pub(crate) entries: Mutex>, + /// The rooms Rust computed with nothing to send. + pub(crate) empty: Vec>, + pub(crate) fallbacks: Vec<(Arc, String)>, + pub(crate) errors: Vec<(Arc, String)>, +} + +#[pymethods] +impl RoomsResult { + /// The entries as `(RoomSyncResult, upto_from_now, upto_room_key)`. A + /// second call returns an empty list. + fn take_entries<'py>(&self, py: Python<'py>) -> PyResult> { + let entries = std::mem::take(&mut *lock_ignoring_poison(&self.entries)); + let items = entries + .into_iter() + .map(|(data, from_now, room_key)| { + PyTuple::new( + py, + [ + Bound::new(py, RoomSyncResult { data })?.into_any(), + PyBool::new(py, from_now).to_owned().into_any(), + room_key.into_pyobject(py)?.into_any(), + ], + ) + }) + .collect::>>()?; + PyList::new(py, items) + } + + #[getter] + fn empty<'py>(&self, py: Python<'py>) -> PyResult> { + room_ids_to_py(py, &self.empty) + } + + /// `(room_id, reason)` for each room Python must compute. + #[getter] + fn fallbacks(&self) -> Vec<(String, String)> { + self.fallbacks + .iter() + .map(|(room_id, reason)| (room_id.to_string(), reason.clone())) + .collect() + } + + /// `(room_id, error)` for each room Rust failed on. + #[getter] + fn errors(&self) -> Vec<(String, String)> { + self.errors + .iter() + .map(|(room_id, error)| (room_id.to_string(), error.clone())) + .collect() + } +} + +impl RoomsResult { + /// The result for `outcomes`, the rooms of `changes`. + pub fn new(changes: &RoomChangesData, outcomes: Vec<(Arc, RoomOutcome)>) -> Self { + let mut entries = Vec::new(); + let mut empty = Vec::new(); + let mut fallbacks = Vec::new(); + let mut errors = Vec::new(); + for (room_id, outcome) in outcomes { + match outcome { + RoomOutcome::Entry(Some(data)) => { + let token = &changes.entries[changes.entry_index[&room_id]].upto; + entries.push((*data, token.base == TokenBase::Now, token.room_key.clone())); + } + RoomOutcome::Entry(None) => 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, + fallbacks, + errors, + } + } +} diff --git a/rust/src/visibility.rs b/rust/src/visibility.rs index 311c3d08da..bb8a9eed7d 100644 --- a/rust/src/visibility.rs +++ b/rust/src/visibility.rs @@ -100,7 +100,7 @@ pub struct FilteredEvent { } /// The `membership` of a membership event's content. -fn membership_of(event: &Event) -> Option<&str> { +pub(crate) fn membership_of(event: &Event) -> Option<&str> { event .content() .get_field(MEMBERSHIP) diff --git a/synapse/synapse_rust/sync.pyi b/synapse/synapse_rust/sync.pyi index 41ca5b805c..12e9b27695 100644 --- a/synapse/synapse_rust/sync.pyi +++ b/synapse/synapse_rust/sync.pyi @@ -10,7 +10,7 @@ # See the GNU Affero General Public License for more details: # . -from collections.abc import Mapping, Sequence +from collections.abc import Collection, Mapping, Sequence from twisted.internet.defer import Deferred @@ -19,10 +19,6 @@ from synapse.synapse_rust.events import Event from synapse.types import JsonDict, JsonMapping, RoomStreamToken from synapse.types.state import StateFilter -class RustSyncFallback(Exception): - """The Rust sync handler cannot compute this room's entry. Compute it - with the Python implementation.""" - class SyncHandler: """Computes sync responses with the Rust implementation, reading the database through the native Rust Postgres driver. @@ -55,18 +51,43 @@ class SyncHandler: """Record that `event_id` is being sent to the device as the membership of `member`, in the response ending at `next_batch`.""" - def generate_room_entry( - self, request: "SyncRequest", room: "RoomSyncInput" - ) -> "Deferred[RoomSyncResult | None]": - """`_generate_room_entry` for one joined or archived room. Resolves to - None when the room has nothing to send. Fails with `RustSyncFallback` - for a room the Python implementation must compute. Cancelling the + def room_changes( + self, request: "SyncRequest", input: "RoomsInput" + ) -> "Deferred[RoomChanges]": + """`_get_room_changes_for_incremental_sync` or + `_get_room_changes_for_initial_sync`. Consumes `input`. Cancelling the deferred stops the computation.""" - def generate_room_entry_blocking( - self, request: "SyncRequest", room: "RoomSyncInput" - ) -> "RoomSyncResult | None": - """As `generate_room_entry`, but blocks. For tests.""" + def room_changes_blocking( + self, request: "SyncRequest", input: "RoomsInput" + ) -> "RoomChanges": + """As `room_changes`, but blocks. For tests.""" + + def generate_room_entries( + self, + request: "SyncRequest", + changes: "RoomChanges", + *, + ephemeral_by_room: Mapping[str, Sequence[JsonMapping]], + tags_by_room: Mapping[str, Mapping[str, JsonMapping]], + account_data_by_room: Mapping[str, Mapping[str, JsonMapping]], + sticky_by_room: Mapping[str, Sequence[str]], + ) -> "Deferred[RoomsResult]": + """The entries of the joined and archived rooms of `changes`, ten at + a time. The `*_by_room` mappings are what `handle_room_entries` reads + for each room. Cancelling the deferred stops the computation.""" + + def generate_room_entries_blocking( + self, + request: "SyncRequest", + changes: "RoomChanges", + *, + ephemeral_by_room: Mapping[str, Sequence[JsonMapping]], + tags_by_room: Mapping[str, Mapping[str, JsonMapping]], + account_data_by_room: Mapping[str, Mapping[str, JsonMapping]], + sticky_by_room: Mapping[str, Sequence[str]], + ) -> "RoomsResult": + """As `generate_room_entries`, but blocks. For tests.""" class SyncRequest: """One `/sync` request, as the per-room computation sees it.""" @@ -87,29 +108,79 @@ class SyncRequest: msc4354_enabled: bool, ) -> None: ... -class RoomSyncInput: - """One room's `RoomSyncResultBuilder`, plus what `handle_room_entries` - passes alongside it.""" +class RoomsInput: + """What `room_changes` reads besides the request, from the + `SyncResultBuilder`.""" def __init__( self, *, - room_id: str, - rtype: str, - events: Sequence[Event] | None, - newly_joined: bool, - full_state: bool, since: RoomStreamToken | None, - upto: RoomStreamToken, - end: RoomStreamToken, - out_of_band: bool, - ephemeral: Sequence[JsonMapping], - tags: Mapping[str, JsonMapping] | None, - account_data: Mapping[str, JsonMapping], - sticky_event_ids: Sequence[str], - always_include: bool, - room_max_token: RoomStreamToken, - ) -> None: ... + full_state: bool, + joined_room_ids: Sequence[str], + changed_room_ids: Collection[str], + forced_newly_joined_room_ids: Sequence[str], + membership_change_events: Sequence[Event], + initial_rooms: Sequence[tuple[str, str, str, str, int]], + ) -> None: + """`changed_room_ids` are the joined rooms that may have events after + `since`. `initial_rooms` are `(room_id, sender, membership, event_id, + stream)` for each of the user's rooms in a known room version, for an + initial sync.""" + +class RoomChanges: + """A `_RoomChanges`, with every invite: Python drops those of ignored + users and those the invite rules ignore.""" + + @property + def newly_joined_rooms(self) -> list[str]: ... + @property + def newly_left_rooms(self) -> list[str]: ... + @property + def invited(self) -> list[tuple[str, Event]]: ... + @property + def knocked(self) -> list[tuple[str, Event]]: ... + def __len__(self) -> int: + """The number of joined and archived rooms to compute.""" + + def room_entry( + self, room_id: str + ) -> tuple[ + str, + list[Event] | None, + bool, + bool, + bool, + bool, + RoomStreamToken, + bool, + RoomStreamToken, + bool, + ]: + """The `RoomSyncResultBuilder` fields of a room, for Python to compute + it: `(rtype, events, newly_joined, full_state, has_since, + upto_from_now, upto_room_key, end_from_now, end_room_key, + out_of_band)`.""" + +class RoomsResult: + """What `generate_room_entries` made of the rooms.""" + + def take_entries(self) -> list[tuple["RoomSyncResult", bool, RoomStreamToken]]: + """The entries as `(entry, upto_from_now, upto_room_key)`: the room's + upto token is the request's now or since token with that room key. A + second call returns an empty list.""" + + @property + def empty(self) -> list[str]: + """The rooms computed with nothing to send.""" + + @property + def fallbacks(self) -> list[tuple[str, str]]: + """`(room_id, reason)` for each room Python must compute.""" + + @property + def errors(self) -> list[tuple[str, str]]: + """`(room_id, error)` for each room the Rust handler failed on.""" class RoomSyncResult: """A room's entry, less the bundled aggregations."""