chore(sync/v3): Use more descriptive names for SyncContext properties

This commit is contained in:
Ginger
2025-12-03 16:04:32 +00:00
parent eba5f16e09
commit be743ec70a
4 changed files with 151 additions and 112 deletions
+66 -60
View File
@@ -58,20 +58,15 @@ pub(super) async fn load_joined_room(
ref room_id: OwnedRoomId, ref room_id: OwnedRoomId,
) -> Result<(JoinedRoom, DeviceListUpdates)> { ) -> Result<(JoinedRoom, DeviceListUpdates)> {
let SyncContext { let SyncContext {
sender_user, syncing_user,
since, last_sync_end_count,
next_batch, current_count,
full_state, full_state,
.. ..
} = sync_context; } = sync_context;
// the global count as of the end of the last sync.
// this will be None if we are doing an initial sync.
let previous_sync_end_count = since.map(PduCount::Normal);
let next_batchcount = PduCount::Normal(next_batch);
let mut device_list_updates = DeviceListUpdates::new(); let mut device_list_updates = DeviceListUpdates::new();
// the room state right now // the room state as of `next_batch`.
let current_shortstatehash = services let current_shortstatehash = services
.rooms .rooms
.state .state
@@ -81,36 +76,39 @@ pub(super) async fn load_joined_room(
// the room state as of the end of the last sync. // the room state as of the end of the last sync.
// this will be None if we are doing an initial sync or if we just joined this // this will be None if we are doing an initial sync or if we just joined this
// room. // room.
let previous_sync_end_shortstatehash = OptionFuture::from(since.map(|since| { let last_sync_end_shortstatehash =
services OptionFuture::from(last_sync_end_count.map(|last_sync_end_count| {
.rooms services
.user .rooms
.get_token_shortstatehash(room_id, since) .user
.ok() .get_token_shortstatehash(room_id, last_sync_end_count)
})) .ok()
.map(Option::flatten) }))
.map(Ok); .map(Option::flatten)
.map(Ok);
let (current_shortstatehash, previous_sync_end_shortstatehash) = let (current_shortstatehash, last_sync_end_shortstatehash) =
try_join(current_shortstatehash, previous_sync_end_shortstatehash).await?; try_join(current_shortstatehash, last_sync_end_shortstatehash).await?;
// load recent timeline events.
let timeline = load_timeline( let timeline = load_timeline(
services, services,
sender_user, syncing_user,
room_id, room_id,
previous_sync_end_count, last_sync_end_count.map(PduCount::Normal),
Some(next_batchcount), Some(PduCount::Normal(current_count)),
services.config.incremental_sync_max_timeline_size, services.config.incremental_sync_max_timeline_size,
); );
let receipt_events = services let receipt_events = services
.rooms .rooms
.read_receipt .read_receipt
.readreceipts_since(room_id, since) .readreceipts_since(room_id, last_sync_end_count)
.filter_map(|(read_user, _, edu)| async move { .filter_map(|(read_user, _, edu)| async move {
services services
.users .users
.user_is_ignored(read_user, sender_user) .user_is_ignored(read_user, syncing_user)
.await .await
.or_some((read_user.to_owned(), edu)) .or_some((read_user.to_owned(), edu))
}) })
@@ -142,13 +140,13 @@ pub(super) async fn load_joined_room(
services services
.rooms .rooms
.user .user
.last_notification_read(sender_user, room_id) .last_notification_read(syncing_user, room_id)
}) })
.into(); .into();
// the syncing user's membership event during the last sync. // the syncing user's membership event during the last sync.
// this will be None if `previous_sync_end_shortstatehash` is None. // this will be None if `previous_sync_end_shortstatehash` is None.
let membership_during_previous_sync: OptionFuture<_> = previous_sync_end_shortstatehash let membership_during_previous_sync: OptionFuture<_> = last_sync_end_shortstatehash
.map(|shortstatehash| { .map(|shortstatehash| {
services services
.rooms .rooms
@@ -156,7 +154,7 @@ pub(super) async fn load_joined_room(
.state_get_content( .state_get_content(
shortstatehash, shortstatehash,
&StateEventType::RoomMember, &StateEventType::RoomMember,
sender_user.as_str(), syncing_user.as_str(),
) )
.ok() .ok()
}) })
@@ -199,16 +197,16 @@ pub(super) async fn load_joined_room(
*or* we just joined this room, `calculate_state_initial` will be used, otherwise `calculate_state_incremental` *or* we just joined this room, `calculate_state_initial` will be used, otherwise `calculate_state_incremental`
will be used. will be used.
*/ */
let mut state_events = if let Some(previous_sync_end_count) = previous_sync_end_count let mut state_events = if let Some(last_sync_end_count) = last_sync_end_count
&& let Some(previous_sync_end_shortstatehash) = previous_sync_end_shortstatehash && let Some(last_sync_end_shortstatehash) = last_sync_end_shortstatehash
&& !full_state && !full_state
{ {
calculate_state_incremental( calculate_state_incremental(
services, services,
sender_user, syncing_user,
room_id, room_id,
previous_sync_end_count, PduCount::Normal(last_sync_end_count),
previous_sync_end_shortstatehash, last_sync_end_shortstatehash,
timeline_start_shortstatehash, timeline_start_shortstatehash,
current_shortstatehash, current_shortstatehash,
&timeline, &timeline,
@@ -219,7 +217,7 @@ pub(super) async fn load_joined_room(
} else { } else {
calculate_state_initial( calculate_state_initial(
services, services,
sender_user, syncing_user,
timeline_start_shortstatehash, timeline_start_shortstatehash,
lazily_loaded_members.as_ref(), lazily_loaded_members.as_ref(),
) )
@@ -228,7 +226,7 @@ pub(super) async fn load_joined_room(
}; };
// for incremental syncs, calculate updates to E2EE device lists // for incremental syncs, calculate updates to E2EE device lists
if previous_sync_end_count.is_some() && is_encrypted_room { if last_sync_end_count.is_some() && is_encrypted_room {
calculate_device_list_updates( calculate_device_list_updates(
services, services,
sync_context, sync_context,
@@ -243,8 +241,9 @@ pub(super) async fn load_joined_room(
// only compute room counts and heroes (aka the summary) if the room's members // only compute room counts and heroes (aka the summary) if the room's members
// changed since the last sync // changed since the last sync
let (joined_member_count, invited_member_count, heroes) = let (joined_member_count, invited_member_count, heroes) =
// calculate counts if any of the state events we're syncing are of type `m.room.member`
if state_events.iter().any(|event| event.kind == RoomMember) { if state_events.iter().any(|event| event.kind == RoomMember) {
calculate_counts(services, room_id, sender_user).await? calculate_counts(services, room_id, syncing_user).await?
} else { } else {
(None, None, None) (None, None, None)
}; };
@@ -254,7 +253,7 @@ pub(super) async fn load_joined_room(
&& pdu && pdu
.state_key .state_key
.as_deref() .as_deref()
.is_some_and(is_equal_to!(sender_user.as_str())) .is_some_and(is_equal_to!(syncing_user.as_str()))
}; };
// the membership event of the syncing user, if they joined since the last sync // the membership event of the syncing user, if they joined since the last sync
@@ -272,7 +271,7 @@ pub(super) async fn load_joined_room(
let prev_batch = timeline.pdus.front().map(at!(0)).or_else(|| { let prev_batch = timeline.pdus.front().map(at!(0)).or_else(|| {
sender_join_membership_event sender_join_membership_event
.is_some() .is_some()
.and(since) .and(last_sync_end_count)
.map(Into::into) .map(Into::into)
}); });
@@ -281,7 +280,7 @@ pub(super) async fn load_joined_room(
.into_iter() .into_iter()
.stream() .stream()
// filter out ignored events from the timeline // filter out ignored events from the timeline
.wide_filter_map(|item| ignored_filter(services, item, sender_user)) .wide_filter_map(|item| ignored_filter(services, item, syncing_user))
.map(at!(1)) .map(at!(1))
// if the syncing user just joined, add their membership event to the timeline // if the syncing user just joined, add their membership event to the timeline
.chain(sender_join_membership_event.into_iter().stream()) .chain(sender_join_membership_event.into_iter().stream())
@@ -290,7 +289,7 @@ pub(super) async fn load_joined_room(
let account_data_events = services let account_data_events = services
.account_data .account_data
.changes_since(Some(room_id), sender_user, since, Some(next_batch)) .changes_since(Some(room_id), syncing_user, last_sync_end_count, Some(current_count))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room)) .ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Room))
.collect(); .collect();
@@ -301,7 +300,8 @@ pub(super) async fn load_joined_room(
3. the last notification the user saw has changed since the last sync 3. the last notification the user saw has changed since the last sync
*/ */
let send_notification_counts = last_notification_read.is_none_or(|last_notification_read| { let send_notification_counts = last_notification_read.is_none_or(|last_notification_read| {
since.is_none_or(|since| last_notification_read > since) last_sync_end_count
.is_none_or(|last_sync_end_count| last_notification_read > last_sync_end_count)
}); });
let notification_count: OptionFuture<_> = send_notification_counts let notification_count: OptionFuture<_> = send_notification_counts
@@ -309,7 +309,7 @@ pub(super) async fn load_joined_room(
services services
.rooms .rooms
.user .user
.notification_count(sender_user, room_id) .notification_count(syncing_user, room_id)
.map(TryInto::try_into) .map(TryInto::try_into)
.unwrap_or(uint!(0)) .unwrap_or(uint!(0))
}) })
@@ -320,7 +320,7 @@ pub(super) async fn load_joined_room(
services services
.rooms .rooms
.user .user
.highlight_count(sender_user, room_id) .highlight_count(syncing_user, room_id)
.map(TryInto::try_into) .map(TryInto::try_into)
.unwrap_or(uint!(0)) .unwrap_or(uint!(0))
}) })
@@ -331,14 +331,15 @@ pub(super) async fn load_joined_room(
.typing .typing
.last_typing_update(room_id) .last_typing_update(room_id)
.and_then(|count| async move { .and_then(|count| async move {
if since.is_some_and(|since| count <= since) { if last_sync_end_count.is_some_and(|last_sync_end_count| count <= last_sync_end_count)
{
return Ok(Vec::<Raw<AnySyncEphemeralRoomEvent>>::new()); return Ok(Vec::<Raw<AnySyncEphemeralRoomEvent>>::new());
} }
let typings = services let typings = services
.rooms .rooms
.typing .typing
.typings_event_for_user(room_id, sender_user) .typings_event_for_user(room_id, syncing_user)
.await?; .await?;
Ok(vec![serde_json::from_str(&serde_json::to_string(&typings)?)?]) Ok(vec![serde_json::from_str(&serde_json::to_string(&typings)?)?])
@@ -352,12 +353,12 @@ pub(super) async fn load_joined_room(
let (room_events, account_data_events, typing_events) = events; let (room_events, account_data_events, typing_events) = events;
let (notification_count, highlight_count) = unread_notifications; let (notification_count, highlight_count) = unread_notifications;
let last_privateread_update = if let Some(since) = since { let last_privateread_update = if let Some(last_sync_end_count) = last_sync_end_count {
services services
.rooms .rooms
.read_receipt .read_receipt
.last_privateread_update(sender_user, room_id) .last_privateread_update(syncing_user, room_id)
.await > since .await > last_sync_end_count
} else { } else {
true true
}; };
@@ -366,7 +367,7 @@ pub(super) async fn load_joined_room(
services services
.rooms .rooms
.read_receipt .read_receipt
.private_read_get(room_id, sender_user) .private_read_get(room_id, syncing_user)
.await .await
.ok() .ok()
} else { } else {
@@ -383,7 +384,7 @@ pub(super) async fn load_joined_room(
services services
.rooms .rooms
.user .user
.associate_token_shortstatehash(room_id, next_batch, current_shortstatehash) .associate_token_shortstatehash(room_id, current_count, current_shortstatehash)
.await; .await;
let joined_room = JoinedRoom { let joined_room = JoinedRoom {
@@ -417,7 +418,12 @@ pub(super) async fn load_joined_room(
async fn calculate_device_list_updates( async fn calculate_device_list_updates(
services: &Services, services: &Services,
SyncContext { sender_user, since, next_batch, .. }: SyncContext<'_>, SyncContext {
syncing_user,
last_sync_end_count: since,
current_count,
..
}: SyncContext<'_>,
room_id: &RoomId, room_id: &RoomId,
device_list_updates: &mut DeviceListUpdates, device_list_updates: &mut DeviceListUpdates,
state_events: &Vec<PduEvent>, state_events: &Vec<PduEvent>,
@@ -426,7 +432,7 @@ async fn calculate_device_list_updates(
// add users with changed keys to the `changed` list // add users with changed keys to the `changed` list
services services
.users .users
.room_keys_changed(room_id, since, Some(next_batch)) .room_keys_changed(room_id, since, Some(current_count))
.map(at!(0)) .map(at!(0))
.map(ToOwned::to_owned) .map(ToOwned::to_owned)
.ready_for_each(|user_id| { .ready_for_each(|user_id| {
@@ -456,7 +462,7 @@ async fn calculate_device_list_updates(
if matches!(content.membership, Leave | Join) { if matches!(content.membership, Leave | Join) {
let shares_encrypted_room = let shares_encrypted_room =
share_encrypted_room(services, sender_user, &user_id, Some(room_id)) share_encrypted_room(services, syncing_user, &user_id, Some(room_id))
.await; .await;
match content.membership { match content.membership {
| Leave if !shares_encrypted_room => { | Leave if !shares_encrypted_room => {
@@ -476,7 +482,7 @@ async fn calculate_device_list_updates(
async fn calculate_counts( async fn calculate_counts(
services: &Services, services: &Services,
room_id: &RoomId, room_id: &RoomId,
sender_user: &UserId, syncing_user: &UserId,
) -> Result<(Option<u64>, Option<u64>, Option<Vec<OwnedUserId>>)> { ) -> Result<(Option<u64>, Option<u64>, Option<Vec<OwnedUserId>>)> {
let joined_member_count = services let joined_member_count = services
.rooms .rooms
@@ -496,7 +502,7 @@ async fn calculate_counts(
let small_room = joined_member_count.saturating_add(invited_member_count) <= 5; let small_room = joined_member_count.saturating_add(invited_member_count) <= 5;
let heroes: OptionFuture<_> = small_room let heroes: OptionFuture<_> = small_room
.then(|| calculate_heroes(services, room_id, sender_user)) .then(|| calculate_heroes(services, room_id, syncing_user))
.into(); .into();
Ok((Some(joined_member_count), Some(invited_member_count), heroes.await)) Ok((Some(joined_member_count), Some(invited_member_count), heroes.await))
@@ -505,15 +511,15 @@ async fn calculate_counts(
async fn calculate_heroes( async fn calculate_heroes(
services: &Services, services: &Services,
room_id: &RoomId, room_id: &RoomId,
sender_user: &UserId, syncing_user: &UserId,
) -> Vec<OwnedUserId> { ) -> Vec<OwnedUserId> {
services services
.rooms .rooms
.timeline .timeline
.all_pdus(sender_user, room_id) .all_pdus(syncing_user, room_id)
.ready_filter(|(_, pdu)| pdu.kind == RoomMember) .ready_filter(|(_, pdu)| pdu.kind == RoomMember)
.fold_default(|heroes: Vec<_>, (_, pdu)| { .fold_default(|heroes: Vec<_>, (_, pdu)| {
fold_hero(heroes, services, room_id, sender_user, pdu) fold_hero(heroes, services, room_id, syncing_user, pdu)
}) })
.await .await
} }
@@ -522,7 +528,7 @@ async fn fold_hero(
mut heroes: Vec<OwnedUserId>, mut heroes: Vec<OwnedUserId>,
services: &Services, services: &Services,
room_id: &RoomId, room_id: &RoomId,
sender_user: &UserId, syncing_user: &UserId,
pdu: PduEvent, pdu: PduEvent,
) -> Vec<OwnedUserId> { ) -> Vec<OwnedUserId> {
let Some(user_id): Option<&UserId> = let Some(user_id): Option<&UserId> =
@@ -531,7 +537,7 @@ async fn fold_hero(
return heroes; return heroes;
}; };
if user_id == sender_user { if user_id == syncing_user {
return heroes; return heroes;
} }
+20 -14
View File
@@ -41,14 +41,18 @@ pub(super) async fn load_left_room(
leave_pdu: Option<PduEvent>, leave_pdu: Option<PduEvent>,
) -> Result<Option<LeftRoom>> { ) -> Result<Option<LeftRoom>> {
let SyncContext { let SyncContext {
sender_user, since, next_batch, filter, .. syncing_user,
last_sync_end_count,
current_count,
filter,
..
} = sync_context; } = sync_context;
// the global count as of the moment the user left the room // the global count as of the moment the user left the room
let Some(left_count) = services let Some(left_count) = services
.rooms .rooms
.state_cache .state_cache
.get_left_count(room_id, sender_user) .get_left_count(room_id, syncing_user)
.await .await
.ok() .ok()
else { else {
@@ -62,13 +66,13 @@ pub(super) async fn load_left_room(
// return early if we haven't gotten to this leave yet. // 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 // this can happen if the user leaves while a sync response is being generated
if next_batch < left_count { if current_count < left_count {
return Ok(None); return Ok(None);
} }
// return early if this is an incremental sync, and we've already synced this // 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. // leave to the user, and `include_leave` isn't set on the filter.
if !include_leave && since.is_some_and(|since| since >= left_count) { if !include_leave && last_sync_end_count >= Some(left_count) {
return Ok(None); return Ok(None);
} }
@@ -96,7 +100,7 @@ pub(super) async fn load_left_room(
// we have this room in our DB, and can fetch the state and timeline from when // 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. // the user left if they're allowed to see it.
let leave_state_key = sender_user; let leave_state_key = syncing_user;
debug_assert_eq!( debug_assert_eq!(
Some(leave_state_key.as_str()), Some(leave_state_key.as_str()),
leave_pdu.state_key(), leave_pdu.state_key(),
@@ -130,9 +134,11 @@ pub(super) async fn load_left_room(
// if the user went from `join` to `leave`, they should be able to view the // if the user went from `join` to `leave`, they should be able to view the
// timeline. // timeline.
let timeline_start_count = if let Some(since) = since { let timeline_start_count = if let Some(last_sync_end_count) =
last_sync_end_count
{
// for incremental syncs, start the timeline after `since` // for incremental syncs, start the timeline after `since`
PduCount::Normal(since) PduCount::Normal(last_sync_end_count)
} else { } else {
// for initial syncs, start the timeline at the previous membership event // for initial syncs, start the timeline at the previous membership event
services services
@@ -150,7 +156,7 @@ pub(super) async fn load_left_room(
let timeline = load_timeline( let timeline = load_timeline(
services, services,
sender_user, syncing_user,
room_id, room_id,
Some(timeline_start_count), Some(timeline_start_count),
Some(timeline_end_count), Some(timeline_end_count),
@@ -188,7 +194,7 @@ pub(super) async fn load_left_room(
// more processing than strictly necessary. // more processing than strictly necessary.
let state = calculate_state_initial( let state = calculate_state_initial(
services, services,
sender_user, syncing_user,
timeline_start_shortstatehash, timeline_start_shortstatehash,
lazily_loaded_members.as_ref(), lazily_loaded_members.as_ref(),
) )
@@ -236,7 +242,7 @@ pub(super) async fn load_left_room(
.into_iter() .into_iter()
.stream() .stream()
// filter out ignored events from the timeline // filter out ignored events from the timeline
.wide_filter_map(|item| ignored_filter(services, item, sender_user)) .wide_filter_map(|item| ignored_filter(services, item, syncing_user))
.map(at!(1)) .map(at!(1))
.map(Event::into_format) .map(Event::into_format)
.collect::<Vec<_>>() .collect::<Vec<_>>()
@@ -246,7 +252,7 @@ pub(super) async fn load_left_room(
account_data: RoomAccountData { events: Vec::new() }, account_data: RoomAccountData { events: Vec::new() },
timeline: Timeline { timeline: Timeline {
limited: timeline.limited, limited: timeline.limited,
prev_batch: Some(next_batch.to_string()), prev_batch: Some(current_count.to_string()),
events: raw_timeline_pdus, events: raw_timeline_pdus,
}, },
state: State { state: State {
@@ -257,7 +263,7 @@ pub(super) async fn load_left_room(
fn create_dummy_leave_event( fn create_dummy_leave_event(
services: &Services, services: &Services,
SyncContext { sender_user, .. }: SyncContext<'_>, SyncContext { syncing_user, .. }: SyncContext<'_>,
room_id: &RoomId, room_id: &RoomId,
) -> PduEvent { ) -> PduEvent {
// TODO: because this event ID is random, it could cause caching issues with // TODO: because this event ID is random, it could cause caching issues with
@@ -265,14 +271,14 @@ fn create_dummy_leave_event(
// events, or they could be stored as outliers? // events, or they could be stored as outliers?
PduEvent { PduEvent {
event_id: EventId::new(services.globals.server_name()), event_id: EventId::new(services.globals.server_name()),
sender: sender_user.to_owned(), sender: syncing_user.to_owned(),
origin: None, origin: None,
origin_server_ts: utils::millis_since_unix_epoch() origin_server_ts: utils::millis_since_unix_epoch()
.try_into() .try_into()
.expect("Timestamp is valid js_int value"), .expect("Timestamp is valid js_int value"),
kind: TimelineEventType::RoomMember, kind: TimelineEventType::RoomMember,
content: RawValue::from_string(r#"{"membership": "leave"}"#.to_owned()).unwrap(), content: RawValue::from_string(r#"{"membership": "leave"}"#.to_owned()).unwrap(),
state_key: Some(sender_user.as_str().into()), state_key: Some(syncing_user.as_str().into()),
unsigned: None, unsigned: None,
// The following keys are dropped on conversion // The following keys are dropped on conversion
room_id: Some(room_id.to_owned()), room_id: Some(room_id.to_owned()),
+61 -34
View File
@@ -52,6 +52,7 @@ use crate::{
}, },
}; };
/// A collection of updates to users' device lists, used for E2EE.
struct DeviceListUpdates { struct DeviceListUpdates {
changed: HashSet<OwnedUserId>, changed: HashSet<OwnedUserId>,
left: HashSet<OwnedUserId>, left: HashSet<OwnedUserId>,
@@ -80,23 +81,39 @@ impl From<DeviceListUpdates> for DeviceLists {
} }
} }
/// References to common data needed to calculate the sync response.
#[derive(Clone, Copy)] #[derive(Clone, Copy)]
struct SyncContext<'a> { struct SyncContext<'a> {
sender_user: &'a UserId, /// The ID of the user requesting this sync.
sender_device: &'a DeviceId, syncing_user: &'a UserId,
since: Option<u64>, /// The ID of the device requesting this sync, which will belong to
next_batch: u64, /// `syncing_user`.
syncing_device: &'a DeviceId,
/// The global count at the end of the previous sync response.
/// The previous sync's `current_count` will become the next sync's
/// `last_sync_end_count`. This will be None if no `since` query parameter
/// was specified, indicating an initial sync.
last_sync_end_count: Option<u64>,
/// The global count as of when we started building the sync response.
/// This is used as an upper bound when querying the database to ensure the
/// response represents a snapshot in time and doesn't include data which
/// appeared while the response was being built.
current_count: u64,
/// The `full_state` query parameter, used when syncing state for joined and
/// left rooms.
full_state: bool, full_state: bool,
/// The sync filter, which the client uses to specify what data should be
/// included in the sync response.
filter: &'a FilterDefinition, filter: &'a FilterDefinition,
} }
impl<'a> SyncContext<'a> { impl<'a> SyncContext<'a> {
fn lazy_loading_context(&self, room_id: &'a RoomId) -> lazy_loading::Context<'a> { fn lazy_loading_context(&self, room_id: &'a RoomId) -> lazy_loading::Context<'a> {
lazy_loading::Context { lazy_loading::Context {
user_id: self.sender_user, user_id: self.syncing_user,
device_id: Some(self.sender_device), device_id: Some(self.syncing_device),
room_id, room_id,
token: self.since, token: self.last_sync_end_count,
options: Some(&self.filter.room.state.lazy_load_options), options: Some(&self.filter.room.state.lazy_load_options),
} }
} }
@@ -196,10 +213,12 @@ pub(crate) async fn build_sync_events(
services: &Services, services: &Services,
body: &Ruma<sync_events::v3::Request>, body: &Ruma<sync_events::v3::Request>,
) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> { ) -> Result<sync_events::v3::Response, RumaResponse<UiaaResponse>> {
let (sender_user, sender_device) = body.sender(); let (syncing_user, syncing_device) = body.sender();
let next_batch = services.globals.current_count()?; let current_count = services.globals.current_count()?;
let since = body
// the `since` token is the last sync end count stringified
let last_sync_end_count = body
.body .body
.since .since
.as_ref() .as_ref()
@@ -209,20 +228,23 @@ pub(crate) async fn build_sync_events(
// FilterDefinition is very large (0x1000 bytes), let's put it on the heap // FilterDefinition is very large (0x1000 bytes), let's put it on the heap
let filter = Box::new(match body.body.filter.as_ref() { let filter = Box::new(match body.body.filter.as_ref() {
// use the default filter if none was specified
| None => FilterDefinition::default(), | None => FilterDefinition::default(),
// use inline filters directly
| Some(Filter::FilterDefinition(filter)) => filter.clone(), | Some(Filter::FilterDefinition(filter)) => filter.clone(),
// look up filter IDs from the database
| Some(Filter::FilterId(filter_id)) => services | Some(Filter::FilterId(filter_id)) => services
.users .users
.get_filter(sender_user, filter_id) .get_filter(syncing_user, filter_id)
.await .await
.unwrap_or_default(), .unwrap_or_default(),
}); });
let context = SyncContext { let context = SyncContext {
sender_user, syncing_user,
sender_device, syncing_device,
since, last_sync_end_count,
next_batch, current_count,
full_state, full_state,
filter: &filter, filter: &filter,
}; };
@@ -230,7 +252,7 @@ pub(crate) async fn build_sync_events(
let joined_rooms = services let joined_rooms = services
.rooms .rooms
.state_cache .state_cache
.rooms_joined(sender_user) .rooms_joined(syncing_user)
.map(ToOwned::to_owned) .map(ToOwned::to_owned)
.broad_filter_map(|room_id| async { .broad_filter_map(|room_id| async {
let joined_room = load_joined_room(services, context, room_id.clone()).await; let joined_room = load_joined_room(services, context, room_id.clone()).await;
@@ -259,7 +281,7 @@ pub(crate) async fn build_sync_events(
let left_rooms = services let left_rooms = services
.rooms .rooms
.state_cache .state_cache
.rooms_left(sender_user) .rooms_left(syncing_user)
.broad_filter_map(|(room_id, leave_pdu)| { .broad_filter_map(|(room_id, leave_pdu)| {
load_left_room(services, context, room_id.clone(), leave_pdu) load_left_room(services, context, room_id.clone(), leave_pdu)
.map_ok(move |left_room| (room_id, left_room)) .map_ok(move |left_room| (room_id, left_room))
@@ -271,9 +293,9 @@ pub(crate) async fn build_sync_events(
let invited_rooms = services let invited_rooms = services
.rooms .rooms
.state_cache .state_cache
.rooms_invited(sender_user) .rooms_invited(syncing_user)
.wide_filter_map(async |(room_id, invite_state)| { .wide_filter_map(async |(room_id, invite_state)| {
if is_ignored_invite(services, sender_user, &room_id).await { if is_ignored_invite(services, syncing_user, &room_id).await {
None None
} else { } else {
Some((room_id, invite_state)) Some((room_id, invite_state))
@@ -283,12 +305,12 @@ pub(crate) async fn build_sync_events(
let invite_count = services let invite_count = services
.rooms .rooms
.state_cache .state_cache
.get_invite_count(&room_id, sender_user) .get_invite_count(&room_id, syncing_user)
.await .await
.ok(); .ok();
// only sync this invite if it was sent after the last /sync call // only sync this invite if it was sent after the last /sync call
if since < invite_count { if last_sync_end_count < invite_count {
let invited_room = InvitedRoom { let invited_room = InvitedRoom {
invite_state: InviteState { events: invite_state }, invite_state: InviteState { events: invite_state },
}; };
@@ -301,17 +323,17 @@ pub(crate) async fn build_sync_events(
let knocked_rooms = services let knocked_rooms = services
.rooms .rooms
.state_cache .state_cache
.rooms_knocked(sender_user) .rooms_knocked(syncing_user)
.fold_default(|mut knocked_rooms: BTreeMap<_, _>, (room_id, knock_state)| async move { .fold_default(|mut knocked_rooms: BTreeMap<_, _>, (room_id, knock_state)| async move {
let knock_count = services let knock_count = services
.rooms .rooms
.state_cache .state_cache
.get_knock_count(&room_id, sender_user) .get_knock_count(&room_id, syncing_user)
.await .await
.ok(); .ok();
// only sync this knock if it was sent after the last /sync call // only sync this knock if it was sent after the last /sync call
if since < knock_count { if last_sync_end_count < knock_count {
let knocked_room = KnockedRoom { let knocked_room = KnockedRoom {
knock_state: KnockState { events: knock_state }, knock_state: KnockState { events: knock_state },
}; };
@@ -324,36 +346,41 @@ pub(crate) async fn build_sync_events(
let presence_updates: OptionFuture<_> = services let presence_updates: OptionFuture<_> = services
.config .config
.allow_local_presence .allow_local_presence
.then(|| process_presence_updates(services, since, sender_user)) .then(|| process_presence_updates(services, last_sync_end_count, syncing_user))
.into(); .into();
let account_data = services let account_data = services
.account_data .account_data
.changes_since(None, sender_user, since, Some(next_batch)) .changes_since(None, syncing_user, last_sync_end_count, Some(current_count))
.ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global)) .ready_filter_map(|e| extract_variant!(e, AnyRawAccountDataEvent::Global))
.collect(); .collect();
// Look for device list updates of this account // Look for device list updates of this account
let keys_changed = services let keys_changed = services
.users .users
.keys_changed(sender_user, since, Some(next_batch)) .keys_changed(syncing_user, last_sync_end_count, Some(current_count))
.map(ToOwned::to_owned) .map(ToOwned::to_owned)
.collect::<HashSet<_>>(); .collect::<HashSet<_>>();
let to_device_events = services let to_device_events = services
.users .users
.get_to_device_events(sender_user, sender_device, since, Some(next_batch)) .get_to_device_events(
syncing_user,
syncing_device,
last_sync_end_count,
Some(current_count),
)
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let device_one_time_keys_count = services let device_one_time_keys_count = services
.users .users
.count_one_time_keys(sender_user, sender_device); .count_one_time_keys(syncing_user, syncing_device);
// Remove all to-device events the device received *last time* // Remove all to-device events the device received *last time*
let remove_to_device_events = let remove_to_device_events =
services services
.users .users
.remove_to_device_events(sender_user, sender_device, since); .remove_to_device_events(syncing_user, syncing_device, last_sync_end_count);
let rooms = join4(joined_rooms, left_rooms, invited_rooms, knocked_rooms); let rooms = join4(joined_rooms, left_rooms, invited_rooms, knocked_rooms);
let ephemeral = join3(remove_to_device_events, to_device_events, presence_updates); let ephemeral = join3(remove_to_device_events, to_device_events, presence_updates);
@@ -373,7 +400,7 @@ pub(crate) async fn build_sync_events(
device_one_time_keys_count, device_one_time_keys_count,
// Fallback keys are not yet supported // Fallback keys are not yet supported
device_unused_fallback_key_types: None, device_unused_fallback_key_types: None,
next_batch: next_batch.to_string(), next_batch: current_count.to_string(),
presence: Presence { presence: Presence {
events: presence_updates events: presence_updates
.into_iter() .into_iter()
@@ -398,12 +425,12 @@ pub(crate) async fn build_sync_events(
#[tracing::instrument(name = "presence", level = "debug", skip_all)] #[tracing::instrument(name = "presence", level = "debug", skip_all)]
async fn process_presence_updates( async fn process_presence_updates(
services: &Services, services: &Services,
since: Option<u64>, last_sync_end_count: Option<u64>,
syncing_user: &UserId, syncing_user: &UserId,
) -> PresenceUpdates { ) -> PresenceUpdates {
services services
.presence .presence
.presence_since(since.unwrap_or(0)) // send all presences on initial sync .presence_since(last_sync_end_count.unwrap_or(0)) // send all presences on initial sync
.filter(|(user_id, ..)| { .filter(|(user_id, ..)| {
services services
.rooms .rooms
@@ -433,7 +460,7 @@ async fn prepare_lazily_loaded_members(
// reset lazy loading state on initial sync. // reset lazy loading state on initial sync.
// do this even if lazy loading is disabled so future lazy loads // do this even if lazy loading is disabled so future lazy loads
// will have the correct members. // will have the correct members.
if sync_context.since.is_none() { if sync_context.last_sync_end_count.is_none() {
services services
.rooms .rooms
.lazy_loading .lazy_loading
+4 -4
View File
@@ -93,8 +93,8 @@ pub(super) async fn calculate_state_incremental<'a>(
services: &Services, services: &Services,
sender_user: &'a UserId, sender_user: &'a UserId,
room_id: &RoomId, room_id: &RoomId,
previous_sync_end_count: PduCount, last_sync_end_count: PduCount,
previous_sync_end_shortstatehash: ShortStateHash, last_sync_end_shortstatehash: ShortStateHash,
timeline_start_shortstatehash: ShortStateHash, timeline_start_shortstatehash: ShortStateHash,
timeline_end_shortstatehash: ShortStateHash, timeline_end_shortstatehash: ShortStateHash,
timeline: &TimelinePdus, timeline: &TimelinePdus,
@@ -117,7 +117,7 @@ pub(super) async fn calculate_state_incremental<'a>(
let last_pdu_of_last_sync = services let last_pdu_of_last_sync = services
.rooms .rooms
.timeline .timeline
.pdus_rev(Some(sender_user), room_id, Some(previous_sync_end_count.saturating_add(1))) .pdus_rev(Some(sender_user), room_id, Some(last_sync_end_count.saturating_add(1)))
.boxed() .boxed()
.next() .next()
.await .await
@@ -237,7 +237,7 @@ pub(super) async fn calculate_state_incremental<'a>(
services services
.rooms .rooms
.state_accessor .state_accessor
.state_added((previous_sync_end_shortstatehash, timeline_end_shortstatehash)) .state_added((last_sync_end_shortstatehash, timeline_end_shortstatehash))
.await? .await?
.stream() .stream()
.ready_filter_map(|(_, shorteventid)| { .ready_filter_map(|(_, shorteventid)| {