From 162e6eb92fef40614b5fd4ae101cefc1cbcd93e1 Mon Sep 17 00:00:00 2001 From: timedout Date: Thu, 28 May 2026 19:59:33 +0100 Subject: [PATCH] fix: Remove short-term memory loss I keep writing forgetful code, it's a problem --- .../rooms/event_handler/fetch_state.rs | 66 ++++++++++++------- .../event_handler/upgrade_outlier_pdu.rs | 8 ++- 2 files changed, 48 insertions(+), 26 deletions(-) diff --git a/src/service/rooms/event_handler/fetch_state.rs b/src/service/rooms/event_handler/fetch_state.rs index f4e19fe96..710ee6254 100644 --- a/src/service/rooms/event_handler/fetch_state.rs +++ b/src/service/rooms/event_handler/fetch_state.rs @@ -1,6 +1,8 @@ use std::collections::{HashMap, hash_map}; -use conduwuit::{Err, Event, PduEvent, Result, debug, debug_warn, err, utils::IterStream, warn}; +use conduwuit::{ + Err, Event, PduEvent, Result, debug, debug_warn, err, trace, utils::IterStream, warn, +}; use futures::StreamExt; use ruma::{ EventId, OwnedEventId, RoomId, ServerName, @@ -30,6 +32,7 @@ pub(super) async fn fetch_state( room_id: &RoomId, event_id: &EventId, ) -> Result>> { + trace!(%origin, "Asking remote for state_ids"); let res: get_room_state_ids::v1::Response = self .services .sending @@ -40,7 +43,7 @@ pub(super) async fn fetch_state( .await .inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?; - debug!("Fetching state events"); + debug!(events = res.pdu_ids.len(), "Fetching state events"); let mut state_events: HashMap = HashMap::with_capacity(res.pdu_ids.len()); let to_fetch: Vec = res @@ -57,32 +60,46 @@ pub(super) async fn fetch_state( }) .collect() .await; - if !to_fetch.is_empty() { - if to_fetch.len() >= 100 { - // That's a lot of events to fetch, just ask for the full state - // at that point. - debug_warn!( - to_fetch = to_fetch.len(), - "Fetching full state from remote server for event" - ); - state_events.extend( - self.fetch_full_state(origin, create_event, room_id, event_id) - .await?, - ); - } else { - debug!( - to_fetch = to_fetch.len(), - "Fetching missing events for state from remote" - ); - state_events.extend( - self.fetch_and_handle_missing_events(origin, to_fetch, create_event, room_id) - .await, - ); - } + if to_fetch.is_empty() { + debug!("All required state events are already known."); + state_events = res + .pdu_ids + .iter() + .stream() + .broad_filter_map(|event_id| async move { + Some(( + event_id.clone(), + self.services + .timeline + .get_pdu(event_id) + .await + .expect("Event disappeared between filtering and fetching"), + )) + }) + .collect() + .await; + } else if to_fetch.len() >= 100 { + // That's a lot of events to fetch, just ask for the full state + // at that point. + debug_warn!( + to_fetch = to_fetch.len(), + "Fetching full state from remote server for event" + ); + state_events.extend( + self.fetch_full_state(origin, create_event, room_id, event_id) + .await?, + ); + } else { + debug!(to_fetch = to_fetch.len(), "Fetching missing events for state from remote"); + state_events.extend( + self.fetch_and_handle_missing_events(origin, to_fetch, create_event, room_id) + .await, + ); } let mut state: HashMap = HashMap::with_capacity(state_events.len()); + debug!(events = state_events.len(), "Processing state events"); for (event_id, pdu) in state_events { let state_key = pdu.state_key().ok_or_else(|| { err!(Database("Found non-state pdu in state events: {event_id}")) @@ -136,6 +153,7 @@ pub(super) async fn fetch_full_state( room_id: &RoomId, event_id: &EventId, ) -> Result> { + trace!("Fetching full state from remote server"); let res: get_room_state::v1::Response = self .services .sending diff --git a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs index 253a6905e..89966b43b 100644 --- a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs +++ b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs @@ -1,8 +1,9 @@ use std::{borrow::Borrow, sync::Arc, time::Instant}; use conduwuit::{ - Err, Result, debug, debug_info, err, implement, info, is_equal_to, + Err, Result, debug, debug_error, debug_info, err, implement, info, is_equal_to, matrix::{Event, EventTypeExt, PduEvent, StateKey, state_res}, + result::DebugInspect, trace, utils::{ IterStream, @@ -83,11 +84,14 @@ pub(super) async fn upgrade_outlier_to_timeline_pdu( trace!("Could not calculate incoming state, asking remote {origin} for it"); state_at_incoming_event = self .fetch_state(origin, create_event, room_id, incoming_pdu.event_id()) - .await?; + .await + .debug_inspect_err(|e| debug_error!("Could not fetch state from {origin}: {e}"))?; } let state_at_incoming_event = state_at_incoming_event.expect("we always set this to some above"); + assert!(!state_at_incoming_event.is_empty(), "Event has no incoming state"); + trace!(state_events = state_at_incoming_event.len(), "Calculated incoming state"); debug!( event_id = %incoming_pdu.event_id,