git-mirrors/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2025-09-09 14:33:03 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: Use handle_incoming_pdu directly to keep remote PDUs as outliers

Browse source
This commit is contained in:
Ginger 2025-09-02 15:12:03 -04:00 • committed by nex
commit f3824ffc3d
1 changed files with 16 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

23
src/service/rooms/timeline/backfill.rs
View file

@ -2,7 +2,7 @@ use std::iter::once;
use conduwuit::{Err, PduEvent}; use conduwuit::{Err, PduEvent};
use conduwuit_core::{ use conduwuit_core::{
Result, debug, debug_warn, implement, info, Result, debug, debug_warn, err, implement, info,
matrix::{ matrix::{
event::Event, event::Event,
pdu::{PduCount, PduId, RawPduId}, pdu::{PduCount, PduId, RawPduId},
@ -12,7 +12,7 @@ use conduwuit_core::{
}; };
use futures::{FutureExt, StreamExt}; use futures::{FutureExt, StreamExt};
use ruma::{ use ruma::{
EventId, RoomId, ServerName, CanonicalJsonObject, EventId, RoomId, ServerName,
api::federation, api::federation,
events::{ events::{
StateEventType, TimelineEventType, room::power_levels::RoomPowerLevelsEventContent, StateEventType, TimelineEventType, room::power_levels::RoomPowerLevelsEventContent,
@ -210,17 +210,26 @@ pub async fn get_remote_pdu(&self, room_id: &RoomId, event_id: &EventId) -> Resu
while let Some(ref backfill_server) = servers.next().await { while let Some(ref backfill_server) = servers.next().await {
info!("Asking {backfill_server} for event {}", event_id); info!("Asking {backfill_server} for event {}", event_id);
let response = self let value = self
.services .services
.sending .sending
.send_federation_request(backfill_server, federation::event::get_event::v1::Request { .send_federation_request(backfill_server, federation::event::get_event::v1::Request {
event_id: event_id.to_owned(), event_id: event_id.to_owned(),
include_unredacted_content: Some(false), include_unredacted_content: Some(false),
}) })
.await; .await
let pdu = match response { .and_then(|response| {
| Ok(response) => { serde_json::from_str::<CanonicalJsonObject>(response.pdu.get()).map_err(|e| {
self.backfill_pdu(backfill_server, response.pdu) err!(BadServerResponse(debug_warn!(
"Error parsing incoming event {e:?} from {backfill_server}"
)))
})
});
let pdu = match value {
| Ok(value) => {
self.services
.event_handler
.handle_incoming_pdu(backfill_server, &room_id, &event_id, value, false)
.boxed() .boxed()
.await?; .await?;
debug!("Successfully backfilled {event_id} from {backfill_server}"); debug!("Successfully backfilled {event_id} from {backfill_server}");