diff --git a/Cargo.lock b/Cargo.lock index 1e88e7434..733a76301 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4969,7 +4969,7 @@ dependencies = [ [[package]] name = "ruma" version = "0.17.2" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "assign", "js_int", @@ -4988,7 +4988,7 @@ dependencies = [ [[package]] name = "ruma-appservice-api" version = "0.17.1" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "http", "js_int", @@ -5001,7 +5001,7 @@ dependencies = [ [[package]] name = "ruma-client-api" version = "0.25.2" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "as_variant", "assign", @@ -5023,7 +5023,7 @@ dependencies = [ [[package]] name = "ruma-common" version = "0.20.2" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "as_variant", "base64 0.23.1", @@ -5056,7 +5056,7 @@ dependencies = [ [[package]] name = "ruma-events" version = "0.35.2" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "as_variant", "indexmap 2.14.2", @@ -5077,7 +5077,7 @@ dependencies = [ [[package]] name = "ruma-federation-api" version = "0.16.1" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "bytes", "http", @@ -5099,7 +5099,7 @@ dependencies = [ [[package]] name = "ruma-identifiers-validation" version = "0.14.0" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "js_int", "thiserror 2.0.21", @@ -5108,7 +5108,7 @@ dependencies = [ [[package]] name = "ruma-macros" version = "0.20.2" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "as_variant", "cfg-if", @@ -5124,7 +5124,7 @@ dependencies = [ [[package]] name = "ruma-push-gateway-api" version = "0.16.0" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "js_int", "ruma-common", @@ -5136,7 +5136,7 @@ dependencies = [ [[package]] name = "ruma-signatures" version = "0.22.1" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "base64 0.23.1", "ed25519-dalek", @@ -5153,7 +5153,7 @@ dependencies = [ [[package]] name = "ruma-state-res" version = "0.18.0" -source = "git+https://github.com/ruma/ruma.git?rev=0ed7fba34213f18a628a56f527c6cc795cd7eef5#0ed7fba34213f18a628a56f527c6cc795cd7eef5" +source = "git+https://github.com/ruma/ruma.git?rev=472a23ef86bc0bb40a22d856fc0af539eb9d71b7#472a23ef86bc0bb40a22d856fc0af539eb9d71b7" dependencies = [ "js_int", "ruma-common", @@ -5625,7 +5625,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d43fb82d550b07e6f929a58d7d0935182602f48ddead875f5ee313c41c8f07fe" dependencies = [ "annotate-snippets", - "base64 0.23.1", + "base64 0.22.1", "encoding_rs_io", "granit-parser", "nohash-hasher", diff --git a/Cargo.toml b/Cargo.toml index 05831568a..a683d2058 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -357,7 +357,7 @@ version = "1.1.1" # Used for matrix spec type definitions and helpers [workspace.dependencies.ruma] git = "https://github.com/ruma/ruma.git" -rev = "0ed7fba34213f18a628a56f527c6cc795cd7eef5" +rev = "472a23ef86bc0bb40a22d856fc0af539eb9d71b7" features = [ "appservice-api-c", "client-api", diff --git a/src/api/client/dehydrated_device.rs b/src/api/client/dehydrated_device.rs index 72ae2b955..124b90f19 100644 --- a/src/api/client/dehydrated_device.rs +++ b/src/api/client/dehydrated_device.rs @@ -3,10 +3,10 @@ use conduwuit::{Err, Result, at}; use futures::StreamExt; use ruma::{ api::client::dehydrated_device::{ - delete_dehydrated_device::unstable as delete_dehydrated_device, - get_dehydrated_device::unstable as get_dehydrated_device, - get_events::unstable as get_events, - put_dehydrated_device::unstable as put_dehydrated_device, + delete_dehydrated_device::msc3814 as delete_dehydrated_device, + get_dehydrated_device::msc3814 as get_dehydrated_device, + get_events::msc3814 as get_events, + put_dehydrated_device::msc3814 as put_dehydrated_device, }, assign, }; diff --git a/src/api/client/media.rs b/src/api/client/media.rs index 824d98541..669cda35d 100644 --- a/src/api/client/media.rs +++ b/src/api/client/media.rs @@ -30,9 +30,9 @@ pub(crate) async fn get_media_config_route( State(services): State, _body: Ruma, ) -> Result { - Ok(get_media_config::v1::Response::new(ruma_from_usize( + Ok(get_media_config::v1::Response::new(Some(ruma_from_usize( services.server.config.max_request_size, - ))) + )))) } /// # `POST /_matrix/media/v3/upload` diff --git a/src/api/client/media_legacy.rs b/src/api/client/media_legacy.rs index ddc7e9c92..85eaabec9 100644 --- a/src/api/client/media_legacy.rs +++ b/src/api/client/media_legacy.rs @@ -25,9 +25,9 @@ pub(crate) async fn get_media_config_legacy_route( State(services): State, _body: Ruma, ) -> Result { - Ok(get_media_config::v3::Response::new(ruma_from_usize( + Ok(get_media_config::v3::Response::new(Some(ruma_from_usize( services.server.config.max_request_size, - ))) + )))) } /// # `GET /_matrix/media/v1/config` diff --git a/src/api/client/mutual_rooms.rs b/src/api/client/mutual_rooms.rs index 532a5cc79..265238545 100644 --- a/src/api/client/mutual_rooms.rs +++ b/src/api/client/mutual_rooms.rs @@ -13,8 +13,8 @@ use crate::Ruma; #[tracing::instrument(skip_all, name = "mutual_rooms", level = "info")] pub(crate) async fn get_mutual_rooms_unstable_route( State(services): State, - body: Ruma, -) -> Result { + body: Ruma, +) -> Result { let sender_user = body.identity.expect_sender_user()?; if sender_user == body.user_id { @@ -28,7 +28,7 @@ pub(crate) async fn get_mutual_rooms_unstable_route( .collect() .await; - Ok(get_mutual_rooms::unstable::Response::new(mutual_rooms)) + Ok(get_mutual_rooms::msc2666::Response::new(mutual_rooms)) } /// # `GET /_matrix/client/v1/mutual_rooms` diff --git a/src/api/client/room/stripped_state.rs b/src/api/client/room/stripped_state.rs index d30655721..cf686ef2f 100644 --- a/src/api/client/room/stripped_state.rs +++ b/src/api/client/room/stripped_state.rs @@ -15,11 +15,11 @@ use crate::{Ruma, router::ClientIdentity}; #[tracing::instrument(skip_all, name = "stripped_state", level = "info")] pub(crate) async fn get_stripped_state_route( State(services): State, - body: Ruma, -) -> Result { + body: Ruma, +) -> Result { let user_id = body.identity.as_ref().and_then(ClientIdentity::sender_user); - if body.via.len() > get_stripped_state::unstable::MAX_VIAS { + if body.via.len() > get_stripped_state::msc4560::MAX_VIAS { return Err!(Request(TooLarge("Too many vias, please be less indecisive."))); } @@ -59,7 +59,7 @@ pub(crate) async fn get_stripped_state_route( via, RoomStrippedState { state, num_joined_members, .. }, )) => Ok( - assign!(get_stripped_state::unstable::Response::new(via), { state, num_joined_members }), + assign!(get_stripped_state::msc4560::Response::new(via), { state, num_joined_members }), ), | Accessibility::Inaccessible => { Err!(Request(Forbidden("You may not request this room's stripped state."))) diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index 4f21c1930..4ce7d9c1c 100644 --- a/src/api/client/sync/v5.rs +++ b/src/api/client/sync/v5.rs @@ -24,7 +24,7 @@ use conduwuit::{ use conduwuit_service::{ Services, rooms::read_receipt::pack_receipts, - sync::{SnakeConnectionsKey, into_snake_key}, + sync::{SnakeConnectionsKey, StickyDelivered, into_snake_key}, }; use futures::{ FutureExt, StreamExt, TryFutureExt, @@ -34,7 +34,7 @@ use futures::{ use ruma::{ DeviceId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedRoomId, RoomId, UInt, UserId, api::client::sync::sync_events::{ - self, DeviceLists, UnreadNotificationsCount, v5::request::ExtensionRoomConfig, + self, DeviceLists, UnreadNotificationsCount, msc4186::request::ExtensionRoomConfig, }, assign, directory::RoomTypeFilter, @@ -63,7 +63,7 @@ use crate::{ client_ip::ClientIp, }; -type SyncInfo<'a> = (&'a UserId, &'a DeviceId, u64, &'a sync_events::v5::Request); +type SyncInfo<'a> = (&'a UserId, &'a DeviceId, u64, &'a sync_events::msc4186::Request); type TodoRooms = BTreeMap, usize, u64)>; type KnownRooms = BTreeMap>; type KnownRoomUpdates = BTreeMap>; @@ -86,8 +86,9 @@ fn num_live_events<'a>( } struct SyncCollection { - response: sync_events::v5::Response, + response: sync_events::msc4186::Response, known_room_updates: KnownRoomUpdates, + sticky_delivered: StickyDelivered, } /// `POST /_matrix/client/unstable/org.matrix.simplified_msc3575/sync` @@ -103,8 +104,8 @@ struct SyncCollection { pub(crate) async fn sync_events_v5_route( State(ref services): State, ClientIp(client_ip): ClientIp, // NOTE: Required for updating device metadata - body: Ruma, -) -> Result { + body: Ruma, +) -> Result { let sender_user = body.identity.expect_sender_user()?; let sender_device = body.identity.expect_sender_device()?; @@ -142,7 +143,7 @@ pub(crate) async fn sync_events_v5_route( } // Get sticky parameters from cache - let known_rooms = + let (known_rooms, sticky_delivered) = services .sync .update_snake_sync_request_with_cache(&snake_key, globalsince, &mut body); @@ -155,6 +156,7 @@ pub(crate) async fn sync_events_v5_route( &body, globalsince, &known_rooms, + &sticky_delivered, ) .await?; @@ -173,6 +175,7 @@ pub(crate) async fn sync_events_v5_route( &body, globalsince, &known_rooms, + &sticky_delivered, ) .await?; } @@ -202,9 +205,10 @@ async fn collect_sync_response( services: &Services, sender_user: &UserId, sender_device: &DeviceId, - body: &sync_events::v5::Request, + body: &sync_events::msc4186::Request, globalsince: u64, known_rooms: &KnownRooms, + sticky_delivered: &StickyDelivered, ) -> Result { // Bounds every read below and becomes the `pos` returned to the client. // Zero would be echoed back as a request for an initial sync. @@ -262,7 +266,6 @@ 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); @@ -277,16 +280,16 @@ async fn collect_sync_response( let (account_data, e2ee, to_device, receipts) = try_join4(account_data, e2ee, to_device, receipts).await?; - let extensions = assign!(sync_events::v5::response::Extensions::default(), { + let extensions = assign!(sync_events::msc4186::response::Extensions::default(), { account_data, e2ee, to_device, receipts, - typing: sync_events::v5::response::Typing::default(), - sticky_events: sync_events::v5::response::StickyEvents::default(), + typing: sync_events::msc4186::response::Typing::default(), + sticky_events: sync_events::msc4186::response::StickyEvents::default(), }); - let mut response = assign!(sync_events::v5::Response::new(pos), { + let mut response = assign!(sync_events::msc4186::Response::new(pos), { txn_id: body.txn_id.clone(), lists: BTreeMap::new(), rooms: BTreeMap::new(), @@ -319,7 +322,6 @@ async fn collect_sync_response( all_rooms, &direct_rooms, &mut todo_rooms, - &mut sticky_rooms, known_rooms, &mut response, ) @@ -329,7 +331,6 @@ 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( @@ -346,25 +347,31 @@ async fn collect_sync_response( ) .await?; - response.extensions.sticky_events = collect_sticky_events( + let (sticky_events, sticky_delivered) = collect_sticky_events( services, sender_user, body, - &sticky_rooms, + next_batch, + sticky_delivered, + &known_room_updates, &joined_room_ids, - &todo_rooms, &timeline_event_ids, ) - .await?; + .await; + response.extensions.sticky_events = sticky_events; response.extensions.typing = collect_typing_events(services, sender_user, body, &todo_rooms, &known_room_updates) .await?; - Ok(SyncCollection { response, known_room_updates }) + Ok(SyncCollection { + response, + known_room_updates, + sticky_delivered, + }) } -fn response_is_empty(response: &sync_events::v5::Response) -> bool { +fn response_is_empty(response: &sync_events::msc4186::Response) -> bool { let no_account_data = response.extensions.account_data.global.is_empty() && response .extensions @@ -395,7 +402,7 @@ async fn commit_sync_collection( services: &Services, sender_user: &UserId, sender_device: &DeviceId, - body: &sync_events::v5::Request, + body: &sync_events::msc4186::Request, globalsince: u64, snake_key: &SnakeConnectionsKey, collection: &SyncCollection, @@ -408,9 +415,10 @@ async fn commit_sync_collection( .parse() .expect("generated sync position"), collection.known_room_updates.clone(), + collection.sticky_delivered.clone(), ); - if body.extensions.to_device.enabled.unwrap_or(false) { + if body.extensions.to_device.enabled { services .users .remove_to_device_events(sender_user, sender_device, globalsince) @@ -420,7 +428,7 @@ async fn commit_sync_collection( async fn fetch_subscriptions( services: &Services, - body: &sync_events::v5::Request, + body: &sync_events::msc4186::Request, known_rooms: &KnownRooms, allowed_rooms: &BTreeSet, todo_rooms: &mut TodoRooms, @@ -480,15 +488,14 @@ async fn fetch_subscriptions( async fn handle_lists<'a, Rooms, AllRooms>( services: &Services, sender_user: &UserId, - body: &sync_events::v5::Request, + body: &sync_events::msc4186::Request, all_invited_rooms: Rooms, all_joined_rooms: Rooms, 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, + response: &'_ mut sync_events::msc4186::Response, ) -> KnownRoomUpdates where Rooms: Iterator + Clone + Send + 'a, @@ -579,7 +586,6 @@ where .await; sort_rooms_by_activity(&mut active_rooms); let active_rooms: Vec<_> = active_rooms.into_iter().map(at!(0)).collect(); - sticky_rooms.extend(active_rooms.iter().map(|room_id| (*room_id).to_owned())); let mut new_known_rooms: BTreeSet = BTreeSet::new(); @@ -631,7 +637,7 @@ where } response.lists.insert( list_id.clone(), - assign!(sync_events::v5::response::List::default(), { + assign!(sync_events::msc4186::response::List::default(), { count: ruma_from_usize(active_rooms.len()), }), ); @@ -749,10 +755,10 @@ async fn process_rooms<'a, Rooms>( all_invited_rooms: Rooms, all_knocked_rooms: Rooms, todo_rooms: &TodoRooms, - response: &mut sync_events::v5::Response, - body: &sync_events::v5::Request, + response: &mut sync_events::msc4186::Response, + body: &sync_events::msc4186::Request, timeline_event_ids: &mut BTreeSet, -) -> Result> +) -> Result> where Rooms: Iterator + Clone + Send + 'a, { @@ -787,7 +793,7 @@ where rooms.insert( room_id.clone(), - assign!(sync_events::v5::response::Room::new(), { + assign!(sync_events::msc4186::response::Room::new(), { initial: Some(roomsince == &0), invite_state, limited: true, @@ -821,7 +827,7 @@ where rooms.insert( room_id.clone(), - assign!(sync_events::v5::response::Room::new(), { + assign!(sync_events::msc4186::response::Room::new(), { initial: Some(roomsince == &0), invite_state: Some(knock_state), limited: true, @@ -988,7 +994,7 @@ where .state_accessor .get_member(room_id, &user_id) .map_ok(|member_event| { - assign!(sync_events::v5::response::Hero::new(user_id.clone()), { + assign!(sync_events::msc4186::response::Hero::new(user_id.clone()), { name: member_event.displayname, avatar: member_event.avatar_url, }) @@ -1032,7 +1038,7 @@ where rooms.insert( room_id.clone(), - assign!(sync_events::v5::response::Room::new(), { + assign!(sync_events::msc4186::response::Room::new(), { name: services .rooms .state_accessor @@ -1103,64 +1109,61 @@ where Ok(rooms) } +#[expect(clippy::too_many_arguments)] async fn collect_sticky_events( services: &Services, sender_user: &UserId, - body: &sync_events::v5::Request, - sticky_rooms: &BTreeSet, + body: &sync_events::msc4186::Request, + next_batch: u64, + sticky_delivered: &StickyDelivered, + known_room_updates: &KnownRoomUpdates, 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()); +) -> (sync_events::msc4186::response::StickyEvents, StickyDelivered) { + let mut response = sync_events::msc4186::response::StickyEvents::default(); + let mut delivered = StickyDelivered::new(); + + let sticky_events = &body.extensions.sticky_events; + if !services.config.allow_sticky_events || !sticky_events.enabled.unwrap_or(false) { + return (response, delivered); } - let since = body - .extensions - .sticky_events - .since - .as_deref() - .and_then(|token| token.parse::().ok()) - .unwrap_or(0); - let limit = body - .extensions - .sticky_events + let scope = extension_scope( + sticky_events.lists.as_deref(), + sticky_events.rooms.as_deref(), + body.lists.keys().map(String::as_str), + known_room_updates, + ); + let limit = 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); + for room_id in scope.intersection(joined_room_ids) { + let since = sticky_delivered.get(room_id).copied(); + let mut pending = Vec::new(); 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; }; + if count > next_batch { + continue; + } + + // Nothing older than this can still be sticky, and everything up to + // `since` has already been delivered + if u64::from(pdu.origin_server_ts) < oldest_sticky_ts + || since.is_some_and(|since| count <= since) + { + break; + } + let is_sticky = pdu .sticky .as_deref() @@ -1177,38 +1180,26 @@ async fn collect_sticky_events( continue; } - if count > since { - stream.push((count, room_id.clone(), pdu)); - } else { - backlog.push((room_id.clone(), pdu)); - } + pending.push((count, pdu)); } - } - stream.sort_by_key(|(count, ..)| *count); - let mut response = sync_events::v5::response::StickyEvents::default(); + pending.reverse(); + let truncated = pending.len() > limit; + let mut last_count = next_batch; + for (count, pdu) in pending.into_iter().take(limit) { + push_sticky_event(services, &mut response, sender_user, room_id.clone(), pdu).await; + last_count = count; + } - // MSC4480 lets us ignore the limit for a newly visible room's backlog - for (room_id, pdu) in backlog { - push_sticky_event(services, &mut response, sender_user, room_id, pdu).await; + delivered.insert(room_id.clone(), if truncated { last_count } else { next_batch }); } - let mut next_batch = None; - for (count, room_id, pdu) in stream.into_iter().take(limit) { - push_sticky_event(services, &mut response, sender_user, room_id, pdu).await; - next_batch = Some(count); - } - - if !response.rooms.is_empty() { - response.next_batch = Some(next_batch.unwrap_or(since).to_string()); - } - - Ok(response) + (response, delivered) } async fn push_sticky_event( services: &Services, - response: &mut sync_events::v5::response::StickyEvents, + response: &mut sync_events::msc4186::response::StickyEvents, sender_user: &UserId, room_id: OwnedRoomId, mut pdu: PduEvent, @@ -1275,12 +1266,12 @@ async fn collect_required_state( async fn collect_typing_events( services: &Services, sender_user: &UserId, - body: &sync_events::v5::Request, + body: &sync_events::msc4186::Request, todo_rooms: &TodoRooms, known_room_updates: &KnownRoomUpdates, -) -> Result { +) -> Result { if !body.extensions.typing.enabled.unwrap_or(false) { - return Ok(sync_events::v5::response::Typing::default()); + return Ok(sync_events::msc4186::response::Typing::default()); } let typing = &body.extensions.typing; let scope = extension_scope( @@ -1290,10 +1281,10 @@ async fn collect_typing_events( known_room_updates, ); if scope.is_empty() { - return Ok(sync_events::v5::response::Typing::default()); + return Ok(sync_events::msc4186::response::Typing::default()); } - let mut typing_response = sync_events::v5::response::Typing::default(); + let mut typing_response = sync_events::msc4186::response::Typing::default(); for (room_id, (_, _, roomsince)) in todo_rooms { if !scope.contains(room_id) { continue; @@ -1373,12 +1364,17 @@ fn extension_scope<'a>( async fn collect_account_data( services: &Services, - (sender_user, _, globalsince, body): (&UserId, &DeviceId, u64, &sync_events::v5::Request), -) -> sync_events::v5::response::AccountData { - let mut account_data = sync_events::v5::response::AccountData::default(); + (sender_user, _, globalsince, body): ( + &UserId, + &DeviceId, + u64, + &sync_events::msc4186::Request, + ), +) -> sync_events::msc4186::response::AccountData { + let mut account_data = sync_events::msc4186::response::AccountData::default(); if !body.extensions.account_data.enabled.unwrap_or(false) { - return sync_events::v5::response::AccountData::default(); + return sync_events::msc4186::response::AccountData::default(); } account_data.global = services @@ -1413,15 +1409,15 @@ async fn collect_e2ee<'a, Rooms>( &UserId, &DeviceId, u64, - &sync_events::v5::Request, + &sync_events::msc4186::Request, ), all_joined_rooms: Rooms, -) -> Result +) -> Result where Rooms: Iterator + Send + 'a, { - if !body.extensions.e2ee.enabled.unwrap_or(false) { - return Ok(sync_events::v5::response::E2EE::default()); + if !body.extensions.e2ee.enabled { + return Ok(sync_events::msc4186::response::E2EE::default()); } let mut left_encrypted_users = HashSet::new(); // Users that have left any encrypted rooms the sender was in let mut device_list_changes = HashSet::new(); @@ -1608,7 +1604,7 @@ where } } - Ok(assign!(sync_events::v5::response::E2EE::default(), { + Ok(assign!(sync_events::msc4186::response::E2EE::default(), { device_unused_fallback_key_types: None, device_one_time_keys_count: services @@ -1627,12 +1623,12 @@ async fn collect_to_device( services: &Services, (sender_user, sender_device, globalsince, body): SyncInfo<'_>, next_batch: u64, -) -> Option { - if !body.extensions.to_device.enabled.unwrap_or(false) { +) -> Option { + if !body.extensions.to_device.enabled { return None; } - Some(assign!(sync_events::v5::response::ToDevice::default(), { + Some(assign!(sync_events::msc4186::response::ToDevice::default(), { next_batch: next_batch.to_string(), events: services .users @@ -1643,9 +1639,9 @@ async fn collect_to_device( })) } -async fn collect_receipts(_services: &Services) -> sync_events::v5::response::Receipts { +async fn collect_receipts(_services: &Services) -> sync_events::msc4186::response::Receipts { // TODO: get explicitly requested read receipts - sync_events::v5::response::Receipts::default() + sync_events::msc4186::response::Receipts::default() } #[cfg(test)] @@ -1658,23 +1654,25 @@ mod tests { fn custom_room_type() -> RoomTypeFilter { RoomTypeFilter::from(Some("com.example")) } - fn response() -> sync_events::v5::Response { sync_events::v5::Response::new("1".to_owned()) } + fn response() -> sync_events::msc4186::Response { + sync_events::msc4186::Response::new("1".to_owned()) + } - fn typing_scope_request() -> sync_events::v5::Request { - let mut request = sync_events::v5::Request::new(); + fn typing_scope_request() -> sync_events::msc4186::Request { + let mut request = sync_events::msc4186::Request::new(); request .lists - .insert("first".to_owned(), sync_events::v5::request::List::default()); + .insert("first".to_owned(), sync_events::msc4186::request::List::default()); request .lists - .insert("second".to_owned(), sync_events::v5::request::List::default()); + .insert("second".to_owned(), sync_events::msc4186::request::List::default()); request.room_subscriptions.insert( owned_room_id!("!subscription-a:example.com"), - sync_events::v5::request::RoomSubscription::default(), + sync_events::msc4186::request::RoomSubscription::default(), ); request.room_subscriptions.insert( owned_room_id!("!subscription-b:example.com"), - sync_events::v5::request::RoomSubscription::default(), + sync_events::msc4186::request::RoomSubscription::default(), ); request } @@ -1768,9 +1766,10 @@ mod tests { #[test] fn a_room_with_no_events_is_still_empty() { let mut response = response(); - response - .rooms - .insert(owned_room_id!("!a:example.com"), sync_events::v5::response::Room::default()); + response.rooms.insert( + owned_room_id!("!a:example.com"), + sync_events::msc4186::response::Room::default(), + ); assert!(response_is_empty(&response)); } @@ -1781,7 +1780,7 @@ mod tests { let mut response = response(); response .rooms - .insert(room_id.to_owned(), sync_events::v5::response::Room::default()); + .insert(room_id.to_owned(), sync_events::msc4186::response::Room::default()); response .extensions .receipts @@ -1846,7 +1845,7 @@ mod tests { #[test] fn to_device_without_events_is_empty() { let mut response = response(); - response.extensions.to_device = Some(sync_events::v5::response::ToDevice::default()); + response.extensions.to_device = Some(sync_events::msc4186::response::ToDevice::default()); assert!(response_is_empty(&response)); } diff --git a/src/api/client/well_known.rs b/src/api/client/well_known.rs index 749571f99..9e2552307 100644 --- a/src/api/client/well_known.rs +++ b/src/api/client/well_known.rs @@ -43,9 +43,9 @@ pub(crate) async fn well_known_client( /// homeserver, implementing MSC4143. pub(crate) async fn get_rtc_transports( State(services): State, - _body: Ruma, -) -> Result { - Ok(ruma::api::client::rtc::transports::v1::Response::new( + _body: Ruma, +) -> Result { + Ok(ruma::api::client::rtc::transports::msc4143::Response::new( services.config.matrix_rtc.foci.clone(), )) } diff --git a/src/api/server/stripped_state.rs b/src/api/server/stripped_state.rs index 9680583ce..8fb0dc83d 100644 --- a/src/api/server/stripped_state.rs +++ b/src/api/server/stripped_state.rs @@ -14,8 +14,8 @@ use crate::Ruma; /// [MSC4560]: https://github.com/matrix-org/matrix-spec-proposals/pull/4560 pub(crate) async fn get_stripped_state_route( State(services): State, - body: Ruma, -) -> Result { + body: Ruma, +) -> Result { if let Some(forbidden_type) = body .state .iter() @@ -42,7 +42,7 @@ pub(crate) async fn get_stripped_state_route( .await { | Accessibility::Accessible(RoomStrippedState { state, num_joined_members, .. }) => Ok( - assign!(get_stripped_state::unstable::Response::default(), { state, num_joined_members }), + assign!(get_stripped_state::msc4560::Response::default(), { state, num_joined_members }), ), | Accessibility::Inaccessible => { Err!(Request(Forbidden("You may not access this room's stripped state."))) diff --git a/src/service/rooms/summary/mod.rs b/src/service/rooms/summary/mod.rs index 42a277c17..ee643b6a6 100644 --- a/src/service/rooms/summary/mod.rs +++ b/src/service/rooms/summary/mod.rs @@ -666,7 +666,7 @@ impl Service { return Accessibility::NotFound; } - let request = assign!(get_stripped_state::unstable::Request::new(room_id.to_owned()), { + let request = assign!(get_stripped_state::msc4560::Request::new(room_id.to_owned()), { state: state.to_owned(), num_joined_members }); diff --git a/src/service/sync/mod.rs b/src/service/sync/mod.rs index e3a07c40d..46e56894b 100644 --- a/src/service/sync/mod.rs +++ b/src/service/sync/mod.rs @@ -6,7 +6,8 @@ use std::{ use conduwuit::{Result, SyncMutex, trace}; use futures::StreamExt; use ruma::{ - OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId, api::client::sync::sync_events::v5, + OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId, + api::client::sync::sync_events::msc4186, }; use tokio::sync::watch; @@ -24,21 +25,32 @@ struct Services { #[allow(unused, reason = "TODO refactor")] struct SlidingSyncCache { - lists: BTreeMap, - subscriptions: BTreeMap, + lists: BTreeMap, + subscriptions: BTreeMap, // For every room, the roomsince number known_rooms: BTreeMap>, - extensions: v5::request::Extensions, + extensions: msc4186::request::Extensions, } +/// For every in-scope room, the stream position up to which sticky events +/// have been delivered on this connection. +pub type StickyDelivered = BTreeMap; + #[derive(Default)] struct SnakeSyncCache { - lists: BTreeMap, - subscriptions: BTreeMap, + lists: BTreeMap, + subscriptions: BTreeMap, known_rooms: BTreeMap>, + sticky_delivered: StickyDelivered, acknowledged_pos: u64, - pending: BTreeMap>>, - extensions: v5::request::Extensions, + pending: BTreeMap, + extensions: msc4186::request::Extensions, +} + +#[derive(Default)] +struct PendingState { + known_rooms: BTreeMap>, + sticky_delivered: StickyDelivered, } impl SnakeSyncCache { @@ -47,13 +59,19 @@ impl SnakeSyncCache { } fn acknowledge(&mut self, pos: u64) { - if let Some(known_rooms) = self.pending.remove(&pos) { - self.known_rooms = known_rooms; + if let Some(pending) = self.pending.remove(&pos) { + self.known_rooms = pending.known_rooms; + self.sticky_delivered = pending.sticky_delivered; self.acknowledged_pos = pos; } } - fn stage_pending(&mut self, pos: u64, known_rooms: BTreeMap>) { + fn stage_pending( + &mut self, + pos: u64, + known_rooms: BTreeMap>, + sticky_delivered: StickyDelivered, + ) { let mut pending_rooms = self.known_rooms.clone(); for (list_id, rooms) in known_rooms { let list = pending_rooms.entry(list_id).or_default(); @@ -61,7 +79,10 @@ impl SnakeSyncCache { list.insert(room_id, pos); } } - self.pending.insert(pos, pending_rooms); + self.pending.insert(pos, PendingState { + known_rooms: pending_rooms, + sticky_delivered, + }); } } @@ -177,8 +198,8 @@ impl Service { &self, snake_key: &SnakeConnectionsKey, pos: u64, - request: &mut v5::Request, - ) -> BTreeMap> { + request: &mut msc4186::Request, + ) -> (BTreeMap>, StickyDelivered) { let mut cache = self.snake_connections.lock(); let cached = Arc::clone( cache @@ -190,7 +211,7 @@ impl Service { cached.acknowledge(pos); - //v5::Request::try_from_http_request(req, path_args); + //msc4186::Request::try_from_http_request(req, path_args); for (list_id, list) in &mut request.lists { if let Some(cached_list) = cached.lists.get(list_id) { list_or_sticky( @@ -224,18 +245,6 @@ impl Service { .room_subscriptions .extend(cached.subscriptions.clone()); - request.extensions.e2ee.enabled = request - .extensions - .e2ee - .enabled - .or(cached.extensions.e2ee.enabled); - - request.extensions.to_device.enabled = request - .extensions - .to_device - .enabled - .or(cached.extensions.to_device.enabled); - request.extensions.account_data.enabled = request .extensions .account_data @@ -275,9 +284,25 @@ impl Service { &mut request.extensions.receipts.lists, cached.extensions.receipts.lists.clone(), ); + some_or_sticky( + &mut request.extensions.sticky_events.enabled, + cached.extensions.sticky_events.enabled, + ); + some_or_sticky( + &mut request.extensions.sticky_events.limit, + cached.extensions.sticky_events.limit, + ); + some_or_sticky( + &mut request.extensions.sticky_events.rooms, + cached.extensions.sticky_events.rooms.clone(), + ); + some_or_sticky( + &mut request.extensions.sticky_events.lists, + cached.extensions.sticky_events.lists.clone(), + ); cached.extensions = request.extensions.clone(); - cached.known_rooms.clone() + (cached.known_rooms.clone(), cached.sticky_delivered.clone()) } pub fn store_snake_sync_pending( @@ -285,6 +310,7 @@ impl Service { key: &SnakeConnectionsKey, pos: u64, known_rooms: BTreeMap>, + sticky_delivered: StickyDelivered, ) { let mut cache = self.snake_connections.lock(); let cached = Arc::clone( @@ -295,13 +321,13 @@ impl Service { let cached = &mut cached.lock(); drop(cache); - cached.stage_pending(pos, known_rooms); + cached.stage_pending(pos, known_rooms, sticky_delivered); } pub fn update_snake_sync_subscriptions( &self, key: &SnakeConnectionsKey, - subscriptions: BTreeMap, + subscriptions: BTreeMap, ) { let mut cache = self.snake_connections.lock(); let cached = Arc::clone( @@ -412,10 +438,13 @@ mod tests { fn pending_rooms_are_committed_only_when_the_response_pos_is_acknowledged() { let room_id = OwnedRoomId::try_from("!room:example.com").unwrap(); let mut cache = SnakeSyncCache { - pending: BTreeMap::from([( - 42, - BTreeMap::from([("list".to_owned(), BTreeMap::from([(room_id, 42)]))]), - )]), + pending: BTreeMap::from([(42, PendingState { + known_rooms: BTreeMap::from([( + "list".to_owned(), + BTreeMap::from([(room_id, 42)]), + )]), + sticky_delivered: StickyDelivered::new(), + })]), ..Default::default() }; @@ -438,7 +467,11 @@ mod tests { let dropped = OwnedRoomId::try_from("!dropped:example.com").unwrap(); let mut cache = SnakeSyncCache::default(); let stage = |cache: &mut SnakeSyncCache, pos, rooms: BTreeSet| { - cache.stage_pending(pos, BTreeMap::from([("all_rooms".to_owned(), rooms)])); + cache.stage_pending( + pos, + BTreeMap::from([("all_rooms".to_owned(), rooms)]), + StickyDelivered::new(), + ); cache.acknowledge(pos); }; @@ -458,7 +491,10 @@ mod tests { fn overlapping_responses_accept_either_response_position() { let mut cache = SnakeSyncCache { acknowledged_pos: 100, - pending: BTreeMap::from([(200, BTreeMap::new()), (201, BTreeMap::new())]), + pending: BTreeMap::from([ + (200, PendingState::default()), + (201, PendingState::default()), + ]), ..Default::default() }; diff --git a/src/service/users/dehydrated_device.rs b/src/service/users/dehydrated_device.rs index 83ab1b343..065d9013e 100644 --- a/src/service/users/dehydrated_device.rs +++ b/src/service/users/dehydrated_device.rs @@ -3,7 +3,7 @@ use conduwuit_database::{Deserialized, Json}; use ruma::{ DeviceId, OwnedDeviceId, UserId, api::client::dehydrated_device::{ - DehydratedDeviceData, put_dehydrated_device::unstable::Request, + DehydratedDeviceData, put_dehydrated_device::msc3814::Request, }, serde::Raw, };