From b2072baf5922967c7d354d7cb469c327381aed75 Mon Sep 17 00:00:00 2001 From: timedout Date: Mon, 21 Sep 2026 00:32:02 +0100 Subject: [PATCH] perf: Strip unsigned over federation and generate it on the fly locally --- src/api/client/context.rs | 24 ++++++- src/api/client/message.rs | 12 +++- src/api/client/room/event.rs | 7 +- src/api/client/room/initial_sync.rs | 12 +++- src/api/client/sync/mod.rs | 34 ++++++++-- src/api/client/sync/v3/joined.rs | 7 +- src/api/client/sync/v5.rs | 14 ++-- src/api/server/backfill.rs | 10 --- src/core/matrix/pdu/tests.rs | 2 +- src/core/matrix/pdu/unsigned.rs | 93 ++++++++++++++------------ src/service/rooms/pdu_metadata/data.rs | 9 ++- src/service/rooms/search/mod.rs | 16 ++++- src/service/rooms/threads/mod.rs | 12 +++- src/service/rooms/timeline/mod.rs | 90 ++++++++++++++++++++++++- 14 files changed, 263 insertions(+), 79 deletions(-) diff --git a/src/api/client/context.rs b/src/api/client/context.rs index e6064f496..28547df9e 100644 --- a/src/api/client/context.rs +++ b/src/api/client/context.rs @@ -98,7 +98,17 @@ pub(crate) async fn get_context_route( .pdus_rev(room_id, Some(base_count)) .ignore_err() .then(async |mut pdu| { - pdu.1.set_unsigned(Some(sender_user)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu.1, Some(sender_user)) + .await; + pdu.1.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); if let Err(e) = services .rooms .pdu_metadata @@ -121,7 +131,17 @@ pub(crate) async fn get_context_route( .pdus(room_id, Some(base_count)) .ignore_err() .then(async |mut pdu| { - pdu.1.set_unsigned(Some(sender_user)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu.1, Some(sender_user)) + .await; + pdu.1.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); if let Err(e) = services .rooms .pdu_metadata diff --git a/src/api/client/message.rs b/src/api/client/message.rs index 7f0628c2e..84d2d0269 100644 --- a/src/api/client/message.rs +++ b/src/api/client/message.rs @@ -140,7 +140,17 @@ pub(crate) async fn get_message_events_route( .wide_filter_map(|item| visibility_filter(&services, item, sender_user)) .take(limit) .then(async |mut pdu| { - pdu.1.set_unsigned(Some(sender_user)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu.1, Some(sender_user)) + .await; + pdu.1.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); if let Err(e) = services .rooms .pdu_metadata diff --git a/src/api/client/room/event.rs b/src/api/client/room/event.rs index af5098436..fe7d7f106 100644 --- a/src/api/client/room/event.rs +++ b/src/api/client/room/event.rs @@ -58,7 +58,12 @@ pub(crate) async fn get_room_event_route( debug_warn!("Failed to add bundled aggregations to event: {e}"); } - event.set_unsigned(Some(body.identity.expect_sender_user()?)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&event, Some(sender_user)) + .await; + event.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because); Ok(get_room_event::v3::Response::new(event.into_format())) } diff --git a/src/api/client/room/initial_sync.rs b/src/api/client/room/initial_sync.rs index 393557223..415d07399 100644 --- a/src/api/client/room/initial_sync.rs +++ b/src/api/client/room/initial_sync.rs @@ -53,7 +53,17 @@ pub(crate) async fn room_initial_sync_route( .pdus_rev(room_id, None) .try_take(limit) .and_then(async |mut pdu| { - pdu.1.set_unsigned(Some(sender_user)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu.1, Some(sender_user)) + .await; + pdu.1.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); if let Err(e) = services .rooms .pdu_metadata diff --git a/src/api/client/sync/mod.rs b/src/api/client/sync/mod.rs index 4110e24e0..0b1025de5 100644 --- a/src/api/client/sync/mod.rs +++ b/src/api/client/sync/mod.rs @@ -7,7 +7,7 @@ use conduwuit::{ Event, PduCount, Result, debug_warn, err, matrix::pdu::PduEvent, ref_at, trace, - utils::stream::{BroadbandExt, ReadyExt, TryIgnore}, + utils::stream::{BroadbandExt, ReadyExt, TryIgnore, WidebandExt}, }; use conduwuit_service::Services; use futures::StreamExt; @@ -83,9 +83,19 @@ async fn load_timeline( .pdus_rev(room_id, ending_count.map(|count| count.saturating_add(1))) .ignore_err() .ready_take_while(move |&(pducount, _)| pducount > starting_count) - .map(move |mut pdu| { - pdu.1.set_unsigned(Some(sender_user)); - pdu + .wide_filter_map(async |mut pdu| { + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu.1, Some(sender_user)) + .await; + pdu.1.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); + Some(pdu) }) .then(async move |mut pdu| { if let Err(e) = services @@ -108,9 +118,19 @@ async fn load_timeline( .timeline .pdus_rev(room_id, ending_count.map(|count| count.saturating_add(1))) .ignore_err() - .map(move |mut pdu| { - pdu.1.set_unsigned(Some(sender_user)); - pdu + .wide_filter_map(async |mut pdu| { + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu.1, Some(sender_user)) + .await; + pdu.1.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); + Some(pdu) }) .then(async move |mut pdu| { if let Err(e) = services diff --git a/src/api/client/sync/v3/joined.rs b/src/api/client/sync/v3/joined.rs index c04da2420..3fe1d3ee2 100644 --- a/src/api/client/sync/v3/joined.rs +++ b/src/api/client/sync/v3/joined.rs @@ -322,7 +322,12 @@ async fn build_sticky_events( continue; } - pdu.set_unsigned(Some(syncing_user)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu, Some(sync_context.syncing_user)) + .await; + pdu.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because); events.push(Event::into_format(pdu)); } diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index 458f26cc8..8411a81e5 100644 --- a/src/api/client/sync/v5.rs +++ b/src/api/client/sync/v5.rs @@ -1189,12 +1189,12 @@ async fn collect_sticky_events( // MSC4480 lets us ignore the limit for a newly visible room's backlog for (room_id, pdu) in backlog { - push_sticky_event(&mut response, sender_user, room_id, pdu); + push_sticky_event(services, &mut response, sender_user, room_id, pdu).await; } let mut next_batch = None; for (count, room_id, pdu) in stream.into_iter().take(limit) { - push_sticky_event(&mut response, sender_user, room_id, pdu); + push_sticky_event(services, &mut response, sender_user, room_id, pdu).await; next_batch = Some(count); } @@ -1205,13 +1205,19 @@ async fn collect_sticky_events( Ok(response) } -fn push_sticky_event( +async fn push_sticky_event( + services: &Services, response: &mut sync_events::v5::response::StickyEvents, sender_user: &UserId, room_id: OwnedRoomId, mut pdu: PduEvent, ) { - pdu.set_unsigned(Some(sender_user)); + let ctx = services + .rooms + .timeline + .get_unsigned_context(&pdu, Some(sender_user)) + .await; + pdu.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because); response .rooms .entry(room_id) diff --git a/src/api/server/backfill.rs b/src/api/server/backfill.rs index e44db5611..4fade6b54 100644 --- a/src/api/server/backfill.rs +++ b/src/api/server/backfill.rs @@ -3,7 +3,6 @@ use std::cmp; use axum::extract::State; use conduwuit::{ Err, Event, PduCount, Result, info, - result::LogErr, utils::{IterStream, ReadyExt, stream::TryTools}, }; use futures::{FutureExt, StreamExt, TryStreamExt}; @@ -80,15 +79,6 @@ pub(crate) async fn get_backfill_route( .await .then_some(pdu)) }) - .and_then(async |mut pdu| { - // Strip the transaction ID, as that is private - pdu.remove_transaction_id().log_err().ok(); - // Add age, as this is specified - pdu.add_age().log_err().ok(); - // It's not clear if we should strip or add any more data, leave as - // is. In particular: Redaction? - Ok(pdu) - }) .try_filter_map(|pdu| async move { Ok(services .rooms diff --git a/src/core/matrix/pdu/tests.rs b/src/core/matrix/pdu/tests.rs index d5af4f9d4..a535be89f 100644 --- a/src/core/matrix/pdu/tests.rs +++ b/src/core/matrix/pdu/tests.rs @@ -100,7 +100,7 @@ fn set_unsigned_adds_the_sticky_ttl() { // section let mut pdu = sticky_pdu(&serde_json::json!({ "duration_ms": 300_000 })); pdu.origin_server_ts = ruma::MilliSecondsSinceUnixEpoch::now().get(); - pdu.set_unsigned(None); + pdu.set_unsigned(None, None, None, None); let unsigned: serde_json::Value = serde_json::from_str(pdu.unsigned.as_deref().expect("unsigned is set").get()) diff --git a/src/core/matrix/pdu/unsigned.rs b/src/core/matrix/pdu/unsigned.rs index eff5a8247..47bd1f9ff 100644 --- a/src/core/matrix/pdu/unsigned.rs +++ b/src/core/matrix/pdu/unsigned.rs @@ -1,10 +1,16 @@ -use std::{borrow::Borrow, collections::BTreeMap}; +use std::collections::BTreeMap; -use ruma::MilliSecondsSinceUnixEpoch; +use ruma::{ + MilliSecondsSinceUnixEpoch, + events::{AnyTimelineEvent, room::member::MembershipState}, + serde::Raw, +}; use serde_json::value::{RawValue as RawJsonValue, Value as JsonValue, to_raw_value}; use super::{Pdu, sticky}; -use crate::{Result, err, result::LogErr}; +use crate::{Event, Result, err, result::LogErr}; + +type Unsigned = BTreeMap>; impl Pdu { /// Set the `unsigned` field of the PDU using only information in the PDU. @@ -12,51 +18,52 @@ impl Pdu { /// threads). Once this is done, other data must be calculated from the /// database (eg. relations) This is for server-to-client events. /// Backfill handles this itself. - pub fn set_unsigned(&mut self, user_id: Option<&ruma::UserId>) { - if Some(self.sender.borrow()) != user_id { - self.remove_transaction_id().log_err().ok(); - } - self.add_age().log_err().ok(); - self.add_sticky_duration_ttl().log_err().ok(); - } - - pub fn remove_transaction_id(&mut self) -> Result { - use BTreeMap as Map; - - let Some(unsigned) = &self.unsigned else { - return Ok(()); + pub fn set_unsigned( + &mut self, + user_id: Option<&ruma::UserId>, + membership: Option, + prev_content: Option>, + redacted_because: Option>, + ) { + // See: https://spec.matrix.org/v1.19/client-server-api/#definition-clientevent_unsigneddata + let Some(mut unsigned) = self + .unsigned() + .and_then(|u| serde_json::from_str::(u.get()).ok()) + else { + return; }; - let mut unsigned: Map<&str, Box> = serde_json::from_str(unsigned.get()) - .map_err(|e| err!(Database("Invalid unsigned in pdu event: {e}")))?; + let now: ruma::Int = MilliSecondsSinceUnixEpoch::now().get().into(); + unsigned.insert( + "age".to_owned(), + to_raw_value(&now.saturating_sub(self.origin_server_ts.into())).unwrap(), + ); - unsigned.remove("transaction_id"); - self.unsigned = to_raw_value(&unsigned) - .map(Some) - .expect("unsigned is valid"); + if let Some(membership) = membership { + unsigned.insert("membership".to_owned(), to_raw_value(&membership).unwrap()); + } - Ok(()) - } + if let Some(prev_content) = prev_content { + unsigned.insert( + "prev_content".to_owned(), + to_raw_value(&prev_content) + .expect("prev_content must be a valid JSON value for unsigned"), + ); + } + if let Some(redacted_because) = redacted_because { + unsigned.remove("redacted_because_id"); + unsigned.insert( + "redacted_because".to_owned(), + to_raw_value(&redacted_because) + .expect("redacted_because must be a valid JSON value for unsigned"), + ); + } - pub fn add_age(&mut self) -> Result { - use BTreeMap as Map; - - let mut unsigned: Map<&str, Box> = self - .unsigned - .as_deref() - .map(RawJsonValue::get) - .map_or_else(|| Ok(Map::new()), serde_json::from_str) - .map_err(|e| err!(Database("Invalid unsigned in pdu event: {e}")))?; - - // deliberately allowing for the possibility of negative age - let now: i128 = MilliSecondsSinceUnixEpoch::now().get().into(); - let then: i128 = self.origin_server_ts.into(); - let this_age = now.saturating_sub(then); - - unsigned.insert("age", to_raw_value(&this_age)?); - self.unsigned = Some(to_raw_value(&unsigned)?); - - Ok(()) + // Remove transaction_id unless the user is the sender + if user_id.is_none_or(|u| u != self.sender()) { + unsigned.remove("transaction_id"); + } + self.add_sticky_duration_ttl().log_err().ok(); } pub fn add_sticky_duration_ttl(&mut self) -> Result { diff --git a/src/service/rooms/pdu_metadata/data.rs b/src/service/rooms/pdu_metadata/data.rs index 1aee7e19f..18ef01f10 100644 --- a/src/service/rooms/pdu_metadata/data.rs +++ b/src/service/rooms/pdu_metadata/data.rs @@ -2,7 +2,7 @@ use std::{mem::size_of, sync::Arc}; use conduwuit::{ arrayvec::ArrayVec, - matrix::{Event, PduCount}, + matrix::PduCount, utils::{ ReadyExt, stream::{TryIgnore, WidebandExt}, @@ -94,7 +94,12 @@ impl Data { let mut pdu = self.services.timeline.get_pdu_from_id(&pdu_id).await.ok()?; - pdu.as_mut_pdu().set_unsigned(Some(user_id)); + let ctx = self + .services + .timeline + .get_unsigned_context(&pdu, Some(user_id)) + .await; + pdu.set_unsigned(ctx.user_id, ctx.membership, ctx.prev_content, ctx.redacted_because); Some((shorteventid, pdu)) }) diff --git a/src/service/rooms/search/mod.rs b/src/service/rooms/search/mod.rs index d155d34ad..24f7ca688 100644 --- a/src/service/rooms/search/mod.rs +++ b/src/service/rooms/search/mod.rs @@ -136,10 +136,20 @@ impl Service { }) .skip(query.skip) .take(query.limit) - .map(move |mut pdu| { - pdu.set_unsigned(query.user_id); + .wide_filter_map(async move |mut pdu| { + let ctx = self + .services + .timeline + .get_unsigned_context(&pdu, Some(sender_user)) + .await; + pdu.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); - pdu + Some(pdu) }) .then(async move |mut pdu| { if let Err(e) = self diff --git a/src/service/rooms/threads/mod.rs b/src/service/rooms/threads/mod.rs index 4809e38f0..6afd11b4d 100644 --- a/src/service/rooms/threads/mod.rs +++ b/src/service/rooms/threads/mod.rs @@ -164,7 +164,17 @@ impl Service { let mut pdu = self.services.timeline.get_pdu_from_id(&pdu_id).await.ok()?; let pdu_id: PduId = pdu_id.into(); - pdu.as_mut_pdu().set_unsigned(Some(user_id)); + let ctx = self + .services + .timeline + .get_unsigned_context(&pdu, Some(user_id)) + .await; + pdu.set_unsigned( + ctx.user_id, + ctx.membership, + ctx.prev_content, + ctx.redacted_because, + ); Some((pdu_id.shorteventid, pdu)) }); diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index 715d52855..d3d05488d 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/src/service/rooms/timeline/mod.rs @@ -21,8 +21,12 @@ use conduwuit_core::{ }; use futures::{Future, Stream, TryStreamExt, pin_mut}; use ruma::{ - CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, RoomId, - events::room::encrypted::Relation, + CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, RoomId, UserId, + events::{ + AnyTimelineEvent, + room::{encrypted::Relation, member::MembershipState}, + }, + serde::Raw, }; use serde::Deserialize; @@ -61,6 +65,14 @@ pub struct Service { pub mutex_insert: RoomMutexMap, } +#[derive(Default, Clone)] +pub struct UnsignedContext<'a> { + pub user_id: Option<&'a UserId>, + pub membership: Option, + pub prev_content: Option>, + pub redacted_because: Option>, +} + struct Services { server: Arc, account_data: Dep, @@ -272,4 +284,78 @@ impl Service { ) -> impl Stream> + Send + 'a { self.db.pdus(room_id, from.unwrap_or_else(PduCount::min)) } + + /// Fetches contextual unsigned information, like prev content, membership, + /// and redacted because. + pub async fn get_unsigned_context<'a, E>( + &self, + event: E, + user_id: Option<&'a UserId>, + ) -> UnsignedContext<'a> + where + E: Event, + { + let ssh = self + .services + .state_accessor + .pdu_shortstatehash(event.event_id()) + .await; + let (membership, prev_content) = if let Some(sender_user) = user_id + && let Ok(ssh) = ssh + { + let membership = Some( + self.services + .state_accessor + .user_membership(ssh, sender_user) + .await, + ); + let prev_content = match event.get_unsigned_property::("replaces_state") + { + | Ok(event_id) => { + // Only include prev content if the sender is allowed to see + // the event + let can_see = self + .services + .state_accessor + .user_can_see_event( + sender_user, + &event.room_id_or_hash(), + event.event_id(), + ) + .await; + if can_see { + self.get_pdu(&event_id).await.map(|p| p.content).ok() + } else { + None + } + }, + | _ => None, + }; + + (membership, prev_content) + } else { + (None, None) + }; + + // N.B. Theoretically, the redaction event should be included in a + // redacted capacity if the requesting user isn't allowed to see the + // redaction event. However, the cost of looking up that information + // is extraordinary compared to the value it would provide, so is + // deliberately not done. + let redacted_because = + match event.get_unsigned_property::("redacted_because_id") { + | Ok(event_id) => self + .get_pdu(&event_id) + .await + .map(Event::into_format::>) + .ok(), + | _ => None, + }; + UnsignedContext { + user_id, + membership, + prev_content, + redacted_because, + } + } }