From fbf81fcdebaf39a677ba6caf601c4f8c1d368336 Mon Sep 17 00:00:00 2001 From: timedout Date: Sat, 11 Jul 2026 20:14:46 +0100 Subject: [PATCH] fix: Opportunistically re-use room format rules when parsing incoming PDUs --- src/admin/debug/commands.rs | 4 +- src/api/server/invite.rs | 3 +- src/api/server/send.rs | 2 +- src/api/server/utils.rs | 8 ++-- .../fetch_and_handle_outliers.rs | 7 ++-- src/service/rooms/event_handler/fetch_auth.rs | 18 ++++---- .../rooms/event_handler/fetch_state.rs | 4 +- .../rooms/event_handler/parse_incoming_pdu.rs | 42 ++++++++----------- src/service/rooms/timeline/backfill.rs | 7 +++- 9 files changed, 48 insertions(+), 47 deletions(-) diff --git a/src/admin/debug/commands.rs b/src/admin/debug/commands.rs index e1102d4ed..57a9095cb 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -486,7 +486,7 @@ impl crate::Context<'_> { .services .rooms .event_handler - .parse_incoming_pdu(&response.pdu) + .parse_incoming_pdu(&response.pdu, None) .boxed() .await; @@ -832,7 +832,7 @@ impl crate::Context<'_> { .services .rooms .event_handler - .parse_incoming_pdu(&pdu) + .parse_incoming_pdu(&pdu, Some(&room_version_rules)) .await { | Ok(t) => t, diff --git a/src/api/server/invite.rs b/src/api/server/invite.rs index 490f0fdf0..392bfa7a9 100644 --- a/src/api/server/invite.rs +++ b/src/api/server/invite.rs @@ -250,6 +250,7 @@ async fn validate_invite_membership_event( room_id: OwnedRoomId, event_id: OwnedEventId, ) -> Result<(CanonicalJsonObject, OwnedUserId, OwnedUserId)> { + trace!(?body, "Invite membership event"); let (pdu, target_membership, sender_user, recipient_user) = validate_any_membership_event( services, body, @@ -309,7 +310,7 @@ async fn validate_invite_state( let (state_event_room_id, state_event_id, state_event_json) = services .rooms .event_handler - .parse_incoming_pdu(&raw_pdu) + .parse_incoming_pdu(&raw_pdu, Some(room_version_rules)) .await .map_err(|e| { err!(Request(InvalidParam(debug_warn!("Invalid PDU in invite state: {e}")))) diff --git a/src/api/server/send.rs b/src/api/server/send.rs index cb3a8eeea..bc1dbf32a 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -150,7 +150,7 @@ async fn process_inbound_transaction( .pdus .iter() .stream() - .broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu)) + .broad_then(|pdu| services.rooms.event_handler.parse_incoming_pdu(pdu, None)) .inspect_err(|e| warn!("Could not parse incoming PDU: {e}")) .ready_filter_map(Result::ok); diff --git a/src/api/server/utils.rs b/src/api/server/utils.rs index b420c59a9..774675de7 100644 --- a/src/api/server/utils.rs +++ b/src/api/server/utils.rs @@ -94,7 +94,7 @@ pub(crate) async fn validate_any_membership_event( let (template_room_id, template_event_id, pdu) = services .rooms .event_handler - .parse_incoming_pdu(body) + .parse_incoming_pdu(body, Some(room_version_rules)) .await .map_err(|e| err!(Request(BadJson("Invalid membership PDU: {e}"))))?; @@ -102,9 +102,11 @@ pub(crate) async fn validate_any_membership_event( return Err!(Request(InvalidParam("Membership event does not belong to requested room"))); } if template_event_id != expected_event_id { - return Err!(Request(InvalidParam( + return Err!(Request(InvalidParam(debug_warn!( + %template_event_id, + %expected_event_id, "Membership event ID does not match provided event ID" - ))); + )))); } services 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 67c5a4a16..f8c56acf2 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -241,7 +241,7 @@ impl super::Service { latest_events.clear(); for raw_event in response.events { - let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event).await?; + let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event, None).await?; let pdu = PduEvent::from_id_val(&event_id, pdu_json).map_err(|e| { err!(Request(BadJson("Failed to parse gapfilled event {event_id}: {e}"))) })?; @@ -384,8 +384,9 @@ impl super::Service { .send_federation_request(&remote, get_event::v1::Request::new(event_id.clone())) .await?; - let (calculated_event_id, value) = - Self::parse_incoming_pdu_with_known_room(&res.pdu, room_version_rules)?; + let (_, calculated_event_id, value) = self + .parse_incoming_pdu(&res.pdu, Some(room_version_rules)) + .await?; if calculated_event_id != event_id { Err!(Request(BadJson(warn!( diff --git a/src/service/rooms/event_handler/fetch_auth.rs b/src/service/rooms/event_handler/fetch_auth.rs index cf4d95d49..6ee77548c 100644 --- a/src/service/rooms/event_handler/fetch_auth.rs +++ b/src/service/rooms/event_handler/fetch_auth.rs @@ -48,14 +48,16 @@ impl super::Service { let mut auth_chain_map = HashMap::with_capacity(event_auth.auth_chain.len()); for auth_pdu_json in event_auth.auth_chain { - let (auth_event_room_id, auth_event_id, auth_pdu_json) = - match self.parse_incoming_pdu(&auth_pdu_json).await { - | Ok(parsed) => parsed, - | Err(e) => { - warn!(error=?e, "Dropping auth chain event as it could not be parsed"); - continue; - }, - }; + let (auth_event_room_id, auth_event_id, auth_pdu_json) = match self + .parse_incoming_pdu(&auth_pdu_json, Some(room_version_rules)) + .await + { + | Ok(parsed) => parsed, + | Err(e) => { + warn!(error=?e, "Dropping auth chain event as it could not be parsed"); + continue; + }, + }; if let Err(e) = Self::pdu_format_check_1( &auth_pdu_json, room_version_rules, diff --git a/src/service/rooms/event_handler/fetch_state.rs b/src/service/rooms/event_handler/fetch_state.rs index dbd0b75d6..9606e50ef 100644 --- a/src/service/rooms/event_handler/fetch_state.rs +++ b/src/service/rooms/event_handler/fetch_state.rs @@ -314,7 +314,7 @@ impl super::Service { .iter() .stream() .broad_filter_map(|raw_event_json| async { - if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok() + if let Some(parsed) = self.parse_incoming_pdu(raw_event_json, None).await.ok() && parsed.0 == room_id { Some(parsed) @@ -351,7 +351,7 @@ impl super::Service { .iter() .stream() .broad_filter_map(|raw_event_json| async { - if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok() + if let Some(parsed) = self.parse_incoming_pdu(raw_event_json, None).await.ok() && parsed.0 == room_id { Some(parsed) diff --git a/src/service/rooms/event_handler/parse_incoming_pdu.rs b/src/service/rooms/event_handler/parse_incoming_pdu.rs index 66122bf96..e01667e80 100644 --- a/src/service/rooms/event_handler/parse_incoming_pdu.rs +++ b/src/service/rooms/event_handler/parse_incoming_pdu.rs @@ -82,25 +82,14 @@ pub(super) fn expect_event_id_array( } impl super::Service { - /// Parses an incoming PDU JSON object, generating an event ID for it. Does - /// not insert the event ID into the returned object. Does not discover the - /// room ID. - pub(super) fn parse_incoming_pdu_with_known_room( - pdu: &RawJsonValue, - room_version_rules: &RoomVersionRules, - ) -> Result<(OwnedEventId, CanonicalJsonObject)> { - let (event_id, value) = - gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| { - err!(Request(InvalidParam("Could not convert event to canonical json: {e}"))) - })?; - // NOTE: validation checks are now performed by `pdu_format_check_1`. - Ok((event_id, value)) - } - /// Parses an incoming PDU JSON object, generating an event ID for it and /// attempts to discover the associated room ID. Does not insert the event /// ID into the returned object. - pub async fn parse_incoming_pdu(&self, pdu: &RawJsonValue) -> Result { + pub async fn parse_incoming_pdu( + &self, + pdu: &RawJsonValue, + room_version_rules: Option<&RoomVersionRules>, + ) -> Result { let value = serde_json::from_str::(pdu.get()).map_err(|e| { err!(BadServerResponse(debug_warn!("Error parsing incoming event {e:?}"))) })?; @@ -111,17 +100,20 @@ impl super::Service { let room_id = extract_room_id(event_type, &value)?; - let room_version_rules = self - .services - .state - .get_room_version(&room_id) - .await - .unwrap_or(RoomVersionId::V1) - .rules() - .unwrap(); + let room_version_rules = match room_version_rules { + | Some(r) => r, + | None => &self + .services + .state + .get_room_version(&room_id) + .await + .unwrap_or(RoomVersionId::V1) + .rules() + .unwrap(), + }; let (event_id, value) = - gen_event_id_canonical_json(pdu, &room_version_rules).map_err(|e| { + gen_event_id_canonical_json(pdu, room_version_rules).map_err(|e| { err!(Request(InvalidParam("Could not convert event to canonical json: {e}"))) })?; // NOTE: validation checks are now performed by `pdu_format_check_1`. diff --git a/src/service/rooms/timeline/backfill.rs b/src/service/rooms/timeline/backfill.rs index e40cf7c5e..36b854a67 100644 --- a/src/service/rooms/timeline/backfill.rs +++ b/src/service/rooms/timeline/backfill.rs @@ -166,8 +166,11 @@ impl super::Service { /// Backfills a single PDU. #[tracing::instrument(skip(self, pdu), level = "debug")] pub async fn backfill_pdu(&self, origin: &ServerName, pdu: Box) -> Result<()> { - let (room_id, event_id, value) = - self.services.event_handler.parse_incoming_pdu(&pdu).await?; + let (room_id, event_id, value) = self + .services + .event_handler + .parse_incoming_pdu(&pdu, None) + .await?; // Lock so we cannot backfill the same pdu twice at the same time let mutex_lock = self