diff --git a/src/admin/room/directory/mod.rs b/src/admin/room/directory/mod.rs index 24b672ab88..a44ebb14c8 100644 --- a/src/admin/room/directory/mod.rs +++ b/src/admin/room/directory/mod.rs @@ -43,8 +43,9 @@ fn local_alias<'a>( room: &'a RoomOrAliasId, ) -> Result> { match <&RoomAliasId>::try_from(room).ok() { - | Some(alias) if !services.globals.alias_is_local(alias) => - Err!("Alias {alias} is not local to this server; use the room id instead"), + | Some(alias) if !services.globals.alias_is_local(alias) => { + Err!("Alias {alias} is not local to this server; use the room id instead") + }, | alias => Ok(alias), } } diff --git a/src/api/client/message.rs b/src/api/client/message.rs index d98a26ad14..da034df0ce 100644 --- a/src/api/client/message.rs +++ b/src/api/client/message.rs @@ -1,5 +1,10 @@ +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; + use axum::extract::State; -use futures::{FutureExt, StreamExt, TryFutureExt, future::Either, pin_mut}; +use futures::{StreamExt, TryFutureExt, future::Either, pin_mut}; use ruma::{ DeviceId, RoomId, UInt, UserId, api::{ @@ -13,7 +18,7 @@ use ruma::{ serde::Raw, }; use tuwunel_core::{ - Err, PduId, Result, at, + Err, PduId, Result, at, debug, matrix::{ event::{Event, Matches}, pdu::{PduCount, PduEvent}, @@ -83,6 +88,41 @@ type RelTypes = SmallVec<[RelationType; 1]>; const LIMIT_MAX: usize = 1000; const LIMIT_DEFAULT: usize = 10; +#[derive(Default)] +struct MessageFilterStats { + iterated: AtomicUsize, + event_filter_dropped: AtomicUsize, + related_by_filter_dropped: AtomicUsize, + ignored_dropped: AtomicUsize, + visibility_dropped: AtomicUsize, +} + +struct MessagePagination { + from: PduCount, + to: Option, + limit: usize, +} + +struct MessageCollectionContext<'a> { + services: &'a Services, + room_id: &'a RoomId, + sender_user: &'a UserId, + filter: &'a RoomEventFilter, + dir: Direction, + bypass_visibility: bool, + shortroomid: ShortRoomId, + encrypted: bool, +} + +struct MessageFilterContext<'a> { + services: &'a Services, + sender_user: &'a UserId, + filter: &'a RoomEventFilter, + bypass_visibility: bool, + shortroomid: ShortRoomId, + to: Option, +} + /// # `GET /_matrix/client/r0/rooms/{roomId}/messages` /// /// Allows paginating through room history. @@ -126,6 +166,49 @@ pub(crate) async fn get_messages( bypass_visibility, } = args; + validate_messages_request(services, room_id, sender_user, bypass_visibility).await?; + let pagination = parse_message_pagination(from, to, dir, limit)?; + maybe_backfill_messages(services, room_id, dir, pagination.from, pagination.to).await; + + let encrypted = services + .state_accessor + .is_encrypted_room(room_id) + .await; + + let shortroomid = services.short.get_shortroomid(room_id).await?; + let stats = Arc::new(MessageFilterStats::default()); + let context = MessageCollectionContext { + services, + room_id, + sender_user, + filter, + dir, + bypass_visibility, + shortroomid, + encrypted, + }; + let events = collect_message_events(&context, &pagination, &stats).await; + + let state = collect_lazy_loaded_state( + services, + room_id, + sender_user, + sender_device, + filter, + pagination.from, + &events, + ) + .await; + + build_messages_response(room_id, dir, &pagination, events, state, &stats) +} + +async fn validate_messages_request( + services: &Services, + room_id: &RoomId, + sender_user: &UserId, + bypass_visibility: bool, +) -> Result { if !services.metadata.exists(room_id).await { return Err!(Request(Forbidden("Room does not exist to this server"))); } @@ -139,7 +222,16 @@ pub(crate) async fn get_messages( return Err!(Request(Forbidden("You don't have permission to view this room."))); } - let from: PduCount = from + Ok(()) +} + +fn parse_message_pagination( + from: Option<&str>, + to: Option<&str>, + dir: Direction, + limit: Option, +) -> Result { + let from = from .map(str::parse) .transpose()? .unwrap_or_else(|| match dir { @@ -147,23 +239,82 @@ pub(crate) async fn get_messages( | Direction::Backward => PduCount::max(), }); - let to: Option = to.map(str::parse).flat_ok(); - - let limit: usize = limit + let to = to.map(str::parse).flat_ok(); + let limit = limit .and_then(|limit| limit.try_into().ok()) .unwrap_or(LIMIT_DEFAULT) .min(LIMIT_MAX); + Ok(MessagePagination { from, to, limit }) +} + +async fn maybe_backfill_messages( + services: &Services, + room_id: &RoomId, + dir: Direction, + from: PduCount, + to: Option, +) { if matches!(dir, Direction::Backward) { services .timeline - .backfill_if_required(room_id, from) + .backfill_if_required(room_id, from, to) .await .log_err() .ok(); } +} + +async fn collect_message_events( + context: &MessageCollectionContext<'_>, + pagination: &MessagePagination, + stats: &Arc, +) -> Vec { + let MessagePagination { from, to, limit } = *pagination; + let it = message_timeline_iter( + context.services, + context.room_id, + context.sender_user, + context.dir, + from, + ); + let filter_context = MessageFilterContext { + services: context.services, + sender_user: context.sender_user, + filter: context.filter, + bypass_visibility: context.bypass_visibility, + shortroomid: context.shortroomid, + to, + }; + + apply_message_filters(it, &filter_context, stats) + .take(limit) + .wide_then(|item| { + add_membership_unsigned( + context.services, + item, + context.sender_user, + context.encrypted, + ) + }) + .wide_then(|item| { + bundle_message_aggregations(context.services, context.sender_user, item) + }) + .collect() + .await +} - let it = match dir { +fn message_timeline_iter<'a>( + services: &'a Services, + room_id: &'a RoomId, + sender_user: &'a UserId, + dir: Direction, + from: PduCount, +) -> Either< + impl futures::Stream + 'a, + impl futures::Stream + 'a, +> { + match dir { | Direction::Forward => Either::Left( services .timeline @@ -176,67 +327,170 @@ pub(crate) async fn get_messages( .pdus_rev(Some(sender_user), room_id, Some(from)) .ignore_err(), ), - }; - - let encrypted = services - .state_accessor - .is_encrypted_room(room_id) - .await; - - let shortroomid = services.short.get_shortroomid(room_id).await?; + } +} - let events: Vec<_> = it - .ready_take_while(|(count, _)| Some(*count) != to) - .ready_filter_map(|item| event_filter(item, filter)) - .wide_filter_map(|item| related_by_filter(services, shortroomid, filter, item)) - .wide_filter_map(|item| event_filters(services, sender_user, item, bypass_visibility)) - .take(limit) - .wide_then(|item| add_membership_unsigned(services, item, sender_user, encrypted)) - .wide_then(async |(count, pdu)| { - let pdu = services - .pdu_metadata - .bundle_aggregations(sender_user, pdu) - .await; - - (count, pdu) +fn apply_message_filters<'a, S>( + it: S, + context: &'a MessageFilterContext<'a>, + stats: &Arc, +) -> impl futures::Stream + Send + 'a +where + S: futures::Stream + Send + 'a, +{ + let stats = Arc::clone(stats); + let to = context.to; + it.ready_take_while(move |(count, _)| Some(*count) != to) + .inspect({ + let stats = Arc::clone(&stats); + move |_| { + stats.iterated.fetch_add(1, Ordering::Relaxed); + } }) - .collect() - .await; + .ready_filter_map({ + let stats = Arc::clone(&stats); + move |item| event_filter_counted(item, context.filter, stats.as_ref()) + }) + .wide_filter_map({ + let stats = Arc::clone(&stats); + move |item| { + let stats = Arc::clone(&stats); + async move { + related_by_filter_counted( + context.services, + context.shortroomid, + context.filter, + item, + stats.as_ref(), + ) + .await + } + } + }) + .wide_filter_map({ + let stats = Arc::clone(&stats); + move |item| { + let stats = Arc::clone(&stats); + async move { + event_filters_counted( + context.services, + context.sender_user, + item, + context.bypass_visibility, + stats.as_ref(), + ) + .await + } + } + }) +} + +async fn collect_lazy_loaded_state( + services: &Services, + room_id: &RoomId, + sender_user: &UserId, + sender_device: Option<&DeviceId>, + filter: &RoomEventFilter, + from: PduCount, + events: &[PdusIterItem], +) -> Vec> { + let lazy_loading_context = + lazy_loading_context(room_id, sender_user, sender_device, filter, from); + let witness = + collect_lazy_loading_witness(services, &lazy_loading_context, filter, events).await; + + collect_lazy_loading_state_events(services, room_id, witness).await +} - let lazy_loading_context = lazy_loading::Context { +fn lazy_loading_context<'a>( + room_id: &'a RoomId, + sender_user: &'a UserId, + sender_device: Option<&'a DeviceId>, + filter: &'a RoomEventFilter, + from: PduCount, +) -> lazy_loading::Context<'a> { + lazy_loading::Context { user_id: sender_user, device_id: sender_device, room_id, token: Some(from.into_unsigned()), options: Some(&filter.lazy_load_options), mode: lazy_loading::Mode::Update, - }; + } +} - let witness = filter +async fn collect_lazy_loading_witness( + services: &Services, + lazy_loading_context: &lazy_loading::Context<'_>, + filter: &RoomEventFilter, + events: &[PdusIterItem], +) -> Option { + filter .lazy_load_options .is_enabled() - .then_async(|| lazy_loading_witness(services, &lazy_loading_context, events.iter())); - - let state = witness - .map(Option::into_iter) - .map(|option| option.flat_map(Witness::into_iter)) - .map(IterStream::stream) - .into_stream() - .flatten() + .then_async(|| lazy_loading_witness(services, lazy_loading_context, events.iter())) + .await +} + +async fn collect_lazy_loading_state_events( + services: &Services, + room_id: &RoomId, + witness: Option, +) -> Vec> { + witness + .into_iter() + .flat_map(Witness::into_iter) + .stream() .broad_filter_map(async |user_id| get_member_event(services, room_id, &user_id).await) .collect() + .await +} + +async fn bundle_message_aggregations( + services: &Services, + sender_user: &UserId, + (count, pdu): PdusIterItem, +) -> PdusIterItem { + let pdu = services + .pdu_metadata + .bundle_aggregations(sender_user, pdu) .await; + (count, pdu) +} + +fn build_messages_response( + room_id: &RoomId, + dir: Direction, + pagination: &MessagePagination, + events: Vec, + state: Vec>, + stats: &Arc, +) -> Result { let next_token = events.last().map(at!(0)); - let chunk = events + let chunk: Vec<_> = events .into_iter() .map(at!(1)) .map(Event::into_format) .collect(); + debug!( + room_id = %room_id, + ?dir, + limit = pagination.limit, + iterated = stats.iterated.load(Ordering::Relaxed), + event_filter_dropped = stats.event_filter_dropped.load(Ordering::Relaxed), + related_by_filter_dropped = stats.related_by_filter_dropped.load(Ordering::Relaxed), + ignored_dropped = stats.ignored_dropped.load(Ordering::Relaxed), + visibility_dropped = stats.visibility_dropped.load(Ordering::Relaxed), + returned = chunk.len(), + next_token = ?next_token, + "Returning messages page" + ); + Ok(get_message_events::v3::Response { - start: from.to_string(), + start: pagination.from.to_string(), end: next_token.as_ref().map(ToString::to_string), chunk, state, @@ -309,13 +563,48 @@ pub(crate) async fn event_filters( user_id: &UserId, item: PdusIterItem, bypass_visibility: bool, +) -> Option { + event_filters_inner(services, user_id, item, bypass_visibility, None).await +} + +async fn event_filters_counted( + services: &Services, + user_id: &UserId, + item: PdusIterItem, + bypass_visibility: bool, + stats: &MessageFilterStats, +) -> Option { + event_filters_inner(services, user_id, item, bypass_visibility, Some(stats)).await +} + +async fn event_filters_inner( + services: &Services, + user_id: &UserId, + item: PdusIterItem, + bypass_visibility: bool, + stats: Option<&MessageFilterStats>, ) -> Option { if bypass_visibility { return Some(item); } - let item = ignored_filter(services, item, user_id).await?; - let item = visibility_filter(services, item, user_id).await?; + let Some(item) = ignored_filter(services, item, user_id).await else { + if let Some(stats) = stats { + stats + .ignored_dropped + .fetch_add(1, Ordering::Relaxed); + } + return None; + }; + + let Some(item) = visibility_filter(services, item, user_id).await else { + if let Some(stats) = stats { + stats + .visibility_dropped + .fetch_add(1, Ordering::Relaxed); + } + return None; + }; Some(item) } @@ -350,6 +639,24 @@ pub(crate) async fn related_by_filter( .then_some(item) } +async fn related_by_filter_counted( + services: &Services, + shortroomid: ShortRoomId, + filter: &RoomEventFilter, + item: PdusIterItem, + stats: &MessageFilterStats, +) -> Option { + let result = related_by_filter(services, shortroomid, filter, item).await; + + if result.is_none() { + stats + .related_by_filter_dropped + .fetch_add(1, Ordering::Relaxed); + } + + result +} + #[inline] pub(crate) async fn ignored_filter( services: &Services, @@ -418,6 +725,22 @@ pub(crate) fn event_filter(item: PdusIterItem, filter: &RoomEventFilter) -> Opti filter.matches(pdu).then_some(item) } +fn event_filter_counted( + item: PdusIterItem, + filter: &RoomEventFilter, + stats: &MessageFilterStats, +) -> Option { + let result = event_filter(item, filter); + + if result.is_none() { + stats + .event_filter_dropped + .fetch_add(1, Ordering::Relaxed); + } + + result +} + /// MSC4115: stamp `unsigned.membership` on a served PDU with the requesting /// user's membership at the time of the event. The MSC permits omitting the /// property when calculating it is expensive, so the project restricts it to diff --git a/src/api/client/push/pushrules_rule.rs b/src/api/client/push/pushrules_rule.rs index 7d10fb3c71..c4e7364334 100644 --- a/src/api/client/push/pushrules_rule.rs +++ b/src/api/client/push/pushrules_rule.rs @@ -67,11 +67,13 @@ pub(crate) async fn set_pushrule_route( | BeforeHigherThanAfter => Err!(Request(InvalidParam( "The before rule has a higher priority than the after rule." ))), - | InvalidRuleId => - Err!(Request(InvalidParam("Rule ID containing invalid characters."))), + | InvalidRuleId => { + Err!(Request(InvalidParam("Rule ID containing invalid characters."))) + }, - | UnknownRuleId => - Err!(Request(NotFound("The before or after rule could not be found."))), + | UnknownRuleId => { + Err!(Request(NotFound("The before or after rule could not be found."))) + }, | _ => Err!(Request(InvalidParam("Invalid data."))), }; @@ -98,8 +100,9 @@ pub(crate) async fn delete_pushrule_route( .remove(body.kind.clone(), &body.rule_id) { return match error { - | RemovePushRuleError::ServerDefault => - Err!(Request(InvalidParam("Cannot delete a server-default pushrule."))), + | RemovePushRuleError::ServerDefault => { + Err!(Request(InvalidParam("Cannot delete a server-default pushrule."))) + }, | RemovePushRuleError::NotFound => Err!(Request(NotFound("Push rule not found."))), diff --git a/src/api/client/session/sso.rs b/src/api/client/session/sso.rs index 09c71ecdbe..87faf8ea4e 100644 --- a/src/api/client/session/sso.rs +++ b/src/api/client/session/sso.rs @@ -453,8 +453,9 @@ pub(crate) async fn sso_callback_route( | true => "sso", // Present in LDAP is an existing user to provision, not a new registration. | false if ldap_user_exists(&services, &user_id).await => "ldap", - | false => - return Err!(Request(Forbidden("Registration from this provider is disabled"))), + | false => { + return Err!(Request(Forbidden("Registration from this provider is disabled"))); + }, }; register_user(&services, &provider, &session, &userinfo, &user_id, origin).await?; diff --git a/src/api/oidc/account.rs b/src/api/oidc/account.rs index 034fe361f1..10582d70a5 100644 --- a/src/api/oidc/account.rs +++ b/src/api/oidc/account.rs @@ -259,11 +259,12 @@ async fn handle_account_callback( | "org.matrix.sessions_list" => consume_login_token(services, login_token).await?, | _ if method == Method::POST => consume_login_token(services, login_token).await?, | _ if method == Method::GET => peek_login_token(services, login_token).await?, - | _ => + | _ => { return Err!(HttpJson(METHOD_NOT_ALLOWED, { "errcode": "M_UNRECOGNIZED", "error": "Unsupported account management method", - })), + })); + }, }; match action { diff --git a/src/api/oidc/authorize.rs b/src/api/oidc/authorize.rs index 1eb3b98b7f..1885f4eb1f 100644 --- a/src/api/oidc/authorize.rs +++ b/src/api/oidc/authorize.rs @@ -60,11 +60,13 @@ pub(crate) async fn authorize_route( // RFC 7636 / MSC2964: require an explicit S256 challenge; bare `plain` is // rejected. match (¶ms.code_challenge, params.code_challenge_method.as_deref()) { - | (None, _) if services.config.oidc_require_pkce => - return Err!(Request(InvalidParam("code_challenge is required (PKCE with S256)"))), + | (None, _) if services.config.oidc_require_pkce => { + return Err!(Request(InvalidParam("code_challenge is required (PKCE with S256)"))); + }, - | (Some(_), method) if method != Some("S256") => - return Err!(Request(InvalidParam("Only code_challenge_method=S256 is supported"))), + | (Some(_), method) if method != Some("S256") => { + return Err!(Request(InvalidParam("Only code_challenge_method=S256 is supported"))); + }, | _ => {}, } @@ -99,8 +101,9 @@ pub(crate) async fn authorize_route( let idp_id = match (serve_native, resolved_idp) { | (true, _) => None, | (false, Some(idp_id)) => Some(idp_id), - | (false, None) => - return Err!(Config("identity_provider", "No identity provider configured")), + | (false, None) => { + return Err!(Config("identity_provider", "No identity provider configured")); + }, }; let auth_req = AuthRequest { diff --git a/src/api/oidc/registration.rs b/src/api/oidc/registration.rs index 49697c94db..61b488ccf7 100644 --- a/src/api/oidc/registration.rs +++ b/src/api/oidc/registration.rs @@ -122,8 +122,9 @@ fn validate_client_metadata(body: &DcrRequest, require_client_uri: bool) -> Resu { match parse_https(uri).as_ref().and_then(Url::host_str) { | Some(host) if shares_base(host, base) => {}, - | Some(_) => - return Err(DcrError::Metadata("a metadata URI must share the client_uri host")), + | Some(_) => { + return Err(DcrError::Metadata("a metadata URI must share the client_uri host")); + }, | None => return Err(DcrError::Metadata("a metadata URI must be an https URL")), } } diff --git a/src/api/router/auth.rs b/src/api/router/auth.rs index 34435d2ae1..db1393041a 100644 --- a/src/api/router/auth.rs +++ b/src/api/router/auth.rs @@ -169,8 +169,9 @@ fn check_auth_still_required(services: &Services, token: &Token, route: TypeId) { match token { | Token::Appservice(_) | Token::User(_) => Ok(()), - | Token::None | Token::Expired(_) | Token::Invalid => - Err!(Request(MissingToken("Missing or invalid access token."))), + | Token::None | Token::Expired(_) | Token::Invalid => { + Err!(Request(MissingToken("Missing or invalid access token."))) + }, } } else { Ok(()) diff --git a/src/api/router/auth/dispatch.rs b/src/api/router/auth/dispatch.rs index 93e8636bdd..a2862829e7 100644 --- a/src/api/router/auth/dispatch.rs +++ b/src/api/router/auth/dispatch.rs @@ -170,8 +170,9 @@ impl AuthDispatch for AppserviceToken { match token { | Token::Invalid => unknown_token(), | Token::Expired(access_token) => expired_token(services, &access_token).await, - | Token::User(_) => - Err!(Request(Unauthorized("Appservice tokens must be used on this endpoint."))), + | Token::User(_) => { + Err!(Request(Unauthorized("Appservice tokens must be used on this endpoint."))) + }, | Token::Appservice(info) => Ok(Auth { appservice_info: Some(*info), ..Auth::default() @@ -222,8 +223,9 @@ impl AuthDispatch for ServerSignatures { match token { | Token::Invalid => unknown_token(), | Token::Expired(access_token) => expired_token(services, &access_token).await, - | Token::Appservice(_) | Token::User(_) => - Err!(Request(Unauthorized("Server signatures must be used on this endpoint."))), + | Token::Appservice(_) | Token::User(_) => { + Err!(Request(Unauthorized("Server signatures must be used on this endpoint."))) + }, | Token::None => Ok(auth_server(services, request, json_body).await?), } } diff --git a/src/api/server/backfill.rs b/src/api/server/backfill.rs index 411146e555..69a766a292 100644 --- a/src/api/server/backfill.rs +++ b/src/api/server/backfill.rs @@ -67,17 +67,18 @@ pub(crate) async fn get_backfill_route( pdus: services .timeline .pdus_rev(None, &body.room_id, Some(from.saturating_add(1))) - .try_filter_map(async |(_, pdu)| { - Ok(services + .try_filter_map(async |pdu| { + if !services .state_accessor - .server_can_see_event(body.origin(), &pdu.room_id, &pdu.event_id) + .server_can_see_event(body.origin(), &body.room_id, &pdu.1.event_id) .await - .then_some(pdu)) - }) - .try_filter_map(async |pdu| { + { + return Ok(None); + } + Ok(services .timeline - .get_pdu_json(&pdu.event_id) + .get_pdu_json(&pdu.1.event_id) .await .ok()) }) diff --git a/src/api/server/get_missing_events.rs b/src/api/server/get_missing_events.rs index 641a7acc6d..9d573082a0 100644 --- a/src/api/server/get_missing_events.rs +++ b/src/api/server/get_missing_events.rs @@ -1,4 +1,6 @@ -use std::collections::{BTreeSet, VecDeque}; +use std::{ + collections::{BTreeSet, VecDeque}, +}; use axum::extract::State; use futures::{ diff --git a/src/main/tests/pusher_notify.rs b/src/main/tests/pusher_notify.rs index 51b2a2adc6..6937f47c01 100644 --- a/src/main/tests/pusher_notify.rs +++ b/src/main/tests/pusher_notify.rs @@ -202,8 +202,9 @@ async fn verify_badge_recovery(services: &Services, recovery: &mut BadgeRecovery match recovery.rx.try_recv() { | Err(TryRecvError::Empty) => Ok(()), - | Err(TryRecvError::Disconnected) => - Err!("stub gateway channel closed after badge recovery"), + | Err(TryRecvError::Disconnected) => { + Err!("stub gateway channel closed after badge recovery") + }, | Ok(_) => Err!("stale badge wake produced a duplicate notification"), } } @@ -515,8 +516,9 @@ async fn badge_count_opt_out(fixture: &Fixture<'_>) -> Result { match rx.try_recv() { | Err(TryRecvError::Empty) => Ok(()), - | Err(TryRecvError::Disconnected) => - Err!("stub gateway channel closed after badge opt-out"), + | Err(TryRecvError::Disconnected) => { + Err!("stub gateway channel closed after badge opt-out") + }, | Ok(_) => Err!("badge opt-out emitted a counts-only notification"), } } diff --git a/src/main/tests/short_id_allocation.rs b/src/main/tests/short_id_allocation.rs index 9ef9a9e86c..92e758cb34 100644 --- a/src/main/tests/short_id_allocation.rs +++ b/src/main/tests/short_id_allocation.rs @@ -1,15 +1,24 @@ #![cfg(test)] -use std::{env::var, fs::remove_dir_all, path::PathBuf, process::id as process_id}; +use std::{ + env::var, fs::remove_dir_all, net::TcpListener, path::PathBuf, process::id as process_id, + time::Duration, +}; use futures::{StreamExt, pin_mut}; +use serde_json::{Value, json}; +use tokio::time::{sleep, timeout}; use tuwunel::{Args, Runtime, Server, async_run, async_start, async_stop}; use tuwunel_core::{ - Err, Result, - ruma::{OwnedEventId, event_id}, + Err, Result, err, + matrix::pdu::PduBuilder, + ruma::{ + OwnedEventId, OwnedRoomId, RoomVersionId, UserId, event_id, + events::room::create::RoomCreateEventContent, room_id, + }, utils::stream::ReadyExt, }; -use tuwunel_service::Services; +use tuwunel_service::{Services, users::Register}; const OCCURRENCES: usize = 8; @@ -21,6 +30,9 @@ impl Drop for DatabasePath { #[test] fn batch_duplicates_share_one_shorteventid() -> Result { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let port = listener.local_addr()?.port(); + let root = var("TMPDIR").unwrap_or_else(|_| "/nvme/target/tmp".into()); let db_path = DatabasePath( PathBuf::from(root).join(format!("tuwunel-short-id-allocation-{}", process_id())), @@ -28,22 +40,34 @@ fn batch_duplicates_share_one_shorteventid() -> Result { let mut args = Args::default_test(&["fresh", "cleanup"]); args.maintenance = true; - args.option - .push(format!("database_path={:?}", db_path.0)); + args.option.extend([ + format!("database_path={:?}", db_path.0), + "address=[\"127.0.0.1\"]".to_owned(), + format!("port={port}"), + "listening=true".to_owned(), + ]); let runtime = Runtime::new(Some(&args))?; let server = Server::new(Some(&args), Some(&runtime))?; let result = runtime.block_on(async { let services = async_start(&server).await?; - let outcome = exercise(&services).await; - let shutdown = server.server.shutdown(); + let base = format!("http://127.0.0.1:{port}"); - drop(services); + drop(listener); + + let exercise = async { + let outcome = exercise(&services, &base).await; + let shutdown = server.server.shutdown(); + + outcome.and(shutdown) + }; + let run = async_run(&server); - let run = async_run(&server).await; + let (outcome, run) = tokio::join!(exercise, run); + drop(services); let stop = async_stop(&server).await; - outcome.and(shutdown).and(run).and(stop) + outcome.and(run).and(stop) }); drop(runtime); @@ -51,7 +75,10 @@ fn batch_duplicates_share_one_shorteventid() -> Result { result } -async fn exercise(services: &Services) -> Result { +async fn exercise(services: &Services, base: &str) -> Result { + create_hash_and_sign_does_not_allocate_short_id(services).await?; + repeated_identical_state_resend_does_not_allocate_short_id(services, base).await?; + let event_id = event_id!("$short-id-allocation-batch:localhost"); // a repeated event misses the batched lookup on every occurrence let batch = [event_id; OCCURRENCES]; @@ -81,3 +108,223 @@ async fn exercise(services: &Services) -> Result { Ok(()) } + +async fn repeated_identical_state_resend_does_not_allocate_short_id( + services: &Services, + base: &str, +) -> Result { + wait_until_ready(services, base).await?; + + let user_id = UserId::parse_with_server_name("shortidalice", services.globals.server_name())?; + let token = "short-id-allocation-token-0000000000000000"; + + services + .users + .full_register(Register { + user_id: Some(&user_id), + password: Some("short-id-allocation-password"), + ..Default::default() + }) + .await?; + + services + .users + .create_device(&user_id, None, (Some(token), None), None, None, None) + .await?; + + let room_id = create_room(services, base, token).await?; + let content = json!({"topic": "Short ID resend regression"}); + + let first_event_id = send_state_event(services, base, token, &room_id, &content).await?; + let second_event_id = send_state_event(services, base, token, &room_id, &content).await?; + + if second_event_id != first_event_id.as_str() { + return Err!("identical state resend returned a different event id"); + } + + let current_event_id = current_state_event_id(services, base, token, &room_id).await?; + + if current_event_id != first_event_id.as_str() { + return Err!("identical state resend overwrote the room state"); + } + + Ok(()) +} + +async fn create_hash_and_sign_does_not_allocate_short_id(services: &Services) -> Result { + let sender = services.globals.server_user.as_ref(); + if !services.users.exists(sender).await { + services.users.create(sender, None, None).await?; + } + + let room_id = room_id!("!short-id-no-append:localhost"); + let state_lock = services.state.mutex.lock(room_id).await; + services + .short + .get_or_create_shortroomid(room_id) + .await; + let (pdu, pdu_json, _prev_state) = services + .timeline + .create_hash_and_sign_event( + PduBuilder::state(String::new(), &RoomCreateEventContent { + federate: true, + predecessor: None, + room_version: RoomVersionId::V11, + ..RoomCreateEventContent::new_v11() + }), + sender, + room_id, + &state_lock, + ) + .await?; + let event_id = pdu.event_id.clone(); + + if services + .short + .get_shorteventid(&event_id) + .await + .is_ok() + { + return Err!("create_hash_and_sign_event allocated a short event id before append"); + } + + services + .timeline + .append_created_pdu(pdu, pdu_json, sender, &state_lock) + .await?; + + if services + .short + .get_shorteventid(&event_id) + .await + .is_err() + { + return Err!("append_created_pdu did not allocate a short event id"); + } + + Ok(()) +} + +async fn wait_until_ready(services: &Services, base: &str) -> Result { + let url = format!("{base}/_matrix/client/versions"); + + timeout(Duration::from_secs(10), async { + loop { + let response = services + .client + .clients + .default + .get(&url) + .send() + .await; + + if matches!(response, Ok(response) if response.status().is_success()) { + break; + } + + sleep(Duration::from_millis(20)).await; + } + }) + .await + .map_err(|_| err!("server listener did not become ready"))?; + + Ok(()) +} + +async fn create_room(services: &Services, base: &str, token: &str) -> Result { + let response = services + .client + .clients + .default + .post(format!("{base}/_matrix/client/v3/createRoom")) + .bearer_auth(token) + .json(&json!({})) + .send() + .await? + .error_for_status()? + .json::() + .await?; + + let room_id = response + .get("room_id") + .and_then(Value::as_str) + .ok_or_else(|| err!("createRoom response omitted room_id"))?; + + Ok(room_id.try_into()?) +} + +async fn send_state_event( + services: &Services, + base: &str, + token: &str, + room_id: &OwnedRoomId, + content: &Value, +) -> Result { + let response = services + .client + .clients + .default + .put(format!( + "{base}/_matrix/client/v3/rooms/{room_id}/state/m.room.topic/short-id-allocation" + )) + .bearer_auth(token) + .json(content) + .send() + .await? + .error_for_status()? + .json::() + .await?; + + response + .get("event_id") + .and_then(Value::as_str) + .map(str::to_owned) + .ok_or_else(|| err!("state PUT response omitted event_id")) +} + +async fn current_state_event_id( + services: &Services, + base: &str, + token: &str, + room_id: &OwnedRoomId, +) -> Result { + let response = services + .client + .clients + .default + .get(current_state_event_id_url(base, room_id)) + .bearer_auth(token) + .send() + .await? + .error_for_status()? + .json::() + .await?; + + response + .get("event_id") + .and_then(Value::as_str) + .map(str::to_owned) + .ok_or_else(|| err!("state GET response omitted event_id")) +} + +fn current_state_event_id_url(base: &str, room_id: &tuwunel_core::ruma::RoomId) -> String { + format!( + "{base}/_matrix/client/v3/rooms/{room_id}/state/m.room.topic/short-id-allocation?\ + format=event" + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn current_state_event_id_url_keeps_exact_query() { + let room_id = room_id!("!short-id-allocation:example.org"); + + assert_eq!( + current_state_event_id_url("http://localhost:8008", room_id), + "http://localhost:8008/_matrix/client/v3/rooms/!short-id-allocation:example.org/state/m.room.topic/short-id-allocation?format=event", + ); + } +} diff --git a/src/service/membership/invite.rs b/src/service/membership/invite.rs index 0e67ba26b7..feb9c60940 100644 --- a/src/service/membership/invite.rs +++ b/src/service/membership/invite.rs @@ -64,7 +64,7 @@ async fn remote_invite( .fill_profile_data(user_id, &mut content) .await; - let (pdu, pdu_json) = self + let (pdu, pdu_json, _prev_state) = self .services .timeline .create_hash_and_sign_event( diff --git a/src/service/membership/join.rs b/src/service/membership/join.rs index f9d90437fa..ed9b6ad20d 100644 --- a/src/service/membership/join.rs +++ b/src/service/membership/join.rs @@ -56,6 +56,14 @@ pub struct Join<'a> { pub extra_content: Option, } +fn membership_event_id_snapshot(event_id: Result) -> Result> { + match event_id { + | Ok(event_id) => Ok(Some(event_id)), + | Err(e) if e.is_not_found() => Ok(None), + | Err(e) => Err(e), + } +} + #[implement(Service)] #[async_noinline] #[tracing::instrument( @@ -192,6 +200,118 @@ async fn join_remote( ) -> Result { info!("Joining {room_id} over federation."); + let initial_membership = self + .services + .state_cache + .user_membership(sender_user, room_id) + .await; + let initial_membership_event_id = membership_event_id_snapshot( + self.services + .state_accessor + .room_state_get_id(room_id, &StateEventType::RoomMember, sender_user.as_str()) + .await, + )?; + + // Release the caller's room-state lock before the network round trip below. + // Holding it for the whole call creates a lock-order cycle with inbound + // federation `/send`: `send.rs`'s `handle_room` takes `mutex_federation` + // first and only then `state.mutex` (via `upgrade_outlier_to_timeline_pdu`), + // while this function used to hold `state.mutex` (acquired by our caller) + // across the entire call before taking `mutex_federation` below. If an + // inbound transaction for this room arrives while a remote join for it is + // in flight, each side ends up waiting on the lock the other already holds. + // We only actually need the room-state lock for the commit at the end, + // once `mutex_federation` is already held -- see the fresh acquisition + // further down, which keeps the same order the inbound path uses. + drop(state_lock); + + let ( + room_version_id, + room_version_rules, + mut join_event, + event_id, + join_authorized_via_users_server, + remote_server, + ) = self + .prepare_remote_join(sender_user, room_id, reason, servers, extra_content) + .await?; + + // Once send_join hits the remote server it may start sending us events which + // have to be belayed until we process this response first. + let _federation_lock = self + .services + .event_handler + .mutex_federation + .lock(room_id) + .await; + + let response = self + .fetch_and_prepare_send_join_response( + SendJoinRequest { + remote_server: &remote_server, + room_id, + event_id: &event_id, + servers, + room_version_id: &room_version_id, + join_authorized_via_users_server: join_authorized_via_users_server.as_ref(), + }, + &mut join_event, + ) + .await?; + + let (parsed_join_pdu, join_event, state) = self + .ingest_remote_join_response( + room_id, + &event_id, + join_event, + &room_version_id, + &room_version_rules, + &response, + ) + .await?; + + self.auth_check_send_join_response(&room_version_rules, &parsed_join_pdu, &state) + .await?; + + self.commit_remote_join( + sender_user, + room_id, + initial_membership, + initial_membership_event_id, + state, + parsed_join_pdu, + join_event, + ) + .await?; + + Ok(()) +} + +struct SendJoinRequest<'a> { + remote_server: &'a OwnedServerName, + room_id: &'a RoomId, + event_id: &'a OwnedEventId, + servers: &'a [OwnedServerName], + room_version_id: &'a RoomVersionId, + join_authorized_via_users_server: Option<&'a OwnedUserId>, +} + +#[implement(Service)] +async fn prepare_remote_join( + &self, + sender_user: &UserId, + room_id: &RoomId, + reason: Option, + servers: &[OwnedServerName], + extra_content: Option, +) -> Result<( + RoomVersionId, + RoomVersionRules, + CanonicalJsonObject, + OwnedEventId, + Option, + OwnedServerName, +)> { let (make_join_response, remote_server) = self .make_join_request(sender_user, room_id, servers) .await?; @@ -200,7 +320,7 @@ async fn join_remote( let room_version_id = self.require_supported_remote_room_version(&make_join_response)?; let room_version_rules = room_version::rules(&room_version_id)?; - let (mut join_event, event_id, join_authorized_via_users_server) = self + let (join_event, event_id, join_authorized_via_users_server) = self .create_join_event( room_id, sender_user, @@ -212,40 +332,66 @@ async fn join_remote( ) .await?; - // Once send_join hits the remote server it may start sending us events which - // have to be belayed until we process this response first. - let _federation_lock = self - .services - .event_handler - .mutex_federation - .lock(room_id) - .await; + Ok(( + room_version_id, + room_version_rules, + join_event, + event_id, + join_authorized_via_users_server, + remote_server, + )) +} +#[implement(Service)] +async fn fetch_and_prepare_send_join_response( + &self, + request: SendJoinRequest<'_>, + join_event: &mut CanonicalJsonObject, +) -> Result { let mut response = self .execute_send_join( - &remote_server, - room_id, - &event_id, + request.remote_server, + request.room_id, + request.event_id, join_event.clone(), - &room_version_id, + request.room_version_id, ) .await?; if response.members_omitted { - self.fetch_omitted_state(&remote_server, room_id, &event_id, servers, &mut response) - .await?; + self.fetch_omitted_state( + request.remote_server, + request.room_id, + request.event_id, + request.servers, + &mut response, + ) + .await?; } - if join_authorized_via_users_server.is_some() { + if request.join_authorized_via_users_server.is_some() { merge_restricted_signature( - &remote_server, - &event_id, - &room_version_id, + request.remote_server, + request.event_id, + request.room_version_id, &response, - &mut join_event, + join_event, )?; } + Ok(response) +} + +#[implement(Service)] +async fn ingest_remote_join_response( + &self, + room_id: &RoomId, + event_id: &OwnedEventId, + join_event: CanonicalJsonObject, + room_version_id: &RoomVersionId, + room_version_rules: &RoomVersionRules, + response: &federation::membership::create_join_event::v2::RoomState, +) -> Result<(Pdu, CanonicalJsonObject, HashMap)> { let shortroomid = self .services .short @@ -258,7 +404,7 @@ async fn join_remote( "Initialized room. Parsing join event..." ); let (parsed_join_pdu, join_event) = - Pdu::from_object_federation(room_id, &event_id, join_event, &room_version_rules)?; + Pdu::from_object_federation(room_id, event_id, join_event, room_version_rules)?; info!( events = response @@ -278,21 +424,31 @@ async fn join_remote( .await; let state = self - .ingest_send_join_state(room_id, &room_version_id, &room_version_rules, &response.state) + .ingest_send_join_state(room_id, room_version_id, room_version_rules, &response.state) .await; self.ingest_send_join_auth_chain( room_id, - &room_version_id, - &room_version_rules, + room_version_id, + room_version_rules, &response.auth_chain, ) .await; + Ok((parsed_join_pdu, join_event, state)) +} + +#[implement(Service)] +async fn auth_check_send_join_response( + &self, + room_version_rules: &RoomVersionRules, + parsed_join_pdu: &Pdu, + state: &HashMap, +) -> Result { debug!("Running send_join auth check..."); state_res::auth_check( - &room_version_rules, - &parsed_join_pdu, + room_version_rules, + parsed_join_pdu, &async |event_id| self.services.timeline.get_pdu(&event_id).await, &async |event_type, state_key| { let shortstatekey = self @@ -310,14 +466,64 @@ async fn join_remote( ) .inspect_err(|e| error!("send_join auth check failed: {e:?}")) .boxed() - .await?; + .await +} + +#[implement(Service)] +#[expect(clippy::too_many_arguments)] +async fn commit_remote_join( + &self, + sender_user: &UserId, + room_id: &RoomId, + initial_membership: Option, + initial_membership_event_id: Option, + state: HashMap, + parsed_join_pdu: Pdu, + join_event: CanonicalJsonObject, +) -> Result { + // Reacquire the room-state lock only now, for the commit below. We already + // hold `mutex_federation` in the caller, so this preserves the same + // `mutex_federation` -> `state.mutex` order the inbound federation `/send` + // path uses. + let state_lock = self.services.state.mutex.lock(room_id).await; + + let current_membership = self + .services + .state_cache + .user_membership(sender_user, room_id) + .await; + let current_membership_event_id = membership_event_id_snapshot( + self.services + .state_accessor + .room_state_get_id(room_id, &StateEventType::RoomMember, sender_user.as_str()) + .await, + )?; + + if current_membership == Some(MembershipState::Join) { + debug!( + %sender_user, + %room_id, + "Skipping stale remote join commit because the user is already joined" + ); + + return Ok(()); + } + + if current_membership_event_id != initial_membership_event_id { + debug_warn!( + %sender_user, + %room_id, + current_membership = ?current_membership, + initial_membership = ?initial_membership, + "Skipping stale remote join commit after a newer local membership change" + ); + + return Err!(Conflict("Join was superseded by a newer membership change.")); + } self.apply_send_join_state(room_id, &state, &state_lock) .await?; - // We append to state before appending the pdu, so we don't have a moment in - // time with the pdu without it's state. This is okay because append_pdu can't - // fail. let statehash_after_join = self .services .state @@ -339,8 +545,6 @@ async fn join_remote( ) .await?; - // We set the room state after inserting the pdu, so that we never have a moment - // in time where events in the current room state do not exist self.services .state .set_room_state(room_id, statehash_after_join, &state_lock); @@ -1136,3 +1340,33 @@ pub(super) async fn get_servers_for_room( debug_info!(?servers); Ok(servers) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn membership_event_id_snapshot_keeps_found_event() { + let event_id = ruma::event_id!("$test:example.org").to_owned(); + + let snapshot = + membership_event_id_snapshot(Ok(event_id.clone())).expect("snapshot should succeed"); + + assert_eq!(snapshot, Some(event_id)); + } + + #[test] + fn membership_event_id_snapshot_treats_missing_state_as_none() { + let snapshot = membership_event_id_snapshot(Err(err!(Request(NotFound("missing"))))) + .expect("not found should be treated as absent"); + + assert_eq!(snapshot, None); + } + + #[test] + fn membership_event_id_snapshot_preserves_other_errors() { + let snapshot = membership_event_id_snapshot(Err(err!(Database("boom")))); + + assert!(snapshot.is_err(), "database failures must not be collapsed"); + } +} diff --git a/src/service/migrations/injectivity/repair.rs b/src/service/migrations/injectivity/repair.rs index 731c400912..33f58d4f13 100644 --- a/src/service/migrations/injectivity/repair.rs +++ b/src/service/migrations/injectivity/repair.rs @@ -118,7 +118,7 @@ async fn patch_statediffs(services: &Services, scan: &Scan) -> Result { .ready_fold(Digests::new(), |mut digests, (key, value)| { let infected = short_of(value) .filter(|state| scan.infected.contains(state)) - .and_then(|state| key.try_into().ok().map(|digest| (state, digest))); + .zip(key.try_into().ok()); if let Some((state, digest)) = infected { digests.entry(state).or_default().push(digest); diff --git a/src/service/migrations/mod.rs b/src/service/migrations/mod.rs index 286911816a..eea7f72c0c 100644 --- a/src/service/migrations/mod.rs +++ b/src/service/migrations/mod.rs @@ -356,8 +356,9 @@ async fn migrate(services: &Services, foreign_lineage: bool) -> Result { .bump_database_version(target_version); match discovered.cmp(&target_version) { - | Ordering::Less => - info!("Database: migrated schema version from {discovered} to {target_version}."), + | Ordering::Less => { + info!("Database: migrated schema version from {discovered} to {target_version}."); + }, | Ordering::Greater => warn!( "Database: stamped schema version {target_version} over a higher discovered version \ {discovered} (forced downgrade or foreign import)." diff --git a/src/service/rooms/event_handler/policy_server.rs b/src/service/rooms/event_handler/policy_server.rs index 3243b6ffdd..f49e5c9d39 100644 --- a/src/service/rooms/event_handler/policy_server.rs +++ b/src/service/rooms/event_handler/policy_server.rs @@ -247,8 +247,9 @@ where let event_id = pdu.event_id(); match self.cached_policy_state(event_id).await { - | Some(PolicySigState::Refused { .. }) => - return Err!(Request(Forbidden("Event was rejected by the room's policy server."))), + | Some(PolicySigState::Refused { .. }) => { + return Err!(Request(Forbidden("Event was rejected by the room's policy server."))); + }, | Some(PolicySigState::BackoffUntil { until_secs }) if until_secs > now_secs() => { debug!(via = %policy.via, until_secs, "skipping outbound /sign during policy backoff"); diff --git a/src/service/rooms/spaces/mod.rs b/src/service/rooms/spaces/mod.rs index 762a33a619..6bcee0e79a 100644 --- a/src/service/rooms/spaces/mod.rs +++ b/src/service/rooms/spaces/mod.rs @@ -103,8 +103,9 @@ pub async fn get_summary_and_children( .or_else(async |e| match e { | _ if !e.is_not_found() => Err(e), - | _ if via.is_empty() => - Err!(Request(NotFound("Space room not found locally; not querying federation"))), + | _ if via.is_empty() => { + Err!(Request(NotFound("Space room not found locally; not querying federation"))) + }, | _ => self.get_summary_and_children_federation(room_id, sender, via) diff --git a/src/service/rooms/state_accessor/user_can.rs b/src/service/rooms/state_accessor/user_can.rs index f009599edc..f5985ac39f 100644 --- a/src/service/rooms/state_accessor/user_can.rs +++ b/src/service/rooms/state_accessor/user_can.rs @@ -122,7 +122,7 @@ pub async fn user_can_see_event( | HistoryVisibility::Shared | _ => self.services .state_cache - .is_joined(user_id, room_id) + .once_joined(user_id, room_id) .await, } } diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 73ac847485..a62ebf5c06 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -330,12 +330,13 @@ async fn append_pdu_effects( } } }, - | TimelineEventType::RoomTopic => + | TimelineEventType::RoomTopic => { if let Some(topic) = pdu.get_content().ok().and_then(plain_text_topic) { self.services .search .index_pdu(shortroomid, &pdu_id, &topic); - }, + } + }, | _ => {}, } diff --git a/src/service/rooms/timeline/backfill.rs b/src/service/rooms/timeline/backfill.rs index 1d4c1c8871..2a36633ce5 100644 --- a/src/service/rooms/timeline/backfill.rs +++ b/src/service/rooms/timeline/backfill.rs @@ -6,8 +6,10 @@ use futures::{ }; use rand::seq::SliceRandom; use ruma::{ - CanonicalJsonObject, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName, - api::Direction, events::TimelineEventType, + CanonicalJsonObject, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId, OwnedServerName, + RoomId, ServerName, UserId, + api::Direction, + events::{StateEventType, TimelineEventType}, }; use serde::Deserialize; use serde_json::value::RawValue as RawJsonValue; @@ -45,16 +47,28 @@ struct TimestampHit { #[implement(super::Service)] #[tracing::instrument(name = "backfill", level = "debug", skip(self))] -pub async fn backfill_if_required(&self, room_id: &RoomId, from: PduCount) -> Result { - let (first_pdu_count, first_pdu) = self - .first_item_in_room(room_id) - .await - .expect("Room is not empty"); - - // No backfill required, there are still events between them - if first_pdu_count < from { - return Ok(()); - } +pub async fn backfill_if_required( + &self, + room_id: &RoomId, + from: PduCount, + to: Option, +) -> Result { + let first_pdu = if from == PduCount::max() { + // The first backward `/messages` page starts from the room head, so + // backfill from the newest local event rather than the oldest one. + self.latest_item_in_room(None, room_id).await? + } else { + let (first_pdu_count, first_pdu) = self.first_item_in_room(room_id).await?; + + // If the request range stays entirely newer than the oldest local event, + // the existing history already covers it and no federation backfill is + // needed. + if request_is_local_only(first_pdu_count, from, to) { + return Ok(()); + } + + first_pdu + }; // No backfill required, reached the end. if *first_pdu.event_type() == TimelineEventType::RoomCreate { @@ -122,6 +136,15 @@ pub async fn backfill_if_required(&self, room_id: &RoomId, from: PduCount) -> Re Ok(()) } +#[inline] +fn request_is_local_only( + first_pdu_count: PduCount, + from: PduCount, + to: Option, +) -> bool { + first_pdu_count < from && to.is_none_or(|to| first_pdu_count <= to) +} + #[implement(super::Service)] async fn backfill_candidates(&self, room_id: &RoomId) -> Candidates { let canonical_alias = self @@ -136,6 +159,26 @@ async fn backfill_candidates(&self, room_id: &RoomId) -> Candidates { let (canonical_alias, power_levels) = join(canonical_alias, power_levels).await; + let state_member_servers = self + .services + .state_accessor + .room_state_keys(room_id, &StateEventType::RoomMember) + .filter_map(async |state_key| { + let Ok(state_key) = state_key else { + return None; + }; + + let Ok(user_id) = UserId::parse(state_key.as_str()) else { + return None; + }; + + (!self.services.globals.user_is_local(&user_id)) + .then_some(user_id.server_name().to_owned()) + }) + .collect::>() + .await; + let state_member_server_candidates = state_member_servers.iter().cloned().stream(); + let power_servers = power_levels .iter() .flat_map(|power| { @@ -184,21 +227,31 @@ async fn backfill_candidates(&self, room_id: &RoomId) -> Candidates { .map(ToOwned::to_owned) .stream(); - power_servers + state_member_server_candidates + .chain(power_servers) .chain(canonical_room_alias_server) .chain(trusted_servers) .ready_filter(|server_name| !self.services.globals.server_is_ours(server_name)) .filter_map(async |server_name| { - self.services + (self + .services .state_cache .server_in_room(&server_name, room_id) - .await - .then_some(server_name) + .await || state_member_servers.contains(&server_name)) + .then_some(server_name) }) - .collect() + .ready_fold(Candidates::new(), push_unique) .await } +fn push_unique(mut candidates: Candidates, server: OwnedServerName) -> Candidates { + if !candidates.contains(&server) { + candidates.push(server); + } + + candidates +} + #[implement(super::Service)] pub async fn get_event_id_near_ts_with_fallback( &self, @@ -398,12 +451,13 @@ pub async fn backfill_pdu( .index_pdu(shortroomid, &pdu_id, &body); } }, - | TimelineEventType::RoomTopic => + | TimelineEventType::RoomTopic => { if let Some(topic) = pdu.get_content().ok().and_then(plain_text_topic) { self.services .search .index_pdu(shortroomid, &pdu_id, &topic); - }, + } + }, | _ => {}, } @@ -434,3 +488,19 @@ fn prepend_backfill_pdu( txn.execute(); } + +#[cfg(test)] +mod tests { + use tuwunel_core::matrix::PduCount; + + use super::request_is_local_only; + + #[test] + fn local_coverage_check_includes_the_oldest_requested_event() { + let oldest = PduCount::Normal(10); + let newer = PduCount::Normal(20); + + assert!(request_is_local_only(oldest, newer, Some(oldest))); + assert!(!request_is_local_only(oldest, newer, Some(PduCount::Normal(9)))); + } +} diff --git a/src/service/rooms/timeline/build.rs b/src/service/rooms/timeline/build.rs index 01acfd1413..55ceaae5f6 100644 --- a/src/service/rooms/timeline/build.rs +++ b/src/service/rooms/timeline/build.rs @@ -2,16 +2,20 @@ use std::{collections::HashSet, iter::once}; use futures::{FutureExt, StreamExt}; use ruma::{ - OwnedEventId, OwnedServerName, RoomId, UserId, + CanonicalJsonObject, OwnedEventId, OwnedServerName, RoomId, UserId, events::{ - TimelineEventType, + StateEventType, TimelineEventType, room::member::{MembershipState, RoomMemberEventContent}, }, }; use serde_json::value::to_raw_value; use tuwunel_core::{ Err, Result, implement, - matrix::{event::Event, pdu::PduBuilder, room_version}, + matrix::{ + event::Event, + pdu::{PduBuilder, PduEvent}, + room_version, + }, utils::{IterStream, ReadyExt}, }; @@ -35,15 +39,46 @@ pub async fn build_and_append_pdu( state_lock: &RoomMutexGuard, ) -> Result { if pdu_builder.event_type == TimelineEventType::RoomMember { - self.sanitize_member_authorisation(&mut pdu_builder, room_id) + self.normalize_member_authorisation(&mut pdu_builder, room_id) .boxed() .await?; } - let (pdu, mut pdu_json) = self + let (pdu, pdu_json, _prev_state) = self .create_hash_and_sign_event(pdu_builder, sender, room_id, state_lock) .await?; + self.append_created_pdu(pdu, pdu_json, sender, state_lock) + .boxed() + .await +} + +/// Authorizes and persists a PDU already built by `create_hash_and_sign_event` +/// as the newest event in the room. Split out from `build_and_append_pdu` so +/// callers that need to inspect the built PDU before committing to persist it +/// (e.g. `/state`'s identical-resend short-circuit, which still needs the +/// auth_check inside `create_hash_and_sign_event` to have run) can reuse the +/// standard persistence path once they decide to. +#[implement(super::Service)] +#[tracing::instrument( + name = "append" + level = "debug", + skip(self, pdu_json, state_lock), + ret, +)] +pub async fn append_created_pdu( + &self, + pdu: PduEvent, + mut pdu_json: CanonicalJsonObject, + sender: &UserId, + state_lock: &RoomMutexGuard, +) -> Result { + let _shorteventid = self + .services + .short + .get_or_create_shorteventid(&pdu.event_id) + .await; + //TODO: Use proper room version here if *pdu.kind() == TimelineEventType::RoomCreate && pdu.room_id().server_name().is_none() { let _short_id = self @@ -151,7 +186,7 @@ pub async fn build_and_append_pdu( #[implement(super::Service)] #[tracing::instrument(skip_all, level = "debug")] -async fn sanitize_member_authorisation( +pub async fn normalize_member_authorisation( &self, pdu_builder: &mut PduBuilder, room_id: &RoomId, @@ -175,11 +210,16 @@ async fn sanitize_member_authorisation( .and_then(|key| UserId::parse(key).ok()) && self .services - .state_cache - .user_membership(&target, room_id) + .state_accessor + .room_state_get_content::( + room_id, + &StateEventType::RoomMember, + target.as_str(), + ) .await - .is_some_and(|m| matches!(m, MembershipState::Join | MembershipState::Invite)) - { + .is_ok_and(|event| { + matches!(event.membership, MembershipState::Join | MembershipState::Invite) + }) { let mut object = pdu_builder.content.deserialize()?; object.remove("join_authorised_via_users_server"); pdu_builder.content = to_raw_value(&object)?.into(); diff --git a/src/service/rooms/timeline/create.rs b/src/service/rooms/timeline/create.rs index 50d43f0c21..9a349b0473 100644 --- a/src/service/rooms/timeline/create.rs +++ b/src/service/rooms/timeline/create.rs @@ -25,6 +25,13 @@ use tuwunel_core::{ use super::RoomMutexGuard; use crate::rooms::state_res; +/// Builds, authorizes (via `state_res::auth_check`), hashes, and signs a PDU +/// without persisting it. The third return value is the state event this PDU +/// would replace, if any -- already fetched here to populate `unsigned`, so +/// callers that need it (e.g. an identical-resend short-circuit) can reuse it +/// instead of issuing another lookup. Because auth_check has already run by +/// the time this returns, a caller deciding to skip persistence never hands +/// out a success the sender wasn't currently authorized for. #[implement(super::Service)] pub async fn create_hash_and_sign_event( &self, @@ -33,7 +40,7 @@ pub async fn create_hash_and_sign_event( room_id: &RoomId, // Take mutex guard to make sure users get the room state mutex _mutex_lock: &RoomMutexGuard, -) -> Result<(PduEvent, CanonicalJsonObject)> { +) -> Result<(PduEvent, CanonicalJsonObject, Option)> { let PduBuilder { event_type, content, @@ -95,7 +102,7 @@ pub async fn create_hash_and_sign_event( .saturating_add(uint!(1)); let mut unsigned = unsigned.unwrap_or_default(); - if let Some(state_key) = &state_key + let prev_state = if let Some(state_key) = &state_key && let Ok(prev_pdu) = self .services .state_accessor @@ -105,7 +112,10 @@ pub async fn create_hash_and_sign_event( unsigned.insert("prev_content".to_owned(), prev_pdu.get_content_as_value()); unsigned.insert("prev_sender".to_owned(), serde_json::to_value(prev_pdu.sender())?); unsigned.insert("replaces_state".to_owned(), serde_json::to_value(prev_pdu.event_id())?); - } + Some(prev_pdu) + } else { + None + }; let unsigned = unsigned .is_empty() @@ -191,14 +201,7 @@ pub async fn create_hash_and_sign_event( check_rules(&pdu_json, &version_rules.event_format)?; - // Generate short event id - let _shorteventid = self - .services - .short - .get_or_create_shorteventid(&pdu.event_id) - .await; - - Ok((pdu, pdu_json)) + Ok((pdu, pdu_json, prev_state)) } #[implement(super::Service)] diff --git a/tests/complement/results.jsonl b/tests/complement/results.jsonl index 38acb3c2f8..215128c5c5 100644 --- a/tests/complement/results.jsonl +++ b/tests/complement/results.jsonl @@ -382,12 +382,12 @@ {"Action":"pass","Test":"TestMembersLocal/Parallel/Existing_members_see_new_members'_presence_(in_initial_sync)"} {"Action":"pass","Test":"TestMembersLocal/Parallel/New_room_members_see_their_own_join_event"} {"Action":"fail","Test":"TestMembershipOnEvents"} -{"Action":"fail","Test":"TestMessagesOverFederation"} +{"Action":"pass","Test":"TestMessagesOverFederation"} {"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_joining_new_room_(backfill)"} {"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_joining_new_room_(backfill)/`messagesRequestLimit`_is_greater_than_the_number_of_messages_backfilled_(in_Synapse,_100)"} {"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_joining_new_room_(backfill)/`messagesRequestLimit`_is_lower_than_the_number_of_messages_backfilled_(assumed)"} -{"Action":"fail","Test":"TestMessagesOverFederation/Visible_shared_history_after_re-joining_room_(backfill)"} -{"Action":"fail","Test":"TestMessagesOverFederation/Visible_shared_history_after_re-joining_room_(backfill)/`messagesRequestLimit`_is_lower_than_the_number_of_messages_backfilled_(assumed)"} +{"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_re-joining_room_(backfill)"} +{"Action":"pass","Test":"TestMessagesOverFederation/Visible_shared_history_after_re-joining_room_(backfill)/`messagesRequestLimit`_is_lower_than_the_number_of_messages_backfilled_(assumed)"} {"Action":"pass","Test":"TestNetworkPartitionOrdering"} {"Action":"pass","Test":"TestNotPresentUserCannotBanOthers"} {"Action":"pass","Test":"TestOlderLeftRoomsNotInLeaveSection"}