diff --git a/Cargo.toml b/Cargo.toml index 7fe0906e8..547309229 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -378,6 +378,7 @@ features = [ "unstable-msc4406", "unstable-msc4439", "unstable-msc4354", + "unstable-msc4480", "unstable-msc4466", "unstable-extensible-events", ] diff --git a/changelog.d/+3ec50619.feature.md b/changelog.d/+3ec50619.feature.md new file mode 100644 index 000000000..d1aba9801 --- /dev/null +++ b/changelog.d/+3ec50619.feature.md @@ -0,0 +1 @@ +Added MSC4480 sticky events to simplified sliding sync. Contributed by @eleboucher. diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index 3bf6ac693..c2dd705da 100644 --- a/src/api/client/sync/v5.rs +++ b/src/api/client/sync/v5.rs @@ -7,7 +7,10 @@ use std::{ use axum::extract::State; use conduwuit::{ Err, Error, Result, at, error, extract_variant, is_equal_to, - matrix::{Event, TypeStateKey, pdu::PduCount}, + matrix::{ + Event, TypeStateKey, + pdu::{PduCount, PduEvent, sticky}, + }, trace, utils::{ BoolExt, FutureBoolExt, IterStream, ReadyExt, TryFutureExtExt, @@ -28,7 +31,7 @@ use futures::{ pin_mut, }; use ruma::{ - DeviceId, OwnedEventId, OwnedRoomId, RoomId, UInt, UserId, + DeviceId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, RoomId, UInt, UserId, api::client::sync::sync_events::{ self, DeviceLists, UnreadNotificationsCount, v5::request::ExtensionRoomConfig, }, @@ -64,6 +67,9 @@ type TodoRooms = BTreeMap, usize, u64)>; type KnownRooms = BTreeMap>; type KnownRoomUpdates = BTreeMap>; +/// Default and maximum number of sticky events returned per response. +const DEFAULT_STICKY_LIMIT: usize = 100; + struct SyncCollection { response: sync_events::v5::Response, known_room_updates: KnownRoomUpdates, @@ -221,6 +227,10 @@ async fn collect_sync_response( .collect(); let all_joined_rooms = all_joined_rooms.iter().map(AsRef::as_ref); + let joined_room_ids = all_joined_rooms + .clone() + .map(ToOwned::to_owned) + .collect::>(); let all_invited_rooms = all_invited_rooms.iter().map(AsRef::as_ref); let all_knocked_rooms = all_knocked_rooms.iter().map(AsRef::as_ref); let all_rooms = all_joined_rooms @@ -231,6 +241,7 @@ async fn collect_sync_response( let pos = next_batch.to_string(); let mut todo_rooms: TodoRooms = BTreeMap::new(); + let mut sticky_rooms = BTreeSet::new(); let sync_info: SyncInfo<'_> = (sender_user, sender_device, globalsince, body); @@ -251,6 +262,7 @@ async fn collect_sync_response( to_device, receipts, typing: sync_events::v5::response::Typing::default(), + sticky_events: sync_events::v5::response::StickyEvents::default(), }); let mut response = assign!(sync_events::v5::Response::new(pos), { @@ -286,6 +298,7 @@ async fn collect_sync_response( all_rooms, &direct_rooms, &mut todo_rooms, + &mut sticky_rooms, known_rooms, &mut response, ) @@ -295,7 +308,9 @@ async fn collect_sync_response( "subscriptions".to_owned(), fetch_subscriptions(services, body, known_rooms, &allowed_rooms, &mut todo_rooms).await, ); + sticky_rooms.extend(body.room_subscriptions.keys().cloned()); + let mut timeline_event_ids = BTreeSet::new(); response.rooms = process_rooms( services, sender_user, @@ -305,6 +320,18 @@ async fn collect_sync_response( &todo_rooms, &mut response, body, + &mut timeline_event_ids, + ) + .await?; + + response.extensions.sticky_events = collect_sticky_events( + services, + sender_user, + body, + &sticky_rooms, + &joined_room_ids, + &todo_rooms, + &timeline_event_ids, ) .await?; @@ -334,7 +361,10 @@ fn response_is_empty(response: &sync_events::v5::Response) -> bool { .as_ref() .is_none_or(|to| to.events.is_empty()); - no_account_data && no_room_data && no_to_device_messages + no_account_data + && no_room_data + && no_to_device_messages + && response.extensions.sticky_events.is_empty() } async fn commit_sync_collection( @@ -435,6 +465,7 @@ async fn handle_lists<'a, Rooms, AllRooms>( all_rooms: AllRooms, direct_rooms: &HashSet, todo_rooms: &'a mut TodoRooms, + sticky_rooms: &mut BTreeSet, known_rooms: &'a KnownRooms, response: &'_ mut sync_events::v5::Response, ) -> KnownRoomUpdates @@ -511,6 +542,7 @@ where active_rooms.push(room_id); } + sticky_rooms.extend(active_rooms.iter().map(|room_id| (*room_id).to_owned())); let mut new_known_rooms: BTreeSet = BTreeSet::new(); @@ -669,6 +701,7 @@ async fn process_rooms<'a, Rooms>( todo_rooms: &TodoRooms, response: &mut sync_events::v5::Response, body: &sync_events::v5::Request, + timeline_event_ids: &mut BTreeSet, ) -> Result> where Rooms: Iterator + Clone + Send + 'a, @@ -872,6 +905,7 @@ where .map(Event::into_format) .collect() .await; + timeline_event_ids.extend(timeline_pdus.iter().map(|(_, pdu)| pdu.event_id.clone())); for (_, pdu) in timeline_pdus { let ts = pdu.origin_server_ts; @@ -1009,6 +1043,124 @@ where Ok(rooms) } +async fn collect_sticky_events( + services: &Services, + sender_user: &UserId, + body: &sync_events::v5::Request, + sticky_rooms: &BTreeSet, + joined_room_ids: &BTreeSet, + todo_rooms: &TodoRooms, + timeline_event_ids: &BTreeSet, +) -> Result { + if !services.config.allow_sticky_events + || !body.extensions.sticky_events.enabled.unwrap_or(false) + { + return Ok(sync_events::v5::response::StickyEvents::default()); + } + + let since = body + .extensions + .sticky_events + .since + .as_deref() + .and_then(|token| token.parse::().ok()) + .unwrap_or(0); + let limit = body + .extensions + .sticky_events + .limit + .map_or(DEFAULT_STICKY_LIMIT, usize_from_ruma) + .clamp(1, DEFAULT_STICKY_LIMIT); + let now = u64::from(MilliSecondsSinceUnixEpoch::now().get()); + let oldest_sticky_ts = now.saturating_sub(sticky::MAX_DURATION_MS); + + let mut backlog = Vec::new(); + let mut stream = Vec::new(); + for room_id in sticky_rooms + .iter() + .filter(|room_id| joined_room_ids.contains(*room_id)) + { + // a room the client has not seen before needs its whole backlog, which a + // single stream position cannot express + let initial = todo_rooms + .get(room_id) + .is_some_and(|&(_, _, roomsince)| roomsince == 0); + + let pdus = services.rooms.timeline.pdus_rev(room_id, None).ignore_err(); + pin_mut!(pdus); + while let Some((count, pdu)) = pdus.next().await { + // nothing older than this can still be sticky, and on the stream we + // have already delivered everything up to `since` + if u64::from(pdu.origin_server_ts) < oldest_sticky_ts + || (!initial && count <= PduCount::Normal(since)) + { + break; + } + + let PduCount::Normal(count) = count else { + continue; + }; + + let is_sticky = pdu + .sticky + .as_deref() + .is_some_and(|sticky| sticky::is_sticky(pdu.origin_server_ts, sticky, now)); + + if !is_sticky || timeline_event_ids.contains(&pdu.event_id) { + continue; + } + if services + .users + .user_is_ignored(pdu.sender(), sender_user) + .await + { + continue; + } + + if count > since { + stream.push((count, room_id.clone(), pdu)); + } else { + backlog.push((room_id.clone(), pdu)); + } + } + } + stream.sort_by_key(|(count, ..)| *count); + + let mut response = sync_events::v5::response::StickyEvents::default(); + + // MSC4480 lets us ignore the limit for a newly visible room's backlog + for (room_id, pdu) in backlog { + push_sticky_event(&mut response, sender_user, room_id, pdu); + } + + let mut next_batch = None; + for (count, room_id, pdu) in stream.into_iter().take(limit) { + push_sticky_event(&mut response, sender_user, room_id, pdu); + next_batch = Some(count); + } + + if !response.rooms.is_empty() { + response.next_batch = Some(next_batch.unwrap_or(since).to_string()); + } + + Ok(response) +} + +fn push_sticky_event( + response: &mut sync_events::v5::response::StickyEvents, + sender_user: &UserId, + room_id: OwnedRoomId, + mut pdu: PduEvent, +) { + pdu.set_unsigned(Some(sender_user)); + response + .rooms + .entry(room_id) + .or_default() + .events + .push(Event::into_format(pdu)); +} + /// Collect the required state events for a room async fn collect_required_state( services: &Services,