Mirrors/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2026-05-26 20:49:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: Properly sync left rooms

Browse Source 
- Remove most usages of `update_membership` in favor
  of directly calling the `mark_as_*` functions
- Store the leave membership event as the value in the
  `userroomid_leftstate` table
- Use the `userroomid_leftstate` table to synchronize the
  timeline and state for left rooms if possible
This commit is contained in:
Ginger
2025-10-27 17:24:02 -04:00
parent afd115eedc
commit 1ad60df7a6
14 changed files with 759 additions and 665 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/client/sync/mod.rs
+13 -3
View File
@@ -4,15 +4,15 @@ mod v5;
use std::collections::VecDeque;
use conduwuit::{
PduCount, Result,
Event, PduCount, Result,
matrix::pdu::PduEvent,
trace,
ref_at, trace,
utils::stream::{BroadbandExt, ReadyExt, TryIgnore},
};
use conduwuit_service::Services;
use futures::StreamExt;
use ruma::{
RoomId, UserId,
OwnedUserId, RoomId, UserId,
events::TimelineEventType::{
self, Beacon, CallInvite, PollStart, RoomEncrypted, RoomMessage, Sticker,
},
@@ -29,6 +29,16 @@ pub(crate) struct TimelinePdus {
pub limited: bool,
}
impl TimelinePdus {
fn senders(&self) -> impl Iterator<Item = OwnedUserId> {
self.pdus
.iter()
.map(ref_at!(1))
.map(Event::sender)
.map(Into::into)
}
}
async fn load_timeline(
services: &Services,
sender_user: &UserId,
src/api/client/sync/v3/joined.rs
+23 -293
View File
@@ -1,7 +1,4 @@
use std::{
collections::{BTreeMap, BTreeSet, HashMap},
ops::ControlFlow,
};
use std::collections::{BTreeMap, HashMap};
use conduwuit::{
Result, at, err, extract_variant, is_equal_to,
@@ -9,29 +6,20 @@ use conduwuit::{
Event,
pdu::{PduCount, PduEvent},
},
ref_at,
result::FlatOk,
utils::{
BoolExt, IterStream, ReadyExt, TryFutureExtExt,
math::ruma_from_u64,
stream::{BroadbandExt, Tools, TryIgnore, WidebandExt},
},
};
use conduwuit_service::{
Services,
rooms::{
lazy_loading,
lazy_loading::{MemberSet, Options},
short::ShortStateHash,
stream::{Tools, WidebandExt},
},
};
use conduwuit_service::{Services, rooms::lazy_loading::MemberSet};
use futures::{
FutureExt, StreamExt, TryFutureExt,
future::{OptionFuture, join, join3, join4, try_join},
};
use itertools::Itertools;
use ruma::{
OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, UserId,
OwnedRoomId, OwnedUserId, RoomId, UserId,
api::client::sync::sync_events::{
UnreadNotificationsCount,
v3::{Ephemeral, JoinedRoom, RoomAccountData, RoomSummary, State as RoomState, Timeline},
@@ -44,13 +32,14 @@ use ruma::{
serde::Raw,
uint,
};
use service::rooms::short::ShortEventId;
use tracing::trace;
use super::{load_timeline, share_encrypted_room};
use crate::client::{
TimelinePdus, ignored_filter,
sync::v3::{DeviceListUpdates, SyncContext},
ignored_filter,
sync::v3::{
DeviceListUpdates, SyncContext,
state::{calculate_state_incremental, calculate_state_initial},
},
};
/// Generate the sync response for a room the user is joined to.
@@ -80,11 +69,10 @@ pub(super) async fn load_joined_room(
let SyncContext {
sender_user,
sender_device,
since,
next_batch,
full_state,
filter,
..
} = sync_context;
// the global count as of the end of the last sync.
@@ -211,38 +199,23 @@ pub(super) async fn load_joined_room(
|content: RoomMemberEventContent| content.membership != MembershipState::Join,
);
// lazy loading is only enabled if the filter allows for it and we aren't
// requesting the full state
let lazy_loading_enabled = (filter.room.state.lazy_load_options.is_enabled()
|| filter.room.timeline.lazy_load_options.is_enabled())
&& !full_state;
let lazy_loading_context = &lazy_loading::Context {
user_id: sender_user,
device_id: Some(sender_device),
room_id,
token: since,
options: Some(&filter.room.state.lazy_load_options),
};
let lazy_loading_context = &sync_context.lazy_loading_context(room_id);
// the user IDs of members whose membership needs to be sent to the client, if
// lazy-loading is enabled.
let lazily_loaded_members = OptionFuture::from(lazy_loading_enabled.then(|| {
let witness: MemberSet = timeline
.pdus
.iter()
.map(ref_at!(1))
.map(Event::sender)
.map(Into::into)
.chain(receipt_events.keys().map(Into::into))
.collect();
let lazily_loaded_members =
OptionFuture::from(sync_context.lazy_loading_enabled().then(|| {
let timeline_and_receipt_members: MemberSet = timeline
.senders()
.chain(receipt_events.keys().map(Into::into))
.collect();
services
.rooms
.lazy_loading
.retain_lazy_members(witness, lazy_loading_context)
}))
.await;
services
.rooms
.lazy_loading
.retain_lazy_members(timeline_and_receipt_members, lazy_loading_context)
}))
.await;
// reset lazy loading state on initial sync
if previous_sync_end_count.is_none() {
@@ -473,249 +446,6 @@ pub(super) async fn load_joined_room(
Ok((joined_room, device_list_updates))
}
/// Calculate the state events to include in an initial sync response.
///
/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
/// Vec will include the membership events of exclusively the members in
/// `lazily_loaded_members`.
#[tracing::instrument(
name = "initial",
level = "trace",
skip_all,
fields(current_shortstatehash)
)]
#[allow(clippy::too_many_arguments)]
async fn calculate_state_initial(
services: &Services,
sender_user: &UserId,
timeline_start_shortstatehash: ShortStateHash,
lazily_loaded_members: Option<&MemberSet>,
) -> Result<Vec<PduEvent>> {
// load the keys and event IDs of the state events at the start of the timeline
let (shortstatekeys, event_ids): (Vec<_>, Vec<_>) = services
.rooms
.state_accessor
.state_full_ids(timeline_start_shortstatehash)
.unzip()
.await;
trace!("performing initial sync of {} state events", event_ids.len());
services
.rooms
.short
// look up the full state keys
.multi_get_statekey_from_short(shortstatekeys.into_iter().stream())
.zip(event_ids.into_iter().stream())
.ready_filter_map(|item| Some((item.0.ok()?, item.1)))
.ready_filter_map(|((event_type, state_key), event_id)| {
if let Some(lazily_loaded_members) = lazily_loaded_members {
/*
if lazy loading is enabled, filter out membership events which aren't for a user
included in `lazily_loaded_members` or for the user requesting the sync.
*/
let event_is_redundant = event_type == StateEventType::RoomMember
&& state_key.as_str().try_into().is_ok_and(|user_id: &UserId| {
sender_user != user_id && !lazily_loaded_members.contains(user_id)
});
event_is_redundant.or_some(event_id)
} else {
Some(event_id)
}
})
.broad_filter_map(|event_id: OwnedEventId| async move {
services.rooms.timeline.get_pdu(&event_id).await.ok()
})
.collect()
.map(Ok)
.await
}
/// Calculate the state events to include in an incremental sync response.
///
/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
/// Vec will include the membership events of all the members in
/// `lazily_loaded_members`.
#[tracing::instrument(name = "incremental", level = "trace", skip_all)]
#[allow(clippy::too_many_arguments)]
async fn calculate_state_incremental<'a>(
services: &Services,
sender_user: &'a UserId,
room_id: &RoomId,
previous_sync_end_count: PduCount,
previous_sync_end_shortstatehash: ShortStateHash,
timeline_start_shortstatehash: ShortStateHash,
timeline_end_shortstatehash: ShortStateHash,
timeline: &TimelinePdus,
lazily_loaded_members: Option<&'a MemberSet>,
) -> Result<Vec<PduEvent>> {
// NB: a limited sync is one where `timeline.limited == true`. Synapse calls
// this a "gappy" sync internally.
/*
the state events returned from an incremental sync which isn't limited are usually empty.
however, if an event in the timeline (`timeline.pdus`) merges a split in the room's DAG (i.e. has multiple `prev_events`),
the state at the _end_ of the timeline may include state events which were merged in and don't exist in the state
at the _start_ of the timeline. because this is uncommon, we check here to see if any events in the timeline
merged a split in the DAG.
see: https://github.com/element-hq/synapse/issues/16941
*/
let timeline_is_linear = timeline.pdus.is_empty() || {
let last_pdu_of_last_sync = services
.rooms
.timeline
.pdus_rev(Some(sender_user), room_id, Some(previous_sync_end_count.saturating_add(1)))
.boxed()
.next()
.await
.transpose()
.expect("last sync should have had some PDUs")
.map(at!(1));
// make sure the prev_events of each pdu in the timeline refer only to the
// previous pdu
timeline
.pdus
.iter()
.try_fold(last_pdu_of_last_sync.map(|pdu| pdu.event_id), |prev_event_id, (_, pdu)| {
if let Ok(pdu_prev_event_id) = pdu.prev_events.iter().exactly_one() {
if prev_event_id
.as_ref()
.is_none_or(is_equal_to!(pdu_prev_event_id))
{
return ControlFlow::Continue(Some(pdu_prev_event_id.to_owned()));
}
}
trace!(
"pdu {:?} has split prev_events (expected {:?}): {:?}",
pdu.event_id, prev_event_id, pdu.prev_events
);
ControlFlow::Break(())
})
.is_continue()
};
if timeline_is_linear && !timeline.limited {
// if there are no splits in the DAG and the timeline isn't limited, then
// `state` will always be empty unless lazy loading is enabled.
if let Some(lazily_loaded_members) = lazily_loaded_members
&& !timeline.pdus.is_empty()
{
// lazy loading is enabled, so we return the membership events which were
// requested by the caller.
let lazy_membership_events: Vec<_> = lazily_loaded_members
.iter()
.stream()
.broad_filter_map(|user_id| async move {
if user_id == sender_user {
return None;
}
services
.rooms
.state_accessor
.state_get(
timeline_start_shortstatehash,
&StateEventType::RoomMember,
user_id.as_str(),
)
.ok()
.await
})
.collect()
.await;
if !lazy_membership_events.is_empty() {
trace!(
"syncing lazy membership events for members: {:?}",
lazy_membership_events
.iter()
.map(|pdu| pdu.state_key().unwrap())
);
}
return Ok(lazy_membership_events);
}
// lazy loading is disabled, `state` is empty.
return Ok(vec![]);
}
/*
at this point, either the timeline is `limited` or the DAG has a split in it. this necessitates
computing the incremental state (which may be empty).
NOTE: this code path does not apply lazy-load filtering to membership state events. the spec forbids lazy-load filtering
if the timeline is `limited`, and DAG splits which require sending extra membership state events are (probably) uncommon
enough that the performance penalty is acceptable.
*/
trace!(?timeline_is_linear, ?timeline.limited, "computing state for incremental sync");
// fetch the shorteventids of state events in the timeline
let state_events_in_timeline: BTreeSet<ShortEventId> = services
.rooms
.short
.multi_get_or_create_shorteventid(timeline.pdus.iter().filter_map(|(_, pdu)| {
if pdu.state_key().is_some() {
Some(pdu.event_id.as_ref())
} else {
None
}
}))
.collect()
.await;
trace!("{} state events in timeline", state_events_in_timeline.len());
/*
fetch the state events which were added since the last sync.
specifically we fetch the difference between the state at the last sync and the state at the _end_
of the timeline, and then we filter out state events in the timeline itself using the shorteventids we fetched.
this is necessary to account for splits in the DAG, as explained above.
*/
let state_diff = services
.rooms
.short
.multi_get_eventid_from_short::<'_, OwnedEventId, _>(
services
.rooms
.state_accessor
.state_added((previous_sync_end_shortstatehash, timeline_end_shortstatehash))
.await?
.stream()
.ready_filter_map(|(_, shorteventid)| {
if state_events_in_timeline.contains(&shorteventid) {
None
} else {
Some(shorteventid)
}
}),
)
.ignore_err();
// finally, fetch the PDU contents and collect them into a vec
let state_diff_pdus = state_diff
.broad_filter_map(|event_id| async move {
services
.rooms
.timeline
.get_non_outlier_pdu(&event_id)
.await
.ok()
})
.collect::<Vec<_>>()
.await;
trace!(?state_diff_pdus, "collected state PDUs for incremental sync");
Ok(state_diff_pdus)
}
async fn calculate_device_list_updates(
services: &Services,
SyncContext { sender_user, since, next_batch, .. }: SyncContext<'_>,
src/api/client/sync/v3/left.rs
+227 -145
View File
@@ -1,21 +1,29 @@
use std::collections::HashMap;
use conduwuit::{
Event, PduEvent, Result, error,
Event, PduCount, PduEvent, Result, at, debug_warn,
pdu::EventHash,
utils::{self, FutureBoolExt, TryFutureExtExt, future::ReadyEqExt},
warn,
trace,
utils::{self, IterStream, future::ReadyEqExt, stream::WidebandExt as _},
};
use futures::{FutureExt, StreamExt, pin_mut};
use futures::{StreamExt, future::join};
use ruma::{
EventId, OwnedEventId, OwnedRoomId, UserId,
EventId, OwnedRoomId, RoomId,
api::client::sync::sync_events::v3::{LeftRoom, RoomAccountData, State, Timeline},
events::{StateEventType, TimelineEventType::*},
events::{
StateEventType, TimelineEventType,
room::member::{MembershipChange, RoomMemberEventContent},
},
uint,
};
use service::{Services, rooms::lazy_loading::Options};
use serde_json::value::RawValue;
use service::Services;
use crate::client::sync::v3::SyncContext;
use crate::client::{
TimelinePdus, ignored_filter,
sync::{
load_timeline,
v3::{SyncContext, prepare_lazily_loaded_members, state::calculate_state_initial},
},
};
#[tracing::instrument(
name = "left",
@@ -23,174 +31,248 @@ use crate::client::sync::v3::SyncContext;
skip_all,
fields(
room_id = %room_id,
full = %full_state,
),
)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn load_left_room(
services: &Services,
SyncContext {
sender_user,
since,
next_batch,
full_state,
filter,
..
}: SyncContext<'_>,
sync_context: SyncContext<'_>,
ref room_id: OwnedRoomId,
leave_pdu: Option<PduEvent>,
) -> Result<Option<LeftRoom>> {
let left_count = services
let SyncContext {
sender_user, since, next_batch, filter, ..
} = sync_context;
// the global count as of the moment the user left the room
let Some(left_count) = services
.rooms
.state_cache
.get_left_count(room_id, sender_user)
.await
.ok();
.ok()
else {
// if we get here, the membership cache is incorrect, likely due to a state
// reset
debug_warn!("attempting to sync left room but no left count exists");
return Ok(None);
};
// Left before last sync
let include_leave = filter.room.include_leave;
if (since >= left_count && !include_leave) || Some(next_batch) < left_count {
// return early if we haven't gotten to this leave yet.
// this can happen if the user leaves while a sync response is being generated
if next_batch < left_count {
return Ok(None);
}
let is_not_found = services.rooms.metadata.exists(room_id).eq(&false);
let is_disabled = services.rooms.metadata.is_disabled(room_id);
let is_banned = services.rooms.metadata.is_banned(room_id);
pin_mut!(is_not_found, is_disabled, is_banned);
if is_not_found.or(is_disabled).or(is_banned).await {
// This is just a rejected invite, not a room we know
// Insert a leave event anyways for the client
let event = PduEvent {
event_id: EventId::new(services.globals.server_name()),
sender: sender_user.to_owned(),
origin: None,
origin_server_ts: utils::millis_since_unix_epoch()
.try_into()
.expect("Timestamp is valid js_int value"),
kind: RoomMember,
content: serde_json::from_str(r#"{"membership":"leave"}"#)
.expect("this is valid JSON"),
state_key: Some(sender_user.as_str().into()),
unsigned: None,
// The following keys are dropped on conversion
room_id: Some(room_id.clone()),
prev_events: vec![],
depth: uint!(1),
auth_events: vec![],
redacts: None,
hashes: EventHash { sha256: String::new() },
signatures: None,
};
return Ok(Some(LeftRoom {
account_data: RoomAccountData { events: Vec::new() },
timeline: Timeline {
limited: false,
prev_batch: Some(next_batch.to_string()),
events: Vec::new(),
},
state: State { events: vec![event.into_format()] },
}));
// return early if this is an incremental sync, and we've already synced this
// leave to the user, and `include_leave` isn't set on the filter.
if !include_leave && since.is_some_and(|since| since >= left_count) {
return Ok(None);
}
let mut left_state_events = Vec::new();
let since_state_ids = async {
let since_shortstatehash = services
.rooms
.user
.get_token_shortstatehash(room_id, since?)
.ok()
.await?;
services
.rooms
.state_accessor
.state_full_ids(since_shortstatehash)
.collect::<HashMap<_, OwnedEventId>>()
.map(Some)
.await
if let Some(ref leave_pdu) = leave_pdu {
debug_assert_eq!(leave_pdu.kind, TimelineEventType::RoomMember);
}
.await
.unwrap_or_default();
let Ok(left_event_id): Result<OwnedEventId> = services
.rooms
.state_accessor
.room_state_get_id(room_id, &StateEventType::RoomMember, sender_user.as_str())
.await
else {
warn!("Left {room_id} but no left state event");
return Ok(None);
};
let does_not_exist = services.rooms.metadata.exists(room_id).eq(&false).await;
let Ok(left_shortstatehash) = services
.rooms
.state_accessor
.pdu_shortstatehash(&left_event_id)
.await
else {
warn!(event_id = %left_event_id, "Leave event has no state in {room_id}");
return Ok(None);
};
let (timeline, state_events) = match leave_pdu {
| Some(leave_pdu) if does_not_exist => {
/*
we have none PDUs with left beef for this room, likely because it was a rejected invite to a room
which nobody on this homeserver is in. `leave_pdu` is the remote-assisted outlier leave event for the room,
which is all we can send to the client.
*/
trace!("syncing remote-assisted leave PDU");
(TimelinePdus::default(), vec![leave_pdu])
},
| Some(leave_pdu) => {
// we have this room in our DB, and can fetch the state and timeline from when
// the user left if they're allowed to see it.
let mut left_state_ids: HashMap<_, _> = services
.rooms
.state_accessor
.state_full_ids(left_shortstatehash)
.collect()
.await;
let leave_state_key = sender_user;
debug_assert_eq!(Some(leave_state_key.as_str()), leave_pdu.state_key());
let leave_shortstatekey = services
.rooms
.short
.get_or_create_shortstatekey(&StateEventType::RoomMember, sender_user.as_str())
.await;
left_state_ids.insert(leave_shortstatekey, left_event_id);
for (shortstatekey, event_id) in left_state_ids {
if full_state || since_state_ids.get(&shortstatekey) != Some(&event_id) {
let (event_type, state_key) = services
let leave_shortstatehash = services
.rooms
.short
.get_statekey_from_short(shortstatekey)
.state_accessor
.pdu_shortstatehash(&leave_pdu.event_id)
.await?;
if filter.room.state.lazy_load_options.is_enabled()
&& event_type == StateEventType::RoomMember
&& !full_state
&& state_key
.as_str()
.try_into()
.is_ok_and(|user_id: &UserId| sender_user != user_id)
{
continue;
let prev_member_event = services
.rooms
.state_accessor
.state_get(
leave_shortstatehash,
&StateEventType::RoomMember,
leave_state_key.as_str(),
)
.await?;
let current_membership: RoomMemberEventContent = leave_pdu.get_content()?;
let prev_membership: RoomMemberEventContent = prev_member_event.get_content()?;
match current_membership.membership_change(
Some(prev_membership.details()),
&leave_pdu.sender,
leave_state_key,
) {
| MembershipChange::Left => {
// if the user went from `join` to `leave`, they should be able to view the
// timeline.
let timeline_start_count = if let Some(since) = since {
// for incremental syncs, start the timeline after `since`
PduCount::Normal(since)
} else {
// for initial syncs, start the timeline at the previous membership event
services
.rooms
.timeline
.get_pdu_count(&prev_member_event.event_id)
.await?
.saturating_sub(1)
};
let timeline_end_count = services
.rooms
.timeline
.get_pdu_count(leave_pdu.event_id())
.await?;
let timeline = load_timeline(
services,
sender_user,
room_id,
Some(timeline_start_count),
Some(timeline_end_count),
10_usize,
)
.await?;
let timeline_start_shortstatehash = async {
if let Some((_, pdu)) = timeline.pdus.front() {
if let Ok(shortstatehash) = services
.rooms
.state_accessor
.pdu_shortstatehash(&pdu.event_id)
.await
{
return shortstatehash;
}
}
leave_shortstatehash
};
let lazily_loaded_members = prepare_lazily_loaded_members(
services,
sync_context,
room_id,
timeline.senders(),
);
let (timeline_start_shortstatehash, lazily_loaded_members) =
join(timeline_start_shortstatehash, lazily_loaded_members).await;
// TODO: calculate incremental state for incremental syncs.
// always calculating initial state _works_ but returns more data and does
// more processing than strictly necessary.
let state = calculate_state_initial(
services,
sender_user,
timeline_start_shortstatehash,
lazily_loaded_members.as_ref(),
)
.await?;
trace!(
?timeline_start_count,
?timeline_end_count,
"syncing {} timeline events (limited = {}) and {} state events",
timeline.pdus.len(),
timeline.limited,
state.len()
);
(timeline, state)
},
| other_membership => {
// otherwise, the user should not be able to view the timeline.
// only return their leave event.
trace!(
?other_membership,
"user did not leave happily, only syncing leave event"
);
(TimelinePdus::default(), vec![leave_pdu])
},
}
},
| None => {
/*
no leave event was actually sent in this room, but we still need to pretend
like the user left it. this is usually because the room was banned by a server admin.
generate a fake leave event to placate the client.
*/
trace!("syncing dummy leave event");
(TimelinePdus::default(), vec![create_dummy_leave_event(
services,
sync_context,
room_id,
)])
},
};
let Ok(pdu) = services.rooms.timeline.get_pdu(&event_id).await else {
error!("Pdu in state not found: {event_id}");
continue;
};
if !include_leave && pdu.sender == sender_user {
continue;
}
left_state_events.push(pdu.into_format());
}
}
let raw_timeline_pdus = timeline
.pdus
.into_iter()
.stream()
// filter out ignored events from the timeline
.wide_filter_map(|item| ignored_filter(services, item, sender_user))
.map(at!(1))
.map(Event::into_format)
.collect::<Vec<_>>()
.await;
Ok(Some(LeftRoom {
account_data: RoomAccountData { events: Vec::new() },
timeline: Timeline {
// TODO: support left timeline events so we dont need to set limited to true
limited: true,
limited: timeline.limited,
prev_batch: Some(next_batch.to_string()),
events: Vec::new(), // and so we dont need to set this to empty vec
events: raw_timeline_pdus,
},
state: State {
events: state_events.into_iter().map(Event::into_format).collect(),
},
state: State { events: left_state_events },
}))
}
fn create_dummy_leave_event(
services: &Services,
SyncContext { sender_user, .. }: SyncContext<'_>,
room_id: &RoomId,
) -> PduEvent {
// TODO: because this event ID is random, it could cause caching issues with
// clients. perhaps a database table could be created to hold these dummy
// events, or they could be stored as outliers?
PduEvent {
event_id: EventId::new(services.globals.server_name()),
sender: sender_user.to_owned(),
origin: None,
origin_server_ts: utils::millis_since_unix_epoch()
.try_into()
.expect("Timestamp is valid js_int value"),
kind: TimelineEventType::RoomMember,
content: RawValue::from_string(r#"{"membership": "leave"}"#.to_owned()).unwrap(),
state_key: Some(sender_user.as_str().into()),
unsigned: None,
// The following keys are dropped on conversion
room_id: Some(room_id.to_owned()),
prev_events: vec![],
depth: uint!(1),
auth_events: vec![],
redacts: None,
hashes: EventHash { sha256: String::new() },
signatures: None,
}
}
src/api/client/sync/v3/mod.rs
+55 -3
View File
@@ -1,5 +1,6 @@
mod joined;
mod left;
mod state;
use std::{
cmp::{self},
@@ -22,7 +23,7 @@ use futures::{
future::{OptionFuture, join3, join4, join5},
};
use ruma::{
DeviceId, OwnedUserId, UserId,
DeviceId, OwnedUserId, RoomId, UserId,
api::client::{
filter::FilterDefinition,
sync::sync_events::{
@@ -40,6 +41,7 @@ use ruma::{
},
serde::Raw,
};
use service::rooms::lazy_loading::{self, MemberSet, Options as _};
use super::{load_timeline, share_encrypted_room};
use crate::{
@@ -88,6 +90,25 @@ struct SyncContext<'a> {
filter: &'a FilterDefinition,
}
impl<'a> SyncContext<'a> {
fn lazy_loading_context(&self, room_id: &'a RoomId) -> lazy_loading::Context<'a> {
lazy_loading::Context {
user_id: self.sender_user,
device_id: Some(self.sender_device),
room_id,
token: self.since,
options: Some(&self.filter.room.state.lazy_load_options),
}
}
#[inline]
fn lazy_loading_enabled(&self) -> bool {
(self.filter.room.state.lazy_load_options.is_enabled()
|| self.filter.room.timeline.lazy_load_options.is_enabled())
&& !self.full_state
}
}
type PresenceUpdates = HashMap<OwnedUserId, PresenceEventContent>;
/// # `GET /_matrix/client/r0/sync`
@@ -239,8 +260,8 @@ pub(crate) async fn build_sync_events(
.rooms
.state_cache
.rooms_left(sender_user)
.broad_filter_map(|(room_id, _)| {
load_left_room(services, context, room_id.clone())
.broad_filter_map(|(room_id, leave_pdu)| {
load_left_room(services, context, room_id.clone(), leave_pdu)
.map_ok(move |left_room| (room_id, left_room))
.ok()
})
@@ -400,3 +421,34 @@ async fn process_presence_updates(
.collect()
.await
}
async fn prepare_lazily_loaded_members(
services: &Services,
sync_context: SyncContext<'_>,
room_id: &RoomId,
timeline_members: impl Iterator<Item = OwnedUserId>,
) -> Option<MemberSet> {
let lazy_loading_context = &sync_context.lazy_loading_context(room_id);
// the user IDs of members whose membership needs to be sent to the client, if
// lazy-loading is enabled.
let lazily_loaded_members =
OptionFuture::from(sync_context.lazy_loading_enabled().then(|| {
services
.rooms
.lazy_loading
.retain_lazy_members(timeline_members.collect(), lazy_loading_context)
}))
.await;
// reset lazy loading state on initial sync
if sync_context.since.is_none() {
services
.rooms
.lazy_loading
.reset(lazy_loading_context)
.await;
}
lazily_loaded_members
}
src/api/client/sync/v3/state.rs
+268
View File
@@ -0,0 +1,268 @@
use std::{collections::BTreeSet, ops::ControlFlow};
use conduwuit::{
Result, at, is_equal_to,
matrix::{
Event,
pdu::{PduCount, PduEvent},
},
utils::{
BoolExt, IterStream, ReadyExt, TryFutureExtExt,
stream::{BroadbandExt, TryIgnore},
},
};
use conduwuit_service::{
Services,
rooms::{lazy_loading::MemberSet, short::ShortStateHash},
};
use futures::{FutureExt, StreamExt};
use itertools::Itertools;
use ruma::{OwnedEventId, RoomId, UserId, events::StateEventType};
use service::rooms::short::ShortEventId;
use tracing::trace;
use crate::client::TimelinePdus;
/// Calculate the state events to include in an initial sync response.
///
/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
/// Vec will include the membership events of exclusively the members in
/// `lazily_loaded_members`.
#[tracing::instrument(
name = "initial",
level = "trace",
skip_all,
fields(current_shortstatehash)
)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn calculate_state_initial(
services: &Services,
sender_user: &UserId,
timeline_start_shortstatehash: ShortStateHash,
lazily_loaded_members: Option<&MemberSet>,
) -> Result<Vec<PduEvent>> {
// load the keys and event IDs of the state events at the start of the timeline
let (shortstatekeys, event_ids): (Vec<_>, Vec<_>) = services
.rooms
.state_accessor
.state_full_ids(timeline_start_shortstatehash)
.unzip()
.await;
trace!("performing initial sync of {} state events", event_ids.len());
services
.rooms
.short
// look up the full state keys
.multi_get_statekey_from_short(shortstatekeys.into_iter().stream())
.zip(event_ids.into_iter().stream())
.ready_filter_map(|item| Some((item.0.ok()?, item.1)))
.ready_filter_map(|((event_type, state_key), event_id)| {
if let Some(lazily_loaded_members) = lazily_loaded_members {
/*
if lazy loading is enabled, filter out membership events which aren't for a user
included in `lazily_loaded_members` or for the user requesting the sync.
*/
let event_is_redundant = event_type == StateEventType::RoomMember
&& state_key.as_str().try_into().is_ok_and(|user_id: &UserId| {
sender_user != user_id && !lazily_loaded_members.contains(user_id)
});
event_is_redundant.or_some(event_id)
} else {
Some(event_id)
}
})
.broad_filter_map(|event_id: OwnedEventId| async move {
services.rooms.timeline.get_pdu(&event_id).await.ok()
})
.collect()
.map(Ok)
.await
}
/// Calculate the state events to include in an incremental sync response.
///
/// If lazy-loading is enabled (`lazily_loaded_members` is Some), the returned
/// Vec will include the membership events of all the members in
/// `lazily_loaded_members`.
#[tracing::instrument(name = "incremental", level = "trace", skip_all)]
#[allow(clippy::too_many_arguments)]
pub(super) async fn calculate_state_incremental<'a>(
services: &Services,
sender_user: &'a UserId,
room_id: &RoomId,
previous_sync_end_count: PduCount,
previous_sync_end_shortstatehash: ShortStateHash,
timeline_start_shortstatehash: ShortStateHash,
timeline_end_shortstatehash: ShortStateHash,
timeline: &TimelinePdus,
lazily_loaded_members: Option<&'a MemberSet>,
) -> Result<Vec<PduEvent>> {
// NB: a limited sync is one where `timeline.limited == true`. Synapse calls
// this a "gappy" sync internally.
/*
the state events returned from an incremental sync which isn't limited are usually empty.
however, if an event in the timeline (`timeline.pdus`) merges a split in the room's DAG (i.e. has multiple `prev_events`),
the state at the _end_ of the timeline may include state events which were merged in and don't exist in the state
at the _start_ of the timeline. because this is uncommon, we check here to see if any events in the timeline
merged a split in the DAG.
see: https://github.com/element-hq/synapse/issues/16941
*/
let timeline_is_linear = timeline.pdus.is_empty() || {
let last_pdu_of_last_sync = services
.rooms
.timeline
.pdus_rev(Some(sender_user), room_id, Some(previous_sync_end_count.saturating_add(1)))
.boxed()
.next()
.await
.transpose()
.expect("last sync should have had some PDUs")
.map(at!(1));
// make sure the prev_events of each pdu in the timeline refer only to the
// previous pdu
timeline
.pdus
.iter()
.try_fold(last_pdu_of_last_sync.map(|pdu| pdu.event_id), |prev_event_id, (_, pdu)| {
if let Ok(pdu_prev_event_id) = pdu.prev_events.iter().exactly_one() {
if prev_event_id
.as_ref()
.is_none_or(is_equal_to!(pdu_prev_event_id))
{
return ControlFlow::Continue(Some(pdu_prev_event_id.to_owned()));
}
}
trace!(
"pdu {:?} has split prev_events (expected {:?}): {:?}",
pdu.event_id, prev_event_id, pdu.prev_events
);
ControlFlow::Break(())
})
.is_continue()
};
if timeline_is_linear && !timeline.limited {
// if there are no splits in the DAG and the timeline isn't limited, then
// `state` will always be empty unless lazy loading is enabled.
if let Some(lazily_loaded_members) = lazily_loaded_members
&& !timeline.pdus.is_empty()
{
// lazy loading is enabled, so we return the membership events which were
// requested by the caller.
let lazy_membership_events: Vec<_> = lazily_loaded_members
.iter()
.stream()
.broad_filter_map(|user_id| async move {
if user_id == sender_user {
return None;
}
services
.rooms
.state_accessor
.state_get(
timeline_start_shortstatehash,
&StateEventType::RoomMember,
user_id.as_str(),
)
.ok()
.await
})
.collect()
.await;
if !lazy_membership_events.is_empty() {
trace!(
"syncing lazy membership events for members: {:?}",
lazy_membership_events
.iter()
.map(|pdu| pdu.state_key().unwrap())
.collect::<Vec<_>>()
);
}
return Ok(lazy_membership_events);
}
// lazy loading is disabled, `state` is empty.
return Ok(vec![]);
}
/*
at this point, either the timeline is `limited` or the DAG has a split in it. this necessitates
computing the incremental state (which may be empty).
NOTE: this code path does not apply lazy-load filtering to membership state events. the spec forbids lazy-load filtering
if the timeline is `limited`, and DAG splits which require sending extra membership state events are (probably) uncommon
enough that the performance penalty is acceptable.
*/
trace!(?timeline_is_linear, ?timeline.limited, "computing state for incremental sync");
// fetch the shorteventids of state events in the timeline
let state_events_in_timeline: BTreeSet<ShortEventId> = services
.rooms
.short
.multi_get_or_create_shorteventid(timeline.pdus.iter().filter_map(|(_, pdu)| {
if pdu.state_key().is_some() {
Some(pdu.event_id.as_ref())
} else {
None
}
}))
.collect()
.await;
trace!("{} state events in timeline", state_events_in_timeline.len());
/*
fetch the state events which were added since the last sync.
specifically we fetch the difference between the state at the last sync and the state at the _end_
of the timeline, and then we filter out state events in the timeline itself using the shorteventids we fetched.
this is necessary to account for splits in the DAG, as explained above.
*/
let state_diff = services
.rooms
.short
.multi_get_eventid_from_short::<'_, OwnedEventId, _>(
services
.rooms
.state_accessor
.state_added((previous_sync_end_shortstatehash, timeline_end_shortstatehash))
.await?
.stream()
.ready_filter_map(|(_, shorteventid)| {
if state_events_in_timeline.contains(&shorteventid) {
None
} else {
Some(shorteventid)
}
}),
)
.ignore_err();
// finally, fetch the PDU contents and collect them into a vec
let state_diff_pdus = state_diff
.broad_filter_map(|event_id| async move {
services
.rooms
.timeline
.get_non_outlier_pdu(&event_id)
.await
.ok()
})
.collect::<Vec<_>>()
.await;
trace!(?state_diff_pdus, "collected state PDUs for incremental sync");
Ok(state_diff_pdus)
}