From b51a34c1c1a148e9f05d1f87bc6c8c287bcfd826 Mon Sep 17 00:00:00 2001 From: timedout Date: Sat, 11 Jul 2026 04:30:14 +0100 Subject: [PATCH] fix: Address review feedback --- src/api/client/membership/invite.rs | 2 +- src/api/server/send.rs | 2 +- src/api/server/send_join.rs | 2 +- src/api/server/send_knock.rs | 2 +- src/api/server/send_leave.rs | 2 +- .../fetch_and_handle_outliers.rs | 26 ++- src/service/rooms/event_handler/fetch_auth.rs | 4 +- src/service/rooms/event_handler/fetch_prev.rs | 6 +- .../event_handler/handle_incoming_pdu.rs | 36 +++- .../rooms/event_handler/handle_outlier_pdu.rs | 13 +- src/service/rooms/event_handler/pdu_checks.rs | 8 +- .../event_handler/upgrade_outlier_pdu.rs | 156 +------------- src/service/rooms/timeline/append.rs | 195 +++++++++++++++--- src/service/rooms/timeline/backfill.rs | 4 +- src/service/rooms/timeline/mod.rs | 3 + 15 files changed, 250 insertions(+), 211 deletions(-) diff --git a/src/api/client/membership/invite.rs b/src/api/client/membership/invite.rs index 788f431ae..f4d310ea6 100644 --- a/src/api/client/membership/invite.rs +++ b/src/api/client/membership/invite.rs @@ -221,7 +221,7 @@ pub(crate) async fn invite_helper( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, true) + .handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, false) .boxed() .await? .ok_or_else(|| { diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 1615d513b..32bcfb3bb 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -327,7 +327,7 @@ async fn handle_room( let result = services .rooms .event_handler - .handle_incoming_pdu(origin, room_id, &event_id, value, true) + .handle_incoming_pdu(origin, room_id, &event_id, value, false) .boxed() .await .map(|_| ()); diff --git a/src/api/server/send_join.rs b/src/api/server/send_join.rs index 5498d03d3..b72c4c18a 100644 --- a/src/api/server/send_join.rs +++ b/src/api/server/send_join.rs @@ -200,7 +200,7 @@ async fn create_join_event( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), true) + .handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), false) .boxed() .await? .ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?; diff --git a/src/api/server/send_knock.rs b/src/api/server/send_knock.rs index 76e8f363f..377a8792d 100644 --- a/src/api/server/send_knock.rs +++ b/src/api/server/send_knock.rs @@ -166,7 +166,7 @@ pub(crate) async fn create_knock_event_v1_route( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), true) + .handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), false) .boxed() .await? .ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?; diff --git a/src/api/server/send_leave.rs b/src/api/server/send_leave.rs index 7b1ec9c03..121683224 100644 --- a/src/api/server/send_leave.rs +++ b/src/api/server/send_leave.rs @@ -153,7 +153,7 @@ async fn create_leave_event( let pdu_id = services .rooms .event_handler - .handle_incoming_pdu(origin, room_id, &event_id, value, true) + .handle_incoming_pdu(origin, room_id, &event_id, value, false) .boxed() .await? .ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?; diff --git a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs index c5a7fe1e2..a2855c750 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -1,5 +1,6 @@ use std::{ collections::{HashMap, HashSet, VecDeque}, + fmt::{Display, Formatter}, time::Instant, }; @@ -33,6 +34,20 @@ pub enum DagBuilderTree { AuthEvents, } +impl DagBuilderTree { + #[must_use] + pub fn as_str(&self) -> &str { + match self { + | Self::AuthEvents => "auth_events", + | Self::PrevEvents => "prev_events", + } + } +} + +impl Display for DagBuilderTree { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(self.as_str()) } +} + /// Attempts to build a localised directed acyclic graph out of the given PDUs, /// returning them in a topologically sorted order. /// @@ -45,20 +60,19 @@ pub async fn build_local_dag( pdu_map: &HashMap, tree: DagBuilderTree, ) -> Result> { - debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs"); + if pdu_map.len() <= 1 { + return Ok(pdu_map.keys().cloned().collect()); + } + let mut dag: HashMap> = HashMap::with_capacity(pdu_map.len()); let mut id_origin_ts: HashMap = HashMap::with_capacity(pdu_map.len()); - let tree = match tree { - | DagBuilderTree::AuthEvents => "auth_events", - | DagBuilderTree::PrevEvents => "prev_events", - }; for (event_id, value) in pdu_map { // Parse all prev events as event IDs - if they are missing, return an error (we // can't sanely continue in this case), otherwise skip invalid prev events. let prev_events = value - .get(tree) + .get(tree.as_str()) .and_then(CanonicalJsonValue::as_array) .ok_or_else(|| err!(Request(BadJson("event JSON for {event_id} is missing {tree}"))))? .iter() diff --git a/src/service/rooms/event_handler/fetch_auth.rs b/src/service/rooms/event_handler/fetch_auth.rs index ea99b4afb..48ff38a9c 100644 --- a/src/service/rooms/event_handler/fetch_auth.rs +++ b/src/service/rooms/event_handler/fetch_auth.rs @@ -56,9 +56,9 @@ impl super::Service { continue; }, }; - if !Self::pdu_format_check_1(&auth_pdu_json) { + if let Err(e) = Self::pdu_format_check_1(&auth_pdu_json) { // drop this PDU - warn!(%auth_event_id, "Dropping auth chain event as it is too large"); + warn!(%auth_event_id, error=?e, "Dropping auth chain event as it violates the room event format"); continue; } let auth_pdu_json = match self diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index 387d56031..cf85f10f4 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/src/service/rooms/event_handler/fetch_prev.rs @@ -84,11 +84,7 @@ impl super::Service { }) .collect::>(); - let to_persist = if mapped.len() <= 1 { - mapped.keys().map(ToOwned::to_owned).collect() - } else { - build_local_dag(&mapped, DagBuilderTree::PrevEvents).await? - }; + let to_persist = build_local_dag(&mapped, DagBuilderTree::PrevEvents).await?; let job_start = Instant::now(); trace!("Starting to persist {} prev events", to_persist.len()); diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index 1d3cb355d..aa4ce5d24 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs @@ -15,6 +15,26 @@ use crate::rooms::timeline::RawPduId; impl super::Service { /// Handles an incoming PDU from federation. + /// + /// First checks that we want to receive this PDU. If we already have it as + /// a timeline PDU, or we don't want to receive the PDU (e.g. origin ACL'd, + /// room disabled/unknown), abort. + /// + /// The PDU is then handled as an outlier event, which performs [PDU checks] + /// 1 through 4. See: `handle_outlier_pdu`. + /// + /// Once handled as an outlier, any missing prev events are fetched, and + /// then the PDU will be promoted/upgraded from an outlier to a timeline + /// event clients can see. See: `upgrade_outlier_to_timeline_pdu`. After + /// this finishes, the PDU is either accepted or left as an outlier. + /// + /// If the PDU is successfully upgraded, the remaining extremity count of + /// the room is checked. If there are a potentially problematic number of + /// forward extremities, a squasher task is started with a debounce period, + /// which will eventually send a dummy event that ties up as many DAG forks + /// as possible. + /// + /// [PDU checks]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu #[tracing::instrument( name = "pdu", skip_all, @@ -26,7 +46,7 @@ impl super::Service { room_id: &'a RoomId, event_id: &'a EventId, value: BTreeMap, - is_timeline_event: bool, + is_backfilled_event: bool, ) -> Result> { // Skip the PDU if we already have it as a timeline event. We still re-process // outliers in this scenario. @@ -65,7 +85,7 @@ impl super::Service { ))); } - if !room_exists && !is_interesting_member_event { + if !room_exists { if is_interesting_member_event { // TODO: handle interesting membership events where we aren't in // the room @@ -95,9 +115,10 @@ impl super::Service { .handle_outlier_pdu(origin, create_event, event_id, room_id, value) .await?; - // If this is not a timeline event, stop now, as we don't want to de-outlier it. - if !is_timeline_event { - debug!("Not promoting incoming event as it is not a timeline event"); + // If this event is being processed as part of backfill, we don't want to end up + // *appending* it during the upgrade process, so we return early. + if is_backfilled_event { + debug!("Not promoting incoming event as it is being backfilled"); return Ok(None); } @@ -110,7 +131,10 @@ impl super::Service { .await? .origin_server_ts(); if incoming_pdu.origin_server_ts() < first_ts_in_room { - debug!("Not promoting incoming event as it is sent before we joined the room"); + debug_warn!( + "Not promoting incoming event as it is sent before we joined the room (but was \ + not backfilled)" + ); return Ok(None); } diff --git a/src/service/rooms/event_handler/handle_outlier_pdu.rs b/src/service/rooms/event_handler/handle_outlier_pdu.rs index 3763f02d1..665c1ab7d 100644 --- a/src/service/rooms/event_handler/handle_outlier_pdu.rs +++ b/src/service/rooms/event_handler/handle_outlier_pdu.rs @@ -1,7 +1,7 @@ use std::collections::{BTreeMap, HashMap, hash_map}; use conduwuit::{ - Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace, warn, + Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace, }; use ruma::{CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName}; @@ -44,13 +44,12 @@ impl super::Service { // 1. Check that the PDU follows the format for the room version // (in this case, just size check) - if !Self::pdu_format_check_1(&value) { - warn!( - "Dropping incoming PDU {event_id} in room {room_id} from {origin} because it \ - exceeds 65535 bytes or is otherwise too large." + Self::pdu_format_check_1(&value).inspect_err(|e| { + info!( + err=?e, + "Dropping incoming PDU from {origin} because it violates the room event format" ); - return Err!(Request(TooLarge("PDU is too large"))); - } + })?; value.remove("unsigned"); let room_version_rules = get_room_version_rules(create_event)?; diff --git a/src/service/rooms/event_handler/pdu_checks.rs b/src/service/rooms/event_handler/pdu_checks.rs index 71ed6234c..26005c17e 100644 --- a/src/service/rooms/event_handler/pdu_checks.rs +++ b/src/service/rooms/event_handler/pdu_checks.rs @@ -16,10 +16,14 @@ impl super::Service { /// Checks that the PDU conforms to the PDU format (check 1). This is /// already mostly done during deserialisation, so this function just checks /// that the PDU isn't a too large. - pub(super) fn pdu_format_check_1(pdu_json: &CanonicalJsonObject) -> bool { + pub(super) fn pdu_format_check_1(pdu_json: &CanonicalJsonObject) -> Result<()> { // NOTE: if we do any more validation outside of deserialisation, it has to be // done here. - pdu_fits(pdu_json) + if !pdu_fits(pdu_json) { + Err!(Request(TooLarge("PDU is too large"))) + } else { + Ok(()) + } } /// Checks that the PDU has a valid signature (check 2), and redacts it if diff --git a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs index 3ee8ccb82..09aa9e41e 100644 --- a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs +++ b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs @@ -1,27 +1,15 @@ -use std::{borrow::Borrow, collections::HashMap, sync::Arc, time::Instant}; +use std::time::Instant; use conduwuit::{ - Err, Result, debug, debug_info, debug_warn, is_equal_to, is_true, + Err, Result, debug, debug_info, debug_warn, is_true, matrix::{Event, PduEvent}, trace, - utils::{ - IterStream, - mutex_map::Guard, - stream::{BroadbandExt, ReadyExt}, - }, -}; -use futures::{FutureExt, StreamExt}; -use ruma::{ - CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId, ServerName, events::StateEventType, - room_version_rules::RoomVersionRules, }; +use ruma::{CanonicalJsonObject, RoomId, ServerName, events::StateEventType}; use tokio::join; use super::get_room_version_rules; -use crate::rooms::{ - state_compressor::{CompressedState, HashSetCompressStateEvent}, - timeline::RawPduId, -}; +use crate::rooms::timeline::RawPduId; impl super::Service { #[tracing::instrument(name="upgrade_outlier", skip_all, fields(event_id=%incoming_pdu.event_id()))] @@ -170,55 +158,20 @@ impl super::Service { // The PDU has now passed all checks! We can now promote it (or soft-fail it if // the verdict is such). trace!("Appending pdu to timeline"); - let mut extremities: Vec<_> = self - .services - .state - .get_forward_extremities(room_id) - .collect() - .await; - if !should_soft_fail { - // Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events - // are not added as forward extremities. - // This also means we set the state here. - // We do this BEFORE setting the extremities so that there's never a point in - // time where we have fresh extremities referencing stale state. - extremities = self - .progress_state_and_extremities( - &incoming_pdu, - &room_version_rules, - state_before.clone(), - extremities, - &state_lock, - ) - .await?; - } - - let state_ids_compressed: Arc = self - .services - .state_compressor - .compress_state_events(state_before.iter().map(|(ssk, eid)| (ssk, eid.borrow()))) - .collect() - .map(Arc::new) - .await; let pdu_id = self .services .timeline .append_incoming_pdu( &incoming_pdu, val, - extremities.iter().map(Borrow::borrow), - state_ids_compressed, + &room_version_rules, + state_before, should_soft_fail, &state_lock, - room_id, ) .await?; if should_soft_fail { - self.services - .pdu_metadata - .mark_event_soft_failed(incoming_pdu.event_id()); - debug_info!( elapsed = ?timer.elapsed(), event_id = %incoming_pdu.event_id, @@ -242,101 +195,4 @@ impl super::Service { Ok(pdu_id) } - - /// Derives new room state from the incoming event and filters forward - /// extremities accordingly. Does not set forward extremities. - /// - /// Only call this function if the incoming PDU is not soft-failed or - /// rejected. - async fn progress_state_and_extremities( - &self, - incoming_pdu: &PduEvent, - room_version_rules: &RoomVersionRules, - state_before: HashMap, - forward_extremities: Vec, - state_lock: &Guard, - ) -> Result> { - if incoming_pdu.state_key().is_some() { - debug!("Event is a state-event. Deriving new room state"); - self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock) - .await?; - } - - // Now we calculate the set of extremities this room has after the incoming - // event has been applied. We start with the previous extremities - trace!("Calculating extremities"); - let mut forward_extremities = forward_extremities - .into_iter() - .stream() - .ready_filter(|event_id| { - // Remove any that are referenced by this incoming event's prev_events - !incoming_pdu.prev_events().any(is_equal_to!(event_id)) - }) - .broad_filter_map(|event_id| async move { - // Only keep those extremities were not referenced yet - self.services - .pdu_metadata - .is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id) - .await - .eq(&false) - .then_some(event_id) - }) - .collect::>() - .await; - forward_extremities.push(incoming_pdu.event_id().to_owned()); - debug!( - "Retained {} extremities checked against {} prev_events", - forward_extremities.len(), - incoming_pdu.prev_events().count() - ); - assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty"); - Ok(forward_extremities) - } - - /// Derives a new room state by adding the incoming PDU to the state before - /// it to create the state at, which then becomes the current room state. - /// - /// The caller MUST ensure forward extremities are set appropriately, - /// including this incoming pdu, either before or after calling this - /// function. Failing to do so will result in an inconsistent current state - /// cache, which may affect event authentication. - async fn derive_new_state( - &self, - incoming_pdu: &PduEvent, - room_version_rules: &RoomVersionRules, - state_before: HashMap, - state_lock: &Guard, - ) -> Result { - let room_id = incoming_pdu.room_id_or_hash(); - // We also add state after incoming event to the fork states - let mut state_at_incoming_event = state_before; - let shortstatekey = self - .services - .short - .get_or_create_shortstatekey( - &incoming_pdu.kind().to_string().into(), - incoming_pdu.state_key().unwrap(), - ) - .await; - - let event_id = incoming_pdu.event_id(); - state_at_incoming_event.insert(shortstatekey, event_id.to_owned()); - - debug!("Resolving new room state"); - let new_room_state = self - .resolve_state(&room_id, room_version_rules, state_at_incoming_event) - .await?; - - debug!("Forcing new room state"); - let HashSetCompressStateEvent { shortstatehash, added, removed } = self - .services - .state_compressor - .save_state(&room_id, new_room_state) - .await?; - - self.services - .state - .force_state(&room_id, shortstatehash, added, removed, state_lock) - .await - } } diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 3d7dcbcc1..c1959c94d 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -1,69 +1,113 @@ use std::{ - collections::{BTreeMap, HashSet}, + borrow::Borrow, + collections::{BTreeMap, HashMap, HashSet}, sync::Arc, }; use conduwuit::{ - debug_warn, + Result, debug, debug_warn, is_equal_to, + matrix::{Event, PduEvent}, pdu::{Count, ShortRoomId}, trace, - utils::{IterStream, TryFutureExtExt, stream::BroadbandExt}, + utils::{ + IterStream, TryFutureExtExt, + mutex_map::Guard, + stream::{BroadbandExt, ReadyExt}, + }, warn, }; use conduwuit_core::{ - Result, err, error, - matrix::{ - event::Event, - pdu::{PduCount, PduEvent, PduId, RawPduId}, - }, - utils::{self, ReadyExt}, + err, error, + matrix::pdu::{PduCount, PduId, RawPduId}, + utils::{self}, }; -use futures::{StreamExt, TryFutureExt}; +use futures::{FutureExt, StreamExt, TryFutureExt}; use ruma::{ - CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, UserId, + CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId, + RoomVersionId, UserId, events::{ GlobalAccountDataEventType, TimelineEventType, push_rules::PushRulesEvent, room::{encrypted::Relation, redaction::RoomRedactionEventContent}, }, push::{Action, Ruleset, Tweak}, + room_version_rules::RoomVersionRules, }; use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard}; -use crate::{appservice::RegistrationInfo, rooms::state_compressor::CompressedState}; +use crate::{ + appservice::RegistrationInfo, + rooms::state_compressor::{CompressedState, HashSetCompressStateEvent}, +}; impl super::Service { /// Append the incoming event setting the state snapshot to the state from /// the server that sent the event. - #[allow(clippy::too_many_arguments)] - pub async fn append_incoming_pdu<'a, Leaves>( + pub async fn append_incoming_pdu<'a>( &'a self, pdu: &'a PduEvent, pdu_json: CanonicalJsonObject, - new_room_leaves: Leaves, - state_ids_compressed: Arc, - soft_fail: bool, + room_version_rules: &RoomVersionRules, + state_before: HashMap, + should_soft_fail: bool, state_lock: &'a RoomMutexGuard, - room_id: &'a RoomId, - ) -> Result> - where - Leaves: Iterator + Send + 'a, - { + ) -> Result> { + let room_id = pdu.room_id_or_hash(); + + let sb = state_before.iter().map(|(ssk, eid)| (ssk, eid.as_ref())); + let state_ids_compressed: Arc = self + .services + .state_compressor + .compress_state_events(sb) + .collect() + .map(Arc::new) + .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. self.services .state - .set_event_state(&pdu.event_id, room_id, state_ids_compressed) + .set_event_state(&pdu.event_id, &room_id, state_ids_compressed) .await?; - if soft_fail { + if should_soft_fail { // Nothing else to do with a soft-failed event. + self.services + .pdu_metadata + .mark_event_soft_failed(pdu.event_id()); return Ok(None); } + // Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events + // are not added as forward extremities. + // This also means we set the state here. + // We do this BEFORE setting the extremities so that there's never a point in + // time where we have fresh extremities referencing stale state. + let mut extremities: Vec<_> = self + .services + .state + .get_forward_extremities(&room_id) + .collect() + .await; + extremities = self + .progress_state_and_extremities( + pdu, + room_version_rules, + state_before.clone(), + extremities, + state_lock, + ) + .await?; + let pdu_id = self - .append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id) + .append_pdu( + pdu, + pdu_json, + extremities.iter().map(Borrow::borrow), + state_lock, + &room_id, + ) .await?; // Process admin commands for federation events @@ -86,11 +130,110 @@ impl super::Service { } } - self.services.sync.wake_all_joined(room_id).await; + self.services.sync.wake_all_joined(&room_id).await; Ok(Some(pdu_id)) } + /// Derives new room state from the incoming event and filters forward + /// extremities accordingly. Does not set forward extremities. + /// + /// Only call this function if the incoming PDU is not soft-failed or + /// rejected. + async fn progress_state_and_extremities( + &self, + incoming_pdu: &PduEvent, + room_version_rules: &RoomVersionRules, + state_before: HashMap, + forward_extremities: Vec, + state_lock: &Guard, + ) -> Result> { + if incoming_pdu.state_key().is_some() { + debug!("Event is a state-event. Deriving new room state"); + self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock) + .await?; + } + + // Now we calculate the set of extremities this room has after the incoming + // event has been applied. We start with the previous extremities + trace!("Calculating extremities"); + let mut forward_extremities = forward_extremities + .into_iter() + .stream() + .ready_filter(|event_id| { + // Remove any that are referenced by this incoming event's prev_events + !incoming_pdu.prev_events().any(is_equal_to!(event_id)) + }) + .broad_filter_map(|event_id| async move { + // Only keep those extremities were not referenced yet + self.services + .pdu_metadata + .is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id) + .await + .eq(&false) + .then_some(event_id) + }) + .collect::>() + .await; + forward_extremities.push(incoming_pdu.event_id().to_owned()); + debug!( + "Retained {} extremities checked against {} prev_events", + forward_extremities.len(), + incoming_pdu.prev_events().count() + ); + assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty"); + Ok(forward_extremities) + } + + /// Derives a new room state by adding the incoming PDU to the state before + /// it to create the state at, which then becomes the current room state. + /// + /// The caller MUST ensure forward extremities are set appropriately, + /// including this incoming pdu, either before or after calling this + /// function. Failing to do so will result in an inconsistent current state + /// cache, which may affect event authentication. + async fn derive_new_state( + &self, + incoming_pdu: &PduEvent, + room_version_rules: &RoomVersionRules, + state_before: HashMap, + state_lock: &Guard, + ) -> Result { + let room_id = incoming_pdu.room_id_or_hash(); + // We also add state after incoming event to the fork states + let mut state_at_incoming_event = state_before; + let shortstatekey = self + .services + .short + .get_or_create_shortstatekey( + &incoming_pdu.kind().to_string().into(), + incoming_pdu.state_key().unwrap(), + ) + .await; + + let event_id = incoming_pdu.event_id(); + state_at_incoming_event.insert(shortstatekey, event_id.to_owned()); + + debug!("Resolving new room state"); + let new_room_state = self + .services + .event_handler + .resolve_state(&room_id, room_version_rules, state_at_incoming_event) + .await?; + + debug!("Forcing new room state"); + let HashSetCompressStateEvent { shortstatehash, added, removed } = self + .services + .state_compressor + .save_state(&room_id, new_room_state) + .await?; + + self.services + .state + .force_state(&room_id, shortstatehash, added, removed, state_lock) + .await + } + /// Populates the unsigned data of a PDU async fn populate_unsigned(&self, pdu: &PduEvent, pdu_json: &mut CanonicalJsonObject) { let Some(state_key) = pdu.state_key() else { diff --git a/src/service/rooms/timeline/backfill.rs b/src/service/rooms/timeline/backfill.rs index 36b01c8e8..e40cf7c5e 100644 --- a/src/service/rooms/timeline/backfill.rs +++ b/src/service/rooms/timeline/backfill.rs @@ -143,7 +143,7 @@ impl super::Service { | Ok(value) => { self.services .event_handler - .handle_incoming_pdu(&backfill_server, room_id, event_id, value, false) + .handle_incoming_pdu(&backfill_server, room_id, event_id, value, true) .boxed() .await?; debug!("Successfully backfilled {event_id} from {backfill_server}"); @@ -185,7 +185,7 @@ impl super::Service { self.services .event_handler - .handle_incoming_pdu(origin, &room_id, &event_id, value, false) + .handle_incoming_pdu(origin, &room_id, &event_id, value, true) .boxed() .await?; diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index c96707b43..a3d0fbd7c 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/src/service/rooms/timeline/mod.rs @@ -77,6 +77,7 @@ struct Services { short: Dep, state: Dep, state_accessor: Dep, + state_compressor: Dep, state_cache: Dep, sync: Dep, threads: Dep, @@ -110,6 +111,8 @@ impl crate::Service for Service { state: args.depend::("rooms::state"), state_accessor: args .depend::("rooms::state_accessor"), + state_compressor: args + .depend::("rooms::state_compressor"), state_cache: args.depend::("rooms::state_cache"), sync: args.depend::("sync"), threads: args.depend::("rooms::threads"),