fix: Address review feedback

src/api/client/membership/invite.rs

@@ -221,7 +221,7 @@ pub(crate) async fn invite_helper(
 		let pdu_id = services
 			.rooms
 			.event_handler
-			.handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, true)
+			.handle_incoming_pdu(recipient_user.server_name(), room_id, &event_id, value, false)
 			.boxed()
 			.await?
 			.ok_or_else(|| {

src/api/server/send.rs

@@ -327,7 +327,7 @@ async fn handle_room(
 		let result = services
 			.rooms
 			.event_handler
-			.handle_incoming_pdu(origin, room_id, &event_id, value, true)
+			.handle_incoming_pdu(origin, room_id, &event_id, value, false)
 			.boxed()
 			.await
 			.map(|_| ());

src/api/server/send_join.rs

@@ -200,7 +200,7 @@ async fn create_join_event(
 	let pdu_id = services
 		.rooms
 		.event_handler
-		.handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), true)
+		.handle_incoming_pdu(sender.server_name(), room_id, &event_id, value.clone(), false)
 		.boxed()
 		.await?
 		.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;

src/api/server/send_knock.rs

@@ -166,7 +166,7 @@ pub(crate) async fn create_knock_event_v1_route(
 	let pdu_id = services
 		.rooms
 		.event_handler
-		.handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), true)
+		.handle_incoming_pdu(sender.server_name(), &body.room_id, &event_id, value.clone(), false)
 		.boxed()
 		.await?
 		.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;

src/api/server/send_leave.rs

@@ -153,7 +153,7 @@ async fn create_leave_event(
 	let pdu_id = services
 		.rooms
 		.event_handler
-		.handle_incoming_pdu(origin, room_id, &event_id, value, true)
+		.handle_incoming_pdu(origin, room_id, &event_id, value, false)
 		.boxed()
 		.await?
 		.ok_or_else(|| err!(Request(InvalidParam("Could not accept as timeline event."))))?;

src/service/rooms/event_handler/fetch_and_handle_outliers.rs

@@ -1,5 +1,6 @@
 use std::{
 	collections::{HashMap, HashSet, VecDeque},
+	fmt::{Display, Formatter},
 	time::Instant,
 };
 
@@ -33,6 +34,20 @@ pub enum DagBuilderTree {
 	AuthEvents,
 }
 
+impl DagBuilderTree {
+	#[must_use]
+	pub fn as_str(&self) -> &str {
+		match self {
+			| Self::AuthEvents => "auth_events",
+			| Self::PrevEvents => "prev_events",
+		}
+	}
+}
+
+impl Display for DagBuilderTree {
+	fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { f.write_str(self.as_str()) }
+}
+
 /// Attempts to build a localised directed acyclic graph out of the given PDUs,
 /// returning them in a topologically sorted order.
 ///
@@ -45,20 +60,19 @@ pub async fn build_local_dag(
 	pdu_map: &HashMap<OwnedEventId, CanonicalJsonObject>,
 	tree: DagBuilderTree,
 ) -> Result<Vec<OwnedEventId>> {
-	debug_assert!(pdu_map.len() >= 2, "needless call to build_local_dag with less than 2 PDUs");
+	if pdu_map.len() <= 1 {
+		return Ok(pdu_map.keys().cloned().collect());
+	}
+
 	let mut dag: HashMap<OwnedEventId, HashSet<OwnedEventId>> =
 		HashMap::with_capacity(pdu_map.len());
 	let mut id_origin_ts: HashMap<OwnedEventId, _> = HashMap::with_capacity(pdu_map.len());
-	let tree = match tree {
-		| DagBuilderTree::AuthEvents => "auth_events",
-		| DagBuilderTree::PrevEvents => "prev_events",
-	};
 
 	for (event_id, value) in pdu_map {
 		// Parse all prev events as event IDs - if they are missing, return an error (we
 		// can't sanely continue in this case), otherwise skip invalid prev events.
 		let prev_events = value
-			.get(tree)
+			.get(tree.as_str())
 			.and_then(CanonicalJsonValue::as_array)
 			.ok_or_else(|| err!(Request(BadJson("event JSON for {event_id} is missing {tree}"))))?
 			.iter()

src/service/rooms/event_handler/fetch_auth.rs

@@ -56,9 +56,9 @@ impl super::Service {
 						continue;
 					},
 				};
-			if !Self::pdu_format_check_1(&auth_pdu_json) {
+			if let Err(e) = Self::pdu_format_check_1(&auth_pdu_json) {
 				// drop this PDU
-				warn!(%auth_event_id, "Dropping auth chain event as it is too large");
+				warn!(%auth_event_id, error=?e, "Dropping auth chain event as it violates the room event format");
 				continue;
 			}
 			let auth_pdu_json = match self

src/service/rooms/event_handler/fetch_prev.rs

@@ -84,11 +84,7 @@ impl super::Service {
 			})
 			.collect::<HashMap<_, _>>();
 
-		let to_persist = if mapped.len() <= 1 {
-			mapped.keys().map(ToOwned::to_owned).collect()
-		} else {
-			build_local_dag(&mapped, DagBuilderTree::PrevEvents).await?
-		};
+		let to_persist = build_local_dag(&mapped, DagBuilderTree::PrevEvents).await?;
 
 		let job_start = Instant::now();
 		trace!("Starting to persist {} prev events", to_persist.len());

src/service/rooms/event_handler/handle_incoming_pdu.rs

@@ -15,6 +15,26 @@ use crate::rooms::timeline::RawPduId;
 
 impl super::Service {
 	/// Handles an incoming PDU from federation.
+	///
+	/// First checks that we want to receive this PDU. If we already have it as
+	/// a timeline PDU, or we don't want to receive the PDU (e.g. origin ACL'd,
+	/// room disabled/unknown), abort.
+	///
+	/// The PDU is then handled as an outlier event, which performs [PDU checks]
+	/// 1 through 4. See: `handle_outlier_pdu`.
+	///
+	/// Once handled as an outlier, any missing prev events are fetched, and
+	/// then the PDU will be promoted/upgraded from an outlier to a timeline
+	/// event clients can see. See: `upgrade_outlier_to_timeline_pdu`. After
+	/// this finishes, the PDU is either accepted or left as an outlier.
+	///
+	/// If the PDU is successfully upgraded, the remaining extremity count of
+	/// the room is checked. If there are a potentially problematic number of
+	/// forward extremities, a squasher task is started with a debounce period,
+	/// which will eventually send a dummy event that ties up as many DAG forks
+	/// as possible.
+	///
+	/// [PDU checks]: https://spec.matrix.org/v1.19/server-server-api/#checks-performed-on-receipt-of-a-pdu
 	#[tracing::instrument(
 		name = "pdu",
 		skip_all,
@@ -26,7 +46,7 @@ impl super::Service {
 		room_id: &'a RoomId,
 		event_id: &'a EventId,
 		value: BTreeMap<String, CanonicalJsonValue>,
-		is_timeline_event: bool,
+		is_backfilled_event: bool,
 	) -> Result<Option<RawPduId>> {
 		// Skip the PDU if we already have it as a timeline event. We still re-process
 		// outliers in this scenario.
@@ -65,7 +85,7 @@ impl super::Service {
 			)));
 		}
 
-		if !room_exists && !is_interesting_member_event {
+		if !room_exists {
 			if is_interesting_member_event {
 				// TODO: handle interesting membership events where we aren't in
 				// the room
@@ -95,9 +115,10 @@ impl super::Service {
 			.handle_outlier_pdu(origin, create_event, event_id, room_id, value)
 			.await?;
 
-		// If this is not a timeline event, stop now, as we don't want to de-outlier it.
-		if !is_timeline_event {
-			debug!("Not promoting incoming event as it is not a timeline event");
+		// If this event is being processed as part of backfill, we don't want to end up
+		// *appending* it during the upgrade process, so we return early.
+		if is_backfilled_event {
+			debug!("Not promoting incoming event as it is being backfilled");
 			return Ok(None);
 		}
 
@@ -110,7 +131,10 @@ impl super::Service {
 			.await?
 			.origin_server_ts();
 		if incoming_pdu.origin_server_ts() < first_ts_in_room {
-			debug!("Not promoting incoming event as it is sent before we joined the room");
+			debug_warn!(
+				"Not promoting incoming event as it is sent before we joined the room (but was \
+				 not backfilled)"
+			);
 			return Ok(None);
 		}
 

src/service/rooms/event_handler/handle_outlier_pdu.rs

@@ -1,7 +1,7 @@
 use std::collections::{BTreeMap, HashMap, hash_map};
 
 use conduwuit::{
-	Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace, warn,
+	Err, Event, EventTypeExt, PduEvent, Result, debug, debug_warn, err, info, trace,
 };
 use ruma::{CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, ServerName};
 
@@ -44,13 +44,12 @@ impl super::Service {
 
 		// 1. Check that the PDU follows the format for the room version
 		// (in this case, just size check)
-		if !Self::pdu_format_check_1(&value) {
-			warn!(
-				"Dropping incoming PDU {event_id} in room {room_id} from {origin} because it \
-				 exceeds 65535 bytes or is otherwise too large."
+		Self::pdu_format_check_1(&value).inspect_err(|e| {
+			info!(
+				err=?e,
+				"Dropping incoming PDU from {origin} because it violates the room event format"
 			);
-			return Err!(Request(TooLarge("PDU is too large")));
-		}
+		})?;
 
 		value.remove("unsigned");
 		let room_version_rules = get_room_version_rules(create_event)?;

src/service/rooms/event_handler/pdu_checks.rs

@@ -16,10 +16,14 @@ impl super::Service {
 	/// Checks that the PDU conforms to the PDU format (check 1). This is
 	/// already mostly done during deserialisation, so this function just checks
 	/// that the PDU isn't a too large.
-	pub(super) fn pdu_format_check_1(pdu_json: &CanonicalJsonObject) -> bool {
+	pub(super) fn pdu_format_check_1(pdu_json: &CanonicalJsonObject) -> Result<()> {
 		// NOTE: if we do any more validation outside of deserialisation, it has to be
 		// done here.
-		pdu_fits(pdu_json)
+		if !pdu_fits(pdu_json) {
+			Err!(Request(TooLarge("PDU is too large")))
+		} else {
+			Ok(())
+		}
 	}
 
 	/// Checks that the PDU has a valid signature (check 2), and redacts it if

src/service/rooms/event_handler/upgrade_outlier_pdu.rs

@@ -1,27 +1,15 @@
-use std::{borrow::Borrow, collections::HashMap, sync::Arc, time::Instant};
+use std::time::Instant;
 
 use conduwuit::{
-	Err, Result, debug, debug_info, debug_warn, is_equal_to, is_true,
+	Err, Result, debug, debug_info, debug_warn, is_true,
 	matrix::{Event, PduEvent},
 	trace,
-	utils::{
-		IterStream,
-		mutex_map::Guard,
-		stream::{BroadbandExt, ReadyExt},
-	},
-};
-use futures::{FutureExt, StreamExt};
-use ruma::{
-	CanonicalJsonObject, OwnedEventId, OwnedRoomId, RoomId, ServerName, events::StateEventType,
-	room_version_rules::RoomVersionRules,
 };
+use ruma::{CanonicalJsonObject, RoomId, ServerName, events::StateEventType};
 use tokio::join;
 
 use super::get_room_version_rules;
-use crate::rooms::{
-	state_compressor::{CompressedState, HashSetCompressStateEvent},
-	timeline::RawPduId,
-};
+use crate::rooms::timeline::RawPduId;
 
 impl super::Service {
 	#[tracing::instrument(name="upgrade_outlier", skip_all, fields(event_id=%incoming_pdu.event_id()))]
@@ -170,55 +158,20 @@ impl super::Service {
 		// The PDU has now passed all checks! We can now promote it (or soft-fail it if
 		// the verdict is such).
 		trace!("Appending pdu to timeline");
-		let mut extremities: Vec<_> = self
-			.services
-			.state
-			.get_forward_extremities(room_id)
-			.collect()
-			.await;
-		if !should_soft_fail {
-			// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
-			// are not added as forward extremities.
-			// This also means we set the state here.
-			// We do this BEFORE setting the extremities so that there's never a point in
-			// time where we have fresh extremities referencing stale state.
-			extremities = self
-				.progress_state_and_extremities(
-					&incoming_pdu,
-					&room_version_rules,
-					state_before.clone(),
-					extremities,
-					&state_lock,
-				)
-				.await?;
-		}
-
-		let state_ids_compressed: Arc<CompressedState> = self
-			.services
-			.state_compressor
-			.compress_state_events(state_before.iter().map(|(ssk, eid)| (ssk, eid.borrow())))
-			.collect()
-			.map(Arc::new)
-			.await;
 		let pdu_id = self
 			.services
 			.timeline
 			.append_incoming_pdu(
 				&incoming_pdu,
 				val,
-				extremities.iter().map(Borrow::borrow),
-				state_ids_compressed,
+				&room_version_rules,
+				state_before,
 				should_soft_fail,
 				&state_lock,
-				room_id,
 			)
 			.await?;
 
 		if should_soft_fail {
-			self.services
-				.pdu_metadata
-				.mark_event_soft_failed(incoming_pdu.event_id());
-
 			debug_info!(
 				elapsed = ?timer.elapsed(),
 				event_id = %incoming_pdu.event_id,
@@ -242,101 +195,4 @@ impl super::Service {
 
 		Ok(pdu_id)
 	}
-
-	/// Derives new room state from the incoming event and filters forward
-	/// extremities accordingly. Does not set forward extremities.
-	///
-	/// Only call this function if the incoming PDU is not soft-failed or
-	/// rejected.
-	async fn progress_state_and_extremities(
-		&self,
-		incoming_pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		state_before: HashMap<u64, OwnedEventId>,
-		forward_extremities: Vec<OwnedEventId>,
-		state_lock: &Guard<OwnedRoomId, ()>,
-	) -> Result<Vec<OwnedEventId>> {
-		if incoming_pdu.state_key().is_some() {
-			debug!("Event is a state-event. Deriving new room state");
-			self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock)
-				.await?;
-		}
-
-		// Now we calculate the set of extremities this room has after the incoming
-		// event has been applied. We start with the previous extremities
-		trace!("Calculating extremities");
-		let mut forward_extremities = forward_extremities
-			.into_iter()
-			.stream()
-			.ready_filter(|event_id| {
-				// Remove any that are referenced by this incoming event's prev_events
-				!incoming_pdu.prev_events().any(is_equal_to!(event_id))
-			})
-			.broad_filter_map(|event_id| async move {
-				// Only keep those extremities were not referenced yet
-				self.services
-					.pdu_metadata
-					.is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id)
-					.await
-					.eq(&false)
-					.then_some(event_id)
-			})
-			.collect::<Vec<_>>()
-			.await;
-		forward_extremities.push(incoming_pdu.event_id().to_owned());
-		debug!(
-			"Retained {} extremities checked against {} prev_events",
-			forward_extremities.len(),
-			incoming_pdu.prev_events().count()
-		);
-		assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty");
-		Ok(forward_extremities)
-	}
-
-	/// Derives a new room state by adding the incoming PDU to the state before
-	/// it to create the state at, which then becomes the current room state.
-	///
-	/// The caller MUST ensure forward extremities are set appropriately,
-	/// including this incoming pdu, either before or after calling this
-	/// function. Failing to do so will result in an inconsistent current state
-	/// cache, which may affect event authentication.
-	async fn derive_new_state(
-		&self,
-		incoming_pdu: &PduEvent,
-		room_version_rules: &RoomVersionRules,
-		state_before: HashMap<u64, OwnedEventId>,
-		state_lock: &Guard<OwnedRoomId, ()>,
-	) -> Result {
-		let room_id = incoming_pdu.room_id_or_hash();
-		// We also add state after incoming event to the fork states
-		let mut state_at_incoming_event = state_before;
-		let shortstatekey = self
-			.services
-			.short
-			.get_or_create_shortstatekey(
-				&incoming_pdu.kind().to_string().into(),
-				incoming_pdu.state_key().unwrap(),
-			)
-			.await;
-
-		let event_id = incoming_pdu.event_id();
-		state_at_incoming_event.insert(shortstatekey, event_id.to_owned());
-
-		debug!("Resolving new room state");
-		let new_room_state = self
-			.resolve_state(&room_id, room_version_rules, state_at_incoming_event)
-			.await?;
-
-		debug!("Forcing new room state");
-		let HashSetCompressStateEvent { shortstatehash, added, removed } = self
-			.services
-			.state_compressor
-			.save_state(&room_id, new_room_state)
-			.await?;
-
-		self.services
-			.state
-			.force_state(&room_id, shortstatehash, added, removed, state_lock)
-			.await
-	}
 }

src/service/rooms/timeline/append.rs

@@ -1,69 +1,113 @@
 use std::{
-	collections::{BTreeMap, HashSet},
+	borrow::Borrow,
+	collections::{BTreeMap, HashMap, HashSet},
 	sync::Arc,
 };
 
 use conduwuit::{
-	debug_warn,
+	Result, debug, debug_warn, is_equal_to,
+	matrix::{Event, PduEvent},
 	pdu::{Count, ShortRoomId},
 	trace,
-	utils::{IterStream, TryFutureExtExt, stream::BroadbandExt},
+	utils::{
+		IterStream, TryFutureExtExt,
+		mutex_map::Guard,
+		stream::{BroadbandExt, ReadyExt},
+	},
 	warn,
 };
 use conduwuit_core::{
-	Result, err, error,
-	matrix::{
-		event::Event,
-		pdu::{PduCount, PduEvent, PduId, RawPduId},
-	},
-	utils::{self, ReadyExt},
+	err, error,
+	matrix::pdu::{PduCount, PduId, RawPduId},
+	utils::{self},
 };
-use futures::{StreamExt, TryFutureExt};
+use futures::{FutureExt, StreamExt, TryFutureExt};
 use ruma::{
-	CanonicalJsonObject, CanonicalJsonValue, EventId, RoomId, RoomVersionId, UserId,
+	CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, RoomId,
+	RoomVersionId, UserId,
 	events::{
 		GlobalAccountDataEventType, TimelineEventType,
 		push_rules::PushRulesEvent,
 		room::{encrypted::Relation, redaction::RoomRedactionEventContent},
 	},
 	push::{Action, Ruleset, Tweak},
+	room_version_rules::RoomVersionRules,
 };
 
 use super::{ExtractBody, ExtractRelatesTo, ExtractRelatesToEventId, RoomMutexGuard};
-use crate::{appservice::RegistrationInfo, rooms::state_compressor::CompressedState};
+use crate::{
+	appservice::RegistrationInfo,
+	rooms::state_compressor::{CompressedState, HashSetCompressStateEvent},
+};
 
 impl super::Service {
 	/// Append the incoming event setting the state snapshot to the state from
 	/// the server that sent the event.
-	#[allow(clippy::too_many_arguments)]
-	pub async fn append_incoming_pdu<'a, Leaves>(
+	pub async fn append_incoming_pdu<'a>(
 		&'a self,
 		pdu: &'a PduEvent,
 		pdu_json: CanonicalJsonObject,
-		new_room_leaves: Leaves,
-		state_ids_compressed: Arc<CompressedState>,
-		soft_fail: bool,
+		room_version_rules: &RoomVersionRules,
+		state_before: HashMap<u64, OwnedEventId>,
+		should_soft_fail: bool,
 		state_lock: &'a RoomMutexGuard,
-		room_id: &'a RoomId,
-	) -> Result<Option<RawPduId>>
-	where
-		Leaves: Iterator<Item = &'a EventId> + Send + 'a,
-	{
+	) -> Result<Option<RawPduId>> {
+		let room_id = pdu.room_id_or_hash();
+
+		let sb = state_before.iter().map(|(ssk, eid)| (ssk, eid.as_ref()));
+		let state_ids_compressed: Arc<CompressedState> = self
+			.services
+			.state_compressor
+			.compress_state_events(sb)
+			.collect()
+			.map(Arc::new)
+			.await;
+
 		// We append to state before appending the pdu, so we don't have a moment in
 		// time with the pdu without it's state. This is okay because append_pdu can't
 		// fail.
 		self.services
 			.state
-			.set_event_state(&pdu.event_id, room_id, state_ids_compressed)
+			.set_event_state(&pdu.event_id, &room_id, state_ids_compressed)
 			.await?;
 
-		if soft_fail {
+		if should_soft_fail {
 			// Nothing else to do with a soft-failed event.
+			self.services
+				.pdu_metadata
+				.mark_event_soft_failed(pdu.event_id());
 			return Ok(None);
 		}
 
+		// Per https://spec.matrix.org/unstable/server-server-api/#soft-failure, soft-failed events
+		// are not added as forward extremities.
+		// This also means we set the state here.
+		// We do this BEFORE setting the extremities so that there's never a point in
+		// time where we have fresh extremities referencing stale state.
+		let mut extremities: Vec<_> = self
+			.services
+			.state
+			.get_forward_extremities(&room_id)
+			.collect()
+			.await;
+		extremities = self
+			.progress_state_and_extremities(
+				pdu,
+				room_version_rules,
+				state_before.clone(),
+				extremities,
+				state_lock,
+			)
+			.await?;
+
 		let pdu_id = self
-			.append_pdu(pdu, pdu_json, new_room_leaves, state_lock, room_id)
+			.append_pdu(
+				pdu,
+				pdu_json,
+				extremities.iter().map(Borrow::borrow),
+				state_lock,
+				&room_id,
+			)
 			.await?;
 
 		// Process admin commands for federation events
@@ -86,11 +130,110 @@ impl super::Service {
 			}
 		}
 
-		self.services.sync.wake_all_joined(room_id).await;
+		self.services.sync.wake_all_joined(&room_id).await;
 
 		Ok(Some(pdu_id))
 	}
 
+	/// Derives new room state from the incoming event and filters forward
+	/// extremities accordingly. Does not set forward extremities.
+	///
+	/// Only call this function if the incoming PDU is not soft-failed or
+	/// rejected.
+	async fn progress_state_and_extremities(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		state_before: HashMap<u64, OwnedEventId>,
+		forward_extremities: Vec<OwnedEventId>,
+		state_lock: &Guard<OwnedRoomId, ()>,
+	) -> Result<Vec<OwnedEventId>> {
+		if incoming_pdu.state_key().is_some() {
+			debug!("Event is a state-event. Deriving new room state");
+			self.derive_new_state(incoming_pdu, room_version_rules, state_before, state_lock)
+				.await?;
+		}
+
+		// Now we calculate the set of extremities this room has after the incoming
+		// event has been applied. We start with the previous extremities
+		trace!("Calculating extremities");
+		let mut forward_extremities = forward_extremities
+			.into_iter()
+			.stream()
+			.ready_filter(|event_id| {
+				// Remove any that are referenced by this incoming event's prev_events
+				!incoming_pdu.prev_events().any(is_equal_to!(event_id))
+			})
+			.broad_filter_map(|event_id| async move {
+				// Only keep those extremities were not referenced yet
+				self.services
+					.pdu_metadata
+					.is_event_referenced(&incoming_pdu.room_id_or_hash(), &event_id)
+					.await
+					.eq(&false)
+					.then_some(event_id)
+			})
+			.collect::<Vec<_>>()
+			.await;
+		forward_extremities.push(incoming_pdu.event_id().to_owned());
+		debug!(
+			"Retained {} extremities checked against {} prev_events",
+			forward_extremities.len(),
+			incoming_pdu.prev_events().count()
+		);
+		assert!(!forward_extremities.is_empty(), "resolved extremities cannot be empty");
+		Ok(forward_extremities)
+	}
+
+	/// Derives a new room state by adding the incoming PDU to the state before
+	/// it to create the state at, which then becomes the current room state.
+	///
+	/// The caller MUST ensure forward extremities are set appropriately,
+	/// including this incoming pdu, either before or after calling this
+	/// function. Failing to do so will result in an inconsistent current state
+	/// cache, which may affect event authentication.
+	async fn derive_new_state(
+		&self,
+		incoming_pdu: &PduEvent,
+		room_version_rules: &RoomVersionRules,
+		state_before: HashMap<u64, OwnedEventId>,
+		state_lock: &Guard<OwnedRoomId, ()>,
+	) -> Result {
+		let room_id = incoming_pdu.room_id_or_hash();
+		// We also add state after incoming event to the fork states
+		let mut state_at_incoming_event = state_before;
+		let shortstatekey = self
+			.services
+			.short
+			.get_or_create_shortstatekey(
+				&incoming_pdu.kind().to_string().into(),
+				incoming_pdu.state_key().unwrap(),
+			)
+			.await;
+
+		let event_id = incoming_pdu.event_id();
+		state_at_incoming_event.insert(shortstatekey, event_id.to_owned());
+
+		debug!("Resolving new room state");
+		let new_room_state = self
+			.services
+			.event_handler
+			.resolve_state(&room_id, room_version_rules, state_at_incoming_event)
+			.await?;
+
+		debug!("Forcing new room state");
+		let HashSetCompressStateEvent { shortstatehash, added, removed } = self
+			.services
+			.state_compressor
+			.save_state(&room_id, new_room_state)
+			.await?;
+
+		self.services
+			.state
+			.force_state(&room_id, shortstatehash, added, removed, state_lock)
+			.await
+	}
+
 	/// Populates the unsigned data of a PDU
 	async fn populate_unsigned(&self, pdu: &PduEvent, pdu_json: &mut CanonicalJsonObject) {
 		let Some(state_key) = pdu.state_key() else {

src/service/rooms/timeline/backfill.rs

@@ -143,7 +143,7 @@ impl super::Service {
 				| Ok(value) => {
 					self.services
 						.event_handler
-						.handle_incoming_pdu(&backfill_server, room_id, event_id, value, false)
+						.handle_incoming_pdu(&backfill_server, room_id, event_id, value, true)
 						.boxed()
 						.await?;
 					debug!("Successfully backfilled {event_id} from {backfill_server}");
@@ -185,7 +185,7 @@ impl super::Service {
 
 		self.services
 			.event_handler
-			.handle_incoming_pdu(origin, &room_id, &event_id, value, false)
+			.handle_incoming_pdu(origin, &room_id, &event_id, value, true)
 			.boxed()
 			.await?;
 

src/service/rooms/timeline/mod.rs

@@ -77,6 +77,7 @@ struct Services {
 	short: Dep<rooms::short::Service>,
 	state: Dep<rooms::state::Service>,
 	state_accessor: Dep<rooms::state_accessor::Service>,
+	state_compressor: Dep<rooms::state_compressor::Service>,
 	state_cache: Dep<rooms::state_cache::Service>,
 	sync: Dep<sync::Service>,
 	threads: Dep<rooms::threads::Service>,
@@ -110,6 +111,8 @@ impl crate::Service for Service {
 				state: args.depend::<rooms::state::Service>("rooms::state"),
 				state_accessor: args
 					.depend::<rooms::state_accessor::Service>("rooms::state_accessor"),
+				state_compressor: args
+					.depend::<rooms::state_compressor::Service>("rooms::state_compressor"),
 				state_cache: args.depend::<rooms::state_cache::Service>("rooms::state_cache"),
 				sync: args.depend::<sync::Service>("sync"),
 				threads: args.depend::<rooms::threads::Service>("rooms::threads"),