From 4b673692f6fc9dfffc9214f29098d1c9cc48c40e Mon Sep 17 00:00:00 2001 From: timedout Date: Wed, 27 May 2026 00:43:37 +0100 Subject: [PATCH] fix: Be smarter when re-receiving already-seen PDUs --- src/service/rooms/event_handler/fetch_prev.rs | 4 +- .../event_handler/handle_incoming_pdu.rs | 3 +- .../rooms/event_handler/handle_outlier_pdu.rs | 54 +++++++++++-------- 3 files changed, 36 insertions(+), 25 deletions(-) diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index f25b90085..25fdf9554 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/src/service/rooms/event_handler/fetch_prev.rs @@ -2,11 +2,11 @@ use conduwuit::{ Event, PduEvent, debug, debug_info, - utils::{BoolExt, IterStream, math::try_into, stream::BroadbandExt}, + utils::{BoolExt, IterStream, stream::BroadbandExt}, warn, }; use futures::StreamExt; -use ruma::{RoomId, ServerName, UInt}; +use ruma::{RoomId, ServerName}; use crate::rooms::event_handler::build_local_dag; diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index 216a989d4..163779814 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs @@ -1,8 +1,7 @@ use std::{collections::BTreeMap, time::Instant}; use conduwuit::{ - Err, Event, PduEvent, Result, debug::INFO_SPAN_LEVEL, debug_error, debug_info, defer, err, - info, trace, warn, + Err, Event, PduEvent, Result, debug_error, debug_info, defer, err, info, trace, warn, }; use futures::{ FutureExt, diff --git a/src/service/rooms/event_handler/handle_outlier_pdu.rs b/src/service/rooms/event_handler/handle_outlier_pdu.rs index 5431b490c..c4d912cd9 100644 --- a/src/service/rooms/event_handler/handle_outlier_pdu.rs +++ b/src/service/rooms/event_handler/handle_outlier_pdu.rs @@ -7,7 +7,8 @@ use futures::future::ready; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName, - api::federation::authorization::get_event_authorization, events::StateEventType, + api::federation::authorization::get_event_authorization, canonical_json::redact, + events::StateEventType, }; use super::{check_room_id, get_room_version_rules}; @@ -48,27 +49,46 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( .verify_event(&value, &room_version_rules) .await { - | Ok(ruma::signatures::Verified::All) => value, + | Ok(ruma::signatures::Verified::All) => { + if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await { + debug!( + "Already have event {event_id} as an outlier or timeline event, not \ + re-processing" + ); + value.insert( + "event_id".to_owned(), + CanonicalJsonValue::String(event_id.as_str().to_owned()), + ); + check_room_id(room_id, &pdu_event)?; + return Ok((pdu_event, value)); + } + value + }, | Ok(ruma::signatures::Verified::Signatures) => { - // Redact debug_info!("Calculated hash does not match (redaction): {event_id}"); - let Ok(obj) = - ruma::canonical_json::redact(value, &room_version_rules.redaction, None) - else { - return Err!(Request(InvalidParam("Redaction failed"))); + let mut obj = match redact(value, &room_version_rules.redaction, None) { + | Ok(obj) => obj, + | Err(e) => + return Err!(Request(BadJson("Failed to redact {event_id}: {e}"))), }; - // Skip the PDU if it is redacted and we already have it as an outlier event - if self.services.timeline.pdu_exists(event_id).await { - return Err!(Request(InvalidParam( - "Event was redacted and we already knew about it" - ))); + if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await { + debug!( + "Received a redacted copy of {event_id}, but we already knew about it. \ + Re-using known content instead." + ); + obj.insert( + "event_id".to_owned(), + CanonicalJsonValue::String(event_id.as_str().to_owned()), + ); + check_room_id(room_id, &pdu_event)?; + return Ok((pdu_event, obj)); } obj }, | Err(e) => { - return Err!(Request(InvalidParam(debug_error!( + return Err!(Request(Forbidden(debug_error!( "Signature verification failed for {event_id}: {e}" )))); }, @@ -81,14 +101,6 @@ pub(super) async fn handle_outlier_pdu<'a, Pdu>( CanonicalJsonValue::String(event_id.as_str().to_owned()), ); - if let Ok(pdu_event) = self.services.timeline.get_pdu(event_id).await { - debug!( - "Already have event {event_id} as an outlier or timeline event, not \ - re-processing" - ); - return Ok((pdu_event, incoming_pdu)); - } - let pdu_event = serde_json::from_value::( serde_json::to_value(&incoming_pdu).expect("CanonicalJsonObj is a valid JsonValue"), )