diff --git a/src/database/maps.rs b/src/database/maps.rs index d1092c176..013fc9270 100644 --- a/src/database/maps.rs +++ b/src/database/maps.rs @@ -171,7 +171,7 @@ pub(super) static MAPS: &[Descriptor] = &[ }, Descriptor { name: "referencedevents", - ..descriptor::DROPPED + ..descriptor::RANDOM }, Descriptor { name: "refreshtoken_refreshtokeninfo", diff --git a/src/service/rooms/pdu_metadata/data.rs b/src/service/rooms/pdu_metadata/data.rs index 2f5cc38f7..2b40f6e52 100644 --- a/src/service/rooms/pdu_metadata/data.rs +++ b/src/service/rooms/pdu_metadata/data.rs @@ -11,7 +11,7 @@ use conduwuit::{ }; use database::Map; use futures::{Stream, StreamExt}; -use ruma::{EventId, api::Direction}; +use ruma::{EventId, RoomId, api::Direction}; use crate::{ Dep, @@ -26,6 +26,7 @@ pub(super) struct Data { tofrom_relation: Arc, softfailedeventids: Arc, rejectedeventids: Arc, + referencedevents: Arc, services: Services, } @@ -40,6 +41,7 @@ impl Data { tofrom_relation: db["tofrom_relation"].clone(), softfailedeventids: db["softfailedeventids"].clone(), rejectedeventids: db["rejectedeventids"].clone(), + referencedevents: db["referencedevents"].clone(), services: Services { timeline: args.depend::("rooms::timeline"), }, @@ -132,4 +134,16 @@ impl Data { self.unmark_event_rejected(event_id); self.unmark_event_soft_failed(event_id); } + + #[inline] + pub(super) fn mark_as_referenced(&self, room_id: &RoomId, event_id: &EventId) { + let key = (room_id, event_id); + self.referencedevents.put_raw(key, []); + } + + #[inline] + pub(super) async fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> bool { + let key = (room_id, event_id); + self.referencedevents.qry(&key).await.is_ok() + } } diff --git a/src/service/rooms/pdu_metadata/mod.rs b/src/service/rooms/pdu_metadata/mod.rs index 6a0244cb1..0ff50402c 100644 --- a/src/service/rooms/pdu_metadata/mod.rs +++ b/src/service/rooms/pdu_metadata/mod.rs @@ -207,6 +207,14 @@ impl Service { self.db.is_event_rejected(event_id).await } + pub async fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> bool { + self.db.is_event_referenced(room_id, event_id).await + } + + pub fn mark_as_referenced(&self, room_id: &RoomId, event_id: &EventId) { + self.db.mark_as_referenced(room_id, event_id); + } + pub fn mark_event_rejected(&self, event_id: &EventId) { self.db.mark_event_rejected(event_id); } diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index e1044514a..d0cfd058e 100644 --- a/src/service/rooms/timeline/append.rs +++ b/src/service/rooms/timeline/append.rs @@ -145,9 +145,17 @@ impl super::Service { // Note that we add it *again* as a no-op in append_pdu. This is done // now so that there's never a point in time where we have zero forward // extremities. - self.services - .state - .add_forward_extremity(&incoming_pdu.room_id_or_hash(), incoming_pdu.event_id()); + let effective_room_id = incoming_pdu.room_id_or_hash(); + if !self + .services + .pdu_metadata + .is_event_referenced(&effective_room_id, incoming_pdu.event_id()) + .await + { + self.services + .state + .add_forward_extremity(&incoming_pdu.room_id_or_hash(), incoming_pdu.event_id()); + } let Some(room_id) = incoming_pdu.room_id() else { // If there's no room ID, this is a v12+ create event, which never // has any predecessors, and consequently will be the @@ -177,19 +185,37 @@ impl super::Service { // Now we collect all the prev events of this predecessor and filter // for ones that we 1. know, 2. are not backfilled, and 3. have // rejected/soft-failed. + // Additionally, drop any events that were referenced late. let outlier_prevs = prev_event .prev_events() .map(ToOwned::to_owned) .stream() .broad_filter_map(|prev_id| async { - let (exists, count, is_accepted) = tokio::join!( + let (exists, count, is_accepted, is_referenced) = tokio::join!( self.pdu_exists(&prev_id), self.get_pdu_count(&prev_id) .unwrap_or(PduCount::Backfilled(i64::MIN)), self.services.pdu_metadata.is_event_accepted(&prev_id), + self.services + .pdu_metadata + .is_event_referenced(room_id, &prev_id) ); - (exists && !matches!(count, PduCount::Backfilled(_)) && !is_accepted) - .then_some(prev_id) + let is_backfilled = matches!(count, PduCount::Backfilled(_)); + if is_referenced { + self.services + .state + .remove_forward_extremity(room_id, &prev_id); + } else { + // If we got here, that means there's now at least an + // indirect reference to the event. + self.services + .pdu_metadata + .mark_as_referenced(room_id, &prev_id); + } + if is_referenced || !exists || !is_accepted || is_backfilled { + return None; + } + Some(prev_id) }) .collect::>() .await;