Work out a sync response's rooms in Rust

Phase 4 of the sync-in-Rust plan. `SyncHandler.room_changes` ports
`_get_room_changes_for_incremental_sync` and
`_get_room_changes_for_initial_sync`: which rooms the response reports, and
which the user joined or left since the last sync. It fetches the timelines
of the joined rooms itself, as `get_room_events_stream_for_rooms` does.
`SyncHandler.generate_room_entries` then computes every joined and archived
room's entry, ten at a time. A room it cannot compute, or fails on, comes
back in the result for Python to compute. A panic in a room is that room's
error, so it cannot fail the rooms that have recorded their lazy-loaded
members.

The `state_after` rollback now reads the state deltas with no upper bound,
after the current state, so that it covers every change the current state
may hold.

These replace the per-room `generate_room_entry` and `RoomSyncInput`. The
store gains `event_position`, for the leave events of archived rooms.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb

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<bool> {
+    pub fn is_out_of_band_membership(&self) -> PyResult<bool> {
         Ok(self.read_inner()?.is_out_of_band_membership())
     }
 

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<Option<PersistedEventPosition>> {
+        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<String> = 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(

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<Mutex<LazyLoadedMembersCaches>>,
 }
 
-/// One room's entry, started. See [`SyncHandler::start_room`].
-type RoomFuture = Pin<Box<dyn Future<Output = PyResult<Option<RoomSyncResult>>> + Send>>;
-
 impl SyncHandler {
     pub fn new(
         db_pool: Py<PyAny>,
@@ -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<RoomFuture> {
+    /// The connection pool to read through, or an error if it is closed.
+    fn pool(&self, py: Python<'_>) -> PyResult<ConnectionPool> {
         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::<PyConnectionPool>()?.get().pool.clone();
+        Ok(pool.cast::<PyConnectionPool>()?.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<BoxFuture<'static, PyResult<RoomChanges>>> {
+        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<BoxFuture<'static, PyResult<RoomsResult>>> {
+        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::<PyResult<Vec<_>>>()?;
+        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<T: Send>(
+        &self,
+        py: Python<'_>,
+        future: BoxFuture<'static, PyResult<T>>,
+    ) -> PyResult<T> {
+        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::<Fallback>() {
-        Ok(fallback) => RustSyncFallback::new_err(fallback.0),
-        Err(err) => match err.downcast::<PyErr>() {
-            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::<PyErr>() {
+        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<Bound<'py, PyAny>> {
-        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<Option<RoomSyncResult>> {
-        let future = self.start_room(py, request, room)?;
-        let handle = self.runtime.tokio_handle()?;
-        py.detach(|| handle.block_on(future))
+        input: &Bound<'_, RoomsInput>,
+    ) -> PyResult<RoomChanges> {
+        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<Bound<'py, PyAny>> {
+        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<RoomsResult> {
+        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::<filter::FilterCollection>()?;
     child_module.add_class::<ll_members::LazyLoadedMembersCache>()?;
     child_module.add_class::<SyncRequest>()?;
-    child_module.add_class::<RoomSyncInput>()?;
     child_module.add_class::<RoomSyncResult>()?;
+    child_module.add_class::<RoomsInput>()?;
+    child_module.add_class::<RoomChanges>()?;
+    child_module.add_class::<RoomsResult>()?;
     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::<RustSyncFallback>())?;
 
     m.add_submodule(&child_module)?;
 

rust/src/sync/request.rs

@@ -12,10 +12,10 @@
  * <https://www.gnu.org/licenses/agpl-3.0.html>.
  */
 
-//! 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<Arc<str>>,
     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<RoomSyncInputData>,
-}
-
-#[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<Vec<Bound<'_, Event>>>,
-        newly_joined: bool,
-        full_state: bool,
-        since: Option<RoomStreamToken>,
-        upto: RoomStreamToken,
-        end: RoomStreamToken,
-        out_of_band: bool,
-        ephemeral: Vec<Bound<'_, PyAny>>,
-        tags: Option<Bound<'_, PyAny>>,
-        account_data: &Bound<'_, PyAny>,
-        sticky_event_ids: Vec<String>,
-        always_include: bool,
-        room_max_token: RoomStreamToken,
-    ) -> PyResult<Self> {
-        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::<Result<_, _>>()?,
-                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.

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));
             }

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:
+ * <https://www.gnu.org/licenses/agpl-3.0.html>.
+ */
+
+//! 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<str>,
+    pub rtype: RoomType,
+    /// The events between the since and upto tokens, if fetched.
+    pub events: Option<Vec<Arc<Event>>>,
+    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<str>,
+    pub sender: Arc<str>,
+    pub membership: Arc<str>,
+    pub event_id: Arc<str>,
+    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<RoomStreamToken>,
+    /// The request's `full_state`.
+    pub full_state: bool,
+    pub joined_room_ids: Vec<Arc<str>>,
+    /// The joined rooms that may have events after `since`, by the stream
+    /// change cache. Incremental syncs only.
+    pub changed_room_ids: HashSet<Arc<str>>,
+    pub forced_newly_joined_room_ids: Vec<Arc<str>>,
+    /// The user's membership events after `since`, oldest first.
+    pub membership_change_events: Vec<Arc<Event>>,
+    /// The user's rooms, in a known room version and not excluded. Initial
+    /// syncs only.
+    pub initial_rooms: Vec<RoomForUser>,
+}
+
+/// A `_RoomChanges`, less the invites Python drops.
+pub struct RoomChangesData {
+    pub since: Option<RoomStreamToken>,
+    pub full_state: bool,
+    pub entries: Vec<RoomEntryPlan>,
+    /// The index in `entries` of each room.
+    pub entry_index: HashMap<Arc<str>, usize>,
+    pub invited: Vec<(Arc<str>, Arc<Event>)>,
+    pub knocked: Vec<(Arc<str>, Arc<Event>)>,
+    pub newly_joined_rooms: Vec<Arc<str>>,
+    pub newly_left_rooms: Vec<Arc<str>>,
+}
+
+fn entry_index(entries: &[RoomEntryPlan]) -> HashMap<Arc<str>, 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<RoomChangesData> {
+    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<Option<Event>> {
+    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<RoomChangesData> {
+    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<Arc<str>> = Vec::new();
+    let mut by_room: HashMap<Arc<str>, Vec<&Arc<Event>>> = HashMap::new();
+    for event in &input.membership_change_events {
+        let room_id: Arc<str> = 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<Arc<str>> = input.forced_newly_joined_room_ids.clone();
+    // The same rooms, to look them up.
+    let mut newly_joined: HashSet<Arc<str>> = newly_joined_rooms.iter().cloned().collect();
+    let mut newly_left_rooms: Vec<Arc<str>> = Vec::new();
+    let mut entries: Vec<RoomEntryPlan> = 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<Event>> = 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<str>> = 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<Arc<str>, (Vec<Event>, 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<str>,
+    now: &RoomStreamToken,
+    since: &RoomStreamToken,
+    limit: i64,
+) -> anyhow::Result<(Arc<str>, (Vec<Event>, 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<RoomChangesData> {
+    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<str>, Arc<str>, 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<Arc<str>> = to_load.iter().map(|(_, id, _)| Arc::clone(id)).collect();
+    let loaded: HashMap<Arc<str>, Arc<Event>> = 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<serde_json::Value>,
+    pub tags: Option<serde_json::Value>,
+    pub account_data: Vec<(String, serde_json::Value)>,
+    pub sticky_event_ids: Vec<Arc<str>>,
+}
+
+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<Self> {
+        let py = ephemeral_by_room.py();
+        fn get<'py>(
+            mapping: &Bound<'py, PyAny>,
+            room_id: &str,
+        ) -> PyResult<Option<Bound<'py, PyAny>>> {
+            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::<Vec<String>>()?
+                .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<Box<RoomSyncResultData>>),
+    /// 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<LazyLoadedMembersCaches>,
+    room: RoomSyncInputData,
+) -> (Arc<str>, 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::<Fallback>() {
+            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::<String>() {
+        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<LazyLoadedMembersCaches>,
+    rooms: Vec<RoomSyncInputData>,
+) -> Vec<(Arc<str>, 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<Option<RoomsInputData>>,
+}
+
+#[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<RoomStreamToken>,
+        full_state: bool,
+        joined_room_ids: Vec<String>,
+        changed_room_ids: HashSet<String>,
+        forced_newly_joined_room_ids: Vec<String>,
+        membership_change_events: Vec<Bound<'_, Event>>,
+        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<RoomsInputData> {
+        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<str>, Arc<Event>)],
+) -> PyResult<Bound<'py, PyList>> {
+    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::<PyResult<Vec<_>>>()?;
+    PyList::new(py, items)
+}
+
+fn room_ids_to_py<'py>(py: Python<'py>, room_ids: &[Arc<str>]) -> PyResult<Bound<'py, PyList>> {
+    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<RoomChangesData>,
+}
+
+#[pymethods]
+impl RoomChanges {
+    #[getter]
+    fn newly_joined_rooms<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
+        room_ids_to_py(py, &self.data.newly_joined_rooms)
+    }
+
+    #[getter]
+    fn newly_left_rooms<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
+        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<Bound<'py, PyList>> {
+        events_to_py(py, &self.data.invited)
+    }
+
+    /// The knocks as `(room_id, event)`.
+    #[getter]
+    fn knocked<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
+        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<Bound<'py, PyTuple>> {
+        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::<PyResult<Vec<_>>>()?,
+            )?
+            .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<Vec<(RoomSyncResultData, bool, RoomStreamToken)>>,
+    /// The rooms Rust computed with nothing to send.
+    pub(crate) empty: Vec<Arc<str>>,
+    pub(crate) fallbacks: Vec<(Arc<str>, String)>,
+    pub(crate) errors: Vec<(Arc<str>, 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<Bound<'py, PyList>> {
+        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::<PyResult<Vec<_>>>()?;
+        PyList::new(py, items)
+    }
+
+    #[getter]
+    fn empty<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyList>> {
+        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<str>, 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,
+        }
+    }
+}

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)

synapse/synapse_rust/sync.pyi

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